Deep Dive: How AWS VPC Lattice Works with EKS 1.36 – Service Networking Internals

// Copyright 2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
// Source: https://github.com/aws/aws-vpc-lattice-eks-addon/blob/main/pkg/webhook/mutate.go
// MutatingWebhook handles pod admission requests to inject VPC Lattice Envoy sidecars
package webhook

import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"

    admissionv1 "k8s.io/api/admission/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/rest"
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/service/lattice"
    "github.com/prometheus/client_golang/prometheus"
)

var (
    sidecarInjectionCounter = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "lattice_sidecar_injections_total",
            Help: "Total number of VPC Lattice sidecar injection attempts",
        },
        []string{"namespace", "status"},
    )
)

type MutatingWebhook struct {
    client     *kubernetes.Clientset
    latticeSvc *lattice.Client
}

// NewMutatingWebhook initializes a new webhook handler with K8s and Lattice clients
func NewMutatingWebhook() (*MutatingWebhook, error) {
    config, err := rest.InClusterConfig()
    if err != nil {
        return nil, fmt.Errorf("failed to load in-cluster config: %w", err)
    }
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        return nil, fmt.Errorf("failed to create kubernetes clientset: %w", err)
    }
    // Initialize Lattice client with IRSA credentials
    awsCfg, err := awsConfig.LoadDefaultConfig(context.Background())
    if err != nil {
        return nil, fmt.Errorf("failed to load AWS config: %w", err)
    }
    latticeSvc := lattice.NewFromConfig(awsCfg)

    prometheus.MustRegister(sidecarInjectionCounter)
    return &MutatingWebhook{
        client:     clientset,
        latticeSvc: latticeSvc,
    }, nil
}

// HandleAdmission processes incoming pod admission requests
func (m *MutatingWebhook) HandleAdmission(w http.ResponseWriter, r *http.Request) {
    var admissionReq admissionv1.AdmissionReview
    if err := json.NewDecoder(r.Body).Decode(&admissionReq); err != nil {
        http.Error(w, fmt.Sprintf("failed to decode admission request: %v", err), http.StatusBadRequest)
        sidecarInjectionCounter.WithLabelValues("unknown", "decode_error").Inc()
        return
    }

    // Only process pod creation requests
    if admissionReq.Request.Operation != admissionv1.Create {
        sendAdmissionResponse(w, admissionReq.Request.UID, true, "non-create operation skipped")
        return
    }

    var pod corev1.Pod
    if err := json.Unmarshal(admissionReq.Request.Object.Raw, &pod); err != nil {
        http.Error(w, fmt.Sprintf("failed to unmarshal pod: %v", err), http.StatusBadRequest)
        sidecarInjectionCounter.WithLabelValues(pod.Namespace, "unmarshal_error").Inc()
        return
    }

    // Check if pod has Lattice service annotation
    svcName, ok := pod.Annotations["lattice.amazonaws.com/service-name"]
    if !ok {
        sendAdmissionResponse(w, admissionReq.Request.UID, true, "no lattice service annotation")
        return
    }

    // Verify service exists in Lattice control plane
    _, err := m.latticeSvc.GetService(context.Background(), &lattice.GetServiceInput{
        ServiceIdentifier: aws.String(svcName),
    })
    if err != nil {
        sendAdmissionResponse(w, admissionReq.Request.UID, false, fmt.Sprintf("service %s not found in Lattice: %v", svcName, err))
        sidecarInjectionCounter.WithLabelValues(pod.Namespace, "service_not_found").Inc()
        return
    }

    // Inject Envoy sidecar container
    mutatedPod, err := injectSidecar(pod)
    if err != nil {
        sendAdmissionResponse(w, admissionReq.Request.UID, false, fmt.Sprintf("sidecar injection failed: %v", err))
        sidecarInjectionCounter.WithLabelValues(pod.Namespace, "injection_error").Inc()
        return
    }

    // Marshal mutated pod and send patch response
    mutatedRaw, err := json.Marshal(mutatedPod)
    if err != nil {
        http.Error(w, fmt.Sprintf("failed to marshal mutated pod: %v", err), http.StatusInternalServerError)
        return
    }

    patch := []map[string]interface{}{
        {
            "op":    "replace",
            "path":  "/spec",
            "value": mutatedPod.Spec,
        },
    }
    patchRaw, err := json.Marshal(patch)
    if err != nil {
        http.Error(w, fmt.Sprintf("failed to marshal patch: %v", err), http.StatusInternalServerError)
        return
    }

    admissionResp := admissionv1.AdmissionReview{
        TypeMeta: metav1.TypeMeta{
            APIVersion: "admission.k8s.io/v1",
            Kind:       "AdmissionReview",
        },
        Response: &admissionv1.AdmissionResponse{
            UID:     admissionReq.Request.UID,
            Allowed: true,
            Patch:   patchRaw,
            PatchType: func() *admissionv1.PatchType {
                pt := admissionv1.PatchTypeJSONPatch
                return &pt
            }(),
        },
    }

    if err := json.NewEncoder(w).Encode(admissionResp); err != nil {
        http.Error(w, fmt.Sprintf("failed to encode admission response: %v", err), http.StatusInternalServerError)
        return
    }
    sidecarInjectionCounter.WithLabelValues(pod.Namespace, "success").Inc()
}

// injectSidecar adds the VPC Lattice Envoy sidecar to the pod spec
func injectSidecar(pod corev1.Pod) (corev1.Pod, error) {
    sidecar := corev1.Container{
        Name:  "lattice-envoy",
        Image: "public.ecr.aws/aws-vpc-lattice/lattice-envoy:1.29.1-v1.2.4",
        Ports: []corev1.ContainerPort{
            {
                ContainerPort: 15001,
                Name:          "inbound",
                Protocol:      corev1.ProtocolTCP,
            },
            {
                ContainerPort: 15006,
                Name:          "outbound",
                Protocol:      corev1.ProtocolTCP,
            },
        },
        Env: []corev1.EnvVar{
            {
                Name: "POD_NAMESPACE",
                ValueFrom: &corev1.EnvVarSource{
                    FieldRef: &corev1.ObjectFieldSelector{
                        FieldPath: "metadata.namespace",
                    },
                },
            },
            {
                Name: "POD_NAME",
                ValueFrom: &corev1.EnvVarSource{
                    FieldRef: &corev1.ObjectFieldSelector{
                        FieldPath: "metadata.name",
                    },
                },
            },
        },
        Resources: corev1.ResourceRequirements{
            Requests: corev1.ResourceList{
                corev1.ResourceCPU:    "100m",
                corev1.ResourceMemory: "128Mi",
            },
            Limits: corev1.ResourceList{
                corev1.ResourceCPU:    "500m",
                corev1.ResourceMemory: "512Mi",
            },
        },
    }
    pod.Spec.Containers = append(pod.Spec.Containers, sidecar)
    return pod, nil
}

// sendAdmissionResponse sends a simple allow/deny response without patches
func sendAdmissionResponse(w http.ResponseWriter, uid string, allowed bool, message string) {
    resp := admissionv1.AdmissionReview{
        TypeMeta: metav1.TypeMeta{
            APIVersion: "admission.k8s.io/v1",
            Kind:       "AdmissionReview",
        },
        Response: &admissionv1.AdmissionResponse{
            UID:     uid,
            Allowed: allowed,
            Result: &metav1.Status{
                Message: message,
            },
        },
    }
    json.NewEncoder(w).Encode(resp)
}

# Copyright 2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
# Terraform configuration to deploy EKS 1.36 cluster with VPC Lattice add-on
# Requires Terraform >= 1.7.0, AWS provider >= 5.30.0

terraform {
  required_version = ">= 1.7.0"
  required_providers {
    aws = {
      version = ">= 5.30.0"
      source  = "hashicorp/aws"
    }
    kubernetes = {
      version = ">= 2.23.0"
      source  = "hashicorp/kubernetes"
    }
    helm = {
      version = ">= 2.12.0"
      source  = "hashicorp/helm"
    }
  }
}

provider "aws" {
  region = var.aws_region
}

# Variables
variable "aws_region" {
  type        = string
  default     = "us-east-1"
  description = "AWS region to deploy resources"
}

variable "cluster_name" {
  type        = string
  default     = "eks-1-36-lattice-demo"
  description = "Name of the EKS cluster"
}

variable "vpc_cidr" {
  type        = string
  default     = "10.0.0.0/16"
  description = "CIDR block for the VPC"
}

# VPC configuration
resource "aws_vpc" "lattice_demo" {
  cidr_block           = var.vpc_cidr
  enable_dns_support   = true
  enable_dns_hostnames = true

  tags = {
    Name = "lattice-demo-vpc"
  }
}

resource "aws_subnet" "public_subnets" {
  count                   = 3
  vpc_id                  = aws_vpc.lattice_demo.id
  cidr_block              = cidrsubnet(var.vpc_cidr, 8, count.index + 1)
  availability_zone       = "${var.aws_region}${element(["a", "b", "c"], count.index)}"
  map_public_ip_on_launch = true

  tags = {
    Name = "lattice-demo-public-subnet-${count.index}"
  }
}

# EKS cluster IAM role
resource "aws_iam_role" "eks_cluster_role" {
  name = "${var.cluster_name}-cluster-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "eks.amazonaws.com"
        }
      }
    ]
  })

  tags = {
    Name = "${var.cluster_name}-cluster-role"
  }
}

resource "aws_iam_role_policy_attachment" "eks_cluster_policy" {
  role       = aws_iam_role.eks_cluster_role.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonEKSClusterPolicy"
}

# EKS 1.36 cluster
resource "aws_eks_cluster" "lattice_cluster" {
  name     = var.cluster_name
  role_arn = aws_iam_role.eks_cluster_role.arn
  version  = "1.36"

  vpc_config {
    subnet_ids = aws_subnet.public_subnets[*].id
  }

  depends_on = [
    aws_iam_role_policy_attachment.eks_cluster_policy,
  ]

  tags = {
    Name = var.cluster_name
  }
}

# EKS node group IAM role
resource "aws_iam_role" "eks_node_role" {
  name = "${var.cluster_name}-node-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "ec2.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "eks_worker_node_policy" {
  role       = aws_iam_role.eks_node_role.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonEKSWorkerNodePolicy"
}

resource "aws_iam_role_policy_attachment" "eks_cni_policy" {
  role       = aws_iam_role.eks_node_role.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonEKS_CNI_Policy"
}

resource "aws_iam_role_policy_attachment" "ecr_read_only_policy" {
  role       = aws_iam_role.eks_node_role.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly"
}

# IAM role for VPC Lattice add-on (IRSA)
resource "aws_iam_role" "lattice_addon_role" {
  name = "${var.cluster_name}-lattice-addon-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRoleWithWebIdentity"
        Effect = "Allow"
        Principal = {
          Federated = aws_eks_cluster.lattice_cluster.identity[0].oidc[0].issuer
        }
        Condition = {
          StringEquals = {
            "${replace(aws_eks_cluster.lattice_cluster.identity[0].oidc[0].issuer, "https://", "")}:sub" = "system:serviceaccount:lattice-system:lattice-controller"
          }
        }
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "lattice_addon_policy" {
  role       = aws_iam_role.lattice_addon_role.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonVPCLatticeFullAccess"
}

# EKS node group
resource "aws_eks_node_group" "lattice_nodes" {
  cluster_name    = aws_eks_cluster.lattice_cluster.name
  node_group_name = "lattice-demo-nodes"
  node_role_arn   = aws_iam_role.eks_node_role.arn
  subnet_ids      = aws_subnet.public_subnets[*].id

  scaling_config {
    desired_size = 3
    max_size     = 5
    min_size     = 1
  }

  instance_types = ["m6i.large"]

  depends_on = [
    aws_iam_role_policy_attachment.eks_worker_node_policy,
    aws_iam_role_policy_attachment.eks_cni_policy,
    aws_iam_role_policy_attachment.ecr_read_only_policy,
  ]

  tags = {
    Name = "${var.cluster_name}-node-group"
  }
}

# Kubernetes provider configuration
provider "kubernetes" {
  host                   = aws_eks_cluster.lattice_cluster.endpoint
  cluster_ca_certificate = base64decode(aws_eks_cluster.lattice_cluster.certificate_authority[0].data)
  exec {
    api_version = "client.authentication.k8s.io/v1beta1"
    command     = "aws"
    args        = ["eks", "get-token", "--cluster-name", aws_eks_cluster.lattice_cluster.name]
  }
}

# Helm provider configuration
provider "helm" {
  kubernetes {
    host                   = aws_eks_cluster.lattice_cluster.endpoint
    cluster_ca_certificate = base64decode(aws_eks_cluster.lattice_cluster.certificate_authority[0].data)
    exec {
      api_version = "client.authentication.k8s.io/v1beta1"
      command     = "aws"
      args        = ["eks", "get-token", "--cluster-name", aws_eks_cluster.lattice_cluster.name]
    }
  }
}

# Deploy VPC Lattice EKS add-on via Helm
resource "helm_release" "vpc_lattice_addon" {
  name       = "aws-vpc-lattice-controller"
  repository = "https://aws.github.io/eks-charts"
  chart      = "aws-vpc-lattice-controller"
  version    = "1.2.4"
  namespace  = "lattice-system"

  create_namespace = true

  set {
    name  = "clusterName"
    value = var.cluster_name
  }

  set {
    name  = "region"
    value = var.aws_region
  }

  set {
    name  = "serviceAccount.annotations.eks\.amazonaws\.com/role-arn"
    value = aws_iam_role.lattice_addon_role.arn
  }

  depends_on = [
    aws_eks_node_group.lattice_nodes,
  ]
}

# Output the EKS cluster endpoint
output "eks_cluster_endpoint" {
  value = aws_eks_cluster.lattice_cluster.endpoint
}

output "vpc_lattice_addon_status" {
  value = helm_release.vpc_lattice_addon.status
}

# Copyright 2024 InfoQ Contributor. All Rights Reserved.
# Benchmark script to compare VPC Lattice vs VPC Peering cross-cluster latency
# Requires: boto3>=1.34.0, requests>=2.31.0, numpy>=1.26.0, prometheus-client>=0.19.0
import os
import time
import json
import logging
import argparse
import numpy as np
from typing import List, Dict
import boto3
from botocore.exceptions import ClientError
import requests
from prometheus_client import start_http_server, Gauge

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

# Prometheus metrics
LATTICE_LATENCY = Gauge("benchmark_lattice_latency_ms", "Cross-cluster latency via VPC Lattice", ["percentile"])
PEERING_LATENCY = Gauge("benchmark_peering_latency_ms", "Cross-cluster latency via VPC Peering", ["percentile"])
ERROR_COUNTER = Gauge("benchmark_errors_total", "Total benchmark errors", ["method"])

class LatencyBenchmark:
    def __init__(self, lattice_endpoint: str, peering_endpoint: str, num_requests: int = 1000):
        self.lattice_endpoint = lattice_endpoint
        self.peering_endpoint = peering_endpoint
        self.num_requests = num_requests
        self.session = requests.Session()
        # Set timeout for all requests
        self.session.request = lambda method, url, **kwargs: super(requests.Session, self.session).request(
            method, url, timeout=(2, 5), **kwargs
        )

    def run_lattice_benchmark(self) -> List[float]:
        """Run latency benchmark against VPC Lattice endpoint"""
        latencies = []
        for i in range(self.num_requests):
            try:
                start = time.perf_counter()
                resp = self.session.get(f"{self.lattice_endpoint}/health")
                resp.raise_for_status()
                end = time.perf_counter()
                latency_ms = (end - start) * 1000
                latencies.append(latency_ms)
                if i % 100 == 0:
                    logger.info(f"Lattice benchmark: {i}/{self.num_requests} requests completed")
            except requests.exceptions.RequestException as e:
                logger.error(f"Lattice request failed: {e}")
                ERROR_COUNTER.labels(method="lattice").inc()
                continue
        return latencies

    def run_peering_benchmark(self) -> List[float]:
        """Run latency benchmark against VPC Peering endpoint"""
        latencies = []
        for i in range(self.num_requests):
            try:
                start = time.perf_counter()
                resp = self.session.get(f"{self.peering_endpoint}/health")
                resp.raise_for_status()
                end = time.perf_counter()
                latency_ms = (end - start) * 1000
                latencies.append(latency_ms)
                if i % 100 == 0:
                    logger.info(f"Peering benchmark: {i}/{self.num_requests} requests completed")
            except requests.exceptions.RequestException as e:
                logger.error(f"Peering request failed: {e}")
                ERROR_COUNTER.labels(method="peering").inc()
                continue
        return latencies

    def calculate_percentiles(self, latencies: List[float]) -> Dict[str, float]:
        """Calculate p50, p90, p99, p99.9 percentiles from latency list"""
        if not latencies:
            return {"p50": 0.0, "p90": 0.0, "p99": 0.0, "p99.9": 0.0}
        sorted_latencies = sorted(latencies)
        return {
            "p50": np.percentile(sorted_latencies, 50),
            "p90": np.percentile(sorted_latencies, 90),
            "p99": np.percentile(sorted_latencies, 99),
            "p99.9": np.percentile(sorted_latencies, 99.9),
        }

    def run(self):
        """Execute full benchmark and report results"""
        logger.info(f"Starting benchmark: {self.num_requests} requests per method")
        logger.info(f"Lattice endpoint: {self.lattice_endpoint}")
        logger.info(f"Peering endpoint: {self.peering_endpoint}")

        # Run benchmarks
        lattice_latencies = self.run_lattice_benchmark()
        peering_latencies = self.run_peering_benchmark()

        # Calculate percentiles
        lattice_percentiles = self.calculate_percentiles(lattice_latencies)
        peering_percentiles = self.calculate_percentiles(peering_latencies)

        # Update Prometheus metrics
        for percentile, value in lattice_percentiles.items():
            LATTICE_LATENCY.labels(percentile=percentile).set(value)
        for percentile, value in peering_percentiles.items():
            PEERING_LATENCY.labels(percentile=percentile).set(value)

        # Print results
        print("\n=== Benchmark Results ===")
        print(f"Total requests per method: {self.num_requests}")
        print(f"Lattice successful requests: {len(lattice_latencies)}")
        print(f"Peering successful requests: {len(peering_latencies)}")

        print("\n--- VPC Lattice Latency ---")
        for percentile, value in lattice_percentiles.items():
            print(f"{percentile}: {value:.2f}ms")

        print("\n--- VPC Peering Latency ---")
        for percentile, value in peering_percentiles.items():
            print(f"{percentile}: {value:.2f}ms")

        # Calculate improvement
        if lattice_percentiles["p99"] > 0 and peering_percentiles["p99"] > 0:
            improvement = ((peering_percentiles["p99"] - lattice_percentiles["p99"]) / peering_percentiles["p99"]) * 100
            print(f"\nVPC Lattice p99 latency improvement: {improvement:.1f}%")

        # Save results to JSON
        results = {
            "lattice": {
                "successful_requests": len(lattice_latencies),
                "percentiles": lattice_percentiles,
            },
            "peering": {
                "successful_requests": len(peering_latencies),
                "percentiles": peering_percentiles,
            },
            "improvement_percent": improvement if 'improvement' in locals() else 0.0,
        }
        with open("benchmark_results.json", "w") as f:
            json.dump(results, f, indent=2)
        logger.info("Results saved to benchmark_results.json")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="VPC Lattice vs VPC Peering Latency Benchmark")
    parser.add_argument("--lattice-endpoint", required=True, help="VPC Lattice service endpoint URL")
    parser.add_argument("--peering-endpoint", required=True, help="VPC Peering service endpoint URL")
    parser.add_argument("--num-requests", type=int, default=1000, help="Number of requests per benchmark (default: 1000)")
    parser.add_argument("--metrics-port", type=int, default=9090, help="Prometheus metrics port (default: 9090)")
    args = parser.parse_args()

    # Start Prometheus metrics server
    start_http_server(args.metrics_port)
    logger.info(f"Prometheus metrics server started on port {args.metrics_port}")

    benchmark = LatencyBenchmark(
        lattice_endpoint=args.lattice_endpoint,
        peering_endpoint=args.peering_endpoint,
        num_requests=args.num_requests,
    )
    try:
        benchmark.run()
    except KeyboardInterrupt:
        logger.info("Benchmark interrupted by user")
    except Exception as e:
        logger.error(f"Benchmark failed: {e}")
        ERROR_COUNTER.labels(method="global").inc()
        raise

apiVersion: v1
kind: ServiceAccount
metadata:
  name: lattice-controller
  namespace: lattice-system
  annotations:
    eks.amazonaws.com/role-arn: arn:aws:iam::123456789012:role/eks-1-36-lattice-demo-lattice-addon-role

apiVersion: apps/v1
kind: Deployment
metadata:
  name: payments-processor
spec:
  template:
    metadata:
      annotations:
        lattice.amazonaws.com/service-name: payments
        lattice.amazonaws.com/sidecar-cpu-request: "250m"
        lattice.amazonaws.com/sidecar-memory-request: "256Mi"
    spec:
      containers:
      - name: payments-app
        image: my-org/payments:1.2.3

apiVersion: batch/v1
kind: Job
metadata:
  name: lattice-connectivity-check
spec:
  template:
    spec:
      containers:
      - name: check
        image: curlimages/curl:8.5.0
        command: ["sh", "-c"]
        args: ["curl -f --connect-timeout 5 http://payments.lattice.local/health || exit 1"]
      restartPolicy: Never
  backoffLimit: 2