Repository: open-telemetry/opamp-go
Pull request #114 (2022-07-28 19:07:47)

Use ReportsRemoteConfig capability
Implements spec change https://github.com/open-telemetry/opamp-spec/pull/115

codecov[bot]: # [Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/114?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) Report
> Merging [#114](https://codecov.io/gh/open-telemetry/opamp-go/pull/114?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) (6f10b4c) into [main](https://codecov.io/gh/open-telemetry/opamp-go/commit/3d4cbdc975d341c1765050a9a59882ef89dda006?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) (3d4cbdc) will **decrease** coverage by `0.13%`.
> The diff coverage is `0.00%`.

```diff
@@            Coverage Diff             @@
##             main     #114      +/-   ##
==========================================
- Coverage   74.89%   74.75%   -0.14%     
==========================================
  Files          22       22              
  Lines        1629     1632       +3     
==========================================
  Hits         1220     1220              
- Misses        307      309       +2     
- Partials      102      103       +1     
```


| [Impacted Files](https://codecov.io/gh/open-telemetry/opamp-go/pull/114?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) | Coverage Δ | |
|---|---|---|
| [client/internal/clientcommon.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/114/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL2NsaWVudGNvbW1vbi5nbw==) | `78.32% <0.00%> (-1.18%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/114?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/114?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry). Last update [3d4cbdc...6f10b4c](https://codecov.io/gh/open-telemetry/opamp-go/pull/114?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry).


Patch:
diff --git a/client/httpclient.go b/client/httpclient.go
index 63a7bc1..7e1d5b5 100644
--- a/client/httpclient.go
+++ b/client/httpclient.go
@@ -42,6 +42,8 @@ func (c *httpClient) Start(ctx context.Context, settings types.StartSettings) er
 
 	c.opAMPServerURL = settings.OpAMPServerURL
 
+	// TODO: implement compression based on settings.EnableCompression value.
+
 	// Prepare Server connection settings.
 	c.sender.SetRequestHeader(settings.Header)
 
diff --git a/client/internal/clientcommon.go b/client/internal/clientcommon.go
index bc8211f..c734abf 100644
--- a/client/internal/clientcommon.go
+++ b/client/internal/clientcommon.go
@@ -16,6 +16,7 @@ var (
 	ErrAgentDescriptionNoAttributes = errors.New("AgentDescription has no attributes defined")
 	ErrAgentHealthMissing           = errors.New("AgentHealth is nil")
 	ErrReportsEffectiveConfigNotSet = errors.New("ReportsEffectiveConfig capability is not set")
+	ErrReportsRemoteConfigNotSet    = errors.New("ReportsRemoteConfig capability is not set")
 	ErrPackagesStateProviderNotSet  = errors.New("PackagesStateProvider must be set")
 	ErrAcceptsPackagesNotSet        = errors.New("AcceptsPackages and ReportsPackageStatuses must be set")
 
@@ -293,6 +294,10 @@ func (c *ClientCommon) UpdateEffectiveConfig(ctx context.Context) error {
 // It also remembers the new RemoteConfigStatus in the client state so that it can be
 // sent to the Server when the Server asks for it.
 func (c *ClientCommon) SetRemoteConfigStatus(status *protobufs.RemoteConfigStatus) error {
+	if c.Capabilities&protobufs.AgentCapabilities_ReportsRemoteConfig == 0 {
+		return ErrReportsRemoteConfigNotSet
+	}
+
 	if status.LastRemoteConfigHash == nil {
 		return errLastRemoteConfigHashNil
 	}
diff --git a/client/internal/mockserver.go b/client/internal/mockserver.go
index 274cea1..10b47cd 100644
--- a/client/internal/mockserver.go
+++ b/client/internal/mockserver.go
@@ -28,16 +28,15 @@ type MockServer struct {
 	OnMessage   func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent
 	srv         *httptest.Server
 
-	expectedHandlers chan receivedMessageHandler
-	expectedComplete chan struct{}
-	isExpectMode     bool
+	expectedHandlers  chan receivedMessageHandler
+	expectedComplete  chan struct{}
+	isExpectMode      bool
+	enableCompression bool
 }
 
 const headerContentType = "Content-Type"
 const contentTypeProtobuf = "application/x-protobuf"
 
-var upgrader = websocket.Upgrader{}
-
 func StartMockServer(t *testing.T) *MockServer {
 	srv := &MockServer{
 		t:                t,
@@ -101,7 +100,15 @@ func (m *MockServer) handlePlainHttp(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (m *MockServer) EnableCompression() {
+	m.enableCompression = true
+}
+
 func (m *MockServer) handleWebSocket(t *testing.T, w http.ResponseWriter, r *http.Request) {
+	var upgrader = websocket.Upgrader{
+		EnableCompression: m.enableCompression,
+	}
+
 	conn, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
 		return
diff --git a/client/internal/sender.go b/client/internal/sender.go
index 28a283d..f418b4c 100644
--- a/client/internal/sender.go
+++ b/client/internal/sender.go
@@ -3,6 +3,7 @@ package internal
 import (
 	"errors"
 
+	"github.com/oklog/ulid/v2"
 	"github.com/open-telemetry/opamp-go/protobufs"
 )
 
@@ -69,6 +70,10 @@ func (h *SenderCommon) SetInstanceUid(instanceUid string) error {
 		return errors.New("cannot set instance uid to empty value")
 	}
 
+	if _, err := ulid.ParseStrict(instanceUid); err != nil {
+		return err
+	}
+
 	h.nextMessage.Update(
 		func(msg *protobufs.AgentToServer) {
 			msg.InstanceUid = instanceUid
diff --git a/client/internal/tcpproxy.go b/client/internal/tcpproxy.go
new file mode 100644
index 0000000..c371ab8
--- /dev/null
+++ b/client/internal/tcpproxy.go
@@ -0,0 +1,113 @@
+package internal
+
+import (
+	"log"
+	"net"
+	"sync/atomic"
+)
+
+// TCPProxy is used for intercepting WebSocket connections and counting
+// the number of bytes transferred.
+type TCPProxy struct {
+	// Destination endpoint to connect to.
+	destHostPort string
+	// Incoming endpoint to accept connections on.
+	incomingHostPort string
+
+	stopSignal chan struct{}
+
+	// Byte counters in both directions.
+	clientToServerBytes int64
+	serverToClientBytes int64
+}
+
+func NewProxy(destHostPort string) *TCPProxy {
+	return &TCPProxy{destHostPort: destHostPort}
+}
+
+func (p *TCPProxy) Start() error {
+	// Begin listening on an available TCP port.
+	ln, err := net.Listen("tcp", "127.0.0.1:")
+	if err != nil {
+		return err
+	}
+
+	// Remember the port that we listen on.
+	p.incomingHostPort = ln.Addr().String()
+
+	p.stopSignal = make(chan struct{})
+
+	go func() {
+		for {
+			select {
+			case <-p.stopSignal:
+				ln.Close()
+				return
+			default:
+			}
+			conn, err := ln.Accept()
+			if err != nil {
+				log.Printf("Failed to Accept TCP connection: %v\n", err.Error())
+				return
+			}
+			go p.forwardBothWays(conn)
+		}
+	}()
+
+	return nil
+}
+
+func (p *TCPProxy) Stop() {
+	close(p.stopSignal)
+}
+
+func (p *TCPProxy) IncomingEndpoint() string {
+	return p.incomingHostPort
+}
+
+func (p *TCPProxy) forwardBothWays(in net.Conn) {
+	// We have an incoming connection. Establish an outgoing connection
+	// to the destination endpoint.
+	out, err := net.Dial("tcp", p.destHostPort)
+	if err != nil {
+		return
+	}
+
+	defer out.Close()
+	defer in.Close()
+
+	// Forward TCP stream bytes from incoming to outgoing connection direction.
+	go p.forwardConn(in, out, &p.clientToServerBytes)
+
+	// Forward TCP stream bytes in the reverse direction.
+	p.forwardConn(out, in, &p.serverToClientBytes)
+}
+
+func (p *TCPProxy) ServerToClientBytes() int {
+	return int(atomic.LoadInt64(&p.serverToClientBytes))
+}
+
+func (p *TCPProxy) ClientToServerBytes() int {
+	return int(atomic.LoadInt64(&p.clientToServerBytes))
+}
+
+func (p *TCPProxy) forwardConn(in, out net.Conn, byteCounter *int64) {
+	buf := make([]byte, 1024)
+	for {
+		select {
+		case <-p.stopSignal:
+			return
+		default:
+		}
+
+		n, err := in.Read(buf)
+		if err != nil {
+			break
+		}
+		n, err = out.Write(buf[:n])
+		if err != nil {
+			break
+		}
+		atomic.AddInt64(byteCounter, int64(n))
+	}
+}
diff --git a/client/types/startsettings.go b/client/types/startsettings.go
index 8a2e4c2..30e2003 100644
--- a/client/types/startsettings.go
+++ b/client/types/startsettings.go
@@ -42,4 +42,9 @@ type StartSettings struct {
 	// Defines the capabilities of the Agent. AgentCapabilities_ReportsStatus bit does not need to
 	// be set in this field, it will be set automatically since it is required by OpAMP protocol.
 	Capabilities protobufs.AgentCapabilities
+
+	// EnableCompression can be set to true to enable the compression. Note that for WebSocket transport
+	// the compression is only effectively enabled if the Server also supports compression.
+	// The data will be compressed in both directions.
+	EnableCompression bool
 }
diff --git a/client/wsclient.go b/client/wsclient.go
index 1ac0de3..d07f7bd 100644
--- a/client/wsclient.go
+++ b/client/wsclient.go
@@ -65,6 +65,8 @@ func (c *wsClient) Start(ctx context.Context, settings types.StartSettings) erro
 		return err
 	}
 
+	c.dialer.EnableCompression = settings.EnableCompression
+
 	if settings.TLSConfig != nil {
 		c.url.Scheme = "wss"
 	}
diff --git a/internal/examples/agent/agent/agent.go b/internal/examples/agent/agent/agent.go
index fbfc274..65987a5 100644
--- a/internal/examples/agent/agent/agent.go
+++ b/internal/examples/agent/agent/agent.go
@@ -104,6 +104,7 @@ func (agent *Agent) start() error {
 		},
 		RemoteConfigStatus: agent.remoteConfigStatus,
 		Capabilities: protobufs.AgentCapabilities_AcceptsRemoteConfig |
+			protobufs.AgentCapabilities_ReportsRemoteConfig |
 			protobufs.AgentCapabilities_ReportsEffectiveConfig |
 			protobufs.AgentCapabilities_ReportsOwnMetrics,
 	}
diff --git a/internal/examples/server/data/agent.go b/internal/examples/server/data/agent.go
index f684632..cf53f37 100644
--- a/internal/examples/server/data/agent.go
+++ b/internal/examples/server/data/agent.go
@@ -211,7 +211,7 @@ func (agent *Agent) processStatusUpdate(
 	// Check if any fields were omitted in the status report.
 	effectiveConfigOmitted := agent.hasCapability(protobufs.AgentCapabilities_ReportsEffectiveConfig) && newStatus.EffectiveConfig == nil
 	packageStatusesOmitted := agent.hasCapability(protobufs.AgentCapabilities_ReportsPackageStatuses) && newStatus.PackageStatuses == nil
-	remoteConfigStatusOmitted := agent.hasCapability(protobufs.AgentCapabilities_AcceptsRemoteConfig) && newStatus.RemoteConfigStatus == nil
+	remoteConfigStatusOmitted := agent.hasCapability(protobufs.AgentCapabilities_ReportsRemoteConfig) && newStatus.RemoteConfigStatus == nil
 	healthOmitted := agent.hasCapability(protobufs.AgentCapabilities_ReportsHealth) && newStatus.Health == nil
 
 	// True if the status was not fully reported.
diff --git a/internal/examples/supervisor/supervisor/supervisor.go b/internal/examples/supervisor/supervisor/supervisor.go
index 161e1a4..0b8f630 100644
--- a/internal/examples/supervisor/supervisor/supervisor.go
+++ b/internal/examples/supervisor/supervisor/supervisor.go
@@ -152,6 +152,7 @@ func (s *Supervisor) startOpAMP() error {
 			OnMessageFunc: s.onMessage,
 		},
 		Capabilities: protobufs.AgentCapabilities_AcceptsRemoteConfig |
+			protobufs.AgentCapabilities_ReportsRemoteConfig |
 			protobufs.AgentCapabilities_ReportsEffectiveConfig |
 			protobufs.AgentCapabilities_ReportsOwnMetrics |
 			protobufs.AgentCapabilities_ReportsHealth,
diff --git a/internal/proto/opamp.proto b/internal/proto/opamp.proto
index 851a7e7..61ed289 100644
--- a/internal/proto/opamp.proto
+++ b/internal/proto/opamp.proto
@@ -540,6 +540,8 @@ enum AgentCapabilities {
     AcceptsRestartCommand          = 0x00000400;
     // The Agent will report Health via AgentToServer.health field.
     ReportsHealth                  = 0x00000800;
+    // The Agent will report RemoteConfig status via AgentToServer.remote_config_status field.
+    ReportsRemoteConfig            = 0x00001000;
 
     // Add new capabilities here, continuing with the least significant unused bit.
 }
diff --git a/protobufs/opamp.pb.go b/protobufs/opamp.pb.go
index ca75af2..273e8c0 100644
--- a/protobufs/opamp.pb.go
+++ b/protobufs/opamp.pb.go
@@ -140,6 +140,8 @@ const (
 	AgentCapabilities_AcceptsRestartCommand AgentCapabilities = 1024
 	// The Agent will report Health via AgentToServer.health field.
 	AgentCapabilities_ReportsHealth AgentCapabilities = 2048
+	// The Agent will report RemoteConfig status via AgentToServer.remote_config_status field.
+	AgentCapabilities_ReportsRemoteConfig AgentCapabilities = 4096
 )
 
 // Enum value maps for AgentCapabilities.
@@ -158,6 +160,7 @@ var (
 		512:  "AcceptsOtherConnectionSettings",
 		1024: "AcceptsRestartCommand",
 		2048: "ReportsHealth",
+		4096: "ReportsRemoteConfig",
 	}
 	AgentCapabilities_value = map[string]int32{
 		"UnspecifiedAgentCapability":     0,
@@ -173,6 +176,7 @@ var (
 		"AcceptsOtherConnectionSettings": 512,
 		"AcceptsRestartCommand":          1024,
 		"ReportsHealth":                  2048,
+		"ReportsRemoteConfig":            4096,
 	}
 )
 
@@ -2957,7 +2961,7 @@ var file_opamp_proto_rawDesc = []byte{
 	0x63, 0x63, 0x65, 0x70, 0x74, 0x73, 0x50, 0x61, 0x63, 0x6b, 0x61, 0x67, 0x65, 0x73, 0x53, 0x74,
 	0x61, 0x74, 0x75, 0x73, 0x10, 0x10, 0x12, 0x1c, 0x0a, 0x18, 0x4f, 0x66, 0x66, 0x65, 0x72, 0x73,
 	0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e,
-	0x67, 0x73, 0x10, 0x20, 0x2a, 0xe8, 0x02, 0x0a, 0x11, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x43, 0x61,
+	0x67, 0x73, 0x10, 0x20, 0x2a, 0x82, 0x03, 0x0a, 0x11, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x43, 0x61,
 	0x70, 0x61, 0x62, 0x69, 0x6c, 0x69, 0x74, 0x69, 0x65, 0x73, 0x12, 0x1e, 0x0a, 0x1a, 0x55, 0x6e,
 	0x73, 0x70, 0x65, 0x63, 0x69, 0x66, 0x69, 0x65, 0x64, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x43, 0x61,
 	0x70, 0x61, 0x62, 0x69, 0x6c, 0x69, 0x74, 0x79, 0x10, 0x00, 0x12, 0x11, 0x0a, 0x0d, 0x52, 0x65,
@@ -2979,11 +2983,13 @@ var file_opamp_proto_rawDesc = []byte{
 	0x74, 0x69, 0x6f, 0x6e, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x73, 0x10, 0x80, 0x04, 0x12,
 	0x1a, 0x0a, 0x15, 0x41, 0x63, 0x63, 0x65, 0x70, 0x74, 0x73, 0x52, 0x65, 0x73, 0x74, 0x61, 0x72,
 	0x74, 0x43, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x10, 0x80, 0x08, 0x12, 0x12, 0x0a, 0x0d, 0x52,
-	0x65, 0x70, 0x6f, 0x72, 0x74, 0x73, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x10, 0x80, 0x10, 0x42,
-	0x2e, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x6f, 0x70,
-	0x65, 0x6e, 0x2d, 0x74, 0x65, 0x6c, 0x65, 0x6d, 0x65, 0x74, 0x72, 0x79, 0x2f, 0x6f, 0x70, 0x61,
-	0x6d, 0x70, 0x2d, 0x67, 0x6f, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x73, 0x62,
-	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x65, 0x70, 0x6f, 0x72, 0x74, 0x73, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x10, 0x80, 0x10, 0x12,
+	0x18, 0x0a, 0x13, 0x52, 0x65, 0x70, 0x6f, 0x72, 0x74, 0x73, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65,
+	0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x10, 0x80, 0x20, 0x42, 0x2e, 0x5a, 0x2c, 0x67, 0x69, 0x74,
+	0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x6f, 0x70, 0x65, 0x6e, 0x2d, 0x74, 0x65, 0x6c,
+	0x65, 0x6d, 0x65, 0x74, 0x72, 0x79, 0x2f, 0x6f, 0x70, 0x61, 0x6d, 0x70, 0x2d, 0x67, 0x6f, 0x2f,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x33,
 }
 
 var (
diff --git a/server/serverimpl.go b/server/serverimpl.go
index 7f4dd11..4fee254 100644
--- a/server/serverimpl.go
+++ b/server/serverimpl.go
@@ -47,6 +47,7 @@ func New(logger types.Logger) *server {
 
 func (s *server) Attach(settings Settings) (HTTPHandlerFunc, error) {
 	s.settings = settings
+	// TODO: Add support for compression using Upgrader.EnableCompression field.
 	s.wsUpgrader = websocket.Upgrader{}
 	return s.httpHandler, nil
 }


Test patch:
diff --git a/client/clientimpl_test.go b/client/clientimpl_test.go
index 226b0e6..5cdc1d9 100644
--- a/client/clientimpl_test.go
+++ b/client/clientimpl_test.go
@@ -117,6 +117,17 @@ func TestConnectNoServer(t *testing.T) {
 	})
 }
 
+func TestInvalidInstanceId(t *testing.T) {
+	testClients(t, func(t *testing.T, client OpAMPClient) {
+		settings := createNoServerSettings()
+		prepareClient(t, &settings, client)
+		settings.InstanceUid = "invalidid"
+
+		err := client.Start(context.Background(), settings)
+		assert.Error(t, err)
+	})
+}
+
 func TestOnConnectFail(t *testing.T) {
 	testClients(t, func(t *testing.T, client OpAMPClient) {
 		var connectErr atomic.Value
@@ -915,7 +926,7 @@ func verifyRemoteConfigUpdate(t *testing.T, successCase bool, expectStatus *prot
 					}
 				},
 			},
-			Capabilities: protobufs.AgentCapabilities_AcceptsRemoteConfig,
+			Capabilities: protobufs.AgentCapabilities_AcceptsRemoteConfig | protobufs.AgentCapabilities_ReportsRemoteConfig,
 		}
 		prepareClient(t, &settings, client)
 
diff --git a/client/wsclient_test.go b/client/wsclient_test.go
index b10b90a..2f08df7 100644
--- a/client/wsclient_test.go
+++ b/client/wsclient_test.go
@@ -2,18 +2,23 @@ package client
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"sync/atomic"
 	"testing"
 
 	"github.com/gorilla/websocket"
 	"github.com/open-telemetry/opamp-go/client/internal"
 	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/protobufs"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
 )
 
 func TestDisconnectWSByServer(t *testing.T) {
 	// Start a Server.
 	srv := internal.StartMockServer(t)
+
 	var conn atomic.Value
 	srv.OnWSConnect = func(c *websocket.Conn) {
 		conn.Store(c)
@@ -51,3 +56,122 @@ func TestDisconnectWSByServer(t *testing.T) {
 	err := client.Stop(context.Background())
 	assert.NoError(t, err)
 }
+
+func TestVerifyWSCompress(t *testing.T) {
+
+	tests := []bool{false, true}
+	for _, withCompression := range tests {
+		t.Run(fmt.Sprintf("%v", withCompression), func(t *testing.T) {
+
+			// Start a Server.
+			srv := internal.StartMockServer(t)
+			srv.EnableExpectMode()
+			if withCompression {
+				srv.EnableCompression()
+			}
+
+			// We use a transparent TCP proxy to be able to count the actual bytes transferred so that
+			// we can test the number of actual bytes vs number of expected bytes with and without compression.
+			proxy := internal.NewProxy(srv.Endpoint)
+			assert.NoError(t, proxy.Start())
+
+			// Start an OpAMP/WebSocket client.
+			var clientGotRemoteConfig atomic.Value
+			settings := types.StartSettings{
+				Callbacks: types.CallbacksStruct{
+					OnMessageFunc: func(ctx context.Context, msg *types.MessageData) {
+						if msg.RemoteConfig != nil {
+							clientGotRemoteConfig.Store(msg.RemoteConfig)
+						}
+					},
+					GetEffectiveConfigFunc: func(ctx context.Context) (*protobufs.EffectiveConfig, error) {
+						// If the client already received a remote config offer make sure to report
+						// the effective config back to the server.
+						var effCfg []byte
+						remoteCfg, _ := clientGotRemoteConfig.Load().(*protobufs.AgentRemoteConfig)
+						if remoteCfg != nil {
+							effCfg = remoteCfg.Config.ConfigMap[""].Body
+						}
+						return &protobufs.EffectiveConfig{
+							ConfigMap: &protobufs.AgentConfigMap{
+								ConfigMap: map[string]*protobufs.AgentConfigFile{
+									"key": {
+										Body: effCfg,
+									},
+								},
+							},
+						}, nil
+					},
+				},
+				Capabilities: protobufs.AgentCapabilities_AcceptsRemoteConfig |
+					protobufs.AgentCapabilities_ReportsEffectiveConfig,
+			}
+			settings.OpAMPServerURL = "ws://" + proxy.IncomingEndpoint()
+
+			if withCompression {
+				settings.EnableCompression = true
+			}
+
+			client := NewWebSocket(nil)
+			startClient(t, settings, client)
+
+			// Use highly compressible config body.
+			uncompressedCfg := []byte(strings.Repeat("test", 10000))
+
+			remoteCfg := &protobufs.AgentRemoteConfig{
+				Config: &protobufs.AgentConfigMap{
+					ConfigMap: map[string]*protobufs.AgentConfigFile{
+						"": &protobufs.AgentConfigFile{
+							Body: uncompressedCfg,
+						},
+					},
+				},
+				ConfigHash: []byte{1, 2, 3, 4},
+			}
+
+			// ---> Server
+			srv.Expect(
+				func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
+					assert.EqualValues(t, 0, msg.SequenceNum)
+					// The first status report after Start must have full AgentDescription.
+					assert.True(t, proto.Equal(client.AgentDescription(), msg.AgentDescription))
+					return &protobufs.ServerToAgent{
+						InstanceUid:  msg.InstanceUid,
+						RemoteConfig: remoteCfg,
+					}
+				},
+			)
+
+			// Wait to receive remote config
+			eventually(t, func() bool { return clientGotRemoteConfig.Load() != nil })
+
+			_ = client.UpdateEffectiveConfig(context.Background())
+
+			// ---> Server
+			srv.Expect(
+				func(msg *protobufs.AgentToServer) *protobufs.ServerToAgent {
+					return &protobufs.ServerToAgent{InstanceUid: msg.InstanceUid}
+				},
+			)
+
+			// Stop the client.
+			err := client.Stop(context.Background())
+			assert.NoError(t, err)
+
+			proxy.Stop()
+
+			fmt.Printf("sent %d, received %d\n", proxy.ClientToServerBytes(), proxy.ServerToClientBytes())
+
+			if withCompression {
+				// With compression the entire bytes exchanged should be less than the config body.
+				// This is only possible if there is any compression happening.
+				assert.Less(t, proxy.ServerToClientBytes(), len(uncompressedCfg))
+				assert.Less(t, proxy.ClientToServerBytes(), len(uncompressedCfg))
+			} else {
+				// Without compression the entire bytes exchanged should be more than the config body.
+				assert.Greater(t, proxy.ServerToClientBytes(), len(uncompressedCfg))
+				assert.Greater(t, proxy.ClientToServerBytes(), len(uncompressedCfg))
+			}
+		})
+	}
+}