1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
|
"""Scoped pre-traffic smoke + post-promote probe (spec §7.1 steps 2/5).
Probes ONE version's pages plus apex special paths through a host, presenting a
CF Access service token. Header contract (§3.3): content probes assert
X-Docs-Build == --expect-sha; apex probes assert X-Apex-Build presence only.
Phase-2 obligation (authorized addition, not in the original spec text): the
version's index.html probe also asserts the response body carries the
`#vyos-search` mount div (docs/_templates/searchbox.html), guarding the
Pagefind gate's silent-degrade failure mode — a build that forgot to set
DOCS_VERSION_SLUG would otherwise ship stock RTD search without CI noticing.
"""
from __future__ import annotations
import argparse
import dataclasses
import json
import sys
import time
import urllib.request
APEX_PATHS = ["/versions.json", "/healthz", "/robots.txt", "/sitemap.xml"]
SEARCH_MOUNT_MARKER = 'id="vyos-search"'
# Explicit UA so the gate never depends on a Cloudflare edge exemption for the default
# Python-urllib UA. The Browser Integrity Check blocked that UA until a skip rule was added;
# the gate must not silently rely on that rule surviving.
USER_AGENT = "vyos-docs-smoke/1.0 (+https://github.com/vyos/vyos-documentation)"
# Round-based retry (module-level so tests can shrink them). A freshly-deployed worker version
# can lose a propagation race: for a few minutes a probe may be served by the PREVIOUS version
# (wrong status / stale X-Docs-Build). Rather than fail-fast, each round re-probes ONLY the
# still-failing probes — this preserves the full per-probe failure enumeration (diagnostic
# value) while adding at most (MAX_ROUNDS - 1) inter-round sleeps. Envelope widened to 5 rounds
# x 30s after an observed propagation wave outlasted 3 rounds x 20s (a path still stale at
# round 3): 4 x 30s = 2 min now covers the observed 1-2+ min waves. The green path still costs
# zero extra time (no retries), and DEADLINE_SECONDS=480 still bounds the worst case.
MAX_ROUNDS = 5
RETRY_SLEEP_SECONDS = 30
DEADLINE_SECONDS = 480
# Per-socket-op timeout (connect + read), capped down to the remaining deadline budget on each
# probe so a probe that starts late cannot overshoot DEADLINE_SECONDS. Pages are small, so this
# socket-op timeout also bounds body reads adequately — no separate body-read deadline is needed.
PROBE_TIMEOUT_SECONDS = 30
class _NoRedirect(urllib.request.HTTPRedirectHandler):
"""Probes assert an EXACT status per-path (200 or 404) — following a 3xx would
silently swap the probed status for whatever the redirect target returns,
masking an accidental redirect where a direct 200/404 was expected. Mirrors
parity.py's _NoRedirect/_OPENER pattern."""
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D401
return None
_OPENER = urllib.request.build_opener(_NoRedirect)
@dataclasses.dataclass
class Probe:
path: str
expect_status: int
assert_docs_build: bool
assert_apex_build: bool
assert_search_mount: bool = False
def probe_plan(slug: str, pdf: str | None, critical: list[str]) -> list[Probe]:
# `critical` may itself list "index.html" (it does in critical-pages.txt); drop it so the
# index page is probed exactly once — as plan[0], the sole search-mount probe below.
critical = [rel for rel in critical if rel != "index.html"]
plan = [Probe(f"/en/{slug}/{rel}", 200, True, False) for rel in ["index.html", *critical]]
plan.append(Probe(f"/en/{slug}/pagefind/pagefind.js", 200, True, False))
if pdf:
plan.append(Probe(pdf, 200, True, False))
plan.append(Probe(f"/en/{slug}/definitely-missing-page-xyz.html", 404, False, False))
plan += [Probe(p, 200, False, True) for p in APEX_PATHS]
plan[0].assert_search_mount = True # plan[0] is always /en/<slug>/index.html
return plan
def docs_build_ok(header_value: str | None, expect_sha: str) -> bool:
"""SKIP sentinel (nightly sweep): header presence only; otherwise exact match."""
if expect_sha == "SKIP":
return header_value is not None
return header_value == expect_sha
def search_mount_present(html: str) -> bool:
return SEARCH_MOUNT_MARKER in html
def _probe_once(host: str, probe: Probe, expect_sha: str, access_id: str, access_secret: str,
timeout: float = PROBE_TIMEOUT_SECONDS,
) -> tuple[bool, int | None, str | None, str | None]:
"""One probe attempt. Returns (ok, status, docs_build, detail). `detail` names the failed
check(s) — "status" / "docs-build" / "apex-build" / "search-mount" joined by "+", or the
transport error text — and is None when ok. `timeout` is the per-socket-op deadline (connect
+ read); run() caps it to the remaining budget so a late probe can't overshoot the overall
deadline. ANY exception in the open OR body-read path (including a transport error DURING
HTTPError.read()) is contained and yields (False, None, None, <error text>): a retryable
failure, never a traceback. The HTTPError response stream is always closed — it owns a
socket, so a bare e.read() would leak it."""
req = urllib.request.Request(f"https://{host}{probe.path}", method="GET")
req.add_header("CF-Access-Client-Id", access_id)
req.add_header("CF-Access-Client-Secret", access_secret)
req.add_header("User-Agent", USER_AGENT)
try:
try:
with _OPENER.open(req, timeout=timeout) as resp:
status, headers, body = resp.status, resp.headers, resp.read()
except urllib.error.HTTPError as e: # non-2xx still carries headers/body
with e: # HTTPError is file-like and owns the response socket — always close it
status, headers, body = e.code, e.headers, e.read()
except Exception as e: # noqa: BLE001 — open OR read failure → retryable probe result
return False, None, None, str(e)
reasons: list[str] = []
if status != probe.expect_status:
reasons.append("status")
if probe.assert_docs_build and not docs_build_ok(headers.get("X-Docs-Build"), expect_sha):
reasons.append("docs-build")
if probe.assert_apex_build and not headers.get("X-Apex-Build"):
reasons.append("apex-build")
if probe.assert_search_mount and not search_mount_present(
body.decode("utf-8", errors="replace")):
reasons.append("search-mount")
detail = "+".join(reasons) if reasons else None
return not reasons, status, headers.get("X-Docs-Build"), detail
def run(host: str, slug: str, expect_sha: str, access_id: str, access_secret: str,
pdf: str | None, critical: list[str]) -> int:
"""Probe the whole plan, then re-probe ONLY the still-failing probes each round (up to
MAX_ROUNDS, one RETRY_SLEEP_SECONDS between rounds). A probe passing in ANY round passes;
a single propagation blip served by the previous worker version cannot fail the gate.
DEADLINE_SECONDS bounds total wall-clock — on breach, unresolved probes count as failed."""
plan = probe_plan(slug, pdf, critical)
start = time.monotonic()
deadline = start + DEADLINE_SECONDS # absolute — a hard upper bound on total wall-clock
pending = list(plan) # probes not yet passed
detail_by_path: dict[str, str] = {} # last failure detail per path, for logging
deadline_hit = False
def _remaining() -> float:
return deadline - time.monotonic()
for round_num in range(1, MAX_ROUNDS + 1):
if not pending:
break
still_failing: list[Probe] = []
unprobed: list[Probe] = []
for i, probe in enumerate(pending):
remaining = _remaining() # checked before each probe
if remaining < 1: # < 1s budget: don't start a probe that could overshoot
deadline_hit = True
unprobed = pending[i:] # not reached this round → still unresolved
break
ok, status, docs_build, detail = _probe_once(
host, probe, expect_sha, access_id, access_secret,
timeout=min(PROBE_TIMEOUT_SECONDS, max(1, remaining)))
if ok:
continue
still_failing.append(probe)
detail_by_path[probe.path] = (
f"status={status} docs-build={docs_build} detail={detail}")
pending = still_failing + unprobed
if deadline_hit or not pending or round_num == MAX_ROUNDS:
break
remaining = _remaining() # checked before the inter-round sleep
if remaining <= 0: # no budget left → deadline path (skip the sleep)
deadline_hit = True
break
for probe in pending:
print(f"SMOKE-RETRY {probe.path}: round {round_num} "
f"{detail_by_path[probe.path]}", file=sys.stderr)
time.sleep(min(RETRY_SLEEP_SECONDS, remaining)) # never sleep past the deadline
if deadline_hit:
print("SMOKE-DEADLINE: overall deadline reached — remaining probes counted as failed",
file=sys.stderr)
for probe in pending:
print(f"SMOKE-FAIL {probe.path}: {detail_by_path.get(probe.path, 'unresolved')}",
file=sys.stderr)
failures = len(pending)
print(json.dumps({"failures": failures}))
return 1 if failures else 0
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--host", required=True)
ap.add_argument("--slug", required=True)
ap.add_argument("--expect-sha", required=True)
ap.add_argument("--access-id", required=True)
ap.add_argument("--access-secret", required=True)
ap.add_argument("--pdf", default=None)
ap.add_argument("--critical-list", default="scripts/docs_gates/critical-pages.txt")
a = ap.parse_args()
critical = [line.strip() for line in open(a.critical_list).read().splitlines()
if line.strip() and not line.startswith("#")]
return run(a.host, a.slug, a.expect_sha, a.access_id, a.access_secret, a.pdf, critical)
if __name__ == "__main__":
raise SystemExit(main())
|
