[#131]: feature: experimental tracing support

.github/workflows/linux.yml

@@ -78,7 +78,7 @@ jobs:
       - name: Run golang tests with coverage
         run: |
           cd protoc_plugins
-          
+     
           go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=$(cat ../tests/pkgs.txt) -coverprofile=../tests/coverage-ci/protoc_gen.out -covermode=atomic ./...
 
       - name: Run golang tests with coverage
@@ -95,8 +95,11 @@ jobs:
           mkcert -client localhost 127.0.0.1 ::1 
           cp -r localhost+2-client-key.pem localhost+2-client.pem localhost+2-key.pem localhost+2.pem test-certs/
           cp -r $(mkcert -CAROOT)/rootCA.pem test-certs/
-
+          
+          docker-compose -f env/docker-compose-otel.yaml up -d
+          sleep 30
           go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=$(cat pkgs.txt) -coverprofile=./coverage-ci/grpc.out -covermode=atomic ./...
+          docker-compose -f env/docker-compose-otel.yaml down
 
       - name: Archive code coverage results
         uses: actions/upload-artifact@v3

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,9 @@ 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/contrib/propagators/jaeger v1.21.1
+	go.opentelemetry.io/otel v1.21.0
+	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,10 +43,9 @@ 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.opentelemetry.io/otel/trace v1.21.0 // indirect
 	go.uber.org/goleak v1.3.0 // indirect

go.sum

@@ -67,18 +67,22 @@ github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKl
 github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 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/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/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=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1/go.mod h1:4UoMYEZOC0yN/sPGH76KPkkU7zgiEWYWL9vwmbnTJPE=
+go.opentelemetry.io/contrib/propagators/jaeger v1.21.1 h1:f4beMGDKiVzg9IcX7/VuWVy+oGdjx3dNJ72YehmtY5k=
+go.opentelemetry.io/contrib/propagators/jaeger v1.21.1/go.mod h1:U9jhkEl8d1LL+QXY7q3kneJWJugiN3kZJV2OWz3hkBY=
 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=
@@ -95,8 +99,6 @@ golang.org/x/sync v0.5.0 h1:60k92dhOjHxJkrqnwsfl8KuaHbn/5dl0lUPUklKo3qE=
 golang.org/x/sync v0.5.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.1.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/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=

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-duplicate-2.yaml

@@ -32,4 +32,4 @@ grpc:
         num_workers: 2
         max_jobs: 0
         allocate_timeout: 60s
-        destroy_timeout: 60
+        destroy_timeout: 60s

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

@@ -59,4 +59,4 @@ grpc:
         num_workers: 2
         max_jobs: 0
         allocate_timeout: 60s
-        destroy_timeout: 60
+        destroy_timeout: 60s

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

@@ -32,4 +32,4 @@ grpc:
         num_workers: 2
         max_jobs: 0
         allocate_timeout: 60s
-        destroy_timeout: 60
+        destroy_timeout: 60s

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
@@ -26,4 +26,4 @@ grpc:
     num_workers: 2
     max_jobs: 0
     allocate_timeout: 60s
-    destroy_timeout: 60
+    destroy_timeout: 60s

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

@@ -33,7 +33,7 @@ grpc:
     num_workers: 2
     max_jobs: 0
     allocate_timeout: 60s
-    destroy_timeout: 60
+    destroy_timeout: 60s
 
 metrics:
   address: 127.0.0.1:2112

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

@@ -0,0 +1,35 @@
+version: '3'
+
+rpc:
+  listen: "tcp://127.0.0.1:6001"
+
+server:
+  command: "php php_test_files/worker-grpc.php"
+  relay: "pipes"
+  relay_timeout: "20s"
+
+logs:
+  mode: development
+  level: debug
+
+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: 60s
+
+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: true
+  compress: true
+  exporter: otlp
+  endpoint: 127.0.0.1:4319

tests/configs/.rr-grpc-rq-exception.yaml

@@ -62,4 +62,4 @@ grpc:
     num_workers: 2
     max_jobs: 0
     allocate_timeout: 60s
-    destroy_timeout: 60
+    destroy_timeout: 60s

tests/configs/.rr-grpc-rq-issue1193.yaml

@@ -34,4 +34,4 @@ grpc:
     num_workers: 2
     max_jobs: 0
     allocate_timeout: 60s
-    destroy_timeout: 60
+    destroy_timeout: 60s

tests/configs/.rr-grpc-rq-multiple.yaml

@@ -63,4 +63,4 @@ grpc:
     num_workers: 2
     max_jobs: 0
     allocate_timeout: 60s
-    destroy_timeout: 60
+    destroy_timeout: 60s

tests/configs/.rr-grpc-rq-otlp.yaml

@@ -10,7 +10,7 @@ server:
 
 logs:
   mode: development
-  level: error
+  level: debug
 
 # GRPC service configuration
 grpc:
@@ -21,9 +21,6 @@ grpc:
 
   max_send_msg_size: 50
   max_recv_msg_size: 50
-  max_connection_idle: 0s
-  max_connection_age: 0s
-  max_connection_age_grace: 0s
   max_concurrent_streams: 10
   ping_time: 1s
   timeout: 200s
@@ -32,11 +29,14 @@ grpc:
     num_workers: 2
     max_jobs: 0
     allocate_timeout: 60s
-    destroy_timeout: 60
+    destroy_timeout: 60s
 
 otel:
   insecure: false
   compress: true
   exporter: stderr
-  service_name: rr_test
-  service_version: 1.0.0
+  resource:
+    service_name: "rr_test_grpc"
+    service_version: "1.0.0"
+    service_namespace: "RR-gRPC"
+    service_instance_id: "UUID-super-long-unique-id"

tests/configs/.rr-grpc-rq-tls-rootca.yaml

@@ -67,4 +67,4 @@ grpc:
     num_workers: 2
     max_jobs: 0
     allocate_timeout: 60s
-    destroy_timeout: 60
+    destroy_timeout: 60s

tests/configs/.rr-grpc-rq-tls.yaml

@@ -65,4 +65,4 @@ grpc:
     num_workers: 2
     max_jobs: 0
     allocate_timeout: 60s
-    destroy_timeout: 60
+    destroy_timeout: 60s

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

@@ -32,4 +32,4 @@ grpc:
     num_workers: 2
     max_jobs: 0
     allocate_timeout: 60s
-    destroy_timeout: 60
+    destroy_timeout: 60s