Merge pull request #141 from Netflix-Skunkworks/vrayini/vertical-scale-param

Restrict Instance Type Param + Change Tier Target CPU Bounds

Author: rayiniv-nflx

service_capacity_modeling/models/org/netflix/kafka.py

@@ -18,6 +18,7 @@
 from service_capacity_modeling.interface import certain_int
 from service_capacity_modeling.interface import Clusters
 from service_capacity_modeling.interface import Consistency
+from service_capacity_modeling.interface import CurrentZoneClusterCapacity
 from service_capacity_modeling.interface import DataShape
 from service_capacity_modeling.interface import Drive
 from service_capacity_modeling.interface import GIB_IN_BYTES
@@ -47,13 +48,30 @@ def target_cpu_utilization(tier: int) -> float:
     """
     Returns the target average cluster CPU utilization for a given tier
     """
-    if tier == 0:
-        return 0.25
-    if tier == 1:
-        return 0.30
-    if tier == 2:
-        return 0.35
-    return 0.40
+    if tier in (0, 1):
+        return 0.40
+    return 0.50
+
+
+def _get_current_zonal_cluster(
+    desires: CapacityDesires,
+) -> Optional[CurrentZoneClusterCapacity]:
+    return (
+        None
+        if desires.current_clusters is None
+        else (
+            desires.current_clusters.zonal[0]
+            if len(desires.current_clusters.zonal)
+            else None
+        )
+    )
+
+
+def _is_same_instance_family(cluster, target_family):
+    """Check if cluster has a different instance family than the target."""
+    if not cluster or not cluster.cluster_instance:
+        return False
+    return cluster.cluster_instance.family == target_family
 
 
 def _estimate_kafka_requirement(  # pylint: disable=too-many-positional-arguments
@@ -90,15 +108,7 @@ def _estimate_kafka_requirement(  # pylint: disable=too-many-positional-argument
         write_mib_per_second
     )
     # use the current cluster capacity if available
-    current_zonal_cluster = (
-        None
-        if desires.current_clusters is None
-        else (
-            desires.current_clusters.zonal[0]
-            if len(desires.current_clusters.zonal)
-            else None
-        )
-    )
+    current_zonal_cluster = _get_current_zonal_cluster(desires)
 
     if (
         current_zonal_cluster
@@ -230,6 +240,7 @@ def _estimate_kafka_cluster_zonal(  # pylint: disable=too-many-positional-argume
     max_local_disk_gib: int = 1024 * 5,
     min_instance_cpu: int = 2,
     min_instance_memory_gib: int = 12,
+    require_same_instance_family: bool = True,
 ) -> Optional[CapacityPlan]:
 
     # Kafka doesn't like to deploy on single CPU instances or with < 12 GiB of ram
@@ -248,6 +259,15 @@ def _estimate_kafka_cluster_zonal(  # pylint: disable=too-many-positional-argume
     if instance.drive is None and drive.name != "gp3":
         return None
 
+    # If there is a current cluster, check if we are restricted to same instance family
+    current_zonal_cluster = _get_current_zonal_cluster(desires)
+    if (
+        current_zonal_cluster
+        and require_same_instance_family
+        and not _is_same_instance_family(current_zonal_cluster, instance.family)
+    ):
+        return None
+
     requirement, regrets = _estimate_kafka_requirement(
         instance=instance,
         desires=desires,
@@ -461,6 +481,10 @@ def capacity_plan(
         required_zone_size: Optional[int] = extra_model_arguments.get(
             "required_zone_size", None
         )
+        # By default, for existing clusters, restrict to only using same instance family
+        require_same_instance_family: bool = extra_model_arguments.get(
+            "require_same_instance_family", True
+        )
 
         return _estimate_kafka_cluster_zonal(
             instance=instance,
@@ -476,6 +500,7 @@ def capacity_plan(
             min_instance_cpu=min_instance_cpu,
             min_instance_memory_gib=min_instance_memory_gib,
             hot_retention_seconds=hot_retention_seconds,
+            require_same_instance_family=require_same_instance_family,
         )
 
     @staticmethod

tests/netflix/test_kafka.py

@@ -425,6 +425,7 @@ def test_plan_certain_data_shape():
             "retention": "PT8H",
             "require_attached_disks": True,
             "required_zone_size": cluster_capacity.cluster_instance_count.mid,
+            "require_same_instance_family": False,
         },
     )
 
@@ -435,3 +436,113 @@ def test_plan_certain_data_shape():
     assert lr_clusters[0].count == cluster_capacity.cluster_instance_count.high
     for lr in cap_plan:
         print(lr.candidate_clusters.zonal[0])
+    families = set(
+        map(
+            lambda curr_plan: curr_plan.candidate_clusters.zonal[0].instance.family,
+            cap_plan,
+        )
+    )
+    # check that we did not restrict the instance family to only r7a
+    assert families != {"r7a"}
+
+
+def test_plan_certain_data_shape_same_instance_type():
+    """
+    Use current clusters cpu utilization to determine instance types directly as
+    supposed to extrapolating it from the Data Shape
+    """
+    cluster_capacity = CurrentZoneClusterCapacity(
+        cluster_instance_name="r7a.4xlarge",
+        cluster_drive=Drive(
+            name="gp3",
+            drive_type=DriveType.attached_ssd,
+            size_gib=5000,
+            block_size_kib=16,
+        ),
+        cluster_instance_count=Interval(low=15, mid=15, high=15, confidence=1),
+        cpu_utilization=Interval(
+            low=5.441147804260254,
+            mid=13.548842955300195,
+            high=25.11203956604004,
+            confidence=1,
+        ),
+        memory_utilization_gib=Interval(low=0, mid=0, high=0, confidence=1),
+        network_utilization_mbps=Interval(
+            low=4580.919447446355,
+            mid=19451.59814477331,
+            high=42963.441154527085,
+            confidence=1,
+        ),
+        disk_utilization_gib=Interval(
+            low=1341.579345703125,
+            mid=1940.8741284013684,
+            high=2437.607421875,
+            confidence=1,
+        ),
+    )
+
+    desires = CapacityDesires(
+        service_tier=1,
+        current_clusters=CurrentClusters(zonal=[cluster_capacity]),
+        query_pattern=QueryPattern(
+            access_pattern=AccessPattern(AccessPattern.latency),
+            # 2 consumers
+            estimated_read_per_second=Interval(low=2, mid=2, high=4, confidence=1),
+            # 1 producer
+            estimated_write_per_second=Interval(low=1, mid=1, high=1, confidence=0.98),
+            estimated_mean_read_latency_ms=Interval(low=1, mid=1, high=1, confidence=1),
+            estimated_mean_write_latency_ms=Interval(
+                low=1, mid=1, high=1, confidence=1
+            ),
+            estimated_mean_read_size_bytes=Interval(
+                low=1024, mid=1024, high=1024, confidence=1
+            ),
+            estimated_mean_write_size_bytes=Interval(
+                low=125000000, mid=579000000, high=1351000000, confidence=0.98
+            ),
+            estimated_read_parallelism=Interval(low=1, mid=1, high=1, confidence=1),
+            estimated_write_parallelism=Interval(low=1, mid=1, high=1, confidence=1),
+            read_latency_slo_ms=FixedInterval(low=0.4, mid=4, high=10, confidence=0.98),
+            write_latency_slo_ms=FixedInterval(
+                low=0.4, mid=4, high=10, confidence=0.98
+            ),
+        ),
+        data_shape=DataShape(
+            estimated_state_size_gib=Interval(
+                low=44000, mid=86000, high=91000, confidence=1
+            ),
+        ),
+    )
+
+    cap_plan = planner.plan_certain(
+        model_name="org.netflix.kafka",
+        region="us-east-1",
+        num_results=3,
+        num_regions=4,
+        desires=desires,
+        extra_model_arguments={
+            "cluster_type": ClusterType.ha,
+            "retention": "PT8H",
+            "require_attached_disks": True,
+            "required_zone_size": cluster_capacity.cluster_instance_count.mid,
+            "require_same_instance_family": True,
+        },
+    )
+
+    assert len(cap_plan) >= 1
+    lr_clusters = cap_plan[0].candidate_clusters.zonal
+    assert len(lr_clusters) >= 1
+    print(lr_clusters[0].instance.name)
+    assert lr_clusters[0].count == cluster_capacity.cluster_instance_count.high
+
+    families = set(
+        map(
+            lambda curr_plan: curr_plan.candidate_clusters.zonal[0].instance.family,
+            cap_plan,
+        )
+    )
+    # check that we restricted the instance family to only r7a
+    assert families == {"r7a"}
+
+    for lr in cap_plan:
+        print(lr.candidate_clusters.zonal[0])