fix(ray): propagate original function name to Ray metrics instead of generic _capture [ENG-358]#128
Conversation
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
There was a problem hiding this comment.
Pull request overview
This PR updates the Ray executor’s remote capture wrapper creation so Ray reports the original packet function name in Ray metrics/dashboard (instead of the generic _capture), improving observability in Grafana.
Changes:
- Pass the target function name into
RayExecutor._get_remote_fn()and cache remote wrappers by(remote_opts, fn_name). - Extend
make_capture_wrapper()to optionally set the wrapper function’s__name__/__qualname__to the original function name.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
src/orcapod/core/executors/ray.py |
Creates/caches Ray-remote wrappers per function name so Ray metrics reflect the executed function’s name. |
src/orcapod/core/executors/capture_wrapper.py |
Adds an optional name parameter to rename the capture wrapper function for remote-framework reporting. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| def make_capture_wrapper(name: str | None = None) -> Callable[..., Any]: | ||
| """Return a capture wrapper suitable for remote execution. | ||
|
|
||
| Args: | ||
| name: If provided, the wrapper's ``__name__`` and ``__qualname__`` | ||
| are set to this value so that remote frameworks (e.g. Ray) report | ||
| the original function name in dashboards and metrics rather than | ||
| the generic ``_capture``. |
There was a problem hiding this comment.
New behavior (setting wrapper name/qualname when name is provided) isn't covered by tests. Please add a unit test asserting that make_capture_wrapper(name="...") returns a callable whose __name__ and __qualname__ match the provided name (and that default None keeps the original _capture name).
| opts = self._build_remote_opts() | ||
| cache_key = self._normalize_opts(opts) | ||
| cache_key = (self._normalize_opts(opts), fn_name) | ||
| if cache_key not in self._remote_fn_cache: | ||
| with self._remote_fn_cache_lock: | ||
| # Double-checked: another thread may have filled the slot | ||
| # while we waited for the lock. | ||
| if cache_key not in self._remote_fn_cache: | ||
| wrapper = make_capture_wrapper() | ||
| wrapper = make_capture_wrapper(name=fn_name) | ||
| self._remote_fn_cache[cache_key] = ray.remote(**opts)(wrapper) |
There was a problem hiding this comment.
_get_remote_fn() now caches remote wrappers per (remote_opts, fn_name). If callers can generate many distinct function names (e.g., dynamically generated callables), this cache can grow without bound over the lifetime of the process. Consider bounding/evicting entries (e.g., LRU) or normalizing the name to a stable identifier (such as module + qualname) to avoid unexpected memory growth.
There was a problem hiding this comment.
Acknowledged — in practice the set of function names is bounded by the pipeline's topology (a fixed, small number of packet functions defined at construction time), so unbounded growth isn't a realistic concern here. Dismissing as not applicable.
| @@ -206,7 +206,7 @@ async def async_execute_callable( | |||
| import ray | |||
|
|
|||
| self._ensure_ray_initialized() | |||
| remote_fn = self._get_remote_fn(ray) | |||
| remote_fn = self._get_remote_fn(ray, fn.__name__) | |||
| ref = remote_fn.options(name=fn.__name__).remote(fn, kwargs) | |||
There was a problem hiding this comment.
The metrics-related change (propagating fn.__name__ into the Ray-remote wrapper via _get_remote_fn(..., fn.__name__)) isn't covered by tests. Adding a unit test with a mocked ray module that asserts make_capture_wrapper(name=...) is used and that different fn_name values result in distinct cached remote functions would help prevent regressions.
brian-arnold
changed the title
This PR reimplements the feature that the function passed to Ray retains it's original name, instead of getting passed as _capture. This way, functions show of as distinct entities in Grafana dashboards.