This PR adds Export API support for Ray Train state events.

## Key Changes

- Added new proto messages `ExportTrainRunEventData` and
`ExportTrainRunAttemptEventData` to capture training state
- Created `EventLogType` enum to manage different types of export event
logs
- Updated `TrainStateActor` to export Train state events when export is
enabled
- Modified timestamp fields from milliseconds to nanoseconds (for both
proto and python schema)
  - `start_time_ms` → `start_time_ns`
  - `end_time_ms` → `end_time_ns`

## Implementation Details

- Train run and attempt events are now written to the
`event_EXPORT_TRAIN_STATE.log` log file when the export API is enabled
- Export can be enabled either globally or specifically for Train events
using environment variables:
  - `RAY_enable_export_api_write=1` (all events)
- `RAY_enable_export_api_write_config=EXPORT_TRAIN_RUN` (Train run
events only)
- `RAY_enable_export_api_write_config=EXPORT_TRAIN_RUN_ATTEMPT` (Train
run attempt events only)

Based off of #47888.
Follows the new schema added in
#50515.

---------

Signed-off-by: Matthew Deng <matt@anyscale.com>
Signed-off-by: Alan Guo <aguo@anyscale.com>
Signed-off-by: Justin Yu <justinvyu@anyscale.com>
Co-authored-by: Alan Guo <aguo@anyscale.com>
Co-authored-by: Justin Yu <justinvyu@anyscale.com>
Signed-off-by: Abrar Sheikh <abrar@anyscale.com>

This PR implements the export API for Ray Train V1 state. This builds on
top of #50622, which implements
the export API for Ray Train V2.


## Key Changes

- Added `export.py` with conversion functions between Train V1 state
models and Train (V2) state export protobuf
- Updated `TrainRunInfo` and `TrainWorkerInfo` schemas with additional
fields for compatibility:
  - Log file paths for controller and workers
- Note that these point to the Ray worker stderr logs, rather than
specific train logs.
  - Resource allocation information
  - Made worker status a required field
     - Note that it is always set as ACTIVE for now.


Signed-off-by: Matthew Deng <matt@anyscale.com>

This PR adds Export API support for Ray Train state events.

## Key Changes

- Added new proto messages `ExportTrainRunEventData` and
`ExportTrainRunAttemptEventData` to capture training state
- Created `EventLogType` enum to manage different types of export event
logs
- Updated `TrainStateActor` to export Train state events when export is
enabled
- Modified timestamp fields from milliseconds to nanoseconds (for both
proto and python schema)
  - `start_time_ms` → `start_time_ns`
  - `end_time_ms` → `end_time_ns`

## Implementation Details

- Train run and attempt events are now written to the
`event_EXPORT_TRAIN_STATE.log` log file when the export API is enabled
- Export can be enabled either globally or specifically for Train events
using environment variables:
  - `RAY_enable_export_api_write=1` (all events)
- `RAY_enable_export_api_write_config=EXPORT_TRAIN_RUN` (Train run
events only)
- `RAY_enable_export_api_write_config=EXPORT_TRAIN_RUN_ATTEMPT` (Train
run attempt events only)

Based off of ray-project#47888.
Follows the new schema added in
ray-project#50515.

---------

Signed-off-by: Matthew Deng <matt@anyscale.com>
Signed-off-by: Alan Guo <aguo@anyscale.com>
Signed-off-by: Justin Yu <justinvyu@anyscale.com>
Co-authored-by: Alan Guo <aguo@anyscale.com>
Co-authored-by: Justin Yu <justinvyu@anyscale.com>

This PR implements the export API for Ray Train V1 state. This builds on
top of ray-project#50622, which implements
the export API for Ray Train V2.


## Key Changes

- Added `export.py` with conversion functions between Train V1 state
models and Train (V2) state export protobuf
- Updated `TrainRunInfo` and `TrainWorkerInfo` schemas with additional
fields for compatibility:
  - Log file paths for controller and workers
- Note that these point to the Ray worker stderr logs, rather than
specific train logs.
  - Resource allocation information
  - Made worker status a required field
     - Note that it is always set as ACTIVE for now.


Signed-off-by: Matthew Deng <matt@anyscale.com>

This PR implements the export API for Ray Train V1 state. This builds on
top of ray-project#50622, which implements
the export API for Ray Train V2.

## Key Changes

- Added `export.py` with conversion functions between Train V1 state
models and Train (V2) state export protobuf
- Updated `TrainRunInfo` and `TrainWorkerInfo` schemas with additional
fields for compatibility:
  - Log file paths for controller and workers
- Note that these point to the Ray worker stderr logs, rather than
specific train logs.
  - Resource allocation information
  - Made worker status a required field
     - Note that it is always set as ACTIVE for now.

Signed-off-by: Matthew Deng <matt@anyscale.com>
Signed-off-by: Dhakshin Suriakannu <d_suriakannu@apple.com>

## Why are these changes needed?
_Inspired / based on the equivalent train work
[here](#50622

We want to be able to persist the metadata for a dataset using the
export api. Initially this will include the dag toplogy, but in the
future we can add more to this. It will be exported once upon
registering the dataset to the stats actor.

## Example
Test script:
```python
import ray

ds = ray.data.range(1000)
ds = ds.map(lambda x: {"id": x["id"], "squared": x["id"]*x["id"]})
ds = ds.filter(lambda x: x["id"] % 3 == 0)
ds = ds.random_shuffle(seed=42)
ds = ds.map(lambda x: {**x, "cubed": x["id"]**3})


for _ in ds.iter_batches():
    pass
```

Verifying events are being exported:
```
❯ cat test_topo.py
import ray

ds = ray.data.range(1000)
ds = ds.map(lambda x: {"id": x["id"], "squared": x["id"]*x["id"]})
ds = ds.filter(lambda x: x["id"] % 3 == 0)
ds = ds.random_shuffle(seed=42)
ds = ds.map(lambda x: {**x, "cubed": x["id"]**3})


for _ in ds.iter_batches():
    pass
❯ RAY_enable_export_api_write=True python test_topo.py
2025-04-10 14:41:53,395	INFO worker.py:1849 -- Started a local Ray instance. View the dashboard at 127.0.0.1:8265
/Users/mowen/code/ray/python/ray/data/dataset.py:1402: UserWarning: Use 'expr' instead of 'fn' when possible for performant filters.
  warnings.warn(
2025-04-10 14:41:54,380	INFO streaming_executor.py:108 -- Starting execution of Dataset. Full logs are in /tmp/ray/session_2025-04-10_14-41-51_402358_17728/logs/ray-data
2025-04-10 14:41:54,380	INFO streaming_executor.py:109 -- Execution plan of Dataset: InputDataBuffer[Input] -> AllToAllOperator[ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle] -> TaskPoolMapOperator[Map(<lambda>)]
Running 0: 0.00 row [00:00, ? row/s]        2025-04-10 14:41:55,104	WARNING progress_bar.py:120 -- Truncating long operator name to 100 characters. To disable this behavior, set `ray.data.DataContext.get_current().DEFAULT_ENABLE_PROGRESS_BAR_NAME_TRUNCATION = False`.0%|                                                                             | 0.00/1.00 [00:00<?, ? row/s]
✔️  Dataset execution finished in 0.76 seconds: 100%|████████████████████████████████████████| 334/334 [00:00<00:00, 441 row/s]

- ReadRange->...->RandomShuffle: Tasks: 0; Queued blocks: 0; Resources: 0.0 CPU, 0.0B object store; 334 rows output: : 0.00 row- ReadRange->...->RandomShuffle: Tasks: 0; Queued blocks: 0; Resources: 0.0 CPU, 0.0B object store; 334 rows output: 100%|█| 33  *- Shuffle Map: 100%|██████████████████████████████████████████████████████████████████████| 334/334 [00:00<00:00, 450 row/s]
  *- Shuffle Reduce: 100%|███████████████████████████████████████████████████████████████████| 334/334 [00:00<00:00, 449 row/s]
- Map(<lambda>): Tasks: 0; Queued blocks: 0; Resources: 0.0 CPU, 2.8KB object store: 100%|███| 334/334 [00:00<00:00, 449 row/s]
❯ cat /tmp/ray/session_2025-04-16_19-47-48_523587_36185/logs/export_events/event_EXPORT_DATA_METADATA.log
{"event_id": "dfb2EF5c345aE39d58", "timestamp": 1744858071, "source_type": "EXPORT_DATA_METADATA", "event_data": {"dag": {"operators": [{"name": "Input", "id": "Input_0", "uuid": "1356dab5-1a24-4afc-9bb5-f520b90fb76a", "input_dependencies": [], "sub_operators": []}, {"name": "ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle", "id": "ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle_1", "uuid": "39e11798-20e0-4b06-b7cb-345fc0b74f63", "input_dependencies": ["Input_0"], "sub_operators": [{"name": "Shuffle Map", "id": "ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle_1_sub_0"}, {"name": "Shuffle Reduce", "id": "ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle_1_sub_1"}]}, {"name": "Map(<lambda>)", "id": "Map(<lambda>)_2", "uuid": "46da676f-48a1-43f7-894e-93b8a08d1eb2", "input_dependencies": ["ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle_1"], "sub_operators": []}]}, "dataset_id": "dataset_4", "job_id": "01000000", "start_time": "1744858071"}}
```

Formatted export event JSON
```
{
  "event_id": "dfb2EF5c345aE39d58",
  "timestamp": 1744858071,
  "source_type": "EXPORT_DATA_METADATA",
  "event_data": {
    "dag": {
      "operators": [
        {
          "name": "Input",
          "id": "Input_0",
          "uuid": "1356dab5-1a24-4afc-9bb5-f520b90fb76a",
          "input_dependencies": [],
          "sub_operators": []
        },
        {
          "name": "ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle",
          "id": "ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle_1",
          "uuid": "39e11798-20e0-4b06-b7cb-345fc0b74f63",
          "input_dependencies": [
            "Input_0"
          ],
          "sub_operators": [
            {
              "name": "Shuffle Map",
              "id": "ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle_1_sub_0"
            },
            {
              "name": "Shuffle Reduce",
              "id": "ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle_1_sub_1"
            }
          ]
        },
        {
          "name": "Map(<lambda>)",
          "id": "Map(<lambda>)_2",
          "uuid": "46da676f-48a1-43f7-894e-93b8a08d1eb2",
          "input_dependencies": [
            "ReadRange->Map(<lambda>)->Filter(<lambda>)->RandomShuffle_1"
          ],
          "sub_operators": []
        }
      ]
    },
    "dataset_id": "dataset_4",
    "job_id": "01000000",
    "start_time": "1744858071"
  }
}
```

## Related issue number

<!-- For example: "Closes #1234" -->

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for
https://docs.ray.io/en/master/.
- [ ] I've added any new APIs to the API Reference. For example, if I
added a
method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a
few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

---------

Signed-off-by: Matthew Owen <mowen@anyscale.com>