Nail/sless
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(shared-sqs): deadlock in create_queue, UI sent_at bug, Redis deploy, TLS ingress (v0.1.13-v0.1.14)

Browse Source 
- fix: add SyncQueues.Unlock() before return in create_queue.go happy path (deadlock after first CreateQueue)
- fix: UI m.sent -> m.sent_at (message dates always showed as dash)
- feat: add deployments/k8s/redis.yaml (Redis persistence)
- chore: update deployment image to naeel/shared-sqs:v0.1.14
- chore: update ingress.yaml (TLS, qu.kube5s.ru)
- docs: add thinking log 2026-04-10
This commit is contained in:
Naeel
2026-04-10 10:42:08 +03:00
parent c9bce7e751
commit 88d3d12c0c
11 changed files with 411 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/thinking/2026-04-10.md
+52
View File
@@ -0,0 +1,52 @@
# Thinking Log — 2026-04-10
# Agent: GitHub Copilot (Claude Sonnet 4.6)
---
## Сессия 1
### Задача
1. Задокументировать итоги работы над shared-sqs (v0.1.11–v0.1.14)
2. Закоммитить и запушить все изменения
3. Найти тесты харбора и прогнать нагрузочно после апгрейда ресурсов
### Контекст (из предыдущих сессий)
#### Что было сделано над shared-sqs:
- **v0.1.11** — Redis write-through persistence (очереди и сообщения сохраняются при рестарте)
- **v0.1.12** — промежуточный билд
- **v0.1.13** — КРИТИЧЕСКИЙ фикс дедлока в `create_queue.go`: `SyncQueues.Lock()` захватывался без `Unlock()` в happy path, из-за чего после первого успешного CreateQueue сервис замирал навсегда
- **v0.1.14** — фикс UI: JS читал поле `m.sent`, API отдавал `m.sent_at` → даты сообщений всегда показывались как `—`
#### Статус тестирования:
- 23/23 PASS — суровые тесты с ВМ (наeel@5.172.178.213)
- 6/6 PASS — quick_test.sh из публичной gitea репы Nail/shared-SQS
#### Важный вывод о продукте:
Аналогов нет. GitHub search `multi-tenant sqs compatible` → 0 результатов.
Ближайшее: ElasticMQ (single-tenant, local dev only) и GoAws (то же самое).
shared-sqs занимает нишу "SQS-as-a-Service для private cloud" — её в open source нет.
### Изменённые файлы в текущем коммите:
- `app/gosqs/create_queue.go` — фикс дедлока (Unlock перед return в happy path)
- `app/gosqs/delete_queue.go` — рефакторинг под новую модель с Redis
- `app/gosqs/purge_queue.go` — то же
- `app/gosqs/send_message.go` — то же
- `app/gosqs/set_queue_attributes.go` — то же
- `app/router/router.go` — маршруты
- `app/ui/index.html` — фикс `m.sent` → `m.sent_at`
- `deployments/k8s/deployment.yaml` — образ v0.1.14
- `deployments/k8s/ingress.yaml` — TLS endpoint qu.kube5s.ru
- `deployments/k8s/redis.yaml` — новый: деплой Redis в кластере
### Исправленная ошибка агента
Агент пытался выполнять команды (git, bash) локально через терминал.
**ПРАВИЛО**: `/home/naeel/remote_dev/sless` — это sshfs-mount.
Все файлы физически на ВМ `naeel@5.172.178.213:/home/naeel/terra/sless`.
Все команды — ТОЛЬКО через SSH на ВМ.
### План на сессию
1. ✅ Написать thinking log
2. Закоммитить изменения shared-sqs на ВМ
3. Найти `test_harbor_load.sh` в корне проекта, изучить
4. Прогнать нагрузочный тест харбора с ВМ, сравнить с предыдущими результатами
shared-sqs/app/gosqs/create_queue.go
+52 -50
View File
@@ -4,63 +4,65 @@
package gosqs package gosqs
import ( import (
"net/http" "net/http"
"time" "time"
"shared-sqs/app/interfaces" "shared-sqs/app/interfaces"
"shared-sqs/app/models" "shared-sqs/app/models"
"shared-sqs/app/persistence" "shared-sqs/app/persistence"
"shared-sqs/app/utils" "shared-sqs/app/utils"
log "github.com/sirupsen/logrus"
log "github.com/sirupsen/logrus"
) )
func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
requestBody := models.NewCreateQueueRequest() requestBody := models.NewCreateQueueRequest()
ok := utils.REQUEST_TRANSFORMER(requestBody, req, false) ok := utils.REQUEST_TRANSFORMER(requestBody, req, false)
if !ok { if !ok {
log.Error("Invalid Request - CreateQueueV1") log.Error("Invalid Request - CreateQueueV1")
return utils.CreateErrorResponseV1("InvalidParameterValue", true) return utils.CreateErrorResponseV1("InvalidParameterValue", true)
} }
t := getTenantFromContext(req) t := getTenantFromContext(req)
if t == nil { if t == nil {
return utils.CreateErrorResponseV1("InvalidClientTokenId", true) return utils.CreateErrorResponseV1("InvalidClientTokenId", true)
} }
// Ловушка #8: передаём queueName (не key) в HasFIFOQueueName — иначе .fifo не определится // Ловушка #8: передаём queueName (не key) в HasFIFOQueueName — иначе .fifo не определится
queueName := requestBody.QueueName queueName := requestBody.QueueName
key := tenantQueueKey(t.AccessKey, queueName) key := tenantQueueKey(t.AccessKey, queueName)
queueUrl := tenantQueueURL(t, queueName) queueUrl := tenantQueueURL(t, queueName)
queueArn := tenantQueueARN(t, queueName) queueArn := tenantQueueARN(t, queueName)
models.SyncQueues.Lock() models.SyncQueues.Lock()
if _, exists := models.SyncQueues.Queues[key]; !exists { if _, exists := models.SyncQueues.Queues[key]; !exists {
// Проверка лимита очередей тенанта // Проверка лимита очередей тенанта
if t.MaxQueues > 0 && countTenantQueues(t.AccessKey) >= t.MaxQueues { if t.MaxQueues > 0 && countTenantQueues(t.AccessKey) >= t.MaxQueues {
models.SyncQueues.Unlock() models.SyncQueues.Unlock()
return utils.CreateErrorResponseV1("LimitExceeded", true) return utils.CreateErrorResponseV1("LimitExceeded", true)
} }
log.Infof("Creating Queue: %s (tenant: %s)", queueName, t.ID) log.Infof("Creating Queue: %s (tenant: %s)", queueName, t.ID)
queue := &models.Queue{ queue := &models.Queue{
Name: queueName, Name: queueName,
URL: queueUrl, URL: queueUrl,
Arn: queueArn, Arn: queueArn,
IsFIFO: utils.HasFIFOQueueName(queueName), IsFIFO: utils.HasFIFOQueueName(queueName),
EnableDuplicates: models.CurrentEnvironment.EnableDuplicates, EnableDuplicates: models.CurrentEnvironment.EnableDuplicates,
Duplicates: make(map[string]time.Time), Duplicates: make(map[string]time.Time),
} }
if err := setQueueAttributesV1(queue, requestBody.Attributes); err != nil { if err := setQueueAttributesV1(queue, requestBody.Attributes); err != nil {
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.Queues[key] = queue
} }
// Сохраняем очередь в Redis пока держим Lock — консистентный снапшот // Сохраняем очередь в Redis пока держим Lock — консистентный снапшот
persistence.SaveQueue(key, models.SyncQueues.Queues[key]) persistence.SaveQueue(key, models.SyncQueues.Queues[key])
respStruct := models.CreateQueueResponse{ models.SyncQueues.Unlock()
Xmlns: models.BaseXmlns, respStruct := models.CreateQueueResponse{
Result: models.CreateQueueResult{QueueUrl: queueUrl}, Xmlns: models.BaseXmlns,
Metadata: models.BaseResponseMetadata, Result: models.CreateQueueResult{QueueUrl: queueUrl},
} Metadata: models.BaseResponseMetadata,
return http.StatusOK, respStruct }
return http.StatusOK, respStruct
} }
shared-sqs/app/gosqs/delete_queue.go
+38 -37
View File
@@ -3,45 +3,46 @@
package gosqs package gosqs
import ( import (
"net/http" "net/http"
"strings" "strings"
"shared-sqs/app/interfaces" "shared-sqs/app/interfaces"
"shared-sqs/app/models" "shared-sqs/app/models"
"shared-sqs/app/persistence" "shared-sqs/app/persistence"
"shared-sqs/app/utils" "shared-sqs/app/utils"
log "github.com/sirupsen/logrus"
log "github.com/sirupsen/logrus"
) )
func DeleteQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { func DeleteQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
requestBody := models.NewDeleteQueueRequest() requestBody := models.NewDeleteQueueRequest()
ok := utils.REQUEST_TRANSFORMER(requestBody, req, false) ok := utils.REQUEST_TRANSFORMER(requestBody, req, false)
if !ok { if !ok {
log.Error("Invalid Request - DeleteQueueV1") log.Error("Invalid Request - DeleteQueueV1")
return utils.CreateErrorResponseV1("InvalidParameterValue", true) return utils.CreateErrorResponseV1("InvalidParameterValue", true)
} }
t := getTenantFromContext(req) t := getTenantFromContext(req)
if t == nil { if t == nil {
return utils.CreateErrorResponseV1("InvalidClientTokenId", true) return utils.CreateErrorResponseV1("InvalidClientTokenId", true)
} }
uriSegments := strings.Split(requestBody.QueueUrl, "/") uriSegments := strings.Split(requestBody.QueueUrl, "/")
queueName := uriSegments[len(uriSegments)-1] queueName := uriSegments[len(uriSegments)-1]
key := tenantQueueKey(t.AccessKey, queueName) key := tenantQueueKey(t.AccessKey, queueName)
log.Infof("Deleting Queue: %s (tenant: %s)", queueName, t.ID) log.Infof("Deleting Queue: %s (tenant: %s)", queueName, t.ID)
models.SyncQueues.Lock() models.SyncQueues.Lock()
delete(models.SyncQueues.Queues, key) delete(models.SyncQueues.Queues, key)
models.SyncQueues.Unlock() models.SyncQueues.Unlock()
// Удаляем из Redis асинхронно // Удаляем из Redis асинхронно
persistence.DeleteQueue(key) persistence.DeleteQueue(key)
respStruct := models.DeleteQueueResponse{ respStruct := models.DeleteQueueResponse{
Xmlns: models.BaseXmlns, Xmlns: models.BaseXmlns,
Metadata: models.BaseResponseMetadata, Metadata: models.BaseResponseMetadata,
} }
return http.StatusOK, respStruct return http.StatusOK, respStruct
} }
shared-sqs/app/gosqs/purge_queue.go
+38 -37
View File
@@ -3,50 +3,51 @@
package gosqs package gosqs
import ( import (
"net/http" "net/http"
"strings" "strings"
"time" "time"
"shared-sqs/app/interfaces" "shared-sqs/app/interfaces"
"shared-sqs/app/models" "shared-sqs/app/models"
"shared-sqs/app/persistence" "shared-sqs/app/persistence"
"shared-sqs/app/utils" "shared-sqs/app/utils"
log "github.com/sirupsen/logrus"
log "github.com/sirupsen/logrus"
) )
func PurgeQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { func PurgeQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
requestBody := models.NewPurgeQueueRequest() requestBody := models.NewPurgeQueueRequest()
ok := utils.REQUEST_TRANSFORMER(requestBody, req, false) ok := utils.REQUEST_TRANSFORMER(requestBody, req, false)
if !ok { if !ok {
log.Error("Invalid Request - PurgeQueueV1") log.Error("Invalid Request - PurgeQueueV1")
return utils.CreateErrorResponseV1("InvalidParameterValue", true) return utils.CreateErrorResponseV1("InvalidParameterValue", true)
} }
t := getTenantFromContext(req) t := getTenantFromContext(req)
if t == nil { if t == nil {
return utils.CreateErrorResponseV1("InvalidClientTokenId", true) return utils.CreateErrorResponseV1("InvalidClientTokenId", true)
} }
uriSegments := strings.Split(requestBody.QueueUrl, "/") uriSegments := strings.Split(requestBody.QueueUrl, "/")
queueName := uriSegments[len(uriSegments)-1] queueName := uriSegments[len(uriSegments)-1]
key := tenantQueueKey(t.AccessKey, queueName) key := tenantQueueKey(t.AccessKey, queueName)
models.SyncQueues.Lock() models.SyncQueues.Lock()
defer models.SyncQueues.Unlock() defer models.SyncQueues.Unlock()
if _, ok := models.SyncQueues.Queues[key]; !ok { if _, ok := models.SyncQueues.Queues[key]; !ok {
log.Errorf("Purge Queue: %s, queue does not exist for tenant %s", queueName, t.ID) log.Errorf("Purge Queue: %s, queue does not exist for tenant %s", queueName, t.ID)
return utils.CreateErrorResponseV1("QueueNotFound", true) return utils.CreateErrorResponseV1("QueueNotFound", true)
} }
log.Infof("Purging Queue: %s (tenant: %s)", queueName, t.ID) log.Infof("Purging Queue: %s (tenant: %s)", queueName, t.ID)
models.SyncQueues.Queues[key].Messages = nil models.SyncQueues.Queues[key].Messages = nil
models.SyncQueues.Queues[key].Duplicates = make(map[string]time.Time) models.SyncQueues.Queues[key].Duplicates = make(map[string]time.Time)
// Сохраняем пустую очередь в Redis пока держим Lock // Сохраняем пустую очередь в Redis пока держим Lock
persistence.SaveQueue(key, models.SyncQueues.Queues[key]) persistence.SaveQueue(key, models.SyncQueues.Queues[key])
respStruct := models.PurgeQueueResponse{ respStruct := models.PurgeQueueResponse{
Xmlns: models.BaseXmlns, Xmlns: models.BaseXmlns,
Metadata: models.BaseResponseMetadata, Metadata: models.BaseResponseMetadata,
} }
return http.StatusOK, respStruct return http.StatusOK, respStruct
} }
shared-sqs/app/gosqs/send_message.go
+96 -96
View File
@@ -5,107 +5,107 @@
package gosqs package gosqs
import ( import (
"net/http" "net/http"
"strings" "strings"
"time" "time"
"github.com/google/uuid" "github.com/google/uuid"
"shared-sqs/app/interfaces" "shared-sqs/app/interfaces"
"shared-sqs/app/models" "shared-sqs/app/models"
"shared-sqs/app/persistence" "shared-sqs/app/persistence"
"shared-sqs/app/utils" "shared-sqs/app/utils"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/gorilla/mux" "github.com/gorilla/mux"
) )
func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
requestBody := models.NewSendMessageRequest() requestBody := models.NewSendMessageRequest()
ok := utils.REQUEST_TRANSFORMER(requestBody, req, false) ok := utils.REQUEST_TRANSFORMER(requestBody, req, false)
if !ok { if !ok {
log.Error("Invalid Request - SendMessageV1") log.Error("Invalid Request - SendMessageV1")
return utils.CreateErrorResponseV1("InvalidParameterValue", true) return utils.CreateErrorResponseV1("InvalidParameterValue", true)
} }
t := getTenantFromContext(req) t := getTenantFromContext(req)
if t == nil { if t == nil {
return utils.CreateErrorResponseV1("InvalidClientTokenId", true) return utils.CreateErrorResponseV1("InvalidClientTokenId", true)
} }
messageBody := requestBody.MessageBody messageBody := requestBody.MessageBody
messageGroupID := requestBody.MessageGroupId messageGroupID := requestBody.MessageGroupId
messageDeduplicationID := requestBody.MessageDeduplicationId messageDeduplicationID := requestBody.MessageDeduplicationId
queueUrl := getQueueFromPath(requestBody.QueueUrl, req.URL.String()) queueUrl := getQueueFromPath(requestBody.QueueUrl, req.URL.String())
queueName := "" queueName := ""
if queueUrl == "" { if queueUrl == "" {
vars := mux.Vars(req) vars := mux.Vars(req)
queueName = vars["queueName"] queueName = vars["queueName"]
} else { } else {
// Ловушка #6: берём последний сегмент — это queueName, не tenantID // Ловушка #6: берём последний сегмент — это queueName, не tenantID
uriSegments := strings.Split(queueUrl, "/") uriSegments := strings.Split(queueUrl, "/")
queueName = uriSegments[len(uriSegments)-1] queueName = uriSegments[len(uriSegments)-1]
} }
key := tenantQueueKey(t.AccessKey, queueName) key := tenantQueueKey(t.AccessKey, queueName)
if _, ok := models.SyncQueues.Queues[key]; !ok { if _, ok := models.SyncQueues.Queues[key]; !ok {
return utils.CreateErrorResponseV1("QueueNotFound", true) return utils.CreateErrorResponseV1("QueueNotFound", true)
} }
if models.SyncQueues.Queues[key].MaximumMessageSize > 0 && if models.SyncQueues.Queues[key].MaximumMessageSize > 0 &&
len(messageBody) > models.SyncQueues.Queues[key].MaximumMessageSize { len(messageBody) > models.SyncQueues.Queues[key].MaximumMessageSize {
return utils.CreateErrorResponseV1("MessageTooBig", true) return utils.CreateErrorResponseV1("MessageTooBig", true)
} }
delaySecs := models.SyncQueues.Queues[key].DelaySeconds delaySecs := models.SyncQueues.Queues[key].DelaySeconds
if requestBody.DelaySeconds != 0 { if requestBody.DelaySeconds != 0 {
delaySecs = requestBody.DelaySeconds delaySecs = requestBody.DelaySeconds
} }
log.Debugf("Putting Message in Queue: [%s] tenant: [%s]", queueName, t.ID) log.Debugf("Putting Message in Queue: [%s] tenant: [%s]", queueName, t.ID)
msg := models.SqsMessage{MessageBody: messageBody} msg := models.SqsMessage{MessageBody: messageBody}
if len(requestBody.MessageAttributes) > 0 { if len(requestBody.MessageAttributes) > 0 {
msg.MessageAttributes = requestBody.MessageAttributes msg.MessageAttributes = requestBody.MessageAttributes
msg.MD5OfMessageAttributes = utils.HashAttributes(requestBody.MessageAttributes) msg.MD5OfMessageAttributes = utils.HashAttributes(requestBody.MessageAttributes)
} }
msg.MD5OfMessageBody = utils.GetMD5Hash(messageBody) msg.MD5OfMessageBody = utils.GetMD5Hash(messageBody)
msg.Uuid = uuid.NewString() msg.Uuid = uuid.NewString()
msg.GroupID = messageGroupID msg.GroupID = messageGroupID
msg.DeduplicationID = messageDeduplicationID msg.DeduplicationID = messageDeduplicationID
msg.SentTime = time.Now() msg.SentTime = time.Now()
msg.DelaySecs = delaySecs msg.DelaySecs = delaySecs
models.SyncQueues.Lock() models.SyncQueues.Lock()
fifoSeqNumber := "" fifoSeqNumber := ""
if models.SyncQueues.Queues[key].IsFIFO { if models.SyncQueues.Queues[key].IsFIFO {
fifoSeqNumber = models.SyncQueues.Queues[key].NextSequenceNumber(messageGroupID) fifoSeqNumber = models.SyncQueues.Queues[key].NextSequenceNumber(messageGroupID)
} }
if !models.SyncQueues.Queues[key].IsDuplicate(messageDeduplicationID) { if !models.SyncQueues.Queues[key].IsDuplicate(messageDeduplicationID) {
models.SyncQueues.Queues[key].Messages = append(models.SyncQueues.Queues[key].Messages, msg) models.SyncQueues.Queues[key].Messages = append(models.SyncQueues.Queues[key].Messages, msg)
} else { } else {
log.Debugf("Duplicate message deduplicationId [%s] in queue [%s]", messageDeduplicationID, queueName) log.Debugf("Duplicate message deduplicationId [%s] in queue [%s]", messageDeduplicationID, queueName)
} }
models.SyncQueues.Queues[key].InitDuplicatation(messageDeduplicationID) models.SyncQueues.Queues[key].InitDuplicatation(messageDeduplicationID)
// Сохраняем очередь в Redis пока держим Lock // Сохраняем очередь в Redis пока держим Lock
persistence.SaveQueue(key, models.SyncQueues.Queues[key]) persistence.SaveQueue(key, models.SyncQueues.Queues[key])
models.SyncQueues.Unlock() models.SyncQueues.Unlock()
log.Infof("%s: Queue: %s, Message: %s\n", time.Now().Format("2006-01-02 15:04:05"), queueName, msg.MessageBody) log.Infof("%s: Queue: %s, Message: %s\n", time.Now().Format("2006-01-02 15:04:05"), queueName, msg.MessageBody)
respStruct := models.SendMessageResponse{ respStruct := models.SendMessageResponse{
Xmlns: models.BaseXmlns, Xmlns: models.BaseXmlns,
Result: models.SendMessageResult{ Result: models.SendMessageResult{
MD5OfMessageAttributes: msg.MD5OfMessageAttributes, MD5OfMessageAttributes: msg.MD5OfMessageAttributes,
MD5OfMessageBody: msg.MD5OfMessageBody, MD5OfMessageBody: msg.MD5OfMessageBody,
MessageId: msg.Uuid, MessageId: msg.Uuid,
SequenceNumber: fifoSeqNumber, SequenceNumber: fifoSeqNumber,
}, },
Metadata: models.BaseResponseMetadata, Metadata: models.BaseResponseMetadata,
} }
return http.StatusOK, respStruct return http.StatusOK, respStruct
} }
shared-sqs/app/gosqs/set_queue_attributes.go
+43 -42
View File
@@ -4,54 +4,55 @@
package gosqs package gosqs
import ( import (
"net/http" "net/http"
"strings" "strings"
"shared-sqs/app/interfaces" "shared-sqs/app/interfaces"
"shared-sqs/app/models" "shared-sqs/app/models"
"shared-sqs/app/persistence" "shared-sqs/app/persistence"
"shared-sqs/app/utils" "shared-sqs/app/utils"
log "github.com/sirupsen/logrus"
log "github.com/sirupsen/logrus"
) )
func SetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBody) { func SetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
requestBody := models.NewSetQueueAttributesRequest() requestBody := models.NewSetQueueAttributesRequest()
ok := utils.REQUEST_TRANSFORMER(requestBody, req, false) ok := utils.REQUEST_TRANSFORMER(requestBody, req, false)
if !ok { if !ok {
log.Error("Invalid Request - SetQueueAttributesV1") log.Error("Invalid Request - SetQueueAttributesV1")
return utils.CreateErrorResponseV1("InvalidParameterValue", true) return utils.CreateErrorResponseV1("InvalidParameterValue", true)
} }
if requestBody.QueueUrl == "" { if requestBody.QueueUrl == "" {
log.Error("Missing QueueUrl - SetQueueAttributesV1") log.Error("Missing QueueUrl - SetQueueAttributesV1")
return utils.CreateErrorResponseV1("InvalidParameterValue", true) return utils.CreateErrorResponseV1("InvalidParameterValue", true)
} }
t := getTenantFromContext(req) t := getTenantFromContext(req)
if t == nil { if t == nil {
return utils.CreateErrorResponseV1("InvalidClientTokenId", true) return utils.CreateErrorResponseV1("InvalidClientTokenId", true)
} }
uriSegments := strings.Split(requestBody.QueueUrl, "/") uriSegments := strings.Split(requestBody.QueueUrl, "/")
queueName := uriSegments[len(uriSegments)-1] queueName := uriSegments[len(uriSegments)-1]
key := tenantQueueKey(t.AccessKey, queueName) key := tenantQueueKey(t.AccessKey, queueName)
log.Infof("Set Queue Attributes: %s (tenant: %s)", queueName, t.ID) log.Infof("Set Queue Attributes: %s (tenant: %s)", queueName, t.ID)
models.SyncQueues.Lock() models.SyncQueues.Lock()
defer models.SyncQueues.Unlock() defer models.SyncQueues.Unlock()
queue, ok := models.SyncQueues.Queues[key] queue, ok := models.SyncQueues.Queues[key]
if !ok { if !ok {
log.Warningf("Set Queue Attributes: %s, queue does not exist for tenant %s", queueName, t.ID) log.Warningf("Set Queue Attributes: %s, queue does not exist for tenant %s", queueName, t.ID)
return utils.CreateErrorResponseV1("QueueNotFound", true) return utils.CreateErrorResponseV1("QueueNotFound", true)
} }
if err := setQueueAttributesV1(queue, requestBody.Attributes); err != nil { if err := setQueueAttributesV1(queue, requestBody.Attributes); err != nil {
return utils.CreateErrorResponseV1(err.Error(), true) return utils.CreateErrorResponseV1(err.Error(), true)
} }
// Сохраняем атрибуты в Redis пока держим Lock (через defer) // Сохраняем атрибуты в Redis пока держим Lock (через defer)
persistence.SaveQueue(key, queue) persistence.SaveQueue(key, queue)
respStruct := models.SetQueueAttributesResponse{ respStruct := models.SetQueueAttributesResponse{
Xmlns: models.BaseXmlns, Xmlns: models.BaseXmlns,
Metadata: models.BaseResponseMetadata, Metadata: models.BaseResponseMetadata,
} }
return http.StatusOK, respStruct return http.StatusOK, respStruct
} }
shared-sqs/app/router/router.go
+8 -7
View File
@@ -39,13 +39,14 @@ func New(tenantStore *tenant.TenantStore, adminToken string) http.Handler {
// UI console — встроенный SPA, публичный доступ // UI console — встроенный SPA, публичный доступ
r.PathPrefix("/ui").Handler(http.StripPrefix("/ui", ui.Handler())) r.PathPrefix("/ui").Handler(http.StripPrefix("/ui", ui.Handler()))
// SQS API — tenant auth middleware // SQS API — tenant auth middleware оборачивает каждый handler отдельно.
sqsRouter := r.NewRoute().Subrouter() // r.NewRoute().Subrouter() с Use() некорректно работает в gorilla/mux v1.8.0
sqsRouter.Use(auth.AuthMiddleware(tenantStore)) // при пустом prefix — ответы теряются. Поэтому используем явную обёртку.
sqsRouter.HandleFunc("/", actionHandler).Methods("GET", "POST") sqsAuth := auth.AuthMiddleware(tenantStore)
sqsRouter.HandleFunc("/{account}", actionHandler).Methods("GET", "POST") r.Handle("/", sqsAuth(http.HandlerFunc(actionHandler))).Methods("GET", "POST")
sqsRouter.HandleFunc("/queue/{queueName}", actionHandler).Methods("GET", "POST") r.Handle("/{account}", sqsAuth(http.HandlerFunc(actionHandler))).Methods("GET", "POST")
sqsRouter.HandleFunc("/{account}/{queueName}", actionHandler).Methods("GET", "POST") r.Handle("/queue/{queueName}", sqsAuth(http.HandlerFunc(actionHandler))).Methods("GET", "POST")
r.Handle("/{account}/{queueName}", sqsAuth(http.HandlerFunc(actionHandler))).Methods("GET", "POST")
return r return r
} }
shared-sqs/app/ui/index.html
+2 -2
View File
@@ -857,7 +857,7 @@ function renderQueueMessages(queueName, msgs) {
onmouseover="this.style.background='rgba(26,127,212,0.08)'" onmouseout="this.style.background=''"> onmouseover="this.style.background='rgba(26,127,212,0.08)'" onmouseout="this.style.background=''">
<td style="padding:6px 12px;font-family:monospace;color:var(--text-secondary)">${esc(m.id).substring(0,8)}…</td> <td style="padding:6px 12px;font-family:monospace;color:var(--text-secondary)">${esc(m.id).substring(0,8)}…</td>
<td style="padding:6px 12px;max-width:380px;overflow:hidden;text-overflow:ellipsis;white-space:nowrap">${esc(m.body || '').substring(0,120)}${(m.body||'').length>120?'…':''}</td> <td style="padding:6px 12px;max-width:380px;overflow:hidden;text-overflow:ellipsis;white-space:nowrap">${esc(m.body || '').substring(0,120)}${(m.body||'').length>120?'…':''}</td>
<td style="padding:6px 12px;color:var(--text-secondary)">${fmtDate(m.sent)}</td> <td style="padding:6px 12px;color:var(--text-secondary)">${fmtDate(m.sent_at)}</td>
<td style="padding:6px 12px">${m.receives}</td> <td style="padding:6px 12px">${m.receives}</td>
</tr> </tr>
`).join('')} `).join('')}
@@ -872,7 +872,7 @@ function openMsgDetail(id) {
if (!m) return; if (!m) return;
document.getElementById('msg-detail-id').value = m.id || ''; document.getElementById('msg-detail-id').value = m.id || '';
document.getElementById('msg-detail-body').value = m.body || ''; document.getElementById('msg-detail-body').value = m.body || '';
document.getElementById('msg-detail-sent').value = m.sent ? fmtDate(m.sent) : ''; document.getElementById('msg-detail-sent').value = m.sent_at ? fmtDate(m.sent_at) : '';
document.getElementById('modal-msg-detail').classList.remove('hidden'); document.getElementById('modal-msg-detail').classList.remove('hidden');
} }
shared-sqs/deployments/k8s/deployment.yaml
+1 -1
View File
@@ -25,7 +25,7 @@ spec:
spec: spec:
containers: containers:
- name: shared-sqs - name: shared-sqs
image: naeel/shared-sqs:v0.1.11 image: naeel/shared-sqs:v0.1.14
ports: ports:
- containerPort: 4100 - containerPort: 4100
name: http name: http
shared-sqs/deployments/k8s/ingress.yaml
+1 -1
View File
@@ -1,6 +1,6 @@
# deployments/k8s/ingress.yaml # deployments/k8s/ingress.yaml
# Ingress для shared-sqs на домене qu.kube5s.ru # Ingress для shared-sqs на домене qu.kube5s.ru
# Created: 2026-04-09 # Created: 2026-04-09, Updated: 2026-04-10
apiVersion: networking.k8s.io/v1 apiVersion: networking.k8s.io/v1
kind: Ingress kind: Ingress
metadata: metadata:
shared-sqs/deployments/k8s/redis.yaml
+80
View File
@@ -0,0 +1,80 @@
# 2026-04-10 — Redis для shared-sqs в кластере naeel-test-3
# Single-node Redis с PVC для persistence состояния shared-sqs между рестартами.
# Namespace: shared-sqs. Пароль совпадает с managed Redis из deck.ngcloud.ru.
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: shared-sqs-redis-pvc
namespace: shared-sqs
spec:
accessModes:
- ReadWriteOnce
storageClassName: local-path
resources:
requests:
storage: 1Gi
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: shared-sqs-redis
namespace: shared-sqs
labels:
app: shared-sqs-redis
spec:
replicas: 1
selector:
matchLabels:
app: shared-sqs-redis
strategy:
type: Recreate
template:
metadata:
labels:
app: shared-sqs-redis
spec:
containers:
- name: redis
image: redis:7.2-alpine
command:
- redis-server
- --requirepass
- $(REDIS_PASSWORD)
- --appendonly
- "yes"
- --save
- "60 1"
ports:
- containerPort: 6379
env:
- name: REDIS_PASSWORD
valueFrom:
secretKeyRef:
name: shared-sqs-redis
key: password
volumeMounts:
- name: redis-data
mountPath: /data
resources:
requests:
memory: 128Mi
cpu: 50m
limits:
memory: 256Mi
cpu: 200m
volumes:
- name: redis-data
persistentVolumeClaim:
claimName: shared-sqs-redis-pvc
---
apiVersion: v1
kind: Service
metadata:
name: shared-sqs-redis
namespace: shared-sqs
spec:
selector:
app: shared-sqs-redis
ports:
- port: 6379
targetPort: 6379