fix: mem leak due to unbounded metrics cardinality

[toddbaert]

Fixes a memory leak that can be seen with many remote clients due to default OTel metrics attributes included remote peer and ephemeral port.

See: https://github.com/connectrpc/otelconnect-go?tab=readme-ov-file#reducing-metrics-and-tracing-cardinality

I tested this with this integration test (I can see the memory leak and confirmed this fixes it).

Details

package service

import (
	"bytes"
	"context"
	"fmt"
	"net/http"
	"runtime"
	"testing"
	"time"

	mock "github.com/open-feature/flagd/core/pkg/evaluator/mock"
	"github.com/open-feature/flagd/core/pkg/logger"
	"github.com/open-feature/flagd/core/pkg/model"
	iservice "github.com/open-feature/flagd/core/pkg/service"
	"github.com/open-feature/flagd/core/pkg/telemetry"
	"github.com/stretchr/testify/require"
	"go.opentelemetry.io/otel/sdk/metric"
	"go.opentelemetry.io/otel/sdk/resource"
	"go.uber.org/mock/gomock"
)

// TestMemoryGrowthFromPeerAttributes is a long-running test that demonstrates
// unbounded memory growth when otelconnect records net.peer.port in metrics.
//
// Run with: go test -v -run TestMemoryGrowthFromPeerAttributes -timeout=120s ./flagd/pkg/service/flag-evaluation/
func TestMemoryGrowthFromPeerAttributes(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping long-running memory test in short mode")
	}

	const (
		port       = 18490
		batchSize  = 500
		batches    = 20
		resolveURL = "http://localhost:%d/flagd.evaluation.v1.Service/ResolveBoolean"
	)

	ctrl := gomock.NewController(t)
	eval := mock.NewMockIEvaluator(ctrl)
	eval.EXPECT().ResolveBooleanValue(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(
		true, "on", model.DefaultReason, map[string]interface{}{}, nil,
	).AnyTimes()

	reader := metric.NewManualReader()
	rs := resource.NewWithAttributes("testSchema")
	metricRecorder := telemetry.NewOTelRecorder(reader, rs, "memory-test")

	options, err := telemetry.BuildConnectOptions(telemetry.Config{})
	require.NoError(t, err)

	svc := NewConnectService(logger.NewLogger(nil, false), eval, nil, metricRecorder)

	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)

	go func() {
		_ = svc.Serve(ctx, iservice.Configuration{
			ReadinessProbe: func() bool { return true },
			Port:           port,
			Options:        options,
		})
	}()

	url := fmt.Sprintf(resolveURL, port)
	require.Eventually(t, func() bool {
		resp, err := http.Get(fmt.Sprintf("http://localhost:%d/flagd.evaluation.v1.Service/ResolveAll", port))
		if err == nil && resp != nil {
			resp.Body.Close()
		}
		return err == nil && resp != nil
	}, 3*time.Second, 100*time.Millisecond)

	body := []byte(`{"flagKey":"testFlag"}`)

	// baseline
	runtime.GC()
	var baseline runtime.MemStats
	runtime.ReadMemStats(&baseline)

	t.Logf("")
	t.Logf("%-10s  %-12s  %-12s  %-12s", "Requests", "HeapInuse", "HeapAlloc", "Delta")
	t.Logf("%-10s  %-12s  %-12s  %-12s", "--------", "---------", "---------", "-----")

	totalRequests := 0
	for batch := 0; batch < batches; batch++ {
		for i := 0; i < batchSize; i++ {
			client := &http.Client{
				Transport: &http.Transport{
					DisableKeepAlives: true,
				},
			}
			req, _ := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
			req.Header.Set("Content-Type", "application/json")
			resp, err := client.Do(req)
			if err == nil {
				resp.Body.Close()
			}
		}
		totalRequests += batchSize

		runtime.GC()
		var m runtime.MemStats
		runtime.ReadMemStats(&m)
		delta := int64(m.HeapInuse) - int64(baseline.HeapInuse)

		t.Logf("%-10d  %-12s  %-12s  %+.1f MB",
			totalRequests,
			formatMB(m.HeapInuse),
			formatMB(m.HeapAlloc),
			float64(delta)/1024/1024,
		)
	}

	// final summary
	runtime.GC()
	var final runtime.MemStats
	runtime.ReadMemStats(&final)
	totalDelta := float64(int64(final.HeapInuse)-int64(baseline.HeapInuse)) / 1024 / 1024

	t.Logf("")
	t.Logf("Total: %d requests, heap grew by %.1f MB", totalRequests, totalDelta)

	if totalDelta > 10 {
		t.Logf("WARNING: heap grew by %.1f MB; this indicates unbounded cardinality growth", totalDelta)
	} else {
		t.Logf("Heap growth is bounded (%.1f MB); cardinality is under control", totalDelta)
	}
}

func formatMB(b uint64) string {
	return fmt.Sprintf("%.1f MB", float64(b)/1024/1024)
}

[netlify]

✅ Deploy Preview for polite-licorice-3db33c canceled.

Name	Link
🔨 Latest commit	04521d6
🔍 Latest deploy log	https://app.netlify.com/projects/polite-licorice-3db33c/deploys/69d52259034db900084253ab

[toddbaert]

Primary fix.

[toddbaert]

Additional guard to prevent metrics leak for some other custom metrics, etc.

[beeme1mr]

Could you please mention this in the docs too?

https://flagd.dev/reference/monitoring/#metrics

[gemini-code-assist]

Code Review

This pull request updates telemetry settings by disabling server peer attributes in the connect interceptor and attempting to implement a metric cardinality limit. A critical issue was identified where the msdk.WithCardinalityLimit function is not available in the stable OpenTelemetry Go SDK, which will cause a compilation failure. The feedback suggests removing this programmatic configuration and instead relying on the OTEL_GO_X_CARDINALITY_LIMIT environment variable for cardinality management.

[juanparadox]

Thank you for jumping on this 😄

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud