diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 99cec2f8..6942753e 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -1,3 +1,7 @@ +# 1.2.8 (2026-07-27) + +- `asynctaskext`/`busext` 新增 Prometheus 处理耗时/QPS 埋点（`asynctask_task_duration_seconds`/`bus_task_duration_seconds`），config `monitor_enable` 开关可选开启，默认关闭零开销；与 Python coast 库同名同 label 同 buckets，可跨语言合并查询 + # 1.2.7 (2026-06-16) - 移除 shanbay/go-redis fork 的 replace 指令，改用官方 `github.com/redis/go-redis/extra/redisotel/v9` v9.18.0 及 `rediscmd/v9` v9.18.0 diff --git a/docs/ext_amqp_cn.md b/docs/ext_amqp_cn.md index 15ac881f..7e7b7707 100644 --- a/docs/ext_amqp_cn.md +++ b/docs/ext_amqp_cn.md @@ -208,3 +208,29 @@ func TestHelloHandler_Run(t *testing.T) { } } ``` + +## 监控指标 + +在 config.yaml 里加一行即可开启处理耗时/QPS 埋点（默认关闭，零开销）： + +```yaml + bus_monitor_enable: true +``` + +开启后会记录 Prometheus Histogram 指标 `bus_task_duration_seconds`（label：`task_name`、`queue`、`status`；两者都取消息的 routing key，因为 busext 目前没有独立于 routing key 的"任务名"概念）。`status` 是真实的处理结果：`success` / `failure`（`handler.Run()` 返回 error）/ `parse_error`（payload 解析失败）/ `invalid_message`（headers、content-type、encoding、未注册 routing key 等消息本身有问题）。gobay 本身**不会**自建 `/metrics` HTTP server，需要使用方应用自己起一个（如 `promhttp.Handler()`），照搬 `cachext` 的既有惯例。 + +这个指标和 Python coast 库（`coast/celery.py`）里的 `bus_task_duration_seconds` 是**同一个指标**（同名、同 label、同 buckets），可以在同一个 Prometheus/Grafana 查询里合并 Go 和 Python 两边的数据，不需要额外处理。 + +常用 PromQL： + +```promql +# QPS +sum(rate(bus_task_duration_seconds_count[1m])) by (task_name) + +# 失败率 +sum(rate(bus_task_duration_seconds_count{status="failure"}[5m])) by (task_name) + / sum(rate(bus_task_duration_seconds_count[5m])) by (task_name) + +# P95 耗时 +histogram_quantile(0.95, sum(rate(bus_task_duration_seconds_bucket[5m])) by (le, task_name)) +``` diff --git a/docs/ext_asynctask_cn.md b/docs/ext_asynctask_cn.md index 7faf7300..03dd9e58 100644 --- a/docs/ext_asynctask_cn.md +++ b/docs/ext_asynctask_cn.md @@ -116,3 +116,29 @@ func SomeAsyncTask(ctx context.Context, relatedItemIDToProcess uint64) error { ``` ## 测试代码只需测试 SomeAsyncTask 函数本身即可 + +## 监控指标 + +在 config.yaml 里加一行即可开启处理耗时/QPS 埋点（默认关闭，零开销）： + +```yaml + asynctask_monitor_enable: true +``` + +开启后会记录 Prometheus Histogram 指标 `asynctask_task_duration_seconds`（label：`task_name`、`queue`、`status`；`status` 固定为 `"unknown"`——machinery 的全局 hook 拿不到每次任务调用的成功/失败信息，只能保证记录到耗时和 QPS）。gobay 本身**不会**自建 `/metrics` HTTP server，需要使用方应用自己起一个（如 `promhttp.Handler()`），照搬 `cachext` 的既有惯例。 + +这个指标和 Python coast 库（`coast/celery.py`）里的 `asynctask_task_duration_seconds` 是**同一个指标**（同名、同 label、同 buckets），可以在同一个 Prometheus/Grafana 查询里合并 Go 和 Python 两边的数据，不需要额外处理。 + +常用 PromQL： + +```promql +# QPS +sum(rate(asynctask_task_duration_seconds_count[1m])) by (task_name) + +# 平均耗时 +sum(rate(asynctask_task_duration_seconds_sum[5m])) by (task_name) + / sum(rate(asynctask_task_duration_seconds_count[5m])) by (task_name) + +# P95 耗时 +histogram_quantile(0.95, sum(rate(asynctask_task_duration_seconds_bucket[5m])) by (le, task_name)) +``` diff --git a/extensions/asynctaskext/asynctaskext.go b/extensions/asynctaskext/asynctaskext.go index 74762a0d..fe2d7f59 100644 --- a/extensions/asynctaskext/asynctaskext.go +++ b/extensions/asynctaskext/asynctaskext.go @@ -23,6 +23,7 @@ import ( "github.com/google/uuid" "github.com/mitchellh/mapstructure" "github.com/shanbay/gobay" + "github.com/shanbay/gobay/observability" ) const ( @@ -39,6 +40,10 @@ type AsyncTaskExt struct { lock sync.Mutex healthCheckCompleteChan chan string healthHandlerRegistered bool + + monitorEnabled bool + taskStartTimes map[string]time.Time + taskStartTimesM sync.Mutex } func (t *AsyncTaskExt) Object() interface{} { @@ -70,6 +75,12 @@ func (t *AsyncTaskExt) Init(app *gobay.Application) error { return err } t.server = server + + if config.GetBool("monitor_enable") { + t.monitorEnabled = true + t.taskStartTimes = make(map[string]time.Time) + } + return t.registerHealthCheck() } @@ -102,6 +113,13 @@ func (t *AsyncTaskExt) StartWorker(queue string, concurrency int, enableHealthCh worker.Queue = queue t.workers = append(t.workers, worker) + if t.monitorEnabled { + worker.SetPreTaskHandler(t.recordTaskStart) + worker.SetPostTaskHandler(func(sig *tasks.Signature) { + t.recordTaskDuration(sig, queue) + }) + } + // run health check http server if enableHealthCheck && !t.healthHandlerRegistered { t.healthHandlerRegistered = true @@ -146,6 +164,32 @@ func (t *AsyncTaskExt) genConsumerTag(queue string) string { return fmt.Sprintf("%s@%s", queue, hostName) } +// recordTaskStart 记录任务开始处理的时间点，按 signature.UUID 索引。 +func (t *AsyncTaskExt) recordTaskStart(sig *tasks.Signature) { + t.taskStartTimesM.Lock() + defer t.taskStartTimesM.Unlock() + t.taskStartTimes[sig.UUID] = time.Now() +} + +// recordTaskDuration 从 taskStartTimes 弹出起点算 duration，Observe 到 +// observability.AsyncTaskDurationSeconds。status 固定写死 "unknown"——machinery +// 的 SetPostTaskHandler 全局 hook 拿不到该次调用的 error，无法在并发 worker 下 +// 精确归因到具体任务，为避免引入 reflect.MakeFunc 包装 handler 的复杂度和风险， +// 明确放弃精确 status（详见 plan §1.3）。 +func (t *AsyncTaskExt) recordTaskDuration(sig *tasks.Signature, queue string) { + t.taskStartTimesM.Lock() + start, ok := t.taskStartTimes[sig.UUID] + if ok { + delete(t.taskStartTimes, sig.UUID) + } + t.taskStartTimesM.Unlock() + if !ok { + return + } + observability.AsyncTaskDurationSeconds.WithLabelValues(sig.Name, queue, "unknown"). + Observe(time.Since(start).Seconds()) +} + func (t *AsyncTaskExt) registerHealthCheck() error { return t.server.RegisterTask(healthCheckTaskName, func(healthCheckUUID string) error { select { diff --git a/extensions/asynctaskext/asynctaskext_test.go b/extensions/asynctaskext/asynctaskext_test.go index 463c0f84..dc438a7d 100644 --- a/extensions/asynctaskext/asynctaskext_test.go +++ b/extensions/asynctaskext/asynctaskext_test.go @@ -1,14 +1,34 @@ +/* +# Partition table (ECP + BVA) — step 01-asynctask-bus-metrics (asynctaskext side) +# +# 参数/状态 | 等价类 | 类型 | 代表值/场景 | 期望输出 | 对应契约条目 +# monitorEnabled | 关闭（默认/未设置） | 有效 | one_asynctask_ 在 "testing" env 下未设置 monitor_enable | 任务正常执行不受影响，不 panic，asynctask_task_duration_seconds 不含该 task_name 的记录 | 数据层 API 锚点 1 +# monitorEnabled | 开启 | 有效 | three_asynctask_monitor_enable: true | Init 不 panic，monitorEnabled == true | 数据层 API 锚点 2/3/9 +# 任务执行结果 (task outcome)| 成功 | 有效 | TaskAddThree 正常返回 | 指标 _count 增加、_sum > 0，status 仍为 "unknown" | 数据层 API 锚点 2、行为契约"status 固定 unknown" +# 任务执行结果 (task outcome)| 返回 error | 有效（错误路径） | TaskFailThree 返回 error | 指标依然被 Observe（不因失败跳过），status 仍为 "unknown" | 数据层 API 锚点 3 +# 同类型多实例 (multi-instance)| 两个实例都开启 monitor_enable | 边界/并发场景 | three_asynctask_ + four_asynctask_ 同时 Init | 不 panic（验证包级单例、无重复 promauto 注册） | 数据层 API 锚点 9、MUST "包级单例" +# +# 说明：本 step 的 asynctaskext 侧只有 2 个独立参数（monitorEnabled ×2、task outcome ×2）， +# 未达到 pairwise 触发阈值（≥3 参数 × 每个 ≥2 取值），因此未调用 pairwise 脚本， +# 采用穷举的 2×2 组合（关闭+成功 已由锚点1覆盖；开启+成功、开启+失败 由锚点2/3覆盖）。 +*/ package asynctaskext import ( "context" + "errors" + "io" "log" "net/http" + "strconv" + "strings" + "sync" "testing" "time" "github.com/RichardKnop/machinery/v1/backends/result" "github.com/RichardKnop/machinery/v1/tasks" + "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/stretchr/testify/assert" "github.com/shanbay/gobay" @@ -55,6 +75,81 @@ func TaskSubWithContext(ctx context.Context, arg1, arg2 int64) (int64, error) { return arg1 - arg2, nil } +// TaskAddThree / TaskFailThree are used exclusively by the monitor tests below, +// with names that never collide with "add"/"sub"/"subCtx" registered on taskOne, +// so metric assertions can't accidentally match unrelated task executions. +func TaskAddThree(args ...int64) (int64, error) { + sum := int64(0) + for _, arg := range args { + sum += arg + } + return sum, nil +} + +func TaskFailThree(arg int64) (int64, error) { + return 0, errors.New("intentional failure for asynctaskext monitor test") +} + +var ( + taskThree AsyncTaskExt + taskFour AsyncTaskExt + + asyncTaskMetricsOnce sync.Once +) + +const asyncTaskMetricsAddr = "127.0.0.1:2113" + +// startAsyncTaskMetricsServer exposes prometheus.DefaultRegisterer via +// promhttp.Handler(), mirroring extensions/cachext's TestCacheExt_Cached_Monitor +// pattern (config-gated instrumentation + no self-built /metrics server in +// production code, only in the test harness). +func startAsyncTaskMetricsServer() { + asyncTaskMetricsOnce.Do(func() { + go func() { + http.Handle("/metrics", promhttp.Handler()) + if err := http.ListenAndServe(asyncTaskMetricsAddr, nil); err != nil { + log.Fatalf("error when start prometheus server: %v\n", err) + } + }() + time.Sleep(200 * time.Millisecond) + }) +} + +func fetchAsyncTaskMetrics(t *testing.T) string { + resp, err := http.Get("http://" + asyncTaskMetricsAddr + "/metrics") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + data, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + return string(data) +} + +// extractMetricValue looks up "{} " in the exposition +// text and parses the numeric value, so tests can assert e.g. "_sum > 0" +// instead of only checking for substring presence. +func extractMetricValue(t *testing.T, metrics, family, labels string) float64 { + prefix := family + "{" + labels + "} " + idx := strings.Index(metrics, prefix) + if idx == -1 { + t.Fatalf("metric not found: %s", prefix) + } + rest := metrics[idx+len(prefix):] + end := strings.IndexByte(rest, '\n') + if end == -1 { + end = len(rest) + } + valueStr := strings.TrimSpace(rest[:end]) + value, err := strconv.ParseFloat(valueStr, 64) + if err != nil { + t.Fatalf("failed to parse metric value %q: %v", valueStr, err) + } + return value +} + func TestPushConsume(t *testing.T) { if err := taskOne.RegisterWorkerHandlers(map[string]interface{}{ "add": TaskAdd, "sub": TaskSub, "subCtx": TaskSubWithContext, @@ -184,3 +279,125 @@ func TestMultiTaskExtStartWorker(t *testing.T) { }) }) } + +// TestAsyncTaskExt_Monitor_Disabled 覆盖锚点 1：monitor_enable 默认关闭时， +// monitorEnabled 保持 false，任务正常执行不受影响、不 panic，且不产生任何 +// asynctask_task_duration_seconds 记录。 +func TestAsyncTaskExt_Monitor_Disabled(t *testing.T) { + startAsyncTaskMetricsServer() + + assert.False(t, taskOne.monitorEnabled) + + sign := &tasks.Signature{ + Name: "add", + Args: []tasks.Arg{ + {Type: "int64", Value: 5}, + {Type: "int64", Value: 6}, + }, + } + asyncResult, err := taskOne.SendTask(sign) + if err != nil { + t.Fatal(err) + } + if _, err := asyncResult.Get(5 * time.Millisecond); err != nil { + t.Fatal(err) + } + time.Sleep(200 * time.Millisecond) + + data := fetchAsyncTaskMetrics(t) + assert.NotContains(t, data, + `asynctask_task_duration_seconds_count{queue="gobay.task.one",status="unknown",task_name="add"}`) +} + +// TestAsyncTaskExt_Monitor covers 锚点 2/3/9：monitor_enable=true 下， +// 两个同类型实例（three_asynctask_/four_asynctask_）各自 Init 不 panic， +// 成功任务和失败任务都会被 Observe，status 始终固定 "unknown"。 +func TestAsyncTaskExt_Monitor(t *testing.T) { + startAsyncTaskMetricsServer() + + taskThree = AsyncTaskExt{NS: "three_asynctask_"} + taskFour = AsyncTaskExt{NS: "four_asynctask_"} + + t.Run("9: 同类型两个实例都开启 monitor_enable 各自 Init 不 panic", func(t *testing.T) { + assert.NotPanics(t, func() { + app, err := gobay.CreateApp( + "../../testdata", + "asynctaskmonitored", + map[gobay.Key]gobay.Extension{ + "taskthree": &taskThree, + "taskfour": &taskFour, + }, + ) + if err != nil { + t.Fatal(err) + } + if err := app.Init(); err != nil { + t.Fatal(err) + } + }) + assert.True(t, taskThree.monitorEnabled) + assert.True(t, taskFour.monitorEnabled) + }) + + if err := taskThree.RegisterWorkerHandlers(map[string]interface{}{ + "addThree": TaskAddThree, + "failThree": TaskFailThree, + }); err != nil { + t.Fatal(err) + } + go func() { + // use default queue "gobay.task.three" + if err := taskThree.StartWorker("", 1, false); err != nil { + t.Error(err) + } + }() + time.Sleep(500 * time.Millisecond) // make sure the worker is started + + t.Run("2: monitor_enable=true 下执行已注册任务成功 -> _count 增加、_sum > 0", func(t *testing.T) { + sign := &tasks.Signature{ + Name: "addThree", + Args: []tasks.Arg{ + {Type: "int64", Value: 3}, + {Type: "int64", Value: 4}, + }, + } + asyncResult, err := taskThree.SendTask(sign) + if err != nil { + t.Fatal(err) + } + if results, err := asyncResult.Get(5 * time.Millisecond); err != nil { + t.Fatal(err) + } else if res, ok := results[0].Interface().(int64); !ok || res != 7 { + t.Fatalf("unexpected task result: %v", results) + } + time.Sleep(300 * time.Millisecond) + + data := fetchAsyncTaskMetrics(t) + labels := `queue="gobay.task.three",status="unknown",task_name="addThree"` + assert.Contains(t, data, `asynctask_task_duration_seconds_count{`+labels+`} 1`) + sum := extractMetricValue(t, data, "asynctask_task_duration_seconds_sum", labels) + assert.Greater(t, sum, 0.0) + }) + + t.Run("3: monitor_enable=true 下执行返回 error 的任务仍被 Observe，status 仍为 unknown", func(t *testing.T) { + sign := &tasks.Signature{ + Name: "failThree", + Args: []tasks.Arg{ + {Type: "int64", Value: 1}, + }, + } + asyncResult, err := taskThree.SendTask(sign) + if err != nil { + t.Fatal(err) + } + // TaskFailThree itself returns an error; we only care whether the + // metric was recorded regardless of the task outcome, matching the + // "不因失败而跳过" 行为契约. + _, _ = asyncResult.Get(5 * time.Millisecond) + time.Sleep(300 * time.Millisecond) + + data := fetchAsyncTaskMetrics(t) + labels := `queue="gobay.task.three",status="unknown",task_name="failThree"` + assert.Contains(t, data, `asynctask_task_duration_seconds_count{`+labels+`} 1`) + }) +} diff --git a/extensions/busext/amqp.go b/extensions/busext/amqp.go index 6e14e67a..ab8f73f0 100644 --- a/extensions/busext/amqp.go +++ b/extensions/busext/amqp.go @@ -15,6 +15,7 @@ import ( "github.com/streadway/amqp" "github.com/shanbay/gobay" + "github.com/shanbay/gobay/observability" ) var ( @@ -68,6 +69,7 @@ type BusExt struct { publishFunc func(exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error brokerUrl string notifyChanBlock chan error + monitorEnabled bool } func (b *BusExt) Object() interface{} { @@ -98,6 +100,7 @@ func (b *BusExt) Init(app *gobay.Application) error { b.pushTimeout = b.config.GetDuration("push_timeout") b.pushFunc = b.doPush b.notifyChanBlock = make(chan error) + b.monitorEnabled = b.config.GetBool("monitor_enable") var tlsConfig *tls.Config if b.config.GetBool("tls") { @@ -277,26 +280,14 @@ func (b *BusExt) Consume() error { b.deliveryAck(delivery) log.Printf("Receive delivery: %s from queue: %v\n", delivery.Headers["id"], chName) - var handler Handler - var ok bool - if delivery.Headers == nil { - b.ErrorLogger.Println("Not support v1 celery protocol yet") - } else if delivery.ContentType != "application/json" { - b.ErrorLogger.Println("Only json encoding is allowed") - } else if delivery.ContentEncoding != "utf-8" { - b.ErrorLogger.Println("Unsupported content encoding") - } else if handler, ok = b.consumers[delivery.RoutingKey]; !ok { - b.ErrorLogger.Println("Receive unregistered message") + if b.monitorEnabled { + start := time.Now() + status := b.dispatch(delivery) + observability.BusTaskDurationSeconds.WithLabelValues( + delivery.RoutingKey, delivery.RoutingKey, status, + ).Observe(time.Since(start).Seconds()) } else { - var payload []json.RawMessage - if err := json.Unmarshal(delivery.Body, &payload); err != nil { - b.ErrorLogger.Printf("json decode error: %v\n", err) - } else if err := handler.ParsePayload(payload[0], - payload[1]); err != nil { - b.ErrorLogger.Printf("handler parse payload error: %v\n", err) - } else if err := handler.Run(); err != nil { - b.ErrorLogger.Printf("handler run task failed: %v\n", err) - } + b.dispatch(delivery) } } } @@ -306,6 +297,45 @@ func (b *BusExt) Consume() error { return nil } +// dispatch 处理一条已经 ack 过的 delivery，逻辑与改动前的 Consume() 内联 +// if/else-if 链完全一致（日志文案、分支顺序不变），额外返回一个状态字符串 +// 供 monitor_enable 开启时打 Prometheus 指标用： +// +// "invalid_message" — headers 为空 / content-type 非 json / encoding 非 utf-8 / 未注册的 routing key +// "parse_error" — json.Unmarshal 失败 或 handler.ParsePayload 失败 +// "failure" — handler.Run() 返回 error +// "success" — 全部通过 +func (b *BusExt) dispatch(delivery amqp.Delivery) string { + var handler Handler + var ok bool + if delivery.Headers == nil { + b.ErrorLogger.Println("Not support v1 celery protocol yet") + return "invalid_message" + } else if delivery.ContentType != "application/json" { + b.ErrorLogger.Println("Only json encoding is allowed") + return "invalid_message" + } else if delivery.ContentEncoding != "utf-8" { + b.ErrorLogger.Println("Unsupported content encoding") + return "invalid_message" + } else if handler, ok = b.consumers[delivery.RoutingKey]; !ok { + b.ErrorLogger.Println("Receive unregistered message") + return "invalid_message" + } + + var payload []json.RawMessage + if err := json.Unmarshal(delivery.Body, &payload); err != nil { + b.ErrorLogger.Printf("json decode error: %v\n", err) + return "parse_error" + } else if err := handler.ParsePayload(payload[0], payload[1]); err != nil { + b.ErrorLogger.Printf("handler parse payload error: %v\n", err) + return "parse_error" + } else if err := handler.Run(); err != nil { + b.ErrorLogger.Printf("handler run task failed: %v\n", err) + return "failure" + } + return "success" +} + func (b *BusExt) handleReconnect(brokerUrl string, tlsConfig *tls.Config) { for { b.isReady = false diff --git a/extensions/busext/amqp_test.go b/extensions/busext/amqp_test.go index 85845104..e1cf4289 100644 --- a/extensions/busext/amqp_test.go +++ b/extensions/busext/amqp_test.go @@ -1,11 +1,32 @@ +/* +# Partition table (ECP + BVA) — step 01-asynctask-bus-metrics (busext side) +# +# 参数/状态 | 等价类 | 类型 | 代表值/场景 | 期望输出 | 对应契约条目 +# monitorEnabled | 关闭（默认/未设置） | 有效 | busmonoff_ 在 "testing" env 下未设置 monitor_enable | Consume() 正常处理消息不受影响，不 Observe | 数据层 API 锚点 4 +# monitorEnabled | 开启 | 有效 | busmon_monitor_enable: true | monitorEnabled == true | 数据层 API 锚点 5/6/7/8 +# dispatch status | success | 有效（正常路径） | 已注册 handler，ParsePayload/Run 都成功 | status="success" 计数增加 | 数据层 API 锚点 5 +# dispatch status | failure | 有效（错误路径） | handler.Run() 返回 error | status="failure" 计数增加 | 数据层 API 锚点 6 +# dispatch status | invalid_message | 无效（错误路径） | 消息路由到已绑定队列，但 routing key 未注册消费者 | status="invalid_message" 计数增加 | 数据层 API 锚点 7 +# dispatch status | parse_error | 无效（错误路径） | handler.ParsePayload() 返回 error | status="parse_error" 计数增加 | 数据层 API 锚点 8 +# +# 说明：busext 侧的独立参数只有 2 个（monitorEnabled ×2、dispatch status ×4）， +# 未达到 pairwise 触发阈值（≥3 参数 × 每个 ≥2 取值），因此未调用 pairwise 脚本， +# 改为对 dispatch status 的 4 个等价类逐一穷举（success/failure/invalid_message/ +# parse_error 各一个断言），并单独覆盖 monitorEnabled=false 的等价类代表。 +*/ package busext import ( "encoding/json" + "errors" + "io" "log" + "net/http" + "sync" "testing" "time" + "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/shanbay/gobay" "github.com/shanbay/gobay/extensions/sentryext/custom_logger" "github.com/streadway/amqp" @@ -142,3 +163,172 @@ func (o *TestHandler) Run() error { result = append(result, o) return nil } + +// monitorSuccessHandler / monitorFailureHandler / monitorParseErrorHandler are +// used exclusively by the busext monitor tests below, each representing one +// equivalence class of the dispatch() status partition (success / failure / +// parse_error). The "invalid_message" class doesn't need a handler at all — +// it's triggered by deliberately leaving a bound routing key unregistered. +type monitorSuccessHandler struct{} + +func (h *monitorSuccessHandler) ParsePayload(args []byte, kwargs []byte) error { return nil } +func (h *monitorSuccessHandler) Run() error { return nil } + +type monitorFailureHandler struct{} + +func (h *monitorFailureHandler) ParsePayload(args []byte, kwargs []byte) error { return nil } +func (h *monitorFailureHandler) Run() error { + return errors.New("intentional failure for busext monitor test") +} + +type monitorParseErrorHandler struct{} + +func (h *monitorParseErrorHandler) ParsePayload(args []byte, kwargs []byte) error { + return errors.New("intentional parse error for busext monitor test") +} +func (h *monitorParseErrorHandler) Run() error { return nil } + +var busMetricsOnce sync.Once + +const busMetricsAddr = "127.0.0.1:2114" + +// startBusMetricsServer exposes prometheus.DefaultRegisterer via +// promhttp.Handler(), mirroring extensions/cachext's TestCacheExt_Cached_Monitor +// pattern (config-gated instrumentation + no self-built /metrics server in +// production code, only in the test harness). +func startBusMetricsServer() { + busMetricsOnce.Do(func() { + go func() { + http.Handle("/metrics", promhttp.Handler()) + if err := http.ListenAndServe(busMetricsAddr, nil); err != nil { + log.Fatalf("error when start prometheus server: %v\n", err) + } + }() + time.Sleep(200 * time.Millisecond) + }) +} + +func fetchBusMetrics(t *testing.T) string { + resp, err := http.Get("http://" + busMetricsAddr + "/metrics") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + data, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + return string(data) +} + +// TestBusExt_Monitor_Disabled 覆盖锚点 4：monitor_enable 默认关闭时， +// monitorEnabled 保持 false，Consume() 正常处理消息不受影响，且不产生任何 +// bus_task_duration_seconds 记录。 +func TestBusExt_Monitor_Disabled(t *testing.T) { + startBusMetricsServer() + + busMonOff := BusExt{NS: "busmonoff_"} + busMonOff.ErrorLogger = custom_logger.NewSentryErrorLogger() + offApp, err := gobay.CreateApp( + "../../testdata", + "testing", + map[gobay.Key]gobay.Extension{"busmonoff": &busMonOff}, + ) + assert.Nil(t, err) + assert.Nil(t, offApp.Init()) + assert.False(t, busMonOff.monitorEnabled) + + routingKey := "gobay.buses.busmonoff" + busMonOff.Register(routingKey, &monitorSuccessHandler{}) + go func() { + if err := busMonOff.Consume(); err != nil { + t.Error(err) + } + }() + time.Sleep(500 * time.Millisecond) + + msg, err := BuildMsg(routingKey, []interface{}{}, map[string]interface{}{"n": 1}) + assert.Nil(t, err) + assert.Nil(t, busMonOff.Push("sbay-exchange", routingKey, *msg)) + time.Sleep(1 * time.Second) + + data := fetchBusMetrics(t) + assert.NotContains(t, data, + `bus_task_duration_seconds_count{queue="`+routingKey+`",status="success",task_name="`+routingKey+`"}`) +} + +// TestBusExt_Monitor covers 锚点 5/6/7/8：monitor_enable=true 下， +// dispatch() 的四种状态（success/failure/invalid_message/parse_error） +// 都会被各自记录到 bus_task_duration_seconds。 +func TestBusExt_Monitor(t *testing.T) { + startBusMetricsServer() + + busMon := BusExt{NS: "busmon_"} + busMon.ErrorLogger = custom_logger.NewSentryErrorLogger() + monApp, err := gobay.CreateApp( + "../../testdata", + "busmonitored", + map[gobay.Key]gobay.Extension{"busmon": &busMon}, + ) + assert.Nil(t, err) + assert.Nil(t, monApp.Init()) + assert.True(t, busMon.monitorEnabled) + + busMon.Register("gobay.buses.busmon", &monitorSuccessHandler{}) + busMon.Register("gobay.buses.busmon.failure", &monitorFailureHandler{}) + busMon.Register("gobay.buses.busmon.parseerror", &monitorParseErrorHandler{}) + // gobay.buses.busmon.unregistered 故意不注册消费者，用于验证 invalid_message 分支 + + go func() { + if err := busMon.Consume(); err != nil { + t.Error(err) + } + }() + time.Sleep(500 * time.Millisecond) + + push := func(t *testing.T, routingKey string) { + msg, err := BuildMsg(routingKey, []interface{}{}, map[string]interface{}{"n": 1}) + assert.Nil(t, err) + assert.Nil(t, busMon.Push("sbay-exchange", routingKey, *msg)) + } + + t.Run("5: 成功处理的消息 -> status=success 计数增加", func(t *testing.T) { + routingKey := "gobay.buses.busmon" + push(t, routingKey) + time.Sleep(1 * time.Second) + + data := fetchBusMetrics(t) + labels := `queue="` + routingKey + `",status="success",task_name="` + routingKey + `"` + assert.Contains(t, data, `bus_task_duration_seconds_count{`+labels+`} 1`) + }) + + t.Run("6: handler.Run() 返回 error -> status=failure", func(t *testing.T) { + routingKey := "gobay.buses.busmon.failure" + push(t, routingKey) + time.Sleep(1 * time.Second) + + data := fetchBusMetrics(t) + labels := `queue="` + routingKey + `",status="failure",task_name="` + routingKey + `"` + assert.Contains(t, data, `bus_task_duration_seconds_count{`+labels+`} 1`) + }) + + t.Run("7: 未注册 routing key 的消息 -> status=invalid_message", func(t *testing.T) { + routingKey := "gobay.buses.busmon.unregistered" + push(t, routingKey) + time.Sleep(1 * time.Second) + + data := fetchBusMetrics(t) + labels := `queue="` + routingKey + `",status="invalid_message",task_name="` + routingKey + `"` + assert.Contains(t, data, `bus_task_duration_seconds_count{`+labels+`} 1`) + }) + + t.Run("8: payload 无法解析的消息 -> status=parse_error", func(t *testing.T) { + routingKey := "gobay.buses.busmon.parseerror" + push(t, routingKey) + time.Sleep(1 * time.Second) + + data := fetchBusMetrics(t) + labels := `queue="` + routingKey + `",status="parse_error",task_name="` + routingKey + `"` + assert.Contains(t, data, `bus_task_duration_seconds_count{`+labels+`} 1`) + }) +} diff --git a/observability/metrics.go b/observability/metrics.go new file mode 100644 index 00000000..3ee27faf --- /dev/null +++ b/observability/metrics.go @@ -0,0 +1,35 @@ +package observability + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +// taskDurationBuckets 与 Python coast 库 coast/celery.py 里的 +// _TASK_DURATION_BUCKETS 完全一致，覆盖秒级 bus 消息到分钟级 asynctask 长任务， +// 保证跨语言 histogram_quantile 聚合不失真。 +var taskDurationBuckets = []float64{0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600} + +// AsyncTaskDurationSeconds 与 Python coast 库的 ASYNC_TASK_DURATION_SECONDS +// 同名同 label 同 buckets，可以在同一个 Prometheus/Grafana 查询里合并两边的数据。 +// 包级 var，Go 保证进程内只初始化一次——即便同一个 extension 类型有多个实例 +// （如 one_asynctask_/two_asynctask_）反复 Init 也不会重复 promauto 注册导致 panic。 +var AsyncTaskDurationSeconds = promauto.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "asynctask_task_duration_seconds", + Help: "Time spent processing an asynctask message", + Buckets: taskDurationBuckets, + }, + []string{"task_name", "queue", "status"}, +) + +// BusTaskDurationSeconds 与 Python coast 库的 BUS_TASK_DURATION_SECONDS +// 同名同 label 同 buckets。 +var BusTaskDurationSeconds = promauto.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "bus_task_duration_seconds", + Help: "Time spent processing a bus message", + Buckets: taskDurationBuckets, + }, + []string{"task_name", "queue", "status"}, +) diff --git a/testdata/config.yaml b/testdata/config.yaml index d794e8a3..5edea0fc 100644 --- a/testdata/config.yaml +++ b/testdata/config.yaml @@ -71,6 +71,25 @@ defaults: &defaults redisnoprefix_password: "" redisnoprefix_db: 0 redisnoprefix_prefix: "" + + # busmonoff_ is used to verify busext metrics stay silent when monitor_enable + # is left at its default (false/unset), sharing the "testing" env so it does + # not need its own config block. + busmonoff_broker_url: "amqp://guest:guest@127.0.0.1:5672/" + busmonoff_reconnect_delay: "2s" + busmonoff_reinit_delay: "1s" + busmonoff_exchanges: + - sbay-exchange + busmonoff_queues: + - busmonoff_test + busmonoff_resend_delay: "1s" + busmonoff_publish_retry: 5 + busmonoff_push_timeout: "3s" + busmonoff_prefetch: 10 + busmonoff_bindings: + - exchange: sbay-exchange + queue: busmonoff_test + binding_key: gobay.buses.busmonoff testing: <<: *defaults db_driver: mysql @@ -87,6 +106,58 @@ grpcmocked: cachemonitored: <<: *defaults cache_monitor_enable: true +asynctaskmonitored: + <<: *defaults + # three_asynctask_ / four_asynctask_ are dedicated NS prefixes (distinct + # from one_asynctask_/two_asynctask_ used by the "testing" env) so the + # monitor-enabled tests get their own queues and don't collide with the + # always-on TestPushConsume workers. + three_asynctask_concurrency: 10 + three_asynctask_broker: "redis://127.0.0.1:6379/8" + three_asynctask_default_queue: "gobay.task.three" + three_asynctask_result_backend: "redis://127.0.0.1:6379/8" + three_asynctask_results_expire_in: 1 + three_asynctask_redis: {} + three_asynctask_monitor_enable: true + + four_asynctask_concurrency: 10 + four_asynctask_broker: "redis://127.0.0.1:6379/8" + four_asynctask_default_queue: "gobay.task.four" + four_asynctask_result_backend: "redis://127.0.0.1:6379/8" + four_asynctask_results_expire_in: 1 + four_asynctask_redis: {} + four_asynctask_monitor_enable: true +busmonitored: + <<: *defaults + busmon_broker_url: "amqp://guest:guest@127.0.0.1:5672/" + busmon_reconnect_delay: "2s" + busmon_reinit_delay: "1s" + busmon_exchanges: + - sbay-exchange + busmon_queues: + - busmon_test + busmon_resend_delay: "1s" + busmon_publish_retry: 5 + busmon_push_timeout: "3s" + busmon_prefetch: 10 + busmon_bindings: + - exchange: sbay-exchange + queue: busmon_test + binding_key: gobay.buses.busmon + - exchange: sbay-exchange + queue: busmon_test + binding_key: gobay.buses.busmon.failure + - exchange: sbay-exchange + queue: busmon_test + binding_key: gobay.buses.busmon.parseerror + # gobay.buses.busmon.unregistered is bound to the queue but deliberately + # never registered as a consumer (see busmon.Register calls in + # amqp_test.go), so a message published with this routing key exercises + # the "invalid_message" (unregistered routing key) dispatch branch. + - exchange: sbay-exchange + queue: busmon_test + binding_key: gobay.buses.busmon.unregistered + busmon_monitor_enable: true development: <<: *defaults production: