Merge branch 'master' into llm-ner-test

Author: GokuMohandas

cpp/BUILD.bazel

@@ -361,7 +361,7 @@ cc_test(
         "metric_example.so",
     ],
     linkstatic = True,
-    tags = ["team:serverless"],
+    tags = ["team:core"],
     deps = [
         "ray_cpp_lib",
         "@boost//:callable_traits",

doc/source/cluster/kubernetes/user-guides/kuberay-gcs-persistent-ft.md

@@ -110,7 +110,7 @@ to set `appendfsync` to `always` so Redis stores all writes immediately.
 ## Putting it together
 
 Edit
-[the full YAML](https://github.com/ray-project/kuberay/blob/master/config/samples/ray-cluster.persistent-redis.yaml)
+[the full YAML](https://github.com/ray-project/kuberay/blob/release-1.3/ray-operator/config/samples/ray-cluster.persistent-redis.yaml)
 to your satisfaction and apply it:
 
 ```

doc/source/data/working-with-llms.rst

@@ -243,3 +243,39 @@ Data for the following features and attributes is collected to improve Ray Data
 
 If you would like to opt-out from usage data collection, you can follow :ref:`Ray usage stats <ref-usage-stats>`
 to turn it off.
+
+.. _production_guide:
+
+Production guide
+--------------------------------------------------
+
+.. _model_cache:
+
+Caching model weight to remote object storage
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+While deploying Ray Data LLM to large scale clusters, model loading may be rate
+limited by HuggingFace. In this case, you can cache the model to remote object
+storage (AWS S3 or Google Cloud Storage) for more stable model loading.
+
+Ray Data LLM provides the following utility to help uploading models to remote object storage.
+
+.. testcode::
+
+    # Download model from HuggingFace, and upload to GCS
+    python -m ray.llm.utils.upload_model \
+        --model-source facebook/opt-350m \
+        --bucket-uri gs://my-bucket/path/to/facebook-opt-350m
+    # Or upload a local custom model to S3
+    python -m ray.llm.utils.upload_model \
+        --model-source local/path/to/model \
+        --bucket-uri s3://my-bucket/path/to/model_name
+
+And later you can use remote object store URI as `model_source` in the config.
+
+.. testcode::
+
+    config = vLLMEngineProcessorConfig(
+        model_source="gs://my-bucket/path/to/facebook-opt-350m",  # or s3://my-bucket/path/to/model_name
+        ...
+    )

doc/source/workflows/comparison.rst

@@ -21,12 +21,12 @@ Specifying the workflow id allows for resuming of the workflow by its id in case
 Other Workflow Engines
 ----------------------
 
-Note: these comparisons are inspired by the `Serverless workflows comparisons repo <https://github.com/serverlessworkflow/specification/tree/main/comparisons>`__.
+Note: these comparisons are inspired by the `Serverless workflows comparisons repo <https://github.com/serverlessworkflow/specification/tree/0.9.x/comparisons>`__.
 
 Argo API Comparison
 ~~~~~~~~~~~~~~~~~~~
 
-The original source of these comparisons can be `found here <https://github.com/serverlessworkflow/specification/blob/main/comparisons/comparison-argo.md>`__.
+The original source of these comparisons can be `found here <https://github.com/serverlessworkflow/specification/blob/0.9.x/comparisons/comparison-argo.md>`__.
 
 Conditionals
 ^^^^^^^^^^^^
@@ -162,7 +162,7 @@ Trip Booking
 Google Cloud Workflows API Comparison
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-The original source of these comparisons can be `found here <https://github.com/serverlessworkflow/specification/blob/main/comparisons/comparison-google-cloud-workflows.md>`__.
+The original source of these comparisons can be `found here <https://github.com/serverlessworkflow/specification/blob/0.9.x/comparisons/comparison-google-cloud-workflows.md>`__.
 
 Data Conditional
 ^^^^^^^^^^^^^^^^

python/ray/_private/label_utils.py

@@ -0,0 +1,128 @@
+import json
+import re
+import yaml
+from typing import Dict
+
+import ray._private.ray_constants as ray_constants
+
+# Regex patterns used to validate that labels conform to Kubernetes label syntax rules.
+# https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#syntax-and-character-set
+
+# Regex for mandatory name (DNS label) or value
+# Examples:
+#   Valid matches: "a", "label-name", "a-._b", "123", "this_is_a_valid_label"
+#   Invalid matches: "-abc", "abc-", "my@label", "a" * 64
+LABEL_REGEX = re.compile(r"[a-zA-Z0-9]([a-zA-Z0-9_.-]*[a-zA-Z0-9]){0,62}")
+
+# Regex for optional prefix (DNS subdomain)
+# Examples:
+#   Valid matches: "abc", "sub.domain.example", "my-label", "123.456.789"
+#   Invalid matches: "-abc", "prefix_", "sub..domain", sub.$$.example
+LABEL_PREFIX_REGEX = rf"^({LABEL_REGEX.pattern}?(\.{LABEL_REGEX.pattern}?)*)$"
+
+
+def parse_node_labels_json(labels_json: str) -> Dict[str, str]:
+    labels = json.loads(labels_json)
+    if not isinstance(labels, dict):
+        raise ValueError("The format after deserialization is not a key-value pair map")
+    for key, value in labels.items():
+        if not isinstance(key, str):
+            raise ValueError("The key is not string type.")
+        if not isinstance(value, str):
+            raise ValueError(f'The value of the "{key}" is not string type')
+
+    return labels
+
+
+def parse_node_labels_string(labels_str: str) -> Dict[str, str]:
+    labels = {}
+
+    # Remove surrounding quotes if they exist
+    if len(labels_str) > 1 and labels_str.startswith('"') and labels_str.endswith('"'):
+        labels_str = labels_str[1:-1]
+
+    if labels_str == "":
+        return labels
+
+    # Labels argument should consist of a string of key=value pairs
+    # separated by commas. Labels follow Kubernetes label syntax.
+    label_pairs = labels_str.split(",")
+    for pair in label_pairs:
+        # Split each pair by `=`
+        key_value = pair.split("=")
+        if len(key_value) != 2:
+            raise ValueError("Label string is not a key-value pair.")
+        key = key_value[0].strip()
+        value = key_value[1].strip()
+        labels[key] = value
+
+    # Validate parsed node labels follow expected Kubernetes label syntax
+    validate_node_label_syntax(labels)
+
+    return labels
+
+
+def parse_node_labels_from_yaml_file(path: str) -> Dict[str, str]:
+    if path == "":
+        return {}
+    with open(path, "r") as file:
+        # Expects valid YAML content
+        labels = yaml.safe_load(file)
+        if not isinstance(labels, dict):
+            raise ValueError(
+                "The format after deserialization is not a key-value pair map."
+            )
+        for key, value in labels.items():
+            if not isinstance(key, str):
+                raise ValueError("The key is not string type.")
+            if not isinstance(value, str):
+                raise ValueError(f'The value of "{key}" is not string type.')
+
+    # Validate parsed node labels follow expected Kubernetes label syntax
+    validate_node_label_syntax(labels)
+
+    return labels
+
+
+def validate_node_labels(labels: Dict[str, str]):
+    if labels is None:
+        return
+    for key in labels.keys():
+        if key.startswith(ray_constants.RAY_DEFAULT_LABEL_KEYS_PREFIX):
+            raise ValueError(
+                f"Custom label keys `{key}` cannot start with the prefix "
+                f"`{ray_constants.RAY_DEFAULT_LABEL_KEYS_PREFIX}`. "
+                f"This is reserved for Ray defined labels."
+            )
+
+
+# TODO (ryanaoleary@): This function will replace `validate_node_labels` after
+# the migration from NodeLabelSchedulingPolicy to the Label Selector API is complete.
+def validate_node_label_syntax(labels: Dict[str, str]):
+    if labels is None:
+        return
+    for key in labels.keys():
+        if "/" in key:
+            prefix, name = key.rsplit("/")
+            if len(prefix) > 253 or not re.match(LABEL_PREFIX_REGEX, prefix):
+                raise ValueError(
+                    f"Invalid label key prefix `{prefix}`. Prefix must be a series of DNS labels "
+                    f"separated by dots (.),not longer than 253 characters in total."
+                )
+        else:
+            name = key
+        if len(name) > 63 or not re.match(LABEL_REGEX, name):
+            raise ValueError(
+                f"Invalid label key name `{name}`. Name must be 63 chars or less beginning and ending "
+                f"with an alphanumeric character ([a-z0-9A-Z]) with dashes (-), underscores (_),"
+                f"dots (.), and alphanumerics between."
+            )
+        value = labels.get(key)
+        if value is None or value == "":
+            return
+        if len(value) > 63 or not re.match(LABEL_REGEX, value):
+            raise ValueError(
+                f"Invalid label key value `{value}`. Value must be 63 chars or less beginning and ending "
+                f"with an alphanumeric character ([a-z0-9A-Z]) with dashes (-), underscores (_),"
+                f"dots (.), and alphanumerics between."
+            )

python/ray/_private/node.py

@@ -26,6 +26,7 @@
 from ray._raylet import GcsClient, get_session_key_from_storage
 from ray._private.resource_spec import ResourceSpec
 from ray._private.services import serialize_config, get_address
+from ray._private.resource_isolation_config import ResourceIsolationConfig
 from ray._private.utils import (
     open_log,
     try_to_create_directory,
@@ -89,6 +90,9 @@ def __init__(
         self.kernel_fate_share = bool(
             spawn_reaper and ray._private.utils.detect_fate_sharing_support()
         )
+        self.resource_isolation_config: ResourceIsolationConfig = (
+            ray_params.resource_isolation_config
+        )
         self.all_processes: dict = {}
         self.removal_lock = threading.Lock()
 
@@ -1236,7 +1240,6 @@ def start_raylet(
         object_store_memory: int,
         use_valgrind: bool = False,
         use_profiler: bool = False,
-        enable_physical_mode: bool = False,
     ):
         """Start the raylet.
 
@@ -1320,7 +1323,7 @@ def start_raylet(
             node_name=self._ray_params.node_name,
             webui=self._webui_url,
             labels=self._get_node_labels(),
-            enable_physical_mode=enable_physical_mode,
+            resource_isolation_config=self.resource_isolation_config,
         )
         assert ray_constants.PROCESS_TYPE_RAYLET not in self.all_processes
         self.all_processes[ray_constants.PROCESS_TYPE_RAYLET] = [process_info]
@@ -1489,6 +1492,7 @@ def start_ray_processes(self):
         # Make sure we don't call `determine_plasma_store_config` multiple
         # times to avoid printing multiple warnings.
         resource_spec = self.get_resource_spec()
+
         (
             plasma_directory,
             fallback_directory,
@@ -1500,6 +1504,11 @@ def start_ray_processes(self):
             fallback_directory=self._fallback_directory,
             huge_pages=self._ray_params.huge_pages,
         )
+
+        # add plasma store memory to the total system reserved memory
+        if self.resource_isolation_config.is_enabled():
+            self.resource_isolation_config.add_object_store_memory(object_store_memory)
+
         self.start_raylet(plasma_directory, fallback_directory, object_store_memory)
         if self._ray_params.include_log_monitor:
             self.start_log_monitor()

python/ray/_private/parameter.py

@@ -3,10 +3,10 @@
 from typing import Dict, List, Optional
 
 import ray._private.ray_constants as ray_constants
-from ray._private.utils import (
-    validate_node_labels,
-    check_ray_client_dependencies_installed,
-)
+
+from ray._private.resource_isolation_config import ResourceIsolationConfig
+from ray._private.label_utils import validate_node_labels
+from ray._private.utils import check_ray_client_dependencies_installed
 
 
 logger = logging.getLogger(__name__)
@@ -127,9 +127,8 @@ class RayParams:
         session_name: The name of the session of the ray cluster.
         webui: The url of the UI.
         cluster_id: The cluster ID in hex string.
-        enable_physical_mode: Whether physical mode is enabled, which applies
-            constraint to tasks' resource consumption. As of now, only memory resource
-            is supported.
+        resource_isolation_config: settings for cgroupv2 based isolation of ray
+            system processes (defaults to no isolation if config not provided)
     """
 
     def __init__(
@@ -193,7 +192,7 @@ def __init__(
         webui: Optional[str] = None,
         cluster_id: Optional[str] = None,
         node_id: Optional[str] = None,
-        enable_physical_mode: bool = False,
+        resource_isolation_config: Optional[ResourceIsolationConfig] = None,
     ):
         self.redis_address = redis_address
         self.gcs_address = gcs_address
@@ -257,7 +256,12 @@ def __init__(
         self._check_usage()
         self.cluster_id = cluster_id
         self.node_id = node_id
-        self.enable_physical_mode = enable_physical_mode
+
+        self.resource_isolation_config = resource_isolation_config
+        if not self.resource_isolation_config:
+            self.resource_isolation_config = ResourceIsolationConfig(
+                enable_resource_isolation=False
+            )
 
         # Set the internal config options for object reconstruction.
         if enable_object_reconstruction:

python/ray/_private/ray_constants.py

@@ -70,13 +70,38 @@ def env_set_by_user(key):
 # The default maximum number of bytes to allocate to the object store unless
 # overridden by the user.
 DEFAULT_OBJECT_STORE_MAX_MEMORY_BYTES = env_integer(
-    "RAY_DEFAULT_OBJECT_STORE_MAX_MEMORY_BYTES", 200 * 10**9  # 200 GB
+    "RAY_DEFAULT_OBJECT_STORE_MAX_MEMORY_BYTES", (200) * (10**9)  # 200 GB
 )
 # The default proportion of available memory allocated to the object store
 DEFAULT_OBJECT_STORE_MEMORY_PROPORTION = env_float(
     "RAY_DEFAULT_OBJECT_STORE_MEMORY_PROPORTION",
     0.3,
 )
+
+# The following values are only used when resource isolation is enabled
+# ===== The default number of bytes to reserve for ray system processes
+DEFAULT_SYSTEM_RESERVED_MEMORY_BYTES = env_integer(
+    "RAY_DEFAULT_DEFAULT_SYSTEM_RESERVED_MEMORY_BYTES", (25) * (10**9)
+)
+# The default proportion available memory to reserve for ray system processes
+DEFAULT_SYSTEM_RESERVED_MEMORY_PROPORTION = env_integer(
+    "RAY_DEFAULT_SYSTEM_RESERVED_MEMORY_PROPORTION", 0.10
+)
+# The default number of cpu cores to reserve for ray system processes
+DEFAULT_SYSTEM_RESERVED_CPU_CORES = env_float(
+    "RAY_DEFAULT_SYSTEM_RESERVED_CPU_CORES", 1.0
+)
+# The default proportion of cpu cores to reserve for ray system processes
+DEFAULT_SYSTEM_RESERVED_CPU_PROPORTION = env_float(
+    "RAY_DEFAULT_SYSTEM_RESERVED_CPU_PROPORTION", 0.05
+)
+# The smallest number of cores that ray system processes can be guaranteed
+MINIMUM_SYSTEM_RESERVED_CPU_CORES = 0.5
+# The smallest number of bytes that ray system processes can be guaranteed
+MINIMUM_SYSTEM_RESERVED_MEMORY_BYTES = (100) * (10**6)
+# The default path for cgroupv2
+DEFAULT_CGROUP_PATH = "/sys/fs/cgroup"
+
 # The smallest cap on the memory used by the object store that we allow.
 # This must be greater than MEMORY_RESOURCE_UNIT_BYTES
 OBJECT_STORE_MINIMUM_MEMORY_BYTES = 75 * 1024 * 1024
@@ -95,7 +120,7 @@ def env_set_by_user(key):
 # (see https://github.com/ray-project/ray/issues/20388 for details)
 # The workaround here is to limit capacity to 2GB for Mac by default,
 # and raise error if the capacity is overwritten by user.
-MAC_DEGRADED_PERF_MMAP_SIZE_LIMIT = 2 * 2**30
+MAC_DEGRADED_PERF_MMAP_SIZE_LIMIT = (2) * (2**30)
 # If a user does not specify a port for the primary Ray service,
 # we attempt to start the service running at this port.
 DEFAULT_PORT = 6379