rock/kmsvc-manage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: queue-operator dropped reconcile namespace and overflowed hash-range status

Browse Source 
main.go called Reconcile(ctx, req.Name) without req.Namespace, so the
Get against the namespaced Queue CRD always 404'd and was silently
swallowed as success -- no shard topics or Redis state were ever created.

Separately, ShardStatus.HashRangeStart/End were uint32, but controller-gen
maps that to OpenAPI format:int32, whose max (2147483647) is smaller than
FullHashRangeEnd (0xFFFFFFFF), so the apiserver rejected every status
update with the (misleadingly empty-looking) "must be of type integer with
format int32" error. Widened to int64, regenerated the CRD, and synced the
chart's bundled copy.
This commit is contained in:
riotpiaole
2026-06-22 12:23:23 -07:00
parent 173a4935ab
commit 9fa5420b22
7 changed files with 40 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apis/kmsvc/v1/queue_types.go
+6 -3
View File
@@ -102,9 +102,12 @@ type ShardStatus struct {
Topic string `json:"topic"` Topic string `json:"topic"`
// HashRangeStart/HashRangeEnd define the [start, end) murmur2 hash range // HashRangeStart/HashRangeEnd define the [start, end) murmur2 hash range
// this shard owns over the 32-bit key space. // this shard owns over the 32-bit key space. Stored as int64 (not uint32)
HashRangeStart uint32 `json:"hashRangeStart"` // because controller-gen maps Go uint32 to OpenAPI format:int32, whose max
HashRangeEnd uint32 `json:"hashRangeEnd"` // (2147483647) is smaller than FullHashRangeEnd (0xFFFFFFFF) and the
// apiserver rejects the status update.
HashRangeStart int64 `json:"hashRangeStart"`
HashRangeEnd int64 `json:"hashRangeEnd"`
// Phase is this shard's lifecycle state. // Phase is this shard's lifecycle state.
// +kubebuilder:validation:Enum=Active;Closing;Closed // +kubebuilder:validation:Enum=Active;Closing;Closed
cmd/queue-operator/main.go
+1 -1
View File
@@ -72,7 +72,7 @@ func main() {
err = ctrl.NewControllerManagedBy(mgr). err = ctrl.NewControllerManagedBy(mgr).
For(&kmsvcv1.Queue{}). For(&kmsvcv1.Queue{}).
Complete(reconcile.Func(func(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { Complete(reconcile.Func(func(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
if err := reconciler.Reconcile(ctx, req.Name); err != nil { if err := reconciler.Reconcile(ctx, req.Namespace, req.Name); err != nil {
return reconcile.Result{}, err return reconcile.Result{}, err
} }
return reconcile.Result{}, nil return reconcile.Result{}, nil
config/crd/kmsvc.io_queues.yaml
+6 -3
View File
@@ -218,13 +218,16 @@ spec:
format: date-time format: date-time
type: string type: string
hashRangeEnd: hashRangeEnd:
format: int32 format: int64
type: integer type: integer
hashRangeStart: hashRangeStart:
description: |- description: |-
HashRangeStart/HashRangeEnd define the [start, end) murmur2 hash range HashRangeStart/HashRangeEnd define the [start, end) murmur2 hash range
this shard owns over the 32-bit key space. this shard owns over the 32-bit key space. Stored as int64 (not uint32)
format: int32 because controller-gen maps Go uint32 to OpenAPI format:int32, whose max
(2147483647) is smaller than FullHashRangeEnd (0xFFFFFFFF) and the
apiserver rejects the status update.
format: int64
type: integer type: integer
id: id:
description: ID is the shard's identifier, used in its topic description: ID is the shard's identifier, used in its topic
internal/operator/queue_controller.go
+7 -7
View File
@@ -57,14 +57,14 @@ func (r *QueueReconciler) now() time.Time {
} }
// Reconcile implements the controller-runtime reconcile loop. // Reconcile implements the controller-runtime reconcile loop.
func (r *QueueReconciler) Reconcile(ctx context.Context, name string) error { func (r *QueueReconciler) Reconcile(ctx context.Context, namespace, name string) error {
var queue kmsvcv1.Queue var queue kmsvcv1.Queue
err := r.Client.Get(ctx, client.ObjectKey{Name: name}, &queue) err := r.Client.Get(ctx, client.ObjectKey{Namespace: namespace, Name: name}, &queue)
if apierrors.IsNotFound(err) { if apierrors.IsNotFound(err) {
return nil return nil
} }
if err != nil { if err != nil {
return fmt.Errorf("get queue %s: %w", name, err) return fmt.Errorf("get queue %s/%s: %w", namespace, name, err)
} }
if !queue.DeletionTimestamp.IsZero() { if !queue.DeletionTimestamp.IsZero() {
@@ -113,8 +113,8 @@ func (r *QueueReconciler) newShard(id, parentID string, start, end uint32, queue
return kmsvcv1.ShardStatus{ return kmsvcv1.ShardStatus{
ID: id, ID: id,
Topic: kafka.ShardTopicName(queue.Name, queue.Spec.FIFOQueue, id), Topic: kafka.ShardTopicName(queue.Name, queue.Spec.FIFOQueue, id),
HashRangeStart: start, HashRangeStart: int64(start),
HashRangeEnd: end, HashRangeEnd: int64(end),
Phase: kmsvcv1.ShardPhaseActive, Phase: kmsvcv1.ShardPhaseActive,
ParentID: parentID, ParentID: parentID,
CreatedAt: metav1.NewTime(r.now()), CreatedAt: metav1.NewTime(r.now()),
@@ -168,8 +168,8 @@ func (r *QueueReconciler) publishRedisState(ctx context.Context, queue *kmsvcv1.
shards = append(shards, kafka.Shard{ shards = append(shards, kafka.Shard{
ID: s.ID, ID: s.ID,
Topic: s.Topic, Topic: s.Topic,
HashRangeStart: s.HashRangeStart, HashRangeStart: uint32(s.HashRangeStart),
HashRangeEnd: s.HashRangeEnd, HashRangeEnd: uint32(s.HashRangeEnd),
Phase: string(s.Phase), Phase: string(s.Phase),
}) })
} }
internal/operator/queue_controller_test.go
+11 -11
View File
@@ -76,7 +76,7 @@ func TestReconcileCreatesShardZeroAndPublishesRedisState(t *testing.T) {
r, admin := newTestReconciler(t, queue) r, admin := newTestReconciler(t, queue)
ctx := context.Background() ctx := context.Background()
if err := r.Reconcile(ctx, "orders"); err != nil { if err := r.Reconcile(ctx, "", "orders"); err != nil {
t.Fatalf("Reconcile: %v", err) t.Fatalf("Reconcile: %v", err)
} }
@@ -118,7 +118,7 @@ func TestReconcileRejectsDLQSelfReference(t *testing.T) {
r, _ := newTestReconciler(t, queue) r, _ := newTestReconciler(t, queue)
ctx := context.Background() ctx := context.Background()
err := r.Reconcile(ctx, "orders") err := r.Reconcile(ctx, "", "orders")
if err == nil { if err == nil {
t.Fatal("expected reconcile to fail for self-referencing DLQ") t.Fatal("expected reconcile to fail for self-referencing DLQ")
} }
@@ -137,7 +137,7 @@ func TestReconcileDeleteCleansUpTopicsAndRedis(t *testing.T) {
r, admin := newTestReconciler(t, queue) r, admin := newTestReconciler(t, queue)
ctx := context.Background() ctx := context.Background()
if err := r.Reconcile(ctx, "orders"); err != nil { if err := r.Reconcile(ctx, "", "orders"); err != nil {
t.Fatalf("initial reconcile: %v", err) t.Fatalf("initial reconcile: %v", err)
} }
@@ -150,7 +150,7 @@ func TestReconcileDeleteCleansUpTopicsAndRedis(t *testing.T) {
if err := r.Client.Delete(ctx, &got); err != nil { if err := r.Client.Delete(ctx, &got); err != nil {
t.Fatalf("delete queue: %v", err) t.Fatalf("delete queue: %v", err)
} }
if err := r.Reconcile(ctx, "orders"); err != nil { if err := r.Reconcile(ctx, "", "orders"); err != nil {
t.Fatalf("delete reconcile: %v", err) t.Fatalf("delete reconcile: %v", err)
} }
@@ -174,7 +174,7 @@ func TestReconcileSplitsShardOverThreshold(t *testing.T) {
r.Now = func() time.Time { return now } r.Now = func() time.Time { return now }
ctx := context.Background() ctx := context.Background()
if err := r.Reconcile(ctx, "orders"); err != nil { if err := r.Reconcile(ctx, "", "orders"); err != nil {
t.Fatalf("initial reconcile: %v", err) t.Fatalf("initial reconcile: %v", err)
} }
var got kmsvcv1.Queue var got kmsvcv1.Queue
@@ -185,7 +185,7 @@ func TestReconcileSplitsShardOverThreshold(t *testing.T) {
admin.setOffsetSum(shard0Topic, 0) admin.setOffsetSum(shard0Topic, 0)
// First sample establishes the baseline (no delta to compare yet). // First sample establishes the baseline (no delta to compare yet).
if err := r.Reconcile(ctx, "orders"); err != nil { if err := r.Reconcile(ctx, "", "orders"); err != nil {
t.Fatalf("baseline reconcile: %v", err) t.Fatalf("baseline reconcile: %v", err)
} }
@@ -194,7 +194,7 @@ func TestReconcileSplitsShardOverThreshold(t *testing.T) {
now = now.Add(time.Second) now = now.Add(time.Second)
admin.setOffsetSum(shard0Topic, 10) admin.setOffsetSum(shard0Topic, 10)
if err := r.Reconcile(ctx, "orders"); err != nil { if err := r.Reconcile(ctx, "", "orders"); err != nil {
t.Fatalf("split-triggering reconcile: %v", err) t.Fatalf("split-triggering reconcile: %v", err)
} }
@@ -220,13 +220,13 @@ func TestReconcileSplitsShardOverThreshold(t *testing.T) {
if len(children) != 2 { if len(children) != 2 {
t.Fatalf("children = %+v, want 2", children) t.Fatalf("children = %+v, want 2", children)
} }
mid := kafka.SplitHashRange(parent.HashRangeStart, parent.HashRangeEnd) mid := kafka.SplitHashRange(uint32(parent.HashRangeStart), uint32(parent.HashRangeEnd))
gotRanges := map[[2]uint32]bool{} gotRanges := map[[2]uint32]bool{}
for _, c := range children { for _, c := range children {
if c.Phase != kmsvcv1.ShardPhaseActive { if c.Phase != kmsvcv1.ShardPhaseActive {
t.Errorf("child %s phase = %v, want Active", c.ID, c.Phase) t.Errorf("child %s phase = %v, want Active", c.ID, c.Phase)
} }
gotRanges[[2]uint32{c.HashRangeStart, c.HashRangeEnd}] = true gotRanges[[2]uint32{uint32(c.HashRangeStart), uint32(c.HashRangeEnd)}] = true
} }
if !gotRanges[[2]uint32{0, mid}] || !gotRanges[[2]uint32{mid, kafka.FullHashRangeEnd}] { if !gotRanges[[2]uint32{0, mid}] || !gotRanges[[2]uint32{mid, kafka.FullHashRangeEnd}] {
t.Errorf("child ranges = %+v, want [0,%d) and [%d,%d)", children, mid, mid, kafka.FullHashRangeEnd) t.Errorf("child ranges = %+v, want [0,%d) and [%d,%d)", children, mid, mid, kafka.FullHashRangeEnd)
@@ -241,7 +241,7 @@ func TestReconcileDrainsClosingShardWhenLagZero(t *testing.T) {
r.Now = func() time.Time { return now } r.Now = func() time.Time { return now }
ctx := context.Background() ctx := context.Background()
if err := r.Reconcile(ctx, "orders"); err != nil { if err := r.Reconcile(ctx, "", "orders"); err != nil {
t.Fatalf("initial reconcile: %v", err) t.Fatalf("initial reconcile: %v", err)
} }
var got kmsvcv1.Queue var got kmsvcv1.Queue
@@ -259,7 +259,7 @@ func TestReconcileDrainsClosingShardWhenLagZero(t *testing.T) {
topic := got.Status.Shards[0].Topic topic := got.Status.Shards[0].Topic
admin.setLag(kafka.ConsumerGroup("orders"), topic, 0) admin.setLag(kafka.ConsumerGroup("orders"), topic, 0)
if err := r.Reconcile(ctx, "orders"); err != nil { if err := r.Reconcile(ctx, "", "orders"); err != nil {
t.Fatalf("drain reconcile: %v", err) t.Fatalf("drain reconcile: %v", err)
} }
internal/operator/shard_split.go
+3 -3
View File
@@ -68,13 +68,13 @@ func (r *QueueReconciler) reconcileSplits(ctx context.Context, queue *kmsvcv1.Qu
continue continue
} }
mid := kafka.SplitHashRange(s.HashRangeStart, s.HashRangeEnd) mid := kafka.SplitHashRange(uint32(s.HashRangeStart), uint32(s.HashRangeEnd))
childAID := nextShardID(queue.Status.Shards) childAID := nextShardID(queue.Status.Shards)
childA := r.newShard(childAID, s.ID, s.HashRangeStart, mid, queue) childA := r.newShard(childAID, s.ID, uint32(s.HashRangeStart), mid, queue)
childA.CreatedAt = metav1.NewTime(now) childA.CreatedAt = metav1.NewTime(now)
childAIDNum, _ := strconv.Atoi(childAID) childAIDNum, _ := strconv.Atoi(childAID)
childBID := strconv.Itoa(childAIDNum + 1) childBID := strconv.Itoa(childAIDNum + 1)
childB := r.newShard(childBID, s.ID, mid, s.HashRangeEnd, queue) childB := r.newShard(childBID, s.ID, mid, uint32(s.HashRangeEnd), queue)
childB.CreatedAt = metav1.NewTime(now) childB.CreatedAt = metav1.NewTime(now)
s.Phase = kmsvcv1.ShardPhaseClosing s.Phase = kmsvcv1.ShardPhaseClosing
k8s/charts/queue-crd/templates/crd.yaml
+6 -3
View File
@@ -218,13 +218,16 @@ spec:
format: date-time format: date-time
type: string type: string
hashRangeEnd: hashRangeEnd:
format: int32 format: int64
type: integer type: integer
hashRangeStart: hashRangeStart:
description: |- description: |-
HashRangeStart/HashRangeEnd define the [start, end) murmur2 hash range HashRangeStart/HashRangeEnd define the [start, end) murmur2 hash range
this shard owns over the 32-bit key space. this shard owns over the 32-bit key space. Stored as int64 (not uint32)
format: int32 because controller-gen maps Go uint32 to OpenAPI format:int32, whose max
(2147483647) is smaller than FullHashRangeEnd (0xFFFFFFFF) and the
apiserver rejects the status update.
format: int64
type: integer type: integer
id: id:
description: ID is the shard's identifier, used in its topic description: ID is the shard's identifier, used in its topic