diff --git a/config/events/worker-events.yaml b/config/events/worker-events.yaml index a604d72f0..b97db91d4 100644 --- a/config/events/worker-events.yaml +++ b/config/events/worker-events.yaml @@ -508,4 +508,28 @@ events: action: None type: Warning reportingController: worker - message: Gateway recycling failed \ No newline at end of file + message: Gateway recycling failed + - name: ControllerEndpointChanged + reason: ControllerEndpointChanged + action: None + type: Normal + reportingController: worker + message: This worker followed a resolved controller failover and is now reconciling against a different hub endpoint + - name: ControllerConnected + reason: Connected + action: None + type: Normal + reportingController: worker + message: Connection to the hub controller is healthy + - name: ControllerConnectionLost + reason: DialFailed + action: None + type: Warning + reportingController: worker + message: The active hub changed and this worker is restarting to reconnect; the previous connection may already be lost + - name: CertVerificationFailed + reason: CertVerificationFailed + action: None + type: Warning + reportingController: worker + message: TLS handshake with the hub controller failed certificate verification; check the configured CA bundle \ No newline at end of file diff --git a/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..c9f0e4eeb --- /dev/null +++ b/docs/hub-failover.md @@ -0,0 +1,149 @@ +# 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. + +## Connection observability (#469) + +The worker's own `Cluster` CR on the hub carries two Conditions: + +| Condition | Meaning | +|---|---| +| `ControllerConnected` | Is this worker currently reconciling against a hub | +| `ControllerEndpointSynced` | Does the connected endpoint match the currently resolved active hub | + +Reasons: `Connected` (steady state), `ReconnectedAfterFailover` (the first reconcile after +following a resolved switch), `EndpointNotConfigured` (no secondary hub configured — the normal +non-HA case, status `Unknown`, not an error), `Reconnecting` (best-effort, written just before the +process restarts to follow a switch), `DialFailed`/`CertVerificationFailed`. + +`DialFailed` and `CertVerificationFailed` are real reason values — see +`pkg/hub/hubclient.ClassifyConnectionError`, unit-tested against fake dial errors — but by +construction they will rarely persist on the CR: writing either one needs the same connection that +just failed. Metrics and logs remain the reliable live-detection channel: + +- `kubeslice_worker_controller_reconnect_attempts_total{result}` — startup connection decisions + (`primary`, `resolved-switch`, `unresolved`), next to the existing `kubeslice_worker_hub_switches_total` + and `kubeslice_worker_hub_probe_errors_total` in `pkg/hub/failover/failover.go`. +- `kubeslice_worker_controller_last_sync_time_seconds` — when the last startup decision was made. + +Four Events, all on the worker's `Cluster` CR: `ControllerEndpointChanged` (fires once, the same +reconcile that reports `ReconnectedAfterFailover`), `ControllerConnected` (fires on the transition +into `Connected`), `ControllerConnectionLost` (best-effort, written from the `Watch` callback right +before the process restarts to follow a switch), and `CertVerificationFailed` (defined, not +currently fired from any live path — same limitation as the condition reason above). diff --git a/events/events_generated.go b/events/events_generated.go index 602aa1e7d..18721892a 100644 --- a/events/events_generated.go +++ b/events/events_generated.go @@ -702,6 +702,38 @@ var EventsMap = map[events.EventName]*events.EventSchema{ ReportingController: "worker", Message: "Gateway recycling failed", }, + "ControllerEndpointChanged": { + Name: "ControllerEndpointChanged", + Reason: "ControllerEndpointChanged", + Action: "None", + Type: events.EventTypeNormal, + ReportingController: "worker", + Message: "This worker followed a resolved controller failover and is now reconciling against a different hub endpoint", + }, + "ControllerConnected": { + Name: "ControllerConnected", + Reason: "Connected", + Action: "None", + Type: events.EventTypeNormal, + ReportingController: "worker", + Message: "Connection to the hub controller is healthy", + }, + "ControllerConnectionLost": { + Name: "ControllerConnectionLost", + Reason: "DialFailed", + Action: "None", + Type: events.EventTypeWarning, + ReportingController: "worker", + Message: "The active hub changed and this worker is restarting to reconnect; the previous connection may already be lost", + }, + "CertVerificationFailed": { + Name: "CertVerificationFailed", + Reason: "CertVerificationFailed", + Action: "None", + Type: events.EventTypeWarning, + ReportingController: "worker", + Message: "TLS handshake with the hub controller failed certificate verification; check the configured CA bundle", + }, } var ( @@ -790,4 +822,8 @@ var ( EventTriggeredFSMToRecycleGateways events.EventName = "TriggeredFSMToRecycleGateways" EventGatewayRecyclingSuccessful events.EventName = "GatewayRecyclingSuccessful" EventGatewayRecyclingFailed events.EventName = "GatewayRecyclingFailed" + EventControllerEndpointChanged events.EventName = "ControllerEndpointChanged" + EventControllerConnected events.EventName = "ControllerConnected" + EventControllerConnectionLost events.EventName = "ControllerConnectionLost" + EventCertVerificationFailed events.EventName = "CertVerificationFailed" ) diff --git a/go.mod b/go.mod index dd67f0aef..ae2689a53 100644 --- a/go.mod +++ b/go.mod @@ -102,3 +102,5 @@ require ( sigs.k8s.io/structured-merge-diff/v4 v4.4.2 // indirect sigs.k8s.io/yaml v1.4.0 // indirect ) + +replace github.com/kubeslice/apis => github.com/sumanthd032/apis v0.5.1-0.20260825105516-d7d920d4b404 diff --git a/go.sum b/go.sum index e7b3672f1..b0a5be56c 100644 --- a/go.sum +++ b/go.sum @@ -264,8 +264,6 @@ github.com/kr/pty v1.1.5/go.mod h1:9r2w37qlBe7rQ6e1fg1S/9xpWHSnaqNdHD3WcMdbPDA= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/kubeslice/apis v0.4.0 h1:nU66JoA2OQx48bZnXDWH8iHG+5R1ELX5ikc3l/fn5II= -github.com/kubeslice/apis v0.4.0/go.mod h1:F1hXnAt3Dk4Sto5yQDoMnqgXX5ImL1bRBiAmrW6TG00= github.com/kubeslice/gateway-sidecar v0.2.0 h1:Ja3fIUivuSjUFQ4lPCt79ATq99BxslvAFYUwV9Urpy4= github.com/kubeslice/gateway-sidecar v0.2.0/go.mod h1:nM1+Wjud2vk44cUg+9iwBbWTpqI+2Ecbn9NuaHEs9aY= github.com/kubeslice/kubeslice-monitoring v0.2.1 h1:wtmIEigpQoKzuckof7QRqdsaa4lV/rqxd/FcmOj5N5Q= @@ -374,6 +372,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/sumanthd032/apis v0.5.1-0.20260825105516-d7d920d4b404 h1:vlGmBE+ZeGdPhc5TwOZgd/0dQSSvLdW4sdoPG/VJu34= +github.com/sumanthd032/apis v0.5.1-0.20260825105516-d7d920d4b404/go.mod h1:F1hXnAt3Dk4Sto5yQDoMnqgXX5ImL1bRBiAmrW6TG00= github.com/vishvananda/netlink v1.2.1-beta.2.0.20220812183158-d44b87fd4d3f h1:kTcjiSlfxwj/o7ezwNrsYjydHYu4Bgu0OkF+q46Vb3k= github.com/vishvananda/netlink v1.2.1-beta.2.0.20220812183158-d44b87fd4d3f/go.mod h1:cAAsePK2e15YDAMJNyOpGYEWNe4sIghTY7gpz4cX/Ik= github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0= diff --git a/main.go b/main.go index a95f367e9..c52f10cbf 100644 --- a/main.go +++ b/main.go @@ -19,6 +19,7 @@ package main import ( + "context" "flag" "os" "strings" @@ -41,6 +42,8 @@ import ( netop "github.com/kubeslice/worker-operator/pkg/netop" router "github.com/kubeslice/worker-operator/pkg/router" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" @@ -64,9 +67,13 @@ import ( "github.com/kubeslice/worker-operator/controllers/serviceimport" "github.com/kubeslice/worker-operator/controllers/slice" "github.com/kubeslice/worker-operator/controllers/slicegateway" + hubv1alpha1 "github.com/kubeslice/apis/pkg/controller/v1alpha1" ossEvents "github.com/kubeslice/worker-operator/events" + hubCluster "github.com/kubeslice/worker-operator/pkg/hub/controllers/cluster" + "github.com/kubeslice/worker-operator/pkg/hub/failover" hub "github.com/kubeslice/worker-operator/pkg/hub/hubclient" "github.com/kubeslice/worker-operator/pkg/hub/manager" + "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 +172,24 @@ 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 + var reconnected bool + if failoverCfg.Enabled() { + hubFollower, err = failover.New(failoverCfg, hubConn, ctrl.Log.WithName("hub-failover"), nil) + if err != nil { + setupLog.With("error", err).Error("could not configure hub failover following") + os.Exit(1) + } + hubConn, reconnected = hubFollower.StartupConnection(context.Background()) + } + connInfo := hubCluster.ConnectionInfo{Enabled: failoverCfg.Enabled(), Reconnected: reconnected} + + 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 +217,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,12 +338,56 @@ func main() { } go func() { setupLog.Info("starting hub manager") - manager.Start(clientForHubMgr, hubClient, ctx) + manager.Start(clientForHubMgr, hubClient, ctx, hubConn, connInfo) }() + 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") + reportConnectionLost(hubClient, &sliceEventRecorder) + stopForHubSwitch() + }) + } + setupLog.Info("starting manager") if err := mgr.Start(ctx); err != nil { setupLog.With("error", err).Error("problem running manager") os.Exit(1) } } + +// reportConnectionLost makes one best-effort attempt to record, on the hub +// this worker is about to leave, that it is doing so (issue #469's +// ControllerConnectionLost event and Reconnecting condition). Best-effort +// because the only connection available to write with is the one this +// worker is abandoning — if that hub is itself the reason for the switch, +// there is nothing to write to, and that failure is expected, not fatal. +func reportConnectionLost(hubClient client.Client, er *monitoringEvents.EventRecorder) { + cr := &hubv1alpha1.Cluster{} + err := hubClient.Get(context.Background(), client.ObjectKey{ + Name: controllers.ClusterName, + Namespace: hub.ProjectNamespace, + }, cr) + if err != nil { + setupLog.With("error", err).Info("could not report connection loss before reconnecting; the hub may already be unreachable") + return + } + meta.SetStatusCondition(&cr.Status.Conditions, metav1.Condition{ + Type: hubCluster.ConditionControllerConnected, + Status: metav1.ConditionUnknown, + Reason: hubCluster.ReasonReconnecting, + Message: "following a resolved hub failover; reconnecting", + }) + if err := hubClient.Status().Update(context.Background(), cr); err != nil { + setupLog.With("error", err).Info("could not persist the Reconnecting condition before restart") + return + } + utils.RecordEvent(context.Background(), er, cr, nil, ossEvents.EventControllerConnectionLost, "hub-failover") +} diff --git a/pkg/hub/controllers/cluster/cluster_suite_test.go b/pkg/hub/controllers/cluster/cluster_suite_test.go index c7b140b06..6d97d6a5d 100644 --- a/pkg/hub/controllers/cluster/cluster_suite_test.go +++ b/pkg/hub/controllers/cluster/cluster_suite_test.go @@ -132,6 +132,7 @@ var _ = BeforeSuite(func() { k8sClient, &spokeClusterEventRecorder, mf, + ConnectionInfo{}, ) clusterReconciler.ReconcileInterval = 5 * time.Second err = builder. diff --git a/pkg/hub/controllers/cluster/conditions.go b/pkg/hub/controllers/cluster/conditions.go new file mode 100644 index 000000000..449c43654 --- /dev/null +++ b/pkg/hub/controllers/cluster/conditions.go @@ -0,0 +1,107 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package cluster + +import ( + "context" + + hubv1alpha1 "github.com/kubeslice/apis/pkg/controller/v1alpha1" + ossEvents "github.com/kubeslice/worker-operator/events" + "github.com/kubeslice/worker-operator/pkg/utils" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// Condition types and reasons for issue #469. Reason strings are shared with +// pkg/hub/hubclient.ClassifyConnectionError where they overlap +// (DialFailed/CertVerificationFailed) so a caller wiring one into the other +// cannot typo a mismatch. +const ( + ConditionControllerConnected = "ControllerConnected" + ConditionControllerEndpointSynced = "ControllerEndpointSynced" + + ReasonConnected = "Connected" + ReasonReconnectedAfterFailover = "ReconnectedAfterFailover" + ReasonEndpointNotConfigured = "EndpointNotConfigured" + ReasonReconnecting = "Reconnecting" + ReasonEndpointUpToDate = "EndpointUpToDate" +) + +// ConnectionInfo is what main.go already knows about this worker's hub +// connection before the hub manager starts: whether failover-following is +// configured at all, and whether the connection now in use was reached by +// following a resolved switch rather than the configured primary. It is a +// startup-time fact, not a live signal. +type ConnectionInfo struct { + Enabled bool + Reconnected bool +} + +// setConnectionConditions records this worker's hub-connection health on its +// own Cluster CR, via the standard idempotent SetStatusCondition helper (a +// repeated no-op call never bumps LastTransitionTime). +// +// This only ever sets Connected, ReconnectedAfterFailover or +// EndpointNotConfigured — never DialFailed or CertVerificationFailed. Those +// two are real reason values (see pkg/hub/hubclient.ClassifyConnectionError, +// unit-tested there) but this method only runs because Reconcile just read cr +// successfully: a hub this worker cannot reach is a hub it cannot write +// "unreachable" to either, on that same hub's own copy of the object. That is +// the same conclusion issue #467 reached before deferring the durable surface +// here. kubeslice_worker_hub_probe_errors_total and logs remain the live +// down-detection channel; this condition is the durable, hub-side-visible +// record of the connection's last known good state. +func (r *Reconciler) setConnectionConditions(ctx context.Context, cr *hubv1alpha1.Cluster) { + connected := metav1.Condition{ + Type: ConditionControllerConnected, + Status: metav1.ConditionTrue, + Reason: ReasonConnected, + Message: "worker is reconciling against this hub", + } + synced := metav1.Condition{ + Type: ConditionControllerEndpointSynced, + Status: metav1.ConditionTrue, + Reason: ReasonEndpointUpToDate, + Message: "connected to the currently resolved active hub", + } + + switch { + case !r.connInfo.Enabled: + connected.Status = metav1.ConditionUnknown + connected.Reason = ReasonEndpointNotConfigured + connected.Message = "no secondary hub configured; failover-following is inactive" + synced.Status = metav1.ConditionUnknown + synced.Reason = ReasonEndpointNotConfigured + synced.Message = "no secondary hub configured; failover-following is inactive" + case r.connInfo.Reconnected && !r.reportedReconnect: + connected.Reason = ReasonReconnectedAfterFailover + connected.Message = "resumed reconciliation after following a resolved hub failover" + } + + changed := meta.SetStatusCondition(&cr.Status.Conditions, connected) + meta.SetStatusCondition(&cr.Status.Conditions, synced) + + if r.connInfo.Reconnected && !r.reportedReconnect { + utils.RecordEvent(ctx, r.EventRecorder, cr, nil, ossEvents.EventControllerEndpointChanged, controllerName) + r.reportedReconnect = true + } + if changed && connected.Reason == ReasonConnected { + utils.RecordEvent(ctx, r.EventRecorder, cr, nil, ossEvents.EventControllerConnected, controllerName) + } +} diff --git a/pkg/hub/controllers/cluster/conditions_test.go b/pkg/hub/controllers/cluster/conditions_test.go new file mode 100644 index 000000000..98fe72547 --- /dev/null +++ b/pkg/hub/controllers/cluster/conditions_test.go @@ -0,0 +1,149 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package cluster + +import ( + "context" + "testing" + + hubv1alpha1 "github.com/kubeslice/apis/pkg/controller/v1alpha1" + mevents "github.com/kubeslice/kubeslice-monitoring/pkg/events" + "github.com/kubeslice/kubeslice-monitoring/pkg/metrics" + ossEvents "github.com/kubeslice/worker-operator/events" + hub "github.com/kubeslice/worker-operator/pkg/hub/hubclient" + utilmock "github.com/kubeslice/worker-operator/pkg/mocks" + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/scheme" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" +) + +// newConditionsReconciler builds a Reconciler wired to a mock client that +// accepts any Event Create, so tests can focus on cr.Status.Conditions and on +// how many events fired rather than on mock plumbing. +func newConditionsReconciler(t *testing.T, connInfo ConnectionInfo) (*Reconciler, *utilmock.MockClient) { + t.Helper() + client := utilmock.NewClient() + client.On("Create", mock.Anything, mock.Anything, mock.Anything).Return(nil) + + testScheme := runtime.NewScheme() + if err := scheme.AddToScheme(testScheme); err != nil { + t.Fatalf("adding core scheme: %v", err) + } + testScheme.AddKnownTypeWithName(hubv1alpha1.GroupVersion.WithKind("Cluster"), &hubv1alpha1.Cluster{}) + + er := mevents.NewEventRecorder(client, testScheme, ossEvents.EventsMap, mevents.EventRecorderOptions{ + Cluster: "worker-1", + Project: "avesha", + Component: "worker-operator", + }) + mf, err := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) + if err != nil { + t.Fatalf("NewMetricsFactory: %v", err) + } + return NewReconciler(client, client, &er, mf, connInfo), client +} + +func newTestCluster() *hubv1alpha1.Cluster { + return &hubv1alpha1.Cluster{ + ObjectMeta: metav1.ObjectMeta{Name: "worker-1", Namespace: "avesha"}, + } +} + +// TestSetConnectionConditions_NonHA covers #469's EndpointNotConfigured case: +// every existing non-HA deployment, which must see an explicit Unknown state +// rather than either a misleading True/False or a silently absent condition. +func TestSetConnectionConditions_NonHA(t *testing.T) { + r, client := newConditionsReconciler(t, ConnectionInfo{Enabled: false}) + cr := newTestCluster() + + r.setConnectionConditions(context.Background(), cr) + + connected := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected) + assert.NotNil(t, connected) + assert.Equal(t, metav1.ConditionUnknown, connected.Status) + assert.Equal(t, ReasonEndpointNotConfigured, connected.Reason) + + synced := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerEndpointSynced) + assert.NotNil(t, synced) + assert.Equal(t, metav1.ConditionUnknown, synced.Status) + + client.AssertNotCalled(t, "Create", mock.Anything, mock.Anything, mock.Anything) +} + +// TestSetConnectionConditions_ReportsReconnectOnceThenSteadyState is #468 +// scenario 1 (clean failover/reconnect) at the reconciler level: the first +// reconcile after a resolved switch must say so, and every reconcile after +// that must report plain Connected, not repeat ReconnectedAfterFailover. +func TestSetConnectionConditions_ReportsReconnectOnceThenSteadyState(t *testing.T) { + r, client := newConditionsReconciler(t, ConnectionInfo{Enabled: true, Reconnected: true}) + cr := newTestCluster() + + r.setConnectionConditions(context.Background(), cr) + first := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected) + assert.Equal(t, metav1.ConditionTrue, first.Status) + assert.Equal(t, ReasonReconnectedAfterFailover, first.Reason) + + r.setConnectionConditions(context.Background(), cr) + second := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected) + assert.Equal(t, ReasonConnected, second.Reason, "every reconcile after the first must report plain Connected") + + // ControllerEndpointChanged fires once on the first call (announcing the + // failover), then ControllerConnected fires once on the second call (the + // transition into steady Connected) — not on the first, since its reason + // there is ReconnectedAfterFailover, not Connected. + client.AssertNumberOfCalls(t, "Create", 2) +} + +// TestSetConnectionConditions_Idempotent is #469's explicit acceptance +// criterion: calling this with the same outcome twice must not bump +// LastTransitionTime, or every steady-state reconcile would look like a +// fresh transition to anyone reading the CR. +func TestSetConnectionConditions_Idempotent(t *testing.T) { + r, client := newConditionsReconciler(t, ConnectionInfo{Enabled: true}) + cr := newTestCluster() + + r.setConnectionConditions(context.Background(), cr) + firstTransition := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected).LastTransitionTime + + r.setConnectionConditions(context.Background(), cr) + secondTransition := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected).LastTransitionTime + + assert.Equal(t, firstTransition, secondTransition, "an unchanged outcome must not bump LastTransitionTime") + // The second no-op reconcile must not fire another event. + client.AssertNumberOfCalls(t, "Create", 1) +} + +// TestSetConnectionConditions_NeverReportsLiveFailure documents the +// architectural limit explained in conditions.go: this method only ever +// leaves DialFailed/CertVerificationFailed absent, because it only runs at +// all on a hub this worker could just reach. +func TestSetConnectionConditions_NeverReportsLiveFailure(t *testing.T) { + r, _ := newConditionsReconciler(t, ConnectionInfo{Enabled: true}) + cr := newTestCluster() + + r.setConnectionConditions(context.Background(), cr) + + connected := meta.FindStatusCondition(cr.Status.Conditions, ConditionControllerConnected) + assert.NotEqual(t, hub.ReasonDialFailed, connected.Reason) + assert.NotEqual(t, hub.ReasonCertVerificationFailed, connected.Reason) +} diff --git a/pkg/hub/controllers/cluster/deregister_unit_test.go b/pkg/hub/controllers/cluster/deregister_unit_test.go index dde36f003..5d978f85d 100644 --- a/pkg/hub/controllers/cluster/deregister_unit_test.go +++ b/pkg/hub/controllers/cluster/deregister_unit_test.go @@ -76,7 +76,7 @@ func TestGetOperatorClusterRole(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterRoleKey := types.NamespacedName{Name: operatorClusterRoleName} @@ -108,7 +108,7 @@ func TestCreateDeregisterJobPositiveScenarios(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -215,7 +215,7 @@ func TestReconcilerFailToUpdateClusterRegistrationStatus(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -258,7 +258,7 @@ func TestReconcilerFailToCreateServiceAccount(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -316,7 +316,7 @@ func TestReconcilerFailToFetchOperatorClusterRole(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -384,7 +384,7 @@ func TestReconcilerFailToCreateClusterRole(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -462,7 +462,7 @@ func TestReconcilerFailToCreateClusterRoleBinding(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -545,7 +545,7 @@ func TestReconcilerFailToCreateConfigmap(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -642,7 +642,7 @@ func TestReconcilerFailToDeleteJob(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -750,7 +750,7 @@ func TestReconcilerFailToCreateDeregisterJob(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} diff --git a/pkg/hub/controllers/cluster/reconciler.go b/pkg/hub/controllers/cluster/reconciler.go index 7c4d25026..f380cfb3e 100644 --- a/pkg/hub/controllers/cluster/reconciler.go +++ b/pkg/hub/controllers/cluster/reconciler.go @@ -67,9 +67,17 @@ type Reconciler struct { gaugeComponentUp *prometheus.GaugeVec ReconcileInterval time.Duration + + // connInfo and reportedReconnect back setConnectionConditions (issue + // #469). reportedReconnect flips true the first time this process reports + // ReconnectedAfterFailover, so every later reconcile reports plain + // Connected instead — connInfo.Reconnected itself never changes for the + // life of the process. + connInfo ConnectionInfo + reportedReconnect bool } -func NewReconciler(c client.Client, mc client.Client, er *events.EventRecorder, mf metrics.MetricsFactory) *Reconciler { +func NewReconciler(c client.Client, mc client.Client, er *events.EventRecorder, mf metrics.MetricsFactory, connInfo ConnectionInfo) *Reconciler { gaugeClusterUp := mf.NewGauge("cluster_up", "Kubeslice cluster health status", []string{}) gaugeComponentUp := mf.NewGauge("cluster_component_up", "Kubeslice cluster component health status", []string{"slice_cluster_component"}) @@ -82,6 +90,8 @@ func NewReconciler(c client.Client, mc client.Client, er *events.EventRecorder, gaugeComponentUp: gaugeComponentUp, ReconcileInterval: 120 * time.Second, + + connInfo: connInfo, } } @@ -173,6 +183,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco } r.updateClusterMetrics(cr) + r.setConnectionConditions(ctx, cr) if time.Since(cr.Status.ClusterHealth.LastUpdated.Time) > r.ReconcileInterval { cr.Status.ClusterHealth.LastUpdated = metav1.Now() diff --git a/pkg/hub/controllers/cluster/reconciler_unit_test.go b/pkg/hub/controllers/cluster/reconciler_unit_test.go index 595341eb2..e5e556572 100644 --- a/pkg/hub/controllers/cluster/reconciler_unit_test.go +++ b/pkg/hub/controllers/cluster/reconciler_unit_test.go @@ -65,7 +65,7 @@ func TestReconcileToReturnErrorWhileFetchingControllerCluster(t *testing.T) { ).Return(errors.New("object not found")) mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) result, err := reconciler.Reconcile(expected.ctx, expected.req) if expected.res != result { @@ -106,7 +106,7 @@ func TestReconcileToCallHandleClusterDeletion(t *testing.T) { ).Return(errors.New("failed to update cluster CR")) mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) result, err := reconciler.Reconcile(expected.ctx, expected.req) if expected.res != result { @@ -133,7 +133,7 @@ func TestReconcilerHandleClusterDeletion(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObj) @@ -166,7 +166,7 @@ func TestReconcilerHandleExternalDependency(t *testing.T) { client := utilmock.NewClient() mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, nil, mf) + reconciler := NewReconciler(client, client, nil, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObjWithFinalizer) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} @@ -283,7 +283,7 @@ func TestReconcilerToFailWhileCallingCreateDeregisterJob(t *testing.T) { Namespace: controllers.ControlPlaneNamespace, }) mf, _ := metrics.NewMetricsFactory(prometheus.NewRegistry(), metrics.MetricsFactoryOptions{}) - reconciler := NewReconciler(client, client, &testClusterEventRecorder, mf) + reconciler := NewReconciler(client, client, &testClusterEventRecorder, mf, ConnectionInfo{}) reconciler.InjectClient(client) ctx := context.WithValue(context.Background(), types.NamespacedName{Name: testClusterName, Namespace: testProjectNamespace}, testClusterObjWithFinalizer) clusterKey := types.NamespacedName{Namespace: testProjectNamespace, Name: testClusterName} diff --git a/pkg/hub/failover/failover.go b/pkg/hub/failover/failover.go new file mode 100644 index 000000000..0a5ccd4ac --- /dev/null +++ b/pkg/hub/failover/failover.go @@ -0,0 +1,310 @@ +/* + * 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"}) + // controllerReconnectAttemptsTotal counts every resolved winner this + // worker acted on at startup, by whether it matched the configured + // primary or a resolved switch. It is deliberately not "every poll" — + // gather() already runs every tick and hubProbeErrorsTotal covers probe + // failures; this metric is about connection *decisions*, issue #469's ask. + controllerReconnectAttemptsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "kubeslice_worker_controller_reconnect_attempts_total", + Help: "Startup hub-connection decisions this worker has made, by result.", + }, []string{"result"}) + controllerLastSyncTime = prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "kubeslice_worker_controller_last_sync_time_seconds", + Help: "Unix time of the last resolved hub-connection decision at startup.", + }) +) + +func init() { + ctrlmetrics.Registry.MustRegister(hubSwitchesTotal, hubProbeErrorsTotal, + controllerReconnectAttemptsTotal, controllerLastSyncTime) +} + +// 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, plus +// whether that hub was reached by following a resolved switch rather than +// falling back to the configured primary. The bool is issue #469's +// "ReconnectedAfterFailover vs Connected" distinction: main.go has no other +// way to know, since a fresh process looks identical whether it just started +// normally or was just restarted to follow a failover. +// +// Falling back to the primary when nothing resolves is deliberate. A worker +// that refused to start because it could not reach a hub would turn a hub +// outage into a worker outage, and the primary is the same hub it would have +// used before any of this existed. +func (f *Follower) StartupConnection(ctx context.Context) (hub.Connection, bool) { + claim := f.resolver.Resolve(ctx) + if claim == nil { + f.log.Info("no active hub resolved at startup; using the configured primary", + "endpoint", f.primary.Endpoint) + controllerReconnectAttemptsTotal.WithLabelValues("unresolved").Inc() + controllerLastSyncTime.SetToCurrentTime() + return f.primary, false + } + 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) + controllerReconnectAttemptsTotal.WithLabelValues("unresolved").Inc() + controllerLastSyncTime.SetToCurrentTime() + return f.primary, false + } + f.log.Info("connecting to the resolved active hub", + "identity", claim.Identity, "endpoint", conn.Endpoint) + viaSwitch := conn.Endpoint != f.primary.Endpoint + result := "primary" + if viaSwitch { + result = "resolved-switch" + } + controllerReconnectAttemptsTotal.WithLabelValues(result).Inc() + controllerLastSyncTime.SetToCurrentTime() + return conn, viaSwitch +} + +// Watch polls until ctx is done, calling onSwitch once the resolved active hub +// 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..54ecafe5a --- /dev/null +++ b/pkg/hub/failover/failover_test.go @@ -0,0 +1,291 @@ +/* + * 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, reconnected := f.StartupConnection(context.Background()) + assert.Equal(t, secondaryEndpoint, conn.Endpoint) + assert.Equal(t, "/creds/b/token", conn.TokenFile, "the winner's token must travel with its endpoint") + assert.Equal(t, "/creds/b/ca.crt", conn.CAFile, "the winner's CA must travel with its endpoint") + assert.True(t, reconnected, "a winner other than the primary is a resolved switch") +} + +func TestStartupConnection_SteadyStateStaysOnThePrimary(t *testing.T) { + 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, reconnected := f.StartupConnection(context.Background()) + assert.Equal(t, primaryConn(), conn, "both hubs naming hub A is agreement, not a conflict") + assert.False(t, reconnected, "staying on the primary is not a resolved switch") +} + +// TestStartupConnection_FallsBackWhenNothingResolves: a worker that refused to +// 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) + conn, reconnected := f.StartupConnection(context.Background()) + assert.Equal(t, primaryConn(), conn) + assert.False(t, reconnected, "falling back to the primary is not a resolved switch") + }) + } +} + +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/classify.go b/pkg/hub/hubclient/classify.go new file mode 100644 index 000000000..47690667c --- /dev/null +++ b/pkg/hub/hubclient/classify.go @@ -0,0 +1,47 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package hub + +import ( + "crypto/x509" + "errors" +) + +// Connection error reasons, matching worker-operator issue #469's +// ControllerConnected reason table. DialFailed is the default for anything +// that isn't specifically a certificate problem: a refused connection, a DNS +// failure, and a timeout all look the same to an operator deciding what to +// check first, and all point at network/endpoint configuration rather than +// the CA bundle. +const ( + ReasonDialFailed = "DialFailed" + ReasonCertVerificationFailed = "CertVerificationFailed" +) + +// ClassifyConnectionError reports which of the two reasons a hub-connection +// error matches. err must be non-nil. +func ClassifyConnectionError(err error) string { + var unknownAuthority x509.UnknownAuthorityError + var certInvalid x509.CertificateInvalidError + var hostnameErr x509.HostnameError + if errors.As(err, &unknownAuthority) || errors.As(err, &certInvalid) || errors.As(err, &hostnameErr) { + return ReasonCertVerificationFailed + } + return ReasonDialFailed +} diff --git a/pkg/hub/hubclient/classify_test.go b/pkg/hub/hubclient/classify_test.go new file mode 100644 index 000000000..9b5e1b037 --- /dev/null +++ b/pkg/hub/hubclient/classify_test.go @@ -0,0 +1,68 @@ +/* + * Copyright (c) 2022 Avesha, Inc. All rights reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package hub + +import ( + "crypto/x509" + "errors" + "fmt" + "net" + "net/url" + "testing" +) + +func TestClassifyConnectionError_CertFailuresAreDistinguished(t *testing.T) { + cases := []struct { + name string + err error + }{ + {"unknown authority", x509.UnknownAuthorityError{}}, + {"certificate invalid", x509.CertificateInvalidError{Reason: x509.Expired}}, + {"hostname mismatch", x509.HostnameError{Certificate: &x509.Certificate{}, Host: "hub.example"}}, + {"wrapped in a url.Error, as a real client-go transport failure would be", + &url.Error{Op: "Get", URL: "https://hub.example", Err: x509.UnknownAuthorityError{}}}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got := ClassifyConnectionError(c.err) + if got != ReasonCertVerificationFailed { + t.Fatalf("ClassifyConnectionError(%v) = %q, want %q", c.err, got, ReasonCertVerificationFailed) + } + }) + } +} + +func TestClassifyConnectionError_EverythingElseIsADialFailure(t *testing.T) { + cases := []struct { + name string + err error + }{ + {"connection refused", &net.OpError{Op: "dial", Err: errors.New("connection refused")}}, + {"generic timeout", fmt.Errorf("context deadline exceeded")}, + {"not found, unrelated to connectivity", errors.New("clusters.kubeslice.io \"worker-1\" not found")}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got := ClassifyConnectionError(c.err) + if got != ReasonDialFailed { + t.Fatalf("ClassifyConnectionError(%v) = %q, want %q", c.err, got, ReasonDialFailed) + } + }) + } +} 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..503da8a93 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,17 @@ 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. +// connInfo carries what main.go already knows about failover-following +// (issue #469) for the cluster reconciler's ControllerConnected condition. +func Start(meshClient client.Client, hubClient client.Client, ctx context.Context, conn hub.Connection, connInfo hubCluster.ConnectionInfo) { + config := conn.RestConfig() var log = log.Log.WithName("hub") - 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, @@ -206,6 +205,7 @@ func Start(meshClient client.Client, hubClient client.Client, ctx context.Contex meshClient, &workerSliceEventRecorder, mf, + connInfo, ) err = builder. ControllerManagedBy(mgr). diff --git a/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) + } +} diff --git a/tests/hub/hub_suite_test.go b/tests/hub/hub_suite_test.go index c0658d08e..70dbeacab 100644 --- a/tests/hub/hub_suite_test.go +++ b/tests/hub/hub_suite_test.go @@ -222,6 +222,7 @@ var _ = BeforeSuite(func() { k8sClient, &testSliceEventRecorder, mf, + cluster.ConnectionInfo{}, ) clusterReconciler.ReconcileInterval = 5 * time.Second err = builder. diff --git a/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/cluster_types.go b/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/cluster_types.go index dda4a7f21..ad61ba3d7 100644 --- a/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/cluster_types.go +++ b/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/cluster_types.go @@ -137,6 +137,58 @@ type ClusterStatus struct { // VCPURestriction is the restriction on the cluster disabling the creation of new pods VCPURestriction *VCPURestriction `json:"vCPURestriction,omitempty"` GPURestriction *GPURestriction `json:"GPURestriction,omitempty"` + // StorageCapabilities contains auto-detected storage capabilities reported by the worker operator. + // Populated only when the worker operator's storage-capability reconciler is active. + StorageCapabilities *StorageCapabilities `json:"storageCapabilities,omitempty"` + // ActiveController identifies the hub controller that currently holds leadership. + // Populated only on an Active/Standby HA deployment; absent otherwise, so a + // non-HA worker sees no behaviour change. + ActiveController *ActiveControllerInfo `json:"activeController,omitempty"` + // Conditions describe this worker's connection to its hub controller, e.g. + // ControllerConnected and ControllerEndpointSynced. Written by the worker + // operator about itself; absent on a worker that has never reported one. + Conditions []metav1.Condition `json:"conditions,omitempty"` +} + +// ActiveControllerInfo describes the hub controller currently holding leadership. +// +// Each hub writes this field about itself, on its own API server, and only while +// it holds leadership. A Standby's copy is populated by the state mirror from the +// Active, so it names the Active rather than itself — which lets a worker watching +// both hubs identify the Active by the rule "trust whichever endpoint is reachable +// and reports an ActiveIdentity matching that endpoint's own identity", without +// needing to know which role either hub currently holds. +type ActiveControllerInfo struct { + // Endpoint is the API server endpoint of the hub currently holding leadership + Endpoint string `json:"endpoint,omitempty"` + // CABundle is the base64-encoded PEM CA bundle for Endpoint + CABundle string `json:"caBundle,omitempty"` + // ActiveIdentity is the HA identity of the hub that wrote this field about itself + ActiveIdentity string `json:"activeIdentity,omitempty"` + // LastUpdated is the timestamp when this declaration was last written. It gives + // a consumer a deterministic tie-break if both hubs self-declare simultaneously. + LastUpdated metav1.Time `json:"lastUpdated,omitempty"` +} + +// StorageCapabilities holds auto-detected RWX-capable storage classes on the worker cluster. +// To add support for a new storage system, append its CSI provisioner string to the +// worker operator's rwxProvisioners list — no changes to this struct are required. +type StorageCapabilities struct { + // RWXStorageClasses lists all ReadWriteMany-capable StorageClasses detected on the cluster + RWXStorageClasses []RWXStorageClass `json:"rwxStorageClasses,omitempty"` + // DefaultStorageClass is the name of the StorageClass annotated with + // storageclass.kubernetes.io/is-default-class: "true" on the worker cluster + DefaultStorageClass string `json:"defaultStorageClass,omitempty"` + // LastUpdated is the timestamp when capabilities were last detected + LastUpdated metav1.Time `json:"lastUpdated,omitempty"` +} + +// RWXStorageClass describes a single ReadWriteMany-capable StorageClass. +type RWXStorageClass struct { + // Name is the StorageClass name (e.g. "rook-cephfs", "juicefs-sc") + Name string `json:"name"` + // Provisioner is the CSI provisioner string (e.g. "rook-ceph.cephfs.csi.ceph.com") + Provisioner string `json:"provisioner"` } type GPURestriction struct { diff --git a/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/zz_generated.deepcopy.go b/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/zz_generated.deepcopy.go index baeea08ce..6215f0f86 100644 --- a/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/zz_generated.deepcopy.go +++ b/vendor/github.com/kubeslice/apis/pkg/controller/v1alpha1/zz_generated.deepcopy.go @@ -20,9 +20,26 @@ package v1alpha1 import ( + "k8s.io/apimachinery/pkg/apis/meta/v1" runtime "k8s.io/apimachinery/pkg/runtime" ) +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ActiveControllerInfo) DeepCopyInto(out *ActiveControllerInfo) { + *out = *in + in.LastUpdated.DeepCopyInto(&out.LastUpdated) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ActiveControllerInfo. +func (in *ActiveControllerInfo) DeepCopy() *ActiveControllerInfo { + if in == nil { + return nil + } + out := new(ActiveControllerInfo) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Cluster) DeepCopyInto(out *Cluster) { *out = *in @@ -175,6 +192,23 @@ func (in *ClusterStatus) DeepCopyInto(out *ClusterStatus) { *out = new(GPURestriction) (*in).DeepCopyInto(*out) } + if in.StorageCapabilities != nil { + in, out := &in.StorageCapabilities, &out.StorageCapabilities + *out = new(StorageCapabilities) + (*in).DeepCopyInto(*out) + } + if in.ActiveController != nil { + in, out := &in.ActiveController, &out.ActiveController + *out = new(ActiveControllerInfo) + (*in).DeepCopyInto(*out) + } + if in.Conditions != nil { + in, out := &in.Conditions, &out.Conditions + *out = make([]v1.Condition, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterStatus. @@ -473,6 +507,21 @@ func (in *QOSProfile) DeepCopy() *QOSProfile { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *RWXStorageClass) DeepCopyInto(out *RWXStorageClass) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RWXStorageClass. +func (in *RWXStorageClass) DeepCopy() *RWXStorageClass { + if in == nil { + return nil + } + out := new(RWXStorageClass) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ServiceAccess) DeepCopyInto(out *ServiceAccess) { *out = *in @@ -917,6 +966,27 @@ func (in *StatusOfKeyRotation) DeepCopy() *StatusOfKeyRotation { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *StorageCapabilities) DeepCopyInto(out *StorageCapabilities) { + *out = *in + if in.RWXStorageClasses != nil { + in, out := &in.RWXStorageClasses, &out.RWXStorageClasses + *out = make([]RWXStorageClass, len(*in)) + copy(*out, *in) + } + in.LastUpdated.DeepCopyInto(&out.LastUpdated) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new StorageCapabilities. +func (in *StorageCapabilities) DeepCopy() *StorageCapabilities { + if in == nil { + return nil + } + out := new(StorageCapabilities) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Telemetry) DeepCopyInto(out *Telemetry) { *out = *in diff --git a/vendor/modules.txt b/vendor/modules.txt index 21e06e176..dfaf43aa1 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -116,7 +116,7 @@ github.com/josharian/intern # github.com/json-iterator/go v1.1.12 ## explicit; go 1.12 github.com/json-iterator/go -# github.com/kubeslice/apis v0.4.0 +# github.com/kubeslice/apis v0.4.0 => github.com/sumanthd032/apis v0.5.1-0.20260825105516-d7d920d4b404 ## explicit; go 1.24.0 github.com/kubeslice/apis/pkg/controller/v1alpha1 github.com/kubeslice/apis/pkg/worker/v1alpha1 @@ -977,3 +977,4 @@ sigs.k8s.io/structured-merge-diff/v4/value ## explicit; go 1.12 sigs.k8s.io/yaml sigs.k8s.io/yaml/goyaml.v2 +# github.com/kubeslice/apis => github.com/sumanthd032/apis v0.5.1-0.20260825105516-d7d920d4b404