diff --git a/raydar/dashboard/dashboard.py b/raydar/dashboard/dashboard.py index 3500208..f6c2017 100644 --- a/raydar/dashboard/dashboard.py +++ b/raydar/dashboard/dashboard.py @@ -151,19 +151,21 @@ async def _perspective_socket(self, websocket: WebSocket) -> None: def apply(self, batch: dict) -> None: """Apply a drained :class:`~raydar.ops.OpBuffer` batch to the tables.""" changed = False - for tablename, schema in (batch.get("schemas") or {}).items(): - changed |= self.tables.new_table(tablename, schema) - for tablename in batch.get("cleared") or (): - self.tables.clear(tablename) - changed = True - for tablename, rows in (batch.get("updates") or {}).items(): - if rows: - self.tables.update(tablename, rows) + try: + for tablename, schema in (batch.get("schemas") or {}).items(): + changed |= self.tables.new_table(tablename, schema) + for tablename in batch.get("cleared") or (): + self.tables.clear(tablename) changed = True - # Schemas are replayed on every drain, so most batches are empty; only - # touch the synced model when something actually moved. - if changed: - self._refresh_state() + for tablename, rows in (batch.get("updates") or {}).items(): + if rows: + self.tables.update(tablename, rows) + changed = True + finally: + # Sync whatever landed before a bad row raised, otherwise those tables + # stay invisible: replayed schemas report no change on the next batch. + if changed: + self._refresh_state() def _refresh_state(self) -> None: names = self.tables.names() diff --git a/raydar/task_tracker/schema.py b/raydar/task_tracker/schema.py index a245657..ae63cc8 100644 --- a/raydar/task_tracker/schema.py +++ b/raydar/task_tracker/schema.py @@ -6,7 +6,7 @@ "name": pl.Utf8, "state": pl.Utf8, "job_id": pl.Utf8, - "actor_id": pl.Float32, + "actor_id": pl.Utf8, "type": pl.Utf8, "func_or_class_name": pl.Utf8, "parent_task_id": pl.Utf8, @@ -29,7 +29,7 @@ ), ] ), - "placement_group_id": pl.Float32, + "placement_group_id": pl.Utf8, "events": pl.List(pl.Struct([pl.Field("state", pl.Utf8), pl.Field("created_ms", pl.Float64)])), "profiling_data": pl.Struct( [ diff --git a/raydar/task_tracker/task_tracker.py b/raydar/task_tracker/task_tracker.py index bcd9eb4..1e80d1e 100644 --- a/raydar/task_tracker/task_tracker.py +++ b/raydar/task_tracker/task_tracker.py @@ -131,7 +131,7 @@ def __init__( "name": "string", "state": "string", "job_id": "string", - "actor_id": "float", + "actor_id": "string", "type": "string", "func_or_class_name": "string", "parent_task_id": "string", @@ -139,7 +139,7 @@ def __init__( "worker_id": "string", "error_type": "string", "language": "string", - "placement_group_id": "float", + "placement_group_id": "string", "creation_time_ms": "datetime", "start_time_ms": "datetime", "end_time_ms": "datetime", @@ -357,18 +357,18 @@ def __init__( **kwargs, ) + # `get_if_exists` returns a pre-existing actor and drops these constructor + # args, so a mode mismatch would otherwise show as an empty dashboard. + active = ray.get(self.tracker.get_dashboard_mode.remote()) + if active != dashboard: + logger.warning( + f'Actor "{name}" in namespace "{namespace}" already exists with dashboard={active!r}, ' + f"not the requested {dashboard!r}. Use a new name or namespace." + ) + if dashboard == "local": from raydar.dashboard import LocalDashboard - # `get_if_exists` returns a pre-existing actor and drops these constructor - # args, so a mode mismatch would otherwise show as an empty dashboard. - active = ray.get(self.tracker.get_dashboard_mode.remote()) - if active != dashboard: - logger.warning( - f'Actor "{name}" in namespace "{namespace}" already exists with dashboard={active!r}, ' - f"so it will not feed a {dashboard!r} dashboard. Use a new name or namespace." - ) - self.dashboard = LocalDashboard( drain=lambda: ray.get(self.tracker.drain.remote()), host=dashboard_host, diff --git a/raydar/tests/test_dashboard.py b/raydar/tests/test_dashboard.py index d115e7f..42dd4d1 100644 --- a/raydar/tests/test_dashboard.py +++ b/raydar/tests/test_dashboard.py @@ -68,6 +68,19 @@ def test_update_of_an_unknown_table_raises(self, dashboard): with pytest.raises(KeyError): dashboard.apply({"updates": {"nope": [{"a": 1}]}}) + def test_a_failed_batch_still_syncs_what_landed(self, dashboard): + with pytest.raises(KeyError): + dashboard.apply( + { + "schemas": {"t": SCHEMA}, + "updates": {"t": [{"a": 1, "b": "x"}], "nope": [{"a": 1}]}, + } + ) + + # The good table landed, so it must be visible even though the batch raised. + assert dashboard.state.tables == ["t"] + assert dashboard.state.rows == "1" + def test_limit_caps_retained_rows(self): dashboard = Dashboard(limit=2) dashboard.apply({"schemas": {"t": SCHEMA}, "updates": {"t": [{"a": i, "b": "x"} for i in range(5)]}}) diff --git a/raydar/tests/test_task_tracker.py b/raydar/tests/test_task_tracker.py index 08265cc..b6962b0 100644 --- a/raydar/tests/test_task_tracker.py +++ b/raydar/tests/test_task_tracker.py @@ -13,6 +13,13 @@ def do_some_work(): return True +@ray.remote +class SomeActor: + def do_some_work(self): + time.sleep(0.1) + return True + + def wait_for(fetch, ready, timeout=120, interval=0.5): """Poll `fetch` until `ready` accepts the value, then return it.""" deadline = time.time() + timeout @@ -44,6 +51,25 @@ def test_dashboard_is_off_by_default(self): task_tracker = RayTaskTracker() assert task_tracker.dashboard_url is None + def test_actor_task_ids_survive_as_strings(self): + # Ray reports actor_id as a hex string. Declaring it numeric made get_df + # raise and rendered every id as 0.0 in the dashboard. + task_tracker = RayTaskTracker(dashboard="local") + try: + actor = SomeActor.remote() + refs = [actor.do_some_work.remote() for _ in range(3)] + task_tracker.process(refs) + ray.get(refs) + + df = wait_for(task_tracker.get_df, lambda d: not d.is_empty()) + assert not df.is_empty(), "tracker recorded no finished actor tasks" + + actor_ids = [a for a in df["actor_id"].to_list() if a] + assert actor_ids, "actor_id was not recorded" + assert all(isinstance(a, str) and int(a, 16) for a in actor_ids) + finally: + task_tracker.dashboard.stop() + def test_dashboard_options_reach_the_dashboard(self): layout = {"sizes": [1], "viewers": {}} task_tracker = RayTaskTracker(dashboard="local", dashboard_options={"title": "custom", "layout": layout}) @@ -65,8 +91,11 @@ def test_local_dashboard_serves_tables_pulled_from_the_actor(self): tables = task_tracker.dashboard.dashboard.tables expected = ["custom", "task_tracker_data"] names = wait_for(lambda: sorted(tables.names()), lambda n: n == expected, timeout=60) - assert names == expected + + # Wait on the row too: names alone pass before the update is applied. + rows = wait_for(lambda: tables._tables["custom"].size(), lambda n: n > 0, timeout=60) + assert rows == 1 assert httpx.get(task_tracker.dashboard_url).status_code == 200 finally: task_tracker.dashboard.stop()