Fix default discovery crash on Windows (#5202)

Author: pjanotti

cmd/otelcol/main.go

@@ -18,6 +18,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"os"
@@ -36,10 +37,14 @@ import (
 )
 
 func main() {
+	runFromCmdLine(os.Args)
+}
+
+func runFromCmdLine(args []string) {
 	// TODO: Use same format as the collector
 	log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)
 
-	collectorSettings, err := settings.New(os.Args[1:])
+	collectorSettings, err := settings.New(args[1:])
 	if err != nil {
 		// Exit if --help flag was supplied and usage help was displayed.
 		if err == flag.ErrHelp {
@@ -80,14 +85,21 @@ func main() {
 		},
 	}
 
-	os.Args = append(os.Args[:1], collectorSettings.ColCoreArgs()...)
+	allArgs := args[:1]
+	allArgs = append(allArgs, collectorSettings.ColCoreArgs()...)
+	os.Args = allArgs
 	if err = run(serviceSettings); err != nil {
 		log.Fatal(err)
 	}
 }
 
+var otelcolCmdTestCtx context.Context // Use to control termination during tests.
+
 func runInteractive(settings otelcol.CollectorSettings) error {
 	cmd := otelcol.NewCommand(settings)
+	if otelcolCmdTestCtx != nil {
+		cmd.SetContext(otelcolCmdTestCtx)
+	}
 	if err := cmd.Execute(); err != nil {
 		return fmt.Errorf("application run finished with error: %w", err)
 	}

cmd/otelcol/main_test.go

@@ -14,3 +14,111 @@
 // limitations under the License.
 
 package main
+
+import (
+	"context"
+	"net"
+	"os"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRunFromCmdLine(t *testing.T) {
+	tests := []struct {
+		name     string
+		panicMsg string
+		skipMsg  string
+		args     []string
+		timeout  time.Duration
+	}{
+		{
+			name:    "agent",
+			args:    []string{"otelcol", "--config=config/collector/agent_config.yaml"},
+			timeout: 15 * time.Second,
+		},
+		{
+			name:    "gateway",
+			args:    []string{"otelcol", "--config=config/collector/gateway_config.yaml"},
+			timeout: 15 * time.Second,
+		},
+		{
+			name:    "default_discovery",
+			args:    []string{"otelcol", "--discovery", "--config=config/collector/agent_config.yaml"},
+			timeout: 30 * time.Second,
+		},
+		// Running the discovery with --dry-run in CI is not desirable because of the use of os.Exit(0) to end the execution.
+		// That prevents the test from releasing resources like ports. The test needs to catch the panic to not fail the test,
+		// however, the resources won't be properly released for the remaining tests that may use the same resources.
+		// Skipping the test by default but keeping it around to deliberate runs on dev box.
+		{
+			name:     "dry-run_discovery",
+			args:     []string{"otelcol", "--discovery", "--dry-run", "--config=config/collector/agent_config.yaml"},
+			timeout:  30 * time.Second,
+			panicMsg: "unexpected call to os.Exit(0) during test", // os.Exit(0) in the normal execution is expected for '--dry-run'.
+			skipMsg:  "Skipping this test by default because --dry-run uses os.Exit(0) to end the execution",
+		},
+	}
+
+	// Set execution environment
+	requiredEnvVars := map[string]string{
+		"SPLUNK_ACCESS_TOKEN":     "access_token",
+		"SPLUNK_HEC_TOKEN":        "hec_token",
+		"SPLUNK_REALM":            "test_realm",
+		"SPLUNK_LISTEN_INTERFACE": "127.0.0.1",
+		"NO_WINDOWS_SERVICE":      "true", // Avoid using the Windows service manager
+	}
+	for key, value := range requiredEnvVars {
+		os.Setenv(key, value)
+		defer os.Unsetenv(key)
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if tt.skipMsg != "" {
+				t.Skip(tt.skipMsg)
+			}
+
+			// GH darwin runners don't have docker installed, skip discovery tests on them
+			// given that the docker_observer is enabled by default.
+			if runtime.GOOS == "darwin" && (tt.name == "default_discovery" || tt.name == "dry-run_discovery") {
+				if os.Getenv("GITHUB_ACTIONS") == "true" {
+					t.Skip("skipping discovery tests on darwin runners since they don't have docker installed")
+				}
+			}
+
+			testCtx, cancel := context.WithTimeout(context.Background(), tt.timeout)
+			defer cancel()
+
+			otelcolCmdTestCtx = testCtx
+			defer func() {
+				otelcolCmdTestCtx = nil
+			}()
+
+			// Wait for the ConfigServer to be down after the test.
+			defer waitForPort(t, "55554")
+
+			if tt.panicMsg != "" {
+				assert.PanicsWithValue(t, tt.panicMsg, func() { runFromCmdLine(tt.args) })
+				return
+			}
+
+			waitForPort(t, "55554")
+			runFromCmdLine(tt.args)
+		})
+	}
+}
+
+func waitForPort(t *testing.T, port string) {
+	require.Eventually(t, func() bool {
+		ln, err := net.Listen("tcp", "localhost:"+port)
+		if err == nil {
+			ln.Close()
+			return true
+		}
+		return false
+	}, 60*time.Second, 500*time.Millisecond)
+}

internal/confmapprovider/discovery/discoverer.go

@@ -19,6 +19,7 @@ import (
 	"encoding/base64"
 	"fmt"
 	"os"
+	"runtime"
 	"sort"
 	"strconv"
 	"strings"
@@ -182,6 +183,19 @@ func (d *discoverer) discover(cfg *Config) (map[string]any, error) {
 		return nil, nil
 	}
 
+	err = d.performDiscovery(discoveryReceivers, discoveryObservers)
+	if err != nil {
+		return nil, err
+	}
+
+	discoveryConfig, err := d.discoveryConfig(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed constructing discovery config: %w", err)
+	}
+	return discoveryConfig, nil
+}
+
+func (d *discoverer) performDiscovery(discoveryReceivers map[component.ID]otelcolreceiver.Logs, discoveryObservers map[component.ID]otelcolextension.Extension) error {
 	var cancels []context.CancelFunc
 
 	defer func() {
@@ -199,19 +213,35 @@ func (d *discoverer) discover(cfg *Config) (map[string]any, error) {
 				fmt.Sprintf("%q startup failed. Won't proceed with %q-based discovery", observerID, observerID.Type()),
 				zap.Error(e),
 			)
+			return e
 		}
+		defer func(obsID component.ID, obsExt otelcolextension.Extension) {
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			cancels = append(cancels, cancel)
+			if e := obsExt.Shutdown(ctx); e != nil {
+				d.logger.Warn(fmt.Sprintf("error shutting down observer %q", obsID), zap.Error(e))
+			}
+		}(observerID, observer)
 	}
 
 	for receiverID, receiver := range discoveryReceivers {
 		d.logger.Debug(fmt.Sprintf("starting receiver %q", receiverID))
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		cancels = append(cancels, cancel)
-		if err = receiver.Start(ctx, d); err != nil {
+		if err := receiver.Start(ctx, d); err != nil {
 			d.logger.Warn(
 				fmt.Sprintf("%q startup failed.", receiverID),
 				zap.Error(err),
 			)
+			return err
 		}
+		defer func(rcvID component.ID, rcv otelcolreceiver.Logs) {
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			cancels = append(cancels, cancel)
+			if e := rcv.Shutdown(ctx); e != nil {
+				d.logger.Warn(fmt.Sprintf("error shutting down receiver %q", rcvID), zap.Error(e))
+			}
+		}(receiverID, receiver)
 	}
 
 	_, _ = fmt.Fprintf(os.Stderr, "Discovering for next %s...\n", d.duration)
@@ -221,26 +251,7 @@ func (d *discoverer) discover(cfg *Config) (map[string]any, error) {
 	}
 	_, _ = fmt.Fprintf(os.Stderr, "Discovery complete.\n")
 
-	for receiverID, receiver := range discoveryReceivers {
-		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-		cancels = append(cancels, cancel)
-		if e := receiver.Shutdown(ctx); e != nil {
-			d.logger.Warn(fmt.Sprintf("error shutting down receiver %q", receiverID), zap.Error(e))
-		}
-	}
-	for observerID, observer := range discoveryObservers {
-		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-		cancels = append(cancels, cancel)
-		if e := observer.Shutdown(ctx); e != nil {
-			d.logger.Warn(fmt.Sprintf("error shutting down observer %q", observerID), zap.Error(e))
-		}
-	}
-
-	discoveryConfig, err := d.discoveryConfig(cfg)
-	if err != nil {
-		return nil, fmt.Errorf("failed constructing discovery config: %w", err)
-	}
-	return discoveryConfig, nil
+	return nil
 }
 
 func (d *discoverer) createDiscoveryReceiversAndObservers(cfg *Config) (map[component.ID]otelcolreceiver.Logs, map[component.ID]otelcolextension.Extension, error) {
@@ -449,11 +460,15 @@ func (d *discoverer) updateReceiverForObserver(receiverID component.ID, receiver
 
 func factoryForObserverType(extType component.Type) (otelcolextension.Factory, error) {
 	factories := map[component.Type]otelcolextension.Factory{
-		component.MustNewType("docker_observer"):   dockerobserver.NewFactory(),
 		component.MustNewType("host_observer"):     hostobserver.NewFactory(),
 		component.MustNewType("k8s_observer"):      k8sobserver.NewFactory(),
 		component.MustNewType("ecs_task_observer"): ecstaskobserver.NewFactory(),
 	}
+	if runtime.GOOS != "windows" {
+		// Docker observer currently always crashes on Windows with the default configuration.
+		// The observer is being temporarily disabled until it is fixed upstream.
+		factories[component.MustNewType("docker_observer")] = dockerobserver.NewFactory()
+	}
 	ef, ok := factories[extType]
 	if !ok {
 		return nil, fmt.Errorf("unsupported discovery observer %q. Please remove its .discovery.yaml from your config directory", extType)