Get ApiClient from predefined locations. Args: k8s_config_file(str): Path to kubernetes config file. Raises: RuntimeError: K8s api client resolve failed. Returns: An kubernetes ApiClient object, initialized with the client args. The order of resolution as
(k8s_config_file=None)
| 35 | |
| 36 | |
| 37 | def resolve_api_client(k8s_config_file=None): |
| 38 | """Get ApiClient from predefined locations. |
| 39 | |
| 40 | Args: |
| 41 | k8s_config_file(str): Path to kubernetes config file. |
| 42 | |
| 43 | Raises: |
| 44 | RuntimeError: K8s api client resolve failed. |
| 45 | |
| 46 | Returns: |
| 47 | An kubernetes ApiClient object, initialized with the client args. |
| 48 | |
| 49 | The order of resolution as follows: |
| 50 | 1. load from kubernetes config file or, |
| 51 | 2. load from incluster configuration or, |
| 52 | 3. set api address from env if `KUBE_API_ADDRESS` exist. |
| 53 | RuntimeError will be raised if resolution failed. |
| 54 | """ |
| 55 | try: |
| 56 | # load from kubernetes config file |
| 57 | kube_config.load_kube_config(k8s_config_file) |
| 58 | except: # noqa: E722 |
| 59 | try: |
| 60 | # load from incluster configuration |
| 61 | kube_config.load_incluster_config() |
| 62 | except: # noqa: E722 |
| 63 | if "KUBE_API_ADDRESS" in os.environ: |
| 64 | # try to load from env `KUBE_API_ADDRESS` |
| 65 | config = kube_client.Configuration() |
| 66 | config.host = os.environ["KUBE_API_ADDRESS"] |
| 67 | return kube_client.ApiClient(config) |
| 68 | raise RuntimeError("Resolve kube api client failed.") |
| 69 | return kube_client.ApiClient() |
| 70 | |
| 71 | |
| 72 | def parse_readable_memory(value): |
no test coverage detected