Posted inProgramming

Xây dựng Workflow Orchestration bền bỉ với Temporal và Go: Xử lý Business Logic phân tán chịu lỗi

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

Bối cảnh & Lý do: Quản lý Business Logic phân tán ở quy mô lớn

Sáu tháng trước, đường ống xử lý đơn hàng (order processing pipeline) của chúng tôi bắt đầu quá tải trước khoảng 45.000 giao dịch mỗi ngày. Giống như nhiều hệ thống microservice khác, luồng checkout của chúng tôi dựa vào sự kết hợp chắp vá giữa các Kafka topic, cờ trạng thái trong Postgres và các cron job quét định kỳ mỗi 5 phút.

Bất cứ khi nào Stripe bị timeout hoặc inventory service khởi động lại trong quá trình checkout, các đơn hàng đều bị kẹt ở trạng thái bất nhất. Đội ngũ kỹ sư trực on-call phải mất từ 10 đến 15 giờ mỗi tuần để chạy các script bù trừ thủ công, gỡ rối dead-letter queue và đối soát các dòng dữ liệu mồ côi trong cơ sở dữ liệu.

Giao dịch phân tán (distributed transaction) vốn nổi tiếng là mỏng manh và dễ lỗi. Mạng rớt gói tin, container tự khởi động lại không báo trước, còn các API bên thứ ba ở phía upstream thì thỉnh thoảng lại trả về lỗi 502. Để xử lý thủ công vấn đề này, bạn thường phải tự dựng các state machine tùy biến trong SQL, truyền idempotency key qua hàng loạt service khác nhau và chèn thêm các vòng lặp retry vụn vặt vào từng HTTP client. Chẳng mấy chốc, đống boilerplate mang tính phòng thủ đó sẽ nuốt chửng phần lớn codebase của bạn.

Temporal giải quyết triệt để vấn đề này thông qua mô hình thực thi bền bỉ (durable execution). Bạn chỉ cần viết code Go tuần tự, tường minh thay vì phải tự theo dõi các bước chuyển trạng thái và biến đếm số lần retry. Bên dưới hệ thống, cluster của Temporal sẽ ghi lại các bước chuyển trạng thái vào một nhật ký sự kiện chỉ ghi nối tiếp (append-only event log). Nếu một tiến trình worker bị crash ngay giữa hàm hoặc máy ảo bên dưới đột ngột biến mất, một worker khác sẽ tiếp quản tiến trình thực thi ngay lập tức. Nó tái tạo lại trạng thái thực thi và tiếp tục chạy từ đúng dòng code vừa bị gián đoạn.

Sau khi chuyển đổi pipeline thanh toán sang Temporal, số lượng giao dịch ma giảm từ khoảng 120 vụ mỗi tháng xuống con số 0. Những thông báo khẩn on-call liên quan đến điều hướng đơn hàng gần như biến mất hoàn toàn. Tuyệt vời hơn nữa, chu kỳ review pull request của chúng tôi đã nhanh hơn 30% do các kỹ sư không còn phải thiết kế các luồng bù trừ riêng biệt cho từng sự cố hạ tầng nhỏ lẻ.

Cài đặt: Thiết lập Temporal Server và Go SDK

Khởi chạy một Temporal cluster trên môi trường local chỉ mất chưa đầy hai phút. Temporal CLI đi kèm sẵn một development server độc lập tích hợp kho dữ liệu SQLite nhúng cùng web console.

Cài đặt CLI thông qua script cài đặt chính thức hoặc Homebrew:

# Linux và macOS qua curl
curl -sSf https://temporal.download/cli.sh | sh

# macOS qua Homebrew
brew install temporal

Khởi động instance server local:

temporal server start-dev

Lệnh này sẽ khởi chạy gRPC engine tại 127.0.0.1:7233 và mở web UI tại http://localhost:8233. Hãy truy cập URL đó trên trình duyệt để kiểm tra các workflow đang chạy, kiểm toán dòng thời gian sự kiện (event timeline) và theo dõi thông lượng (throughput) của task queue.

Khi server đã hoạt động, hãy khởi tạo Go module và cài đặt SDK chính thức:

mkdir temporal-go-orders
cd temporal-go-orders
go mod init temporal-go-orders

go get go.temporal.io/sdk

Cấu hình: Tổ chức Workflow, Activity và Worker trong Go

Temporal tổ chức điều phối xoay quanh hai khối kiến trúc cốt lõi: Workflow và Activity.

  • Workflows: Nơi chứa toàn bộ logic điều phối thuần túy. Chúng bắt buộc phải mang tính tất định (deterministic)—tuyệt đối không thực hiện các lệnh gọi mạng trực tiếp, không truy vấn đồng hồ hệ thống trực tiếp hoặc tạo các giá trị ngẫu nhiên không có seed bên trong hàm workflow.
  • Activities: Thực thi các tác vụ có side effect như trừ tiền thẻ tín dụng, cập nhật SQL hoặc gọi API bên ngoài. Temporal sẽ tự động retry các activity theo chính sách backoff mà bạn cấu hình.

1. Định nghĩa các Activity

Trong activities.go, định nghĩa các thao tác thực hiện tương tác với bên ngoài:

package main

import (
	"context"
	"errors"
	"fmt"
	"math/rand"
)

type OrderRequest struct {
	OrderID string
	Amount  float64
	Email   string
}

type OrderActivities struct{}

func (a *OrderActivities) ProcessPayment(ctx context.Context, order OrderRequest) (string, error) {
	// Mô phỏng độ trễ cổng thanh toán và lỗi chập chờn ngắt quãng (tỷ lệ lỗi 30%)
	if rand.Float32() < 0.3 {
		return "", errors.New("timeout cổng thanh toán: kết nối bị reset bởi peer")
	}
	txnID := fmt.Sprintf("TXN-%s-OK", order.OrderID)
	return txnID, nil
}

func (a *OrderActivities) ReserveInventory(ctx context.Context, orderID string) error {
	// Gọi service tồn kho nội bộ
	return nil
}

func (a *OrderActivities) SendConfirmationEmail(ctx context.Context, email string, orderID string) error {
	// Kích hoạt email giao dịch qua Resend hoặc SendGrid
	return nil
}

2. Viết Deterministic Workflow

Tiếp theo, hãy định nghĩa luồng nghiệp vụ trong workflow.go. Hãy chú ý đến các thông số timeout tường minh và chính sách exponential retry được thiết lập trong workflow.ActivityOptions:

package main

import (
	"time"

	"go.temporal.io/sdk/temporal"
	"go.temporal.io/sdk/workflow"
)

func OrderWorkflow(ctx workflow.Context, order OrderRequest) (string, error) {
	retryPolicy := &temporal.RetryPolicy{
		InitialInterval:    time.Second * 1,
		BackoffCoefficient: 2.0,
		MaximumInterval:    time.Second * 30,
		MaximumAttempts:    5,
	}

	actOptions := workflow.ActivityOptions{
		StartToCloseTimeout: time.Minute * 2,
		RetryPolicy:         retryPolicy,
	}
	ctx = workflow.WithActivityOptions(ctx, actOptions)

	var activities *OrderActivities

	// Bước 1: Trừ tiền thẻ tín dụng của khách hàng với cơ chế retry tự động
	var transactionID string
	err := workflow.ExecuteActivity(ctx, activities.ProcessPayment, order).Get(ctx, &transactionID)
	if err != nil {
		return "", err
	}

	// Bước 2: Giữ hàng trong kho
	err = workflow.ExecuteActivity(ctx, activities.ReserveInventory, order.OrderID).Get(ctx, nil)
	if err != nil {
		return "", err
	}

	// Bước 3: Gửi email xác nhận
	err = workflow.ExecuteActivity(ctx, activities.SendConfirmationEmail, order.Email, order.OrderID).Get(ctx, nil)
	if err != nil {
		return "", err
	}

	return transactionID, nil
}

3. Xây dựng tiến trình Worker

Worker là các binary Go chạy nền liên tục để thăm dò task queue của Temporal qua gRPC. Hãy đặt phần triển khai worker sau vào file main.go:

package main

import (
	"context"
	"log"

	"go.temporal.io/sdk/client"
	"go.temporal.io/sdk/worker"
)

const TaskQueueName = "orders-queue"

func main() {
	c, err := client.Dial(client.Options{
		HostPort: client.DefaultHostPort,
	})
	if err != nil {
		log.Fatalln("Không thể tạo Temporal client:", err)
	}
	defer c.Close()

	w := worker.New(c, TaskQueueName, worker.Options{})

	act := &OrderActivities{}
	w.RegisterWorkflow(OrderWorkflow)
	w.RegisterActivity(act)

	go func() {
		// Kích hoạt một workflow execution mẫu sau khi worker khởi động
		workflowOptions := client.StartWorkflowOptions{
			ID:        "order-sample-1001",
			TaskQueue: TaskQueueName,
		}
		req := OrderRequest{
			OrderID: "1001",
			Amount:  89.99,
			Email:   "[email protected]",
		}
		we, startErr := c.ExecuteWorkflow(context.Background(), workflowOptions, OrderWorkflow, req)
		if startErr != nil {
			log.Println("Không thể khởi chạy workflow:", startErr)
			return
		}
		log.Printf("Đã khởi chạy WorkflowID: %s, RunID: %s\n", we.GetID(), we.GetRunID())
	}()

	log.Println("Worker đang lắng nghe trên task queue:", TaskQueueName)
	err = w.Run(worker.InterruptCh())
	if err != nil {
		log.Fatalln("Worker gặp lỗi nghiêm trọng:", err)
	}
}

Kiểm thử & Giám sát: Kiểm tra khả năng xử lý lỗi và Observability

Hãy cùng kiểm chứng cách hệ thống phản ứng khi tiến trình gặp sự cố đột ngột. Khởi chạy worker:

go run .

Mở http://localhost:8233 trên trình duyệt và chọn execution có tên order-sample-1001.

Kiểm chứng khả năng chịu lỗi (Fault-Tolerance)

Mô phỏng tình huống tiến trình bị crash đột ngột ngay từ terminal:

  1. Chạy go run . để bắt đầu workflow.
  2. Nhấn Ctrl+C trong khi ProcessPayment đang chạy để làm crash tiến trình worker ngay lập tức.
  3. Kiểm tra web console của Temporal. Workflow vẫn ở trạng thái mở và hoạt động tốt thay vì bị fail. Temporal chỉ đơn giản là chờ một worker đang hoạt động vào poll task queue.
  4. Khởi động lại worker bằng lệnh go run .. Temporal sẽ phân phối lại task trong queue. Workflow sẽ tiếp tục từ đúng điểm bị gián đoạn mà không thực hiện lặp lại các tác vụ trước đó.

Giám sát môi trường Production và các chỉ số trọng yếu

Vận hành kiến trúc này trong môi trường có throughput cao đòi hỏi bạn phải theo dõi hai chỉ số thiết yếu:

  • Schedule-To-Start Latency: Theo dõi khoảng chênh lệch thời gian từ lúc Temporal đưa một task vào hàng đợi đến khi worker tiếp nhận task đó. Trong một cluster khỏe mạnh, chỉ số này luôn duy trì dưới 50ms. Nếu chỉ số này bắt đầu tăng vượt quá 1 đến 2 giây, hãy scale thêm số lượng worker instance.
  • Activity Failure Rates: Kiểm tra cây lịch sử (history tree) trên Temporal UI khi các nhà cung cấp dịch vụ phía upstream gặp sự cố. Giao diện cung cấp đầy đủ stack trace, input payload và nguyên nhân thất bại cho mỗi lần retry, giúp bạn chẩn đoán vấn đề ngay lập tức mà không cần phải bới tìm trong đống log phi cấu trúc.

Sự kết hợp giữa cơ chế quản lý trạng thái event-sourced của Temporal cùng mô hình xử lý đồng thời (concurrency) gọn nhẹ của Go mang lại cho bạn những workflow rõ ràng, bền bỉ và có khả năng chống chịu tình trạng mất kết nối mạng hay crash server một cách tự nhiên mà không cần cấu hình phức tạp.

Share:
Tags: