Start to implement k8s deployment

Author: flmarrero

.env

@@ -1,4 +1,3 @@
 COMPOSE_PATH_SEPARATOR=:
-COMPOSE_FILE=docker-compose.yml:clickhouse/compose.clickhouse.yml:metastore/compose.metastore.yml
-
-. ./.secrets
+COMPOSE_FILE=docker-compose.yml:clickhouse/compose.clickhouse.yml:nessie/compose.nessie.yml:openmetadata/compose.openmetadata.yml:dagster/compose.dagster.yml:spark/compose.spark.yml:pipelines/compose.pipelines.yml:kafka/compose.kafka.yml
+#:cube/compose.cube.yml:prometheus/compose.prometheus.yml

.envrc

@@ -1 +1 @@
-. ./.env
+use flake

.gitignore

@@ -1,2 +1,14 @@
-.ipynb_checkpoints
-.secrets
+.ipynb_checkpoints/
+__pycache__/
+.logs_queue/
+.nux/
+.telemetry/
+pipelines/storage/
+pipelines/schedules/
+pipelines/history/
+metastore_db/
+env/
+!doc/env/
+*.log
+.venv/
+.direnv/

.secrets.template

@@ -32,6 +32,8 @@ services:
       retries: 10
     restart:
       always
+    env_file:
+      - env/minio.env
 
   clickhouse-02:
     image: "clickhouse/clickhouse-server:24"
@@ -66,6 +68,8 @@ services:
       retries: 10
     restart:
       always
+    env_file:
+      - env/minio.env
 
   clickhouse-03:
     image: "clickhouse/clickhouse-server:24"
@@ -100,6 +104,8 @@ services:
       retries: 10
     restart:
       always
+    env_file:
+      - env/minio.env
 
   clickhouse-04:
     image: "clickhouse/clickhouse-server:24"
@@ -134,6 +140,8 @@ services:
       retries: 10
     restart:
       always
+    env_file:
+      - env/minio.env
 
   clickhouse-keeper-01:
     image: "clickhouse/clickhouse-keeper:24"

clickhouse/compose.clickhouse.yml

@@ -1,17 +1,19 @@
 services:
   dagster_postgresql:
     image: postgres:16
-    container_name: dagster-postgresql
-    environment:
-      DAGSTER_POSTGRES_USER: ${DAGSTER_DATABASE_USERNAME}
-      DAGSTER_POSTGRES_PASSWORD: ${DAGSTER_DATABASE_PASSWORD}
-      DAGSTER_POSTGRES_DB: ${DAGSTER_DATABASE}
+    container_name: dagster_postgresql
+    hostname: dagster_postgresql
     networks:
       - dagster-internal
+      - public
+    env_file:
+      - env/dagster.env
+      - env/pipeline.env
 
   dagster_webserver:
     build: dagster/dagster
     container_name: dagster_webserver
+    hostname: dagster_webserver
     restart: on-failure
     entrypoint:
       - dagster-webserver
@@ -25,37 +27,39 @@ services:
       - "3070"
     ports:
       - "3070:3070"
-    environment:
-      DAGSTER_POSTGRES_USER: ${DAGSTER_DATABASE_USERNAME}
-      DAGSTER_POSTGRES_PASSWORD: ${DAGSTER_DATABASE_PASSWORD}
-      DAGSTER_POSTGRES_DB: ${DAGSTER_DATABASE}
     volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
       - /tmp/io_manager_storage:/tmp/io_manager_storage
       - ./pipelines/home:/opt/dagster/dagster_home/
     networks:
       - dagster-internal
       - public
     depends_on:
       - dagster_postgresql
+    env_file:
+      - env/dagster.env
+      - env/pipeline.env
 
   dagster_daemon:
     build: dagster/dagster
     entrypoint:
       - dagster-daemon
       - run
-    container_name: dagster-daemon
+    container_name: dagster_daemon
+    hostname: dagster_daemon
     restart: on-failure
-    environment:
-      DAGSTER_POSTGRES_USER: ${DAGSTER_DATABASE_USERNAME}
-      DAGSTER_POSTGRES_PASSWORD: ${DAGSTER_DATABASE_PASSWORD}
-      DAGSTER_POSTGRES_DB: ${DAGSTER_DATABASE}
     volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
       - /tmp/io_manager_storage:/tmp/io_manager_storage
       - ./pipelines/home:/opt/dagster/dagster_home/
     networks:
       - dagster-internal
+      - public
     depends_on:
       - dagster_postgresql
+    env_file:
+      - env/dagster.env
+      - env/pipeline.env
 
 networks:
   dagster-internal:

dagster/compose.dagster.yml

@@ -12,13 +12,12 @@ RUN pip install \
 
 RUN apt update && apt upgrade -y && apt install default-jdk -y
 
+# Set $DAGSTER_HOME and copy dagster instance and workspace YAML there
 ENV PYSPARK_SUBMIT_ARGS="--master spark://spark:7077"
 ENV JAVA_HOME=/usr/lib/jvm/default-java/
 ENV DAGSTER_DIR=/var/lib/fxttr/dagster
 ENV DAGSTER_HOME=/opt/dagster/dagster_home/
 
 RUN mkdir -p ${DAGSTER_HOME} ${DAGSTER_DIR}
 
-COPY conf/dagster.yaml ${DAGSTER_HOME}/dagster.yaml
-
 WORKDIR $DAGSTER_HOME

dagster/dagster/Dockerfile

@@ -3,6 +3,7 @@ services:
     build: playground/
     container_name: playground
     volumes:
+      - ./data/stage:/var/lib/fxttr/stage
       - ./data/playground/notebooks:/var/lib/fxttr/notebooks
     ports:
       - '8888:8888'
@@ -17,6 +18,7 @@ services:
     build: trino/
     container_name: trino
     volumes:
+      - ./data/stage:/var/lib/fxttr/stage
       - ./trino/conf/trino.pass:/trino.pass
       - ./trino/conf/cert.pem:/cert.pem
     ports:
@@ -29,6 +31,9 @@ services:
       nofile:
         soft: 20000
         hard: 40000
+    env_file:
+      - env/clickhouse.env
+      - env/minio.env
 
 networks:
   public:

dagster/dagster/conf/dagster.yaml

@@ -0,0 +1,32 @@
+from pathlib import Path
+
+
+def __resolve_provider__() -> str:
+    """Determine the current provider"""
+    def __is_valid_command__(name):
+        """Check whether `name` is on PATH and marked as executable."""
+
+        from shutil import which
+
+        return which(name) is not None
+
+    if __is_valid_command__("talosctl"):
+        return "talosctl"
+        
+    if __is_valid_command__("microk8s"):
+        return "microk8s"
+    
+    raise Exception("Could not find any provider!")
+
+def __is_first_run__() -> bool:
+    lock_file = Path(__file__).joinpath("/../.dsadm_lockfile")
+    result = False
+
+    if not lock_file.exists():
+        result = True
+        lock_file.touch()
+
+    return result
+
+PROVIDER = __resolve_provider__()
+#IS_FIRST_RUN = __is_first_run__()

docker-compose.yml

@@ -0,0 +1,27 @@
+from ds_adm import PROVIDER
+from ds_adm.helm.chart import HelmChart
+from ds_adm.helm.plugin import HelmPlugin
+from ds_adm.helm.repository import HelmRepository
+from ds_adm.kube.operator import KubeOperator
+import yaml
+
+class Config(yaml.YAMLObject):
+    yaml_loader = yaml.SafeLoader
+    yaml_tag = u'!dsadm'
+
+    def __init__(self,
+                 repos: list[HelmRepository] = [], 
+                 plugins: list[HelmPlugin] = [],
+                 charts: list[HelmChart] = [],
+                 operators: list[KubeOperator] = []):
+        self.provider = PROVIDER
+        self.repos = repos
+        self.plugins = plugins
+        self.charts = charts
+        self.operators = operators
+
+def parse(path: str = "ds_config.yaml") -> Config:
+    with open(path, 'r') as file:
+        return yaml.safe_load(file)
+    
+

ds_adm/__init__.py

@@ -0,0 +1,14 @@
+from ds_adm.runtime import helm
+
+
+class HelmChart:
+    def __init__(self, name: str, namespace: str, repo: str, values: str):
+        self.name = name
+        self.namespace = namespace
+        self.repo = repo
+        self.values = values
+
+    def install(self):
+        print(f"🚀 Installing helm chart {self.name} in the {self.namespace} namespace...")
+        helm(f"upgrade --install {self.name} {self.repo} --namespace {self.namespace} -f k8s/values/{self.values}")
+

ds_adm/config.py

@@ -0,0 +1,17 @@
+from ds_adm.runtime import helm
+
+
+class HelmPlugin:
+    def __init__(self, name: str, url: str):
+        self.name = name
+        self.url = url
+
+    def __is_installed__(self):
+        helm(f"plugin list | grep -qw {self.name}").returncode == 0
+
+    def install(self, reinstall: bool = False):
+        if not reinstall and self.__is_installed__():
+            pass
+
+        print(f"📦 Adding helm plugin {self.name}...")
+        helm(f"plugin install {self.url}")

ds_adm/helm/chart.py

@@ -0,0 +1,19 @@
+from ds_adm.runtime import helm
+
+
+class HelmRepository:
+    def __init__(self, name: str, url: str):
+        self.name = name
+        self.url = url
+    
+    def __is_installed__(self):
+        helm(f"repo list | grep -qw {self.name}").returncode == 0
+
+    def install(self, reinstall: bool = False):
+        if not reinstall and self.__is_installed__():
+            pass
+
+        print(f"📦 Adding helm repository {self.name}...")
+        helm(f"repo add {self.name} {self.url}")
+        helm(f"repo update")
+

ds_adm/helm/plugin.py

@@ -0,0 +1,6 @@
+from ds_adm.runtime import kubectl
+
+
+def apply():
+    print("🔧 Adding config maps...")
+    kubectl("apply -f k8s/manifests/")