Use ElasticMQ instead of fake_sqs

changelog.d/0-release-notes/elasticmq

@@ -0,0 +1,8 @@
+* Replace fake-sqs with ElasticMQ
+
+ElasticMQ is an actively maintained project, fake-sqs hasn't seen a commit since
+2018. This is not expected to have any noticeable effect on deployments that
+don't have any extra configurations for the SQS queues. If the fake-aws-sqs
+chart had configured custom queue names, they have couple of extra limitations:
+- The queue names must only contain alphanumeric characters and hyphens.
+- The FIFO queue names must end in `.fifo`.

charts/fake-aws-sqs/templates/configmap.yaml

@@ -0,0 +1,60 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ template "fullname" . }}
+  labels:
+    app: {{ template "fullname" . }}
+    chart: "{{ .Chart.Name }}-{{ .Chart.Version }}"
+    release: "{{ .Release.Name }}"
+    heritage: "{{ .Release.Service }}"
+data:
+  elasticmq.conf: |
+    include classpath("application.conf")
+
+    # What is the outside visible address of this ElasticMQ node
+    # Used to create the queue URL (may be different from bind address!)
+    node-address {
+      protocol = http
+      host = localhost
+      port = {{ .Values.service.httpPort }}
+      context-path = ""
+    }
+
+    rest-sqs {
+      enabled = true
+      bind-port = {{ .Values.service.httpPort }}
+      bind-hostname = "0.0.0.0"
+      # Possible values: relaxed, strict
+      sqs-limits = strict
+    }
+
+    rest-stats {
+      enabled = true
+      bind-port = 9325
+      bind-hostname = "0.0.0.0"
+    }
+
+    # Should the node-address be generated from the bind port/hostname
+    # Set this to true e.g. when assigning port automatically by using port 0.
+    generate-node-address = false
+
+    queues {
+    {{- range $i, $queueName := .Values.queueNames }}
+      "{{ $queueName }}" {
+      {{- if hasSuffix ".fifo" $queueName }}
+        fifo = true
+      {{- end }}
+      }
+    {{- end }}
+    }
+
+    messages-storage {
+      enabled = true
+      uri = "jdbc:h2:/data/elasticmq.db"
+    }
+
+    # Region and accountId which will be included in resource ids
+    aws {
+      region = eu-west-1
+      accountId = 000000000000
+    }

charts/fake-aws-sqs/templates/deployment.yaml

@@ -16,6 +16,8 @@ spec:
     metadata:
       labels:
         app: {{ template "fullname" . }}
+      annotations:
+        checksum/configmap: {{ include (print .Template.BasePath "/configmap.yaml") . | sha256sum }}
     spec:
       containers:
       - name: fake-aws-sqs
@@ -24,45 +26,22 @@ spec:
         - containerPort: {{ .Values.service.httpPort }}
           name: http
           protocol: TCP
+        command:
+        - /sbin/tini
+        - --
+        - /opt/elasticmq/bin/elasticmq-native-server
+        - -Dconfig.file=/config/elasticmq.conf
+        - -Dlogback.configurationFile=/opt/logback.xml
         volumeMounts:
         - name: storage
           mountPath: /data
+        - name: config
+          mountPath: /config
         resources:
 {{ toYaml .Values.resources | indent 12 }}
-      - name: initiate-fake-aws-sqs
-        image: mesosphere/aws-cli:1.14.5
-        command: [/bin/sh]
-        args:
-        - -c
-        - |
-          exec_until_ready() {
-              until $1; do echo 'service not ready yet'; sleep 1; done
-          }
-          queue_exists() {
-              # NOTE: we use the '"' to match the queue name more exactly (otherwise there is some overlap)
-              OUTPUT=$(aws --endpoint-url=http://localhost:{{ $.Values.service.httpPort }} sqs list-queues | grep $1'"' | wc -l)
-              echo $OUTPUT
-          }
-
-          echo 'Creating AWS resources'
-          aws configure set aws_access_key_id dummy
-          aws configure set aws_secret_access_key dummy
-          aws configure set region eu-west-1
-
-          while true
-          do
-              # Recreate resources if needed
-              {{ range $i, $queueName := .Values.queueNames }}
-                  QUEUE=$(queue_exists "{{ $queueName }}")
-                  if [ "$QUEUE" == "1" ]
-                    then echo "Queue {{ $queueName }} exists, no need to re-create"
-                    else exec_until_ready "aws --endpoint-url=http://localhost:{{ $.Values.service.httpPort }} sqs create-queue --queue-name {{ $queueName }}"
-                  fi
-              {{ end }}
-
-              echo 'Sleeping 10'
-              sleep 10
-          done
       volumes:
         - emptyDir: {}
           name: "storage"
+        - name: config
+          configMap:
+            name: {{ template "fullname" . }}

charts/fake-aws-sqs/values.yaml

@@ -1,6 +1,6 @@
 image:
-  repository: airdock/fake-sqs
-  tag: 0.3.1
+  repository: softwaremill/elasticmq-native
+  tag: 1.5.2
 
 # TODO: in a wire-server chart, these queue names should match the ones defined in galley/brig/gundeck (i.e. only be defined once)
 queueNames:

deploy/dockerephemeral/docker-compose.yaml

@@ -34,10 +34,12 @@ services:
 
   fake_sqs:
     container_name: demo_wire_sqs
-#    image: airdock/fake-sqs:0.3.1
-    image: julialongtin/airdock_fakesqs:0.0.9
+    image: softwaremill/elasticmq-native:1.5.2
     ports:
       - 127.0.0.1:4568:4568
+      - 127.0.0.1:9325:9325
+    volumes:
+      - ./docker/elasticmq.conf:/opt/elasticmq.conf
     networks:
       - demo_wire
 

deploy/dockerephemeral/docker/elasticmq.conf

@@ -0,0 +1,80 @@
+include classpath("application.conf")
+
+# What is the outside visible address of this ElasticMQ node
+# Used to create the queue URL (may be different from bind address!)
+node-address {
+  protocol = http
+  host = localhost
+  port = 4568
+  context-path = ""
+}
+
+rest-sqs {
+  enabled = true
+  bind-port = 4568
+  bind-hostname = "0.0.0.0"
+  # Possible values: relaxed, strict
+  sqs-limits = strict
+}
+
+rest-stats {
+  enabled = true
+  bind-port = 9325
+  bind-hostname = "0.0.0.0"
+}
+
+# Should the node-address be generated from the bind port/hostname
+# Set this to true e.g. when assigning port automatically by using port 0.
+generate-node-address = false
+
+queues {
+  default-queue-template {
+    defaultVisibilityTimeout = 1s
+  }
+
+  fifo-queue-template {
+    defaultVisibilityTimeout = 1s
+    fifo = true
+  }
+
+  integration-brig-events =  ${queues.default-queue-template}
+  integration-brig-events2 = ${queues.default-queue-template}
+  integration-brig-events3 = ${queues.default-queue-template}
+  integration-brig-events4 = ${queues.default-queue-template}
+  integration-brig-events5 = ${queues.default-queue-template}
+  integration-brig-events-federation-v0 = ${queues.default-queue-template}
+
+  integration-brig-events-internal =  ${queues.default-queue-template}
+  integration-brig-events-internal2 = ${queues.default-queue-template}
+  integration-brig-events-internal3 = ${queues.default-queue-template}
+  integration-brig-events-internal4 = ${queues.default-queue-template}
+  integration-brig-events-internal5 = ${queues.default-queue-template}
+  integration-brig-events-internal-federation-v0 = ${queues.default-queue-template}
+
+  "integration-user-events.fifo" = ${queues.fifo-queue-template}
+  "integration-user-events2.fifo" = ${queues.fifo-queue-template}
+  "integration-user-events3.fifo" = ${queues.fifo-queue-template}
+  "integration-user-events4.fifo" = ${queues.fifo-queue-template}
+  "integration-user-events5.fifo" = ${queues.fifo-queue-template}
+  "integration-user-events-federation-v0.fifo" = ${queues.fifo-queue-template}
+
+  integration-gundeck-events = ${queues.default-queue-template}
+  integration-gundeck-events2 = ${queues.default-queue-template}
+  integration-gundeck-events3 = ${queues.default-queue-template}
+  integration-gundeck-events4 = ${queues.default-queue-template}
+  integration-gundeck-events5 = ${queues.default-queue-template}
+  integration-gundeck-events-federation-v0 = ${queues.default-queue-template}
+
+  "integration-team-events.fifo" = ${queues.fifo-queue-template}
+  "integration-team-events2.fifo" = ${queues.fifo-queue-template}
+  "integration-team-events3.fifo" = ${queues.fifo-queue-template}
+  "integration-team-events4.fifo" = ${queues.fifo-queue-template}
+  "integration-team-events5.fifo" = ${queues.fifo-queue-template}
+  "integration-team-events-federation-v0.fifo" = ${queues.fifo-queue-template}
+}
+
+# Region and accountId which will be included in resource ids
+aws {
+  region = eu-west-1
+  accountId = 000000000000
+}

deploy/dockerephemeral/federation-v0/brig.yaml

@@ -36,7 +36,7 @@ federatorInternal:
 
 # You can set up local SQS/Dynamo running e.g. `../../deploy/dockerephemeral/run.sh`
 aws:
-  userJournalQueue: integration-user-events.fifo-federation-v0
+  userJournalQueue: integration-user-events-federation-v0.fifo
   # ^ Comment this out if you don't want to journal user events
   prekeyTable: integration-brig-prekeys-federation-v0
   sqsEndpoint: http://fake_sqs:4568 # https://sqs.eu-west-1.amazonaws.com

deploy/dockerephemeral/federation-v0/galley.yaml

@@ -88,6 +88,6 @@ logLevel: Warn
 logNetStrings: false
 
 journal: # if set, journals; if not set, disables journaling
-  queueName: integration-team-events.fifo-federation-v0
+  queueName: integration-team-events-federation-v0.fifo
   endpoint: http://demo_wire_sqs:4568 # https://sqs.eu-west-1.amazonaws.com
   region: eu-west-1

deploy/dockerephemeral/init.sh

@@ -22,25 +22,10 @@ for suffix in "" "2" "3" "4" "5" "-federation-v0"; do
     aws --endpoint-url=http://dynamodb:8000 dynamodb delete-table --table-name integration-brig-userkey-blacklist$suffix || true
     aws --endpoint-url=http://dynamodb:8000 dynamodb delete-table --table-name integration-brig-prekeys$suffix || true
 
-    # Create Dynamo/SQS resources
+    # Create Dynamo resources
     exec_until_ready "aws --endpoint-url=http://dynamodb:8000 dynamodb create-table --table-name integration-brig-userkey-blacklist$suffix --attribute-definitions AttributeName=key,AttributeType=S --key-schema AttributeName=key,KeyType=HASH --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5"
     exec_until_ready "aws --endpoint-url=http://dynamodb:8000 dynamodb create-table --table-name integration-brig-prekeys$suffix --attribute-definitions AttributeName=client,AttributeType=S --key-schema AttributeName=client,KeyType=HASH --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5"
 
-    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs create-queue --queue-name integration-brig-events$suffix"
-    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs set-queue-attributes --queue-url http://sqs:4568/integration-brig-events$suffix --attributes VisibilityTimeout=1"
-    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs create-queue --queue-name integration-brig-events-internal$suffix"
-    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs set-queue-attributes --queue-url http://sqs:4568/integration-brig-events-internal$suffix --attributes VisibilityTimeout=1"
-    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs create-queue --queue-name integration-user-events.fifo$suffix"
-    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs set-queue-attributes --queue-url http://sqs:4568/integration-user-events.fifo$suffix --attributes VisibilityTimeout=1"
-
-    # Gundeck's feedback queue
-    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs create-queue --queue-name integration-gundeck-events$suffix"
-    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs set-queue-attributes --queue-url http://sqs:4568/integration-gundeck-events$suffix --attributes VisibilityTimeout=1"
-
-    # Galley's team event queue
-    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs create-queue --queue-name integration-team-events.fifo$suffix"
-    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs set-queue-attributes --queue-url http://sqs:4568/integration-team-events.fifo$suffix --attributes VisibilityTimeout=1"
-
     # Verify sender's email address (ensure the sender address is in sync with the config in brig)
     exec_until_ready "aws --endpoint-url=http://ses:4579 ses verify-email-identity --email-address backend-integration$suffix@wire.com"
 
@@ -53,6 +38,13 @@ for suffix in "" "2" "3" "4" "5" "-federation-v0"; do
     # TODO: Lifecycle configuration for the bucket, if supported.
     aws --endpoint-url=http://s3:9000 s3api create-bucket --bucket "dummy-bucket$suffix"
     aws --endpoint-url=http://s3:9000 s3api wait bucket-exists --bucket "dummy-bucket$suffix"
+
+    # Check that SQS resources are created
+    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs get-queue-url --queue-name integration-brig-events$suffix"
+    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs get-queue-url --queue-name integration-brig-events-internal$suffix"
+    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs get-queue-url --queue-name integration-user-events$suffix.fifo"
+    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs get-queue-url --queue-name integration-gundeck-events"
+    exec_until_ready "aws --endpoint-url=http://sqs:4568 sqs get-queue-url --queue-name integration-team-events$suffix.fifo"
 done
 
 echo 'AWS resources created successfully!'

integration/test/Testlib/ResourcePool.hs

@@ -94,14 +94,14 @@ backendResources dynConfs =
                     berFederatorInternal = Ports.portForDyn (Ports.ServiceInternal FederatorInternal) i,
                     berFederatorExternal = dynConf.federatorExternalPort,
                     berDomain = dynConf.domain,
-                    berAwsUserJournalQueue = "integration-user-events.fifo" <> suffix i,
+                    berAwsUserJournalQueue = "integration-user-events" <> suffix i <> ".fifo",
                     berAwsPrekeyTable = "integration-brig-prekeys" <> suffix i,
                     berAwsS3Bucket = "dummy-bucket" <> suffix i,
                     berAwsQueueName = "integration-gundeck-events" <> suffix i,
                     berBrigInternalEvents = "integration-brig-events-internal" <> suffix i,
                     berEmailSMSSesQueue = "integration-brig-events" <> suffix i,
                     berEmailSMSEmailSender = "backend-integration" <> suffix i <> "@wire.com",
-                    berGalleyJournal = "integration-team-events.fifo" <> suffix i,
+                    berGalleyJournal = "integration-team-events" <> suffix i <> ".fifo",
                     berVHost = dynConf.domain,
                     berNginzSslPort = Ports.portForDyn Ports.NginzSSL i,
                     berNginzHttp2Port = Ports.portForDyn Ports.NginzHttp2 i,
@@ -151,14 +151,14 @@ backendB =
       berFederatorInternal = Ports.port (Ports.ServiceInternal FederatorInternal) BackendB,
       berFederatorExternal = Ports.port Ports.FederatorExternal BackendB,
       berDomain = "b.example.com",
-      berAwsUserJournalQueue = "integration-user-events.fifo2",
+      berAwsUserJournalQueue = "integration-user-events2.fifo",
       berAwsPrekeyTable = "integration-brig-prekeys2",
       berAwsS3Bucket = "dummy-bucket2",
       berAwsQueueName = "integration-gundeck-events2",
       berBrigInternalEvents = "integration-brig-events-internal2",
       berEmailSMSSesQueue = "integration-brig-events2",
       berEmailSMSEmailSender = "backend-integration2@wire.com",
-      berGalleyJournal = "integration-team-events.fifo2",
+      berGalleyJournal = "integration-team-events2.fifo",
       -- FUTUREWORK: set up vhosts in dev/ci for example.com and b.example.com
       -- in case we want backendA and backendB to federate with a third backend
       -- (because otherwise both queues will overlap)

libs/types-common-aws/src/Util/Test/SQS.hs

@@ -54,14 +54,18 @@ data SQSWatcher a = SQSWatcher
 -- the queue has too many things in it before the tests start.
 -- Note that the purgeQueue command is not guaranteed to be instant (can take up to 60 seconds)
 -- Hopefully, the fake-aws implementation used during tests is fast enough.
-watchSQSQueue :: Message a => AWS.Env -> Text -> IO (SQSWatcher a)
-watchSQSQueue env queueUrl = do
+watchSQSQueue :: (Message a) => AWS.Env -> Text -> IO (SQSWatcher a)
+watchSQSQueue env queueName = do
   eventsRef <- newIORef []
-  ensureEmpty
-  process <- async $ recieveLoop eventsRef
+
+  queueUrlRes <- execute env . sendEnv $ SQS.newGetQueueUrl queueName
+  let queueUrl = view SQS.getQueueUrlResponse_queueUrl queueUrlRes
+
+  ensureEmpty queueUrl
+  process <- async $ recieveLoop queueUrl eventsRef
   pure $ SQSWatcher process eventsRef
   where
-    recieveLoop ref = do
+    recieveLoop queueUrl ref = do
       let rcvReq =
             SQS.newReceiveMessage queueUrl
               & set SQS.receiveMessage_waitTimeSeconds (Just 100)
@@ -74,10 +78,10 @@ watchSQSQueue env queueUrl = do
         [] -> pure ()
         _ -> atomicModifyIORef ref $ \xs ->
           (parsedMsgs <> xs, ())
-      recieveLoop ref
+      recieveLoop queueUrl ref
 
-    ensureEmpty :: IO ()
-    ensureEmpty = void $ execute env $ sendEnv (SQS.newPurgeQueue queueUrl)
+    ensureEmpty :: Text -> IO ()
+    ensureEmpty queueUrl = void $ execute env $ sendEnv (SQS.newPurgeQueue queueUrl)
 
 -- | Waits for a message matching a predicate for a given number of seconds.
 waitForMessage :: (MonadUnliftIO m, Eq a) => SQSWatcher a -> Int -> (a -> Bool) -> m (Maybe a)
@@ -95,7 +99,7 @@ waitForMessage watcher seconds predicate = timeout (seconds * 1_000_000) poll
 -- an assertion on such a message.
 assertMessage :: (MonadUnliftIO m, Eq a, HasCallStack) => SQSWatcher a -> String -> (a -> Bool) -> (String -> Maybe a -> m ()) -> m ()
 assertMessage watcher label predicate callback = do
-  matched <- waitForMessage watcher 10 predicate
+  matched <- waitForMessage watcher 5 predicate
   callback label matched
 
 -----------------------------------------------------------------------------