From 78dc2f6cbcb27f5e99fa10614e62b416c24256b9 Mon Sep 17 00:00:00 2001 From: Michael McGuinness Date: Wed, 5 Feb 2025 18:33:06 +0000 Subject: [PATCH] Merged in feature/localstackcompose (pull request #49) Add Localstack to Compose * localstack * addedlocalstackpluscleanup# --- deployments/compose.yaml | 23 ++++++++++++ devbox.json | 14 ++++++-- devbox.lock | 64 ++++++++++++++++++++++++++++++++++ internal/server/runner/poll.go | 56 +++++++---------------------- scripts/compose.yml | 10 ++++-- scripts/database.yml | 4 +-- 6 files changed, 121 insertions(+), 50 deletions(-) diff --git a/deployments/compose.yaml b/deployments/compose.yaml index 6b48afa6..7997efbc 100644 --- a/deployments/compose.yaml +++ b/deployments/compose.yaml @@ -19,6 +19,29 @@ services: retries: 5 networks: - server-network + localstack: + image: localstack/localstack:4.1.0 + ports: + - 4566:4566 + environment: + AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID} + AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY} + AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN} + AWS_REGION: ${AWS_REGION} + SERVICES: "s3,sqs" + SKIP_SSL_CERT_DOWNLOAD: "1" + LOCALSTACK_HOST: localstack + SQS_ENDPOINT_STRATEGY: "path" + EAGER_SERVICE_LOADING: "1" + expose: + - 4566 + # healthcheck: + # test: ["CMD-SHELL", "pg_isready -U ${DB_USER}"] + # interval: 10s + # timeout: 5s + # retries: 5 + networks: + - server-network volumes: db-data: networks: diff --git a/devbox.json b/devbox.json index 722094b6..4bed49ed 100644 --- a/devbox.json +++ b/devbox.json @@ -16,7 +16,8 @@ "gotools@0.25.0", "nodePackages.jsonlint@1.6.3", "oapi-codegen@2.4.1", - "vacuum-go@0.14.1" + "vacuum-go@0.14.1", + "awscli2@2.19.0" ], "shell": { "init_hook": [ @@ -34,7 +35,16 @@ "DB_NOSSL": "true", "AWS_ACCESS_KEY_ID": "test", "AWS_SECRET_ACCESS_KEY": "test", - "AWS_REGION": "us-east-1" + "AWS_SESSION_TOKEN": "", + "AWS_REGION": "us-east-1", + "AWS_ENDPOINT_URL": "http://localhost:4566", + "QNAME_DOCUMENT_CLEAN": "document_clean", + "QNAME_DOCUMENT_INIT": "document_init", + "QNAME_QUERY_RUNNER": "query_runner", + "DOCUMENT_CLEAN_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_clean", + "DOCUMENT_INIT_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_init", + "QUERYRUNNER_URL": "http://localstack:4566/queue/us-east-1/000000000000/query_runner", + "BUCKET_IN": "documentin" }, "env_from": ".env" } diff --git a/devbox.lock b/devbox.lock index 9ab80907..1ba5d6cc 100644 --- a/devbox.lock +++ b/devbox.lock @@ -1,6 +1,70 @@ { "lockfile_version": "1", "packages": { + "awscli2@2.19.0": { + "last_modified": "2024-12-03T12:40:06Z", + "resolved": "github:NixOS/nixpkgs/566e53c2ad750c84f6d31f9ccb9d00f823165550#awscli2", + "source": "devbox-search", + "version": "2.19.0", + "systems": { + "aarch64-darwin": { + "outputs": [ + { + "name": "out", + "path": "/nix/store/ywgz36mwv7q52m3lnsj8d1h7x70nyylc-awscli2-2.19.0", + "default": true + }, + { + "name": "dist", + "path": "/nix/store/254shki4ihfaqw0h1v5xj9vqcc0y5vkz-awscli2-2.19.0-dist" + } + ], + "store_path": "/nix/store/ywgz36mwv7q52m3lnsj8d1h7x70nyylc-awscli2-2.19.0" + }, + "aarch64-linux": { + "outputs": [ + { + "name": "out", + "path": "/nix/store/irx8wcn8hqar9nnq1h9b92cssl7plj1g-awscli2-2.19.0", + "default": true + }, + { + "name": "dist", + "path": "/nix/store/yqnb36a2b6bmlpsbxx28sb5857r6j9dz-awscli2-2.19.0-dist" + } + ], + "store_path": "/nix/store/irx8wcn8hqar9nnq1h9b92cssl7plj1g-awscli2-2.19.0" + }, + "x86_64-darwin": { + "outputs": [ + { + "name": "out", + "path": "/nix/store/6jgmxb4gc0zbhhdd2h566vz4v08gz19g-awscli2-2.19.0", + "default": true + }, + { + "name": "dist", + "path": "/nix/store/3zxc4fc3jqbd2m19p37ydpnj8w3s3mpm-awscli2-2.19.0-dist" + } + ], + "store_path": "/nix/store/6jgmxb4gc0zbhhdd2h566vz4v08gz19g-awscli2-2.19.0" + }, + "x86_64-linux": { + "outputs": [ + { + "name": "out", + "path": "/nix/store/g2w36iqqxj51nlfpid40dq102ag6kq8y-awscli2-2.19.0", + "default": true + }, + { + "name": "dist", + "path": "/nix/store/r5rkanfiin4i6cicbhl5jfswydq9kqgw-awscli2-2.19.0-dist" + } + ], + "store_path": "/nix/store/g2w36iqqxj51nlfpid40dq102ag6kq8y-awscli2-2.19.0" + } + } + }, "go-migrate@4.18.1": { "last_modified": "2024-12-23T21:10:33Z", "resolved": "github:NixOS/nixpkgs/de1864217bfa9b5845f465e771e0ecb48b30e02d#go-migrate", diff --git a/internal/server/runner/poll.go b/internal/server/runner/poll.go index a08512a9..d64a1e1c 100644 --- a/internal/server/runner/poll.go +++ b/internal/server/runner/poll.go @@ -20,7 +20,6 @@ func (c *Server) Listen(ctx context.Context) { } }() - // TODO need to slow down the polling so we are not spinning the cpu or burning up API calls. for { select { case <-ctx.Done(): @@ -35,31 +34,6 @@ func (c *Server) Listen(ctx context.Context) { } -// Alternative without go func if no pool is needed. -// Just for comment for the next function which needs to be made safe. -//func (c *Server) pollMessage(ctx context.Context) error { -// cfg := &queue.Config{ -// URL: c.queueURL, -// Client: c.client, -// } -// result, err := queue.Receive(ctx, cfg, []string{"type"}) -// if err != nil { -// return fmt.Errorf("message fetch fail: %v", err) -// } -// -// for _, message := range result.Messages { -// if err := c.controller.Process(ctx, &message); err != nil { -// return fmt.Errorf("message process fail: %v", err) -// } -// -// if err := queue.Delete(ctx, cfg, &message); err != nil { -// return fmt.Errorf("message delete fail: %v", err) -// } -// } -// -// return nil -//} - func (c *Server) pollMessage(ctx context.Context) error { result, err := c.cfg.ReceiveFromQueue(ctx, &queue.ReceiveParams{ QueueURL: c.cfg.GetQueueURL(), @@ -69,24 +43,20 @@ func (c *Server) pollMessage(ctx context.Context) error { } for _, message := range result.Messages { - // TODO Need to remove this go func() !! - // Must not have uncontrolled go routine creation. See commented example above if - // simple case. - // if process takes too long to complete then we need to use a thread pool. - go func() { - err := c.cfg.GetController().Process(ctx, &message) - if err != nil { - slog.Error("message process fail", "err", err) - } + slog.Info("processing message", "id", message.MessageId) - err = c.cfg.DeleteFromQueue(ctx, &queue.DeleteParams{ - QueueURL: c.cfg.GetQueueURL(), - ReceiptHandle: message.ReceiptHandle, - }) - if err != nil { - slog.Error("message delete fail", "err", err) - } - }() + err := c.cfg.GetController().Process(ctx, &message) + if err != nil { + slog.Error("message process fail", "err", err) + } + + err = c.cfg.DeleteFromQueue(ctx, &queue.DeleteParams{ + QueueURL: c.cfg.GetQueueURL(), + ReceiptHandle: message.ReceiptHandle, + }) + if err != nil { + slog.Error("message delete fail", "err", err) + } } return nil diff --git a/scripts/compose.yml b/scripts/compose.yml index 43cfd13f..846d28a3 100644 --- a/scripts/compose.yml +++ b/scripts/compose.yml @@ -11,11 +11,15 @@ tasks: cmds: - docker compose -f {{.COMPOSE_FILE}} build up: - cmds: - - docker compose -f {{.COMPOSE_FILE}} up --no-recreate - up:bg: cmds: - docker compose -f {{.COMPOSE_FILE}} up --no-recreate -d + - task compose:init + init: + cmds: + - aws sqs create-queue --queue-name $QNAME_DOCUMENT_CLEAN + - aws sqs create-queue --queue-name $QNAME_DOCUMENT_INIT + - aws sqs create-queue --queue-name $QNAME_QUERY_RUNNER + - aws s3 mb s3://$BUCKET_IN down: cmds: - docker compose -f {{.COMPOSE_FILE}} down diff --git a/scripts/database.yml b/scripts/database.yml index 42b2e3bb..804d3428 100644 --- a/scripts/database.yml +++ b/scripts/database.yml @@ -13,7 +13,7 @@ tasks: - sqlc generate --file sqlc.yml lint: cmds: - - task compose:up:bg + - task compose:up - sleep 2 - sqlc vet --file sqlc.yml mig:create: @@ -24,6 +24,6 @@ tasks: migrate create -ext sql -dir {{.MIGRATIONS}} $name mig: cmds: - - task compose:up:bg + - task compose:up - sleep 2 - migrate -path {{.MIGRATIONS}} -database {{.DB_URI}} up