|
1 | 1 | import enum |
| 2 | +import functools |
2 | 3 | import json |
3 | 4 | import logging |
4 | 5 | import os |
|
7 | 8 | import time |
8 | 9 | import typing |
9 | 10 | import yaml |
10 | | -import functools |
11 | 11 |
|
12 | 12 | from abc import ABC |
13 | 13 | from abc import abstractmethod |
@@ -56,6 +56,36 @@ class _OperationState(enum.Enum): |
56 | 56 | STOPPED = 5 |
57 | 57 |
|
58 | 58 |
|
| 59 | +@functools.cache |
| 60 | +def detect_default_resource_name_cached( |
| 61 | + client: K8sClient, |
| 62 | + namespace: str, |
| 63 | + secondary_network_nad: Optional[str], |
| 64 | +) -> Optional[str]: |
| 65 | + if secondary_network_nad is not None: |
| 66 | + if "/" in secondary_network_nad: |
| 67 | + ns, nad = secondary_network_nad.split("/", 1) |
| 68 | + else: |
| 69 | + ns, nad = namespace, secondary_network_nad |
| 70 | + data = client.oc_get( |
| 71 | + f"network-attachment-definition/{nad}", |
| 72 | + namespace=ns, |
| 73 | + ) |
| 74 | + else: |
| 75 | + data = None |
| 76 | + resource_name = None |
| 77 | + |
| 78 | + if data is not None: |
| 79 | + try: |
| 80 | + r = data["metadata"]["annotations"]["k8s.v1.cni.cncf.io/resourceName"] |
| 81 | + if isinstance(r, str) and r: |
| 82 | + resource_name = r |
| 83 | + except Exception: |
| 84 | + pass |
| 85 | + logger.info(f"autodetected resource_name as {repr(resource_name)}") |
| 86 | + return resource_name |
| 87 | + |
| 88 | + |
59 | 89 | class TaskOperation: |
60 | 90 | @typing.overload |
61 | 91 | def __init__( |
@@ -308,38 +338,10 @@ def out_file_yaml(self) -> str: |
308 | 338 | raise RuntimeError("task cannot generate a pod yaml") |
309 | 339 | return tftbase.get_manifest_renderpath(self.pod_name + ".yaml") |
310 | 340 |
|
311 | | - @functools.cache |
312 | | - @staticmethod |
313 | | - def _fetch_default_resource_name( |
314 | | - client: K8sClient, namespace: str, secondary_network_nad: Optional[str] |
315 | | - ) -> Optional[str]: |
316 | | - if secondary_network_nad is not None: |
317 | | - if "/" in secondary_network_nad: |
318 | | - ns, nad = secondary_network_nad.split("/", 1) |
319 | | - else: |
320 | | - ns, nad = namespace, secondary_network_nad |
321 | | - data = client.oc_get( |
322 | | - f"network-attachment-definition/{nad}", |
323 | | - namespace=ns, |
324 | | - ) |
325 | | - else: |
326 | | - data = None |
327 | | - resource_name = None |
328 | | - |
329 | | - if data is not None: |
330 | | - try: |
331 | | - r = data["metadata"]["annotations"]["k8s.v1.cni.cncf.io/resourceName"] |
332 | | - if isinstance(r, str) and r: |
333 | | - resource_name = r |
334 | | - except Exception: |
335 | | - pass |
336 | | - logger.info(f"autodetected resource_name as {repr(resource_name)}") |
337 | | - return resource_name |
338 | | - |
339 | 341 | def get_template_args(self) -> dict[str, str | list[str] | bool]: |
340 | 342 | resource_name = ( |
341 | 343 | self.ts.connection.resource_name |
342 | | - or Task._fetch_default_resource_name( |
| 344 | + or detect_default_resource_name_cached( |
343 | 345 | self.client, |
344 | 346 | self.get_namespace(), |
345 | 347 | self.ts.connection.secondary_network_nad, |
|
0 commit comments