diff --git a/pkg/chipingress/README.md b/pkg/chipingress/README.md index 5c45c1a33a..c39440cab9 100644 --- a/pkg/chipingress/README.md +++ b/pkg/chipingress/README.md @@ -130,6 +130,8 @@ client, err := chipingress.NewClient("example.com:9090", chipingress.WithTracerProvider(tracerProvider)) ``` +Use `chipingress.WithClientName("my_service")` to identify the client: the name is sent as the gRPC user-agent and recorded as a `client_name` attribute on the otelgrpc spans and metrics. + The client uses `otelgrpc.NewClientHandler()` to automatically create spans for all gRPC calls, including metrics for request duration, message sizes, and error rates. ### Retries diff --git a/pkg/chipingress/client.go b/pkg/chipingress/client.go index 344981e440..1334f4fedc 100644 --- a/pkg/chipingress/client.go +++ b/pkg/chipingress/client.go @@ -9,6 +9,7 @@ import ( "github.com/google/uuid" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/trace" "google.golang.org/grpc" @@ -57,6 +58,7 @@ type clientConfig struct { meterProvider metric.MeterProvider tracerProvider trace.TracerProvider nopInfoHeaderProvider HeaderProvider + clientName string retryPolicy *RetryPolicy } @@ -98,6 +100,14 @@ func NewClient(address string, opts ...Opt) (Client, error) { otelOpts = append(otelOpts, otelgrpc.WithTracerProvider(cfg.tracerProvider)) } + if cfg.clientName != "" { + clientNameAttr := attribute.String("client_name", cfg.clientName) + otelOpts = append(otelOpts, + otelgrpc.WithSpanAttributes(clientNameAttr), + otelgrpc.WithMetricAttributes(clientNameAttr), + ) + } + grpcOpts := []grpc.DialOption{ grpc.WithTransportCredentials(cfg.transportCredentials), grpc.WithStatsHandler(otelgrpc.NewClientHandler(otelOpts...)), @@ -108,6 +118,9 @@ func NewClient(address string, opts ...Opt) (Client, error) { PermitWithoutStream: true, }), } + if cfg.clientName != "" { + grpcOpts = append(grpcOpts, grpc.WithUserAgent(cfg.clientName)) + } // Retries are opt-in (see WithRetryPolicy): they can replay non-idempotent RPCs such as // Publish, so they are disabled unless the caller enables them. The service config is built // from typed structs (see retry_policy.go) rather than a hand-written JSON literal - a @@ -264,6 +277,13 @@ func WithMeterProvider(provider metric.MeterProvider) Opt { return func(c *clientConfig) { c.meterProvider = provider } } +// WithClientName identifies this client to chip-ingress and in telemetry: it is sent as the gRPC +// user-agent and recorded as a client_name attribute on the otelgrpc spans and metrics. An empty +// name is a no-op. +func WithClientName(name string) Opt { + return func(c *clientConfig) { c.clientName = name } +} + // WithTracerProvider sets a custom OpenTelemetry TracerProvider for distributed tracing. // If not set, the global tracer provider will be used. func WithTracerProvider(provider trace.TracerProvider) Opt { diff --git a/pkg/chipingress/client_test.go b/pkg/chipingress/client_test.go index a50f6f9bd8..9b90393a48 100644 --- a/pkg/chipingress/client_test.go +++ b/pkg/chipingress/client_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "net" + "strings" "testing" "time" @@ -641,6 +642,40 @@ func TestOptions(t *testing.T) { require.NoError(t, err) assert.Equal(t, "true", headers["x-include-nop-info"]) }) + + t.Run("WithClientName", func(t *testing.T) { + config := defaultCfg + WithClientName("durable_emitter")(&config) + assert.Equal(t, "durable_emitter", config.clientName) + }) +} + +// WithClientName must reach the server as part of the gRPC user-agent, and an unset name must +// leave the default user-agent untouched. +func TestClient_ClientNameUserAgent(t *testing.T) { + userAgent := func(t *testing.T, opts ...Opt) string { + lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + defer lis.Close() + + srv := gp.NewServer() + capture := &capturingServer{} + pb.RegisterChipIngressServer(srv, capture) + go func() { _ = srv.Serve(lis) }() + defer srv.Stop() + + client, err := NewClient(lis.Addr().String(), append([]Opt{WithInsecureConnection()}, opts...)...) + require.NoError(t, err) + defer client.Close() //nolint:errcheck + + _, err = client.Ping(t.Context(), &EmptyRequest{}) + require.NoError(t, err) + require.NotNil(t, capture.lastMD) + return strings.Join(capture.lastMD.Get("user-agent"), " ") + } + + assert.Contains(t, userAgent(t, WithClientName("durable_emitter")), "durable_emitter") + assert.NotContains(t, userAgent(t, WithClientName("")), "durable_emitter") } func TestHeaderInterceptor(t *testing.T) {