Backport changes to contrib stackdriverexporter

exporter/stackdriverexporter/config.go

@@ -23,4 +23,5 @@ type Config struct {
 	EnableTracing                 bool                     `mapstructure:"enable-tracing"`
 	EnableMetrics                 bool                     `mapstructure:"enable-metrics"`
 	Prefix                        string                   `mapstructure:"metric-prefix"`
+	Endpoint                      string                   `mapstructure:"endpoint"`
 }

exporter/stackdriverexporter/config_test.go

@@ -26,13 +26,13 @@ import (
 )
 
 func TestLoadConfig(t *testing.T) {
-	receivers, processors, exporters, err := config.ExampleComponents()
+	facotries, err := config.ExampleComponents()
 	assert.Nil(t, err)
 
 	factory := &Factory{}
-	exporters[typeStr] = factory
+	facotries.Exporters[typeStr] = factory
 	cfg, err := config.LoadConfigFile(
-		t, path.Join(".", "testdata", "config.yaml"), receivers, processors, exporters,
+		t, path.Join(".", "testdata", "config.yaml"), facotries,
 	)
 
 	require.NoError(t, err)
@@ -44,9 +44,13 @@ func TestLoadConfig(t *testing.T) {
 	assert.Equal(t, r0, factory.CreateDefaultConfig())
 
 	r1 := cfg.Exporters["stackdriver/customname"].(*Config)
-	assert.Equal(t, r1.ExporterSettings,
-		configmodels.ExporterSettings{
-			TypeVal: typeStr,
-			NameVal: "stackdriver/customname",
+	assert.Equal(t, r1,
+		&Config{
+			ExporterSettings: configmodels.ExporterSettings{TypeVal: typeStr, NameVal: "stackdriver/customname"},
+			ProjectID:        "my-project",
+			EnableTracing:    true,
+			EnableMetrics:    true,
+			Prefix:           "prefix",
+			Endpoint:         "test-endpoint",
 		})
 }

exporter/stackdriverexporter/factory.go

@@ -18,7 +18,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-service/config/configmodels"
-	"github.com/open-telemetry/opentelemetry-service/consumer"
 	"github.com/open-telemetry/opentelemetry-service/exporter"
 )
 
@@ -31,6 +30,8 @@ const (
 type Factory struct {
 }
 
+var _ (exporter.Factory) = (*Factory)(nil)
+
 // Type gets the type of the Exporter config created by this factory.
 func (f *Factory) Type() string {
 	return typeStr
@@ -47,19 +48,19 @@ func (f *Factory) CreateDefaultConfig() configmodels.Exporter {
 }
 
 // CreateTraceExporter creates a trace exporter based on this config.
-func (f *Factory) CreateTraceExporter(logger *zap.Logger, cfg configmodels.Exporter) (consumer.TraceConsumer, exporter.StopFunc, error) {
+func (f *Factory) CreateTraceExporter(logger *zap.Logger, cfg configmodels.Exporter) (exporter.TraceExporter, error) {
 	eCfg := cfg.(*Config)
 	if !eCfg.EnableTracing {
-		return nil, nil, nil
+		return nil, nil
 	}
-	return newStackdriverTraceExporter(eCfg.ProjectID, eCfg.Prefix)
+	return newStackdriverTraceExporter(eCfg)
 }
 
 // CreateMetricsExporter creates a metrics exporter based on this config.
-func (f *Factory) CreateMetricsExporter(logger *zap.Logger, cfg configmodels.Exporter) (consumer.MetricsConsumer, exporter.StopFunc, error) {
+func (f *Factory) CreateMetricsExporter(logger *zap.Logger, cfg configmodels.Exporter) (exporter.MetricsExporter, error) {
 	eCfg := cfg.(*Config)
 	if !eCfg.EnableMetrics {
-		return nil, nil, nil
+		return nil, nil
 	}
-	return newStackdriverMetricsExporter(eCfg.ProjectID, eCfg.Prefix)
+	return newStackdriverMetricsExporter(eCfg)
 }

exporter/stackdriverexporter/factory_test.go

@@ -31,9 +31,9 @@ func TestCreateExporter(t *testing.T) {
 	factory := Factory{}
 	cfg := factory.CreateDefaultConfig()
 
-	_, _, err := factory.CreateTraceExporter(zap.NewNop(), cfg)
+	_, err := factory.CreateTraceExporter(zap.NewNop(), cfg)
 	assert.Nil(t, err)
 
-	_, _, err = factory.CreateMetricsExporter(zap.NewNop(), cfg)
+	_, err = factory.CreateMetricsExporter(zap.NewNop(), cfg)
 	assert.Nil(t, err)
 }

exporter/stackdriverexporter/go.mod

@@ -4,8 +4,9 @@ go 1.12
 
 require (
 	contrib.go.opencensus.io/exporter/stackdriver v0.12.6
-	github.com/open-telemetry/opentelemetry-service v0.0.2-0.20190808233811-8c8603fad686
+	github.com/open-telemetry/opentelemetry-service v0.0.2-0.20190904165913-41a7afa548c8
 	github.com/stretchr/testify v1.3.0
-	go.opencensus.io v0.22.0
+	go.opencensus.io v0.22.1 // indirect
 	go.uber.org/zap v1.10.0
+	google.golang.org/api v0.7.0
 )

exporter/stackdriverexporter/go.sum

@@ -85,6 +85,8 @@ github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekf
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/groupcache v0.0.0-20180924190550-6f2cf27854a4/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
+github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6 h1:ZgQEtGgCBiWRM39fZuwSd1LwSqqSW0hOdXCYYDX0R3I=
+github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y=
@@ -191,13 +193,12 @@ github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRW
 github.com/oklog/oklog v0.0.0-20170918173356-f857583a70c3/go.mod h1:FCV+B7mhrz4o+ueLpx+KqkyXRGMWOYEvfiXtdGtbWGs=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
 github.com/olekukonko/tablewriter v0.0.0-20180912035003-be2c049b30cc/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo=
-github.com/omnition/scribe-go v0.0.0-20190131012523-9e3c68f31124/go.mod h1:GnPmaNTr3pdt/V0JmVNVgDq+JEMb/oXxNlsG+pN6gg4=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/gomega v1.4.1/go.mod h1:C1qb7wdrVGGVU+Z6iS04AVkA3Q65CEZX59MT0QO5uiA=
 github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
-github.com/open-telemetry/opentelemetry-service v0.0.2-0.20190808233811-8c8603fad686 h1:mStshityB+e8YsR9ukfsG1uX8KhEUC0D4nooCrHJs1s=
-github.com/open-telemetry/opentelemetry-service v0.0.2-0.20190808233811-8c8603fad686/go.mod h1:4lgEb8KPAigABad+b3dwOM77L97s0rwxsxzI3ja/op8=
+github.com/open-telemetry/opentelemetry-service v0.0.2-0.20190904165913-41a7afa548c8 h1:LV9EPsTfig/OA6KQPKRaPvKnxdRHOIBm4/danwt+6/o=
+github.com/open-telemetry/opentelemetry-service v0.0.2-0.20190904165913-41a7afa548c8/go.mod h1:GY8PME33TPhlBLyggcwaP8d5gNeF9iPNkoWXC/k8U9k=
 github.com/opentracing-contrib/go-stdlib v0.0.0-20170113013457-1de4cc2120e7/go.mod h1:PLldrQSroqzH70Xl+1DQcGnefIbqsKR7UDaiux3zV+w=
 github.com/opentracing/basictracer-go v1.0.0/go.mod h1:QfBfYuafItcjQuMwinw9GhYKwFXS9KnPs5lxoYwgW74=
 github.com/opentracing/opentracing-go v1.0.1/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
@@ -281,6 +282,8 @@ go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 go.opencensus.io v0.22.0 h1:C9hSCOW830chIVkdja34wa6Ky+IzWllkUinR+BtRZd4=
 go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
+go.opencensus.io v0.22.1 h1:8dP3SGL7MPB94crU3bEPplMPe83FI4EouesJUeFHv50=
+go.opencensus.io v0.22.1/go.mod h1:Ap50jQcDJrx6rB6VgeeFPtuPIf3wMRvRfrfYDO6+BmA=
 go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI=

exporter/stackdriverexporter/stackdriver.go

@@ -19,96 +19,105 @@ package stackdriverexporter
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"contrib.go.opencensus.io/exporter/stackdriver"
-	"go.opencensus.io/trace"
+	"google.golang.org/api/option"
 
-	"github.com/open-telemetry/opentelemetry-service/consumer"
 	"github.com/open-telemetry/opentelemetry-service/consumer/consumerdata"
-	"github.com/open-telemetry/opentelemetry-service/exporter/exporterwrapper"
+	"github.com/open-telemetry/opentelemetry-service/exporter"
+	"github.com/open-telemetry/opentelemetry-service/exporter/exporterhelper"
+	"github.com/open-telemetry/opentelemetry-service/oterr"
+	spandatatranslator "github.com/open-telemetry/opentelemetry-service/translator/trace/spandata"
 )
 
-// TODO: Add metrics support to the exporterwrapper.
+// stackdriverExporter is a wrapper struct of Stackdriver exporter
 type stackdriverExporter struct {
 	exporter *stackdriver.Exporter
 }
 
-var _ consumer.MetricsConsumer = (*stackdriverExporter)(nil)
-
-func newStackdriverTraceExporter(ProjectID, MetricPrefix string) (consumer.TraceConsumer, func() error, error) {
-	sde, serr := newStackdriverExporter(ProjectID, MetricPrefix)
-	if serr != nil {
-		return nil, nil, fmt.Errorf("cannot configure Stackdriver Trace exporter: %v", serr)
-	}
-
-	tExp, err := exporterwrapper.NewExporterWrapper("stackdriver_trace", "ocservice.exporter.Stackdriver.ConsumeTraceData", sde)
-	if err != nil {
-		return nil, nil, err
-	}
-	// TODO: Examine "contrib.go.opencensus.io/exporter/stackdriver" to see
-	// if trace.ExportSpan was constraining and if perhaps the Stackdriver
-	// upload can use the context and information from the Node.
-
-	doneFn := func() error {
-		sde.Flush()
-		return nil
-	}
+func (*stackdriverExporter) Name() string {
+	return "stackdriver"
+}
 
-	return tExp, doneFn, nil
+func (se *stackdriverExporter) Shutdown() error {
+	se.exporter.Flush()
+	se.exporter.StopMetricsExporter()
+	return nil
 }
 
-func newStackdriverMetricsExporter(ProjectID, MetricPrefix string) (consumer.MetricsConsumer, func() error, error) {
-	sde, serr := newStackdriverExporter(ProjectID, MetricPrefix)
+func newStackdriverTraceExporter(cfg *Config) (exporter.TraceExporter, error) {
+	options := getStackdriverOptions(cfg.ProjectID, cfg.Prefix)
+	if cfg.Endpoint != "" {
+		options.TraceClientOptions = []option.ClientOption{option.WithEndpoint(cfg.Endpoint)}
+	}
+	sde, serr := stackdriver.NewExporter(options)
 	if serr != nil {
-		return nil, nil, fmt.Errorf("cannot configure Stackdriver metric exporter: %v", serr)
+		return nil, fmt.Errorf("cannot configure Stackdriver Trace exporter: %v", serr)
 	}
+	tExp := &stackdriverExporter{exporter: sde}
+
+	return exporterhelper.NewTraceExporter(
+		cfg,
+		tExp.pushTraceData,
+		exporterhelper.WithTracing(true),
+		exporterhelper.WithMetrics(true),
+		exporterhelper.WithShutdown(tExp.Shutdown))
+}
 
-	mExp := &stackdriverExporter{
-		exporter: sde,
+func newStackdriverMetricsExporter(cfg *Config) (exporter.MetricsExporter, error) {
+	options := getStackdriverOptions(cfg.ProjectID, cfg.Prefix)
+	if cfg.Endpoint != "" {
+		options.MonitoringClientOptions = []option.ClientOption{option.WithEndpoint(cfg.Endpoint)}
 	}
-
-	doneFn := func() error {
-		sde.Flush()
-		return nil
+	sde, serr := stackdriver.NewExporter(options)
+	if serr != nil {
+		return nil, fmt.Errorf("cannot configure Stackdriver metric exporter: %v", serr)
 	}
-
-	return mExp, doneFn, nil
+	mExp := &stackdriverExporter{exporter: sde}
+
+	return exporterhelper.NewMetricsExporter(
+		cfg,
+		mExp.pushMetricsData,
+		exporterhelper.WithTracing(true),
+		exporterhelper.WithMetrics(true),
+		exporterhelper.WithShutdown(mExp.Shutdown))
 }
 
-func newStackdriverExporter(ProjectID, MetricPrefix string) (*stackdriver.Exporter, error) {
+func getStackdriverOptions(ProjectID, MetricPrefix string) stackdriver.Options {
 	// TODO:  For each ProjectID, create a different exporter
 	// or at least a unique Stackdriver client per ProjectID.
 
-	return stackdriver.NewExporter(stackdriver.Options{
+	return stackdriver.Options{
 		// If the project ID is an empty string, it will be set by default based on
 		// the project this is running on in GCP.
 		ProjectID: ProjectID,
 
 		MetricPrefix: MetricPrefix,
 
-		// Stackdriver Metrics mandates a minimum of 60 seconds for
-		// reporting metrics. We have to enforce this as per the advisory
-		// at https://cloud.google.com/monitoring/custom-metrics/creating-metrics#writing-ts
-		// which says:
-		//
-		// "If you want to write more than one point to the same time series, then use a separate call
-		//  to the timeSeries.create method for each point. Don't make the calls faster than one time per
-		//  minute. If you are adding data points to different time series, then there is no rate limitation."
-		BundleDelayThreshold: 61 * time.Second,
-	})
+		// Set DefaultMonitoringLabels to an empty map to avoid getting the "opencensus_task" label
+		DefaultMonitoringLabels: &stackdriver.Labels{},
+	}
 }
 
-func (sde *stackdriverExporter) ConsumeMetricsData(ctx context.Context, md consumerdata.MetricsData) error {
-	ctx, span := trace.StartSpan(ctx,
-		"opentelemetry.service.exporter.stackdriver.ExportMetricsData",
-		trace.WithSampler(trace.NeverSample()))
-	defer span.End()
+// pushMetricsData is a wrapper method on StackdriverExporter.PushMetricsProto
+func (se *stackdriverExporter) pushMetricsData(ctx context.Context, md consumerdata.MetricsData) (int, error) {
+	return se.exporter.PushMetricsProto(ctx, md.Node, md.Resource, md.Metrics)
+}
 
-	err := sde.exporter.ExportMetricsProto(ctx, md.Node, md.Resource, md.Metrics)
-	if err != nil {
-		span.SetStatus(trace.Status{Code: trace.StatusCodeInternal, Message: err.Error()})
+// TODO(songya): add an interface PushSpanProto to Stackdriver exporter and remove this method
+// pushTraceData is a wrapper method on StackdriverExporter.PushSpans
+func (se *stackdriverExporter) pushTraceData(ctx context.Context, td consumerdata.TraceData) (int, error) {
+	var errs []error
+	droppedSpans := 0
+	for _, span := range td.Spans {
+		sd, err := spandatatranslator.ProtoSpanToOCSpanData(span)
+		if err == nil {
+			se.exporter.ExportSpan(sd)
+		} else {
+			droppedSpans++
+			errs = append(errs, err)
+		}
 	}
 
-	return err
+	return droppedSpans, oterr.CombineErrors(errs)
 }

exporter/stackdriverexporter/testdata/config.yaml

@@ -11,6 +11,7 @@ exporters:
     enable-tracing: true
     enable-metrics: true
     metric-prefix: prefix
+    endpoint: test-endpoint
   stackdriver/disabled: # will be ignored
     disabled: true