Scaling Strategies

Patterns for scaling agent systems: horizontal, vertical, queue-based, and adaptive.

Scaling Taxonomy

Scaling Strategies
├── Horizontal (Scale Out)
│   ├── Stateless Replicas
│   ├── Sharding
│   ├── Partitioning
│   └── Load Balancing
├── Vertical (Scale Up)
│   ├── Resource Limits
│   ├── Instance Sizing
│   └── Hardware Acceleration
├── Queue-Based
│   ├── Work Queues
│   ├── Priority Queues
│   ├── Dead Letter Queues
│   └── Backpressure
├── Adaptive
│   ├── Predictive Scaling
│   ├── Schedule-Based
│   ├── Metric-Based
│   └── ML-Based
└── Database
    ├── Read Replicas
    ├── Sharding
    ├── Connection Pooling
    └── Caching Layers

Horizontal Scaling

Stateless Agent Replicas

# Kubernetes HPA for agent replicas
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: agent-service-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: agent-service
  minReplicas: 3
  maxReplicas: 100
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Pods
    pods:
      metric:
        name: http_requests_per_second
      target:
        type: AverageValue
        averageValue: "50"
  - type: External
    external:
      metric:
        name: queue_depth
        selector:
          matchLabels:
            queue: agent-tasks
      target:
        type: Value
        value: "100"
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 10
        periodSeconds: 60
    scaleUp:
      stabilizationWindowSeconds: 0
      policies:
      - type: Percent
        value: 100
        periodSeconds: 15
      - type: Pods
        value: 4
        periodSeconds: 15
      selectPolicy: Max

Sharding Strategies

class ShardingStrategy:
    def __init__(self, num_shards: int):
        self.num_shards = num_shards

    def get_shard(self, key: str) -> int:
        """Consistent hashing for session affinity"""
        return int(hashlib.md5(key.encode()).hexdigest(), 16) % self.num_shards

    def get_shard_for_user(self, user_id: str) -> int:
        """User-based sharding for data locality"""
        return self.get_shard(f"user:{user_id}")

    def get_shard_for_session(self, session_id: str) -> int:
        """Session-based sharding"""
        return self.get_shard(f"session:{session_id}")

class ShardedAgentPool:
    def __init__(self, num_shards: int, agent_factory: Callable):
        self.shards = [agent_factory() for _ in range(num_shards)]
        self.strategy = ShardingStrategy(len(self.shards))

    async def execute(self, task: Task, shard_key: str = None) -> Result:
        if shard_key is None:
            shard_key = task.session_id or task.user_id

        shard_id = self.strategy.get_shard(shard_key)
        agent = self.shards[shard_id]
        return await agent.execute(task)

Load Balancing

class LoadBalancer:
    def __init__(self, agents: list[Agent]):
        self.agents = agents
        self.strategy = "least_connections"

    async def route(self, request: Request) -> Agent:
        if self.strategy == "round_robin":
            return self.round_robin()
        elif self.strategy == "least_connections":
            return min(self.agents, key=lambda a: a.active_connections)
        elif self.strategy == "least_latency":
            return min(self.agents, key=lambda a: a.avg_latency_ms)
        elif self.strategy == "weighted":
            return self.weighted_random()
        elif self.strategy == "consistent_hash":
            return self.consistent_hash(request.session_id)

    def round_robin(self) -> Agent:
        agent = self.agents[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.agents)
        return agent

    def weighted_random(self) -> Agent:
        total_weight = sum(a.weight for a in self.agents)
        r = random.uniform(0, total_weight)
        for agent in self.agents:
            r -= agent.weight
            if r <= 0:
                return agent
        return self.agents[-1]

Vertical Scaling

Resource Limits

# Kubernetes resource management
resources:
  requests:
    memory: "512Mi"
    cpu: "250m"
    nvidia.com/gpu: "1"  # For GPU workloads
  limits:
    memory: "2Gi"
    cpu: "2000m"
    nvidia.com/gpu: "1"

# Vertical Pod Autoscaler
apiVersion: autoscaling.k8s.io/v1
kind: VerticalPodAutoscaler
metadata:
  name: agent-service-vpa
spec:
  targetRef:
    apiVersion: "apps/v1"
    kind: Deployment
    name: agent-service
  updateMode: "Auto"
  resourcePolicy:
    containerPolicies:
    - containerName: agent
      minAllowed:
        memory: "256Mi"
        cpu: "100m"
      maxAllowed:
        memory: "4Gi"
        cpu: "4000m"

Instance Sizing Guidelines

Queue-Based Scaling

Work Queue Architecture

class WorkQueue:
    def __init__(self, redis: Redis, queue_name: str):
        self.redis = redis
        self.queue_name = queue_name
        self.dlq_name = f"{queue_name}:dlq"

    async def enqueue(self, task: Task, priority: int = 0) -> str:
        task_id = str(uuid4())
        task_data = json.dumps({
            "id": task_id,
            "task": task.to_dict(),
            "created_at": datetime.utcnow().isoformat(),
            "retries": 0
        })

        # Priority queue: higher score = higher priority
        score = -priority  # Negative for max-heap behavior
        await self.redis.zadd(self.queue_name, {task_data: score})
        return task_id

    async def dequeue(self, worker_id: str, count: int = 10) -> list[Task]:
        # Atomic pop with Lua script
        lua_script = """
        local tasks = redis.call('ZRANGE', KEYS[1], 0, ARGV[1]-1)
        if #tasks > 0 then
            redis.call('ZREM', KEYS[1], unpack(tasks))
        end
        return tasks
        """
        tasks_json = await self.redis.eval(
            lua_script, 1, self.queue_name, count
        )
        return [json.loads(t) for t in tasks_json]

    async def requeue(self, task: Task, error: str):
        task.retries += 1
        if task.retries >= 3:
            await self.move_to_dlq(task, error)
        else:
            await self.enqueue(task, priority=-1)  # Lower priority for retry

class WorkerPool:
    def __init__(self, queue: WorkQueue, agent_factory: Callable, 
                 max_workers: int = 10):
        self.queue = queue
        self.agent_factory = agent_factory
        self.max_workers = max_workers
        self.workers: list[asyncio.Task] = []

    async def start(self):
        for i in range(self.max_workers):
            worker = asyncio.create_task(self.worker_loop(f"worker-{i}"))
            self.workers.append(worker)

    async def worker_loop(self, worker_id: str):
        agent = self.agent_factory()
        while True:
            tasks = await self.queue.dequeue(worker_id, count=5)
            for task_data in tasks:
                try:
                    task = Task.from_dict(task_data["task"])
                    result = await agent.execute(task)
                    await self.handle_result(task_data["id"], result)
                except Exception as e:
                    await self.queue.requeue(Task.from_dict(task_data["task"]), str(e))

Priority Queues

class PriorityQueue:
    def __init__(self, redis: Redis, base_name: str):
        self.redis = redis
        self.queues = {
            "critical": f"{base_name}:critical",
            "high": f"{base_name}:high",
            "normal": f"{base_name}:normal",
            "low": f"{base_name}:low"
        }
        self.weights = {"critical": 100, "high": 10, "normal": 1, "low": 0.1}

    async def enqueue(self, task: Task, priority: str = "normal"):
        await self.redis.zadd(self.queues[priority], 
                             {json.dumps(task.to_dict()): time.time()})

    async def dequeue(self, count: int = 10) -> list[Task]:
        # Weighted round-robin across priorities
        for priority in ["critical", "high", "normal", "low"]:
            tasks = await self.redis.zpopmin(self.queues[priority], count)
            if tasks:
                return [json.loads(t[0]) for t in tasks]
        return []

Backpressure & Rate Limiting

class BackpressureController:
    def __init__(self, max_queue_size: int = 10000, 
                 max_latency_ms: int = 5000):
        self.max_queue_size = max_queue_size
        self.max_latency_ms = max_latency_ms
        self.current_load = 0

    async def check_backpressure(self) -> BackpressureSignal:
        queue_size = await self.get_queue_size()
        p99_latency = await self.get_p99_latency()

        if queue_size > self.max_queue_size * 0.9:
            return BackpressureSignal.REJECT
        elif queue_size > self.max_queue_size * 0.7:
            return BackpressureSignal.SLOW_DOWN
        elif p99_latency > self.max_latency_ms:
            return BackpressureSignal.SLOW_DOWN
        return BackpressureSignal.NORMAL

    async def apply_backpressure(self, signal: BackpressureSignal):
        if signal == BackpressureSignal.REJECT:
            # Return 503 with Retry-After
            raise BackpressureException(retry_after=30)
        elif signal == BackpressureSignal.SLOW_DOWN:
            # Add delay
            await asyncio.sleep(0.1)

Adaptive Scaling

Predictive Scaling

class PredictiveScaler:
    def __init__(self, history_window: int = 3600):  # 1 hour
        self.history_window = history_window
        self.model = None

    async def train(self, metrics_history: list[MetricsPoint]):
        """Train forecasting model on historical data"""
        # Features: time_of_day, day_of_week, recent_load, trends
        # Target: required_replicas
        X, y = self.prepare_features(metrics_history)
        self.model = GradientBoostingRegressor()
        self.model.fit(X, y)

    async def predict_replicas(self, current_metrics: MetricsPoint) -> int:
        if not self.model:
            return self.default_replicas

        features = self.extract_features(current_metrics)
        predicted = self.model.predict([features])[0]

        # Clamp to bounds
        return max(self.min_replicas, min(self.max_replicas, predicted))

    def extract_features(self, m: MetricsPoint) -> list[float]:
        now = datetime.utcnow()
        return [
            m.current_rps,
            m.avg_latency_ms,
            m.queue_depth,
            m.error_rate,
            m.cpu_utilization,
            m.memory_utilization,
            now.hour,
            now.weekday(),
            # Rolling stats
            self.rolling_avg(m.rps_history, 5),
            self.rolling_avg(m.latency_history, 5),
            self.trend(m.rps_history, 10)
        ]

Schedule-Based Scaling

class ScheduleScaler:
    def __init__(self):
        self.schedules: list[ScheduleRule] = []

    def add_schedule(self, rule: ScheduleRule):
        self.schedules.append(rule)

    def get_desired_replicas(self, now: datetime = None) -> int:
        now = now or datetime.utcnow()

        for rule in self.schedules:
            if rule.matches(now):
                return rule.replicas

        return self.default_replicas

@dataclass
class ScheduleRule:
    name: str
    cron: str  # Cron expression
    replicas: int
    timezone: str = "UTC"

    def matches(self, dt: datetime) -> bool:
        tz = pytz.timezone(self.timezone)
        local_dt = dt.astimezone(tz)
        return croniter.match(self.cron, local_dt)

Metric-Based Scaling (KEDA)

# ScaledObject for KEDA
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: agent-service-scaledobject
spec:
  scaleTargetRef:
    name: agent-service
  pollingInterval: 15
  cooldownPeriod: 300
  minReplicaCount: 3
  maxReplicaCount: 50
  triggers:
  - type: prometheus
    metadata:
      serverAddress: http://prometheus:9090
      metricName: http_requests_per_second
      query: sum(rate(http_requests_total{job="agent-service"}[1m])) by (pod)
      threshold: "50"
      activateOnValue: "10"
  - type: cpu
    metadata:
      type: Utilization
      value: "70"
  - type: memory
    metadata:
      type: Utilization
      value: "80"
  - type: external
    metadata:
      scalerName: keda-scaler-kafka
      topic: agent-tasks
      consumerGroup: agent-workers
      lagThreshold: "100"

Database Scaling

Read Replicas

class DatabasePool:
    def __init__(self, primary: Engine, replicas: list[Engine]):
        self.primary = primary
        self.replicas = replicas
        self.replica_index = 0

    def get_connection(self, read_only: bool = True) -> Connection:
        if read_only and self.replicas:
            # Round-robin
            conn = self.replicas[self.replica_index]
            self.replica_index = (self.replica_index + 1) % len(self.replicas)
            return conn
        return self.primary

    async def execute_read(self, query: str, params: dict = None):
        async with self.get_connection(read_only=True) as conn:
            return await conn.execute(query, params)

    async def execute_write(self, query: str, params: dict = None):
        async with self.primary.begin() as conn:
            return await conn.execute(query, params)

Connection Pooling (PgBouncer)

# pgbouncer.ini
[databases]
agent_db = host=primary.db.example.com port=5432 dbname=agent_db

[pgbouncer]
pool_mode = transaction
max_client_conn = 1000
default_pool_size = 100
min_pool_size = 10
reserve_pool_size = 10
reserve_pool_timeout = 5.0

# Connection limits
max_db_connections = 200
max_user_connections = 100

# Timeouts
server_lifetime = 3600
server_idle_timeout = 600
server_connect_timeout = 15
server_login_retry = 15
query_timeout = 30
query_wait_timeout = 120

# Logging
log_connections = 1
log_disconnections = 1
log_pooler_errors = 1

Sharding

class ShardedDatabase:
    def __init__(self, shards: list[DatabasePool]):
        self.shards = shards
        self.router = ShardingRouter(len(shards))

    def get_shard(self, key: str) -> DatabasePool:
        shard_id = self.router.get_shard(key)
        return self.shards[shard_id]

    async def execute(self, key: str, query: str, params: dict = None, 
                      read_only: bool = True):
        shard = self.get_shard(key)
        return await shard.execute_read(query, params) if read_only \
               else await shard.execute_write(query, params)

# Sharding strategies
class ShardingRouter:
    def __init__(self, num_shards: int, strategy: str = "hash"):
        self.num_shards = num_shards
        self.strategy = strategy

    def get_shard(self, key: str) -> int:
        if self.strategy == "hash":
            return int(hashlib.md5(key.encode()).hexdigest(), 16) % self.num_shards
        elif self.strategy == "range":
            return self.range_based(key)
        elif self.strategy == "consistent_hash":
            return self.consistent_hash(key)
        return 0

Caching Layers

Multi-Level Caching

class MultiLevelCache:
    def __init__(self):
        self.l1 = LRUCache(maxsize=1000)      # In-memory
        self.l2 = RedisCache(ttl=300)         # Redis
        self.l3 = DatabaseCache(ttl=3600)     # Database

    async def get(self, key: str) -> Any:
        # L1
        value = self.l1.get(key)
        if value is not None:
            return value

        # L2
        value = await self.l2.get(key)
        if value is not None:
            self.l1.set(key, value)
            return value

        # L3
        value = await self.l3.get(key)
        if value is not None:
            self.l1.set(key, value)
            await self.l2.set(key, value)
            return value

        return None

    async def set(self, key: str, value: Any, ttl: int = 300):
        self.l1.set(key, value)
        await self.l2.set(key, value, ttl=ttl)
        await self.l3.set(key, value, ttl=ttl)

Cache Invalidation

class CacheInvalidator:
    def __init__(self, cache: MultiLevelCache, pubsub: PubSub):
        self.cache = cache
        self.pubsub = pubsub
        await self.pubsub.subscribe("cache:invalidate", self.on_invalidate)

    async def on_invalidate(self, message: dict):
        keys = message.get("keys", [])
        tags = message.get("tags", [])

        for key in keys:
            await self.cache.delete(key)

        # Tag-based invalidation
        if tags:
            keys_to_invalidate = await self.get_keys_by_tags(tags)
            for key in keys_to_invalidate:
                await self.cache.delete(key)

    async def invalidate(self, keys: list[str] = None, tags: list[str] = None):
        message = {"keys": keys or [], "tags": tags or []}
        await self.pubsub.publish("cache:invalidate", message)

Cost-Aware Scaling

Cost-Aware Autoscaler

class CostAwareAutoscaler:
    def __init__(self, max_budget_usd_per_hour: float = 100.0):
        self.max_budget = max_budget_usd_per_hour
        self.instance_costs = {
            "small": 0.05,    # $0.05/hr
            "medium": 0.15,   # $0.15/hr
            "large": 0.40,    # $0.40/hr
            "xlarge": 0.80,   # $0.80/hr
            "gpu": 2.50       # $2.50/hr
        }

    def calculate_optimal_replicas(self, 
                                   load: LoadMetrics,
                                   current_replicas: int) -> int:
        # Estimate required capacity
        required_capacity = self.estimate_capacity(load)

        # Find cheapest combination
        best_combo = self.find_optimal_combination(required_capacity)

        # Check budget
        estimated_cost = sum(self.instance_costs[t] * c 
                            for t, c in best_combo.items())

        if estimated_cost > self.max_budget:
            # Scale down to fit budget
            return self.scale_to_budget(current_replicas, self.max_budget)

        return sum(best_combo.values())

    def find_optimal_combination(self, capacity: float) -> dict[str, int]:
        # Dynamic programming for knapsack-like problem
        # Maximize capacity within budget
        dp = {}
        # ... implementation
        return best_combination

Related Notes

• Deployment Architecture — Deployment patterns
• Observability — Monitoring scaling
• Reliability — Circuit breakers, retries
• Cost Optimization — Cost optimization
• Compliance — Compliance
• CI CD Best Practices — CI/CD
• Infrastructure as Code — IaC
• Observability — Observability
• Git Workflow — GitOps
• Supply Chain Security — Supply chain
• CI CD Best Practices — CI/CD
• Research Directions — Scaling research

Sources: Kubernetes HPA/VPA, KEDA, Prometheus Adaptive Scaling, AWS Auto Scaling, GCP Autoscaler, Azure Autoscale, Netflix Scaling, Uber Scaling, Uber Michelangelo, Uber Michelangelo, Lyft Scaling, DoorDash Scaling, Shopify Scaling, Kubernetes Best Practices, HPA/VPA Best Practices, KEDA Best Practices, Prometheus Adaptive Scaling, Custom Metrics Autoscaling, Cluster Autoscaler, Node Autoscaling, Pod Disruption Budgets, Pod Topology Spread Constraints, Priority Classes, Resource Quotas, Limit Ranges, Quality of Service Classes.