Production & Observability

Workflow ที่ทำงานถูกต้องบนเครื่อง dev กับระบบที่รับธุรกรรมจริงวันละ 1 ล้านรายการ เป็นคนละเรื่องกัน บทนี้ว่าด้วยสิ่งที่ต้องเตรียมก่อนเปิดให้ลูกค้าใช้จริง — ตั้ง worker ให้พอดี, มองเห็นสิ่งที่เกิดขึ้น, และไม่ให้ร้านค้ารายใหญ่กินคิวจนรายเล็กไม่ได้รัน

Worker tuning#

Worker คือ process ของเรา ที่ poll งานจาก task queue มาทำ ค่า default เหมาะกับการเริ่มต้นแต่ไม่เหมาะกับ production

w := worker.New(c, "wallet-core", worker.Options{
	// จำนวน activity ที่รันพร้อมกันได้สูงสุดใน process นี้ (default 1000)
	MaxConcurrentActivityExecutionSize: 200,

	// จำนวน workflow task ที่ประมวลผลพร้อมกัน (default 1000)
	MaxConcurrentWorkflowTaskExecutionSize: 200,

	// จำนวน poller (default 2 ทั้งคู่)
	MaxConcurrentActivityTaskPollers: 8,
	MaxConcurrentWorkflowTaskPollers: 8,

	// จำกัด activity ที่ยิงออกต่อวินาทีทั้ง worker — กันถล่ม API ธนาคาร
	WorkerActivitiesPerSecond: 50,

	// รอให้งานที่ทำอยู่จบก่อนปิด — สำคัญมากตอน rolling deploy
	WorkerStopTimeout: 30 * time.Second,
})

แยก task queue ตามลักษณะงาน#

อย่าเอา activity ที่คุยกับธนาคาร (ช้า, มี rate limit) ไปอยู่คิวเดียวกับงานที่เร็ว เพราะงานช้าจะกิน slot จนงานเร็วอด

// worker สำหรับ orchestration — เบา เร็ว
walletWorker := worker.New(c, "wallet-core", worker.Options{
	MaxConcurrentWorkflowTaskExecutionSize: 500,
})
walletWorker.RegisterWorkflow(workflows.TopUpWorkflow)

// worker สำหรับงานคุยธนาคาร — จำกัด rate ตามโควตาที่ธนาคารให้
bankWorker := worker.New(c, "bank-io", worker.Options{
	MaxConcurrentActivityExecutionSize: 20,
	WorkerActivitiesPerSecond:          30,
})
bankWorker.RegisterActivity(&activities.BankActivities{})

แล้วชี้ activity ไปคิวที่ถูกต้องจากใน workflow:

bankCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
	TaskQueue:           "bank-io",
	StartToCloseTimeout: 60 * time.Second,
})

Metrics ที่ต้องดู#

ต่อ metrics handler เข้ากับ client ตอนสร้าง:

c, err := client.Dial(client.Options{
	MetricsHandler: sdktally.NewMetricsHandler(newPrometheusScope(prometheus.Configuration{
		ListenAddress: "0.0.0.0:9090",
		TimerType:     "histogram",
	})),
})

Logging#

ใช้ logger ของ SDK เท่านั้นใน workflow — มันรู้ว่ากำลัง replay อยู่จึงไม่ log ซ้ำ

func TopUpWorkflow(ctx workflow.Context, req TopUpRequest) (TopUpResult, error) {
	// ติด field ที่ต้องใช้ทุกบรรทัดไว้ครั้งเดียว
	logger := log.With(workflow.GetLogger(ctx),
		"txnID", req.TransactionID,
		"walletID", req.WalletID,
	)

	logger.Info("top-up started", "amountSatang", req.AmountSatang)
	// ...
}

ต่อกับ slog เพื่อให้ log ของ SDK ไปรวมกับ log ของแอปเรา:

handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})
c, err := client.Dial(client.Options{
	Logger: log.NewStructuredLogger(slog.New(handler)),
})

fmt.Println และ log.Println ใน workflow จะพิมพ์ซ้ำทุกครั้งที่ replay ทำให้ log ปลอมและอ่านไม่รู้เรื่อง — ถ้าจำเป็นจริงๆ ให้เช็ค workflow.IsReplaying(ctx) ก่อน

Search Attributes#

Web UI ค้นด้วย Workflow ID ได้อยู่แล้ว แต่ ops มักถามคำถามแบบ "ขอรายการของร้าน M-042 ที่ค้างเกิน 1 ชั่วโมง" ซึ่งต้องใช้ Search Attributes

สร้าง attribute ใน namespace ก่อน (ทำครั้งเดียว):

temporal operator search-attribute create --name MerchantId --type Keyword
temporal operator search-attribute create --name TxnStage  --type Keyword
temporal operator search-attribute create --name AmountSatang --type Int

ประกาศ key แบบมี type แล้วใช้ได้ทั้งฝั่ง client และ workflow:

var (
	MerchantIDKey   = temporal.NewSearchAttributeKeyKeyword("MerchantId")
	TxnStageKey     = temporal.NewSearchAttributeKeyKeyword("TxnStage")
	AmountSatangKey = temporal.NewSearchAttributeKeyInt64("AmountSatang")
)

// ตอนเริ่ม workflow
_, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
	ID:        "topup-" + txnID,
	TaskQueue: "wallet-core",
	TypedSearchAttributes: temporal.NewSearchAttributes(
		MerchantIDKey.ValueSet("M-042"),
		AmountSatangKey.ValueSet(100_00),
	),
}, workflows.TopUpWorkflow, req)

// อัปเดตระหว่างทางจากใน workflow
if err := workflow.UpsertTypedSearchAttributes(ctx,
	TxnStageKey.ValueSet("awaiting_otp")); err != nil {
	return TopUpResult{}, err
}

แล้ว ops ก็ค้นได้:

temporal workflow list --query \
  'MerchantId = "M-042" AND TxnStage = "awaiting_otp" AND ExecutionStatus = "Running"'

Priority & Fairness — ปัญหา multi-tenant#

ระบบ wallet ที่ให้บริการหลายร้านค้าจะเจอปัญหานี้แน่นอน: ร้านใหญ่ยิงงานเข้ามา 100,000 รายการตอนปิดรอบ ส่วนร้านเล็กส่งมา 10 รายการ — ด้วยคิวแบบ FIFO ธรรมดา ร้านเล็กต้องรอจนกว่างานของร้านใหญ่จะหมด

System map

Fairness แก้ตรงนี้ด้วยการให้แต่ละ key มี virtual queue ของตัวเอง แล้ว dispatch สลับกันไป ร้านเล็กจึงได้รันแทรกเสมอ

_, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
	ID:        "topup-" + txnID,
	TaskQueue: "wallet-core",
	Priority: temporal.Priority{
		PriorityKey:    1,          // 1–5, เลขน้อย = สำคัญกว่า, default 3
		FairnessKey:    merchantID, // 1 key = 1 virtual queue
		FairnessWeight: 2.0,        // น้ำหนักเทียบกับ key อื่น, default 1.0
	},
}, workflows.TopUpWorkflow, req)

Priority เป็นคนละเรื่อง — มันจัดลำดับความสำคัญของประเภทงาน ในคิวเดียวกัน

ตัวอย่างการแบ่งชั้นบริการ:

จำกัด rate ต่อคิวและต่อ key ได้ด้วย:

temporal task-queue config set \
  --task-queue bank-io \
  --task-queue-type activity \
  --queue-rps-limit 500 \
  --queue-rps-limit-reason "โควตารวมที่ธนาคารให้" \
  --fairness-key-rps-limit-default 20 \
  --fairness-key-rps-limit-reason "กันร้านเดียวยิงจนเต็มโควตา"

ข้อจำกัดเรื่องขนาด#

ค่าเหล่านี้ปรับได้ระดับ cluster แต่อย่าไปแก้เพื่อหลบปัญหา เพราะรากของปัญหา คือการส่งข้อมูลก้อนใหญ่เข้าออก workflow ซึ่งทำให้ replay ช้าลงทุกครั้งด้วย

วิธีที่ถูกคือส่ง reference แทนข้อมูล:

// อย่าทำ — statement ทั้งเดือนเป็น MB เข้าไปอยู่ใน history ตลอดกาล
var statement []Transaction
_ = workflow.ExecuteActivity(actCtx, a.FetchStatement, bank, day).Get(actCtx, &statement)

// ทำแบบนี้ — activity เก็บลง object storage แล้วคืนแค่ key
var ref StatementRef // { Bucket, Key, RowCount, Checksum }
_ = workflow.ExecuteActivity(actCtx, a.FetchStatementToStorage, bank, day).Get(actCtx, &ref)
_ = workflow.ExecuteActivity(actCtx, a.ReconcileFromStorage, ref).Get(actCtx, nil)

ส่วน history ที่โตจากจำนวนรอบการทำงาน แก้ด้วย Continue-As-New ตามบทที่ 8

ตารางแก้ปัญหาหน้างาน#