lightkube_informer.py

lightkube_informer.py · 3.8 KiB · Python Raw

"""Информер для обновления кэша ресурсов NodePool"""

import asyncio
import sys

from config import CHUNK_SIZE, DEFAULT_SCHEDULER, logger
from lightkube import AsyncClient, NotReadyError
from lightkube.generic_resource import create_global_resource

# создаем класс для ресурса типа nodepool
NodePool = create_global_resource("k8s.dpc24.ru", "v1", "NodePool", "nodepools")

class NodePoolInformer:
    """Информер для NodePool (паттерн list-and-watch)"""

def __init__(self, client: AsyncClient):
        # клиент для k8s
        self.client = client
        # внутренний кэш (# nodepool_name -> scheduler_name)
        self._cache = dict()
        # блокировка
        self._lock = asyncio.Lock()

async def fill(self):
        """Начальное наполнение кэша информера"""
        # предварительная очистка
        self._cache.clear()
        # запуск заполнения словаря
        async for nodepool in self.client.list(NodePool, chunk_size=CHUNK_SIZE):
            # имя nodepool
            nodepool_name: str = nodepool.metadata.name
            # имя планировщика
            scheduler_name: str = nodepool.spec.get("schedulerName", DEFAULT_SCHEDULER)
            # ставим блокировку
            async with self._lock:
                # заносим объект в кэш
                self._cache[nodepool_name] = scheduler_name

async def get(self):
        """Получить данные из кэша"""
        # блокировка, чтобы не испортить данные
        async with self._lock:
            # данные из кэша
            data = self._cache
        return data

async def watch(self):
        """Обновление кэша"""
        # запускаем watch
        while True:
            try:
                async for event, nodepool in self.client.watch(NodePool):
                    try:
                        # имя nodepool
                        nodepool_name: str = nodepool.metadata.name
                        # имя планировщика
                        scheduler_name: str = nodepool.spec.get(
                            "schedulerName", DEFAULT_SCHEDULER
                        )
                        # реакция на основе событий
                        match event:
                            case "ADDED" | "MODIFIED":
                                async with self._lock:
                                    self._cache[nodepool_name] = scheduler_name
                                logger.debug(
                                    f"Nodepool added: {nodepool_name} -> {scheduler_name}"
                                )
                            case "DELETED":
                                async with self._lock:
                                    self._cache.pop(nodepool_name, None)
                                logger.debug(f"Nodepool deleted: : {nodepool_name}")
                            case _:
                                pass
                    except (KeyError, NotReadyError):
                        # пустой watch, отсутствие ресурса в кэше и пропуск события
                        pass
                    except Exception as error:
                        logger.error(f"Error in watch_nodepool_crs: {error}")
                        sys.exit(1)
            except Exception as error:
                logger.error(
                    f"Watch error for Nodepool: {error}. Reconnecting in 5s..."
                )
                await asyncio.sleep(5)
            logger.debug(f"Nodepool cache: {self._cache}")

1	"""Информер для обновления кэша ресурсов NodePool"""
2
3	import asyncio
4	import sys
5
6	from config import CHUNK_SIZE, DEFAULT_SCHEDULER, logger
7	from lightkube import AsyncClient, NotReadyError
8	from lightkube.generic_resource import create_global_resource
9
10	# создаем класс для ресурса типа nodepool
11	NodePool = create_global_resource("k8s.dpc24.ru", "v1", "NodePool", "nodepools")
12
13
14	class NodePoolInformer:
15	"""Информер для NodePool (паттерн list-and-watch)"""
16
17	def __init__(self, client: AsyncClient):
18	# клиент для k8s
19	self.client = client
20	# внутренний кэш (# nodepool_name -> scheduler_name)
21	self._cache = dict()
22	# блокировка
23	self._lock = asyncio.Lock()
24
25	async def fill(self):
26	"""Начальное наполнение кэша информера"""
27	# предварительная очистка
28	self._cache.clear()
29	# запуск заполнения словаря
30	async for nodepool in self.client.list(NodePool, chunk_size=CHUNK_SIZE):
31	# имя nodepool
32	nodepool_name: str = nodepool.metadata.name
33	# имя планировщика
34	scheduler_name: str = nodepool.spec.get("schedulerName", DEFAULT_SCHEDULER)
35	# ставим блокировку
36	async with self._lock:
37	# заносим объект в кэш
38	self._cache[nodepool_name] = scheduler_name
39
40	async def get(self):
41	"""Получить данные из кэша"""
42	# блокировка, чтобы не испортить данные
43	async with self._lock:
44	# данные из кэша
45	data = self._cache
46	return data
47
48	async def watch(self):
49	"""Обновление кэша"""
50	# запускаем watch
51	while True:
52	try:
53	async for event, nodepool in self.client.watch(NodePool):
54	try:
55	# имя nodepool
56	nodepool_name: str = nodepool.metadata.name
57	# имя планировщика
58	scheduler_name: str = nodepool.spec.get(
59	"schedulerName", DEFAULT_SCHEDULER
60	)
61	# реакция на основе событий
62	match event:
63	case "ADDED" \| "MODIFIED":
64	async with self._lock:
65	self._cache[nodepool_name] = scheduler_name
66	logger.debug(
67	f"Nodepool added: {nodepool_name} -> {scheduler_name}"
68	)
69	case "DELETED":
70	async with self._lock:
71	self._cache.pop(nodepool_name, None)
72	logger.debug(f"Nodepool deleted: : {nodepool_name}")
73	case _:
74	pass
75	except (KeyError, NotReadyError):
76	# пустой watch, отсутствие ресурса в кэше и пропуск события
77	pass
78	except Exception as error:
79	logger.error(f"Error in watch_nodepool_crs: {error}")
80	sys.exit(1)
81	except Exception as error:
82	logger.error(
83	f"Watch error for Nodepool: {error}. Reconnecting in 5s..."
84	)
85	await asyncio.sleep(5)
86	logger.debug(f"Nodepool cache: {self._cache}")