Работа с информерами Java kubernetes client10.07.2024 10:15
Оглавление
Введение
Описание создания проекта с нуля
Создание клиентов API для получения объектов kubernetes
Инициализация информеров для получения Pods, Nodes и Ingresses
Создание Listener для запуска информеров
Извлечение данных из информеров и их подготовка для отправки через API
Проверка результата
Небольшой дисклеймер
Проект создан для учебных целей, поэтому код в некоторых местах намеренно (или случайно) упрощен.
Например, path для ingress взят просто первый из списка, на каждый вызов API создается отдельный экземпляр клиента и т. д.
Введение
Привет, меня зовут Сергей, старший разработчик 80 уровня компании DataBlend (группа компаний GlowByte). Наша команда занимается разработкой продукта ClusterManager, который управляет поведением и мониторит состояния таких продуктов, как GreenPlum, ClickHouse, DWH, Nova и т. д.
Около полутора лет назад у нас появилась необходимость собирать и отображать в удобном виде и разрезах метрики и данные об объектах кластеров Kubernetes, в которых развернут продукт Nova.
Для этих целей был выбран официальный kubernetes-client для Java.
Поначалу мы пошли по пути сбора данных о нодах, подах и так далее по расписанию и сохранению их в БД в удобном виде. Но, как это часто бывает, цели и желания со временем меняются, и жизнь заставила перейти к мгновенному получению и отображению изменений.
Лучше всего для этой цели подходит механизм информеров kubernetes-client.
И сейчас мы посмотрим, с какой стороны их лучше начинать есть.
Напишем простое приложение, которое в реальном времени отслеживает состояние Pods, Nodes и Ingresses и по запросу отдает нам информацию о них. Для этого мы повесим информеры на указанные ресурсы Kubernetes.
Если нужно отслеживать CRD-ресурсы, то информеры, к сожалению, не подойдут.
Получать и хранить информацию о ресурсах Kubernetes будем в памяти приложения.
«А у нас этой памяти — завались, у нас папа на фабрике по производству чипов памяти работает»
как сказал бы кот Матроскин.
Описание создания проекта с нуля
Этот пункт не имеет прямого отношения к kubernetes-client, если нет нужды повторять проект, этот пункт можно пропустить и скачать уже готовый проект.
Ссылка на GitHub проекта.
Для запуска приложения необходимо из папки проекта выполнить команду:
Первым делом мы создаем новый проект на основе Spring Boot:
Далее добавляем в pom.xml зависимости для генерации контроллеров из спецификации OpenApi 3.0, mapstruct для маппинга DTO и, собственно, kubernetes-client:
Добавляем плагины для генерации контроллеров и DTO из спецификации OpenApi 3.0, а также maven plugin, в котором прописываем процессоры lombok и mapstruct, не забудьте заменить пути на свои:
В переменную kubernetes.config-file.path мы будем передавать путь к нашему конфиг-файлу Kubernetes.
Теперь генерируем интерфейсы контроллеров и модели командой: mvn clean generate-sources
Если все прошло успешно, то в папке target мы увидим такую картину:
Теперь помечаем package generated-sources как »Generated Source Root», в Intellij IDEA для этого вызываем контекстное меню на нужной папке и выбираем »Mark Directory as»/ «Generated Source Root». Папка в интерфейсе посинеет.
Если у вас не IDEA, мои полномочия все, разбирайтесь.
@RestController
@RequiredArgsConstructor
public class ResourceListController implements ResourceListApi {
private final PodListService podListService;
private final NodeListService nodeListService;
private final IngressListService ingressListService;
@Override
public ResponseEntity getIngresses() {
return ResponseEntity.ok(ingressListService.getIngresses());
}
@Override
public ResponseEntity getNodes() {
return ResponseEntity.ok(nodeListService.getNodes());
}
@Override
public ResponseEntity getPods() {
return ResponseEntity.ok(podListService.getPods());
}
}
Классы PodListService, NodeListService и IngressListService будут созданы в пункте 6.
На этом подготовка проекта завершена, можно переходить к тому, ради чего это затевалось.
Создание клиентов API для получения объектов Kubernetes
На момент написания статьи последняя версия клиента — 20.0.1. От версии к версии функционал библиотеки, структура классов, модели данных и т. д. у клиента может меняться, учитывайте это. Радует, что выпускаются также новые версии клиента с поддержкой старой структуры данных и методов. Например, версия 20.0.1-legacy поддерживает код, написанный для 18 версии клиента.
Также версия 20.0.1 прекратила поддержку Java 8.
Структура нашего приложения будет выглядеть таким образом:
Создаем класс KubernetesResourceService, отвечающий за создание и выдачу api для подключения к Kubernetes
@Service
public class KubernetesResourceService {
@Value("${kubernetes.config-file.path}")
private String configFilePath;
private static final Logger LOGGER = LoggerFactory.getLogger(KubernetesResourceService.class);
public Optional getCoreV1Api() {
Optional clientOptional = getApiClient();
if (clientOptional.isEmpty()) {
return Optional.empty();
}
ApiClient client = clientOptional.get();
CoreV1Api api = new CoreV1Api(client);
return Optional.of(api);
}
public Optional getSharedInformerFactory() {
Optional clientOptional = getApiClient();
if (clientOptional.isEmpty()) {
LOGGER.warn("ApiClient is null.");
return Optional.empty();
}
ApiClient client = clientOptional.get();
client.setReadTimeout(0);
SharedInformerFactory factory = new SharedInformerFactory(client);
return Optional.of(factory);
}
public Optional getNetworkingApi() {
Optional clientOptional = getApiClient();
if (clientOptional.isEmpty()) {
return Optional.empty();
}
ApiClient client = clientOptional.get();
NetworkingV1Api api = new NetworkingV1Api();
api.setApiClient(client);
return Optional.of(api);
}
private Optional configFile() {
try {
Path filePath = Paths.get(configFilePath);
byte[] fileBytes = Files.readAllBytes(filePath);
String configFile = new String(fileBytes);
return Optional.of(configFile);
} catch (IOException e) {
LOGGER.error("Error while getting Kubernetes configFile: {}", e.getMessage());
return Optional.empty();
}
}
private Optional getApiClient() {
Optional configFileOptional = configFile();
if (configFileOptional.isEmpty()) {
LOGGER.error("Config file is empty or null.");
return Optional.empty();
}
String configFile = configFileOptional.get();
try (Reader reader = new StringReader(configFile)){
KubeConfig kubeConfig = KubeConfig.loadKubeConfig(reader);
ApiClient client = ClientBuilder.kubeconfig(kubeConfig).build();
client.setReadTimeout(60000);
return Optional.ofNullable(client);
} catch (IOException e) {
LOGGER.error("Error while getting kubernetes client from config file");
return Optional.empty();
}
}
}
private Optional getApiClient() {
Optional configFileOptional = configFile();
if (configFileOptional.isEmpty()) {
LOGGER.error("Config file is empty or null.");
return Optional.empty();
}
String configFile = configFileOptional.get();
try (Reader reader = new StringReader(configFile)){
KubeConfig kubeConfig = KubeConfig.loadKubeConfig(reader);
ApiClient client = ClientBuilder.kubeconfig(kubeConfig).build();
client.setReadTimeout(60000);
return Optional.ofNullable(client);
} catch (IOException e) {
LOGGER.error("Error while getting kubernetes client from config file");
return Optional.empty();
}
}
}
Переменная configFilePath – это наш путь к конфиг-файлу Kubernetes, значение которой мы получаем из системного свойства, указанного при запуске.
Метод getConfig получает содержимое файла.
Метод getApiClient создает клиент на основе конфиг-файла.
ApiClient: Базовый клиент для взаимодействия с Kubernetes API. Он управляет соединениями, аутентификацией и общими настройками.
При необходимости в нем можно также указать URL, Credentials и т. д.
Строка »client.setReadTimeout (60000); » настраивает время ожидания (timeout) для операций чтения на клиенте Kubernetes (ApiClient). Мы установили время ожидания 60 секунд. Значение 0 означает, что время ожидания будет бесконечным.
При создании SharedInformerFactory нужно установить client.setReadTimeout (0); — так как его соединение должно быть бессрочным.
На основе ApiClient создаются другие объекты в методах getCoreV1Api, getNetworkingApi, getSharedInformerFactory — с их помощью мы обращаемся к Kubernetes.
Инициализация информеров для получения Pods, Nodes и Ingresses
Теперь нам нужно создать информеры, которые будут запускаться вместе с приложением, загружать нужные нам типы объектов и следить за их актуальностью.
Создаем класс InitKubernetesResourceService
@Service
@RequiredArgsConstructor
public class InitKubernetesResourceService {
private static final Logger LOGGER = LoggerFactory.getLogger(InitKubernetesResourceService.class);
private final KubernetesResourceService kubernetesResourceService;
private final KubernetesResourceInformerFactoryService informerFactoryService;
private final KubernetesResourceInformerContextBuilderService contextBuilderService;
private final KubernetesResourceInformerContextManager contextManager;
public void watchResources() {
Optional informerFactoryOptional = kubernetesResourceService.getSharedInformerFactory();
if (informerFactoryOptional.isEmpty()) {
LOGGER.error("Failed to initialize KubernetesApiFactory due to missing SharedInformerFactory.");
return;
}
SharedInformerFactory informerFactory = informerFactoryOptional.get();
Optional coreV1ApiOptional = kubernetesResourceService.getCoreV1Api();
if (coreV1ApiOptional.isEmpty()) {
LOGGER.error("Failed to initialize KubernetesApiFactory due to missing CoreV1Api.");
return;
}
CoreV1Api coreV1Api = coreV1ApiOptional.get();
Optional networkingV1ApiOptional = kubernetesResourceService.getNetworkingApi();
if (networkingV1ApiOptional.isEmpty()) {
LOGGER.error("Failed to initialize KubernetesApiFactory due to missing NetworkingV1Api.");
return;
}
NetworkingV1Api networkingV1Api = networkingV1ApiOptional.get();
informerFactoryService.registerInformers(informerFactory, coreV1Api, networkingV1Api);
KubernetesResourceInformerContext context = contextBuilderService.buildContext(informerFactory);
contextManager.putContext(context);
informerFactory.startAllRegisteredInformers();
}
}
Здесь мы получаем объекты SharedInformerFactory, CoreV1Api, NetworkingV1Api, с их помощью зарегистрируем информеры в классе KubernetesResourceInformerFactoryService, сохраним ссылки на них в классе KubernetesResourceInformerContext. После чего передадим объект KubernetesResourceInformerContext для хранения и выдачи в KubernetesResourceInformerContextManager.
В конце запустим все информеры informerFactory.startAllRegisteredInformers ();
По-хорошему, когда информеры больше не нужны, их нужно остановить методом informerFactory.stopAllRegisteredInformers (true); , но я буду плохим.
Для того, чтобы создать информеры, мы должны создать объекты CallGenerator для каждого информера.
Также нам нужен отдельный информер для каждого типа ресурсов.
Переменная RESYNC_PERIOD_MILLISECONDS отвечает за период времени в миллисекундах между повторными синхронизациями.
Информеры меняют состояние объектов у себя внутри сразу при изменении их в Kubernetes, но также время от времени проводят полную ресинхронизацию на случай, если какое-то событие было пропущено.
Также хочу заметить, что в данном примере мы вешаем информеры на все Pods, Nodes, Ingresses кластера, но есть возможность фильтровать их по неймспейсам, лейблам и т. д. еще на этапе создания информера. Также можно создать несколько информеров для одного ресурса, чтобы забирать Pods только из определенных неймспейсов с указанными лейблами, например.
Регистрируем информеры методом informerFactory.sharedIndexInformerFor (podCallGenerator, V1Pod.class, V1PodList.class, RESYNC_PERIOD_MILLISECONDS), указывая соответствующие генераторы и классы для них.
Также можно создать обработчики событий для информеров.
В нашем примере мы создадим такой обработчик для информера Pods, чтобы логировать события, происходящие с подами.
Для этого получим сам информер podInformer из фабрики и передадим его в ResourceEventHandlerBuilder. Полученный обработчик podResourceEventHandler добавим в информер podInformer.addEventHandler (podResourceEventHandler)
@Service
@RequiredArgsConstructor
public class ResourceEventHandlerBuilder {
private static final Logger LOGGER = LoggerFactory.getLogger(ResourceEventHandlerBuilder.class);
public ResourceEventHandler getPodResourceEventHandler(SharedIndexInformer podInformer) {
return new ResourceEventHandler() {
@Override
public void onAdd(V1Pod pod) {
if(podInformer.hasSynced()){
LOGGER.info("Pod {} added", pod.getMetadata().getName());
}
}
@Override
public void onUpdate(V1Pod oldPod, V1Pod newPod) {
if(podInformer.hasSynced()){
LOGGER.info("Pod {} updated", newPod.getMetadata().getName());
}
}
@Override
public void onDelete(V1Pod pod, boolean deletedFinalStateUnknown) {
if(podInformer.hasSynced()){
LOGGER.info("Pod {} deleted", pod.getMetadata().getName());
}
}
};
}
}
Сам билдер достаточно прост. Мы создаем новый объект ResourceEventHandler и переопределяем три метода, добавляя в них логирование, которое срабатывает тогда, когда информер уже синхронизирован.
podInformer.hasSynced () возвращает true в случае, если информер уже был первоначально синхронизирован. Это важно, если код, выполняемый внутри методов ResourceEventHandler, зависит от состава собираемых ресурсов.
Например, у нас в проекте внутри такого обработчика выполняется сложный обсчет статусов сервисов, которые зависят от состава подов и пишутся в историю изменений. Без такого условия при запуске приложения мы получали стопку записей об изменении статусов, в то время как кластер спокойно продолжал работать не меняясь.
Девопс кластера Nova, когда увидел 50 FAILED и WARNING в истории статусов
4.2 Хранение ссылок на информеры
Ссылки на информеры хранятся в record KubernetesResourceInformerContext
public record KubernetesResourceInformerContext(
SharedIndexInformer podInformer,
SharedIndexInformer nodeInformer,
SharedIndexInformer ingressInformer
){
}
В общем-то мы можем хранить ссылку на саму фабрику SharedInformerFactory, получая информеры непосредственно из нее, но это не очень удобно.
Управляет объектом KubernetesResourceInformerContext класс KubernetesResourceInformerContextManager
@Service
public class KubernetesResourceInformerContextManager {
private final AtomicReference informerContextRef = new AtomicReference<>();
public KubernetesResourceInformerContext getContext() {
return this.informerContextRef.get();
}
public void putContext(KubernetesResourceInformerContext context) {
this.informerContextRef.set(context);
}
}
Создание Listener для запуска информеров
Теперь создадим Listener, который после запуска приложения создаст информеры и сложит их в KubernetesResourceInformerContext
@Service
@RequiredArgsConstructor
public class AppStartUpEventListener {
public static final Logger LOGGER = LoggerFactory.getLogger(AppStartUpEventListener.class);
private final InitKubernetesResourceService kubernetesResourceService;
@EventListener(ApplicationReadyEvent.class)
public void applicationReady() {
LOGGER.info("Start [{}]", ApplicationReadyEvent.class.getName());
CompletableFuture
.runAsync(() -> {})
.thenRunAsync(() -> {
try {
kubernetesResourceService.watchResources();
} catch (Exception e) {
LOGGER.error("ERROR on application start up event", e);
}
});
LOGGER.info("Stop [{}]", ApplicationReadyEvent.class.getName());
}
}
Извлечение данных из информеров и их подготовка для отправки через API
Мы молодцы. Информеры создаются при запуске приложения и собирают инфу о Pods, Nodes, Ingresses.
Теперь мы создадим класс KubernetesObjectsFetcherService, который будет извлекать эти важные данные.
@Service
@RequiredArgsConstructor
public class KubernetesObjectsFetcherService {
private static final Logger LOGGER = LoggerFactory.getLogger(KubernetesObjectsFetcherService.class);
private final KubernetesResourceInformerContextManager contextManager;
public List getPods() {
KubernetesResourceInformerContext context = contextManager.getContext();
if (context == null) {
LOGGER.error("PodInformerContext is null");
return List.of();
}
return getV1Pods(context);
}
public List getNodes() {
KubernetesResourceInformerContext context = contextManager.getContext();
if (context == null) {
LOGGER.error("NodeInformerContext is null");
return List.of();
}
return getV1Nodes(context);
}
public List getNamespacedIngresses(String namespace) {
KubernetesResourceInformerContext context = contextManager.getContext();
if (context == null) {
LOGGER.error("IngressInformerContext is null");
return List.of();
}
List ingresses = getV1Ingresses(context);
return filteredIngresses(namespace, ingresses);
}
protected List filteredIngresses(String namespace, List ingresses) {
return Optional.ofNullable(ingresses).orElse(List.of())
.stream()
.filter(ingress -> ingress.getMetadata() != null)
.filter(ingress -> ingress.getMetadata().getNamespace() != null)
.filter(ingress -> ingress.getMetadata().getNamespace().equals(namespace))
.toList();
}
protected List getV1Pods(KubernetesResourceInformerContext context) {
return Optional.ofNullable(context).map(KubernetesResourceInformerContext::podInformer).map(SharedIndexInformer::getIndexer).map(Store::list).orElse(List.of());
}
protected List getV1Nodes(KubernetesResourceInformerContext context) {
return Optional.ofNullable(context).map(KubernetesResourceInformerContext::nodeInformer).map(SharedIndexInformer::getIndexer).map(Store::list).orElse(List.of());
}
protected List getV1Ingresses(KubernetesResourceInformerContext context) {
return Optional.ofNullable(context).map(KubernetesResourceInformerContext::ingressInformer).map(SharedIndexInformer::getIndexer).map(Store::list).orElse(List.of());
}
}
Для ингрессов я создал дополнительную фильтрацию по неймспейсам, так как наша спецификация предполагает это.
Далее создадим 3 класса бизнес-логики, которые отвечают за извлечение и преобразование объектов в нужный вид.