Merge pull request #139 from Netflix-Skunkworks/vrayini/kafka-fix-replication

Fix Kafka Capacity Model Replication Factor

Author: rayiniv-nflx

service_capacity_modeling/models/org/netflix/kafka.py

@@ -159,7 +159,7 @@ def _estimate_kafka_requirement(  # pylint: disable=too-many-positional-argument
     needed_network_mbps = max(bw_in, bw_out) * 1.40
 
     needed_disk = math.ceil(
-        desires.data_shape.estimated_state_size_gib.mid * copies_per_region,
+        desires.data_shape.estimated_state_size_gib.mid,
     )
 
     # Keep the last N seconds hot in cache
@@ -416,6 +416,10 @@ class NflxKafkaArguments(BaseModel):
 
 
 class NflxKafkaCapacityModel(CapacityModel):
+
+    HA_DEFAULT_REPLICATION_FACTOR = 2
+    SC_DEFAULT_REPLICATION_FACTOR = 3
+
     @staticmethod
     def capacity_plan(
         instance: Instance,
@@ -427,11 +431,11 @@ def capacity_plan(
         cluster_type: ClusterType = ClusterType(
             extra_model_arguments.get("cluster_type", "high-availability")
         )
-        default_replication = 2
+        replication_factor = NflxKafkaCapacityModel.HA_DEFAULT_REPLICATION_FACTOR
         if cluster_type == ClusterType.strong:
-            default_replication = 3
+            replication_factor = NflxKafkaCapacityModel.SC_DEFAULT_REPLICATION_FACTOR
         copies_per_region: int = extra_model_arguments.get(
-            "copies_per_region", default_replication
+            "copies_per_region", replication_factor
         )
         if cluster_type == ClusterType.strong and copies_per_region < 3:
             raise ValueError("Strong consistency and RF<3 doesn't work")
@@ -503,8 +507,13 @@ def default_desires(user_desires, extra_model_arguments: Dict[str, Any]):
         retention = extra_model_arguments.get("retention", "PT8H")
         retention_secs = iso_to_seconds(retention)
 
-        # write throughput * retention = usage
-        state_gib = (write_bytes.mid * retention_secs) / GIB_IN_BYTES
+        # write throughput * retention * replication factor = usage
+        replication_factor = NflxKafkaCapacityModel.HA_DEFAULT_REPLICATION_FACTOR
+        if extra_model_arguments.get("cluster_type", None) == ClusterType.strong:
+            replication_factor = NflxKafkaCapacityModel.SC_DEFAULT_REPLICATION_FACTOR
+        state_gib = (
+            write_bytes.mid * retention_secs * replication_factor
+        ) / GIB_IN_BYTES
 
         return CapacityDesires(
             query_pattern=QueryPattern(

tests/netflix/test_kafka.py

@@ -1,8 +1,13 @@
 from service_capacity_modeling.capacity_planner import planner
+from service_capacity_modeling.interface import AccessPattern
 from service_capacity_modeling.interface import CapacityDesires
 from service_capacity_modeling.interface import certain_float
 from service_capacity_modeling.interface import CurrentClusters
 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 DriveType
+from service_capacity_modeling.interface import FixedInterval
 from service_capacity_modeling.interface import Interval
 from service_capacity_modeling.interface import QueryPattern
 from service_capacity_modeling.models.common import normalize_cores
@@ -65,7 +70,7 @@ def test_kafka_large_scale():
         region="us-east-1",
         desires=desires,
         extra_model_arguments={
-            "cluster_type": "strong",
+            "cluster_type": ClusterType.strong,
             "retention": "PT4H",
         },
     )
@@ -339,3 +344,94 @@ def test_plan_certain():
     assert len(lr_clusters) >= 1
     print(lr_clusters[0].instance.name)
     assert lr_clusters[0].count == cluster_capacity.cluster_instance_count.high
+
+
+def test_plan_certain_data_shape():
+    """
+    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,
+        },
+    )
+
+    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
+    for lr in cap_plan:
+        print(lr.candidate_clusters.zonal[0])