Оператором в Kubernetes принято называть развертывание, которое самостоятельно управляет ресурсами кластера, регистрирует новые Custom Resource Definition (CRD) и, в некоторых случаях, добавляется для наблюдения за существующими ресурсами (через механизмы Dynamic Admission Control). В этой статье на примере создания оператора для развертывания и управления кластером Aerospike мы попробуем разобраться с этапами создания оператора, способами взаимодействия с кластером и проблемами, с которыми можно встретиться в реальной практике. Всех практикующих DevOps и желающих поднять автоматизацию развертывания своих сервисов на новый уровень приглашаю под кат.
Прежде всего нужно определить, какую функциональность будет поддерживать будущий оператор. Поскольку база данных Aerospike допускает работу в режиме кластера, будет разумно создать CRD для описания характеристик кластера (количество подов, ограничения по памяти и процессору и др), конфигурации пространства имен, а также настройки резервного копирования. Также оператор должен отслеживать изменения в существующих ресурсах, связанных с ним, и динамически изменять топологию системы и/или настройки подов.
Для разработки оператора мы будем использовать язык программирования Go и набор инструментов Operator SDK, который нужно будет предварительно установить:
git clone https://github.com/operator-framework/operator-sdk
cd operator-sdk
make installПосле установки SDK можно выполнить инициализацию заготовки проекта оператора:
operator-sdk init --domain dzolotov.tech --owner "Dmitrii Zolotov"Поскольку мы предполагаем создание дополнительных типов ресурсов, необходимо определить идентификатор версии API, который представляет собой URI на произвольном домене. Для удобства использования operator-sdk можно сразу установить поддержку автодополнения в shell Debian/Ubuntu:
apt install bash-completion
echo 'source <(operator-sdk completion bash)' >>~/.bashrcOperator SDK поддерживает создание операторов для разных DevOps-решений, включая Ansible и Helm, для этого при инициализации можно указывать необходимые плагины. Также инструмент командной строки позволяет выполнять развертывание оператора (operator-sdk olm), запуск оператора в выбранном контексте (operator-sdk run) и выполнять тесты (operator-sdk scorecard). Также можно создавать фрагменты кода для реализации разных типов Kubernetes-ресурсов. Начнем с создания версии apiVersion:
operator-sdk create api --group aerospike --version v1alpha1 --kind AerospikeCluster --resource --controllerГенератор создает несколько файлов для взаимодействия с Kubernetes API:
api/v1alpha1/aerospikecluster_types.go- описание типов (CRD);controllers/aerospikecluster_controller.go- контроллер для создания и отслеживания типов CRD;/go.mod- содержит зависимости для подключения к API (k8s.io/apimachinery - доступ к API для управления ресурсами, k8s.io/client-go - Go-клиент для взаимодействия с API, sigs.k8s.io/controller-runtime - базовые классы для реализации контроллера, github.com/onsi/ginkgo - тестовый фреймворк для проверки оператора).
Попробуем сгенерировать манифесты для Kubernetes, которые будут определять Custom Resource Definition и развертывания оператора: make manifests. Результатом выполнения будет генерация файлов в crd/bases - регистрация CRD-ресурсов (apiextensions.k8s.io/v1/CustomResourceDefinition) с названием aerospikeclusters.aerospike.dzolotov.tech для создаваемого API (группа aerospike.my.domain, тип ресурса AerospikeCluster). По умолчанию схема ресурса содержит строковое поле foo, далее мы переопределим схему в коде описания типов на Go:
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
annotations:
controller-gen.kubebuilder.io/version: v0.8.0
name: aerospikeclusters.aerospike.my.domain
spec:
group: aerospike.dzolotov.tech
names:
kind: AerospikeCluster
listKind: AerospikeClusterList
plural: aerospikeclusters
singular: aerospikecluster
# будет локальным для namespace (может быть Cluster для создания ресурса на уровне кластера, без привязки к namespace)
scope: Namespaced
versions:
# создаваемая версия CRD
- name: v1alpha1
schema:
# описание схемы yaml-документа
openAPIV3Schema:
description: AerospikeCluster is the Schema for the aerospikeclusters API
properties:
# версия api (позволит делать разные версии для API)
apiVersion:
description: 'APIVersion defines the versioned schema of this representation
of an object. Servers should convert recognized schemas to the latest
internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources'
type: string
# тип ресурса (в нашем случае будет AerospikeCluster)
kind:
description: 'Kind is a string value representing the REST resource this
object represents. Servers may infer this from the endpoint the client
submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
type: string
# описание метаданных (.meta) - определяется по схеме metav1.ObjectMeta
metadata:
type: object
# описание спецификации кластера (.spec) - в соответствии со схемой
spec:
description: AerospikeClusterSpec defines the desired state of AerospikeCluster
properties:
# свойства спецификации кластера (.spec)
foo:
description: Foo is an example field of AerospikeCluster. Edit aerospikecluster_types.go
to remove/update
type: string
type: object
# назначение ключа объекта yaml .status (пустой объект)
status:
description: AerospikeClusterStatus defines the observed state of AerospikeCluster
type: object
type: object
# флаг актуальности версии CRD
served: true
# версия по умолчанию
storage: true
Примером корректного ресурса с использованием указанного CRD:
apiVersion: aerospike.my.domain/alpha1
kind: AerospikeCluster
metadata:
name: demo
spec:
foo: bar
status: {}Прежде чем мы пойдем рассматривать код контроллера дополним схему данных и добавим следующие поля:
deployments (int) - количество серверов в кластере;
maxMemory (int) - ограничение по размеру памяти (в Гб);
maxCPU (string) - ограничение по использованию процессора;
logging (string) - уровень протоколирования;
compressionLevel (int) - уровень сжатия;
replicationFactor (int) - фактор репликации для кластера;
strongConsistency (bool) - включение режима строгой согласованности (в ущерб производительности);
heartbeat:
interval (int) - интервал для проверки доступности соседних узлов (в миллисекундах).
keepalive:
enabled (bool) - разрешено использование keepalive-сообщений;
interval (int) - интервал в секундах между сообщениями.
Для демонстрации такого набора будет достаточно, в реальном кластере разумеется нужно предусмотреть наличие всех необходимых опций для конфигурации узлов. Структура данных для спецификации кластера определяется в файле api/alphav1/aerospikecluster_types.go, сейчас там даны определения для всех создаваемых типов данных (кластер, список кластеров и связанные с ними типы):
package v1alpha1
import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
//ожидаемое состояние кластера (спецификация)
type AerospikeClusterSpec struct {
Foo string `json:"foo,omitempty"`
}
//текущее состояние кластера
type AerospikeClusterStatus struct {
}
//+kubebuilder:object:root=true
//+kubebuilder:subresource:status
type AerospikeCluster struct {
metav1.TypeMeta `json:",inline"`
//стандартное определение метаданных (namespace, name, annotations, ...)
metav1.ObjectMeta `json:"metadata,omitempty"`
//кластер содержит спецификацию и текущий статус
Spec AerospikeClusterSpec `json:"spec,omitempty"`
Status AerospikeClusterStatus `json:"status,omitempty"`
}
//+kubebuilder:object:root=true
type AerospikeClusterList struct {
metav1.TypeMeta `json:",inline"`
metav1.ListMeta `json:"metadata,omitempty"`
Items []AerospikeCluster `json:"items"`
}
//регистрация типов объектов и их схем
func init() {
SchemeBuilder.Register(&AerospikeCluster{}, &AerospikeClusterList{})
}Добавим необходимые поля (и структуры) в структуру AerospikeClusterSpec и обозначим в статусе текущее состояние кластера: deployed (bool) - устанавливается в true когда развертывание завершено, brokenNodes (int) - количество недоступных узлов. Обновим определение типов и выполним повторно команду make в корневом каталоге проекта, это приведет к выполнению форматирования и проверки синтаксиса файлов.
type AerospikeHeartBeat struct {
// Heartbeat interval in milliseconds
Interval int `json:"interval,omitempty"`
}
type AerospikeKeepAlive struct {
// Keep alive is enabled
Enabled bool `json:"enabled,omitempty"`
// Keep alive messages interval in seconds
Interval int `json:"interval,omitempty"`
}
// ожидаемое состояние кластера (спецификация)
type AerospikeClusterSpec struct {
// Deployments count
Deployments int `json:"deployments,omitempty"`
// Maximum memory limit in GB (for example, 4)
MaxMemory *int `json:"maxMemory,omitempty"`
// Maximum cpu limit (100 = full power)
MaxCPU *int `json:"maxCPU,omitempty"`
// Logging level (https://docs.aerospike.com/reference/configuration)
Logging *string `json:"logging,omitempty"`
// Compression level (1 - lower compression, 9 - higher, but slower compression)
CompressionLevel *int `json:"compressionLevel,omitempty"`
// Min replicas count
ReplicationFactor int `json:"replicationFactor,omitempty"`
// Enable strong consistency mode
StrongConsistency bool `json:"strongConsistency,omitempty"`
// Heartbeat configuration
HeartBeat *AerospikeHeartBeat `json:"heartbeat,omitempty"`
// Keepalive configuration
Keepalive *AerospikeKeepAlive `json:"keepalive,omitempty"`
}
// текущее состояние кластера
type AerospikeClusterStatus struct {
// All nodes are prepared and ready
Deployed bool `json:"deployed"`
// How many nodes isn't available
BrokenNodes int `json:"brokenNodes"`
}После применения make manifests можем убедиться, что описание создания CRD-ресурса соответствует измененной схеме (фрагмент приведен ниже):
compressionLevel:
description: Compression level (1 - lower compression, 9 - higher,
but slower compression)
type: integer
deployments:
description: Deployments count
type: integer
heartbeat:
description: Heartbeat configuration
properties:
interval:
description: Heartbeat interval in milliseconds
type: integer
type: objectПри необходимости создать дополнительную версию существующего ресурса или создать новый можно использовать как инструмент operator-sdk create api, так и создавать дополнительные каталоги и файлы на основе существующего шаблона.
Дальше посмотрим на контроллер и тут ключевую роль играет Reconciler, который выполняет изменение конфигурации в соответствии с разностью между текущим и ожидаемым состоянием. Кроме это для проверки корректности согласованности спецификации может быть добавлено использование webhook и здесь нужно сделать небольшое введение в Dynamic Admission Control. Динамическое управление позволяет зарегистрировать два вида дополнительных обработчиков во время операций над ресурсами - ValidatingWebhook (проверяет корректность схемы создаваемого ресурса, позволяет отменить операцию при нарушении схемы) и MutatingWebhook (допускает возможность внесения изменений в ресурс перед развертыванием, например используется для встраивания sidecar-контейнеров в операторе управления резервным копированием Stash, который мы рассматривали в этой статье). Operator SDK позволяет также создавать и регистрировать webhook и даже генерировать заготовку ValidatingWebhook для проверки согласованности данных .
operator-sdk create webhook --group aerospike --version v1alpha1 --kind AerospikeCluster --defaulting --programmatic-validationПосле вызова генератора создается файл aerospikecluster_webhook.go, в котором нужно будет реализовать функции проверки конфигурации при создании ValidateCreate(), при обновлении ValidateUpdate() и при удалении ValidateDelete(). Добавим дополнительную проверку, что количество реплик не превышает количество узлов и количество узлов является натуральным числом. В результате получим следующий код:
package v1alpha1
import (
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/validation/field"
ctrl "sigs.k8s.io/controller-runtime"
logf "sigs.k8s.io/controller-runtime/pkg/log"
"sigs.k8s.io/controller-runtime/pkg/webhook"
"strconv"
)
// log is for logging in this package.
var aerospikeclusterlog = logf.Log.WithName("aerospikecluster-resource")
func (r *AerospikeCluster) SetupWebhookWithManager(mgr ctrl.Manager) error {
return ctrl.NewWebhookManagedBy(mgr).
For(r).
Complete()
}
//+kubebuilder:webhook:path=/mutate-aerospike-dzolotov-tech-v1alpha1-aerospikecluster,mutating=true,failurePolicy=fail,sideEffects=None,groups=aerospike.dzolotov.tech,resources=aerospikeclusters,verbs=create;update,versions=v1alpha1,name=maerospikecluster.kb.io,admissionReviewVersions=v1
var _ webhook.Defaulter = &AerospikeCluster{}
//здесь можно применить значения по умолчанию (выполняется до валидации)
func (r *AerospikeCluster) Default() {
aerospikeclusterlog.Info("default", "name", r.Name)
}
// TODO(user): change verbs to "verbs=create;update;delete" if you want to enable deletion validation.
//+kubebuilder:webhook:path=/validate-aerospike-dzolotov-tech-v1alpha1-aerospikecluster,mutating=false,failurePolicy=fail,sideEffects=None,groups=aerospike.dzolotov.tech,resources=aerospikeclusters,verbs=create;update,versions=v1alpha1,name=vaerospikecluster.kb.io,admissionReviewVersions=v1
var _ webhook.Validator = &AerospikeCluster{}
func (r *AerospikeCluster) validateReplicas() error {
var allErrs field.ErrorList
if r.Spec.ReplicationFactor > r.Spec.Deployments {
fldPath := field.NewPath("spec").Child("replicationFactor")
allErrs = append(allErrs, field.Invalid(fldPath, strconv.Itoa(r.Spec.ReplicationFactor), "Replication factor must be lower than deployments count"))
}
if r.Spec.Deployments < 1 {
fldPath := field.NewPath("spec").Child("deployments")
allErrs = append(allErrs, field.Invalid(fldPath, strconv.Itoa(r.Spec.ReplicationFactor), "Deployments counter must be greater than zero"))
}
if len(allErrs) == 0 {
return nil
}
return apierrors.NewInvalid(schema.GroupKind{Group: "aerospike.dzolotov.tech", Kind: "AerospikeCluster"}, r.Name, allErrs)
}
//проверка при создании нового ресурса
func (r *AerospikeCluster) ValidateCreate() error {
aerospikeclusterlog.Info("validate create", "name", r.Name)
return r.validateReplicas()
}
//проверка при обновлении ресурса
func (r *AerospikeCluster) ValidateUpdate(old runtime.Object) error {
aerospikeclusterlog.Info("validate update", "name", r.Name)
return r.validateReplicas()
}
//проверка при удалении ресурса (сейчас отключена в метаданных kubebuilder)
func (r *AerospikeCluster) ValidateDelete() error {
aerospikeclusterlog.Info("validate delete", "name", r.Name)
return nil
}Здесь мы используем механизм apierrors, рекомендуемый Kubernetes для подробного описания ошибки и места ее возникновения, который позволяет объединить все ошибки в один объект, но можно было использовать и обычный объект ошибки. Обратите внимание на комментарии //+kubebuilder:webhook, они содержат метаданные, которые используются для генерации манифеста и здесь, например, можно изменить список валидаций (сейчас используются только create,update, но при необходимости можно сделать валидацию на удаление (например, так можно запретить удаление ресурса, но лучше такого не делать). После вызова make / make manifests посмотрим на сгенерированный файл манифеста для установки AdmissionWebhooks:
apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
metadata:
creationTimestamp: null
name: mutating-webhook-configuration
webhooks:
- admissionReviewVersions:
- v1
clientConfig:
service:
name: webhook-service
namespace: system
path: /mutate-aerospike-dzolotov-tech-v1alpha1-aerospikecluster
failurePolicy: Fail
name: maerospikecluster.kb.io
rules:
- apiGroups:
- aerospike.dzolotov.tech
apiVersions:
- v1alpha1
operations:
- CREATE
- UPDATE
resources:
- aerospikeclusters
sideEffects: None
---
apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
creationTimestamp: null
name: validating-webhook-configuration
webhooks:
- admissionReviewVersions:
- v1
clientConfig:
service:
name: webhook-service
namespace: system
path: /validate-aerospike-dzolotov-tech-v1alpha1-aerospikecluster
failurePolicy: Fail
name: vaerospikecluster.kb.io
rules:
- apiGroups:
- aerospike.dzolotov.tech
apiVersions:
- v1alpha1
operations:
- CREATE
- UPDATE
resources:
- aerospikeclusters
sideEffects: NoneЗдесь можно увидеть, что и MutatingWebhook и ValidatingWebhook применяются на ресурсы в apiVersion: aerospike.dzolotov.tech/v1alpha1 при операциях создания и обновления на ресурсах aerospikecluster. Теперь посмотрим на другие подкаталоги в config и поговорим о правах доступа (RBAC) и развертывании контроллера.
certmanager- манифест для издания сертификата для валидации webhook (создается автоматически), привязывается к сервису контроллера;crd- описание типов ресурсов (CRD), мы рассмотрели его выше, но добавлю что кроме наших определений есть еще патчи, которые позволяют пробросить созданный через certmanager сертификат в аннотации ресурса для использования в коде;default- содержит патчи для контроллера, которые публикуют webhook-сервер и разворачивают system/controller-manager из gcr.io/kubebuilder/kube-rbac-proxy:v0.11.0;manifests- конфигурация для kustomization, который создает манифесты при сборке финального артефакта для установки оператора;prometheus- регистрация ServiceMonitor для экспорта метрик из controller-manager в prometheus;rbac- набор прав для обеспечения доступа контроллера к управлению ресурсами aerospikeclusters, а также для публикации метрик, сервисной записи для выполнения controller-manager и др.;scorecard- конфигурация для запуска тестов через OLM (Operator Lifecycle Manager, позволяет управлять запуском и тестирование операторов внутри существующего кластера kubernetes);webhook- конфигурация для регистрации AdminissionWebhooks (рассмотрели ранее).
Коротко об RBAC (Role-Based Access Control) - это внутренний механизм контроля доступа Kubernetes, который определяет понятие Role (или ClusterRole, если без привязки к пространству имен), перечисляющее какие действия и над какими ресурсами могут быть выполнены и RoleBinding (ClusterRoleBinding) для связывания учетной записи пользователи или сервисной записи (ServiceAccount) с соответствующей ролью. Таким образом, чтобы оператор мог создавать новые ресурсы Deployment, требуется определить роль с разрешениями на create/update/patch/delete для ресурсов kind: Deployment в apiVersion: apps/v1 и связать ее с сервисной записью, под которой происходит развертывание оператора. Для управление нашим контроллером будет запущен процесс controller-manager, он также будет отслеживать доступность контроллера и перезапускать его при необходимости.
Вернемся теперь к контроллеру (controllers/aerospikecluster_controller.go) и посмотрим внимательно на его структуру:
package controllers
import (
"context"
"k8s.io/apimachinery/pkg/runtime"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/log"
aerospikev1alpha1 "github.com/operator-framework/operator-sdk/api/v1alpha1"
)
// Объект реконсилера для приведения кластера к ожидаемому состоянию
type AerospikeClusterReconciler struct {
client.Client
Scheme *runtime.Scheme
}
// Описание разрешений (будут транслированы в роли)
//+kubebuilder:rbac:groups=aerospike.dzolotov.tech,resources=aerospikeclusters,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=aerospike.dzolotov.tech,resources=aerospikeclusters/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=aerospike.dzolotov.tech,resources=aerospikeclusters/finalizers,verbs=update
func (r *AerospikeClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
_ = log.FromContext(ctx)
//здесь будет наш код управления кластером
return ctrl.Result{}, nil
}
// Регистрация с использованием ControllerManager
func (r *AerospikeClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&aerospikev1alpha1.AerospikeCluster{}).
Complete(r)
}Поскольку мы будем управлять развертыванием кластера серверов (будем использовать StatefulSet, поскольку нам важно сохранить связь экземпляра пода и его местоположения, также нам будет нужно иметь возможность создания и изменения ConfigMap для определения конфигурации сервера по параметрам из манифеста), то нам нужно будет сообщить контроллеру о том, что мы будем наблюдать за ресурсами StatefulSet и Pod и добавить соответствующие правила в комментариях +kubebuilder. Прежде всего добавим импорт структур, описывающих пространство имен apps/v1 и core/v1 (для ConfigMap):
import (
"context"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/runtime"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/log"
aerospikev1alpha1 "github.com/operator-framework/operator-sdk/api/v1alpha1"
)И сообщим менеджеру о том, что мы будем управлять объектами StatefulSet и ConfigMap:
func (r *AerospikeClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&aerospikev1alpha1.AerospikeCluster{}).
Owns(&appsv1.StatefulSet{}).
Owns(&corev1.ConfigMap{}).
Complete(r)
}
И добавим метаданные перед методом Reconcile для описания требуемых прав доступа:
//+kubebuilder:rbac:groups=apps/v1,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=apps/v1,resources=pods,verbs=get;list
//+kubebuilder:rbac:groups=v1,resources=configmaps,verbs=get;list;create;update;patch;deleteСама реализация будет использовать публичные методы API Kubernetes, прежде всего для нас интересны методы:
r.List(ctx, objectList, opts)- возвращает список объектов в objectList с отбором по opts (требует разрешения list на этот тип объектов в RBAC-роли), например:
podList := &v1.PodList{}
opts := []client.ListOption{
client.InNamespace(request.NamespacedName.Namespace),
client.MatchingLabels{"instance": request.NamespacedName.Name},
client.MatchingFields{"status.phase": "Running"},
}
err := r.List(ctx, podList, opts...)Выбирает все запущенные поды с меткой instance и значением, совпадающих с именем кластера (поиск осуществляется в том же пространстве имен, где развернут ресурс AerospikeCluster).
r.Get(ctx, namespacedName, object)- сохраняет объект в object, идентификацию объекта выполняет NamespacedName, включающий пространство имен Namespace и имя внутри него Name (требует разрешение get на этот тип объектов в RBAC-роли);
r.Create(ctx, object, opts)- создает новый объект в кластере (требуется наличие разрешений create на этот тип объекта в RBAC-роли), например:
func (r *AerospikeClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
_ = log.FromContext(ctx)
//прочитаем ресурс описания кластера
cluster := &aerospikev1alpha1.AerospikeCluster{}
err := r.Get(ctx, req.NamespacedName, cluster)
if err != nil {
return ctrl.Result{}, err
}
//преобразование числовых полей в строки и в тип int32 для передачи в структуру
//описания StatefulSet
var deployments int32
deployments = int32(cluster.Spec.Deployments)
var replicationFactor string
replicationFactor = strconv.Itoa(cluster.Spec.ReplicationFactor)
var maxMemory string
maxMemory = strconv.Itoa(*cluster.Spec.MaxMemory)
var maxCpu string
maxCpu = strconv.Itoa(*cluster.Spec.MaxCPU)
//описание нового объекта StatefulSet
sts := appsv1.StatefulSet{ObjectMeta: ctrl.ObjectMeta{Namespace: req.NamespacedName.Namespace, Name: req.NamespacedName.Name}, Spec: appsv1.StatefulSetSpec{
Replicas: &deployments,
Selector: &v1.LabelSelector{
MatchLabels: map[string]string{"label": req.Name},
},
Template: corev1.PodTemplateSpec{
Spec: corev1.PodSpec{
Containers: []corev1.Container{
{
Name: "aerospike",
Image: "aerospike",
Args: nil,
EnvFrom: nil,
Env: []corev1.EnvVar{
{
Name: "REPL_FACTOR",
Value replicationFactor,
},
{
Name: "MEM_GB",
Value: maxMemory,
},
},
Resources: corev1.ResourceRequirements{
Limits: corev1.ResourceList{
corev1.ResourceCPU: resource.MustParse(maxCpu),
},
},
},
},
},
},
UpdateStrategy: appsv1.StatefulSetUpdateStrategy{},
}}
r.Create(ctx, &sts)
return ctrl.Result{Requeue: true}, nil
}
r.Update(ctx, object, opts)- обновление конфигурации объекта (для StatefulSet например можно изменить переменные окружения и количество реплик), требуется разрешение update на этот тип ресурсов в RBAC-роли;r.Patch(ctx, object, patch)- внесение исправлений в структуру, описывающую запущенный объект;r.Delete(ctx, objects, opts)- удаление объекта из кластера (требуется разрешение delete на этот тип ресурсов в RBAC-роли);r.Status().Update(ctx, cluster)- дает доступ к модификации объекта состояния (виден в манифесте и в объекте cluster как .status).
Поведение reconciler в нашем случае может быть описано несколькими возможными сценариями:
объект кластера существует, но нет statefulset - создаем новый statefulset и configmap;
объект кластера существует и есть statefulset - обновляем параметры конфигурации (лимиты, переменные окружения, количество реплик) и пересоздаем configmap;
объект кластера не существует - удаляем statefulset и configmap (игнорируем ошибки).
В целом код можно разделить на несколько частей: генерация и обновление configmap, обработка удаления кластера, обновление конфигурации существующего кластера. Код может выглядеть таким образом. Обновление или создание configmap:
func (r *AerospikeClusterReconciler) generateConfigMap(ctx context.Context, cluster *aerospikev1alpha1.AerospikeCluster) error {
//проверим на наличие существующего configmap
var cm corev1.ConfigMap
err := r.Get(ctx, types.NamespacedName{
Namespace: cluster.Namespace,
Name: cluster.Name,
}, &cm)
if err != nil {
if errors.IsNotFound(err) {
//создаем новый configmap
newConfigMap := corev1.ConfigMap{
ObjectMeta: v1.ObjectMeta{
Namespace: cluster.Namespace,
Name: cluster.Name,
},
Data: map[string]string{
"aerospike": createConfigMap(cluster),
},
}
err := r.Create(ctx, &newConfigMap)
return err
} else {
return err
}
}
//если объект уже существует, то изменяем данные
cm.Data = map[string]string{
"aerospike": createConfigMap(cluster),
}
err = r.Update(ctx, &cm)
return err
}
func createConfigMap(cluster *aerospikev1alpha1.AerospikeCluster) string {
result := make([]string, 0, 0)
if cluster.Spec.Logging != nil {
result = append(result, "logging "+*cluster.Spec.Logging)
}
result = append(result, "strong-consistency "+strconv.FormatBool(cluster.Spec.StrongConsistency))
if cluster.Spec.CompressionLevel != nil {
result = append(result, "compression-level "+strconv.Itoa(*cluster.Spec.CompressionLevel))
}
result = append(result, "interval "+strconv.Itoa(cluster.Spec.HeartBeat.Interval))
if cluster.Spec.Keepalive != nil {
result = append(result, "keepalive-enabled "+strconv.FormatBool(cluster.Spec.Keepalive.Enabled))
result = append(result, "keepalive-intvl "+strconv.Itoa(cluster.Spec.Keepalive.Interval))
}
return strings.Join(result, "\n")
}Подготовка объекта StatefulSet по описанию кластера:
func (r *AerospikeClusterReconciler) getConfiguration(cluster *aerospikev1alpha1.AerospikeCluster) *appsv1.StatefulSet {
var deployments int32
deployments = int32(cluster.Spec.Deployments)
var replicationFactor string
replicationFactor = strconv.Itoa(cluster.Spec.ReplicationFactor)
var maxMemory string
maxMemory = strconv.Itoa(*cluster.Spec.MaxMemory)
var maxCpu string
maxCpu = strconv.Itoa(*cluster.Spec.MaxCPU)
sts := appsv1.StatefulSet{ObjectMeta: ctrl.ObjectMeta{Namespace: cluster.Namespace, Name: cluster.Name}, Spec: appsv1.StatefulSetSpec{
Replicas: &deployments,
Selector: &v1.LabelSelector{
MatchLabels: map[string]string{"instance": cluster.Name},
},
Template: corev1.PodTemplateSpec{
ObjectMeta: ctrl.ObjectMeta{
Labels: map[string]string{"instance": cluster.Name},
},
Spec: corev1.PodSpec{
Containers: []corev1.Container{
{
Name: "aerospike",
Image: "aerospike",
Args: nil,
EnvFrom: nil,
VolumeMounts: []corev1.VolumeMount{
{
Name: "config",
MountPath: "/etc/aerospike/",
},
},
Env: []corev1.EnvVar{
{
Name: "REPL_FACTOR",
Value: replicationFactor,
},
{
Name: "MEM_GB",
Value: maxMemory,
},
},
Resources: corev1.ResourceRequirements{
Limits: corev1.ResourceList{
corev1.ResourceCPU: resource.MustParse(maxCpu),
},
},
},
},
Volumes: []corev1.Volume{
{
Name: "config",
VolumeSource: corev1.VolumeSource{
ConfigMap: &corev1.ConfigMapVolumeSource{
LocalObjectReference: corev1.LocalObjectReference{Name: cluster.Name},
Items: []corev1.KeyToPath{
{
Key: "aerospike",
Path: "aerospike.conf",
},
},
},
},
},
},
},
},
UpdateStrategy: appsv1.StatefulSetUpdateStrategy{},
}}
return &sts
}Логика удаления, создания и обновления ресурсов кластера и его состояния:
func (r *AerospikeClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
_ = log.FromContext(ctx) //можно сохранить в logger и использовать для сообщений об ошибках или диагностики
cluster := &aerospikev1alpha1.AerospikeCluster{}
err := r.Get(ctx, req.NamespacedName, cluster)
if err != nil {
if errors.IsNotFound(err) {
//кластер был удален, удалим развертывание
sts := &appsv1.StatefulSet{}
err = r.Get(ctx, types.NamespacedName{Name: cluster.Name, Namespace: cluster.Namespace}, sts)
if err == nil {
err = r.Delete(ctx, sts)
if err == nil {
cm := &corev1.ConfigMap{}
err = r.Get(ctx, types.NamespacedName{Name: cluster.Name, Namespace: cluster.Namespace}, cm)
}
}
}
//и завершаем reconfile успешно (или с ошибкой)
return ctrl.Result{}, err
}
// Check if the deployment already exists, if not create a new deployment.
found := &appsv1.StatefulSet{}
err = r.Get(ctx, types.NamespacedName{Name: cluster.Name, Namespace: cluster.Namespace}, found)
if err != nil {
if errors.IsNotFound(err) {
// существующего развертывания нет - создаем новое
err2 := r.generateConfigMap(ctx, cluster) //создаем configmap
if err2 != nil {
return ctrl.Result{}, err
}
dep := r.getConfiguration(cluster)
if err = r.Create(ctx, dep); err != nil {
return ctrl.Result{}, err
}
cluster.Status.BrokenNodes = 0
cluster.Status.Deployed = true
r.Status().Update(ctx, cluster)
return ctrl.Result{Requeue: true}, nil
} else {
return ctrl.Result{}, err
}
}
deployments := int32(cluster.Spec.Deployments)
//здесь можно сравнить текущий и предыдущий размер кластера и изменить конфигурацию, если требуется
//но у нас могут поменяться обновременно и переменные окружения и configmap
found.Spec.Replicas = &deployments
var replicationFactor string
replicationFactor = strconv.Itoa(cluster.Spec.ReplicationFactor)
var maxMemory string
maxMemory = strconv.Itoa(*cluster.Spec.MaxMemory)
found.Spec.Template.Spec.Containers[0].Env = []corev1.EnvVar{
{
Name: "REPL_FACTOR",
Value: replicationFactor,
},
{
Name: "MEM_GB",
Value: maxMemory,
},
}
found.Spec.Template.Spec.Containers[0].Resources.Limits.Cpu().Set(int64(*cluster.Spec.MaxCPU))
err = r.generateConfigMap(ctx, cluster)
if err != nil {
return ctrl.Result{}, err
}
// получим активное количество подов и обновим BrokenNodes в состоянии
podList := &corev1.PodList{}
listOpts := []client.ListOption{
client.InNamespace(cluster.Namespace),
client.MatchingLabels{"instance": cluster.Name},
client.MatchingFields{"status.phase": "Running"},
}
if err = r.List(ctx, podList, listOpts...); err != nil {
return ctrl.Result{}, err
}
// обновим количество неактивных узлов
cluster.Status.BrokenNodes = cluster.Spec.Deployments - len(podList.Items)
err2 := r.Status().Update(ctx, cluster)
return ctrl.Result{}, err2
}Для применения созданного оператора можно использовать make deploy в корневом каталоге проекта или через OLM:
operator-sdk olm install
make bundle IMG="dzolotov.tech/aerospike-operator:v0.0.1"
make bundle-build bundle-push BUNDLE_IMG="dzolotov.tech/aerospike-operator-bundle:v0.0.1"
operator-sdk run bundle dzolotov.tech/aerospike-operator-bundle:v0.0.1Здесь будет собран образ контейнера (его можно отправить, например, в Docker hub) и запущен оператор на текущем кластере Kubernetes. Для остановки оператора применяются действия make undeployили operator-sdk cleanup aerospike-operator (для OLM).
Проверить работу оператора можно через создание ресурса, который был определен ранее:
cat <<< EOF | kubectl apply -f -
apiVersion: dzolotov.tech/v1alpha1
kind: AerospikeCluster
metadata:
namespace: cluster
name: demo
spec:
deployments: 2
maxMemory: 2
compressionLevel: 4
replicationFactor: 1
strongConsistency: false
heartbeat:
interval: 10000
EOFС использованием операторов возможно реализовать сложную логику по управлению распределенными системами или обеспечить возможность конфигурирования разворачиваемых приложений в терминах предметной области. Большое количество операторов можно найти на https://operatorhub.io/, а документацию и статьи по разработке в этой коллекции ссылок.
Также хочу пригласить всех заинтересованных на бесплатный вебинар, в рамках которого мы комплексно рассмотрим основные векторы по обеспечению безопасности kubernetes кластера и подробно остановимся на каждом из них. Затронем тему безопасности docker-образов, безопасность в рантайме, network и application security.
Зарегистрироваться на вебинар
