前言
前文介绍Pod无论是启动时还是关闭时，处理是由kubelet的主循环syncLoop开始执行逻辑，而syncLoop的入参是一条传递变更Pod的通道，显然syncLoop往后的逻辑属于消费者一方，如何发现Pod的变更往通道里面传递变更消息的一方目前还没明朗，故本次来看一下kubelet是如何发现Pod的变更的。
调用链回溯
syncLoop的通道参数updates是经过在startKubelet函数（代码位于/cmd/kubelet/app/server.go）传入，
func startKubelet(...){
go k.Run(podCfg.Updates())
}
podCfg.Updates()方法只返回PodConfig对象的updates字段，updates通道是在构造PodConfig对象时创建出来的，它在构造PodStorage时传作入参，而后者又在构造Mux对象时传作入参，从而推断往updates通道传入Pod变更的跟Mux和Storage都会有关系，需要理清这三者间的关系。
代码位于/pkg/kubelet/config/config.go
func (c *PodConfig) Updates() <-chan kubetypes.PodUpdate {
return c.updates
}
//PodConfig的结构定义
type PodConfig struct {
pods *podStorage
mux *config.Mux
// the channel of denormalized changes passed to listeners
updates chan kubetypes.PodUpdate
// contains the list of all configured sources
sourcesLock sync.Mutex
sources sets.String
}
//PodConfig的构造函数
func NewPodConfig(mode PodConfigNotificationMode, recorder record.EventRecorder) *PodConfig {
updates := make(chan kubetypes.PodUpdate, 50)
storage := newPodStorage(updates, mode, recorder)
podConfig := &PodConfig{
pods: storage,
mux: config.NewMux(storage),
updates: updates,
sources: sets.String{},
}
return podConfig
}
PodConfig相关对象内部结构
podStorage
podStorage的构造函数及结构定义如下，由结构名得知它主要是负责pod的存储，且它的成员中有一个用于存储pod对象的map,查看了对updates通道的引用是往里面塞入对象，主要通过PodStorage的Merge方法传入
代码位于/pkg/kubelet/config/config.go
func newPodStorage(updates chan<- kubetypes.PodUpdate, mode PodConfigNotificationMode, recorder record.EventRecorder) *podStorage {
return &podStorage{
pods: make(map[string]map[types.UID]*v1.Pod),
mode: mode,
updates: updates,
sourcesSeen: sets.String{},
recorder: recorder,
}
}
type podStorage struct {
podLock sync.RWMutex
// map of source name to pod uid to pod reference
pods map[string]map[types.UID]*v1.Pod
mode PodConfigNotificationMode
// ensures that updates are delivered in strict order
// on the updates channel
updateLock sync.Mutex
updates chan<- kubetypes.PodUpdate
// contains the set of all sources that have sent at least one SET
sourcesSeenLock sync.RWMutex
sourcesSeen sets.String
// the EventRecorder to use
recorder record.EventRecorder
}
暂且不管Merge方法里面的逻辑，Merge方法是podStorage实现Merger接口的方法，而调用Merger接口的地方，有且只有Mux的listen方法
func (m *Mux) listen(source string, listenChannel <-chan interface{}) {
for update := range listenChannel {
m.merger.Merge(source, update)
}
}
下面也了解一下Mux这个结构
Mux
Mux的结构和构造函数如下所示
type Mux struct {
// Invoked when an update is sent to a source.
merger Merger
// Sources and their lock.
sourceLock sync.RWMutex
// Maps source names to channels
sources map[string]chan interface{}
}
func NewMux(merger Merger) *Mux {
mux := &Mux{
sources: make(map[string]chan interface{}),
merger: merger,
}
return mux
}
在Mux的结构的成员及其构造函数得知podStorage在Mux这里只充当一个Merger的作用，此外它还有一个sources的map用于记录通道信息，作用待后续了解。
Mux的listen方法也只有一个地方用到，就是Channel方法，观其大意是用于记录每个新来的source到本地的sources这个map中，新来的source会为其开辟一条新的通道，每当收到内容就会调用merger的Merger进行合并操作
func (m *Mux) Channel(source string) chan interface{} {
if len(source) == 0 {
panic("Channel given an empty name")
}
m.sourceLock.Lock()
defer m.sourceLock.Unlock()
channel, exists := m.sources[source]
if exists {
return channel
}
newChannel := make(chan interface{})
m.sources[source] = newChannel
go wait.Until(func() { m.listen(source, newChannel) }, 0, wait.NeverStop)
return newChannel
}
这个Channel方法也只在PodConfig的Channel方法被调用而已
func (c *PodConfig) Channel(source string) chan<- interface{} {
c.sourcesLock.Lock()
defer c.sourcesLock.Unlock()
c.sources.Insert(source)
return c.mux.Channel(source)
}
再往上溯就来到makePodSourceConfig函数了，这个函数在kubelet启动的流程中，构件kubelet对象的NewMainKubelet函数中被调用，那么Pod变更来源的关键就在这个makePodSourceConfig函数里面了
func makePodSourceConfig(kubeCfg *kubeletconfiginternal.KubeletConfiguration, kubeDeps *Dependencies, nodeName types.NodeName) (*config.PodConfig, error) {
manifestURLHeader := make(http.Header)
if len(kubeCfg.StaticPodURLHeader) > 0 {
for k, v := range kubeCfg.StaticPodURLHeader {
for i := range v {
manifestURLHeader.Add(k, v[i])
}
}
}
// source of all configuration
cfg := config.NewPodConfig(config.PodConfigNotificationIncremental, kubeDeps.Recorder)
// define file config source
if kubeCfg.StaticPodPath != "" {
klog.Infof("Adding pod path: %v", kubeCfg.StaticPodPath)
config.NewSourceFile(kubeCfg.StaticPodPath, nodeName, kubeCfg.FileCheckFrequency.Duration, cfg.Channel(kubetypes.FileSource))
}
// define url config source
if kubeCfg.StaticPodURL != "" {
klog.Infof("Adding pod url %q with HTTP header %v", kubeCfg.StaticPodURL, manifestURLHeader)
config.NewSourceURL(kubeCfg.StaticPodURL, manifestURLHeader, nodeName, kubeCfg.HTTPCheckFrequency.Duration, cfg.Channel(kubetypes.HTTPSource))
}
var updatechannel chan<- interface{}
if kubeDeps.KubeClient != nil {
klog.Infof("Watching apiserver")
if updatechannel == nil {
updatechannel = cfg.Channel(kubetypes.ApiserverSource)
}
config.NewSourceApiserver(kubeDeps.KubeClient, nodeName, updatechannel)
}
return cfg, nil
}
在这里看到Pod有三个来源，且在syncLoop方法也提及updates通道会传递来自file, apiserver和 http三种来源的Pod变更事件，下面则了解这三种Pod的来源
Pod的来源分类
- FileSource:Kubelet会读取文件里面定义的pod进行创建．常常我们使用来定义kubelet管理的static pod ，从配置参数处获得，作为master则有master几个组件的pod定义
- HttpSource:通过http Get该manifest url获取到pod的定义，配置中没指定，暂未发现其使用场景
- ApiSource:通过定义跟kube-apiserver进行通过的kubeclient, 从kube-apiserver中获取需要本节点创建的pod的信息。通常kubelet获得pod的途径
FileSource的路径是在/etc/kubernetes/manifests/
ls /etc/kubernetes/manifests/
etcd-external.yaml kube-apiserver.yaml kube-controller-manager.yaml kube-scheduler.yaml
路径的来源是在kubetel的一个配置文件中获取，配置文件路径通过kubelet的启动参数--config=/var/lib/kubelet/config.yaml，打开这个文件里面有一项就是staticPodPath
cat /var/lib/kubelet/config.yaml|grep staticPodPath
staticPodPath: /etc/kubernetes/manifests
静态Pod
静态 Pod 在指定的节点上由 kubelet 守护进程直接管理，不需要 API 服务器 监管。 与由控制面管理的 Pod（例如，Deployment） 不同；kubelet 监视每个静态 Pod（在它崩溃之后重新启动）。
静态 Pod 永远都会绑定到一个指定节点上的 Kubelet。
kubelet 会尝试通过 Kubernetes API 服务器为每个静态 Pod 自动创建一个 镜像 Pod。 这意味着节点上运行的静态 Pod 对 API 服务来说是可见的，但是不能通过 API 服务器来控制。 Pod 名称将把以连字符开头的节点主机名作为后缀。
从yaml文件中可以得知k8s集群中，etcd,api-server,controller-manager,scheduler这几个控制面的pod都是以静态Pod的形式运行在master节点上。
PodConfig应对3种类型的pod的变更
本地yaml文件
makePodSourceConfig /pkg/kubelet/kubelet.go
|--config.NewSourceFile /pkg/kubelet/config/file.go
|--config.run()
拿到了channel内部就从指定路径中读取所有yaml,无论是否读到，都会往通道中塞一个数据，而且Op全部都是Set，内部是执行以下方法，开的协程有两个作用，定时List以下目录里面有那些文件更改，更新缓存；另外的startWatch和watchEvents通道就是对路径进行监控，凡是对该路径的文件进行任何修改都会触发，更新缓存。这个操作很像Informer的ListAndWatch。
func (s *sourceFile) run() {
listTicker := time.NewTicker(s.period)
go func() {
// Read path immediately to speed up startup.
if err := s.listConfig(); err != nil {
klog.Errorf("Unable to read config path %q: %v", s.path, err)
}
for {
select {
case <-listTicker.C:
if err := s.listConfig(); err != nil {
klog.Errorf("Unable to read config path %q: %v", s.path, err)
}
case e := <-s.watchEvents:
if err := s.consumeWatchEvent(e); err != nil {
klog.Errorf("Unable to process watch event: %v", err)
}
}
}
}()
s.startWatch()
}
指定URL的yaml
makePodSourceConfig /pkg/kubelet/kubelet.go
|--config.NewSourceURL /pkg/kubelet/config/http.go
|--config.run()
这里也是跟本地yaml文件的很类似，定期往指定URL处拿pod的yaml,先把他当作单体的pod进行反序列化，不行则当作podList进行反序列化，得到的结果同样以Set形式往通道塞数据。
func (s *sourceURL) run() {
if err := s.extractFromURL(); err != nil {
// Don't log this multiple times per minute. The first few entries should be
// enough to get the point across.
if s.failureLogs < 3 {
klog.Warningf("Failed to read pods from URL: %v", err)
} else if s.failureLogs == 3 {
klog.Warningf("Failed to read pods from URL. Dropping verbosity of this message to V(4): %v", err)
} else {
klog.V(4).Infof("Failed to read pods from URL: %v", err)
}
s.failureLogs++
} else {
if s.failureLogs > 0 {
klog.Info("Successfully read pods from URL.")
s.failureLogs = 0
}
}
}
处理api-server的list&watch
跟api-server对接并没有用client-go的Informer,而是更直接地使用底层的reflector,将list&watch发现有变更的Pod塞到通道里面
代码位于/pkg/kubelet/config/apiserver.go
func NewSourceApiserver(c clientset.Interface, nodeName types.NodeName, updates chan<- interface{}) {
lw := cache.NewListWatchFromClient(c.CoreV1().RESTClient(), "pods", metav1.NamespaceAll, fields.OneTermEqualSelector(api.PodHostField, string(nodeName)))
newSourceApiserverFromLW(lw, updates)
}
// newSourceApiserverFromLW holds creates a config source that watches and pulls from the apiserver.
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}
}
r := cache.NewReflector(lw, &v1.Pod{}, cache.NewUndeltaStore(send, cache.MetaNamespaceKeyFunc), 0)
go r.Run(wait.NeverStop)
}
PosStorage的Merge方法
目前已经查明三种Pod变更的来源，但是从源头到syncLoop之间还缺了一环，就是之前先不纠结的PosStorage.Merge方法，因为三个来源的只是把Pod往通道里面塞，但是并没有明确本次变更的Pod是属于哪一种变更类型(新增，修改，删除，移除，调谐)
func (s *podStorage) Merge(source string, change interface{}) error {
s.updateLock.Lock()
defer s.updateLock.Unlock()
seenBefore := s.sourcesSeen.Has(source)
adds, updates, deletes, removes, reconciles := s.merge(source, change)
//传入的change是SET时才会是firstSet
firstSet := !seenBefore && s.sourcesSeen.Has(source)
// deliver update notifications
switch s.mode {
///按podStorage只会有这种cast
case PodConfigNotificationIncremental:
if len(removes.Pods) > 0 {
s.updates <- *removes
}
if len(adds.Pods) > 0 {
s.updates <- *adds
}
if len(updates.Pods) > 0 {
s.updates <- *updates
}
if len(deletes.Pods) > 0 {
s.updates <- *deletes
}
if firstSet && len(adds.Pods) == 0 && len(updates.Pods) == 0 && len(deletes.Pods) == 0 {
// Send an empty update when first seeing the source and there are
// no ADD or UPDATE or DELETE pods from the source. This signals kubelet that
// the source is ready.
s.updates <- *adds
}
// Only add reconcile support here, because kubelet doesn't support Snapshot update now.
if len(reconciles.Pods) > 0 {
s.updates <- *reconciles
}
...
}
return nil
}
podStorage.Merge是调用它的内部方法merge将变更的pod分成add,update,delete,remove,reconcile几类，然后就按照podStorage的模式来执行后续的操作，而在makePodSourceConfig处构造PodConfig时就指定使用PodConfigNotificationIncremental模式。但也只是把不同类型的Pod变更放入updates通道而已，区分pod变更的类型主要逻辑还是在内部方法merge中
func (s *podStorage) merge(source string, change interface{}) (adds, updates, deletes, removes, reconciles *kubetypes.PodUpdate) {
addPods := []*v1.Pod{}
updatePods := []*v1.Pod{}
deletePods := []*v1.Pod{}
removePods := []*v1.Pod{}
reconcilePods := []*v1.Pod{}
pods := s.pods[source]
if pods == nil {
pods = make(map[types.UID]*v1.Pod)
}
update := change.(kubetypes.PodUpdate)
switch update.Op {
....
case kubetypes.SET:
klog.V(4).Infof("Setting pods for source %s", source)
s.markSourceSet(source)
// Clear the old map entries by just creating a new map
oldPods := pods
pods = make(map[types.UID]*v1.Pod)
///按set的方式，只有update.Pods的内容才会被保留，原本只有cache才有的pod都会被移除
updatePodsFunc(update.Pods, oldPods, pods)
for uid, existing := range oldPods {
if _, found := pods[uid]; !found {
// this is a delete
removePods = append(removePods, existing)
}
}
default:
klog.Warningf("Received invalid update type: %v", update)
}
s.pods[source] = pods
adds = &kubetypes.PodUpdate{Op: kubetypes.ADD, Pods: copyPods(addPods), Source: source}
updates = &kubetypes.PodUpdate{Op: kubetypes.UPDATE, Pods: copyPods(updatePods), Source: source}
deletes = &kubetypes.PodUpdate{Op: kubetypes.DELETE, Pods: copyPods(deletePods), Source: source}
removes = &kubetypes.PodUpdate{Op: kubetypes.REMOVE, Pods: copyPods(removePods), Source: source}
reconciles = &kubetypes.PodUpdate{Op: kubetypes.RECONCILE, Pods: copyPods(reconcilePods), Source: source}
return adds, updates, deletes, removes, reconciles
}
内部方法merge先从指定来源的缓存中把pod取出来，根据本次变更的类型，执行不一样的操作，但是看之前三个来源都发现update.Op的值都是kubetypes.SET。将原本缓存的pod作为ooldPod，另外再创建一个新的空的集合,调用局部函数updatePodsFunc，执行完毕后则把刚创建的pod集合整个放入缓存中，pod变更类型判定的逻辑在updatePodsFunc里面
//根据oldPods和newPods整合成新的pods用于缓存，同时填充add,update,delete,reconcile四个集合
updatePodsFunc := func(newPods []*v1.Pod, oldPods, pods map[types.UID]*v1.Pod) {
//只是对newPods进行去重
filtered := filterInvalidPods(newPods, source, s.recorder)
for _, ref := range filtered {
// Annotate the pod with the source before any comparison.
if ref.Annotations == nil {
ref.Annotations = make(map[string]string)
}
ref.Annotations[kubetypes.ConfigSourceAnnotationKey] = source
if existing, found := oldPods[ref.UID]; found {
pods[ref.UID] = existing
//比较两个新pod与缓存pod,如果其他一致，只是status不同，则单纯reconcile
//否则就更新缓存pod,如果有删除时间的（DeletionTimestamp）就执行删除
needUpdate, needReconcile, needGracefulDelete := checkAndUpdatePod(existing, ref)
if needUpdate {
updatePods = append(updatePods, existing)
} else if needReconcile {
reconcilePods = append(reconcilePods, existing)
} else if needGracefulDelete {
deletePods = append(deletePods, existing)
}
continue
}
//oldPods里面无的就是新的
recordFirstSeenTime(ref)
pods[ref.UID] = ref
addPods = append(addPods, ref)
}
}
对入参newPods集合先调用filterInvalidPods函数进行按pod的fullname去重，然后尝试从缓存中查找有否UID一样的pod,如果没有则表明是新增操作，放入新增集合中。如果有就是剩下几个类型变更，需要与上一个版本的Pod进行比较方可判定，如果只是status不同，则属于reconcile操作，放入reconcile集合。如果有删除时间的（DeletionTimestamp）属于删除操作，放入删除集合。剩下就是更新操作，放入更新集合。
小结
本篇旨在衔接kubelet启动与kubelet处理pod的启动、关闭之间的逻辑。通过回溯调用链的形式找到pod的变更是如何发现，然后分辨pod变更的类型，最后传递到syncLoop方法针对不同的变更类型执行相应的操作。
其中Pod的来源有三种：本地yaml文件，指定URL上的yaml，还有从api-server中list&watch的结果
三种来源监控任务都由PodConfig中的Mux来承担，它为每种来源类型都开辟一个协程来执行监控任务
当发现有Pod的变更事件就会往通道里传递对象，会传到PodConfig的podStorage里作分辨变更类型的逻辑
最后把区分好变更类型的pod放入PodConfig的updates通道中传递给syncLoop
如要回顾本系列的文章可点击
kubelet源码分析——kubelet简介与启动
kubelet源码分析——启动Pod
kubelet源码分析——关闭Pod
kubelet源码分析——监控Pod变更