[History Server][Feat] Add metadata collect logic

[popojk]

Why are these changes needed?

We need to implement metadata collect logic in order to implement dead cluster behavior for bellow history server API endpoints:

/api/data/datasets/{job_id}
/api/serve/applications/
/api/v0/placement_groups/

Related issue number

#4384
#4385
#4386

Tests

The new added features are more about web api calls, which is hard to do unit test. So we do bellow manual tests:

1. Run ray cluster by executing kubectl apply -f historyserver/config/raycluster.yaml, which will run our collector as sidecar container

2.Execute kubectl port-forward svc/raycluster-historyserver-head-svc 8265:8265 in order to access our ray cluster

3.Execute bellow command to add a placement group to ray cluster

ray job submit --address http://localhost:8265 -- python -c "                                                                                                                                  
import ray
ray.init(address='auto')
pg = ray.util.placement_group(name='my_pg', strategy='PACK', bundles=[{'CPU': 0.2}])
ray.get(pg.ready())
print(f'Created: {pg.id.hex()}')
"

4.Check minio, the placement group data is collected and stored

5.Execute curl -sS http://localhost:8265/api/v0/placement_groups to get data from ray cluster directly, we get the same data

6. Execute kubectl apply -f historyserver/config/rayjob.yaml to test if datasets data will be collected

7. Datasets data were collected as expected

8. Execute bellow command to serve an example app:

ray job submit --address http://localhost:8265 -- python -c "
from ray import serve
from ray.serve import Application

@serve.deployment
class MyDeployment:
  pass

app: Application = MyDeployment.bind()
serve.run(app)
"

9. The application showed up in minio as expected

Checks

• I've made sure the tests are passing.
• Testing Strategy
  • Unit tests
  • Manual tests
  • This PR is not tested :(

[400Ping]

I think overall looks good

[fweilun]

nit. Overall, looks great.

[fweilun]

Is &types.UrlInfo removable?
The slice type is already declared as []*types.UrlInfo.

[fweilun]

Looks like a typo here. (Persist)

[400Ping]

Overall LGTM, please resolve conflicts.

[cursor]

Unsafe type assertion may cause panic

Medium Severity

The type assertion jobinfo.(map[string]interface{}) uses the bare form without checking if the assertion succeeded. If the API returns unexpected data (e.g., a null element in the jobs array or a different data structure), this will panic. Other code in this codebase uses the ok-form pattern for similar assertions.

 

[Future-Outlier]

Hi, @popojk @400Ping @fweilun can you give me an example to test these 3 endpoints?

currently I only see 1 example.

ray job submit --address http://localhost:8265 -- python -c "                                                                                                                                  
import ray
ray.init(address='auto')
pg = ray.util.placement_group(name='my_pg', strategy='PACK', bundles=[{'CPU': 0.2}])
ray.get(pg.ready())
print(f'Created: {pg.id.hex()}')
"

[Future-Outlier]

1. I need you to provide an example that can run end-to-end and lets me view, via the Ray Dashboard, information for:

   • Serve applications
   • Dataset endpoints
   • Placement group endpoints
     And please record a video showing it.

   a. If possible, add a new file named rayjob-workaround.yaml and include the example there.
   Try to make it look as similar to rayjob.yaml as possible.

2. There’s a mistake in the PR description: running kubectl apply -f historyserver/config/historyserver.yaml will not create a RayCluster.

3. For the collector, please add an environment variable to enable this “persist metadata” function (when enabled, it should try to fetch data from blob storage).

env:
  - name: SUPPORT_RAY_EVENT_UNSUPPORTED_DATA
    value: 1 # or true

4. The endpoints will hit port 8265. Is there a way to know the Ray Dashboard port ahead of time? Hardcoding it doesn’t feel great.

[cursor]

Race condition modifying struct fields outside mutex

Medium Severity

After calling jobResourcesUrlInfo.Get(), the returned *types.JobUrlInfo pointer is modified (existingJob.Status and urlInfo.StopPersist) outside of any lock. The Get method releases its read lock before the modifications occur, creating a data race if these fields are accessed concurrently.

Additional Locations (1)

• historyserver/pkg/collector/logcollector/runtime/logcollector/meta.go#L238-L241

 

[400Ping]

Here's the yaml I used to test:

# RayJob that demonstrates endpoints used by the "persist metadata" workaround:
# - Placement groups (api/v0/placement_groups)
# - Serve applications (api/serve/applications/)
# - Datasets (api/jobs/ + api/data/datasets/{job_id})
#
# Use this with the History Server setup guide (Step 6). Ensure the collector
# has SUPPORT_RAY_EVENT_UNSUPPORTED_DATA enabled so it persists these to blob storage.
# After the job runs, you can view the data in Ray Dashboard, and after cluster
# teardown, via History Server (from persisted meta).
#
apiVersion: ray.io/v1
kind: RayJob
metadata:
  name: rayjob-workaround
spec:
  entrypoint: |
    python -c "
    import ray
    import time

    ray.init()

    # --- 1. Placement group (visible under api/v0/placement_groups) ---
    pg = ray.util.placement_group(
        name='workaround_pg',
        strategy='PACK',
        bundles=[{'CPU': 0.2}]
    )
    ray.get(pg.ready())
    print('Placement group created:', pg.id.hex())

    # --- 2. Serve application (visible under api/serve/applications/) ---
    from ray import serve

    @serve.deployment(num_replicas=1)
    def hello(request):
        return {'message': 'hello from workaround'}

    serve.run(hello.bind(), name='workaround_app')
    print('Serve application deployed: workaround_app')

    # --- 3. Dataset (creates job + dataset; visible under api/jobs/ and api/data/datasets/) ---
    ds = ray.data.range(10)
    n = ds.count()
    print('Dataset created, count:', n)

    # Keep running so the collector can poll Dashboard API and persist meta (e.g. every 5s).
    # Wait long enough for at least one full persist cycle.
    print('Waiting 15s for collector to persist metadata...')
    time.sleep(15)

    # Align with existing guide: allow events to be sent before shutdown.
    time.sleep(5)
    print('Done.')
    "
  clusterSelector:
    ray.io/cluster: raycluster-historyserver

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

[cursor]

Worker nodes panic when RAY_DASHBOARD_ADDRESS is unset

Medium Severity

The collector unconditionally panics if RAY_DASHBOARD_ADDRESS is not set, regardless of node role. However, the dashboard address is only needed when SupportRayEventUnSupportData is enabled, which only applies to Head nodes. Worker nodes will crash unnecessarily if this environment variable is missing, even though they never use it for metadata collection.

 

[popojk]

1. I need you to provide an example that can run end-to-end and lets me view, via the Ray Dashboard, information for:

   • Serve applications
   • Dataset endpoints
   • Placement group endpoints
     And please record a video showing it.

   a. If possible, add a new file named rayjob-workaround.yaml and include the example there.
   Try to make it look as similar to rayjob.yaml as possible.

2. There’s a mistake in the PR description: running kubectl apply -f historyserver/config/historyserver.yaml will not create a RayCluster.

3. For the collector, please add an environment variable to enable this “persist metadata” function (when enabled, it should try to fetch data from blob storage).

env:
  - name: SUPPORT_RAY_EVENT_UNSUPPORTED_DATA
    value: 1 # or true

4. The endpoints will hit port 8265. Is there a way to know the Ray Dashboard port ahead of time? Hardcoding it doesn’t feel great.

@Future-Outlier Changes made:

1. Add rayjob_workaround.yaml to apply placement group, application, and datasets.
2. Fixed PR description
3. Add SUPPORT_RAY_EVENT_UNSUPPORTED_DATA ENVVAR to ray cluster to enable metadata collect
4. Add RAY_DASHBOARD_ADDRESS ENVVAR to ray cluster to set dashboard address for collector