Repository: open-telemetry/opamp-go
Pull request #126 (2022-08-17 19:50:30)

Implement gzip compression for HTTP server
Contributes to https://github.com/open-telemetry/opamp-go/issues/112

codecov[bot]: # [Codecov](https://codecov.io/gh/open-telemetry/opamp-go/pull/126?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) Report
> Merging [#126](https://codecov.io/gh/open-telemetry/opamp-go/pull/126?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) (be3b029) into [main](https://codecov.io/gh/open-telemetry/opamp-go/commit/0c1d2c1682608f75854121c8e4f12f9e4acc53fe?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) (0c1d2c1) will **increase** coverage by `0.59%`.
> The diff coverage is `48.71%`.

```diff
@@            Coverage Diff             @@
##             main     #126      +/-   ##
==========================================
+ Coverage   75.36%   75.95%   +0.59%     
==========================================
  Files          23       23              
  Lines        1709     1747      +38     
==========================================
+ Hits         1288     1327      +39     
+ Misses        315      314       -1     
  Partials      106      106              
```


| [Impacted Files](https://codecov.io/gh/open-telemetry/opamp-go/pull/126?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=open-telemetry) | Coverage Δ | |
|---|---|---|
| [server/serverimpl.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/126/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.37% <48.71%> (-2.29%)` | :arrow_down: |
| [client/internal/receivedprocessor.go](https://codecov.io/gh/open-telemetry/opamp-go/pull/126/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)


Patch:
diff --git a/server/serverimpl.go b/server/serverimpl.go
index 4fee254..f1a3d34 100644
--- a/server/serverimpl.go
+++ b/server/serverimpl.go
@@ -1,6 +1,8 @@
 package server
 
 import (
+	"bytes"
+	"compress/gzip"
 	"context"
 	"errors"
 	"io"
@@ -21,6 +23,9 @@ var (
 
 const defaultOpAMPPath = "/v1/opamp"
 const headerContentType = "Content-Type"
+const headerContentEncoding = "Content-Encoding"
+const headerAcceptEncoding = "Accept-Encoding"
+const contentEncodingGzip = "gzip"
 const contentTypeProtobuf = "application/x-protobuf"
 
 type server struct {
@@ -228,8 +233,45 @@ func (s *server) handleWSConnection(wsConn *websocket.Conn) {
 	}
 }
 
+func decompressGzip(data []byte) ([]byte, error) {
+	r, err := gzip.NewReader(bytes.NewBuffer(data))
+	if err != nil {
+		return nil, err
+	}
+	defer r.Close()
+	return io.ReadAll(r)
+}
+
+func (s *server) readReqBody(req *http.Request) ([]byte, error) {
+	data, err := io.ReadAll(req.Body)
+	if err != nil {
+		return nil, err
+	}
+	if req.Header.Get(headerContentEncoding) == contentEncodingGzip {
+		data, err = decompressGzip(data)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return data, nil
+}
+
+func compressGzip(data []byte) ([]byte, error) {
+	var buf bytes.Buffer
+	w := gzip.NewWriter(&buf)
+	_, err := w.Write(data)
+	if err != nil {
+		return nil, err
+	}
+	err = w.Close()
+	if err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
 func (s *server) handlePlainHTTPRequest(req *http.Request, w http.ResponseWriter) {
-	bytes, err := io.ReadAll(req.Body)
+	bytes, err := s.readReqBody(req)
 	if err != nil {
 		s.logger.Debugf("Cannot read HTTP body: %v", err)
 		w.WriteHeader(http.StatusBadRequest)
@@ -280,6 +322,15 @@ func (s *server) handlePlainHTTPRequest(req *http.Request, w http.ResponseWriter
 
 	// Send the response.
 	w.Header().Set(headerContentType, contentTypeProtobuf)
+	if req.Header.Get(headerAcceptEncoding) == contentEncodingGzip {
+		bytes, err = compressGzip(bytes)
+		if err != nil {
+			s.logger.Errorf("Cannot compress response: %v", err)
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set(headerContentEncoding, contentEncodingGzip)
+	}
 	_, err = w.Write(bytes)
 
 	if err != nil {


Test patch:
diff --git a/server/serverimpl_test.go b/server/serverimpl_test.go
index ef0e722..00af23e 100644
--- a/server/serverimpl_test.go
+++ b/server/serverimpl_test.go
@@ -295,3 +295,153 @@ func TestServerAttachAcceptConnection(t *testing.T) {
 	conn.Close()
 	eventually(t, func() bool { return atomic.LoadInt32(&connectionCloseCalled) == 1 })
 }
+
+func TestServerHonoursClientRequestContentEncoding(t *testing.T) {
+
+	hc := http.Client{}
+	var rcvMsg atomic.Value
+	var onConnectedCalled, onCloseCalled int32
+	callbacks := CallbacksStruct{
+		OnConnectingFunc: func(request *http.Request) types.ConnectionResponse {
+			return types.ConnectionResponse{Accept: true}
+		},
+		OnConnectedFunc: func(conn types.Connection) {
+			atomic.StoreInt32(&onConnectedCalled, 1)
+		},
+		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: protobufs.ServerCapabilities_AcceptsStatus,
+			}
+			return &response
+		},
+		OnConnectionCloseFunc: func(conn types.Connection) {
+			atomic.StoreInt32(&onCloseCalled, 1)
+		},
+	}
+
+	// Start a Server.
+	settings := &StartSettings{Settings: Settings{Callbacks: callbacks}}
+	srv := startServer(t, settings)
+	defer srv.Stop(context.Background())
+
+	// Send a message to the Server.
+	sendMsg := protobufs.AgentToServer{
+		InstanceUid: "12345678",
+	}
+	b, err := proto.Marshal(&sendMsg)
+	require.NoError(t, err)
+
+	// gzip compress the request payload
+	b, err = compressGzip(b)
+	require.NoError(t, err)
+
+	req, err := http.NewRequest("POST", "http://"+settings.ListenEndpoint+settings.ListenPath, bytes.NewReader(b))
+	req.Header.Set(headerContentType, contentTypeProtobuf)
+	req.Header.Set(headerContentEncoding, contentEncodingGzip)
+	resp, err := hc.Do(req)
+	require.NoError(t, err)
+
+	// Wait until Server receives the message.
+	eventually(t, func() bool { return rcvMsg.Load() != nil })
+	assert.True(t, atomic.LoadInt32(&onConnectedCalled) == 1)
+
+	// Verify the received message is what was sent.
+	assert.True(t, proto.Equal(rcvMsg.Load().(proto.Message), &sendMsg))
+
+	// Read Server's response.
+	b, err = io.ReadAll(resp.Body)
+	require.NoError(t, err)
+
+	assert.EqualValues(t, http.StatusOK, resp.StatusCode)
+	assert.EqualValues(t, contentTypeProtobuf, resp.Header.Get(headerContentType))
+
+	// Decode the response.
+	var response protobufs.ServerToAgent
+	err = proto.Unmarshal(b, &response)
+	require.NoError(t, err)
+
+	// Verify the response.
+	assert.EqualValues(t, sendMsg.InstanceUid, response.InstanceUid)
+	assert.EqualValues(t, protobufs.ServerCapabilities_AcceptsStatus, response.Capabilities)
+
+	eventually(t, func() bool { return atomic.LoadInt32(&onCloseCalled) == 1 })
+}
+
+func TestServerHonoursAcceptEncoding(t *testing.T) {
+
+	hc := http.Client{}
+	var rcvMsg atomic.Value
+	var onConnectedCalled, onCloseCalled int32
+	callbacks := CallbacksStruct{
+		OnConnectingFunc: func(request *http.Request) types.ConnectionResponse {
+			return types.ConnectionResponse{Accept: true}
+		},
+		OnConnectedFunc: func(conn types.Connection) {
+			atomic.StoreInt32(&onConnectedCalled, 1)
+		},
+		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: protobufs.ServerCapabilities_AcceptsStatus,
+			}
+			return &response
+		},
+		OnConnectionCloseFunc: func(conn types.Connection) {
+			atomic.StoreInt32(&onCloseCalled, 1)
+		},
+	}
+
+	// Start a Server.
+	settings := &StartSettings{Settings: Settings{Callbacks: callbacks}}
+	srv := startServer(t, settings)
+	defer srv.Stop(context.Background())
+
+	// Send a message to the Server.
+	sendMsg := protobufs.AgentToServer{
+		InstanceUid: "12345678",
+	}
+	b, err := proto.Marshal(&sendMsg)
+	require.NoError(t, err)
+	req, err := http.NewRequest("POST", "http://"+settings.ListenEndpoint+settings.ListenPath, bytes.NewReader(b))
+	req.Header.Set(headerContentType, contentTypeProtobuf)
+	req.Header.Set(headerAcceptEncoding, contentEncodingGzip)
+	resp, err := hc.Do(req)
+	require.NoError(t, err)
+
+	// Wait until Server receives the message.
+	eventually(t, func() bool { return rcvMsg.Load() != nil })
+	assert.True(t, atomic.LoadInt32(&onConnectedCalled) == 1)
+
+	// Verify the received message is what was sent.
+	assert.True(t, proto.Equal(rcvMsg.Load().(proto.Message), &sendMsg))
+
+	// Read Server's response.
+	b, err = io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	// Decompress the gzip response
+	b, err = decompressGzip(b)
+	require.NoError(t, err)
+
+	assert.EqualValues(t, http.StatusOK, resp.StatusCode)
+	assert.EqualValues(t, contentTypeProtobuf, resp.Header.Get(headerContentType))
+
+	// Decode the response.
+	var response protobufs.ServerToAgent
+	err = proto.Unmarshal(b, &response)
+	require.NoError(t, err)
+
+	// Verify the response.
+	assert.EqualValues(t, sendMsg.InstanceUid, response.InstanceUid)
+	assert.EqualValues(t, protobufs.ServerCapabilities_AcceptsStatus, response.Capabilities)
+
+	eventually(t, func() bool { return atomic.LoadInt32(&onCloseCalled) == 1 })
+}