[sdk/trace] support trace exporter that uses a batch processor and otlp exporter

go.mod

@@ -3,14 +3,27 @@ module github.com/lightstep/otel-launcher-go
 go 1.18
 
 require (
+	github.com/f5/otel-arrow-adapter v0.0.0-20230522233737-dc34685b4828
+	github.com/google/go-cmp v0.5.9
 	github.com/lightstep/otel-launcher-go/lightstep/sdk/metric v1.17.1
 	github.com/lightstep/otel-launcher-go/pipelines v1.17.1
 	github.com/sethvargo/go-envconfig v0.8.3
 	github.com/stretchr/testify v1.8.3
+	go.opentelemetry.io/collector v0.78.1
+	go.opentelemetry.io/collector/component v0.78.1
+	go.opentelemetry.io/collector/consumer v0.78.1
+	go.opentelemetry.io/collector/exporter v0.78.1
+	go.opentelemetry.io/collector/pdata v1.0.0-rcv0012
+	go.opentelemetry.io/collector/processor/batchprocessor v0.78.1
+	go.opentelemetry.io/collector/receiver v0.78.1
 	go.opentelemetry.io/otel v1.15.1
 	go.opentelemetry.io/otel/metric v0.38.1
 	go.opentelemetry.io/otel/sdk v1.15.1
 	go.opentelemetry.io/otel/trace v1.15.1
+	go.opentelemetry.io/proto/otlp v0.19.0
+	go.uber.org/multierr v1.11.0
+	go.uber.org/zap v1.24.0
+	google.golang.org/protobuf v1.30.0
 )
 
 require (
@@ -23,7 +36,7 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13 // indirect
 	github.com/dgryski/go-metro v0.0.0-20180109044635-280f6062b5bc // indirect
-	github.com/f5/otel-arrow-adapter v0.0.0-20230522233737-dc34685b4828 // indirect
+	github.com/felixge/httpsnoop v1.0.3 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.4.0 // indirect
 	github.com/go-logr/logr v1.2.4 // indirect
@@ -54,6 +67,7 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.17 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
+	github.com/rs/cors v1.9.0 // indirect
 	github.com/shirou/gopsutil/v3 v3.23.4 // indirect
 	github.com/shoenig/go-m1cpu v0.1.5 // indirect
 	github.com/tklauser/go-sysconf v0.3.11 // indirect
@@ -62,17 +76,11 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
 	go.opencensus.io v0.24.0 // indirect
-	go.opentelemetry.io/collector v0.78.1 // indirect
-	go.opentelemetry.io/collector/component v0.78.1 // indirect
 	go.opentelemetry.io/collector/confmap v0.78.1 // indirect
-	go.opentelemetry.io/collector/consumer v0.78.1 // indirect
-	go.opentelemetry.io/collector/exporter v0.78.1 // indirect
 	go.opentelemetry.io/collector/featuregate v1.0.0-rcv0012 // indirect
-	go.opentelemetry.io/collector/pdata v1.0.0-rcv0012 // indirect
-	go.opentelemetry.io/collector/processor/batchprocessor v0.78.1 // indirect
-	go.opentelemetry.io/collector/receiver v0.78.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.41.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/host v0.41.1 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.41.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/runtime v0.41.1 // indirect
 	go.opentelemetry.io/contrib/propagators/b3 v1.16.1 // indirect
 	go.opentelemetry.io/contrib/propagators/ot v1.16.1 // indirect
@@ -82,10 +90,7 @@ require (
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.15.1 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.14.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v0.38.1 // indirect
-	go.opentelemetry.io/proto/otlp v0.19.0 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
-	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.24.0 // indirect
 	golang.org/x/exp v0.0.0-20230321023759-10a507213a29 // indirect
 	golang.org/x/mod v0.9.0 // indirect
 	golang.org/x/net v0.10.0 // indirect
@@ -95,7 +100,6 @@ require (
 	golang.org/x/xerrors v0.0.0-20220609144429-65e65417b02f // indirect
 	google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
 	google.golang.org/grpc v1.55.0 // indirect
-	google.golang.org/protobuf v1.30.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 

go.sum

@@ -120,6 +120,7 @@ github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5Kwzbycv
 github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
 github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga6PJ7M=
 github.com/felixge/httpsnoop v1.0.3 h1:s/nj+GCswXYzN5v2DpNMuMQYe+0DDwt5WVCU6CWBdXk=
+github.com/felixge/httpsnoop v1.0.3/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/fogleman/gg v1.2.1-0.20190220221249-0403632d5b90/go.mod h1:R/bRT+9gY/C5z7JzPU0zXsXHKM4/ayA+zqcVNZzPa1k=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
@@ -393,6 +394,7 @@ github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6L
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/rs/cors v1.9.0 h1:l9HGsTsHJcvW14Nk7J9KFz8bzeAWXn3CG6bgt7LsrAE=
+github.com/rs/cors v1.9.0/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/ryanuber/columnize v2.1.0+incompatible/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/ryanuber/go-glob v1.0.0/go.mod h1:807d1WSdnB0XRJzKNil9Om6lcp/3a0v4qIHxIXzX/Yc=
@@ -470,11 +472,13 @@ go.opentelemetry.io/collector/processor/batchprocessor v0.78.1 h1:rrKVGvbnzWls1U
 go.opentelemetry.io/collector/processor/batchprocessor v0.78.1/go.mod h1:6rdnvQXPzo0sIhrUVK4bpZ7EmDdR+bSpTdf7gmChhUU=
 go.opentelemetry.io/collector/receiver v0.78.1 h1:MSgnfyIk2OATOvVLJhTFH29rObN6umCEqbeuGLfdUhw=
 go.opentelemetry.io/collector/receiver v0.78.1/go.mod h1:3uz9mktAkTBFcAkwapirH6InuPUL44Oa6MnoXmQJ2ms=
+go.opentelemetry.io/collector/semconv v0.78.1 h1:YlhokDVTP+gw6yKA0Jc2FcfddhD+a6E5Ixmby5xBWs0=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.41.1 h1:Ei1FUQ5CbSNkl2o/XAiksXSyQNAeJBX3ivqJpJ254Ak=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.41.1/go.mod h1:f7TOPTlEcliCBlOYPuNnZTuND71MVTAoINWIt1SmP/c=
 go.opentelemetry.io/contrib/instrumentation/host v0.41.1 h1:B/s7HG33iwus4cwd8NsPk9H6nLKJpLcLgGV2D9RkfsI=
 go.opentelemetry.io/contrib/instrumentation/host v0.41.1/go.mod h1:OdaD/6LjpX5H0ZCKf9EgR6yKFQihVizQMBRLFGlyTP4=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.41.1 h1:pX+lppB8PArapyhS6nBStyQmkaDUPWdQf0UmEGRCQ54=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.41.1/go.mod h1:2FmkXne0k9nkp27LD/m+uoh8dNlstsiCJ7PLc/S72aI=
 go.opentelemetry.io/contrib/instrumentation/runtime v0.41.1 h1:KXWR7rFIuQLMo/dHu/DTev7qdQdTuGCxTfEUZdLaMfs=
 go.opentelemetry.io/contrib/instrumentation/runtime v0.41.1/go.mod h1:fNv3vYJmbcUl4qVQK9tbAVBQjTAbfDDQa/cFgpOTm50=
 go.opentelemetry.io/contrib/propagators/b3 v1.16.1 h1:Y9Dk1kR93eSHadRTkqnm+QyQVhHthCcvTkoP/Afh7+4=

lightstep/sdk/trace/exporters/otlp/otelcol/client.go

@@ -0,0 +1,212 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// 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
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package otelcol
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/f5/otel-arrow-adapter/collector/gen/exporter/otlpexporter"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configcompression"
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/config/configopaque"
+	"go.opentelemetry.io/collector/config/configtls"
+	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/batchprocessor"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/sdk/trace"
+	"go.uber.org/multierr"
+	"go.uber.org/zap"
+)
+
+type Option func(*Config)
+
+type Config struct {
+	Batcher  batchprocessor.Config
+	Exporter otlpexporter.Config
+}
+
+type client struct {
+	once     sync.Once
+	resource pcommon.Resource
+
+	exporter exporter.Traces
+	batcher  processor.Traces
+	settings exporter.CreateSettings
+}
+
+func (c *client) ExportSpans(ctx context.Context, spans []trace.ReadOnlySpan) error {
+	return c.batcher.ConsumeTraces(ctx, c.d2pd(spans))
+}
+
+func (c *client) Shutdown(ctx context.Context) error {
+	var err error
+	if err1 := c.batcher.Shutdown(ctx); err1 != nil {
+		err = multierr.Append(err, err1)
+	}
+	if err2 := c.exporter.Shutdown(ctx); err2 != nil {
+		err = multierr.Append(err, err2)
+	}
+	return err
+}
+
+func NewDefaultConfig() Config {
+	return Config{
+		Batcher: batchprocessor.Config{
+			Timeout:          0,
+			SendBatchSize:    0,
+			SendBatchMaxSize: 10000,
+		},
+		Exporter: otlpexporter.Config{
+			TimeoutSettings: exporterhelper.TimeoutSettings{
+				Timeout: 10 * time.Second,
+			},
+			RetrySettings: exporterhelper.RetrySettings{
+				Enabled: false,
+			},
+			QueueSettings: exporterhelper.QueueSettings{
+				Enabled: false,
+			},
+			GRPCClientSettings: configgrpc.GRPCClientSettings{
+				Headers:         map[string]configopaque.String{},
+				Compression:     configcompression.Zstd,
+				WriteBufferSize: 512 * 1024,
+				WaitForReady:    true,
+			},
+			Arrow: otlpexporter.ArrowSettings{
+				Disabled:         true,
+				NumStreams:       1,
+				DisableDowngrade: true,
+			},
+		},
+	}
+}
+
+func NewConfig(opts ...Option) Config {
+	cfg := NewDefaultConfig()
+	for _, option := range opts {
+		if option != nil {
+			option(&cfg)
+		}
+	}
+	return cfg
+}
+
+func WithEndpoint(addr string) Option {
+	return func(cfg *Config) {
+		cfg.Exporter.GRPCClientSettings.Endpoint = addr
+	}
+}
+
+func WithHeaders(hdrs map[string]string) Option {
+	return func(cfg *Config) {
+		for key, val := range hdrs {
+			cfg.Exporter.GRPCClientSettings.Headers[key] = configopaque.String(val)
+		}
+	}
+}
+
+func WithCompressor(comp string) Option {
+	return func(cfg *Config) {
+		cfg.Exporter.GRPCClientSettings.Compression = configcompression.CompressionType(comp)
+	}
+}
+
+func WithInsecure() Option {
+	return func(cfg *Config) {
+		cfg.Exporter.GRPCClientSettings.TLSSetting.Insecure = true
+	}
+}
+
+func WithTLSSetting(tlss configtls.TLSClientSetting) Option {
+	return func(cfg *Config) {
+		cfg.Exporter.GRPCClientSettings.TLSSetting = tlss
+	}
+}
+
+func NewExporter(ctx context.Context, cfg Config) (trace.SpanExporter, error) {
+	c := &client{}
+
+	if !cfg.Exporter.Arrow.Disabled {
+		c.settings.ID = component.NewID("otlp/arrow")
+	} else {
+		c.settings.ID = component.NewID("otlp/proto")
+	}
+	logger, err := zap.NewProduction()
+	if err != nil {
+		return nil, err
+	}
+
+	c.settings.TelemetrySettings.Logger = logger
+
+	// Note: this may be too much tracing.
+	c.settings.TelemetrySettings.TracerProvider = otel.GetTracerProvider()
+
+	exp, err := otlpexporter.NewFactory().CreateTracesExporter(ctx, c.settings, &cfg.Exporter)
+	if err != nil {
+		return nil, err
+	}
+
+	bset := processor.CreateSettings{
+		ID:                component.NewID("otlp/batch"),
+		TelemetrySettings: c.settings.TelemetrySettings,
+		BuildInfo:         c.settings.BuildInfo,
+	}
+
+	bat, err := batchprocessor.NewFactory().CreateTracesProcessor(ctx, bset, &cfg.Batcher, exp)
+	if err != nil {
+		return nil, err
+	}
+
+	c.exporter = exp
+	c.batcher = bat
+
+	err = exp.Start(ctx, c)
+	if err != nil {
+		return nil, err
+	}
+
+	err = bat.Start(ctx, c)
+	if err != nil {
+		return nil, err
+	}
+
+	return c, nil
+}
+
+// ReportFatalError implements component.Host.
+func (c *client) ReportFatalError(err error) {
+	c.settings.Logger.Fatal("exporter fatal", zap.Error(err))
+}
+
+// GetFactory implements component.Host.
+func (c *client) GetFactory(component.Kind, component.Type) component.Factory {
+	return nil
+}
+
+// GetExtensions implements component.Host.
+func (c *client) GetExtensions() map[component.ID]component.Component {
+	return nil
+}
+
+// GetExporters implements component.Host.
+func (c *client) GetExporters() map[component.DataType]map[component.ID]component.Component {
+	return nil
+}