How throughput with Mistral 2 and Kubernetes 1.30: Lessons Learned

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

def create_mistral_vllm_deployment():
    \"\"\"Deploys a tuned vLLM 0.4.0 pod serving Mistral 2 7B to Kubernetes 1.30.
    Includes resource limits, startup probes, and nftables-compatible network config.\
    \"\"\"
    # Load kubeconfig from default path or in-cluster config
    try:
        config.load_kube_config()
    except Exception as e:
        print(f"Failed to load kubeconfig, trying in-cluster config: {e}")
        try:
            config.load_incluster_config()
        except Exception as e:
            print(f"Failed to load in-cluster config: {e}")
            sys.exit(1)

    # Initialize API clients
    apps_v1 = client.AppsV1Api()
    core_v1 = client.CoreV1Api()

    # Define the Deployment object
    deployment = client.V1Deployment(
        metadata=client.V1ObjectMeta(
            name="mistral2-vllm",
            namespace="inference",
            labels={"app": "mistral2", "runtime": "vllm"}
        ),
        spec=client.V1DeploymentSpec(
            replicas=3,
            selector=client.V1LabelSelector(
                match_labels={"app": "mistral2", "runtime": "vllm"}
            ),
            template=client.V1PodTemplateSpec(
                metadata=client.V1ObjectMeta(
                    labels={"app": "mistral2", "runtime": "vllm"},
                    annotations={"prometheus.io/scrape": "true", "prometheus.io/port": "9090"}
                ),
                spec=client.V1PodSpec(
                    containers=[
                        client.V1Container(
                            name="vllm",
                            image="vllm/vllm-openai:0.4.0",
                            args=[
                                "--model", "mistralai/Mistral-2-7B-Instruct-v0.2",
                                "--dtype", "half",
                                "--max-model-len", "2048",
                                "--gpu-memory-utilization", "0.9",
                                "--enable-prefix-caching"
                            ],
                            ports=[client.V1ContainerPort(container_port=8000, name="http")],
                            resources=client.V1ResourceRequirements(
                                requests={"cpu": "2", "memory": "4Gi"},
                                limits={"cpu": "4", "memory": "8Gi"}
                            ),
                            startup_probe=client.V1StartupProbe(
                                http_get=client.V1HTTPGetAction(
                                    path="/health",
                                    port=8000
                                ),
                                initial_delay_seconds=30,
                                period_seconds=5,
                                failure_threshold=10
                            ),
                            readiness_probe=client.V1ReadinessProbe(
                                http_get=client.V1HTTPGetAction(
                                    path="/health",
                                    port=8000
                                ),
                                initial_delay_seconds=5,
                                period_seconds=3
                            ),
                            env=[
                                client.V1EnvVar(
                                    name="HUGGING_FACE_HUB_TOKEN",
                                    value_from=client.V1EnvVarSource(
                                        secret_key_ref=client.V1SecretKeySelector(
                                            name="huggingface-secret",
                                            key="token"
                                        )
                                    )
                                )
                            ]
                        )
                    ],
                    node_selector={"kubernetes.io/arch": "amd64"},
                    tolerations=[
                        client.V1Toleration(
                            key="nvidia.com/gpu",
                            operator="exists",
                            effect="NoSchedule"
                        )
                    ]
                )
            )
        )
    )

    # Create the deployment
    try:
        apps_v1.create_namespaced_deployment(
            namespace="inference",
            body=deployment
        )
        print("Successfully created Mistral 2 vLLM deployment")
    except ApiException as e:
        if e.status == 409:
            print(f"Deployment already exists, updating...")
            apps_v1.replace_namespaced_deployment(
                name="mistral2-vllm",
                namespace="inference",
                body=deployment
            )
        else:
            print(f"API exception when creating deployment: {e}")
            sys.exit(1)

    # Create the ClusterIP service
    service = client.V1Service(
        metadata=client.V1ObjectMeta(
            name="mistral2-vllm-svc",
            namespace="inference",
            labels={"app": "mistral2", "runtime": "vllm"}
        ),
        spec=client.V1ServiceSpec(
            selector={"app": "mistral2", "runtime": "vllm"},
            ports=[client.V1ServicePort(port=8000, target_port=8000, name="http")],
            type="ClusterIP"
        )
    )

    try:
        core_v1.create_namespaced_service(
            namespace="inference",
            body=service
        )
        print("Successfully created Mistral 2 vLLM service")
    except ApiException as e:
        if e.status == 409:
            print(f"Service already exists, updating...")
            core_v1.replace_namespaced_service(
                name="mistral2-vllm-svc",
                namespace="inference",
                body=service
            )
        else:
            print(f"API exception when creating service: {e}")
            sys.exit(1)

if __name__ == "__main__":
    # Ensure the inference namespace exists
    core_v1 = client.CoreV1Api()
    try:
        core_v1.create_namespace(
            body=client.V1Namespace(
                metadata=client.V1ObjectMeta(name="inference")
            )
        )
        print("Created inference namespace")
    except ApiException as e:
        if e.status != 409:
            print(f"Failed to create namespace: {e}")
            sys.exit(1)
    create_mistral_vllm_deployment()

import http from 'k6/http';
import { check, sleep, trend, rate } from 'k6';
import { randomString } from 'https://jslib.k6.io/k6-utils/1.4.0/index.js';

// Custom metrics for Mistral 2 specific tracking
const mistralLatency = new trend('mistral_latency_ms');
const mistralErrorRate = new rate('mistral_error_rate');
const mistralTokenRate = new trend('mistral_tokens_per_second');

export const options = {
  stages: [
    { duration: '30s', target: 500 }, // Ramp up to 500 RPS
    { duration: '5m', target: 500 },  // Sustained load
    { duration: '30s', target: 0 },   // Ramp down
  ],
  thresholds: {
    'http_req_duration': ['p(99)<200'], // p99 latency must be under 200ms
    'mistral_error_rate': ['rate<0.01'], // Error rate under 1%
  },
  ext: {
    loadimpact: {
      projectID: 123456, // Replace with real project ID
      name: 'Mistral 2 Kubernetes 1.30 Throughput Test'
    }
  }
};

// Fixed 1024-token input prompt (truncated for brevity, real prompt is 1024 tokens)
const BASE_PROMPT = `You are a helpful assistant. Explain the difference between throughput and latency in distributed systems, providing three real-world examples from cloud-native deployments.`;

export default function () {
  const url = 'http://mistral2-vllm-svc.inference.svc.cluster.local:8000/v1/completions';
  const payload = JSON.stringify({
    model: 'mistralai/Mistral-2-7B-Instruct-v0.2',
    prompt: BASE_PROMPT,
    max_tokens: 256,
    temperature: 0.7,
    stream: false
  });

  const params = {
    headers: {
      'Content-Type': 'application/json',
    },
    timeout: '30s', // 30 second timeout for Mistral inference
  };

  const startTime = new Date().getTime();
  const response = http.post(url, payload, params);
  const endTime = new Date().getTime();
  const latencyMs = endTime - startTime;

  // Track custom metrics
  mistralLatency.add(latencyMs);

  // Check response validity
  const isSuccess = check(response, {
    'status is 200': (r) => r.status === 200,
    'response has completion': (r) => {
      try {
        const body = JSON.parse(r.body);
        return body.choices && body.choices.length > 0 && body.choices[0].text;
      } catch (e) {
        return false;
      }
    },
    'latency under 500ms': (r) => latencyMs < 500,
  });

  if (!isSuccess) {
    mistralErrorRate.add(1);
    console.error(`Request failed: ${response.status} ${response.body}`);
  } else {
    mistralErrorRate.add(0);
    // Calculate tokens per second if usage data is present
    try {
      const body = JSON.parse(response.body);
      if (body.usage && body.usage.completion_tokens) {
        const tokensPerSecond = (body.usage.completion_tokens / (latencyMs / 1000)).toFixed(2);
        mistralTokenRate.add(Number(tokensPerSecond));
      }
    } catch (e) {
      // Ignore parsing errors for token rate
    }
  }

  sleep(0.1); // 100ms sleep between requests per VU
}

export function handleSummary(data) {
  return {
    'stdout': textSummary(data, { indent: ' ', enableColors: true }),
    'mistral-benchmark-summary.json': JSON.stringify(data),
  };
}

// Helper to generate text summary
function textSummary(data, options) {
  const indent = options.indent || '  ';
  const color = options.enableColors ? (s) => `\x1b[32m${s}\x1b[0m` : (s) => s;
  let summary = `${indent}Mistral 2 Throughput Benchmark Summary\n`;
  summary += `${indent}=========================================\n`;
  summary += `${indent}Total Requests: ${data.metrics.http_reqs.values.count}\n`;
  summary += `${indent}Mean Throughput (RPS): ${(data.metrics.http_reqs.values.rate).toFixed(2)}\n`;
  summary += `${indent}p99 Latency (ms): ${data.metrics.http_req_duration.values['p(99)']}\n`;
  summary += `${indent}Error Rate: ${(data.metrics.mistral_error_rate.values.rate * 100).toFixed(2)}%\n`;
  return color(summary);
}

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/kedacore/keda/v2/pkg/scalers/externalscaler"
    "github.com/prometheus/client_golang/api"
    v1 "github.com/prometheus/client_golang/api/prometheus/v1"
    "github.com/prometheus/common/model"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
)

const (
    prometheusURL = "http://prometheus-k8s.monitoring.svc.cluster.local:9090"
    targetRPS     = 800.0 // Target 800 RPS per pod
    minReplicas   = 1
    maxReplicas   = 10
    deploymentNS  = "inference"
    deploymentName = "mistral2-vllm"
)

// MistralScaler implements the KEDA External Scaler interface
type MistralScaler struct {
    clientset *kubernetes.Clientset
    promAPI   v1.API
}

// NewMistralScaler creates a new scaler with k8s and Prometheus clients
func NewMistralScaler() (*MistralScaler, error) {
    // Load kubeconfig
    config, err := clientcmd.BuildConfigFromFlags("", os.Getenv("KUBECONFIG"))
    if err != nil {
        // Try in-cluster config if kubeconfig fails
        config, err = clientcmd.BuildConfigFromFlags("", "")
        if err != nil {
            return nil, fmt.Errorf("failed to load kubeconfig: %v", err)
        }
    }

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

    // Create Prometheus client
    promConfig := api.Config{Address: prometheusURL}
    promClient, err := api.NewClient(promConfig)
    if err != nil {
        return nil, fmt.Errorf("failed to create Prometheus client: %v", err)
    }
    promAPI := v1.NewAPI(promClient)

    return &MistralScaler{
        clientset: clientset,
        promAPI:   promAPI,
    }, nil
}

// GetMetrics returns the current RPS for the Mistral deployment
func (s *MistralScaler) GetMetrics(ctx context.Context, metricName string) ([]externalscaler.MetricValue, error) {
    // Query Prometheus for Mistral 2 RPS
    query := fmt.Sprintf(`sum(rate(http_requests_total{app="mistral2", runtime="vllm"}[1m]))`)
    result, warnings, err := s.promAPI.Query(ctx, query, time.Now())
    if err != nil {
        return nil, fmt.Errorf("prometheus query failed: %v", err)
    }
    if len(warnings) > 0 {
        log.Printf("Prometheus warnings: %v", warnings)
    }

    // Parse result
    var currentRPS float64
    switch r := result.(type) {
    case model.Vector:
        if len(r) == 0 {
            currentRPS = 0
        } else {
            currentRPS = float64(r[0].Value)
        }
    default:
        return nil, fmt.Errorf("unexpected Prometheus result type: %T", result)
    }

    // Calculate desired replicas
    desiredReplicas := int32(currentRPS / targetRPS)
    if desiredReplicas < minReplicas {
        desiredReplicas = minReplicas
    }
    if desiredReplicas > maxReplicas {
        desiredReplicas = maxReplicas
    }

    // Update deployment replicas
    deployment, err := s.clientset.AppsV1().Deployments(deploymentNS).Get(ctx, deploymentName, metav1.GetOptions{})
    if err != nil {
        return nil, fmt.Errorf("failed to get deployment: %v", err)
    }

    if *deployment.Spec.Replicas != desiredReplicas {
        log.Printf("Scaling deployment from %d to %d replicas (current RPS: %.2f)", *deployment.Spec.Replicas, desiredReplicas, currentRPS)
        deployment.Spec.Replicas = &desiredReplicas
        _, err = s.clientset.AppsV1().Deployments(deploymentNS).Update(ctx, deployment, metav1.UpdateOptions{})
        if err != nil {
            return nil, fmt.Errorf("failed to update deployment: %v", err)
        }
    }

    return []externalscaler.MetricValue{{
        MetricName:       metricName,
        Value:            int64(currentRPS),
        DesiredReplicas:  desiredReplicas,
    }}, nil
}

// IsActive returns true if there is any traffic to the deployment
func (s *MistralScaler) IsActive(ctx context.Context) (bool, error) {
    query := fmt.Sprintf(`sum(rate(http_requests_total{app="mistral2", runtime="vllm"}[5m])) > 0`)
    result, _, err := s.promAPI.Query(ctx, query, time.Now())
    if err != nil {
        return false, fmt.Errorf("prometheus query failed: %v", err)
    }

    switch r := result.(type) {
    case model.Vector:
        return len(r) > 0, nil
    default:
        return false, nil
    }
}

// GetMetricSpec returns the metric spec for KEDA
func (s *MistralScaler) GetMetricSpec() ([]externalscaler.MetricSpec, error) {
    return []externalscaler.MetricSpec{{
        MetricName:   "mistral-rps",
        TargetValue:  targetRPS,
    }}, nil
}

func main() {
    scaler, err := NewMistralScaler()
    if err != nil {
        log.Fatalf("Failed to create scaler: %v", err)
    }

    ctx := context.Background()
    ticker := time.NewTicker(30 * time.Second) // Check every 30 seconds
    defer ticker.Stop()

    log.Println("Starting Mistral 2 KEDA scaler...")
    for {
        select {
        case <-ticker.C:
            _, err := scaler.GetMetrics(ctx, "mistral-rps")
            if err != nil {
                log.Printf("Error getting metrics: %v", err)
            }
        }
    }
}

kubectl edit configmap kube-proxy -n kube-system
# Change mode: "iptables" to mode: "nftables"
# Then restart kube-proxy pods:
kubectl rollout restart daemonset kube-proxy -n kube-system

vllm serve mistralai/Mistral-2-7B-Instruct-v0.2 \
  --dtype half \
  --gpu-memory-utilization 0.9 \
  --enable-prefix-caching \
  --max-model-len 2048

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: mistral2-vllm-scaler
  namespace: inference
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: mistral2-vllm
  minReplicaCount: 1
  maxReplicaCount: 10
  triggers:
  - type: prometheus
    metadata:
      serverAddress: http://prometheus-k8s.monitoring.svc.cluster.local:9090
      metricName: mistral-rps
      threshold: '800'
      query: sum(rate(http_requests_total{app="mistral2"}[1m]))