Skip to content

Commit

Permalink
add OTLP export support for internal traces
Browse files Browse the repository at this point in the history 
This PR adds support for exporting internal collector traces via the
OTel Go OTLP exporters.

Closes #8106

Follows #8117

Signed-off-by: Alex Boten <[email protected]>
  • Loading branch information
Alex Boten committed Jul 21, 2023
1 parent 5613523 commit 411ab40
Show file tree
Hide file tree
Showing 6 changed files with 133 additions and 28 deletions.
3 changes: 3 additions & 0 deletions cmd/otelcorecol/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,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
Expand Down
8 changes: 7 additions & 1 deletion cmd/otelcorecol/go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1034,6 +1034,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=
Expand All @@ -1055,7 +1061,7 @@ go.opentelemetry.io/proto/otlp v0.19.0/go.mod h1:H7XAot3MsfNsj7EXtrA2q5xSNQ10UqI
go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=
go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI=
go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
Expand Down
3 changes: 3 additions & 0 deletions go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -83,6 +85,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
go.uber.org/atomic v1.10.0 // indirect
golang.org/x/net v0.12.0 // indirect
Expand Down
8 changes: 7 additions & 1 deletion go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -434,6 +434,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=
Expand All @@ -454,7 +460,7 @@ go.opentelemetry.io/proto/otlp v0.19.0/go.mod h1:H7XAot3MsfNsj7EXtrA2q5xSNQ10UqI
go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=
go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI=
go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
Expand Down
137 changes: 112 additions & 25 deletions service/internal/proctelemetry/config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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(
Expand All @@ -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 %s", processor.Batch.Exporter.Otlp.Protocol)

Check warning on line 108 in service/internal/proctelemetry/config.go

View check run for this annotation

Codecov / codecov/patch

service/internal/proctelemetry/config.go#L100-L108 

Added lines #L100 - L108 were not covered by tests
}
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

Check warning on line 111 in service/internal/proctelemetry/config.go

View check run for this annotation

Codecov / codecov/patch

service/internal/proctelemetry/config.go#L110-L111 

Added lines #L110 - L111 were not covered by tests
}
return sdktrace.NewBatchSpanProcessor(exp, opts...), nil
return initBatchSpanProcessor(processor.Batch, exp)

Check warning on line 113 in service/internal/proctelemetry/config.go

View check run for this annotation

Codecov / codecov/patch

service/internal/proctelemetry/config.go#L113 

Added line #L113 was not covered by tests
}
return nil, errNoValidSpanExporter
}
Expand Down Expand Up @@ -335,3 +328,97 @@ 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())
}

Check warning on line 343 in service/internal/proctelemetry/config.go

View check run for this annotation

Codecov / codecov/patch

service/internal/proctelemetry/config.go#L332-L343 

Added lines #L332 - L343 were not covered by tests
}

if otlpConfig.Compression != nil {
opts = append(opts, otlptracegrpc.WithCompressor(*otlpConfig.Compression))
}
if otlpConfig.Timeout != nil {
opts = append(opts, otlptracegrpc.WithTimeout(time.Millisecond*time.Duration(*otlpConfig.Timeout)))
}
if len(otlpConfig.Headers) > 0 {
opts = append(opts, otlptracegrpc.WithHeaders(otlpConfig.Headers))
}

Check warning on line 354 in service/internal/proctelemetry/config.go

View check run for this annotation

Codecov / codecov/patch

service/internal/proctelemetry/config.go#L346-L354 

Added lines #L346 - L354 were not covered by tests

return otlptracegrpc.New(ctx, opts...)

Check warning on line 356 in service/internal/proctelemetry/config.go

View check run for this annotation

Codecov / codecov/patch

service/internal/proctelemetry/config.go#L356 

Added line #L356 was not covered by tests
}

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))
}

Check warning on line 374 in service/internal/proctelemetry/config.go

View check run for this annotation

Codecov / codecov/patch

service/internal/proctelemetry/config.go#L359-L374 

Added lines #L359 - L374 were not covered by tests
}
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)

Check warning on line 383 in service/internal/proctelemetry/config.go

View check run for this annotation

Codecov / codecov/patch

service/internal/proctelemetry/config.go#L376-L383 

Added lines #L376 - L383 were not covered by tests
}
}
if otlpConfig.Timeout != nil {
opts = append(opts, otlptracehttp.WithTimeout(time.Millisecond*time.Duration(*otlpConfig.Timeout)))
}
if len(otlpConfig.Headers) > 0 {
opts = append(opts, otlptracehttp.WithHeaders(otlpConfig.Headers))
}

Check warning on line 391 in service/internal/proctelemetry/config.go

View check run for this annotation

Codecov / codecov/patch

service/internal/proctelemetry/config.go#L386-L391 

Added lines #L386 - L391 were not covered by tests

return otlptracehttp.New(ctx, opts...)

Check warning on line 393 in service/internal/proctelemetry/config.go

View check run for this annotation

Codecov / codecov/patch

service/internal/proctelemetry/config.go#L393 

Added line #L393 was not covered by tests
}

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

}
2 changes: 1 addition & 1 deletion service/telemetry/config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ func (sp *SpanProcessor) Unmarshal(conf *confmap.Conf) error {
}

if sp.Batch != nil {
if sp.Batch.Exporter.Console == nil {
if sp.Batch.Exporter.Console == nil && sp.Batch.Exporter.Otlp == nil {
return fmt.Errorf("invalid exporter configuration")
}
return nil
Expand Down

0 comments on commit 411ab40

Please sign in to comment.