Repository: skypilot-org/skypilot
Pull request #6006 (2025-06-15 21:40:33)

add priority to resources
<!-- Describe the changes in this PR -->



<!-- 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):

- [ ] Code formatting: install pre-commit (auto-check on commit) or `bash format.sh`
- [ ] Any manual or new tests for this PR (please specify below)
- [ ] All smoke tests: `/smoke-test` (CI) or `pytest tests/test_smoke.py` (local)
- [ ] Relevant individual tests: `/smoke-test -k test_name` (CI) or `pytest tests/test_smoke.py::test_name` (local)
- [ ] Backward compatibility: `/quicktest-core` (CI) or `pytest tests/smoke_tests/test_backward_compat.py` (local)

<!-- CI commands (/-prefixed) can only be triggered by repo members -->


cg505: /quicktest-core
cg505: /smoke-test

Patch:
diff --git a/sky/jobs/server/core.py b/sky/jobs/server/core.py
index b852ead72..c4c9e00d4 100644
--- a/sky/jobs/server/core.py
+++ b/sky/jobs/server/core.py
@@ -102,14 +102,47 @@ def launch(
                     'name only and comment out the task names (so that they '
                     'will be auto-generated) .')
         task_names.add(task_.name)
-        if task_.job_priority is not None:
-            if (priority is not None and priority != task_.job_priority):
+
+        # Check for priority in resources first, then fall back to job priority
+        task_priority = None
+        if task_.resources:
+            # Convert set to list to access elements by index
+            resources_list = list(task_.resources)
+            # Take first resource's priority as reference
+            task_priority = resources_list[0].priority
+
+            # Check all other resources have same priority
+            for resource in resources_list[1:]:
+                if resource.priority != task_priority:
+                    with ux_utils.print_exception_no_traceback():
+                        raise ValueError(
+                            f'Task {task_.name!r}: All resources must have the '
+                            'same priority. Found priority '
+                            f'{resource.priority} but expected {task_priority}.'
+                        )
+
+            # Check for conflict between resources priority and job
+            # priority
+            if task_.job_priority is not None:
+                with ux_utils.print_exception_no_traceback():
+                    raise ValueError(
+                        f'Task {task_.name!r}: Cannot specify both '
+                        f'resources.priority ({task_priority}) and '
+                        f'job.priority ({task_.job_priority}). Please use only '
+                        'one priority specification method.')
+
+        # Fall back to job priority if no resources priority found
+        if task_priority is None:
+            task_priority = task_.job_priority
+
+        if task_priority is not None:
+            if (priority is not None and priority != task_priority):
                 with ux_utils.print_exception_no_traceback():
                     raise ValueError(
                         'Multiple tasks in the DAG have different priorities. '
                         'Either specify a priority in only one task, or set '
                         'the same priority for each task.')
-            priority = task_.job_priority
+            priority = task_priority
 
     if priority is None:
         priority = managed_job_constants.DEFAULT_PRIORITY
diff --git a/sky/resources.py b/sky/resources.py
index c72aec488..914383017 100644
--- a/sky/resources.py
+++ b/sky/resources.py
@@ -101,7 +101,7 @@ class Resources:
     """
     # If any fields changed, increment the version. For backward compatibility,
     # modify the __setstate__ method to handle the old version.
-    _VERSION = 26
+    _VERSION = 27
 
     def __init__(
         self,
@@ -124,6 +124,7 @@ class Resources:
         ports: Optional[Union[int, str, List[str], Tuple[str]]] = None,
         labels: Optional[Dict[str, str]] = None,
         autostop: Union[bool, int, Dict[str, Any], None] = None,
+        priority: Optional[int] = None,
         volumes: Optional[List[Dict[str, Any]]] = None,
         # Internal use only.
         # pylint: disable=invalid-name
@@ -217,6 +218,9 @@ class Resources:
             not supported and will be ignored.
           autostop: the autostop configuration to use. For launched resources,
             may or may not correspond to the actual current autostop config.
+          priority: the priority for this resource configuration. Must be an
+            integer from 0 to 1000, where higher values indicate higher priority.
+            If None, no priority is set.
           volumes: the volumes to mount on the instance.
           _docker_login_config: the docker configuration to use. This includes
             the docker username, password, and registry server. If None, skip
@@ -357,10 +361,14 @@ class Resources:
         self._cluster_config_overrides = _cluster_config_overrides
         self._cached_repr: Optional[str] = None
 
+        # Initialize _priority before calling the setter
+        self._priority: Optional[int] = None
+
         self._set_cpus(cpus)
         self._set_memory(memory)
         self._set_accelerators(accelerators, accelerator_args)
         self._set_autostop_config(autostop)
+        self._set_priority(priority)
         self._set_volumes(volumes)
 
     def validate(self):
@@ -617,6 +625,14 @@ class Resources:
         """
         return self._autostop_config
 
+    @property
+    def priority(self) -> Optional[int]:
+        """The priority for this resource configuration.
+
+        Higher values indicate higher priority. Valid range is 0-1000.
+        """
+        return self._priority
+
     @property
     def is_image_managed(self) -> Optional[bool]:
         return self._is_image_managed
@@ -800,6 +816,20 @@ class Resources:
     ) -> None:
         self._autostop_config = AutostopConfig.from_yaml_config(autostop)
 
+    def _set_priority(self, priority: Optional[int]) -> None:
+        """Sets the priority for this resource configuration.
+
+        Args:
+            priority: Priority value from 0 to 1000, where higher values
+                indicate higher priority. If None, no priority is set.
+        """
+        if priority is not None:
+            if not 0 <= priority <= 1000:
+                with ux_utils.print_exception_no_traceback():
+                    raise ValueError(f'Priority must be between 0 and 1000. '
+                                     f'Found: {priority}')
+        self._priority = priority
+
     def _set_volumes(
         self,
         volumes: Optional[List[Dict[str, Any]]],
@@ -1716,6 +1746,7 @@ class Resources:
             ports=override.pop('ports', self.ports),
             labels=override.pop('labels', self.labels),
             autostop=override.pop('autostop', current_autostop_config),
+            priority=override.pop('priority', self.priority),
             volumes=override.pop('volumes', self.volumes),
             infra=override.pop('infra', None),
             _docker_login_config=override.pop('_docker_login_config',
@@ -1936,6 +1967,7 @@ class Resources:
         resources_fields['ports'] = config.pop('ports', None)
         resources_fields['labels'] = config.pop('labels', None)
         resources_fields['autostop'] = config.pop('autostop', None)
+        resources_fields['priority'] = config.pop('priority', None)
         resources_fields['volumes'] = config.pop('volumes', None)
         resources_fields['_docker_login_config'] = config.pop(
             '_docker_login_config', None)
@@ -2006,6 +2038,7 @@ class Resources:
             config['volumes'] = volumes
         if self._autostop_config is not None:
             config['autostop'] = self._autostop_config.to_yaml_config()
+        add_if_not_none('priority', self.priority)
         if self._docker_login_config is not None:
             config['_docker_login_config'] = dataclasses.asdict(
                 self._docker_login_config)
@@ -2174,6 +2207,9 @@ class Resources:
         if version < 26:
             self._network_tier = state.get('_network_tier', None)
 
+        if version < 27:
+            self._priority = None
+
         self.__dict__.update(state)
 
 
diff --git a/sky/utils/command_runner.py b/sky/utils/command_runner.py
index b6e2bbb9e..f74f80666 100644
--- a/sky/utils/command_runner.py
+++ b/sky/utils/command_runner.py
@@ -561,7 +561,7 @@ class SSHCommandRunner(CommandRunner):
         if self.ssh_control_name is not None:
             control_path = _ssh_control_path(self.ssh_control_name)
             if control_path is not None:
-                # Suppress the `Exit request sent.` output for this comamnd
+                # Suppress the `Exit request sent.` output for this command
                 # which would interrupt the CLI spinner.
                 cmd = (f'ssh -O exit -S {control_path}/%C '
                        f'{self.ssh_user}@{self.ip} > /dev/null 2>&1')
diff --git a/sky/utils/context.py b/sky/utils/context.py
index 9cf132545..d74ceb074 100644
--- a/sky/utils/context.py
+++ b/sky/utils/context.py
@@ -262,7 +262,7 @@ F = TypeVar('F', bound=Callable[..., Any])
 
 
 def contextual(func: F) -> F:
-    """Decorator to intiailize a context before executing the function.
+    """Decorator to initialize a context before executing the function.
 
     If a context is already initialized, this decorator will reset the context,
     i.e. all contextual variables set previously will be cleared.
diff --git a/sky/utils/schemas.py b/sky/utils/schemas.py
index 3aa197121..b4e4cd70f 100644
--- a/sky/utils/schemas.py
+++ b/sky/utils/schemas.py
@@ -273,6 +273,11 @@ def _get_single_resources_schema():
                 }]
             },
             'autostop': _AUTOSTOP_SCHEMA,
+            'priority': {
+                'type': 'integer',
+                'minimum': 0,
+                'maximum': 1000,
+            },
             # The following fields are for internal use only. Should not be
             # specified in the task config.
             '_docker_login_config': {


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 394486f7d..71a4870d0 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -169,7 +169,7 @@ def pytest_addoption(parser):
     parser.addoption(
         '--helm-package',
         type=str,
-        default='skypilot-nightly',
+        default='',
         help='Package name to use for Helm tests',
     )
 
diff --git a/tests/kubernetes/scripts/helm_gcp.sh b/tests/kubernetes/scripts/helm_gcp.sh
index ea387ebbc..4a5f401f2 100644
--- a/tests/kubernetes/scripts/helm_gcp.sh
+++ b/tests/kubernetes/scripts/helm_gcp.sh
@@ -128,10 +128,21 @@ if [ $RETRY_COUNT -eq $MAX_RETRIES ]; then
     exit 1
 fi
 
-# Extract version from response and verify it matches
+# Extract version from response and verify it matches (only if not using 'latest')
 RETURNED_VERSION=$(echo $HEALTH_RESPONSE | jq -r '.version')
-if [ "$RETURNED_VERSION" != "$HELM_VERSION" ]; then
-    echo "Error: Version mismatch! Expected: $HELM_VERSION, Got: $RETURNED_VERSION"
+
+# Verify that we got a valid version string
+if [ -z "$RETURNED_VERSION" ] || [ ${#RETURNED_VERSION} -le 1 ]; then
+    echo "Error: Invalid version returned from API. Got: '$RETURNED_VERSION'"
     exit 1
 fi
-echo "Version verification successful! Expected version $HELM_VERSION matches returned version $RETURNED_VERSION"
+
+if [ "$HELM_VERSION" != "latest" ]; then
+    if [ "$RETURNED_VERSION" != "$HELM_VERSION" ]; then
+        echo "Error: Version mismatch! Expected: $HELM_VERSION, Got: $RETURNED_VERSION"
+        exit 1
+    fi
+    echo "Version verification successful! Expected version $HELM_VERSION matches returned version $RETURNED_VERSION"
+else
+    echo "Using latest version. Skipping version verification. Deployed version: $RETURNED_VERSION"
+fi
diff --git a/tests/unit_tests/test_resources.py b/tests/unit_tests/test_resources.py
index d637d30c9..ef1fdfb10 100644
--- a/tests/unit_tests/test_resources.py
+++ b/tests/unit_tests/test_resources.py
@@ -772,3 +772,193 @@ def test_autostop_config():
     assert r.autostop_config.enabled is False  # should remain disabled
     assert r.autostop_config.down is True
     assert r.autostop_config.idle_minutes == 55
+
+
+def test_priority_basic():
+    """Test basic priority field functionality."""
+    # Test with no priority specified (defaults to None)
+    r = Resources()
+    assert r.priority is None
+
+    # Test with valid priority values
+    r = Resources(priority=0)
+    assert r.priority == 0
+
+    r = Resources(priority=500)
+    assert r.priority == 500
+
+    r = Resources(priority=1000)
+    assert r.priority == 1000
+
+    # Test with None explicitly
+    r = Resources(priority=None)
+    assert r.priority is None
+
+
+def test_priority_validation():
+    """Test priority field validation with invalid values."""
+    # Test invalid priority - below range
+    with pytest.raises(ValueError, match='Priority must be between 0 and 1000'):
+        Resources(priority=-1)
+
+    # Test invalid priority - above range
+    with pytest.raises(ValueError, match='Priority must be between 0 and 1000'):
+        Resources(priority=1001)
+
+
+def test_priority_yaml_serialization():
+    """Test priority YAML serialization and deserialization."""
+    # Test priority serialization
+    r = Resources(priority=750)
+    yaml_config = r.to_yaml_config()
+    assert yaml_config['priority'] == 750
+
+    loaded_resources = list(Resources.from_yaml_config(yaml_config))[0]
+    assert loaded_resources.priority == 750
+
+    # Test None priority (should not appear in yaml)
+    r = Resources(priority=None)
+    yaml_config = r.to_yaml_config()
+    assert 'priority' not in yaml_config
+
+    # Test priority with other fields
+    r = Resources(cpus=4, memory='8', priority=200)
+    yaml_config = r.to_yaml_config()
+    assert yaml_config['priority'] == 200
+    assert yaml_config['cpus'] == '4'
+    assert yaml_config['memory'] == '8'
+
+    loaded_resources = list(Resources.from_yaml_config(yaml_config))[0]
+    assert loaded_resources.priority == 200
+    assert loaded_resources.cpus == '4'
+    assert loaded_resources.memory == '8'
+
+
+def test_priority_copy():
+    """Test priority preservation in copy operations."""
+    r = Resources(priority=300, cpus=4)
+    r_copy = r.copy()
+    assert r_copy.priority == 300
+
+    # Test overriding priority in copy
+    r_override = r.copy(priority=600)
+    assert r_override.priority == 600
+    assert r_override.cpus == '4'  # Other properties preserved
+
+    # Test copying with None priority
+    r_none = Resources(priority=None, cpus=2)
+    r_copy = r_none.copy()
+    assert r_copy.priority is None
+
+    # Test overriding None priority in copy
+    r_override = r_none.copy(priority=100)
+    assert r_override.priority == 100
+    assert r_override.cpus == '2'
+
+
+def test_priority_with_any_of():
+    """Test priority field works with any_of configuration."""
+    config = {
+        'priority': 500,  # Base priority
+        'any_of': [
+            {
+                'cpus': 8,
+                'memory': 16
+            },
+            {
+                'cpus': 4,
+                'memory': 32,
+                'priority': 800  # Override priority
+            },
+        ]
+    }
+    resources_set = Resources.from_yaml_config(config)
+
+    assert isinstance(resources_set, set)
+    assert len(resources_set) == 2
+
+    resources_list = list(resources_set)
+
+    # Find resources by properties
+    r_cpus8 = next((r for r in resources_list if r.cpus == '8'), None)
+    r_cpus4 = next((r for r in resources_list if r.cpus == '4'), None)
+
+    assert r_cpus8 is not None
+    assert r_cpus8.priority == 500  # Base priority
+
+    assert r_cpus4 is not None
+    assert r_cpus4.priority == 800  # Override priority
+
+
+def test_priority_with_ordered():
+    """Test priority field works with ordered configuration."""
+    config = {
+        'priority': 300,  # Base priority
+        'ordered': [
+            {
+                'infra': 'gcp',
+                'accelerators': 'A100:8'
+            },
+            {
+                'infra': 'aws',
+                'accelerators': 'V100:8',
+                'priority': 700  # Override priority
+            },
+        ]
+    }
+    resources_list = Resources.from_yaml_config(config)
+
+    assert isinstance(resources_list, list)
+    assert len(resources_list) == 2
+
+    # Ordered resources should preserve order
+    assert resources_list[0].infra.cloud.lower() == 'gcp'
+    assert resources_list[0].priority == 300  # Base priority
+
+    assert resources_list[1].infra.cloud.lower() == 'aws'
+    assert resources_list[1].priority == 700  # Override priority
+
+
+@pytest.mark.parametrize(['resources_kwargs', 'expected_yaml_config'], [
+    ({
+        'infra': 'aws/us-east-1',
+        'accelerators': 'A10',
+        'priority': 400
+    }, {
+        'infra': 'aws/us-east-1',
+        'accelerators': {
+            'A10': 1
+        },
+        'priority': 400,
+        'disk_size': 256,
+    }),
+    ({
+        'cpus': 4,
+        'memory': '8+',
+        'priority': 0
+    }, {
+        'cpus': '4',
+        'memory': '8+',
+        'priority': 0,
+        'disk_size': 256,
+    }),
+    ({
+        'cpus': 2,
+        'priority': 1000
+    }, {
+        'cpus': '2',
+        'priority': 1000,
+        'disk_size': 256,
+    }),
+])
+def test_priority_to_yaml_and_load(resources_kwargs, expected_yaml_config):
+    """Test priority field in to_yaml_config and from_yaml_config."""
+    r = Resources(**resources_kwargs)
+    yaml_config = r.to_yaml_config()
+    assert yaml_config == expected_yaml_config
+
+    loaded_r = list(Resources.from_yaml_config(yaml_config))[0]
+    assert loaded_r.priority == r.priority
+    assert loaded_r.cpus == r.cpus
+    if 'accelerators' in expected_yaml_config:
+        assert loaded_r.accelerators == r.accelerators
diff --git a/tests/unit_tests/test_sky/utils/test_schemas.py b/tests/unit_tests/test_sky/utils/test_schemas.py
index 3fdfa089c..4c1a2e77d 100644
--- a/tests/unit_tests/test_sky/utils/test_schemas.py
+++ b/tests/unit_tests/test_sky/utils/test_schemas.py
@@ -108,6 +108,63 @@ class TestResourcesSchema(unittest.TestCase):
                     msg=f"Expected '{config['infra']}' to be rejected"):
                 jsonschema.validate(instance=config, schema=resources_schema)
 
+    def test_valid_priority_configs(self):
+        """Test validation of valid priority field configs."""
+        resources_schema = schemas.get_resources_schema()
+
+        # Valid priority configurations
+        valid_priority_configs = [
+            {
+                'priority': 0
+            },  # Minimum value
+            {
+                'priority': 500
+            },  # Middle value
+            {
+                'priority': 1000
+            },  # Maximum value
+            {
+                'cpus': 4,
+                'priority': 750
+            },  # With other fields
+        ]
+
+        for config in valid_priority_configs:
+            # Should not raise an exception
+            jsonschema.validate(instance=config, schema=resources_schema)
+
+    def test_invalid_priority_configs(self):
+        """Test validation rejects invalid priority field configs."""
+        resources_schema = schemas.get_resources_schema()
+
+        # Invalid priority configurations
+        invalid_priority_configs = [
+            {
+                'priority': -1
+            },  # Below minimum
+            {
+                'priority': 1001
+            },  # Above maximum
+            {
+                'priority': 'high'
+            },  # Not an integer
+            {
+                'priority': 500.5
+            },  # Not an integer
+            {
+                'priority': None
+            },  # None (should be omitted instead)
+            {
+                'priority': True
+            },  # Boolean
+        ]
+
+        for config in invalid_priority_configs:
+            with self.assertRaises(
+                    jsonschema.exceptions.ValidationError,
+                    msg=f"Expected priority config {config} to be rejected"):
+                jsonschema.validate(instance=config, schema=resources_schema)
+
 
 class TestWorkspaceSchema(unittest.TestCase):
     """Tests for the workspace schema in schemas.py"""