chore(deployment): casting k8s objects

deployments/sequencer/.gitignore

@@ -0,0 +1,2 @@
+.idea/
+__pycache__/

deployments/sequencer/app/service.py

@@ -5,13 +5,15 @@
 from cdk8s import Names, ApiObjectMetadata
 from imports import k8s
 from imports.com.google import cloud as google
+from imports.k8s import ResourceClaim, Quantity
 
 from services import topology
+from services.objects import ContainerResources
 
 
 class ServiceApp(Construct):
     def __init__(
-        self, scope: Construct, id: str, *, namespace: str, topology: topology.ServiceTopology
+            self, scope: Construct, id: str, *, namespace: str, topology: topology.ServiceTopology
     ):
         super().__init__(scope, id)
 
@@ -80,73 +82,15 @@ def set_k8s_deployment(self):
                             k8s.Container(
                                 name=f"{self.node.id}-{container.name}",
                                 image=container.image,
-                                # command=["sleep", "infinity"],
+                                command=container.command,
                                 args=container.args,
-                                ports=[
-                                    k8s.ContainerPort(container_port=port.container_port)
-                                    for port in container.ports
-                                ],
-                                startup_probe=k8s.Probe(
-                                    http_get=k8s.HttpGetAction(
-                                        path=container.startup_probe.path,
-                                        port=k8s.IntOrString.from_string(
-                                            container.startup_probe.port
-                                        )
-                                        if isinstance(container.startup_probe.port, str)
-                                        else k8s.IntOrString.from_number(
-                                            container.startup_probe.port
-                                        ),
-                                    ),
-                                    period_seconds=container.startup_probe.period_seconds,
-                                    failure_threshold=container.startup_probe.failure_threshold,
-                                    timeout_seconds=container.startup_probe.timeout_seconds,
-                                )
-                                if container.startup_probe is not None
-                                else None,
-                                readiness_probe=k8s.Probe(
-                                    http_get=k8s.HttpGetAction(
-                                        path=container.readiness_probe.path,
-                                        port=k8s.IntOrString.from_string(
-                                            container.readiness_probe.port
-                                        )
-                                        if isinstance(container.readiness_probe.port, str)
-                                        else k8s.IntOrString.from_number(
-                                            container.readiness_probe.port
-                                        ),
-                                    ),
-                                    period_seconds=container.readiness_probe.period_seconds,
-                                    failure_threshold=container.readiness_probe.failure_threshold,
-                                    timeout_seconds=container.readiness_probe.timeout_seconds,
-                                )
-                                if container.readiness_probe is not None
-                                else None,
-                                liveness_probe=k8s.Probe(
-                                    http_get=k8s.HttpGetAction(
-                                        path=container.liveness_probe.path,
-                                        port=k8s.IntOrString.from_string(
-                                            container.liveness_probe.port
-                                        )
-                                        if isinstance(container.liveness_probe.port, str)
-                                        else k8s.IntOrString.from_number(
-                                            container.liveness_probe.port
-                                        ),
-                                    ),
-                                    period_seconds=container.liveness_probe.period_seconds,
-                                    failure_threshold=container.liveness_probe.failure_threshold,
-                                    timeout_seconds=container.liveness_probe.timeout_seconds,
-                                )
-                                if container.liveness_probe is not None
-                                else None,
-                                volume_mounts=[
-                                    k8s.VolumeMount(
-                                        name=mount.name,
-                                        mount_path=mount.mount_path,
-                                        read_only=mount.read_only,
-                                    )
-                                    for mount in container.volume_mounts
-                                ],
-                            )
-                            for container in self.topology.deployment.containers
+                                ports=[container_port.to_k8s() for container_port in container.ports],
+                                resources=container.resources.to_k8s(),
+                                startup_probe=container.startup_probe.to_k8s(),
+                                readiness_probe=container.readiness_probe.to_k8s(),
+                                liveness_probe=container.liveness_probe.to_k8s(),
+                                volume_mounts=[mount.to_k8s() for mount in container.volume_mounts],
+                            ) for container in self.topology.deployment.containers
                         ],
                         volumes=list(
                             chain(

deployments/sequencer/services/objects.py

@@ -1,19 +1,49 @@
 import dataclasses
 from typing import Optional, List, Dict, Any, Mapping, Sequence
+
+from imports import k8s
 from services import const
 
 
+def to_quantity(value: str | int | float) -> k8s.Quantity:
+    if isinstance(value, str):
+        return k8s.Quantity.from_string(value)
+    elif isinstance(value, (int, float)):
+        return k8s.Quantity.from_number(value)
+    else:
+        raise ValueError("Value must be of type int, float or str.")
+
+
+def to_int_or_string(value: str | int | float):
+    if isinstance(value, str):
+        return k8s.IntOrString.from_string(value)
+    elif isinstance(value, (int, float)):
+        return k8s.IntOrString.from_number(value)
+    else:
+        raise ValueError("Value must be of type int, float or str.")
+
+
 @dataclasses.dataclass
-class Probe:
+class HttpProbe:
     port: int | str
     path: str
     period_seconds: int
     failure_threshold: int
     timeout_seconds: int
 
     def __post_init__(self):
-        assert not isinstance(self.port, (bool)), "Port must be of type int or str, not bool."
-
+        assert not isinstance(self.port, bool), "Port must be of type int or str, not bool."
+
+    def to_k8s(self) -> k8s.Probe:
+        return k8s.Probe(
+            http_get=k8s.HttpGetAction(
+                port=to_int_or_string(self.port),
+                path=self.path,
+            ),
+            period_seconds=self.period_seconds,
+            failure_threshold=self.failure_threshold,
+            timeout_seconds=self.timeout_seconds
+        )
 
 @dataclasses.dataclass
 class PortMapping:
@@ -87,6 +117,13 @@ class VolumeMount:
     mount_path: str
     read_only: bool
 
+    def to_k8s(self) -> k8s.VolumeMount:
+        return k8s.VolumeMount(
+            name=self.name,
+            mount_path=self.mount_path,
+            read_only=self.read_only
+        )
+
 
 @dataclasses.dataclass
 class ConfigMapVolume:
@@ -99,21 +136,52 @@ class PvcVolume:
     read_only: bool
 
 
+@dataclasses.dataclass
+class ContainerResources:
+    requests_cpu: str | int
+    requests_memory: str
+    limits_cpu: str | int
+    limits_memory: str
+
+    def to_k8s(self) -> k8s.ResourceRequirements:
+        return k8s.ResourceRequirements(
+            requests={
+                "cpu": to_quantity(self.requests_cpu),
+                "memory": to_quantity(self.requests_memory),
+            },
+            limits={
+                "cpu": to_quantity(self.limits_cpu),
+                "memory": to_quantity(self.limits_memory),
+            }
+        )
+
+
 @dataclasses.dataclass
 class ContainerPort:
-    container_port: int
+    port: int
+    name: Optional[str] = None
+    protocol: Optional[str] = None
+
+    def to_k8s(self) -> k8s.ContainerPort:
+        return k8s.ContainerPort(
+            container_port=self.port,
+            name=self.name,
+            protocol=self.protocol
+        )
 
 
 @dataclasses.dataclass
 class Container:
     name: str
     image: str
-    args: List[str]
     ports: Sequence[ContainerPort]
-    startup_probe: Optional[Probe]
-    readiness_probe: Optional[Probe]
-    liveness_probe: Optional[Probe]
+    resources: ContainerResources
+    startup_probe: Optional[HttpProbe]
+    readiness_probe: Optional[HttpProbe]
+    liveness_probe: Optional[HttpProbe]
     volume_mounts: Sequence[VolumeMount]
+    args: Optional[List[str]] = None
+    command: Optional[List[str]] = None
 
 
 @dataclasses.dataclass

deployments/sequencer/services/topology_helpers.py

@@ -1,5 +1,6 @@
 from services import objects, const, helpers
 from config.sequencer import SequencerDevConfig
+from services.objects import ContainerResources
 
 
 def get_pvc() -> objects.PersistentVolumeClaim:
@@ -81,25 +82,31 @@ def get_deployment() -> objects.Deployment:
                 image="us.gcr.io/starkware-dev/sequencer-node-test:0.0.1-dev.3",
                 args=["--config_file", "/config/sequencer/presets/config"],
                 ports=[
-                    objects.ContainerPort(container_port=const.HTTP_CONTAINER_PORT),
-                    objects.ContainerPort(container_port=const.RPC_CONTAINER_PORT),
-                    objects.ContainerPort(container_port=const.MONITORING_CONTAINER_PORT),
+                    objects.ContainerPort(port=const.HTTP_CONTAINER_PORT),
+                    objects.ContainerPort(port=const.RPC_CONTAINER_PORT),
+                    objects.ContainerPort(port=const.MONITORING_CONTAINER_PORT),
                 ],
-                startup_probe=objects.Probe(
+                resources=objects.ContainerResources(
+                    requests_cpu=4,
+                    requests_memory="8Gi",
+                    limits_cpu=8,
+                    limits_memory="16Gi"
+                ),
+                startup_probe=objects.HttpProbe(
                     port=const.MONITORING_CONTAINER_PORT,
                     path="/monitoring/nodeVersion",
                     period_seconds=10,
                     failure_threshold=10,
                     timeout_seconds=5,
                 ),
-                readiness_probe=objects.Probe(
+                readiness_probe=objects.HttpProbe(
                     port=const.MONITORING_CONTAINER_PORT,
                     path="/monitoring/ready",
                     period_seconds=10,
                     failure_threshold=5,
                     timeout_seconds=5,
                 ),
-                liveness_probe=objects.Probe(
+                liveness_probe=objects.HttpProbe(
                     port=const.MONITORING_CONTAINER_PORT,
                     path="/monitoring/alive",
                     period_seconds=10,