Handle partial-success responses for OTLP trace

CHANGELOG.md

@@ -12,6 +12,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 - Support Go 1.19.
   Include compatibility testing and document support. (#3077)
+- Support the OTLP ExportTracePartialSuccess and ExportMetricsPartialSuccess
+  responses; these are passed to the registered error handler. (#3106)
 - Upgrade go.opentelemetry.io/proto/otlp from v0.18.0 to v0.19.0 (#3107)
 
 ### Changed

exporters/jaeger/agent_test.go

@@ -4,7 +4,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-//     http://www.apache.org/licenses/LICENSE-2.0
+//	http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

exporters/jaeger/reconnecting_udp_client_test.go

@@ -4,7 +4,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-//     http://www.apache.org/licenses/LICENSE-2.0
+//	http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

exporters/otlp/otlpmetric/otlpmetricgrpc/client.go

@@ -26,6 +26,8 @@ import (
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/otlp"
 	"go.opentelemetry.io/otel/exporters/otlp/internal/retry"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/internal/otlpconfig"
@@ -196,9 +198,16 @@ func (c *client) UploadMetrics(ctx context.Context, protoMetrics *metricpb.Resou
 	defer cancel()
 
 	return c.requestFunc(ctx, func(iCtx context.Context) error {
-		_, err := c.msc.Export(iCtx, &colmetricpb.ExportMetricsServiceRequest{
+		resp, err := c.msc.Export(iCtx, &colmetricpb.ExportMetricsServiceRequest{
 			ResourceMetrics: []*metricpb.ResourceMetrics{protoMetrics},
 		})
+		if resp != nil && resp.PartialSuccess != nil {
+			otel.Handle(otlp.PartialSuccessToError(
+				otlp.MetricsPartialSuccess,
+				resp.PartialSuccess.RejectedDataPoints,
+				resp.PartialSuccess.ErrorMessage,
+			))
+		}
 		// nil is converted to OK.
 		if status.Code(err) == codes.OK {
 			// Success.

exporters/otlp/otlpmetric/otlpmetricgrpc/client_test.go

@@ -29,10 +29,12 @@ import (
 	"google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/grpc/status"
 
+	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/internal/otlpmetrictest"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
 	"go.opentelemetry.io/otel/sdk/resource"
+	collectormetricpb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
 )
 
 var (
@@ -329,3 +331,28 @@ func TestFailedMetricTransform(t *testing.T) {
 
 	assert.Error(t, exp.Export(ctx, testResource, otlpmetrictest.FailReader{}))
 }
+
+func TestNewExporterWithPartialSuccess(t *testing.T) {
+	mc := runMockCollectorWithConfig(t, &mockConfig{
+		partial: &collectormetricpb.ExportMetricsPartialSuccess{
+			RejectedDataPoints: 2,
+			ErrorMessage:       "partially successful",
+		},
+	})
+	defer func() {
+		_ = mc.stop()
+	}()
+	errors := new([]error)
+	otel.SetErrorHandler(otel.ErrorHandlerFunc(func(err error) {
+		*errors = append(*errors, err)
+	}))
+	ctx := context.Background()
+	exp := newGRPCExporter(t, ctx, mc.endpoint)
+	require.NoError(t, exp.Export(ctx, testResource, oneRecord))
+	defer func() {
+		_ = exp.Shutdown(ctx)
+	}()
+	require.Equal(t, 1, len(*errors))
+	require.Contains(t, (*errors)[0].Error(), "partially successful")
+	require.Contains(t, (*errors)[0].Error(), "2 metric data points rejected")
+}

exporters/otlp/otlpmetric/otlpmetricgrpc/mock_collector_test.go

@@ -36,6 +36,7 @@ func makeMockCollector(t *testing.T, mockConfig *mockConfig) *mockCollector {
 		metricSvc: &mockMetricService{
 			storage: otlpmetrictest.NewMetricsStorage(),
 			errors:  mockConfig.errors,
+			partial: mockConfig.partial,
 		},
 	}
 }
@@ -45,6 +46,7 @@ type mockMetricService struct {
 
 	requests int
 	errors   []error
+	partial  *collectormetricpb.ExportMetricsPartialSuccess
 
 	headers metadata.MD
 	mu      sync.RWMutex
@@ -75,7 +77,9 @@ func (mms *mockMetricService) Export(ctx context.Context, exp *collectormetricpb
 		mms.mu.Unlock()
 	}()
 
-	reply := &collectormetricpb.ExportMetricsServiceResponse{}
+	reply := &collectormetricpb.ExportMetricsServiceResponse{
+		PartialSuccess: mms.partial,
+	}
 	if mms.requests < len(mms.errors) {
 		idx := mms.requests
 		return reply, mms.errors[idx]
@@ -99,6 +103,7 @@ type mockCollector struct {
 type mockConfig struct {
 	errors   []error
 	endpoint string
+	partial  *collectormetricpb.ExportMetricsPartialSuccess
 }
 
 var _ collectormetricpb.MetricsServiceServer = (*mockMetricService)(nil)

exporters/otlp/otlpmetric/otlpmetrichttp/client.go

@@ -29,6 +29,8 @@ import (
 
 	"google.golang.org/protobuf/proto"
 
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/otlp"
 	"go.opentelemetry.io/otel/exporters/otlp/internal/retry"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/internal/otlpconfig"
@@ -170,6 +172,26 @@ func (d *client) UploadMetrics(ctx context.Context, protoMetrics *metricpb.Resou
 			rErr = fmt.Errorf("failed to send %s to %s: %s", d.name, request.URL, resp.Status)
 		}
 
+		// Read the partial success message, if any.
+		var respData bytes.Buffer
+		_, ioerr := io.Copy(&respData, resp.Body)
+		if ioerr != nil || (ioerr == nil && respData.Len() != 0) {
+			if ioerr == nil {
+				var respProto colmetricpb.ExportMetricsServiceResponse
+				ioerr = proto.Unmarshal(respData.Bytes(), &respProto)
+				if ioerr == nil && respProto.PartialSuccess != nil {
+					ioerr = otlp.PartialSuccessToError(
+						otlp.MetricsPartialSuccess,
+						respProto.PartialSuccess.RejectedDataPoints,
+						respProto.PartialSuccess.ErrorMessage,
+					)
+				}
+			}
+			if ioerr != nil {
+				otel.Handle(ioerr)
+			}
+		}
+
 		if err := resp.Body.Close(); err != nil {
 			return err
 		}

exporters/otlp/otlpmetric/otlpmetrichttp/client_test.go

@@ -24,10 +24,12 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/internal/otlpmetrictest"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
 	"go.opentelemetry.io/otel/sdk/resource"
+	collectormetricpb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
 )
 
 const (
@@ -269,3 +271,34 @@ func TestStopWhileExporting(t *testing.T) {
 	assert.NoError(t, err)
 	<-doneCh
 }
+
+func TestExportPartialSuccess(t *testing.T) {
+	mcCfg := mockCollectorConfig{
+		Partial: &collectormetricpb.ExportMetricsPartialSuccess{
+			RejectedDataPoints: 2,
+			ErrorMessage:       "partially successful",
+		},
+	}
+	mc := runMockCollector(t, mcCfg)
+	defer mc.MustStop(t)
+	driver := otlpmetrichttp.NewClient(
+		otlpmetrichttp.WithEndpoint(mc.Endpoint()),
+		otlpmetrichttp.WithInsecure(),
+	)
+	ctx := context.Background()
+	exporter, err := otlpmetric.New(ctx, driver)
+	require.NoError(t, err)
+	defer func() {
+		assert.NoError(t, exporter.Shutdown(ctx))
+	}()
+
+	errors := new([]error)
+	otel.SetErrorHandler(otel.ErrorHandlerFunc(func(err error) {
+		*errors = append(*errors, err)
+	}))
+	err = exporter.Export(ctx, testResource, oneRecord)
+	assert.NoError(t, err)
+	require.Equal(t, 1, len(*errors))
+	require.Contains(t, (*errors)[0].Error(), "partially successful")
+	require.Contains(t, (*errors)[0].Error(), "2 metric data points rejected")
+}

exporters/otlp/otlpmetric/otlpmetrichttp/go.mod

@@ -4,6 +4,7 @@ go 1.17
 
 require (
 	github.com/stretchr/testify v1.7.1
+	go.opentelemetry.io/otel v1.9.0
 	go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.9.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric v0.31.0
 	go.opentelemetry.io/otel/sdk v1.9.0
@@ -19,7 +20,6 @@ require (
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	go.opentelemetry.io/otel v1.9.0 // indirect
 	go.opentelemetry.io/otel/metric v0.31.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v0.31.0 // indirect
 	go.opentelemetry.io/otel/trace v1.9.0 // indirect

exporters/otlp/otlpmetric/otlpmetrichttp/mock_collector_test.go

@@ -45,6 +45,7 @@ type mockCollector struct {
 
 	injectHTTPStatus  []int
 	injectContentType string
+	partial           *collectormetricpb.ExportMetricsPartialSuccess
 	delay             <-chan struct{}
 
 	clientTLSConfig *tls.Config
@@ -86,7 +87,9 @@ func (c *mockCollector) serveMetrics(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
-	response := collectormetricpb.ExportMetricsServiceResponse{}
+	response := collectormetricpb.ExportMetricsServiceResponse{
+		PartialSuccess: c.partial,
+	}
 	rawResponse, err := proto.Marshal(&response)
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)
@@ -187,6 +190,7 @@ type mockCollectorConfig struct {
 	Delay             <-chan struct{}
 	WithTLS           bool
 	ExpectedHeaders   map[string]string
+	Partial           *collectormetricpb.ExportMetricsPartialSuccess
 }
 
 func (c *mockCollectorConfig) fillInDefaults() {
@@ -207,6 +211,7 @@ func runMockCollector(t *testing.T, cfg mockCollectorConfig) *mockCollector {
 		injectHTTPStatus:  cfg.InjectHTTPStatus,
 		injectContentType: cfg.InjectContentType,
 		delay:             cfg.Delay,
+		partial:           cfg.Partial,
 		expectedHeaders:   cfg.ExpectedHeaders,
 	}
 	mux := http.NewServeMux()

exporters/otlp/otlptrace/otlptracegrpc/client.go

@@ -26,6 +26,8 @@ import (
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/otlp"
 	"go.opentelemetry.io/otel/exporters/otlp/internal/retry"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/internal/otlpconfig"
@@ -196,9 +198,16 @@ func (c *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.Resourc
 	defer cancel()
 
 	return c.requestFunc(ctx, func(iCtx context.Context) error {
-		_, err := c.tsc.Export(iCtx, &coltracepb.ExportTraceServiceRequest{
+		resp, err := c.tsc.Export(iCtx, &coltracepb.ExportTraceServiceRequest{
 			ResourceSpans: protoSpans,
 		})
+		if resp != nil && resp.PartialSuccess != nil {
+			otel.Handle(otlp.PartialSuccessToError(
+				otlp.TracingPartialSuccess,
+				resp.PartialSuccess.RejectedSpans,
+				resp.PartialSuccess.ErrorMessage,
+			))
+		}
 		// nil is converted to OK.
 		if status.Code(err) == codes.OK {
 			// Success.

exporters/otlp/otlptrace/otlptracegrpc/client_test.go

@@ -4,7 +4,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-//     http://www.apache.org/licenses/LICENSE-2.0
+//	http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -30,12 +30,14 @@ import (
 	"google.golang.org/grpc/encoding/gzip"
 	"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/internal/otlptracetest"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+	coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1"
 	commonpb "go.opentelemetry.io/proto/otlp/common/v1"
 )
 
@@ -386,3 +388,26 @@ func TestEmptyData(t *testing.T) {
 
 	assert.NoError(t, exp.ExportSpans(ctx, nil))
 }
+
+func TestPartialSuccess(t *testing.T) {
+	mc := runMockCollectorWithConfig(t, &mockConfig{
+		partial: &coltracepb.ExportTracePartialSuccess{
+			RejectedSpans: 2,
+			ErrorMessage:  "partially successful",
+		},
+	})
+	t.Cleanup(func() { require.NoError(t, mc.stop()) })
+
+	errors := new([]error)
+	otel.SetErrorHandler(otel.ErrorHandlerFunc(func(err error) {
+		*errors = append(*errors, err)
+	}))
+	ctx := context.Background()
+	exp := newGRPCExporter(t, ctx, mc.endpoint)
+	t.Cleanup(func() { require.NoError(t, exp.Shutdown(ctx)) })
+	require.NoError(t, exp.ExportSpans(ctx, roSpans))
+
+	require.Equal(t, 1, len(*errors))
+	require.Contains(t, (*errors)[0].Error(), "partially successful")
+	require.Contains(t, (*errors)[0].Error(), "2 spans rejected")
+}

exporters/otlp/otlptrace/otlptracegrpc/mock_collector_test.go

@@ -36,6 +36,7 @@ func makeMockCollector(t *testing.T, mockConfig *mockConfig) *mockCollector {
 		traceSvc: &mockTraceService{
 			storage: otlptracetest.NewSpansStorage(),
 			errors:  mockConfig.errors,
+			partial: mockConfig.partial,
 		},
 	}
 }
@@ -44,6 +45,7 @@ type mockTraceService struct {
 	collectortracepb.UnimplementedTraceServiceServer
 
 	errors      []error
+	partial     *collectortracepb.ExportTracePartialSuccess
 	requests    int
 	mu          sync.RWMutex
 	storage     otlptracetest.SpansStorage
@@ -82,7 +84,9 @@ func (mts *mockTraceService) Export(ctx context.Context, exp *collectortracepb.E
 		<-mts.exportBlock
 	}
 
-	reply := &collectortracepb.ExportTraceServiceResponse{}
+	reply := &collectortracepb.ExportTraceServiceResponse{
+		PartialSuccess: mts.partial,
+	}
 	if mts.requests < len(mts.errors) {
 		idx := mts.requests
 		return reply, mts.errors[idx]
@@ -106,6 +110,7 @@ type mockCollector struct {
 type mockConfig struct {
 	errors   []error
 	endpoint string
+	partial  *collectortracepb.ExportTracePartialSuccess
 }
 
 var _ collectortracepb.TraceServiceServer = (*mockTraceService)(nil)