Repository navigation
REP: Ray Active-Passive Head Architecture - #65
Conversation
Signed-off-by: Yuchen Zhou <yczhou@google.com>
5766f6e to
47c687c
Compare
Signed-off-by: Yuchen Zhou <yczhou@google.com>
3caa578 to
ded3bd1
Compare
| #### 5. KubeRay Changes | ||
| 1. **API change**: specifying the number of desired head replicas. | ||
| 2. **Configuration changes**: | ||
| - Readiness Probe for passive heads to fail. |
There was a problem hiding this comment.
IIRC, probes are immutable. So, probably the change should be in GCS?
There was a problem hiding this comment.
Yeah, I was originally reaching to k8s APIServer directly to fetch the lease status for readiness check but it could be part of the gcs_health endpoint. In this way, no need to update the readiness probe. Updated.
| 3. **Fencing Token**: The last barrier to prevent data inconsistency. It is also one of the best practices for implementing lease-based leader election in production. At the end of the day, we must make sure the data stored in redis is pure. | ||
| - **The Token**: Lease transition, it is updated when a new leader takes over the lease | ||
| - **Write to Redis**: the leader GCS attaches the token with the query | ||
| - **Fencing**: The Redis rejects the request if the request_epoch is smaller than its current_epoch |
There was a problem hiding this comment.
@YoyinZyc how exactly does redis reject requests based on the token? Is it a feature in Redis or something else?
There was a problem hiding this comment.
I think this can be implemented with redis lua script, but I am wondering if doing fencing in redis is enough. Shouldn't we also have this fencing in all GCS -> raylet communication?
There was a problem hiding this comment.
Yes, it is with lua script, added more details and the example lua script. GCS -> raylet communication involves more design and consideration, I added it as the a phase 2 item which should fall into later iteration.
| The latter two are Ray-specific optimizations for the GCS HA scenario. Techenically, they could provide additional data consistency guarantees. However, it may involve other potential risk and additional implementation work.(e.g. communication between heads/workers could be unreliable) We can leave it to the future iterations. | ||
| ##### Metrics and Monitoring | ||
| - `gcs_is_leader`: Emits a 1 if the current instance is the leader, and a 0 if it is a passive head. | ||
| - `gcs_leader_transitions_total`: Incremented every time a leadership change is detected |
There was a problem hiding this comment.
where would this be exported?
There was a problem hiding this comment.
From the gcs process.
| ##### Metrics and Monitoring | ||
| - `gcs_is_leader`: Emits a 1 if the current instance is the leader, and a 0 if it is a passive head. | ||
| - `gcs_leader_transitions_total`: Incremented every time a leadership change is detected | ||
| - `gcs_recovery_latency_ms`: The time it takes for a passive head to recover the state and become the leader. |
There was a problem hiding this comment.
same for this, where/how is it recorded?
There was a problem hiding this comment.
It is recorded from the gcs server. gcs_is_leader is a gauge metrics tracking local leadership status; gcs_leader_transitions_total increaments when the passive becomes the leader;
gcs_recovery_latency_ms tracks the passive head recovery latency from the moment it acquires the leadership to the moment the head is functional.
| The first three mechanisms are production golden standards for leader election implementations. They are generic and must be included in the first phase of this REP. | ||
|
|
||
| The latter two are Ray-specific optimizations for the GCS HA scenario. Techenically, they could provide additional data consistency guarantees. However, it may involve other potential risk and additional implementation work.(e.g. communication between heads/workers could be unreliable) We can leave it to the future iterations. | ||
| ##### Metrics and Monitoring |
There was a problem hiding this comment.
are there any other metrics around API server latency / health that we should observe?
There was a problem hiding this comment.
APIServer has a bunch of metrics but those are pretty generic like apiserver_request_total, apiserver_request_duration_seconds and apiserver_current_inflight_requests
There was a problem hiding this comment.
It is probably worth also exporting similar metrics from our lease client
There was a problem hiding this comment.
What similar metrics are you referring to? These are k8s APIServer metrics, how could the lease client export them?
There was a problem hiding this comment.
I mean our lease client should also export metrics like the latency for its requests to the API server and a failed request count
There was a problem hiding this comment.
Ah I see, added two more metrics for lease renew latency and failure count.
| - **Raylet** | ||
| Safe to run only if launched with `--num-cpus=0` and `--num-gpus=0`, ensuring the active primary never schedules user tasks onto the standby head node. |
There was a problem hiding this comment.
Setting these is not strong enough -- actors/tasks with no resource requests can still be scheduled there.
Can we gate the Raylet from joining the cluster until after the GCS is elected the leader? This is similar to the problem w/ the autoscaler that I pointed out above
There was a problem hiding this comment.
yes, autoscaler must be suppressed. Raylet may be fine since it anyway only talks to the local gcs but we should restrict its registration which incurs the redis write. Added more detail about components behavior.
Signed-off-by: Yuchen Zhou <yczhou@google.com>
18017f5 to
e7b5d1a
Compare
Signed-off-by: Yuchen Zhou <yczhou@google.com>
Signed-off-by: Yuchen Zhou <yczhou@google.com>
Signed-off-by: Yuchen Zhou <yczhou@google.com>
Signed-off-by: Gang Li <ganglica@google.com>
Update autoscaler section in Active-Passive REP
| - **Raylet**: it is required to be alive to answer the readiness and liveness probes. | ||
| - **Startup**: Upon boot, the local standby Raylet targets the local GCS and sends a RegisterNode gRPC request. On passive head, the PUT request should not reach to Redis. | ||
| - **Passive State**: The Raylet continuously sends heartbeats to the local GCS to keep its node liveness. | ||
| - **Promotion**: During promotion, the Raylet should register itself and physically writes it to Redis. | ||
| - **Autoscaler**: it is a critical, high-risk background component that must be suppressed on passive head. | ||
| - **Startup**: Upon boot, the local autoscaler targets the local GCS and periodically sends a GetClusterResourceState gRPC request to check if the cluster resource state in GCS is ready. The cluster resource state should only be ready when the GCS is in active mode. | ||
| - **Active State**: Autoscaler gets the cluster resource state from GCS and continues with the normal autoscaling logic including updating KubeRay CRDs, reporting the autoscaling state back to GCS. | ||
| - **Passive State**: Autoscaler gets the cluster resource state from GCS, which indicates the cluster resource state is not ready. The autoscaler skips the rest of the logic. |
There was a problem hiding this comment.
Will each of these also be monitoring that its GCS is the leader and fast-failing when it loses the lease?
There was a problem hiding this comment.
@gangli113 helped adding more details about autoscaler, let us know if there is any other concerns. Thanks!
edoakes
left a comment
There was a problem hiding this comment.
Largely looks good. Primary open feedback:
- Please document the full scope of KubeRay changes (especially any user-facing changes).
- We will need the same safety guarantees for the autoscaler process. It will either need to fast fail when the GCS fails or also directly check w/ a lease client.
| // +kubebuilder:default:=1 | ||
| // +kubebuilder:validation:Minimum:=1 | ||
| // +kubebuilder:validation:Maximum:=2 | ||
| Replicas *int32 `json:"replicas,omitempty"` |
There was a problem hiding this comment.
Are we introducing replicas now because we anticipate it can be more than 2 head nodes in the future?
There was a problem hiding this comment.
(As opposed to something like mode: SingleReplica|ActivePassive)
There was a problem hiding this comment.
I think replicas is confusing because it implies a user can set it to any arbitrary value.
A field in gcsFaultToleranceOption is probably better?
spec:
gcsFaultToleranceOptions:
mode: ActivePassive
..
...
This is slightly more intuitive because active-passive mode also requires enabling Redis persistence anyways right?
There was a problem hiding this comment.
I was also thinking about just using a boolean or mode to control the enablement. But using replicas is more kubernetes native for controller to reconcile and can be extended if ray eventually wants to support more replicas?
There was a problem hiding this comment.
But using replicas is more kubernetes native for controller to reconcile and can be extended if ray eventually wants to support more replicas?
I think we can just introduce replicas field in the future when we can support > 2 replicas. Putting replicas outside of gcsFaultToleranceOptions implies you can scale to N replicas without enabling GCS persistence, which is not the case. Putting the mode in gcsFaultToleranceOptions makes it clearer that you can only enable active-passive replica mode when GCS persistence is also enabled
There was a problem hiding this comment.
Ok changed to using EnableActivePassiveHead under gcsFaultToleranceOptions to gate the enablement of the feature.
Signed-off-by: Yuchen Zhou <yczhou@google.com>
Signed-off-by: Yuchen Zhou <yczhou@google.com>
…sive REP Signed-off-by: Gang Li <ganglica@google.com>
Update Autoscaler section for GCS failure and promotion in Active-Passive REP
| - **Active State**: Autoscaler gets the cluster resource state from GCS and continues with the normal autoscaling logic including updating KubeRay CRDs, reporting the autoscaling state back to GCS. | ||
| - **Passive State**: Autoscaler gets the cluster resource state from GCS, which indicates the cluster resource state is not ready. The autoscaler skips the rest of the logic. | ||
| - **Dashboard**: the local Dashboard process is started by KubeRay pointing to the local GCS. Because it queries the passive local GCS, the dashboard UI initially displays a single-node cluster (only itself) with no active jobs or actors. Since the standby pod is Not Ready, its dashboard API is completely unreachable by external clients with the current k8s service setup. | ||
| - **Job API**: with right k8s service configuration, the `ray job submit` request should not be able to reach the passive head. |
There was a problem hiding this comment.
We should not rely on this as the only mechanism. Users may submit job requests in a different way (e.g., exec'ing into the pod and running the job submit command there)
| - **Passive State**: Autoscaler gets the cluster resource state from GCS, which indicates the cluster resource state is not ready. The autoscaler skips the rest of the logic. | ||
| - **Dashboard**: the local Dashboard process is started by KubeRay pointing to the local GCS. Because it queries the passive local GCS, the dashboard UI initially displays a single-node cluster (only itself) with no active jobs or actors. Since the standby pod is Not Ready, its dashboard API is completely unreachable by external clients with the current k8s service setup. | ||
| - **Job API**: with right k8s service configuration, the `ray job submit` request should not be able to reach the passive head. | ||
| - **Serve Controller**: The Ray Serve Controller is a standard Ray actor. Since GCS in standby mode has no active workers and no running jobs, no Serve Controller is ever spawned on the standby head. |
There was a problem hiding this comment.
It doesn't matter if the serve controller isn't run on the standby head, the serve API endpoints in the dashboard on a passive head could still make requests to the active head
Most likely I think we need a shared bit for API server components to check if they are the active or passive and fail requests when passive. This can be used for serve API & jobs API
There was a problem hiding this comment.
We could even abstract it into a utility/decorator that performs the check & rejects requests with an appropriate status code
There was a problem hiding this comment.
I may need more clarification on this. what do you mean about "serve API endpoints in the dashboard on a passive head could still make requests to the active head"?
And you are suggesting that whenever there is an unexpected request hit Job API or Serve API, it should ask its local gcs to check if it is leader or passive. If it is passive, reject the request. Am I understanding correctly?
There was a problem hiding this comment.
also in terms of the decorator, are you saying something on gcs side like a rpc or API for leadership status checking?
There was a problem hiding this comment.
you are suggesting that whenever there is an unexpected request hit Job API or Serve API, it should ask its local gcs to check if it is leader or passive. If it is passive, reject the request. Am I understanding correctly?
Yes that's right.
also in terms of the decorator, are you saying something on gcs side like a rpc or API for leadership status checking?
I am saying we could have a shared utility or python decorator in the dashboard logic that automatically performs this check before accepting any requests
| **1. Single-container co-located components**: | ||
| These components are running in the same container as different subprocesses. The gcs termination caused by lease lost will terminate other processes. | ||
| - **GCS Server** | ||
| Safe on Standby. Global tracking managers are instantiated empty; state hydration is deferred until promotion. The rpc server can only answer read requests in passive mode. All mutation requests should not be accepted and the rejection should be handled gracefully on the client side. (With the proper traffic routing, the requests will be sent to the active head only. But it is another layer of protection to avoid any potential split brain issue.) |
There was a problem hiding this comment.
The rpc server can only answer read requests in passive mode
Nit: It could be implementation detail, but I think it might be better just to identity the endpoints that needs to be called and allow them instead of open all the read endpoints.
There was a problem hiding this comment.
At least GetClusterID and healthcheck rpc have to be enabled on passive. I may uncover more during the implementation.
| - **Dashboard**: the local Dashboard process is started by KubeRay pointing to the local GCS. Because it queries the passive local GCS, the dashboard UI initially displays a single-node cluster (only itself) with no active jobs or actors. Since the standby pod is Not Ready, its dashboard API is completely unreachable by external clients with the current k8s service setup. | ||
| - **Job API**: with right k8s service configuration, the `ray job submit` request should not be able to reach the passive head. | ||
| - **Serve Controller**: The Ray Serve Controller is a standard Ray actor. Since GCS in standby mode has no active workers and no running jobs, no Serve Controller is ever spawned on the standby head. | ||
| - **Dashboard Agent**: Runs locally and reports metrics to the co-located passive GCS. Upon promotion, it continues reporting to the now-promoted local GCS. Ready to initialize runtime environments if new jobs are submitted post-failover. |
There was a problem hiding this comment.
nit: We might also want to disable event generation in the passive mode. There might not be many events generated but disabling it as a safe guard
There was a problem hiding this comment.
I think the event generation is controlled by some envs(like RAY_enable_ray_event or RAY_DASHBOARD_AGGREGATOR_AGENT_PUBLISH_EVENTS_TO_GCS ). Is it possible the re-enable the event generation at run time after promotion? Or are you referring to other approach to disable the event generation?
There was a problem hiding this comment.
I wasn't referring to any specific mechanism but more of a high level safe guard.
You are right. The configurations are all specified at the start time. We probably need to add an additional config in the aggregator agent to control whether to emit events at the runtime.
There was a problem hiding this comment.
Then the most intuitive approach is to ask the dashboard agent to watch and poll the leadership status periodically (in low frequency, 5s) and emit event post-promotion only.
…assive enablement Signed-off-by: Yuchen Zhou <yczhou@google.com>
Signed-off-by: Yuchen Zhou <yczhou@google.com>
edoakes
left a comment
There was a problem hiding this comment.
Looks good to me as a starting point for this feature; I'm sure there will be some more low level details to iron out as we go.
…ion, observability example - Expand the Acceptance Criteria subsection to include the four criteria from @andrewsykim (Ray release tests, Ray Core GCS FT test equivalent, Ray scale tests, Ray Serve chaos testing) and incorporate @edoakes's clarification that the Redis-equivalent test should be a postmerge CI job using a local-file RocksDB, labeled skip-premerge with path-based triggers on src/ray/gcs/**. - Add an "Interaction with Active-Passive Head (REP ray-project#65)" subsection covering the ReadWriteMany topology, two-layer split-brain prevention (lease + RocksDB LOCK file), failover sequence, and RWX latency trade-off. Update Follow-on Work item 2 to mark Active/Passive GCS as in flight and link to REP ray-project#65. - Add observability data (event logs, task/actor history, log indexes) as an explicit example of non-critical, high-volume state that the future cold-paging optimization could keep primarily on disk. Signed-off-by: Santosh Jha <santosh.m.jha@gmail.com>
This REP proposes the transition from a single-point-of-failure GCS model to a highly available Active-Passive setup. The design aims to reduce recovery latency from minutes to seconds.
Scope:
Requested reviewers: @andrewsykim @edoakes @MengjinYan