kubernetes源码阅读之kubelet pod之来源

怼烎@ 2022-06-07 01:27 151阅读 0赞

kubelet最重要的职责就是pod（多个容器）的生命周期管理，那pod的来源是哪里呢？看一下创建中一个容易忽略的函数

源码路径

> /k8s.io/kubernetes/pkg/kubelet/kubelet.go

func NewMainKubelet(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *KubeletDeps, standaloneMode bool) (*Kubelet, error) {
    ...
        if kubeDeps.PodConfig == nil {
            var err error
            kubeDeps.PodConfig, err = makePodSourceConfig(kubeCfg, kubeDeps, nodeName)
            if err != nil {
                return nil, err
            }
        }
    ...

**makePodSourceConfig这个函数。它里面定义了pod源。具体如下：**

// makePodSourceConfig creates a config.PodConfig from the given
    // KubeletConfiguration or returns an error.
    func makePodSourceConfig(kubeCfg *componentconfig.KubeletConfiguration, kubeDeps *KubeletDeps, nodeName types.NodeName) (*config.PodConfig, error) {
        manifestURLHeader := make(http.Header)
        if kubeCfg.ManifestURLHeader != "" {
            pieces := strings.Split(kubeCfg.ManifestURLHeader, ":")
            if len(pieces) != 2 {
                return nil, fmt.Errorf("manifest-url-header must have a single ':' key-value separator, got %q", kubeCfg.ManifestURLHeader)
            }
            manifestURLHeader.Set(pieces[0], pieces[1])
        }
    
        // source of all configuration
        cfg := config.NewPodConfig(config.PodConfigNotificationIncremental, kubeDeps.Recorder)
    
        // define file config source
        if kubeCfg.PodManifestPath != "" {
            glog.Infof("Adding manifest file: %v", kubeCfg.PodManifestPath)
            config.NewSourceFile(kubeCfg.PodManifestPath, nodeName, kubeCfg.FileCheckFrequency.Duration, cfg.Channel(kubetypes.FileSource))
        }
    
        // define url config source
        if kubeCfg.ManifestURL != "" {
            glog.Infof("Adding manifest url %q with HTTP header %v", kubeCfg.ManifestURL, manifestURLHeader)
            config.NewSourceURL(kubeCfg.ManifestURL, manifestURLHeader, nodeName, kubeCfg.HTTPCheckFrequency.Duration, cfg.Channel(kubetypes.HTTPSource))
        }
        if kubeDeps.KubeClient != nil {
            glog.Infof("Watching apiserver")
            config.NewSourceApiserver(kubeDeps.KubeClient, nodeName, cfg.Channel(kubetypes.ApiserverSource))
        }
        return cfg, nil
    }

这个函数里面支持PodManifestPath、ManifestURL、KubeClient三种方式获取pod源，其实前面两种都是让kubelet处于standalone模式，和k8s的api脱离管理，单独运行，通过mainfest创建的pod被称为本地pod，它无法通过apiserver去管理。最重要途径还是通过KubeClient的方式获取容器信息。并在此处watch apiserver，监听api关于本主机上面容器的变化。

func NewSourceApiserver(c clientset.Interface, nodeName types.NodeName, updates chan<- interface{ }) { lw := cache.NewListWatchFromClient(c.Core().RESTClient(), "pods", metav1.NamespaceAll, fields.OneTermEqualSelector(api.PodHostField, string(nodeName))) newSourceApiserverFromLW(lw, updates) }

这里面创建了一个listwatch，它监听在本主机上面所有命名空间pods的变化。并把变化更新都放到updates的channel里面。  
下面在深入看一下创建结构体lw的函数NewListWatchFromClient：

func NewListWatchFromClient(c Getter, resource string, namespace string, fieldSelector fields.Selector) *ListWatch {
        listFunc := func(options api.ListOptions) (runtime.Object, error) {
            return c.Get().
                Namespace(namespace).
                Resource(resource).
                VersionedParams(&options, api.ParameterCodec).
                FieldsSelectorParam(fieldSelector).
                Do().
                Get()
        }
        watchFunc := func(options api.ListOptions) (watch.Interface, error) {
            return c.Get().
                Prefix("watch").
                Namespace(namespace).
                Resource(resource).
                VersionedParams(&options, api.ParameterCodec).
                FieldsSelectorParam(fieldSelector).
                Watch()
        }
        return &ListWatch{ListFunc: listFunc, WatchFunc: watchFunc}
    }

里面定义了listFunc和wtachFunc。下面在看怎么newSourceApiserverFromLW怎样把更新放到updates里面。

func newSourceApiserverFromLW(lw cache.ListerWatcher, updates chan<- interface{}) {
        send := func(objs []interface{}) {
            var pods []*v1.Pod
            for _, o := range objs {
                pods = append(pods, o.(*v1.Pod))
            }
            updates <- kubetypes.PodUpdate{Pods: pods, Op: kubetypes.SET, Source: kubetypes.ApiserverSource}
        }
        cache.NewReflector(lw, &v1.Pod{}, cache.NewUndeltaStore(send, cache.MetaNamespaceKeyFunc), 0).Run()
    }

就是通过send方法吧pods变化发送给updates channel的。那send函数怎么被触发的呢？继续看，首先是创建一个reflector然后是运行run。

func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
        r := &Reflector{
            name:          name,
            listerWatcher: lw,
            store:         store,
            expectedType:  reflect.TypeOf(expectedType),
            period:        time.Second,
            resyncPeriod:  resyncPeriod,
            clock:         &clock.RealClock{},
        }
     return r
    }

上面创建一个reflector的结构体，reflector里面有个store，它是存储pod的信息的。上面的send方法就是store里面的pushFunc,这个方法很有意思，它会在store数据发生变化的时候被调用，从而达到监控的目的。

func NewUndeltaStore(pushFunc func([]interface{}), keyFunc KeyFunc) *UndeltaStore {
        return &UndeltaStore{
            Store:    NewStore(keyFunc),
            PushFunc: pushFunc,
        }
    }

创建完reflector后，就是Run运行了，继续看代码：

func (r *Reflector) Run() {
        glog.V(3).Infof("Starting reflector %v (%s) from %s", r.expectedType, r.resyncPeriod, r.name)
        go wait.Until(func() {
            if err := r.ListAndWatch(wait.NeverStop); err != nil { 
                utilruntime.HandleError(err)
            }
        }, r.period, wait.NeverStop)
    }

Run里面周期的调用ListAndWatch方法，看具体方法实现，

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
        glog.V(3).Infof("Listing and watching %v from %s", r.expectedType, r.name)
        var resourceVersion string
        resyncCh, cleanup := r.resyncChan()
        defer cleanup()
    
        // Explicitly set "0" as resource version - it's fine for the List()
        // to be served from cache and potentially be delayed relative to
        // etcd contents. Reflector framework will catch up via Watch() eventually.
        options := metav1.ListOptions{ResourceVersion: "0"}
        list, err := r.listerWatcher.List(options)
        if err != nil {
            return fmt.Errorf("%s: Failed to list %v: %v", r.name, r.expectedType, err)
        }
        listMetaInterface, err := meta.ListAccessor(list)
        if err != nil {
            return fmt.Errorf("%s: Unable to understand list result %#v: %v", r.name, list, err)
        }
        resourceVersion = listMetaInterface.GetResourceVersion()
        items, err := meta.ExtractList(list)
        if err != nil {
            return fmt.Errorf("%s: Unable to understand list result %#v (%v)", r.name, list, err)
        }
        if err := r.syncWith(items, resourceVersion); err != nil {
            return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
        }
        r.setLastSyncResourceVersion(resourceVersion)
    
        resyncerrc := make(chan error, 1)
        cancelCh := make(chan struct{})
        defer close(cancelCh)
        go func() {
            for {
                select {
                case <-resyncCh:
                case <-stopCh:
                    return
                case <-cancelCh:
                    return
                }
                if r.ShouldResync == nil || r.ShouldResync() {
                    glog.V(4).Infof("%s: forcing resync", r.name)
                    if err := r.store.Resync(); err != nil {
                        resyncerrc <- err
                        return
                    }
                }
                cleanup()
                resyncCh, cleanup = r.resyncChan()
            }
        }()
    
        for {
            timemoutseconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
            options = metav1.ListOptions{
                ResourceVersion: resourceVersion,
                // We want to avoid situations of hanging watchers. Stop any wachers that do not
                // receive any events within the timeout window.
                TimeoutSeconds: &timemoutseconds,
            }
    
            w, err := r.listerWatcher.Watch(options)
            if err != nil {
                switch err {
                case io.EOF:
                    // watch closed normally
                case io.ErrUnexpectedEOF:
                    glog.V(1).Infof("%s: Watch for %v closed with unexpected EOF: %v", r.name, r.expectedType, err)
                default:
                    utilruntime.HandleError(fmt.Errorf("%s: Failed to watch %v: %v", r.name, r.expectedType, err))
                }
                // If this is "connection refused" error, it means that most likely apiserver is not responsive.
                // It doesn't make sense to re-list all objects because most likely we will be able to restart
                // watch where we ended.
                // If that's the case wait and resend watch request.
                if urlError, ok := err.(*url.Error); ok {
                    if opError, ok := urlError.Err.(*net.OpError); ok {
                        if errno, ok := opError.Err.(syscall.Errno); ok && errno == syscall.ECONNREFUSED {
                            time.Sleep(time.Second)
                            continue
                        }
                    }
                }
                return nil
            }
    
            if err := r.watchHandler(w, &resourceVersion, resyncerrc, stopCh); err != nil {
                if err != errorStopRequested {
                    glog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedType, err)
                }
                return nil
            }
        }
    }

具体实现在看watchHandler方法里面，截取关键的部分：

switch event.Type {
                case watch.Added:
                    err := r.store.Add(event.Object)
                    if err != nil {
                        utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", r.name, event.Object, err))
                    }
                case watch.Modified:
                    err := r.store.Update(event.Object)
                    if err != nil {
                        utilruntime.HandleError(fmt.Errorf("%s: unable to update watch event object (%#v) to store: %v", r.name, event.Object, err))
                    }
                case watch.Deleted:
                    // TODO: Will any consumers need access to the "last known
                    // state", which is passed in event.Object? If so, may need
                    // to change this.
                    err := r.store.Delete(event.Object)
                    if err != nil {
                        utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", r.name, event.Object, err))
                    }
                default:
                    utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
                }

通过判断事件类型分别调用store的方法，而更新store数据又会触发send方法，从而把更新信息放到updates channel里面。