fix: load healthcheck v2 once when edot is running as subprocess (#9848)

* ci: extend unit-test to validate number of healthcheck extensions in edot subprocess mode

* fix: instantiate and use healthcheck extension id once

Author: pkoutsovasilis

internal/pkg/otel/manager/execution_subprocess.go

@@ -13,6 +13,8 @@ import (
 	"os/exec"
 	"time"
 
+	"github.com/gofrs/uuid/v5"
+	"go.opentelemetry.io/collector/component"
 	"gopkg.in/yaml.v3"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status"
@@ -34,22 +36,34 @@ const (
 	OtelSupervisedLoggingLevelFlagName = "supervised.logging.level"
 )
 
-func newSubprocessExecution(logLevel logp.Level, collectorPath string) *subprocessExecution {
+func newSubprocessExecution(logLevel logp.Level, collectorPath string) (*subprocessExecution, error) {
+	nsUUID, err := uuid.NewV4()
+	if err != nil {
+		return nil, fmt.Errorf("cannot generate UUID: %w", err)
+	}
+	componentType, err := component.NewType(healthCheckExtensionName)
+	if err != nil {
+		return nil, fmt.Errorf("cannot create component type: %w", err)
+	}
+	healthCheckExtensionID := component.NewIDWithName(componentType, nsUUID.String()).String()
+
 	return &subprocessExecution{
 		collectorPath: collectorPath,
 		collectorArgs: []string{
 			"otel",
 			fmt.Sprintf("--%s", OtelSetSupervisedFlagName),
 			fmt.Sprintf("--%s=%s", OtelSupervisedLoggingLevelFlagName, logLevel.String()),
 		},
-		logLevel: logLevel,
-	}
+		logLevel:               logLevel,
+		healthCheckExtensionID: healthCheckExtensionID,
+	}, nil
 }
 
 type subprocessExecution struct {
-	collectorPath string
-	collectorArgs []string
-	logLevel      logp.Level
+	collectorPath          string
+	collectorArgs          []string
+	logLevel               logp.Level
+	healthCheckExtensionID string
 }
 
 // startCollector starts a supervised collector and monitors its health. Process exit errors are sent to the
@@ -75,7 +89,7 @@ func (r *subprocessExecution) startCollector(ctx context.Context, logger *logger
 		return nil, fmt.Errorf("could not find port for http health check: %w", err)
 	}
 
-	if err := injectHeathCheckV2Extension(cfg, httpHealthCheckPort); err != nil {
+	if err := injectHeathCheckV2Extension(cfg, r.healthCheckExtensionID, httpHealthCheckPort); err != nil {
 		return nil, fmt.Errorf("failed to inject health check extension: %w", err)
 	}
 

internal/pkg/otel/manager/healthcheck.go

@@ -13,8 +13,6 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/gofrs/uuid/v5"
-	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/confmap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status"
@@ -197,14 +195,8 @@ func aggregateStatus(sts componentstatus.Status, err error) *status.AggregateSta
 }
 
 // injectHeathCheckV2Extension injects the healthcheckv2 extension into the provided configuration.
-func injectHeathCheckV2Extension(conf *confmap.Conf, httpHealthCheckPort int) error {
-	nsUUID, err := uuid.NewV4()
-	if err != nil {
-		return fmt.Errorf("cannot generate UUID V4: %w", err)
-	}
-	componentType := component.MustNewType(healthCheckExtensionName)
-	healthCheckExtensionID := component.NewIDWithName(componentType, nsUUID.String()).String()
-	err = conf.Merge(confmap.NewFromStringMap(map[string]interface{}{
+func injectHeathCheckV2Extension(conf *confmap.Conf, healthCheckExtensionID string, httpHealthCheckPort int) error {
+	err := conf.Merge(confmap.NewFromStringMap(map[string]interface{}{
 		"extensions": map[string]interface{}{
 			healthCheckExtensionID: map[string]interface{}{
 				"use_v2": true,

internal/pkg/otel/manager/manager.go

@@ -123,7 +123,10 @@ func NewOTelManager(
 			return nil, fmt.Errorf("failed to get the path to the collector executable: %w", err)
 		}
 		recoveryTimer = newRecoveryBackoff(100*time.Nanosecond, 10*time.Second, time.Minute)
-		exec = newSubprocessExecution(logLevel, executable)
+		exec, err = newSubprocessExecution(logLevel, executable)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create subprocess execution: %w", err)
+		}
 	case EmbeddedExecutionMode:
 		recoveryTimer = newRestarterNoop()
 		exec = newExecutionEmbedded()

internal/pkg/otel/manager/manager_test.go

@@ -11,6 +11,7 @@ import (
 	"errors"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -264,6 +265,22 @@ func (t *EventTime[T]) Time() time.Time {
 	return t.time
 }
 
+func countHealthCheckExtensionStatuses(status *status.AggregateStatus) uint {
+	extensions, ok := status.ComponentStatusMap["extensions"]
+	if !ok {
+		return 0
+	}
+
+	count := uint(0)
+	for key := range extensions.ComponentStatusMap {
+		if strings.HasPrefix(key, "extension:healthcheckv2/") {
+			count++
+		}
+	}
+
+	return count
+}
+
 func TestOTelManager_Run(t *testing.T) {
 	wd, erWd := os.Getwd()
 	require.NoError(t, erWd, "cannot get working directory")
@@ -273,14 +290,16 @@ func TestOTelManager_Run(t *testing.T) {
 
 	for _, tc := range []struct {
 		name                string
-		exec                *testExecution
+		exec                func() (collectorExecution, error)
 		restarter           collectorRecoveryTimer
 		skipListeningErrors bool
 		testFn              func(t *testing.T, m *OTelManager, e *EventListener, exec *testExecution)
 	}{
 		{
-			name:      "embedded collector config updates",
-			exec:      &testExecution{exec: newExecutionEmbedded()},
+			name: "embedded collector config updates",
+			exec: func() (collectorExecution, error) {
+				return newExecutionEmbedded(), nil
+			},
 			restarter: newRestarterNoop(),
 			testFn: func(t *testing.T, m *OTelManager, e *EventListener, exec *testExecution) {
 				// ensure that it got healthy
@@ -302,8 +321,10 @@ func TestOTelManager_Run(t *testing.T) {
 			},
 		},
 		{
-			name:      "subprocess collector config updates",
-			exec:      &testExecution{exec: newSubprocessExecution(logp.DebugLevel, testBinary)},
+			name: "subprocess collector config updates",
+			exec: func() (collectorExecution, error) {
+				return newSubprocessExecution(logp.DebugLevel, testBinary)
+			},
 			restarter: newRecoveryBackoff(100*time.Nanosecond, 10*time.Second, time.Minute),
 			testFn: func(t *testing.T, m *OTelManager, e *EventListener, exec *testExecution) {
 				// ensure that it got healthy
@@ -326,8 +347,10 @@ func TestOTelManager_Run(t *testing.T) {
 			},
 		},
 		{
-			name:      "embedded collector stopped gracefully outside manager",
-			exec:      &testExecution{exec: newExecutionEmbedded()},
+			name: "embedded collector stopped gracefully outside manager",
+			exec: func() (collectorExecution, error) {
+				return newExecutionEmbedded(), nil
+			},
 			restarter: newRestarterNoop(),
 			testFn: func(t *testing.T, m *OTelManager, e *EventListener, exec *testExecution) {
 				// ensure that it got healthy
@@ -350,8 +373,10 @@ func TestOTelManager_Run(t *testing.T) {
 			},
 		},
 		{
-			name:      "subprocess collector stopped gracefully outside manager",
-			exec:      &testExecution{exec: newSubprocessExecution(logp.DebugLevel, testBinary)},
+			name: "subprocess collector stopped gracefully outside manager",
+			exec: func() (collectorExecution, error) {
+				return newSubprocessExecution(logp.DebugLevel, testBinary)
+			},
 			restarter: newRecoveryBackoff(100*time.Nanosecond, 10*time.Second, time.Minute),
 			testFn: func(t *testing.T, m *OTelManager, e *EventListener, exec *testExecution) {
 				// ensure that it got healthy
@@ -365,6 +390,7 @@ func TestOTelManager_Run(t *testing.T) {
 				require.NotNil(t, exec.handle, "exec handle should not be nil")
 				exec.handle.Stop(t.Context())
 				e.EnsureHealthy(t, updateTime)
+				assert.EqualValues(t, 1, countHealthCheckExtensionStatuses(e.getStatus()), "health check extension status count should be 1")
 
 				// no configuration should stop the runner
 				updateTime = time.Now()
@@ -375,15 +401,18 @@ func TestOTelManager_Run(t *testing.T) {
 			},
 		},
 		{
-			name:      "subprocess collector killed outside manager",
-			exec:      &testExecution{exec: newSubprocessExecution(logp.DebugLevel, testBinary)},
+			name: "subprocess collector killed outside manager",
+			exec: func() (collectorExecution, error) {
+				return newSubprocessExecution(logp.DebugLevel, testBinary)
+			},
 			restarter: newRecoveryBackoff(100*time.Nanosecond, 10*time.Second, time.Minute),
 			testFn: func(t *testing.T, m *OTelManager, e *EventListener, exec *testExecution) {
 				// ensure that it got healthy
 				cfg := confmap.NewFromStringMap(testConfig)
 				updateTime := time.Now()
 				m.Update(cfg, nil)
 				e.EnsureHealthy(t, updateTime)
+				assert.EqualValues(t, 1, countHealthCheckExtensionStatuses(e.getStatus()), "health check extension status count should be 1")
 
 				var oldPHandle *procHandle
 				// repeatedly kill the collector
@@ -401,6 +430,7 @@ func TestOTelManager_Run(t *testing.T) {
 					// the collector should restart and report healthy
 					updateTime = time.Now()
 					e.EnsureHealthy(t, updateTime)
+					assert.EqualValues(t, 1, countHealthCheckExtensionStatuses(e.getStatus()), "health check extension status count should be 1")
 				}
 
 				seenRecoveredTimes := m.recoveryRetries.Load()
@@ -414,8 +444,10 @@ func TestOTelManager_Run(t *testing.T) {
 			},
 		},
 		{
-			name:      "subprocess collector panics",
-			exec:      &testExecution{exec: newSubprocessExecution(logp.DebugLevel, testBinary)},
+			name: "subprocess collector panics",
+			exec: func() (collectorExecution, error) {
+				return newSubprocessExecution(logp.DebugLevel, testBinary)
+			},
 			restarter: newRecoveryBackoff(100*time.Nanosecond, 10*time.Second, time.Minute),
 			testFn: func(t *testing.T, m *OTelManager, e *EventListener, exec *testExecution) {
 				err := os.Setenv("TEST_SUPERVISED_COLLECTOR_PANIC", (3 * time.Second).String())
@@ -448,8 +480,10 @@ func TestOTelManager_Run(t *testing.T) {
 			},
 		},
 		{
-			name:                "embedded collector invalid config",
-			exec:                &testExecution{exec: newExecutionEmbedded()},
+			name: "embedded collector invalid config",
+			exec: func() (collectorExecution, error) {
+				return newExecutionEmbedded(), nil
+			},
 			restarter:           newRestarterNoop(),
 			skipListeningErrors: true,
 			testFn: func(t *testing.T, m *OTelManager, e *EventListener, exec *testExecution) {
@@ -487,8 +521,10 @@ func TestOTelManager_Run(t *testing.T) {
 			},
 		},
 		{
-			name:                "subprocess collector invalid config",
-			exec:                &testExecution{exec: newSubprocessExecution(logp.DebugLevel, testBinary)},
+			name: "subprocess collector invalid config",
+			exec: func() (collectorExecution, error) {
+				return newSubprocessExecution(logp.DebugLevel, testBinary)
+			},
 			restarter:           newRecoveryBackoff(100*time.Nanosecond, 10*time.Second, time.Minute),
 			skipListeningErrors: true,
 			testFn: func(t *testing.T, m *OTelManager, e *EventListener, exec *testExecution) {
@@ -531,6 +567,11 @@ func TestOTelManager_Run(t *testing.T) {
 			defer cancel()
 			l, _ := loggertest.New("otel")
 			base, obs := loggertest.New("otel")
+
+			executionMode, err := tc.exec()
+			require.NoError(t, err, "failed to create execution mode")
+			testExecutionMode := &testExecution{exec: executionMode}
+
 			m := &OTelManager{
 				logger:            l,
 				baseLogger:        base,
@@ -540,7 +581,7 @@ func TestOTelManager_Run(t *testing.T) {
 				componentStateCh:  make(chan []runtime.ComponentComponentState, 1),
 				doneChan:          make(chan struct{}),
 				recoveryTimer:     tc.restarter,
-				execution:         tc.exec,
+				execution:         testExecutionMode,
 			}
 
 			eListener := &EventListener{}
@@ -573,7 +614,7 @@ func TestOTelManager_Run(t *testing.T) {
 				runErr = m.Run(ctx)
 			}()
 
-			tc.testFn(t, m, eListener, tc.exec)
+			tc.testFn(t, m, eListener, testExecutionMode)
 
 			cancel()
 			runWg.Wait()