[cmd/opampsupervisor] Persist collector remote config & telemetry set… (

open-telemetry#30807)

…tings

**Description:** <Describe what has changed.>
Persists the configs contained in the AgentRemoteConfig and
ConnectionSettings messages to be used on startup before any messages
have been received from the OpAMP server. If the capability
corresponding to the message for the config is disabled, then the config
is not merged into the Collector's effective configuration

**Link to tracking Issue:** Part of
open-telemetry#21078

**Testing:**

Added e2e test

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

.chloggen/supervisor-persists-remote-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: Persist collector remote config & telemetry settings
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [21078]
+
+# (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

@@ -73,11 +73,13 @@ type testingOpAMPServer struct {
 	addr                string
 	supervisorConnected chan bool
 	sendToSupervisor    func(*protobufs.ServerToAgent)
+	shutdown            func()
 }
 
 func newOpAMPServer(t *testing.T, connectingCallback onConnectingFuncFactory, callbacks server.ConnectionCallbacksStruct) *testingOpAMPServer {
 	var agentConn atomic.Value
 	var isAgentConnected atomic.Bool
+	var didShutdown atomic.Bool
 	connectedChan := make(chan bool)
 	s := server.New(testLogger{t: t})
 	onConnectedFunc := callbacks.OnConnectedFunc
@@ -108,10 +110,14 @@ func newOpAMPServer(t *testing.T, connectingCallback onConnectingFuncFactory, ca
 	httpSrv := httptest.NewServer(mux)
 
 	shutdown := func() {
-		t.Log("Shutting down")
-		err := s.Stop(context.Background())
-		assert.NoError(t, err)
-		httpSrv.Close()
+		if !didShutdown.Load() {
+			waitForSupervisorConnection(connectedChan, false)
+			t.Log("Shutting down")
+			err := s.Stop(context.Background())
+			assert.NoError(t, err)
+			httpSrv.Close()
+		}
+		didShutdown.Store(true)
 	}
 	send := func(msg *protobufs.ServerToAgent) {
 		if !isAgentConnected.Load() {
@@ -121,13 +127,13 @@ func newOpAMPServer(t *testing.T, connectingCallback onConnectingFuncFactory, ca
 		agentConn.Load().(types.Connection).Send(context.Background(), msg)
 	}
 	t.Cleanup(func() {
-		waitForSupervisorConnection(connectedChan, false)
 		shutdown()
 	})
 	return &testingOpAMPServer{
 		addr:                httpSrv.Listener.Addr().String(),
 		supervisorConnected: connectedChan,
 		sendToSupervisor:    send,
+		shutdown:            shutdown,
 	}
 }
 
@@ -600,3 +606,87 @@ func TestSupervisorOpAMPConnectionSettings(t *testing.T) {
 		return connectedToNewServer.Load() == true
 	}, 10*time.Second, 500*time.Millisecond, "Collector did not connect to new OpAMP server")
 }
+
+func TestSupervisorRestartsWithLastReceivedConfig(t *testing.T) {
+	// Create a temporary directory to store the test config file.
+	tempDir := t.TempDir()
+
+	var agentConfig atomic.Value
+	initialServer := 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, "persistence", map[string]string{"url": initialServer.addr, "storage_dir": tempDir})
+
+	waitForSupervisorConnection(initialServer.supervisorConnected, true)
+
+	cfg, hash, _, _ := createSimplePipelineCollectorConf(t)
+
+	initialServer.sendToSupervisor(&protobufs.ServerToAgent{
+		RemoteConfig: &protobufs.AgentRemoteConfig{
+			Config: &protobufs.AgentConfigMap{
+				ConfigMap: map[string]*protobufs.AgentConfigFile{
+					"": {Body: cfg.Bytes()},
+				},
+			},
+			ConfigHash: hash,
+		},
+	})
+
+	require.Eventually(t, func() bool {
+		// Check if the config file was written to the storage directory
+		_, err := os.Stat(path.Join(tempDir, "last_recv_remote_config.dat"))
+		return err == nil
+	}, 5*time.Second, 250*time.Millisecond, "Config file was not written to persistent storage directory")
+
+	agentConfig.Store("")
+	s.Shutdown()
+	initialServer.shutdown()
+
+	newServer := 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{}
+			},
+		})
+	defer newServer.shutdown()
+
+	s1 := newSupervisor(t, "persistence", map[string]string{"url": newServer.addr, "storage_dir": tempDir})
+	defer s1.Shutdown()
+
+	waitForSupervisorConnection(newServer.supervisorConnected, true)
+
+	newServer.sendToSupervisor(&protobufs.ServerToAgent{
+		Flags: uint64(protobufs.ServerToAgentFlags_ServerToAgentFlags_ReportFullState),
+	})
+
+	// Check that the new Supervisor instance starts with the configuration from the last received remote config
+	require.Eventually(t, func() bool {
+		loadedConfig, ok := agentConfig.Load().(string)
+		if !ok {
+			return false
+		}
+
+		return strings.Contains(loadedConfig, "filelog")
+	}, 10*time.Second, 500*time.Millisecond, "Collector was not started with the last received remote config")
+
+}

cmd/opampsupervisor/go.mod

@@ -16,6 +16,7 @@ require (
 	go.opentelemetry.io/collector/semconv v0.99.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.27.0
+	google.golang.org/protobuf v1.33.0
 )
 
 require (
@@ -31,6 +32,5 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/net v0.23.0 // indirect
 	golang.org/x/sys v0.18.0 // indirect
-	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

cmd/opampsupervisor/supervisor/config/config.go

@@ -14,6 +14,12 @@ type Supervisor struct {
 	Server       *OpAMPServer
 	Agent        *Agent
 	Capabilities *Capabilities `mapstructure:"capabilities"`
+	Storage      *Storage      `mapstructure:"storage"`
+}
+
+type Storage struct {
+	// Directory is the directory where the Supervisor will store its data.
+	Directory string `mapstructure:"directory"`
 }
 
 // Capabilities is the set of capabilities that the Supervisor supports.