add self observability for stdout exporter

CHANGELOG.md

@@ -41,6 +41,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Add `go.opentelemetry.io/otel/semconv/v1.41.0` package.
   The package contains semantic conventions from the `v1.41.0` version of the OpenTelemetry Semantic Conventions.
   See the [migration documentation](./semconv/v1.41.0/MIGRATION.md) for information on how to upgrade from `go.opentelemetry.io/otel/semconv/v1.40.0`. (#8324)
+- Add experimental self-observability metrics in `go.opentelemetry.io/otel/exporters/stdout/stdoutlog`. (#8263)
 - Add Observable variants of instruments to `go.opentelemetry.io/otel/semconv/v1.41.0` package. (#8350)
 - Generate explicit histogram bucket boundaries from weaver configuration for HTTP and RPC duration instruments in `go.opentelemetry.io/otel/semconv/v1.41.0`. (#8002)
 

exporters/stdout/stdoutlog/exporter.go

@@ -8,6 +8,9 @@ import (
 	"encoding/json"
 	"sync/atomic"
 
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutlog/internal/counter"
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutlog/internal/observ"
+
 	"go.opentelemetry.io/otel/sdk/log"
 )
 
@@ -18,6 +21,7 @@ var _ log.Exporter = &Exporter{}
 type Exporter struct {
 	encoder    atomic.Pointer[json.Encoder]
 	timestamps bool
+	inst       *observ.Instrumentation
 }
 
 // New creates an [Exporter].
@@ -29,21 +33,31 @@ func New(options ...Option) (*Exporter, error) {
 		enc.SetIndent("", "\t")
 	}
 
-	e := Exporter{
+	e := &Exporter{
 		timestamps: cfg.Timestamps,
 	}
 	e.encoder.Store(enc)
 
-	return &e, nil
+	var err error
+	e.inst, err = observ.NewInstrumentation(counter.NextExporterID())
+	return e, err
 }
 
 // Export exports log records to writer.
-func (e *Exporter) Export(ctx context.Context, records []log.Record) error {
+func (e *Exporter) Export(ctx context.Context, records []log.Record) (err error) {
 	enc := e.encoder.Load()
 	if enc == nil {
 		return nil
 	}
 
+	var success int64
+	if e.inst != nil {
+		op := e.inst.ExportLogs(ctx, int64(len(records)))
+		defer func() {
+			op.End(success, err)
+		}()
+	}
+
 	for _, record := range records {
 		// Honor context cancellation.
 		if err := ctx.Err(); err != nil {
@@ -55,6 +69,7 @@ func (e *Exporter) Export(ctx context.Context, records []log.Record) error {
 		if err := enc.Encode(recordJSON); err != nil {
 			return err
 		}
+		success++
 	}
 	return nil
 }

exporters/stdout/stdoutlog/exporter_test.go

@@ -7,19 +7,29 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutlog/internal"
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutlog/internal/counter"
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutlog/internal/observ"
 	"go.opentelemetry.io/otel/log"
 	"go.opentelemetry.io/otel/sdk/instrumentation"
 	sdklog "go.opentelemetry.io/otel/sdk/log"
 	"go.opentelemetry.io/otel/sdk/log/logtest"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
 	"go.opentelemetry.io/otel/sdk/resource"
+	semconv "go.opentelemetry.io/otel/semconv/v1.41.0"
+	"go.opentelemetry.io/otel/semconv/v1.41.0/otelconv"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -458,3 +468,216 @@ func TestValueMarshalJSON(t *testing.T) {
 		})
 	}
 }
+
+func TestObservability(t *testing.T) {
+	tests := []struct {
+		name    string
+		enabled bool
+		test    func(t *testing.T, scopeMetrics func() metricdata.ScopeMetrics)
+	}{
+		{
+			name:    "Disabled",
+			enabled: false,
+			test: func(t *testing.T, _ func() metricdata.ScopeMetrics) {
+				var buf bytes.Buffer
+				exporter, err := New(WithWriter(&buf))
+				require.NoError(t, err)
+				assert.Nil(t, exporter.inst)
+			},
+		},
+		{
+			name:    "upload success",
+			enabled: true,
+			test: func(t *testing.T, scopeMetrics func() metricdata.ScopeMetrics) {
+				ctx := t.Context()
+				var buf bytes.Buffer
+				exporter, err := New(WithWriter(&buf))
+				require.NoError(t, err)
+				err = exporter.Export(ctx, []sdklog.Record{getRecord(time.Now())})
+				require.NoError(t, err)
+				assertStdoutLogObservabilityMetrics(t, scopeMetrics(), 1, 1, nil)
+			},
+		},
+		{
+			name:    "upload failed",
+			enabled: true,
+			test: func(t *testing.T, scopeMetrics func() metricdata.ScopeMetrics) {
+				ctx := t.Context()
+				writeErr := errors.New("write failed")
+				exporter, err := New(WithWriter(&failingWriter{err: writeErr}))
+				require.NoError(t, err)
+				err = exporter.Export(ctx, []sdklog.Record{getRecord(time.Now())})
+				require.ErrorIs(t, err, writeErr)
+				assertStdoutLogObservabilityMetrics(t, scopeMetrics(), 1, 0, writeErr)
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.enabled {
+				t.Setenv("OTEL_GO_X_OBSERVABILITY", "true")
+				_ = counter.SetExporterID(0)
+			}
+			provider := otel.GetMeterProvider()
+			t.Cleanup(func() {
+				otel.SetMeterProvider(provider)
+			})
+			r := metric.NewManualReader()
+			mp := metric.NewMeterProvider(metric.WithReader(r))
+			otel.SetMeterProvider(mp)
+
+			scopeMetrics := func() metricdata.ScopeMetrics {
+				var got metricdata.ResourceMetrics
+				err := r.Collect(t.Context(), &got)
+				require.NoError(t, err)
+				require.Len(t, got.ScopeMetrics, 1)
+				return got.ScopeMetrics[0]
+			}
+			tc.test(t, scopeMetrics)
+		})
+	}
+}
+
+func BenchmarkExporterObservability(b *testing.B) {
+	ctx := b.Context()
+	rec := getRecord(time.Date(2020, 1, 1, 12, 0, 0, 0, time.UTC))
+	records := []sdklog.Record{rec}
+
+	b.Run("Disabled", func(b *testing.B) {
+		b.Setenv("OTEL_GO_X_OBSERVABILITY", "false")
+		var buf bytes.Buffer
+		exporter, err := New(WithWriter(&buf))
+		require.NoError(b, err)
+
+		b.ReportAllocs()
+		b.ResetTimer()
+		for b.Loop() {
+			buf.Reset()
+			_ = exporter.Export(ctx, records)
+		}
+	})
+
+	setupObservability := func(b *testing.B) {
+		b.Helper()
+		b.Setenv("OTEL_GO_X_OBSERVABILITY", "true")
+		_ = counter.SetExporterID(0)
+		provider := otel.GetMeterProvider()
+		b.Cleanup(func() {
+			otel.SetMeterProvider(provider)
+		})
+		r := metric.NewManualReader()
+		mp := metric.NewMeterProvider(metric.WithReader(r))
+		otel.SetMeterProvider(mp)
+	}
+
+	b.Run("UploadSuccess", func(b *testing.B) {
+		setupObservability(b)
+		var buf bytes.Buffer
+		exporter, err := New(WithWriter(&buf))
+		require.NoError(b, err)
+		b.ReportAllocs()
+		b.ResetTimer()
+		for b.Loop() {
+			buf.Reset()
+			_ = exporter.Export(ctx, records)
+		}
+	})
+}
+
+// failingWriter implements [io.Writer] and always returns an error.
+type failingWriter struct {
+	err error
+}
+
+func (w *failingWriter) Write([]byte) (int, error) {
+	return 0, w.err
+}
+
+func stdoutObservAttrSet(err error) attribute.Set {
+	attrs := []attribute.KeyValue{
+		semconv.OTelComponentName(observ.GetComponentName(0)),
+		semconv.OTelComponentTypeKey.String(observ.ComponentType),
+	}
+	if err != nil {
+		attrs = append(attrs, semconv.ErrorType(err))
+	}
+	return attribute.NewSet(attrs...)
+}
+
+func stdoutLogInflightMetric() metricdata.Metrics {
+	inflight := otelconv.SDKExporterLogInflight{}
+	return metricdata.Metrics{
+		Name:        inflight.Name(),
+		Description: inflight.Description(),
+		Unit:        inflight.Unit(),
+		Data: metricdata.Sum[int64]{
+			Temporality: metricdata.CumulativeTemporality,
+			DataPoints: []metricdata.DataPoint[int64]{
+				{Attributes: stdoutObservAttrSet(nil), Value: 0},
+			},
+		},
+	}
+}
+
+func stdoutLogExportedMetric(success, total int64, err error) metricdata.Metrics {
+	dp := []metricdata.DataPoint[int64]{
+		{Attributes: stdoutObservAttrSet(nil), Value: success},
+	}
+	if err != nil {
+		dp = append(dp, metricdata.DataPoint[int64]{
+			Attributes: stdoutObservAttrSet(err),
+			Value:      total - success,
+		})
+	}
+	exported := otelconv.SDKExporterLogExported{}
+	return metricdata.Metrics{
+		Name:        exported.Name(),
+		Description: exported.Description(),
+		Unit:        exported.Unit(),
+		Data: metricdata.Sum[int64]{
+			Temporality: metricdata.CumulativeTemporality,
+			IsMonotonic: true,
+			DataPoints:  dp,
+		},
+	}
+}
+
+func stdoutLogDurationMetric(err error) metricdata.Metrics {
+	duration := otelconv.SDKExporterOperationDuration{}
+	return metricdata.Metrics{
+		Name:        duration.Name(),
+		Description: duration.Description(),
+		Unit:        duration.Unit(),
+		Data: metricdata.Histogram[float64]{
+			Temporality: metricdata.CumulativeTemporality,
+			DataPoints: []metricdata.HistogramDataPoint[float64]{
+				{Attributes: stdoutObservAttrSet(err)},
+			},
+		},
+	}
+}
+
+func assertStdoutLogObservabilityMetrics(
+	t *testing.T,
+	got metricdata.ScopeMetrics,
+	logs int64,
+	success int64,
+	err error,
+) {
+	t.Helper()
+	wantScope := instrumentation.Scope{
+		Name:      observ.ScopeName,
+		Version:   internal.Version,
+		SchemaURL: semconv.SchemaURL,
+	}
+	assert.Equal(t, wantScope, got.Scope)
+
+	m := got.Metrics
+	require.Len(t, m, 3)
+
+	o := metricdatatest.IgnoreTimestamp()
+	metricdatatest.AssertEqual(t, stdoutLogInflightMetric(), m[0], o)
+	metricdatatest.AssertEqual(t, stdoutLogExportedMetric(success, logs, err), m[1], o)
+	metricdatatest.AssertEqual(t, stdoutLogDurationMetric(err), m[2], metricdatatest.IgnoreValue(), o)
+}