Вставка данных в Redis при запуске контейнера в Kubernetes кластере

Довольно часто встречаются варианты конфигурирования и запуска Redis сервиса в кластере Kubernetes классифицируемые как stateless (т.е. без сохранения данных). В данной статье рассмотрим вариант вставки данных в Redis на старте пода в кластере - давайте разберемся!

Сразу отмечу, что кейс довольно специфический, и если ваши приложения зависят от данных в Redis (или вы хотите вставить пару миллионов ключей за 1 секунду - об этом будет отдельная статья), то лучше использовать решения для stateful приложений - операторы, StatefulSet и т. д.

Итак, в нашем конкретном случае используется самый настоящий stateless Redis - деплоймент с одной репликой. То есть мы готовы к потере данных - это некритично, но “было бы хорошо” иметь в редисе одну пару ключ/значение сразу же после запуска контейнера.

Очевидно, что вариант с initContainers не подходит (Redis уже должен быть запущен на момент вставки данных), а вариант с переписыванием Dockerfile и/или entrypoint-скриптов, мягко говоря, не элегантный.

Самым интересным вариантом для данного кейса оказалось написание собственного контроллера, который, будучи подключен к кластеру Kubernetes “слушает” интересующие нас события и выполняет предусмотренные операции.

Go клиент для Kubernetes (client-go) содержит пакет, который позволяет с легкостью получать события (events) используя Kubernetes API: k8s.io/client-go/tools/cache. Кроме того, этот пакет позволяет нам легко добавлять функции, которые будут вызваны при получении определенного события и хранить объекты в памяти (в так называемом хранилище, Store).

Хотя пакет cache предоставляет необходимые нам инструменты, его инициализация и использование могут быть избыточными и несколько обременительными, когда речь идет о получении простых обновлений от Kubernetes API. Практически все статьи по использованию k8s.io/client-go/tools/cache, которые мне удалось найти, предлагают использовать этот пакет “как есть”.

Однако существует еще один пакет, который объединяет концепции, которые предоставленные пакетом cache, в один: k8s.io/client-go/informers. Он поставляется с простой фабрикой для всех ресурсов Kubernetes - то, что нам нужно!

Инициализация фабрики выглядит примерно так:

...
	kubeInformerFactory := informers.NewSharedInformerFactoryWithOptions(clientset, time.Second*30,
		informers.WithNamespace("default"),
		informers.WithTweakListOptions(func(options *v1.ListOptions) {
			options.LabelSelector = "app=ads-redis-statistic"
		}))
...

Так мы получаем новый экземпляр SharedInformerFactory с дополнительными опциями (неймспейс default и LabelSelector “app=ads-redis-statistic”). Первым агрументом является наш клиент, который подключается и взаимодействует с Kubernetes API, а вторым - периодичность, с которой информер будет синхронизировать данные.

Теперь мы можем с легкостью начать получать необходимые нам события о интересующем нас ресурсе - новом поде в кластере Kubernetes. Выглядеть это будет примерно так:

...
	podInformer := kubeInformerFactory.Core().V1().Pods().Informer()

	podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: onAdd,
	})
...

Первая строка инициализирует информер, который настроен на получение событий о подах групы Core/v1 API (фильтрация по неймспейсу/селектору у нас указана еще на этапе создания фабрики).

Вторая строка добавляет функцию-обработчик (handler), которая будет вызываться каждый раз, когда информет получает от Kubernetes API интересующее нас событие (а именно, добавление в неймспейсе default нового пода с селектором app: ads-redis-statistic).

Сама функция onAdd выглядит следующим образом:

...
func onAdd(obj interface{}) {
	rc := redis.InitRedisClient(
		helpers.GetEnvironmentVariableAsString("REDIS_HOST", "127.0.0.1:6379"),
		helpers.GetEnvironmentVariableAsString("REDIS_PASSWORD", ""),
		helpers.GetEnvironmentVariableAsInteger("REDIS_DB", 4))

	rc.Connect()
	rc.SetValue(
		"flow-rules-key",
		"[{\"resource\":\"loopme.grpc.ssp.v0.AdsTxtRecordService/GetAdsTxtRelationships\",\"count\":100.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"loopme.grpc.ssp.v0.PublisherAccountService/GetPublisherById\",\"count\":5.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"loopme.grpc.ssp.v1.PublisherAccountService/GetPublisherById\",\"count\":5.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"loopme.grpc.ssp.v0.BundleLegacyService/GetBundleByKey\",\"count\":20.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"loopme.lsm.ssp.v0.BundleService/GetBundleById\",\"count\":20.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"loopme.lsm.ssp.v0.BundleService/QueryBundle\",\"count\":20.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"loopme.grpc.ssp.v0.AppLegacyService/GetAppById\",\"count\":10.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"loopme.grpc.ssp.v0.AppLegacyService/GetAppIdByKey\",\"count\":10.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"loopme.grpc.ssp.v0.AppLegacyService/GetAppIdByContainerKey\",\"count\":16.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"loopme.grpc.ssp.v0.AppLegacyService/GetAppByContainerKey\",\"count\":10.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"ExchangeThrottleRateService/GetThrottleRatesByKeys\",\"count\":20.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"dsp-fetcher\",\"count\":25.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"exchange-fetcher\",\"count\":300.0,\"grade\":\"THREAD\",\"limit-app\":\"default\"},{\"resource\":\"kafka_dmp_ads_requests_info\",\"count\":500.0,\"grade\":\"QPS\",\"limit-app\":\"default\"}]")

	fmt.Println(rc.QueryValue("flow-rules-key"))
	rc.Disconnect()
...
}

Функция инициализирует redis-клиент с заданными параметрами, выполняет подключение к сервису, вставляет данные в БД, для проверки получает значение для указанного ключа и отключается от данного экземпляра Redis.

Полный пример кода (а также пример деплоймента для данного контроллера) доступен здесь.