Integrate metric items into raylet (#4602)

src/ray/gcs/redis_context.cc

@@ -4,6 +4,9 @@
 
 #include <sstream>
 
+#include "ray/stats/stats.h"
+#include "ray/util/util.h"
+
 extern "C" {
 #include "ray/thirdparty/hiredis/adapters/ae.h"
 #include "ray/thirdparty/hiredis/async.h"
@@ -18,13 +21,21 @@ namespace {
 /// A helper function to call the callback and delete it from the callback
 /// manager if necessary.
 void ProcessCallback(int64_t callback_index, const std::string &data) {
-  if (callback_index >= 0) {
-    bool delete_callback =
-        ray::gcs::RedisCallbackManager::instance().get(callback_index)(data);
-    // Delete the callback if necessary.
-    if (delete_callback) {
-      ray::gcs::RedisCallbackManager::instance().remove(callback_index);
-    }
+  RAY_CHECK(callback_index >= 0) << "The callback index must be greater than 0, "
+                                 << "but it actually is " << callback_index;
+  auto callback_item = ray::gcs::RedisCallbackManager::instance().get(callback_index);
+  if (!callback_item.is_subscription) {
+    // Record the redis latency for non-subscription redis operations.
+    auto end_time = current_sys_time_us();
+    ray::stats::RedisLatency().Record(end_time - callback_item.start_time);
+  }
+  // Invoke the callback.
+  if (callback_item.callback != nullptr) {
+    callback_item.callback(data);
+  }
+  if (!callback_item.is_subscription) {
+    // Delete the callback if it's not a subscription callback.
+    ray::gcs::RedisCallbackManager::instance().remove(callback_index);
   }
 }
 
@@ -104,18 +115,20 @@ void SubscribeRedisCallback(void *c, void *r, void *privdata) {
   ProcessCallback(callback_index, data);
 }
 
-int64_t RedisCallbackManager::add(const RedisCallback &function) {
-  callbacks_.emplace(num_callbacks_, function);
+int64_t RedisCallbackManager::add(const RedisCallback &function, bool is_subscription) {
+  auto start_time = current_sys_time_us();
+  callback_items_.emplace(num_callbacks_,
+                          CallbackItem(function, is_subscription, start_time));
   return num_callbacks_++;
 }
 
-RedisCallback &RedisCallbackManager::get(int64_t callback_index) {
-  RAY_CHECK(callbacks_.find(callback_index) != callbacks_.end());
-  return callbacks_[callback_index];
+RedisCallbackManager::CallbackItem &RedisCallbackManager::get(int64_t callback_index) {
+  RAY_CHECK(callback_items_.find(callback_index) != callback_items_.end());
+  return callback_items_[callback_index];
 }
 
 void RedisCallbackManager::remove(int64_t callback_index) {
-  callbacks_.erase(callback_index);
+  callback_items_.erase(callback_index);
 }
 
 #define REDIS_CHECK_ERROR(CONTEXT, REPLY)                     \
@@ -217,8 +230,7 @@ Status RedisContext::RunAsync(const std::string &command, const UniqueID &id,
                               const uint8_t *data, int64_t length,
                               const TablePrefix prefix, const TablePubsub pubsub_channel,
                               RedisCallback redisCallback, int log_length) {
-  int64_t callback_index =
-      redisCallback != nullptr ? RedisCallbackManager::instance().add(redisCallback) : -1;
+  int64_t callback_index = RedisCallbackManager::instance().add(redisCallback, false);
   if (length > 0) {
     if (log_length >= 0) {
       std::string redis_command = command + " %d %d %b %b %d";
@@ -278,7 +290,7 @@ Status RedisContext::SubscribeAsync(const ClientID &client_id,
   RAY_CHECK(pubsub_channel != TablePubsub::NO_PUBLISH)
       << "Client requested subscribe on a table that does not support pubsub";
 
-  int64_t callback_index = RedisCallbackManager::instance().add(redisCallback);
+  int64_t callback_index = RedisCallbackManager::instance().add(redisCallback, true);
   RAY_CHECK(out_callback_index != nullptr);
   *out_callback_index = callback_index;
   int status = 0;

src/ray/gcs/redis_context.h

@@ -19,9 +19,8 @@ namespace ray {
 
 namespace gcs {
 /// Every callback should take in a vector of the results from the Redis
-/// operation and return a bool indicating whether the callback should be
-/// deleted once called.
-using RedisCallback = std::function<bool(const std::string &)>;
+/// operation.
+using RedisCallback = std::function<void(const std::string &)>;
 
 class RedisCallbackManager {
  public:
@@ -30,9 +29,24 @@ class RedisCallbackManager {
     return instance;
   }
 
-  int64_t add(const RedisCallback &function);
+  struct CallbackItem {
+    CallbackItem() = default;
 
-  RedisCallback &get(int64_t callback_index);
+    CallbackItem(const RedisCallback &callback, bool is_subscription,
+                 int64_t start_time) {
+      this->callback = callback;
+      this->is_subscription = is_subscription;
+      this->start_time = start_time;
+    }
+
+    RedisCallback callback;
+    bool is_subscription;
+    int64_t start_time;
+  };
+
+  int64_t add(const RedisCallback &function, bool is_subscription);
+
+  CallbackItem &get(int64_t callback_index);
 
   /// Remove a callback.
   void remove(int64_t callback_index);
@@ -43,7 +57,7 @@ class RedisCallbackManager {
   ~RedisCallbackManager() {}
 
   int64_t num_callbacks_ = 0;
-  std::unordered_map<int64_t, RedisCallback> callbacks_;
+  std::unordered_map<int64_t, CallbackItem> callback_items_;
 };
 
 class RedisContext {

src/ray/gcs/tables.cc

@@ -48,7 +48,6 @@ Status Log<ID, Data>::Append(const DriverID &driver_id, const ID &id,
     if (done != nullptr) {
       (done)(client_, id, *dataT);
     }
-    return true;
   };
   flatbuffers::FlatBufferBuilder fbb;
   fbb.ForceDefaults(true);
@@ -73,7 +72,6 @@ Status Log<ID, Data>::AppendAt(const DriverID &driver_id, const ID &id,
         (failure)(client_, id, *dataT);
       }
     }
-    return true;
   };
   flatbuffers::FlatBufferBuilder fbb;
   fbb.ForceDefaults(true);
@@ -102,7 +100,6 @@ Status Log<ID, Data>::Lookup(const DriverID &driver_id, const ID &id,
       }
       lookup(client_, id, results);
     }
-    return true;
   };
   std::vector<uint8_t> nil;
   return GetRedisContext(id)->RunAsync("RAY.TABLE_LOOKUP", id, nil.data(), nil.size(),
@@ -154,10 +151,8 @@ Status Log<ID, Data>::Subscribe(const DriverID &driver_id, const ClientID &clien
         subscribe(client_, id, root->notification_mode(), results);
       }
     }
-    // We do not delete the callback after calling it since there may be
-    // more subscription messages.
-    return false;
   };
+
   subscribe_callback_index_ = 1;
   for (auto &context : shard_contexts_) {
     RAY_RETURN_NOT_OK(context->SubscribeAsync(client_id, pubsub_channel_, callback,
@@ -230,7 +225,6 @@ Status Table<ID, Data>::Add(const DriverID &driver_id, const ID &id,
     if (done != nullptr) {
       (done)(client_, id, *dataT);
     }
-    return true;
   };
   flatbuffers::FlatBufferBuilder fbb;
   fbb.ForceDefaults(true);
@@ -296,7 +290,6 @@ Status Set<ID, Data>::Add(const DriverID &driver_id, const ID &id,
     if (done != nullptr) {
       (done)(client_, id, *dataT);
     }
-    return true;
   };
   flatbuffers::FlatBufferBuilder fbb;
   fbb.ForceDefaults(true);
@@ -314,7 +307,6 @@ Status Set<ID, Data>::Remove(const DriverID &driver_id, const ID &id,
     if (done != nullptr) {
       (done)(client_, id, *dataT);
     }
-    return true;
   };
   flatbuffers::FlatBufferBuilder fbb;
   fbb.ForceDefaults(true);

src/ray/object_manager/connection_pool.cc

@@ -147,4 +147,25 @@ std::string ConnectionPool::DebugString() const {
   return result.str();
 }
 
+void ConnectionPool::RecordMetrics() const {
+  stats::ConnectionPoolStats().Record(
+      message_send_connections_.size(),
+      {{stats::ValueTypeKey, "num_message_send_connections"}});
+  stats::ConnectionPoolStats().Record(
+      transfer_send_connections_.size(),
+      {{stats::ValueTypeKey, "num_transfer_send_connections"}});
+  stats::ConnectionPoolStats().Record(
+      available_transfer_send_connections_.size(),
+      {{stats::ValueTypeKey, "num_avail_message_send_connections"}});
+  stats::ConnectionPoolStats().Record(
+      available_transfer_send_connections_.size(),
+      {{stats::ValueTypeKey, "num_avail_transfer_send_connections"}});
+  stats::ConnectionPoolStats().Record(
+      message_receive_connections_.size(),
+      {{stats::ValueTypeKey, "num_message_receive_connections"}});
+  stats::ConnectionPoolStats().Record(
+      transfer_receive_connections_.size(),
+      {{stats::ValueTypeKey, "num_transfer_receive_connections"}});
+}
+
 }  // namespace ray

src/ray/object_manager/connection_pool.h

@@ -19,6 +19,7 @@
 #include "ray/object_manager/format/object_manager_generated.h"
 #include "ray/object_manager/object_directory.h"
 #include "ray/object_manager/object_manager_client_connection.h"
+#include "ray/stats/stats.h"
 
 namespace asio = boost::asio;
 
@@ -95,6 +96,9 @@ class ConnectionPool {
   /// \return string.
   std::string DebugString() const;
 
+  /// Record metrics.
+  void RecordMetrics() const;
+
   /// This object cannot be copied for thread-safety.
   RAY_DISALLOW_COPY_AND_ASSIGN(ConnectionPool);
 

src/ray/object_manager/object_manager.cc

@@ -1,5 +1,6 @@
 #include "ray/object_manager/object_manager.h"
 #include "ray/common/common_protocol.h"
+#include "ray/stats/stats.h"
 #include "ray/util/util.h"
 
 namespace asio = boost::asio;
@@ -966,4 +967,19 @@ std::string ObjectManager::DebugString() const {
   return result.str();
 }
 
+void ObjectManager::RecordMetrics() const {
+  stats::ObjectManagerStats().Record(local_objects_.size(),
+                                     {{stats::ValueTypeKey, "num_local_objects"}});
+  stats::ObjectManagerStats().Record(active_wait_requests_.size(),
+                                     {{stats::ValueTypeKey, "num_active_wait_requests"}});
+  stats::ObjectManagerStats().Record(
+      unfulfilled_push_requests_.size(),
+      {{stats::ValueTypeKey, "num_unfulfilled_push_requests"}});
+  stats::ObjectManagerStats().Record(pull_requests_.size(),
+                                     {{stats::ValueTypeKey, "num_pull_requests"}});
+  stats::ObjectManagerStats().Record(profile_events_.size(),
+                                     {{stats::ValueTypeKey, "num_profile_events"}});
+  connection_pool_.RecordMetrics();
+}
+
 }  // namespace ray

src/ray/object_manager/object_manager.h

@@ -187,6 +187,9 @@ class ObjectManager : public ObjectManagerInterface {
   /// \return string.
   std::string DebugString() const;
 
+  /// Record metrics.
+  void RecordMetrics() const;
+
  private:
   friend class TestObjectManager;
 

src/ray/raylet/lineage_cache.cc

@@ -1,4 +1,5 @@
 #include "lineage_cache.h"
+#include "ray/stats/stats.h"
 
 #include <sstream>
 
@@ -471,6 +472,17 @@ std::string LineageCache::DebugString() const {
   return result.str();
 }
 
+void LineageCache::RecordMetrics() const {
+  stats::LineageCacheStats().Record(committed_tasks_.size(),
+                                    {{stats::ValueTypeKey, "num_committed_tasks"}});
+  stats::LineageCacheStats().Record(lineage_.GetChildrenSize(),
+                                    {{stats::ValueTypeKey, "num_children"}});
+  stats::LineageCacheStats().Record(subscribed_tasks_.size(),
+                                    {{stats::ValueTypeKey, "num_subscribed_tasks"}});
+  stats::LineageCacheStats().Record(lineage_.GetEntries().size(),
+                                    {{stats::ValueTypeKey, "num_lineages"}});
+}
+
 }  // namespace raylet
 
 }  // namespace ray

src/ray/raylet/lineage_cache.h

@@ -299,6 +299,9 @@ class LineageCache {
   /// \return string.
   std::string DebugString() const;
 
+  /// Record metrics.
+  void RecordMetrics() const;
+
  private:
   FRIEND_TEST(LineageCacheTest, BarReturnsZeroOnNull);
   /// Flush a task that is in UNCOMMITTED_READY state.

src/ray/raylet/node_manager.cc

@@ -34,6 +34,32 @@ int64_t GetExpectedTaskCounter(
   return expected_task_counter;
 };
 
+struct ActorStats {
+  int live_actors = 0;
+  int dead_actors = 0;
+  int reconstructing_actors = 0;
+  int max_num_handles = 0;
+};
+
+/// A helper function to return the statistical data of actors in this node manager.
+ActorStats GetActorStatisticalData(
+    std::unordered_map<ray::ActorID, ray::raylet::ActorRegistration> actor_registry) {
+  ActorStats item;
+  for (auto &pair : actor_registry) {
+    if (pair.second.GetState() == ActorState::ALIVE) {
+      item.live_actors += 1;
+    } else if (pair.second.GetState() == ActorState::RECONSTRUCTING) {
+      item.reconstructing_actors += 1;
+    } else {
+      item.dead_actors += 1;
+    }
+    if (pair.second.NumHandles() > item.max_num_handles) {
+      item.max_num_handles = pair.second.NumHandles();
+    }
+  }
+  return item;
+}
+
 }  // namespace
 
 namespace ray {
@@ -276,6 +302,7 @@ void NodeManager::Heartbeat() {
   if (debug_dump_period_ > 0 &&
       static_cast<int64_t>(now_ms - last_debug_dump_at_ms_) > debug_dump_period_) {
     DumpDebugState();
+    RecordMetrics();
     last_debug_dump_at_ms_ = now_ms;
   }
 
@@ -2151,7 +2178,7 @@ void NodeManager::ForwardTask(const Task &task, const ClientID &node_id,
       });
 }
 
-void NodeManager::DumpDebugState() {
+void NodeManager::DumpDebugState() const {
   std::fstream fs;
   fs.open(temp_dir_ + "/debug_state.txt", std::fstream::out | std::fstream::trunc);
   fs << DebugString();
@@ -2175,26 +2202,13 @@ std::string NodeManager::DebugString() const {
   result << "\n" << task_dependency_manager_.DebugString();
   result << "\n" << lineage_cache_.DebugString();
   result << "\nActorRegistry:";
-  int live_actors = 0;
-  int dead_actors = 0;
-  int reconstructing_actors = 0;
-  int max_num_handles = 0;
-  for (auto &pair : actor_registry_) {
-    if (pair.second.GetState() == ActorState::ALIVE) {
-      live_actors += 1;
-    } else if (pair.second.GetState() == ActorState::RECONSTRUCTING) {
-      reconstructing_actors += 1;
-    } else {
-      dead_actors += 1;
-    }
-    if (pair.second.NumHandles() > max_num_handles) {
-      max_num_handles = pair.second.NumHandles();
-    }
-  }
-  result << "\n- num live actors: " << live_actors;
-  result << "\n- num reconstructing actors: " << live_actors;
-  result << "\n- num dead actors: " << dead_actors;
-  result << "\n- max num handles: " << max_num_handles;
+
+  auto statistical_data = GetActorStatisticalData(actor_registry_);
+  result << "\n- num live actors: " << statistical_data.live_actors;
+  result << "\n- num reconstructing actors: " << statistical_data.reconstructing_actors;
+  result << "\n- num dead actors: " << statistical_data.dead_actors;
+  result << "\n- max num handles: " << statistical_data.max_num_handles;
+
   result << "\nRemoteConnections:";
   for (auto &pair : remote_server_connections_) {
     result << "\n" << pair.first.hex() << ": " << pair.second->DebugString();
@@ -2203,6 +2217,44 @@ std::string NodeManager::DebugString() const {
   return result.str();
 }
 
+void NodeManager::RecordMetrics() const {
+  if (stats::StatsConfig::instance().IsStatsDisabled()) {
+    return;
+  }
+
+  // Record available resources of this node.
+  const auto &available_resources =
+      cluster_resource_map_.at(client_id_).GetAvailableResources().GetResourceMap();
+  for (const auto &pair : available_resources) {
+    stats::LocalAvailableResource().Record(pair.second,
+                                           {{stats::ResourceNameKey, pair.first}});
+  }
+  // Record total resources of this node.
+  const auto &total_resources =
+      cluster_resource_map_.at(client_id_).GetTotalResources().GetResourceMap();
+  for (const auto &pair : total_resources) {
+    stats::LocalTotalResource().Record(pair.second,
+                                       {{stats::ResourceNameKey, pair.first}});
+  }
+
+  object_manager_.RecordMetrics();
+  worker_pool_.RecordMetrics();
+  local_queues_.RecordMetrics();
+  reconstruction_policy_.RecordMetrics();
+  task_dependency_manager_.RecordMetrics();
+  lineage_cache_.RecordMetrics();
+
+  auto statistical_data = GetActorStatisticalData(actor_registry_);
+  stats::ActorStats().Record(statistical_data.live_actors,
+                             {{stats::ValueTypeKey, "live_actors"}});
+  stats::ActorStats().Record(statistical_data.reconstructing_actors,
+                             {{stats::ValueTypeKey, "reconstructing_actors"}});
+  stats::ActorStats().Record(statistical_data.dead_actors,
+                             {{stats::ValueTypeKey, "dead_actors"}});
+  stats::ActorStats().Record(statistical_data.max_num_handles,
+                             {{stats::ValueTypeKey, "max_num_handles"}});
+}
+
 }  // namespace raylet
 
 }  // namespace ray

src/ray/raylet/node_manager.h

@@ -103,6 +103,9 @@ class NodeManager {
   /// \return string.
   std::string DebugString() const;
 
+  /// Record metrics.
+  void RecordMetrics() const;
+
  private:
   /// Methods for handling clients.
 
@@ -121,7 +124,7 @@ class NodeManager {
   void Heartbeat();
 
   /// Write out debug state to a file.
-  void DumpDebugState();
+  void DumpDebugState() const;
 
   /// Get profiling information from the object manager and push it to the GCS.
   ///

src/ray/raylet/reconstruction_policy.cc

@@ -1,5 +1,7 @@
 #include "reconstruction_policy.h"
 
+#include "ray/stats/stats.h"
+
 namespace ray {
 
 namespace raylet {
@@ -210,6 +212,11 @@ std::string ReconstructionPolicy::DebugString() const {
   return result.str();
 }
 
+void ReconstructionPolicy::RecordMetrics() const {
+  stats::ReconstructionPolicyStats().Record(
+      listening_tasks_.size(), {{stats::ValueTypeKey, "num_reconstructing_tasks"}});
+}
+
 }  // namespace raylet
 
 }  // end namespace ray

src/ray/raylet/reconstruction_policy.h

@@ -76,6 +76,9 @@ class ReconstructionPolicy : public ReconstructionPolicyInterface {
   /// \return string.
   std::string DebugString() const;
 
+  /// Record metrics.
+  void RecordMetrics() const;
+
  private:
   struct ReconstructionTask {
     ReconstructionTask(boost::asio::io_service &io_service)

src/ray/raylet/scheduling_queue.cc

@@ -2,13 +2,14 @@
 
 #include <sstream>
 
+#include "ray/stats/stats.h"
 #include "ray/status.h"
 
 namespace {
 
 static constexpr const char *task_state_strings[] = {
     "placeable", "waiting",    "ready",
-    "running",   "infeasible", "waiting for actor creation"};
+    "running",   "infeasible", "waiting_for_actor_creation"};
 static_assert(sizeof(task_state_strings) / sizeof(const char *) ==
                   static_cast<int>(ray::raylet::TaskState::kNumTaskQueues),
               "Must specify a TaskState name for every task queue");
@@ -395,6 +396,18 @@ std::string SchedulingQueue::DebugString() const {
   return result.str();
 }
 
+void SchedulingQueue::RecordMetrics() const {
+  for (const auto &task_state : {
+           TaskState::PLACEABLE, TaskState::WAITING, TaskState::READY, TaskState::RUNNING,
+           TaskState::INFEASIBLE, TaskState::WAITING_FOR_ACTOR_CREATION,
+       }) {
+    stats::SchedulingQueueStats().Record(
+        static_cast<double>(GetTaskQueue(task_state)->GetTasks().size()),
+        {{stats::ValueTypeKey,
+          std::string("num_") + GetTaskStateString(task_state) + "_tasks"}});
+  }
+}
+
 }  // namespace raylet
 
 }  // namespace ray

src/ray/raylet/scheduling_queue.h

@@ -294,6 +294,9 @@ class SchedulingQueue {
   /// \return string.
   std::string DebugString() const;
 
+  /// Record metrics.
+  void RecordMetrics() const;
+
  private:
   /// Get the task queue in the given state. The requested task state must
   /// correspond to one of the task queues (has value <

src/ray/raylet/task_dependency_manager.cc

@@ -1,5 +1,7 @@
 #include "task_dependency_manager.h"
 
+#include "ray/stats/stats.h"
+
 namespace ray {
 
 namespace raylet {
@@ -347,6 +349,19 @@ std::string TaskDependencyManager::DebugString() const {
   return result.str();
 }
 
+void TaskDependencyManager::RecordMetrics() const {
+  stats::TaskDependencyManagerStats().Record(
+      task_dependencies_.size(), {{stats::ValueTypeKey, "num_task_dependencies"}});
+  stats::TaskDependencyManagerStats().Record(
+      required_tasks_.size(), {{stats::ValueTypeKey, "num_required_tasks"}});
+  stats::TaskDependencyManagerStats().Record(
+      required_objects_.size(), {{stats::ValueTypeKey, "num_required_objects"}});
+  stats::TaskDependencyManagerStats().Record(
+      local_objects_.size(), {{stats::ValueTypeKey, "num_local_objects"}});
+  stats::TaskDependencyManagerStats().Record(
+      pending_tasks_.size(), {{stats::ValueTypeKey, "num_pending_tasks"}});
+}
+
 }  // namespace raylet
 
 }  // namespace ray

src/ray/raylet/task_dependency_manager.h

@@ -119,6 +119,9 @@ class TaskDependencyManager {
   /// \return string.
   std::string DebugString() const;
 
+  /// Record metrics.
+  void RecordMetrics() const;
+
  private:
   using ObjectDependencyMap = std::unordered_map<ray::ObjectID, std::vector<ray::TaskID>>;
 

src/ray/raylet/worker_pool.cc

@@ -132,8 +132,6 @@ void WorkerPool::StartWorkerProcess(const Language &language) {
     RAY_LOG(DEBUG) << "Started worker process with pid " << pid;
     state.starting_worker_processes.emplace(
         std::make_pair(pid, num_workers_per_process_));
-    stats::CurrentWorker().Record(pid, {{stats::LanguageKey, EnumNameLanguage(language)},
-                                        {stats::WorkerPidKey, std::to_string(pid)}});
     return;
   }
 }
@@ -247,6 +245,9 @@ bool WorkerPool::DisconnectWorker(const std::shared_ptr<Worker> &worker) {
 void WorkerPool::DisconnectDriver(const std::shared_ptr<Worker> &driver) {
   auto &state = GetStateForLanguage(driver->GetLanguage());
   RAY_CHECK(RemoveWorker(state.registered_drivers, driver));
+  stats::CurrentDriver().Record(
+      0, {{stats::LanguageKey, EnumNameLanguage(driver->GetLanguage())},
+          {stats::WorkerPidKey, std::to_string(driver->Pid())}});
 }
 
 inline WorkerPool::State &WorkerPool::GetStateForLanguage(const Language &language) {
@@ -302,6 +303,24 @@ std::string WorkerPool::DebugString() const {
   return result.str();
 }
 
+void WorkerPool::RecordMetrics() const {
+  for (const auto &entry : states_by_lang_) {
+    // Record worker.
+    for (auto worker : entry.second.registered_workers) {
+      stats::CurrentWorker().Record(
+          worker->Pid(), {{stats::LanguageKey, EnumNameLanguage(worker->GetLanguage())},
+                          {stats::WorkerPidKey, std::to_string(worker->Pid())}});
+    }
+
+    // Record driver.
+    for (auto driver : entry.second.registered_drivers) {
+      stats::CurrentDriver().Record(
+          driver->Pid(), {{stats::LanguageKey, EnumNameLanguage(driver->GetLanguage())},
+                          {stats::WorkerPidKey, std::to_string(driver->Pid())}});
+    }
+  }
+}
+
 }  // namespace raylet
 
 }  // namespace ray

src/ray/raylet/worker_pool.h

@@ -123,6 +123,9 @@ class WorkerPool {
   /// \return string.
   std::string DebugString() const;
 
+  /// Record metrics.
+  void RecordMetrics() const;
+
   /// Generate a warning about the number of workers that have registered or
   /// started if appropriate.
   ///

src/ray/stats/metric_defs.h

@@ -12,16 +12,54 @@
 /// You can follow these examples to define your metrics.
 
 static Gauge CurrentWorker("current_worker",
-                           "This metric is used for report states of workers. "
+                           "This metric is used for reporting states of workers."
                            "Through this, we can see the worker's state on dashboard.",
-                           "1 pcs", {NodeAddressKey, LanguageKey, WorkerPidKey});
+                           "1 pcs", {LanguageKey, WorkerPidKey});
+
+static Gauge CurrentDriver("current_driver",
+                           "This metric is used for reporting states of drivers.",
+                           "1 pcs", {LanguageKey, DriverPidKey});
 
 static Count TaskCountReceived("task_count_received",
-                               "The count that the raylet received.", "pcs",
-                               {NodeAddressKey});
+                               "Number of tasks received by raylet.", "pcs", {});
 
 static Histogram RedisLatency("redis_latency", "The latency of a Redis operation.", "us",
                               {100, 200, 300, 400, 500, 600, 700, 800, 900, 1000},
-                              {NodeAddressKey, CustomKey});
+                              {CustomKey});
+
+static Gauge LocalAvailableResource("local_available_resource",
+                                    "The available resources on this node.", "pcs",
+                                    {ResourceNameKey});
+
+static Gauge LocalTotalResource("local_total_resource",
+                                "The total resources on this node.", "pcs",
+                                {ResourceNameKey});
+
+static Gauge ActorStats("actor_stats", "Stat metrics of the actors in raylet.", "pcs",
+                        {ValueTypeKey});
+
+static Gauge ObjectManagerStats("object_manager_stats",
+                                "Stat the metric values of object in raylet", "pcs",
+                                {ValueTypeKey});
+
+static Gauge LineageCacheStats("lineage_cache_stats",
+                               "Stats the metric values of lineage cache.", "pcs",
+                               {ValueTypeKey});
+
+static Gauge TaskDependencyManagerStats("task_dependency_manager_stats",
+                                        "Stat the metric values of task dependency.",
+                                        "pcs", {ValueTypeKey});
+
+static Gauge SchedulingQueueStats("scheduling_queue_stats",
+                                  "Stats the metric values of scheduling queue.", "pcs",
+                                  {ValueTypeKey});
+
+static Gauge ReconstructionPolicyStats(
+    "reconstruction_policy_stats", "Stats the metric values of reconstruction policy.",
+    "pcs", {ValueTypeKey});
+
+static Gauge ConnectionPoolStats("connection_pool_stats",
+                                 "Stats the connection pool metrics.", "pcs",
+                                 {ValueTypeKey});
 
 #endif  // RAY_STATS_METRIC_DEFS_H

src/ray/stats/stats_test.cc

@@ -44,7 +44,7 @@ class MockExporter : public opencensus::stats::StatsExporter::Handler {
 class StatsTest : public ::testing::Test {
  public:
   void SetUp() {
-    ray::stats::Init("127.0.0.1:8888", {}, false);
+    ray::stats::Init("127.0.0.1:8888", {{stats::NodeAddressKey, "Localhost"}}, false);
     MockExporter::Register();
   }
 
@@ -54,7 +54,7 @@ class StatsTest : public ::testing::Test {
 TEST_F(StatsTest, F) {
   for (size_t i = 0; i < 500; ++i) {
     std::this_thread::sleep_for(std::chrono::milliseconds(50));
-    stats::CurrentWorker().Record(2345, {{stats::NodeAddressKey, "Localhost"}});
+    stats::CurrentWorker().Record(2345);
   }
 }
 

src/ray/stats/tag_defs.h

@@ -19,4 +19,10 @@ static const TagKeyType LanguageKey = TagKeyType::Register("Language");
 
 static const TagKeyType WorkerPidKey = TagKeyType::Register("WorkerPid");
 
+static const TagKeyType DriverPidKey = TagKeyType::Register("DriverPid");
+
+static const TagKeyType ResourceNameKey = TagKeyType::Register("ResourceName");
+
+static const TagKeyType ValueTypeKey = TagKeyType::Register("ValueType");
+
 #endif  // RAY_STATS_TAG_DEFS_H