feat(billing): add queue_name column to usage records
This commit is contained in:
@@ -82,12 +82,18 @@ func autoMigrate() error {
|
|||||||
id BIGSERIAL PRIMARY KEY,
|
id BIGSERIAL PRIMARY KEY,
|
||||||
tenant_id TEXT NOT NULL,
|
tenant_id TEXT NOT NULL,
|
||||||
operation TEXT NOT NULL,
|
operation TEXT NOT NULL,
|
||||||
|
queue_name TEXT NOT NULL DEFAULT '',
|
||||||
msg_count INTEGER DEFAULT 1,
|
msg_count INTEGER DEFAULT 1,
|
||||||
msg_bytes BIGINT DEFAULT 0,
|
msg_bytes BIGINT DEFAULT 0,
|
||||||
recorded_at TIMESTAMPTZ DEFAULT NOW()
|
recorded_at TIMESTAMPTZ DEFAULT NOW()
|
||||||
);
|
);
|
||||||
CREATE INDEX IF NOT EXISTS idx_sqs_usage_tenant_time
|
CREATE INDEX IF NOT EXISTS idx_sqs_usage_tenant_time
|
||||||
ON sqs_usage_records(tenant_id, recorded_at);
|
ON sqs_usage_records(tenant_id, recorded_at);
|
||||||
|
-- Миграция: добавляем queue_name если таблица уже существует без него
|
||||||
|
DO $$ BEGIN
|
||||||
|
ALTER TABLE sqs_usage_records ADD COLUMN queue_name TEXT NOT NULL DEFAULT '';
|
||||||
|
EXCEPTION WHEN duplicate_column THEN NULL;
|
||||||
|
END $$;
|
||||||
`)
|
`)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -95,15 +101,15 @@ func autoMigrate() error {
|
|||||||
// RecordUsage — записывает одну SQS-операцию в таблицу биллинга.
|
// RecordUsage — записывает одну SQS-операцию в таблицу биллинга.
|
||||||
// Вызывается асинхронно (горутина) чтобы не добавлять latency к SQS-ответу.
|
// Вызывается асинхронно (горутина) чтобы не добавлять latency к SQS-ответу.
|
||||||
// Если billing отключён — no-op. Ошибки логируются, SQS-операция не ломается.
|
// Если billing отключён — no-op. Ошибки логируются, SQS-операция не ломается.
|
||||||
func RecordUsage(tenantID, operation string, msgCount int, msgBytes int64) {
|
func RecordUsage(tenantID, operation, queueName string, msgCount int, msgBytes int64) {
|
||||||
if db == nil {
|
if db == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
_, err := db.Exec(
|
_, err := db.Exec(
|
||||||
`INSERT INTO sqs_usage_records (tenant_id, operation, msg_count, msg_bytes) VALUES ($1, $2, $3, $4)`,
|
`INSERT INTO sqs_usage_records (tenant_id, operation, queue_name, msg_count, msg_bytes) VALUES ($1, $2, $3, $4, $5)`,
|
||||||
tenantID, operation, msgCount, msgBytes,
|
tenantID, operation, queueName, msgCount, msgBytes,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("billing: failed to record usage [%s/%s]: %v", tenantID, operation, err)
|
log.Errorf("billing: failed to record usage [%s/%s]: %v", tenantID, operation, err)
|
||||||
|
|||||||
@@ -139,7 +139,12 @@ func actionHandler(w http.ResponseWriter, req *http.Request) {
|
|||||||
if msgBytes < 0 {
|
if msgBytes < 0 {
|
||||||
msgBytes = 0
|
msgBytes = 0
|
||||||
}
|
}
|
||||||
billing.RecordUsage(t.ID, action, 1, msgBytes)
|
// Извлекаем имя очереди: из URL path vars или из form-параметра QueueName (CreateQueue)
|
||||||
|
queueName := mux.Vars(req)["queueName"]
|
||||||
|
if queueName == "" {
|
||||||
|
queueName = req.FormValue("QueueName")
|
||||||
|
}
|
||||||
|
billing.RecordUsage(t.ID, action, queueName, 1, msgBytes)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
|
|||||||
Reference in New Issue
Block a user