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. 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") + } +} 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 new file mode 100644 index 000000000..d17dedad2 --- /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..2de4c55fd --- /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..56c894831 --- /dev/null +++ b/pkg/hub/resolver/resolver.go @@ -0,0 +1,343 @@ +/* + * 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 { + // 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. + 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..039ed2170 --- /dev/null +++ b/pkg/hub/resolver/resolver_test.go @@ -0,0 +1,361 @@ +/* + * 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) + + got := resolveN(r, 1) + 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. + 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 +} + +// 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) + } +}