From fb833ef01c1ffc2046f1a253d3b0ad81201848f0 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Tue, 6 Oct 2026 18:19:17 +0300 Subject: [PATCH] =?UTF-8?q?P2.5:=20=D0=B1=D0=B8=D0=BB=D0=BB=D0=B8=D0=BD?= =?UTF-8?q?=D0=B3=20=E2=80=94=20=D0=B2=D0=BE=D1=80=D0=BA=D0=B5=D1=80+?= =?UTF-8?q?=D0=B1=D1=83=D1=84=D0=B5=D1=80=20=D0=B2=D0=BC=D0=B5=D1=81=D1=82?= =?UTF-8?q?=D0=BE=20=D0=B3=D0=BE=D1=80=D1=83=D1=82=D0=B8=D0=BD=D1=8B=20?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=BE=D0=BF=D0=B5=D1=80=D0=B0=D1=86=D0=B8=D1=8E?= =?UTF-8?q?;=20=D0=BA=D0=BE=D1=80=D1=80=D0=B5=D0=BA=D1=82=D0=BD=D1=8B?= =?UTF-8?q?=D0=B9=20msg=5Fcount?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RecordUsage теперь кладёт запись в буферизованный канал (1024), один воркер пишет в PG — убран навал горутин, ждущих коннект пула (было 5). При переполнении буфера запись дропается с логом (биллинг жертвуем ради стабильности). Для batch-операций (SendMessageBatch/DeleteMessageBatch/ChangeMessageVisibilityBatch) msg_count теперь реальное число записей, а не всегда 1. --- app/billing/billing.go | 57 ++++++++++++++++++++++++++++++++---------- app/router/router.go | 42 +++++++++++++++++++++++++++++-- 2 files changed, 84 insertions(+), 15 deletions(-) diff --git a/app/billing/billing.go b/app/billing/billing.go index 2510a8e..caab50c 100644 --- a/app/billing/billing.go +++ b/app/billing/billing.go @@ -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 diff --git a/app/router/router.go b/app/router/router.go index 8d9105e..8eecf85 100644 --- a/app/router/router.go +++ b/app/router/router.go @@ -200,6 +200,36 @@ func metricsAuth(adminToken string, next http.Handler) http.Handler { }) } +// countBatchEntries — подсчитывает число записей в batch-запросе SQS. +// Для query-протокола записи идут как SendMessageBatchRequestEntry.N.Id (N=1..). +// Для JSON-протокола считаем по количеству записей в массиве Entries (requestBody). +// Возвращает 0 если определить не удалось (тогда биллинг запишет msg_count=1). +func countBatchEntries(req *http.Request, action string) int { + // Query-формат: считаем уникальные N в SendMessageBatchRequestEntry.N.Id и т.п. + prefix := "SendMessageBatchRequestEntry." + if action == "DeleteMessageBatch" { + prefix = "DeleteMessageBatchRequestEntry." + } else if action == "ChangeMessageVisibilityBatch" { + prefix = "ChangeMessageVisibilityBatchRequestEntry." + } + if req.PostForm != nil || req.Form != nil { + form := req.Form + if form == nil { + form = req.PostForm + } + count := 0 + for key := range form { + if strings.HasPrefix(key, prefix) && strings.HasSuffix(key, ".Id") { + count++ + } + } + if count > 0 { + return count + } + } + return 0 +} + func actionHandler(w http.ResponseWriter, req *http.Request) { action := extractAction(req) start := time.Now() @@ -227,8 +257,16 @@ func actionHandler(w http.ResponseWriter, req *http.Request) { if queueName == "" { queueName = req.FormValue("QueueName") } - // Billing: запись в PostgreSQL (async, no-op если billing выключен) - billing.RecordUsage(t.ID, action, queueName, 1, msgBytes) + // Billing: запись в PostgreSQL (async, no-op если billing выключен). + // msgCount для batch-операций — число сообщений в батче; для прочих — 1. + msgCount := 1 + if action == "SendMessageBatch" || action == "DeleteMessageBatch" || + action == "ChangeMessageVisibilityBatch" { + if n := countBatchEntries(req, action); n > 0 { + msgCount = n + } + } + billing.RecordUsage(t.ID, action, queueName, msgCount, msgBytes) // Prometheus counters: requests + bytes metrics.RequestsTotal.WithLabelValues(t.ID, action).Inc() metrics.RequestBytesTotal.WithLabelValues(t.ID, action).Add(float64(msgBytes))