feat: per-queue drain concurrency variables (#53)

Author: haluz

main.tf

@@ -131,6 +131,15 @@ module "ecs" {
   drain_security_group        = var.drain_security_group_id
   drain_container_definitions = var.drain_container_definitions
 
+  drain_concurrency_async_jobs      = var.drain_concurrency_async_jobs
+  drain_concurrency_async_jobs_fifo = var.drain_concurrency_async_jobs_fifo
+  drain_concurrency_cronjobs        = var.drain_concurrency_cronjobs
+  drain_concurrency_dlq             = var.drain_concurrency_dlq
+  drain_concurrency_dlq_fifo        = var.drain_concurrency_dlq_fifo
+  drain_concurrency_events          = var.drain_concurrency_events
+  drain_concurrency_iot             = var.drain_concurrency_iot
+  drain_concurrency_webhooks        = var.drain_concurrency_webhooks
+
   scheduler_cpu                  = var.scheduler_cpu
   scheduler_desired_count        = var.scheduler_desired_count
   scheduler_log_configuration    = var.scheduler_log_configuration

modules/ecs/container_definitions/drain.tf

@@ -42,6 +42,38 @@ locals {
               {
                 name  = "SPACELIFT_PUBLIC_API"
                 value = var.enable_automatic_usage_data_reporting ? local.spacelift_public_api : ""
+              },
+              {
+                name  = "DRAIN_CONCURRENCY_ASYNC_JOBS"
+                value = tostring(var.drain_concurrency_async_jobs)
+              },
+              {
+                name  = "DRAIN_CONCURRENCY_ASYNC_JOBS_FIFO"
+                value = tostring(var.drain_concurrency_async_jobs_fifo)
+              },
+              {
+                name  = "DRAIN_CONCURRENCY_CRONJOBS"
+                value = tostring(var.drain_concurrency_cronjobs)
+              },
+              {
+                name  = "DRAIN_CONCURRENCY_DLQ"
+                value = tostring(var.drain_concurrency_dlq)
+              },
+              {
+                name  = "DRAIN_CONCURRENCY_DLQ_FIFO"
+                value = tostring(var.drain_concurrency_dlq_fifo)
+              },
+              {
+                name  = "DRAIN_CONCURRENCY_EVENTS"
+                value = tostring(var.drain_concurrency_events)
+              },
+              {
+                name  = "DRAIN_CONCURRENCY_IOT"
+                value = tostring(var.drain_concurrency_iot)
+              },
+              {
+                name  = "DRAIN_CONCURRENCY_WEBHOOKS"
+                value = tostring(var.drain_concurrency_webhooks)
               }
             ],
             var.sqs_queues != null ? [

modules/ecs/container_definitions/variables.tf

@@ -39,6 +39,46 @@ variable "drain_log_configuration" {
   description = "Log configuration for drain container"
 }
 
+variable "drain_concurrency_async_jobs" {
+  type        = number
+  description = "Number of concurrent receivers for the async-jobs queue per drain pod."
+}
+
+variable "drain_concurrency_async_jobs_fifo" {
+  type        = number
+  description = "Number of concurrent receivers for the async-jobs.fifo queue per drain pod."
+}
+
+variable "drain_concurrency_cronjobs" {
+  type        = number
+  description = "Number of concurrent receivers for the cronjobs queue per drain pod."
+}
+
+variable "drain_concurrency_dlq" {
+  type        = number
+  description = "Number of concurrent receivers for the DLQ queue per drain pod."
+}
+
+variable "drain_concurrency_dlq_fifo" {
+  type        = number
+  description = "Number of concurrent receivers for the DLQ FIFO queue per drain pod."
+}
+
+variable "drain_concurrency_events" {
+  type        = number
+  description = "Number of concurrent receivers for the events-inbox queue per drain pod."
+}
+
+variable "drain_concurrency_iot" {
+  type        = number
+  description = "Number of concurrent receivers for the IoT queue per drain pod."
+}
+
+variable "drain_concurrency_webhooks" {
+  type        = number
+  description = "Number of concurrent receivers for the webhooks queue per drain pod."
+}
+
 variable "scheduler_log_configuration" {
   type        = any
   description = "Log configuration for scheduler container"

modules/ecs/task_definitions.tf

@@ -185,6 +185,16 @@ module "container_definitions" {
   vcs_gateway_external_port     = var.vcs_gateway_external_port
   vcs_gateway_internal_port     = var.vcs_gateway_internal_port
 
+  # Drain per-queue concurrency
+  drain_concurrency_async_jobs      = var.drain_concurrency_async_jobs
+  drain_concurrency_async_jobs_fifo = var.drain_concurrency_async_jobs_fifo
+  drain_concurrency_cronjobs        = var.drain_concurrency_cronjobs
+  drain_concurrency_dlq             = var.drain_concurrency_dlq
+  drain_concurrency_dlq_fifo        = var.drain_concurrency_dlq_fifo
+  drain_concurrency_events          = var.drain_concurrency_events
+  drain_concurrency_iot             = var.drain_concurrency_iot
+  drain_concurrency_webhooks        = var.drain_concurrency_webhooks
+
   # Authentication
   admin_username = var.admin_username
   admin_password = var.admin_password

modules/ecs/variables.tf

@@ -124,6 +124,46 @@ variable "drain_security_group" {
   description = "The security group to attach to the drain service."
 }
 
+variable "drain_concurrency_async_jobs" {
+  type        = number
+  description = "Number of concurrent receivers for the async-jobs queue per drain pod."
+}
+
+variable "drain_concurrency_async_jobs_fifo" {
+  type        = number
+  description = "Number of concurrent receivers for the async-jobs.fifo queue per drain pod."
+}
+
+variable "drain_concurrency_cronjobs" {
+  type        = number
+  description = "Number of concurrent receivers for the cronjobs queue per drain pod."
+}
+
+variable "drain_concurrency_dlq" {
+  type        = number
+  description = "Number of concurrent receivers for the DLQ queue per drain pod."
+}
+
+variable "drain_concurrency_dlq_fifo" {
+  type        = number
+  description = "Number of concurrent receivers for the DLQ FIFO queue per drain pod."
+}
+
+variable "drain_concurrency_events" {
+  type        = number
+  description = "Number of concurrent receivers for the events-inbox queue per drain pod."
+}
+
+variable "drain_concurrency_iot" {
+  type        = number
+  description = "Number of concurrent receivers for the IoT queue per drain pod."
+}
+
+variable "drain_concurrency_webhooks" {
+  type        = number
+  description = "Number of concurrent receivers for the webhooks queue per drain pod."
+}
+
 variable "server_security_group" {
   type        = string
   description = "The security group to attach to the server service."

variables.tf

@@ -257,6 +257,54 @@ variable "drain_security_group_id" {
   description = "The security group ID to use for the drain service."
 }
 
+variable "drain_concurrency_async_jobs" {
+  type        = number
+  description = "Number of concurrent receivers for the async-jobs queue per drain pod."
+  default     = 1
+}
+
+variable "drain_concurrency_async_jobs_fifo" {
+  type        = number
+  description = "Number of concurrent receivers for the async-jobs.fifo queue per drain pod."
+  default     = 1
+}
+
+variable "drain_concurrency_cronjobs" {
+  type        = number
+  description = "Number of concurrent receivers for the cronjobs queue per drain pod."
+  default     = 1
+}
+
+variable "drain_concurrency_dlq" {
+  type        = number
+  description = "Number of concurrent receivers for the DLQ queue per drain pod."
+  default     = 1
+}
+
+variable "drain_concurrency_dlq_fifo" {
+  type        = number
+  description = "Number of concurrent receivers for the DLQ FIFO queue per drain pod."
+  default     = 1
+}
+
+variable "drain_concurrency_events" {
+  type        = number
+  description = "Number of concurrent receivers for the events-inbox queue per drain pod."
+  default     = 1
+}
+
+variable "drain_concurrency_iot" {
+  type        = number
+  description = "Number of concurrent receivers for the IoT queue per drain pod."
+  default     = 1
+}
+
+variable "drain_concurrency_webhooks" {
+  type        = number
+  description = "Number of concurrent receivers for the webhooks queue per drain pod."
+  default     = 1
+}
+
 variable "scheduler_container_definition" {
   type        = string
   description = "The container definition for the scheduler service. If empty, a default container definition will be used."