Add support for OTEL collector endpoint in Promscale tracing telemetr…

…y exporter

In addition to Jaeger, now Promscale can send its telemetry data to OTEL
collector endpoint. Since this is an endpoint that Promscale itself exposes,
we can actually ingest tracing data from Promscale into Promscale itself.

CHANGELOG.md

@@ -17,6 +17,7 @@ We use the following categories for changes:
 ### Added
 - Add Prometheus metrics support for Tracing [#1102]
 - Add ingested spans-count to telemetry [#1155]
+- Add OTEL collector exporter endpoint support to Promscale tracing telemetry exporter [#1148]
 
 ### Fixed
 - Fix spans with end < start. Start and end are swapped in this case. [#1096]

docs/cli.md

@@ -54,8 +54,11 @@ You can also find information on flags with `promscale_<version> -help`.
 | telemetry.log.format | string | logfmt | Log format to use from [ "logfmt", "json" ]. |
 | telemetry.log.level | string | debug | Log level to use from [ "error", "warn", "info", "debug" ]. |
 | telemetry.log.throughput-report-interval | duration | 0 second | Duration interval at which throughput should be reported. Setting duration to `0` will disable reporting throughput, otherwise, an interval with unit must be provided, e.g. `10s` or `3m`. |
+| telemetry.trace.otel-endpoint | string | "" (empty) | OpenTelemetry tracing collector GRPC URL endpoint to send telemetry to. (i.e. otel-collector:4317) |
+| telemetry.trace.otel-tls-cert-file | string | "" (empty) | TLS Certificate file used for client authentication against the OTEL tracing collector GRPC endpoint, leave blank to disable TLS. |
+| telemetry.trace.otel-tls-key-file | string | "" (empty) | TLS Key file for client authentication against the OTEL tracing collector GRPC endpoint, leave blank to disable TLS. |
 | telemetry.trace.jaeger-endpoint | string | "" (empty) | Jaeger tracing collector thrift HTTP URL endpoint to send telemetry to (e.g. https://jaeger-collector:14268/api/traces). |
-| telemetry.trace.sample-ratio | float | 0.0 | Trace sampling ratio, amount of spans to send to collector. Valid values from 0.0 (none) to 1.0 (all). |
+| telemetry.trace.sample-ratio | float | 1.0 | Trace sampling ratio, amount of spans to send to collector. Valid values from 0.0 (none) to 1.0 (all). |
 
 ## Metrics specific flags flags
 

go.mod

@@ -45,12 +45,14 @@ require (
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.28.0
 	go.opentelemetry.io/otel v1.3.0
 	go.opentelemetry.io/otel/exporters/jaeger v1.3.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.3.0
 	go.opentelemetry.io/otel/sdk v1.3.0
 	go.opentelemetry.io/otel/trace v1.3.0
 	go.uber.org/automaxprocs v1.4.0
 	go.uber.org/goleak v1.1.12
 	golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27
 	google.golang.org/grpc v1.44.0
+	google.golang.org/protobuf v1.27.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0
 )
 

go.sum

@@ -308,6 +308,7 @@ github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QH
 github.com/cenkalti/backoff/v4 v4.0.2/go.mod h1:eEew/i+1Q6OrCDZh3WiXYv3+nJwBASZ8Bog/87DQnVg=
 github.com/cenkalti/backoff/v4 v4.1.0/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/cenkalti/backoff/v4 v4.1.1/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
+github.com/cenkalti/backoff/v4 v4.1.2 h1:6Yo7N8UP2K6LWZnW94DLVSSrbobcWdVzAYOisuDPIFo=
 github.com/cenkalti/backoff/v4 v4.1.2/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/census-instrumentation/opencensus-proto v0.3.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -2051,6 +2052,12 @@ go.opentelemetry.io/otel v1.3.0 h1:APxLf0eiBwLl+SOXiJJCVYzA1OOJNyAoV8C5RNRyy7Y=
 go.opentelemetry.io/otel v1.3.0/go.mod h1:PWIKzi6JCp7sM0k9yZ43VX+T345uNbAkDKwHVjb2PTs=
 go.opentelemetry.io/otel/exporters/jaeger v1.3.0 h1:HfydzioALdtcB26H5WHc4K47iTETJCdloL7VN579/L0=
 go.opentelemetry.io/otel/exporters/jaeger v1.3.0/go.mod h1:KoYHi1BtkUPncGSRtCe/eh1ijsnePhSkxwzz07vU0Fc=
+go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.3.0 h1:R/OBkMoGgfy2fLhs2QhkCI1w4HLEQX92GCcJB6SSdNk=
+go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.3.0/go.mod h1:VpP4/RMn8bv8gNo9uK7/IMY4mtWLELsS+JIP0inH0h4=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.3.0 h1:giGm8w67Ja7amYNfYMdme7xSp2pIxThWopw8+QP51Yk=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.3.0/go.mod h1:hO1KLR7jcKaDDKDkvI9dP/FIhpmna5lkqPUQdEjFAM8=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.3.0 h1:VQbUHoJqytHHSJ1OZodPH9tvZZSVzUHjPHpkO85sT6k=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.3.0/go.mod h1:keUU7UfnwWTWpJ+FWnyqmogPa82nuU5VUANFq49hlMY=
 go.opentelemetry.io/otel/exporters/prometheus v0.26.0/go.mod h1:0/uJZI7H2y0FgMVCgCWdPzZpxPx3X3F5uInY32I9foI=
 go.opentelemetry.io/otel/internal/metric v0.26.0 h1:dlrvawyd/A+X8Jp0EBT4wWEe4k5avYaXsXrBr4dbfnY=
 go.opentelemetry.io/otel/internal/metric v0.26.0/go.mod h1:CbBP6AxKynRs3QCbhklyLUtpfzbqCLiafV9oY2Zj1Jk=
@@ -2063,6 +2070,8 @@ go.opentelemetry.io/otel/sdk/metric v0.26.0/go.mod h1:2VIeK0kS1YvRLFg3J58ptZTXYp
 go.opentelemetry.io/otel/trace v1.3.0 h1:doy8Hzb1RJ+I3yFhtDmwNc7tIyw1tNMOIsyPzp1NOGY=
 go.opentelemetry.io/otel/trace v1.3.0/go.mod h1:c/VDhno8888bvQYmbYLqe41/Ldmr/KKunbvWM4/fEjk=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
+go.opentelemetry.io/proto/otlp v0.11.0 h1:cLDgIBTf4lLOlztkhzAEdQsJ4Lj+i5Wc9k6Nn0K1VyU=
+go.opentelemetry.io/proto/otlp v0.11.0/go.mod h1:QpEjXPrNQzrFDZgoTo49dgHR9RYRSrg3NAKnUGl9YpQ=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=

pkg/runner/runner.go

@@ -70,7 +70,7 @@ func Run(cfg *Config) error {
 		tput.InitWatcher(cfg.ThroughputInterval)
 	}
 
-	if cfg.TracerCfg.JaegerCollectorEndpoint != "" {
+	if cfg.TracerCfg.CanExportTraces() {
 		tp, err := tracer.InitProvider(&cfg.TracerCfg)
 		if err != nil {
 			log.Error("msg", "aborting startup due to tracer provider error", "err", err.Error())

pkg/tracer/codec.go

@@ -0,0 +1,31 @@
+package tracer
+
+import (
+	"google.golang.org/grpc/encoding"
+	"google.golang.org/protobuf/proto"
+)
+
+// clientCodec is used to force the gRPC client into the
+// specific codec that works with OTEL exporter.
+type clientCodec struct{}
+
+var _ encoding.Codec = (*clientCodec)(nil)
+
+func newClientCodec() *clientCodec {
+	return &clientCodec{}
+}
+
+// Name implements encoding.Codec
+func (c *clientCodec) Name() string {
+	return "proto"
+}
+
+// Marshal implements encoding.Codec
+func (c *clientCodec) Marshal(v interface{}) ([]byte, error) {
+	return proto.Marshal(v.(proto.Message))
+}
+
+// Unmarshal implements encoding.Codec
+func (c *clientCodec) Unmarshal(data []byte, v interface{}) error {
+	return proto.Unmarshal(data, v.(proto.Message))
+}

pkg/tracer/trace.go

@@ -1,13 +1,23 @@
 package tracer
 
 import (
+	"context"
 	"flag"
+	"fmt"
+	"time"
 
 	"go.opentelemetry.io/otel"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+
+	//"go.opentelemetry.io/otel/exporters/jaeger"
 	"go.opentelemetry.io/otel/exporters/jaeger"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
 	"go.opentelemetry.io/otel/sdk/resource"
 	tracesdk "go.opentelemetry.io/otel/sdk/trace"
 	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -17,14 +27,27 @@ const (
 
 // Config represents a tracing configuration used upon initialization.
 type Config struct {
+	OtelCollectorEndpoint    string
+	OtelCollectorTLSCertFile string
+	OtelCollectorTLSKeyFile  string
+
 	JaegerCollectorEndpoint string
 	SamplingRatio           float64
 }
 
+// CanExportTraces checks if we have the necessary configurations to export traces telemetry.
+func (c Config) CanExportTraces() bool {
+	return c.OtelCollectorEndpoint != "" ||
+		c.JaegerCollectorEndpoint != ""
+}
+
 // ParseFlags parses the configuration flags for tracing.
 func ParseFlags(fs *flag.FlagSet, cfg *Config) *Config {
+	fs.StringVar(&cfg.OtelCollectorEndpoint, "telemetry.trace.otel-endpoint", "", "OpenTelemetry tracing collector GRPC URL endpoint to send telemetry to. (i.e. otel-collector:4317)")
+	fs.StringVar(&cfg.OtelCollectorTLSCertFile, "telemetry.trace.otel-tls-cert-file", "", "TLS Certificate file used for client authentication against the OTEL tracing collectore GRPC endpoint, leave blank to disable TLS.")
+	fs.StringVar(&cfg.OtelCollectorTLSKeyFile, "telemetry.trace.otel-tls-key-file", "", "TLS Key file for client authentication against the OTEL tracing collectore GRPC endpoint, leave blank to disable TLS.")
 	fs.StringVar(&cfg.JaegerCollectorEndpoint, "telemetry.trace.jaeger-endpoint", "", "Jaeger tracing collector thrift HTTP URL endpoint to send telemetry to (e.g. https://jaeger-collector:14268/api/traces).")
-	fs.Float64Var(&cfg.SamplingRatio, "telemetry.trace.sampling-ratio", 0, "Trace sampling ratio, amount of spans to send to collector. Valid values from 0.0 (none) to 1.0 (all).")
+	fs.Float64Var(&cfg.SamplingRatio, "telemetry.trace.sampling-ratio", 1, "Trace sampling ratio, amount of spans to send to collector. Valid values from 0.0 (none) to 1.0 (all).")
 	return cfg
 }
 
@@ -33,20 +56,70 @@ func Default() trace.Tracer {
 }
 
 func InitProvider(cfg *Config) (*tracesdk.TracerProvider, error) {
-	// Create the Jaeger exporter
-	exp, err := jaeger.New(jaeger.WithCollectorEndpoint(jaeger.WithEndpoint(cfg.JaegerCollectorEndpoint)))
-	if err != nil {
-		return nil, err
+	var (
+		err  error
+		opts = make([]tracesdk.TracerProviderOption, 0)
+	)
+
+	if cfg.JaegerCollectorEndpoint != "" {
+		// Create the Jaeger exporter.
+		exp, err := jaeger.New(jaeger.WithCollectorEndpoint(jaeger.WithEndpoint(cfg.JaegerCollectorEndpoint)))
+		if err != nil {
+			return nil, err
+		}
+
+		// Always be sure to batch in production.
+		opts = append(opts, tracesdk.WithBatcher(exp))
 	}
-	tp := tracesdk.NewTracerProvider(
+
+	if cfg.OtelCollectorEndpoint != "" {
+		creds := insecure.NewCredentials()
+		if cfg.OtelCollectorTLSCertFile != "" {
+			creds, err = credentials.NewClientTLSFromFile(cfg.OtelCollectorTLSCertFile, cfg.OtelCollectorTLSKeyFile)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		// Create GRPC connection to collector.
+		conn, err := grpc.DialContext(
+			ctx,
+			cfg.OtelCollectorEndpoint,
+			grpc.WithTransportCredentials(creds),
+			grpc.WithDefaultCallOptions(grpc.ForceCodec(newClientCodec())),
+			grpc.WithBlock(),
+		)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create gRPC connection to collector: %w", err)
+		}
+
+		// Create the OTEL exporter.
+		exp, err := otlptracegrpc.New(ctx, otlptracegrpc.WithGRPCConn(conn))
+		if err != nil {
+			return nil, fmt.Errorf("failed to create trace exporter: %w", err)
+		}
+
 		// Always be sure to batch in production.
-		tracesdk.WithBatcher(exp),
+		opts = append(opts, tracesdk.WithBatcher(exp))
+	}
+
+	if len(opts) == 0 {
+		return nil, fmt.Errorf("tracing telemetry requires setting either the open telemetry collector endpoint or the jaeger endpoint")
+	}
+
+	opts = append(
+		opts,
 		tracesdk.WithSampler(tracesdk.TraceIDRatioBased(cfg.SamplingRatio)),
 		// Record information about this application in an Resource.
 		tracesdk.WithResource(resource.NewWithAttributes(
 			semconv.SchemaURL,
 			semconv.ServiceNameKey.String("promscale-connector-service"),
 		)),
 	)
+
+	tp := tracesdk.NewTracerProvider(opts...)
 	return tp, nil
 }