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 {