1import logging
2import yaml
3from typing import List, Optional, ClassVar, Dict, Any
4from kubernetes import client, config
5from kubernetes.client.rest import ApiException
6
7from sagemaker.hyperpod.common.utils import (
8 handle_exception,
9 get_default_namespace,
10 verify_kubernetes_version_compatibility
11)
12from sagemaker.hyperpod.common.telemetry.telemetry_logging import (
13 _hyperpod_telemetry_emitter,
14)
15from sagemaker.hyperpod.common.telemetry.constants import Feature
16from sagemaker.hyperpod.cli.constants.space_template_constants import (
17 SPACE_TEMPLATE_GROUP,
18 SPACE_TEMPLATE_VERSION,
19 SPACE_TEMPLATE_PLURAL,
20)
21
22
[docs]
23class HPSpaceTemplate:
24 """HyperPod Space Template on Amazon SageMaker HyperPod clusters.
25
26 This class provides methods to create, manage, and monitor space templates
27 on SageMaker HyperPod clusters orchestrated by Amazon EKS. Space templates
28 define reusable configurations for creating spaces with predefined settings,
29 resources, and constraints.
30
31 **Attributes:**
32
33 .. list-table::
34 :header-rows: 1
35 :widths: 20 20 60
36
37 * - Attribute
38 - Type
39 - Description
40 * - config_data
41 - Dict[str, Any]
42 - Dictionary containing the complete template configuration
43 * - name
44 - str
45 - Name of the space template extracted from metadata
46 * - namespace
47 - str
48 - Kubernetes namespace of the template extracted from metadata
49
50 .. dropdown:: Usage Examples
51 :open:
52
53 .. code-block:: python
54
55 >>> # Create template from YAML file
56 >>> template = HPSpaceTemplate(file_path="template.yaml")
57 >>> template.create()
58
59 >>> # List all templates
60 >>> templates = HPSpaceTemplate.list()
61 >>> for template in templates:
62 ... print(f"Template: {template.name}")
63 """
64
65 is_kubeconfig_loaded: ClassVar[bool] = False
66
[docs]
67 def __init__(self, *, file_path: Optional[str] = None, config_data: Optional[Dict[str, Any]] = None):
68 """Initialize space template with config YAML file path or dictionary data.
69
70 Creates a new HPSpaceTemplate instance from either a YAML configuration file
71 or a dictionary containing configuration data. Exactly one of the parameters
72 must be provided.
73
74 **Parameters:**
75
76 .. list-table::
77 :header-rows: 1
78 :widths: 20 20 60
79
80 * - Parameter
81 - Type
82 - Description
83 * - file_path
84 - str, optional
85 - Path to YAML configuration file (keyword-only)
86 * - config_data
87 - Dict[str, Any], optional
88 - Dictionary containing configuration data (keyword-only)
89
90 **Raises:**
91
92 ValueError: If both or neither parameters are provided, or if YAML parsing fails
93 FileNotFoundError: If the specified file path does not exist
94
95 .. dropdown:: Usage Examples
96 :open:
97
98 .. code-block:: python
99
100 >>> # Initialize from YAML file
101 >>> template = HPSpaceTemplate(file_path="my-template.yaml")
102
103 >>> # Initialize from dictionary (e.g., from API response)
104 >>> config = {"metadata": {"name": "my-template"}, "spec": {...}}
105 >>> template = HPSpaceTemplate(config_data=config)
106 """
107 if (file_path is None) == (config_data is None):
108 raise ValueError("Exactly one of 'file_path' or 'config_data' must be provided")
109
110 if file_path is not None:
111 # Initialize from file path
112 try:
113 with open(file_path, 'r') as f:
114 self.config_data = yaml.safe_load(f)
115 except FileNotFoundError:
116 raise FileNotFoundError(f"File '{file_path}' not found")
117 except yaml.YAMLError as e:
118 raise ValueError(f"Error parsing YAML file: {e}")
119 else:
120 # Initialize from dictionary data (e.g., from Kubernetes API response)
121 self.config_data = config_data
122
123 self.name = self.config_data.get('metadata', {}).get('name')
124 self.namespace = self.config_data.get('metadata', {}).get('namespace')
125
126 @classmethod
127 def get_logger(cls):
128 """Get logger for the HPSpaceTemplate class.
129
130 **Returns:**
131
132 logging.Logger: Logger instance configured for the HPSpaceTemplate class
133
134 .. dropdown:: Usage Examples
135 :open:
136
137 .. code-block:: python
138
139 >>> logger = HPSpaceTemplate.get_logger()
140 >>> logger.info("Template operation completed")
141 """
142 return logging.getLogger(__name__)
143
144 @classmethod
145 def verify_kube_config(cls):
146 """Verify and load Kubernetes configuration.
147
148 Loads the Kubernetes configuration from the default kubeconfig location
149 and verifies compatibility with the cluster. This method is called
150 automatically by other methods that interact with the Kubernetes API.
151
152 **Raises:**
153
154 Exception: If the kubeconfig cannot be loaded or is invalid
155
156 .. dropdown:: Usage Examples
157 :open:
158
159 .. code-block:: python
160
161 >>> # Verify kubeconfig before operations
162 >>> HPSpaceTemplate.verify_kube_config()
163 """
164 if not cls.is_kubeconfig_loaded:
165 config.load_kube_config()
166 cls.is_kubeconfig_loaded = True
167 verify_kubernetes_version_compatibility(cls.get_logger())
168
[docs]
169 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "create_space_template")
170 def create(self) -> "HPSpaceTemplate":
171 """Create the space template in the Kubernetes cluster.
172
173 Submits the space template configuration to the Kubernetes cluster and
174 creates a new template resource. Updates the instance with the server
175 response including generated metadata.
176
177 **Returns:**
178
179 HPSpaceTemplate: Updated HPSpaceTemplate instance with server response data
180
181 **Raises:**
182
183 ApiException: If the Kubernetes API call fails
184 Exception: If template creation fails for other reasons
185
186 .. dropdown:: Usage Examples
187 :open:
188
189 .. code-block:: python
190
191 >>> # Create template from file
192 >>> template = HPSpaceTemplate(file_path="template.yaml")
193 >>> created_template = template.create()
194 >>> print(f"Created template: {created_template.name}")
195 """
196 self.verify_kube_config()
197
198 try:
199 api_instance = client.CustomObjectsApi()
200 response = api_instance.create_namespaced_custom_object(
201 group=SPACE_TEMPLATE_GROUP,
202 version=SPACE_TEMPLATE_VERSION,
203 namespace=self.namespace,
204 plural=SPACE_TEMPLATE_PLURAL,
205 body=self.config_data
206 )
207
208 self.config_data = response
209 self.get_logger().info(f"Space template '{self.name}' created successfully")
210
211 except ApiException as e:
212 handle_exception(e, self.name, None)
213 except Exception as e:
214 self.get_logger().error(f"Error creating space template: {e}")
215 raise
216
[docs]
217 @classmethod
218 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "list_space_templates")
219 def list(cls, namespace: Optional[str] = None) -> List["HPSpaceTemplate"]:
220 """List all space templates in the specified namespace.
221
222 Retrieves all space template resources from the Kubernetes cluster in the
223 specified namespace. If no namespace is provided, uses the default namespace
224 from the current Kubernetes context.
225
226 **Parameters:**
227
228 .. list-table::
229 :header-rows: 1
230 :widths: 20 20 60
231
232 * - Parameter
233 - Type
234 - Description
235 * - namespace
236 - str, optional
237 - The Kubernetes namespace to list space templates from. If None, uses the default namespace from current context
238
239 **Returns:**
240
241 List[HPSpaceTemplate]: List of HPSpaceTemplate instances found in the namespace
242
243 **Raises:**
244
245 ApiException: If the Kubernetes API call fails
246 Exception: If template listing fails for other reasons
247
248 .. dropdown:: Usage Examples
249 :open:
250
251 .. code-block:: python
252
253 >>> # List templates in default namespace
254 >>> templates = HPSpaceTemplate.list()
255 >>> print(f"Found {len(templates)} templates")
256
257 >>> # List templates in specific namespace
258 >>> templates = HPSpaceTemplate.list(namespace="production")
259 >>> for template in templates:
260 ... print(f"Template: {template.name}")
261 """
262 cls.verify_kube_config()
263
264 if not namespace:
265 namespace = get_default_namespace()
266
267 try:
268 api_instance = client.CustomObjectsApi()
269 response = api_instance.list_namespaced_custom_object(
270 group=SPACE_TEMPLATE_GROUP,
271 version=SPACE_TEMPLATE_VERSION,
272 namespace=namespace,
273 plural=SPACE_TEMPLATE_PLURAL
274 )
275
276 templates = []
277 for item in response.get("items", []):
278 templates.append(cls(config_data=item))
279
280 return templates
281
282 except ApiException as e:
283 handle_exception(e, "list", None)
284 except Exception as e:
285 cls.get_logger().error(f"Error listing space templates: {e}")
286 raise
287
[docs]
288 @classmethod
289 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "get_space_template")
290 def get(cls, name: str, namespace: Optional[str] = None) -> "HPSpaceTemplate":
291 """Get a specific space template by name.
292
293 Retrieves a single space template resource from the Kubernetes cluster
294 by name. Removes managedFields from the metadata for cleaner output.
295
296 **Parameters:**
297
298 .. list-table::
299 :header-rows: 1
300 :widths: 20 20 60
301
302 * - Parameter
303 - Type
304 - Description
305 * - name
306 - str
307 - Name of the space template to retrieve
308 * - namespace
309 - str, optional
310 - The Kubernetes namespace. If None, uses the default namespace from current context
311
312 **Returns:**
313
314 HPSpaceTemplate: The space template instance with configuration data
315
316 **Raises:**
317
318 ApiException: If the template is not found or Kubernetes API call fails
319 Exception: If template retrieval fails for other reasons
320
321 .. dropdown:: Usage Examples
322 :open:
323
324 .. code-block:: python
325
326 >>> # Get template from default namespace
327 >>> template = HPSpaceTemplate.get("my-template")
328 >>> print(f"Template display name: {template.config_data['spec']['displayName']}")
329
330 >>> # Get template from specific namespace
331 >>> template = HPSpaceTemplate.get("my-template", namespace="production")
332 >>> print(template.to_yaml())
333 """
334 cls.verify_kube_config()
335
336 if not namespace:
337 namespace = get_default_namespace()
338
339 try:
340 api_instance = client.CustomObjectsApi()
341 response = api_instance.get_namespaced_custom_object(
342 group=SPACE_TEMPLATE_GROUP,
343 version=SPACE_TEMPLATE_VERSION,
344 namespace=namespace,
345 plural=SPACE_TEMPLATE_PLURAL,
346 name=name
347 )
348
349 # Remove managedFields for cleaner output
350 if 'metadata' in response:
351 response['metadata'].pop('managedFields', None)
352
353 return cls(config_data=response)
354
355 except ApiException as e:
356 handle_exception(e, name, None)
357 except Exception as e:
358 cls.get_logger().error(f"Error getting space template '{name}': {e}")
359 raise
360
[docs]
361 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "delete_space_template")
362 def delete(self) -> None:
363 """Delete the space template from the Kubernetes cluster.
364
365 Permanently removes the space template resource from the Kubernetes cluster.
366 This operation cannot be undone. Any spaces created from this template
367 will continue to exist but will no longer reference the template.
368
369 **Raises:**
370
371 ApiException: If the deletion fails or Kubernetes API call fails
372 Exception: If template deletion fails for other reasons
373
374 .. dropdown:: Usage Examples
375 :open:
376
377 .. code-block:: python
378
379 >>> # Delete a template
380 >>> template = HPSpaceTemplate.get("my-template")
381 >>> template.delete()
382 """
383 self.verify_kube_config()
384
385 try:
386 api_instance = client.CustomObjectsApi()
387 api_instance.delete_namespaced_custom_object(
388 group=SPACE_TEMPLATE_GROUP,
389 version=SPACE_TEMPLATE_VERSION,
390 namespace=self.namespace,
391 plural=SPACE_TEMPLATE_PLURAL,
392 name=self.name
393 )
394
395 self.get_logger().info(f"Space template '{self.name}' deleted successfully")
396
397 except ApiException as e:
398 handle_exception(e, self.name, None)
399 except Exception as e:
400 self.get_logger().error(f"Error deleting space template '{self.name}': {e}")
401 raise
402
[docs]
403 @_hyperpod_telemetry_emitter(Feature.HYPERPOD, "update_space_template")
404 def update(self, file_path: str) -> "HPSpaceTemplate":
405 """Update the space template from a YAML configuration file.
406
407 Updates the existing space template with new configuration from a YAML file.
408 Validates that the template name in the file matches the current template name
409 and removes immutable fields before applying the update.
410
411 **Parameters:**
412
413 .. list-table::
414 :header-rows: 1
415 :widths: 20 20 60
416
417 * - Parameter
418 - Type
419 - Description
420 * - file_path
421 - str
422 - Path to the YAML configuration file containing updated template configuration
423
424 **Returns:**
425
426 HPSpaceTemplate: Updated HPSpaceTemplate instance with server response data
427
428 **Raises:**
429
430 FileNotFoundError: If the specified file path does not exist
431 ValueError: If YAML parsing fails or template name mismatch occurs
432 ApiException: If the Kubernetes API call fails
433 Exception: If template update fails for other reasons
434
435 .. dropdown:: Usage Examples
436 :open:
437
438 .. code-block:: python
439
440 >>> # Update template from file
441 >>> template = HPSpaceTemplate.get("my-template")
442 >>> updated_template = template.update("updated-template.yaml")
443 >>> print(f"Updated template: {updated_template.name}")
444 """
445 self.verify_kube_config()
446
447 try:
448 with open(file_path, 'r') as f:
449 config_data = yaml.safe_load(f)
450
451 # Validate that the name matches
452 yaml_name = config_data.get('metadata', {}).get('name')
453 if yaml_name and yaml_name != self.name:
454 raise ValueError(f"Name mismatch. Template name '{self.name}' does not match YAML name '{yaml_name}'")
455
456 # Remove immutable fields
457 if 'metadata' in config_data:
458 for field in ['resourceVersion', 'uid', 'creationTimestamp', 'managedFields']:
459 config_data['metadata'].pop(field, None)
460
461 api_instance = client.CustomObjectsApi()
462 response = api_instance.patch_namespaced_custom_object(
463 group=SPACE_TEMPLATE_GROUP,
464 version=SPACE_TEMPLATE_VERSION,
465 namespace=self.namespace,
466 plural=SPACE_TEMPLATE_PLURAL,
467 name=self.name,
468 body=config_data
469 )
470
471 self.config_data = response
472 self.get_logger().info(f"Space template '{self.name}' updated successfully")
473
474 except FileNotFoundError:
475 raise FileNotFoundError(f"File '{file_path}' not found")
476 except yaml.YAMLError as e:
477 raise ValueError(f"Error parsing YAML file: {e}")
478 except ApiException as e:
479 handle_exception(e, self.name, None)
480 except Exception as e:
481 self.get_logger().error(f"Error updating space template '{self.name}': {e}")
482 raise
483
[docs]
484 def to_yaml(self) -> str:
485 """Convert the space template to YAML format.
486
487 Serializes the template configuration data to a YAML string representation
488 with readable formatting (non-flow style).
489
490 **Returns:**
491
492 str: YAML string representation of the template configuration
493
494 .. dropdown:: Usage Examples
495 :open:
496
497 .. code-block:: python
498
499 >>> # Convert template to YAML
500 >>> template = HPSpaceTemplate.get("my-template")
501 >>> yaml_content = template.to_yaml()
502 >>> print(yaml_content)
503
504 >>> # Save template to file
505 >>> with open("exported-template.yaml", "w") as f:
506 ... f.write(template.to_yaml())
507 """
508 return yaml.dump(self.config_data, default_flow_style=False)
509
[docs]
510 def to_dict(self) -> Dict[str, Any]:
511 """Convert the space template to dictionary format.
512
513 Returns the template configuration data as a dictionary, which can be
514 used for programmatic access to template properties or serialization
515 to other formats.
516
517 **Returns:**
518
519 Dict[str, Any]: Dictionary representation of the template configuration
520
521 .. dropdown:: Usage Examples
522 :open:
523
524 .. code-block:: python
525
526 >>> # Get template as dictionary
527 >>> template = HPSpaceTemplate.get("my-template")
528 >>> config_dict = template.to_dict()
529 >>> print(f"Template spec: {config_dict['spec']}")
530
531 >>> # Access specific configuration values
532 >>> display_name = config_dict['spec']['displayName']
533 >>> default_image = config_dict['spec']['defaultImage']
534 """
535 return self.config_data
