Merge pull request #103218 from dashpole/otel_clientgo

Add tracing to apiserver client-go requests

staging/src/k8s.io/apiserver/pkg/server/options/coreapi.go

@@ -21,11 +21,14 @@ import (
 	"time"
 
 	"github.com/spf13/pflag"
+	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/server"
+	"k8s.io/apiserver/pkg/util/feature"
 	clientgoinformers "k8s.io/client-go/informers"
 	clientgoclientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/component-base/traces"
 )
 
 // CoreAPIOptions contains options to configure the connection to a core API Kubernetes apiserver.
@@ -69,6 +72,9 @@ func (o *CoreAPIOptions) ApplyTo(config *server.RecommendedConfig) error {
 			return err
 		}
 	}
+	if feature.DefaultFeatureGate.Enabled(features.APIServerTracing) {
+		kubeconfig.Wrap(traces.WrapperFor(config.TracerProvider))
+	}
 	clientgoExternalClient, err := clientgoclientset.NewForConfig(kubeconfig)
 	if err != nil {
 		return fmt.Errorf("failed to create Kubernetes clientset: %v", err)

staging/src/k8s.io/apiserver/pkg/server/options/recommended.go

@@ -104,6 +104,14 @@ func (o *RecommendedOptions) ApplyTo(config *server.RecommendedConfig) error {
 	if err := o.Etcd.ApplyTo(&config.Config); err != nil {
 		return err
 	}
+	if err := o.EgressSelector.ApplyTo(&config.Config); err != nil {
+		return err
+	}
+	if feature.DefaultFeatureGate.Enabled(features.APIServerTracing) {
+		if err := o.Traces.ApplyTo(config.Config.EgressSelector, &config.Config); err != nil {
+			return err
+		}
+	}
 	if err := o.SecureServing.ApplyTo(&config.Config.SecureServing, &config.Config.LoopbackClientConfig); err != nil {
 		return err
 	}
@@ -127,9 +135,6 @@ func (o *RecommendedOptions) ApplyTo(config *server.RecommendedConfig) error {
 	} else if err := o.Admission.ApplyTo(&config.Config, config.SharedInformerFactory, config.ClientConfig, o.FeatureGate, initializers...); err != nil {
 		return err
 	}
-	if err := o.EgressSelector.ApplyTo(&config.Config); err != nil {
-		return err
-	}
 	if feature.DefaultFeatureGate.Enabled(features.APIPriorityAndFairness) {
 		if config.ClientConfig != nil {
 			if config.MaxRequestsInFlight+config.MaxMutatingRequestsInFlight <= 0 {
@@ -146,11 +151,6 @@ func (o *RecommendedOptions) ApplyTo(config *server.RecommendedConfig) error {
 			klog.Warningf("Neither kubeconfig is provided nor service-account is mounted, so APIPriorityAndFairness will be disabled")
 		}
 	}
-	if feature.DefaultFeatureGate.Enabled(features.APIServerTracing) {
-		if err := o.Traces.ApplyTo(config.Config.EgressSelector, &config.Config); err != nil {
-			return err
-		}
-	}
 	return nil
 }
 

staging/src/k8s.io/apiserver/pkg/server/options/tracing.go

@@ -106,6 +106,9 @@ func (o *TracingOptions) ApplyTo(es *egressselector.EgressSelector, c *server.Co
 	}
 	tp := traces.NewProvider(context.Background(), sampler, resourceOpts, opts...)
 	c.TracerProvider = &tp
+	if c.LoopbackClientConfig != nil {
+		c.LoopbackClientConfig.Wrap(traces.WrapperFor(c.TracerProvider))
+	}
 	return nil
 }
 

staging/src/k8s.io/component-base/go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/spf13/cobra v1.1.3
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.7.0
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.20.0
 	go.opentelemetry.io/otel v0.20.0
 	go.opentelemetry.io/otel/exporters/otlp v0.20.0
 	go.opentelemetry.io/otel/sdk v0.20.0

staging/src/k8s.io/component-base/traces/utils.go

@@ -18,14 +18,17 @@ package traces
 
 import (
 	"context"
+	"net/http"
 
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 	"go.opentelemetry.io/otel/exporters/otlp"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpgrpc"
 	"go.opentelemetry.io/otel/propagation"
 	"go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/trace"
 
+	"k8s.io/client-go/transport"
 	"k8s.io/klog/v2"
 )
 
@@ -52,6 +55,25 @@ func NewProvider(ctx context.Context, baseSampler sdktrace.Sampler, resourceOpts
 	)
 }
 
+// WrapperFor can be used to add tracing to a *rest.Config. Example usage:
+//     tp := traces.NewProvider(...)
+//     config, _ := rest.InClusterConfig()
+//     config.Wrap(traces.WrapperFor(&tp))
+//     kubeclient, _ := clientset.NewForConfig(config)
+func WrapperFor(tp *trace.TracerProvider) transport.WrapperFunc {
+	return func(rt http.RoundTripper) http.RoundTripper {
+		opts := []otelhttp.Option{
+			otelhttp.WithPropagators(Propagators()),
+		}
+		if tp != nil {
+			opts = append(opts, otelhttp.WithTracerProvider(*tp))
+		}
+		// Even if there is no TracerProvider, the otelhttp still handles context propagation.
+		// See https://github.com/open-telemetry/opentelemetry-go/tree/main/example/passthrough
+		return otelhttp.NewTransport(rt, opts...)
+	}
+}
+
 // Propagators returns the recommended set of propagators.
 func Propagators() propagation.TextMapPropagator {
 	return propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{})