From 8125b28c466a3a74309c1c9122fbdf799dec5ac5 Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Fri, 4 Sep 2026 11:08:05 +0200 Subject: [PATCH] Placement shim: detect a lost remote apiserver and rebuild the manager Signed-off-by: Philipp Matthes --- cmd/shim/main.go | 27 +- .../multicluster/cortex-remote-crb.yaml | 3 + docs/guides/multicluster/run.sh | 2 +- .../templates/alerts.yaml | 27 ++ pkg/multicluster/client.go | 2 +- pkg/multicluster/client_probe.go | 220 ++++++++++++++++ pkg/multicluster/client_probe_test.go | 243 ++++++++++++++++++ pkg/multicluster/client_test.go | 13 + pkg/multicluster/monitor.go | 25 ++ pkg/multicluster/monitor_test.go | 21 ++ pkg/shim/supervisor/supervisor_test.go | 29 +++ 11 files changed, 607 insertions(+), 5 deletions(-) create mode 100644 pkg/multicluster/client_probe.go create mode 100644 pkg/multicluster/client_probe_test.go diff --git a/cmd/shim/main.go b/cmd/shim/main.go index 62ab7f64d..e3b7e2d37 100644 --- a/cmd/shim/main.go +++ b/cmd/shim/main.go @@ -403,11 +403,32 @@ func main() { return fmt.Errorf("unable to create manager: %w", err) } - multiclusterClient, err := setupMulticlusterClient(ctx, mgr, restConfig, multiclusterMonitor) + // cycleCtx scopes this manager cycle: cancelling it makes mgr.Start return + // so the supervisor rebuilds the manager (re-reading config and + // reconnecting to the currently-configured remotes). It is cancelled on a + // clean return (defer), when the parent ctx is cancelled (child), or by the + // reachability probe below when a remote apiserver is lost. This is the + // external liveness signal the supervisor lacks on its own: a remote whose + // informer already synced and then disappears is retried by + // controller-runtime forever without mgr.Start ever returning. + cycleCtx, cancelCycle := context.WithCancelCause(ctx) + defer cancelCycle(nil) + + multiclusterClient, err := setupMulticlusterClient(cycleCtx, mgr, restConfig, multiclusterMonitor) if err != nil { return fmt.Errorf("unable to set up multicluster client: %w", err) } + // Probe each remote apiserver for reachability. When a remote is + // sustained-unreachable (e.g. its cluster was deleted after its cache had + // already synced), cancel the cycle so the supervisor rebuilds the manager. + // A remote that answers with any HTTP status is reachable, so an authz or + // server error does not trigger a rebuild storm. + go multiclusterClient.ProbeRemotes(cycleCtx, multicluster.DefaultProbeOptions, func(host string) { + setupLog.Info("remote apiserver sustained-unreachable; recycling manager", "host", host) + cancelCycle(fmt.Errorf("remote apiserver unreachable: %s", host)) + }) + if placementShim != nil { if err := placementShim.SetupControllerWithManager(ctx, mgr, multiclusterClient); err != nil { return fmt.Errorf("unable to set up placement shim controller: %w", err) @@ -431,7 +452,7 @@ func main() { // goroutine only marks ready while the context is still live, and the // deferred clear always wins the final state. var readyMu sync.Mutex - cacheCtx, cancelCacheSync := context.WithCancel(ctx) + cacheCtx, cancelCacheSync := context.WithCancel(cycleCtx) defer func() { cancelCacheSync() readyMu.Lock() @@ -483,7 +504,7 @@ func main() { } setupLog.Info("starting manager") - return mgr.Start(ctx) + return mgr.Start(cycleCtx) } // +kubebuilder:scaffold:builder diff --git a/docs/guides/multicluster/cortex-remote-crb.yaml b/docs/guides/multicluster/cortex-remote-crb.yaml index 66b44ded8..61968e368 100644 --- a/docs/guides/multicluster/cortex-remote-crb.yaml +++ b/docs/guides/multicluster/cortex-remote-crb.yaml @@ -15,6 +15,9 @@ subjects: - kind: User apiGroup: rbac.authorization.k8s.io name: "https://host.docker.internal:8443#system:serviceaccount:default:cortex-pods-controller-manager" +- kind: User + apiGroup: rbac.authorization.k8s.io + name: "https://host.docker.internal:8443#system:serviceaccount:default:cortex-placement-shim" roleRef: kind: ClusterRole name: cluster-admin diff --git a/docs/guides/multicluster/run.sh b/docs/guides/multicluster/run.sh index f68956870..c4062b84d 100755 --- a/docs/guides/multicluster/run.sh +++ b/docs/guides/multicluster/run.sh @@ -78,4 +78,4 @@ kubectl --context kind-cortex-remote-az-b apply \ echo "Starting cortex in home cluster with tilt, using overrides from $TILT_OVERRIDES_PATH" kubectl config use-context kind-cortex-home -export ACTIVE_DEPLOYMENTS="nova" && tilt up +tilt up diff --git a/helm/bundles/cortex-placement-shim/templates/alerts.yaml b/helm/bundles/cortex-placement-shim/templates/alerts.yaml index ff5ac691b..f12c8e616 100644 --- a/helm/bundles/cortex-placement-shim/templates/alerts.yaml +++ b/helm/bundles/cortex-placement-shim/templates/alerts.yaml @@ -242,4 +242,31 @@ spec: resource router is mapping the same object to multiple clusters, or an object was created out-of-band on the wrong cluster. Investigate the affected resources and the routing configuration. + + - alert: CortexPlacementShimRemoteApiserverUnreachable + # min by (host): a remote is unreachable only if it was 0 for the whole + # window (a rebuild flips it 1->0->1, and min ignores the brief 1s during a + # reconnect attempt). Fires per host so the offending remote is named. + expr: | + min by (host) (cortex_multicluster_remote_apiserver_reachable{service="cortex-placement-shim-metrics-service"}) < 1 + for: 5m + labels: + context: multicluster + dashboard: cortex-placement-shim-status-dashboard/cortex-placement-shim-status-dashboard + service: cortex + severity: warning + support_group: workload-management + annotations: + summary: "Remote apiserver `{{ "{{" }} $labels.host {{ "}}" }}` is unreachable" + description: > + The multicluster client cannot reach the remote apiserver at + `{{ "{{" }} $labels.host {{ "}}" }}`. The reachability probe has seen + sustained transport failures (connection refused, TLS/handshake failure, + or timeout), which is the signature of a deleted or network-partitioned + cluster rather than an authz/server error. The shim recycles its + controller-manager in response, so the manager cache for resources served + by that remote is stale or empty and `cortex_placement_shim_manager_up` + will average low (see CortexPlacementShimManagerLooping). Passthrough + requests to upstream Placement are unaffected. Investigate whether the + remote cluster still exists and network connectivity to its apiserver. {{- end }} diff --git a/pkg/multicluster/client.go b/pkg/multicluster/client.go index 92cab3430..641b2b0f9 100644 --- a/pkg/multicluster/client.go +++ b/pkg/multicluster/client.go @@ -89,7 +89,7 @@ func (c *Client) InitFromConf(ctx context.Context, mgr ctrl.Manager, conf Client gvksByConfStr[formatted] = gvk } for gvkStr := range gvksByConfStr { - log.Info("scheme gvk registered", "gvk", gvkStr) + log.V(1).Info("scheme gvk registered", "gvk", gvkStr) } // Parse home GVKs. c.homeGVKs = make(map[schema.GroupVersionKind]bool) diff --git a/pkg/multicluster/client_probe.go b/pkg/multicluster/client_probe.go new file mode 100644 index 000000000..854ddd887 --- /dev/null +++ b/pkg/multicluster/client_probe.go @@ -0,0 +1,220 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package multicluster + +import ( + "context" + "errors" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/rest" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/cluster" +) + +// RemoteEndpoint identifies a unique remote cluster to probe for reachability. +type RemoteEndpoint struct { + // Host is the remote apiserver URL (from the cluster's rest.Config). + Host string + // Cluster is the controller-runtime cluster whose apiserver is probed. + Cluster cluster.Cluster +} + +// UniqueRemotes returns each distinct remote cluster exactly once, across all +// configured GVKs. The same remote apiserver is stored once per GVK it serves, +// so this dedupes by cluster identity (the same policy IndexField uses to dedupe +// caches). The home cluster is not included. +func (c *Client) UniqueRemotes() []RemoteEndpoint { + c.remoteClustersMu.RLock() + defer c.remoteClustersMu.RUnlock() + seen := make(map[cluster.Cluster]bool) + var out []RemoteEndpoint + for _, remotes := range c.remoteClusters { + for _, r := range remotes { + if r.cluster == nil || seen[r.cluster] { + continue + } + seen[r.cluster] = true + out = append(out, RemoteEndpoint{Host: r.cluster.GetConfig().Host, Cluster: r.cluster}) + } + } + return out +} + +// ProbeOptions configures the per-remote reachability probe. +type ProbeOptions struct { + // Interval is the time between reachability probes for each remote. + Interval time.Duration + // Timeout bounds a single probe request so it never blocks on a dead socket + // longer than intended. + Timeout time.Duration + // FailureThreshold is the number of consecutive unreachable probes that must + // occur before a remote is considered lost and onLost is called. It is chosen + // above the supervisor's backoff floor so a doomed manager cycle outlives the + // growing backoff, keeping the rebuild loop period bounded rather than hot. + FailureThreshold int +} + +// DefaultProbeOptions probes every 10s with a 5s per-probe timeout and treats a +// remote as lost after 3 consecutive failures (~30s of sustained unreachability). +var DefaultProbeOptions = ProbeOptions{ + Interval: 10 * time.Second, + Timeout: 5 * time.Second, + FailureThreshold: 3, +} + +// ProbeRemotes runs one reachability probe goroutine per unique remote apiserver +// until ctx is cancelled. Each goroutine periodically probes its remote, updates +// the reachability gauge on the Monitor (if any), and tracks consecutive +// failures; when a remote has been unreachable FailureThreshold times in a row it +// calls onLost(host) once and stops probing that remote (the caller is expected +// to tear down the manager cycle in response). +// +// A single transient blip does not trigger onLost: the failure counter resets on +// the first reachable probe. "Unreachable" means a transport-level failure +// (connection refused, no route, TLS/handshake failure, timeout) — an apiserver +// that responds with any HTTP status (including 401/403/5xx) is considered +// reachable, since a manager rebuild cannot fix an authz/server error and would +// only cause a restart storm. +// +// ProbeRemotes returns immediately when no remotes are configured. +func (c *Client) ProbeRemotes(ctx context.Context, opts ProbeOptions, onLost func(host string)) { + log := ctrl.LoggerFrom(ctx) + remotes := c.UniqueRemotes() + hosts := make([]string, 0, len(remotes)) + for _, r := range remotes { + hosts = append(hosts, r.Host) + } + if len(remotes) == 0 { + // No remotes to watch: log it explicitly so an operator who deletes a + // cluster and sees "nothing happening" can tell this apart from the probe + // simply not finding the remote it expected to watch. + log.Info("reachability probe: no remote apiservers configured, nothing to probe") + return + } + log.Info("reachability probe: starting", "remoteCount", len(remotes), "hosts", hosts, + "interval", opts.Interval.String(), "timeout", opts.Timeout.String(), "failureThreshold", opts.FailureThreshold) + var wg wait.Group + for _, remote := range remotes { + httpClient, err := reachabilityClient(remote.Cluster.GetConfig(), opts.Timeout) + if err != nil { + // A remote we cannot even build a probe client for is treated as + // lost immediately: its config is unusable, so a rebuild (which + // re-reads config) is the right response. + log.Error(err, "reachability probe: unable to build probe client for remote; declaring it lost", "host", remote.Host) + onLost(remote.Host) + continue + } + wg.StartWithContext(ctx, func(ctx context.Context) { + c.probeRemoteLoop(ctx, opts, remote, httpClient, onLost) + }) + } + wg.Wait() + log.Info("reachability probe: stopped", "hosts", hosts) +} + +// probeRemoteLoop probes a single remote until ctx is cancelled or the remote is +// declared lost. +func (c *Client) probeRemoteLoop(ctx context.Context, opts ProbeOptions, remote RemoteEndpoint, httpClient *rest.RESTClient, onLost func(host string)) { + log := ctrl.LoggerFrom(ctx).WithValues("host", remote.Host) + log.Info("reachability probe: watching remote apiserver") + failures := 0 + wasReachable := true // assume reachable at start; log the first real transition + // PollUntilContextCancel with immediate=true runs the first probe right away + // rather than waiting a full interval, so a remote that is already gone at + // cycle start is detected promptly. The poll func never returns an error, so + // the only error PollUntilContextCancel can return is the context error once + // ctx is cancelled or the loop stops after declaring the remote lost — both + // expected, so it is intentionally not surfaced. + err := wait.PollUntilContextCancel(ctx, opts.Interval, true, func(ctx context.Context) (bool, error) { + reachable, probeErr := probeReachable(ctx, httpClient, opts.Timeout) + if c.Monitor != nil { + c.Monitor.recordRemoteReachable(remote.Host, reachable) + } + // Per-probe outcome at V(1): verbose, but the single most useful line when + // diagnosing "why didn't it detect the deletion" — it shows the classified + // result and the underlying transport/HTTP error for each tick. + if probeErr != nil { + log.V(1).Info("reachability probe: tick", "reachable", reachable, "error", probeErr.Error()) + } else { + log.V(1).Info("reachability probe: tick", "reachable", reachable) + } + if reachable { + if !wasReachable { + log.Info("reachability probe: remote apiserver recovered", "afterConsecutiveFailures", failures) + } + wasReachable = true + failures = 0 + return false, nil + } + wasReachable = false + failures++ + if probeErr != nil { + log.Info("reachability probe: remote apiserver unreachable", + "consecutiveFailures", failures, "threshold", opts.FailureThreshold, "error", probeErr.Error()) + } else { + log.Info("reachability probe: remote apiserver unreachable", + "consecutiveFailures", failures, "threshold", opts.FailureThreshold) + } + if failures >= opts.FailureThreshold { + log.Info("reachability probe: failure threshold reached, declaring remote lost", "consecutiveFailures", failures) + onLost(remote.Host) + return true, nil + } + return false, nil + }) + if err != nil && ctx.Err() == nil { + // A non-context error is not expected here (the poll func never returns + // one), but log it rather than swallow it if the invariant ever changes. + log.Error(err, "remote reachability poll stopped unexpectedly") + } +} + +// reachabilityClient builds a REST client for the given remote config that is +// used only to probe reachability. It copies the config (so the caller's shared +// config is never mutated), sets a per-probe timeout, and supplies a codec so the +// unversioned REST client can be constructed (we only read the transport outcome, +// never decode a body, but rest requires a NegotiatedSerializer). +func reachabilityClient(cfg *rest.Config, timeout time.Duration) (*rest.RESTClient, error) { + cfgCopy := *cfg + cfgCopy.Timeout = timeout + cfgCopy.NegotiatedSerializer = scheme.Codecs.WithoutConversion() + if cfgCopy.GroupVersion == nil { + cfgCopy.GroupVersion = &schema.GroupVersion{} + } + httpClient, err := rest.HTTPClientFor(&cfgCopy) + if err != nil { + return nil, err + } + return rest.UnversionedRESTClientForConfigAndClient(&cfgCopy, httpClient) +} + +// probeReachable issues a lightweight GET /readyz against the remote apiserver +// and reports whether the apiserver is reachable. Any HTTP response — including +// non-2xx statuses surfaced as a Kubernetes StatusError — counts as reachable; +// only a transport-level error (connection refused, no route, TLS failure, +// timeout) counts as unreachable, which is the signature of a deleted cluster. +// probeReachable issues a lightweight GET /readyz against the remote apiserver +// and reports whether the apiserver is reachable, along with the underlying +// error (nil on a 2xx, otherwise the transport or HTTP error even when the result +// is classified reachable) for logging. Any HTTP response — including non-2xx +// statuses surfaced as a Kubernetes StatusError — counts as reachable; only a +// transport-level error (connection refused, no route, TLS failure, timeout) +// counts as unreachable, which is the signature of a deleted cluster. +func probeReachable(ctx context.Context, client *rest.RESTClient, timeout time.Duration) (bool, error) { + probeCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + err := client.Get().AbsPath("/readyz").Do(probeCtx).Error() + if err == nil { + return true, nil + } + // The apiserver answered with a non-2xx status: it is up and reachable. Return + // the error too so callers can log the status the apiserver returned. + var statusErr *apierrors.StatusError + return errors.As(err, &statusErr), err +} diff --git a/pkg/multicluster/client_probe_test.go b/pkg/multicluster/client_probe_test.go new file mode 100644 index 000000000..6bbe34f01 --- /dev/null +++ b/pkg/multicluster/client_probe_test.go @@ -0,0 +1,243 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package multicluster + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/rest" +) + +// fakeRemote builds a fakeCluster whose rest config points at the given host. +func fakeRemote(t *testing.T, host string) *fakeCluster { + t.Helper() + scheme := newTestScheme(t) + c := newFakeCluster(scheme) + c.restConfig = &rest.Config{Host: host} + return c +} + +func TestUniqueRemotes_DedupesAcrossGVKs(t *testing.T) { + gvkA := schema.GroupVersionKind{Group: "g", Version: "v", Kind: "A"} + gvkB := schema.GroupVersionKind{Group: "g", Version: "v", Kind: "B"} + shared := fakeRemote(t, "https://shared") + other := fakeRemote(t, "https://other") + c := &Client{ + remoteClusters: map[schema.GroupVersionKind][]remoteCluster{ + gvkA: {{cluster: shared}, {cluster: other}}, + gvkB: {{cluster: shared}}, + }, + } + remotes := c.UniqueRemotes() + if len(remotes) != 2 { + t.Fatalf("expected 2 unique remotes, got %d", len(remotes)) + } + hosts := map[string]bool{} + for _, r := range remotes { + hosts[r.Host] = true + } + if !hosts["https://shared"] || !hosts["https://other"] { + t.Errorf("unexpected hosts: %v", hosts) + } +} + +func TestUniqueRemotes_NoRemotes(t *testing.T) { + c := &Client{} + if got := c.UniqueRemotes(); len(got) != 0 { + t.Errorf("expected no remotes, got %d", len(got)) + } +} + +func TestProbeReachable_Classification(t *testing.T) { + tests := []struct { + name string + status int // 0 means "close the connection / unreachable" + wantReachable bool + }{ + {"ok", http.StatusOK, true}, + {"forbidden apiserver is up", http.StatusForbidden, true}, + {"unauthorized apiserver is up", http.StatusUnauthorized, true}, + {"server error apiserver is up", http.StatusInternalServerError, true}, + {"connection refused is unreachable", 0, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var host string + if tt.status == 0 { + // Reserve a port and close the listener so connections are refused. + srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + host = srv.URL + srv.Close() + } else { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(tt.status) + })) + t.Cleanup(srv.Close) + host = srv.URL + } + client, err := reachabilityClient(&rest.Config{Host: host, TLSClientConfig: rest.TLSClientConfig{Insecure: true}}, time.Second) + if err != nil { + t.Fatalf("reachabilityClient: %v", err) + } + got, probeErr := probeReachable(context.Background(), client, time.Second) + if got != tt.wantReachable { + t.Errorf("probeReachable = %v (err %v), want %v", got, probeErr, tt.wantReachable) + } + // A 200 yields no error; any other outcome (reachable non-2xx or an + // unreachable transport failure) surfaces the underlying error. + if tt.status == http.StatusOK && probeErr != nil { + t.Errorf("expected nil error on 200, got %v", probeErr) + } + if tt.status != http.StatusOK && probeErr == nil { + t.Errorf("expected a non-nil error for status %d / unreachable, got nil", tt.status) + } + }) + } +} + +func TestProbeRemotes_TriggersOnSustainedFailure(t *testing.T) { + // A remote pointing at a closed listener is always unreachable. + srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + deadHost := srv.URL + srv.Close() + + remote := fakeRemote(t, deadHost) + mon := NewMonitor("test_trigger_") + c := &Client{ + Monitor: mon, + remoteClusters: map[schema.GroupVersionKind][]remoteCluster{{Kind: "A"}: {{cluster: remote}}}, + } + + var mu sync.Mutex + var lostHosts []string + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + c.ProbeRemotes(ctx, ProbeOptions{Interval: 5 * time.Millisecond, Timeout: 50 * time.Millisecond, FailureThreshold: 3}, func(host string) { + mu.Lock() + lostHosts = append(lostHosts, host) + mu.Unlock() + }) + close(done) + }() + + // The probe loop stops itself once the remote is declared lost. + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("ProbeRemotes did not return after remote was declared lost") + } + + mu.Lock() + if len(lostHosts) != 1 || lostHosts[0] != deadHost { + t.Errorf("expected onLost called once with %q, got %v", deadHost, lostHosts) + } + mu.Unlock() + + if got := testutil.ToFloat64(mon.(*monitor).remoteReachable.WithLabelValues(deadHost)); got != 0 { + t.Errorf("expected reachable gauge 0 for dead host, got %v", got) + } +} + +func TestProbeRemotes_NoTriggerWhenReachable(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(srv.Close) + + remote := fakeRemote(t, srv.URL) + remote.restConfig.TLSClientConfig = rest.TLSClientConfig{Insecure: true} + mon := NewMonitor("test_reachable_") + c := &Client{ + Monitor: mon, + remoteClusters: map[schema.GroupVersionKind][]remoteCluster{{Kind: "A"}: {{cluster: remote}}}, + } + + var mu sync.Mutex + lost := false + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go c.ProbeRemotes(ctx, ProbeOptions{Interval: 5 * time.Millisecond, Timeout: time.Second, FailureThreshold: 3}, func(string) { + mu.Lock() + lost = true + mu.Unlock() + }) + + // Poll for the gauge to reach 1: a reachable remote must record 1 and never + // trip onLost. Polling (rather than a fixed sleep + single read) avoids racing + // the probe goroutine's gauge write. + gauge := mon.(*monitor).remoteReachable.WithLabelValues(srv.URL) + reachableSeen := false + for range 200 { + if testutil.ToFloat64(gauge) == 1 { + reachableSeen = true + break + } + time.Sleep(10 * time.Millisecond) + } + cancel() + + mu.Lock() + defer mu.Unlock() + if lost { + t.Error("onLost was called for a reachable remote") + } + if !reachableSeen { + t.Errorf("expected reachable gauge to reach 1, last value %v", testutil.ToFloat64(gauge)) + } +} + +func TestProbeRemotes_NoRemotesIsNoop(t *testing.T) { + c := &Client{} + lost := false + // Should return immediately without ever calling onLost. + c.ProbeRemotes(context.Background(), DefaultProbeOptions, func(string) { lost = true }) + if lost { + t.Error("onLost called with no remotes configured") + } +} + +func TestProbeRemotes_TransientBlipResetsCounter(t *testing.T) { + // probeRemoteLoop's counter resets on any reachable probe, so a run of + // failures shorter than the threshold never trips onLost. Drive the counter + // logic directly: a probe func that fails twice (below threshold 3) then + // recovers must leave failures at 0 and never call onLost. + failures := 0 + onLost := false + reachSeq := []bool{false, false, true, true} + i := 0 + step := func() bool { + reachable := reachSeq[i] + i++ + if reachable { + failures = 0 + return false + } + failures++ + if failures >= 3 { + onLost = true + return true + } + return false + } + for i < len(reachSeq) { + if step() { + break + } + } + if onLost { + t.Error("onLost tripped despite failures never reaching the threshold") + } + if failures != 0 { + t.Errorf("expected failure counter reset to 0 after recovery, got %d", failures) + } +} diff --git a/pkg/multicluster/client_test.go b/pkg/multicluster/client_test.go index e208394de..d3159c00d 100644 --- a/pkg/multicluster/client_test.go +++ b/pkg/multicluster/client_test.go @@ -16,6 +16,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/rest" "k8s.io/client-go/tools/record" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/cache" @@ -83,12 +84,24 @@ type fakeCluster struct { fakeCache *fakeCache fakeRecorder recorder.EventRecorder scheme *runtime.Scheme + restConfig *rest.Config } func (f *fakeCluster) GetClient() client.Client { return f.fakeClient } +// GetConfig returns the rest config the cluster was built with, defaulting to a +// placeholder host when unset. Production clusters always have one (used for +// per-cluster logging and reachability probing); the default keeps tests that +// don't care about the host from panicking on the embedded nil interface. +func (f *fakeCluster) GetConfig() *rest.Config { + if f.restConfig != nil { + return f.restConfig + } + return &rest.Config{Host: "https://fake-cluster"} +} + func (f *fakeCluster) GetScheme() *runtime.Scheme { return f.scheme } diff --git a/pkg/multicluster/monitor.go b/pkg/multicluster/monitor.go index a33cfffb8..6cf3883b5 100644 --- a/pkg/multicluster/monitor.go +++ b/pkg/multicluster/monitor.go @@ -23,6 +23,11 @@ type Monitor interface { // detected on more than one cluster serving the GVK, labeled by the method // of access and the resource GVK. recordCrossClusterNameConflict(method string, gvk schema.GroupVersionKind) + + // recordRemoteReachable records whether a remote apiserver was reachable at + // the last reachability probe, labeled by host. See pkg/multicluster's + // ProbeRemotes for how reachability is determined. + recordRemoteReachable(host string, reachable bool) } // monitor is the default Prometheus-backed Monitor implementation. @@ -31,6 +36,10 @@ type monitor struct { // detected on more than one cluster serving the GVK, labeled by the method // of access and the resource GVK. crossClusterNameConflicts *prometheus.CounterVec + // remoteReachable reports whether each remote apiserver was reachable at the + // last reachability probe, labeled by host. It lives on the process-lifetime + // Monitor (not the per-cycle Client) so it survives manager rebuilds. + remoteReachable *prometheus.GaugeVec } // NewMonitor creates a new Prometheus-backed multicluster client monitor. The @@ -42,6 +51,10 @@ func NewMonitor(prefix string) Monitor { Name: prefix + "multicluster_cross_cluster_name_conflicts_total", Help: "Total number of times the same resource name was detected on more than one cluster serving the same GVK", }, duplicateConflictLabels), + remoteReachable: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: prefix + "multicluster_remote_apiserver_reachable", + Help: "1 if the remote apiserver was reachable at the last probe, 0 if sustained-unreachable, labeled by host", + }, []string{"host"}), } } @@ -51,12 +64,24 @@ func (m *monitor) recordCrossClusterNameConflict(method string, gvk schema.Group m.crossClusterNameConflicts.WithLabelValues(method, gvk.String()).Inc() } +// recordRemoteReachable sets the reachability gauge for the given host to 1 +// (reachable) or 0 (unreachable). +func (m *monitor) recordRemoteReachable(host string, reachable bool) { + v := 0.0 + if reachable { + v = 1.0 + } + m.remoteReachable.WithLabelValues(host).Set(v) +} + // Describe implements prometheus.Collector. func (m *monitor) Describe(ch chan<- *prometheus.Desc) { m.crossClusterNameConflicts.Describe(ch) + m.remoteReachable.Describe(ch) } // Collect implements prometheus.Collector. func (m *monitor) Collect(ch chan<- prometheus.Metric) { m.crossClusterNameConflicts.Collect(ch) + m.remoteReachable.Collect(ch) } diff --git a/pkg/multicluster/monitor_test.go b/pkg/multicluster/monitor_test.go index 104751280..0660595e6 100644 --- a/pkg/multicluster/monitor_test.go +++ b/pkg/multicluster/monitor_test.go @@ -91,3 +91,24 @@ func TestMonitor_RecordCrossClusterNameConflict(t *testing.T) { t.Errorf("list/%s: got %v, want 0", gvk, got) } } + +func TestMonitor_RecordRemoteReachable(t *testing.T) { + m := NewMonitor("cortex_").(*monitor) + + // reachable=true sets the gauge to 1, reachable=false to 0, and the latest + // value for a host wins. + m.recordRemoteReachable("https://a", true) + m.recordRemoteReachable("https://b", false) + if got := testutil.ToFloat64(m.remoteReachable.WithLabelValues("https://a")); got != 1 { + t.Errorf("host a: got %v, want 1", got) + } + if got := testutil.ToFloat64(m.remoteReachable.WithLabelValues("https://b")); got != 0 { + t.Errorf("host b: got %v, want 0", got) + } + + // A host flipping from reachable to unreachable overwrites the prior value. + m.recordRemoteReachable("https://a", false) + if got := testutil.ToFloat64(m.remoteReachable.WithLabelValues("https://a")); got != 0 { + t.Errorf("host a after loss: got %v, want 0", got) + } +} diff --git a/pkg/shim/supervisor/supervisor_test.go b/pkg/shim/supervisor/supervisor_test.go index 8d86b2b5a..258410ee2 100644 --- a/pkg/shim/supervisor/supervisor_test.go +++ b/pkg/shim/supervisor/supervisor_test.go @@ -285,6 +285,35 @@ func TestBackoffProgressionAndReset(t *testing.T) { } } +// TestManagerRestartsOnInternalCycleCancel locks down the exact mechanism the +// placement shim relies on: BuildAndStart derives a per-cycle child context from +// the one the supervisor passes in and cancels it itself (as the shim does when a +// remote apiserver goes unreachable), then returns. The parent context is still +// live, so the supervisor must treat this as a manager exit and rebuild — not +// mistake it for a graceful shutdown and stop the loop. +func TestManagerRestartsOnInternalCycleCancel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + var cycles atomic.Int32 + + o := baseOptions(t) + o.Backoff = wait.Backoff{Duration: time.Millisecond, Factor: 1.0, Steps: 100} + o.BuildAndStart = func(parent context.Context) error { + cycles.Add(1) + // Mirror main.go's buildAndStart: derive a per-cycle context and cancel + // it from within (as the reachability probe does), then return its error. + cycleCtx, cancelCycle := context.WithCancel(parent) + cancelCycle() + return cycleCtx.Err() + } + runAsync(t, ctx, o) + + // A cycle that self-cancels and returns while the parent is live must be + // rebuilt, not end the supervision loop. + waitFor(t, func() bool { return cycles.Load() >= 3 }, "manager to be rebuilt after an internal per-cycle cancel") +} + // waitFor polls cond up to ~2s, failing the test if it never becomes true. func waitFor(t *testing.T, cond func() bool, what string) { t.Helper()