Retrospective: Running 10k Concurrent Ollama 0.4 Local LLM Instances for Internal Developer Assistants

import asyncio
import aiohttp
import logging
import os
import uuid
from dataclasses import dataclass
from typing import Optional, Dict, Any

# Configure logging for provisioner audit trail
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("ollama-provisioner")

@dataclass
class OllamaInstanceConfig:
    """Configuration for a single Ollama 0.4 instance"""
    model: str = "llama3.1:8b-q4_0"
    cpu_shares: int = 2048  # 2 CPU cores equivalent
    memory_mb: int = 8192   # 8GB RAM
    port: Optional[int] = None
    instance_id: str = ""

@dataclass
class OllamaInstance:
    """Runtime state of a provisioned Ollama instance"""
    config: OllamaInstanceConfig
    container_id: str
    host_port: int
    healthy: bool = False

class OllamaProvisioner:
    """Manages lifecycle of Ollama 0.4 instances via containerd API"""
    def __init__(self, containerd_socket: str = "/run/containerd/containerd.sock"):
        self.containerd_socket = containerd_socket
        self.instances: Dict[str, OllamaInstance] = {}
        self._session: Optional[aiohttp.ClientSession] = None

    async def __aenter__(self):
        """Initialize aiohttp session for containerd gRPC-HTTP bridge"""
        self._session = aiohttp.ClientSession(
            connector=aiohttp.UnixConnector(path=self.containerd_socket)
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Clean up session and unprovisioned instances"""
        if self._session:
            await self._session.close()
        # Unprovision all instances on exit for cleanup
        await asyncio.gather(*[self.terminate_instance(i) for i in list(self.instances.keys())])

    async def provision_instance(self, config: Optional[OllamaInstanceConfig] = None) -> OllamaInstance:
        """Provision a new Ollama 0.4 instance with retries"""
        config = config or OllamaInstanceConfig()
        config.instance_id = config.instance_id or f"ollama-{uuid.uuid4().hex[:8]}"
        config.port = config.port or self._get_free_port()

        logger.info(f"Provisioning instance {config.instance_id} with model {config.model}")

        # Retry up to 3 times for container creation failures
        for attempt in range(3):
            try:
                container_id = await self._create_container(config)
                await self._start_container(container_id)
                # Wait for health check to pass
                healthy = await self._wait_for_health(container_id, config.host_port)
                instance = OllamaInstance(
                    config=config,
                    container_id=container_id,
                    host_port=config.host_port,
                    healthy=healthy
                )
                self.instances[config.instance_id] = instance
                logger.info(f"Successfully provisioned instance {config.instance_id} (container {container_id})")
                return instance
            except Exception as e:
                logger.warning(f"Attempt {attempt+1} failed for {config.instance_id}: {str(e)}")
                if attempt == 2:
                    logger.error(f"Failed to provision {config.instance_id} after 3 attempts")
                    raise
                await asyncio.sleep(1 * (2 ** attempt))  # Exponential backoff

    async def _create_container(self, config: OllamaInstanceConfig) -> str:
        """Create containerd container for Ollama 0.4"""
        # containerd HTTP API payload for Ollama 0.4 image
        payload = {
            "id": config.instance_id,
            "image": "docker.io/ollama/ollama:0.4.0",
            "args": ["serve", "--model", config.model, "--port", "11434"],
            "env": [
                f"OLLAMA_HOST=0.0.0.0:11434",
                f"OLLAMA_NUM_PARALLEL=1",  # 1 concurrent request per instance
                f"OLLAMA_MAX_LOADED_MODELS=1"
            ],
            "resources": {
                "cpu_shares": config.cpu_shares,
                "memory_mb": config.memory_mb
            },
            "port_mappings": [
                {"container_port": 11434, "host_port": config.port, "protocol": "tcp"}
            ]
        }
        async with self._session.post("http://localhost/v1/containers", json=payload) as resp:
            if resp.status != 201:
                raise RuntimeError(f"Container creation failed: {await resp.text()}")
            data = await resp.json()
            return data["id"]

    async def _start_container(self, container_id: str) -> None:
        """Start a created container"""
        async with self._session.post(f"http://localhost/v1/containers/{container_id}/start") as resp:
            if resp.status != 200:
                raise RuntimeError(f"Container start failed: {await resp.text()}")

    async def _wait_for_health(self, container_id: str, port: int, timeout: int = 30) -> bool:
        """Check Ollama /health endpoint until ready or timeout"""
        start = asyncio.get_event_loop().time()
        while (asyncio.get_event_loop().time() - start) < timeout:
            try:
                async with self._session.get(f"http://localhost:{port}/health") as resp:
                    if resp.status == 200:
                        return True
            except Exception:
                pass
            await asyncio.sleep(1)
        return False

    def _get_free_port(self) -> int:
        """Get a random free ephemeral port"""
        import socket
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(("", 0))
            return s.getsockname()[1]

    async def terminate_instance(self, instance_id: str) -> None:
        """Terminate a running Ollama instance"""
        if instance_id not in self.instances:
            return
        instance = self.instances[instance_id]
        try:
            async with self._session.post(f"http://localhost/v1/containers/{instance.container_id}/stop") as resp:
                if resp.status != 200:
                    logger.warning(f"Stop failed for {instance_id}: {await resp.text()}")
            async with self._session.delete(f"http://localhost/v1/containers/{instance.container_id}") as resp:
                if resp.status != 200:
                    logger.warning(f"Delete failed for {instance_id}: {await resp.text()}")
            del self.instances[instance_id]
            logger.info(f"Terminated instance {instance_id}")
        except Exception as e:
            logger.error(f"Error terminating {instance_id}: {str(e)}")

# Example usage: Provision 10k instances (truncated for brevity, full loop in production)
async def main():
    async with OllamaProvisioner() as provisioner:
        # Provision 10 instances as a test (scale to 10k in production)
        tasks = [provisioner.provision_instance() for _ in range(10)]
        instances = await asyncio.gather(*tasks, return_exceptions=True)
        successful = [i for i in instances if isinstance(i, OllamaInstance)]
        logger.info(f"Provisioned {len(successful)} instances successfully")

if __name__ == "__main__":
    asyncio.run(main())

from locust import HttpUser, task, between, events
import json
import logging
import time
import random
from typing import Dict, List

# Configure logging for load test results
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("ollama-load-test")

# Global registry of provisioned Ollama instances (populated by provisioner)
INSTANCE_REGISTRY: List[Dict[str, str]] = []
INSTANCE_LOCK = None  # Set to asyncio.Lock() in async context

class OllamaLoadTestUser(HttpUser):
    """Locust user that sends requests to random Ollama 0.4 instances"""
    wait_time = between(0.1, 0.5)  # Simulate developer typing cadence
    _current_instance: Dict[str, str] = {}

    def on_start(self):
        """Assign a random Ollama instance to this user on startup"""
        global INSTANCE_REGISTRY
        if not INSTANCE_REGISTRY:
            logger.error("No Ollama instances available in registry")
            self.environment.runner.quit()
        self._current_instance = random.choice(INSTANCE_REGISTRY)
        logger.debug(f"Assigned user to instance {self._current_instance['id']}")

    @task(3)
    def generate_code(self):
        """Simulate code generation request (most common dev assistant use case)"""
        prompt = "Write a Python function to calculate the Fibonacci sequence iteratively"
        payload = {
            "model": "llama3.1:8b-q4_0",
            "prompt": prompt,
            "stream": False,
            "options": {
                "temperature": 0.2,
                "max_tokens": 256
            }
        }
        start_time = time.time()
        try:
            with self.client.post(
                f"{self._current_instance['url']}/api/generate",
                json=payload,
                catch_response=True,
                timeout=10
            ) as response:
                if response.status_code != 200:
                    response.failure(f"Status code {response.status_code}")
                    return
                data = response.json()
                if "response" not in data:
                    response.failure("Missing response field in Ollama output")
                    return
                # Validate response contains valid code
                if "def fibonacci" not in data["response"]:
                    response.failure("Response does not contain expected Fibonacci function")
                    return
                response.success()
                latency = time.time() - start_time
                logger.debug(f"Code gen request succeeded in {latency:.2f}s")
        except Exception as e:
            logger.error(f"Code gen request failed: {str(e)}")
            self.environment.runner.stats.log_error("POST", "/api/generate", str(e))

    @task(1)
    def chat_query(self):
        """Simulate general chat query (less common use case)"""
        prompt = "Explain the difference between a mutex and a semaphore in 2 sentences"
        payload = {
            "model": "llama3.1:8b-q4_0",
            "prompt": prompt,
            "stream": False,
            "options": {
                "temperature": 0.5,
                "max_tokens": 128
            }
        }
        start_time = time.time()
        try:
            with self.client.post(
                f"{self._current_instance['url']}/api/generate",
                json=payload,
                catch_response=True,
                timeout=10
            ) as response:
                if response.status_code != 200:
                    response.failure(f"Status code {response.status_code}")
                    return
                data = response.json()
                if "response" not in data:
                    response.failure("Missing response field in Ollama output")
                    return
                response.success()
                latency = time.time() - start_time
                logger.debug(f"Chat query succeeded in {latency:.2f}s")
        except Exception as e:
            logger.error(f"Chat query failed: {str(e)}")
            self.environment.runner.stats.log_error("POST", "/api/generate", str(e))

    @task(1)
    def health_check(self):
        """Periodic health check for assigned instance"""
        try:
            with self.client.get(
                f"{self._current_instance['url']}/health",
                catch_response=True,
                timeout=2
            ) as response:
                if response.status_code != 200:
                    response.failure(f"Health check failed: {response.status_code}")
                    # Reassign instance if unhealthy
                    global INSTANCE_REGISTRY
                    if INSTANCE_REGISTRY:
                        self._current_instance = random.choice(INSTANCE_REGISTRY)
                    return
                response.success()
        except Exception as e:
            logger.error(f"Health check failed: {str(e)}")
            self.environment.runner.stats.log_error("GET", "/health", str(e))

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    """Populate instance registry from provisioner API on test start"""
    global INSTANCE_REGISTRY, INSTANCE_LOCK
    INSTANCE_LOCK = asyncio.Lock()
    logger.info("Loading Ollama instance registry from provisioner API")
    try:
        import aiohttp
        async def fetch_instances():
            async with aiohttp.ClientSession() as session:
                async with session.get("http://provisioner:8080/instances") as resp:
                    if resp.status == 200:
                        instances = await resp.json()
                        return [{"id": i["id"], "url": f"http://localhost:{i['port']}"} for i in instances]
            return []
        loop = asyncio.get_event_loop()
        INSTANCE_REGISTRY = loop.run_until_complete(fetch_instances())
        logger.info(f"Loaded {len(INSTANCE_REGISTRY)} instances into registry")
    except Exception as e:
        logger.error(f"Failed to load instance registry: {str(e)}")
        environment.runner.quit()

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    """Log final test stats"""
    stats = environment.runner.stats
    logger.info(f"Load test completed. Total requests: {stats.total.num_requests}")
    logger.info(f"p50 latency: {stats.total.get_response_time_percentile(0.5):.2f}s")
    logger.info(f"p99 latency: {stats.total.get_response_time_percentile(0.99):.2f}s")
    logger.info(f"Failure rate: {stats.total.fail_ratio * 100:.2f}%")

if __name__ == "__main__":
    # Run with: locust -f load_test.py --headless -u 10000 -r 1000 --run-time 30m
    pass

from prometheus_client import start_http_server, Gauge, Counter, Histogram
import asyncio
import aiohttp
import logging
import time
from typing import Dict, List
from dataclasses import dataclass

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

# Prometheus metrics definitions
OLLAMA_INSTANCE_COUNT = Gauge(
    "ollama_instances_total",
    "Total number of provisioned Ollama instances",
    ["status"]  # status: healthy, unhealthy, provisioning
)
OLLAMA_REQUEST_LATENCY = Histogram(
    "ollama_request_latency_seconds",
    "Latency of Ollama API requests",
    ["instance_id", "endpoint"],
    buckets=[0.1, 0.25, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 5.0]
)
OLLAMA_TOKEN_THROUGHPUT = Gauge(
    "ollama_token_throughput_total",
    "Total tokens generated per second across all instances",
    ["model"]
)
OLLAMA_ERRORS = Counter(
    "ollama_errors_total",
    "Total number of Ollama instance errors",
    ["instance_id", "error_type"]
)

@dataclass
class InstanceMetrics:
    """Metrics for a single Ollama instance"""
    instance_id: str
    healthy: bool
    tokens_per_second: float
    latency_p99: float
    error_count: int

class OllamaMetricsExporter:
    """Exports Ollama 0.4 instance metrics to Prometheus"""
    def __init__(self, poll_interval: int = 10, provisioner_api: str = "http://provisioner:8080"):
        self.poll_interval = poll_interval
        self.provisioner_api = provisioner_api
        self._session: Optional[aiohttp.ClientSession] = None
        self._instance_metrics: Dict[str, InstanceMetrics] = {}

    async def start(self):
        """Start metrics collection loop and Prometheus HTTP server"""
        start_http_server(9090)  # Prometheus scrape endpoint
        self._session = aiohttp.ClientSession()
        logger.info(f"Started metrics exporter on port 9090, polling every {self.poll_interval}s")
        while True:
            try:
                await self._collect_metrics()
                self._update_prometheus_metrics()
            except Exception as e:
                logger.error(f"Metrics collection failed: {str(e)}")
            await asyncio.sleep(self.poll_interval)

    async def _collect_metrics(self):
        """Fetch instance list and collect metrics from each Ollama instance"""
        # Fetch instance list from provisioner
        try:
            async with self._session.get(f"{self.provisioner_api}/instances") as resp:
                if resp.status != 200:
                    raise RuntimeError(f"Provisioner API error: {await resp.text()}")
                instances = await resp.json()
        except Exception as e:
            logger.error(f"Failed to fetch instance list: {str(e)}")
            return

        # Collect metrics from each instance concurrently
        tasks = [self._collect_instance_metrics(i) for i in instances]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Update instance metrics registry
        for result in results:
            if isinstance(result, InstanceMetrics):
                self._instance_metrics[result.instance_id] = result
            elif isinstance(result, Exception):
                logger.error(f"Instance metrics collection failed: {str(result)}")

    async def _collect_instance_metrics(self, instance: Dict) -> InstanceMetrics:
        """Collect metrics from a single Ollama instance"""
        instance_id = instance["id"]
        port = instance["port"]
        healthy = False
        tokens_per_second = 0.0
        latency_p99 = 0.0
        error_count = 0

        # Check health
        try:
            async with self._session.get(f"http://localhost:{port}/health", timeout=2) as resp:
                healthy = resp.status == 200
        except Exception:
            healthy = False

        # Collect token throughput via /api/metrics endpoint (Ollama 0.4+)
        if healthy:
            try:
                async with self._session.get(f"http://localhost:{port}/api/metrics", timeout=5) as resp:
                    if resp.status == 200:
                        metrics = await resp.json()
                        tokens_per_second = metrics.get("tokens_per_second", 0.0)
                        latency_p99 = metrics.get("latency_p99", 0.0)
                        error_count = metrics.get("error_count", 0)
            except Exception as e:
                logger.warning(f"Failed to collect metrics from {instance_id}: {str(e)}")
                error_count += 1

        return InstanceMetrics(
            instance_id=instance_id,
            healthy=healthy,
            tokens_per_second=tokens_per_second,
            latency_p99=latency_p99,
            error_count=error_count
        )

    def _update_prometheus_metrics(self):
        """Update Prometheus metrics with collected data"""
        # Reset instance count gauges
        OLLAMA_INSTANCE_COUNT.labels(status="healthy").set(0)
        OLLAMA_INSTANCE_COUNT.labels(status="unhealthy").set(0)
        OLLAMA_INSTANCE_COUNT.labels(status="provisioning").set(0)

        total_throughput = 0.0
        model_throughput: Dict[str, float] = {}

        for instance_id, metrics in self._instance_metrics.items():
            status = "healthy" if metrics.healthy else "unhealthy"
            OLLAMA_INSTANCE_COUNT.labels(status=status).inc()

            if metrics.healthy:
                total_throughput += metrics.tokens_per_second
                # Assume all instances run llama3.1:8b-q4_0 for this example
                model = "llama3.1:8b-q4_0"
                model_throughput[model] = model_throughput.get(model, 0.0) + metrics.tokens_per_second

            # Update error counter (only increment if new errors)
            OLLAMA_ERRORS.labels(instance_id=instance_id, error_type="request").inc(metrics.error_count)

        # Update token throughput gauge
        for model, throughput in model_throughput.items():
            OLLAMA_TOKEN_THROUGHPUT.labels(model=model).set(throughput)

        logger.debug(f"Updated metrics: {len(self._instance_metrics)} instances, {total_throughput:.2f} tokens/sec total")

    async def stop(self):
        """Clean up resources"""
        if self._session:
            await self._session.close()
        logger.info("Metrics exporter stopped")

async def main():
    exporter = OllamaMetricsExporter(poll_interval=10)
    try:
        await exporter.start()
    except KeyboardInterrupt:
        await exporter.stop()

if __name__ == "__main__":
    asyncio.run(main())

[plugins."io.containerd.grpc.v1.cri".containerd.runtimes.crun]
  runtime_type = "io.containerd.runc.v2"
  [plugins."io.containerd.grpc.v1.cri".containerd.runtimes.crun.options]
    BinaryName = "/usr/bin/crun"
    Root = "/run/containerd/runc"

ollama pull llama3.1:8b-q4_0

limit_req_zone $instance_id zone=ollama:10m rate=1r/s;
server {
    location /api/generate {
        limit_req zone=ollama burst=2 nodelay;
        proxy_pass http://ollama-backend;
    }
}