[Core] Put pg state to kv store when pg rescheduling

[larrylian]

Why are these changes needed?

When a PG fails over but has not been scheduled successfully, the restart of gcs will cause the PG to no longer be rescheduled.

1. A node is down, triggering the rescheduling of the PG bundle on this node
2. However, due to insufficient resources, this PG bunlde cannot be scheduled successfully
3. The gcs server sent FO
4. In the end, even if the resources are sufficient, the PG bundle is still not rescheduled.

Reproduce command:

pytest -sv python/ray/tests/test_placement_group_failover.py::test_gcs_restart_when_placement_group_failover

Because the rescheduling state of PG is lost when gcs restarts.
solution:
It is necessary to save the PG to kvstore when the PG is changed to the rescheduling state.

Related issue number

When a PG fails over but has not been scheduled successfully, the restart of gcs will cause the PG to no longer be rescheduled. #34468

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[rkooo567]

This makes sense! Btw can you also add a cpp test to simulate both cases?

1. rescheduling -> node dead -> GCS restarts -> rescheduling succeeds
2. rescheduling started -> GCS restarts -> rescheduling succeeds

[larrylian]

Btw can you also add a cpp test to simulate both cases?

@rkooo567
In fact, my personal opinion is not to add this use case to gcs_placement_group_manager_test.

1. My current python "test_gcs_restart_when_placement_group_failover" use case has been able to fully cover this scenario, and it is also a really effective use case.
2. Many operations in the current gcs_placement_group_manager_test use case are mocked and very complicated. After I add a use case with "gcs restart" logic to it, the later maintenance cost will be even greater.
3. Now many operations in the gcs_placement_group_manager_test use case are mocked. Even if I add the use case you mentioned, its protection effect is not as good as the python use case.

[rkooo567]

I don't think it is complicated to test it? It seems like we already have the way to test GCS restart

TEST_F(GcsPlacementGroupManagerTest, TestSchedulerReinitializeAfterGcsRestart) {
  // Create a placement group and make sure it has been created successfully.
  auto request = Mocker::GenCreatePlacementGroupRequest();
  std::atomic<int> registered_placement_group_count(0);
  RegisterPlacementGroup(request, [&registered_placement_group_count](Status status) {
    ++registered_placement_group_count;
  });
  ASSERT_EQ(registered_placement_group_count, 1);
  WaitForExpectedPgCount(1);

  auto placement_group = mock_placement_group_scheduler_->placement_groups_.back();
  placement_group->GetMutableBundle(0)->set_node_id(NodeID::FromRandom().Binary());
  placement_group->GetMutableBundle(1)->set_node_id(NodeID::FromRandom().Binary());
  mock_placement_group_scheduler_->placement_groups_.pop_back();
  OnPlacementGroupCreationSuccess(placement_group);
  ASSERT_EQ(placement_group->GetState(), rpc::PlacementGroupTableData::CREATED);
  // Reinitialize the placement group manager and test the node dead case.
  auto gcs_init_data = LoadDataFromDataStorage();
  ASSERT_EQ(1, gcs_init_data->PlacementGroups().size());
  EXPECT_TRUE(
      gcs_init_data->PlacementGroups().find(placement_group->GetPlacementGroupID()) !=
      gcs_init_data->PlacementGroups().end());
  EXPECT_CALL(*mock_placement_group_scheduler_, ReleaseUnusedBundles(_)).Times(1);
  EXPECT_CALL(
      *mock_placement_group_scheduler_,
      Initialize(testing::Contains(testing::Key(placement_group->GetPlacementGroupID()))))
      .Times(1);
  gcs_placement_group_manager_->Initialize(*gcs_init_data);
}

The pro of testing it in cpp is we can test the exact edge cases. The python level test depends on the timing, and it is easy to miss edge cases.

[larrylian]

The pro of testing it in cpp is we can test the exact edge cases. The python level test depends on the timing, and it is easy to miss edge cases.

@rkooo567
I agree with this point of view.
Thank you for the example. I'll resubmit a new PR to complement this C++ test case.
This PR has now been run through CI, I think it will be more efficient to merge it first.

[rkooo567]

We will merge this PR first and @larrylian will submit a follow up PR for unit test!

[rkooo567]

Failures are unlikely related