Coverage for cli/aws_client.py: 100.00%

1"""

2AWS Client utilities for GCO CLI.

4Provides authenticated access to AWS services with SigV4 signing,

5stack discovery, and region management.

6"""

8import ast

9import json

10import logging

11import os

12import re

13import time

14from dataclasses import dataclass

15from datetime import datetime

16from typing import Any, cast

17from urllib.parse import quote, urlsplit

19import boto3

20import requests

21from botocore.auth import SigV4Auth

22from botocore.awsrequest import AWSRequest

23from botocore.exceptions import ClientError

24from botocore.regions import EndpointResolver

25from botocore.session import get_session as get_botocore_session

27from .config import GCOConfig, get_config

29logger = logging.getLogger(__name__)

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

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

33_MAX_RETRIES = 3

34_RETRY_BACKOFF_BASE = 1.0 # seconds

37class APIRequestError(RuntimeError):

38 """HTTP API failure that preserves status for policy-aware callers."""

40 def __init__(self, status_code: int, message: str):

41 super().__init__(f"API request failed: {message}")

42 self.status_code = status_code

45class RegionalApiDiscoveryError(RuntimeError):

46 """Regional endpoint discovery failed without confirming stack absence."""

49def _cloudformation_stack_missing(exc: ClientError) -> bool:

50 """Return whether CloudFormation authoritatively reports an absent stack."""

51 error = exc.response.get("Error", {})

52 return bool(

53 error.get("Code") == "ValidationError"

54 and "does not exist" in str(error.get("Message", "")).lower()

55 )

58def _aws_credential_context(session: Any) -> str:

59 """Describe configured profile/role hints without claiming provider selection."""

60 hints = []

61 profile_name = getattr(session, "profile_name", None)

62 if isinstance(profile_name, str) and profile_name.strip():

63 hints.append(f"session profile {profile_name.strip()[:128]!r}")

65 role_arn = os.getenv("AWS_ROLE_ARN", "").strip()

66 if role_arn:

67 hints.append(f"AWS_ROLE_ARN is set to {role_arn[:256]!r}")

69 return "; ".join(hints) or "no profile or role hint is available"

72def _safe_aws_error_message(error: dict[str, Any]) -> str:

73 """Return bounded service-provided detail without terminal control bytes."""

74 raw_message = error.get("Message")

75 if not isinstance(raw_message, str):

76 return "AWS did not return an error message"

77 printable = "".join(

78 character if character.isprintable() and character != "\x1b" else " "

79 for character in raw_message

80 )

81 normalized = " ".join(printable.split())

82 return normalized[:512] or "AWS did not return an error message"

85def _execute_api_service_hostname(region: str) -> str:

86 """Resolve the partition-correct execute-api hostname from botocore data.

88 Builds an :class:`EndpointResolver` over the bundled ``endpoints`` data

89 file rather than fetching the session's resolver through

90 ``get_component``, which botocore deprecated as an internal interface.

91 """

92 resolver = EndpointResolver(get_botocore_session().get_data("endpoints"))

93 endpoint = resolver.construct_endpoint("execute-api", region)

94 hostname = endpoint.get("hostname") if isinstance(endpoint, dict) else None

95 if not isinstance(hostname, str):

96 raise ValueError("execute-api endpoint metadata is unavailable")

97 return hostname.lower()

100def _normalize_regional_api_endpoint(value: Any, region: str) -> tuple[str, str]:

101 """Validate a regional stack output as this region's execute-api prod URL."""

102 if not isinstance(value, str):

103 raise ValueError("regional endpoint output is not a string")

104 try:

105 parsed = urlsplit(value.strip())

106 port = parsed.port

107 except ValueError as exc:

108 raise ValueError("regional endpoint output is not a valid URL") from exc

109

110 host = (parsed.hostname or "").lower()

111 service_hostname = _execute_api_service_hostname(region)

112 host_suffix = f".{service_hostname}"

113 api_id = host[: -len(host_suffix)] if host.endswith(host_suffix) else ""

114 if (

115 parsed.scheme != "https"

116 or parsed.username is not None

117 or parsed.password is not None

118 or port not in {None, 443}

119 or re.fullmatch(r"[a-z0-9]+", api_id) is None

120 or parsed.path.rstrip("/") != "/prod"

121 or parsed.query

122 or parsed.fragment

123 ):

124 raise ValueError("regional endpoint output is not an execute-api prod URL")

125 return f"https://{host}/prod", api_id

126

127

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

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

130 if max_attempts is not None and (

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

132 ):

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

134

135

136def _decode_log_payload(payload: Any) -> str:

137 """Return log text, decoding bytes and exact Python bytes-literal envelopes."""

138 if isinstance(payload, str):

139 if len(payload) >= 3 and payload[0] == "b" and payload[1] in {"'", '"'}:

140 try:

141 return cast(bytes, ast.literal_eval(payload)).decode("utf-8", errors="replace")

142 except SyntaxError, ValueError:

143 return payload

144 return payload

145 if isinstance(payload, (bytes, bytearray)):

146 return bytes(payload).decode("utf-8", errors="replace")

147 raise TypeError(f"API returned an unsupported log payload: {type(payload)!r}")

148

149

150@dataclass

151class RegionalStack:

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

153

154 region: str

155 stack_name: str

156 cluster_name: str

157 status: str

158 api_endpoint: str | None = None

159 efs_file_system_id: str | None = None

160 fsx_file_system_id: str | None = None

161 created_time: datetime | None = None

162

163

164@dataclass

165class ApiEndpoint:

166 """API Gateway endpoint information."""

167

168 url: str

169 region: str

170 api_id: str

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

172

173

174class GCOAWSClient:

175 """

176 AWS client for GCO operations.

177

178 Handles:

179 - Stack discovery across regions

180 - Authenticated API requests with SigV4

181 - CloudFormation stack queries

182 - EKS cluster information

183 """

184

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

186 self.config = config or get_config()

187 self._session = boto3.Session()

188 self._api_endpoint_cache: ApiEndpoint | None = None

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

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

191 self._cache_timestamp: float | None = None

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

193

194 def _is_cache_valid(self) -> bool:

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

196 if self._cache_timestamp is None:

197 return False

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

199

200 def _invalidate_cache(self) -> None:

201 """Invalidate all caches."""

202 self._api_endpoint_cache = None

203 self._regional_api_cache = {}

204 self._regional_stacks_cache = None

205 self._cache_timestamp = None

206

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

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

209

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

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

212 public access is disabled.

213

214 Args:

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

216 """

217 self._use_regional_api = use_regional

218

219 def get_regional_api_endpoint(

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

221 ) -> ApiEndpoint | None:

222 """Get the regional API Gateway endpoint for a specific region.

223

224 ``None`` is reserved for CloudFormation's authoritative confirmation

225 that the regional API stack does not exist. Credential, authorization,

226 transport, and malformed-response failures raise instead, so callers

227 never misreport an observation failure as missing infrastructure.

228

229 Args:

230 region: AWS region

231 force_refresh: Force refresh from CloudFormation

232

233 Returns:

234 ApiEndpoint with URL and metadata, or None when CloudFormation

235 confirms the regional API stack is absent

236

237 Raises:

238 RegionalApiDiscoveryError: If discovery cannot run or the existing

239 stack does not publish a usable endpoint

240 """

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

242 return self._regional_api_cache[region]

243

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

245 credential_context = _aws_credential_context(self._session)

246

247 try:

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

249 response = cfn.describe_stacks(StackName=stack_name)

250 except ClientError as exc:

251 if _cloudformation_stack_missing(exc):

252 return None

253 error = exc.response.get("Error", {})

254 raw_error_code = error.get("Code")

255 error_code = (

256 raw_error_code

257 if isinstance(raw_error_code, str)

258 and re.fullmatch(r"[A-Za-z0-9_.-]{1,128}", raw_error_code)

259 else "ClientError"

260 )

261 error_message = _safe_aws_error_message(error)

262 raise RegionalApiDiscoveryError(

263 f"Regional API endpoint discovery could not run for stack '{stack_name}' "

264 f"in {region}. Credential context: {credential_context}. "

265 f"AWS error {error_code}: {error_message}"

266 ) from exc

267 except Exception as exc:

268 failure_type = re.sub(r"[^A-Za-z0-9_.-]", "", type(exc).__name__)[:128]

269 raise RegionalApiDiscoveryError(

270 f"Regional API endpoint discovery could not run for stack '{stack_name}' "

271 f"in {region}. Credential context: {credential_context}. Discovery failed "

272 f"with {failure_type or 'UnexpectedError'}; check AWS credential and network "

273 "configuration"

274 ) from exc

275

276 if not isinstance(response, dict):

277 raise RegionalApiDiscoveryError(

278 f"Regional API endpoint discovery returned an invalid CloudFormation "

279 f"response for stack '{stack_name}' in {region}"

280 )

281 stacks = response.get("Stacks")

282 if not isinstance(stacks, list) or len(stacks) != 1 or not isinstance(stacks[0], dict):

283 raise RegionalApiDiscoveryError(

284 f"Regional API endpoint discovery returned no unique usable stack record for "

285 f"'{stack_name}' in {region}"

286 )

287 stack = stacks[0]

288

289 outputs = stack.get("Outputs", [])

290 if not isinstance(outputs, list):

291 raise RegionalApiDiscoveryError(

292 f"Regional API endpoint discovery returned malformed CloudFormation Outputs "

293 f"for stack '{stack_name}' in {region}"

294 )

295

296 endpoint_output: Any = None

297 endpoint_output_found = False

298 for output in outputs:

299 if not isinstance(output, dict) or not isinstance(output.get("OutputKey"), str):

300 raise RegionalApiDiscoveryError(

301 f"Regional API endpoint discovery returned malformed CloudFormation "

302 f"Outputs for stack '{stack_name}' in {region}"

303 )

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

305 endpoint_output_found = True

306 endpoint_output = output.get("OutputValue")

307 break

308

309 if not endpoint_output_found:

310 raise RegionalApiDiscoveryError(

311 f"Regional API stack '{stack_name}' exists in {region} but does not publish "

312 "a RegionalApiEndpoint output; the bridge deployment may be incomplete"

313 )

314

315 try:

316 api_url, api_id = _normalize_regional_api_endpoint(endpoint_output, region)

317 except Exception as exc:

318 raise RegionalApiDiscoveryError(

319 f"Regional API stack '{stack_name}' in {region} publishes an invalid "

320 "RegionalApiEndpoint output"

321 ) from exc

322

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

324 self._regional_api_cache[region] = endpoint

325 return endpoint

326

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

328 """

329 Get the global API Gateway endpoint.

330

331 Args:

332 force_refresh: Force refresh from CloudFormation

333

334 Returns:

335 ApiEndpoint with URL and metadata

336 """

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

338 return self._api_endpoint_cache

339

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

341

342 try:

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

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

345

346 api_url = None

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

348 if output["OutputKey"] == "ApiEndpoint":

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

350 break

351

352 if not api_url:

353 raise ValueError(

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

355 )

356

357 # Extract API ID from URL

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

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

360

361 self._api_endpoint_cache = ApiEndpoint(

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

363 )

364 self._cache_timestamp = time.time()

365

366 return self._api_endpoint_cache

367

368 except Exception as e:

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

370

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

372 """

373 Discover all regional GCO stacks.

374

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

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

377

378 Args:

379 force_refresh: Force refresh from CloudFormation

380

381 Returns:

382 Dictionary mapping region to RegionalStack

383 """

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

385 return self._regional_stacks_cache

386

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

388

389 # Try configured regions first (fast path)

390 configured_regions = self._get_configured_regions()

391 if configured_regions:

392 for region in configured_regions:

393 stack = self._probe_regional_stack(region)

394 if stack:

395 regional_stacks[region] = stack

396

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

398 if not regional_stacks:

399 # Fall back to scanning all regions

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

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

402 regions_response = ec2.describe_regions()

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

404

405 for region in all_regions:

406 if region in configured_regions:

407 continue # Already checked

408 stack = self._probe_regional_stack(region)

409 if stack:

410 regional_stacks[region] = stack

411

412 self._regional_stacks_cache = regional_stacks

413 self._cache_timestamp = time.time()

414

415 return regional_stacks

416

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

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

419 from .config import _load_cdk_json

420

421 cdk_regions = _load_cdk_json()

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

423 return regions

424

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

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

427

428 Args:

429 region: AWS region to check

430

431 Returns:

432 RegionalStack if found, None otherwise

433 """

434 try:

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

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

437

438 try:

439 response = cfn.describe_stacks(StackName=stack_name)

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

441

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

443

444 return RegionalStack(

445 region=region,

446 stack_name=stack_name,

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

448 status=stack["StackStatus"],

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

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

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

452 )

453 except cfn.exceptions.ClientError:

454 return None

455

456 except Exception as e:

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

458 return None

459

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

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

462 stacks = self.discover_regional_stacks()

463 return stacks.get(region)

464

465 def call_api(

466 self,

467 method: str,

468 path: str,

469 region: str | None = None,

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

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

472 *,

473 max_attempts: int | None = None,

474 ) -> dict[str, Any]:

475 """

476 Make an API call and return the JSON response.

477

478 This is a convenience wrapper around make_authenticated_request.

479

480 Args:

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

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

483 region: Target region for the request

484 body: Request body (will be JSON encoded)

485 params: Query parameters

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

487 requests always make exactly one attempt. Defaults to the

488 existing retry limit.

489

490 Returns:

491 JSON response as dictionary

492

493 Raises:

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

495 ValueError: If max_attempts is not a positive integer

496 """

497 _validate_max_attempts(max_attempts)

498

499 # Add URL-encoded query parameters to path

500 if params:

501 encoded_pairs = [

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

503 for k, v in params.items()

504 if v is not None

505 ]

506 if encoded_pairs:

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

508

509 response = self.make_authenticated_request(

510 method=method,

511 path=path,

512 body=body,

513 target_region=region,

514 max_attempts=max_attempts,

515 )

516

517 if not response.ok:

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

519 try:

520 error_data = response.json()

521 if "error" in error_data:

522 error_msg = error_data["error"]

523 elif "message" in error_data:

524 error_msg = error_data["message"]

525 elif "detail" in error_data:

526 error_msg = error_data["detail"]

527 except json.JSONDecodeError, KeyError:

528 error_msg = response.text or error_msg

529 raise APIRequestError(response.status_code, str(error_msg))

530

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

532 return result

533

534 def make_authenticated_request(

535 self,

536 method: str,

537 path: str,

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

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

540 target_region: str | None = None,

541 stream: bool = False,

542 *,

543 max_attempts: int | None = None,

544 ) -> requests.Response:

545 """

546 Make an authenticated request to the GCO API.

547

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

549 exact region pinning is enforced without sending routing headers through

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

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

552 aggregation paths are unavailable in regional mode. Missing regional

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

554

555 Args:

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

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

558 body: Request body (will be JSON encoded)

559 headers: Additional headers

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

561 uses that region's API Gateway directly.

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

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

564 requests always make exactly one attempt. Defaults to the

565 existing retry limit.

566

567 Returns:

568 requests.Response object

569

570 Raises:

571 ValueError: If max_attempts is not a positive integer

572 """

573 _validate_max_attempts(max_attempts)

574

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

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

577 # regional bridge and surface an opaque 404.

578 if self._use_regional_api and (

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

580 ):

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

582

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

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

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

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

587 if self._use_regional_api:

588 effective_region = (

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

590 )

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

592 raise ValueError(

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

594 )

595 target_region = effective_region.strip()

596 endpoint = self.get_regional_api_endpoint(target_region)

597 elif target_region:

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

599 # proxy is intentionally not VPC-attached and rejects

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

601 # success or weakening isolation.

602 endpoint = self.get_regional_api_endpoint(target_region)

603 else:

604 endpoint = self.get_api_endpoint()

605

606 if endpoint is None:

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

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

609 assert target_region is not None

610 raise RuntimeError(

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

612 "exact region routing requires the regional API bridge"

613 )

614

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

616

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

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

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

620 method = method.upper()

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

622

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

624 request_headers = dict(headers or {})

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

626

627 # Prepare body

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

629

630 # Create AWS request for signing

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

632

633 # Sign the request with the endpoint's region

634 credentials = self._session.get_credentials()

635 if credentials is None:

636 raise RuntimeError(

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

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

639 )

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

641

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

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

644 # attempt and return its response unchanged.

645 retried_auth = False

646 attempt_limit = (

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

648 )

649 attempt = 0

650 while True:

651 response = requests.request(

652 method=method,

653 url=url,

654 headers=dict(aws_request.headers),

655 data=body_str,

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

657 stream=stream,

658 )

659

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

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

662 if (

663 response.status_code == 403

664 and retryable_method

665 and not retried_auth

666 and attempt < attempt_limit - 1

667 ):

668 retried_auth = True

669 logger.warning(

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

671 path,

672 )

673 # Force a new session to pick up refreshed credentials

674 self._session = boto3.Session()

675 aws_request = AWSRequest(

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

677 )

678 credentials = self._session.get_credentials()

679 if credentials is None:

680 return response # No credentials available, return the 403

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

682 response.close()

683 attempt += 1

684 continue

685

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

687 return response

688 if attempt == attempt_limit - 1:

689 return response

690

691 # Retryable read-only error — close before backoff, then re-sign.

692 wait_time = _RETRY_BACKOFF_BASE * (2**attempt)

693 logger.warning(

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

695 path,

696 response.status_code,

697 wait_time,

698 attempt + 1,

699 attempt_limit,

700 )

701 response.close()

702 time.sleep(wait_time)

703

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

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

706 credentials = self._session.get_credentials()

707 if credentials is None:

708 return response

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

710 attempt += 1

711

712 def submit_manifests(

713 self,

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

715 namespace: str | None = None,

716 target_region: str | None = None,

717 dry_run: bool = False,

718 ) -> dict[str, Any]:

719 """

720 Submit manifests to the GCO API.

721

722 Args:

723 manifests: List of Kubernetes manifest dictionaries

724 namespace: Default namespace for manifests

725 target_region: Target region for job execution

726 dry_run: If True, validate without applying

727

728 Returns:

729 API response dictionary

730

731 Raises:

732 RuntimeError: If submission fails with descriptive error message

733 """

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

735

736 if namespace:

737 body["namespace"] = namespace

738

739 response = self.make_authenticated_request(

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

741 )

742

743 # Parse response and provide descriptive error messages

744 if not response.ok:

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

746 try:

747 error_data = response.json()

748 # Extract meaningful error details from the response

749 if "resources" in error_data:

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

751 if failed:

752 messages = [

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

754 ]

755 error_msg = "; ".join(messages)

756 elif "error" in error_data:

757 error_msg = error_data["error"]

758 elif "message" in error_data:

759 error_msg = error_data["message"]

760 except json.JSONDecodeError, KeyError:

761 error_msg = response.text or error_msg

762 raise RuntimeError(error_msg)

763

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

765 return result

766

767 def get_jobs(

768 self,

769 region: str | None = None,

770 namespace: str | None = None,

771 status: str | None = None,

772 ) -> dict[str, Any]:

773 """

774 Get jobs from GCO clusters.

775

776 Args:

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

778 namespace: Filter by namespace

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

780

781 Returns:

782 The ``GET /api/v1/jobs`` envelope: the Job objects under

783 ``"jobs"`` plus the page metadata (``total``, ``count``,

784 ``limit``, ``offset``, ``has_more``) and the answering

785 cluster's ``cluster_id`` / ``region``.

786 """

787 params = []

788 if namespace:

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

790 if status:

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

792

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

794

795 response = self.make_authenticated_request(

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

797 )

798

799 response.raise_for_status()

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

801 return result

802

803 def get_job_details(

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

805 ) -> dict[str, Any]:

806 """

807 Get detailed information about a specific job.

808

809 Args:

810 job_name: Name of the job

811 namespace: Namespace of the job

812 region: Region where the job is running

813

814 Returns:

815 Job details dictionary

816 """

817 response = self.make_authenticated_request(

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

819 )

820

821 response.raise_for_status()

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

823 return result

824

825 def get_job_logs(

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

827 ) -> str:

828 """

829 Get logs from a job.

830

831 Args:

832 job_name: Name of the job

833 namespace: Namespace of the job

834 region: Region where the job is running

835 tail_lines: Number of lines to return from the end

836

837 Returns:

838 Log content as string

839 """

840 response = self.make_authenticated_request(

841 method="GET",

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

843 target_region=region,

844 )

845

846 if not response.ok:

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

848 try:

849 error_data = response.json()

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

851 except Exception:

852 detail = response.text or response.reason

853 raise RuntimeError(detail)

854

855 return _decode_log_payload(response.json().get("logs", ""))

856

857 def delete_job(

858 self,

859 job_name: str,

860 namespace: str,

861 region: str | None = None,

862 expected_uid: str | None = None,

863 ) -> dict[str, Any]:

864 """

865 Delete a job.

866

867 Args:

868 job_name: Name of the job

869 namespace: Namespace of the job

870 region: Region where the job is running

871

872 Returns:

873 Deletion result dictionary

874 """

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

876 if expected_uid is not None:

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

878 response = self.make_authenticated_request(

879 method="DELETE",

880 path=path,

881 target_region=region,

882 )

883

884 response.raise_for_status()

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

886 return result

887

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

889 """

890 Get the ALB endpoint for a specific region.

891

892 Args:

893 region: AWS region

894

895 Returns:

896 ALB DNS name or None if not found

897 """

898 stack = self.get_regional_stack(region)

899 if not stack:

900 return None

901

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

903 try:

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

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

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

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

908 except Exception as e:

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

910 return None

911

912 # =========================================================================

913 # Global Aggregation Methods (Cross-Region)

914 # =========================================================================

915

916 def get_global_jobs(

917 self,

918 namespace: str | None = None,

919 status: str | None = None,

920 limit: int = 50,

921 ) -> dict[str, Any]:

922 """

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

924

925 Args:

926 namespace: Filter by namespace

927 status: Filter by status

928 limit: Maximum jobs to return

929

930 Returns:

931 Aggregated job list with region information

932 """

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

934 if namespace:

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

936 if status:

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

938

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

940

941 response = self.make_authenticated_request(

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

943 )

944

945 response.raise_for_status()

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

947 return result

948

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

950 """

951 Get health status across all regions.

952

953 Returns:

954 Aggregated health status from all regional clusters

955 """

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

957

958 response.raise_for_status()

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

960 return result

961

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

963 """

964 Get cluster status across all regions.

965

966 Returns:

967 Aggregated status from all regional clusters

968 """

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

970

971 response.raise_for_status()

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

973 return result

974

975 def bulk_delete_global(

976 self,

977 namespace: str | None = None,

978 status: str | None = None,

979 older_than_days: int | None = None,

980 label_selector: str | None = None,

981 dry_run: bool = True,

982 ) -> dict[str, Any]:

983 """

984 Bulk delete jobs across all regions.

985

986 Args:

987 namespace: Filter by namespace

988 status: Filter by status

989 older_than_days: Delete jobs older than N days

990 label_selector: Kubernetes label selector

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

992

993 Returns:

994 Deletion results from all regions

995 """

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

997 if namespace:

998 body["namespace"] = namespace

999 if status:

1000 body["status"] = status

1001 if older_than_days:

1002 body["older_than_days"] = older_than_days

1003 if label_selector:

1004 body["label_selector"] = label_selector

1005

1006 response = self.make_authenticated_request(

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

1008 )

1009

1010 response.raise_for_status()

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

1012 return result

1013

1014 # =========================================================================

1015 # Regional Job Operations (New API Endpoints)

1016 # =========================================================================

1017

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

1019 """

1020 Get Kubernetes events for a job.

1021

1022 Args:

1023 job_name: Name of the job

1024 namespace: Namespace of the job

1025 region: Region where the job is running

1026

1027 Returns:

1028 Events related to the job

1029 """

1030 response = self.make_authenticated_request(

1031 method="GET",

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

1033 target_region=region,

1034 )

1035

1036 response.raise_for_status()

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

1038 return result

1039

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

1041 """

1042 Get pods for a job.

1043

1044 Args:

1045 job_name: Name of the job

1046 namespace: Namespace of the job

1047 region: Region where the job is running

1048

1049 Returns:

1050 Pod details for the job

1051 """

1052 response = self.make_authenticated_request(

1053 method="GET",

1054 path=f"/api/v1/jobs/{namespace}/{job_name}/pods",

1055 target_region=region,

1056 )

1057

1058 response.raise_for_status()

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

1060 return result

1061

1062 def get_pod_logs(

1063 self,

1064 job_name: str,

1065 pod_name: str,

1066 namespace: str,

1067 region: str,

1068 tail_lines: int = 100,

1069 container: str | None = None,

1070 ) -> dict[str, Any]:

1071 """

1072 Get logs from a specific pod of a job.

1073

1074 Args:

1075 job_name: Name of the job

1076 pod_name: Name of the pod

1077 namespace: Namespace of the job

1078 region: Region where the job is running

1079 tail_lines: Number of lines to return from the end

1080 container: Container name (for multi-container pods)

1081

1082 Returns:

1083 Pod logs response

1084 """

1085 params = [f"tail={tail_lines}"]

1086 if container:

1087 params.append(f"container={container}")

1088

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

1090

1091 response = self.make_authenticated_request(

1092 method="GET",

1093 path=f"/api/v1/jobs/{namespace}/{job_name}/pods/{pod_name}/logs{query_string}",

1094 target_region=region,

1095 )

1096

1097 response.raise_for_status()

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

1099 if "logs" in result:

1100 result["logs"] = _decode_log_payload(result["logs"])

1101 return result

1102

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

1104 """

1105 Get resource metrics for a job.

1106

1107 Args:

1108 job_name: Name of the job

1109 namespace: Namespace of the job

1110 region: Region where the job is running

1111

1112 Returns:

1113 Resource usage metrics for the job's pods

1114 """

1115 response = self.make_authenticated_request(

1116 method="GET",

1117 path=f"/api/v1/jobs/{namespace}/{job_name}/metrics",

1118 target_region=region,

1119 )

1120

1121 response.raise_for_status()

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

1123 return result

1124

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

1126 """

1127 Retry a failed job.

1128

1129 Creates a new job from the failed job's spec with a new name.

1130

1131 Args:

1132 job_name: Name of the failed job

1133 namespace: Namespace of the job

1134 region: Region where the job is running

1135

1136 Returns:

1137 Result with new job name

1138 """

1139 response = self.make_authenticated_request(

1140 method="POST",

1141 path=f"/api/v1/jobs/{namespace}/{job_name}/retry",

1142 target_region=region,

1143 )

1144

1145 response.raise_for_status()

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

1147 return result

1148

1149 def bulk_delete_jobs(

1150 self,

1151 namespace: str | None = None,

1152 status: str | None = None,

1153 older_than_days: int | None = None,

1154 label_selector: str | None = None,

1155 region: str | None = None,

1156 dry_run: bool = True,

1157 ) -> dict[str, Any]:

1158 """

1159 Bulk delete jobs in a region.

1160

1161 Args:

1162 namespace: Filter by namespace

1163 status: Filter by status

1164 older_than_days: Delete jobs older than N days

1165 label_selector: Kubernetes label selector

1166 region: Target region

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

1168

1169 Returns:

1170 Deletion results

1171 """

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

1173 if namespace:

1174 body["namespace"] = namespace

1175 if status:

1176 body["status"] = status

1177 if older_than_days:

1178 body["older_than_days"] = older_than_days

1179 if label_selector:

1180 body["label_selector"] = label_selector

1181

1182 response = self.make_authenticated_request(

1183 method="DELETE", path="/api/v1/jobs", body=body, target_region=region

1184 )

1185

1186 response.raise_for_status()

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

1188 return result

1189

1190 def get_health(self, region: str) -> dict[str, Any]:

1191 """

1192 Get health status for a specific region.

1193

1194 Args:

1195 region: Target region

1196

1197 Returns:

1198 Health status for the regional cluster

1199 """

1200 response = self.make_authenticated_request(

1201 method="GET", path="/api/v1/health", target_region=region

1202 )

1203

1204 response.raise_for_status()

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

1206 return result

1207

1208 def get_job_validation_policy(self, region: str) -> dict[str, Any]:

1209 """

1210 Get the job validation policy a region actually enforces.

1211

1212 Reads the deployed manifest processor's live configuration, not a

1213 local ``cdk.json`` — the two can diverge whenever a stack was deployed

1214 from a different checkout, and CDK augments ``trusted_registries``

1215 with the project's own ECR hostnames at synth time.

1216

1217 Args:

1218 region: Target region

1219

1220 Returns:

1221 The region's effective policy plus its live namespace

1222 ResourceQuota / LimitRange ceilings

1223 """

1224 response = self.make_authenticated_request(

1225 method="GET", path="/api/v1/policy", target_region=region

1226 )

1227

1228 response.raise_for_status()

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

1230 return result

1231

1232

1233def get_aws_client(config: GCOConfig | None = None) -> GCOAWSClient:

1234 """Get a configured AWS client instance."""

1235 return GCOAWSClient(config)

Coverage for cli / aws_client.py: 100.00%

465 statements