[k8s] SkyServe on Kubernetes

sky/backends/backend_utils.py

@@ -810,7 +810,7 @@ def write_cluster_config(
             if fnmatch.fnmatchcase(cluster_name, list(profile.keys())[0]):
                 remote_identity = list(profile.values())[0]
                 break
-    if remote_identity != 'LOCAL_CREDENTIALS':
+    if remote_identity != schemas.RemoteIdentityOptions.LOCAL_CREDENTIALS.value:
         if not cloud.supports_service_account_on_remote():
             raise exceptions.InvalidCloudConfigs(
                 'remote_identity: SERVICE_ACCOUNT is specified in '
@@ -2721,13 +2721,20 @@ def check_stale_runtime_on_remote(returncode: int, stderr: str,
 
 
 def get_endpoints(cluster: str,
-                  port: Optional[Union[int, str]] = None) -> Dict[int, str]:
+                  port: Optional[Union[int, str]] = None,
+                  skip_status_check: bool = False) -> Dict[int, str]:
     """Gets the endpoint for a given cluster and port number (endpoint).
 
     Args:
         cluster: The name of the cluster.
         port: The port number to get the endpoint for. If None, endpoints
             for all ports are returned.
+        skip_status_check: Whether to skip the status check for the cluster.
+            This is useful when the cluster is known to be in a INIT state
+            and the caller wants to query the endpoints. Used by serve
+            controller to query endpoints during cluster launch when multiple
+            services may be getting launched in parallel (and as a result,
+            the controller may be in INIT status due to a concurrent launch).
 
     Returns: A dictionary of port numbers to endpoints. If endpoint is None,
         the dictionary will contain all ports:endpoints exposed on the cluster.
@@ -2751,7 +2758,8 @@ def get_endpoints(cluster: str,
                                    refresh=False,
                                    cluster_names=[cluster])
     cluster_record = cluster_records[0]
-    if cluster_record['status'] != status_lib.ClusterStatus.UP:
+    if (not skip_status_check and
+            cluster_record['status'] != status_lib.ClusterStatus.UP):
         with ux_utils.print_exception_no_traceback():
             raise exceptions.ClusterNotUpError(
                 f'Cluster {cluster_record["name"]!r} '

sky/cli.py

@@ -52,7 +52,6 @@
 from sky import exceptions
 from sky import global_user_state
 from sky import jobs as managed_jobs
-from sky import provision as provision_lib
 from sky import serve as serve_lib
 from sky import sky_logging
 from sky import status_lib

sky/clouds/kubernetes.py

@@ -271,11 +271,13 @@ def make_deploy_resources_variables(
 
         remote_identity = skypilot_config.get_nested(
             ('kubernetes', 'remote_identity'), schemas.REMOTE_IDENTITY_DEFAULT)
-        if remote_identity == 'LOCAL_CREDENTIALS':
+        if (remote_identity ==
+                schemas.RemoteIdentityOptions.LOCAL_CREDENTIALS.value):
             # SA name doesn't matter since automounting credentials is disabled
             k8s_service_account_name = 'default'
             k8s_automount_sa_token = 'false'
-        elif remote_identity == 'SERVICE_ACCOUNT':
+        elif (remote_identity ==
+              schemas.RemoteIdentityOptions.SERVICE_ACCOUNT.value):
             # Use the default service account
             k8s_service_account_name = self.SKY_DEFAULT_SERVICE_ACCOUNT_NAME
             k8s_automount_sa_token = 'true'

sky/provision/kubernetes/utils.py

@@ -635,8 +635,8 @@ def is_kubeconfig_exec_auth() -> Tuple[bool, Optional[str]]:
 
     remote_identity = skypilot_config.get_nested(
         ('kubernetes', 'remote_identity'), schemas.REMOTE_IDENTITY_DEFAULT)
-    if ('exec' in user_details.get('user', {}) and
-            remote_identity == 'LOCAL_CREDENTIALS'):
+    if ('exec' in user_details.get('user', {}) and remote_identity
+            == schemas.RemoteIdentityOptions.LOCAL_CREDENTIALS.value):
         ctx_name = current_context['name']
         exec_msg = ('exec-based authentication is used for '
                     f'Kubernetes context {ctx_name!r}.'

sky/serve/core.py

@@ -255,7 +255,9 @@ def up(
             lb_port = serve_utils.load_service_initialization_result(
                 lb_port_payload)
             endpoint = backend_utils.get_endpoints(
-                controller_handle.cluster_name, lb_port)[lb_port]
+                controller_handle.cluster_name, lb_port,
+                skip_status_check=True).get(lb_port)
+            assert endpoint is not None, 'Did not get endpoint for controller.'
 
         sky_logging.print(
             f'{fore.CYAN}Service name: '

sky/utils/common_utils.py

@@ -106,7 +106,9 @@ def _is_valid_user_hash(user_hash: Optional[str]) -> bool:
     os.makedirs(os.path.dirname(_USER_HASH_FILE), exist_ok=True)
     if not force_fresh_hash:
         # Do not cache to file if force_fresh_hash is True since the file may
-        # be intentionally using a different hash.
+        # be intentionally using a different hash, e.g. we want to keep the
+        # user_hash for usage collection the same on the jobs/serve controller
+        # as users' local client.
         with open(_USER_HASH_FILE, 'w', encoding='utf-8') as f:
             f.write(user_hash)
     return user_hash

sky/utils/controller_utils.py

@@ -242,8 +242,9 @@ def _get_cloud_dependencies_installation_commands(
                 'fi" && '
                 # Install kubectl
                 '(command -v kubectl &>/dev/null || '
-                '(curl -LO "https://dl.k8s.io/release/$(curl -L -s '
-                'https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl" && '
+                '(curl -LO "https://dl.k8s.io/release/'
+                '$(curl -L -s https://dl.k8s.io/release/stable.txt)'
+                '/bin/linux/amd64/kubectl" && '
                 'sudo install -o root -g root -m 0755 '
                 'kubectl /usr/local/bin/kubectl))')
         if controller == Controllers.JOBS_CONTROLLER:

sky/utils/schemas.py

@@ -3,6 +3,7 @@
 Schemas conform to the JSON Schema specification as defined at
 https://json-schema.org/
 """
+import enum
 
 
 def _check_not_both_fields_present(field1: str, field2: str):
@@ -522,10 +523,26 @@ def get_cluster_schema():
     }
 }
 
+
+class RemoteIdentityOptions(enum.Enum):
+    """Enum for remote identity types.
+
+    Some clouds (e.g., AWS, Kubernetes) also allow string values for remote
+    identity, which map to the service account/role to use. Those are not
+    included in this enum.
+    """
+    LOCAL_CREDENTIALS = 'LOCAL_CREDENTIALS'
+    SERVICE_ACCOUNT = 'SERVICE_ACCOUNT'
+
+
+REMOTE_IDENTITY_DEFAULT = RemoteIdentityOptions.LOCAL_CREDENTIALS.value
+
 _REMOTE_IDENTITY_SCHEMA = {
     'remote_identity': {
         'type': 'string',
-        'case_insensitive_enum': ['LOCAL_CREDENTIALS', 'SERVICE_ACCOUNT']
+        'case_insensitive_enum': [
+            option.value for option in RemoteIdentityOptions
+        ]
     }
 }
 
@@ -562,8 +579,6 @@ def get_cluster_schema():
     },
 }
 
-REMOTE_IDENTITY_DEFAULT = 'LOCAL_CREDENTIALS'
-
 
 def get_config_schema():
     # pylint: disable=import-outside-toplevel

tests/skyserve/auto_restart.yaml

@@ -8,7 +8,6 @@ service:
 resources:
   ports: 8080
   cloud: gcp
-  zone: us-central1-a
   cpus: 2+
 
 workdir: examples/serve/http_server

tests/test_smoke.py

@@ -3244,7 +3244,7 @@ def test_skyserve_azure_http():
 
 
 @pytest.mark.kubernetes
-@pytest.mark.sky_serve
+@pytest.mark.serve
 def test_skyserve_kubernetes_http():
     """Test skyserve on Kubernetes"""
     name = _get_service_name()
@@ -3407,6 +3407,7 @@ def test_skyserve_user_bug_restart(generic_cloud: str):
 
 
 @pytest.mark.serve
+@pytest.mark.no_kubernetes  # Replicas on k8s may be running on the same node and have the same public IP
 def test_skyserve_load_balancer(generic_cloud: str):
     """Test skyserve load balancer round-robin policy"""
     name = _get_service_name()
@@ -3573,7 +3574,7 @@ def test_skyserve_fast_update(generic_cloud: str):
             f'{_SERVE_ENDPOINT_WAIT.format(name=name)}; curl -L http://$endpoint | grep "Hi, SkyPilot here"',
             f'sky serve update {name} --cloud {generic_cloud} --mode blue_green -y tests/skyserve/update/bump_version_after.yaml',
             # sleep to wait for update to be registered.
-            'sleep 120',
+            'sleep 30',
             # 2 on-deamnd (ready) + 1 on-demand (provisioning).
             (
                 _check_replica_in_status(
@@ -3587,7 +3588,7 @@ def test_skyserve_fast_update(generic_cloud: str):
             # Test rolling update
             f'sky serve update {name} --cloud {generic_cloud} -y tests/skyserve/update/bump_version_before.yaml',
             # sleep to wait for update to be registered.
-            'sleep 30',
+            'sleep 15',
             # 2 on-deamnd (ready) + 1 on-demand (shutting down).
             _check_replica_in_status(name, [(2, False, 'READY'),
                                             (1, False, 'SHUTTING_DOWN')]),

tests/test_yamls/minimal.yaml

@@ -2,6 +2,7 @@ name: min
 
 setup: |
   echo "running setup"
+  sudo apt-get install -y jq
 
 run: |
   conda env list