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 httpx 

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-09-11T22:27:39Z 

91# Generated from Git commit: 5d1a9122b6630246e01cafdaf458d11c2da8b4ae 

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 if family not in {socket.AF_INET, socket.AF_INET6}: 

204 return None, "DNS resolution returned an unsupported address family" 

205 raw_address = sockaddr[0] 

206 if not isinstance(raw_address, str): 

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

208 address = _globally_routable_unicast(raw_address) 

209 if address is None: 

210 return None, ( 

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

212 ) 

213 addresses.add(str(address)) 

214 if not addresses: 

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

216 

217 return ( 

218 _ValidatedWebhookTarget( 

219 original_url=url, 

220 parsed=parsed, 

221 hostname=hostname, 

222 port=port, 

223 addresses=tuple(sorted(addresses)), 

224 ), 

225 None, 

226 ) 

227 

228 

229def validate_webhook_url( 

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

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

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

233 target, error = _resolve_webhook_target(url, allowed_domains) 

234 return target is not None, error 

235 

236 

237class WebhookEvent(StrEnum): 

238 """Webhook event types.""" 

239 

240 JOB_STARTED = "job.started" 

241 JOB_COMPLETED = "job.completed" 

242 JOB_FAILED = "job.failed" 

243 

244 

245@dataclass 

246class WebhookDeliveryResult: 

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

248 

249 webhook_id: str 

250 url: str 

251 event: str 

252 success: bool 

253 status_code: int | None = None 

254 error: str | None = None 

255 attempts: int = 1 

256 duration_ms: float = 0.0 

257 

258 

259@dataclass 

260class JobStateCache: 

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

262 

263 # Map of job_uid -> last known status 

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

265 

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

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

268 return self.job_states.get(job_uid) 

269 

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

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

272 previous = self.job_states.get(job_uid) 

273 self.job_states[job_uid] = state 

274 return previous 

275 

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

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

278 self.job_states.pop(job_uid, None) 

279 

280 

281class WebhookDispatcher: 

282 """ 

283 Dispatches webhook notifications for Kubernetes job events. 

284 

285 This class monitors job status changes and sends HTTP notifications 

286 to registered webhook endpoints. 

287 """ 

288 

289 def __init__( 

290 self, 

291 cluster_id: str, 

292 region: str, 

293 webhook_store: WebhookStore | None = None, 

294 timeout: int = 30, 

295 max_retries: int = 3, 

296 retry_delay: int = 5, 

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

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

299 leader_lease: LeaseIdentity | None = None, 

300 standby_poll_seconds: float = 5.0, 

301 ): 

302 """Initialize the webhook dispatcher. 

303 

304 Args: 

305 cluster_id: EKS cluster identifier 

306 region: AWS region 

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

308 timeout: HTTP timeout for webhook calls in seconds 

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

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

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

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

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

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

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

316 """ 

317 self.cluster_id = cluster_id 

318 self.region = region 

319 self.webhook_store = webhook_store or get_webhook_store() 

320 self.timeout = timeout 

321 self.max_retries = max_retries 

322 self.retry_delay = retry_delay 

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

324 self.allowed_domains = allowed_domains or [] 

325 

326 # Initialize Kubernetes client 

327 try: 

328 config.load_incluster_config() 

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

330 except config.ConfigException: 

331 try: 

332 config.load_kube_config() 

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

334 except config.ConfigException as e: 

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

336 raise 

337 

338 self.batch_v1 = client.BatchV1Api() 

339 self.coordination_v1 = client.CoordinationV1Api() 

340 

341 # Timeout for Kubernetes API calls (seconds) 

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

343 

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

345 self._leader_lease = leader_lease 

346 self._standby_poll_seconds = standby_poll_seconds 

347 self._is_leader = False 

348 

349 # State tracking 

350 self._job_state_cache = JobStateCache() 

351 self._running = False 

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

353 

354 # Metrics 

355 self._deliveries_total = 0 

356 self._deliveries_success = 0 

357 self._deliveries_failed = 0 

358 

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

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

361 status = job.status 

362 conditions = status.conditions or [] 

363 

364 for condition in conditions: 

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

366 return "succeeded" 

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

368 return "failed" 

369 

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

371 return "running" 

372 

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

374 return "succeeded" 

375 

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

377 return "failed" 

378 

379 return "pending" 

380 

381 def _determine_event( 

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

383 ) -> WebhookEvent | None: 

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

385 if previous_status is None: 

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

387 if current_status == "running": 

388 return WebhookEvent.JOB_STARTED 

389 return None 

390 

391 # Status transitions 

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

393 return WebhookEvent.JOB_STARTED 

394 

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

396 return WebhookEvent.JOB_COMPLETED 

397 

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

399 return WebhookEvent.JOB_FAILED 

400 

401 return None 

402 

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

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

405 metadata = job.metadata 

406 status = job.status 

407 

408 return { 

409 "event": event.value, 

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

411 "cluster_id": self.cluster_id, 

412 "region": self.region, 

413 "job": { 

414 "name": metadata.name, 

415 "namespace": metadata.namespace, 

416 "uid": metadata.uid, 

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

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

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

420 "completion_time": ( 

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

422 ), 

423 "active": status.active or 0, 

424 "succeeded": status.succeeded or 0, 

425 "failed": status.failed or 0, 

426 }, 

427 } 

428 

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

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

431 signature = hmac.new( 

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

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

434 hashlib.sha256, 

435 ).hexdigest() 

436 return f"sha256={signature}" 

437 

438 async def _deliver_webhook( 

439 self, 

440 webhook: dict[str, Any], 

441 payload: dict[str, Any], 

442 ) -> WebhookDeliveryResult: 

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

444 raw_webhook_id = webhook.get("id") 

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

446 raw_url = webhook.get("url") 

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

448 raw_event = payload.get("event") 

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

450 secret = webhook.get("secret") 

451 start_time = datetime.now(UTC) 

452 self._deliveries_total += 1 

453 

454 target: _ValidatedWebhookTarget | None = None 

455 attempts = 0 

456 last_error: str | None = None 

457 last_status_code: int | None = None 

458 successful_status_code: int | None = None 

459 validation_error: str | None = None 

460 try: 

461 if not isinstance(raw_url, str): 

462 validation_error = "Webhook URL must be a string" 

463 else: 

464 target, validation_error = _resolve_webhook_target( 

465 raw_url, self.allowed_domains or None 

466 ) 

467 if target is None: 

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

469 logger.warning( 

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

471 webhook_id, 

472 validation_error, 

473 ) 

474 else: 

475 payload_json = json.dumps(payload) 

476 headers = { 

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

478 "Host": target.host_header, 

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

480 "X-GCO-Event": event, 

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

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

483 } 

484 if secret: 

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

486 

487 async with httpx.AsyncClient(timeout=self.timeout) as client: 

488 while attempts < self.max_retries: 

489 attempts += 1 

490 try: 

491 response = await client.post( 

492 target.pinned_url(attempts), 

493 content=payload_json, 

494 headers=headers, 

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

496 ) 

497 last_status_code = response.status_code 

498 if 200 <= response.status_code < 300: 

499 successful_status_code = response.status_code 

500 break 

501 

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

503 if response.status_code >= 500: 

504 logger.warning( 

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

506 webhook_id, 

507 target.log_identity, 

508 response.status_code, 

509 attempts, 

510 ) 

511 if attempts < self.max_retries: 

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

513 continue 

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

515 

516 except httpx.TimeoutException: 

517 last_error = "Request timed out" 

518 logger.warning( 

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

520 webhook_id, 

521 target.log_identity, 

522 attempts, 

523 ) 

524 if attempts < self.max_retries: 

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

526 except httpx.RequestError as exc: 

527 last_error = type(exc).__name__ 

528 logger.warning( 

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

530 "error_type=%s attempt=%s", 

531 webhook_id, 

532 target.log_identity, 

533 type(exc).__name__, 

534 attempts, 

535 ) 

536 if attempts < self.max_retries: 

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

538 

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

540 # success only after teardown completes so one delivery has 

541 # exactly one terminal metric classification. 

542 if successful_status_code is not None: 

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

544 logger.info( 

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

546 webhook_id, 

547 target.log_identity, 

548 successful_status_code, 

549 attempts, 

550 ) 

551 self._deliveries_success += 1 

552 return WebhookDeliveryResult( 

553 webhook_id=webhook_id, 

554 url=url, 

555 event=event, 

556 success=True, 

557 status_code=successful_status_code, 

558 attempts=attempts, 

559 duration_ms=duration, 

560 ) 

561 except asyncio.CancelledError: 

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

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

564 self._deliveries_failed += 1 

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

566 raise 

567 except Exception as exc: # noqa: BLE001 — sanitize the logical delivery boundary 

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

569 logger.error( 

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

571 webhook_id, 

572 type(exc).__name__, 

573 ) 

574 

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

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

577 logger.error( 

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

579 webhook_id, 

580 log_identity, 

581 attempts, 

582 last_status_code, 

583 last_error, 

584 ) 

585 self._deliveries_failed += 1 

586 return WebhookDeliveryResult( 

587 webhook_id=webhook_id, 

588 url=url, 

589 event=event, 

590 success=False, 

591 status_code=last_status_code, 

592 error=last_error, 

593 attempts=attempts, 

594 duration_ms=duration, 

595 ) 

596 

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

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

599 namespace = job.metadata.namespace 

600 payload = self._build_payload(event, job) 

601 

602 # Get webhooks subscribed to this event 

603 try: 

604 # Get webhooks for this specific namespace 

605 namespace_webhooks = self.webhook_store.get_webhooks_for_event( 

606 event.value, namespace=namespace 

607 ) 

608 # Get global webhooks (no namespace filter) 

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

610 

611 # Combine and deduplicate 

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

613 for w in global_webhooks: 

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

615 all_webhooks[w["id"]] = w 

616 

617 webhooks = list(all_webhooks.values()) 

618 

619 except Exception as exc: 

620 logger.error( 

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

622 event.value, 

623 type(exc).__name__, 

624 ) 

625 return [] 

626 

627 if not webhooks: 

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

629 return [] 

630 

631 logger.info( 

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

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

634 ) 

635 

636 # Dispatch all webhooks concurrently. _deliver_webhook owns normal 

637 # exception conversion; this boundary protects accounting and redaction 

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

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

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

641 

642 delivery_results: list[WebhookDeliveryResult] = [] 

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

644 if isinstance(result, WebhookDeliveryResult): 

645 delivery_results.append(result) 

646 continue 

647 if isinstance(result, asyncio.CancelledError): 

648 raise result 

649 raw_webhook_id = webhook.get("id") 

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

651 raw_url = webhook.get("url") 

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

653 logger.error( 

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

655 webhook_id, 

656 type(result).__name__, 

657 ) 

658 self._deliveries_total += 1 

659 self._deliveries_failed += 1 

660 delivery_results.append( 

661 WebhookDeliveryResult( 

662 webhook_id=webhook_id, 

663 url=url, 

664 event=event.value, 

665 success=False, 

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

667 attempts=0, 

668 duration_ms=0.0, 

669 ) 

670 ) 

671 

672 return delivery_results 

673 

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

675 """Process a Kubernetes job event.""" 

676 job_uid = job.metadata.uid 

677 job_name = job.metadata.name 

678 namespace = job.metadata.namespace 

679 

680 # Skip system namespaces 

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

682 return 

683 

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

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

686 return 

687 

688 current_status = self._compute_job_status(job) 

689 

690 if event_type == "DELETED": 

691 self._job_state_cache.remove(job_uid) 

692 return 

693 

694 # Get previous state and update cache 

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

696 

697 # Determine if we should fire an event 

698 webhook_event = self._determine_event(previous_status, current_status) 

699 

700 if webhook_event: 

701 logger.info( 

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

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

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

705 ) 

706 await self._dispatch_event(webhook_event, job) 

707 

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

709 """ 

710 Synchronous job watcher that yields batches of events. 

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

712 """ 

713 w = Watch() 

714 events = [] 

715 

716 try: 

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

718 for event in w.stream( 

719 self.batch_v1.list_job_for_all_namespaces, 

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

721 ): 

722 if not self._running: 

723 break 

724 

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

726 

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

728 if len(events) >= 10: 

729 break 

730 

731 except ApiException as e: 

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

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

734 else: 

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

736 raise 

737 except Exception as e: 

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

739 raise 

740 

741 return events 

742 

743 async def _acquire_leadership(self) -> bool: 

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

745 

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

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

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

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

750 deliberately dropped rather than duplicated. 

751 """ 

752 if self._leader_lease is None: 

753 return True 

754 held = await asyncio.to_thread( 

755 try_acquire_lease, 

756 self.coordination_v1, 

757 self._leader_lease, 

758 label="webhook", 

759 ) 

760 if held and not self._is_leader: 

761 logger.info( 

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

763 self._leader_lease.holder, 

764 ) 

765 await self._initialize_job_cache() 

766 elif not held and self._is_leader: 

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

768 self._is_leader = held 

769 return held 

770 

771 async def _watch_jobs(self) -> None: 

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

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

774 

775 while self._running: 

776 try: 

777 if not await self._acquire_leadership(): 

778 await asyncio.sleep(self._standby_poll_seconds) 

779 continue 

780 

781 # Run the synchronous watch in a thread executor 

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

783 

784 # Process collected events 

785 for event_type, job in events: 

786 if not self._running: 

787 break 

788 

789 try: 

790 await self._process_job_event(event_type, job) 

791 except Exception as e: 

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

793 

794 # Small delay between watch cycles if no events 

795 if not events: 

796 await asyncio.sleep(1) 

797 

798 except Exception as e: 

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

800 await asyncio.sleep(5) 

801 

802 async def start(self) -> None: 

803 """Start the webhook dispatcher.""" 

804 if self._running: 

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

806 return 

807 

808 self._running = True 

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

810 

811 if self._leader_lease is None: 

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

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

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

815 await self._initialize_job_cache() 

816 

817 # Start the watch task 

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

819 

820 async def stop(self) -> None: 

821 """Stop the webhook dispatcher.""" 

822 logger.info("Stopping webhook dispatcher") 

823 self._running = False 

824 

825 if self._watch_task: 

826 self._watch_task.cancel() 

827 with contextlib.suppress(asyncio.CancelledError): 

828 await self._watch_task 

829 self._watch_task = None 

830 

831 async def _initialize_job_cache(self) -> None: 

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

833 try: 

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

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

836 jobs = await asyncio.to_thread( 

837 self.batch_v1.list_job_for_all_namespaces, 

838 _request_timeout=self._k8s_timeout, 

839 ) 

840 for job in jobs.items: 

841 namespace = job.metadata.namespace 

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

843 continue 

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

845 continue 

846 

847 job_uid = job.metadata.uid 

848 status = self._compute_job_status(job) 

849 self._job_state_cache.set_state(job_uid, status) 

850 

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

852 except Exception as e: 

853 logger.error(f"Failed to initialize job cache: {e}") 

854 

855 def get_metrics(self) -> dict[str, Any]: 

856 """Get dispatcher metrics.""" 

857 return {