diff --git a/historyserver/config/raycluster-azureblob.yaml b/historyserver/config/raycluster-azureblob.yaml index 80150df58db..a66ef56cb89 100644 --- a/historyserver/config/raycluster-azureblob.yaml +++ b/historyserver/config/raycluster-azureblob.yaml @@ -6,6 +6,10 @@ metadata: name: raycluster-historyserver namespace: default spec: + # Enable Ray token authentication. Requires Ray 2.52.0+ + # rayVersion: "2.56.0" + # authOptions: + # mode: token headGroupSpec: rayStartParams: dashboard-host: 0.0.0.0 @@ -72,6 +76,13 @@ spec: value: "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite-service.azurite-dev.svc.cluster.local:10000/devstoreaccount1;" - name: AZURE_STORAGE_CONTAINER value: ray-historyserver + # Uncomment when the RayCluster sets `spec.authOptions.mode: token`; the collector + # needs the token to call the Dashboard. + # - name: RAY_AUTH_TOKEN + # valueFrom: + # secretKeyRef: + # name: raycluster-historyserver # defaults to the RayCluster name + # key: auth_token volumeMounts: - name: historyserver mountPath: *rayTmpRoot @@ -146,6 +157,13 @@ spec: value: "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite-service.azurite-dev.svc.cluster.local:10000/devstoreaccount1;" - name: AZURE_STORAGE_CONTAINER value: ray-historyserver + # Uncomment when the RayCluster sets `spec.authOptions.mode: token`; the collector + # needs the token to call the Dashboard. + # - name: RAY_AUTH_TOKEN + # valueFrom: + # secretKeyRef: + # name: raycluster-historyserver # defaults to the RayCluster name + # key: auth_token volumeMounts: - name: historyserver mountPath: *rayTmpRoot diff --git a/historyserver/config/raycluster-gcs.yaml b/historyserver/config/raycluster-gcs.yaml index 6cd8dc50bbf..f574e0d5bc2 100644 --- a/historyserver/config/raycluster-gcs.yaml +++ b/historyserver/config/raycluster-gcs.yaml @@ -8,6 +8,10 @@ metadata: name: raycluster-historyserver namespace: default spec: + # Enable Ray token authentication. Requires Ray 2.52.0+ + # rayVersion: "2.56.0" + # authOptions: + # mode: token headGroupSpec: rayStartParams: dashboard-host: 0.0.0.0 @@ -93,6 +97,13 @@ spec: value: "gcs" - name: EVENTS_PORT value: "8084" + # Uncomment when the RayCluster sets `spec.authOptions.mode: token`; the collector + # needs the token to call the Dashboard. + # - name: RAY_AUTH_TOKEN + # valueFrom: + # secretKeyRef: + # name: ${RAY_CLUSTER} # defaults to the RayCluster name + # key: auth_token volumeMounts: - name: historyserver mountPath: *rayTmpRoot @@ -165,6 +176,13 @@ spec: value: "gcs" - name: EVENTS_PORT value: "8084" + # Uncomment when the RayCluster sets `spec.authOptions.mode: token`; the collector + # needs the token to call the Dashboard. + # - name: RAY_AUTH_TOKEN + # valueFrom: + # secretKeyRef: + # name: ${RAY_CLUSTER} # defaults to the RayCluster name + # key: auth_token volumeMounts: - name: historyserver mountPath: *rayTmpRoot diff --git a/historyserver/config/raycluster-kubernetes-auth.yaml b/historyserver/config/raycluster-kubernetes-auth.yaml new file mode 100644 index 00000000000..d1d8e07ac25 --- /dev/null +++ b/historyserver/config/raycluster-kubernetes-auth.yaml @@ -0,0 +1,274 @@ +apiVersion: ray.io/v1 +kind: RayCluster +metadata: + labels: + ray.io/cluster: raycluster-historyserver-kubernetes-auth + name: raycluster-historyserver-kubernetes-auth +spec: + # enableK8sTokenAuth requires Ray 2.55.0+. authOptions.secretName must NOT be + # set at the same time. + rayVersion: '2.56.0' + authOptions: + mode: 'token' + enableK8sTokenAuth: true + headGroupSpec: + rayStartParams: + dashboard-host: 0.0.0.0 + num-cpus: "0" + template: + spec: + serviceAccountName: raycluster-historyserver-kubernetes-auth + containers: + - name: ray-head + imagePullPolicy: IfNotPresent + image: rayproject/ray:2.56.0 + env: + - name: RAY_TMP_ROOT + value: &rayTmpRoot /tmp/ray + - name: RAY_enable_ray_event + value: "true" + - name: RAY_enable_core_worker_ray_event_to_aggregator + value: "true" + - name: RAY_DASHBOARD_AGGREGATOR_AGENT_EVENTS_EXPORT_ADDR + value: "http://localhost:8084/v1/events" + - name: RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES + value: "ALL" + resources: + limits: + cpu: "5" + memory: "10Gi" + requests: + cpu: "50m" + memory: "1Gi" + ports: + - containerPort: 6379 + name: gcs-server + - containerPort: 8265 + name: dashboard + - containerPort: 10001 + name: client + volumeMounts: + - name: historyserver + mountPath: *rayTmpRoot + - name: collector + image: collector:v0.1.0 + imagePullPolicy: IfNotPresent + env: + - name: RAY_CLUSTER_NAME + valueFrom: + fieldRef: + fieldPath: metadata.labels['ray.io/cluster'] + - name: RAY_CLUSTER_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: POD_IP + valueFrom: + fieldRef: + fieldPath: status.podIP + - name: FQ_RAY_IP + value: $(RAY_CLUSTER_NAME)-head-svc.$(RAY_CLUSTER_NAMESPACE).svc.cluster.local + - name: RAY_TMP_ROOT + value: *rayTmpRoot + # RAY_DASHBOARD_ADDRESS is used by the head collector to fetch endpoints' results + # (e.g., /api/v0/cluster_metadata) from the Ray Dashboard running in the same pod. + # Only the head collector uses this; worker collectors do not need it. + - name: RAY_DASHBOARD_ADDRESS + value: "http://localhost:8265" + - name: RAY_AUTH_MODE + value: "token" + - name: RAY_ENABLE_K8S_TOKEN_AUTH + value: "true" + - name: AWS_ACCESS_KEY_ID + value: minioadmin + - name: AWS_SECRET_ACCESS_KEY + value: minioadmin + - name: AWS_SESSION_TOKEN + value: "" + - name: S3_BUCKET + value: "ray-historyserver" + - name: S3_ENDPOINT + value: "minio-service.minio-dev:9000" + - name: S3_REGION + value: "test" + - name: S3FORCE_PATH_STYLE + value: "true" + - name: S3DISABLE_SSL + value: "true" + - name: RAY_ROLE + value: "Head" + - name: STORAGE_BACKEND + value: "s3" + - name: RAY_ROOT_DIR + value: "log" + resources: + limits: + cpu: "200m" + memory: "256Mi" + requests: + cpu: "50m" + memory: "64Mi" + volumeMounts: + - name: historyserver + mountPath: *rayTmpRoot + - name: ray-token + mountPath: /var/run/secrets/ray.io/serviceaccount + readOnly: true + volumes: + - name: historyserver + emptyDir: {} + workerGroupSpecs: + - replicas: 1 + minReplicas: 1 + maxReplicas: 5 + groupName: workergroup + rayStartParams: {} + template: + spec: + serviceAccountName: raycluster-historyserver-kubernetes-auth + containers: + - name: ray-worker + imagePullPolicy: IfNotPresent + image: rayproject/ray:2.56.0 + env: + - name: RAY_TMP_ROOT + value: *rayTmpRoot + - name: RAY_enable_ray_event + value: "true" + - name: RAY_enable_core_worker_ray_event_to_aggregator + value: "true" + - name: RAY_DASHBOARD_AGGREGATOR_AGENT_EVENTS_EXPORT_ADDR + value: "http://localhost:8084/v1/events" + - name: RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES + value: "ALL" + resources: + limits: + cpu: "2" + memory: "2Gi" + requests: + cpu: "50m" + memory: "1Gi" + volumeMounts: + - name: historyserver + mountPath: *rayTmpRoot + - name: collector + image: collector:v0.1.0 + imagePullPolicy: IfNotPresent + env: + - name: RAY_CLUSTER_NAME + valueFrom: + fieldRef: + fieldPath: metadata.labels['ray.io/cluster'] + - name: RAY_CLUSTER_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: POD_IP + valueFrom: + fieldRef: + fieldPath: status.podIP + - name: FQ_RAY_IP + value: $(RAY_CLUSTER_NAME)-head-svc.$(RAY_CLUSTER_NAMESPACE).svc.cluster.local + - name: RAY_TMP_ROOT + value: *rayTmpRoot + - name: RAY_AUTH_MODE + value: "token" + - name: RAY_ENABLE_K8S_TOKEN_AUTH + value: "true" + - name: AWS_ACCESS_KEY_ID + value: minioadmin + - name: AWS_SECRET_ACCESS_KEY + value: minioadmin + - name: AWS_SESSION_TOKEN + value: "" + - name: S3_BUCKET + value: "ray-historyserver" + - name: S3_ENDPOINT + value: "minio-service.minio-dev:9000" + - name: S3_REGION + value: "test" + - name: S3FORCE_PATH_STYLE + value: "true" + - name: S3DISABLE_SSL + value: "true" + - name: RAY_ROLE + value: "Worker" + - name: STORAGE_BACKEND + value: "s3" + - name: RAY_ROOT_DIR + value: "log" + resources: + limits: + cpu: "200m" + memory: "256Mi" + requests: + cpu: "50m" + memory: "64Mi" + volumeMounts: + - name: historyserver + mountPath: *rayTmpRoot + - name: ray-token + mountPath: /var/run/secrets/ray.io/serviceaccount + readOnly: true + volumes: + - name: historyserver + emptyDir: {} +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: raycluster-historyserver-kubernetes-auth +--- +# The Ray Dashboard runs TokenReview/SubjectAccessReview to validate incoming +# tokens, so the cluster's ServiceAccount needs permission to create them. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: ray-authenticator +rules: +- apiGroups: ["authentication.k8s.io"] + resources: + - 'tokenreviews' + verbs: ["create"] +- apiGroups: ["authorization.k8s.io"] + resources: + - 'subjectaccessreviews' + verbs: ["create"] +--- +# Grants the caller ray:write on rayclusters, which is what the Dashboard +# authorizes the presented token against. +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: ray-writer +rules: +- apiGroups: ["ray.io"] + resources: + - 'rayclusters' + verbs: ["ray:write"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: ray-authenticator +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: ray-authenticator +subjects: +- kind: ServiceAccount + name: raycluster-historyserver-kubernetes-auth + namespace: default +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: raycluster-historyserver-kubernetes-auth +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: ray-writer +subjects: +- kind: ServiceAccount + name: raycluster-historyserver-kubernetes-auth + namespace: default diff --git a/historyserver/config/raycluster.yaml b/historyserver/config/raycluster.yaml index fc7fea4d32c..5c899b9e0a2 100644 --- a/historyserver/config/raycluster.yaml +++ b/historyserver/config/raycluster.yaml @@ -7,7 +7,7 @@ metadata: namespace: default spec: # Enable Ray token authentication. Requires Ray 2.52.0+ - # rayVersion: "2.55.0" + # rayVersion: "2.56.0" # authOptions: # mode: token headGroupSpec: @@ -108,6 +108,13 @@ spec: value: "s3" - name: RAY_ROOT_DIR value: "log" + # Uncomment when the RayCluster sets `spec.authOptions.mode: token`; the collector + # needs the token to call the Dashboard. + # - name: RAY_AUTH_TOKEN + # valueFrom: + # secretKeyRef: + # name: raycluster-historyserver # defaults to the RayCluster name + # key: auth_token volumeMounts: - name: historyserver mountPath: *rayTmpRoot @@ -193,6 +200,13 @@ spec: value: "s3" - name: RAY_ROOT_DIR value: "log" + # Uncomment when the RayCluster sets `spec.authOptions.mode: token`; the collector + # needs the token to call the Dashboard. + # - name: RAY_AUTH_TOKEN + # valueFrom: + # secretKeyRef: + # name: raycluster-historyserver # defaults to the RayCluster name + # key: auth_token volumeMounts: - name: historyserver mountPath: *rayTmpRoot diff --git a/historyserver/docs/README.md b/historyserver/docs/README.md index 1180fa49157..d005a9edd90 100644 --- a/historyserver/docs/README.md +++ b/historyserver/docs/README.md @@ -80,6 +80,7 @@ Sample configs are in the `config/` directory: | `azurite.yaml` | Azurite deployment for Azure Blob Storage emulation | | `raycluster.yaml` | Ray cluster with collector sidecar (S3/MinIO) | | `raycluster-azureblob.yaml` | Ray cluster with collector sidecar (Azure Blob) | +| `raycluster-kubernetes-auth.yaml` | Ray cluster with collector sidecar using Kubernetes token authentication (S3/MinIO) | | `rayjob.yaml` | Sample Ray job for testing | | `historyserver.yaml` | History Server deployment (S3/MinIO) | | `historyserver-azureblob.yaml` | History Server deployment (Azure Blob) | diff --git a/historyserver/pkg/collector/logcollector/runtime/logcollector/endpoint_fetch_once.go b/historyserver/pkg/collector/logcollector/runtime/logcollector/endpoint_fetch_once.go index 93d9c92c5f9..2844ea8671c 100644 --- a/historyserver/pkg/collector/logcollector/runtime/logcollector/endpoint_fetch_once.go +++ b/historyserver/pkg/collector/logcollector/runtime/logcollector/endpoint_fetch_once.go @@ -76,6 +76,12 @@ func (r *RayLogHandler) fetchAndStoreEndpoint(cfg endpointFetchConfig) { return } + if err := utils.SetRayAuthHeader(req); err != nil { + cancel() + logrus.Errorf("Failed to authenticate request for %s: %v", cfg.endpoint, err) + return + } + client := r.HttpClient if client == nil { client = http.DefaultClient @@ -105,6 +111,13 @@ func (r *RayLogHandler) fetchAndStoreEndpoint(cfg endpointFetchConfig) { continue } + // Unlike a Dashboard that is still starting up, a rejected credential never resolves on + // its own. Retrying until shutdown would bury the cause under an endless warning loop. + if utils.IsAuthFailure(resp.StatusCode) { + logrus.Errorf("%s returned status %d: the collector is not authenticated against the Ray Dashboard, giving up on this endpoint. Check that token auth is configured for the collector container.", cfg.endpoint, resp.StatusCode) + return + } + if resp.StatusCode != http.StatusOK { logrus.Warnf("%s returned status %d, retrying in %v", cfg.endpoint, resp.StatusCode, retryInterval) if !r.sleepOrShutdown(retryInterval) { diff --git a/historyserver/pkg/collector/logcollector/runtime/logcollector/poll.go b/historyserver/pkg/collector/logcollector/runtime/logcollector/poll.go index 18179444624..ee7765a0c92 100644 --- a/historyserver/pkg/collector/logcollector/runtime/logcollector/poll.go +++ b/historyserver/pkg/collector/logcollector/runtime/logcollector/poll.go @@ -111,6 +111,12 @@ func (r *RayLogHandler) pollSingleEndpoint(endpoint, sessionName string) { return } + if err := utils.SetRayAuthHeader(req); err != nil { + cancel() + logrus.Errorf("Failed to authenticate request for additional endpoint %s: %v", endpoint, err) + return + } + resp, err := r.HttpClient.Do(req) if err != nil { cancel() @@ -126,6 +132,13 @@ func (r *RayLogHandler) pollSingleEndpoint(endpoint, sessionName string) { return } + // An auth failure drops this endpoint's data on every cycle for the lifetime of the Pod, so + // it must not be reported at the same level as a transient non-200. + if utils.IsAuthFailure(resp.StatusCode) { + logrus.Errorf("Additional endpoint %s returned status %d: the collector is not authenticated against the Ray Dashboard, so this endpoint's data will keep being dropped. Check that token auth is configured for the collector container.", endpoint, resp.StatusCode) + return + } + if resp.StatusCode != http.StatusOK { logrus.Warnf("Additional endpoint %s returned status %d", endpoint, resp.StatusCode) return diff --git a/historyserver/pkg/historyserver/router.go b/historyserver/pkg/historyserver/router.go index 28eb9046ffd..3652eb4d86e 100644 --- a/historyserver/pkg/historyserver/router.go +++ b/historyserver/pkg/historyserver/router.go @@ -447,7 +447,7 @@ func (s *ServerHandler) redirectRequest(req *restful.Request, resp *restful.Resp } // In auth-token mode, drop the client-supplied x-ray-authorization so it cannot bypass // the server-managed token injected below. - if s.useAuthTokenMode && strings.EqualFold(key, "x-ray-authorization") { + if s.useAuthTokenMode && strings.EqualFold(key, utils.RayAuthHeader) { continue } // Authorization has to be dropped in two cases. In auth-token mode, Ray only falls back to @@ -472,7 +472,7 @@ func (s *ServerHandler) redirectRequest(req *restful.Request, resp *restful.Resp // so no header is needed; auth-enabled clusters without a token fail before reaching here. if authTokenAttr := req.Attribute(ATTRIBUTE_AUTH_TOKEN); authTokenAttr != nil { if authToken, ok := authTokenAttr.(string); ok && authToken != "" { - proxyReq.Header.Set("x-ray-authorization", fmt.Sprintf("Bearer %s", authToken)) + proxyReq.Header.Set(utils.RayAuthHeader, fmt.Sprintf("Bearer %s", authToken)) } } } diff --git a/historyserver/pkg/utils/auth.go b/historyserver/pkg/utils/auth.go new file mode 100644 index 00000000000..b7896b28060 --- /dev/null +++ b/historyserver/pkg/utils/auth.go @@ -0,0 +1,76 @@ +package utils + +import ( + "errors" + "fmt" + "net/http" + "os" + "strconv" + "strings" +) + +const ( + RayTokenMountPath = "/var/run/secrets/ray.io/serviceaccount" + RAY_ENABLE_K8S_TOKEN_AUTH_ENV_VAR = "RAY_ENABLE_K8S_TOKEN_AUTH" + RAY_AUTH_TOKEN_ENV_VAR = "RAY_AUTH_TOKEN" + RAY_AUTH_MODE_ENV_VAR = "RAY_AUTH_MODE" + + RAY_AUTH_TOKEN_SECRET_KEY = "auth_token" + RayAuthModeToken = "token" + RayAuthHeader = "x-ray-authorization" + RayK8sTokenPath = RayTokenMountPath + "/token" +) + +// errRayAuthConfig marks a failure to resolve the collector's own Dashboard +// credential, as opposed to the Dashboard rejecting a credential we did send. +var errRayAuthConfig = errors.New("ray dashboard credential is misconfigured") + +func isRayAuthEnabled() bool { + return strings.EqualFold(os.Getenv(RAY_AUTH_MODE_ENV_VAR), RayAuthModeToken) +} + +func isK8sTokenAuthEnabled() bool { + enabled, err := strconv.ParseBool(os.Getenv(RAY_ENABLE_K8S_TOKEN_AUTH_ENV_VAR)) + return err == nil && enabled +} + +func staticAuthToken() string { + return strings.TrimSpace(os.Getenv(RAY_AUTH_TOKEN_ENV_VAR)) +} + +func getRayAuthToken() (string, error) { + switch { + case isK8sTokenAuthEnabled(): + raw, err := os.ReadFile(RayK8sTokenPath) + if err != nil { + return "", fmt.Errorf("%w: failed to read projected ServiceAccount token at %s: %w", errRayAuthConfig, RayK8sTokenPath, err) + } + token := strings.TrimSpace(string(raw)) + if token == "" { + return "", fmt.Errorf("%w: projected ServiceAccount token at %s is empty", errRayAuthConfig, RayK8sTokenPath) + } + return token, nil + case staticAuthToken() != "": + return staticAuthToken(), nil + case isRayAuthEnabled(): + return "", fmt.Errorf("%w: %s=%s but no credential is available: set %s, or %s=true with the projected token volume mounted at %s", + errRayAuthConfig, RAY_AUTH_MODE_ENV_VAR, RayAuthModeToken, RAY_AUTH_TOKEN_ENV_VAR, RAY_ENABLE_K8S_TOKEN_AUTH_ENV_VAR, RayTokenMountPath) + default: + return "", nil + } +} + +func SetRayAuthHeader(req *http.Request) error { + token, err := getRayAuthToken() + if err != nil { + return err + } + if token != "" { + req.Header.Set(RayAuthHeader, "Bearer "+token) + } + return nil +} + +func IsAuthFailure(statusCode int) bool { + return statusCode == http.StatusUnauthorized || statusCode == http.StatusForbidden +} diff --git a/historyserver/pkg/utils/auth_test.go b/historyserver/pkg/utils/auth_test.go new file mode 100644 index 00000000000..cc674623108 --- /dev/null +++ b/historyserver/pkg/utils/auth_test.go @@ -0,0 +1,125 @@ +package utils + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// clearAuthEnv makes every auth-related env var explicitly empty so tests do not +// depend on the environment they run in. +func clearAuthEnv(t *testing.T) { + t.Helper() + t.Setenv(RAY_AUTH_MODE_ENV_VAR, "") + t.Setenv(RAY_AUTH_TOKEN_ENV_VAR, "") + t.Setenv(RAY_ENABLE_K8S_TOKEN_AUTH_ENV_VAR, "") +} + +func TestIsRayAuthEnabled(t *testing.T) { + tests := []struct { + scenario string + authMode string + expected bool + }{ + {scenario: "unset", authMode: "", expected: false}, + {scenario: "token mode", authMode: "token", expected: true}, + {scenario: "token mode is case insensitive", authMode: "Token", expected: true}, + {scenario: "unknown mode", authMode: "disabled", expected: false}, + } + + for _, tc := range tests { + t.Run(tc.scenario, func(t *testing.T) { + clearAuthEnv(t) + t.Setenv(RAY_AUTH_MODE_ENV_VAR, tc.authMode) + assert.Equal(t, tc.expected, isRayAuthEnabled()) + }) + } +} + +func TestIsK8sTokenAuthEnabled(t *testing.T) { + tests := []struct { + scenario string + value string + expected bool + }{ + {scenario: "unset", value: "", expected: false}, + {scenario: "true", value: "true", expected: true}, + {scenario: "1", value: "1", expected: true}, + {scenario: "TRUE", value: "TRUE", expected: true}, + {scenario: "false", value: "false", expected: false}, + {scenario: "not a bool", value: "yes", expected: false}, + } + + for _, tc := range tests { + t.Run(tc.scenario, func(t *testing.T) { + clearAuthEnv(t) + t.Setenv(RAY_ENABLE_K8S_TOKEN_AUTH_ENV_VAR, tc.value) + assert.Equal(t, tc.expected, isK8sTokenAuthEnabled()) + }) + } +} + +func TestGetRayAuthToken(t *testing.T) { + t.Run("auth disabled returns no token", func(t *testing.T) { + clearAuthEnv(t) + token, err := getRayAuthToken() + require.NoError(t, err) + assert.Empty(t, token) + }) + + t.Run("static token is trimmed", func(t *testing.T) { + clearAuthEnv(t) + t.Setenv(RAY_AUTH_MODE_ENV_VAR, RayAuthModeToken) + t.Setenv(RAY_AUTH_TOKEN_ENV_VAR, " secret\n") + token, err := getRayAuthToken() + require.NoError(t, err) + assert.Equal(t, "secret", token) + }) + + t.Run("static token is used even without token auth mode", func(t *testing.T) { + clearAuthEnv(t) + t.Setenv(RAY_AUTH_TOKEN_ENV_VAR, "secret") + token, err := getRayAuthToken() + require.NoError(t, err) + assert.Equal(t, "secret", token) + }) + + t.Run("token auth mode without any credential fails", func(t *testing.T) { + clearAuthEnv(t) + t.Setenv(RAY_AUTH_MODE_ENV_VAR, RayAuthModeToken) + _, err := getRayAuthToken() + require.Error(t, err) + require.ErrorIs(t, err, errRayAuthConfig) + assert.Contains(t, err.Error(), "no credential is available") + }) + + t.Run("missing projected token fails", func(t *testing.T) { + clearAuthEnv(t) + t.Setenv(RAY_AUTH_MODE_ENV_VAR, RayAuthModeToken) + t.Setenv(RAY_ENABLE_K8S_TOKEN_AUTH_ENV_VAR, "true") + _, err := getRayAuthToken() + require.Error(t, err) + require.ErrorIs(t, err, errRayAuthConfig) + assert.Contains(t, err.Error(), RayK8sTokenPath) + }) +} + +func TestSetRayAuthHeader(t *testing.T) { + t.Run("sets bearer header when a token exists", func(t *testing.T) { + clearAuthEnv(t) + t.Setenv(RAY_AUTH_TOKEN_ENV_VAR, "secret") + req := httptest.NewRequest(http.MethodGet, "http://example.com", nil) + require.NoError(t, SetRayAuthHeader(req)) + assert.Equal(t, "Bearer secret", req.Header.Get(RayAuthHeader)) + }) + + t.Run("leaves header unset when auth is disabled", func(t *testing.T) { + clearAuthEnv(t) + req := httptest.NewRequest(http.MethodGet, "http://example.com", nil) + require.NoError(t, SetRayAuthHeader(req)) + assert.Empty(t, req.Header.Get(RayAuthHeader)) + }) +} diff --git a/historyserver/pkg/utils/constant.go b/historyserver/pkg/utils/constant.go index 28eee8582e1..4a38d4074b2 100644 --- a/historyserver/pkg/utils/constant.go +++ b/historyserver/pkg/utils/constant.go @@ -14,8 +14,6 @@ const ( RayServiceKind = "rayservice" RayClusterKind = "raycluster" - RAY_AUTH_TOKEN_SECRET_KEY = "auth_token" - // RayContainerIndex is the index of the Ray container in the head pod template. RayContainerIndex = 0 // DashboardPortName is the name the ray-operator gives the dashboard port. diff --git a/historyserver/pkg/utils/utils.go b/historyserver/pkg/utils/utils.go index 1885af85d4f..9cd3cd011d5 100644 --- a/historyserver/pkg/utils/utils.go +++ b/historyserver/pkg/utils/utils.go @@ -30,8 +30,9 @@ const ( ) var ( - sessionIDRegex = regexp.MustCompile(SESSION_ID_REGEX) - hexRegex = regexp.MustCompile(HEX_REGEX) + sessionIDRegex = regexp.MustCompile(SESSION_ID_REGEX) + hexRegex = regexp.MustCompile(HEX_REGEX) + errDashboardAuthFailed = errors.New("ray dashboard rejected the request: authentication failed") ) // EndpointPathToStorageKey converts a Ray Dashboard API endpoint path to a @@ -232,11 +233,22 @@ func FetchCurrentNodeID() (string, error) { endpoint := fmt.Sprintf("%s%s/api/v0/nodes?limit=10000", scheme, strings.TrimRight(addr, "/")) client := &http.Client{Timeout: 1 * time.Second} - resp, err := client.Get(endpoint) + req, err := http.NewRequest(http.MethodGet, endpoint, nil) + if err != nil { + return "", err + } + if err := SetRayAuthHeader(req); err != nil { + return "", fmt.Errorf("failed to authenticate request to %s: %w", endpoint, err) + } + + resp, err := client.Do(req) if err != nil { return "", err } defer resp.Body.Close() + if IsAuthFailure(resp.StatusCode) { + return "", fmt.Errorf("%w: HTTP status %d from endpoint %s", errDashboardAuthFailed, resp.StatusCode, endpoint) + } if resp.StatusCode != http.StatusOK { return "", fmt.Errorf("HTTP status %d from endpoint %s", resp.StatusCode, endpoint) } @@ -261,9 +273,11 @@ func GetNodeRayIDWithFQIP() (string, error) { if err == nil { return nodeID, nil } - if errors.Is(err, ErrPodIPNotSet) || errors.Is(err, ErrFQRayIPNotSet) { + if errors.Is(err, ErrPodIPNotSet) || errors.Is(err, ErrFQRayIPNotSet) || + errors.Is(err, errDashboardAuthFailed) || errors.Is(err, errRayAuthConfig) { return "", err } + logrus.Warnf("Attempt %d/12 to discover Ray NodeID failed: %v, retrying in 5s", i+1, err) lastErr = err time.Sleep(5 * time.Second) } diff --git a/historyserver/pkg/utils/utils_test.go b/historyserver/pkg/utils/utils_test.go index 17a12ac8aff..381b9f37691 100644 --- a/historyserver/pkg/utils/utils_test.go +++ b/historyserver/pkg/utils/utils_test.go @@ -9,6 +9,9 @@ import ( "strings" "testing" "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestSlice(t *testing.T) { @@ -171,6 +174,33 @@ func TestGetNodeRayIDWithFQIP_EnvVars(t *testing.T) { } } +// A credential that cannot be resolved never becomes valid, so discovery must +// surface it immediately instead of spending the full 60s retry budget and then +// reporting a misleading timeout. +func TestGetNodeRayIDWithFQIP_FailsFastOnAuthConfigError(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"data":{"result":{"result":[{"node_id":"node-id-abc","node_ip":"10.0.0.1","state":"ALIVE"}]}}}`)) + })) + defer ts.Close() + + t.Setenv("POD_IP", "10.0.0.1") + t.Setenv("FQ_RAY_IP", ts.URL) + // Token auth requested, but neither RAY_AUTH_TOKEN nor the projected token volume is available. + t.Setenv(RAY_AUTH_MODE_ENV_VAR, RayAuthModeToken) + t.Setenv(RAY_AUTH_TOKEN_ENV_VAR, "") + t.Setenv(RAY_ENABLE_K8S_TOKEN_AUTH_ENV_VAR, "") + + start := time.Now() + _, err := GetNodeRayIDWithFQIP() + elapsed := time.Since(start) + + require.Error(t, err) + require.ErrorIs(t, err, errRayAuthConfig) + assert.NotContains(t, err.Error(), "timeout", "misconfiguration must not be reported as a discovery timeout") + assert.Less(t, elapsed, 5*time.Second, "must not enter the retry loop") +} + func TestGetSessionDir_Symlinks(t *testing.T) { tmpDir, err := os.MkdirTemp("", "s-*") if err != nil { diff --git a/historyserver/test/e2e/collector_test.go b/historyserver/test/e2e/collector_test.go index 26d24f810d0..8f79f1e8da1 100644 --- a/historyserver/test/e2e/collector_test.go +++ b/historyserver/test/e2e/collector_test.go @@ -80,6 +80,10 @@ func TestCollector(t *testing.T) { name: "Timezone: collector should fetch and store timezone endpoint data once on startup", testFunc: testCollectorStoresTimezone, }, + { + name: "Token auth: collector should authenticate against the Ray Dashboard instead of crash-looping", + testFunc: testCollectorWithTokenAuth, + }, } for _, tt := range tests { @@ -386,13 +390,17 @@ func testCollectorStoresClusterMetadata(test Test, g *WithT, namespace *corev1.N // // The test case follows these steps: // 1. Prepare test environment by applying a Ray cluster with the collector -// 2. Get the sessionID from the head pod to build the expected S3 key -// 3. Wait for the timezone file to appear in S3 at {sessionName}/fetched_endpoints/restful__timezone -// 4. Read the file and verify it contains valid JSON with the expected schema (offset, value) -// 5. Delete S3 bucket to ensure test isolation +// 2. Assert the timezone data reaches S3 with the expected key and schema +// 3. Delete S3 bucket to ensure test isolation func testCollectorStoresTimezone(test Test, g *WithT, namespace *corev1.Namespace, s3Client *s3.S3) { rayCluster := PrepareTestEnv(test, g, namespace, s3Client) + assertTimezoneStored(test, g, rayCluster, s3Client) + + DeleteS3Bucket(test, g, s3Client) +} + +func assertTimezoneStored(test Test, g *WithT, rayCluster *rayv1.RayCluster, s3Client *s3.S3) { sessionID := GetSessionIDFromHeadPod(test, g, rayCluster) storageKey := utils.EndpointPathToStorageKey(EndpointTimezone) sessionDir := clusterlogs.SessionDir("log", "", "", rayCluster.Namespace, rayCluster.Name, sessionID) @@ -428,8 +436,6 @@ func testCollectorStoresTimezone(test Test, g *WithT, namespace *corev1.Namespac }, TestTimeoutMedium).Should(Succeed()) LogWithTimestamp(test.T(), "Timezone data stored successfully: %s", string(timezoneBody)) - - DeleteS3Bucket(test, g, s3Client) } // testCollectorStoresPlacementGroups verifies that the Head collector periodically polls @@ -649,3 +655,31 @@ func assertAllEventTypesCovered(test Test, g Gomega, events []rayEvent) { g.Expect(foundEventTypes[string(eventType)]).To(BeTrue(), "Event type %s not found", eventType) } } + +func testCollectorWithTokenAuth(test Test, g *WithT, namespace *corev1.Namespace, s3Client *s3.S3) { + rayCluster := ApplyRayClusterWithCollectorTokenAuth(test, g, namespace) + + 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` + stdout, stderr := ExecPodCmd(test, headPod, "ray-head", []string{"sh", "-c", probe}) + g.Expect(stderr.String()).To(BeEmpty()) + g.Expect(strings.TrimSpace(stdout.String())).To(Equal("401")) + + // Data in storage is the real proof: the collector only writes the timezone object after an + // authenticated Dashboard call succeeds. + assertTimezoneStored(test, g, rayCluster, s3Client) + + DeleteS3Bucket(test, g, s3Client) +} diff --git a/historyserver/test/support/raycluster.go b/historyserver/test/support/raycluster.go index dac77e17aed..54d90af345e 100644 --- a/historyserver/test/support/raycluster.go +++ b/historyserver/test/support/raycluster.go @@ -13,26 +13,53 @@ import ( . "github.com/ray-project/kuberay/ray-operator/test/support" ) -const RayClusterManifestPath = "../../config/raycluster.yaml" +const ( + RayClusterManifestPath = "../../config/raycluster.yaml" + + // RayVersionForTokenAuth is the minimum rayVersion the operator accepts for token auth. + RayVersionForTokenAuth = "2.52.0" +) // ApplyRayClusterWithCollectorWithEnvs deploys a Ray cluster with the collector sidecar into the test namespace, // adding the specified environment variables to the head pod. func ApplyRayClusterWithCollectorWithEnvs(test Test, g *WithT, namespace *corev1.Namespace, envs map[string]string) *rayv1.RayCluster { - rayClusterFromYaml := DeserializeRayClusterYAML(test, RayClusterManifestPath) - rayClusterFromYaml.Namespace = namespace.Name + return applyRayClusterWithCollector(test, g, namespace, func(rayCluster *rayv1.RayCluster) { + headContainer := &rayCluster.Spec.HeadGroupSpec.Template.Spec.Containers[utils.RayContainerIndex] + if len(headContainer.Env) == 0 { + headContainer.Env = []corev1.EnvVar{} + } - headContainer := &rayClusterFromYaml.Spec.HeadGroupSpec.Template.Spec.Containers[utils.RayContainerIndex] - if len(headContainer.Env) == 0 { - headContainer.Env = []corev1.EnvVar{} - } + for key, value := range envs { + env := corev1.EnvVar{ + Name: key, + Value: value, + } + headContainer.Env = append(headContainer.Env, env) + } + }) +} - for key, value := range envs { - env := corev1.EnvVar{ - Name: key, - Value: value, +// ApplyRayClusterWithCollectorTokenAuth deploys a Ray cluster whose Dashboard requires token +// authentication, and wires the collector sidecars to the auth token. +func ApplyRayClusterWithCollectorTokenAuth(test Test, g *WithT, namespace *corev1.Namespace) *rayv1.RayCluster { + return applyRayClusterWithCollector(test, g, namespace, func(rayCluster *rayv1.RayCluster) { + rayCluster.Spec.RayVersion = RayVersionForTokenAuth + rayCluster.Spec.AuthOptions = &rayv1.AuthOptions{Mode: rayv1.AuthModeToken} + + // reconcileAuthSecret generates this Secret when authOptions.secretName is unset. + secretName := utils.CheckName(rayCluster.Name) + injectCollectorAuthToken(rayCluster.Spec.HeadGroupSpec.Template.Spec.Containers, secretName) + for wg := range rayCluster.Spec.WorkerGroupSpecs { + injectCollectorAuthToken(rayCluster.Spec.WorkerGroupSpecs[wg].Template.Spec.Containers, secretName) } - headContainer.Env = append(headContainer.Env, env) - } + }) +} + +func applyRayClusterWithCollector(test Test, g *WithT, namespace *corev1.Namespace, mutate func(*rayv1.RayCluster)) *rayv1.RayCluster { + rayClusterFromYaml := DeserializeRayClusterYAML(test, RayClusterManifestPath) + rayClusterFromYaml.Namespace = namespace.Name + + mutate(rayClusterFromYaml) // Inject namespace name as the ray-cluster-namespace for head group collector injectCollectorRayClusterNamespaceAndEnvVar(rayClusterFromYaml.Spec.HeadGroupSpec.Template.Spec.Containers, rayClusterFromYaml.Name, namespace.Name) @@ -78,6 +105,21 @@ func injectCollectorRayClusterNamespaceAndEnvVar(containers []corev1.Container, } } +// injectCollectorAuthToken points every collector sidecar at the auth token in the given Secret. +func injectCollectorAuthToken(containers []corev1.Container, secretName string) { + for i := range containers { + if containers[i].Name != "collector" { + continue + } + setOrAppendEnv(&containers[i], utils.RAY_AUTH_TOKEN_ENV_VAR, "", &corev1.EnvVarSource{ + SecretKeyRef: &corev1.SecretKeySelector{ + LocalObjectReference: corev1.LocalObjectReference{Name: secretName}, + Key: utils.RAY_AUTH_TOKEN_SECRET_KEY, + }, + }) + } +} + // setOrAppendEnv updates an environment variable in-place if it already exists (e.g., from static YAML manifests) // or appends a new entry if missing. This prevents duplicate environment variable entries in the pod spec. func setOrAppendEnv(container *corev1.Container, name string, val string, valFrom *corev1.EnvVarSource) { diff --git a/ray-operator/controllers/ray/common/pod.go b/ray-operator/controllers/ray/common/pod.go index 21d5d459afd..679775f3f2c 100644 --- a/ray-operator/controllers/ray/common/pod.go +++ b/ray-operator/controllers/ray/common/pod.go @@ -304,6 +304,13 @@ func DefaultHeadPodTemplate(ctx context.Context, instance rayv1.RayCluster, head if features.Enabled(features.RayClusterHistoryServer) && instance.Spec.HistoryServerOptions != nil && instance.Spec.HistoryServerOptions.CollectorOptions != nil { fqdnRayIP := utils.GenerateFQDNServiceName(ctx, instance, instance.Namespace) collectorContainer := BuildCollectorContainer(instance.Spec.HistoryServerOptions.CollectorOptions, rayv1.HeadNode, instance.Name, instance.Namespace, fqdnRayIP, instance.Labels) + + // 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) + } + podTemplate.Spec.Containers = append(podTemplate.Spec.Containers, collectorContainer) } @@ -700,6 +707,11 @@ func DefaultWorkerPodTemplate(ctx context.Context, instance rayv1.RayCluster, wo if features.Enabled(features.RayClusterHistoryServer) && instance.Spec.HistoryServerOptions != nil && instance.Spec.HistoryServerOptions.CollectorOptions != nil { collectorContainer := BuildCollectorContainer(instance.Spec.HistoryServerOptions.CollectorOptions, rayv1.WorkerNode, instance.Name, instance.Namespace, fqdnRayIP, instance.Labels) + + if utils.IsAuthEnabled(&instance.Spec) { + SetContainerTokenAuthEnvVars(instance.Name, &collectorContainer, instance.Spec.AuthOptions) + } + podTemplate.Spec.Containers = append(podTemplate.Spec.Containers, collectorContainer) }