From 64af55a365263bdad9d4f18750a458e20aa563f4 Mon Sep 17 00:00:00 2001 From: "austin.barrington" Date: Wed, 15 Jul 2026 19:09:09 +0100 Subject: [PATCH 1/4] a --- api/v1alpha1/hyperbytedbcluster_types.go | 64 +++- api/v1alpha1/zz_generated.deepcopy.go | 21 ++ ...b.hyperbyte.cloud_hyperbytedbclusters.yaml | 62 +++- .../hyperbytedbcluster_controller.go | 332 ++++++++++++++++++ internal/hyperbytedb/client.go | 79 +++++ internal/hyperbytedb/configmap.go | 16 +- internal/hyperbytedb/proxy.go | 5 + 7 files changed, 567 insertions(+), 12 deletions(-) diff --git a/api/v1alpha1/hyperbytedbcluster_types.go b/api/v1alpha1/hyperbytedbcluster_types.go index b18f109..135844f 100644 --- a/api/v1alpha1/hyperbytedbcluster_types.go +++ b/api/v1alpha1/hyperbytedbcluster_types.go @@ -229,10 +229,21 @@ type ChDBSpec struct { // +optional SessionDataPath string `json:"sessionDataPath,omitempty"` - // Ignored by hyperbytedb (libchdb is a process-global singleton). Retained - // for API stability; always written as 1 in config.toml. - // +kubebuilder:default=1 + // Sets both query and write pool sizes when QueryPoolSize and WritePoolSize + // are unset. Defaults to 1 when all pool fields are unset. + // +optional + // +kubebuilder:validation:Minimum=1 PoolSize int32 `json:"poolSize,omitempty"` + + // chDB connections reserved for queries. Isolated from ingest/flush. + // +optional + // +kubebuilder:validation:Minimum=1 + QueryPoolSize int32 `json:"queryPoolSize,omitempty"` + + // chDB connections reserved for ingest and flush (Arrow WAL build, INSERTs). + // +optional + // +kubebuilder:validation:Minimum=1 + WritePoolSize int32 `json:"writePoolSize,omitempty"` } type AuthSpec struct { @@ -264,7 +275,7 @@ type ClusterTuningSpec struct { // +kubebuilder:default=5 ReplicationMaxRetries int32 `json:"replicationMaxRetries,omitempty"` - // +kubebuilder:default=300 + // +kubebuilder:default=1000 RaftHeartbeatIntervalMs int32 `json:"raftHeartbeatIntervalMs,omitempty"` // +kubebuilder:default=1000 @@ -306,6 +317,12 @@ type ClusterTuningSpec struct { // TLS for inter-node replication traffic. // +optional TLS *TLSSpec `json:"tls,omitempty"` + + // Seconds to wait after excluding a backend from the proxy before + // deleting its pod. Gives in-flight requests time to drain. + // +kubebuilder:default=10 + // +optional + DrainWaitSecs int32 `json:"drainWaitSecs,omitempty"` } // ReplicationSpec controls coordinator-side replication (how this node's @@ -543,6 +560,40 @@ const ( ClusterPhaseFailed ClusterPhase = "Failed" ) +// RollingRestartPhase tracks which step of the proxy-coordinated rolling +// restart the operator is currently executing. +type RollingRestartPhase string + +const ( + RollingRestartExcluding RollingRestartPhase = "Excluding" + RollingRestartDraining RollingRestartPhase = "Draining" + RollingRestartWaitingReady RollingRestartPhase = "WaitingReady" + RollingRestartIncluding RollingRestartPhase = "Including" + RollingRestartCompleted RollingRestartPhase = "Completed" +) + +// RollingRestartState tracks pod-by-pod proxy exclusion during rolling upgrades. +// Nil when no rolling restart is in progress. +type RollingRestartState struct { + // Ordinal of the pod currently being restarted. + CurrentOrdinal int32 `json:"currentOrdinal"` + + // Total number of pod ordinals to cycle through. + TotalOrdinals int32 `json:"totalOrdinals"` + + // Current phase of the restart state machine. + Phase RollingRestartPhase `json:"phase"` + + // When the current phase started (used to compute drain wait). + PhaseStartedAt metav1.Time `json:"phaseStartedAt"` + + // IP of the pod being excluded (set during Excluding phase). + OldPodIP string `json:"oldPodIP,omitempty"` + + // True once the proxy confirms the backend is excluded. + ExcludeConfirmed bool `json:"excludeConfirmed"` +} + // MemberStatus describes the observed state of a single cluster member. type MemberStatus struct { // Stable identifier derived from the StatefulSet ordinal. @@ -604,6 +655,11 @@ type HyperbytedbClusterStatus struct { // +optional ConfigHash string `json:"configHash,omitempty"` + // Tracks pod-by-pod proxy exclusion during rolling upgrades. + // Nil when no rolling restart is in progress. + // +optional + RollingRestart *RollingRestartState `json:"rollingRestart,omitempty"` + // +listType=map // +listMapKey=type // +optional diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index d28862f..f3ef7ae 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -482,6 +482,11 @@ func (in *HyperbytedbClusterStatus) DeepCopyInto(out *HyperbytedbClusterStatus) (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.RollingRestart != nil { + in, out := &in.RollingRestart, &out.RollingRestart + *out = new(RollingRestartState) + (*in).DeepCopyInto(*out) + } if in.Conditions != nil { in, out := &in.Conditions, &out.Conditions *out = make([]v1.Condition, len(*in)) @@ -812,6 +817,22 @@ func (in *RetentionSpec) DeepCopy() *RetentionSpec { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *RollingRestartState) DeepCopyInto(out *RollingRestartState) { + *out = *in + in.PhaseStartedAt.DeepCopyInto(&out.PhaseStartedAt) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RollingRestartState. +func (in *RollingRestartState) DeepCopy() *RollingRestartState { + if in == nil { + return nil + } + out := new(RollingRestartState) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *S3BackupSpec) DeepCopyInto(out *S3BackupSpec) { *out = *in diff --git a/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml b/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml index e89299b..fa7ae24 100644 --- a/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml +++ b/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml @@ -2977,20 +2977,39 @@ spec: chdb: properties: poolSize: - default: 1 description: |- - Ignored by hyperbytedb (libchdb is a process-global singleton). Retained - for API stability; always written as 1 in config.toml. + Sets both query and write pool sizes when QueryPoolSize and WritePoolSize + are unset. Defaults to 1 when all pool fields are unset. format: int32 + minimum: 1 + type: integer + queryPoolSize: + description: chDB connections reserved for queries. Isolated from + ingest/flush. + format: int32 + minimum: 1 type: integer sessionDataPath: description: |- chDB session directory inside the data volume. Defaults to /var/lib/hyperbytedb/chdb when unset. type: string + writePoolSize: + description: chDB connections reserved for ingest and flush (Arrow + WAL build, INSERTs). + format: int32 + minimum: 1 + type: integer type: object cluster: properties: + drainWaitSecs: + default: 10 + description: |- + Seconds to wait after excluding a backend from the proxy before + deleting its pod. Gives in-flight requests time to drain. + format: int32 + type: integer heartbeatIntervalSecs: default: 2 format: int32 @@ -3004,7 +3023,7 @@ spec: format: int32 type: integer raftHeartbeatIntervalMs: - default: 300 + default: 1000 format: int32 type: integer raftSnapshotThreshold: @@ -3977,6 +3996,41 @@ spec: description: 'Replication convergence state: Healthy, Lagging, Diverged, Unknown.' type: string + rollingRestart: + description: |- + Tracks pod-by-pod proxy exclusion during rolling upgrades. + Nil when no rolling restart is in progress. + properties: + currentOrdinal: + description: Ordinal of the pod currently being restarted. + format: int32 + type: integer + excludeConfirmed: + description: True once the proxy confirms the backend is excluded. + type: boolean + oldPodIP: + description: IP of the pod being excluded (set during Excluding + phase). + type: string + phase: + description: Current phase of the restart state machine. + type: string + phaseStartedAt: + description: When the current phase started (used to compute drain + wait). + format: date-time + type: string + totalOrdinals: + description: Total number of pod ordinals to cycle through. + format: int32 + type: integer + required: + - currentOrdinal + - excludeConfirmed + - phase + - phaseStartedAt + - totalOrdinals + type: object type: object required: - spec diff --git a/internal/controller/hyperbytedbcluster_controller.go b/internal/controller/hyperbytedbcluster_controller.go index d4ca64e..886374f 100644 --- a/internal/controller/hyperbytedbcluster_controller.go +++ b/internal/controller/hyperbytedbcluster_controller.go @@ -34,6 +34,7 @@ import ( "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/client-go/tools/record" + "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" @@ -180,6 +181,18 @@ func (r *HyperbytedbClusterReconciler) Reconcile(ctx context.Context, req ctrl.R return r.setFailedStatus(ctx, cluster, "StatefulSetFailed", err) } + // 6b. Operator-driven rolling restart coordination. When the StatefulSet + // has a pending rolling update and the proxy is enabled, the operator + // orchestrates pod-by-pod exclusion from the proxy, drain wait, pod + // deletion, and inclusion after the replacement pod is healthy. + if hyperbytedb.ProxyEnabled(cluster) { + if result, err := r.reconcileRollingRestart(ctx, cluster, stsResult, replicas); err != nil { + return r.setFailedStatus(ctx, cluster, "RollingRestartFailed", err) + } else if result != nil { + return *result, nil + } + } + // 7. Handle scaling if stsResult.SpecReplicas != replicas { if replicas > stsResult.SpecReplicas { @@ -395,6 +408,19 @@ func (r *HyperbytedbClusterReconciler) reconcileStatefulSet(ctx context.Context, existing.Spec.Template.Labels = desired.Spec.Template.Labels existing.Spec.UpdateStrategy = desired.Spec.UpdateStrategy + // When the proxy is enabled the operator drives pod lifecycle itself + // (exclude → drain → delete → include) so there is no need to pin + // the StatefulSet partition. Clear any leftover partition from a + // previous operator version so the STS controller can reconcile + // normally after the operator coordination finishes. + if existing.Spec.UpdateStrategy.RollingUpdate != nil && + existing.Spec.UpdateStrategy.RollingUpdate.Partition != nil && + *existing.Spec.UpdateStrategy.RollingUpdate.Partition != 0 { + existing.Spec.UpdateStrategy.RollingUpdate = &appsv1.RollingUpdateStatefulSetStrategy{ + Partition: ptr.To(int32(0)), + } + } + if err := r.Update(ctx, existing); err != nil { return stsReconcileResult{}, err } @@ -405,6 +431,312 @@ func (r *HyperbytedbClusterReconciler) reconcileStatefulSet(ctx context.Context, }, nil } +// ---------- Rolling Restart Coordination ---------- + +// reconcileRollingRestart orchestrates pod-by-pod proxy exclusion during a +// StatefulSet rolling upgrade. Returns a non-nil *ctrl.Result when the +// caller should return early (requeue), or nil when no coordination is +// needed (either no upgrade in progress or coordination is complete). +func (r *HyperbytedbClusterReconciler) reconcileRollingRestart( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + stsResult stsReconcileResult, + replicas int32, +) (*ctrl.Result, error) { + log := logf.FromContext(ctx) + + // Check if a rolling upgrade is actually in progress. + curSTS := &appsv1.StatefulSet{} + if err := r.Get(ctx, types.NamespacedName{ + Name: hyperbytedb.StatefulSetName(cluster), Namespace: cluster.Namespace, + }, curSTS); err != nil { + return nil, err + } + rollingUpgrade := curSTS.Status.UpdateRevision != "" && + curSTS.Status.UpdateRevision != curSTS.Status.CurrentRevision + + // No rolling upgrade and no active restart state → nothing to do. + if !rollingUpgrade && cluster.Status.RollingRestart == nil { + return nil, nil + } + + // If the upgrade finished (all replicas ready and up-to-date) AND the + // State machine reached Completed — clear coordination state once all + // pods are ready and up-to-date. + if cluster.Status.RollingRestart != nil && + cluster.Status.RollingRestart.Phase == hyperbytedbv1alpha1.RollingRestartCompleted { + if stsResult.ReadyReplicas == stsResult.SpecReplicas && + curSTS.Status.ReadyReplicas == curSTS.Status.Replicas && + curSTS.Status.UpdatedReplicas == curSTS.Status.Replicas { + cluster.Status.RollingRestart = nil + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + log.Info("Rolling restart complete, cleared coordination state") + return nil, nil + } + } + + proxyPort := hyperbytedb.ServerPort(cluster) + stsName := hyperbytedb.StatefulSetName(cluster) + + // Initialise state if this is the start of a new rolling upgrade. + if cluster.Status.RollingRestart == nil { + cluster.Status.RollingRestart = &hyperbytedbv1alpha1.RollingRestartState{ + CurrentOrdinal: 0, + TotalOrdinals: replicas, + Phase: hyperbytedbv1alpha1.RollingRestartExcluding, + PhaseStartedAt: metav1.Now(), + } + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + log.Info("Starting operator-driven rolling restart", "totalOrdinals", replicas) + } + + state := cluster.Status.RollingRestart + if state == nil { + // CRD schema didn't persist the field (stale CRD or server-side pruning). + // Re-initialize so the state machine can proceed. + state = &hyperbytedbv1alpha1.RollingRestartState{ + CurrentOrdinal: 0, + TotalOrdinals: replicas, + Phase: hyperbytedbv1alpha1.RollingRestartExcluding, + PhaseStartedAt: metav1.Now(), + } + cluster.Status.RollingRestart = state + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + } + podName := fmt.Sprintf("%s-%d", stsName, state.CurrentOrdinal) + + requeue3s := ctrl.Result{RequeueAfter: 3 * time.Second} + + switch state.Phase { + case hyperbytedbv1alpha1.RollingRestartExcluding: + // Get the pod's IP. + pod := &corev1.Pod{} + if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { + if apierrors.IsNotFound(err) { + // Pod not yet created; skip to WaitingReady (StatefulSet is still rolling). + state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + return &requeue3s, nil + } + return nil, err + } + + podIP := pod.Status.PodIP + if podIP == "" { + log.Info("Pod has no IP yet, waiting", "pod", podName) + return &requeue3s, nil + } + state.OldPodIP = podIP + + // Call proxy exclude on ALL proxy pods (not just one via Service round-robin). + if err := r.excludeFromAllProxies(ctx, cluster, podIP, proxyPort); err != nil { + log.V(1).Info("Could not exclude backend from proxy, will retry", "pod", podName, "ip", podIP, "error", err) + return &requeue3s, nil + } + log.Info("Excluded backend from proxy", "pod", podName, "ip", podIP) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendExcluded", + "Excluded pod %s (%s) from proxy routing", podName, podIP) + state.ExcludeConfirmed = true + state.Phase = hyperbytedbv1alpha1.RollingRestartDraining + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + return &requeue3s, nil + + case hyperbytedbv1alpha1.RollingRestartDraining: + // Wait for drain time. + drainWait := time.Duration(cluster.Spec.Cluster.DrainWaitSecs) * time.Second + if drainWait <= 0 { + drainWait = 10 * time.Second + } + elapsed := time.Since(state.PhaseStartedAt.Time) + if elapsed < drainWait { + log.Info("Waiting for in-flight requests to drain", + "pod", podName, "elapsed", elapsed.Truncate(time.Second), "wait", drainWait) + return &ctrl.Result{RequeueAfter: drainWait - elapsed + time.Second}, nil + } + + // Delete the pod to trigger recreation from the updated StatefulSet template. + pod := &corev1.Pod{} + if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { + if apierrors.IsNotFound(err) { + // Already gone, move to WaitingReady. + state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + return &requeue3s, nil + } + return nil, err + } + + if err := r.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) { + return nil, fmt.Errorf("deleting pod %s for rolling restart: %w", podName, err) + } + log.Info("Deleted pod for rolling restart", "pod", podName) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "PodRestarted", + "Deleted pod %s for rolling upgrade", podName) + state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + return &requeue3s, nil + + case hyperbytedbv1alpha1.RollingRestartWaitingReady: + // Wait for the new pod to be Ready. + pod := &corev1.Pod{} + if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { + if apierrors.IsNotFound(err) { + log.Info("Waiting for replacement pod to appear", "pod", podName) + return &requeue3s, nil + } + return nil, err + } + + if !isPodReady(pod) || pod.DeletionTimestamp != nil { + if pod.DeletionTimestamp != nil { + log.Info("Old pod still terminating, waiting for replacement", "pod", podName) + } else { + log.Info("Replacement pod not ready yet", "pod", podName) + } + return &requeue3s, nil + } + + // Pod is ready and not terminating — get its IP and include it. + newIP := pod.Status.PodIP + if newIP == "" { + log.Info("Pod is ready but has no IP, waiting", "pod", podName) + return &requeue3s, nil + } + + if err := r.includeFromAllProxies(ctx, cluster, newIP, proxyPort); err != nil { + log.V(1).Info("Could not include backend in proxy, will retry", "pod", podName, "ip", newIP, "error", err) + return &requeue3s, nil + } + log.Info("Included backend in proxy", "pod", podName, "ip", newIP) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendIncluded", + "Included pod %s (%s) in proxy routing", podName, newIP) + + // Move to next ordinal or finish. + state.CurrentOrdinal++ + if state.CurrentOrdinal >= state.TotalOrdinals { + // All pods have been cycled. Set Completed phase and let the + // "upgrade finished" check clear the state once the STS settles. + state.Phase = hyperbytedbv1alpha1.RollingRestartCompleted + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + log.Info("Rolling restart coordination complete") + return &ctrl.Result{RequeueAfter: 3 * time.Second}, nil + } + state.Phase = hyperbytedbv1alpha1.RollingRestartExcluding + state.PhaseStartedAt = metav1.Now() + state.ExcludeConfirmed = false + state.OldPodIP = "" + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + return &requeue3s, nil + } + + // Coordination complete but STS hasn't fully settled yet — just wait. + if state.Phase == hyperbytedbv1alpha1.RollingRestartCompleted { + return &requeue3s, nil + } + + return nil, nil +} + +// isPodReady returns true if the pod has a Ready condition with Status=True. +func isPodReady(pod *corev1.Pod) bool { + for _, cond := range pod.Status.Conditions { + if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue { + return true + } + } + return false +} + +// excludeFromAllProxies sends the exclude command to every proxy pod +// individually, so the exclusion takes effect on all proxy instances (the +// ClusterIP Service only round-robins to one). Returns an error if any +// proxy pod is not ready so the caller retries. +func (r *HyperbytedbClusterReconciler) excludeFromAllProxies( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + backendIP string, + proxyPort int32, +) error { + proxyPods := &corev1.PodList{} + if err := r.List(ctx, proxyPods, + client.InNamespace(cluster.Namespace), + client.MatchingLabels{ + "app.kubernetes.io/name": "hyperbytedb-proxy", + "app.kubernetes.io/instance": cluster.Name, + "app.kubernetes.io/managed-by": "hyperbytedb-operator", + "app.kubernetes.io/component": "proxy", + }, + ); err != nil { + return fmt.Errorf("listing proxy pods: %w", err) + } + for i := range proxyPods.Items { + pod := &proxyPods.Items[i] + if pod.Status.PodIP == "" || !isPodReady(pod) { + return fmt.Errorf("proxy pod %s not ready (IP=%s, ready=%v), will retry", pod.Name, pod.Status.PodIP, isPodReady(pod)) + } + if err := r.Members.Client.ExcludeProxyBackend(ctx, pod.Status.PodIP, proxyPort, backendIP); err != nil { + return fmt.Errorf("exclude on proxy pod %s: %w", pod.Name, err) + } + } + return nil +} + +// includeFromAllProxies sends the include command to every proxy pod. +// Returns an error if any proxy pod is not ready, so the caller retries — +// skipping a proxy would leave a stale exclusion on that instance. +func (r *HyperbytedbClusterReconciler) includeFromAllProxies( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + backendIP string, + proxyPort int32, +) error { + proxyPods := &corev1.PodList{} + if err := r.List(ctx, proxyPods, + client.InNamespace(cluster.Namespace), + client.MatchingLabels{ + "app.kubernetes.io/name": "hyperbytedb-proxy", + "app.kubernetes.io/instance": cluster.Name, + "app.kubernetes.io/managed-by": "hyperbytedb-operator", + "app.kubernetes.io/component": "proxy", + }, + ); err != nil { + return fmt.Errorf("listing proxy pods: %w", err) + } + for i := range proxyPods.Items { + pod := &proxyPods.Items[i] + if pod.Status.PodIP == "" || !isPodReady(pod) { + return fmt.Errorf("proxy pod %s not ready (IP=%s, ready=%v), will retry", pod.Name, pod.Status.PodIP, isPodReady(pod)) + } + if err := r.Members.Client.IncludeProxyBackend(ctx, pod.Status.PodIP, proxyPort, backendIP); err != nil { + return fmt.Errorf("include on proxy pod %s: %w", pod.Name, err) + } + } + return nil +} + // ---------- Scale-down handling ---------- // runScaleDownClusterHooks drains pods that will be removed (highest ordinals first) and asks diff --git a/internal/hyperbytedb/client.go b/internal/hyperbytedb/client.go index 2de95b4..2a59101 100644 --- a/internal/hyperbytedb/client.go +++ b/internal/hyperbytedb/client.go @@ -400,3 +400,82 @@ func (c *Client) LeaveNode(ctx context.Context, host string, port int32, departe } return nil } + +// ---------- Proxy backend exclusion ---------- + +// ProxyBackendStatus mirrors the JSON returned by the proxy's GET /admin/pool. +type ProxyBackendStatus struct { + Addr string `json:"addr"` + Port int `json:"port"` + Health string `json:"health"` + Excluded bool `json:"excluded"` + Inflight int `json:"inflight"` + ConsecutiveFailures int `json:"consecutive_failures"` +} + +// excludeIncludeResponse is the envelope from POST /admin/backends/{ip}/exclude|include. +type excludeIncludeResponse struct { + Status string `json:"status"` + IP string `json:"ip"` +} + +// ExcludeProxyBackend tells the proxy to stop routing to the given backend IP. +func (c *Client) ExcludeProxyBackend(ctx context.Context, proxyHost string, port int32, ip string) error { + url := fmt.Sprintf("http://%s:%d/admin/backends/%s/exclude", proxyHost, port, ip) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil) + if err != nil { + return err + } + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("exclude backend %s: %w", ip, err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(resp.Body) + return fmt.Errorf("exclude backend %s returned %d: %s", ip, resp.StatusCode, string(respBody)) + } + return nil +} + +// IncludeProxyBackend tells the proxy to resume routing to the given backend IP. +func (c *Client) IncludeProxyBackend(ctx context.Context, proxyHost string, port int32, ip string) error { + url := fmt.Sprintf("http://%s:%d/admin/backends/%s/include", proxyHost, port, ip) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil) + if err != nil { + return err + } + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("include backend %s: %w", ip, err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(resp.Body) + return fmt.Errorf("include backend %s returned %d: %s", ip, resp.StatusCode, string(respBody)) + } + return nil +} + +// GetProxyPoolState retrieves the full pool status from the proxy's admin API. +func (c *Client) GetProxyPoolState(ctx context.Context, proxyHost string, port int32) ([]ProxyBackendStatus, error) { + url := fmt.Sprintf("http://%s:%d/admin/pool", proxyHost, port) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("get proxy pool state: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("proxy pool state returned %d: %s", resp.StatusCode, string(respBody)) + } + var pool []ProxyBackendStatus + if err := json.NewDecoder(resp.Body).Decode(&pool); err != nil { + return nil, fmt.Errorf("decoding proxy pool state: %w", err) + } + return pool, nil +} diff --git a/internal/hyperbytedb/configmap.go b/internal/hyperbytedb/configmap.go index 8c12317..f076a15 100644 --- a/internal/hyperbytedb/configmap.go +++ b/internal/hyperbytedb/configmap.go @@ -161,11 +161,19 @@ func writeChdbSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) sessionPath = spec.ChDB.SessionDataPath } fmt.Fprintf(b, "session_data_path = \"%s\"\n", sessionPath) - poolSize := int32(1) - if spec.ChDB.PoolSize > 0 { - poolSize = spec.ChDB.PoolSize + if spec.ChDB.QueryPoolSize > 0 { + fmt.Fprintf(b, "query_pool_size = %d\n", spec.ChDB.QueryPoolSize) + } + if spec.ChDB.WritePoolSize > 0 { + 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 + } + fmt.Fprintf(b, "pool_size = %d\n", poolSize) } - fmt.Fprintf(b, "pool_size = %d\n", poolSize) } func writeAuthSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) { diff --git a/internal/hyperbytedb/proxy.go b/internal/hyperbytedb/proxy.go index 3ba846b..869b747 100644 --- a/internal/hyperbytedb/proxy.go +++ b/internal/hyperbytedb/proxy.go @@ -26,6 +26,11 @@ func ProxyServiceName(cluster *v1alpha1.HyperbytedbCluster) string { return cluster.Name + "-proxy" } +// ProxyServiceAddr returns the in-cluster DNS name for the proxy Service. +func ProxyServiceAddr(cluster *v1alpha1.HyperbytedbCluster) string { + return fmt.Sprintf("%s.%s.svc.cluster.local", ProxyServiceName(cluster), cluster.Namespace) +} + // ProxyEnabled reports whether the operator should reconcile proxy resources. // The proxy is enabled by default; set spec.proxy.enabled=false to opt out. func ProxyEnabled(cluster *v1alpha1.HyperbytedbCluster) bool { From 17a045b20264f1c62a101911ab05d6cc35fddfe1 Mon Sep 17 00:00:00 2001 From: "austin.barrington" Date: Wed, 15 Jul 2026 19:27:46 +0100 Subject: [PATCH 2/4] fix ci to use github runners instead of self-hosted ones --- .github/actions/install-ci-deps/action.yml | 2 +- .github/actions/install-ci-deps/install-ci-deps.sh | 9 ++++----- .github/workflows/lint.yml | 2 +- .github/workflows/release.yml | 8 ++++---- .github/workflows/test-chart.yml | 2 +- .github/workflows/test-e2e.yml | 2 +- .github/workflows/test.yml | 2 +- 7 files changed, 13 insertions(+), 14 deletions(-) diff --git a/.github/actions/install-ci-deps/action.yml b/.github/actions/install-ci-deps/action.yml index 7c091a8..83340b0 100644 --- a/.github/actions/install-ci-deps/action.yml +++ b/.github/actions/install-ci-deps/action.yml @@ -1,5 +1,5 @@ name: Install CI dependencies -description: Install system packages and tools for ARC/Kubernetes GitHub Actions runners +description: Install system packages and tools for GitHub Actions runners inputs: profile: diff --git a/.github/actions/install-ci-deps/install-ci-deps.sh b/.github/actions/install-ci-deps/install-ci-deps.sh index 76c6975..80e7913 100755 --- a/.github/actions/install-ci-deps/install-ci-deps.sh +++ b/.github/actions/install-ci-deps/install-ci-deps.sh @@ -1,10 +1,9 @@ #!/usr/bin/env bash -# Install CI dependencies for GitHub Actions on ARC/Kubernetes runners. +# Install CI dependencies for GitHub Actions runners. # -# ARC runner pods use the actions-runner image, which is minimal and typically -# runs as a non-root user without sudo. Job-level `container:` directives are -# not supported unless the runner scale set is configured for container jobs, so -# workflows install what they need directly into the runner pod. +# GitHub-hosted ubuntu runners have sudo and apt-get. Self-hosted/ARC runner +# pods may be minimal and non-root; this script falls back to user-local +# installs when package managers are unavailable. set -euo pipefail PROFILE="${1:-k8s-runner}" diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index d139270..9790a3a 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -13,7 +13,7 @@ permissions: jobs: lint: name: Lint - runs-on: hyperbytedb-operator-controller + runs-on: ubuntu-latest steps: - name: Clone the code uses: actions/checkout@v4 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 8bcd00c..22cbc50 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -34,7 +34,7 @@ env: jobs: versions: name: Compute versions - runs-on: hyperbytedb-operator-controller + runs-on: ubuntu-latest outputs: chart_version: ${{ steps.compute.outputs.chart_version }} image_tag: ${{ steps.compute.outputs.image_tag }} @@ -59,7 +59,7 @@ jobs: image: name: Build & push operator image needs: versions - runs-on: hyperbytedb-operator-controller + runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -87,7 +87,7 @@ jobs: chart: name: Package & push helm chart needs: [versions, image] - runs-on: hyperbytedb-operator-controller + runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -141,7 +141,7 @@ jobs: name: GitHub Release needs: [versions, chart] if: github.ref_type == 'tag' - runs-on: hyperbytedb-operator-controller + runs-on: ubuntu-latest steps: - uses: actions/download-artifact@v4 with: diff --git a/.github/workflows/test-chart.yml b/.github/workflows/test-chart.yml index 13e55fc..20333a8 100644 --- a/.github/workflows/test-chart.yml +++ b/.github/workflows/test-chart.yml @@ -13,7 +13,7 @@ permissions: jobs: test-chart: name: Test Chart - runs-on: hyperbytedb-operator-controller + runs-on: ubuntu-latest timeout-minutes: 30 env: KIND_CLUSTER_NAME: hyperbytedb-chart-${{ github.run_id }}-${{ github.run_attempt }} diff --git a/.github/workflows/test-e2e.yml b/.github/workflows/test-e2e.yml index cc97d6f..a2f767a 100644 --- a/.github/workflows/test-e2e.yml +++ b/.github/workflows/test-e2e.yml @@ -13,7 +13,7 @@ permissions: jobs: test-e2e: name: E2E Tests - runs-on: hyperbytedb-operator-controller + runs-on: ubuntu-latest timeout-minutes: 30 env: KIND_CLUSTER: hyperbytedb-e2e-${{ github.run_id }}-${{ github.run_attempt }} diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index a8896d2..f61d553 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -13,7 +13,7 @@ permissions: jobs: test: name: Test - runs-on: hyperbytedb-operator-controller + runs-on: ubuntu-latest steps: - name: Clone the code uses: actions/checkout@v4 From aec7896e1706f6e8dab6ebf929408be53b8dcb30 Mon Sep 17 00:00:00 2001 From: "austin.barrington" Date: Wed, 15 Jul 2026 19:36:17 +0100 Subject: [PATCH 3/4] fix linting errors --- .../hyperbytedbcluster_controller.go | 336 ++++++++++-------- internal/hyperbytedb/client.go | 6 - 2 files changed, 188 insertions(+), 154 deletions(-) diff --git a/internal/controller/hyperbytedbcluster_controller.go b/internal/controller/hyperbytedbcluster_controller.go index 886374f..5a24a22 100644 --- a/internal/controller/hyperbytedbcluster_controller.go +++ b/internal/controller/hyperbytedbcluster_controller.go @@ -443,9 +443,6 @@ func (r *HyperbytedbClusterReconciler) reconcileRollingRestart( stsResult stsReconcileResult, replicas int32, ) (*ctrl.Result, error) { - log := logf.FromContext(ctx) - - // Check if a rolling upgrade is actually in progress. curSTS := &appsv1.StatefulSet{} if err := r.Get(ctx, types.NamespacedName{ Name: hyperbytedb.StatefulSetName(cluster), Namespace: cluster.Namespace, @@ -455,32 +452,72 @@ func (r *HyperbytedbClusterReconciler) reconcileRollingRestart( rollingUpgrade := curSTS.Status.UpdateRevision != "" && curSTS.Status.UpdateRevision != curSTS.Status.CurrentRevision - // No rolling upgrade and no active restart state → nothing to do. if !rollingUpgrade && cluster.Status.RollingRestart == nil { return nil, nil } - // If the upgrade finished (all replicas ready and up-to-date) AND the - // State machine reached Completed — clear coordination state once all - // pods are ready and up-to-date. - if cluster.Status.RollingRestart != nil && - cluster.Status.RollingRestart.Phase == hyperbytedbv1alpha1.RollingRestartCompleted { - if stsResult.ReadyReplicas == stsResult.SpecReplicas && - curSTS.Status.ReadyReplicas == curSTS.Status.Replicas && - curSTS.Status.UpdatedReplicas == curSTS.Status.Replicas { - cluster.Status.RollingRestart = nil - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - log.Info("Rolling restart complete, cleared coordination state") - return nil, nil - } + if cleared, err := r.tryClearCompletedRollingRestart(ctx, cluster, stsResult, curSTS); err != nil { + return nil, err + } else if cleared { + return nil, nil } proxyPort := hyperbytedb.ServerPort(cluster) stsName := hyperbytedb.StatefulSetName(cluster) - // Initialise state if this is the start of a new rolling upgrade. + state, err := r.ensureRollingRestartState(ctx, cluster, replicas) + if err != nil { + return nil, err + } + + podName := fmt.Sprintf("%s-%d", stsName, state.CurrentOrdinal) + requeue3s := ctrl.Result{RequeueAfter: 3 * time.Second} + + switch state.Phase { + case hyperbytedbv1alpha1.RollingRestartExcluding: + return r.rollingRestartExcluding(ctx, cluster, state, podName, proxyPort, requeue3s) + case hyperbytedbv1alpha1.RollingRestartDraining: + return r.rollingRestartDraining(ctx, cluster, state, podName, requeue3s) + case hyperbytedbv1alpha1.RollingRestartWaitingReady: + return r.rollingRestartWaitingReady(ctx, cluster, state, podName, proxyPort, requeue3s) + case hyperbytedbv1alpha1.RollingRestartCompleted: + return &requeue3s, nil + default: + return nil, nil + } +} + +func (r *HyperbytedbClusterReconciler) tryClearCompletedRollingRestart( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + stsResult stsReconcileResult, + curSTS *appsv1.StatefulSet, +) (bool, error) { + log := logf.FromContext(ctx) + state := cluster.Status.RollingRestart + if state == nil || state.Phase != hyperbytedbv1alpha1.RollingRestartCompleted { + return false, nil + } + if stsResult.ReadyReplicas != stsResult.SpecReplicas || + curSTS.Status.ReadyReplicas != curSTS.Status.Replicas || + curSTS.Status.UpdatedReplicas != curSTS.Status.Replicas { + return false, nil + } + + cluster.Status.RollingRestart = nil + if err := r.Status().Update(ctx, cluster); err != nil { + return false, err + } + log.Info("Rolling restart complete, cleared coordination state") + return true, nil +} + +func (r *HyperbytedbClusterReconciler) ensureRollingRestartState( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + replicas int32, +) (*hyperbytedbv1alpha1.RollingRestartState, error) { + log := logf.FromContext(ctx) if cluster.Status.RollingRestart == nil { cluster.Status.RollingRestart = &hyperbytedbv1alpha1.RollingRestartState{ CurrentOrdinal: 0, @@ -496,8 +533,6 @@ func (r *HyperbytedbClusterReconciler) reconcileRollingRestart( state := cluster.Status.RollingRestart if state == nil { - // CRD schema didn't persist the field (stale CRD or server-side pruning). - // Re-initialize so the state machine can proceed. state = &hyperbytedbv1alpha1.RollingRestartState{ CurrentOrdinal: 0, TotalOrdinals: replicas, @@ -509,155 +544,160 @@ func (r *HyperbytedbClusterReconciler) reconcileRollingRestart( return nil, err } } - podName := fmt.Sprintf("%s-%d", stsName, state.CurrentOrdinal) - - requeue3s := ctrl.Result{RequeueAfter: 3 * time.Second} + return state, nil +} - switch state.Phase { - case hyperbytedbv1alpha1.RollingRestartExcluding: - // Get the pod's IP. - pod := &corev1.Pod{} - if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { - if apierrors.IsNotFound(err) { - // Pod not yet created; skip to WaitingReady (StatefulSet is still rolling). - state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - return &requeue3s, nil +func (r *HyperbytedbClusterReconciler) rollingRestartExcluding( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + state *hyperbytedbv1alpha1.RollingRestartState, + podName string, + proxyPort int32, + requeue ctrl.Result, +) (*ctrl.Result, error) { + log := logf.FromContext(ctx) + pod := &corev1.Pod{} + if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { + if apierrors.IsNotFound(err) { + state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err } - return nil, err + return &requeue, nil } + return nil, err + } - podIP := pod.Status.PodIP - if podIP == "" { - log.Info("Pod has no IP yet, waiting", "pod", podName) - return &requeue3s, nil - } - state.OldPodIP = podIP + podIP := pod.Status.PodIP + if podIP == "" { + log.Info("Pod has no IP yet, waiting", "pod", podName) + return &requeue, nil + } + state.OldPodIP = podIP - // Call proxy exclude on ALL proxy pods (not just one via Service round-robin). - if err := r.excludeFromAllProxies(ctx, cluster, podIP, proxyPort); err != nil { - log.V(1).Info("Could not exclude backend from proxy, will retry", "pod", podName, "ip", podIP, "error", err) - return &requeue3s, nil - } - log.Info("Excluded backend from proxy", "pod", podName, "ip", podIP) - r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendExcluded", - "Excluded pod %s (%s) from proxy routing", podName, podIP) - state.ExcludeConfirmed = true - state.Phase = hyperbytedbv1alpha1.RollingRestartDraining - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - return &requeue3s, nil + if err := r.excludeFromAllProxies(ctx, cluster, podIP, proxyPort); err != nil { + log.V(1).Info("Could not exclude backend from proxy, will retry", "pod", podName, "ip", podIP, "error", err) + return &requeue, nil + } + log.Info("Excluded backend from proxy", "pod", podName, "ip", podIP) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendExcluded", + "Excluded pod %s (%s) from proxy routing", podName, podIP) + state.ExcludeConfirmed = true + state.Phase = hyperbytedbv1alpha1.RollingRestartDraining + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + return &requeue, nil +} - case hyperbytedbv1alpha1.RollingRestartDraining: - // Wait for drain time. - drainWait := time.Duration(cluster.Spec.Cluster.DrainWaitSecs) * time.Second - if drainWait <= 0 { - drainWait = 10 * time.Second - } - elapsed := time.Since(state.PhaseStartedAt.Time) - if elapsed < drainWait { - log.Info("Waiting for in-flight requests to drain", - "pod", podName, "elapsed", elapsed.Truncate(time.Second), "wait", drainWait) - return &ctrl.Result{RequeueAfter: drainWait - elapsed + time.Second}, nil - } +func (r *HyperbytedbClusterReconciler) rollingRestartDraining( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + state *hyperbytedbv1alpha1.RollingRestartState, + podName string, + requeue ctrl.Result, +) (*ctrl.Result, error) { + log := logf.FromContext(ctx) + drainWait := time.Duration(cluster.Spec.Cluster.DrainWaitSecs) * time.Second + if drainWait <= 0 { + drainWait = 10 * time.Second + } + elapsed := time.Since(state.PhaseStartedAt.Time) + if elapsed < drainWait { + log.Info("Waiting for in-flight requests to drain", + "pod", podName, "elapsed", elapsed.Truncate(time.Second), "wait", drainWait) + return &ctrl.Result{RequeueAfter: drainWait - elapsed + time.Second}, nil + } - // Delete the pod to trigger recreation from the updated StatefulSet template. - pod := &corev1.Pod{} - if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { - if apierrors.IsNotFound(err) { - // Already gone, move to WaitingReady. - state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - return &requeue3s, nil + pod := &corev1.Pod{} + if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { + if apierrors.IsNotFound(err) { + state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err } - return nil, err + return &requeue, nil } + return nil, err + } - if err := r.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) { - return nil, fmt.Errorf("deleting pod %s for rolling restart: %w", podName, err) - } - log.Info("Deleted pod for rolling restart", "pod", podName) - r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "PodRestarted", - "Deleted pod %s for rolling upgrade", podName) - state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - return &requeue3s, nil + if err := r.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) { + return nil, fmt.Errorf("deleting pod %s for rolling restart: %w", podName, err) + } + log.Info("Deleted pod for rolling restart", "pod", podName) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "PodRestarted", + "Deleted pod %s for rolling upgrade", podName) + state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + return &requeue, nil +} - case hyperbytedbv1alpha1.RollingRestartWaitingReady: - // Wait for the new pod to be Ready. - pod := &corev1.Pod{} - if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { - if apierrors.IsNotFound(err) { - log.Info("Waiting for replacement pod to appear", "pod", podName) - return &requeue3s, nil - } - return nil, err +func (r *HyperbytedbClusterReconciler) rollingRestartWaitingReady( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + state *hyperbytedbv1alpha1.RollingRestartState, + podName string, + proxyPort int32, + requeue ctrl.Result, +) (*ctrl.Result, error) { + log := logf.FromContext(ctx) + pod := &corev1.Pod{} + if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { + if apierrors.IsNotFound(err) { + log.Info("Waiting for replacement pod to appear", "pod", podName) + return &requeue, nil } + return nil, err + } - if !isPodReady(pod) || pod.DeletionTimestamp != nil { - if pod.DeletionTimestamp != nil { - log.Info("Old pod still terminating, waiting for replacement", "pod", podName) - } else { - log.Info("Replacement pod not ready yet", "pod", podName) - } - return &requeue3s, nil + if !isPodReady(pod) || pod.DeletionTimestamp != nil { + if pod.DeletionTimestamp != nil { + log.Info("Old pod still terminating, waiting for replacement", "pod", podName) + } else { + log.Info("Replacement pod not ready yet", "pod", podName) } + return &requeue, nil + } - // Pod is ready and not terminating — get its IP and include it. - newIP := pod.Status.PodIP - if newIP == "" { - log.Info("Pod is ready but has no IP, waiting", "pod", podName) - return &requeue3s, nil - } + newIP := pod.Status.PodIP + if newIP == "" { + log.Info("Pod is ready but has no IP, waiting", "pod", podName) + return &requeue, nil + } - if err := r.includeFromAllProxies(ctx, cluster, newIP, proxyPort); err != nil { - log.V(1).Info("Could not include backend in proxy, will retry", "pod", podName, "ip", newIP, "error", err) - return &requeue3s, nil - } - log.Info("Included backend in proxy", "pod", podName, "ip", newIP) - r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendIncluded", - "Included pod %s (%s) in proxy routing", podName, newIP) - - // Move to next ordinal or finish. - state.CurrentOrdinal++ - if state.CurrentOrdinal >= state.TotalOrdinals { - // All pods have been cycled. Set Completed phase and let the - // "upgrade finished" check clear the state once the STS settles. - state.Phase = hyperbytedbv1alpha1.RollingRestartCompleted - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - log.Info("Rolling restart coordination complete") - return &ctrl.Result{RequeueAfter: 3 * time.Second}, nil - } - state.Phase = hyperbytedbv1alpha1.RollingRestartExcluding + if err := r.includeFromAllProxies(ctx, cluster, newIP, proxyPort); err != nil { + log.V(1).Info("Could not include backend in proxy, will retry", "pod", podName, "ip", newIP, "error", err) + return &requeue, nil + } + log.Info("Included backend in proxy", "pod", podName, "ip", newIP) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendIncluded", + "Included pod %s (%s) in proxy routing", podName, newIP) + + state.CurrentOrdinal++ + if state.CurrentOrdinal >= state.TotalOrdinals { + state.Phase = hyperbytedbv1alpha1.RollingRestartCompleted state.PhaseStartedAt = metav1.Now() - state.ExcludeConfirmed = false - state.OldPodIP = "" if err := r.Status().Update(ctx, cluster); err != nil { return nil, err } - return &requeue3s, nil + log.Info("Rolling restart coordination complete") + return &ctrl.Result{RequeueAfter: 3 * time.Second}, nil } - // Coordination complete but STS hasn't fully settled yet — just wait. - if state.Phase == hyperbytedbv1alpha1.RollingRestartCompleted { - return &requeue3s, nil + state.Phase = hyperbytedbv1alpha1.RollingRestartExcluding + state.PhaseStartedAt = metav1.Now() + state.ExcludeConfirmed = false + state.OldPodIP = "" + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err } - - return nil, nil + return &requeue, nil } // isPodReady returns true if the pod has a Ready condition with Status=True. diff --git a/internal/hyperbytedb/client.go b/internal/hyperbytedb/client.go index 2a59101..e0c8dd0 100644 --- a/internal/hyperbytedb/client.go +++ b/internal/hyperbytedb/client.go @@ -413,12 +413,6 @@ type ProxyBackendStatus struct { ConsecutiveFailures int `json:"consecutive_failures"` } -// excludeIncludeResponse is the envelope from POST /admin/backends/{ip}/exclude|include. -type excludeIncludeResponse struct { - Status string `json:"status"` - IP string `json:"ip"` -} - // ExcludeProxyBackend tells the proxy to stop routing to the given backend IP. func (c *Client) ExcludeProxyBackend(ctx context.Context, proxyHost string, port int32, ip string) error { url := fmt.Sprintf("http://%s:%d/admin/backends/%s/exclude", proxyHost, port, ip) From 25bcf8c760353bfbf3e4fbe1a8536f026369b356 Mon Sep 17 00:00:00 2001 From: "austin.barrington" Date: Wed, 15 Jul 2026 19:44:20 +0100 Subject: [PATCH 4/4] fix linting again --- .../hyperbytedbcluster_controller.go | 336 ++++++++++-------- internal/hyperbytedb/client.go | 6 - 2 files changed, 188 insertions(+), 154 deletions(-) diff --git a/internal/controller/hyperbytedbcluster_controller.go b/internal/controller/hyperbytedbcluster_controller.go index 886374f..5a24a22 100644 --- a/internal/controller/hyperbytedbcluster_controller.go +++ b/internal/controller/hyperbytedbcluster_controller.go @@ -443,9 +443,6 @@ func (r *HyperbytedbClusterReconciler) reconcileRollingRestart( stsResult stsReconcileResult, replicas int32, ) (*ctrl.Result, error) { - log := logf.FromContext(ctx) - - // Check if a rolling upgrade is actually in progress. curSTS := &appsv1.StatefulSet{} if err := r.Get(ctx, types.NamespacedName{ Name: hyperbytedb.StatefulSetName(cluster), Namespace: cluster.Namespace, @@ -455,32 +452,72 @@ func (r *HyperbytedbClusterReconciler) reconcileRollingRestart( rollingUpgrade := curSTS.Status.UpdateRevision != "" && curSTS.Status.UpdateRevision != curSTS.Status.CurrentRevision - // No rolling upgrade and no active restart state → nothing to do. if !rollingUpgrade && cluster.Status.RollingRestart == nil { return nil, nil } - // If the upgrade finished (all replicas ready and up-to-date) AND the - // State machine reached Completed — clear coordination state once all - // pods are ready and up-to-date. - if cluster.Status.RollingRestart != nil && - cluster.Status.RollingRestart.Phase == hyperbytedbv1alpha1.RollingRestartCompleted { - if stsResult.ReadyReplicas == stsResult.SpecReplicas && - curSTS.Status.ReadyReplicas == curSTS.Status.Replicas && - curSTS.Status.UpdatedReplicas == curSTS.Status.Replicas { - cluster.Status.RollingRestart = nil - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - log.Info("Rolling restart complete, cleared coordination state") - return nil, nil - } + if cleared, err := r.tryClearCompletedRollingRestart(ctx, cluster, stsResult, curSTS); err != nil { + return nil, err + } else if cleared { + return nil, nil } proxyPort := hyperbytedb.ServerPort(cluster) stsName := hyperbytedb.StatefulSetName(cluster) - // Initialise state if this is the start of a new rolling upgrade. + state, err := r.ensureRollingRestartState(ctx, cluster, replicas) + if err != nil { + return nil, err + } + + podName := fmt.Sprintf("%s-%d", stsName, state.CurrentOrdinal) + requeue3s := ctrl.Result{RequeueAfter: 3 * time.Second} + + switch state.Phase { + case hyperbytedbv1alpha1.RollingRestartExcluding: + return r.rollingRestartExcluding(ctx, cluster, state, podName, proxyPort, requeue3s) + case hyperbytedbv1alpha1.RollingRestartDraining: + return r.rollingRestartDraining(ctx, cluster, state, podName, requeue3s) + case hyperbytedbv1alpha1.RollingRestartWaitingReady: + return r.rollingRestartWaitingReady(ctx, cluster, state, podName, proxyPort, requeue3s) + case hyperbytedbv1alpha1.RollingRestartCompleted: + return &requeue3s, nil + default: + return nil, nil + } +} + +func (r *HyperbytedbClusterReconciler) tryClearCompletedRollingRestart( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + stsResult stsReconcileResult, + curSTS *appsv1.StatefulSet, +) (bool, error) { + log := logf.FromContext(ctx) + state := cluster.Status.RollingRestart + if state == nil || state.Phase != hyperbytedbv1alpha1.RollingRestartCompleted { + return false, nil + } + if stsResult.ReadyReplicas != stsResult.SpecReplicas || + curSTS.Status.ReadyReplicas != curSTS.Status.Replicas || + curSTS.Status.UpdatedReplicas != curSTS.Status.Replicas { + return false, nil + } + + cluster.Status.RollingRestart = nil + if err := r.Status().Update(ctx, cluster); err != nil { + return false, err + } + log.Info("Rolling restart complete, cleared coordination state") + return true, nil +} + +func (r *HyperbytedbClusterReconciler) ensureRollingRestartState( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + replicas int32, +) (*hyperbytedbv1alpha1.RollingRestartState, error) { + log := logf.FromContext(ctx) if cluster.Status.RollingRestart == nil { cluster.Status.RollingRestart = &hyperbytedbv1alpha1.RollingRestartState{ CurrentOrdinal: 0, @@ -496,8 +533,6 @@ func (r *HyperbytedbClusterReconciler) reconcileRollingRestart( state := cluster.Status.RollingRestart if state == nil { - // CRD schema didn't persist the field (stale CRD or server-side pruning). - // Re-initialize so the state machine can proceed. state = &hyperbytedbv1alpha1.RollingRestartState{ CurrentOrdinal: 0, TotalOrdinals: replicas, @@ -509,155 +544,160 @@ func (r *HyperbytedbClusterReconciler) reconcileRollingRestart( return nil, err } } - podName := fmt.Sprintf("%s-%d", stsName, state.CurrentOrdinal) - - requeue3s := ctrl.Result{RequeueAfter: 3 * time.Second} + return state, nil +} - switch state.Phase { - case hyperbytedbv1alpha1.RollingRestartExcluding: - // Get the pod's IP. - pod := &corev1.Pod{} - if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { - if apierrors.IsNotFound(err) { - // Pod not yet created; skip to WaitingReady (StatefulSet is still rolling). - state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - return &requeue3s, nil +func (r *HyperbytedbClusterReconciler) rollingRestartExcluding( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + state *hyperbytedbv1alpha1.RollingRestartState, + podName string, + proxyPort int32, + requeue ctrl.Result, +) (*ctrl.Result, error) { + log := logf.FromContext(ctx) + pod := &corev1.Pod{} + if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { + if apierrors.IsNotFound(err) { + state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err } - return nil, err + return &requeue, nil } + return nil, err + } - podIP := pod.Status.PodIP - if podIP == "" { - log.Info("Pod has no IP yet, waiting", "pod", podName) - return &requeue3s, nil - } - state.OldPodIP = podIP + podIP := pod.Status.PodIP + if podIP == "" { + log.Info("Pod has no IP yet, waiting", "pod", podName) + return &requeue, nil + } + state.OldPodIP = podIP - // Call proxy exclude on ALL proxy pods (not just one via Service round-robin). - if err := r.excludeFromAllProxies(ctx, cluster, podIP, proxyPort); err != nil { - log.V(1).Info("Could not exclude backend from proxy, will retry", "pod", podName, "ip", podIP, "error", err) - return &requeue3s, nil - } - log.Info("Excluded backend from proxy", "pod", podName, "ip", podIP) - r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendExcluded", - "Excluded pod %s (%s) from proxy routing", podName, podIP) - state.ExcludeConfirmed = true - state.Phase = hyperbytedbv1alpha1.RollingRestartDraining - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - return &requeue3s, nil + if err := r.excludeFromAllProxies(ctx, cluster, podIP, proxyPort); err != nil { + log.V(1).Info("Could not exclude backend from proxy, will retry", "pod", podName, "ip", podIP, "error", err) + return &requeue, nil + } + log.Info("Excluded backend from proxy", "pod", podName, "ip", podIP) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendExcluded", + "Excluded pod %s (%s) from proxy routing", podName, podIP) + state.ExcludeConfirmed = true + state.Phase = hyperbytedbv1alpha1.RollingRestartDraining + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + return &requeue, nil +} - case hyperbytedbv1alpha1.RollingRestartDraining: - // Wait for drain time. - drainWait := time.Duration(cluster.Spec.Cluster.DrainWaitSecs) * time.Second - if drainWait <= 0 { - drainWait = 10 * time.Second - } - elapsed := time.Since(state.PhaseStartedAt.Time) - if elapsed < drainWait { - log.Info("Waiting for in-flight requests to drain", - "pod", podName, "elapsed", elapsed.Truncate(time.Second), "wait", drainWait) - return &ctrl.Result{RequeueAfter: drainWait - elapsed + time.Second}, nil - } +func (r *HyperbytedbClusterReconciler) rollingRestartDraining( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + state *hyperbytedbv1alpha1.RollingRestartState, + podName string, + requeue ctrl.Result, +) (*ctrl.Result, error) { + log := logf.FromContext(ctx) + drainWait := time.Duration(cluster.Spec.Cluster.DrainWaitSecs) * time.Second + if drainWait <= 0 { + drainWait = 10 * time.Second + } + elapsed := time.Since(state.PhaseStartedAt.Time) + if elapsed < drainWait { + log.Info("Waiting for in-flight requests to drain", + "pod", podName, "elapsed", elapsed.Truncate(time.Second), "wait", drainWait) + return &ctrl.Result{RequeueAfter: drainWait - elapsed + time.Second}, nil + } - // Delete the pod to trigger recreation from the updated StatefulSet template. - pod := &corev1.Pod{} - if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { - if apierrors.IsNotFound(err) { - // Already gone, move to WaitingReady. - state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - return &requeue3s, nil + pod := &corev1.Pod{} + if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { + if apierrors.IsNotFound(err) { + state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err } - return nil, err + return &requeue, nil } + return nil, err + } - if err := r.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) { - return nil, fmt.Errorf("deleting pod %s for rolling restart: %w", podName, err) - } - log.Info("Deleted pod for rolling restart", "pod", podName) - r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "PodRestarted", - "Deleted pod %s for rolling upgrade", podName) - state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - return &requeue3s, nil + if err := r.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) { + return nil, fmt.Errorf("deleting pod %s for rolling restart: %w", podName, err) + } + log.Info("Deleted pod for rolling restart", "pod", podName) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "PodRestarted", + "Deleted pod %s for rolling upgrade", podName) + state.Phase = hyperbytedbv1alpha1.RollingRestartWaitingReady + state.PhaseStartedAt = metav1.Now() + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err + } + return &requeue, nil +} - case hyperbytedbv1alpha1.RollingRestartWaitingReady: - // Wait for the new pod to be Ready. - pod := &corev1.Pod{} - if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { - if apierrors.IsNotFound(err) { - log.Info("Waiting for replacement pod to appear", "pod", podName) - return &requeue3s, nil - } - return nil, err +func (r *HyperbytedbClusterReconciler) rollingRestartWaitingReady( + ctx context.Context, + cluster *hyperbytedbv1alpha1.HyperbytedbCluster, + state *hyperbytedbv1alpha1.RollingRestartState, + podName string, + proxyPort int32, + requeue ctrl.Result, +) (*ctrl.Result, error) { + log := logf.FromContext(ctx) + pod := &corev1.Pod{} + if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: cluster.Namespace}, pod); err != nil { + if apierrors.IsNotFound(err) { + log.Info("Waiting for replacement pod to appear", "pod", podName) + return &requeue, nil } + return nil, err + } - if !isPodReady(pod) || pod.DeletionTimestamp != nil { - if pod.DeletionTimestamp != nil { - log.Info("Old pod still terminating, waiting for replacement", "pod", podName) - } else { - log.Info("Replacement pod not ready yet", "pod", podName) - } - return &requeue3s, nil + if !isPodReady(pod) || pod.DeletionTimestamp != nil { + if pod.DeletionTimestamp != nil { + log.Info("Old pod still terminating, waiting for replacement", "pod", podName) + } else { + log.Info("Replacement pod not ready yet", "pod", podName) } + return &requeue, nil + } - // Pod is ready and not terminating — get its IP and include it. - newIP := pod.Status.PodIP - if newIP == "" { - log.Info("Pod is ready but has no IP, waiting", "pod", podName) - return &requeue3s, nil - } + newIP := pod.Status.PodIP + if newIP == "" { + log.Info("Pod is ready but has no IP, waiting", "pod", podName) + return &requeue, nil + } - if err := r.includeFromAllProxies(ctx, cluster, newIP, proxyPort); err != nil { - log.V(1).Info("Could not include backend in proxy, will retry", "pod", podName, "ip", newIP, "error", err) - return &requeue3s, nil - } - log.Info("Included backend in proxy", "pod", podName, "ip", newIP) - r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendIncluded", - "Included pod %s (%s) in proxy routing", podName, newIP) - - // Move to next ordinal or finish. - state.CurrentOrdinal++ - if state.CurrentOrdinal >= state.TotalOrdinals { - // All pods have been cycled. Set Completed phase and let the - // "upgrade finished" check clear the state once the STS settles. - state.Phase = hyperbytedbv1alpha1.RollingRestartCompleted - state.PhaseStartedAt = metav1.Now() - if err := r.Status().Update(ctx, cluster); err != nil { - return nil, err - } - log.Info("Rolling restart coordination complete") - return &ctrl.Result{RequeueAfter: 3 * time.Second}, nil - } - state.Phase = hyperbytedbv1alpha1.RollingRestartExcluding + if err := r.includeFromAllProxies(ctx, cluster, newIP, proxyPort); err != nil { + log.V(1).Info("Could not include backend in proxy, will retry", "pod", podName, "ip", newIP, "error", err) + return &requeue, nil + } + log.Info("Included backend in proxy", "pod", podName, "ip", newIP) + r.Recorder.Eventf(cluster, corev1.EventTypeNormal, "BackendIncluded", + "Included pod %s (%s) in proxy routing", podName, newIP) + + state.CurrentOrdinal++ + if state.CurrentOrdinal >= state.TotalOrdinals { + state.Phase = hyperbytedbv1alpha1.RollingRestartCompleted state.PhaseStartedAt = metav1.Now() - state.ExcludeConfirmed = false - state.OldPodIP = "" if err := r.Status().Update(ctx, cluster); err != nil { return nil, err } - return &requeue3s, nil + log.Info("Rolling restart coordination complete") + return &ctrl.Result{RequeueAfter: 3 * time.Second}, nil } - // Coordination complete but STS hasn't fully settled yet — just wait. - if state.Phase == hyperbytedbv1alpha1.RollingRestartCompleted { - return &requeue3s, nil + state.Phase = hyperbytedbv1alpha1.RollingRestartExcluding + state.PhaseStartedAt = metav1.Now() + state.ExcludeConfirmed = false + state.OldPodIP = "" + if err := r.Status().Update(ctx, cluster); err != nil { + return nil, err } - - return nil, nil + return &requeue, nil } // isPodReady returns true if the pod has a Ready condition with Status=True. diff --git a/internal/hyperbytedb/client.go b/internal/hyperbytedb/client.go index 2a59101..e0c8dd0 100644 --- a/internal/hyperbytedb/client.go +++ b/internal/hyperbytedb/client.go @@ -413,12 +413,6 @@ type ProxyBackendStatus struct { ConsecutiveFailures int `json:"consecutive_failures"` } -// excludeIncludeResponse is the envelope from POST /admin/backends/{ip}/exclude|include. -type excludeIncludeResponse struct { - Status string `json:"status"` - IP string `json:"ip"` -} - // ExcludeProxyBackend tells the proxy to stop routing to the given backend IP. func (c *Client) ExcludeProxyBackend(ctx context.Context, proxyHost string, port int32, ip string) error { url := fmt.Sprintf("http://%s:%d/admin/backends/%s/exclude", proxyHost, port, ip)