228 lines
5.6 KiB
Markdown
228 lines
5.6 KiB
Markdown
# Temporal Workflow Orchestration
|
|||
|
|
|
||
|
|
**Server:** `temporal.temporal.svc.cluster.local:7233` (cluster-internal)
|
||
|
|
**Web UI:** `kubectl port-forward -n temporal svc/temporal-web 8088:8088`
|
||
|
|
**Namespace:** `temporal`
|
||
|
|
|
||
|
|
## When to Use
|
||
|
|
|
||
|
|
- **Long-running operations** — Tasks that take minutes/hours (send email, process batch, retry with backoff)
|
||
|
|
- **State machines** — Multi-step workflows with decision logic
|
||
|
|
- **Retries & timeouts** — Built-in exponential backoff, automatic retry
|
||
|
|
- **Audit trail** — Full history of workflow executions (why it happened, when, by whom)
|
||
|
|
|
||
|
|
## Quick Start
|
||
|
|
|
||
|
|
**1. Access Temporal Web UI:**
|
||
|
|
```bash
|
||
|
|
k port-forward -n temporal svc/temporal-web 8088:8088
|
||
|
|
# http://localhost:8088
|
||
|
|
```
|
||
|
|
|
||
|
|
**2. Define workflow (Go example):**
|
||
|
|
```go
|
||
|
|
package workflows
|
||
|
|
|
||
|
|
import (
|
||
|
|
"time"
|
||
|
|
"go.temporal.io/sdk/workflow"
|
||
|
|
"go.temporal.io/sdk/activity"
|
||
|
|
)
|
||
|
|
|
||
|
|
type Inputs struct {
|
||
|
|
OrderID string
|
||
|
|
Amount float64
|
||
|
|
}
|
||
|
|
|
||
|
|
// Workflow definition
|
||
|
|
func OrderProcessing(ctx workflow.Context, input Inputs) (string, error) {
|
||
|
|
// Step 1: Charge payment
|
||
|
|
chargeResult := ""
|
||
|
|
err := workflow.ExecuteActivity(
|
||
|
|
ctx,
|
||
|
|
ChargePayment,
|
||
|
|
input.OrderID,
|
||
|
|
input.Amount,
|
||
|
|
).Get(ctx, &chargeResult)
|
||
|
|
if err != nil {
|
||
|
|
return "", err
|
||
|
|
}
|
||
|
|
|
||
|
|
// Step 2: Send confirmation email (retry 3x on failure)
|
||
|
|
emailResult := ""
|
||
|
|
opts := workflow.ActivityOptions{
|
||
|
|
StartToCloseTimeout: time.Minute,
|
||
|
|
RetryPolicy: &temporal.RetryPolicy{
|
||
|
|
InitialInterval: time.Second,
|
||
|
|
BackoffCoefficient: 2,
|
||
|
|
MaximumAttempts: 3,
|
||
|
|
},
|
||
|
|
}
|
||
|
|
ctx = workflow.WithActivityOptions(ctx, opts)
|
||
|
|
|
||
|
|
err = workflow.ExecuteActivity(ctx, SendConfirmationEmail, input.OrderID).Get(ctx, &emailResult)
|
||
|
|
if err != nil {
|
||
|
|
return "", err
|
||
|
|
}
|
||
|
|
|
||
|
|
return "order_processed", nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// Activity: payment processing
|
||
|
|
func ChargePayment(ctx context.Context, orderID string, amount float64) (string, error) {
|
||
|
|
// Call payment gateway
|
||
|
|
return "payment_successful", nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// Activity: email notification
|
||
|
|
func SendConfirmationEmail(ctx context.Context, orderID string) (string, error) {
|
||
|
|
// Send email
|
||
|
|
return "email_sent", nil
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
**3. Register & start workflow:**
|
||
|
|
```go
|
||
|
|
import "go.temporal.io/sdk/client"
|
||
|
|
|
||
|
|
client, _ := client.Dial(client.Options{
|
||
|
|
HostPort: "temporal.temporal.svc.cluster.local:7233",
|
||
|
|
})
|
||
|
|
|
||
|
|
// Start workflow execution
|
||
|
|
run, _ := client.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
|
||
|
|
ID: "order-123",
|
||
|
|
TaskQueue: "orders",
|
||
|
|
}, OrderProcessing, Inputs{OrderID: "123", Amount: 99.99})
|
||
|
|
|
||
|
|
// Wait for result
|
||
|
|
var result string
|
||
|
|
run.Get(ctx, &result)
|
||
|
|
```
|
||
|
|
|
||
|
|
## Configuration
|
||
|
|
|
||
|
|
| Key | Value |
|
||
|
|
|-----|-------|
|
||
|
|
| Server | `temporal.temporal.svc.cluster.local:7233` |
|
||
|
|
| Web UI | `localhost:8088` (via port-forward) |
|
||
|
|
| Database | PostgreSQL (managed by helmfile) |
|
||
|
|
| Task queue | `default`, `orders`, `emails` (custom per app) |
|
||
|
|
| Retention | 30 days (configurable) |
|
||
|
|
|
||
|
|
## Common Patterns
|
||
|
|
|
||
|
|
**Retry with exponential backoff:**
|
||
|
|
```go
|
||
|
|
opts := workflow.ActivityOptions{
|
||
|
|
StartToCloseTimeout: 5 * time.Minute,
|
||
|
|
RetryPolicy: &temporal.RetryPolicy{
|
||
|
|
InitialInterval: time.Second,
|
||
|
|
BackoffCoefficient: 2.0, // double wait time each retry
|
||
|
|
MaximumInterval: time.Minute, // cap at 1 min between retries
|
||
|
|
MaximumAttempts: 5, // give up after 5 tries
|
||
|
|
},
|
||
|
|
}
|
||
|
|
ctx = workflow.WithActivityOptions(ctx, opts)
|
||
|
|
```
|
||
|
|
|
||
|
|
**Wait for signal (user approval):**
|
||
|
|
```go
|
||
|
|
// Workflow waits for approval signal
|
||
|
|
approval := ""
|
||
|
|
workflow.GetSignalChannel(ctx, "approval").Receive(ctx, &approval)
|
||
|
|
|
||
|
|
if approval == "approved" {
|
||
|
|
// Continue workflow
|
||
|
|
} else {
|
||
|
|
return "", errors.New("request denied")
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
**Parallel activities:**
|
||
|
|
```go
|
||
|
|
// Execute email & SMS in parallel
|
||
|
|
emailFuture := workflow.ExecuteActivity(ctx, SendEmail, userID)
|
||
|
|
smsFuture := workflow.ExecuteActivity(ctx, SendSMS, userID)
|
||
|
|
|
||
|
|
// Wait for both to complete
|
||
|
|
emailFuture.Get(ctx, nil)
|
||
|
|
smsFuture.Get(ctx, nil)
|
||
|
|
```
|
||
|
|
|
||
|
|
**Scheduled workflow (cron):**
|
||
|
|
```go
|
||
|
|
run, _ := client.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
|
||
|
|
ID: "daily-report",
|
||
|
|
CronSchedule: "0 9 * * MON-FRI", // 9 AM weekdays
|
||
|
|
WorkflowTaskTimeout: time.Hour,
|
||
|
|
}, GenerateDailyReport, nil)
|
||
|
|
```
|
||
|
|
|
||
|
|
## Monitoring
|
||
|
|
|
||
|
|
**Web UI:**
|
||
|
|
- List workflows: http://localhost:8088/namespaces/default/workflows
|
||
|
|
- View execution history: Click workflow ID
|
||
|
|
- See activity logs, errors, retry attempts
|
||
|
|
|
||
|
|
**Grafana dashboard:** `svc-temporal` (auto-configured)
|
||
|
|
|
||
|
|
**Key metrics:**
|
||
|
|
- `temporal_workflow_execution_duration_seconds` — workflow time
|
||
|
|
- `temporal_activity_execution_duration_seconds` — activity time
|
||
|
|
- `temporal_activity_execution_failed_total` — failed activities
|
||
|
|
|
||
|
|
## Integration with Story Crater
|
||
|
|
|
||
|
|
**Example: Process message via Temporal:**
|
||
|
|
```go
|
||
|
|
// In message handler
|
||
|
|
client, _ := temporal.Dial(/* ... */)
|
||
|
|
|
||
|
|
run, _ := client.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
|
||
|
|
ID: fmt.Sprintf("msg-%s", messageID),
|
||
|
|
TaskQueue: "story-crater",
|
||
|
|
}, ProcessMessageWorkflow, Message{
|
||
|
|
ID: messageID,
|
||
|
|
Body: body,
|
||
|
|
Source: "kafka-queue",
|
||
|
|
})
|
||
|
|
|
||
|
|
// Non-blocking: workflow runs independently
|
||
|
|
// Check status later
|
||
|
|
```
|
||
|
|
|
||
|
|
## Troubleshooting
|
||
|
|
|
||
|
|
**Workflow stuck:**
|
||
|
|
```bash
|
||
|
|
# Check Temporal server health
|
||
|
|
k get pods -n temporal
|
||
|
|
|
||
|
|
# View workflow execution history (via Web UI or CLI)
|
||
|
|
tctl workflow show --workflow-id order-123
|
||
|
|
|
||
|
|
# Terminate stuck workflow
|
||
|
|
tctl workflow terminate --workflow-id order-123
|
||
|
|
```
|
||
|
|
|
||
|
|
**Activity retrying endlessly:**
|
||
|
|
```go
|
||
|
|
// Add max attempts or timeout
|
||
|
|
RetryPolicy: &temporal.RetryPolicy{
|
||
|
|
MaximumAttempts: 5, // must have this!
|
||
|
|
}
|
||
|
|
```
|
||
|
|
|
||
|
|
**PostgreSQL connection fails:**
|
||
|
|
```bash
|
||
|
|
# Check temporal pod logs
|
||
|
|
k logs -n temporal pod/temporal-0
|
||
|
|
|
||
|
|
# Verify database is running
|
||
|
|
k get pods -n ddb
|
||
|
|
```
|
||
|
|
|
||
|
|
See `/TROUBLESHOOTING.md` for full incident guide.
|