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.session import get_session as get_botocore_session

26from .config import GCOConfig, get_config

28logger = logging.getLogger(__name__)

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

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

32_MAX_RETRIES = 3

33_RETRY_BACKOFF_BASE = 1.0 # seconds

36class APIRequestError(RuntimeError):

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

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

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

41 self.status_code = status_code

44class RegionalApiDiscoveryError(RuntimeError):

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

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

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

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

51 return bool(

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

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

54 )

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

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

59 hints = []

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

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

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

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

65 if role_arn:

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

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

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

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

73 raw_message = error.get("Message")

74 if not isinstance(raw_message, str):

75 return "AWS did not return an error message"

76 printable = "".join(

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

78 for character in raw_message

79 )

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

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

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

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

86 resolver = get_botocore_session().get_component("endpoint_resolver")

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

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

89 if not isinstance(hostname, str):

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

91 return hostname.lower()

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

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

96 if not isinstance(value, str):

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

98 try:

99 parsed = urlsplit(value.strip())

100 port = parsed.port

101 except ValueError as exc:

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

103

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

105 service_hostname = _execute_api_service_hostname(region)

106 host_suffix = f".{service_hostname}"

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

108 if (

109 parsed.scheme != "https"

110 or parsed.username is not None

111 or parsed.password is not None

112 or port not in {None, 443}

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

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

115 or parsed.query

116 or parsed.fragment

117 ):

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

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

120

121

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

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

124 if max_attempts is not None and (

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

126 ):

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

128

129

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

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

132 if isinstance(payload, str):

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

134 try:

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

136 except SyntaxError, ValueError:

137 return payload

138 return payload

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

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

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

142

143

144@dataclass

145class RegionalStack:

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

147

148 region: str

149 stack_name: str

150 cluster_name: str

151 status: str

152 api_endpoint: str | None = None

153 efs_file_system_id: str | None = None

154 fsx_file_system_id: str | None = None

155 created_time: datetime | None = None

156

157

158@dataclass

159class ApiEndpoint:

160 """API Gateway endpoint information."""

161

162 url: str

163 region: str

164 api_id: str

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

166

167

168class GCOAWSClient:

169 """

170 AWS client for GCO operations.

171

172 Handles:

173 - Stack discovery across regions

174 - Authenticated API requests with SigV4

175 - CloudFormation stack queries

176 - EKS cluster information

177 """

178

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

180 self.config = config or get_config()

181 self._session = boto3.Session()

182 self._api_endpoint_cache: ApiEndpoint | None = None

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

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

185 self._cache_timestamp: float | None = None

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

187

188 def _is_cache_valid(self) -> bool:

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

190 if self._cache_timestamp is None:

191 return False

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

193

194 def _invalidate_cache(self) -> None:

195 """Invalidate all caches."""

196 self._api_endpoint_cache = None

197 self._regional_api_cache = {}

198 self._regional_stacks_cache = None

199 self._cache_timestamp = None

200

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

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

203

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

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

206 public access is disabled.

207

208 Args:

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

210 """

211 self._use_regional_api = use_regional

212

213 def get_regional_api_endpoint(

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

215 ) -> ApiEndpoint | None:

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

217

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

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

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

221 never misreport an observation failure as missing infrastructure.

222

223 Args:

224 region: AWS region

225 force_refresh: Force refresh from CloudFormation

226

227 Returns:

228 ApiEndpoint with URL and metadata, or None when CloudFormation

229 confirms the regional API stack is absent

230

231 Raises:

232 RegionalApiDiscoveryError: If discovery cannot run or the existing

233 stack does not publish a usable endpoint

234 """

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

236 return self._regional_api_cache[region]

237

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

239 credential_context = _aws_credential_context(self._session)

240

241 try:

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

243 response = cfn.describe_stacks(StackName=stack_name)

244 except ClientError as exc:

245 if _cloudformation_stack_missing(exc):

246 return None

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

248 raw_error_code = error.get("Code")

249 error_code = (

250 raw_error_code

251 if isinstance(raw_error_code, str)

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

253 else "ClientError"

254 )

255 error_message = _safe_aws_error_message(error)

256 raise RegionalApiDiscoveryError(

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

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

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

260 ) from exc

261 except Exception as exc:

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

263 raise RegionalApiDiscoveryError(

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

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

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

267 "configuration"

268 ) from exc

269

270 if not isinstance(response, dict):

271 raise RegionalApiDiscoveryError(

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

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

274 )

275 stacks = response.get("Stacks")

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

277 raise RegionalApiDiscoveryError(

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

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

280 )

281 stack = stacks[0]

282

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

284 if not isinstance(outputs, list):

285 raise RegionalApiDiscoveryError(

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

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

288 )

289

290 endpoint_output: Any = None

291 endpoint_output_found = False

292 for output in outputs:

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

294 raise RegionalApiDiscoveryError(

295 f"Regional API endpoint discovery returned malformed CloudFormation "

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

297 )

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

299 endpoint_output_found = True

300 endpoint_output = output.get("OutputValue")

301 break

302

303 if not endpoint_output_found:

304 raise RegionalApiDiscoveryError(

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

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

307 )

308

309 try:

310 api_url, api_id = _normalize_regional_api_endpoint(endpoint_output, region)

311 except Exception as exc:

312 raise RegionalApiDiscoveryError(

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

314 "RegionalApiEndpoint output"

315 ) from exc

316

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

318 self._regional_api_cache[region] = endpoint

319 return endpoint

320

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

322 """

323 Get the global API Gateway endpoint.

324

325 Args:

326 force_refresh: Force refresh from CloudFormation

327

328 Returns:

329 ApiEndpoint with URL and metadata

330 """

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

332 return self._api_endpoint_cache

333

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

335

336 try:

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

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

339

340 api_url = None

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

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

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

344 break

345

346 if not api_url:

347 raise ValueError(

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

349 )

350

351 # Extract API ID from URL

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

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

354

355 self._api_endpoint_cache = ApiEndpoint(

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

357 )

358 self._cache_timestamp = time.time()

359

360 return self._api_endpoint_cache

361

362 except Exception as e:

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

364

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

366 """

367 Discover all regional GCO stacks.

368

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

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

371

372 Args:

373 force_refresh: Force refresh from CloudFormation

374

375 Returns:

376 Dictionary mapping region to RegionalStack

377 """

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

379 return self._regional_stacks_cache

380

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

382

383 # Try configured regions first (fast path)

384 configured_regions = self._get_configured_regions()

385 if configured_regions:

386 for region in configured_regions:

387 stack = self._probe_regional_stack(region)

388 if stack:

389 regional_stacks[region] = stack

390

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

392 if not regional_stacks:

393 # Fall back to scanning all regions

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

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

396 regions_response = ec2.describe_regions()

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

398

399 for region in all_regions:

400 if region in configured_regions:

401 continue # Already checked

402 stack = self._probe_regional_stack(region)

403 if stack:

404 regional_stacks[region] = stack

405

406 self._regional_stacks_cache = regional_stacks

407 self._cache_timestamp = time.time()

408

409 return regional_stacks

410

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

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

413 from .config import _load_cdk_json

414

415 cdk_regions = _load_cdk_json()

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

417 return regions

418

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

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

421

422 Args:

423 region: AWS region to check

424

425 Returns:

426 RegionalStack if found, None otherwise

427 """

428 try:

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

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

431

432 try:

433 response = cfn.describe_stacks(StackName=stack_name)

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

435

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

437

438 return RegionalStack(

439 region=region,

440 stack_name=stack_name,

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

442 status=stack["StackStatus"],

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

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

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

446 )

447 except cfn.exceptions.ClientError:

448 return None

449

450 except Exception as e:

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

452 return None

453

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

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

456 stacks = self.discover_regional_stacks()

457 return stacks.get(region)

458

459 def call_api(

460 self,

461 method: str,

462 path: str,

463 region: str | None = None,

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

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

466 *,

467 max_attempts: int | None = None,

468 ) -> dict[str, Any]:

469 """

470 Make an API call and return the JSON response.

471

472 This is a convenience wrapper around make_authenticated_request.

473

474 Args:

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

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

477 region: Target region for the request

478 body: Request body (will be JSON encoded)

479 params: Query parameters

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

481 requests always make exactly one attempt. Defaults to the

482 existing retry limit.

483

484 Returns:

485 JSON response as dictionary

486

487 Raises:

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

489 ValueError: If max_attempts is not a positive integer

490 """

491 _validate_max_attempts(max_attempts)

492

493 # Add URL-encoded query parameters to path

494 if params:

495 encoded_pairs = [

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

497 for k, v in params.items()

498 if v is not None

499 ]

500 if encoded_pairs:

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

502

503 response = self.make_authenticated_request(

504 method=method,

505 path=path,

506 body=body,

507 target_region=region,

508 max_attempts=max_attempts,

509 )

510

511 if not response.ok:

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

513 try:

514 error_data = response.json()

515 if "error" in error_data:

516 error_msg = error_data["error"]

517 elif "message" in error_data:

518 error_msg = error_data["message"]

519 elif "detail" in error_data:

520 error_msg = error_data["detail"]

521 except json.JSONDecodeError, KeyError:

522 error_msg = response.text or error_msg

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

524

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

526 return result

527

528 def make_authenticated_request(

529 self,

530 method: str,

531 path: str,

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

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

534 target_region: str | None = None,

535 stream: bool = False,

536 *,

537 max_attempts: int | None = None,

538 ) -> requests.Response:

539 """

540 Make an authenticated request to the GCO API.

541

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

543 exact region pinning is enforced without sending routing headers through

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

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

546 aggregation paths are unavailable in regional mode. Missing regional

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

548

549 Args:

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

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

552 body: Request body (will be JSON encoded)

553 headers: Additional headers

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

555 uses that region's API Gateway directly.

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

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

558 requests always make exactly one attempt. Defaults to the

559 existing retry limit.

560

561 Returns:

562 requests.Response object

563

564 Raises:

565 ValueError: If max_attempts is not a positive integer

566 """

567 _validate_max_attempts(max_attempts)

568

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

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

571 # regional bridge and surface an opaque 404.

572 if self._use_regional_api and (

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

574 ):

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

576

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

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

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

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

581 if self._use_regional_api:

582 effective_region = (

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

584 )

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

586 raise ValueError(

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

588 )

589 target_region = effective_region.strip()

590 endpoint = self.get_regional_api_endpoint(target_region)

591 elif target_region:

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

593 # proxy is intentionally not VPC-attached and rejects

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

595 # success or weakening isolation.

596 endpoint = self.get_regional_api_endpoint(target_region)

597 else:

598 endpoint = self.get_api_endpoint()

599

600 if endpoint is None:

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

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

603 assert target_region is not None

604 raise RuntimeError(

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

606 "exact region routing requires the regional API bridge"

607 )

608

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

610

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

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

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

614 method = method.upper()

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

616

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

618 request_headers = dict(headers or {})

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

620

621 # Prepare body

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

623

624 # Create AWS request for signing

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

626

627 # Sign the request with the endpoint's region

628 credentials = self._session.get_credentials()

629 if credentials is None:

630 raise RuntimeError(

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

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

633 )

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

635

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

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

638 # attempt and return its response unchanged.

639 retried_auth = False

640 attempt_limit = (

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

642 )

643 attempt = 0

644 while True:

645 response = requests.request(

646 method=method,

647 url=url,

648 headers=dict(aws_request.headers),

649 data=body_str,

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

651 stream=stream,

652 )

653

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

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

656 if (

657 response.status_code == 403

658 and retryable_method

659 and not retried_auth

660 and attempt < attempt_limit - 1

661 ):

662 retried_auth = True

663 logger.warning(

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

665 path,

666 )

667 # Force a new session to pick up refreshed credentials

668 self._session = boto3.Session()

669 aws_request = AWSRequest(

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

671 )

672 credentials = self._session.get_credentials()

673 if credentials is None:

674 return response # No credentials available, return the 403

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

676 response.close()

677 attempt += 1

678 continue

679

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

681 return response

682 if attempt == attempt_limit - 1:

683 return response

684

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

686 wait_time = _RETRY_BACKOFF_BASE * (2**attempt)

687 logger.warning(

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

689 path,

690 response.status_code,

691 wait_time,

692 attempt + 1,

693 attempt_limit,

694 )

695 response.close()

696 time.sleep(wait_time)

697

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

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

700 credentials = self._session.get_credentials()

701 if credentials is None:

702 return response

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

704 attempt += 1

705

706 def submit_manifests(

707 self,

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

709 namespace: str | None = None,

710 target_region: str | None = None,

711 dry_run: bool = False,

712 ) -> dict[str, Any]:

713 """

714 Submit manifests to the GCO API.

715

716 Args:

717 manifests: List of Kubernetes manifest dictionaries

718 namespace: Default namespace for manifests

719 target_region: Target region for job execution

720 dry_run: If True, validate without applying

721

722 Returns:

723 API response dictionary

724

725 Raises:

726 RuntimeError: If submission fails with descriptive error message

727 """

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

729

730 if namespace:

731 body["namespace"] = namespace

732

733 response = self.make_authenticated_request(

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

735 )

736

737 # Parse response and provide descriptive error messages

738 if not response.ok:

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

740 try:

741 error_data = response.json()

742 # Extract meaningful error details from the response

743 if "resources" in error_data:

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

745 if failed:

746 messages = [

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

748 ]

749 error_msg = "; ".join(messages)

750 elif "error" in error_data:

751 error_msg = error_data["error"]

752 elif "message" in error_data:

753 error_msg = error_data["message"]

754 except json.JSONDecodeError, KeyError:

755 error_msg = response.text or error_msg

756 raise RuntimeError(error_msg)

757

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

759 return result

760

761 def get_jobs(

762 self,

763 region: str | None = None,

764 namespace: str | None = None,

765 status: str | None = None,

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

767 """

768 Get jobs from GCO clusters.

769

770 Args:

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

772 namespace: Filter by namespace

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

774

775 Returns:

776 List of job information dictionaries

777 """

778 params = []

779 if namespace:

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

781 if status:

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

783

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

785

786 response = self.make_authenticated_request(

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

788 )

789

790 response.raise_for_status()

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

792 return result

793

794 def get_job_details(

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

796 ) -> dict[str, Any]:

797 """

798 Get detailed information about a specific job.

799

800 Args:

801 job_name: Name of the job

802 namespace: Namespace of the job

803 region: Region where the job is running

804

805 Returns:

806 Job details dictionary

807 """

808 response = self.make_authenticated_request(

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

810 )

811

812 response.raise_for_status()

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

814 return result

815

816 def get_job_logs(

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

818 ) -> str:

819 """

820 Get logs from a job.

821

822 Args:

823 job_name: Name of the job

824 namespace: Namespace of the job

825 region: Region where the job is running

826 tail_lines: Number of lines to return from the end

827

828 Returns:

829 Log content as string

830 """

831 response = self.make_authenticated_request(

832 method="GET",

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

834 target_region=region,

835 )

836

837 if not response.ok:

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

839 try:

840 error_data = response.json()

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

842 except Exception:

843 detail = response.text or response.reason

844 raise RuntimeError(detail)

845

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

847

848 def delete_job(

849 self,

850 job_name: str,

851 namespace: str,

852 region: str | None = None,

853 expected_uid: str | None = None,

854 ) -> dict[str, Any]:

855 """

856 Delete a job.

857

858 Args:

859 job_name: Name of the job

860 namespace: Namespace of the job

861 region: Region where the job is running

862

863 Returns:

864 Deletion result dictionary

865 """

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

867 if expected_uid is not None:

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

869 response = self.make_authenticated_request(

870 method="DELETE",

871 path=path,

872 target_region=region,

873 )

874

875 response.raise_for_status()

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

877 return result

878

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

880 """

881 Get the ALB endpoint for a specific region.

882

883 Args:

884 region: AWS region

885

886 Returns:

887 ALB DNS name or None if not found

888 """

889 stack = self.get_regional_stack(region)

890 if not stack:

891 return None

892

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

894 try:

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

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

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

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

899 except Exception as e:

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

901 return None

902

903 # =========================================================================

904 # Global Aggregation Methods (Cross-Region)

905 # =========================================================================

906

907 def get_global_jobs(

908 self,

909 namespace: str | None = None,

910 status: str | None = None,

911 limit: int = 50,

912 ) -> dict[str, Any]:

913 """

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

915

916 Args:

917 namespace: Filter by namespace

918 status: Filter by status

919 limit: Maximum jobs to return

920

921 Returns:

922 Aggregated job list with region information

923 """

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

925 if namespace:

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

927 if status:

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

929

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

931

932 response = self.make_authenticated_request(

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

934 )

935

936 response.raise_for_status()

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

938 return result

939

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

941 """

942 Get health status across all regions.

943

944 Returns:

945 Aggregated health status from all regional clusters

946 """

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

948

949 response.raise_for_status()

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

951 return result

952

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

954 """

955 Get cluster status across all regions.

956

957 Returns:

958 Aggregated status from all regional clusters

959 """

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

961

962 response.raise_for_status()

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

964 return result

965

966 def bulk_delete_global(

967 self,

968 namespace: str | None = None,

969 status: str | None = None,

970 older_than_days: int | None = None,

971 label_selector: str | None = None,

972 dry_run: bool = True,

973 ) -> dict[str, Any]:

974 """

975 Bulk delete jobs across all regions.

976

977 Args:

978 namespace: Filter by namespace

979 status: Filter by status

980 older_than_days: Delete jobs older than N days

981 label_selector: Kubernetes label selector

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

983

984 Returns:

985 Deletion results from all regions

986 """

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

988 if namespace:

989 body["namespace"] = namespace

990 if status:

991 body["status"] = status

992 if older_than_days:

993 body["older_than_days"] = older_than_days

994 if label_selector:

995 body["label_selector"] = label_selector

996

997 response = self.make_authenticated_request(

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

999 )

1000

1001 response.raise_for_status()

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

1003 return result

1004

1005 # =========================================================================

1006 # Regional Job Operations (New API Endpoints)

1007 # =========================================================================

1008

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

1010 """

1011 Get Kubernetes events for a job.

1012

1013 Args:

1014 job_name: Name of the job

1015 namespace: Namespace of the job

1016 region: Region where the job is running

1017

1018 Returns:

1019 Events related to the job

1020 """

1021 response = self.make_authenticated_request(

1022 method="GET",

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

1024 target_region=region,

1025 )

1026

1027 response.raise_for_status()

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

1029 return result

1030

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

1032 """

1033 Get pods for a job.

1034

1035 Args:

1036 job_name: Name of the job

1037 namespace: Namespace of the job

1038 region: Region where the job is running

1039

1040 Returns:

1041 Pod details for the job

1042 """

1043 response = self.make_authenticated_request(

1044 method="GET",

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

1046 target_region=region,

1047 )

1048

1049 response.raise_for_status()

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

1051 return result

1052

1053 def get_pod_logs(

1054 self,

1055 job_name: str,

1056 pod_name: str,

1057 namespace: str,

1058 region: str,

1059 tail_lines: int = 100,

1060 container: str | None = None,

1061 ) -> dict[str, Any]:

1062 """

1063 Get logs from a specific pod of a job.

1064

1065 Args:

1066 job_name: Name of the job

1067 pod_name: Name of the pod

1068 namespace: Namespace of the job

1069 region: Region where the job is running

1070 tail_lines: Number of lines to return from the end

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

1072

1073 Returns:

1074 Pod logs response

1075 """

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

1077 if container:

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

1079

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

1081

1082 response = self.make_authenticated_request(

1083 method="GET",

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

1085 target_region=region,

1086 )

1087

1088 response.raise_for_status()

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

1090 if "logs" in result:

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

1092 return result

1093

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

1095 """

1096 Get resource metrics for a job.

1097

1098 Args:

1099 job_name: Name of the job

1100 namespace: Namespace of the job

1101 region: Region where the job is running

1102

1103 Returns:

1104 Resource usage metrics for the job's pods

1105 """

1106 response = self.make_authenticated_request(

1107 method="GET",

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

1109 target_region=region,

1110 )

1111

1112 response.raise_for_status()

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

1114 return result

1115

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

1117 """

1118 Retry a failed job.

1119

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

1121

1122 Args:

1123 job_name: Name of the failed job

1124 namespace: Namespace of the job

1125 region: Region where the job is running

1126

1127 Returns:

1128 Result with new job name

1129 """

1130 response = self.make_authenticated_request(

1131 method="POST",

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

1133 target_region=region,

1134 )

1135

1136 response.raise_for_status()

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

1138 return result

1139

1140 def bulk_delete_jobs(

1141 self,

1142 namespace: str | None = None,

1143 status: str | None = None,

1144 older_than_days: int | None = None,

1145 label_selector: str | None = None,

1146 region: str | None = None,

1147 dry_run: bool = True,

1148 ) -> dict[str, Any]:

1149 """

1150 Bulk delete jobs in a region.

1151

1152 Args:

1153 namespace: Filter by namespace

1154 status: Filter by status

1155 older_than_days: Delete jobs older than N days

1156 label_selector: Kubernetes label selector

1157 region: Target region

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

1159

1160 Returns:

1161 Deletion results

1162 """

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

1164 if namespace:

1165 body["namespace"] = namespace

1166 if status:

1167 body["status"] = status

1168 if older_than_days:

1169 body["older_than_days"] = older_than_days

1170 if label_selector:

1171 body["label_selector"] = label_selector

1172

1173 response = self.make_authenticated_request(

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

1175 )

1176

1177 response.raise_for_status()

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

1179 return result

1180

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

1182 """

1183 Get health status for a specific region.

1184

1185 Args:

1186 region: Target region

1187

1188 Returns:

1189 Health status for the regional cluster

1190 """

1191 response = self.make_authenticated_request(

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

1193 )

1194

1195 response.raise_for_status()

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

1197 return result

1198

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

1200 """

1201 Get the job validation policy a region actually enforces.

1202

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

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

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

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

1207

1208 Args:

1209 region: Target region

1210

1211 Returns:

1212 The region's effective policy plus its live namespace

1213 ResourceQuota / LimitRange ceilings

1214 """

1215 response = self.make_authenticated_request(

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

1217 )

1218

1219 response.raise_for_status()

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

1221 return result

1222

1223

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

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

1226 return GCOAWSClient(config)

Coverage for cli / aws_client.py: 100.00%

464 statements