diff --git a/TEMPORAL_INTEGRATION.md b/TEMPORAL_INTEGRATION.md new file mode 100644 index 0000000..c275ae8 --- /dev/null +++ b/TEMPORAL_INTEGRATION.md @@ -0,0 +1,262 @@ +# Temporal + kmsvc Integration Design + +## Vision +Use **kmsvc Queue CRDs** as the source of truth for Temporal worker provisioning. When a Queue is created, the Temporal worker operator automatically spawns a corresponding worker Deployment that listens to the queue's task queue. + +## Architecture + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ Developer: kubectl apply Queue CRD │ +│ (queue: orders-fifo, fifo: true, partitions: 6) │ +└──────────────────────┬──────────────────────────────────────────┘ + │ + v + ┌─────────────────────────────────────--─┐ + │ TemporalWorker Operator (kmsvc-manage)| + │ • Watches Queue CRDs │ + │ • Detects temporal.io/enabled label │ + │ • Creates/scales Deployments │ + └──────────────────────────────────────--┘ + │ + ┌──────────────┴──────────────┐ + │ │ + v v + ┌─────────────────┐ ┌─────────────────┐ + │ Temporal Worker │ │ Temporal Worker │ + │ Deployment │ ... │ StatefulSet │ + │ (Replicas: N) │ │ (DLQ processor) │ + └────────┬────────┘ └────────┬────────┘ + │ │ + └──────────────┬──────────-┘ + │ + ┌──────v───────-┐ + │ Temporal │ + │ Frontend │ + │ (Task Queues)│ + └───────────────┘ +``` + +## CRD: TemporalWorker + +**Namespace:** `temporal` (co-located with Temporal cluster) + +```yaml +apiVersion: temporal.kmsvc.io/v1 +kind: TemporalWorker +metadata: + name: worker-orders-fifo # derived from Queue name + suffix + namespace: temporal + ownerReferences: + - apiVersion: kmsvc.io/v1 + kind: Queue + name: orders-fifo # 1:1 reference to source Queue + uid: +spec: + # Source queue (read-only, set by operator) + queueRef: + name: orders-fifo + namespace: sqs # Queue lives in sqs namespace + + # Temporal task queue name (defaults to Queue name if omitted) + taskQueueName: orders-fifo # Temporal sees this task queue + + # Worker deployment config + image: story-crater-backend:latest # must have Temporal SDK initialized + imagePullPolicy: IfNotPresent + + # Replica count (can be overridden per-queue) + replicas: 2 # default: 1, autoscale later + + # Resource constraints + resources: + requests: + cpu: 500m + memory: 512Mi + limits: + cpu: 2000m + memory: 2Gi + + # Pod placement + nodeSelector: {} # default: any worker node + affinity: {} # optional: custom affinity rules + tolerations: [] + + # Lifecycle hooks (optional) + lifecycle: + postStartCommand: [] # e.g., ["register-activities.sh"] + preStopCommand: [] # e.g., ["drain-in-flight.sh"] + +status: + phase: Ready # Pending | Ready | Failed + replicas: 2 + readyReplicas: 2 + conditions: + - type: WorkerDeploymentReady + status: "True" + lastTransitionTime: "2026-07-10T12:34:56Z" + reason: DeploymentReady + message: "Worker Deployment worker-orders-fifo is running 2/2 replicas" + - type: TemporalTaskQueueRegistered + status: "True" + lastTransitionTime: "2026-07-10T12:34:56Z" + reason: QueueAvailable + message: "Task queue 'orders-fifo' is available in Temporal Frontend" +``` + +## Implementation Roadmap + +### Phase 1 (MVP): Manual TemporalWorker CRD (user creates explicitly) +1. Define `TemporalWorker` CRD in Go + Kubernetes schema +2. Implement controller that watches TemporalWorker objects +3. For each TemporalWorker: + - Create a Deployment with the specified image/replicas/resources + - Pod template includes: + - Environment variables: `TEMPORAL_FRONTEND_ADDRESS`, `TEMPORAL_TASK_QUEUE`, `TEMPORAL_NAMESPACE` + - Init container: wait for Temporal Frontend to be ready (DNS check: `temporal-frontend.temporal.svc.cluster.local:7233`) + - Set owner reference back to the source Queue (for cleanup on Queue deletion) +4. Create example TemporalWorker CRD (e.g., `k8s/temporal/workers/worker-orders-fifo.yaml`) +5. Deploy via helmfile postsync hook: `kubectl apply -f k8s/temporal/workers/` + +### Phase 2 (Future): Auto-provisioning from Queue CRDs +1. Extend kmsvc Queue CRD with optional label: `temporal.io/worker-enabled: "true"` +2. Extend queue-operator to watch Queue CRDs +3. On Queue creation with the label, auto-create a TemporalWorker CRD +4. Auto-derived fields: + - `taskQueueName` = Queue name + - `image` = default worker image (from configurable CM or env var) + - `replicas` = default (e.g., 1, or derived from Queue.spec.partitionsPerShard) + +### Phase 3 (Future): Autoscaling +1. Operator samples Queue depth via `kmsvc.io/metrics` endpoints +2. Adjust TemporalWorker replicas based on lag (similar to HPA but custom logic) +3. Min/max replicas configurable per worker + +## Integration Points + +### 1. Deployment Pod Template (What workers run) + +The application container must: +- Initialize Temporal SDK worker: `go.temporal.io/sdk/worker.New(...)` +- Register activity & workflow functions with the worker +- Listen on task queue = TemporalWorker.spec.taskQueueName (env var: `TEMPORAL_TASK_QUEUE`) +- Connect to Temporal Frontend: `TEMPORAL_FRONTEND_ADDRESS=temporal-frontend.temporal.svc.cluster.local:7233` + +**Example (story-crater-backend):** +```go +package main + +import ( + "fmt" + "os" + "go.temporal.io/client" + "go.temporal.io/sdk/worker" +) + +func main() { + // Read from TemporalWorker env vars + frontendAddr := os.Getenv("TEMPORAL_FRONTEND_ADDRESS") // temporal-frontend.temporal:7233 + taskQueue := os.Getenv("TEMPORAL_TASK_QUEUE") // orders-fifo + namespace := os.Getenv("TEMPORAL_NAMESPACE") // default + + c, _ := client.Dial(client.Options{ + HostPort: frontendAddr, + }) + defer c.Close() + + w := worker.New(c, namespace, taskQueue, worker.Options{}) + + // Register activities & workflows + w.RegisterActivity(activities.ProcessOrder) + w.RegisterWorkflow(workflows.OrderWorkflow) + + w.Run(worker.InterruptCh()) // block forever +} +``` + +### 2. Kubernetes Environment + +The TemporalWorker controller injects these env vars into the Deployment: +- `TEMPORAL_FRONTEND_ADDRESS` = `temporal-frontend.temporal.svc.cluster.local:7233` +- `TEMPORAL_NAMESPACE` = `default` (or from TemporalWorker.spec.namespace) +- `TEMPORAL_TASK_QUEUE` = TemporalWorker.spec.taskQueueName +- Inherited from Pod: `POD_NAME`, `POD_NAMESPACE`, `NODE_NAME` (via `downwardAPI`) + +### 3. Temporal Server Expectations + +No changes needed. Temporal Frontend auto-discovers task queues as workers connect. A worker that connects to task queue `orders-fifo` will: +- Show up in Temporal UI under `/namespaces/default/task-queues` +- Receive workflows routed to that task queue +- Return activity results to the workflow + +### 4. DNS & Network + +- Workers must resolve `temporal-frontend.temporal.svc.cluster.local:7233` (CoreDNS, Kubernetes standard) +- Temporal Frontend Service (already exists, type: ClusterIP) +- No ingress needed for worker→Temporal communication (cluster-internal) + +## Files to Create/Modify + +``` +kmsvc-manage/ + api/ + v1/ + temporalworker_types.go # CRD schema (Phase 1) + controllers/ + temporalworker_controller.go # Reconciliation logic (Phase 1) + config/ + crd/temporal.kmsvc.io_temporalworkers.yaml # CRD definition YAML + +homelab/ + k8s/ + temporal/ + workers/ + worker-orders-fifo.yaml # Example TemporalWorker instance + worker-orders-dlq.yaml # DLQ processor (optional) + temporal-worker-rbac.yaml # ServiceAccount + RBAC for worker Deployments +``` + +## Validation & Testing + +1. **Unit tests:** CRD validation, controller reconciliation loops +2. **Integration test (manual):** + ```bash + # 1. Deploy Temporal cluster (already done) + helmfile apply -l name=temporal + + # 2. Deploy kmsvc queue-operator with TemporalWorker controller + helmfile apply -l name=kmsvc-manage # (future Helm chart) + + # 3. Create a TemporalWorker CRD + kubectl apply -f k8s/temporal/workers/worker-orders-fifo.yaml + + # 4. Verify Deployment was created + kubectl get deploy -n temporal + kubectl get pods -n temporal -l app=temporal-worker-orders-fifo + + # 5. Verify worker is visible in Temporal UI + curl https://temporal.riotpiao.homelab.com/api/v1/task-queues + + # 6. Start a workflow targeting the task queue + temporal workflow start --task-queue=orders-fifo --type OrderWorkflow + + # 7. Verify worker executes the workflow + kubectl logs -n temporal -f deploy/temporal-worker-orders-fifo + ``` + +## Rollout Plan + +1. **MVP (kmsvc-manage Phase 1):** + - Define TemporalWorker CRD + controller + - Build CRUD reconciliation (create/update/delete Deployments) + - Document example TemporalWorker manifests + - User manually creates TemporalWorker CRDs for each queue + +2. **Phase 2 (kmsvc-manage Phase 2):** + - Extend Queue CRD with `temporal.io/worker-enabled` label + - Auto-generate TemporalWorker on Queue creation + - User just: `kubectl apply -f queue-orders-fifo.yaml` → worker auto-provisioned + +3. **Phase 3 (kmsvc-manage Phase 3):** + - Hook into Prometheus metrics (queue depth, lag) + - Scale replicas based on workload diff --git a/apis/kmsvc/v1/temporalworker_types.go b/apis/kmsvc/v1/temporalworker_types.go new file mode 100644 index 0000000..0a1a76e --- /dev/null +++ b/apis/kmsvc/v1/temporalworker_types.go @@ -0,0 +1,226 @@ +package v1 + +import ( + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" +) + +// TemporalWorkerPhase tracks the reconciliation state of a TemporalWorker. +type TemporalWorkerPhase string + +const ( + TemporalWorkerPhasePending TemporalWorkerPhase = "Pending" + TemporalWorkerPhaseReady TemporalWorkerPhase = "Ready" + TemporalWorkerPhaseFailed TemporalWorkerPhase = "Failed" +) + +// TemporalWorkerSpec defines the desired state of a TemporalWorker. +// One TemporalWorker per Temporal namespace handles all task queues in that namespace. +type TemporalWorkerSpec struct { + // Namespace is the Temporal namespace this worker connects to. + // The worker will process all task queues in this namespace. + Namespace string `json:"namespace"` + + // Image is the worker container image (must include Temporal SDK and + // activity/workflow registration logic for all task queues in the namespace). + Image string `json:"image"` + + // ImagePullPolicy controls when the image is pulled. + // +kubebuilder:validation:Enum=Always;Never;IfNotPresent + // +kubebuilder:default=IfNotPresent + // +optional + ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy,omitempty"` + + // Replicas is the desired number of worker Deployment replicas. + // Scale horizontally to handle aggregate queue depth across all task queues in the namespace. + // +kubebuilder:validation:Minimum=1 + // +kubebuilder:default=1 + // +optional + Replicas *int32 `json:"replicas,omitempty"` + + // Resources defines CPU/memory requests and limits for each worker pod. + // +optional + Resources *corev1.ResourceRequirements `json:"resources,omitempty"` + + // NodeSelector for pod placement (e.g., preferring worker nodes). + // +optional + NodeSelector map[string]string `json:"nodeSelector,omitempty"` + + // Affinity for advanced pod placement rules. + // +optional + Affinity *corev1.Affinity `json:"affinity,omitempty"` + + // Tolerations for pod placement on tainted nodes. + // +optional + Tolerations []corev1.Toleration `json:"tolerations,omitempty"` +} + +// TemporalWorkerStatus defines the observed state of a TemporalWorker. +type TemporalWorkerStatus struct { + // Phase is the current reconciliation phase. + // +kubebuilder:validation:Enum=Pending;Ready;Failed + Phase TemporalWorkerPhase `json:"phase,omitempty"` + + // Replicas is the current replica count from the underlying Deployment. + // +optional + Replicas int32 `json:"replicas,omitempty"` + + // ReadyReplicas is the number of ready replicas from the underlying Deployment. + // +optional + ReadyReplicas int32 `json:"readyReplicas,omitempty"` + + // Conditions hold detailed status information. + // +optional + Conditions []metav1.Condition `json:"conditions,omitempty"` +} + +// +kubebuilder:object:root=true +// +kubebuilder:subresource:status +// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase` +// +kubebuilder:printcolumn:name="Ready",type=string,JSONPath=`.status.readyReplicas` +// +kubebuilder:printcolumn:name="Desired",type=string,JSONPath=`.status.replicas` +// +kubebuilder:resource:shortName=worker;workers;tw + +// TemporalWorker is auto-created when a Queue CRD has the `temporal.io/namespace` label. +// The controller creates a Deployment that runs Temporal worker containers listening +// to the Temporal namespace specified in the label. +type TemporalWorker struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec TemporalWorkerSpec `json:"spec,omitempty"` + Status TemporalWorkerStatus `json:"status,omitempty"` +} + +// +kubebuilder:object:root=true + +// TemporalWorkerList contains a list of TemporalWorker. +type TemporalWorkerList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []TemporalWorker `json:"items"` +} + +func init() { + SchemeBuilder.Register(&TemporalWorker{}, &TemporalWorkerList{}) +} + +// DeepCopyInto copies the receiver, writing into out. +func (in *TemporalWorker) DeepCopyInto(out *TemporalWorker) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) + in.Status.DeepCopyInto(&out.Status) +} + +// DeepCopy returns a deep copy of the object. +func (in *TemporalWorker) DeepCopy() *TemporalWorker { + if in == nil { + return nil + } + out := new(TemporalWorker) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject returns a deep copy as runtime.Object. +func (in *TemporalWorker) DeepCopyObject() runtime.Object { + return in.DeepCopy() +} + +// DeepCopyInto copies the receiver, writing into out. +func (in *TemporalWorkerSpec) DeepCopyInto(out *TemporalWorkerSpec) { + *out = *in + if in.Replicas != nil { + in, out := &in.Replicas, &out.Replicas + *out = new(int32) + **out = **in + } + if in.Resources != nil { + in, out := &in.Resources, &out.Resources + *out = new(corev1.ResourceRequirements) + (*in).DeepCopyInto(*out) + } + if in.NodeSelector != nil { + in, out := &in.NodeSelector, &out.NodeSelector + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } + if in.Affinity != nil { + in, out := &in.Affinity, &out.Affinity + *out = new(corev1.Affinity) + (*in).DeepCopyInto(*out) + } + if in.Tolerations != nil { + in, out := &in.Tolerations, &out.Tolerations + *out = make([]corev1.Toleration, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy returns a deep copy of the object. +func (in *TemporalWorkerSpec) DeepCopy() *TemporalWorkerSpec { + if in == nil { + return nil + } + out := new(TemporalWorkerSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto copies the receiver, writing into out. +func (in *TemporalWorkerStatus) DeepCopyInto(out *TemporalWorkerStatus) { + *out = *in + if in.Conditions != nil { + in, out := &in.Conditions, &out.Conditions + *out = make([]metav1.Condition, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy returns a deep copy of the object. +func (in *TemporalWorkerStatus) DeepCopy() *TemporalWorkerStatus { + if in == nil { + return nil + } + out := new(TemporalWorkerStatus) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto copies the receiver, writing into out. +func (in *TemporalWorkerList) DeepCopyInto(out *TemporalWorkerList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]TemporalWorker, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy returns a deep copy of the object. +func (in *TemporalWorkerList) DeepCopy() *TemporalWorkerList { + if in == nil { + return nil + } + out := new(TemporalWorkerList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject returns a deep copy as runtime.Object. +func (in *TemporalWorkerList) DeepCopyObject() runtime.Object { + return in.DeepCopy() +} diff --git a/internal/operator/temporal_worker_controller.go b/internal/operator/temporal_worker_controller.go new file mode 100644 index 0000000..2be95d0 --- /dev/null +++ b/internal/operator/temporal_worker_controller.go @@ -0,0 +1,155 @@ +package operator + +import ( + "context" + "fmt" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + 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" +) + +const temporalWorkerFinalizerName = "kmsvc.io/temporal-worker" + +// TemporalWorkerReconciler reconciles TemporalWorker objects by creating and +// managing corresponding Kubernetes Deployments. +type TemporalWorkerReconciler struct { + Client client.Client +} + +// Reconcile creates/updates/deletes a Deployment based on the TemporalWorker object. +func (r *TemporalWorkerReconciler) Reconcile(ctx context.Context, namespace, name string) error { + logger := ctrllog.FromContext(ctx) + var worker kmsvcv1.TemporalWorker + if err := r.Client.Get(ctx, client.ObjectKey{Namespace: namespace, Name: name}, &worker); err != nil { + if apierrors.IsNotFound(err) { + return nil + } + return fmt.Errorf("get temporalworker %s/%s: %w", namespace, name, err) + } + + if !worker.DeletionTimestamp.IsZero() { + return r.reconcileDelete(ctx, &worker) + } + + if !controllerutil.ContainsFinalizer(&worker, temporalWorkerFinalizerName) { + controllerutil.AddFinalizer(&worker, temporalWorkerFinalizerName) + if err := r.Client.Update(ctx, &worker); err != nil { + return fmt.Errorf("add finalizer: %w", err) + } + } + + deployment := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: worker.Name, + Namespace: worker.Namespace, + }, + } + + replicas := int32(1) + if worker.Spec.Replicas != nil { + replicas = *worker.Spec.Replicas + } + + imagePullPolicy := corev1.PullIfNotPresent + if worker.Spec.ImagePullPolicy != "" { + imagePullPolicy = worker.Spec.ImagePullPolicy + } + + if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, deployment, func() error { + if err := controllerutil.SetControllerReference(&worker, deployment, r.Client.Scheme()); err != nil { + return fmt.Errorf("set controller reference: %w", err) + } + + deployment.Spec.Replicas = &replicas + deployment.Spec.Selector = &metav1.LabelSelector{ + MatchLabels: map[string]string{ + "app.kubernetes.io/name": "temporal-worker", + "app.kubernetes.io/instance": worker.Name, + "app.kubernetes.io/managed-by": "kmsvc-temporal-operator", + }, + } + deployment.Spec.Template.ObjectMeta.Labels = map[string]string{ + "app.kubernetes.io/name": "temporal-worker", + "app.kubernetes.io/instance": worker.Name, + "app.kubernetes.io/managed-by": "kmsvc-temporal-operator", + } + + container := corev1.Container{ + Name: "worker", + Image: worker.Spec.Image, + ImagePullPolicy: imagePullPolicy, + Env: []corev1.EnvVar{ + { + Name: "TEMPORAL_FRONTEND_ADDRESS", + Value: "temporal-frontend.temporal.svc.cluster.local:7233", + }, + { + Name: "TEMPORAL_NAMESPACE", + Value: worker.Spec.Namespace, + }, + { + Name: "TEMPORAL_TASK_QUEUE", + Value: worker.Name, + }, + }, + } + + if worker.Spec.Resources != nil { + container.Resources = *worker.Spec.Resources + } + + deployment.Spec.Template.Spec.Containers = []corev1.Container{container} + + if len(worker.Spec.NodeSelector) > 0 { + deployment.Spec.Template.Spec.NodeSelector = worker.Spec.NodeSelector + } + if worker.Spec.Affinity != nil { + deployment.Spec.Template.Spec.Affinity = worker.Spec.Affinity + } + if len(worker.Spec.Tolerations) > 0 { + deployment.Spec.Template.Spec.Tolerations = worker.Spec.Tolerations + } + + return nil + }); err != nil { + logger.Error(err, "failed to create or update deployment", "deployment", deployment.Name) + worker.Status.Phase = kmsvcv1.TemporalWorkerPhaseFailed + if err := r.Client.Status().Update(ctx, &worker); err != nil { + return fmt.Errorf("update status failed: %w", err) + } + return err + } + + worker.Status.Replicas = *deployment.Spec.Replicas + worker.Status.ReadyReplicas = deployment.Status.ReadyReplicas + if deployment.Status.ReadyReplicas == *deployment.Spec.Replicas { + worker.Status.Phase = kmsvcv1.TemporalWorkerPhaseReady + } else { + worker.Status.Phase = kmsvcv1.TemporalWorkerPhasePending + } + + if err := r.Client.Status().Update(ctx, &worker); err != nil { + return fmt.Errorf("update status: %w", err) + } + + return nil +} + +func (r *TemporalWorkerReconciler) reconcileDelete(ctx context.Context, worker *kmsvcv1.TemporalWorker) error { + if !controllerutil.ContainsFinalizer(worker, temporalWorkerFinalizerName) { + return nil + } + + controllerutil.RemoveFinalizer(worker, temporalWorkerFinalizerName) + if err := r.Client.Update(ctx, worker); err != nil { + return fmt.Errorf("remove finalizer: %w", err) + } + return nil +}