rockandpoimen
76d8c518f0
CI / CI (push) Successful in 4m11s
## Changes - `internal/temporal/client.go` — Robust Temporal client with retry (exp backoff), TLS, health check - `internal/temporal/worker.go` — Worker creation, activity/workflow registration, lifecycle - `internal/temporal/context.go` — Timeout helpers - `k8s/worker-deployment.yaml` — 2-10 replica HPA, liveness/readiness probes, security context, pod anti-affinity - `k8s/workflow-runner-deployment.yaml` — Singleton runner with probes - `k8s/kustomization.yaml` — Updated resource list Co-authored-by: poimen <[email protected]>
85 lines
2.0 KiB
Go
85 lines
2.0 KiB
Go
package temporal
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"go.temporal.io/sdk/client"
|
|
"go.temporal.io/sdk/worker"
|
|
)
|
|
|
|
// WorkerConfig holds configuration for worker creation.
|
|
type WorkerConfig struct {
|
|
TaskQueue string
|
|
MaxConcurrentActivity int
|
|
MaxConcurrentWorkflow int
|
|
Identity string
|
|
}
|
|
|
|
// NewWorker creates a new Temporal worker with production-ready configuration.
|
|
//
|
|
// Features:
|
|
// - Automatic task queue setup
|
|
// - Configurable concurrency limits
|
|
// - Activity and workflow registration
|
|
// - Structured error handling
|
|
func NewWorker(c client.Client, cfg WorkerConfig) (worker.Worker, error) {
|
|
if cfg.TaskQueue == "" {
|
|
cfg.TaskQueue = "poimen-taskqueue"
|
|
}
|
|
if cfg.MaxConcurrentActivity == 0 {
|
|
cfg.MaxConcurrentActivity = 10
|
|
}
|
|
if cfg.MaxConcurrentWorkflow == 0 {
|
|
cfg.MaxConcurrentWorkflow = 10
|
|
}
|
|
if cfg.Identity == "" {
|
|
cfg.Identity = "poimen-worker-default"
|
|
}
|
|
|
|
workerOptions := worker.Options{
|
|
Identity: cfg.Identity,
|
|
MaxConcurrentActivityExecutionSize: cfg.MaxConcurrentActivity,
|
|
MaxConcurrentWorkflowTaskExecutionSize: cfg.MaxConcurrentWorkflow,
|
|
}
|
|
|
|
w := worker.New(c, cfg.TaskQueue, workerOptions)
|
|
if w == nil {
|
|
return nil, fmt.Errorf("failed to create worker for task queue: %s", cfg.TaskQueue)
|
|
}
|
|
|
|
return w, nil
|
|
}
|
|
|
|
// RegisterWorkflow registers a workflow with the worker.
|
|
func RegisterWorkflow(w worker.Worker, workflow interface{}) error {
|
|
if w == nil {
|
|
return fmt.Errorf("worker is nil")
|
|
}
|
|
w.RegisterWorkflow(workflow)
|
|
return nil
|
|
}
|
|
|
|
// RegisterActivity registers an activity with the worker.
|
|
func RegisterActivity(w worker.Worker, activity interface{}) error {
|
|
if w == nil {
|
|
return fmt.Errorf("worker is nil")
|
|
}
|
|
w.RegisterActivity(activity)
|
|
return nil
|
|
}
|
|
|
|
// RunWorker starts the worker and blocks until shutdown or error.
|
|
func RunWorker(w worker.Worker) error {
|
|
if w == nil {
|
|
return fmt.Errorf("worker is nil")
|
|
}
|
|
return w.Run(worker.InterruptCh())
|
|
}
|
|
|
|
// StopWorker gracefully stops the worker.
|
|
func StopWorker(w worker.Worker) {
|
|
if w != nil {
|
|
w.Stop()
|
|
}
|
|
}
|