Repository: skypilot-org/skypilot
Pull request: Introduce High Availability Service Controller
<!-- Describe the changes in this PR -->

Resolves #4563 

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [x] Code formatting: `bash format.sh`
- [x] Any manual or new tests for this PR (please specify below)
- [ ] All smoke tests: `pytest tests/test_smoke.py` 
- [ ] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_fill_in_the_name` 
- [ ] Backward compatibility tests: `conda deactivate; bash -i tests/backward_compatibility_tests.sh`


Problem statement:
[Serve] Make SkyServe Controller More Persistent
# Design Docs: Persistent Service in SkyPilot

## Overview
The Persistent Service feature aims to enhance SkyServe's reliability by enabling service recovery after controller failures or system shutdowns. This design addresses how to restore services with their previous states preserved, making services more resilient to interruptions and allowing for planned downtimes.

### Service Recovery Flow

1. Service Recovery Trigger:
    - Primary Option (& our current implementation): Kubernetes deployment with automatic recovery via pod initialization
    - Cloud Provider Option: Cloud-native serverless functions (AWS Lambda, Azure Functions, Google Cloud Functions) for monitoring and auto-recovery
    - Fallback Option (*TBD*): Reuse sky serve up command for manual recovery in unsupported environments with `FAILED_CONTROLLER` state

2. When the controller is launched:
Check database states whether the service exists to realize that it’s on recover mode.

3. Recovery Process:
   - Skip service registration
   - Load Controller and LoadBalancer ports from `serve_state`
       - LoadBalancer ports should definitely be unchanged
   - Restart **Controller** (seems stateless already)
   - Load **AutoScaler** state from database
	 - ~~`request_timestamps`~~
   - Reconstruct **LoadBalancer** with persisted replica URLs synced from Controller
	 - ~~`client_pool` (TBD, or we can also wait for syncing from Controller, which takes longer to recover to previous states)~~
	 - ~~`_request_aggregator` (TBD, or just ignore it, since it flushes each time syncing with Controller)~~
   - Restore **ReplicaManager** process pools from database
	 - `launch_process_pool` and `down_process_pool`
	 - ~~We can possibly check if they become zombies.~~
	 - We simply store launching cluster, and start a new launch when recovery, since our `sky launch` and `sky down` are already robust enough

### Future Work

The system should consider recovery from broken operations, such as suddenly terminating an ongoing `update_service`. That's negligible, but would be more robust on thread safety if we implement this.

We should save unfinished requests, so that we can print something like `Unfinished requests: …` when recovering, providing a nice add-on when Controller and LoadBalancer are separated, since the unfinished http requests would not be interrupted when Controller failed, causing some confusion to users.

---

This design provides a foundation for making SkyServe services more resilient while enabling users to manage service life cycles more effectively, particularly for cost optimization during off-hours.

Hints from issue discussion:

Commit message: refactor: common util judging if controller by cluster name

Patch to review:
diff --git a/docs/source/reference/config.rst b/docs/source/reference/config.rst
index a76dc473206..a18ea7ef380 100644
--- a/docs/source/reference/config.rst
+++ b/docs/source/reference/config.rst
@@ -14,6 +14,8 @@ Such configurations apply to all new clusters and do not affect existing cluster
 Spec: ``~/.sky/config.yaml``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
+For details about SkyServe controller and its customization, see :ref:`customizing-sky-serve-controller-resources`.
+
 Available fields and semantics:
 
 .. code-block:: yaml
diff --git a/docs/source/serving/sky-serve.rst b/docs/source/serving/sky-serve.rst
index 9e8ab0e4a61..93fbe2fe0b3 100644
--- a/docs/source/serving/sky-serve.rst
+++ b/docs/source/serving/sky-serve.rst
@@ -508,6 +508,22 @@ To achieve the above, you can specify custom configs in :code:`~/.sky/config.yam
     # NOTE: these settings only take effect for a new SkyServe controller, not if
     # you have an existing one.
     controller:
+      # Enable high availability mode for the controller (optional).
+      #
+      # When set to true, the controller will be deployed with high availability
+      # capabilities on Kubernetes using Deployments. This allows the controller and load balancer
+      # to automatically recover from failures (e.g., node failures, pod crashes)
+      # and maintain service continuity.
+      #
+      # NOTE: This feature is ONLY supported when Kubernetes cloud is enabled. To enable kubernetes, see :ref:`Kubernetes Setup <kubernetes-setup>`.
+      # The service controller cluster will be scheduled to kubernetes. The k8s deployment needs to be always-on to keep the controller running. If using a local kubernetes deployment (e.g. `sky local up`), keeping the laptop/machines up is required.
+      # When enabled:
+      # - The controller is deployed as a Kubernetes Deployment instead of a Pod
+      # - Automatic pod rescheduling and recovery is handled by Kubernetes
+      #
+      # Default: false.
+      high_availability: true
+
       resources:
         # All configs below are optional.
         # Specify the location of the SkyServe controller.
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 0b6f06254d0..0ccc8992094 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -2026,6 +2026,8 @@ def provision_with_retries(
                 # Recheck cluster name as the 'except:' block below may
                 # change the cloud assignment.
                 common_utils.check_cluster_name_is_valid(cluster_name)
+
+                assert to_provision.cloud is not None
                 if dryrun:
                     cloud_user = None
                 else:
@@ -2452,6 +2454,11 @@ def get_command_runners(self,
                     'Tried to use cached cluster info, but it\'s missing for '
                     f'cluster "{self.cluster_name}"')
             self._update_cluster_info()
+        # TODO(andyl): must force refresh here?
+        if common_utils.is_cluster_name_indicating_serve_controller(
+                self.cluster_name):
+            self._update_cluster_info()
+
         assert self.cached_cluster_info is not None, self
         runners = provision_lib.get_command_runners(
             self.cached_cluster_info.provider_name, self.cached_cluster_info,
@@ -2659,6 +2666,7 @@ def register_info(self, **kwargs) -> None:
             self._optimize_target) or optimizer.OptimizeTarget.COST
         self._requested_features = kwargs.pop('requested_features',
                                               self._requested_features)
+        self._dump_final_script = kwargs.pop('dump_final_script', False)
         assert not kwargs, f'Unexpected kwargs: {kwargs}'
 
     def check_resources_fit_cluster(
@@ -3233,23 +3241,29 @@ def _setup_node(node_id: int) -> None:
                                                          env_vars=setup_envs)
             encoded_script = shlex.quote(setup_script)
 
-            def _dump_setup_script(setup_script: str) -> None:
+            def _dump_final_script(
+                    setup_script: str,
+                    target_dir: str = remote_setup_file_name) -> None:
                 with tempfile.NamedTemporaryFile('w', prefix='sky_setup_') as f:
                     f.write(setup_script)
                     f.flush()
                     setup_sh_path = f.name
                     runner.rsync(source=setup_sh_path,
-                                 target=remote_setup_file_name,
+                                 target=target_dir,
                                  up=True,
                                  stream_logs=False)
 
             if detach_setup or _is_command_length_over_limit(encoded_script):
-                _dump_setup_script(setup_script)
+                _dump_final_script(setup_script)
                 create_script_code = 'true'
             else:
                 create_script_code = (f'{{ echo {encoded_script} > '
                                       f'{remote_setup_file_name}; }}')
 
+            if self._dump_final_script:
+                _dump_final_script(setup_script,
+                                   constants.PERSISTENT_SETUP_SCRIPT_PATH)
+
             if detach_setup:
                 return
 
@@ -3296,7 +3310,7 @@ def _run_setup(setup_cmd: str) -> int:
                         'Failed to run setup command inline due to '
                         'command length limit. Dumping setup script to '
                         'file and running it with SSH.')
-                    _dump_setup_script(setup_script)
+                    _dump_final_script(setup_script)
                     returncode = _run_setup(setup_cmd)
 
             def error_message() -> str:
@@ -3390,14 +3404,14 @@ def _exec_code_on_head(
         code = job_lib.JobLibCodeGen.queue_job(job_id, job_submit_cmd)
         job_submit_cmd = ' && '.join([mkdir_code, create_script_code, code])
 
-        def _dump_code_to_file(codegen: str) -> None:
+        def _dump_code_to_file(codegen: str,
+                               target_dir: str = SKY_REMOTE_APP_DIR) -> None:
             runners = handle.get_command_runners()
             head_runner = runners[0]
             with tempfile.NamedTemporaryFile('w', prefix='sky_app_') as fp:
                 fp.write(codegen)
                 fp.flush()
-                script_path = os.path.join(SKY_REMOTE_APP_DIR,
-                                           f'sky_job_{job_id}')
+                script_path = os.path.join(target_dir, f'sky_job_{job_id}')
                 # We choose to sync code + exec, because the alternative of 'ray
                 # submit' may not work as it may use system python (python2) to
                 # execute the script. Happens for AWS.
@@ -3423,6 +3437,10 @@ def _dump_code_to_file(codegen: str) -> None:
             # commands to be scheduled on the job controller in high-load cases.
             job_submit_cmd = job_submit_cmd + ' && ' + managed_job_code
 
+        if self._dump_final_script:
+            _dump_code_to_file(job_submit_cmd,
+                               constants.PERSISTENT_RUN_SCRIPT_DIR)
+
         returncode, stdout, stderr = self.run_on_head(handle,
                                                       job_submit_cmd,
                                                       stream_logs=False,
diff --git a/sky/backends/local_docker_backend.py b/sky/backends/local_docker_backend.py
index c10e51e7975..875820ae241 100644
--- a/sky/backends/local_docker_backend.py
+++ b/sky/backends/local_docker_backend.py
@@ -276,7 +276,6 @@ def _execute(self,
                  detach_run: bool,
                  dryrun: bool = False) -> None:
         """ Launches the container."""
-
         if detach_run:
             raise NotImplementedError('detach_run=True is not supported in '
                                       'LocalDockerBackend.')
@@ -364,7 +363,7 @@ def _update_state(self):
                 if k.startswith(_DOCKER_LABEL_PREFIX):
                     # Remove 'skymeta_' from key
                     metadata[k[len(_DOCKER_LABEL_PREFIX):]] = v
-            self.images[c.name] = [c.image, metadata]
+            self.images[c.name] = (c.image, metadata)
             self.containers[c.name] = c
 
     def _execute_task_one_node(self, handle: LocalDockerResourceHandle,
diff --git a/sky/cli.py b/sky/cli.py
index 8ce3051e7df..cfb6fc757c4 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -1516,7 +1516,10 @@ def _status_kubernetes(show_all: bool):
                    f'{colorama.Style.RESET_ALL}')
         msg = managed_jobs.format_job_table(all_jobs, show_all=show_all)
         click.echo(msg)
-    if any(['sky-serve-controller' in c.cluster_name for c in all_clusters]):
+    if any([
+            common_utils.is_cluster_name_indicating_serve_controller(
+                c.cluster_name) for c in all_clusters
+    ]):
         # TODO: Parse serve controllers and show services separately.
         #  Currently we show a hint that services are shown as clusters.
         click.echo(f'\n{colorama.Style.DIM}Hint: SkyServe replica pods are '
diff --git a/sky/clouds/cloud.py b/sky/clouds/cloud.py
index 2cb45ca14fc..99168bb990a 100644
--- a/sky/clouds/cloud.py
+++ b/sky/clouds/cloud.py
@@ -45,6 +45,8 @@ class CloudImplementationFeatures(enum.Enum):
     OPEN_PORTS = 'open_ports'
     STORAGE_MOUNTING = 'storage_mounting'
     HOST_CONTROLLERS = 'host_controllers'  # Can run jobs/serve controllers
+    HIGH_AVAILABILITY_CONTROLLERS = ('high_availability_controllers'
+                                    )  # Controller can auto-restart
     AUTO_TERMINATE = 'auto_terminate'  # Pod/VM can stop or down itself
 
 
@@ -605,6 +607,15 @@ def check_features_are_supported(
         unsupported_features2reason = cls._unsupported_features_for_resources(
             resources)
 
+        # pylint: disable=import-outside-toplevel
+        from sky.clouds import kubernetes
+        if cls is not kubernetes.Kubernetes:
+            unsupported_features2reason.update({
+                CloudImplementationFeatures.HIGH_AVAILABILITY_CONTROLLERS:
+                    ('Currently, for all clouds except Kubernetes, '
+                     'high availability controllers are not supported.')
+            })
+
         # Docker image is not compatible with ssh proxy command.
         if skypilot_config.get_nested(
             (str(cls._REPR).lower(), 'ssh_proxy_command'), None) is not None:
diff --git a/sky/clouds/kubernetes.py b/sky/clouds/kubernetes.py
index 3bab7687f48..cbef9ca041b 100644
--- a/sky/clouds/kubernetes.py
+++ b/sky/clouds/kubernetes.py
@@ -357,22 +357,26 @@ def make_deploy_resources_variables(
         acc_count = k.accelerator_count if k.accelerator_count else 0
         acc_type = k.accelerator_type if k.accelerator_type else None
 
-        image_id_dict = resources.image_id
-        if image_id_dict is not None:
-            # Use custom image specified in resources
-            if None in image_id_dict:
-                image_id = image_id_dict[None]
+        def _get_image_id(resources: 'resources_lib.Resources') -> str:
+            image_id_dict = resources.image_id
+            if image_id_dict is not None:
+                # Use custom image specified in resources
+                if None in image_id_dict:
+                    image_id = image_id_dict[None]
+                else:
+                    assert resources.region in image_id_dict, image_id_dict
+                    image_id = image_id_dict[resources.region]
+                if image_id.startswith('docker:'):
+                    image_id = image_id[len('docker:'):]
             else:
-                assert resources.region in image_id_dict, image_id_dict
-                image_id = image_id_dict[resources.region]
-            if image_id.startswith('docker:'):
-                image_id = image_id[len('docker:'):]
-        else:
-            # Select image based on whether we are using GPUs or not.
-            image_id = self.IMAGE_GPU if acc_count > 0 else self.IMAGE_CPU
-            # Get the container image ID from the service catalog.
-            image_id = service_catalog.get_image_id_from_tag(
-                image_id, clouds='kubernetes')
+                # Select image based on whether we are using GPUs or not.
+                image_id = self.IMAGE_GPU if acc_count > 0 else self.IMAGE_CPU
+                # Get the container image ID from the service catalog.
+                image_id = service_catalog.get_image_id_from_tag(
+                    image_id, clouds='kubernetes')
+            return image_id
+
+        image_id = _get_image_id(resources)
         # TODO(romilb): Create a lightweight image for SSH jump host
         ssh_jump_image = service_catalog.get_image_id_from_tag(
             self.IMAGE_CPU, clouds='kubernetes')
diff --git a/sky/execution.py b/sky/execution.py
index bb8ba725314..2e09cbd2556 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -200,9 +200,19 @@ def _execute(
     # Requested features that some clouds support and others don't.
     requested_features = set()
 
-    if controller_utils.Controllers.from_name(cluster_name) is not None:
+    is_controller_high_availability_supported = False
+
+    controller = controller_utils.Controllers.from_name(cluster_name)
+    if controller is not None:
         requested_features.add(
             clouds.CloudImplementationFeatures.HOST_CONTROLLERS)
+        if controller_utils.get_controller_high_availability_supported(
+                controller):
+            requested_features.add(clouds.CloudImplementationFeatures.
+                                   HIGH_AVAILABILITY_CONTROLLERS)
+            # If we provision a cluster that supports high availability
+            # controllers, we can use the high availability controller.
+            is_controller_high_availability_supported = True
 
     # Add requested features from the task
     requested_features |= task.get_required_cloud_features()
@@ -285,9 +295,14 @@ def _execute(
                     task = dag.tasks[0]  # Keep: dag may have been deep-copied.
                     assert task.best_resources is not None, task
 
-    backend.register_info(dag=dag,
-                          optimize_target=optimize_target,
-                          requested_features=requested_features)
+    backend.register_info(
+        dag=dag,
+        optimize_target=optimize_target,
+        requested_features=requested_features,
+        # That's because we want to do commands in task.setup and task.run again
+        # after K8S pod recovers from a crash.
+        # See `kubernetes-ray.yml.j2` for more details.
+        dump_final_script=is_controller_high_availability_supported)
 
     if task.storage_mounts is not None:
         # Optimizer should eventually choose where to store bucket
diff --git a/sky/global_user_state.py b/sky/global_user_state.py
index 2c1ecc33575..c476c4bd51c 100644
--- a/sky/global_user_state.py
+++ b/sky/global_user_state.py
@@ -350,6 +350,7 @@ def remove_cluster(cluster_name: str, terminate: bool) -> None:
         # Must invalidate IP list to avoid directly trying to ssh into a
         # stopped VM, which leads to timeout.
         if hasattr(handle, 'stable_internal_external_ips'):
+            handle = typing.cast('backends.CloudVmRayResourceHandle', handle)
             handle.stable_internal_external_ips = None
         current_time = int(time.time())
         _DB.cursor.execute(
diff --git a/sky/provision/common.py b/sky/provision/common.py
index 1d58174090e..71508e9415b 100644
--- a/sky/provision/common.py
+++ b/sky/provision/common.py
@@ -207,11 +207,8 @@ def _get_ips(self, use_internal_ips: bool) -> List[str]:
 
     def get_feasible_ips(self, force_internal_ips: bool = False) -> List[str]:
         """Get internal or external IPs depends on the settings."""
-        if self.provider_config is not None:
-            use_internal_ips = self.provider_config.get('use_internal_ips',
-                                                        False)
-        else:
-            use_internal_ips = False
+        use_internal_ips = (self.provider_config is not None and
+                            self.provider_config.get('use_internal_ips', False))
         return self._get_ips(use_internal_ips or not self.has_external_ips() or
                              force_internal_ips)
 
diff --git a/sky/provision/instance_setup.py b/sky/provision/instance_setup.py
index c5df48c8bae..7d08696d0d0 100644
--- a/sky/provision/instance_setup.py
+++ b/sky/provision/instance_setup.py
@@ -499,7 +499,7 @@ def start_skylet_on_head_node(cluster_name: str,
 def _internal_file_mounts(file_mounts: Dict,
                           runner: command_runner.CommandRunner,
                           log_path: str) -> None:
-    if file_mounts is None or not file_mounts:
+    if not file_mounts:
         return
 
     for dst, src in file_mounts.items():
diff --git a/sky/provision/kubernetes/instance.py b/sky/provision/kubernetes/instance.py
index 3e39230dc57..13fc5e0a6a8 100644
--- a/sky/provision/kubernetes/instance.py
+++ b/sky/provision/kubernetes/instance.py
@@ -12,7 +12,6 @@
 from sky.adaptors import kubernetes
 from sky.provision import common
 from sky.provision import constants
-from sky.provision import docker_utils
 from sky.provision.kubernetes import config as config_lib
 from sky.provision.kubernetes import network_utils
 from sky.provision.kubernetes import utils as kubernetes_utils
@@ -33,14 +32,32 @@
 TAG_SKYPILOT_CLUSTER_NAME = 'skypilot-cluster-name'
 TAG_POD_INITIALIZED = 'skypilot-initialized'
 
+# Please be careful when changing this.
+# When mounting, Kubernetes changes the ownership of the parent directory
+# to root:root.
+# pylint: disable=line-too-long
+# See https://stackoverflow.com/questions/50818029/mounted-folder-created-as-root-instead-of-current-user-in-docker/50820023#50820023.
+DEPLOYMENT_VOLUME_MOUNTS = {
+    # volume name -> mount path
+    'sky-data': '/home/sky',
+}
+
+
+def _is_head(pod) -> bool:
+    return pod.metadata.labels.get(constants.TAG_RAY_NODE_KIND) == 'head'
+
 
 def _get_head_pod_name(pods: Dict[str, Any]) -> Optional[str]:
-    head_pod_name = None
-    for pod_name, pod in pods.items():
-        if pod.metadata.labels[constants.TAG_RAY_NODE_KIND] == 'head':
-            head_pod_name = pod_name
-            break
-    return head_pod_name
+    return next((pod_name for pod_name, pod in pods.items() if _is_head(pod)),
+                None)
+
+
+def _get_pvc_name(cluster_name: str, volume_name: str) -> str:
+    return f'{cluster_name}-{volume_name}'
+
+
+def _get_deployment_name(cluster_name: str) -> str:
+    return f'{cluster_name}-deployment'
 
 
 def head_service_selector(cluster_name: str) -> Dict[str, str]:
@@ -382,195 +399,6 @@ def _check_init_containers(pod):
         time.sleep(1)
 
 
-def _run_function_with_retries(func: Callable,
-                               operation_name: str,
-                               max_retries: int = _MAX_RETRIES,
-                               retry_delay: int = 5) -> Any:
-    """Runs a function with retries on Kubernetes errors.
-
-    Args:
-        func: Function to retry
-        operation_name: Name of the operation for logging
-        max_retries: Maximum number of retry attempts
-        retry_delay: Delay between retries in seconds
-
-    Raises:
-        The last exception encountered if all retries fail.
-    """
-    for attempt in range(max_retries + 1):
-        try:
-            return func()
-        except config_lib.KubernetesError:
-            if attempt < max_retries:
-                logger.warning(f'Failed to {operation_name} - '
-                               f'retrying in {retry_delay} seconds.')
-                time.sleep(retry_delay)
-            else:
-                raise
-
-
-@timeline.event
-def pre_init(namespace: str, context: Optional[str], new_nodes: List) -> None:
-    """Pre-initialization step for SkyPilot pods.
-
-    This step is run in the pod right after it is created and before the
-    SkyPilot runtime is setup.
-
-    This step includes three key steps:
-
-    1. Privilege check: Checks if the default user has sufficient privilege
-    to set up the kubernetes instance pod.
-    2. SSH setup: Sets up SSH for the pod instance.
-    3. Environment variable setup to populate k8s env vars in the pod.
-
-    Make sure commands used in these methods are generic and work
-    on most base images. E.g., do not use Python, since that may not
-    be installed by default.
-
-    If you run any apt commands, be sure to check if the lock is available.
-    It is possible the `apt update` run in the pod container args may still
-    be running.
-
-    Args:
-        namespace (str): Kubernetes namespace.
-        context (Optional[str]): Kubernetes context.
-        new_nodes (List): List of new pod instances.
-
-    Raises:
-        config_lib.KubernetesError: If user privileges are insufficient or
-          setup fails.
-    """
-
-    check_k8s_user_sudo_cmd = (
-        'if [ $(id -u) -eq 0 ]; then'
-        # If user is root, create an alias for sudo used in skypilot setup
-        '  echo \'alias sudo=""\' >> ~/.bashrc; echo succeed;'
-        'else '
-        '  if command -v sudo >/dev/null 2>&1; then '
-        '    timeout 2 sudo -l >/dev/null 2>&1 && echo succeed || '
-        f'    ( echo {exceptions.INSUFFICIENT_PRIVILEGES_CODE!r}; '
-        f'      exit {exceptions.INSUFFICIENT_PRIVILEGES_CODE}; ); '
-        '  else '
-        f'    ( echo {exceptions.INSUFFICIENT_PRIVILEGES_CODE!r}; '
-        f'      exit {exceptions.INSUFFICIENT_PRIVILEGES_CODE}; ); '
-        '  fi; '
-        'fi;')
-
-    # Kubernetes automatically populates containers with critical
-    # environment variables, such as those for discovering services running
-    # in the cluster and CUDA/nvidia environment variables. We need to
-    # make sure these env vars are available in every task and ssh session.
-    # This is needed for GPU support and service discovery.
-    # See https://github.com/skypilot-org/skypilot/issues/2287 for more details.
-    # To do so, we capture env vars from the pod's runtime and write them to
-    # /etc/profile.d/, making them available for all users in future
-    # shell sessions.
-    set_k8s_env_var_cmd = docker_utils.SETUP_ENV_VARS_CMD
-
-    check_apt_update_complete_cmd = (
-        'echo "Checking if apt update from container init is complete..."; '
-        'timeout_secs=600; '
-        'start_time=$(date +%s); '
-        'while ! grep -q "Fetched" /tmp/apt-update.log 2>/dev/null; do '
-        '  echo "apt update still running. Logs:"; '
-        '  cat /tmp/apt-update.log || true; '
-        '  current_time=$(date +%s); '
-        '  elapsed=$((current_time - start_time)); '
-        '  if [ $elapsed -ge $timeout_secs ]; then '
-        '    echo "Timed out waiting for apt update"; '
-        '    exit 1; '
-        '  fi; '
-        '  sleep 5; '
-        'done; '
-        'echo "apt update complete."; ')
-
-    install_ssh_k8s_cmd = (
-        'prefix_cmd() '
-        '{ if [ $(id -u) -ne 0 ]; then echo "sudo"; else echo ""; fi; }; '
-        'export DEBIAN_FRONTEND=noninteractive;'
-        'echo "Installing missing packages..."; '
-        'for i in {1..5}; do '
-        '  output=$($(prefix_cmd) apt install openssh-server rsync -y 2>&1); '
-        '  rc=$?; '
-        '  if [ $rc -eq 0 ]; then '
-        '    break; '
-        '  fi; '
-        '  echo "$output" | grep -qi "could not get lock" || '
-        '  grep -qi "Unable to acquire the dpkg frontend lock"; '
-        '  if [ $? -eq 0 ]; then '
-        '    echo "apt install failed due to lock, retrying. (Attempt $i/5)"; '
-        '    sleep 5; '
-        '  else '
-        '    echo "apt install failed for a non-lock reason: $output"; '
-        '    exit $rc; '
-        '  fi; '
-        'done; '
-        'if [ $rc -ne 0 ]; then '
-        '    echo "apt install failed after 5 attempts due to lock errors."; '
-        '    exit $rc; '
-        'fi; '
-        '$(prefix_cmd) mkdir -p /var/run/sshd; '
-        '$(prefix_cmd) '
-        'sed -i "s/PermitRootLogin prohibit-password/PermitRootLogin yes/" '
-        '/etc/ssh/sshd_config; '
-        '$(prefix_cmd) sed '
-        '"s@session\\s*required\\s*pam_loginuid.so@session optional '
-        'pam_loginuid.so@g" -i /etc/pam.d/sshd; '
-        'cd /etc/ssh/ && $(prefix_cmd) ssh-keygen -A; '
-        '$(prefix_cmd) mkdir -p ~/.ssh; '
-        '$(prefix_cmd) chown -R $(whoami) ~/.ssh;'
-        '$(prefix_cmd) chmod 700 ~/.ssh; '
-        '$(prefix_cmd) cat /etc/secret-volume/ssh-publickey* > '
-        '~/.ssh/authorized_keys; '
-        '$(prefix_cmd) chmod 644 ~/.ssh/authorized_keys; '
-        '$(prefix_cmd) service ssh restart; '
-        # Eliminate the error
-        # `mesg: ttyname failed: inappropriate ioctl for device`.
-        # See https://www.educative.io/answers/error-mesg-ttyname-failed-inappropriate-ioctl-for-device  # pylint: disable=line-too-long
-        '$(prefix_cmd) sed -i "s/mesg n/tty -s \\&\\& mesg n/" ~/.profile;')
-
-    pre_init_cmd = ('set -ex; ' + check_k8s_user_sudo_cmd +
-                    set_k8s_env_var_cmd + check_apt_update_complete_cmd +
-                    install_ssh_k8s_cmd)
-
-    def _pre_init_thread(new_node):
-        pod_name = new_node.metadata.name
-        logger.info(f'{"-"*20}Start: Pre-init in pod {pod_name!r} {"-"*20}')
-        runner = command_runner.KubernetesCommandRunner(
-            ((namespace, context), pod_name))
-
-        # Run the combined pre-init command
-        rc, stdout, _ = runner.run(pre_init_cmd,
-                                   require_outputs=True,
-                                   stream_logs=False)
-        if rc == exceptions.INSUFFICIENT_PRIVILEGES_CODE:
-            raise config_lib.KubernetesError(
-                'Insufficient system privileges detected. '
-                'Ensure the default user has root access or '
-                '"sudo" is installed and the user is added to the sudoers '
-                'from the image.')
-
-        op_name = 'pre-init'
-        _raise_command_running_error(op_name, pre_init_cmd, pod_name, rc,
-                                     stdout)
-
-        logger.info(f'{"-"*20}End: Pre-init in pod {pod_name!r} {"-"*20}')
-
-    # Run pre_init in parallel across all new_nodes
-    subprocess_utils.run_in_parallel(_pre_init_thread, new_nodes, _NUM_THREADS)
-
-
-def _label_pod(namespace: str, context: Optional[str], pod_name: str,
-               label: Dict[str, str]) -> None:
-    """Label a pod."""
-    kubernetes.core_api(context).patch_namespaced_pod(
-        pod_name,
-        namespace, {'metadata': {
-            'labels': label
-        }},
-        _request_timeout=kubernetes.API_TIMEOUT)
-
-
 @timeline.event
 def _create_namespaced_pod_with_retries(namespace: str, pod_spec: dict,
                                         context: Optional[str]) -> Any:
@@ -650,6 +478,144 @@ def _create_namespaced_pod_with_retries(namespace: str, pod_spec: dict,
             raise e
 
 
+def _create_persistent_volume_claim(namespace: str, context: Optional[str],
+                                    pvc_name: str) -> None:
+    """Creates a persistent volume claim for SkyServe controller."""
+    try:
+        kubernetes.core_api(context).read_namespaced_persistent_volume_claim(
+            name=pvc_name, namespace=namespace)
+        return
+    except kubernetes.api_exception() as e:
+        if e.status != 404:  # Not found
+            raise
+
+    pvc_spec = {
+        'apiVersion': 'v1',
+        'kind': 'PersistentVolumeClaim',
+        'metadata': {
+            'name': pvc_name,
+        },
+        'spec': {
+            'accessModes': ['ReadWriteOnce'],
+            'resources': {
+                'requests': {
+                    'storage': '10Gi'  # TODO(andyl): use a constant here
+                }
+            }
+        }
+    }
+
+    kubernetes.core_api(context).create_namespaced_persistent_volume_claim(
+        namespace=namespace, body=pvc_spec)
+
+
+def _create_serve_controller_deployment(
+        pod_spec: Dict[str, Any], cluster_name_on_cloud: str, namespace: str,
+        context: Optional[str]) -> Dict[str, Any]:
+    """Creates a deployment for SkyServe controller with persistence."""
+    assert len(
+        pod_spec['spec']['containers']) == 1, 'Only one container is supported'
+
+    # The pod template part of pod_spec is used in the deployment
+    # spec.template.spec
+    spec = pod_spec.pop('spec')
+    container = spec['containers'][0]
+
+    if DEPLOYMENT_VOLUME_MOUNTS:
+        # TODO(andyl): use defaultdict
+        if spec.get('volumes') is None:
+            spec['volumes'] = []
+        if container.get('volumeMounts') is None:
+            container['volumeMounts'] = []
+
+    for volume_name, mount_path in DEPLOYMENT_VOLUME_MOUNTS.items():
+        pvc_name = _get_pvc_name(cluster_name_on_cloud, volume_name)
+        _create_persistent_volume_claim(namespace, context, pvc_name)
+        spec['volumes'].append({
+            'name': volume_name,
+            'persistentVolumeClaim': {
+                'claimName': pvc_name
+            }
+        })
+        container['volumeMounts'].append({
+            'mountPath': mount_path,
+            'name': volume_name
+        })
+
+    template_metadata = pod_spec.pop('metadata')
+    assert pod_spec.keys() == {'apiVersion', 'kind'
+                              }, (f'pod_spec has fields {pod_spec.keys()}')
+
+    deployment_labels = {
+        'app': cluster_name_on_cloud,
+    }
+    template_metadata['labels'].update(deployment_labels)
+
+    image_id = container['image']
+    assert image_id is not None, 'Image ID is None'
+    init_containers = spec.get('initContainers', [])
+    init_containers.append({
+        'name': 'init-copy-home',
+        'image': image_id,
+        'command': ['/bin/sh', '-c'],
+        'args': [
+            'echo "Copying home directory to /mnt/home"; '
+            'rsync -a /home/sky/ /mnt/home; '
+            'echo "Copy completed.";'
+            'ls -la /mnt/home'
+        ],
+        'volumeMounts': [{
+            'name': 'sky-data',
+            'mountPath': '/mnt/home'
+        }]
+    })
+
+    deployment_spec = {
+        'apiVersion': 'apps/v1',
+        'kind': 'Deployment',
+        'metadata': {
+            'name': _get_deployment_name(cluster_name_on_cloud),
+            'namespace': namespace,
+        },
+        'spec': {
+            'replicas': 1,
+            'selector': {
+                'matchLabels': deployment_labels
+            },
+            'template': {
+                'metadata': template_metadata,
+                'spec': {
+                    'initContainers': init_containers,
+                    **spec, 'restartPolicy': 'Always'
+                }
+            }
+        }
+    }
+
+    return deployment_spec
+
+
+@timeline.event
+def _wait_for_deployment_pod(context, namespace, deployment, timeout=60):
+    label_selector = ','.join([
+        f'{key}={value}'
+        for key, value in deployment.spec.selector.match_labels.items()
+    ])
+    target_replicas = deployment.spec.replicas
+    start_time = time.time()
+    while time.time() - start_time < timeout:
+        pods = kubernetes.core_api(context).list_namespaced_pod(
+            namespace, label_selector=label_selector).items
+        # TODO(andyl): not sure if this necessary
+        if len(pods) == target_replicas:
+            return pods
+        time.sleep(2)
+
+    raise TimeoutError(
+        f'Timeout: Not all Pods for Deployment {deployment.metadata.name!r}'
+        ' are created.')
+
+
 @timeline.event
 def _create_pods(region: str, cluster_name_on_cloud: str,
                  config: common.ProvisionConfig) -> common.ProvisionRecord:
@@ -661,6 +627,7 @@ def _create_pods(region: str, cluster_name_on_cloud: str,
     tags = {
         TAG_RAY_CLUSTER_NAME: cluster_name_on_cloud,
     }
+
     pod_spec['metadata']['namespace'] = namespace
     if 'labels' in pod_spec['metadata']:
         pod_spec['metadata']['labels'].update(tags)
@@ -748,7 +715,8 @@ def _create_pod_thread(i: int):
             pod_spec_copy['metadata']['labels'].update(constants.HEAD_NODE_TAGS)
             head_selector = head_service_selector(cluster_name_on_cloud)
             pod_spec_copy['metadata']['labels'].update(head_selector)
-            pod_spec_copy['metadata']['name'] = f'{cluster_name_on_cloud}-head'
+            pod_spec_copy['metadata'][
+                'name'] = f'{cluster_name_on_cloud}-head'  #!
         else:
             # Worker pods
             pod_spec_copy['metadata']['labels'].update(
@@ -799,6 +767,17 @@ def _create_pod_thread(i: int):
             }
             pod_spec_copy['spec']['tolerations'] = [tpu_toleration]
 
+        if kubernetes_utils.is_serve_controller(cluster_name_on_cloud):
+            deployment_spec = _create_serve_controller_deployment(
+                pod_spec_copy, cluster_name_on_cloud, namespace, context)
+            try:
+                return kubernetes.apps_api(
+                    context).create_namespaced_deployment(
+                        namespace, deployment_spec)
+            except Exception as e:
+                print('Deployment failed', e)
+                raise e
+
         return _create_namespaced_pod_with_retries(namespace, pod_spec_copy,
                                                    context)
 
@@ -807,11 +786,17 @@ def _create_pod_thread(i: int):
                                             list(range(to_start_count)),
                                             _NUM_THREADS)
 
+    if kubernetes_utils.is_serve_controller(cluster_name_on_cloud):
+        deployments = copy.deepcopy(pods)
+        pods.clear()  # Since it's not pods. What created here are true pods.
+        for deployment in deployments:
+            pods.extend(_wait_for_deployment_pod(context, namespace,
+                                                 deployment))
+
     # Process created pods
     for pod in pods:
         created_pods[pod.metadata.name] = pod
-        if head_pod_name is None and pod.metadata.labels.get(
-                constants.TAG_RAY_NODE_KIND) == 'head':
+        if head_pod_name is None and _is_head(pod):
             head_pod_name = pod.metadata.name
 
     networking_mode = network_utils.get_networking_mode(
@@ -878,55 +863,69 @@ def stop_instances(
     raise NotImplementedError()
 
 
+def _delete_k8s_resource_with_retry(delete_func: Callable, resource_type: str,
+                                    resource_name: str) -> None:
+    """Helper to delete Kubernetes resources with 404 handling and retries.
+
+    Args:
+        delete_func: Function to call to delete the resource
+        resource_type: Type of resource being deleted (e.g. 'service'),
+            used in logging
+        resource_name: Name of the resource being deleted, used in logging
+    """
+    max_retries = 3
+    retry_delay = 5  # seconds
+
+    for attempt in range(max_retries):
+        try:
+            delete_func()
+            return
+        except kubernetes.api_exception() as e:
+            if e.status == 404:
+                logger.warning(
+                    f'terminate_instances: Tried to delete {resource_type} '
+                    f'{resource_name}, but the {resource_type} was not '
+                    'found (404).')
+                return
+            elif attempt < max_retries - 1:
+                logger.warning(f'terminate_instances: Failed to delete '
+                               f'{resource_type} {resource_name} (attempt '
+                               f'{attempt + 1}/{max_retries}). Error: {e}. '
+                               f'Retrying in {retry_delay} seconds...')
+                time.sleep(retry_delay)
+            else:
+                raise
+
+
+def _delete_services(name_prefix: str, namespace: str,
+                     context: Optional[str]) -> None:
+    """Delete services with the given name prefix.
+
+    Args:
+        name_prefix: Prefix of the service names to delete
+        namespace: Kubernetes namespace
+        context: Kubernetes context
+    """
+    for service_suffix in ['', '-ssh']:
+        service_name = f'{name_prefix}{service_suffix}'
+        # Since we are not saving this lambda, it's a false positive.
+        # TODO(andyl): Wait for https://github.com/pylint-dev/pylint/issues/5263.
+        # pylint: disable=cell-var-from-loop
+        _delete_k8s_resource_with_retry(delete_func=lambda: kubernetes.core_api(
+            context).delete_namespaced_service(name=service_name,
+                                               namespace=namespace,
+                                               _request_timeout=config_lib.
+                                               DELETION_TIMEOUT),
+                                        resource_type='service',
+                                        resource_name=service_name)
+
+
 def _terminate_node(namespace: str, context: Optional[str],
                     pod_name: str) -> None:
-    """Terminate a pod."""
+    """Terminate a pod and its associated services."""
     logger.debug('terminate_instances: calling delete_namespaced_pod')
 
-    def _delete_k8s_resource_with_retry(delete_func: Callable,
-                                        resource_type: str,
-                                        resource_name: str) -> None:
-        """Helper to delete Kubernetes resources with 404 handling and retries.
-
-        Args:
-            delete_func: Function to call to delete the resource
-            resource_type: Type of resource being deleted (e.g. 'service'),
-                used in logging
-            resource_name: Name of the resource being deleted, used in logging
-        """
-        max_retries = 3
-        retry_delay = 5  # seconds
-
-        for attempt in range(max_retries):
-            try:
-                delete_func()
-                return
-            except kubernetes.api_exception() as e:
-                if e.status == 404:
-                    logger.warning(
-                        f'terminate_instances: Tried to delete {resource_type} '
-                        f'{resource_name}, but the {resource_type} was not '
-                        'found (404).')
-                    return
-                elif attempt < max_retries - 1:
-                    logger.warning(f'terminate_instances: Failed to delete '
-                                   f'{resource_type} {resource_name} (attempt '
-                                   f'{attempt + 1}/{max_retries}). Error: {e}. '
-                                   f'Retrying in {retry_delay} seconds...')
-                    time.sleep(retry_delay)
-                else:
-                    raise
-
-    # Delete services for the pod
-    for service_name in [pod_name, f'{pod_name}-ssh']:
-        _delete_k8s_resource_with_retry(
-            delete_func=lambda name=service_name: kubernetes.core_api(
-                context).delete_namespaced_service(name=name,
-                                                   namespace=namespace,
-                                                   _request_timeout=config_lib.
-                                                   DELETION_TIMEOUT),
-            resource_type='service',
-            resource_name=service_name)
+    _delete_services(pod_name, namespace, context)
 
     # Note - delete pod after all other resources are deleted.
     # This is to ensure there are no leftover resources if this down is run
@@ -940,6 +939,35 @@ def _delete_k8s_resource_with_retry(delete_func: Callable,
         resource_name=pod_name)
 
 
+def _terminate_deployment(cluster_name: str, namespace: str,
+                          context: Optional[str]) -> None:
+    """Terminate a deployment."""
+    # Delete services first
+    _delete_services(f'{cluster_name}-head', namespace, context)
+
+    # Delete deployment
+    deployment_name = _get_deployment_name(cluster_name)
+    _delete_k8s_resource_with_retry(delete_func=lambda: kubernetes.apps_api(
+        context).delete_namespaced_deployment(name=deployment_name,
+                                              namespace=namespace,
+                                              _request_timeout=config_lib.
+                                              DELETION_TIMEOUT),
+                                    resource_type='deployment',
+                                    resource_name=deployment_name)
+
+    # Delete PVCs
+    for volume_name in DEPLOYMENT_VOLUME_MOUNTS:
+        pvc_name = _get_pvc_name(cluster_name, volume_name)
+        # pylint: disable=cell-var-from-loop
+        _delete_k8s_resource_with_retry(delete_func=lambda: kubernetes.core_api(
+            context).delete_namespaced_persistent_volume_claim(
+                name=pvc_name,
+                namespace=namespace,
+                _request_timeout=config_lib.DELETION_TIMEOUT),
+                                        resource_type='pvc',
+                                        resource_name=pvc_name)
+
+
 def terminate_instances(
     cluster_name_on_cloud: str,
     provider_config: Dict[str, Any],
@@ -965,8 +993,9 @@ def terminate_instances(
             logger.warning('terminate_instances: Error occurred when analyzing '
                            f'SSH Jump pod: {e}')
 
-    def _is_head(pod) -> bool:
-        return pod.metadata.labels[constants.TAG_RAY_NODE_KIND] == 'head'
+    if kubernetes_utils.is_serve_controller(cluster_name_on_cloud):
+        _terminate_deployment(cluster_name_on_cloud, namespace, context)
+        return
 
     def _terminate_pod_thread(pod_info):
         pod_name, pod = pod_info
@@ -1023,7 +1052,7 @@ def get_cluster_info(
                 tags=pod.metadata.labels,
             )
         ]
-        if pod.metadata.labels[constants.TAG_RAY_NODE_KIND] == 'head':
+        if _is_head(pod):
             head_pod_name = pod_name
             head_spec = pod.spec
             assert head_spec is not None, pod
diff --git a/sky/provision/kubernetes/network.py b/sky/provision/kubernetes/network.py
index a0c51624216..4e3287116db 100644
--- a/sky/provision/kubernetes/network.py
+++ b/sky/provision/kubernetes/network.py
@@ -78,7 +78,7 @@ def _open_ports_using_ingress(
             'https://github.com/kubernetes/ingress-nginx/blob/main/docs/deploy/index.md.'  # pylint: disable=line-too-long
         )
 
-    # Prepare service names, ports,  for template rendering
+    # Prepare service names, ports, for template rendering
     service_details = [
         (f'{cluster_name_on_cloud}--skypilot-svc--{port}', port,
          _PATH_PREFIX.format(
diff --git a/sky/provision/kubernetes/utils.py b/sky/provision/kubernetes/utils.py
index 594e3752b41..50b715864ef 100644
--- a/sky/provision/kubernetes/utils.py
+++ b/sky/provision/kubernetes/utils.py
@@ -1395,6 +1395,16 @@ def __str__(self):
         return self.name
 
 
+def is_serve_controller(cluster_name_on_cloud: str) -> bool:
+    return common_utils.is_cluster_name_indicating_serve_controller(
+        cluster_name_on_cloud)
+
+
+def get_head_pod_name_from_deployment(cluster_name: str) -> str:
+    """Get the name of the head pod from the deployment."""
+    return f'{cluster_name}-deployment-*'
+
+
 def construct_ssh_jump_command(
         private_key_path: str,
         ssh_jump_ip: str,
@@ -2576,7 +2586,8 @@ def process_skypilot_pods(
             # because hash is different across users
             if 'sky-jobs-controller' in cluster_name_on_cloud:
                 jobs_controllers.append(cluster_info)
-            elif 'sky-serve-controller' in cluster_name_on_cloud:
+            elif common_utils.is_cluster_name_indicating_serve_controller(
+                    cluster_name_on_cloud):
                 serve_controllers.append(cluster_info)
         else:
             # Update start_time if this pod started earlier
diff --git a/sky/resources.py b/sky/resources.py
index c4f04ea5543..c7b7513542b 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -1062,6 +1062,7 @@ def make_deploy_variables(self, cluster_name: resources_utils.ClusterName,
         docker_image = self.extract_docker_image()
 
         # Cloud specific variables
+        assert self.cloud is not None
         cloud_specific_variables = self.cloud.make_deploy_resources_variables(
             self, cluster_name, region, zones, num_nodes, dryrun)
 
diff --git a/sky/serve/replica_managers.py b/sky/serve/replica_managers.py
index 981a446b9a2..5e7e17a175e 100644
--- a/sky/serve/replica_managers.py
+++ b/sky/serve/replica_managers.py
@@ -632,10 +632,44 @@ def __init__(self, service_name: str, spec: 'service_spec.SkyServiceSpec',
         self._down_process_pool: serve_utils.ThreadSafeDict[
             int, multiprocessing.Process] = serve_utils.ThreadSafeDict()
 
+        self._recover_replica_operations()
+
         threading.Thread(target=self._process_pool_refresher).start()
         threading.Thread(target=self._job_status_fetcher).start()
         threading.Thread(target=self._replica_prober).start()
 
+    def _recover_replica_operations(self):
+        """Let's see are there something to do for ReplicaManager in a
+        recovery run"""
+        assert (not self._launch_process_pool and not self._down_process_pool
+               ), 'We should not have any running processes in a recovery run'
+
+        # Pending replicas can be dropped safely, and we're only risk
+        # of launching less replicas than expected.
+        # But for provisioning and shutting down replicas, we need to
+        # be careful to prevent resource leak.
+        to_down_replicas = serve_state.get_replicas_at_statuses(
+            self._service_name, [serve_state.ReplicaStatus.SHUTTING_DOWN])
+
+        to_down_replicas += serve_state.get_replicas_at_statuses(
+            self._service_name, [
+                serve_state.ReplicaStatus.PENDING,
+                serve_state.ReplicaStatus.PROVISIONING
+            ])
+
+        for replica_info in to_down_replicas:
+            replica_id = replica_info.replica_id
+            try:
+                self._terminate_replica(replica_id,
+                                        sync_down_logs=False,
+                                        purge=True,
+                                        replica_drain_delay_seconds=0)
+            # pylint: disable=broad-except
+            except Exception as e:
+                logger.error(
+                    f'Failed to terminate replica {replica_id} in recovery'
+                    f'run: {e}')
+
     ################################
     # Replica management functions #
     ################################
@@ -834,7 +868,9 @@ def _refresh_process_pool(self) -> None:
         the fly. If any of them finished, it will update the status of the
         corresponding replica.
         """
-        for replica_id, p in list(self._launch_process_pool.items()):
+        # To avoid `dictionary changed size during iteration` error.
+        launch_process_pool_snapshot = list(self._launch_process_pool.items())
+        for replica_id, p in launch_process_pool_snapshot:
             if not p.is_alive():
                 info = serve_state.get_replica_info_from_id(
                     self._service_name, replica_id)
@@ -876,7 +912,8 @@ def _refresh_process_pool(self) -> None:
                     self._terminate_replica(replica_id,
                                             sync_down_logs=True,
                                             replica_drain_delay_seconds=0)
-        for replica_id, p in list(self._down_process_pool.items()):
+        down_process_pool_snapshot = list(self._down_process_pool.items())
+        for replica_id, p in down_process_pool_snapshot:
             if not p.is_alive():
                 logger.info(
                     f'Terminate process for replica {replica_id} finished.')
diff --git a/sky/serve/serve_state.py b/sky/serve/serve_state.py
index 1ea2a0d6a97..7db4cbe7759 100644
--- a/sky/serve/serve_state.py
+++ b/sky/serve/serve_state.py
@@ -6,7 +6,7 @@
 import pickle
 import sqlite3
 import typing
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
 
 import colorama
 
@@ -485,6 +485,14 @@ def total_number_provisioning_replicas() -> int:
     return provisioning_count
 
 
+def get_replicas_at_statuses(
+    service_name: str, statuses: Union[ReplicaStatus, Sequence[ReplicaStatus]]
+) -> List['replica_managers.ReplicaInfo']:
+    replicas = get_replica_infos(service_name)
+    statuses = [statuses] if isinstance(statuses, ReplicaStatus) else statuses
+    return [replica for replica in replicas if replica.status in statuses]
+
+
 # === Version functions ===
 def add_version(service_name: str) -> int:
     """Adds a version to the database."""
@@ -555,3 +563,25 @@ def delete_all_versions(service_name: str) -> None:
             """\
             DELETE FROM version_specs
             WHERE service_name=(?)""", (service_name,))
+
+
+def get_service_controller_port(service_name: str) -> int:
+    """Gets the controller port of a service."""
+    with db_utils.safe_cursor(_DB_PATH) as cursor:
+        cursor.execute('SELECT controller_port FROM services WHERE name = ?',
+                       (service_name,))
+        row = cursor.fetchone()
+        if row is None:
+            raise ValueError(f'Service {service_name} does not exist.')
+        return row[0]
+
+
+def get_service_load_balancer_port(service_name: str) -> int:
+    """Gets the load balancer port of a service."""
+    with db_utils.safe_cursor(_DB_PATH) as cursor:
+        cursor.execute('SELECT load_balancer_port FROM services WHERE name = ?',
+                       (service_name,))
+        row = cursor.fetchone()
+        if row is None:
+            raise ValueError(f'Service {service_name} does not exist.')
+        return row[0]
diff --git a/sky/serve/service.py b/sky/serve/service.py
index f075990a6a0..783c847e46d 100644
--- a/sky/serve/service.py
+++ b/sky/serve/service.py
@@ -6,6 +6,7 @@
 import multiprocessing
 import os
 import pathlib
+from pathlib import Path
 import shutil
 import time
 import traceback
@@ -25,6 +26,7 @@
 from sky.serve import replica_managers
 from sky.serve import serve_state
 from sky.serve import serve_utils
+from sky.skylet import constants as skylet_constants
 from sky.utils import common_utils
 from sky.utils import subprocess_utils
 from sky.utils import ux_utils
@@ -130,6 +132,19 @@ def cleanup_version_storage(version: int) -> bool:
     return failed
 
 
+def is_recovery_mode(service_name: str) -> bool:
+    """Check if service exists in database to determine recovery mode.
+
+    Args:
+        service_name: Name of the service to check
+
+    Returns:
+        True if service exists in database, indicating recovery mode
+    """
+    service = serve_state.get_service_from_name(service_name)
+    return service is not None
+
+
 def _start(service_name: str, tmp_task_yaml: str, job_id: int):
     """Starts the service."""
     # Generate ssh key pair to avoid race condition when multiple sky.launch
@@ -141,28 +156,34 @@ def _start(service_name: str, tmp_task_yaml: str, job_id: int):
     # Already checked before submit to controller.
     assert task.service is not None, task
     service_spec = task.service
-    if len(serve_state.get_services()) >= serve_utils.NUM_SERVICE_THRESHOLD:
-        cleanup_storage(tmp_task_yaml)
-        with ux_utils.print_exception_no_traceback():
-            raise RuntimeError('Max number of services reached.')
-    success = serve_state.add_service(
-        service_name,
-        controller_job_id=job_id,
-        policy=service_spec.autoscaling_policy_str(),
-        requested_resources_str=backend_utils.get_task_resources_str(task),
-        load_balancing_policy=service_spec.load_balancing_policy,
-        status=serve_state.ServiceStatus.CONTROLLER_INIT,
-        tls_encrypted=service_spec.tls_credential is not None)
-    # Directly throw an error here. See sky/serve/api.py::up
-    # for more details.
-    if not success:
-        cleanup_storage(tmp_task_yaml)
-        with ux_utils.print_exception_no_traceback():
-            raise ValueError(f'Service {service_name} already exists.')
-
-    # Add initial version information to the service state.
-    serve_state.add_or_update_version(service_name, constants.INITIAL_VERSION,
-                                      service_spec)
+
+    is_recovery = is_recovery_mode(service_name)
+    logger.info(f'It\'s a {"first" if not is_recovery else "recovery"} run')
+
+    if not is_recovery:
+        if len(serve_state.get_services()) >= serve_utils.NUM_SERVICE_THRESHOLD:
+            cleanup_storage(tmp_task_yaml)
+            with ux_utils.print_exception_no_traceback():
+                raise RuntimeError('Max number of services reached.')
+        success = serve_state.add_service(
+            service_name,
+            controller_job_id=job_id,
+            policy=service_spec.autoscaling_policy_str(),
+            requested_resources_str=backend_utils.get_task_resources_str(task),
+            load_balancing_policy=service_spec.load_balancing_policy,
+            status=serve_state.ServiceStatus.CONTROLLER_INIT,
+            tls_encrypted=service_spec.tls_credential is not None)
+        # Directly throw an error here. See sky/serve/api.py::up
+        # for more details.
+        if not success:
+            cleanup_storage(tmp_task_yaml)
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(f'Service {service_name} already exists.')
+
+        # Add initial version information to the service state.
+        serve_state.add_or_update_version(service_name,
+                                          constants.INITIAL_VERSION,
+                                          service_spec)
 
     # Create the service working directory.
     service_dir = os.path.expanduser(
@@ -188,8 +209,15 @@ def _start(service_name: str, tmp_task_yaml: str, job_id: int):
     try:
         with filelock.FileLock(
                 os.path.expanduser(constants.PORT_SELECTION_FILE_LOCK_PATH)):
-            controller_port = common_utils.find_free_port(
-                constants.CONTROLLER_PORT_START)
+            if is_recovery:
+                # In recovery mode, use the ports from the database
+                controller_port = serve_state.get_service_controller_port(
+                    service_name)
+                load_balancer_port = serve_state.get_service_load_balancer_port(
+                    service_name)
+            else:
+                controller_port = common_utils.find_free_port(
+                    constants.CONTROLLER_PORT_START)
 
             # We expose the controller to the public network when running
             # inside a kubernetes cluster to allow external load balancers
@@ -212,13 +240,15 @@ def _get_host():
                 args=(service_name, service_spec, task_yaml, controller_host,
                       controller_port))
             controller_process.start()
-            serve_state.set_service_controller_port(service_name,
-                                                    controller_port)
+            if not is_recovery:
+                serve_state.set_service_controller_port(service_name,
+                                                        controller_port)
 
             controller_addr = f'http://{controller_host}:{controller_port}'
 
-            load_balancer_port = common_utils.find_free_port(
-                constants.LOAD_BALANCER_PORT_START)
+            if not is_recovery:
+                load_balancer_port = common_utils.find_free_port(
+                    constants.LOAD_BALANCER_PORT_START)
 
             # Extract the load balancing policy from the service spec
             policy_name = service_spec.load_balancing_policy
@@ -234,8 +264,9 @@ def _get_host():
                 args=(controller_addr, load_balancer_port, policy_name,
                       service_spec.tls_credential))
             load_balancer_process.start()
-            serve_state.set_service_load_balancer_port(service_name,
-                                                       load_balancer_port)
+            if not is_recovery:
+                serve_state.set_service_load_balancer_port(
+                    service_name, load_balancer_port)
 
         while True:
             _handle_signal(service_name)
@@ -265,8 +296,26 @@ def _get_host():
             serve_state.delete_all_versions(service_name)
             logger.info(f'Service {service_name} terminated successfully.')
 
+        task_run_dir = Path(
+            skylet_constants.PERSISTENT_RUN_SCRIPT_DIR).expanduser()
+        if task_run_dir.exists():
+            # Please see `kubernetes-ray.yml.j2` for more details.
+            this_task_run_script = task_run_dir / f'sky_job_{job_id}'
+            if this_task_run_script.exists():
+                this_task_run_script.unlink()
+                logger.info(f'Task run script {this_task_run_script} removed')
+            else:
+                logger.warning(
+                    f'Task run script {this_task_run_script} not found')
+
 
 if __name__ == '__main__':
+    if not pathlib.Path('/home/sky/.sky/k8s_container_ready').exists():
+        print('k8s_container_ready not found, exiting')
+        exit(0)
+
+    print('k8s_container_ready found, starting service')
+
     parser = argparse.ArgumentParser(description='Sky Serve Service')
     parser.add_argument('--service-name',
                         type=str,
diff --git a/sky/skylet/constants.py b/sky/skylet/constants.py
index 4d37e9682e2..28016612f4e 100644
--- a/sky/skylet/constants.py
+++ b/sky/skylet/constants.py
@@ -320,3 +320,8 @@
 ROLE_ASSIGNMENT_FAILURE_ERROR_MSG = (
     'Failed to assign Storage Blob Data Owner role to the '
     'storage account {storage_account_name}.')
+
+# Constants for path in K8S pod to store persistent setup and run scripts
+# so that we can run them again after the pod restarts.
+PERSISTENT_SETUP_SCRIPT_PATH = '~/.sky/.controller_recovery_setup_commands.sh'
+PERSISTENT_RUN_SCRIPT_DIR = '~/.sky/.controller_recovery_task_run'
diff --git a/sky/task.py b/sky/task.py
index bbf6d59b2ae..de60052fb0c 100644
--- a/sky/task.py
+++ b/sky/task.py
@@ -279,7 +279,7 @@ def __init__(
         self.service_name: Optional[str] = None
 
         # Filled in by the optimizer.  If None, this Task is not planned.
-        self.best_resources = None
+        self.best_resources: Optional[sky.Resources] = None
         # Check if the task is legal.
         self._validate()
 
diff --git a/sky/templates/kubernetes-ray.yml.j2 b/sky/templates/kubernetes-ray.yml.j2
index c246141c03f..4af34a4f6f0 100644
--- a/sky/templates/kubernetes-ray.yml.j2
+++ b/sky/templates/kubernetes-ray.yml.j2
@@ -415,6 +415,7 @@ available_node_types:
                 done
                 {{ conda_installation_commands }}
                 {{ ray_installation_commands }}
+
                 VIRTUAL_ENV=~/skypilot-runtime ~/.local/bin/uv pip install skypilot[kubernetes,remote]
                 touch /tmp/ray_skypilot_installation_complete
                 echo "=== Ray and skypilot installation completed ==="
@@ -518,6 +519,26 @@ available_node_types:
                   sleep 0.1
                 done
               }
+
+              mkdir -p ~/.sky/.controller_recovery_task_run
+              if [ -f ~/.sky/k8s_container_ready ]; then
+                # ! Keep this aligned with `CloudVmRayBackend._setup()`
+                # Suppose all `task.setup` are the same for skyserve controller task.
+                # So be careful for compatibility issue once you change it.
+                chmod +x ~/.sky/.controller_recovery_setup_commands.sh
+                /bin/bash --login -c "true && export OMP_NUM_THREADS=1 PYTHONWARNINGS='ignore' && ~/.sky/.controller_recovery_setup_commands.sh > /tmp/controller_recovery_setup_commands.log 2>&1"
+                echo "=== Controller setup commands completed for recovery ==="
+
+                for file in ~/.sky/.controller_recovery_task_run/*; do
+                  # ! Keep this aligned with `CloudVmRayBackend._execute()`
+                  chmod +x $file
+                  /bin/bash --login -c "true && export OMP_NUM_THREADS=1 PYTHONWARNINGS='ignore' && $file > /tmp/task_run_$(basename $file).log 2>&1"
+                  echo "=== Controller task run for service (file: $file) completed for recovery ==="
+                done
+              fi
+
+              touch ~/.sky/k8s_container_ready
+
               trap : TERM INT; log_tail || sleep infinity & wait
 
           ports:
diff --git a/sky/utils/command_runner.py b/sky/utils/command_runner.py
index 92d1f2749d7..2864fbc0301 100644
--- a/sky/utils/command_runner.py
+++ b/sky/utils/command_runner.py
@@ -289,6 +289,8 @@ def _get_remote_home_dir_with_retry():
             resolved_target = target
             if target.startswith('~'):
                 remote_home_dir = _get_remote_home_dir_with_retry()
+                assert remote_home_dir is not None, ('remote_home_dir '
+                                                     'returns None')
                 resolved_target = target.replace('~', remote_home_dir)
             full_source_str = str(resolved_source)
             if resolved_source.is_dir():
@@ -301,6 +303,8 @@ def _get_remote_home_dir_with_retry():
             resolved_source = source
             if source.startswith('~'):
                 remote_home_dir = _get_remote_home_dir_with_retry()
+                assert remote_home_dir is not None, ('remote_home_dir '
+                                                     'returns None')
                 resolved_source = source.replace('~', remote_home_dir)
             rsync_command.extend([
                 f'{node_destination}:{resolved_source!r}',
@@ -309,8 +313,12 @@ def _get_remote_home_dir_with_retry():
         command = ' '.join(rsync_command)
         logger.debug(f'Running rsync command: {command}')
 
+        print(f'Running rsync command: {command}')
+        logger.info(f'Running rsync command: {command}')
+
         backoff = common_utils.Backoff(initial_backoff=5, max_backoff_factor=5)
         assert max_retry > 0, f'max_retry {max_retry} must be positive.'
+        stdout, stderr = '', ''
         while max_retry >= 0:
             returncode, stdout, stderr = log_lib.run_with_log(
                 command,
@@ -326,6 +334,7 @@ def _get_remote_home_dir_with_retry():
         direction = 'up' if up else 'down'
         error_msg = (f'Failed to rsync {direction}: {source} -> {target}. '
                      'Ensure that the network is stable, then retry.')
+
         subprocess_utils.handle_returncode(returncode,
                                            command,
                                            error_msg,
diff --git a/sky/utils/common_utils.py b/sky/utils/common_utils.py
index 5ec4011a691..edb015be931 100644
--- a/sky/utils/common_utils.py
+++ b/sky/utils/common_utils.py
@@ -46,6 +46,11 @@
 _usage_run_id = None
 
 
+def is_cluster_name_indicating_serve_controller(cluster_name: str) -> bool:
+    """Check if the cluster name is indicating a serve controller."""
+    return cluster_name.startswith('sky-serve-controller-')
+
+
 def get_usage_run_id() -> str:
     """Returns a unique run id for each 'run'.
 
diff --git a/sky/utils/controller_utils.py b/sky/utils/controller_utils.py
index 116515acebb..6a03f848a8b 100644
--- a/sky/utils/controller_utils.py
+++ b/sky/utils/controller_utils.py
@@ -435,6 +435,25 @@ def shared_controller_vars_to_fill(
     return vars_to_fill
 
 
+def get_controller_high_availability_supported(
+    controller: Controllers,) -> bool:
+    """Check if the controller high availability is supported.
+    """
+    if skypilot_config.loaded():
+        high_availability_specified = skypilot_config.get_nested(
+            (controller.value.controller_type, 'controller',
+             'high_availability'), False)
+        if high_availability_specified:
+            if controller != Controllers.SKY_SERVE_CONTROLLER:
+                print(f'{colorama.Fore.RED}High availability controller is only'
+                      'supported for SkyServe controller. It cannot be enabled '
+                      f'for {controller.value.name}. Skipping this flag.'
+                      f'{colorama.Style.RESET_ALL}')
+            else:
+                return True
+    return False
+
+
 def get_controller_resources(
     controller: Controllers,
     task_resources: Iterable['resources.Resources'],
diff --git a/sky/utils/schemas.py b/sky/utils/schemas.py
index 06364686800..f421877f96a 100644
--- a/sky/utils/schemas.py
+++ b/sky/utils/schemas.py
@@ -725,6 +725,9 @@ def get_config_schema():
                 'additionalProperties': False,
                 'properties': {
                     'resources': resources_schema,
+                    'high_availability': {
+                        'type': 'boolean',
+                    },
                 }
             },
             'bucket': {
diff --git a/tests/smoke_tests/test_sky_serve.py b/tests/smoke_tests/test_sky_serve.py
index ddc3ea8a92a..ad2f9ed08ae 100644
--- a/tests/smoke_tests/test_sky_serve.py
+++ b/tests/smoke_tests/test_sky_serve.py
@@ -67,8 +67,8 @@ def _get_service_name() -> str:
     ' done; }}; echo "Got service status $s";'
     f'sleep {serve.LB_CONTROLLER_SYNC_INTERVAL_SECONDS + 2};')
 _IP_REGEX = r'([0-9]{1,3}\.){3}[0-9]{1,3}'
-_AWK_ALL_LINES_BELOW_REPLICAS = r'/Replicas/{flag=1; next} flag'
-_SERVICE_LAUNCHING_STATUS_REGEX = 'PROVISIONING\|STARTING'
+_AWK_ALL_LINES_BELOW_REPLICAS = '/Replicas/{flag=1; next} flag'
+_SERVICE_LAUNCHING_STATUS_REGEX = r'PROVISIONING\|STARTING'
 # Since we don't allow terminate the service if the controller is INIT,
 # which is common for simultaneous pytest, we need to wait until the
 # controller is UP before we can terminate the service.
@@ -79,7 +79,7 @@ def _get_service_name() -> str:
     '     s=$(sky serve down -y {name});'
     '     echo "Trying to terminate {name}";'
     '     echo "$s";'
-    '     echo "$s" | grep -q "scheduled to be terminated\|No service to terminate" && break;'
+    r'     echo "$s" | grep -q "scheduled to be terminated\|No service to terminate" && break;'
     '     sleep 10;'
     '     [ $i -eq 20 ] && echo "Failed to terminate service {name}";'
     'done)')
@@ -175,7 +175,7 @@ def _check_replica_in_status(name: str, check_tuples: List[Tuple[int, bool,
                          ] and not status.startswith('FAILED'):
             spot_str = ''
             if is_spot:
-                spot_str = '\[Spot\]'
+                spot_str = r'\[Spot\]'
             resource_str = f'({spot_str}vCPU=2)'
         check_cmd += (f' echo "$s" | grep "{resource_str}" | '
                       f'grep "{status}" | wc -l | grep {count} || exit 1;')
@@ -188,7 +188,7 @@ def _check_service_version(service_name: str, version: str) -> str:
     # Grep the lines before 'Service Replicas' and check if the service version
     # is correct.
     return (f'echo "$s" | grep -B1000 "Service Replicas" | '
-            f'grep -E "{service_name}\s+{version}" || exit 1; ')
+            rf'grep -E "{service_name}\s+{version}" || exit 1; ')
 
 
 @pytest.mark.gcp
@@ -239,12 +239,12 @@ def test_skyserve_oci_http():
 
 @pytest.mark.no_fluidstack  # Fluidstack does not support T4 gpus for now
 @pytest.mark.no_vast  # Vast has low availability of T4 GPUs
-@pytest.mark.parametrize('accelerator', [{'do': 'H100'}])
+@pytest.mark.parametrize('accelerator_dict', [{'do': 'H100'}])
 @pytest.mark.serve
 @pytest.mark.resource_heavy
-def test_skyserve_llm(generic_cloud: str, accelerator: Dict[str, str]):
+def test_skyserve_llm(generic_cloud: str, accelerator_dict: Dict[str, str]):
     """Test skyserve with real LLM usecase"""
-    accelerator = accelerator.get(generic_cloud, 'T4')
+    accelerator = accelerator_dict.get(generic_cloud, 'T4')
     name = _get_service_name()
 
     def generate_llm_test_command(prompt: str, expected_output: str) -> str:
@@ -260,7 +260,7 @@ def generate_llm_test_command(prompt: str, expected_output: str) -> str:
         prompt2output = json.load(f)
 
     test = smoke_tests_utils.Test(
-        f'test-skyserve-llm',
+        'test-skyserve-llm',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} --gpus {accelerator} -y tests/skyserve/llm/service.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=1),
@@ -282,7 +282,7 @@ def test_skyserve_spot_recovery():
     zone = 'us-central1-a'
 
     test = smoke_tests_utils.Test(
-        f'test-skyserve-spot-recovery-gcp',
+        'test-skyserve-spot-recovery-gcp',
         [
             f'sky serve up -n {name} -y tests/skyserve/spot/recovery.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=1),
@@ -307,7 +307,7 @@ def test_skyserve_spot_recovery():
 def test_skyserve_base_ondemand_fallback(generic_cloud: str):
     name = _get_service_name()
     test = smoke_tests_utils.Test(
-        f'test-skyserve-base-ondemand-fallback',
+        'test-skyserve-base-ondemand-fallback',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/spot/base_ondemand_fallback.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=2),
@@ -327,15 +327,15 @@ def test_skyserve_dynamic_ondemand_fallback():
     zone = 'us-central1-a'
 
     test = smoke_tests_utils.Test(
-        f'test-skyserve-dynamic-ondemand-fallback',
+        'test-skyserve-dynamic-ondemand-fallback',
         [
             f'sky serve up -n {name} --cloud gcp -y tests/skyserve/spot/dynamic_ondemand_fallback.yaml',
-            f'sleep 40',
+            'sleep 40',
             # 2 on-demand (provisioning) + 2 Spot (provisioning).
             f'{_SERVE_STATUS_WAIT.format(name=name)}; echo "$s";'
             'echo "$s" | grep -q "0/4" || exit 1',
             # Wait for the provisioning starts
-            f'sleep 40',
+            'sleep 40',
             _check_replica_in_status(name, [
                 (2, True, _SERVICE_LAUNCHING_STATUS_REGEX + '\|READY'),
                 (2, False, _SERVICE_LAUNCHING_STATUS_REGEX + '\|SHUTTING_DOWN')
@@ -346,7 +346,7 @@ def test_skyserve_dynamic_ondemand_fallback():
             _check_replica_in_status(name, [(2, True, 'READY'),
                                             (0, False, '')]),
             smoke_tests_utils.terminate_gcp_replica(name, zone, 1),
-            f'sleep 40',
+            'sleep 40',
             # 1 on-demand (provisioning) + 1 Spot (ready) + 1 spot (provisioning).
             f'{_SERVE_STATUS_WAIT.format(name=name)}; '
             'echo "$s" | grep -q "1/3"',
@@ -377,7 +377,7 @@ def test_skyserve_user_bug_restart(generic_cloud: str):
     # TODO(zhwu): this behavior needs some rethinking.
     name = _get_service_name()
     test = smoke_tests_utils.Test(
-        f'test-skyserve-user-bug-restart',
+        'test-skyserve-user-bug-restart',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/restart/user_bug.yaml',
             f's=$(sky serve status {name}); echo "$s";'
@@ -411,7 +411,7 @@ def test_skyserve_load_balancer(generic_cloud: str):
     """Test skyserve load balancer round-robin policy"""
     name = _get_service_name()
     test = smoke_tests_utils.Test(
-        f'test-skyserve-load-balancer',
+        'test-skyserve-load-balancer',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/load_balancer/service.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=3),
@@ -436,7 +436,7 @@ def test_skyserve_auto_restart():
     name = _get_service_name()
     zone = 'us-central1-a'
     test = smoke_tests_utils.Test(
-        f'test-skyserve-auto-restart',
+        'test-skyserve-auto-restart',
         [
             # TODO(tian): we can dynamically generate YAML from template to
             # avoid maintaining too many YAML files
@@ -446,7 +446,7 @@ def test_skyserve_auto_restart():
             'request_output=$(curl $endpoint); echo "$request_output"; echo "$request_output" | grep "Hi, SkyPilot here"',
             # sleep for 20 seconds (initial delay) to make sure it will
             # be restarted
-            f'sleep 20',
+            'sleep 20',
             smoke_tests_utils.terminate_gcp_replica(name, zone, 1),
             # Wait for consecutive failure timeout passed.
             # If the cluster is not using spot, it won't check the cluster status
@@ -479,7 +479,7 @@ def test_skyserve_cancel(generic_cloud: str):
     name = _get_service_name()
 
     test = smoke_tests_utils.Test(
-        f'test-skyserve-cancel',
+        'test-skyserve-cancel',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/cancel/cancel.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=1),
@@ -505,7 +505,7 @@ def test_skyserve_streaming(generic_cloud: str):
     """Test skyserve with streaming"""
     name = _get_service_name()
     test = smoke_tests_utils.Test(
-        f'test-skyserve-streaming',
+        'test-skyserve-streaming',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/streaming/streaming.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=1),
@@ -525,7 +525,7 @@ def test_skyserve_readiness_timeout_fail(generic_cloud: str):
     """Test skyserve with large readiness probe latency, expected to fail"""
     name = _get_service_name()
     test = smoke_tests_utils.Test(
-        f'test-skyserve-readiness-timeout-fail',
+        'test-skyserve-readiness-timeout-fail',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/readiness_timeout/task.yaml',
             # None of the readiness probe will pass, so the service will be
@@ -550,7 +550,7 @@ def test_skyserve_large_readiness_timeout(generic_cloud: str):
     """Test skyserve with customized large readiness timeout"""
     name = _get_service_name()
     test = smoke_tests_utils.Test(
-        f'test-skyserve-large-readiness-timeout',
+        'test-skyserve-large-readiness-timeout',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/readiness_timeout/task_large_timeout.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=1),
@@ -573,7 +573,7 @@ def test_skyserve_update(generic_cloud: str):
     """Test skyserve with update"""
     name = _get_service_name()
     test = smoke_tests_utils.Test(
-        f'test-skyserve-update',
+        'test-skyserve-update',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/update/old.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=2),
@@ -609,7 +609,7 @@ def test_skyserve_rolling_update(generic_cloud: str):
         name, [(2, False, 'READY'), (1, False, _SERVICE_LAUNCHING_STATUS_REGEX),
                (1, False, 'SHUTTING_DOWN')])
     test = smoke_tests_utils.Test(
-        f'test-skyserve-rolling-update',
+        'test-skyserve-rolling-update',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/update/old.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=2),
@@ -646,7 +646,7 @@ def test_skyserve_fast_update(generic_cloud: str):
     name = _get_service_name()
 
     test = smoke_tests_utils.Test(
-        f'test-skyserve-fast-update',
+        'test-skyserve-fast-update',
         [
             f'sky serve up -n {name} -y --cloud {generic_cloud} tests/skyserve/update/bump_version_before.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=2),
@@ -688,7 +688,7 @@ def test_skyserve_update_autoscale(generic_cloud: str):
     """Test skyserve update with autoscale"""
     name = _get_service_name()
     test = smoke_tests_utils.Test(
-        f'test-skyserve-update-autoscale',
+        'test-skyserve-update-autoscale',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/update/num_min_two.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=2) +
@@ -766,7 +766,7 @@ def test_skyserve_new_autoscaler_update(mode: str, generic_cloud: str):
             's=$(curl $endpoint); echo "$s"; echo "$s" | grep "Hi, SkyPilot here"',
             f'sky serve update {name} --cloud {generic_cloud} --mode {mode} -y tests/skyserve/update/new_autoscaler_after.yaml',
             # Wait for update to be registered
-            f'sleep 90',
+            'sleep 90',
             wait_until_no_pending,
             _check_replica_in_status(
                 name, [(4, True, _SERVICE_LAUNCHING_STATUS_REGEX + '\|READY'),
@@ -851,7 +851,7 @@ def test_skyserve_https(generic_cloud: str):
             f'-subj "/" -keyout {keyfile} -out {certfile}')
 
         test = smoke_tests_utils.Test(
-            f'test-skyserve-https',
+            'test-skyserve-https',
             [
                 f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/https/service.yaml '
                 f'--env TLS_KEYFILE_ENV_VAR={keyfile} --env TLS_CERTFILE_ENV_VAR={certfile}',
@@ -881,7 +881,7 @@ def test_skyserve_multi_ports(generic_cloud: str):
     """Test skyserve with multiple ports"""
     name = _get_service_name()
     test = smoke_tests_utils.Test(
-        f'test-skyserve-multi-ports',
+        'test-skyserve-multi-ports',
         [
             f'sky serve up -n {name} --cloud {generic_cloud} -y tests/skyserve/multi_ports.yaml',
             _SERVE_WAIT_UNTIL_READY.format(name=name, replica_num=1),


Code review comments:
diff_hunk: @@ -605,6 +607,15 @@ def check_features_are_supported(
         unsupported_features2reason = cls._unsupported_features_for_resources(
             resources)
 
+        # pylint: disable=import-outside-toplevel
+        from sky.clouds import kubernetes
+        if cls is not kubernetes.Kubernetes:
path: sky/clouds/cloud.py
text: @user1:
pls check `is_same_cloud`

@author:
`is_same_cloud` is used for instance, not for class object

diff_hunk: @@ -1395,6 +1395,16 @@ def __str__(self):
         return self.name
 
 
+def is_serve_controller(cluster_name_on_cloud: str) -> bool:
+    return common_utils.is_cluster_name_indicating_serve_controller(
+        cluster_name_on_cloud)
+
+
+def get_head_pod_name_from_deployment(cluster_name: str) -> str:
path: sky/provision/kubernetes/utils.py
text: @user1:
```suggestion
def get_head_pod_name_template_from_deployment(cluster_name: str) -> str:
```

diff_hunk: @@ -632,10 +632,44 @@ def __init__(self, service_name: str, spec: 'service_spec.SkyServiceSpec',
         self._down_process_pool: serve_utils.ThreadSafeDict[
             int, multiprocessing.Process] = serve_utils.ThreadSafeDict()
 
+        self._recover_replica_operations()
+
         threading.Thread(target=self._process_pool_refresher).start()
         threading.Thread(target=self._job_status_fetcher).start()
         threading.Thread(target=self._replica_prober).start()
 
+    def _recover_replica_operations(self):
+        """Let's see are there something to do for ReplicaManager in a
+        recovery run"""
+        assert (not self._launch_process_pool and not self._down_process_pool
+               ), 'We should not have any running processes in a recovery run'
+
+        # Pending replicas can be dropped safely, and we're only risk
+        # of launching less replicas than expected.
+        # But for provisioning and shutting down replicas, we need to
+        # be careful to prevent resource leak.
+        to_down_replicas = serve_state.get_replicas_at_statuses(
+            self._service_name, [serve_state.ReplicaStatus.SHUTTING_DOWN])
+
+        to_down_replicas += serve_state.get_replicas_at_statuses(
+            self._service_name, [
+                serve_state.ReplicaStatus.PENDING,
+                serve_state.ReplicaStatus.PROVISIONING
+            ])
path: sky/serve/replica_managers.py
text: @user1:
for pending & provisioning replicas, why not launch them?

@author:
That's a great question. For provisioning replicas, we can't be certain which step of `sky.launch` they were in right before the controller crashed. There’s no guarantee that provisioning was completed and without any internal inconsistencies, nor can we be sure that the cloud API wasn’t already called, potentially leading to the cluster starting on its own later.

So are we confident that our `sky.launch` is robust enough to handle cases where the last launch was interrupted at any point?

@user1:
sky launch should be able to fix the cluster into a correct status. cc @user2 here

@author:
Just asked @user2 . Should safely use `sky.launch` here

diff_hunk: @@ -6,6 +6,7 @@
 import multiprocessing
 import os
 import pathlib
+from pathlib import Path
path: sky/serve/service.py
text: @user1:
nit: import module instead of class

diff_hunk: @@ -309,8 +313,12 @@ def _get_remote_home_dir_with_retry():
         command = ' '.join(rsync_command)
         logger.debug(f'Running rsync command: {command}')
 
+        print(f'Running rsync command: {command}')
+        logger.info(f'Running rsync command: {command}')
+
path: sky/utils/command_runner.py
text: @user1:
accidentally?

@author:
Good catch!

diff_hunk: @@ -265,8 +296,26 @@ def _get_host():
             serve_state.delete_all_versions(service_name)
             logger.info(f'Service {service_name} terminated successfully.')
 
+        task_run_dir = Path(
+            skylet_constants.PERSISTENT_RUN_SCRIPT_DIR).expanduser()
+        if task_run_dir.exists():
+            # Please see `kubernetes-ray.yml.j2` for more details.
+            this_task_run_script = task_run_dir / f'sky_job_{job_id}'
+            if this_task_run_script.exists():
+                this_task_run_script.unlink()
+                logger.info(f'Task run script {this_task_run_script} removed')
+            else:
+                logger.warning(
+                    f'Task run script {this_task_run_script} not found')
+
 
 if __name__ == '__main__':
+    if not pathlib.Path('/home/sky/.sky/k8s_container_ready').exists():
+        print('k8s_container_ready not found, exiting')
+        exit(0)
path: sky/serve/service.py
text: @user1:
can we add some comment for what this file for?

also, we should have a constant for this ;)

@user1:
Actually, does this file guaranteed to exist?

@author:
theoretically, yes. It's a sanity check.

@author:
Some rare scenerios, like the pvc mounting isn't work, may get into this.

@user1:
Then lets use assertion?

diff_hunk: @@ -33,14 +32,32 @@
 TAG_SKYPILOT_CLUSTER_NAME = 'skypilot-cluster-name'
 TAG_POD_INITIALIZED = 'skypilot-initialized'
 
+# Please be careful when changing this.
+# When mounting, Kubernetes changes the ownership of the parent directory
+# to root:root.
+# pylint: disable=line-too-long
+# See https://stackoverflow.com/questions/50818029/mounted-folder-created-as-root-instead-of-current-user-in-docker/50820023#50820023.
+DEPLOYMENT_VOLUME_MOUNTS = {
path: sky/provision/kubernetes/instance.py
text: @user1:
add high avail in the var name?

diff_hunk: @@ -650,6 +478,144 @@ def _create_namespaced_pod_with_retries(namespace: str, pod_spec: dict,
             raise e
 
 
+def _create_persistent_volume_claim(namespace: str, context: Optional[str],
+                                    pvc_name: str) -> None:
+    """Creates a persistent volume claim for SkyServe controller."""
+    try:
+        kubernetes.core_api(context).read_namespaced_persistent_volume_claim(
+            name=pvc_name, namespace=namespace)
+        return
+    except kubernetes.api_exception() as e:
+        if e.status != 404:  # Not found
+            raise
+
+    pvc_spec = {
+        'apiVersion': 'v1',
+        'kind': 'PersistentVolumeClaim',
+        'metadata': {
+            'name': pvc_name,
+        },
+        'spec': {
+            'accessModes': ['ReadWriteOnce'],
+            'resources': {
+                'requests': {
+                    'storage': '10Gi'  # TODO(andyl): use a constant here
+                }
+            }
+        }
+    }
+
+    kubernetes.core_api(context).create_namespaced_persistent_volume_claim(
+        namespace=namespace, body=pvc_spec)
+
+
+def _create_serve_controller_deployment(
+        pod_spec: Dict[str, Any], cluster_name_on_cloud: str, namespace: str,
+        context: Optional[str]) -> Dict[str, Any]:
+    """Creates a deployment for SkyServe controller with persistence."""
+    assert len(
+        pod_spec['spec']['containers']) == 1, 'Only one container is supported'
+
+    # The pod template part of pod_spec is used in the deployment
+    # spec.template.spec
+    spec = pod_spec.pop('spec')
+    container = spec['containers'][0]
+
+    if DEPLOYMENT_VOLUME_MOUNTS:
+        # TODO(andyl): use defaultdict
+        if spec.get('volumes') is None:
+            spec['volumes'] = []
+        if container.get('volumeMounts') is None:
+            container['volumeMounts'] = []
+
+    for volume_name, mount_path in DEPLOYMENT_VOLUME_MOUNTS.items():
+        pvc_name = _get_pvc_name(cluster_name_on_cloud, volume_name)
+        _create_persistent_volume_claim(namespace, context, pvc_name)
+        spec['volumes'].append({
+            'name': volume_name,
+            'persistentVolumeClaim': {
+                'claimName': pvc_name
+            }
+        })
+        container['volumeMounts'].append({
+            'mountPath': mount_path,
+            'name': volume_name
+        })
+
+    template_metadata = pod_spec.pop('metadata')
+    assert pod_spec.keys() == {'apiVersion', 'kind'
+                              }, (f'pod_spec has fields {pod_spec.keys()}')
+
+    deployment_labels = {
+        'app': cluster_name_on_cloud,
+    }
+    template_metadata['labels'].update(deployment_labels)
+
+    image_id = container['image']
+    assert image_id is not None, 'Image ID is None'
+    init_containers = spec.get('initContainers', [])
+    init_containers.append({
+        'name': 'init-copy-home',
+        'image': image_id,
+        'command': ['/bin/sh', '-c'],
+        'args': [
+            'echo "Copying home directory to /mnt/home"; '
+            'rsync -a /home/sky/ /mnt/home; '
+            'echo "Copy completed.";'
+            'ls -la /mnt/home'
path: sky/provision/kubernetes/instance.py
text: @user1:
Can we add some comments for why we need to rsync it to /mnt/home?

iiuc only /home/sky is enough?

@author:
Oh thats because we are copying from the docker container to pvc (to prevent pvc mounting overriding the directory that docker image provides). And /mnt/home is on the pvc, while /home/sky is in the docker container.

@user1:
Do you think we can directly use the mount mode instead of copy it?

@author:
Well we need to do this copying to prevent the docker image being overridden by pvc mounting.

@user1:
Could you elaborate?

@author:
Yes. Like when do a PVC mounting, the default behavior is that the pvc will take up this place, so the original directory inherited from the docker image is overriden. To prevent that, we need to do a copying before mounting.

@user1:
What is the problem of just letting it override?

@author:
Mentioned before. `.profile`, `skypilot_runtime`, these files. Should be used in container commands.

diff_hunk: @@ -748,7 +715,8 @@ def _create_pod_thread(i: int):
             pod_spec_copy['metadata']['labels'].update(constants.HEAD_NODE_TAGS)
             head_selector = head_service_selector(cluster_name_on_cloud)
             pod_spec_copy['metadata']['labels'].update(head_selector)
-            pod_spec_copy['metadata']['name'] = f'{cluster_name_on_cloud}-head'
+            pod_spec_copy['metadata'][
+                'name'] = f'{cluster_name_on_cloud}-head'  #!
path: sky/provision/kubernetes/instance.py
text: @user1:
what is this #!

@author:
typo. fixed

diff_hunk: @@ -2452,6 +2454,11 @@ def get_command_runners(self,
                     'Tried to use cached cluster info, but it\'s missing for '
                     f'cluster "{self.cluster_name}"')
             self._update_cluster_info()
+        # TODO(andyl): must force refresh here?
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
https://github.com/skypilot-org/skypilot/pull/4564#discussion_r1951800510