[cmd/opampsupervisor] chore: Store RemoteConfigStatus in persistent state (#40467)

Author: dpaasman00

.chloggen/supervisor-persistently-stores-remotecfgstatus.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: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: opampsupervisor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Supervisor now persists the remote config status to disk. This allows more accurate reporting of the remote config status. Also reports healthy status when not running the agent due to empty config (previous performance optimization).
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40467]
+
+# (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

@@ -457,7 +457,6 @@ func TestSupervisorStartsCollectorWithRemoteConfigAndExecParams(t *testing.T) {
 
 	// create server
 	var agentConfig atomic.Value
-	var remoteConfigStatus atomic.Value
 	server := newOpAMPServer(
 		t,
 		defaultConnectingHandler,
@@ -469,9 +468,6 @@ func TestSupervisorStartsCollectorWithRemoteConfigAndExecParams(t *testing.T) {
 						agentConfig.Store(string(config.Body))
 					}
 				}
-				if message.RemoteConfigStatus != nil {
-					remoteConfigStatus.Store(message.RemoteConfigStatus)
-				}
 				return &protobufs.ServerToAgent{}
 			},
 		})
@@ -502,12 +498,6 @@ func TestSupervisorStartsCollectorWithRemoteConfigAndExecParams(t *testing.T) {
 
 	waitForSupervisorConnection(server.supervisorConnected, true)
 
-	// verify that the supervisor reports the remote config as applied and the correct hash
-	require.Eventually(t, func() bool {
-		status := remoteConfigStatus.Load().(*protobufs.RemoteConfigStatus)
-		return status != nil && status.Status == protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLIED && bytes.Equal(status.LastRemoteConfigHash, hash)
-	}, 20*time.Second, 500*time.Millisecond, "Remote config status never became applied")
-
 	for _, port := range []int{healthcheckPort, secondHealthcheckPort} {
 		require.Eventually(t, func() bool {
 			resp, err := http.DefaultClient.Get(fmt.Sprintf("http://localhost:%d", port))
@@ -1364,6 +1354,7 @@ func TestSupervisorRestartsWithLastReceivedConfig(t *testing.T) {
 	tempDir := t.TempDir()
 
 	var agentConfig atomic.Value
+	var initialRemoteConfigStatus atomic.Value
 	initialServer := newOpAMPServer(
 		t,
 		defaultConnectingHandler,
@@ -1375,6 +1366,9 @@ func TestSupervisorRestartsWithLastReceivedConfig(t *testing.T) {
 						agentConfig.Store(string(config.Body))
 					}
 				}
+				if message.RemoteConfigStatus != nil {
+					initialRemoteConfigStatus.Store(message.RemoteConfigStatus)
+				}
 				return &protobufs.ServerToAgent{}
 			},
 		})
@@ -1385,16 +1379,16 @@ func TestSupervisorRestartsWithLastReceivedConfig(t *testing.T) {
 
 	waitForSupervisorConnection(initialServer.supervisorConnected, true)
 
-	cfg, hash, _, _ := createSimplePipelineCollectorConf(t)
+	simplePipelineCFG, simplePipelineHash, _, _ := createSimplePipelineCollectorConf(t)
 
 	initialServer.sendToSupervisor(&protobufs.ServerToAgent{
 		RemoteConfig: &protobufs.AgentRemoteConfig{
 			Config: &protobufs.AgentConfigMap{
 				ConfigMap: map[string]*protobufs.AgentConfigFile{
-					"": {Body: cfg.Bytes()},
+					"": {Body: simplePipelineCFG.Bytes()},
 				},
 			},
-			ConfigHash: hash,
+			ConfigHash: simplePipelineHash,
 		},
 	})
 
@@ -1404,10 +1398,17 @@ func TestSupervisorRestartsWithLastReceivedConfig(t *testing.T) {
 		return err == nil
 	}, 5*time.Second, 250*time.Millisecond, "Config file was not written to persistent storage directory")
 
+	// wait for remote config status to be applied
+	require.Eventually(t, func() bool {
+		status := initialRemoteConfigStatus.Load().(*protobufs.RemoteConfigStatus)
+		return status != nil && status.Status == protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLIED && bytes.Equal(status.LastRemoteConfigHash, simplePipelineHash)
+	}, 20*time.Second, 500*time.Millisecond, "Initial remote config never became applied")
+
 	agentConfig.Store("")
 	s.Shutdown()
 	initialServer.shutdown()
 
+	var remoteConfigStatus atomic.Value
 	newServer := newOpAMPServer(
 		t,
 		defaultConnectingHandler,
@@ -1419,6 +1420,9 @@ func TestSupervisorRestartsWithLastReceivedConfig(t *testing.T) {
 						agentConfig.Store(string(config.Body))
 					}
 				}
+				if message.RemoteConfigStatus != nil {
+					remoteConfigStatus.Store(message.RemoteConfigStatus)
+				}
 				return &protobufs.ServerToAgent{}
 			},
 		})
@@ -1435,6 +1439,12 @@ func TestSupervisorRestartsWithLastReceivedConfig(t *testing.T) {
 		Flags: uint64(protobufs.ServerToAgentFlags_ServerToAgentFlags_ReportFullState),
 	})
 
+	// verify that the supervisor reports the existing remote config status
+	require.Eventually(t, func() bool {
+		status := remoteConfigStatus.Load().(*protobufs.RemoteConfigStatus)
+		return status != nil && status.Status == protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLIED && bytes.Equal(status.LastRemoteConfigHash, simplePipelineHash)
+	}, 20*time.Second, 500*time.Millisecond, "Remote config status never became applied")
+
 	// 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)
@@ -1635,6 +1645,8 @@ func TestSupervisorWritesAgentFilesToStorageDir(t *testing.T) {
 
 func TestSupervisorStopsAgentProcessWithEmptyConfigMap(t *testing.T) {
 	agentCfgChan := make(chan string, 1)
+	var healthReport atomic.Value
+	var remoteConfigStatus atomic.Value
 	server := newOpAMPServer(
 		t,
 		defaultConnectingHandler,
@@ -1649,7 +1661,12 @@ func TestSupervisorStopsAgentProcessWithEmptyConfigMap(t *testing.T) {
 						}
 					}
 				}
-
+				if message.Health != nil {
+					healthReport.Store(message.Health)
+				}
+				if message.RemoteConfigStatus != nil {
+					remoteConfigStatus.Store(message.RemoteConfigStatus)
+				}
 				return &protobufs.ServerToAgent{}
 			},
 		})
@@ -1723,6 +1740,18 @@ func TestSupervisorStopsAgentProcessWithEmptyConfigMap(t *testing.T) {
 			assert.ErrorContains(tt, err, "No connection could be made")
 		}
 	}, 3*time.Second, 250*time.Millisecond)
+
+	// Verify we have a healthy status (if it was ran with the empty config it would be healthy)
+	require.Eventually(t, func() bool {
+		health, ok := healthReport.Load().(*protobufs.ComponentHealth)
+		return ok && health.Healthy
+	}, 3*time.Second, 250*time.Millisecond)
+
+	// Verify the status is set to APPLIED (if it was ran with the empty config it would be APPLIED)
+	require.Eventually(t, func() bool {
+		status, ok := remoteConfigStatus.Load().(*protobufs.RemoteConfigStatus)
+		return ok && status.Status == protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLIED
+	}, 3*time.Second, 250*time.Millisecond)
 }
 
 type logEntry struct {

cmd/opampsupervisor/opampsupervisor

@@ -4,27 +4,68 @@
 package supervisor
 
 import (
+	"encoding/hex"
 	"errors"
 	"os"
 
 	"github.com/google/uuid"
+	"github.com/open-telemetry/opamp-go/protobufs"
+	"go.uber.org/zap"
 	"gopkg.in/yaml.v3"
 )
 
 // persistentState represents persistent state for the supervisor
 type persistentState struct {
-	InstanceID uuid.UUID `yaml:"instance_id"`
+	InstanceID             uuid.UUID           `yaml:"instance_id"`
+	LastRemoteConfigStatus *RemoteConfigStatus `yaml:"last_remote_config_status"`
 
 	// Path to the config file that the state should be saved to.
 	// This is not marshaled.
-	configPath string `yaml:"-"`
+	configPath string      `yaml:"-"`
+	logger     *zap.Logger `yaml:"-"`
+}
+
+// RemoteConfigStatus is a custom struct that is used to marshal/unmarshal the remote config status.
+// LastRemoteConfigHash is a hex encoded string of the last remote config hash for human readability.
+type RemoteConfigStatus struct {
+	// Status is the status of the last remote config.
+	Status protobufs.RemoteConfigStatuses `yaml:"status"`
+	// LastRemoteConfigHash is a hex encoded string of the last remote config hash for human readability.
+	LastRemoteConfigHash string `yaml:"last_remote_config_hash"`
+	// ErrorMessage is the error message of the last remote config.
+	ErrorMessage string `yaml:"error_message"`
 }
 
 func (p *persistentState) SetInstanceID(id uuid.UUID) error {
 	p.InstanceID = id
 	return p.writeState()
 }
 
+func (p *persistentState) SetLastRemoteConfigStatus(status *protobufs.RemoteConfigStatus) error {
+	p.LastRemoteConfigStatus = &RemoteConfigStatus{
+		Status:               status.Status,
+		LastRemoteConfigHash: hex.EncodeToString(status.LastRemoteConfigHash),
+		ErrorMessage:         status.ErrorMessage,
+	}
+	return p.writeState()
+}
+
+func (p *persistentState) GetLastRemoteConfigStatus() *protobufs.RemoteConfigStatus {
+	if p.LastRemoteConfigStatus == nil {
+		return nil
+	}
+	lastRemoteConfigHash, err := hex.DecodeString(p.LastRemoteConfigStatus.LastRemoteConfigHash)
+	if err != nil {
+		p.logger.Error("Failed to decode last remote config hash, returning empty status", zap.Error(err))
+		return nil
+	}
+	return &protobufs.RemoteConfigStatus{
+		Status:               p.LastRemoteConfigStatus.Status,
+		LastRemoteConfigHash: lastRemoteConfigHash,
+		ErrorMessage:         p.LastRemoteConfigStatus.ErrorMessage,
+	}
+}
+
 func (p *persistentState) writeState() error {
 	by, err := yaml.Marshal(p)
 	if err != nil {
@@ -36,19 +77,19 @@ func (p *persistentState) writeState() error {
 
 // loadOrCreatePersistentState attempts to load the persistent state from disk. If it doesn't
 // exist, a new persistent state file is created.
-func loadOrCreatePersistentState(file string) (*persistentState, error) {
-	state, err := loadPersistentState(file)
+func loadOrCreatePersistentState(file string, logger *zap.Logger) (*persistentState, error) {
+	state, err := loadPersistentState(file, logger)
 	switch {
 	case errors.Is(err, os.ErrNotExist):
-		return createNewPersistentState(file)
+		return createNewPersistentState(file, logger)
 	case err != nil:
 		return nil, err
 	default:
 		return state, nil
 	}
 }
 
-func loadPersistentState(file string) (*persistentState, error) {
+func loadPersistentState(file string, logger *zap.Logger) (*persistentState, error) {
 	var state *persistentState
 
 	by, err := os.ReadFile(file)
@@ -61,11 +102,12 @@ func loadPersistentState(file string) (*persistentState, error) {
 	}
 
 	state.configPath = file
+	state.logger = logger
 
 	return state, nil
 }
 
-func createNewPersistentState(file string) (*persistentState, error) {
+func createNewPersistentState(file string, logger *zap.Logger) (*persistentState, error) {
 	id, err := uuid.NewV7()
 	if err != nil {
 		return nil, err
@@ -74,6 +116,7 @@ func createNewPersistentState(file string) (*persistentState, error) {
 	p := &persistentState{
 		InstanceID: id,
 		configPath: file,
+		logger:     logger,
 	}
 
 	return p, p.writeState()