fix: Improve compression performance via new provider (#532)

Author: mmatecki

go.mod

@@ -60,6 +60,7 @@ require (
 
 require (
 	github.com/alecthomas/units v0.0.0-20231202071711-9a357b53e9c9
+	github.com/klauspost/compress v1.18.0
 	github.com/launchdarkly/api-client-go/v13 v13.0.1-0.20230420175109-f5469391a13e
 )
 
@@ -84,7 +85,6 @@ require (
 	github.com/google/s2a-go v0.1.7 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
 	github.com/googleapis/gax-go/v2 v2.12.0 // indirect
-	github.com/gorilla/handlers v1.5.2 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-rootcerts v1.0.2 // indirect

go.sum

@@ -234,8 +234,6 @@ github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/googleapis/gax-go/v2 v2.12.0 h1:A+gCJKdRfqXkr+BIRGtZLibNXf0m1f9E4HG56etFpas=
 github.com/googleapis/gax-go/v2 v2.12.0/go.mod h1:y+aIqrI5eb1YGMVJfuV3185Ts/D7qKpsEkdD5+I6QGU=
-github.com/gorilla/handlers v1.5.2 h1:cLTUSsNkgcwhgRqvCNmdbRWG0A3N4F+M2nWKdScwyEE=
-github.com/gorilla/handlers v1.5.2/go.mod h1:dX+xVpaxdSw+q0Qek8SSsl3dfMk3jNddUkMzo0GtH0w=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 h1:+ngKgrYPPJrOjhax5N+uePQ0Fh1Z7PheYoUI/0nzkPA=

relay/relay_routes.go

@@ -14,8 +14,8 @@ import (
 	"github.com/launchdarkly/go-sdk-common/v3/ldlog"
 	ldevents "github.com/launchdarkly/go-sdk-events/v3"
 
-	h "github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
+	h "github.com/klauspost/compress/gzhttp"
 )
 
 const (
@@ -36,7 +36,9 @@ func (r *Relay) makeRouter() *mux.Router {
 		router.Use(logging.RequestLoggerMiddleware(r.loggers))
 	}
 	if r.config.HTTP.EnableCompression {
-		router.Use(h.CompressHandler)
+		router.Use(func(next http.Handler) http.Handler {
+			return h.GzipHandler(next)
+		})
 	}
 	router.Handle("/status", statusHandler(r)).Methods("GET")
 

relay/relay_test.go

@@ -3,17 +3,23 @@ package relay
 import (
 	"bytes"
 	"compress/gzip"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"runtime"
+	"runtime/debug"
+	"sync"
 	"testing"
+	"time"
 
 	c "github.com/launchdarkly/ld-relay/v8/config"
 
 	"github.com/launchdarkly/go-configtypes"
 	"github.com/launchdarkly/go-sdk-common/v3/ldlog"
 	"github.com/launchdarkly/go-test-helpers/v3/httphelpers"
 
+	"github.com/klauspost/compress/gzhttp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -185,9 +191,10 @@ func TestCompressionIsAppliedWhenEnabled(t *testing.T) {
 	}
 
 	withStartedRelay(t, configWithCompression, func(p relayTestParams) {
-		// Create a request to the status endpoint
-		req, _ := http.NewRequest("GET", "/status", nil)
+		// Create a request to the flags endpoint which returns more data
+		req, _ := http.NewRequest("GET", "/sdk/flags", nil)
 		req.Header.Set("Accept-Encoding", "gzip")
+		req.Header.Set("Authorization", "test-key")
 
 		w := httptest.NewRecorder()
 		p.relay.ServeHTTP(w, req)
@@ -208,6 +215,9 @@ func TestCompressionIsAppliedWhenEnabled(t *testing.T) {
 		decompressed, err := io.ReadAll(reader)
 		assert.NoError(t, err, "Should be able to read decompressed content")
 		assert.Greater(t, len(decompressed), 0, "Decompressed content should not be empty")
+
+		// Verify the decompressed content is substantial enough to test compression
+		assert.Greater(t, len(decompressed), gzhttp.DefaultMinSize, fmt.Sprintf("Decompressed content should be larger than %d bytes to properly test compression", gzhttp.DefaultMinSize))
 	})
 }
 
@@ -225,9 +235,10 @@ func TestCompressionIsNotAppliedWhenDisabled(t *testing.T) {
 	}
 
 	withStartedRelay(t, configWithoutCompression, func(p relayTestParams) {
-		// Create a request to the status endpoint
-		req, _ := http.NewRequest("GET", "/status", nil)
+		// Create a request to the flags endpoint which returns more data
+		req, _ := http.NewRequest("GET", "/sdk/flags", nil)
 		req.Header.Set("Accept-Encoding", "gzip")
+		req.Header.Set("Authorization", "test-key")
 
 		w := httptest.NewRecorder()
 		p.relay.ServeHTTP(w, req)
@@ -240,8 +251,132 @@ func TestCompressionIsNotAppliedWhenDisabled(t *testing.T) {
 		body := w.Body.Bytes()
 		assert.Greater(t, len(body), 0, "Response body should not be empty")
 
+		// Verify the uncompressed content is substantial enough
+		assert.Greater(t, len(body), gzhttp.DefaultMinSize, fmt.Sprintf("Uncompressed content should be larger than %d bytes", gzhttp.DefaultMinSize))
+
+		t.Logf("Body size: %d bytes", len(body))
 		// Try to decompress the body - this should fail since it's not compressed
 		_, err := gzip.NewReader(io.NopCloser(bytes.NewReader(body)))
 		assert.Error(t, err, "Response should not be gzip content when compression is disabled")
 	})
 }
+
+func TestLoadCompression(t *testing.T) {
+	// Set minimal memory limit for the test
+	originalLimit := debug.SetMemoryLimit(10 * 1024 * 1024) // 10MB limit
+	defer debug.SetMemoryLimit(originalLimit)
+
+	// Force garbage collection before starting
+	runtime.GC()
+
+	// Test with compression enabled
+	configWithCompression := c.Config{
+		HTTP: c.HTTPConfig{
+			EnableCompression: true,
+		},
+		Environment: map[string]*c.EnvConfig{
+			"test": {
+				SDKKey: "test-key",
+			},
+		},
+	}
+
+	withStartedRelay(t, configWithCompression, func(p relayTestParams) {
+		var wg sync.WaitGroup
+		var mu sync.Mutex
+		var requestCount int
+		var lastError error
+		var failed bool
+		var maxAlloc uint64
+
+		// Start with a reasonable number of concurrent requests
+		concurrency := 100
+		maxRequests := 1000
+
+		for i := 0; i < concurrency; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+
+				for {
+					mu.Lock()
+					if failed || requestCount >= maxRequests {
+						mu.Unlock()
+						return
+					}
+					currentCount := requestCount
+					requestCount++
+					mu.Unlock()
+
+					// Create a request to the flags endpoint
+					req, err := http.NewRequest("GET", "/sdk/flags", nil)
+					if err != nil {
+						mu.Lock()
+						lastError = err
+						failed = true
+						mu.Unlock()
+						return
+					}
+
+					req.Header.Set("Accept-Encoding", "gzip")
+					req.Header.Set("Authorization", "test-key")
+
+					w := httptest.NewRecorder()
+					p.relay.ServeHTTP(w, req)
+
+					// Check if the request succeeded
+					if w.Result().StatusCode != http.StatusOK {
+						mu.Lock()
+						lastError = fmt.Errorf("request failed with status: %d", w.Result().StatusCode)
+						failed = true
+						mu.Unlock()
+						return
+					}
+
+					// Verify compression is working
+					if w.Header().Get("Content-Encoding") != "gzip" {
+						mu.Lock()
+						lastError = fmt.Errorf("compression not applied at request %d", currentCount)
+						failed = true
+						mu.Unlock()
+						return
+					}
+
+					// Check memory usage
+					var m runtime.MemStats
+					runtime.ReadMemStats(&m)
+
+					// Track maximum memory usage
+					mu.Lock()
+					if m.Alloc > maxAlloc {
+						maxAlloc = m.Alloc
+					}
+					mu.Unlock()
+
+					// Small delay to prevent overwhelming the system
+					time.Sleep(1 * time.Millisecond)
+				}
+			}()
+		}
+
+		wg.Wait()
+
+		// Report results
+		t.Logf("Load test completed: %d requests processed", requestCount)
+
+		if failed {
+			t.Fatalf("Test failed: %v", lastError)
+		}
+
+		t.Logf("Test completed successfully with %d requests", requestCount)
+		assert.Greater(t, requestCount, 0, "Should have processed requests successfully")
+
+		// Final memory stats
+		var m runtime.MemStats
+		runtime.ReadMemStats(&m)
+		t.Logf("Final memory usage: Alloc=%d MB, TotalAlloc=%d MB, NumGC=%d",
+			m.Alloc/1024/1024, m.TotalAlloc/1024/1024, m.NumGC)
+		t.Logf("Peak memory usage: MaxAlloc=%d MB",
+			maxAlloc/1024/1024)
+	})
+}