diff --git a/api/v1alpha1/conversion_test.go b/api/v1alpha1/conversion_test.go index c2a198a5..6bb1c65e 100644 --- a/api/v1alpha1/conversion_test.go +++ b/api/v1alpha1/conversion_test.go @@ -10,6 +10,7 @@ import ( "testing" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" "github.com/keiailab/valkey-operator/api/v1alpha1" "github.com/keiailab/valkey-operator/api/v1alpha2" @@ -104,7 +105,9 @@ func TestValkeyCluster_ConvertTo_v1alpha2(t *testing.T) { ObjectMeta: metav1.ObjectMeta{Name: "cluster-test"}, Spec: v1alpha1.ValkeyClusterSpec{ Shards: 3, - ReplicasPerShard: 1, + ReplicasPerShard: ptr.To[int32](0), // defect ④: masters-only 명시 0 round-trip (nil 과 구별). + // defect ⑥: cluster modules round-trip (v1alpha1 ↔ v1alpha2). + Modules: []v1alpha1.ModuleSpec{{Name: "valkey-search"}}, }, } dst := &v1alpha2.ValkeyCluster{} @@ -115,7 +118,46 @@ func TestValkeyCluster_ConvertTo_v1alpha2(t *testing.T) { if dst.Spec.Shards != 3 { t.Errorf("Shards = %d, want 3", dst.Spec.Shards) } - if dst.Spec.ReplicasPerShard != 1 { - t.Errorf("ReplicasPerShard = %d, want 1", dst.Spec.ReplicasPerShard) + if dst.Spec.ReplicasPerShard == nil || *dst.Spec.ReplicasPerShard != 0 { + t.Errorf("ReplicasPerShard = %v, want explicit 0 (masters-only must round-trip, not nil)", dst.Spec.ReplicasPerShard) + } + if len(dst.Spec.Modules) != 1 || dst.Spec.Modules[0].Name != "valkey-search" { + t.Errorf("Modules = %+v, want [valkey-search]", dst.Spec.Modules) + } + + // 역방향 round-trip — v1alpha2 → v1alpha1 도 보존. + back := &v1alpha1.ValkeyCluster{} + if err := back.ConvertFrom(dst); err != nil { + t.Fatalf("ConvertFrom failed: %v", err) + } + if back.Spec.ReplicasPerShard == nil || *back.Spec.ReplicasPerShard != 0 { + t.Errorf("round-trip ReplicasPerShard = %v, want explicit 0", back.Spec.ReplicasPerShard) + } + if len(back.Spec.Modules) != 1 || back.Spec.Modules[0].Name != "valkey-search" { + t.Errorf("round-trip Modules = %+v, want [valkey-search]", back.Spec.Modules) + } +} + +// TestValkeyCluster_Convert_NilReplicasPerShard — defect ④: nil(미지정) 은 +// round-trip 내내 nil 로 유지되어야 한다 (controller/webhook 가 1 로 default). +// nil 이 도중에 0 으로 바뀌면 omit→masters-only 로 의미가 뒤집힌다. +func TestValkeyCluster_Convert_NilReplicasPerShard(t *testing.T) { + src := &v1alpha1.ValkeyCluster{ + ObjectMeta: metav1.ObjectMeta{Name: "cluster-nil-rps"}, + Spec: v1alpha1.ValkeyClusterSpec{Shards: 3, ReplicasPerShard: nil}, + } + dst := &v1alpha2.ValkeyCluster{} + if err := src.ConvertTo(dst); err != nil { + t.Fatalf("ConvertTo failed: %v", err) + } + if dst.Spec.ReplicasPerShard != nil { + t.Errorf("nil ReplicasPerShard converted to %v, want nil (omitted must stay omitted)", *dst.Spec.ReplicasPerShard) + } + back := &v1alpha1.ValkeyCluster{} + if err := back.ConvertFrom(dst); err != nil { + t.Fatalf("ConvertFrom failed: %v", err) + } + if back.Spec.ReplicasPerShard != nil { + t.Errorf("round-trip nil ReplicasPerShard = %v, want nil", *back.Spec.ReplicasPerShard) } } diff --git a/api/v1alpha1/types_helpers_test.go b/api/v1alpha1/types_helpers_test.go index de93a4b2..8ae14467 100644 --- a/api/v1alpha1/types_helpers_test.go +++ b/api/v1alpha1/types_helpers_test.go @@ -10,7 +10,11 @@ Licensed under the MIT License. See the LICENSE file for details. package v1alpha1 -import "testing" +import ( + "testing" + + "k8s.io/utils/ptr" +) func TestValkeySpecIsAutoFailoverEnabled(t *testing.T) { t.Parallel() @@ -97,13 +101,15 @@ func TestValkeyClusterSpecTotalNodes(t *testing.T) { cases := []struct { name string shards int32 - replicasPerShard int32 + replicasPerShard *int32 want int32 }{ - {"3 shards, 1 replica each → 6", 3, 1, 6}, - {"6 shards, 2 replicas each → 18", 6, 2, 18}, - {"1 shard, 0 replicas → 1", 1, 0, 1}, - {"0 shards → 0", 0, 1, 0}, + {"3 shards, 1 replica each → 6", 3, ptr.To[int32](1), 6}, + {"6 shards, 2 replicas each → 18", 6, ptr.To[int32](2), 18}, + {"1 shard, explicit 0 replicas → 1 (masters-only)", 1, ptr.To[int32](0), 1}, + {"3 shards, explicit 0 replicas → 3 (masters-only)", 3, ptr.To[int32](0), 3}, + {"3 shards, nil replicas → 6 (nil defaults to 1)", 3, nil, 6}, + {"0 shards → 0", 0, ptr.To[int32](1), 0}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { @@ -115,3 +121,27 @@ func TestValkeyClusterSpecTotalNodes(t *testing.T) { }) } } + +// TestValkeyClusterSpecGetReplicasPerShard — defect ④: nil(미지정)→1, +// 명시 0(masters-only)→0, 명시 2→2. 명시 0 이 default 1 로 덮이지 않음을 증명. +func TestValkeyClusterSpecGetReplicasPerShard(t *testing.T) { + t.Parallel() + cases := []struct { + name string + rps *int32 + want int32 + }{ + {"nil → 1 (omitted defaults to 1)", nil, 1}, + {"explicit 0 → 0 (masters-only preserved)", ptr.To[int32](0), 0}, + {"explicit 2 → 2", ptr.To[int32](2), 2}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + t.Parallel() + s := &ValkeyClusterSpec{ReplicasPerShard: c.rps} + if got := s.GetReplicasPerShard(); got != c.want { + t.Fatalf("GetReplicasPerShard() = %d, want %d", got, c.want) + } + }) + } +} diff --git a/api/v1alpha1/valkeycluster_types.go b/api/v1alpha1/valkeycluster_types.go index 5ebf1cf3..122935c6 100644 --- a/api/v1alpha1/valkeycluster_types.go +++ b/api/v1alpha1/valkeycluster_types.go @@ -40,11 +40,17 @@ type ValkeyClusterSpec struct { // +kubebuilder:default=3 Shards int32 `json:"shards"` - // shard 당 replica 수. 기본 1 → 총 노드 = shards*(1+replicasPerShard). + // shard 당 replica 수. 총 노드 = shards*(1+replicasPerShard). + // + // *pointer* 인 이유 (defect ④): non-pointer int32 + CRD `+kubebuilder:default=1` 은 + // apiserver/admission 이 "필드 미지정" 과 "명시 0" 을 구별하지 못해 명시 0 + // (masters-only 토폴로지) 을 default 1 로 덮어쓴다. pointer 로 두면 nil(미지정) + // 과 명시 0 이 구별된다. CRD default 는 제거하고 nil→1 defaulting 은 코드 + // (GetReplicasPerShard 접근자 + mutating webhook) 에서 처리한다. // +kubebuilder:validation:Minimum=0 // +kubebuilder:validation:Maximum=5 - // +kubebuilder:default=1 - ReplicasPerShard int32 `json:"replicasPerShard"` + // +optional + ReplicasPerShard *int32 `json:"replicasPerShard,omitempty"` // +kubebuilder:default=true AutoFailover bool `json:"autoFailover,omitempty"` @@ -83,6 +89,13 @@ type ValkeyClusterSpec struct { // +optional AdditionalConfig map[string]string `json:"additionalConfig,omitempty"` + // Modules — Valkey 공식 BSD module(valkey-search/json/bloom) 또는 BYO module 로딩. + // 외부 Redis Stack(RediSearch/RedisJSON, RSALv2/SSPL)은 라이선스 비호환으로 미지원 + // (ADR-0032). Name 만 지정 시 공식 preset 자동 resolve, Image 지정 시 BYO. Valkey CR + // (ValkeySpec.Modules) 와 동일 타입/검증 — cluster 샤드 pod 전체에 동일 적용. + // +optional + Modules []ModuleSpec `json:"modules,omitempty"` + // +optional SlowLog *SlowLogSpec `json:"slowLog,omitempty"` @@ -163,4 +176,15 @@ func init() { func (v *ValkeyCluster) GetConditions() *[]metav1.Condition { return &v.Status.Conditions } func (v *ValkeyCluster) SetPhase(phase string) { v.Status.Phase = ClusterPhase(phase) } -func (s *ValkeyClusterSpec) TotalNodes() int32 { return s.Shards * (1 + s.ReplicasPerShard) } +// GetReplicasPerShard — nil(미지정) → 1, 명시값(0 포함) → 그대로. +// +// defect ④: 명시 0 은 masters-only 토폴로지이므로 보존하고, 미지정만 1 로 +// defaulting 한다. ReplicasPerShard 를 읽는 모든 곳은 이 접근자를 거친다. +func (s *ValkeyClusterSpec) GetReplicasPerShard() int32 { + if s.ReplicasPerShard == nil { + return 1 + } + return *s.ReplicasPerShard +} + +func (s *ValkeyClusterSpec) TotalNodes() int32 { return s.Shards * (1 + s.GetReplicasPerShard()) } diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 6f6dd81d..26d5c523 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -1203,6 +1203,11 @@ func (in *ValkeyClusterList) DeepCopyObject() runtime.Object { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ValkeyClusterSpec) DeepCopyInto(out *ValkeyClusterSpec) { *out = *in + if in.ReplicasPerShard != nil { + in, out := &in.ReplicasPerShard, &out.ReplicasPerShard + *out = new(int32) + **out = **in + } out.Version = in.Version in.Storage.DeepCopyInto(&out.Storage) in.Resources.DeepCopyInto(&out.Resources) @@ -1254,6 +1259,13 @@ func (in *ValkeyClusterSpec) DeepCopyInto(out *ValkeyClusterSpec) { (*out)[key] = val } } + if in.Modules != nil { + in, out := &in.Modules, &out.Modules + *out = make([]ModuleSpec, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } if in.SlowLog != nil { in, out := &in.SlowLog, &out.SlowLog *out = new(SlowLogSpec) diff --git a/api/v1alpha2/types_helpers_test.go b/api/v1alpha2/types_helpers_test.go index ede1de90..75e73472 100644 --- a/api/v1alpha2/types_helpers_test.go +++ b/api/v1alpha2/types_helpers_test.go @@ -10,7 +10,11 @@ Licensed under the MIT License. See the LICENSE file for details. package v1alpha2 -import "testing" +import ( + "testing" + + "k8s.io/utils/ptr" +) func TestValkeySpecIsAutoFailoverEnabled(t *testing.T) { t.Parallel() @@ -97,13 +101,15 @@ func TestValkeyClusterSpecTotalNodes(t *testing.T) { cases := []struct { name string shards int32 - replicasPerShard int32 + replicasPerShard *int32 want int32 }{ - {"3 shards, 1 replica each → 6", 3, 1, 6}, - {"6 shards, 2 replicas each → 18", 6, 2, 18}, - {"1 shard, 0 replicas → 1", 1, 0, 1}, - {"0 shards → 0", 0, 1, 0}, + {"3 shards, 1 replica each → 6", 3, ptr.To[int32](1), 6}, + {"6 shards, 2 replicas each → 18", 6, ptr.To[int32](2), 18}, + {"1 shard, explicit 0 replicas → 1 (masters-only)", 1, ptr.To[int32](0), 1}, + {"3 shards, explicit 0 replicas → 3 (masters-only)", 3, ptr.To[int32](0), 3}, + {"3 shards, nil replicas → 12 (nil defaults to 1)", 3, nil, 6}, + {"0 shards → 0", 0, ptr.To[int32](1), 0}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { @@ -115,3 +121,41 @@ func TestValkeyClusterSpecTotalNodes(t *testing.T) { }) } } + +// TestValkeyClusterSpecGetReplicasPerShard — defect ④: nil(미지정)→1, +// 명시 0(masters-only)→0, 명시 2→2. 명시 0 이 default 1 로 덮이지 않음을 증명. +func TestValkeyClusterSpecGetReplicasPerShard(t *testing.T) { + t.Parallel() + cases := []struct { + name string + rps *int32 + want int32 + }{ + {"nil → 1 (omitted defaults to 1)", nil, 1}, + {"explicit 0 → 0 (masters-only preserved)", ptr.To[int32](0), 0}, + {"explicit 2 → 2", ptr.To[int32](2), 2}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + t.Parallel() + s := &ValkeyClusterSpec{ReplicasPerShard: c.rps} + if got := s.GetReplicasPerShard(); got != c.want { + t.Fatalf("GetReplicasPerShard() = %d, want %d", got, c.want) + } + }) + } +} + +// TestValkeyClusterSpecTotalNodesMastersOnly — 명시 0 의 TotalNodes 가 shards +// 와 같음(masters-only)을 명시 검증. nil 은 shards*2. +func TestValkeyClusterSpecTotalNodesMastersOnly(t *testing.T) { + t.Parallel() + mastersOnly := &ValkeyClusterSpec{Shards: 5, ReplicasPerShard: ptr.To[int32](0)} + if got := mastersOnly.TotalNodes(); got != 5 { + t.Fatalf("masters-only TotalNodes() = %d, want 5 (== shards)", got) + } + omitted := &ValkeyClusterSpec{Shards: 5, ReplicasPerShard: nil} + if got := omitted.TotalNodes(); got != 10 { + t.Fatalf("omitted TotalNodes() = %d, want 10 (shards*2, nil→1)", got) + } +} diff --git a/api/v1alpha2/valkeycluster_types.go b/api/v1alpha2/valkeycluster_types.go index 8a68dc69..5384c334 100644 --- a/api/v1alpha2/valkeycluster_types.go +++ b/api/v1alpha2/valkeycluster_types.go @@ -40,11 +40,17 @@ type ValkeyClusterSpec struct { // +kubebuilder:default=3 Shards int32 `json:"shards"` - // shard 당 replica 수. 기본 1 → 총 노드 = shards*(1+replicasPerShard). + // shard 당 replica 수. 총 노드 = shards*(1+replicasPerShard). + // + // *pointer* 인 이유 (defect ④): non-pointer int32 + CRD `+kubebuilder:default=1` 은 + // apiserver/admission 이 "필드 미지정" 과 "명시 0" 을 구별하지 못해 명시 0 + // (masters-only 토폴로지) 을 default 1 로 덮어쓴다. pointer 로 두면 nil(미지정) + // 과 명시 0 이 구별된다. CRD default 는 제거하고 nil→1 defaulting 은 코드 + // (GetReplicasPerShard 접근자 + mutating webhook) 에서 처리한다. // +kubebuilder:validation:Minimum=0 // +kubebuilder:validation:Maximum=5 - // +kubebuilder:default=1 - ReplicasPerShard int32 `json:"replicasPerShard"` + // +optional + ReplicasPerShard *int32 `json:"replicasPerShard,omitempty"` // +kubebuilder:default=true AutoFailover bool `json:"autoFailover,omitempty"` @@ -83,6 +89,13 @@ type ValkeyClusterSpec struct { // +optional AdditionalConfig map[string]string `json:"additionalConfig,omitempty"` + // Modules — Valkey 공식 BSD module(valkey-search/json/bloom) 또는 BYO module 로딩. + // 외부 Redis Stack(RediSearch/RedisJSON, RSALv2/SSPL)은 라이선스 비호환으로 미지원 + // (ADR-0032). Name 만 지정 시 공식 preset 자동 resolve, Image 지정 시 BYO. Valkey CR + // (ValkeySpec.Modules) 와 동일 타입/검증 — cluster 샤드 pod 전체에 동일 적용. + // +optional + Modules []ModuleSpec `json:"modules,omitempty"` + // RevisionHistoryLimit — StatefulSet rollout history 보존 개수. // +kubebuilder:validation:Minimum=0 // +optional @@ -153,4 +166,15 @@ func init() { func (v *ValkeyCluster) GetConditions() *[]metav1.Condition { return &v.Status.Conditions } func (v *ValkeyCluster) SetPhase(phase string) { v.Status.Phase = ClusterPhase(phase) } -func (s *ValkeyClusterSpec) TotalNodes() int32 { return s.Shards * (1 + s.ReplicasPerShard) } +// GetReplicasPerShard — nil(미지정) → 1, 명시값(0 포함) → 그대로. +// +// defect ④: 명시 0 은 masters-only 토폴로지이므로 보존하고, 미지정만 1 로 +// defaulting 한다. ReplicasPerShard 를 읽는 모든 곳은 이 접근자를 거친다. +func (s *ValkeyClusterSpec) GetReplicasPerShard() int32 { + if s.ReplicasPerShard == nil { + return 1 + } + return *s.ReplicasPerShard +} + +func (s *ValkeyClusterSpec) TotalNodes() int32 { return s.Shards * (1 + s.GetReplicasPerShard()) } diff --git a/api/v1alpha2/zz_generated.deepcopy.go b/api/v1alpha2/zz_generated.deepcopy.go index 833156ab..e441de16 100644 --- a/api/v1alpha2/zz_generated.deepcopy.go +++ b/api/v1alpha2/zz_generated.deepcopy.go @@ -1111,6 +1111,11 @@ func (in *ValkeyClusterList) DeepCopyObject() runtime.Object { // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ValkeyClusterSpec) DeepCopyInto(out *ValkeyClusterSpec) { *out = *in + if in.ReplicasPerShard != nil { + in, out := &in.ReplicasPerShard, &out.ReplicasPerShard + *out = new(int32) + **out = **in + } out.Version = in.Version in.Storage.DeepCopyInto(&out.Storage) in.Resources.DeepCopyInto(&out.Resources) @@ -1162,6 +1167,13 @@ func (in *ValkeyClusterSpec) DeepCopyInto(out *ValkeyClusterSpec) { (*out)[key] = val } } + if in.Modules != nil { + in, out := &in.Modules, &out.Modules + *out = make([]ModuleSpec, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } if in.RevisionHistoryLimit != nil { in, out := &in.RevisionHistoryLimit, &out.RevisionHistoryLimit *out = new(int32) diff --git a/charts/valkey-operator/crds/cache.keiailab.io_valkeyclusters.yaml b/charts/valkey-operator/crds/cache.keiailab.io_valkeyclusters.yaml index c9e11a85..dfdc7a6b 100644 --- a/charts/valkey-operator/crds/cache.keiailab.io_valkeyclusters.yaml +++ b/charts/valkey-operator/crds/cache.keiailab.io_valkeyclusters.yaml @@ -180,6 +180,36 @@ spec: 빈 값이면 상시 허용. 자정 넘김(예: "22:00-02:00") 지원. type: string type: object + modules: + description: |- + Modules — Valkey 공식 BSD module(valkey-search/json/bloom) 또는 BYO module 로딩. + 외부 Redis Stack(RediSearch/RedisJSON, RSALv2/SSPL)은 라이선스 비호환으로 미지원 + (ADR-0032). Name 만 지정 시 공식 preset 자동 resolve, Image 지정 시 BYO. Valkey CR + (ValkeySpec.Modules) 와 동일 타입/검증 — cluster 샤드 pod 전체에 동일 적용. + items: + description: "ModuleSpec — Valkey module 정의 (ADR-0032). 컨트롤러 hub(v1alpha1) + 미러.\n\n\tName 만: 공식 BSD preset (allow-list 검증 + 공식 image 자동 resolve)\n\tImage: + \ bring-your-own custom module (init container 가 .so 를 emptyDir + mount)" + properties: + image: + description: Image — custom module image (optional). 미지정 시 공식 + preset 자동 resolve. + type: string + loadModuleArgs: + description: LoadModuleArgs — `loadmodule ` 의 args + (optional). + items: + type: string + type: array + name: + description: 'Name — module 식별자 (예: "valkey-search").' + pattern: ^[a-z][a-z0-9-]+$ + type: string + required: + - name + type: object + type: array monitoring: description: MonitoringSpec — Prometheus exporter sidecar + ServiceMonitor. properties: @@ -2554,8 +2584,14 @@ spec: - enabled type: object replicasPerShard: - default: 1 - description: shard 당 replica 수. 기본 1 → 총 노드 = shards*(1+replicasPerShard). + description: |- + shard 당 replica 수. 총 노드 = shards*(1+replicasPerShard). + + *pointer* 인 이유 (defect ④): non-pointer int32 + CRD `+kubebuilder:default=1` 은 + apiserver/admission 이 "필드 미지정" 과 "명시 0" 을 구별하지 못해 명시 0 + (masters-only 토폴로지) 을 default 1 로 덮어쓴다. pointer 로 두면 nil(미지정) + 과 명시 0 이 구별된다. CRD default 는 제거하고 nil→1 defaulting 은 코드 + (GetReplicasPerShard 접근자 + mutating webhook) 에서 처리한다. format: int32 maximum: 5 minimum: 0 @@ -2828,7 +2864,6 @@ spec: - version type: object required: - - replicasPerShard - shards - version type: object @@ -3167,6 +3202,45 @@ spec: 빈 값이면 상시 허용. 자정 넘김(예: "22:00-02:00") 지원. type: string type: object + modules: + description: |- + Modules — Valkey 공식 BSD module(valkey-search/json/bloom) 또는 BYO module 로딩. + 외부 Redis Stack(RediSearch/RedisJSON, RSALv2/SSPL)은 라이선스 비호환으로 미지원 + (ADR-0032). Name 만 지정 시 공식 preset 자동 resolve, Image 지정 시 BYO. Valkey CR + (ValkeySpec.Modules) 와 동일 타입/검증 — cluster 샤드 pod 전체에 동일 적용. + items: + description: |- + ModuleSpec — Valkey module 정의 (Plan §2 D9, ADR-0032). + + 두 모드: + - Name 만 지정: Valkey 공식 module preset (예: "valkey-search", + "valkey-json", "valkey-bloom"). operator 가 alllow-list 검증 + + 공식 image 자동 resolve. + - Image 명시: bring-your-own custom module. init container 가 + 해당 image 의 /modules/.so 를 emptyDir 로 mount, valkey + container 가 `--loadmodule /modules/.so ` 로 적재. + + 보안: PSS Restricted (ADR-0036) 와 정합 — module image 가 privileged + syscall 요구 시 webhook 거부. Sonatype Trust Score 검증 권장. + properties: + image: + description: Image — custom module image (optional). 미지정 시 공식 + preset 자동 resolve. + type: string + loadModuleArgs: + description: LoadModuleArgs — `loadmodule ` 의 args + (optional). + items: + type: string + type: array + name: + description: 'Name — module 식별자 (예: "valkey-search").' + pattern: ^[a-z][a-z0-9-]+$ + type: string + required: + - name + type: object + type: array monitoring: description: MonitoringSpec — Prometheus exporter sidecar + ServiceMonitor. properties: @@ -5581,8 +5655,14 @@ spec: - enabled type: object replicasPerShard: - default: 1 - description: shard 당 replica 수. 기본 1 → 총 노드 = shards*(1+replicasPerShard). + description: |- + shard 당 replica 수. 총 노드 = shards*(1+replicasPerShard). + + *pointer* 인 이유 (defect ④): non-pointer int32 + CRD `+kubebuilder:default=1` 은 + apiserver/admission 이 "필드 미지정" 과 "명시 0" 을 구별하지 못해 명시 0 + (masters-only 토폴로지) 을 default 1 로 덮어쓴다. pointer 로 두면 nil(미지정) + 과 명시 0 이 구별된다. CRD default 는 제거하고 nil→1 defaulting 은 코드 + (GetReplicasPerShard 접근자 + mutating webhook) 에서 처리한다. format: int32 maximum: 5 minimum: 0 @@ -5799,7 +5879,6 @@ spec: - version type: object required: - - replicasPerShard - shards - version type: object diff --git a/config/crd/bases/cache.keiailab.io_valkeyclusters.yaml b/config/crd/bases/cache.keiailab.io_valkeyclusters.yaml index c9e11a85..dfdc7a6b 100644 --- a/config/crd/bases/cache.keiailab.io_valkeyclusters.yaml +++ b/config/crd/bases/cache.keiailab.io_valkeyclusters.yaml @@ -180,6 +180,36 @@ spec: 빈 값이면 상시 허용. 자정 넘김(예: "22:00-02:00") 지원. type: string type: object + modules: + description: |- + Modules — Valkey 공식 BSD module(valkey-search/json/bloom) 또는 BYO module 로딩. + 외부 Redis Stack(RediSearch/RedisJSON, RSALv2/SSPL)은 라이선스 비호환으로 미지원 + (ADR-0032). Name 만 지정 시 공식 preset 자동 resolve, Image 지정 시 BYO. Valkey CR + (ValkeySpec.Modules) 와 동일 타입/검증 — cluster 샤드 pod 전체에 동일 적용. + items: + description: "ModuleSpec — Valkey module 정의 (ADR-0032). 컨트롤러 hub(v1alpha1) + 미러.\n\n\tName 만: 공식 BSD preset (allow-list 검증 + 공식 image 자동 resolve)\n\tImage: + \ bring-your-own custom module (init container 가 .so 를 emptyDir + mount)" + properties: + image: + description: Image — custom module image (optional). 미지정 시 공식 + preset 자동 resolve. + type: string + loadModuleArgs: + description: LoadModuleArgs — `loadmodule ` 의 args + (optional). + items: + type: string + type: array + name: + description: 'Name — module 식별자 (예: "valkey-search").' + pattern: ^[a-z][a-z0-9-]+$ + type: string + required: + - name + type: object + type: array monitoring: description: MonitoringSpec — Prometheus exporter sidecar + ServiceMonitor. properties: @@ -2554,8 +2584,14 @@ spec: - enabled type: object replicasPerShard: - default: 1 - description: shard 당 replica 수. 기본 1 → 총 노드 = shards*(1+replicasPerShard). + description: |- + shard 당 replica 수. 총 노드 = shards*(1+replicasPerShard). + + *pointer* 인 이유 (defect ④): non-pointer int32 + CRD `+kubebuilder:default=1` 은 + apiserver/admission 이 "필드 미지정" 과 "명시 0" 을 구별하지 못해 명시 0 + (masters-only 토폴로지) 을 default 1 로 덮어쓴다. pointer 로 두면 nil(미지정) + 과 명시 0 이 구별된다. CRD default 는 제거하고 nil→1 defaulting 은 코드 + (GetReplicasPerShard 접근자 + mutating webhook) 에서 처리한다. format: int32 maximum: 5 minimum: 0 @@ -2828,7 +2864,6 @@ spec: - version type: object required: - - replicasPerShard - shards - version type: object @@ -3167,6 +3202,45 @@ spec: 빈 값이면 상시 허용. 자정 넘김(예: "22:00-02:00") 지원. type: string type: object + modules: + description: |- + Modules — Valkey 공식 BSD module(valkey-search/json/bloom) 또는 BYO module 로딩. + 외부 Redis Stack(RediSearch/RedisJSON, RSALv2/SSPL)은 라이선스 비호환으로 미지원 + (ADR-0032). Name 만 지정 시 공식 preset 자동 resolve, Image 지정 시 BYO. Valkey CR + (ValkeySpec.Modules) 와 동일 타입/검증 — cluster 샤드 pod 전체에 동일 적용. + items: + description: |- + ModuleSpec — Valkey module 정의 (Plan §2 D9, ADR-0032). + + 두 모드: + - Name 만 지정: Valkey 공식 module preset (예: "valkey-search", + "valkey-json", "valkey-bloom"). operator 가 alllow-list 검증 + + 공식 image 자동 resolve. + - Image 명시: bring-your-own custom module. init container 가 + 해당 image 의 /modules/.so 를 emptyDir 로 mount, valkey + container 가 `--loadmodule /modules/.so ` 로 적재. + + 보안: PSS Restricted (ADR-0036) 와 정합 — module image 가 privileged + syscall 요구 시 webhook 거부. Sonatype Trust Score 검증 권장. + properties: + image: + description: Image — custom module image (optional). 미지정 시 공식 + preset 자동 resolve. + type: string + loadModuleArgs: + description: LoadModuleArgs — `loadmodule ` 의 args + (optional). + items: + type: string + type: array + name: + description: 'Name — module 식별자 (예: "valkey-search").' + pattern: ^[a-z][a-z0-9-]+$ + type: string + required: + - name + type: object + type: array monitoring: description: MonitoringSpec — Prometheus exporter sidecar + ServiceMonitor. properties: @@ -5581,8 +5655,14 @@ spec: - enabled type: object replicasPerShard: - default: 1 - description: shard 당 replica 수. 기본 1 → 총 노드 = shards*(1+replicasPerShard). + description: |- + shard 당 replica 수. 총 노드 = shards*(1+replicasPerShard). + + *pointer* 인 이유 (defect ④): non-pointer int32 + CRD `+kubebuilder:default=1` 은 + apiserver/admission 이 "필드 미지정" 과 "명시 0" 을 구별하지 못해 명시 0 + (masters-only 토폴로지) 을 default 1 로 덮어쓴다. pointer 로 두면 nil(미지정) + 과 명시 0 이 구별된다. CRD default 는 제거하고 nil→1 defaulting 은 코드 + (GetReplicasPerShard 접근자 + mutating webhook) 에서 처리한다. format: int32 maximum: 5 minimum: 0 @@ -5799,7 +5879,6 @@ spec: - version type: object required: - - replicasPerShard - shards - version type: object diff --git a/go.mod b/go.mod index 27a3acc3..1bed94fa 100644 --- a/go.mod +++ b/go.mod @@ -24,6 +24,7 @@ require ( k8s.io/api v0.36.1 k8s.io/apimachinery v0.36.1 k8s.io/client-go v0.36.1 + k8s.io/utils v0.0.0-20260210185600-b8788abfbbc2 sigs.k8s.io/controller-runtime v0.24.1 sigs.k8s.io/yaml v1.6.0 ) @@ -138,7 +139,6 @@ require ( k8s.io/klog/v2 v2.140.0 // indirect k8s.io/kube-openapi v0.0.0-20260317180543-43fb72c5454a // indirect k8s.io/streaming v0.36.1 // indirect - k8s.io/utils v0.0.0-20260210185600-b8788abfbbc2 // indirect sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.34.0 // indirect sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect sigs.k8s.io/randfill v1.0.0 // indirect diff --git a/internal/controller/cluster_helpers_test.go b/internal/controller/cluster_helpers_test.go index 11f7a624..b2aaa9b1 100644 --- a/internal/controller/cluster_helpers_test.go +++ b/internal/controller/cluster_helpers_test.go @@ -11,12 +11,14 @@ package controller import ( "context" "reflect" + "strings" "testing" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -53,7 +55,7 @@ func TestPodAddresses_ordering(t *testing.T) { vc.Name = "vk" vc.Namespace = "ns" vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) got := podAddresses(vc) want := []string{ @@ -83,7 +85,7 @@ func TestBuildShardStatus_3x1(t *testing.T) { vc := &cachev1alpha1.ValkeyCluster{} vc.Name = "vk" vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) got := buildShardStatus(vc) if len(got) != 3 { @@ -136,7 +138,7 @@ func TestBuildShardStatus_3x2_replicaMapping(t *testing.T) { vc := &cachev1alpha1.ValkeyCluster{} vc.Name = "vk" vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 2 + vc.Spec.ReplicasPerShard = ptr.To[int32](2) got := buildShardStatus(vc) cases := [][]string{ @@ -199,6 +201,58 @@ func TestBuildShardStatusFromNodes_3x1(t *testing.T) { } } +// defect ⑥: ValkeyClusterSpec.Modules → cluster STS 에 module init-container + +// --loadmodule 적용. cluster reconcile 이 STSParams.Modules = vc.Spec.Modules 로 +// 전달하므로, cluster mode STS 빌드가 모듈을 반영하는지 검증. +func TestClusterSTS_modulesWired(t *testing.T) { + vc := &cachev1alpha1.ValkeyCluster{} + vc.Name = "vk" + vc.Namespace = "ns" + vc.Spec.Shards = 3 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) + vc.Spec.Modules = []cachev1alpha1.ModuleSpec{{Name: "valkey-search"}} + + sts := resources.BuildStatefulSet(resources.STSParams{ + CRName: vc.Name, + Namespace: vc.Namespace, + Replicas: vc.Spec.TotalNodes(), + Image: "valkey:9", + ClusterMode: true, + Modules: vc.Spec.Modules, // reconcile 의 STSParams 와 동일 wiring. + }) + ps := sts.Spec.Template.Spec + if len(ps.InitContainers) != 1 { + t.Fatalf("cluster STS module init-container 1 기대, got %d", len(ps.InitContainers)) + } + // cluster mode 에서는 clusterAnnounceCommand(#298) 가 args 를 sh -c Command + // 문자열로 접어 넣으므로 --loadmodule 은 Command 또는 Args 양쪽에 있을 수 있다. + hasLoad := false + for _, tok := range append(append([]string{}, ps.Containers[0].Command...), ps.Containers[0].Args...) { + if strings.Contains(tok, "--loadmodule") { + hasLoad = true + } + } + if !hasLoad { + t.Fatalf("cluster STS valkey container 에 --loadmodule 기대: cmd=%v args=%v", ps.Containers[0].Command, ps.Containers[0].Args) + } +} + +// defect ④: masters-only — TotalNodes() 가 shards 만 (replica 0). +func TestTotalNodes_mastersOnly(t *testing.T) { + vc := &cachev1alpha1.ValkeyCluster{} + vc.Spec.Shards = 3 + vc.Spec.ReplicasPerShard = ptr.To[int32](0) + if got := vc.Spec.TotalNodes(); got != 3 { + t.Fatalf("masters-only TotalNodes(): got %d want 3", got) + } + // podAddresses 도 shards 개 (replica 주소 없음). + vc.Name = "vk" + vc.Namespace = "ns" + if got := podAddresses(vc); len(got) != 3 { + t.Fatalf("masters-only podAddresses: got %d want 3", len(got)) + } +} + // decidePhase — phase 결정 우선순위 행렬. func TestDecidePhase_matrix(t *testing.T) { mkVC := func(spec, status string) *cachev1alpha1.ValkeyCluster { diff --git a/internal/controller/cluster_reintegrate.go b/internal/controller/cluster_reintegrate.go new file mode 100644 index 00000000..e7cd2bb0 --- /dev/null +++ b/internal/controller/cluster_reintegrate.go @@ -0,0 +1,459 @@ +/* +Copyright 2026 Keiailab. + +Licensed under the MIT License. See the LICENSE file for details. +*/ + +package controller + +import ( + "context" + "fmt" + "strings" + + "github.com/redis/go-redis/v9" + corev1 "k8s.io/api/core/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/log" + + cachev1alpha1 "github.com/keiailab/valkey-operator/api/v1alpha1" + "github.com/keiailab/valkey-operator/internal/observability" + "github.com/keiailab/valkey-operator/internal/resources" + vk "github.com/keiailab/valkey-operator/internal/valkey" +) + +// 결함 ③ — cluster 멤버십 자가복구. +// +// 증상: 노드가 재시작 후 새 node id 를 얻거나 nodes.conf 를 잃어 멤버십에서 이탈하면 +// (CLUSTER NODES 에 안 보임), operator 가 다시 합류시키지 못한다. 기존 health gate 는 +// slot 레벨(cluster_state:ok)만 검사하고 *replica 수 / 멤버십* 은 보지 않아 idle 상태로 +// shard 가 desired 보다 적은 replica 로 운영된다. +// +// 본 path 는: +// 1. detectReintegration: CLUSTER NODES + desired 토폴로지를 비교해 *어떤 pod 가 +// 멤버가 아닌지* + *각 replica 가 어느 primary 를 따라야 하는지* 결정 (순수 함수). +// 2. reintegratePods: 각 누락 pod 에 대해 CLUSTER MEET (현재 pod IP) → 필요 시 +// CLUSTER REPLICATE . 기존 ensureClusterMeet / CreateCluster 의 +// 멤버십 helper 와 동일한 client 추상화 위에서 동작하며 멱등하다 (이미 멤버이며 +// 올바른 master 를 가리키는 노드는 건드리지 않는다). + +// reintegrationAction — 단일 pod 의 재합류 계획. +type reintegrationAction struct { + // Ordinal — STS pod ordinal (0..total-1). + Ordinal int + // IsReplica — true 면 MEET 후 ReplicateTargetOrdinal 의 primary 를 따르게 한다. + IsReplica bool + // ReplicateTargetOrdinal — replica 가 따라야 할 primary 의 pod ordinal (desired + // 토폴로지 기준). IsReplica=false (primary 자신 이탈) 면 의미 없음. + ReplicateTargetOrdinal int +} + +// observedMember — detectReintegration 의 입력. 관측된 cluster 멤버십을 +// pod ordinal 기준으로 정규화한 것. controller 가 CLUSTER NODES + Pod IP 매핑으로 채운다. +type observedMember struct { + // IsMember — 해당 ordinal 의 pod 가 CLUSTER NODES 에 (자신의 IP 로) 보이는가. + IsMember bool + // MasterOrdinal — replica 인 경우 현재 따르고 있는 primary 의 ordinal. 모르면 -1. + MasterOrdinal int +} + +// desiredMasterOrdinal — replica pod ordinal → 따라야 할 primary pod ordinal. +// +// CreateCluster 의 배치 규칙과 동일: replica index j (= ordinal - shards) 는 +// primary (j % shards) = pod ordinal (j % shards) 를 따른다. +func desiredMasterOrdinal(ordinal, shards int) int { + j := ordinal - shards + return j % shards +} + +// detectReintegration — 순수 결정 로직. desired 토폴로지(shards, replicasPerShard)와 +// 관측된 멤버십(ordinal→observedMember)을 비교해 재합류가 필요한 pod 들의 계획을 만든다. +// +// 규칙: +// - primary pod (ordinal < shards) 가 멤버 아님 → MEET (replica 아님). +// - replica pod (ordinal >= shards) 가 멤버 아님 → MEET + REPLICATE(desired master). +// - replica pod 가 멤버이지만 *틀린 master* 를 따름 → REPLICATE(desired master). +// - 이미 올바른 멤버 → skip (멱등). +// +// 반환 순서: primary 먼저, 그다음 replica (master 가 먼저 멤버여야 replica 가 붙는다). +func detectReintegration(shards, replicasPerShard int, observed map[int]observedMember) []reintegrationAction { + if shards <= 0 { + return nil + } + total := shards * (1 + replicasPerShard) + var primaries, replicas []reintegrationAction + + for ord := 0; ord < total; ord++ { + m, seen := observed[ord] + isReplica := ord >= shards + if !isReplica { + // primary — 멤버가 아니면 MEET 필요. + if !seen || !m.IsMember { + primaries = append(primaries, reintegrationAction{Ordinal: ord, IsReplica: false}) + } + continue + } + wantMaster := desiredMasterOrdinal(ord, shards) + switch { + case !seen || !m.IsMember: + // 멤버 이탈 — MEET + REPLICATE. + replicas = append(replicas, reintegrationAction{ + Ordinal: ord, IsReplica: true, ReplicateTargetOrdinal: wantMaster, + }) + case m.MasterOrdinal != wantMaster: + // 멤버지만 틀린 master (또는 master 모름) → REPLICATE 로 교정. + replicas = append(replicas, reintegrationAction{ + Ordinal: ord, IsReplica: true, ReplicateTargetOrdinal: wantMaster, + }) + } + } + return append(primaries, replicas...) +} + +// buildObservedMembers — CLUSTER NODES 결과(nodes) + pod ordinal→IP:port 매핑으로 +// observedMember 맵을 만든다. desired total 만큼의 ordinal 을 채운다. +// +// addrByOrdinal[ord] = "ip:port" (pod 의 현재 IP). 비어 있으면 IsMember=false. +// nodeIDByOrdinal 은 부수적으로 ordinal→nodeID (멤버인 경우) 매핑을 채워 반환한다 — +// REPLICATE 시 target master 의 node id 가 필요하다. +func buildObservedMembers( + shards, replicasPerShard int, + nodes []vk.NodeView, + addrByOrdinal map[int]string, +) (map[int]observedMember, map[int]string) { + total := shards * (1 + replicasPerShard) + + // addr → NodeView 인덱스. + byAddr := make(map[string]*vk.NodeView, len(nodes)) + for i := range nodes { + byAddr[nodes[i].Addr] = &nodes[i] + } + // nodeID → ordinal (master 매핑 역참조용). + idToOrdinal := make(map[string]int, total) + for ord := 0; ord < total; ord++ { + if addr := addrByOrdinal[ord]; addr != "" { + if nv := byAddr[addr]; nv != nil { + idToOrdinal[nv.ID] = ord + } + } + } + + observed := make(map[int]observedMember, total) + nodeIDByOrdinal := make(map[int]string, total) + for ord := 0; ord < total; ord++ { + addr := addrByOrdinal[ord] + nv := byAddr[addr] + if addr == "" || nv == nil { + observed[ord] = observedMember{IsMember: false, MasterOrdinal: -1} + continue + } + nodeIDByOrdinal[ord] = nv.ID + masterOrd := -1 + if nv.IsReplica() && nv.MasterID != "" && nv.MasterID != "-" { + if mo, ok := idToOrdinal[nv.MasterID]; ok { + masterOrd = mo + } + } + observed[ord] = observedMember{IsMember: true, MasterOrdinal: masterOrd} + } + return observed, nodeIDByOrdinal +} + +// ensureClusterMembership — 결함 ③ 자가복구 진입점. allReady && cluster_state=ok 인 +// 정상 cluster 에서도 *멤버십 / replica 수* 를 추가 검사해, desired 토폴로지에 비해 +// 누락된 멤버를 재합류시킨다. +// +// 멱등: detectReintegration 이 이미 올바른 멤버는 제외하므로 반복 호출이 안전하다. +// 누락이 없으면 즉시 반환(no-op). +func (r *ValkeyClusterReconciler) ensureClusterMembership( + ctx context.Context, vc *cachev1alpha1.ValkeyCluster, password string, +) (int, error) { + ctx, span := observability.StartCallSpan(ctx, "ValkeyCluster/EnsureClusterMembership") + defer span.End() + logger := log.FromContext(ctx) + + shards := int(vc.Spec.Shards) + rps := int(vc.Spec.GetReplicasPerShard()) + if shards <= 0 { + return 0, nil + } + + info, nodes, err := r.queryAnyNode(ctx, vc, password) + if err != nil || info == nil { + // cluster 에 도달 불가 — bootstrap path 가 별도로 처리. 여기선 no-op. + return 0, err + } + // ordinal → 현재 pod IP:port 매핑 (CLUSTER MEET 은 IP 를 요구, announce-ip/PR #298 정합). + addrByOrdinal := r.podIPByOrdinal(ctx, vc) + + observed, nodeIDByOrdinal := buildObservedMembers(shards, rps, nodes, addrByOrdinal) + + // 명백히 죽은 ghost(fail,noaddr / orphan) 정리 — 새 node id 로 재합류한 노드의 + // 옛 id 가 gossip 에 fail,noaddr 로 남아 cluster_known_nodes 가 부풀려지는 것을 + // 방지한다. 이 gated path(allReady && state=ok)에서만 실행해 bootstrap 와 레이스 없음. + // 보수적 — detectStaleNodes 가 myself / 현 멤버 / handshake 를 모두 제외한다. + expectedAddrs := make(map[string]bool, len(addrByOrdinal)) + for _, addr := range addrByOrdinal { + if addr != "" { + expectedAddrs[addr] = true + } + } + if staleIDs := detectStaleNodes(nodes, expectedAddrs); len(staleIDs) > 0 { + logger.Info("Stale/ghost cluster nodes detected; forgetting", + "ghostCount", len(staleIDs), + "knownNodes", info.KnownNodes) + if _, fErr := r.forgetStaleNodes(ctx, vc, password, staleIDs, addrByOrdinal, nodeIDByOrdinal); fErr != nil { + // best-effort — forget 실패는 재합류를 막지 않는다. 다음 reconcile 재시도. + logger.Error(fErr, "Stale node forget pending — will retry") + } + } + + actions := detectReintegration(shards, rps, observed) + if len(actions) == 0 { + return 0, nil + } + + logger.Info("ValkeyCluster membership drift detected; re-integrating", + "missingOrReplicaDrift", len(actions), + "clusterState", info.State, + "knownNodes", info.KnownNodes) + + return r.reintegratePods(ctx, vc, password, actions, addrByOrdinal, nodeIDByOrdinal) +} + +// reintegratePods — actions 를 실제 CLUSTER MEET / REPLICATE 명령으로 실행한다. +// +// MEET 은 멤버십을 가진 임의의 healthy 노드(seed)에서 발행한다. REPLICATE 는 대상 +// pod 자신에게 발행하며, target master 의 *현재 node id* 가 필요하다 — gossip 수렴 +// 직후라 모를 수 있으므로 replicateWithRetry 패턴(vk 패키지)을 재사용한다. +func (r *ValkeyClusterReconciler) reintegratePods( + ctx context.Context, + vc *cachev1alpha1.ValkeyCluster, + password string, + actions []reintegrationAction, + addrByOrdinal map[int]string, + nodeIDByOrdinal map[int]string, +) (int, error) { + tlsCfg, err := r.tlsConfigForCluster(ctx, vc) + if err != nil { + return 0, fmt.Errorf("tls config: %w", err) + } + dial := func(addr string) *redis.Client { return dialPod(addr, password, tlsCfg) } + + // seed — MEET 발행 노드. 첫 멤버(가급적 primary)를 고른다. + seedAddr := r.firstMemberAddr(addrByOrdinal, nodeIDByOrdinal, int(vc.Spec.Shards)) + if seedAddr == "" { + return 0, fmt.Errorf("no healthy seed member to issue CLUSTER MEET") + } + + var done int + for _, a := range actions { + addr := addrByOrdinal[a.Ordinal] + if addr == "" { + // pod IP 미상 (아직 스케줄 전 등) — 다음 reconcile 재시도. + continue + } + // 1) MEET (seed → 누락 노드). 이미 멤버면 valkey 가 no-op 처리. + if err := vk.MeetNode(ctx, dial, seedAddr, addr); err != nil { + return done, fmt.Errorf("cluster meet ordinal %d (%s): %w", a.Ordinal, addr, err) + } + if !a.IsReplica { + done++ + continue + } + // 2) REPLICATE — target master 의 현재 node id. + targetID := nodeIDByOrdinal[a.ReplicateTargetOrdinal] + if targetID == "" { + // master 자체가 아직 멤버가 아니거나 id 미상 — 다음 reconcile 에 수렴. + // (actions 는 primary 를 먼저 처리하므로 통상 동일 reconcile 내에서 해소.) + continue + } + if err := vk.ReplicateTo(ctx, dial, addr, targetID); err != nil { + return done, fmt.Errorf("cluster replicate ordinal %d → master ordinal %d: %w", + a.Ordinal, a.ReplicateTargetOrdinal, err) + } + done++ + } + return done, nil +} + +// detectStaleNodes — 순수 결정 로직. CLUSTER NODES 스냅샷에서 *명백히 죽은 ghost* +// node id 들을 골라낸다 (CLUSTER FORGET 대상). +// +// 배경: 노드가 CLUSTER RESET HARD 등으로 새 node id 로 재합류하면, gossip 테이블에 +// 옛/죽은 node id 가 `fail,noaddr` (또는 `slave,fail,noaddr`) ghost 로 남아 +// cluster_known_nodes 가 valkey 의 느린 auto-eviction 전까지 부풀려진다. 본 함수는 +// 그런 ghost 만 보수적으로 골라 forget 대상으로 반환한다. +// +// 규칙 (보수적 · 멱등): +// - myself 는 절대 forget 하지 않는다. +// - 다음을 모두 만족하는 노드만 forget: +// 1) `fail` flag 가 있다 (gossip 이 죽었다고 합의). +// 2) `noaddr` flag 가 있거나, addr 가 *현재 기대 pod* 중 어느 것과도 매칭되지 않는다 +// (= 이번 incarnation 에 속하지 않는 orphan). +// - addr 가 현재 기대 pod 와 매칭되면 (정당한 현 멤버) — 일시적 fail 이어도 건드리지 않는다. +// - `handshake` 노드는 *제외*: 아직 수렴 중일 수 있어 (MEET 직후) 성급히 forget 하면 +// 방금 재합류시킨 노드를 도로 쫓아낼 수 있다. handshake 는 valkey 가 자체 timeout 으로 정리. +// +// expectedAddrs: 현재 기대되는 pod 주소 집합 ("ip:port"). buildObservedMembers 가 쓰는 +// addrByOrdinal 의 값들과 동일. +func detectStaleNodes(nodes []vk.NodeView, expectedAddrs map[string]bool) []string { + var stale []string + for i := range nodes { + n := &nodes[i] + if n.Flags["myself"] { + continue + } + if n.Flags["handshake"] { + // 수렴 중 — valkey 자체 timeout 에 위임. + continue + } + if !n.Flags["fail"] { + // gossip 이 죽었다고 합의하지 않음 — 정당한 멤버이거나 일시 장애. + continue + } + if n.Addr != "" && expectedAddrs[n.Addr] { + // 현 incarnation 의 정당한 pod — fail 이어도 forget 하지 않는다. + continue + } + // 여기 도달 = fail + (noaddr 이거나 addr 가 현재 기대 pod 와 매칭 안 됨). + // 둘 다 이번 incarnation 에 속하지 않는 명백한 ghost/orphan. + if n.ID == "" { + continue + } + stale = append(stale, n.ID) + } + return stale +} + +// forgetStaleNodes — detectStaleNodes 가 고른 ghost id 들을 *모든 healthy primary* 에서 +// CLUSTER FORGET 한다 (scale-in / gracefulClusterTeardown 의 forget 패턴 미러링). +// +// forget 은 gossip 전파를 위해 살아있는 모든 노드에서 발행해야 효과가 지속된다 — +// 한 노드에서만 forget 하면 다른 노드의 gossip 이 다시 알려준다. 여기선 현재 멤버인 +// 모든 ordinal(primary 우선 포함)에 발행한다. best-effort — 일부 실패는 다음 reconcile 재시도. +func (r *ValkeyClusterReconciler) forgetStaleNodes( + ctx context.Context, + vc *cachev1alpha1.ValkeyCluster, + password string, + staleIDs []string, + addrByOrdinal map[int]string, + nodeIDByOrdinal map[int]string, +) (int, error) { + if len(staleIDs) == 0 { + return 0, nil + } + tlsCfg, err := r.tlsConfigForCluster(ctx, vc) + if err != nil { + return 0, fmt.Errorf("tls config: %w", err) + } + // forget 대상 집합 (live 멤버 자신을 실수로 forget 하지 않도록 한 번 더 가드). + stale := make(map[string]bool, len(staleIDs)) + liveIDs := make(map[string]bool, len(nodeIDByOrdinal)) + for _, id := range nodeIDByOrdinal { + liveIDs[id] = true + } + for _, id := range staleIDs { + if id != "" && !liveIDs[id] { + stale[id] = true + } + } + if len(stale) == 0 { + return 0, nil + } + + logger := log.FromContext(ctx) + var forgotten int + // 현재 멤버인 모든 ordinal 에서 발행. + for ord, id := range nodeIDByOrdinal { + if id == "" { + continue + } + addr := addrByOrdinal[ord] + if addr == "" { + continue + } + c := dialPod(addr, password, tlsCfg) + for sid := range stale { + if err := vk.ForgetNode(ctx, c, sid); err != nil { + // 이미 forget 됐거나(Unknown node) 일시 에러 — best-effort, 계속. + logger.V(1).Info("CLUSTER FORGET attempt failed (best-effort)", + "node", sid, "via", addr, "error", err.Error()) + continue + } + forgotten++ + } + _ = c.Close() + } + if forgotten > 0 { + logger.Info("Forgot stale/ghost cluster nodes", + "ghostIDs", len(stale), "forgetCallsSucceeded", forgotten) + } + return forgotten, nil +} + +// firstMemberAddr — MEET 발행에 쓸 seed 주소. primary ordinal 중 멤버를 우선, +// 없으면 멤버인 아무 ordinal. +func (r *ValkeyClusterReconciler) firstMemberAddr( + addrByOrdinal, nodeIDByOrdinal map[int]string, shards int, +) string { + for ord := 0; ord < shards; ord++ { + if nodeIDByOrdinal[ord] != "" { + return addrByOrdinal[ord] + } + } + for ord, id := range nodeIDByOrdinal { + if id != "" { + return addrByOrdinal[ord] + } + } + return "" +} + +// podIPByOrdinal — STS pod 들을 조회해 ordinal → "podIP:port" 매핑을 만든다. +// CLUSTER MEET 은 hostname 을 거부하고 IP 만 받으므로 FQDN 이 아닌 *현재 pod IP* 를 +// 쓴다 (재시작 후 IP 변경 + announce-ip, PR #298 와 정합). IP 미상 pod 는 생략. +func (r *ValkeyClusterReconciler) podIPByOrdinal(ctx context.Context, vc *cachev1alpha1.ValkeyCluster) map[int]string { + pods := &corev1.PodList{} + selector := client.MatchingLabels(resources.SelectorLabels(vc.Name)) + out := make(map[int]string) + if err := r.List(ctx, pods, client.InNamespace(vc.Namespace), selector); err != nil { + return out + } + port := resources.PortClient + if vc.Spec.TLS != nil && vc.Spec.TLS.Enabled { + port = resources.PortTLS + } + prefix := resources.StatefulSetName(vc.Name) + "-" + for i := range pods.Items { + p := &pods.Items[i] + if p.Status.PodIP == "" { + continue + } + ord, ok := ordinalFromPodName(p.Name, prefix) + if !ok { + continue + } + out[ord] = fmt.Sprintf("%s:%d", p.Status.PodIP, port) + } + return out +} + +// ordinalFromPodName — "vk-3" → 3. prefix("vk-") 이후 정수 파싱. +func ordinalFromPodName(name, prefix string) (int, bool) { + if !strings.HasPrefix(name, prefix) { + return 0, false + } + suffix := name[len(prefix):] + n := 0 + if suffix == "" { + return 0, false + } + for _, c := range suffix { + if c < '0' || c > '9' { + return 0, false + } + n = n*10 + int(c-'0') + } + return n, true +} diff --git a/internal/controller/cluster_reintegrate_test.go b/internal/controller/cluster_reintegrate_test.go new file mode 100644 index 00000000..314aaa3d --- /dev/null +++ b/internal/controller/cluster_reintegrate_test.go @@ -0,0 +1,291 @@ +/* +Copyright 2026 Keiailab. + +Licensed under the MIT License. See the LICENSE file for details. +*/ + +// 결함 ③ — 멤버십 재합류 *순수 결정 로직* 단위테스트. +// 실제 Valkey gossip 없이 (envtest 불가 영역) 어떤 pod 가 재합류 대상인지 + +// 각 replica 가 어느 master 를 따라야 하는지의 알고리즘 회귀를 차단한다. +package controller + +import ( + "reflect" + "testing" + + vk "github.com/keiailab/valkey-operator/internal/valkey" +) + +func TestDesiredMasterOrdinal(t *testing.T) { + // 3 shards × 2 rps: replica ordinal 3..8 → master ordinal. + // ord 3 (j=0) → 0, ord 4 (j=1) → 1, ord 5 (j=2) → 2 + // ord 6 (j=3) → 0, ord 7 (j=4) → 1, ord 8 (j=5) → 2 + cases := map[int]int{3: 0, 4: 1, 5: 2, 6: 0, 7: 1, 8: 2} + for ord, want := range cases { + if got := desiredMasterOrdinal(ord, 3); got != want { + t.Errorf("desiredMasterOrdinal(%d,3)=%d want %d", ord, got, want) + } + } +} + +func TestDetectReintegration_allHealthy_noop(t *testing.T) { + // 3×1, 모두 멤버 + replica 가 올바른 master. + observed := map[int]observedMember{ + 0: {IsMember: true, MasterOrdinal: -1}, + 1: {IsMember: true, MasterOrdinal: -1}, + 2: {IsMember: true, MasterOrdinal: -1}, + 3: {IsMember: true, MasterOrdinal: 0}, + 4: {IsMember: true, MasterOrdinal: 1}, + 5: {IsMember: true, MasterOrdinal: 2}, + } + if got := detectReintegration(3, 1, observed); len(got) != 0 { + t.Fatalf("healthy cluster should yield no actions, got %v", got) + } +} + +func TestDetectReintegration_missingReplica(t *testing.T) { + // 3×1, replica ord 4 가 멤버에서 이탈 (재시작 후 새 id). + observed := map[int]observedMember{ + 0: {IsMember: true, MasterOrdinal: -1}, + 1: {IsMember: true, MasterOrdinal: -1}, + 2: {IsMember: true, MasterOrdinal: -1}, + 3: {IsMember: true, MasterOrdinal: 0}, + 4: {IsMember: false, MasterOrdinal: -1}, // 이탈. + 5: {IsMember: true, MasterOrdinal: 2}, + } + got := detectReintegration(3, 1, observed) + want := []reintegrationAction{ + {Ordinal: 4, IsReplica: true, ReplicateTargetOrdinal: 1}, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %+v want %+v", got, want) + } +} + +func TestDetectReintegration_replicaWrongMaster(t *testing.T) { + // replica ord 5 가 멤버지만 틀린 master(0)를 따름 → 2 로 교정. + observed := map[int]observedMember{ + 0: {IsMember: true, MasterOrdinal: -1}, + 1: {IsMember: true, MasterOrdinal: -1}, + 2: {IsMember: true, MasterOrdinal: -1}, + 3: {IsMember: true, MasterOrdinal: 0}, + 4: {IsMember: true, MasterOrdinal: 1}, + 5: {IsMember: true, MasterOrdinal: 0}, // 틀림 — 2 여야. + } + got := detectReintegration(3, 1, observed) + want := []reintegrationAction{ + {Ordinal: 5, IsReplica: true, ReplicateTargetOrdinal: 2}, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %+v want %+v", got, want) + } +} + +func TestDetectReintegration_missingPrimaryOrderedFirst(t *testing.T) { + // primary ord 1 이탈 + replica ord 4 (그 primary 의 replica) 이탈. + // primary 가 먼저 MEET 되도록 actions 순서가 primary→replica 여야 한다. + observed := map[int]observedMember{ + 0: {IsMember: true, MasterOrdinal: -1}, + 1: {IsMember: false, MasterOrdinal: -1}, + 2: {IsMember: true, MasterOrdinal: -1}, + 3: {IsMember: true, MasterOrdinal: 0}, + 4: {IsMember: false, MasterOrdinal: -1}, + 5: {IsMember: true, MasterOrdinal: 2}, + } + got := detectReintegration(3, 1, observed) + want := []reintegrationAction{ + {Ordinal: 1, IsReplica: false, ReplicateTargetOrdinal: 0}, + {Ordinal: 4, IsReplica: true, ReplicateTargetOrdinal: 1}, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %+v want %+v", got, want) + } + // 첫 액션은 반드시 primary (replica 가 붙기 전 master 가 멤버여야 함). + if got[0].IsReplica { + t.Fatal("primary action must come before replica action") + } +} + +func TestDetectReintegration_mastersOnly_noReplicaActions(t *testing.T) { + // rps=0 — replica 자체가 없으므로 누락 primary 만 MEET, replica 액션 없음. + observed := map[int]observedMember{ + 0: {IsMember: true, MasterOrdinal: -1}, + 1: {IsMember: false, MasterOrdinal: -1}, + 2: {IsMember: true, MasterOrdinal: -1}, + } + got := detectReintegration(3, 0, observed) + want := []reintegrationAction{ + {Ordinal: 1, IsReplica: false, ReplicateTargetOrdinal: 0}, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %+v want %+v", got, want) + } +} + +func TestBuildObservedMembers_mapsMasterOrdinal(t *testing.T) { + // 3×1, 모든 노드 멤버. NODES 가 replica 의 MasterID 로 master ordinal 을 역참조. + addrByOrdinal := map[int]string{ + 0: "10.0.0.1:6379", 1: "10.0.0.2:6379", 2: "10.0.0.3:6379", + 3: "10.0.0.4:6379", 4: "10.0.0.5:6379", 5: "10.0.0.6:6379", + } + nodes := []vk.NodeView{ + {ID: "p0", Addr: "10.0.0.1:6379", Flags: map[string]bool{"master": true}}, + {ID: "p1", Addr: "10.0.0.2:6379", Flags: map[string]bool{"master": true}}, + {ID: "p2", Addr: "10.0.0.3:6379", Flags: map[string]bool{"master": true}}, + {ID: "r0", Addr: "10.0.0.4:6379", Flags: map[string]bool{"slave": true}, MasterID: "p0"}, + {ID: "r1", Addr: "10.0.0.5:6379", Flags: map[string]bool{"slave": true}, MasterID: "p1"}, + {ID: "r2", Addr: "10.0.0.6:6379", Flags: map[string]bool{"slave": true}, MasterID: "p2"}, + } + observed, ids := buildObservedMembers(3, 1, nodes, addrByOrdinal) + + // replica ordinal 3 → master ordinal 0, 4 → 1, 5 → 2. + for ord, wantMaster := range map[int]int{3: 0, 4: 1, 5: 2} { + if !observed[ord].IsMember { + t.Errorf("ordinal %d should be member", ord) + } + if observed[ord].MasterOrdinal != wantMaster { + t.Errorf("ordinal %d masterOrdinal=%d want %d", ord, observed[ord].MasterOrdinal, wantMaster) + } + } + if ids[0] != "p0" || ids[5] != "r2" { + t.Errorf("nodeIDByOrdinal mapping wrong: %v", ids) + } + // 멤버 + 올바른 master → 재합류 액션 0. + if acts := detectReintegration(3, 1, observed); len(acts) != 0 { + t.Errorf("healthy → no actions, got %v", acts) + } +} + +func TestBuildObservedMembers_missingPodIsNonMember(t *testing.T) { + // ordinal 4 의 pod IP 미상 (재시작 직후) → 비멤버. + addrByOrdinal := map[int]string{ + 0: "10.0.0.1:6379", 1: "10.0.0.2:6379", 2: "10.0.0.3:6379", + 3: "10.0.0.4:6379", 5: "10.0.0.6:6379", + } + nodes := []vk.NodeView{ + {ID: "p0", Addr: "10.0.0.1:6379", Flags: map[string]bool{"master": true}}, + {ID: "p1", Addr: "10.0.0.2:6379", Flags: map[string]bool{"master": true}}, + {ID: "p2", Addr: "10.0.0.3:6379", Flags: map[string]bool{"master": true}}, + {ID: "r0", Addr: "10.0.0.4:6379", Flags: map[string]bool{"slave": true}, MasterID: "p0"}, + {ID: "r2", Addr: "10.0.0.6:6379", Flags: map[string]bool{"slave": true}, MasterID: "p2"}, + } + observed, _ := buildObservedMembers(3, 1, nodes, addrByOrdinal) + if observed[4].IsMember { + t.Fatal("ordinal 4 (no pod IP) must be non-member") + } + got := detectReintegration(3, 1, observed) + want := []reintegrationAction{{Ordinal: 4, IsReplica: true, ReplicateTargetOrdinal: 1}} + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %+v want %+v", got, want) + } +} + +// expectedAddrsFromMap — addrByOrdinal 의 값 집합 (detectStaleNodes 입력). +func expectedAddrsFromMap(m map[int]string) map[string]bool { + out := make(map[string]bool, len(m)) + for _, a := range m { + if a != "" { + out[a] = true + } + } + return out +} + +func TestDetectStaleNodes_ghostForgotten(t *testing.T) { + // 3 healthy 멤버 + replica 가 CLUSTER RESET HARD 후 새 id 로 재합류 → + // 옛 id "deadghost" 가 slave,fail,noaddr 로 gossip 에 남음. 정확히 그 id 만 forget. + expected := expectedAddrsFromMap(map[int]string{ + 0: "10.0.0.1:6379", 1: "10.0.0.2:6379", 2: "10.0.0.3:6379", + 3: "10.0.0.9:6379", // 새 incarnation 의 replica (새 id). + }) + nodes := []vk.NodeView{ + {ID: "p0", Addr: "10.0.0.1:6379", Flags: map[string]bool{"myself": true, "master": true}}, + {ID: "p1", Addr: "10.0.0.2:6379", Flags: map[string]bool{"master": true}}, + {ID: "p2", Addr: "10.0.0.3:6379", Flags: map[string]bool{"master": true}}, + {ID: "rnew", Addr: "10.0.0.9:6379", Flags: map[string]bool{"slave": true}, MasterID: "p0"}, + // ghost: 옛 replica id, addr 사라짐. + {ID: "deadghost", Addr: "", Flags: map[string]bool{"slave": true, "fail": true, "noaddr": true}}, + } + got := detectStaleNodes(nodes, expected) + want := []string{"deadghost"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %v want %v", got, want) + } +} + +func TestDetectStaleNodes_healthyOnly_noop(t *testing.T) { + expected := expectedAddrsFromMap(map[int]string{ + 0: "10.0.0.1:6379", 1: "10.0.0.2:6379", 2: "10.0.0.3:6379", + }) + nodes := []vk.NodeView{ + {ID: "p0", Addr: "10.0.0.1:6379", Flags: map[string]bool{"myself": true, "master": true}}, + {ID: "p1", Addr: "10.0.0.2:6379", Flags: map[string]bool{"master": true}}, + {ID: "p2", Addr: "10.0.0.3:6379", Flags: map[string]bool{"master": true}}, + } + if got := detectStaleNodes(nodes, expected); len(got) != 0 { + t.Fatalf("healthy-only snapshot must forget nothing, got %v", got) + } +} + +func TestDetectStaleNodes_neverForgetMyselfOrCurrentPod(t *testing.T) { + // myself 가 fail,noaddr 로 잘못 표기돼도(이론상) 절대 forget 안 함. + // 현재 기대 pod 가 일시적 fail 이어도(주소는 그대로) forget 안 함. + expected := expectedAddrsFromMap(map[int]string{ + 0: "10.0.0.1:6379", 1: "10.0.0.2:6379", + }) + nodes := []vk.NodeView{ + {ID: "p0", Addr: "10.0.0.1:6379", Flags: map[string]bool{"myself": true, "master": true, "fail": true, "noaddr": true}}, + // 현 멤버지만 일시 fail (주소 유지) — 정당한 pod, forget 금지. + {ID: "p1", Addr: "10.0.0.2:6379", Flags: map[string]bool{"master": true, "fail": true}}, + } + if got := detectStaleNodes(nodes, expected); len(got) != 0 { + t.Fatalf("must never forget myself or current-pod-backed node, got %v", got) + } +} + +func TestDetectStaleNodes_handshakeExcluded(t *testing.T) { + // 방금 MEET 한 노드가 handshake 중 (아직 수렴) — forget 하면 도로 쫓아냄. 제외. + expected := expectedAddrsFromMap(map[int]string{0: "10.0.0.1:6379"}) + nodes := []vk.NodeView{ + {ID: "p0", Addr: "10.0.0.1:6379", Flags: map[string]bool{"myself": true, "master": true}}, + {ID: "hs", Addr: "10.0.0.7:6379", Flags: map[string]bool{"handshake": true, "fail": true}}, + } + if got := detectStaleNodes(nodes, expected); len(got) != 0 { + t.Fatalf("handshake node must be excluded, got %v", got) + } +} + +func TestDetectStaleNodes_orphanWithAddrNotInExpected(t *testing.T) { + // fail 인데 addr 가 남아있지만 현재 기대 pod 어디에도 없음 (옛 incarnation IP) → forget. + expected := expectedAddrsFromMap(map[int]string{0: "10.0.0.1:6379"}) + nodes := []vk.NodeView{ + {ID: "p0", Addr: "10.0.0.1:6379", Flags: map[string]bool{"myself": true, "master": true}}, + {ID: "orphan", Addr: "10.0.99.99:6379", Flags: map[string]bool{"slave": true, "fail": true}}, + } + got := detectStaleNodes(nodes, expected) + want := []string{"orphan"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %v want %v", got, want) + } +} + +func TestOrdinalFromPodName(t *testing.T) { + cases := []struct { + name, prefix string + want int + ok bool + }{ + {"vk-0", "vk-", 0, true}, + {"vk-3", "vk-", 3, true}, + {"vk-12", "vk-", 12, true}, + {"vk-", "vk-", 0, false}, + {"other-1", "vk-", 0, false}, + {"vk-x", "vk-", 0, false}, + } + for _, c := range cases { + got, ok := ordinalFromPodName(c.name, c.prefix) + if ok != c.ok || (ok && got != c.want) { + t.Errorf("ordinalFromPodName(%q,%q)=(%d,%v) want (%d,%v)", c.name, c.prefix, got, ok, c.want, c.ok) + } + } +} diff --git a/internal/controller/multipod_volumesnapshot_guard_test.go b/internal/controller/multipod_volumesnapshot_guard_test.go index 1ab7e1b8..a7df1a82 100644 --- a/internal/controller/multipod_volumesnapshot_guard_test.go +++ b/internal/controller/multipod_volumesnapshot_guard_test.go @@ -13,6 +13,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/tools/events" + "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -40,7 +41,7 @@ func TestHandlePending_VolumeSnapshot_with_ValkeyCluster_target_rejected(t *test ObjectMeta: metav1.ObjectMeta{Name: "vc-shard", Namespace: "ns"}, Spec: cachev1alpha1.ValkeyClusterSpec{ Shards: 3, - ReplicasPerShard: 1, + ReplicasPerShard: ptr.To[int32](1), Version: cachev1alpha1.ValkeyVersion{Version: "8.1.6"}, }, } diff --git a/internal/controller/valkeybackup_phase_test.go b/internal/controller/valkeybackup_phase_test.go index 0c196bdd..3ea1f651 100644 --- a/internal/controller/valkeybackup_phase_test.go +++ b/internal/controller/valkeybackup_phase_test.go @@ -17,6 +17,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -148,7 +149,7 @@ func TestBackup_buildJob_ValkeyClusterUsesPerShardPrimaryHosts(t *testing.T) { ObjectMeta: metav1.ObjectMeta{Name: "vk", Namespace: "ns"}, Spec: cachev1alpha1.ValkeyClusterSpec{ Shards: 3, - ReplicasPerShard: 1, + ReplicasPerShard: ptr.To[int32](1), }, Status: cachev1alpha1.ValkeyClusterStatus{ Shards: []cachev1alpha1.ShardStatus{ diff --git a/internal/controller/valkeycluster_controller.go b/internal/controller/valkeycluster_controller.go index 3b9c7470..e4c86469 100644 --- a/internal/controller/valkeycluster_controller.go +++ b/internal/controller/valkeycluster_controller.go @@ -225,6 +225,7 @@ func (r *ValkeyClusterReconciler) Reconcile(ctx context.Context, req ctrl.Reques Pod: vc.Spec.Pod, AuthSecretHash: hashAuthSecret(password), RevisionHistoryLimit: vc.Spec.RevisionHistoryLimit, + Modules: vc.Spec.Modules, } if vc.Spec.TLS != nil && vc.Spec.TLS.Enabled { // CertManager 와 CustomCert 둘 다 동일 secret 마운트 — webhook 이 둘 중 하나만 @@ -281,7 +282,7 @@ func (r *ValkeyClusterReconciler) Reconcile(ctx context.Context, req ctrl.Reques perShard := vc.Spec.PodDisruptionBudget != nil && vc.Spec.PodDisruptionBudget.PerShard if perShard { // CDEX-M2 per-shard PDB loop. shardReplicas = 1 primary + ReplicasPerShard. - shardReplicas := int32(1) + vc.Spec.ReplicasPerShard + shardReplicas := int32(1) + vc.Spec.GetReplicasPerShard() for i := 0; i < int(vc.Spec.Shards); i++ { pdb := resources.BuildShardPDB(vc.Name, vc.Namespace, i, shardReplicas, vc.Spec.PodDisruptionBudget) if err := commonsapply.PDB(ctx, r.Client, r.Scheme, vc, pdb); err != nil { @@ -370,6 +371,26 @@ func (r *ValkeyClusterReconciler) Reconcile(ctx context.Context, req ctrl.Reques } } + // 10.5 멤버십 자가복구 (결함 ③) — slot 레벨 health gate(cluster_state=ok)가 + // 통과해도 *멤버십/replica 수* 는 별도다. 노드가 재시작 후 새 node id 를 + // 얻거나 nodes.conf 를 잃어 멤버십에서 이탈하면 shard 가 desired 보다 적은 + // replica 로 운영된다. allReady && state=ok 인 정상 cluster 에서도 CLUSTER + // NODES 를 desired 토폴로지와 비교해 누락 멤버를 MEET + REPLICATE 로 재합류. + // 멱등 — 이미 올바른 멤버는 건드리지 않는다. + if allReady && info != nil && info.State == "ok" && vc.Spec.GetReplicasPerShard() > 0 { + if reintegrated, mErr := r.ensureClusterMembership(ctx, vc, password); mErr != nil { + logger.Error(mErr, "Cluster membership re-integration pending — will retry") + } else if reintegrated > 0 { + logger.Info("Re-integrated cluster members", "count", reintegrated) + commonsevents.Emitf(r.Recorder, vc, "Reintegration", + "cluster 멤버십 자가복구: %d 노드 재합류 (CLUSTER MEET + REPLICATE)", reintegrated) + // 멤버십이 막 바뀌었으니 NODES 를 다시 읽어 shard status 정확도를 높인다. + if i2, n2, e2 := r.pollClusterState(ctx, vc, password); e2 == nil && i2 != nil { + info, nodes = i2, n2 + } + } + } + // 11. Shard status 빌드 + metrics 갱신. // ADR-0004 후속: NODES 응답이 있으면 *실제 토폴로지* 기반 — failover 정확. // 없으면 spec 기반 fallback (부트스트랩 직후 / NODES 조회 실패 시). @@ -702,7 +723,7 @@ func (r *ValkeyClusterReconciler) ensureClusterMeet( dial := func(addr string) *redis.Client { return dialPod(addr, password, tlsCfg) } createCtx, createSpan := observability.StartCallSpan(ctx, "ValkeyCluster/CreateCluster") defer createSpan.End() - if err := vk.CreateCluster(createCtx, dial, addresses, int(vc.Spec.Shards), int(vc.Spec.ReplicasPerShard)); err != nil { + if err := vk.CreateCluster(createCtx, dial, addresses, int(vc.Spec.Shards), int(vc.Spec.GetReplicasPerShard())); err != nil { createSpan.RecordError(err) return err } @@ -1010,7 +1031,7 @@ func joinRanges(rs []string) string { // slot range 도 CreateCluster 와 동일한 균등 분배 (마지막 shard 가 잔여 흡수). func buildShardStatus(vc *cachev1alpha1.ValkeyCluster) []cachev1alpha1.ShardStatus { shards := int(vc.Spec.Shards) - rps := int(vc.Spec.ReplicasPerShard) + rps := int(vc.Spec.GetReplicasPerShard()) if shards == 0 { return nil } diff --git a/internal/controller/valkeycluster_controller_test.go b/internal/controller/valkeycluster_controller_test.go index 672eb942..ff612b16 100644 --- a/internal/controller/valkeycluster_controller_test.go +++ b/internal/controller/valkeycluster_controller_test.go @@ -17,6 +17,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/reconcile" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" cachev1alpha1 "github.com/keiailab/valkey-operator/api/v1alpha1" "github.com/keiailab/valkey-operator/internal/resources" @@ -45,7 +46,7 @@ var _ = Describe("ValkeyCluster Controller", func() { }, Spec: cachev1alpha1.ValkeyClusterSpec{ Shards: 3, - ReplicasPerShard: 1, + ReplicasPerShard: ptr.To[int32](1), Version: cachev1alpha1.ValkeyVersion{Version: "8.1.6"}, }, } @@ -101,7 +102,7 @@ var _ = Describe("ValkeyCluster Controller", func() { ObjectMeta: metav1.ObjectMeta{Name: resourceName, Namespace: "default"}, Spec: cachev1alpha1.ValkeyClusterSpec{ Shards: 3, - ReplicasPerShard: 1, + ReplicasPerShard: ptr.To[int32](1), Version: cachev1alpha1.ValkeyVersion{Version: "8.1.6"}, }, } diff --git a/internal/controller/valkeyrestore_controller_test.go b/internal/controller/valkeyrestore_controller_test.go index b1c365d1..c5072848 100644 --- a/internal/controller/valkeyrestore_controller_test.go +++ b/internal/controller/valkeyrestore_controller_test.go @@ -20,6 +20,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -624,7 +625,7 @@ func standaloneCluster(name, ns string, shards, repsPerShard int32) *cachev1alph ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns}, Spec: cachev1alpha1.ValkeyClusterSpec{ Shards: shards, - ReplicasPerShard: repsPerShard, + ReplicasPerShard: ptr.To(repsPerShard), }, } } diff --git a/internal/resources/servicemonitor_test.go b/internal/resources/servicemonitor_test.go index 4e8323b8..87eba51a 100644 --- a/internal/resources/servicemonitor_test.go +++ b/internal/resources/servicemonitor_test.go @@ -6,6 +6,7 @@ Licensed under the MIT License. See the LICENSE file for details. package resources import ( + "k8s.io/utils/ptr" "strings" "testing" @@ -96,7 +97,7 @@ func TestConfigMap_autoFailoverFalse_setsDirective(t *testing.T) { vc.Name = "vk" vc.Namespace = "ns" vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.NodeTimeoutMillis = 15000 vc.Spec.AutoFailover = false @@ -115,7 +116,7 @@ func TestConfigMap_autoFailoverTrue_omitsDirective(t *testing.T) { vc.Name = "vk" vc.Namespace = "ns" vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.NodeTimeoutMillis = 15000 vc.Spec.AutoFailover = true @@ -163,7 +164,7 @@ func TestConfigMap_persistenceMode(t *testing.T) { vc.Name = "vk" vc.Namespace = "ns" vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.Persistence = &cachev1alpha1.PersistencePolicy{Mode: tc.mode} cm, err := BuildConfigMapForValkeyCluster(vc, "pwd") @@ -191,7 +192,7 @@ func TestConfigMap_additionalConfig(t *testing.T) { vc.Name = "vk" vc.Namespace = "ns" vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.AdditionalConfig = map[string]string{ "maxclients": "10000", "hash-max-ziplist-entries": "512", diff --git a/internal/valkey/cluster.go b/internal/valkey/cluster.go index 76573ab9..d3bd903b 100644 --- a/internal/valkey/cluster.go +++ b/internal/valkey/cluster.go @@ -272,6 +272,35 @@ func findMyself(nodes []NodeView) *NodeView { return nil } +// MeetNode — 결함 ③ 자가복구: seed 노드에서 target 노드로 CLUSTER MEET 발행. +// +// 멱등: target 이 이미 멤버면 valkey 가 사실상 no-op 처리한다. ensureMeet 와 동일하게 +// hostname 을 IP 로 정규화(CLUSTER MEET 은 IP 만 수용)한다. seedAddr / targetAddr 는 +// "host:port" 또는 "ip:port". +func MeetNode(ctx context.Context, dial func(addr string) *redis.Client, seedAddr, targetAddr string) error { + ipAddr, err := resolveAddrIP(ctx, targetAddr) + if err != nil { + return fmt.Errorf("cluster meet resolve %s: %w", targetAddr, err) + } + host, portStr, ok := strings.Cut(ipAddr, ":") + if !ok { + return fmt.Errorf("invalid resolved address: %q", ipAddr) + } + c := dial(seedAddr) + defer func() { _ = c.Close() }() + if err := c.ClusterMeet(ctx, host, portStr).Err(); err != nil { + return fmt.Errorf("cluster meet %s (resolved %s): %w", targetAddr, ipAddr, err) + } + return nil +} + +// ReplicateTo — 결함 ③ 자가복구: addr 노드가 masterID 를 따르도록 CLUSTER REPLICATE. +// replicateWithRetry 를 재사용해 gossip 수렴 지연("Unknown node")을 흡수하고, 이미 +// 올바른 master 를 가리키면 skip(멱등). +func ReplicateTo(ctx context.Context, dial func(addr string) *redis.Client, addr, masterID string) error { + return replicateWithRetry(ctx, dial, addr, masterID) +} + // ForgetNode — scale-in 시 node 제거. 모든 잔존 primary 에서 호출 필요. func ForgetNode(ctx context.Context, c *redis.Client, nodeID string) error { if err := c.ClusterForget(ctx, nodeID).Err(); err != nil { diff --git a/internal/webhook/v1alpha1/additionalconfig_validation_test.go b/internal/webhook/v1alpha1/additionalconfig_validation_test.go index a69e02cd..d0f803b8 100644 --- a/internal/webhook/v1alpha1/additionalconfig_validation_test.go +++ b/internal/webhook/v1alpha1/additionalconfig_validation_test.go @@ -18,6 +18,7 @@ import ( "testing" "k8s.io/apimachinery/pkg/util/validation/field" + "k8s.io/utils/ptr" cachev1alpha1 "github.com/keiailab/valkey-operator/api/v1alpha1" ) @@ -158,7 +159,7 @@ func TestValidateClusterSpec_AdditionalConfig(t *testing.T) { t.Parallel() vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.AdditionalConfig = map[string]string{"tls-port": "9999"} errs := validateClusterSpec(vc) found := false diff --git a/internal/webhook/v1alpha1/admission_roundtrip_test.go b/internal/webhook/v1alpha1/admission_roundtrip_test.go index 39d5d852..d9efcf81 100644 --- a/internal/webhook/v1alpha1/admission_roundtrip_test.go +++ b/internal/webhook/v1alpha1/admission_roundtrip_test.go @@ -17,6 +17,7 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" cachev1alpha1 "github.com/keiailab/valkey-operator/api/v1alpha1" ) @@ -82,17 +83,16 @@ var _ = Describe("Valkey webhook admission round-trip", func() { Expect(err.Error()).To(ContainSubstring("passwordSecretRef")) }) - // 발견: 'autoFailover + ReplicasPerShard=0' webhook invariant 는 *real - // apiserver 통해 도달 불가능* — CRD 의 'kubebuilder:default=1' 가 admission - // 전에 0→1 변환. webhook 의 본 검증 path 는 *dead code* (CRD default 가 항상 - // 1+ 로 보장). 추후 cleanup ADR candidate (Surgical §3 — 발견사항만 보고). + // defect ④: 'autoFailover + replicasPerShard=0' reject invariant 는 제거됐다. + // masters-only 토폴로지는 유효하며 명시 0 은 apiserver(CRD default=1 은 *부재* 시만 + // 적용) + mutating defaulter(0→1 clobber 제거) 양쪽에서 보존된다. It("rejects ValkeyCluster with totalNodes > 100", func() { vc := &cachev1alpha1.ValkeyCluster{ ObjectMeta: metav1.ObjectMeta{Name: "rt-toomany", Namespace: "default"}, Spec: cachev1alpha1.ValkeyClusterSpec{ Shards: 50, - ReplicasPerShard: 2, // total = 50 * 3 = 150 + ReplicasPerShard: ptr.To[int32](2), // total = 50 * 3 = 150 Storage: cachev1alpha1.StorageSpec{Size: resource.MustParse("8Gi")}, }, } @@ -107,7 +107,7 @@ var _ = Describe("Valkey webhook admission round-trip", func() { ObjectMeta: metav1.ObjectMeta{Name: "rt-clusterhappy", Namespace: "default"}, Spec: cachev1alpha1.ValkeyClusterSpec{ Shards: 3, - ReplicasPerShard: 1, + ReplicasPerShard: ptr.To[int32](1), Storage: cachev1alpha1.StorageSpec{Size: resource.MustParse("8Gi")}, }, } diff --git a/internal/webhook/v1alpha1/valkey_validation_test.go b/internal/webhook/v1alpha1/valkey_validation_test.go index e3d620aa..c563d96c 100644 --- a/internal/webhook/v1alpha1/valkey_validation_test.go +++ b/internal/webhook/v1alpha1/valkey_validation_test.go @@ -15,6 +15,7 @@ import ( corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" + "k8s.io/utils/ptr" cachev1alpha1 "github.com/keiailab/valkey-operator/api/v1alpha1" ) @@ -235,27 +236,28 @@ func TestValidateValkeySpec(t *testing.T) { // validateClusterSpec 회귀 보호 (cycle 131). func TestValidateClusterSpec(t *testing.T) { t.Parallel() - t.Run("autoFailover=true + replicasPerShard=0 → error", func(t *testing.T) { + t.Run("masters-only replicasPerShard=0 → accepted (defect ④)", func(t *testing.T) { t.Parallel() vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 0 + vc.Spec.ReplicasPerShard = ptr.To[int32](0) vc.Spec.AutoFailover = true + vc.Spec.Version.Version = "8.1.6" errs := validateClusterSpec(vc) - if len(errs) == 0 { - t.Error("autoFailover=true + replicasPerShard=0 → expected error") + if len(errs) != 0 { + t.Errorf("masters-only (replicasPerShard=0) should be accepted, got %v", errs) } }) t.Run("totalNodes > 100 → error", func(t *testing.T) { t.Parallel() vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 50 - vc.Spec.ReplicasPerShard = 1 // total = 50 * 2 = 100, OK. + vc.Spec.ReplicasPerShard = ptr.To[int32](1) // total = 50 * 2 = 100, OK. errs := validateClusterSpec(vc) if len(errs) > 0 { t.Errorf("100 nodes → expected ok, got %v", errs) } - vc.Spec.ReplicasPerShard = 2 // total = 50 * 3 = 150, error. + vc.Spec.ReplicasPerShard = ptr.To[int32](2) // total = 50 * 3 = 150, error. errs = validateClusterSpec(vc) var hasOver100 bool for _, e := range errs { @@ -271,7 +273,7 @@ func TestValidateClusterSpec(t *testing.T) { t.Parallel() vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.TLS = &cachev1alpha1.TLSSpec{ Enabled: true, CertManager: &cachev1alpha1.CertManagerSpec{IssuerRef: cachev1alpha1.CertIssuerRef{Name: "ca"}}, @@ -292,7 +294,7 @@ func TestValidateClusterSpec(t *testing.T) { t.Parallel() vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.Auth.Enabled = false vc.Spec.Auth.Users = []cachev1alpha1.ValkeyUser{{Name: "alice"}} errs := validateClusterSpec(vc) @@ -304,7 +306,7 @@ func TestValidateClusterSpec(t *testing.T) { t.Parallel() vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.AutoFailover = true errs := validateClusterSpec(vc) if len(errs) > 0 { @@ -772,7 +774,7 @@ func TestValidateClusterSpec_StorageClassName(t *testing.T) { vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Version.Version = cachev1alpha1.DefaultValkeyVersion vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.Storage.Size = resource.MustParse("2Gi") vc.Spec.Storage.StorageClassName = "Bad_Name" errs := validateClusterSpec(vc) diff --git a/internal/webhook/v1alpha1/valkeycluster_webhook.go b/internal/webhook/v1alpha1/valkeycluster_webhook.go index 175f643c..ecf7639d 100644 --- a/internal/webhook/v1alpha1/valkeycluster_webhook.go +++ b/internal/webhook/v1alpha1/valkeycluster_webhook.go @@ -73,11 +73,14 @@ func (d *ValkeyClusterCustomDefaulter) Default(_ context.Context, obj *cachev1al if obj.Spec.Shards == 0 { obj.Spec.Shards = 3 } - // ReplicasPerShard: AutoFailover=true 가 default 인데 0 이면 webhook validator - // 가 reject 한다. CRD default=1 이지만 omitempty 부재로 schema default 가 skip - // 되므로 여기서 채움. - if obj.Spec.ReplicasPerShard == 0 { - obj.Spec.ReplicasPerShard = 1 + // ReplicasPerShard: 이제 *pointer* (*int32) — nil(미지정) 과 명시 0(masters-only) + // 이 구별된다. CRD default=1 은 제거했고 (non-pointer 시절 명시 0 을 무력화했던 + // defect ④), 미지정(nil)→1 defaulting 은 여기서 처리한다. *명시 0 은 절대 손대지 + // 않는다* — pointer 가 non-nil 이면 그 값(0 포함)을 그대로 보존한다. webhook 이 + // 항상 명시값을 채우므로 저장된 객체는 nil 이 아닌 explicit 값을 갖는다. + if obj.Spec.ReplicasPerShard == nil { + one := int32(1) + obj.Spec.ReplicasPerShard = &one } if obj.Spec.Version.Version == "" { obj.Spec.Version.Version = cachev1alpha1.DefaultValkeyVersion @@ -139,24 +142,12 @@ func validateClusterSpec(vc *cachev1alpha1.ValkeyCluster) field.ErrorList { errs = append(errs, err) } - // AutoFailover=true + ReplicasPerShard=0 → failover 불가 (replica 부재). - // - // ADR-0017 Type A' (조건부 unreachable, defensive 유지): - // - ValkeyClusterSpec.ReplicasPerShard 는 'json:"replicasPerShard"' (no - // omitempty), CRD '+kubebuilder:default=1', mutating defaulter (Default - // 함수 위) 가 명시 0→1 보강. - // - webhook.enabled=true 환경 (operator 정상 운영) 에서는 mutating defaulter - // 가 0→1 변환 → 본 invariant 도달 안 함. - // - webhook.enabled=false 환경 (CRD-only 모드, helm opt-out) 에서는 mutating - // defaulter 우회 + CRD default 가 *명시 0* 무력화 못 함 → reachable. - // - 따라서 *제거 금지* — defensive 가드. it47 commit 5f3f91c 의 envtest - // 'unreachable' 분석은 webhook.enabled=true 한정 시나리오. - if vc.Spec.AutoFailover && vc.Spec.ReplicasPerShard == 0 { - errs = append(errs, field.Forbidden( - specPath.Child("autoFailover"), - "autoFailover=true requires replicasPerShard >= 1 (no replicas means no failover possible)", - )) - } + // ReplicasPerShard=0 (masters-only 토폴로지, defect ④) 은 *유효* 하다. + // AutoFailover 는 non-pointer bool 로 CRD default=true — 명시 false 와 zero 를 + // 구별할 수 없으므로 'autoFailover=true && rps=0' 를 reject 하면 모든 masters-only + // 사용자가 autoFailover=false 를 강제로 함께 명시해야 한다. failover 가 replica + // 부재 시 불가능한 것은 masters-only 토폴로지에 *내재된* tradeoff 이지 모순 조합이 + // 아니므로 admission 에서 거부하지 않는다 (replica 0 → 단순히 failover 비활성). // 총 노드 수 상한 — Valkey cluster 권장 (>100 노드는 운영 부담 + gossip 오버헤드). total := vc.Spec.TotalNodes() @@ -234,6 +225,9 @@ func validateClusterSpec(vc *cachev1alpha1.ValkeyCluster) field.ErrorList { // additionalConfig directive 주입 + operator-managed override 차단 (Valkey CR 와 sister). errs = append(errs, validateAdditionalConfig(specPath.Child("additionalConfig"), vc.Spec.AdditionalConfig)...) + // modules allow-list 검증 (ADR-0032) — 외부 Redis Stack 거부 (Valkey CR 와 동일 검증). + errs = append(errs, validateModules(specPath.Child("modules"), vc.Spec.Modules)...) + return errs } diff --git a/internal/webhook/v1alpha1/webhook_validation_test.go b/internal/webhook/v1alpha1/webhook_validation_test.go index fea0bc17..94912aeb 100644 --- a/internal/webhook/v1alpha1/webhook_validation_test.go +++ b/internal/webhook/v1alpha1/webhook_validation_test.go @@ -14,20 +14,87 @@ import ( "testing" "k8s.io/apimachinery/pkg/api/resource" + "k8s.io/utils/ptr" cachev1alpha1 "github.com/keiailab/valkey-operator/api/v1alpha1" ) -func TestValkeyClusterValidate_AutoFailover_requires_replicas(t *testing.T) { +// defect ④: masters-only 토폴로지 (replicasPerShard=0) 는 autoFailover default(true) +// 와 함께여도 *허용* 된다. failover 가 replica 부재 시 불가능한 것은 내재된 tradeoff. +func TestValkeyClusterValidate_mastersOnly_accepted(t *testing.T) { v := &ValkeyClusterCustomValidator{} vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 vc.Spec.AutoFailover = true - vc.Spec.ReplicasPerShard = 0 // 모순 — autoFailover 가 replica 없이 동작 불가. + vc.Spec.ReplicasPerShard = ptr.To[int32](0) // masters-only — 명시 0 보존. vc.Spec.Version.Version = "8.1.6" + if _, err := v.ValidateCreate(context.Background(), vc); err != nil { + t.Fatalf("masters-only (replicasPerShard=0) should be accepted: %v", err) + } +} + +// defect ④: mutating defaulter 는 명시 0 을 1 로 clobber 하지 않는다 (masters-only 보존). +// 동시에 unset (CRD apiserver default 가 채우는 경로) 도 손대지 않는다. +func TestValkeyClusterDefaulter_preserves_explicit_zero_replicas(t *testing.T) { + d := &ValkeyClusterCustomDefaulter{} + vc := &cachev1alpha1.ValkeyCluster{} + vc.Spec.Shards = 3 + vc.Spec.ReplicasPerShard = ptr.To[int32](0) // 명시 masters-only. + if err := d.Default(context.Background(), vc); err != nil { + t.Fatalf("default: %v", err) + } + if vc.Spec.ReplicasPerShard == nil || *vc.Spec.ReplicasPerShard != 0 { + t.Errorf("explicit replicasPerShard=0 must be preserved, got %v", vc.Spec.ReplicasPerShard) + } + if got := vc.Spec.TotalNodes(); got != 3 { + t.Errorf("masters-only TotalNodes(): got %d want 3 (shards masters, 0 replicas)", got) + } +} + +// defect ④: mutating defaulter 는 nil(미지정) ReplicasPerShard 를 명시 1 로 채운다. +// (CRD default=1 제거 후 nil→1 defaulting 책임이 webhook 으로 이동.) +func TestValkeyClusterDefaulter_defaults_nil_replicas_to_one(t *testing.T) { + d := &ValkeyClusterCustomDefaulter{} + vc := &cachev1alpha1.ValkeyCluster{} + vc.Spec.Shards = 3 + vc.Spec.ReplicasPerShard = nil // 미지정. + if err := d.Default(context.Background(), vc); err != nil { + t.Fatalf("default: %v", err) + } + if vc.Spec.ReplicasPerShard == nil || *vc.Spec.ReplicasPerShard != 1 { + t.Errorf("nil replicasPerShard must be defaulted to 1, got %v", vc.Spec.ReplicasPerShard) + } + if got := vc.Spec.TotalNodes(); got != 6 { + t.Errorf("defaulted TotalNodes(): got %d want 6 (3 shards * (1+1))", got) + } +} + +// defect ⑥: ValkeyCluster 도 modules 검증 — 외부 Redis Stack(비공식 preset) 거부. +func TestValkeyClusterValidate_rejects_external_module(t *testing.T) { + v := &ValkeyClusterCustomValidator{} + vc := &cachev1alpha1.ValkeyCluster{} + vc.Spec.Shards = 3 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) + vc.Spec.Version.Version = "8.1.6" + vc.Spec.Modules = []cachev1alpha1.ModuleSpec{{Name: "redisearch"}} // 외부 Redis Stack. + if _, err := v.ValidateCreate(context.Background(), vc); err == nil { - t.Fatal("expected validation error for AutoFailover=true + ReplicasPerShard=0") + t.Fatal("expected validation error for external Redis Stack module on ValkeyCluster") + } +} + +// defect ⑥: 공식 BSD preset 은 ValkeyCluster 에서도 허용. +func TestValkeyClusterValidate_accepts_official_module(t *testing.T) { + v := &ValkeyClusterCustomValidator{} + vc := &cachev1alpha1.ValkeyCluster{} + vc.Spec.Shards = 3 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) + vc.Spec.Version.Version = "8.1.6" + vc.Spec.Modules = []cachev1alpha1.ModuleSpec{{Name: "valkey-search"}} + + if _, err := v.ValidateCreate(context.Background(), vc); err != nil { + t.Fatalf("official preset valkey-search should be accepted on ValkeyCluster: %v", err) } } @@ -35,7 +102,7 @@ func TestValkeyClusterValidate_total_node_limit(t *testing.T) { v := &ValkeyClusterCustomValidator{} vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 50 - vc.Spec.ReplicasPerShard = 5 // 50 * 6 = 300 > 100. + vc.Spec.ReplicasPerShard = ptr.To[int32](5) // 50 * 6 = 300 > 100. if _, err := v.ValidateCreate(context.Background(), vc); err == nil { t.Fatal("expected validation error for total > 100") @@ -46,7 +113,7 @@ func TestValkeyClusterValidate_TLS_requires_certManager_or_customCert(t *testing v := &ValkeyClusterCustomValidator{} vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.TLS = &cachev1alpha1.TLSSpec{Enabled: true} // 둘 다 미명시. if _, err := v.ValidateCreate(context.Background(), vc); err == nil { @@ -58,7 +125,7 @@ func TestValkeyClusterValidate_TLS_mutually_exclusive(t *testing.T) { v := &ValkeyClusterCustomValidator{} vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.TLS = &cachev1alpha1.TLSSpec{ Enabled: true, CertManager: &cachev1alpha1.CertManagerSpec{IssuerRef: cachev1alpha1.CertIssuerRef{Name: "issuer"}}, @@ -74,7 +141,7 @@ func TestValkeyClusterValidate_Auth_users_requires_enabled(t *testing.T) { v := &ValkeyClusterCustomValidator{} vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.Auth = cachev1alpha1.AuthSpec{ Enabled: false, Users: []cachev1alpha1.ValkeyUser{ @@ -91,7 +158,7 @@ func TestValkeyClusterValidate_Update_storageClass_immutable(t *testing.T) { v := &ValkeyClusterCustomValidator{} old := &cachev1alpha1.ValkeyCluster{} old.Spec.Shards = 3 - old.Spec.ReplicasPerShard = 1 + old.Spec.ReplicasPerShard = ptr.To[int32](1) old.Spec.Storage.StorageClassName = "fast-ssd" new := old.DeepCopy() new.Spec.Storage.StorageClassName = "slow-hdd" @@ -107,7 +174,7 @@ func TestValkeyClusterValidate_Update_tls_false_to_true_allowed(t *testing.T) { v := &ValkeyClusterCustomValidator{} old := &cachev1alpha1.ValkeyCluster{} old.Spec.Shards = 3 - old.Spec.ReplicasPerShard = 1 + old.Spec.ReplicasPerShard = ptr.To[int32](1) old.Spec.Version.Version = "8.1.6" old.Spec.TLS = &cachev1alpha1.TLSSpec{Enabled: false} new := old.DeepCopy() @@ -125,7 +192,7 @@ func TestValkeyClusterValidate_Update_tls_true_to_false_forbidden(t *testing.T) v := &ValkeyClusterCustomValidator{} old := &cachev1alpha1.ValkeyCluster{} old.Spec.Shards = 3 - old.Spec.ReplicasPerShard = 1 + old.Spec.ReplicasPerShard = ptr.To[int32](1) old.Spec.Version.Version = "8.1.6" old.Spec.TLS = &cachev1alpha1.TLSSpec{ Enabled: true, @@ -143,7 +210,7 @@ func TestValkeyClusterValidate_valid_passes(t *testing.T) { v := &ValkeyClusterCustomValidator{} vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.AutoFailover = true vc.Spec.Version.Version = "8.1.6" @@ -241,7 +308,7 @@ func TestValkeyClusterValidate_Update_storageSize_shrink_rejected(t *testing.T) v := &ValkeyClusterCustomValidator{} old := &cachev1alpha1.ValkeyCluster{} old.Spec.Shards = 3 - old.Spec.ReplicasPerShard = 1 + old.Spec.ReplicasPerShard = ptr.To[int32](1) old.Spec.Version.Version = "8.1.6" old.Spec.Storage.Size = resource.MustParse("16Gi") new := old.DeepCopy() @@ -260,7 +327,7 @@ func TestValkeyClusterValidate_Update_storageSize_grow_accepted(t *testing.T) { v := &ValkeyClusterCustomValidator{} old := &cachev1alpha1.ValkeyCluster{} old.Spec.Shards = 3 - old.Spec.ReplicasPerShard = 1 + old.Spec.ReplicasPerShard = ptr.To[int32](1) old.Spec.Version.Version = "8.1.6" old.Spec.Storage.Size = resource.MustParse("8Gi") new := old.DeepCopy() @@ -309,7 +376,7 @@ func TestValkeyCluster_TLS_AutoSelfSigned_alone_passes(t *testing.T) { v := &ValkeyClusterCustomValidator{} vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.Version.Version = "8.1.6" vc.Spec.TLS = &cachev1alpha1.TLSSpec{ Enabled: true, @@ -325,7 +392,7 @@ func TestValkeyCluster_TLS_AutoSelfSigned_with_IssuerRef_rejected(t *testing.T) v := &ValkeyClusterCustomValidator{} vc := &cachev1alpha1.ValkeyCluster{} vc.Spec.Shards = 3 - vc.Spec.ReplicasPerShard = 1 + vc.Spec.ReplicasPerShard = ptr.To[int32](1) vc.Spec.Version.Version = "8.1.6" vc.Spec.TLS = &cachev1alpha1.TLSSpec{ Enabled: true,