From 9e2863eb15e9fc2cf631b1b8eae57af720a2b3eb Mon Sep 17 00:00:00 2001 From: euniceamoni Date: Sat, 27 Jun 2026 22:12:03 +0000 Subject: [PATCH 1/3] fix: scope VPC security groups to minimum required ports (#960) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Move ecs_tasks SG to root module to avoid circular dependencies - ALB SG: restrict egress to container port → ecs_tasks SG only - ecs_tasks SG: egress limited to 5432 (RDS), 6379 (Redis), 443 (AWS APIs) - RDS SG: replace 10.0.0.0/8 ingress with ecs_tasks SG reference, remove broad egress - Redis SG: replace 10.0.0.0/8 ingress with ecs_tasks SG reference, remove broad egress - Add Checkov CI scan job that fails on HIGH/CRITICAL findings before terraform plan --- .github/workflows/deploy.yml | 22 +++++++- infrastructure/terraform/main.tf | 50 +++++++++++++++++ infrastructure/terraform/modules/ecs/main.tf | 54 ++++++++----------- infrastructure/terraform/modules/rds/main.tf | 25 +++++---- .../terraform/modules/redis/main.tf | 25 +++++---- 5 files changed, 122 insertions(+), 54 deletions(-) diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 45b688d7..392b8e07 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -50,10 +50,30 @@ jobs: echo "✅ All required secrets are configured" + security-scan: + name: Checkov Security Scan + runs-on: ubuntu-latest + needs: validate-secrets + steps: + - uses: actions/checkout@v4 + + - name: Run Checkov + uses: bridgecrewio/checkov-action@v12 + with: + directory: infrastructure/terraform + framework: terraform + # Fail on HIGH and CRITICAL severity findings only + soft_fail_on: LOW,MEDIUM + output_format: cli + # Skip checks that are intentional/acceptable in this architecture: + # CKV_AWS_91 - ALB access logs (cost tradeoff, tracked separately) + # CKV2_AWS_28 - WAF on ALB (tracked separately) + skip_check: CKV_AWS_91,CKV2_AWS_28 + plan: name: Terraform Plan runs-on: ubuntu-latest - needs: validate-secrets + needs: [validate-secrets, security-scan] strategy: matrix: environment: [dev, staging, prod] diff --git a/infrastructure/terraform/main.tf b/infrastructure/terraform/main.tf index 6b61c450..35e5b03e 100644 --- a/infrastructure/terraform/main.tf +++ b/infrastructure/terraform/main.tf @@ -25,6 +25,53 @@ provider "aws" { } } +# The ECS tasks SG is created at root level to break the circular dependency +# between the ecs module (which needs the ALB SG to wire the ingress rule) and +# rds/redis modules (which need this SG ID for their ingress rules). +# Egress rules are added separately via aws_security_group_rule so that they +# can reference the rds/redis module outputs without creating a cycle. +resource "aws_security_group" "ecs_tasks" { + name = "predictiq-${var.environment}-ecs-tasks-sg" + vpc_id = module.vpc.vpc_id + + tags = { + Name = "predictiq-${var.environment}-ecs-tasks-sg" + Project = "predictiq" + Environment = var.environment + ManagedBy = "terraform" + } +} + +# Outbound to RDS (PostgreSQL) +resource "aws_security_group_rule" "ecs_tasks_egress_rds" { + type = "egress" + from_port = 5432 + to_port = 5432 + protocol = "tcp" + security_group_id = aws_security_group.ecs_tasks.id + source_security_group_id = module.rds.sg_id +} + +# Outbound to Redis +resource "aws_security_group_rule" "ecs_tasks_egress_redis" { + type = "egress" + from_port = 6379 + to_port = 6379 + protocol = "tcp" + security_group_id = aws_security_group.ecs_tasks.id + source_security_group_id = module.redis.sg_id +} + +# Outbound HTTPS for AWS API calls (Secrets Manager, ECR, CloudWatch) +resource "aws_security_group_rule" "ecs_tasks_egress_https" { + type = "egress" + from_port = 443 + to_port = 443 + protocol = "tcp" + security_group_id = aws_security_group.ecs_tasks.id + cidr_blocks = ["0.0.0.0/0"] +} + module "vpc" { source = "./modules/vpc" @@ -44,6 +91,7 @@ module "rds" { db_instance_class = var.db_instance_class allocated_storage = var.allocated_storage backup_retention = var.backup_retention_days + ecs_tasks_sg_id = aws_security_group.ecs_tasks.id } module "redis" { @@ -55,6 +103,7 @@ module "redis" { node_type = var.redis_node_type num_cache_nodes = var.redis_num_nodes engine_version = var.redis_engine_version + ecs_tasks_sg_id = aws_security_group.ecs_tasks.id } module "ecs" { @@ -71,6 +120,7 @@ module "ecs" { api_memory = var.api_memory database_url = module.rds.database_url redis_url = module.redis.redis_url + ecs_tasks_sg_id = aws_security_group.ecs_tasks.id } module "monitoring" { diff --git a/infrastructure/terraform/modules/ecs/main.tf b/infrastructure/terraform/modules/ecs/main.tf index da4db3fb..cc500252 100644 --- a/infrastructure/terraform/modules/ecs/main.tf +++ b/infrastructure/terraform/modules/ecs/main.tf @@ -44,6 +44,11 @@ variable "redis_url" { sensitive = true } +variable "ecs_tasks_sg_id" { + type = string + description = "Security group ID of the ECS tasks (managed at root level)" +} + locals { common_tags = { Project = "predictiq" @@ -141,6 +146,7 @@ resource "aws_security_group" "alb" { name = "predictiq-${var.environment}-alb-sg" vpc_id = var.vpc_id + # Allow inbound HTTP/HTTPS from the public internet ingress { from_port = 80 to_port = 80 @@ -155,11 +161,12 @@ resource "aws_security_group" "alb" { cidr_blocks = ["0.0.0.0/0"] } + # Restrict egress to the container port on ECS tasks only egress { - from_port = 0 - to_port = 0 - protocol = "-1" - cidr_blocks = ["0.0.0.0/0"] + from_port = var.api_container_port + to_port = var.api_container_port + protocol = "tcp" + security_groups = [var.ecs_tasks_sg_id] } tags = merge( @@ -170,6 +177,17 @@ resource "aws_security_group" "alb" { ) } +# Allow inbound from the ALB on the container port — added as a rule on the +# externally-managed ecs_tasks SG to avoid a circular module dependency. +resource "aws_security_group_rule" "ecs_tasks_ingress_alb" { + type = "ingress" + from_port = var.api_container_port + to_port = var.api_container_port + protocol = "tcp" + security_group_id = var.ecs_tasks_sg_id + source_security_group_id = aws_security_group.alb.id +} + resource "aws_lb" "main" { name = "predictiq-${var.environment}-alb" internal = false @@ -220,32 +238,6 @@ resource "aws_lb_listener" "api" { } } -resource "aws_security_group" "ecs_tasks" { - name = "predictiq-${var.environment}-ecs-tasks-sg" - vpc_id = var.vpc_id - - ingress { - from_port = var.api_container_port - to_port = var.api_container_port - protocol = "tcp" - security_groups = [aws_security_group.alb.id] - } - - egress { - from_port = 0 - to_port = 0 - protocol = "-1" - cidr_blocks = ["0.0.0.0/0"] - } - - tags = merge( - local.common_tags, - { - Name = "predictiq-${var.environment}-ecs-tasks-sg" - } - ) -} - resource "aws_ecs_service" "api" { name = "predictiq-${var.environment}-api" cluster = aws_ecs_cluster.main.id @@ -255,7 +247,7 @@ resource "aws_ecs_service" "api" { network_configuration { subnets = var.private_subnet_ids - security_groups = [aws_security_group.ecs_tasks.id] + security_groups = [var.ecs_tasks_sg_id] assign_public_ip = false } diff --git a/infrastructure/terraform/modules/rds/main.tf b/infrastructure/terraform/modules/rds/main.tf index d94e20df..7b2d8b97 100644 --- a/infrastructure/terraform/modules/rds/main.tf +++ b/infrastructure/terraform/modules/rds/main.tf @@ -36,6 +36,11 @@ variable "backup_retention" { type = number } +variable "ecs_tasks_sg_id" { + type = string + description = "Security group ID of the ECS tasks that are allowed to connect" +} + locals { common_tags = { Project = "predictiq" @@ -61,18 +66,12 @@ resource "aws_security_group" "rds" { name = "predictiq-${var.environment}-rds-sg" vpc_id = var.vpc_id + # Inbound PostgreSQL from ECS tasks only ingress { - from_port = 5432 - to_port = 5432 - protocol = "tcp" - cidr_blocks = ["10.0.0.0/8"] - } - - egress { - from_port = 0 - to_port = 0 - protocol = "-1" - cidr_blocks = ["0.0.0.0/0"] + from_port = 5432 + to_port = 5432 + protocol = "tcp" + security_groups = [var.ecs_tasks_sg_id] } tags = merge( @@ -113,6 +112,10 @@ resource "aws_db_instance" "main" { ) } +output "sg_id" { + value = aws_security_group.rds.id +} + output "endpoint" { value = aws_db_instance.main.endpoint sensitive = true diff --git a/infrastructure/terraform/modules/redis/main.tf b/infrastructure/terraform/modules/redis/main.tf index b8518c71..9abd1128 100644 --- a/infrastructure/terraform/modules/redis/main.tf +++ b/infrastructure/terraform/modules/redis/main.tf @@ -22,6 +22,11 @@ variable "engine_version" { type = string } +variable "ecs_tasks_sg_id" { + type = string + description = "Security group ID of the ECS tasks that are allowed to connect" +} + locals { common_tags = { Project = "predictiq" @@ -47,18 +52,12 @@ resource "aws_security_group" "redis" { name = "predictiq-${var.environment}-redis-sg" vpc_id = var.vpc_id + # Inbound Redis from ECS tasks only ingress { - from_port = 6379 - to_port = 6379 - protocol = "tcp" - cidr_blocks = ["10.0.0.0/8"] - } - - egress { - from_port = 0 - to_port = 0 - protocol = "-1" - cidr_blocks = ["0.0.0.0/0"] + from_port = 6379 + to_port = 6379 + protocol = "tcp" + security_groups = [var.ecs_tasks_sg_id] } tags = merge( @@ -95,6 +94,10 @@ resource "aws_elasticache_cluster" "main" { ) } +output "sg_id" { + value = aws_security_group.redis.id +} + output "endpoint" { value = aws_elasticache_cluster.main.cache_nodes[0].address } From ff7b787908ab0bbf5f555285f045fd47809aadc6 Mon Sep 17 00:00:00 2001 From: euniceamoni Date: Sat, 27 Jun 2026 22:28:34 +0000 Subject: [PATCH 2/3] feat: expose email_queue_depth as Prometheus metric with Grafana panel and alert - Add email_queue_depth IntGauge to Metrics struct - Emit the gauge in the email worker loop after each dequeue cycle - Also update the gauge on each /api/v1/email/queue/stats request - Add Grafana panel for queue depth visualization - Add Prometheus alert when queue depth exceeds 100 for 5 minutes Resolves #961 --- performance/config/alerts.yaml | 14 +++++++ performance/config/grafana-dashboard.json | 46 +++++++++++++++++++++++ services/api/src/email/queue.rs | 18 +++++++++ services/api/src/handlers.rs | 1 + services/api/src/main.rs | 3 +- services/api/src/metrics.rs | 13 +++++++ 6 files changed, 94 insertions(+), 1 deletion(-) diff --git a/performance/config/alerts.yaml b/performance/config/alerts.yaml index 778ff282..c3905c19 100644 --- a/performance/config/alerts.yaml +++ b/performance/config/alerts.yaml @@ -259,3 +259,17 @@ groups: summary: "Performance degradation alert threshold reached" description: "Response time has degraded by more than 5% compared to 24h ago (alert threshold)" runbook_url: "https://docs.predictiq.com/runbooks/significant-performance-degradation" + + - name: email_queue + interval: 30s + rules: + - alert: EmailQueueDepthHigh + expr: email_queue_depth > 100 + for: 5m + labels: + severity: warning + component: email + annotations: + summary: "Email queue depth is high" + description: "Email queue depth is {{ $value }}, exceeding 100 for more than 5 minutes" + runbook_url: "https://docs.predictiq.com/runbooks/email-queue-depth-high" diff --git a/performance/config/grafana-dashboard.json b/performance/config/grafana-dashboard.json index 3810f27c..27dcecef 100644 --- a/performance/config/grafana-dashboard.json +++ b/performance/config/grafana-dashboard.json @@ -450,6 +450,52 @@ "notifications": [] } }, + { + "id": 11, + "title": "Email Queue Depth", + "type": "graph", + "gridPos": { "x": 0, "y": 36, "w": 12, "h": 8 }, + "targets": [ + { + "expr": "email_queue_depth", + "legendFormat": "Queue Depth", + "refId": "A" + } + ], + "yaxes": [ + { + "format": "short", + "label": "Jobs" + } + ], + "alert": { + "name": "Email Queue Depth High", + "conditions": [ + { + "evaluator": { + "params": [100], + "type": "gt" + }, + "operator": { + "type": "and" + }, + "query": { + "params": ["A", "5m", "now"] + }, + "reducer": { + "params": [], + "type": "avg" + }, + "type": "query" + } + ], + "executionErrorState": "alerting", + "frequency": "1m", + "message": "Email queue depth exceeded 100 for more than 5 minutes", + "noDataState": "no_data", + "notifications": [] + } + }, { "id": 9, "title": "System Health Overview", diff --git a/services/api/src/email/queue.rs b/services/api/src/email/queue.rs index 44c78e01..7bac43bf 100644 --- a/services/api/src/email/queue.rs +++ b/services/api/src/email/queue.rs @@ -10,6 +10,7 @@ use crate::cache::RedisCache; use crate::db::Database; use crate::email::service::idempotency_key; use crate::email::types::{EmailJobStatus, EmailJobType}; +use crate::metrics::Metrics; use crate::shutdown::ShutdownCoordinator; const EMAIL_QUEUE_KEY: &str = "email:queue"; @@ -345,6 +346,16 @@ impl EmailQueue { }) } + /// Get the current depth of the main email queue. + pub async fn get_queue_depth(&self) -> Result { + let mut conn = self.cache.get_connection().await?; + let depth: usize = conn + .zcard(EMAIL_QUEUE_KEY) + .await + .context("Failed to get queue depth")?; + Ok(depth) + } + /// Re-queue any jobs stuck in the processing set (e.g. from a previous crash). /// /// Recovers jobs that have been in processing longer than the configured @@ -416,6 +427,7 @@ impl EmailQueue { shutdown: CancellationToken, coordinator: ShutdownCoordinator, stale_job_threshold_secs: u64, + metrics: Option, ) { tracing::info!("Email queue worker started"); @@ -464,6 +476,12 @@ impl EmailQueue { } } } + + if let Some(ref m) = metrics { + if let Ok(depth) = self.get_queue_depth().await { + m.set_email_queue_depth(depth as i64); + } + } } tracing::info!("Email queue worker stopped"); diff --git a/services/api/src/handlers.rs b/services/api/src/handlers.rs index bbea7005..7002c47d 100644 --- a/services/api/src/handlers.rs +++ b/services/api/src/handlers.rs @@ -987,6 +987,7 @@ pub async fn email_queue_stats( .map_err(into_api_error)?; state.metrics.set_dlq_size(stats.dead_letter as i64); + state.metrics.set_email_queue_depth(stats.pending as i64); Ok((StatusCode::OK, Json(stats))) } diff --git a/services/api/src/main.rs b/services/api/src/main.rs index 115a5963..bac1ff90 100644 --- a/services/api/src/main.rs +++ b/services/api/src/main.rs @@ -173,9 +173,10 @@ async fn main() -> anyhow::Result<()> { let email_token = email_coordinator.token(); let email_coord = email_coordinator.clone(); let stale_threshold = state.config.email_stale_job_threshold_secs; + let metrics_worker = metrics.clone(); tokio::spawn(async move { queue_worker - .start_worker(service_worker, email_token, email_coord, stale_threshold) + .start_worker(service_worker, email_token, email_coord, stale_threshold, Some(metrics_worker)) .await; }); diff --git a/services/api/src/metrics.rs b/services/api/src/metrics.rs index e9c8b542..9fed3b50 100644 --- a/services/api/src/metrics.rs +++ b/services/api/src/metrics.rs @@ -14,6 +14,7 @@ pub struct Metrics { rpc_fallbacks: IntCounterVec, db_timeouts: IntCounterVec, email_dlq_size: IntGauge, + email_queue_depth: IntGauge, db_pool_connections_active: IntGaugeVec, db_pool_connections_idle: IntGaugeVec, db_pool_acquire_duration: HistogramVec, @@ -81,6 +82,12 @@ impl Metrics { ) .context("email_dlq_size metric")?; + let email_queue_depth = IntGauge::new( + "email_queue_depth", + "Number of email jobs currently in the main queue", + ) + .context("email_queue_depth metric")?; + let db_pool_connections_active = IntGaugeVec::new( prometheus::Opts::new( "db_pool_connections_active", @@ -128,6 +135,7 @@ impl Metrics { registry.register(Box::new(rpc_fallbacks.clone()))?; registry.register(Box::new(db_timeouts.clone()))?; registry.register(Box::new(email_dlq_size.clone()))?; + registry.register(Box::new(email_queue_depth.clone()))?; registry.register(Box::new(db_pool_connections_active.clone()))?; registry.register(Box::new(db_pool_connections_idle.clone()))?; registry.register(Box::new(db_pool_acquire_duration.clone()))?; @@ -143,6 +151,7 @@ impl Metrics { rpc_fallbacks, db_timeouts, email_dlq_size, + email_queue_depth, db_pool_connections_active, db_pool_connections_idle, db_pool_acquire_duration, @@ -191,6 +200,10 @@ impl Metrics { self.email_dlq_size.set(n); } + pub fn set_email_queue_depth(&self, n: i64) { + self.email_queue_depth.set(n); + } + pub fn observe_tx_eviction(&self, count: u64) { if count > 0 { self.invalidations From f20416f297020783ad1426aba72b8fbe84767d77 Mon Sep 17 00:00:00 2001 From: euniceamoni Date: Sat, 27 Jun 2026 22:40:04 +0000 Subject: [PATCH 3/3] fix: add CacheCircuitBreakerOpen Prometheus alert - Add cache_circuit_breaker_state IntGauge metric (0=closed, 1=open, 2=half_open) - Update health handler to set circuit breaker state metric - Add CacheCircuitBreakerOpen alert with severity page and 2m for duration Fixes #966 --- performance/config/alerts.yaml | 11 +++++++++++ services/api/src/handlers.rs | 11 +++++++---- services/api/src/metrics.rs | 13 +++++++++++++ 3 files changed, 31 insertions(+), 4 deletions(-) diff --git a/performance/config/alerts.yaml b/performance/config/alerts.yaml index c3905c19..5b8ced37 100644 --- a/performance/config/alerts.yaml +++ b/performance/config/alerts.yaml @@ -63,6 +63,17 @@ groups: description: "Cache hit rate is {{ $value | humanizePercentage }}, below 80% threshold" runbook_url: "https://docs.predictiq.com/runbooks/low-cache-hit-rate" + - alert: CacheCircuitBreakerOpen + expr: cache_circuit_breaker_state == 1 + for: 2m + labels: + severity: page + component: cache + annotations: + summary: "Redis cache circuit breaker is open" + description: "The Redis cache circuit breaker has been open for more than 2 minutes. All cache reads are failing open, causing increased load on the database and RPC endpoints. Immediate operator response required." + runbook_url: "https://docs.predictiq.com/runbooks/cache-circuit-breaker-open" + - name: database_performance interval: 30s rules: diff --git a/services/api/src/handlers.rs b/services/api/src/handlers.rs index 7002c47d..7c5d9733 100644 --- a/services/api/src/handlers.rs +++ b/services/api/src/handlers.rs @@ -119,12 +119,15 @@ pub async fn health(State(state): State>, headers: HeaderMap) -> i .and_then(|v| v.to_str().ok()) .unwrap_or("-"); - let cb_state = match state.cache.circuit_state() { - CircuitState::Closed => "closed", - CircuitState::Open => "open", - CircuitState::HalfOpen => "half_open", + let (cb_state, cb_state_val) = match state.cache.circuit_state() { + CircuitState::Closed => ("closed", 0), + CircuitState::Open => ("open", 1), + CircuitState::HalfOpen => ("half_open", 2), }; let pool = state.cache.pool_status(); + state + .metrics + .set_circuit_breaker_state(cb_state_val); let mut health_status = serde_json::json!({ "status": "ok", diff --git a/services/api/src/metrics.rs b/services/api/src/metrics.rs index 9fed3b50..15ec81ff 100644 --- a/services/api/src/metrics.rs +++ b/services/api/src/metrics.rs @@ -19,6 +19,7 @@ pub struct Metrics { db_pool_connections_idle: IntGaugeVec, db_pool_acquire_duration: HistogramVec, rate_limit_rejections: IntCounterVec, + cache_circuit_breaker_state: IntGauge, } impl Metrics { @@ -127,6 +128,12 @@ impl Metrics { ) .context("rate_limit_rejections metric")?; + let cache_circuit_breaker_state = IntGauge::new( + "cache_circuit_breaker_state", + "Current Redis cache circuit breaker state (0=closed, 1=open, 2=half_open)", + ) + .context("cache_circuit_breaker_state metric")?; + registry.register(Box::new(cache_hits.clone()))?; registry.register(Box::new(cache_misses.clone()))?; registry.register(Box::new(invalidations.clone()))?; @@ -140,6 +147,7 @@ impl Metrics { registry.register(Box::new(db_pool_connections_idle.clone()))?; registry.register(Box::new(db_pool_acquire_duration.clone()))?; registry.register(Box::new(rate_limit_rejections.clone()))?; + registry.register(Box::new(cache_circuit_breaker_state.clone()))?; Ok(Self { registry, @@ -156,6 +164,7 @@ impl Metrics { db_pool_connections_idle, db_pool_acquire_duration, rate_limit_rejections, + cache_circuit_breaker_state, }) } @@ -239,6 +248,10 @@ impl Metrics { .inc(); } + pub fn set_circuit_breaker_state(&self, state: i64) { + self.cache_circuit_breaker_state.set(state); + } + pub fn render(&self) -> anyhow::Result { let mut buffer = vec![]; let encoder = TextEncoder::new();