[observability][export-api] Write actor events (ray-project#47529)

- Add back code changes from [observability][export-api] Write actor events ray-project#47303
- Separate out actor manager export event test into a separate file so we can skip on windows. Update BUILD rule so all tests in src/ray/gcs/gcs_server/test/export_api are skipped on windows

Signed-off-by: Nikita Vemuri <[email protected]>
Co-authored-by: Nikita Vemuri <[email protected]>
Signed-off-by: ujjawal-khare <[email protected]>

BUILD.bazel

@@ -1654,25 +1654,6 @@ ray_cc_test(
     ],
 )
 
-ray_cc_test(
-    name = "gcs_node_manager_export_event_test",
-    size = "small",
-    srcs = [
-        "src/ray/gcs/gcs_server/test/gcs_node_manager_export_event_test.cc",
-    ],
-    tags = [
-        "no_windows",
-        "team:core"
-    ],
-    deps = [
-        ":gcs_server_lib",
-        ":gcs_server_test_util",
-        ":gcs_test_util_lib",
-        ":ray_mock",
-        "@com_google_googletest//:gtest_main",
-    ],
-)
-
 ray_cc_test(
     name = "gcs_job_manager_test",
     size = "small",

src/ray/gcs/gcs_server/gcs_actor_manager.h

@@ -30,6 +30,7 @@
 #include "ray/rpc/gcs_server/gcs_rpc_server.h"
 #include "ray/rpc/worker/core_worker_client.h"
 #include "ray/util/counter_map.h"
+#include "ray/util/event.h"
 #include "src/ray/protobuf/gcs_service.pb.h"
 
 namespace ray {
@@ -188,6 +189,9 @@ class GcsActor {
   /// Get the mutable ActorTableData of this actor.
   rpc::ActorTableData *GetMutableActorTableData();
   rpc::TaskSpec *GetMutableTaskSpec();
+  /// Write an event containing this actor's ActorTableData
+  /// to file for the Export API.
+  void WriteActorExportEvent() const;
 
   const ResourceRequest &GetAcquiredResources() const;
   void SetAcquiredResources(ResourceRequest &&resource_request);
@@ -214,6 +218,27 @@ class GcsActor {
     last_metric_state_ = cur_state;
   }
 
+  rpc::ExportActorData::ActorState ConvertActorStateToExport(
+      rpc::ActorTableData::ActorState actor_state) const {
+    switch (actor_state) {
+    case rpc::ActorTableData::DEPENDENCIES_UNREADY:
+      return rpc::ExportActorData::DEPENDENCIES_UNREADY;
+    case rpc::ActorTableData::PENDING_CREATION:
+      return rpc::ExportActorData::PENDING_CREATION;
+    case rpc::ActorTableData::ALIVE:
+      return rpc::ExportActorData::ALIVE;
+    case rpc::ActorTableData::RESTARTING:
+      return rpc::ExportActorData::RESTARTING;
+    case rpc::ActorTableData::DEAD:
+      return rpc::ExportActorData::DEAD;
+    default:
+      // Unknown rpc::ActorTableData::ActorState value
+      RAY_LOG(FATAL) << "Invalid value for rpc::ActorTableData::ActorState"
+                     << rpc::ActorTableData::ActorState_Name(actor_state);
+      return rpc::ExportActorData::DEAD;
+    }
+  }
+
   /// The actor meta data which contains the task specification as well as the state of
   /// the gcs actor and so on (see gcs.proto).
   rpc::ActorTableData actor_table_data_;