Infrastructure Scaling, Capacity Planning, and High Availability Questions

Translate the 99% freshness SLO (data available within 1 hour of ingestion) into measurable capacity, alerts, and runbook steps as follows.

Capacity thresholds

• Baseline: measure current ingest rate, median and 95th/99th pipeline latency. Set capacity to handle 2x peak ingestion for an hour without latency >1h.
• Thresholds:
  • Green: 95th latency <30m and system CPU/memory <60%, backlog <5% of hourly volume.
  • Yellow (warning): 95th latency 30–45m OR backlog 5–15% OR CPU 60–80%.
  • Red (breach risk): 95th latency 45–60m OR backlog >15% OR CPU >80% — imminent SLO failure.

Monitoring & alerts

• Synthetic freshness metric: percent of records with ingestion-to-availability <=1h, evaluated per 10m window. Alert when rolling 1h value <99% (page on <98%, notify on 98–99%).
• Leading indicators (early warnings): growing ingest backlog (records/minute), increasing task retry rates, rising processing queue length, GC/pod restarts, consumer lag (Kafka) >50k.
• Alert types:
  • Pager (P1): synthetic freshness <98% or red thresholds triggered.
  • Alert (P2): yellow thresholds or any leading indicator exceeding warning levels.
  • Info: transient spikes cleared within 10m.

Automatic remediation

• Autoscale workers (spark executors / consumer instances) when backlog >10% or consumer lag rising — add nodes up to predefined cap.
• Throttle noncritical workloads (backfill/analytics) via rate limits when CPU >75% or latency approaches 45m.
• Circuit-breaker to pause retries if error rates spike; route failing messages to dead-letter for manual inspection.
• Fall-back: switch to warm standby pipeline (simpler transformation) to restore freshness quickly.

On-call playbook (P1)

1. Acknowledge within 5 minutes; confirm metric (synthetic freshness & leading indicators).
2. Check ingest sources, broker health (Kafka partition leaders, broker CPU), and consumer lag.
3. If autoscale didn’t trigger, trigger manual scale-up; monitor freshness delta for 5–10m.
4. If resource saturation persists, throttle noncritical jobs and enable standby pipeline.
5. If data corruption/errors present, pause affected source, route to DLQ, and trigger data owner.
6. Post-incident: run root-cause, adjust thresholds, and increase capacity or optimize transforms as permanent fix.

Metrics to report post-incident: SLO breach duration, % records late, root cause, remediation applied, and plan to prevent recurrence.

To estimate monthly vCPU-hours: compute how many vCPUs are needed at peak QPS given the CPU time per request, then multiply by hours per month.

def estimate_monthly_compute_hours(peak_qps, avg_cpu_per_request_ms, avg_request_duration_ms=None, hours_per_month=730):
    """
    Estimate total vCPU-hours per month.

    Assumptions:
    - Single-threaded vCPU executes one request at a time.
    - We ignore IO waits, so only CPU time matters -> use avg_cpu_per_request_ms.
    - avg_request_duration_ms is optional and kept for clarity but not used when ignoring IO.
    - peak_qps is the maximum concurrent requests per second to plan for.

    Formula:
    vCPUs_needed = peak_qps * (avg_cpu_per_request_ms / 1000)
    monthly_vcpu_hours = vCPUs_needed * hours_per_month

    Returns float vCPU-hours for the month.
    """
    if peak_qps < 0 or avg_cpu_per_request_ms < 0 or hours_per_month <= 0:
        raise ValueError("Inputs must be non-negative and hours_per_month > 0")

    # Convert CPU-ms to seconds-per-request, multiply by requests/sec to get vCPUs
    vcpus_needed = peak_qps * (avg_cpu_per_request_ms / 1000.0)

    monthly_vcpu_hours = vcpus_needed * hours_per_month
    return monthly_vcpu_hours

Examples:

• peak_qps=100, avg_cpu_per_request_ms=50 -> vCPUs = 100 * 0.05 = 5 vCPUs; monthly = 5 * 730 = 3650 vCPU-hours.
• peak_qps=1000, avg_cpu_per_request_ms=20 -> 10000.02=20 vCPUs -> 20730=14600 vCPU-hours.

Key notes:

• If you wanted to include concurrency effects from request duration (wall time) or IO waits, use avg_request_duration_ms instead or combine both to model blocking vs CPU-bound behavior.
• Always validate units (ms vs s) and plan for headroom (e.g., +20% buffer). Complexity O(1), constant time and space.

Requirements & constraints:

• Local pre-processing (transform/filter/enrich) in nearest region with low latency (tens–hundreds ms).
• Global aggregates must converge and be consistent within 5 minutes.
• Multi-region fault tolerance, cost-aware bandwidth, resilient to region surge events.
• Prefer cloud-managed services where possible (Kinesis/PubSub/Event Hubs, Kafka, managed DBs).

High-level architecture:
Clients → Regional Ingest (API Gateway / Edge) → Regional Pre-process (stream processors e.g., Flink/Beam/Managed KDA) → Regional buffer (partitioned Kafka/topic or cloud stream) → Cross-region replication → Global Aggregation Service → Global store (OLAP/time-series) → Consumers

Local vs Global scaling triggers:

• Local scaling (per region)
  • Input rate per partition, CPU/memory per processor, input queue lag (consumer lag), request latency.
  • Autoscale rules: scale out when lag > threshold or CPU>70% for X minutes; scale in when lag drops and CPU<40%.
  • Fast-reacting horizontal scale (container group, scaling Kafka consumers) + warm pool of standby workers to reduce cold start.
• Global scaling (cross-region/global aggregators)
  • Triggered by aggregate backlog across regions, increased inter-region replication lag, or global aggregate error rates.
  • Scale global aggregator horizontally when global window join/aggregation latency approaches 5-minute SLA or when outstanding backlog > capacity threshold.

Cross-region replication architecture:

• Use durable append-only streams (Kafka with MirrorMaker 2 / Confluent Replicator or cloud native replication).
• Replicate only compacted topics or pre-aggregated deltas to reduce bandwidth.
• Send per-region deltas (e.g., incremental partial aggregates) rather than raw event fan-out to all regions.
• Use per-tenant/partition routing to ensure balanced replication; use compression and batching.

Network cost & bandwidth implications:

• Minimize cross-region bytes: replicate compacted summaries or sketches (HyperLogLog, Count-Min Sketch) rather than raw events.
• Batch and compress replication (snappy/gzip), use protobuf/avro.
• Estimate bandwidth: if raw = 100MB/s across 5 regions → 100MB/s * 4 replicates = 400MB/s exit traffic → costly. Instead pre-aggregate to 5MB/s of deltas reduces cost 20x.
• Monitor egress costs and set QoS: backpressure to local buffers when upstream throughput constrained.

Leader election for global aggregation:

• Use a consensus system (Zookeeper/etcd/managed consul) or cloud leader-election (e.g., DynamoDB conditional writes) to pick regional leader(s).
• Design: partition global space (shards) and elect leader per shard to avoid single global leader bottleneck.
• Leaders receive replicated deltas, perform final merge/resolve conflicts, publish global aggregates.
• Use epoch-based merges and idempotent writes; include vector clocks/timestamps to handle late-arriving events.
• If leader fails, quick re-election (<30s) with state checkpointing (snapshot offsets) to resume.

Consistency model & convergence:

• Use eventual consistency with bounded convergence: guarantee all region deltas applied within 5 minutes.
• Use windowed aggregation with watermarking: accept late events up to allowed lateness (e.g., 3–4 minutes) and emit final result at 5 minutes.
• For strong invariants, use CRDTs or commutative-associative reduce functions so merges are deterministic.

Handling extreme region surge:

• Local strategies:
  • Autoscale aggressively within region (with budgeted burst capacity).
  • Apply ingestion shedding: prioritized tenants/events; soft-quota enforcement; graceful degradation of non-critical enrichment.
  • Buffer to durable storage (S3/GCS) and process asynchronously if compute saturated.
• Cross-region strategies:
  • Offload some processing to other regions if network/latency allows: send raw/partial events to global aggregators or neighboring region workers.
  • Increase replication frequency of deltas to leaders, but prefer sending compressed summaries.
  • Trigger temporary leader re-assignment if surge region leader can't keep up.
• Operational guardrails:
  • Circuit breakers to avoid overwhelming global aggregator.
  • Auto-throttle per-client/API key.
  • Alerting/Runbooks + automated rollback of non-essential features.

Metrics & SLOs to monitor:

• Regional ingestion rate, consumer lag, processing latency percentiles, replication lag, egress bytes, aggregate convergence time, error rates.
• Automate cost-aware policies: cap replication rate if egress cost breaches threshold.

Trade-offs:

• Full raw replication: simpler, but high bandwidth/cost and slower.
• Summaries/CRDTs: lower cost, faster convergence, but limited queryability of raw events.
• Sharded leaders reduce contention but add complexity for cross-shard joins.

This design prioritizes low-latency local processing, bounded global convergence using compact deltas and CRDT-style merges, pragmatic autoscaling triggers at local and global tiers, and surge-handling through buffering, shedding, and cross-region offload to meet the 5-minute consistency objective while controlling cost.

Approach: create a reusable Terraform module that builds a Launch Template (mixed instance types + spot overrides), an Auto Scaling Group with MixedInstancesPolicy, IAM role/policy for instance actions (SSM, cloudwatch, drain script), a Lifecycle Hook that triggers a Lambda (or SSM Run Command) to gracefully drain Spark executors before EC2 termination.

Example key snippets (conceptual):

Inputs:

• name_prefix, vpc_subnet_ids, security_group_ids
• launch_instance_types (list), on_demand_percentage, spot_allocation_strategy
• ami_id, instance_profile_name, spark_master_endpoint, drain_timeout_seconds
• desired_count, min_size, max_size, instance_types_overrides

Outputs:

• asg_id, launch_template_id, autoscaling_group_name, iam_instance_profile

Key resources (conceptual HCL):

resource "aws_launch_template" "spark_workers" {
  name_prefix = "${var.name_prefix}-lt-"
  image_id    = var.ami_id
  instance_type = var.default_instance_type
  iam_instance_profile { name = aws_iam_instance_profile.spark.name }
  user_data = base64encode(templatefile("${path.module}/userdata/drain.sh.tpl", {
    spark_master = var.spark_master_endpoint
  }))
  tag_specifications { ... }
}

resource "aws_autoscaling_group" "spark_workers" {
  name                 = "${var.name_prefix}-asg"
  max_size             = var.max_size
  min_size             = var.min_size
  desired_capacity     = var.desired_count
  mixed_instances_policy {
    launch_template {
      launch_template_specification { launch_template_id = aws_launch_template.spark_workers.id, version = "$Latest" }
    }
    instances_distribution {
      on_demand_percentage_above_base_capacity = var.on_demand_percentage
      spot_allocation_strategy = var.spot_allocation_strategy
    }
    override = [
      for t in var.launch_instance_types : { instance_type = t }
    ]
  }
  vpc_zone_identifier = var.vpc_subnet_ids
  tag { key = "Name" value = "${var.name_prefix}-worker" propagate_at_launch = true }
  lifecycle {
    create_before_destroy = true
  }
}

resource "aws_iam_role" "instance" { ... }          # allow ssm:SendCommand, ec2:Describe*, cloudwatch logs
resource "aws_iam_instance_profile" "spark" { role = aws_iam_role.instance.name }

resource "aws_autoscaling_lifecycle_hook" "pre_terminate" {
  name                   = "${var.name_prefix}-drain-hook"
  autoscaling_group_name = aws_autoscaling_group.spark_workers.name
  lifecycle_transition   = "autoscaling:EC2_INSTANCE_TERMINATING"
  heartbeat_timeout      = var.drain_timeout_seconds
  default_result         = "CONTINUE"
  notification_target_arn = aws_sns_topic.drains.arn  # or SQS / EventBridge for Lambda
  role_arn = aws_iam_role.lifecycle.arn
}

Drain flow & reasoning:

• Lifecycle hook pauses termination; sends notification (SNS/EventBridge) to a Lambda or step-function.
• Lambda calls SSM RunCommand targeting the instance to run a drain script (user-data places spark utilities). The drain script uses spark/yarn APIs to decommission executors, wait for in-flight tasks, then signals completion by calling CompleteLifecycleAction via AWS SDK.
• IAM roles: instance profile (SSM, CloudWatch), lifecycle role (autoscaling:CompleteLifecycleAction), Lambda role (ssm:SendCommand, autoscaling:CompleteLifecycleAction).

Edge cases & best practices:

• Use multiple instance_type overrides to improve spot flexibility.
• Set appropriate heartbeat_timeout > expected drain time + buffer.
• Use SSM rather than SSH for robust command execution and security.
• Emit metrics/logs to CloudWatch for failed drains and retry logic.
• Ensure graceful decommission uses Spark/YARN recommended REST APIs to avoid data loss.

Requirements & constraints:

• Real-time/near-real-time forecasts for capacity (hourly/day), high availability, low false positives (prevent churn in provisioning).
• Support scale: millions of time series (per service/region), cloud-native stack (e.g., Kafka, Spark/Flink, BigQuery/S3).

Drift signals (what to detect):

• Increased residuals: rolling mean/percentile of |observed - predicted| and normalized error (MAPE, RMSE) over window W.
• Change in seasonality: differences in estimated seasonal components (Fourier/ETS) via cosine similarity or subspace distance.
• Trend shift: change in slope of residual-corrected series (CUSUM).
• Concept shift in features: distributional change in exogenous features (traffic mix, promotions) via population statistics.

Monitoring metrics:

• Per-series: rolling MAPE, RMSE, standardized residual z-score, prediction interval coverage (PICP), Pinball loss for quantiles.
• Aggregate: % series with error > threshold, median lead-time to breach, rate of seasonality change.
• Data quality: missingness, latency, feature drift KS/PSI scores.

Statistical tests & detection logic:

• Residual increases: two-sample t-test or Mann-Whitney on residuals (baseline vs recent window) plus effect-size (Cohen’s d).
• Seasonality/trend: compare decomposed components with Hotelling’s T2 or cosine similarity; use Fisher’s exact for categorical feature shifts.
• Change-point: Bayesian online changepoint, or ADWIN/CUSUM for streaming.
• Combine via scoring: weighted ensemble that outputs severity score and confidence.

Automation & alerting:

• Alert tiers: info (early warning), action-required, critical. Include root signals, affected services, magnitude, timestamps, and rollback plan.
• Integrations: send to PagerDuty/Slack, create incident in ticketing (Jira), and record to observability (Grafana, Prometheus).
• Rate-limit/aggregation to reduce noise (group by service, region).

Retraining & deployment triggers:

• Soft trigger: severity score > retrain_threshold and sustained for T_consistent → schedule retrain in staging with latest data.
• Hard trigger: critical failures (prediction interval breaches causing SLA risk) → expedited retrain.
• Retrain pipeline: create reproducible train dataset snapshot, automated model training (CI), evaluation on holdout & backtest windows, compare against production with champion-challenger.
• Promotion criteria: statistically significant improvement on business metrics (e.g., lower cost of over-provisioning or SLA violations), stability checks.

Safety nets to avoid unstable provisioning:

• Staging & canary: deploy model to a small subset (e.g., 5% traffic/instances), A/B test for at least N cycles before full rollout.
• Smooth rollout: gradual weight shift using ensemble blending (e.g., 90:10 to 50:50).
• Guardrails: require improvement in prediction interval calibration and a max-permitted change in provisioning recommendations per window (rate-limiter delta percent).
• Human-in-loop: require manual approval for high-impact changes; provide explainability artifacts (feature importances, seasonal deltas).
• Automated rollback: if canary causes provisioning oscillation or increases SLA risk, auto-revert to previous model.

Implementation notes for a Data Engineer:

• Streaming ingestion: Kafka → Spark/Flink jobs compute residuals, KS/PSI, CUSUM; write metrics to time-series DB (Prometheus/Influx) and analytics store (BigQuery/S3).
• Batch retrain orchestration: Airflow/Argo workflows produce snapshots, run training on scalable infra (Spark/TF), push candidates to model registry (MLflow).
• Observability: dashboards (Grafana), alerts via Prometheus Alertmanager; store artifacts and audit logs for compliance.
• Performance: shard metrics, downsample when needed, and use sketching (TDigest) for quantiles at scale.

Key trade-offs:

• Sensitivity vs noise: tune windows, thresholds and require sustained signal to avoid churn.
• Compute cost vs freshness: balance retrain frequency with operational cost; prefer online detectors for early warnings and batch retrain for model updates.