From a1fc2bb437187053609fd605c0847db3bc0bf0ed Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Sun, 2 Aug 2026 09:37:33 +0530 Subject: [PATCH 01/10] feat(hub): add the active-hub resolver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A worker cannot learn about a controller failover from the hub that just failed, because the promotion is recorded on the other hub. So each hub publishes status.activeController on this worker's own Cluster CR while it holds leadership, and a Standby's mirrored copy repeats the Active's declaration. This is the worker's half: read that field from both pre-provisioned endpoints and decide who to talk to. Wired to nothing. No caller, no behaviour change, and a non-HA worker is byte-identical. The decision logic is the part with real substance, it reviews on its own, and landing it separately keeps the change that touches the hub connection small when it comes. The rule, in order: an unreachable hub has no say; a hub that published nothing has no say, which is what every non-HA deployment looks like from here; a claim naming an endpoint outside the configured candidate set is rejected, because the field selects among endpoints an operator provisioned rather than pointing the worker at arbitrary addresses; agreement between the hubs wins, and agreement is the normal case since the Standby mirrors the Active's declaration; disagreement prefers the fresher declaration, which keeps behaviour single-valued during the split brain the design does not claim to solve. No usable claim means change nothing — a worker that disconnected whenever it was unsure would turn every hub blip into a worker outage. Two properties worth their own tests. A switch needs consecutive confirming polls, so one divergent poll cannot move a worker, and the comparison excludes LastUpdated: the Active republishes on a timer, so including it would reset the counter every poll and no switch could ever confirm. Every read is deadline-bounded, because an API server that accepts a connection and then stops answering hangs until the OS TCP timeout otherwise; the controller side of this feature shipped that bug and measured a single read blocking ~12s against a stopped API server. The field is read unstructured rather than through the shared github.com/kubeslice/apis types. It is four scalars out of one status field, and reading them untyped keeps a third repository's release cadence off the critical path of a package that is otherwise self-contained. Nothing in go.mod, go.sum or vendor/ changes as a result. Part of #467 Signed-off-by: Sumanth D --- pkg/hub/resolver/probe.go | 147 +++++++++++++ pkg/hub/resolver/probe_test.go | 207 ++++++++++++++++++ pkg/hub/resolver/resolver.go | 322 ++++++++++++++++++++++++++++ pkg/hub/resolver/resolver_test.go | 339 ++++++++++++++++++++++++++++++ 4 files changed, 1015 insertions(+) create mode 100644 pkg/hub/resolver/probe.go create mode 100644 pkg/hub/resolver/probe_test.go create mode 100644 pkg/hub/resolver/resolver.go create mode 100644 pkg/hub/resolver/resolver_test.go diff --git a/pkg/hub/resolver/probe.go b/pkg/hub/resolver/probe.go new file mode 100644 index 000000000..bd2052d5f --- /dev/null +++ b/pkg/hub/resolver/probe.go @@ -0,0 +1,147 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 resolver + +import ( + "context" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// clusterGVK is the hub-side Cluster CR carrying status.activeController. +// +// Read unstructured rather than through the shared github.com/kubeslice/apis +// types on purpose. This package needs four scalars out of one status field, +// and reading them untyped means it does not depend on an apis release shipping +// ActiveControllerInfo — which would otherwise put a third repository's release +// cadence on the critical path of a feature that is otherwise self-contained. +// The typed watch the worker already runs over its own Cluster CR is untouched. +var clusterGVK = schema.GroupVersionKind{ + Group: "controller.kubeslice.io", + Version: "v1alpha1", + Kind: "Cluster", +} + +// ClusterReader reads one object from one hub. Narrowed to what the probe needs +// so a caller can hand over a controller-runtime client, and a test can hand +// over anything. +type ClusterReader interface { + Get(ctx context.Context, key types.NamespacedName, obj *unstructured.Unstructured) error +} + +// clientReader adapts a controller-runtime client to ClusterReader. +type clientReader struct{ c client.Client } + +func (r clientReader) Get(ctx context.Context, key types.NamespacedName, obj *unstructured.Unstructured) error { + return r.c.Get(ctx, key, obj) +} + +// NewClusterReader adapts a controller-runtime client for use as a probe target. +func NewClusterReader(c client.Client) ClusterReader { + return clientReader{c: c} +} + +// ProbeConfig describes which object a probe reads and how long it may take. +type ProbeConfig struct { + // ClusterName is this worker's own Cluster CR, the object every hub keeps a + // copy of and stamps activeController onto. + ClusterName string + // Namespace is the hub-side project namespace holding that CR. + Namespace string + // Timeout bounds each read. Zero means DefaultProbeTimeout. + Timeout time.Duration +} + +// NewProbe returns a prober that reads each candidate's copy of this worker's +// Cluster CR through the reader built for that candidate by readerFor. +// +// readerFor is a function rather than a prepared map because a candidate's +// client is built from its own endpoint and credentials, and the caller owns +// that construction. Building them once at startup and never rebuilding them is +// the intent: these probes must keep working across a failover, so they are +// deliberately independent of the worker's primary hub connection, which is the +// thing a failover replaces. +func NewProbe(readerFor func(HubCandidate) (ClusterReader, error), cfg ProbeConfig) prober { + if cfg.Timeout <= 0 { + cfg.Timeout = DefaultProbeTimeout + } + return func(ctx context.Context, candidate HubCandidate) Verdict { + reader, err := readerFor(candidate) + if err != nil { + return Verdict{Candidate: candidate, Err: err} + } + + readCtx, cancel := context.WithTimeout(ctx, cfg.Timeout) + defer cancel() + + obj := &unstructured.Unstructured{} + obj.SetGroupVersionKind(clusterGVK) + key := types.NamespacedName{Name: cfg.ClusterName, Namespace: cfg.Namespace} + if err := reader.Get(readCtx, key, obj); err != nil { + if apierrors.IsNotFound(err) { + // The hub answered; this worker simply has no Cluster CR there. + // That is a reachable hub with nothing to say, not an outage — + // and treating it as unreachable would hide a real + // misconfiguration behind a connectivity error. + return Verdict{Candidate: candidate, Reachable: true, Err: err} + } + return Verdict{Candidate: candidate, Err: err} + } + + return Verdict{ + Candidate: candidate, + Reachable: true, + Claim: claimFrom(obj, candidate.Name), + } + } +} + +// claimFrom decodes status.activeController, returning nil when the field is +// absent or unusable. +// +// A partially-written field is treated as absent rather than as an error: an +// endpoint or identity that is missing cannot be acted on, and a worker that +// refused to resolve at all because one hub published something malformed would +// be broken by the other hub's bug. +func claimFrom(obj *unstructured.Unstructured, source string) *Claim { + raw, found, err := unstructured.NestedMap(obj.Object, "status", "activeController") + if err != nil || !found || raw == nil { + return nil + } + endpoint, _, _ := unstructured.NestedString(raw, "endpoint") + identity, _, _ := unstructured.NestedString(raw, "activeIdentity") + if endpoint == "" || identity == "" { + return nil + } + claim := &Claim{Endpoint: endpoint, Identity: identity, Source: source} + if stamp, ok, _ := unstructured.NestedString(raw, "lastUpdated"); ok && stamp != "" { + // Absence or a malformed stamp leaves the zero time, which orders last + // in the tie-break. A hub that cannot say when it last declared itself + // should not win against one that can. + if parsed, err := time.Parse(time.RFC3339, stamp); err == nil { + claim.LastUpdated = parsed + } + } + return claim +} diff --git a/pkg/hub/resolver/probe_test.go b/pkg/hub/resolver/probe_test.go new file mode 100644 index 000000000..4ed18f8d7 --- /dev/null +++ b/pkg/hub/resolver/probe_test.go @@ -0,0 +1,207 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 resolver + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" +) + +const ( + testCluster = "worker-1" + testNamespace = "kubeslice-avesha" +) + +// readerFunc adapts a plain function to ClusterReader. +type readerFunc func(ctx context.Context, key types.NamespacedName, obj *unstructured.Unstructured) error + +func (f readerFunc) Get(ctx context.Context, key types.NamespacedName, obj *unstructured.Unstructured) error { + return f(ctx, key, obj) +} + +// clusterWith builds the hub-side Cluster CR carrying an activeController. +func clusterWith(activeController map[string]interface{}) *unstructured.Unstructured { + obj := &unstructured.Unstructured{Object: map[string]interface{}{}} + obj.SetGroupVersionKind(clusterGVK) + obj.SetName(testCluster) + obj.SetNamespace(testNamespace) + if activeController != nil { + _ = unstructured.SetNestedMap(obj.Object, activeController, "status", "activeController") + } + return obj +} + +// probeReturning builds a probe whose single reader serves obj (or err). +func probeReturning(t *testing.T, obj *unstructured.Unstructured, err error, timeout time.Duration) prober { + t.Helper() + reader := readerFunc(func(ctx context.Context, key types.NamespacedName, into *unstructured.Unstructured) error { + assert.Equal(t, testCluster, key.Name, "the probe must read this worker's own Cluster CR") + assert.Equal(t, testNamespace, key.Namespace) + if err != nil { + return err + } + into.Object = obj.Object + return nil + }) + return NewProbe( + func(HubCandidate) (ClusterReader, error) { return reader, nil }, + ProbeConfig{ClusterName: testCluster, Namespace: testNamespace, Timeout: timeout}, + ) +} + +func TestProbe_DecodesAFullDeclaration(t *testing.T) { + stamp := time.Now().UTC().Truncate(time.Second) + probe := probeReturning(t, clusterWith(map[string]interface{}{ + "endpoint": hubA.Endpoint, + "activeIdentity": "hub-a-1", + "lastUpdated": stamp.Format(time.RFC3339), + }), nil, time.Second) + + got := probe(context.Background(), hubA) + assert.True(t, got.Reachable) + claim := mustClaim(t, got.Claim, "a complete activeController must decode") + assert.Equal(t, hubA.Endpoint, claim.Endpoint) + assert.Equal(t, "hub-a-1", claim.Identity) + assert.True(t, stamp.Equal(claim.LastUpdated), "lastUpdated must round-trip") + assert.Equal(t, hubA.Name, claim.Source, "the claim records which hub answered") +} + +// TestProbe_NoFieldIsNotAnError is the non-HA path: a hub that answers and +// publishes nothing is healthy and has nothing to say. Treating that as a +// failure would make every existing worker log errors forever. +func TestProbe_NoFieldIsNotAnError(t *testing.T) { + probe := probeReturning(t, clusterWith(nil), nil, time.Second) + + got := probe(context.Background(), hubA) + assert.True(t, got.Reachable, "the hub answered") + assert.Nil(t, got.Claim) + assert.NoError(t, got.Err) +} + +// TestProbe_PartialDeclarationIsIgnored: a claim missing either half cannot be +// acted on. Treating it as an error instead would let one hub's bug stop the +// worker from resolving against the other. +func TestProbe_PartialDeclarationIsIgnored(t *testing.T) { + for name, ac := range map[string]map[string]interface{}{ + "no endpoint": {"activeIdentity": "hub-a-1"}, + "no identity": {"endpoint": hubA.Endpoint}, + "both empty": {"endpoint": "", "activeIdentity": ""}, + } { + t.Run(name, func(t *testing.T) { + got := probeReturning(t, clusterWith(ac), nil, time.Second)(context.Background(), hubA) + assert.True(t, got.Reachable) + assert.Nil(t, got.Claim, "a half-written declaration must not be acted on") + }) + } +} + +// TestProbe_UnparseableTimestampStillYieldsAClaim: the stamp only orders +// conflicting claims. Dropping an otherwise-valid declaration because its +// timestamp was malformed would be worse than ordering it last. +func TestProbe_UnparseableTimestampStillYieldsAClaim(t *testing.T) { + probe := probeReturning(t, clusterWith(map[string]interface{}{ + "endpoint": hubA.Endpoint, + "activeIdentity": "hub-a-1", + "lastUpdated": "not-a-timestamp", + }), nil, time.Second) + + claim := mustClaim(t, probe(context.Background(), hubA).Claim, "a bad stamp must not void the claim") + assert.True(t, claim.LastUpdated.IsZero(), "an unusable stamp orders last in the tie-break") +} + +// TestProbe_NotFoundIsReachable separates "this hub is down" from "this worker +// is not registered on this hub". Only the first is a connectivity problem, and +// conflating them would hide a real misconfiguration behind a network error. +func TestProbe_NotFoundIsReachable(t *testing.T) { + notFound := apierrors.NewNotFound(schema.GroupResource{Group: clusterGVK.Group, Resource: "clusters"}, testCluster) + got := probeReturning(t, nil, notFound, time.Second)(context.Background(), hubA) + + assert.True(t, got.Reachable, "the API server answered; it just has no such object") + assert.Nil(t, got.Claim) + assert.Error(t, got.Err) +} + +func TestProbe_TransportFailureIsUnreachable(t *testing.T) { + got := probeReturning(t, nil, fmt.Errorf("connection refused"), time.Second)(context.Background(), hubA) + assert.False(t, got.Reachable) + assert.Error(t, got.Err) + assert.Nil(t, got.Claim) +} + +func TestProbe_ReaderConstructionFailureIsReported(t *testing.T) { + probe := NewProbe( + func(HubCandidate) (ClusterReader, error) { return nil, fmt.Errorf("bad kubeconfig") }, + ProbeConfig{ClusterName: testCluster, Namespace: testNamespace}, + ) + got := probe(context.Background(), hubA) + assert.False(t, got.Reachable) + assert.Error(t, got.Err) +} + +// TestProbe_BoundsAHangingRead is the whole reason every read here is wrapped. +// An API server that accepts a connection and then stops answering — a +// powered-off node, a partition dropping packets — leaves an unbounded read +// hanging until the OS TCP timeout, minutes later. The controller side of this +// feature shipped exactly that bug and measured a single read blocking ~12s +// against a stopped API server. +func TestProbe_BoundsAHangingRead(t *testing.T) { + hang := readerFunc(func(ctx context.Context, _ types.NamespacedName, _ *unstructured.Unstructured) error { + <-ctx.Done() // never answers; only the probe's own deadline ends this + return ctx.Err() + }) + probe := NewProbe( + func(HubCandidate) (ClusterReader, error) { return hang, nil }, + ProbeConfig{ClusterName: testCluster, Namespace: testNamespace, Timeout: 50 * time.Millisecond}, + ) + + start := time.Now() + got := probe(context.Background(), hubA) + elapsed := time.Since(start) + + assert.False(t, got.Reachable) + assert.Error(t, got.Err) + assert.Less(t, elapsed, 2*time.Second, + "a hanging hub must not block the poll loop; it returned after %s", elapsed) +} + +func TestProbe_DefaultsTheTimeout(t *testing.T) { + var seen time.Duration + reader := readerFunc(func(ctx context.Context, _ types.NamespacedName, _ *unstructured.Unstructured) error { + deadline, ok := ctx.Deadline() + assert.True(t, ok, "every read must carry a deadline") + seen = time.Until(deadline) + return fmt.Errorf("done") + }) + probe := NewProbe( + func(HubCandidate) (ClusterReader, error) { return reader, nil }, + ProbeConfig{ClusterName: testCluster, Namespace: testNamespace}, // no Timeout + ) + + probe(context.Background(), hubA) + assert.Greater(t, seen, time.Duration(0)) + assert.LessOrEqual(t, seen, DefaultProbeTimeout) +} diff --git a/pkg/hub/resolver/resolver.go b/pkg/hub/resolver/resolver.go new file mode 100644 index 000000000..d1d5afa8d --- /dev/null +++ b/pkg/hub/resolver/resolver.go @@ -0,0 +1,322 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 resolver decides which of a worker's two pre-provisioned hub clusters +// is currently the Active one, so the worker can follow a controller failover +// without manual intervention. +// +// A worker cannot learn about a failover from the hub that just failed: the +// promotion is recorded on the *other* hub. So each hub publishes +// status.activeController on this worker's own Cluster CR while it holds +// leadership, and a Standby's mirrored copy repeats the Active's declaration. +// This package reads that field from both endpoints and applies one rule to +// decide who to talk to. +// +// It resolves; it does not connect. Nothing here opens or closes the worker's +// hub connections — that is the caller's job, and keeping the two apart is what +// makes this testable without a cluster. See worker-operator issue #467 and +// kubeslice-controller issue #297. +package resolver + +import ( + "context" + "fmt" + "sort" + "time" + + "github.com/go-logr/logr" +) + +const ( + // DefaultProbeTimeout bounds a single read of one hub's copy of the Cluster + // CR. Every networked read here is bounded, and not as a formality: an API + // server that accepts a connection and then stops answering (a powered-off + // node, a partition dropping packets) leaves an unbounded read hanging until + // the OS TCP timeout, which is minutes. The kubeslice-controller side of + // this feature shipped that bug and measured a single read blocking for ~12 + // seconds against a stopped API server before the connection broke. + DefaultProbeTimeout = 5 * time.Second + + // DefaultSwitchConfirmations is how many consecutive polls must agree on a + // new winner before Resolve reports the switch. A worker that re-points its + // hub connection on a single divergent poll would flap through every + // transient blip; requiring agreement costs one poll interval of latency and + // removes that entire class of behaviour. + DefaultSwitchConfirmations = 2 +) + +// HubCandidate is one pre-provisioned hub endpoint and the credentials to reach +// it. The set of candidates is fixed at startup and never derived from cluster +// state — that is what makes it a trust boundary rather than a suggestion. +type HubCandidate struct { + // Name is a stable label for logs and metrics ("primary", "secondary"). + // It has no relationship to which hub is currently Active. + Name string + Endpoint string + TokenFile string + CAFile string +} + +// Claim is one hub's statement about which controller currently holds +// leadership. It is the decoded form of status.activeController. +type Claim struct { + // Endpoint is where the Active hub says it is reachable. + Endpoint string + // Identity is the Active hub's leader identity, stable across its restarts. + Identity string + // LastUpdated is when the Active last refreshed the declaration. It exists + // so two conflicting claims can be ordered; see Resolve. + LastUpdated time.Time + // Source names the candidate this claim was read from, which is not + // necessarily the hub the claim is about — a Standby's mirrored copy names + // the Active. + Source string +} + +// Verdict is the outcome of probing one candidate. +type Verdict struct { + Candidate HubCandidate + // Reachable reports whether the probe got an answer at all. + Reachable bool + // Claim is nil when the hub answered but published nothing. That is the + // ordinary non-HA case, not a failure. + Claim *Claim + Err error +} + +// prober reads one candidate's copy of this worker's Cluster CR. It is a field +// on Resolver rather than a hardcoded call so the decision logic can be tested +// without a cluster, and so a caller can supply a client built however it likes. +type prober func(ctx context.Context, candidate HubCandidate) Verdict + +// Options configures a Resolver. Zero-valued fields fall back to the Default* +// constants. +type Options struct { + // ProbeTimeout bounds each individual candidate read. + ProbeTimeout time.Duration + // SwitchConfirmations is how many consecutive agreeing polls are required + // before a change of winner is reported. + SwitchConfirmations int + Log logr.Logger +} + +// Resolver applies the active-hub selection rule across a fixed candidate set. +// It is not safe for concurrent use: it keeps the confirmation counter that +// suppresses flapping, and it is meant to be driven by a single polling loop. +type Resolver struct { + candidates []HubCandidate + probe prober + + confirmations int + log logr.Logger + + // current is the winner Resolve last reported, and pending/pendingCount are + // the challenger accumulating confirmations against it. + current *Claim + pending *Claim + pendingCount int +} + +// New builds a Resolver over a fixed candidate set. +func New(candidates []HubCandidate, probe prober, opts Options) (*Resolver, error) { + if len(candidates) == 0 { + return nil, fmt.Errorf("resolver requires at least one hub candidate") + } + if probe == nil { + return nil, fmt.Errorf("resolver requires a probe function") + } + seen := make(map[string]struct{}, len(candidates)) + for _, c := range candidates { + if c.Endpoint == "" { + return nil, fmt.Errorf("hub candidate %q has no endpoint", c.Name) + } + if _, dup := seen[c.Endpoint]; dup { + return nil, fmt.Errorf("duplicate hub candidate endpoint %q", c.Endpoint) + } + seen[c.Endpoint] = struct{}{} + } + if opts.SwitchConfirmations <= 0 { + opts.SwitchConfirmations = DefaultSwitchConfirmations + } + // A zero logr.Logger has a nil sink and panics on first use, so an Options + // literal that simply omits Log would take the worker down on the first + // poll rather than at construction. + if opts.Log.GetSink() == nil { + opts.Log = logr.Discard() + } + return &Resolver{ + candidates: candidates, + probe: probe, + confirmations: opts.SwitchConfirmations, + log: opts.Log, + }, nil +} + +// Current returns the winner most recently confirmed, or nil if none ever has +// been. It never blocks and performs no I/O. +func (r *Resolver) Current() *Claim { + return r.current +} + +// Resolve probes every candidate once and returns the hub the worker should be +// talking to, or nil to mean "change nothing". +// +// Returning nil is a real answer, not an error: with no usable claim the +// correct behaviour is to leave the existing connection alone. A worker that +// disconnected whenever it was unsure would turn every hub blip into a worker +// outage, which is strictly worse than talking to a hub that might be stale. +func (r *Resolver) Resolve(ctx context.Context) *Claim { + claims := r.gather(ctx) + winner := r.pick(claims) + if winner == nil { + // Deliberately does NOT reset the pending challenger. A single poll in + // which both hubs happen to be unreachable should not undo confirmations + // already accumulated; only a competing claim displaces one. + return r.current + } + return r.confirm(winner) +} + +// gather probes every candidate and returns the claims worth considering. +func (r *Resolver) gather(ctx context.Context) []Claim { + var claims []Claim + for _, candidate := range r.candidates { + verdict := r.probe(ctx, candidate) + switch { + case !verdict.Reachable: + r.log.V(1).Info("hub candidate unreachable", "hub", candidate.Name, + "endpoint", candidate.Endpoint, "error", verdict.Err) + case verdict.Claim == nil: + // The hub answered and published nothing. This is what a non-HA + // deployment looks like from here, and it must stay silent at info + // level or every worker in every existing cluster logs a warning + // forever. + r.log.V(1).Info("hub candidate published no activeController", "hub", candidate.Name) + case !r.known(verdict.Claim.Endpoint): + // The trust boundary. The field chooses among endpoints an operator + // pre-provisioned; it does not get to point this worker at an + // address nobody configured. Anything else would make write access + // to one Cluster CR enough to redirect a worker's hub connection. + r.log.Info("ignoring activeController naming an unconfigured endpoint", + "hub", candidate.Name, "declaredEndpoint", verdict.Claim.Endpoint, + "declaredIdentity", verdict.Claim.Identity) + default: + claims = append(claims, *verdict.Claim) + } + } + return claims +} + +// known reports whether endpoint belongs to a configured candidate. +func (r *Resolver) known(endpoint string) bool { + for _, c := range r.candidates { + if c.Endpoint == endpoint { + return true + } + } + return false +} + +// pick reduces the surviving claims to at most one winner. +// +// Agreement is the normal case, not the exception, and misreading that is the +// easiest way to get this wrong: in steady state the Active declares itself and +// the Standby's mirrored copy repeats the same declaration, so both hubs +// answer with the same identity. Two claims naming one hub is one winner. +// +// Disagreement means two hubs each believe they are Active — most plausibly an +// old Active that came back after a promotion and still names itself. Split +// brain is an explicit non-goal of the design and this does not resolve it; it +// only keeps the worker's behaviour single-valued while it lasts, by preferring +// the fresher declaration. +func (r *Resolver) pick(claims []Claim) *Claim { + if len(claims) == 0 { + return nil + } + distinct := make(map[string]struct{}, len(claims)) + for _, c := range claims { + distinct[c.Identity] = struct{}{} + } + if len(distinct) == 1 { + winner := claims[0] + return &winner + } + + // Sorted rather than max-scanned so the outcome is total and repeatable: + // identical LastUpdated values must not resolve differently between polls, + // or the anti-flap counter below could never accumulate. + sorted := make([]Claim, len(claims)) + copy(sorted, claims) + sort.SliceStable(sorted, func(i, j int) bool { + if !sorted[i].LastUpdated.Equal(sorted[j].LastUpdated) { + return sorted[i].LastUpdated.After(sorted[j].LastUpdated) + } + return sorted[i].Identity < sorted[j].Identity + }) + r.log.Info("hubs disagree about which controller is active; preferring the freshest declaration", + "chosen", sorted[0].Identity, "chosenLastUpdated", sorted[0].LastUpdated, + "rejected", sorted[1].Identity, "rejectedLastUpdated", sorted[1].LastUpdated) + winner := sorted[0] + return &winner +} + +// confirm applies the anti-flap rule and returns the claim the caller should +// act on: either the established winner, or a challenger that has now agreed +// with itself enough times to replace it. +func (r *Resolver) confirm(winner *Claim) *Claim { + if r.current != nil && sameTarget(r.current, winner) { + // Re-confirmation of the status quo clears any half-accumulated + // challenger; a challenger has to win consecutive polls, not cumulative + // ones. + r.pending = nil + r.pendingCount = 0 + return r.current + } + + if r.pending != nil && sameTarget(r.pending, winner) { + r.pendingCount++ + } else { + r.pending = winner + r.pendingCount = 1 + } + + if r.pendingCount < r.confirmations { + r.log.Info("candidate active hub not yet confirmed; keeping the current one", + "candidate", winner.Identity, "seen", r.pendingCount, "need", r.confirmations) + return r.current + } + + previous := "" + if r.current != nil { + previous = r.current.Identity + } + r.current = winner + r.pending = nil + r.pendingCount = 0 + r.log.Info("active hub resolved", "identity", winner.Identity, + "endpoint", winner.Endpoint, "previous", previous) + return r.current +} + +// sameTarget reports whether two claims name the same hub. Compared on identity +// and endpoint only: LastUpdated advances on every republication and would make +// every claim differ from itself, so including it would reset the confirmation +// counter on every poll and no switch would ever be confirmed. +func sameTarget(a, b *Claim) bool { + return a.Identity == b.Identity && a.Endpoint == b.Endpoint +} diff --git a/pkg/hub/resolver/resolver_test.go b/pkg/hub/resolver/resolver_test.go new file mode 100644 index 000000000..43095a8a4 --- /dev/null +++ b/pkg/hub/resolver/resolver_test.go @@ -0,0 +1,339 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 resolver + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +var ( + hubA = HubCandidate{Name: "primary", Endpoint: "https://hub-a.example:6443"} + hubB = HubCandidate{Name: "secondary", Endpoint: "https://hub-b.example:6443"} +) + +func candidates() []HubCandidate { return []HubCandidate{hubA, hubB} } + +// declares builds the verdict a hub gives when it reports someone as Active. +// The declared hub is often not the hub answering: a Standby's mirrored copy +// names the Active, which is the whole basis of the resolution rule. +func declares(answering HubCandidate, about HubCandidate, identity string, age time.Duration) Verdict { + return Verdict{ + Candidate: answering, + Reachable: true, + Claim: &Claim{ + Endpoint: about.Endpoint, + Identity: identity, + LastUpdated: time.Now().Add(-age), + Source: answering.Name, + }, + } +} + +func silent(c HubCandidate) Verdict { return Verdict{Candidate: c, Reachable: true} } +func unreachable(c HubCandidate) Verdict { + return Verdict{Candidate: c, Err: fmt.Errorf("simulated: connection refused")} +} + +// staticProbe answers each candidate from a fixed table. +func staticProbe(verdicts map[string]Verdict) prober { + return func(_ context.Context, c HubCandidate) Verdict { + if v, ok := verdicts[c.Endpoint]; ok { + return v + } + return unreachable(c) + } +} + +func newResolver(t *testing.T, probe prober, confirmations int) *Resolver { + t.Helper() + r, err := New(candidates(), probe, Options{SwitchConfirmations: confirmations}) + mustNoErr(t, err) + return r +} + +// resolveN drives the resolver n times against the same verdicts, which is how +// the confirmation counter is exercised. +func resolveN(r *Resolver, n int) *Claim { + var last *Claim + for i := 0; i < n; i++ { + last = r.Resolve(context.Background()) + } + return last +} + +func TestNew_Validation(t *testing.T) { + probe := staticProbe(nil) + _, err := New(nil, probe, Options{}) + assert.Error(t, err, "a resolver with no candidates can never resolve anything") + + _, err = New(candidates(), nil, Options{}) + assert.Error(t, err, "a resolver with no probe can never resolve anything") + + _, err = New([]HubCandidate{{Name: "broken"}}, probe, Options{}) + assert.Error(t, err, "a candidate with no endpoint is unusable") + + _, err = New([]HubCandidate{hubA, {Name: "dup", Endpoint: hubA.Endpoint}}, probe, Options{}) + assert.Error(t, err, "duplicate endpoints would let one hub vote twice in the tie-break") +} + +// TestResolve_SteadyState is the ordinary case and the one most easily misread: +// the Active declares itself and the Standby's mirror repeats that same +// declaration, so BOTH hubs answer naming hub A. Two claims about one hub is a +// unanimous result, not a conflict. +func TestResolve_SteadyState(t *testing.T) { + r := newResolver(t, staticProbe(map[string]Verdict{ + hubA.Endpoint: declares(hubA, hubA, "hub-a-1", 2*time.Second), + hubB.Endpoint: declares(hubB, hubA, "hub-a-1", 3*time.Second), + }), 1) + + got := r.Resolve(context.Background()) + mustClaim(t, got, "") + assert.Equal(t, "hub-a-1", got.Identity) + assert.Equal(t, hubA.Endpoint, got.Endpoint) +} + +// TestResolve_OnlyTheActiveIsReachable covers failover's own shape: the Active +// is gone and only the promoted hub answers, naming itself. +func TestResolve_OnlyTheActiveIsReachable(t *testing.T) { + r := newResolver(t, staticProbe(map[string]Verdict{ + hubA.Endpoint: unreachable(hubA), + hubB.Endpoint: declares(hubB, hubB, "hub-b-1", time.Second), + }), 1) + + got := r.Resolve(context.Background()) + mustClaim(t, got, "") + assert.Equal(t, "hub-b-1", got.Identity) + assert.Equal(t, hubB.Endpoint, got.Endpoint) +} + +// TestResolve_DisagreementPrefersTheFresherDeclaration is the recovered-old- +// Active case: both hubs are up and each names itself. Not split-brain +// resolution, which is an explicit non-goal — just a single-valued answer. +func TestResolve_DisagreementPrefersTheFresherDeclaration(t *testing.T) { + r := newResolver(t, staticProbe(map[string]Verdict{ + hubA.Endpoint: declares(hubA, hubA, "hub-a-1", time.Hour), + hubB.Endpoint: declares(hubB, hubB, "hub-b-1", time.Second), + }), 1) + + got := r.Resolve(context.Background()) + mustClaim(t, got, "") + assert.Equal(t, "hub-b-1", got.Identity, "the newer declaration must win") +} + +// TestResolve_EqualTimestampsAreDeterministic guards the nastiest edge: if two +// equally-stale conflicting claims resolved differently between polls, the +// confirmation counter could never accumulate and the worker would never settle. +func TestResolve_EqualTimestampsAreDeterministic(t *testing.T) { + stamp := time.Now().Add(-time.Minute) + probe := staticProbe(map[string]Verdict{ + hubA.Endpoint: {Candidate: hubA, Reachable: true, + Claim: &Claim{Endpoint: hubA.Endpoint, Identity: "hub-a-1", LastUpdated: stamp}}, + hubB.Endpoint: {Candidate: hubB, Reachable: true, + Claim: &Claim{Endpoint: hubB.Endpoint, Identity: "hub-b-1", LastUpdated: stamp}}, + }) + + first := newResolver(t, probe, 1).Resolve(context.Background()) + mustClaim(t, first, "") + for i := 0; i < 20; i++ { + again := newResolver(t, probe, 1).Resolve(context.Background()) + mustClaim(t, again, "") + assert.Equal(t, first.Identity, again.Identity, + "an equal-timestamp tie must resolve the same way every time") + } +} + +// TestResolve_RejectsAnUnconfiguredEndpoint pins the trust boundary. Without +// it, write access to one Cluster CR would be enough to point a worker's hub +// connection at any address at all. +func TestResolve_RejectsAnUnconfiguredEndpoint(t *testing.T) { + rogue := HubCandidate{Name: "rogue", Endpoint: "https://attacker.example:6443"} + r := newResolver(t, staticProbe(map[string]Verdict{ + hubA.Endpoint: declares(hubA, rogue, "rogue-1", time.Second), + hubB.Endpoint: unreachable(hubB), + }), 1) + + assert.Nil(t, r.Resolve(context.Background()), + "a claim naming an endpoint nobody configured must never be followed") +} + +// TestResolve_NoClaimsLeavesTheCallerAlone covers the two ways a worker ends up +// with nothing to act on. Both must mean "change nothing" rather than +// "disconnect", or a hub blip becomes a worker outage. +func TestResolve_NoClaimsLeavesTheCallerAlone(t *testing.T) { + t.Run("non-HA hubs publish nothing", func(t *testing.T) { + r := newResolver(t, staticProbe(map[string]Verdict{ + hubA.Endpoint: silent(hubA), + hubB.Endpoint: silent(hubB), + }), 1) + assert.Nil(t, r.Resolve(context.Background())) + assert.Nil(t, r.Current(), "the non-HA path must stay a clean no-op") + }) + + t.Run("neither hub is reachable", func(t *testing.T) { + r := newResolver(t, staticProbe(map[string]Verdict{ + hubA.Endpoint: unreachable(hubA), + hubB.Endpoint: unreachable(hubB), + }), 1) + assert.Nil(t, r.Resolve(context.Background())) + }) +} + +// TestResolve_TotalOutageKeepsTheEstablishedWinner is the one that matters most +// operationally: once a worker knows who the Active is, losing sight of both +// hubs must not retract that. Otherwise every network hiccup would tear down a +// working connection. +func TestResolve_TotalOutageKeepsTheEstablishedWinner(t *testing.T) { + verdicts := map[string]Verdict{ + hubA.Endpoint: declares(hubA, hubA, "hub-a-1", time.Second), + hubB.Endpoint: declares(hubB, hubA, "hub-a-1", time.Second), + } + r := newResolver(t, func(_ context.Context, c HubCandidate) Verdict { return verdicts[c.Endpoint] }, 1) + mustClaim(t, r.Resolve(context.Background()), "") + + verdicts[hubA.Endpoint] = unreachable(hubA) + verdicts[hubB.Endpoint] = unreachable(hubB) + + got := r.Resolve(context.Background()) + mustClaim(t, got, "a total outage must not retract a known Active") + assert.Equal(t, "hub-a-1", got.Identity) +} + +// TestResolve_RequiresConsecutiveConfirmations is the anti-flap rule. A single +// divergent poll must not move a worker's hub connection. +func TestResolve_RequiresConsecutiveConfirmations(t *testing.T) { + verdicts := map[string]Verdict{ + hubA.Endpoint: declares(hubA, hubA, "hub-a-1", time.Second), + hubB.Endpoint: declares(hubB, hubA, "hub-a-1", time.Second), + } + r := newResolver(t, func(_ context.Context, c HubCandidate) Verdict { return verdicts[c.Endpoint] }, 3) + + assert.Nil(t, resolveN(r, 2), "even the first winner must earn its confirmations") + got := resolveN(r, 1) + mustClaim(t, got, "") + assert.Equal(t, "hub-a-1", got.Identity) + + // hub B takes over. + verdicts[hubA.Endpoint] = unreachable(hubA) + verdicts[hubB.Endpoint] = declares(hubB, hubB, "hub-b-1", time.Second) + + for i := 1; i < 3; i++ { + got = r.Resolve(context.Background()) + assert.Equal(t, "hub-a-1", got.Identity, + "poll %d of 3: the switch must not be reported before it is confirmed", i) + } + got = r.Resolve(context.Background()) + assert.Equal(t, "hub-b-1", got.Identity, "the switch lands on the confirming poll") +} + +// TestResolve_AFlappingChallengerNeverWins pins that confirmations must be +// CONSECUTIVE. A hub that wins every other poll is exactly the instability the +// counter exists to absorb, and a cumulative counter would eventually let it +// through. +func TestResolve_AFlappingChallengerNeverWins(t *testing.T) { + verdicts := map[string]Verdict{ + hubA.Endpoint: declares(hubA, hubA, "hub-a-1", time.Second), + hubB.Endpoint: declares(hubB, hubA, "hub-a-1", time.Second), + } + r := newResolver(t, func(_ context.Context, c HubCandidate) Verdict { return verdicts[c.Endpoint] }, 3) + mustClaim(t, resolveN(r, 3), "") + + challenger := map[string]Verdict{ + hubA.Endpoint: unreachable(hubA), + hubB.Endpoint: declares(hubB, hubB, "hub-b-1", time.Second), + } + incumbent := map[string]Verdict{ + hubA.Endpoint: declares(hubA, hubA, "hub-a-1", time.Second), + hubB.Endpoint: declares(hubB, hubA, "hub-a-1", time.Second), + } + for i := 0; i < 10; i++ { + for k, v := range challenger { + verdicts[k] = v + } + r.Resolve(context.Background()) + for k, v := range incumbent { + verdicts[k] = v + } + got := r.Resolve(context.Background()) + assert.Equal(t, "hub-a-1", got.Identity, + "an alternating challenger must never accumulate its way to a switch") + } +} + +// TestResolve_RepublicationDoesNotResetConfirmations guards a subtle way the +// anti-flap counter could deadlock: the Active republishes on a timer, so +// LastUpdated differs on every poll. If sameTarget compared it, no challenger +// would ever agree with itself twice and no switch could ever be confirmed. +func TestResolve_RepublicationDoesNotResetConfirmations(t *testing.T) { + age := 10 * time.Second + r := newResolver(t, func(_ context.Context, c HubCandidate) Verdict { + age -= time.Second // every poll sees a fresher declaration + if c.Endpoint == hubA.Endpoint { + return unreachable(hubA) + } + return declares(hubB, hubB, "hub-b-1", age) + }, 3) + + got := resolveN(r, 3) + mustClaim(t, got, "a steadily-republishing hub must still confirm") + assert.Equal(t, "hub-b-1", got.Identity) +} + +// TestResolve_ProbeRespectsContext checks the resolver passes its context down +// and does not swallow cancellation, so a shutting-down worker is not held up +// by a hub that stopped answering. +func TestResolve_ProbeRespectsContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + seen := false + r := newResolver(t, func(probeCtx context.Context, c HubCandidate) Verdict { + seen = true + assert.Error(t, probeCtx.Err(), "the caller's context must reach the probe") + return unreachable(c) + }, 1) + + assert.Nil(t, r.Resolve(ctx)) + assert.True(t, seen, "the probe must actually have been called") +} + +// mustNoErr and mustClaim give the fail-fast behaviour testify's require would, +// without importing it: only testify/assert and testify/mock are vendored here, +// and adding a package to vendor/ for two helpers is not worth the diff. +func mustNoErr(t *testing.T, err error) { + t.Helper() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } +} + +func mustClaim(t *testing.T, got *Claim, msg string) *Claim { + t.Helper() + if got == nil { + if msg == "" { + msg = "expected a resolved active hub, got none" + } + t.Fatal(msg) + } + return got +} From ec23cccf96ff9397a0dbd6477c276f8ad6e7dc75 Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Sun, 2 Aug 2026 13:35:16 +0530 Subject: [PATCH 02/10] refactor(hub): pass the hub connection instead of reading the environment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The endpoint and the credentials for a hub are only valid as a pair, but they were reaching the client builders by two different routes. HUB_HOST_ENDPOINT was read inside NewHubClientConfig and manager.Start, while HubTokenFile and HubCAFile are package-level vars evaluated before main runs. That difference does not matter with one hub and cannot be worked around with two. Overriding the environment from main would move the address without moving the token, producing a client aimed at one hub authenticating as the other — which fails as a TLS or authorization error and reads like a network fault. Both now take a hub.Connection carrying all three together. Callers with a single hub pass PrimaryConnection(), built from the same environment lookups as before, so the resulting rest.Config is identical field for field. Also exports resolver.Prober, which was unexported and therefore unnameable by the first package to build one. Left alone deliberately: the hub manager's webhook server still takes its Host from the HubEndpoint package var. It registers no webhooks, so the server never starts and the value is dead config; changing it here would mean altering behaviour in a commit whose whole point is not to. Part of #467 Signed-off-by: Sumanth D --- pkg/hub/hubclient/connection.go | 62 ++++++++++++++++++++++++++++ pkg/hub/hubclient/connection_test.go | 56 +++++++++++++++++++++++++ pkg/hub/hubclient/hubclient.go | 14 +++---- pkg/hub/manager/manager.go | 17 ++++---- pkg/hub/resolver/probe.go | 4 +- pkg/hub/resolver/probe_test.go | 2 +- pkg/hub/resolver/resolver.go | 8 ++-- pkg/hub/resolver/resolver_test.go | 4 +- 8 files changed, 140 insertions(+), 27 deletions(-) create mode 100644 pkg/hub/hubclient/connection.go create mode 100644 pkg/hub/hubclient/connection_test.go diff --git a/pkg/hub/hubclient/connection.go b/pkg/hub/hubclient/connection.go new file mode 100644 index 000000000..a5ce6cc0c --- /dev/null +++ b/pkg/hub/hubclient/connection.go @@ -0,0 +1,62 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 hub + +import "k8s.io/client-go/rest" + +// Connection is everything needed to reach one hub cluster: where it is, and +// the credentials for it. It travels together for a reason — the endpoint and +// the token are only valid as a pair, and a client built from one hub's address +// with the other hub's token fails authentication in a way that looks like a +// network problem. +// +// It is passed by value rather than read from the environment at the point of +// use, which matters once more than one hub is configured. The endpoint is read +// late (inside the functions that build a client) but HubTokenFile and +// HubCAFile are package-level vars initialised before main runs, so overriding +// the environment from main would move the address without moving the +// credentials. Passing the pair removes that whole class of mistake. +type Connection struct { + Endpoint string + TokenFile string + CAFile string +} + +// PrimaryConnection is the hub this worker was configured with, from the +// environment the deployment has always set. It is what a worker with no +// secondary hub configured uses, and is identical to what the client-building +// code read directly before connections were passed around. +func PrimaryConnection() Connection { + return Connection{ + Endpoint: HubEndpoint, + TokenFile: HubTokenFile, + CAFile: HubCAFile, + } +} + +// RestConfig builds the client-go config for this hub. +func (c Connection) RestConfig() *rest.Config { + return &rest.Config{ + Host: c.Endpoint, + BearerTokenFile: c.TokenFile, + TLSClientConfig: rest.TLSClientConfig{ + CAFile: c.CAFile, + }, + } +} diff --git a/pkg/hub/hubclient/connection_test.go b/pkg/hub/hubclient/connection_test.go new file mode 100644 index 000000000..85aa83295 --- /dev/null +++ b/pkg/hub/hubclient/connection_test.go @@ -0,0 +1,56 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 hub + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +// TestPrimaryConnection_MatchesTheConfiguredEnvironment is the no-regression +// test for passing connections around instead of reading the environment at the +// point of use: a worker with one hub must end up with exactly the endpoint and +// credentials it has always used. +func TestPrimaryConnection_MatchesTheConfiguredEnvironment(t *testing.T) { + conn := PrimaryConnection() + assert.Equal(t, HubEndpoint, conn.Endpoint) + assert.Equal(t, HubTokenFile, conn.TokenFile) + assert.Equal(t, HubCAFile, conn.CAFile) +} + +// TestConnection_RestConfigCarriesTheWholeTriple guards the mistake the type +// exists to prevent. The endpoint used to be read late, inside the client +// builders, while HubTokenFile and HubCAFile are package vars fixed before main +// runs — so moving only the endpoint produced a client aimed at one hub +// authenticating as the other, which fails looking like a network problem. +func TestConnection_RestConfigCarriesTheWholeTriple(t *testing.T) { + conn := Connection{ + Endpoint: "https://hub-b.example:6443", + TokenFile: "/creds/b/token", + CAFile: "/creds/b/ca.crt", + } + + cfg := conn.RestConfig() + assert.Equal(t, conn.Endpoint, cfg.Host) + assert.Equal(t, conn.TokenFile, cfg.BearerTokenFile, + "the token must follow the endpoint it belongs to") + assert.Equal(t, conn.CAFile, cfg.TLSClientConfig.CAFile, + "the CA must follow the endpoint it belongs to") +} diff --git a/pkg/hub/hubclient/hubclient.go b/pkg/hub/hubclient/hubclient.go index 272a558ea..f8e871dec 100644 --- a/pkg/hub/hubclient/hubclient.go +++ b/pkg/hub/hubclient/hubclient.go @@ -31,7 +31,6 @@ import ( "k8s.io/apimachinery/pkg/types" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" - "k8s.io/client-go/rest" "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" @@ -76,13 +75,12 @@ type HubClientRpc interface { UpdateLBIPsForSliceGwServer(ctx context.Context, lbIP []string, sliceGwName string) error } -func NewHubClientConfig(er *monitoring.EventRecorder) (*HubClientConfig, error) { - hubClient, err := client.New(&rest.Config{ - Host: os.Getenv("HUB_HOST_ENDPOINT"), - BearerTokenFile: HubTokenFile, - TLSClientConfig: rest.TLSClientConfig{ - CAFile: HubCAFile, - }}, +// NewHubClientConfig builds the uncached hub client. conn says which hub to +// reach and with what; pass PrimaryConnection() for the single-hub case, which +// is byte-for-byte the configuration this function used to read from the +// environment itself. +func NewHubClientConfig(er *monitoring.EventRecorder, conn Connection) (*HubClientConfig, error) { + hubClient, err := client.New(conn.RestConfig(), client.Options{ Scheme: scheme, }, diff --git a/pkg/hub/manager/manager.go b/pkg/hub/manager/manager.go index 47410fca4..99cb97741 100644 --- a/pkg/hub/manager/manager.go +++ b/pkg/hub/manager/manager.go @@ -26,7 +26,6 @@ import ( "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" - "k8s.io/client-go/rest" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/cache" "sigs.k8s.io/controller-runtime/pkg/client" @@ -47,6 +46,7 @@ import ( "github.com/kubeslice/worker-operator/pkg/hub/controllers" hubCluster "github.com/kubeslice/worker-operator/pkg/hub/controllers/cluster" "github.com/kubeslice/worker-operator/pkg/hub/controllers/vpnkeyrotation" + hub "github.com/kubeslice/worker-operator/pkg/hub/hubclient" "github.com/kubeslice/worker-operator/pkg/hub/controllers/workerslicegwrecycler" "github.com/kubeslice/worker-operator/pkg/logger" @@ -66,18 +66,15 @@ func init() { utilruntime.Must(hubv1alpha1.AddToScheme(scheme)) } -func Start(meshClient client.Client, hubClient client.Client, ctx context.Context) { - config := &rest.Config{ - Host: os.Getenv("HUB_HOST_ENDPOINT"), - BearerTokenFile: HubTokenFile, - TLSClientConfig: rest.TLSClientConfig{ - CAFile: HubCAFile, - }, - } +// Start runs the hub-side manager against the hub described by conn. Callers +// with a single hub pass hub.PrimaryConnection(), which is the same endpoint +// and credentials this function used to assemble from the environment itself. +func Start(meshClient client.Client, hubClient client.Client, ctx context.Context, conn hub.Connection) { + config := conn.RestConfig() var log = log.Log.WithName("hub") - log.Info("Connecting to hub cluster", "endpoint", HubEndpoint, "ns", ProjectNamespace) + log.Info("Connecting to hub cluster", "endpoint", conn.Endpoint, "ns", ProjectNamespace) webhookServer := webhook.NewServer(webhook.Options{ Host: HubEndpoint, diff --git a/pkg/hub/resolver/probe.go b/pkg/hub/resolver/probe.go index bd2052d5f..d17dedad2 100644 --- a/pkg/hub/resolver/probe.go +++ b/pkg/hub/resolver/probe.go @@ -73,7 +73,7 @@ type ProbeConfig struct { Timeout time.Duration } -// NewProbe returns a prober that reads each candidate's copy of this worker's +// NewProbe returns a Prober that reads each candidate's copy of this worker's // Cluster CR through the reader built for that candidate by readerFor. // // readerFor is a function rather than a prepared map because a candidate's @@ -82,7 +82,7 @@ type ProbeConfig struct { // the intent: these probes must keep working across a failover, so they are // deliberately independent of the worker's primary hub connection, which is the // thing a failover replaces. -func NewProbe(readerFor func(HubCandidate) (ClusterReader, error), cfg ProbeConfig) prober { +func NewProbe(readerFor func(HubCandidate) (ClusterReader, error), cfg ProbeConfig) Prober { if cfg.Timeout <= 0 { cfg.Timeout = DefaultProbeTimeout } diff --git a/pkg/hub/resolver/probe_test.go b/pkg/hub/resolver/probe_test.go index 4ed18f8d7..2de4c55fd 100644 --- a/pkg/hub/resolver/probe_test.go +++ b/pkg/hub/resolver/probe_test.go @@ -56,7 +56,7 @@ func clusterWith(activeController map[string]interface{}) *unstructured.Unstruct } // probeReturning builds a probe whose single reader serves obj (or err). -func probeReturning(t *testing.T, obj *unstructured.Unstructured, err error, timeout time.Duration) prober { +func probeReturning(t *testing.T, obj *unstructured.Unstructured, err error, timeout time.Duration) Prober { t.Helper() reader := readerFunc(func(ctx context.Context, key types.NamespacedName, into *unstructured.Unstructured) error { assert.Equal(t, testCluster, key.Name, "the probe must read this worker's own Cluster CR") diff --git a/pkg/hub/resolver/resolver.go b/pkg/hub/resolver/resolver.go index d1d5afa8d..3d5d89d1d 100644 --- a/pkg/hub/resolver/resolver.go +++ b/pkg/hub/resolver/resolver.go @@ -99,10 +99,10 @@ type Verdict struct { Err error } -// prober reads one candidate's copy of this worker's Cluster CR. It is a field +// Prober reads one candidate's copy of this worker's Cluster CR. It is a field // on Resolver rather than a hardcoded call so the decision logic can be tested // without a cluster, and so a caller can supply a client built however it likes. -type prober func(ctx context.Context, candidate HubCandidate) Verdict +type Prober func(ctx context.Context, candidate HubCandidate) Verdict // Options configures a Resolver. Zero-valued fields fall back to the Default* // constants. @@ -120,7 +120,7 @@ type Options struct { // suppresses flapping, and it is meant to be driven by a single polling loop. type Resolver struct { candidates []HubCandidate - probe prober + probe Prober confirmations int log logr.Logger @@ -133,7 +133,7 @@ type Resolver struct { } // New builds a Resolver over a fixed candidate set. -func New(candidates []HubCandidate, probe prober, opts Options) (*Resolver, error) { +func New(candidates []HubCandidate, probe Prober, opts Options) (*Resolver, error) { if len(candidates) == 0 { return nil, fmt.Errorf("resolver requires at least one hub candidate") } diff --git a/pkg/hub/resolver/resolver_test.go b/pkg/hub/resolver/resolver_test.go index 43095a8a4..70e4202c5 100644 --- a/pkg/hub/resolver/resolver_test.go +++ b/pkg/hub/resolver/resolver_test.go @@ -56,7 +56,7 @@ func unreachable(c HubCandidate) Verdict { } // staticProbe answers each candidate from a fixed table. -func staticProbe(verdicts map[string]Verdict) prober { +func staticProbe(verdicts map[string]Verdict) Prober { return func(_ context.Context, c HubCandidate) Verdict { if v, ok := verdicts[c.Endpoint]; ok { return v @@ -65,7 +65,7 @@ func staticProbe(verdicts map[string]Verdict) prober { } } -func newResolver(t *testing.T, probe prober, confirmations int) *Resolver { +func newResolver(t *testing.T, probe Prober, confirmations int) *Resolver { t.Helper() r, err := New(candidates(), probe, Options{SwitchConfirmations: confirmations}) mustNoErr(t, err) From 523b57ee5143e59c27bd3967a57dcce621007ce4 Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Sun, 2 Aug 2026 13:35:33 +0530 Subject: [PATCH 03/10] feat(hub): follow a controller failover to the promoted hub MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A worker pinned to one hub endpoint cannot survive that hub being promoted away from. With the controller running Active/Standby, leadership can move, and until now the only way to point a worker at the new Active was to edit its deployment. The worker now resolves which hub holds leadership before it opens any connection, and keeps watching. When the answer changes and holds across consecutive polls, it logs, counts it, and shuts down cleanly; the kubelet restarts it and startup resolution picks the hub that is now Active. Restarting rather than rebuilding in place is the deliberate choice. Both hub connections are assembled once from a rest.Config, and manager.Start already exits the process on any hub error, so a clean restart is both the smaller change and the one this process is built for. The data plane is untouched either way: gateways and tunnels run in their own pods. Everything is gated on HUB_SECONDARY_HOST_ENDPOINT. Unset, which is every deployment today, no resolver is built, no extra client is opened, and the connection is the same one the environment has always described. The resolution rule refuses claims naming any endpoint outside the two configured hubs, so a Cluster CR cannot redirect a worker somewhere nobody provisioned. Two metrics: kubeslice_worker_hub_switches_total, and kubeslice_worker_hub_probe_errors_total by hub slot. The second is the one worth alerting on — a hub that has been quietly unreachable for days is a problem to hear about before a failover rather than during one. Not implemented: issue #467 asks for a ControllerConnected condition. ClusterStatus has no Conditions field, so that needs a further change to the shared apis module, and a hub-side condition can only be written while the hub is reachable — it can never report the state anyone wants to see. Local metrics and logs carry it instead. A durable local health surface belongs with #469. Part of #467 Signed-off-by: Sumanth D --- main.go | 42 ++++- pkg/hub/failover/failover.go | 280 +++++++++++++++++++++++++++++ pkg/hub/failover/failover_test.go | 287 ++++++++++++++++++++++++++++++ 3 files changed, 606 insertions(+), 3 deletions(-) create mode 100644 pkg/hub/failover/failover.go create mode 100644 pkg/hub/failover/failover_test.go diff --git a/main.go b/main.go index a95f367e9..69a2d79ed 100644 --- a/main.go +++ b/main.go @@ -19,6 +19,7 @@ package main import ( + "context" "flag" "os" "strings" @@ -65,8 +66,10 @@ import ( "github.com/kubeslice/worker-operator/controllers/slice" "github.com/kubeslice/worker-operator/controllers/slicegateway" ossEvents "github.com/kubeslice/worker-operator/events" + "github.com/kubeslice/worker-operator/pkg/hub/failover" hub "github.com/kubeslice/worker-operator/pkg/hub/hubclient" "github.com/kubeslice/worker-operator/pkg/hub/manager" + "github.com/kubeslice/worker-operator/pkg/hub/resolver" "github.com/kubeslice/worker-operator/pkg/logger" "github.com/kubeslice/worker-operator/pkg/networkpolicy" "github.com/kubeslice/worker-operator/pkg/utils" @@ -165,7 +168,22 @@ func main() { //view.SetReportingPeriod(10 * time.Millisecond) } - hubClient, err := hub.NewHubClientConfig(er) + // Decide which hub to talk to before any client is built. Inert unless + // HUB_SECONDARY_HOST_ENDPOINT is set, in which case hubConn is exactly the + // primary connection this worker has always used. + hubConn := hub.PrimaryConnection() + failoverCfg := failover.ConfigFromEnv() + var hubFollower *failover.Follower + if failoverCfg.Enabled() { + hubFollower, err = failover.New(failoverCfg, hubConn, ctrl.Log.WithName("hub-failover"), nil) + if err != nil { + setupLog.With("error", err).Error("could not configure hub failover following") + os.Exit(1) + } + hubConn = hubFollower.StartupConnection(context.Background()) + } + + hubClient, err := hub.NewHubClientConfig(er, hubConn) if err != nil { setupLog.With("error", err).Error("could not create hub client for slice gateway reconciler") os.Exit(1) @@ -193,7 +211,11 @@ func main() { Scheme: scheme, }) - ctx := ctrl.SetupSignalHandler() + // Cancellable so a resolved hub failover can shut this process down the same + // way a signal would: the manager drains, the process exits 0, and the + // kubelet restarts it against the hub that is now Active. + ctx, stopForHubSwitch := context.WithCancel(ctrl.SetupSignalHandler()) + defer stopForHubSwitch() mf, err := metrics.NewMetricsFactory(ctrlmetrics.Registry, metrics.MetricsFactoryOptions{ Cluster: controllers.ClusterName, @@ -310,9 +332,23 @@ func main() { } go func() { setupLog.Info("starting hub manager") - manager.Start(clientForHubMgr, hubClient, ctx) + manager.Start(clientForHubMgr, hubClient, ctx, hubConn) }() + if hubFollower != nil { + go hubFollower.Watch(ctx, hubConn, func(claim resolver.Claim) { + // Restart rather than rebuild. Both hub connections are assembled + // once during startup from a rest.Config, and manager.Start exits + // the process on any hub error already, so a clean restart is both + // the smaller change and the one this process is already built for. + // The data plane is untouched: gateways and tunnels run in their + // own pods. + setupLog.With("endpoint", claim.Endpoint, "identity", claim.Identity). + Info("active hub changed; shutting down to reconnect") + stopForHubSwitch() + }) + } + setupLog.Info("starting manager") if err := mgr.Start(ctx); err != nil { setupLog.With("error", err).Error("problem running manager") diff --git a/pkg/hub/failover/failover.go b/pkg/hub/failover/failover.go new file mode 100644 index 000000000..fa39a1203 --- /dev/null +++ b/pkg/hub/failover/failover.go @@ -0,0 +1,280 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 failover connects pkg/hub/resolver to the worker's actual hub +// connections: it decides which hub to start against, and notices when that +// answer changes. +// +// Everything here is inert unless HUB_SECONDARY_HOST_ENDPOINT is set. A worker +// with one hub configured — every deployment that exists today — builds no +// resolver, opens no extra clients, and connects exactly as it always has. +package failover + +import ( + "context" + "fmt" + "strconv" + "time" + + "github.com/go-logr/logr" + "github.com/prometheus/client_golang/prometheus" + "sigs.k8s.io/controller-runtime/pkg/client" + ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" + + hub "github.com/kubeslice/worker-operator/pkg/hub/hubclient" + "github.com/kubeslice/worker-operator/pkg/hub/resolver" + "github.com/kubeslice/worker-operator/pkg/utils" +) + +// Candidate names used in logs and metric labels. They identify the configured +// slot, not the role a hub currently holds — which hub is Active is exactly the +// thing being resolved, and may change while the process runs. +const ( + primaryName = "primary" + secondaryName = "secondary" +) + +const ( + defaultSecondaryTokenFile = "/var/run/secrets/kubernetes.io/hub-secondary-serviceaccount/token" + defaultSecondaryCAFile = "/var/run/secrets/kubernetes.io/hub-secondary-serviceaccount/ca.crt" + defaultInterval = 10 * time.Second + defaultConfirmations = 2 +) + +var ( + hubSwitchesTotal = prometheus.NewCounter(prometheus.CounterOpts{ + Name: "kubeslice_worker_hub_switches_total", + Help: "Number of times this worker resolved a different active hub and restarted to follow it.", + }) + hubProbeErrorsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "kubeslice_worker_hub_probe_errors_total", + Help: "Failed reads of a hub's copy of this worker's Cluster CR, by configured hub slot.", + }, []string{"hub"}) +) + +func init() { + ctrlmetrics.Registry.MustRegister(hubSwitchesTotal, hubProbeErrorsTotal) +} + +// Config is the failover-following configuration, read from the environment. +type Config struct { + SecondaryEndpoint string + SecondaryTokenFile string + SecondaryCAFile string + ClusterName string + Namespace string + Interval time.Duration + Timeout time.Duration + Confirmations int +} + +// ConfigFromEnv reads the configuration. Only SecondaryEndpoint has no usable +// default: without a second hub to compare against there is nothing to resolve, +// which is why it doubles as the feature's on switch. +func ConfigFromEnv() Config { + return Config{ + SecondaryEndpoint: utils.GetEnvOrDefault("HUB_SECONDARY_HOST_ENDPOINT", ""), + SecondaryTokenFile: utils.GetEnvOrDefault("HUB_SECONDARY_TOKEN_FILE", defaultSecondaryTokenFile), + SecondaryCAFile: utils.GetEnvOrDefault("HUB_SECONDARY_CA_FILE", defaultSecondaryCAFile), + ClusterName: hub.ClusterName, + Namespace: hub.ProjectNamespace, + Interval: durationFromEnv("HUB_RESOLVE_INTERVAL", defaultInterval), + Timeout: durationFromEnv("HUB_RESOLVE_TIMEOUT", resolver.DefaultProbeTimeout), + Confirmations: intFromEnv("HUB_SWITCH_CONFIRMATIONS", defaultConfirmations), + } +} + +// Enabled reports whether a second hub is configured. Everything in this +// package is a no-op when it is not. +func (c Config) Enabled() bool { return c.SecondaryEndpoint != "" } + +func durationFromEnv(key string, fallback time.Duration) time.Duration { + raw := utils.GetEnvOrDefault(key, "") + if raw == "" { + return fallback + } + // An unparseable value falls back rather than failing: this is a tuning + // knob, and refusing to start over a malformed one would be a worse + // outcome than running at the default. + parsed, err := time.ParseDuration(raw) + if err != nil || parsed <= 0 { + return fallback + } + return parsed +} + +func intFromEnv(key string, fallback int) int { + raw := utils.GetEnvOrDefault(key, "") + if raw == "" { + return fallback + } + parsed, err := strconv.Atoi(raw) + if err != nil || parsed <= 0 { + return fallback + } + return parsed +} + +// Follower resolves which hub is Active and reports when that changes. +type Follower struct { + resolver *resolver.Resolver + byEndpoint map[string]hub.Connection + primary hub.Connection + interval time.Duration + log logr.Logger +} + +// New builds a Follower over the primary hub and the configured secondary. +// +// readerFor exists so tests can supply their own hub readers; production passes +// nil and gets real clients. The clients it builds are opened once and never +// rebuilt: they have to keep working across a failover, which is precisely what +// makes them useless if they are tied to the connection a failover replaces. +func New(cfg Config, primary hub.Connection, log logr.Logger, + readerFor func(hub.Connection) (resolver.ClusterReader, error)) (*Follower, error) { + + if !cfg.Enabled() { + return nil, fmt.Errorf("no secondary hub configured") + } + secondary := hub.Connection{ + Endpoint: cfg.SecondaryEndpoint, + TokenFile: cfg.SecondaryTokenFile, + CAFile: cfg.SecondaryCAFile, + } + if secondary.Endpoint == primary.Endpoint { + return nil, fmt.Errorf("the secondary hub endpoint %q is the same as the primary", secondary.Endpoint) + } + if readerFor == nil { + readerFor = newClusterReader + } + + byEndpoint := map[string]hub.Connection{ + primary.Endpoint: primary, + secondary.Endpoint: secondary, + } + candidates := []resolver.HubCandidate{ + {Name: primaryName, Endpoint: primary.Endpoint, TokenFile: primary.TokenFile, CAFile: primary.CAFile}, + {Name: secondaryName, Endpoint: secondary.Endpoint, TokenFile: secondary.TokenFile, CAFile: secondary.CAFile}, + } + + probe := resolver.NewProbe( + func(c resolver.HubCandidate) (resolver.ClusterReader, error) { + return readerFor(hub.Connection{Endpoint: c.Endpoint, TokenFile: c.TokenFile, CAFile: c.CAFile}) + }, + resolver.ProbeConfig{ClusterName: cfg.ClusterName, Namespace: cfg.Namespace, Timeout: cfg.Timeout}, + ) + // Count probe failures without changing the verdict: an unreachable hub is + // already handled by the rule, but a hub that is quietly unreachable for + // days is the thing an operator needs to see before a failover, not after. + counted := func(ctx context.Context, c resolver.HubCandidate) resolver.Verdict { + v := probe(ctx, c) + if !v.Reachable { + hubProbeErrorsTotal.WithLabelValues(c.Name).Inc() + } + return v + } + + r, err := resolver.New(candidates, counted, resolver.Options{ + SwitchConfirmations: cfg.Confirmations, + Log: log, + }) + if err != nil { + return nil, err + } + interval := cfg.Interval + if interval <= 0 { + interval = defaultInterval + } + return &Follower{ + resolver: r, + byEndpoint: byEndpoint, + primary: primary, + interval: interval, + log: log, + }, nil +} + +// StartupConnection resolves once and returns the hub to connect to. +// +// Falling back to the primary when nothing resolves is deliberate. A worker +// that refused to start because it could not reach a hub would turn a hub +// outage into a worker outage, and the primary is the same hub it would have +// used before any of this existed. +func (f *Follower) StartupConnection(ctx context.Context) hub.Connection { + claim := f.resolver.Resolve(ctx) + if claim == nil { + f.log.Info("no active hub resolved at startup; using the configured primary", + "endpoint", f.primary.Endpoint) + return f.primary + } + conn, ok := f.byEndpoint[claim.Endpoint] + if !ok { + // Unreachable in practice: the resolver already refuses claims naming + // endpoints outside the candidate set, and the candidates are built + // from this same map. Handled anyway so a future change to either side + // cannot silently produce a connection with mismatched credentials. + f.log.Info("resolved hub is not a configured connection; using the primary", + "resolvedEndpoint", claim.Endpoint) + return f.primary + } + f.log.Info("connecting to the resolved active hub", + "identity", claim.Identity, "endpoint", conn.Endpoint) + return conn +} + +// Watch polls until ctx is done, calling onSwitch once the resolved active hub +// differs from the one this process started against. +// +// Acting on the change is the caller's business, and in practice means shutting +// the process down so it restarts and resolves again. This function does not +// exit the process itself: a package that reads state should not also decide to +// terminate, and keeping them apart is what makes this testable. +func (f *Follower) Watch(ctx context.Context, startedWith hub.Connection, onSwitch func(resolver.Claim)) { + ticker := time.NewTicker(f.interval) + defer ticker.Stop() + f.log.Info("watching for hub failover", "interval", f.interval, "connectedTo", startedWith.Endpoint) + + for { + select { + case <-ctx.Done(): + f.log.Info("hub failover watch stopped", "reason", ctx.Err()) + return + case <-ticker.C: + claim := f.resolver.Resolve(ctx) + if claim == nil || claim.Endpoint == startedWith.Endpoint { + continue + } + hubSwitchesTotal.Inc() + f.log.Info("the active hub changed; this worker must reconnect", + "from", startedWith.Endpoint, "to", claim.Endpoint, "identity", claim.Identity) + onSwitch(*claim) + return + } + } +} + +// newClusterReader opens a client to one hub. Deliberately uncached: this reads +// a single object on a timer, and a cache would add an informer, a watch and a +// resync against a hub this worker may not even be talking to. +func newClusterReader(conn hub.Connection) (resolver.ClusterReader, error) { + c, err := client.New(conn.RestConfig(), client.Options{}) + if err != nil { + return nil, err + } + return resolver.NewClusterReader(c), nil +} diff --git a/pkg/hub/failover/failover_test.go b/pkg/hub/failover/failover_test.go new file mode 100644 index 000000000..828f6de9a --- /dev/null +++ b/pkg/hub/failover/failover_test.go @@ -0,0 +1,287 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 failover + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/types" + + hub "github.com/kubeslice/worker-operator/pkg/hub/hubclient" + "github.com/kubeslice/worker-operator/pkg/hub/resolver" +) + +const ( + primaryEndpoint = "https://hub-a.example:6443" + secondaryEndpoint = "https://hub-b.example:6443" + clusterName = "worker-1" + namespace = "kubeslice-avesha" +) + +func primaryConn() hub.Connection { + return hub.Connection{Endpoint: primaryEndpoint, TokenFile: "/creds/a/token", CAFile: "/creds/a/ca.crt"} +} + +func testConfig() Config { + return Config{ + SecondaryEndpoint: secondaryEndpoint, + SecondaryTokenFile: "/creds/b/token", + SecondaryCAFile: "/creds/b/ca.crt", + ClusterName: clusterName, + Namespace: namespace, + Interval: 5 * time.Millisecond, + Timeout: time.Second, + Confirmations: 1, + } +} + +// fakeReader serves one hub's copy of the Cluster CR. +type fakeReader struct { + activeController map[string]interface{} + err error + // sawCreds records the credentials the reader was built with, so a test can + // prove the endpoint and its token travelled together. + sawCreds hub.Connection +} + +func (f *fakeReader) Get(_ context.Context, _ types.NamespacedName, obj *unstructured.Unstructured) error { + if f.err != nil { + return f.err + } + obj.Object = map[string]interface{}{} + if f.activeController != nil { + _ = unstructured.SetNestedMap(obj.Object, f.activeController, "status", "activeController") + } + return nil +} + +// declaring builds an activeController naming endpoint. +func declaring(endpoint, identity string) map[string]interface{} { + return map[string]interface{}{ + "endpoint": endpoint, + "activeIdentity": identity, + "lastUpdated": time.Now().UTC().Format(time.RFC3339), + } +} + +// readersFor wires a per-endpoint reader table into the readerFor callback, +// recording the connection each reader was constructed from. +func readersFor(table map[string]*fakeReader) func(hub.Connection) (resolver.ClusterReader, error) { + var mu sync.Mutex + return func(conn hub.Connection) (resolver.ClusterReader, error) { + mu.Lock() + defer mu.Unlock() + r, ok := table[conn.Endpoint] + if !ok { + return nil, fmt.Errorf("no reader for %s", conn.Endpoint) + } + r.sawCreds = conn + return r, nil + } +} + +func newFollower(t *testing.T, cfg Config, table map[string]*fakeReader) *Follower { + t.Helper() + f, err := New(cfg, primaryConn(), logr.Discard(), readersFor(table)) + if err != nil { + t.Fatalf("New: %v", err) + } + return f +} + +func TestConfigFromEnv_DisabledWithoutASecondaryHub(t *testing.T) { + t.Setenv("HUB_SECONDARY_HOST_ENDPOINT", "") + cfg := ConfigFromEnv() + assert.False(t, cfg.Enabled(), + "with no secondary hub there is nothing to resolve, and every existing deployment must stay on this path") +} + +func TestConfigFromEnv_DefaultsAndOverrides(t *testing.T) { + t.Setenv("HUB_SECONDARY_HOST_ENDPOINT", secondaryEndpoint) + cfg := ConfigFromEnv() + assert.True(t, cfg.Enabled()) + assert.Equal(t, defaultSecondaryTokenFile, cfg.SecondaryTokenFile) + assert.Equal(t, defaultInterval, cfg.Interval) + assert.Equal(t, defaultConfirmations, cfg.Confirmations) + + t.Setenv("HUB_RESOLVE_INTERVAL", "3s") + t.Setenv("HUB_SWITCH_CONFIRMATIONS", "5") + cfg = ConfigFromEnv() + assert.Equal(t, 3*time.Second, cfg.Interval) + assert.Equal(t, 5, cfg.Confirmations) +} + +// TestConfigFromEnv_MalformedValuesFallBack: these are tuning knobs. Refusing to +// start over a typo in one would be a worse outcome than running at the default. +func TestConfigFromEnv_MalformedValuesFallBack(t *testing.T) { + t.Setenv("HUB_SECONDARY_HOST_ENDPOINT", secondaryEndpoint) + t.Setenv("HUB_RESOLVE_INTERVAL", "not-a-duration") + t.Setenv("HUB_SWITCH_CONFIRMATIONS", "-4") + cfg := ConfigFromEnv() + assert.Equal(t, defaultInterval, cfg.Interval) + assert.Equal(t, defaultConfirmations, cfg.Confirmations) +} + +func TestNew_Rejects(t *testing.T) { + _, err := New(Config{}, primaryConn(), logr.Discard(), readersFor(nil)) + assert.Error(t, err, "a disabled config must not produce a follower") + + same := testConfig() + same.SecondaryEndpoint = primaryEndpoint + _, err = New(same, primaryConn(), logr.Discard(), readersFor(nil)) + assert.Error(t, err, "two candidates at one endpoint cannot be told apart") +} + +// TestStartupConnection_UsesTheWinnersOwnCredentials is the test this whole +// design exists for. The endpoint is read late, inside the client builders, but +// the token and CA paths are package-level vars fixed before main runs — so +// moving the endpoint alone would produce a client pointed at hub B +// authenticating as hub A, which fails in a way that looks like a network fault. +func TestStartupConnection_UsesTheWinnersOwnCredentials(t *testing.T) { + table := map[string]*fakeReader{ + primaryEndpoint: {err: fmt.Errorf("simulated: hub A is gone")}, + secondaryEndpoint: {activeController: declaring(secondaryEndpoint, "hub-b-1")}, + } + f := newFollower(t, testConfig(), table) + + conn := f.StartupConnection(context.Background()) + assert.Equal(t, secondaryEndpoint, conn.Endpoint) + assert.Equal(t, "/creds/b/token", conn.TokenFile, "the winner's token must travel with its endpoint") + assert.Equal(t, "/creds/b/ca.crt", conn.CAFile, "the winner's CA must travel with its endpoint") +} + +func TestStartupConnection_SteadyStateStaysOnThePrimary(t *testing.T) { + table := map[string]*fakeReader{ + primaryEndpoint: {activeController: declaring(primaryEndpoint, "hub-a-1")}, + secondaryEndpoint: {activeController: declaring(primaryEndpoint, "hub-a-1")}, // the mirror + } + f := newFollower(t, testConfig(), table) + + conn := f.StartupConnection(context.Background()) + assert.Equal(t, primaryConn(), conn, "both hubs naming hub A is agreement, not a conflict") +} + +// TestStartupConnection_FallsBackWhenNothingResolves: a worker that refused to +// start because it could not reach a hub would turn a hub outage into a worker +// outage. The primary is the same hub it would have used before any of this. +func TestStartupConnection_FallsBackWhenNothingResolves(t *testing.T) { + for name, table := range map[string]map[string]*fakeReader{ + "both unreachable": { + primaryEndpoint: {err: fmt.Errorf("down")}, + secondaryEndpoint: {err: fmt.Errorf("down")}, + }, + "neither publishes": { + primaryEndpoint: {}, + secondaryEndpoint: {}, + }, + } { + t.Run(name, func(t *testing.T) { + f := newFollower(t, testConfig(), table) + assert.Equal(t, primaryConn(), f.StartupConnection(context.Background())) + }) + } +} + +func TestWatch_FiresOnceWhenTheActiveHubMoves(t *testing.T) { + table := map[string]*fakeReader{ + primaryEndpoint: {err: fmt.Errorf("simulated: hub A died")}, + secondaryEndpoint: {activeController: declaring(secondaryEndpoint, "hub-b-1")}, + } + f := newFollower(t, testConfig(), table) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + var mu sync.Mutex + var calls []resolver.Claim + f.Watch(ctx, primaryConn(), func(c resolver.Claim) { + mu.Lock() + defer mu.Unlock() + calls = append(calls, c) + }) + + mu.Lock() + defer mu.Unlock() + if len(calls) != 1 { + t.Fatalf("expected exactly one switch notification, got %d", len(calls)) + } + assert.Equal(t, secondaryEndpoint, calls[0].Endpoint) + assert.Equal(t, "hub-b-1", calls[0].Identity) +} + +// TestWatch_QuietWhileTheActiveHubIsUnchanged: the steady state must produce no +// notification at all, or the worker would restart itself on a timer. +func TestWatch_QuietWhileTheActiveHubIsUnchanged(t *testing.T) { + table := map[string]*fakeReader{ + primaryEndpoint: {activeController: declaring(primaryEndpoint, "hub-a-1")}, + secondaryEndpoint: {activeController: declaring(primaryEndpoint, "hub-a-1")}, + } + f := newFollower(t, testConfig(), table) + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Millisecond) + defer cancel() + + switched := false + f.Watch(ctx, primaryConn(), func(resolver.Claim) { switched = true }) + assert.False(t, switched, "an unchanged active hub must never trigger a reconnect") +} + +// TestWatch_SilentHubsDoNotTriggerARestart is the non-HA safety net: if the +// field is absent everywhere, the worker must sit still rather than restart. +func TestWatch_SilentHubsDoNotTriggerARestart(t *testing.T) { + f := newFollower(t, testConfig(), map[string]*fakeReader{ + primaryEndpoint: {}, + secondaryEndpoint: {}, + }) + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Millisecond) + defer cancel() + + switched := false + f.Watch(ctx, primaryConn(), func(resolver.Claim) { switched = true }) + assert.False(t, switched) +} + +func TestWatch_StopsWithItsContext(t *testing.T) { + f := newFollower(t, testConfig(), map[string]*fakeReader{ + primaryEndpoint: {}, + secondaryEndpoint: {}, + }) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + f.Watch(ctx, primaryConn(), func(resolver.Claim) {}) + close(done) + }() + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Watch did not return when its context was cancelled") + } +} From 0d9d4e3d36e247e4b13dea7b5ead83cb84c34d0c Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Sun, 2 Aug 2026 13:35:47 +0530 Subject: [PATCH 04/10] docs(hub): document following a failover, and the credential it needs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Describes the resolution rule, the configuration, the metrics, and how to verify the whole thing in Kind. config/manager/manager.yaml gains the env, volume mount and volume for the second hub, commented out, so the shape is visible where an operator would look for it. The part worth being explicit about is the second credential. A worker authenticates with a token the hub it talks to minted, so following a failover needs a Standby-valid credential mounted before the Active fails — afterwards nothing is left to hand it one. The controller side already produces it: the Standby mirrors the worker's ServiceAccount and an empty token Secret shell, and its own token controller fills that shell in. Getting it from there onto the worker belongs to cluster registration and the charts, which live in neither repository, so it currently has no owner. The manual procedure is written out rather than left implied, including a check that the token actually authenticates before it is installed — a credential that is present but invalid is worse than none, because it fails only at failover. Part of #467 Signed-off-by: Sumanth D --- config/manager/manager.yaml | 22 +++++++ docs/hub-failover.md | 119 ++++++++++++++++++++++++++++++++++++ 2 files changed, 141 insertions(+) create mode 100644 docs/hub-failover.md diff --git a/config/manager/manager.yaml b/config/manager/manager.yaml index fd01ab1ea..6ef0a614b 100644 --- a/config/manager/manager.yaml +++ b/config/manager/manager.yaml @@ -45,10 +45,27 @@ spec: value: "cluster-1" - name: NODE_IP value: "1.2.3.4" + # Controller failover (worker-operator #467). Setting the endpoint of + # a second hub turns on active-hub resolution: the worker reads + # status.activeController from both hubs, connects to whichever one + # currently holds leadership, and restarts to follow a promotion. + # Leave unset for a single-hub deployment — nothing below applies and + # behaviour is unchanged. + # - name: HUB_SECONDARY_HOST_ENDPOINT + # value: "https://10.1.80.12:6443" + # The second hub needs its own credential, valid on that hub. See + # docs/hub-failover.md for how to mint and install it. + # - name: HUB_SECONDARY_TOKEN_FILE + # value: "/var/run/secrets/kubernetes.io/hub-secondary-serviceaccount/token" + # - name: HUB_SECONDARY_CA_FILE + # value: "/var/run/secrets/kubernetes.io/hub-secondary-serviceaccount/ca.crt" volumeMounts: - mountPath: /var/run/secrets/kubernetes.io/hub-serviceaccount name: hub-token readOnly: true + # - mountPath: /var/run/secrets/kubernetes.io/hub-secondary-serviceaccount + # name: hub-secondary-token + # readOnly: true - mountPath: /etc/webhook/certs name: webhook-certs readOnly: true @@ -84,3 +101,8 @@ spec: secret: defaultMode: 420 secretName: hub-avesha-tenant-token + # Paired with the HUB_SECONDARY_* env above; see docs/hub-failover.md. + # - name: hub-secondary-token + # secret: + # defaultMode: 420 + # secretName: hub-secondary-tenant-token diff --git a/docs/hub-failover.md b/docs/hub-failover.md new file mode 100644 index 000000000..336ecceff --- /dev/null +++ b/docs/hub-failover.md @@ -0,0 +1,119 @@ +# Following a controller failover (worker-operator #467) + +When the KubeSlice controller runs Active/Standby across two hub clusters, a worker has to notice +that leadership moved and reconnect to the hub that now holds it. This describes how the worker +does that, and what an operator has to provide for it to work. + +Off by default. A worker with a single hub configured behaves exactly as it always has — no +resolution, no extra connections, no change. + +## How it decides + +Each hub publishes `status.activeController` on this worker's `Cluster` CR while it holds +leadership, and a Standby's mirrored copy repeats the Active's declaration. The worker polls both +pre-provisioned hub endpoints and applies one rule: + +1. An unreachable hub has no say. +2. A hub that published nothing has no say. This is what a non-HA hub looks like. +3. A declaration naming an endpoint that is not one of the two configured hubs is **rejected**. The + field chooses between endpoints you provisioned; it cannot point the worker somewhere else. +4. If the hubs agree, that is the answer. Agreement is the normal case — the Standby mirrors the + Active's declaration, so both name the same hub. +5. If they disagree, the fresher declaration wins. This happens when a recovered old Active still + names itself. It keeps the worker's behaviour single-valued; it does not resolve split brain, + which the controller design lists as a non-goal. +6. If nothing usable comes back, the worker changes nothing and keeps its current connection. + +A change has to hold across consecutive polls before the worker acts, so a single blip cannot move +it. When it does act, the worker shuts down cleanly and the kubelet restarts it; startup resolution +then picks the new hub. Gateways and tunnels run in their own pods, so the data plane is not +affected by that restart. + +## Configuration + +| Variable | Default | Meaning | +|---|---|---| +| `HUB_SECONDARY_HOST_ENDPOINT` | *(unset)* | The other hub's API server. **Unset disables everything here.** | +| `HUB_SECONDARY_TOKEN_FILE` | `/var/run/secrets/kubernetes.io/hub-secondary-serviceaccount/token` | Credential for that hub | +| `HUB_SECONDARY_CA_FILE` | `/var/run/secrets/kubernetes.io/hub-secondary-serviceaccount/ca.crt` | CA for that hub | +| `HUB_RESOLVE_INTERVAL` | `10s` | How often to re-check | +| `HUB_RESOLVE_TIMEOUT` | `5s` | Bound on each read of a hub | +| `HUB_SWITCH_CONFIRMATIONS` | `2` | Consecutive agreeing polls before reconnecting | + +`config/manager/manager.yaml` carries the env, volume mount and volume for the second credential, +commented out. + +Two metrics are exported: `kubeslice_worker_hub_switches_total` and +`kubeslice_worker_hub_probe_errors_total{hub="primary|secondary"}`. The second is the one to alert +on — a hub that has been quietly unreachable for days is a problem you want to hear about before a +failover, not during one. + +## The second credential + +The worker authenticates to a hub with a token that hub minted. A token from hub A is not valid on +hub B, so the worker needs a **second** credential, for the Standby, mounted **before** the Active +fails — afterwards there is nothing left to hand it one. + +**This step has no owner in either repository.** It belongs to the cluster registration flow and +the Helm charts, not to worker-operator or kubeslice-controller. Until it is part of registration, +install it by hand. The steps below are what the Kind demo uses. + +The controller side already does its half: the Standby mirrors the worker's `ServiceAccount` and an +empty token `Secret` shell, and its own token controller fills that shell with a token valid on the +Standby. So the credential already exists on the Standby — it just has to be copied to the worker. + +Set these to match your deployment: + +```bash +export STANDBY=kind-hub-standby +export WORKER=kind-worker-1 +export PROJECT_NS=kubeslice-avesha +export WORKER_SA=kubeslice-worker-worker-1 +export WORKER_NS=kubeslice-system +``` + +Read the Standby-minted token and CA: + +```bash +kubectl --context $STANDBY get secret $WORKER_SA \ + -n $PROJECT_NS -o jsonpath='{.data.token}' | base64 -d > /tmp/hub-b-token +``` + +```bash +kubectl --context $STANDBY get secret $WORKER_SA \ + -n $PROJECT_NS -o jsonpath='{.data.ca\.crt}' | base64 -d > /tmp/hub-b-ca.crt +``` + +Confirm it actually authenticates on the Standby before installing it. A token that is present but +invalid is worse than none, because it fails only at failover: + +```bash +kubectl --server=$(kubectl --context $STANDBY config view -o \ + jsonpath='{.clusters[?(@.name=="'${STANDBY#kind-}'")].cluster.server}') \ + --certificate-authority=/tmp/hub-b-ca.crt \ + --token="$(cat /tmp/hub-b-token)" auth whoami +``` + +Install it on the worker: + +```bash +kubectl --context $WORKER create secret generic hub-secondary-tenant-token \ + -n $WORKER_NS \ + --from-file=token=/tmp/hub-b-token \ + --from-file=ca.crt=/tmp/hub-b-ca.crt +``` + +Then uncomment the `HUB_SECONDARY_*` env, the volume mount and the volume in the worker's +deployment, set `HUB_SECONDARY_HOST_ENDPOINT` to the Standby's API server address, and restart the +worker. + +```bash +rm -f /tmp/hub-b-token /tmp/hub-b-ca.crt +``` + +## Checking it works + +With both hubs up, the worker logs which hub it resolved at startup and then stays quiet. Stop the +Active controller; once the Standby promotes itself, the worker logs `active hub changed`, +increments `kubeslice_worker_hub_switches_total`, exits, and comes back connected to the promoted +hub. Gateway pods should not restart at any point — that is the part worth watching. From 02e8146ccb1cadf691a63c43fdaa890eefc7f723 Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Sun, 2 Aug 2026 15:45:38 +0530 Subject: [PATCH 05/10] fix(hub): resolve the first active hub without waiting for confirmations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Startup resolution never worked at any confirmation setting above one. The caller resolves once before opening a connection, but the very first winner had to clear the same consecutive-poll threshold as a later switch — and a single call can never reach it. The worker fell back to its configured primary every time, including when the primary is the hub that just lost leadership, which is the one case startup resolution exists for. The threshold guards a *change* of hub: it stops one divergent poll re-pointing a worker that is already connected somewhere. Before anything is established there is no connection to protect and nothing to flap between, so the first winner is now taken as it stands. Later changes are unaffected. Found by running the failover demo against live clusters, where startup logged "candidate active hub not yet confirmed, seen 1 need 2" and then ignored a hub that had plainly declared itself. No unit test caught it because every existing test drove the resolver in a loop, which is exactly what startup does not do. TestResolve_RequiresConsecutiveConfirmations asserted the old behaviour in passing and has been corrected. Part of #467 Signed-off-by: Sumanth D --- pkg/hub/resolver/resolver.go | 23 ++++++++++++++++++++++- pkg/hub/resolver/resolver_test.go | 26 ++++++++++++++++++++++++-- 2 files changed, 46 insertions(+), 3 deletions(-) diff --git a/pkg/hub/resolver/resolver.go b/pkg/hub/resolver/resolver.go index 3d5d89d1d..56c894831 100644 --- a/pkg/hub/resolver/resolver.go +++ b/pkg/hub/resolver/resolver.go @@ -279,7 +279,28 @@ func (r *Resolver) pick(claims []Claim) *Claim { // act on: either the established winner, or a challenger that has now agreed // with itself enough times to replace it. func (r *Resolver) confirm(winner *Claim) *Claim { - if r.current != nil && sameTarget(r.current, winner) { + // The confirmation rule guards a *change* of hub: it exists so a single + // divergent poll cannot re-point a worker that is already connected + // somewhere. With nothing established yet there is no connection to + // protect and nothing to flap between, so the first winner is taken as it + // stands. + // + // Requiring confirmations here instead makes startup resolution useless at + // any setting above one: the caller resolves once before opening a + // connection, that single call can never reach the threshold, and the + // worker falls back to its configured primary every time — including when + // the primary is the hub that just lost leadership. Found by running the + // failover demo, where startup logged "not yet confirmed, seen 1 need 2" + // and then ignored a correctly resolved Active. + if r.current == nil { + r.current = winner + r.pending = nil + r.pendingCount = 0 + r.log.Info("active hub resolved", "identity", winner.Identity, + "endpoint", winner.Endpoint, "previous", "") + return r.current + } + if sameTarget(r.current, winner) { // Re-confirmation of the status quo clears any half-accumulated // challenger; a challenger has to win consecutive polls, not cumulative // ones. diff --git a/pkg/hub/resolver/resolver_test.go b/pkg/hub/resolver/resolver_test.go index 70e4202c5..039ed2170 100644 --- a/pkg/hub/resolver/resolver_test.go +++ b/pkg/hub/resolver/resolver_test.go @@ -228,9 +228,8 @@ func TestResolve_RequiresConsecutiveConfirmations(t *testing.T) { } r := newResolver(t, func(_ context.Context, c HubCandidate) Verdict { return verdicts[c.Endpoint] }, 3) - assert.Nil(t, resolveN(r, 2), "even the first winner must earn its confirmations") got := resolveN(r, 1) - mustClaim(t, got, "") + mustClaim(t, got, "the first winner is taken at once; there is no connection to protect yet") assert.Equal(t, "hub-a-1", got.Identity) // hub B takes over. @@ -337,3 +336,26 @@ func mustClaim(t *testing.T, got *Claim, msg string) *Claim { } return got } + +// TestResolve_FirstWinnerNeedsNoConfirmation is the regression test for a bug +// the live failover demo found. Callers resolve once before opening a +// connection, so if the very first selection also had to clear the confirmation +// threshold, that single call could never reach it and startup resolution would +// silently fall back to the configured primary at any setting above one — +// including when the primary is the hub that just lost leadership. +// +// The threshold guards a *change* of hub. Before anything is established there +// is no connection to protect and nothing to flap between. +func TestResolve_FirstWinnerNeedsNoConfirmation(t *testing.T) { + for _, confirmations := range []int{1, 2, 5} { + r := newResolver(t, staticProbe(map[string]Verdict{ + hubA.Endpoint: unreachable(hubA), + hubB.Endpoint: declares(hubB, hubB, "hub-b-1", time.Second), + }), confirmations) + + got := r.Resolve(context.Background()) + mustClaim(t, got, "a single startup resolve must produce an answer") + assert.Equal(t, "hub-b-1", got.Identity, + "confirmations=%d: startup must not fall back when a hub plainly declares itself", confirmations) + } +} From fbb6845085737c11dab973b4ecf03da5482e4550 Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Tue, 25 Aug 2026 16:26:42 +0530 Subject: [PATCH 06/10] chore: vendor apis with ClusterStatus.Conditions Pulls in ActiveController/StorageCapabilities (already merged locally, never vendored) and the new Conditions field #469 needs. Pinned to the fork commit behind kubeslice/apis#47; drop the replace once that merges and a release carries the field. Signed-off-by: Sumanth D --- go.mod | 2 + go.sum | 4 +- .../pkg/controller/v1alpha1/cluster_types.go | 52 ++++++++++++++ .../v1alpha1/zz_generated.deepcopy.go | 70 +++++++++++++++++++ vendor/modules.txt | 3 +- 5 files changed, 128 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index dd67f0aef..ae2689a53 100644 --- a/go.mod +++ b/go.mod @@ -102,3 +102,5 @@ require ( sigs.k8s.io/structured-merge-diff/v4 v4.4.2 // indirect sigs.k8s.io/yaml v1.4.0 // indirect ) + +replace github.com/kubeslice/apis => github.com/sumanthd032/apis v0.5.1-0.20260825105516-d7d920d4b404 diff --git a/go.sum b/go.sum index e7b3672f1..b0a5be56c 100644 --- a/go.sum +++ b/go.sum @@ -264,8 +264,6 @@ github.com/kr/pty v1.1.5/go.mod h1:9r2w37qlBe7rQ6e1fg1S/9xpWHSnaqNdHD3WcMdbPDA= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/kubeslice/apis v0.4.0 h1:nU66JoA2OQx48bZnXDWH8iHG+5R1ELX5ikc3l/fn5II= -github.com/kubeslice/apis v0.4.0/go.mod h1:F1hXnAt3Dk4Sto5yQDoMnqgXX5ImL1bRBiAmrW6TG00= github.com/kubeslice/gateway-sidecar v0.2.0 h1:Ja3fIUivuSjUFQ4lPCt79ATq99BxslvAFYUwV9Urpy4= github.com/kubeslice/gateway-sidecar v0.2.0/go.mod h1:nM1+Wjud2vk44cUg+9iwBbWTpqI+2Ecbn9NuaHEs9aY= github.com/kubeslice/kubeslice-monitoring v0.2.1 h1:wtmIEigpQoKzuckof7QRqdsaa4lV/rqxd/FcmOj5N5Q= @@ -374,6 +372,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/sumanthd032/apis v0.5.1-0.20260825105516-d7d920d4b404 h1:vlGmBE+ZeGdPhc5TwOZgd/0dQSSvLdW4sdoPG/VJu34= +github.com/sumanthd032/apis v0.5.1-0.20260825105516-d7d920d4b404/go.mod h1:F1hXnAt3Dk4Sto5yQDoMnqgXX5ImL1bRBiAmrW6TG00= github.com/vishvananda/netlink v1.2.1-beta.2.0.20220812183158-d44b87fd4d3f h1:kTcjiSlfxwj/o7ezwNrsYjydHYu4Bgu0OkF+q46Vb3k= github.com/vishvananda/netlink v1.2.1-beta.2.0.20220812183158-d44b87fd4d3f/go.mod h1:cAAsePK2e15YDAMJNyOpGYEWNe4sIghTY7gpz4cX/Ik= github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0= diff --git a/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/cluster_types.go b/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/cluster_types.go index dda4a7f21..ad61ba3d7 100644 --- a/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/cluster_types.go +++ b/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/cluster_types.go @@ -137,6 +137,58 @@ type ClusterStatus struct { // VCPURestriction is the restriction on the cluster disabling the creation of new pods VCPURestriction *VCPURestriction `json:"vCPURestriction,omitempty"` GPURestriction *GPURestriction `json:"GPURestriction,omitempty"` + // StorageCapabilities contains auto-detected storage capabilities reported by the worker operator. + // Populated only when the worker operator's storage-capability reconciler is active. + StorageCapabilities *StorageCapabilities `json:"storageCapabilities,omitempty"` + // ActiveController identifies the hub controller that currently holds leadership. + // Populated only on an Active/Standby HA deployment; absent otherwise, so a + // non-HA worker sees no behaviour change. + ActiveController *ActiveControllerInfo `json:"activeController,omitempty"` + // Conditions describe this worker's connection to its hub controller, e.g. + // ControllerConnected and ControllerEndpointSynced. Written by the worker + // operator about itself; absent on a worker that has never reported one. + Conditions []metav1.Condition `json:"conditions,omitempty"` +} + +// ActiveControllerInfo describes the hub controller currently holding leadership. +// +// Each hub writes this field about itself, on its own API server, and only while +// it holds leadership. A Standby's copy is populated by the state mirror from the +// Active, so it names the Active rather than itself — which lets a worker watching +// both hubs identify the Active by the rule "trust whichever endpoint is reachable +// and reports an ActiveIdentity matching that endpoint's own identity", without +// needing to know which role either hub currently holds. +type ActiveControllerInfo struct { + // Endpoint is the API server endpoint of the hub currently holding leadership + Endpoint string `json:"endpoint,omitempty"` + // CABundle is the base64-encoded PEM CA bundle for Endpoint + CABundle string `json:"caBundle,omitempty"` + // ActiveIdentity is the HA identity of the hub that wrote this field about itself + ActiveIdentity string `json:"activeIdentity,omitempty"` + // LastUpdated is the timestamp when this declaration was last written. It gives + // a consumer a deterministic tie-break if both hubs self-declare simultaneously. + LastUpdated metav1.Time `json:"lastUpdated,omitempty"` +} + +// StorageCapabilities holds auto-detected RWX-capable storage classes on the worker cluster. +// To add support for a new storage system, append its CSI provisioner string to the +// worker operator's rwxProvisioners list — no changes to this struct are required. +type StorageCapabilities struct { + // RWXStorageClasses lists all ReadWriteMany-capable StorageClasses detected on the cluster + RWXStorageClasses []RWXStorageClass `json:"rwxStorageClasses,omitempty"` + // DefaultStorageClass is the name of the StorageClass annotated with + // storageclass.kubernetes.io/is-default-class: "true" on the worker cluster + DefaultStorageClass string `json:"defaultStorageClass,omitempty"` + // LastUpdated is the timestamp when capabilities were last detected + LastUpdated metav1.Time `json:"lastUpdated,omitempty"` +} + +// RWXStorageClass describes a single ReadWriteMany-capable StorageClass. +type RWXStorageClass struct { + // Name is the StorageClass name (e.g. "rook-cephfs", "juicefs-sc") + Name string `json:"name"` + // Provisioner is the CSI provisioner string (e.g. "rook-ceph.cephfs.csi.ceph.com") + Provisioner string `json:"provisioner"` } type GPURestriction struct { diff --git a/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/zz_generated.deepcopy.go b/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/zz_generated.deepcopy.go index baeea08ce..6215f0f86 100644 --- a/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/zz_generated.deepcopy.go +++ b/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/zz_generated.deepcopy.go @@ -20,9 +20,26 @@ package v1alpha1 import ( + "k8s.io/apimachinery/pkg/apis/meta/v1" runtime "k8s.io/apimachinery/pkg/runtime" ) +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ActiveControllerInfo) DeepCopyInto(out *ActiveControllerInfo) { + *out = *in + in.LastUpdated.DeepCopyInto(&out.LastUpdated) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ActiveControllerInfo. +func (in *ActiveControllerInfo) DeepCopy() *ActiveControllerInfo { + if in == nil { + return nil + } + out := new(ActiveControllerInfo) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Cluster) DeepCopyInto(out *Cluster) { *out = *in @@ -175,6 +192,23 @@ func (in *ClusterStatus) DeepCopyInto(out *ClusterStatus) { *out = new(GPURestriction) (*in).DeepCopyInto(*out) } + if in.StorageCapabilities != nil { + in, out := &in.StorageCapabilities, &out.StorageCapabilities + *out = new(StorageCapabilities) + (*in).DeepCopyInto(*out) + } + if in.ActiveController != nil { + in, out := &in.ActiveController, &out.ActiveController + *out = new(ActiveControllerInfo) + (*in).DeepCopyInto(*out) + } + if in.Conditions != nil { + in, out := &in.Conditions, &out.Conditions + *out = make([]v1.Condition, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterStatus. @@ -473,6 +507,21 @@ func (in *QOSProfile) DeepCopy() *QOSProfile { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *RWXStorageClass) DeepCopyInto(out *RWXStorageClass) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RWXStorageClass. +func (in *RWXStorageClass) DeepCopy() *RWXStorageClass { + if in == nil { + return nil + } + out := new(RWXStorageClass) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ServiceAccess) DeepCopyInto(out *ServiceAccess) { *out = *in @@ -917,6 +966,27 @@ func (in *StatusOfKeyRotation) DeepCopy() *StatusOfKeyRotation { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *StorageCapabilities) DeepCopyInto(out *StorageCapabilities) { + *out = *in + if in.RWXStorageClasses != nil { + in, out := &in.RWXStorageClasses, &out.RWXStorageClasses + *out = make([]RWXStorageClass, len(*in)) + copy(*out, *in) + } + in.LastUpdated.DeepCopyInto(&out.LastUpdated) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new StorageCapabilities. +func (in *StorageCapabilities) DeepCopy() *StorageCapabilities { + if in == nil { + return nil + } + out := new(StorageCapabilities) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Telemetry) DeepCopyInto(out *Telemetry) { *out = *in diff --git a/vendor/modules.txt b/vendor/modules.txt index 21e06e176..dfaf43aa1 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -116,7 +116,7 @@ github.com/josharian/intern # github.com/json-iterator/go v1.1.12 ## explicit; go 1.12 github.com/json-iterator/go -# github.com/kubeslice/apis v0.4.0 +# github.com/kubeslice/apis v0.4.0 => github.com/sumanthd032/apis v0.5.1-0.20260825105516-d7d920d4b404 ## explicit; go 1.24.0 github.com/kubeslice/apis/pkg/controller/v1alpha1 github.com/kubeslice/apis/pkg/worker/v1alpha1 @@ -977,3 +977,4 @@ sigs.k8s.io/structured-merge-diff/v4/value ## explicit; go 1.12 sigs.k8s.io/yaml sigs.k8s.io/yaml/goyaml.v2 +# github.com/kubeslice/apis => github.com/sumanthd032/apis v0.5.1-0.20260825105516-d7d920d4b404 From 0641fa6e5eaa51d83ff766924e572e27352019c7 Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Tue, 25 Aug 2026 16:56:12 +0530 Subject: [PATCH 07/10] feat(hub): classify a hub-connection error as DialFailed or CertVerificationFailed Distinguishes a TLS/cert-handshake failure from everything else, for issue #469's ControllerConnected reason table. Signed-off-by: Sumanth D --- pkg/hub/hubclient/classify.go | 47 +++++++++++++++++++++ pkg/hub/hubclient/classify_test.go | 68 ++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+) create mode 100644 pkg/hub/hubclient/classify.go create mode 100644 pkg/hub/hubclient/classify_test.go diff --git a/pkg/hub/hubclient/classify.go b/pkg/hub/hubclient/classify.go new file mode 100644 index 000000000..47690667c --- /dev/null +++ b/pkg/hub/hubclient/classify.go @@ -0,0 +1,47 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 hub + +import ( + "crypto/x509" + "errors" +) + +// Connection error reasons, matching worker-operator issue #469's +// ControllerConnected reason table. DialFailed is the default for anything +// that isn't specifically a certificate problem: a refused connection, a DNS +// failure, and a timeout all look the same to an operator deciding what to +// check first, and all point at network/endpoint configuration rather than +// the CA bundle. +const ( + ReasonDialFailed = "DialFailed" + ReasonCertVerificationFailed = "CertVerificationFailed" +) + +// ClassifyConnectionError reports which of the two reasons a hub-connection +// error matches. err must be non-nil. +func ClassifyConnectionError(err error) string { + var unknownAuthority x509.UnknownAuthorityError + var certInvalid x509.CertificateInvalidError + var hostnameErr x509.HostnameError + if errors.As(err, &unknownAuthority) || errors.As(err, &certInvalid) || errors.As(err, &hostnameErr) { + return ReasonCertVerificationFailed + } + return ReasonDialFailed +} diff --git a/pkg/hub/hubclient/classify_test.go b/pkg/hub/hubclient/classify_test.go new file mode 100644 index 000000000..9b5e1b037 --- /dev/null +++ b/pkg/hub/hubclient/classify_test.go @@ -0,0 +1,68 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 hub + +import ( + "crypto/x509" + "errors" + "fmt" + "net" + "net/url" + "testing" +) + +func TestClassifyConnectionError_CertFailuresAreDistinguished(t *testing.T) { + cases := []struct { + name string + err error + }{ + {"unknown authority", x509.UnknownAuthorityError{}}, + {"certificate invalid", x509.CertificateInvalidError{Reason: x509.Expired}}, + {"hostname mismatch", x509.HostnameError{Certificate: &x509.Certificate{}, Host: "hub.example"}}, + {"wrapped in a url.Error, as a real client-go transport failure would be", + &url.Error{Op: "Get", URL: "https://hub.example", Err: x509.UnknownAuthorityError{}}}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got := ClassifyConnectionError(c.err) + if got != ReasonCertVerificationFailed { + t.Fatalf("ClassifyConnectionError(%v) = %q, want %q", c.err, got, ReasonCertVerificationFailed) + } + }) + } +} + +func TestClassifyConnectionError_EverythingElseIsADialFailure(t *testing.T) { + cases := []struct { + name string + err error + }{ + {"connection refused", &net.OpError{Op: "dial", Err: errors.New("connection refused")}}, + {"generic timeout", fmt.Errorf("context deadline exceeded")}, + {"not found, unrelated to connectivity", errors.New("clusters.kubeslice.io \"worker-1\" not found")}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got := ClassifyConnectionError(c.err) + if got != ReasonDialFailed { + t.Fatalf("ClassifyConnectionError(%v) = %q, want %q", c.err, got, ReasonDialFailed) + } + }) + } +} From 50fa8012ee6e568527c5a4179903f4de1d649abe Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Tue, 25 Aug 2026 16:56:24 +0530 Subject: [PATCH 08/10] feat(hub): report ControllerConnected/ControllerEndpointSynced on the worker's Cluster CR Adds the two conditions, four events and two metrics from issue #469, wired through main.go/manager.Start/the cluster reconciler. Only ever writes Connected, ReconnectedAfterFailover or EndpointNotConfigured live; DialFailed/CertVerificationFailed are unit-tested but architecturally can't persist on a hub this worker can't reach. Signed-off-by: Sumanth D --- config/events/worker-events.yaml | 26 ++++- events/events_generated.go | 36 ++++++ main.go | 40 ++++++- .../controllers/cluster/cluster_suite_test.go | 1 + pkg/hub/controllers/cluster/conditions.go | 107 ++++++++++++++++++ .../cluster/deregister_unit_test.go | 20 ++-- pkg/hub/controllers/cluster/reconciler.go | 13 ++- .../cluster/reconciler_unit_test.go | 10 +- pkg/hub/failover/failover.go | 42 ++++++- pkg/hub/failover/failover_test.go | 10 +- pkg/hub/manager/manager.go | 5 +- tests/hub/hub_suite_test.go | 1 + 12 files changed, 282 insertions(+), 29 deletions(-) create mode 100644 pkg/hub/controllers/cluster/conditions.go diff --git a/config/events/worker-events.yaml b/config/events/worker-events.yaml index a604d72f0..b97db91d4 100644 --- a/config/events/worker-events.yaml +++ b/config/events/worker-events.yaml @@ -508,4 +508,28 @@ events: action: None type: Warning reportingController: worker - message: Gateway recycling failed \ No newline at end of file + message: Gateway recycling failed + - name: ControllerEndpointChanged + reason: ControllerEndpointChanged + action: None + type: Normal + reportingController: worker + message: This worker followed a resolved controller failover and is now reconciling against a different hub endpoint + - name: ControllerConnected + reason: Connected + action: None + type: Normal + reportingController: worker + message: Connection to the hub controller is healthy + - name: ControllerConnectionLost + reason: DialFailed + action: None + type: Warning + reportingController: worker + message: The active hub changed and this worker is restarting to reconnect; the previous connection may already be lost + - name: CertVerificationFailed + reason: CertVerificationFailed + action: None + type: Warning + reportingController: worker + message: TLS handshake with the hub controller failed certificate verification; check the configured CA bundle \ No newline at end of file diff --git a/events/events_generated.go b/events/events_generated.go index 602aa1e7d..18721892a 100644 --- a/events/events_generated.go +++ b/events/events_generated.go @@ -702,6 +702,38 @@ var EventsMap = map[events.EventName]*events.EventSchema{ ReportingController: "worker", Message: "Gateway recycling failed", }, + "ControllerEndpointChanged": { + Name: "ControllerEndpointChanged", + Reason: "ControllerEndpointChanged", + Action: "None", + Type: events.EventTypeNormal, + ReportingController: "worker", + Message: "This worker followed a resolved controller failover and is now reconciling against a different hub endpoint", + }, + "ControllerConnected": { + Name: "ControllerConnected", + Reason: "Connected", + Action: "None", + Type: events.EventTypeNormal, + ReportingController: "worker", + Message: "Connection to the hub controller is healthy", + }, + "ControllerConnectionLost": { + Name: "ControllerConnectionLost", + Reason: "DialFailed", + Action: "None", + Type: events.EventTypeWarning, + ReportingController: "worker", + Message: "The active hub changed and this worker is restarting to reconnect; the previous connection may already be lost", + }, + "CertVerificationFailed": { + Name: "CertVerificationFailed", + Reason: "CertVerificationFailed", + Action: "None", + Type: events.EventTypeWarning, + ReportingController: "worker", + Message: "TLS handshake with the hub controller failed certificate verification; check the configured CA bundle", + }, } var ( @@ -790,4 +822,8 @@ var ( EventTriggeredFSMToRecycleGateways events.EventName = "TriggeredFSMToRecycleGateways" EventGatewayRecyclingSuccessful events.EventName = "GatewayRecyclingSuccessful" EventGatewayRecyclingFailed events.EventName = "GatewayRecyclingFailed" + EventControllerEndpointChanged events.EventName = "ControllerEndpointChanged" + EventControllerConnected events.EventName = "ControllerConnected" + EventControllerConnectionLost events.EventName = "ControllerConnectionLost" + EventCertVerificationFailed events.EventName = "CertVerificationFailed" ) diff --git a/main.go b/main.go index 69a2d79ed..c52f10cbf 100644 --- a/main.go +++ b/main.go @@ -42,6 +42,8 @@ import ( netop "github.com/kubeslice/worker-operator/pkg/netop" router "github.com/kubeslice/worker-operator/pkg/router" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" @@ -65,7 +67,9 @@ import ( "github.com/kubeslice/worker-operator/controllers/serviceimport" "github.com/kubeslice/worker-operator/controllers/slice" "github.com/kubeslice/worker-operator/controllers/slicegateway" + hubv1alpha1 "github.com/kubeslice/apis/pkg/controller/v1alpha1" ossEvents "github.com/kubeslice/worker-operator/events" + hubCluster "github.com/kubeslice/worker-operator/pkg/hub/controllers/cluster" "github.com/kubeslice/worker-operator/pkg/hub/failover" hub "github.com/kubeslice/worker-operator/pkg/hub/hubclient" "github.com/kubeslice/worker-operator/pkg/hub/manager" @@ -174,14 +178,16 @@ func main() { hubConn := hub.PrimaryConnection() failoverCfg := failover.ConfigFromEnv() var hubFollower *failover.Follower + var reconnected bool if failoverCfg.Enabled() { hubFollower, err = failover.New(failoverCfg, hubConn, ctrl.Log.WithName("hub-failover"), nil) if err != nil { setupLog.With("error", err).Error("could not configure hub failover following") os.Exit(1) } - hubConn = hubFollower.StartupConnection(context.Background()) + hubConn, reconnected = hubFollower.StartupConnection(context.Background()) } + connInfo := hubCluster.ConnectionInfo{Enabled: failoverCfg.Enabled(), Reconnected: reconnected} hubClient, err := hub.NewHubClientConfig(er, hubConn) if err != nil { @@ -332,7 +338,7 @@ func main() { } go func() { setupLog.Info("starting hub manager") - manager.Start(clientForHubMgr, hubClient, ctx, hubConn) + manager.Start(clientForHubMgr, hubClient, ctx, hubConn, connInfo) }() if hubFollower != nil { @@ -345,6 +351,7 @@ func main() { // own pods. setupLog.With("endpoint", claim.Endpoint, "identity", claim.Identity). Info("active hub changed; shutting down to reconnect") + reportConnectionLost(hubClient, &sliceEventRecorder) stopForHubSwitch() }) } @@ -355,3 +362,32 @@ func main() { os.Exit(1) } } + +// reportConnectionLost makes one best-effort attempt to record, on the hub +// this worker is about to leave, that it is doing so (issue #469's +// ControllerConnectionLost event and Reconnecting condition). Best-effort +// because the only connection available to write with is the one this +// worker is abandoning — if that hub is itself the reason for the switch, +// there is nothing to write to, and that failure is expected, not fatal. +func reportConnectionLost(hubClient client.Client, er *monitoringEvents.EventRecorder) { + cr := &hubv1alpha1.Cluster{} + err := hubClient.Get(context.Background(), client.ObjectKey{ + Name: controllers.ClusterName, + Namespace: hub.ProjectNamespace, + }, cr) + if err != nil { + setupLog.With("error", err).Info("could not report connection loss before reconnecting; the hub may already be unreachable") + return + } + meta.SetStatusCondition(&cr.Status.Conditions, metav1.Condition{ + Type: hubCluster.ConditionControllerConnected, + Status: metav1.ConditionUnknown, + Reason: hubCluster.ReasonReconnecting, + Message: "following a resolved hub failover; reconnecting", + }) + if err := hubClient.Status().Update(context.Background(), cr); err != nil { + setupLog.With("error", err).Info("could not persist the Reconnecting condition before restart") + return + } + utils.RecordEvent(context.Background(), er, cr, nil, ossEvents.EventControllerConnectionLost, "hub-failover") +} diff --git a/pkg/hub/controllers/cluster/cluster_suite_test.go b/pkg/hub/controllers/cluster/cluster_suite_test.go index c7b140b06..6d97d6a5d 100644 --- a/pkg/hub/controllers/cluster/cluster_suite_test.go +++ b/pkg/hub/controllers/cluster/cluster_suite_test.go @@ -132,6 +132,7 @@ var _ = BeforeSuite(func() { k8sClient, &spokeClusterEventRecorder, mf, + ConnectionInfo{}, ) clusterReconciler.ReconcileInterval = 5 * time.Second err = builder. diff --git a/pkg/hub/controllers/cluster/conditions.go b/pkg/hub/controllers/cluster/conditions.go new file mode 100644 index 000000000..449c43654 --- /dev/null +++ b/pkg/hub/controllers/cluster/conditions.go @@ -0,0 +1,107 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 cluster + +import ( + "context" + + hubv1alpha1 "github.com/kubeslice/apis/pkg/controller/v1alpha1" + ossEvents "github.com/kubeslice/worker-operator/events" + "github.com/kubeslice/worker-operator/pkg/utils" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// Condition types and reasons for issue #469. Reason strings are shared with +// pkg/hub/hubclient.ClassifyConnectionError where they overlap +// (DialFailed/CertVerificationFailed) so a caller wiring one into the other +// cannot typo a mismatch. +const ( + ConditionControllerConnected = "ControllerConnected" + ConditionControllerEndpointSynced = "ControllerEndpointSynced" + + ReasonConnected = "Connected" + ReasonReconnectedAfterFailover = "ReconnectedAfterFailover" + ReasonEndpointNotConfigured = "EndpointNotConfigured" + ReasonReconnecting = "Reconnecting" + ReasonEndpointUpToDate = "EndpointUpToDate" +) + +// ConnectionInfo is what main.go already knows about this worker's hub +// connection before the hub manager starts: whether failover-following is +// configured at all, and whether the connection now in use was reached by +// following a resolved switch rather than the configured primary. It is a +// startup-time fact, not a live signal. +type ConnectionInfo struct { + Enabled bool + Reconnected bool +} + +// setConnectionConditions records this worker's hub-connection health on its +// own Cluster CR, via the standard idempotent SetStatusCondition helper (a +// repeated no-op call never bumps LastTransitionTime). +// +// This only ever sets Connected, ReconnectedAfterFailover or +// EndpointNotConfigured — never DialFailed or CertVerificationFailed. Those +// two are real reason values (see pkg/hub/hubclient.ClassifyConnectionError, +// unit-tested there) but this method only runs because Reconcile just read cr +// successfully: a hub this worker cannot reach is a hub it cannot write +// "unreachable" to either, on that same hub's own copy of the object. That is +// the same conclusion issue #467 reached before deferring the durable surface +// here. kubeslice_worker_hub_probe_errors_total and logs remain the live +// down-detection channel; this condition is the durable, hub-side-visible +// record of the connection's last known good state. +func (r *Reconciler) setConnectionConditions(ctx context.Context, cr *hubv1alpha1.Cluster) { + connected := metav1.Condition{ + Type: ConditionControllerConnected, + Status: metav1.ConditionTrue, + Reason: ReasonConnected, + Message: "worker is reconciling against this hub", + } + synced := metav1.Condition{ + Type: ConditionControllerEndpointSynced, + Status: metav1.ConditionTrue, + Reason: ReasonEndpointUpToDate, + Message: "connected to the currently resolved active hub", + } + + switch { + case !r.connInfo.Enabled: + connected.Status = metav1.ConditionUnknown + connected.Reason = ReasonEndpointNotConfigured + connected.Message = "no secondary hub configured; failover-following is inactive" + synced.Status = metav1.ConditionUnknown + synced.Reason = ReasonEndpointNotConfigured + synced.Message = "no secondary hub configured; failover-following is inactive" + case r.connInfo.Reconnected && !r.reportedReconnect: + connected.Reason = ReasonReconnectedAfterFailover + connected.Message = "resumed reconciliation after following a resolved hub failover" + } + + changed := meta.SetStatusCondition(&cr.Status.Conditions, connected) + meta.SetStatusCondition(&cr.Status.Conditions, synced) + + if r.connInfo.Reconnected && !r.reportedReconnect { + utils.RecordEvent(ctx, r.EventRecorder, cr, nil, ossEvents.EventControllerEndpointChanged, controllerName) + r.reportedReconnect = true + } + if changed && connected.Reason == ReasonConnected { + utils.RecordEvent(ctx, r.EventRecorder, cr, nil, ossEvents.EventControllerConnected, controllerName) + } +} diff --git a/pkg/hub/controllers/cluster/deregister_unit_test.go b/pkg/hub/controllers/cluster/deregister_unit_test.go index dde36f003..5d978f85d 100644 --- a/pkg/hub/controllers/cluster/deregister_unit_test.go +++ b/pkg/hub/controllers/cluster/deregister_unit_test.go @@ -76,7 +76,7 @@ func TestGetOperatorClusterRole(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterRoleKey := types.NamespacedName{Name: operatorClusterRoleName} @@ -108,7 +108,7 @@ func TestCreateDeregisterJobPositiveScenarios(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -215,7 +215,7 @@ func TestReconcilerFailToUpdateClusterRegistrationStatus(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -258,7 +258,7 @@ func TestReconcilerFailToCreateServiceAccount(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -316,7 +316,7 @@ func TestReconcilerFailToFetchOperatorClusterRole(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -384,7 +384,7 @@ func TestReconcilerFailToCreateClusterRole(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -462,7 +462,7 @@ func TestReconcilerFailToCreateClusterRoleBinding(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -545,7 +545,7 @@ func TestReconcilerFailToCreateConfigmap(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -642,7 +642,7 @@ func TestReconcilerFailToDeleteJob(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -750,7 +750,7 @@ func TestReconcilerFailToCreateDeregisterJob(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} diff --git a/pkg/hub/controllers/cluster/reconciler.go b/pkg/hub/controllers/cluster/reconciler.go index 7c4d25026..f380cfb3e 100644 --- a/pkg/hub/controllers/cluster/reconciler.go +++ b/pkg/hub/controllers/cluster/reconciler.go @@ -67,9 +67,17 @@ type Reconciler struct { gaugeComponentUp *prometheus.GaugeVec ReconcileInterval time.Duration + + // connInfo and reportedReconnect back setConnectionConditions (issue + // #469). reportedReconnect flips true the first time this process reports + // ReconnectedAfterFailover, so every later reconcile reports plain + // Connected instead — connInfo.Reconnected itself never changes for the + // life of the process. + connInfo ConnectionInfo + reportedReconnect bool } -func NewReconciler(c client.Client, mc client.Client, er *events.EventRecorder, mf metrics.MetricsFactory) *Reconciler { +func NewReconciler(c client.Client, mc client.Client, er *events.EventRecorder, mf metrics.MetricsFactory, connInfo ConnectionInfo) *Reconciler { gaugeClusterUp := mf.NewGauge("cluster_up", "Kubeslice cluster health status", []string{}) gaugeComponentUp := mf.NewGauge("cluster_component_up", "Kubeslice cluster component health status", []string{"slice_cluster_component"}) @@ -82,6 +90,8 @@ func NewReconciler(c client.Client, mc client.Client, er *events.EventRecorder, gaugeComponentUp: gaugeComponentUp, ReconcileInterval: 120 * time.Second, + + connInfo: connInfo, } } @@ -173,6 +183,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco } r.updateClusterMetrics(cr) + r.setConnectionConditions(ctx, cr) if time.Since(cr.Status.ClusterHealth.LastUpdated.Time) > r.ReconcileInterval { cr.Status.ClusterHealth.LastUpdated = metav1.Now() diff --git a/pkg/hub/controllers/cluster/reconciler_unit_test.go b/pkg/hub/controllers/cluster/reconciler_unit_test.go index 595341eb2..e5e556572 100644 --- a/pkg/hub/controllers/cluster/reconciler_unit_test.go +++ b/pkg/hub/controllers/cluster/reconciler_unit_test.go @@ -65,7 +65,7 @@ func TestReconcileToReturnErrorWhileFetchingControllerCluster(t *testing.T) { ).Return(errors.New("object not found")) mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) result, err := reconciler.Reconcile(expected.ctx, expected.req) if expected.res != result { @@ -106,7 +106,7 @@ func TestReconcileToCallHandleClusterDeletion(t *testing.T) { ).Return(errors.New("failed to update cluster CR")) mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) result, err := reconciler.Reconcile(expected.ctx, expected.req) if expected.res != result { @@ -133,7 +133,7 @@ func TestReconcilerHandleClusterDeletion(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) @@ -166,7 +166,7 @@ func TestReconcilerHandleExternalDependency(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObjWithFinalizer) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -283,7 +283,7 @@ func TestReconcilerToFailWhileCallingCreateDeregisterJob(t *testing.T) { Namespace: controllers.ControlPlaneNamespace, }) mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, &testClusterEventRecorder, mf) + reconciler := NewReconciler(client, client, &testClusterEventRecorder, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObjWithFinalizer) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} diff --git a/pkg/hub/failover/failover.go b/pkg/hub/failover/failover.go index fa39a1203..0a5ccd4ac 100644 --- a/pkg/hub/failover/failover.go +++ b/pkg/hub/failover/failover.go @@ -65,10 +65,24 @@ var ( Name: "kubeslice_worker_hub_probe_errors_total", Help: "Failed reads of a hub's copy of this worker's Cluster CR, by configured hub slot.", }, []string{"hub"}) + // controllerReconnectAttemptsTotal counts every resolved winner this + // worker acted on at startup, by whether it matched the configured + // primary or a resolved switch. It is deliberately not "every poll" — + // gather() already runs every tick and hubProbeErrorsTotal covers probe + // failures; this metric is about connection *decisions*, issue #469's ask. + controllerReconnectAttemptsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "kubeslice_worker_controller_reconnect_attempts_total", + Help: "Startup hub-connection decisions this worker has made, by result.", + }, []string{"result"}) + controllerLastSyncTime = prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "kubeslice_worker_controller_last_sync_time_seconds", + Help: "Unix time of the last resolved hub-connection decision at startup.", + }) ) func init() { - ctrlmetrics.Registry.MustRegister(hubSwitchesTotal, hubProbeErrorsTotal) + ctrlmetrics.Registry.MustRegister(hubSwitchesTotal, hubProbeErrorsTotal, + controllerReconnectAttemptsTotal, controllerLastSyncTime) } // Config is the failover-following configuration, read from the environment. @@ -209,18 +223,25 @@ func New(cfg Config, primary hub.Connection, log logr.Logger, }, nil } -// StartupConnection resolves once and returns the hub to connect to. +// StartupConnection resolves once and returns the hub to connect to, plus +// whether that hub was reached by following a resolved switch rather than +// falling back to the configured primary. The bool is issue #469's +// "ReconnectedAfterFailover vs Connected" distinction: main.go has no other +// way to know, since a fresh process looks identical whether it just started +// normally or was just restarted to follow a failover. // // Falling back to the primary when nothing resolves is deliberate. A worker // that refused to start because it could not reach a hub would turn a hub // outage into a worker outage, and the primary is the same hub it would have // used before any of this existed. -func (f *Follower) StartupConnection(ctx context.Context) hub.Connection { +func (f *Follower) StartupConnection(ctx context.Context) (hub.Connection, bool) { claim := f.resolver.Resolve(ctx) if claim == nil { f.log.Info("no active hub resolved at startup; using the configured primary", "endpoint", f.primary.Endpoint) - return f.primary + controllerReconnectAttemptsTotal.WithLabelValues("unresolved").Inc() + controllerLastSyncTime.SetToCurrentTime() + return f.primary, false } conn, ok := f.byEndpoint[claim.Endpoint] if !ok { @@ -230,11 +251,20 @@ func (f *Follower) StartupConnection(ctx context.Context) hub.Connection { // cannot silently produce a connection with mismatched credentials. f.log.Info("resolved hub is not a configured connection; using the primary", "resolvedEndpoint", claim.Endpoint) - return f.primary + controllerReconnectAttemptsTotal.WithLabelValues("unresolved").Inc() + controllerLastSyncTime.SetToCurrentTime() + return f.primary, false } f.log.Info("connecting to the resolved active hub", "identity", claim.Identity, "endpoint", conn.Endpoint) - return conn + viaSwitch := conn.Endpoint != f.primary.Endpoint + result := "primary" + if viaSwitch { + result = "resolved-switch" + } + controllerReconnectAttemptsTotal.WithLabelValues(result).Inc() + controllerLastSyncTime.SetToCurrentTime() + return conn, viaSwitch } // Watch polls until ctx is done, calling onSwitch once the resolved active hub diff --git a/pkg/hub/failover/failover_test.go b/pkg/hub/failover/failover_test.go index 828f6de9a..54ecafe5a 100644 --- a/pkg/hub/failover/failover_test.go +++ b/pkg/hub/failover/failover_test.go @@ -167,10 +167,11 @@ func TestStartupConnection_UsesTheWinnersOwnCredentials(t *testing.T) { } f := newFollower(t, testConfig(), table) - conn := f.StartupConnection(context.Background()) + conn, reconnected := f.StartupConnection(context.Background()) assert.Equal(t, secondaryEndpoint, conn.Endpoint) assert.Equal(t, "/creds/b/token", conn.TokenFile, "the winner's token must travel with its endpoint") assert.Equal(t, "/creds/b/ca.crt", conn.CAFile, "the winner's CA must travel with its endpoint") + assert.True(t, reconnected, "a winner other than the primary is a resolved switch") } func TestStartupConnection_SteadyStateStaysOnThePrimary(t *testing.T) { @@ -180,8 +181,9 @@ func TestStartupConnection_SteadyStateStaysOnThePrimary(t *testing.T) { } f := newFollower(t, testConfig(), table) - conn := f.StartupConnection(context.Background()) + conn, reconnected := f.StartupConnection(context.Background()) assert.Equal(t, primaryConn(), conn, "both hubs naming hub A is agreement, not a conflict") + assert.False(t, reconnected, "staying on the primary is not a resolved switch") } // TestStartupConnection_FallsBackWhenNothingResolves: a worker that refused to @@ -200,7 +202,9 @@ func TestStartupConnection_FallsBackWhenNothingResolves(t *testing.T) { } { t.Run(name, func(t *testing.T) { f := newFollower(t, testConfig(), table) - assert.Equal(t, primaryConn(), f.StartupConnection(context.Background())) + conn, reconnected := f.StartupConnection(context.Background()) + assert.Equal(t, primaryConn(), conn) + assert.False(t, reconnected, "falling back to the primary is not a resolved switch") }) } } diff --git a/pkg/hub/manager/manager.go b/pkg/hub/manager/manager.go index 99cb97741..503da8a93 100644 --- a/pkg/hub/manager/manager.go +++ b/pkg/hub/manager/manager.go @@ -69,7 +69,9 @@ func init() { // Start runs the hub-side manager against the hub described by conn. Callers // with a single hub pass hub.PrimaryConnection(), which is the same endpoint // and credentials this function used to assemble from the environment itself. -func Start(meshClient client.Client, hubClient client.Client, ctx context.Context, conn hub.Connection) { +// connInfo carries what main.go already knows about failover-following +// (issue #469) for the cluster reconciler's ControllerConnected condition. +func Start(meshClient client.Client, hubClient client.Client, ctx context.Context, conn hub.Connection, connInfo hubCluster.ConnectionInfo) { config := conn.RestConfig() var log = log.Log.WithName("hub") @@ -203,6 +205,7 @@ func Start(meshClient client.Client, hubClient client.Client, ctx context.Contex meshClient, &workerSliceEventRecorder, mf, + connInfo, ) err = builder. ControllerManagedBy(mgr). diff --git a/tests/hub/hub_suite_test.go b/tests/hub/hub_suite_test.go index c0658d08e..70dbeacab 100644 --- a/tests/hub/hub_suite_test.go +++ b/tests/hub/hub_suite_test.go @@ -222,6 +222,7 @@ var _ = BeforeSuite(func() { k8sClient, &testSliceEventRecorder, mf, + cluster.ConnectionInfo{}, ) clusterReconciler.ReconcileInterval = 5 * time.Second err = builder. From d4529e33254c8e43f22786485cd6b04290cdbeb0 Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Tue, 25 Aug 2026 16:56:29 +0530 Subject: [PATCH 09/10] docs(hub): document the new connection-health conditions, events and metrics Signed-off-by: Sumanth D --- docs/hub-failover.md | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/docs/hub-failover.md b/docs/hub-failover.md index 336ecceff..c9f0e4eeb 100644 --- a/docs/hub-failover.md +++ b/docs/hub-failover.md @@ -117,3 +117,33 @@ With both hubs up, the worker logs which hub it resolved at startup and then sta Active controller; once the Standby promotes itself, the worker logs `active hub changed`, increments `kubeslice_worker_hub_switches_total`, exits, and comes back connected to the promoted hub. Gateway pods should not restart at any point — that is the part worth watching. + +## Connection observability (#469) + +The worker's own `Cluster` CR on the hub carries two Conditions: + +| Condition | Meaning | +|---|---| +| `ControllerConnected` | Is this worker currently reconciling against a hub | +| `ControllerEndpointSynced` | Does the connected endpoint match the currently resolved active hub | + +Reasons: `Connected` (steady state), `ReconnectedAfterFailover` (the first reconcile after +following a resolved switch), `EndpointNotConfigured` (no secondary hub configured — the normal +non-HA case, status `Unknown`, not an error), `Reconnecting` (best-effort, written just before the +process restarts to follow a switch), `DialFailed`/`CertVerificationFailed`. + +`DialFailed` and `CertVerificationFailed` are real reason values — see +`pkg/hub/hubclient.ClassifyConnectionError`, unit-tested against fake dial errors — but by +construction they will rarely persist on the CR: writing either one needs the same connection that +just failed. Metrics and logs remain the reliable live-detection channel: + +- `kubeslice_worker_controller_reconnect_attempts_total{result}` — startup connection decisions + (`primary`, `resolved-switch`, `unresolved`), next to the existing `kubeslice_worker_hub_switches_total` + and `kubeslice_worker_hub_probe_errors_total` in `pkg/hub/failover/failover.go`. +- `kubeslice_worker_controller_last_sync_time_seconds` — when the last startup decision was made. + +Four Events, all on the worker's `Cluster` CR: `ControllerEndpointChanged` (fires once, the same +reconcile that reports `ReconnectedAfterFailover`), `ControllerConnected` (fires on the transition +into `Connected`), `ControllerConnectionLost` (best-effort, written from the `Watch` callback right +before the process restarts to follow a switch), and `CertVerificationFailed` (defined, not +currently fired from any live path — same limitation as the condition reason above). From 15cb0930bf661afe138b2e3435c65e8260703865 Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Tue, 25 Aug 2026 17:02:26 +0530 Subject: [PATCH 10/10] test(hub): cover setConnectionConditions idempotency and the reconnect handoff Fills the reconciler-level test #469 asked for but the prior commit skipped: non-HA, the one-shot ReconnectedAfterFailover-then-Connected handoff, LastTransitionTime idempotency, and that a live failure reason never gets set. Signed-off-by: Sumanth D --- .../controllers/cluster/conditions_test.go | 149 ++++++++++++++++++ 1 file changed, 149 insertions(+) create mode 100644 pkg/hub/controllers/cluster/conditions_test.go diff --git a/pkg/hub/controllers/cluster/conditions_test.go b/pkg/hub/controllers/cluster/conditions_test.go new file mode 100644 index 000000000..98fe72547 --- /dev/null +++ b/pkg/hub/controllers/cluster/conditions_test.go @@ -0,0 +1,149 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * 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 cluster + +import ( + "context" + "testing" + + hubv1alpha1 "github.com/kubeslice/apis/pkg/controller/v1alpha1" + mevents "github.com/kubeslice/kubeslice-monitoring/pkg/events" + "github.com/kubeslice/kubeslice-monitoring/pkg/metrics" + ossEvents "github.com/kubeslice/worker-operator/events" + hub "github.com/kubeslice/worker-operator/pkg/hub/hubclient" + utilmock "github.com/kubeslice/worker-operator/pkg/mocks" + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/scheme" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" +) + +// newConditionsReconciler builds a Reconciler wired to a mock client that +// accepts any Event Create, so tests can focus on cr.Status.Conditions and on +// how many events fired rather than on mock plumbing. +func newConditionsReconciler(t *testing.T, connInfo ConnectionInfo) (*Reconciler, *utilmock.MockClient) { + t.Helper() + client := utilmock.NewClient() + client.On("Create", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + testScheme := runtime.NewScheme() + if err := scheme.AddToScheme(testScheme); err != nil { + t.Fatalf("adding core scheme: %v", err) + } + testScheme.AddKnownTypeWithName(hubv1alpha1.GroupVersion.WithKind("Cluster"), &hubv1alpha1.Cluster{}) + + er := mevents.NewEventRecorder(client, testScheme, ossEvents.EventsMap, mevents.EventRecorderOptions{ + Cluster: "worker-1", + Project: "avesha", + Component: "worker-operator", + }) + mf, err := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) + if err != nil { + t.Fatalf("NewMetricsFactory: %v", err) + } + return NewReconciler(client, client, &er, mf, connInfo), client +} + +func newTestCluster() *hubv1alpha1.Cluster { + return &hubv1alpha1.Cluster{ + ObjectMeta: metav1.ObjectMeta{Name: "worker-1", Namespace: "avesha"}, + } +} + +// TestSetConnectionConditions_NonHA covers #469's EndpointNotConfigured case: +// every existing non-HA deployment, which must see an explicit Unknown state +// rather than either a misleading True/False or a silently absent condition. +func TestSetConnectionConditions_NonHA(t *testing.T) { + r, client := newConditionsReconciler(t, ConnectionInfo{Enabled: false}) + cr := newTestCluster() + + r.setConnectionConditions(context.Background(), cr) + + connected := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected) + assert.NotNil(t, connected) + assert.Equal(t, metav1.ConditionUnknown, connected.Status) + assert.Equal(t, ReasonEndpointNotConfigured, connected.Reason) + + synced := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerEndpointSynced) + assert.NotNil(t, synced) + assert.Equal(t, metav1.ConditionUnknown, synced.Status) + + client.AssertNotCalled(t, "Create", mock.Anything, mock.Anything, mock.Anything) +} + +// TestSetConnectionConditions_ReportsReconnectOnceThenSteadyState is #468 +// scenario 1 (clean failover/reconnect) at the reconciler level: the first +// reconcile after a resolved switch must say so, and every reconcile after +// that must report plain Connected, not repeat ReconnectedAfterFailover. +func TestSetConnectionConditions_ReportsReconnectOnceThenSteadyState(t *testing.T) { + r, client := newConditionsReconciler(t, ConnectionInfo{Enabled: true, Reconnected: true}) + cr := newTestCluster() + + r.setConnectionConditions(context.Background(), cr) + first := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected) + assert.Equal(t, metav1.ConditionTrue, first.Status) + assert.Equal(t, ReasonReconnectedAfterFailover, first.Reason) + + r.setConnectionConditions(context.Background(), cr) + second := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected) + assert.Equal(t, ReasonConnected, second.Reason, "every reconcile after the first must report plain Connected") + + // ControllerEndpointChanged fires once on the first call (announcing the + // failover), then ControllerConnected fires once on the second call (the + // transition into steady Connected) — not on the first, since its reason + // there is ReconnectedAfterFailover, not Connected. + client.AssertNumberOfCalls(t, "Create", 2) +} + +// TestSetConnectionConditions_Idempotent is #469's explicit acceptance +// criterion: calling this with the same outcome twice must not bump +// LastTransitionTime, or every steady-state reconcile would look like a +// fresh transition to anyone reading the CR. +func TestSetConnectionConditions_Idempotent(t *testing.T) { + r, client := newConditionsReconciler(t, ConnectionInfo{Enabled: true}) + cr := newTestCluster() + + r.setConnectionConditions(context.Background(), cr) + firstTransition := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected).LastTransitionTime + + r.setConnectionConditions(context.Background(), cr) + secondTransition := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected).LastTransitionTime + + assert.Equal(t, firstTransition, secondTransition, "an unchanged outcome must not bump LastTransitionTime") + // The second no-op reconcile must not fire another event. + client.AssertNumberOfCalls(t, "Create", 1) +} + +// TestSetConnectionConditions_NeverReportsLiveFailure documents the +// architectural limit explained in conditions.go: this method only ever +// leaves DialFailed/CertVerificationFailed absent, because it only runs at +// all on a hub this worker could just reach. +func TestSetConnectionConditions_NeverReportsLiveFailure(t *testing.T) { + r, _ := newConditionsReconciler(t, ConnectionInfo{Enabled: true}) + cr := newTestCluster() + + r.setConnectionConditions(context.Background(), cr) + + connected := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected) + assert.NotEqual(t, hub.ReasonDialFailed, connected.Reason) + assert.NotEqual(t, hub.ReasonCertVerificationFailed, connected.Reason) +}