feat: stamp Queue shard status with availability zones

Resolves each shard topic's replica broker IDs (internal/kafka.Admin.
ReplicaBrokerIDs) to the topology.kubernetes.io/zone labels of the nodes
hosting those brokers (ZoneLocator), and writes the result into
ShardStatus.AvailabilityZones each reconcile. Uses mgr.GetAPIReader()
rather than the cached client for the Pod/Node lookups, since the cached
client would otherwise require cluster-wide list/watch RBAC on Pods just
to serve occasional point Gets.

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.

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).

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

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)

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)
 }

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()

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 {

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:
+// "<clusterName>-<poolName>-<brokerID>" (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
+}

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

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 }}

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

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 "<kafkaClusterName>-<kafkaPoolName>-<brokerID>" broker pod names
+# for AZ-aware Queue status (design.md §2a).
+kafkaClusterName: kmsvc
+kafkaPoolName: kmsvc-pool
+
 resources:
   requests:
     cpu: 100m