From a1fc2bb437187053609fd605c0847db3bc0bf0ed Mon Sep 17 00:00:00 2001 From: Sumanth D Date: Sun, 2 Aug 2026 09:37:33 +0530 Subject: [PATCH] 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 +}