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 58c288b7406b2..98dd81140071a 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 @@ -243,7 +243,8 @@ func (r *proxyHandler) updateAPIService(apiService *apiregistrationv1api.APIServ servicePort: *apiService.Spec.Service.Port, serviceAvailable: apiregistrationv1apihelper.IsAPIServiceConditionTrue(apiService, apiregistrationv1api.Available), } - newInfo.proxyRoundTripper, newInfo.transportBuildingError = transport.New(newInfo.transportConfig) + // UPSTREAM: : use fast http2 health checking for aggregated API backend connections + newInfo.proxyRoundTripper, newInfo.transportBuildingError = newAggregatedAPIBackendRoundTripper(newInfo.transportConfig) if newInfo.transportBuildingError != nil { klog.Warning(newInfo.transportBuildingError.Error()) } diff --git a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/patch_backend_transport.go b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/patch_backend_transport.go new file mode 100644 index 0000000000000..94c765052b19e --- /dev/null +++ b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/patch_backend_transport.go @@ -0,0 +1,72 @@ +package apiserver + +import ( + "net/http" + "time" + + "golang.org/x/net/http2" + + "k8s.io/client-go/transport" + "k8s.io/klog/v2" +) + +// UPSTREAM: : fast http2 health checking for aggregated API backend connections +// +// The aggregator proxies requests to aggregated apiservers over pooled http2 +// connections. When such a connection is silently broken (for instance when it was +// established while the pod network on a freshly rebooted control plane node was +// still converging), the default health check parameters +// (ReadIdleTimeout=30s/PingTimeout=15s, see k8s.io/apimachinery/pkg/util/net) keep +// the dead connection pinned for up to ~45 seconds while every request multiplexed +// onto it fails with "http2: client connection lost". Observed as 10-15s of +// aggregated API disruption during upgrades (OCPBUGS-100065). +// +// Detect and drop dead backend connections within seconds instead. The values are +// intentionally aggressive: aggregated apiservers are same-cluster backends with +// sub-second round trips, so a connection that cannot answer a ping for a few +// seconds is broken for practical purposes and re-dialing is cheap. +const ( + aggregatedAPIBackendReadIdleTimeout = 5 * time.Second + aggregatedAPIBackendPingTimeout = 5 * time.Second +) + +// newAggregatedAPIBackendRoundTripper builds the round tripper used to proxy +// requests to an aggregated apiserver. It mirrors transport.New for the transport +// construction, but configures aggressive http2 connection health checking so that +// broken backend connections are abandoned within seconds. On any unexpected +// configuration it falls back to the default transport.New behavior. +func newAggregatedAPIBackendRoundTripper(cfg *transport.Config) (http.RoundTripper, error) { + tlsConfig, err := transport.TLSConfigFor(cfg) + if err != nil { + return nil, err + } + if tlsConfig == nil || cfg.Transport != nil { + // no TLS settings or a custom transport: nothing for us to tune, keep the default behavior + return transport.New(cfg) + } + + // mirror the transport constructed by client-go's transport.New/tlsCache.get + t := &http.Transport{ + Proxy: http.ProxyFromEnvironment, + TLSClientConfig: tlsConfig, + TLSHandshakeTimeout: 10 * time.Second, + MaxIdleConnsPerHost: 25, + IdleConnTimeout: 90 * time.Second, + } + if cfg.DialHolder != nil { + t.DialContext = cfg.DialHolder.Dial + } + + t2, err := http2.ConfigureTransports(t) + if err != nil { + // should not happen; fall back to the default construction rather than failing the APIService + klog.Warningf("failed to configure http2 health checking for aggregated API backend transport, falling back to defaults: %v", err) + return transport.New(cfg) + } + t2.ReadIdleTimeout = aggregatedAPIBackendReadIdleTimeout + t2.PingTimeout = aggregatedAPIBackendPingTimeout + + // apply the same wrappers (user agent, auth, WrapTransport such as the x509 + // metrics wrapper) that transport.New would apply + return transport.HTTPWrappersForConfig(cfg, t) +} diff --git a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/patch_backend_transport_test.go b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/patch_backend_transport_test.go new file mode 100644 index 0000000000000..038c66ccbff03 --- /dev/null +++ b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/patch_backend_transport_test.go @@ -0,0 +1,64 @@ +package apiserver + +import ( + "net/http" + "net/http/httptest" + "testing" + + "k8s.io/client-go/transport" +) + +func TestNewAggregatedAPIBackendRoundTripperServesRequests(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + rt, err := newAggregatedAPIBackendRoundTripper(&transport.Config{ + TLS: transport.TLSConfig{Insecure: true}, + }) + if err != nil { + t.Fatalf("unexpected error building round tripper: %v", err) + } + + req, err := http.NewRequest(http.MethodGet, server.URL, nil) + if err != nil { + t.Fatalf("unexpected error building request: %v", err) + } + resp, err := rt.RoundTrip(req) + if err != nil { + t.Fatalf("unexpected round trip error: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Errorf("expected 200, got %d", resp.StatusCode) + } +} + +func TestNewAggregatedAPIBackendRoundTripperFallsBackWithoutTLS(t *testing.T) { + // no TLS settings: must fall back to transport.New without error + rt, err := newAggregatedAPIBackendRoundTripper(&transport.Config{}) + if err != nil { + t.Fatalf("unexpected error building fallback round tripper: %v", err) + } + if rt == nil { + t.Fatalf("expected a round tripper") + } +} + +func TestNewAggregatedAPIBackendRoundTripperAppliesWrappers(t *testing.T) { + wrapped := false + cfg := &transport.Config{ + TLS: transport.TLSConfig{Insecure: true}, + } + cfg.Wrap(func(rt http.RoundTripper) http.RoundTripper { + wrapped = true + return rt + }) + if _, err := newAggregatedAPIBackendRoundTripper(cfg); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !wrapped { + t.Errorf("expected the config WrapTransport to be applied") + } +}