Skip to content

Commit 30995b5

Browse files
Wei-ShawWei-Shaw
authored
Merge pull request Wei-Shaw#936 from xvhuan/fix/ops-write-pressure-20260311
降低 ops_error_logs 与 scheduler_outbox 的数据库写放大
2 parents eb60f67 + fbd73f2 commit 30995b5

8 files changed

Lines changed: 498 additions & 103 deletions

File tree

‎backend/internal/handler/ops_error_logger.go‎

Lines changed: 71 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -31,13 +31,15 @@ const (
3131
const (
3232
opsErrorLogTimeout = 5 * time.Second
3333
opsErrorLogDrainTimeout = 10 * time.Second
34+
opsErrorLogBatchWindow = 200 * time.Millisecond
3435

3536
opsErrorLogMinWorkerCount = 4
3637
opsErrorLogMaxWorkerCount = 32
3738

3839
opsErrorLogQueueSizePerWorker = 128
3940
opsErrorLogMinQueueSize = 256
4041
opsErrorLogMaxQueueSize = 8192
42+
opsErrorLogBatchSize = 32
4143
)
4244

4345
type opsErrorLogJob struct {
@@ -82,27 +84,82 @@ func startOpsErrorLogWorkers() {
8284
for i := 0; i < workerCount; i++ {
8385
go func() {
8486
defer opsErrorLogWorkersWg.Done()
85-
for job := range opsErrorLogQueue {
86-
opsErrorLogQueueLen.Add(-1)
87-
if job.ops == nil || job.entry == nil {
88-
continue
87+
for {
88+
job, ok := <-opsErrorLogQueue
89+
if !ok {
90+
return
8991
}
90-
func() {
91-
defer func() {
92-
if r := recover(); r != nil {
93-
log.Printf("[OpsErrorLogger] worker panic: %v\n%s", r, debug.Stack())
92+
opsErrorLogQueueLen.Add(-1)
93+
batch := make([]opsErrorLogJob, 0, opsErrorLogBatchSize)
94+
batch = append(batch, job)
95+
96+
timer := time.NewTimer(opsErrorLogBatchWindow)
97+
batchLoop:
98+
for len(batch) < opsErrorLogBatchSize {
99+
select {
100+
case nextJob, ok := <-opsErrorLogQueue:
101+
if !ok {
102+
if !timer.Stop() {
103+
select {
104+
case <-timer.C:
105+
default:
106+
}
107+
}
108+
flushOpsErrorLogBatch(batch)
109+
return
94110
}
95-
}()
96-
ctx, cancel := context.WithTimeout(context.Background(), opsErrorLogTimeout)
97-
_ = job.ops.RecordError(ctx, job.entry, nil)
98-
cancel()
99-
opsErrorLogProcessed.Add(1)
100-
}()
111+
opsErrorLogQueueLen.Add(-1)
112+
batch = append(batch, nextJob)
113+
case <-timer.C:
114+
break batchLoop
115+
}
116+
}
117+
if !timer.Stop() {
118+
select {
119+
case <-timer.C:
120+
default:
121+
}
122+
}
123+
flushOpsErrorLogBatch(batch)
101124
}
102125
}()
103126
}
104127
}
105128

129+
func flushOpsErrorLogBatch(batch []opsErrorLogJob) {
130+
if len(batch) == 0 {
131+
return
132+
}
133+
defer func() {
134+
if r := recover(); r != nil {
135+
log.Printf("[OpsErrorLogger] worker panic: %v\n%s", r, debug.Stack())
136+
}
137+
}()
138+
139+
grouped := make(map[*service.OpsService][]*service.OpsInsertErrorLogInput, len(batch))
140+
var processed int64
141+
for _, job := range batch {
142+
if job.ops == nil || job.entry == nil {
143+
continue
144+
}
145+
grouped[job.ops] = append(grouped[job.ops], job.entry)
146+
processed++
147+
}
148+
if processed == 0 {
149+
return
150+
}
151+
152+
for opsSvc, entries := range grouped {
153+
if opsSvc == nil || len(entries) == 0 {
154+
continue
155+
}
156+
ctx, cancel := context.WithTimeout(context.Background(), opsErrorLogTimeout)
157+
_ = opsSvc.RecordErrorBatch(ctx, entries)
158+
cancel()
159+
}
160+
opsErrorLogProcessed.Add(processed)
161+
}
162+
106163
func enqueueOpsErrorLog(ops *service.OpsService, entry *service.OpsInsertErrorLogInput) {
107164
if ops == nil || entry == nil {
108165
return

‎backend/internal/repository/ops_repo.go‎

Lines changed: 68 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -16,19 +16,7 @@ type opsRepository struct {
1616
db *sql.DB
1717
}
1818

19-
func NewOpsRepository(db *sql.DB) service.OpsRepository {
20-
return &opsRepository{db: db}
21-
}
22-
23-
func (r *opsRepository) InsertErrorLog(ctx context.Context, input *service.OpsInsertErrorLogInput) (int64, error) {
24-
if r == nil || r.db == nil {
25-
return 0, fmt.Errorf("nil ops repository")
26-
}
27-
if input == nil {
28-
return 0, fmt.Errorf("nil input")
29-
}
30-
31-
q := `
19+
const insertOpsErrorLogSQL = `
3220
INSERT INTO ops_error_logs (
3321
request_id,
3422
client_request_id,
@@ -70,12 +58,77 @@ INSERT INTO ops_error_logs (
7058
created_at
7159
) VALUES (
7260
$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31,$32,$33,$34,$35,$36,$37,$38
73-
) RETURNING id`
61+
)`
62+
63+
func NewOpsRepository(db *sql.DB) service.OpsRepository {
64+
return &opsRepository{db: db}
65+
}
66+
67+
func (r *opsRepository) InsertErrorLog(ctx context.Context, input *service.OpsInsertErrorLogInput) (int64, error) {
68+
if r == nil || r.db == nil {
69+
return 0, fmt.Errorf("nil ops repository")
70+
}
71+
if input == nil {
72+
return 0, fmt.Errorf("nil input")
73+
}
7474

7575
var id int64
7676
err := r.db.QueryRowContext(
7777
ctx,
78-
q,
78+
insertOpsErrorLogSQL+" RETURNING id",
79+
opsInsertErrorLogArgs(input)...,
80+
).Scan(&id)
81+
if err != nil {
82+
return 0, err
83+
}
84+
return id, nil
85+
}
86+
87+
func (r *opsRepository) BatchInsertErrorLogs(ctx context.Context, inputs []*service.OpsInsertErrorLogInput) (int64, error) {
88+
if r == nil || r.db == nil {
89+
return 0, fmt.Errorf("nil ops repository")
90+
}
91+
if len(inputs) == 0 {
92+
return 0, nil
93+
}
94+
95+
tx, err := r.db.BeginTx(ctx, nil)
96+
if err != nil {
97+
return 0, err
98+
}
99+
defer func() {
100+
if err != nil {
101+
_ = tx.Rollback()
102+
}
103+
}()
104+
105+
stmt, err := tx.PrepareContext(ctx, insertOpsErrorLogSQL)
106+
if err != nil {
107+
return 0, err
108+
}
109+
defer func() {
110+
_ = stmt.Close()
111+
}()
112+
113+
var inserted int64
114+
for _, input := range inputs {
115+
if input == nil {
116+
continue
117+
}
118+
if _, err = stmt.ExecContext(ctx, opsInsertErrorLogArgs(input)...); err != nil {
119+
return inserted, err
120+
}
121+
inserted++
122+
}
123+
124+
if err = tx.Commit(); err != nil {
125+
return inserted, err
126+
}
127+
return inserted, nil
128+
}
129+
130+
func opsInsertErrorLogArgs(input *service.OpsInsertErrorLogInput) []any {
131+
return []any{
79132
opsNullString(input.RequestID),
80133
opsNullString(input.ClientRequestID),
81134
opsNullInt64(input.UserID),
@@ -114,11 +167,7 @@ INSERT INTO ops_error_logs (
114167
input.IsRetryable,
115168
input.RetryCount,
116169
input.CreatedAt,
117-
).Scan(&id)
118-
if err != nil {
119-
return 0, err
120170
}
121-
return id, nil
122171
}
123172

124173
func (r *opsRepository) ListErrorLogs(ctx context.Context, filter *service.OpsErrorLogFilter) (*service.OpsErrorLogList, error) {

‎backend/internal/repository/ops_write_pressure_integration_test.go‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
//go:build integration
2+
3+
package repository
4+
5+
import (
6+
"context"
7+
"testing"
8+
"time"
9+
10+
"github.com/Wei-Shaw/sub2api/internal/service"
11+
"github.com/stretchr/testify/require"
12+
)
13+
14+
func TestOpsRepositoryBatchInsertErrorLogs(t *testing.T) {
15+
ctx := context.Background()
16+
_, _ = integrationDB.ExecContext(ctx, "TRUNCATE ops_error_logs RESTART IDENTITY")
17+
18+
repo := NewOpsRepository(integrationDB).(*opsRepository)
19+
now := time.Now().UTC()
20+
inserted, err := repo.BatchInsertErrorLogs(ctx, []*service.OpsInsertErrorLogInput{
21+
{
22+
RequestID: "batch-ops-1",
23+
ErrorPhase: "upstream",
24+
ErrorType: "upstream_error",
25+
Severity: "error",
26+
StatusCode: 429,
27+
ErrorMessage: "rate limited",
28+
CreatedAt: now,
29+
},
30+
{
31+
RequestID: "batch-ops-2",
32+
ErrorPhase: "internal",
33+
ErrorType: "api_error",
34+
Severity: "error",
35+
StatusCode: 500,
36+
ErrorMessage: "internal error",
37+
CreatedAt: now.Add(time.Millisecond),
38+
},
39+
})
40+
require.NoError(t, err)
41+
require.EqualValues(t, 2, inserted)
42+
43+
var count int
44+
require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM ops_error_logs WHERE request_id IN ('batch-ops-1', 'batch-ops-2')").Scan(&count))
45+
require.Equal(t, 2, count)
46+
}
47+
48+
func TestEnqueueSchedulerOutbox_DeduplicatesIdempotentEvents(t *testing.T) {
49+
ctx := context.Background()
50+
_, _ = integrationDB.ExecContext(ctx, "TRUNCATE scheduler_outbox RESTART IDENTITY")
51+
52+
accountID := int64(12345)
53+
require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountChanged, &accountID, nil, nil))
54+
require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountChanged, &accountID, nil, nil))
55+
56+
var count int
57+
require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountChanged).Scan(&count))
58+
require.Equal(t, 1, count)
59+
60+
time.Sleep(schedulerOutboxDedupWindow + 150*time.Millisecond)
61+
require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountChanged, &accountID, nil, nil))
62+
require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountChanged).Scan(&count))
63+
require.Equal(t, 2, count)
64+
}
65+
66+
func TestEnqueueSchedulerOutbox_DoesNotDeduplicateLastUsed(t *testing.T) {
67+
ctx := context.Background()
68+
_, _ = integrationDB.ExecContext(ctx, "TRUNCATE scheduler_outbox RESTART IDENTITY")
69+
70+
accountID := int64(67890)
71+
payload1 := map[string]any{"last_used": map[string]int64{"67890": 100}}
72+
payload2 := map[string]any{"last_used": map[string]int64{"67890": 200}}
73+
require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountLastUsed, &accountID, nil, payload1))
74+
require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountLastUsed, &accountID, nil, payload2))
75+
76+
var count int
77+
require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountLastUsed).Scan(&count))
78+
require.Equal(t, 2, count)
79+
}

‎backend/internal/repository/scheduler_outbox_repo.go‎

Lines changed: 33 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"database/sql"
66
"encoding/json"
7+
"time"
78

89
"github.com/Wei-Shaw/sub2api/internal/service"
910
)
@@ -12,6 +13,8 @@ type schedulerOutboxRepository struct {
1213
db *sql.DB
1314
}
1415

16+
const schedulerOutboxDedupWindow = time.Second
17+
1518
func NewSchedulerOutboxRepository(db *sql.DB) service.SchedulerOutboxRepository {
1619
return &schedulerOutboxRepository{db: db}
1720
}
@@ -88,9 +91,37 @@ func enqueueSchedulerOutbox(ctx context.Context, exec sqlExecutor, eventType str
8891
}
8992
payloadArg = encoded
9093
}
91-
_, err := exec.ExecContext(ctx, `
94+
query := `
9295
INSERT INTO scheduler_outbox (event_type, account_id, group_id, payload)
9396
VALUES ($1, $2, $3, $4)
94-
`, eventType, accountID, groupID, payloadArg)
97+
`
98+
args := []any{eventType, accountID, groupID, payloadArg}
99+
if schedulerOutboxEventSupportsDedup(eventType) {
100+
query = `
101+
INSERT INTO scheduler_outbox (event_type, account_id, group_id, payload)
102+
SELECT $1, $2, $3, $4
103+
WHERE NOT EXISTS (
104+
SELECT 1
105+
FROM scheduler_outbox
106+
WHERE event_type = $1
107+
AND account_id IS NOT DISTINCT FROM $2
108+
AND group_id IS NOT DISTINCT FROM $3
109+
AND created_at >= NOW() - make_interval(secs => $5)
110+
)
111+
`
112+
args = append(args, schedulerOutboxDedupWindow.Seconds())
113+
}
114+
_, err := exec.ExecContext(ctx, query, args...)
95115
return err
96116
}
117+
118+
func schedulerOutboxEventSupportsDedup(eventType string) bool {
119+
switch eventType {
120+
case service.SchedulerOutboxEventAccountChanged,
121+
service.SchedulerOutboxEventGroupChanged,
122+
service.SchedulerOutboxEventFullRebuild:
123+
return true
124+
default:
125+
return false
126+
}
127+
}

‎backend/internal/service/ops_port.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77

88
type OpsRepository interface {
99
InsertErrorLog(ctx context.Context, input *OpsInsertErrorLogInput) (int64, error)
10+
BatchInsertErrorLogs(ctx context.Context, inputs []*OpsInsertErrorLogInput) (int64, error)
1011
ListErrorLogs(ctx context.Context, filter *OpsErrorLogFilter) (*OpsErrorLogList, error)
1112
GetErrorLogByID(ctx context.Context, id int64) (*OpsErrorLogDetail, error)
1213
ListRequestDetails(ctx context.Context, filter *OpsRequestDetailFilter) ([]*OpsRequestDetail, int64, error)

‎backend/internal/service/ops_repo_mock_test.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,16 +7,28 @@ import (
77

88
// opsRepoMock is a test-only OpsRepository implementation with optional function hooks.
99
type opsRepoMock struct {
10+
InsertErrorLogFn func(ctx context.Context, input *OpsInsertErrorLogInput) (int64, error)
11+
BatchInsertErrorLogsFn func(ctx context.Context, inputs []*OpsInsertErrorLogInput) (int64, error)
1012
BatchInsertSystemLogsFn func(ctx context.Context, inputs []*OpsInsertSystemLogInput) (int64, error)
1113
ListSystemLogsFn func(ctx context.Context, filter *OpsSystemLogFilter) (*OpsSystemLogList, error)
1214
DeleteSystemLogsFn func(ctx context.Context, filter *OpsSystemLogCleanupFilter) (int64, error)
1315
InsertSystemLogCleanupAuditFn func(ctx context.Context, input *OpsSystemLogCleanupAudit) error
1416
}
1517

1618
func (m *opsRepoMock) InsertErrorLog(ctx context.Context, input *OpsInsertErrorLogInput) (int64, error) {
19+
if m.InsertErrorLogFn != nil {
20+
return m.InsertErrorLogFn(ctx, input)
21+
}
1722
return 0, nil
1823
}
1924

25+
func (m *opsRepoMock) BatchInsertErrorLogs(ctx context.Context, inputs []*OpsInsertErrorLogInput) (int64, error) {
26+
if m.BatchInsertErrorLogsFn != nil {
27+
return m.BatchInsertErrorLogsFn(ctx, inputs)
28+
}
29+
return int64(len(inputs)), nil
30+
}
31+
2032
func (m *opsRepoMock) ListErrorLogs(ctx context.Context, filter *OpsErrorLogFilter) (*OpsErrorLogList, error) {
2133
return &OpsErrorLogList{Errors: []*OpsErrorLog{}, Page: 1, PageSize: 20}, nil
2234
}

0 commit comments

Comments
 (0)