Coverage for scripts/live_release_validation/checks/tracing.py: 100.00%

1"""Checks behind the ``tracing`` action: GCO's spans reach CloudWatch Transaction Search.

3The four API services (health-monitor, manifest-processor, inference-proxy,

4cost-monitor) export OpenTelemetry spans to their Region's X-Ray OTLP

5endpoint. With Transaction Search on, X-Ray stores every span as a structured

6log event in the Region's ``aws/spans`` log group, and that is where this

7check looks. Per regional Region it proves:

91. the X-Ray trace segment destination is ``CloudWatchLogs`` and ``ACTIVE``

10 (the regional stack's enabler switched it; spans are searchable only then);

112. authenticated API traffic, sent through the same per-Region transport the

12 other API actions use, reaches the health monitor (``/api/v1/metrics``:

13 ``/api/v1/health`` is excluded from tracing), the manifest processor

14 (``/api/v1/policy``) and, when cost monitoring is on,

15 ``/api/v1/cost/status``, which the manifest processor answers by calling

16 the cost monitor over verified HTTPS, so the W3C context its client span

17 injects puts the cost monitor's server span into the same trace; and

183. CloudWatch Logs Insights, polled against a deadline because X-Ray delivers

19 spans minutes late, finds spans for every expected ``service.name``

20 resource attribute (``service.namespace`` ``gco``) and at least one trace

21 holding both a manifest-processor and a cost-monitor span.

23The inference proxy is expected only in the Region where this run's

24``inference`` action sent real requests through it, counted from that

25action's start; the run has no running endpoint of its own by the time this

26action runs, so anywhere else it is skipped with the reason. Evidence is

27counts only: no trace, span, or request identifiers.

29Every CDK invocation of a release run carries the ``tracing_overrides``

30context ``{"sample_ratio": 1.0}`` (:func:`tracing_overrides_json`), so every

31request is sampled; with a lower effective ratio the traffic rounds grow so

32that each request kind is still sampled with high probability.

33"""

35from __future__ import annotations

37import contextlib

38import json

39import math

40import time

41from collections.abc import Iterable

42from datetime import datetime

43from typing import Any

45from botocore.exceptions import ClientError

47from ..context import _job_transport_region

48from ..models import RunContext

49from ..ownership.transaction_search import _read_trace_segment_destination

50from .opencost import _cost_monitoring_configured

52#: ``service.name`` of every traced GCO service, in manifest order.

53TRACED_SERVICES = ("health-monitor", "manifest-processor", "inference-proxy", "cost-monitor")

54#: The ``tracing_overrides`` context every release-run CDK invocation carries.

55RELEASE_TRACING_OVERRIDES: dict[str, Any] = {"sample_ratio": 1.0}

56SPAN_LOG_GROUP = "aws/spans"

58#: cdk.json ``tracing`` defaults when a key (or the whole block) is absent.

59_DEFAULT_TRACING: dict[str, Any] = {

60 "enabled": True,

61 "sample_ratio": 0.05,

62 "enable_transaction_search": True,

63}

64_HEALTH_MONITOR_PATH = "/api/v1/metrics"

65_MANIFEST_PROCESSOR_PATH = "/api/v1/policy"

66_COST_STATUS_PATH = "/api/v1/cost/status"

67_MIN_TRAFFIC_ROUNDS = 3

68_MAX_TRAFFIC_ROUNDS = 200

69#: With a ratio below 1, rounds are sized so each request kind is sampled at

70#: least once with this probability.

71_SAMPLING_CONFIDENCE = 0.999

72#: The enabler switches the destination at deploy; ``PENDING`` can last minutes.

73_DESTINATION_WAIT_SECONDS = 900

74#: X-Ray delivers spans to ``aws/spans`` minutes after export.

75_SPAN_WAIT_SECONDS = 900

76_SPAN_POLL_SECONDS = 30

77_QUERY_WAIT_SECONDS = 120

78_QUERY_POLL_SECONDS = 2

79#: Slack on both ends of a query window for clock skew and ingestion time.

80_WINDOW_MARGIN_SECONDS = 120

81_QUERY_RESULT_LIMIT = 10_000

82_TERMINAL_QUERY_STATUSES = frozenset({"Failed", "Cancelled", "Timeout", "Unknown"})

83#: ``StartQuery`` errors a later poll may not see again: the span log group is

84#: created on first delivery, and Logs Insights caps concurrent queries.

85_TRANSIENT_QUERY_ERRORS = frozenset(

86 {

87 "LimitExceededException",

88 "ResourceNotFoundException",

89 "ServiceUnavailableException",

90 "ThrottlingException",

91 }

92)

93_SERVICE_FIELD = "`resource.attributes.service.name`"

94_NAMESPACE_FIELD = "`resource.attributes.service.namespace`"

95_CORRELATED_SERVICES = frozenset({"manifest-processor", "cost-monitor"})

98class TracingValidationError(RuntimeError):

99 """The deployed services' spans did not reach Transaction Search as promised."""

100

101

102class _TransientQueryError(RuntimeError):

103 """One Logs Insights round failed in a way a later round may not repeat."""

104

105

106def tracing_overrides_json() -> str:

107 """Canonical JSON for the ``tracing_overrides`` context and the resume identity."""

108 return json.dumps(RELEASE_TRACING_OVERRIDES, sort_keys=True, separators=(",", ":"))

109

110

111def effective_tracing_config(ctx: RunContext) -> dict[str, Any]:

112 """cdk.json ``tracing`` merged with the run's ``tracing_overrides``, validated."""

113 block = ctx.cdk_context.get("tracing")

114 if block is not None and not isinstance(block, dict):

115 raise TracingValidationError("cdk.json context.tracing must be an object")

116 raw_overrides = str(getattr(ctx.settings, "tracing_overrides_json", "") or "")

117 try:

118 overrides = json.loads(raw_overrides) if raw_overrides else {}

119 except json.JSONDecodeError as exc:

120 raise TracingValidationError(f"tracing_overrides is not valid JSON: {exc}") from exc

121 if not isinstance(overrides, dict):

122 raise TracingValidationError("tracing_overrides must be a JSON object")

123 merged = {**_DEFAULT_TRACING, **(block or {}), **overrides}

124 enabled = merged["enabled"]

125 transaction_search = merged["enable_transaction_search"]

126 ratio = merged["sample_ratio"]

127 if not isinstance(enabled, bool) or not isinstance(transaction_search, bool):

128 raise TracingValidationError(

129 "tracing.enabled and tracing.enable_transaction_search must be booleans"

130 )

131 if isinstance(ratio, bool) or not isinstance(ratio, (int, float)) or not 0 <= ratio <= 1:

132 raise TracingValidationError("tracing.sample_ratio must be a number in [0, 1]")

133 return {

134 "enabled": enabled,

135 "sample_ratio": float(ratio),

136 "enable_transaction_search": transaction_search,

137 }

138

139

140def _traffic_rounds(sample_ratio: float) -> int:

141 """Rounds of requests so each kind is sampled at least once with high probability."""

142 if sample_ratio >= 1:

143 return _MIN_TRAFFIC_ROUNDS

144 needed = math.ceil(math.log(1 - _SAMPLING_CONFIDENCE) / math.log(1 - sample_ratio))

145 return max(_MIN_TRAFFIC_ROUNDS, min(_MAX_TRAFFIC_ROUNDS, needed))

146

147

148def _epoch_ms(timestamp: str) -> int:

149 try:

150 return int(datetime.fromisoformat(timestamp).timestamp() * 1000)

151 except (TypeError, ValueError) as exc:

152 raise TracingValidationError(f"Invalid action timestamp {timestamp!r}") from exc

153

154

155def _inference_proxy_expectation(ctx: RunContext, region: str) -> int | str:

156 """When the run's inference traffic crossed this Region's proxy (epoch ms), or why not."""

157 settings = ctx.settings

158 if not getattr(settings, "inference_enabled", False):

159 return (

160 "the inference action is not in this run's scope, so no inference request "

161 "reached the proxy"

162 )

163 selected = str(getattr(settings, "selected_region", "") or "")

164 if region != selected:

165 return f"this run's inference requests went through the proxy in {selected}"

166 result = ctx.checkpoint.action_results.get("inference")

167 if result is None or result.status != "passed":

168 return (

169 "the inference action has not passed in this run, so no served inference "

170 "request is known to have reached the proxy"

171 )

172 return _epoch_ms(result.started_at)

173

174

175def _expected_services(

176 ctx: RunContext, region: str

177) -> tuple[list[str], dict[str, str], int | None]:

178 """Return ``(expected services, skipped with reason, inference traffic start ms)``."""

179 skipped: dict[str, str] = {}

180 if not _cost_monitoring_configured(ctx):

181 skipped["cost-monitor"] = (

182 "cost monitoring is disabled in cdk.json (cost_monitoring or cluster_observability)"

183 )

184 inference = _inference_proxy_expectation(ctx, region)

185 if isinstance(inference, str):

186 skipped["inference-proxy"] = inference

187 expected = [service for service in TRACED_SERVICES if service not in skipped]

188 return expected, skipped, None if isinstance(inference, str) else inference

189

190

191def _wait_for_transaction_search(

192 ctx: RunContext,

193 region: str,

194 config: dict[str, Any],

195) -> tuple[dict[str, str], str | None]:

196 """Return ``(destination, skip reason)`` once spans are searchable in the Region."""

197 xray = ctx.session.client("xray", region_name=region)

198 deadline = time.monotonic() + _DESTINATION_WAIT_SECONDS

199 while True:

200 state = _read_trace_segment_destination(xray)

201 if state["destination"] == "XRay":

202 if not config["enable_transaction_search"]:

203 return state, (

204 "tracing.enable_transaction_search is false and the Region's trace "

205 f"segment destination is XRay, so spans are not stored in {SPAN_LOG_GROUP}"

206 )

207 raise TracingValidationError(

208 f"{region}: the trace segment destination is XRay although "

209 "tracing.enable_transaction_search is on; the Transaction Search enabler "

210 "did not switch it"

211 )

212 if state["status"] == "ACTIVE":

213 return state, None

214 if time.monotonic() >= deadline:

215 raise TracingValidationError(

216 f"{region}: Transaction Search stayed {state['status']} for "

217 f"{_DESTINATION_WAIT_SECONDS}s"

218 )

219 time.sleep(_SPAN_POLL_SECONDS)

220

221

222def _drive_traffic(

223 ctx: RunContext,

224 region: str,

225 *,

226 rounds: int,

227 cost: bool,

228) -> dict[str, dict[str, Any]]:

229 """Send authenticated GETs to each traced API service; return per-path counts."""

230 transport = _job_transport_region(ctx, region)

231 paths = [_HEALTH_MONITOR_PATH, _MANIFEST_PROCESSOR_PATH]

232 if cost:

233 paths.append(_COST_STATUS_PATH)

234 traffic: dict[str, dict[str, Any]] = {

235 path: {"requests": 0, "status_codes": {}} for path in paths

236 }

237 for _round in range(rounds):

238 for path in paths:

239 response = ctx.aws_client.make_authenticated_request(

240 method="GET",

241 path=path,

242 target_region=transport,

243 )

244 entry = traffic[path]

245 code = str(response.status_code)

246 entry["requests"] += 1

247 entry["status_codes"][code] = entry["status_codes"].get(code, 0) + 1

248 if not response.ok:

249 raise TracingValidationError(

250 f"{region}: GET {path} answered HTTP {response.status_code} while "

251 "driving traced traffic"

252 )

253 return traffic

254

255

256def _service_filter(services: Iterable[str]) -> str:

257 names = ", ".join(json.dumps(name) for name in services)

258 return f'{_NAMESPACE_FIELD} = "gco" and {_SERVICE_FIELD} in [{names}]'

259

260

261def _span_count_query(services: Iterable[str]) -> str:

262 return (

263 f"fields {_SERVICE_FIELD} as service\n"

264 f"| filter {_service_filter(services)}\n"

265 "| stats count(*) as spans by service"

266 )

267

268

269def _correlation_query() -> str:

270 return (

271 f"fields {_SERVICE_FIELD} as service, traceId\n"

272 f"| filter {_service_filter(sorted(_CORRELATED_SERVICES))}\n"

273 "| stats count(*) as spans by traceId, service\n"

274 f"| limit {_QUERY_RESULT_LIMIT}"

275 )

276

277

278def _run_query(logs: Any, query: str, *, start_s: int, end_s: int) -> list[dict[str, str]]:

279 """Run one Logs Insights query on ``aws/spans`` to completion; rows as dicts."""

280 try:

281 started = logs.start_query(

282 logGroupName=SPAN_LOG_GROUP,

283 startTime=start_s,

284 endTime=end_s,

285 queryString=query,

286 limit=_QUERY_RESULT_LIMIT,

287 )

288 except ClientError as exc:

289 code = str(exc.response.get("Error", {}).get("Code") or "")

290 if code in _TRANSIENT_QUERY_ERRORS:

291 raise _TransientQueryError(f"StartQuery failed with {code}") from exc

292 raise

293 query_id = str(started.get("queryId") or "")

294 if not query_id:

295 raise TracingValidationError("CloudWatch Logs Insights returned no query ID")

296 deadline = time.monotonic() + _QUERY_WAIT_SECONDS

297 while True:

298 response = logs.get_query_results(queryId=query_id)

299 status = str(response.get("status") or "")

300 if status == "Complete":

301 return [

302 {

303 str(field.get("field")): str(field.get("value") or "")

304 for field in row

305 if isinstance(field, dict)

306 }

307 for row in response.get("results") or []

308 if isinstance(row, list)

309 ]

310 if status in _TERMINAL_QUERY_STATUSES:

311 raise _TransientQueryError(f"Logs Insights query ended {status}")

312 if time.monotonic() >= deadline:

313 with contextlib.suppress(ClientError):

314 logs.stop_query(queryId=query_id)

315 raise _TransientQueryError(

316 f"Logs Insights query did not complete within {_QUERY_WAIT_SECONDS}s"

317 )

318 time.sleep(_QUERY_POLL_SECONDS)

319

320

321def _count(value: str) -> int:

322 try:

323 return int(float(value))

324 except OverflowError, ValueError:

325 return 0

326

327

328def _span_snapshot(

329 logs: Any,

330 expected: list[str],

331 *,

332 window_start_ms: int,

333 traffic_start_ms: int,

334 correlate: bool,

335) -> tuple[dict[str, int], int | None]:

336 """Return per-service span counts and, when asked, the correlated trace count."""

337 end_s = int(time.time()) + _WINDOW_MARGIN_SECONDS

338 counts = dict.fromkeys(expected, 0)

339 for row in _run_query(

340 logs,

341 _span_count_query(expected),

342 start_s=window_start_ms // 1000 - _WINDOW_MARGIN_SECONDS,

343 end_s=end_s,

344 ):

345 service = row.get("service", "")

346 if service in counts:

347 counts[service] += _count(row.get("spans", ""))

348 if not correlate:

349 return counts, None

350 services_by_trace: dict[str, set[str]] = {}

351 for row in _run_query(

352 logs,

353 _correlation_query(),

354 start_s=traffic_start_ms // 1000 - _WINDOW_MARGIN_SECONDS,

355 end_s=end_s,

356 ):

357 trace_id = row.get("traceId", "")

358 service = row.get("service", "")

359 if trace_id and service:

360 services_by_trace.setdefault(trace_id, set()).add(service)

361 correlated = sum(

362 1 for services in services_by_trace.values() if services >= _CORRELATED_SERVICES

363 )

364 return counts, correlated

365

366

367def _wait_for_spans(

368 ctx: RunContext,

369 region: str,

370 expected: list[str],

371 *,

372 window_start_ms: int,

373 traffic_start_ms: int,

374 correlate: bool,

375) -> dict[str, Any]:

376 """Poll ``aws/spans`` until every expected service (and the cost trace) shows up."""

377 logs = ctx.session.client("logs", region_name=region)

378 deadline = time.monotonic() + _SPAN_WAIT_SECONDS

379 counts = dict.fromkeys(expected, 0)

380 correlated: int | None = 0 if correlate else None

381 polls = 0

382 while True:

383 polls += 1

384 last_error: str | None = None

385 try:

386 counts, correlated = _span_snapshot(

387 logs,

388 expected,

389 window_start_ms=window_start_ms,

390 traffic_start_ms=traffic_start_ms,

391 correlate=correlate,

392 )

393 except _TransientQueryError as exc:

394 last_error = str(exc)

395 missing = [service for service in expected if counts.get(service, 0) < 1]

396 uncorrelated = correlate and not correlated

397 if last_error is None and not missing and not uncorrelated:

398 return {"span_counts": counts, "correlated_cost_traces": correlated, "polls": polls}

399 if time.monotonic() >= deadline:

400 problems = []

401 if missing:

402 problems.append("no spans for " + ", ".join(missing))

403 if uncorrelated:

404 problems.append("no trace holds both a manifest-processor and a cost-monitor span")

405 if last_error:

406 problems.append(f"last query: {last_error}")

407 raise TracingValidationError(

408 f"{region}: {'; '.join(problems)} in {SPAN_LOG_GROUP} within "

409 f"{_SPAN_WAIT_SECONDS}s ({polls} polls)"

410 )

411 time.sleep(_SPAN_POLL_SECONDS)

412

413

414def verify_region_tracing(

415 ctx: RunContext,

416 region: str,

417 config: dict[str, Any],

418) -> dict[str, Any]:

419 """Drive traced traffic in one Region and require its spans in Transaction Search."""

420 with ctx.state_lock:

421 record: dict[str, Any] = ctx.checkpoint.state.setdefault("tracing", {}).setdefault(

422 region, {}

423 )

424 try:

425 evidence = _verify_region(ctx, region, config, record)

426 except TracingValidationError as exc:

427 record["failure"] = str(exc)

428 ctx.persist()

429 raise

430 record["evidence"] = evidence

431 ctx.persist()

432 return evidence

433

434

435def _verify_region(

436 ctx: RunContext,

437 region: str,

438 config: dict[str, Any],

439 record: dict[str, Any],

440) -> dict[str, Any]:

441 if config["sample_ratio"] <= 0:

442 return {"status": "skipped", "reason": "tracing.sample_ratio is 0, so no span is sampled"}

443 destination, skip = _wait_for_transaction_search(ctx, region, config)

444 if skip is not None:

445 return {"status": "skipped", "reason": skip, "transaction_search": destination}

446 expected, skipped, inference_start_ms = _expected_services(ctx, region)

447 rounds = _traffic_rounds(config["sample_ratio"])

448 cost = "cost-monitor" in expected

449 traffic_start_ms = int(time.time() * 1000)

450 traffic = _drive_traffic(ctx, region, rounds=rounds, cost=cost)

451 record["traffic"] = traffic

452 ctx.persist()

453 window_start_ms = (

454 traffic_start_ms

455 if inference_start_ms is None

456 else min(traffic_start_ms, inference_start_ms)

457 )

458 spans = _wait_for_spans(

459 ctx,

460 region,

461 expected,

462 window_start_ms=window_start_ms,

463 traffic_start_ms=traffic_start_ms,

464 correlate=cost,

465 )

466 return {

467 "status": "passed",

468 "transaction_search": destination,

469 "sample_ratio": config["sample_ratio"],

470 "expected_services": expected,

471 "skipped_services": skipped,

472 "traffic_rounds": rounds,

473 "traffic": traffic,

474 **spans,

475 }

Coverage for scripts / live_release_validation / checks / tracing.py: 100.00%

227 statements