[exporter/otlp] Report runtime status

.chloggen/exp-status.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# 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. otlpreceiver)
+component: otlpexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add runtime status reporting for OTLP exporters.
+
+# One or more tracking issues or pull requests related to the change
+issues: [9957]
+
+# (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:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

exporter/otlpexporter/go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector v0.111.0
 	go.opentelemetry.io/collector/component v0.111.0
+	go.opentelemetry.io/collector/component/componentstatus v0.111.0
 	go.opentelemetry.io/collector/config/configauth v0.111.0
 	go.opentelemetry.io/collector/config/configcompression v1.17.0
 	go.opentelemetry.io/collector/config/configgrpc v0.111.0

exporter/otlpexporter/otlp.go

@@ -17,6 +17,7 @@ import (
 	"google.golang.org/grpc/status"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componentstatus"
 	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/exporter"
@@ -41,6 +42,7 @@ type baseExporter struct {
 	metadata       metadata.MD
 	callOptions    []grpc.CallOption
 
+	host     component.Host
 	settings component.TelemetrySettings
 
 	// Default user-agent header.
@@ -74,6 +76,7 @@ func (e *baseExporter) start(ctx context.Context, host component.Host) (err erro
 	e.callOptions = []grpc.CallOption{
 		grpc.WaitForReady(e.config.ClientConfig.WaitForReady),
 	}
+	e.host = host
 
 	return
 }
@@ -85,11 +88,14 @@ func (e *baseExporter) shutdown(context.Context) error {
 	return nil
 }
 
-func (e *baseExporter) pushTraces(ctx context.Context, td ptrace.Traces) error {
+func (e *baseExporter) pushTraces(ctx context.Context, td ptrace.Traces) (err error) {
+	defer func() {
+		e.reportStatusFromError(err)
+	}()
 	req := ptraceotlp.NewExportRequestFromTraces(td)
 	resp, respErr := e.traceExporter.Export(e.enhanceContext(ctx), req, e.callOptions...)
-	if err := processError(respErr); err != nil {
-		return err
+	if err = e.processError(respErr); err != nil {
+		return
 	}
 	partialSuccess := resp.PartialSuccess()
 	if !(partialSuccess.ErrorMessage() == "" && partialSuccess.RejectedSpans() == 0) {
@@ -98,14 +104,17 @@ func (e *baseExporter) pushTraces(ctx context.Context, td ptrace.Traces) error {
 			zap.Int64("dropped_spans", resp.PartialSuccess().RejectedSpans()),
 		)
 	}
-	return nil
+	return
 }
 
-func (e *baseExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) error {
+func (e *baseExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) (err error) {
+	defer func() {
+		e.reportStatusFromError(err)
+	}()
 	req := pmetricotlp.NewExportRequestFromMetrics(md)
 	resp, respErr := e.metricExporter.Export(e.enhanceContext(ctx), req, e.callOptions...)
-	if err := processError(respErr); err != nil {
-		return err
+	if err = e.processError(respErr); err != nil {
+		return
 	}
 	partialSuccess := resp.PartialSuccess()
 	if !(partialSuccess.ErrorMessage() == "" && partialSuccess.RejectedDataPoints() == 0) {
@@ -114,14 +123,17 @@ func (e *baseExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) erro
 			zap.Int64("dropped_data_points", resp.PartialSuccess().RejectedDataPoints()),
 		)
 	}
-	return nil
+	return
 }
 
-func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
+func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) (err error) {
+	defer func() {
+		e.reportStatusFromError(err)
+	}()
 	req := plogotlp.NewExportRequestFromLogs(ld)
 	resp, respErr := e.logExporter.Export(e.enhanceContext(ctx), req, e.callOptions...)
-	if err := processError(respErr); err != nil {
-		return err
+	if err = e.processError(respErr); err != nil {
+		return
 	}
 	partialSuccess := resp.PartialSuccess()
 	if !(partialSuccess.ErrorMessage() == "" && partialSuccess.RejectedLogRecords() == 0) {
@@ -130,7 +142,7 @@ func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
 			zap.Int64("dropped_log_records", resp.PartialSuccess().RejectedLogRecords()),
 		)
 	}
-	return nil
+	return
 }
 
 func (e *baseExporter) enhanceContext(ctx context.Context) context.Context {
@@ -140,7 +152,7 @@ func (e *baseExporter) enhanceContext(ctx context.Context) context.Context {
 	return ctx
 }
 
-func processError(err error) error {
+func (e *baseExporter) processError(err error) error {
 	if err == nil {
 		// Request is successful, we are done.
 		return nil
@@ -154,6 +166,10 @@ func processError(err error) error {
 	}
 
 	// Now, this is a real error.
+	if isComponentPermanentError(st) {
+		componentstatus.ReportStatus(e.host, componentstatus.NewPermanentErrorEvent(err))
+	}
+
 	retryInfo := getRetryInfo(st)
 
 	if !shouldRetry(st.Code(), retryInfo) {
@@ -172,6 +188,14 @@ func processError(err error) error {
 	return err
 }
 
+func (e *baseExporter) reportStatusFromError(err error) {
+	if err != nil {
+		componentstatus.ReportStatus(e.host, componentstatus.NewRecoverableErrorEvent(err))
+		return
+	}
+	componentstatus.ReportStatus(e.host, componentstatus.NewEvent(componentstatus.StatusOK))
+}
+
 func shouldRetry(code codes.Code, retryInfo *errdetails.RetryInfo) bool {
 	switch code {
 	case codes.Canceled,
@@ -209,3 +233,21 @@ func getThrottleDuration(t *errdetails.RetryInfo) time.Duration {
 	}
 	return 0
 }
+
+// A component status of PermanentError indicates the component is in a state that will require user
+// intervention to fix. Typically this is a misconfiguration detected at runtime. A component
+// PermanentError has different semantics than a consumererror. For more information, see:
+// https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-status.md
+func isComponentPermanentError(st *status.Status) bool {
+	switch st.Code() {
+	case codes.NotFound:
+		return true
+	case codes.PermissionDenied:
+		return true
+	case codes.Unauthenticated:
+		return true
+	default:
+		return false
+	}
+
+}

exporter/otlpexporter/otlp_test.go

@@ -25,6 +25,8 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/durationpb"
 
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componentstatus"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/configopaque"
@@ -795,3 +797,171 @@ func TestSendLogData(t *testing.T) {
 	assert.Len(t, observed.FilterLevelExact(zap.WarnLevel).All(), 1)
 	assert.Contains(t, observed.FilterLevelExact(zap.WarnLevel).All()[0].Message, "Partial success")
 }
+
+func TestComponentStatus(t *testing.T) {
+	tests := []struct {
+		name            string
+		exportError     error
+		componentStatus componentstatus.Status
+	}{
+		{
+			name:            "No Error",
+			exportError:     nil,
+			componentStatus: componentstatus.StatusOK,
+		},
+		{
+			name:            "Permission Denied",
+			exportError:     status.Error(codes.PermissionDenied, "permission denied"),
+			componentStatus: componentstatus.StatusPermanentError,
+		},
+		{
+			name:            "Not Found",
+			exportError:     status.Error(codes.NotFound, "not found"),
+			componentStatus: componentstatus.StatusPermanentError,
+		},
+		{
+			name:            "Unauthenticated",
+			exportError:     status.Error(codes.Unauthenticated, "unauthenticated"),
+			componentStatus: componentstatus.StatusPermanentError,
+		},
+		{
+			name:            "Resource Exhausted",
+			exportError:     status.Error(codes.ResourceExhausted, "resource exhausted"),
+			componentStatus: componentstatus.StatusRecoverableError,
+		},
+	}
+
+	t.Run("traces", func(t *testing.T) {
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				ln, err := net.Listen("tcp", "localhost:")
+				require.NoError(t, err)
+				rcv, _ := otlpTracesReceiverOnGRPCServer(ln, false)
+				rcv.setExportError(tt.exportError)
+				defer rcv.srv.GracefulStop()
+
+				factory := NewFactory()
+				cfg := factory.CreateDefaultConfig().(*Config)
+				cfg.QueueConfig.Enabled = false
+				cfg.ClientConfig = configgrpc.ClientConfig{
+					Endpoint: ln.Addr().String(),
+					TLSSetting: configtls.ClientConfig{
+						Insecure: true,
+					},
+				}
+
+				set := exportertest.NewNopSettings()
+				host := &testHost{Host: componenttest.NewNopHost()}
+
+				exp, err := factory.CreateTracesExporter(context.Background(), set, cfg)
+				require.NoError(t, err)
+				require.NotNil(t, exp)
+				require.NoError(t, exp.Start(context.Background(), host))
+
+				defer func() {
+					assert.NoError(t, exp.Shutdown(context.Background()))
+				}()
+
+				td := ptrace.NewTraces()
+				err = exp.ConsumeTraces(context.Background(), td)
+
+				assert.Equal(t, tt.componentStatus != componentstatus.StatusOK, err != nil)
+				assert.Equal(t, tt.componentStatus, host.lastStatus)
+			})
+		}
+	})
+
+	t.Run("metrics", func(t *testing.T) {
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				ln, err := net.Listen("tcp", "localhost:")
+				require.NoError(t, err)
+
+				rcv := otlpMetricsReceiverOnGRPCServer(ln)
+				rcv.setExportError(tt.exportError)
+				defer rcv.srv.GracefulStop()
+
+				factory := NewFactory()
+				cfg := factory.CreateDefaultConfig().(*Config)
+				cfg.QueueConfig.Enabled = false
+				cfg.ClientConfig = configgrpc.ClientConfig{
+					Endpoint: ln.Addr().String(),
+					TLSSetting: configtls.ClientConfig{
+						Insecure: true,
+					},
+				}
+
+				set := exportertest.NewNopSettings()
+				host := &testHost{Host: componenttest.NewNopHost()}
+
+				exp, err := factory.CreateMetricsExporter(context.Background(), set, cfg)
+				require.NoError(t, err)
+				require.NotNil(t, exp)
+				require.NoError(t, exp.Start(context.Background(), host))
+
+				defer func() {
+					assert.NoError(t, exp.Shutdown(context.Background()))
+				}()
+
+				md := pmetric.NewMetrics()
+				err = exp.ConsumeMetrics(context.Background(), md)
+				assert.Equal(t, tt.componentStatus != componentstatus.StatusOK, err != nil)
+				assert.Equal(t, tt.componentStatus, host.lastStatus)
+			})
+		}
+	})
+
+	t.Run("logs", func(t *testing.T) {
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				ln, err := net.Listen("tcp", "localhost:")
+				require.NoError(t, err)
+
+				rcv := otlpLogsReceiverOnGRPCServer(ln)
+				rcv.setExportError(tt.exportError)
+				defer rcv.srv.GracefulStop()
+
+				factory := NewFactory()
+				cfg := factory.CreateDefaultConfig().(*Config)
+				cfg.QueueConfig.Enabled = false
+				cfg.ClientConfig = configgrpc.ClientConfig{
+					Endpoint: ln.Addr().String(),
+					TLSSetting: configtls.ClientConfig{
+						Insecure: true,
+					},
+				}
+
+				set := exportertest.NewNopSettings()
+				host := &testHost{Host: componenttest.NewNopHost()}
+
+				exp, err := factory.CreateLogsExporter(context.Background(), set, cfg)
+				require.NoError(t, err)
+				require.NotNil(t, exp)
+				require.NoError(t, exp.Start(context.Background(), host))
+
+				defer func() {
+					assert.NoError(t, exp.Shutdown(context.Background()))
+				}()
+
+				ld := plog.NewLogs()
+				err = exp.ConsumeLogs(context.Background(), ld)
+				assert.Equal(t, tt.componentStatus != componentstatus.StatusOK, err != nil)
+				assert.Equal(t, tt.componentStatus, host.lastStatus)
+			})
+		}
+	})
+}
+
+var _ component.Host = (*testHost)(nil)
+var _ componentstatus.Reporter = (*testHost)(nil)
+
+type testHost struct {
+	component.Host
+	lastStatus componentstatus.Status
+}
+
+func (h *testHost) Report(ev *componentstatus.Event) {
+	if h.lastStatus != componentstatus.StatusPermanentError {
+		h.lastStatus = ev.Status()
+	}
+}

exporter/otlphttpexporter/go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector v0.111.0
 	go.opentelemetry.io/collector/component v0.111.0
+	go.opentelemetry.io/collector/component/componentstatus v0.111.0
 	go.opentelemetry.io/collector/config/configcompression v1.17.0
 	go.opentelemetry.io/collector/config/confighttp v0.111.0
 	go.opentelemetry.io/collector/config/configopaque v1.17.0