From 9fa5420b225ce34629ed4218f60f96d3dd839629 Mon Sep 17 00:00:00 2001 From: riotpiaole <19826264+Riotpiaole@users.noreply.github.com> Date: Mon, 22 Jun 2026 12:23:23 -0700 Subject: [PATCH] fix: queue-operator dropped reconcile namespace and overflowed hash-range status 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. --- apis/kmsvc/v1/queue_types.go | 9 ++++++--- cmd/queue-operator/main.go | 2 +- config/crd/kmsvc.io_queues.yaml | 9 ++++++--- internal/operator/queue_controller.go | 14 +++++++------- internal/operator/queue_controller_test.go | 22 +++++++++++----------- internal/operator/shard_split.go | 6 +++--- k8s/charts/queue-crd/templates/crd.yaml | 9 ++++++--- 7 files changed, 40 insertions(+), 31 deletions(-) diff --git a/apis/kmsvc/v1/queue_types.go b/apis/kmsvc/v1/queue_types.go index 60235d6..00c392d 100644 --- a/apis/kmsvc/v1/queue_types.go +++ b/apis/kmsvc/v1/queue_types.go @@ -102,9 +102,12 @@ type ShardStatus struct { Topic string `json:"topic"` // HashRangeStart/HashRangeEnd define the [start, end) murmur2 hash range - // this shard owns over the 32-bit key space. - HashRangeStart uint32 `json:"hashRangeStart"` - HashRangeEnd uint32 `json:"hashRangeEnd"` + // this shard owns over the 32-bit key space. Stored as int64 (not uint32) + // 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. + HashRangeStart int64 `json:"hashRangeStart"` + HashRangeEnd int64 `json:"hashRangeEnd"` // Phase is this shard's lifecycle state. // +kubebuilder:validation:Enum=Active;Closing;Closed diff --git a/cmd/queue-operator/main.go b/cmd/queue-operator/main.go index 36910e3..b773574 100644 --- a/cmd/queue-operator/main.go +++ b/cmd/queue-operator/main.go @@ -72,7 +72,7 @@ func main() { err = ctrl.NewControllerManagedBy(mgr). For(&kmsvcv1.Queue{}). 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{}, nil diff --git a/config/crd/kmsvc.io_queues.yaml b/config/crd/kmsvc.io_queues.yaml index 1f5eefe..862c40c 100644 --- a/config/crd/kmsvc.io_queues.yaml +++ b/config/crd/kmsvc.io_queues.yaml @@ -218,13 +218,16 @@ spec: format: date-time type: string hashRangeEnd: - format: int32 + format: int64 type: integer hashRangeStart: description: |- HashRangeStart/HashRangeEnd define the [start, end) murmur2 hash range - this shard owns over the 32-bit key space. - format: int32 + this shard owns over the 32-bit key space. Stored as int64 (not uint32) + 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 id: description: ID is the shard's identifier, used in its topic diff --git a/internal/operator/queue_controller.go b/internal/operator/queue_controller.go index 5bf1ad4..2768aad 100644 --- a/internal/operator/queue_controller.go +++ b/internal/operator/queue_controller.go @@ -57,14 +57,14 @@ func (r *QueueReconciler) now() time.Time { } // 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 - 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) { return 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() { @@ -113,8 +113,8 @@ func (r *QueueReconciler) newShard(id, parentID string, start, end uint32, queue return kmsvcv1.ShardStatus{ ID: id, Topic: kafka.ShardTopicName(queue.Name, queue.Spec.FIFOQueue, id), - HashRangeStart: start, - HashRangeEnd: end, + HashRangeStart: int64(start), + HashRangeEnd: int64(end), Phase: kmsvcv1.ShardPhaseActive, ParentID: parentID, CreatedAt: metav1.NewTime(r.now()), @@ -168,8 +168,8 @@ func (r *QueueReconciler) publishRedisState(ctx context.Context, queue *kmsvcv1. shards = append(shards, kafka.Shard{ ID: s.ID, Topic: s.Topic, - HashRangeStart: s.HashRangeStart, - HashRangeEnd: s.HashRangeEnd, + HashRangeStart: uint32(s.HashRangeStart), + HashRangeEnd: uint32(s.HashRangeEnd), Phase: string(s.Phase), }) } diff --git a/internal/operator/queue_controller_test.go b/internal/operator/queue_controller_test.go index b96b149..b0f7ee5 100644 --- a/internal/operator/queue_controller_test.go +++ b/internal/operator/queue_controller_test.go @@ -76,7 +76,7 @@ func TestReconcileCreatesShardZeroAndPublishesRedisState(t *testing.T) { r, admin := newTestReconciler(t, queue) ctx := context.Background() - if err := r.Reconcile(ctx, "orders"); err != nil { + if err := r.Reconcile(ctx, "", "orders"); err != nil { t.Fatalf("Reconcile: %v", err) } @@ -118,7 +118,7 @@ func TestReconcileRejectsDLQSelfReference(t *testing.T) { r, _ := newTestReconciler(t, queue) ctx := context.Background() - err := r.Reconcile(ctx, "orders") + err := r.Reconcile(ctx, "", "orders") if err == nil { t.Fatal("expected reconcile to fail for self-referencing DLQ") } @@ -137,7 +137,7 @@ func TestReconcileDeleteCleansUpTopicsAndRedis(t *testing.T) { r, admin := newTestReconciler(t, queue) 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) } @@ -150,7 +150,7 @@ func TestReconcileDeleteCleansUpTopicsAndRedis(t *testing.T) { if err := r.Client.Delete(ctx, &got); err != nil { 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) } @@ -174,7 +174,7 @@ func TestReconcileSplitsShardOverThreshold(t *testing.T) { r.Now = func() time.Time { return now } 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) } var got kmsvcv1.Queue @@ -185,7 +185,7 @@ func TestReconcileSplitsShardOverThreshold(t *testing.T) { admin.setOffsetSum(shard0Topic, 0) // 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) } @@ -194,7 +194,7 @@ func TestReconcileSplitsShardOverThreshold(t *testing.T) { now = now.Add(time.Second) 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) } @@ -220,13 +220,13 @@ func TestReconcileSplitsShardOverThreshold(t *testing.T) { if len(children) != 2 { 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{} for _, c := range children { if c.Phase != kmsvcv1.ShardPhaseActive { 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}] { 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 } 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) } var got kmsvcv1.Queue @@ -259,7 +259,7 @@ func TestReconcileDrainsClosingShardWhenLagZero(t *testing.T) { topic := got.Status.Shards[0].Topic 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) } diff --git a/internal/operator/shard_split.go b/internal/operator/shard_split.go index 9a84aca..a2879f2 100644 --- a/internal/operator/shard_split.go +++ b/internal/operator/shard_split.go @@ -68,13 +68,13 @@ func (r *QueueReconciler) reconcileSplits(ctx context.Context, queue *kmsvcv1.Qu continue } - mid := kafka.SplitHashRange(s.HashRangeStart, s.HashRangeEnd) + mid := kafka.SplitHashRange(uint32(s.HashRangeStart), uint32(s.HashRangeEnd)) 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) childAIDNum, _ := strconv.Atoi(childAID) 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) s.Phase = kmsvcv1.ShardPhaseClosing diff --git a/k8s/charts/queue-crd/templates/crd.yaml b/k8s/charts/queue-crd/templates/crd.yaml index 1f5eefe..862c40c 100644 --- a/k8s/charts/queue-crd/templates/crd.yaml +++ b/k8s/charts/queue-crd/templates/crd.yaml @@ -218,13 +218,16 @@ spec: format: date-time type: string hashRangeEnd: - format: int32 + format: int64 type: integer hashRangeStart: description: |- HashRangeStart/HashRangeEnd define the [start, end) murmur2 hash range - this shard owns over the 32-bit key space. - format: int32 + this shard owns over the 32-bit key space. Stored as int64 (not uint32) + 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 id: description: ID is the shard's identifier, used in its topic