[Serve] Add label_selector and bundle_label_selector to Serve API

python/ray/_private/label_utils.py

@@ -228,3 +228,45 @@ def validate_fallback_strategy(
                 return error_message
 
     return None
+
+
+def match_label_selector_value(node_value: Optional[str], selector_value: str) -> bool:
+    """Evaluates if a node's label value matches a selector expression.
+
+    Supports:
+    - Equality: "value" matches if node_value == "value"
+    - Not Equal: "!value" matches if node_value != "value"
+    - In: "in(v1, v2)" matches if node_value in ["v1", "v2"]
+    - Not In: "!in(v1, v2)" matches if node_value not in ["v1", "v2"]
+    """
+    if not isinstance(selector_value, str):
+        return False
+
+    # !in operator
+    if selector_value.startswith("!in(") and selector_value.endswith(")"):
+        content = selector_value[4:-1]
+        values = [v.strip() for v in content.split(",")]
+        return node_value is None or node_value not in values
+
+    # in operator
+    if selector_value.startswith("in(") and selector_value.endswith(")"):
+        content = selector_value[3:-1]
+        values = [v.strip() for v in content.split(",")]
+        return node_value in values
+
+    # not equal operator
+    if selector_value.startswith("!"):
+        target_val = selector_value[1:]
+        return node_value != target_val
+
+    # equals operator
+    return node_value == selector_value
+
+
+def match_label_selector(node_labels: Dict[str, str], selector: Dict[str, str]) -> bool:
+    """Returns True if node_labels satisfy all selector constraints."""
+    for key, selector_value in selector.items():
+        node_value = node_labels.get(key)
+        if not match_label_selector_value(node_value, selector_value):
+            return False
+    return True

python/ray/serve/_private/application_state.py

@@ -1707,6 +1707,14 @@ def override_deployment_info(
         override_max_replicas_per_node = options.pop(
             "max_replicas_per_node", replica_config.max_replicas_per_node
         )
+        override_bundle_label_selector = options.pop(
+            "placement_group_bundle_label_selector",
+            replica_config.placement_group_bundle_label_selector,
+        )
+        override_fallback_strategy = options.pop(
+            "placement_group_fallback_strategy",
+            replica_config.placement_group_fallback_strategy,
+        )
 
         # Record telemetry for container runtime env feature at deployment level
         if override_actor_options.get("runtime_env") and (
@@ -1725,6 +1733,8 @@ def override_deployment_info(
             placement_group_bundles=override_placement_group_bundles,
             placement_group_strategy=override_placement_group_strategy,
             max_replicas_per_node=override_max_replicas_per_node,
+            placement_group_bundle_label_selector=override_bundle_label_selector,
+            placement_group_fallback_strategy=override_fallback_strategy,
         )
         override_options["replica_config"] = replica_config
 

python/ray/serve/_private/cluster_node_info_cache.py

@@ -89,6 +89,10 @@ def get_available_resources_per_node(self) -> Dict[str, Union[float, Dict]]:
 
         return self._cached_available_resources_per_node
 
+    def get_node_labels(self, node_id: str) -> Dict[str, str]:
+        """Get the labels for a specific node from the cache."""
+        return self._cached_node_labels.get(node_id, {})
+
 
 class DefaultClusterNodeInfoCache(ClusterNodeInfoCache):
     def __init__(self, gcs_client: GcsClient):

python/ray/serve/_private/common.py

@@ -805,6 +805,8 @@ class CreatePlacementGroupRequest:
     target_node_id: str
     name: str
     runtime_env: Optional[str] = None
+    bundle_label_selector: Optional[List[Dict[str, str]]] = None
+    fallback_strategy: Optional[List[Dict[str, Any]]] = None
 
 
 # This error is used to raise when a by-value DeploymentResponse is converted to an

python/ray/serve/_private/config.py

@@ -480,6 +480,8 @@ def __init__(
         ray_actor_options: Dict,
         placement_group_bundles: Optional[List[Dict[str, float]]] = None,
         placement_group_strategy: Optional[str] = None,
+        placement_group_bundle_label_selector: Optional[List[Dict[str, str]]] = None,
+        placement_group_fallback_strategy: Optional[List[Dict[str, Any]]] = None,
         max_replicas_per_node: Optional[int] = None,
         needs_pickle: bool = True,
     ):
@@ -505,6 +507,10 @@ def __init__(
 
         self.placement_group_bundles = placement_group_bundles
         self.placement_group_strategy = placement_group_strategy
+        self.placement_group_bundle_label_selector = (
+            placement_group_bundle_label_selector
+        )
+        self.placement_group_fallback_strategy = placement_group_fallback_strategy
 
         self.max_replicas_per_node = max_replicas_per_node
 
@@ -535,12 +541,18 @@ def update(
         ray_actor_options: dict,
         placement_group_bundles: Optional[List[Dict[str, float]]] = None,
         placement_group_strategy: Optional[str] = None,
+        placement_group_bundle_label_selector: Optional[List[Dict[str, str]]] = None,
+        placement_group_fallback_strategy: Optional[List[Dict[str, Any]]] = None,
         max_replicas_per_node: Optional[int] = None,
     ):
         self.ray_actor_options = ray_actor_options
 
         self.placement_group_bundles = placement_group_bundles
         self.placement_group_strategy = placement_group_strategy
+        self.placement_group_bundle_label_selector = (
+            placement_group_bundle_label_selector
+        )
+        self.placement_group_fallback_strategy = placement_group_fallback_strategy
 
         self.max_replicas_per_node = max_replicas_per_node
 
@@ -557,6 +569,8 @@ def create(
         ray_actor_options: Optional[Dict] = None,
         placement_group_bundles: Optional[List[Dict[str, float]]] = None,
         placement_group_strategy: Optional[str] = None,
+        placement_group_bundle_label_selector: Optional[List[Dict[str, str]]] = None,
+        placement_group_fallback_strategy: Optional[List[Dict[str, Any]]] = None,
         max_replicas_per_node: Optional[int] = None,
         deployment_def_name: Optional[str] = None,
     ):
@@ -597,17 +611,23 @@ def create(
                 deployment_def_name = deployment_def.__name__
 
         config = cls(
-            deployment_def_name,
-            pickle_dumps(
+            deployment_def_name=deployment_def_name,
+            serialized_deployment_def=pickle_dumps(
                 deployment_def,
                 f"Could not serialize the deployment {repr(deployment_def)}",
             ),
-            pickle_dumps(init_args, "Could not serialize the deployment init args"),
-            pickle_dumps(init_kwargs, "Could not serialize the deployment init kwargs"),
-            ray_actor_options,
-            placement_group_bundles,
-            placement_group_strategy,
-            max_replicas_per_node,
+            serialized_init_args=pickle_dumps(
+                init_args, "Could not serialize the deployment init args"
+            ),
+            serialized_init_kwargs=pickle_dumps(
+                init_kwargs, "Could not serialize the deployment init kwargs"
+            ),
+            ray_actor_options=ray_actor_options,
+            placement_group_bundles=placement_group_bundles,
+            placement_group_strategy=placement_group_strategy,
+            placement_group_bundle_label_selector=placement_group_bundle_label_selector,
+            placement_group_fallback_strategy=placement_group_fallback_strategy,
+            max_replicas_per_node=max_replicas_per_node,
         )
 
         config._deployment_def = deployment_def
@@ -633,6 +653,8 @@ def _validate_ray_actor_options(self):
             "resources",
             # Other options
             "runtime_env",
+            "label_selector",
+            "fallback_strategy",
         }
 
         for option in self.ray_actor_options:
@@ -674,11 +696,37 @@ def _validate_placement_group_options(self) -> None:
                     "`placement_group_bundles` must also be provided."
                 )
 
+        if self.placement_group_fallback_strategy is not None:
+            if self.placement_group_bundles is None:
+                raise ValueError(
+                    "If `placement_group_fallback_strategy` is provided, "
+                    "`placement_group_bundles` must also be provided."
+                )
+            if not isinstance(self.placement_group_fallback_strategy, list):
+                raise TypeError(
+                    "placement_group_fallback_strategy must be a list of dictionaries. "
+                    f"Got: {type(self.placement_group_fallback_strategy)}."
+                )
+            for i, strategy in enumerate(self.placement_group_fallback_strategy):
+                if not isinstance(strategy, dict):
+                    raise TypeError(
+                        f"placement_group_fallback_strategy entry at index {i} must be a dictionary. "
+                        f"Got: {type(strategy)}."
+                    )
+
+        if self.placement_group_bundle_label_selector is not None:
+            if self.placement_group_bundles is None:
+                raise ValueError(
+                    "If `placement_group_bundle_label_selector` is provided, "
+                    "`placement_group_bundles` must also be provided."
+                )
+
         if self.placement_group_bundles is not None:
             validate_placement_group(
                 bundles=self.placement_group_bundles,
                 strategy=self.placement_group_strategy or "PACK",
                 lifetime="detached",
+                bundle_label_selector=self.placement_group_bundle_label_selector,
             )
 
             resource_error_prefix = (
@@ -772,19 +820,37 @@ def init_kwargs(self) -> Optional[Tuple[Any]]:
     @classmethod
     def from_proto(cls, proto: ReplicaConfigProto, needs_pickle: bool = True):
         return ReplicaConfig(
-            proto.deployment_def_name,
-            proto.deployment_def,
-            proto.init_args if proto.init_args != b"" else None,
-            proto.init_kwargs if proto.init_kwargs != b"" else None,
-            json.loads(proto.ray_actor_options),
-            json.loads(proto.placement_group_bundles)
-            if proto.placement_group_bundles
-            else None,
-            proto.placement_group_strategy
-            if proto.placement_group_strategy != ""
-            else None,
-            proto.max_replicas_per_node if proto.max_replicas_per_node else None,
-            needs_pickle,
+            deployment_def_name=proto.deployment_def_name,
+            serialized_deployment_def=proto.deployment_def,
+            serialized_init_args=(proto.init_args if proto.init_args != b"" else None),
+            serialized_init_kwargs=(
+                proto.init_kwargs if proto.init_kwargs != b"" else None
+            ),
+            ray_actor_options=json.loads(proto.ray_actor_options),
+            placement_group_bundles=(
+                json.loads(proto.placement_group_bundles)
+                if proto.placement_group_bundles
+                else None
+            ),
+            placement_group_strategy=(
+                proto.placement_group_strategy
+                if proto.placement_group_strategy != ""
+                else None
+            ),
+            placement_group_bundle_label_selector=(
+                json.loads(proto.placement_group_bundle_label_selector)
+                if proto.placement_group_bundle_label_selector
+                else None
+            ),
+            placement_group_fallback_strategy=(
+                json.loads(proto.placement_group_fallback_strategy)
+                if proto.placement_group_fallback_strategy
+                else None
+            ),
+            max_replicas_per_node=(
+                proto.max_replicas_per_node if proto.max_replicas_per_node else None
+            ),
+            needs_pickle=needs_pickle,
         )
 
     @classmethod
@@ -793,19 +859,39 @@ def from_proto_bytes(cls, proto_bytes: bytes, needs_pickle: bool = True):
         return cls.from_proto(proto, needs_pickle)
 
     def to_proto(self):
+        placement_group_bundles = (
+            json.dumps(self.placement_group_bundles)
+            if self.placement_group_bundles is not None
+            else ""
+        )
+
+        bundle_label_selector = (
+            json.dumps(self.placement_group_bundle_label_selector)
+            if self.placement_group_bundle_label_selector is not None
+            else ""
+        )
+
+        fallback_strategy = (
+            json.dumps(self.placement_group_fallback_strategy)
+            if self.placement_group_fallback_strategy is not None
+            else ""
+        )
+
+        max_replicas_per_node = (
+            self.max_replicas_per_node if self.max_replicas_per_node is not None else 0
+        )
+
         return ReplicaConfigProto(
             deployment_def_name=self.deployment_def_name,
             deployment_def=self.serialized_deployment_def,
             init_args=self.serialized_init_args,
             init_kwargs=self.serialized_init_kwargs,
             ray_actor_options=json.dumps(self.ray_actor_options),
-            placement_group_bundles=json.dumps(self.placement_group_bundles)
-            if self.placement_group_bundles is not None
-            else "",
+            placement_group_bundles=placement_group_bundles,
             placement_group_strategy=self.placement_group_strategy,
-            max_replicas_per_node=self.max_replicas_per_node
-            if self.max_replicas_per_node is not None
-            else 0,
+            placement_group_bundle_label_selector=bundle_label_selector,
+            placement_group_fallback_strategy=fallback_strategy,
+            max_replicas_per_node=max_replicas_per_node,
         )
 
     def to_proto_bytes(self):
@@ -818,6 +904,8 @@ def to_dict(self):
             "ray_actor_options": self.ray_actor_options,
             "placement_group_bundles": self.placement_group_bundles,
             "placement_group_strategy": self.placement_group_strategy,
+            "placement_group_bundle_label_selector": self.placement_group_bundle_label_selector,
+            "placement_group_fallback_strategy": self.placement_group_fallback_strategy,
             "max_replicas_per_node": self.max_replicas_per_node,
         }
 

python/ray/serve/_private/default_impl.py

@@ -63,6 +63,7 @@ def _default_create_placement_group(
         _soft_target_node_id=request.target_node_id,
         name=request.name,
         lifetime="detached",
+        bundle_label_selector=request.bundle_label_selector,
     )