diff --git a/api/v1alpha1/hyperbytedbcluster_types.go b/api/v1alpha1/hyperbytedbcluster_types.go index 7faa463..14c2c22 100644 --- a/api/v1alpha1/hyperbytedbcluster_types.go +++ b/api/v1alpha1/hyperbytedbcluster_types.go @@ -76,6 +76,11 @@ type HyperbytedbClusterSpec struct { // +optional Cluster ClusterTuningSpec `json:"cluster,omitempty"` + // Experimental series sharding. When set, the operator writes a `[sharding]` + // block into config.toml. Enabling it requires replicas > 1 (cluster mode). + // +optional + Sharding *ShardingSpec `json:"sharding,omitempty"` + // +optional Cardinality CardinalitySpec `json:"cardinality,omitempty"` @@ -319,6 +324,81 @@ type ClusterTuningSpec struct { DrainWaitSecs int32 `json:"drainWaitSecs,omitempty"` } +// ShardingSpec maps to HyperbyteDB `[sharding]` (experimental series_id range +// sharding). Omitted fields are left out of config.toml so the server defaults apply. +// When enabled, config validation requires regionMergeSeries < regionSplitSeries +// < regionMaxSeries (same inequalities as hyperbytedb). +type ShardingSpec struct { + // Master switch. Written as sharding.enabled. + Enabled bool `json:"enabled"` + + // Target replica count per shard region. + // +optional + // +kubebuilder:validation:Minimum=1 + ReplicationFactor int32 `json:"replicationFactor,omitempty"` + + // Target series per region before split. + // +optional + // +kubebuilder:validation:Minimum=1 + RegionSplitSeries int64 `json:"regionSplitSeries,omitempty"` + + // Hard split threshold (must be > regionSplitSeries when both are set). + // +optional + // +kubebuilder:validation:Minimum=1 + RegionMaxSeries int64 `json:"regionMaxSeries,omitempty"` + + // Merge when adjacent regions fall below this (must be < regionSplitSeries when both are set). + // +optional + // +kubebuilder:validation:Minimum=1 + RegionMergeSeries int64 `json:"regionMergeSeries,omitempty"` + + // Cooldown between split/merge on a region. + // +optional + // +kubebuilder:validation:Minimum=0 + SplitMergeIntervalSecs int64 `json:"splitMergeIntervalSecs,omitempty"` + + // Max concurrent split/move/merge operators. + // +optional + // +kubebuilder:validation:Minimum=1 + ScheduleLimit int32 `json:"scheduleLimit,omitempty"` + + // Region-stats report interval and shard scheduler tick (same duration). + // +optional + // +kubebuilder:validation:Minimum=1 + HeartbeatIntervalSecs int64 `json:"heartbeatIntervalSecs,omitempty"` + + // Sync bootstrap RPC timeout. + // +optional + // +kubebuilder:validation:Minimum=1 + BootstrapTimeoutMs int64 `json:"bootstrapTimeoutMs,omitempty"` + + // Seconds before the Raft leader proposes TransferPrimary for an unhealthy primary. + // +optional + // +kubebuilder:validation:Minimum=1 + PrimaryFailoverAfterSecs int64 `json:"primaryFailoverAfterSecs,omitempty"` + + // Per-peer HTTP timeout for sharded query/write/metadata scatter. + // +optional + // +kubebuilder:validation:Minimum=1 + ScatterPeerTimeoutMs int64 `json:"scatterPeerTimeoutMs,omitempty"` + + // Max Active peers tried per region per scatter request. + // +optional + // +kubebuilder:validation:Minimum=1 + ScatterMaxPeerAttempts int32 `json:"scatterMaxPeerAttempts,omitempty"` + + // Load-based split QPS threshold; 0 = disabled. Emitted only when set so the + // server default (0) applies when omitted. + // +optional + // +kubebuilder:validation:Minimum=0 + LoadSplitQpsThreshold *int64 `json:"loadSplitQpsThreshold,omitempty"` + + // Hard cap on regions per measurement. + // +optional + // +kubebuilder:validation:Minimum=1 + MaxRegionsPerMeasurement int32 `json:"maxRegionsPerMeasurement,omitempty"` +} + // ReplicationSpec controls coordinator-side replication (how this node's // accepted client writes are replicated to peers). type ReplicationSpec struct { diff --git a/api/v1alpha1/hyperbytedbcluster_webhook.go b/api/v1alpha1/hyperbytedbcluster_webhook.go index a3769ce..ce3b4a7 100644 --- a/api/v1alpha1/hyperbytedbcluster_webhook.go +++ b/api/v1alpha1/hyperbytedbcluster_webhook.go @@ -150,5 +150,28 @@ func validateCluster(cluster *HyperbytedbCluster) (admission.Warnings, error) { warnings = append(warnings, "2-node clusters cannot tolerate any node failure; consider 3+ replicas") } + if err := validateSharding(cluster, replicas); err != nil { + return warnings, err + } + return warnings, nil } + +func validateSharding(cluster *HyperbytedbCluster, replicas int32) error { + s := cluster.Spec.Sharding + if s == nil { + return nil + } + if s.Enabled && replicas < 2 { + return fmt.Errorf("sharding.enabled requires replicas > 1 (cluster mode)") + } + if s.RegionMergeSeries > 0 && s.RegionSplitSeries > 0 && s.RegionMergeSeries >= s.RegionSplitSeries { + return fmt.Errorf("sharding.regionMergeSeries (%d) must be < regionSplitSeries (%d)", + s.RegionMergeSeries, s.RegionSplitSeries) + } + if s.RegionSplitSeries > 0 && s.RegionMaxSeries > 0 && s.RegionSplitSeries >= s.RegionMaxSeries { + return fmt.Errorf("sharding.regionSplitSeries (%d) must be < regionMaxSeries (%d)", + s.RegionSplitSeries, s.RegionMaxSeries) + } + return nil +} diff --git a/api/v1alpha1/hyperbytedbcluster_webhook_test.go b/api/v1alpha1/hyperbytedbcluster_webhook_test.go new file mode 100644 index 0000000..ececc13 --- /dev/null +++ b/api/v1alpha1/hyperbytedbcluster_webhook_test.go @@ -0,0 +1,89 @@ +package v1alpha1 + +import ( + "strings" + "testing" + + "k8s.io/utils/ptr" +) + +func TestValidateSharding(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + replicas int32 + sharding *ShardingSpec + wantErr string + }{ + {name: "nil sharding", replicas: 1}, + { + name: "enabled requires cluster", + replicas: 1, + sharding: &ShardingSpec{Enabled: true}, + wantErr: "replicas > 1", + }, + { + name: "enabled on 3 replicas", + replicas: 3, + sharding: &ShardingSpec{Enabled: true, ReplicationFactor: 2}, + }, + { + name: "merge not less than split", + replicas: 3, + sharding: &ShardingSpec{ + Enabled: true, + RegionMergeSeries: 10, + RegionSplitSeries: 5, + }, + wantErr: "regionMergeSeries", + }, + { + name: "split not less than max", + replicas: 3, + sharding: &ShardingSpec{ + Enabled: true, + RegionSplitSeries: 10, + RegionMaxSeries: 10, + }, + wantErr: "regionSplitSeries", + }, + { + name: "kind split-test thresholds", + replicas: 6, + sharding: &ShardingSpec{ + Enabled: true, + RegionSplitSeries: 5, + RegionMaxSeries: 10, + RegionMergeSeries: 2, + }, + }, + { + name: "disabled on single replica is ok", + replicas: 1, + sharding: &ShardingSpec{Enabled: false}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + cluster := &HyperbytedbCluster{ + Spec: HyperbytedbClusterSpec{ + Replicas: ptr.To(tt.replicas), + Sharding: tt.sharding, + }, + } + err := validateSharding(cluster, tt.replicas) + if tt.wantErr == "" { + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("want error containing %q, got %v", tt.wantErr, err) + } + }) + } +} diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index f3ef7ae..55275c9 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -392,6 +392,11 @@ func (in *HyperbytedbClusterSpec) DeepCopyInto(out *HyperbytedbClusterSpec) { out.Logging = in.Logging in.Resources.DeepCopyInto(&out.Resources) in.Cluster.DeepCopyInto(&out.Cluster) + if in.Sharding != nil { + in, out := &in.Sharding, &out.Sharding + *out = new(ShardingSpec) + (*in).DeepCopyInto(*out) + } out.Cardinality = in.Cardinality in.StatementSummary.DeepCopyInto(&out.StatementSummary) in.HintedHandoff.DeepCopyInto(&out.HintedHandoff) @@ -868,6 +873,26 @@ func (in *ServerSpec) DeepCopy() *ServerSpec { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ShardingSpec) DeepCopyInto(out *ShardingSpec) { + *out = *in + if in.LoadSplitQpsThreshold != nil { + in, out := &in.LoadSplitQpsThreshold, &out.LoadSplitQpsThreshold + *out = new(int64) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ShardingSpec. +func (in *ShardingSpec) DeepCopy() *ShardingSpec { + if in == nil { + return nil + } + out := new(ShardingSpec) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *StatementSummarySpec) DeepCopyInto(out *StatementSummarySpec) { *out = *in diff --git a/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml b/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml index 007e56c..64b6f6d 100644 --- a/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml +++ b/config/crd/bases/hyperbytedb.hyperbyte.cloud_hyperbytedbclusters.yaml @@ -3589,6 +3589,89 @@ spec: - enabled type: object type: object + sharding: + description: |- + Experimental series sharding. When set, the operator writes a `[sharding]` + block into config.toml. Enabling it requires replicas > 1 (cluster mode). + properties: + bootstrapTimeoutMs: + description: Sync bootstrap RPC timeout. + format: int64 + minimum: 1 + type: integer + enabled: + description: Master switch. Written as sharding.enabled. + type: boolean + heartbeatIntervalSecs: + description: Region-stats report interval and shard scheduler + tick (same duration). + format: int64 + minimum: 1 + type: integer + loadSplitQpsThreshold: + description: |- + Load-based split QPS threshold; 0 = disabled. Emitted only when set so the + server default (0) applies when omitted. + format: int64 + minimum: 0 + type: integer + maxRegionsPerMeasurement: + description: Hard cap on regions per measurement. + format: int32 + minimum: 1 + type: integer + primaryFailoverAfterSecs: + description: Seconds before the Raft leader proposes TransferPrimary + for an unhealthy primary. + format: int64 + minimum: 1 + type: integer + regionMaxSeries: + description: Hard split threshold (must be > regionSplitSeries + when both are set). + format: int64 + minimum: 1 + type: integer + regionMergeSeries: + description: Merge when adjacent regions fall below this (must + be < regionSplitSeries when both are set). + format: int64 + minimum: 1 + type: integer + regionSplitSeries: + description: Target series per region before split. + format: int64 + minimum: 1 + type: integer + replicationFactor: + description: Target replica count per shard region. + format: int32 + minimum: 1 + type: integer + scatterMaxPeerAttempts: + description: Max Active peers tried per region per scatter request. + format: int32 + minimum: 1 + type: integer + scatterPeerTimeoutMs: + description: Per-peer HTTP timeout for sharded query/write/metadata + scatter. + format: int64 + minimum: 1 + type: integer + scheduleLimit: + description: Max concurrent split/move/merge operators. + format: int32 + minimum: 1 + type: integer + splitMergeIntervalSecs: + description: Cooldown between split/merge on a region. + format: int64 + minimum: 0 + type: integer + required: + - enabled + type: object statementSummary: description: |- StatementSummarySpec controls collection of per-statement execution stats diff --git a/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_cluster.yaml b/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_cluster.yaml index caa05b1..fde677e 100644 --- a/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_cluster.yaml +++ b/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_cluster.yaml @@ -43,6 +43,13 @@ spec: replication: mode: async ackTimeoutMs: 5000 + sharding: + enabled: true + replicationFactor: 3 + regionSplitSeries: 100000 + regionMaxSeries: 150000 + regionMergeSeries: 20000 + heartbeatIntervalSecs: 10 monitoring: enabled: true serviceMonitor: true diff --git a/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_ha.yaml b/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_ha.yaml index fae4725..61f995b 100644 --- a/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_ha.yaml +++ b/config/samples/hyperbytedb_v1alpha1_hyperbytedbcluster_ha.yaml @@ -65,6 +65,13 @@ spec: ackTimeoutMs: 5000 syncQuorum: minAcks: majority + sharding: + enabled: true + replicationFactor: 3 + regionSplitSeries: 100000 + regionMaxSeries: 150000 + regionMergeSeries: 20000 + heartbeatIntervalSecs: 10 monitoring: enabled: true serviceMonitor: true diff --git a/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbyte.cloud.yaml b/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbyte.cloud.yaml index 007e56c..64b6f6d 100644 --- a/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbyte.cloud.yaml +++ b/dist/chart/templates/crd/hyperbytedbclusters.hyperbytedb.hyperbyte.cloud.yaml @@ -3589,6 +3589,89 @@ spec: - enabled type: object type: object + sharding: + description: |- + Experimental series sharding. When set, the operator writes a `[sharding]` + block into config.toml. Enabling it requires replicas > 1 (cluster mode). + properties: + bootstrapTimeoutMs: + description: Sync bootstrap RPC timeout. + format: int64 + minimum: 1 + type: integer + enabled: + description: Master switch. Written as sharding.enabled. + type: boolean + heartbeatIntervalSecs: + description: Region-stats report interval and shard scheduler + tick (same duration). + format: int64 + minimum: 1 + type: integer + loadSplitQpsThreshold: + description: |- + Load-based split QPS threshold; 0 = disabled. Emitted only when set so the + server default (0) applies when omitted. + format: int64 + minimum: 0 + type: integer + maxRegionsPerMeasurement: + description: Hard cap on regions per measurement. + format: int32 + minimum: 1 + type: integer + primaryFailoverAfterSecs: + description: Seconds before the Raft leader proposes TransferPrimary + for an unhealthy primary. + format: int64 + minimum: 1 + type: integer + regionMaxSeries: + description: Hard split threshold (must be > regionSplitSeries + when both are set). + format: int64 + minimum: 1 + type: integer + regionMergeSeries: + description: Merge when adjacent regions fall below this (must + be < regionSplitSeries when both are set). + format: int64 + minimum: 1 + type: integer + regionSplitSeries: + description: Target series per region before split. + format: int64 + minimum: 1 + type: integer + replicationFactor: + description: Target replica count per shard region. + format: int32 + minimum: 1 + type: integer + scatterMaxPeerAttempts: + description: Max Active peers tried per region per scatter request. + format: int32 + minimum: 1 + type: integer + scatterPeerTimeoutMs: + description: Per-peer HTTP timeout for sharded query/write/metadata + scatter. + format: int64 + minimum: 1 + type: integer + scheduleLimit: + description: Max concurrent split/move/merge operators. + format: int32 + minimum: 1 + type: integer + splitMergeIntervalSecs: + description: Cooldown between split/merge on a region. + format: int64 + minimum: 0 + type: integer + required: + - enabled + type: object statementSummary: description: |- StatementSummarySpec controls collection of per-statement execution stats diff --git a/dist/install.yaml b/dist/install.yaml index 472fb06..6465c3c 100644 --- a/dist/install.yaml +++ b/dist/install.yaml @@ -3792,6 +3792,89 @@ spec: - enabled type: object type: object + sharding: + description: |- + Experimental series sharding. When set, the operator writes a `[sharding]` + block into config.toml. Enabling it requires replicas > 1 (cluster mode). + properties: + bootstrapTimeoutMs: + description: Sync bootstrap RPC timeout. + format: int64 + minimum: 1 + type: integer + enabled: + description: Master switch. Written as sharding.enabled. + type: boolean + heartbeatIntervalSecs: + description: Region-stats report interval and shard scheduler + tick (same duration). + format: int64 + minimum: 1 + type: integer + loadSplitQpsThreshold: + description: |- + Load-based split QPS threshold; 0 = disabled. Emitted only when set so the + server default (0) applies when omitted. + format: int64 + minimum: 0 + type: integer + maxRegionsPerMeasurement: + description: Hard cap on regions per measurement. + format: int32 + minimum: 1 + type: integer + primaryFailoverAfterSecs: + description: Seconds before the Raft leader proposes TransferPrimary + for an unhealthy primary. + format: int64 + minimum: 1 + type: integer + regionMaxSeries: + description: Hard split threshold (must be > regionSplitSeries + when both are set). + format: int64 + minimum: 1 + type: integer + regionMergeSeries: + description: Merge when adjacent regions fall below this (must + be < regionSplitSeries when both are set). + format: int64 + minimum: 1 + type: integer + regionSplitSeries: + description: Target series per region before split. + format: int64 + minimum: 1 + type: integer + replicationFactor: + description: Target replica count per shard region. + format: int32 + minimum: 1 + type: integer + scatterMaxPeerAttempts: + description: Max Active peers tried per region per scatter request. + format: int32 + minimum: 1 + type: integer + scatterPeerTimeoutMs: + description: Per-peer HTTP timeout for sharded query/write/metadata + scatter. + format: int64 + minimum: 1 + type: integer + scheduleLimit: + description: Max concurrent split/move/merge operators. + format: int32 + minimum: 1 + type: integer + splitMergeIntervalSecs: + description: Cooldown between split/merge on a region. + format: int64 + minimum: 0 + type: integer + required: + - enabled + type: object statementSummary: description: |- StatementSummarySpec controls collection of per-statement execution stats diff --git a/internal/hyperbytedb/configmap.go b/internal/hyperbytedb/configmap.go index 8029ab5..32cdc07 100644 --- a/internal/hyperbytedb/configmap.go +++ b/internal/hyperbytedb/configmap.go @@ -77,6 +77,7 @@ func renderConfigTOMLWithClusterEnabled(cluster *v1alpha1.HyperbytedbCluster, cl writeRateLimitSection(&b, spec) writeRetentionSection(&b, spec) writeClusterSection(&b, cluster, clusterEnabled) + writeShardingSection(&b, spec) return b.String() } @@ -356,3 +357,51 @@ func writeMinAcks(b *strings.Builder, v *intstr.IntOrString) { fmt.Fprintf(b, "min_acks = \"%s\"\n", v.StrVal) } } + +func writeShardingSection(b *strings.Builder, spec *v1alpha1.HyperbytedbClusterSpec) { + if spec.Sharding == nil { + return + } + s := spec.Sharding + b.WriteString("\n[sharding]\n") + fmt.Fprintf(b, "enabled = %t\n", s.Enabled) + if s.ReplicationFactor > 0 { + fmt.Fprintf(b, "replication_factor = %d\n", s.ReplicationFactor) + } + if s.RegionSplitSeries > 0 { + fmt.Fprintf(b, "region_split_series = %d\n", s.RegionSplitSeries) + } + if s.RegionMaxSeries > 0 { + fmt.Fprintf(b, "region_max_series = %d\n", s.RegionMaxSeries) + } + if s.RegionMergeSeries > 0 { + fmt.Fprintf(b, "region_merge_series = %d\n", s.RegionMergeSeries) + } + if s.SplitMergeIntervalSecs > 0 { + fmt.Fprintf(b, "split_merge_interval_secs = %d\n", s.SplitMergeIntervalSecs) + } + if s.ScheduleLimit > 0 { + fmt.Fprintf(b, "schedule_limit = %d\n", s.ScheduleLimit) + } + if s.HeartbeatIntervalSecs > 0 { + fmt.Fprintf(b, "heartbeat_interval_secs = %d\n", s.HeartbeatIntervalSecs) + } + if s.BootstrapTimeoutMs > 0 { + fmt.Fprintf(b, "bootstrap_timeout_ms = %d\n", s.BootstrapTimeoutMs) + } + if s.PrimaryFailoverAfterSecs > 0 { + fmt.Fprintf(b, "primary_failover_after_secs = %d\n", s.PrimaryFailoverAfterSecs) + } + if s.ScatterPeerTimeoutMs > 0 { + fmt.Fprintf(b, "scatter_peer_timeout_ms = %d\n", s.ScatterPeerTimeoutMs) + } + if s.ScatterMaxPeerAttempts > 0 { + fmt.Fprintf(b, "scatter_max_peer_attempts = %d\n", s.ScatterMaxPeerAttempts) + } + if s.LoadSplitQpsThreshold != nil { + fmt.Fprintf(b, "load_split_qps_threshold = %d\n", *s.LoadSplitQpsThreshold) + } + if s.MaxRegionsPerMeasurement > 0 { + fmt.Fprintf(b, "max_regions_per_measurement = %d\n", s.MaxRegionsPerMeasurement) + } +} diff --git a/internal/hyperbytedb/configmap_test.go b/internal/hyperbytedb/configmap_test.go index c5d9c3b..bac0e7a 100644 --- a/internal/hyperbytedb/configmap_test.go +++ b/internal/hyperbytedb/configmap_test.go @@ -46,6 +46,7 @@ func TestRenderConfigTOML_singleNode(t *testing.T) { "data_dir", "anti_entropy", "wal_replication", + "[sharding]", } { if strings.Contains(out, absent) { t.Fatalf("expected config NOT to contain %q\n\ngot:\n%s", absent, out) @@ -108,3 +109,44 @@ func TestConfigHash_ignoresReplicaCount(t *testing.T) { t.Fatal("ConfigHash must not change when only replica count changes") } } + +func TestRenderConfigTOML_sharding(t *testing.T) { + cluster := &v1alpha1.HyperbytedbCluster{ + Spec: v1alpha1.HyperbytedbClusterSpec{ + Replicas: ptr.To(int32(3)), + Sharding: &v1alpha1.ShardingSpec{ + Enabled: true, + ReplicationFactor: 2, + RegionSplitSeries: 5, + RegionMaxSeries: 10, + RegionMergeSeries: 2, + HeartbeatIntervalSecs: 10, + LoadSplitQpsThreshold: ptr.To(int64(0)), + MaxRegionsPerMeasurement: 128, + }, + }, + } + + out := renderConfigTOML(cluster) + for _, want := range []string{ + "[sharding]", + "enabled = true", + "replication_factor = 2", + "region_split_series = 5", + "region_max_series = 10", + "region_merge_series = 2", + "heartbeat_interval_secs = 10", + "load_split_qps_threshold = 0", + "max_regions_per_measurement = 128", + } { + if !strings.Contains(out, want) { + t.Fatalf("expected config to contain %q\n\ngot:\n%s", want, out) + } + } + + off := &v1alpha1.HyperbytedbCluster{Spec: cluster.Spec} + off.Spec.Sharding = &v1alpha1.ShardingSpec{Enabled: false} + if ConfigHash(cluster) == ConfigHash(off) { + t.Fatal("ConfigHash must change when sharding.enabled changes") + } +}