controller的实现归档 - 一只安静的猫

一、概述

namespace 是 kubernetes 中一个比较重要的概念，是对一组资源和对象的抽象，也常用来作不同用户的隔离。namespace 下有很多资源，比如我们常用的 deployment, pods, service, ingress, configmap 等等。

当然本篇文章的重点在于删除 namespace 时发生了什么？一个很典型的场景是在终端中执行 kubectl delete ns test 时，我们会观察到，在执行命令后，test 命名空间会立刻进入 terminating 状态，在几秒钟之后，才会被真正删除。即使 test 命名空间中没有任何资源。

NAME              STATUS   AGE
default           Active   2d2h
docker            Active   2d2h
kube-node-lease   Active   2d2h
kube-public       Active   2d2h
kube-system       Active   2d2h
test              Active   4s
test              Terminating   18s
test              Terminating   23s
test              Terminating   23s

因此，我们在下面会探究以下几点：

api-server 如何处理 namespace 的删除请求
删除 namespace 时如何处理其中的资源

二、api server 如何处理 namespace 删除请求

和其他资源不同，namespace 在删除时，需要先清空 namespace 下资源。因此 namespace 有两种状态，即 active 和 terminating。当 namespace 处于 terminating 时，说明其下的资源还没有被确认删除干净。因此，api-server 在收到 namespace 的删除请求时，并不会立刻将其从 etcd 中删除，而是先检查 metadata.DeletionTimestamp 是否为空，如果为空，则是先将 metadata.DeletionTimestamp 置为当前时间，然后将 status.Phase 置为 terminating。如果 metadata.DeletionTimestamp 不为空，还要再判断 spec.Finalizers 是否为空。如果为空，才会真正的删除该 namespace。

这样的处理方式，就保证了在 spec.Finalizers 不为空时，namespace 不会被删除。那么 finalizer 是在什么时候添加的呢？具体的作用是怎么体现的？

三、finalizer 机制

namespace 的 finalizer 其实在创建的时候就已经添加上去了。处理逻辑可见以下代码：

// PrepareForCreate clears fields that are not allowed to be set by end users on creation.
func (namespaceStrategy) PrepareForCreate(ctx context.Context, obj runtime.Object) {
    // on create, status is active
    namespace := obj.(*api.Namespace)
    namespace.Status = api.NamespaceStatus{
        Phase: api.NamespaceActive,
    }
    // on create, we require the kubernetes value
    // we cannot use this in defaults conversion because we let it get removed over life of object
    hasKubeFinalizer := false
    for i := range namespace.Spec.Finalizers {
        if namespace.Spec.Finalizers[i] == api.FinalizerKubernetes {
            hasKubeFinalizer = true
            break
        }
    }
    if !hasKubeFinalizer {
        if len(namespace.Spec.Finalizers) == 0 {
            namespace.Spec.Finalizers = []api.FinalizerName{api.FinalizerKubernetes}
        } else {
            namespace.Spec.Finalizers = append(namespace.Spec.Finalizers, api.FinalizerKubernetes)
        }
    }
}

然后在删除时 namespace 变更到 terminating 状态，namespace controller 就开始发挥作用了。namespace controller 属于 controller manager，其会监听 namespace 的 add 和 update 事件

    // configure the namespace informer event handlers
    namespaceInformer.Informer().AddEventHandlerWithResyncPeriod(
        cache.ResourceEventHandlerFuncs{
            AddFunc: func(obj interface{}) {
                namespace := obj.(*v1.Namespace)
                namespaceController.enqueueNamespace(namespace)
            },
            UpdateFunc: func(oldObj, newObj interface{}) {
                namespace := newObj.(*v1.Namespace)
                namespaceController.enqueueNamespace(namespace)
            },
        },
        resyncPeriod,
    )

并且会使用 workqueue 来保存每一个 namespace 的变化事件。然后统统触发 nm.namespacedResourcesDeleter.Delete(namespace.Name)。当然，如果 namespace 不存在或者 namespace.DeletionTimestamp 为空，则会退出：

    namespace, err := d.nsClient.Get(context.TODO(), nsName, metav1.GetOptions{})
    if err != nil {
        if errors.IsNotFound(err) {
            return nil
        }
        return err
    }
    if namespace.DeletionTimestamp == nil {
        return nil
    }

否则无论如何都会先将 namespace 的 phase 先置为 terminating。这也就是说，如果一个 namespace 已经处于 terminating 了，你就无法通过仅仅修改该 phase 来改变该 namespace 的状态。我之前在遇到过 namespace 一直处于 terminating 时，手动修改了 phase 为 active，但是 namespace 会立刻变为 terminating，原因大概就是如此了。

// updateNamespaceStatusFunc will verify that the status of the namespace is correct
func (d *namespacedResourcesDeleter) updateNamespaceStatusFunc(namespace *v1.Namespace) (*v1.Namespace, error) {
    if namespace.DeletionTimestamp.IsZero() || namespace.Status.Phase == v1.NamespaceTerminating {
        return namespace, nil
    }
    newNamespace := v1.Namespace{}
    newNamespace.ObjectMeta = namespace.ObjectMeta
    newNamespace.Status = *namespace.Status.DeepCopy()
    newNamespace.Status.Phase = v1.NamespaceTerminating
    return d.nsClient.UpdateStatus(context.TODO(), &newNamespace, metav1.UpdateOptions{})
}

之后就开始尝试清空该 namespace 下的所有内容：

    // there may still be content for us to remove
    estimate, err := d.deleteAllContent(namespace)
    if err != nil {
        return err
    }
    if estimate > 0 {
        return &ResourcesRemainingError{estimate}
    }

四、DiscoveryInterface 的工作机制

现在我们面临的一个问题就是如何清理该 namespace 下的所有资源呢？平时如果我们要删除一个 pod，我们可以调用 client-go 提供的 PodInterface 接口来删除，其实就是 RESTful 的 HTTP DELETE 动作的封装。但是现在因为我们不知道 namespace 下有哪些资源，所以就没有办法直接调用删除的接口。

所以 client-go 还提供了一个 DiscoveryInterface，顾名思义，DicoveryInterface 可以用来发现集群中的 API groups，versions, resources。在取得集群中所有的接口资源列表口，我们就可以对这些资源进行查询和删除了。DicoveryInterface 接口如下:

// DiscoveryInterface holds the methods that discover server-supported API groups,
// versions and resources.
type DiscoveryInterface interface {
    RESTClient() restclient.Interface
    ServerGroupsInterface
    ServerResourcesInterface
    ServerVersionInterface
    OpenAPISchemaInterface
}

其中 ServerGroupInterface 提供了获取集群中所有接口组的能力，具体的函数签名如下：

    // ServerGroups returns the supported groups, with information like supported versions and the
    // preferred version.
    ServerGroups() (*metav1.APIGroupList, error)

ServerVersionInterface 可以用来获取服务的版本信息，具体的函数签名如下：

    // ServerVersion retrieves and parses the server's version (git version).
    ServerVersion() (*version.Info, error)

然后我们需要关注的是 ServerResourcesInterface 这个接口

// ServerResourcesInterface has methods for obtaining supported resources on the API server
type ServerResourcesInterface interface {
    // ServerResourcesForGroupVersion returns the supported resources for a group and version.
    ServerResourcesForGroupVersion(groupVersion string) (*metav1.APIResourceList, error)
    // ServerResources returns the supported resources for all groups and versions.
    //
    // The returned resource list might be non-nil with partial results even in the case of
    // non-nil error.
    //
    // Deprecated: use ServerGroupsAndResources instead.
    ServerResources() ([]*metav1.APIResourceList, error)
    // ServerResources returns the supported groups and resources for all groups and versions.
    //
    // The returned group and resource lists might be non-nil with partial results even in the
    // case of non-nil error.
    ServerGroupsAndResources() ([]*metav1.APIGroup, []*metav1.APIResourceList, error)
    // ServerPreferredResources returns the supported resources with the version preferred by the
    // server.
    //
    // The returned group and resource lists might be non-nil with partial results even in the
    // case of non-nil error.
    ServerPreferredResources() ([]*metav1.APIResourceList, error)
    // ServerPreferredNamespacedResources returns the supported namespaced resources with the
    // version preferred by the server.
    //
    // The returned resource list might be non-nil with partial results even in the case of
    // non-nil error.
    ServerPreferredNamespacedResources() ([]*metav1.APIResourceList, error)
}

这里我们可以用 ServerPreferredNamespacedResources 来获取所有属于 namespace 的资源列表。然后过滤出支持 DELETE 的资源。最后获取这些资源的 GroupVersionResources（简称 GVR ）。

    resources, err := d.discoverResourcesFn()
    if err != nil {
        // discovery errors are not fatal.  We often have some set of resources we can operate against even if we don't have a complete list
        errs = append(errs, err)
        conditionUpdater.ProcessDiscoverResourcesErr(err)
    }
    // TODO(sttts): get rid of opCache and pass the verbs (especially "deletecollection") down into the deleter
    deletableResources := discovery.FilteredBy(discovery.SupportsAllVerbs{Verbs: []string{"delete"}}, resources)
    groupVersionResources, err := discovery.GroupVersionResources(deletableResources)
    if err != nil {
        // discovery errors are not fatal.  We often have some set of resources we can operate against even if we don't have a complete list
        errs = append(errs, err)
        conditionUpdater.ProcessGroupVersionErr(err)
    }

最后遍历这些 GVR 进行删除：

    for gvr := range groupVersionResources {
        gvrDeletionMetadata, err := d.deleteAllContentForGroupVersionResource(gvr, namespace, namespaceDeletedAt)
    }

五、为什么 namespace 会长时间处于 terminating 状态

要探究 namespace 长时间处于 terminating 状态的原因，我们先看下面一段很短的代码：

    // there may still be content for us to remove
    estimate, err := d.deleteAllContent(namespace)
    if err != nil {
        return err
    }
    if estimate > 0 {
        return &ResourcesRemainingError{estimate}
    }

在删除命名空间下所有资源的时候，如果返回了错误，或者预估删除完所有资源的时间大于 0 的话，就会继续处于 terminating 状态。比如说 pod 会有一个 terminationGracePeriodSeconds，那么在删除 pod 的时候就可能要等待这个周期过去。但是这也造成不了什么问题，我们常常遇到的头疼问题是，namespace 一直无法删除。简单来说，就是 namespace 下肯定还有资源没法删除，可能性有以下几种。

部分资源有 admission 阻止了删除，因为所有的删除请求都要先进过 admission webhook，那么可能因为 admission 的原因导致部分资源无法直接删除。

apiservice 出问题了。这个问题我们可以通过 kubectl get apiservice 来确认，在 AVAILABLE 一列中，如果有 false 的话，我们就要去检查这个 apiservice 无法使用的原因了。因为 apiservice 出了问题，就会导致这个 apiservice 下的资源无法通过 HTTP 请求去查询或操作，那么自然无法确认是否还有这部分资源遗留，也就无法彻底删除了。

最后，关于 namespace 无法删除的解决方案，网上给出的方案往往是通过置空 namespace 的 spec.finalizers 来做，但是这是治标不治本的方法。因为如果 namespace 无法删除，就一定说明你的集群中存在缺陷或问题，还是要找出真正的原因才是解决之道。你也可以尝试这个工具找出问题所在：https://github.com/thyarles/knsk

简介

informer 是 client-go 提供的一个工具，主要是用来在 api-server 和程序之间同步指定的资源，并作为本地缓存，比如 Pod, Deployment 等等。

我们都知道，kubernetes 中有很多个 controller 在运行，来保证它们关注的资源处于符合期望的状态。比如 ReplicasSet，会保证该 ReplicaSet 有期望的副本数一直运行。这是通过一个不会终止的循环，不断监控当前集群的状态，然后调整的期望的状态。如下代码所示:

for {
    current := getCurrentState()
    desired := getDesiredState()
    reconcile(current, desired)
}

因为这样的需求，所以 controller 需要不停的获取集群中一些资源的状态，然后调整到期望的状态。如果我们是通过网络不停的查询集群状态，将是一个性能很差的方案。为了性能，可以使用缓存，来将指定的资源保存在本地，只要我们及时的更新缓存，就不需要通过网络向集群查询了。

这就是 informer 出现的原因。它通过 List&Watch 来实时同步 api-server 中的资源，然后将资源分成三种事件来触发不同的处理。这三种事件是: Add, Update 和 Delete。同时它还提供了一个抽象的 Store 来提供本地缓存的查询。

最后，它还可以配合 workqueue 来实现本地的重试等等。informer 是一个非常强大的工具，在我们做 kubernetes 上 controller 的开发时必不可少，但是因为 controller 的编写本身就是一件比较复杂的工作，我们必须要对 informer 本身，以及其周边的工具有清晰的理解，才能写出质量更好的代码。

工作流程

informer

这里先放上一张图来做参考。一般我们在使用 informer 时，会使用如下的代码：

// filterd
lw := cache.ListWatch{
    ListFunc: func(options metav1.ListOptions) (object runtime.Object, err error) {
        return k8sCli.CoreV1().Pods(metav1.NamespaceAll).List(context.TODO(), options)
    },
    WatchFunc: func(options metav1.ListOptions) (w watch.Interface, err error) {
        return k8sCli.CoreV1().Pods(metav1.NamespaceAll).Watch(context.TODO(), options)
    },
}
// indexerInformer, shareInformer
store, ctrl := cache.NewInformer(&lw, &v1.Pod{}, 0, cache.ResourceEventHandlerFuncs{
    AddFunc:    handleAddPod,
    UpdateFunc: handleUpdatePod,
    DeleteFunc: handleDeletePod,
})
stopChan := signals.SetupSignalHandler()
go ctrl.Run(stopChan)
if sync := cache.WaitForCacheSync(stopChan, ctrl.HasSynced); !sync {
    log.Println("not sync")
}
log.Println("synchronized finish")

首先，我们定义了 ListWatch 的方法，informer 会用 List 方法来获取所有的 Pod 资源，然后使用 Watch 来监听之后 Pod 资源的更新。

之后我们实例化了一个 informer。第二个参数是资源类型。第三个参数是重新同步的周期，0为不同步，否则会在每个周期开始时重新 List 所有的资源。第四个参数是 ResourceEventHandlerFuncs，这里的 AddFunc, UpdateFunc,DeleteFunc 是本地缓存在增加，更新和删除时触发的事件。

NewInformer 返回了 store 和 ctrl 两个值，store 就是 pod 的本地缓存，我们可以通过查询 store 来代替直接向 api-server 查询。这个返回的 store 实现了如下的接口:

type Store interface {
    Add(obj interface{}) error
    Update(obj interface{}) error
    Delete(obj interface{}) error
    List() []interface{}
    ListKeys() []string
    Get(obj interface{}) (item interface{}, exists bool, err error)
    GetByKey(key string) (item interface{}, exists bool, err error)

    // Replace will delete the contents of the store, using instead the
    // given list. Store takes ownership of the list, you should not reference
    // it after calling this function.
    Replace([]interface{}, string) error
    Resync() error
}

在 controller 中，我们一般使用 List*, Get* 方法，可以用来查询本地的缓存。同时不要使用其他的方法，这会导致一些不可预知的问题。

另外一个返回值 ctrl，实现了如下的接口：

type Controller interface {
    Run(stopCh <-chan struct{})
    HasSynced() bool
    LastSyncResourceVersion() string
}

这个接口很简单，Run 用来开始启动同步，stopCh 用来随时停止同步，HasSynced 用来判断同步是否完成。LastSyncResourceVersion 用来获取最新同步的资源 version。

cache.WaitForCacheSync 用来等待同步完成。

总结

关于 informer 的基本使用就先介绍这么多。后面会对 informer 中涉及的代码进行详细的分析，包括 List&Watch 的机制、DeltaFIFO 的实现、本地缓存(Store) 的实现等等。

分类： controller的实现

k8s 中删除 namespace 时发生了什么