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