Code Story: How We Implemented Multi-Region Active-Active Deployment with Kubernetes 1.32 and Submariner 0.19

#!/bin/bash
# deploy-submariner.sh: Automates Submariner 0.19 deployment across two K8s 1.32 clusters
# Requires: kubectl 1.32+, helm 3.14+, submariner-operator 0.19+
set -euo pipefail

# Configuration - update these values for your environment
readonly PRIMARY_CLUSTER="us-east-1-prod"
readonly PRIMARY_REGION="us-east-1"
readonly PRIMARY_KUBECONFIG="${HOME}/.kube/${PRIMARY_CLUSTER}.kubeconfig"
readonly SECONDARY_CLUSTER="eu-west-1-prod"
readonly SECONDARY_REGION="eu-west-1"
readonly SECONDARY_KUBECONFIG="${HOME}/.kube/${SECONDARY_CLUSTER}.kubeconfig"
readonly SUBMARINER_VERSION="0.19.0"
readonly OPERATOR_NAMESPACE="submariner-operator"
readonly BROKER_NAMESPACE="submariner-broker"
readonly IPSEC_PSK="your-secure-psk-here" # Replace with strong PSK, use vault in prod

# Validate prerequisites
validate_prereqs() {
  echo "Validating prerequisites..."
  for cmd in kubectl helm; do
    if ! command -v "${cmd}" &> /dev/null; then
      echo "ERROR: ${cmd} is not installed. Install ${cmd} 1.32+ before proceeding."
      exit 1
    fi
  done

  # Check kubectl version
  local kube_version
  kube_version=$(kubectl version --client -o json | jq -r '.clientVersion.gitVersion' | cut -d'v' -f2 | cut -d'+' -f1)
  if [[ "$(printf '%s\n' "1.32.0" "$kube_version" | sort -V | head -n1)" != "1.32.0" ]]; then
    echo "ERROR: kubectl version must be 1.32.0 or higher. Found: ${kube_version}"
    exit 1
  fi

  # Check helm version
  local helm_version
  helm_version=$(helm version --short | cut -d':' -f2 | cut -d'+' -f1)
  if [[ "$(printf '%s\n' "3.14.0" "$helm_version" | sort -V | head -n1)" != "3.14.0" ]]; then
    echo "ERROR: helm version must be 3.14.0 or higher. Found: ${helm_version}"
    exit 1
  fi

  # Validate kubeconfig files exist
  for kubeconfig in "${PRIMARY_KUBECONFIG}" "${SECONDARY_KUBECONFIG}"; do
    if [[ ! -f "${kubeconfig}" ]]; then
      echo "ERROR: Kubeconfig file ${kubeconfig} not found. Ensure kubeconfigs are in place."
      exit 1
    fi
  done
  echo "Prerequisites validated successfully."
}

# Deploy Submariner broker on primary cluster
deploy_broker() {
  echo "Deploying Submariner broker on ${PRIMARY_CLUSTER}..."
  export KUBECONFIG="${PRIMARY_KUBECONFIG}"

  # Create broker namespace
  kubectl create namespace "${BROKER_NAMESPACE}" --dry-run=client -o yaml | kubectl apply -f -

  # Add submariner helm repo
  helm repo add submariner https://submariner-io.github.io/submariner-charts
  helm repo update

  # Install broker
  helm upgrade --install submariner-broker submariner/submariner-broker \
    --namespace "${BROKER_NAMESPACE}" \
    --version "${SUBMARINER_VERSION}" \
    --set ipsec.psk="${IPSEC_PSK}" \
    --wait --timeout 10m

  echo "Broker deployed successfully. Saving broker info..."
  kubectl get secret -n "${BROKER_NAMESPACE}" submariner-broker-info -o yaml > broker-info.secret.yaml
  echo "Broker info saved to broker-info.secret.yaml"
}

# Deploy Submariner operator on both clusters
deploy_operator() {
  echo "Deploying Submariner operator on both clusters..."

  # Primary cluster
  export KUBECONFIG="${PRIMARY_KUBECONFIG}"
  kubectl create namespace "${OPERATOR_NAMESPACE}" --dry-run=client -o yaml | kubectl apply -f -
  helm upgrade --install submariner-operator submariner/submariner-operator \
    --namespace "${OPERATOR_NAMESPACE}" \
    --version "${SUBMARINER_VERSION}" \
    --wait --timeout 10m

  # Secondary cluster
  export KUBECONFIG="${SECONDARY_KUBECONFIG}"
  kubectl create namespace "${OPERATOR_NAMESPACE}" --dry-run=client -o yaml | kubectl apply -f -
  helm upgrade --install submariner-operator submariner/submariner-operator \
    --namespace "${OPERATOR_NAMESPACE}" \
    --version "${SUBMARINER_VERSION}" \
    --wait --timeout 10m

  echo "Operators deployed successfully on both clusters."
}

# Join clusters to Submariner
join_clusters() {
  echo "Joining clusters to Submariner..."

  # Join primary cluster
  export KUBECONFIG="${PRIMARY_KUBECONFIG}"
  kubectl apply -f broker-info.secret.yaml
  kubectl label namespace default submariner.io/gateway=true --overwrite

  # Join secondary cluster
  export KUBECONFIG="${SECONDARY_KUBECONFIG}"
  kubectl apply -f broker-info.secret.yaml
  kubectl label namespace default submariner.io/gateway=true --overwrite

  echo "Clusters joined successfully."
}

# Validate deployment
validate_deployment() {
  echo "Validating Submariner deployment..."
  export KUBECONFIG="${PRIMARY_KUBECONFIG}"
  if ! kubectl get pods -n "${OPERATOR_NAMESPACE}" -l app=submariner-gateway -o jsonpath='{.items[0].status.phase}' | grep -q Running; then
    echo "ERROR: Submariner gateway pod not running on primary cluster."
    exit 1
  fi

  export KUBECONFIG="${SECONDARY_KUBECONFIG}"
  if ! kubectl get pods -n "${OPERATOR_NAMESPACE}" -l app=submariner-gateway -o jsonpath='{.items[0].status.phase}' | grep -q Running; then
    echo "ERROR: Submariner gateway pod not running on secondary cluster."
    exit 1
  fi

  echo "Deployment validated. Submariner 0.19 is running on both K8s 1.32 clusters."
}

# Main execution
main() {
  validate_prereqs
  deploy_broker
  deploy_operator
  join_clusters
  validate_deployment
}

main

// latency-benchmark.go: Benchmarks cross-region service latency for active-active K8s clusters with Submariner 0.19
// Build: go build -o latency-benchmark latency-benchmark.go
// Run: ./latency-benchmark --primary-kubeconfig=us-east-1.kubeconfig --secondary-kubeconfig=eu-west-1.kubeconfig --service=nginx --port=80 --requests=1000
package main

import (
    "context"
    "flag"
    "fmt"
    "io"
    "net/http"
    "os"
    "sort"
    "sync"
    "time"

    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// Config holds benchmark configuration
type Config struct {
    PrimaryKubeconfig   string
    SecondaryKubeconfig string
    ServiceName         string
    ServicePort         int
    RequestCount        int
    RequestTimeout      time.Duration
}

// LatencyResult holds individual request latency data
type LatencyResult struct {
    Region    string
    Latency   time.Duration
    Error     error
    Timestamp time.Time
}

func main() {
    // Parse flags
    primaryKubeconfig := flag.String("primary-kubeconfig", "", "Path to primary cluster kubeconfig")
    secondaryKubeconfig := flag.String("secondary-kubeconfig", "", "Path to secondary cluster kubeconfig")
    serviceName := flag.String("service", "nginx", "Name of service to benchmark")
    servicePort := flag.Int("port", 80, "Port of service to benchmark")
    requestCount := flag.Int("requests", 1000, "Number of requests to send per region")
    requestTimeout := flag.Duration("timeout", 5*time.Second, "Timeout per request")
    flag.Parse()

    // Validate flags
    if *primaryKubeconfig == "" || *secondaryKubeconfig == "" {
        fmt.Fprintf(os.Stderr, "ERROR: primary-kubeconfig and secondary-kubeconfig are required\n")
        flag.Usage()
        os.Exit(1)
    }

    cfg := Config{
        PrimaryKubeconfig:   *primaryKubeconfig,
        SecondaryKubeconfig: *secondaryKubeconfig,
        ServiceName:         *serviceName,
        ServicePort:         *servicePort,
        RequestCount:        *requestCount,
        RequestTimeout:      *requestTimeout,
    }

    // Get service ClusterIPs from both clusters
    primaryIP, err := getServiceIP(cfg.PrimaryKubeconfig, cfg.ServiceName, cfg.ServicePort)
    if err != nil {
        fmt.Fprintf(os.Stderr, "ERROR: Failed to get service IP from primary cluster: %v\n", err)
        os.Exit(1)
    }
    secondaryIP, err := getServiceIP(cfg.SecondaryKubeconfig, cfg.ServiceName, cfg.ServicePort)
    if err != nil {
        fmt.Fprintf(os.Stderr, "ERROR: Failed to get service IP from secondary cluster: %v\n", err)
        os.Exit(1)
    }

    fmt.Printf("Starting benchmark: %d requests per region\n", cfg.RequestCount)
    fmt.Printf("Primary service IP: %s:%d\n", primaryIP, cfg.ServicePort)
    fmt.Printf("Secondary service IP: %s:%d\n", secondaryIP, cfg.ServicePort)

    // Run benchmarks
    primaryResults := runBenchmark("us-east-1", primaryIP, cfg.ServicePort, cfg.RequestCount, cfg.RequestTimeout)
    secondaryResults := runBenchmark("eu-west-1", secondaryIP, cfg.ServicePort, cfg.RequestCount, cfg.RequestTimeout)

    // Calculate and print stats
    printStats("Primary (us-east-1)", primaryResults)
    printStats("Secondary (eu-west-1)", secondaryResults)
}

// getServiceIP retrieves the ClusterIP of a service in a K8s cluster
func getServiceIP(kubeconfigPath, serviceName string, port int) (string, error) {
    config, err := clientcmd.BuildConfigFromFlags("", kubeconfigPath)
    if err != nil {
        return "", fmt.Errorf("failed to build kubeconfig: %w", err)
    }

    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        return "", fmt.Errorf("failed to create k8s client: %w", err)
    }

    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    service, err := clientset.CoreV1().Services("default").Get(ctx, serviceName, v1.GetOptions{})
    if err != nil {
        return "", fmt.Errorf("failed to get service %s: %w", serviceName, err)
    }

    // Find port in service
    var targetPort int32
    for _, svcPort := range service.Spec.Ports {
        if svcPort.Port == int32(port) {
            targetPort = svcPort.TargetPort.IntVal
            break
        }
    }
    if targetPort == 0 {
        return "", fmt.Errorf("port %d not found in service %s", port, serviceName)
    }

    return service.Spec.ClusterIP, nil
}

// runBenchmark sends concurrent requests to a service and records latency
func runBenchmark(region, ip string, port, requestCount int, timeout time.Duration) []LatencyResult {
    results := make([]LatencyResult, requestCount)
    var wg sync.WaitGroup
    wg.Add(requestCount)

    client := &http.Client{
        Timeout: timeout,
    }

    for i := 0; i < requestCount; i++ {
        go func(idx int) {
            defer wg.Done()
            start := time.Now()
            resp, err := client.Get(fmt.Sprintf("http://%s:%d", ip, port))
            latency := time.Since(start)

            result := LatencyResult{
                Region:    region,
                Latency:   latency,
                Timestamp: start,
            }

            if err != nil {
                result.Error = err
            } else {
                io.Copy(io.Discard, resp.Body)
                resp.Body.Close()
            }

            results[idx] = result
        }(i)
    }

    wg.Wait()
    return results
}

// printStats calculates and prints latency statistics
func printStats(region string, results []LatencyResult) {
    // Filter successful requests
    var latencies []time.Duration
    var errorCount int
    for _, res := range results {
        if res.Error != nil {
            errorCount++
            continue
        }
        latencies = append(latencies, res.Latency)
    }

    if len(latencies) == 0 {
        fmt.Printf("\n%s: No successful requests\n", region)
        return
    }

    // Sort latencies for percentile calculation
    sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] })

    p50 := latencies[len(latencies)/2]
    p99 := latencies[int(float64(len(latencies))*0.99)]
    avg := time.Duration(0)
    for _, l := range latencies {
        avg += l
    }
    avg = avg / time.Duration(len(latencies))

    fmt.Printf("\n%s Latency Stats (successful: %d/%d)\n", region, len(latencies), len(results))
    fmt.Printf("  Avg: %v\n", avg)
    fmt.Printf("  P50: %v\n", p50)
    fmt.Printf("  P99: %v\n", p99)
    fmt.Printf("  Error rate: %.2f%%\n", float64(errorCount)/float64(len(results))*100)
}

#!/usr/bin/env python3
# active-active-router.py: Configures weighted traffic routing for active-active K8s 1.32 clusters via Submariner 0.19
# Requires: kubernetes 28.1.0+, python 3.10+
import argparse
import logging
import sys
import time
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 ActiveActiveRouter:
    """Manages weighted traffic routing between active-active K8s clusters."""

    def __init__(self, primary_kubeconfig: str, secondary_kubeconfig: str, namespace: str = "default"):
        self.namespace = namespace
        # Load primary cluster config
        config.load_kube_config(config_file=primary_kubeconfig)
        self.primary_clientset = client.ApiClient()
        self.primary_v1 = client.CoreV1Api(self.primary_clientset)
        self.primary_discovery = client.DiscoveryV1Api(self.primary_clientset)

        # Load secondary cluster config
        config.load_kube_config(config_file=secondary_kubeconfig)
        self.secondary_clientset = client.ApiClient()
        self.secondary_v1 = client.CoreV1Api(self.secondary_clientset)
        self.secondary_discovery = client.DiscoveryV1Api(self.secondary_clientset)

        self.endpoint_slice_version = "discovery.k8s.io/v1"  # K8s 1.32 uses v1 EndpointSlice

    def create_endpoint_slice(self, cluster: str, service_name: str, port: int, weight: int) -> None:
        """Creates an EndpointSlice for a service in a cluster with a given traffic weight."""
        try:
            # Get service to retrieve selector
            if cluster == "primary":
                svc = self.primary_v1.read_namespaced_service(service_name, self.namespace)
                discovery_api = self.primary_discovery
            else:
                svc = self.secondary_v1.read_namespaced_service(service_name, self.namespace)
                discovery_api = self.secondary_discovery

            # Get pods matching service selector
            selector = ",".join([f"{k}={v}" for k, v in svc.spec.selector.items()])
            if cluster == "primary":
                pods = self.primary_v1.list_namespaced_pod(self.namespace, label_selector=selector)
            else:
                pods = self.secondary_v1.list_namespaced_pod(self.namespace, label_selector=selector)

            if not pods.items:
                logger.warning(f"No pods found for service {service_name} in {cluster} cluster")
                return

            # Build EndpointSlice
            endpoints = []
            for pod in pods.items:
                if pod.status.pod_ip is None:
                    continue
                # Get container port
                container_port = None
                for c in pod.spec.containers:
                    for p in c.ports:
                        if p.container_port == port:
                            container_port = p.container_port
                            break
                    if container_port:
                        break
                if not container_port:
                    continue
                endpoints.append({
                    "addresses": [pod.status.pod_ip],
                    "conditions": {"ready": True, "serving": True},
                    "ports": [{"name": "http", "port": container_port}],
                    "topology": {"kubernetes.io/hostname": pod.spec.node_name}
                })

            if not endpoints:
                logger.warning(f"No ready endpoints found for {service_name} in {cluster}")
                return

            endpoint_slice = {
                "apiVersion": self.endpoint_slice_version,
                "kind": "EndpointSlice",
                "metadata": {
                    "name": f"{service_name}-{cluster}",
                    "namespace": self.namespace,
                    "labels": {
                        "kubernetes.io/service-name": service_name,
                        "traffic.router/weight": str(weight),
                        "cluster": cluster
                    }
                },
                "addressType": "IPv4",
                "endpoints": endpoints,
                "ports": [{"name": "http", "port": port, "protocol": "TCP"}]
            }

            # Create or update EndpointSlice
            try:
                discovery_api.create_namespaced_endpoint_slice(self.namespace, endpoint_slice)
                logger.info(f"Created EndpointSlice {service_name}-{cluster} in {cluster} cluster")
            except ApiException as e:
                if e.status == 409:
                    discovery_api.replace_namespaced_endpoint_slice(f"{service_name}-{cluster}", self.namespace, endpoint_slice)
                    logger.info(f"Updated EndpointSlice {service_name}-{cluster} in {cluster} cluster")
                else:
                    raise

        except ApiException as e:
            logger.error(f"K8s API error creating EndpointSlice for {cluster}: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error creating EndpointSlice for {cluster}: {e}")
            raise

    def update_traffic_weights(self, service_name: str, primary_weight: int, secondary_weight: int) -> None:
        """Updates traffic weights for active-active service across both clusters."""
        if primary_weight + secondary_weight != 100:
            logger.error("Total traffic weight must be 100")
            sys.exit(1)

        logger.info(f"Updating traffic weights for {service_name}: primary={primary_weight}%, secondary={secondary_weight}%")

        # Get service port (assume first port for simplicity)
        try:
            svc = self.primary_v1.read_namespaced_service(service_name, self.namespace)
            port = svc.spec.ports[0].port
        except ApiException as e:
            logger.error(f"Failed to get service {service_name}: {e}")
            sys.exit(1)

        # Create/update EndpointSlices with weights
        self.create_endpoint_slice("primary", service_name, port, primary_weight)
        self.create_endpoint_slice("secondary", service_name, port, secondary_weight)

        logger.info(f"Traffic weights updated successfully for {service_name}")

    def validate_routing(self, service_name: str) -> None:
        """Validates that traffic is routed correctly based on weights."""
        logger.info(f"Validating routing for {service_name}...")
        # In production, this would send test requests and verify distribution
        # For brevity, we check EndpointSlice labels
        primary_slice = self.primary_discovery.read_namespaced_endpoint_slice(f"{service_name}-primary", self.namespace)
        secondary_slice = self.secondary_discovery.read_namespaced_endpoint_slice(f"{service_name}-secondary", self.namespace)

        primary_weight = int(primary_slice.metadata.labels["traffic.router/weight"])
        secondary_weight = int(secondary_slice.metadata.labels["traffic.router/weight"])

        logger.info(f"Validation passed: Primary weight={primary_weight}%, Secondary weight={secondary_weight}%")
        assert primary_weight + secondary_weight == 100, "Total weight is not 100"

def main():
    parser = argparse.ArgumentParser(description="Configure active-active traffic routing for K8s 1.32 clusters")
    parser.add_argument("--primary-kubeconfig", required=True, help="Path to primary cluster kubeconfig")
    parser.add_argument("--secondary-kubeconfig", required=True, help="Path to secondary cluster kubeconfig")
    parser.add_argument("--service", required=True, help="Name of service to configure")
    parser.add_argument("--primary-weight", type=int, default=70, help="Primary cluster traffic weight (default: 70)")
    parser.add_argument("--secondary-weight", type=int, default=30, help="Secondary cluster traffic weight (default: 30)")
    parser.add_argument("--namespace", default="default", help="Service namespace (default: default)")

    args = parser.parse_args()

    # Validate weights
    if args.primary_weight + args.secondary_weight != 100:
        logger.error("Primary and secondary weights must sum to 100")
        sys.exit(1)

    router = ActiveActiveRouter(
        primary_kubeconfig=args.primary_kubeconfig,
        secondary_kubeconfig=args.secondary_kubeconfig,
        namespace=args.namespace
    )

    router.update_traffic_weights(
        service_name=args.service,
        primary_weight=args.primary_weight,
        secondary_weight=args.secondary_weight
    )

    router.validate_routing(args.service)

if __name__ == "__main__":
    main()

# Validate Submariner connectivity between clusters
export KUBECONFIG=us-east-1.kubeconfig
subctl diagnose --remote-kubeconfig=eu-west-1.kubeconfig --namespace submariner-operator
# Check tunnel status
kubectl get pods -n submariner-operator -l app=submariner-gateway -o wide
subctl gateway status

apiVersion: discovery.k8s.io/v1
kind: EndpointSlice
metadata:
  name: nginx-us-east-1
  namespace: default
  labels:
    kubernetes.io/service-name: nginx
    traffic.router/weight: "70"
    cluster: us-east-1
addressType: IPv4
endpoints:
- addresses: ["10.0.1.12"]
  conditions:
    ready: true
    serving: true
  ports:
  - name: http
    port: 80
    protocol: TCP

# Prometheus scrape config for Submariner gateway metrics
scrape_configs:
- job_name: submariner-gateway
  kubernetes_sd_configs:
  - role: pod
  relabel_configs:
  - source_labels: [__meta_kubernetes_pod_label_app]
    action: keep
    regex: submariner-gateway
  - source_labels: [__meta_kubernetes_namespace]
    action: keep
    regex: submariner-operator
  metrics_path: /metrics
  port: 8080