Scalability Fundamentals Questions

import math

def compute_instances(peak_rps, per_instance_rps, redundancy_factor=1.0, safety_margin=0.0):
    """
    Returns number of instances required and explanation for rounding.
    - peak_rps: observed or expected peak requests per second (float)
    - per_instance_rps: capacity of one instance (float)
    - redundancy_factor: multiplicative factor for redundancy (e.g., 2 for N+1)
    - safety_margin: fractional safety margin (e.g., 0.2 for +20%)
    """
    if peak_rps < 0 or per_instance_rps <= 0 or redundancy_factor < 1 or safety_margin < 0:
        raise ValueError("Invalid inputs: peak>=0, per_instance>0, redundancy>=1, safety_margin>=0")

    required_capacity = peak_rps * (1 + safety_margin) * redundancy_factor
    raw_instances = required_capacity / per_instance_rps
    instances = math.ceil(raw_instances)

    explanation = (
        f"Peak RPS: {peak_rps}\n"
        f"Safety margin: {safety_margin*100:.1f}% -> scaled peak: {peak_rps*(1+safety_margin):.2f}\n"
        f"Redundancy factor: {redundancy_factor} -> required capacity: {required_capacity:.2f} RPS\n"
        f"Per-instance capacity: {per_instance_rps} RPS -> raw instances: {raw_instances:.3f}\n"
        f"Rounding: ceil to {instances} instances to ensure capacity meets targets (cannot provision fractional instances)\n"
    )
    return instances, explanation

# Sample invocation and sample output:
if __name__ == "__main__":
    inst, note = compute_instances(peak_rps=1200, per_instance_rps=400, redundancy_factor=2, safety_margin=0.15)
    print(f"Instances required: {inst}\n")
    print(note)

"""
Sample output:
Instances required: 7

Peak RPS: 1200
Safety margin: 15.0% -> scaled peak: 1380.00
Redundancy factor: 2 -> required capacity: 2760.00 RPS
Per-instance capacity: 400 RPS -> raw instances: 6.900
Rounding: ceil to 7 instances to ensure capacity meets targets (cannot provision fractional instances)
"""

import math
import unittest

def capacity_plan(incoming_rps, cache_hit_rate, downstream_requests_per_miss, instance_capacity):
    """
    Returns (backend_rps, number_of_instances_needed)
    - incoming_rps: total incoming requests per second (float >=0)
    - cache_hit_rate: fraction [0,1] of requests served by cache
    - downstream_requests_per_miss: average downstream calls triggered per miss (>=0)
    - instance_capacity: downstream request handling capacity per instance (requests/sec, >0)
    """
    if incoming_rps < 0:
        raise ValueError("incoming_rps must be >= 0")
    if not (0 <= cache_hit_rate <= 1):
        raise ValueError("cache_hit_rate must be between 0 and 1")
    if downstream_requests_per_miss < 0:
        raise ValueError("downstream_requests_per_miss must be >= 0")
    if instance_capacity <= 0:
        raise ValueError("instance_capacity must be > 0")
    miss_rate = 1.0 - cache_hit_rate
    backend_rps = incoming_rps * miss_rate * downstream_requests_per_miss
    instances = math.ceil(backend_rps / instance_capacity) if backend_rps > 0 else 0
    return backend_rps, instances

class TestCapacityPlan(unittest.TestCase):
    def test_basic(self):
        brps, inst = capacity_plan(1000, 0.9, 1, 100)
        # 10% misses -> 100 misses * 1 downstream = 100 backend rps -> needs 1 instance
        self.assertEqual(brps, 100.0)
        self.assertEqual(inst, 1)

    def test_multiple_downstream(self):
        brps, inst = capacity_plan(500, 0.8, 3, 200)
        # misses=100 -> backend=300 -> instances=2
        self.assertEqual(brps, 300.0)
        self.assertEqual(inst, 2)

    def test_zero_traffic(self):
        brps, inst = capacity_plan(0, 0.5, 5, 10)
        self.assertEqual(brps, 0.0)
        self.assertEqual(inst, 0)

    def test_full_cache(self):
        brps, inst = capacity_plan(100, 1.0, 10, 10)
        self.assertEqual(brps, 0.0)
        self.assertEqual(inst, 0)

    def test_invalid_inputs(self):
        with self.assertRaises(ValueError):
            capacity_plan(-1, 0.5, 1, 10)
        with self.assertRaises(ValueError):
            capacity_plan(10, 1.5, 1, 10)
        with self.assertRaises(ValueError):
            capacity_plan(10, 0.5, -1, 10)
        with self.assertRaises(ValueError):
            capacity_plan(10, 0.5, 1, 0)

if __name__ == "__main__":
    unittest.main()