From a11289ee69f25a8915f3a7ff2f64330ad7e5fc73 Mon Sep 17 00:00:00 2001 From: "austin.barrington" Date: Wed, 24 Jun 2026 10:47:58 +0100 Subject: [PATCH] feat: Update operator to support Hyperbytedb 0.8.3 --- api/v1alpha1/hyperbytedbcluster_types.go | 35 ++++++---- api/v1alpha1/hyperbytedbcluster_webhook.go | 10 +-- api/v1alpha1/zz_generated.deepcopy.go | 14 ++-- ...db.hyperbytedb.io_hyperbytedbclusters.yaml | 36 ++++++---- ...bytedb_v1alpha1_hyperbytedbcluster_ha.yaml | 2 - ...dbclusters.hyperbytedb.hyperbytedb.io.yaml | 14 +++- .../hyperbytedbbackup_controller.go | 5 +- .../hyperbytedbcluster_controller.go | 30 ++++++-- .../hyperbytedbrestore_controller.go | 5 +- internal/hyperbytedb/client.go | 4 +- internal/hyperbytedb/configmap.go | 26 +++---- internal/hyperbytedb/configmap_test.go | 4 +- internal/hyperbytedb/image.go | 65 +++++++++++++++++ internal/hyperbytedb/image_test.go | 70 +++++++++++++++++++ internal/hyperbytedb/proxy.go | 15 ++-- internal/hyperbytedb/statefulset.go | 61 +++++++++++++--- 16 files changed, 304 insertions(+), 92 deletions(-) create mode 100644 internal/hyperbytedb/image.go create mode 100644 internal/hyperbytedb/image_test.go diff --git a/api/v1alpha1/hyperbytedbcluster_types.go b/api/v1alpha1/hyperbytedbcluster_types.go index 66b5b1e..c74fa35 100644 --- a/api/v1alpha1/hyperbytedbcluster_types.go +++ b/api/v1alpha1/hyperbytedbcluster_types.go @@ -29,9 +29,15 @@ type HyperbytedbClusterSpec struct { // +kubebuilder:default=1 Replicas *int32 `json:"replicas,omitempty"` - // +kubebuilder:default="hyperbytedb:latest" + // Container image reference. When set with a tag (contains ':'), used as-is. + // When set without a tag, treated as the repository and combined with Version. + // When empty, defaults to hyperbytedb:{Version} or hyperbytedb:latest. + // +optional Image string `json:"image,omitempty"` + // Application version. Drives the container image tag for hyperbytedb and + // hyperbytedb-proxy (e.g. version "0.8.3" → hyperbytedb:v0.8.3). Changing + // this field triggers a rolling upgrade. // +optional Version string `json:"version,omitempty"` @@ -169,6 +175,12 @@ type CertManagerIssuerRef struct { type StorageSpec struct { // +optional VolumeClaimTemplate *PersistentVolumeClaimSpec `json:"volumeClaimTemplate,omitempty"` + + // WAL encoding format: "bincode" (default) or "arrow_ipc". + // +optional + // +kubebuilder:validation:Enum=bincode;arrow_ipc + // +kubebuilder:default="bincode" + WALFormat string `json:"walFormat,omitempty"` } type PersistentVolumeClaimSpec struct { @@ -204,6 +216,11 @@ type FlushSpec struct { // +optional // +kubebuilder:validation:Minimum=0 WALBatchDelayUs int64 `json:"walBatchDelayUs,omitempty"` + + // Keep chDB-ready Arrow batches in an in-memory WAL cache for zero-copy flush. + // +optional + // +kubebuilder:default=true + ArrowWALEnabled *bool `json:"arrowWALEnabled,omitempty"` } type ChDBSpec struct { @@ -235,18 +252,6 @@ type LoggingSpec struct { // +kubebuilder:default="text" // +kubebuilder:validation:Enum=text;json Format string `json:"format,omitempty"` - - // Emit per-phase performance logs (write/query/flush). Off by default. - // +optional - DetailedTrace *bool `json:"detailedTrace,omitempty"` - - // OTLP HTTP endpoint for trace export (e.g. http://alloy-logs:4318). - // +optional - OtlpEndpoint string `json:"otlpEndpoint,omitempty"` - - // Fraction of traces exported to OTLP (0.0–1.0). Default 1.0 when OTLP is set. - // +optional - OtlpSampleRatio string `json:"otlpSampleRatio,omitempty"` } type ClusterTuningSpec struct { @@ -447,7 +452,7 @@ type ProxySpec struct { // When false, the operator does not create or reconcile any proxy // resources. Existing proxy Deployment/Service (if any) are left alone // so they can be cleaned up out-of-band. - // +kubebuilder:default=false + // +kubebuilder:default=true Enabled bool `json:"enabled"` // +kubebuilder:default="hyperbytedb-proxy:latest" @@ -479,7 +484,7 @@ type ProxySpec struct { // request with 503. Bigger values mean rolling restarts are smoother but // individual stuck requests sit longer. // +optional - // +kubebuilder:default=10 + // +kubebuilder:default=30 // +kubebuilder:validation:Minimum=0 HoldTimeoutSecs int32 `json:"holdTimeoutSecs,omitempty"` diff --git a/api/v1alpha1/hyperbytedbcluster_webhook.go b/api/v1alpha1/hyperbytedbcluster_webhook.go index 067d3f1..d29cd38 100644 --- a/api/v1alpha1/hyperbytedbcluster_webhook.go +++ b/api/v1alpha1/hyperbytedbcluster_webhook.go @@ -43,10 +43,6 @@ func (w *HyperbytedbClusterWebhook) Default(_ context.Context, obj *HyperbytedbC obj.Spec.Replicas = ptr.To(int32(1)) } - if obj.Spec.Image == "" { - obj.Spec.Image = "hyperbytedb:latest" - } - if obj.Spec.Server.Port == 0 { obj.Spec.Server.Port = 8086 } @@ -84,6 +80,12 @@ func (w *HyperbytedbClusterWebhook) Default(_ context.Context, obj *HyperbytedbC obj.Spec.Retention.Interval = "60s" } + // hyperbytedb-proxy is enabled by default for health-aware routing during + // rolling upgrades. Opt out with spec.proxy.enabled=false. + if obj.Spec.Proxy == nil { + obj.Spec.Proxy = &ProxySpec{Enabled: true} + } + return nil } diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 222f46d..d28862f 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -161,6 +161,11 @@ func (in *FailoverSpec) DeepCopy() *FailoverSpec { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *FlushSpec) DeepCopyInto(out *FlushSpec) { *out = *in + if in.ArrowWALEnabled != nil { + in, out := &in.ArrowWALEnabled, &out.ArrowWALEnabled + *out = new(bool) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FlushSpec. @@ -381,10 +386,10 @@ func (in *HyperbytedbClusterSpec) DeepCopyInto(out *HyperbytedbClusterSpec) { } in.Server.DeepCopyInto(&out.Server) in.Storage.DeepCopyInto(&out.Storage) - out.Flush = in.Flush + in.Flush.DeepCopyInto(&out.Flush) out.ChDB = in.ChDB out.Auth = in.Auth - in.Logging.DeepCopyInto(&out.Logging) + out.Logging = in.Logging in.Resources.DeepCopyInto(&out.Resources) in.Cluster.DeepCopyInto(&out.Cluster) out.Cardinality = in.Cardinality @@ -627,11 +632,6 @@ func (in *HyperbytedbRestoreStatus) DeepCopy() *HyperbytedbRestoreStatus { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *LoggingSpec) DeepCopyInto(out *LoggingSpec) { *out = *in - if in.DetailedTrace != nil { - in, out := &in.DetailedTrace, &out.DetailedTrace - *out = new(bool) - **out = **in - } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new LoggingSpec. diff --git a/config/crd/bases/hyperbytedb.hyperbytedb.io_hyperbytedbclusters.yaml b/config/crd/bases/hyperbytedb.hyperbytedb.io_hyperbytedbclusters.yaml index 507e184..86aab52 100644 --- a/config/crd/bases/hyperbytedb.hyperbytedb.io_hyperbytedbclusters.yaml +++ b/config/crd/bases/hyperbytedb.hyperbytedb.io_hyperbytedbclusters.yaml @@ -3137,6 +3137,11 @@ spec: type: object flush: properties: + arrowWALEnabled: + default: true + description: Keep chDB-ready Arrow batches in an in-memory WAL + cache for zero-copy flush. + type: boolean intervalSecs: default: 10 format: int32 @@ -3189,7 +3194,10 @@ spec: type: integer type: object image: - default: hyperbytedb:latest + description: |- + Container image reference. When set with a tag (contains ':'), used as-is. + When set without a tag, treated as the repository and combined with Version. + When empty, defaults to hyperbytedb:{Version} or hyperbytedb:latest. type: string imagePullPolicy: description: PullPolicy describes a policy for if/when to pull a container @@ -3215,10 +3223,6 @@ spec: type: array logging: properties: - detailedTrace: - description: Emit per-phase performance logs (write/query/flush). - Off by default. - type: boolean format: default: text enum: @@ -3234,13 +3238,6 @@ spec: - warn - error type: string - otlpEndpoint: - description: OTLP HTTP endpoint for trace export (e.g. http://alloy-logs:4318). - type: string - otlpSampleRatio: - description: Fraction of traces exported to OTLP (0.0–1.0). Default - 1.0 when OTLP is set. - type: string type: object monitoring: properties: @@ -3271,7 +3268,7 @@ spec: returning errors to clients. properties: enabled: - default: false + default: true description: |- When false, the operator does not create or reconcile any proxy resources. Existing proxy Deployment/Service (if any) are left alone @@ -3283,7 +3280,7 @@ spec: Set to `/health/ready` for the deeper chDB-aware readiness check. type: string holdTimeoutSecs: - default: 10 + default: 30 description: |- How long the proxy waits for a backend to come back before failing a request with 503. Bigger values mean rolling restarts are smoother but @@ -3609,6 +3606,13 @@ spec: storageClassName: type: string type: object + walFormat: + default: bincode + description: 'WAL encoding format: "bincode" (default) or "arrow_ipc".' + enum: + - bincode + - arrow_ipc + type: string type: object tolerations: items: @@ -3823,6 +3827,10 @@ spec: type: object type: array version: + description: |- + Application version. Drives the container image tag for hyperbytedb and + hyperbytedb-proxy (e.g. version "0.8.3" → hyperbytedb:v0.8.3). Changing + this field triggers a rolling upgrade. type: string type: object status: diff --git a/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_ha.yaml b/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_ha.yaml index 76c2324..e52fd0c 100644 --- a/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_ha.yaml +++ b/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_ha.yaml @@ -27,8 +27,6 @@ spec: logging: level: info format: json - otlpEndpoint: http://alloy-logs:4318 - otlpSampleRatio: "0.1" cardinality: maxTagValuesPerMeasurement: 100000 maxMeasurementsPerDatabase: 10000 diff --git a/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbytedb.io.yaml b/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbytedb.io.yaml index 507e184..0be6400 100644 --- a/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbytedb.io.yaml +++ b/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbytedb.io.yaml @@ -1,4 +1,5 @@ --- +--- apiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition metadata: @@ -3189,7 +3190,10 @@ spec: type: integer type: object image: - default: hyperbytedb:latest + description: |- + Container image reference. When set with a tag (contains ':'), used as-is. + When set without a tag, treated as the repository and combined with Version. + When empty, defaults to hyperbytedb:{Version} or hyperbytedb:latest. type: string imagePullPolicy: description: PullPolicy describes a policy for if/when to pull a container @@ -3271,7 +3275,7 @@ spec: returning errors to clients. properties: enabled: - default: false + default: true description: |- When false, the operator does not create or reconcile any proxy resources. Existing proxy Deployment/Service (if any) are left alone @@ -3283,7 +3287,7 @@ spec: Set to `/health/ready` for the deeper chDB-aware readiness check. type: string holdTimeoutSecs: - default: 10 + default: 30 description: |- How long the proxy waits for a backend to come back before failing a request with 503. Bigger values mean rolling restarts are smoother but @@ -3823,6 +3827,10 @@ spec: type: object type: array version: + description: |- + Application version. Drives the container image tag for hyperbytedb and + hyperbytedb-proxy (e.g. version "0.8.3" → hyperbytedb:v0.8.3). Changing + this field triggers a rolling upgrade. type: string type: object status: diff --git a/internal/controller/hyperbytedbbackup_controller.go b/internal/controller/hyperbytedbbackup_controller.go index 0ac6997..64d6f1e 100644 --- a/internal/controller/hyperbytedbbackup_controller.go +++ b/internal/controller/hyperbytedbbackup_controller.go @@ -207,10 +207,7 @@ func (r *HyperbytedbBackupReconciler) reconcileCronJob(ctx context.Context, back } func (r *HyperbytedbBackupReconciler) buildBackupJob(backup *hyperbytedbv1alpha1.HyperbytedbBackup, cluster *hyperbytedbv1alpha1.HyperbytedbCluster, name string) *batchv1.Job { - image := cluster.Spec.Image - if image == "" { - image = "hyperbytedb:latest" - } + image := hyperbytedb.ResolveHyperbytedbImage(cluster) s3 := backup.Spec.Destination.S3 s3Path := r.buildBackupS3Path(backup) diff --git a/internal/controller/hyperbytedbcluster_controller.go b/internal/controller/hyperbytedbcluster_controller.go index b66acbf..a444a65 100644 --- a/internal/controller/hyperbytedbcluster_controller.go +++ b/internal/controller/hyperbytedbcluster_controller.go @@ -216,9 +216,8 @@ func (r *HyperbytedbClusterReconciler) Reconcile(ctx context.Context, req ctrl.R log.Error(err, "Failed to reconcile HPA") } - // 11b. Proxy (optional). When disabled (default) we don't reconcile — - // any pre-existing proxy resources are left intact so users can clean - // them up out-of-band. + // 11b. Proxy (optional). Enabled by default; set spec.proxy.enabled=false + // to skip reconciliation. if hyperbytedb.ProxyEnabled(cluster) { if err := r.reconcileProxy(ctx, cluster); err != nil { log.Error(err, "Failed to reconcile proxy") @@ -527,7 +526,10 @@ func (r *HyperbytedbClusterReconciler) reconcilePDB(ctx context.Context, cluster pdbName := cluster.Name + "-pdb" - if replicas < 3 { + // Single-node clusters don't need a PDB. For 2+ replicas keep at least + // N-1 pods available during voluntary disruption so rolling upgrades and + // node drains never take the whole cluster offline at once. + if replicas < 2 { existing := &policyv1.PodDisruptionBudget{} err := r.Get(ctx, types.NamespacedName{Name: pdbName, Namespace: cluster.Namespace}, existing) if apierrors.IsNotFound(err) { @@ -990,6 +992,15 @@ func (r *HyperbytedbClusterReconciler) updateStatus(ctx context.Context, cluster cluster.Status.ReadyReplicas = sts.ReadyReplicas cluster.Status.ConfigHash = configHash + rollingUpgrade := false + curSTS := &appsv1.StatefulSet{} + if err := r.Get(ctx, types.NamespacedName{ + Name: hyperbytedb.StatefulSetName(cluster), Namespace: cluster.Namespace, + }, curSTS); err == nil { + rollingUpgrade = curSTS.Status.UpdateRevision != "" && + curSTS.Status.UpdateRevision != curSTS.Status.CurrentRevision + } + if sts.ReadyReplicas == sts.SpecReplicas && sts.SpecReplicas > 0 { cluster.Status.Phase = hyperbytedbv1alpha1.ClusterPhaseRunning meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{ @@ -1000,9 +1011,14 @@ func (r *HyperbytedbClusterReconciler) updateStatus(ctx context.Context, cluster LastTransitionTime: metav1.Now(), }) } else { - if cluster.Status.Phase != hyperbytedbv1alpha1.ClusterPhaseFailed && - cluster.Status.Phase != hyperbytedbv1alpha1.ClusterPhaseScaling && - cluster.Status.Phase != hyperbytedbv1alpha1.ClusterPhaseUpgrading { + switch { + case cluster.Status.Phase == hyperbytedbv1alpha1.ClusterPhaseFailed: + // preserve + case cluster.Status.Phase == hyperbytedbv1alpha1.ClusterPhaseScaling: + // preserve + case rollingUpgrade: + cluster.Status.Phase = hyperbytedbv1alpha1.ClusterPhaseUpgrading + default: cluster.Status.Phase = hyperbytedbv1alpha1.ClusterPhaseInitializing } meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{ diff --git a/internal/controller/hyperbytedbrestore_controller.go b/internal/controller/hyperbytedbrestore_controller.go index f902c84..862b59b 100644 --- a/internal/controller/hyperbytedbrestore_controller.go +++ b/internal/controller/hyperbytedbrestore_controller.go @@ -246,10 +246,7 @@ func (r *HyperbytedbRestoreReconciler) buildRestoreJob( s3Source *hyperbytedbv1alpha1.S3BackupSpec, ordinal int32, ) *batchv1.Job { - image := cluster.Spec.Image - if image == "" { - image = "hyperbytedb:latest" - } + image := hyperbytedb.ResolveHyperbytedbImage(cluster) s3Path := s3Source.Bucket if s3Source.Prefix != "" { diff --git a/internal/hyperbytedb/client.go b/internal/hyperbytedb/client.go index 5a7d275..2de95b4 100644 --- a/internal/hyperbytedb/client.go +++ b/internal/hyperbytedb/client.go @@ -108,7 +108,9 @@ func (c *Client) DrainNode(ctx context.Context, host string, port int32) error { return err } defer func() { _ = resp.Body.Close() }() - if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { + if resp.StatusCode != http.StatusOK && + resp.StatusCode != http.StatusNoContent && + resp.StatusCode != http.StatusAccepted { respBody, _ := io.ReadAll(resp.Body) return fmt.Errorf("drain returned %d: %s", resp.StatusCode, string(respBody)) } diff --git a/internal/hyperbytedb/configmap.go b/internal/hyperbytedb/configmap.go index 935cd77..8c12317 100644 --- a/internal/hyperbytedb/configmap.go +++ b/internal/hyperbytedb/configmap.go @@ -66,7 +66,7 @@ func renderConfigTOMLWithClusterEnabled(cluster *v1alpha1.HyperbytedbCluster, cl var b strings.Builder writeServerSection(&b, spec) - writeStorageSection(&b) + writeStorageSection(&b, spec) writeFlushSection(&b, spec) writeChdbSection(&b, spec) writeAuthSection(&b, spec) @@ -108,10 +108,15 @@ func writeServerSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpe } } -func writeStorageSection(b *strings.Builder) { +func writeStorageSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) { b.WriteString("\n[storage]\n") fmt.Fprintf(b, "wal_dir = \"%s\"\n", defaultWalDir) fmt.Fprintf(b, "meta_dir = \"%s\"\n", defaultMetaDir) + walFormat := "bincode" + if spec.Storage.WALFormat != "" { + walFormat = spec.Storage.WALFormat + } + fmt.Fprintf(b, "wal_format = \"%s\"\n", walFormat) } func writeFlushSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) { @@ -142,6 +147,11 @@ func writeFlushSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec if spec.Flush.WALBatchDelayUs > 0 { fmt.Fprintf(b, "wal_batch_delay_us = %d\n", spec.Flush.WALBatchDelayUs) } + arrowWALEnabled := true + if spec.Flush.ArrowWALEnabled != nil { + arrowWALEnabled = *spec.Flush.ArrowWALEnabled + } + fmt.Fprintf(b, "arrow_wal_enabled = %t\n", arrowWALEnabled) } func writeChdbSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) { @@ -189,15 +199,7 @@ func writeLoggingSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSp format = spec.Logging.Format } fmt.Fprintf(b, "format = \"%s\"\n", format) - if spec.Logging.DetailedTrace != nil { - fmt.Fprintf(b, "detailed_trace = %t\n", *spec.Logging.DetailedTrace) - } - if spec.Logging.OtlpEndpoint != "" { - fmt.Fprintf(b, "otlp_endpoint = \"%s\"\n", spec.Logging.OtlpEndpoint) - } - if spec.Logging.OtlpSampleRatio != "" { - fmt.Fprintf(b, "otlp_sample_ratio = %s\n", spec.Logging.OtlpSampleRatio) - } + } func writeStatementSummarySection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) { @@ -251,7 +253,7 @@ func writeRetentionSection(b *strings.Builder, spec *v1alpha1.HyperbytedbCluster if spec.Retention.Enabled != nil { enabled = *spec.Retention.Enabled } - interval := "60s" + interval := "12h" if spec.Retention.Interval != "" { interval = spec.Retention.Interval } diff --git a/internal/hyperbytedb/configmap_test.go b/internal/hyperbytedb/configmap_test.go index 83c7dc2..c5d9c3b 100644 --- a/internal/hyperbytedb/configmap_test.go +++ b/internal/hyperbytedb/configmap_test.go @@ -31,7 +31,9 @@ func TestRenderConfigTOML_singleNode(t *testing.T) { "[cluster]", "enabled = false", "[retention]", - `interval = "60s"`, + `interval = "12h"`, + `wal_format = "bincode"`, + "arrow_wal_enabled = true", } { if !strings.Contains(out, want) { t.Fatalf("expected config to contain %q\n\ngot:\n%s", want, out) diff --git a/internal/hyperbytedb/image.go b/internal/hyperbytedb/image.go new file mode 100644 index 0000000..e86416c --- /dev/null +++ b/internal/hyperbytedb/image.go @@ -0,0 +1,65 @@ +package hyperbytedb + +import ( + "fmt" + "strings" + + v1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" +) + +const ( + defaultHyperbytedbRepo = "hyperbytedb" + defaultProxyRepo = "hyperbytedb-proxy" + defaultImageTag = "latest" +) + +// NormalizeVersionTag maps a semver-ish version string to an OCI image tag. +// "0.8.3" and "v0.8.3" both become "v0.8.3" to match published container tags. +func NormalizeVersionTag(version string) string { + v := strings.TrimSpace(version) + if v == "" { + return defaultImageTag + } + if !strings.HasPrefix(v, "v") { + v = "v" + v + } + return v +} + +// ResolveHyperbytedbImage returns the container image for database pods. +// +// Resolution order: +// - spec.image with a tag (contains ':') → used verbatim (local/kind overrides) +// - spec.image without a tag → repository, tag from spec.version (or latest) +// - spec.version only → hyperbytedb:{tag} +// - neither → hyperbytedb:latest +func ResolveHyperbytedbImage(cluster *v1alpha1.HyperbytedbCluster) string { + return resolveComponentImage(cluster.Spec.Image, cluster.Spec.Version, defaultHyperbytedbRepo) +} + +// ResolveProxyImage returns the container image for hyperbytedb-proxy pods. +// Uses the same tag rules as ResolveHyperbytedbImage so proxy and database +// versions stay aligned when spec.proxy.image is not pinned. +func ResolveProxyImage(cluster *v1alpha1.HyperbytedbCluster) string { + repoOverride := "" + if cluster.Spec.Proxy != nil { + repoOverride = cluster.Spec.Proxy.Image + } + return resolveComponentImage(repoOverride, cluster.Spec.Version, defaultProxyRepo) +} + +func resolveComponentImage(imageField, version, defaultRepo string) string { + tag := defaultImageTag + if version != "" { + tag = NormalizeVersionTag(version) + } + + img := strings.TrimSpace(imageField) + if img == "" { + return fmt.Sprintf("%s:%s", defaultRepo, tag) + } + if strings.Contains(img, ":") { + return img + } + return fmt.Sprintf("%s:%s", img, tag) +} diff --git a/internal/hyperbytedb/image_test.go b/internal/hyperbytedb/image_test.go new file mode 100644 index 0000000..cd12095 --- /dev/null +++ b/internal/hyperbytedb/image_test.go @@ -0,0 +1,70 @@ +package hyperbytedb + +import ( + "testing" + + v1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" +) + +func TestNormalizeVersionTag(t *testing.T) { + tests := []struct { + in string + want string + }{ + {"", "latest"}, + {"0.8.3", "v0.8.3"}, + {"v0.8.3", "v0.8.3"}, + {" 1.0.0 ", "v1.0.0"}, + } + for _, tc := range tests { + if got := NormalizeVersionTag(tc.in); got != tc.want { + t.Errorf("NormalizeVersionTag(%q) = %q, want %q", tc.in, got, tc.want) + } + } +} + +func TestResolveHyperbytedbImage(t *testing.T) { + cluster := func(image, version string) *v1alpha1.HyperbytedbCluster { + return &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{ + Image: image, + Version: version, + }, + } + } + + tests := []struct { + name string + c *v1alpha1.HyperbytedbCluster + want string + }{ + {"version only", cluster("", "0.8.3"), "hyperbytedb:v0.8.3"}, + {"explicit tag wins", cluster("hyperbytedb:local", "0.8.3"), "hyperbytedb:local"}, + {"repo plus version", cluster("ghcr.io/org/hyperbytedb", "0.8.3"), "ghcr.io/org/hyperbytedb:v0.8.3"}, + {"default latest", cluster("", ""), "hyperbytedb:latest"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := ResolveHyperbytedbImage(tc.c); got != tc.want { + t.Errorf("ResolveHyperbytedbImage() = %q, want %q", got, tc.want) + } + }) + } +} + +func TestResolveProxyImage(t *testing.T) { + c := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{ + Version: "0.8.3", + Proxy: &v1alpha1.ProxySpec{}, + }, + } + if got := ResolveProxyImage(c); got != "hyperbytedb-proxy:v0.8.3" { + t.Fatalf("ResolveProxyImage() = %q, want hyperbytedb-proxy:v0.8.3", got) + } + + c.Spec.Proxy.Image = "hyperbytedb-proxy:local" + if got := ResolveProxyImage(c); got != "hyperbytedb-proxy:local" { + t.Fatalf("explicit proxy image = %q, want hyperbytedb-proxy:local", got) + } +} diff --git a/internal/hyperbytedb/proxy.go b/internal/hyperbytedb/proxy.go index 68c37c7..3ba846b 100644 --- a/internal/hyperbytedb/proxy.go +++ b/internal/hyperbytedb/proxy.go @@ -26,9 +26,13 @@ func ProxyServiceName(cluster *v1alpha1.HyperbytedbCluster) string { return cluster.Name + "-proxy" } -// ProxyEnabled reports whether the user opted-in to proxy reconciliation. +// 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 { - return cluster.Spec.Proxy != nil && cluster.Spec.Proxy.Enabled + if cluster.Spec.Proxy == nil { + return true + } + return cluster.Spec.Proxy.Enabled } // proxyLabels intentionally uses `name=hyperbytedb-proxy` instead of @@ -106,10 +110,7 @@ func BuildProxyDeployment(cluster *v1alpha1.HyperbytedbCluster) *appsv1.Deployme replicas = *spec.Replicas } - image := spec.Image - if image == "" { - image = "hyperbytedb-proxy:latest" - } + image := ResolveProxyImage(cluster) pullPolicy := spec.ImagePullPolicy if pullPolicy == "" { pullPolicy = corev1.PullIfNotPresent @@ -120,7 +121,7 @@ func BuildProxyDeployment(cluster *v1alpha1.HyperbytedbCluster) *appsv1.Deployme backendService := fmt.Sprintf("%s.%s.svc.cluster.local", HeadlessServiceName(cluster), cluster.Namespace) - holdSecs := int32(10) + holdSecs := int32(30) if spec.HoldTimeoutSecs > 0 { holdSecs = spec.HoldTimeoutSecs } diff --git a/internal/hyperbytedb/statefulset.go b/internal/hyperbytedb/statefulset.go index ae4dfe0..adcdb80 100644 --- a/internal/hyperbytedb/statefulset.go +++ b/internal/hyperbytedb/statefulset.go @@ -14,6 +14,15 @@ import ( v1alpha1 "github.com/hyperbyte-cloud/hyperbytedb-operator/api/v1alpha1" ) +const ( + // drainAckWaitSecs mirrors hyperbytedb DrainService::wait_for_replication_acks + // max_wait. preStop must outlive that window so the pod is not SIGKILL'd mid-drain. + drainAckWaitSecs = 90 + // podTerminationGraceSecs is the kubelet grace budget (includes preStop). Keep + // comfortably above drainAckWaitSecs + headroom for flush + SIGTERM cleanup. + podTerminationGraceSecs = 120 +) + func StatefulSetName(cluster *v1alpha1.HyperbytedbCluster) string { return cluster.Name } @@ -30,10 +39,7 @@ func BuildStatefulSet(cluster *v1alpha1.HyperbytedbCluster, configHash string) * clusterEnabled := replicas > 1 headlessSvc := HeadlessServiceName(cluster) - image := cluster.Spec.Image - if image == "" { - image = "hyperbytedb:latest" - } + image := ResolveHyperbytedbImage(cluster) pullPolicy := cluster.Spec.ImagePullPolicy if pullPolicy == "" { @@ -138,10 +144,6 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve {Name: "HYPERBYTEDB__STORAGE__META_DIR", Value: defaultMetaDir}, {Name: "HYPERBYTEDB__CHDB__SESSION_DATA_PATH", Value: chdbPath}, } - if cluster.Spec.Logging.OtlpEndpoint != "" { - env = append(env, corev1.EnvVar{Name: "OTEL_SERVICE_NAME", Value: cluster.Name}) - } - // 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. @@ -195,6 +197,28 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve TimeoutSeconds: 3, } + // Startup can be slow on a node with large series cardinality or a cold + // chDB that must be rebuilt from the WAL: the HTTP listener only binds + // after the dedup-cache warm and startup WAL replay complete. Without a + // startup probe, the liveness probe (15s delay + 3×10s) fires at ~35s and + // SIGKILLs the pod mid-warm, which restarts and re-does the same heavy + // work — a self-amplifying CPU/memory loop. The startup probe holds + // liveness off until /ping first answers, giving the listener up to + // FailureThreshold×PeriodSeconds (600s) to come up before any restart. + startupProbe := &corev1.Probe{ + ProbeHandler: corev1.ProbeHandler{ + HTTPGet: &corev1.HTTPGetAction{ + Path: "/ping", + Port: intstr.FromString("http"), + Scheme: probeScheme, + }, + }, + InitialDelaySeconds: 5, + PeriodSeconds: 5, + TimeoutSeconds: 3, + FailureThreshold: 120, + } + // --------------- main container --------------- mainContainer := corev1.Container{ Name: "hyperbytedb", @@ -210,6 +234,7 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve }, }, VolumeMounts: volumeMounts, + StartupProbe: startupProbe, LivenessProbe: livenessProbe, ReadinessProbe: readinessProbe, Resources: cluster.Spec.Resources, @@ -219,8 +244,7 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve mainContainer.Lifecycle = &corev1.Lifecycle{ PreStop: &corev1.LifecycleHandler{ Exec: &corev1.ExecAction{ - Command: []string{"sh", "-c", - fmt.Sprintf("curl -sf -X POST http://localhost:%d/internal/drain || true; sleep 45", port)}, + Command: []string{"sh", "-c", clusterPreStopScript(port)}, }, }, } @@ -258,7 +282,7 @@ exec hyperbytedb --config /etc/hyperbytedb/config.toml serve // --------------- pod spec --------------- podSpec := corev1.PodSpec{ - TerminationGracePeriodSeconds: ptr.To(int64(60)), + TerminationGracePeriodSeconds: ptr.To(int64(podTerminationGraceSecs)), ImagePullSecrets: cluster.Spec.ImagePullSecrets, InitContainers: initContainers, Containers: []corev1.Container{mainContainer}, @@ -335,3 +359,18 @@ func podManagementPolicy(clusterEnabled bool) appsv1.PodManagementPolicyType { } return appsv1.ParallelPodManagement } + +// clusterPreStopScript initiates graceful drain and blocks until the node +// 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 +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 + exit 0 + fi + sleep 1 +done +exit 0`, port, drainAckWaitSecs) +}