Coverage for cli/nodepools.py: 100.00%

1"""

2NodePool management utilities for GCO CLI.

4Provides functionality to create and manage EKS Auto Mode NodePools with

5support for On-Demand Capacity Reservations (ODCRs) and Capacity Blocks.

7Key Features:

8- Generate NodeClass + NodePool manifests for ODCR- and Capacity Block-backed

9 capacity, bound through the Auto Mode NodeClass's

10 ``capacityReservationSelectorTerms``

11- Taint reserved GPU and Neuron nodes the way the shipped NodePools do, with

12 each instance type's accelerator class read from EC2 (``DescribeInstanceTypes``)

13 and, when that lookup fails, from the reviewed accelerator catalog the

14 monthly dependency scan keeps current

15- List, describe and delete NodePools in EKS clusters

16- Support for fallback to on-demand when the reservation is exhausted

18See: https://docs.aws.amazon.com/eks/latest/userguide/auto-odcr.html

19"""

21import base64

22import functools

23import json

24import logging

25import os

26import tempfile

27import weakref

28from collections.abc import Mapping, Sequence

29from contextlib import suppress

30from dataclasses import dataclass

31from pathlib import Path

32from typing import Any

34import boto3

35import yaml

36from kubernetes.client import CustomObjectsApi

38import gco

40logger = logging.getLogger(__name__)

42# Default vCPU count when instance type lookup fails (conservative estimate)

43DEFAULT_VCPUS_PER_NODE = 96

45#: Where an instance type's accelerator class came from (``InstanceTypeFacts.source``).

46FACTS_SOURCE_EC2 = "ec2"

47FACTS_SOURCE_CATALOG = "catalog"

48FACTS_SOURCE_UNKNOWN = "unknown"

50#: The reviewed accelerator catalog, shipped inside the ``gco.config`` package

51#: (pyproject's package-data). The monthly dependency scan compares its

52#: families with EC2 and the maintainer refreshes it (docs/MAINTENANCE.md), so

53#: it is the offline answer to "does this family carry an NVIDIA GPU or Neuron

54#: devices" and nothing in this module keeps its own list of families. Located

55#: from the ``gco`` package directory rather than imported through

56#: ``gco.config``, whose ``__init__`` pulls in the CDK (an optional extra).

57_ACCELERATOR_CATALOG_PATH = (

58 Path(gco.__file__).resolve().parent / "config" / "accelerator_catalog.json"

59)

60_ACCELERATOR_CLASSES = frozenset({"nvidia", "neuron"})

63@dataclass(frozen=True)

64class InstanceTypeFacts:

65 """What the NodePool generator needs to know about one EC2 instance type.

67 ``accelerator`` is ``"nvidia"``, ``"neuron"`` or ``None``; ``source`` says

68 whether it was read live from EC2, from the reviewed catalog because the

69 EC2 lookup failed, or could not be determined at all.

70 """

72 instance_type: str

73 vcpus: int

74 accelerator: str | None

75 source: str

77 @property

78 def family(self) -> str:

79 return self.instance_type.partition(".")[0]

81 def describe(self) -> str:

82 """One manifest-comment line: the class, the vCPUs and where they came from."""

83 if self.accelerator == "nvidia":

84 hardware = "NVIDIA GPU"

85 elif self.accelerator == "neuron":

86 hardware = "AWS Neuron"

87 elif self.source == FACTS_SOURCE_UNKNOWN:

88 hardware = "accelerator unknown"

89 else:

90 hardware = "no accelerator"

91 provenance = {

92 FACTS_SOURCE_EC2: "EC2 DescribeInstanceTypes",

93 FACTS_SOURCE_CATALOG: "reviewed catalog; EC2 lookup failed",

94 }.get(

95 self.source,

96 "EC2 lookup failed and the family is not in the reviewed catalog; no accelerator taint",

97 )

98 return f"# {self.instance_type}: {hardware}, {self.vcpus} vCPUs ({provenance})"

100

101def accelerator_from_ec2_record(record: Mapping[str, Any]) -> str | None:

102 """Classify one ``DescribeInstanceTypes`` record the way the accelerator catalog does.

103

104 An NVIDIA GPU (``GpuInfo.Gpus[].Manufacturer``) wins over Neuron devices;

105 AMD GPUs, FPGAs and media accelerators are not accelerators GCO schedules.

106 """

107 gpu_info = record.get("GpuInfo")

108 gpus = gpu_info.get("Gpus") if isinstance(gpu_info, Mapping) else None

109 for gpu in gpus if isinstance(gpus, list) else []:

110 manufacturer = gpu.get("Manufacturer") if isinstance(gpu, Mapping) else None

111 if isinstance(manufacturer, str) and manufacturer.casefold() == "nvidia":

112 return "nvidia"

113 neuron_info = record.get("NeuronInfo")

114 devices = neuron_info.get("NeuronDevices") if isinstance(neuron_info, Mapping) else None

115 if isinstance(devices, list) and devices:

116 return "neuron"

117 return None

118

119

120@functools.cache

121def _catalog_family_accelerators() -> dict[str, str]:

122 """``family -> accelerator`` from the reviewed catalog; empty when it can't be read."""

123 try:

124 parsed = json.loads(_ACCELERATOR_CATALOG_PATH.read_text(encoding="utf-8"))

125 except (OSError, ValueError) as exc:

126 logger.debug("Reviewed accelerator catalog unavailable: %s", exc)

127 return {}

128 families = parsed.get("families") if isinstance(parsed, dict) else None

129 if not isinstance(families, dict):

130 return {}

131 return {

132 str(name): str(policy["accelerator"])

133 for name, policy in families.items()

134 if isinstance(policy, dict) and policy.get("accelerator") in _ACCELERATOR_CLASSES

135 }

136

137

138def accelerator_from_catalog(instance_type: str) -> str | None:

139 """The reviewed catalog's accelerator class for an instance type's family, if any."""

140 return _catalog_family_accelerators().get(instance_type.partition(".")[0])

141

142

143def describe_instance_type(instance_type: str, region: str = "us-east-1") -> InstanceTypeFacts:

144 """Read an instance type's vCPUs and accelerator class from EC2, with fallbacks.

145

146 One ``DescribeInstanceTypes`` call answers both. When it fails (no

147 credentials, an unknown type in this Region, throttling) the vCPU count

148 falls back to :data:`DEFAULT_VCPUS_PER_NODE` and the accelerator class to

149 the reviewed catalog's entry for the family.

150 """

151 try:

152 ec2 = boto3.client("ec2", region_name=region)

153 response = ec2.describe_instance_types(InstanceTypes=[instance_type])

154 records = response.get("InstanceTypes") or []

155 if records:

156 record = records[0]

157 return InstanceTypeFacts(

158 instance_type=instance_type,

159 vcpus=int(record["VCpuInfo"]["DefaultVCpus"]),

160 accelerator=accelerator_from_ec2_record(record),

161 source=FACTS_SOURCE_EC2,

162 )

163 logger.debug("EC2 returned no record for instance type %s in %s", instance_type, region)

164 except Exception as exc:

165 logger.debug("Failed to describe instance type %s in %s: %s", instance_type, region, exc)

166 accelerator = accelerator_from_catalog(instance_type)

167 return InstanceTypeFacts(

168 instance_type=instance_type,

169 vcpus=DEFAULT_VCPUS_PER_NODE,

170 accelerator=accelerator,

171 source=FACTS_SOURCE_CATALOG if accelerator is not None else FACTS_SOURCE_UNKNOWN,

172 )

173

174

175def _unlink_temp_ca_cert(path: str) -> None:

176 """Best-effort cleanup for a CA file owned by a Kubernetes API client."""

177 try:

178 os.unlink(path)

179 except FileNotFoundError:

180 # Already gone (e.g. removed by another cleanup path) — the goal is an

181 # absent file, so this is success, not a failure to log or retry.

182 pass

183 except OSError as exc:

184 logger.debug("Failed to remove temporary Kubernetes CA certificate %s: %s", path, exc)

185

186

187def calculate_cpu_limit(facts: Sequence[InstanceTypeFacts], max_nodes: int) -> int:

188 """The NodePool CPU limit: ``max_nodes`` times the largest instance type's vCPUs.

189

190 Without instance types (any type the reservation offers) the conservative

191 :data:`DEFAULT_VCPUS_PER_NODE` stands in for the largest type.

192 """

193 if not facts:

194 return max_nodes * DEFAULT_VCPUS_PER_NODE

195 return max_nodes * max(fact.vcpus for fact in facts)

196

197

198@dataclass

199class NodePoolInfo:

200 """Information about a Karpenter NodePool."""

201

202 name: str

203 capacity_type: str # "on-demand", "spot", "reserved"

204 instance_types: list[str]

205 max_nodes: int | None

206 status: str

207 node_count: int

208 capacity_reservation_id: str | None = None

209

210

211#: EKS Auto Mode's node class API. GCO clusters run Auto Mode, which serves

212#: ``NodeClass`` objects in this group; self-managed Karpenter's

213#: ``karpenter.k8s.aws`` ``EC2NodeClass`` CRD does not exist on them.

214AUTO_MODE_NODECLASS_GROUP = "eks.amazonaws.com"

215AUTO_MODE_NODECLASS_VERSION = "v1"

216AUTO_MODE_NODECLASS_KIND = "NodeClass"

217AUTO_MODE_NODECLASS_PLURAL = "nodeclasses"

218

219#: Header lines every generated reservation manifest carries.

220_RESERVATION_MANIFEST_NOTES = (

221 "#",

222 "# Apply with: kubectl apply -f <this-file>.yaml. The EKS endpoint is private by",

223 "# default: run `gco cluster tunnel --via-ssm auto` and add the kubectl flags it prints.",

224 "# Target the pool from a pod with nodeSelector karpenter.sh/nodepool: {name},",

225 "# plus a toleration for the pool's taint when it has one.",

226 "#",

227 "# Once any NodeClass selects capacity reservations, EKS Auto Mode stops",

228 "# launching into open ODCRs on its own for every NodeClass in the cluster.",

229 "# See: https://docs.aws.amazon.com/eks/latest/userguide/auto-odcr.html",

230)

231

232

233@dataclass(frozen=True)

234class AutoModeNodeClassInputs:

235 """What a custom Auto Mode NodeClass needs to launch nodes like the built-in one.

236

237 ``role`` is the cluster's Auto Mode node role, which already has the EC2

238 access entry nodes join with; ``subnet_ids`` and ``security_group_ids`` are

239 the cluster's own subnets and cluster security group, which the built-in

240 ``default`` NodeClass uses too.

241 """

242

243 role: str

244 subnet_ids: tuple[str, ...]

245 security_group_ids: tuple[str, ...]

246

247

248def resolve_auto_mode_node_class_inputs(cluster_name: str, region: str) -> AutoModeNodeClassInputs:

249 """Read the node role, subnets and security group of an Auto Mode cluster.

250

251 The CDK app names none of these (CloudFormation generates the role name),

252 so they come from ``eks:DescribeCluster`` at generation time.

253

254 Raises:

255 ValueError: The cluster does not run EKS Auto Mode, or reports no node

256 role, subnet, or cluster security group.

257 """

258 eks = boto3.client("eks", region_name=region)

259 cluster = eks.describe_cluster(name=cluster_name)["cluster"]

260 compute = cluster.get("computeConfig") or {}

261 role_arn = compute.get("nodeRoleArn")

262 if not compute.get("enabled") or not isinstance(role_arn, str) or not role_arn:

263 raise ValueError(

264 f"EKS cluster {cluster_name} in {region} does not run EKS Auto Mode with a "

265 "node role, so it has no NodeClass API to bind a reservation to"

266 )

267 vpc_config = cluster.get("resourcesVpcConfig") or {}

268 subnet_ids = tuple(

269 subnet for subnet in vpc_config.get("subnetIds") or () if isinstance(subnet, str) and subnet

270 )

271 security_group = vpc_config.get("clusterSecurityGroupId")

272 if not subnet_ids or not isinstance(security_group, str) or not security_group:

273 raise ValueError(

274 f"EKS cluster {cluster_name} in {region} reports no subnets or cluster security group"

275 )

276 return AutoModeNodeClassInputs(

277 role=role_arn.rsplit("/", 1)[-1],

278 subnet_ids=subnet_ids,

279 security_group_ids=(security_group,),

280 )

281

282

283def _reservation_node_class(

284 name: str,

285 capacity_reservation_id: str,

286 node_class: AutoModeNodeClassInputs,

287 reservation_label: str,

288) -> dict[str, Any]:

289 """The Auto Mode NodeClass that binds one capacity reservation."""

290 return {

291 "apiVersion": f"{AUTO_MODE_NODECLASS_GROUP}/{AUTO_MODE_NODECLASS_VERSION}",

292 "kind": AUTO_MODE_NODECLASS_KIND,

293 "metadata": {

294 "name": f"{name}-nodeclass",

295 "labels": {

296 "app.kubernetes.io/part-of": "gco",

297 "gco.io/nodepool": name,

298 reservation_label: capacity_reservation_id,

299 },

300 },

301 "spec": {

302 "role": node_class.role,

303 "subnetSelectorTerms": [{"id": subnet} for subnet in node_class.subnet_ids],

304 "securityGroupSelectorTerms": [

305 {"id": group} for group in node_class.security_group_ids

306 ],

307 "capacityReservationSelectorTerms": [{"id": capacity_reservation_id}],

308 },

309 }

310

311

312def _reservation_nodepool(

313 name: str,

314 facts: Sequence[InstanceTypeFacts],

315 max_nodes: int,

316 fallback_on_demand: bool,

317 efa: bool,

318 template_labels: dict[str, str],

319 disruption: dict[str, Any],

320) -> dict[str, Any]:

321 """The NodePool that launches reserved capacity through ``<name>-nodeclass``.

322

323 Reserved nodes get the same NoSchedule taint the shipped GPU and Neuron

324 NodePools carry, so only pods that tolerate it (the workloads the

325 reservation was bought for) consume them. Which taint follows from each

326 instance type's accelerator class in ``facts``, read from EC2 or the

327 reviewed catalog; no family list is kept here.

328 """

329 capacity_types = ["reserved", "on-demand"] if fallback_on_demand else ["reserved"]

330 requirements: list[dict[str, Any]] = [

331 {"key": "karpenter.sh/capacity-type", "operator": "In", "values": capacity_types},

332 ]

333 if facts:

334 # The reservation's instance types pin the architecture, Arm included.

335 requirements.append(

336 {

337 "key": "node.kubernetes.io/instance-type",

338 "operator": "In",

339 "values": [fact.instance_type for fact in facts],

340 }

341 )

342 else:

343 requirements.append({"key": "kubernetes.io/arch", "operator": "In", "values": ["amd64"]})

344

345 labels = dict(template_labels)

346 taints: list[dict[str, str]] = []

347 accelerators = {fact.accelerator for fact in facts}

348 if "nvidia" in accelerators:

349 taints.append({"key": "nvidia.com/gpu", "value": "true", "effect": "NoSchedule"})

350 if "neuron" in accelerators:

351 taints.append({"key": "aws.amazon.com/neuron", "value": "true", "effect": "NoSchedule"})

352 if efa:

353 labels["efa"] = "true"

354 labels["workload-type"] = "gpu-efa"

355 if taints:

356 taints.append({"key": "vpc.amazonaws.com/efa", "value": "true", "effect": "NoSchedule"})

357

358 template_spec: dict[str, Any] = {

359 "nodeClassRef": {

360 "group": AUTO_MODE_NODECLASS_GROUP,

361 "kind": AUTO_MODE_NODECLASS_KIND,

362 "name": f"{name}-nodeclass",

363 },

364 "requirements": requirements,

365 }

366 if taints:

367 template_spec["taints"] = taints

368

369 return {

370 "apiVersion": "karpenter.sh/v1",

371 "kind": "NodePool",

372 "metadata": {"name": name, "labels": {"app.kubernetes.io/part-of": "gco"}},

373 "spec": {

374 "template": {"metadata": {"labels": labels}, "spec": template_spec},

375 "limits": {"cpu": str(calculate_cpu_limit(facts, max_nodes))},

376 "disruption": disruption,

377 },

378 }

379

380

381def _render_reservation_manifest(

382 header: list[str],

383 name: str,

384 facts: Sequence[InstanceTypeFacts],

385 node_class: dict[str, Any],

386 nodepool: dict[str, Any],

387) -> str:

388 notes = [line.format(name=name) for line in _RESERVATION_MANIFEST_NOTES]

389 if facts:

390 notes = ["# Instance types:", *(fact.describe() for fact in facts), *notes]

391 return "\n".join(

392 [

393 *header,

394 *notes,

395 "---",

396 yaml.dump(node_class, default_flow_style=False, sort_keys=False),

397 "---",

398 yaml.dump(nodepool, default_flow_style=False, sort_keys=False),

399 ]

400 )

401

402

403def _instance_type_facts(

404 instance_types: Sequence[str] | None, region: str

405) -> tuple[InstanceTypeFacts, ...]:

406 return tuple(describe_instance_type(item, region) for item in instance_types or ())

407

408

409def generate_odcr_nodepool_manifest(

410 name: str,

411 region: str,

412 capacity_reservation_id: str,

413 *,

414 node_class: AutoModeNodeClassInputs,

415 instance_types: list[str] | None = None,

416 max_nodes: int = 100,

417 fallback_on_demand: bool = False,

418 efa: bool = False,

419) -> str:

420 """

421 Generate an EKS Auto Mode NodeClass and NodePool for ODCR-backed capacity.

422

423 The NodeClass binds the reservation through

424 ``capacityReservationSelectorTerms``; nodes launched into it carry

425 ``karpenter.sh/capacity-type: reserved`` and the

426 ``eks.amazonaws.com/capacity-reservation-id`` label.

427

428 Args:

429 name: Name for the NodePool (the NodeClass is ``<name>-nodeclass``)

430 region: AWS region

431 capacity_reservation_id: EC2 Capacity Reservation ID (cr-xxx)

432 node_class: The cluster's node role, subnets and security group, from

433 :func:`resolve_auto_mode_node_class_inputs`

434 instance_types: List of instance types (if None, any amd64 type the

435 reservation offers)

436 max_nodes: Maximum number of nodes

437 fallback_on_demand: Whether to fall back to on-demand if ODCR exhausted

438 efa: Whether to enable EFA support (adds EFA taint and labels)

439

440 Returns:

441 YAML manifest string for the NodeClass and NodePool

442 """

443 disruption: dict[str, Any] = {

444 "consolidationPolicy": "WhenEmptyOrUnderutilized",

445 "consolidateAfter": "30s",

446 "budgets": [{"nodes": "10%"}],

447 }

448 if efa:

449 # WhenEmpty for EFA workloads: never disrupt a long-running

450 # distributed training job to consolidate.

451 disruption = {

452 "consolidationPolicy": "WhenEmpty",

453 "consolidateAfter": "300s",

454 "budgets": [{"nodes": "10%"}],

455 }

456 facts = _instance_type_facts(instance_types, region)

457 nodepool = _reservation_nodepool(

458 name,

459 facts,

460 max_nodes,

461 fallback_on_demand,

462 efa,

463 {

464 "workload-type": "reserved-capacity",

465 "project": "gco",

466 "gco.io/capacity-reservation": capacity_reservation_id,

467 },

468 disruption,

469 )

470 header = [

471 "# ODCR-backed NodePool for GCO (EKS Auto Mode)",

472 f"# Capacity Reservation: {capacity_reservation_id}",

473 f"# Region: {region}",

474 ]

475 if fallback_on_demand:

476 header.append("# Fallback: on-demand (when ODCR exhausted)")

477 return _render_reservation_manifest(

478 header,

479 name,

480 facts,

481 _reservation_node_class(

482 name, capacity_reservation_id, node_class, "gco.io/capacity-reservation"

483 ),

484 nodepool,

485 )

486

487

488def generate_capacity_block_nodepool_manifest(

489 name: str,

490 region: str,

491 capacity_reservation_id: str,

492 *,

493 node_class: AutoModeNodeClassInputs,

494 instance_types: list[str] | None = None,

495 max_nodes: int = 100,

496 fallback_on_demand: bool = False,

497 efa: bool = False,

498) -> str:

499 """

500 Generate an EKS Auto Mode NodeClass and NodePool for a Capacity Block.

501

502 The Capacity Block counterpart to :func:`generate_odcr_nodepool_manifest`.

503 Purchasing a Capacity Block ("gco capacity reserve") yields a normal EC2

504 Capacity Reservation id (``cr-...``), which Auto Mode consumes through the

505 same ``capacityReservationSelectorTerms`` / ``reserved`` capacity type as an

506 ODCR; Auto Mode never uses a Capacity Block without such a selector. The

507 difference is intent: a Capacity Block is prepaid for a fixed term, so the

508 generated NodePool defaults to holding that capacity (``WhenEmpty``

509 consolidation with a long delay) rather than consolidating aggressively.

510

511 Args:

512 name: Name for the NodePool (the NodeClass is ``<name>-nodeclass``).

513 region: AWS region.

514 capacity_reservation_id: Capacity Reservation ID (cr-xxx) of the purchased

515 Capacity Block (from ``gco capacity reserve`` / reservation-check).

516 node_class: The cluster's node role, subnets and security group, from

517 :func:`resolve_auto_mode_node_class_inputs`.

518 instance_types: List of instance types (if None, any amd64 type).

519 max_nodes: Maximum number of nodes.

520 fallback_on_demand: Whether to fall back to on-demand when the block is

521 exhausted or after it expires. Off by default — a Capacity Block is

522 fixed-term guaranteed capacity, so silent on-demand fallback can

523 surprise-bill; opt in explicitly.

524 efa: Whether to enable EFA support (adds EFA taint and labels).

525

526 Returns:

527 YAML manifest string for the NodeClass and NodePool.

528 """

529 facts = _instance_type_facts(instance_types, region)

530 nodepool = _reservation_nodepool(

531 name,

532 facts,

533 max_nodes,

534 fallback_on_demand,

535 efa,

536 {

537 "workload-type": "capacity-block",

538 "project": "gco",

539 "gco.io/capacity-type": "capacity-block",

540 "gco.io/capacity-block": capacity_reservation_id,

541 },

542 # A Capacity Block is prepaid for a fixed term — hold the nodes rather

543 # than consolidating them away, so the paid capacity stays available

544 # for the whole block.

545 {

546 "consolidationPolicy": "WhenEmpty",

547 "consolidateAfter": "600s",

548 "budgets": [{"nodes": "10%"}],

549 },

550 )

551 header = [

552 "# Capacity Block-backed NodePool for GCO (EKS Auto Mode)",

553 f"# Capacity Reservation (from Capacity Block): {capacity_reservation_id}",

554 f"# Region: {region}",

555 ]

556 if fallback_on_demand:

557 header.append("# Fallback: on-demand (when Capacity Block exhausted/expired)")

558 return _render_reservation_manifest(

559 header,

560 name,

561 facts,

562 _reservation_node_class(name, capacity_reservation_id, node_class, "gco.io/capacity-block"),

563 nodepool,

564 )

565

566

567def get_eks_token(cluster_name: str, region: str) -> str:

568 """Generate EKS authentication token using STS presigned URL."""

569 from botocore.signers import RequestSigner

570

571 session = boto3.Session()

572 sts_client = session.client("sts", region_name=region)

573 service_id = sts_client.meta.service_model.service_id

574

575 signer = RequestSigner(

576 service_id, region, "sts", "v4", session.get_credentials(), session.events

577 )

578

579 params = {

580 "method": "GET",

581 "url": f"https://sts.{region}.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15",

582 "body": {},

583 "headers": {"x-k8s-aws-id": cluster_name},

584 "context": {},

585 }

586

587 url = signer.generate_presigned_url(

588 params, region_name=region, expires_in=60, operation_name=""

589 )

590

591 token_b64 = base64.urlsafe_b64encode(url.encode("utf-8")).decode("utf-8").rstrip("=")

592 return f"k8s-aws-v1.{token_b64}"

593

594

595def get_k8s_client(

596 cluster_name: str,

597 region: str,

598 *,

599 server: str | None = None,

600 tls_server_name: str | None = None,

601) -> CustomObjectsApi:

602 """Get configured Kubernetes client for EKS cluster.

603

604 ``server`` and ``tls_server_name`` point the client at an SSM tunnel to the

605 cluster's private endpoint (``gco cluster tunnel``): requests go to the

606 local tunnel address while TLS keeps the endpoint's own host name for SNI

607 and certificate verification, so the cluster CA still validates.

608 """

609 from kubernetes import client

610

611 eks = boto3.client("eks", region_name=region)

612 cluster_info = eks.describe_cluster(name=cluster_name)

613 cluster = cluster_info["cluster"]

614

615 configuration = client.Configuration()

616 configuration.host = server or cluster["endpoint"]

617 if tls_server_name:

618 configuration.tls_server_name = tls_server_name

619 configuration.verify_ssl = True

620

621 # The ApiClient reads this path throughout its lifetime. Keep the file until

622 # that client is collected, while cleaning every partially constructed path.

623 ca_cert = base64.b64decode(cluster["certificateAuthority"]["data"])

624 fd, ca_cert_path = tempfile.mkstemp(suffix=".crt")

625 fd_owned = True

626 api_client = None

627 try:

628 ca_file = os.fdopen(fd, "wb")

629 fd_owned = False # os.fdopen transferred descriptor ownership to ca_file.

630 with ca_file:

631 ca_file.write(ca_cert)

632 ca_file.flush()

633 configuration.ssl_ca_cert = ca_cert_path

634

635 # Generate EKS token

636 eks_token = get_eks_token(cluster_name, region)

637 configuration.api_key = {"authorization": f"Bearer {eks_token}"}

638

639 # Create API client with the configuration explicitly

640 api_client = client.ApiClient(configuration)

641 custom_api = client.CustomObjectsApi(api_client)

642 except BaseException:

643 if fd_owned:

644 with suppress(OSError):

645 os.close(fd)

646 if api_client is not None:

647 with suppress(Exception):

648 api_client.close()

649 _unlink_temp_ca_cert(ca_cert_path)

650 raise

651

652 # CustomObjectsApi retains api_client, so the certificate remains available

653 # for lazy TLS setup and is removed when the owning client is released.

654 weakref.finalize(api_client, _unlink_temp_ca_cert, ca_cert_path)

655 return custom_api

656

657

658def list_cluster_nodepools(

659 cluster_name: str,

660 region: str,

661 *,

662 server: str | None = None,

663 tls_server_name: str | None = None,

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

665 """

666 List NodePools in an EKS cluster.

667

668 Args:

669 cluster_name: EKS cluster name

670 region: AWS region

671 server: API server override (an SSM tunnel address); see :func:`get_k8s_client`

672 tls_server_name: The endpoint host name TLS verifies against through the tunnel

673

674 Returns:

675 List of NodePool information dictionaries

676 """

677 try:

678 custom_api = get_k8s_client(

679 cluster_name, region, server=server, tls_server_name=tls_server_name

680 )

681

682 nodepools = custom_api.list_cluster_custom_object(

683 group="karpenter.sh",

684 version="v1",

685 plural="nodepools",

686 )

687

688 result = []

689 for np in nodepools.get("items", []):

690 spec = np.get("spec", {})

691 template = spec.get("template", {}).get("spec", {})

692 requirements = template.get("requirements", [])

693

694 # Extract capacity types

695 capacity_types = []

696 instance_types = []

697 for req in requirements:

698 if req.get("key") == "karpenter.sh/capacity-type":

699 capacity_types = req.get("values", [])

700 elif req.get("key") == "node.kubernetes.io/instance-type":

701 instance_types = req.get("values", [])

702

703 # Get status

704 status = np.get("status", {})

705 conditions = status.get("conditions", [])

706 ready_condition: dict[str, Any] = next(

707 (c for c in conditions if c.get("type") == "Ready"), {}

708 )

709

710 result.append(

711 {

712 "name": np["metadata"]["name"],

713 "capacity_types": ", ".join(capacity_types) or "on-demand",

714 "instance_types": ", ".join(instance_types[:3])

715 + ("..." if len(instance_types) > 3 else "")

716 or "any",

717 "status": "Ready" if ready_condition.get("status") == "True" else "NotReady",

718 "limits": spec.get("limits", {}),

719 }

720 )

721

722 return result

723

724 except Exception as e:

725 raise RuntimeError(f"Failed to list NodePools: {e}") from e

726

727

728def describe_cluster_nodepool(

729 cluster_name: str,

730 region: str,

731 nodepool_name: str,

732 *,

733 server: str | None = None,

734 tls_server_name: str | None = None,

735) -> dict[str, Any] | None:

736 """

737 Describe a specific NodePool in an EKS cluster.

738

739 Args:

740 cluster_name: EKS cluster name

741 region: AWS region

742 nodepool_name: Name of the NodePool

743 server: API server override (an SSM tunnel address); see :func:`get_k8s_client`

744 tls_server_name: The endpoint host name TLS verifies against through the tunnel

745

746 Returns:

747 NodePool details or None if not found

748 """

749 try:

750 custom_api = get_k8s_client(

751 cluster_name, region, server=server, tls_server_name=tls_server_name

752 )

753

754 nodepool = custom_api.get_cluster_custom_object(

755 group="karpenter.sh",

756 version="v1",

757 plural="nodepools",

758 name=nodepool_name,

759 )

760

761 if isinstance(nodepool, dict):

762 return nodepool

763 return None

764

765 except Exception as e:

766 if "404" in str(e):

767 return None

768 raise RuntimeError(f"Failed to describe NodePool: {e}") from e

769

770

771def delete_cluster_nodepool(

772 cluster_name: str,

773 region: str,

774 nodepool_name: str,

775 *,

776 server: str | None = None,

777 tls_server_name: str | None = None,

778) -> dict[str, Any]:

779 """

780 Delete a NodePool (and its paired NodeClass) from an EKS cluster.

781

782 Deletes the NodePool ``nodepool_name`` and, when present, the Auto Mode

783 NodeClass named ``<nodepool_name>-nodeclass`` that the GCO manifest

784 generators create alongside it. EKS Auto Mode drains and terminates any

785 nodes the NodePool provisioned once it is removed. A missing NodeClass

786 (custom name, or already deleted) is not treated as an error.

787

788 Args:

789 cluster_name: EKS cluster name

790 region: AWS region

791 nodepool_name: Name of the NodePool to delete

792 server: API server override (an SSM tunnel address); see :func:`get_k8s_client`

793 tls_server_name: The endpoint host name TLS verifies against through the tunnel

794

795 Returns:

796 Dict describing what was deleted: ``{"nodepool": <name>,

797 "nodeclass": <name-or-None>}``.

798 """

799 try:

800 custom_api = get_k8s_client(

801 cluster_name, region, server=server, tls_server_name=tls_server_name

802 )

803

804 custom_api.delete_cluster_custom_object(

805 group="karpenter.sh",

806 version="v1",

807 plural="nodepools",

808 name=nodepool_name,

809 )

810 deleted: dict[str, Any] = {"nodepool": nodepool_name, "nodeclass": None}

811

812 nodeclass_name = f"{nodepool_name}-nodeclass"

813 try:

814 custom_api.delete_cluster_custom_object(

815 group=AUTO_MODE_NODECLASS_GROUP,

816 version=AUTO_MODE_NODECLASS_VERSION,

817 plural=AUTO_MODE_NODECLASS_PLURAL,

818 name=nodeclass_name,

819 )

820 deleted["nodeclass"] = nodeclass_name

821 except Exception as e: # best effort; the NodePool is the primary target

822 if "404" not in str(e):

823 logger.warning("Could not delete NodeClass %s: %s", nodeclass_name, e)

824

825 return deleted

826

827 except Exception as e:

828 raise RuntimeError(f"Failed to delete NodePool: {e}") from e

Coverage for cli / nodepools.py: 100.00%

253 statements