support to clean worker table with maximum_gcs_dead_worker_cached_count

src/ray/common/ray_config_def.h

@@ -481,6 +481,8 @@ RAY_CONFIG(int64_t, task_events_report_interval_ms, 1000)
 /// Setting the value to -1 allows for unlimited task events stored in GCS.
 RAY_CONFIG(int64_t, task_events_max_num_task_in_gcs, 100000)
 
+RAY_CONFIG(int64_t, maximum_gcs_dead_worker_cached_count, 1000)
+
 /// The number of task attempts being dropped per job tracked at GCS. When GCS is forced
 /// to stop tracking some task attempts that are lost, this will incur potential partial
 /// data loss for a single task attempt (e.g. some task events were dropped, but some were

src/ray/gcs/gcs_server/gcs_init_data.cc

@@ -27,6 +27,8 @@ void GcsInitData::AsyncLoad(const EmptyCallback &on_done) {
     }
   };
 
+  AsyncLoadWorkerTableData(on_load_finished);
+
   AsyncLoadJobTableData(on_load_finished);
 
   AsyncLoadNodeTableData(on_load_finished);
@@ -38,6 +40,18 @@ void GcsInitData::AsyncLoad(const EmptyCallback &on_done) {
   AsyncLoadPlacementGroupTableData(on_load_finished);
 }
 
+void GcsInitData::AsyncLoadWorkerTableData(const EmptyCallback &on_done) {
+  RAY_LOG(INFO) << "Loading worker table data.";
+  auto load_worker_table_data_callback =
+      [this, on_done](absl::flat_hash_map<WorkerID, rpc::WorkerTableData> &&result) {
+        worker_table_data_ = std::move(result);
+        RAY_LOG(INFO) << "Finished loading worker table data, size = "
+                      << worker_table_data_.size();
+        on_done();
+  };
+  RAY_CHECK_OK(gcs_table_storage_->WorkerTable().GetAll(load_worker_table_data_callback));
+}
+
 void GcsInitData::AsyncLoadJobTableData(const EmptyCallback &on_done) {
   RAY_LOG(INFO) << "Loading job table data.";
   auto load_job_table_data_callback =

src/ray/gcs/gcs_server/gcs_init_data.h

@@ -39,6 +39,11 @@ class GcsInitData {
   /// \param on_done The callback when all metadatas are loaded successfully.
   void AsyncLoad(const EmptyCallback &on_done);
 
+  /// Get worker metadata.
+  const absl::flat_hash_map<WorkerID, rpc::WorkerTableData> &Workers() const {
+    return worker_table_data_;
+  }
+
   /// Get job metadata.
   const absl::flat_hash_map<JobID, rpc::JobTableData> &Jobs() const {
     return job_table_data_;
@@ -65,6 +70,8 @@ class GcsInitData {
   }
 
  private:
+  void AsyncLoadWorkerTableData(const EmptyCallback &on_done);
+
   /// Load job metadata from the store into memory asynchronously.
   ///
   /// \param on_done The callback when job metadata is loaded successfully.
@@ -91,6 +98,9 @@ class GcsInitData {
   /// The gcs table storage.
   std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage_;
 
+  /// Worker metadata.
+  absl::flat_hash_map<WorkerID, rpc::WorkerTableData> worker_table_data_;
+
   /// Job metadata.
   absl::flat_hash_map<JobID, rpc::JobTableData> job_table_data_;
 

src/ray/gcs/gcs_server/gcs_server.cc

@@ -221,7 +221,7 @@ void GcsServer::DoStart(const GcsInitData &gcs_init_data) {
   InitGcsActorManager(gcs_init_data);
 
   // Init gcs worker manager.
-  InitGcsWorkerManager();
+  InitGcsWorkerManager(gcs_init_data);
 
   // Init GCS task manager.
   InitGcsTaskManager();
@@ -630,9 +630,13 @@ void GcsServer::InitRuntimeEnvManager() {
   rpc_server_.RegisterService(*runtime_env_service_);
 }
 
-void GcsServer::InitGcsWorkerManager() {
+
+void GcsServer::InitGcsWorkerManager(const GcsInitData &gcs_init_data) {
   gcs_worker_manager_ =
-      std::make_unique<GcsWorkerManager>(gcs_table_storage_, gcs_publisher_);
+      std::make_unique<GcsWorkerManager>(RayConfig::instance().maximum_gcs_dead_worker_cached_count(), gcs_table_storage_, gcs_publisher_);
+
+  // Initialize by gcs tables data.
+  gcs_worker_manager_->Initialize(gcs_init_data);
   // Register service.
   worker_info_service_.reset(new rpc::WorkerInfoGrpcService(
       io_context_provider_.GetDefaultIOContext(), *gcs_worker_manager_));

src/ray/gcs/gcs_server/gcs_server.h

@@ -155,7 +155,7 @@ class GcsServer {
   void InitGcsPlacementGroupManager(const GcsInitData &gcs_init_data);
 
   /// Initialize gcs worker manager.
-  void InitGcsWorkerManager();
+  void InitGcsWorkerManager(const GcsInitData &gcs_init_data);
 
   /// Initialize gcs task manager.
   void InitGcsTaskManager();

src/ray/gcs/gcs_server/gcs_worker_manager.cc

@@ -129,6 +129,20 @@ void GcsWorkerManager::HandleReportWorkerFailure(
             usage_stats_client_->RecordExtraUsageCounter(key, count);
           }
         }
+
+        // worker IDs and clean up expired workers
+        sorted_dead_worker_list_.emplace_back(worker_id, worker_failure_data->timestamp());
+
+        // If limit enforced, replace one.
+        if (max_num_dead_workers_ > 0 && sorted_dead_worker_list_.size() > max_num_dead_workers_) {
+          RAY_LOG_EVERY_MS(WARNING, 10000)
+              << "Max number of dead workers event (" << max_num_worker_events_
+              << ") allowed is reached. Old worker events will be overwritten. Set "
+                 "`RAY_maximum_gcs_dead_worker_cached_count` to a higher value to store more.";
+          const auto &clean_worker_id = sorted_dead_worker_list_.front().first;
+          RAY_CHECK_OK(gcs_table_storage_->WorkerTable().Delete(clean_worker_id, nullptr));
+          sorted_dead_worker_list_.pop_front();
+        }
       });
 }
 
@@ -327,6 +341,21 @@ void GcsWorkerManager::HandleUpdateWorkerNumPausedThreads(
   }
 }
 
+void GcsWorkerManager::Initialize(const GcsInitData &gcs_init_data) {
+  RAY_LOG(INFO) << "Initializing gcs_worker_manager for load dead workers table.";
+  for (const auto &[worker_id, worker_table_data] : gcs_init_data.Workers()) {
+    if (!worker_table_data.is_alive()) {
+      sorted_dead_worker_list_.emplace_back(worker_id, worker_table_data.timestamp());
+    }
+  }
+  // Sort by expiration time, prioritize clearing the history first
+  sorted_dead_worker_list_.sort([](const std::pair<WorkerID, int64_t> &left,
+                                       const std::pair<WorkerID, int64_t> &right) {
+    return left.second < right.second;
+  });
+  RAY_LOG(INFO) << "Finished initialize gcs_worker_manager for dead workers size : " << sorted_dead_worker_list_.size();
+}
+
 void GcsWorkerManager::AddWorkerDeadListener(
     std::function<void(std::shared_ptr<WorkerTableData>)> listener) {
   RAY_CHECK(listener != nullptr);

src/ray/gcs/gcs_server/gcs_worker_manager.h

@@ -19,16 +19,25 @@
 #include "ray/gcs/gcs_server/usage_stats_client.h"
 #include "ray/gcs/pubsub/gcs_pub_sub.h"
 #include "ray/rpc/gcs_server/gcs_rpc_server.h"
+#include "ray/util/counter_map.h"
+#include "ray/gcs/gcs_server/gcs_init_data.h"
 
 namespace ray {
 namespace gcs {
 
+enum GcsWorkerManagerCounter {
+  kNumWorkerDeadEventsStored,
+};
+
 /// This implementation class of `WorkerInfoHandler`.
 class GcsWorkerManager : public rpc::WorkerInfoHandler {
  public:
-  explicit GcsWorkerManager(std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage,
+  explicit GcsWorkerManager(size_t max_num_worker_events,
+                            std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage,
                             std::shared_ptr<GcsPublisher> &gcs_publisher)
-      : gcs_table_storage_(gcs_table_storage), gcs_publisher_(gcs_publisher) {}
+      : gcs_table_storage_(gcs_table_storage),
+        gcs_publisher_(gcs_publisher),
+        max_num_dead_workers_(max_num_dead_workers) {}
 
   void HandleReportWorkerFailure(rpc::ReportWorkerFailureRequest request,
                                  rpc::ReportWorkerFailureReply *reply,
@@ -56,6 +65,8 @@ class GcsWorkerManager : public rpc::WorkerInfoHandler {
       rpc::UpdateWorkerNumPausedThreadsReply *reply,
       rpc::SendReplyCallback send_reply_callback) override;
 
+  void Initialize(const GcsInitData &gcs_init_data);
+
   void AddWorkerDeadListener(
       std::function<void(std::shared_ptr<WorkerTableData>)> listener);
 
@@ -74,6 +85,13 @@ class GcsWorkerManager : public rpc::WorkerInfoHandler {
   std::vector<std::function<void(std::shared_ptr<WorkerTableData>)>>
       worker_dead_listeners_;
 
+  /// The workers are sorted according to the timestamp, and the oldest is at the head of
+  /// the list.
+  std::list<std::pair<WorkerID, int64_t>> sorted_dead_worker_list_;
+
+  /// Max number of dead workers allowed in the storage.
+  const size_t max_num_dead_workers_ = 1000;
+
   /// Tracks the number of occurences of worker crash due to system error
   int32_t worker_crash_system_error_count_ = 0;