Kubernetes Operator

Kubernetes Operator and Controller Pattern 

Kubernets Operator는 Controller 패턴을 따릅니다.

Controller는 적어도 하나의 쿠버네티스 리소스 타입을 추적합니다. 해당 리소스에 대한 컨트롤러는 현재 상태를 desired 상태에 더 가깝게 만드는 역할을 합니다.  컨트롤러는 작업 자체도 수행할 수 있는데 Kubernetes에서 컨트롤러는 유용한 side effect가 있는 메시지를 API 서버에 보냅니다. Kubernetes에서 컨트롤러는 API 서버를 통해 클러스터의 shared state를 감시하고 current state를 desired  state로 이동하려고 시도하는 변경을 하는 control-loop 입니다. 

 

Kubernetes에는 다양한 Controller가 존재합니다. 예를 들어, Node의 up/down을 알리고 대응하는 Node Controller, Job 오브젝트를 관찰하다가 API Server에 명령을 내려 파드를 만들어 실행하는 Job Controller, EndpointSlice Controller, ServiceAccount Controller 등이 있습니다. 

Kubernetes Operator 패턴 개념을 사용하면 컨트롤러를 하나 이상의 Custom Resource 에 연결하여 Kubernetes 자체의 코드를 수정하지 않고도 클러스터의 동작을 확장할 수 있습니다. Operators는 Custom Resource의 컨트롤러 역할을 하는 Kubernetes API의 클라이언트입니다.

Controller는 감시 대상 object를 관찰하며 reconcile 루프를 사용하여 desired state를 current state와 지속적으로 비교합니다.

 

기술적으로, 일반적인 Controller와 Operator 사이에는 차이가 없습니다. 종종 언급되는 차이점은 Operator에 포함된 작동 지식입니다. 따라서, Custom Resource가 생성될 때 Pod가 Spin up하고 이후 Pod가 destroy되는 컨트롤러를 단순히 Controller라고 할 수 있습니다. Controller가 오류를 업그레이드하거나 해결하는 방법과 같은 추가적인 작동 지식을 가지고 있는 경우 Operator로 분류하곤 합니다.

 

Custom Resource & Custom Resource Description

Custom Resource는 기본 Kubernetes API의 확장으로 Kubernetes에서 구조화된 데이터를 저장하고 검색하는 데 사용됩니다. Operator의 경우, Custom Resource는 리소스의 desired state를 포함하지만 구현 로직은 포함하지 않습니다. CRD는 이러한 오브젝트의 외부(예: 어떤 필드가 있는지, CRD의 이름은 어떻게 지정되는지)를 정의합니다. 이러한 CRD는 (operator SDK와 같은) 도구를 사용하여 만들거나 손으로 작성할 수 있습니다.

 

Deep dive into Kubernetes Controller

컨트롤러에는 두 가지 주요 구성 요소가 있습니다. Informer/SharedInformer 그리고 Workqueue 입니다. Informer/SharedInformer는 Kubernetes 오브젝트의 현재 상태에 대한 변경 사항을 모니터링하고 Workqueue로 이벤트를 전송하며, 여기서 worker(s)는 이벤트를 pop하여 처리할 수 있습니다.

Informer

Kubernetes Controller의 중요한 역할은 오브젝트에서 desired state와 current state를 관찰한 다음 current state가 desired state와 더 비슷해지도록 명령을 보내는 것입니다. 오브젝트의 정보를 검색하기 위해 컨트롤러는 요청을 Kubernetes API 서버로 보냅니다.

그러나 API server에서 정보를 반복적으로 검색하는 것은 비용이 많이 들 수 있어, 코드에서 객체를 여러 번 가져 와서 list up하기 위해 Kubernetes 개발자는 client-go 라이브러리에서 이미 제공하고 있는 캐시를 사용하게됩니다. 또한 컨트롤러도 실제로 request를 계속 보내고 싶진 않고, 오브젝트가 생성, 수정 또는 삭제된 경우의 이벤트에만 관심이 있습니다. client-go 라이브러리는 initial list 을 가져오고 특정 리소스 watch를 시작하는 Listwatcher 인터페이스를 제공합니다.

lw := cache.NewListWatchFromClient(
      client,
      &v1.Pod{},
      api.NamespaceAll,
      fieldSelector)

 

이 모든 것들이 Informer에서 consume됩니다. Informer의 일반적인 구조는 아래와 같습니다.

store, controller := cache.NewInformer {
    &cache.ListWatch{},
    &v1.Pod{},
    resyncPeriod,
    cache.ResourceEventHandlerFuncs{},

현재는 SharedInformer가 Informer 대신 쓰이고 있습니다.

ListWatcher

Listwatcher는 특정 namespace에서 특정 리소스에 대한 list function와 watch function의 조합입니다. 이를 통해 컨트롤러가 보고 싶은 특정 리소스에만 집중할 수 있습니다. field selector는 컨트롤러가 특정 필드와 일치하는 리소스를 검색하려는 것 같이 리소스 검색 결과를 좁히는 filter 타입 입니다. Listwatcher의 구조는 다음과 같습니다.

cache.ListWatch {
    listFunc := func(options metav1.ListOptions) (runtime.Object, error) {
        return client.Get().
            Namespace(namespace).
            Resource(resource).
            VersionedParams(&options, metav1.ParameterCodec).
            FieldsSelectorParam(fieldSelector).
            Do().
            Get()
    }
    watchFunc := func(options metav1.ListOptions) (watch.Interface, error) {
        options.Watch = true
        return client.Get().
            Namespace(namespace).
            Resource(resource).
            VersionedParams(&options, metav1.ParameterCodec).
            FieldsSelectorParam(fieldSelector).
            Watch()
    }
}

Resource Event Handler

Resource Event Handler는 컨트롤러가 특정 리소스의 변경 사항에 대한 알림을 처리하는 곳입니다.

type ResourceEventHandlerFuncs struct {
    AddFunc    func(obj interface{})
    UpdateFunc func(oldObj, newObj interface{})
    DeleteFunc func(obj interface{})
}

- AddFunc: 새 리소스가 생성되면 호출됩니다.

- UpdateFunc: 기존 리소스가 수정되면 호출됩니다. re-synchronization과정에서도 호출되며 아무일이 없어도 수행됩니다.

- DeleteFunc: 기존 리소스가 삭제되면 호출됩니다. 

ResyncPeriod

Resync period는 컨트롤러가 캐시에 남아 있는 모든 항목을 무시하고 UpdateFunc를 다시 실행하는 빈도를 정의합니다. 이것은 현재 상태를 주기적으로 확인하고 원하는 상태처럼 만들기 위한 일종의 configuration을 제공합니다.이 기능은 컨트롤러가 업데이트를 누락했거나 이전 작업이 실패한 경우에 매우 유용합니다. 그러나 custom controller를 구축하는 경우 period가 너무 짧으면 CPU load를 주의해야 합니다.

SharedInformer

Informer는 자체적으로만 사용되는 리소스 set에 대해서만 로컬 캐시를 구성합니다. 그러나 Kubernetes에는 여러 종류의 리소스를 실행하고 관리하는 controller bundle이 있습니다. 즉, 하나의 리소스가 둘 이상의 컨트롤러에 의해 관리되고 있을 수 있음을 의미합니다.

이 경우 SharedInformer는 컨트롤러 간에 single shared cache를 만드는 데 도움이됩니다. 즉, 캐시된 리소스가 복제되지 않으므로 시스템의 메모리 부하가 줄어듭니다. 게다가 각 SharedInformer는 얼마나 많은 downstream consumer가 informer의 이벤트를 읽고 있는지에 관계없이 upstream server에서 단일 watch만 만듭니다. 이렇게 하면 업스트림 서버의 부하도 줄어듭니다. 이는 많은 내부 컨트롤러가 있는 kube-controller-manager 에서 일반적입니다.

SharedInformer는 특정 리소스를 추가, 업데이트 및 삭제하는 알림을 수신하기 위한 후크를 이미 제공했습니다. 또한 shared cache에 액세스하고 캐시가 준비되는 시기를 결정하는 편리한 기능을 제공합니다. 이를 통해 API 서버에 대한 연결, 서버 측 중복 직렬화 비용, 컨트롤러 측 중복 역직렬화 비용 및 컨트롤러 측 중복 캐싱 비용을 절감할 수 있습니다.

lw := cache.NewListWatchFromClient(…)
sharedInformer := cache.NewSharedInformer(lw, &api.Pod{}, resyncPeriod)

Workqueue

SharedInformer는 공유되고 있기 때문에 각 컨트롤러의 위치를 추적할 수 없으므로 컨트롤러가 자체 큐잉 및 재시도 메커니즘을 제공해야 합니다. 따라서 대부분의 Resource Event Handlers는 단순히 per-consumer workqueue에 items을 배치합니다.

리소스가 변경될 때마다 Resource Event Handler는 workqueue에 key를 push합니다. 키는 <resource_namespace>/<resource_name> 형식을 사용합니다. <resource_namespace>가 비어 있으면 <resource_name> 형식입니다. 이렇게하면 이벤트가 키로 모아져 각 소비자가 worker(s)를 사용하여 키를 pop할 수 있으며 이 작업은 순차적으로 수행됩니다. 이것은 두 명의 workers가 동시에 같은 키에서 작업을 수행하지 않을 것을 보장합니다.

Workqueue는 client-go 라이브러리의 client-go/util/workqueue에서 제공됩니다. 지연 대기열, 시간 제한 대기열 및 속도 제한 대기열을 포함하여 여러 종류의 대기열이 지원됩니다.

속도 제한 대기열 예시:

queue :=
workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())

Workqueue에서 key의 수명주기:

workqueue에서 key를 완전히 삭제하려면 controller는 Done() function을 트리거해야합니다.

컨트롤러가 workqueue를 처리하는 workers를 언제부터 시작해야 할까요? 컨트롤러가 최신 상태를 달성하기 위해 캐시가 완전히 동기화될 때까지 기다려야 하는 두 가지 이유가 있습니다.

1. 캐시가 동기화 완료될 때까지 모든 리소스를 list-up하는 것은 정확하지 않습니다. 
2. 단일 리소스에 대한 여러 개의 빠른 업데이트가 캐시/큐에 의해 최신 버전으로 수렴됩니다. 따라서 중간 상태에서 낭비되는 작업을 피하기 위해 실제로 항목을 처리하기 전에 캐시가 idle 상태가 될 때까지 기다려야 합니다.

controller.informer = cache.NewSharedInformer(...)
controller.queue = workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())

controller.informer.Run(stopCh)

if !cache.WaitForCacheSync(stopCh, controller.HasSynched)
{
    log.Errorf("Timed out waiting for caches to sync"))
}

// Now start processing
controller.runWorker()

Java kubernetes client

go로는 설명이 잘 되어있는 것 같아 Java에서는 어떻게 구현했는지 궁금해 찾아보았습니다.

아래와 같이 example에서는  sharedInformerFactory를 생성 후 node-informer를 등록합니다.

// instantiating an informer-factory, and there should be only one informer-factory
    // globally.
    SharedInformerFactory informerFactory = new SharedInformerFactory();
    // registering node-informer into the informer-factory.
    SharedIndexInformer<V1Node> nodeInformer =
        informerFactory.sharedIndexInformerFor(
            (CallGeneratorParams params) -> {
              return coreV1Api.listNodeCall(
                  null,
                  null,
                  null,
                  null,
                  null,
                  null,
                  params.resourceVersion,
                  null,
                  params.timeoutSeconds,
                  params.watch,
                  null);
            },
            V1Node.class,
            V1NodeList.class);
    informerFactory.startAllRegisteredInformers();

https://github.com/kubernetes-client/java/blob/master/util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java#L59 에서 SharedInformerFactory 생성자 내부에서 cachedThreadPool을 생성해 반환합니다. sharedIndexInformerFor를 통해 listwatcher도 생성합니다.

  /** Constructor w/ api client specified and default thread pool. */
  public SharedInformerFactory(ApiClient apiClient) {
    this(apiClient, Executors.newCachedThreadPool());
  }
  
  public synchronized <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
      SharedIndexInformer<ApiType> sharedIndexInformerFor(
          GenericKubernetesApi<ApiType, ApiListType> genericKubernetesApi,
          Class<ApiType> apiTypeClass,
          long resyncPeriodInMillis,
          String namespace) {
    ListerWatcher<ApiType, ApiListType> listerWatcher =
        listerWatcherFor(genericKubernetesApi, namespace);
    return sharedIndexInformerFor(listerWatcher, apiTypeClass, resyncPeriodInMillis);
  }

그 후 example 코드로 돌아가보면

// Use builder library to construct a default controller.
    Controller controller =
        ControllerBuilder.defaultBuilder(informerFactory)
            .watch(
                (workQueue) ->
                    ControllerBuilder.controllerWatchBuilder(V1Node.class, workQueue)
                        .withWorkQueueKeyFunc(
                            (V1Node node) ->
                                new Request(node.getMetadata().getName())) // optional, default to
                        .withOnAddFilter(
                            (V1Node createdNode) ->
                                createdNode
                                    .getMetadata()
                                    .getName()
                                    .startsWith("docker-")) // optional, set onAdd filter
                        .withOnUpdateFilter(
                            (V1Node oldNode, V1Node newNode) ->
                                newNode
                                    .getMetadata()
                                    .getName()
                                    .startsWith("docker-")) // optional, set onUpdate filter
                        .withOnDeleteFilter(
                            (V1Node deletedNode, Boolean stateUnknown) ->
                                deletedNode
                                    .getMetadata()
                                    .getName()
                                    .startsWith("docker-")) // optional, set onDelete filter
                        .build())
            .withReconciler(nodeReconciler) // required, set the actual reconciler
            .withName("node-printing-controller") // optional, set name for controller
            .withWorkerCount(4) // optional, set worker thread count
            .withReadyFunc(nodeInformer::hasSynced) // optional, only starts controller when the
            // cache has synced up
            .build();

Resource Event Handler에서 봤듯이 add, update, delete function이 workqueue와 함께 ControllerBuilder에 등록됩니다.

https://github.com/kubernetes-client/java/blob/master/extended/src/main/java/io/kubernetes/client/extended/controller/builder/DefaultControllerBuilder.java#L71 에서 볼 수 있듯이 watch하는 로직은 다양하고 위의 go 코드와 유사합니다.  resource에 대한 watch 오브젝트를 만들고 아마 이후 생성된 Request는 watchqueue에 들어가게 될 것입니다.

/**
   * Starts building watches over resource.
   *
   * @param <ApiType> the type parameter for the singular response
   * @param controllerWatchGetter the controller watch getter
   * @return the controller builder . controller watch builder
   */
  public <ApiType extends KubernetesObject> DefaultControllerBuilder watch(
      Function<WorkQueue<Request>, ControllerWatch<ApiType>> controllerWatchGetter) {
    ControllerWatch<ApiType> watch = controllerWatchGetter.apply(this.workQueue);
    Class<ApiType> apiTypeClass = watch.getResourceClass();
    SharedIndexInformer<ApiType> informer =
        informerFactory.getExistingSharedIndexInformer(apiTypeClass);
    if (informer == null) {
      throw new IllegalStateException(
          String.format(
              "Missing informer for resource %s, "
                  + "check if informer already constructed in the informerFactory",
              apiTypeClass));
    }
    informer.addEventHandlerWithResyncPeriod(
        watch.getResourceEventHandler(), watch.getResyncPeriod().toMillis());
    return this;
  }

 

 

 

참고자료

- https://github.com/cncf/tag-app-delivery/blob/eece8f7307f2970f46f100f51932db106db46968/operator-wg/whitepaper/Operator-WhitePaper_v1-0.md#operator-design-pattern

GitHub - cncf/tag-app-delivery: 📨🚚CNCF App Delivery TAG

- https://docs.bitnami.com/tutorials/a-deep-dive-into-kubernetes-controllers

A deep dive into Kubernetes controllers

- https://kubernetes.io/docs/concepts/extend-kubernetes/operator/

Operator pattern