kube controller manager之root-ca-

作用

当新建或更新namespace时,新建或更新该namespace下名为kube-root-ca.crt的configmap
当某个namespace下的名为kube-root-ca.crt的configmap删除或更新时,重建或更新该namespace下名为kube-root-ca.crt的configmap

图

相关代码

前置相关代码

cmd/kube-controller-manager/app/controllermanager.go

func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {
    ...
    controllers["root-ca-cert-publisher"] = startRootCACertPublisher
    ...
    return controllers
}

cmd/kube-controller-manager/app/certificates.go

func startRootCACertPublisher(ctx context.Context, controllerContext ControllerContext) (controller.Interface, bool, error) {
    var (
        rootCA []byte
        err    error
    )
    if controllerContext.ComponentConfig.SAController.RootCAFile != "" {
        if rootCA, err = readCA(controllerContext.ComponentConfig.SAController.RootCAFile); err != nil {
            return nil, true, fmt.Errorf("error parsing root-ca-file at %s: %v", controllerContext.ComponentConfig.SAController.RootCAFile, err)
        }
    } else {
        rootCA = controllerContext.ClientBuilder.ConfigOrDie("root-ca-cert-publisher").CAData
    }

    sac, err := rootcacertpublisher.NewPublisher(
        controllerContext.InformerFactory.Core().V1().ConfigMaps(),
        controllerContext.InformerFactory.Core().V1().Namespaces(),
        controllerContext.ClientBuilder.ClientOrDie("root-ca-cert-publisher"),
        rootCA,
    )
    if err != nil {
        return nil, true, fmt.Errorf("error creating root CA certificate publisher: %v", err)
    }
    go sac.Run(1, ctx.Done())
    return nil, true, nil
}

pkg/controller/certificates/rootcacertpublisher/publisher.go

func NewPublisher(cmInformer coreinformers.ConfigMapInformer, nsInformer coreinformers.NamespaceInformer, cl clientset.Interface, rootCA []byte) (*Publisher, error) {
    e := &Publisher{
        client: cl,
        rootCA: rootCA,
        queue:  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "root_ca_cert_publisher"),
    }
    if cl.CoreV1().RESTClient().GetRateLimiter() != nil {
        if err := ratelimiter.RegisterMetricAndTrackRateLimiterUsage("root_ca_cert_publisher", cl.CoreV1().RESTClient().GetRateLimiter()); err != nil {
            return nil, err
        }
    }

    cmInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        DeleteFunc: e.configMapDeleted,
        UpdateFunc: e.configMapUpdated,
    })
    e.cmLister = cmInformer.Lister()
    e.cmListerSynced = cmInformer.Informer().HasSynced

    nsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    e.namespaceAdded,
        UpdateFunc: e.namespaceUpdated,
    })
    e.nsListerSynced = nsInformer.Informer().HasSynced

    e.syncHandler = e.syncNamespace

    return e, nil

}

func (c *Publisher) Run(workers int, stopCh <-chan struct{}) {
    defer utilruntime.HandleCrash()
    defer c.queue.ShutDown()

    klog.Infof("Starting root CA certificate configmap publisher")
    defer klog.Infof("Shutting down root CA certificate configmap publisher")

    if !cache.WaitForNamedCacheSync("crt configmap", stopCh, c.cmListerSynced) {
        return
    }

    for i := 0; i < workers; i++ {
        go wait.Until(c.runWorker, time.Second, stopCh)
    }

    <-stopCh
}

主要代码

namespace新建或更新

pkg/controller/certificates/rootcacertpublisher/publisher.go

func (c *Publisher) syncNamespace(ns string) (err error) {
    startTime := time.Now()
    defer func() {
        recordMetrics(startTime, err)
        klog.V(4).Infof("Finished syncing namespace %q (%v)", ns, time.Since(startTime))
    }()

    cm, err := c.cmLister.ConfigMaps(ns).Get(RootCACertConfigMapName)
    switch {
    case apierrors.IsNotFound(err):
        _, err = c.client.CoreV1().ConfigMaps(ns).Create(context.TODO(), &v1.ConfigMap{
            ObjectMeta: metav1.ObjectMeta{
                Name:        RootCACertConfigMapName,
                Annotations: map[string]string{DescriptionAnnotation: Description},
            },
            Data: map[string]string{
                "ca.crt": string(c.rootCA),
            },
        }, metav1.CreateOptions{})
        // don't retry a create if the namespace doesn't exist or is terminating
        if apierrors.IsNotFound(err) || apierrors.HasStatusCause(err, v1.NamespaceTerminatingCause) {
            return nil
        }
        return err
    case err != nil:
        return err
    }

    data := map[string]string{
        "ca.crt": string(c.rootCA),
    }

    // ensure the data and the one annotation describing usage of this configmap match.
    if reflect.DeepEqual(cm.Data, data) && len(cm.Annotations[DescriptionAnnotation]) > 0 {
        return nil
    }

    // copy so we don't modify the cache's instance of the configmap
    cm = cm.DeepCopy()
    cm.Data = data
    if cm.Annotations == nil {
        cm.Annotations = map[string]string{}
    }
    cm.Annotations[DescriptionAnnotation] = Description

    _, err = c.client.CoreV1().ConfigMaps(ns).Update(context.TODO(), cm, metav1.UpdateOptions{})
    return err
}

configmap删除或更新

pkg/controller/certificates/rootcacertpublisher/publisher.go

func (c *Publisher) configMapDeleted(obj interface{}) {
    cm, err := convertToCM(obj)
    if err != nil {
        utilruntime.HandleError(err)
        return
    }
    if cm.Name != RootCACertConfigMapName {
        return
    }
    c.queue.Add(cm.Namespace)
}

func (c *Publisher) configMapUpdated(_, newObj interface{}) {
    cm, err := convertToCM(newObj)
    if err != nil {
        utilruntime.HandleError(err)
        return
    }
    if cm.Name != RootCACertConfigMapName {
        return
    }
    c.queue.Add(cm.Namespace)
}