Merge pull request containerd#10186 from cpuguy83/propagate_traces_to_shim

Propagate trace contexts to shims

Author: Maksym Pavlenko

cmd/containerd-shim-runc-v2/main_tracing.go

@@ -0,0 +1,21 @@
+//go:build shim_tracing
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package main
+
+import _ "github.com/containerd/containerd/v2/pkg/tracing/plugin"

cmd/containerd-shim-runc-v2/manager/manager_linux.go

@@ -104,6 +104,7 @@ func newCommand(ctx context.Context, id, containerdAddress, containerdTTRPCAddre
 	cmd := exec.Command(self, args...)
 	cmd.Dir = cwd
 	cmd.Env = append(os.Environ(), "GOMAXPROCS=4")
+	cmd.Env = append(cmd.Env, "OTEL_SERVICE_NAME=containerd-shim-"+id)
 	cmd.SysProcAttr = &syscall.SysProcAttr{
 		Setpgid: true,
 	}

cmd/containerd/server/server_linux.go

@@ -26,6 +26,7 @@ import (
 	srvconfig "github.com/containerd/containerd/v2/cmd/containerd/server/config"
 	"github.com/containerd/containerd/v2/pkg/sys"
 	"github.com/containerd/log"
+	"github.com/containerd/otelttrpc"
 	"github.com/containerd/ttrpc"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 )
@@ -66,5 +67,8 @@ func apply(ctx context.Context, config *srvconfig.Config) error {
 }
 
 func newTTRPCServer() (*ttrpc.Server, error) {
-	return ttrpc.NewServer(ttrpc.WithServerHandshaker(ttrpc.UnixSocketRequireSameUser()))
+	return ttrpc.NewServer(
+		ttrpc.WithServerHandshaker(ttrpc.UnixSocketRequireSameUser()),
+		ttrpc.WithUnaryServerInterceptor(otelttrpc.UnaryServerInterceptor()),
+	)
 }

cmd/containerd/server/server_windows.go

@@ -20,6 +20,7 @@ import (
 	"context"
 
 	srvconfig "github.com/containerd/containerd/v2/cmd/containerd/server/config"
+	"github.com/containerd/otelttrpc"
 	"github.com/containerd/ttrpc"
 )
 
@@ -28,5 +29,7 @@ func apply(_ context.Context, _ *srvconfig.Config) error {
 }
 
 func newTTRPCServer() (*ttrpc.Server, error) {
-	return ttrpc.NewServer()
+	return ttrpc.NewServer(
+		ttrpc.WithUnaryServerInterceptor(otelttrpc.UnaryServerInterceptor()),
+	)
 }

core/runtime/v2/shim.go

@@ -28,6 +28,7 @@ import (
 	"strings"
 	"time"
 
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
@@ -46,6 +47,7 @@ import (
 	"github.com/containerd/containerd/v2/pkg/timeout"
 	"github.com/containerd/errdefs"
 	"github.com/containerd/log"
+	"github.com/containerd/otelttrpc"
 	"github.com/containerd/ttrpc"
 	"github.com/containerd/typeurl/v2"
 )
@@ -273,10 +275,16 @@ func makeConnection(ctx context.Context, id string, params client.BootstrapParam
 			}
 		}()
 
-		return ttrpc.NewClient(conn, ttrpc.WithOnClose(onClose)), nil
+		return ttrpc.NewClient(
+			conn,
+			ttrpc.WithOnClose(onClose),
+			ttrpc.WithUnaryClientInterceptor(otelttrpc.UnaryClientInterceptor()),
+		), nil
 	case "grpc":
 		gopts := []grpc.DialOption{
 			grpc.WithTransportCredentials(insecure.NewCredentials()),
+			grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()),   //nolint:staticcheck // Ignore SA1019. Deprecation assumes use of [grpc.NewClient] but we are not using that here.
+			grpc.WithStreamInterceptor(otelgrpc.StreamClientInterceptor()), //nolint:staticcheck // Ignore SA1019. Deprecation assumes use of [grpc.NewClient] but we are not using that here.
 		}
 		return grpcDialContext(params.Address, onClose, gopts...)
 	default:

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/containerd/imgcrypt v1.2.0-rc1
 	github.com/containerd/log v0.1.0
 	github.com/containerd/nri v0.6.1
+	github.com/containerd/otelttrpc v0.0.0-20240305015340-ea5083fda723
 	github.com/containerd/platforms v0.2.1
 	github.com/containerd/plugin v0.1.0
 	github.com/containerd/ttrpc v1.2.5

go.sum

@@ -43,8 +43,20 @@ type item struct {
 	count int
 }
 
+type publisherConfig struct {
+	ttrpcOpts []ttrpc.ClientOpts
+}
+
+type PublisherOpts func(*publisherConfig)
+
+func WithPublishTTRPCOpts(opts ...ttrpc.ClientOpts) PublisherOpts {
+	return func(cfg *publisherConfig) {
+		cfg.ttrpcOpts = append(cfg.ttrpcOpts, opts...)
+	}
+}
+
 // NewPublisher creates a new remote events publisher
-func NewPublisher(address string) (*RemoteEventsPublisher, error) {
+func NewPublisher(address string, opts ...PublisherOpts) (*RemoteEventsPublisher, error) {
 	client, err := ttrpcutil.NewClient(address)
 	if err != nil {
 		return nil, err

pkg/shim/publisher.go

@@ -43,6 +43,7 @@ import (
 	"github.com/containerd/containerd/v2/plugins"
 	"github.com/containerd/containerd/v2/version"
 	"github.com/containerd/log"
+	"github.com/containerd/otelttrpc"
 	"github.com/containerd/plugin"
 	"github.com/containerd/plugin/registry"
 	"github.com/containerd/ttrpc"
@@ -248,7 +249,9 @@ func run(ctx context.Context, manager Manager, config Config) error {
 	}
 
 	ttrpcAddress := os.Getenv(ttrpcAddressEnv)
-	publisher, err := NewPublisher(ttrpcAddress)
+	publisher, err := NewPublisher(ttrpcAddress, WithPublishTTRPCOpts(
+		ttrpc.WithUnaryClientInterceptor(otelttrpc.UnaryClientInterceptor()),
+	))
 	if err != nil {
 		return err
 	}
@@ -398,6 +401,8 @@ func run(ctx context.Context, manager Manager, config Config) error {
 		return fmt.Errorf("required that ttrpc service")
 	}
 
+	ttrpcUnaryInterceptors = append(ttrpcUnaryInterceptors, otelttrpc.UnaryServerInterceptor())
+
 	unaryInterceptor := chainUnaryServerInterceptors(ttrpcUnaryInterceptors...)
 	server, err := newServer(ttrpc.WithUnaryServerInterceptor(unaryInterceptor))
 	if err != nil {

pkg/shim/shim.go

@@ -203,6 +203,9 @@ func newTracer(ctx context.Context, procs []trace.SpanProcessor) (io.Closer, err
 }
 
 func warnTraceConfig(ic *plugin.InitContext) error {
+	if ic.Config == nil {
+		return nil
+	}
 	ctx := ic.Context
 	cfg := ic.Config.(*TraceConfig)
 	var warn bool
@@ -227,6 +230,9 @@ func warnTraceConfig(ic *plugin.InitContext) error {
 }
 
 func warnOTLPConfig(ic *plugin.InitContext) error {
+	if ic.Config == nil {
+		return nil
+	}
 	ctx := ic.Context
 	cfg := ic.Config.(*OTLPConfig)
 	var warn bool