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

.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.