Repinoid
c5b2c11f5d
- models.Queue: deadIdx/receiptIdx/uuidIdx/liveCount/deadCount + примитивы (AppendMessage/RemoveIndex/Compact/CompactIfNeeded/Reset/EnsureIndexes/ IndexOfReceipt/IndexOfUUID/MarkInFlight/ClearInFlight/InFlightIndexes/LiveCount). Удаление — tombstone за O(1), физическая чистка амортизированная. - DeleteMessage/ChangeMessageVisibility(±Batch): поиск по индексу вместо скана среза; удаление без сдвига остатка. - DeleteMessageBatch: tombstone вместо полной пересборки среза. - processQueueTick: visibility/DLQ по индексу in-flight (O(in-flight)), retention — tombstone, в конце — CompactIfNeeded. - send-пути: AppendMessage + LiveCount (OOM-лимит по живым). - admin/metrics/persistence: учёт живых сообщений, индексы после загрузки из Redis. - Redis-схема не менялась (сообщения и так в per-message HASH).
70 lines
2.4 KiB
Go
70 lines
2.4 KiB
Go
// app/metrics/gauge_updater.go
|
|
// Периодическое обновление gauge-метрик (количество очередей и сообщений per tenant).
|
|
// Запускается как горутина из main, обновляет каждые 15 секунд.
|
|
// Не блокирует SQS-запросы — читает через RLock.
|
|
// Created: 2026-04-12
|
|
package metrics
|
|
|
|
import (
|
|
"strings"
|
|
"time"
|
|
|
|
"shared-sqs/app/models"
|
|
|
|
log "github.com/sirupsen/logrus"
|
|
)
|
|
|
|
// StartGaugeUpdater — запускает горутину, которая каждые interval секунд
|
|
// пересчитывает количество очередей и сообщений по тенантам для Prometheus gauge.
|
|
// Останавливается при закрытии канала quit.
|
|
func StartGaugeUpdater(interval time.Duration, quit <-chan bool) {
|
|
go func() {
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
updateGauges()
|
|
case <-quit:
|
|
log.Debug("metrics: gauge updater stopped")
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
log.Infof("metrics: gauge updater started, interval=%s", interval)
|
|
}
|
|
|
|
// updateGauges — пересчитывает sqs_queues_count и sqs_messages_count per tenant.
|
|
// Формат ключа очереди: "{accessKey}:{queueName}" — берём часть до первого ":".
|
|
func updateGauges() {
|
|
// Снимок указателей под коротким RLock, затем содержимое — под per-queue RLock
|
|
models.SyncQueues.RLock()
|
|
snapshot := make(map[string]*models.Queue, len(models.SyncQueues.Queues))
|
|
for key, queue := range models.SyncQueues.Queues {
|
|
snapshot[key] = queue
|
|
}
|
|
models.SyncQueues.RUnlock()
|
|
|
|
tenantQueues := make(map[string]int)
|
|
tenantMessages := make(map[string]int)
|
|
for key, queue := range snapshot {
|
|
// Ключ формата "accessKey:queueName" — tenant определяется по accessKey
|
|
parts := strings.SplitN(key, ":", 2)
|
|
tenantKey := parts[0]
|
|
tenantQueues[tenantKey]++
|
|
queue.RLockMu()
|
|
tenantMessages[tenantKey] += queue.LiveCount()
|
|
queue.RUnlockMu()
|
|
}
|
|
|
|
// Сбрасываем старые значения и записываем новые
|
|
QueuesGauge.Reset()
|
|
MessagesGauge.Reset()
|
|
for tenant, count := range tenantQueues {
|
|
QueuesGauge.WithLabelValues(tenant).Set(float64(count))
|
|
}
|
|
for tenant, count := range tenantMessages {
|
|
MessagesGauge.WithLabelValues(tenant).Set(float64(count))
|
|
}
|
|
}
|