riotpiaole
9fa5420b22
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.
276 lines
8.4 KiB
Go
276 lines
8.4 KiB
Go
package operator
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/alicebob/miniredis/v2"
|
|
goredis "github.com/redis/go-redis/v9"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
clientgoscheme "k8s.io/client-go/kubernetes/scheme"
|
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
|
"sigs.k8s.io/controller-runtime/pkg/client/fake"
|
|
|
|
kmsvcv1 "github.com/rockliang/kafka-management-service/apis/kmsvc/v1"
|
|
"github.com/rockliang/kafka-management-service/internal/kafka"
|
|
kmsvcredis "github.com/rockliang/kafka-management-service/internal/redis"
|
|
)
|
|
|
|
func newTestScheme(t *testing.T) *runtime.Scheme {
|
|
t.Helper()
|
|
scheme := runtime.NewScheme()
|
|
if err := clientgoscheme.AddToScheme(scheme); err != nil {
|
|
t.Fatalf("add client-go scheme: %v", err)
|
|
}
|
|
if err := kmsvcv1.AddToScheme(scheme); err != nil {
|
|
t.Fatalf("add kmsvc scheme: %v", err)
|
|
}
|
|
return scheme
|
|
}
|
|
|
|
func newTestRedis(t *testing.T) *goredis.Client {
|
|
t.Helper()
|
|
mr, err := miniredis.Run()
|
|
if err != nil {
|
|
t.Fatalf("starting miniredis: %v", err)
|
|
}
|
|
t.Cleanup(mr.Close)
|
|
return goredis.NewClient(&goredis.Options{Addr: mr.Addr()})
|
|
}
|
|
|
|
func newTestReconciler(t *testing.T, objs ...client.Object) (*QueueReconciler, *fakeAdmin) {
|
|
t.Helper()
|
|
scheme := newTestScheme(t)
|
|
cl := fake.NewClientBuilder().
|
|
WithScheme(scheme).
|
|
WithStatusSubresource(&kmsvcv1.Queue{}).
|
|
WithObjects(objs...).
|
|
Build()
|
|
admin := newFakeAdmin()
|
|
return &QueueReconciler{
|
|
Client: cl,
|
|
Admin: admin,
|
|
Redis: newTestRedis(t),
|
|
Now: time.Now,
|
|
}, admin
|
|
}
|
|
|
|
func baseQueue(name string) *kmsvcv1.Queue {
|
|
return &kmsvcv1.Queue{
|
|
ObjectMeta: metav1.ObjectMeta{Name: name},
|
|
Spec: kmsvcv1.QueueSpec{
|
|
VisibilityTimeoutSeconds: 30,
|
|
MessageRetentionPeriodSeconds: 345600,
|
|
MaxReceiveCount: 5,
|
|
PartitionsPerShard: 6,
|
|
MinShards: 1,
|
|
MaxShards: 8,
|
|
},
|
|
}
|
|
}
|
|
|
|
func TestReconcileCreatesShardZeroAndPublishesRedisState(t *testing.T) {
|
|
queue := baseQueue("orders")
|
|
r, admin := newTestReconciler(t, queue)
|
|
ctx := context.Background()
|
|
|
|
if err := r.Reconcile(ctx, "", "orders"); err != nil {
|
|
t.Fatalf("Reconcile: %v", err)
|
|
}
|
|
|
|
var got kmsvcv1.Queue
|
|
if err := r.Client.Get(ctx, client.ObjectKey{Name: "orders"}, &got); err != nil {
|
|
t.Fatalf("get queue: %v", err)
|
|
}
|
|
if got.Status.Phase != kmsvcv1.QueuePhaseReady {
|
|
t.Fatalf("phase = %v, want Ready", got.Status.Phase)
|
|
}
|
|
if len(got.Status.Shards) != 1 || got.Status.Shards[0].ID != "0" {
|
|
t.Fatalf("shards = %+v, want one shard-0", got.Status.Shards)
|
|
}
|
|
wantTopic := kafka.ShardTopicName("orders", false, "0")
|
|
if got.Status.Shards[0].Topic != wantTopic {
|
|
t.Errorf("shard-0 topic = %q, want %q", got.Status.Shards[0].Topic, wantTopic)
|
|
}
|
|
if !admin.hasTopic(wantTopic) {
|
|
t.Errorf("expected topic %q to be created", wantTopic)
|
|
}
|
|
|
|
meta, ok, err := kmsvcredis.GetQueueMeta(ctx, r.Redis, "orders")
|
|
if err != nil || !ok {
|
|
t.Fatalf("GetQueueMeta: ok=%v err=%v", ok, err)
|
|
}
|
|
if meta.MaxReceiveCount != 5 {
|
|
t.Errorf("meta.MaxReceiveCount = %d, want 5", meta.MaxReceiveCount)
|
|
}
|
|
|
|
shards, ok, err := kmsvcredis.GetShardMap(ctx, r.Redis, "orders")
|
|
if err != nil || !ok || len(shards) != 1 {
|
|
t.Fatalf("GetShardMap: shards=%+v ok=%v err=%v", shards, ok, err)
|
|
}
|
|
}
|
|
|
|
func TestReconcileRejectsDLQSelfReference(t *testing.T) {
|
|
queue := baseQueue("orders")
|
|
queue.Spec.DeadLetterTargetQueue = "orders"
|
|
r, _ := newTestReconciler(t, queue)
|
|
ctx := context.Background()
|
|
|
|
err := r.Reconcile(ctx, "", "orders")
|
|
if err == nil {
|
|
t.Fatal("expected reconcile to fail for self-referencing DLQ")
|
|
}
|
|
|
|
var got kmsvcv1.Queue
|
|
if err := r.Client.Get(ctx, client.ObjectKey{Name: "orders"}, &got); err != nil {
|
|
t.Fatalf("get queue: %v", err)
|
|
}
|
|
if got.Status.Phase != kmsvcv1.QueuePhaseFailed {
|
|
t.Errorf("phase = %v, want Failed", got.Status.Phase)
|
|
}
|
|
}
|
|
|
|
func TestReconcileDeleteCleansUpTopicsAndRedis(t *testing.T) {
|
|
queue := baseQueue("orders")
|
|
r, admin := newTestReconciler(t, queue)
|
|
ctx := context.Background()
|
|
|
|
if err := r.Reconcile(ctx, "", "orders"); err != nil {
|
|
t.Fatalf("initial reconcile: %v", err)
|
|
}
|
|
|
|
var got kmsvcv1.Queue
|
|
if err := r.Client.Get(ctx, client.ObjectKey{Name: "orders"}, &got); err != nil {
|
|
t.Fatalf("get queue: %v", err)
|
|
}
|
|
topic := got.Status.Shards[0].Topic
|
|
|
|
if err := r.Client.Delete(ctx, &got); err != nil {
|
|
t.Fatalf("delete queue: %v", err)
|
|
}
|
|
if err := r.Reconcile(ctx, "", "orders"); err != nil {
|
|
t.Fatalf("delete reconcile: %v", err)
|
|
}
|
|
|
|
if admin.hasTopic(topic) {
|
|
t.Errorf("expected topic %q to be deleted", topic)
|
|
}
|
|
if _, ok, _ := kmsvcredis.GetQueueMeta(ctx, r.Redis, "orders"); ok {
|
|
t.Error("expected queue meta to be removed")
|
|
}
|
|
if _, ok, _ := kmsvcredis.GetShardMap(ctx, r.Redis, "orders"); ok {
|
|
t.Error("expected shard map to be removed")
|
|
}
|
|
}
|
|
|
|
func TestReconcileSplitsShardOverThreshold(t *testing.T) {
|
|
queue := baseQueue("orders")
|
|
queue.Spec.ShardSplitThresholdBytesPerSec = 1000
|
|
queue.Spec.ShardSplitCooldownSeconds = 0
|
|
now := time.Now()
|
|
r, admin := newTestReconciler(t, queue)
|
|
r.Now = func() time.Time { return now }
|
|
ctx := context.Background()
|
|
|
|
if err := r.Reconcile(ctx, "", "orders"); err != nil {
|
|
t.Fatalf("initial reconcile: %v", err)
|
|
}
|
|
var got kmsvcv1.Queue
|
|
if err := r.Client.Get(ctx, client.ObjectKey{Name: "orders"}, &got); err != nil {
|
|
t.Fatalf("get queue: %v", err)
|
|
}
|
|
shard0Topic := got.Status.Shards[0].Topic
|
|
admin.setOffsetSum(shard0Topic, 0)
|
|
|
|
// First sample establishes the baseline (no delta to compare yet).
|
|
if err := r.Reconcile(ctx, "", "orders"); err != nil {
|
|
t.Fatalf("baseline reconcile: %v", err)
|
|
}
|
|
|
|
// Advance time and offsets enough to exceed 1000 bytes/sec at the
|
|
// 1024-byte/record estimate: 10 records over 1s ≈ 10240 bytes/sec.
|
|
now = now.Add(time.Second)
|
|
admin.setOffsetSum(shard0Topic, 10)
|
|
|
|
if err := r.Reconcile(ctx, "", "orders"); err != nil {
|
|
t.Fatalf("split-triggering reconcile: %v", err)
|
|
}
|
|
|
|
if err := r.Client.Get(ctx, client.ObjectKey{Name: "orders"}, &got); err != nil {
|
|
t.Fatalf("get queue after split: %v", err)
|
|
}
|
|
if len(got.Status.Shards) != 3 {
|
|
t.Fatalf("shards = %+v, want 3 (1 closing parent + 2 active children)", got.Status.Shards)
|
|
}
|
|
var parent *kmsvcv1.ShardStatus
|
|
var children []kmsvcv1.ShardStatus
|
|
for i := range got.Status.Shards {
|
|
s := &got.Status.Shards[i]
|
|
if s.ID == "0" {
|
|
parent = s
|
|
} else {
|
|
children = append(children, *s)
|
|
}
|
|
}
|
|
if parent == nil || parent.Phase != kmsvcv1.ShardPhaseClosing {
|
|
t.Fatalf("parent shard = %+v, want Closing", parent)
|
|
}
|
|
if len(children) != 2 {
|
|
t.Fatalf("children = %+v, want 2", children)
|
|
}
|
|
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{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)
|
|
}
|
|
}
|
|
|
|
func TestReconcileDrainsClosingShardWhenLagZero(t *testing.T) {
|
|
queue := baseQueue("orders")
|
|
queue.Spec.MessageRetentionPeriodSeconds = 60
|
|
now := time.Now()
|
|
r, admin := newTestReconciler(t, queue)
|
|
r.Now = func() time.Time { return now }
|
|
ctx := context.Background()
|
|
|
|
if err := r.Reconcile(ctx, "", "orders"); err != nil {
|
|
t.Fatalf("initial reconcile: %v", err)
|
|
}
|
|
var got kmsvcv1.Queue
|
|
if err := r.Client.Get(ctx, client.ObjectKey{Name: "orders"}, &got); err != nil {
|
|
t.Fatalf("get queue: %v", err)
|
|
}
|
|
|
|
// Manually mark shard-0 as Closing (simulating a prior split) to isolate
|
|
// drain behavior from split behavior.
|
|
got.Status.Shards[0].Phase = kmsvcv1.ShardPhaseClosing
|
|
got.Status.Shards[0].CreatedAt = metav1.NewTime(now.Add(-2 * time.Minute))
|
|
if err := r.Client.Status().Update(ctx, &got); err != nil {
|
|
t.Fatalf("seed closing shard: %v", err)
|
|
}
|
|
topic := got.Status.Shards[0].Topic
|
|
admin.setLag(kafka.ConsumerGroup("orders"), topic, 0)
|
|
|
|
if err := r.Reconcile(ctx, "", "orders"); err != nil {
|
|
t.Fatalf("drain reconcile: %v", err)
|
|
}
|
|
|
|
if err := r.Client.Get(ctx, client.ObjectKey{Name: "orders"}, &got); err != nil {
|
|
t.Fatalf("get queue after drain: %v", err)
|
|
}
|
|
if got.Status.Shards[0].Phase != kmsvcv1.ShardPhaseClosed {
|
|
t.Errorf("shard-0 phase = %v, want Closed", got.Status.Shards[0].Phase)
|
|
}
|
|
if admin.hasTopic(topic) {
|
|
t.Errorf("expected drained topic %q to be deleted", topic)
|
|
}
|
|
}
|