Handle partial success response from OTLP server in otlpmetric exporters

CHANGELOG.md

@@ -44,6 +44,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Exported `Status` codes in the `go.opentelemetry.io/otel/exporters/zipkin` exporter are now exported as all upper case values. (#3340)
 - Reenabled Attribute Filters in the Metric SDK. (#3396)
 - Do not report empty partial-success responses in the `go.opentelemetry.io/otel/exporters/otlp` exporters. (#3438, #3432)
+- Handle partial success responses in `go.opentelemetry.io/otel/exporters/otlp/otlpmetric` exporters. (#3162, #3440)
 
 ## [1.11.1/0.33.0] 2022-10-19
 

exporters/otlp/otlpmetric/internal/otest/client.go

@@ -16,6 +16,7 @@ package otest // import "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/inte
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
@@ -24,9 +25,11 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 
+	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric"
 	"go.opentelemetry.io/otel/metric/unit"
 	semconv "go.opentelemetry.io/otel/semconv/v1.10.0"
+	collpb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
 	cpb "go.opentelemetry.io/proto/otlp/common/v1"
 	mpb "go.opentelemetry.io/proto/otlp/metrics/v1"
 	rpb "go.opentelemetry.io/proto/otlp/resource/v1"
@@ -168,7 +171,10 @@ var (
 // otlpmetric.Client implementation that is connected to also returned
 // Collector implementation. The Client is ready to upload metric data to the
 // Collector which is ready to store that data.
-type ClientFactory func() (otlpmetric.Client, Collector)
+//
+// If resultCh is not nil, the returned Collector needs to use the responses
+// from that channel to send back to the client for every export request.
+type ClientFactory func(resultCh <-chan ExportResult) (otlpmetric.Client, Collector)
 
 // RunClientTests runs a suite of Client integration tests. For example:
 //
@@ -177,17 +183,17 @@ func RunClientTests(f ClientFactory) func(*testing.T) {
 	return func(t *testing.T) {
 		t.Run("ClientHonorsContextErrors", func(t *testing.T) {
 			t.Run("Shutdown", testCtxErrs(func() func(context.Context) error {
-				c, _ := f()
+				c, _ := f(nil)
 				return c.Shutdown
 			}))
 
 			t.Run("ForceFlush", testCtxErrs(func() func(context.Context) error {
-				c, _ := f()
+				c, _ := f(nil)
 				return c.ForceFlush
 			}))
 
 			t.Run("UploadMetrics", testCtxErrs(func() func(context.Context) error {
-				c, _ := f()
+				c, _ := f(nil)
 				return func(ctx context.Context) error {
 					return c.UploadMetrics(ctx, nil)
 				}
@@ -196,7 +202,7 @@ func RunClientTests(f ClientFactory) func(*testing.T) {
 
 		t.Run("ForceFlushFlushes", func(t *testing.T) {
 			ctx := context.Background()
-			client, collector := f()
+			client, collector := f(nil)
 			require.NoError(t, client.UploadMetrics(ctx, resourceMetrics))
 
 			require.NoError(t, client.ForceFlush(ctx))
@@ -211,7 +217,7 @@ func RunClientTests(f ClientFactory) func(*testing.T) {
 
 		t.Run("UploadMetrics", func(t *testing.T) {
 			ctx := context.Background()
-			client, coll := f()
+			client, coll := f(nil)
 
 			require.NoError(t, client.UploadMetrics(ctx, resourceMetrics))
 			require.NoError(t, client.Shutdown(ctx))
@@ -222,6 +228,51 @@ func RunClientTests(f ClientFactory) func(*testing.T) {
 				t.Fatalf("unexpected ResourceMetrics:\n%s", diff)
 			}
 		})
+
+		t.Run("PartialSuccess", func(t *testing.T) {
+			const n, msg = 2, "bad data"
+			rCh := make(chan ExportResult, 3)
+			rCh <- ExportResult{
+				Response: &collpb.ExportMetricsServiceResponse{
+					PartialSuccess: &collpb.ExportMetricsPartialSuccess{
+						RejectedDataPoints: n,
+						ErrorMessage:       msg,
+					},
+				},
+			}
+			rCh <- ExportResult{
+				Response: &collpb.ExportMetricsServiceResponse{
+					PartialSuccess: &collpb.ExportMetricsPartialSuccess{
+						// Should not be logged.
+						RejectedDataPoints: 0,
+						ErrorMessage:       "",
+					},
+				},
+			}
+			rCh <- ExportResult{
+				Response: &collpb.ExportMetricsServiceResponse{},
+			}
+
+			ctx := context.Background()
+			client, _ := f(rCh)
+
+			defer func(orig otel.ErrorHandler) {
+				otel.SetErrorHandler(orig)
+			}(otel.GetErrorHandler())
+
+			errs := []error{}
+			eh := otel.ErrorHandlerFunc(func(e error) { errs = append(errs, e) })
+			otel.SetErrorHandler(eh)
+
+			require.NoError(t, client.UploadMetrics(ctx, resourceMetrics))
+			require.NoError(t, client.UploadMetrics(ctx, resourceMetrics))
+			require.NoError(t, client.UploadMetrics(ctx, resourceMetrics))
+			require.NoError(t, client.Shutdown(ctx))
+
+			require.Equal(t, 1, len(errs))
+			want := fmt.Sprintf("%s (%d metric data points rejected)", msg, n)
+			assert.ErrorContains(t, errs[0], want)
+		})
 	}
 }
 

exporters/otlp/otlpmetric/internal/otest/client_test.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"testing"
 
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/otlp/internal"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric"
 	"go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/metric/aggregation"
@@ -28,6 +30,7 @@ import (
 )
 
 type client struct {
+	rCh     <-chan ExportResult
 	storage *Storage
 }
 
@@ -47,15 +50,26 @@ func (c *client) UploadMetrics(ctx context.Context, rm *mpb.ResourceMetrics) err
 	c.storage.Add(&cpb.ExportMetricsServiceRequest{
 		ResourceMetrics: []*mpb.ResourceMetrics{rm},
 	})
+	if c.rCh != nil {
+		r := <-c.rCh
+		if r.Response != nil && r.Response.GetPartialSuccess() != nil {
+			msg := r.Response.GetPartialSuccess().GetErrorMessage()
+			n := r.Response.GetPartialSuccess().GetRejectedDataPoints()
+			if msg != "" || n > 0 {
+				otel.Handle(internal.MetricPartialSuccessError(n, msg))
+			}
+		}
+		return r.Err
+	}
 	return ctx.Err()
 }
 
 func (c *client) ForceFlush(ctx context.Context) error { return ctx.Err() }
 func (c *client) Shutdown(ctx context.Context) error   { return ctx.Err() }
 
 func TestClientTests(t *testing.T) {
-	factory := func() (otlpmetric.Client, Collector) {
-		c := &client{storage: NewStorage()}
+	factory := func(rCh <-chan ExportResult) (otlpmetric.Client, Collector) {
+		c := &client{rCh: rCh, storage: NewStorage()}
 		return c, c
 	}
 

exporters/otlp/otlpmetric/internal/otest/collector.go

@@ -50,6 +50,11 @@ type Collector interface {
 	Collect() *Storage
 }
 
+type ExportResult struct {
+	Response *collpb.ExportMetricsServiceResponse
+	Err      error
+}
+
 // Storage stores uploaded OTLP metric data in their proto form.
 type Storage struct {
 	dataMu sync.Mutex
@@ -86,7 +91,7 @@ type GRPCCollector struct {
 	headers   metadata.MD
 	storage   *Storage
 
-	errCh    <-chan error
+	resultCh <-chan ExportResult
 	listener net.Listener
 	srv      *grpc.Server
 }
@@ -100,14 +105,14 @@ type GRPCCollector struct {
 // If errCh is not nil, the collector will respond to Export calls with errors
 // sent on that channel. This means that if errCh is not nil Export calls will
 // block until an error is received.
-func NewGRPCCollector(endpoint string, errCh <-chan error) (*GRPCCollector, error) {
+func NewGRPCCollector(endpoint string, resultCh <-chan ExportResult) (*GRPCCollector, error) {
 	if endpoint == "" {
 		endpoint = "localhost:0"
 	}
 
 	c := &GRPCCollector{
-		storage: NewStorage(),
-		errCh:   errCh,
+		storage:  NewStorage(),
+		resultCh: resultCh,
 	}
 
 	var err error
@@ -155,11 +160,14 @@ func (c *GRPCCollector) Export(ctx context.Context, req *collpb.ExportMetricsSer
 		c.headersMu.Unlock()
 	}
 
-	var err error
-	if c.errCh != nil {
-		err = <-c.errCh
+	if c.resultCh != nil {
+		r := <-c.resultCh
+		if r.Response == nil {
+			return &collpb.ExportMetricsServiceResponse{}, r.Err
+		}
+		return r.Response, r.Err
 	}
-	return &collpb.ExportMetricsServiceResponse{}, err
+	return &collpb.ExportMetricsServiceResponse{}, nil
 }
 
 var emptyExportMetricsServiceResponse = func() []byte {
@@ -189,7 +197,7 @@ type HTTPCollector struct {
 	headers   http.Header
 	storage   *Storage
 
-	errCh    <-chan error
+	resultCh <-chan ExportResult
 	listener net.Listener
 	srv      *http.Server
 }
@@ -207,7 +215,7 @@ type HTTPCollector struct {
 // If errCh is not nil, the collector will respond to HTTP requests with errors
 // sent on that channel. This means that if errCh is not nil Export calls will
 // block until an error is received.
-func NewHTTPCollector(endpoint string, errCh <-chan error) (*HTTPCollector, error) {
+func NewHTTPCollector(endpoint string, resultCh <-chan ExportResult) (*HTTPCollector, error) {
 	u, err := url.Parse(endpoint)
 	if err != nil {
 		return nil, err
@@ -220,9 +228,9 @@ func NewHTTPCollector(endpoint string, errCh <-chan error) (*HTTPCollector, erro
 	}
 
 	c := &HTTPCollector{
-		headers: http.Header{},
-		storage: NewStorage(),
-		errCh:   errCh,
+		headers:  http.Header{},
+		storage:  NewStorage(),
+		resultCh: resultCh,
 	}
 
 	c.listener, err = net.Listen("tcp", u.Host)
@@ -276,22 +284,25 @@ func (c *HTTPCollector) handler(w http.ResponseWriter, r *http.Request) {
 	c.respond(w, c.record(r))
 }
 
-func (c *HTTPCollector) record(r *http.Request) error {
+func (c *HTTPCollector) record(r *http.Request) ExportResult {
 	// Currently only supports protobuf.
 	if v := r.Header.Get("Content-Type"); v != "application/x-protobuf" {
-		return fmt.Errorf("content-type not supported: %s", v)
+		err := fmt.Errorf("content-type not supported: %s", v)
+		return ExportResult{Err: err}
 	}
 
 	body, err := c.readBody(r)
 	if err != nil {
-		return err
+		return ExportResult{Err: err}
 	}
 	pbRequest := &collpb.ExportMetricsServiceRequest{}
 	err = proto.Unmarshal(body, pbRequest)
 	if err != nil {
-		return &HTTPResponseError{
-			Err:    err,
-			Status: http.StatusInternalServerError,
+		return ExportResult{
+			Err: &HTTPResponseError{
+				Err:    err,
+				Status: http.StatusInternalServerError,
+			},
 		}
 	}
 	c.storage.Add(pbRequest)
@@ -304,10 +315,10 @@ func (c *HTTPCollector) record(r *http.Request) error {
 	}
 	c.headersMu.Unlock()
 
-	if c.errCh != nil {
-		err = <-c.errCh
+	if c.resultCh != nil {
+		return <-c.resultCh
 	}
-	return err
+	return ExportResult{Err: err}
 }
 
 func (c *HTTPCollector) readBody(r *http.Request) (body []byte, err error) {
@@ -345,12 +356,12 @@ func (c *HTTPCollector) readBody(r *http.Request) (body []byte, err error) {
 	return body, err
 }
 
-func (c *HTTPCollector) respond(w http.ResponseWriter, err error) {
-	if err != nil {
+func (c *HTTPCollector) respond(w http.ResponseWriter, resp ExportResult) {
+	if resp.Err != nil {
 		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
 		w.Header().Set("X-Content-Type-Options", "nosniff")
 		var e *HTTPResponseError
-		if errors.As(err, &e) {
+		if errors.As(resp.Err, &e) {
 			for k, vals := range e.Header {
 				for _, v := range vals {
 					w.Header().Add(k, v)
@@ -360,14 +371,22 @@ func (c *HTTPCollector) respond(w http.ResponseWriter, err error) {
 			fmt.Fprintln(w, e.Error())
 		} else {
 			w.WriteHeader(http.StatusBadRequest)
-			fmt.Fprintln(w, err.Error())
+			fmt.Fprintln(w, resp.Err.Error())
 		}
 		return
 	}
 
 	w.Header().Set("Content-Type", "application/x-protobuf")
 	w.WriteHeader(http.StatusOK)
-	_, _ = w.Write(emptyExportMetricsServiceResponse)
+	if resp.Response == nil {
+		_, _ = w.Write(emptyExportMetricsServiceResponse)
+	} else {
+		r, err := proto.Marshal(resp.Response)
+		if err != nil {
+			panic(err)
+		}
+		_, _ = w.Write(r)
+	}
 }
 
 type mathRandReader struct{}

exporters/otlp/otlpmetric/otlpmetricgrpc/client.go

@@ -24,6 +24,8 @@ import (
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/otlp/internal"
 	"go.opentelemetry.io/otel/exporters/otlp/internal/retry"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/internal/oconf"
@@ -163,9 +165,17 @@ 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 {
+			msg := resp.PartialSuccess.GetErrorMessage()
+			n := resp.PartialSuccess.GetRejectedDataPoints()
+			if n != 0 || msg != "" {
+				err := internal.MetricPartialSuccessError(n, msg)
+				otel.Handle(err)
+			}
+		}
 		// nil is converted to OK.
 		if status.Code(err) == codes.OK {
 			// Success.