[Feature] [history server] Support token auth in the history server collector

historyserver/config/raycluster-azureblob.yaml

@@ -6,6 +6,12 @@ metadata:
   name: raycluster-historyserver
   namespace: default
 spec:
+  # Uncomment to make the Dashboard reject unauthenticated requests, which the collector
+  # sidecars below must survive (ray-project/kuberay#5056). The operator rejects
+  # `authOptions.mode: token` unless `rayVersion` is set and is at least 2.52.0.
+  # rayVersion: "2.52.0"
+  # authOptions:
+  #   mode: token
   headGroupSpec:
     rayStartParams:
       dashboard-host: 0.0.0.0
@@ -65,6 +71,17 @@ 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
+          # Ray token auth (optional): uncomment when the RayCluster sets
+          # `spec.authOptions.mode: token` (which also requires `spec.rayVersion`, unset here).
+          # Without the token the collector gets 401 from the Dashboard during NodeID discovery.
+          # - name: RAY_AUTH_TOKEN
+          #   valueFrom:
+          #     secretKeyRef:
+          #       name: raycluster-historyserver  # defaults to the RayCluster name
+          #       key: auth_token
+          # For `spec.authOptions.enableK8sTokenAuth` (Ray 2.55.0+) set
+          # RAY_ENABLE_K8S_TOKEN_AUTH="true" instead, plus a projected ServiceAccount token
+          # volume at /var/run/secrets/ray.io/serviceaccount.
           command: [collector, --role=Head, --runtime-class-name=azureblob, --ray-cluster-name=raycluster-historyserver, --ray-root-dir=log, --events-port=8084]
           volumeMounts:
           - name: historyserver
@@ -136,6 +153,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
+          # Ray token auth (optional): workers query the Dashboard through FQ_RAY_IP for their
+          # Ray NodeID, so they need the same token as the head collector.
+          # - name: RAY_AUTH_TOKEN
+          #   valueFrom:
+          #     secretKeyRef:
+          #       name: raycluster-historyserver  # defaults to the RayCluster name
+          #       key: auth_token
           command: [collector, --role=Worker, --runtime-class-name=azureblob, --ray-cluster-name=raycluster-historyserver, --ray-root-dir=log, --events-port=8084]
           volumeMounts:
           - name: historyserver

historyserver/config/raycluster-gcs.yaml

@@ -6,6 +6,12 @@ metadata:
   name: raycluster-historyserver
   namespace: default
 spec:
+  # Uncomment to make the Dashboard reject unauthenticated requests, which the collector
+  # sidecars below must survive (ray-project/kuberay#5056). The operator rejects
+  # `authOptions.mode: token` unless `rayVersion` is set and is at least 2.52.0.
+  # rayVersion: "2.52.0"
+  # authOptions:
+  #   mode: token
   headGroupSpec:
     rayStartParams:
       dashboard-host: 0.0.0.0
@@ -84,6 +90,17 @@ spec:
           # endpoints above. Accepts Go duration format (e.g., "30s", "1m", "5m").
           - name: RAY_COLLECTOR_POLL_INTERVAL
             value: "30s"
+          # Ray token auth (optional): uncomment when the RayCluster sets
+          # `spec.authOptions.mode: token` (which also requires `spec.rayVersion`, unset here).
+          # Without the token the collector gets 401 from the Dashboard during NodeID discovery.
+          # - name: RAY_AUTH_TOKEN
+          #   valueFrom:
+          #     secretKeyRef:
+          #       name: ${RAY_CLUSTER}  # defaults to the RayCluster name
+          #       key: auth_token
+          # For `spec.authOptions.enableK8sTokenAuth` (Ray 2.55.0+) set
+          # RAY_ENABLE_K8S_TOKEN_AUTH="true" instead, plus a projected ServiceAccount token
+          # volume at /var/run/secrets/ray.io/serviceaccount.
           command:
           - collector
           - --role=Head
@@ -158,6 +175,13 @@ spec:
             value: *rayTmpRoot
           - name: GCS_BUCKET
             value: "${GCS_BUCKET}"
+          # Ray token auth (optional): workers query the Dashboard through FQ_RAY_IP for their
+          # Ray NodeID, so they need the same token as the head collector.
+          # - name: RAY_AUTH_TOKEN
+          #   valueFrom:
+          #     secretKeyRef:
+          #       name: ${RAY_CLUSTER}  # defaults to the RayCluster name
+          #       key: auth_token
           command:
           - collector
           - --role=Worker

historyserver/config/raycluster-k8s-auth.yaml

@@ -0,0 +1,173 @@
+apiVersion: ray.io/v1
+kind: RayCluster
+metadata:
+  labels:
+    ray.io/cluster: raycluster-historyserver-k8sauth
+  name: raycluster-historyserver-k8sauth
+spec:
+  # enableK8sTokenAuth requires Ray 2.55.0+. authOptions.secretName must NOT be
+  # set at the same time.
+  rayVersion: '2.55.0'
+  authOptions:
+    mode: 'token'
+    enableK8sTokenAuth: true
+  historyServerOptions:
+    collectorOptions:
+      image: collector:v0.1.0
+      imagePullPolicy: IfNotPresent
+      resources:
+        limits:
+          cpu: "200m"
+          memory: "256Mi"
+        requests:
+          cpu: "50m"
+          memory: "64Mi"
+      env:
+      - name: STORAGE_BACKEND
+        value: "s3"
+      - name: RAY_ROOT_DIR
+        value: "log"
+      - name: S3DISABLE_SSL
+        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"
+  headGroupSpec:
+    rayStartParams:
+      dashboard-host: 0.0.0.0
+      num-cpus: "0"
+    template:
+      spec:
+        serviceAccountName: raycluster-historyserver-k8sauth
+        containers:
+        - name: ray-head
+          imagePullPolicy: IfNotPresent
+          image: rayproject/ray:2.55.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_PUBLISHER_HTTP_ENDPOINT_EXPOSABLE_EVENT_TYPES
+            value: &eventTypes "TASK_DEFINITION_EVENT,TASK_LIFECYCLE_EVENT,ACTOR_TASK_DEFINITION_EVENT,
+                    TASK_PROFILE_EVENT,DRIVER_JOB_DEFINITION_EVENT,DRIVER_JOB_LIFECYCLE_EVENT,
+                    ACTOR_DEFINITION_EVENT,ACTOR_LIFECYCLE_EVENT,NODE_DEFINITION_EVENT,NODE_LIFECYCLE_EVENT"
+          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
+  workerGroupSpecs:
+  - replicas: 1
+    minReplicas: 1
+    maxReplicas: 5
+    groupName: workergroup
+    rayStartParams: {}
+    template:
+      spec:
+        serviceAccountName: raycluster-historyserver-k8sauth
+        containers:
+        - name: ray-worker
+          imagePullPolicy: IfNotPresent
+          image: rayproject/ray:2.55.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_PUBLISHER_HTTP_ENDPOINT_EXPOSABLE_EVENT_TYPES
+            value: *eventTypes
+          resources:
+            limits:
+              cpu: "2"
+              memory: "2Gi"
+            requests:
+              cpu: "50m"
+              memory: "1Gi"
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: raycluster-historyserver-k8sauth
+---
+# 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-k8sauth
+  namespace: default
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: raycluster-historyserver-k8sauth
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: ray-writer
+subjects:
+- kind: ServiceAccount
+  name: raycluster-historyserver-k8sauth
+  namespace: default

historyserver/config/raycluster.yaml

@@ -103,6 +103,19 @@ spec:
             value: "test"
           - name: S3FORCE_PATH_STYLE
             value: "true"
+          # Ray token auth (optional): uncomment when the RayCluster sets
+          # `spec.authOptions.mode: token` (which also requires `spec.rayVersion`, commented out
+          # above). Without the token the collector gets 401 from the Dashboard during NodeID
+          # discovery and crash-loops. The operator generates this Secret from the RayCluster name
+          # when `authOptions.secretName` is unset.
+          # - name: RAY_AUTH_TOKEN
+          #   valueFrom:
+          #     secretKeyRef:
+          #       name: raycluster-historyserver  # defaults to the RayCluster name
+          #       key: auth_token
+          # For `spec.authOptions.enableK8sTokenAuth` (Ray 2.55.0+) set
+          # RAY_ENABLE_K8S_TOKEN_AUTH="true" instead, plus a projected ServiceAccount token
+          # volume at /var/run/secrets/ray.io/serviceaccount.
           command:
           - collector
           - --role=Head
@@ -193,6 +206,13 @@ spec:
             value: "true"
           - name: S3DISABLE_SSL
             value: "true"
+          # Ray token auth (optional): workers query the Dashboard through FQ_RAY_IP for their
+          # Ray NodeID, so they need the same token as the head collector.
+          # - name: RAY_AUTH_TOKEN
+          #   valueFrom:
+          #     secretKeyRef:
+          #       name: raycluster-historyserver  # defaults to the RayCluster name
+          #       key: auth_token
           command:
           - collector
           - --role=Worker

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) {

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