Coverage for cli / files.py: 100.00%
295 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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 )
626 if status_result.stdout.strip() == "Running":
627 break
628 time.sleep(2) # nosemgrep: arbitrary-sleep
629 waited += 2
631 if waited >= max_wait:
632 raise RuntimeError("Helper pod did not become ready in time")
634 # List contents using kubectl exec
635 list_result = subprocess.run(
636 [
637 "kubectl",
638 "exec",
639 helper_pod_name,
640 "-n",
641 namespace,
642 "--",
643 "ls",
644 "-la",
645 "--",
646 full_remote_path,
647 ],
648 capture_output=True,
649 text=True,
650 )
652 if list_result.returncode != 0:
653 return {
654 "status": "error",
655 "path": remote_path,
656 "storage_type": storage_type,
657 "contents": [],
658 "message": f"Path not found or empty: {list_result.stderr.strip()}",
659 }
661 # Parse ls output
662 contents = []
663 for line in list_result.stdout.strip().split("\n"):
664 if line.startswith("total") or not line.strip():
665 continue
666 parts = line.split()
667 if len(parts) >= 9:
668 name = " ".join(parts[8:])
669 is_dir = line.startswith("d")
670 size = int(parts[4]) if parts[4].isdigit() else 0
671 contents.append(
672 {
673 "name": name,
674 "is_directory": is_dir,
675 "size_bytes": size,
676 "permissions": parts[0],
677 }
678 )
680 return {
681 "status": "success",
682 "path": remote_path,
683 "storage_type": storage_type,
684 "contents": contents,
685 "message": f"Found {len(contents)} items",
686 }
688 except subprocess.CalledProcessError as e:
689 error_msg = e.stderr if e.stderr else str(e)
690 raise RuntimeError(f"List failed: {error_msg}") from e
691 except FileNotFoundError as e:
692 raise RuntimeError(
693 "kubectl not found. Please install kubectl and ensure it's in your PATH."
694 ) from e
695 finally:
696 # Always clean up the helper pod
697 import contextlib
699 with contextlib.suppress(Exception):
700 subprocess.run(
701 [
702 "kubectl",
703 "delete",
704 "pod",
705 helper_pod_name,
706 "-n",
707 namespace,
708 "--ignore-not-found",
709 ],
710 capture_output=True,
711 text=True,
712 )
714 def download_from_storage(
715 self,
716 region: str,
717 remote_path: str,
718 local_path: str,
719 storage_type: str = "efs",
720 namespace: str = "gco-jobs",
721 pvc_name: str | None = None,
722 ) -> dict[str, Any]:
723 """
724 Download files from EFS/FSx storage using a temporary helper pod.
726 This creates a temporary pod that mounts the storage, copies files via
727 kubectl cp, then cleans up. Works even after the original job pod is gone.
729 Args:
730 region: AWS region where the cluster is located
731 remote_path: Path inside the storage (e.g., /efs-output-example/results.json)
732 local_path: Local destination path
733 storage_type: "efs" or "fsx" (default: efs)
734 namespace: Kubernetes namespace (default: gco-jobs)
735 pvc_name: PVC name to mount (default: gco-shared-storage for EFS,
736 gco-fsx-storage for FSx)
738 Returns:
739 Dict with download status and details
740 """
741 import os
742 import subprocess
743 import time
744 import uuid
746 if storage_type not in ("efs", "fsx"):
747 raise ValueError("storage_type must be 'efs' or 'fsx'")
748 namespace = _validated_kubernetes_name(namespace, "namespace", allow_subdomains=False)
749 local_path = _validated_local_destination(local_path)
751 # Determine PVC name based on storage type
752 if pvc_name is None:
753 pvc_name = "gco-shared-storage" if storage_type == "efs" else "gco-fsx-storage"
754 pvc_name = _validated_kubernetes_name(pvc_name, "pvc_name", allow_subdomains=True)
756 # Determine mount path based on storage type
757 mount_path = "/efs" if storage_type == "efs" else "/fsx"
758 storage_sub_path = _storage_sub_path(remote_path)
759 full_remote_path = _storage_remote_path(mount_path, remote_path)
760 _create_local_destination_parent(local_path)
762 # Generate unique pod name
763 helper_pod_name = f"gco-download-helper-{uuid.uuid4().hex[:8]}"
765 # Update kubeconfig for the cluster
766 cluster_name = f"{self.config.project_name}-{region}"
767 update_kubeconfig(cluster_name, region)
769 # Create helper pod manifest.
770 #
771 # Explicit ``resources`` block avoids the LimitRange admission plugin
772 # substituting ``max.nvidia.com/gpu`` as an implicit request — see the
773 # ``ls`` helper above for the full rationale.
774 pod_manifest = _helper_pod_manifest(
775 pod_name=helper_pod_name,
776 namespace=namespace,
777 pvc_name=pvc_name,
778 mount_path=full_remote_path,
779 app_label="gco-download-helper",
780 storage_sub_path=storage_sub_path,
781 )
783 try:
784 # Create the helper pod
785 subprocess.run(
786 ["kubectl", "apply", "-f", "-"],
787 input=pod_manifest,
788 capture_output=True,
789 text=True,
790 check=True,
791 )
793 # Wait for pod to be ready
794 max_wait = 60
795 waited = 0
796 while waited < max_wait:
797 status_result = subprocess.run(
798 [
799 "kubectl",
800 "get",
801 "pod",
802 helper_pod_name,
803 "-n",
804 namespace,
805 "-o",
806 "jsonpath={.status.phase}",
807 ],
808 capture_output=True,
809 text=True,
810 )
811 if status_result.stdout.strip() == "Running":
812 break
813 time.sleep(2) # nosemgrep: arbitrary-sleep
814 waited += 2
816 if waited >= max_wait:
817 raise RuntimeError("Helper pod did not become ready in time")
819 # Copy files from the helper pod
820 source = f"{namespace}/{helper_pod_name}:{full_remote_path}"
821 cmd = ["kubectl", "cp", "--", source, local_path]
823 subprocess.run(
824 cmd, check=True, capture_output=True, text=True
825 ) # 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
827 # Get file info
828 if os.path.isfile(local_path):
829 size = os.path.getsize(local_path)
830 elif os.path.isdir(local_path):
831 size = sum(
832 os.path.getsize(os.path.join(dirpath, filename))
833 for dirpath, _, filenames in os.walk(local_path)
834 for filename in filenames
835 )
836 else:
837 raise RuntimeError(
838 "kubectl cp reported success but did not create the local destination: "
839 f"{local_path}"
840 )
842 return {
843 "status": "success",
844 "source": f"{storage_type}:{remote_path}",
845 "destination": local_path,
846 "size_bytes": size,
847 "storage_type": storage_type,
848 "message": "Download completed successfully",
849 }
851 except subprocess.CalledProcessError as e:
852 error_msg = e.stderr if e.stderr else str(e)
853 raise RuntimeError(f"Download failed: {error_msg}") from e
854 except FileNotFoundError as e:
855 raise RuntimeError(
856 "kubectl not found. Please install kubectl and ensure it's in your PATH."
857 ) from e
858 finally:
859 # Always clean up the helper pod
860 import contextlib
862 with contextlib.suppress(Exception):
863 subprocess.run(
864 [
865 "kubectl",
866 "delete",
867 "pod",
868 helper_pod_name,
869 "-n",
870 namespace,
871 "--ignore-not-found",
872 ],
873 capture_output=True,
874 text=True,
875 )
878def get_file_system_client(config: GCOConfig | None = None) -> FileSystemClient:
879 """Get a configured file system client instance."""
880 return FileSystemClient(config)