diff --git a/README.md b/README.md index 4440529..ffb3b09 100644 --- a/README.md +++ b/README.md @@ -55,20 +55,45 @@ 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`. `get_deployment_logs_range()` fetches a specific time window -(epoch-millisecond bounds, both optional) and, with `pod=None`, merges every pod's -stream chronologically. The same paging is available statelessly through -`get_deployment_logs(before=..., after=...)`, anchored on events you already hold -or on a bare epoch-millisecond boundary: +`fetch_logs()` is the one way to read deployment logs: it fetches one pod's stored +log lines within a time window (`start_time`/`end_time`, epoch ms, inclusive) and +yields them lazily, oldest first, as chunks of `DeploymentLogEvent` — each line at +most once, holding only a short dedup window in memory however long the stream. +`chunk_size` (1 to 5000) is the number of lines requested from the server per +round trip and each server page that carries window lines becomes one chunk, so +a bulk read of history wants a large `chunk_size` — the log read path is +rate-limited upstream, and a small `chunk_size` over a large window multiplies +requests. Discover pod names with `get_deployment_pods()` +(terminated pods still within log retention are included). `start_time` defaults +to the moment of the call; with `end_time` set the iterator terminates once the +window is delivered or the store has no more lines to give, whichever comes +first: + +```python +for chunk in cclient.fetch_logs(DEPLOYMENT_ID, REVISION, pod, start_time=t1_ms, end_time=t2_ms): + for event in chunk: + print(event.message) +``` -```bash -python examples/sdk/get_deployment_logs.py +Without `end_time` the same generator tails: it never terminates, and once caught +up it yields an empty chunk each time nothing new is stored yet — the caller +decides when to sleep or break: + +```python +import time + +for chunk in cclient.fetch_logs(DEPLOYMENT_ID, REVISION, pod): + if not chunk: + time.sleep(2) + continue + for event in chunk: + print(event.message) ``` +`python examples/sdk/get_deployment_logs.py` runs both. `get_deployment_logs()`, +`get_deployment_logs_range()` and `deployment_log_session()` still work but are +deprecated in favor of `fetch_logs()` and emit a `DeprecationWarning` on use. + ### Un-installation To uninstall `centml`, simply do: diff --git a/centml/sdk/api.py b/centml/sdk/api.py index afa8b26..3419652 100644 --- a/centml/sdk/api.py +++ b/centml/sdk/api.py @@ -1,7 +1,10 @@ +import random +import time from bisect import insort from contextlib import contextmanager from dataclasses import dataclass -from typing import List, Optional, Union +from functools import partial +from typing import Callable, Iterator, List, Optional, Union import platform_api_python_client from platform_api_python_client import ( @@ -18,6 +21,7 @@ InviteUserRequest, Metric, ) +from typing_extensions import deprecated from centml.sdk import auth from centml.sdk.config import settings @@ -26,9 +30,39 @@ DEFAULT_LOG_PAGE_LINES = 100 # server-side default for max_lines MAX_LOG_PAGE_LINES = 5000 # server-side ceiling for max_lines +# fetch_logs asks for less than the server's own default because the common call tails a +# running deployment, where the store has only a handful of new lines to hand over per poll +# whatever the page size is. Reading a backlog wants a far larger chunk_size. +DEFAULT_LOG_CHUNK_LINES = 10 # 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 +# The log read path is rate limited upstream on a bucket shared by every caller, and the +# API reports a saturated bucket the same way it reports a sick store: HTTP 503. +LOG_BUSY_STATUS = 503 +# Doubling from half a second spends about seven seconds over these attempts — long enough +# to outlast a bucket refill without looking hung. +LOG_RETRY_ATTEMPTS = 5 +LOG_RETRY_BASE_SECONDS = 0.5 +# Fraction of each backoff to vary it by: the bucket is shared, so clients backing off in +# lockstep would re-collide on it every round. +LOG_RETRY_JITTER = 0.25 + + +def _with_busy_retry(fetch_page: Callable[[], list]) -> list: + """Call fetch_page, retrying while the store reports itself busy. There is no + server-issued cursor, so a page request is a pure function of its anchor and + re-issuing it can neither duplicate nor skip lines.""" + for attempt in range(LOG_RETRY_ATTEMPTS - 1): + try: + return fetch_page() + except ApiException as exc: + if exc.status != LOG_BUSY_STATUS: + raise + backoff = LOG_RETRY_BASE_SECONDS * 2**attempt + time.sleep(backoff * (1 + random.uniform(-LOG_RETRY_JITTER, LOG_RETRY_JITTER))) + # The last attempt propagates whatever it raises. + return fetch_page() def _recent_anchor(events: list) -> list: @@ -42,6 +76,16 @@ def _recent_anchor(events: list) -> list: return events[first_recent:] +@dataclass(frozen=True) +class _LogAnchor: + """The two fields an after anchor uses: the id it dedupes by and the timestamp it + takes its boundary and its retention cutoff from. Holding these instead of whole + events keeps a long tail off the message text it has already handed out.""" + + id: str + timestamp: int + + @dataclass(frozen=True) class DeploymentLogEvent: """One log line with its pod attached — logs_v4 events carry no pod name, so @@ -230,6 +274,7 @@ def get_deployment_pods(self, deployment_id: int, revision_number: int) -> List[ ).pods # pylint: disable=R0917 + @deprecated("get_deployment_logs() is deprecated; use fetch_logs() instead") def get_deployment_logs( self, deployment_id: int, @@ -239,7 +284,9 @@ def get_deployment_logs( after: Optional[Union[list, int]] = None, max_lines: int = DEFAULT_LOG_PAGE_LINES, ) -> list: - """Fetch one page of a pod's logs, oldest-first. Use get_deployment_pods() to + """Deprecated: use fetch_logs() instead. + + 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 @@ -257,6 +304,22 @@ def get_deployment_logs( through undeduplicated. An empty anchor list raises ValueError. Pages never split a millisecond, so a delivered boundary millisecond is always complete. """ + return self._fetch_log_page( + deployment_id, revision_number, pod, before=before, after=after, max_lines=max_lines + ) + + # pylint: disable=R0917 + def _fetch_log_page( + self, + deployment_id: int, + revision_number: int, + pod: str, + before: Optional[Union[list, int]] = None, + after: Optional[Union[list, int]] = None, + max_lines: int = DEFAULT_LOG_PAGE_LINES, + ) -> list: + """The page primitive behind fetch_logs and the deprecated readers — the + contract get_deployment_logs() documents, without the deprecation warning.""" if before is not None and after is not None: raise ValueError("before and after are mutually exclusive") @@ -294,6 +357,7 @@ def get_deployment_logs( return [event for event in response.events if event.id not in held_event_ids] # pylint: disable=R0917 + @deprecated("get_deployment_logs_range() is deprecated; use fetch_logs() instead") def get_deployment_logs_range( self, deployment_id: int, @@ -302,7 +366,9 @@ def get_deployment_logs_range( start_time: Optional[int] = None, end_time: Optional[int] = None, ) -> List[DeploymentLogEvent]: - """Fetch every log line in [start_time, end_time] (epoch ms, inclusive; both + """Deprecated: use fetch_logs() instead. + + Fetch every log line in [start_time, end_time] (epoch ms, inclusive; both optional — an open end reads to the beginning or the present), oldest first. pod=None reads all pods of the revision and merges the streams chronologically; each returned event carries its pod name.""" @@ -317,7 +383,7 @@ def get_deployment_logs_range( # after is exclusive, so start_time - 1 admits lines at start_time itself; # start_time 0 (or None) means the whole window — scan from the head. anchor: Union[list, int] = _recent_anchor(events) if events else (start_time - 1 if start_time else 0) - page = self.get_deployment_logs( + page = self._fetch_log_page( deployment_id, revision_number, pod_name, after=anchor, max_lines=MAX_LOG_PAGE_LINES ) if not page: @@ -334,18 +400,156 @@ def get_deployment_logs_range( merged.sort(key=lambda event: event.id) return merged + # pylint: disable=R0917 + def fetch_logs( + self, + deployment_id: int, + revision_number: int, + pod: str, + start_time: Optional[int] = None, + end_time: Optional[int] = None, + chunk_size: int = DEFAULT_LOG_CHUNK_LINES, + ) -> Iterator[List[DeploymentLogEvent]]: + """Fetch one pod's stored log lines within [start_time, end_time] (epoch ms, + inclusive), yielded lazily oldest first as chunks of DeploymentLogEvent, + each stored line at most once. Discover pod names with + get_deployment_pods(). + + chunk_size is the number of lines requested from the server per round + trip (1 to MAX_LOG_PAGE_LINES), and each server page that carries window + lines becomes one yielded chunk, so a bulk read of history wants a large + chunk_size — the log read path is rate-limited upstream, and a small + chunk_size over a large window multiplies requests. A chunk usually holds + up to chunk_size lines but can be smaller (lines below start_time or + already delivered are filtered out of the page) or larger (the server + never splits one millisecond across pages, so a millisecond holding more + than chunk_size lines arrives whole). A page filtered away entirely + yields nothing at all rather than an empty chunk, which means only that + the stream is caught up. + + start_time defaults to the current time, resolved once when fetch_logs is + called (not at the first next()), so lines logged while the generator sits + unstarted are not skipped; pass an earlier start_time to read history. + + With end_time set the iterator terminates once the window is delivered or + the store has no more lines to give, whichever comes first — an end_time + in the future does not keep it polling until then. + Without end_time it never terminates — once caught up it yields an empty + chunk each time nothing new is stored yet, and the caller decides when to + sleep or break: + + for chunk in cclient.fetch_logs(dep, rev, pod): + if not chunk: + time.sleep(2) + continue + ... + + Nothing is fetched before the first next(), and memory is bounded by the + dedup window rather than by the length of the stream: the anchor holds + only the ids and timestamps of the last LOG_DEDUP_RETENTION_MS, a + generous margin over the server's re-delivery span, so no line is + delivered twice — across empty chunks too. + + A line the log store received late lands in a later chunk than its + timestamp position, never duplicated, as long as it lands inside that + re-delivery span (~15s). This reader only ever pages forward, and the + server re-delivers the span behind the boundary only; a line whose + timestamp falls further than the span behind the newest line already + delivered is never returned at all. Lines inside each chunk are always + in ascending (timestamp, id) order. + + A page request the store answers as busy (HTTP 503) is retried with + exponential backoff and jitter; the anchor survives the retry, so an + upstream rate limit costs a pause rather than the stream. + + If a page request fails for any other reason, or the retries run out, + the iterator raises and, like any generator, cannot be resumed — but + every chunk already yielded is complete and none is left half-built. + Resume with a new fetch_logs whose start_time is the last delivered + event's timestamp: bounds are inclusive, so the only lines delivered + again are the ones sharing that millisecond, which the caller already + holds. + """ + if not 1 <= chunk_size <= MAX_LOG_PAGE_LINES: + raise ValueError( + f"chunk_size must be between 1 and {MAX_LOG_PAGE_LINES} lines " + "(chunk_size is also the per-request page size, and the server " + f"caps max_lines at {MAX_LOG_PAGE_LINES})" + ) + if (start_time is not None and start_time < 0) or (end_time is not None and end_time < 0): + raise ValueError("start_time and end_time are epoch milliseconds and must not be negative") + if start_time is not None and end_time is not None and start_time > end_time: + raise ValueError("start_time must not exceed end_time") + start_ms = int(time.time() * 1000) if start_time is None else start_time + + def chunks() -> Iterator[List[DeploymentLogEvent]]: + held: list = [] + # after is exclusive, so start_ms - 1 admits lines at start_ms itself. + initial_boundary = max(start_ms - 1, 0) + while True: + anchor: Union[list, int] = _recent_anchor(held) if held else initial_boundary + page = _with_busy_retry( + partial( + self._fetch_log_page, deployment_id, revision_number, pod, after=anchor, max_lines=chunk_size + ) + ) + past_end = False + chunk: List[DeploymentLogEvent] = [] + for raw in page: + anchor_event = _LogAnchor(id=raw.id, timestamp=raw.timestamp) + if held and raw.id <= held[-1].id: + # Late arrival inside the look-behind span: keep the held window + # id-ordered (id order == time order) so trimming stays correct. + insort(held, anchor_event, key=lambda held_event: held_event.id) + else: + held.append(anchor_event) + # Anchoring at start_ms - 1 re-delivers the look-behind span below + # start_ms; those ids must be held for dedup but never emitted. + if raw.timestamp < start_ms: + continue + if end_time is not None and raw.timestamp > end_time: + past_end = True + break + event = DeploymentLogEvent(id=raw.id, timestamp=raw.timestamp, message=raw.message, pod=pod) + # The server orders a page by nanosecond timestamp only, never by + # the id's hash suffix, so lines sharing one nanosecond can arrive + # in either id order; insort keeps every chunk ascending. + if chunk and (event.timestamp, event.id) < (chunk[-1].timestamp, chunk[-1].id): + insort(chunk, event, key=lambda chunk_event: (chunk_event.timestamp, chunk_event.id)) + else: + chunk.append(event) + if page: + held = _recent_anchor(held) + if chunk: + yield chunk + if past_end or (not page and end_time is not None): + break + if not page: + # Caught up with no end bound: signal "nothing new yet" until + # new lines are stored. + yield [] + + # The nested generator closes over the validated arguments, so the + # ValueErrors above raise at the call rather than at the first next(). + return chunks() + + @deprecated("deployment_log_session() is deprecated; use fetch_logs() instead") 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 + """Deprecated: use fetch_logs() instead. + + 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) +@deprecated("DeploymentLogSession is deprecated; use CentMLClient.fetch_logs() instead") class DeploymentLogSession: - """Maintains a contiguous, ordered window of one pod's logs across fetches. + """Deprecated: use CentMLClient.fetch_logs() instead. + 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). @@ -372,7 +576,7 @@ 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( + page = self._client._fetch_log_page( self._deployment_id, self._revision_number, self._pod, @@ -390,7 +594,7 @@ def fetch_newer(self, max_lines: int = DEFAULT_LOG_PAGE_LINES) -> list: 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( + delta = self._client._fetch_log_page( self._deployment_id, self._revision_number, self._pod, diff --git a/examples/sdk/get_deployment_logs.py b/examples/sdk/get_deployment_logs.py index 0dd7ec5..ad56d74 100644 --- a/examples/sdk/get_deployment_logs.py +++ b/examples/sdk/get_deployment_logs.py @@ -6,8 +6,9 @@ # --- Configuration --- DEPLOYMENT_ID = 1234 # Replace with your deployment ID REVISION_NUMBER = 10 -TAIL_SECONDS = 30 # How long to keep polling for new lines after reading history -TAIL_LINES = 20 # How much history to print before tailing +WINDOW_MINUTES = 10 # How far back the window read looks +TAIL_LINES = 20 # How many tailed lines to print before stopping the tail loop +POLL_SECONDS = 2.0 def format_event(event) -> str: @@ -17,44 +18,58 @@ def format_event(event) -> str: def main(): with get_centml_client() as cclient: - # Logs are read per pod: discover the pods that have logged for this revision - # (terminated pods within log retention are included). + # Discover pod names; terminated pods still 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 - events = session.events - print(f"Found {len(events)} log entries; showing the last {TAIL_LINES}:\n") - for event in events[-TAIL_LINES:]: - 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(): + # One call reads one pod, so here is the rest of the revision's roster. + print(f"Pods with logs: {', '.join(pods)}\n") + + # A window read: start_time/end_time are epoch ms, inclusive. With end_time + # set the iterator terminates once the window is delivered or the store has + # no more lines to give. fetch_logs is lazy — each server page that carries + # window lines is yielded as one chunk, and only a short dedup window is held + # however large the read. chunk_size is also the number of lines requested per + # round trip, so a bulk read wants a large value. + now_ms = int(time.time() * 1000) + print(f"Last {WINDOW_MINUTES} minutes of pod {pod}:\n") + count = 0 + for chunk in cclient.fetch_logs( + DEPLOYMENT_ID, + REVISION_NUMBER, + pod, + start_time=now_ms - WINDOW_MINUTES * 60_000, + end_time=now_ms, + chunk_size=1000, + ): + for event in chunk: + print(format_event(event)) + count += len(chunk) + print(f"\nThe window holds {count} lines.") + + # A tail: without end_time the same generator never terminates. start_time + # defaults to the moment of the call, and once caught up the generator + # yields an empty chunk each time nothing new is stored yet — the caller + # decides when to sleep or break. No line is ever delivered twice. + # + # Passing an earlier start_time here rather than omitting it delivers the + # backlog first and then follows, in one pass. That is the shape to reach + # for when both are wanted: a tail started after a separate window read + # begins at its own "now", losing whatever was logged in between. + print(f"\nTailing pod {pod}; stopping after {TAIL_LINES} new lines...") + printed = 0 + for chunk in cclient.fetch_logs(DEPLOYMENT_ID, REVISION_NUMBER, pod): + if not chunk: + time.sleep(POLL_SECONDS) + continue + for event in chunk: print(format_event(event)) - time.sleep(2) - - # 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) # empty return = beginning - # newer = cclient.get_deployment_logs(..., pod=pod, after=page) # empty return = nothing new - # A specific time window (all pods merged, oldest first, pod on each event): - # window = cclient.get_deployment_logs_range( - # DEPLOYMENT_ID, REVISION_NUMBER, start_time=t1_ms, end_time=t2_ms - # ) + printed += len(chunk) + if printed >= TAIL_LINES: + break if __name__ == "__main__": diff --git a/requirements.txt b/requirements.txt index 6629fff..1ccea02 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,3 +7,4 @@ websockets>=16.0 pyte>=0.8.0 platform-api-python-client==4.28.0 click>=8.4.1 +typing-extensions>=4.5 diff --git a/tests/test_sdk_api.py b/tests/test_sdk_api.py index ab1c55d..11d8ae1 100644 --- a/tests/test_sdk_api.py +++ b/tests/test_sdk_api.py @@ -1,3 +1,4 @@ +import warnings from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -13,7 +14,14 @@ ) from centml.sdk import ApiException -from centml.sdk.api import CentMLClient, get_centml_client +from centml.sdk.api import ( + LOG_DEDUP_RETENTION_MS, + LOG_RETRY_ATTEMPTS, + MAX_LOG_PAGE_LINES, + CentMLClient, + DeploymentLogSession, + get_centml_client, +) from centml.sdk.config import settings @@ -270,7 +278,8 @@ def test_get_deployment_logs_returns_tail_page_when_unanchored(): ) client = CentMLClient(api) - events = client.get_deployment_logs(123, 2, pod="pod-a") + with pytest.warns(DeprecationWarning): + 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( @@ -284,7 +293,8 @@ def test_get_deployment_logs_before_pages_older_from_oldest_anchor(): 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) + with pytest.warns(DeprecationWarning): + 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 @@ -298,7 +308,8 @@ def test_get_deployment_logs_before_empty_page_signals_beginning_of_history(): 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)]) == [] + with pytest.warns(DeprecationWarning): + 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(): @@ -307,7 +318,8 @@ def test_get_deployment_logs_after_fetches_newer_from_newest_anchor(): 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) + with pytest.warns(DeprecationWarning): + 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 @@ -324,7 +336,8 @@ def test_get_deployment_logs_after_drops_redelivered_lines_but_keeps_late_arriva 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) + with pytest.warns(DeprecationWarning): + events = client.get_deployment_logs(123, 2, pod="pod-a", after=held) assert [e.id for e in events] == ["15-l", "3-c"] @@ -334,7 +347,8 @@ def test_get_deployment_logs_zero_after_anchor_reads_from_head(): 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=0) + with pytest.warns(DeprecationWarning): + events = client.get_deployment_logs(123, 2, pod="pod-a", after=0) assert [e.id for e in events] == ["1-a"] call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args @@ -347,7 +361,7 @@ def test_get_deployment_logs_rejects_empty_anchor_lists(): client = CentMLClient(api) for kwargs in ({"before": []}, {"after": []}): - with pytest.raises(ValueError): + with pytest.warns(DeprecationWarning), pytest.raises(ValueError): client.get_deployment_logs(123, 2, pod="pod-a", **kwargs) api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_not_called() @@ -358,7 +372,8 @@ def test_get_deployment_logs_after_empty_page_signals_nothing_new(): 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)]) == [] + with pytest.warns(DeprecationWarning): + 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(): @@ -366,7 +381,8 @@ def test_get_deployment_logs_passes_max_lines_through(): 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) + with pytest.warns(DeprecationWarning): + 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 @@ -376,14 +392,15 @@ def test_get_deployment_logs_rejects_before_and_after_together(): api = MagicMock() client = CentMLClient(api) - with pytest.raises(ValueError): + with pytest.warns(DeprecationWarning), pytest.raises(ValueError): client.get_deployment_logs(123, 2, pod="pod-a", before=[_log_event("1-a", 1000)], after=0) 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) + with pytest.warns(DeprecationWarning): + return CentMLClient(api).deployment_log_session(123, 2, "pod-a", events=events) def test_log_session_first_fetch_is_tail_for_both_directions(): @@ -493,12 +510,14 @@ def test_get_deployment_logs_accepts_timestamp_anchors(): api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page(_log_event("2-b", 2000)) client = CentMLClient(api) - events = client.get_deployment_logs(123, 2, pod="pod-a", after=1999) + with pytest.warns(DeprecationWarning): + events = client.get_deployment_logs(123, 2, pod="pod-a", after=1999) assert [e.id for e in events] == ["2-b"] 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"] == 1999 - client.get_deployment_logs(123, 2, pod="pod-a", before=5000) + with pytest.warns(DeprecationWarning): + client.get_deployment_logs(123, 2, pod="pod-a", before=5000) 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"] == 5000 @@ -512,7 +531,8 @@ def test_get_deployment_logs_range_trims_to_window(): ] client = CentMLClient(api) - events = client.get_deployment_logs_range(123, 2, pod="pod-a", start_time=1000, end_time=3000) + with pytest.warns(DeprecationWarning): + events = client.get_deployment_logs_range(123, 2, pod="pod-a", start_time=1000, end_time=3000) assert [(e.id, e.pod) for e in events] == [("1-a", "pod-a"), ("2-b", "pod-a"), ("3-c", "pod-a")] calls = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list @@ -530,7 +550,8 @@ def test_get_deployment_logs_range_open_ended_reads_full_history(): ] client = CentMLClient(api) - events = client.get_deployment_logs_range(123, 2, pod="pod-a") + with pytest.warns(DeprecationWarning): + events = client.get_deployment_logs_range(123, 2, pod="pod-a") assert [e.id for e in events] == ["1-a", "2-b"] calls = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list @@ -553,7 +574,8 @@ def pages(**kwargs): api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = pages client = CentMLClient(api) - events = client.get_deployment_logs_range(123, 2) + with pytest.warns(DeprecationWarning): + events = client.get_deployment_logs_range(123, 2) assert [(e.id, e.pod, e.message) for e in events] == [ ("1-a", "pod-a", "line"), @@ -568,7 +590,8 @@ def test_get_deployment_logs_range_returns_empty_when_no_pod_has_logged(): api.get_deployment_pods_deployments_pods_deployment_id_revision_number_get.return_value = SimpleNamespace(pods=[]) client = CentMLClient(api) - assert client.get_deployment_logs_range(123, 2) == [] + with pytest.warns(DeprecationWarning): + assert client.get_deployment_logs_range(123, 2) == [] api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_not_called() @@ -581,7 +604,8 @@ def test_get_deployment_logs_range_single_millisecond_window(): ] client = CentMLClient(api) - events = client.get_deployment_logs_range(123, 2, pod="pod-a", start_time=2000, end_time=2000) + with pytest.warns(DeprecationWarning): + events = client.get_deployment_logs_range(123, 2, pod="pod-a", start_time=2000, end_time=2000) assert [e.id for e in events] == ["2-b"] calls = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list @@ -592,7 +616,7 @@ def test_get_deployment_logs_range_rejects_inverted_window(): api = MagicMock() client = CentMLClient(api) - with pytest.raises(ValueError): + with pytest.warns(DeprecationWarning), pytest.raises(ValueError): client.get_deployment_logs_range(123, 2, pod="pod-a", start_time=2000, end_time=1000) api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_not_called() @@ -633,3 +657,481 @@ def test_log_session_long_window_keeps_boundary_and_dedup_correct(): session.fetch_older() 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"] == 1 + + +def _flatten(chunks): + return [event for chunk in chunks for event in chunk] + + +def _replaying_log_server(all_events, look_behind_ms=15_000, page_cap=None): + """Mimic the server's fetch-newer contract: newer-than-boundary lines up to + max_lines, plus the re-delivered look-behind span at and before the boundary + (uncounted). page_cap simulates a server that fills pages with fewer lines + than asked.""" + + def respond(**kwargs): + limit = min(kwargs["max_lines"], page_cap or kwargs["max_lines"]) + boundary = kwargs["timestamp"] + newer = [e for e in all_events if e.timestamp > boundary][:limit] + look_behind = [e for e in all_events if boundary - look_behind_ms < e.timestamp <= boundary] + return _log_page(*sorted(look_behind + newer, key=lambda e: e.id)) + + return respond + + +def test_fetch_logs_yields_first_chunk_before_fetching_the_next_page(): + 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)), + _log_page(_log_event("3-c", 3000)), + _log_page(), + ] + client = CentMLClient(api) + + stream = client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000, chunk_size=2) + first = next(stream) + + assert [(e.id, e.pod) for e in first] == [("1-a", "pod-a"), ("2-b", "pod-a")] + assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_count == 1 + # The first fetch never passes an (invalid) empty anchor list: it is a bare + # int boundary just below start_time (after is exclusive). + first_call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list[0] + assert first_call.kwargs["fetch_newer"] is True and first_call.kwargs["timestamp"] == 0 + + assert [[e.id for e in chunk] for chunk in stream] == [["3-c"]] + assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_count == 3 + + +def test_fetch_logs_requests_the_callers_chunk_size_as_max_lines(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("1-a", 1000)), + _log_page(), + ] + client = CentMLClient(api) + + list(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000, chunk_size=7)) + + # chunk_size is the server page size: every request carries it as max_lines. + calls = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list + assert [call.kwargs["max_lines"] for call in calls] == [7, 7] + + +def test_fetch_logs_millisecond_burst_larger_than_chunk_size_arrives_whole(): + # The server never splits one millisecond across pages, so a millisecond + # holding more than chunk_size lines comes back — and is yielded — whole. + burst = [_log_event(f"1000-{suffix}", 1000) for suffix in "abcdefg"] + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [_log_page(*burst), _log_page()] + client = CentMLClient(api) + + chunks = list(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000, chunk_size=3)) + + assert [len(chunk) for chunk in chunks] == [7] + assert [e.id for e in chunks[0]] == [e.id for e in burst] + + +def test_fetch_logs_chunk_comes_back_short_when_the_page_is_filtered(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + # The first page re-delivers the look-behind span below start_time; those + # lines are held for dedup but not emitted, so the chunk is short. + _log_page( + _log_event("04990-w", 4990), + _log_event("04995-x", 4995), + _log_event("05000-y", 5000), + _log_event("06000-z", 6000), + ), + _log_page(), + ] + client = CentMLClient(api) + + chunks = list(client.fetch_logs(123, 2, pod="pod-a", start_time=5000, end_time=10_000, chunk_size=4)) + + assert [[e.id for e in chunk] for chunk in chunks] == [["05000-y", "06000-z"]] + + +def test_fetch_logs_start_time_defaults_to_now_resolved_at_the_call(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page() + client = CentMLClient(api) + + with patch("centml.sdk.api.time.time", return_value=5000.0): + stream = client.fetch_logs(123, 2, pod="pod-a") + # "now" is pinned at the call, not the first next(): iterating under a later + # clock still anchors just below the call-time millisecond (after is exclusive). + with patch("centml.sdk.api.time.time", return_value=9000.0): + assert next(stream) == [] + 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"] == 5_000_000 - 1 + + +def test_fetch_logs_end_time_truncates_the_window_and_stops_fetching(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = _replaying_log_server( + [_log_event(f"{1000 * i}-x", 1000 * i) for i in range(1, 9)] + ) + client = CentMLClient(api) + + events = _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=2000, end_time=5000)) + + assert [e.id for e in events] == ["2000-x", "3000-x", "4000-x", "5000-x"] + # The page carrying lines past end_time already proves the window is complete: + # no further request is issued. + assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_count == 1 + + +def test_fetch_logs_bounded_read_terminates_when_caught_up(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("1-a", 1000)), + _log_page(), + ] + client = CentMLClient(api) + + # end_time still in the future: an empty page means the store is caught up, + # and with an end bound set that terminates the read. + events = _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000)) + + assert [e.id for e in events] == ["1-a"] + assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_count == 2 + + +def test_fetch_logs_open_ended_yields_empty_chunks_and_resumes_without_duplicates(): + 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)), + _log_page(), # caught up: "nothing new yet" + _log_page(), # still nothing + _log_page(_log_event("2-b", 2000), _log_event("3-c", 3000)), # look-behind re-delivers 2-b + _log_page(), + ] + client = CentMLClient(api) + + stream = client.fetch_logs(123, 2, pod="pod-a", start_time=1) + + # Without end_time the generator never returns: an empty chunk is the + # "nothing new yet" signal and the caller decides when to sleep or break. + assert [e.id for e in next(stream)] == ["1-a", "2-b"] + assert next(stream) == [] + assert next(stream) == [] + # It resumes delivering once new lines are stored, and the line re-delivered + # inside the look-behind span never crosses the empty-chunk boundary twice. + assert [e.id for e in next(stream)] == ["3-c"] + assert next(stream) == [] + + +def test_fetch_logs_does_not_livelock_on_look_behind_redelivery(): + # A bare int boundary alone would never terminate: the re-delivered look-behind + # span keeps every page non-empty and the boundary never advances past it. The + # held anchor dedupes the span away, so the bounded read terminates. + all_events = [_log_event(f"{1000 + i:05d}-x", 1000 + i) for i in range(30)] + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = _replaying_log_server( + all_events, page_cap=10 + ) + client = CentMLClient(api) + + events = _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=100_000)) + + assert [e.id for e in events] == [e.id for e in all_events] # every line once, in order + # 3 data pages + 1 empty page proving catch-up; a livelock would exceed this. + assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_count == 4 + + +def test_fetch_logs_start_time_holds_look_behind_lines_below_the_window(): + all_events = [ + _log_event("04990-w", 4990), + _log_event("04995-x", 4995), + _log_event("05000-y", 5000), + _log_event("06000-z", 6000), + ] + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = _replaying_log_server(all_events) + client = CentMLClient(api) + + events = _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=5000, end_time=100_000)) + + # The look-behind lines below start_time are held for dedup but never emitted. + assert [e.id for e in events] == ["05000-y", "06000-z"] + first_call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list[0] + assert first_call.kwargs["timestamp"] == 4999 # after is exclusive: admits start_time itself + + +def test_fetch_logs_page_filtered_away_entirely_yields_nothing_not_an_empty_chunk(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("04990-w", 4990), _log_event("04995-x", 4995)), + _log_page(_log_event("06000-z", 6000)), + ] + client = CentMLClient(api) + + stream = client.fetch_logs(123, 2, pod="pod-a", start_time=5000) + + # An empty chunk means "caught up", so a page holding only look-behind lines + # below start_time must fetch again rather than claim the stream is idle. + assert [e.id for e in next(stream)] == ["06000-z"] + assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_count == 2 + + +def test_fetch_logs_held_state_stays_within_the_dedup_window(): + step_ms = 10_000 + pages = [ + _log_page(*(_log_event(f"{(p * 100 + i) * step_ms:09d}-x", (p * 100 + i) * step_ms) for i in range(100))) + for p in range(1, 50) + ] + [_log_page()] + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = pages + client = CentMLClient(api) + + anchor_sizes = [] + original = CentMLClient._fetch_log_page + + def spying_fetch_log_page(self, *args, **kwargs): + if isinstance(kwargs.get("after"), list): + anchor_sizes.append(len(kwargs["after"])) + return original(self, *args, **kwargs) + + with patch.object(CentMLClient, "_fetch_log_page", spying_fetch_log_page): + events = _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10**12)) + + assert len(events) == 4900 + # Held state is the trimmed dedup window, not the accumulated stream. + assert max(anchor_sizes) <= LOG_DEDUP_RETENTION_MS // step_ms + 1 + + +def test_fetch_logs_single_millisecond_window(): + all_events = [_log_event(f"{1000 * i:05d}-x", 1000 * i) for i in range(1, 4)] + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = _replaying_log_server(all_events) + client = CentMLClient(api) + + events = _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=2000, end_time=2000)) + + assert [e.id for e in events] == ["02000-x"] + + +def test_fetch_logs_late_arrival_lands_in_a_later_chunk_still_ascending(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("01000-a", 1000), _log_event("02000-b", 2000), _log_event("03000-c", 3000)), + # A late arrival re-delivered inside the look-behind span lands in the + # chunk of the page that carried it, below that page's fresh lines. + _log_page(_log_event("02500-l", 2500), _log_event("04000-d", 4000)), + _log_page(), + ] + client = CentMLClient(api) + + chunks = list(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000, chunk_size=3)) + + assert [[e.id for e in chunk] for chunk in chunks] == [["01000-a", "02000-b", "03000-c"], ["02500-l", "04000-d"]] + + +def test_fetch_logs_chunk_stays_ascending_when_the_server_ties_on_timestamp(): + # The server sorts a page by nanosecond timestamp only, so two lines sharing + # one nanosecond can arrive in either id order; the chunk must still come out + # ascending in (timestamp, id). + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("1000-b", 1000), _log_event("1000-a", 1000), _log_event("2000-c", 2000)), + _log_page(), + ] + client = CentMLClient(api) + + chunks = list(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000, chunk_size=5)) + + assert [[e.id for e in chunk] for chunk in chunks] == [["1000-a", "1000-b", "2000-c"]] + + +def test_fetch_logs_validates_eagerly_at_the_call_not_the_first_next(): + api = MagicMock() + client = CentMLClient(api) + + # Each ValueError is raised by the call itself — never deferred to next() — + # so a stored or passed-around iterator cannot surface it far from the bad call. + for kwargs in ( + {"start_time": 2000, "end_time": 1000}, + {"start_time": -1}, + {"end_time": -1}, + {"chunk_size": 0}, + # chunk_size is the on-the-wire page size, so it inherits the server's + # ceiling — rejected here, not by a generated-model pydantic error. + {"chunk_size": MAX_LOG_PAGE_LINES + 1}, + ): + with pytest.raises(ValueError): + client.fetch_logs(123, 2, pod="pod-a", **kwargs) + + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_not_called() + + +def test_fetch_logs_accepts_both_ends_of_the_chunk_size_range(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page() + client = CentMLClient(api) + + for chunk_size in (1, MAX_LOG_PAGE_LINES): + list(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000, chunk_size=chunk_size)) + + sent = [ + call.kwargs["max_lines"] + for call in api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list + ] + assert sent == [1, MAX_LOG_PAGE_LINES] + + +def test_fetch_logs_start_time_zero_clamps_the_boundary(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page() + client = CentMLClient(api) + + list(client.fetch_logs(123, 2, pod="pod-a", start_time=0, end_time=10_000)) + + # after is exclusive, but the server rejects a negative boundary, so start_time 0 + # clamps to 0 — which already admits every stored line. + first_call = api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_args_list[0] + assert first_call.kwargs["timestamp"] == 0 + + +def test_fetch_logs_failed_page_leaves_delivered_chunks_whole(): + 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)), + ApiException(status=400), + ] + client = CentMLClient(api) + + stream = client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000, chunk_size=2) + + assert [e.id for e in next(stream)] == ["1-a", "2-b"] + with pytest.raises(ApiException): + next(stream) + # The failure ends the iterator, as it would any generator; the caller resumes + # with a new fetch_logs anchored on the last event it holds. + with pytest.raises(StopIteration): + next(stream) + + +def test_fetch_logs_survives_a_busy_store_without_losing_its_anchor(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("1-a", 1000)), + ApiException(status=503), + _log_page(_log_event("1-a", 1000), _log_event("2-b", 2000)), + _log_page(), + ] + client = CentMLClient(api) + + with patch("centml.sdk.api.time.sleep"): + events = _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000)) + + # The retry re-issues the same anchored request, so the line it re-delivers is + # deduped rather than emitted twice, and the stream survives the rate limit. + assert [e.id for e in events] == ["1-a", "2-b"] + + +def test_fetch_logs_gives_up_on_a_busy_store_after_the_retry_budget(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("1-a", 1000)) + ] + [ApiException(status=503)] * LOG_RETRY_ATTEMPTS + client = CentMLClient(api) + + stream = client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000, chunk_size=1) + assert [e.id for e in next(stream)] == ["1-a"] + + with patch("centml.sdk.api.time.sleep") as sleep: + with pytest.raises(ApiException): + next(stream) + + # The whole budget goes on the one page, and the chunk already delivered stands. + assert sleep.call_count == LOG_RETRY_ATTEMPTS - 1 + + +def test_fetch_logs_backs_off_further_on_each_busy_answer(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ApiException(status=503)] * ( + LOG_RETRY_ATTEMPTS - 1 + ) + [_log_page()] + client = CentMLClient(api) + + with patch("centml.sdk.api.time.sleep") as sleep: + assert _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000)) == [] + + waited = [call.args[0] for call in sleep.call_args_list] + assert len(waited) == LOG_RETRY_ATTEMPTS - 1 + # Each wait is longer than the one before it, so a store that stays busy is asked + # less and less often rather than hammered at a fixed interval. The last wait also + # dwarfs the first: a fixed interval, however jittered, could not span that, while + # doubling across these attempts clears it with room to spare. + assert all(earlier < later for earlier, later in zip(waited, waited[1:])) + assert waited[-1] > waited[0] * 4 + + +def test_fetch_logs_does_not_retry_a_request_the_store_rejects(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = ApiException(status=404) + client = CentMLClient(api) + + with patch("centml.sdk.api.time.sleep") as sleep: + with pytest.raises(ApiException): + next(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000)) + + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_called_once() + sleep.assert_not_called() + + +def test_fetch_logs_is_lazy_until_the_first_next(): + 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) + + stream = client.fetch_logs(123, 2, pod="pod-a", start_time=1, chunk_size=1) + + # Nothing is fetched before the first next(). + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_not_called() + + next(stream) + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_called_once() + + +def test_fetch_logs_does_not_warn_on_its_own_internal_calls(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ + _log_page(_log_event("1-a", 1000)), + _log_page(), + ] + client = CentMLClient(api) + + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + events = _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000)) + + assert len(events) == 1 + + +def test_deprecated_log_readers_warn_and_name_the_replacement(): + api = MagicMock() + api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.return_value = _log_page() + api.get_deployment_pods_deployments_pods_deployment_id_revision_number_get.return_value = SimpleNamespace(pods=[]) + client = CentMLClient(api) + + with pytest.warns(DeprecationWarning, match="fetch_logs"): + client.get_deployment_logs(123, 2, pod="pod-a") + with pytest.warns(DeprecationWarning, match="fetch_logs"): + client.get_deployment_logs_range(123, 2) + with pytest.warns(DeprecationWarning, match="fetch_logs") as caught: + session = client.deployment_log_session(123, 2, "pod-a") + # Two warnings, one per deprecated surface the call crosses: the factory the + # caller named, and the session class it hands back for the caller to keep using. + assert [str(warning.message) for warning in caught] == [ + "deployment_log_session() is deprecated; use fetch_logs() instead", + "DeploymentLogSession is deprecated; use CentMLClient.fetch_logs() instead", + ] + with pytest.warns(DeprecationWarning, match="fetch_logs"): + DeploymentLogSession(client, 123, 2, "pod-a") + + # The deprecated paths still work, and a constructed session fetches without + # re-warning on the SDK's own internal page calls. + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + assert session.fetch_older() == []