add OTLP export support for internal traces (#8119)

This PR adds support for exporting internal collector traces via the OTel Go OTLP exporters. Closes #8106 ~Follows #8117 --------- Signed-off-by: Alex Boten <aboten@lightstep.com> Co-authored-by: Anthony Mirabella <a9@aneurysm9.com>
open-telemetry · Aug 24, 2023 · f453c51 · f453c51
1 parent cdbe8e7
commit f453c51
Show file tree

Hide file tree

Showing 9 changed files with 461 additions and 41 deletions.
diff --git a/.chloggen/codeboten_add-otlp-trace-exporter.yaml b/.chloggen/codeboten_add-otlp-trace-exporter.yaml
@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: service
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: add OTLP export for internal traces
+
+# One or more tracking issues or pull requests related to the change
+issues: [8106]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
diff --git a/cmd/otelcorecol/go.mod b/cmd/otelcorecol/go.mod
@@ -99,6 +99,9 @@ require (
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric v0.39.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v0.39.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v0.39.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.16.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.16.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.16.0 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.39.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v0.39.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.16.0 // indirect

diff --git a/cmd/otelcorecol/go.sum b/cmd/otelcorecol/go.sum
@@ -936,6 +936,12 @@ go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v0.39.0 h1:rm+
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v0.39.0/go.mod h1:sWFbI3jJ+6JdjOVepA5blpv/TJ20Hw+26561iMbWcwU=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v0.39.0 h1:IZXpCEtI7BbX01DRQEWTGDkvjMB6hEhiEZXS+eg2YqY=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v0.39.0/go.mod h1:xY111jIZtWb+pUUgT4UiiSonAaY2cD2Ts5zvuKLki3o=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.16.0 h1:cbsD4cUcviQGXdw8+bo5x2wazq10SKz8hEbtCRPcU78=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.16.0/go.mod h1:JgXSGah17croqhJfhByOLVY719k1emAXC8MVhCIJlRs=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.16.0 h1:TVQp/bboR4mhZSav+MdgXB8FaRho1RC8UwVn3T0vjVc=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.16.0/go.mod h1:I33vtIe0sR96wfrUcilIzLoA3mLHhRmz9S9Te0S3gDo=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.16.0 h1:iqjq9LAB8aK++sKVcELezzn655JnBNdsDhghU4G/So8=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.16.0/go.mod h1:hGXzO5bhhSHZnKvrDaXB82Y9DRFour0Nz/KrBh7reWw=
 go.opentelemetry.io/otel/exporters/prometheus v0.39.0 h1:whAaiHxOatgtKd+w0dOi//1KUxj3KoPINZdtDaDj3IA=
 go.opentelemetry.io/otel/exporters/prometheus v0.39.0/go.mod h1:4jo5Q4CROlCpSPsXLhymi+LYrDXd2ObU5wbKayfZs7Y=
 go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v0.39.0 h1:fl2WmyenEf6LYYlfHAtCUEDyGcpwJNqD4dHGO7PVm4w=
@@ -954,7 +960,7 @@ go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqe
 go.opentelemetry.io/proto/otlp v0.15.0/go.mod h1:H7XAot3MsfNsj7EXtrA2q5xSNQ10UqI405h3+duxN4U=
 go.opentelemetry.io/proto/otlp v0.19.0 h1:IVN6GR+mhC4s5yfcTbmzHYODqvWAp3ZedA2SJPI1Nnw=
 go.opentelemetry.io/proto/otlp v0.19.0/go.mod h1:H7XAot3MsfNsj7EXtrA2q5xSNQ10UqI405h3+duxN4U=
-go.uber.org/goleak v1.2.0 h1:xqgm/S+aQvhWFTtR0XK3Jvg7z8kGV8P4X14IzwN3Eqk=
+go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.25.0 h1:4Hvk6GtkucQ790dqmj7l1eEnRdKm3k3ZUrUMS2d5+5c=

diff --git a/go.mod b/go.mod
@@ -31,6 +31,8 @@ require (
 	go.opentelemetry.io/otel/bridge/opencensus v0.39.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v0.39.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v0.39.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.16.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.16.0
 	go.opentelemetry.io/otel/exporters/prometheus v0.39.0
 	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v0.39.0
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.16.0
@@ -84,6 +86,7 @@ require (
 	go.opentelemetry.io/contrib/zpages v0.42.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.16.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric v0.39.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.16.0 // indirect
 	go.opentelemetry.io/proto/otlp v0.19.0 // indirect
 	golang.org/x/exp v0.0.0-20230711023510-fffb14384f22 // indirect
 	golang.org/x/net v0.14.0 // indirect

diff --git a/go.sum b/go.sum
@@ -332,6 +332,12 @@ go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v0.39.0 h1:rm+
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v0.39.0/go.mod h1:sWFbI3jJ+6JdjOVepA5blpv/TJ20Hw+26561iMbWcwU=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v0.39.0 h1:IZXpCEtI7BbX01DRQEWTGDkvjMB6hEhiEZXS+eg2YqY=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v0.39.0/go.mod h1:xY111jIZtWb+pUUgT4UiiSonAaY2cD2Ts5zvuKLki3o=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.16.0 h1:cbsD4cUcviQGXdw8+bo5x2wazq10SKz8hEbtCRPcU78=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.16.0/go.mod h1:JgXSGah17croqhJfhByOLVY719k1emAXC8MVhCIJlRs=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.16.0 h1:TVQp/bboR4mhZSav+MdgXB8FaRho1RC8UwVn3T0vjVc=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.16.0/go.mod h1:I33vtIe0sR96wfrUcilIzLoA3mLHhRmz9S9Te0S3gDo=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.16.0 h1:iqjq9LAB8aK++sKVcELezzn655JnBNdsDhghU4G/So8=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.16.0/go.mod h1:hGXzO5bhhSHZnKvrDaXB82Y9DRFour0Nz/KrBh7reWw=
 go.opentelemetry.io/otel/exporters/prometheus v0.39.0 h1:whAaiHxOatgtKd+w0dOi//1KUxj3KoPINZdtDaDj3IA=
 go.opentelemetry.io/otel/exporters/prometheus v0.39.0/go.mod h1:4jo5Q4CROlCpSPsXLhymi+LYrDXd2ObU5wbKayfZs7Y=
 go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v0.39.0 h1:fl2WmyenEf6LYYlfHAtCUEDyGcpwJNqD4dHGO7PVm4w=
@@ -349,7 +355,7 @@ go.opentelemetry.io/otel/trace v1.16.0/go.mod h1:Yt9vYq1SdNz3xdjZZK7wcXv1qv2pwLk
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 go.opentelemetry.io/proto/otlp v0.19.0 h1:IVN6GR+mhC4s5yfcTbmzHYODqvWAp3ZedA2SJPI1Nnw=
 go.opentelemetry.io/proto/otlp v0.19.0/go.mod h1:H7XAot3MsfNsj7EXtrA2q5xSNQ10UqI405h3+duxN4U=
-go.uber.org/goleak v1.2.0 h1:xqgm/S+aQvhWFTtR0XK3Jvg7z8kGV8P4X14IzwN3Eqk=
+go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.25.0 h1:4Hvk6GtkucQ790dqmj7l1eEnRdKm3k3ZUrUMS2d5+5c=

diff --git a/service/internal/proctelemetry/config.go b/service/internal/proctelemetry/config.go
@@ -20,6 +20,8 @@ import (
 	"go.opentelemetry.io/otel/bridge/opencensus"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
 	otelprom "go.opentelemetry.io/otel/exporters/prometheus"
 	"go.opentelemetry.io/otel/exporters/stdout/stdoutmetric"
 	"go.opentelemetry.io/otel/exporters/stdout/stdouttrace"
@@ -83,7 +85,7 @@ func InitMetricReader(ctx context.Context, reader telemetry.MetricReader, asyncE
 	return nil, nil, fmt.Errorf("unsupported metric reader type %v", reader)
 }
 
-func InitSpanProcessor(_ context.Context, processor telemetry.SpanProcessor) (sdktrace.SpanProcessor, error) {
+func InitSpanProcessor(ctx context.Context, processor telemetry.SpanProcessor) (sdktrace.SpanProcessor, error) {
 	if processor.Batch != nil {
 		if processor.Batch.Exporter.Console != nil {
 			exp, err := stdouttrace.New(
@@ -92,32 +94,23 @@ func InitSpanProcessor(_ context.Context, processor telemetry.SpanProcessor) (sd
 			if err != nil {
 				return nil, err
 			}
-			opts := []sdktrace.BatchSpanProcessorOption{}
-			if processor.Batch.ExportTimeout != nil {
-				if *processor.Batch.ExportTimeout < 0 {
-					return nil, fmt.Errorf("invalid export timeout %d", *processor.Batch.ExportTimeout)
-				}
-				opts = append(opts, sdktrace.WithExportTimeout(time.Millisecond*time.Duration(*processor.Batch.ExportTimeout)))
-			}
-			if processor.Batch.MaxExportBatchSize != nil {
-				if *processor.Batch.MaxExportBatchSize < 0 {
-					return nil, fmt.Errorf("invalid batch size %d", *processor.Batch.MaxExportBatchSize)
-				}
-				opts = append(opts, sdktrace.WithMaxExportBatchSize(*processor.Batch.MaxExportBatchSize))
-			}
-			if processor.Batch.MaxQueueSize != nil {
-				if *processor.Batch.MaxQueueSize < 0 {
-					return nil, fmt.Errorf("invalid queue size %d", *processor.Batch.MaxQueueSize)
-				}
-				opts = append(opts, sdktrace.WithMaxQueueSize(*processor.Batch.MaxQueueSize))
+			return initBatchSpanProcessor(processor.Batch, exp)
+		}
+		if processor.Batch.Exporter.Otlp != nil {
+			var err error
+			var exp sdktrace.SpanExporter
+			switch processor.Batch.Exporter.Otlp.Protocol {
+			case protocolProtobufHTTP:
+				exp, err = initOTLPHTTPSpanExporter(ctx, processor.Batch.Exporter.Otlp)
+			case protocolProtobufGRPC:
+				exp, err = initOTLPgRPCSpanExporter(ctx, processor.Batch.Exporter.Otlp)
+			default:
+				return nil, fmt.Errorf("unsupported protocol %q", processor.Batch.Exporter.Otlp.Protocol)
 			}
-			if processor.Batch.ScheduleDelay != nil {
-				if *processor.Batch.ScheduleDelay < 0 {
-					return nil, fmt.Errorf("invalid schedule delay %d", *processor.Batch.ScheduleDelay)
-				}
-				opts = append(opts, sdktrace.WithBatchTimeout(time.Millisecond*time.Duration(*processor.Batch.ScheduleDelay)))
+			if err != nil {
+				return nil, err
 			}
-			return sdktrace.NewBatchSpanProcessor(exp, opts...), nil
+			return initBatchSpanProcessor(processor.Batch, exp)
 		}
 		return nil, errNoValidSpanExporter
 	}
@@ -287,7 +280,14 @@ func initOTLPgRPCExporter(ctx context.Context, otlpConfig *telemetry.OtlpMetric)
 	}
 
 	if otlpConfig.Compression != nil {
-		opts = append(opts, otlpmetricgrpc.WithCompressor(*otlpConfig.Compression))
+		switch *otlpConfig.Compression {
+		case "gzip":
+			opts = append(opts, otlpmetricgrpc.WithCompressor(*otlpConfig.Compression))
+		case "none":
+			break
+		default:
+			return nil, fmt.Errorf("unsupported compression %q", *otlpConfig.Compression)
+		}
 	}
 	if otlpConfig.Timeout != nil {
 		opts = append(opts, otlpmetricgrpc.WithTimeout(time.Millisecond*time.Duration(*otlpConfig.Timeout)))
@@ -335,3 +335,104 @@ func initOTLPHTTPExporter(ctx context.Context, otlpConfig *telemetry.OtlpMetric)
 
 	return otlpmetrichttp.New(ctx, opts...)
 }
+
+func initOTLPgRPCSpanExporter(ctx context.Context, otlpConfig *telemetry.Otlp) (sdktrace.SpanExporter, error) {
+	opts := []otlptracegrpc.Option{}
+
+	if len(otlpConfig.Endpoint) > 0 {
+		u, err := url.ParseRequestURI(normalizeEndpoint(otlpConfig.Endpoint))
+		if err != nil {
+			return nil, err
+		}
+		opts = append(opts, otlptracegrpc.WithEndpoint(u.Host))
+		if u.Scheme == "http" {
+			opts = append(opts, otlptracegrpc.WithInsecure())
+		}
+	}
+
+	if otlpConfig.Compression != nil {
+		switch *otlpConfig.Compression {
+		case "gzip":
+			opts = append(opts, otlptracegrpc.WithCompressor(*otlpConfig.Compression))
+		case "none":
+			break
+		default:
+			return nil, fmt.Errorf("unsupported compression %q", *otlpConfig.Compression)
+		}
+	}
+	if otlpConfig.Timeout != nil && *otlpConfig.Timeout > 0 {
+		opts = append(opts, otlptracegrpc.WithTimeout(time.Millisecond*time.Duration(*otlpConfig.Timeout)))
+	}
+	if len(otlpConfig.Headers) > 0 {
+		opts = append(opts, otlptracegrpc.WithHeaders(otlpConfig.Headers))
+	}
+
+	return otlptracegrpc.New(ctx, opts...)
+}
+
+func initOTLPHTTPSpanExporter(ctx context.Context, otlpConfig *telemetry.Otlp) (sdktrace.SpanExporter, error) {
+	opts := []otlptracehttp.Option{}
+
+	if len(otlpConfig.Endpoint) > 0 {
+		u, err := url.ParseRequestURI(normalizeEndpoint(otlpConfig.Endpoint))
+		if err != nil {
+			return nil, err
+		}
+		opts = append(opts, otlptracehttp.WithEndpoint(u.Host))
+
+		if u.Scheme == "http" {
+			opts = append(opts, otlptracehttp.WithInsecure())
+		}
+		if len(u.Path) > 0 {
+			opts = append(opts, otlptracehttp.WithURLPath(u.Path))
+		}
+	}
+	if otlpConfig.Compression != nil {
+		switch *otlpConfig.Compression {
+		case "gzip":
+			opts = append(opts, otlptracehttp.WithCompression(otlptracehttp.GzipCompression))
+		case "none":
+			opts = append(opts, otlptracehttp.WithCompression(otlptracehttp.NoCompression))
+		default:
+			return nil, fmt.Errorf("unsupported compression %q", *otlpConfig.Compression)
+		}
+	}
+	if otlpConfig.Timeout != nil && *otlpConfig.Timeout > 0 {
+		opts = append(opts, otlptracehttp.WithTimeout(time.Millisecond*time.Duration(*otlpConfig.Timeout)))
+	}
+	if len(otlpConfig.Headers) > 0 {
+		opts = append(opts, otlptracehttp.WithHeaders(otlpConfig.Headers))
+	}
+
+	return otlptracehttp.New(ctx, opts...)
+}
+
+func initBatchSpanProcessor(bsp *telemetry.BatchSpanProcessor, exp sdktrace.SpanExporter) (sdktrace.SpanProcessor, error) {
+	opts := []sdktrace.BatchSpanProcessorOption{}
+	if bsp.ExportTimeout != nil {
+		if *bsp.ExportTimeout < 0 {
+			return nil, fmt.Errorf("invalid export timeout %d", *bsp.ExportTimeout)
+		}
+		opts = append(opts, sdktrace.WithExportTimeout(time.Millisecond*time.Duration(*bsp.ExportTimeout)))
+	}
+	if bsp.MaxExportBatchSize != nil {
+		if *bsp.MaxExportBatchSize < 0 {
+			return nil, fmt.Errorf("invalid batch size %d", *bsp.MaxExportBatchSize)
+		}
+		opts = append(opts, sdktrace.WithMaxExportBatchSize(*bsp.MaxExportBatchSize))
+	}
+	if bsp.MaxQueueSize != nil {
+		if *bsp.MaxQueueSize < 0 {
+			return nil, fmt.Errorf("invalid queue size %d", *bsp.MaxQueueSize)
+		}
+		opts = append(opts, sdktrace.WithMaxQueueSize(*bsp.MaxQueueSize))
+	}
+	if bsp.ScheduleDelay != nil {
+		if *bsp.ScheduleDelay < 0 {
+			return nil, fmt.Errorf("invalid schedule delay %d", *bsp.ScheduleDelay)
+		}
+		opts = append(opts, sdktrace.WithBatchTimeout(time.Millisecond*time.Duration(*bsp.ScheduleDelay)))
+	}
+	return sdktrace.NewBatchSpanProcessor(exp, opts...), nil
+
+}