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)
This commit is contained in:
@@ -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 — проверка пропускается).
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
+16
-7
@@ -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
|
||||
}
|
||||
|
||||
|
||||
+61
-7
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user