War Story: How We Implemented Disaster Recovery for Our AWS EKS Cluster

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "strings"
    "time"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/eks"
    "github.com/aws/aws-sdk-go-v2/service/s3"
)

// DRReadinessReport stores the results of EKS DR readiness checks
type DRReadinessReport struct {
    ClusterName     string    `json:"cluster_name"`
    Region          string    `json:"region"`
    BackupEnabled   bool      `json:"backup_enabled"`
    CrossRegionRepl bool      `json:"cross_region_replication"`
    EtcdHealth      string    `json:"etcd_health"`
    LastBackupTime  time.Time `json:"last_backup_time"`
    Errors          []string  `json:"errors,omitempty"`
}

func main() {
    // Parse command line arguments for cluster name and primary region
    if len(os.Args) < 3 {
        log.Fatal("Usage: dr-checker  ")
    }
    clusterName := os.Args[1]
    primaryRegion := os.Args[2]
    drRegion := os.Getenv("DR_REGION")
    if drRegion == "" {
        drRegion = "us-west-2" // default DR region
    }

    cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion(primaryRegion))
    if err != nil {
        log.Fatalf("Failed to load AWS config: %v", err)
    }

    // Initialize EKS and S3 clients
    eksClient := eks.NewFromConfig(cfg)
    s3Client := s3.NewFromConfig(cfg)

    report := DRReadinessReport{
        ClusterName: clusterName,
        Region:      primaryRegion,
        Errors:      []string{},
    }

    // Check if EKS cluster exists and is active
    clusterResp, err := eksClient.DescribeCluster(context.TODO(), &eks.DescribeClusterInput{
        Name: aws.String(clusterName),
    })
    if err != nil {
        report.Errors = append(report.Errors, fmt.Sprintf("Failed to describe cluster: %v", err))
        printReport(report)
        os.Exit(1)
    }
    if *clusterResp.Cluster.Status != "ACTIVE" {
        report.Errors = append(report.Errors, fmt.Sprintf("Cluster status is %s, expected ACTIVE", *clusterResp.Cluster.Status))
    }

    // Check Velero backup status (assuming Velero backup bucket is named velero-)
    backupBucket := fmt.Sprintf("velero-%s-backups", clusterName)
    _, err = s3Client.HeadBucket(context.TODO(), &s3.HeadBucketInput{
        Bucket: aws.String(backupBucket),
    })
    if err != nil {
        report.Errors = append(report.Errors, fmt.Sprintf("Velero backup bucket %s not found: %v", backupBucket, err))
        report.BackupEnabled = false
    } else {
        report.BackupEnabled = true
        // Check last backup time by listing objects in bucket
        listResp, err := s3Client.ListObjectsV2(context.TODO(), &s3.ListObjectsV2Input{
            Bucket: aws.String(backupBucket),
            Prefix: aws.String("backups/"),
            MaxKeys: aws.Int32(1),
        })
        if err != nil {
            report.Errors = append(report.Errors, fmt.Sprintf("Failed to list backup objects: %v", err))
        } else if len(listResp.Contents) > 0 {
            report.LastBackupTime = *listResp.Contents[0].LastModified
        }
    }

    // Check cross-region replication for backup bucket
    replCfg, err := s3Client.GetBucketReplication(context.TODO(), &s3.GetBucketReplicationInput{
        Bucket: aws.String(backupBucket),
    })
    if err != nil {
        report.Errors = append(report.Errors, fmt.Sprintf("Failed to get replication config: %v", err))
        report.CrossRegionRepl = false
    } else {
        // Check if replication to DR region is configured
        for _, rule := range replCfg.ReplicationConfiguration.Rules {
            if rule.Destination != nil && rule.Destination.Bucket != nil {
                if strings.Contains(*rule.Destination.Bucket, drRegion) {
                    report.CrossRegionRepl = true
                    break
                }
            }
        }
    }

    // Check etcd health via EKS describe cluster (simplified, real impl would use k8s client)
    // In production, we'd use k8s client-go to hit etcd endpoints directly
    report.EtcdHealth = "HEALTHY" // Placeholder for brevity, real check would query etcd
    if len(report.Errors) > 0 {
        report.EtcdHealth = "DEGRADED"
    }

    printReport(report)
}

func printReport(report DRReadinessReport) {
    jsonData, err := json.MarshalIndent(report, "", "  ")
    if err != nil {
        log.Fatalf("Failed to marshal report: %v", err)
    }
    fmt.Println(string(jsonData))
}

import os
import sys
import time
import logging
from typing import List, Dict, Optional

import boto3
from kubernetes import client, config
from kubernetes.client.rest import ApiException

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class EKSFailoverOrchestrator:
    """Automates failover of EKS cluster from primary to DR region"""

    def __init__(self, primary_cluster: str, dr_cluster: str, primary_region: str, dr_region: str):
        self.primary_cluster = primary_cluster
        self.dr_cluster = dr_cluster
        self.primary_region = primary_region
        self.dr_region = dr_region

        # Initialize AWS clients
        self.eks_primary = boto3.client("eks", region_name=primary_region)
        self.eks_dr = boto3.client("eks", region_name=dr_region)
        self.elbv2 = boto3.client("elbv2", region_name=primary_region)

        # Load k8s config for primary cluster
        try:
            config.load_kube_config(context=f"arn:aws:eks:{primary_region}:123456789012:cluster/{primary_cluster}")
            self.k8s_apps = client.AppsV1Api()
            self.k8s_core = client.CoreV1Api()
        except Exception as e:
            logger.error(f"Failed to load k8s config for primary cluster: {e}")
            sys.exit(1)

    def check_primary_health(self) -> bool:
        """Check if primary cluster is actually unhealthy, not a false alarm"""
        try:
            # Check EKS cluster status
            resp = self.eks_primary.describe_cluster(name=self.primary_cluster)
            if resp["cluster"]["status"] != "ACTIVE":
                logger.warning(f"Primary cluster status: {resp['cluster']['status']}")
                return False

            # Check API server availability
            try:
                self.k8s_core.list_namespace(limit=1)
                logger.info("Primary cluster API server is responsive")
                return True
            except ApiException as e:
                logger.error(f"Primary API server unavailable: {e}")
                return False
        except Exception as e:
            logger.error(f"Failed to check primary health: {e}")
            return False

    def scale_down_primary_workloads(self) -> None:
        """Scale all deployments to 0 in primary cluster to prevent split-brain"""
        try:
            deployments = self.k8s_apps.list_deployment_for_all_namespaces()
            for deploy in deployments.items:
                if deploy.metadata.namespace in ["kube-system", "velero-system"]:
                    continue  # Skip system workloads
                logger.info(f"Scaling down deployment {deploy.metadata.namespace}/{deploy.metadata.name}")
                self.k8s_apps.patch_namespaced_deployment_scale(
                    name=deploy.metadata.name,
                    namespace=deploy.metadata.namespace,
                    body={"spec": {"replicas": 0}}
                )
            logger.info("All primary workloads scaled to 0")
        except ApiException as e:
            logger.error(f"Failed to scale down workloads: {e}")
            raise

    def trigger_dr_failover(self) -> None:
        """Activate DR cluster and update DNS"""
        try:
            # Check DR cluster status
            dr_resp = self.eks_dr.describe_cluster(name=self.dr_cluster)
            if dr_resp["cluster"]["status"] != "ACTIVE":
                raise RuntimeError(f"DR cluster status is {dr_resp['cluster']['status']}")

            # Update Route53 DNS to point to DR cluster ALB
            # Simplified: in production, use Route53 client to update record sets
            logger.info("Updating DNS to point to DR cluster ALB")
            # Placeholder for Route53 update logic
            time.sleep(5)  # Simulate DNS propagation

            # Scale up DR workloads (Velero restore would handle this, but fallback)
            logger.info("DR cluster activated, workloads restoring via Velero")
        except Exception as e:
            logger.error(f"Failover failed: {e}")
            raise

    def run_failover(self) -> None:
        """Full failover workflow"""
        logger.info(f"Starting failover from {self.primary_cluster} to {self.dr_cluster}")

        # Step 1: Verify primary is actually down
        if self.check_primary_health():
            logger.error("Primary cluster is healthy, aborting failover")
            sys.exit(1)

        # Step 2: Scale down primary to prevent split-brain
        self.scale_down_primary_workloads()

        # Step 3: Trigger DR activation
        self.trigger_dr_failover()

        logger.info("Failover completed successfully")

if __name__ == "__main__":
    if len(sys.argv) != 5:
        logger.error("Usage: failover.py    ")
        sys.exit(1)

    orchestrator = EKSFailoverOrchestrator(
        primary_cluster=sys.argv[1],
        dr_cluster=sys.argv[2],
        primary_region=sys.argv[3],
        dr_region=sys.argv[4]
    )
    orchestrator.run_failover()

package main

import (
    "context"
    "fmt"
    "log"
    "net/http"
    "os"
    "time"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/eks"
    "github.com/aws/aws-sdk-go-v2/service/s3"
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
)

// Prometheus metrics
var (
    drBackupSuccess = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "eks_dr_backup_success_total",
            Help: "Total number of successful DR backups",
        },
        []string{"cluster", "region"},
    )
    drBackupFailure = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "eks_dr_backup_failure_total",
            Help: "Total number of failed DR backups",
        },
        []string{"cluster", "region"},
    )
    drLastBackupTimestamp = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "eks_dr_last_backup_timestamp_seconds",
            Help: "Timestamp of last successful DR backup",
        },
        []string{"cluster", "region"},
    )
    drCrossRegionReplStatus = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "eks_dr_cross_region_replication_enabled",
            Help: "1 if cross-region replication is enabled, 0 otherwise",
        },
        []string{"cluster", "region", "dr_region"},
    )
)

func init() {
    // Register Prometheus metrics
    prometheus.MustRegister(drBackupSuccess)
    prometheus.MustRegister(drBackupFailure)
    prometheus.MustRegister(drLastBackupTimestamp)
    prometheus.MustRegister(drCrossRegionReplStatus)
}

type DRMetricsCollector struct {
    ClusterName string
    PrimaryRegion string
    DRRegion string
    EksClient *eks.Client
    S3Client *s3.Client
}

func (c *DRMetricsCollector) CollectMetrics(ctx context.Context) {
    backupBucket := fmt.Sprintf("velero-%s-backups", c.ClusterName)

    // Check cross-region replication status
    replCfg, err := c.S3Client.GetBucketReplication(ctx, &s3.GetBucketReplicationInput{
        Bucket: aws.String(backupBucket),
    })
    if err != nil {
        log.Printf("Failed to get replication config: %v", err)
        drCrossRegionReplStatus.WithLabelValues(c.ClusterName, c.PrimaryRegion, c.DRRegion).Set(0)
    } else {
        replEnabled := 0
        for _, rule := range replCfg.ReplicationConfiguration.Rules {
            if rule.Destination != nil && rule.Destination.Bucket != nil {
                if strings.Contains(*rule.Destination.Bucket, c.DRRegion) {
                    replEnabled = 1
                    break
                }
            }
        }
        drCrossRegionReplStatus.WithLabelValues(c.ClusterName, c.PrimaryRegion, c.DRRegion).Set(float64(replEnabled))
    }

    // Check last backup time
    listResp, err := c.S3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
        Bucket: aws.String(backupBucket),
        Prefix: aws.String("backups/"),
        MaxKeys: aws.Int32(1),
    })
    if err != nil {
        log.Printf("Failed to list backup objects: %v", err)
        drBackupFailure.WithLabelValues(c.ClusterName, c.PrimaryRegion).Inc()
    } else if len(listResp.Contents) > 0 {
        lastBackup := *listResp.Contents[0].LastModified
        drLastBackupTimestamp.WithLabelValues(c.ClusterName, c.PrimaryRegion).Set(float64(lastBackup.Unix()))
        drBackupSuccess.WithLabelValues(c.ClusterName, c.PrimaryRegion).Inc()
    }
}

func contains(s, substr string) bool {
    return len(s) >= len(substr) && (s == substr || len(s) > 0 && (s[0:len(substr)] == substr || contains(s[1:], substr)))
}

func main() {
    clusterName := os.Getenv("CLUSTER_NAME")
    if clusterName == "" {
        log.Fatal("CLUSTER_NAME environment variable not set")
    }
    primaryRegion := os.Getenv("PRIMARY_REGION")
    if primaryRegion == "" {
        primaryRegion = "us-east-1"
    }
    drRegion := os.Getenv("DR_REGION")
    if drRegion == "" {
        drRegion = "us-west-2"
    }
    metricsPort := os.Getenv("METRICS_PORT")
    if metricsPort == "" {
        metricsPort = "9090"
    }

    cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion(primaryRegion))
    if err != nil {
        log.Fatalf("Failed to load AWS config: %v", err)
    }

    eksClient := eks.NewFromConfig(cfg)
    s3Client := s3.NewFromConfig(cfg)

    collector := &DRMetricsCollector{
        ClusterName: clusterName,
        PrimaryRegion: primaryRegion,
        DRRegion: drRegion,
        EksClient: eksClient,
        S3Client: s3Client,
    }

    // Collect metrics every 5 minutes
    go func() {
        ticker := time.NewTicker(5 * time.Minute)
        defer ticker.Stop()
        for {
            select {
            case <-ticker.C:
                collector.CollectMetrics(context.TODO())
            }
        }
    }()

    // Expose Prometheus metrics
    http.Handle("/metrics", promhttp.Handler())
    log.Printf("Starting metrics server on port %s", metricsPort)
    if err := http.ListenAndServe(fmt.Sprintf(":%s", metricsPort), nil); err != nil {
        log.Fatalf("Failed to start metrics server: %v", err)
    }
}

etcdctl snapshot restore --data-dir /var/lib/etcd-restore snapshot.db
etcdctl snapshot status snapshot.db -w table

from kubernetes import client
apps = client.AppsV1Api()
apps.patch_namespaced_deployment_scale(
    name="order-service",
    namespace="production",
    body={"spec": {"replicas": 0}}
)

import http from 'k6/http';
export const options = { stages: [{ duration: '5m', target: 10000 }] };
export default function() { http.get('https://my-app.com/health'); }