Always build pipeline image for Kubernetes orchestrator

src/zenml/integrations/kubernetes/flavors/kubernetes_orchestrator_flavor.py

@@ -69,6 +69,8 @@ class KubernetesOrchestratorSettings(BaseSettings):
             scheduling a pipeline.
         prevent_orchestrator_pod_caching: If `True`, the orchestrator pod will
             not try to compute cached steps before starting the step pods.
+        always_build_pipeline_image: If `True`, the orchestrator will always
+            build the pipeline image, even if all steps have a custom build.
     """
 
     synchronous: bool = True
@@ -88,6 +90,7 @@ class KubernetesOrchestratorSettings(BaseSettings):
     failed_jobs_history_limit: Optional[NonNegativeInt] = None
     ttl_seconds_after_finished: Optional[NonNegativeInt] = None
     prevent_orchestrator_pod_caching: bool = False
+    always_build_pipeline_image: bool = False
 
 
 class KubernetesOrchestratorConfig(

src/zenml/integrations/kubernetes/orchestrators/kubernetes_orchestrator.py

@@ -70,7 +70,11 @@
 from zenml.stack import StackValidator
 
 if TYPE_CHECKING:
-    from zenml.models import PipelineDeploymentResponse, PipelineRunResponse
+    from zenml.models import (
+        PipelineDeploymentBase,
+        PipelineDeploymentResponse,
+        PipelineRunResponse,
+    )
     from zenml.stack import Stack
 
 logger = get_logger(__name__)
@@ -84,6 +88,19 @@ class KubernetesOrchestrator(ContainerizedOrchestrator):
 
     _k8s_client: Optional[k8s_client.ApiClient] = None
 
+    def should_build_pipeline_image(
+        self, deployment: "PipelineDeploymentBase"
+    ) -> bool:
+        """Whether to always build the pipeline image.
+
+        Returns:
+            Whether to always build the pipeline image.
+        """
+        settings = cast(
+            KubernetesOrchestratorSettings, self.get_settings(deployment)
+        )
+        return settings.always_build_pipeline_image
+
     def get_kube_client(
         self, incluster: Optional[bool] = None
     ) -> k8s_client.ApiClient:

src/zenml/orchestrators/containerized_orchestrator.py

@@ -53,6 +53,16 @@ def get_image(
             component_key=ORCHESTRATOR_DOCKER_IMAGE_KEY, step=step_name
         )
 
+    def should_build_pipeline_image(
+        self, deployment: "PipelineDeploymentBase"
+    ) -> bool:
+        """Whether to build the pipeline image.
+
+        Returns:
+            Whether to build the pipeline image.
+        """
+        return False
+
     def get_docker_builds(
         self, deployment: "PipelineDeploymentBase"
     ) -> List["BuildConfiguration"]:
@@ -87,4 +97,13 @@ def get_docker_builds(
                 builds.append(pipeline_build)
                 included_pipeline_build = True
 
+        if not included_pipeline_build and self.should_build_pipeline_image(
+            deployment
+        ):
+            pipeline_build = BuildConfiguration(
+                key=ORCHESTRATOR_DOCKER_IMAGE_KEY,
+                settings=pipeline_settings,
+            )
+            builds.append(pipeline_build)
+
         return builds