diff --git a/src/ray/gcs/redis_context.cc b/src/ray/gcs/redis_context.cc index 368c5fa13..fe61df288 100644 --- a/src/ray/gcs/redis_context.cc +++ b/src/ray/gcs/redis_context.cc @@ -4,6 +4,9 @@ #include +#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; diff --git a/src/ray/gcs/redis_context.h b/src/ray/gcs/redis_context.h index 1fcfd55ad..0af5a121e 100644 --- a/src/ray/gcs/redis_context.h +++ b/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; +/// operation. +using RedisCallback = std::function; 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 callbacks_; + std::unordered_map callback_items_; }; class RedisContext { diff --git a/src/ray/gcs/tables.cc b/src/ray/gcs/tables.cc index 93e40a392..e0876aa73 100644 --- a/src/ray/gcs/tables.cc +++ b/src/ray/gcs/tables.cc @@ -48,7 +48,6 @@ Status Log::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::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::Lookup(const DriverID &driver_id, const ID &id, } lookup(client_, id, results); } - return true; }; std::vector nil; return GetRedisContext(id)->RunAsync("RAY.TABLE_LOOKUP", id, nil.data(), nil.size(), @@ -154,10 +151,8 @@ Status Log::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::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::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::Remove(const DriverID &driver_id, const ID &id, if (done != nullptr) { (done)(client_, id, *dataT); } - return true; }; flatbuffers::FlatBufferBuilder fbb; fbb.ForceDefaults(true); diff --git a/src/ray/object_manager/connection_pool.cc b/src/ray/object_manager/connection_pool.cc index d1261d7da..6b029cbe8 100644 --- a/src/ray/object_manager/connection_pool.cc +++ b/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 diff --git a/src/ray/object_manager/connection_pool.h b/src/ray/object_manager/connection_pool.h index 69fbd5261..628769d38 100644 --- a/src/ray/object_manager/connection_pool.h +++ b/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); diff --git a/src/ray/object_manager/object_manager.cc b/src/ray/object_manager/object_manager.cc index 71ff1ead1..3d829027b 100644 --- a/src/ray/object_manager/object_manager.cc +++ b/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 diff --git a/src/ray/object_manager/object_manager.h b/src/ray/object_manager/object_manager.h index b905f6c2c..0d3010523 100644 --- a/src/ray/object_manager/object_manager.h +++ b/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; diff --git a/src/ray/raylet/lineage_cache.cc b/src/ray/raylet/lineage_cache.cc index 57761cd06..94f1dc11f 100644 --- a/src/ray/raylet/lineage_cache.cc +++ b/src/ray/raylet/lineage_cache.cc @@ -1,4 +1,5 @@ #include "lineage_cache.h" +#include "ray/stats/stats.h" #include @@ -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 diff --git a/src/ray/raylet/lineage_cache.h b/src/ray/raylet/lineage_cache.h index 1816d97f7..045ded107 100644 --- a/src/ray/raylet/lineage_cache.h +++ b/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. diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 675278989..f901c6800 100644 --- a/src/ray/raylet/node_manager.cc +++ b/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 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(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 diff --git a/src/ray/raylet/node_manager.h b/src/ray/raylet/node_manager.h index ad2091c14..edd456dbe 100644 --- a/src/ray/raylet/node_manager.h +++ b/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. /// diff --git a/src/ray/raylet/reconstruction_policy.cc b/src/ray/raylet/reconstruction_policy.cc index ee328e027..4274ff5a2 100644 --- a/src/ray/raylet/reconstruction_policy.cc +++ b/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 diff --git a/src/ray/raylet/reconstruction_policy.h b/src/ray/raylet/reconstruction_policy.h index f18290aa3..fc441de8a 100644 --- a/src/ray/raylet/reconstruction_policy.h +++ b/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) diff --git a/src/ray/raylet/scheduling_queue.cc b/src/ray/raylet/scheduling_queue.cc index dc6316868..986ede199 100644 --- a/src/ray/raylet/scheduling_queue.cc +++ b/src/ray/raylet/scheduling_queue.cc @@ -2,13 +2,14 @@ #include +#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(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(GetTaskQueue(task_state)->GetTasks().size()), + {{stats::ValueTypeKey, + std::string("num_") + GetTaskStateString(task_state) + "_tasks"}}); + } +} + } // namespace raylet } // namespace ray diff --git a/src/ray/raylet/scheduling_queue.h b/src/ray/raylet/scheduling_queue.h index 89d623306..3f7bab123 100644 --- a/src/ray/raylet/scheduling_queue.h +++ b/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 < diff --git a/src/ray/raylet/task_dependency_manager.cc b/src/ray/raylet/task_dependency_manager.cc index fe4364c44..4fbebb8df 100644 --- a/src/ray/raylet/task_dependency_manager.cc +++ b/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 diff --git a/src/ray/raylet/task_dependency_manager.h b/src/ray/raylet/task_dependency_manager.h index afb146af6..9691b3d64 100644 --- a/src/ray/raylet/task_dependency_manager.h +++ b/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>; diff --git a/src/ray/raylet/worker_pool.cc b/src/ray/raylet/worker_pool.cc index 44f8e39f7..3138b88cf 100644 --- a/src/ray/raylet/worker_pool.cc +++ b/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) { void WorkerPool::DisconnectDriver(const std::shared_ptr &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 diff --git a/src/ray/raylet/worker_pool.h b/src/ray/raylet/worker_pool.h index ccad659f5..03443447c 100644 --- a/src/ray/raylet/worker_pool.h +++ b/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. /// diff --git a/src/ray/stats/metric_defs.h b/src/ray/stats/metric_defs.h index bd4aeb00c..961dacb2c 100644 --- a/src/ray/stats/metric_defs.h +++ b/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 diff --git a/src/ray/stats/stats_test.cc b/src/ray/stats/stats_test.cc index 706478a73..da8b7ffc3 100644 --- a/src/ray/stats/stats_test.cc +++ b/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); } } diff --git a/src/ray/stats/tag_defs.h b/src/ray/stats/tag_defs.h index 60bfcafae..9fb9c9d31 100644 --- a/src/ray/stats/tag_defs.h +++ b/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