Executor trace b3 propagate

executor/api/tracing/trace.go

@@ -0,0 +1,52 @@
+package tracing
+
+import (
+	"io"
+	"os"
+
+	"github.com/opentracing/opentracing-go"
+	jaegercfg "github.com/uber/jaeger-client-go/config"
+	"github.com/uber/jaeger-client-go/zipkin"
+)
+
+func InitTracing() (io.Closer, error) {
+	//Initialise tracing
+	cfg, err := jaegercfg.FromEnv()
+	if err != nil {
+		// parsing errors might happen here, such as when we get a string where we expect a number
+		return nil, err
+	}
+
+	if cfg.ServiceName == "" {
+		cfg.ServiceName = "executor"
+	}
+
+	propagation := os.Getenv("JAEGER_TRACE_PROPAGATION_TYPE")
+	closer, err := newTraceByPropagation(propagation, cfg)
+	if err != nil {
+		return nil, err
+	}
+	return closer, nil
+}
+
+func newTraceByPropagation(propagation string, cfg *jaegercfg.Configuration) (closer io.Closer, err error) {
+	switch propagation {
+	case "b3":
+		return newB3Tracer(cfg)
+	}
+	return newDefaultTracer(cfg)
+}
+
+func newB3Tracer(cfg *jaegercfg.Configuration) (io.Closer, error) {
+	zipkinPropagator := zipkin.NewZipkinB3HTTPHeaderPropagator()
+	closer, err := cfg.InitGlobalTracer(cfg.ServiceName,
+		jaegercfg.Injector(opentracing.HTTPHeaders, zipkinPropagator),
+		jaegercfg.Extractor(opentracing.HTTPHeaders, zipkinPropagator),
+		jaegercfg.ZipkinSharedRPCSpan(true))
+
+	return closer, err
+}
+func newDefaultTracer(cfg *jaegercfg.Configuration) (io.Closer, error) {
+	closer, err := cfg.InitGlobalTracer(cfg.ServiceName)
+	return closer, err
+}

executor/api/tracing/trace_test.go

@@ -0,0 +1,44 @@
+package tracing
+
+import (
+	"github.com/opentracing/opentracing-go"
+	"github.com/uber/jaeger-client-go"
+	"os"
+	"testing"
+)
+
+var (
+	nonRootNonSampledHeader = opentracing.TextMapCarrier{
+		"x-b3-traceid":      "1",
+		"x-b3-spanid":       "2",
+		"x-b3-parentspanid": "1",
+		"x-b3-sampled":      "0",
+	}
+)
+
+func newSpanContext(traceID, spanID, parentID uint64, sampled bool, baggage map[string]string) jaeger.SpanContext {
+	return jaeger.NewSpanContext(
+		jaeger.TraceID{Low: traceID},
+		jaeger.SpanID(spanID),
+		jaeger.SpanID(parentID),
+		sampled,
+		baggage,
+	)
+}
+
+func TestInitTracing(t *testing.T) {
+	os.Setenv("JAEGER_TRACE_PROPAGATION_TYPE", "b3")
+	closer, err := InitTracing()
+	if err != nil {
+		t.Error("init trace err", err.Error())
+	}
+	defer closer.Close()
+	if !opentracing.IsGlobalTracerRegistered() {
+		t.Error("trace is not registered")
+	}
+	tracer := opentracing.GlobalTracer()
+	_, err = tracer.Extract(opentracing.HTTPHeaders, nonRootNonSampledHeader)
+	if err != nil {
+		t.Error("trace extract error", err.Error())
+	}
+}

executor/main.go

@@ -8,7 +8,6 @@ import (
 	"fmt"
 	"github.com/ghodss/yaml"
 	"github.com/go-logr/logr"
-	"github.com/opentracing/opentracing-go"
 	"github.com/prometheus/common/log"
 	"github.com/seldonio/seldon-core/executor/api"
 	seldonclient "github.com/seldonio/seldon-core/executor/api/client"
@@ -17,11 +16,10 @@ import (
 	"github.com/seldonio/seldon-core/executor/api/grpc/seldon/proto"
 	"github.com/seldonio/seldon-core/executor/api/grpc/tensorflow"
 	"github.com/seldonio/seldon-core/executor/api/rest"
+	"github.com/seldonio/seldon-core/executor/api/tracing"
 	loghandler "github.com/seldonio/seldon-core/executor/logger"
 	"github.com/seldonio/seldon-core/executor/proto/tensorflow/serving"
 	"github.com/seldonio/seldon-core/operator/apis/machinelearning/v1"
-	jaegercfg "github.com/uber/jaeger-client-go/config"
-	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -161,28 +159,6 @@ func runGrpcServer(logger logr.Logger, predictor *v1.PredictorSpec, client seldo
 	}
 }
 
-func initTracing() io.Closer {
-	//Initialise tracing
-	cfg, err := jaegercfg.FromEnv()
-	if err != nil {
-		// parsing errors might happen here, such as when we get a string where we expect a number
-		log.Fatal("Could not parse Jaeger env vars", err.Error())
-	}
-
-	if cfg.ServiceName == "" {
-		cfg.ServiceName = "executor"
-	}
-
-	tracer, closer, err := cfg.NewTracer()
-	if err != nil {
-		log.Fatal("Could not initialize jaeger tracer:", err.Error())
-	}
-
-	opentracing.SetGlobalTracer(tracer)
-
-	return closer
-}
-
 func main() {
 	flag.Parse()
 
@@ -245,7 +221,10 @@ func main() {
 	loghandler.StartDispatcher(*logWorkers, logger, *sdepName, *namespace, *predictorName)
 
 	//Init Tracing
-	closer := initTracing()
+	closer, err := tracing.InitTracing()
+	if err != nil {
+		log.Fatal("Could not initialize jaeger tracer", err.Error())
+	}
 	defer closer.Close()
 
 	if *transport == "rest" {