[cmd/opampsupervisor] Receive and report effective config (#33462)

**Description:**
Adds the ability to receive and run remote configurations from an OpAMP
server, as well as to report the remote configuration status.

This PR is just bringing #31641 up-to-date.

**Link to tracking Issue:** Resolves #30622

**Testing:** <Describe what testing was performed and which tests were
added.>
Unit tests

**Documentation:** <Describe the documentation added.>

---------

Co-authored-by: Evan Bradley <[email protected]>
Co-authored-by: Evan Bradley <[email protected]>
Co-authored-by: Srikanth Chekuri <[email protected]>
Co-authored-by: Tigran Najaryan <[email protected]>

Author: 4 people

.chloggen/supervisor-effective-config.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: cmd/opampsupervisor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Receive and report effective config to the OpAMP server
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [30622]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

cmd/opampsupervisor/e2e_test.go

@@ -56,7 +56,7 @@ func (tl testLogger) Errorf(_ context.Context, format string, args ...any) {
 }
 
 func defaultConnectingHandler(connectionCallbacks server.ConnectionCallbacksStruct) func(request *http.Request) types.ConnectionResponse {
-	return func(request *http.Request) types.ConnectionResponse {
+	return func(_ *http.Request) types.ConnectionResponse {
 		return types.ConnectionResponse{
 			Accept:              true,
 			ConnectionCallbacks: connectionCallbacks,
@@ -125,7 +125,8 @@ func newOpAMPServer(t *testing.T, connectingCallback onConnectingFuncFactory, ca
 			require.Fail(t, "Agent connection has not been established")
 		}
 
-		agentConn.Load().(types.Connection).Send(context.Background(), msg)
+		err = agentConn.Load().(types.Connection).Send(context.Background(), msg)
+		require.NoError(t, err)
 	}
 	t.Cleanup(func() {
 		shutdown()
@@ -217,8 +218,9 @@ func TestSupervisorStartsCollectorWithRemoteConfig(t *testing.T) {
 		cfg, ok := agentConfig.Load().(string)
 		if ok {
 			// The effective config may be structurally different compared to what was sent,
-			// so just check that it includes some strings we know to be unique to the remote config.
-			return strings.Contains(cfg, inputFile.Name()) && strings.Contains(cfg, outputFile.Name())
+			// and will also have some data redacted,
+			// so just check that it includes the filelog receiver
+			return strings.Contains(cfg, "filelog")
 		}
 
 		return false
@@ -340,9 +342,9 @@ func TestSupervisorConfiguresCapabilities(t *testing.T) {
 	waitForSupervisorConnection(server.supervisorConnected, true)
 
 	require.Eventually(t, func() bool {
-		cap := capabilities.Load()
+		caps := capabilities.Load()
 
-		return cap == uint64(protobufs.AgentCapabilities_AgentCapabilities_ReportsStatus)
+		return caps == uint64(protobufs.AgentCapabilities_AgentCapabilities_ReportsStatus)
 	}, 5*time.Second, 250*time.Millisecond)
 }
 
@@ -418,6 +420,82 @@ func TestSupervisorBootstrapsCollector(t *testing.T) {
 	}, 5*time.Second, 250*time.Millisecond)
 }
 
+func TestSupervisorReportsEffectiveConfig(t *testing.T) {
+	var agentConfig atomic.Value
+	server := newOpAMPServer(
+		t,
+		defaultConnectingHandler,
+		server.ConnectionCallbacksStruct{
+			OnMessageFunc: func(_ context.Context, _ types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent {
+				if message.EffectiveConfig != nil {
+					config := message.EffectiveConfig.ConfigMap.ConfigMap[""]
+					if config != nil {
+						agentConfig.Store(string(config.Body))
+					}
+				}
+
+				return &protobufs.ServerToAgent{}
+			},
+		})
+
+	s := newSupervisor(t, "basic", map[string]string{"url": server.addr})
+	defer s.Shutdown()
+
+	waitForSupervisorConnection(server.supervisorConnected, true)
+
+	// Create input and output files so we can "communicate" with a Collector binary.
+	// The testing package will automatically clean these up after each test.
+	tempDir := t.TempDir()
+	testKeyFile, err := os.CreateTemp(tempDir, "confKey")
+	require.NoError(t, err)
+	n, err := testKeyFile.Write([]byte(testKeyFile.Name()))
+	require.NoError(t, err)
+	require.NotZero(t, n)
+
+	colCfgTpl, err := os.ReadFile(path.Join("testdata", "collector", "split_config.yaml"))
+	require.NoError(t, err)
+
+	templ, err := template.New("").Parse(string(colCfgTpl))
+	require.NoError(t, err)
+
+	var cfg bytes.Buffer
+	err = templ.Execute(
+		&cfg,
+		map[string]string{
+			"TestKeyFile": testKeyFile.Name(),
+		},
+	)
+	require.NoError(t, err)
+
+	h := sha256.New()
+	if _, err := io.Copy(h, bytes.NewBuffer(cfg.Bytes())); err != nil {
+		t.Fatal(err)
+	}
+
+	server.sendToSupervisor(&protobufs.ServerToAgent{
+		RemoteConfig: &protobufs.AgentRemoteConfig{
+			Config: &protobufs.AgentConfigMap{
+				ConfigMap: map[string]*protobufs.AgentConfigFile{
+					"": {Body: cfg.Bytes()},
+				},
+			},
+			ConfigHash: h.Sum(nil),
+		},
+	})
+
+	require.Eventually(t, func() bool {
+		cfg, ok := agentConfig.Load().(string)
+		if ok {
+			// The effective config may be structurally different compared to what was sent,
+			// and currently has most values redacted,
+			// so just check that it includes some strings we know to be unique to the remote config.
+			return strings.Contains(cfg, "test_key:")
+		}
+
+		return false
+	}, 5*time.Second, 500*time.Millisecond, "Collector never reported effective config")
+}
+
 func TestSupervisorAgentDescriptionConfigApplies(t *testing.T) {
 	// Load the Supervisor config so we can get the location of
 	// the Collector that will be run.
@@ -673,7 +751,7 @@ func TestSupervisorOpAMPConnectionSettings(t *testing.T) {
 			OnConnectedFunc: func(_ context.Context, _ types.Connection) {
 				connectedToNewServer.Store(true)
 			},
-			OnMessageFunc: func(_ context.Context, _ types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent {
+			OnMessageFunc: func(_ context.Context, _ types.Connection, _ *protobufs.AgentToServer) *protobufs.ServerToAgent {
 				return &protobufs.ServerToAgent{}
 			},
 		})

cmd/opampsupervisor/examples/supervisor_darwin.yaml

@@ -3,7 +3,7 @@ server:
   tls:
     # Disable verification to test locally.
     # Don't do this in production.
-    insecure_skip_verify: true 
+    insecure_skip_verify: true
     # For more TLS settings see config/configtls.ClientConfig
 
 capabilities:
@@ -15,3 +15,6 @@ capabilities:
 
 agent:
   executable: ../../bin/otelcontribcol_darwin_amd64
+
+storage:
+  directory: .

cmd/opampsupervisor/examples/supervisor_linux.yaml

@@ -3,7 +3,7 @@ server:
   tls:
     # Disable verification to test locally.
     # Don't do this in production.
-    insecure_skip_verify: true 
+    insecure_skip_verify: true
     # For more TLS settings see config/configtls.ClientConfig
 
 capabilities:
@@ -15,3 +15,6 @@ capabilities:
 
 agent:
   executable: ../../bin/otelcontribcol_linux_amd64
+
+storage:
+  directory: .

cmd/opampsupervisor/examples/supervisor_windows.yaml

@@ -3,7 +3,7 @@ server:
   tls:
     # Disable verification to test locally.
     # Don't do this in production.
-    insecure_skip_verify: true 
+    insecure_skip_verify: true
     # For more TLS settings see config/configtls.ClientConfig
 
 capabilities:
@@ -15,3 +15,6 @@ capabilities:
 
 agent:
   executable: ../../bin/otelcontribcol_windows_amd64.exe
+
+storage:
+  directory: .

cmd/opampsupervisor/go.mod

@@ -5,6 +5,7 @@ go 1.21.0
 require (
 	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/google/uuid v1.6.0
+	github.com/knadh/koanf/maps v0.1.1
 	github.com/knadh/koanf/parsers/yaml v0.1.0
 	github.com/knadh/koanf/providers/file v0.1.0
 	github.com/knadh/koanf/providers/rawbytes v0.1.0
@@ -26,7 +27,6 @@ require (
 	github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/gorilla/websocket v1.5.1 // indirect
-	github.com/knadh/koanf/maps v0.1.1 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect

cmd/opampsupervisor/supervisor/pid.go

@@ -0,0 +1,17 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package supervisor
+
+import "os"
+
+// pidProvider provides the PID of the current process
+type pidProvider interface {
+	PID() int
+}
+
+type defaultPIDProvider struct{}
+
+func (defaultPIDProvider) PID() int {
+	return os.Getpid()
+}