|
| 1 | +from bisect import insort |
1 | 2 | from contextlib import contextmanager |
| 3 | +from typing import List, Optional |
2 | 4 |
|
3 | 5 | import platform_api_python_client |
4 | 6 | from platform_api_python_client import ( |
|
21 | 23 |
|
22 | 24 | STATUS_V3_DEPLOYMENT_TYPES = {DeploymentType.INFERENCE_V3, DeploymentType.CSERVE_V3} |
23 | 25 |
|
| 26 | +DEFAULT_LOG_PAGE_LINES = 100 # server-side default for max_lines |
| 27 | +# The server re-delivers a ~15s look-behind window on fetch-newer requests; only the |
| 28 | +# caller's events within this generous margin of the boundary can be re-delivered. |
| 29 | +LOG_DEDUP_RETENTION_MS = 300_000 |
| 30 | + |
24 | 31 |
|
25 | 32 | class CentMLClient: |
26 | 33 | def __init__(self, api): |
@@ -191,46 +198,127 @@ def get_deployment_revisions(self, deployment_id: int): |
191 | 198 | deployment_id=deployment_id |
192 | 199 | ).results |
193 | 200 |
|
| 201 | + def get_deployment_pods(self, deployment_id: int, revision_number: int) -> List[str]: |
| 202 | + """List pods that have logged for a deployment revision, including terminated |
| 203 | + pods still within log retention. A fresh deployment may return an empty list.""" |
| 204 | + return self._api.get_deployment_pods_deployments_pods_deployment_id_revision_number_get( |
| 205 | + deployment_id=deployment_id, revision_number=revision_number |
| 206 | + ).pods |
| 207 | + |
| 208 | + # pylint: disable=R0917 |
194 | 209 | def get_deployment_logs( |
195 | 210 | self, |
196 | 211 | deployment_id: int, |
197 | 212 | revision_number: int, |
198 | | - start_time: int, |
199 | | - end_time: int, |
200 | | - line_count: int = 100, |
201 | | - start_from_head: bool = True, |
202 | | - stream: bool = False, |
203 | | - ): |
204 | | - """Fetch logs for a deployment within a time window, handling pagination automatically. |
205 | | -
|
206 | | - start_time and end_time are Unix timestamps in milliseconds. |
207 | | - Use get_deployment_revisions() to find the current revision number. |
208 | | -
|
209 | | - If stream=True, returns a generator that yields events as each page is fetched. |
210 | | - If stream=False (default), returns a flat list of all events. |
| 213 | + pod: str, |
| 214 | + before: Optional[list] = None, |
| 215 | + after: Optional[list] = None, |
| 216 | + max_lines: int = DEFAULT_LOG_PAGE_LINES, |
| 217 | + ) -> list: |
| 218 | + """Fetch one page of a pod's logs, oldest-first. Use get_deployment_pods() to |
| 219 | + discover pod names and get_deployment_revisions() for the revision number. |
| 220 | +
|
| 221 | + before and after anchor the page to events a previous call returned for the |
| 222 | + same pod (pass your accumulated list; only the relevant boundary is used): |
| 223 | + - neither: the newest page (tail). |
| 224 | + - before=<events>: the page strictly older than the oldest of them; |
| 225 | + an empty result means the beginning of history is reached. |
| 226 | + - after=<events>: lines strictly newer than the newest of them; an empty |
| 227 | + result means nothing new yet — call again later to keep tailing. Late |
| 228 | + lines still landing near that boundary are included on top of max_lines |
| 229 | + and may sort below events you already hold (order by id if that matters). |
211 | 230 | """ |
| 231 | + if before is not None and after is not None: |
| 232 | + raise ValueError("before and after are mutually exclusive") |
| 233 | + |
| 234 | + fetch_newer = after is not None |
| 235 | + anchor_events = after if fetch_newer else before |
| 236 | + boundary_timestamp = None |
| 237 | + if anchor_events: |
| 238 | + timestamps = [event.timestamp for event in anchor_events] |
| 239 | + boundary_timestamp = max(timestamps) if fetch_newer else min(timestamps) |
| 240 | + |
| 241 | + response = self._api.get_deployment_logs_v4_logs_deployment_id_revision_number_get( |
| 242 | + deployment_id=deployment_id, |
| 243 | + revision_number=revision_number, |
| 244 | + pod=pod, |
| 245 | + fetch_newer=fetch_newer, |
| 246 | + timestamp=boundary_timestamp, |
| 247 | + max_lines=max_lines, |
| 248 | + ) |
| 249 | + if not fetch_newer or not anchor_events: |
| 250 | + return response.events |
| 251 | + |
| 252 | + # fetch_newer re-delivers a look-behind window at and before the boundary |
| 253 | + # (late-arrival protection); drop the lines the caller already holds by id. |
| 254 | + cutoff = max(event.timestamp for event in anchor_events) - LOG_DEDUP_RETENTION_MS |
| 255 | + held_event_ids = {event.id for event in anchor_events if event.timestamp >= cutoff} |
| 256 | + return [event for event in response.events if event.id not in held_event_ids] |
| 257 | + |
| 258 | + def deployment_log_session( |
| 259 | + self, deployment_id: int, revision_number: int, pod: str, events: Optional[list] = None |
| 260 | + ) -> "DeploymentLogSession": |
| 261 | + """Stateful reader for one pod's logs that tracks fetched pages and anchors |
| 262 | + every request itself — see DeploymentLogSession. Seed events with logs a |
| 263 | + previous session (or get_deployment_logs) returned for the same pod.""" |
| 264 | + return DeploymentLogSession(self, deployment_id, revision_number, pod, events) |
| 265 | + |
| 266 | + |
| 267 | +class DeploymentLogSession: |
| 268 | + """Maintains a contiguous, ordered window of one pod's logs across fetches. |
| 269 | +
|
| 270 | + Every fetch is anchored on the window itself, so pages can never overlap or |
| 271 | + leave gaps inside it (within log retention; an undetectable gap forms if the |
| 272 | + session idles past retention before fetching newer lines). |
| 273 | + """ |
212 | 274 |
|
213 | | - def _iter_events(): |
214 | | - next_page_token = None |
215 | | - while True: |
216 | | - response = self._api.get_deployment_logs_v3_deployments_logs_v3_deployment_id_revision_number_get( |
217 | | - deployment_id=deployment_id, |
218 | | - revision_number=revision_number, |
219 | | - start_time=start_time, |
220 | | - end_time=end_time, |
221 | | - next_page_token=next_page_token, |
222 | | - start_from_head=start_from_head, |
223 | | - line_count=line_count, |
224 | | - ) |
225 | | - yield from response.events |
226 | | - next_page_token = response.next_page_token |
227 | | - if not next_page_token: |
228 | | - break |
229 | | - |
230 | | - if stream: |
231 | | - return _iter_events() |
232 | | - |
233 | | - return list(_iter_events()) |
| 275 | + # pylint: disable=R0917 |
| 276 | + def __init__(self, client: CentMLClient, deployment_id: int, revision_number: int, pod: str, events=None): |
| 277 | + self._client = client |
| 278 | + self._deployment_id = deployment_id |
| 279 | + self._revision_number = revision_number |
| 280 | + self._pod = pod |
| 281 | + # Seeded events come from outside the session: canonicalize to unique ids in |
| 282 | + # chronological order (id order == time order at nanosecond precision). |
| 283 | + unique_events = {event.id: event for event in events or []} |
| 284 | + self._events = [unique_events[event_id] for event_id in sorted(unique_events)] |
| 285 | + |
| 286 | + @property |
| 287 | + def events(self) -> list: |
| 288 | + """Copy of the window fetched so far, oldest first. Complete from the beginning |
| 289 | + of history only once fetch_older() has returned an empty list.""" |
| 290 | + return list(self._events) |
| 291 | + |
| 292 | + def fetch_older(self, max_lines: int = DEFAULT_LOG_PAGE_LINES) -> list: |
| 293 | + """Fetch the page older than the window and prepend it; on an empty session |
| 294 | + fetches the newest page (tail). Returns the page; empty list = no older |
| 295 | + lines exist (yet).""" |
| 296 | + page = self._client.get_deployment_logs( |
| 297 | + self._deployment_id, self._revision_number, self._pod, before=self._events, max_lines=max_lines |
| 298 | + ) |
| 299 | + self._events[:0] = page |
| 300 | + return page |
| 301 | + |
| 302 | + def fetch_newer(self, max_lines: int = DEFAULT_LOG_PAGE_LINES) -> list: |
| 303 | + """Fetch lines newer than the window and merge them in; on an empty session |
| 304 | + fetches the newest page (tail) — to read from the beginning of history |
| 305 | + instead, loop fetch_older() until it returns an empty list. Returns only |
| 306 | + the new lines; empty list = nothing new yet, call again later to keep |
| 307 | + tailing. Rare late arrivals sort into the window below its newest lines.""" |
| 308 | + if not self._events: |
| 309 | + return self.fetch_older(max_lines=max_lines) |
| 310 | + delta = self._client.get_deployment_logs( |
| 311 | + self._deployment_id, self._revision_number, self._pod, after=self._events, max_lines=max_lines |
| 312 | + ) |
| 313 | + for event in delta: |
| 314 | + if event.id > self._events[-1].id: |
| 315 | + self._events.append(event) |
| 316 | + else: |
| 317 | + # A late arrival may even precede the window's oldest line (tail page |
| 318 | + # cut inside the look-behind span); the server delivers that span |
| 319 | + # completely on top of max_lines, so the window stays contiguous. |
| 320 | + insort(self._events, event, key=lambda held: held.id) |
| 321 | + return delta |
234 | 322 |
|
235 | 323 |
|
236 | 324 | @contextmanager |
|
0 commit comments