Spaces:
Runtime error
Runtime error
File size: 4,193 Bytes
105b369 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
from pathlib import Path
from typing import List, Optional, Any, Dict
from pydantic import BaseModel, Field, ConfigDict
from phi.utils.log import logger
class KubeconfigClusterConfig(BaseModel):
server: str
certificate_authority_data: str = Field(..., alias="certificate-authority-data")
model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)
class KubeconfigCluster(BaseModel):
name: str
cluster: KubeconfigClusterConfig
class KubeconfigUser(BaseModel):
name: str
user: Dict[str, Any]
class KubeconfigContextSpec(BaseModel):
"""Each Kubeconfig context is made of (cluster, user, namespace).
It should be read as:
Use the credentials of the "user"
to access the "namespace"
of the "cluster"
"""
cluster: Optional[str]
user: Optional[str]
namespace: Optional[str] = None
class KubeconfigContext(BaseModel):
"""A context element in a kubeconfig file is used to group access parameters under a
convenient name. Each context has three parameters: cluster, namespace, and user.
By default, the kubectl command-line tool uses parameters from the current context
to communicate with the cluster.
"""
name: str
context: KubeconfigContextSpec
class Kubeconfig(BaseModel):
"""
We configure access to K8s clusters using a Kubeconfig.
This configuration can be stored in a file or an object.
A Kubeconfig stores information about clusters, users, namespaces, and authentication mechanisms,
Locally the kubeconfig file is usually stored at ~/.kube/config
View your local kubeconfig using `kubectl config view`
References:
* Docs:
https://kubernetes.io/docs/tasks/access-application-cluster/configure-access-multiple-clusters/
* Go Doc: https://godoc.org/k8s.io/client-go/tools/clientcmd/api#Config
"""
api_version: str = Field("v1", alias="apiVersion")
kind: str = "Config"
clusters: List[KubeconfigCluster] = []
users: List[KubeconfigUser] = []
contexts: List[KubeconfigContext] = []
current_context: Optional[str] = Field(None, alias="current-context")
preferences: dict = {}
model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)
@classmethod
def read_from_file(cls: Any, file_path: Path, create_if_not_exists: bool = True) -> Optional[Any]:
if file_path is not None:
if not file_path.exists():
if create_if_not_exists:
logger.info(f"Creating: {file_path}")
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.touch()
else:
logger.error(f"File does not exist: {file_path}")
return None
if file_path.exists() and file_path.is_file():
try:
import yaml
logger.info(f"Reading {file_path}")
kubeconfig_dict = yaml.safe_load(file_path.read_text())
if kubeconfig_dict is not None and isinstance(kubeconfig_dict, dict):
kubeconfig = cls(**kubeconfig_dict)
return kubeconfig
except Exception as e:
logger.error(f"Error reading {file_path}")
logger.error(e)
else:
logger.warning(f"Kubeconfig invalid: {file_path}")
return None
def write_to_file(self, file_path: Path) -> bool:
"""Writes the kubeconfig to file_path"""
if file_path is not None:
try:
import yaml
kubeconfig_dict = self.model_dump(exclude_none=True, by_alias=True)
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(yaml.safe_dump(kubeconfig_dict))
logger.info(f"Updated: {file_path}")
return True
except Exception as e:
logger.error(f"Error writing {file_path}")
logger.error(e)
else:
logger.error(f"Kubeconfig invalid: {file_path}")
return False
|