[Feature] [history server] Support token auth in the history server collector - #5078
[Feature] [history server] Support token auth in the history server collector#5078CheyuWu wants to merge 22 commits into
Conversation
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
|
@win5923 @machichima PTAL |
|
Could you update the PR description to show:
|
|
Please correct me if I'm wrong, #4520 only supports secret-based token auth and explicitly rejects enableK8sTokenAuth: kuberay/historyserver/pkg/historyserver/clientmanager.go Lines 103 to 107 in a6f0035 Should we keep the collector consistent and focus on secret mode here |
Yes, but I think the collector should support both auth modes, since users can able to view the collected data for dead clusters. |
SG, let's keep it. For history server one we can do in follow-up |
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
| func testCollectorWithTokenAuth(test Test, g *WithT, namespace *corev1.Namespace, s3Client *s3.S3) { | ||
| rayCluster := ApplyRayClusterWithCollectorTokenAuth(test, g, namespace) | ||
|
|
||
| // The operator generates the auth Secret; without it the sidecars could not resolve a token. | ||
| secretName := utils.CheckName(rayCluster.Name) | ||
| g.Eventually(func(gg Gomega) { | ||
| secret, err := test.Client().Core().CoreV1().Secrets(namespace.Name).Get(test.Ctx(), secretName, metav1.GetOptions{}) | ||
| gg.Expect(err).NotTo(HaveOccurred()) | ||
| gg.Expect(secret.Data).To(HaveKey(utils.RAY_AUTH_TOKEN_SECRET_KEY)) | ||
| gg.Expect(secret.Data[utils.RAY_AUTH_TOKEN_SECRET_KEY]).NotTo(BeEmpty()) | ||
| }, TestTimeoutShort).Should(Succeed()) | ||
|
|
||
| headPod, err := GetHeadPod(test, rayCluster) | ||
| g.Expect(err).NotTo(HaveOccurred()) | ||
|
|
||
| // Guard against a vacuous pass: if the Dashboard served this cluster without credentials the | ||
| // rest of the assertions would hold even with the fix reverted. Probed with Python because | ||
| // the Ray images ship no curl. | ||
| probe := `python3 - <<'PY' | ||
| import urllib.error, urllib.request | ||
| try: | ||
| urllib.request.urlopen("http://localhost:8265/api/v0/nodes?limit=1", timeout=10) | ||
| print("200") | ||
| except urllib.error.HTTPError as err: | ||
| print(err.code) | ||
| PY` | ||
| unauthenticated, _ := ExecPodCmd(test, headPod, "ray-head", []string{"sh", "-c", probe}) | ||
| g.Expect(strings.TrimSpace(unauthenticated.String())).To(Equal("401"), | ||
| "Dashboard must reject unauthenticated requests, otherwise this test proves nothing") | ||
|
|
||
| // The crash loop from #5056 took ~60s to surface, so a plain Expect right after startup would | ||
| // pass even against the broken build. Hold the assertion open instead. | ||
| LogWithTimestamp(test.T(), "Verifying collector containers do not crash-loop under token auth") | ||
| g.Consistently(func(gg Gomega) { | ||
| pod, err := GetHeadPod(test, rayCluster) | ||
| gg.Expect(err).NotTo(HaveOccurred()) | ||
|
|
||
| var found bool | ||
| for _, status := range pod.Status.ContainerStatuses { | ||
| if status.Name != "collector" { | ||
| continue | ||
| } | ||
| found = true | ||
| gg.Expect(status.RestartCount).To(BeZero(), "collector restarted, it is likely failing to authenticate") | ||
| gg.Expect(status.State.Running).NotTo(BeNil(), "collector is not running") | ||
| } | ||
| gg.Expect(found).To(BeTrue(), "collector container status not reported") | ||
| }, TestTimeoutShort, 5*time.Second).Should(Succeed()) | ||
|
|
||
| // Data in storage is the real proof: reaching it requires an authenticated Dashboard call. | ||
| sessionID := GetSessionIDFromHeadPod(test, g, rayCluster) | ||
| storageKey := utils.EndpointPathToStorageKey(EndpointTimezone) | ||
| sessionDir := clusterlogs.SessionDir("log", "", "", rayCluster.Namespace, rayCluster.Name, sessionID) | ||
| timezoneKey := fmt.Sprintf("%s/%s/%s", sessionDir, utils.RAY_SESSIONDIR_FETCHED_ENDPOINTS_NAME, storageKey) | ||
|
|
||
| LogWithTimestamp(test.T(), "Waiting for authenticated endpoint data at S3 key: %s", timezoneKey) | ||
| g.Eventually(func(gg Gomega) { | ||
| result, err := s3Client.GetObject(&s3.GetObjectInput{ | ||
| Bucket: aws.String(S3BucketName), | ||
| Key: new(timezoneKey), | ||
| }) | ||
| gg.Expect(err).NotTo(HaveOccurred()) | ||
| defer result.Body.Close() | ||
|
|
||
| body, err := io.ReadAll(result.Body) | ||
| gg.Expect(err).NotTo(HaveOccurred()) | ||
| gg.Expect(body).NotTo(BeEmpty()) | ||
|
|
||
| var timezone map[string]any | ||
| gg.Expect(json.Unmarshal(body, &timezone)).To(Succeed()) | ||
| gg.Expect(timezone).To(HaveKey("value")) | ||
| }, TestTimeoutMedium).Should(Succeed()) | ||
|
|
||
| DeleteS3Bucket(test, g, s3Client) | ||
| } |
There was a problem hiding this comment.
I think we can simplify this test:
ApplyRayClusterWithCollectorTokenAuthalready waits for RayCluster Ready, which means the collector must already be up and authenticating.FetchAndStoreTimezonehits the Dashboard at startup and writes to S3 so the S3 object is the proof that an authenticated Dashboard call succeeded. That's the assertion worth keeping. Let's extract its S3 check into a helper and reuse it here instead of duplicating.
func testCollectorWithTokenAuth(...) {
rayCluster := ApplyRayClusterWithCollectorTokenAuth(test, g, namespace)
// verify dashboard rejects the unauthenticated requests
out, err := ExecPodCmd(test, headPod, "ray-head", []string{"sh", "-c", probe})
g.Expect(err).NotTo(HaveOccurred())
g.Expect(strings.TrimSpace(out.String())).To(Equal("401"))
// resue timezone assertion from testCollectorStoresTimezone to test the S3 key
assertTimezoneStored(test, g, rayCluster, s3Client)
}Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
|
Hi @win5923 PTAL Few things beyond the exact suggestions:
By the way, do we need to add e2e test for |
Sure, I will open the follow-up issue, since this PR has been merged |
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
|
@win5923 PTAL |
Can be a follow-up. |
There was a problem hiding this comment.
| proxyReq.Header.Set(utils.RayAuthHeader, fmt.Sprintf("Bearer %s", authToken)) |
| namespace: default | ||
| spec: | ||
| # Enable Ray token authentication. Requires Ray 2.52.0+ | ||
| # rayVersion: "2.52.0" |
There was a problem hiding this comment.
| # rayVersion: "2.52.0" | |
| # rayVersion: "2.56.0" |
Follow the ray image that we use.
| namespace: default | ||
| spec: | ||
| # Enable Ray token authentication. Requires Ray 2.52.0+ | ||
| # rayVersion: "2.52.0" |
There was a problem hiding this comment.
| # rayVersion: "2.52.0" | |
| # rayVersion: "2.56.0" |
ditto
| containers: | ||
| - name: ray-head | ||
| imagePullPolicy: IfNotPresent | ||
| image: rayproject/ray:2.55.0 |
| value: "http://localhost:8084/v1/events" | ||
| - name: RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES | ||
| value: "ALL" | ||
| image: rayproject/ray:2.55.0 |
There was a problem hiding this comment.
Should change to ray 2.56.0, right?
There was a problem hiding this comment.
NVM, there’s already a PR that will update the Ray image to 2.56.0 in the sample YAML.
#5111
| // The collector queries the Ray Dashboard, so it needs the same credentials as the Ray | ||
| // container. | ||
| if utils.IsAuthEnabled(&instance.Spec) { | ||
| SetContainerTokenAuthEnvVars(instance.Name, &collectorContainer, instance.Spec.AuthOptions) | ||
| } |
There was a problem hiding this comment.
Have you also tested the RayClusterHistoryServer feature gate with the ray-cluster.historyserver.yaml sample and token auth enabled?
There was a problem hiding this comment.
I have tested the ray-cluster.historyserver.yaml and update the description.
To make sure its easier for user to use. I also add the comment in 083e8c4
| // Worker collectors also reach the Dashboard (via FQ_RAY_IP) to discover their Ray NodeID. | ||
| if utils.IsAuthEnabled(&instance.Spec) { | ||
| SetContainerTokenAuthEnvVars(instance.Name, &collectorContainer, instance.Spec.AuthOptions) | ||
| } |
There was a problem hiding this comment.
This comment is unrelated to the following.
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
Reviewed by Cursor Bugbot for commit 6d71cbb. Configure here.
Signed-off-by: Cheyu Wu <cheyu1220@gmail.com>
There was a problem hiding this comment.
LGTM, thank you so much for the verify! cc @rueian and @andrewsykim for the final review.
| value: "http://localhost:8084/v1/events" | ||
| - name: RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES | ||
| value: "ALL" | ||
| image: rayproject/ray:2.55.0 |
There was a problem hiding this comment.
NVM, there’s already a PR that will update the Ray image to 2.56.0 in the sample YAML.
#5111
| # set at the same time. | ||
| # authOptions: | ||
| # mode: 'token' | ||
| # enableK8sTokenAuth: true |
There was a problem hiding this comment.
Could you remove the auth config from this YAML? I'd like to keep the file simple.
historyserver/config/raycluster-kubernetes-auth.yaml is enough.
This reverts commit 083e8c4.


Why are these changes needed?
The collector talks to the Ray Dashboard but has no token support at all, so it breaks on any
RayCluster with
spec.authOptions.mode: tokenRelated issue number
Closes #5056
Labels
doc-updates-requiredlabel.breaking-changelabel.Checks
Manual test instructions
Basic setup
Follow the instruction to setup operator and MinIO
https://github.com/ray-project/kuberay/blob/master/historyserver/DEVELOPMENT.md#step-1-set-up-kind-and-kuberay-operator
Launch collector
uncomment the block in
historyserver/config/raycluster.yamlSpec
headGroupSpec
workerGroupSpecs
Apply the yaml file -
historyserver/config/raycluster.yamlSend the unauth request to ray dashboard -> 401
Check the restart count is zero
Send the request with auth to ray dashboard -> 200
port forward minio and check the log exists
kubectl -n minio-dev port-forward svc/minio-service 9001:9001 9000:9000kubernetes auth for history server raycluster
Apply the yaml
Check the status code
verify the status
Minio log
Apply the yaml with historyServerOptions ->
ray-operator/config/samples/ray-cluster.historyserver.yamlUncomment the ray token portion
Change the env settings to
Uncomment the
serviceAccountNameinheadGroupSpecandworkerGroupSpecsUpdate the ray image version to
rayproject/ray:2.56.0in head and workerUncomment the serviceAccount portion
load the latest image
Enable the feature gate
Apply the yaml
kubectl apply -f ray-operator/config/samples/ray-cluster.historyserver.yamlVerify the result
kubectl get pods -l ray.io/cluster=raycluster-historyserverGo to MinIO console
