Allow skypilot to configure step or run full pipeline in one VM

docs/book/stacks-and-components/component-guide/orchestrators/skypilot-vm.md

@@ -40,6 +40,12 @@ on-demand and managed spot VMs. While you can select the VM type you want to use
 also includes an optimizer that automatically selects the cheapest VM/zone/region/cloud for your workloads.
 Finally, the orchestrator includes an autostop feature that cleans up idle clusters, preventing unnecessary cloud costs.
 
+{% hint style="info" %}
+You can configure the SkyPilot VM Orchestrator to use a specific VM type, and 
+resources for each step of your pipeline can be configured individually.
+Read more about how to configure step-specific resources [here](#configuring-step-specific-resources).
+{% endhint %}
+
 {% hint style="warning" %}
 The SkyPilot VM Orchestrator does not currently support the ability to [schedule pipelines runs](/docs/book/user-guide/advanced-guide/pipelining-features/schedule-pipeline-runs.md)
 {% endhint %}
@@ -378,7 +384,7 @@ One of the key features of the SkyPilot VM Orchestrator is the ability to run ea
 
 The SkyPilot VM Orchestrator allows you to configure resources for each step individually. This means you can specify different VM types, CPU and memory requirements, and even use spot instances for certain steps while using on-demand instances for others.
 
-To configure step-specific resources, you can pass a `SkypilotBaseOrchestratorSettings` object to the `settings` parameter of the `@step` decorator. This object allows you to define various attributes such as `instance_type`, `cpus`, `memory`, `use_spot`, `region`, and more.
+In order to enable this, you will need to update your orchestrator configuration to use `configure_step_resources=True` or create a new orchestrator with this setting enabled. This setting allows the orchestrator to configure resources for each step individually. Once your orchestrator is configured to allow step-specific resources, you can pass a `SkypilotBaseOrchestratorSettings` object to the `settings` parameter of the `@step` decorator. This object allows you to define various attributes such as `instance_type`, `cpus`, `memory`, `use_spot`, `region`, and more.
 
 Here's an example of how to configure specific resources for a step for the AWS cloud:
 

src/zenml/integrations/skypilot/flavors/skypilot_orchestrator_base_vm_config.py

@@ -113,7 +113,17 @@ class SkypilotBaseOrchestratorSettings(BaseSettings):
 class SkypilotBaseOrchestratorConfig(  # type: ignore[misc] # https://github.com/pydantic/pydantic/issues/4173
     BaseOrchestratorConfig, SkypilotBaseOrchestratorSettings
 ):
-    """Skypilot orchestrator base config."""
+    """Skypilot orchestrator base config.
+
+    Attributes:
+        configure_step_resources: Enables the orchestrator to run configured steps.
+            This will be used to determine whether to run the entire pipeline
+            in one single VM or to run each step in a separate VM if the
+            orchestrator is configured to run steps separately with some custom
+            resources.
+    """
+
+    configure_step_resources: bool = False
 
     @property
     def is_local(self) -> bool:

src/zenml/integrations/skypilot/orchestrators/skypilot_base_vm_orchestrator.py

@@ -20,9 +20,11 @@
 
 import sky
 
+from zenml.entrypoints import PipelineEntrypointConfiguration
 from zenml.enums import StackComponentType
 from zenml.environment import Environment
 from zenml.integrations.skypilot.flavors.skypilot_orchestrator_base_vm_config import (
+    SkypilotBaseOrchestratorConfig,
     SkypilotBaseOrchestratorSettings,
 )
 from zenml.integrations.skypilot.orchestrators.skypilot_orchestrator_entrypoint_configuration import (
@@ -109,6 +111,15 @@ def get_orchestrator_run_id(self) -> str:
                 f"{ENV_ZENML_SKYPILOT_ORCHESTRATOR_RUN_ID}."
             )
 
+    @property
+    def config(self) -> SkypilotBaseOrchestratorConfig:
+        """Returns the `SkypilotBaseOrchestratorConfig` config.
+
+        Returns:
+            The configuration.
+        """
+        return cast(SkypilotBaseOrchestratorConfig, self._config)
+
     @property
     @abstractmethod
     def cloud(self) -> sky.clouds.Cloud:
@@ -189,14 +200,21 @@ def prepare_or_run_pipeline(
                 deployment=deployment, step_name=pipeline_step_name
             )
 
-        # Build entrypoint command and args for the orchestrator pod.
-        # This will internally also build the command/args for all step pods.
-        command = SkypilotOrchestratorEntrypointConfiguration.get_entrypoint_command()
+        if self.config.configure_step_resources:
+            # Run each step in a separate VM if configured.
+            # Build entrypoint command and args for the orchestrator VM.
+            # This will internally also build the command/args for all step pods.
+            command = SkypilotOrchestratorEntrypointConfiguration.get_entrypoint_command()
+            args = SkypilotOrchestratorEntrypointConfiguration.get_entrypoint_arguments(
+                run_name=orchestrator_run_name,
+                deployment_id=deployment.id,
+            )
+        else:
+            command = PipelineEntrypointConfiguration.get_entrypoint_command()
+            args = PipelineEntrypointConfiguration.get_entrypoint_arguments(
+                deployment_id=deployment.id
+            )
         entrypoint_str = " ".join(command)
-        args = SkypilotOrchestratorEntrypointConfiguration.get_entrypoint_arguments(
-            run_name=orchestrator_run_name,
-            deployment_id=deployment.id,
-        )
         arguments_str = " ".join(args)
 
         docker_environment_str = " ".join(