Add Job and secrets support for k8s-kind deployments (#995)

Part of https://plan.wireit.in/deepstack/browse/VUL-315

Reviewed-on: #995
Co-authored-by: Prathamesh Musale <prathamesh.musale0@gmail.com>
Co-committed-by: Prathamesh Musale <prathamesh.musale0@gmail.com>

stack_orchestrator/deploy/k8s/cluster_info.py

@@ -72,15 +72,17 @@ def to_k8s_resource_requirements(resources: Resources) -> client.V1ResourceRequi
 
 class ClusterInfo:
     parsed_pod_yaml_map: Any
+    parsed_job_yaml_map: Any
     image_set: Set[str] = set()
     app_name: str
+    stack_name: str
     environment_variables: DeployEnvVars
     spec: Spec
 
     def __init__(self) -> None:
-        pass
+        self.parsed_job_yaml_map = {}
 
-    def int(self, pod_files: List[str], compose_env_file, deployment_name, spec: Spec):
+    def int(self, pod_files: List[str], compose_env_file, deployment_name, spec: Spec, stack_name=""):
         self.parsed_pod_yaml_map = parsed_pod_files_map_from_file_names(pod_files)
         # Find the set of images in the pods
         self.image_set = images_for_deployment(pod_files)
@@ -90,10 +92,23 @@ class ClusterInfo:
         }
         self.environment_variables = DeployEnvVars(env_vars)
         self.app_name = deployment_name
+        self.stack_name = stack_name
         self.spec = spec
         if opts.o.debug:
             print(f"Env vars: {self.environment_variables.map}")
 
+    def init_jobs(self, job_files: List[str]):
+        """Initialize parsed job YAML map from job compose files."""
+        self.parsed_job_yaml_map = parsed_pod_files_map_from_file_names(job_files)
+        if opts.o.debug:
+            print(f"Parsed job yaml map: {self.parsed_job_yaml_map}")
+
+    def _all_named_volumes(self) -> list:
+        """Return named volumes from both pod and job compose files."""
+        volumes = named_volumes_from_pod_files(self.parsed_pod_yaml_map)
+        volumes.extend(named_volumes_from_pod_files(self.parsed_job_yaml_map))
+        return volumes
+
     def get_nodeports(self):
         nodeports = []
         for pod_name in self.parsed_pod_yaml_map:
@@ -257,7 +272,7 @@ class ClusterInfo:
     def get_pvcs(self):
         result = []
         spec_volumes = self.spec.get_volumes()
-        named_volumes = named_volumes_from_pod_files(self.parsed_pod_yaml_map)
+        named_volumes = self._all_named_volumes()
         resources = self.spec.get_volume_resources()
         if not resources:
             resources = DEFAULT_VOLUME_RESOURCES
@@ -301,7 +316,7 @@ class ClusterInfo:
     def get_configmaps(self):
         result = []
         spec_configmaps = self.spec.get_configmaps()
-        named_volumes = named_volumes_from_pod_files(self.parsed_pod_yaml_map)
+        named_volumes = self._all_named_volumes()
         for cfg_map_name, cfg_map_path in spec_configmaps.items():
             if cfg_map_name not in named_volumes:
                 if opts.o.debug:
@@ -337,7 +352,7 @@ class ClusterInfo:
     def get_pvs(self):
         result = []
         spec_volumes = self.spec.get_volumes()
-        named_volumes = named_volumes_from_pod_files(self.parsed_pod_yaml_map)
+        named_volumes = self._all_named_volumes()
         resources = self.spec.get_volume_resources()
         if not resources:
             resources = DEFAULT_VOLUME_RESOURCES
@@ -424,15 +439,25 @@ class ClusterInfo:
         # 3. Fall back to spec.yml global (already resolved with DEFAULT fallback)
         return global_resources
 
-    # TODO: put things like image pull policy into an object-scope struct
-    def get_deployment(self, image_pull_policy: Optional[str] = None):
+    def _build_containers(
+        self,
+        parsed_yaml_map: Any,
+        image_pull_policy: Optional[str] = None,
+    ) -> tuple:
+        """Build k8s container specs from parsed compose YAML.
+
+        Returns a tuple of (containers, services, volumes) where:
+        - containers: list of V1Container objects
+        - services: the last services dict processed (used for annotations/labels)
+        - volumes: list of V1Volume objects
+        """
         containers = []
         services = {}
         global_resources = self.spec.get_container_resources()
         if not global_resources:
             global_resources = DEFAULT_CONTAINER_RESOURCES
-        for pod_name in self.parsed_pod_yaml_map:
-            pod = self.parsed_pod_yaml_map[pod_name]
+        for pod_name in parsed_yaml_map:
+            pod = parsed_yaml_map[pod_name]
             services = pod["services"]
             for service_name in services:
                 container_name = service_name
@@ -489,7 +514,7 @@ class ClusterInfo:
                     else image
                 )
                 volume_mounts = volume_mounts_for_service(
-                    self.parsed_pod_yaml_map, service_name
+                    parsed_yaml_map, service_name
                 )
                 # Handle command/entrypoint from compose file
                 # In docker-compose: entrypoint -> k8s command, command -> k8s args
@@ -513,6 +538,16 @@ class ClusterInfo:
                         )
                     )
                 ]
+                # Mount user-declared secrets from spec.yml
+                for user_secret_name in self.spec.get_secrets():
+                    env_from.append(
+                        client.V1EnvFromSource(
+                            secret_ref=client.V1SecretEnvSource(
+                                name=user_secret_name,
+                                optional=True,
+                            )
+                        )
+                    )
                 container_resources = self._resolve_container_resources(
                     container_name, service_info, global_resources
                 )
@@ -538,7 +573,14 @@ class ClusterInfo:
                 )
                 containers.append(container)
         volumes = volumes_for_pod_files(
-            self.parsed_pod_yaml_map, self.spec, self.app_name
+            parsed_yaml_map, self.spec, self.app_name
+        )
+        return containers, services, volumes
+
+    # TODO: put things like image pull policy into an object-scope struct
+    def get_deployment(self, image_pull_policy: Optional[str] = None):
+        containers, services, volumes = self._build_containers(
+            self.parsed_pod_yaml_map, image_pull_policy
         )
         registry_config = self.spec.get_image_registry_config()
         if registry_config:
@@ -549,6 +591,8 @@ class ClusterInfo:
 
         annotations = None
         labels = {"app": self.app_name}
+        if self.stack_name:
+            labels["app.kubernetes.io/stack"] = self.stack_name
         affinity = None
         tolerations = None
 
@@ -628,3 +672,75 @@ class ClusterInfo:
             spec=spec,
         )
         return deployment
+
+    def get_jobs(self, image_pull_policy: Optional[str] = None) -> List[client.V1Job]:
+        """Build k8s Job objects from parsed job compose files.
+
+        Each job compose file produces a V1Job with:
+        - restartPolicy: Never
+        - backoffLimit: 0
+        - Name: {app_name}-job-{job_name}
+        """
+        if not self.parsed_job_yaml_map:
+            return []
+
+        jobs = []
+        registry_config = self.spec.get_image_registry_config()
+        if registry_config:
+            secret_name = f"{self.app_name}-registry"
+            image_pull_secrets = [client.V1LocalObjectReference(name=secret_name)]
+        else:
+            image_pull_secrets = []
+
+        for job_file in self.parsed_job_yaml_map:
+            # Build containers for this single job file
+            single_job_map = {job_file: self.parsed_job_yaml_map[job_file]}
+            containers, _services, volumes = self._build_containers(
+                single_job_map, image_pull_policy
+            )
+
+            # Derive job name from file path: docker-compose-<name>.yml -> <name>
+            base = os.path.basename(job_file)
+            # Strip docker-compose- prefix and .yml suffix
+            job_name = base
+            if job_name.startswith("docker-compose-"):
+                job_name = job_name[len("docker-compose-"):]
+            if job_name.endswith(".yml"):
+                job_name = job_name[: -len(".yml")]
+            elif job_name.endswith(".yaml"):
+                job_name = job_name[: -len(".yaml")]
+
+            # Use a distinct app label for job pods so they don't get
+            # picked up by pods_in_deployment() which queries app={app_name}.
+            pod_labels = {
+                "app": f"{self.app_name}-job",
+                **({"app.kubernetes.io/stack": self.stack_name} if self.stack_name else {}),
+            }
+            template = client.V1PodTemplateSpec(
+                metadata=client.V1ObjectMeta(
+                    labels=pod_labels
+                ),
+                spec=client.V1PodSpec(
+                    containers=containers,
+                    image_pull_secrets=image_pull_secrets,
+                    volumes=volumes,
+                    restart_policy="Never",
+                ),
+            )
+            job_spec = client.V1JobSpec(
+                template=template,
+                backoff_limit=0,
+            )
+            job_labels = {"app": self.app_name, **({"app.kubernetes.io/stack": self.stack_name} if self.stack_name else {})}
+            job = client.V1Job(
+                api_version="batch/v1",
+                kind="Job",
+                metadata=client.V1ObjectMeta(
+                    name=f"{self.app_name}-job-{job_name}",
+                    labels=job_labels,
+                ),
+                spec=job_spec,
+            )
+            jobs.append(job)
+
+        return jobs

stack_orchestrator/deploy/k8s/deploy_k8s.py

@@ -95,6 +95,7 @@ class K8sDeployer(Deployer):
     type: str
     core_api: client.CoreV1Api
     apps_api: client.AppsV1Api
+    batch_api: client.BatchV1Api
     networking_api: client.NetworkingV1Api
     k8s_namespace: str
     kind_cluster_name: str
@@ -110,6 +111,7 @@ class K8sDeployer(Deployer):
         compose_files,
         compose_project_name,
         compose_env_file,
+        job_compose_files=None,
     ) -> None:
         self.type = type
         self.skip_cluster_management = False
@@ -124,15 +126,24 @@ class K8sDeployer(Deployer):
         # Use deployment-specific namespace for resource isolation and easy cleanup
         self.k8s_namespace = f"laconic-{compose_project_name}"
         self.cluster_info = ClusterInfo()
+        # stack.name may be an absolute path (from spec "stack:" key after
+        # path resolution). Extract just the directory basename for labels.
+        raw_name = deployment_context.stack.name if deployment_context else ""
+        stack_name = Path(raw_name).name if raw_name else ""
         self.cluster_info.int(
             compose_files,
             compose_env_file,
             compose_project_name,
             deployment_context.spec,
+            stack_name=stack_name,
         )
+        # Initialize job compose files if provided
+        if job_compose_files:
+            self.cluster_info.init_jobs(job_compose_files)
         if opts.o.debug:
             print(f"Deployment dir: {deployment_context.deployment_dir}")
             print(f"Compose files: {compose_files}")
+            print(f"Job compose files: {job_compose_files}")
             print(f"Project name: {compose_project_name}")
             print(f"Env file: {compose_env_file}")
             print(f"Type: {type}")
@@ -150,6 +161,7 @@ class K8sDeployer(Deployer):
         self.core_api = client.CoreV1Api()
         self.networking_api = client.NetworkingV1Api()
         self.apps_api = client.AppsV1Api()
+        self.batch_api = client.BatchV1Api()
         self.custom_obj_api = client.CustomObjectsApi()
 
     def _ensure_namespace(self):
@@ -256,6 +268,11 @@ class K8sDeployer(Deployer):
                     print(f"{cfg_rsp}")
 
     def _create_deployment(self):
+        # Skip if there are no pods to deploy (e.g. jobs-only stacks)
+        if not self.cluster_info.parsed_pod_yaml_map:
+            if opts.o.debug:
+                print("No pods defined, skipping Deployment creation")
+            return
         # Process compose files into a Deployment
         deployment = self.cluster_info.get_deployment(
             image_pull_policy=None if self.is_kind() else "Always"
@@ -293,6 +310,26 @@ class K8sDeployer(Deployer):
                 print("Service created:")
                 print(f"{service_resp}")
 
+    def _create_jobs(self):
+        # Process job compose files into k8s Jobs
+        jobs = self.cluster_info.get_jobs(
+            image_pull_policy=None if self.is_kind() else "Always"
+        )
+        for job in jobs:
+            if opts.o.debug:
+                print(f"Sending this job: {job}")
+            if not opts.o.dry_run:
+                job_resp = self.batch_api.create_namespaced_job(
+                    body=job, namespace=self.k8s_namespace
+                )
+                if opts.o.debug:
+                    print("Job created:")
+                    if job_resp.metadata:
+                        print(
+                            f"  {job_resp.metadata.namespace} "
+                            f"{job_resp.metadata.name}"
+                        )
+
     def _find_certificate_for_host_name(self, host_name):
         all_certificates = self.custom_obj_api.list_namespaced_custom_object(
             group="cert-manager.io",
@@ -384,6 +421,7 @@ class K8sDeployer(Deployer):
 
         self._create_volume_data()
         self._create_deployment()
+        self._create_jobs()
 
         http_proxy_info = self.cluster_info.spec.get_http_proxy()
         # Note: we don't support tls for kind (enabling tls causes errors)
@@ -426,6 +464,11 @@ class K8sDeployer(Deployer):
                     print("NodePort created:")
                     print(f"{nodeport_resp}")
 
+        # Call start() hooks — stacks can create additional k8s resources
+        if self.deployment_context:
+            from stack_orchestrator.deploy.deployment_create import call_stack_deploy_start
+            call_stack_deploy_start(self.deployment_context)
+
     def down(self, timeout, volumes, skip_cluster_management):
         self.skip_cluster_management = skip_cluster_management
         self.connect_api()
@@ -574,14 +617,14 @@ class K8sDeployer(Deployer):
 
     def logs(self, services, tail, follow, stream):
         self.connect_api()
-        pods = pods_in_deployment(self.core_api, self.cluster_info.app_name)
+        pods = pods_in_deployment(self.core_api, self.cluster_info.app_name, namespace=self.k8s_namespace)
         if len(pods) > 1:
             print("Warning: more than one pod in the deployment")
         if len(pods) == 0:
             log_data = "******* Pods not running ********\n"
         else:
             k8s_pod_name = pods[0]
-            containers = containers_in_pod(self.core_api, k8s_pod_name)
+            containers = containers_in_pod(self.core_api, k8s_pod_name, namespace=self.k8s_namespace)
             # If pod not started, logs request below will throw an exception
             try:
                 log_data = ""
@@ -599,6 +642,10 @@ class K8sDeployer(Deployer):
         return log_stream_from_string(log_data)
 
     def update(self):
+        if not self.cluster_info.parsed_pod_yaml_map:
+            if opts.o.debug:
+                print("No pods defined, skipping update")
+            return
         self.connect_api()
         ref_deployment = self.cluster_info.get_deployment()
         if not ref_deployment or not ref_deployment.metadata:
@@ -659,26 +706,43 @@ class K8sDeployer(Deployer):
 
     def run_job(self, job_name: str, helm_release: Optional[str] = None):
         if not opts.o.dry_run:
-            from stack_orchestrator.deploy.k8s.helm.job_runner import run_helm_job
-
             # Check if this is a helm-based deployment
             chart_dir = self.deployment_dir / "chart"
-            if not chart_dir.exists():
-                # TODO: Implement job support for compose-based K8s deployments
-                raise Exception(
-                    f"Job support is only available for helm-based "
-                    f"deployments. Chart directory not found: {chart_dir}"
-                )
+            if chart_dir.exists():
+                from stack_orchestrator.deploy.k8s.helm.job_runner import run_helm_job
 
-            # Run the job using the helm job runner
-            run_helm_job(
-                chart_dir=chart_dir,
-                job_name=job_name,
-                release=helm_release,
-                namespace=self.k8s_namespace,
-                timeout=600,
-                verbose=opts.o.verbose,
-            )
+                # Run the job using the helm job runner
+                run_helm_job(
+                    chart_dir=chart_dir,
+                    job_name=job_name,
+                    release=helm_release,
+                    namespace=self.k8s_namespace,
+                    timeout=600,
+                    verbose=opts.o.verbose,
+                )
+            else:
+                # Non-Helm path: create job from ClusterInfo
+                self.connect_api()
+                jobs = self.cluster_info.get_jobs(
+                    image_pull_policy=None if self.is_kind() else "Always"
+                )
+                # Find the matching job by name
+                target_name = f"{self.cluster_info.app_name}-job-{job_name}"
+                matched_job = None
+                for job in jobs:
+                    if job.metadata and job.metadata.name == target_name:
+                        matched_job = job
+                        break
+                if matched_job is None:
+                    raise Exception(
+                        f"Job '{job_name}' not found. Available jobs: "
+                        f"{[j.metadata.name for j in jobs if j.metadata]}"
+                    )
+                if opts.o.debug:
+                    print(f"Creating job: {target_name}")
+                self.batch_api.create_namespaced_job(
+                    body=matched_job, namespace=self.k8s_namespace
+                )
 
     def is_kind(self):
         return self.type == "k8s-kind"

stack_orchestrator/deploy/k8s/helpers.py

@@ -393,10 +393,10 @@ def load_images_into_kind(kind_cluster_name: str, image_set: Set[str]):
             raise DeployerException(f"kind load docker-image failed: {result}")
 
 
-def pods_in_deployment(core_api: client.CoreV1Api, deployment_name: str):
+def pods_in_deployment(core_api: client.CoreV1Api, deployment_name: str, namespace: str = "default"):
     pods = []
     pod_response = core_api.list_namespaced_pod(
-        namespace="default", label_selector=f"app={deployment_name}"
+        namespace=namespace, label_selector=f"app={deployment_name}"
     )
     if opts.o.debug:
         print(f"pod_response: {pod_response}")
@@ -406,10 +406,10 @@ def pods_in_deployment(core_api: client.CoreV1Api, deployment_name: str):
     return pods
 
 
-def containers_in_pod(core_api: client.CoreV1Api, pod_name: str) -> List[str]:
+def containers_in_pod(core_api: client.CoreV1Api, pod_name: str, namespace: str = "default") -> List[str]:
     containers: List[str] = []
     pod_response = cast(
-        client.V1Pod, core_api.read_namespaced_pod(pod_name, namespace="default")
+        client.V1Pod, core_api.read_namespaced_pod(pod_name, namespace=namespace)
     )
     if opts.o.debug:
         print(f"pod_response: {pod_response}")