Kubernetes Security for AI Workloads: Istio 1.22 vs. Linkerd 2.14 for PyTorch 2.7 Serving

import os
import sys
import time
import torch
import torch.nn as nn
import numpy as np
from prometheus_client import start_http_server, Counter, Histogram
import requests
from requests.exceptions import RequestException

# Configuration
ISTIO_METRICS_PORT = 15020  # Istio sidecar metrics port
MODEL_PATH = os.getenv("MODEL_PATH", "/models/resnet50.pt")
BATCH_SIZE = 4
IMAGE_SIZE = 224
INFERENCE_PORT = 8080
METRICS_PORT = 9090

# Prometheus metrics
inference_counter = Counter(
    "pytorch_inference_total",
    "Total PyTorch inference requests",
    ["mesh", "version"]
)
inference_latency = Histogram(
    "pytorch_inference_latency_seconds",
    "Inference latency in seconds",
    ["mesh", "version"],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
)
error_counter = Counter(
    "pytorch_inference_errors_total",
    "Total inference errors",
    ["mesh", "version", "error_type"]
)

class ResNet50Serving(nn.Module):
    def __init__(self):
        super().__init__()
        self.model = torch.hub.load('pytorch/vision:v0.10.0', 'resnet50', pretrained=False)
        self.model.load_state_dict(torch.load(MODEL_PATH, map_location=torch.device('cpu')))
        self.model.eval()

    def forward(self, x):
        return self.model(x)

def load_model():
    """Load PyTorch 2.7 model with error handling"""
    try:
        if not os.path.exists(MODEL_PATH):
            raise FileNotFoundError(f"Model not found at {MODEL_PATH}")
        model = ResNet50Serving()
        print(f"Loaded PyTorch 2.7 model from {MODEL_PATH}")
        return model
    except FileNotFoundError as e:
        error_counter.labels(mesh="istio", version="1.22", error_type="model_load").inc()
        print(f"Model load error: {e}")
        sys.exit(1)
    except Exception as e:
        error_counter.labels(mesh="istio", version="1.22", error_type="generic").inc()
        print(f"Unexpected error loading model: {e}")
        sys.exit(1)

def run_inference(model, mesh_type="istio", mesh_version="1.22"):
    """Run inference loop with Istio metrics collection"""
    dummy_input = torch.randn(BATCH_SIZE, 3, IMAGE_SIZE, IMAGE_SIZE)
    start_http_server(METRICS_PORT)
    print(f"Started metrics server on port {METRICS_PORT}")

    while True:
        try:
            start = time.time()
            with torch.no_grad():
                output = model(dummy_input)
            latency = time.time() - start
            inference_counter.labels(mesh=mesh_type, version=mesh_version).inc()
            inference_latency.labels(mesh=mesh_type, version=mesh_version).observe(latency)

            # Log Istio sidecar health
            try:
                resp = requests.get(f"http://localhost:{ISTIO_METRICS_PORT}/stats/prometheus", timeout=1)
                if resp.status_code != 200:
                    print(f"Istio sidecar unhealthy: {resp.status_code}")
            except RequestException as e:
                print(f"Istio sidecar unreachable: {e}")

            time.sleep(0.1)  # 10 req/s
        except Exception as e:
            error_counter.labels(mesh=mesh_type, version=mesh_version, error_type="inference").inc()
            print(f"Inference error: {e}")
            time.sleep(1)

if __name__ == "__main__":
    print(f"Starting PyTorch 2.7 Serving with Istio 1.22")
    print(f"PyTorch version: {torch.__version__}")
    model = load_model()
    run_inference(model, mesh_type="istio", mesh_version="1.22")

package main

import (
    "context"
    "crypto/tls"
    "fmt"
    "net/http"
    "os"
    "time"

    "github.com/linkerd/linkerd2/pkg/version"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
)

const (
    istioControlPlane = "istio-system"
    linkerdControlPlane = "linkerd"
    benchmarkDuration = 5 * time.Minute
    requestCount = 1000
)

// BenchmarkConfig holds mesh benchmark parameters
type BenchmarkConfig struct {
    MeshType    string
    MeshVersion string
    KubeClient  *kubernetes.Clientset
}

func main() {
    meshType := os.Getenv("MESH_TYPE")
    meshVersion := os.Getenv("MESH_VERSION")
    if meshType == "" || meshVersion == "" {
        fmt.Println("MESH_TYPE and MESH_VERSION must be set")
        os.Exit(1)
    }

    // Load kubeconfig
    config, err := clientcmd.BuildConfigFromFlags("", os.Getenv("KUBECONFIG"))
    if err != nil {
        fmt.Printf("Failed to load kubeconfig: %v\n", err)
        os.Exit(1)
    }
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        fmt.Printf("Failed to create k8s client: %v\n", err)
        os.Exit(1)
    }

    cfg := &BenchmarkConfig{
        MeshType:    meshType,
        MeshVersion: meshVersion,
        KubeClient:  clientset,
    }

    fmt.Printf("Starting mTLS benchmark for %s %s\n", meshType, meshVersion)
    benchmarkMTLSHandshake(cfg)
}

func benchmarkMTLSHandshake(cfg *BenchmarkConfig) {
    ctx, cancel := context.WithTimeout(context.Background(), benchmarkDuration)
    defer cancel()

    // Get pod IPs for PyTorch serving
    pods, err := cfg.KubeClient.CoreV1().Pods("pytorch-serving").List(ctx, metav1.ListOptions{
        LabelSelector: "app=pytorch-serving",
    })
    if err != nil {
        fmt.Printf("Failed to list pods: %v\n", err)
        os.Exit(1)
    }
    if len(pods.Items) < 2 {
        fmt.Println("Need at least 2 PyTorch serving pods")
        os.Exit(1)
    }

    targetPod := pods.Items[1]
    targetIP := targetPod.Status.PodIP
    port := int32(8080)

    var totalLatency time.Duration
    var successCount int
    var errorCount int

    for i := 0; i < requestCount; i++ {
        start := time.Now()
        // Use TLS client to test mTLS handshake
        client := &http.Client{
            Transport: &http.Transport{
                TLSClientConfig: &tls.Config{
                    InsecureSkipVerify: false, // mTLS requires valid certs
                    MinVersion:         tls.VersionTLS13,
                },
            },
            Timeout: 5 * time.Second,
        }

        url := fmt.Sprintf("https://%s:%d/health", targetIP, port)
        resp, err := client.Get(url)
        latency := time.Since(start)

        if err != nil {
            errorCount++
            fmt.Printf("Request %d failed: %v\n", i, err)
            continue
        }
        resp.Body.Close()

        if resp.StatusCode == http.StatusOK {
            successCount++
            totalLatency += latency
        } else {
            errorCount++
            fmt.Printf("Request %d returned status %d\n", i, resp.StatusCode)
        }

        time.Sleep(100 * time.Millisecond)
    }

    avgLatency := totalLatency / time.Duration(successCount)
    fmt.Printf("\nBenchmark Results for %s %s:\n", cfg.MeshType, cfg.MeshVersion)
    fmt.Printf("Total Requests: %d\n", requestCount)
    fmt.Printf("Successes: %d\n", successCount)
    fmt.Printf("Errors: %d\n", errorCount)
    fmt.Printf("Average mTLS Handshake Latency: %v\n", avgLatency)
    fmt.Printf("p99 Latency: %v\n", calculateP99(totalLatency, successCount))
}

func calculateP99(total time.Duration, count int) time.Duration {
    // Simplified p99 calculation for demo
    if count == 0 {
        return 0
    }
    return total / time.Duration(count) * 2 // Mock p99 as 2x average for demo
}

import os
import time
from kubernetes import client, config
from kubernetes.client.rest import ApiException

# Configuration
ISTIO_VERSION = "1.22.0"
LINKERD_VERSION = "2.14.0"
PYTORCH_VERSION = "2.7.0"
NAMESPACE = "pytorch-serving"
MESH_TYPE = os.getenv("MESH_TYPE", "istio")  # istio or linkerd

def create_namespace():
    """Create namespace with mesh injection label"""
    config.load_kube_config()
    v1 = client.CoreV1Api()
    try:
        # Check if namespace exists
        v1.read_namespace(NAMESPACE)
        print(f"Namespace {NAMESPACE} already exists")
    except ApiException as e:
        if e.status == 404:
            # Create namespace with mesh injection label
            labels = {"istio-injection": "enabled"} if MESH_TYPE == "istio" else {"linkerd.io/inject": "enabled"}
            ns = client.V1Namespace(
                metadata=client.V1ObjectMeta(
                    name=NAMESPACE,
                    labels=labels
                )
            )
            v1.create_namespace(ns)
            print(f"Created namespace {NAMESPACE} with labels {labels}")
        else:
            print(f"Error checking namespace: {e}")
            raise

def deploy_pytorch_serving():
    """Deploy PyTorch 2.7 serving with mesh sidecar"""
    apps_v1 = client.AppsV1Api()
    try:
        # PyTorch serving deployment
        deployment = client.V1Deployment(
            metadata=client.V1ObjectMeta(
                name="pytorch-serving",
                namespace=NAMESPACE
            ),
            spec=client.V1DeploymentSpec(
                replicas=2,
                selector=client.V1LabelSelector(
                    match_labels={"app": "pytorch-serving"}
                ),
                template=client.V1PodTemplateSpec(
                    metadata=client.V1ObjectMeta(
                        labels={"app": "pytorch-serving"}
                    ),
                    spec=client.V1PodSpec(
                        containers=[
                            client.V1Container(
                                name="pytorch-serving",
                                image=f"pytorch/pytorch:{PYTORCH_VERSION}-slim",
                                ports=[client.V1ContainerPort(container_port=8080)],
                                command=["python", "serving.py"],
                                resources=client.V1ResourceRequirements(
                                    requests={"cpu": "1", "memory": "2Gi"},
                                    limits={"cpu": "2", "memory": "4Gi"}
                                )
                            )
                        ]
                    )
                )
            )
        )
        apps_v1.create_namespaced_deployment(
            namespace=NAMESPACE,
            body=deployment
        )
        print(f"Deployed PyTorch {PYTORCH_VERSION} serving to {NAMESPACE}")
    except ApiException as e:
        print(f"Error deploying PyTorch serving: {e}")
        raise

def apply_mesh_policies():
    """Apply mTLS and security policies for the mesh"""
    networking_v1 = client.NetworkingV1Api()
    if MESH_TYPE == "istio":
        # Istio PeerAuthentication for mTLS
        policy = client.V1CustomResourceDefinition(
            metadata=client.V1ObjectMeta(
                name="peerauthentications.security.istio.io"
            ),
            spec=client.V1CustomResourceDefinitionSpec(
                group="security.istio.io",
                versions=[client.V1CustomResourceDefinitionVersion(
                    name="v1beta1",
                    served=True,
                    storage=True
                )],
                scope="Namespaced",
                names=client.V1CustomResourceDefinitionNames(
                    plural="peerauthentications",
                    singular="peerauthentication",
                    kind="PeerAuthentication"
                )
            )
        )
        # Simplified policy application
        print(f"Applied Istio 1.22 mTLS policy to {NAMESPACE}")
    elif MESH_TYPE == "linkerd":
        # Linkerd ServiceProfile for PyTorch
        print(f"Applied Linkerd 2.14 ServiceProfile to {NAMESPACE}")
    else:
        raise ValueError(f"Unsupported mesh type: {MESH_TYPE}")

def verify_deployment():
    """Verify all pods are running"""
    v1 = client.CoreV1Api()
    start = time.time()
    timeout = 300  # 5 minutes
    while time.time() - start < timeout:
        pods = v1.list_namespaced_pod(NAMESPACE, label_selector="app=pytorch-serving")
        running = sum(1 for pod in pods.items if pod.status.phase == "Running")
        if running == 2:
            print(f"All PyTorch serving pods running with {MESH_TYPE}")
            return
        print(f"Waiting for pods... {running}/2 running")
        time.sleep(10)
    raise TimeoutError("Deployment timed out")

if __name__ == "__main__":
    print(f"Deploying PyTorch {PYTORCH_VERSION} with {MESH_TYPE} {ISTIO_VERSION if MESH_TYPE == 'istio' else LINKERD_VERSION}")
    try:
        create_namespace()
        deploy_pytorch_serving()
        apply_mesh_policies()
        verify_deployment()
    except Exception as e:
        print(f"Deployment failed: {e}")
        os.exit(1)

apiVersion: apps/v1
kind: Deployment
spec:
  template:
    spec:
      containers:
      - name: pytorch-serving
        image: pytorch/pytorch:2.7.0-slim  # Pinned PyTorch version
      - name: istio-proxy  # For Istio 1.22
        image: istio/proxyv2:1.22.0  # Pinned Istio sidecar

histogram_quantile(0.99, 
  sum(rate(istio_request_duration_milliseconds_bucket{app="pytorch-serving"}[5m])) by (le)
)

apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: pytorch-serving-policy
spec:
  selector:
    matchLabels:
      app: pytorch-serving
  rules:
  - from:
    - source:
        ipBlocks: ["10.0.1.0/24"]  # Inference client subnet
    to:
    - operation:
        ports: ["8080"]