Correct SDK trace Exporter interface (open-telemetry#1078)

* Update trace export interface

Move to conforming to the specification.

* Update documentation in export trace

* Update sdk trace provider to support new trace exporter

* Update SpanProcessors

Support the Provider changes and new trace exporter.

* Update the SDK to support the changes

* Update trace Provider to not return an error

* Update sdk with new Provider return

Also fix the testExporter ExportSpans method

* Update exporters with changes

* Update examples with changes

* Update Changelog

* Move error handling to end of shutdown

* Update exporter interface

Rename to SpanExporter to match specification. Add an error return value
to the Shutdown method based on feedback. Propagate these changes.

Remove the Stop method from the OTLP exporter to avoid confusion and
redundancy.

* Add test to check OTLP Shutdown honors context

* Add Jaeger exporter test for shutdown

* Fix race in Jaeger test

* Unify shutdown behavior and testing

* Update sdk/trace/simple_span_processor.go

Co-authored-by: Anthony Mirabella <a9@aneurysm9.com>

Co-authored-by: Anthony Mirabella <a9@aneurysm9.com>

CHANGELOG.md

@@ -35,11 +35,12 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ### Added
 
+- Support for exporting array-valued attributes via OTLP. (#992)
 - `Noop` and `InMemory` `SpanBatcher` implementations to help with testing integrations. (#994)
-- Integration tests for more OTel Collector Attribute types. (#1062)
-- A dimensionality-reducing metric Processor. (#1057)
 - Support for filtering metric label sets. (#1047)
-- Support for exporting array-valued attributes via OTLP. (#992)
+- A dimensionality-reducing metric Processor. (#1057)
+- Integration tests for more OTel Collector Attribute types. (#1062)
+- A new `WithSpanProcessor` `ProviderOption` is added to the `go.opentelemetry.io/otel/sdk/trace` package to create a `Provider` and automatically register the `SpanProcessor`. (#1078)
 
 ### Changed
 
@@ -58,6 +59,13 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Unify Callback Function Naming.
    Rename `*Callback` with `*Func`. (#1061)
 - CI builds validate against last two versions of Go, dropping 1.13 and adding 1.15. (#1064)
+- The `go.opentelemetry.io/otel/sdk/export/trace` interfaces `SpanSyncer` and `SpanBatcher` have been replaced with a specification compliant `Exporter` interface.
+   This interface still supports the export of `SpanData`, but only as a slice.
+   Implementation are also required now to return any error from `ExportSpans` if one occurs as well as implement a `Shutdown` method for exporter clean-up. (#1078)
+- The `go.opentelemetry.io/otel/sdk/trace` `NewBatchSpanProcessor` function no longer returns an error.
+   If a `nil` exporter is passed as an argument to this function, instead of it returning an error, it now returns a `BatchSpanProcessor` that handles the export of `SpanData` by not taking any action. (#1078)
+- The `go.opentelemetry.io/otel/sdk/trace` `NewProvider` function to create a `Provider` no longer returns an error, instead only a `*Provider`.
+   This change is related to `NewBatchSpanProcessor` not returning an error which was the only error this function would return. (#1078)
 
 ### Removed
 

example/namedtracer/main.go

@@ -43,11 +43,14 @@ func initTracer() {
 		log.Panicf("failed to initialize stdout exporter %v\n", err)
 		return
 	}
-	tp, err = sdktrace.NewProvider(sdktrace.WithBatcher(exp),
-		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}))
-	if err != nil {
-		log.Panicf("failed to initialize trace provider %v\n", err)
-	}
+	tp = sdktrace.NewProvider(
+		sdktrace.WithConfig(
+			sdktrace.Config{
+				DefaultSampler: sdktrace.AlwaysSample(),
+			},
+		),
+		sdktrace.WithBatcher(exp),
+	)
 	global.SetTracerProvider(tp)
 }
 

example/otel-collector/main.go

@@ -54,15 +54,14 @@ func initProvider() (*otlp.Exporter, *push.Controller) {
 	)
 	handleErr(err, "failed to create exporter")
 
-	tracerProvider, err := sdktrace.NewProvider(
+	tracerProvider := sdktrace.NewProvider(
 		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
 		sdktrace.WithResource(resource.New(
 			// the service name used to display traces in backends
 			semconv.ServiceNameKey.String("test-service"),
 		)),
 		sdktrace.WithBatcher(exp),
 	)
-	handleErr(err, "failed to create trace provider")
 
 	pusher := push.New(
 		basic.New(
@@ -84,7 +83,9 @@ func main() {
 	log.Printf("Waiting for connection...")
 
 	exp, pusher := initProvider()
-	defer func() { handleErr(exp.Stop(), "failed to stop exporter") }()
+	defer func() {
+		handleErr(exp.Shutdown(context.Background()), "failed to stop exporter")
+	}()
 	defer pusher.Stop() // pushes any last exports to the receiver
 
 	tracer := global.Tracer("test-tracer")

exporters/otlp/README.md

@@ -41,17 +41,13 @@ func main() {
 	}()
 
 	// Note: The exporter can also be used as a Batcher. E.g.
-	//   tracerProvider, err := sdktrace.NewProvider(
+	//   tracerProvider := sdktrace.NewProvider(
 	//   	sdktrace.WithBatcher(exporter,
 	//   		sdktrace.WithBatchTimeout(time.Second*15),
 	//   		sdktrace.WithMaxExportBatchSize(100),
 	//   	),
 	//   )
-	tracerProvider, err := sdktrace.NewProvider(sdktrace.WithBatcher(exporter))
-	if err != nil {
-		log.Fatal("failed to create trace provider: %v", err)
-	}
-
+	tracerProvider := sdktrace.NewProvider(sdktrace.WithBatcher(exporter))
 	pusher := push.New(simple.NewWithExactDistribution(), exporter)
 	pusher.Start()
 	metricProvider := pusher.Provider()

exporters/otlp/example_test.go

@@ -33,19 +33,22 @@ func Example_insecure() {
 		log.Fatalf("Failed to create the collector exporter: %v", err)
 	}
 	defer func() {
-		_ = exp.Stop()
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		if err := exp.Shutdown(ctx); err != nil {
+			global.Handle(err)
+		}
 	}()
 
-	tp, _ := sdktrace.NewProvider(
+	tp := sdktrace.NewProvider(
 		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
-		sdktrace.WithBatcher(exp, // add following two options to ensure flush
+		sdktrace.WithBatcher(
+			exp,
+			// add following two options to ensure flush
 			sdktrace.WithBatchTimeout(5),
 			sdktrace.WithMaxExportBatchSize(10),
-		))
-	if err != nil {
-		log.Fatalf("error creating trace provider: %v\n", err)
-	}
-
+		),
+	)
 	global.SetTracerProvider(tp)
 
 	tracer := global.Tracer("test-tracer")
@@ -74,19 +77,22 @@ func Example_withTLS() {
 		log.Fatalf("failed to create the collector exporter: %v", err)
 	}
 	defer func() {
-		_ = exp.Stop()
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		if err := exp.Shutdown(ctx); err != nil {
+			global.Handle(err)
+		}
 	}()
 
-	tp, err := sdktrace.NewProvider(
+	tp := sdktrace.NewProvider(
 		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
-		sdktrace.WithBatcher(exp, // add following two options to ensure flush
+		sdktrace.WithBatcher(
+			exp,
+			// add following two options to ensure flush
 			sdktrace.WithBatchTimeout(5),
 			sdktrace.WithMaxExportBatchSize(10),
-		))
-	if err != nil {
-		log.Fatalf("error creating trace provider: %v\n", err)
-	}
-
+		),
+	)
 	global.SetTracerProvider(tp)
 
 	tracer := global.Tracer("test-tracer")

exporters/otlp/otlp.go

@@ -28,7 +28,6 @@ import (
 	colmetricpb "go.opentelemetry.io/otel/exporters/otlp/internal/opentelemetry-proto-gen/collector/metrics/v1"
 	coltracepb "go.opentelemetry.io/otel/exporters/otlp/internal/opentelemetry-proto-gen/collector/trace/v1"
 
-	"go.opentelemetry.io/otel/api/global"
 	"go.opentelemetry.io/otel/api/metric"
 	"go.opentelemetry.io/otel/exporters/otlp/internal/transform"
 	metricsdk "go.opentelemetry.io/otel/sdk/export/metric"
@@ -57,7 +56,7 @@ type Exporter struct {
 	metadata metadata.MD
 }
 
-var _ tracesdk.SpanBatcher = (*Exporter)(nil)
+var _ tracesdk.SpanExporter = (*Exporter)(nil)
 var _ metricsdk.Exporter = (*Exporter)(nil)
 
 func configureOptions(cfg *Config, opts ...ExporterOption) {
@@ -195,10 +194,14 @@ func (e *Exporter) dialToCollector() (*grpc.ClientConn, error) {
 	return grpc.DialContext(ctx, addr, dialOpts...)
 }
 
-// Stop shuts down all the connections and resources
-// related to the exporter.
-// If the exporter is not started then this func does nothing.
-func (e *Exporter) Stop() error {
+// closeStopCh is used to wrap the exporters stopCh channel closing for testing.
+var closeStopCh = func(stopCh chan bool) {
+	close(stopCh)
+}
+
+// Shutdown closes all connections and releases resources currently being used
+// by the exporter. If the exporter is not started this does nothing.
+func (e *Exporter) Shutdown(ctx context.Context) error {
 	e.mu.RLock()
 	cc := e.grpcClientConn
 	started := e.started
@@ -208,20 +211,24 @@ func (e *Exporter) Stop() error {
 		return nil
 	}
 
-	// Now close the underlying gRPC connection.
 	var err error
 	if cc != nil {
+		// Clean things up before checking this error.
 		err = cc.Close()
 	}
 
 	// At this point we can change the state variable started
 	e.mu.Lock()
 	e.started = false
 	e.mu.Unlock()
-	close(e.stopCh)
+	closeStopCh(e.stopCh)
 
 	// Ensure that the backgroundConnector returns
-	<-e.backgroundConnectionDoneCh
+	select {
+	case <-e.backgroundConnectionDoneCh:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 
 	return err
 }
@@ -272,27 +279,22 @@ func (e *Exporter) ExportKindFor(*metric.Descriptor, aggregation.Kind) metricsdk
 	return metricsdk.PassThroughExporter
 }
 
-func (e *Exporter) ExportSpan(ctx context.Context, sd *tracesdk.SpanData) {
-	e.uploadTraces(ctx, []*tracesdk.SpanData{sd})
+func (e *Exporter) ExportSpans(ctx context.Context, sds []*tracesdk.SpanData) error {
+	return e.uploadTraces(ctx, sds)
 }
 
-func (e *Exporter) ExportSpans(ctx context.Context, sds []*tracesdk.SpanData) {
-	e.uploadTraces(ctx, sds)
-}
-
-func (e *Exporter) uploadTraces(ctx context.Context, sdl []*tracesdk.SpanData) {
+func (e *Exporter) uploadTraces(ctx context.Context, sdl []*tracesdk.SpanData) error {
 	select {
 	case <-e.stopCh:
-		return
-
+		return nil
 	default:
 		if !e.connected() {
-			return
+			return nil
 		}
 
 		protoSpans := transform.SpanData(sdl)
 		if len(protoSpans) == 0 {
-			return
+			return nil
 		}
 
 		e.senderMu.Lock()
@@ -302,7 +304,8 @@ func (e *Exporter) uploadTraces(ctx context.Context, sdl []*tracesdk.SpanData) {
 		e.senderMu.Unlock()
 		if err != nil {
 			e.setStateDisconnected(err)
-			global.Handle(err)
+			return err
 		}
 	}
+	return nil
 }