feat: billing — usage tracking per SQS operation in PostgreSQL
- New package app/billing: Init/RecordUsage/Close with auto-migrate - Integration in actionHandler: record tenant_id, operation, msg_count, msg_bytes - Helm chart: billing section in values.yaml, secret-billing.yaml, env vars in deployment - Optional: billing disabled by default (BILLING_PG_HOST not set = no-op) - Table: sqs_usage_records with index on (tenant_id, recorded_at)
This commit is contained in:
Naeel
+10
-2
@@ -1,7 +1,7 @@
|
||||
// app/admin/admin.go
|
||||
// Admin API handlers for shared-sqs management
|
||||
// Created: 2026-04-09
|
||||
// Updated: 2026-04-12 09:28 MSK — demo UI token и изоляция UI API одним tenant-ом
|
||||
// Updated: 2026-04-12 10:12 MSK — пометки о временном demo showcase режиме
|
||||
package admin
|
||||
|
||||
import (
|
||||
@@ -30,6 +30,9 @@ const (
|
||||
uiTenantContextKey uiContextKey = "ui-tenant"
|
||||
)
|
||||
|
||||
// ВАЖНО: значения ниже относятся только к временному demo/showcase режиму.
|
||||
// Перед production rollout без публичного demo-режима этот блок должен быть удалён
|
||||
// вместе с веткой authenticateUIDemoToken и связанными UI-подсказками.
|
||||
const (
|
||||
defaultUIDemoToken = "demo-ui-shared-sqs-ngcloud-2026"
|
||||
defaultUIDemoTenantID = "t-demo-shared-sqs-ngcloud"
|
||||
@@ -67,7 +70,8 @@ func NewHandler(store *tenant.TenantStore, adminToken string) *Handler {
|
||||
return &Handler{store: store, adminToken: adminToken, nubesEndpoint: nubesEndpoint}
|
||||
}
|
||||
|
||||
// uiDemoToken — возвращает публичный demo token для UI, если он не переопределён через env.
|
||||
// uiDemoToken — возвращает публичный demo token для UI.
|
||||
// Временный showcase-only путь: в production без demo user этот метод нужно удалить.
|
||||
func (h *Handler) uiDemoToken() string {
|
||||
if token := os.Getenv("SHARED_SQS_UI_DEMO_TOKEN"); token != "" {
|
||||
return token
|
||||
@@ -76,6 +80,7 @@ func (h *Handler) uiDemoToken() string {
|
||||
}
|
||||
|
||||
// uiDemoTenantID — возвращает tenant ID, к которому привязан demo token.
|
||||
// Существует только для showcase-режима, не для постоянной production auth-модели.
|
||||
func (h *Handler) uiDemoTenantID() string {
|
||||
if tenantID := os.Getenv("SHARED_SQS_UI_DEMO_TENANT_ID"); tenantID != "" {
|
||||
return tenantID
|
||||
@@ -84,6 +89,7 @@ func (h *Handler) uiDemoTenantID() string {
|
||||
}
|
||||
|
||||
// uiDemoEmail — возвращает отображаемый email для demo UI session.
|
||||
// Нужен только для публичного демо-логина и должен уйти вместе с demo path.
|
||||
func (h *Handler) uiDemoEmail() string {
|
||||
if email := os.Getenv("SHARED_SQS_UI_DEMO_EMAIL"); email != "" {
|
||||
return email
|
||||
@@ -92,6 +98,8 @@ func (h *Handler) uiDemoEmail() string {
|
||||
}
|
||||
|
||||
// authenticateUIDemoToken — маппит публичный demo token на заранее сидированный demo tenant.
|
||||
// Почему так: это быстрый showcase-вход для заказчика. Для production без demo user
|
||||
// функция должна быть удалена, чтобы в коде не осталось публичного bypass-пути.
|
||||
func (h *Handler) authenticateUIDemoToken(token string) (*tenant.Tenant, string, error) {
|
||||
demoToken := h.uiDemoToken()
|
||||
if demoToken == "" || subtle.ConstantTimeCompare([]byte(token), []byte(demoToken)) != 1 {
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
// app/billing/billing.go
|
||||
// Модуль учёта использования SQS-операций для биллинга.
|
||||
// Записывает каждую успешную SQS-операцию в PostgreSQL: tenant_id, operation, msg_count, msg_bytes.
|
||||
// Если PostgreSQL не сконфигурирован — billing отключён, SQS работает как раньше.
|
||||
// Created: 2026-04-12
|
||||
package billing
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
// PostgreSQL драйвер — регистрируется в database/sql через init()
|
||||
_ "github.com/lib/pq"
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// db — подключение к PostgreSQL для записи usage-данных.
|
||||
// nil если billing отключён.
|
||||
var db *sql.DB
|
||||
|
||||
// Init — подключается к PostgreSQL и создаёт таблицу sqs_usage_records если не существует.
|
||||
// Env переменные: BILLING_PG_HOST, BILLING_PG_PORT, BILLING_PG_DATABASE, BILLING_PG_USER,
|
||||
// BILLING_PG_PASSWORD, BILLING_PG_SSLMODE.
|
||||
// Если BILLING_PG_HOST не задан — billing отключён, сервис работает без него.
|
||||
func Init() {
|
||||
host := os.Getenv("BILLING_PG_HOST")
|
||||
if host == "" {
|
||||
log.Info("billing: BILLING_PG_HOST not set, usage tracking disabled")
|
||||
return
|
||||
}
|
||||
|
||||
port := os.Getenv("BILLING_PG_PORT")
|
||||
if port == "" {
|
||||
port = "5432"
|
||||
}
|
||||
dbname := os.Getenv("BILLING_PG_DATABASE")
|
||||
user := os.Getenv("BILLING_PG_USER")
|
||||
password := os.Getenv("BILLING_PG_PASSWORD")
|
||||
sslmode := os.Getenv("BILLING_PG_SSLMODE")
|
||||
if sslmode == "" {
|
||||
sslmode = "require"
|
||||
}
|
||||
|
||||
dsn := fmt.Sprintf("host=%s port=%s dbname=%s user=%s password=%s sslmode=%s",
|
||||
host, port, dbname, user, password, sslmode)
|
||||
|
||||
var err error
|
||||
db, err = sql.Open("postgres", dsn)
|
||||
if err != nil {
|
||||
log.Errorf("billing: failed to open PostgreSQL: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Проверяем реальное подключение (Open не подключается)
|
||||
if err = db.Ping(); err != nil {
|
||||
log.Errorf("billing: failed to connect to PostgreSQL: %v", err)
|
||||
db.Close()
|
||||
db = nil
|
||||
return
|
||||
}
|
||||
|
||||
// Ограничиваем пул — billing не должен отжирать коннекты у основной БД
|
||||
db.SetMaxOpenConns(5)
|
||||
db.SetMaxIdleConns(2)
|
||||
|
||||
if err = autoMigrate(); err != nil {
|
||||
log.Errorf("billing: failed to create table: %v", err)
|
||||
db.Close()
|
||||
db = nil
|
||||
return
|
||||
}
|
||||
|
||||
log.Infof("billing: connected to PostgreSQL %s:%s/%s, usage tracking enabled", host, port, dbname)
|
||||
}
|
||||
|
||||
// autoMigrate — создаёт таблицу и индекс если не существуют.
|
||||
// Идемпотентно — безопасно вызывать при каждом старте.
|
||||
func autoMigrate() error {
|
||||
_, err := db.Exec(`
|
||||
CREATE TABLE IF NOT EXISTS sqs_usage_records (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
tenant_id TEXT NOT NULL,
|
||||
operation TEXT NOT NULL,
|
||||
msg_count INTEGER DEFAULT 1,
|
||||
msg_bytes BIGINT DEFAULT 0,
|
||||
recorded_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_sqs_usage_tenant_time
|
||||
ON sqs_usage_records(tenant_id, recorded_at);
|
||||
`)
|
||||
return err
|
||||
}
|
||||
|
||||
// RecordUsage — записывает одну SQS-операцию в таблицу биллинга.
|
||||
// Вызывается асинхронно (горутина) чтобы не добавлять latency к SQS-ответу.
|
||||
// Если billing отключён — no-op. Ошибки логируются, SQS-операция не ломается.
|
||||
func RecordUsage(tenantID, operation string, msgCount int, msgBytes int64) {
|
||||
if db == nil {
|
||||
return
|
||||
}
|
||||
|
||||
go func() {
|
||||
_, err := db.Exec(
|
||||
`INSERT INTO sqs_usage_records (tenant_id, operation, msg_count, msg_bytes) VALUES ($1, $2, $3, $4)`,
|
||||
tenantID, operation, msgCount, msgBytes,
|
||||
)
|
||||
if err != nil {
|
||||
log.Errorf("billing: failed to record usage [%s/%s]: %v", tenantID, operation, err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Enabled — возвращает true если billing подключён к PostgreSQL
|
||||
func Enabled() bool {
|
||||
return db != nil
|
||||
}
|
||||
|
||||
// Close — закрывает подключение к PostgreSQL. Вызывается при graceful shutdown.
|
||||
func Close() {
|
||||
if db != nil {
|
||||
db.Close()
|
||||
db = nil
|
||||
log.Info("billing: PostgreSQL connection closed")
|
||||
}
|
||||
}
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"shared-sqs/app/billing"
|
||||
"shared-sqs/app/conf"
|
||||
"shared-sqs/app/gosqs"
|
||||
"shared-sqs/app/models"
|
||||
@@ -124,6 +125,10 @@ func main() {
|
||||
if os.Getenv("SHARED_SQS_SEED_DEMO") == "true" {
|
||||
seedDemoData(tenantStore)
|
||||
}
|
||||
|
||||
// Billing: подключение к PostgreSQL для учёта использования.
|
||||
// Если BILLING_PG_HOST не задан — billing отключён, SQS работает без него.
|
||||
billing.Init()
|
||||
// Роутер с tenant auth и admin API
|
||||
r := router.New(tenantStore, adminToken)
|
||||
|
||||
@@ -163,6 +168,9 @@ func main() {
|
||||
// Остановить PeriodicTasks
|
||||
close(quit)
|
||||
|
||||
// Закрыть billing (если был подключён)
|
||||
billing.Close()
|
||||
|
||||
// Дать 10 секунд на завершение текущих HTTP запросов
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
|
||||
"shared-sqs/app/admin"
|
||||
"shared-sqs/app/auth"
|
||||
"shared-sqs/app/billing"
|
||||
sqs "shared-sqs/app/gosqs"
|
||||
"shared-sqs/app/interfaces"
|
||||
"shared-sqs/app/tenant"
|
||||
@@ -131,6 +132,16 @@ func actionHandler(w http.ResponseWriter, req *http.Request) {
|
||||
if ok {
|
||||
statusCode, responseBody := jsonFn(req)
|
||||
encodeResponse(w, req, statusCode, responseBody)
|
||||
// Billing: записываем каждую успешную SQS-операцию (async, no-op если billing выключен)
|
||||
if statusCode < 400 {
|
||||
if t, _ := req.Context().Value(auth.TenantContextKey).(*tenant.Tenant); t != nil {
|
||||
msgBytes := req.ContentLength
|
||||
if msgBytes < 0 {
|
||||
msgBytes = 0
|
||||
}
|
||||
billing.RecordUsage(t.ID, action, 1, msgBytes)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
log.Warnf("Bad Request - Action: %s", action)
|
||||
|
||||
Reference in New Issue
Block a user