[core][autoscaler][IPPR] Initial impl for resizing Pods in-place to the maximum configured by the user

Signed-off-by: Rueian <[email protected]>

Author: rueian

ci/lint/pydoclint-baseline.txt

@@ -662,10 +662,6 @@ python/ray/autoscaler/v2/autoscaler.py
     DOC103: Method `Autoscaler.__init__`: Docstring arguments are different from function arguments. (Or could be other formatting issues: https://jsh9.github.io/pydoclint/violation_codes.html#notes-on-doc103 ). Arguments in the function signature but not in the docstring: [config_reader: IConfigReader, event_logger: Optional[AutoscalerEventLogger], gcs_client: GcsClient, metrics_reporter: Optional[AutoscalerMetricsReporter], session_name: str].
 --------------------
 python/ray/autoscaler/v2/instance_manager/cloud_providers/kuberay/cloud_provider.py
-    DOC001: Method `__init__` Potential formatting errors in docstring. Error message: No specification for "Args": ""
-    DOC001: Function/method `__init__`: Potential formatting errors in docstring. Error message: No specification for "Args": "" (Note: DOC001 could trigger other unrelated violations under this function/method too. Please fix the docstring formatting first.)
-    DOC101: Method `KubeRayProvider.__init__`: Docstring contains fewer arguments than in function signature.
-    DOC103: Method `KubeRayProvider.__init__`: Docstring arguments are different from function arguments. (Or could be other formatting issues: https://jsh9.github.io/pydoclint/violation_codes.html#notes-on-doc103 ). Arguments in the function signature but not in the docstring: [cluster_name: str, k8s_api_client: Optional[IKubernetesHttpApiClient], provider_config: Dict[str, Any]].
     DOC101: Method `KubeRayProvider._get_workers_delete_info`: Docstring contains fewer arguments than in function signature.
     DOC103: Method `KubeRayProvider._get_workers_delete_info`: Docstring arguments are different from function arguments. (Or could be other formatting issues: https://jsh9.github.io/pydoclint/violation_codes.html#notes-on-doc103 ). Arguments in the function signature but not in the docstring: [node_set: Set[CloudInstanceId], ray_cluster_spec: Dict[str, Any]].
     DOC201: Method `KubeRayProvider._cloud_instance_from_pod` does not have a return section in docstring

python/ray/autoscaler/_private/kuberay/node_provider.py

@@ -259,7 +259,12 @@ def get(self, path: str) -> Dict[str, Any]:
         pass
 
     @abstractmethod
-    def patch(self, path: str, payload: List[Dict[str, Any]]) -> Dict[str, Any]:
+    def patch(
+        self,
+        path: str,
+        payload: List[Dict[str, Any]],
+        content_type: str = "application/json-patch+json",
+    ) -> Dict[str, Any]:
         """Wrapper for REST PATCH of resource with proper headers."""
         pass
 
@@ -311,12 +316,18 @@ def get(self, path: str) -> Dict[str, Any]:
             result.raise_for_status()
         return result.json()
 
-    def patch(self, path: str, payload: List[Dict[str, Any]]) -> Dict[str, Any]:
+    def patch(
+        self,
+        path: str,
+        payload: List[Dict[str, Any]],
+        content_type: str = "application/json-patch+json",
+    ) -> Dict[str, Any]:
         """Wrapper for REST PATCH of resource with proper headers
 
         Args:
             path: The part of the resource path that starts with the resource type.
             payload: The JSON patch payload.
+            content_type: The content type of the merge strategy.
 
         Returns:
             The JSON response of the PATCH request.
@@ -333,7 +344,7 @@ def patch(self, path: str, payload: List[Dict[str, Any]]) -> Dict[str, Any]:
         result = requests.patch(
             url,
             json.dumps(payload),
-            headers={**headers, "Content-type": "application/json-patch+json"},
+            headers={**headers, "Content-type": content_type},
             verify=verify,
         )
         if not result.status_code == 200:

python/ray/autoscaler/v2/BUILD

@@ -22,6 +22,7 @@ py_test_module_list(
         "tests/test_instance_manager.py",
         "tests/test_instance_storage.py",
         "tests/test_instance_util.py",
+        "tests/test_ippr_provider.py",
         "tests/test_metrics_reporter.py",
         "tests/test_node_provider.py",
         "tests/test_ray_installer.py",

python/ray/autoscaler/v2/autoscaler.py

@@ -101,6 +101,7 @@ def _init_cloud_instance_provider(
             self._cloud_instance_provider = KubeRayProvider(
                 config.get_config("cluster_name"),
                 provider_config,
+                gcs_client=self._gcs_client,
             )
         elif config.provider == Provider.READ_ONLY:
             provider_config["gcs_address"] = self._gcs_client.address

python/ray/autoscaler/v2/instance_manager/cloud_providers/kuberay/cloud_provider.py

@@ -23,6 +23,10 @@
     worker_delete_patch,
     worker_replica_patch,
 )
+from ray._raylet import GcsClient
+from ray.autoscaler.v2.instance_manager.cloud_providers.kuberay.ippr_provider import (
+    KubeRayIPPRProvider,
+)
 from ray.autoscaler.v2.instance_manager.node_provider import (
     CloudInstance,
     CloudInstanceId,
@@ -32,7 +36,7 @@
     NodeKind,
     TerminateNodeError,
 )
-from ray.autoscaler.v2.schema import NodeType
+from ray.autoscaler.v2.schema import IPPRSpecs, IPPRStatus, NodeType
 
 logger = logging.getLogger(__name__)
 
@@ -50,14 +54,19 @@ def __init__(
         self,
         cluster_name: str,
         provider_config: Dict[str, Any],
+        gcs_client: Optional[GcsClient] = None,
         k8s_api_client: Optional[IKubernetesHttpApiClient] = None,
     ):
         """
+        Initializes a new KubeRayProvider.
+
         Args:
             cluster_name: The name of the RayCluster resource.
-            provider_config: The namespace of the RayCluster.
-            k8s_api_client: The client to the Kubernetes API server.
-                This could be used to mock the Kubernetes API server for testing.
+            provider_config: The configuration dictionary
+                for the RayCluster (e.g., namespace and provider-specific settings).
+            gcs_client: The client to the GCS server.
+            k8s_api_client: The client to the Kubernetes
+                API server. This can be used to mock the Kubernetes API server for testing.
         """
         self._cluster_name = cluster_name
         self._namespace = provider_config["namespace"]
@@ -74,6 +83,9 @@ def __init__(
         # Below are states that are fetched from the Kubernetes API server.
         self._ray_cluster = None
         self._cached_instances: Dict[CloudInstanceId, CloudInstance]
+        self._ippr_provider = KubeRayIPPRProvider(
+            gcs_client=gcs_client, k8s_api_client=self._k8s_api_client
+        )
 
     @dataclass
     class ScaleRequest:
@@ -182,6 +194,31 @@ def poll_errors(self) -> List[CloudInstanceProviderError]:
         self._terminate_errors_queue = []
         return errors
 
+    def get_ippr_specs(self) -> IPPRSpecs:
+        """Return the cached, validated IPPR specs for the cluster.
+
+        The IPPR specs are refreshed during the provider's periodic sync with the
+        API server by reading the RayCluster annotation and validating it against
+        the IPPR schema.
+        """
+        return self._ippr_provider.get_ippr_specs()
+
+    def get_ippr_statuses(self) -> Dict[str, IPPRStatus]:
+        """Return the latest per-pod IPPR statuses keyed by pod name.
+
+        These statuses are refreshed from the current pod list during the provider's
+        periodic sync with the API server.
+        """
+        return self._ippr_provider.get_ippr_statuses()
+
+    def do_ippr_requests(self, resizes: List[IPPRStatus]) -> None:
+        """Execute IPPR resize requests via the underlying IPPR provider.
+
+        Args:
+            resizes: The list of per-pod IPPR actions produced by the scheduler.
+        """
+        self._ippr_provider.do_ippr_requests(resizes)
+
     ############################
     # Private
     ############################
@@ -393,7 +430,9 @@ def _add_terminate_errors(
     def _sync_with_api_server(self) -> None:
         """Fetches the RayCluster resource from the Kubernetes API server."""
         self._ray_cluster = self._get(f"rayclusters/{self._cluster_name}")
+        self._ippr_provider.validate_and_set_ippr_specs(self._ray_cluster)
         self._cached_instances = self._fetch_instances()
+        self._ippr_provider.sync_with_raylets()
 
     @property
     def ray_cluster(self) -> Dict[str, Any]:
@@ -499,6 +538,9 @@ def _fetch_instances(self) -> Dict[CloudInstanceId, CloudInstance]:
             cloud_instance = self._cloud_instance_from_pod(pod)
             if cloud_instance:
                 cloud_instances[pod_name] = cloud_instance
+
+        self._ippr_provider.sync_ippr_status_from_pods(pod_list["items"])
+
         return cloud_instances
 
     @staticmethod