Skip to content
25 changes: 16 additions & 9 deletions historyserver/cmd/collector/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,8 @@ func main() {
}
}

// RAY_COLLECTOR_ADDITIONAL_ENDPOINTS is optional: the head collector always
// polls its built-in endpoints, and anything listed here is polled on top.
var additionalEndpoints []string
if epStr := os.Getenv("RAY_COLLECTOR_ADDITIONAL_ENDPOINTS"); epStr != "" {
for _, ep := range strings.Split(epStr, ",") {
Expand All @@ -170,16 +172,15 @@ func main() {
}
}

// Fall back instead of exiting: crash-looping this sidecar would take the head pod
// out of its Service endpoints.
Comment on lines +175 to +176

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: maybe move this to line 114 above, as this is the behavior of all env var parsing logic

https://github.com/ray-project/kuberay/pull/5093/changes#diff-49caa92ffb620cd5fa29b0eaa2bd82362831df957085fc12520b1d4ec23f88adR114

@Future-Outlier Future-Outlier Aug 6, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Env parsing isn't all in one place — RAY_COLLECTOR_EVENT_* (131-155), ADDITIONAL_ENDPOINTS (166), STORAGE_BACKEND (197) and RAY_DASHBOARD_ADDRESS (235) all come after 113. The block ending at 113 is the core flags validateFlags consumes, while this one pairs with ADDITIONAL_ENDPOINTS (both endpoint-polling knobs, adjacent config fields).

Keeping it here, but happy to move all the trailing blocks in a separate cleanup if you prefer.

endpointPollInterval := 30 * time.Second
if intervalStr := os.Getenv("RAY_COLLECTOR_POLL_INTERVAL"); intervalStr != "" {
parsed, parseErr := time.ParseDuration(intervalStr)
if parseErr != nil {
logrus.Fatalf("Failed to parse RAY_COLLECTOR_POLL_INTERVAL: %v", parseErr)
}
if parsed <= 0 {
logrus.Fatalf("RAY_COLLECTOR_POLL_INTERVAL must be positive, got: %s", intervalStr)
if v := os.Getenv("RAY_COLLECTOR_POLL_INTERVAL"); v != "" {
if parsed, err := time.ParseDuration(v); err == nil && parsed > 0 {
endpointPollInterval = parsed
} else {
logrus.Warnf("Invalid RAY_COLLECTOR_POLL_INTERVAL=%s, using default %s", v, endpointPollInterval)
}
endpointPollInterval = parsed
}
Comment thread
win5923 marked this conversation as resolved.

jsonData := make(map[string]interface{})
Expand Down Expand Up @@ -229,6 +230,12 @@ func main() {

sessionName := path.Base(activeSessionDir)

// The head collector shares a pod with the dashboard; override for non-default ports.
dashboardAddress := "http://localhost:8265"
if v := os.Getenv("RAY_DASHBOARD_ADDRESS"); v != "" {
dashboardAddress = v
}

globalConfig := types.RayCollectorConfig{
RootDir: rayRootDir,
SessionDir: activeSessionDir,
Expand All @@ -238,7 +245,7 @@ func main() {
RayClusterNamespace: rayClusterNamespace,
PushInterval: pushInterval,
LogBatching: logBatching,
DashboardAddress: os.Getenv("RAY_DASHBOARD_ADDRESS"),
DashboardAddress: dashboardAddress,
OwnerKind: ownerKind,
OwnerName: ownerName,

Expand Down
116 changes: 116 additions & 0 deletions historyserver/config/ray-data.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
apiVersion: ray.io/v1
kind: RayJob
metadata:
name: rayjob-ray-data
spec:
# Self-contained: brings up its own cluster; also exercises the collector's shutdown path.
shutdownAfterJobFinishes: true
# Long enough for one polling cycle after the job succeeds.
ttlSecondsAfterFinished: 30
entrypoint: |
python -c "
import ray
ray.init()

# materialize() is required: an unexecuted Dataset produces no stats.
ds = ray.data.range(100).map_batches(lambda batch: batch).materialize()
print(f'Dataset rows: {ds.count()}')
"
rayClusterSpec:
# Head-only: worker collectors would need the generated head Service FQDN in FQ_RAY_IP.
headGroupSpec:
rayStartParams:
dashboard-host: 0.0.0.0
serviceType: ClusterIP
template:
spec:
containers:
- 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"
# in ray 2.52.0, we need to set RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES
# in ray 2.53.0 (noy yet done). we need to set RAY_DASHBOARD_AGGREGATOR_AGENT_PUBLISHER_HTTP_ENDPOINT_EXPOSABLE_EVENT_TYPES
- name: RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES
value: "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"
Comment on lines +40 to +42

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's update the ray image version to 2.56.0 and use "ALL" here

image: rayproject/ray:2.52.0
imagePullPolicy: IfNotPresent
name: ray-head
securityContext:
allowPrivilegeEscalation: true
privileged: true
resources:
limits:
cpu: "5"
memory: 10G
requests:
cpu: "50m"
memory: 1G
volumeMounts:
- name: historyserver
mountPath: *rayTmpRoot
- name: collector
image: collector:v0.1.0
imagePullPolicy: IfNotPresent
env:
- name: POD_IP
valueFrom:
fieldRef:
fieldPath: status.podIP
# KubeRay generates the cluster name; read it back from the pod label.
- name: RAY_CLUSTER_NAME
valueFrom:
fieldRef:
fieldPath: metadata.labels['ray.io/cluster']
- name: RAY_CLUSTER_NAMESPACE
valueFrom:
fieldRef:
fieldPath: metadata.namespace
# ray.io/originated-from-* labels are on the RayCluster, not the pod; must match metadata.name.
- name: OWNER_KIND
value: "RayJob"
- name: OWNER_NAME
value: "rayjob-ray-data"
# Only used to look up this pod's Ray NodeID; the dashboard is in this pod.
- name: FQ_RAY_IP
value: "localhost"
- name: RAY_TMP_ROOT
value: *rayTmpRoot
# Shorter than the 30s default so a cycle fits inside ttlSecondsAfterFinished.
- name: RAY_COLLECTOR_POLL_INTERVAL
value: "5s"
- 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"
command:
- collector
- --role=Head
- --runtime-class-name=s3
- --ray-root-dir=log
- --events-port=8084
volumeMounts:
- name: historyserver
mountPath: *rayTmpRoot
volumes:
- name: historyserver
emptyDir: {}
10 changes: 10 additions & 0 deletions historyserver/config/raycluster-azureblob.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,16 @@ spec:
value: raycluster-historyserver-head-svc.default.svc.cluster.local
- name: RAY_TMP_ROOT
value: *rayTmpRoot
# Optional; defaults to http://localhost:8265 (head only).
# - name: RAY_DASHBOARD_ADDRESS
# value: "http://localhost:9265"
# Optional; defaults to 30s.
# - name: RAY_COLLECTOR_POLL_INTERVAL
# value: "1m"
# Optional extras on top of the built-in endpoints (Serve, placement groups, Ray Data).
# Paths must match the frontend request URI, query string included.
# - name: RAY_COLLECTOR_ADDITIONAL_ENDPOINTS
# value: "/nodes?view=summary"
Comment on lines +69 to +72

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm curious why the sample value is /nodes?view=summary. Only endpoints under /api/** that don't have dedicated handlers can be replayed.

ws.Path("/api").Consumes(restful.MIME_JSON).Produces(restful.MIME_JSON).Filter(RequestLogFilter) //.Filter(s.loginWrapper)

# reference: https://learn.microsoft.com/en-us/azure/storage/common/storage-use-azurite#connect-to-the-emulator-by-using-the-azure-storage-explorer
- name: AZURE_STORAGE_CONNECTION_STRING
value: "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite-service.azurite-dev.svc.cluster.local:10000/devstoreaccount1;"
Expand Down
31 changes: 10 additions & 21 deletions historyserver/config/raycluster-gcs.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -63,27 +63,16 @@ spec:
value: *rayTmpRoot
- name: GCS_BUCKET
value: "${GCS_BUCKET}"
# 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.
# If your Ray Dashboard uses a non-default port (not 8265), update this value accordingly.
- name: RAY_DASHBOARD_ADDRESS
value: "http://localhost:8265"
# RAY_COLLECTOR_ADDITIONAL_ENDPOINTS is a comma-separated list of Ray Dashboard
# API endpoint paths that the head collector will periodically poll and store.
# Use this for endpoints whose data cannot be obtained via Ray events.
# You can add more endpoints, e.g., "/api/v0/placement_groups,/api/serve/applications/"
# Note: Only static endpoints (no dynamic path parameters like {job_id}) are supported.
- name: RAY_COLLECTOR_ADDITIONAL_ENDPOINTS
value: "/api/v0/placement_groups?detail=1&limit=10000"
# Query params must match the Ray Dashboard frontend request exactly
# (see https://github.com/ray-project/ray/blob/cb9c80fee6a700efe61ea97987248ce82e3fa2e2/python/ray/dashboard/client/src/service/placementGroup.ts):
# detail=1 — include bundles and stats fields required by PlacementGroupTable
# limit=10000 — match the frontend's default limit to avoid truncation
# RAY_COLLECTOR_POLL_INTERVAL controls how often the collector polls the additional
# endpoints above. Accepts Go duration format (e.g., "30s", "1m", "5m").
- name: RAY_COLLECTOR_POLL_INTERVAL
value: "30s"
# Optional; defaults to http://localhost:8265 (head only).
# - name: RAY_DASHBOARD_ADDRESS
# value: "http://localhost:9265"
# Optional; defaults to 30s.
# - name: RAY_COLLECTOR_POLL_INTERVAL
# value: "1m"
# Optional extras on top of the built-in endpoints (Serve, placement groups, Ray Data).
# Paths must match the frontend request URI, query string included.
# - name: RAY_COLLECTOR_ADDITIONAL_ENDPOINTS
# value: "/nodes?view=summary"
command:
- collector
- --role=Head
Expand Down
31 changes: 10 additions & 21 deletions historyserver/config/raycluster.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -66,27 +66,16 @@ spec:
value: raycluster-historyserver-head-svc.default.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.
# If your Ray Dashboard uses a non-default port (not 8265), update this value accordingly.
- name: RAY_DASHBOARD_ADDRESS
value: "http://localhost:8265"
# RAY_COLLECTOR_ADDITIONAL_ENDPOINTS is a comma-separated list of Ray Dashboard
# API endpoint paths that the head collector will periodically poll and store.
# Use this for endpoints whose data cannot be obtained via Ray events.
# You can add more endpoints, e.g., "/api/v0/placement_groups,/api/serve/applications/"
# Note: Only static endpoints (no dynamic path parameters like {job_id}) are supported.
- name: RAY_COLLECTOR_ADDITIONAL_ENDPOINTS
# Query params must match the Ray Dashboard frontend request exactly
# (see https://github.com/ray-project/ray/blob/cb9c80fee6a700efe61ea97987248ce82e3fa2e2/python/ray/dashboard/client/src/service/placementGroup.ts):
# detail=1 — include bundles and stats fields required by PlacementGroupTable
# limit=10000 — match the frontend's default limit to avoid truncation
value: "/api/v0/placement_groups?detail=1&limit=10000"
# RAY_COLLECTOR_POLL_INTERVAL controls how often the collector polls the additional
# endpoints above. Accepts Go duration format (e.g., "30s", "1m", "5m").
- name: RAY_COLLECTOR_POLL_INTERVAL
value: "30s"
# Optional; defaults to http://localhost:8265 (head only).
# - name: RAY_DASHBOARD_ADDRESS
# value: "http://localhost:9265"
# Optional; defaults to 30s.
# - name: RAY_COLLECTOR_POLL_INTERVAL
# value: "1m"
# Optional extras on top of the built-in endpoints (Serve, placement groups, Ray Data).
# Paths must match the frontend request URI, query string included.
# - name: RAY_COLLECTOR_ADDITIONAL_ENDPOINTS
# value: "/nodes?view=summary"
- name: S3DISABLE_SSL
value: "true"
- name: AWS_ACCESS_KEY_ID
Expand Down
124 changes: 124 additions & 0 deletions historyserver/config/rayservice.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
apiVersion: ray.io/v1
kind: RayService
metadata:
name: rayservice-historyserver
spec:
serveConfigV2: |
proxy_location: EveryNode
applications:
- name: history-e2e
route_prefix: /
import_path: microbenchmarks.no_ops:app_builder
args:
num_forwards: 0
runtime_env:
working_dir: https://github.com/ray-project/serve_workloads/archive/a9f184f4d9ddb7f9a578502ae106470f87a702ef.zip
deployments:
- name: NoOp
num_replicas: 1
ray_actor_options:
num_cpus: 0.5
rayClusterConfig:
# Head-only: worker collectors would need the generated head Service FQDN in FQ_RAY_IP.
headGroupSpec:
rayStartParams:
dashboard-host: 0.0.0.0
serviceType: ClusterIP
template:
spec:
containers:
- 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"
# in ray 2.52.0, we need to set RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES
# in ray 2.53.0 (noy yet done). we need to set RAY_DASHBOARD_AGGREGATOR_AGENT_PUBLISHER_HTTP_ENDPOINT_EXPOSABLE_EVENT_TYPES
- name: RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES
value: "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"
image: rayproject/ray:2.52.0
imagePullPolicy: IfNotPresent
name: ray-head
securityContext:
allowPrivilegeEscalation: true
privileged: true
ports:
- containerPort: 6379
name: gcs-server
- containerPort: 8265
name: dashboard
- containerPort: 10001
name: client
- containerPort: 8000
name: serve
resources:
limits:
cpu: "5"
memory: 10G
requests:
cpu: "50m"
memory: 1G
volumeMounts:
- name: historyserver
mountPath: *rayTmpRoot
- name: collector
image: collector:v0.1.0
imagePullPolicy: IfNotPresent
env:
- name: POD_IP
valueFrom:
fieldRef:
fieldPath: status.podIP
# KubeRay generates the cluster name; read it back from the pod label.
- name: RAY_CLUSTER_NAME
valueFrom:
fieldRef:
fieldPath: metadata.labels['ray.io/cluster']
- name: RAY_CLUSTER_NAMESPACE
valueFrom:
fieldRef:
fieldPath: metadata.namespace
# ray.io/originated-from-* labels are on the RayCluster, not the pod; must match metadata.name.
- name: OWNER_KIND
value: "RayService"
- name: OWNER_NAME
value: "rayservice-historyserver"
# Only used to look up this pod's Ray NodeID; the dashboard is in this pod.
- name: FQ_RAY_IP
value: "localhost"
- name: RAY_TMP_ROOT
value: *rayTmpRoot

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we also want to set RAY_COLLECTOR_POLL_INTERVAL to 5 sec here so test historyserver/test/support/rayservice.go can end quicker?

- 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"
command:
- collector
- --role=Head
- --runtime-class-name=s3
- --ray-root-dir=log
- --events-port=8084
volumeMounts:
- name: historyserver
mountPath: *rayTmpRoot
volumes:
- name: historyserver
emptyDir: {}
Loading
Loading