Add support for running EDOT inside of running Elastic Agent (#5767)

.github/workflows/golangci-lint.yml

@@ -25,10 +25,10 @@ jobs:
           go-version-file: .go-version
 
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v6
+        uses: golangci/golangci-lint-action@v6.1.1
         with:
           # Optional: version of golangci-lint to use in form of v1.2 or v1.2.3 or `latest` to use the latest version
-          version: v1.55.2
+          version: v1.61.0
 
           # Give the job more time to execute.
           # Regarding `--whole-files`, the linter is supposed to support linting of changed a patch only but,

changelog/fragments/1729011748-Add-EDOT-hybrid-mode.yaml

@@ -0,0 +1,32 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Add ability to run Elastic Distribution of OTel Collector at the same time as other inputs
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description:
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: https://github.com/elastic/elastic-agent/pull/5767
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+#issue: https://github.com/owner/repo/1234

control_v2.proto

@@ -23,6 +23,18 @@ enum State {
   ROLLBACK = 8;
 }
 
+// CollectorComponentStatus used for OTel collector components.
+enum CollectorComponentStatus {
+    StatusNone = 0;
+    StatusStarting = 1;
+    StatusOK = 2;
+    StatusRecoverableError = 3;
+    StatusPermanentError = 4;
+    StatusFatalError = 5;
+    StatusStopping = 6;
+    StatusStopped = 7;
+}
+
 // Unit Type running inside a component.
 enum UnitType {
   INPUT = 0;
@@ -173,6 +185,18 @@ message StateAgentInfo {
   bool isManaged = 8;
 }
 
+// CollectorComponent is the status of an OTel collector component.
+message CollectorComponent {
+  // Status of the component.
+  CollectorComponentStatus status = 1;
+  // Error is set to the reported error.
+  string error = 2;
+  // Timestamp of status.
+  string timestamp = 3;
+  // Status information for sub-components of this component.
+  map<string, CollectorComponent> ComponentStatusMap = 4;
+}
+
 // StateResponse is the current state of Elastic Agent.
 // Next unused id: 8
 message StateResponse {
@@ -194,6 +218,9 @@ message StateResponse {
 
   // Upgrade details
   UpgradeDetails upgrade_details = 7;
+
+  // OTel collector component status information.
+  CollectorComponent collector = 8;
 }
 
 // UpgradeDetails captures the details of an ongoing Agent upgrade.

go.mod

@@ -44,6 +44,7 @@ require (
 	github.com/oklog/ulid/v2 v2.1.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/healthcheckextension v0.113.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/pprofextension v0.113.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status v0.113.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/jaegerreceiver v0.113.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver v0.113.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/zipkinreceiver v0.113.0
@@ -62,6 +63,7 @@ require (
 	go.elastic.co/apm/v2 v2.6.0
 	go.elastic.co/ecszap v1.0.2
 	go.elastic.co/go-licence-detector v0.7.0
+	go.opentelemetry.io/collector/component/componentstatus v0.113.0
 	go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.113.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/crypto v0.28.0
@@ -109,7 +111,6 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kubeletstatsreceiver v0.113.0
 	go.opentelemetry.io/collector/component v0.113.0
 	go.opentelemetry.io/collector/confmap v1.19.0
-	go.opentelemetry.io/collector/confmap/converter/expandconverter v0.113.0
 	go.opentelemetry.io/collector/confmap/provider/envprovider v1.19.0
 	go.opentelemetry.io/collector/confmap/provider/fileprovider v1.19.0
 	go.opentelemetry.io/collector/confmap/provider/httpprovider v1.19.0
@@ -497,7 +498,6 @@ require (
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/collector v0.113.0 // indirect
 	go.opentelemetry.io/collector/client v1.19.0 // indirect
-	go.opentelemetry.io/collector/component/componentstatus v0.113.0 // indirect
 	go.opentelemetry.io/collector/config/configauth v0.113.0 // indirect
 	go.opentelemetry.io/collector/config/configcompression v1.19.0 // indirect
 	go.opentelemetry.io/collector/config/configgrpc v0.113.0 // indirect

go.sum

@@ -1153,6 +1153,8 @@ github.com/open-telemetry/opentelemetry-collector-contrib/pkg/sampling v0.113.0
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/sampling v0.113.0/go.mod h1:X25Nhlw6xhuNSd/C0FeEwmD4PGmcXDl7pa2jR0UREkU=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza v0.113.0 h1:G8w+wg4nnqBqe297fBWnjJ5Tg2OYDVEMsdWA9/3ozxQ=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza v0.113.0/go.mod h1:m3hDVsXPQzQfeji3+hn7NYJPHDRlHhQRNd5T7N5wZqc=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status v0.113.0 h1:3cHaq0xbqzoEBxcHDl8h0YpUZ1W3St7UG5YQ8f9qCxw=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status v0.113.0/go.mod h1:aJlolKULr8dNC4PlPkqpnBYGHrbanp4+cODG/8V/GW8=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.113.0 h1:mFYOvag34kGXceVj29k0ZpBUyjEX7VZq+KctUSfNiG0=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.113.0/go.mod h1:54P38b2i1CgHvZLxD3EAzVccqreamGEz2U4pqy9DuHw=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheus v0.113.0 h1:vKtNSM3VQBTJx1ecf+I1iqn4kj7fKif1SpBLQ+numf8=
@@ -1519,8 +1521,6 @@ go.opentelemetry.io/collector/config/internal v0.113.0 h1:9RAzH8v7ItFT1npHpvP0Sv
 go.opentelemetry.io/collector/config/internal v0.113.0/go.mod h1:yC7E4h1Uj0SubxcFImh6OvBHFTjMh99+A5PuyIgDWqc=
 go.opentelemetry.io/collector/confmap v1.19.0 h1:TQ0lZpAKqgsE0EKk+u4JA+uBbPYeFRmWP3GH43w40CY=
 go.opentelemetry.io/collector/confmap v1.19.0/go.mod h1:GgNu1ElPGmLn9govqIfjaopvdspw4PJ9KeDtWC4E2Q4=
-go.opentelemetry.io/collector/confmap/converter/expandconverter v0.113.0 h1:DBjWXlVzdwVbs1ZOH+k1vmoBt3TLx8NvRK8ZK3nKdmo=
-go.opentelemetry.io/collector/confmap/converter/expandconverter v0.113.0/go.mod h1:/YDWibNLalyfd0BR0V5ixiParsNCvVKkA58f3bcu/AA=
 go.opentelemetry.io/collector/confmap/provider/envprovider v1.19.0 h1:f8O/I5pVRN86Gx5mHekNx92S6fGdOS4VcooRJKWe6Bs=
 go.opentelemetry.io/collector/confmap/provider/envprovider v1.19.0/go.mod h1:AiaW5YW1LD0/WlZuc8eZuZPBH6PA9QqsiAYRX1iC6T0=
 go.opentelemetry.io/collector/confmap/provider/fileprovider v1.19.0 h1:TYwyk4ea3U+5MYcEjrzZAaonBcLlabQu8CZeB7ekAYY=

internal/pkg/agent/application/actions/handlers/handler_action_upgrade_test.go

@@ -110,7 +110,7 @@ func TestUpgradeHandler(t *testing.T) {
 				return nil, nil
 			},
 		},
-		nil, nil, nil, nil, nil, false)
+		nil, nil, nil, nil, nil, false, nil)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
 
@@ -169,7 +169,7 @@ func TestUpgradeHandlerSameVersion(t *testing.T) {
 				return nil, err
 			},
 		},
-		nil, nil, nil, nil, nil, false)
+		nil, nil, nil, nil, nil, false, nil)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
 
@@ -230,7 +230,7 @@ func TestUpgradeHandlerNewVersion(t *testing.T) {
 				return nil, nil
 			},
 		},
-		nil, nil, nil, nil, nil, false)
+		nil, nil, nil, nil, nil, false, nil)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
 

internal/pkg/agent/application/apm_config_modifier.go

@@ -97,7 +97,7 @@ func getAPMConfigFromMap(cfg map[string]any) (*monitoringcfg.APMConfig, error) {
 	}
 
 	monitoringConfig := new(monitoringcfg.APMConfig)
-	err = newConfigFrom.Unpack(monitoringConfig)
+	err = newConfigFrom.UnpackTo(monitoringConfig)
 	if err != nil {
 		return nil, fmt.Errorf("error unpacking apm config: %w", err)
 	}

internal/pkg/agent/application/application.go

@@ -9,10 +9,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/elastic/elastic-agent/pkg/features"
-	"github.com/elastic/elastic-agent/pkg/limits"
-	"github.com/elastic/elastic-agent/version"
-
 	"go.elastic.co/apm/v2"
 
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -28,10 +24,14 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/capabilities"
 	"github.com/elastic/elastic-agent/internal/pkg/composable"
 	"github.com/elastic/elastic-agent/internal/pkg/config"
+	otelmanager "github.com/elastic/elastic-agent/internal/pkg/otel/manager"
 	"github.com/elastic/elastic-agent/internal/pkg/release"
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/component/runtime"
 	"github.com/elastic/elastic-agent/pkg/core/logger"
+	"github.com/elastic/elastic-agent/pkg/features"
+	"github.com/elastic/elastic-agent/pkg/limits"
+	"github.com/elastic/elastic-agent/version"
 )
 
 // New creates a new Agent and bootstrap the required subsystem.
@@ -176,13 +176,13 @@ func New(
 		}
 	}
 
-	// no need for vars in otel mode
 	varsManager, err := composable.New(log, rawConfig, composableManaged)
 	if err != nil {
 		return nil, nil, nil, errors.New(err, "failed to initialize composable controller")
 	}
 
-	coord := coordinator.New(log, cfg, logLevel, agentInfo, specs, reexec, upgrader, runtime, configMgr, varsManager, caps, monitor, isManaged, compModifiers...)
+	otelManager := otelmanager.NewOTelManager(log.Named("otel_manager"))
+	coord := coordinator.New(log, cfg, logLevel, agentInfo, specs, reexec, upgrader, runtime, configMgr, varsManager, caps, monitor, isManaged, otelManager, compModifiers...)
 	if managed != nil {
 		// the coordinator requires the config manager as well as in managed-mode the config manager requires the
 		// coordinator, so it must be set here once the coordinator is created

internal/pkg/agent/application/config_test.go

@@ -24,7 +24,7 @@ func testMgmtMode(t *testing.T) {
 	t.Run("succeed when local mode is selected", func(t *testing.T) {
 		c := mustWithConfigMode(true)
 		m := localConfig{}
-		err := c.Unpack(&m)
+		err := c.UnpackTo(&m)
 		require.NoError(t, err)
 		assert.Equal(t, false, m.Fleet.Enabled)
 		assert.Equal(t, true, configuration.IsStandalone(m.Fleet))
@@ -34,7 +34,7 @@ func testMgmtMode(t *testing.T) {
 	t.Run("succeed when fleet mode is selected", func(t *testing.T) {
 		c := mustWithConfigMode(false)
 		m := localConfig{}
-		err := c.Unpack(&m)
+		err := c.UnpackTo(&m)
 		require.NoError(t, err)
 		assert.Equal(t, true, m.Fleet.Enabled)
 		assert.Equal(t, false, configuration.IsStandalone(m.Fleet))
@@ -49,15 +49,15 @@ func testLocalConfig(t *testing.T) {
 		})
 
 		m := configuration.ReloadConfig{}
-		err := c.Unpack(&m)
+		err := c.UnpackTo(&m)
 		assert.Error(t, err)
 
 		c = config.MustNewConfigFrom(map[string]interface{}{
 			"enabled": true,
 			"period":  1,
 		})
 
-		err = c.Unpack(&m)
+		err = c.UnpackTo(&m)
 		assert.NoError(t, err)
 		assert.Equal(t, 1*time.Second, m.Period)
 	})