feat: expose mysql transaction retry metrics
This commit is contained in:
@@ -40,6 +40,8 @@ func catalog() map[string]descriptor {
|
||||
}
|
||||
c["orbit_db_wait_count_total"] = descriptor{counter, "Cumulative waits for a MySQL connection", nil}
|
||||
c["orbit_db_wait_duration_seconds_total"] = descriptor{counter, "Cumulative seconds waiting for a MySQL connection", nil}
|
||||
c["orbit_db_transaction_retries_total"] = descriptor{counter, "Cumulative pre-commit MySQL transaction retries", nil}
|
||||
c["orbit_db_transaction_retry_exhausted_total"] = descriptor{counter, "Cumulative pre-commit MySQL transaction retry exhaustion", nil}
|
||||
c["orbit_pulse_tasks_state"] = descriptor{gauge, "Bounded count of Pulse tasks by durable state", map[string][]string{"state": taskStates}}
|
||||
c["orbit_pulse_tasks_oldest_age_seconds"] = descriptor{gauge, "Age of the oldest sampled unfinished Pulse task", nil}
|
||||
c["orbit_result_deliveries_state"] = descriptor{gauge, "Bounded count of result deliveries by durable state", map[string][]string{"state": deliveryStates}}
|
||||
|
||||
@@ -102,8 +102,9 @@ func TestRuntimeObservationDoesNotExposeErrors(t *testing.T) {
|
||||
r.RecordBatch("workflow", time.Now(), errors.New("private-dsn"))
|
||||
r.RecordBatch("schedule", time.Now(), context.Canceled)
|
||||
r.RecordDatabase(sql.DBStats{MaxOpenConnections: 20, OpenConnections: 4, InUse: 3, Idle: 1, WaitCount: 9, WaitDuration: 2 * time.Second})
|
||||
r.RecordTransactionStats(TransactionStats{Retries: 3, Exhausted: 1})
|
||||
out := r.Render()
|
||||
for _, want := range []string{`orbit_runtime_batch_total{component="workflow",outcome="success"} 1`, `orbit_runtime_batch_total{component="workflow",outcome="error"} 1`, `orbit_runtime_batch_total{component="schedule",outcome="cancelled"} 1`, "orbit_db_connections_in_use 3\n", "orbit_db_wait_duration_seconds_total 2\n"} {
|
||||
for _, want := range []string{`orbit_runtime_batch_total{component="workflow",outcome="success"} 1`, `orbit_runtime_batch_total{component="workflow",outcome="error"} 1`, `orbit_runtime_batch_total{component="schedule",outcome="cancelled"} 1`, "orbit_db_connections_in_use 3\n", "orbit_db_wait_duration_seconds_total 2\n", "orbit_db_transaction_retries_total 3\n", "orbit_db_transaction_retry_exhausted_total 1\n"} {
|
||||
if !strings.Contains(out, want) {
|
||||
t.Fatal("missing runtime evidence", want)
|
||||
}
|
||||
|
||||
@@ -102,6 +102,9 @@ func (r *Registry) RecordBatch(component string, started time.Time, err error) {
|
||||
|
||||
// RecordDatabase reads the local connection-pool snapshot; it issues no SQL.
|
||||
func (r *Registry) RecordDatabase(s sql.DBStats) {
|
||||
if r == nil {
|
||||
return
|
||||
}
|
||||
r.SetGauge("orbit_db_connections_open", float64(s.OpenConnections), nil)
|
||||
r.SetGauge("orbit_db_connections_in_use", float64(s.InUse), nil)
|
||||
r.SetGauge("orbit_db_connections_idle", float64(s.Idle), nil)
|
||||
@@ -109,3 +112,18 @@ func (r *Registry) RecordDatabase(s sql.DBStats) {
|
||||
r.SetCounter("orbit_db_wait_count_total", float64(s.WaitCount), nil)
|
||||
r.SetCounter("orbit_db_wait_duration_seconds_total", s.WaitDuration.Seconds(), nil)
|
||||
}
|
||||
|
||||
// TransactionStats 是本进程提交前事务重试的累计观测值。
|
||||
type TransactionStats struct {
|
||||
Retries uint64 // 已实际执行的重试次数。
|
||||
Exhausted uint64 // 达到重试上限的事务次数。
|
||||
}
|
||||
|
||||
// RecordTransactionStats records bounded database transaction retry counters.
|
||||
func (r *Registry) RecordTransactionStats(stats TransactionStats) {
|
||||
if r == nil {
|
||||
return
|
||||
}
|
||||
r.SetCounter("orbit_db_transaction_retries_total", float64(stats.Retries), nil)
|
||||
r.SetCounter("orbit_db_transaction_retry_exhausted_total", float64(stats.Exhausted), nil)
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
driver "github.com/go-sql-driver/mysql"
|
||||
@@ -53,6 +54,24 @@ func Open(ctx context.Context, dsn string, maxOpen, maxIdle int) (*sql.DB, error
|
||||
|
||||
const transactionRetryLimit = 2
|
||||
|
||||
var transactionRetries atomic.Uint64
|
||||
var transactionRetryExhausted atomic.Uint64
|
||||
|
||||
// TransactionStats 是本进程提交前事务重试的累计快照。
|
||||
// 计数器只记录明确的 MySQL 锁等待超时或死锁回调失败,不记录提交回执不明。
|
||||
type TransactionStats struct {
|
||||
Retries uint64 // 已实际执行的重试次数。
|
||||
Exhausted uint64 // 达到重试上限的事务次数。
|
||||
}
|
||||
|
||||
// ReadTransactionStats 返回本进程事务重试计数,不访问数据库。
|
||||
func ReadTransactionStats() TransactionStats {
|
||||
return TransactionStats{
|
||||
Retries: transactionRetries.Load(),
|
||||
Exhausted: transactionRetryExhausted.Load(),
|
||||
}
|
||||
}
|
||||
|
||||
// WithTx retries only a callback failure caused by a MySQL deadlock or lock wait
|
||||
// timeout before COMMIT. It never retries Begin/Commit errors because a commit
|
||||
// response may be lost after the database has already applied the transaction;
|
||||
@@ -61,11 +80,15 @@ func WithTx(ctx context.Context, db *sql.DB, fn func(*sql.Tx) error) error {
|
||||
for attempt := 0; attempt <= transactionRetryLimit; attempt++ {
|
||||
err, retryable := withTxAttempt(ctx, db, fn)
|
||||
if err == nil || !retryable || attempt == transactionRetryLimit {
|
||||
if retryable && attempt == transactionRetryLimit {
|
||||
transactionRetryExhausted.Add(1)
|
||||
}
|
||||
return err
|
||||
}
|
||||
if err := waitTransactionRetry(ctx, attempt); err != nil {
|
||||
return err
|
||||
}
|
||||
transactionRetries.Add(1)
|
||||
}
|
||||
return errors.New("transaction retry loop exhausted")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user