[extension/jaegerremotesampling] gRPC remote mode propagates configured HTTP headers

.chloggen/zcross_jaegerremotesampling_support-header-enhancement-outbound-grpc.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: extension/jaegerremotesampling
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: gRPC remote source usage in jaegerremotesampling extension propagates HTTP headers if set in gRPC client config
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [24414]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

extension/jaegerremotesampling/extension.go

@@ -7,7 +7,6 @@ import (
 	"context"
 	"fmt"
 
-	grpcStore "github.com/jaegertracing/jaeger/cmd/agent/app/configmanager/grpc"
 	"github.com/jaegertracing/jaeger/cmd/collector/app/sampling/strategystore"
 	"github.com/jaegertracing/jaeger/plugin/sampling/strategystore/static"
 	"go.opentelemetry.io/collector/component"
@@ -62,13 +61,10 @@ func (jrse *jrsExtension) Start(ctx context.Context, host component.Host) error
 	if jrse.cfg.Source.Remote != nil {
 		conn, err := jrse.cfg.Source.Remote.ToClientConn(ctx, host, jrse.telemetry)
 		if err != nil {
-			return fmt.Errorf("error while connecting to the remote sampling source: %w", err)
+			return fmt.Errorf("failed to create the remote strategy store: %w", err)
 		}
-
-		jrse.samplingStore = grpcStore.NewConfigManager(conn)
-		jrse.closers = append(jrse.closers, func() error {
-			return conn.Close()
-		})
+		jrse.closers = append(jrse.closers, conn.Close)
+		jrse.samplingStore = internal.NewRemoteStrategyStore(conn, jrse.cfg.Source.Remote)
 	}
 
 	if jrse.cfg.HTTPServerSettings != nil {

extension/jaegerremotesampling/extension_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"net/http"
 	"path/filepath"
 	"testing"
 
@@ -15,7 +16,10 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/config/configopaque"
+	"go.opentelemetry.io/collector/config/configtls"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
 )
 
 func TestNewExtension(t *testing.T) {
@@ -41,44 +45,97 @@ func TestStartAndShutdownLocalFile(t *testing.T) {
 	assert.NoError(t, e.Shutdown(context.Background()))
 }
 
-func TestStartAndShutdownRemote(t *testing.T) {
-	// prepare the socket the mock server will listen at
-	lis, err := net.Listen("tcp", "127.0.0.1:0")
-	require.NoError(t, err)
-
-	// create the mock server
-	server := grpc.NewServer()
-
-	// register the service
-	api_v2.RegisterSamplingManagerServer(server, &samplingServer{})
-
-	go func() {
-		err = server.Serve(lis)
-		require.NoError(t, err)
-	}()
-
-	// create the config, pointing to the mock server
-	cfg := testConfig()
-	cfg.GRPCServerSettings.NetAddr.Endpoint = "127.0.0.1:0"
-	cfg.Source.Remote = &configgrpc.GRPCClientSettings{
-		Endpoint:     fmt.Sprintf("127.0.0.1:%d", lis.Addr().(*net.TCPAddr).Port),
-		WaitForReady: true,
+func TestStartAndCallAndShutdownRemote(t *testing.T) {
+	for _, tc := range []struct {
+		name                     string
+		remoteClientHeaderConfig map[string]configopaque.String
+	}{
+		{
+			name: "no configured header additions",
+		},
+		{
+			name: "configured header additions",
+			remoteClientHeaderConfig: map[string]configopaque.String{
+				"testheadername":    "testheadervalue",
+				"anotherheadername": "anotherheadervalue",
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+
+			// prepare the socket the mock server will listen at
+			lis, err := net.Listen("tcp", "127.0.0.1:0")
+			require.NoError(t, err)
+
+			// create the mock server
+			server := grpc.NewServer()
+
+			// register the service
+			mockServer := &samplingServer{}
+			api_v2.RegisterSamplingManagerServer(server, mockServer)
+
+			go func() {
+				err = server.Serve(lis)
+				require.NoError(t, err)
+			}()
+
+			// create the config, pointing to the mock server
+			cfg := testConfig()
+			cfg.GRPCServerSettings.NetAddr.Endpoint = "127.0.0.1:0"
+			cfg.Source.Remote = &configgrpc.GRPCClientSettings{
+				Endpoint: fmt.Sprintf("127.0.0.1:%d", lis.Addr().(*net.TCPAddr).Port),
+				TLSSetting: configtls.TLSClientSetting{
+					Insecure: true, // test only
+				},
+				WaitForReady: true,
+				Headers:      tc.remoteClientHeaderConfig,
+			}
+
+			// create the extension
+			e := newExtension(cfg, componenttest.NewNopTelemetrySettings())
+			require.NotNil(t, e)
+
+			// start the server
+			assert.NoError(t, e.Start(context.Background(), componenttest.NewNopHost()))
+
+			// make a call
+			resp, err := http.Get("http://127.0.0.1:5778/sampling?service=foo")
+			assert.NoError(t, err)
+			assert.Equal(t, 200, resp.StatusCode)
+
+			// shut down the server
+			assert.NoError(t, e.Shutdown(context.Background()))
+
+			// verify observed calls
+			assert.Len(t, mockServer.observedCalls, 1)
+			singleCall := mockServer.observedCalls[0]
+			assert.Equal(t, &api_v2.SamplingStrategyParameters{
+				ServiceName: "foo",
+			}, singleCall.params)
+			md, ok := metadata.FromIncomingContext(singleCall.ctx)
+			assert.True(t, ok)
+			for expectedHeaderName, expectedHeaderValue := range tc.remoteClientHeaderConfig {
+				assert.Equal(t, []string{string(expectedHeaderValue)}, md.Get(expectedHeaderName))
+			}
+		})
 	}
-
-	// create the extension
-	e := newExtension(cfg, componenttest.NewNopTelemetrySettings())
-	require.NotNil(t, e)
-
-	// test
-	assert.NoError(t, e.Start(context.Background(), componenttest.NewNopHost()))
-	assert.NoError(t, e.Shutdown(context.Background()))
 }
 
 type samplingServer struct {
 	api_v2.UnimplementedSamplingManagerServer
+	observedCalls []observedCall
+}
+
+type observedCall struct {
+	ctx    context.Context
+	params *api_v2.SamplingStrategyParameters
 }
 
-func (s samplingServer) GetSamplingStrategy(_ context.Context, _ *api_v2.SamplingStrategyParameters) (*api_v2.SamplingStrategyResponse, error) {
+func (s *samplingServer) GetSamplingStrategy(ctx context.Context, params *api_v2.SamplingStrategyParameters) (*api_v2.SamplingStrategyResponse, error) {
+	s.observedCalls = append(s.observedCalls, observedCall{
+		ctx:    ctx,
+		params: params,
+	})
 	return &api_v2.SamplingStrategyResponse{
 		StrategyType: api_v2.SamplingStrategyType_PROBABILISTIC,
 	}, nil

extension/jaegerremotesampling/go.mod

@@ -9,6 +9,8 @@ require (
 	go.opentelemetry.io/collector/config/configgrpc v0.82.0
 	go.opentelemetry.io/collector/config/confighttp v0.82.0
 	go.opentelemetry.io/collector/config/confignet v0.82.0
+	go.opentelemetry.io/collector/config/configopaque v0.82.0
+	go.opentelemetry.io/collector/config/configtls v0.82.0
 	go.opentelemetry.io/collector/confmap v0.82.0
 	go.opentelemetry.io/collector/extension v0.82.0
 	go.uber.org/zap v1.24.0
@@ -51,9 +53,7 @@ require (
 	go.opentelemetry.io/collector v0.82.0 // indirect
 	go.opentelemetry.io/collector/config/configauth v0.82.0 // indirect
 	go.opentelemetry.io/collector/config/configcompression v0.82.0 // indirect
-	go.opentelemetry.io/collector/config/configopaque v0.82.0 // indirect
 	go.opentelemetry.io/collector/config/configtelemetry v0.82.0 // indirect
-	go.opentelemetry.io/collector/config/configtls v0.82.0 // indirect
 	go.opentelemetry.io/collector/config/internal v0.82.0 // indirect
 	go.opentelemetry.io/collector/extension/auth v0.82.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.0.0-rcv0014 // indirect

extension/jaegerremotesampling/internal/remote_strategy_store.go

@@ -0,0 +1,48 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package internal // import "github.com/open-telemetry/opentelemetry-collector-contrib/extension/jaegerremotesampling/internal"
+
+import (
+	"context"
+
+	grpcstore "github.com/jaegertracing/jaeger/cmd/agent/app/configmanager/grpc"
+	"github.com/jaegertracing/jaeger/cmd/collector/app/sampling/strategystore"
+	"github.com/jaegertracing/jaeger/thrift-gen/sampling"
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/config/configopaque"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+)
+
+type grpcRemoteStrategyStore struct {
+	headerAdditions map[string]configopaque.String
+	delegate        *grpcstore.SamplingManager
+}
+
+// NewRemoteStrategyStore returns a StrategyStore that delegates to the configured Jaeger gRPC endpoint, making
+// extension-configured enhancements (header additions only for now) to the gRPC context of every outbound gRPC call.
+// Note: it would be nice to expand the configuration surface to include an optional TTL-based caching behavior
+// for service-specific outbound GetSamplingStrategy calls.
+func NewRemoteStrategyStore(
+	conn *grpc.ClientConn,
+	grpcClientSettings *configgrpc.GRPCClientSettings,
+) strategystore.StrategyStore {
+	return &grpcRemoteStrategyStore{
+		headerAdditions: grpcClientSettings.Headers,
+		delegate:        grpcstore.NewConfigManager(conn),
+	}
+}
+
+func (g *grpcRemoteStrategyStore) GetSamplingStrategy(ctx context.Context, serviceName string) (*sampling.SamplingStrategyResponse, error) {
+	return g.delegate.GetSamplingStrategy(g.enhanceContext(ctx), serviceName)
+}
+
+// This function is used to add the extension configuration defined HTTP headers to a given outbound gRPC call's context.
+func (g *grpcRemoteStrategyStore) enhanceContext(ctx context.Context) context.Context {
+	md := metadata.New(nil)
+	for k, v := range g.headerAdditions {
+		md.Set(k, string(v))
+	}
+	return metadata.NewOutgoingContext(ctx, md)
+}