Coverage for cli / files.py: 100.00%
295 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2File system operations for GCO CLI.
4Provides functionality to interact with EFS and FSx for Lustre file systems
5attached to GCO regional stacks.
6"""
8import json
9import re
10from dataclasses import dataclass
11from datetime import UTC, datetime
12from pathlib import Path, PurePosixPath
13from typing import Any
15import boto3
16from botocore.exceptions import ClientError
18from ._image_uri import aws_partition, aws_url_suffix
19from .aws_client import get_aws_client
20from .config import GCOConfig, get_config
21from .kubectl_helpers import update_kubeconfig
23_DNS_LABEL_RE = re.compile(r"[a-z0-9](?:[-a-z0-9]*[a-z0-9])?")
24_DNS_SUBDOMAIN_RE = re.compile(
25 r"[a-z0-9](?:[-a-z0-9]*[a-z0-9])?(?:\.[a-z0-9](?:[-a-z0-9]*[a-z0-9])?)*"
26)
29def _validated_kubernetes_name(
30 value: str,
31 field: str,
32 *,
33 allow_subdomains: bool,
34) -> str:
35 """Return a Kubernetes name after strict DNS syntax validation."""
36 max_length = 253 if allow_subdomains else 63
37 pattern = _DNS_SUBDOMAIN_RE if allow_subdomains else _DNS_LABEL_RE
38 if not isinstance(value, str) or not value or len(value) > max_length:
39 raise ValueError(f"{field} must be a non-empty Kubernetes DNS name")
40 if pattern.fullmatch(value) is None:
41 raise ValueError(f"{field} contains characters not allowed in a Kubernetes DNS name")
42 return value
45def _validated_copy_path(value: str, field: str) -> str:
46 """Reject empty paths and control characters before passing them to kubectl."""
47 if not isinstance(value, str) or not value:
48 raise ValueError(f"{field} must be a non-empty path")
49 if any(ord(char) < 32 or ord(char) == 127 for char in value):
50 raise ValueError(f"{field} must not contain control characters")
51 return value
54def _validated_local_destination(local_path: str) -> str:
55 """Ensure kubectl cannot reinterpret the local destination as a pod spec."""
56 local_path = _validated_copy_path(local_path, "local_path")
57 if ":" in local_path:
58 raise ValueError("local_path must not contain ':' because kubectl treats it as a pod path")
59 return local_path
62def _create_local_destination_parent(local_path: str) -> None:
63 """Create only the parent kubectl needs, preserving leaf copy semantics."""
64 Path(local_path).parent.mkdir(parents=True, exist_ok=True)
67def _validated_pod_remote_path(remote_path: str) -> str:
68 """Require an absolute pod path so remote tar sees no ambiguous input."""
69 remote_path = _validated_copy_path(remote_path, "remote_path")
70 if "\\" in remote_path:
71 raise ValueError("remote_path must not contain backslashes")
72 if not remote_path.startswith("/"):
73 raise ValueError("remote_path must be absolute when copying from a pod")
74 return remote_path
77def _storage_sub_path(remote_path: str) -> str | None:
78 """Return a safe PVC subPath, or ``None`` when the root was requested."""
79 remote_path = _validated_copy_path(remote_path, "remote_path")
80 if "\\" in remote_path:
81 raise ValueError("remote_path must not contain backslashes")
82 relative = remote_path.lstrip("/")
83 if not relative:
84 return None
85 if any(part in (".", "..") for part in relative.split("/")):
86 raise ValueError("remote_path must stay beneath the mounted storage root")
87 return str(PurePosixPath(relative))
90def _storage_remote_path(mount_path: str, remote_path: str) -> str:
91 """Resolve a user path beneath a helper pod's storage mount.
93 The same validated relative path is mounted with Kubernetes ``subPath`` in
94 the helper pod. Kubelet's symlink-safe subPath resolution makes the runtime
95 boundary stronger than a lexical ``..`` check alone.
96 """
97 sub_path = _storage_sub_path(remote_path)
98 if sub_path is None:
99 return mount_path
100 return str(PurePosixPath(mount_path).joinpath(sub_path))
103def _helper_pod_manifest(
104 *,
105 pod_name: str,
106 namespace: str,
107 pvc_name: str,
108 mount_path: str,
109 app_label: str,
110 storage_sub_path: str | None = None,
111) -> str:
112 """Build a JSON Kubernetes manifest without YAML string interpolation."""
113 volume_mount: dict[str, Any] = {
114 "name": "storage",
115 "mountPath": mount_path,
116 "readOnly": True,
117 }
118 if storage_sub_path is not None:
119 volume_mount["subPath"] = storage_sub_path
120 manifest = {
121 "apiVersion": "v1",
122 "kind": "Pod",
123 "metadata": {
124 "name": pod_name,
125 "namespace": namespace,
126 "labels": {"app": app_label},
127 },
128 "spec": {
129 "automountServiceAccountToken": False,
130 "enableServiceLinks": False,
131 "restartPolicy": "Never",
132 "containers": [
133 {
134 "name": "helper",
135 "image": "busybox:1.38.0",
136 "command": ["sleep", "300"],
137 "resources": {
138 "requests": {"cpu": "50m", "memory": "64Mi"},
139 "limits": {"cpu": "200m", "memory": "256Mi"},
140 },
141 "volumeMounts": [volume_mount],
142 }
143 ],
144 "volumes": [
145 {
146 "name": "storage",
147 "persistentVolumeClaim": {"claimName": pvc_name},
148 }
149 ],
150 },
151 }
152 return json.dumps(manifest)
155@dataclass
156class FileSystemInfo:
157 """Information about a file system."""
159 file_system_id: str
160 file_system_type: str # "efs" or "fsx"
161 region: str
162 dns_name: str
163 mount_target_ip: str | None = None
164 size_bytes: int | None = None
165 status: str = "available"
166 created_time: datetime | None = None
167 tags: dict[str, str] | None = None
169 def __post_init__(self) -> None:
170 if self.tags is None:
171 self.tags = {}
174@dataclass
175class FileInfo:
176 """Information about a file or directory."""
178 path: str
179 name: str
180 is_directory: bool
181 size_bytes: int = 0
182 modified_time: datetime | None = None
183 owner: str | None = None
186class FileSystemClient:
187 """
188 Client for interacting with GCO file systems.
190 Supports:
191 - Listing file systems (EFS/FSx) in GCO stacks
192 - Getting file system information and access points
193 - Downloading files from pods via kubectl cp
194 """
196 def __init__(self, config: GCOConfig | None = None):
197 self.config = config or get_config()
198 self._session = boto3.Session()
199 self._aws_client = get_aws_client(config)
201 def get_file_systems(self, region: str | None = None) -> list[FileSystemInfo]:
202 """
203 Get all file systems associated with GCO stacks.
205 Args:
206 region: Specific region to query (None for all regions)
208 Returns:
209 List of FileSystemInfo objects
210 """
211 file_systems = []
213 # Get regional stacks
214 stacks = self._aws_client.discover_regional_stacks()
216 if region:
217 stacks = {k: v for k, v in stacks.items() if k == region}
219 for stack_region, stack in stacks.items():
220 # Get EFS file systems
221 if stack.efs_file_system_id:
222 efs_info = self._get_efs_info(stack.efs_file_system_id, stack_region)
223 if efs_info:
224 file_systems.append(efs_info)
226 # Get FSx file systems
227 if stack.fsx_file_system_id:
228 fsx_info = self._get_fsx_info(stack.fsx_file_system_id, stack_region)
229 if fsx_info:
230 file_systems.append(fsx_info)
232 return file_systems
234 def _get_efs_info(self, file_system_id: str, region: str) -> FileSystemInfo | None:
235 """Get information about an EFS file system."""
236 try:
237 efs = self._session.client("efs", region_name=region)
239 response = efs.describe_file_systems(FileSystemId=file_system_id)
240 if not response["FileSystems"]:
241 return None
243 fs = response["FileSystems"][0]
245 # Get mount targets for DNS name
246 mt_response = efs.describe_mount_targets(FileSystemId=file_system_id)
247 mount_target_ip = None
248 if mt_response["MountTargets"]:
249 mount_target_ip = mt_response["MountTargets"][0].get("IpAddress")
251 # Get tags
252 tags_response = efs.describe_tags(FileSystemId=file_system_id)
253 tags = {t["Key"]: t["Value"] for t in tags_response.get("Tags", [])}
255 return FileSystemInfo(
256 file_system_id=file_system_id,
257 file_system_type="efs",
258 region=region,
259 dns_name=f"{file_system_id}.efs.{region}.{aws_url_suffix(region)}",
260 mount_target_ip=mount_target_ip,
261 size_bytes=fs.get("SizeInBytes", {}).get("Value"),
262 status=fs["LifeCycleState"],
263 created_time=fs.get("CreationTime"),
264 tags=tags,
265 )
266 except ClientError:
267 return None
269 def _get_fsx_info(self, file_system_id: str, region: str) -> FileSystemInfo | None:
270 """Get information about an FSx for Lustre file system."""
271 try:
272 fsx = self._session.client("fsx", region_name=region)
274 response = fsx.describe_file_systems(FileSystemIds=[file_system_id])
275 if not response["FileSystems"]:
276 return None
278 fs = response["FileSystems"][0]
280 # Get DNS name from Lustre configuration
281 dns_name = fs.get("DNSName", "")
283 # Get tags
284 tags = {t["Key"]: t["Value"] for t in fs.get("Tags", [])}
286 return FileSystemInfo(
287 file_system_id=file_system_id,
288 file_system_type="fsx",
289 region=region,
290 dns_name=dns_name,
291 size_bytes=fs.get("StorageCapacity", 0) * 1024 * 1024 * 1024, # GB to bytes
292 status=fs["Lifecycle"],
293 created_time=fs.get("CreationTime"),
294 tags=tags,
295 )
296 except ClientError:
297 return None
299 def get_file_system_by_region(self, region: str, fs_type: str = "efs") -> FileSystemInfo | None:
300 """
301 Get file system for a specific region.
303 Args:
304 region: AWS region
305 fs_type: "efs" or "fsx"
307 Returns:
308 FileSystemInfo or None
309 """
310 file_systems = self.get_file_systems(region)
311 for fs in file_systems:
312 if fs.file_system_type == fs_type:
313 return fs
314 return None
316 def create_datasync_download_task(
317 self,
318 file_system_id: str,
319 region: str,
320 source_path: str,
321 destination_bucket: str,
322 destination_prefix: str = "",
323 ) -> str:
324 """
325 Create a DataSync task to download files from EFS/FSx to S3.
327 This is useful for downloading large amounts of data from file systems
328 that aren't directly accessible.
330 Args:
331 file_system_id: EFS or FSx file system ID
332 region: AWS region
333 source_path: Path within the file system
334 destination_bucket: S3 bucket name
335 destination_prefix: S3 key prefix
337 Returns:
338 DataSync task ARN
339 """
340 datasync = self._session.client("datasync", region_name=region)
341 partition = aws_partition(region)
343 # Determine file system type
344 fs_info = None
345 for fs in self.get_file_systems(region):
346 if fs.file_system_id == file_system_id:
347 fs_info = fs
348 break
350 if not fs_info:
351 raise ValueError(f"File system {file_system_id} not found in region {region}")
353 # Create source location
354 if fs_info.file_system_type == "efs":
355 source_location = datasync.create_location_efs(
356 EfsFilesystemArn=f"arn:{partition}:elasticfilesystem:{region}:{self._get_account_id()}:file-system/{file_system_id}",
357 Subdirectory=source_path,
358 Ec2Config={
359 "SubnetArn": self._get_subnet_arn(region),
360 "SecurityGroupArns": [self._get_security_group_arn(region)],
361 },
362 )
363 source_arn = source_location["LocationArn"]
364 else:
365 source_location = datasync.create_location_fsx_lustre(
366 FsxFilesystemArn=f"arn:{partition}:fsx:{region}:{self._get_account_id()}:file-system/{file_system_id}",
367 Subdirectory=source_path,
368 SecurityGroupArns=[self._get_security_group_arn(region)],
369 )
370 source_arn = source_location["LocationArn"]
372 # Create destination location (S3)
373 dest_location = datasync.create_location_s3(
374 S3BucketArn=f"arn:{partition}:s3:::{destination_bucket}",
375 Subdirectory=destination_prefix,
376 S3Config={"BucketAccessRoleArn": self._get_datasync_role_arn(region)},
377 )
378 dest_arn = dest_location["LocationArn"]
380 # Create task
381 task = datasync.create_task(
382 SourceLocationArn=source_arn,
383 DestinationLocationArn=dest_arn,
384 Name=f"gco-download-{datetime.now(UTC).strftime('%Y%m%d-%H%M%S')}",
385 Options={
386 "VerifyMode": "ONLY_FILES_TRANSFERRED",
387 "OverwriteMode": "ALWAYS",
388 "PreserveDeletedFiles": "REMOVE",
389 "TransferMode": "CHANGED",
390 },
391 )
393 return str(task["TaskArn"])
395 def _get_account_id(self) -> str:
396 """Get current AWS account ID."""
397 sts = self._session.client("sts")
398 return str(sts.get_caller_identity()["Account"])
400 def _get_subnet_arn(self, _region: str) -> str:
401 """Get a subnet ARN for DataSync in the given region."""
402 # This would need to be implemented based on your VPC setup
403 # For now, return a placeholder
404 raise NotImplementedError("Subnet ARN lookup not implemented - configure via stack outputs")
406 def _get_security_group_arn(self, _region: str) -> str:
407 """Get a security group ARN for DataSync in the given region."""
408 raise NotImplementedError(
409 "Security group ARN lookup not implemented - configure via stack outputs"
410 )
412 def _get_datasync_role_arn(self, region: str) -> str:
413 """Get the DataSync IAM role ARN."""
414 raise NotImplementedError(
415 "DataSync role ARN lookup not implemented - configure via stack outputs"
416 )
418 def get_access_point_info(self, file_system_id: str, region: str) -> list[dict[str, Any]]:
419 """
420 Get EFS access points for a file system.
422 Args:
423 file_system_id: EFS file system ID
424 region: AWS region
426 Returns:
427 List of access point information
428 """
429 try:
430 efs = self._session.client("efs", region_name=region)
431 response = efs.describe_access_points(FileSystemId=file_system_id)
433 return [
434 {
435 "access_point_id": ap["AccessPointId"],
436 "name": ap.get("Name", ""),
437 "path": ap.get("RootDirectory", {}).get("Path", "/"),
438 "posix_user": ap.get("PosixUser", {}),
439 "status": ap["LifeCycleState"],
440 }
441 for ap in response.get("AccessPoints", [])
442 ]
443 except ClientError:
444 return []
446 def download_from_pod(
447 self,
448 region: str,
449 pod_name: str,
450 remote_path: str,
451 local_path: str,
452 namespace: str = "gco-jobs",
453 container: str | None = None,
454 ) -> dict[str, Any]:
455 """
456 Download files from a pod using kubectl cp.
458 This uses kubectl port-forward internally to copy files from a pod's
459 mounted file system (EFS/FSx) to the local machine.
461 Args:
462 region: AWS region where the cluster is located
463 pod_name: Name of the pod to copy from
464 remote_path: Path inside the pod (e.g., /mnt/efs/outputs)
465 local_path: Local destination path
466 namespace: Kubernetes namespace (default: gco-jobs)
467 container: Container name (optional, for multi-container pods)
469 Returns:
470 Dict with download status and details
471 """
472 import os
473 import subprocess
475 namespace = _validated_kubernetes_name(namespace, "namespace", allow_subdomains=False)
476 pod_name = _validated_kubernetes_name(pod_name, "pod_name", allow_subdomains=True)
477 remote_path = _validated_pod_remote_path(remote_path)
478 local_path = _validated_local_destination(local_path)
479 if container is not None:
480 container = _validated_kubernetes_name(container, "container", allow_subdomains=False)
481 _create_local_destination_parent(local_path)
483 # Update kubeconfig for the cluster
484 cluster_name = f"{self.config.project_name}-{region}"
485 update_kubeconfig(cluster_name, region)
487 # Build kubectl cp command
488 # Format: kubectl cp <namespace>/<pod>:<remote_path> <local_path>
489 source = f"{namespace}/{pod_name}:{remote_path}"
490 cmd = ["kubectl", "cp"]
492 if container:
493 cmd.extend(["-c", container])
494 cmd.extend(["--", source, local_path])
496 try:
497 subprocess.run(
498 cmd, check=True, capture_output=True, text=True
499 ) # nosemgrep: dangerous-subprocess-use-audit - cmd is a list ["kubectl","cp",source,local_path]; source is namespace/pod:path, local_path is caller-provided destination
500 if os.path.isfile(local_path):
501 size = os.path.getsize(local_path)
502 elif os.path.isdir(local_path):
503 size = sum(
504 os.path.getsize(os.path.join(dirpath, filename))
505 for dirpath, _, filenames in os.walk(local_path)
506 for filename in filenames
507 )
508 else:
509 raise RuntimeError(
510 "kubectl cp reported success but did not create the local destination: "
511 f"{local_path}"
512 )
514 return {
515 "status": "success",
516 "source": source,
517 "destination": local_path,
518 "size_bytes": size,
519 "message": "Download completed successfully",
520 }
522 except subprocess.CalledProcessError as e:
523 raise RuntimeError(f"kubectl cp failed: {e.stderr}") from e
524 except FileNotFoundError as e:
525 raise RuntimeError(
526 "kubectl not found. Please install kubectl and ensure it's in your PATH."
527 ) from e
529 def list_storage_contents(
530 self,
531 region: str,
532 remote_path: str = "/",
533 storage_type: str = "efs",
534 namespace: str = "gco-jobs",
535 pvc_name: str | None = None,
536 ) -> dict[str, Any]:
537 """
538 List contents of EFS/FSx storage using a temporary helper pod.
540 This creates a temporary pod that mounts the storage, lists contents,
541 then cleans up. Useful for discovering what directories/files exist.
543 Args:
544 region: AWS region where the cluster is located
545 remote_path: Path inside the storage to list (default: root)
546 storage_type: "efs" or "fsx" (default: efs)
547 namespace: Kubernetes namespace (default: gco-jobs)
548 pvc_name: PVC name to mount (default: gco-shared-storage for EFS,
549 gco-fsx-storage for FSx)
551 Returns:
552 Dict with listing status and contents
553 """
554 import subprocess
555 import time
556 import uuid
558 if storage_type not in ("efs", "fsx"):
559 raise ValueError("storage_type must be 'efs' or 'fsx'")
560 namespace = _validated_kubernetes_name(namespace, "namespace", allow_subdomains=False)
562 # Determine PVC name based on storage type
563 if pvc_name is None:
564 pvc_name = "gco-shared-storage" if storage_type == "efs" else "gco-fsx-storage"
565 pvc_name = _validated_kubernetes_name(pvc_name, "pvc_name", allow_subdomains=True)
567 # Determine mount path based on storage type
568 mount_path = "/efs" if storage_type == "efs" else "/fsx"
569 storage_sub_path = _storage_sub_path(remote_path)
570 full_remote_path = _storage_remote_path(mount_path, remote_path)
572 # Generate unique pod name
573 helper_pod_name = f"gco-list-helper-{uuid.uuid4().hex[:8]}"
575 # Update kubeconfig for the cluster
576 cluster_name = f"{self.config.project_name}-{region}"
577 update_kubeconfig(cluster_name, region)
579 # Create helper pod manifest.
580 #
581 # We set an explicit ``resources`` block with CPU + memory but no GPU
582 # so the gco-jobs LimitRange admission plugin does not substitute its
583 # ``max`` value as an implicit request. Without this, a namespace
584 # that already has all 32 GPUs in use (typical during a demo or
585 # heavy workload burst) would reject the helper pod — even though
586 # listing files doesn't need a GPU — because K8s quota admission
587 # would attribute the LimitRange's ``max.nvidia.com/gpu`` to the
588 # pod's request.
589 pod_manifest = _helper_pod_manifest(
590 pod_name=helper_pod_name,
591 namespace=namespace,
592 pvc_name=pvc_name,
593 mount_path=full_remote_path,
594 app_label="gco-list-helper",
595 storage_sub_path=storage_sub_path,
596 )
598 try:
599 # Create the helper pod
600 subprocess.run(
601 ["kubectl", "apply", "-f", "-"],
602 input=pod_manifest,
603 capture_output=True,
604 text=True,
605 check=True,
606 )
608 # Wait for pod to be ready
609 max_wait = 60
610 waited = 0
611 while waited < max_wait:
612 status_result = subprocess.run(
613 [
614 "kubectl",
615 "get",
616 "pod",
617 helper_pod_name,
618 "-n",
619 namespace,
620 "-o",
621 "jsonpath={.status.phase}",
622 ],
623 capture_output=True,
624 text=True,
625 check=False,
626 )
627 if status_result.stdout.strip() == "Running":
628 break
629 time.sleep(2) # nosemgrep: arbitrary-sleep
630 waited += 2
632 if waited >= max_wait:
633 raise RuntimeError("Helper pod did not become ready in time")
635 # List contents using kubectl exec
636 list_result = subprocess.run(
637 [
638 "kubectl",
639 "exec",
640 helper_pod_name,
641 "-n",
642 namespace,
643 "--",
644 "ls",
645 "-la",
646 "--",
647 full_remote_path,
648 ],
649 capture_output=True,
650 text=True,
651 check=False,
652 )
654 if list_result.returncode != 0:
655 return {
656 "status": "error",
657 "path": remote_path,
658 "storage_type": storage_type,
659 "contents": [],
660 "message": f"Path not found or empty: {list_result.stderr.strip()}",
661 }
663 # Parse ls output
664 contents = []
665 for line in list_result.stdout.strip().split("\n"):
666 if line.startswith("total") or not line.strip():
667 continue
668 parts = line.split()
669 if len(parts) >= 9:
670 name = " ".join(parts[8:])
671 is_dir = line.startswith("d")
672 size = int(parts[4]) if parts[4].isdigit() else 0
673 contents.append(
674 {
675 "name": name,
676 "is_directory": is_dir,
677 "size_bytes": size,
678 "permissions": parts[0],
679 }
680 )
682 return {
683 "status": "success",
684 "path": remote_path,
685 "storage_type": storage_type,
686 "contents": contents,
687 "message": f"Found {len(contents)} items",
688 }
690 except subprocess.CalledProcessError as e:
691 error_msg = e.stderr if e.stderr else str(e)
692 raise RuntimeError(f"List failed: {error_msg}") from e
693 except FileNotFoundError as e:
694 raise RuntimeError(
695 "kubectl not found. Please install kubectl and ensure it's in your PATH."
696 ) from e
697 finally:
698 # Always clean up the helper pod
699 import contextlib
701 with contextlib.suppress(Exception):
702 subprocess.run(
703 [
704 "kubectl",
705 "delete",
706 "pod",
707 helper_pod_name,
708 "-n",
709 namespace,
710 "--ignore-not-found",
711 ],
712 capture_output=True,
713 text=True,
714 check=False,
715 )
717 def download_from_storage(
718 self,
719 region: str,
720 remote_path: str,
721 local_path: str,
722 storage_type: str = "efs",
723 namespace: str = "gco-jobs",
724 pvc_name: str | None = None,
725 ) -> dict[str, Any]:
726 """
727 Download files from EFS/FSx storage using a temporary helper pod.
729 This creates a temporary pod that mounts the storage, copies files via
730 kubectl cp, then cleans up. Works even after the original job pod is gone.
732 Args:
733 region: AWS region where the cluster is located
734 remote_path: Path inside the storage (e.g., /efs-output-example/results.json)
735 local_path: Local destination path
736 storage_type: "efs" or "fsx" (default: efs)
737 namespace: Kubernetes namespace (default: gco-jobs)
738 pvc_name: PVC name to mount (default: gco-shared-storage for EFS,
739 gco-fsx-storage for FSx)
741 Returns:
742 Dict with download status and details
743 """
744 import os
745 import subprocess
746 import time
747 import uuid
749 if storage_type not in ("efs", "fsx"):
750 raise ValueError("storage_type must be 'efs' or 'fsx'")
751 namespace = _validated_kubernetes_name(namespace, "namespace", allow_subdomains=False)
752 local_path = _validated_local_destination(local_path)
754 # Determine PVC name based on storage type
755 if pvc_name is None:
756 pvc_name = "gco-shared-storage" if storage_type == "efs" else "gco-fsx-storage"
757 pvc_name = _validated_kubernetes_name(pvc_name, "pvc_name", allow_subdomains=True)
759 # Determine mount path based on storage type
760 mount_path = "/efs" if storage_type == "efs" else "/fsx"
761 storage_sub_path = _storage_sub_path(remote_path)
762 full_remote_path = _storage_remote_path(mount_path, remote_path)
763 _create_local_destination_parent(local_path)
765 # Generate unique pod name
766 helper_pod_name = f"gco-download-helper-{uuid.uuid4().hex[:8]}"
768 # Update kubeconfig for the cluster
769 cluster_name = f"{self.config.project_name}-{region}"
770 update_kubeconfig(cluster_name, region)
772 # Create helper pod manifest.
773 #
774 # Explicit ``resources`` block avoids the LimitRange admission plugin
775 # substituting ``max.nvidia.com/gpu`` as an implicit request — see the
776 # ``ls`` helper above for the full rationale.
777 pod_manifest = _helper_pod_manifest(
778 pod_name=helper_pod_name,
779 namespace=namespace,
780 pvc_name=pvc_name,
781 mount_path=full_remote_path,
782 app_label="gco-download-helper",
783 storage_sub_path=storage_sub_path,
784 )
786 try:
787 # Create the helper pod
788 subprocess.run(
789 ["kubectl", "apply", "-f", "-"],
790 input=pod_manifest,
791 capture_output=True,
792 text=True,
793 check=True,
794 )
796 # Wait for pod to be ready
797 max_wait = 60
798 waited = 0
799 while waited < max_wait:
800 status_result = subprocess.run(
801 [
802 "kubectl",
803 "get",
804 "pod",
805 helper_pod_name,
806 "-n",
807 namespace,
808 "-o",
809 "jsonpath={.status.phase}",
810 ],
811 capture_output=True,
812 text=True,
813 check=False,
814 )
815 if status_result.stdout.strip() == "Running":
816 break
817 time.sleep(2) # nosemgrep: arbitrary-sleep
818 waited += 2
820 if waited >= max_wait:
821 raise RuntimeError("Helper pod did not become ready in time")
823 # Copy files from the helper pod
824 source = f"{namespace}/{helper_pod_name}:{full_remote_path}"
825 cmd = ["kubectl", "cp", "--", source, local_path]
827 subprocess.run(
828 cmd, check=True, capture_output=True, text=True
829 ) # nosemgrep: dangerous-subprocess-use-audit - cmd is a list ["kubectl","cp",source,local_path]; source is namespace/pod:path, local_path is caller-provided destination
831 # Get file info
832 if os.path.isfile(local_path):
833 size = os.path.getsize(local_path)
834 elif os.path.isdir(local_path):
835 size = sum(
836 os.path.getsize(os.path.join(dirpath, filename))
837 for dirpath, _, filenames in os.walk(local_path)
838 for filename in filenames
839 )
840 else:
841 raise RuntimeError(
842 "kubectl cp reported success but did not create the local destination: "
843 f"{local_path}"
844 )
846 return {
847 "status": "success",
848 "source": f"{storage_type}:{remote_path}",
849 "destination": local_path,
850 "size_bytes": size,
851 "storage_type": storage_type,
852 "message": "Download completed successfully",
853 }
855 except subprocess.CalledProcessError as e:
856 error_msg = e.stderr if e.stderr else str(e)
857 raise RuntimeError(f"Download failed: {error_msg}") from e
858 except FileNotFoundError as e:
859 raise RuntimeError(
860 "kubectl not found. Please install kubectl and ensure it's in your PATH."
861 ) from e
862 finally:
863 # Always clean up the helper pod
864 import contextlib
866 with contextlib.suppress(Exception):
867 subprocess.run(
868 [
869 "kubectl",
870 "delete",
871 "pod",
872 helper_pod_name,
873 "-n",
874 namespace,
875 "--ignore-not-found",
876 ],
877 capture_output=True,
878 text=True,
879 check=False,
880 )
883def get_file_system_client(config: GCOConfig | None = None) -> FileSystemClient:
884 """Get a configured file system client instance."""
885 return FileSystemClient(config)