connect sqs and log

Author: menduz

package.json

@@ -16,10 +16,13 @@
     "semi": false
   },
   "dependencies": {
+    "@dcl/schemas": "^7.4.1",
     "@well-known-components/env-config-provider": "^1.1.1",
     "@well-known-components/http-server": "^2.0.0",
     "@well-known-components/interfaces": "^1.4.1",
     "@well-known-components/logger": "^3.1.2",
-    "@well-known-components/metrics": "^2.0.1-20220909150423.commit-8f7e5bc"
+    "@well-known-components/metrics": "^2.0.1-20220909150423.commit-8f7e5bc",
+    "@well-known-components/pushable-channel": "^1.0.3",
+    "aws-sdk": "^2.1407.0"
   }
 }

src/adapters/runner.ts

@@ -0,0 +1,32 @@
+import { IBaseComponent } from "@well-known-components/interfaces";
+
+export type RunnerComponentArg = {
+  readonly isRunning: boolean
+}
+
+export type IRunnerComponent = IBaseComponent & {
+  runTask(delegate: (opt: RunnerComponentArg) => Promise<void>): void
+}
+
+// this component runs a loop while the application is enabled. and waits until it
+// finishes the loop to stop the service
+export function createRunnerComponent(): IRunnerComponent {
+
+  let delegates: Promise<any>[] = []
+  let isRunning = false
+
+  return {
+    async start() {
+      if (isRunning) throw new Error('Cannot run twice')
+      isRunning = true
+    },
+    async stop() {
+      isRunning = false
+      await Promise.all(delegates)
+    },
+    runTask(delegate) {
+      if (!isRunning) throw new Error('You can only run tasks while the component is running')
+      delegates.push(delegate({ get isRunning() { return isRunning } }))
+    },
+  }
+}

src/adapters/sqs.ts

@@ -0,0 +1,164 @@
+import { IBaseComponent, IMetricsComponent } from '@well-known-components/interfaces'
+import { validateMetricsDeclaration } from '@well-known-components/metrics'
+import { AsyncQueue } from '@well-known-components/pushable-channel'
+import { SQS } from 'aws-sdk'
+import { AppComponents } from '../types'
+
+export interface TaskQueueMessage {
+  id: string
+}
+
+export interface ITaskQueue<T> {
+  // publishes a job for the queue
+  publish(job: T): Promise<TaskQueueMessage>
+  // awaits for a job. then calls and waits for the taskRunner argument.
+  // the result is then returned to the wrapper function.
+  consumeAndProcessJob<R>(taskRunner: (job: T, message: TaskQueueMessage) => Promise<R>): Promise<{ result: R | undefined }>
+}
+
+export const queueMetrics = validateMetricsDeclaration({
+  job_queue_duration_seconds: {
+    type: IMetricsComponent.HistogramType,
+    help: 'Duration of each job in seconds',
+    labelNames: ["queue_name"],
+    buckets: [1, 10, 100, 200, 300, 400, 500, 600, 700, 1000, 1200, 1600, 1800, 3600]
+  },
+  job_queue_enqueue_total: {
+    type: IMetricsComponent.CounterType,
+    help: "Total amount of enqueued jobs",
+    labelNames: ["queue_name"],
+  },
+  job_queue_failures_total: {
+    type: IMetricsComponent.CounterType,
+    help: "Total amount of failed tasks",
+    labelNames: ["queue_name"],
+  },
+})
+
+type SNSOverSQSMessage = {
+  Message: string
+}
+
+
+export function createMemoryQueueAdapter<T>(components: Pick<AppComponents, "logs" | 'metrics'>, options: { queueName: string }): ITaskQueue<T> & IBaseComponent {
+  type InternalElement = { message: TaskQueueMessage, job: T }
+  const q = new AsyncQueue<InternalElement>((action) => void 0)
+  let lastJobId = 0
+
+  const logger = components.logs.getLogger(options.queueName)
+
+  return {
+    async stop() {
+      q.close()
+    },
+    async publish(job) {
+      const id = 'job-' + (++lastJobId).toString()
+      const message: TaskQueueMessage = { id }
+      q.enqueue({ job, message })
+      logger.info(`Publishing job`, { id })
+      components.metrics.increment('job_queue_enqueue_total', { queue_name: options.queueName })
+      return message
+    },
+    async consumeAndProcessJob(taskRunner) {
+      const it: InternalElement = (await q.next()).value
+      if (it) {
+        const { end } = components.metrics.startTimer('job_queue_duration_seconds', { queue_name: options.queueName })
+        try {
+          logger.info(`Processing job`, { id: it.message.id })
+          const result = await taskRunner(it.job, it.message)
+          logger.info(`Processed job`, { id: it.message.id })
+          return { result, message: it.message }
+        } catch (err: any) {
+          components.metrics.increment('job_queue_failures_total', { queue_name: options.queueName })
+          logger.error(err, { id: it.message.id })
+          // q.enqueue(it)
+        } finally {
+          end()
+        }
+      }
+      return { result: undefined }
+    },
+  }
+}
+
+
+export function createSqsAdapter<T>(components: Pick<AppComponents, "logs" | 'metrics'>, options: { queueUrl: string, queueRegion?: string }): ITaskQueue<T> {
+  const logger = components.logs.getLogger(options.queueUrl)
+
+  const sqs = new SQS({ apiVersion: 'latest', region: options.queueRegion })
+
+  return {
+    async publish(job) {
+      const snsOverSqs: SNSOverSQSMessage = {
+        Message: JSON.stringify(job)
+      }
+
+      const published = await sqs.sendMessage(
+        {
+          QueueUrl: options.queueUrl,
+          MessageBody: JSON.stringify(snsOverSqs),
+        }).promise()
+
+      const m: TaskQueueMessage = { id: published.MessageId! }
+
+      logger.info(`Publishing job`, m as any)
+
+      components.metrics.increment('job_queue_enqueue_total', { queue_name: options.queueUrl })
+      return m
+    },
+    async consumeAndProcessJob(taskRunner) {
+      while (true) {
+        const params: AWS.SQS.ReceiveMessageRequest = {
+          AttributeNames: ['SentTimestamp'],
+          MaxNumberOfMessages: 1,
+          MessageAttributeNames: ['All'],
+          QueueUrl: options.queueUrl,
+          WaitTimeSeconds: 15,
+          VisibilityTimeout: 3 * 3600 // 3 hours
+        }
+
+        try {
+          const response = await Promise.race([
+            sqs.receiveMessage(params).promise(),
+            timeout(30 * 60 * 1000, 'Timed out sqs.receiveMessage')
+          ])
+
+          if (response.Messages && response.Messages.length > 0) {
+            for (const it of response.Messages) {
+              const message: TaskQueueMessage = { id: it.MessageId! }
+              const { end } = components.metrics.startTimer('job_queue_duration_seconds', { queue_name: options.queueUrl })
+              try {
+                const snsOverSqs: SNSOverSQSMessage = JSON.parse(it.Body!)
+                logger.info(`Processing job`, { id: message.id, message: snsOverSqs.Message })
+                const result = await taskRunner(JSON.parse(snsOverSqs.Message), message)
+                logger.info(`Processed job`, { id: message.id })
+                return { result, message }
+              } catch (err: any) {
+                logger.error(err)
+
+                components.metrics.increment('job_queue_failures_total', { queue_name: options.queueUrl })
+
+                return { result: undefined, message }
+              } finally {
+                await sqs.deleteMessage({ QueueUrl: options.queueUrl, ReceiptHandle: it.ReceiptHandle! }).promise()
+                end()
+              }
+            }
+          }
+          logger.info(`No new messages in queue. Retrying for 15 seconds`)
+        } catch (err: any) {
+          logger.error(err)
+          await sleep(1000)
+        }
+      }
+    },
+  }
+}
+
+export async function sleep(ms: number) {
+  return new Promise<void>((ok) => setTimeout(ok, ms))
+}
+
+export async function timeout(ms: number, message: string) {
+  return new Promise<never>((_, reject) => setTimeout(() => reject(new Error(message)), ms))
+}

src/components.ts

@@ -5,24 +5,46 @@ import { createFetchComponent } from "./adapters/fetch"
 import { createMetricsComponent, instrumentHttpServerWithMetrics } from "@well-known-components/metrics"
 import { AppComponents, GlobalContext } from "./types"
 import { metricDeclarations } from "./metrics"
+import { createMemoryQueueAdapter, createSqsAdapter } from "./adapters/sqs"
+import { DeploymentToSqs } from "@dcl/schemas/dist/misc/deployments-to-sqs"
+import AWS from "aws-sdk"
+import { createRunnerComponent } from "./adapters/runner"
+import mitt from "mitt"
 
 // Initialize all the components of the app
 export async function initComponents(): Promise<AppComponents> {
   const config = await createDotEnvConfigComponent({ path: [".env.default", ".env"] })
   const metrics = await createMetricsComponent(metricDeclarations, { config })
   const logs = await createLogComponent({ metrics })
-  const server = await createServerComponent<GlobalContext>({ config, logs }, {})
+  const server = await createServerComponent<GlobalContext>({ config, logs }, {
+    cors: {}
+  })
   const statusChecks = await createStatusCheckComponent({ server, config })
   const fetch = await createFetchComponent()
 
   await instrumentHttpServerWithMetrics({ metrics, server, config })
 
+  const AWS_REGION = await config.getString('AWS_REGION')
+  if (AWS_REGION) {
+    AWS.config.update({ region: AWS_REGION })
+  }
+
+  const sqsQueue = await config.getString('TASK_QUEUE')
+  const taskQueue = sqsQueue ?
+  createSqsAdapter<DeploymentToSqs>({ logs, metrics }, { queueUrl: sqsQueue, queueRegion: AWS_REGION }) :
+  createMemoryQueueAdapter<DeploymentToSqs>({ logs, metrics }, { queueName: "ConversionTaskQueue" })
+
+  const runner = createRunnerComponent()
+
   return {
     config,
     logs,
     server,
     statusChecks,
     fetch,
     metrics,
+    taskQueue,
+    runner,
+    deploymentsByPointer: mitt()
   }
 }

src/metrics.ts

@@ -1,10 +1,12 @@
 import { IMetricsComponent } from "@well-known-components/interfaces"
 import { getDefaultHttpMetrics, validateMetricsDeclaration } from "@well-known-components/metrics"
 import { metricDeclarations as logsMetricsDeclarations } from "@well-known-components/logger"
+import { queueMetrics } from "./adapters/sqs"
 
 export const metricDeclarations = {
   ...getDefaultHttpMetrics(),
   ...logsMetricsDeclarations,
+  ...queueMetrics,
   test_ping_counter: {
     help: "Count calls to ping",
     type: IMetricsComponent.CounterType,

src/service.ts

@@ -20,4 +20,14 @@ export async function main(program: Lifecycle.EntryPointParameters<AppComponents
 
   // start ports: db, listeners, synchronizations, etc
   await startComponents()
+
+  const logger = components.logs.getLogger('main-loop')
+
+  components.runner.runTask(async (opt) => {
+    while (opt.isRunning) {
+      await components.taskQueue.consumeAndProcessJob(async (job, message) => {
+        logger.log("Deployment " + JSON.stringify(job))
+      })
+    }
+  })
 }

src/types.ts

@@ -7,6 +7,10 @@ import type {
   IMetricsComponent,
 } from "@well-known-components/interfaces"
 import { metricDeclarations } from "./metrics"
+import { ITaskQueue } from "./adapters/sqs"
+import { DeploymentToSqs } from "@dcl/schemas/dist/misc/deployments-to-sqs"
+import { IRunnerComponent } from "./adapters/runner"
+import { Emitter } from "mitt"
 
 export type GlobalContext = {
   components: BaseComponents
@@ -19,6 +23,9 @@ export type BaseComponents = {
   server: IHttpServerComponent<GlobalContext>
   fetch: IFetchComponent
   metrics: IMetricsComponent<keyof typeof metricDeclarations>
+  taskQueue: ITaskQueue<DeploymentToSqs>
+  runner: IRunnerComponent
+  deploymentsByPointer: Emitter<Record<string /* pointer */, DeploymentToSqs>>
 }
 
 // components used in runtime

tsconfig.json

@@ -12,6 +12,7 @@
     "noImplicitReturns": true,
     "noImplicitAny": true,
     "noImplicitThis": true,
+    "esModuleInterop": true,
     "forceConsistentCasingInFileNames": true
   },
   "include": [