diff --git a/README.md b/README.md index 3955e95..92dba10 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,7 @@ configurable values. ```sh helm upgrade hyperbytedb-operator \ oci://ghcr.io/hyperbyte-cloud/charts/hyperbytedb-operator \ - --version 0.2.0 \ + --version 0.3.0 \ --namespace hyperbytedb-operator-system \ --reuse-values ``` @@ -142,6 +142,13 @@ The operator manages three CRDs: See `config/samples/` for example manifests. +## HyperbyteDB versions + +The operator works out which HyperbyteDB version each cluster runs (from +`spec.version`, the image tag, or the `hyperbytedb.hyperbyte.cloud/database-version` +annotation) and renders only what that version supports. See +[docs/compatibility.md](docs/compatibility.md) for the matrix. + ## Development ```sh diff --git a/api/v1alpha1/hyperbytedbcluster_types.go b/api/v1alpha1/hyperbytedbcluster_types.go index ee235c4..f6b1d21 100644 --- a/api/v1alpha1/hyperbytedbcluster_types.go +++ b/api/v1alpha1/hyperbytedbcluster_types.go @@ -230,7 +230,8 @@ type ChDBSpec struct { SessionDataPath string `json:"sessionDataPath,omitempty"` // Sets both query and write pool sizes when QueryPoolSize and WritePoolSize - // are unset. Defaults to 1 when all pool fields are unset. + // are unset. When all pool fields are unset the query pool is 12 and the + // write pool 1. // +optional // +kubebuilder:validation:Minimum=1 PoolSize int32 `json:"poolSize,omitempty"` @@ -265,6 +266,7 @@ type LoggingSpec struct { Format string `json:"format,omitempty"` } +// +kubebuilder:validation:XValidation:rule="!has(self.raftHeartbeatIntervalMs) || !has(self.raftElectionTimeoutMs) || self.raftHeartbeatIntervalMs < self.raftElectionTimeoutMs",message="raftHeartbeatIntervalMs must be below raftElectionTimeoutMs" type ClusterTuningSpec struct { // +kubebuilder:default=2 HeartbeatIntervalSecs int32 `json:"heartbeatIntervalSecs,omitempty"` @@ -275,7 +277,10 @@ type ClusterTuningSpec struct { // +kubebuilder:default=5 ReplicationMaxRetries int32 `json:"replicationMaxRetries,omitempty"` - // +kubebuilder:default=1000 + // Must be below raftElectionTimeoutMs: Raft refuses to start otherwise. + // The default was 1000, equal to the election timeout's, and a cluster + // that kept both defaults crash-looped. + // +kubebuilder:default=300 RaftHeartbeatIntervalMs int32 `json:"raftHeartbeatIntervalMs,omitempty"` // +kubebuilder:default=1000 @@ -388,6 +393,23 @@ type ShardingSpec struct { // +kubebuilder:validation:Minimum=1 ScatterMaxPeerAttempts int32 `json:"scatterMaxPeerAttempts,omitempty"` + // Propose AddPeer shard-map ops. Defaults to true when unset. + // + // Set false for the duration of a rolling upgrade of a sharded cluster: a + // node running a build that predates the op cannot decode it, rejects the + // Raft append RPC carrying it, and stops replicating until every node is + // upgraded. Re-enable once the rollout completes. + // +optional + AddPeerProposalsEnabled *bool `json:"addPeerProposalsEnabled,omitempty"` + + // Propose RemovePeer shard-map ops. Defaults to true when unset. + // + // Carries the identical rolling-upgrade hazard as AddPeerProposalsEnabled, + // and scale-down evacuation depends on it: leaving it false prevents a + // decommissioning node from ever releasing its regions. + // +optional + RemovePeerProposalsEnabled *bool `json:"removePeerProposalsEnabled,omitempty"` + // Load-based split QPS threshold; 0 = disabled. Emitted only when set so the // server default (0) applies when omitted. // +optional diff --git a/api/v1alpha1/hyperbytedbcluster_webhook.go b/api/v1alpha1/hyperbytedbcluster_webhook.go index 33db06f..a37b765 100644 --- a/api/v1alpha1/hyperbytedbcluster_webhook.go +++ b/api/v1alpha1/hyperbytedbcluster_webhook.go @@ -68,10 +68,6 @@ func (w *HyperbytedbClusterWebhook) Default(_ context.Context, obj *HyperbytedbC } } - if obj.Spec.ChDB.PoolSize == 0 { - obj.Spec.ChDB.PoolSize = 1 - } - if obj.Spec.Retention.Interval == "" { obj.Spec.Retention.Interval = "60s" } diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 55275c9..0e29633 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -876,6 +876,16 @@ func (in *ServerSpec) DeepCopy() *ServerSpec { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ShardingSpec) DeepCopyInto(out *ShardingSpec) { *out = *in + if in.AddPeerProposalsEnabled != nil { + in, out := &in.AddPeerProposalsEnabled, &out.AddPeerProposalsEnabled + *out = new(bool) + **out = **in + } + if in.RemovePeerProposalsEnabled != nil { + in, out := &in.RemovePeerProposalsEnabled, &out.RemovePeerProposalsEnabled + *out = new(bool) + **out = **in + } if in.LoadSplitQpsThreshold != nil { in, out := &in.LoadSplitQpsThreshold, &out.LoadSplitQpsThreshold *out = new(int64) diff --git a/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml b/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml index b76eab2..cebcb54 100644 --- a/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml +++ b/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml @@ -2979,7 +2979,8 @@ spec: poolSize: description: |- Sets both query and write pool sizes when QueryPoolSize and WritePoolSize - are unset. Defaults to 1 when all pool fields are unset. + are unset. When all pool fields are unset the query pool is 12 and the + write pool 1. format: int32 minimum: 1 type: integer @@ -3023,7 +3024,11 @@ spec: format: int32 type: integer raftHeartbeatIntervalMs: - default: 1000 + default: 300 + description: |- + Must be below raftElectionTimeoutMs: Raft refuses to start otherwise. + The default was 1000, equal to the election timeout's, and a cluster + that kept both defaults crash-looped. format: int32 type: integer raftSnapshotThreshold: @@ -3131,6 +3136,10 @@ spec: - enabled type: object type: object + x-kubernetes-validations: + - message: raftHeartbeatIntervalMs must be below raftElectionTimeoutMs + rule: '!has(self.raftHeartbeatIntervalMs) || !has(self.raftElectionTimeoutMs) + || self.raftHeartbeatIntervalMs < self.raftElectionTimeoutMs' failover: description: FailoverSpec controls automatic failure detection and recovery. @@ -3595,6 +3604,15 @@ spec: block into config.toml. A 1-replica CR with sharding.enabled is a 1-member cluster ([cluster] enabled=true). Defaults stay off. properties: + addPeerProposalsEnabled: + description: |- + Propose AddPeer shard-map ops. Defaults to true when unset. + + Set false for the duration of a rolling upgrade of a sharded cluster: a + node running a build that predates the op cannot decode it, rejects the + Raft append RPC carrying it, and stops replicating until every node is + upgraded. Re-enable once the rollout completes. + type: boolean bootstrapTimeoutMs: description: Sync bootstrap RPC timeout. format: int64 @@ -3644,6 +3662,14 @@ spec: format: int64 minimum: 1 type: integer + removePeerProposalsEnabled: + description: |- + Propose RemovePeer shard-map ops. Defaults to true when unset. + + Carries the identical rolling-upgrade hazard as AddPeerProposalsEnabled, + and scale-down evacuation depends on it: leaving it false prevents a + decommissioning node from ever releasing its regions. + type: boolean replicationFactor: description: Target replica count per shard region. format: int32 diff --git a/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_cluster.yaml b/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_cluster.yaml index fde677e..5390e55 100644 --- a/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_cluster.yaml +++ b/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_cluster.yaml @@ -20,7 +20,7 @@ spec: maxPointsPerBatch: 50000 chdb: sessionDataPath: /var/lib/hyperbytedb/chdb - poolSize: 1 + queryPoolSize: 12 retention: enabled: true interval: 60s diff --git a/dist/chart/Chart.yaml b/dist/chart/Chart.yaml index 9b9b610..7e332ee 100644 --- a/dist/chart/Chart.yaml +++ b/dist/chart/Chart.yaml @@ -7,8 +7,8 @@ type: application # release workflow (.github/workflows/release.yml) stamps them with the git # tag (or a `0.0.0-sha.` snapshot for main branch builds) when the # chart is packaged and pushed to ghcr.io. -version: 0.1.2 -appVersion: "0.1.2" +version: 0.3.0 +appVersion: "0.3.0" home: https://hyperbyte-cloud.github.io/hyperbytedb/ sources: diff --git a/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbyte.cloud.yaml b/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbyte.cloud.yaml index 64b6f6d..cebcb54 100644 --- a/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbyte.cloud.yaml +++ b/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbyte.cloud.yaml @@ -2979,7 +2979,8 @@ spec: poolSize: description: |- Sets both query and write pool sizes when QueryPoolSize and WritePoolSize - are unset. Defaults to 1 when all pool fields are unset. + are unset. When all pool fields are unset the query pool is 12 and the + write pool 1. format: int32 minimum: 1 type: integer @@ -3023,7 +3024,11 @@ spec: format: int32 type: integer raftHeartbeatIntervalMs: - default: 1000 + default: 300 + description: |- + Must be below raftElectionTimeoutMs: Raft refuses to start otherwise. + The default was 1000, equal to the election timeout's, and a cluster + that kept both defaults crash-looped. format: int32 type: integer raftSnapshotThreshold: @@ -3131,6 +3136,10 @@ spec: - enabled type: object type: object + x-kubernetes-validations: + - message: raftHeartbeatIntervalMs must be below raftElectionTimeoutMs + rule: '!has(self.raftHeartbeatIntervalMs) || !has(self.raftElectionTimeoutMs) + || self.raftHeartbeatIntervalMs < self.raftElectionTimeoutMs' failover: description: FailoverSpec controls automatic failure detection and recovery. @@ -3592,8 +3601,18 @@ spec: sharding: description: |- Experimental series sharding. When set, the operator writes a `[sharding]` - block into config.toml. Enabling it requires replicas > 1 (cluster mode). + block into config.toml. A 1-replica CR with sharding.enabled is a + 1-member cluster ([cluster] enabled=true). Defaults stay off. properties: + addPeerProposalsEnabled: + description: |- + Propose AddPeer shard-map ops. Defaults to true when unset. + + Set false for the duration of a rolling upgrade of a sharded cluster: a + node running a build that predates the op cannot decode it, rejects the + Raft append RPC carrying it, and stops replicating until every node is + upgraded. Re-enable once the rollout completes. + type: boolean bootstrapTimeoutMs: description: Sync bootstrap RPC timeout. format: int64 @@ -3643,6 +3662,14 @@ spec: format: int64 minimum: 1 type: integer + removePeerProposalsEnabled: + description: |- + Propose RemovePeer shard-map ops. Defaults to true when unset. + + Carries the identical rolling-upgrade hazard as AddPeerProposalsEnabled, + and scale-down evacuation depends on it: leaving it false prevents a + decommissioning node from ever releasing its regions. + type: boolean replicationFactor: description: Target replica count per shard region. format: int32 diff --git a/dist/install.yaml b/dist/install.yaml index 6465c3c..88bba93 100644 --- a/dist/install.yaml +++ b/dist/install.yaml @@ -3182,7 +3182,8 @@ spec: poolSize: description: |- Sets both query and write pool sizes when QueryPoolSize and WritePoolSize - are unset. Defaults to 1 when all pool fields are unset. + are unset. When all pool fields are unset the query pool is 12 and the + write pool 1. format: int32 minimum: 1 type: integer @@ -3226,7 +3227,11 @@ spec: format: int32 type: integer raftHeartbeatIntervalMs: - default: 1000 + default: 300 + description: |- + Must be below raftElectionTimeoutMs: Raft refuses to start otherwise. + The default was 1000, equal to the election timeout's, and a cluster + that kept both defaults crash-looped. format: int32 type: integer raftSnapshotThreshold: @@ -3334,6 +3339,10 @@ spec: - enabled type: object type: object + x-kubernetes-validations: + - message: raftHeartbeatIntervalMs must be below raftElectionTimeoutMs + rule: '!has(self.raftHeartbeatIntervalMs) || !has(self.raftElectionTimeoutMs) + || self.raftHeartbeatIntervalMs < self.raftElectionTimeoutMs' failover: description: FailoverSpec controls automatic failure detection and recovery. @@ -3795,8 +3804,18 @@ spec: sharding: description: |- Experimental series sharding. When set, the operator writes a `[sharding]` - block into config.toml. Enabling it requires replicas > 1 (cluster mode). + block into config.toml. A 1-replica CR with sharding.enabled is a + 1-member cluster ([cluster] enabled=true). Defaults stay off. properties: + addPeerProposalsEnabled: + description: |- + Propose AddPeer shard-map ops. Defaults to true when unset. + + Set false for the duration of a rolling upgrade of a sharded cluster: a + node running a build that predates the op cannot decode it, rejects the + Raft append RPC carrying it, and stops replicating until every node is + upgraded. Re-enable once the rollout completes. + type: boolean bootstrapTimeoutMs: description: Sync bootstrap RPC timeout. format: int64 @@ -3846,6 +3865,14 @@ spec: format: int64 minimum: 1 type: integer + removePeerProposalsEnabled: + description: |- + Propose RemovePeer shard-map ops. Defaults to true when unset. + + Carries the identical rolling-upgrade hazard as AddPeerProposalsEnabled, + and scale-down evacuation depends on it: leaving it false prevents a + decommissioning node from ever releasing its regions. + type: boolean replicationFactor: description: Target replica count per shard region. format: int32 diff --git a/docs/compatibility.md b/docs/compatibility.md new file mode 100644 index 0000000..b197559 --- /dev/null +++ b/docs/compatibility.md @@ -0,0 +1,70 @@ +# HyperbyteDB compatibility + +One operator manages clusters on several HyperbyteDB versions. For each +cluster it works out which version runs and renders the config and pods for +that version: a HyperbyteDB 0.9 rejects config keys it does not know, and a +0.8 ignores them and runs on its defaults, so neither is safe to get keys +meant for the other. + +## Which version a cluster runs + +The first of these that is a version (`0.8.5-beta`, `v0.9.0-beta`, `0.9.0`): + +1. the annotation `hyperbytedb.hyperbyte.cloud/database-version`, +2. `spec.version`, +3. the tag of the database image. + +If none is (a local image tagged `e2e`, say), the cluster is rendered for the +newest version this operator knows. The `DatabaseVersion` condition in the +cluster's status says which version was used, where it came from, and what +was left out for it: + +```sh +kubectl get hyperbytedbcluster my-cluster \ + -o jsonpath='{.status.conditions[?(@.type=="DatabaseVersion")]}' +``` + +Reasons: `Supported`, `Assumed` (no version found), `NewerThanKnown` +(rendered for the newest this operator knows), `OlderThanSupported` +(condition status `False`; rendered as the oldest supported). + +## Matrix + +| Operator | Renders for HyperbyteDB | Newest known | +|---|---|---| +| v0.3.0 | 0.8.0 and later | 0.9.1-beta | +| v0.2.0 | 0.8.0 and later | 0.9.0-beta | +| v0.1.x | not version-aware: renders the newest keys for every cluster (0.8.x ignores the ones it does not know) | — | + +What v0.3.0 renders per version: + +| Capability | HyperbyteDB | For older versions the operator | +|---|---|---| +| Split chDB pools (`query_pool_size`, `write_pool_size`) | 0.9.0-beta | writes `pool_size` for both: `spec.chdb.poolSize`, else `queryPoolSize`, else 1 | +| Query pool sized by the database from its CPUs | 0.9.0-beta | writes `pool_size = 1` when no pool field is set, as v0.1.x did | +| Sharding (`[sharding]`, `spec.sharding`) | 0.9.0-beta | omits the section; `spec.sharding` has no effect | +| Control listener (`control-listener` annotation) | 0.9.0-beta | ignores the annotation; liveness stays on the server port | +| Evacuate together (`evacuate-together` annotation) | 0.9.0-beta | ignores the annotation; departing nodes leave one at a time | +| Cluster auth (`spec.auth.enabled`: auth Secret, cluster token, bootstrap admin) | 0.9.1-beta | renders `auth.enabled` only; no Secret, so a cluster with auth on cannot reach its own internal routes | +| Parallel pod management (`parallel-pods` annotation) | any | — (operator only) | + +The source of truth is `CompatibilityMatrix` in +`internal/hyperbytedb/compat.go`; keep this page in step with it. + +## Upgrading + +- v0.2.0 → v0.3.0: cluster auth needs HyperbyteDB 0.9.1-beta (0.9.0-beta + has no cluster token and refuses the key). A cluster with + `spec.auth.enabled` that resolves to 0.9.0-beta gets no auth Secret env, + so move it to 0.9.1-beta (version and image) before or with the operator. +- Upgrading the operator does not restart a cluster unless its rendered + config or pods change. v0.1.x → v0.2.0 renders the same config for a + 0.8.5-beta cluster and for a 0.9 development cluster that sets its pool + sizes; a 0.9 cluster with no pool field changes (it stops writing + `pool_size = 1` and lets the database size its query pool), which rolls + its pods once. +- Upgrading HyperbyteDB: change `spec.version` (or the image). The operator + renders for the new version in the same reconcile, so the new pods start + with a config they accept. +- Do not run `SET` statements while a HyperbyteDB rolling upgrade is in + progress: nodes on the older build cannot decode them. diff --git a/go.mod b/go.mod index 7ff0747..8cea5e5 100644 --- a/go.mod +++ b/go.mod @@ -3,16 +3,18 @@ module github.com/hyperbyte-cloud/hyperbytedb-operator go 1.25.3 require ( + github.com/Masterminds/semver/v3 v3.4.0 github.com/onsi/ginkgo/v2 v2.27.2 github.com/onsi/gomega v1.38.2 + k8s.io/api v0.35.0 k8s.io/apimachinery v0.35.0 k8s.io/client-go v0.35.0 + k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 sigs.k8s.io/controller-runtime v0.23.1 ) require ( cel.dev/expr v0.24.0 // indirect - github.com/Masterminds/semver/v3 v3.4.0 // indirect github.com/antlr4-go/antlr/v4 v4.13.0 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/blang/semver/v4 v4.0.0 // indirect @@ -85,13 +87,11 @@ require ( gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect - k8s.io/api v0.35.0 // indirect k8s.io/apiextensions-apiserver v0.35.0 // indirect k8s.io/apiserver v0.35.0 // indirect k8s.io/component-base v0.35.0 // indirect k8s.io/klog/v2 v2.130.1 // indirect k8s.io/kube-openapi v0.0.0-20250910181357-589584f1c912 // indirect - k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 // indirect sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2 // indirect sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect sigs.k8s.io/randfill v1.0.0 // indirect diff --git a/internal/controller/departed_state_test.go b/internal/controller/departed_state_test.go new file mode 100644 index 0000000..8392706 --- /dev/null +++ b/internal/controller/departed_state_test.go @@ -0,0 +1,19 @@ +package controller + +import "testing" + +// A pod the leader still lists as departed must be sent add-node, or a node +// scaled down and back up keeps its persisted `leaving` state and never passes +// readiness. Anything else is a current member and must be left alone. +func TestIsDepartedState(t *testing.T) { + for _, s := range []string{"leaving", "decommissioning"} { + if !isDepartedState(s) { + t.Errorf("%q must count as departed so the node is re-added", s) + } + } + for _, s := range []string{"active", "syncing", "joining", "draining", "disconnected", ""} { + if isDepartedState(s) { + t.Errorf("%q is a current member and must not be re-added", s) + } + } +} diff --git a/internal/controller/hyperbytedbcluster_controller.go b/internal/controller/hyperbytedbcluster_controller.go index 5a24a22..937f3c7 100644 --- a/internal/controller/hyperbytedbcluster_controller.go +++ b/internal/controller/hyperbytedbcluster_controller.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "maps" + "slices" "time" appsv1 "k8s.io/api/apps/v1" @@ -146,6 +147,16 @@ func (r *HyperbytedbClusterReconciler) Reconcile(ctx context.Context, req ctrl.R return r.setFailedStatus(ctx, cluster, "TLSFailed", err) } + // 1b. Auth Secret. Every call the operator makes to the cluster from here + // on sends its cluster token. + if hyperbytedb.ClusterAuthEnabled(cluster) { + token, err := r.reconcileAuthSecret(ctx, cluster) + if err != nil { + return r.setFailedStatus(ctx, cluster, "AuthSecretFailed", err) + } + ctx = hyperbytedb.WithClusterToken(ctx, token) + } + // 2. ConfigMap if err := r.reconcileConfigMap(ctx, cluster); err != nil { return r.setFailedStatus(ctx, cluster, "ConfigMapFailed", err) @@ -170,7 +181,12 @@ func (r *HyperbytedbClusterReconciler) Reconcile(ctx context.Context, req ctrl.R // 5b. Scale-down: drain departing pods and notify survivors before the StatefulSet shrinks. // Errors per-pod are already logged inside the hook and intentionally non-fatal. - r.runScaleDownClusterHooks(ctx, cluster, prevSTSReplicas, replicas) + if !r.runScaleDownClusterHooks(ctx, cluster, prevSTSReplicas, replicas) { + // Regions are still moving off a departing node. Hold .spec.replicas + // where it is and check again shortly; shrinking now would delete a + // pod, and later its volume, mid-transfer. + return ctrl.Result{RequeueAfter: 10 * time.Second}, nil + } // 6. StatefulSet stsResult, err := r.reconcileStatefulSet(ctx, cluster, configHash) @@ -180,6 +196,14 @@ func (r *HyperbytedbClusterReconciler) Reconcile(ctx context.Context, req ctrl.R } return r.setFailedStatus(ctx, cluster, "StatefulSetFailed", err) } + if stsResult.Recreating { + return ctrl.Result{RequeueAfter: 2 * time.Second}, nil + } + + // 6a. Finish a scale-down the StatefulSet controller cannot. + if err := r.deleteStuckDepartedPods(ctx, cluster, stsResult); err != nil { + return r.setFailedStatus(ctx, cluster, "ScaleDownFailed", err) + } // 6b. Operator-driven rolling restart coordination. When the StatefulSet // has a pending rolling update and the proxy is enabled, the operator @@ -242,6 +266,20 @@ func (r *HyperbytedbClusterReconciler) Reconcile(ctx context.Context, req ctrl.R memberStatuses := r.Members.CollectMemberStatuses(ctx, cluster, cluster.Namespace, replicas) cluster.Status.Members = memberStatuses cluster.Status.ClusterState = hyperbytedb.DeriveClusterState(memberStatuses) + // Which HyperbyteDB version the config and pods are rendered for, and + // what is left out for it (see hyperbytedb.CompatibilityMatrix). + versionReason, versionMessage := hyperbytedb.CompatibilitySummary(cluster) + versionStatus := metav1.ConditionTrue + if versionReason == "OlderThanSupported" { + versionStatus = metav1.ConditionFalse + } + meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{ + Type: "DatabaseVersion", + Status: versionStatus, + Reason: versionReason, + Message: versionMessage, + ObservedGeneration: cluster.Generation, + }) // 13. Cluster membership reconciliation (multi-node only). // Asks the Raft leader (via API) to add any pod that is reachable but @@ -305,6 +343,48 @@ func (r *HyperbytedbClusterReconciler) reconcileTLS(ctx context.Context, cluster return r.Create(ctx, secret) } +// reconcileAuthSecret makes sure the auth Secret exists with every key, +// generating what is missing, and returns the cluster token. A Secret the +// operator names (-auth) is owned by the cluster and goes with it; +// one named in spec.auth.credentialsSecretName is the user's and is kept. +func (r *HyperbytedbClusterReconciler) reconcileAuthSecret(ctx context.Context, cluster *hyperbytedbv1alpha1.HyperbytedbCluster) (string, error) { + name := hyperbytedb.AuthSecretName(cluster) + secret := &corev1.Secret{} + err := r.Get(ctx, types.NamespacedName{Name: name, Namespace: cluster.Namespace}, secret) + switch { + case apierrors.IsNotFound(err): + secret, err = hyperbytedb.BuildAuthSecret(cluster) + if err != nil { + return "", fmt.Errorf("generating auth secret: %w", err) + } + if cluster.Spec.Auth.CredentialsSecretName == "" { + if err := controllerutil.SetControllerReference(cluster, secret, r.Scheme); err != nil { + return "", err + } + } + if err := r.Create(ctx, secret); err != nil { + return "", err + } + r.Recorder.Event(cluster, corev1.EventTypeNormal, "AuthSecretCreated", + fmt.Sprintf("Generated cluster token and admin login in Secret %s", name)) + case err != nil: + return "", err + default: + changed, err := hyperbytedb.FillAuthSecret(secret) + if err != nil { + return "", err + } + if changed { + if err := r.Update(ctx, secret); err != nil { + return "", err + } + r.Recorder.Event(cluster, corev1.EventTypeNormal, "AuthSecretUpdated", + fmt.Sprintf("Added missing keys to Secret %s", name)) + } + } + return string(secret.Data[hyperbytedb.AuthSecretClusterTokenKey]), nil +} + // ---------- ConfigMap ---------- func (r *HyperbytedbClusterReconciler) reconcileConfigMap(ctx context.Context, cluster *hyperbytedbv1alpha1.HyperbytedbCluster) error { @@ -374,6 +454,12 @@ func (r *HyperbytedbClusterReconciler) reconcileService(ctx context.Context, clu type stsReconcileResult struct { SpecReplicas int32 ReadyReplicas int32 + // StatusReplicas is how many pods the StatefulSet currently has. It + // stays above SpecReplicas while a scale-down is still deleting pods. + StatusReplicas int32 + // Recreating is set while the StatefulSet is being deleted to change a + // field that cannot be updated in place; the reconcile waits for it. + Recreating bool } func (r *HyperbytedbClusterReconciler) reconcileStatefulSet(ctx context.Context, cluster *hyperbytedbv1alpha1.HyperbytedbCluster, configHash string) (stsReconcileResult, error) { @@ -395,6 +481,19 @@ func (r *HyperbytedbClusterReconciler) reconcileStatefulSet(ctx context.Context, return stsReconcileResult{}, err } + // podManagementPolicy cannot be updated in place. Delete the StatefulSet + // leaving its pods running (orphaned); the next reconcile recreates it, + // and the new StatefulSet adopts the pods by their labels. + if existing.Spec.PodManagementPolicy != desired.Spec.PodManagementPolicy { + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "RecreatingStatefulSet", + "Recreating StatefulSet to change pod management from %s to %s; pods keep running", + existing.Spec.PodManagementPolicy, desired.Spec.PodManagementPolicy) + if err := r.Delete(ctx, existing, client.PropagationPolicy(metav1.DeletePropagationOrphan)); err != nil && !apierrors.IsNotFound(err) { + return stsReconcileResult{}, err + } + return stsReconcileResult{Recreating: true}, nil + } + // Update mutable fields existing.Spec.Replicas = desired.Spec.Replicas existing.Spec.Template.Spec.Containers = desired.Spec.Template.Spec.Containers @@ -426,8 +525,9 @@ func (r *HyperbytedbClusterReconciler) reconcileStatefulSet(ctx context.Context, } return stsReconcileResult{ - SpecReplicas: *existing.Spec.Replicas, - ReadyReplicas: existing.Status.ReadyReplicas, + SpecReplicas: *existing.Spec.Replicas, + ReadyReplicas: existing.Status.ReadyReplicas, + StatusReplicas: existing.Status.Replicas, }, nil } @@ -783,9 +883,18 @@ func (r *HyperbytedbClusterReconciler) includeFromAllProxies( // surviving members to drop them from membership. Must run before .spec.replicas is reduced. // Per-pod failures are logged and ignored: the StatefulSet scale-down still proceeds because // the membership/drain endpoints are best-effort hints to surviving nodes. -func (r *HyperbytedbClusterReconciler) runScaleDownClusterHooks(ctx context.Context, cluster *hyperbytedbv1alpha1.HyperbytedbCluster, prevReplicas, desiredReplicas int32) { +// runScaleDownClusterHooks decommissions the pods that are leaving and reports +// whether they have finished evacuating. +// +// Returns false while any departing node still owns regions. The caller must +// not reduce .spec.replicas until it returns true: deleting the pod (and, on +// the next pass, its PVC) while regions are still moving off it destroys the +// only remaining copy of whatever had not yet transferred. Evacuation time is +// a function of how much data the node holds, so no fixed grace period is +// safe -- the node itself reports when it is done, by reaching "leaving". +func (r *HyperbytedbClusterReconciler) runScaleDownClusterHooks(ctx context.Context, cluster *hyperbytedbv1alpha1.HyperbytedbCluster, prevReplicas, desiredReplicas int32) bool { if desiredReplicas >= prevReplicas || prevReplicas < 1 { - return + return true } log := logf.FromContext(ctx) @@ -793,12 +902,68 @@ func (r *HyperbytedbClusterReconciler) runScaleDownClusterHooks(ctx context.Cont stsName := hyperbytedb.StatefulSetName(cluster) headlessSvc := hyperbytedb.HeadlessServiceName(cluster) + // Highest ordinal first, matching the order the StatefulSet will remove + // them. One at a time unless the cluster opts into evacuating together + // (see departurePlan). + together := hyperbytedb.EvacuateTogether(cluster) + var departing []departingNode + var departHosts []string for i := prevReplicas - 1; i >= desiredReplicas; i-- { host := fmt.Sprintf("%s-%d.%s.%s.svc.cluster.local", stsName, i, headlessSvc, cluster.Namespace) - if err := r.Members.Client.DrainNode(ctx, host, port); err != nil { - log.V(1).Info("Could not drain departing pod", "ordinal", i, "host", host, "error", err) + health, err := r.Members.Client.GetNodeHealth(ctx, host, port) + state := "" + if health != nil { + state = health.State } + if err != nil { + log.V(1).Info("Departing pod unreachable; not gating on it", + "ordinal", i, "host", host, "error", err) + } + departing = append(departing, departingNode{state: state, reachable: err == nil}) + departHosts = append(departHosts, host) + if !together && err == nil && state != nodeStateLeaving { + // Sequential: nodes after the first one still holding data are + // not looked at until it has released its regions. + break + } + } + tell, evacuated := departurePlan(departing, together) + for _, i := range tell { + if err := r.Members.Client.DecommissionNode(ctx, departHosts[i], port); err != nil { + log.V(1).Info("Could not decommission departing pod", + "host", departHosts[i], "error", err) + } + } + if !evacuated { + log.Info("Waiting for departing nodes to finish evacuating", + "departing", len(departHosts), "together", together) + } + + if !evacuated { + return false + } + + // Drop the departing nodes from the Raft voter set before the StatefulSet + // deletes them. Without this they stay voters forever: shrinking 6 -> 3 + // left 3 of 6 voters alive, which is no quorum, so the smaller cluster + // could not commit anything. Worse, scaling back up brought the same node + // ids back with empty disks while the leader still held their old + // progress; openraft does not support that ("log reversion") and the + // returning nodes crashed on their first apply. + hosts := make(map[uint64]string, prevReplicas) + for i := range prevReplicas { + hosts[uint64(i+1)] = fmt.Sprintf("%s-%d.%s.%s.svc.cluster.local", + stsName, i, headlessSvc, cluster.Namespace) + } + departed := make([]uint64, 0, prevReplicas-desiredReplicas) + for i := prevReplicas - 1; i >= desiredReplicas; i-- { + departed = append(departed, uint64(i+1)) + } + if err := removeDepartedVoters(ctx, r.Members.Client, hosts, port, departed); err != nil { + log.Info("Holding scale-down: could not remove departing nodes from raft voters", + "error", err) + return false } for i := desiredReplicas; i < prevReplicas; i++ { @@ -812,6 +977,159 @@ func (r *HyperbytedbClusterReconciler) runScaleDownClusterHooks(ctx context.Cont } } } + return true +} + +// deleteStuckDepartedPods deletes the unready pods a scale-down has left +// behind. +// +// .spec.replicas only shrinks after runScaleDownClusterHooks has confirmed +// the departing nodes evacuated and removed them from the Raft voters, so +// every pod at or above .spec.replicas is already out of the cluster. Such a +// pod reports unready: it keeps campaigning for a cluster it is no longer in. +// With OrderedReady the StatefulSet controller will not delete a pod while a +// lower-ordinal pod is unready, so 12 -> 3 stalled with pods 3..11 alive and +// pods 4..11 unready. The StatefulSet would delete these pods anyway; this +// only removes the ordering block. Ready pods are left to the StatefulSet. +func (r *HyperbytedbClusterReconciler) deleteStuckDepartedPods(ctx context.Context, cluster *hyperbytedbv1alpha1.HyperbytedbCluster, sts stsReconcileResult) error { + if sts.StatusReplicas <= sts.SpecReplicas { + return nil + } + stsName := hyperbytedb.StatefulSetName(cluster) + pods := make([]*corev1.Pod, 0, sts.StatusReplicas-sts.SpecReplicas) + for i := sts.SpecReplicas; i < sts.StatusReplicas; i++ { + pod := &corev1.Pod{} + name := fmt.Sprintf("%s-%d", stsName, i) + if err := r.Get(ctx, types.NamespacedName{Name: name, Namespace: cluster.Namespace}, pod); err != nil { + if apierrors.IsNotFound(err) { + continue + } + return err + } + pods = append(pods, pod) + } + for _, pod := range stuckDepartedPods(pods) { + if err := r.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("deleting departed pod %s: %w", pod.Name, err) + } + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "ScaleDown", + "Deleted departed pod %s that blocked the StatefulSet scale-down", pod.Name) + } + return nil +} + +// stuckDepartedPods picks, from pods above .spec.replicas, the ones the +// StatefulSet controller would wait on forever: unready and not already +// being deleted. +func stuckDepartedPods(pods []*corev1.Pod) []*corev1.Pod { + var stuck []*corev1.Pod + for _, pod := range pods { + if pod.DeletionTimestamp != nil || isPodReady(pod) { + continue + } + stuck = append(stuck, pod) + } + return stuck +} + +// removeDepartedVoters removes each id in `departed` from the Raft voter set, +// in order. The leader is looked up again before every removal because +// removing the leader itself makes it step down. +// +// A node that is already not a voter counts as removed, so a retry after a +// partial failure is safe. +func removeDepartedVoters(ctx context.Context, c *hyperbytedb.Client, hosts map[uint64]string, port int32, departed []uint64) error { + for _, id := range departed { + leaderHost, err := raftLeaderHost(ctx, c, hosts, port) + if err != nil { + return err + } + if _, err := c.RemoveClusterNode(ctx, leaderHost, port, id); err != nil { + return fmt.Errorf("remove node %d from raft voters via %s: %w", id, leaderHost, err) + } + } + return nil +} + +// raftLeaderHost asks each host who the leader is and returns the leader's +// host. `hosts` maps node id to host. +func raftLeaderHost(ctx context.Context, c *hyperbytedb.Client, hosts map[uint64]string, port int32) (string, error) { + ids := make([]uint64, 0, len(hosts)) + for id := range hosts { + ids = append(ids, id) + } + slices.Sort(ids) + for _, id := range ids { + info, err := c.GetClusterLeader(ctx, hosts[id], port) + if err != nil || info.LeaderID == nil { + continue + } + if h, ok := hosts[*info.LeaderID]; ok { + return h, nil + } + } + return "", fmt.Errorf("no raft leader visible") +} + +const ( + // Node states reported by /health's "state" field. + nodeStateDecommissioning = "decommissioning" + nodeStateLeaving = "leaving" +) + +// departingNodeAction decides what to do with one node that is being removed. +// +// `gate` false means .spec.replicas must not shrink yet. An unreachable node +// never gates: it may already be gone, and the cluster's own dead-peer path +// restores replication factor, so blocking forever on a pod that will never +// answer would wedge every future scale-down. +type departingNode struct { + state string + reachable bool +} + +// departurePlan says which departing nodes (highest ordinal first) to tell to +// decommission now, and whether every one has evacuated. +// +// One at a time by default: a node is told only once every node before it +// has released its regions. Together, all are told at once. One at a time, +// every node evacuated still counts as a placement target, so its regions +// land on the next nodes to leave and move again: on n3 an 8 -> 3 scale-down +// moved node 8's copies onto nodes 4-7, and node 6 then held 61 copies to +// move. Together, the scheduler places every copy on a staying node once. +// spec.replicas shrinks only after all have evacuated either way. +func departurePlan(nodes []departingNode, together bool) (tell []int, evacuated bool) { + evacuated = true + for i, n := range nodes { + decommission, gate := departingNodeAction(n.state, n.reachable) + if decommission { + tell = append(tell, i) + } + if gate { + evacuated = false + if !together { + break + } + } + } + return tell, evacuated +} + +func departingNodeAction(state string, reachable bool) (decommission bool, gate bool) { + if !reachable { + return false, false + } + switch state { + case nodeStateLeaving: + // Fully evacuated; safe to delete the pod and its volume. + return false, false + case nodeStateDecommissioning: + // Already told; still moving regions. + return false, true + default: + // Still holding data and not yet told to leave. + return true, true + } } // ---------- Auto-failover ---------- @@ -1157,10 +1475,23 @@ func (r *HyperbytedbClusterReconciler) reconcileClusterMembership( } // The leader's view is authoritative for who is currently a member. + // + // A node the leader still lists as departed is NOT a current member. A + // StatefulSet keeps its PVCs on scale-down, so scaling back up restarts the + // same node id with its old `leaving` state, which is terminal on the + // database side. Treating it as known skipped add-node, so the node failed + // readiness forever and a cluster that had scaled down could not scale back + // up. Sending add-node for it is what lets the leader re-admit it. + // + // Safe against undoing a scale-down: `views` holds only ordinals below + // spec.replicas, so a pod still being evacuated is never in it. knownIDs := make(map[uint64]bool) for _, v := range views { if v.host == leaderHost { for _, n := range v.nodes { + if isDepartedState(n.State) { + continue + } knownIDs[uint64(n.NodeID)] = true } } @@ -1437,3 +1768,14 @@ func (r *HyperbytedbClusterReconciler) SetupWithManager(mgr ctrl.Manager) error Named("hyperbytedbcluster"). Complete(r) } + +// isDepartedState reports whether a membership state means the node was +// removed (decommissioned or retired) rather than merely restarting. +func isDepartedState(state string) bool { + switch state { + case "leaving", "decommissioning": + return true + default: + return false + } +} diff --git a/internal/controller/hyperbytedbcluster_controller_test.go b/internal/controller/hyperbytedbcluster_controller_test.go index f01c891..6e30023 100644 --- a/internal/controller/hyperbytedbcluster_controller_test.go +++ b/internal/controller/hyperbytedbcluster_controller_test.go @@ -200,6 +200,147 @@ var _ = Describe("HyperbytedbCluster Controller", func() { }) }) + Context("When a cluster opts into parallel pod management", func() { + const resourceName = "test-parallel" + + ctx := context.Background() + + typeNamespacedName := types.NamespacedName{ + Name: resourceName, + Namespace: "default", + } + + BeforeEach(func() { + resource := &hyperbytedbv1alpha1.HyperbytedbCluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: resourceName, + Namespace: "default", + }, + Spec: hyperbytedbv1alpha1.HyperbytedbClusterSpec{ + Replicas: ptr.To(int32(3)), + Image: "hyperbytedb:latest", + }, + } + Expect(k8sClient.Create(ctx, resource)).To(Succeed()) + }) + + AfterEach(func() { + resource := &hyperbytedbv1alpha1.HyperbytedbCluster{} + if err := k8sClient.Get(ctx, typeNamespacedName, resource); err == nil { + Expect(k8sClient.Delete(ctx, resource)).To(Succeed()) + } + }) + + // The policy cannot change in place, so the StatefulSet is deleted + // with its pods orphaned (left running) and recreated. envtest runs + // no garbage collector, so the delete stays pending on the orphan + // finalizer: that pending state is what this checks. + It("recreates the StatefulSet leaving its pods running", func() { + controllerReconciler := &HyperbytedbClusterReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Recorder: record.NewFakeRecorder(32), + Members: hyperbytedb.NewMemberManager(), + } + _, err := controllerReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName}) + Expect(err).NotTo(HaveOccurred()) + + sts := &appsv1.StatefulSet{} + Expect(k8sClient.Get(ctx, typeNamespacedName, sts)).To(Succeed()) + Expect(sts.Spec.PodManagementPolicy).To(Equal(appsv1.OrderedReadyPodManagement)) + + By("opting in") + cluster := &hyperbytedbv1alpha1.HyperbytedbCluster{} + Expect(k8sClient.Get(ctx, typeNamespacedName, cluster)).To(Succeed()) + cluster.Annotations = map[string]string{hyperbytedb.ParallelPodsAnnotation: "true"} + Expect(k8sClient.Update(ctx, cluster)).To(Succeed()) + + result, err := controllerReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName}) + Expect(err).NotTo(HaveOccurred()) + Expect(result.RequeueAfter).To(BeNumerically(">", 0)) + + Expect(k8sClient.Get(ctx, typeNamespacedName, sts)).To(Succeed()) + Expect(sts.DeletionTimestamp).NotTo(BeNil()) + Expect(sts.Finalizers).To(ContainElement(metav1.FinalizerOrphanDependents)) + }) + }) + + Context("When a scale-down leaves pods above .spec.replicas", func() { + const resourceName = "test-scaledown" + + ctx := context.Background() + + typeNamespacedName := types.NamespacedName{ + Name: resourceName, + Namespace: "default", + } + + BeforeEach(func() { + resource := &hyperbytedbv1alpha1.HyperbytedbCluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: resourceName, + Namespace: "default", + }, + Spec: hyperbytedbv1alpha1.HyperbytedbClusterSpec{ + Replicas: ptr.To(int32(3)), + Image: "hyperbytedb:latest", + }, + } + Expect(k8sClient.Create(ctx, resource)).To(Succeed()) + }) + + It("deletes the unready ones", func() { + controllerReconciler := &HyperbytedbClusterReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Recorder: record.NewFakeRecorder(32), + Members: hyperbytedb.NewMemberManager(), + } + _, err := controllerReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName}) + Expect(err).NotTo(HaveOccurred()) + + By("faking a StatefulSet that still has 5 pods after shrinking to 3") + sts := &appsv1.StatefulSet{} + Expect(k8sClient.Get(ctx, typeNamespacedName, sts)).To(Succeed()) + sts.Status.Replicas = 5 + Expect(k8sClient.Status().Update(ctx, sts)).To(Succeed()) + + makePod := func(name string, ready bool) { + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}, + Spec: corev1.PodSpec{Containers: []corev1.Container{ + {Name: "hyperbytedb", Image: "hyperbytedb:latest"}, + }}, + } + Expect(k8sClient.Create(ctx, pod)).To(Succeed()) + status := corev1.ConditionFalse + if ready { + status = corev1.ConditionTrue + } + pod.Status.Conditions = []corev1.PodCondition{{Type: corev1.PodReady, Status: status}} + Expect(k8sClient.Status().Update(ctx, pod)).To(Succeed()) + } + makePod(resourceName+"-3", true) + makePod(resourceName+"-4", false) + DeferCleanup(func() { + for _, name := range []string{resourceName + "-3", resourceName + "-4"} { + _ = k8sClient.Delete(ctx, &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}}) + } + }) + + _, err = controllerReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName}) + Expect(err).NotTo(HaveOccurred()) + + By("the unready departed pod is gone; the ready one is left to the StatefulSet") + gone := &corev1.Pod{} + err = k8sClient.Get(ctx, types.NamespacedName{Name: resourceName + "-4", Namespace: "default"}, gone) + Expect(errors.IsNotFound(err) || gone.DeletionTimestamp != nil).To(BeTrue()) + kept := &corev1.Pod{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: resourceName + "-3", Namespace: "default"}, kept)).To(Succeed()) + Expect(kept.DeletionTimestamp).To(BeNil()) + }) + }) + Context("When the cluster is paused", func() { const resourceName = "test-paused" diff --git a/internal/controller/raft_timing_crd_test.go b/internal/controller/raft_timing_crd_test.go new file mode 100644 index 0000000..5eccfa9 --- /dev/null +++ b/internal/controller/raft_timing_crd_test.go @@ -0,0 +1,41 @@ +package controller + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + + hyperbytedbv1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" +) + +// The CRD defaulted both Raft timings to 1000, which Raft refuses: a UAT +// cluster created without them crash-looped on every node. +var _ = Describe("Raft timing defaults", func() { + cr := func(name string) *hyperbytedbv1alpha1.HyperbytedbCluster { + return &hyperbytedbv1alpha1.HyperbytedbCluster{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}, + Spec: hyperbytedbv1alpha1.HyperbytedbClusterSpec{Replicas: ptr.To(int32(3))}, + } + } + + It("defaults the heartbeat below the election timeout", func() { + c := cr("raft-defaults") + Expect(k8sClient.Create(ctx, c)).To(Succeed()) + DeferCleanup(func() { _ = k8sClient.Delete(ctx, c) }) + got := &hyperbytedbv1alpha1.HyperbytedbCluster{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Name: c.Name, Namespace: "default"}, got)).To(Succeed()) + Expect(got.Spec.Cluster.RaftHeartbeatIntervalMs).To(BeNumerically("<", got.Spec.Cluster.RaftElectionTimeoutMs)) + }) + + It("refuses a heartbeat at or above the election timeout", func() { + c := cr("raft-equal") + c.Spec.Cluster.RaftHeartbeatIntervalMs = 1000 + c.Spec.Cluster.RaftElectionTimeoutMs = 1000 + err := k8sClient.Create(ctx, c) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("raftHeartbeatIntervalMs must be below raftElectionTimeoutMs")) + }) +}) diff --git a/internal/controller/remove_voters_test.go b/internal/controller/remove_voters_test.go new file mode 100644 index 0000000..72d4e45 --- /dev/null +++ b/internal/controller/remove_voters_test.go @@ -0,0 +1,124 @@ +package controller + +import ( + "context" + "encoding/json" + "fmt" + "net" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "sync" + "testing" + + "github.com/hyperbyte-cloud/hyperbytedb-operator/internal/hyperbytedb" +) + +// fakeRaft is a shared view of one Raft group served by several fake nodes. +type fakeRaft struct { + mu sync.Mutex + leader uint64 + voters map[uint64]bool + removed []uint64 +} + +// handler serves the leader and remove-node endpoints as node `id` sees them. +func (f *fakeRaft) handler(id uint64) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/cluster/leader": + _, _ = fmt.Fprintf(w, `{"leader_id":%d,"this_node_id":%d,"is_leader":%t,"term":1}`, + f.leader, id, f.leader == id) + case "/cluster/membership/remove-node": + if f.leader != id { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = fmt.Fprintf(w, `{"error":"not raft leader","leader_id":%d}`, f.leader) + return + } + var req struct { + NodeID uint64 `json:"node_id"` + } + _ = json.NewDecoder(r.Body).Decode(&req) + was := f.voters[req.NodeID] + delete(f.voters, req.NodeID) + if was { + f.removed = append(f.removed, req.NodeID) + } + // A leader that removes itself steps down; the lowest survivor + // takes over. + if req.NodeID == f.leader { + f.leader = 0 + for v := range f.voters { + if f.leader == 0 || v < f.leader { + f.leader = v + } + } + } + _, _ = fmt.Fprintf(w, `{"node_id":%d,"removed_from_voters":%t,"leader_id":%d,"remaining_voters":[]}`, + req.NodeID, was, id) + default: + w.WriteHeader(http.StatusNotFound) + } + } +} + +func TestRemoveDepartedVotersFollowsTheLeader(t *testing.T) { + // Node 6 leads and is itself departing, the case that forces a second + // leader lookup. 4..6 must all leave the voter set, highest first. + f := &fakeRaft{leader: 6, voters: map[uint64]bool{1: true, 2: true, 3: true, 4: true, 5: true, 6: true}} + + // The client uses one port for every node, so give each node its own + // loopback address on a shared port. + hosts := map[uint64]string{} + var port int32 + for id := uint64(1); id <= 6; id++ { + srv := httptest.NewUnstartedServer(f.handler(id)) + addr := fmt.Sprintf("127.0.0.%d:0", id) + if port != 0 { + addr = fmt.Sprintf("127.0.0.%d:%d", id, port) + } + l, err := net.Listen("tcp", addr) + if err != nil { + t.Skipf("cannot bind %s: %v", addr, err) + } + srv.Listener = l + srv.Start() + t.Cleanup(srv.Close) + u, _ := url.Parse(srv.URL) + p, _ := strconv.Atoi(u.Port()) + port = int32(p) + hosts[id] = u.Hostname() + } + + err := removeDepartedVoters(context.Background(), hyperbytedb.NewClient(), hosts, port, []uint64{6, 5, 4}) + if err != nil { + t.Fatalf("removeDepartedVoters: %v", err) + } + if got := fmt.Sprint(f.removed); got != "[6 5 4]" { + t.Errorf("removed = %s, want [6 5 4]", got) + } + for id := uint64(1); id <= 3; id++ { + if !f.voters[id] { + t.Errorf("survivor %d was removed", id) + } + } + + // Retrying after success is a no-op, not an error: the operator re-runs + // the hook every reconcile until the StatefulSet has shrunk. + if err := removeDepartedVoters(context.Background(), hyperbytedb.NewClient(), hosts, port, []uint64{6, 5, 4}); err != nil { + t.Fatalf("retry: %v", err) + } +} + +func TestRemoveDepartedVotersFailsWithoutALeader(t *testing.T) { + // No leader visible must hold the scale-down, not skip the removal. + hosts := map[uint64]string{1: "127.0.0.1"} + err := removeDepartedVoters(context.Background(), hyperbytedb.NewClient(), hosts, 1, []uint64{2}) + if err == nil { + t.Fatal("expected an error when no leader is reachable") + } +} diff --git a/internal/controller/scaledown_gate_test.go b/internal/controller/scaledown_gate_test.go new file mode 100644 index 0000000..68f0c8e --- /dev/null +++ b/internal/controller/scaledown_gate_test.go @@ -0,0 +1,123 @@ +package controller + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "testing" + + "github.com/hyperbyte-cloud/hyperbytedb-operator/internal/hyperbytedb" +) + +// healthServer serves /health reporting a fixed node state, and records +// whether /internal/decommission was called. +func healthServer(t *testing.T, state string, calls *int) (host string, port int32) { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/internal/decommission" { + *calls++ + w.WriteHeader(http.StatusAccepted) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(w, `{"status":"warn","message":"node is %s","state":%q}`, state, state) + })) + t.Cleanup(srv.Close) + + u, err := url.Parse(srv.URL) + if err != nil { + t.Fatalf("parse test server url: %v", err) + } + p, err := strconv.Atoi(u.Port()) + if err != nil { + t.Fatalf("parse test server port: %v", err) + } + return u.Hostname(), int32(p) +} + +func TestHealthStateIsParsedFromTheBody(t *testing.T) { + // The operator gates scale-down on `state`, not `status`: a node that is + // draining and one that has fully evacuated both report status "warn". + for _, want := range []string{"decommissioning", "leaving", "active"} { + host, port := healthServer(t, want, new(int)) + got, err := hyperbytedb.NewClient().GetNodeHealth(context.Background(), host, port) + if err != nil { + t.Fatalf("GetNodeHealth(%s): %v", want, err) + } + if got.State != want { + t.Errorf("state = %q, want %q", got.State, want) + } + } +} + +func TestDecommissionNodeHitsTheEndpoint(t *testing.T) { + calls := 0 + host, port := healthServer(t, "active", &calls) + if err := hyperbytedb.NewClient().DecommissionNode(context.Background(), host, port); err != nil { + t.Fatalf("DecommissionNode: %v", err) + } + if calls != 1 { + t.Errorf("decommission endpoint called %d times, want 1", calls) + } +} + +func TestDepartingNodeAction(t *testing.T) { + // The gate exists because evacuation time is a function of how much data a + // node holds, so no fixed grace period is safe. The node itself says when + // it is done, by reaching "leaving". + tests := []struct { + name string + state string + reachable bool + wantDecommission bool + wantGate bool + }{ + {"still serving: tell it to go, and wait", "active", true, true, true}, + {"already told: keep waiting", nodeStateDecommissioning, true, false, true}, + {"evacuated: safe to delete pod and volume", nodeStateLeaving, true, false, false}, + {"draining is a restart, not a departure", "draining", true, true, true}, + {"unreachable: never gate, or scale-down wedges forever", "", false, false, false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + decommission, gate := departingNodeAction(tc.state, tc.reachable) + if decommission != tc.wantDecommission { + t.Errorf("decommission = %v, want %v", decommission, tc.wantDecommission) + } + if gate != tc.wantGate { + t.Errorf("gate = %v, want %v", gate, tc.wantGate) + } + }) + } +} + +// One at a time, only the first node still holding data is told; together, +// every one is, so copies never land on a node that is about to leave. +func TestDeparturePlanTellsOneAtATimeOrAllTogether(t *testing.T) { + active := departingNode{state: "active", reachable: true} + draining := departingNode{state: nodeStateDecommissioning, reachable: true} + gone := departingNode{state: nodeStateLeaving, reachable: true} + cases := []struct { + name string + nodes []departingNode + together bool + tell []int + evacuated bool + }{ + {"one at a time tells the first", []departingNode{active, active, active}, false, []int{0}, false}, + {"one at a time skips evacuated nodes", []departingNode{gone, active, active}, false, []int{1}, false}, + {"one at a time waits on a draining node", []departingNode{draining, active}, false, nil, false}, + {"together tells every node still holding data", []departingNode{active, draining, active, gone}, true, []int{0, 2}, false}, + {"together is done when all have left", []departingNode{gone, gone}, true, nil, true}, + } + for _, tc := range cases { + tell, evacuated := departurePlan(tc.nodes, tc.together) + if fmt.Sprint(tell) != fmt.Sprint(tc.tell) || evacuated != tc.evacuated { + t.Errorf("%s: tell=%v evacuated=%v, want tell=%v evacuated=%v", + tc.name, tell, evacuated, tc.tell, tc.evacuated) + } + } +} diff --git a/internal/controller/stuck_departed_pods_test.go b/internal/controller/stuck_departed_pods_test.go new file mode 100644 index 0000000..7b7c978 --- /dev/null +++ b/internal/controller/stuck_departed_pods_test.go @@ -0,0 +1,46 @@ +package controller + +import ( + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func departedPod(name string, ready, deleting bool) *corev1.Pod { + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: name}} + status := corev1.ConditionFalse + if ready { + status = corev1.ConditionTrue + } + pod.Status.Conditions = []corev1.PodCondition{{Type: corev1.PodReady, Status: status}} + if deleting { + now := metav1.Now() + pod.DeletionTimestamp = &now + } + return pod +} + +// The 12 -> 3 stall: pod 3 still ready, pods 4.. unready, one already going. +func TestStuckDepartedPodsAreTheUnreadyOnes(t *testing.T) { + pods := []*corev1.Pod{ + departedPod("db-3", true, false), + departedPod("db-4", false, false), + departedPod("db-5", false, true), + departedPod("db-11", false, false), + } + got := stuckDepartedPods(pods) + if len(got) != 2 || got[0].Name != "db-4" || got[1].Name != "db-11" { + names := make([]string, 0, len(got)) + for _, p := range got { + names = append(names, p.Name) + } + t.Fatalf("stuck pods = %v, want [db-4 db-11]", names) + } +} + +func TestNoStuckPodsWhenAllDepartedPodsAreReady(t *testing.T) { + if got := stuckDepartedPods([]*corev1.Pod{departedPod("db-3", true, false)}); len(got) != 0 { + t.Fatalf("got %d stuck pods, want 0", len(got)) + } +} diff --git a/internal/hyperbytedb/auth.go b/internal/hyperbytedb/auth.go new file mode 100644 index 0000000..0fa6fa8 --- /dev/null +++ b/internal/hyperbytedb/auth.go @@ -0,0 +1,136 @@ +package hyperbytedb + +import ( + "context" + "crypto/rand" + "encoding/base64" + "net/http" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + v1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" +) + +// With spec.auth.enabled, a cluster's internal routes need an admin login. +// Nodes and the operator have none, so they send a shared cluster token +// instead. The operator keeps the token and a bootstrap admin login in a +// Secret, passes them to the pods as env, and sends the token on its own +// calls. + +// Keys of the auth Secret. +const ( + AuthSecretClusterTokenKey = "cluster-token" + AuthSecretAdminUsernameKey = "admin-username" + AuthSecretAdminPasswordKey = "admin-password" +) + +// ClusterTokenHeader carries the cluster token on internal routes. +const ClusterTokenHeader = "x-hyperbytedb-cluster-token" + +// AuthSecretName is spec.auth.credentialsSecretName, else -auth. +func AuthSecretName(cluster *v1alpha1.HyperbytedbCluster) string { + if cluster.Spec.Auth.CredentialsSecretName != "" { + return cluster.Spec.Auth.CredentialsSecretName + } + return cluster.Name + "-auth" +} + +// ClusterAuthEnabled is whether the operator wires the auth Secret into +// the cluster: auth is on and the database knows the cluster token. +func ClusterAuthEnabled(cluster *v1alpha1.HyperbytedbCluster) bool { + return cluster.Spec.Auth.Enabled && Supports(cluster, CapabilityClusterAuth) +} + +func randomSecret(bytes int) (string, error) { + b := make([]byte, bytes) + if _, err := rand.Read(b); err != nil { + return "", err + } + return base64.RawURLEncoding.EncodeToString(b), nil +} + +// FillAuthSecret adds any missing key to secret's data: a random token, the +// admin user "admin" and a random password. Existing values are kept, so a +// user-supplied Secret needs only the keys it wants to choose. Returns +// whether it changed anything. +func FillAuthSecret(secret *corev1.Secret) (bool, error) { + if secret.Data == nil { + secret.Data = map[string][]byte{} + } + changed := false + for _, k := range []struct { + key string + value func() (string, error) + }{ + {AuthSecretClusterTokenKey, func() (string, error) { return randomSecret(32) }}, + {AuthSecretAdminUsernameKey, func() (string, error) { return "admin", nil }}, + {AuthSecretAdminPasswordKey, func() (string, error) { return randomSecret(24) }}, + } { + if len(secret.Data[k.key]) > 0 { + continue + } + v, err := k.value() + if err != nil { + return false, err + } + secret.Data[k.key] = []byte(v) + changed = true + } + return changed, nil +} + +// BuildAuthSecret is a new auth Secret with every key generated. +func BuildAuthSecret(cluster *v1alpha1.HyperbytedbCluster) (*corev1.Secret, error) { + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: AuthSecretName(cluster), + Namespace: cluster.Namespace, + Labels: CommonLabels(cluster), + }, + Type: corev1.SecretTypeOpaque, + } + if _, err := FillAuthSecret(secret); err != nil { + return nil, err + } + return secret, nil +} + +// authEnv passes the auth Secret to the database. +func authEnv(cluster *v1alpha1.HyperbytedbCluster) []corev1.EnvVar { + from := func(name, key string) corev1.EnvVar { + return corev1.EnvVar{ + Name: name, + ValueFrom: &corev1.EnvVarSource{SecretKeyRef: &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{Name: AuthSecretName(cluster)}, + Key: key, + }}, + } + } + return []corev1.EnvVar{ + from("HYPERBYTEDB__AUTH__CLUSTER_TOKEN", AuthSecretClusterTokenKey), + from("HYPERBYTEDB__AUTH__ADMIN_USERNAME", AuthSecretAdminUsernameKey), + from("HYPERBYTEDB__AUTH__ADMIN_PASSWORD", AuthSecretAdminPasswordKey), + } +} + +type clusterTokenKey struct{} + +// WithClusterToken makes the Client send token on requests made with ctx. +// One Client serves every cluster; the token is per cluster. +func WithClusterToken(ctx context.Context, token string) context.Context { + return context.WithValue(ctx, clusterTokenKey{}, token) +} + +// tokenTransport adds the request context's cluster token, if any. +type tokenTransport struct { + base http.RoundTripper +} + +func (t tokenTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if token, _ := req.Context().Value(clusterTokenKey{}).(string); token != "" { + req = req.Clone(req.Context()) + req.Header.Set(ClusterTokenHeader, token) + } + return t.base.RoundTrip(req) +} diff --git a/internal/hyperbytedb/auth_test.go b/internal/hyperbytedb/auth_test.go new file mode 100644 index 0000000..71909e9 --- /dev/null +++ b/internal/hyperbytedb/auth_test.go @@ -0,0 +1,136 @@ +package hyperbytedb + +import ( + "context" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + + v1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" +) + +func authCluster(version string, enabled bool) *v1alpha1.HyperbytedbCluster { + c := versionedCluster(version, "", nil) + c.Name = "uat" + c.Namespace = "ns" + c.Spec.Auth.Enabled = enabled + return c +} + +func hyperbytedbContainer(t *testing.T, cluster *v1alpha1.HyperbytedbCluster) corev1.Container { + t.Helper() + for _, c := range BuildStatefulSet(cluster, "hash").Spec.Template.Spec.Containers { + if c.Name == containerName { + return c + } + } + t.Fatal("no hyperbytedb container") + return corev1.Container{} +} + +func TestFillAuthSecretKeepsWhatIsThereAndAddsTheRest(t *testing.T) { + s := &corev1.Secret{Data: map[string][]byte{AuthSecretAdminPasswordKey: []byte("chosen")}} + changed, err := FillAuthSecret(s) + if err != nil || !changed { + t.Fatalf("changed=%v err=%v", changed, err) + } + if string(s.Data[AuthSecretAdminPasswordKey]) != "chosen" { + t.Error("a chosen password was replaced") + } + if string(s.Data[AuthSecretAdminUsernameKey]) != "admin" { + t.Errorf("admin username = %q", s.Data[AuthSecretAdminUsernameKey]) + } + if len(s.Data[AuthSecretClusterTokenKey]) < 40 { + t.Errorf("token too short: %q", s.Data[AuthSecretClusterTokenKey]) + } + again, _ := FillAuthSecret(s) + if again { + t.Error("a full Secret was changed") + } + other, _ := BuildAuthSecret(authCluster("0.9.1-beta", true)) + if string(other.Data[AuthSecretClusterTokenKey]) == string(s.Data[AuthSecretClusterTokenKey]) { + t.Error("two generated tokens are equal") + } +} + +func TestAuthSecretNameDefaultsToTheCluster(t *testing.T) { + c := authCluster("0.9.1-beta", true) + if got := AuthSecretName(c); got != "uat-auth" { + t.Errorf("got %q", got) + } + c.Spec.Auth.CredentialsSecretName = "mine" + if got := AuthSecretName(c); got != "mine" { + t.Errorf("got %q", got) + } +} + +// With auth on, the pods get the Secret as env and the drain script sends +// the token. Off, or on a database without cluster auth, the pod template is +// what it was, so existing clusters are not restarted. +func TestClusterAuthReachesThePodOnlyWhenOnAndSupported(t *testing.T) { + on := hyperbytedbContainer(t, authCluster("0.9.1-beta", true)) + refs := map[string]string{} + for _, e := range on.Env { + if e.ValueFrom != nil && e.ValueFrom.SecretKeyRef != nil { + refs[e.Name] = e.ValueFrom.SecretKeyRef.Name + "/" + e.ValueFrom.SecretKeyRef.Key + } + } + want := map[string]string{ + "HYPERBYTEDB__AUTH__CLUSTER_TOKEN": "uat-auth/cluster-token", + "HYPERBYTEDB__AUTH__ADMIN_USERNAME": "uat-auth/admin-username", + "HYPERBYTEDB__AUTH__ADMIN_PASSWORD": "uat-auth/admin-password", + } + if !reflect.DeepEqual(refs, want) { + t.Errorf("secret env = %v, want %v", refs, want) + } + preStop := strings.Join(on.Lifecycle.PreStop.Exec.Command, " ") + if strings.Count(preStop, ClusterTokenHeader+": $HYPERBYTEDB__AUTH__CLUSTER_TOKEN") != 2 { + t.Errorf("preStop curls must send the token:\n%s", preStop) + } + + off := hyperbytedbContainer(t, authCluster("0.9.1-beta", false)) + old := hyperbytedbContainer(t, authCluster("0.8.5-beta", true)) + // 0.9.0-beta has no auth.cluster_token and refuses unknown config keys. + noToken := hyperbytedbContainer(t, authCluster("0.9.0-beta", true)) + for name, c := range map[string]corev1.Container{"auth off": off, "0.8.5-beta": old, "0.9.0-beta": noToken} { + for _, e := range c.Env { + if strings.HasPrefix(e.Name, "HYPERBYTEDB__AUTH__") { + t.Errorf("%s: unexpected env %s", name, e.Name) + } + } + if p := strings.Join(c.Lifecycle.PreStop.Exec.Command, " "); strings.Contains(p, ClusterTokenHeader) { + t.Errorf("%s: preStop changed:\n%s", name, p) + } + } + if !reflect.DeepEqual(off.Lifecycle, old.Lifecycle) { + t.Error("preStop differs between auth off and an old database") + } +} + +func TestTheClientSendsTheTokenOnlyWithOne(t *testing.T) { + var got []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got = append(got, r.Header.Get(ClusterTokenHeader)) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + c := NewClient() + for _, ctx := range []context.Context{ + context.Background(), + WithClusterToken(context.Background(), "tok"), + } { + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil) + resp, err := c.httpClient.Do(req) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + } + if !reflect.DeepEqual(got, []string{"", "tok"}) { + t.Errorf("headers seen = %q", got) + } +} diff --git a/internal/hyperbytedb/client.go b/internal/hyperbytedb/client.go index e0c8dd0..871c949 100644 --- a/internal/hyperbytedb/client.go +++ b/internal/hyperbytedb/client.go @@ -17,7 +17,10 @@ type Client struct { func NewClient() *Client { return &Client{ - httpClient: &http.Client{Timeout: 30 * time.Second}, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + Transport: tokenTransport{base: http.DefaultTransport}, + }, } } @@ -40,9 +43,15 @@ type NodeInfo struct { // HealthStatus represents the structured response from /health. type HealthStatus struct { - Status string `json:"status"` // "pass" (healthy), "syncing", "joining", "draining", "leaving" + Status string `json:"status"` // "pass" (healthy), "warn", "fail" Message string `json:"message,omitempty"` NodeID int `json:"node_id,omitempty"` + // State is the node's own membership record: "active", "syncing", + // "joining", "draining", "decommissioning", "leaving". Omitted before + // bootstrap or on a standalone node. This is the field to gate on, not + // Status -- Status only says whether the node is serving traffic, and a + // draining node and a fully evacuated one both report "warn". + State string `json:"state,omitempty"` } // clusterNodesResponse wraps the /cluster/nodes JSON envelope. @@ -117,6 +126,30 @@ func (c *Client) DrainNode(ctx context.Context, host string, port int32) error { return nil } +// DecommissionNode tells a node it is leaving permanently: hand off its +// primaries, then let the cluster evacuate its regions. Unlike DrainNode this +// makes the node ineligible to hold data, so it must only be called on a node +// that is actually being removed -- never on a restart. +func (c *Client) DecommissionNode(ctx context.Context, host string, port int32) error { + url := fmt.Sprintf("http://%s:%d/internal/decommission", host, port) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil) + if err != nil { + return err + } + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK && + resp.StatusCode != http.StatusNoContent && + resp.StatusCode != http.StatusAccepted { + respBody, _ := io.ReadAll(resp.Body) + return fmt.Errorf("decommission returned %d: %s", resp.StatusCode, string(respBody)) + } + return nil +} + // GetNodeHealth returns structured health information including node state. func (c *Client) GetNodeHealth(ctx context.Context, host string, port int32) (*HealthStatus, error) { url := fmt.Sprintf("http://%s:%d/health", host, port) @@ -371,6 +404,59 @@ func (c *Client) AddClusterNode(ctx context.Context, host string, port int32, no return &out, nil } +// RemoveNodeResponse mirrors the JSON returned by POST /cluster/membership/remove-node. +type RemoveNodeResponse struct { + NodeID uint64 `json:"node_id"` + RemovedFromVoter bool `json:"removed_from_voters"` + LeaderID *uint64 `json:"leader_id"` + RemainingVoters []uint64 `json:"remaining_voters"` +} + +type removeNodeRequest struct { + NodeID uint64 `json:"node_id"` +} + +// RemoveClusterNode asks `host` (which must be the current Raft leader) to +// drop `nodeID` from the Raft voter set. Removing a node that is not a voter +// succeeds with RemovedFromVoter=false. Returns *ErrNotLeader if the target +// is not currently the leader. +func (c *Client) RemoveClusterNode(ctx context.Context, host string, port int32, nodeID uint64) (*RemoveNodeResponse, error) { + url := fmt.Sprintf("http://%s:%d/cluster/membership/remove-node", host, port) + body, err := json.Marshal(removeNodeRequest{NodeID: nodeID}) + if err != nil { + return nil, fmt.Errorf("marshalling remove-node request: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode == http.StatusServiceUnavailable { + var errResp struct { + Error string `json:"error"` + LeaderID *uint64 `json:"leader_id"` + } + _ = json.NewDecoder(resp.Body).Decode(&errResp) + return nil, &ErrNotLeader{LeaderID: errResp.LeaderID} + } + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("membership/remove-node returned %d: %s", resp.StatusCode, string(respBody)) + } + + var out RemoveNodeResponse + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return nil, fmt.Errorf("decoding remove-node response: %w", err) + } + return &out, nil +} + // leaveRequest is the body for POST /internal/membership/leave. type leaveRequest struct { NodeID uint64 `json:"node_id"` diff --git a/internal/hyperbytedb/compat.go b/internal/hyperbytedb/compat.go new file mode 100644 index 0000000..f3d6b4e --- /dev/null +++ b/internal/hyperbytedb/compat.go @@ -0,0 +1,172 @@ +package hyperbytedb + +import ( + "fmt" + "strings" + + "github.com/Masterminds/semver/v3" + + v1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" +) + +// The operator renders a cluster's config and pods for the HyperbyteDB +// version it runs, so one operator manages clusters on several versions. +// A newer database rejects config keys it does not know, and an older one +// silently ignores keys it does not know (running on its defaults), so +// neither direction is safe without knowing the version. + +// DatabaseVersionAnnotation pins the HyperbyteDB version the operator +// renders for, when neither spec.version nor the image tag is a version +// (a locally built image tagged "e2e", say). +const DatabaseVersionAnnotation = "hyperbytedb.hyperbyte.cloud/database-version" + +// LatestDatabaseVersion is the newest HyperbyteDB release this operator +// knows. A cluster whose version cannot be determined is rendered for it. +const LatestDatabaseVersion = "0.9.1-beta" + +// OldestDatabaseVersion is the oldest HyperbyteDB release this operator +// renders for. Older ones are rendered as this one. +const OldestDatabaseVersion = "0.8.0" + +// Capability is something the operator only renders for a database that +// has it. +type Capability string + +const ( + // [chdb] query_pool_size and write_pool_size. Before it there is only + // pool_size, for both. + CapabilitySplitChdbPools Capability = "split-chdb-pools" + // An unset query pool is sized by the database from its CPUs, so the + // operator leaves it unset. + CapabilityAutoQueryPool Capability = "auto-query-pool" + // The [sharding] section and sharded clusters. + CapabilitySharding Capability = "sharding" + // The control listener on server port + 1 (control_port). + CapabilityControlListener Capability = "control-listener" + // Evacuating several departing nodes at once. + CapabilityEvacuateTogether Capability = "evacuate-together" + // auth.cluster_token and auth.admin_username / admin_password: a + // cluster with auth on can reach its own internal routes. + CapabilityClusterAuth Capability = "cluster-auth" +) + +// CompatibilityEntry is one row of the compatibility matrix. +type CompatibilityEntry struct { + Capability Capability + // Since is the first HyperbyteDB version that has it. + Since string + // Without says what the operator does for an older database. + Without string +} + +// CompatibilityMatrix lists what each capability needs. docs/compatibility.md +// is generated from the same facts; keep them in step. +var CompatibilityMatrix = []CompatibilityEntry{ + {CapabilitySplitChdbPools, "0.9.0-beta", "writes pool_size (spec.chdb.poolSize, else queryPoolSize, else 1) for both pools"}, + {CapabilityAutoQueryPool, "0.9.0-beta", "an unset query pool is written as pool_size = 1, as operator v0.1.x did"}, + {CapabilitySharding, "0.9.0-beta", "omits the [sharding] section; spec.sharding has no effect"}, + {CapabilityControlListener, "0.9.0-beta", "ignores the control-listener annotation (no control port, liveness on the server port)"}, + {CapabilityEvacuateTogether, "0.9.0-beta", "ignores the evacuate-together annotation (departing nodes leave one at a time)"}, + {CapabilityClusterAuth, "0.9.1-beta", "renders auth.enabled only; no auth Secret, so a cluster with auth on cannot reach its internal routes"}, +} + +// DatabaseVersion is the HyperbyteDB version a cluster is rendered for. +type DatabaseVersion struct { + Version *semver.Version + // Where it came from: "annotation", "spec.version", "image tag", or + // "assumed" (the newest this operator knows). + Source string + // Raw is what Source said, before parsing. + Raw string +} + +func (d DatabaseVersion) String() string { + return d.Version.Original() +} + +func parseDatabaseVersion(raw string) (*semver.Version, bool) { + v, err := semver.NewVersion(strings.TrimPrefix(strings.TrimSpace(raw), "v")) + if err != nil { + return nil, false + } + return v, true +} + +func imageTag(image string) string { + image = strings.TrimSpace(image) + if at := strings.Index(image, "@"); at >= 0 { + image = image[:at] + } + slash := strings.LastIndex(image, "/") + colon := strings.LastIndex(image, ":") + if colon <= slash { + return "" + } + return image[colon+1:] +} + +// ResolveDatabaseVersion is the version a cluster runs: the annotation, +// spec.version, or the image tag, the first that is a version; else the +// newest this operator knows. +func ResolveDatabaseVersion(cluster *v1alpha1.HyperbytedbCluster) DatabaseVersion { + candidates := []struct{ source, raw string }{ + {"annotation", cluster.Annotations[DatabaseVersionAnnotation]}, + {"spec.version", cluster.Spec.Version}, + {"image tag", imageTag(ResolveHyperbytedbImage(cluster))}, + } + for _, c := range candidates { + if c.raw == "" { + continue + } + if v, ok := parseDatabaseVersion(c.raw); ok { + return DatabaseVersion{Version: v, Source: c.source, Raw: c.raw} + } + } + v, _ := parseDatabaseVersion(LatestDatabaseVersion) + return DatabaseVersion{Version: v, Source: "assumed", Raw: LatestDatabaseVersion} +} + +// Supports reports whether the cluster's database version has the capability. +func Supports(cluster *v1alpha1.HyperbytedbCluster, capability Capability) bool { + return versionSupports(ResolveDatabaseVersion(cluster).Version, capability) +} + +func versionSupports(v *semver.Version, capability Capability) bool { + for _, e := range CompatibilityMatrix { + if e.Capability != capability { + continue + } + since, ok := parseDatabaseVersion(e.Since) + return ok && v.Compare(since) >= 0 + } + return false +} + +// CompatibilitySummary describes, for a status condition, the version a +// cluster is rendered for and what is left out for it. +func CompatibilitySummary(cluster *v1alpha1.HyperbytedbCluster) (reason, message string) { + dv := ResolveDatabaseVersion(cluster) + var missing []string + for _, e := range CompatibilityMatrix { + if !versionSupports(dv.Version, e.Capability) { + missing = append(missing, string(e.Capability)) + } + } + reason = "Supported" + switch { + case dv.Source == "assumed": + reason = "Assumed" + case dv.Version.Compare(semver.MustParse(OldestDatabaseVersion)) < 0: + reason = "OlderThanSupported" + case dv.Version.Compare(semver.MustParse(LatestDatabaseVersion)) > 0: + reason = "NewerThanKnown" + } + message = fmt.Sprintf("HyperbyteDB %s (from %s)", dv, dv.Source) + if dv.Source == "assumed" { + message += fmt.Sprintf("; set %s or a version image tag to pin it", DatabaseVersionAnnotation) + } + if len(missing) > 0 { + message += "; not rendered for this version: " + strings.Join(missing, ", ") + } + return reason, message +} diff --git a/internal/hyperbytedb/compat_test.go b/internal/hyperbytedb/compat_test.go new file mode 100644 index 0000000..b6b5f1b --- /dev/null +++ b/internal/hyperbytedb/compat_test.go @@ -0,0 +1,127 @@ +package hyperbytedb + +import ( + "strings" + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + + v1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" +) + +func versionedCluster(version, image string, annotations map[string]string) *v1alpha1.HyperbytedbCluster { + return &v1alpha1.HyperbytedbCluster{ + ObjectMeta: metav1.ObjectMeta{Annotations: annotations}, + Spec: v1alpha1.HyperbytedbClusterSpec{ + Replicas: ptr.To(int32(3)), + Version: version, + Image: image, + Sharding: &v1alpha1.ShardingSpec{Enabled: true}, + }, + } +} + +func TestResolveDatabaseVersion_precedence(t *testing.T) { + cases := []struct { + name string + cluster *v1alpha1.HyperbytedbCluster + wantVersion, wantSrc string + }{ + {"annotation wins", versionedCluster("0.8.5-beta", "", map[string]string{DatabaseVersionAnnotation: "0.9.0-beta"}), "0.9.0-beta", "annotation"}, + {"spec.version, v prefix dropped", versionedCluster("v0.8.5-beta", "", nil), "0.8.5-beta", "spec.version"}, + {"image tag when version is not one", versionedCluster("e2e", "ghcr.io/x/hyperbytedb:0.8.4-beta", nil), "0.8.4-beta", "image tag"}, + {"registry port is not a tag", versionedCluster("", "localhost:5000/hyperbytedb", nil), LatestDatabaseVersion, "assumed"}, + {"dev tag is assumed latest", versionedCluster("e2e", "docker.io/library/hyperbytedb", nil), LatestDatabaseVersion, "assumed"}, + } + for _, c := range cases { + got := ResolveDatabaseVersion(c.cluster) + if got.String() != c.wantVersion || got.Source != c.wantSrc { + t.Errorf("%s: got %s from %s, want %s from %s", c.name, got, got.Source, c.wantVersion, c.wantSrc) + } + } +} + +func TestSupports_followsTheMatrix(t *testing.T) { + for _, version := range []string{"0.8.0", "0.8.5", "0.8.5-beta"} { + c := versionedCluster(version, "", nil) + for _, e := range CompatibilityMatrix { + if Supports(c, e.Capability) { + t.Errorf("%s must not support %s", version, e.Capability) + } + } + } + for _, version := range []string{"0.9.0-beta", "0.9.0-beta.2", "0.9.0"} { + c := versionedCluster(version, "", nil) + for _, e := range CompatibilityMatrix { + if got, want := Supports(c, e.Capability), e.Capability != CapabilityClusterAuth; got != want { + t.Errorf("%s supports %s = %v, want %v", version, e.Capability, got, want) + } + } + } + for _, version := range []string{"0.9.1-beta", "0.9.1-beta.2", "0.9.1", "1.2.0"} { + c := versionedCluster(version, "", nil) + for _, e := range CompatibilityMatrix { + if !Supports(c, e.Capability) { + t.Errorf("%s must support %s", version, e.Capability) + } + } + } +} + +// A 0.8 database ignores keys it does not know and runs on its defaults, so +// it must get the keys it reads; a 0.9 one must not get a fixed query pool. +func TestRenderConfigTOML_perDatabaseVersion(t *testing.T) { + old := versionedCluster("0.8.5-beta", "", map[string]string{ + ControlListenerAnnotation: "true", + EvacuateTogetherAnnotation: "true", + }) + old.Spec.ChDB = v1alpha1.ChDBSpec{QueryPoolSize: 4} + out := renderConfigTOML(old) + for _, want := range []string{"\npool_size = 4\n"} { + if !strings.Contains(out, want) { + t.Errorf("0.8.5: expected %q\n%s", want, out) + } + } + for _, absent := range []string{"query_pool_size", "write_pool_size", "[sharding]", "control_port"} { + if strings.Contains(out, absent) { + t.Errorf("0.8.5: expected no %q\n%s", absent, out) + } + } + if EvacuateTogether(old) || ControlListenerEnabled(old, true) { + t.Error("0.8.5: the annotations must be ignored") + } + + cur := versionedCluster("0.9.0-beta", "", map[string]string{ControlListenerAnnotation: "true"}) + out = renderConfigTOML(cur) + for _, want := range []string{"write_pool_size = 1\n", "[sharding]", "control_port = "} { + if !strings.Contains(out, want) { + t.Errorf("0.9.0-beta: expected %q\n%s", want, out) + } + } + for _, absent := range []string{"query_pool_size", "\npool_size"} { + if strings.Contains(out, absent) { + t.Errorf("0.9.0-beta: expected no %q (the database sizes it)\n%s", absent, out) + } + } +} + +func TestCompatibilitySummary_reasons(t *testing.T) { + cases := []struct { + cluster *v1alpha1.HyperbytedbCluster + reason string + mention string + }{ + {versionedCluster("0.9.0-beta", "", nil), "Supported", "HyperbyteDB 0.9.0-beta (from spec.version)"}, + {versionedCluster("0.8.5-beta", "", nil), "Supported", "not rendered for this version: split-chdb-pools"}, + {versionedCluster("0.7.0", "", nil), "OlderThanSupported", "0.7.0"}, + {versionedCluster("2.0.0", "", nil), "NewerThanKnown", "2.0.0"}, + {versionedCluster("e2e", "", nil), "Assumed", DatabaseVersionAnnotation}, + } + for _, c := range cases { + reason, message := CompatibilitySummary(c.cluster) + if reason != c.reason || !strings.Contains(message, c.mention) { + t.Errorf("%s: got %s %q, want %s mentioning %q", c.cluster.Spec.Version, reason, message, c.reason, c.mention) + } + } +} diff --git a/internal/hyperbytedb/configmap.go b/internal/hyperbytedb/configmap.go index 4763cb6..ded9d10 100644 --- a/internal/hyperbytedb/configmap.go +++ b/internal/hyperbytedb/configmap.go @@ -73,7 +73,7 @@ func renderConfigTOMLWithClusterEnabled(cluster *v1alpha1.HyperbytedbCluster, cl writeServerSection(&b, spec) writeStorageSection(&b, spec) writeFlushSection(&b, spec) - writeChdbSection(&b, spec) + writeChdbSection(&b, cluster) writeAuthSection(&b, spec) writeCardinalitySection(&b, spec) writeLoggingSection(&b, spec) @@ -82,7 +82,7 @@ func renderConfigTOMLWithClusterEnabled(cluster *v1alpha1.HyperbytedbCluster, cl writeRateLimitSection(&b, spec) writeRetentionSection(&b, spec) writeClusterSection(&b, cluster, clusterEnabled) - writeShardingSection(&b, spec) + writeShardingSection(&b, cluster) return b.String() } @@ -150,13 +150,26 @@ func writeFlushSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec fmt.Fprintf(b, "arrow_wal_enabled = %t\n", arrowWALEnabled) } -func writeChdbSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) { +func writeChdbSection(b *strings.Builder, cluster *v1alpha1.HyperbytedbCluster) { + spec := &cluster.Spec b.WriteString("\n[chdb]\n") sessionPath := defaultChdbSessionPath if spec.ChDB.SessionDataPath != "" { sessionPath = spec.ChDB.SessionDataPath } fmt.Fprintf(b, "session_data_path = \"%s\"\n", sessionPath) + if !Supports(cluster, CapabilitySplitChdbPools) { + // One pool for both; query_pool_size would be ignored. + pool := spec.ChDB.PoolSize + if pool <= 0 { + pool = spec.ChDB.QueryPoolSize + } + if pool <= 0 { + pool = 1 + } + fmt.Fprintf(b, "pool_size = %d\n", pool) + return + } if spec.ChDB.QueryPoolSize > 0 { fmt.Fprintf(b, "query_pool_size = %d\n", spec.ChDB.QueryPoolSize) } @@ -164,11 +177,17 @@ func writeChdbSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) fmt.Fprintf(b, "write_pool_size = %d\n", spec.ChDB.WritePoolSize) } if spec.ChDB.QueryPoolSize <= 0 && spec.ChDB.WritePoolSize <= 0 { - poolSize := int32(1) - if spec.ChDB.PoolSize > 0 { - poolSize = spec.ChDB.PoolSize + switch { + case spec.ChDB.PoolSize > 0: + fmt.Fprintf(b, "pool_size = %d\n", spec.ChDB.PoolSize) + case Supports(cluster, CapabilityAutoQueryPool): + // The database sizes the query pool from its CPU limit (and + // `SET query_pool_size` changes it); a fixed number here would + // override that. The write pool keeps operator v0.1.x's 1. + b.WriteString("write_pool_size = 1\n") + default: + b.WriteString("pool_size = 1\n") } - fmt.Fprintf(b, "pool_size = %d\n", poolSize) } } @@ -193,7 +212,7 @@ func writeCardinalitySection(b *strings.Builder, spec *v1alpha1.HyperbytedbClust func writeLoggingSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) { b.WriteString("\n[logging]\n") - level := "info" + level := defaultLogLevel if spec.Logging.Level != "" { level = spec.Logging.Level } @@ -273,6 +292,9 @@ func writeClusterSection(b *strings.Builder, cluster *v1alpha1.HyperbytedbCluste b.WriteString("peers = \"\"\n") fmt.Fprintf(b, "replication_log_dir = \"%s\"\n", defaultReplLogDir) fmt.Fprintf(b, "raft_dir = \"%s\"\n", defaultRaftDir) + if ControlListenerEnabled(cluster, clusterEnabled) { + fmt.Fprintf(b, "control_port = %d\n", ControlPort(cluster)) + } heartbeatInterval := int32(2) if spec.Cluster.HeartbeatIntervalSecs > 0 { @@ -363,8 +385,9 @@ func writeMinAcks(b *strings.Builder, v *intstr.IntOrString) { } } -func writeShardingSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) { - if spec.Sharding == nil { +func writeShardingSection(b *strings.Builder, cluster *v1alpha1.HyperbytedbCluster) { + spec := &cluster.Spec + if spec.Sharding == nil || !Supports(cluster, CapabilitySharding) { return } s := spec.Sharding @@ -397,6 +420,14 @@ func writeShardingSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterS if s.PrimaryFailoverAfterSecs > 0 { fmt.Fprintf(b, "primary_failover_after_secs = %d\n", s.PrimaryFailoverAfterSecs) } + // Emitted only when explicitly set, so the server keeps its own default + // (on) rather than this silently pinning the value. + if s.AddPeerProposalsEnabled != nil { + fmt.Fprintf(b, "add_peer_proposals_enabled = %t\n", *s.AddPeerProposalsEnabled) + } + if s.RemovePeerProposalsEnabled != nil { + fmt.Fprintf(b, "remove_peer_proposals_enabled = %t\n", *s.RemovePeerProposalsEnabled) + } if s.ScatterPeerTimeoutMs > 0 { fmt.Fprintf(b, "scatter_peer_timeout_ms = %d\n", s.ScatterPeerTimeoutMs) } diff --git a/internal/hyperbytedb/configmap_test.go b/internal/hyperbytedb/configmap_test.go index 9d5e95a..3c70f1c 100644 --- a/internal/hyperbytedb/configmap_test.go +++ b/internal/hyperbytedb/configmap_test.go @@ -207,3 +207,87 @@ func TestRenderConfigTOML_sharding(t *testing.T) { t.Fatal("ConfigHash must change when sharding.enabled changes") } } + +func TestShardingUpgradeGatesReachTheConfig(t *testing.T) { + // The rollout procedure for a sharded cluster is "turn these off, upgrade + // every node, turn them back on". Until they were exposed on the CRD there + // was no way to do that through the operator: the generated config never + // carried them and the server fell back to on, while reconcileConfigMap + // overwrote any hand edit. The mitigation existed in the binary and not in + // the product. + cluster := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{ + Replicas: ptr.To(int32(3)), + Sharding: &v1alpha1.ShardingSpec{ + Enabled: true, + AddPeerProposalsEnabled: ptr.To(false), + RemovePeerProposalsEnabled: ptr.To(false), + }, + }, + } + out := renderConfigTOML(cluster) + for _, want := range []string{ + "add_peer_proposals_enabled = false", + "remove_peer_proposals_enabled = false", + } { + if !strings.Contains(out, want) { + t.Fatalf("expected config to contain %q\n\ngot:\n%s", want, out) + } + } +} + +func TestShardingUpgradeGatesOmittedWhenUnset(t *testing.T) { + // Unset must stay absent rather than being written as an explicit true. + // Emitting a value here would pin the setting from the operator side and + // silently take ownership of a server default. + cluster := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{ + Replicas: ptr.To(int32(3)), + Sharding: &v1alpha1.ShardingSpec{Enabled: true}, + }, + } + out := renderConfigTOML(cluster) + for _, absent := range []string{ + "add_peer_proposals_enabled", + "remove_peer_proposals_enabled", + } { + if strings.Contains(out, absent) { + t.Fatalf("expected config NOT to contain %q\n\ngot:\n%s", absent, out) + } + } +} + +// For a database that sizes its own query pool, unset pools leave it unset +// and keep the 1-connection write pool. Any pool field the spec sets is +// written exactly as before. +func TestRenderConfigTOML_chdbPools(t *testing.T) { + cases := []struct { + name string + chdb v1alpha1.ChDBSpec + want []string + absent []string + }{ + {"unset", v1alpha1.ChDBSpec{}, + []string{"write_pool_size = 1"}, []string{"query_pool_size", "\npool_size"}}, + {"legacy poolSize", v1alpha1.ChDBSpec{PoolSize: 6}, + []string{"\npool_size = 6"}, []string{"query_pool_size", "write_pool_size"}}, + {"queryPoolSize only", v1alpha1.ChDBSpec{QueryPoolSize: 4}, + []string{"query_pool_size = 4"}, []string{"write_pool_size", "\npool_size"}}, + } + for _, c := range cases { + cluster := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{Replicas: ptr.To(int32(1)), ChDB: c.chdb}, + } + out := renderConfigTOML(cluster) + for _, w := range c.want { + if !strings.Contains(out, w) { + t.Fatalf("%s: expected %q\n\ngot:\n%s", c.name, w, out) + } + } + for _, a := range c.absent { + if strings.Contains(out, a) { + t.Fatalf("%s: expected no %q\n\ngot:\n%s", c.name, a, out) + } + } + } +} diff --git a/internal/hyperbytedb/control_port_test.go b/internal/hyperbytedb/control_port_test.go new file mode 100644 index 0000000..a3cd03b --- /dev/null +++ b/internal/hyperbytedb/control_port_test.go @@ -0,0 +1,77 @@ +package hyperbytedb + +import ( + "strings" + "testing" + + "k8s.io/utils/ptr" + + v1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" +) + +func mainContainerOf(t *testing.T, cluster *v1alpha1.HyperbytedbCluster) (livenessPort string, ports []string) { + t.Helper() + sts := BuildStatefulSet(cluster, "hash") + for _, c := range sts.Spec.Template.Spec.Containers { + if c.Name != containerName { + continue + } + for _, p := range c.Ports { + ports = append(ports, p.Name) + } + return c.LivenessProbe.HTTPGet.Port.String(), ports + } + t.Fatal("no hyperbytedb container") + return "", nil +} + +// Liveness must use the control listener on a clustered node: on the main +// port, /ping queued behind write work past the 3s timeout and kubelet killed +// busy but healthy nodes. +func TestClusteredLivenessUsesTheControlPort(t *testing.T) { + cluster := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{Replicas: ptr.To(int32(3))}, + } + cluster.Annotations = map[string]string{ControlListenerAnnotation: "true"} + port, ports := mainContainerOf(t, cluster) + if port != "control" { + t.Errorf("liveness port = %q, want control", port) + } + if strings.Join(ports, ",") != "http,control" { + t.Errorf("container ports = %v, want [http control]", ports) + } + toml := renderConfigTOML(cluster) + if !strings.Contains(toml, "control_port = 8087\n") { + t.Errorf("config lacks control_port = 8087:\n%s", toml) + } +} + +// A single node has no control listener; nothing changes for it. +func TestSingleNodeKeepsLivenessOnTheMainPort(t *testing.T) { + cluster := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{Replicas: ptr.To(int32(1))}, + } + port, ports := mainContainerOf(t, cluster) + if port != httpPortName || strings.Join(ports, ",") != httpPortName { + t.Errorf("single node: liveness %q ports %v, want http / [http]", port, ports) + } + if strings.Contains(renderConfigTOML(cluster), "control_port") { + t.Error("single node config must not set control_port") + } +} + +// Opt-in: an existing cluster on a database build without the control +// listener must keep liveness on the main port. Defaulting it on left a +// 0.8.5-beta cluster restarting in a loop. +func TestClusteredLivenessStaysOnTheMainPortWithoutOptIn(t *testing.T) { + cluster := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{Replicas: ptr.To(int32(3))}, + } + port, ports := mainContainerOf(t, cluster) + if port != httpPortName || strings.Join(ports, ",") != httpPortName { + t.Errorf("no opt-in: liveness %q ports %v, want http / [http]", port, ports) + } + if strings.Contains(renderConfigTOML(cluster), "control_port") { + t.Error("no opt-in: config must not set control_port") + } +} diff --git a/internal/hyperbytedb/service.go b/internal/hyperbytedb/service.go index f463687..49e5a62 100644 --- a/internal/hyperbytedb/service.go +++ b/internal/hyperbytedb/service.go @@ -28,6 +28,26 @@ func ServerPort(cluster *v1alpha1.HyperbytedbCluster) int32 { return 8086 } +// ControlPort is the node's control listener: liveness and Raft RPCs, +// served on a dedicated runtime in the database so write saturation cannot +// starve them. Always the server port + 1. +func ControlPort(cluster *v1alpha1.HyperbytedbCluster) int32 { + return ServerPort(cluster) + 1 +} + +// ControlListenerAnnotation opts a cluster into the control listener. +const ControlListenerAnnotation = "hyperbytedb.hyperbyte.cloud/control-listener" + +// ControlListenerEnabled reports whether this cluster's pods get +// control_port and a liveness probe on it. Opt-in, because only database +// builds with the control listener serve that port: turning it on for every +// clustered CR left an existing 0.8.5-beta cluster failing liveness on a +// port nothing listened on, restarting in a loop. +func ControlListenerEnabled(cluster *v1alpha1.HyperbytedbCluster, clusterEnabled bool) bool { + return clusterEnabled && cluster.Annotations[ControlListenerAnnotation] == annotationTrue && + Supports(cluster, CapabilityControlListener) +} + func BuildHeadlessService(cluster *v1alpha1.HyperbytedbCluster) *corev1.Service { port := serverPort(cluster) return &corev1.Service{ @@ -42,9 +62,9 @@ func BuildHeadlessService(cluster *v1alpha1.HyperbytedbCluster) *corev1.Service PublishNotReadyAddresses: true, Ports: []corev1.ServicePort{ { - Name: "http", + Name: httpPortName, Port: port, - TargetPort: intstr.FromString("http"), + TargetPort: intstr.FromString(httpPortName), Protocol: corev1.ProtocolTCP, }, }, @@ -65,9 +85,9 @@ func BuildClientService(cluster *v1alpha1.HyperbytedbCluster) *corev1.Service { Selector: CommonLabels(cluster), Ports: []corev1.ServicePort{ { - Name: "http", + Name: httpPortName, Port: port, - TargetPort: intstr.FromString("http"), + TargetPort: intstr.FromString(httpPortName), Protocol: corev1.ProtocolTCP, }, }, diff --git a/internal/hyperbytedb/servicemonitor.go b/internal/hyperbytedb/servicemonitor.go index 471d40d..6d9b69e 100644 --- a/internal/hyperbytedb/servicemonitor.go +++ b/internal/hyperbytedb/servicemonitor.go @@ -25,7 +25,7 @@ func BuildServiceMonitor(cluster *v1alpha1.HyperbytedbCluster) *unstructured.Uns }, "endpoints": []any{ map[string]any{ - "port": "http", + "port": httpPortName, "path": "/metrics", "interval": "15s", }, diff --git a/internal/hyperbytedb/statefulset.go b/internal/hyperbytedb/statefulset.go index 2250bb0..c4fc321 100644 --- a/internal/hyperbytedb/statefulset.go +++ b/internal/hyperbytedb/statefulset.go @@ -21,6 +21,14 @@ const ( // podTerminationGraceSecs is the kubelet grace budget (includes preStop). Keep // comfortably above drainAckWaitSecs + headroom for flush + SIGTERM cleanup. podTerminationGraceSecs = 120 + // containerName is the database container's name in every pod. + containerName = "hyperbytedb" + // httpPortName names the database's HTTP port on pods and Services. + httpPortName = "http" + // defaultLogLevel is the log level without spec.logging.level. + defaultLogLevel = "info" + // annotationTrue is the value that turns an opt-in annotation on. + annotationTrue = "true" ) func StatefulSetName(cluster *v1alpha1.HyperbytedbCluster) string { @@ -139,10 +147,25 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve if cluster.Spec.ChDB.SessionDataPath != "" { chdbPath = cluster.Spec.ChDB.SessionDataPath } + // The image bakes HYPERBYTEDB__LOGGING__LEVEL=info, and the HYPERBYTEDB__* + // env overlay is applied OVER the mounted config.toml. Rendering the level + // into the ConfigMap alone therefore did nothing -- spec.logging.level was + // inert, and a sharded cluster could not be put into debug to diagnose a + // placement or staging failure at all. Set it as env, like the four other + // image defaults this block already restates, so the CR field wins. + logLevel := defaultLogLevel + if cluster.Spec.Logging.Level != "" { + logLevel = cluster.Spec.Logging.Level + } env := []corev1.EnvVar{ {Name: "HYPERBYTEDB__STORAGE__WAL_DIR", Value: defaultWalDir}, {Name: "HYPERBYTEDB__STORAGE__META_DIR", Value: defaultMetaDir}, {Name: "HYPERBYTEDB__CHDB__SESSION_DATA_PATH", Value: chdbPath}, + {Name: "HYPERBYTEDB__LOGGING__LEVEL", Value: logLevel}, + } + clusterAuth := ClusterAuthEnabled(cluster) + if clusterAuth { + env = append(env, authEnv(cluster)...) } // Cluster mode and paths come from mounted config.toml (hot-updated on scale); // avoid replica-dependent env vars so the pod template stays stable across scale events. @@ -171,12 +194,23 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve probeScheme = corev1.URISchemeHTTPS } + // Liveness goes to the control listener when the cluster opts in: it is + // served on the database's control runtime, so a node saturated with + // writes still answers. On the main port, /ping queued behind write work + // past the 3s timeout and kubelet killed busy but healthy nodes. + livenessPort := intstr.FromString(httpPortName) + livenessScheme := probeScheme + controlListener := ControlListenerEnabled(cluster, clusterEnabled) + if controlListener { + livenessPort = intstr.FromString("control") + livenessScheme = corev1.URISchemeHTTP + } livenessProbe := &corev1.Probe{ ProbeHandler: corev1.ProbeHandler{ HTTPGet: &corev1.HTTPGetAction{ Path: "/ping", - Port: intstr.FromString("http"), - Scheme: probeScheme, + Port: livenessPort, + Scheme: livenessScheme, }, }, InitialDelaySeconds: 15, @@ -188,7 +222,7 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve ProbeHandler: corev1.ProbeHandler{ HTTPGet: &corev1.HTTPGetAction{ Path: "/health", - Port: intstr.FromString("http"), + Port: intstr.FromString(httpPortName), Scheme: probeScheme, }, }, @@ -209,7 +243,7 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve ProbeHandler: corev1.ProbeHandler{ HTTPGet: &corev1.HTTPGetAction{ Path: "/ping", - Port: intstr.FromString("http"), + Port: intstr.FromString(httpPortName), Scheme: probeScheme, }, }, @@ -221,30 +255,24 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve // --------------- main container --------------- mainContainer := corev1.Container{ - Name: "hyperbytedb", + Name: containerName, Image: image, ImagePullPolicy: pullPolicy, Command: []string{"sh", "/entrypoint/start.sh"}, Env: env, - Ports: []corev1.ContainerPort{ - { - Name: "http", - ContainerPort: port, - Protocol: corev1.ProtocolTCP, - }, - }, - VolumeMounts: volumeMounts, - StartupProbe: startupProbe, - LivenessProbe: livenessProbe, - ReadinessProbe: readinessProbe, - Resources: cluster.Spec.Resources, + Ports: containerPorts(port, controlListener, ControlPort(cluster)), + VolumeMounts: volumeMounts, + StartupProbe: startupProbe, + LivenessProbe: livenessProbe, + ReadinessProbe: readinessProbe, + Resources: cluster.Spec.Resources, } if clusterEnabled { mainContainer.Lifecycle = &corev1.Lifecycle{ PreStop: &corev1.LifecycleHandler{ Exec: &corev1.ExecAction{ - Command: []string{"sh", "-c", clusterPreStopScript(port)}, + Command: []string{"sh", "-c", clusterPreStopScript(port, clusterAuth)}, }, }, } @@ -328,7 +356,7 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve Selector: &metav1.LabelSelector{ MatchLabels: CommonLabels(cluster), }, - PodManagementPolicy: podManagementPolicy(clusterEnabled), + PodManagementPolicy: podManagementPolicy(cluster, clusterEnabled), UpdateStrategy: appsv1.StatefulSetUpdateStrategy{ Type: appsv1.RollingUpdateStatefulSetStrategyType, }, @@ -353,8 +381,28 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve return sts } -func podManagementPolicy(clusterEnabled bool) appsv1.PodManagementPolicyType { - if clusterEnabled { +// EvacuateTogetherAnnotation opts a cluster into decommissioning every +// departing node of a scale-down at once instead of one at a time. +const EvacuateTogetherAnnotation = "hyperbytedb.hyperbyte.cloud/evacuate-together" + +// EvacuateTogether reports whether a scale-down evacuates all departing nodes +// at once. Opt-in: an older database may not expect several at a time. +func EvacuateTogether(cluster *v1alpha1.HyperbytedbCluster) bool { + return cluster.Annotations[EvacuateTogetherAnnotation] == annotationTrue && + Supports(cluster, CapabilityEvacuateTogether) +} + +// ParallelPodsAnnotation opts a cluster into Parallel pod management. +const ParallelPodsAnnotation = "hyperbytedb.hyperbyte.cloud/parallel-pods" + +// podManagementPolicy is OrderedReady for a cluster unless it opts into +// Parallel. Under OrderedReady a departing node reports unready, and the +// StatefulSet will not create a pod above an unready one: on n3 a pod deleted +// during a scale-down was never recreated, and a full restart of 8 pods took +// over an hour, one pod at a time. Opt-in, because an existing cluster's +// policy cannot change in place (the controller recreates its StatefulSet). +func podManagementPolicy(cluster *v1alpha1.HyperbytedbCluster, clusterEnabled bool) appsv1.PodManagementPolicyType { + if clusterEnabled && cluster.Annotations[ParallelPodsAnnotation] != annotationTrue { return appsv1.OrderedReadyPodManagement } return appsv1.ParallelPodManagement @@ -364,13 +412,38 @@ func podManagementPolicy(clusterEnabled bool) appsv1.PodManagementPolicyType { // reports state=leaving (flush + replication acks + peer leave complete) or // until drainAckWaitSecs elapses. A fixed sleep is insufficient because // DrainService may wait up to 60s for peer replication acks alone. -func clusterPreStopScript(port int32) string { - return fmt.Sprintf(`curl -sf -X POST http://localhost:%[1]d/internal/drain || true +// +// With cluster auth the curls send the cluster token from the pod's env; +// without it the script is unchanged, so existing pods are not restarted. +func clusterPreStopScript(port int32, clusterAuth bool) string { + auth := "" + if clusterAuth { + auth = ` -H "` + ClusterTokenHeader + `: $HYPERBYTEDB__AUTH__CLUSTER_TOKEN"` + } + return fmt.Sprintf(`curl -sf%[3]s -X POST http://localhost:%[1]d/internal/drain || true for i in $(seq 1 %[2]d); do - if curl -sf http://localhost:%[1]d/cluster/metrics 2>/dev/null | grep -q '"state":"leaving"'; then + if curl -sf%[3]s http://localhost:%[1]d/cluster/metrics 2>/dev/null | grep -q '"state":"leaving"'; then exit 0 fi sleep 1 done -exit 0`, port, drainAckWaitSecs) +exit 0`, port, drainAckWaitSecs, auth) +} + +// containerPorts lists the main port and, when enabled, the control +// listener's port. +func containerPorts(port int32, controlListener bool, controlPort int32) []corev1.ContainerPort { + ports := []corev1.ContainerPort{{ + Name: "http", + ContainerPort: port, + Protocol: corev1.ProtocolTCP, + }} + if controlListener { + ports = append(ports, corev1.ContainerPort{ + Name: "control", + ContainerPort: controlPort, + Protocol: corev1.ProtocolTCP, + }) + } + return ports } diff --git a/internal/hyperbytedb/statefulset_test.go b/internal/hyperbytedb/statefulset_test.go new file mode 100644 index 0000000..83a3de6 --- /dev/null +++ b/internal/hyperbytedb/statefulset_test.go @@ -0,0 +1,94 @@ +package hyperbytedb + +import ( + "testing" + + appsv1 "k8s.io/api/apps/v1" + "k8s.io/utils/ptr" + + v1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" +) + +func podEnv(t *testing.T, cluster *v1alpha1.HyperbytedbCluster, name string) (string, bool) { + t.Helper() + sts := BuildStatefulSet(cluster, "hash") + for _, c := range sts.Spec.Template.Spec.Containers { + if c.Name != containerName { + continue + } + for _, e := range c.Env { + if e.Name == name { + return e.Value, true + } + } + } + return "", false +} + +// The image bakes HYPERBYTEDB__LOGGING__LEVEL=info and the HYPERBYTEDB__* env +// overlay is applied OVER the mounted config.toml. Rendering the level into the +// ConfigMap alone was therefore inert: spec.logging.level had no effect, and a +// sharded cluster could not be put into debug to diagnose a placement or +// staging failure. The env var is the only thing that can win. +func TestLoggingLevelReachesThePodEnv(t *testing.T) { + cluster := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{ + Replicas: ptr.To(int32(3)), + Logging: v1alpha1.LoggingSpec{Level: "debug"}, + }, + } + got, ok := podEnv(t, cluster, "HYPERBYTEDB__LOGGING__LEVEL") + if !ok { + t.Fatal("HYPERBYTEDB__LOGGING__LEVEL absent from pod env; the image default would silently win") + } + if got != "debug" { + t.Fatalf("HYPERBYTEDB__LOGGING__LEVEL = %q, want %q", got, "debug") + } +} + +// Unset must still pin a level explicitly rather than leaving the image's +// baked value to decide -- otherwise the operator's behaviour depends on which +// image tag happens to be deployed. +func TestLoggingLevelDefaultsToInfoInPodEnv(t *testing.T) { + cluster := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{Replicas: ptr.To(int32(3))}, + } + got, ok := podEnv(t, cluster, "HYPERBYTEDB__LOGGING__LEVEL") + if !ok { + t.Fatal("HYPERBYTEDB__LOGGING__LEVEL absent from pod env") + } + if got != "info" { + t.Fatalf("HYPERBYTEDB__LOGGING__LEVEL = %q, want %q", got, "info") + } +} + +// A cluster keeps OrderedReady unless it opts in; a single node is Parallel. +func TestPodManagementIsParallelOnlyWhenAClusterOptsIn(t *testing.T) { + cluster := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{Replicas: ptr.To(int32(3))}, + } + if got := BuildStatefulSet(cluster, "h").Spec.PodManagementPolicy; got != appsv1.OrderedReadyPodManagement { + t.Errorf("cluster without the annotation: %s, want OrderedReady", got) + } + cluster.Annotations = map[string]string{ParallelPodsAnnotation: "true"} + if got := BuildStatefulSet(cluster, "h").Spec.PodManagementPolicy; got != appsv1.ParallelPodManagement { + t.Errorf("cluster with the annotation: %s, want Parallel", got) + } + single := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{Replicas: ptr.To(int32(1))}, + } + if got := BuildStatefulSet(single, "h").Spec.PodManagementPolicy; got != appsv1.ParallelPodManagement { + t.Errorf("single node: %s, want Parallel", got) + } +} + +func TestEvacuateTogetherIsOptIn(t *testing.T) { + cluster := &v1alpha1.HyperbytedbCluster{} + if EvacuateTogether(cluster) { + t.Error("on without the annotation") + } + cluster.Annotations = map[string]string{EvacuateTogetherAnnotation: "true"} + if !EvacuateTogether(cluster) { + t.Error("off with the annotation") + } +}