Login

aquafire / lightkube_informer.py

Last active 6 months ago

Like 0

aquafire revised this gist 10 months ago · 51fa92a

1 file changed, 86 insertions

lightkube_informer.py (file created)
@@ -0,0 +1,86 @@
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}")