HPC на коленке: кастомный планировщик pod’ов для Kubernetes с NUMA-awareness

Привет, Хабр!

Сегодня я расскажу о разработке кастомного планировщика pod’ов для HPC‑нагрузок в Kubernetes с учётом NUMA и специфичных требований. Рассмотрим код с примером для магазина корма для собачек и всеми нюансами реализации.

Но для начала… зачем вообще нужен кастомный планировщик для HPC?

Kubernetes в своём стандартном виде отлично справляется с большинством задач. Но когда речь идёт о высокопроизводительных вычислениях (HPC), стандартный scheduler может не выдержать требования:

Тщательное распределение ресурсов. HPC‑приложения чувствительны к задержкам, фрагментации памяти и неправильному распределению вычислительных ядер.
Учёт NUMA‑топологии. В архитектуре NUMA скорость доступа к памяти зависит от её физического расположения относительно процессора — и тут ошибка может стоить дорого.
Специфичные требования к CPU, памяти и даже специализированному оборудованию (GPU, RDMA и т. п.).

Поэтому напишем собственный планировщик, который:

Отбирает pod’ы с меткой hpc=true.
Анализирует аннотации узлов с описанием NUMA‑топологии.
Оценивает узлы по доступным ресурсам и специфическим требованиям pod’ов.
Надёжно связывает pod с выбранным узлом через Kubernetes API.

Основные функции нашего кастомного HPC-планировщика

Фильтрация pod’ов по меткам HPC

Не все pod’ы требуют особой обработки. Будем отбирать только те, что действительно настроены для HPC, используя метку hpc=true. Таким образом, стандартный scheduler продолжит работать для остальных задач, а наш планировщик займётся только критичными нагрузками.

Анализ NUMA‑топологии узлов

Каждый узел в кластере снабжён аннотацией, описывающей его NUMA‑топологию, например:

[   {"socketId": 0, "cores": 8, "memory": 32768},   {"socketId": 1, "cores": 8, "memory": 32768} ]

Это позволяет учитывать не только общее количество ядер и памяти, но и распределение этих ресурсов по сокетам, что критично для HPC‑задач.

Оценка узлов

После фильтрации узлов планировщик оценивает их по нескольким параметрам:

Доступность требуемых ядер и памяти.
Оптимальное распределение по NUMA‑сокетам.
Наличие дополнительных ресурсов (например, GPU).
Специфичные требования pod’а, передаваемые через аннотации (например, минимальное число ядер для приложения DogFoodShop).

Безопасное связывание pod»ов с узлами

Наконец, когда выбран наилучший узел, планировщик связывает pod с ним с помощью Kubernetes API. Обработка ошибок на каждом этапе гарантирует, что никакие сбои не останутся незамеченными.

Архитектура и реализация

Будем писать на моем любимом Go, ведь это язык, на котором написан сам Kubernetes. Структура проекта довольно проста:

Фильтрация pod’ов: отслеживание событий создания pod’ов с помощью информеров.
Оценка узлов: считывание аннотаций узлов, разбор NUMA‑топологии и вычисление баллов.
Binding: связывание pod’а с выбранным узлом через API Kubernetes.

Объявление пакета и импорт необходимых библиотек

Объявляем пакет main и подключаем все нужные пакеты. Стандартные пакеты (например, context, encoding/json, flag, log, sort и time) понадобятся для работы с контекстами, JSON, аргументами командной строки, логированием, сортировкой и таймерами. Библиотеки из k8s.io — это мост к Kubernetes API, информерам и настройке клиента.

package main  import ( "context" "encoding/json" "flag" "fmt" "log" "sort" "time"  v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "k8s.io/client-go/tools/cache" "k8s.io/client-go/tools/clientcmd" )

Определение структур: NumaInfo и Scheduler

В этой части кода мы объявляем структуру NumaInfo, которая описывает характеристики одного NUMA‑сокета (идентификатор сокета, количество ядер и объём памяти). Также создаём структуру Scheduler — кастомный планировщик, который будет хранить клиента Kubernetes и информер для отслеживания событий pod’ов.

apiVersion: v1 kind: Pod metadata:   name: dogfoodshop-hpc   labels:     app: dogfoodshop     hpc: "true"  # Обязательно для нашего кастомного планировщика!   annotations:     dogshop.minCores: "16"  # Минимальное требование к количеству ядер spec:   containers:     - name: dogfoodshop       image: dogfoodshop/hpc:latest       resources:         limits:           cpu: "16"           memory: "32Gi"         requests:           cpu: "16"           memory: "32Gi"

Конструктор планировщика

Здесь создаём функцию‑конструктор, которая принимает клиент Kubernetes и создаёт фабрику информеров для отслеживания pod’ов. Именно информер будет следить за событиями создания pod’ов и передавать их нашему планировщику.

apiVersion: v1 kind: Node metadata:   name: node-1   annotations:     hpc.numa/topology: '[{"socketId":0,"cores":8,"memory":32768},{"socketId":1,"cores":8,"memory":32768}]'

Запуск планировщика и обработка событий

Функция Run запускает информер и регистрирует обработчик событий. Обработчик onAddPod вызывается при создании нового pod’а. Здесь фильтруем pod’ы по метке hpc=true, чтобы работать только с нужными HPC‑задачами.

// Run запускает планировщик func (s *Scheduler) Run(stopCh <-chan struct{}) { defer func() { if r := recover(); r != nil { log.Printf("Recovered from panic: %v", r) } }() log.Println("Запускаем кастомный HPC-планировщик...") s.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: s.onAddPod, }) s.informer.Run(stopCh) }  // onAddPod вызывается при создании нового pod’а func (s *Scheduler) onAddPod(obj interface{}) { pod, ok := obj.(*v1.Pod) if !ok { return }  // Фильтруем pod’ы, которые не предназначены для HPC if val, ok := pod.Labels["hpc"]; !ok || val != "true" { return }  log.Printf("Найден HPC-pod: %s/%s", pod.Namespace, pod.Name) // Пробуем назначить pod на подходящий узел err := s.schedulePod(context.Background(), pod) if err != nil { log.Printf("Ошибка при планировании pod’а %s/%s: %v", pod.Namespace, pod.Name, err) } }

Выбор узла для pod’а и связывание

Здесь переходим к «умному» выбору узла: функция schedulePod получает список узлов, фильтрует их по готовности и NUMA‑аннотациям, оценивая их баллами, чтобы выбрать лучший кандидат; при этом filterAndScoreNodes для каждого узла проверяет его готовность, считывает NUMA‑информацию и суммирует количество ядер, учитывая специальные требования (например, для DogFoodShop требуется минимум 16 ядер), а функция isNodeReady просто определяет, находится ли узел в рабочем состоянии; далее getNodeNumaInfo декодирует NUMA‑топологию, сохранённую в аннотациях узла, и, наконец, bindPod создаёт объект binding и связывает pod с выбранным узлом через API.

// schedulePod – логика выбора узла и связывания pod’а с ним func (s *Scheduler) schedulePod(ctx context.Context, pod *v1.Pod) error { // Получаем список всех узлов nodes, err := s.clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) if err != nil { return fmt.Errorf("не удалось получить список узлов: %w", err) }  // Фильтруем узлы с учётом NUMA и ресурсных требований candidateNodes := s.filterAndScoreNodes(nodes.Items, pod) if len(candidateNodes) == 0 { return fmt.Errorf("нет подходящих узлов для pod %s/%s", pod.Namespace, pod.Name) }  // Выбираем лучший узел (с максимальным баллом) bestNode := candidateNodes[0].nodeName log.Printf("Выбран узел '%s' для pod %s/%s (балл: %d)", bestNode, pod.Namespace, pod.Name, candidateNodes[0].score)  // Выполняем binding pod’а к выбранному узлу return s.bindPod(ctx, pod, bestNode) }  // nodeScore – структура для хранения оценки узла type nodeScore struct { nodeName string score    int }  // filterAndScoreNodes фильтрует узлы и присваивает им баллы func (s *Scheduler) filterAndScoreNodes(nodes []v1.Node, pod *v1.Pod) []nodeScore { var scores []nodeScore  for _, node := range nodes { // Пропускаем недоступные узлы if !isNodeReady(&node) { continue }  // Получаем NUMA-информацию из аннотаций узла numaInfo, err := getNodeNumaInfo(node) if err != nil { log.Printf("Ошибка получения NUMA-информации с узла %s: %v", node.Name, err) continue }  // Оцениваем узел: суммируем количество ядер всех сокетов totalCores := 0 for _, socket := range numaInfo { totalCores += socket.Cores }  // Здесь можно учитывать дополнительные требования pod’а (например, память, GPU, RDMA) score := totalCores  // Пример: если pod предназначен для DogFoodShop и требует минимум 16 ядер, узлы с меньшим количеством получат штраф if val, ok := pod.Annotations["dogshop.minCores"]; ok && val == "16" && totalCores < 16 { score = 0 }  if score > 0 { scores = append(scores, nodeScore{ nodeName: node.Name, score:    score, }) } }  // Сортируем узлы по убыванию баллов sort.Slice(scores, func(i, j int) bool { return scores[i].score > scores[j].score })  return scores }  // isNodeReady проверяет, находится ли узел в состоянии Ready func isNodeReady(node *v1.Node) bool { for _, condition := range node.Status.Conditions { if condition.Type == v1.NodeReady && condition.Status == v1.ConditionTrue { return true } } return false }  // getNodeNumaInfo декодирует NUMA-информацию из аннотаций узла func getNodeNumaInfo(node v1.Node) ([]NumaInfo, error) { annotation, ok := node.Annotations["hpc.numa/topology"] if !ok { return nil, fmt.Errorf("аннотация hpc.numa/topology не найдена") }  var numaInfo []NumaInfo err := json.Unmarshal([]byte(annotation), &numaInfo) if err != nil { return nil, fmt.Errorf("ошибка декодирования NUMA-информации: %w", err) } return numaInfo, nil }  // bindPod связывает pod с выбранным узлом func (s *Scheduler) bindPod(ctx context.Context, pod *v1.Pod, nodeName string) error { // Создаем объект Binding binding := &v1.Binding{ ObjectMeta: metav1.ObjectMeta{ Namespace: pod.Namespace, Name:      pod.Name, }, Target: v1.ObjectReference{ Kind: "Node", Name: nodeName, }, }  // Выполняем вызов API для binding-а err := s.clientset.CoreV1().Pods(pod.Namespace).Bind(ctx, binding, metav1.CreateOptions{}) if err != nil { return fmt.Errorf("не удалось выполнить binding: %w", err) } log.Printf("Pod %s/%s успешно связан с узлом %s", pod.Namespace, pod.Name, nodeName) return nil }

Функция main

В функции main настраиваем подключение к Kubernetes. Если указан kubeconfig, используем его для локальной разработки, иначе — предполагаем, что код работает внутри кластера (in‑cluster config). Затем создаём клиента, инициализируем планировщик и запускаем его в отдельном горутине. В конце блокируем выполнение, чтобы приложение не завершилось.

func main() { // Читаем параметры командной строки kubeconfig := flag.String("kubeconfig", "", "Путь до kubeconfig файла") flag.Parse()  var config *rest.Config var err error if *kubeconfig != "" { // Локальная разработка: используем kubeconfig config, err = clientcmd.BuildConfigFromFlags("", *kubeconfig) if err != nil { log.Fatalf("Ошибка создания конфигурации из kubeconfig: %v", err) } } else { // На продакшене: предполагаем работу внутри кластера config, err = rest.InClusterConfig() if err != nil { log.Fatalf("Ошибка создания in-cluster конфигурации: %v", err) } }  clientset, err := kubernetes.NewForConfig(config) if err != nil { log.Fatalf("Ошибка создания клиента Kubernetes: %v", err) }  scheduler := NewScheduler(clientset) stopCh := make(chan struct{}) go scheduler.Run(stopCh)  // Блокируем выполнение, чтобы приложение не завершалось select {} }

Полный код в спойлере

package main  import ( "context" "encoding/json" "flag" "fmt" "log" "sort" "time"  v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "k8s.io/client-go/tools/cache" "k8s.io/client-go/tools/clientcmd" )  // NumaInfo описывает характеристики одного NUMA-сокета type NumaInfo struct { SocketID int `json:"socketId"` Cores    int `json:"cores"` Memory   int `json:"memory"` // в мегабайтах }  // Scheduler – структура нашего кастомного планировщика type Scheduler struct { clientset kubernetes.Interface informer  cache.SharedIndexInformer }  // NewScheduler создаёт новый экземпляр Scheduler func NewScheduler(clientset kubernetes.Interface) *Scheduler { // Создаём фабрику информеров для отслеживания pod’ов factory := informers.NewSharedInformerFactory(clientset, time.Minute) podInformer := factory.Core().V1().Pods().Informer()  return &Scheduler{ clientset: clientset, informer:  podInformer, } }  // Run запускает планировщик func (s *Scheduler) Run(stopCh <-chan struct{}) { defer func() { if r := recover(); r != nil { log.Printf("Recovered from panic: %v", r) } }() log.Println("Запускаем кастомный HPC-планировщик...") s.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: s.onAddPod, }) s.informer.Run(stopCh) }  // onAddPod вызывается при создании нового pod’а func (s *Scheduler) onAddPod(obj interface{}) { pod, ok := obj.(*v1.Pod) if !ok { return }  // Фильтруем pod’ы, которые не предназначены для HPC if val, ok := pod.Labels["hpc"]; !ok || val != "true" { return }  log.Printf("Найден HPC-pod: %s/%s", pod.Namespace, pod.Name) // Пробуем назначить pod на подходящий узел err := s.schedulePod(context.Background(), pod) if err != nil { log.Printf("Ошибка при планировании pod’а %s/%s: %v", pod.Namespace, pod.Name, err) } }  // schedulePod – логика выбора узла и связывания pod’а с ним func (s *Scheduler) schedulePod(ctx context.Context, pod *v1.Pod) error { // Получаем список всех узлов nodes, err := s.clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) if err != nil { return fmt.Errorf("не удалось получить список узлов: %w", err) }  // Фильтруем узлы с учётом NUMA и ресурсных требований candidateNodes := s.filterAndScoreNodes(nodes.Items, pod) if len(candidateNodes) == 0 { return fmt.Errorf("нет подходящих узлов для pod %s/%s", pod.Namespace, pod.Name) }  // Выбираем лучший узел (с максимальным баллом) bestNode := candidateNodes[0].nodeName log.Printf("Выбран узел '%s' для pod %s/%s (балл: %d)", bestNode, pod.Namespace, pod.Name, candidateNodes[0].score)  // Выполняем binding pod’а к выбранному узлу return s.bindPod(ctx, pod, bestNode) }  // nodeScore – структура для хранения оценки узла type nodeScore struct { nodeName string score    int }  // filterAndScoreNodes фильтрует узлы и присваивает им баллы func (s *Scheduler) filterAndScoreNodes(nodes []v1.Node, pod *v1.Pod) []nodeScore { var scores []nodeScore  for _, node := range nodes { // Пропускаем недоступные узлы if !isNodeReady(&node) { continue }  // Получаем NUMA-информацию из аннотаций узла numaInfo, err := getNodeNumaInfo(node) if err != nil { log.Printf("Ошибка получения NUMA-информации с узла %s: %v", node.Name, err) continue }  // Оцениваем узел: суммируем количество ядер всех сокетов totalCores := 0 for _, socket := range numaInfo { totalCores += socket.Cores }  // Здесь можно учитывать дополнительные требования pod’а (например, память, GPU, RDMA) score := totalCores  // Пример: если pod предназначен для DogFoodShop и требует минимум 16 ядер, узлы с меньшим количеством получат штраф if val, ok := pod.Annotations["dogshop.minCores"]; ok && val == "16" && totalCores < 16 { score = 0 }  if score > 0 { scores = append(scores, nodeScore{ nodeName: node.Name, score:    score, }) } }  // Сортируем узлы по убыванию баллов sort.Slice(scores, func(i, j int) bool { return scores[i].score > scores[j].score })  return scores }  // isNodeReady проверяет, находится ли узел в состоянии Ready func isNodeReady(node *v1.Node) bool { for _, condition := range node.Status.Conditions { if condition.Type == v1.NodeReady && condition.Status == v1.ConditionTrue { return true } } return false }  // getNodeNumaInfo декодирует NUMA-информацию из аннотаций узла func getNodeNumaInfo(node v1.Node) ([]NumaInfo, error) { annotation, ok := node.Annotations["hpc.numa/topology"] if !ok { return nil, fmt.Errorf("аннотация hpc.numa/topology не найдена") }  var numaInfo []NumaInfo err := json.Unmarshal([]byte(annotation), &numaInfo) if err != nil { return nil, fmt.Errorf("ошибка декодирования NUMA-информации: %w", err) } return numaInfo, nil }  // bindPod связывает pod с выбранным узлом func (s *Scheduler) bindPod(ctx context.Context, pod *v1.Pod, nodeName string) error { // Создаем объект Binding binding := &v1.Binding{ ObjectMeta: metav1.ObjectMeta{ Namespace: pod.Namespace, Name:      pod.Name, }, Target: v1.ObjectReference{ Kind: "Node", Name: nodeName, }, }  // Выполняем вызов API для binding-а err := s.clientset.CoreV1().Pods(pod.Namespace).Bind(ctx, binding, metav1.CreateOptions{}) if err != nil { return fmt.Errorf("не удалось выполнить binding: %w", err) } log.Printf("Pod %s/%s успешно связан с узлом %s", pod.Namespace, pod.Name, nodeName) return nil }  func main() { // Читаем параметры командной строки kubeconfig := flag.String("kubeconfig", "", "Путь до kubeconfig файла") flag.Parse()  var config *rest.Config var err error if *kubeconfig != "" { // Локальная разработка: используем kubeconfig config, err = clientcmd.BuildConfigFromFlags("", *kubeconfig) if err != nil { log.Fatalf("Ошибка создания конфигурации из kubeconfig: %v", err) } } else { // На продакшене: предполагаем работу внутри кластера config, err = rest.InClusterConfig() if err != nil { log.Fatalf("Ошибка создания in-cluster конфигурации: %v", err) } }  clientset, err := kubernetes.NewForConfig(config) if err != nil { log.Fatalf("Ошибка создания клиента Kubernetes: %v", err) }  scheduler := NewScheduler(clientset) stopCh := make(chan struct{}) go scheduler.Run(stopCh)  // Блокируем выполнение, чтобы приложение не завершалось select {} }

Пример: магазин корма для собачек с HPC-нагрузками

Представим, что есть приложение для магазина корма для собачек — DogFoodShop. Манифест для pod’а DogFoodShop:

apiVersion: v1 kind: Pod metadata:   name: dogfoodshop-hpc   labels:     app: dogfoodshop     hpc: "true"  # Обязательно для нашего кастомного планировщика!   annotations:     dogshop.minCores: "16"  # Минимальное требование к количеству ядер spec:   containers:     - name: dogfoodshop       image: dogfoodshop/hpc:latest       resources:         limits:           cpu: "16"           memory: "32Gi"         requests:           cpu: "16"           memory: "32Gi"

Для корректной работы scheduler ожидает, что узлы будут иметь аннотации с NUMA‑топологией, например:

apiVersion: v1 kind: Node metadata:   name: node-1   annotations:     hpc.numa/topology: '[{"socketId":0,"cores":8,"memory":32768},{"socketId":1,"cores":8,"memory":32768}]'

Работали ли вы с чем‑то подобным? Делитесь в комментариях.

Уже сегодня, 20 февраля в 20:00, пройдёт открытый урок, посвящённый CI/CD. Всего за 100 секунд вы увидите, как можно перейти от пустого проекта к работающей CI/CD-платформе. После этого мы разберём каждый этап: создание пайплайнов, настройку тестирования, автоматический деплой, обработку ошибок и масштабирование.

Записаться можно на странице курса «Инфраструктурная платформа на основе Kubernetes».

ссылка на оригинал статьи https://habr.com/ru/articles/884012/