From 39560700da10121501f6d1777a03bee64e388010 Mon Sep 17 00:00:00 2001 From: Babak Salamat Date: Fri, 20 Feb 2026 00:53:47 +0000 Subject: [PATCH] 1. Optionally Enforce TLS Validation: modified the created http.Transport to configure CAData. The `Insecure` field is now conditionally set using apiService spec. 2. Invalidate Bad Connections: when the availability check receives an unexpected response from the configured endpoints, a timeout, or network dial/reachability error, the controller now forces the cached connections inside the HTTP transport to be dropped. This avoids scenarios where a connection is reused when moving or recreating pods behind a node but without reaching kube-proxy properly. 3. Tests Added: Added tests to assert the overall flow behaves normally and expectedly for such situations as we now close connections. --- .../pkg/apiserver/handler_proxy.go | 12 +-- .../remote/remote_available_controller.go | 39 ++++++---- .../remote_available_controller_test.go | 73 +++++++++++++++++++ 3 files changed, 101 insertions(+), 23 deletions(-) diff --git a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy.go b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy.go index d95a271af76..ecf1800e05b 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy.go +++ b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy.go @@ -37,6 +37,7 @@ import ( "k8s.io/klog/v2" apiregistrationv1api "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1" apiregistrationv1apihelper "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1/helper" + "k8s.io/kube-aggregator/pkg/controllers/status/remote" ) const ( @@ -219,16 +220,7 @@ func (r *proxyHandler) updateAPIService(apiService *apiregistrationv1api.APIServ proxyClientCert, proxyClientKey := r.proxyCurrentCertKeyContent() - transportConfig := &transport.Config{ - TLS: transport.TLSConfig{ - Insecure: apiService.Spec.InsecureSkipTLSVerify, - ServerName: apiService.Spec.Service.Name + "." + apiService.Spec.Service.Namespace + ".svc", - CertData: proxyClientCert, - KeyData: proxyClientKey, - CAData: apiService.Spec.CABundle, - }, - DialHolder: r.proxyTransportDial, - } + transportConfig := remote.BuildTransportConfig(r.proxyTransportDial, proxyClientCert, proxyClientKey, apiService) transportConfig.Wrap(x509metrics.NewDeprecatedCertificateRoundTripperWrapperConstructor( x509MissingSANCounter, x509InsecureSHA1Counter, diff --git a/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller.go b/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller.go index 22e78b3b9d8..c31ec33254b 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller.go +++ b/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller.go @@ -32,6 +32,7 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + utilnet "k8s.io/apimachinery/pkg/util/net" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apiserver/pkg/util/proxy" @@ -157,6 +158,25 @@ func New( return c, nil } +// BuildTransportConfig builds a transport.Config for an APIService. +// It ignores TLS verification if InsecureSkipTLSVerify is true. +func BuildTransportConfig( + proxyTransportDial *transport.DialHolder, + proxyClientCert, proxyClientKey []byte, + apiService *apiregistrationv1.APIService, +) *transport.Config { + return &transport.Config{ + TLS: transport.TLSConfig{ + Insecure: apiService.Spec.InsecureSkipTLSVerify, + ServerName: apiService.Spec.Service.Name + "." + apiService.Spec.Service.Namespace + ".svc", + CertData: proxyClientCert, + KeyData: proxyClientKey, + CAData: apiService.Spec.CABundle, + }, + DialHolder: proxyTransportDial, + } +} + func (c *AvailableConditionController) sync(key string) error { originalAPIService, err := c.apiServiceLister.Get(key) if apierrors.IsNotFound(err) { @@ -247,21 +267,12 @@ func (c *AvailableConditionController) sync(key string) error { // actually try to hit the discovery endpoint when it isn't local and when we're routing as a service. if apiService.Spec.Service != nil && c.serviceResolver != nil { // if a particular transport was specified, use that otherwise build one - // construct an http client that will ignore TLS verification (if someone owns the network and messes with your status - // that's not so bad) and sets a very short timeout. This is a best effort GET that provides no additional information - transportConfig := &transport.Config{ - TLS: transport.TLSConfig{ - Insecure: true, - }, - DialHolder: c.proxyTransportDial, - } - + var proxyClientCert, proxyClientKey []byte if c.proxyCurrentCertKeyContent != nil { - proxyClientCert, proxyClientKey := c.proxyCurrentCertKeyContent() - - transportConfig.TLS.CertData = proxyClientCert - transportConfig.TLS.KeyData = proxyClientKey + proxyClientCert, proxyClientKey = c.proxyCurrentCertKeyContent() } + + transportConfig := BuildTransportConfig(c.proxyTransportDial, proxyClientCert, proxyClientKey, apiService) restTransport, err := transport.New(transportConfig) if err != nil { return err @@ -318,6 +329,7 @@ func (c *AvailableConditionController) sync(key string) error { select { case err = <-errCh: if err != nil { + utilnet.CloseIdleConnectionsFor(restTransport) results <- fmt.Errorf("failing or missing response from %v: %w", discoveryURL, err) return } @@ -325,6 +337,7 @@ func (c *AvailableConditionController) sync(key string) error { // we had trouble with slow dial and DNS responses causing us to wait too long. // we added this as insurance case <-time.After(6 * time.Second): + utilnet.CloseIdleConnectionsFor(restTransport) results <- fmt.Errorf("timed out waiting for %v", discoveryURL) return } diff --git a/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller_test.go b/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller_test.go index 20e62975dac..d0faad7a709 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller_test.go +++ b/staging/src/k8s.io/kube-aggregator/pkg/controllers/status/remote/remote_available_controller_test.go @@ -18,6 +18,7 @@ package remote import ( "fmt" + "net" "net/http" "net/http/httptest" "net/url" @@ -537,3 +538,75 @@ func TestUpdateAPIServiceStatus(t *testing.T) { func emptyCert() []byte { return []byte{} } + +func TestCloseIdleConnections(t *testing.T) { + apiServiceName := "remote.group" + apiServices := []runtime.Object{newRemoteAPIService(apiServiceName)} + services := []*v1.Service{newService("foo", "bar", testServicePort, testServicePortName)} + endpointSlices := []*discoveryv1.EndpointSlice{newEndpointSliceWithAddress("foo", "bar", testServicePort, testServicePortName)} + + fakeClient := fake.NewSimpleClientset(apiServices...) + apiServiceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + serviceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + endpointSliceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + for _, obj := range apiServices { + if err := apiServiceIndexer.Add(obj); err != nil { + t.Fatalf("failed to add APIService: %v", err) + } + } + for _, obj := range services { + if err := serviceIndexer.Add(obj); err != nil { + t.Fatalf("failed to add service: %v", err) + } + } + for _, obj := range endpointSlices { + if err := endpointSliceIndexer.Add(obj); err != nil { + t.Fatalf("failed to add endpointSlice: %v", err) + } + } + + endpointSliceGetter, err := proxy.NewEndpointSliceListerGetter(discoveryv1listers.NewEndpointSliceLister(endpointSliceIndexer)) + if err != nil { + t.Fatalf("error creating endpointSliceGetter: %v", err) + } + + closed := make(chan struct{}, 1) + testServer := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadGateway) + })) + // We want to verify that the client closes the connection. + // httptest.Server doesn't expose ConnState, but the underlying http.Server does. + testServer.Config.ConnState = func(c net.Conn, state http.ConnState) { + if state == http.StateClosed { + select { + case closed <- struct{}{}: + default: + } + } + } + testServer.Start() + defer testServer.Close() + + c := AvailableConditionController{ + apiServiceClient: fakeClient.ApiregistrationV1(), + apiServiceLister: listers.NewAPIServiceLister(apiServiceIndexer), + serviceLister: v1listers.NewServiceLister(serviceIndexer), + endpointSliceGetter: endpointSliceGetter, + serviceResolver: &fakeServiceResolver{url: testServer.URL}, + proxyCurrentCertKeyContent: func() ([]byte, []byte) { return emptyCert(), emptyCert() }, + metrics: availabilitymetrics.New(), + } + + // This should trigger the bad gateway response and (with the fix) close the connection. + err = c.sync(apiServiceName) + if err == nil { + t.Fatal("expected error from sync") + } + + select { + case <-closed: + // success, connection was closed + case <-time.After(30 * time.Second): + t.Fatal("timeout waiting for connection to be closed") + } +}