From 52b86ab8e8648606e1734aad9951b692532fd881 Mon Sep 17 00:00:00 2001 From: Story Crater Bot <19826264+Riotpiaole@users.noreply.github.com> Date: Mon, 13 Jul 2026 13:02:40 -0700 Subject: [PATCH] feat: queue-operator auto-registers Temporal namespace before creating TemporalWorker A Queue's temporal.io/namespace label was trusted as-is -- if the referenced Temporal namespace was never registered (or typo'd), the failure only surfaced as a worker pod silently polling a namespace that doesn't exist. Now reconcileTemporalWorker calls RegisterNamespace (idempotent, ignores AlreadyExists) via a direct WorkflowService gRPC client before creating the TemporalWorker, so namespace and worker always come into existence together. Also grant queue-operator's ClusterRole create/delete on temporalworkers (previously missing, causing forbidden errors on the create-then-delete path). --- cmd/queue-operator/main.go | 15 ++- config/rbac/role.yaml | 2 +- go.mod | 2 + go.sum | 4 + hack/test-queues-and-workflow.sh | 111 +++++++++++++++++++++ internal/operator/fake_temporal.go | 41 ++++++++ internal/operator/queue_controller.go | 13 ++- internal/operator/queue_controller_test.go | 57 +++++++++-- internal/operator/temporal_namespace.go | 14 +++ internal/temporal/client.go | 55 ++++++++++ 10 files changed, 299 insertions(+), 15 deletions(-) create mode 100755 hack/test-queues-and-workflow.sh create mode 100644 internal/operator/fake_temporal.go create mode 100644 internal/operator/temporal_namespace.go create mode 100644 internal/temporal/client.go diff --git a/cmd/queue-operator/main.go b/cmd/queue-operator/main.go index 2ae66ff..262af31 100644 --- a/cmd/queue-operator/main.go +++ b/cmd/queue-operator/main.go @@ -23,6 +23,7 @@ import ( kmsvcv1 "github.com/rockliang/kafka-management-service/apis/kmsvc/v1" "github.com/rockliang/kafka-management-service/internal/kafka" "github.com/rockliang/kafka-management-service/internal/operator" + kmsvctemporal "github.com/rockliang/kafka-management-service/internal/temporal" ) func main() { @@ -66,11 +67,17 @@ func main() { exitf("connecting to redis at %s: %v", redisAddr, err) } + temporalClient, err := kmsvctemporal.NewClient(getEnv("KMSVC_TEMPORAL_FRONTEND_ADDRESS", "temporal-frontend.temporal.svc.cluster.local:7233")) + if err != nil { + exitf("creating temporal client: %v", err) + } + reconciler := &operator.QueueReconciler{ - Client: mgr.GetClient(), - Admin: admin, - Redis: rdb, - Now: time.Now, + Client: mgr.GetClient(), + Admin: admin, + Redis: rdb, + Now: time.Now, + Temporal: temporalClient, Zones: &operator.ZoneLocator{ // GetAPIReader(), not GetClient(): the latter is cache-backed and // would make controller-runtime List+Watch all Pods/Nodes diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index 7637d55..e778355 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -14,7 +14,7 @@ rules: verbs: ["update"] - apiGroups: ["kmsvc.io"] resources: ["temporalworkers"] - verbs: ["get", "list", "watch", "update", "patch"] + verbs: ["get", "list", "watch", "create", "update", "patch", "delete"] - apiGroups: ["kmsvc.io"] resources: ["temporalworkers/status"] verbs: ["get", "update", "patch"] diff --git a/go.mod b/go.mod index bcbb65b..29040c7 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/twmb/franz-go v1.21.3 github.com/twmb/franz-go/pkg/kadm v1.18.0 github.com/twmb/franz-go/pkg/kfake v0.0.0-20260615024848-f17c00130060 + go.temporal.io/api v1.63.3 google.golang.org/grpc v1.81.1 google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af k8s.io/api v0.36.2 @@ -48,6 +49,7 @@ require ( github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/nexus-rpc/nexus-proto-annotations v0.1.0 // indirect github.com/pierrec/lz4/v4 v4.1.26 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/prometheus/client_golang v1.23.2 // indirect diff --git a/go.sum b/go.sum index 0fe8e94..e116f0c 100644 --- a/go.sum +++ b/go.sum @@ -101,6 +101,8 @@ github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFd github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/nexus-rpc/nexus-proto-annotations v0.1.0 h1:2fELd+9sqUtNu6Fg//pw8YFsxOvp8vZ8hfP0nHhNI80= +github.com/nexus-rpc/nexus-proto-annotations v0.1.0/go.mod h1:n3UjF1bPCW8llR8tHvbxJ+27yPWrhpo8w/Yg1IOuY0Y= github.com/onsi/ginkgo/v2 v2.27.4 h1:fcEcQW/A++6aZAZQNUmNjvA9PSOzefMJBerHJ4t8v8Y= github.com/onsi/ginkgo/v2 v2.27.4/go.mod h1:ArE1D/XhNXBXCBkKOLkbsb2c81dQHCRcF5zwn/ykDRo= github.com/onsi/gomega v1.39.0 h1:y2ROC3hKFmQZJNFeGAMeHZKkjBL65mIZcvrLQBF9k6Q= @@ -166,6 +168,8 @@ go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfC go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +go.temporal.io/api v1.63.3 h1:09yoemfjnk1YHV6g402lMW1vZccUd9Au/NfQBEZC0Eo= +go.temporal.io/api v1.63.3/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= diff --git a/hack/test-queues-and-workflow.sh b/hack/test-queues-and-workflow.sh new file mode 100755 index 0000000..5b9129f --- /dev/null +++ b/hack/test-queues-and-workflow.sh @@ -0,0 +1,111 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Smoke test for the queue-operator -> TemporalWorker pipeline: +# 1. send/receive/delete a message through each Queue in +# k8s/temporal/queues/example-queue.yaml (story-crater-tasks, +# story-crater-notifications) via kmsvc-cli. +# 2. start a helloworld workflow via the temporal CLI against the +# Temporal namespace those queues point at (temporal.io/namespace +# label, "production"), on the task queue the auto-created +# TemporalWorker polls ("worker-production"). +# +# Workflow execution will not complete -- there is no real worker image +# registering a "HelloWorldWorkflow" handler yet (TemporalWorker's Deployment +# is running the KMSVC_TEMPORAL_WORKER_IMAGE placeholder). This only proves +# the control-plane path: namespace exists, task queue routing works, and +# StartWorkflowExecution succeeds end-to-end through the operator-managed +# pipeline. `temporal workflow describe` afterward should show it stuck in +# Running/WorkflowTaskScheduled, which is the expected signal that this half +# of the pipe is wired correctly. +# +# Env overrides: +# NAMESPACE k8s namespace the Queue CRDs + management-service live in (default: sqs) +# TEMPORAL_NAMESPACE Temporal namespace the queues are labeled with (default: production) +# TASK_QUEUE Temporal task queue the TemporalWorker polls (default: worker-production) +# LOCAL_PORT local port for the management-service port-forward (default: 9090) +# TEMPORAL_LOCAL_PORT local port for the temporal-frontend port-forward (default: 7233) +# AUTHENTIK_TOKEN_URL Authentik OAuth2 token endpoint + +QUEUES=("story-crater-tasks" "story-crater-notifications") +NAMESPACE="${NAMESPACE:-sqs}" +TEMPORAL_NAMESPACE="${TEMPORAL_NAMESPACE:-production}" +TASK_QUEUE="${TASK_QUEUE:-worker-production}" +LOCAL_PORT="${LOCAL_PORT:-9090}" +TEMPORAL_LOCAL_PORT="${TEMPORAL_LOCAL_PORT:-7233}" +SERVER="127.0.0.1:${LOCAL_PORT}" +AUTHENTIK_TOKEN_URL="${AUTHENTIK_TOKEN_URL:-https://authentik.riotpiao.homelab.com/application/o/token/}" + +CLIENT_ID=$(talos get cluster/AUTHENTIK_KAFAKA_CLIENT_ID --key AUTHENTIK_KAFAKA_CLIENT_ID) +CLIENT_SECRET=$(talos get cluster/AUTHENTIK_KAFAKA_CLIENT_SECRET --key AUTHENTIK_KAFAKA_CLIENT_SECRET) + +TOKEN=$(curl -sf -X POST "${AUTHENTIK_TOKEN_URL}" \ + -d grant_type=client_credentials \ + -d client_id="${CLIENT_ID}" \ + -d client_secret="${CLIENT_SECRET}" \ + | jq -r '.access_token') + +if [[ -z "${TOKEN}" || "${TOKEN}" == "null" ]]; then + echo "FAIL: could not obtain access token from Authentik" >&2 + exit 1 +fi + +kubectl port-forward -n "${NAMESPACE}" svc/management-service "${LOCAL_PORT}:9090" >/tmp/kmsvc-port-forward.log 2>&1 & +KMSVC_PF_PID=$! +kubectl port-forward -n temporal svc/temporal-frontend "${TEMPORAL_LOCAL_PORT}:7233" >/tmp/temporal-port-forward.log 2>&1 & +TEMPORAL_PF_PID=$! +sleep 2 + +cleanup() { + kill "${KMSVC_PF_PID}" "${TEMPORAL_PF_PID}" 2>/dev/null || true +} +trap cleanup EXIT + +KMSVC=(kmsvc --server "${SERVER}" --token "${TOKEN}" --insecure) + +for QUEUE_NAME in "${QUEUES[@]}"; do + echo "=== ${QUEUE_NAME} ===" + + PHASE=$(kubectl get queue "${QUEUE_NAME}" -n "${NAMESPACE}" -o jsonpath='{.status.phase}' 2>/dev/null || true) + if [[ "${PHASE}" != "Ready" ]]; then + echo "FAIL: queue ${QUEUE_NAME} phase=${PHASE}, want Ready" >&2 + exit 1 + fi + + echo "--- send ---" + "${KMSVC[@]}" message send --queue "${QUEUE_NAME}" --body "hello from ${QUEUE_NAME}" + + echo "--- receive ---" + RECEIVE_OUT=$("${KMSVC[@]}" message receive --queue "${QUEUE_NAME}" --max-messages 1 --wait 10 --output json) + echo "${RECEIVE_OUT}" + + RECEIPT_HANDLE=$(echo "${RECEIVE_OUT}" | jq -r '.[0].receipt_handle // .[0].ReceiptHandle') + if [[ -z "${RECEIPT_HANDLE}" || "${RECEIPT_HANDLE}" == "null" ]]; then + echo "FAIL: no message received from ${QUEUE_NAME}" >&2 + exit 1 + fi + + echo "--- delete (ack) ---" + "${KMSVC[@]}" message delete --queue "${QUEUE_NAME}" --receipt-handle "${RECEIPT_HANDLE}" + + echo "OK: ${QUEUE_NAME} round-trip succeeded" +done + +echo "=== temporal: start helloworld workflow ===" +WORKFLOW_ID="hack-helloworld-$(date +%s)" +temporal workflow start \ + --address "127.0.0.1:${TEMPORAL_LOCAL_PORT}" \ + --namespace "${TEMPORAL_NAMESPACE}" \ + --task-queue "${TASK_QUEUE}" \ + --type HelloWorldWorkflow \ + --workflow-id "${WORKFLOW_ID}" \ + --input '"hack smoke test"' + +echo "--- describe ---" +temporal workflow describe \ + --address "127.0.0.1:${TEMPORAL_LOCAL_PORT}" \ + --namespace "${TEMPORAL_NAMESPACE}" \ + --workflow-id "${WORKFLOW_ID}" + +echo "OK: workflow ${WORKFLOW_ID} started on namespace=${TEMPORAL_NAMESPACE} task-queue=${TASK_QUEUE}" +echo "NOTE: it will not complete -- no worker is registering HelloWorldWorkflow yet (placeholder image on worker-production)." diff --git a/internal/operator/fake_temporal.go b/internal/operator/fake_temporal.go new file mode 100644 index 0000000..48a52b2 --- /dev/null +++ b/internal/operator/fake_temporal.go @@ -0,0 +1,41 @@ +package operator + +import ( + "context" + "sync" +) + +// fakeTemporal is an in-memory TemporalNamespaceRegisterer for reconciler +// tests — avoids needing a real Temporal frontend just to exercise reconcile +// logic. +type fakeTemporal struct { + mu sync.Mutex + registered map[string]int + err error +} + +func newFakeTemporal() *fakeTemporal { + return &fakeTemporal{registered: map[string]int{}} +} + +func (f *fakeTemporal) RegisterNamespace(_ context.Context, namespace string) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.err != nil { + return f.err + } + f.registered[namespace]++ + return nil +} + +func (f *fakeTemporal) setErr(err error) { + f.mu.Lock() + defer f.mu.Unlock() + f.err = err +} + +func (f *fakeTemporal) count(namespace string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.registered[namespace] +} diff --git a/internal/operator/queue_controller.go b/internal/operator/queue_controller.go index c28988b..2ca9729 100644 --- a/internal/operator/queue_controller.go +++ b/internal/operator/queue_controller.go @@ -36,10 +36,11 @@ const minInsyncReplicas = 2 // QueueReconciler reconciles Queue objects (design.md §2a). type QueueReconciler struct { - Client client.Client - Admin TopicAdmin - Redis *goredis.Client - Now func() time.Time + Client client.Client + Admin TopicAdmin + Redis *goredis.Client + Now func() time.Time + Temporal TemporalNamespaceRegisterer // Zones resolves shard topics' broker placement to availability zones // (design.md §2a AZ-awareness). Nil disables zone annotation entirely -- @@ -248,6 +249,10 @@ func (r *QueueReconciler) reconcileTemporalWorker(ctx context.Context, queue *km return fmt.Errorf("invalid kubernetes name %q: %w", workerName, err) } + if err := r.Temporal.RegisterNamespace(ctx, namespace); err != nil { + return fmt.Errorf("register temporal namespace %s: %w", namespace, err) + } + replicas := int32(1) workerNamespace := getEnvOrDefault("KMSVC_TEMPORAL_NAMESPACE", "temporal") workerImage := getEnvOrDefault("KMSVC_TEMPORAL_WORKER_IMAGE", "story-crater-backend:latest") diff --git a/internal/operator/queue_controller_test.go b/internal/operator/queue_controller_test.go index 23cb30d..a63d75b 100644 --- a/internal/operator/queue_controller_test.go +++ b/internal/operator/queue_controller_test.go @@ -2,6 +2,7 @@ package operator import ( "context" + "fmt" "strings" "testing" "time" @@ -52,6 +53,12 @@ func newTestSchemeWithAppsV1(t *testing.T) *runtime.Scheme { } func newTestReconciler(t *testing.T, objs ...client.Object) (*QueueReconciler, *fakeAdmin) { + t.Helper() + r, admin, _ := newTestReconcilerWithTemporal(t, objs...) + return r, admin +} + +func newTestReconcilerWithTemporal(t *testing.T, objs ...client.Object) (*QueueReconciler, *fakeAdmin, *fakeTemporal) { t.Helper() scheme := newTestScheme(t) cl := fake.NewClientBuilder(). @@ -60,12 +67,14 @@ func newTestReconciler(t *testing.T, objs ...client.Object) (*QueueReconciler, * WithObjects(objs...). Build() admin := newFakeAdmin() + temporal := newFakeTemporal() return &QueueReconciler{ - Client: cl, - Admin: admin, - Redis: newTestRedis(t), - Now: time.Now, - }, admin + Client: cl, + Admin: admin, + Redis: newTestRedis(t), + Now: time.Now, + Temporal: temporal, + }, admin, temporal } func baseQueue(name string) *kmsvcv1.Queue { @@ -288,7 +297,7 @@ func TestReconcileDrainsClosingShardWhenLagZero(t *testing.T) { func TestReconcileTemporalWorkerCreatesWhenLabelPresent(t *testing.T) { queue := baseQueue("orders") queue.Labels = map[string]string{"temporal.io/namespace": "default"} - r, _ := newTestReconciler(t, queue) + r, _, temporal := newTestReconcilerWithTemporal(t, queue) ctx := context.Background() if err := r.Reconcile(ctx, "", "orders"); err != nil { @@ -302,6 +311,42 @@ func TestReconcileTemporalWorkerCreatesWhenLabelPresent(t *testing.T) { if worker.Spec.Namespace != "default" { t.Errorf("worker namespace = %q, want default", worker.Spec.Namespace) } + if got := temporal.count("default"); got != 1 { + t.Errorf("RegisterNamespace(%q) called %d times, want 1", "default", got) + } +} + +func TestReconcileTemporalWorkerRegistersNamespaceBeforeCreating(t *testing.T) { + queue := baseQueue("orders") + queue.Labels = map[string]string{"temporal.io/namespace": "checkout"} + r, _, temporal := newTestReconcilerWithTemporal(t, queue) + ctx := context.Background() + + if err := r.Reconcile(ctx, "", "orders"); err != nil { + t.Fatalf("Reconcile: %v", err) + } + if got := temporal.count("checkout"); got != 1 { + t.Errorf("RegisterNamespace(%q) called %d times, want 1", "checkout", got) + } +} + +func TestReconcileTemporalWorkerFailsWhenNamespaceRegistrationFails(t *testing.T) { + queue := baseQueue("orders") + queue.Labels = map[string]string{"temporal.io/namespace": "checkout"} + r, _, temporal := newTestReconcilerWithTemporal(t, queue) + temporal.setErr(fmt.Errorf("frontend unreachable")) + ctx := context.Background() + + err := r.Reconcile(ctx, "", "orders") + if err == nil { + t.Fatal("expected Reconcile to fail when namespace registration fails") + } + + var worker kmsvcv1.TemporalWorker + getErr := r.Client.Get(ctx, client.ObjectKey{Name: "worker-checkout", Namespace: "temporal"}, &worker) + if getErr == nil { + t.Error("expected no TemporalWorker to be created when namespace registration fails") + } } func TestReconcileTemporalWorkerValidatesNamespaceLabel(t *testing.T) { diff --git a/internal/operator/temporal_namespace.go b/internal/operator/temporal_namespace.go new file mode 100644 index 0000000..e48a387 --- /dev/null +++ b/internal/operator/temporal_namespace.go @@ -0,0 +1,14 @@ +package operator + +import "context" + +// TemporalNamespaceRegisterer registers a Temporal namespace. reconcileTemporalWorker +// calls this before creating a TemporalWorker so a Queue's temporal.io/namespace +// label always has a real namespace behind it — previously that label was +// trusted as-is, and a typo'd or never-registered namespace would only +// surface as a silently-stuck worker pod polling a namespace that doesn't +// exist. +type TemporalNamespaceRegisterer interface { + // RegisterNamespace registers namespace, treating AlreadyExists as success. + RegisterNamespace(ctx context.Context, namespace string) error +} diff --git a/internal/temporal/client.go b/internal/temporal/client.go new file mode 100644 index 0000000..34296ea --- /dev/null +++ b/internal/temporal/client.go @@ -0,0 +1,55 @@ +// Package temporal wraps the narrow slice of Temporal's WorkflowService that +// queue-operator needs (namespace registration) directly over gRPC, instead +// of pulling in the full Temporal Go SDK for one RPC. +package temporal + +import ( + "context" + "fmt" + "time" + + "go.temporal.io/api/workflowservice/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/durationpb" +) + +// defaultRetentionPeriod is applied to namespaces queue-operator registers on +// a Queue's behalf. Namespaces created deliberately by an operator (e.g. via +// the temporal CLI) can still override this by re-registering with different +// settings — RegisterNamespace on an existing namespace is a no-op here. +const defaultRetentionPeriod = 72 * time.Hour + +// Client wraps a Temporal frontend's WorkflowService. +type Client struct { + svc workflowservice.WorkflowServiceClient +} + +// NewClient dials a Temporal frontend at address (e.g. +// "temporal-frontend.temporal.svc.cluster.local:7233"). The connection is +// plaintext, matching how TemporalWorkerReconciler's worker pods already +// talk to the same frontend (see temporal_worker_controller.go). +func NewClient(address string) (*Client, error) { + conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return nil, fmt.Errorf("dial temporal frontend %s: %w", address, err) + } + return &Client{svc: workflowservice.NewWorkflowServiceClient(conn)}, nil +} + +// RegisterNamespace registers namespace, treating AlreadyExists as success. +func (c *Client) RegisterNamespace(ctx context.Context, namespace string) error { + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + + _, err := c.svc.RegisterNamespace(ctx, &workflowservice.RegisterNamespaceRequest{ + Namespace: namespace, + WorkflowExecutionRetentionPeriod: durationpb.New(defaultRetentionPeriod), + }) + if err != nil && status.Code(err) != codes.AlreadyExists { + return fmt.Errorf("register temporal namespace %s: %w", namespace, err) + } + return nil +}