diff --git a/apis/kmsvc/v1/queue_types.go b/apis/kmsvc/v1/queue_types.go index 00c392d..3df3324 100644 --- a/apis/kmsvc/v1/queue_types.go +++ b/apis/kmsvc/v1/queue_types.go @@ -121,6 +121,13 @@ type ShardStatus struct { // CreatedAt timestamps when this shard was created, used to enforce // ShardSplitCooldownSeconds. CreatedAt metav1.Time `json:"createdAt,omitempty"` + + // AvailabilityZones lists the topology.kubernetes.io/zone values of every + // node currently hosting a Kafka replica of this shard's topic, resolved + // from the broker pods' node placement each reconcile. Empty until the + // first successful resolution (e.g. node lookup failed transiently). + // +optional + AvailabilityZones []string `json:"availabilityZones,omitempty"` } // QueueStatus defines the observed state of a Queue. diff --git a/cmd/queue-operator/main.go b/cmd/queue-operator/main.go index b773574..97a3a76 100644 --- a/cmd/queue-operator/main.go +++ b/cmd/queue-operator/main.go @@ -67,6 +67,16 @@ func main() { Admin: admin, Redis: rdb, Now: time.Now, + Zones: &operator.ZoneLocator{ + // GetAPIReader(), not GetClient(): the latter is cache-backed and + // would make controller-runtime List+Watch all Pods/Nodes + // cluster-wide just to serve occasional point Gets here. A direct + // (uncached) reader needs only "get" RBAC, not "list"/"watch". + Client: mgr.GetAPIReader(), + Namespace: getEnv("KMSVC_NAMESPACE", "sqs"), + ClusterName: getEnv("KMSVC_KAFKA_CLUSTER_NAME", "kmsvc"), + PoolName: getEnv("KMSVC_KAFKA_POOL_NAME", "kmsvc-pool"), + }, } err = ctrl.NewControllerManagedBy(mgr). diff --git a/config/crd/kmsvc.io_queues.yaml b/config/crd/kmsvc.io_queues.yaml index 862c40c..df684c8 100644 --- a/config/crd/kmsvc.io_queues.yaml +++ b/config/crd/kmsvc.io_queues.yaml @@ -211,6 +211,15 @@ spec: description: ShardStatus describes one shard backing a Queue (design.md §2a/§2c). properties: + availabilityZones: + description: |- + AvailabilityZones lists the topology.kubernetes.io/zone values of every + node currently hosting a Kafka replica of this shard's topic, resolved + from the broker pods' node placement each reconcile. Empty until the + first successful resolution (e.g. node lookup failed transiently). + items: + type: string + type: array createdAt: description: |- CreatedAt timestamps when this shard was created, used to enforce diff --git a/internal/kafka/admin.go b/internal/kafka/admin.go index 72c3a4c..d231599 100644 --- a/internal/kafka/admin.go +++ b/internal/kafka/admin.go @@ -81,6 +81,33 @@ func (a *Admin) DescribeTopic(ctx context.Context, topic string) (partitions int return int32(len(detail.Partitions)), true, nil } +// ReplicaBrokerIDs returns the set of broker IDs holding any replica of any +// partition of topic, used by the queue-operator to resolve which +// availability zones (node labels) a shard's data actually lives on +// (design.md §2a AZ-awareness). +func (a *Admin) ReplicaBrokerIDs(ctx context.Context, topic string) ([]int32, error) { + td, err := a.client.ListTopics(ctx, topic) + if err != nil { + return nil, fmt.Errorf("list topics %q: %w", topic, err) + } + detail, found := td[topic] + if !found || detail.Err != nil { + return nil, nil + } + + seen := make(map[int32]bool) + var ids []int32 + for _, p := range detail.Partitions { + for _, id := range p.Replicas { + if !seen[id] { + seen[id] = true + ids = append(ids, id) + } + } + } + return ids, nil +} + // LogEndOffsetSum returns the sum of the high-watermark offsets across every // partition of a topic — a monotonically increasing proxy for total records // produced, used by the queue-operator's shard-split sampler (design.md §2c) diff --git a/internal/operator/admin.go b/internal/operator/admin.go index 0111222..a31182d 100644 --- a/internal/operator/admin.go +++ b/internal/operator/admin.go @@ -13,4 +13,5 @@ type TopicAdmin interface { DeleteTopic(ctx context.Context, topic string) error LogEndOffsetSum(ctx context.Context, topic string) (int64, error) ConsumerLag(ctx context.Context, group, topic string) (int64, error) + ReplicaBrokerIDs(ctx context.Context, topic string) ([]int32, error) } diff --git a/internal/operator/fake_admin.go b/internal/operator/fake_admin.go index 1dc8f2f..9df6189 100644 --- a/internal/operator/fake_admin.go +++ b/internal/operator/fake_admin.go @@ -51,6 +51,15 @@ func (f *fakeAdmin) ConsumerLag(_ context.Context, group, topic string) (int64, return f.lag[group+"/"+topic], nil } +func (f *fakeAdmin) ReplicaBrokerIDs(_ context.Context, topic string) ([]int32, error) { + f.mu.Lock() + defer f.mu.Unlock() + if _, ok := f.topics[topic]; !ok { + return nil, nil + } + return []int32{0}, nil +} + func (f *fakeAdmin) setOffsetSum(topic string, v int64) { f.mu.Lock() defer f.mu.Unlock() diff --git a/internal/operator/queue_controller.go b/internal/operator/queue_controller.go index 2768aad..2608080 100644 --- a/internal/operator/queue_controller.go +++ b/internal/operator/queue_controller.go @@ -15,6 +15,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + ctrllog "sigs.k8s.io/controller-runtime/pkg/log" kmsvcv1 "github.com/rockliang/kafka-management-service/apis/kmsvc/v1" "github.com/rockliang/kafka-management-service/internal/kafka" @@ -39,6 +40,11 @@ type QueueReconciler struct { Redis *goredis.Client Now func() time.Time + // Zones resolves shard topics' broker placement to availability zones + // (design.md §2a AZ-awareness). Nil disables zone annotation entirely -- + // tests and any deployment without zone-labeled nodes can leave it unset. + Zones *ZoneLocator + // sampleState tracks the last (offsetSum, time) seen per shard topic, used // to compute a throughput estimate between reconciles. Keyed by topic name. sampleState map[string]sample @@ -90,6 +96,8 @@ func (r *QueueReconciler) Reconcile(ctx context.Context, namespace, name string) return r.setFailed(ctx, &queue, "EnsureTopics", err) } + r.annotateAvailabilityZones(ctx, &queue) + if err := r.reconcileSplits(ctx, &queue); err != nil { return r.setFailed(ctx, &queue, "ShardSplit", err) } @@ -139,6 +147,43 @@ func (r *QueueReconciler) ensureShardTopics(ctx context.Context, queue *kmsvcv1. return nil } +// annotateAvailabilityZones resolves and stamps each non-closed shard's +// AvailabilityZones (design.md §2a). Best-effort: a resolution failure for +// one shard is logged via setFailed-style swallowing -- it must never block +// the rest of reconciliation, since AZ info is status metadata, not +// load-bearing for the queue's actual operation. +func (r *QueueReconciler) annotateAvailabilityZones(ctx context.Context, queue *kmsvcv1.Queue) { + if r.Zones == nil { + return + } + logger := ctrllog.FromContext(ctx) + for i := range queue.Status.Shards { + s := &queue.Status.Shards[i] + if s.Phase == kmsvcv1.ShardPhaseClosed { + continue + } + brokerIDs, err := r.Admin.ReplicaBrokerIDs(ctx, s.Topic) + if err != nil { + logger.Error(err, "resolve replica broker IDs", "topic", s.Topic) + continue + } + if len(brokerIDs) == 0 { + logger.Info("no replica broker IDs returned", "topic", s.Topic) + continue + } + zones, err := r.Zones.ZonesForBrokers(ctx, brokerIDs) + if err != nil { + logger.Error(err, "resolve zones for brokers", "topic", s.Topic, "brokerIDs", brokerIDs) + continue + } + if len(zones) == 0 { + logger.Info("no zones resolved", "topic", s.Topic, "brokerIDs", brokerIDs) + continue + } + s.AvailabilityZones = zones + } +} + func (r *QueueReconciler) setFailed(ctx context.Context, queue *kmsvcv1.Queue, reason string, cause error) error { queue.Status.Phase = kmsvcv1.QueuePhaseFailed if err := r.Client.Status().Update(ctx, queue); err != nil { diff --git a/internal/operator/zone_locator.go b/internal/operator/zone_locator.go new file mode 100644 index 0000000..bc7dcf9 --- /dev/null +++ b/internal/operator/zone_locator.go @@ -0,0 +1,93 @@ +package operator + +import ( + "context" + "fmt" + "sort" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// zoneLabel is the standard k8s topology label this cluster's Talos nodes +// carry (confirmed via `kubectl get nodes --show-labels`). +const zoneLabel = "topology.kubernetes.io/zone" + +// ZoneLocator resolves which availability zones (node labels) a set of +// Kafka broker IDs currently run in, for AZ-aware Queue status (design.md +// §2a). Broker pods follow Strimzi's StrimziPodSet naming convention: +// "--" (confirmed live: broker id 0 is pod +// kmsvc-kmsvc-pool-0, etc. -- broker ID equals the pod's ordinal suffix). +type ZoneLocator struct { + // Client only needs Get, so callers can pass an uncached client.Reader + // (e.g. mgr.GetAPIReader()) and avoid requiring cluster-wide list/watch + // RBAC on Pods/Nodes just to serve occasional point lookups here. + Client client.Reader + Namespace string + ClusterName string + PoolName string + + // nodeZoneCache avoids a Node lookup per broker per reconcile -- node + // zone labels don't change while a node is up, and this is best-effort + // (cleared implicitly by process restart) rather than watched for + // correctness. + nodeZoneCache map[string]string +} + +// ZonesForBrokers returns the sorted, de-duplicated set of zones the given +// broker IDs currently run in. A broker whose pod or node can't be resolved +// is skipped rather than failing the whole call -- AZ info is best-effort +// status, not load-bearing for reconciliation. +func (z *ZoneLocator) ZonesForBrokers(ctx context.Context, brokerIDs []int32) ([]string, error) { + if z.nodeZoneCache == nil { + z.nodeZoneCache = make(map[string]string) + } + + seen := make(map[string]bool) + var zones []string + for _, id := range brokerIDs { + nodeName, err := z.nodeNameForBroker(ctx, id) + if err != nil || nodeName == "" { + continue + } + zone, err := z.zoneForNode(ctx, nodeName) + if err != nil || zone == "" { + continue + } + if !seen[zone] { + seen[zone] = true + zones = append(zones, zone) + } + } + sort.Strings(zones) + return zones, nil +} + +func (z *ZoneLocator) nodeNameForBroker(ctx context.Context, brokerID int32) (string, error) { + podName := fmt.Sprintf("%s-%s-%d", z.ClusterName, z.PoolName, brokerID) + var pod corev1.Pod + if err := z.Client.Get(ctx, client.ObjectKey{Namespace: z.Namespace, Name: podName}, &pod); err != nil { + if apierrors.IsNotFound(err) { + return "", nil + } + return "", fmt.Errorf("get broker pod %s: %w", podName, err) + } + return pod.Spec.NodeName, nil +} + +func (z *ZoneLocator) zoneForNode(ctx context.Context, nodeName string) (string, error) { + if zone, ok := z.nodeZoneCache[nodeName]; ok { + return zone, nil + } + var node corev1.Node + if err := z.Client.Get(ctx, client.ObjectKey{Name: nodeName}, &node); err != nil { + if apierrors.IsNotFound(err) { + return "", nil + } + return "", fmt.Errorf("get node %s: %w", nodeName, err) + } + zone := node.Labels[zoneLabel] + z.nodeZoneCache[nodeName] = zone + return zone, nil +} diff --git a/k8s/charts/queue-crd/templates/crd.yaml b/k8s/charts/queue-crd/templates/crd.yaml index 862c40c..df684c8 100644 --- a/k8s/charts/queue-crd/templates/crd.yaml +++ b/k8s/charts/queue-crd/templates/crd.yaml @@ -211,6 +211,15 @@ spec: description: ShardStatus describes one shard backing a Queue (design.md §2a/§2c). properties: + availabilityZones: + description: |- + AvailabilityZones lists the topology.kubernetes.io/zone values of every + node currently hosting a Kafka replica of this shard's topic, resolved + from the broker pods' node placement each reconcile. Empty until the + first successful resolution (e.g. node lookup failed transiently). + items: + type: string + type: array createdAt: description: |- CreatedAt timestamps when this shard was created, used to enforce diff --git a/k8s/charts/queue-crd/templates/operator-deployment.yaml b/k8s/charts/queue-crd/templates/operator-deployment.yaml index 3f886ca..22c9a57 100644 --- a/k8s/charts/queue-crd/templates/operator-deployment.yaml +++ b/k8s/charts/queue-crd/templates/operator-deployment.yaml @@ -25,5 +25,13 @@ spec: value: {{ .Values.redisAddr | quote }} - name: GOMEMLIMIT value: {{ .Values.goMemLimit | quote }} + - name: KMSVC_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: KMSVC_KAFKA_CLUSTER_NAME + value: {{ .Values.kafkaClusterName | quote }} + - name: KMSVC_KAFKA_POOL_NAME + value: {{ .Values.kafkaPoolName | quote }} resources: {{- toYaml .Values.resources | nindent 12 }} diff --git a/k8s/charts/queue-crd/templates/rbac.yaml b/k8s/charts/queue-crd/templates/rbac.yaml index 4939664..39a46d7 100644 --- a/k8s/charts/queue-crd/templates/rbac.yaml +++ b/k8s/charts/queue-crd/templates/rbac.yaml @@ -24,6 +24,9 @@ rules: - apiGroups: [""] resources: ["events"] verbs: ["create", "patch"] + - apiGroups: [""] + resources: ["pods", "nodes"] + verbs: ["get"] --- apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding diff --git a/k8s/charts/queue-crd/values.yaml b/k8s/charts/queue-crd/values.yaml index 1069077..6828ba8 100644 --- a/k8s/charts/queue-crd/values.yaml +++ b/k8s/charts/queue-crd/values.yaml @@ -8,6 +8,12 @@ image: kafkaBrokers: "kmsvc-kafka-bootstrap.sqs.svc.cluster.local:9092" redisAddr: "kmsvc-redis-master.sqs.svc.cluster.local:6379" +# Must match kafka-cluster chart's clusterName/derived pool name -- used to +# resolve "--" broker pod names +# for AZ-aware Queue status (design.md §2a). +kafkaClusterName: kmsvc +kafkaPoolName: kmsvc-pool + resources: requests: cpu: 100m