[core] Add cluster ID to the Python layer #37583

vitsai · 2023-07-19T21:19:56Z

Why are these changes needed?

An earlier change added a new RPC to each GCS client, causing some stress tests to fail. Instead of adding O(num_worker) RPCs, add O(num_node) RPCs by plumbing through the Python layer.

Related issue number

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 :(

src/ray/raylet/main.cc

fishbone · 2023-07-19T22:27:58Z

src/ray/gcs/gcs_client/gcs_client.h

+  inline grpc::ClientContext &&PrepareContext(int64_t timeout_ms) {
+    grpc::ClientContext context;
+    if (timeout_ms != -1) {
+      context.set_deadline(std::chrono::system_clock::now() +
+                           std::chrono::milliseconds(timeout_ms));
+    }
+    if (!cluster_id_.IsNil()) {
+      context.AddMetadata(kClusterIdKey, cluster_id_.Hex());
+    }
+    return std::move(context);
+  }


Suggested change

inline grpc::ClientContext &&PrepareContext(int64_t timeout_ms) {

grpc::ClientContext context;

if (timeout_ms != -1) {

context.set_deadline(std::chrono::system_clock::now() +

std::chrono::milliseconds(timeout_ms));

}

if (!cluster_id_.IsNil()) {

context.AddMetadata(kClusterIdKey, cluster_id_.Hex());

}

return std::move(context);

}

grpc::ClientContext PrepareContext(int64_t timeout_ms) {

grpc::ClientContext context;

if (timeout_ms != -1) {

context.set_deadline(std::chrono::system_clock::now() +

std::chrono::milliseconds(timeout_ms));

}

if (!cluster_id_.IsNil()) {

context.AddMetadata(kClusterIdKey, cluster_id_.Hex());

}

return context;

}

Compiler complains because there's no copy constructor for ClientContext. Moved the ClientContext out of the helper the way it was before.

hmmm, interesting... because you actually can return a unique_ptr directly.

fishbone · 2023-07-19T22:29:10Z

src/ray/gcs/gcs_client/gcs_client.cc

@@ -432,8 +416,7 @@ Status PythonGcsClient::GetAllJobInfo(int64_t timeout_ms,
 Status PythonGcsClient::RequestClusterResourceConstraint(
    int64_t timeout_ms,
    const std::vector<std::unordered_map<std::string, double>> &bundles) {
-  grpc::ClientContext context;
-  GrpcClientContextWithTimeoutMs(context, timeout_ms);
+  grpc::ClientContext &&context = PrepareContext(timeout_ms);


Suggested change

grpc::ClientContext &&context = PrepareContext(timeout_ms);

grpc::ClientContext context = PrepareContext(timeout_ms);

python/ray/includes/common.pxd

src/ray/gcs/gcs_client/gcs_client.cc

fishbone

Overall it looks good.

fishbone · 2023-07-19T22:45:51Z

Could you run the failed release test to make sure it's working?

src/ray/gcs/gcs_client/gcs_client.cc

src/ray/raylet/main.cc

jjyao · 2023-07-20T04:51:57Z

python/ray/_private/node.py

@@ -83,6 +83,9 @@ def __init__(
        )
        self.all_processes: dict = {}
        self.removal_lock = threading.Lock()
+        self.cluster_id = (
+            ray_params.cluster_id if hasattr(ray_params, "cluster_id") else None


Why hasattr?

Originally it was because there was no enforcement that the attr was even declared, and when it's not declared, it throws exception instead of returning None (i.e. if ray_params.cluster_id = cluster_id is only called sometimes).

But RayParams is fixed now per your other comment.

python/ray/_private/workers/default_worker.py

python/ray/_private/node.py

python/ray/_private/worker.py

src/ray/gcs/gcs_client/gcs_client.cc

vitsai

good comments

vitsai · 2023-07-20T08:11:43Z

src/ray/gcs/gcs_client/gcs_client.h

+  inline grpc::ClientContext &&PrepareContext(int64_t timeout_ms) {
+    grpc::ClientContext context;
+    if (timeout_ms != -1) {
+      context.set_deadline(std::chrono::system_clock::now() +
+                           std::chrono::milliseconds(timeout_ms));
+    }
+    if (!cluster_id_.IsNil()) {
+      context.AddMetadata(kClusterIdKey, cluster_id_.Hex());
+    }
+    return std::move(context);
+  }


Compiler complains because there's no copy constructor for ClientContext. Moved the ClientContext out of the helper the way it was before.

vitsai · 2023-07-20T08:11:51Z

python/ray/_private/node.py

@@ -83,6 +83,9 @@ def __init__(
        )
        self.all_processes: dict = {}
        self.removal_lock = threading.Lock()
+        self.cluster_id = (
+            ray_params.cluster_id if hasattr(ray_params, "cluster_id") else None


Originally it was because there was no enforcement that the attr was even declared, and when it's not declared, it throws exception instead of returning None (i.e. if ray_params.cluster_id = cluster_id is only called sometimes).

But RayParams is fixed now per your other comment.

fishbone · 2023-07-20T20:11:18Z

src/ray/gcs/gcs_client/gcs_client.h

@@ -231,7 +232,20 @@ class RAY_EXPORT PythonGcsClient {
      const std::vector<std::unordered_map<std::string, double>> &bundles);
  Status GetClusterStatus(int64_t timeout_ms, std::string &serialized_reply);

+  ClusterID GetClusterId() const { return cluster_id_; }


nit: const ClusterID& maybe? not big deal btw.

fishbone

Thanks! Let's make sure the broken release test passes

jjyao

LGTM

python/ray/_private/node.py

python/ray/_private/parameter.py

python/ray/includes/unique_ids.pxd

src/ray/gcs/gcs_client/gcs_client.cc

jjyao · 2023-07-21T22:52:19Z

src/ray/gcs/gcs_client/gcs_client.cc

+
+    auto status =
+        GrpcStatusToRayStatus(node_info_stub_->GetClusterId(&context, request, &reply));
+    while (!status.IsTimedOut()) {


Should we fail after several retries?

Looks like all the other methods use timeouts

Added retries to emulate the @_auto_reconnect on the python side

src/ray/raylet/main.cc

python/ray/_private/node.py

fishbone · 2023-08-03T19:59:43Z

dashboard/modules/job/tests/test_job_agent.py

-    assert "(raylet)" not in out_str
-    assert "(raylet)" not in err_str
+
+    def check_output(blob):


Not quite sure, but the test is to make sure no logs are streamed to worker from raylet. SHould we keep that?

But we added new logs in this change that aren't errors, so we broke that invariant. Probably doesn't make sense to test for it anymore

fishbone · 2023-08-03T20:00:44Z

python/ray/_private/node.py

-                f"Please check {os.path.join(self._logs_dir, 'gcs_server.out')}"
-                " for details"
-            )
+            if hasattr(self, "_logs_dir"):


Why we need to check the attr now? Could you add comment there?

I think this was always a bug, but the path never got triggered. In the init path _logs_dir is inited after GcsClient

fishbone · 2023-08-03T20:02:26Z

python/ray/_private/worker.py

@@ -1126,6 +1126,7 @@ def init(
    namespace: Optional[str] = None,
    runtime_env: Optional[Union[Dict[str, Any], "RuntimeEnv"]] = None,  # noqa: F821
    storage: Optional[str] = None,
+    no_gcs: bool = False,


We shouldn't change the API.

should we have a global variable then? initialization order might be tricky

So the problem is that some tests don't start GCS, and then the newly added RPC hangs.

fishbone

We shouldn't change the API. Still in progress of reviewing.

python/ray/tests/test_ray_init.py

python/ray/tests/test_autoscaler.py

Signed-off-by: vitsai <victoria@anyscale.com>

Signed-off-by: vitsai <vitsai@cs.stanford.edu>

This reverts commit cfe608c. Signed-off-by: Edward Oakes <ed.nmi.oakes@gmail.com>

An earlier[ change](ray-project#37399) added a new RPC to each GCS client, causing some stress tests to fail. Instead of adding O(num_worker) RPCs, add O(num_node) RPCs by plumbing through the Python layer. Signed-off-by: Shreyas Krishnaswamy <shrekris@anyscale.com>

ray-project#38320) This reverts commit cfe608c. Signed-off-by: Edward Oakes <ed.nmi.oakes@gmail.com> Signed-off-by: Shreyas Krishnaswamy <shrekris@anyscale.com>

Redo #37583 Signed-off-by: vitsai <victoria@anyscale.com> Signed-off-by: vitsai <vitsai@cs.stanford.edu>

An earlier[ change](ray-project#37399) added a new RPC to each GCS client, causing some stress tests to fail. Instead of adding O(num_worker) RPCs, add O(num_node) RPCs by plumbing through the Python layer. Signed-off-by: NripeshN <nn2012@hw.ac.uk>

ray-project#38320) This reverts commit cfe608c. Signed-off-by: Edward Oakes <ed.nmi.oakes@gmail.com> Signed-off-by: NripeshN <nn2012@hw.ac.uk>

Redo ray-project#37583 Signed-off-by: vitsai <victoria@anyscale.com> Signed-off-by: vitsai <vitsai@cs.stanford.edu> Signed-off-by: NripeshN <nn2012@hw.ac.uk>

An earlier[ change](ray-project#37399) added a new RPC to each GCS client, causing some stress tests to fail. Instead of adding O(num_worker) RPCs, add O(num_node) RPCs by plumbing through the Python layer. Signed-off-by: e428265 <arvind.chandramouli@lmco.com>

ray-project#38320) This reverts commit cfe608c. Signed-off-by: Edward Oakes <ed.nmi.oakes@gmail.com> Signed-off-by: e428265 <arvind.chandramouli@lmco.com>

Redo ray-project#37583 Signed-off-by: vitsai <victoria@anyscale.com> Signed-off-by: vitsai <vitsai@cs.stanford.edu> Signed-off-by: e428265 <arvind.chandramouli@lmco.com>

An earlier[ change](ray-project#37399) added a new RPC to each GCS client, causing some stress tests to fail. Instead of adding O(num_worker) RPCs, add O(num_node) RPCs by plumbing through the Python layer. Signed-off-by: Victor <vctr.y.m@example.com>

ray-project#38320) This reverts commit cfe608c. Signed-off-by: Edward Oakes <ed.nmi.oakes@gmail.com> Signed-off-by: Victor <vctr.y.m@example.com>

Redo ray-project#37583 Signed-off-by: vitsai <victoria@anyscale.com> Signed-off-by: vitsai <vitsai@cs.stanford.edu> Signed-off-by: Victor <vctr.y.m@example.com>

vitsai requested a review from a team as a code owner July 19, 2023 21:19

vitsai force-pushed the gcs-raylet-plumb branch from 16b5e03 to 5e60f9a Compare July 19, 2023 21:35

fishbone self-assigned this Jul 19, 2023

fishbone reviewed Jul 19, 2023

View reviewed changes

src/ray/raylet/main.cc Outdated Show resolved Hide resolved

fishbone reviewed Jul 19, 2023

View reviewed changes

python/ray/includes/common.pxd Outdated Show resolved Hide resolved

fishbone reviewed Jul 19, 2023

View reviewed changes

src/ray/gcs/gcs_client/gcs_client.cc Outdated Show resolved Hide resolved

fishbone reviewed Jul 19, 2023

View reviewed changes

jjyao reviewed Jul 20, 2023

View reviewed changes

vitsai force-pushed the gcs-raylet-plumb branch from 5e60f9a to 983eb0c Compare July 20, 2023 08:32

vitsai commented Jul 20, 2023

View reviewed changes

fishbone reviewed Jul 20, 2023

View reviewed changes

fishbone approved these changes Jul 20, 2023

View reviewed changes

jjyao reviewed Jul 21, 2023

View reviewed changes

jjyao approved these changes Jul 24, 2023

View reviewed changes

python/ray/_private/node.py Outdated Show resolved Hide resolved

vitsai force-pushed the gcs-raylet-plumb branch 2 times, most recently from e50f292 to f7028b9 Compare July 31, 2023 21:12

vitsai force-pushed the gcs-raylet-plumb branch from 68b3cd5 to 9b6f671 Compare August 1, 2023 21:03

vitsai requested review from wuisawesome, DmitriGekhtman and ericl as code owners August 2, 2023 20:13

vitsai force-pushed the gcs-raylet-plumb branch from 9be9cf3 to 40c36cf Compare August 3, 2023 04:34

fishbone reviewed Aug 3, 2023

View reviewed changes

fishbone requested changes Aug 3, 2023

View reviewed changes

jjyao reviewed Aug 4, 2023

View reviewed changes

python/ray/tests/test_ray_init.py Outdated Show resolved Hide resolved

fishbone reviewed Aug 4, 2023

View reviewed changes

python/ray/tests/test_autoscaler.py Show resolved Hide resolved

vitsai added 3 commits August 8, 2023 23:44

preserve error types of existing logic

e442955

Signed-off-by: vitsai <victoria@anyscale.com>

increase timeout for init test

38070fc

Signed-off-by: vitsai <victoria@anyscale.com>

rebase fixes

af111c7

Signed-off-by: vitsai <victoria@anyscale.com>

vitsai force-pushed the gcs-raylet-plumb branch from 6702492 to af111c7 Compare August 9, 2023 06:56

vitsai added 2 commits August 9, 2023 22:10

fix gcs fault tolerance

b0d51b3

Signed-off-by: vitsai <vitsai@cs.stanford.edu>

increase size of threshold tests

d00c49e

Signed-off-by: vitsai <vitsai@cs.stanford.edu>

fishbone merged commit cfe608c into ray-project:master Aug 10, 2023

edoakes added a commit to edoakes/ray that referenced this pull request Aug 10, 2023

Revert "[core] Add cluster ID to the Python layer (ray-project#37583)"

48c7731

This reverts commit cfe608c. Signed-off-by: Edward Oakes <ed.nmi.oakes@gmail.com>

edoakes added a commit that referenced this pull request Aug 10, 2023

Revert "[core] Add cluster ID to the Python layer (#37583)" (#38320)

03be66c

This reverts commit cfe608c. Signed-off-by: Edward Oakes <ed.nmi.oakes@gmail.com>

vitsai mentioned this pull request Aug 11, 2023

[core] Add cluster ID to python layer #38350

Merged

8 tasks

jjyao pushed a commit that referenced this pull request Aug 11, 2023

[core] Add cluster ID to python layer (#38350)

409fb86

Redo #37583 Signed-off-by: vitsai <victoria@anyscale.com> Signed-off-by: vitsai <vitsai@cs.stanford.edu>

vitsai mentioned this pull request Aug 16, 2023

[core] Potential performance regression during ray start due to new RPC #38509

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[core] Add cluster ID to the Python layer #37583

[core] Add cluster ID to the Python layer #37583

vitsai commented Jul 19, 2023

fishbone Jul 19, 2023

vitsai Jul 20, 2023

fishbone Jul 20, 2023

fishbone Jul 19, 2023

fishbone left a comment

fishbone commented Jul 19, 2023

jjyao Jul 20, 2023

vitsai Jul 20, 2023

vitsai left a comment

vitsai Jul 20, 2023

vitsai Jul 20, 2023

fishbone Jul 20, 2023

fishbone left a comment •

edited

Loading

jjyao left a comment

jjyao Jul 21, 2023

vitsai Jul 24, 2023

vitsai Jul 26, 2023

fishbone Aug 3, 2023

vitsai Aug 3, 2023

vitsai Aug 3, 2023

fishbone Aug 3, 2023

vitsai Aug 3, 2023

fishbone Aug 3, 2023

vitsai Aug 3, 2023

vitsai Aug 3, 2023 •

edited

Loading

fishbone left a comment

	grpc::ClientContext &&context = PrepareContext(timeout_ms);
	grpc::ClientContext context = PrepareContext(timeout_ms);

[core] Add cluster ID to the Python layer #37583

[core] Add cluster ID to the Python layer #37583

Conversation

vitsai commented Jul 19, 2023

Why are these changes needed?

Related issue number

Checks

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fishbone left a comment

Choose a reason for hiding this comment

fishbone commented Jul 19, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vitsai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fishbone left a comment • edited Loading

Choose a reason for hiding this comment

jjyao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vitsai Aug 3, 2023 • edited Loading

Choose a reason for hiding this comment

fishbone left a comment

Choose a reason for hiding this comment

fishbone left a comment •

edited

Loading

vitsai Aug 3, 2023 •

edited

Loading