test(e2e): add worker service test suite (#2879)

efekarakus · web-flow · commit 4efabf4070d6 · 2021-09-29T17:41:16.000Z
The test ensures:
1. The worker service receives and consumes sqs messages
2. Can contact other services via service discovery

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the Apache 2.0 License.
diff --git a/.release/buildspec_e2e.yml b/.release/buildspec_e2e.yml
@@ -120,7 +120,15 @@ batch:
           APP_REGION: eu-west-1
           TESTENV_REGION: eu-west-1
           PRODENV_REGION: eu-central-1
-
+    - identifier: worker
+      env:
+        privileged-mode: true
+        type: LINUX_CONTAINER
+        compute-type: BUILD_GENERAL1_LARGE
+        image: aws/codebuild/standard:5.0
+        variables:
+          TEST_SUITE: worker
+          APP_REGION: eu-central-1
 
 phases:
   install:
diff --git a/e2e/internal/client/cli.go b/e2e/internal/client/cli.go
@@ -80,11 +80,12 @@ type EnvShowRequest struct {
 
 // SvcInitRequest contains the parameters for calling copilot svc init.
 type SvcInitRequest struct {
-	Name       string
-	SvcType    string
-	Dockerfile string
-	Image      string
-	SvcPort    string
+	Name               string
+	SvcType            string
+	Dockerfile         string
+	Image              string
+	SvcPort            string
+	TopicSubscriptions []string
 }
 
 // SvcShowRequest contains the parameters for calling copilot svc show.
@@ -314,6 +315,9 @@ func (cli *CLI) SvcInit(opts *SvcInitRequest) (string, error) {
 	if opts.Image != "" {
 		args = append(args, "--image", opts.Image)
 	}
+	if len(opts.TopicSubscriptions) > 0 {
+		args = append(args, "--subscribe-topics", strings.Join(opts.TopicSubscriptions, ","))
+	}
 	return cli.exec(
 		exec.Command(cli.path, args...))
 }
diff --git a/e2e/worker/README.md b/e2e/worker/README.md
@@ -0,0 +1,7 @@
+# Worker Service E2E tests
+
+The goal of the `worker` e2e test is to validate the most common usecases for a
+[Worker Service](https://aws.github.io/copilot-cli/docs/concepts/services/#worker-service).
+
+- Receiving and deleting messages from the default SQS queue: `COPILOT_QUEUE_URI`
+- Making requests via service discovery to other services in the application.
diff --git a/e2e/worker/frontend/Dockerfile b/e2e/worker/frontend/Dockerfile
@@ -0,0 +1,7 @@
+FROM node:14-slim
+WORKDIR /usr/src/app
+COPY package*.json ./
+RUN npm install
+COPY . .
+EXPOSE 8080
+CMD [ "node", "server.js" ]
diff --git a/e2e/worker/frontend/package.json b/e2e/worker/frontend/package.json
@@ -0,0 +1,13 @@
+{
+  "name": "frontend",
+  "version": "1.0.0",
+  "author": "The ECS DevX team",
+  "main": "server.js",
+  "scripts": {
+    "start": "node server.js"
+  },
+  "dependencies": {
+    "@aws-sdk/client-sns": "^3.28.0",
+    "express": "^4.16.1"
+  }
+}
diff --git a/e2e/worker/frontend/server.js b/e2e/worker/frontend/server.js
@@ -0,0 +1,38 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+'use strict';
+
+const express = require('express');
+const { SNSClient, PublishCommand } = require("@aws-sdk/client-sns");
+
+const PORT = 8080;
+const HOST = '0.0.0.0';
+const client = new SNSClient({ region: process.env.AWS_DEFAULT_REGION });
+const app = express();
+
+// Start the service waiting for an ack message.
+let status = 'waiting on acknowledgement';
+
+// Each health check request from the ALB will result in publishing an event.
+app.get('/', async (req, res) => {
+  const {events} = JSON.parse(process.env.COPILOT_SNS_TOPIC_ARNS);
+  const out = await client.send(new PublishCommand({
+    Message: "healthcheck",
+    TopicArn: events,
+  }));
+  console.log(JSON.stringify(out));
+  res.send('hello');
+});
+
+app.get('/status', (req, res) => {
+  res.send(status);
+});
+
+app.post('/ack', async (req, res) => {
+  status = 'consumed';
+  res.send('ok');
+});
+
+app.listen(PORT, HOST);
+console.log(`Running on http://${HOST}:${PORT}`);
diff --git a/e2e/worker/worker/Dockerfile b/e2e/worker/worker/Dockerfile
@@ -0,0 +1,6 @@
+FROM node:14-slim
+WORKDIR /usr/src/app
+COPY package*.json ./
+RUN npm install
+COPY . .
+CMD [ "node", "index.js" ]
diff --git a/e2e/worker/worker/index.js b/e2e/worker/worker/index.js
@@ -0,0 +1,41 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+const { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } = require("@aws-sdk/client-sqs");
+const axios = require("axios");
+const client = new SQSClient({ region: process.env.AWS_DEFAULT_REGION });
+
+console.log(`COPILOT_QUEUE_URI: ${process.env.COPILOT_QUEUE_URI}`);
+
+const eventsQueue = process.env.COPILOT_QUEUE_URI;
+
+function sleep(ms) {
+  return new Promise(resolve => setTimeout(resolve, ms));
+}
+
+(async () => {
+  console.log(`The queue url created is: ${eventsQueue}`);
+  while(true) {
+    try {
+      await sleep(300);
+      const out = await client.send(new ReceiveMessageCommand({
+        QueueUrl: eventsQueue,
+        WaitTimeSeconds: 10,
+      }));
+      console.log(`results: ${JSON.stringify(out)}`);
+
+      if (out.Messages === undefined || out.Messages.length === 0) {
+        continue;
+      }
+
+      const resp = await axios.post(`http://frontend.${process.env.COPILOT_SERVICE_DISCOVERY_ENDPOINT}:8080/ack`);
+      console.log(`response from frontend service: ${JSON.stringify(resp)}`);
+
+      await client.send( new DeleteMessageCommand({
+        QueueUrl: eventsQueue,
+        ReceiptHandle: out.Messages[0].ReceiptHandle,
+      }));
+    } catch (err) {
+      console.error(err);
+    }
+  }
+})();
diff --git a/e2e/worker/worker/package.json b/e2e/worker/worker/package.json
@@ -0,0 +1,13 @@
+{
+  "name": "worker",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "dependencies": {
+    "@aws-sdk/client-sqs": "^3.27.0",
+    "axios": "^0.21.1"
+  }
+}
diff --git a/e2e/worker/worker_suite_test.go b/e2e/worker/worker_suite_test.go
@@ -0,0 +1,43 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package worker_test
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/aws/copilot-cli/e2e/internal/client"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+var (
+	cli     *client.CLI
+	appName string
+)
+
+const (
+	lbwsServiceName   = "frontend"
+	workerServiceName = "worker"
+
+	envName = "test"
+)
+
+func TestWorker(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "App Runner Suite")
+}
+
+var _ = BeforeSuite(func() {
+	var err error
+	cli, err = client.NewCLI()
+	Expect(err).NotTo(HaveOccurred())
+	appName = fmt.Sprintf("e2e-worker-%d", time.Now().Unix())
+})
+
+var _ = AfterSuite(func() {
+	_, err := cli.AppDelete()
+	Expect(err).NotTo(HaveOccurred())
+})
diff --git a/e2e/worker/worker_test.go b/e2e/worker/worker_test.go
@@ -0,0 +1,160 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package worker_test
+
+import (
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+
+	"github.com/aws/copilot-cli/e2e/internal/client"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Worker Service E2E Test", func() {
+	Context("create an application", func() {
+		It("app init succeeds", func() {
+			_, err := cli.AppInit(&client.AppInitRequest{
+				AppName: appName,
+			})
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("app init creates a copilot directory", func() {
+			Expect("./copilot").Should(BeADirectory())
+		})
+
+		It("app show includes app name", func() {
+			appShowOutput, err := cli.AppShow(appName)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(appShowOutput.Name).To(Equal(appName))
+			Expect(appShowOutput.URI).To(BeEmpty())
+		})
+	})
+
+	Context("create an environment", func() {
+		It("env init should succeed", func() {
+			_, err := cli.EnvInit(&client.EnvInitRequest{
+				AppName: appName,
+				EnvName: envName,
+				Profile: "default",
+			})
+			Expect(err).NotTo(HaveOccurred())
+		})
+	})
+
+	Context("create a load balanced web service", func() {
+		It("svc init should succeed", func() {
+			_, err := cli.SvcInit(&client.SvcInitRequest{
+				Name:       lbwsServiceName,
+				SvcType:    "Load Balanced Web Service",
+				Dockerfile: "./frontend/Dockerfile",
+			})
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("adds a topic to push to the service", func() {
+			Expect("./copilot/frontend/manifest.yml").Should(BeAnExistingFile())
+			f, err := os.OpenFile("./copilot/frontend/manifest.yml", os.O_APPEND|os.O_WRONLY, 0644)
+			Expect(err).NotTo(HaveOccurred())
+			defer f.Close()
+			// Append publish section to manifest.
+			_, err = f.WriteString(`
+publish:
+  topics:
+  - name: events
+`)
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		It("deploys the load balanced web service", func() {
+			_, err := cli.SvcDeploy(&client.SvcDeployInput{
+				Name:     lbwsServiceName,
+				EnvName:  envName,
+				ImageTag: "gallopinggurdey",
+			})
+			Expect(err).NotTo(HaveOccurred())
+		})
+	})
+
+	Context("deploys the worker service", func() {
+		It("svc init should succeed", func() {
+			_, err := cli.SvcInit(&client.SvcInitRequest{
+				Name:               workerServiceName,
+				SvcType:            "Worker Service",
+				Dockerfile:         "./worker/Dockerfile",
+				TopicSubscriptions: []string{fmt.Sprintf("%s:%s", lbwsServiceName, "events")},
+			})
+			Expect(err).NotTo(HaveOccurred())
+		})
+		It("deploys the worker service", func() {
+			_, err := cli.SvcDeploy(&client.SvcDeployInput{
+				Name:     workerServiceName,
+				EnvName:  envName,
+				ImageTag: "gallopinggurdey",
+			})
+			Expect(err).NotTo(HaveOccurred())
+		})
+		It("should have the SQS queue and service discovery injected as env vars", func() {
+			svc, err := cli.SvcShow(&client.SvcShowRequest{
+				AppName: appName,
+				Name:    workerServiceName,
+			})
+			Expect(err).NotTo(HaveOccurred())
+			var hasQueueURI bool
+			var svcDiscovery bool
+			for _, envVar := range svc.Variables {
+				switch envVar.Name {
+				case "COPILOT_QUEUE_URI":
+					hasQueueURI = true
+				case "COPILOT_SERVICE_DISCOVERY_ENDPOINT":
+					svcDiscovery = true
+				}
+			}
+			if !hasQueueURI {
+				Expect(errors.New("worker service is missing env var 'COPILOT_QUEUE_URI'")).NotTo(HaveOccurred())
+			}
+			if !svcDiscovery {
+				Expect(errors.New("worker service is missing env var 'COPILOT_SERVICE_DISCOVERY_ENDPOINT'")).NotTo(HaveOccurred())
+			}
+		})
+	})
+
+	Context("should have consumed messages", func() {
+		It("frontend service should have received acknowledgement", func() {
+			svc, err := cli.SvcShow(&client.SvcShowRequest{
+				AppName: appName,
+				Name:    lbwsServiceName,
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(len(svc.Routes)).To(Equal(1))
+			route := svc.Routes[0]
+			Expect(route.Environment).To(Equal(envName))
+			Eventually(func() error {
+				resp, err := http.Get(fmt.Sprintf("%s/status", route.URL))
+				if err != nil {
+					return err
+				}
+				if resp.StatusCode != http.StatusOK {
+					return fmt.Errorf("response status is %d and not %d", resp.StatusCode, http.StatusOK)
+				}
+
+				body, err := ioutil.ReadAll(resp.Body)
+				if err != nil {
+					return err
+				}
+				fmt.Printf("response: %s\n", string(body))
+				// The LBWS sets its state to "consumed" when the worker service processes a message
+				if string(body) != "consumed" {
+					return fmt.Errorf("the message content is '%s', but expected '%s'", string(body), "consumed")
+				}
+				return nil
+			}, "60s", "1s").ShouldNot(HaveOccurred())
+		})
+	})
+})
