Added PortMappings dataclass for better type hinting

deployments/sequencer/config/sequencer.py

@@ -1,29 +1,20 @@
-from typing import Dict, Any
 import os
 import json
 import jsonschema
 
-ROOT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), '../../../')
-CONFIG_DIR = os.path.join(ROOT_DIR, 'config/papyrus/')
-
-
-class Config():
-    def __init__(self, schema: Dict[Any, Any], config: Dict[Any, Any]):
-        self.schema = schema
-        self.config = config
+from services.objects import Config
 
-    def get(self):
-        return self.config
 
-    def validate(self):
-        pass
+ROOT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), '../../../')
+CONFIG_DIR = os.path.join(ROOT_DIR, 'config/papyrus/')
 
 
 class SequencerDevConfig(Config):
-    def __init__(self):
+    def __init__(self, mount_path: str):
         super().__init__(
             schema=json.loads(open(os.path.join(CONFIG_DIR, 'default_config.json'), 'r').read()),
-            config=json.loads(open(os.path.join(CONFIG_DIR, 'presets', 'sepolia_testnet.json'), 'r').read())
+            config=json.loads(open(os.path.join(CONFIG_DIR, 'presets', 'sepolia_testnet.json'), 'r').read()),
+            mount_path = mount_path
         )
 
     def validate(self):

deployments/sequencer/main.py

@@ -9,7 +9,7 @@
 from services.service import Service
 from config.sequencer import Config, SequencerDevConfig
 from services.objects import (
-    HealthCheck, ServiceType, Probe, PersistentVolumeClaim
+    HealthCheck, ServiceType, Probe, PersistentVolumeClaim, PortMappings
 )
 from services import defaults
 
@@ -19,8 +19,7 @@ class SystemStructure:
     topology: str = "mesh"
     replicas: str = "2"
     size: str = "small"
-    config: Config = SequencerDevConfig()
-    health_check: Optional[HealthCheck] = None
+    config: Optional[Config] = None
 
     def __post_init__(self):
         self.config.validate()
@@ -40,15 +39,15 @@ def __init__(
         self.service = Service(
             self,
             "sequencer-node",
-            image="",
+            image="us.gcr.io/starkware-dev/sequencer-node:0.0.1",
             port_mappings=[
-                {"port": 80, "container_port": 8080},
-                {"port": 8081, "container_port": 8081},
-                {"port": 8082, "container_port": 8082}
+                PortMappings(name="http", port=80, container_port=8080),
+                PortMappings(name="rpc", port=8081, container_port=8081),
+                PortMappings(name="monitoring", port=8082, container_port=8082)
             ],
             service_type=ServiceType.CLUSTER_IP,
             replicas=1,
-            config=config.get(),
+            config=config,
             health_check=HealthCheck(
                 startup_probe=Probe(port=8082, path="/monitoring/NodeVersion", period_seconds=10, failure_threshold=10, timeout_seconds=5),
                 readiness_probe=Probe(port=8082, path="/monitoring/ready", period_seconds=10, failure_threshold=5, timeout_seconds=5),
@@ -81,46 +80,32 @@ def __init__(
             "mempool",
             image="paulbouwer/hello-kubernetes:1.7",
             replicas=2,
-            config=system_structure.config.get(),
+            config=system_structure.config,
             health_check=defaults.health_check
         )
         self.batcher = Service(
             self, 
             "batcher", 
-            image="ghost", 
-            container_port=2368, 
+            image="ghost",
+            port_mappings=[{"port": 80, "container_port": 2368}],
             health_check=defaults.health_check
         )
-        self.sequencer_node = Service(
-            self, 
-            "sequencer-node", 
-            image="",
-            container_port=8082,
-            replicas=1,
-            config=system_structure.config.get(),
-            service_type=ServiceType.CLUSTER_IP,
-            health_check=HealthCheck(
-                startup_probe=Probe(port="http", path="/monitoring/NodeVersion", period_seconds=10, failure_threshold=10, timeout_seconds=5),
-                readiness_probe=Probe(port="http", path="/monitoring/ready", period_seconds=10, failure_threshold=5, timeout_seconds=5),
-                liveness_probe=Probe(port="http", path="/monitoring/alive", period_seconds=10, failure_threshold=5, timeout_seconds=5)
-            )
-        )
 
 
 app = App(
-    yaml_output_type=YamlOutputType.FILE_PER_RESOURCE
+    yaml_output_type=YamlOutputType.FOLDER_PER_CHART_FILE_PER_RESOURCE
 )
 sequencer_node = SequencerNode(
     scope=app,
     name="sequencer-node",
     namespace="sequencer",
-    config=SequencerDevConfig()
+    config=SequencerDevConfig(mount_path="/app/config")
+)
+a = SequencerSystem(
+    scope=app,
+    name="sequencer-system",
+    namespace="test-namespace",
+    system_structure=SystemStructure(config=SequencerDevConfig(mount_path="/app/config")),
 )
-# a = SequencerSystem(
-#     scope=app,
-#     name="sequencer-system",
-#     namespace="test-namespace",
-#     system_structure=SystemStructure(),
-# )
 
 app.synth()

deployments/sequencer/services/objects.py

@@ -1,12 +1,12 @@
 import dataclasses
 
-from typing import Optional, Union
+from typing import Optional, Dict, Any, TypedDict
 from enum import Enum
 
 
 @dataclasses.dataclass
 class Probe:
-    port: Union[int, str]
+    port: int | str
     path: str
     period_seconds: int
     failure_threshold: int
@@ -33,11 +33,29 @@ class ServiceType(Enum):
 
 @dataclasses.dataclass
 class PersistentVolumeClaim:
-    storage_class_name: str = None
-    access_modes: list[str] = None
-    volume_mode: str = None
-    storage: str = None
+    storage_class_name: str | None = None
+    access_modes: list[str] | None = None
+    volume_mode: str | None = None
+    storage: str | None = None
     read_only: bool = True
-    mount_path: str = None
+    mount_path: str | None = None
 
 
+@dataclasses.dataclass
+class Config():
+    schema: Dict[Any, Any]
+    config: Dict[Any, Any]
+    mount_path: str
+
+    def get(self):
+        return self.config
+
+    def validate(self):
+        pass
+
+
+@dataclasses.dataclass
+class PortMappings(TypedDict):
+    name: str
+    port: int
+    container_port: int

deployments/sequencer/services/service.py

@@ -1,11 +1,11 @@
 import json
 
-from typing import Optional, Dict
+from typing import Optional, List
 from constructs import Construct
 from cdk8s import Names
 from imports import k8s
 
-from services.objects import HealthCheck, ServiceType, PersistentVolumeClaim
+from services.objects import *
 
 
 class Service(Construct):
@@ -17,8 +17,8 @@ def __init__(
         image: str,
         replicas: int = 1,
         service_type: Optional[ServiceType] = None,
-        port_mappings: Optional[list[Dict[str, int]]] = None,
-        config: Optional[Dict[str, str]] = None,
+        port_mappings: Optional[List[PortMappings]] = None,
+        config: Optional[Config] = None,
         health_check: Optional[HealthCheck] = None,
         pvc: Optional[PersistentVolumeClaim] = None
     ):
@@ -34,7 +34,8 @@ def __init__(
                     type=service_type.value if service_type is not None else None,
                     ports=[
                         k8s.ServicePort(
-                            port=port_map["port"],
+                            name=port_map.get("name"),
+                            port=port_map.get("port"),
                             target_port=k8s.IntOrString.from_number(port_map.get("container_port")),
                         ) for port_map in port_mappings
                     ],
@@ -46,7 +47,7 @@ def __init__(
             k8s.KubeConfigMap(
                 self,
                 "config",
-                data=dict(config=json.dumps(config)),
+                data=dict(config=json.dumps(config.get())),
             )
 
         k8s.KubeDeployment(
@@ -60,7 +61,7 @@ def __init__(
                     spec=k8s.PodSpec(
                         containers=[
                             k8s.Container(
-                                name="web",
+                                name="sequencer",
                                 image=image,
                                 ports=[k8s.ContainerPort(container_port=port_map.get("container_port")) for port_map in port_mappings or []],
                                 startup_probe=k8s.Probe(
@@ -103,7 +104,7 @@ def __init__(
                                     mount for mount in [
                                         k8s.VolumeMount(
                                             name=f"{self.node.id}-config",
-                                            mount_path="/",
+                                            mount_path=config.mount_path,
                                             read_only=True
                                         ) if config is not None else None,