[core][autoscaler] Fix idle resource status accounting from raylet #3…

…8268 (#38285)


---------

Signed-off-by: rickyyx <rickyx@anyscale.com>

python/ray/autoscaler/v2/BUILD

@@ -24,15 +24,15 @@ py_test(
     name = "test_storage",
     size = "small",
     srcs = ["tests/test_storage.py"],
-    tags = ["team:core"],
+    tags = ["team:core", "exclusive"],
     deps = ["//:ray_lib",],
 )
 
 py_test(
     name = "test_config",
     size = "small",
     srcs = ["tests/test_config.py"],
-    tags = ["team:core"],
+    tags = ["team:core", "exclusive"],
     deps = [
         "//:ray_lib",
     ]
@@ -42,39 +42,39 @@ py_test(
     name = "test_node_provider",
     size = "small",
     srcs = ["tests/test_node_provider.py"],
-    tags = ["team:core"],
+    tags = ["team:core", "exclusive"],
     deps = ["//:ray_lib",],
 )
 
 py_test(
     name = "test_ray_installer",
     size = "small",
     srcs = ["tests/test_ray_installer.py"],
-    tags = ["team:core"],
+    tags = ["team:core", "exclusive"],
     deps = ["//:ray_lib",],
 )
 
 py_test(
     name = "test_instance_launcher",
     size = "small",
     srcs = ["tests/test_instance_launcher.py"],
-    tags = ["team:core"],
+    tags = ["team:core", "exclusive"],
     deps = ["//:ray_lib",],
 )
 
 py_test(
     name = "test_reconciler",
     size = "small",
     srcs = ["tests/test_reconciler.py"],
-    tags = ["team:core"],
+    tags = ["team:core", "exclusive"],
     deps = ["//:ray_lib",],
 )
 
 py_test(
     name = "test_threaded_ray_installer",
     size = "small",
     srcs = ["tests/test_threaded_ray_installer.py"],
-    tags = ["team:core"],
+    tags = ["team:core", "exclusive"],
     deps = ["//:ray_lib",],
 )
 
@@ -90,7 +90,7 @@ py_test(
     name = "test_utils",
     size = "small",
     srcs = ["tests/test_utils.py"],
-    tags = ["team:core"],
+    tags = ["team:core", "exclusive"],
     deps = ["//:ray_lib", ":conftest"],
 )
 

python/ray/autoscaler/v2/tests/test_e2e.py

@@ -8,6 +8,7 @@
 from ray._private.test_utils import run_string_as_driver_nonblocking, wait_for_condition
 from ray.autoscaler.v2.sdk import get_cluster_status
 from ray.cluster_utils import AutoscalingCluster
+from ray.util.placement_group import placement_group, remove_placement_group
 from ray.util.state.api import list_placement_groups, list_tasks
 
 
@@ -156,6 +157,45 @@ def pg_created():
         cluster.shutdown()
 
 
+def test_placement_group_removal_idle_node():
+    # Test that nodes become idle after placement group removal.
+    cluster = AutoscalingCluster(
+        head_resources={"CPU": 2},
+        worker_node_types={
+            "type-1": {
+                "resources": {"CPU": 2},
+                "node_config": {},
+                "min_workers": 0,
+                "max_workers": 2,
+            },
+        },
+    )
+    try:
+        cluster.start()
+        ray.init("auto")
+
+        # Schedule a pg on nodes
+        pg = placement_group([{"CPU": 2}] * 3, strategy="STRICT_SPREAD")
+        ray.get(pg.ready())
+
+        time.sleep(2)
+        remove_placement_group(pg)
+
+        time.sleep(1)
+        from ray.autoscaler.v2.sdk import get_cluster_status
+
+        cluster_state = get_cluster_status()
+
+        # Verify that nodes are idle.
+        assert len((cluster_state.healthy_nodes)) == 3
+        for node in cluster_state.healthy_nodes:
+            assert node.node_status == "IDLE"
+            assert node.resource_usage.idle_time_ms >= 1000
+    finally:
+        ray.shutdown()
+        cluster.shutdown()
+
+
 if __name__ == "__main__":
     if os.environ.get("PARALLEL_CI"):
         sys.exit(pytest.main(["-n", "auto", "--boxed", "-vs", __file__]))

src/ray/raylet/scheduling/cluster_resource_scheduler_test.cc

@@ -1036,6 +1036,9 @@ TEST_F(ClusterResourceSchedulerTest, TaskGPUResourceInstancesTest) {
                            available_gpu_instances.end(),
                            expected_available_gpu_instances.begin()));
 
+    ASSERT_FALSE(
+        resource_scheduler.GetLocalResourceManager().GetResourceIdleTime().has_value());
+
     resource_scheduler.GetLocalResourceManager().AddResourceInstances(
         ResourceID::GPU(), allocate_gpu_instances);
     available_gpu_instances = resource_scheduler.GetLocalResourceManager()
@@ -1047,10 +1050,16 @@ TEST_F(ClusterResourceSchedulerTest, TaskGPUResourceInstancesTest) {
                            available_gpu_instances.end(),
                            expected_available_gpu_instances.begin()));
 
+    ASSERT_TRUE(
+        resource_scheduler.GetLocalResourceManager().GetResourceIdleTime().has_value());
+
     allocate_gpu_instances = {1.5, 1.5, .5, 1.5};
     std::vector<double> underflow =
         resource_scheduler.GetLocalResourceManager().SubtractResourceInstances(
             ResourceID::GPU(), allocate_gpu_instances);
+    ASSERT_FALSE(
+        resource_scheduler.GetLocalResourceManager().GetResourceIdleTime().has_value());
+
     std::vector<double> expected_underflow{.5, .5, 0., .5};
     ASSERT_TRUE(
         std::equal(underflow.begin(), underflow.end(), expected_underflow.begin()));
@@ -1067,6 +1076,8 @@ TEST_F(ClusterResourceSchedulerTest, TaskGPUResourceInstancesTest) {
     std::vector<double> overflow =
         resource_scheduler.GetLocalResourceManager().AddResourceInstances(
             ResourceID::GPU(), allocate_gpu_instances);
+    ASSERT_FALSE(
+        resource_scheduler.GetLocalResourceManager().GetResourceIdleTime().has_value());
     std::vector<double> expected_overflow{.0, .0, .5, 0.};
     ASSERT_TRUE(std::equal(overflow.begin(), overflow.end(), expected_overflow.begin()));
     available_gpu_instances = resource_scheduler.GetLocalResourceManager()

src/ray/raylet/scheduling/local_resource_manager.cc

@@ -51,6 +51,7 @@ void LocalResourceManager::AddLocalResourceInstances(
 void LocalResourceManager::DeleteLocalResource(scheduling::ResourceID resource_id) {
   local_resources_.available.Remove(resource_id);
   local_resources_.total.Remove(resource_id);
+  resources_last_idle_time_.erase(resource_id);
   OnResourceChanged();
 }
 
@@ -275,13 +276,21 @@ std::vector<double> LocalResourceManager::SubtractResourceInstances(
       resource_instances_fp,
       local_resources_.available.GetMutable(resource_id),
       allow_going_negative);
+
+  // If there's any non 0 instance delta to be subtracted, the source should be marked as
+  // non-idle.
+  for (const auto &to_subtract_instance : resource_instances_fp) {
+    if (to_subtract_instance > 0) {
+      SetResourceNonIdle(resource_id);
+      break;
+    }
+  }
   OnResourceChanged();
 
   return FixedPointVectorToDouble(underflow);
 }
 
 void LocalResourceManager::SetResourceNonIdle(const scheduling::ResourceID &resource_id) {
-  // We o
   resources_last_idle_time_[resource_id] = absl::nullopt;
 }
 

src/ray/raylet/scheduling/local_resource_manager.h

@@ -290,6 +290,7 @@ class LocalResourceManager : public syncer::ReporterInterface {
   FRIEND_TEST(ClusterResourceSchedulerTest, TaskResourceInstanceWithHardRequestTest);
   FRIEND_TEST(ClusterResourceSchedulerTest, TaskResourceInstanceWithoutCpuUnitTest);
   FRIEND_TEST(ClusterResourceSchedulerTest, CustomResourceInstanceTest);
+  FRIEND_TEST(ClusterResourceSchedulerTest, TaskGPUResourceInstancesTest);
 
   friend class LocalResourceManagerTest;
   FRIEND_TEST(LocalResourceManagerTest, BasicGetResourceUsageMapTest);