vendor: update buildkit to v0.32.0-rc1

Signed-off-by: Tonis Tiigi <tonistiigi@gmail.com>
2026-07-22 15:27:47 -07:00
parent efd9aa1dea
commit 0cf7592d41
464 changed files with 19941 additions and 15772 deletions
@@ -121,12 +121,28 @@ func (f *envVarFeatureGates) Enabled(key Feature) bool {
 // Features set via this method take precedence over
 // the features set via environment variables.
 func (f *envVarFeatureGates) Set(featureName Feature, featureValue bool) error {
+	return f.set(featureName, featureValue, false)
+}
+
+// SetForTesting sets the given feature to the given value. This method
+// bypasses the check for locked features and should only be used for
+// testing purposes.
+//
+// Features set via this method take precedence over
+// the features set via environment variables.
+func (f *envVarFeatureGates) SetForTesting(featureName Feature, featureValue bool) error {
+	return f.set(featureName, featureValue, true)
+}
+
+func (f *envVarFeatureGates) set(featureName Feature, featureValue bool, allowChangingLockedFeatures bool) error {
 	feature, ok := f.known[featureName]
 	if !ok {
 		return fmt.Errorf("feature %q is not registered in FeatureGates %q", featureName, f.callSiteName)
 	}
-	if feature.LockToDefault && feature.Default != featureValue {
-		return fmt.Errorf("cannot set feature gate %q to %v, feature is locked to %v", featureName, featureValue, feature.Default)
+	if !allowChangingLockedFeatures {
+		if feature.LockToDefault && feature.Default != featureValue {
+			return fmt.Errorf("cannot set feature gate %q to %v, feature is locked to %v", featureName, featureValue, feature.Default)
+		}
 	}

 	f.lockEnabledViaSetMethod.Lock()
@@ -141,6 +157,13 @@ func (f *envVarFeatureGates) Set(featureName Feature, featureValue bool) error {
 // read from the corresponding environmental variable.
 func (f *envVarFeatureGates) getEnabledMapFromEnvVar() map[Feature]bool {
 	f.readEnvVarsOnce.Do(func() {
+		// This code does not really support contextual logging. Making it do so has huge
+		// implications for several call chains because the Enabled call then needs
+		// a `*WithLogger` variant. This does not matter in Kubernetes itself because
+		// all Kubernetes components replace the feature gate implementation used
+		// by client-go, but it might matter elsewhere.
+		logger := klog.Background()
+
 		featureGatesState := map[Feature]bool{}
 		for feature, featureSpec := range f.known {
 			featureState, featureStateSet := os.LookupEnv(fmt.Sprintf("KUBE_FEATURE_%s", feature))
@@ -150,10 +173,10 @@ func (f *envVarFeatureGates) getEnabledMapFromEnvVar() map[Feature]bool {
 			boolVal, boolErr := strconv.ParseBool(featureState)
 			switch {
 			case boolErr != nil:
-				utilruntime.HandleError(fmt.Errorf("cannot set feature gate %q to %q, due to %v", feature, featureState, boolErr))
+				utilruntime.HandleErrorWithLogger(logger, boolErr, "Could not set feature gate", "feature", feature, "desiredState", featureState)
 			case featureSpec.LockToDefault:
 				if boolVal != featureSpec.Default {
-					utilruntime.HandleError(fmt.Errorf("cannot set feature gate %q to %q, feature is locked to %v", feature, featureState, featureSpec.Default))
+					utilruntime.HandleErrorWithLogger(logger, nil, "Could not set feature gate, feature is locked", "feature", feature, "desiredState", featureState, "lockedState", featureSpec.Default)
 					break
 				}
 				featureGatesState[feature] = featureSpec.Default
@@ -166,10 +189,10 @@ func (f *envVarFeatureGates) getEnabledMapFromEnvVar() map[Feature]bool {

 		for feature, featureSpec := range f.known {
 			if featureState, ok := featureGatesState[feature]; ok {
-				klog.V(1).InfoS("Feature gate updated state", "feature", feature, "enabled", featureState)
+				logger.V(1).Info("Feature gate updated state", "feature", feature, "enabled", featureState)
 				continue
 			}
-			klog.V(1).InfoS("Feature gate default state", "feature", feature, "enabled", featureSpec.Default)
+			logger.V(1).Info("Feature gate default state", "feature", feature, "enabled", featureSpec.Default)
 		}
 	})
 	return f.enabledViaEnvVar.Load().(map[Feature]bool)
@@ -17,7 +17,7 @@ limitations under the License.
 package features

 import (
-	"errors"
+	"context"
 	"sync/atomic"

 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -127,7 +127,9 @@ func AddVersionedFeaturesToExistingFeatureGates(registry VersionedRegistry) erro
 //	clientgofeaturegate.ReplaceFeatureGates(utilfeature.DefaultMutableFeatureGate)
 func ReplaceFeatureGates(newFeatureGates Gates) {
 	if replaceFeatureGatesWithWarningIndicator(newFeatureGates) {
-		utilruntime.HandleError(errors.New("the default feature gates implementation has already been used and now it's being overwritten. This might lead to unexpected behaviour. Check your initialization order"))
+		// TODO (?): A new API would be needed where callers pass in a context or logger.
+		// Probably not worth it.
+		utilruntime.HandleErrorWithContext(context.TODO(), nil, "The default feature gates implementation has already been used and now it's being overwritten. This might lead to unexpected behaviour. Check your initialization order.")
 	}
 }

@@ -31,6 +31,20 @@ import (
 // of code conflicts because changes are more likely to be scattered
 // across the file.
 const (
+	// owner: @michaelasp
+	// beta: v1.36
+	//
+	// Allow the client to process events atomically rather than a stream of
+	// events for items popped off the FIFO.
+	AtomicFIFO Feature = "AtomicFIFO"
+
+	// owner: @yt2985
+	// beta: 1.36
+	//
+	// If enabled, allows clients to gracefully handle Certificate Authority (CA)
+	// rotations without dropping connections or requiring a restart.
+	ClientsAllowCARotation Feature = "ClientsAllowCARotation"
+
 	// owner: @benluddy
 	// kep: https://kep.k8s.io/4222
 	// alpha: 1.32
@@ -41,6 +55,13 @@ const (
 	// "application/json" or "application/apply-patch+yaml", respectively.
 	ClientsAllowCBOR Feature = "ClientsAllowCBOR"

+	// owner: @enj
+	// beta: v1.36
+	//
+	// If enabled, the client-go TLS transport cache uses weak pointers to allow
+	// garbage collection of unused transports, preventing unbounded cache growth.
+	ClientsAllowTLSCacheGC Feature = "ClientsAllowTLSCacheGC"
+
 	// owner: @benluddy
 	// kep: https://kep.k8s.io/4222
 	// alpha: 1.32
@@ -69,6 +90,12 @@ const (
 	// GA: v1.35
 	InformerResourceVersion Feature = "InformerResourceVersion"

+	// owner: @michaelasp
+	// beta: v1.36
+	//
+	// Allow the FIFO to unlock while processing items to allow other goroutines to add items to the queue.
+	UnlockWhileProcessingFIFO Feature = "UnlockWhileProcessingFIFO"
+
 	// owner: @p0lyn0mial
 	// beta: v1.30
 	//
@@ -82,14 +109,24 @@ const (
 // After registering with the binary, the features are, by default, controllable using environment variables.
 // For more details, please see envVarFeatureGates implementation.
 var defaultVersionedKubernetesFeatureGates = map[Feature]VersionedSpecs{
+	AtomicFIFO: {
+		{Version: version.MustParse("1.36"), Default: true, PreRelease: Beta},
+	},
+	ClientsAllowCARotation: {
+		{Version: version.MustParse("1.36"), Default: true, PreRelease: Beta},
+	},
 	ClientsAllowCBOR: {
 		{Version: version.MustParse("1.32"), Default: false, PreRelease: Alpha},
 	},
+	ClientsAllowTLSCacheGC: {
+		{Version: version.MustParse("1.36"), Default: true, PreRelease: Beta},
+	},
 	ClientsPreferCBOR: {
 		{Version: version.MustParse("1.32"), Default: false, PreRelease: Alpha},
 	},
 	InOrderInformers: {
 		{Version: version.MustParse("1.33"), Default: true, PreRelease: Beta},
+		{Version: version.MustParse("1.36"), Default: true, PreRelease: GA, LockToDefault: true},
 	},
 	InOrderInformersBatchProcess: {
 		{Version: version.MustParse("1.35"), Default: true, PreRelease: Beta},
@@ -98,6 +135,9 @@ var defaultVersionedKubernetesFeatureGates = map[Feature]VersionedSpecs{
 		{Version: version.MustParse("1.30"), Default: false, PreRelease: Alpha},
 		{Version: version.MustParse("1.35"), Default: true, PreRelease: GA},
 	},
+	UnlockWhileProcessingFIFO: {
+		{Version: version.MustParse("1.36"), Default: true, PreRelease: Beta},
+	},
 	WatchListClient: {
 		{Version: version.MustParse("1.30"), Default: false, PreRelease: Beta},
 		{Version: version.MustParse("1.35"), Default: true, PreRelease: Beta},
@@ -38,7 +38,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
-	"k8s.io/apimachinery/pkg/util/dump"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/pkg/apis/clientauthentication"
@@ -51,6 +50,7 @@ import (
 	"k8s.io/client-go/util/connrotation"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/clock"
+	"k8s.io/utils/dump"
 )

 const execInfoEnv = "KUBERNETES_EXEC_INFO"
@@ -185,8 +185,8 @@ func newAuthenticator(c *cache, isTerminalFunc func(int) bool, config *api.ExecC

 	allowlistLookup := sets.New[string]()
 	for _, entry := range config.PluginPolicy.Allowlist {
-		if entry.Name != "" {
-			allowlistLookup.Insert(entry.Name)
+		if entry.Command != "" {
+			allowlistLookup.Insert(entry.Command)
 		}
 	}

@@ -641,14 +641,14 @@ func (a *Authenticator) checkAllowlistLocked(cmd *exec.Cmd) error {
 func (a *Authenticator) resolveAllowListEntriesLocked(commandHint string) {
 	hintName := filepath.Base(commandHint)
 	for _, entry := range a.execPluginPolicy.Allowlist {
-		entryBasename := filepath.Base(entry.Name)
+		entryBasename := filepath.Base(entry.Command)
 		if hintName != "" && hintName != entryBasename {
 			// we got a hint, and this allowlist entry does not match it
 			continue
 		}
-		entryResolvedPath, err := exec.LookPath(entry.Name)
+		entryResolvedPath, err := exec.LookPath(entry.Command)
 		if err != nil {
-			klog.V(5).ErrorS(err, "resolving credential plugin allowlist", "name", entry.Name)
+			klog.V(5).ErrorS(err, "resolving credential plugin allowlist", "name", entry.Command)
 			continue
 		}
 		if entryResolvedPath != "" {
@@ -691,10 +691,10 @@ func validateAllowlist(list []api.AllowlistEntry) error {
 			return fmt.Errorf("misconfigured credential plugin allowlist: empty allowlist entry #%d", i+1)
 		}

-		if cleaned := filepath.Clean(item.Name); cleaned != item.Name {
-			return fmt.Errorf("non-normalized file path: %q vs %q", item.Name, cleaned)
-		} else if item.Name == "" {
-			return fmt.Errorf("empty file path: %q", item.Name)
+		if cleaned := filepath.Clean(item.Command); cleaned != item.Command {
+			return fmt.Errorf("non-normalized file path: %q vs %q", item.Command, cleaned)
+		} else if item.Command == "" {
+			return fmt.Errorf("empty file path: %q", item.Command)
 		}
 	}

@@ -298,12 +298,12 @@ type ExecConfig struct {
 // the logical AND of all checks corresponding to the specified fields within
 // the entry.
 type AllowlistEntry struct {
-	// Name matching is performed by first resolving the absolute path of both
+	// Command matching is performed by first resolving the absolute path of both
 	// the plugin and the name in the allowlist entry using `exec.LookPath`. It
 	// will be called on both, and the resulting strings must be equal. If
-	// either call to `exec.LookPath` results in an error, the `Name` check
+	// either call to `exec.LookPath` results in an error, the `Command` check
 	// will be considered a failure.
-	Name string `json:"-"`
+	Command string `json:"-"`
 }

 // PluginPolicy describes the policy type and allowlist (if any) for client-go
@@ -679,11 +679,13 @@ func (config *inClusterClientConfig) Possible() bool {
 // to the default config.
 func BuildConfigFromFlags(masterUrl, kubeconfigPath string) (*restclient.Config, error) {
 	if kubeconfigPath == "" && masterUrl == "" {
+		//nolint:logcheck // A helper function like this should not log. But this is probably part of the the established client-go API and not worth changing.
 		klog.Warning("Neither --kubeconfig nor --master was specified.  Using the inClusterConfig.  This might not work.")
 		kubeconfig, err := restclient.InClusterConfig()
 		if err == nil {
 			return kubeconfig, nil
 		}
+		//nolint:logcheck // A helper function like this should not log. But this is probably part of the the established client-go API and not worth changing.
 		klog.Warning("error creating inClusterConfig, falling back to default config: ", err)
 	}
 	return NewNonInteractiveDeferredLoadingClientConfig(
@@ -492,6 +492,7 @@ func getConfigFromFile(filename string) (*clientcmdapi.Config, error) {
 func GetConfigFromFileOrDie(filename string) *clientcmdapi.Config {
 	config, err := getConfigFromFile(filename)
 	if err != nil {
+		//nolint:logcheck // A helper function like this should not log. But this is probably part of the the established client-go API and not worth changing.
 		klog.FatalDepth(1, err)
 	}

@@ -137,6 +137,7 @@ type WarningHandler func(error)

 func (handler WarningHandler) Warn(err error) {
 	if handler == nil {
+		//nolint:logcheck // This is the fallback when logging is not initialized. With nothing provided, using the global logger is the only option.
 		klog.V(1).Info(err)
 	} else {
 		handler(err)
@@ -402,6 +403,7 @@ func LoadFromFile(filename string) (*clientcmdapi.Config, error) {
 	if err != nil {
 		return nil, err
 	}
+	//nolint:logcheck // A helper function like this should not log. But this is probably part of the the established client-go API and not worth changing.
 	klog.V(6).Infoln("Config loaded from file: ", filename)

 	// set LocationOfOrigin on every Cluster, User, and Context
@@ -118,6 +118,7 @@ func (config *DeferredLoadingClientConfig) ClientConfig() (*restclient.Config, e

 	// check for in-cluster configuration and use it
 	if config.icc.Possible() {
+		//nolint:logcheck // A helper function like this should not log. But this is probably part of the the established client-go API and not worth changing.
 		klog.V(4).Infof("Using in-cluster configuration")
 		return config.icc.ClientConfig()
 	}
@@ -160,6 +161,7 @@ func (config *DeferredLoadingClientConfig) Namespace() (string, bool, error) {
 		}
 	}

+	//nolint:logcheck // A helper function like this should not log. But this is probably part of the the established client-go API and not worth changing.
 	klog.V(4).Infof("Using in-cluster namespace")

 	// allow the namespace from the service account token directory to be used.
@@ -80,11 +80,29 @@ type TransportCacheMetric interface {
 }

 // TransportCreateCallsMetric counts the number of times a transport is created
-// partitioned by the result of the cache: hit, miss, uncacheable
+// partitioned by the result of the cache: hit, miss, miss-gc, uncacheable
 type TransportCreateCallsMetric interface {
 	Increment(result string)
 }

+// TransportCAReloadsMetric counts the number of times a CA reload is attempted,
+// partitioned by the result and reason.
+type TransportCAReloadsMetric interface {
+	Increment(result, reason string)
+}
+
+// TransportCertRotationGCCallsMetric counts the number of times a cert rotation
+// goroutine cancel func is called via GC cleanup.
+type TransportCertRotationGCCallsMetric interface {
+	Increment()
+}
+
+// TransportCacheGCCallsMetric counts the number of times a GC cleanup
+// attempts to delete a cache entry, partitioned by the result: deleted, skipped.
+type TransportCacheGCCallsMetric interface {
+	Increment(result string)
+}
+
 var (
 	// ClientCertExpiry is the expiry time of a client certificate
 	ClientCertExpiry ExpiryMetric = noopExpiry{}
@@ -117,23 +135,34 @@ var (
 	// TransportCreateCalls is the metric that counts the number of times a new transport
 	// is created
 	TransportCreateCalls TransportCreateCallsMetric = noopTransportCreateCalls{}
+	// TransportCAReloads is the metric that counts the number of times a CA reload is attempted
+	TransportCAReloads TransportCAReloadsMetric = noopTransportCAReloads{}
+	// TransportCertRotationGCCalls counts the number of times a cert rotation goroutine
+	// cancel func is called via GC cleanup
+	TransportCertRotationGCCalls TransportCertRotationGCCallsMetric = noopTransportCertRotationGCCalls{}
+	// TransportCacheGCCalls counts the number of times a GC cleanup attempts
+	// to delete a transport cache entry, partitioned by result: deleted, skipped.
+	TransportCacheGCCalls TransportCacheGCCallsMetric = noopTransportCacheGCCalls{}
 )

 // RegisterOpts contains all the metrics to register. Metrics may be nil.
 type RegisterOpts struct {
-	ClientCertExpiry      ExpiryMetric
-	ClientCertRotationAge DurationMetric
-	RequestLatency        LatencyMetric
-	ResolverLatency       ResolverLatencyMetric
-	RequestSize           SizeMetric
-	ResponseSize          SizeMetric
-	RateLimiterLatency    LatencyMetric
-	RequestResult         ResultMetric
-	ExecPluginCalls       CallsMetric
-	ExecPluginPolicyCalls PolicyCallsMetric
-	RequestRetry          RetryMetric
-	TransportCacheEntries TransportCacheMetric
-	TransportCreateCalls  TransportCreateCallsMetric
+	ClientCertExpiry             ExpiryMetric
+	ClientCertRotationAge        DurationMetric
+	RequestLatency               LatencyMetric
+	ResolverLatency              ResolverLatencyMetric
+	RequestSize                  SizeMetric
+	ResponseSize                 SizeMetric
+	RateLimiterLatency           LatencyMetric
+	RequestResult                ResultMetric
+	ExecPluginCalls              CallsMetric
+	ExecPluginPolicyCalls        PolicyCallsMetric
+	RequestRetry                 RetryMetric
+	TransportCacheEntries        TransportCacheMetric
+	TransportCreateCalls         TransportCreateCallsMetric
+	TransportCAReloads           TransportCAReloadsMetric
+	TransportCertRotationGCCalls TransportCertRotationGCCallsMetric
+	TransportCacheGCCalls        TransportCacheGCCallsMetric
 }

 // Register registers metrics for the rest client to use. This can
@@ -179,6 +208,15 @@ func Register(opts RegisterOpts) {
 		if opts.TransportCreateCalls != nil {
 			TransportCreateCalls = opts.TransportCreateCalls
 		}
+		if opts.TransportCAReloads != nil {
+			TransportCAReloads = opts.TransportCAReloads
+		}
+		if opts.TransportCertRotationGCCalls != nil {
+			TransportCertRotationGCCalls = opts.TransportCertRotationGCCalls
+		}
+		if opts.TransportCacheGCCalls != nil {
+			TransportCacheGCCalls = opts.TransportCacheGCCalls
+		}
 	})
 }

@@ -226,3 +264,15 @@ func (noopTransportCache) Observe(int) {}
 type noopTransportCreateCalls struct{}

 func (noopTransportCreateCalls) Increment(string) {}
+
+type noopTransportCAReloads struct{}
+
+func (noopTransportCAReloads) Increment(result, reason string) {}
+
+type noopTransportCertRotationGCCalls struct{}
+
+func (noopTransportCertRotationGCCalls) Increment() {}
+
+type noopTransportCacheGCCalls struct{}
+
+func (noopTransportCacheGCCalls) Increment(string) {}
@@ -21,6 +21,7 @@ import (
 	"io"

 	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/klog/v2"
 )

 // errorStreamDecoder interprets the data on the error channel and creates a go error object from it.
@@ -32,11 +33,11 @@ type errorStreamDecoder interface {
 // decodes it with the given errorStreamDecoder, sends the decoded error (or nil if the remote
 // command exited successfully) to the returned error channel, and closes it.
 // This function returns immediately.
-func watchErrorStream(errorStream io.Reader, d errorStreamDecoder) chan error {
+func watchErrorStream(logger klog.Logger, errorStream io.Reader, d errorStreamDecoder) chan error {
 	errorChan := make(chan error)

 	go func() {
-		defer runtime.HandleCrash()
+		defer runtime.HandleCrashWithLogger(logger)

 		message, err := io.ReadAll(errorStream)
 		switch {
@@ -53,7 +53,7 @@ func (f *FallbackExecutor) Stream(options StreamOptions) error {
 func (f *FallbackExecutor) StreamWithContext(ctx context.Context, options StreamOptions) error {
 	err := f.primary.StreamWithContext(ctx, options)
 	if err != nil && f.shouldFallback(err) {
-		klog.V(4).Infof("RemoteCommand fallback: %v", err)
+		klog.FromContext(ctx).V(4).Info("RemoteCommand fallback", "err", err)
 		return f.secondary.StreamWithContext(ctx, options)
 	}
 	return err
@@ -21,7 +21,8 @@ import (
 	"io"
 	"net/http"

-	"k8s.io/apimachinery/pkg/util/httpstream"
+	"k8s.io/klog/v2"
+	"k8s.io/streaming/pkg/httpstream"
 )

 // StreamOptions holds information pertaining to the current streaming session:
@@ -54,5 +55,5 @@ type streamCreator interface {
 }

 type streamProtocolHandler interface {
-	stream(conn streamCreator, ready chan<- struct{}) error
+	stream(logger klog.Logger, conn streamCreator, ready chan<- struct{}) error
 }
@@ -22,11 +22,11 @@ import (
 	"net/http"
 	"net/url"

-	"k8s.io/apimachinery/pkg/util/httpstream"
 	"k8s.io/apimachinery/pkg/util/remotecommand"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/transport/spdy"
 	"k8s.io/klog/v2"
+	"k8s.io/streaming/pkg/httpstream"
 )

 // spdyStreamExecutor handles transporting standard shell streams over an httpstream connection.
@@ -109,7 +109,7 @@ func (e *spdyStreamExecutor) newConnectionAndStream(ctx context.Context, options
 			return fmt.Errorf("redirect not allowed")
 		}
 	}
-	conn, protocol, err := spdy.Negotiate(
+	conn, protocol, err := spdy.NegotiateStreaming(
 		e.upgrader,
 		&client,
 		req,
@@ -121,6 +121,7 @@ func (e *spdyStreamExecutor) newConnectionAndStream(ctx context.Context, options

 	var streamer streamProtocolHandler

+	logger := klog.FromContext(ctx)
 	switch protocol {
 	case remotecommand.StreamProtocolV5Name:
 		streamer = newStreamProtocolV5(options)
@@ -131,7 +132,7 @@ func (e *spdyStreamExecutor) newConnectionAndStream(ctx context.Context, options
 	case remotecommand.StreamProtocolV2Name:
 		streamer = newStreamProtocolV2(options)
 	case "":
-		klog.V(4).Infof("The server did not negotiate a streaming protocol version. Falling back to %s", remotecommand.StreamProtocolV1Name)
+		logger.V(4).Info("The server did not negotiate a streaming protocol version, falling back", "protocol", remotecommand.StreamProtocolV1Name)
 		fallthrough
 	case remotecommand.StreamProtocolV1Name:
 		streamer = newStreamProtocolV1(options)
@@ -161,7 +162,7 @@ func (e *spdyStreamExecutor) StreamWithContext(ctx context.Context, options Stre
 		// The SPDY executor does not need to synchronize stream creation, so we pass a nil
 		// ready channel. The underlying spdystream library handles stream multiplexing
 		// without a race condition.
-		errorChan <- streamer.stream(conn, nil)
+		errorChan <- streamer.stream(klog.FromContext(ctx), conn, nil)
 	}()

 	select {
@@ -21,9 +21,9 @@ import (
 	"io"
 	"net/http"

-	"k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/httpstream"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/klog/v2"
+	"k8s.io/streaming/pkg/httpstream"
 )

 // streamProtocolV1 implements the first version of the streaming exec & attach
@@ -47,15 +47,15 @@ func newStreamProtocolV1(options StreamOptions) streamProtocolHandler {
 	}
 }

-func (p *streamProtocolV1) stream(conn streamCreator, ready chan<- struct{}) error {
+func (p *streamProtocolV1) stream(logger klog.Logger, conn streamCreator, ready chan<- struct{}) error {
 	doneChan := make(chan struct{}, 2)
 	errorChan := make(chan error)

 	cp := func(s string, dst io.Writer, src io.Reader) {
-		klog.V(6).Infof("Copying %s", s)
-		defer klog.V(6).Infof("Done copying %s", s)
+		logger.V(6).Info("Copying", "data", s)
+		defer logger.V(6).Info("Done copying", "data", s)
 		if _, err := io.Copy(dst, src); err != nil && err != io.EOF {
-			klog.Errorf("Error copying %s: %v", s, err)
+			logger.Error(err, "Error copying", "data", s)
 		}
 		if s == v1.StreamTypeStdout || s == v1.StreamTypeStderr {
 			doneChan <- struct{}{}
@@ -22,8 +22,9 @@ import (
 	"net/http"
 	"sync"

-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/klog/v2"
 )

 // streamProtocolV2 implements version 2 of the streaming protocol for attach
@@ -87,13 +88,13 @@ func (p *streamProtocolV2) createStreams(conn streamCreator) error {
 	return nil
 }

-func (p *streamProtocolV2) copyStdin() {
+func (p *streamProtocolV2) copyStdin(logger klog.Logger) {
 	if p.Stdin != nil {
 		var once sync.Once

 		// copy from client's stdin to container's stdin
 		go func() {
-			defer runtime.HandleCrash()
+			defer runtime.HandleCrashWithLogger(logger)

 			// if p.stdin is noninteractive, p.g. `echo abc | kubectl exec -i <pod> -- cat`, make sure
 			// we close remoteStdin as soon as the copy from p.stdin to remoteStdin finishes. Otherwise
@@ -101,7 +102,7 @@ func (p *streamProtocolV2) copyStdin() {
 			defer once.Do(func() { p.remoteStdin.Close() })

 			if _, err := io.Copy(p.remoteStdin, readerWrapper{p.Stdin}); err != nil {
-				runtime.HandleError(err)
+				runtime.HandleErrorWithLogger(logger, err, "Copying stdin failed")
 			}
 		}()

@@ -120,26 +121,26 @@ func (p *streamProtocolV2) copyStdin() {
 		// When that happens, we must Close() on our side of remoteStdin, to
 		// allow the copy in hijack to complete, and hijack to return.
 		go func() {
-			defer runtime.HandleCrash()
+			defer runtime.HandleCrashWithLogger(logger)
 			defer once.Do(func() { p.remoteStdin.Close() })

 			// this "copy" doesn't actually read anything - it's just here to wait for
 			// the server to close remoteStdin.
 			if _, err := io.Copy(io.Discard, p.remoteStdin); err != nil {
-				runtime.HandleError(err)
+				runtime.HandleErrorWithLogger(logger, err, "Waiting for server to close stdin failed")
 			}
 		}()
 	}
 }

-func (p *streamProtocolV2) copyStdout(wg *sync.WaitGroup) {
+func (p *streamProtocolV2) copyStdout(logger klog.Logger, wg *sync.WaitGroup) {
 	if p.Stdout == nil {
 		return
 	}

 	wg.Add(1)
 	go func() {
-		defer runtime.HandleCrash()
+		defer runtime.HandleCrashWithLogger(logger)
 		defer wg.Done()
 		// make sure, packet in queue can be consumed.
 		// block in queue may lead to deadlock in conn.server
@@ -147,29 +148,29 @@ func (p *streamProtocolV2) copyStdout(wg *sync.WaitGroup) {
 		defer io.Copy(io.Discard, p.remoteStdout)

 		if _, err := io.Copy(p.Stdout, p.remoteStdout); err != nil {
-			runtime.HandleError(err)
+			runtime.HandleErrorWithLogger(logger, err, "Copying stdout failed")
 		}
 	}()
 }

-func (p *streamProtocolV2) copyStderr(wg *sync.WaitGroup) {
+func (p *streamProtocolV2) copyStderr(logger klog.Logger, wg *sync.WaitGroup) {
 	if p.Stderr == nil || p.Tty {
 		return
 	}

 	wg.Add(1)
 	go func() {
-		defer runtime.HandleCrash()
+		defer runtime.HandleCrashWithLogger(logger)
 		defer wg.Done()
 		defer io.Copy(io.Discard, p.remoteStderr)

 		if _, err := io.Copy(p.Stderr, p.remoteStderr); err != nil {
-			runtime.HandleError(err)
+			runtime.HandleErrorWithLogger(logger, err, "Copying stderr failed")
 		}
 	}()
 }

-func (p *streamProtocolV2) stream(conn streamCreator, ready chan<- struct{}) error {
+func (p *streamProtocolV2) stream(logger klog.Logger, conn streamCreator, ready chan<- struct{}) error {
 	if err := p.createStreams(conn); err != nil {
 		return err
 	}
@@ -181,13 +182,13 @@ func (p *streamProtocolV2) stream(conn streamCreator, ready chan<- struct{}) err

 	// now that all the streams have been created, proceed with reading & copying

-	errorChan := watchErrorStream(p.errorStream, &errorDecoderV2{})
+	errorChan := watchErrorStream(logger, p.errorStream, &errorDecoderV2{})

-	p.copyStdin()
+	p.copyStdin(logger)

 	var wg sync.WaitGroup
-	p.copyStdout(&wg)
-	p.copyStderr(&wg)
+	p.copyStdout(logger, &wg)
+	p.copyStderr(logger, &wg)

 	// we're waiting for stdout/stderr to finish copying
 	wg.Wait()
@@ -22,8 +22,9 @@ import (
 	"net/http"
 	"sync"

-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/klog/v2"
 )

 // streamProtocolV3 implements version 3 of the streaming protocol for attach
@@ -62,12 +63,12 @@ func (p *streamProtocolV3) createStreams(conn streamCreator) error {
 	return nil
 }

-func (p *streamProtocolV3) handleResizes() {
+func (p *streamProtocolV3) handleResizes(logger klog.Logger) {
 	if p.resizeStream == nil || p.TerminalSizeQueue == nil {
 		return
 	}
 	go func() {
-		defer runtime.HandleCrash()
+		defer runtime.HandleCrashWithLogger(logger)

 		encoder := json.NewEncoder(p.resizeStream)
 		for {
@@ -76,13 +77,13 @@ func (p *streamProtocolV3) handleResizes() {
 				return
 			}
 			if err := encoder.Encode(&size); err != nil {
-				runtime.HandleError(err)
+				runtime.HandleErrorWithLogger(logger, err, "Encoding terminal size failed")
 			}
 		}
 	}()
 }

-func (p *streamProtocolV3) stream(conn streamCreator, ready chan<- struct{}) error {
+func (p *streamProtocolV3) stream(logger klog.Logger, conn streamCreator, ready chan<- struct{}) error {
 	if err := p.createStreams(conn); err != nil {
 		return err
 	}
@@ -94,15 +95,15 @@ func (p *streamProtocolV3) stream(conn streamCreator, ready chan<- struct{}) err

 	// now that all the streams have been created, proceed with reading & copying

-	errorChan := watchErrorStream(p.errorStream, &errorDecoderV3{})
+	errorChan := watchErrorStream(logger, p.errorStream, &errorDecoderV3{})

-	p.handleResizes()
+	p.handleResizes(logger)

-	p.copyStdin()
+	p.copyStdin(logger)

 	var wg sync.WaitGroup
-	p.copyStdout(&wg)
-	p.copyStderr(&wg)
+	p.copyStdout(logger, &wg)
+	p.copyStderr(logger, &wg)

 	// we're waiting for stdout/stderr to finish copying
 	wg.Wait()
@@ -26,6 +26,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/remotecommand"
 	"k8s.io/client-go/util/exec"
+	"k8s.io/klog/v2"
 )

 // streamProtocolV4 implements version 4 of the streaming protocol for attach
@@ -47,11 +48,11 @@ func (p *streamProtocolV4) createStreams(conn streamCreator) error {
 	return p.streamProtocolV3.createStreams(conn)
 }

-func (p *streamProtocolV4) handleResizes() {
-	p.streamProtocolV3.handleResizes()
+func (p *streamProtocolV4) handleResizes(logger klog.Logger) {
+	p.streamProtocolV3.handleResizes(logger)
 }

-func (p *streamProtocolV4) stream(conn streamCreator, ready chan<- struct{}) error {
+func (p *streamProtocolV4) stream(logger klog.Logger, conn streamCreator, ready chan<- struct{}) error {
 	if err := p.createStreams(conn); err != nil {
 		return err
 	}
@@ -63,15 +64,15 @@ func (p *streamProtocolV4) stream(conn streamCreator, ready chan<- struct{}) err

 	// now that all the streams have been created, proceed with reading & copying

-	errorChan := watchErrorStream(p.errorStream, &errorDecoderV4{})
+	errorChan := watchErrorStream(logger, p.errorStream, &errorDecoderV4{})

-	p.handleResizes()
+	p.handleResizes(logger)

-	p.copyStdin()
+	p.copyStdin(logger)

 	var wg sync.WaitGroup
-	p.copyStdout(&wg)
-	p.copyStderr(&wg)
+	p.copyStdout(logger, &wg)
+	p.copyStderr(logger, &wg)

 	// we're waiting for stdout/stderr to finish copying
 	wg.Wait()
@@ -16,6 +16,8 @@ limitations under the License.

 package remotecommand

+import "k8s.io/klog/v2"
+
 // streamProtocolV5 add support for V5 of the remote command subprotocol.
 // For the streamProtocolHandler, this version is the same as V4.
 type streamProtocolV5 struct {
@@ -30,6 +32,6 @@ func newStreamProtocolV5(options StreamOptions) streamProtocolHandler {
 	}
 }

-func (p *streamProtocolV5) stream(conn streamCreator, ready chan<- struct{}) error {
-	return p.streamProtocolV4.stream(conn, ready)
+func (p *streamProtocolV5) stream(logger klog.Logger, conn streamCreator, ready chan<- struct{}) error {
+	return p.streamProtocolV4.stream(logger, conn, ready)
 }
@@ -29,11 +29,11 @@ import (
 	gwebsocket "github.com/gorilla/websocket"

 	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/httpstream"
 	"k8s.io/apimachinery/pkg/util/remotecommand"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/transport/websocket"
 	"k8s.io/klog/v2"
+	"k8s.io/streaming/pkg/httpstream"
 )

 // writeDeadline defines the time that a client-side write to the websocket
@@ -130,7 +130,8 @@ func (e *wsStreamExecutor) StreamWithContext(ctx context.Context, options Stream
 	}
 	defer conn.Close()
 	e.negotiated = conn.Subprotocol()
-	klog.V(4).Infof("The subprotocol is %s", e.negotiated)
+	logger := klog.FromContext(ctx)
+	logger.V(4).Info("Subprotocol negotiated", "protocol", e.negotiated)

 	var streamer streamProtocolHandler
 	switch e.negotiated {
@@ -143,7 +144,7 @@ func (e *wsStreamExecutor) StreamWithContext(ctx context.Context, options Stream
 	case remotecommand.StreamProtocolV2Name:
 		streamer = newStreamProtocolV2(options)
 	case "":
-		klog.V(4).Infof("The server did not negotiate a streaming protocol version. Falling back to %s", remotecommand.StreamProtocolV1Name)
+		logger.V(4).Info("The server did not negotiate a streaming protocol version, falling back", "protocol", remotecommand.StreamProtocolV1Name)
 		fallthrough
 	case remotecommand.StreamProtocolV1Name:
 		streamer = newStreamProtocolV1(options)
@@ -159,7 +160,7 @@ func (e *wsStreamExecutor) StreamWithContext(ctx context.Context, options Stream
 		}()

 		readyChan := make(chan struct{})
-		creator := newWSStreamCreator(conn)
+		creator := newWSStreamCreator(logger, conn)
 		go func() {
 			select {
 			// Wait until all streams have been created before starting the readDemuxLoop.
@@ -177,7 +178,7 @@ func (e *wsStreamExecutor) StreamWithContext(ctx context.Context, options Stream
 				e.heartbeatDeadline,
 			)
 		}()
-		errorChan <- streamer.stream(creator, readyChan)
+		errorChan <- streamer.stream(logger, creator, readyChan)
 	}()

 	select {
@@ -191,7 +192,8 @@ func (e *wsStreamExecutor) StreamWithContext(ctx context.Context, options Stream
 }

 type wsStreamCreator struct {
-	conn *gwebsocket.Conn
+	logger klog.Logger
+	conn   *gwebsocket.Conn
 	// Protects writing to websocket connection; reading is lock-free
 	connWriteLock sync.Mutex
 	// map of stream id to stream; multiple streams read/write the connection
@@ -202,8 +204,9 @@ type wsStreamCreator struct {
 	setStreamErr error
 }

-func newWSStreamCreator(conn *gwebsocket.Conn) *wsStreamCreator {
+func newWSStreamCreator(logger klog.Logger, conn *gwebsocket.Conn) *wsStreamCreator {
 	return &wsStreamCreator{
+		logger:  logger,
 		conn:    conn,
 		streams: map[byte]*stream{},
 	}
@@ -238,6 +241,7 @@ func (c *wsStreamCreator) CreateStream(headers http.Header) (httpstream.Stream,
 	}
 	reader, writer := io.Pipe()
 	s := &stream{
+		logger:        klog.LoggerWithValues(c.logger, "id", id),
 		headers:       headers,
 		readPipe:      reader,
 		writePipe:     writer,
@@ -260,11 +264,11 @@ func (c *wsStreamCreator) CreateStream(headers http.Header) (httpstream.Stream,
 // connection reader at a time (a read mutex would provide no benefit).
 func (c *wsStreamCreator) readDemuxLoop(bufferSize int, period time.Duration, deadline time.Duration) {
 	// Initialize and start the ping/pong heartbeat.
-	h := newHeartbeat(c.conn, period, deadline)
+	h := newHeartbeat(c.logger, c.conn, period, deadline)
 	// Set initial timeout for websocket connection reading.
-	klog.V(5).Infof("Websocket initial read deadline: %s", deadline)
+	c.logger.V(5).Info("Websocket read starts", "deadline", deadline)
 	if err := c.conn.SetReadDeadline(time.Now().Add(deadline)); err != nil {
-		klog.Errorf("Websocket initial setting read deadline failed %v", err)
+		c.logger.Error(err, "Websocket initial setting read deadline failed")
 		return
 	}
 	go h.start()
@@ -308,7 +312,7 @@ func (c *wsStreamCreator) readDemuxLoop(bufferSize int, period time.Duration, de
 		streamID := readBuffer[0]
 		s := c.getStream(streamID)
 		if s == nil {
-			klog.Errorf("Unknown stream id %d, discarding message", streamID)
+			c.logger.Error(nil, "Unknown stream, discarding message", "id", streamID)
 			continue
 		}
 		for {
@@ -351,6 +355,7 @@ func (c *wsStreamCreator) closeAllStreamReaders(err error) {
 }

 type stream struct {
+	logger    klog.Logger
 	headers   http.Header
 	readPipe  *io.PipeReader
 	writePipe *io.PipeWriter
@@ -369,8 +374,8 @@ func (s *stream) Read(p []byte) (n int, err error) {

 // Write writes directly to the underlying WebSocket connection.
 func (s *stream) Write(p []byte) (n int, err error) {
-	klog.V(8).Infof("Write() on stream %d", s.id)
-	defer klog.V(8).Infof("Write() done on stream %d", s.id)
+	s.logger.V(8).Info("Write() on stream")
+	defer s.logger.V(8).Info("Write() done on stream")
 	s.connWriteLock.Lock()
 	defer s.connWriteLock.Unlock()
 	if s.conn == nil {
@@ -378,7 +383,7 @@ func (s *stream) Write(p []byte) (n int, err error) {
 	}
 	err = s.conn.SetWriteDeadline(time.Now().Add(writeDeadline))
 	if err != nil {
-		klog.V(4).Infof("Websocket setting write deadline failed %v", err)
+		s.logger.V(4).Info("Websocket setting write deadline failed", "err", err)
 		return 0, err
 	}
 	// Message writer buffers the message data, so we don't need to do that ourselves.
@@ -407,8 +412,8 @@ func (s *stream) Write(p []byte) (n int, err error) {

 // Close half-closes the stream, indicating this side is finished with the stream.
 func (s *stream) Close() error {
-	klog.V(6).Infof("Close() on stream %d", s.id)
-	defer klog.V(6).Infof("Close() done on stream %d", s.id)
+	s.logger.V(6).Info("Close() on stream")
+	defer s.logger.V(6).Info("Close() done on stream")
 	s.connWriteLock.Lock()
 	defer s.connWriteLock.Unlock()
 	if s.conn == nil {
@@ -421,8 +426,8 @@ func (s *stream) Close() error {
 }

 func (s *stream) Reset() error {
-	klog.V(4).Infof("Reset() on stream %d", s.id)
-	defer klog.V(4).Infof("Reset() done on stream %d", s.id)
+	s.logger.V(4).Info("Reset() on stream")
+	defer s.logger.V(4).Info("Reset() done on stream")
 	s.Close()
 	return s.writePipe.Close()
 }
@@ -442,7 +447,8 @@ func (s *stream) Identifier() uint32 {
 // inside the "readDemuxLoop" will return an i/o error prompting a connection close
 // and cleanup.
 type heartbeat struct {
-	conn *gwebsocket.Conn
+	logger klog.Logger
+	conn   *gwebsocket.Conn
 	// period defines how often a "ping" heartbeat message is sent to the other endpoint
 	period time.Duration
 	// closing the "closer" channel will clean up the heartbeat timers
@@ -456,8 +462,9 @@ type heartbeat struct {
 // newHeartbeat creates heartbeat structure encapsulating fields necessary to
 // run the websocket connection ping/pong mechanism and sets up handlers on
 // the websocket connection.
-func newHeartbeat(conn *gwebsocket.Conn, period time.Duration, deadline time.Duration) *heartbeat {
+func newHeartbeat(logger klog.Logger, conn *gwebsocket.Conn, period time.Duration, deadline time.Duration) *heartbeat {
 	h := &heartbeat{
+		logger: logger,
 		conn:   conn,
 		period: period,
 		closer: make(chan struct{}),
@@ -467,10 +474,10 @@ func newHeartbeat(conn *gwebsocket.Conn, period time.Duration, deadline time.Dur
 	// be empty.
 	h.conn.SetPongHandler(func(msg string) error {
 		// Push the read deadline into the future.
-		klog.V(6).Infof("Pong message received (%s)--resetting read deadline", msg)
+		logger.V(6).Info("Pong message received -- resetting read deadline", "message", msg)
 		err := h.conn.SetReadDeadline(time.Now().Add(deadline))
 		if err != nil {
-			klog.Errorf("Websocket setting read deadline failed %v", err)
+			logger.Error(err, "Websocket setting read deadline failed")
 			return err
 		}
 		if len(msg) > 0 {
@@ -502,16 +509,16 @@ func (h *heartbeat) start() {
 	for {
 		select {
 		case <-h.closer:
-			klog.V(5).Infof("closed channel--returning")
+			h.logger.V(5).Info("Closed channel -- returning")
 			return
 		case <-t.C:
 			// "WriteControl" does not need to be protected by a mutex. According to
 			// gorilla/websockets library docs: "The Close and WriteControl methods can
 			// be called concurrently with all other methods."
 			if err := h.conn.WriteControl(gwebsocket.PingMessage, h.message, time.Now().Add(pingReadDeadline)); err == nil {
-				klog.V(6).Infof("Websocket Ping succeeeded")
+				h.logger.V(6).Info("Websocket Ping succeeeded")
 			} else {
-				klog.Errorf("Websocket Ping failed: %v", err)
+				h.logger.Error(err, "Websocket Ping failed")
 				if errors.Is(err, gwebsocket.ErrCloseSent) {
 					// we continue because c.conn.CloseChan will manage closing the connection already
 					continue
@@ -0,0 +1,154 @@
+/*
+Copyright The Kubernetes 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 transport
+
+import (
+	"bytes"
+	"context"
+	"net/http"
+	"os"
+	"sync"
+	"time"
+
+	utilnet "k8s.io/apimachinery/pkg/util/net"
+	"k8s.io/client-go/tools/metrics"
+	"k8s.io/klog/v2"
+	"k8s.io/utils/clock"
+)
+
+var _ utilnet.RoundTripperWrapper = &atomicTransportHolder{}
+
+// atomicTransportHolder holds a transport that can be atomically updated
+// when CA files change, enabling graceful CA rotation without cache complexity
+type atomicTransportHolder struct {
+	caFile        string
+	currentCAData []byte // Track the actual CA data currently in use
+	// clock and caRefreshDuration are used to allow for testing time-based logic.
+	clock             clock.Clock
+	caRefreshDuration time.Duration
+	// mu covers transport and transportLastChecked
+	mu                   sync.RWMutex
+	transport            *http.Transport
+	transportLastChecked time.Time
+}
+
+func (h *atomicTransportHolder) RoundTrip(req *http.Request) (*http.Response, error) {
+	return h.getTransport(req.Context()).RoundTrip(req)
+}
+
+func (h *atomicTransportHolder) WrappedRoundTripper() http.RoundTripper {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+
+	return h.transport
+}
+
+func (h *atomicTransportHolder) getTransport(ctx context.Context) *http.Transport {
+	if rt := h.getTransportIfFresh(); rt != nil {
+		return rt
+	}
+
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	h.tryRefreshTransportLocked(ctx)
+	return h.transport
+}
+
+func (h *atomicTransportHolder) getTransportIfFresh() *http.Transport {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+
+	if h.clock.Since(h.transportLastChecked) < h.caRefreshDuration {
+		return h.transport
+	}
+	return nil
+}
+
+func (h *atomicTransportHolder) tryRefreshTransportLocked(ctx context.Context) {
+	// If some other goroutine already checked/updated the CA
+	if h.clock.Since(h.transportLastChecked) < h.caRefreshDuration {
+		return
+	}
+
+	// only attempt CA reload once per caRefreshDuration, even if the reload fails
+	h.transportLastChecked = h.clock.Now()
+
+	logger := klog.FromContext(ctx).WithValues("caFile", h.caFile)
+
+	logger.V(4).Info("Checking CA file content")
+
+	// Load new CA data from file
+	newCAData, err := os.ReadFile(h.caFile)
+	// Return old transport on read error
+	if err != nil {
+		logger.Error(err, "Failed to read CA data from file")
+		metrics.TransportCAReloads.Increment("failure", "read_error")
+		return
+	}
+
+	if len(newCAData) == 0 {
+		logger.Info("CA file empty, skipping transport rotation")
+		metrics.TransportCAReloads.Increment("failure", "empty")
+		return
+	}
+
+	if bytes.Equal(h.currentCAData, newCAData) {
+		logger.V(4).Info("CA file unchanged, skipping transport rotation")
+		metrics.TransportCAReloads.Increment("success", "unchanged")
+		return
+	}
+
+	logger.V(4).Info("CA content changed, updating transport")
+
+	// Load new CA pool
+	newCAs, err := rootCertPool(newCAData)
+	// Return old transport on parse error
+	if err != nil {
+		logger.Error(err, "Failed to parse CA data from file")
+		metrics.TransportCAReloads.Increment("failure", "ca_parse_error")
+		return
+	}
+	newTransport := h.transport.Clone()
+	newTransport.TLSClientConfig.RootCAs = newCAs
+	oldTransport := h.transport
+	h.transport = newTransport
+	// Update our tracking of current CA data
+	h.currentCAData = newCAData
+
+	// Close idle connections on the old transport to encourage migration
+	oldTransport.CloseIdleConnections()
+
+	logger.V(4).Info("Transport updated for CA rotation")
+	metrics.TransportCAReloads.Increment("success", "updated")
+}
+
+// newAtomicTransportHolder creates a new holder for CA file reloading scenarios.
+// The caFile must be specified.
+// caData may be empty but should correspond to the contents of caFile.
+// transport must have a TLS config and its root CAs should match caData.
+func newAtomicTransportHolder(caFile string, caData []byte, transport *http.Transport) *atomicTransportHolder {
+	c := clock.RealClock{}
+	return &atomicTransportHolder{
+		caFile:               caFile,
+		currentCAData:        caData,
+		clock:                c,
+		caRefreshDuration:    5 * time.Minute,
+		transport:            transport,
+		transportLastChecked: c.Now(),
+	}
+}
@@ -21,12 +21,14 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"runtime"
 	"strings"
 	"sync"
 	"time"
+	"weak"

 	utilnet "k8s.io/apimachinery/pkg/util/net"
-	"k8s.io/apimachinery/pkg/util/wait"
+	clientgofeaturegate "k8s.io/client-go/features"
 	"k8s.io/client-go/tools/metrics"
 	"k8s.io/klog/v2"
 )
@@ -35,22 +37,26 @@ import (
 // same RoundTripper will be returned for configs with identical TLS options If
 // the config has no custom TLS options, http.DefaultTransport is returned.
 type tlsTransportCache struct {
-	mu         sync.Mutex
-	transports map[tlsCacheKey]*http.Transport
+	mu               sync.Mutex
+	transports       map[tlsCacheKey]weak.Pointer[trackedTransport] // GC-enabled
+	strongTransports map[tlsCacheKey]http.RoundTripper              // GC-disabled
 }

-// DialerStopCh is stop channel that is passed down to dynamic cert dialer.
-// It's exposed as variable for testing purposes to avoid testing for goroutine
-// leakages.
-var DialerStopCh = wait.NeverStop
-
 const idleConnsPerHost = 25

-var tlsCache = &tlsTransportCache{transports: make(map[tlsCacheKey]*http.Transport)}
+var tlsCache = newTLSCache()
+
+func newTLSCache() *tlsTransportCache {
+	return &tlsTransportCache{
+		transports:       make(map[tlsCacheKey]weak.Pointer[trackedTransport]),
+		strongTransports: make(map[tlsCacheKey]http.RoundTripper),
+	}
+}

 type tlsCacheKey struct {
 	insecure           bool
 	caData             string
+	caFile             string
 	certData           string
 	keyData            string `datapolicy:"security-key"`
 	certFile           string
@@ -68,8 +74,8 @@ func (t tlsCacheKey) String() string {
 	if len(t.keyData) > 0 {
 		keyText = "<redacted>"
 	}
-	return fmt.Sprintf("insecure:%v, caData:%#v, certData:%#v, keyData:%s, serverName:%s, disableCompression:%t, getCert:%p, dial:%p",
-		t.insecure, t.caData, t.certData, keyText, t.serverName, t.disableCompression, t.getCert, t.dial)
+	return fmt.Sprintf("insecure:%v, caData:%#v, caFile:%s, certData:%#v, keyData:%s, serverName:%s, disableCompression:%t, getCert:%p, dial:%p",
+		t.insecure, t.caData, t.caFile, t.certData, keyText, t.serverName, t.disableCompression, t.getCert, t.dial)
 }

 func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
@@ -82,14 +88,18 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
 		// Ensure we only create a single transport for the given TLS options
 		c.mu.Lock()
 		defer c.mu.Unlock()
-		defer metrics.TransportCacheEntries.Observe(len(c.transports))
+		defer func() { metrics.TransportCacheEntries.Observe(c.lenLocked()) }()

 		// See if we already have a custom transport for this config
-		if t, ok := c.transports[key]; ok {
-			metrics.TransportCreateCalls.Increment("hit")
-			return t, nil
+		if t, ok := c.getLocked(key); ok {
+			if t != nil {
+				metrics.TransportCreateCalls.Increment("hit")
+				return t, nil
+			}
+			metrics.TransportCreateCalls.Increment("miss-gc")
+		} else {
+			metrics.TransportCreateCalls.Increment("miss")
 		}
-		metrics.TransportCreateCalls.Increment("miss")
 	} else {
 		metrics.TransportCreateCalls.Increment("uncacheable")
 	}
@@ -116,6 +126,7 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {

 	// If we use are reloading files, we need to handle certificate rotation properly
 	// TODO(jackkleeman): We can also add rotation here when config.HasCertCallback() is true
+	var cancel context.CancelFunc
 	if config.TLS.ReloadTLSFiles && tlsConfig != nil && tlsConfig.GetClientCertificate != nil {
 		// The TLS cache is a singleton, so sharing the same name for all of its
 		// background activity seems okay.
@@ -123,7 +134,9 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
 		dynamicCertDialer := certRotatingDialer(logger, tlsConfig.GetClientCertificate, dial)
 		tlsConfig.GetClientCertificate = dynamicCertDialer.GetClientCertificate
 		dial = dynamicCertDialer.connDialer.DialContext
-		go dynamicCertDialer.run(DialerStopCh)
+		var ctx context.Context
+		ctx, cancel = context.WithCancel(context.Background())
+		go dynamicCertDialer.run(ctx.Done())
 	}

 	proxy := http.ProxyFromEnvironment
@@ -131,7 +144,7 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
 		proxy = config.Proxy
 	}

-	transport := utilnet.SetTransportDefaults(&http.Transport{
+	httpTransport := utilnet.SetTransportDefaults(&http.Transport{
 		Proxy:               proxy,
 		TLSHandshakeTimeout: 10 * time.Second,
 		TLSClientConfig:     tlsConfig,
@@ -139,13 +152,101 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
 		DialContext:         dial,
 		DisableCompression:  config.DisableCompression,
 	})
+	var transport http.RoundTripper = httpTransport

-	if canCache {
-		// Cache a single transport for these options
-		c.transports[key] = transport
+	if config.TLS.ReloadCAFiles && tlsConfig != nil && tlsConfig.RootCAs != nil && len(config.TLS.CAFile) > 0 {
+		transport = newAtomicTransportHolder(config.TLS.CAFile, config.TLS.CAData, httpTransport)
 	}

-	return transport, nil
+	if !canCache && cancel == nil {
+		return transport, nil // uncacheable config with no cert rotation - nothing to GC
+	}
+
+	if !clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.ClientsAllowTLSCacheGC) {
+		if canCache {
+			c.strongTransports[key] = transport
+		}
+		return transport, nil // cancel is intentionally discarded and the cert rotation go routine leaks
+	}
+
+	transportWithGC := &trackedTransport{rt: transport}
+
+	if cancel != nil {
+		// capture metric as local var so that cleanups do not influence other tests via globals
+		transportCertRotationGCCalls := metrics.TransportCertRotationGCCalls
+		runtime.AddCleanup(transportWithGC, func(_ struct{}) {
+			cancel()
+			transportCertRotationGCCalls.Increment()
+		}, struct{}{})
+	}
+
+	if canCache {
+		wp := weak.Make(transportWithGC)
+		c.transports[key] = wp
+		// capture metrics as local vars so that cleanups do not influence other tests via globals
+		transportCacheGCCalls := metrics.TransportCacheGCCalls
+		transportCacheEntries := metrics.TransportCacheEntries
+		runtime.AddCleanup(transportWithGC, func(key tlsCacheKey) {
+			c.mu.Lock()
+			defer c.mu.Unlock()
+
+			// make sure we only delete the weak pointer created by this specific setLocked call
+			if c.transports[key] != wp {
+				transportCacheGCCalls.Increment("skipped")
+				return
+			}
+			delete(c.transports, key)
+			transportCacheGCCalls.Increment("deleted")
+			transportCacheEntries.Observe(c.lenLocked())
+		}, key)
+	}
+
+	return transportWithGC, nil
+}
+
+func (c *tlsTransportCache) getLocked(key tlsCacheKey) (http.RoundTripper, bool) {
+	if !clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.ClientsAllowTLSCacheGC) {
+		v, ok := c.strongTransports[key]
+		return v, ok
+	}
+
+	wp, ok := c.transports[key]
+	if !ok {
+		return nil, false
+	}
+
+	v := wp.Value()
+
+	if v == nil { // avoid typed nil
+		return nil, true // key exists but value has been garbage collected
+	}
+
+	return v, true
+}
+
+func (c *tlsTransportCache) lenLocked() int {
+	if !clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.ClientsAllowTLSCacheGC) {
+		return len(c.strongTransports)
+	}
+	return len(c.transports)
+}
+
+// trackedTransport wraps an http.RoundTripper to serve as the weak.Pointer
+// target in the TLS transport cache. Dropping all references to this object
+// triggers GC cleanup of the cache entry and any cert rotation goroutine.
+type trackedTransport struct {
+	rt http.RoundTripper
+}
+
+var _ http.RoundTripper = &trackedTransport{}
+var _ utilnet.RoundTripperWrapper = &trackedTransport{}
+
+func (v *trackedTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	return v.rt.RoundTrip(req)
+}
+
+func (v *trackedTransport) WrappedRoundTripper() http.RoundTripper {
+	return v.rt
 }

 // tlsConfigKey returns a unique key for tls.Config objects returned from TLSConfigFor
@@ -162,7 +263,6 @@ func tlsConfigKey(c *Config) (tlsCacheKey, bool, error) {

 	k := tlsCacheKey{
 		insecure:           c.TLS.Insecure,
-		caData:             string(c.TLS.CAData),
 		serverName:         c.TLS.ServerName,
 		nextProtos:         strings.Join(c.TLS.NextProtos, ","),
 		disableCompression: c.DisableCompression,
@@ -178,5 +278,14 @@ func tlsConfigKey(c *Config) (tlsCacheKey, bool, error) {
 		k.keyData = string(c.TLS.KeyData)
 	}

+	if c.TLS.ReloadCAFiles {
+		// When reloading CA files, include CA file path in cache key instead of CA data
+		// This allows the CA to be reloaded from disk on each transport creation
+		k.caFile = c.TLS.CAFile
+	} else {
+		// When not reloading, cache the CA data directly
+		k.caData = string(c.TLS.CAData)
+	}
+
 	return k, true, nil
 }
@@ -134,7 +134,8 @@ type TLSConfig struct {
 	CAFile         string // Path of the PEM-encoded server trusted root certificates.
 	CertFile       string // Path of the PEM-encoded client certificate.
 	KeyFile        string // Path of the PEM-encoded client key.
-	ReloadTLSFiles bool   // Set to indicate that the original config provided files, and that they should be reloaded
+	ReloadTLSFiles bool   // Set to indicate that the original config provided files, and that they should be reloaded.
+	ReloadCAFiles  bool   // Set to indicate that CA files should be reloaded from disk.

 	Insecure   bool   // Server should be accessed without verifying the certificate. For testing only.
 	ServerName string // Override for the server name passed to the server for SNI and used to verify certificates.
@@ -319,7 +319,7 @@ func (rt *bearerAuthRoundTripper) RoundTrip(req *http.Request) (*http.Response,
 			token = refreshedToken.AccessToken
 		}
 	}
-	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
+	req.Header.Set("Authorization", "Bearer "+token)
 	return rt.rt.RoundTrip(req)
 }

@@ -23,8 +23,10 @@ import (
 	"time"

 	"k8s.io/apimachinery/pkg/util/httpstream"
-	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
+	httpstreamspdy "k8s.io/apimachinery/pkg/util/httpstream/spdy"
 	restclient "k8s.io/client-go/rest"
+	"k8s.io/klog/v2"
+	streamhttp "k8s.io/streaming/pkg/httpstream"
 )

 // Upgrader validates a response from the server after a SPDY upgrade.
@@ -43,7 +45,7 @@ func RoundTripperFor(config *restclient.Config) (http.RoundTripper, Upgrader, er
 	if config.Proxy != nil {
 		proxy = config.Proxy
 	}
-	upgradeRoundTripper, err := spdy.NewRoundTripperWithConfig(spdy.RoundTripperConfig{
+	upgradeRoundTripper, err := httpstreamspdy.NewRoundTripperWithConfig(httpstreamspdy.RoundTripperConfig{
 		TLS:              tlsConfig,
 		Proxier:          proxy,
 		PingPeriod:       time.Second * 5,
@@ -79,6 +81,18 @@ func NewDialer(upgrader Upgrader, client *http.Client, method string, url *url.U
 	}
 }

+// NewDialerForStreaming creates a SPDY dialer for in-tree callers that use
+// k8s.io/streaming/pkg/httpstream types.
+func NewDialerForStreaming(upgrader Upgrader, client *http.Client, method string, url *url.URL) streamhttp.Dialer {
+	return &streamingDialerAdapter{delegate: NewDialer(upgrader, client, method, url)}
+}
+
+// NewUpgraderForStreaming adapts a streaming upgrader for callers that need
+// the compatibility Upgrader interface.
+func NewUpgraderForStreaming(upgrader streamhttp.UpgradeRoundTripper) Upgrader {
+	return &compatUpgraderAdapter{delegate: upgrader}
+}
+
 func (d *dialer) Dial(protocols ...string) (httpstream.Connection, string, error) {
 	req, err := http.NewRequest(d.method, d.url.String(), nil)
 	if err != nil {
@@ -105,3 +119,199 @@ func Negotiate(upgrader Upgrader, client *http.Client, req *http.Request, protoc
 	}
 	return conn, resp.Header.Get(httpstream.HeaderProtocolVersion), nil
 }
+
+// NegotiateStreaming is for in-tree callers that still operate on
+// k8s.io/streaming/pkg/httpstream types.
+func NegotiateStreaming(upgrader Upgrader, client *http.Client, req *http.Request, protocols ...string) (streamhttp.Connection, string, error) {
+	conn, protocol, err := Negotiate(upgrader, client, req, protocols...)
+	if err != nil {
+		return nil, "", err
+	}
+	return wrapStreamingConnection(conn), protocol, nil
+}
+
+type streamingDialerAdapter struct {
+	delegate httpstream.Dialer
+}
+
+func (d *streamingDialerAdapter) Dial(protocols ...string) (streamhttp.Connection, string, error) {
+	conn, protocol, err := d.delegate.Dial(protocols...)
+	if err != nil {
+		return nil, "", err
+	}
+	return wrapStreamingConnection(conn), protocol, nil
+}
+
+type compatUpgraderAdapter struct {
+	delegate streamhttp.UpgradeRoundTripper
+}
+
+func (u *compatUpgraderAdapter) NewConnection(resp *http.Response) (httpstream.Connection, error) {
+	conn, err := u.delegate.NewConnection(resp)
+	if err != nil {
+		return nil, err
+	}
+	return wrapCompatConnection(conn), nil
+}
+
+type streamingStreamAdapter struct {
+	delegate httpstream.Stream
+}
+
+func (s *streamingStreamAdapter) Read(p []byte) (int, error) {
+	return s.delegate.Read(p)
+}
+
+func (s *streamingStreamAdapter) Write(p []byte) (int, error) {
+	return s.delegate.Write(p)
+}
+
+func (s *streamingStreamAdapter) Close() error {
+	return s.delegate.Close()
+}
+
+func (s *streamingStreamAdapter) Reset() error {
+	return s.delegate.Reset()
+}
+
+func (s *streamingStreamAdapter) Headers() http.Header {
+	return s.delegate.Headers()
+}
+
+func (s *streamingStreamAdapter) Identifier() uint32 {
+	return s.delegate.Identifier()
+}
+
+type streamingConnectionAdapter struct {
+	delegate httpstream.Connection
+}
+
+func (c *streamingConnectionAdapter) CreateStream(headers http.Header) (streamhttp.Stream, error) {
+	stream, err := c.delegate.CreateStream(headers)
+	if err != nil {
+		return nil, err
+	}
+	return &streamingStreamAdapter{delegate: stream}, nil
+}
+
+func (c *streamingConnectionAdapter) Close() error {
+	return c.delegate.Close()
+}
+
+func (c *streamingConnectionAdapter) CloseChan() <-chan bool {
+	return c.delegate.CloseChan()
+}
+
+func (c *streamingConnectionAdapter) SetIdleTimeout(timeout time.Duration) {
+	c.delegate.SetIdleTimeout(timeout)
+}
+
+func (c *streamingConnectionAdapter) RemoveStreams(streams ...streamhttp.Stream) {
+	compatStreams := make([]httpstream.Stream, 0, len(streams))
+	for _, stream := range streams {
+		if stream == nil {
+			continue
+		}
+		if s, ok := stream.(*streamingStreamAdapter); ok {
+			compatStreams = append(compatStreams, s.delegate)
+			continue
+		}
+		if s, ok := stream.(httpstream.Stream); ok {
+			compatStreams = append(compatStreams, s)
+			continue
+		}
+		klog.V(5).Infof("dropping unadaptable streaming stream %T in RemoveStreams", stream)
+	}
+	c.delegate.RemoveStreams(compatStreams...)
+}
+
+func wrapStreamingConnection(conn httpstream.Connection) streamhttp.Connection {
+	if conn == nil {
+		return nil
+	}
+	if wrapped, ok := conn.(*compatConnectionAdapter); ok {
+		return wrapped.delegate
+	}
+	return &streamingConnectionAdapter{delegate: conn}
+}
+
+type compatStreamAdapter struct {
+	delegate streamhttp.Stream
+}
+
+func (s *compatStreamAdapter) Read(p []byte) (int, error) {
+	return s.delegate.Read(p)
+}
+
+func (s *compatStreamAdapter) Write(p []byte) (int, error) {
+	return s.delegate.Write(p)
+}
+
+func (s *compatStreamAdapter) Close() error {
+	return s.delegate.Close()
+}
+
+func (s *compatStreamAdapter) Reset() error {
+	return s.delegate.Reset()
+}
+
+func (s *compatStreamAdapter) Headers() http.Header {
+	return s.delegate.Headers()
+}
+
+func (s *compatStreamAdapter) Identifier() uint32 {
+	return s.delegate.Identifier()
+}
+
+type compatConnectionAdapter struct {
+	delegate streamhttp.Connection
+}
+
+func (c *compatConnectionAdapter) CreateStream(headers http.Header) (httpstream.Stream, error) {
+	stream, err := c.delegate.CreateStream(headers)
+	if err != nil {
+		return nil, err
+	}
+	return &compatStreamAdapter{delegate: stream}, nil
+}
+
+func (c *compatConnectionAdapter) Close() error {
+	return c.delegate.Close()
+}
+
+func (c *compatConnectionAdapter) CloseChan() <-chan bool {
+	return c.delegate.CloseChan()
+}
+
+func (c *compatConnectionAdapter) SetIdleTimeout(timeout time.Duration) {
+	c.delegate.SetIdleTimeout(timeout)
+}
+
+func (c *compatConnectionAdapter) RemoveStreams(streams ...httpstream.Stream) {
+	streamingStreams := make([]streamhttp.Stream, 0, len(streams))
+	for _, stream := range streams {
+		if stream == nil {
+			continue
+		}
+		if s, ok := stream.(*compatStreamAdapter); ok {
+			streamingStreams = append(streamingStreams, s.delegate)
+			continue
+		}
+		if s, ok := stream.(streamhttp.Stream); ok {
+			streamingStreams = append(streamingStreams, s)
+			continue
+		}
+		klog.V(5).Infof("dropping unadaptable compat stream %T in RemoveStreams", stream)
+	}
+	c.delegate.RemoveStreams(streamingStreams...)
+}
+
+func wrapCompatConnection(conn streamhttp.Connection) httpstream.Connection {
+	if conn == nil {
+		return nil
+	}
+	if wrapped, ok := conn.(*streamingConnectionAdapter); ok {
+		return wrapped.delegate
+	}
+	return &compatConnectionAdapter{delegate: conn}
+}
@@ -28,6 +28,7 @@ import (
 	"time"

 	utilnet "k8s.io/apimachinery/pkg/util/net"
+	clientgofeaturegate "k8s.io/client-go/features"
 	"k8s.io/klog/v2"
 )

@@ -211,17 +212,26 @@ func TLSConfigFor(c *Config) (*tls.Config, error) {
 // KeyData, and CAFile fields, or returns an error. If no error is returned, all three fields are
 // either populated or were empty to start.
 func loadTLSFiles(c *Config) error {
+	// Check that we are purely loading CA from file
+	if clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.ClientsAllowCARotation) {
+		if len(c.TLS.CAFile) > 0 && len(c.TLS.CAData) == 0 {
+			c.TLS.ReloadCAFiles = true
+		}
+	} else if c.TLS.ReloadCAFiles {
+		return fmt.Errorf("ReloadCAFiles=true requires ClientsAllowCARotation to be enabled")
+	}
+
+	// Check that we are purely loading certs and keys from files
+	if len(c.TLS.CertFile) > 0 && len(c.TLS.CertData) == 0 && len(c.TLS.KeyFile) > 0 && len(c.TLS.KeyData) == 0 {
+		c.TLS.ReloadTLSFiles = true
+	}
+
 	var err error
 	c.TLS.CAData, err = dataFromSliceOrFile(c.TLS.CAData, c.TLS.CAFile)
 	if err != nil {
 		return err
 	}

-	// Check that we are purely loading from files
-	if len(c.TLS.CertFile) > 0 && len(c.TLS.CertData) == 0 && len(c.TLS.KeyFile) > 0 && len(c.TLS.KeyData) == 0 {
-		c.TLS.ReloadTLSFiles = true
-	}
-
 	c.TLS.CertData, err = dataFromSliceOrFile(c.TLS.CertData, c.TLS.CertFile)
 	if err != nil {
 		return err
@@ -254,6 +264,11 @@ func rootCertPool(caData []byte) (*x509.CertPool, error) {
 	// code for a look at the platform specific insanity), so we'll use the fact that RootCAs == nil gives us the system values
 	// It doesn't allow trusting either/or, but hopefully that won't be an issue
 	if len(caData) == 0 {
+		// When the ClientsAllowCARotation feature gate is enabled, it returns an empty but non-nil pool.
+		// This ensures we don't fall back to system roots when a user explicitly points CAFile to a zero-byte file.
+		if clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.ClientsAllowCARotation) {
+			return x509.NewCertPool(), nil
+		}
 		return nil, nil
 	}

@@ -31,11 +31,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
-	"k8s.io/apimachinery/pkg/util/httpstream"
-	"k8s.io/apimachinery/pkg/util/httpstream/wsstream"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/transport"
+	"k8s.io/streaming/pkg/httpstream"
+	"k8s.io/streaming/pkg/httpstream/wsstream"
 )

 var (