Skip to content

Commit

Permalink
Revert "[observability][export-api] Write actor events" (ray-project#…
Browse files Browse the repository at this point in the history
…47516)

Reverts ray-project#47303

Signed-off-by: ujjawal-khare <[email protected]>
  • Loading branch information
can-anyscale authored and ujjawal-khare committed Oct 15, 2024
1 parent a6f382c commit eb79da0
Show file tree
Hide file tree
Showing 2 changed files with 1 addition and 67 deletions.
25 changes: 0 additions & 25 deletions src/ray/gcs/gcs_server/gcs_actor_manager.h
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@
#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 {
Expand Down Expand Up @@ -189,9 +188,6 @@ 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);
Expand All @@ -218,27 +214,6 @@ 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_;
Expand Down
43 changes: 1 addition & 42 deletions src/ray/gcs/gcs_server/test/gcs_actor_manager_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.

#include <chrono>
#include <memory>
#include <thread>

// clang-format off
#include "gtest/gtest.h"
Expand All @@ -26,7 +24,6 @@
#include "mock/ray/gcs/gcs_server/gcs_kv_manager.h"
#include "mock/ray/gcs/gcs_server/gcs_node_manager.h"
#include "mock/ray/pubsub/publisher.h"
#include "ray/util/event.h"
// clang-format on

namespace ray {
Expand Down Expand Up @@ -121,8 +118,7 @@ class GcsActorManagerTest : public ::testing::Test {
RayConfig::instance().initialize(
R"(
{
"maximum_gcs_destroyed_actor_cached_count": 10,
"enable_export_api_write": true
"maximum_gcs_destroyed_actor_cached_count": 10
}
)");
std::promise<bool> promise;
Expand Down Expand Up @@ -164,13 +160,11 @@ class GcsActorManagerTest : public ::testing::Test {
auto job_id = JobID::FromInt(i);
job_namespace_table_[job_id] = "";
}
log_dir_ = "event_123";
}

virtual ~GcsActorManagerTest() {
io_service_.stop();
thread_io_service_->join();
std::filesystem::remove_all(log_dir_.c_str());
}

void WaitActorCreated(const ActorID &actor_id) {
Expand Down Expand Up @@ -292,13 +286,9 @@ class GcsActorManagerTest : public ::testing::Test {
std::unique_ptr<gcs::GcsFunctionManager> function_manager_;
std::unique_ptr<gcs::MockInternalKVInterface> kv_;
PeriodicalRunner periodical_runner_;
std::string log_dir_;
};

TEST_F(GcsActorManagerTest, TestBasic) {
std::vector<SourceTypeVariant> source_types = {
rpc::ExportEvent_SourceType::ExportEvent_SourceType_EXPORT_ACTOR};
RayEventInit(source_types, absl::flat_hash_map<std::string, std::string>(), log_dir_);
auto job_id = JobID::FromInt(1);
auto registered_actor = RegisterActor(job_id);
rpc::CreateActorRequest create_actor_request;
Expand Down Expand Up @@ -333,37 +323,6 @@ TEST_F(GcsActorManagerTest, TestBasic) {
ASSERT_EQ(actor->GetState(), rpc::ActorTableData::DEAD);
RAY_CHECK_EQ(gcs_actor_manager_->CountFor(rpc::ActorTableData::ALIVE, ""), 0);
RAY_CHECK_EQ(gcs_actor_manager_->CountFor(rpc::ActorTableData::DEAD, ""), 1);

// Check correct export events are written for each of the 4 state transitions
int num_retry = 5;
int num_export_events = 4;
std::vector<std::string> expected_states = {
"DEPENDENCIES_UNREADY", "PENDING_CREATION", "ALIVE", "DEAD"};
std::vector<std::string> vc;
for (int i = 0; i < num_retry; i++) {
Mocker::ReadContentFromFile(vc, log_dir_ + "/events/event_EXPORT_ACTOR.log");
if ((int)vc.size() == num_export_events) {
for (int event_idx = 0; event_idx < num_export_events; event_idx++) {
json export_event_as_json = json::parse(vc[event_idx]);
json event_data = export_event_as_json["event_data"].get<json>();
ASSERT_EQ(event_data["state"], expected_states[event_idx]);
if (event_idx == num_export_events - 1) {
// Verify death cause for last actor DEAD event
ASSERT_EQ(
event_data["death_cause"]["actor_died_error_context"]["error_message"],
"The actor is dead because all references to the actor were removed.");
}
}
return;
} else {
// Sleep and retry
std::this_thread::sleep_for(std::chrono::seconds(1));
vc.clear();
}
}
Mocker::ReadContentFromFile(vc, log_dir_ + "/events/event_EXPORT_ACTOR.log");
ASSERT_TRUE(false) << "Export API only wrote " << (int)vc.size()
<< " lines, but expecting 4.\n";
}

TEST_F(GcsActorManagerTest, TestDeadCount) {
Expand Down

0 comments on commit eb79da0

Please sign in to comment.