1import logging
2import yaml
3import boto3
4from sagemaker.hyperpod.common.utils import create_boto3_client
5from typing import List, Optional, ClassVar, Dict, Set, Any
6from pydantic import BaseModel, Field, ConfigDict, model_validator
7from kubernetes import client, config
8from kubernetes.client.rest import ApiException
9from kr8s.objects import Pod
10
11from sagemaker.hyperpod.common.config.metadata import Metadata
12from hyperpod_space_template.v1_0.model import ResourceRequirements
13from sagemaker.hyperpod.common.utils import (
14 handle_exception,
15 get_default_namespace,
16 setup_logging,
17 verify_kubernetes_version_compatibility,
18)
19from sagemaker.hyperpod.space.utils import (
20 map_kubernetes_response_to_model,
21 validate_space_mig_resources,
22 validate_mig_profile_in_cluster,
23)
24from sagemaker.hyperpod.common.telemetry.telemetry_logging import (
25 _hyperpod_telemetry_emitter,
26)
27from sagemaker.hyperpod.common.telemetry.constants import Feature
28from sagemaker.hyperpod.cli.constants.space_constants import (
29 SPACE_GROUP,
30 SPACE_VERSION,
31 SPACE_PLURAL,
32 DEFAULT_SPACE_PORT,
33)
34from sagemaker.hyperpod.cli.constants.space_access_constants import (
35 SPACE_ACCESS_GROUP,
36 SPACE_ACCESS_VERSION,
37 SPACE_ACCESS_PLURAL,
38)
39from hyperpod_space_template.v1_0.model import SpaceConfig, ResourceRequirements
40
41
[docs]
42class HPSpace(BaseModel):
43 """HyperPod Space on Amazon SageMaker HyperPod clusters.
44
45 This class provides methods to create, manage, and monitor spaces
46 on SageMaker HyperPod clusters orchestrated by Amazon EKS. Spaces are
47 interactive workspaces that provide development environments with
48 configurable resources, storage, and access controls.
49
50 **Attributes:**
51
52 .. list-table::
53 :header-rows: 1
54 :widths: 20 20 60
55
56 * - Attribute
57 - Type
58 - Description
59 * - config
60 - SpaceConfig
61 - The space configuration using the space parameter model
62 * - raw_resource
63 - Dict[str, Any], optional
64 - The complete Kubernetes resource data including apiVersion, kind, metadata, and status
65
66 .. dropdown:: Usage Examples
67 :open:
68
69 .. code-block:: python
70
71 >>> # Create a new space
72 >>> from hyperpod_space_template.v1_0.model import SpaceConfig
73 >>> config = SpaceConfig(name="my-space", display_name="My Space")
74 >>> space = HPSpace(config=config)
75 >>> space.create()
76
77 >>> # List all spaces
78 >>> spaces = HPSpace.list()
79 >>> for space in spaces:
80 ... print(f"Space: {space.config.name}")
81 """
82
83 is_kubeconfig_loaded: ClassVar[bool] = False
84 model_config = ConfigDict(extra="forbid")
85
86 config: SpaceConfig = Field(
87 description="The space configuration using the space parameter model"
88 )
89
90 raw_resource: Optional[Dict[str, Any]] = Field(
91 default=None,
92 description="The complete Kubernetes resource data including apiVersion, kind, metadata, and status"
93 )
94
95 @classmethod
96 def get_logger(cls):
97 """Get logger for the HPSpace class.
98
99 **Returns:**
100
101 logging.Logger: Logger instance configured for the HPSpace class
102
103 .. dropdown:: Usage Examples
104 :open:
105
106 .. code-block:: python
107
108 >>> logger = HPSpace.get_logger()
109 >>> logger.info("Space operation completed")
110 """
111 return logging.getLogger(__name__)
112
113 @property
114 def api_version(self) -> Optional[str]:
115 """Get the apiVersion from the Kubernetes resource.
116
117 **Returns:**
118
119 str or None: The API version of the Kubernetes resource, or None if raw_resource is not available
120
121 .. dropdown:: Usage Examples
122 :open:
123
124 .. code-block:: python
125
126 >>> space = HPSpace.get("my-space")
127 >>> print(f"API Version: {space.api_version}")
128 """
129 return self.raw_resource.get("apiVersion") if self.raw_resource else None
130
131 @property
132 def kind(self) -> Optional[str]:
133 """Get the kind from the Kubernetes resource.
134
135 **Returns:**
136
137 str or None: The kind of the Kubernetes resource, or None if raw_resource is not available
138
139 .. dropdown:: Usage Examples
140 :open:
141
142 .. code-block:: python
143
144 >>> space = HPSpace.get("my-space")
145 >>> print(f"Resource Kind: {space.kind}")
146 """
147 return self.raw_resource.get("kind") if self.raw_resource else None
148
149 @property
150 def metadata(self) -> Optional[Dict[str, Any]]:
151 """Get the metadata from the Kubernetes resource.
152
153 **Returns:**
154
155 Dict[str, Any] or None: The metadata section of the Kubernetes resource, or None if raw_resource is not available
156
157 .. dropdown:: Usage Examples
158 :open:
159
160 .. code-block:: python
161
162 >>> space = HPSpace.get("my-space")
163 >>> print(f"Creation Time: {space.metadata['creationTimestamp']}")
164 """
165 return self.raw_resource.get("metadata") if self.raw_resource else None
166
167 @property
168 def status(self) -> Optional[Dict[str, Any]]:
169 """Get the status from the Kubernetes resource.
170
171 **Returns:**
172
173 Dict[str, Any] or None: The status section of the Kubernetes resource, or None if raw_resource is not available
174
175 .. dropdown:: Usage Examples
176 :open:
177
178 .. code-block:: python
179
180 >>> space = HPSpace.get("my-space")
181 >>> conditions = space.status.get('conditions', [])
182 >>> for condition in conditions:
183 ... print(f"{condition['type']}: {condition['status']}")
184 """
185 return self.raw_resource.get("status") if self.raw_resource else None
186
187 @classmethod
188 def verify_kube_config(cls):
189 """Verify and load Kubernetes configuration.
190
191 Loads the Kubernetes configuration from the default kubeconfig location
192 and verifies compatibility with the cluster. This method is called
193 automatically by other methods that interact with the Kubernetes API.
194
195 **Raises:**
196
197 RuntimeError: If the kubeconfig cannot be loaded or is invalid
198
199 .. dropdown:: Usage Examples
200 :open:
201
202 .. code-block:: python
203
204 >>> # Verify kubeconfig before operations
205 >>> HPSpace.verify_kube_config()
206 """
207 if not cls.is_kubeconfig_loaded:
208 try:
209 config.load_kube_config()
210 cls.is_kubeconfig_loaded = True
211 verify_kubernetes_version_compatibility(cls.get_logger())
212 except Exception as e:
213 raise RuntimeError(f"Failed to load kubeconfig: {e}")
214
215 @staticmethod
216 def _extract_mig_profiles(resources: Optional[ResourceRequirements]) -> Set[str]:
217 """Extract MIG profile resource keys from resources without validation.
218
219 **Parameters:**
220
221 .. list-table::
222 :header-rows: 1
223 :widths: 20 20 60
224
225 * - Parameter
226 - Type
227 - Description
228 * - resources
229 - ResourceRequirements or None
230 - The resource requirements to extract MIG profiles from
231
232 **Returns:**
233
234 set: Set of MIG profile resource keys found in the resources
235 """
236 if not resources:
237 return set()
238
239 mig_profiles = set()
240
241 if resources.requests:
242 mig_profiles.update([
243 key for key in resources.requests.keys()
244 if key.startswith("nvidia.com/mig-")
245 ])
246
247 if resources.limits:
248 mig_profiles.update([
249 key for key in resources.limits.keys()
250 if key.startswith("nvidia.com/mig-")
251 ])
252
253 return mig_profiles
254
255 def _validate_and_extract_mig_profiles(self, resources: Optional[ResourceRequirements]) -> Set[str]:
256 """Validate MIG resources and extract MIG profiles.
257
258 **Parameters:**
259
260 .. list-table::
261 :header-rows: 1
262 :widths: 20 20 60
263
264 * - Parameter
265 - Type
266 - Description
267 * - resources
268 - ResourceRequirements or None
269 - The resource requirements to validate
270
271 **Returns:**
272
273 set: Set of MIG profile resource keys found in the resources
274
275 **Raises:**
276
277 RuntimeError: If MIG validation fails or profiles are invalid
278 """
279 if not resources:
280 return set()
281
282 # Validate requests
283 if resources.requests:
284 valid, err = validate_space_mig_resources(resources.requests)
285 if not valid:
286 raise RuntimeError(err)
287
288 # Validate limits
289 if resources.limits:
290 valid, err = validate_space_mig_resources(resources.limits)
291 if not valid:
292 raise RuntimeError(err)
293
294 # Extract MIG profiles
295 mig_profiles = self._extract_mig_profiles(resources)
296
297 # Validate that requests and limits use the same MIG profile
298 if len(mig_profiles) > 1:
299 raise RuntimeError(
300 "MIG profile mismatch: requests and limits must use the same MIG profile. "
301 f"Found: {', '.join(mig_profiles)}"
302 )
303
304 # Validate MIG profile exists in cluster
305 if mig_profiles:
306 mig_profile = list(mig_profiles)[0]
307 valid, err = validate_mig_profile_in_cluster(mig_profile)
308 if not valid:
309 raise RuntimeError(err)
310
311 return mig_profiles
312
[docs]
313 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "create_space")
314 def create(self, debug: bool = False):
315 """Create and submit the HyperPod Space to the Kubernetes cluster.
316
317 Creates a new space resource in the Kubernetes cluster based on the
318 configuration provided in the space config. Validates MIG profiles
319 if enabled and converts the configuration to the appropriate domain model.
320
321 **Parameters:**
322
323 .. list-table::
324 :header-rows: 1
325 :widths: 20 20 60
326
327 * - Parameter
328 - Type
329 - Description
330 * - debug
331 - bool, optional
332 - Enable debug logging (default: False)
333
334 **Raises:**
335
336 RuntimeError: If MIG profile validation fails or unsupported profiles are used
337 Exception: If the space creation fails or Kubernetes API call fails
338
339 .. dropdown:: Usage Examples
340 :open:
341
342 .. code-block:: python
343
344 >>> # Create a space with debug logging
345 >>> space = HPSpace(config=space_config)
346 >>> space.create(debug=True)
347
348 >>> # Create a space with default settings
349 >>> space.create()
350 """
351 self.verify_kube_config()
352
353 logger = self.get_logger()
354 logger = setup_logging(logger, debug)
355
356 # Validate and extract MIG profiles
357 self._validate_and_extract_mig_profiles(self.config.resources)
358
359 # Convert config to domain model
360 domain_config = self.config.to_domain()
361 config_body = domain_config["space_spec"]
362
363 logger.debug(
364 "Creating HyperPod Space with config:\n%s",
365 yaml.dump(config_body),
366 )
367
368 custom_api = client.CustomObjectsApi()
369
370 try:
371 custom_api.create_namespaced_custom_object(
372 group=SPACE_GROUP,
373 version=SPACE_VERSION,
374 namespace=self.config.namespace,
375 plural=SPACE_PLURAL,
376 body=config_body,
377 )
378 logger.debug(f"Successfully created HyperPod Space '{self.config.name}'!")
379 except Exception as e:
380 logger.error(f"Failed to create HyperPod Space {self.config.name}!")
381 handle_exception(e, self.config.name, self.config.namespace, debug=debug)
382
[docs]
383 @classmethod
384 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "list_spaces")
385 def list(cls, namespace: Optional[str] = None) -> List["HPSpace"]:
386 """List all HyperPod Spaces in the specified namespace created by the caller.
387
388 Retrieves all spaces that were either created by the current caller (based on
389 AWS STS identity) or are marked as 'Public' ownership type. Uses pagination
390 to handle large numbers of spaces efficiently.
391
392 **Parameters:**
393
394 .. list-table::
395 :header-rows: 1
396 :widths: 20 20 60
397
398 * - Parameter
399 - Type
400 - Description
401 * - namespace
402 - str, optional
403 - The Kubernetes namespace to list spaces from. If None, uses the default namespace from current context
404
405 **Returns:**
406
407 List[HPSpace]: List of HPSpace instances created by the caller or marked as public
408
409 **Raises:**
410
411 Exception: If the Kubernetes API call fails or spaces cannot be retrieved
412
413 .. dropdown:: Usage Examples
414 :open:
415
416 .. code-block:: python
417
418 >>> # List spaces in default namespace
419 >>> spaces = HPSpace.list()
420 >>> print(f"Found {len(spaces)} spaces")
421
422 >>> # List spaces in specific namespace
423 >>> spaces = HPSpace.list(namespace="my-namespace")
424 >>> for space in spaces:
425 ... print(f"Space: {space.config.name}")
426 """
427 cls.verify_kube_config()
428
429 if not namespace:
430 namespace = get_default_namespace()
431
432 # Get caller identity
433 sts_client = create_boto3_client('sts')
434 caller_identity = sts_client.get_caller_identity()
435 caller_arn = caller_identity['Arn']
436
437 custom_api = client.CustomObjectsApi()
438 spaces = []
439 continue_token = None
440
441 try:
442 while True:
443 response = custom_api.list_namespaced_custom_object(
444 group=SPACE_GROUP,
445 version=SPACE_VERSION,
446 namespace=namespace,
447 plural=SPACE_PLURAL,
448 _continue=continue_token
449 )
450
451 for item in response.get("items", []):
452 # Check if space was created by the caller or it's set as 'Public'
453 created_by = item.get('metadata', {}).get('annotations', {}).get('workspace.jupyter.org/created-by')
454 ownership_type = item.get('spec', {}).get('ownershipType', '')
455 if created_by == caller_arn or ownership_type == "Public":
456 config_data = map_kubernetes_response_to_model(item, SpaceConfig)
457 space_config = SpaceConfig(**config_data)
458
459 space = cls(
460 config=space_config,
461 raw_resource=item
462 )
463 spaces.append(space)
464
465 # Check if there are more pages
466 continue_token = response.get('metadata', {}).get('continue')
467 if not continue_token:
468 break
469
470 return spaces
471 except Exception as e:
472 handle_exception(e, "list", namespace)
473
[docs]
474 @classmethod
475 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "get_space")
476 def get(cls, name: str, namespace: str = None) -> "HPSpace":
477 """Get a specific HyperPod Space by name.
478
479 Retrieves a single space resource from the Kubernetes cluster and maps
480 the response to the SpaceConfig model for easy access to configuration
481 and status information.
482
483 **Parameters:**
484
485 .. list-table::
486 :header-rows: 1
487 :widths: 20 20 60
488
489 * - Parameter
490 - Type
491 - Description
492 * - name
493 - str
494 - The name of the space to retrieve
495 * - namespace
496 - str, optional
497 - The Kubernetes namespace. If None, uses the default namespace from current context
498
499 **Returns:**
500
501 HPSpace: The space instance with configuration and raw Kubernetes resource data
502
503 **Raises:**
504
505 Exception: If the space is not found or Kubernetes API call fails
506
507 .. dropdown:: Usage Examples
508 :open:
509
510 .. code-block:: python
511
512 >>> # Get space from default namespace
513 >>> space = HPSpace.get("my-space")
514 >>> print(f"Space status: {space.status}")
515
516 >>> # Get space from specific namespace
517 >>> space = HPSpace.get("my-space", namespace="production")
518 >>> print(f"Display name: {space.config.display_name}")
519 """
520 cls.verify_kube_config()
521
522 if not namespace:
523 namespace = get_default_namespace()
524
525 custom_api = client.CustomObjectsApi()
526
527 try:
528 response = custom_api.get_namespaced_custom_object(
529 group=SPACE_GROUP,
530 version=SPACE_VERSION,
531 namespace=namespace,
532 plural=SPACE_PLURAL,
533 name=name
534 )
535
536 # Use dynamic mapping based on SpaceConfig model
537 config_data = map_kubernetes_response_to_model(response, SpaceConfig)
538
539 space_config = SpaceConfig(**config_data)
540
541 return cls(
542 config=space_config,
543 raw_resource=response
544 )
545 except Exception as e:
546 handle_exception(e, name, namespace)
547
[docs]
548 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "delete_space")
549 def delete(self):
550 """Delete the HyperPod Space from the Kubernetes cluster.
551
552 Permanently removes the space resource from the Kubernetes cluster.
553 This operation cannot be undone and will terminate any running
554 workloads associated with the space.
555
556 **Raises:**
557
558 Exception: If the deletion fails or Kubernetes API call fails
559
560 .. dropdown:: Usage Examples
561 :open:
562
563 .. code-block:: python
564
565 >>> # Delete a space
566 >>> space = HPSpace.get("my-space")
567 >>> space.delete()
568 """
569 self.verify_kube_config()
570 logger = self.get_logger()
571
572 custom_api = client.CustomObjectsApi()
573
574 try:
575 custom_api.delete_namespaced_custom_object(
576 group=SPACE_GROUP,
577 version=SPACE_VERSION,
578 namespace=self.config.namespace,
579 plural=SPACE_PLURAL,
580 name=self.config.name
581 )
582 logger.debug(f"Successfully deleted HyperPod Space '{self.config.name}'!")
583 except Exception as e:
584 logger.error(f"Failed to delete HyperPod Space {self.config.name}!")
585 handle_exception(e, self.config.name, self.config.namespace)
586
[docs]
587 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "update_space")
588 def update(self, **kwargs):
589 """Update the HyperPod Space configuration.
590
591 Updates the space configuration with the provided parameters. Validates
592 MIG profiles if resource updates are requested and ensures compatibility
593 with the current node instance type.
594
595 **Parameters:**
596
597 .. list-table::
598 :header-rows: 1
599 :widths: 20 20 60
600
601 * - Parameter
602 - Type
603 - Description
604 * - **kwargs
605 - Any
606 - Configuration fields to update (e.g., desired_status="Stopped", display_name="New Name")
607
608 **Raises:**
609
610 RuntimeError: If MIG profile validation fails or unsupported profiles are used
611 Exception: If the update fails or Kubernetes API call fails
612
613 .. dropdown:: Usage Examples
614 :open:
615
616 .. code-block:: python
617
618 >>> # Update space status
619 >>> space = HPSpace.get("my-space")
620 >>> space.update(desired_status="Stopped")
621
622 >>> # Update display name and resources
623 >>> space.update(
624 ... display_name="Updated Space",
625 ... resources={"requests": {"cpu": "2", "memory": "4Gi"}}
626 ... )
627 """
628 self.verify_kube_config()
629 logger = self.get_logger()
630
631 # Validate MIG profile configuration
632 if "resources" in kwargs:
633 resources = kwargs["resources"]
634
635 if isinstance(resources, dict):
636 resources = ResourceRequirements(**resources)
637
638 # Validate and extract MIG profiles
639 mig_profiles = self._validate_and_extract_mig_profiles(resources)
640
641 # Remove existing MIG profiles if changing to a different one
642 if mig_profiles:
643 mig_profile = list(mig_profiles)[0]
644
645 existing_config = HPSpace.get(self.config.name, self.config.namespace).config
646 existing_mig_profiles = self._extract_mig_profiles(existing_config.resources)
647
648 if existing_mig_profiles and mig_profile not in existing_mig_profiles:
649 # Remove existing MIG profiles by setting to None
650 for existing_profile in existing_mig_profiles:
651 if existing_profile != mig_profile:
652 kwargs["resources"].setdefault("requests", {})[existing_profile] = None
653 kwargs["resources"].setdefault("limits", {})[existing_profile] = None
654
655 custom_api = client.CustomObjectsApi()
656
657 # Update space config with the input config
658 current_config = self.config.model_dump(by_alias=True)
659 current_config.update(kwargs)
660 self.config = SpaceConfig(**current_config)
661
662 # Convert to domain model and extract spec
663 domain_config = self.config.to_domain()
664 spec_updates = domain_config["space_spec"]["spec"]
665
666 try:
667 custom_api.patch_namespaced_custom_object(
668 group=SPACE_GROUP,
669 version=SPACE_VERSION,
670 namespace=self.config.namespace,
671 plural=SPACE_PLURAL,
672 name=self.config.name,
673 body={"spec": spec_updates}
674 )
675 logger.debug(f"Successfully updated HyperPod Space '{self.config.name}'!")
676 except Exception as e:
677 logger.error(f"Failed to update HyperPod Space {self.config.name}!")
678 handle_exception(e, self.config.name, self.config.namespace)
679
[docs]
680 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "start_space")
681 def start(self):
682 """Start the HyperPod Space by setting desired status to Running.
683
684 Convenience method that updates the space's desired status to "Running",
685 which will cause the Kubernetes operator to start the space workloads.
686
687 .. dropdown:: Usage Examples
688 :open:
689
690 .. code-block:: python
691
692 >>> # Start a space
693 >>> space = HPSpace.get("my-space")
694 >>> space.start()
695 """
696 self.update(desired_status="Running")
697
[docs]
698 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "stop_space")
699 def stop(self):
700 """Stop the HyperPod Space by setting desired status to Stopped.
701
702 Convenience method that updates the space's desired status to "Stopped",
703 which will cause the Kubernetes operator to stop the space workloads.
704
705 .. dropdown:: Usage Examples
706 :open:
707
708 .. code-block:: python
709
710 >>> # Stop a space
711 >>> space = HPSpace.get("my-space")
712 >>> space.stop()
713 """
714 self.update(desired_status="Stopped")
715
[docs]
716 def list_pods(self) -> List[str]:
717 """List all pods associated with this space.
718
719 Retrieves all Kubernetes pods that are labeled as belonging to this
720 space using the workspace-name label selector.
721
722 **Returns:**
723
724 List[str]: List of pod names associated with the space
725
726 **Raises:**
727
728 Exception: If the Kubernetes API call fails
729
730 .. dropdown:: Usage Examples
731 :open:
732
733 .. code-block:: python
734
735 >>> # List pods for a space
736 >>> space = HPSpace.get("my-space")
737 >>> pods = space.list_pods()
738 >>> print(f"Found {len(pods)} pods: {pods}")
739 """
740 self.verify_kube_config()
741 logger = self.get_logger()
742
743 v1 = client.CoreV1Api()
744
745 try:
746 pods = v1.list_namespaced_pod(
747 namespace=self.config.namespace,
748 label_selector=f"{SPACE_GROUP}/workspace-name={self.config.name}"
749 )
750 return [pod.metadata.name for pod in pods.items]
751 except Exception as e:
752 handle_exception(e, self.config.name, self.config.namespace)
753
[docs]
754 def get_logs(self, pod_name: Optional[str] = None, container: Optional[str] = None) -> str:
755 """Get logs from a pod associated with this space.
756
757 Retrieves logs from a specific pod and container. If no pod is specified,
758 uses the first available pod. If no container is specified, defaults to
759 the "workspace" container.
760
761 **Parameters:**
762
763 .. list-table::
764 :header-rows: 1
765 :widths: 20 20 60
766
767 * - Parameter
768 - Type
769 - Description
770 * - pod_name
771 - str, optional
772 - Name of the pod to get logs from. If None, gets logs from the first available pod
773 * - container
774 - str, optional
775 - Name of the container to get logs from. Defaults to "workspace"
776
777 **Returns:**
778
779 str: The pod logs as a string
780
781 **Raises:**
782
783 RuntimeError: If no pods are found for the space
784 Exception: If the Kubernetes API call fails
785
786 .. dropdown:: Usage Examples
787 :open:
788
789 .. code-block:: python
790
791 >>> # Get logs from default pod and container
792 >>> space = HPSpace.get("my-space")
793 >>> logs = space.get_logs()
794 >>> print(logs)
795
796 >>> # Get logs from specific pod and container
797 >>> logs = space.get_logs(pod_name="my-pod", container="sidecar")
798 """
799 self.verify_kube_config()
800 logger = self.get_logger()
801
802 if not pod_name:
803 pods = self.list_pods()
804 if not pods:
805 raise RuntimeError(f"No pods found for space '{self.config.name}'")
806 pod_name = pods[0]
807
808 if not container:
809 container = "workspace"
810
811 v1 = client.CoreV1Api()
812
813 try:
814 return v1.read_namespaced_pod_log(
815 name=pod_name,
816 namespace=self.config.namespace,
817 container=container
818 )
819 except Exception as e:
820 handle_exception(e, pod_name, self.config.namespace)
821
[docs]
822 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "create_space_access")
823 def create_space_access(self, connection_type: str = "vscode-remote") -> Dict[str, str]:
824 """Create a space access for this space.
825
826 Creates a space access resource that provides remote connection capabilities
827 to the space. Supports VS Code remote development and web UI access types.
828
829 **Parameters:**
830
831 .. list-table::
832 :header-rows: 1
833 :widths: 20 20 60
834
835 * - Parameter
836 - Type
837 - Description
838 * - connection_type
839 - str, optional
840 - The IDE type for remote access. Must be "vscode-remote" or "web-ui" (default: "vscode-remote")
841
842 **Returns:**
843
844 Dict[str, str]: Dictionary containing 'SpaceConnectionType' and 'SpaceConnectionUrl' keys
845
846 **Raises:**
847
848 ValueError: If connection_type is not "vscode-remote" or "web-ui"
849 Exception: If the space access creation fails or Kubernetes API call fails
850
851 .. dropdown:: Usage Examples
852 :open:
853
854 .. code-block:: python
855
856 >>> # Create VS Code remote access
857 >>> space = HPSpace.get("my-space")
858 >>> access = space.create_space_access("vscode-remote")
859 >>> print(f"Connection URL: {access['SpaceConnectionUrl']}")
860
861 >>> # Create web UI access
862 >>> access = space.create_space_access("web-ui")
863 >>> print(f"Web UI URL: {access['SpaceConnectionUrl']}")
864 """
865 self.verify_kube_config()
866 logger = self.get_logger()
867
868 if connection_type not in {"vscode-remote", "web-ui"}:
869 raise ValueError("--connection-type must be 'vscode-remote' or 'web-ui'.")
870
871 config = {
872 "metadata": {
873 "namespace": self.config.namespace,
874 },
875 "spec": {
876 "workspaceName": self.config.name,
877 "workspaceConnectionType": connection_type,
878 }
879 }
880
881 custom_api = client.CustomObjectsApi()
882
883 try:
884 response = custom_api.create_namespaced_custom_object(
885 group=SPACE_ACCESS_GROUP,
886 version=SPACE_ACCESS_VERSION,
887 namespace=self.config.namespace,
888 plural=SPACE_ACCESS_PLURAL,
889 body=config
890 )
891 logger.debug(f"Successfully created space access for '{self.config.name}'!")
892 return {
893 "SpaceConnectionType": connection_type,
894 "SpaceConnectionUrl": response["status"]["workspaceConnectionUrl"]
895 }
896 except Exception as e:
897 logger.error(f"Failed to create space access for {self.config.name}!")
898 handle_exception(e, self.config.name, self.config.namespace)
899
[docs]
900 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "portforward_space")
901 def portforward_space(self, local_port: str, remote_port: str = DEFAULT_SPACE_PORT):
902 """Forward local port to the space pod for development access.
903
904 Creates a port forwarding connection from a local port to a remote port
905 on the space pod, enabling direct access to services running inside the
906 space.
907
908 **Parameters:**
909
910 .. list-table::
911 :header-rows: 1
912 :widths: 20 20 60
913
914 * - Parameter
915 - Type
916 - Description
917 * - local_port
918 - str
919 - The local port to forward from
920 * - remote_port
921 - str, optional
922 - The remote port on the space pod to forward to (default: DEFAULT_SPACE_PORT)
923
924 **Raises:**
925
926 RuntimeError: If no pods are found for the space or if the space is not in Available status
927 KeyboardInterrupt: When the user stops the port forwarding with Ctrl+C
928 Exception: If the port forwarding setup fails or Kubernetes API call fails
929
930 .. dropdown:: Usage Examples
931 :open:
932
933 .. code-block:: python
934
935 >>> # Forward local port 8080 to default remote port
936 >>> space = HPSpace.get("myspace")
937 >>> space.portforward_space("8080")
938
939 >>> # Forward local port 3000 to remote port 8888
940 >>> space.portforward_space("3000", "8888")
941
942 >>> # Access forwarded service (in another terminal)
943 >>> # curl http://localhost:8080
944 """
945
946 self.verify_kube_config()
947 logger = self.get_logger()
948
949 # Check if space is in Available status
950 if self.status and self.status.get("conditions"):
951 is_available = False
952 for condition in self.status["conditions"]:
953 if condition.get("type") == "Available" and condition.get("status") == "True":
954 is_available = True
955 break
956
957 if not is_available:
958 raise RuntimeError(f"Space '{self.config.name}' is not in Available status. Port forwarding is only allowed for available spaces.")
959
960 pods = self.list_pods()
961 if not pods:
962 raise RuntimeError(f"No pods found for space '{self.config.name}'")
963
964 pod_name = pods[0]
965 pod = Pod.get(name=pod_name, namespace=self.config.namespace)
966 pf = pod.portforward(remote_port=int(remote_port), local_port=int(local_port))
967
968 logger.debug(f"Forwarding from local port {local_port} to space pod: {pod_name}.")
969
970 try:
971 pf.run_forever()
972 except KeyboardInterrupt:
973 logger.debug("Stopping space port forward...")
974 finally:
975 pf.stop()
