feat: add TemporalWorker CRD and controller
Add TemporalWorker CRD definition with full status tracking and TemporalWorkerReconciler that manages Deployment lifecycle: - Auto-generate DeepCopy methods via kubebuilder markers - Controller creates/updates Deployments matching worker spec - Injects TEMPORAL_FRONTEND_ADDRESS and TEMPORAL_NAMESPACE env vars - Tracks replica count and ready status - Handles graceful deletion via finalizer Design document (TEMPORAL_INTEGRATION.md) describes three-phase roadmap: - Phase 1 (MVP): Manual TemporalWorker CRD creation - Phase 2: Auto-provisioning from Queue labels - Phase 3: Autoscaling based on queue depth Co-Authored-By: Claude Haiku 4.5 <[email protected]>
This commit is contained in:
Story Crater Bot
co-authored by
Claude Haiku 4.5
Claude Haiku 4.5
parent
674fbb6bfe
commit
6577efc100
@@ -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: <uuid>
|
||||
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
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user