Перемещение подов к primary ноде patroni или как использовать taints, labels в k8s

Низкая сетевая задержка между прикладом и базой может стать головной болью для fintech, если вы обслуживаете kubernetes и patroni исключительно на bare metal. Разрыв сетевой связности и последующая смена primary node patroni обычный кейс который может произойти в жизни ops, к примеру когда горят крупные датацентры или уборщица случайно задела шваброй кабель питания. И тут на помощь приходят taints, labels из kubernetes.

Итак что мы имеем:

1. геораспределенный кластер kubernetes

2. несколько подов с нашим приложением, которое отказывается правильно работать при network latency близкое к 20ms

3. геораспреденный кластер patroni который работает как часики, но кто знает что может произойти на этот раз.

К сожалению, взахлеб обчитавшись документации, мануалов и stackoverflow, я пришел к выводу что taint отрабатывает только при планировании подов, никакого выселения не происходит если налету поменять taint и label. И было решено написать писать свой костыль сервис.

Так как я имею какой либо опыт написания на python я выбрал именно его, потому что есть быстрая возможность реализовать свой велосипед быстро и без мук. Логика следующая, давайте сделаем daemonset, где livenes probe это ручка меряющая задержку до мастер ноды патрони (с окном 10 icmp запросов), записываем значения в redis, а затем используя значения вешаем теги и отправляем блестящий rollout restart

Discover primary node

Не должно вызвать какого либо непонимания, так как patroni реализует сервис на 8008 порту, отдающий статус 200 если это мастер и 503 если это реплика.

import requests
from flask import Blueprint, Response
from json import dumps, loads

from app.constants import *

load_dotenv(find_dotenv())
main = Blueprint('main', name)
redis_conn = REDIS_CONN

@main.route('/primary')
def discover_primary():
    res = RESPONSE
    for node in PATRONI_HOSTS:
        r = requests.get(f'{getenv("PATRONI_SCHEMA")}{node}:{getenv("PATRONI_PORT")}')
        if r.status_code == 200:
            res['data']['node'] = node
            res['message'] = 'successful'
            redis_conn.set('primary', node)
            return Response(dumps(res), status=200, mimetype='application/json')
    res['error'] = 1
    res['message'] = 'primary not found'
    return Response(dumps(res), status=500, mimetype='application/json')

Liveness probe

Проверка на работоспособность сервиса и также замер пингов до мастер ноды патрони

#...
from ping3 import ping 
@main.route('/liveness-probe')
def liveness_probe():
    res = RESPONSE
    res['data']['node'] = NODE
    res['data']['round-trip'] = 0
    result_pings = []

    for _ in range(COUNT_PING):
        single_ping = ping(loads(discover_master().data)['data']['node'], unit='ms')
        if single_ping:
            result_pings.append(single_ping)
    try:
        res['data']['round-trip'] = str(sum(result_pings) / len(result_pings))
        res['message'] = 'successful'
    except ZeroDivisionError:
        res['message'] = 'something was going on (ping)'
    redis_conn.set(NODE, res['data']['round-trip'])
    return Response(dumps(res), status=200, mimetype='application/json')

Не забываем про изначальную инициализацию:

@main.route('/')
def init():
    primary = redis_conn.get('primary_host')
    for node in KUBE_NODES:
        if redis_conn.get(node):
            pass
        else:
            redis_conn.set(node, 0)
    if primary is None:
        redis_conn.set('primary_host', 'localhost')
    return "It's a Switchover for patroni and kubernetes"

Cronjob или как модно называть оператор:

Необходима cronjob которая будет по сути следить за стейтом наших taints, label и выставлять необходимые тэги, лейблы в зависимости от расстояния [far,intermediate,closely]

import subprocess

from app.constants import *
from app.main.views import init


redis_conn = REDIS_CONN
check = dict()
min_ping = MIN_PING
max_ping = MAX_PING

try:
    current_db_master = redis_conn.get('primary').decode()
    pre_current_db_master = redis_conn.get('primary_host').decode()
except BaseException as error:
    init()
    current_db_master = redis_conn.get('primary').decode()
    pre_current_db_master = redis_conn.get('primary_host').decode()


def subprocess_wrapper(command):
    return subprocess.run(
        command, text=True, check=False,
        capture_output=True
    )


for node in KUBE_NODES:
    value = redis_conn.get(node)
    if value:
        check[node] = int(float(value.decode()))

for _, value in check.items():
    max_ping = value if max_ping < value else max_ping
    min_ping = value if min_ping > value else min_ping

for node in KUBE_NODES:
    if check.get(node) in range(0, min_ping + WINDOW_PING):
        subprocess_wrapper(
            ['kubectl', 'taint', 'nodes', node, 'switchover:NoExecute', '--overwrite']
        )
        subprocess_wrapper(
            ['kubectl', 'label', 'nodes', node, 'switchover=closely', '--overwrite']
        )
    elif check.get(node) in range(max_ping - WINDOW_PING, max_ping + WINDOW_PING):
        subprocess_wrapper(
            ['kubectl', 'taint', 'nodes', node, 'switchover:NoExecute-']
        )
        subprocess_wrapper(
            ['kubectl', 'label', 'nodes', node, 'switchover=far', '--overwrite']
        )
    else:
        subprocess_wrapper(
            ['kubectl', 'taint', 'nodes', node, 'switchover:NoExecute-']
        )
        subprocess_wrapper(
            ['kubectl', 'label', 'nodes', node, 'switchover=intermediate', '--overwrite']
        )

if pre_current_db_master != current_db_master:
    for app in APPS:
        subprocess_wrapper(
            ['kubectl', 'rollout', 'restart', app]
        )
    redis_conn.set('primary_host', current_db_master)

Я не претендую на открытия чего либо нового, меня мучает вопрос, как опсы в наше время решают данную проблему. Может быть тут найдутся люди готовые поделиться болью или даже лучшим решением. Интересно как люди решают данный кейс.