Elasticsearch Operator 개발하기

이번 글에서는 Operator SDK를 통해 Elasticsearch Cluster를 편리하게 관리할 수 있는 Elasticsearch Operator를 만들어보겠습니다. Operator가 무엇인지, Operator SDK에 대한 기본적인 사용방법이 궁금하신 분께서는 이전 글을 참고해주시면 감사하겠습니다.

Elasticsearch Cluster

먼저 Operator로 관리할 Elasticsearch Operator에 대한 스펙을 간단히 알아보겠습니다. Cluster 구성은 제가 이전에 작성했던 Elasticsearch Helm Chart의 구성을 사용하였습니다.

• Master Node & Master Eligible Node
  마스터 노드와 마스터 노드가 다운됐을 때 승격될 수 있는 후보 노드입니다.

• Hot Data Node
  Elasticsearch의 Hot-Warm 아키텍쳐를 구축하기 위해 Node의 디스크 타입으로 Data Node를 나누었습니다. Hot Data Node는 빠른 Disk I/O 성능이 필요하기 때문에 SSD를 가진 노드에만 배치되어야 합니다.

• Warm Data Node
  Warm Data Node는 Disk I/O 성능이 중요하지 않기 때문에 가격이 비싼 SSD를 사용하지 않아도 됩니다. 따라서 HDD를 가진 노드에만 배치되어야 합니다.

• Coordinating Node
  요청 및 데이터를 라우팅하는 기능을 하는 노드입니다.

• Cerebro
  Elasticsearch Cluster 및 Index 관리를 위한 도구입니다.

• Kibana
  Elasticsearch에 저장된 데이터를 시각화하기 위한 도구입니다.

Elasticsearch Cluster에 대해 궁금하신 분들은 Elasticsearch 문서를 참고하여 자세한 노드의 역할을 확인하실 수 있습니다.

Elasticsearch Operator

CRD API

먼저 CRD를 생성해야 합니다. Elasticsearch CRD가 가지게 될 데이터는 api/v1alpha1/elasticsearch_types.go 에 존재합니다.

package v1alpha1

import (
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// ElasticsearchStatus defines the observed state of Elasticsearch
type ElasticsearchStatus struct {
    // INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
    // Important: Run "make" to regenerate code after modifying this file
    State   string `json:"state,omitempty"`
    Message string `json:"message,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status

// Elasticsearch is the Schema for the elasticsearches API
type Elasticsearch struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   ElasticsearchSpec   `json:"spec,omitempty"`
    Status ElasticsearchStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true

// ElasticsearchList contains a list of Elasticsearch
type ElasticsearchList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []Elasticsearch `json:"items"`
}

// ElasticsearchSpec defines the desired state of Elasticsearch
type ElasticsearchSpec struct {
    // INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
    // Important: Run "make" to regenerate code after modifying this file
    //Master Node Replicas
    MasterReplicas int32 `json:"master-replicas"`

    //Client Node Replicas
    ClientReplicas int32 `json:"client-replicas"`

    //Hot Data Node Replica
    HotDataReplicas int32 `json:"hot-data-replicas"`

    //Warm Data Node Replica
    WarmDataReplicas int32 `json:"warm-data-replicas"`

    //NodeSelector for the pod to be eligible to run on a node (ex. Hot-Warm Architecture)
    NodeSelector map[string]string `json:"nodeSelector,omitempty"`

    //Annotations
    Annotations map[string]string `json:"annotations,omitempty"`

    //Disk Size of Hot Data Node
    HotDataDiskSize string `json:"hot-data-volume"`

    //Disk Size of Warm Data Node
    WarmDataDiskSize string `json:"warm-data-volume"`

    //Elasticsearch Image
    ElasticsearchImage string `json:"elasticsearch-image"`

    //Elasticsearch Cluster Name
    ElasticsearchClusterName string `json:"elasticsearch-cluster-name"`
    //MasterJavaOpt
    MasterJavaOpts string `json:"master-javaOpts"`

    //ClientJavaOpt
    ClientJavaOpts string `json:"client-javaOpts"`

    //HotDataJavaOpt
    HotDataJavaOpts string `json:"hot-data-javaOpts"`

    //WarmDataJavaOpt
    WarmDataJavaOpts string `json:"warm-data-javaOpts"`

    //Cerebro
    Cerebro Cerebro `json:"cerebro"`

    //Kibana
    Kibana Kibana `json:"kibana"`

    //Curator
    Curator Curator `json:"curator"`
}

// Kibana properties (Optional)
type Kibana struct {
    // Defines the image to use for deploying kibana
    Image string `json:"image"`
}

// Cerebro properties (Optional)
type Cerebro struct {
    // Defines the image to use for deploying Cerebro
    Image string `json:"image"`
}

// Curator properties (Optional)
type Curator struct {
    // Defines the image to use for deploying Curator
    Image string `json:"image"`
}

func init() {
    SchemeBuilder.Register(&Elasticsearch{}, &ElasticsearchList{})
}

Elasticsearch Cluster의 Spec은 다음과 같은 정보를 포함합니다.

• Elasticsearch Image
• Node의 Replica 수 (Master / Hot / Warm / Coordinating)
• Data Node Disk 크기 (Hot / Warm)
• Java Options (Master / Hot / Warm / Coordinating)
• Cerebro Image
• Kibana Image

위의 정보를 사용해 생성된 CRD는 다음과 같습니다. (config/crd/bases 폴더 아래 존재합니다.)

---
apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
  annotations:
    controller-gen.kubebuilder.io/version: v0.3.0
  creationTimestamp: null
  name: elasticsearches.sphong.com.my.domain
spec:
  group: sphong.com.my.domain
  names:
    kind: Elasticsearch
    listKind: ElasticsearchList
    plural: elasticsearches
    singular: elasticsearch
  scope: Namespaced
  subresources:
    status: {}
  validation:
    openAPIV3Schema:
      description: Elasticsearch is the Schema for the elasticsearches API
      properties:
        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
        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
        metadata:
          type: object
        spec:
          description: ElasticsearchSpec defines the desired state of Elasticsearch
          properties:
            annotations:
              additionalProperties:
                type: string
              description: Annotations
              type: object
            cerebro:
              description: Cerebro
              properties:
                image:
                  description: Defines the image to use for deploying Cerebro
                  type: string
              required:
              - image
              type: object
            client-javaOpts:
              description: ClientJavaOpt
              type: string
            client-replicas:
              description: Client Node Replicas
              format: int32
              type: integer
            curator:
              description: Curator
              properties:
                image:
                  description: Defines the image to use for deploying Curator
                  type: string
              required:
              - image
              type: object
            elasticsearch-cluster-name:
              description: Elasticsearch Cluster Name
              type: string
            elasticsearch-image:
              description: Elasticsearch Image
              type: string
            hot-data-javaOpts:
              description: HotDataJavaOpt
              type: string
            hot-data-replicas:
              description: Hot Data Node Replica
              format: int32
              type: integer
            hot-data-volume:
              description: Disk Size of Hot Data Node
              type: string
            kibana:
              description: Kibana
              properties:
                image:
                  description: Defines the image to use for deploying kibana
                  type: string
              required:
              - image
              type: object
            master-javaOpts:
              description: MasterJavaOpt
              type: string
            master-replicas:
              description: 'INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
                Important: Run "make" to regenerate code after modifying this file
                Master Node Replicas'
              format: int32
              type: integer
            nodeSelector:
              additionalProperties:
                type: string
              description: NodeSelector for the pod to be eligible to run on a node
                (ex. Hot-Warm Architecture)
              type: object
            warm-data-javaOpts:
              description: WarmDataJavaOpt
              type: string
            warm-data-replicas:
              description: Warm Data Node Replica
              format: int32
              type: integer
            warm-data-volume:
              description: Disk Size of Warm Data Node
              type: string
          required:
          - cerebro
          - client-javaOpts
          - client-replicas
          - curator
          - elasticsearch-cluster-name
          - elasticsearch-image
          - hot-data-javaOpts
          - hot-data-replicas
          - hot-data-volume
          - kibana
          - master-javaOpts
          - master-replicas
          - warm-data-javaOpts
          - warm-data-replicas
          - warm-data-volume
          type: object
        status:
          description: ElasticsearchStatus defines the observed state of Elasticsearch
          properties:
            message:
              type: string
            state:
              description: 'INSERT ADDITIONAL STATUS FIELD - define observed state
                of cluster Important: Run "make" to regenerate code after modifying
                this file'
              type: string
          type: object
      type: object
  version: v1alpha1
  versions:
  - name: v1alpha1
    served: true
    storage: true
status:
  acceptedNames:
    kind: ""
    plural: ""
  conditions: []
  storedVersions: []

Controller

다음으로 Controller에 대해서 살펴보겠습니다. 먼저 SetupWithManager 메서드는 다음과 같습니다.

func (r *ElasticsearchReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&sphongcomv1alpha1.Elasticsearch{}).
        Owns(&v1.Deployment{}).
        Owns(&v1.StatefulSet{}).
        WithOptions(controller.Options{
            MaxConcurrentReconciles: 3,
        }).Complete(r)
}

• 튜토리얼과 유사하지만 Elasticsearch Operator에서 관찰할 리소스가 Deployment 이외에도 StatefulSet도 필요하기 때문에 Owns(&v1.StatefulSet{}) 을 추가하였습니다.

생성한 Elasticsearch CR에 대해 Reconcile Loop를 수행하는 곳은 Reconcil 함수입니다.

//Fetch the Elasticsearch Cluster
    elasticsearch := &sphongcomv1alpha1.Elasticsearch{}
    err := r.Get(ctx, req.NamespacedName, elasticsearch)
    if err != nil {
        if errors.IsNotFound(err) {
            // Case 1
            log.Info("Elasticsearch Resource Not Found!. Ignore since object must be deleted")
            return ctrl.Result{}, nil
        }
        // Case 2
        log.Error(err, "Failed to get Elasticsearch")
        return ctrl.Result{}, err
    }

• Case 1의 경우는 Request가 발생하였는데 찾을 수 없는 경우 즉, 삭제된 경우를 의미한다. 기존 존재하던 리소스는 자동으로 GC되고 필요하다면 이 부분에 별도의 삭제 로직을 구현할 수 있다. 삭제 같은 경우, 이벤트 큐에 다시 Request를 넣지 않으므로 nil을 반환한다.

• Case 2의 경우는 Request를 정상적으로 읽지 못했을 때 발생하는 에러 상황입니다. 이 경우에는 Request를 이벤트 큐에 다시 넣어줘야 하므로 err을 반환합니다.

다음 코드는 Elasticsearch Cluster에 필요한 리소스들이 생성되어 있는지 확인하는 부분입니다.

...

//Create Discovery Service
    foundMasterSvc := &v12.Service{}
    err = r.Get(ctx, types.NamespacedName, foundMasterSvc)

    if err != nil && errors.IsNotFound(err) {
        creationErr := r.createMasterService(elasticsearch)
        if creationErr != nil {
            return reconcile.Result{}, err
        }
    }

...

• 만약 존재하지 않는다면 새로운 리소스를 생성합니다. 만약 생성 중 에러가 발생하게 되면 request를 이벤트 큐에 다시 넣어줘야 하므로 creationErr를 반환합니다.

다음 부분은 Spec과 Status가 동일한지 확인하는 부분입니다.

//scaling status's size => spec's size
    hotDataSize := elasticsearch.Spec.HotDataReplicas
    if *foundHotData.Spec.Replicas != hotDataSize {
        foundHotData.Spec.Replicas = &hotDataSize
        err = r.Client.Update(context.TODO(), foundHotData)
        if err != nil {
            log.Error(err, "Failed to update Deployment's Size (Hot Data Node Replica)")
            return reconcile.Result{}, err
        }
    }

    warmDataSize := elasticsearch.Spec.WarmDataReplicas
    if *foundWarmData.Spec.Replicas != warmDataSize {
        foundWarmData.Spec.Replicas = &warmDataSize
        err = r.Client.Update(context.TODO(), foundWarmData)
        if err != nil {
            log.Error(err, "Failed to update Deployment's Size")
            return reconcile.Result{}, err
        }
    }

• 제가 만든 Elasticsearch Operator에서는 Hot / Warm Data Node 수만 검사하여 Spec과 동일한지 확인합니다. 만약 동일하지 않다면 Request를 이벤트 큐에 다시 넣게 됩니다.

마지막으로 Hot/ Warm Data Node의 Status를 각 Pod의 이름으로 갱신합니다.

hotPodList := &v12.PodList{}
    hotListOpts := []client.ListOption{
        client.InNamespace(elasticsearch.Namespace),
        client.MatchingLabels(labelsForWarmData()),
    }
    if err = r.List(ctx, hotPodList, hotListOpts...); err != nil {
        log.Error(err, "Failed to list pods")
        return ctrl.Result{}, err
    }

    warmPodList := &v12.PodList{}
    warmListOpts := []client.ListOption{
        client.InNamespace(elasticsearch.Namespace),
        client.MatchingLabels(labelsForWarmData()),
    }
    if err = r.List(ctx, warmPodList, warmListOpts...); err != nil {
        log.Error(err, "Failed to list pods")
        return ctrl.Result{}, err
    }
    return ctrl.Result{}, nil

Deploy Operator

이번에는 로컬 클러스터가 아닌 외부 클러스터에 operator를 배포해보겠습니다. 저는 GKE를 사용하여 배포를 진행하였습니다.

먼저, 도커 이미지를 빌드하고 GCS에 푸시합니다.

> make docker-build IMG=<GCS Image> 
> make docker-push IMG=<GCS Image>

다음으로 kustomize에 namespace에 대한 설정을 배포할 namespace로 변경해줍니다. 저는 default namespace에 배포를 진행하였습니다.

> cd config/default/ && kustomize edit set namespace "default" && cd ../..

마지막으로 배포를 진행합니다.

> make deploy IMG=<GCS Image>

kubectl로 확인해보면 elasticsearch operator가 정상적으로 동작하는 것을 알 수 있습니다.

> kubectl get deployment
NAME                                        READY   UP-TO-DATE   AVAILABLE   AGE
elasticsearch-operator-controller-manager   1/1     1            1           1m

Create Elasticsearch CR

config/samples/폴더 아래에 존재하는 CR을 업데이트 한 후 적용해보겠습니다. 먼저 원하는 클러스터 스펙을 입력합니다.

apiVersion: sphong.com.my.domain/v1alpha1
kind: Elasticsearch
metadata:
  name: elasticsearch-sample
  namespace: default
spec:
  cerebro:
    image: asia.gcr.io/--/cerebro:6.8.2
  client-javaOpts: ""
  client-replicas: 1
  curator:
    image: asia.gcr.io/--/cerebro:6.8.2
  elasticsearch-cluster-name: es-cluster
  elasticsearch-image: asia.gcr.io/--/elasticsearch-base:6.8.2
  hot-data-javaOpts: ""
  hot-data-replicas: 1
  hot-data-volume: 1Gi
  kibana:
    image: asia.gcr.io/--/kibana:6.8.2
  master-javaOpts: ""
  master-replicas: 3
  warm-data-javaOpts: ""
  warm-data-replicas: 1
  warm-data-volume: 1Gi

kubectl을 통해 해당 CR을 생성합니다.

$ kubectl apply -f config/samples/elasticsearch-customresource.yaml

마지막으로 생성된 Elasticsearch Cluster를 확인합니다.

$ kubectl get all
NAME                                        READY   UP-TO-DATE   AVAILABLE   AGE
elasticsearch-operator-controller-manager   1/1     1            1           8m
elasticsearch-master                        3/3     3            3           1m
elasticsearch-hot-data-0                    3/3     3            3           1m
...

이상으로 Elasticsearch Cluster를 보다 손쉽게 관리할 수 있고 간단하게 Hot / Warm Data Node 수의 변경을 감지하는 Elasticsearch Operator를 개발하였습니다.

전체 코드는 Github에 존재하며 글의 내용 중 틀린 부분이나 더 좋은 의견이 있다면 언제든지 피드백 부탁드립니다. :)

참고 자료

• https://m.blog.naver.com/alice_k106/221586279079
• https://sdk.operatorframework.io/docs/building-operators/golang/tutorial/