Add unit tests, fix issues found by tests, document

client/client.go

@@ -137,6 +137,11 @@ type OpAMPClient interface {
 
 	// SetAvailableComponents modifies the set of components that are available for configuration
 	// on the agent.
-	// TODO more documentation
+	// May be called any time after Start(), including from the OnMessage handler.
+	// The new components will be sent with the next message to the server.
+	// If components is nil, errReportsAvailableComponentsNotSet will be returned.
+	// If components.Hash is nil or an empty []byte, errNoAvailableComponentHash will be returned.
+	// This method is subject to agent status compression - if components is not
+	// different from the cached agent state, this method is a no-op.
 	SetAvailableComponents(components *protobufs.AvailableComponents) error
 }

client/clientimpl_test.go

@@ -2305,3 +2305,23 @@ func TestSetFlagsBeforeStart(t *testing.T) {
 		assert.NoError(t, err)
 	})
 }
+
+func generateTestAvailableComponents() *protobufs.AvailableComponents {
+	return &protobufs.AvailableComponents{
+		Hash: []byte("fake-hash"),
+		Components: map[string]*protobufs.ComponentDetails{
+			"receivers": {
+				Metadata: []*protobufs.KeyValue{
+					{
+						Key: "component",
+						Value: &protobufs.AnyValue{
+							Value: &protobufs.AnyValue_StringValue{
+								StringValue: "filereceiver",
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}

client/httpclient_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/open-telemetry/opamp-go/client/internal"
@@ -311,3 +312,112 @@ func TestRedirectHTTP(t *testing.T) {
 		})
 	}
 }
+
+func TestHTTPReportsAvailableComponents(t *testing.T) {
+	testCases := []struct {
+		desc                string
+		availableComponents *protobufs.AvailableComponents
+	}{
+		{
+			desc:                "Does not report AvailableComponents",
+			availableComponents: nil,
+		},
+		{
+			desc:                "Reports AvailableComponents",
+			availableComponents: generateTestAvailableComponents(),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			// Start a Server.
+			srv := internal.StartMockServer(t)
+			var rcvCounter atomic.Uint64
+			srv.OnMessage = func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
+				assert.EqualValues(t, rcvCounter.Load(), msg.SequenceNum)
+				rcvCounter.Add(1)
+				time.Sleep(50 * time.Millisecond)
+				if rcvCounter.Load() == 1 {
+					resp := &protobufs.ServerToAgent{
+						InstanceUid: msg.InstanceUid,
+					}
+
+					if tc.availableComponents != nil {
+						// the first message received should contain just the available component hash
+						availableComponents := msg.GetAvailableComponents()
+						require.NotNil(t, availableComponents)
+						require.Nil(t, availableComponents.GetComponents())
+						require.Equal(t, tc.availableComponents.GetHash(), availableComponents.GetHash())
+
+						// add the flag asking for the full available component state to the response
+						resp.Flags = uint64(protobufs.ServerToAgentFlags_ServerToAgentFlags_ReportAvailableComponents)
+					} else {
+						require.Nil(t, msg.GetAvailableComponents())
+					}
+
+					return resp
+				}
+
+				if rcvCounter.Load() == 2 {
+					if tc.availableComponents != nil {
+						// the second message received should contain the full component state
+						availableComponents := msg.GetAvailableComponents()
+						require.NotNil(t, availableComponents)
+						require.Equal(t, tc.availableComponents.GetComponents(), availableComponents.GetComponents())
+						require.Equal(t, tc.availableComponents.GetHash(), availableComponents.GetHash())
+					} else {
+						require.Nil(t, msg.GetAvailableComponents())
+					}
+
+					return nil
+				}
+
+				// all subsequent messages should not have any available components
+				require.Nil(t, msg.GetAvailableComponents())
+				return nil
+			}
+
+			// Start a client.
+			settings := types.StartSettings{}
+			settings.OpAMPServerURL = "http://" + srv.Endpoint
+			if tc.availableComponents != nil {
+				settings.Capabilities = protobufs.AgentCapabilities_AgentCapabilities_ReportsAvailableComponents
+				settings.AvailableComponents = tc.availableComponents
+			}
+
+			client := NewHTTP(nil)
+			prepareClient(t, &settings, client)
+
+			assert.NoError(t, client.Start(context.Background(), settings))
+
+			// Verify that status report is delivered.
+			eventually(t, func() bool {
+				return rcvCounter.Load() == 1
+			})
+
+			if tc.availableComponents != nil {
+				// Verify that status report is delivered again. Polling should ensure this.
+				eventually(t, func() bool {
+					return rcvCounter.Load() == 2
+				})
+			} else {
+				// Verify that no second status report is delivered (polling is too infrequent for this to happen in 3 seconds)
+				require.Never(t, func() bool {
+					return rcvCounter.Load() == 2
+				}, 3*time.Second, 10*time.Millisecond)
+			}
+
+			// Verify that no third status report is delivered (polling is too infrequent for this to happen in 3 seconds)
+			require.Never(t, func() bool {
+				return rcvCounter.Load() == 3
+			}, 3*time.Second, 10*time.Millisecond)
+
+			// Shutdown the Server.
+			srv.Close()
+
+			// Shutdown the client.
+			err := client.Stop(context.Background())
+			assert.NoError(t, err)
+		})
+	}
+}

client/internal/clientcommon.go

@@ -90,8 +90,14 @@
 		return ErrHealthMissing
 	}
 
-	if c.Capabilities&protobufs.AgentCapabilities_AgentCapabilities_ReportsAvailableComponents != 0 && c.ClientSyncedState.AvailableComponents() == nil {
-		return ErrAvailableComponentsMissing
+	if c.Capabilities&protobufs.AgentCapabilities_AgentCapabilities_ReportsAvailableComponents != 0 {
+		if settings.AvailableComponents == nil {
+			return ErrAvailableComponentsMissing
+		}
+
+		if err := c.ClientSyncedState.SetAvailableComponents(settings.AvailableComponents); err != nil {
+			return err
+		}
 	}
 
 	// Prepare remote config status.
@@ -220,8 +226,11 @@
 
 	// initially, do not send the full component state - just send the hash.
 	// full state is available on request from the server using the corresponding ServerToAgent flag
-	availableComponents := &protobufs.AvailableComponents{
-		Hash: c.ClientSyncedState.AvailableComponents().GetHash(),
+	var availableComponents *protobufs.AvailableComponents
+	if c.Capabilities&protobufs.AgentCapabilities_AgentCapabilities_ReportsAvailableComponents != 0 {
+		availableComponents = &protobufs.AvailableComponents{
+			Hash: c.ClientSyncedState.AvailableComponents().GetHash(),
+		}
 	}
 
 	c.sender.NextMessage().Update(
@@ -448,31 +457,31 @@
 }
 
 // SetAvailableComponents sends a message to the server with the available components for the agent
 func (c *ClientCommon) SetAvailableComponents(components *protobufs.AvailableComponents) error {
 	if c.Capabilities&protobufs.AgentCapabilities_AgentCapabilities_ReportsAvailableComponents == 0 {
 		return errReportsAvailableComponentsNotSet
 	}
 
 	if len(components.Hash) == 0 {
 		return errNoAvailableComponentHash
 	}
 
 	// implement agent status compression, don't send the message if it hasn't changed from the previous message
 	availableComponentsChanged := !proto.Equal(c.ClientSyncedState.AvailableComponents(), components)
 
 	if availableComponentsChanged {
 		if err := c.ClientSyncedState.SetAvailableComponents(components); err != nil {
 			return err
 		}
 
 		c.sender.NextMessage().Update(
 			func(msg *protobufs.AgentToServer) {
 				msg.AvailableComponents = c.ClientSyncedState.AvailableComponents()
 			},
 		)
 
 		c.sender.ScheduleSend()
 	}
 
 	return nil
 }

client/types/startsettings.go

@@ -64,4 +64,9 @@ type StartSettings struct {
 	//
 	// If the ReportsHeartbeat capability is disabled, this option has no effect.
 	HeartbeatInterval *time.Duration
+
+	// Defines the available components of the Agent.
+	// Required if the ReportsAvailableComponents capability is set.
+	// If the ReportsAvailableComponents capability is not set, this option has no effect.
+	AvailableComponents *protobufs.AvailableComponents
 }

client/wsclient_test.go

@@ -727,3 +727,96 @@ func TestHandlesConnectionError(t *testing.T) {
 	err = client.Stop(context.Background())
 	require.NoError(t, err)
 }
+
+func TestWSSenderReportsAvailableComponents(t *testing.T) {
+	testCases := []struct {
+		desc                string
+		availableComponents *protobufs.AvailableComponents
+	}{
+		{
+			desc:                "Does not report AvailableComponents",
+			availableComponents: nil,
+		},
+		{
+			desc:                "Reports AvailableComponents",
+			availableComponents: generateTestAvailableComponents(),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			srv := internal.StartMockServer(t)
+
+			var firstMsg atomic.Bool
+			var conn atomic.Value
+			srv.OnWSConnect = func(c *websocket.Conn) {
+				conn.Store(c)
+				firstMsg.Store(true)
+			}
+			var msgCount atomic.Int64
+			srv.OnMessage = func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
+				if firstMsg.Load() {
+					msgCount.Add(1)
+					firstMsg.Store(false)
+					resp := &protobufs.ServerToAgent{
+						InstanceUid: msg.InstanceUid,
+					}
+
+					if tc.availableComponents != nil {
+						availableComponents := msg.GetAvailableComponents()
+						require.NotNil(t, availableComponents)
+						require.Nil(t, availableComponents.GetComponents())
+						require.Equal(t, tc.availableComponents.GetHash(), availableComponents.GetHash())
+
+						resp.Flags = uint64(protobufs.ServerToAgentFlags_ServerToAgentFlags_ReportAvailableComponents)
+					} else {
+						require.Nil(t, msg.GetAvailableComponents())
+					}
+
+					return resp
+				}
+				msgCount.Add(1)
+				if tc.availableComponents != nil {
+					availableComponents := msg.GetAvailableComponents()
+					require.NotNil(t, availableComponents)
+					require.Equal(t, tc.availableComponents.GetHash(), availableComponents.GetHash())
+					require.Equal(t, tc.availableComponents.GetComponents(), availableComponents.GetComponents())
+				} else {
+					require.Error(t, errors.New("should not receive a second message when ReportsAvailableComponents is disabled"))
+				}
+
+				return nil
+			}
+
+			// Start an OpAMP/WebSocket client.
+			settings := types.StartSettings{
+				OpAMPServerURL: "ws://" + srv.Endpoint,
+			}
+			client := NewWebSocket(nil)
+
+			if tc.availableComponents != nil {
+				settings.Capabilities = protobufs.AgentCapabilities_AgentCapabilities_ReportsAvailableComponents
+				settings.AvailableComponents = tc.availableComponents
+			}
+
+			startClient(t, settings, client)
+
+			// Wait for connection to be established.
+			eventually(t, func() bool { return conn.Load() != nil })
+
+			if tc.availableComponents != nil {
+				assert.Eventually(t, func() bool {
+					return msgCount.Load() >= 2
+				}, 5*time.Second, 10*time.Millisecond)
+			} else {
+				assert.Never(t, func() bool {
+					return msgCount.Load() >= 2
+				}, 3*time.Second, 10*time.Millisecond)
+			}
+
+			// Stop the client.
+			err := client.Stop(context.Background())
+			assert.NoError(t, err)
+		})
+	}
+}