Instrument the otlptracegrpc exporter (#7459)

Resolve #7007 

### Benchmarks

```console
> benchstat inst-otlptracegrpc.bmark.result
goos: linux
goarch: amd64
pkg: go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc
cpu: Intel(R) Core(TM) i7-8550U CPU @ 1.80GHz
                                      │ inst-otlptracegrpc.bmark.result │
                                      │             sec/op              │
ExporterExportSpans/Observability-8                         144.3µ ± 4%
ExporterExportSpans/NoObservability-8                       147.3µ ± 4%
geomean                                                     145.8µ

                                      │ inst-otlptracegrpc.bmark.result │
                                      │              B/op               │
ExporterExportSpans/Observability-8                        23.07Ki ± 0%
ExporterExportSpans/NoObservability-8                      22.34Ki ± 0%
geomean                                                    22.70Ki

                                      │ inst-otlptracegrpc.bmark.result │
                                      │            allocs/op            │
ExporterExportSpans/Observability-8                          335.0 ± 0%
ExporterExportSpans/NoObservability-8                        331.0 ± 0%
geomean                                                      333.0
```

Author: MrAlias

CHANGELOG.md

@@ -17,6 +17,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Add experimental observability for the prometheus exporter in `go.opentelemetry.io/otel/exporters/prometheus`.
   Check the `go.opentelemetry.io/otel/exporters/prometheus/internal/x` package documentation for more information. (#7345)
 - Add experimental observability metrics in `go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc`. (#7353)
+- Add experimental observability metrics in `go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc`. (#7459)
 
 ### Fixed
 

exporters/otlp/otlptrace/otlptracegrpc/client.go

@@ -19,6 +19,8 @@ import (
 
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal/counter"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal/observ"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal/otlpconfig"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal/retry"
 )
@@ -44,6 +46,9 @@ type client struct {
 	conn    *grpc.ClientConn
 	tscMu   sync.RWMutex
 	tsc     coltracepb.TraceServiceClient
+
+	instID int64
+	inst   *observ.Instrumentation
 }
 
 // Compile time check *client implements otlptrace.Client.
@@ -67,6 +72,7 @@ func newClient(opts ...Option) *client {
 		stopCtx:       ctx,
 		stopFunc:      cancel,
 		conn:          cfg.GRPCConn,
+		instID:        counter.NextExporterID(),
 	}
 
 	if len(cfg.Traces.Headers) > 0 {
@@ -91,13 +97,24 @@ func (c *client) Start(context.Context) error {
 		c.conn = conn
 	}
 
+	// Initialize the instrumentation if not already done.
+	//
+	// Initialize here instead of NewClient to allow any errors to be passed
+	// back to the caller and so that any setup of the environment variables to
+	// enable instrumentation can be set via code.
+	var err error
+	if c.inst == nil {
+		target := c.conn.CanonicalTarget()
+		c.inst, err = observ.NewInstrumentation(c.instID, target)
+	}
+
 	// The otlptrace.Client interface states this method is called just once,
 	// so no need to check if already started.
 	c.tscMu.Lock()
 	c.tsc = coltracepb.NewTraceServiceClient(c.conn)
 	c.tscMu.Unlock()
 
-	return nil
+	return err
 }
 
 var errAlreadyStopped = errors.New("the client is already stopped")
@@ -188,6 +205,12 @@ func (c *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.Resourc
 	ctx, cancel := c.exportContext(ctx)
 	defer cancel()
 
+	var code codes.Code
+	if c.inst != nil {
+		op := c.inst.ExportSpans(ctx, len(protoSpans))
+		defer func() { op.End(uploadErr, code) }()
+	}
+
 	return c.requestFunc(ctx, func(iCtx context.Context) error {
 		resp, err := c.tsc.Export(iCtx, &coltracepb.ExportTraceServiceRequest{
 			ResourceSpans: protoSpans,
@@ -201,7 +224,8 @@ func (c *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.Resourc
 			}
 		}
 		// nil is converted to OK.
-		if status.Code(err) == codes.OK {
+		code = status.Code(err)
+		if code == codes.OK {
 			// Success.
 			return uploadErr
 		}

exporters/otlp/otlptrace/otlptracegrpc/client_test.go

@@ -20,17 +20,26 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/backoff"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 
+	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal/counter"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal/observ"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal/otlptracetest"
+	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+	"go.opentelemetry.io/otel/semconv/v1.37.0/otelconv"
 )
 
 func TestMain(m *testing.M) {
@@ -115,7 +124,7 @@ func TestWithEndpointURL(t *testing.T) {
 }
 
 func newGRPCExporter(
-	t *testing.T,
+	tb testing.TB,
 	ctx context.Context,
 	endpoint string,
 	additionalOpts ...otlptracegrpc.Option,
@@ -130,7 +139,7 @@ func newGRPCExporter(
 	client := otlptracegrpc.NewClient(opts...)
 	exp, err := otlptrace.New(ctx, client)
 	if err != nil {
-		t.Fatalf("failed to create a new collector exporter: %v", err)
+		tb.Fatalf("failed to create a new collector exporter: %v", err)
 	}
 	return exp
 }
@@ -430,3 +439,161 @@ func TestCustomUserAgent(t *testing.T) {
 	headers := mc.getHeaders()
 	require.Contains(t, headers.Get("user-agent")[0], customUserAgent)
 }
+
+func TestClientInstrumentation(t *testing.T) {
+	// Enable instrumentation for this test.
+	t.Setenv("OTEL_GO_X_OBSERVABILITY", "true")
+
+	// Reset client ID to be deterministic
+	const id = 0
+	counter.SetExporterID(id)
+
+	// Save original meter provider and restore at end of test.
+	orig := otel.GetMeterProvider()
+	t.Cleanup(func() { otel.SetMeterProvider(orig) })
+
+	// Create a new meter provider to capture metrics.
+	reader := metric.NewManualReader()
+	mp := metric.NewMeterProvider(metric.WithReader(reader))
+	otel.SetMeterProvider(mp)
+
+	const n, msg = 2, "partially successful"
+	mc := runMockCollectorWithConfig(t, &mockConfig{
+		endpoint: "localhost:0", // Determine canonical endpoint.
+		partial: &coltracepb.ExportTracePartialSuccess{
+			RejectedSpans: n,
+			ErrorMessage:  msg,
+		},
+	})
+	t.Cleanup(func() { require.NoError(t, mc.stop()) })
+
+	exp := newGRPCExporter(t, t.Context(), mc.endpoint)
+	err := exp.ExportSpans(t.Context(), roSpans)
+	assert.ErrorIs(t, err, internal.TracePartialSuccessError(n, msg))
+	require.NoError(t, exp.Shutdown(t.Context()))
+
+	var got metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(t.Context(), &got))
+
+	attrs := observ.BaseAttrs(id, canonical(t, mc.endpoint))
+	want := metricdata.ScopeMetrics{
+		Scope: instrumentation.Scope{
+			Name:      observ.ScopeName,
+			Version:   observ.Version,
+			SchemaURL: observ.SchemaURL,
+		},
+		Metrics: []metricdata.Metrics{
+			{
+				Name:        otelconv.SDKExporterSpanInflight{}.Name(),
+				Description: otelconv.SDKExporterSpanInflight{}.Description(),
+				Unit:        otelconv.SDKExporterSpanInflight{}.Unit(),
+				Data: metricdata.Sum[int64]{
+					DataPoints: []metricdata.DataPoint[int64]{
+						{Attributes: attribute.NewSet(attrs...)},
+					},
+					Temporality: metricdata.CumulativeTemporality,
+				},
+			},
+			{
+				Name:        otelconv.SDKExporterSpanExported{}.Name(),
+				Description: otelconv.SDKExporterSpanExported{}.Description(),
+				Unit:        otelconv.SDKExporterSpanExported{}.Unit(),
+				Data: metricdata.Sum[int64]{
+					DataPoints: []metricdata.DataPoint[int64]{
+						{Attributes: attribute.NewSet(attrs...)},
+						{Attributes: attribute.NewSet(append(
+							attrs,
+							otelconv.SDKExporterSpanExported{}.AttrErrorType("*errors.joinError"),
+						)...)},
+					},
+					Temporality: 0x1,
+					IsMonotonic: true,
+				},
+			},
+			{
+				Name:        otelconv.SDKExporterOperationDuration{}.Name(),
+				Description: otelconv.SDKExporterOperationDuration{}.Description(),
+				Unit:        otelconv.SDKExporterOperationDuration{}.Unit(),
+				Data: metricdata.Histogram[float64]{
+					DataPoints: []metricdata.HistogramDataPoint[float64]{
+						{Attributes: attribute.NewSet(append(
+							attrs,
+							otelconv.SDKExporterOperationDuration{}.AttrErrorType("*errors.joinError"),
+							otelconv.SDKExporterOperationDuration{}.AttrRPCGRPCStatusCode(
+								otelconv.RPCGRPCStatusCodeOk,
+							),
+						)...)},
+					},
+					Temporality: 0x1,
+				},
+			},
+		},
+	}
+	require.Len(t, got.ScopeMetrics, 1)
+	opt := []metricdatatest.Option{
+		metricdatatest.IgnoreTimestamp(),
+		metricdatatest.IgnoreExemplars(),
+		metricdatatest.IgnoreValue(),
+	}
+	metricdatatest.AssertEqual(t, want, got.ScopeMetrics[0], opt...)
+}
+
+func canonical(t *testing.T, endpoint string) string {
+	t.Helper()
+
+	opt := grpc.WithTransportCredentials(insecure.NewCredentials())
+	c, err := grpc.NewClient(endpoint, opt) // Used to normaliz endpoint.
+	if err != nil {
+		t.Fatalf("failed to create grpc client: %v", err)
+	}
+	out := c.CanonicalTarget()
+	_ = c.Close()
+
+	return out
+}
+
+func BenchmarkExporterExportSpans(b *testing.B) {
+	const n = 10
+
+	run := func(b *testing.B) {
+		mc := runMockCollectorWithConfig(b, &mockConfig{
+			endpoint: "localhost:0",
+			partial: &coltracepb.ExportTracePartialSuccess{
+				RejectedSpans: 5,
+				ErrorMessage:  "partially successful",
+			},
+		})
+		b.Cleanup(func() { require.NoError(b, mc.stop()) })
+
+		exp := newGRPCExporter(b, b.Context(), mc.endpoint)
+		b.Cleanup(func() {
+			//nolint:usetesting // required to avoid getting a canceled context at cleanup.
+			assert.NoError(b, exp.Shutdown(context.Background()))
+		})
+
+		stubs := make([]tracetest.SpanStub, n)
+		for i := range stubs {
+			stubs[i].Name = fmt.Sprintf("Span %d", i)
+		}
+		spans := tracetest.SpanStubs(stubs).Snapshots()
+
+		b.ReportAllocs()
+		b.ResetTimer()
+
+		var err error
+		for b.Loop() {
+			err = exp.ExportSpans(b.Context(), spans)
+		}
+		_ = err
+	}
+
+	b.Run("Observability", func(b *testing.B) {
+		b.Setenv("OTEL_GO_X_OBSERVABILITY", "true")
+		run(b)
+	})
+
+	b.Run("NoObservability", func(b *testing.B) {
+		b.Setenv("OTEL_GO_X_OBSERVABILITY", "false")
+		run(b)
+	})
+}

exporters/otlp/otlptrace/otlptracegrpc/internal/counter/counter.go

@@ -29,3 +29,6 @@ package internal // import "go.opentelemetry.io/otel/exporters/otlp/otlptrace/ot
 
 //go:generate gotmpl --body=../../../../../internal/shared/otlp/observ/target.go.tmpl "--data={ \"pkg\": \"observ\", \"pkg_path\": \"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal/observ\" }" --out=observ/target.go
 //go:generate gotmpl --body=../../../../../internal/shared/otlp/observ/target_test.go.tmpl "--data={ \"pkg\": \"observ\" }" --out=observ/target_test.go
+
+//go:generate gotmpl --body=../../../../../internal/shared/counter/counter.go.tmpl "--data={ \"pkg\": \"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc/internal/counter\" }" --out=counter/counter.go
+//go:generate gotmpl --body=../../../../../internal/shared/counter/counter_test.go.tmpl "--data={}" --out=counter/counter_test.go