From 9f1242aca3b4680f5de029935e04d129a84a65f2 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 10 Oct 2026 17:29:58 +0300 Subject: [PATCH] feat(aws-compat): queue attributes for Terraform (FifoQueue, ContentBasedDeduplication, Policy, real timestamps) - GetQueueAttributes now returns FifoQueue, ContentBasedDeduplication, Policy and real Created/LastModifiedTimestamp (previously missing/stubbed '0000000000') - Queue: add CreatedAt, LastModifiedAt, Policy, ContentBasedDeduplication - parse FifoQueue/ContentBasedDeduplication in both protocols (StringToBool) - Policy/RedriveAllowPolicy: normalize JSON-string form to object (JSON protocol sent them as string -> SetQueueAttributes failed with InvalidParameterValue) - persistence: keep new fields in meta DTO + backfill empty timestamps on load - FIFO dedup is inherent: IsDuplicate/InitDuplicatation gate on IsFIFO only (env flag EnableDuplicates=false disabled dedup for all FIFO queues) --- app/gosqs/create_queue.go | 3 ++ app/gosqs/get_queue_attributes.go | 25 +++++++++++- app/gosqs/queue_attributes.go | 21 ++++++++++ app/models/conversions.go | 28 +++++++++++++ app/models/models.go | 23 +++++++---- app/models/requests.go | 68 +++++++++++++++++++++++++++---- app/persistence/redis.go | 19 +++++++++ 7 files changed, 171 insertions(+), 16 deletions(-) diff --git a/app/gosqs/create_queue.go b/app/gosqs/create_queue.go index b3c29a5..1378007 100644 --- a/app/gosqs/create_queue.go +++ b/app/gosqs/create_queue.go @@ -59,6 +59,9 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { IsFIFO: utils.HasFIFOQueueName(queueName), EnableDuplicates: models.CurrentEnvironment.EnableDuplicates, Duplicates: make(map[string]time.Time), + // CreatedTimestamp/LastModifiedTimestamp в GetQueueAttributes (AWS их отдаёт). + CreatedAt: time.Now().UTC(), + LastModifiedAt: time.Now().UTC(), } // Список имён переданных атрибутов — для whitelist-валидации // (Query-протокол: Attribute.N.Name; JSON: nil — проверка пропускается). diff --git a/app/gosqs/get_queue_attributes.go b/app/gosqs/get_queue_attributes.go index 5e79445..e42dff8 100644 --- a/app/gosqs/get_queue_attributes.go +++ b/app/gosqs/get_queue_attributes.go @@ -7,6 +7,7 @@ import ( "net/http" "strconv" "strings" + "time" "shared-sqs/app/interfaces" "shared-sqs/app/models" @@ -86,6 +87,13 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo if shouldInclude("VisibilityTimeout") { queueAttributes = append(queueAttributes, models.Attribute{Name: "VisibilityTimeout", Value: strconv.Itoa(queue.VisibilityTimeout)}) } + // FIFO-атрибуты: читаются Terraform (aws_sqs_queue) и aws cli; без них — drift. + if shouldInclude("FifoQueue") { + queueAttributes = append(queueAttributes, models.Attribute{Name: "FifoQueue", Value: strconv.FormatBool(queue.IsFIFO)}) + } + if shouldInclude("ContentBasedDeduplication") { + queueAttributes = append(queueAttributes, models.Attribute{Name: "ContentBasedDeduplication", Value: strconv.FormatBool(queue.ContentBasedDeduplication)}) + } if shouldInclude("ApproximateNumberOfMessages") { queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessages", Value: strconv.Itoa(queue.LiveCount())}) } @@ -93,10 +101,10 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessagesNotVisible", Value: strconv.Itoa(numberOfHiddenMessagesInQueue(queue))}) } if shouldInclude("CreatedTimestamp") { - queueAttributes = append(queueAttributes, models.Attribute{Name: "CreatedTimestamp", Value: "0000000000"}) + queueAttributes = append(queueAttributes, models.Attribute{Name: "CreatedTimestamp", Value: epochSeconds(queue.CreatedAt)}) } if shouldInclude("LastModifiedTimestamp") { - queueAttributes = append(queueAttributes, models.Attribute{Name: "LastModifiedTimestamp", Value: "0000000000"}) + queueAttributes = append(queueAttributes, models.Attribute{Name: "LastModifiedTimestamp", Value: epochSeconds(queue.LastModifiedAt)}) } if shouldInclude("QueueArn") { queueAttributes = append(queueAttributes, models.Attribute{Name: "QueueArn", Value: queue.Arn}) @@ -107,6 +115,10 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo Value: fmt.Sprintf(`{"maxReceiveCount":"%d", "deadLetterTargetArn":"%s"}`, queue.MaxReceiveCount, queue.DeadLetterQueue.Arn), }) } + // Policy — отдаём сохранённую JSON-строку как есть (совместимость с Terraform). + if shouldInclude("Policy") && queue.Policy != "" { + queueAttributes = append(queueAttributes, models.Attribute{Name: "Policy", Value: queue.Policy}) + } respStruct := models.GetQueueAttributesResponse{ Xmlns: models.BaseXmlns, @@ -115,3 +127,12 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo } return http.StatusOK, respStruct } + +// epochSeconds — timestamp в формате AWS (unix-секунды строкой). +// Нулевое время отдаём как "0" (стабильно, без drift при чтении). +func epochSeconds(t time.Time) string { + if t.IsZero() { + return "0" + } + return strconv.FormatInt(t.Unix(), 10) +} diff --git a/app/gosqs/queue_attributes.go b/app/gosqs/queue_attributes.go index 3de8c2c..341db23 100644 --- a/app/gosqs/queue_attributes.go +++ b/app/gosqs/queue_attributes.go @@ -1,8 +1,10 @@ package gosqs import ( + "encoding/json" "fmt" "strings" + "time" log "github.com/sirupsen/logrus" @@ -85,5 +87,24 @@ func setQueueAttributesV1(q *models.Queue, attr models.QueueAttributes, provided q.DeadLetterQueue = deadLetterQueue q.MaxReceiveCount = attr.RedrivePolicy.MaxReceiveCount.Int() } + + // Шаг 4: FIFO-атрибуты и Policy — храним как есть (совместимость с Terraform). + // FifoQueue только включаем (выключить FIFO у существующей очереди нельзя — + // как и в AWS). ContentBasedDeduplication и Policy не исполняются, + // но должны возвращаться в GetQueueAttributes, иначе drift в Terraform. + if attr.FifoQueue.Bool() { + q.IsFIFO = true + } + if attr.ContentBasedDeduplication.Bool() { + q.ContentBasedDeduplication = true + } + if attr.Policy != nil { + // Сохраняем канонический JSON (map → marshal сортирует ключи) — + // детерминированно, без drift при чтении. + if data, err := json.Marshal(attr.Policy); err == nil { + q.Policy = string(data) + } + } + q.LastModifiedAt = time.Now().UTC() return nil } diff --git a/app/models/conversions.go b/app/models/conversions.go index 35c7530..c87f1b0 100644 --- a/app/models/conversions.go +++ b/app/models/conversions.go @@ -34,3 +34,31 @@ func (s *StringToInt) UnmarshalJSON(data []byte) error { func (s *StringToInt) Int() int { return int(*s) } + +// StringToBool — аналог StringToInt для булевых атрибутов: принимает и JSON-bool +// (true/false), и строку ("true"/"false"). Query-протокол шлёт строки, +// JSON-протокол — булевы значения; без этого FifoQueue/ContentBasedDeduplication +// из JSON-запроса молча терялись бы. +type StringToBool bool + +func (s *StringToBool) UnmarshalJSON(data []byte) error { + var b bool + if err := json.Unmarshal(data, &b); err == nil { + *s = StringToBool(b) + return nil + } + var str string + if err := json.Unmarshal(data, &str); err != nil { + return err + } + tmp, err := strconv.ParseBool(str) + if err != nil { + return err + } + *s = StringToBool(tmp) + return nil +} + +func (s *StringToBool) Bool() bool { + return bool(*s) +} diff --git a/app/models/models.go b/app/models/models.go index 27d5370..aedf6c3 100644 --- a/app/models/models.go +++ b/app/models/models.go @@ -92,6 +92,16 @@ type Queue struct { // Используется для запрета повторного purge ранее чем через 60 секунд // (требование AWS: PurgeQueueInProgress). Нулевое значение = purge ещё не выполнялся. LastPurgeTime time.Time + // CreatedAt/LastModifiedAt — для атрибутов CreatedTimestamp/LastModifiedTimestamp + // (AWS отдаёт их в GetQueueAttributes; Terraform читает и без них видел бы drift). + CreatedAt time.Time + LastModifiedAt time.Time + // Policy — resource policy (JSON-строка), храним и отдаём как есть для + // совместимости с Terraform. НЕ исполняется (нет enforcement). + Policy string + // ContentBasedDeduplication — FIFO-атрибут. Храним и отдаём значение, + // переданное клиентом (для совместимости), см. раздел про dedup. + ContentBasedDeduplication bool } // LockMu/UnlockMu/RLockMu/RUnlockMu — доступ к per-queue блокировке извне пакета models @@ -150,13 +160,12 @@ func (q *Queue) UnlockGroup(groupId string) { } // IsDuplicate — проверяет, было ли сообщение с таким deduplicationId уже -// добавлено в очередь в пределах deduplication-окна. -// -// Логическая схема: дубликаты отслеживаются ТОЛЬКО для FIFO-очередей с -// включённой дупликацией (EnableDuplicates) и непустым deduplicationId. -// Для остальных случаев дубликаты не отслеживаются (AWS-семантика). +// добавлено в очередь в пределах deduplication-окна (5 минут). +// Дедупликация — НЕОТЪЕМЛЕМОЕ свойство FIFO-очереди (AWS): раньше она зависела +// от env-флага EnableDuplicates=false по умолчанию, и FIFO-очереди не +// дедуплицировали вообще. Теперь — только по IsFIFO и непустому id. func (q *Queue) IsDuplicate(deduplicationId string) bool { - if !q.EnableDuplicates || !q.IsFIFO || deduplicationId == "" { + if !q.IsFIFO || deduplicationId == "" { return false } @@ -170,7 +179,7 @@ func (q *Queue) IsDuplicate(deduplicationId string) bool { // фоновой задачей PeriodicTasks (deduplication period expired). // Для не-FIFO очередей или пустого deduplicationId — no-op. func (q *Queue) InitDuplicatation(deduplicationId string) { - if !q.EnableDuplicates || !q.IsFIFO || deduplicationId == "" { + if !q.IsFIFO || deduplicationId == "" { return } diff --git a/app/models/requests.go b/app/models/requests.go index a1c4d8e..672d87b 100644 --- a/app/models/requests.go +++ b/app/models/requests.go @@ -113,6 +113,20 @@ func (r *CreateQueueRequest) SetAttributesFromForm(values url.Values) { continue } r.Attributes.RedriveAllowPolicy = tmp + case "FifoQueue": + tmp, err := strconv.ParseBool(attrValue) + if err != nil { + log.Debugf("Failed to parse form attribute - %s: %s", attrName, attrValue) + continue + } + r.Attributes.FifoQueue = StringToBool(tmp) + case "ContentBasedDeduplication": + tmp, err := strconv.ParseBool(attrValue) + if err != nil { + log.Debugf("Failed to parse form attribute - %s: %s", attrName, attrValue) + continue + } + r.Attributes.ContentBasedDeduplication = StringToBool(tmp) } } return @@ -381,6 +395,20 @@ func (r *SetQueueAttributesRequest) SetAttributesFromForm(values url.Values) { continue } r.Attributes.RedriveAllowPolicy = tmp + case "FifoQueue": + tmp, err := strconv.ParseBool(attrValue) + if err != nil { + log.Debugf("Failed to parse form attribute - %s: %s", attrName, attrValue) + continue + } + r.Attributes.FifoQueue = StringToBool(tmp) + case "ContentBasedDeduplication": + tmp, err := strconv.ParseBool(attrValue) + if err != nil { + log.Debugf("Failed to parse form attribute - %s: %s", attrName, attrValue) + continue + } + r.Attributes.ContentBasedDeduplication = StringToBool(tmp) } } return @@ -393,12 +421,17 @@ type QueueAttributes struct { DelaySeconds StringToInt `json:"DelaySeconds"` MaximumMessageSize StringToInt `json:"MaximumMessageSize"` MessageRetentionPeriod StringToInt `json:"MessageRetentionPeriod"` // NOTE: not implemented - Policy map[string]interface{} `json:"Policy"` // NOTE: not implemented - ReceiveMessageWaitTimeSeconds StringToInt `json:"ReceiveMessageWaitTimeSeconds"` - VisibilityTimeout StringToInt `json:"VisibilityTimeout"` - // Dead Letter Queues Only - RedrivePolicy RedrivePolicy `json:"RedrivePolicy"` - RedriveAllowPolicy map[string]interface{} `json:"RedriveAllowPolicy"` // NOTE: not implemented + Policy map[string]interface{} `json:"Policy"` // хранится, не исполняется + ReceiveMessageWaitTimeSeconds StringToInt `json:"ReceiveMessageWaitTimeSeconds"` + VisibilityTimeout StringToInt `json:"VisibilityTimeout"` + // Dead Letter Queues Only + RedrivePolicy RedrivePolicy `json:"RedrivePolicy"` + RedriveAllowPolicy map[string]interface{} `json:"RedriveAllowPolicy"` // NOTE: not implemented + // FIFO-атрибуты. FifoQueue у нас также выводится из суффикса .fifo имени, + // но значение от клиента нужно СОХРАНИТЬ и вернуть в GetQueueAttributes — + // иначе Terraform (aws_sqs_queue) видел бы постоянный drift. + FifoQueue StringToBool `json:"FifoQueue"` + ContentBasedDeduplication StringToBool `json:"ContentBasedDeduplication"` } // AttrNameWhitelist — имена атрибутов, допустимые в SetQueueAttributes/CreateQueue. @@ -444,8 +477,29 @@ func (q *QueueAttributes) UnmarshalJSON(data []byte) error { return fmt.Errorf("invalid attribute name: %s", name) } } + // Policy/RedriveAllowPolicy в SQS API — СТРОКА с JSON внутри, а поля у нас — + // map. JSON-протокол присылает их именно строкой, и unmarshal в map падал бы + // (SetQueueAttributes → InvalidParameterValue). Разворачиваем строку в объект. + // Также принимаем вариант, где уже пришёл объект — он и так валиден. + for _, name := range []string{"Policy", "RedriveAllowPolicy"} { + rv, ok := raw[name] + if !ok { + continue + } + var s string + if err := json.Unmarshal(rv, &s); err == nil { + if !json.Valid([]byte(s)) { + return fmt.Errorf("invalid JSON in attribute %s", name) + } + raw[name] = json.RawMessage(s) + } + } + normalized, err := json.Marshal(raw) + if err != nil { + return err + } type plain QueueAttributes - return json.Unmarshal(data, (*plain)(q)) + return json.Unmarshal(normalized, (*plain)(q)) } type RedrivePolicy struct { diff --git a/app/persistence/redis.go b/app/persistence/redis.go index 1e9edfe..7bbaef4 100644 --- a/app/persistence/redis.go +++ b/app/persistence/redis.go @@ -167,6 +167,11 @@ type queueMetaDTO struct { Tags map[string]string LastPurgeTime time.Time DeadLetterQueueName string `json:"dead_letter_queue_name,omitempty"` + // Создание/изменение + совместимые атрибуты (GetQueueAttributes/Terraform). + CreatedAt time.Time + LastModifiedAt time.Time + Policy string + ContentBasedDeduplication bool } // queueToMetaDTO — строит снимок очереди без сообщений, НЕ мутируя исходную очередь. @@ -194,6 +199,10 @@ func queueToMetaDTO(q *models.Queue) *queueMetaDTO { Tags: q.Tags, LastPurgeTime: q.LastPurgeTime, DeadLetterQueueName: q.DeadLetterQueueName, + CreatedAt: q.CreatedAt, + LastModifiedAt: q.LastModifiedAt, + Policy: q.Policy, + ContentBasedDeduplication: q.ContentBasedDeduplication, } if q.DeadLetterQueue != nil { dto.DeadLetterQueueName = q.DeadLetterQueue.Name @@ -452,6 +461,16 @@ func LoadAllQueues() (map[string]*models.Queue, error) { q.FIFOSequenceNumbers = make(map[string]int) } + // Бэкфилл timestamp-ов: у очередей, созданных до появления полей, они нулевые. + // Без этого Terraform видел бы CreatedTimestamp=0 и постоянный drift. + // Заполняем сейчас и сохраним при ближайшем SaveQueue. + if q.CreatedAt.IsZero() { + q.CreatedAt = time.Now().UTC() + } + if q.LastModifiedAt.IsZero() { + q.LastModifiedAt = q.CreatedAt + } + // Миграция v1→v2: если в JSON есть Messages — это старый формат if len(q.Messages) > 0 { migrateKeys = append(migrateKeys, k)