[core][stats-die/03bis] improve scheduler_placement_time_s metric (#58217)

Change the unit of `scheduler_placement_time` from seconds to
mili-seconds. The current bucket is in the range of 0.1s to 2.5 hours
which doesn't make sense. According to a sample of data, the range we
are interested in would be from us to s. Thanks @ZacAttack for pointing
this out.

```
Note: This is an internal (non–public-facing) metric, so we only need to update its usage within Ray (e.g., the dashboard). A simple code change should suffice.
```

<img width="1609" height="421"
alt="505491038-c5d81017-b86c-406f-acf4-614560752062"
src="https://github.com/user-attachments/assets/cc647b97-42ec-42eb-bf01-4d1867940207"
/>

Test:
- CI

Signed-off-by: Cuong Nguyen <[email protected]>

Author: can-anyscale

python/ray/tests/test_scheduling_2.py

@@ -787,17 +787,17 @@ def ready(self):
     placement_metric_condition = get_metric_check_condition(
         [
             MetricSamplePattern(
-                name="ray_scheduler_placement_time_s_bucket",
+                name="ray_scheduler_placement_time_ms_bucket",
                 value=1.0,
                 partial_label_match={"WorkloadType": "Actor"},
             ),
             MetricSamplePattern(
-                name="ray_scheduler_placement_time_s_bucket",
+                name="ray_scheduler_placement_time_ms_bucket",
                 value=1.0,
                 partial_label_match={"WorkloadType": "Task"},
             ),
             MetricSamplePattern(
-                name="ray_scheduler_placement_time_s_bucket",
+                name="ray_scheduler_placement_time_ms_bucket",
                 value=1.0,
                 partial_label_match={"WorkloadType": "PlacementGroup"},
             ),

src/ray/common/metrics.h

@@ -73,15 +73,15 @@ inline ray::stats::Gauge GetObjectStoreMemoryGaugeMetric() {
   };
 }
 
-inline ray::stats::Histogram GetSchedulerPlacementTimeSHistogramMetric() {
+inline ray::stats::Histogram GetSchedulerPlacementTimeMsHistogramMetric() {
   return ray::stats::Histogram{
-      /*name=*/"scheduler_placement_time_s",
+      /*name=*/"scheduler_placement_time_ms",
       /*description=*/
-      "The time it takes for a worklod (task, actor, placement group) to "
+      "The time it takes for a workload (task, actor, placement group) to "
       "be placed. This is the time from when the tasks dependencies are "
       "resolved to when it actually reserves resources on a node to run.",
-      /*unit=*/"s",
-      /*boundaries=*/{0.1, 1, 10, 100, 1000, 10000},
+      /*unit=*/"ms",
+      /*boundaries=*/{1, 10, 100, 1000, 10000},
       /*tag_keys=*/{"WorkloadType"},
   };
 }

src/ray/common/task/task_spec.cc

@@ -612,15 +612,15 @@ bool TaskSpecification::IsRetriable() const {
 }
 
 void TaskSpecification::EmitTaskMetrics(
-    ray::observability::MetricInterface &scheduler_placement_time_s_histogram) const {
-  double duration_s = (GetMessage().lease_grant_timestamp_ms() -
-                       GetMessage().dependency_resolution_timestamp_ms()) /
-                      1000;
+    ray::observability::MetricInterface &scheduler_placement_time_ms_histogram) const {
+  double duration_ms = GetMessage().lease_grant_timestamp_ms() -
+                       GetMessage().dependency_resolution_timestamp_ms();
 
   if (IsActorCreationTask()) {
-    scheduler_placement_time_s_histogram.Record(duration_s, {{"WorkloadType", "Actor"}});
+    scheduler_placement_time_ms_histogram.Record(duration_ms,
+                                                 {{"WorkloadType", "Actor"}});
   } else {
-    scheduler_placement_time_s_histogram.Record(duration_s, {{"WorkloadType", "Task"}});
+    scheduler_placement_time_ms_histogram.Record(duration_ms, {{"WorkloadType", "Task"}});
   }
 }
 

src/ray/common/task/task_spec.h

@@ -365,7 +365,7 @@ class TaskSpecification : public MessageWrapper<rpc::TaskSpec> {
   bool IsRetriable() const;
 
   void EmitTaskMetrics(
-      ray::observability::MetricInterface &scheduler_placement_time_s_histogram) const;
+      ray::observability::MetricInterface &scheduler_placement_time_ms_histogram) const;
 
   /// \return true if task events from this task should be reported.
   bool EnableTaskEvents() const;

src/ray/core_worker/core_worker_process.cc

@@ -552,7 +552,7 @@ std::shared_ptr<CoreWorker> CoreWorkerProcessImpl::CreateCoreWorker(
         return rpc::TensorTransport::OBJECT_STORE;
       },
       boost::asio::steady_timer(io_service_),
-      *scheduler_placement_time_s_histogram_);
+      *scheduler_placement_time_ms_histogram_);
 
   auto report_locality_data_callback = [this](
                                            const ObjectID &object_id,
@@ -797,8 +797,8 @@ CoreWorkerProcessImpl::CoreWorkerProcessImpl(const CoreWorkerOptions &options)
       new ray::stats::Gauge(GetActorByStateGaugeMetric()));
   total_lineage_bytes_gauge_ = std::unique_ptr<ray::stats::Gauge>(
       new ray::stats::Gauge(GetTotalLineageBytesGaugeMetric()));
-  scheduler_placement_time_s_histogram_ = std::unique_ptr<ray::stats::Histogram>(
-      new ray::stats::Histogram(GetSchedulerPlacementTimeSHistogramMetric()));
+  scheduler_placement_time_ms_histogram_ = std::unique_ptr<ray::stats::Histogram>(
+      new ray::stats::Histogram(GetSchedulerPlacementTimeMsHistogramMetric()));
 
   // Initialize event framework before starting up worker.
   if (RayConfig::instance().event_log_reporter_enabled() && !options_.log_dir.empty()) {

src/ray/core_worker/core_worker_process.h

@@ -192,7 +192,7 @@ class CoreWorkerProcessImpl {
   std::unique_ptr<ray::stats::Gauge> task_by_state_gauge_;
   std::unique_ptr<ray::stats::Gauge> actor_by_state_gauge_;
   std::unique_ptr<ray::stats::Gauge> total_lineage_bytes_gauge_;
-  std::unique_ptr<ray::stats::Histogram> scheduler_placement_time_s_histogram_;
+  std::unique_ptr<ray::stats::Histogram> scheduler_placement_time_ms_histogram_;
 };
 }  // namespace core
 }  // namespace ray

src/ray/core_worker/task_submission/normal_task_submitter.cc

@@ -177,7 +177,7 @@ void NormalTaskSubmitter::OnWorkerIdle(
       scheduling_key_entry.num_busy_workers++;
 
       task_spec.GetMutableMessage().set_lease_grant_timestamp_ms(current_sys_time_ms());
-      task_spec.EmitTaskMetrics(scheduler_placement_time_s_histogram_);
+      task_spec.EmitTaskMetrics(scheduler_placement_time_ms_histogram_);
 
       executing_tasks_.emplace(task_spec.TaskId(), addr);
       PushNormalTask(

src/ray/core_worker/task_submission/normal_task_submitter.h

@@ -96,7 +96,7 @@ class NormalTaskSubmitter {
       std::shared_ptr<LeaseRequestRateLimiter> lease_request_rate_limiter,
       const TensorTransportGetter &tensor_transport_getter,
       boost::asio::steady_timer cancel_timer,
-      ray::observability::MetricInterface &scheduler_placement_time_s_histogram)
+      ray::observability::MetricInterface &scheduler_placement_time_ms_histogram)
       : rpc_address_(std::move(rpc_address)),
         local_raylet_client_(std::move(local_raylet_client)),
         raylet_client_pool_(std::move(raylet_client_pool)),
@@ -111,7 +111,7 @@ class NormalTaskSubmitter {
         job_id_(job_id),
         lease_request_rate_limiter_(std::move(lease_request_rate_limiter)),
         cancel_retry_timer_(std::move(cancel_timer)),
-        scheduler_placement_time_s_histogram_(scheduler_placement_time_s_histogram) {}
+        scheduler_placement_time_ms_histogram_(scheduler_placement_time_ms_histogram) {}
 
   /// Schedule a task for direct submission to a worker.
   void SubmitTask(TaskSpecification task_spec);
@@ -365,7 +365,7 @@ class NormalTaskSubmitter {
   // Retries cancelation requests if they were not successful.
   boost::asio::steady_timer cancel_retry_timer_ ABSL_GUARDED_BY(mu_);
 
-  ray::observability::MetricInterface &scheduler_placement_time_s_histogram_;
+  ray::observability::MetricInterface &scheduler_placement_time_ms_histogram_;
 };
 
 }  // namespace core

src/ray/core_worker/task_submission/tests/normal_task_submitter_test.cc

@@ -496,7 +496,7 @@ class NormalTaskSubmitterTest : public testing::Test {
         rate_limiter,
         [](const ObjectID &object_id) { return rpc::TensorTransport::OBJECT_STORE; },
         boost::asio::steady_timer(io_context),
-        fake_scheduler_placement_time_s_histogram_);
+        fake_scheduler_placement_time_ms_histogram_);
   }
 
   NodeID local_node_id;
@@ -513,7 +513,7 @@ class NormalTaskSubmitterTest : public testing::Test {
   std::unique_ptr<MockLeasePolicy> lease_policy;
   MockLeasePolicy *lease_policy_ptr = nullptr;
   instrumented_io_context io_context;
-  ray::observability::FakeHistogram fake_scheduler_placement_time_s_histogram_;
+  ray::observability::FakeHistogram fake_scheduler_placement_time_ms_histogram_;
 };
 
 TEST_F(NormalTaskSubmitterTest, TestLocalityAwareSubmitOneTask) {
@@ -1433,7 +1433,7 @@ void TestSchedulingKey(const std::shared_ptr<CoreWorkerMemoryStore> store,
                        const TaskSpecification &same2,
                        const TaskSpecification &different) {
   rpc::Address address;
-  ray::observability::FakeHistogram fake_scheduler_placement_time_s_histogram_;
+  ray::observability::FakeHistogram fake_scheduler_placement_time_ms_histogram_;
   auto local_node_id = NodeID::FromRandom();
   auto raylet_client = std::make_shared<MockRayletClient>();
   auto raylet_client_pool = std::make_shared<rpc::RayletClientPool>(
@@ -1462,7 +1462,7 @@ void TestSchedulingKey(const std::shared_ptr<CoreWorkerMemoryStore> store,
       std::make_shared<StaticLeaseRequestRateLimiter>(1),
       [](const ObjectID &object_id) { return rpc::TensorTransport::OBJECT_STORE; },
       boost::asio::steady_timer(io_context),
-      fake_scheduler_placement_time_s_histogram_);
+      fake_scheduler_placement_time_ms_histogram_);
 
   submitter.SubmitTask(same1);
   submitter.SubmitTask(same2);

src/ray/core_worker/tests/core_worker_test.cc

@@ -224,7 +224,7 @@ class CoreWorkerTest : public ::testing::Test {
         lease_request_rate_limiter,
         [](const ObjectID &object_id) { return rpc::TensorTransport::OBJECT_STORE; },
         boost::asio::steady_timer(io_service_),
-        fake_scheduler_placement_time_s_histogram_);
+        fake_scheduler_placement_time_ms_histogram_);
 
     auto actor_task_submitter = std::make_unique<ActorTaskSubmitter>(
         *core_worker_client_pool,
@@ -300,7 +300,7 @@ class CoreWorkerTest : public ::testing::Test {
   ray::observability::FakeGauge fake_task_by_state_gauge_;
   ray::observability::FakeGauge fake_actor_by_state_gauge_;
   ray::observability::FakeGauge fake_total_lineage_bytes_gauge_;
-  ray::observability::FakeHistogram fake_scheduler_placement_time_s_histogram_;
+  ray::observability::FakeHistogram fake_scheduler_placement_time_ms_histogram_;
   std::unique_ptr<FakePeriodicalRunner> fake_periodical_runner_;
 
   // Controllable time for testing publisher timeouts