Coverage for cli/aws_client.py: 94.88%

1""" 

2AWS Client utilities for GCO CLI. 

3 

4Provides authenticated access to AWS services with SigV4 signing, 

5stack discovery, and region management. 

6""" 

7 

8import json 

9import logging 

10import time 

11from dataclasses import dataclass 

12from datetime import datetime 

13from typing import Any 

14from urllib.parse import quote 

15 

16import boto3 

17import requests 

18from botocore.auth import SigV4Auth 

19from botocore.awsrequest import AWSRequest 

20 

21from .config import GCOConfig, get_config 

22 

23logger = logging.getLogger(__name__) 

24 

25# HTTP status codes that are safe to retry (transient failures) 

26_RETRYABLE_STATUS_CODES = {429, 502, 503, 504} 

27_MAX_RETRIES = 3 

28_RETRY_BACKOFF_BASE = 1.0 # seconds 

29 

30 

31def _validate_max_attempts(max_attempts: int | None) -> None: 

32 """Validate an explicitly supplied request-attempt limit.""" 

33 if max_attempts is not None and ( 

34 isinstance(max_attempts, bool) or not isinstance(max_attempts, int) or max_attempts <= 0 

35 ): 

36 raise ValueError("max_attempts must be a positive integer") 

37 

38 

39@dataclass 

40class RegionalStack: 

41 """Information about a regional GCO stack.""" 

42 

43 region: str 

44 stack_name: str 

45 cluster_name: str 

46 status: str 

47 api_endpoint: str | None = None 

48 efs_file_system_id: str | None = None 

49 fsx_file_system_id: str | None = None 

50 created_time: datetime | None = None 

51 

52 

53@dataclass 

54class ApiEndpoint: 

55 """API Gateway endpoint information.""" 

56 

57 url: str 

58 region: str 

59 api_id: str 

60 is_regional: bool = False # True if this is a regional API (for private access) 

61 

62 

63class GCOAWSClient: 

64 """ 

65 AWS client for GCO operations. 

66 

67 Handles: 

68 - Stack discovery across regions 

69 - Authenticated API requests with SigV4 

70 - CloudFormation stack queries 

71 - EKS cluster information 

72 """ 

73 

74 def __init__(self, config: GCOConfig | None = None): 

75 self.config = config or get_config() 

76 self._session = boto3.Session() 

77 self._api_endpoint_cache: ApiEndpoint | None = None 

78 self._regional_api_cache: dict[str, ApiEndpoint] = {} 

79 self._regional_stacks_cache: dict[str, RegionalStack] | None = None 

80 self._cache_timestamp: float | None = None 

81 self._use_regional_api = getattr(self.config, "use_regional_api", False) is True 

82 

83 def _is_cache_valid(self) -> bool: 

84 """Check if cache is still valid.""" 

85 if self._cache_timestamp is None: 

86 return False 

87 return (time.time() - self._cache_timestamp) < self.config.cache_ttl_seconds 

88 

89 def _invalidate_cache(self) -> None: 

90 """Invalidate all caches.""" 

91 self._api_endpoint_cache = None 

92 self._regional_api_cache = {} 

93 self._regional_stacks_cache = None 

94 self._cache_timestamp = None 

95 

96 def set_use_regional_api(self, use_regional: bool) -> None: 

97 """Set whether to use regional APIs instead of global API. 

98 

99 When enabled, API calls will be routed through regional API Gateways 

100 that use VPC Lambdas to access internal ALBs. This is required when 

101 public access is disabled. 

102 

103 Args: 

104 use_regional: True to use regional APIs, False for global API 

105 """ 

106 self._use_regional_api = use_regional 

107 

108 def get_regional_api_endpoint( 

109 self, region: str, force_refresh: bool = False 

110 ) -> ApiEndpoint | None: 

111 """ 

112 Get the regional API Gateway endpoint for a specific region. 

113 

114 Regional APIs are used when public access is disabled and the ALB 

115 is internal-only. 

116 

117 Args: 

118 region: AWS region 

119 force_refresh: Force refresh from CloudFormation 

120 

121 Returns: 

122 ApiEndpoint with URL and metadata, or None if not found 

123 """ 

124 if not force_refresh and region in self._regional_api_cache and self._is_cache_valid(): 

125 return self._regional_api_cache[region] 

126 

127 cfn = self._session.client("cloudformation", region_name=region) 

128 stack_name = f"{self.config.project_name}-regional-api-{region}" 

129 

130 try: 

131 response = cfn.describe_stacks(StackName=stack_name) 

132 stack = response["Stacks"][0] 

133 

134 api_url = None 

135 for output in stack.get("Outputs", []): 

136 if output["OutputKey"] == "RegionalApiEndpoint": 

137 api_url = output["OutputValue"].rstrip("/") 

138 break 

139 

140 if not api_url: 

141 return None 

142 

143 # Extract API ID from URL 

144 api_id = api_url.split(".")[0].replace("https://", "") 

145 

146 endpoint = ApiEndpoint(url=api_url, region=region, api_id=api_id, is_regional=True) 

147 self._regional_api_cache[region] = endpoint 

148 return endpoint 

149 

150 except cfn.exceptions.ClientError: 

151 # Stack doesn't exist 

152 return None 

153 except Exception as e: 

154 logger.debug("Failed to get regional API endpoint for %s: %s", region, e) 

155 return None 

156 

157 def get_api_endpoint(self, force_refresh: bool = False) -> ApiEndpoint: 

158 """ 

159 Get the global API Gateway endpoint. 

160 

161 Args: 

162 force_refresh: Force refresh from CloudFormation 

163 

164 Returns: 

165 ApiEndpoint with URL and metadata 

166 """ 

167 if not force_refresh and self._api_endpoint_cache and self._is_cache_valid(): 

168 return self._api_endpoint_cache 

169 

170 cfn = self._session.client("cloudformation", region_name=self.config.api_gateway_region) 

171 

172 try: 

173 response = cfn.describe_stacks(StackName=self.config.api_gateway_stack_name) 

174 stack = response["Stacks"][0] 

175 

176 api_url = None 

177 for output in stack.get("Outputs", []): 

178 if output["OutputKey"] == "ApiEndpoint": 178 ↛ 177line 178 didn't jump to line 177 because the condition on line 178 was always true

179 api_url = output["OutputValue"].rstrip("/") 

180 break 

181 

182 if not api_url: 

183 raise ValueError( 

184 f"ApiEndpoint not found in stack {self.config.api_gateway_stack_name}" 

185 ) 

186 

187 # Extract API ID from URL 

188 # Format: https://{api-id}.execute-api.{region}.amazonaws.com/prod 

189 api_id = api_url.split(".")[0].replace("https://", "") 

190 

191 self._api_endpoint_cache = ApiEndpoint( 

192 url=api_url, region=self.config.api_gateway_region, api_id=api_id 

193 ) 

194 self._cache_timestamp = time.time() 

195 

196 return self._api_endpoint_cache 

197 

198 except Exception as e: 

199 raise RuntimeError(f"Failed to get API endpoint: {e}") from e 

200 

201 def discover_regional_stacks(self, force_refresh: bool = False) -> dict[str, RegionalStack]: 

202 """ 

203 Discover all regional GCO stacks. 

204 

205 Checks configured regions from cdk.json first for fast discovery, 

206 then falls back to scanning all AWS regions if no stacks are found. 

207 

208 Args: 

209 force_refresh: Force refresh from CloudFormation 

210 

211 Returns: 

212 Dictionary mapping region to RegionalStack 

213 """ 

214 if not force_refresh and self._regional_stacks_cache and self._is_cache_valid(): 

215 return self._regional_stacks_cache 

216 

217 regional_stacks: dict[str, RegionalStack] = {} 

218 

219 # Try configured regions first (fast path) 

220 configured_regions = self._get_configured_regions() 

221 if configured_regions: 221 ↛ 228line 221 didn't jump to line 228 because the condition on line 221 was always true

222 for region in configured_regions: 

223 stack = self._probe_regional_stack(region) 

224 if stack: 

225 regional_stacks[region] = stack 

226 

227 # If we found stacks in configured regions, skip the full scan 

228 if not regional_stacks: 

229 # Fall back to scanning all regions 

230 logger.debug("No stacks found in configured regions, scanning all AWS regions") 

231 ec2 = self._session.client("ec2", region_name="us-east-1") 

232 regions_response = ec2.describe_regions() 

233 all_regions = [r["RegionName"] for r in regions_response["Regions"]] 

234 

235 for region in all_regions: 

236 if region in configured_regions: 

237 continue # Already checked 

238 stack = self._probe_regional_stack(region) 

239 if stack: 239 ↛ 235line 239 didn't jump to line 235 because the condition on line 239 was always true

240 regional_stacks[region] = stack 

241 

242 self._regional_stacks_cache = regional_stacks 

243 self._cache_timestamp = time.time() 

244 

245 return regional_stacks 

246 

247 def _get_configured_regions(self) -> list[str]: 

248 """Get the list of configured deployment regions from cdk.json.""" 

249 from .config import _load_cdk_json 

250 

251 cdk_regions = _load_cdk_json() 

252 regions: list[str] = cdk_regions.get("regional", []) 

253 return regions 

254 

255 def _probe_regional_stack(self, region: str) -> RegionalStack | None: 

256 """Probe a single region for a GCO regional stack. 

257 

258 Args: 

259 region: AWS region to check 

260 

261 Returns: 

262 RegionalStack if found, None otherwise 

263 """ 

264 try: 

265 cfn = self._session.client("cloudformation", region_name=region) 

266 stack_name = f"{self.config.regional_stack_prefix}-{region}" 

267 

268 try: 

269 response = cfn.describe_stacks(StackName=stack_name) 

270 stack = response["Stacks"][0] 

271 

272 outputs = {o["OutputKey"]: o["OutputValue"] for o in stack.get("Outputs", [])} 

273 

274 return RegionalStack( 

275 region=region, 

276 stack_name=stack_name, 

277 cluster_name=outputs.get("ClusterName", f"{self.config.project_name}-{region}"), 

278 status=stack["StackStatus"], 

279 efs_file_system_id=outputs.get("EfsFileSystemId"), 

280 fsx_file_system_id=outputs.get("FsxFileSystemId"), 

281 created_time=stack.get("CreationTime"), 

282 ) 

283 except cfn.exceptions.ClientError: 

284 return None 

285 

286 except Exception as e: 

287 logger.debug("Failed to get regional stack info for %s: %s", region, e) 

288 return None 

289 

290 def get_regional_stack(self, region: str) -> RegionalStack | None: 

291 """Get information about a specific regional stack.""" 

292 stacks = self.discover_regional_stacks() 

293 return stacks.get(region) 

294 

295 def call_api( 

296 self, 

297 method: str, 

298 path: str, 

299 region: str | None = None, 

300 body: dict[str, Any] | None = None, 

301 params: dict[str, str] | None = None, 

302 *, 

303 max_attempts: int | None = None, 

304 ) -> dict[str, Any]: 

305 """ 

306 Make an API call and return the JSON response. 

307 

308 This is a convenience wrapper around make_authenticated_request. 

309 

310 Args: 

311 method: HTTP method (GET, POST, DELETE, etc.) 

312 path: API path (e.g., /api/v1/templates) 

313 region: Target region for the request 

314 body: Request body (will be JSON encoded) 

315 params: Query parameters 

316 max_attempts: Maximum attempts for read-only requests. Mutating 

317 requests always make exactly one attempt. Defaults to the 

318 existing retry limit. 

319 

320 Returns: 

321 JSON response as dictionary 

322 

323 Raises: 

324 RuntimeError: If the request fails with a descriptive error message 

325 ValueError: If max_attempts is not a positive integer 

326 """ 

327 _validate_max_attempts(max_attempts) 

328 

329 # Add URL-encoded query parameters to path 

330 if params: 

331 encoded_pairs = [ 

332 f"{quote(str(k), safe='')}={quote(str(v), safe='')}" 

333 for k, v in params.items() 

334 if v is not None 

335 ] 

336 if encoded_pairs: 336 ↛ 339line 336 didn't jump to line 339 because the condition on line 336 was always true

337 path = f"{path}?{'&'.join(encoded_pairs)}" 

338 

339 response = self.make_authenticated_request( 

340 method=method, 

341 path=path, 

342 body=body, 

343 target_region=region, 

344 max_attempts=max_attempts, 

345 ) 

346 

347 if not response.ok: 

348 error_msg = f"{response.status_code} {response.reason}" 

349 try: 

350 error_data = response.json() 

351 if "error" in error_data: 

352 error_msg = error_data["error"] 

353 elif "message" in error_data: 

354 error_msg = error_data["message"] 

355 elif "detail" in error_data: 

356 error_msg = error_data["detail"] 

357 except json.JSONDecodeError, KeyError: 

358 error_msg = response.text or error_msg 

359 raise RuntimeError(f"API request failed: {error_msg}") 

360 

361 result: dict[str, Any] = response.json() 

362 return result 

363 

364 def make_authenticated_request( 

365 self, 

366 method: str, 

367 path: str, 

368 body: dict[str, Any] | None = None, 

369 headers: dict[str, str] | None = None, 

370 target_region: str | None = None, 

371 stream: bool = False, 

372 *, 

373 max_attempts: int | None = None, 

374 ) -> requests.Response: 

375 """ 

376 Make an authenticated request to the GCO API. 

377 

378 Requests with ``target_region`` always use that region's API Gateway so 

379 exact region pinning is enforced without sending routing headers through 

380 the global endpoint. Unpinned requests use the global API unless regional 

381 mode is enabled, in which case they use ``config.default_region``. Global 

382 aggregation paths are unavailable in regional mode. Missing regional 

383 endpoints fail closed instead of silently using the global API. 

384 

385 Args: 

386 method: HTTP method (GET, POST, etc.) 

387 path: API path (e.g., /api/v1/manifests) 

388 body: Request body (will be JSON encoded) 

389 headers: Additional headers 

390 target_region: Exact region for the request. When set, the request 

391 uses that region's API Gateway directly. 

392 stream: Leave the response body unbuffered for incremental consumption. 

393 max_attempts: Maximum attempts for read-only requests. Mutating 

394 requests always make exactly one attempt. Defaults to the 

395 existing retry limit. 

396 

397 Returns: 

398 requests.Response object 

399 

400 Raises: 

401 ValueError: If max_attempts is not a positive integer 

402 """ 

403 _validate_max_attempts(max_attempts) 

404 

405 # Global aggregation endpoints exist only on the global API. Regional 

406 # mode must reject them clearly rather than send a global path to a 

407 # regional bridge and surface an opaque 404. 

408 if self._use_regional_api and ( 

409 path == "/api/v1/global" or path.startswith("/api/v1/global/") 

410 ): 

411 raise ValueError("Global API operations are unavailable in regional API mode") 

412 

413 # Strict regional mode has no global fallback. Resolve an omitted 

414 # optional ``--region`` to the configured default, then require a real 

415 # non-blank Region before attempting endpoint discovery. Keep this as a 

416 # separate branch so ``get_api_endpoint`` is unreachable in strict mode. 

417 if self._use_regional_api: 

418 effective_region = ( 

419 target_region if target_region is not None else self.config.default_region 

420 ) 

421 if not isinstance(effective_region, str) or not effective_region.strip(): 

422 raise ValueError( 

423 "Regional API mode requires a non-empty target or default AWS region" 

424 ) 

425 target_region = effective_region.strip() 

426 endpoint = self.get_regional_api_endpoint(target_region) 

427 elif target_region: 

428 # Exact region pinning always uses the regional API. The global 

429 # proxy is intentionally not VPC-attached and rejects 

430 # X-GCO-Target-Region, so it cannot honor a pin without pretending 

431 # success or weakening isolation. 

432 endpoint = self.get_regional_api_endpoint(target_region) 

433 else: 

434 endpoint = self.get_api_endpoint() 

435 

436 if endpoint is None: 

437 # Only regional discovery returns None; the global endpoint helper 

438 # either returns an endpoint or raises its own actionable error. 

439 assert target_region is not None 

440 raise RuntimeError( 

441 f"Regional API endpoint is not deployed in {target_region}; " 

442 "exact region routing requires the regional API bridge" 

443 ) 

444 

445 url = f"{endpoint.url}{path}" 

446 

447 # Normalize the method once. Only read-only operations are eligible 

448 # for automatic replay; retrying POST/PUT/PATCH/DELETE can duplicate a 

449 # model invocation or state transition after an ambiguous response. 

450 method = method.upper() 

451 retryable_method = method in {"GET", "HEAD", "OPTIONS"} 

452 

453 # Prepare headers without mutating the caller's mapping. 

454 request_headers = dict(headers or {}) 

455 request_headers["Content-Type"] = "application/json" 

456 

457 # Prepare body 

458 body_str = json.dumps(body) if body is not None else "" 

459 

460 # Create AWS request for signing 

461 aws_request = AWSRequest(method=method, url=url, headers=request_headers, data=body_str) 

462 

463 # Sign the request with the endpoint's region 

464 credentials = self._session.get_credentials() 

465 if credentials is None: 

466 raise RuntimeError( 

467 "No AWS credentials found. Configure credentials via environment variables, " 

468 "~/.aws/credentials, IAM role, or SSO (aws sso login)." 

469 ) 

470 SigV4Auth(credentials, "execute-api", endpoint.region).add_auth(aws_request) 

471 

472 # Read-only requests retry transient failures and may refresh expired 

473 # SigV4 credentials once. Mutating requests receive exactly one network 

474 # attempt and return its response unchanged. 

475 last_response = None 

476 retried_auth = False 

477 attempt_limit = ( 

478 (max_attempts if max_attempts is not None else _MAX_RETRIES) if retryable_method else 1 

479 ) 

480 for attempt in range(attempt_limit): 

481 response = requests.request( 

482 method=method, 

483 url=url, 

484 headers=dict(aws_request.headers), 

485 data=body_str, 

486 timeout=(10, 310) if stream else 30, 

487 stream=stream, 

488 ) 

489 last_response = response 

490 

491 # A read-only 403 may mean an expired SigV4 signature. Refresh and 

492 # retry once; mutating requests are never replayed automatically. 

493 if ( 

494 response.status_code == 403 

495 and retryable_method 

496 and not retried_auth 

497 and attempt < attempt_limit - 1 

498 ): 

499 retried_auth = True 

500 logger.warning( 

501 "Request to %s returned 403, refreshing credentials and retrying", 

502 path, 

503 ) 

504 # Force a new session to pick up refreshed credentials 

505 self._session = boto3.Session() 

506 aws_request = AWSRequest( 

507 method=method, url=url, headers=request_headers, data=body_str 

508 ) 

509 credentials = self._session.get_credentials() 

510 if credentials is None: 510 ↛ 511line 510 didn't jump to line 511 because the condition on line 510 was never true

511 return response # No credentials available, return the 403 

512 SigV4Auth(credentials, "execute-api", endpoint.region).add_auth(aws_request) 

513 response.close() 

514 continue 

515 

516 if response.status_code not in _RETRYABLE_STATUS_CODES or not retryable_method: 

517 return response 

518 

519 # Retryable read-only error — back off and retry. 

520 if attempt < attempt_limit - 1: 

521 wait_time = _RETRY_BACKOFF_BASE * (2**attempt) 

522 logger.warning( 

523 "Request to %s returned %d, retrying in %.1fs (attempt %d/%d)", 

524 path, 

525 response.status_code, 

526 wait_time, 

527 attempt + 1, 

528 attempt_limit, 

529 ) 

530 response.close() 

531 time.sleep(wait_time) 

532 

533 # Re-sign the request for the retry (credentials/time may have changed) 

534 aws_request = AWSRequest( 

535 method=method, url=url, headers=request_headers, data=body_str 

536 ) 

537 credentials = self._session.get_credentials() 

538 if credentials is None: 538 ↛ 539line 538 didn't jump to line 539 because the condition on line 538 was never true

539 return last_response 

540 SigV4Auth(credentials, "execute-api", endpoint.region).add_auth(aws_request) 

541 

542 # All retries exhausted — return the last response 

543 return last_response # type: ignore[return-value] 

544 

545 def submit_manifests( 

546 self, 

547 manifests: list[dict[str, Any]], 

548 namespace: str | None = None, 

549 target_region: str | None = None, 

550 dry_run: bool = False, 

551 ) -> dict[str, Any]: 

552 """ 

553 Submit manifests to the GCO API. 

554 

555 Args: 

556 manifests: List of Kubernetes manifest dictionaries 

557 namespace: Default namespace for manifests 

558 target_region: Target region for job execution 

559 dry_run: If True, validate without applying 

560 

561 Returns: 

562 API response dictionary 

563 

564 Raises: 

565 RuntimeError: If submission fails with descriptive error message 

566 """ 

567 body = {"manifests": manifests, "dry_run": dry_run} 

568 

569 if namespace: 569 ↛ 572line 569 didn't jump to line 572 because the condition on line 569 was always true

570 body["namespace"] = namespace 

571 

572 response = self.make_authenticated_request( 

573 method="POST", path="/api/v1/manifests", body=body, target_region=target_region 

574 ) 

575 

576 # Parse response and provide descriptive error messages 

577 if not response.ok: 

578 error_msg = f"{response.status_code} {response.reason}" 

579 try: 

580 error_data = response.json() 

581 # Extract meaningful error details from the response 

582 if "resources" in error_data: 

583 failed = [r for r in error_data["resources"] if r.get("status") == "failed"] 

584 if failed: 584 ↛ 595line 584 didn't jump to line 595 because the condition on line 584 was always true

585 messages = [ 

586 f"{r.get('name')}: {r.get('message', 'Unknown error')}" for r in failed 

587 ] 

588 error_msg = "; ".join(messages) 

589 elif "error" in error_data: 589 ↛ 590line 589 didn't jump to line 590 because the condition on line 589 was never true

590 error_msg = error_data["error"] 

591 elif "message" in error_data: 591 ↛ 595line 591 didn't jump to line 595 because the condition on line 591 was always true

592 error_msg = error_data["message"] 

593 except json.JSONDecodeError, KeyError: 

594 error_msg = response.text or error_msg 

595 raise RuntimeError(error_msg) 

596 

597 result: dict[str, Any] = response.json() 

598 return result 

599 

600 def get_jobs( 

601 self, 

602 region: str | None = None, 

603 namespace: str | None = None, 

604 status: str | None = None, 

605 ) -> list[dict[str, Any]]: 

606 """ 

607 Get jobs from GCO clusters. 

608 

609 Args: 

610 region: Specific region to query (None for all regions) 

611 namespace: Filter by namespace 

612 status: Filter by status (running, completed, failed) 

613 

614 Returns: 

615 List of job information dictionaries 

616 """ 

617 params = [] 

618 if namespace: 618 ↛ 620line 618 didn't jump to line 620 because the condition on line 618 was always true

619 params.append(f"namespace={namespace}") 

620 if status: 

621 params.append(f"status={status}") 

622 

623 query_string = f"?{'&'.join(params)}" if params else "" 

624 

625 response = self.make_authenticated_request( 

626 method="GET", path=f"/api/v1/jobs{query_string}", target_region=region 

627 ) 

628 

629 response.raise_for_status() 

630 result: list[dict[str, Any]] = response.json() 

631 return result 

632 

633 def get_job_details( 

634 self, job_name: str, namespace: str, region: str | None = None 

635 ) -> dict[str, Any]: 

636 """ 

637 Get detailed information about a specific job. 

638 

639 Args: 

640 job_name: Name of the job 

641 namespace: Namespace of the job 

642 region: Region where the job is running 

643 

644 Returns: 

645 Job details dictionary 

646 """ 

647 response = self.make_authenticated_request( 

648 method="GET", path=f"/api/v1/jobs/{namespace}/{job_name}", target_region=region 

649 ) 

650 

651 response.raise_for_status() 

652 result: dict[str, Any] = response.json() 

653 return result 

654 

655 def get_job_logs( 

656 self, job_name: str, namespace: str, region: str | None = None, tail_lines: int = 100 

657 ) -> str: 

658 """ 

659 Get logs from a job. 

660 

661 Args: 

662 job_name: Name of the job 

663 namespace: Namespace of the job 

664 region: Region where the job is running 

665 tail_lines: Number of lines to return from the end 

666 

667 Returns: 

668 Log content as string 

669 """ 

670 response = self.make_authenticated_request( 

671 method="GET", 

672 path=f"/api/v1/jobs/{namespace}/{job_name}/logs?tail={tail_lines}", 

673 target_region=region, 

674 ) 

675 

676 if not response.ok: 

677 # Try to extract a useful error message from the response body 

678 try: 

679 error_data = response.json() 

680 detail = error_data.get("detail", response.reason) 

681 except Exception: 

682 detail = response.text or response.reason 

683 raise RuntimeError(detail) 

684 

685 return str(response.json().get("logs", "")) 

686 

687 def delete_job( 

688 self, 

689 job_name: str, 

690 namespace: str, 

691 region: str | None = None, 

692 expected_uid: str | None = None, 

693 ) -> dict[str, Any]: 

694 """ 

695 Delete a job. 

696 

697 Args: 

698 job_name: Name of the job 

699 namespace: Namespace of the job 

700 region: Region where the job is running 

701 

702 Returns: 

703 Deletion result dictionary 

704 """ 

705 path = f"/api/v1/jobs/{quote(namespace, safe='')}/{quote(job_name, safe='')}" 

706 if expected_uid is not None: 706 ↛ 708line 706 didn't jump to line 708 because the condition on line 706 was always true

707 path += f"?expected_uid={quote(expected_uid, safe='')}" 

708 response = self.make_authenticated_request( 

709 method="DELETE", 

710 path=path, 

711 target_region=region, 

712 ) 

713 

714 response.raise_for_status() 

715 result: dict[str, Any] = response.json() 

716 return result 

717 

718 def get_regional_alb_endpoint(self, region: str) -> str | None: 

719 """ 

720 Get the ALB endpoint for a specific region. 

721 

722 Args: 

723 region: AWS region 

724 

725 Returns: 

726 ALB DNS name or None if not found 

727 """ 

728 stack = self.get_regional_stack(region) 

729 if not stack: 729 ↛ 730line 729 didn't jump to line 730 because the condition on line 729 was never true

730 return None 

731 

732 cfn = self._session.client("cloudformation", region_name=region) 

733 try: 

734 response = cfn.describe_stacks(StackName=stack.stack_name) 

735 stack_data = response["Stacks"][0] 

736 outputs = {o["OutputKey"]: o["OutputValue"] for o in stack_data.get("Outputs", [])} 

737 return outputs.get("AlbDnsName") or outputs.get("LoadBalancerDnsName") 

738 except Exception as e: 

739 logger.debug("Failed to get ALB DNS for %s: %s", region, e) 

740 return None 

741 

742 # ========================================================================= 

743 # Global Aggregation Methods (Cross-Region) 

744 # ========================================================================= 

745 

746 def get_global_jobs( 

747 self, 

748 namespace: str | None = None, 

749 status: str | None = None, 

750 limit: int = 50, 

751 ) -> dict[str, Any]: 

752 """ 

753 Get jobs across all regions via the global aggregation API. 

754 

755 Args: 

756 namespace: Filter by namespace 

757 status: Filter by status 

758 limit: Maximum jobs to return 

759 

760 Returns: 

761 Aggregated job list with region information 

762 """ 

763 params = [f"limit={limit}"] 

764 if namespace: 764 ↛ 766line 764 didn't jump to line 766 because the condition on line 764 was always true

765 params.append(f"namespace={namespace}") 

766 if status: 766 ↛ 769line 766 didn't jump to line 769 because the condition on line 766 was always true

767 params.append(f"status={status}") 

768 

769 query_string = f"?{'&'.join(params)}" 

770 

771 response = self.make_authenticated_request( 

772 method="GET", path=f"/api/v1/global/jobs{query_string}" 

773 ) 

774 

775 response.raise_for_status() 

776 result: dict[str, Any] = response.json() 

777 return result 

778 

779 def get_global_health(self) -> dict[str, Any]: 

780 """ 

781 Get health status across all regions. 

782 

783 Returns: 

784 Aggregated health status from all regional clusters 

785 """ 

786 response = self.make_authenticated_request(method="GET", path="/api/v1/global/health") 

787 

788 response.raise_for_status() 

789 result: dict[str, Any] = response.json() 

790 return result 

791 

792 def get_global_status(self) -> dict[str, Any]: 

793 """ 

794 Get cluster status across all regions. 

795 

796 Returns: 

797 Aggregated status from all regional clusters 

798 """ 

799 response = self.make_authenticated_request(method="GET", path="/api/v1/global/status") 

800 

801 response.raise_for_status() 

802 result: dict[str, Any] = response.json() 

803 return result 

804 

805 def bulk_delete_global( 

806 self, 

807 namespace: str | None = None, 

808 status: str | None = None, 

809 older_than_days: int | None = None, 

810 label_selector: str | None = None, 

811 dry_run: bool = True, 

812 ) -> dict[str, Any]: 

813 """ 

814 Bulk delete jobs across all regions. 

815 

816 Args: 

817 namespace: Filter by namespace 

818 status: Filter by status 

819 older_than_days: Delete jobs older than N days 

820 label_selector: Kubernetes label selector 

821 dry_run: If True, only return what would be deleted 

822 

823 Returns: 

824 Deletion results from all regions 

825 """ 

826 body: dict[str, Any] = {"dry_run": dry_run} 

827 if namespace: 

828 body["namespace"] = namespace 

829 if status: 829 ↛ 831line 829 didn't jump to line 831 because the condition on line 829 was always true

830 body["status"] = status 

831 if older_than_days: 

832 body["older_than_days"] = older_than_days 

833 if label_selector: 

834 body["label_selector"] = label_selector 

835 

836 response = self.make_authenticated_request( 

837 method="DELETE", path="/api/v1/global/jobs", body=body 

838 ) 

839 

840 response.raise_for_status() 

841 result: dict[str, Any] = response.json() 

842 return result 

843 

844 # ========================================================================= 

845 # Regional Job Operations (New API Endpoints) 

846 # ========================================================================= 

847 

848 def get_job_events(self, job_name: str, namespace: str, region: str) -> dict[str, Any]: 

849 """ 

850 Get Kubernetes events for a job. 

851 

852 Args: 

853 job_name: Name of the job 

854 namespace: Namespace of the job 

855 region: Region where the job is running 

856 

857 Returns: 

858 Events related to the job 

859 """ 

860 response = self.make_authenticated_request( 

861 method="GET", 

862 path=f"/api/v1/jobs/{namespace}/{job_name}/events", 

863 target_region=region, 

864 ) 

865 

866 response.raise_for_status() 

867 result: dict[str, Any] = response.json() 

868 return result 

869 

870 def get_job_pods(self, job_name: str, namespace: str, region: str) -> dict[str, Any]: 

871 """ 

872 Get pods for a job. 

873 

874 Args: