Calliope Calliope Hub Performance Optimization Guide

# Calliope Hub/config/jupyterhub_config.py
c.Spawner.poll_interval = 20  # from random.randint(5, 10)
c.Spawner.poll_jitter = 0.2   # Add 20% jitter to prevent thundering herd

# Update Calliope Hub task definition
TaskDefinition:
  cpu: "4096"      # from "2048"
  memory: "8192"   # from "4096"

# Calliope Hub/spawners/helpers/batch_poller.py (NEW FILE)

import asyncio
import boto3
from typing import Dict, List
from datetime import datetime, timezone

class BatchTaskPoller:
    """Batch multiple task status checks into single ECS API calls."""

    def __init__(self, ecs_cluster: str, aws_region: str, logger):
        self.ecs_cluster = ecs_cluster
        self.aws_region = aws_region
        self.log = logger
        self.ecs_client = boto3.client("ecs", region_name=aws_region)

        # Cache for task status (TTL: 10 seconds)
        self._cache = {}
        self._cache_ttl = 10

    async def get_task_status(self, task_arn: str) -> Dict:
        """
        Get task status, using cache if available.

        Args:
            task_arn: ECS task ARN

        Returns:
            Task description dict
        """
        # Check cache first
        if task_arn in self._cache:
            cached_data, cached_time = self._cache[task_arn]
            age = (datetime.now(timezone.utc) - cached_time).total_seconds()

            if age < self._cache_ttl:
                self.log.debug(f"Cache hit for {task_arn[-12:]} (age: {age:.1f}s)")
                return cached_data

        # Cache miss or expired - fetch from ECS
        return await self._fetch_task_status(task_arn)

    async def batch_get_task_status(self, task_arns: List[str]) -> Dict[str, Dict]:
        """
        Batch fetch multiple task statuses in single API call.

        Args:
            task_arns: List of task ARNs to check

        Returns:
            Dict mapping task_arn to task description
        """
        if not task_arns:
            return {}

        # Filter out cached tasks
        uncached_arns = []
        results = {}

        for arn in task_arns:
            if arn in self._cache:
                cached_data, cached_time = self._cache[arn]
                age = (datetime.now(timezone.utc) - cached_time).total_seconds()

                if age < self._cache_ttl:
                    results[arn] = cached_data
                    continue

            uncached_arns.append(arn)

        # Fetch uncached tasks in batches of 100 (ECS API limit)
        if uncached_arns:
            self.log.info(f"Batch fetching {len(uncached_arns)} tasks from ECS")

            for i in range(0, len(uncached_arns), 100):
                batch = uncached_arns[i:i+100]

                loop = asyncio.get_running_loop()
                response = await loop.run_in_executor(
                    None,
                    lambda: self.ecs_client.describe_tasks(
                        cluster=self.ecs_cluster,
                        tasks=batch,
                        include=["TAGS"]
                    )
                )

                # Cache and return results
                now = datetime.now(timezone.utc)
                for task in response.get("tasks", []):
                    task_arn = task["taskArn"]
                    self._cache[task_arn] = (task, now)
                    results[task_arn] = task

        return results

    async def _fetch_task_status(self, task_arn: str) -> Dict:
        """Fetch single task status and cache it."""
        loop = asyncio.get_running_loop()
        response = await loop.run_in_executor(
            None,
            lambda: self.ecs_client.describe_tasks(
                cluster=self.ecs_cluster,
                tasks=[task_arn]
            )
        )

        tasks = response.get("tasks", [])
        if tasks:
            task = tasks[0]
            self._cache[task_arn] = (task, datetime.now(timezone.utc))
            return task

        return None

    def clear_cache(self):
        """Clear the cache (e.g., after spawning new tasks)."""
        self._cache.clear()

# Calliope Hub/spawners/ecs.py

# At module level, create shared batch poller
_batch_poller = None

def get_batch_poller(spawner):
    global _batch_poller
    if _batch_poller is None:
        _batch_poller = BatchTaskPoller(
            spawner.ecs_cluster,
            spawner.aws_region,
            spawner.log
        )
    return _batch_poller

# In poll() method, replace direct describe_tasks:
async def poll(self) -> Optional[int]:
    if not self.task_arn:
        return 0

    # Use batch poller instead of direct ECS call
    batch_poller = get_batch_poller(self)
    task = await batch_poller.get_task_status(self.task_arn)

    if not task:
        return 0  # Task not found

    # Rest of poll logic...

# Calliope Hub/config/jupyterhub_config.py

import asyncio
import random

# Global poll coordinator
class PollCoordinator:
    def __init__(self, poll_interval=15):
        self.poll_interval = poll_interval
        self.last_poll = {}

    async def should_poll(self, spawner_id: str) -> bool:
        """
        Check if spawner should poll now.
        Coordinates polling to prevent simultaneous API calls.
        """
        import time
        now = time.time()

        last_poll = self.last_poll.get(spawner_id, 0)
        if now - last_poll < self.poll_interval:
            return False  # Too soon

        # Add jitter to prevent thundering herd
        jitter = random.uniform(-2, 2)
        if now - last_poll < self.poll_interval + jitter:
            return False

        self.last_poll[spawner_id] = now
        return True

poll_coordinator = PollCoordinator(poll_interval=15)

# In spawner's poll() method:
async def poll(self):
    spawner_id = f"{self.user.name}/{self.name}"

    if not await poll_coordinator.should_poll(spawner_id):
        # Skip this poll cycle
        return None  # Assume still running

    # Proceed with actual poll
    ...

# Calliope Hub/spawners/helpers/ecs_cache.py (NEW FILE)

from datetime import datetime, timezone, timedelta
from typing import Dict, Optional
import asyncio

class ECSResponseCache:
    """Cache ECS API responses to reduce redundant calls."""

    def __init__(self, ttl_seconds: int = 15):
        self.ttl = timedelta(seconds=ttl_seconds)
        self._cache: Dict[str, tuple] = {}
        self._lock = asyncio.Lock()

    async def get(self, key: str) -> Optional[Dict]:
        """Get cached response if not expired."""
        async with self._lock:
            if key in self._cache:
                data, timestamp = self._cache[key]
                if datetime.now(timezone.utc) - timestamp < self.ttl:
                    return data
                else:
                    # Expired, remove
                    del self._cache[key]
        return None

    async def set(self, key: str, data: Dict):
        """Cache response with current timestamp."""
        async with self._lock:
            self._cache[key] = (data, datetime.now(timezone.utc))

    async def clear_expired(self):
        """Background task to clear expired entries."""
        while True:
            await asyncio.sleep(60)  # Every minute
            async with self._lock:
                now = datetime.now(timezone.utc)
                expired = [
                    k for k, (_, ts) in self._cache.items()
                    if now - ts >= self.ttl
                ]
                for k in expired:
                    del self._cache[k]

# Global cache instance
ecs_cache = ECSResponseCache(ttl_seconds=15)

# Start cleanup task
asyncio.create_task(ecs_cache.clear_expired())

# In ECS spawner poll() method
cache_key = f"task:{self.task_arn}"

# Try cache first
cached = await ecs_cache.get(cache_key)
if cached:
    task = cached
else:
    # Fetch from ECS
    response = ecs_client.describe_tasks(...)
    task = response["tasks"][0]
    await ecs_cache.set(cache_key, task)

EventRule:
  Name: ecs-task-state-changes
  EventPattern:
    source:
      - aws.ecs
    detail-type:
      - ECS Task State Change
    detail:
      clusterArn:
        - arn:aws:ecs:us-west-2:xxxx:cluster/development
      lastStatus:
        - RUNNING
        - STOPPED
        - DEPROVISIONING

  Targets:
    - Arn: !GetAtt TaskStatusLambda.Arn
      Id: task-status-processor

# lambda/ecs_task_status.py

import json
import requests
import os

JUPYTERHUB_API_TOKEN = os.environ["JUPYTERHUB_API_TOKEN"]
JUPYTERHUB_API_URL = os.environ["JUPYTERHUB_API_URL"]

def lambda_handler(event, context):
    """Process ECS task state change events and notify Calliope Calliope Hub."""

    detail = event.get("detail", {})
    task_arn = detail.get("taskArn")
    last_status = detail.get("lastStatus")
    desired_status = detail.get("desiredStatus")

    # Extract task metadata
    containers = detail.get("containers", [])

    # Notify Calliope Calliope Hub of state change
    response = requests.post(
        f"{JUPYTERHUB_API_URL}/Calliope Hub/api/tasks/status",
        headers={"Authorization": f"token {JUPYTERHUB_API_TOKEN}"},
        json={
            "task_arn": task_arn,
            "status": last_status,
            "desired_status": desired_status,
            "containers": containers,
        },
        timeout=5,
    )

    return {
        "statusCode": 200,
        "body": json.dumps(f"Processed {task_arn}: {last_status}"),
    }

# Calliope Hub/handlers/task_status_handler.py (NEW FILE)

from Calliope Calliope Hub.handlers import BaseHandler
from Calliope Calliope Hub.utils import token_authenticated

class TaskStatusHandler(BaseHandler):
    """Receive ECS task status updates from EventBridge."""

    @token_authenticated
    async def post(self):
        """Process task status update."""
        data = self.get_json_body()

        task_arn = data.get("task_arn")
        status = data.get("status")

        self.log.info(f"Received task status: {task_arn[-12:]} = {status}")

        # Find spawner managing this task
        for user in self.users.values():
            for spawner in user.spawners.values():
                if getattr(spawner, "task_arn", None) == task_arn:
                    # Update spawner's cached state
                    spawner._cached_task_status = status
                    spawner._cache_time = datetime.now()

                    if status == "STOPPED":
                        # Trigger cleanup
                        self.log.info(f"Task stopped, scheduling cleanup")
                        asyncio.create_task(spawner.stop())

                    self.set_status(200)
                    return

        self.log.warning(f"No spawner found for task {task_arn[-12:]}")
        self.set_status(404)

# Register handler in jupyterhub_config.py:
c.Calliope Calliope Hub.extra_handlers = [
    (r"/Calliope Hub/api/tasks/status", TaskStatusHandler),
]

# services/task_monitor.py

import asyncio
import boto3
from sqlalchemy import create_engine

class TaskMonitor:
    """Dedicated service for monitoring ECS task health."""

    def __init__(self, db_url, ecs_cluster, aws_region):
        self.db = create_engine(db_url)
        self.ecs = boto3.client("ecs", region_name=aws_region)
        self.cluster = ecs_cluster

    async def monitor_loop(self):
        """Main monitoring loop."""
        while True:
            try:
                # Get all spawner tasks from database
                with self.db.connect() as conn:
                    result = conn.execute("""
                        SELECT id, user_id, name, state
                        FROM spawners
                        WHERE server_id IS NOT NULL
                    """)
                    spawners = result.fetchall()

                # Extract task ARNs from state
                task_arns = []
                for spawner in spawners:
                    state = json.loads(spawner.state or "{}")
                    task_arn = state.get("task_arn")
                    if task_arn:
                        task_arns.append((spawner.id, task_arn))

                # Batch check all tasks (100 at a time)
                for i in range(0, len(task_arns), 100):
                    batch = [arn for _, arn in task_arns[i:i+100]]

                    response = self.ecs.describe_tasks(
                        cluster=self.cluster,
                        tasks=batch
                    )

                    # Update database with status
                    for task in response.get("tasks", []):
                        task_arn = task["taskArn"]
                        status = task["lastStatus"]
                        health = task.get("healthStatus", "UNKNOWN")

                        # Find spawner ID
                        spawner_id = next(
                            (sid for sid, arn in task_arns if arn == task_arn),
                            None
                        )

                        if spawner_id:
                            # Update spawner state in DB
                            self._update_spawner_status(spawner_id, status, health)

                # Sleep before next poll
                await asyncio.sleep(15)

            except Exception as e:
                logger.error(f"Monitor loop error: {e}")
                await asyncio.sleep(60)

    def _update_spawner_status(self, spawner_id, status, health):
        """Update spawner status in database."""
        with self.db.connect() as conn:
            conn.execute("""
                UPDATE spawners
                SET state = jsonb_set(
                    COALESCE(state::jsonb, '{}'::jsonb),
                    '{task_status}',
                    to_jsonb($1::text)
                ),
                state = jsonb_set(
                    state::jsonb,
                    '{health_status}',
                    to_jsonb($2::text)
                ),
                state = jsonb_set(
                    state::jsonb,
                    '{last_poll}',
                    to_jsonb($3::text)
                )
                WHERE id = $4
            """, status, health, datetime.now().isoformat(), spawner_id)

Service:
  ServiceName: Calliope Calliope Hub-monitor
  Cluster: development
  TaskDefinition: task-monitor:latest
  DesiredCount: 1  # Single instance

  # No load balancer - internal service

-- Index on user_id for fast user lookups
CREATE INDEX idx_spawners_user_id ON spawners(user_id);

-- Index on server_id for fast server lookups
CREATE INDEX idx_spawners_server_id ON spawners(server_id);

-- Composite index for task ARN lookups in state
CREATE INDEX idx_spawners_state_task_arn ON spawners USING gin (state jsonb_path_ops);

-- Index on last_activity for idle culling
CREATE INDEX idx_spawners_last_activity ON spawners(last_activity);

-- Index on user name for fast username lookups
CREATE INDEX idx_users_name ON users(name);

-- Enable query logging
ALTER DATABASE Calliope Calliope Hub SET log_min_duration_statement = 100;

-- Check slow queries
SELECT query, calls, mean_exec_time, max_exec_time
FROM pg_stat_statements
ORDER BY mean_exec_time DESC
LIMIT 20;

# pgbouncer.ini
[databases]
Calliope Calliope Hub = host=rds-endpoint port=5432 dbname=Calliope Calliope Hub

[pgbouncer]
listen_addr = 0.0.0.0
listen_port = 6432
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt
pool_mode = transaction
max_client_conn = 500
default_pool_size = 20
reserve_pool_size = 10
server_lifetime = 3600
server_idle_timeout = 600

# Connect through pgbouncer instead of direct RDS
c.Calliope Calliope Hub.db_url = "postgresql://user:pass@pgbouncer:6432/Calliope Calliope Hub"

# Reduce pool size (pgbouncer handles pooling)
c.Calliope Calliope Hub.db_pool_size = 5  # from 10
c.Calliope Calliope Hub.db_max_overflow = 5  # from 20

# Calliope Hub/spawners/ecs.py

class ECSSpawner(BaseSpawner):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._state_loaded = False

    async def _ensure_state_loaded(self):
        """Lazy load spawner state from database."""
        if self._state_loaded:
            return

        # Load state from DB
        state = await self._load_state_from_db()
        if state:
            self.load_state(state)

        self._state_loaded = True

    async def poll(self):
        await self._ensure_state_loaded()
        # ... rest of poll logic

# Before (serial):
async def start(self):
    self._create_task_definition()  # 500ms
    self._run_task()                # 1000ms
    self._wait_for_running()        # 2000ms
    self._register_with_proxy()     # 500ms
    # Total: 4000ms

# After (parallel):
async def start(self):
    # Run independent operations in parallel
    task_def, proxy_ready = await asyncio.gather(
        self._create_task_definition(),
        self._prepare_proxy_registration(),
    )

    task_arn = await self._run_task(task_def)

    # Parallel wait
    await asyncio.gather(
        self._wait_for_running(task_arn),
        self._register_with_proxy(task_arn, proxy_ready),
    )
    # Total: 2500ms (40% faster)

# jupyterhub_config.py

import logging
import os

# Environment-based log levels
if os.getenv("ENVIRONMENT") == "production":
    c.Calliope Calliope Hub.log_level = "WARNING"  # Only warnings and errors
    c.Spawner.debug = False
else:
    c.Calliope Calliope Hub.log_level = "INFO"
    c.Spawner.debug = True

# Reduce SQLAlchemy logging
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
logging.getLogger("tornado.access").setLevel(logging.WARNING)

# Keep important loggers at INFO
logging.getLogger("Calliope Calliope Hub.proxy").setLevel(logging.INFO)
logging.getLogger("Calliope Calliope Hub.spawner").setLevel(logging.INFO)

# Calliope Hub/spawners/helpers/metrics.py (NEW FILE)

import json
import time
from datetime import datetime

class MetricsCollector:
    """Collect spawner metrics using EMF for CloudWatch."""

    @staticmethod
    def emit_spawn_duration(service_type: str, duration: float):
        """Emit spawn duration metric."""
        print(json.dumps({
            "_aws": {
                "Timestamp": int(time.time() * 1000),
                "CloudWatchMetrics": [{
                    "Namespace": "Calliope Calliope Hub/Spawner",
                    "Dimensions": [["ServiceType"]],
                    "Metrics": [{"Name": "SpawnDuration", "Unit": "Seconds"}]
                }]
            },
            "ServiceType": service_type,
            "SpawnDuration": duration,
        }))

    @staticmethod
    def emit_poll_duration(duration: float, task_count: int):
        """Emit poll cycle metrics."""
        print(json.dumps({
            "_aws": {
                "Timestamp": int(time.time() * 1000),
                "CloudWatchMetrics": [{
                    "Namespace": "Calliope Calliope Hub/Spawner",
                    "Dimensions": [["Operation"]],
                    "Metrics": [
                        {"Name": "PollDuration", "Unit": "Seconds"},
                        {"Name": "TaskCount", "Unit": "Count"}
                    ]
                }]
            },
            "Operation": "Poll",
            "PollDuration": duration,
            "TaskCount": task_count,
        }))

# In spawner
from helpers.metrics import MetricsCollector

async def start(self):
    start_time = time.time()
    # ... spawn logic ...
    duration = time.time() - start_time

    MetricsCollector.emit_spawn_duration(self.instance_type, duration)

# CloudWatch dashboard to track
# - Average spawn duration by service
# - Poll duration trends
# - API call counts
# - Active spawner count

VPCEndpoints:
  - ServiceName: com.amazonaws.us-west-2.ecs
    VpcEndpointType: Interface
    SubnetIds:
      - subnet-private-1
      - subnet-private-2
    SecurityGroupIds:
      - sg-vpc-endpoints

  - ServiceName: com.amazonaws.us-west-2.ecs-agent
    VpcEndpointType: Interface

  - ServiceName: com.amazonaws.us-west-2.ecr.api
    VpcEndpointType: Interface

  - ServiceName: com.amazonaws.us-west-2.ecr.dkr
    VpcEndpointType: Interface

  - ServiceName: com.amazonaws.us-west-2.s3
    VpcEndpointType: Gateway  # Free!

  - ServiceName: com.amazonaws.us-west-2.logs
    VpcEndpointType: Interface

EFS:
  ThroughputMode: provisioned
  ProvisionedThroughputInMibps: 10  # from bursting

# In spawner pre_spawn_hook
async def pre_spawn_hook(spawner):
    # Pre-create user directory structure on EFS
    # Reduces container startup time
    await provision_user_directory(spawner.user.name)

# In container image, pre-populate common configs
COPY jupyter_config_template.py /opt/jupyter/config/
COPY jupyter_ai_schema.json /opt/jupyter/ai/

# tests/load_test.py

import asyncio
import aiohttp
import time
from typing import List

async def spawn_server(session: aiohttp.ClientSession, username: str, server_name: str):
    """Spawn a single server."""
    start = time.time()

    async with session.post(
        f"https://your-Calliope Hub/Calliope Hub/api/users/{username}/servers/{server_name}",
        headers={"Authorization": f"token {API_TOKEN}"},
        json={"instance_type": "lab", "instance_size": "medium"}
    ) as resp:
        duration = time.time() - start
        return {
            "username": username,
            "server_name": server_name,
            "status": resp.status,
            "duration": duration,
        }

async def load_test(num_servers: int):
    """Spawn N servers concurrently and measure performance."""
    async with aiohttp.ClientSession() as session:
        tasks = []

        # Create spawn tasks
        for i in range(num_servers):
            username = f"loadtest-user-{i:04d}"
            server_name = f"lab-{int(time.time())}"
            tasks.append(spawn_server(session, username, server_name))

        # Execute concurrently
        print(f"Spawning {num_servers} servers...")
        start = time.time()

        results = await asyncio.gather(*tasks, return_exceptions=True)

        total_time = time.time() - start

        # Analyze results
        successful = sum(1 for r in results if isinstance(r, dict) and r["status"] == 202)
        failed = len(results) - successful
        avg_duration = sum(r["duration"] for r in results if isinstance(r, dict)) / len(results)

        print(f"""
Load Test Results:
- Total servers: {num_servers}
- Successful: {successful}
- Failed: {failed}
- Total time: {total_time:.1f}s
- Avg spawn time: {avg_duration:.1f}s
- Spawns/second: {successful / total_time:.2f}
        """)

if __name__ == "__main__":
    asyncio.run(load_test(50))

python load_test.py 50   # Baseline
python load_test.py 100  # 2x
python load_test.py 200  # 4x
python load_test.py 500  # 10x

# In spawner
import time
import logging

logger = logging.getLogger(__name__)

async def poll(self):
    start = time.time()

    # ... poll logic ...

    duration = time.time() - start
    if duration > 5:
        logger.warning(f"Slow poll: {duration:.2f}s for {self.user.name}/{self.name}")

    # Emit metric
    emit_metric("PollDuration", duration)

# For 300-500 servers on 8 vCPU / 16GB Calliope Hub

import random
import os

# Polling configuration
c.Spawner.poll_interval = 20  # Slower polling for stability
c.Spawner.poll_jitter = 0.2   # 20% jitter

# Activity tracking
c.Calliope Calliope Hub.last_activity_interval = 300  # 5 minutes

# Database configuration (PostgreSQL)
c.Calliope Calliope Hub.db_url = os.getenv("JUPYTERHUB_DB_URL")
c.Calliope Calliope Hub.db_pool_size = 10
c.Calliope Calliope Hub.db_max_overflow = 20

# Logging (production)
if os.getenv("ENVIRONMENT") == "production":
    c.Calliope Calliope Hub.log_level = "WARNING"
else:
    c.Calliope Calliope Hub.log_level = "INFO"

# Proxy configuration
c.ConfigurableHTTPProxy.should_start = True
c.ConfigurableHTTPProxy.api_url = "http://127.0.0.1:8001"

# Spawner defaults
c.Spawner.start_timeout = 600  # 10 minutes
c.Spawner.http_timeout = 120   # 2 minutes

TaskDefinition:
  Family: Calliope Hub-calliope-dev-Calliope Calliope Hub-task
  NetworkMode: awsvpc
  RequiresCompatibilities:
    - FARGATE
  Cpu: "4096"      # 4 vCPU for 300-400 servers
  Memory: "8192"   # 8 GB

  ContainerDefinitions:
    - Name: Calliope Calliope Hub
      Image: <your-Calliope Hub-image>
      Essential: true

      PortMappings:
        - ContainerPort: 8000
          Protocol: tcp

      Environment:
        - Name: JUPYTERHUB_SERVICE_MODE
          Value: ecs
        - Name: ENVIRONMENT
          Value: production
        - Name: AWS_DEFAULT_REGION
          Value: us-west-2

      Secrets:
        - Name: JUPYTERHUB_DB_URL
          ValueFrom: arn:aws:secretsmanager:...:JUPYTERHUB_DB_URL::
        - Name: JUPYTERHUB_CRYPT_KEY
          ValueFrom: arn:aws:secretsmanager:...:JUPYTERHUB_CRYPT_KEY::

      LogConfiguration:
        LogDriver: awslogs
        Options:
          awslogs-group: /aws/ecs/Calliope Hub-calliope-dev
          awslogs-region: us-west-2
          awslogs-stream-prefix: Calliope Hub

      HealthCheck:
        Command:
          - CMD-SHELL
          - curl -f http://localhost:8000/Calliope Hub/health || exit 1
        Interval: 30
        Timeout: 5
        Retries: 3
        StartPeriod: 60