diff --git a/README.md b/README.md index c99ede7..9725fc2 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,19 @@ Use `python examples/sdk/get_clusters.py` and Creating the example reserves GPU capacity and may incur usage charges. It does not delete the deployment automatically. +### Deployment logs SDK example + +Logs are read per pod. Discover pod names with `get_deployment_pods()` (terminated +pods still within log retention are included), then read with a +`deployment_log_session()`: `fetch_older()` pages toward the beginning of history and +`fetch_newer()` returns only new lines, while the session keeps the merged, ordered +log in `.events`. The same paging is available statelessly through +`get_deployment_logs(before=..., after=...)`, anchored on events you already hold: + +```bash +python examples/sdk/get_deployment_logs.py +``` + ### Un-installation To uninstall `centml`, simply do: diff --git a/centml/sdk/api.py b/centml/sdk/api.py index 493a750..102ddee 100644 --- a/centml/sdk/api.py +++ b/centml/sdk/api.py @@ -1,4 +1,6 @@ +from bisect import insort from contextlib import contextmanager +from typing import List, Optional import platform_api_python_client from platform_api_python_client import ( @@ -20,6 +22,11 @@ STATUS_V3_DEPLOYMENT_TYPES = {DeploymentType.INFERENCE_V3, DeploymentType.CSERVE_V3} +DEFAULT_LOG_PAGE_LINES = 100 # server-side default for max_lines +# The server re-delivers a ~15s look-behind window on fetch-newer requests; only the +# caller's events within this generous margin of the boundary can be re-delivered. +LOG_DEDUP_RETENTION_MS = 300_000 + class CentMLClient: def __init__(self, api): @@ -187,46 +194,127 @@ def get_deployment_revisions(self, deployment_id: int): deployment_id=deployment_id ).results + def get_deployment_pods(self, deployment_id: int, revision_number: int) -> List[str]: + """List pods that have logged for a deployment revision, including terminated + pods still within log retention. A fresh deployment may return an empty list.""" + return self._api.get_deployment_pods_deployments_pods_deployment_id_revision_number_get( + deployment_id=deployment_id, revision_number=revision_number + ).pods + + # pylint: disable=R0917 def get_deployment_logs( self, deployment_id: int, revision_number: int, - start_time: int, - end_time: int, - line_count: int = 100, - start_from_head: bool = True, - stream: bool = False, - ): - """Fetch logs for a deployment within a time window, handling pagination automatically. - - start_time and end_time are Unix timestamps in milliseconds. - Use get_deployment_revisions() to find the current revision number. - - If stream=True, returns a generator that yields events as each page is fetched. - If stream=False (default), returns a flat list of all events. + pod: str, + before: Optional[list] = None, + after: Optional[list] = None, + max_lines: int = DEFAULT_LOG_PAGE_LINES, + ) -> list: + """Fetch one page of a pod's logs, oldest-first. Use get_deployment_pods() to + discover pod names and get_deployment_revisions() for the revision number. + + before and after anchor the page to events a previous call returned for the + same pod (pass your accumulated list; only the relevant boundary is used): + - neither: the newest page (tail). + - before=: the page strictly older than the oldest of them; + an empty result means the beginning of history is reached. + - after=: lines strictly newer than the newest of them; an empty + result means nothing new yet — call again later to keep tailing. Late + lines still landing near that boundary are included on top of max_lines + and may sort below events you already hold (order by id if that matters). """ + if before is not None and after is not None: + raise ValueError("before and after are mutually exclusive") + + fetch_newer = after is not None + anchor_events = after if fetch_newer else before + boundary_timestamp = None + if anchor_events: + timestamps = [event.timestamp for event in anchor_events] + boundary_timestamp = max(timestamps) if fetch_newer else min(timestamps) + + response = self._api.get_deployment_logs_v4_logs_deployment_id_revision_number_get( + deployment_id=deployment_id, + revision_number=revision_number, + pod=pod, + fetch_newer=fetch_newer, + timestamp=boundary_timestamp, + max_lines=max_lines, + ) + if not fetch_newer or not anchor_events: + return response.events + + # fetch_newer re-delivers a look-behind window at and before the boundary + # (late-arrival protection); drop the lines the caller already holds by id. + cutoff = max(event.timestamp for event in anchor_events) - LOG_DEDUP_RETENTION_MS + held_event_ids = {event.id for event in anchor_events if event.timestamp >= cutoff} + return [event for event in response.events if event.id not in held_event_ids] + + def deployment_log_session( + self, deployment_id: int, revision_number: int, pod: str, events: Optional[list] = None + ) -> "DeploymentLogSession": + """Stateful reader for one pod's logs that tracks fetched pages and anchors + every request itself — see DeploymentLogSession. Seed events with logs a + previous session (or get_deployment_logs) returned for the same pod.""" + return DeploymentLogSession(self, deployment_id, revision_number, pod, events) + + +class DeploymentLogSession: + """Maintains a contiguous, ordered window of one pod's logs across fetches. + + Every fetch is anchored on the window itself, so pages can never overlap or + leave gaps inside it (within log retention; an undetectable gap forms if the + session idles past retention before fetching newer lines). + """ - def _iter_events(): - next_page_token = None - while True: - response = self._api.get_deployment_logs_v3_deployments_logs_v3_deployment_id_revision_number_get( - deployment_id=deployment_id, - revision_number=revision_number, - start_time=start_time, - end_time=end_time, - next_page_token=next_page_token, - start_from_head=start_from_head, - line_count=line_count, - ) - yield from response.events - next_page_token = response.next_page_token - if not next_page_token: - break - - if stream: - return _iter_events() - - return list(_iter_events()) + # pylint: disable=R0917 + def __init__(self, client: CentMLClient, deployment_id: int, revision_number: int, pod: str, events=None): + self._client = client + self._deployment_id = deployment_id + self._revision_number = revision_number + self._pod = pod + # Seeded events come from outside the session: canonicalize to unique ids in + # chronological order (id order == time order at nanosecond precision). + unique_events = {event.id: event for event in events or []} + self._events = [unique_events[event_id] for event_id in sorted(unique_events)] + + @property + def events(self) -> list: + """Copy of the window fetched so far, oldest first. Complete from the beginning + of history only once fetch_older() has returned an empty list.""" + return list(self._events) + + def fetch_older(self, max_lines: int = DEFAULT_LOG_PAGE_LINES) -> list: + """Fetch the page older than the window and prepend it; on an empty session + fetches the newest page (tail). Returns the page; empty list = no older + lines exist (yet).""" + page = self._client.get_deployment_logs( + self._deployment_id, self._revision_number, self._pod, before=self._events, max_lines=max_lines + ) + self._events[:0] = page + return page + + def fetch_newer(self, max_lines: int = DEFAULT_LOG_PAGE_LINES) -> list: + """Fetch lines newer than the window and merge them in; on an empty session + fetches the newest page (tail) — to read from the beginning of history + instead, loop fetch_older() until it returns an empty list. Returns only + the new lines; empty list = nothing new yet, call again later to keep + tailing. Rare late arrivals sort into the window below its newest lines.""" + if not self._events: + return self.fetch_older(max_lines=max_lines) + delta = self._client.get_deployment_logs( + self._deployment_id, self._revision_number, self._pod, after=self._events, max_lines=max_lines + ) + for event in delta: + if event.id > self._events[-1].id: + self._events.append(event) + else: + # A late arrival may even precede the window's oldest line (tail page + # cut inside the look-behind span); the server delivers that span + # completely on top of max_lines, so the window stays contiguous. + insort(self._events, event, key=lambda held: held.id) + return delta @contextmanager diff --git a/examples/sdk/get_deployment_logs.py b/examples/sdk/get_deployment_logs.py index 6c7d7ea..3114413 100644 --- a/examples/sdk/get_deployment_logs.py +++ b/examples/sdk/get_deployment_logs.py @@ -1,74 +1,54 @@ -from datetime import datetime, timezone, timedelta +import time +from datetime import datetime, timezone from centml.sdk.api import get_centml_client # --- Configuration --- DEPLOYMENT_ID = 1234 # Replace with your deployment ID REVISION_NUMBER = 10 -HOURS_BACK = 1 # Fetch logs from the last N hours +TAIL_SECONDS = 30 # How long to keep polling for new lines after reading history -def format_event(event: dict) -> str: - timestamp_ms = ( - event.get("timestamp") - or event.get("time") - or event.get("ts") - or "" - ) - message = ( - event.get("message") - or event.get("msg") - or event.get("log") - or str(event) - ) - if timestamp_ms: - ts = datetime.fromtimestamp(int(timestamp_ms) / 1000, tz=timezone.utc).isoformat() - return f"[{ts}] {message}" - return message +def format_event(event) -> str: + ts = datetime.fromtimestamp(event.timestamp / 1000, tz=timezone.utc).isoformat() + return f"[{ts}] {event.message}" def main(): - stream = True - end_time = int(datetime.now(timezone.utc).timestamp() * 1000) - start_time = end_time - int(timedelta(hours=HOURS_BACK).total_seconds() * 1000) - - print(f"Fetching logs for deployment {DEPLOYMENT_ID}") - print( - f"Time window: " - f"{datetime.fromtimestamp(start_time / 1000, tz=timezone.utc).isoformat()} → " - f"{datetime.fromtimestamp(end_time / 1000, tz=timezone.utc).isoformat()}" - ) - print() - with get_centml_client() as cclient: - if stream: - # Streaming: print events as each page arrives - for event in cclient.get_deployment_logs( - deployment_id=DEPLOYMENT_ID, - revision_number=REVISION_NUMBER, - start_time=start_time, - end_time=end_time, - start_from_head=False, - stream=stream, - ): + # Logs are read per pod: discover the pods that have logged for this revision + # (terminated pods within log retention are included). + pods = cclient.get_deployment_pods(DEPLOYMENT_ID, REVISION_NUMBER) + if not pods: + print("No pods have logged for this revision yet.") + return + + pod = pods[0] + print(f"Reading logs for deployment {DEPLOYMENT_ID} revision {REVISION_NUMBER}, pod {pod}\n") + + # The session tracks what it has fetched and anchors every request itself. + session = cclient.deployment_log_session(DEPLOYMENT_ID, REVISION_NUMBER, pod) + + # Read the full history: newest page first, then page back to the beginning. + while session.fetch_older(): + pass + print(f"Found {len(session.events)} log entries:\n") + for event in session.events: + print(format_event(event)) + + # Keep tailing: each call returns only the lines the session does not hold yet. + print(f"\nPolling for new lines for {TAIL_SECONDS}s...") + deadline = time.monotonic() + TAIL_SECONDS + while time.monotonic() < deadline: + for event in session.fetch_newer(): print(format_event(event)) - else: - # Batch: collect all events then process - events = cclient.get_deployment_logs( - deployment_id=DEPLOYMENT_ID, - revision_number=REVISION_NUMBER, - start_time=start_time, - end_time=end_time, - start_from_head=False, - ) + time.sleep(2) - if not events: - print("No logs found in the given time window.") - return - - print(f"Found {len(events)} log entries:\n") - for event in events: - print(format_event(event)) + # The same paging is available statelessly via get_deployment_logs, anchored + # on events you already hold — useful when you manage storage yourself: + # page = cclient.get_deployment_logs(DEPLOYMENT_ID, REVISION_NUMBER, pod=pod) # tail + # older = cclient.get_deployment_logs(..., pod=pod, before=page) # [] = beginning + # newer = cclient.get_deployment_logs(..., pod=pod, after=page) # [] = nothing new if __name__ == "__main__": diff --git a/requirements.txt b/requirements.txt index 123f832..8dd6027 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,5 +5,5 @@ pyjwt>=2.8.0 cryptography==48.0.1 websockets>=16.0 pyte>=0.8.0 -platform-api-python-client==4.23.1 +platform-api-python-client==4.25.0 click>=8.4.1 diff --git a/tests/test_sdk_api.py b/tests/test_sdk_api.py index 1e2a642..366327d 100644 --- a/tests/test_sdk_api.py +++ b/tests/test_sdk_api.py @@ -209,3 +209,252 @@ def test_delete_hardware_instance_delegates_to_platform_client(): assert response is expected_response api.delete_hardware_instance_hardware_instances_hardware_instance_id_delete.assert_called_once_with(123) + + +def _log_event(event_id, timestamp, message="line"): + return SimpleNamespace(id=event_id, timestamp=timestamp, message=message) + + +def _log_page(*events): + return SimpleNamespace(events=list(events)) + + +def test_generated_client_exposes_logs_v4_contract(): + assert hasattr( + platform_api_python_client.EXTERNALApi, "get_deployment_logs_v4_logs_deployment_id_revision_number_get" + ) + assert hasattr( + platform_api_python_client.EXTERNALApi, "get_deployment_pods_deployments_pods_deployment_id_revision_number_get" + ) + + +def test_get_deployment_pods_returns_pod_names(): + api = MagicMock() + api.get_deployment_pods_deployments_pods_deployment_id_revision_number_get.return_value = SimpleNamespace( + pods=["pod-a", "pod-b"] + ) + client = CentMLClient(api) + + assert client.get_deployment_pods(123, 2) == ["pod-a", "pod-b"] + + api.get_deployment_pods_deployments_pods_deployment_id_revision_number_get.assert_called_once_with( + deployment_id=123, revision_number=2 + ) + + +def test_get_deployment_logs_returns_tail_page_when_unanchored(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page( + _log_event("1-a", 1000), _log_event("2-b", 2000) + ) + client = CentMLClient(api) + + events = client.get_deployment_logs(123, 2, pod="pod-a") + + assert [e.id for e in events] == ["1-a", "2-b"] + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_called_once_with( + deployment_id=123, revision_number=2, pod="pod-a", fetch_newer=False, timestamp=None, max_lines=100 + ) + + +def test_get_deployment_logs_before_pages_older_from_oldest_anchor(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page(_log_event("1-a", 1000)) + client = CentMLClient(api) + held = [_log_event("2-b", 2000), _log_event("3-c", 3000)] + + events = client.get_deployment_logs(123, 2, pod="pod-a", before=held) + + assert [e.id for e in events] == ["1-a"] + call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args + # The boundary is the oldest held timestamp, passed verbatim (exclusive server-side). + assert call.kwargs["fetch_newer"] is False + assert call.kwargs["timestamp"] == 2000 + + +def test_get_deployment_logs_before_empty_page_signals_beginning_of_history(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page() + client = CentMLClient(api) + + assert client.get_deployment_logs(123, 2, pod="pod-a", before=[_log_event("1-a", 1000)]) == [] + + +def test_get_deployment_logs_after_fetches_newer_from_newest_anchor(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page(_log_event("3-c", 3000)) + client = CentMLClient(api) + held = [_log_event("1-a", 1000), _log_event("2-b", 2000)] + + events = client.get_deployment_logs(123, 2, pod="pod-a", after=held) + + assert [e.id for e in events] == ["3-c"] + call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args + assert call.kwargs["fetch_newer"] is True + assert call.kwargs["timestamp"] == 2000 + + +def test_get_deployment_logs_after_drops_redelivered_lines_but_keeps_late_arrivals(): + api = MagicMock() + # The look-behind window re-delivers 2-b (already held) plus a late arrival at 1500. + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page( + _log_event("15-l", 1500), _log_event("2-b", 2000), _log_event("3-c", 3000) + ) + client = CentMLClient(api) + held = [_log_event("1-a", 1000), _log_event("2-b", 2000)] + + events = client.get_deployment_logs(123, 2, pod="pod-a", after=held) + + assert [e.id for e in events] == ["15-l", "3-c"] + + +def test_get_deployment_logs_after_empty_anchor_reads_from_head(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page(_log_event("1-a", 1000)) + client = CentMLClient(api) + + events = client.get_deployment_logs(123, 2, pod="pod-a", after=[]) + + assert [e.id for e in events] == ["1-a"] + call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args + assert call.kwargs["fetch_newer"] is True + assert call.kwargs["timestamp"] is None + + +def test_get_deployment_logs_after_empty_page_signals_nothing_new(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page() + client = CentMLClient(api) + + assert client.get_deployment_logs(123, 2, pod="pod-a", after=[_log_event("1-a", 1000)]) == [] + + +def test_get_deployment_logs_passes_max_lines_through(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page() + client = CentMLClient(api) + + client.get_deployment_logs(123, 2, pod="pod-a", max_lines=7) + + call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args + assert call.kwargs["max_lines"] == 7 + + +def test_get_deployment_logs_rejects_before_and_after_together(): + api = MagicMock() + client = CentMLClient(api) + + try: + client.get_deployment_logs(123, 2, pod="pod-a", before=[_log_event("1-a", 1000)], after=[]) + except ValueError: + pass + else: + raise AssertionError("Expected ValueError") + + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_not_called() + + +def _session(api, events=None): + return CentMLClient(api).deployment_log_session(123, 2, "pod-a", events=events) + + +def test_log_session_first_fetch_is_tail_for_both_directions(): + for method in ("fetch_older", "fetch_newer"): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page( + _log_event("1-a", 1000), _log_event("2-b", 2000) + ) + session = _session(api) + + page = getattr(session, method)() + + assert [e.id for e in page] == ["1-a", "2-b"] + assert [e.id for e in session.events] == ["1-a", "2-b"] + call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args + assert call.kwargs["fetch_newer"] is False and call.kwargs["timestamp"] is None + + +def test_log_session_fetch_older_prepends_until_beginning(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("3-c", 3000), _log_event("4-d", 4000)), # tail + _log_page(_log_event("1-a", 1000), _log_event("2-b", 2000)), # older page + _log_page(), # beginning reached + ] + session = _session(api) + session.fetch_older() + + older = session.fetch_older() + assert [e.id for e in older] == ["1-a", "2-b"] + assert [e.id for e in session.events] == ["1-a", "2-b", "3-c", "4-d"] + # The older fetch anchors on the window's oldest timestamp. + call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args + assert call.kwargs["fetch_newer"] is False and call.kwargs["timestamp"] == 3000 + + assert session.fetch_older() == [] + assert [e.id for e in session.events] == ["1-a", "2-b", "3-c", "4-d"] + + +def test_log_session_fetch_newer_merges_delta_and_sorts_late_arrivals(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("1-a", 1000), _log_event("2-b", 2000)), # tail + # Look-behind re-delivers 2-b (held) plus a late arrival at 1500 and a fresh line. + _log_page(_log_event("15-l", 1500), _log_event("2-b", 2000), _log_event("3-c", 3000)), + ] + session = _session(api) + session.fetch_newer() + + delta = session.fetch_newer() + + assert [e.id for e in delta] == ["15-l", "3-c"] + assert [e.id for e in session.events] == ["1-a", "15-l", "2-b", "3-c"] + call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args + assert call.kwargs["fetch_newer"] is True and call.kwargs["timestamp"] == 2000 + + +def test_log_session_fetch_newer_empty_delta_leaves_window_unchanged(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("1-a", 1000)), + _log_page(), + ] + session = _session(api) + session.fetch_newer() + + assert session.fetch_newer() == [] + assert [e.id for e in session.events] == ["1-a"] + + +def test_log_session_seed_canonicalizes_and_anchors_follow_up_fetches(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page() + seed = [_log_event("2-b", 2000), _log_event("1-a", 1000), _log_event("2-b", 2000)] + session = _session(api, events=seed) + + assert [e.id for e in session.events] == ["1-a", "2-b"] + + session.fetch_newer() + call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args + assert call.kwargs["fetch_newer"] is True and call.kwargs["timestamp"] == 2000 + + +def test_log_session_events_returns_a_copy(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page(_log_event("1-a", 1000)) + session = _session(api) + session.fetch_older() + + view = session.events + view.append(_log_event("9-z", 9000)) + + assert [e.id for e in session.events] == ["1-a"] + + +def test_log_session_passes_max_lines_through(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page() + session = _session(api) + + session.fetch_older(max_lines=7) + assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args.kwargs["max_lines"] == 7