feature: experimental tracing support

common/interfaces.go

@@ -21,6 +21,8 @@ type Configurer interface {
 	UnmarshalKey(name string, out any) error
 	// Has checks if a config section exists.
 	Has(name string) bool
+	// Experimental returns true if experimental mode is enabled.
+	Experimental() bool
 }
 
 type Pool interface {

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/roadrunner-server/sdk/v4 v4.5.3
 	github.com/stretchr/testify v1.8.4
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1
+	go.opentelemetry.io/otel/sdk v1.21.0
 	go.uber.org/zap v1.26.0
 	golang.org/x/net v0.19.0
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20231127180814-3a041ad873d4
@@ -40,8 +41,8 @@ require (
 	github.com/roadrunner-server/tcplisten v1.4.0 // indirect
 	github.com/rogpeppe/go-internal v1.11.0 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
-	github.com/tklauser/go-sysconf v0.3.12 // indirect
-	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/tklauser/go-sysconf v0.3.13 // indirect
+	github.com/tklauser/numcpus v0.7.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.3 // indirect
 	go.opentelemetry.io/otel v1.21.0 // indirect
 	go.opentelemetry.io/otel/metric v1.21.0 // indirect

go.sum

@@ -69,8 +69,12 @@ github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcU
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU=
 github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
+github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08lq3r4=
+github.com/tklauser/go-sysconf v0.3.13/go.mod h1:zwleP4Q4OehZHGn4CYZDipCgg9usW5IJePewFCGVEa0=
 github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk=
 github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY=
+github.com/tklauser/numcpus v0.7.0 h1:yjuerZP127QG9m5Zh/mSO4wqurYil27tHrqwRoRjpr4=
+github.com/tklauser/numcpus v0.7.0/go.mod h1:bb6dMVcj8A42tSE7i32fsIUCbQNllK5iDguyOZRUzAY=
 github.com/yusufpapurcu/wmi v1.2.3 h1:E1ctvB7uKFMOJw3fdOW32DwGE9I7t++CRUEMKvFoFiw=
 github.com/yusufpapurcu/wmi v1.2.3/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1 h1:SpGay3w+nEwMpfVnbqOLH5gY52/foP8RE8UzTZ1pdSE=
@@ -79,6 +83,8 @@ go.opentelemetry.io/otel v1.21.0 h1:hzLeKBZEL7Okw2mGzZ0cc4k/A7Fta0uoPgaJCr8fsFc=
 go.opentelemetry.io/otel v1.21.0/go.mod h1:QZzNPQPm1zLX4gZK4cMi+71eaorMSGT3A4znnUvNNEo=
 go.opentelemetry.io/otel/metric v1.21.0 h1:tlYWfeo+Bocx5kLEloTjbcDwBuELRrIFxwdQ36PlJu4=
 go.opentelemetry.io/otel/metric v1.21.0/go.mod h1:o1p3CA8nNHW8j5yuQLdc1eeqEaPfzug24uvsyIEJRWM=
+go.opentelemetry.io/otel/sdk v1.21.0 h1:FTt8qirL1EysG6sTQRZ5TokkU8d0ugCj8htOgThZXQ8=
+go.opentelemetry.io/otel/sdk v1.21.0/go.mod h1:Nna6Yv7PWTdgJHVRD9hIYywQBRx7pbox6nwBnZIxl/E=
 go.opentelemetry.io/otel/trace v1.21.0 h1:WD9i5gzvoUPuXIXH24ZNBudiarZDKuekPqi/E8fpfLc=
 go.opentelemetry.io/otel/trace v1.21.0/go.mod h1:LGbsEB0f9LGjN+OZaQQ26sohbOmiMR+BaslueVtS/qQ=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=

plugin.go

@@ -6,6 +6,10 @@ import (
 	"sync"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"go.opentelemetry.io/otel/propagation"
+
+	jprop "go.opentelemetry.io/contrib/propagators/jaeger"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 
 	"github.com/roadrunner-server/endure/v2/dep"
 	"github.com/roadrunner-server/errors"
@@ -30,16 +34,25 @@ const (
 	RrMode     string = "RR_MODE"
 )
 
+type Tracer interface {
+	Tracer() *sdktrace.TracerProvider
+}
+
 type Plugin struct {
-	mu            *sync.RWMutex
-	config        *Config
-	gPool         common.Pool
-	opts          []grpc.ServerOption
-	server        *grpc.Server
-	rrServer      common.Server
-	proxyList     []*proxy.Proxy
-	healthServer  *HealthCheckServer
+	mu           *sync.RWMutex
+	config       *Config
+	gPool        common.Pool
+	opts         []grpc.ServerOption
+	server       *grpc.Server
+	rrServer     common.Server
+	proxyList    []*proxy.Proxy
+	healthServer *HealthCheckServer
+
+	experimental bool
+
 	statsExporter *metrics.StatsExporter
+	prop          propagation.TextMapPropagator
+	tracer        *sdktrace.TracerProvider
 
 	queueSize       prometheus.Gauge
 	requestCounter  *prometheus.CounterVec
@@ -108,6 +121,9 @@ func (p *Plugin) Init(cfg common.Configurer, log common.Logger, server common.Se
 		[]string{"grpc_method"},
 	)
 
+	p.prop = propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{}, jprop.Jaeger{})
+	p.tracer = sdktrace.NewTracerProvider()
+	p.experimental = cfg.Experimental()
 	p.interceptors = make(map[string]common.Interceptor)
 
 	return nil
@@ -246,5 +262,8 @@ func (p *Plugin) Collects() []*dep.In {
 			p.interceptors[interceptor.Name()] = interceptor
 			p.mu.Unlock()
 		}, (*common.Interceptor)(nil)),
+		dep.Fits(func(pp any) {
+			p.tracer = pp.(Tracer).Tracer()
+		}, (*Tracer)(nil)),
 	}
 }

proxy/proxy.go

@@ -16,6 +16,7 @@ import (
 	"github.com/roadrunner-server/sdk/v4/payload"
 	"github.com/roadrunner-server/sdk/v4/pool/static_pool"
 	"github.com/roadrunner-server/sdk/v4/worker"
+	"go.opentelemetry.io/otel/propagation"
 	"golang.org/x/net/context"
 	spb "google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/grpc"
@@ -64,6 +65,7 @@ type rpcContext struct {
 // Proxy manages GRPC/RoadRunner bridge.
 type Proxy struct {
 	mu       *sync.RWMutex
+	prop     propagation.TextMapPropagator
 	grpcPool Pool
 	name     string
 	metadata string
@@ -73,9 +75,10 @@ type Proxy struct {
 }
 
 // NewProxy creates new service proxy object.
-func NewProxy(name string, metadata string, grpcPool Pool, mu *sync.RWMutex) *Proxy {
+func NewProxy(name string, metadata string, grpcPool Pool, mu *sync.RWMutex, prop propagation.TextMapPropagator) *Proxy {
 	return &Proxy{
 		mu:       mu,
+		prop:     prop,
 		grpcPool: grpcPool,
 		name:     name,
 		metadata: metadata,
@@ -245,6 +248,7 @@ func (p *Proxy) responseMetadata(resp *payload.Payload) (metadata.MD, error) {
 func (p *Proxy) makePayload(ctx context.Context, method string, body *codec.RawMessage, pld *payload.Payload) error {
 	ctxMD := make(map[string][]string)
 
+	p.prop.Inject(ctx, propagation.HeaderCarrier(ctxMD))
 	if md, ok := metadata.FromIncomingContext(ctx); ok {
 		for k, v := range md {
 			ctxMD[k] = v

server.go

@@ -48,7 +48,9 @@ func (p *Plugin) createGRPCserver(interceptors map[string]common.Interceptor) (*
 	)
 
 	// append OTEL grpc server handler
-	opts = append(opts, grpc.StatsHandler(otelgrpc.NewServerHandler()))
+	if p.experimental {
+		opts = append(opts, grpc.StatsHandler(otelgrpc.NewServerHandler(otelgrpc.WithTracerProvider(p.tracer))))
+	}
 
 	server := grpc.NewServer(opts...)
 
@@ -64,7 +66,7 @@ func (p *Plugin) createGRPCserver(interceptors map[string]common.Interceptor) (*
 		}
 
 		for _, service := range services {
-			px := proxy.NewProxy(fmt.Sprintf("%s.%s", service.Package, service.Name), p.config.Proto[i], p.gPool, p.mu)
+			px := proxy.NewProxy(fmt.Sprintf("%s.%s", service.Package, service.Name), p.config.Proto[i], p.gPool, p.mu, p.prop)
 			for _, m := range service.Methods {
 				px.RegisterMethod(m.Name)
 			}

status.go

@@ -35,8 +35,8 @@ func (p *Plugin) Ready() (*status.Status, error) {
 	workers := p.gPool.Workers()
 
 	for i := 0; i < len(workers); i++ {
-		// If state of the worker is ready (at least 1)
-		// we assume, that plugin's worker pool is ready
+		// If the state of the worker is ready (at least 1)
+		// we assume that plugin's worker pool is ready
 		if workers[i].State().Compare(fsm.StateReady) {
 			return &status.Status{
 				Code: http.StatusOK,

tests/configs/.rr-grpc-init.yaml

@@ -10,9 +10,9 @@ server:
 
 # GRPC service configuration
 grpc:
-  listen: "tcp://127.0.0.1:9001"
+  listen: "tcp://127.0.0.1:9091"
   proto:
-    - "proto/test/test.proto"
+    - "proto/service/service.proto"
   max_send_msg_size: 50
   max_recv_msg_size: 50
   max_connection_idle: 0s

tests/configs/.rr-grpc-otel.yaml

@@ -0,0 +1,29 @@
+version: '3'
+
+rpc:
+  listen: "tcp://127.0.0.1:6001"
+
+server:
+  command: "php php_test_files/worker-grpc.php"
+  relay: "pipes"
+  relay_timeout: "20s"
+
+grpc:
+  listen: "tcp://127.0.0.1:9092"
+  proto:
+    - "proto/service/service.proto"
+  ping_time: 1s
+  timeout: 200s
+  pool:
+    num_workers: 2
+    allocate_timeout: 60s
+    destroy_timeout: 60
+
+otel:
+  resource:
+    service_name: "rr_test_grpc"
+    service_version: "1.0.0"
+    service_namespace: "RR-gRPC"
+    service_instance_id: "UUID-super-long-unique-id"
+  insecure: false
+  exporter: stderr

tests/go.mod

@@ -77,8 +77,8 @@ require (
 	github.com/spf13/viper v1.17.0 // indirect
 	github.com/stretchr/objx v0.5.1 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
-	github.com/tklauser/go-sysconf v0.3.12 // indirect
-	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/tklauser/go-sysconf v0.3.13 // indirect
+	github.com/tklauser/numcpus v0.7.0 // indirect
 	github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.3 // indirect

tests/go.sum

@@ -320,10 +320,10 @@ github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcU
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
-github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU=
-github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
-github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk=
-github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY=
+github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08lq3r4=
+github.com/tklauser/go-sysconf v0.3.13/go.mod h1:zwleP4Q4OehZHGn4CYZDipCgg9usW5IJePewFCGVEa0=
+github.com/tklauser/numcpus v0.7.0 h1:yjuerZP127QG9m5Zh/mSO4wqurYil27tHrqwRoRjpr4=
+github.com/tklauser/numcpus v0.7.0/go.mod h1:bb6dMVcj8A42tSE7i32fsIUCbQNllK5iDguyOZRUzAY=
 github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
 github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
 github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
@@ -543,8 +543,6 @@ golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc=
 golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

tests/grpc_plugin_test.go

@@ -103,12 +103,110 @@ func TestGrpcInit(t *testing.T) {
 		}
 	}()
 
-	time.Sleep(time.Second * 1)
+	time.Sleep(time.Second)
+
+	conn, err := grpc.Dial("127.0.0.1:9091", grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+
+	client := service.NewEchoClient(conn)
+	resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"})
+	require.NoError(t, err)
+	require.Equal(t, "TOST", resp.Msg)
+
 	stopCh <- struct{}{}
 
 	wg.Wait()
 }
 
+func TestGrpcOtel(t *testing.T) {
+	// TODO(rustatian) use the: https://pkg.go.dev/go.opentelemetry.io/otel/sdk/trace/tracetest"
+	rd, wr, err := os.Pipe()
+	assert.NoError(t, err)
+	os.Stderr = wr
+
+	cont := endure.New(slog.LevelDebug)
+
+	cfg := &config.Plugin{
+		Version:              "2023.3.0",
+		ExperimentalFeatures: true,
+		Path:                 "configs/.rr-grpc-otel.yaml",
+		Prefix:               "rr",
+	}
+
+	err = cont.RegisterAll(
+		cfg,
+		&grpcPlugin.Plugin{},
+		&rpcPlugin.Plugin{},
+		&logger.Plugin{},
+		&otel.Plugin{},
+		&server.Plugin{},
+	)
+	assert.NoError(t, err)
+
+	err = cont.Init()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ch, err := cont.Serve()
+	assert.NoError(t, err)
+
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+
+	stopCh := make(chan struct{}, 1)
+
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case e := <-ch:
+				assert.Fail(t, "error", e.Error.Error())
+				err = cont.Stop()
+				assert.NoError(t, err)
+			case <-sig:
+				err = cont.Stop()
+				assert.NoError(t, err)
+				return
+			case <-stopCh:
+				err = cont.Stop()
+				assert.NoError(t, err)
+				return
+			}
+		}
+	}()
+
+	time.Sleep(time.Second)
+
+	conn, err := grpc.Dial("127.0.0.1:9092", grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+
+	client := service.NewEchoClient(conn)
+	resp, err := client.Ping(context.Background(), &service.Message{Msg: "TOST"})
+	require.NoError(t, err)
+	require.Equal(t, "TOST", resp.Msg)
+
+	stopCh <- struct{}{}
+	wg.Wait()
+
+	time.Sleep(time.Second * 3)
+
+	_ = wr.Close()
+	buf := new(bytes.Buffer)
+	_, err = io.Copy(buf, rd)
+	assert.NoError(t, err)
+
+	// contains spans
+	assert.Contains(t, buf.String(), "service.Echo/Ping")
+	assert.Contains(t, buf.String(), "RR-gRPC")
+	assert.Contains(t, buf.String(), "2023.3.0")
+}
+
 func TestGrpcCheckStatus(t *testing.T) {
 	cont := endure.New(slog.LevelDebug)