INNER CODE UNIT · Python
__init__
robusta-dev/krr · robusta_krr/core/integrations/kubernetes/__init__.py:44
def __init__(self, cluster: Optional[str] = None):
self.cluster = cluster
# This executor will be running requests to Kubernetes API
self.executor = ThreadPoolExecutor(settings.max_workers)
self.api_client = settings.get_kube_client(cluster)
self.apps = client.AppsV1Api(api_client=self.api_client)
self.custom_objects = client.CustomObjectsApi(api_client=self.api_client)
self.batch = client.BatchV1Api(api_client=self.api_client)
self.core = client.CoreV1Api(api_client=self.api_client)
self.autoscaling_v1 = client.AutoscalingV1Api(api_client=self.api_client)
self.autoscaling_v2 = client.AutoscalingV2Api(api_client=self.api_client)
self.__kind_available: defaultdict[KindLiteral, bool] = defaultdict(lambda: True)
self.__jobs_for_cronjobs: dict[str, list[V1Job]] = {}
self.__jobs_loading_locks: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
self.__namespaces: Union[list[str, None]] = None