Add OTLP for traces

Signed-off-by: Pavol Loffay <[email protected]>

Author: pavolloffay

api/traces/v1/http.go

@@ -109,7 +109,7 @@ func (n nopInstrumentHandler) NewHandler(labels prometheus.Labels, handler http.
 // The web UI handler is able to rewrite
 // HTML to change the <base> attribute so that it works with the Observatorium-style
 // "/api/v1/traces/{tenant}/" URLs.
-func NewV2Handler(read *url.URL, readTemplate string, tempo *url.URL, upstreamCA []byte, upstreamCert *stdtls.Certificate, opts ...HandlerOption) http.Handler {
+func NewV2Handler(read *url.URL, readTemplate string, tempo *url.URL, writeOTLPHttp *url.URL, upstreamCA []byte, upstreamCert *stdtls.Certificate, opts ...HandlerOption) http.Handler {
 
 	if read == nil && readTemplate == "" && tempo == nil {
 		panic("missing Jaeger read url")
@@ -191,6 +191,39 @@ func NewV2Handler(read *url.URL, readTemplate string, tempo *url.URL, upstreamCA
 		})
 	}
 
+	if writeOTLPHttp != nil {
+		middlewares := proxy.Middlewares(
+			proxy.MiddlewareSetUpstream(writeOTLPHttp),
+			proxy.MiddlewareSetPrefixHeader(),
+			proxy.MiddlewareLogger(c.logger),
+			middlewareMetrics,
+		)
+
+		t := &http.Transport{
+			DialContext: (&net.Dialer{
+				Timeout: dialTimeout,
+			}).DialContext,
+			TLSClientConfig: tls.NewClientConfig(upstreamCA, upstreamCert),
+		}
+
+		proxyOTLP := &httputil.ReverseProxy{
+			Director:  middlewares,
+			ErrorLog:  proxy.Logger(c.logger),
+			Transport: otelhttp.NewTransport(t),
+
+			// This is a key piece, it changes <base href=> tags on text/html content
+			ModifyResponse: jaegerUIResponseModifier,
+		}
+
+		r.Group(func(r chi.Router) {
+			r.Use(c.tempoMiddlewares...)
+			r.Post("/v1/traces", c.instrument.NewHandler(
+				prometheus.Labels{"group": "tracesotlphttpv1api", "handler": "traces"},
+				proxyOTLP))
+		})
+
+	}
+
 	// if tempo upstream is enabled, configure proxy and route
 	if tempo != nil {
 		level.Debug(c.logger).Log("msg", "Configuring upstream Tempo", "queryv2", tempo)

main.go

@@ -178,14 +178,15 @@ type tracesConfig struct {
 	// readTemplateEndpoint is of the form "http://jaeger-{tenant}-query:16686".
 	readTemplateEndpoint string
 
-	readEndpoint         *url.URL
-	writeEndpoint        string
-	tempoEndpoint        *url.URL
-	upstreamWriteTimeout time.Duration
-	upstreamCAFile       string
-	upstreamCertFile     string
-	upstreamKeyFile      string
-	tenantHeader         string
+	readEndpoint          *url.URL
+	writeOTLPGRPCEndpoint string
+	writeOTLPHTTPEndpoint *url.URL
+	tempoEndpoint         *url.URL
+	upstreamWriteTimeout  time.Duration
+	upstreamCAFile        string
+	upstreamCertFile      string
+	upstreamKeyFile       string
+	tenantHeader          string
 	// enable traces if readTemplateEndpoint, readEndpoint, or writeEndpoint is provided.
 	enabled bool
 }
@@ -802,6 +803,7 @@ func main() {
 								cfg.traces.readEndpoint,
 								cfg.traces.readTemplateEndpoint,
 								cfg.traces.tempoEndpoint,
+								cfg.traces.writeOTLPHTTPEndpoint,
 								tracesUpstreamCACert,
 								tracesUpstreamClientCert,
 								tracesv1.Logger(logger),
@@ -1052,7 +1054,8 @@ func parseFlags() (config, error) {
 		rawLogsAuthExtractSelectors    string
 		rawTracesReadEndpoint          string
 		rawTracesTempoEndpoint         string
-		rawTracesWriteEndpoint         string
+		rawTracesWriteOTLPGRPCEndpoint string
+		rawTracesWriteOTLPHTTPEndpoint string
 		rawTracingEndpointType         string
 	)
 
@@ -1142,8 +1145,10 @@ func parseFlags() (config, error) {
 		"The endpoint against which to make HTTP read requests for traces using traceQL (tempo API).")
 	flag.StringVar(&cfg.traces.readTemplateEndpoint, "experimental.traces.read.endpoint-template", "",
 		"A template replacing --read.traces.endpoint, such as http://jaeger-{tenant}-query:16686")
-	flag.StringVar(&rawTracesWriteEndpoint, "traces.write.endpoint", "",
-		"The endpoint against which to make gRPC write requests for traces.")
+	flag.StringVar(&rawTracesWriteOTLPGRPCEndpoint, "traces.write.otlpgrpc.endpoint", "",
+		"The endpoint against which to make OTLP gRPC write requests for traces.")
+	flag.StringVar(&rawTracesWriteOTLPHTTPEndpoint, "traces.write.otlphttp.endpoint", "",
+		"The endpoint against which to make OTLP HTTP write requests for traces.")
 	flag.DurationVar(&cfg.traces.upstreamWriteTimeout, "traces.write-timeout", tracesMiddlewareTimeout,
 		"The HTTP write timeout for proxied requests to the traces endpoint.")
 	flag.StringVar(&cfg.traces.upstreamCAFile, "traces.tls.ca-file", "",
@@ -1344,19 +1349,29 @@ func parseFlags() (config, error) {
 		cfg.traces.tempoEndpoint = tracesTempoEndpoint
 	}
 
-	if rawTracesWriteEndpoint != "" {
+	if rawTracesWriteOTLPGRPCEndpoint != "" {
+		cfg.traces.enabled = true
+
+		_, _, err := net.SplitHostPort(rawTracesWriteOTLPGRPCEndpoint)
+		if err != nil {
+			return cfg, fmt.Errorf("--traces.write.otlpgrpc.endpoint %q is invalid: %w", rawTracesWriteOTLPGRPCEndpoint, err)
+		}
+
+		cfg.traces.writeOTLPGRPCEndpoint = rawTracesWriteOTLPGRPCEndpoint
+	}
+	if rawTracesWriteOTLPHTTPEndpoint != "" {
 		cfg.traces.enabled = true
 
-		_, _, err := net.SplitHostPort(rawTracesWriteEndpoint)
+		tracesOTLPHTTPEndpoint, err := url.ParseRequestURI(rawTracesWriteOTLPHTTPEndpoint)
 		if err != nil {
-			return cfg, fmt.Errorf("--traces.write.endpoint %q is invalid: %w", rawTracesWriteEndpoint, err)
+			return cfg, fmt.Errorf("--traces.write.otlphttp.endpoint %q is invalid: %w", rawTracesWriteOTLPHTTPEndpoint, err)
 		}
 
-		cfg.traces.writeEndpoint = rawTracesWriteEndpoint
+		cfg.traces.writeOTLPHTTPEndpoint = tracesOTLPHTTPEndpoint
 	}
 
 	if cfg.traces.enabled && cfg.server.grpcListen == "" {
-		return cfg, fmt.Errorf("-traces.write.endpoint is set to %q but -grpc.listen is not set", cfg.traces.writeEndpoint)
+		return cfg, fmt.Errorf("-traces.write.endpoint is set to %q but -grpc.listen is not set", cfg.traces.writeOTLPGRPCEndpoint)
 	}
 
 	if !cfg.traces.enabled && cfg.server.grpcListen != "" {
@@ -1446,7 +1461,7 @@ func newGRPCServer(cfg *config, tenantHeader string, tenantIDs map[string]string
 	authorizers map[string]rbac.Authorizer, logger log.Logger, tracesUpstreamCA []byte, tracesUpstreamCert *stdtls.Certificate,
 ) (*grpc.Server, error) {
 	connOtel, err := tracesv1.NewOTelConnection(
-		cfg.traces.writeEndpoint,
+		cfg.traces.writeOTLPGRPCEndpoint,
 		tracesv1.WithLogger(logger),
 		tracesv1.WithUpstreamTLS(tracesUpstreamCA, tracesUpstreamCert),
 	)

test/e2e/configs.go

@@ -228,15 +228,17 @@ func createRulesYAML(
 
 const otelConfigTpl = `
 receivers:
-    otlp/grpc:
+    otlp:
       protocols:
         grpc:
             endpoint: "0.0.0.0:4317"
+        http:
+            endpoint: "0.0.0.0:4318"
 
 exporters:
-    logging:
-        logLevel: debug
-    jaeger:
+    debug:
+        verbosity: detailed
+    otlp:
         endpoint: %[1]s
         tls:
           insecure: true
@@ -250,8 +252,8 @@ service:
 
     pipelines:
         traces/grpc:
-            receivers: [otlp/grpc]
-            exporters: [logging,jaeger]
+            receivers: [otlp]
+            exporters: [debug,otlp]
 `
 
 // createOtelCollectorConfigYAML() creates YAML for an Open Telemetry collector inside the Observatorium API boundary.
@@ -288,8 +290,8 @@ receivers:
             endpoint: 0.0.0.0:4317
 
 exporters:
-    logging:
-      logLevel: debug
+    debug:
+      verbosity: detailed
     otlp:
       endpoint: %[1]s
       # auth:
@@ -310,12 +312,13 @@ service:
     extensions: [health_check]
     telemetry:
       metrics:
-        address: localhost:8889
+        address: :8888
+        level: detailed
     # extensions: [oauth2client]
     pipelines:
       traces:
         receivers: [otlp]
-        exporters: [logging,otlp]
+        exporters: [debug,otlp]
 `
 
 // createOtelForwardingCollectorConfigYAML() creates YAML for an Open Telemetry collector outside the

test/e2e/interactive_test.go

@@ -23,7 +23,7 @@ func TestInteractiveSetup(t *testing.T) {
 	readEndpoint, writeEndpoint, readExtEndpoint := startServicesForMetrics(t, e)
 	logsEndpoint, logsExtEndpoint := startServicesForLogs(t, e)
 	rulesEndpoint := startServicesForRules(t, e)
-	internalOtlpEndpoint, httpExternalQueryEndpoint, httpInternalQueryEndpoint := startServicesForTraces(t, e)
+	internalOTLPGRPCEndpoint, internalOTLPHTTPEndpoint, httpExternalQueryEndpoint, httpInternalQueryEndpoint := startServicesForTraces(t, e)
 
 	api, err := newObservatoriumAPIService(
 		e,
@@ -32,7 +32,8 @@ func TestInteractiveSetup(t *testing.T) {
 		withRulesEndpoint("http://"+rulesEndpoint),
 		withRateLimiter(rateLimiterAddr),
 		withGRPCListenEndpoint(":8317"),
-		withOtelTraceEndpoint(internalOtlpEndpoint),
+		withOTLPGRPCTraceEndpoint(internalOTLPGRPCEndpoint),
+		withOTLPHTTPTraceEndpoint(internalOTLPHTTPEndpoint),
 		withJaegerEndpoint("http://"+httpInternalQueryEndpoint),
 	)
 	testutil.Ok(t, err)

test/e2e/services.go

@@ -26,15 +26,10 @@ const (
 	upImage           = "quay.io/observatorium/up:master-2022-10-27-d8bb06f"
 	alertmanagerImage = "quay.io/prometheus/alertmanager:v0.25.0"
 
-	jaegerAllInOneImage = "jaegertracing/all-in-one:1.31"
-	otelCollectorImage  = "otel/opentelemetry-collector:0.45.0"
+	jaegerAllInOneImage = "jaegertracing/all-in-one:1.57.0"
+	otelCollectorImage  = "ghcr.io/open-telemetry/opentelemetry-collector-releases/opentelemetry-collector-contrib:0.101.0"
 	tempoImage          = "grafana/tempo:2.2.4"
 
-	// Note that if the forwarding collector uses OIDC flow instead of hard-coding
-	// the bearer token we would need
-	// "otel/opentelemetry-collector-contrib:0.45.0" instead.
-	otelFwdCollectorImage = "otel/opentelemetry-collector:0.45.0"
-
 	dexImage              = "dexidp/dex:v2.30.0"
 	opaImage              = "openpolicyagent/opa:0.47.4-static"
 	gubernatorImage       = "ghcr.io/mailgun/gubernator:v2.0.0-rc.36"
@@ -96,16 +91,16 @@ func startServicesForLogs(t *testing.T, e e2e.Environment) (
 	return loki.InternalEndpoint("http"), loki.Endpoint("http")
 }
 
-func startServicesForTraces(t *testing.T, e e2e.Environment) (otlpGRPCEndpoint, jaegerExternalHttpEndpoint, jaegerInternalHttpEndpoint string) {
+func startServicesForTraces(t *testing.T, e e2e.Environment) (otlpGRPCEndpoint, otlpHTTPEndpoint, jaegerExternalHttpEndpoint, jaegerInternalHttpEndpoint string) {
 	prometheus := e2edb.NewPrometheus(e, "prometheus")
 	testutil.Ok(t, e2e.StartAndWaitReady(prometheus))
 
 	jaeger := e.Runnable("jaeger").
 		WithPorts(
 			map[string]int{
-				"jaeger.grpc": 14250, // Receives traces
-				"grpc.query":  16685, // Query
-				"http.query":  16686, // Query
+				"otlp.grpc":  4317,  // Receiving traces
+				"grpc.query": 16685, // Query
+				"http.query": 16686, // Query
 			}).
 		Init(e2e.StartOptions{
 			Image:   jaegerAllInOneImage,
@@ -115,7 +110,7 @@ func startServicesForTraces(t *testing.T, e e2e.Environment) (otlpGRPCEndpoint,
 			},
 		})
 
-	createOtelCollectorConfigYAML(t, e, jaeger.InternalEndpoint("jaeger.grpc"))
+	createOtelCollectorConfigYAML(t, e, jaeger.InternalEndpoint("otlp.grpc"))
 
 	otel := e.Runnable("otel-collector").
 		WithPorts(
@@ -139,7 +134,7 @@ func startServicesForTraces(t *testing.T, e e2e.Environment) (otlpGRPCEndpoint,
 	testutil.Ok(t, e2e.StartAndWaitReady(jaeger))
 	testutil.Ok(t, e2e.StartAndWaitReady(otel))
 
-	return otel.InternalEndpoint("grpc"), jaeger.Endpoint("http.query"), jaeger.InternalEndpoint("http.query")
+	return otel.InternalEndpoint("grpc"), otel.InternalEndpoint("http"), jaeger.Endpoint("http.query"), jaeger.InternalEndpoint("http.query")
 }
 
 func startTempoServicesForTraces(t *testing.T, e e2e.Environment) (tempoDistributorEndpoint, internalTempoQueryEndpoint, tempoQueryEndpoint string) {
@@ -348,16 +343,17 @@ func newOPAService(e e2e.Environment) *e2emon.InstrumentedRunnable {
 }
 
 type apiOptions struct {
-	logsEndpoint         string
-	metricsReadEndpoint  string
-	metricsWriteEndpoint string
-	metricsRulesEndpoint string
-	alertmanagerEndpoint string
-	ratelimiterAddr      string
-	tracesWriteEndpoint  string
-	gRPCListenEndpoint   string
-	jaegerQueryEndpoint  string
-	tempoEndpoint        string
+	logsEndpoint                string
+	metricsReadEndpoint         string
+	metricsWriteEndpoint        string
+	metricsRulesEndpoint        string
+	alertmanagerEndpoint        string
+	ratelimiterAddr             string
+	tracesWriteOTLPGRPCEndpoint string
+	tracesWriteOTLPHTTPEndpoint string
+	gRPCListenEndpoint          string
+	jaegerQueryEndpoint         string
+	tempoEndpoint               string
 
 	// "experimental.traces.read.endpoint-template" value.
 	tracesExperimentalTemplateReadEndpoint string
@@ -378,9 +374,15 @@ func withMetricsEndpoints(readEndpoint string, writeEndpoint string) apiOption {
 	}
 }
 
-func withOtelTraceEndpoint(exportEndpoint string) apiOption {
+func withOTLPGRPCTraceEndpoint(exportEndpoint string) apiOption {
+	return func(o *apiOptions) {
+		o.tracesWriteOTLPGRPCEndpoint = exportEndpoint
+	}
+}
+
+func withOTLPHTTPTraceEndpoint(exportEndpoint string) apiOption {
 	return func(o *apiOptions) {
-		o.tracesWriteEndpoint = exportEndpoint
+		o.tracesWriteOTLPHTTPEndpoint = exportEndpoint
 	}
 }
 
@@ -476,8 +478,11 @@ func newObservatoriumAPIService(
 		args = append(args, "--middleware.rate-limiter.grpc-address="+opts.ratelimiterAddr)
 	}
 
-	if opts.tracesWriteEndpoint != "" {
-		args = append(args, "--traces.write.endpoint="+opts.tracesWriteEndpoint)
+	if opts.tracesWriteOTLPGRPCEndpoint != "" {
+		args = append(args, "--traces.write.otlpgrpc.endpoint="+opts.tracesWriteOTLPGRPCEndpoint)
+	}
+	if opts.tracesWriteOTLPHTTPEndpoint != "" {
+		args = append(args, "--traces.write.otlphttp.endpoint="+opts.tracesWriteOTLPHTTPEndpoint)
 	}
 
 	if opts.tracesExperimentalTemplateReadEndpoint != "" {