Skip to content

Commit 291ea79

Browse files
alfredeenalfredeen
committed
Add a rollout guard to protect against invalid Deleted statuses during pod swaps. Add an is_pod_terminating function but only log the result for now.
1 parent ed10e83 commit 291ea79

2 files changed

Lines changed: 162 additions & 7 deletions

File tree

‎serve_event_listener/status_data.py‎

Lines changed: 34 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
from typing import Any, Dict, Optional, Tuple, Union
77

88
from kubernetes import client
9+
from kubernetes.client import V1Pod
910
from kubernetes.client.exceptions import ApiException
1011
from kubernetes.client.models import V1PodStatus
1112

@@ -58,6 +59,29 @@ def _detect_app_type(pod: Optional[object]) -> Optional[AppType]:
5859
return None
5960

6061

62+
def is_pod_terminating(pod: V1Pod) -> bool:
63+
# 1) canonical: deletionTimestamp means the pod is terminating
64+
# if getattr(getattr(pod, "metadata", None), "deletion_timestamp", None):
65+
# return True
66+
67+
# 2) friendly signal: Ready=False because the pod is being deleted
68+
conds = getattr(getattr(pod, "status", None), "conditions", None) or []
69+
for c in conds:
70+
if getattr(c, "type", "") == "Ready" and getattr(c, "status", "") == "False":
71+
if getattr(c, "reason", "") == "PodDeleting":
72+
return True
73+
74+
# 3) containers are terminating (optional heuristic)
75+
cstats = getattr(getattr(pod, "status", None), "container_statuses", None) or []
76+
for cs in cstats:
77+
st = getattr(cs, "state", None)
78+
term = getattr(st, "terminated", None) if st else None
79+
if term is not None:
80+
return True
81+
82+
return False
83+
84+
6185
class StatusData:
6286
"""StatusData represents event and status information about a k8s app."""
6387

@@ -198,7 +222,8 @@ def fetch_pod_phases_in_release_from_k8s_api(
198222
If no pod matches the release, then returns empty list.
199223
"""
200224
logger.debug(
201-
f"Getting the nr of pods in release {release} directly from k8s via the api client"
225+
"Getting the nr of pods in release %s directly from k8s via the api client",
226+
release,
202227
)
203228

204229
# Using label_selector with app=release because no pods were selected using the
@@ -303,7 +328,14 @@ def update(self, event: dict) -> None:
303328
release = pod.metadata.labels.get("release")
304329

305330
logger.info(
306-
f"--- Event triggered update status data from release {release}"
331+
"--- Event triggered update status data from release %s", release
332+
)
333+
334+
# Consider using a is_pod_terminating() method to determine a custom status Terminating
335+
# As we are not ready to introduce a new status Terminating, we will only log it for now
336+
is_terminating = is_pod_terminating(pod)
337+
logger.info(
338+
"Is release %s identified as Termianting? %s", release, is_terminating
307339
)
308340

309341
status_object = pod.status

‎serve_event_listener/status_queue.py‎

Lines changed: 128 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,10 @@
3131

3232
PROCESS_TICK_SECONDS = 2.0
3333

34+
# how long to defer 'Deleted' after we recently saw 'Running' for the same release
35+
# seconds to tolerate a swap of pods in a release
36+
ROLLOUT_GUARD_SECONDS = float(os.getenv("ROLLOUT_GUARD_SECONDS", "60"))
37+
3438

3539
def _parse_csv_env(name: str) -> set[str]:
3640
raw = os.getenv(name, "") or ""
@@ -47,6 +51,20 @@ def _parse_csv_env(name: str) -> set[str]:
4751
}
4852

4953

54+
def _ts(iso: str | None) -> float:
55+
if not iso:
56+
return 0.0
57+
58+
try:
59+
return (
60+
datetime.strptime(iso, "%Y-%m-%dT%H:%M:%S.%fZ")
61+
.replace(tzinfo=timezone.utc)
62+
.timestamp()
63+
)
64+
except Exception:
65+
return 0.0
66+
67+
5068
class StatusQueue:
5169
"""
5270
StatusQueue represents a queue of k8s pod statuses to process.
@@ -82,7 +100,7 @@ def __init__(
82100
Raises:
83101
ValueError: If `url` is not absolute or `token` is empty.
84102
"""
85-
# Minimal runtime validation (nice for early misconfig)
103+
# minimal runtime validation (nice for early misconfig)
86104
if not token:
87105
raise ValueError("token must be non-empty")
88106
parsed = urlparse(url)
@@ -95,9 +113,16 @@ def __init__(
95113
self.token_fetcher = token_fetcher
96114
self.prober = prober
97115

98-
self.queue = Queue[StatusRecord]() # queue of StatusRecord items
116+
# queue of StatusRecord items
117+
self.queue = Queue[StatusRecord]()
99118
self.stop_event = threading.Event()
100119

120+
# data structures to capture attributes per release needed to handle rollout swaps
121+
# stores the last Running event (epoch)
122+
self._last_seen_running_ts: dict[str, float] = {}
123+
# do not finalize Deleted before this epoch
124+
self._rollout_block_until: dict[str, float] = {}
125+
101126
def add(self, record: StatusRecord) -> None:
102127
"""Enqueue a StatusRecord."""
103128
self.queue.put(record)
@@ -106,6 +131,15 @@ def add(self, record: StatusRecord) -> None:
106131
record.get("release"),
107132
self.queue.qsize() + 1,
108133
)
134+
# special attributes for handling rollout swaps
135+
evt_ts = _ts(record.get("event-ts"))
136+
status_lc = (record.get("status") or "").lower()
137+
if status_lc == "running":
138+
# remember latest running event
139+
if evt_ts > (
140+
self._last_seen_running_ts.get(record.get("release", ""), 0.0)
141+
):
142+
self._last_seen_running_ts[record["release"]] = evt_ts
109143

110144
def process(self) -> None:
111145
"""Process the queue in a loop until stop event is set.
@@ -116,11 +150,21 @@ def process(self) -> None:
116150
- Deleted (shiny/shiny-proxy): require NXDOMAIN NotFound N times within
117151
DELETED_PROBE_WINDOW/INTERVAL, then confirm 'Deleted'.
118152
- If probing is disabled for 'deleted', keep legacy <30s requeue grace.
153+
- Rollout guard — after we see 'Running' for a release, we defer any 'Deleted'
154+
for the same release for a short window to avoid posting spurious deletions
155+
during pod swaps.
119156
"""
120-
# Track logging of empty queue
157+
# track logging of empty queue
121158
q_empty_log = 0
122159

160+
# lazy create tracking maps; keep them small by popping on completion/expiry
161+
if not hasattr(self, "_last_seen_running_ts"):
162+
self._last_seen_running_ts = {} # release -> epoch (float)
163+
if not hasattr(self, "_rollout_block_until"):
164+
self._rollout_block_until = {} # release -> epoch (float)
165+
123166
PERIOD = PROCESS_TICK_SECONDS
167+
guard_seconds = globals().get("ROLLOUT_GUARD_SECONDS", 15.0)
124168

125169
next_tick = time.monotonic()
126170
while not self.stop_event.is_set():
@@ -137,10 +181,60 @@ def process(self) -> None:
137181

138182
release = rec.get("release")
139183
status_lc = (rec.get("status") or "").lower()
184+
now = time.time()
140185

141186
# default action: proceed to POST (may be flipped to requeue)
142187
requeue = False
143188

189+
# ---- ROLLOUT GUARD: track Running and block Deleted briefly ----
190+
if status_lc == "running":
191+
# seeing Running again extends/refreshes the guard
192+
self._last_seen_running_ts[release] = now
193+
self._rollout_block_until[release] = now + float(guard_seconds)
194+
# reset any NXDOMAIN counters from earlier Deleted probes
195+
rec.pop("_nx_consec", None)
196+
197+
elif status_lc == "deleted":
198+
# if we've seen Running very recently for this release, defer 'Deleted'
199+
block_until = self._rollout_block_until.get(release)
200+
if block_until and now < block_until:
201+
# inside guard: defer; optionally probe to see if it flipped back to Running
202+
if (
203+
self.prober
204+
and self._probe_enabled_for(rec)
205+
and self._allow_probe_now(rec)
206+
):
207+
app_url = rec.get("app-url")
208+
pr = self.prober.probe_url(app_url) # type: ignore[arg-type]
209+
rec["curl-probe"] = {
210+
"status": pr.status,
211+
"port80_status": pr.port80_status,
212+
"note": pr.note,
213+
"url": app_url,
214+
}
215+
# if probe says Running, convert this record to Running and refresh guard
216+
if pr.status == "Running":
217+
rec["status"] = "Running"
218+
status_lc = "running"
219+
self._last_seen_running_ts[release] = now
220+
self._rollout_block_until[release] = now + float(
221+
guard_seconds
222+
)
223+
rec.pop("_nx_consec", None)
224+
# always requeue during guard; throttle next attempt
225+
requeue = True
226+
self._schedule_next_probe(rec, "deleted")
227+
self.queue.task_done()
228+
self.queue.put(rec)
229+
time.sleep(0.01)
230+
# prevent unbounded growth: drop expired blocks opportunistically
231+
# (no-op here, because we're still inside guard)
232+
continue
233+
else:
234+
# guard expired → clean out any stale entry
235+
if block_until and now >= block_until:
236+
self._rollout_block_until.pop(release, None)
237+
144238
# probing path (gated)
145239
if status_lc in {"running", "deleted"} and self._probe_enabled_for(rec):
146240

@@ -163,13 +257,20 @@ def process(self) -> None:
163257
"note": pr.note,
164258
"url": app_url,
165259
}
260+
logger.debug(
261+
"Probing completed. Returned status=%s, port80_status=%s, note=%s",
262+
pr.status,
263+
pr.port80_status,
264+
pr.note,
265+
)
166266

167267
if status_lc == "running":
168268
# confirm only on probe Running; otherwise keep probing
169269
if pr.status != "Running":
170270
requeue = True
171271
self._schedule_next_probe(rec, status_lc)
172-
else: # deleted
272+
273+
elif status_lc == "deleted":
173274
# Deleted only confirms on NXDOMAIN NotFound; require N consecutive
174275
if pr.status == "NotFound":
175276
nx = int(rec.get("_nx_consec", 0)) + 1
@@ -183,6 +284,11 @@ def process(self) -> None:
183284
rec["_nx_consec"] = 0
184285
requeue = True
185286
self._schedule_next_probe(rec, status_lc)
287+
else: # not running or deleted
288+
logger.warning(
289+
"The status in the probing logic must be running or deleted but is %s",
290+
status_lc,
291+
)
186292
else:
187293
# not time yet; requeue to avoid blocking
188294
requeue = True
@@ -203,9 +309,14 @@ def process(self) -> None:
203309
self.queue.put(rec)
204310
# yield briefly; actual delay is handled by the next-epoch throttle
205311
time.sleep(0.01)
312+
# defensive cleanup: drop expired guard entries so the dict can't grow unbounded
313+
# (cheap O(1) check per loop)
314+
ru = self._rollout_block_until.get(release)
315+
if ru and time.time() >= ru:
316+
self._rollout_block_until.pop(release, None)
206317
continue
207318

208-
# build payload and POST
319+
# ---------- build payload and POST ----------
209320
if not rec.get("status"):
210321
logger.error(
211322
"record missing status before POST; release=%s keys=%s",
@@ -250,6 +361,18 @@ def process(self) -> None:
250361
else:
251362
logger.debug("POST ok: %s (%s)", resp.status_code, release)
252363

364+
# -------- cleanup: prevent tracking dicts from growing --------
365+
# if we posted a terminal/non-running state or guard expired, drop entries.
366+
if status_lc != "running":
367+
self._last_seen_running_ts.pop(release, None)
368+
# if posted Deleted or done with this release, drop guard if expired or not useful
369+
ru = self._rollout_block_until.get(release)
370+
if ru and time.time() >= ru:
371+
self._rollout_block_until.pop(release, None)
372+
# also drop guard after posting Deleted
373+
if status_lc == "deleted":
374+
self._rollout_block_until.pop(release, None)
375+
253376
logger.debug(
254377
"Processed queue successfully of release %s, new status=%s",
255378
release,

0 commit comments

Comments
 (0)