P1.4: tenant-индекс очередей (O(1) подсчёт/список вместо скана)
Введён тип QueueStore с индексом accessKey→{queueName}. Мутации очередей идут через
AddQueue/RemoveQueue (create/delete/restore/config/seed/admin), подсчёт и список —
через CountTenantQueues/TenantQueueNames. Устранены O(total) скан-по-префиксу в
countTenantQueues, deleteTenant, listTenantQueues, createTenantQueue. API SyncQueues
(Queues, Lock) сохранён — остальной код не затронут.
This commit is contained in:
+17
-21
@@ -405,15 +405,14 @@ func (h *Handler) deleteTenant(w http.ResponseWriter, r *http.Request) {
|
|||||||
json.NewEncoder(w).Encode(map[string]string{"error": "tenant not found"})
|
json.NewEncoder(w).Encode(map[string]string{"error": "tenant not found"})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// Удаляем все очереди тенанта из SyncQueues
|
// Удаляем все очереди тенанта из SyncQueues (через tenant-индекс, без скана всех очередей)
|
||||||
prefix := t.AccessKey + ":"
|
|
||||||
deletedQueueKeys := make([]string, 0)
|
|
||||||
models.SyncQueues.Lock()
|
models.SyncQueues.Lock()
|
||||||
for key := range models.SyncQueues.Queues {
|
tenantQueueNames := models.SyncQueues.TenantQueueNames(t.AccessKey)
|
||||||
if strings.HasPrefix(key, prefix) {
|
deletedQueueKeys := make([]string, 0, len(tenantQueueNames))
|
||||||
|
for _, name := range tenantQueueNames {
|
||||||
|
key := t.AccessKey + ":" + name
|
||||||
deletedQueueKeys = append(deletedQueueKeys, key)
|
deletedQueueKeys = append(deletedQueueKeys, key)
|
||||||
delete(models.SyncQueues.Queues, key)
|
models.SyncQueues.RemoveQueue(key)
|
||||||
}
|
|
||||||
}
|
}
|
||||||
models.SyncQueues.Unlock()
|
models.SyncQueues.Unlock()
|
||||||
for _, queueKey := range deletedQueueKeys {
|
for _, queueKey := range deletedQueueKeys {
|
||||||
@@ -447,9 +446,11 @@ func (h *Handler) listTenantQueues(w http.ResponseWriter, r *http.Request) {
|
|||||||
}
|
}
|
||||||
queues := make([]queueInfo, 0)
|
queues := make([]queueInfo, 0)
|
||||||
models.SyncQueues.RLock()
|
models.SyncQueues.RLock()
|
||||||
for key, q := range models.SyncQueues.Queues {
|
for _, name := range models.SyncQueues.TenantQueueNames(t.AccessKey) {
|
||||||
if strings.HasPrefix(key, prefix) {
|
q := models.SyncQueues.Queues[prefix+name]
|
||||||
name := strings.TrimPrefix(key, prefix)
|
if q == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
visible := 0
|
visible := 0
|
||||||
notVisible := 0
|
notVisible := 0
|
||||||
for _, msg := range q.Messages {
|
for _, msg := range q.Messages {
|
||||||
@@ -468,7 +469,6 @@ func (h *Handler) listTenantQueues(w http.ResponseWriter, r *http.Request) {
|
|||||||
RetentionPeriod: q.MessageRetentionPeriod,
|
RetentionPeriod: q.MessageRetentionPeriod,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
|
||||||
models.SyncQueues.RUnlock()
|
models.SyncQueues.RUnlock()
|
||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
json.NewEncoder(w).Encode(queues)
|
json.NewEncoder(w).Encode(queues)
|
||||||
@@ -504,19 +504,14 @@ func (h *Handler) createTenantQueue(w http.ResponseWriter, r *http.Request) {
|
|||||||
jsonErr(w, http.StatusConflict, "queue already exists")
|
jsonErr(w, http.StatusConflict, "queue already exists")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// Проверяем лимит очередей тенанта
|
// Проверяем лимит очередей тенанта (через tenant-индекс, O(1))
|
||||||
count := 0
|
count := models.SyncQueues.CountTenantQueues(t.AccessKey)
|
||||||
for k := range models.SyncQueues.Queues {
|
|
||||||
if strings.HasPrefix(k, t.AccessKey+":") {
|
|
||||||
count++
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if t.MaxQueues > 0 && count >= t.MaxQueues {
|
if t.MaxQueues > 0 && count >= t.MaxQueues {
|
||||||
models.SyncQueues.Unlock()
|
models.SyncQueues.Unlock()
|
||||||
jsonErr(w, http.StatusForbidden, "queue limit exceeded")
|
jsonErr(w, http.StatusForbidden, "queue limit exceeded")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
models.SyncQueues.Queues[key] = &models.Queue{
|
newQueue := &models.Queue{
|
||||||
Name: req.Name,
|
Name: req.Name,
|
||||||
VisibilityTimeout: 30,
|
VisibilityTimeout: 30,
|
||||||
MaximumMessageSize: 262144,
|
MaximumMessageSize: 262144,
|
||||||
@@ -524,7 +519,8 @@ func (h *Handler) createTenantQueue(w http.ResponseWriter, r *http.Request) {
|
|||||||
Messages: []models.SqsMessage{},
|
Messages: []models.SqsMessage{},
|
||||||
Duplicates: make(map[string]time.Time),
|
Duplicates: make(map[string]time.Time),
|
||||||
}
|
}
|
||||||
persistence.SaveQueue(key, models.SyncQueues.Queues[key])
|
models.SyncQueues.AddQueue(key, newQueue)
|
||||||
|
persistence.SaveQueue(key, newQueue)
|
||||||
models.SyncQueues.Unlock()
|
models.SyncQueues.Unlock()
|
||||||
log.Infof("admin: created queue %s for tenant %s", req.Name, t.ID)
|
log.Infof("admin: created queue %s for tenant %s", req.Name, t.ID)
|
||||||
w.Header().Set("Content-Type", "application/json")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
@@ -549,7 +545,7 @@ func (h *Handler) deleteTenantQueue(w http.ResponseWriter, r *http.Request) {
|
|||||||
jsonErr(w, http.StatusNotFound, "queue not found")
|
jsonErr(w, http.StatusNotFound, "queue not found")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
delete(models.SyncQueues.Queues, key)
|
models.SyncQueues.RemoveQueue(key)
|
||||||
models.SyncQueues.Unlock()
|
models.SyncQueues.Unlock()
|
||||||
persistence.DeleteQueue(key)
|
persistence.DeleteQueue(key)
|
||||||
log.Infof("admin: deleted queue %s for tenant %s", queueName, t.ID)
|
log.Infof("admin: deleted queue %s for tenant %s", queueName, t.ID)
|
||||||
|
|||||||
+1
-1
@@ -116,7 +116,7 @@ func main() {
|
|||||||
} else {
|
} else {
|
||||||
models.SyncQueues.Lock()
|
models.SyncQueues.Lock()
|
||||||
for k, q := range queues {
|
for k, q := range queues {
|
||||||
models.SyncQueues.Queues[k] = q
|
models.SyncQueues.AddQueue(k, q)
|
||||||
}
|
}
|
||||||
models.SyncQueues.Unlock()
|
models.SyncQueues.Unlock()
|
||||||
}
|
}
|
||||||
|
|||||||
+2
-2
@@ -104,7 +104,7 @@ func seedDemoData(store *tenant.TenantStore) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
models.SyncQueues.Lock()
|
models.SyncQueues.Lock()
|
||||||
models.SyncQueues.Queues[key] = &models.Queue{
|
models.SyncQueues.AddQueue(key, &models.Queue{
|
||||||
Name: q.name,
|
Name: q.name,
|
||||||
URL: queueURL,
|
URL: queueURL,
|
||||||
Arn: queueARN,
|
Arn: queueARN,
|
||||||
@@ -113,7 +113,7 @@ func seedDemoData(store *tenant.TenantStore) {
|
|||||||
MessageRetentionPeriod: 345600,
|
MessageRetentionPeriod: 345600,
|
||||||
Messages: msgs,
|
Messages: msgs,
|
||||||
Duplicates: make(map[string]time.Time),
|
Duplicates: make(map[string]time.Time),
|
||||||
}
|
})
|
||||||
models.SyncQueues.Unlock()
|
models.SyncQueues.Unlock()
|
||||||
|
|
||||||
log.Infof("seed: очередь %s (%d сообщений)", q.name, len(q.msgs))
|
log.Infof("seed: очередь %s (%d сообщений)", q.name, len(q.msgs))
|
||||||
|
|||||||
+2
-2
@@ -105,7 +105,7 @@ func LoadYamlConfig(filename string, env string) []string {
|
|||||||
queue.MessageRetentionPeriod = models.CurrentEnvironment.QueueAttributeDefaults.MessageRetentionPeriod
|
queue.MessageRetentionPeriod = models.CurrentEnvironment.QueueAttributeDefaults.MessageRetentionPeriod
|
||||||
}
|
}
|
||||||
|
|
||||||
models.SyncQueues.Queues[queue.Name] = &models.Queue{
|
models.SyncQueues.AddQueue(queue.Name, &models.Queue{
|
||||||
Name: queue.Name,
|
Name: queue.Name,
|
||||||
VisibilityTimeout: queue.VisibilityTimeout,
|
VisibilityTimeout: queue.VisibilityTimeout,
|
||||||
Arn: queueArn,
|
Arn: queueArn,
|
||||||
@@ -116,7 +116,7 @@ func LoadYamlConfig(filename string, env string) []string {
|
|||||||
IsFIFO: utils.HasFIFOQueueName(queue.Name),
|
IsFIFO: utils.HasFIFOQueueName(queue.Name),
|
||||||
EnableDuplicates: models.CurrentEnvironment.EnableDuplicates,
|
EnableDuplicates: models.CurrentEnvironment.EnableDuplicates,
|
||||||
Duplicates: make(map[string]time.Time),
|
Duplicates: make(map[string]time.Time),
|
||||||
}
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// Второй проход — устанавливаем RedrivePolicy, чтобы DLQ были доступны независимо от порядка
|
// Второй проход — устанавливаем RedrivePolicy, чтобы DLQ были доступны независимо от порядка
|
||||||
|
|||||||
@@ -75,7 +75,7 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
|||||||
models.SyncQueues.Unlock()
|
models.SyncQueues.Unlock()
|
||||||
return utils.CreateErrorResponseV1(err.Error(), true)
|
return utils.CreateErrorResponseV1(err.Error(), true)
|
||||||
}
|
}
|
||||||
models.SyncQueues.Queues[key] = queue
|
models.SyncQueues.AddQueue(key, queue)
|
||||||
}
|
}
|
||||||
// Сохраняем очередь в Redis пока держим Lock — консистентный снапшот
|
// Сохраняем очередь в Redis пока держим Lock — консистентный снапшот
|
||||||
persistence.SaveQueue(key, models.SyncQueues.Queues[key])
|
persistence.SaveQueue(key, models.SyncQueues.Queues[key])
|
||||||
|
|||||||
@@ -50,7 +50,7 @@ func DeleteQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
|||||||
log.Warnf("Delete Queue: %s does not exist (tenant: %s)", queueName, t.ID)
|
log.Warnf("Delete Queue: %s does not exist (tenant: %s)", queueName, t.ID)
|
||||||
return utils.CreateErrorResponseV1("QueueNotFound", true)
|
return utils.CreateErrorResponseV1("QueueNotFound", true)
|
||||||
}
|
}
|
||||||
delete(models.SyncQueues.Queues, key)
|
models.SyncQueues.RemoveQueue(key)
|
||||||
|
|
||||||
// Удаляем из Redis асинхронно
|
// Удаляем из Redis асинхронно
|
||||||
persistence.DeleteQueue(key)
|
persistence.DeleteQueue(key)
|
||||||
|
|||||||
@@ -41,16 +41,9 @@ func tenantQueueARN(t *tenant.Tenant, queueName string) string {
|
|||||||
return "arn:aws:sqs:" + models.CurrentEnvironment.Region + ":" + t.ID + ":" + queueName
|
return "arn:aws:sqs:" + models.CurrentEnvironment.Region + ":" + t.ID + ":" + queueName
|
||||||
}
|
}
|
||||||
|
|
||||||
// countTenantQueues — считает количество очередей тенанта в SyncQueues.
|
// countTenantQueues — считает количество очередей тенанта (через tenant-индекс, O(1)).
|
||||||
// Используется для проверки лимита MaxQueues.
|
// Используется для проверки лимита MaxQueues.
|
||||||
// Вызывать под SyncQueues.RLock().
|
// Вызывать под SyncQueues.RLock()/Lock().
|
||||||
func countTenantQueues(tenantAccessKey string) int {
|
func countTenantQueues(tenantAccessKey string) int {
|
||||||
prefix := tenantAccessKey + ":"
|
return models.SyncQueues.CountTenantQueues(tenantAccessKey)
|
||||||
count := 0
|
|
||||||
for key := range models.SyncQueues.Queues {
|
|
||||||
if len(key) > len(prefix) && key[:len(prefix)] == prefix {
|
|
||||||
count++
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return count
|
|
||||||
}
|
}
|
||||||
|
|||||||
+77
-2
@@ -1,6 +1,7 @@
|
|||||||
package models
|
package models
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -17,7 +18,81 @@ var CurrentEnvironment = Environment{
|
|||||||
var LogMessages bool
|
var LogMessages bool
|
||||||
var LogFile string
|
var LogFile string
|
||||||
|
|
||||||
var SyncQueues = struct {
|
// QueueStore — хранилище очередей: блокировка + map ключ→очередь + индекс tenant→очереди.
|
||||||
|
//
|
||||||
|
// Ключ очереди в Queues — "{accessKey}:{queueName}" (tenant-scoped изоляция).
|
||||||
|
// Индекс tenantQueues (accessKey → набор имён очередей) позволяет считать и
|
||||||
|
// перечислять очереди тенанта за O(число его очередей), без скана всех очередей сервиса.
|
||||||
|
//
|
||||||
|
// ВАЖНО: методы AddQueue/RemoveQueue/CountTenantQueues/TenantQueueNames НЕ берут
|
||||||
|
// блокировку — вызывать под уже захваченным SyncQueues.Lock()/RLock()
|
||||||
|
// (соответствует существующей дисциплине блокировок в коде).
|
||||||
|
type QueueStore struct {
|
||||||
sync.RWMutex
|
sync.RWMutex
|
||||||
Queues map[string]*Queue
|
Queues map[string]*Queue
|
||||||
}{Queues: make(map[string]*Queue)}
|
tenantQueues map[string]map[string]struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
var SyncQueues = QueueStore{
|
||||||
|
Queues: make(map[string]*Queue),
|
||||||
|
tenantQueues: make(map[string]map[string]struct{}),
|
||||||
|
}
|
||||||
|
|
||||||
|
// splitQueueKey — разбирает tenant-scoped ключ "{accessKey}:{queueName}".
|
||||||
|
// Легаси-ключи без ":" (yaml-очереди) tenant-независимы → ok=false.
|
||||||
|
func splitQueueKey(key string) (accessKey, queueName string, ok bool) {
|
||||||
|
idx := strings.Index(key, ":")
|
||||||
|
if idx <= 0 || idx >= len(key)-1 {
|
||||||
|
return "", "", false
|
||||||
|
}
|
||||||
|
return key[:idx], key[idx+1:], true
|
||||||
|
}
|
||||||
|
|
||||||
|
// AddQueue — добавляет очередь и обновляет tenant-индекс.
|
||||||
|
// Вызывать под SyncQueues.Lock().
|
||||||
|
func (s *QueueStore) AddQueue(key string, q *Queue) {
|
||||||
|
s.Queues[key] = q
|
||||||
|
if ak, name, ok := splitQueueKey(key); ok {
|
||||||
|
if s.tenantQueues[ak] == nil {
|
||||||
|
s.tenantQueues[ak] = make(map[string]struct{})
|
||||||
|
}
|
||||||
|
s.tenantQueues[ak][name] = struct{}{}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// RemoveQueue — удаляет очередь и обновляет tenant-индекс.
|
||||||
|
// Вызывать под SyncQueues.Lock().
|
||||||
|
func (s *QueueStore) RemoveQueue(key string) {
|
||||||
|
delete(s.Queues, key)
|
||||||
|
if ak, name, ok := splitQueueKey(key); ok {
|
||||||
|
if set, exists := s.tenantQueues[ak]; exists {
|
||||||
|
delete(set, name)
|
||||||
|
if len(set) == 0 {
|
||||||
|
delete(s.tenantQueues, ak)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// CountTenantQueues — число очередей тенанта (по accessKey) за O(1).
|
||||||
|
// Вызывать под SyncQueues.RLock() или Lock().
|
||||||
|
func (s *QueueStore) CountTenantQueues(accessKey string) int {
|
||||||
|
return len(s.tenantQueues[accessKey])
|
||||||
|
}
|
||||||
|
|
||||||
|
// TenantQueueNames — копия списка имён очередей тенанта.
|
||||||
|
// Вызывать под SyncQueues.RLock() или Lock().
|
||||||
|
func (s *QueueStore) TenantQueueNames(accessKey string) []string {
|
||||||
|
set := s.tenantQueues[accessKey]
|
||||||
|
names := make([]string, 0, len(set))
|
||||||
|
for n := range set {
|
||||||
|
names = append(names, n)
|
||||||
|
}
|
||||||
|
return names
|
||||||
|
}
|
||||||
|
|
||||||
|
// ResetQueueIndex — очищает tenant-индекс (для ResetResources/тестов).
|
||||||
|
// Вызывать под SyncQueues.Lock().
|
||||||
|
func (s *QueueStore) ResetQueueIndex() {
|
||||||
|
s.tenantQueues = make(map[string]map[string]struct{})
|
||||||
|
}
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ func ResetApp() {
|
|||||||
func ResetResources() {
|
func ResetResources() {
|
||||||
SyncQueues.Lock()
|
SyncQueues.Lock()
|
||||||
SyncQueues.Queues = make(map[string]*Queue)
|
SyncQueues.Queues = make(map[string]*Queue)
|
||||||
|
SyncQueues.ResetQueueIndex()
|
||||||
SyncQueues.Unlock()
|
SyncQueues.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user