[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/factory.go

@@ -59,7 +59,7 @@ func createTraces(
 	oce := newExporter(cfg, set)
 	oCfg := cfg.(*Config)
 	return exporterhelper.NewTraces(ctx, set, cfg,
-		oce.pushTraces,
+		oce.pushTracesWithStatus,
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
 		exporterhelper.WithTimeout(oCfg.TimeoutConfig),
 		exporterhelper.WithRetry(oCfg.RetryConfig),
@@ -78,7 +78,7 @@ func createMetrics(
 	oce := newExporter(cfg, set)
 	oCfg := cfg.(*Config)
 	return exporterhelper.NewMetrics(ctx, set, cfg,
-		oce.pushMetrics,
+		oce.pushMetricsWithStatus,
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
 		exporterhelper.WithTimeout(oCfg.TimeoutConfig),
 		exporterhelper.WithRetry(oCfg.RetryConfig),
@@ -97,7 +97,7 @@ func createLogs(
 	oce := newExporter(cfg, set)
 	oCfg := cfg.(*Config)
 	return exporterhelper.NewLogs(ctx, set, cfg,
-		oce.pushLogs,
+		oce.pushLogsWithStatus,
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
 		exporterhelper.WithTimeout(oCfg.TimeoutConfig),
 		exporterhelper.WithRetry(oCfg.RetryConfig),
@@ -116,7 +116,7 @@ func createProfilesExporter(
 	oce := newExporter(cfg, set)
 	oCfg := cfg.(*Config)
 	return exporterhelperprofiles.NewProfilesExporter(ctx, set, cfg,
-		oce.pushProfiles,
+		oce.pushProfilesWithStatus,
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
 		exporterhelper.WithTimeout(oCfg.TimeoutConfig),
 		exporterhelper.WithRetry(oCfg.RetryConfig),

exporter/otlpexporter/go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector v0.114.0
 	go.opentelemetry.io/collector/component v0.114.0
+	go.opentelemetry.io/collector/component/componentstatus v0.114.0
 	go.opentelemetry.io/collector/component/componenttest v0.114.0
 	go.opentelemetry.io/collector/config/configauth v0.114.0
 	go.opentelemetry.io/collector/config/configcompression v1.20.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"
@@ -44,6 +45,7 @@ type baseExporter struct {
 	metadata        metadata.MD
 	callOptions     []grpc.CallOption
 
+	host     component.Host
 	settings component.TelemetrySettings
 
 	// Default user-agent header.
@@ -78,6 +80,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
 }
@@ -105,6 +108,15 @@ func (e *baseExporter) pushTraces(ctx context.Context, td ptrace.Traces) error {
 	return nil
 }
 
+func (e *baseExporter) pushTracesWithStatus(ctx context.Context, td ptrace.Traces) error {
+	if err := e.pushTraces(ctx, td); err != nil {
+		componentstatus.ReportStatus(e.host, componentstatus.NewRecoverableErrorEvent(err))
+		return err
+	}
+	componentstatus.ReportStatus(e.host, componentstatus.NewEvent(componentstatus.StatusOK))
+	return nil
+}
+
 func (e *baseExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) error {
 	req := pmetricotlp.NewExportRequestFromMetrics(md)
 	resp, respErr := e.metricExporter.Export(e.enhanceContext(ctx), req, e.callOptions...)
@@ -121,6 +133,15 @@ func (e *baseExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) erro
 	return nil
 }
 
+func (e *baseExporter) pushMetricsWithStatus(ctx context.Context, md pmetric.Metrics) error {
+	if err := e.pushMetrics(ctx, md); err != nil {
+		componentstatus.ReportStatus(e.host, componentstatus.NewRecoverableErrorEvent(err))
+		return err
+	}
+	componentstatus.ReportStatus(e.host, componentstatus.NewEvent(componentstatus.StatusOK))
+	return nil
+}
+
 func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
 	req := plogotlp.NewExportRequestFromLogs(ld)
 	resp, respErr := e.logExporter.Export(e.enhanceContext(ctx), req, e.callOptions...)
@@ -137,6 +158,15 @@ func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
 	return nil
 }
 
+func (e *baseExporter) pushLogsWithStatus(ctx context.Context, ld plog.Logs) error {
+	if err := e.pushLogs(ctx, ld); err != nil {
+		componentstatus.ReportStatus(e.host, componentstatus.NewRecoverableErrorEvent(err))
+		return err
+	}
+	componentstatus.ReportStatus(e.host, componentstatus.NewEvent(componentstatus.StatusOK))
+	return nil
+}
+
 func (e *baseExporter) pushProfiles(ctx context.Context, td pprofile.Profiles) error {
 	req := pprofileotlp.NewExportRequestFromProfiles(td)
 	resp, respErr := e.profileExporter.Export(e.enhanceContext(ctx), req, e.callOptions...)
@@ -153,6 +183,15 @@ func (e *baseExporter) pushProfiles(ctx context.Context, td pprofile.Profiles) e
 	return nil
 }
 
+func (e *baseExporter) pushProfilesWithStatus(ctx context.Context, td pprofile.Profiles) error {
+	if err := e.pushProfiles(ctx, td); err != nil {
+		componentstatus.ReportStatus(e.host, componentstatus.NewRecoverableErrorEvent(err))
+		return err
+	}
+	componentstatus.ReportStatus(e.host, componentstatus.NewEvent(componentstatus.StatusOK))
+	return nil
+}
+
 func (e *baseExporter) enhanceContext(ctx context.Context) context.Context {
 	if e.metadata.Len() > 0 {
 		return metadata.NewOutgoingContext(ctx, e.metadata)
@@ -173,7 +212,6 @@ func processError(err error) error {
 		return nil
 	}
 
-	// Now, this is a real error.
 	retryInfo := getRetryInfo(st)
 
 	if !shouldRetry(st.Code(), retryInfo) {

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"
@@ -1034,3 +1036,196 @@ func TestSendProfilesWhenEndpointHasHttpScheme(t *testing.T) {
 		})
 	}
 }
+
+func TestComponentStatus(t *testing.T) {
+	tests := []struct {
+		name            string
+		exportError     error
+		componentStatus componentstatus.Status
+	}{
+		{
+			name:            "No Error",
+			exportError:     nil,
+			componentStatus: componentstatus.StatusOK,
+		},
+		{
+			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.CreateTraces(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.CreateMetrics(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.CreateLogs(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)
+			})
+		}
+	})
+
+	t.Run("profiles", 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, _ := otlpProfilesReceiverOnGRPCServer(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.(exporterprofiles.Factory).CreateProfiles(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()))
+				}()
+
+				pd := pprofile.NewProfiles()
+				err = exp.ConsumeProfiles(context.Background(), pd)
+				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()
+	}
+}