Repository: GoogleCloudPlatform/opentelemetry-operations-go
Pull request #63 (2020-07-06 21:50:20)

inject labels
Put the logic of injecting labels into span in the trace_proto.go because
1) The semantic of injecting labels is on the level of building spans, not implementing trace exporter.
2) func protoFromSpanData would be used by both func ExportSpan() and func ExportSpans() in trace.go. It would have  duplicate callings if we put the code in trace.go or cloudtrace.go.

YANYZP: Tested it by client.go on GCE and got non-duplicate results

Patch:
diff --git a/Makefile b/Makefile
index a3e8a51..966fab3 100644
--- a/Makefile
+++ b/Makefile
@@ -97,8 +97,8 @@ lint: $(TOOLS_DIR)/golangci-lint $(TOOLS_DIR)/misspell
 	set -e; for dir in $(ALL_GO_MOD_DIRS); do \
 	  echo "golangci-lint in $${dir}"; \
 	  (cd "$${dir}" && \
-	    $(TOOLS_DIR)/golangci-lint run --fix && \
-	    $(TOOLS_DIR)/golangci-lint run); \
+	    $(TOOLS_DIR)/golangci-lint --config $(CURDIR)/golangci.yml run --fix && \
+	    $(TOOLS_DIR)/golangci-lint --config $(CURDIR)/golangci.yml run); \
 	done
 	$(TOOLS_DIR)/misspell -w $(ALL_DOCS)
 	set -e; for dir in $(ALL_GO_MOD_DIRS) $(TOOLS_MOD_DIR); do \
diff --git a/example/trace/http/client/client.go b/example/trace/http/client/client.go
index 0a9737c..5249dce 100644
--- a/example/trace/http/client/client.go
+++ b/example/trace/http/client/client.go
@@ -22,43 +22,39 @@ import (
 	"os"
 
 	"net/http"
-	"time"
 
 	"google.golang.org/grpc/codes"
 
-	texporter "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace"
 	"go.opentelemetry.io/otel/api/correlation"
 	"go.opentelemetry.io/otel/api/global"
 	"go.opentelemetry.io/otel/api/kv"
 	"go.opentelemetry.io/otel/api/trace"
 	"go.opentelemetry.io/otel/plugin/httptrace"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+
+	texporter "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace"
 )
 
-func initTracer() {
+func initTracer() func() {
 	projectID := os.Getenv("PROJECT_ID")
 
 	// Create Google Cloud Trace exporter to be able to retrieve
 	// the collected spans.
-	exporter, err := texporter.NewExporter(
-		texporter.WithProjectID(projectID),
+	_, flush, err := texporter.InstallNewPipeline(
+		[]texporter.Option{texporter.WithProjectID(projectID)},
+		// For this example code we use sdktrace.AlwaysSample sampler to sample all traces.
+		// In a production application, use sdktrace.ProbabilitySampler with a desired probability.
+		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
 	)
 	if err != nil {
 		log.Fatal(err)
 	}
-
-	// For the demonstration, use sdktrace.AlwaysSample sampler to sample all traces.
-	// In a production application, use sdktrace.ProbabilitySampler with a desired probability.
-	tp, err := sdktrace.NewProvider(sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
-		sdktrace.WithSyncer(exporter))
-	if err != nil {
-		log.Fatal(err)
-	}
-	global.SetTraceProvider(tp)
+	return flush
 }
 
 func main() {
-	initTracer()
+	flush := initTracer()
+	defer flush()
 	tr := global.TraceProvider().Tracer("cloudtrace/example/client")
 
 	client := http.DefaultClient
@@ -92,7 +88,5 @@ func main() {
 	}
 
 	fmt.Printf("Response Received: %s\n\n\n", body)
-	fmt.Printf("Waiting for few seconds to export spans ...\n\n")
-	time.Sleep(10 * time.Second)
-	fmt.Println("Check traces on Google Cloud Trace")
+	fmt.Printf("Waiting to export spans ...\n\n")
 }
diff --git a/example/trace/http/server/server.go b/example/trace/http/server/server.go
index 1b82387..13cebd1 100644
--- a/example/trace/http/server/server.go
+++ b/example/trace/http/server/server.go
@@ -20,38 +20,35 @@ import (
 	"net/http"
 	"os"
 
-	cloudtrace "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace"
 	"go.opentelemetry.io/otel/api/correlation"
 	"go.opentelemetry.io/otel/api/global"
 	"go.opentelemetry.io/otel/api/trace"
 	"go.opentelemetry.io/otel/plugin/httptrace"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+
+	cloudtrace "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace"
 )
 
-func initTracer() {
+func initTracer() func() {
 	projectID := os.Getenv("PROJECT_ID")
 
 	// Create Google Cloud Trace exporter to be able to retrieve
 	// the collected spans.
-	exporter, err := cloudtrace.NewExporter(
-		cloudtrace.WithProjectID(projectID),
+	_, flush, err := cloudtrace.InstallNewPipeline(
+		[]cloudtrace.Option{cloudtrace.WithProjectID(projectID)},
+		// For this example code we use sdktrace.AlwaysSample sampler to sample all traces.
+		// In a production application, use sdktrace.ProbabilitySampler with a desired probability.
+		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
 	)
 	if err != nil {
 		log.Fatal(err)
 	}
-
-	// For the demonstration, use sdktrace.AlwaysSample sampler to sample all traces.
-	// In a production application, use sdktrace.ProbabilitySampler with a desired probability.
-	tp, err := sdktrace.NewProvider(sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
-		sdktrace.WithSyncer(exporter))
-	if err != nil {
-		log.Fatal(err)
-	}
-	global.SetTraceProvider(tp)
+	return flush
 }
 
 func main() {
-	initTracer()
+	flush := initTracer()
+	defer flush()
 
 	tr := global.TraceProvider().Tracer("cloudtrace/example/server")
 
diff --git a/exporter/metric/metric.go b/exporter/metric/metric.go
index 09fc7be..8af73b4 100644
--- a/exporter/metric/metric.go
+++ b/exporter/metric/metric.go
@@ -148,15 +148,15 @@ func (me *metricExporter) ExportMetrics(ctx context.Context, cps export.Checkpoi
 func (me *metricExporter) exportMetricDescriptor(ctx context.Context, cps export.CheckpointSet) error {
 	mds := make(map[key]*googlemetricpb.MetricDescriptor)
 	aggError := cps.ForEach(func(r export.Record) error {
-		key := keyOf(r.Descriptor())
+		k := keyOf(r.Descriptor())
 
-		if _, ok := me.mdCache[key]; ok {
+		if _, ok := me.mdCache[k]; ok {
 			return nil
 		}
 
-		if _, localok := mds[key]; !localok {
+		if _, localok := mds[k]; !localok {
 			md := me.recordToMdpb(&r)
-			mds[key] = md
+			mds[k] = md
 		}
 		return nil
 	})
@@ -168,7 +168,7 @@ func (me *metricExporter) exportMetricDescriptor(ctx context.Context, cps export
 	}
 
 	// TODO: This process is synchronous and blocks longer time if records in cps
-	// have many diffrent descriptors. In the cps.ForEach above, it should spawn
+	// have many different descriptors. In the cps.ForEach above, it should spawn
 	// goroutines to send CreateMetricDescriptorRequest asynchronously in the case
 	// the descriptor does not exist in global cache (me.mdCache).
 	// See details in #26.
@@ -330,8 +330,8 @@ func recordToMdpbKindType(r *export.Record) (googlemetricpb.MetricDescriptor_Met
 // recordToMpb converts data from records to Metric proto type for Cloud Monitoring.
 func (me *metricExporter) recordToMpb(r *export.Record) *googlemetricpb.Metric {
 	desc := r.Descriptor()
-	key := keyOf(desc)
-	md, ok := me.mdCache[key]
+	k := keyOf(desc)
+	md, ok := me.mdCache[k]
 	if !ok {
 		md = me.recordToMdpb(r)
 	}
diff --git a/exporter/trace/README.md b/exporter/trace/README.md
index 2066bc5..7712f8a 100644
--- a/exporter/trace/README.md
+++ b/exporter/trace/README.md
@@ -21,28 +21,31 @@ Add `github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace`
 import texporter "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace"
 ```
 
-Once you import the trace exporter package, then register the exporter to the application, and start tracing. If you are running in a GCP environment, the exporter will automatically authenticate using the environment's service account. If not, you will need to follow the instruction in [Authentication](#Authentication).
+Once you import the trace exporter package, create and install a new export pipeline, and then you can start tracing. If you are running in a GCP environment, the exporter will automatically authenticate using the environment's service account. If not, you will need to follow the instruction in [Authentication](#Authentication).
 
 ```go
-// Create exporter.
+// Create exporter and trace provider pipeline, and register provider.
 projectID := os.Getenv("GOOGLE_CLOUD_PROJECT")
-exporter, err := texporter.NewExporter(texporter.WithProjectID(projectID))
+_, flush, err := texporter.InstallNewPipeline(
+        []texporter.Option {
+                texporter.WithProjectID(projectID),
+                // other optional exporter options
+        },
+        // This example code uses sdktrace.AlwaysSample sampler to sample all traces.
+        // In a production environment or high QPS setup please use ProbabilitySampler
+        // set at the desired probability.
+        // Example:
+        // sdktrace.WithConfig(sdktrace.Config {
+        //        DefaultSampler: sdktrace.ProbabilitySampler(0.0001),
+        // })
+        sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
+        // other optional provider options
+)
 if err != nil {
-    log.Fatalf("texporter.NewExporter: %v", err)
+    log.Fatalf("texporter.InstallNewPipeline: %v", err)
 }
-// Create trace provider with the exporter.
-//
-// By default it uses AlwaysSample() which samples all traces.
-// In a production environment or high QPS setup please use
-// ProbabilitySampler set at the desired probability.
-// Example:
-//   config := sdktrace.Config{DefaultSampler:sdktrace.ProbabilitySampler(0.0001)}
-//   tp, err := sdktrace.NewProvider(sdktrace.WithConfig(config), ...)
-tp, err := sdktrace.NewProvider(sdktrace.WithSyncer(exporter))
-if err != nil {
-        log.Fatal(err)
-}
-global.SetTraceProvider(tp)
+// before ending program, wait for all enqueued spans to be exported
+defer flush()
 
 // Create custom span.
 tracer := global.TraceProvider().Tracer("example.com/trace")
diff --git a/exporter/trace/cloudtrace.go b/exporter/trace/cloudtrace.go
index 39affa4..592d675 100644
--- a/exporter/trace/cloudtrace.go
+++ b/exporter/trace/cloudtrace.go
@@ -25,8 +25,11 @@ import (
 	"golang.org/x/oauth2/google"
 	"google.golang.org/api/option"
 
+	"go.opentelemetry.io/otel/api/global"
 	"go.opentelemetry.io/otel/api/kv"
+	apitrace "go.opentelemetry.io/otel/api/trace"
 	export "go.opentelemetry.io/otel/sdk/export/trace"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 )
 
 // Option is function type that is passed to the exporter initialization function.
@@ -148,7 +151,7 @@ func WithOnError(onError func(err error)) func(o *options) {
 	}
 }
 
-// WithBundleDelayThreshold sets the max amount of time the exporter can wait before 
+// WithBundleDelayThreshold sets the max amount of time the exporter can wait before
 // uploading trace spans to the backend.
 func WithBundleDelayThreshold(bundleDelayThreshold time.Duration) func(o *options) {
 	return func(o *options) {
@@ -202,7 +205,7 @@ func WithTimeout(t time.Duration) func(o *options) {
 	}
 }
 
-// WithDisplayNameFormatter sets the way span's display names will be 
+// WithDisplayNameFormatter sets the way span's display names will be
 // generated from SpanData
 func WithDisplayNameFormatter(f DisplayNameFormatter) func(o *options) {
 	return func(o *options) {
@@ -229,6 +232,30 @@ type Exporter struct {
 	traceExporter *traceExporter
 }
 
+// InstallNewPipeline instantiates a NewExportPipeline and registers it globally.
+func InstallNewPipeline(opts []Option, topts ...sdktrace.ProviderOption) (apitrace.Provider, func(), error) {
+	tp, flush, err := NewExportPipeline(opts, topts...)
+	if err != nil {
+		return nil, nil, err
+	}
+	global.SetTraceProvider(tp)
+	return tp, flush, err
+}
+
+// NewExportPipeline sets up a complete export pipeline with the recommended setup
+// for trace provider. Returns provider, flush function, and errors.
+func NewExportPipeline(opts []Option, topts ...sdktrace.ProviderOption) (apitrace.Provider, func(), error) {
+	exporter, err := NewExporter(opts...)
+	if err != nil {
+		return nil, nil, err
+	}
+	tp, err := sdktrace.NewProvider(append(topts, sdktrace.WithSyncer(exporter))...)
+	if err != nil {
+		return nil, nil, err
+	}
+	return tp, exporter.traceExporter.Flush, nil
+}
+
 // NewExporter creates a new Exporter thats implements trace.Exporter.
 //
 // TODO(yoshifumi): add a metrics exporter one the spec definition
@@ -273,11 +300,6 @@ func (e *Exporter) ExportSpan(ctx context.Context, sd *export.SpanData) {
 	e.traceExporter.ExportSpan(ctx, sd)
 }
 
-// ExportSpans exports a slice of SpanData to Stackdriver Trace in batch
-func (e *Exporter) ExportSpans(ctx context.Context, sds []*export.SpanData) {
-	e.traceExporter.ExportSpans(ctx, sds)
-}
-
 func (e *Exporter) sdWithDefaultTraceAttributes(sd *export.SpanData) *export.SpanData {
 	newSD := *sd
 	for k, v := range e.traceExporter.o.DefaultTraceAttributes {
diff --git a/exporter/trace/trace.go b/exporter/trace/trace.go
index 300b238..61a69c6 100644
--- a/exporter/trace/trace.go
+++ b/exporter/trace/trace.go
@@ -22,9 +22,9 @@ import (
 	"time"
 
 	traceclient "cloud.google.com/go/trace/apiv2"
-	tracepb "google.golang.org/genproto/googleapis/devtools/cloudtrace/v2"
-	"google.golang.org/api/support/bundler"
 	"github.com/golang/protobuf/proto"
+	"google.golang.org/api/support/bundler"
+	tracepb "google.golang.org/genproto/googleapis/devtools/cloudtrace/v2"
 
 	export "go.opentelemetry.io/otel/sdk/export/trace"
 )
@@ -38,7 +38,7 @@ type traceExporter struct {
 	// uploadFn defaults in uploadSpans; it can be replaced for tests.
 	uploadFn func(ctx context.Context, spans []*tracepb.Span)
 	overflowLogger
-	client   *traceclient.Client
+	client *traceclient.Client
 }
 
 const defaultBufferedByteLimit = 8 * 1024 * 1024
@@ -94,7 +94,6 @@ func newTraceExporter(o *options) (*traceExporter, error) {
 	return e, nil
 }
 
-
 func (e *traceExporter) checkBundlerError(err error) {
 	switch err {
 	case nil:
@@ -113,24 +112,12 @@ func (e *traceExporter) ExportSpan(ctx context.Context, sd *export.SpanData) {
 	protoSpan := protoFromSpanData(sd, e.projectID, e.o.DisplayNameFormatter)
 	protoSize := proto.Size(protoSpan)
 	err := e.bundler.Add(&contextAndSpans{
-		ctx: ctx, 
+		ctx:   ctx,
 		spans: []*tracepb.Span{protoSpan},
 	}, protoSize)
 	e.checkBundlerError(err)
 }
 
-// ExportSpans exports a slice of SpanData to Stackdriver Trace in batch
-func (e *traceExporter) ExportSpans(ctx context.Context, sds []*export.SpanData) {
-	pbSpans := make([]*tracepb.Span, len(sds))
-	var protoSize int = 0
-	for i, sd := range sds {
-		pbSpans[i] = protoFromSpanData(sd, e.projectID, e.o.DisplayNameFormatter)
-		protoSize += proto.Size(pbSpans[i])
-	}
-	err := e.bundler.Add(&contextAndSpans{ctx, pbSpans}, protoSize)
-	e.checkBundlerError(err)
-}
-
 // uploadSpans sends a set of spans to Stackdriver.
 func (e *traceExporter) uploadSpans(ctx context.Context, spans []*tracepb.Span) {
 	req := tracepb.BatchWriteSpansRequest{
@@ -161,9 +148,13 @@ func (e *traceExporter) uploadSpans(ctx context.Context, spans []*tracepb.Span)
 	}
 }
 
+func (e *traceExporter) Flush() {
+	e.bundler.Flush()
+}
+
 // contextAndSpan stores both a context and spans for use with a bundler.
 type contextAndSpans struct {
-	ctx context.Context
+	ctx   context.Context
 	spans []*tracepb.Span
 }
 
@@ -204,4 +195,4 @@ func (o *overflowLogger) log() {
 	} else {
 		o.accum++
 	}
-}
\ No newline at end of file
+}
diff --git a/exporter/trace/trace_proto.go b/exporter/trace/trace_proto.go
index 13fe5ca..91a01da 100644
--- a/exporter/trace/trace_proto.go
+++ b/exporter/trace/trace_proto.go
@@ -58,7 +58,7 @@ const (
 	version = "0.1.0"
 )
 
-var userAgent = fmt.Sprintf("opentelemetry-go %s; cloudtrace-exporter %s", opentelemetry.Version(), version)
+var userAgent = fmt.Sprintf("opentelemetry-go %s; google-cloud-trace-exporter %s", opentelemetry.Version(), version)
 
 func generateDisplayName(s *export.SpanData, format DisplayNameFormatter) string {
 	if format != nil {
@@ -71,11 +71,33 @@ func generateDisplayName(s *export.SpanData, format DisplayNameFormatter) string
 	}
 }
 
+
+// If there are duplicate keys present in the list of attributes, 
+// then the first value found for the key is preserved.
+func injectLabelsFromResources(sd *export.SpanData) {
+	if sd.Resource.Len() == 0 {
+		return
+	}
+	uniqueAttrs := make(map[kv.Key]bool, len(sd.Attributes))
+	for _, attr := range sd.Attributes {
+		uniqueAttrs[attr.Key] = true	
+	}
+	for _, attr := range sd.Resource.Attributes() {
+		if uniqueAttrs[attr.Key] {
+			continue // skip resource attributes which conflict with span attributes
+		}
+		uniqueAttrs[attr.Key] = true	
+		sd.Attributes = append(sd.Attributes, attr)
+	}
+}
+
 func protoFromSpanData(s *export.SpanData, projectID string, format DisplayNameFormatter) *tracepb.Span {
 	if s == nil {
 		return nil
 	}
 
+	injectLabelsFromResources(s)
+
 	traceIDString := s.SpanContext.TraceID.String()
 	spanIDString := s.SpanContext.SpanID.String()
 
diff --git a/golangci.yml b/golangci.yml
new file mode 100644
index 0000000..9ed1b7a
--- /dev/null
+++ b/golangci.yml
@@ -0,0 +1,127 @@
+# This file contains all available configuration options
+# with their default values.
+
+# options for analysis running
+run:
+  # default concurrency is a available CPU number
+  concurrency: 4
+
+  # timeout for analysis, e.g. 30s, 5m, default is 1m
+  timeout: 1m
+
+  # exit code when at least one issue was found, default is 1
+  issues-exit-code: 1
+
+  # include test files or not, default is true
+  tests: true
+
+  # list of build tags, all linters use it. Default is empty list.
+  build-tags:
+
+  # which dirs to skip: issues from them won't be reported;
+  # can use regexp here: generated.*, regexp is applied on full path;
+  # default value is empty list, but default dirs are skipped independently
+  # from this option's value (see skip-dirs-use-default).
+  # "/" will be replaced by current OS file path separator to properly work
+  # on Windows.
+  skip-dirs:
+
+  # default is true. Enables skipping of directories:
+  #   vendor$, third_party$, testdata$, examples$, Godeps$, builtin$
+  skip-dirs-use-default: false
+
+  # which files to skip: they will be analyzed, but issues from them
+  # won't be reported. Default value is empty list, but there is
+  # no need to include all autogenerated files, we confidently recognize
+  # autogenerated files. If it's not please let us know.
+  # "/" will be replaced by current OS file path separator to properly work
+  # on Windows.
+  skip-files:
+
+  # by default isn't set. If set we pass it to "go list -mod={option}". From "go help modules":
+  # If invoked with -mod=readonly, the go command is disallowed from the implicit
+  # automatic updating of go.mod described above. Instead, it fails when any changes
+  # to go.mod are needed. This setting is most useful to check that go.mod does
+  # not need updates, such as in a continuous integration and testing system.
+  # If invoked with -mod=vendor, the go command assumes that the vendor
+  # directory holds the correct copies of dependencies and ignores
+  # the dependency descriptions in go.mod.
+  modules-download-mode: readonly
+
+  # Allow multiple parallel golangci-lint instances running.
+  # If false (default) - golangci-lint acquires file lock on start.
+  allow-parallel-runners: false
+
+
+# output configuration options
+output:
+  # colored-line-number|line-number|json|tab|checkstyle|code-climate, default is "colored-line-number"
+  format: colored-line-number
+
+  # print lines of code with issue, default is true
+  print-issued-lines: true
+
+  # print linter name in the end of issue text, default is true
+  print-linter-name: true
+
+  # make issues output unique by line, default is true
+  uniq-by-line: true
+
+
+# all available settings of specific linters
+linters-settings:
+  gofmt:
+    # simplify code: gofmt with `-s` option, true by default
+    simplify: true
+  goimports:
+    # put imports beginning with prefix after 3rd-party packages;
+    # it's a comma-separated list of prefixes
+    local-prefixes: github.com/GoogleCloudPlatform/opentelemetry-operations-go
+  golint:
+    # minimal confidence for issues, default is 0.8
+    min-confidence: 0.8
+  govet:
+    # report about shadowed variables
+    check-shadowing: true
+
+    # settings per analyzer
+    settings:
+      printf: # analyzer name, run `go tool vet help` to see all analyzers
+        funcs: # run `go tool vet help printf` to see available settings for `printf` analyzer
+        - (github.com/golangci/golangci-lint/pkg/logutils.Log).Infof
+        - (github.com/golangci/golangci-lint/pkg/logutils.Log).Warnf
+        - (github.com/golangci/golangci-lint/pkg/logutils.Log).Errorf
+        - (github.com/golangci/golangci-lint/pkg/logutils.Log).Fatalf
+  
+    enable-all: true
+  misspell:
+    # Correct spellings using locale preferences for US or UK.
+    # Default is to use a neutral variety of English.
+    # Setting locale to US will correct the British spelling of 'colour' to 'color'.
+    locale: US
+    ignore-words:
+    - cancelled
+    - metre
+    - meter
+    - metres
+    - kilometre
+    - kilometres
+
+linters:
+  disable-all: true
+  enable:
+    - gofmt
+    - goimports
+    - golint
+    - govet
+    - staticcheck
+    - misspell
+    - scopelint
+
+issues:
+  # Excluding configuration per-path, per-linter, per-text and per-source
+  exclude-rules:
+    # Exclude some linters from running on tests files.
+    - path: _test\.go
+      linters:
+        - scopelint


Test patch:
diff --git a/exporter/trace/cloudtrace_test.go b/exporter/trace/cloudtrace_test.go
index a07b64a..0f13d6b 100644
--- a/exporter/trace/cloudtrace_test.go
+++ b/exporter/trace/cloudtrace_test.go
@@ -20,20 +20,22 @@ import (
 	"log"
 	"net"
 	"os"
+	"regexp"
 	"sync"
 	"testing"
 	"time"
 
-	texporter "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace"
 	emptypb "github.com/golang/protobuf/ptypes/empty"
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/api/option"
 	tracepb "google.golang.org/genproto/googleapis/devtools/cloudtrace/v2"
 	"google.golang.org/grpc"
 
+	texporter "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace"
+
 	"go.opentelemetry.io/otel/api/global"
-	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	export "go.opentelemetry.io/otel/sdk/export/trace"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 )
 
 type mockTraceServer struct {
@@ -42,7 +44,7 @@ type mockTraceServer struct {
 	spansUploaded []*tracepb.Span
 	delay         time.Duration
 	// onUpload is called every time BatchWriteSpans is called
-	onUpload      func(ctx context.Context, spans []*tracepb.Span)
+	onUpload func(ctx context.Context, spans []*tracepb.Span)
 }
 
 func (s *mockTraceServer) BatchWriteSpans(ctx context.Context, req *tracepb.BatchWriteSpansRequest) (*emptypb.Empty, error) {
@@ -98,35 +100,29 @@ func TestMain(m *testing.M) {
 	os.Exit(m.Run())
 }
 
-func TestExporter_ExportSpans(t *testing.T) {
+func TestExporter_ExportSpan(t *testing.T) {
 	// Initial test precondition
 	mockTrace.spansUploaded = nil
 	mockTrace.delay = 0
 
 	// Create Google Cloud Trace Exporter
-	exp, err := texporter.NewExporter(
-		texporter.WithProjectID("PROJECT_ID_NOT_REAL"),
-		texporter.WithTraceClientOptions(clientOpt),
-		// handle bundle as soon as span is received
-		texporter.WithBundleCountThreshold(1),
-	)
-	assert.NoError(t, err)
-
-	tp, err := sdktrace.NewProvider(
+	_, flush, err := texporter.InstallNewPipeline(
+		[]texporter.Option{
+			texporter.WithProjectID("PROJECT_ID_NOT_REAL"),
+			texporter.WithTraceClientOptions(clientOpt),
+			// handle bundle as soon as span is received
+			texporter.WithBundleCountThreshold(1),
+		},
 		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
-		sdktrace.WithBatcher(exp, // add following two options to ensure flush
-			sdktrace.WithBatchTimeout(1),
-			sdktrace.WithMaxExportBatchSize(1),
-		))
+	)
 	assert.NoError(t, err)
 
-	global.SetTraceProvider(tp)
 	_, span := global.TraceProvider().Tracer("test-tracer").Start(context.Background(), "test-span")
 	span.End()
 	assert.True(t, span.SpanContext().IsValid())
 
 	// wait exporter to flush
-	time.Sleep(20 * time.Millisecond)
+	flush()
 	assert.EqualValues(t, 1, mockTrace.len())
 }
 
@@ -134,35 +130,32 @@ func TestExporter_DisplayNameFormatter(t *testing.T) {
 	// Initial test precondition
 	mockTrace.spansUploaded = nil
 	mockTrace.delay = 0
-	
+
 	spanName := "span1234"
 	format := func(s *export.SpanData) string {
 		return "TEST_FORMAT" + s.Name
 	}
 
 	// Create Google Cloud Trace Exporter
-	exp, err := texporter.NewExporter(
-		texporter.WithProjectID("PROJECT_ID_NOT_REAL"),
-		texporter.WithTraceClientOptions(clientOpt),
-		texporter.WithBundleCountThreshold(1),
-		texporter.WithDisplayNameFormatter(format),
-	)
-	assert.NoError(t, err)
-
-	tp, err := sdktrace.NewProvider(
+	_, flush, err := texporter.InstallNewPipeline(
+		[]texporter.Option{
+			texporter.WithProjectID("PROJECT_ID_NOT_REAL"),
+			texporter.WithTraceClientOptions(clientOpt),
+			texporter.WithBundleCountThreshold(1),
+			texporter.WithDisplayNameFormatter(format),
+		},
 		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
-		sdktrace.WithSyncer(exp))
+	)
 	assert.NoError(t, err)
 
-	global.SetTraceProvider(tp)
 	_, span := global.TraceProvider().Tracer("test-tracer").Start(context.Background(), spanName)
 	span.End()
 	assert.True(t, span.SpanContext().IsValid())
 
 	// wait exporter to flush
-	time.Sleep(20 * time.Millisecond)
+	flush()
 	assert.EqualValues(t, 1, mockTrace.len())
-	assert.EqualValues(t, "TEST_FORMAT" + spanName, mockTrace.spansUploaded[0].DisplayName.Value)
+	assert.EqualValues(t, "TEST_FORMAT"+spanName, mockTrace.spansUploaded[0].DisplayName.Value)
 }
 
 func TestExporter_Timeout(t *testing.T) {
@@ -170,39 +163,35 @@ func TestExporter_Timeout(t *testing.T) {
 	mockTrace.spansUploaded = nil
 	mockTrace.delay = 20 * time.Millisecond
 	var exportErrors []error
-	ch := make(chan error)
 
 	// Create Google Cloud Trace Exporter
-	exp, err := texporter.NewExporter(
-		texporter.WithProjectID("PROJECT_ID_NOT_REAL"),
-		texporter.WithTraceClientOptions(clientOpt),
-		texporter.WithTimeout(1*time.Millisecond),
-		// handle bundle as soon as span is received
-		texporter.WithBundleCountThreshold(1),
-		texporter.WithOnError(func(err error) {
-			exportErrors = append(exportErrors, err)
-			ch <- err
-		}),
-	)
-	assert.NoError(t, err)
-
-	tp, err := sdktrace.NewProvider(
+	_, flush, err := texporter.InstallNewPipeline(
+		[]texporter.Option{
+			texporter.WithProjectID("PROJECT_ID_NOT_REAL"),
+			texporter.WithTraceClientOptions(clientOpt),
+			texporter.WithTimeout(1 * time.Millisecond),
+			// handle bundle as soon as span is received
+			texporter.WithBundleCountThreshold(1),
+			texporter.WithOnError(func(err error) {
+				exportErrors = append(exportErrors, err)
+			}),
+		},
 		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
-		sdktrace.WithSyncer(exp))
+	)
 	assert.NoError(t, err)
 
-	global.SetTraceProvider(tp)
 	_, span := global.TraceProvider().Tracer("test-tracer").Start(context.Background(), "test-span")
 	span.End()
 	assert.True(t, span.SpanContext().IsValid())
 
 	// wait for error to be handled
-	<-ch
+	flush()
 	assert.EqualValues(t, 0, mockTrace.len())
 	if got, want := len(exportErrors), 1; got != want {
 		t.Fatalf("len(exportErrors) = %q; want %q", got, want)
 	}
-	if got, want := exportErrors[0].Error(), "rpc error: code = DeadlineExceeded desc = context deadline exceeded"; got != want {
+	got, want := exportErrors[0].Error(), "rpc error: code = (DeadlineExceeded|Unknown) desc = context deadline exceeded"
+	if match, _ := regexp.MatchString(want, got); !match {
 		t.Fatalf("err.Error() = %q; want %q", got, want)
 	}
 }
@@ -217,21 +206,17 @@ func TestBundling(t *testing.T) {
 	}
 	mockTrace.mu.Unlock()
 
-	exporter, err := texporter.NewExporter(
-		texporter.WithProjectID("PROJECT_ID_NOT_REAL"),
-		texporter.WithTraceClientOptions(clientOpt),
-		texporter.WithBundleDelayThreshold(time.Second / 10),
-		texporter.WithBundleCountThreshold(10),
-	)
-	assert.NoError(t, err) 
-
-	tp, err := sdktrace.NewProvider(
+	_, _, err := texporter.InstallNewPipeline(
+		[]texporter.Option{
+			texporter.WithProjectID("PROJECT_ID_NOT_REAL"),
+			texporter.WithTraceClientOptions(clientOpt),
+			texporter.WithBundleDelayThreshold(time.Second / 10),
+			texporter.WithBundleCountThreshold(10),
+		},
 		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
-		sdktrace.WithSyncer(exporter))
+	)
 	assert.NoError(t, err)
 
-	global.SetTraceProvider(tp)
-
 	for i := 0; i < 35; i++ {
 		_, span := global.TraceProvider().Tracer("test-tracer").Start(context.Background(), "test-span")
 		span.End()
@@ -258,7 +243,6 @@ func TestBundling(t *testing.T) {
 	}
 }
 
-
 func TestBundling_ConcurrentExports(t *testing.T) {
 	mockTrace.mu.Lock()
 	mockTrace.spansUploaded = nil
@@ -281,22 +265,18 @@ func TestBundling_ConcurrentExports(t *testing.T) {
 	workers := 3
 	spansPerWorker := 50
 	delay := 2 * time.Second
-	exporter, err := texporter.NewExporter(
-		texporter.WithProjectID("PROJECT_ID_NOT_REAL"),
-		texporter.WithTraceClientOptions(clientOpt),
-		texporter.WithBundleDelayThreshold(delay),
-		texporter.WithBundleCountThreshold(spansPerWorker),
-		texporter.WithMaxNumberOfWorkers(workers),
-	)
-	assert.NoError(t, err) 
-
-	tp, err := sdktrace.NewProvider(
+	_, _, err := texporter.InstallNewPipeline(
+		[]texporter.Option{
+			texporter.WithProjectID("PROJECT_ID_NOT_REAL"),
+			texporter.WithTraceClientOptions(clientOpt),
+			texporter.WithBundleDelayThreshold(delay),
+			texporter.WithBundleCountThreshold(spansPerWorker),
+			texporter.WithMaxNumberOfWorkers(workers),
+		},
 		sdktrace.WithConfig(sdktrace.Config{DefaultSampler: sdktrace.AlwaysSample()}),
-		sdktrace.WithSyncer(exporter))
+	)
 	assert.NoError(t, err)
 
-	global.SetTraceProvider(tp)
-
 	waitCh := make(chan struct{})
 	wg.Add(workers)