P2.5: биллинг — воркер+буфер вместо горутины на операцию; корректный msg_count
RecordUsage теперь кладёт запись в буферизованный канал (1024), один воркер пишет в PG — убран навал горутин, ждущих коннект пула (было 5). При переполнении буфера запись дропается с логом (биллинг жертвуем ради стабильности). Для batch-операций (SendMessageBatch/DeleteMessageBatch/ChangeMessageVisibilityBatch) msg_count теперь реальное число записей, а не всегда 1.
This commit is contained in:
+44
-13
@@ -98,23 +98,54 @@ func autoMigrate() error {
|
||||
return err
|
||||
}
|
||||
|
||||
// RecordUsage — записывает одну SQS-операцию в таблицу биллинга.
|
||||
// Вызывается асинхронно (горутина) чтобы не добавлять latency к SQS-ответу.
|
||||
// Если billing отключён — no-op. Ошибки логируются, SQS-операция не ломается.
|
||||
// usageRecord — одна запись биллинга, накапливается в буфере воркера.
|
||||
type usageRecord struct {
|
||||
tenantID string
|
||||
operation string
|
||||
queueName string
|
||||
msgCount int
|
||||
msgBytes int64
|
||||
}
|
||||
|
||||
// usageChan — канал-буфер для асинхронных записей биллинга.
|
||||
// Вместо горутины на каждую операцию (навал горутин под нагрузкой) используется
|
||||
// один воркер, читающий из буферизованного канала.
|
||||
var usageChan = make(chan usageRecord, 1024)
|
||||
|
||||
// init запускает воркер-горутину биллинга. Воркер стартует всегда; если db == nil
|
||||
// (billing отключён) — он просто дропает записи.
|
||||
func init() {
|
||||
go usageWorker()
|
||||
}
|
||||
|
||||
// usageWorker — единственная горутина, пишущая usage-записи в PostgreSQL.
|
||||
func usageWorker() {
|
||||
for rec := range usageChan {
|
||||
if db == nil {
|
||||
continue
|
||||
}
|
||||
_, err := db.Exec(
|
||||
`INSERT INTO sqs_usage_records (tenant_id, operation, queue_name, msg_count, msg_bytes) VALUES ($1, $2, $3, $4, $5)`,
|
||||
rec.tenantID, rec.operation, rec.queueName, rec.msgCount, rec.msgBytes,
|
||||
)
|
||||
if err != nil {
|
||||
log.Errorf("billing: failed to record usage [%s/%s]: %v", rec.tenantID, rec.operation, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// RecordUsage — ставит запись в очередь биллинга (асинхронно, без блокировки SQS).
|
||||
// Если billing отключён — no-op. Если буфер переполнен — запись дропается (биллинг
|
||||
// жертвуем ради стабильности основного пути), ошибка логируется.
|
||||
func RecordUsage(tenantID, operation, queueName string, msgCount int, msgBytes int64) {
|
||||
if db == nil {
|
||||
return
|
||||
}
|
||||
|
||||
go func() {
|
||||
_, err := db.Exec(
|
||||
`INSERT INTO sqs_usage_records (tenant_id, operation, queue_name, msg_count, msg_bytes) VALUES ($1, $2, $3, $4, $5)`,
|
||||
tenantID, operation, queueName, msgCount, msgBytes,
|
||||
)
|
||||
if err != nil {
|
||||
log.Errorf("billing: failed to record usage [%s/%s]: %v", tenantID, operation, err)
|
||||
}
|
||||
}()
|
||||
select {
|
||||
case usageChan <- usageRecord{tenantID, operation, queueName, msgCount, msgBytes}:
|
||||
default:
|
||||
log.Warnf("billing: буфер полон, запись [%s/%s] дропнута", tenantID, operation)
|
||||
}
|
||||
}
|
||||
|
||||
// Enabled — возвращает true если billing подключён к PostgreSQL
|
||||
|
||||
Reference in New Issue
Block a user