Repository: skypilot-org/skypilot

Issue:
[k8s] Using AppArmor causes provisioning failures on certain k8s clusters
User reported using file_mounts for bucket mounting in `sky launch` was causing provisioning to fail on their EKS cluster:

e.g.,
```
file_mounts:
  /data:
    name: romilb-sky-test
    source: ~/tmp-workdir
    mode: MOUNT

run: |
  ls -l /data
```

would cause:

```
HTTP response headers: HTTPHeaderDict({'Audit-Id': '42ea60ef-3305-43e9-a554-ef028e808b2f', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Kubernetes-Pf-Flowschema-Uid': '7f0f3024-9f2f-4abb-af09-80757fbbd065', 'X-Kubernetes-Pf-Prioritylevel-Uid': 'a6ea3db1-56e6-4506-b031-942b90da496c', 'Date': 'Wed, 23 Oct 2024 19:11:07 GMT', 'Transfer-Encoding': 'chunked'})
(xxx-nemo-train-eks, pid=1713) W 10-23 19:11:07 instance.py:607] HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"Pod \"xxx-nemo-train-eks-1-4ce9-9a3a-worker\" is invalid: spec: Forbidden: pod updates may not change fields other than `spec.containers[*].image`,`spec.initContainers[*].image`,`spec.activeDeadlineSeconds`,`spec.tolerations` (only additions to existing tolerations),`spec.terminationGracePeriodSeconds` (allow it to be set to 1 if it was previously negative)\n  core.PodSpec{\n  \tVolumes:        {{Name: \"secret-volume\", VolumeSource: {Secret: \u0026{SecretName: \"sky-ssh-keys\", DefaultMode: \u0026420}}}, {Name: \"dshm\", VolumeSource: {EmptyDir: \u0026{Medium: \"Memory\"}}}, {Name: \"dev-fuse\", VolumeSource: {HostPath: \u0026{Path: \"/dev/fuse\", Type: \u0026\"\"}}}, {Name: \"nvme\", VolumeSource: {HostPath: \u0026{Path: \"/nvme\", Type: \u0026\"Directory\"}}}, ...},\n  \tInitContainers: nil,\n  \tContainers: []core.Container{\n  \t\t{\n  \t\t\t... // 18 identical fields\n  \t\t\tTerminationMessagePolicy: \"File\",\n  \t\t\tImagePullPolicy:          \"IfNotPresent\",\n  \t\t\tSecurityContext: \u0026core.SecurityContext{\n  \t\t\t\t... // 9 identical fields\n  \t\t\t\tProcMount:       nil,\n  \t\t\t\tSeccompProfile:  nil,\n- \t\t\t\tAppArmorProfile: \u0026core.AppArmorProfile{Type: \"Unconfined\"},\n+ \t\t\t\tAppArmorProfile: nil,\n  \t\t\t},\n  \t\t\tStdin:     false,\n  \t\t\tStdinOnce: false,\n  \t\t\tTTY:       false,\n  \t\t},\n  \t},\n  \tEphemeralContainers: nil,\n  \tRestartPolicy:       \"Never\",\n  \t... // 28 identical fields\n  }\n","reason":"Invalid","details":{"name":"xxx-nemo-train-eks-1-4ce9-9a3a-worker","kind":"Pod","causes":[{"reason":"FieldValueForbidden","message":"Forbidden: pod updates may not change fields other than `spec.containers[*].image`,`spec.initContainers[*].image`,`spec.activeDeadlineSeconds`,`spec.tolerations` (only additions to existing tolerations),`spec.terminationGracePeriodSeconds` (allow it to be set to 1 if it was previously negative)\n  core.PodSpec{\n  \tVolumes:        {{Name: \"secret-volume\", VolumeSource: {Secret: \u0026{SecretName: \"sky-ssh-keys\", DefaultMode: \u0026420}}}, {Name: \"dshm\", VolumeSource: {EmptyDir: \u0026{Medium: \"Memory\"}}}, {Name: \"dev-fuse\", VolumeSource: {HostPath: \u0026{Path: \"/dev/fuse\", Type: \u0026\"\"}}}, {Name: \"nvme\", VolumeSource: {HostPath: \u0026{Path: \"/nvme\", Type: \u0026\"Directory\"}}}, ...},\n  \tInitContainers: nil,\n  \tContainers: []core.Container{\n  \t\t{\n  \t\t\t... // 18 identical fields\n  \t\t\tTerminationMessagePolicy: \"File\",\n  \t\t\tImagePullPolicy:          \"IfNotPresent\",\n  \t\t\tSecurityContext: \u0026core.SecurityContext{\n  \t\t\t\t... // 9 identical fields\n  \t\t\t\tProcMount:       nil,\n  \t\t\t\tSeccompProfile:  nil,\n- \t\t\t\tAppArmorProfile: \u0026core.AppArmorProfile{Type: \"Unconfined\"},\n+ \t\t\t\tAppArmorProfile: nil,\n  \t\t\t},\n  \t\t\tStdin:     false,\n  \t\t\tStdinOnce: false,\n  \t\t\tTTY:       false,\n  \t\t},\n  \t},\n  \tEphemeralContainers: nil,\n  \tRestartPolicy:       \"Never\",\n  \t... // 28 identical fields\n  }\n","field":"spec"}]},"code":422}
```

The error is complaining about removal of the AppArmorProfile cannot be done once the pod is already running (`pod updates may not change fields`)
```
- AppArmorProfile: &core.AppArmorProfile{Type: "Unconfined"},
+ AppArmorProfile: nil,
```

The error was traced to the use of `container.apparmor.security.beta.kubernetes.io/ray-node: unconfined` in our kubernetes template when storage is used:

https://github.com/skypilot-org/skypilot/blob/e832dde2c5a7f9ba9e141afad874054deb15732c/sky/templates/kubernetes-ray.yml.j2#L274-L278

This is generally required for FUSE mounting (tested on GKE, EKS and Rancher). However, on the user's cluster, this annotation is not required for fuse mounting and Mandatory Access Control (MAC) is not enforced on their cluster. Using this annotation would cause the above error.

It's unclear if there's an admission controller or something else removing/updating the pod. 

User info:
```
System Info:                                                                                                           │
│   Machine ID:                 ec2f818ba21dddda0dc3507fe141bd49                                                         │
│   System UUID:                ec2f818b-a21d-ddda-0dc3-507fe141bd49                                                     │
│   Boot ID:                    a80b2cdb-f82a-4962-97b3-80a98a88195e                                                     │
│   Kernel Version:             6.8.0-1015-aws                                                                           │
│   OS Image:                   Ubuntu 22.04.5 LTS                                                                       │
│   Operating System:           linux                                                                                    │
│   Architecture:               amd64                                                                                    │
│   Container Runtime Version:  containerd://1.7.12                                                                      │
│   Kubelet Version:            v1.30.2                                                                                  │
│   Kube-Proxy Version:         v1.30.2   
```

```
sky -c
bc30c0b2e5c92dc00e52c6f4240b603fad4b258a
```


Discussion:

Patch:
diff --git a/sky/check.py b/sky/check.py
index 9ac28487..dcaa349d 100644
--- a/sky/check.py
+++ b/sky/check.py
@@ -1,4 +1,5 @@
 """Credential checks: check cloud credentials and enable clouds."""
+import os
 import traceback
 from types import ModuleType
 from typing import Dict, Iterable, List, Optional, Tuple, Union
@@ -194,19 +195,25 @@ def get_cached_enabled_clouds_or_refresh(
 def get_cloud_credential_file_mounts(
         excluded_clouds: Optional[Iterable[sky_clouds.Cloud]]
 ) -> Dict[str, str]:
-    """Returns the files necessary to access all enabled clouds.
+    """Returns the files necessary to access all clouds.
 
     Returns a dictionary that will be added to a task's file mounts
     and a list of patterns that will be excluded (used as rsync_exclude).
     """
-    enabled_clouds = get_cached_enabled_clouds_or_refresh()
+    # Uploading credentials for all clouds instead of only sky check
+    # enabled clouds because users may have partial credentials for some
+    # clouds to access their specific resources (e.g. cloud storage) but
+    # not have the complete credentials to pass sky check.
+    clouds = sky_clouds.CLOUD_REGISTRY.values()
     file_mounts = {}
-    for cloud in enabled_clouds:
+    for cloud in clouds:
         if (excluded_clouds is not None and
                 sky_clouds.cloud_in_iterable(cloud, excluded_clouds)):
             continue
         cloud_file_mounts = cloud.get_credential_file_mounts()
-        file_mounts.update(cloud_file_mounts)
+        for remote_path, local_path in cloud_file_mounts.items():
+            if os.path.exists(os.path.expanduser(local_path)):
+                file_mounts[remote_path] = local_path
     # Currently, get_cached_enabled_clouds_or_refresh() does not support r2 as
     # only clouds with computing instances are marked as enabled by skypilot.
     # This will be removed when cloudflare/r2 is added as a 'cloud'.
diff --git a/sky/clouds/azure.py b/sky/clouds/azure.py
index adffd32a..0852c993 100644
--- a/sky/clouds/azure.py
+++ b/sky/clouds/azure.py
@@ -15,6 +15,7 @@ from sky import exceptions
 from sky import sky_logging
 from sky.adaptors import azure
 from sky.clouds import service_catalog
+from sky.clouds.utils import azure_utils
 from sky.utils import common_utils
 from sky.utils import resources_utils
 from sky.utils import ux_utils
@@ -36,6 +37,15 @@ _MAX_IDENTITY_FETCH_RETRY = 10
 
 _DEFAULT_AZURE_UBUNTU_HPC_IMAGE_GB = 30
 _DEFAULT_AZURE_UBUNTU_2004_IMAGE_GB = 150
+_DEFAULT_SKYPILOT_IMAGE_GB = 30
+
+_DEFAULT_CPU_IMAGE_ID = 'skypilot:custom-cpu-ubuntu-v2'
+_DEFAULT_GPU_IMAGE_ID = 'skypilot:custom-gpu-ubuntu-v2'
+_DEFAULT_V1_IMAGE_ID = 'skypilot:custom-gpu-ubuntu-v1'
+_DEFAULT_GPU_K80_IMAGE_ID = 'skypilot:k80-ubuntu-2004'
+_FALLBACK_IMAGE_ID = 'skypilot:gpu-ubuntu-2204'
+
+_COMMUNITY_IMAGE_PREFIX = '/CommunityGalleries'
 
 
 def _run_output(cmd):
@@ -132,29 +142,56 @@ class Azure(clouds.Cloud):
         cost += 0.0
         return cost
 
+    @classmethod
+    def get_default_instance_type(
+            cls,
+            cpus: Optional[str] = None,
+            memory: Optional[str] = None,
+            disk_tier: Optional[resources_utils.DiskTier] = None
+    ) -> Optional[str]:
+        return service_catalog.get_default_instance_type(cpus=cpus,
+                                                         memory=memory,
+                                                         disk_tier=disk_tier,
+                                                         clouds='azure')
+
     @classmethod
     def get_image_size(cls, image_id: str, region: Optional[str]) -> float:
-        if region is None:
-            # The region used here is only for where to send the query,
-            # not the image location. Azure's image is globally available.
-            region = 'eastus'
-        is_skypilot_image_tag = False
+        # Process skypilot images.
         if image_id.startswith('skypilot:'):
-            is_skypilot_image_tag = True
             image_id = service_catalog.get_image_id_from_tag(image_id,
                                                              clouds='azure')
-        image_id_splitted = image_id.split(':')
-        if len(image_id_splitted) != 4:
-            with ux_utils.print_exception_no_traceback():
-                raise ValueError(f'Invalid image id: {image_id}. Expected '
-                                 'format: <publisher>:<offer>:<sku>:<version>')
-        publisher, offer, sku, version = image_id_splitted
-        if is_skypilot_image_tag:
-            if offer == 'ubuntu-hpc':
-                return _DEFAULT_AZURE_UBUNTU_HPC_IMAGE_GB
+            if image_id.startswith(_COMMUNITY_IMAGE_PREFIX):
+                # Avoid querying the image size from Azure as
+                # all skypilot custom images have the same size.
+                return _DEFAULT_SKYPILOT_IMAGE_GB
             else:
-                return _DEFAULT_AZURE_UBUNTU_2004_IMAGE_GB
+                publisher, offer, sku, version = image_id.split(':')
+                if offer == 'ubuntu-hpc':
+                    return _DEFAULT_AZURE_UBUNTU_HPC_IMAGE_GB
+                else:
+                    return _DEFAULT_AZURE_UBUNTU_2004_IMAGE_GB
+
+        # Process user-specified images.
+        azure_utils.validate_image_id(image_id)
         compute_client = azure.get_client('compute', cls.get_project_id())
+
+        # Community gallery image.
+        if image_id.startswith(_COMMUNITY_IMAGE_PREFIX):
+            if region is None:
+                return 0.0
+            _, _, gallery_name, _, image_name = image_id.split('/')
+            try:
+                return azure_utils.get_community_image_size(
+                    compute_client, gallery_name, image_name, region)
+            except exceptions.ResourcesUnavailableError:
+                return 0.0
+
+        # Marketplace image
+        if region is None:
+            # The region used here is only for where to send the query,
+            # not the image location. Marketplace image is globally available.
+            region = 'eastus'
+        publisher, offer, sku, version = image_id.split(':')
         try:
             image = compute_client.virtual_machine_images.get(
                 region, publisher, offer, sku, version)
@@ -176,40 +213,23 @@ class Azure(clouds.Cloud):
         size_in_gb = size_in_bytes / (1024**3)
         return size_in_gb
 
-    @classmethod
-    def get_default_instance_type(
-            cls,
-            cpus: Optional[str] = None,
-            memory: Optional[str] = None,
-            disk_tier: Optional[resources_utils.DiskTier] = None
-    ) -> Optional[str]:
-        return service_catalog.get_default_instance_type(cpus=cpus,
-                                                         memory=memory,
-                                                         disk_tier=disk_tier,
-                                                         clouds='azure')
-
     def _get_default_image_tag(self, gen_version, instance_type) -> str:
         # ubuntu-2004 v21.08.30, K80 requires image with old NVIDIA driver version
         acc = self.get_accelerators_from_instance_type(instance_type)
         if acc is not None:
             acc_name = list(acc.keys())[0]
             if acc_name == 'K80':
-                return 'skypilot:k80-ubuntu-2004'
-
-        # ubuntu-2004 v21.11.04, the previous image we used in the past for
-        # V1 HyperV instance before we change default image to ubuntu-hpc.
+                return _DEFAULT_GPU_K80_IMAGE_ID
+        # About Gen V1 vs V2:
         # In Azure, all instances with K80 (Standard_NC series), some
         # instances with M60 (Standard_NV series) and some cpu instances
-        # (Basic_A, Standard_D, ...) are V1 instance. For these instances,
-        # we use the previous image.
+        # (Basic_A, Standard_D, ...) are V1 instance.
+        # All A100 instances are V2.
         if gen_version == 'V1':
-            return 'skypilot:v1-ubuntu-2004'
-
-        # nvidia-driver: 535.54.03, cuda: 12.2
-        # see: https://github.com/Azure/azhpc-images/releases/tag/ubuntu-hpc-20230803
-        # All A100 instances is of gen2, so it will always use
-        # the latest ubuntu-hpc:2204 image.
-        return 'skypilot:gpu-ubuntu-2204'
+            return _DEFAULT_V1_IMAGE_ID
+        if acc is None:
+            return _DEFAULT_CPU_IMAGE_ID
+        return _DEFAULT_GPU_IMAGE_ID
 
     @classmethod
     def regions_with_offering(cls, instance_type: str,
@@ -302,17 +322,34 @@ class Azure(clouds.Cloud):
             else:
                 assert region_name in resources.image_id, resources.image_id
                 image_id = resources.image_id[region_name]
+
+        # Checked basic image syntax in resources.py
         if image_id.startswith('skypilot:'):
             image_id = service_catalog.get_image_id_from_tag(image_id,
                                                              clouds='azure')
-        # Already checked in resources.py
-        publisher, offer, sku, version = image_id.split(':')
-        image_config = {
-            'image_publisher': publisher,
-            'image_offer': offer,
-            'image_sku': sku,
-            'image_version': version,
-        }
+            # Fallback if image does not exist in the specified region.
+            # Putting fallback here instead of at image validation
+            # when creating the resource because community images are
+            # regional so we need the correct region when we check whether
+            # the image exists.
+            if image_id.startswith(
+                    _COMMUNITY_IMAGE_PREFIX
+            ) and region_name not in azure_catalog.COMMUNITY_IMAGE_AVAILABLE_REGIONS:
+                logger.info(f'Azure image {image_id} does not exist in region '
+                            f'{region_name} so use the fallback image instead.')
+                image_id = service_catalog.get_image_id_from_tag(
+                    _FALLBACK_IMAGE_ID, clouds='azure')
+
+        if image_id.startswith(_COMMUNITY_IMAGE_PREFIX):
+            image_config = {'community_gallery_image_id': image_id}
+        else:
+            publisher, offer, sku, version = image_id.split(':')
+            image_config = {
+                'image_publisher': publisher,
+                'image_offer': offer,
+                'image_sku': sku,
+                'image_version': version,
+            }
 
         # Setup the A10 nvidia driver.
         need_nvidia_driver_extension = (acc_dict is not None and
@@ -380,7 +417,6 @@ class Azure(clouds.Cloud):
         # Setting disk performance tier for high disk tier.
         if disk_tier == resources_utils.DiskTier.HIGH:
             resources_vars['disk_performance_tier'] = 'P50'
-
         return resources_vars
 
     def _get_feasible_launchable_resources(
diff --git a/sky/clouds/oci.py b/sky/clouds/oci.py
index 810e43fe..0feda467 100644
--- a/sky/clouds/oci.py
+++ b/sky/clouds/oci.py
@@ -468,8 +468,12 @@ class OCI(clouds.Cloud):
             api_key_file = oci_cfg[
                 'key_file'] if 'key_file' in oci_cfg else 'BadConf'
             sky_cfg_file = oci_utils.oci_config.get_sky_user_config_file()
+        # Must catch ImportError before any oci_adaptor.oci.exceptions
+        # because oci_adaptor.oci.exceptions can throw ImportError.
         except ImportError:
             return {}
+        except oci_adaptor.oci.exceptions.ConfigFileNotFound:
+            return {}
 
         # OCI config and API key file are mandatory
         credential_files = [oci_cfg_file, api_key_file]
diff --git a/sky/clouds/service_catalog/azure_catalog.py b/sky/clouds/service_catalog/azure_catalog.py
index 2d323cba..867141f7 100644
--- a/sky/clouds/service_catalog/azure_catalog.py
+++ b/sky/clouds/service_catalog/azure_catalog.py
@@ -7,11 +7,29 @@ import re
 from typing import Dict, List, Optional, Tuple
 
 from sky import clouds as cloud_lib
+from sky import sky_logging
 from sky.clouds import Azure
 from sky.clouds.service_catalog import common
 from sky.utils import resources_utils
 from sky.utils import ux_utils
 
+logger = sky_logging.init_logger(__name__)
+
+# This list should match the list of regions in
+# skypilot image generation Packer script's replication_regions
+# sky/clouds/service_catalog/images/skypilot-azure-cpu-ubuntu.pkr.hcl
+COMMUNITY_IMAGE_AVAILABLE_REGIONS = {
+    'centralus',
+    'eastus',
+    'eastus2',
+    'northcentralus',
+    'southcentralus',
+    'westcentralus',
+    'westus',
+    'westus2',
+    'westus3',
+}
+
 # The frequency of pulling the latest catalog from the cloud provider.
 # Though the catalog update is manual in our skypilot-catalog repo, we
 # still want to pull the latest catalog periodically to make sure the
@@ -176,9 +194,16 @@ def list_accelerators(
 
 def get_image_id_from_tag(tag: str, region: Optional[str]) -> Optional[str]:
     """Returns the image id from the tag."""
-    # Azure images are not region-specific.
-    del region  # Unused.
-    return common.get_image_id_from_tag_impl(_image_df, tag, None)
+    global _image_df
+    image_id = common.get_image_id_from_tag_impl(_image_df, tag, region)
+    if image_id is None:
+        # Refresh the image catalog and try again, if the image tag is not
+        # found.
+        logger.debug('Refreshing the image catalog and trying again.')
+        _image_df = common.read_catalog('azure/images.csv',
+                                        pull_frequency_hours=0)
+        image_id = common.get_image_id_from_tag_impl(_image_df, tag, region)
+    return image_id
 
 
 def is_image_tag_valid(tag: str, region: Optional[str]) -> bool:
diff --git a/sky/clouds/service_catalog/images/README.md b/sky/clouds/service_catalog/images/README.md
index 31ce7c6d..3bdcbf86 100644
--- a/sky/clouds/service_catalog/images/README.md
+++ b/sky/clouds/service_catalog/images/README.md
@@ -10,42 +10,58 @@ packer init plugins.pkr.hcl
 3. Setup cloud credentials
 
 ## Generate Images
-```bash
-export CLOUD=gcp    # Update this
-export TYPE=gpu    # Update this
-export IMAGE=skypilot-${CLOUD}-${TYPE}-ubuntu
-packer build ${IMAGE}.pkr.hcl
-```
-You will see the image ID after the build is complete.
-
-FYI time to packer build an image:
-
+FYI time to packer build images:
 | Cloud | Type | Approx. Time |
 |-------|------|------------------------|
 | AWS   | GPU  | 15 min          |
 | AWS   | CPU  | 10 min          |
 | GCP   | GPU  | 16 min          |
 | GCP   | CPU  | 5 min          |
+| Azure | GPU  | 35 min          |
+| Azure | CPU  | 25 min          |
 
 ### GCP
+1. Build a single global image.
+```bash
+export TYPE=gpu  # Update this
+export IMAGE=skypilot-gcp-${TYPE}-ubuntu
+packer build ${IMAGE}.pkr.hcl
+```
+2. Make the image public
 ```bash
-export IMAGE_NAME=skypilot-gcp-cpu-ubuntu-20241011003407  # Update this
-
 # Make image public
+export IMAGE_NAME=skypilot-gcp-cpu-ubuntu-xxx  # Update this
 export IMAGE_ID=projects/sky-dev-465/global/images/${IMAGE_NAME}
 gcloud compute images add-iam-policy-binding ${IMAGE_NAME} --member='allAuthenticatedUsers' --role='roles/compute.imageUser'
 ```
 
 ### AWS
-1. Generate images for all regions
+1. Generate the source image for a single region.
+```bash
+export TYPE=gpu  # Update this
+export IMAGE=skypilot-aws-${TYPE}-ubuntu
+packer build ${IMAGE}.pkr.hcl
+```
+2. Copy images to all regions
 ```bash
 export IMAGE_ID=ami-0b31b24524afa8e47   # Update this
-
 python aws_utils/image_gen.py --image-id ${IMAGE_ID} --processor ${TYPE}
 ```
-2. Add fallback images if any region failed \
+3. Add fallback images if any region failed \
 Look for "NEED_FALLBACK" in the output `images.csv` and edit. (You can use public [ubuntu images](https://cloud-images.ubuntu.com/locator/ec2/) as fallback.)
 
+### Azure
+1. Generate a client secret for packer [here](https://portal.azure.com/?feature.msaljs=true#view/Microsoft_AAD_RegisteredApps/ApplicationMenuBlade/~/Credentials/appId/1d249f23-c22e-4d02-b62b-a6827bd113fe/isMSAApp~/false).
+```bash
+export SECRET=xxxxxx  # Update this
+```
+2. Build and copy images for all regions and both VM generations (1 and 2).
+```bash
+export VM_GENERATION=2  # Update this
+packer build -force --var vm_generation=${VM_GENERATION} --var client_secret=${SECRET} skypilot-azure-cpu-ubuntu.pkr.hcl
+packer build --var client_secret=${SECRET} skypilot-azure-gpu-ubuntu.pkr.hcl
+```
+
 ## Test Images
 1. Minimal GPU test: `sky launch --image ${IMAGE_ID} --gpus=L4:1 --cloud ${CLOUD}` then run `nvidia-smi` in the launched instance.
 2. Update the image ID in `sky/clouds/gcp.py` and run the test:
@@ -60,13 +76,16 @@ pytest tests/test_smoke.py::test_cancel_gcp
 Submit a PR to update [`SkyPilot Catalog`](https://github.com/skypilot-org/skypilot-catalog/tree/master/catalogs) then clean up the old images to avoid extra iamge storage fees.
 
 ### GCP
-1. Example PR: [#86](https://github.com/skypilot-org/skypilot-catalog/pull/86)
-2. Go to console and delete old images.
+1. Update Catalog with new images: [example PR](https://github.com/skypilot-org/skypilot-catalog/pull/86)
+2. Go to [GCP console](https://console.cloud.google.com/compute/images?tab=images&project=sky-dev-465) and delete old images.
 
 ### AWS
 1. Copy the old custom image rows from Catalog's existing `images.csv` to a local `images.csv` in this folder.
-2. Update Catalog with new images. Example PR: [#89](https://github.com/skypilot-org/skypilot-catalog/pull/89)
+2. Update Catalog with new images: [example PR](https://github.com/skypilot-org/skypilot-catalog/pull/89)
 3. Delete AMIs across regions by running
 ```bash
 python aws_utils/image_delete.py --tag ${TAG}
 ```
+
+### Azure
+1. Update Catalog with new images: [example PR](https://github.com/skypilot-org/skypilot-catalog/pull/92)
diff --git a/sky/clouds/service_catalog/images/skypilot-aws-cpu-ubuntu.pkr.hcl b/sky/clouds/service_catalog/images/skypilot-aws-cpu-ubuntu.pkr.hcl
index c21fbf51..5b049cf3 100644
--- a/sky/clouds/service_catalog/images/skypilot-aws-cpu-ubuntu.pkr.hcl
+++ b/sky/clouds/service_catalog/images/skypilot-aws-cpu-ubuntu.pkr.hcl
@@ -22,9 +22,9 @@ source "amazon-ebs" "cpu-ubuntu" {
     owners      = ["099720109477"]
   }
   launch_block_device_mappings {
-    device_name = "/dev/sda1"
-    volume_size = 8
-    volume_type = "gp2"
+    device_name           = "/dev/sda1"
+    volume_size           = 8
+    volume_type           = "gp2"
     delete_on_termination = true
   }
 }
diff --git a/sky/clouds/service_catalog/images/skypilot-aws-gpu-ubuntu.pkr.hcl b/sky/clouds/service_catalog/images/skypilot-aws-gpu-ubuntu.pkr.hcl
index c4a8efac..45799877 100644
--- a/sky/clouds/service_catalog/images/skypilot-aws-gpu-ubuntu.pkr.hcl
+++ b/sky/clouds/service_catalog/images/skypilot-aws-gpu-ubuntu.pkr.hcl
@@ -22,9 +22,9 @@ source "amazon-ebs" "gpu-ubuntu" {
     owners      = ["099720109477"]
   }
   launch_block_device_mappings {
-    device_name = "/dev/sda1"
-    volume_size = 30
-    volume_type = "gp2"
+    device_name           = "/dev/sda1"
+    volume_size           = 30
+    volume_type           = "gp2"
     delete_on_termination = true
   }
 }
diff --git a/sky/clouds/service_catalog/images/skypilot-azure-cpu-ubuntu.pkr.hcl b/sky/clouds/service_catalog/images/skypilot-azure-cpu-ubuntu.pkr.hcl
new file mode 100644
index 00000000..2a07c41d
--- /dev/null
+++ b/sky/clouds/service_catalog/images/skypilot-azure-cpu-ubuntu.pkr.hcl
@@ -0,0 +1,72 @@
+variable "client_secret" {
+  type        = string
+  description = "The client secret for the packer client registered in Azure (see Azure app registration)"
+}
+
+variable "vm_generation" {
+  type        = number
+  description = "Azure's VM generation, currently support 1 or 2"
+}
+
+locals {
+  timestamp = regex_replace(timestamp(), "[- TZ:]", "")
+  version   = formatdate("YY.MM.DD", timestamp())
+}
+
+source "azure-arm" "cpu-ubuntu" {
+  managed_image_resource_group_name = "skypilot-images"
+  managed_image_name                = "skypilot-azure-cpu-ubuntu-${local.timestamp}"
+
+  subscription_id = "59d8c23c-7ef5-42c7-b2f3-a919ad8026a7"
+  tenant_id       = "7c81f068-46f8-4b26-9a46-2fbec2287e3d"
+  client_id       = "1d249f23-c22e-4d02-b62b-a6827bd113fe"
+  client_secret   = var.client_secret
+
+  os_type         = "Linux"
+  image_publisher = "Canonical"
+  image_offer     = "0001-com-ubuntu-server-jammy"
+  image_sku       = var.vm_generation == 1 ? "22_04-lts" : "22_04-lts-gen2"
+  location        = "centralus"
+  vm_size         = var.vm_generation == 1 ? "Standard_D1_v2" : "Standard_B2s"
+  ssh_username    = "azureuser"
+  azure_tags = {
+    Created_by = "packer"
+    Purpose    = "skypilot"
+  }
+
+  shared_image_gallery_destination {
+    subscription   = "59d8c23c-7ef5-42c7-b2f3-a919ad8026a7"
+    resource_group = "skypilot-images"
+    gallery_name   = "skypilot_image_gallery"
+    image_name     = "skypilot-cpu-gen${var.vm_generation}"
+    image_version  = "${local.version}"
+    replication_regions = [
+      "centralus",
+      "eastus",
+      "eastus2",
+      "northcentralus",
+      "southcentralus",
+      "westcentralus",
+      "westus",
+      "westus2",
+      "westus3"
+    ]
+  }
+}
+
+build {
+  name    = "azure-cpu-ubuntu-build"
+  sources = ["sources.azure-arm.cpu-ubuntu"]
+  provisioner "shell" {
+    script = "./provisioners/docker.sh"
+  }
+  provisioner "shell" {
+    script = "./provisioners/skypilot.sh"
+  }
+  provisioner "shell" {
+    environment_vars = [
+      "CLOUD=azure",
+    ]
+    script = "./provisioners/cloud.sh"
+  }
+}
diff --git a/sky/clouds/service_catalog/images/skypilot-azure-gpu-ubuntu.pkr.hcl b/sky/clouds/service_catalog/images/skypilot-azure-gpu-ubuntu.pkr.hcl
new file mode 100644
index 00000000..97c99b24
--- /dev/null
+++ b/sky/clouds/service_catalog/images/skypilot-azure-gpu-ubuntu.pkr.hcl
@@ -0,0 +1,78 @@
+variable "client_secret" {
+  type        = string
+  description = "The client secret for the packer client registered in Azure (see Azure app registration)"
+}
+
+variable "vm_generation" {
+  type        = number
+  description = "Azure's VM generation, currently support 1 or 2"
+}
+
+locals {
+  timestamp = regex_replace(timestamp(), "[- TZ:]", "")
+  version   = formatdate("YY.MM.DD", timestamp())
+}
+
+source "azure-arm" "gpu-ubuntu" {
+  managed_image_resource_group_name = "skypilot-images"
+  managed_image_name                = "skypilot-azure-gpu-ubuntu-${local.timestamp}"
+
+  subscription_id = "59d8c23c-7ef5-42c7-b2f3-a919ad8026a7"
+  tenant_id       = "7c81f068-46f8-4b26-9a46-2fbec2287e3d"
+  client_id       = "1d249f23-c22e-4d02-b62b-a6827bd113fe"
+  client_secret   = var.client_secret
+
+  os_type         = "Linux"
+  image_publisher = "Canonical"
+  image_offer     = "0001-com-ubuntu-server-jammy"
+  image_sku       = var.vm_generation == 1 ? "22_04-lts" : "22_04-lts-gen2"
+  location        = var.vm_generation == 1 ? "eastus" : "centralus"
+  vm_size         = var.vm_generation == 1 ? "Standard_NC4as_T4_v3" : "Standard_NC24ads_A100_v4"
+  ssh_username    = "azureuser"
+  azure_tags = {
+    Created_by = "packer"
+    Purpose    = "skypilot"
+  }
+
+  shared_image_gallery_destination {
+    subscription   = "59d8c23c-7ef5-42c7-b2f3-a919ad8026a7"
+    resource_group = "skypilot-images"
+    gallery_name   = var.vm_generation == 1 ? "skypilot_images": "skypilot_image_gallery"
+    image_name     = "skypilot-gpu-gen${var.vm_generation}"
+    image_version  = "${local.version}"
+    replication_regions = [
+      "centralus",
+      "eastus",
+      "eastus2",
+      "northcentralus",
+      "southcentralus",
+      "westcentralus",
+      "westus",
+      "westus2",
+      "westus3"
+    ]
+  }
+}
+
+build {
+  name    = "azure-gpu-ubuntu-build"
+  sources = ["sources.azure-arm.gpu-ubuntu"]
+  provisioner "shell" {
+    script = "./provisioners/docker.sh"
+  }
+  provisioner "shell" {
+    script = "./provisioners/cuda.sh"
+  }
+  provisioner "shell" {
+    script = "./provisioners/nvidia-container-toolkit.sh"
+  }
+  provisioner "shell" {
+    script = "./provisioners/skypilot.sh"
+  }
+  provisioner "shell" {
+    environment_vars = [
+      "CLOUD=azure",
+    ]
+    script = "./provisioners/cloud.sh"
+  }
+}
diff --git a/sky/clouds/utils/azure_utils.py b/sky/clouds/utils/azure_utils.py
new file mode 100644
index 00000000..83b86f4d
--- /dev/null
+++ b/sky/clouds/utils/azure_utils.py
@@ -0,0 +1,91 @@
+"""Utilies for Azure"""
+
+import typing
+
+from sky import exceptions
+from sky.adaptors import azure
+from sky.utils import ux_utils
+
+if typing.TYPE_CHECKING:
+    from azure.mgmt import compute as azure_compute
+    from azure.mgmt.compute import models as azure_compute_models
+
+
+def validate_image_id(image_id: str):
+    """Check if the image ID has a valid format.
+
+    Raises:
+        ValueError: If the image ID is invalid.
+    """
+    image_id_colon_splitted = image_id.split(':')
+    image_id_slash_splitted = image_id.split('/')
+    if len(image_id_slash_splitted) != 5 and len(image_id_colon_splitted) != 4:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError(
+                f'Invalid image id for Azure: {image_id}. Expected format: \n'
+                '* Marketplace image ID: <publisher>:<offer>:<sku>:<version>\n'
+                '* Community image ID: '
+                '/CommunityGalleries/<gallery-name>/Images/<image-name>')
+    if len(image_id_slash_splitted) == 5:
+        _, gallery_type, _, image_type, _ = image_id.split('/')
+        if gallery_type != 'CommunityGalleries' or image_type != 'Images':
+            with ux_utils.print_exception_no_traceback():
+                raise ValueError(
+                    f'Invalid community image id for Azure: {image_id}.\n'
+                    'Expected format: '
+                    '/CommunityGalleries/<gallery-name>/Images/<image-name>')
+
+
+def get_community_image(
+        compute_client: 'azure_compute.ComputeManagementClient', image_id: str,
+        region: str) -> 'azure_compute_models.CommunityGalleryImage':
+    """Get community image from cloud.
+
+    Args:
+        image_id: /CommunityGalleries/<gallery-name>/Images/<image-name>
+    Raises:
+        ResourcesUnavailableError
+    """
+    try:
+        _, _, gallery_name, _, image_name = image_id.split('/')
+        return compute_client.community_gallery_images.get(
+            location=region,
+            public_gallery_name=gallery_name,
+            gallery_image_name=image_name)
+    except azure.exceptions().AzureError as e:
+        raise exceptions.ResourcesUnavailableError(
+            f'Community image {image_id} does not exist in region {region}.'
+        ) from e
+
+
+def get_community_image_size(
+        compute_client: 'azure_compute.ComputeManagementClient',
+        gallery_name: str, image_name: str, region: str) -> float:
+    """Get the size of the community image from cloud.
+
+    Args:
+        image_id: /CommunityGalleries/<gallery-name>/Images/<image-name>
+    Raises:
+        ResourcesUnavailableError
+    """
+    try:
+        image_versions = compute_client.community_gallery_image_versions.list(
+            location=region,
+            public_gallery_name=gallery_name,
+            gallery_image_name=image_name,
+        )
+        image_versions = list(image_versions)
+        if not image_versions:
+            raise exceptions.ResourcesUnavailableError(
+                f'No versions available for Azure community image {image_name}')
+        latest_version = image_versions[-1].name
+
+        image_details = compute_client.community_gallery_image_versions.get(
+            location=region,
+            public_gallery_name=gallery_name,
+            gallery_image_name=image_name,
+            gallery_image_version_name=latest_version)
+        return image_details.storage_profile.os_disk_image.disk_size_gb
+    except azure.exceptions().AzureError as e:
+        raise exceptions.ResourcesUnavailableError(
+            f'Failed to get community image size: {e}.') from e
diff --git a/sky/provision/azure/azure-config-template.json b/sky/provision/azure/azure-config-template.json
index 489783fa..c743dd40 100644
--- a/sky/provision/azure/azure-config-template.json
+++ b/sky/provision/azure/azure-config-template.json
@@ -13,6 +13,12 @@
             "metadata": {
                 "description": "Subnet parameters."
             }
+        },
+        "nsgName": {
+            "type": "string",
+            "metadata": {
+                "description": "Name of the Network Security Group associated with the SkyPilot cluster."
+            }
         }
     },
     "variables": {
@@ -20,7 +26,7 @@
         "location": "[resourceGroup().location]",
         "msiName": "[concat('sky-', parameters('clusterId'), '-msi')]",
         "roleAssignmentName": "[concat('sky-', parameters('clusterId'), '-ra')]",
-        "nsgName": "[concat('sky-', parameters('clusterId'), '-nsg')]",
+        "nsgName": "[parameters('nsgName')]",
         "nsg": "[resourceId('Microsoft.Network/networkSecurityGroups', variables('nsgName'))]",
         "vnetName": "[concat('sky-', parameters('clusterId'), '-vnet')]",
         "subnetName": "[concat('sky-', parameters('clusterId'), '-subnet')]"
diff --git a/sky/provision/azure/config.py b/sky/provision/azure/config.py
index 22982a99..afa94b4a 100644
--- a/sky/provision/azure/config.py
+++ b/sky/provision/azure/config.py
@@ -8,7 +8,7 @@ import json
 from pathlib import Path
 import random
 import time
-from typing import Any, Callable
+from typing import Any, Callable, Tuple
 
 from sky import exceptions
 from sky import sky_logging
@@ -22,6 +22,7 @@ UNIQUE_ID_LEN = 4
 _DEPLOYMENT_NAME = 'skypilot-config'
 _LEGACY_DEPLOYMENT_NAME = 'ray-config'
 _RESOURCE_GROUP_WAIT_FOR_DELETION_TIMEOUT = 480  # 8 minutes
+_CLUSTER_ID = '{cluster_name_on_cloud}-{unique_id}'
 
 
 def get_azure_sdk_function(client: Any, function_name: str) -> Callable:
@@ -41,6 +42,19 @@ def get_azure_sdk_function(client: Any, function_name: str) -> Callable:
     return func
 
 
+def get_cluster_id_and_nsg_name(resource_group: str,
+                                cluster_name_on_cloud: str) -> Tuple[str, str]:
+    hasher = hashlib.md5(resource_group.encode('utf-8'))
+    unique_id = hasher.hexdigest()[:UNIQUE_ID_LEN]
+    # We use the cluster name + resource group hash as the
+    # unique ID for the cluster, as we need to make sure that
+    # the deployments have unique names during failover.
+    cluster_id = _CLUSTER_ID.format(cluster_name_on_cloud=cluster_name_on_cloud,
+                                    unique_id=unique_id)
+    nsg_name = f'sky-{cluster_id}-nsg'
+    return cluster_id, nsg_name
+
+
 @common.log_function_start_end
 def bootstrap_instances(
         region: str, cluster_name_on_cloud: str,
@@ -117,12 +131,13 @@ def bootstrap_instances(
 
     logger.info(f'Using cluster name: {cluster_name_on_cloud}')
 
-    hasher = hashlib.md5(provider_config['resource_group'].encode('utf-8'))
-    unique_id = hasher.hexdigest()[:UNIQUE_ID_LEN]
+    cluster_id, nsg_name = get_cluster_id_and_nsg_name(
+        resource_group=provider_config['resource_group'],
+        cluster_name_on_cloud=cluster_name_on_cloud)
     subnet_mask = provider_config.get('subnet_mask')
     if subnet_mask is None:
         # choose a random subnet, skipping most common value of 0
-        random.seed(unique_id)
+        random.seed(cluster_id)
         subnet_mask = f'10.{random.randint(1, 254)}.0.0/16'
     logger.info(f'Using subnet mask: {subnet_mask}')
 
@@ -135,10 +150,10 @@ def bootstrap_instances(
                     'value': subnet_mask
                 },
                 'clusterId': {
-                    # We use the cluster name + resource group hash as the
-                    # unique ID for the cluster, as we need to make sure that
-                    # the deployments have unique names during failover.
-                    'value': f'{cluster_name_on_cloud}-{unique_id}'
+                    'value': cluster_id
+                },
+                'nsgName': {
+                    'value': nsg_name
                 },
             },
         }
diff --git a/sky/provision/azure/instance.py b/sky/provision/azure/instance.py
index f6c865e2..cc2dc692 100644
--- a/sky/provision/azure/instance.py
+++ b/sky/provision/azure/instance.py
@@ -15,6 +15,7 @@ from sky import status_lib
 from sky.adaptors import azure
 from sky.provision import common
 from sky.provision import constants
+from sky.provision.azure import config as config_lib
 from sky.utils import common_utils
 from sky.utils import subprocess_utils
 from sky.utils import ux_utils
@@ -31,6 +32,8 @@ logger = sky_logging.init_logger(__name__)
 # https://github.com/Azure/azure-sdk-for-python/issues/9422
 azure_logger = logging.getLogger('azure')
 azure_logger.setLevel(logging.WARNING)
+Client = Any
+NetworkSecurityGroup = Any
 
 _RESUME_INSTANCE_TIMEOUT = 480  # 8 minutes
 _RESUME_PER_INSTANCE_TIMEOUT = 120  # 2 minutes
@@ -40,6 +43,10 @@ _WAIT_CREATION_TIMEOUT_SECONDS = 600
 
 _RESOURCE_GROUP_NOT_FOUND_ERROR_MESSAGE = 'ResourceGroupNotFound'
 _POLL_INTERVAL = 1
+# TODO(Doyoung): _LEGACY_NSG_NAME can be remove this after 0.8.0 to ignore
+# legacy nsg names.
+_LEGACY_NSG_NAME = 'ray-{cluster_name_on_cloud}-nsg'
+_SECOND_LEGACY_NSG_NAME = 'sky-{cluster_name_on_cloud}-nsg'
 
 
 class AzureInstanceStatus(enum.Enum):
@@ -795,6 +802,32 @@ def query_instances(
     return statuses
 
 
+# TODO(Doyoung): _get_cluster_nsg can be remove this after 0.8.0 to ignore
+# legacy nsg names.
+def _get_cluster_nsg(network_client: Client, resource_group: str,
+                     cluster_name_on_cloud: str) -> NetworkSecurityGroup:
+    """Retrieve the NSG associated with the given name of the cluster."""
+    list_network_security_groups = _get_azure_sdk_function(
+        client=network_client.network_security_groups, function_name='list')
+    legacy_nsg_name = _LEGACY_NSG_NAME.format(
+        cluster_name_on_cloud=cluster_name_on_cloud)
+    second_legacy_nsg_name = _SECOND_LEGACY_NSG_NAME.format(
+        cluster_name_on_cloud=cluster_name_on_cloud)
+    _, nsg_name = config_lib.get_cluster_id_and_nsg_name(
+        resource_group=resource_group,
+        cluster_name_on_cloud=cluster_name_on_cloud)
+    possible_nsg_names = [nsg_name, legacy_nsg_name, second_legacy_nsg_name]
+    for nsg in list_network_security_groups(resource_group):
+        if nsg.name in possible_nsg_names:
+            return nsg
+
+    # Raise an error if no matching NSG is found
+    raise ValueError('Failed to find a matching NSG for cluster '
+                     f'{cluster_name_on_cloud!r} in resource group '
+                     f'{resource_group!r}. Expected NSG names were: '
+                     f'{possible_nsg_names}.')
+
+
 def open_ports(
     cluster_name_on_cloud: str,
     ports: List[str],
@@ -809,58 +842,66 @@ def open_ports(
     update_network_security_groups = _get_azure_sdk_function(
         client=network_client.network_security_groups,
         function_name='create_or_update')
-    list_network_security_groups = _get_azure_sdk_function(
-        client=network_client.network_security_groups, function_name='list')
-    for nsg in list_network_security_groups(resource_group):
-        try:
-            # Wait the NSG creation to be finished before opening a port. The
-            # cluster provisioning triggers the NSG creation, but it may not be
-            # finished yet.
-            backoff = common_utils.Backoff(max_backoff_factor=1)
-            start_time = time.time()
-            while True:
-                if nsg.provisioning_state not in ['Creating', 'Updating']:
-                    break
-                if time.time() - start_time > _WAIT_CREATION_TIMEOUT_SECONDS:
-                    logger.warning(
-                        f'Fails to wait for the creation of NSG {nsg.name} in '
-                        f'{resource_group} within '
-                        f'{_WAIT_CREATION_TIMEOUT_SECONDS} seconds. '
-                        'Skip this NSG.')
-                backoff_time = backoff.current_backoff()
-                logger.info(f'NSG {nsg.name} is not created yet. Waiting for '
-                            f'{backoff_time} seconds before checking again.')
-                time.sleep(backoff_time)
-
-            # Azure NSG rules have a priority field that determines the order
-            # in which they are applied. The priority must be unique across
-            # all inbound rules in one NSG.
-            priority = max(rule.priority
-                           for rule in nsg.security_rules
-                           if rule.direction == 'Inbound') + 1
-            nsg.security_rules.append(
-                azure.create_security_rule(
-                    name=f'sky-ports-{cluster_name_on_cloud}-{priority}',
-                    priority=priority,
-                    protocol='Tcp',
-                    access='Allow',
-                    direction='Inbound',
-                    source_address_prefix='*',
-                    source_port_range='*',
-                    destination_address_prefix='*',
-                    destination_port_ranges=ports,
-                ))
-            poller = update_network_security_groups(resource_group, nsg.name,
-                                                    nsg)
-            poller.wait()
-            if poller.status() != 'Succeeded':
+
+    try:
+        # Wait for the NSG creation to be finished before opening a port. The
+        # cluster provisioning triggers the NSG creation, but it may not be
+        # finished yet.
+        backoff = common_utils.Backoff(max_backoff_factor=1)
+        start_time = time.time()
+        while True:
+            nsg = _get_cluster_nsg(network_client, resource_group,
+                                   cluster_name_on_cloud)
+            if nsg.provisioning_state not in ['Creating', 'Updating']:
+                break
+            if time.time() - start_time > _WAIT_CREATION_TIMEOUT_SECONDS:
                 with ux_utils.print_exception_no_traceback():
-                    raise ValueError(f'Failed to open ports {ports} in NSG '
-                                     f'{nsg.name}: {poller.status()}')
-        except azure.exceptions().HttpResponseError as e:
+                    raise TimeoutError(
+                        f'Timed out while waiting for the Network '
+                        f'Security Group {nsg.name!r} to be ready for '
+                        f'cluster {cluster_name_on_cloud!r} in '
+                        f'resource group {resource_group!r}. The NSG '
+                        f'did not reach a stable state '
+                        '(Creating/Updating) within the allocated '
+                        f'{_WAIT_CREATION_TIMEOUT_SECONDS} seconds. '
+                        'Consequently, the operation to open ports '
+                        f'{ports} failed.')
+
+            backoff_time = backoff.current_backoff()
+            logger.info(f'NSG {nsg.name} is not created yet. Waiting for '
+                        f'{backoff_time} seconds before checking again.')
+            time.sleep(backoff_time)
+
+        # Azure NSG rules have a priority field that determines the order
+        # in which they are applied. The priority must be unique across
+        # all inbound rules in one NSG.
+        priority = max(rule.priority
+                       for rule in nsg.security_rules
+                       if rule.direction == 'Inbound') + 1
+        nsg.security_rules.append(
+            azure.create_security_rule(
+                name=f'sky-ports-{cluster_name_on_cloud}-{priority}',
+                priority=priority,
+                protocol='Tcp',
+                access='Allow',
+                direction='Inbound',
+                source_address_prefix='*',
+                source_port_range='*',
+                destination_address_prefix='*',
+                destination_port_ranges=ports,
+            ))
+        poller = update_network_security_groups(resource_group, nsg.name, nsg)
+        poller.wait()
+        if poller.status() != 'Succeeded':
             with ux_utils.print_exception_no_traceback():
-                raise ValueError(
-                    f'Failed to open ports {ports} in NSG {nsg.name}.') from e
+                raise ValueError(f'Failed to open ports {ports} in NSG '
+                                 f'{nsg.name}: {poller.status()}')
+
+    except azure.exceptions().HttpResponseError as e:
+        with ux_utils.print_exception_no_traceback():
+            raise ValueError(f'Failed to open ports {ports} in NSG for cluster '
+                             f'{cluster_name_on_cloud!r} within resource group '
+                             f'{resource_group!r}.') from e
 
 
 def cleanup_ports(
diff --git a/sky/provision/kubernetes/instance.py b/sky/provision/kubernetes/instance.py
index 6ce7b74d..26ed5f51 100644
--- a/sky/provision/kubernetes/instance.py
+++ b/sky/provision/kubernetes/instance.py
@@ -1,5 +1,6 @@
 """Kubernetes instance provisioning."""
 import copy
+import json
 import time
 from typing import Any, Dict, List, Optional
 import uuid
@@ -425,6 +426,70 @@ def _label_pod(namespace: str, context: Optional[str], pod_name: str,
         _request_timeout=kubernetes.API_TIMEOUT)
 
 
+def _create_namespaced_pod_with_retries(namespace: str, pod_spec: dict,
+                                        context: Optional[str]) -> Any:
+    """Attempts to create a Kubernetes Pod and handle any errors.
+
+    Currently, we handle errors due to the AppArmor annotation and retry if
+    it fails due to the `FieldValueForbidden` error.
+    See https://github.com/skypilot-org/skypilot/issues/4174 for details.
+
+    Returns: The created Pod object.
+    """
+    try:
+        # Attempt to create the Pod with the AppArmor annotation
+        pod = kubernetes.core_api(context).create_namespaced_pod(
+            namespace, pod_spec)
+        return pod
+    except kubernetes.api_exception() as e:
+        try:
+            error_body = json.loads(e.body)
+            error_message = error_body.get('message', '')
+        except json.JSONDecodeError:
+            error_message = str(e.body)
+        # Check if the error is due to the AppArmor annotation and retry.
+        # We add an AppArmor annotation to set it as unconfined in our
+        # base template in kubernetes-ray.yml.j2. This is required for
+        # FUSE to work in the pod on most Kubernetes distributions.
+        # However, some distributions do not support the AppArmor annotation
+        # and will fail to create the pod. In this case, we retry without
+        # the annotation.
+        if (e.status == 422 and 'FieldValueForbidden' in error_message and
+                'AppArmorProfile: nil' in error_message):
+            logger.warning('AppArmor annotation caused pod creation to fail. '
+                           'Retrying without the annotation. '
+                           'Note: this may cause bucket mounting to fail.')
+
+            # Remove the AppArmor annotation
+            annotations = pod_spec.get('metadata', {}).get('annotations', {})
+            if ('container.apparmor.security.beta.kubernetes.io/ray-node'
+                    in annotations):
+                del annotations[
+                    'container.apparmor.security.beta.kubernetes.io/ray-node']
+                pod_spec['metadata']['annotations'] = annotations
+                logger.info('AppArmor annotation removed from Pod spec.')
+            else:
+                logger.warning('AppArmor annotation not found in pod spec, '
+                               'retrying will not help. '
+                               f'Current annotations: {annotations}')
+                raise e
+
+            # Retry Pod creation without the AppArmor annotation
+            try:
+                pod = kubernetes.core_api(context).create_namespaced_pod(
+                    namespace, pod_spec)
+                logger.info(f'Pod {pod.metadata.name} created successfully '
+                            'without AppArmor annotation.')
+                return pod
+            except kubernetes.api_exception() as retry_exception:
+                logger.info('Failed to create Pod without AppArmor annotation: '
+                            f'{retry_exception}')
+                raise retry_exception
+        else:
+            # Re-raise the exception if it's a different error
+            raise e
+
+
 def _create_pods(region: str, cluster_name_on_cloud: str,
                  config: common.ProvisionConfig) -> common.ProvisionRecord:
     """Create pods based on the config."""
@@ -546,8 +611,7 @@ def _create_pods(region: str, cluster_name_on_cloud: str,
                 }
             }
 
-        pod = kubernetes.core_api(context).create_namespaced_pod(
-            namespace, pod_spec)
+        pod = _create_namespaced_pod_with_retries(namespace, pod_spec, context)
         created_pods[pod.metadata.name] = pod
         if head_pod_name is None:
             head_pod_name = pod.metadata.name
diff --git a/sky/resources.py b/sky/resources.py
index 384f2b6a..540cbfb7 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -225,6 +225,7 @@ class Resources:
         self._set_memory(memory)
         self._set_accelerators(accelerators, accelerator_args)
 
+        # TODO: move these out of init to prevent repeated calls.
         self._try_validate_instance_type()
         self._try_validate_cpus_mem()
         self._try_validate_managed_job_attributes()


Test patch:
diff --git a/tests/unit_tests/test_azure_utils.py b/tests/unit_tests/test_azure_utils.py
new file mode 100644
index 00000000..93ef5caa
--- /dev/null
+++ b/tests/unit_tests/test_azure_utils.py
@@ -0,0 +1,21 @@
+import pytest
+
+from sky.clouds.utils import azure_utils
+
+
+def test_validate_image_id():
+    # Valid marketplace image ID
+    azure_utils.validate_image_id("publisher:offer:sku:version")
+
+    # Valid community image ID
+    azure_utils.validate_image_id(
+        "/CommunityGalleries/gallery-name/Images/image-name")
+
+    # Invalid format (neither marketplace nor community)
+    with pytest.raises(ValueError):
+        azure_utils.validate_image_id(
+            "CommunityGalleries/gallery-name/Images/image-name")
+
+    # Invalid marketplace image ID (too few parts)
+    with pytest.raises(ValueError):
+        azure_utils.validate_image_id("publisher:offer:sku")