Repository: open-telemetry/opamp-go
Pull request #145 (2022-11-07 22:54:38)

Implement Websocket server compression
Resolves https://github.com/open-telemetry/opamp-go/issues/121

linux-foundation-easycla[bot]:  <a href="https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/9522512/424240089/145/#/?version=2"><img src="https://s3.amazonaws.com/cla-project-logo-prod/cla-missing-id.svg" alt="CLA Missing ID" align="left" height="28" width="328"></a> <a href="https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/9522512/424240089/145/#/?version=2"><img src="https://s3.amazonaws.com/cla-project-logo-prod/cla-not-signed.svg" alt="CLA Not Signed" align="left" height="28" width="328"></a><br/><br /><ul><li> :x: The commit (cf94d5e314b9e8d999aac6cdf71f337641ce3c5b). This user is missing the User's ID, preventing the EasyCLA check. <a href='https://help.github.com/en/github/committing-changes-to-your-project/why-are-my-commits-linked-to-the-wrong-user' target='_blank'>Consult GitHub Help</a> to resolve.For further assistance with EasyCLA, <a href='https://jira.linuxfoundation.org/servicedesk/customer/portal/4' target='_blank'>please submit a support request ticket</a>.</li></ul>
codecov[bot]: # [Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/145?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) Report
Base: **75.05**% // Head: **76.24**% // Increases project coverage by **`+1.18%`** :tada:
> Coverage data is based on head [(`7f52c0c`)](https://codecov.io/gh/open-telemetry/opamp-go/pull/145?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) compared to base [(`3744deb`)](https://codecov.io/gh/open-telemetry/opamp-go/commit/3744debe2076d4af8fb1fc68f525c27c4841511f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry).
> Patch coverage: 100.00% of modified lines in pull request are covered.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #145      +/-   ##
==========================================
+ Coverage   75.05%   76.24%   +1.18%     
==========================================
  Files          23       23              
  Lines        1792     1793       +1     
==========================================
+ Hits         1345     1367      +22     
+ Misses        334      319      -15     
+ Partials      113      107       -6     
```


| [Impacted Files](https://codecov.io/gh/open-telemetry/opamp-go/pull/145?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) | Coverage Δ | |
|---|---|---|
| [internal/testhelpers/tcpproxy.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/145/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-aW50ZXJuYWwvdGVzdGhlbHBlcnMvdGNwcHJveHkuZ28=) | `73.01% <ø> (ø)` | |
| [server/serverimpl.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/145/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-c2VydmVyL3NlcnZlcmltcGwuZ28=) | `58.55% <100.00%> (+0.18%)` | :arrow_up: |
| [client/internal/receivedprocessor.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/145/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry#diff-Y2xpZW50L2ludGVybmFsL3JlY2VpdmVkcHJvY2Vzc29yLmdv) | `84.28% <0.00%> (+15.00%)` | :arrow_up: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/145?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry).


Patch:
diff --git a/client/internal/tcpproxy.go b/internal/testhelpers/tcpproxy.go
similarity index 99%
rename from client/internal/tcpproxy.go
rename to internal/testhelpers/tcpproxy.go
index c371ab8..1166a86 100644
--- a/client/internal/tcpproxy.go
+++ b/internal/testhelpers/tcpproxy.go
@@ -1,4 +1,4 @@
-package internal
+package testhelpers
 
 import (
 	"log"
diff --git a/server/server.go b/server/server.go
index ce4e329..d28f9b6 100644
--- a/server/server.go
+++ b/server/server.go
@@ -11,6 +11,11 @@ import (
 type Settings struct {
 	// Callbacks that the Server will call after successful Attach/Start.
 	Callbacks types.Callbacks
+
+	// EnableCompression can be set to true to enable the compression. Note that for WebSocket transport
+	// the compression is only effectively enabled if the client also supports compression.
+	// The data will be compressed in both directions.
+	EnableCompression bool
 }
 
 type StartSettings struct {
diff --git a/server/serverimpl.go b/server/serverimpl.go
index f1a3d34..cbb0f1c 100644
--- a/server/serverimpl.go
+++ b/server/serverimpl.go
@@ -52,8 +52,9 @@ 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{}
+	s.wsUpgrader = websocket.Upgrader{
+		EnableCompression: settings.EnableCompression,
+	}
 	return s.httpHandler, nil
 }
 


Test patch:
diff --git a/client/wsclient_test.go b/client/wsclient_test.go
index 2de35b4..715b140 100644
--- a/client/wsclient_test.go
+++ b/client/wsclient_test.go
@@ -8,11 +8,13 @@ import (
 	"testing"
 
 	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
+
 	"github.com/open-telemetry/opamp-go/client/internal"
 	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/internal/testhelpers"
 	"github.com/open-telemetry/opamp-go/protobufs"
-	"github.com/stretchr/testify/assert"
-	"google.golang.org/protobuf/proto"
 )
 
 func TestDisconnectWSByServer(t *testing.T) {
@@ -72,7 +74,7 @@ func TestVerifyWSCompress(t *testing.T) {
 
 			// 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)
+			proxy := testhelpers.NewProxy(srv.Endpoint)
 			assert.NoError(t, proxy.Start())
 
 			// Start an OpAMP/WebSocket client.
diff --git a/server/serverimpl_test.go b/server/serverimpl_test.go
index 11c7f89..03cb8b5 100644
--- a/server/serverimpl_test.go
+++ b/server/serverimpl_test.go
@@ -3,21 +3,24 @@ package server
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
 	sharedinternal "github.com/open-telemetry/opamp-go/internal"
 	"github.com/open-telemetry/opamp-go/internal/testhelpers"
 	"github.com/open-telemetry/opamp-go/protobufs"
 	"github.com/open-telemetry/opamp-go/server/types"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"google.golang.org/protobuf/proto"
 )
 
 func startServer(t *testing.T, settings *StartSettings) *server {
@@ -38,6 +41,8 @@ func startServer(t *testing.T, settings *StartSettings) *server {
 
 func dialClient(serverSettings *StartSettings) (*websocket.Conn, *http.Response, error) {
 	srvUrl := "ws://" + serverSettings.ListenEndpoint + serverSettings.ListenPath
+	dailer := websocket.DefaultDialer
+	dailer.EnableCompression = serverSettings.EnableCompression
 	return websocket.DefaultDialer.Dial(srvUrl, nil)
 }
 
@@ -224,6 +229,103 @@ func TestServerReceiveSendMessage(t *testing.T) {
 	assert.EqualValues(t, protobufs.ServerCapabilities_ServerCapabilities_AcceptsStatus, response.Capabilities)
 }
 
+func TestServerReceiveSendMessageWithCompression(t *testing.T) {
+	// Use highly compressible config body.
+	uncompressedCfg := []byte(strings.Repeat("test", 10000))
+	tests := []bool{false, true}
+	for _, withCompression := range tests {
+		t.Run(fmt.Sprintf("%v", withCompression), func(t *testing.T) {
+			var rcvMsg atomic.Value
+			callbacks := CallbacksStruct{
+				OnConnectingFunc: func(request *http.Request) types.ConnectionResponse {
+					return types.ConnectionResponse{Accept: true}
+				},
+				OnMessageFunc: func(conn types.Connection, message *protobufs.AgentToServer) *protobufs.ServerToAgent {
+					// Remember received message.
+					rcvMsg.Store(message)
+
+					// Send a response.
+					response := protobufs.ServerToAgent{
+						InstanceUid:  message.InstanceUid,
+						Capabilities: uint64(protobufs.ServerCapabilities_ServerCapabilities_AcceptsStatus),
+						RemoteConfig: &protobufs.AgentRemoteConfig{
+							Config: &protobufs.AgentConfigMap{
+								ConfigMap: map[string]*protobufs.AgentConfigFile{
+									"": {Body: uncompressedCfg},
+								},
+							},
+						},
+					}
+					return &response
+				},
+			}
+
+			// Start a Server.
+			settings := &StartSettings{Settings: Settings{Callbacks: callbacks, EnableCompression: withCompression}}
+			srv := startServer(t, settings)
+			defer srv.Stop(context.Background())
+
+			// 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 := testhelpers.NewProxy(settings.ListenEndpoint)
+			assert.NoError(t, proxy.Start())
+
+			serverSettings := *settings
+			serverSettings.ListenEndpoint = proxy.IncomingEndpoint()
+			// Connect using a WebSocket client.
+			conn, _, _ := dialClient(&serverSettings)
+			require.NotNil(t, conn)
+			defer conn.Close()
+
+			// Send a message to the Server.
+			sendMsg := protobufs.AgentToServer{
+				InstanceUid: "10000000",
+				EffectiveConfig: &protobufs.EffectiveConfig{
+					ConfigMap: &protobufs.AgentConfigMap{
+						ConfigMap: map[string]*protobufs.AgentConfigFile{
+							"": {Body: uncompressedCfg},
+						},
+					},
+				},
+			}
+			bytes, err := proto.Marshal(&sendMsg)
+			require.NoError(t, err)
+			err = conn.WriteMessage(websocket.BinaryMessage, bytes)
+			require.NoError(t, err)
+
+			// Wait until Server receives the message.
+			eventually(t, func() bool { return rcvMsg.Load() != nil })
+			assert.True(t, proto.Equal(rcvMsg.Load().(proto.Message), &sendMsg))
+
+			// Read Server's response.
+			mt, bytes, err := conn.ReadMessage()
+			require.NoError(t, err)
+			require.EqualValues(t, websocket.BinaryMessage, mt)
+
+			// Decode the response.
+			var response protobufs.ServerToAgent
+			err = proto.Unmarshal(bytes, &response)
+			require.NoError(t, err)
+
+			fmt.Printf("sent %d, received %d\n", proxy.ClientToServerBytes(), proxy.ServerToClientBytes())
+
+			// Verify the response.
+			assert.EqualValues(t, sendMsg.InstanceUid, response.InstanceUid)
+			assert.EqualValues(t, protobufs.ServerCapabilities_ServerCapabilities_AcceptsStatus, response.Capabilities)
+			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.ClientToServerBytes(), len(uncompressedCfg))
+				assert.Less(t, proxy.ServerToClientBytes(), len(uncompressedCfg))
+			} else {
+				// Without compression the entire bytes exchanged should be more than the config body.
+				assert.Greater(t, proxy.ClientToServerBytes(), len(uncompressedCfg))
+				assert.Greater(t, proxy.ServerToClientBytes(), len(uncompressedCfg))
+			}
+		})
+	}
+}
+
 func TestServerReceiveSendMessagePlainHTTP(t *testing.T) {
 	var rcvMsg atomic.Value
 	var onConnectedCalled, onCloseCalled int32