From 0f1a5ead6b3c998bafb9d87d7833d1920fde28d7 Mon Sep 17 00:00:00 2001 From: zhuyulicfc49 Date: Wed, 24 Jun 2026 13:22:39 -0400 Subject: [PATCH] fix: push-mode informers pick up rotated bearer token without restart Signed-off-by: zhuyulicfc49 --- pkg/util/membercluster_client.go | 16 +- .../membercluster_client_rotation_test.go | 117 +++++++++ pkg/util/round_trippers.go | 93 +++++++ pkg/util/round_trippers_token_test.go | 164 ++++++++++++ test/e2e/suites/base/token_rotation_test.go | 239 ++++++++++++++++++ 5 files changed, 625 insertions(+), 4 deletions(-) create mode 100644 pkg/util/membercluster_client_rotation_test.go create mode 100644 pkg/util/round_trippers_token_test.go create mode 100644 test/e2e/suites/base/token_rotation_test.go diff --git a/pkg/util/membercluster_client.go b/pkg/util/membercluster_client.go index aa9b4f3933b7..99e268e005fd 100644 --- a/pkg/util/membercluster_client.go +++ b/pkg/util/membercluster_client.go @@ -216,11 +216,10 @@ func BuildClusterConfig(clusterName string, return nil, fmt.Errorf("the secret for cluster %s is missing a non-empty value for %q", clusterName, clusterv1alpha1.SecretTokenKey) } - // Initialize cluster configuration. clusterConfig := &rest.Config{ - BearerToken: string(token), - Host: apiEndpoint, - Timeout: defaultTimeout, + // BearerToken omitted: injected by the token-refreshing transport below. + Host: apiEndpoint, + Timeout: defaultTimeout, } // Handle TLS configuration. @@ -248,6 +247,15 @@ func BuildClusterConfig(clusterName string, } } + // Token-refreshing transport — replaces the built-in bearerAuthRoundTripper + // position (outermost) so inner wrappers see auth already set. + clusterConfig.Wrap(NewTokenRefreshingRoundTripperWrapperConstructor( + secretGetter, + cluster.Spec.SecretRef.Namespace, + cluster.Spec.SecretRef.Name, + string(token), + )) + return clusterConfig, nil } diff --git a/pkg/util/membercluster_client_rotation_test.go b/pkg/util/membercluster_client_rotation_test.go new file mode 100644 index 000000000000..524a9b183396 --- /dev/null +++ b/pkg/util/membercluster_client_rotation_test.go @@ -0,0 +1,117 @@ +/* +Copyright 2026 The Karmada 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 util + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + fakeclient "sigs.k8s.io/controller-runtime/pkg/client/fake" + + clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1" + "github.com/karmada-io/karmada/pkg/util/gclient" +) + +type rotatableTokenServer struct { + *httptest.Server + accepted atomic.Value // string + lastAuth atomic.Value // string +} + +func newRotatableTokenServer(initialToken string) *rotatableTokenServer { + s := &rotatableTokenServer{} + s.accepted.Store(initialToken) + s.lastAuth.Store("") + s.Server = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + auth := r.Header.Get("Authorization") + s.lastAuth.Store(auth) + if auth != "Bearer "+s.accepted.Load().(string) { + w.WriteHeader(http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"apiVersion":"v1","kind":"Node","metadata":{"name":"foo"}}`)) + })) + return s +} + +func (s *rotatableTokenServer) accept(token string) { s.accepted.Store(token) } +func (s *rotatableTokenServer) seenAuth() string { return s.lastAuth.Load().(string) } + +const ( + testTokenCacheTTL = 10 * time.Millisecond + tokenRotateTimeout = 100 * time.Millisecond + tokenRotatePollRate = 5 * time.Millisecond +) + +func TestBuildClusterConfig_LongLivedClientPicksUpRotatedToken(t *testing.T) { + origTTL := tokenCacheTTL + tokenCacheTTL = testTokenCacheTTL + t.Cleanup(func() { tokenCacheTTL = origTTL }) + + const clusterName = "member-rotate" + + srv := newRotatableTokenServer("token-A") + defer srv.Close() + + hostClient := fakeclient.NewClientBuilder().WithScheme(gclient.NewSchema()).WithObjects( + &clusterv1alpha1.Cluster{ + ObjectMeta: metav1.ObjectMeta{Name: clusterName}, + Spec: clusterv1alpha1.ClusterSpec{ + APIEndpoint: srv.URL, + SecretRef: &clusterv1alpha1.LocalSecretReference{Namespace: "ns1", Name: "secret1"}, + }, + }, + &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Namespace: "ns1", Name: "secret1"}, + Data: map[string][]byte{ + clusterv1alpha1.SecretTokenKey: []byte("token-A"), + clusterv1alpha1.SecretCADataKey: getCACertFromGTestServer(t, srv.Server), + }, + }, + ).Build() + + clusterClient, err := NewClusterClientSet(clusterName, hostClient, nil) + assert.NoError(t, err) + + _, err = clusterClient.KubeClient.CoreV1().Nodes().Get(context.TODO(), "foo", metav1.GetOptions{}) + assert.NoError(t, err, "baseline request with token-A should succeed") + + // Rotate: update Secret and reject old token on the server. + secret := &corev1.Secret{} + assert.NoError(t, hostClient.Get(context.TODO(), types.NamespacedName{Namespace: "ns1", Name: "secret1"}, secret)) + secret.Data[clusterv1alpha1.SecretTokenKey] = []byte("token-B") + assert.NoError(t, hostClient.Update(context.TODO(), secret)) + srv.accept("token-B") + + assert.Eventually(t, func() bool { + _, err := clusterClient.KubeClient.CoreV1().Nodes().Get(context.TODO(), "foo", metav1.GetOptions{}) + return err == nil + }, tokenRotateTimeout, tokenRotatePollRate, + "after rotation the long-lived client must pick up token-B without a rebuild") + + assert.Equal(t, "Bearer token-B", srv.seenAuth(), + "rotated token must appear on the wire") +} diff --git a/pkg/util/round_trippers.go b/pkg/util/round_trippers.go index 65bc758fe7fb..579b22f6a770 100644 --- a/pkg/util/round_trippers.go +++ b/pkg/util/round_trippers.go @@ -20,8 +20,14 @@ import ( "net/http" "net/textproto" "strings" + "sync" + "time" + corev1 "k8s.io/api/core/v1" "k8s.io/client-go/transport" + "k8s.io/klog/v2" + + clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1" ) type proxyHeaderRoundTripper struct { @@ -65,3 +71,90 @@ func parseProxyHeaders(headers map[string]string) http.Header { } return proxyHeaders } + +// tokenCacheTTL controls how often the token is re-read from the Secret. +var tokenCacheTTL = 30 * time.Second + +// tokenErrorRetryInterval is a shorter TTL used after a failed Secret read, +// so recovery happens faster than a full tokenCacheTTL cycle. +var tokenErrorRetryInterval = 5 * time.Second + +// tokenRefreshingRoundTripper re-reads the member cluster token from the Karmada +// Secret on a TTL so long-lived informer clients survive token rotation. +type tokenRefreshingRoundTripper struct { + inner http.RoundTripper + secretGetter func(namespace, name string) (*corev1.Secret, error) + secretNS string + secretName string + + mu sync.RWMutex + cachedToken string + cacheExpiry time.Time +} + +var _ http.RoundTripper = &tokenRefreshingRoundTripper{} + +// WrappedRoundTripper implements utilnet.RoundTripperWrapper. +func (t *tokenRefreshingRoundTripper) WrappedRoundTripper() http.RoundTripper { return t.inner } + +// NewTokenRefreshingRoundTripperWrapperConstructor returns a WrapperFunc that injects a TTL-refreshed bearer token. +// secretGetter is expected to be informer-backed (in-memory read, not a live API call). +func NewTokenRefreshingRoundTripperWrapperConstructor( + secretGetter func(string, string) (*corev1.Secret, error), + secretNS, secretName, initialToken string, +) transport.WrapperFunc { + return func(rt http.RoundTripper) http.RoundTripper { + return &tokenRefreshingRoundTripper{ + inner: rt, + secretGetter: secretGetter, + secretNS: secretNS, + secretName: secretName, + cachedToken: initialToken, + cacheExpiry: time.Now().Add(tokenCacheTTL), + } + } +} + +// RoundTrip implements the http.RoundTripper interface. +func (t *tokenRefreshingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + req = req.Clone(req.Context()) + req.Header.Set("Authorization", "Bearer "+t.getToken()) + return t.inner.RoundTrip(req) +} + +func (t *tokenRefreshingRoundTripper) getToken() string { + t.mu.RLock() + if time.Now().Before(t.cacheExpiry) { + token := t.cachedToken + t.mu.RUnlock() + return token + } + t.mu.RUnlock() + return t.forceRefresh() +} + +func (t *tokenRefreshingRoundTripper) forceRefresh() string { + t.mu.Lock() + defer t.mu.Unlock() + + if time.Now().Before(t.cacheExpiry) { + return t.cachedToken + } + + secret, err := t.secretGetter(t.secretNS, t.secretName) + if err != nil { + klog.Warningf("tokenRefreshingRoundTripper: failed to refresh token from secret %s/%s, keeping last known token: %v", + t.secretNS, t.secretName, err) + t.cacheExpiry = time.Now().Add(tokenErrorRetryInterval) + return t.cachedToken + } + + if token := string(secret.Data[clusterv1alpha1.SecretTokenKey]); token != "" { + t.cachedToken = token + } else { + klog.Warningf("tokenRefreshingRoundTripper: secret %s/%s has empty token key, keeping last known token", + t.secretNS, t.secretName) + } + t.cacheExpiry = time.Now().Add(tokenCacheTTL) + return t.cachedToken +} diff --git a/pkg/util/round_trippers_token_test.go b/pkg/util/round_trippers_token_test.go new file mode 100644 index 000000000000..4d23437cc62a --- /dev/null +++ b/pkg/util/round_trippers_token_test.go @@ -0,0 +1,164 @@ +/* +Copyright 2026 The Karmada 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 util + +import ( + "errors" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + corev1 "k8s.io/api/core/v1" + + clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1" +) + +type recordingRoundTripper struct { + mu sync.Mutex + lastReq *http.Request +} + +func (r *recordingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + r.mu.Lock() + r.lastReq = req + r.mu.Unlock() + return &http.Response{StatusCode: http.StatusOK, Body: http.NoBody}, nil +} + +func (r *recordingRoundTripper) authHeader() string { + r.mu.Lock() + defer r.mu.Unlock() + return r.lastReq.Header.Get("Authorization") +} + +func tokenSecret(token string) *corev1.Secret { + return &corev1.Secret{Data: map[string][]byte{clusterv1alpha1.SecretTokenKey: []byte(token)}} +} + +func newTokenRT(rec http.RoundTripper, getter func(string, string) (*corev1.Secret, error), initialToken string) *tokenRefreshingRoundTripper { + return NewTokenRefreshingRoundTripperWrapperConstructor(getter, "ns", "name", initialToken)(rec).(*tokenRefreshingRoundTripper) +} + +func doGet(t *testing.T, rt http.RoundTripper) { + t.Helper() + req, err := http.NewRequest(http.MethodGet, "https://example.test/api", nil) + assert.NoError(t, err) + _, err = rt.RoundTrip(req) + assert.NoError(t, err) +} + +func TestTokenRefreshingRoundTripper_InjectsToken(t *testing.T) { + rec := &recordingRoundTripper{} + getter := func(string, string) (*corev1.Secret, error) { return tokenSecret("token-A"), nil } + rt := newTokenRT(rec, getter, "token-A") + + req, err := http.NewRequest(http.MethodGet, "https://example.test/api", nil) + assert.NoError(t, err) + _, err = rt.RoundTrip(req) + assert.NoError(t, err) + + assert.Equal(t, "Bearer token-A", rec.authHeader(), "token must be injected into the request sent downstream") + assert.Empty(t, req.Header.Get("Authorization"), "the caller's original request must not be mutated") +} + +func TestTokenRefreshingRoundTripper_RefreshesAfterTTL(t *testing.T) { + rec := &recordingRoundTripper{} + var mu sync.Mutex + current := "token-A" + getter := func(string, string) (*corev1.Secret, error) { + mu.Lock() + defer mu.Unlock() + return tokenSecret(current), nil + } + rt := newTokenRT(rec, getter, "token-A") + + mu.Lock() + current = "token-B" + mu.Unlock() + rt.cacheExpiry = time.Now().Add(-time.Minute) + + doGet(t, rt) + assert.Equal(t, "Bearer token-B", rec.authHeader(), "rotated token must be picked up after the TTL") +} + +func TestTokenRefreshingRoundTripper_KeepsTokenOnSecretError(t *testing.T) { + rec := &recordingRoundTripper{} + getter := func(string, string) (*corev1.Secret, error) { return nil, errors.New("temporarily unavailable") } + rt := newTokenRT(rec, getter, "token-A") + rt.cacheExpiry = time.Now().Add(-time.Minute) + + doGet(t, rt) + assert.Equal(t, "Bearer token-A", rec.authHeader(), "on Secret read error the last good token must be retained") +} + +func TestTokenRefreshingRoundTripper_IgnoresEmptyToken(t *testing.T) { + rec := &recordingRoundTripper{} + getter := func(string, string) (*corev1.Secret, error) { return tokenSecret(""), nil } + rt := newTokenRT(rec, getter, "token-A") + rt.cacheExpiry = time.Now().Add(-time.Minute) + + doGet(t, rt) + assert.Equal(t, "Bearer token-A", rec.authHeader(), "an empty token from the Secret must not overwrite a good cached token") +} + +func TestTokenRefreshingRoundTripper_ChainsWithProxyHeaders(t *testing.T) { + var capturedReq *http.Request + inner := &http.Transport{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + capturedReq = r + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + getter := func(string, string) (*corev1.Secret, error) { return tokenSecret("my-token"), nil } + proxyWrapper := NewProxyHeaderRoundTripperWrapperConstructor(nil, map[string]string{"Proxy-Authorization": "Basic secret"}) + tokenWrapper := NewTokenRefreshingRoundTripperWrapperConstructor(getter, "ns", "name", "my-token") + chain := tokenWrapper(proxyWrapper(inner)) + + req, err := http.NewRequest(http.MethodGet, server.URL, nil) + assert.NoError(t, err) + _, err = chain.RoundTrip(req) + assert.NoError(t, err) + + assert.Equal(t, "Bearer my-token", capturedReq.Header.Get("Authorization"), + "token wrapper must set the Authorization header") +} + +func TestTokenRefreshingRoundTripper_ConcurrentRefreshDeduplicated(t *testing.T) { + rec := &recordingRoundTripper{} + var calls atomic.Int32 + getter := func(string, string) (*corev1.Secret, error) { + calls.Add(1) + return tokenSecret("token-B"), nil + } + rt := newTokenRT(rec, getter, "token-A") + rt.cacheExpiry = time.Now().Add(-time.Minute) + + var wg sync.WaitGroup + for range 50 { + wg.Go(func() { + doGet(t, rt) + }) + } + wg.Wait() + + assert.Equal(t, int32(1), calls.Load(), "the Secret must be read exactly once per TTL window despite concurrent requests") +} diff --git a/test/e2e/suites/base/token_rotation_test.go b/test/e2e/suites/base/token_rotation_test.go new file mode 100644 index 000000000000..7192e3d00cb5 --- /dev/null +++ b/test/e2e/suites/base/token_rotation_test.go @@ -0,0 +1,239 @@ +/* +Copyright 2026 The Karmada 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 base + +import ( + "context" + "encoding/base64" + "encoding/json" + "strings" + "time" + + "github.com/onsi/ginkgo/v2" + "github.com/onsi/gomega" + appsv1 "k8s.io/api/apps/v1" + authenticationv1 "k8s.io/api/authentication/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/rand" + "k8s.io/klog/v2" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + kindexec "sigs.k8s.io/kind/pkg/exec" + + clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1" + policyv1alpha1 "github.com/karmada-io/karmada/pkg/apis/policy/v1alpha1" + "github.com/karmada-io/karmada/test/e2e/framework" + testhelper "github.com/karmada-io/karmada/test/helper" +) + +const ( + // Time for the apiserver's SA token cache to flush after revocation. + revocationCacheFlushWait = 15 * time.Second + // Timeout for the cluster to return to Ready after container restart. + clusterReadyTimeout = 90 * time.Second + // Duration to assert the cluster remains Ready (short-lived path unaffected). + clusterStableWindow = 15 * time.Second + // Timeout for the informer to observe scaled status after rotation. + informerRecoveryTimeout = 120 * time.Second +) + +// Verifies that push-mode informers recover after a member cluster token rotation. +// Trade-offs to keep the test fast (~60s vs 10+ min for natural expiry): +// - Token revocation (instant) instead of waiting for natural token expiry. +// - Docker restart to force watch reconnection — an open watch survives +// revocation since the API server only re-validates auth on reconnection. +var _ = framework.SerialDescribe("push-mode token rotation", func() { + var ( + targetCluster string + deploymentNS string + deploymentName string + deployment *appsv1.Deployment + policy *policyv1alpha1.PropagationPolicy + ) + + ginkgo.BeforeEach(func() { + pushClusters := framework.ClusterNamesWithSyncMode(clusterv1alpha1.Push) + gomega.Expect(pushClusters).ShouldNot(gomega.BeEmpty(), "need at least one push-mode cluster") + targetCluster = pushClusters[0] + + deploymentNS = testNamespace + deploymentName = deploymentNamePrefix + rand.String(RandomStrLength) + deployment = testhelper.NewDeployment(deploymentNS, deploymentName) + + policy = testhelper.NewPropagationPolicy(deploymentNS, deploymentName, []policyv1alpha1.ResourceSelector{ + {APIVersion: deployment.APIVersion, Kind: deployment.Kind, Name: deployment.Name}, + }, policyv1alpha1.Placement{ + ClusterAffinity: &policyv1alpha1.ClusterAffinity{ + ClusterNames: []string{targetCluster}, + }, + }) + }) + + ginkgo.JustBeforeEach(func() { + framework.CreatePropagationPolicy(karmadaClient, policy) + framework.CreateDeployment(kubeClient, deployment) + ginkgo.DeferCleanup(func() { + framework.RemoveDeployment(kubeClient, deployment.Namespace, deployment.Name) + framework.RemovePropagationPolicy(karmadaClient, policy.Namespace, policy.Name) + }) + }) + + ginkgo.It("informers recover after member cluster token rotates", func() { + ginkgo.By("1. verify workload propagated and status collected (baseline)", func() { + framework.WaitDeploymentPresentOnClusterFitWith(targetCluster, deploymentNS, deploymentName, + func(d *appsv1.Deployment) bool { + return d.Status.ReadyReplicas > 0 + }) + klog.Infof("baseline: deployment %s/%s present on %s with ready replicas", deploymentNS, deploymentName, targetCluster) + }) + + cluster := &clusterv1alpha1.Cluster{} + gomega.Expect(controlPlaneClient.Get(context.TODO(), client.ObjectKey{Name: targetCluster}, cluster)).Should(gomega.Succeed()) + gomega.Expect(cluster.Spec.SecretRef).ShouldNot(gomega.BeNil()) + secretNS := cluster.Spec.SecretRef.Namespace + secretName := cluster.Spec.SecretRef.Name + + secret := &corev1.Secret{} + gomega.Expect(controlPlaneClient.Get(context.TODO(), client.ObjectKey{Namespace: secretNS, Name: secretName}, secret)).Should(gomega.Succeed()) + saNamespace, saName := parseSAFromToken(string(secret.Data[clusterv1alpha1.SecretTokenKey])) + gomega.Expect(saName).ShouldNot(gomega.BeEmpty(), "cluster token must be a parseable SA JWT") + klog.Infof("cluster %s uses SA %s/%s on the member", targetCluster, saNamespace, saName) + + memberClient := framework.GetClusterClient(targetCluster) + gomega.Expect(memberClient).ShouldNot(gomega.BeNil()) + + ginkgo.By("2. rotate: revoke old token, mint new, write to Secret", func() { + // Revoke first so the new token is bound to the recreated SA's UID. + gomega.Expect(memberClient.CoreV1().ServiceAccounts(saNamespace).Delete( + context.TODO(), saName, metav1.DeleteOptions{})).Should(gomega.Succeed()) + _, err := memberClient.CoreV1().ServiceAccounts(saNamespace).Create( + context.TODO(), &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{ + Name: saName, Namespace: saNamespace, + }}, metav1.CreateOptions{}) + gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) + klog.Infof("revoked old tokens by recreating SA %s/%s", saNamespace, saName) + + tokenReq := &authenticationv1.TokenRequest{ + Spec: authenticationv1.TokenRequestSpec{ + ExpirationSeconds: ptr.To[int64](86400), + }, + } + tokenResp, err := memberClient.CoreV1().ServiceAccounts(saNamespace).CreateToken( + context.TODO(), saName, tokenReq, metav1.CreateOptions{}) + gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) + newToken := tokenResp.Status.Token + + secret.Data[clusterv1alpha1.SecretTokenKey] = []byte(newToken) + gomega.Expect(controlPlaneClient.Update(context.TODO(), secret)).Should(gomega.Succeed()) + klog.Infof("rotated token in Secret %s/%s", secretNS, secretName) + + // The apiserver caches SA tokens for ~12s before revocation takes effect. + time.Sleep(revocationCacheFlushWait) + }) + + ginkgo.By("3. force watch reconnection (docker restart member node)", func() { + containerName := targetCluster + "-control-plane" + klog.Infof("restarting kind container %s to force watch reconnection", containerName) + + cmd := kindexec.Command("docker", "restart", containerName) + output, err := kindexec.CombinedOutputLines(cmd) + if err != nil { + klog.Warningf("docker restart failed (output: %v): %v — skipping e2e", output, err) + ginkgo.Skip("cannot force watch reconnection via docker restart — run manual test instead") + } + klog.Infof("container %s restarted", containerName) + + gomega.Eventually(func() bool { + c := &clusterv1alpha1.Cluster{} + if err := controlPlaneClient.Get(context.TODO(), client.ObjectKey{Name: targetCluster}, c); err != nil { + return false + } + for _, cond := range c.Status.Conditions { + if cond.Type == "Ready" && cond.Status == "True" { + return true + } + } + return false + }, clusterReadyTimeout, 2*time.Second).Should(gomega.BeTrue(), + "cluster must return to Ready after container restart") + }) + + ginkgo.By("4. verify short-lived path still works (cluster stays Ready)", func() { + gomega.Consistently(func() bool { + c := &clusterv1alpha1.Cluster{} + if err := controlPlaneClient.Get(context.TODO(), client.ObjectKey{Name: targetCluster}, c); err != nil { + return false + } + for _, cond := range c.Status.Conditions { + if cond.Type == "Ready" && cond.Status == "True" { + return true + } + } + return false + }, clusterStableWindow, pollInterval).Should(gomega.BeTrue(), + "cluster must remain Ready (short-lived health check unaffected)") + }) + + ginkgo.By("5. verify long-lived path recovered (informer sees member changes)", func() { + karmadaDep, err := kubeClient.AppsV1().Deployments(deploymentNS).Get( + context.TODO(), deploymentName, metav1.GetOptions{}) + gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) + targetReplicas := *karmadaDep.Spec.Replicas + 2 + karmadaDep.Spec.Replicas = &targetReplicas + _, err = kubeClient.AppsV1().Deployments(deploymentNS).Update( + context.TODO(), karmadaDep, metav1.UpdateOptions{}) + gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) + klog.Infof("scaled deployment to %d replicas", targetReplicas) + + gomega.Eventually(func() bool { + dep, err := kubeClient.AppsV1().Deployments(deploymentNS).Get( + context.TODO(), deploymentName, metav1.GetOptions{}) + if err != nil { + return false + } + klog.Infof("karmada collected readyReplicas=%d (target %d)", + dep.Status.ReadyReplicas, targetReplicas) + return dep.Status.ReadyReplicas == targetReplicas + }, informerRecoveryTimeout, pollInterval).Should(gomega.BeTrue(), + "informer must observe scaled status after token rotation") + }) + }) +}) + +func parseSAFromToken(token string) (namespace, name string) { + parts := strings.Split(token, ".") + if len(parts) < 2 { + return "", "" + } + payload, err := base64.RawURLEncoding.DecodeString(parts[1]) + if err != nil { + return "", "" + } + var claims struct { + Sub string `json:"sub"` + } + if err := json.Unmarshal(payload, &claims); err != nil { + return "", "" + } + // sub format: "system:serviceaccount::" + segments := strings.Split(claims.Sub, ":") + if len(segments) != 4 { + return "", "" + } + return segments[2], segments[3] +}