Coverage for gco / services / webhook_dispatcher.py: 100.00%

1""" 

2Webhook Dispatcher Service for GCO (Global Capacity Orchestrator on AWS). 

3 

4This service monitors Kubernetes job status changes and dispatches webhook 

5notifications to registered endpoints. It runs as a background task alongside 

6the health monitor or as a standalone service. 

7 

8Key Features: 

9- Watches Kubernetes jobs for status changes (started, completed, failed) 

10- Queries matching webhooks from DynamoDB based on event type and namespace 

11- Dispatches HTTP POST requests with JSON payloads 

12- Signs payloads with HMAC-SHA256 when a secret is configured 

13- Implements retry logic with exponential backoff for failed deliveries 

14- Publishes delivery metrics to CloudWatch 

15 

16Replica model: the health-monitor runs two replicas for availability, but a 

17webhook must fire once per job transition, not once per replica. When a 

18``LeaseIdentity`` is configured the dispatcher elects a single deliverer 

19through the pre-created ``gco-health-monitor-webhooks`` Lease (see 

20``gco.services.leader_lease``): only the holder watches and delivers, a 

21standby re-checks the Lease every few seconds, and a replica that becomes 

22leader re-seeds its job-state cache first so transitions the previous leader 

23already delivered are not fired again. Delivery is therefore at-most-once 

24across a failover; transitions inside the takeover window (bounded by the 

25lease duration) are dropped rather than duplicated. 

26 

27Webhook Payload Format: 

28 { 

29 "event": "job.completed", 

30 "timestamp": "2026-02-04T12:00:00Z", 

31 "cluster_id": "gco-cluster-us-east-1", 

32 "region": "us-east-1", 

33 "job": { 

34 "name": "my-job", 

35 "namespace": "gco-jobs", 

36 "uid": "abc-123", 

37 "status": "succeeded", 

38 "start_time": "2026-02-04T11:55:00Z", 

39 "completion_time": "2026-02-04T12:00:00Z", 

40 "succeeded": 1, 

41 "failed": 0 

42 } 

43 } 

44 

45HMAC Signature: 

46 When a webhook has a secret configured, the payload is signed using 

47 HMAC-SHA256. The signature is included in the X-GCO-Signature header 

48 as "sha256=<hex_digest>". 

49 

50Environment Variables: 

51 CLUSTER_NAME: Name of the EKS cluster 

52 REGION: AWS region of the cluster 

53 WEBHOOK_TIMEOUT: HTTP timeout for webhook calls (default: 30) 

54 WEBHOOK_MAX_RETRIES: Maximum total delivery attempts (default: 3) 

55 WEBHOOK_RETRY_DELAY: Initial retry delay in seconds (default: 5) 

56 WEBHOOKS_TABLE_NAME: DynamoDB table for webhooks 

57""" 

58 

59from __future__ import annotations 

60 

61import asyncio 

62import contextlib 

63import hashlib 

64import hmac 

65import ipaddress 

66import json 

67import logging 

68import os 

69import socket 

70from dataclasses import dataclass, field 

71from datetime import UTC, datetime 

72from enum import StrEnum 

73from typing import Any 

74from urllib.parse import ParseResult, urlparse 

75 

76import httpx2 

77from kubernetes import client, config 

78from kubernetes.client.models import V1Job 

79from kubernetes.client.rest import ApiException 

80from kubernetes.watch import Watch 

81 

82from gco.services.leader_lease import ( 

83 LeaseIdentity, 

84 lease_duration_from_env, 

85 try_acquire_lease, 

86) 

87from gco.services.template_store import WebhookStore, get_webhook_store 

88 

89# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit 

90# Generated at (UTC): 2026-10-08T19:00:07Z 

91# Generated from Git commit: 231725d929e3dfd3c3768eba41d413effef39d67 

92# Flowchart(s) generated from this file: 

93# * ``WebhookDispatcher._deliver_webhook`` -> ``diagrams/code_diagrams/gco/services/webhook_dispatcher.WebhookDispatcher__deliver_webhook.html`` 

94# (PNG: ``diagrams/code_diagrams/gco/services/webhook_dispatcher.WebhookDispatcher__deliver_webhook.png``) 

95# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``. 

96# <pyflowchart-code-diagram> END 

97 

98 

99logging.basicConfig( 

100 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" 

101) 

102logger = logging.getLogger(__name__) 

103 

104# Networks blocked for SSRF prevention 

105BLOCKED_NETWORKS = [ 

106 ipaddress.ip_network("10.0.0.0/8"), 

107 ipaddress.ip_network("172.16.0.0/12"), 

108 ipaddress.ip_network("192.168.0.0/16"), 

109 ipaddress.ip_network("169.254.0.0/16"), 

110 ipaddress.ip_network("127.0.0.0/8"), 

111 ipaddress.ip_network("::1/128"), 

112 ipaddress.ip_network("fc00::/7"), 

113 ipaddress.ip_network("fe80::/10"), 

114] 

115 

116 

117@dataclass(frozen=True) 

118class _ValidatedWebhookTarget: 

119 """One DNS-validated destination whose transport cannot resolve again.""" 

120 

121 original_url: str 

122 parsed: ParseResult 

123 hostname: str 

124 port: int 

125 addresses: tuple[str, ...] 

126 

127 @property 

128 def log_identity(self) -> str: 

129 """Return an operator-useful identity without path/query credentials.""" 

130 return f"host={self.hostname} port={self.port}" 

131 

132 @property 

133 def host_header(self) -> str: 

134 return self.parsed.netloc 

135 

136 def pinned_url(self, attempt: int) -> str: 

137 address = self.addresses[(attempt - 1) % len(self.addresses)] 

138 host = f"[{address}]" if ":" in address else address 

139 return self.parsed._replace(netloc=f"{host}:{self.port}").geturl() 

140 

141 

142def _globally_routable_unicast( 

143 value: str, 

144) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None: 

145 """Return a canonical public-unicast address, including mapped IPv4. 

146 

147 ``is_global`` intentionally does not exclude multicast in ``ipaddress``. 

148 Normalize IPv4-mapped IPv6 first, then require a globally routable, 

149 non-multicast unicast address so loopback, private, link-local, shared, 

150 documentation, reserved, unspecified, and multicast destinations all fail 

151 closed. 

152 """ 

153 try: 

154 address = ipaddress.ip_address(value) 

155 except ValueError: 

156 return None 

157 if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped is not None: 

158 address = address.ipv4_mapped 

159 if ( 

160 not address.is_global 

161 or address.is_multicast 

162 or address.is_unspecified 

163 or address.is_loopback 

164 or address.is_link_local 

165 or address.is_reserved 

166 ): 

167 return None 

168 return address 

169 

170 

171def _resolve_webhook_target( 

172 url: str, 

173 allowed_domains: list[str] | None = None, 

174) -> tuple[_ValidatedWebhookTarget | None, str | None]: 

175 """Resolve and approve every address before any outbound connection.""" 

176 parsed = urlparse(url) 

177 if parsed.scheme != "https": 

178 return None, "Only HTTPS webhook URLs are allowed" 

179 if parsed.username is not None or parsed.password is not None: 

180 return None, "Webhook URLs must not contain userinfo credentials" 

181 

182 hostname = parsed.hostname 

183 if not hostname: 

184 return None, "Webhook URL must include a valid hostname" 

185 normalized_hostname = hostname.rstrip(".").lower() 

186 normalized_allowed = {value.rstrip(".").lower() for value in allowed_domains or []} 

187 if normalized_allowed and normalized_hostname not in normalized_allowed: 

188 return None, f"Domain '{hostname}' not in allowed domains list" 

189 

190 try: 

191 port = parsed.port or 443 

192 except ValueError: 

193 return None, "Webhook URL contains an invalid port" 

194 try: 

195 resolved = socket.getaddrinfo(hostname, port, proto=socket.IPPROTO_TCP) 

196 except socket.gaierror: 

197 return None, f"DNS resolution failed for {hostname}" 

198 if not resolved: 

199 return None, f"DNS resolution returned no results for {hostname}" 

200 

201 addresses: set[str] = set() 

202 for family, _type, _proto, _canonname, sockaddr in resolved: 

203 # typeshed (mypy 2.4) types every getaddrinfo result as AF_INET or 

204 # AF_INET6, so it calls this branch dead; the resolver is C library 

205 # code outside that model, and an SSRF guard fails closed on purpose. 

206 if family not in {socket.AF_INET, socket.AF_INET6}: 

207 return None, "DNS resolution returned an unsupported address family" # type: ignore[unreachable] 

208 raw_address = sockaddr[0] 

209 if not isinstance(raw_address, str): 

210 return None, "DNS resolution returned a non-text IP address" 

211 address = _globally_routable_unicast(raw_address) 

212 if address is None: 

213 return None, ( 

214 f"Resolved IP {raw_address} is blocked because it is not globally routable unicast" 

215 ) 

216 addresses.add(str(address)) 

217 if not addresses: 

218 return None, f"DNS resolution returned no usable addresses for {hostname}" 

219 

220 return ( 

221 _ValidatedWebhookTarget( 

222 original_url=url, 

223 parsed=parsed, 

224 hostname=hostname, 

225 port=port, 

226 addresses=tuple(sorted(addresses)), 

227 ), 

228 None, 

229 ) 

230 

231 

232def validate_webhook_url( 

233 url: str, allowed_domains: list[str] | None = None 

234) -> tuple[bool, str | None]: 

235 """Validate HTTPS, allowlist, DNS, and blocked-network constraints.""" 

236 target, error = _resolve_webhook_target(url, allowed_domains) 

237 return target is not None, error 

238 

239 

240class WebhookEvent(StrEnum): 

241 """Webhook event types.""" 

242 

243 JOB_STARTED = "job.started" 

244 JOB_COMPLETED = "job.completed" 

245 JOB_FAILED = "job.failed" 

246 

247 

248@dataclass 

249class WebhookDeliveryResult: 

250 """Result of a webhook delivery attempt.""" 

251 

252 webhook_id: str 

253 url: str 

254 event: str 

255 success: bool 

256 status_code: int | None = None 

257 error: str | None = None 

258 attempts: int = 1 

259 duration_ms: float = 0.0 

260 

261 

262@dataclass 

263class JobStateCache: 

264 """Cache of job states to detect transitions.""" 

265 

266 # Map of job_uid -> last known status 

267 job_states: dict[str, str] = field(default_factory=dict) 

268 

269 def get_state(self, job_uid: str) -> str | None: 

270 """Get cached state for a job.""" 

271 return self.job_states.get(job_uid) 

272 

273 def set_state(self, job_uid: str, state: str) -> str | None: 

274 """Set state for a job, returns previous state.""" 

275 previous = self.job_states.get(job_uid) 

276 self.job_states[job_uid] = state 

277 return previous 

278 

279 def remove(self, job_uid: str) -> None: 

280 """Remove a job from the cache.""" 

281 self.job_states.pop(job_uid, None) 

282 

283 

284class WebhookDispatcher: 

285 """ 

286 Dispatches webhook notifications for Kubernetes job events. 

287 

288 This class monitors job status changes and sends HTTP notifications 

289 to registered webhook endpoints. 

290 """ 

291 

292 def __init__( 

293 self, 

294 cluster_id: str, 

295 region: str, 

296 webhook_store: WebhookStore | None = None, 

297 timeout: int = 30, 

298 max_retries: int = 3, 

299 retry_delay: int = 5, 

300 namespaces: list[str] | None = None, 

301 allowed_domains: list[str] | None = None, 

302 leader_lease: LeaseIdentity | None = None, 

303 standby_poll_seconds: float = 5.0, 

304 ): 

305 """Initialize the webhook dispatcher. 

306 

307 Args: 

308 cluster_id: EKS cluster identifier 

309 region: AWS region 

310 webhook_store: DynamoDB webhook store (uses singleton if None) 

311 timeout: HTTP timeout for webhook calls in seconds 

312 max_retries: Maximum total delivery attempts (initial request included) 

313 retry_delay: Initial retry delay in seconds (doubles each retry) 

314 namespaces: Namespaces to watch (None = all non-system namespaces) 

315 allowed_domains: Optional list of allowed webhook domains for SSRF prevention 

316 leader_lease: Lease to hold before watching and delivering; ``None`` 

317 runs single-replica mode (standalone use and unit tests) 

318 standby_poll_seconds: how often a non-leader re-checks the Lease 

319 """ 

320 self.cluster_id = cluster_id 

321 self.region = region 

322 self.webhook_store = webhook_store or get_webhook_store() 

323 self.timeout = timeout 

324 self.max_retries = max_retries 

325 self.retry_delay = retry_delay 

326 self.namespaces = namespaces or ["gco-jobs", "default"] 

327 self.allowed_domains = allowed_domains or [] 

328 

329 # Initialize Kubernetes client 

330 try: 

331 config.load_incluster_config() 

332 logger.info("Loaded in-cluster Kubernetes configuration") 

333 except config.ConfigException: 

334 try: 

335 config.load_kube_config() 

336 logger.info("Loaded local Kubernetes configuration") 

337 except config.ConfigException as e: 

338 logger.error(f"Failed to load Kubernetes configuration: {e}") 

339 raise 

340 

341 self.batch_v1 = client.BatchV1Api() 

342 self.coordination_v1 = client.CoordinationV1Api() 

343 

344 # Timeout for Kubernetes API calls (seconds) 

345 self._k8s_timeout = int(os.environ.get("K8S_API_TIMEOUT", "30")) 

346 

347 # Leader election (None = this process is the only deliverer) 

348 self._leader_lease = leader_lease 

349 self._standby_poll_seconds = standby_poll_seconds 

350 self._is_leader = False 

351 

352 # State tracking 

353 self._job_state_cache = JobStateCache() 

354 self._running = False 

355 self._watch_task: asyncio.Task[None] | None = None 

356 

357 # Metrics 

358 self._deliveries_total = 0 

359 self._deliveries_success = 0 

360 self._deliveries_failed = 0 

361 

362 def _compute_job_status(self, job: V1Job) -> str: 

363 """Compute the effective status of a Kubernetes job.""" 

364 status = job.status 

365 conditions = status.conditions or [] 

366 

367 for condition in conditions: 

368 if condition.type == "Complete" and condition.status == "True": 

369 return "succeeded" 

370 if condition.type == "Failed" and condition.status == "True": 

371 return "failed" 

372 

373 if (status.active or 0) > 0: 

374 return "running" 

375 

376 if (status.succeeded or 0) > 0: 

377 return "succeeded" 

378 

379 if (status.failed or 0) > 0: 

380 return "failed" 

381 

382 return "pending" 

383 

384 def _determine_event( 

385 self, previous_status: str | None, current_status: str 

386 ) -> WebhookEvent | None: 

387 """Determine which webhook event to fire based on status transition.""" 

388 if previous_status is None: 

389 # New job - check if it's already running 

390 if current_status == "running": 

391 return WebhookEvent.JOB_STARTED 

392 return None 

393 

394 # Status transitions 

395 if previous_status in ("pending",) and current_status == "running": 

396 return WebhookEvent.JOB_STARTED 

397 

398 if previous_status in ("pending", "running") and current_status == "succeeded": 

399 return WebhookEvent.JOB_COMPLETED 

400 

401 if previous_status in ("pending", "running") and current_status == "failed": 

402 return WebhookEvent.JOB_FAILED 

403 

404 return None 

405 

406 def _build_payload(self, event: WebhookEvent, job: V1Job) -> dict[str, Any]: 

407 """Build the webhook payload for a job event.""" 

408 metadata = job.metadata 

409 status = job.status 

410 

411 return { 

412 "event": event.value, 

413 "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"), 

414 "cluster_id": self.cluster_id, 

415 "region": self.region, 

416 "job": { 

417 "name": metadata.name, 

418 "namespace": metadata.namespace, 

419 "uid": metadata.uid, 

420 "labels": metadata.labels or {}, 

421 "status": self._compute_job_status(job), 

422 "start_time": (status.start_time.isoformat() if status.start_time else None), 

423 "completion_time": ( 

424 status.completion_time.isoformat() if status.completion_time else None 

425 ), 

426 "active": status.active or 0, 

427 "succeeded": status.succeeded or 0, 

428 "failed": status.failed or 0, 

429 }, 

430 } 

431 

432 def _sign_payload(self, payload: str, secret: str) -> str: 

433 """Sign a payload using HMAC-SHA256.""" 

434 signature = hmac.new( 

435 secret.encode("utf-8"), 

436 payload.encode("utf-8"), 

437 hashlib.sha256, 

438 ).hexdigest() 

439 return f"sha256={signature}" 

440 

441 async def _deliver_webhook( 

442 self, 

443 webhook: dict[str, Any], 

444 payload: dict[str, Any], 

445 ) -> WebhookDeliveryResult: 

446 """Deliver one logical webhook with bounded, address-pinned attempts.""" 

447 raw_webhook_id = webhook.get("id") 

448 webhook_id = raw_webhook_id if isinstance(raw_webhook_id, str) else "<unknown>" 

449 raw_url = webhook.get("url") 

450 url = raw_url if isinstance(raw_url, str) else "" 

451 raw_event = payload.get("event") 

452 event = raw_event if isinstance(raw_event, str) else "unknown" 

453 secret = webhook.get("secret") 

454 start_time = datetime.now(UTC) 

455 self._deliveries_total += 1 

456 

457 target: _ValidatedWebhookTarget | None = None 

458 attempts = 0 

459 last_error: str | None = None 

460 last_status_code: int | None = None 

461 successful_status_code: int | None = None 

462 validation_error: str | None = None 

463 try: 

464 if not isinstance(raw_url, str): 

465 validation_error = "Webhook URL must be a string" 

466 else: 

467 target, validation_error = _resolve_webhook_target( 

468 raw_url, self.allowed_domains or None 

469 ) 

470 if target is None: 

471 last_error = f"URL validation failed: {validation_error}" 

472 logger.warning( 

473 "Webhook URL validation failed: webhook_id=%s error=%s", 

474 webhook_id, 

475 validation_error, 

476 ) 

477 else: 

478 payload_json = json.dumps(payload) 

479 headers = { 

480 "Content-Type": "application/json", 

481 "Host": target.host_header, 

482 "User-Agent": f"GCO-Webhook/{self.cluster_id}", 

483 "X-GCO-Event": event, 

484 "X-GCO-Cluster": self.cluster_id, 

485 "X-GCO-Region": self.region, 

486 } 

487 if secret: 

488 headers["X-GCO-Signature"] = self._sign_payload(payload_json, secret) 

489 

490 # Public trust (the system store) and never a tracing transport: 

491 # webhook URL paths and queries routinely carry credentials, 

492 # and a client span would record the full URL. 

493 async with httpx2.AsyncClient(timeout=self.timeout) as client: 

494 while attempts < self.max_retries: 

495 attempts += 1 

496 try: 

497 response = await client.post( 

498 target.pinned_url(attempts), 

499 content=payload_json, 

500 headers=headers, 

501 extensions={"sni_hostname": target.hostname}, 

502 ) 

503 last_status_code = response.status_code 

504 if 200 <= response.status_code < 300: 

505 successful_status_code = response.status_code 

506 break 

507 

508 last_error = f"HTTP {response.status_code}" 

509 if response.status_code >= 500: 

510 logger.warning( 

511 "Webhook attempt failed: webhook_id=%s %s status=%s attempt=%s", 

512 webhook_id, 

513 target.log_identity, 

514 response.status_code, 

515 attempts, 

516 ) 

517 if attempts < self.max_retries: 

518 await asyncio.sleep(self.retry_delay * (2 ** (attempts - 1))) 

519 continue 

520 break # Caller errors are terminal and must not be replayed. 

521 

522 except httpx2.TimeoutException: 

523 last_error = "Request timed out" 

524 logger.warning( 

525 "Webhook attempt timed out: webhook_id=%s %s attempt=%s", 

526 webhook_id, 

527 target.log_identity, 

528 attempts, 

529 ) 

530 if attempts < self.max_retries: 

531 await asyncio.sleep(self.retry_delay * (2 ** (attempts - 1))) 

532 except httpx2.RequestError as exc: 

533 last_error = type(exc).__name__ 

534 logger.warning( 

535 "Webhook transport failed: webhook_id=%s %s " 

536 "error_type=%s attempt=%s", 

537 webhook_id, 

538 target.log_identity, 

539 type(exc).__name__, 

540 attempts, 

541 ) 

542 if attempts < self.max_retries: 

543 await asyncio.sleep(self.retry_delay * (2 ** (attempts - 1))) 

544 

545 # Exiting AsyncClient may itself await and be cancelled. Commit 

546 # success only after teardown completes so one delivery has 

547 # exactly one terminal metric classification. 

548 if successful_status_code is not None: 

549 duration = (datetime.now(UTC) - start_time).total_seconds() * 1000 

550 logger.info( 

551 "Webhook delivered: webhook_id=%s %s status=%s attempts=%s", 

552 webhook_id, 

553 target.log_identity, 

554 successful_status_code, 

555 attempts, 

556 ) 

557 self._deliveries_success += 1 

558 return WebhookDeliveryResult( 

559 webhook_id=webhook_id, 

560 url=url, 

561 event=event, 

562 success=True, 

563 status_code=successful_status_code, 

564 attempts=attempts, 

565 duration_ms=duration, 

566 ) 

567 except asyncio.CancelledError: 

568 # Cancellation is a terminal outcome for this logical delivery. 

569 # Preserve task cancellation while keeping total == success + failed. 

570 self._deliveries_failed += 1 

571 logger.info("Webhook delivery cancelled: webhook_id=%s", webhook_id) 

572 raise 

573 except Exception as exc: # sanitize the logical delivery boundary 

574 last_error = f"Delivery failure: {type(exc).__name__}" 

575 logger.error( 

576 "Webhook delivery raised: webhook_id=%s error_type=%s", 

577 webhook_id, 

578 type(exc).__name__, 

579 ) 

580 

581 duration = (datetime.now(UTC) - start_time).total_seconds() * 1000 

582 log_identity = target.log_identity if target is not None else "target=unvalidated" 

583 logger.error( 

584 "Webhook delivery failed: webhook_id=%s %s attempts=%s status=%s error=%s", 

585 webhook_id, 

586 log_identity, 

587 attempts, 

588 last_status_code, 

589 last_error, 

590 ) 

591 self._deliveries_failed += 1 

592 return WebhookDeliveryResult( 

593 webhook_id=webhook_id, 

594 url=url, 

595 event=event, 

596 success=False, 

597 status_code=last_status_code, 

598 error=last_error, 

599 attempts=attempts, 

600 duration_ms=duration, 

601 ) 

602 

603 async def _dispatch_event(self, event: WebhookEvent, job: V1Job) -> list[WebhookDeliveryResult]: 

604 """Dispatch webhooks for a job event.""" 

605 namespace = job.metadata.namespace 

606 payload = self._build_payload(event, job) 

607 

608 # Get webhooks subscribed to this event 

609 try: 

610 # Get webhooks for this specific namespace 

611 namespace_webhooks = self.webhook_store.get_webhooks_for_event( 

612 event.value, namespace=namespace 

613 ) 

614 # Get global webhooks (no namespace filter) 

615 global_webhooks = self.webhook_store.get_webhooks_for_event(event.value, namespace=None) 

616 

617 # Combine and deduplicate 

618 all_webhooks = {w["id"]: w for w in namespace_webhooks} 

619 for w in global_webhooks: 

620 if w.get("namespace") is None: # Only add truly global webhooks 

621 all_webhooks[w["id"]] = w 

622 

623 webhooks = list(all_webhooks.values()) 

624 

625 except Exception as exc: 

626 logger.error( 

627 "Failed to get webhooks for event %s: error_type=%s", 

628 event.value, 

629 type(exc).__name__, 

630 ) 

631 return [] 

632 

633 if not webhooks: 

634 logger.debug(f"No webhooks registered for event {event.value} in namespace {namespace}") 

635 return [] 

636 

637 logger.info( 

638 f"Dispatching {len(webhooks)} webhooks for {event.value} " 

639 f"(job={job.metadata.name}, namespace={namespace})" 

640 ) 

641 

642 # Dispatch all webhooks concurrently. _deliver_webhook owns normal 

643 # exception conversion; this boundary protects accounting and redaction 

644 # if a future regression (or an injected implementation) escapes it. 

645 tasks = [self._deliver_webhook(webhook, payload) for webhook in webhooks] 

646 results = await asyncio.gather(*tasks, return_exceptions=True) 

647 

648 delivery_results: list[WebhookDeliveryResult] = [] 

649 for webhook, result in zip(webhooks, results, strict=True): 

650 if isinstance(result, WebhookDeliveryResult): 

651 delivery_results.append(result) 

652 continue 

653 if isinstance(result, asyncio.CancelledError): 

654 raise result 

655 raw_webhook_id = webhook.get("id") 

656 webhook_id = raw_webhook_id if isinstance(raw_webhook_id, str) else "<unknown>" 

657 raw_url = webhook.get("url") 

658 url = raw_url if isinstance(raw_url, str) else "" 

659 logger.error( 

660 "Webhook delivery escaped boundary: webhook_id=%s error_type=%s", 

661 webhook_id, 

662 type(result).__name__, 

663 ) 

664 self._deliveries_total += 1 

665 self._deliveries_failed += 1 

666 delivery_results.append( 

667 WebhookDeliveryResult( 

668 webhook_id=webhook_id, 

669 url=url, 

670 event=event.value, 

671 success=False, 

672 error=f"Delivery failure: {type(result).__name__}", 

673 attempts=0, 

674 duration_ms=0.0, 

675 ) 

676 ) 

677 

678 return delivery_results 

679 

680 async def _process_job_event(self, event_type: str, job: V1Job) -> None: 

681 """Process a Kubernetes job event.""" 

682 job_uid = job.metadata.uid 

683 job_name = job.metadata.name 

684 namespace = job.metadata.namespace 

685 

686 # Skip system namespaces 

687 if namespace in ("kube-system", "kube-public", "kube-node-lease"): 

688 return 

689 

690 # Skip if not in watched namespaces (if specified) 

691 if self.namespaces and namespace not in self.namespaces: 

692 return 

693 

694 current_status = self._compute_job_status(job) 

695 

696 if event_type == "DELETED": 

697 self._job_state_cache.remove(job_uid) 

698 return 

699 

700 # Get previous state and update cache 

701 previous_status = self._job_state_cache.set_state(job_uid, current_status) 

702 

703 # Determine if we should fire an event 

704 webhook_event = self._determine_event(previous_status, current_status) 

705 

706 if webhook_event: 

707 logger.info( 

708 f"Job status transition: {job_name} ({namespace}) " 

709 f"{previous_status or 'new'} -> {current_status} " 

710 f"-> firing {webhook_event.value}" 

711 ) 

712 await self._dispatch_event(webhook_event, job) 

713 

714 def _sync_watch_jobs(self) -> list[tuple[str, Any]]: 

715 """ 

716 Synchronous job watcher that yields batches of events. 

717 This runs in a thread executor to avoid blocking the async event loop. 

718 """ 

719 w = Watch() 

720 events = [] 

721 

722 try: 

723 # Watch jobs with a short timeout to allow periodic returns 

724 for event in w.stream( 

725 self.batch_v1.list_job_for_all_namespaces, 

726 timeout_seconds=30, # Short timeout to return control periodically 

727 ): 

728 if not self._running: 

729 break 

730 

731 events.append((event["type"], event["object"])) 

732 

733 # Return batch after collecting some events or if we have any 

734 if len(events) >= 10: 

735 break 

736 

737 except ApiException as e: 

738 if e.status == 410: # Gone - resource version too old 

739 logger.warning("Watch expired, will restart...") 

740 else: 

741 logger.error(f"Kubernetes API error in job watcher: {e}") 

742 raise 

743 except Exception as e: 

744 logger.error(f"Error in sync job watcher: {e}") 

745 raise 

746 

747 return events 

748 

749 async def _acquire_leadership(self) -> bool: 

750 """Hold (or keep holding) the deliverer Lease; ``True`` means deliver. 

751 

752 Off the event loop because the Lease read/replace is a blocking 

753 Kubernetes round trip. Winning the Lease re-seeds the job-state cache: 

754 the new leader must not replay transitions the old one already 

755 delivered, and anything that happened while nobody held the Lease is 

756 deliberately dropped rather than duplicated. 

757 """ 

758 if self._leader_lease is None: 

759 return True 

760 held = await asyncio.to_thread( 

761 try_acquire_lease, 

762 self.coordination_v1, 

763 self._leader_lease, 

764 label="webhook", 

765 ) 

766 if held and not self._is_leader: 

767 logger.info( 

768 "Webhook dispatcher %s became the deliverer; re-seeding job state cache", 

769 self._leader_lease.holder, 

770 ) 

771 await self._initialize_job_cache() 

772 elif not held and self._is_leader: 

773 logger.warning("Webhook dispatcher lost the deliverer Lease; standing by") 

774 self._is_leader = held 

775 return held 

776 

777 async def _watch_jobs(self) -> None: 

778 """Watch Kubernetes jobs for status changes using thread executor.""" 

779 logger.info(f"Starting job watcher for namespaces: {self.namespaces}") 

780 

781 while self._running: 

782 try: 

783 if not await self._acquire_leadership(): 

784 await asyncio.sleep(self._standby_poll_seconds) 

785 continue 

786 

787 # Run the synchronous watch in a thread executor 

788 events = await asyncio.to_thread(self._sync_watch_jobs) 

789 

790 # Process collected events. ``stop()`` flips ``_running`` from 

791 # another task while this one is awaiting the watch thread, so 

792 # the flag is re-read per event; mypy still sees the ``while`` 

793 # condition's narrowing and calls the re-check unreachable. 

794 for event_type, job in events: 

795 if not self._running: 

796 break # type: ignore[unreachable] 

797 

798 try: 

799 await self._process_job_event(event_type, job) 

800 except Exception as e: 

801 logger.error(f"Error processing job event: {e}") 

802 

803 # Small delay between watch cycles if no events 

804 if not events: 

805 await asyncio.sleep(1) 

806 

807 except Exception as e: 

808 logger.error(f"Error in job watcher: {e}") 

809 await asyncio.sleep(5) 

810 

811 async def start(self) -> None: 

812 """Start the webhook dispatcher.""" 

813 if self._running: 

814 logger.warning("Webhook dispatcher already running") 

815 return 

816 

817 self._running = True 

818 logger.info(f"Starting webhook dispatcher for cluster {self.cluster_id}") 

819 

820 if self._leader_lease is None: 

821 # Single deliverer: seed the job state cache with current jobs now. 

822 # With a Lease the cache is seeded at the moment leadership is won, 

823 # so a standby never fires transitions the leader already delivered. 

824 await self._initialize_job_cache() 

825 

826 # Start the watch task 

827 self._watch_task = asyncio.create_task(self._watch_jobs()) 

828 

829 async def stop(self) -> None: 

830 """Stop the webhook dispatcher.""" 

831 logger.info("Stopping webhook dispatcher") 

832 self._running = False 

833 

834 if self._watch_task: 

835 self._watch_task.cancel() 

836 with contextlib.suppress(asyncio.CancelledError): 

837 await self._watch_task 

838 self._watch_task = None 

839 

840 async def _initialize_job_cache(self) -> None: 

841 """Initialize the job state cache with current job states.""" 

842 try: 

843 # Off the event loop: the synchronous client would otherwise stall 

844 # every other task for the duration of the list (and its retries). 

845 jobs = await asyncio.to_thread( 

846 self.batch_v1.list_job_for_all_namespaces, 

847 _request_timeout=self._k8s_timeout, 

848 ) 

849 for job in jobs.items: 

850 namespace = job.metadata.namespace 

851 if namespace in ("kube-system", "kube-public", "kube-node-lease"): 

852 continue 

853 if self.namespaces and namespace not in self.namespaces: 

854 continue 

855 

856 job_uid = job.metadata.uid 

857 status = self._compute_job_status(job) 

858 self._job_state_cache.set_state(job_uid, status) 

859 

860 logger.info(f"Initialized job cache with {len(self._job_state_cache.job_states)} jobs") 

861 except Exception as e: