Treat PartialSuccess as Success (#9260)

Changes the treatment of
[PartialSuccess](https://opentelemetry.io/docs/specs/otlp/#partial-success),
making them successful and logging a warning instead of returning an
error to the caller. These responses are meant to convey successful
receipt of valid data which could not be accepted for other reasons,
specifically to cover situations where the OpenTelemetry SDK and
Collector have done nothing wrong, specifically to avoid retries. While
the existing OTLP exporter returns a permanent error (also avoids
retries), it makes the situation look like a total failure when in fact
it is more nuanced.

As discussed in the tracking issue, it is a lot of work to propagate
these "partial" successes backwards in a pipeline, so the appropriate
simple way to handle these items is to return success.

In this PR, we log a warning. In a future PR, (IMO) as discussed in
open-telemetry/oteps#238, we should count the
spans/metrics/logs that are rejected in this way using a dedicated
outcome label.

**Link to tracking Issue:**
Part of #9243

**Testing:** Tests for the "partial success" warning have been added.

**Documentation:** PartialSuccess behavior was not documented. Given the
level of detail in the README, it feels appropriate to continue not
documenting, otherwise lots of new details should be added.

---------

Co-authored-by: Alex Boten <aboten@lightstep.com>

.chloggen/partialsuccess-is-success.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: bug_fix
+
+# 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: PartialSuccess is treated as success, logged as warning.
+
+# One or more tracking issues or pull requests related to the change
+issues: [9243]
+
+# (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: [user]

exporter/otlpexporter/go.mod

@@ -19,6 +19,7 @@ require (
 	go.opentelemetry.io/otel/metric v1.22.0
 	go.opentelemetry.io/otel/trace v1.22.0
 	go.uber.org/goleak v1.3.0
+	go.uber.org/zap v1.26.0
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20231127180814-3a041ad873d4
 	google.golang.org/grpc v1.61.0
 	google.golang.org/protobuf v1.32.0
@@ -72,7 +73,6 @@ require (
 	go.opentelemetry.io/otel/sdk/metric v1.22.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.26.0 // indirect
 	golang.org/x/net v0.20.0 // indirect
 	golang.org/x/sys v0.16.0 // indirect
 	golang.org/x/text v0.14.0 // indirect

exporter/otlpexporter/otlp.go

@@ -10,6 +10,7 @@ import (
 	"runtime"
 	"time"
 
+	"go.uber.org/zap"
 	"google.golang.org/genproto/googleapis/rpc/errdetails"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -97,7 +98,10 @@ func (e *baseExporter) pushTraces(ctx context.Context, td ptrace.Traces) error {
 	}
 	partialSuccess := resp.PartialSuccess()
 	if !(partialSuccess.ErrorMessage() == "" && partialSuccess.RejectedSpans() == 0) {
-		return consumererror.NewPermanent(fmt.Errorf("OTLP partial success: \"%s\" (%d rejected)", resp.PartialSuccess().ErrorMessage(), resp.PartialSuccess().RejectedSpans()))
+		e.settings.Logger.Warn("Partial success response",
+			zap.String("message", resp.PartialSuccess().ErrorMessage()),
+			zap.Int64("dropped_spans", resp.PartialSuccess().RejectedSpans()),
+		)
 	}
 	return nil
 }
@@ -110,7 +114,10 @@ func (e *baseExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) erro
 	}
 	partialSuccess := resp.PartialSuccess()
 	if !(partialSuccess.ErrorMessage() == "" && partialSuccess.RejectedDataPoints() == 0) {
-		return consumererror.NewPermanent(fmt.Errorf("OTLP partial success: \"%s\" (%d rejected)", resp.PartialSuccess().ErrorMessage(), resp.PartialSuccess().RejectedDataPoints()))
+		e.settings.Logger.Warn("Partial success response",
+			zap.String("message", resp.PartialSuccess().ErrorMessage()),
+			zap.Int64("dropped_data_points", resp.PartialSuccess().RejectedDataPoints()),
+		)
 	}
 	return nil
 }
@@ -123,7 +130,10 @@ func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
 	}
 	partialSuccess := resp.PartialSuccess()
 	if !(partialSuccess.ErrorMessage() == "" && partialSuccess.RejectedLogRecords() == 0) {
-		return consumererror.NewPermanent(fmt.Errorf("OTLP partial success: \"%s\" (%d rejected)", resp.PartialSuccess().ErrorMessage(), resp.PartialSuccess().RejectedLogRecords()))
+		e.settings.Logger.Warn("Partial success response",
+			zap.String("message", resp.PartialSuccess().ErrorMessage()),
+			zap.Int64("dropped_log_records", resp.PartialSuccess().RejectedLogRecords()),
+		)
 	}
 	return nil
 }

exporter/otlpexporter/otlp_test.go

@@ -15,6 +15,8 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zaptest/observer"
 	"google.golang.org/genproto/googleapis/rpc/errdetails"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -247,6 +249,11 @@ func TestSendTraces(t *testing.T) {
 	set := exportertest.NewNopCreateSettings()
 	set.BuildInfo.Description = "Collector"
 	set.BuildInfo.Version = "1.2.3test"
+
+	// For testing the "Partial success" warning.
+	logger, observed := observer.New(zap.DebugLevel)
+	set.TelemetrySettings.Logger = zap.New(logger)
+
 	exp, err := factory.CreateTracesExporter(context.Background(), set, cfg)
 	require.NoError(t, err)
 	require.NotNil(t, exp)
@@ -310,7 +317,9 @@ func TestSendTraces(t *testing.T) {
 	td = testdata.GenerateTraces(2)
 
 	err = exp.ConsumeTraces(context.Background(), td)
-	assert.Error(t, err)
+	assert.NoError(t, err)
+	assert.Len(t, observed.FilterLevelExact(zap.WarnLevel).All(), 1)
+	assert.Contains(t, observed.FilterLevelExact(zap.WarnLevel).All()[0].Message, "Partial success")
 }
 
 func TestSendTracesWhenEndpointHasHttpScheme(t *testing.T) {
@@ -412,6 +421,11 @@ func TestSendMetrics(t *testing.T) {
 	set := exportertest.NewNopCreateSettings()
 	set.BuildInfo.Description = "Collector"
 	set.BuildInfo.Version = "1.2.3test"
+
+	// For testing the "Partial success" warning.
+	logger, observed := observer.New(zap.DebugLevel)
+	set.TelemetrySettings.Logger = zap.New(logger)
+
 	exp, err := factory.CreateMetricsExporter(context.Background(), set, cfg)
 	require.NoError(t, err)
 	require.NotNil(t, exp)
@@ -484,7 +498,9 @@ func TestSendMetrics(t *testing.T) {
 
 	// Send two metrics.
 	md = testdata.GenerateMetrics(2)
-	assert.Error(t, exp.ConsumeMetrics(context.Background(), md))
+	assert.NoError(t, exp.ConsumeMetrics(context.Background(), md))
+	assert.Len(t, observed.FilterLevelExact(zap.WarnLevel).All(), 1)
+	assert.Contains(t, observed.FilterLevelExact(zap.WarnLevel).All()[0].Message, "Partial success")
 }
 
 func TestSendTraceDataServerDownAndUp(t *testing.T) {
@@ -699,6 +715,11 @@ func TestSendLogData(t *testing.T) {
 	set := exportertest.NewNopCreateSettings()
 	set.BuildInfo.Description = "Collector"
 	set.BuildInfo.Version = "1.2.3test"
+
+	// For testing the "Partial success" warning.
+	logger, observed := observer.New(zap.DebugLevel)
+	set.TelemetrySettings.Logger = zap.New(logger)
+
 	exp, err := factory.CreateLogsExporter(context.Background(), set, cfg)
 	require.NoError(t, err)
 	require.NotNil(t, exp)
@@ -770,5 +791,7 @@ func TestSendLogData(t *testing.T) {
 	ld = testdata.GenerateLogs(2)
 
 	err = exp.ConsumeLogs(context.Background(), ld)
-	assert.Error(t, err)
+	assert.NoError(t, err)
+	assert.Len(t, observed.FilterLevelExact(zap.WarnLevel).All(), 1)
+	assert.Contains(t, observed.FilterLevelExact(zap.WarnLevel).All()[0].Message, "Partial success")
 }

exporter/otlphttpexporter/otlp.go

@@ -92,7 +92,7 @@ func (e *baseExporter) pushTraces(ctx context.Context, td ptrace.Traces) error {
 		return consumererror.NewPermanent(err)
 	}
 
-	return e.export(ctx, e.tracesURL, request, tracesPartialSuccessHandler)
+	return e.export(ctx, e.tracesURL, request, e.tracesPartialSuccessHandler)
 }
 
 func (e *baseExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) error {
@@ -101,7 +101,7 @@ func (e *baseExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) erro
 	if err != nil {
 		return consumererror.NewPermanent(err)
 	}
-	return e.export(ctx, e.metricsURL, request, metricsPartialSuccessHandler)
+	return e.export(ctx, e.metricsURL, request, e.metricsPartialSuccessHandler)
 }
 
 func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
@@ -111,7 +111,7 @@ func (e *baseExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
 		return consumererror.NewPermanent(err)
 	}
 
-	return e.export(ctx, e.logsURL, request, logsPartialSuccessHandler)
+	return e.export(ctx, e.logsURL, request, e.logsPartialSuccessHandler)
 }
 
 func (e *baseExporter) export(ctx context.Context, url string, request []byte, partialSuccessHandler partialSuccessHandler) error {
@@ -259,7 +259,7 @@ func handlePartialSuccessResponse(resp *http.Response, partialSuccessHandler par
 
 type partialSuccessHandler func(bytes []byte, contentType string) error
 
-func tracesPartialSuccessHandler(protoBytes []byte, contentType string) error {
+func (e *baseExporter) tracesPartialSuccessHandler(protoBytes []byte, contentType string) error {
 	if contentType != protobufContentType {
 		return nil
 	}
@@ -270,12 +270,15 @@ func tracesPartialSuccessHandler(protoBytes []byte, contentType string) error {
 	}
 	partialSuccess := exportResponse.PartialSuccess()
 	if !(partialSuccess.ErrorMessage() == "" && partialSuccess.RejectedSpans() == 0) {
-		return consumererror.NewPermanent(fmt.Errorf("OTLP partial success: %s (%d rejected)", partialSuccess.ErrorMessage(), partialSuccess.RejectedSpans()))
+		e.logger.Warn("Partial success response",
+			zap.String("message", exportResponse.PartialSuccess().ErrorMessage()),
+			zap.Int64("dropped_spans", exportResponse.PartialSuccess().RejectedSpans()),
+		)
 	}
 	return nil
 }
 
-func metricsPartialSuccessHandler(protoBytes []byte, contentType string) error {
+func (e *baseExporter) metricsPartialSuccessHandler(protoBytes []byte, contentType string) error {
 	if contentType != protobufContentType {
 		return nil
 	}
@@ -286,12 +289,15 @@ func metricsPartialSuccessHandler(protoBytes []byte, contentType string) error {
 	}
 	partialSuccess := exportResponse.PartialSuccess()
 	if !(partialSuccess.ErrorMessage() == "" && partialSuccess.RejectedDataPoints() == 0) {
-		return consumererror.NewPermanent(fmt.Errorf("OTLP partial success: %s (%d rejected)", partialSuccess.ErrorMessage(), partialSuccess.RejectedDataPoints()))
+		e.logger.Warn("Partial success response",
+			zap.String("message", exportResponse.PartialSuccess().ErrorMessage()),
+			zap.Int64("dropped_data_points", exportResponse.PartialSuccess().RejectedDataPoints()),
+		)
 	}
 	return nil
 }
 
-func logsPartialSuccessHandler(protoBytes []byte, contentType string) error {
+func (e *baseExporter) logsPartialSuccessHandler(protoBytes []byte, contentType string) error {
 	if contentType != protobufContentType {
 		return nil
 	}
@@ -302,7 +308,10 @@ func logsPartialSuccessHandler(protoBytes []byte, contentType string) error {
 	}
 	partialSuccess := exportResponse.PartialSuccess()
 	if !(partialSuccess.ErrorMessage() == "" && partialSuccess.RejectedLogRecords() == 0) {
-		return consumererror.NewPermanent(fmt.Errorf("OTLP partial success: %s (%d rejected)", partialSuccess.ErrorMessage(), partialSuccess.RejectedLogRecords()))
+		e.logger.Warn("Partial success response",
+			zap.String("message", exportResponse.PartialSuccess().ErrorMessage()),
+			zap.Int64("dropped_log_records", exportResponse.PartialSuccess().RejectedLogRecords()),
+		)
 	}
 	return nil
 }