[GCS] Add ErrorInfoAccessor to GCS Client (#6749)

src/ray/gcs/accessor.h

@@ -473,6 +473,32 @@ class NodeInfoAccessor {
   NodeInfoAccessor() = default;
 };
 
+/// \class ErrorInfoAccessor
+/// `ErrorInfoAccessor` is a sub-interface of `GcsClient`.
+/// This class includes all the methods that are related to accessing
+/// error information in the GCS.
+class ErrorInfoAccessor {
+ public:
+  virtual ~ErrorInfoAccessor() = default;
+
+  /// Report a job error to GCS asynchronously.
+  /// The error message will be pushed to the driver of a specific if it is
+  /// a job internal error, or broadcast to all drivers if it is a system error.
+  ///
+  /// TODO(rkn): We need to make sure that the errors are unique because
+  /// duplicate messages currently cause failures (the GCS doesn't allow it). A
+  /// natural way to do this is to have finer-grained time stamps.
+  ///
+  /// \param data_ptr The error message that will be reported to GCS.
+  /// \param callback Callback that will be called when report is complete.
+  /// \return Status
+  virtual Status AsyncReportJobError(const std::shared_ptr<rpc::ErrorTableData> &data_ptr,
+                                     const StatusCallback &callback) = 0;
+
+ protected:
+  ErrorInfoAccessor() = default;
+};
+
 /// \class StatsInfoAccessor
 /// `StatsInfoAccessor` is a sub-interface of `GcsClient`.
 /// This class includes all the methods that are related to accessing

src/ray/gcs/gcs_client.h

@@ -95,6 +95,13 @@ class GcsClient : public std::enable_shared_from_this<GcsClient> {
     return *task_accessor_;
   }
 
+  /// Get the sub-interface for accessing error information in GCS.
+  /// This function is thread safe.
+  ErrorInfoAccessor &Errors() {
+    RAY_CHECK(error_accessor_ != nullptr);
+    return *error_accessor_;
+  }
+
   /// Get the sub-interface for accessing stats in GCS.
   /// This function is thread safe.
   StatsInfoAccessor &Stats() {
@@ -118,6 +125,7 @@ class GcsClient : public std::enable_shared_from_this<GcsClient> {
   std::unique_ptr<ObjectInfoAccessor> object_accessor_;
   std::unique_ptr<NodeInfoAccessor> node_accessor_;
   std::unique_ptr<TaskInfoAccessor> task_accessor_;
+  std::unique_ptr<ErrorInfoAccessor> error_accessor_;
   std::unique_ptr<StatsInfoAccessor> stats_accessor_;
 };
 

src/ray/gcs/pb_util.h

@@ -30,6 +30,18 @@ inline std::shared_ptr<ray::rpc::JobTableData> CreateJobTableData(
   return job_info_ptr;
 }
 
+/// Helper function to produce error table data.
+inline std::shared_ptr<ray::rpc::ErrorTableData> CreateErrorTableData(
+    const std::string &error_type, const std::string &error_msg, double timestamp,
+    const JobID &job_id = JobID::Nil()) {
+  auto error_info_ptr = std::make_shared<ray::rpc::ErrorTableData>();
+  error_info_ptr->set_type(error_type);
+  error_info_ptr->set_error_message(error_msg);
+  error_info_ptr->set_timestamp(timestamp);
+  error_info_ptr->set_job_id(job_id.Binary());
+  return error_info_ptr;
+}
+
 /// Helper function to produce actor table data.
 inline std::shared_ptr<ray::rpc::ActorTableData> CreateActorTableData(
     const TaskSpecification &task_spec, const ray::rpc::Address &address,

src/ray/gcs/redis_accessor.cc

@@ -589,6 +589,22 @@ Status RedisNodeInfoAccessor::AsyncSubscribeToResources(
   return resource_sub_executor_.AsyncSubscribeAll(ClientID::Nil(), subscribe, done);
 }
 
+RedisErrorInfoAccessor::RedisErrorInfoAccessor(RedisGcsClient *client_impl)
+    : client_impl_(client_impl) {}
+
+Status RedisErrorInfoAccessor::AsyncReportJobError(
+    const std::shared_ptr<ErrorTableData> &data_ptr, const StatusCallback &callback) {
+  ErrorTable::WriteCallback on_done = nullptr;
+  if (callback != nullptr) {
+    on_done = [callback](RedisGcsClient *client, const JobID &job_id,
+                         const ErrorTableData &data) { callback(Status::OK()); };
+  }
+
+  JobID job_id = JobID::FromBinary(data_ptr->job_id());
+  ErrorTable &error_table = client_impl_->error_table();
+  return error_table.Append(job_id, job_id, data_ptr, on_done);
+}
+
 RedisStatsInfoAccessor::RedisStatsInfoAccessor(RedisGcsClient *client_impl)
     : client_impl_(client_impl) {}
 

src/ray/gcs/redis_accessor.h

@@ -290,6 +290,22 @@ class RedisNodeInfoAccessor : public NodeInfoAccessor {
   HeartbeatBatchSubscriptionExecutor heartbeat_batch_sub_executor_;
 };
 
+/// \class RedisErrorInfoAccessor
+/// RedisErrorInfoAccessor is an implementation of `ErrorInfoAccessor`
+/// that uses Redis as the backend storage.
+class RedisErrorInfoAccessor : public ErrorInfoAccessor {
+ public:
+  explicit RedisErrorInfoAccessor(RedisGcsClient *client_impl);
+
+  virtual ~RedisErrorInfoAccessor() = default;
+
+  Status AsyncReportJobError(const std::shared_ptr<ErrorTableData> &data_ptr,
+                             const StatusCallback &callback) override;
+
+ private:
+  RedisGcsClient *client_impl_{nullptr};
+};
+
 /// \class RedisStatsInfoAccessor
 /// RedisStatsInfoAccessor is an implementation of `StatsInfoAccessor`
 /// that uses Redis as the backend storage.

src/ray/gcs/redis_gcs_client.cc

@@ -147,6 +147,7 @@ Status RedisGcsClient::Connect(boost::asio::io_service &io_service) {
   object_accessor_.reset(new RedisObjectInfoAccessor(this));
   node_accessor_.reset(new RedisNodeInfoAccessor(this));
   task_accessor_.reset(new RedisTaskInfoAccessor(this));
+  error_accessor_.reset(new RedisErrorInfoAccessor(this));
   stats_accessor_.reset(new RedisStatsInfoAccessor(this));
 
   is_connected_ = true;

src/ray/gcs/redis_gcs_client.h

@@ -25,6 +25,7 @@ class RAY_EXPORT RedisGcsClient : public GcsClient {
   friend class RedisTaskInfoAccessor;
   friend class RedisNodeInfoAccessor;
   friend class RedisObjectInfoAccessor;
+  friend class RedisErrorInfoAccessor;
   friend class RedisStatsInfoAccessor;
   friend class SubscriptionExecutorTest;
   friend class LogSubscribeTestHelper;
@@ -64,8 +65,6 @@ class RAY_EXPORT RedisGcsClient : public GcsClient {
   /// Disconnect with GCS Service. Non-thread safe.
   void Disconnect();
 
-  // TODO: Some API for getting the error on the driver
-  ErrorTable &error_table();
   WorkerFailureTable &worker_failure_table();
 
   // We also need something to export generic code to run on workers from the
@@ -106,6 +105,9 @@ class RAY_EXPORT RedisGcsClient : public GcsClient {
   raylet::TaskTable &raylet_task_table();
   TaskLeaseTable &task_lease_table();
   TaskReconstructionLog &task_reconstruction_log();
+  /// This method will be deprecated, use method Errors() instead.
+  // TODO: Some API for getting the error on the driver
+  ErrorTable &error_table();
   /// This method will be deprecated, use method Stats() instead.
   ProfileTable &profile_table();
 

src/ray/gcs/tables.cc

@@ -515,16 +515,6 @@ Status Hash<ID, Data>::Subscribe(const JobID &job_id, const ClientID &client_id,
   return Status::OK();
 }
 
-Status ErrorTable::PushErrorToDriver(const JobID &job_id, const std::string &type,
-                                     const std::string &error_message, double timestamp) {
-  auto data = std::make_shared<ErrorTableData>();
-  data->set_job_id(job_id.Binary());
-  data->set_type(type);
-  data->set_error_message(error_message);
-  data->set_timestamp(timestamp);
-  return Append(job_id, job_id, data, /*done_callback=*/nullptr);
-}
-
 std::string ErrorTable::DebugString() const {
   return Log<JobID, ErrorTableData>::DebugString();
 }

src/ray/gcs/tables.h

@@ -823,7 +823,7 @@ class TaskTable : public Table<TaskID, TaskTableData> {
 
 }  // namespace raylet
 
-class ErrorTable : private Log<JobID, ErrorTableData> {
+class ErrorTable : public Log<JobID, ErrorTableData> {
  public:
   ErrorTable(const std::vector<std::shared_ptr<RedisContext>> &contexts,
              RedisGcsClient *client)
@@ -832,22 +832,6 @@ class ErrorTable : private Log<JobID, ErrorTableData> {
     prefix_ = TablePrefix::ERROR_INFO;
   };
 
-  /// Push an error message for the driver of a specific.
-  ///
-  /// TODO(rkn): We need to make sure that the errors are unique because
-  /// duplicate messages currently cause failures (the GCS doesn't allow it). A
-  /// natural way to do this is to have finer-grained time stamps.
-  ///
-  /// \param job_id The ID of the job that generated the error. If the error
-  /// should be pushed to all drivers, then this should be nil.
-  /// \param type The type of the error.
-  /// \param error_message The error message to push.
-  /// \param timestamp The timestamp of the error.
-  /// \return Status.
-  // TODO(qwang): refactor this API to implement broadcast.
-  Status PushErrorToDriver(const JobID &job_id, const std::string &type,
-                           const std::string &error_message, double timestamp);
-
   /// Returns debug string for class.
   ///
   /// \return string.

src/ray/raylet/monitor.cc

@@ -2,6 +2,7 @@
 
 #include "ray/common/ray_config.h"
 #include "ray/common/status.h"
+#include "ray/gcs/pb_util.h"
 #include "ray/util/util.h"
 
 namespace ray {
@@ -67,9 +68,10 @@ void Monitor::Tick() {
             error_message << "The node with client ID " << node_id
                           << " has been marked dead because the monitor"
                           << " has missed too many heartbeats from it.";
-            // We use the nil JobID to broadcast the message to all drivers.
-            RAY_CHECK_OK(gcs_client_.error_table().PushErrorToDriver(
-                JobID::Nil(), type, error_message.str(), current_time_ms()));
+            auto error_data_ptr =
+                gcs::CreateErrorTableData(type, error_message.str(), current_time_ms());
+            RAY_CHECK_OK(
+                gcs_client_.Errors().AsyncReportJobError(error_data_ptr, nullptr));
           }
         };
         RAY_CHECK_OK(gcs_client_.Nodes().AsyncGetAll(lookup_callback));

src/ray/raylet/node_manager.cc

@@ -408,9 +408,10 @@ void NodeManager::WarnResourceDeadlock() {
         << "To resolve the issue, consider creating fewer actors or increase the "
         << "resources available to this Ray cluster. You can ignore this message "
         << "if this Ray cluster is expected to auto-scale.";
-    RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver(
-        exemplar.GetTaskSpecification().JobId(), "resource_deadlock", error_message.str(),
-        current_time_ms()));
+    auto error_data_ptr = gcs::CreateErrorTableData(
+        "resource_deadlock", error_message.str(), current_time_ms(),
+        exemplar.GetTaskSpecification().JobId());
+    RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr));
     resource_deadlock_warned_ = true;
   }
 }
@@ -1177,8 +1178,9 @@ void NodeManager::ProcessDisconnectClientMessage(
         std::ostringstream error_message;
         error_message << "A worker died or was killed while executing task " << task_id
                       << ".";
-        RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver(
-            job_id, type, error_message.str(), current_time_ms()));
+        auto error_data_ptr = gcs::CreateErrorTableData(type, error_message.str(),
+                                                        current_time_ms(), job_id);
+        RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr));
       }
     }
 
@@ -1342,13 +1344,12 @@ void NodeManager::ProcessWaitForDirectActorCallArgsRequestMessage(
 void NodeManager::ProcessPushErrorRequestMessage(const uint8_t *message_data) {
   auto message = flatbuffers::GetRoot<protocol::PushErrorRequest>(message_data);
 
-  JobID job_id = from_flatbuf<JobID>(*message->job_id());
   auto const &type = string_from_flatbuf(*message->type());
   auto const &error_message = string_from_flatbuf(*message->error_message());
   double timestamp = message->timestamp();
-
-  RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver(job_id, type, error_message,
-                                                            timestamp));
+  JobID job_id = from_flatbuf<JobID>(*message->job_id());
+  auto error_data_ptr = gcs::CreateErrorTableData(type, error_message, timestamp, job_id);
+  RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr));
 }
 
 void NodeManager::ProcessPrepareActorCheckpointRequest(
@@ -1807,9 +1808,10 @@ void NodeManager::ScheduleTasks(
           << "provide the requested resources. To resolve this issue, consider "
           << "reducing the resource requests of this task or add nodes that "
           << "can fit the task.";
-      RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver(
-          task.GetTaskSpecification().JobId(), type, error_message.str(),
-          current_time_ms()));
+      auto error_data_ptr =
+          gcs::CreateErrorTableData(type, error_message.str(), current_time_ms(),
+                                    task.GetTaskSpecification().JobId());
+      RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr));
     }
     // Assert that this placeable task is not feasible locally (necessary but not
     // sufficient).
@@ -1885,8 +1887,9 @@ void NodeManager::MarkObjectsAsFailed(const ErrorType &error_type,
              << " object may hang forever.";
       std::string error_message = stream.str();
       RAY_LOG(WARNING) << error_message;
-      RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver(
-          job_id, "task", error_message, current_time_ms()));
+      auto error_data_ptr =
+          gcs::CreateErrorTableData("task", error_message, current_time_ms(), job_id);
+      RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr));
     }
   }
 }
@@ -2662,9 +2665,10 @@ void NodeManager::ResubmitTask(const Task &task, const ObjectID &required_object
     error_message << "The task with ID " << task.GetTaskSpecification().TaskId()
                   << " is a driver task and so the object created by ray.put "
                   << "could not be reconstructed.";
-    RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver(
-        task.GetTaskSpecification().JobId(), type, error_message.str(),
-        current_time_ms()));
+    auto error_data_ptr =
+        gcs::CreateErrorTableData(type, error_message.str(), current_time_ms(),
+                                  task.GetTaskSpecification().JobId());
+    RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr));
     MarkObjectsAsFailed(ErrorType::OBJECT_UNRECONSTRUCTABLE,
                         {required_object_id.ToPlasmaId()},
                         task.GetTaskSpecification().JobId());

src/ray/raylet/worker_pool.cc

@@ -12,6 +12,7 @@
 #include "ray/common/constants.h"
 #include "ray/common/ray_config.h"
 #include "ray/common/status.h"
+#include "ray/gcs/pb_util.h"
 #include "ray/stats/stats.h"
 #include "ray/util/logging.h"
 #include "ray/util/util.h"
@@ -487,8 +488,9 @@ void WorkerPool::WarnAboutSize() {
                       << "using nested tasks "
                       << "(see https://github.com/ray-project/ray/issues/3644) for "
                       << "some a discussion of workarounds.";
-      RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver(
-          JobID::Nil(), "worker_pool_large", warning_message.str(), current_time_ms()));
+      auto error_data_ptr = gcs::CreateErrorTableData(
+          "worker_pool_large", warning_message.str(), current_time_ms());
+      RAY_CHECK_OK(gcs_client_->Errors().AsyncReportJobError(error_data_ptr, nullptr));
     }
   }
 }