feat: add OTEL tracing span middleware during function proxy

Signed-off-by: Lucas Roesler <[email protected]>

Makefile

@@ -1,10 +1,15 @@
 TAG?=latest
 NS?=openfaas
 
+COMMIT ?= $(shell git rev-parse HEAD)
+
 .PHONY: build-gateway
 build-gateway:
-	(cd gateway;  docker buildx build --platform linux/amd64 -t ${NS}/gateway:latest-dev .)
+	(cd gateway;  docker buildx build --platform linux/amd64 --load -t  ${NS}/gateway:${COMMIT} -t ${NS}/gateway:latest-dev .)
+
 
+kind-load:
+	kind --name of-tracing load docker-image ${NS}/gateway:${COMMIT}
 # .PHONY: test-ci
 # test-ci:
 # 	./contrib/ci.sh

gateway/go.mod

@@ -5,11 +5,18 @@ go 1.16
 require (
 	github.com/docker/distribution v2.7.1+incompatible
 	github.com/gorilla/mux v1.8.0
-	github.com/nats-io/nats-server/v2 v2.3.2 // indirect
-	github.com/nats-io/nats-streaming-server v0.22.0 // indirect
 	github.com/openfaas/faas-provider v0.18.6
 	github.com/openfaas/nats-queue-worker v0.0.0-20210726161954-ada9a31504c9
 	github.com/prometheus/client_golang v1.9.0
 	github.com/prometheus/client_model v0.2.0
-	go.uber.org/goleak v1.1.10
+	go.opentelemetry.io/contrib/propagators/jaeger v1.3.0
+	go.opentelemetry.io/otel v1.3.0
+	go.opentelemetry.io/otel/exporters/jaeger v1.3.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.3.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.3.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.3.0
+	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.3.0
+	go.opentelemetry.io/otel/sdk v1.3.0
+	go.opentelemetry.io/otel/trace v1.3.0
+	go.uber.org/goleak v1.1.12
 )

gateway/main.go

@@ -4,6 +4,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"net/http"
@@ -14,6 +15,7 @@ import (
 	"github.com/openfaas/faas/gateway/handlers"
 	"github.com/openfaas/faas/gateway/metrics"
 	"github.com/openfaas/faas/gateway/pkg/middleware"
+	"github.com/openfaas/faas/gateway/pkg/tracing"
 	"github.com/openfaas/faas/gateway/plugin"
 	"github.com/openfaas/faas/gateway/scaling"
 	"github.com/openfaas/faas/gateway/types"
@@ -46,6 +48,12 @@ func main() {
 
 	log.Printf("Binding to external function provider: %s", config.FunctionsProviderURL)
 
+	shutdown, err := tracing.Provider(context.TODO(), "gateway", version.Version, version.GitCommitMessage)
+	if err != nil {
+		log.Fatalln(err)
+	}
+	defer shutdown(context.TODO())
+
 	// credentials is used for service-to-service auth
 	var credentials *auth.BasicAuthCredentials
 
@@ -154,6 +162,7 @@ func main() {
 		scaler := scaling.NewFunctionScaler(scalingConfig, scalingFunctionCache)
 		functionProxy = handlers.MakeScalingHandler(faasHandlers.Proxy, scaler, scalingConfig, config.Namespace)
 	}
+	functionProxy = tracing.Middleware(tracing.ConstantName("FunctionProxy"), functionProxy)
 
 	if config.UseNATS() {
 		log.Println("Async enabled: Using NATS Streaming.")

gateway/pkg/tracing/middleware.go

@@ -0,0 +1,62 @@
+package tracing
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"strings"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/propagation"
+	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// Middleware returns a http.HandlerFunc that initializes and replaces the OpenTelemetry span for each request.
+func Middleware(nameFormatter func(r *http.Request) string, next http.HandlerFunc) http.HandlerFunc {
+	_, ok := os.LookupEnv("OTEL_EXPORTER")
+	if !ok {
+		return next
+	}
+	log.Println("configuring proxy tracing middleware")
+
+	propagator := otel.GetTextMapPropagator()
+
+	return func(w http.ResponseWriter, r *http.Request) {
+		// get the parent span from the request headers
+		ctx := propagator.Extract(r.Context(), propagation.HeaderCarrier(r.Header))
+		opts := []trace.SpanStartOption{
+			trace.WithAttributes(semconv.NetAttributesFromHTTPRequest("tcp", r)...),
+			trace.WithAttributes(semconv.HTTPServerAttributesFromHTTPRequest("gateway", "", r)...),
+			trace.WithSpanKind(trace.SpanKindServer),
+		}
+
+		ctx, span := otel.Tracer("Gateway").Start(ctx, nameFormatter(r), opts...)
+		defer span.End()
+
+		debug(span, "tracing request %q", r.URL.String())
+
+		r = r.WithContext(ctx)
+		// set the new span as the parent span in the outgoing request context
+		// note that this will overwrite the uber-trace-id and traceparent headers
+		propagator.Inject(ctx, propagation.HeaderCarrier(r.Header))
+		next(w, r)
+	}
+}
+
+// ConstantName geneates the given name for the span based on the request.
+func ConstantName(value string) func(*http.Request) string {
+	return func(r *http.Request) string {
+		return value
+	}
+}
+
+func debug(span trace.Span, format string, args ...interface{}) {
+	value := os.Getenv("OTEL_LOG_LEVEL")
+	if strings.ToLower(value) != "debug" {
+		return
+	}
+
+	log.Printf("%s, trace_id=%s", fmt.Sprintf(format, args...), span.SpanContext().TraceID())
+}

gateway/pkg/tracing/otel.go

@@ -0,0 +1,155 @@
+package tracing
+
+import (
+	"context"
+	"log"
+	"os"
+	"time"
+
+	jaegerprop "go.opentelemetry.io/contrib/propagators/jaeger"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/exporters/jaeger"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
+	"go.opentelemetry.io/otel/exporters/stdout/stdouttrace"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/sdk/resource"
+	tracesdk "go.opentelemetry.io/otel/sdk/trace"
+	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+)
+
+type Exporter string
+
+const (
+	JaegerExporter   Exporter = "jaeger"
+	LogExporter      Exporter = "log"
+	OTELExporter     Exporter = "otel"
+	DisabledExporter Exporter = "disabled"
+)
+
+type Shutdown func(context.Context)
+
+// Provider returns an OpenTelemetry TracerProvider configured to use
+// the Jaeger exporter that will send spans to the provided url. The returned
+// TracerProvider will also use a Resource configured with all the information
+// about the application.
+func Provider(ctx context.Context, name, version, commit string) (shutdown Shutdown, err error) {
+	exporter := Exporter(os.Getenv("OTEL_EXPORTER"))
+
+	var exp tracesdk.TracerProviderOption
+	switch exporter {
+	case JaegerExporter:
+		// configure the collector from the env variables,
+		// OTEL_EXPORTER_JAEGER_ENDPOINT/USER/PASSWORD
+		// see: https://github.com/open-telemetry/opentelemetry-go/tree/main/exporters/jaeger
+		j, e := jaeger.New(jaeger.WithCollectorEndpoint())
+		exp, err = tracesdk.WithBatcher(j), e
+	case LogExporter:
+		w := os.Stdout
+		opts := []stdouttrace.Option{stdouttrace.WithWriter(w)}
+		if truthyEnv("OTEL_EXPORTER_LOG_PRETTY_PRINT") {
+			opts = append(opts, stdouttrace.WithPrettyPrint())
+		}
+		if !truthyEnv("OTEL_EXPORTER_LOG_TIMESTAMPS") {
+			opts = append(opts, stdouttrace.WithoutTimestamps())
+		}
+
+		s, e := stdouttrace.New(opts...)
+		exp, err = tracesdk.WithSyncer(s), e
+	case OTELExporter:
+		// find available env variables for configuration
+		// see: https://github.com/open-telemetry/opentelemetry-go/tree/main/exporters/otlp/otlptrace#environment-variables
+		kind := get("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")
+
+		var client otlptrace.Client
+		switch kind {
+		case "grpc":
+			client = otlptracegrpc.NewClient()
+		case "http":
+			client = otlptracehttp.NewClient()
+		}
+		o, e := otlptrace.New(ctx, client)
+		exp, err = tracesdk.WithBatcher(o), e
+	default:
+		log.Println("tracing disabled")
+		// We explicitly DO NOT set the global TracerProvider using otel.SetTracerProvider().
+		// The unset TracerProvider returns a "non-recording" span, but still passes through context.
+		// return no-op shutdown function
+		return func(_ context.Context) {}, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	otel.SetTextMapPropagator(
+		propagation.NewCompositeTextMapPropagator(
+			propagation.TraceContext{},
+			jaegerprop.Jaeger{},
+			propagation.Baggage{},
+		),
+	)
+
+	resource, err := resource.New(
+		context.Background(),
+		resource.WithFromEnv(),
+		resource.WithHost(),
+		resource.WithOS(),
+		resource.WithTelemetrySDK(),
+		resource.WithAttributes(
+			semconv.ServiceVersionKey.String(version),
+			attribute.String("service.commit", commit),
+			semconv.ServiceNameKey.String("gateway"),
+		),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	provider := tracesdk.NewTracerProvider(
+		// Always be sure to batch in production.
+		exp,
+		tracesdk.WithResource(resource),
+		tracesdk.WithSampler(tracesdk.AlwaysSample()),
+	)
+
+	// Register our TracerProvider as the global so any imported
+	// instrumentation in the future will default to using it.
+	otel.SetTracerProvider(provider)
+
+	shutdown = func(ctx context.Context) {
+		// Do not let the application hang forever when it is shutdown.
+		ctx, cancel := context.WithTimeout(ctx, time.Second*5)
+		defer cancel()
+
+		err := provider.Shutdown(ctx)
+		if err != nil {
+			log.Printf("failed to shutdown tracing provider: %v", err)
+		}
+	}
+
+	return shutdown, nil
+}
+
+func truthyEnv(name string) bool {
+	value, ok := os.LookupEnv(name)
+	if !ok {
+		return false
+	}
+
+	switch value {
+	case "true", "1", "yes", "on":
+		return true
+	default:
+		return false
+	}
+}
+
+func get(name, defaultValue string) string {
+	value, ok := os.LookupEnv(name)
+	if !ok {
+		return defaultValue
+	}
+	return value
+}

gateway/pkg/tracing/span.go

@@ -0,0 +1,18 @@
+package tracing
+
+import (
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+)
+
+func FinishSpan(span trace.Span, err error) {
+	if span == nil {
+		return
+	}
+
+	if err != nil {
+		span.SetStatus(codes.Error, err.Error())
+		span.RecordError(err)
+	}
+	span.End()
+}