[core] Add metrics for object size distribution in object store

src/ray/object_manager/plasma/stats_collector.cc

@@ -35,15 +35,23 @@ void ObjectStatsCollector::OnObjectCreated(const LocalObject &obj) {
   if (kSource == plasma::flatbuf::ObjectSource::CreatedByWorker) {
     num_objects_created_by_worker_++;
     num_bytes_created_by_worker_ += kObjectSize;
+    ray::stats::STATS_object_store_dist.Record(
+        kObjectSize, {{ray::stats::SourceKey, "CreatedByWorker"}});
   } else if (kSource == plasma::flatbuf::ObjectSource::RestoredFromStorage) {
     num_objects_restored_++;
     num_bytes_restored_ += kObjectSize;
+    ray::stats::STATS_object_store_dist.Record(
+        kObjectSize, {{ray::stats::SourceKey, "RestoredFromStorage"}});
   } else if (kSource == plasma::flatbuf::ObjectSource::ReceivedFromRemoteRaylet) {
     num_objects_received_++;
     num_bytes_received_ += kObjectSize;
+    ray::stats::STATS_object_store_dist.Record(
+        kObjectSize, {{ray::stats::SourceKey, "ReceivedFromRemoteRaylet"}});
   } else if (kSource == plasma::flatbuf::ObjectSource::ErrorStoredByRaylet) {
     num_objects_errored_++;
     num_bytes_errored_ += kObjectSize;
+    ray::stats::STATS_object_store_dist.Record(
+        kObjectSize, {{ray::stats::SourceKey, "ErrorStoredByRaylet"}});
   }
 
   RAY_CHECK(!obj.Sealed());
@@ -246,4 +254,4 @@ int64_t ObjectStatsCollector::GetNumObjectsUnsealed() const {
   return num_objects_unsealed_;
 }
 
-}  // namespace plasma
+}  // namespace plasma

src/ray/stats/metric_defs.cc

@@ -98,6 +98,25 @@ DEFINE_stats(
     (),
     ray::stats::GAUGE);
 
+double operator""_MB(unsigned long long int x) {
+  return static_cast<double>(1024L * 1024L * x);
+}
+
+DEFINE_stats(object_store_dist,
+             "The distribution of object size in bytes",
+             ("Source"),
+             ({32_MB,
+               64_MB,
+               128_MB,
+               256_MB,
+               512_MB,
+               1024_MB,
+               2048_MB,
+               4096_MB,
+               8192_MB,
+               16384_MB}),
+             ray::stats::HISTOGRAM);
+
 /// Placement group metrics from the GCS.
 DEFINE_stats(placement_groups,
              "Number of placement groups broken down by state.",

src/ray/stats/metric_defs.h

@@ -108,6 +108,7 @@ DECLARE_stats(gcs_task_manager_task_events_reported);
 
 /// Object Store
 DECLARE_stats(object_store_memory);
+DECLARE_stats(object_store_dist);
 
 /// Placement Group
 DECLARE_stats(gcs_placement_group_creation_latency_ms);

src/ray/stats/tag_defs.cc

@@ -47,5 +47,7 @@ const TagKeyType NameKey = TagKeyType::Register("Name");
 const TagKeyType LocationKey = TagKeyType::Register("Location");
 
 const TagKeyType ObjectStateKey = TagKeyType::Register("ObjectState");
+
+const TagKeyType SourceKey = TagKeyType::Register("Source");
 }  // namespace stats
 }  // namespace ray

src/ray/stats/tag_defs.h

@@ -48,6 +48,8 @@ extern const TagKeyType SessionNameKey;
 
 extern const TagKeyType NameKey;
 
+extern const TagKeyType SourceKey;
+
 // Object store memory location tag constants
 extern const TagKeyType LocationKey;
 constexpr char kObjectLocMmapShm[] = "MMAP_SHM";