深入源码分析kubernetes informer机制(四)DeltaFIFO

这篇具有很好参考价值的文章主要介绍了深入源码分析kubernetes informer机制(四)DeltaFIFO。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。


[阅读指南]
这是该系列第四篇
基于kubernetes 1.27 stage版本
为了方便阅读,后续所有代码均省略了错误处理及与关注逻辑无关的部分。


client-go中的存储结构

如下图,clinet-go中定义了存储类型接口store,用来提供存储对象的基本能力。
深入源码分析kubernetes informer机制(四)DeltaFIFO,kubernetes,容器,云原生,deltaFIFO,informer

queue继承了store接口,并提供了队列的能力,队列中可以保存需要增删改的存储对象的key,它会取出队头元素,调用PopProcessFunc处理。

queue的实现有两个:FIFOdeltaFIFO

deltaFIFO的不同点在于,deltaFIFO队列中,key对应的不是对象本身,而是对象的delta。

另外deltaFIFO除了通过add、update、delete添加元素,还有两种特殊的方式:replaced和sync。replaced一般发生在资源版本更新时,而sync由resync定时发起。

DeltaFIFO

下面是deltaFIFO数据结构的定义

type DeltaFIFO struct {
    // 并发读写锁
	lock sync.RWMutex
	cond sync.Cond

    // `items` maps a key to a Deltas.
    // 资源对象的key与对应的delta数组,每个数组至少都会有一个delta
	items map[string]Deltas

	// 按照FIFO队列顺序存储key,用来给pop()消费。
    // 该数组不会有重复值,并且所有元素都一定在items中
	queue []string

	// 生成key值的函数,默认是 MetaNamespaceKeyFunc
	keyFunc KeyFunc

    // 本地缓存中已知的所有资源对象的key
	knownObjects KeyListerGetter

	......
}

delta

如前面所说,deltaFIFO中key映射的不是对象本身,是delta数组。

根据Delta数据结构的定义,delta包含了一个资源对象的变更类型及变更的内容。这里的Object不一定是完整的资源数据,大部分场景下只会有变更的部分信息。

type Delta struct {
	Type   DeltaType
	Object interface{}
}

type DeltaType string
const (
	Added   DeltaType = "Added"
	Updated DeltaType = "Updated"
	Deleted DeltaType = "Deleted"
	Replaced DeltaType = "Replaced"
	Sync DeltaType = "Sync"
)

举个栗子,本地已经有了一个pod对象,

&Pod{
    Name:      "mypod",
    Namespace: "default",
    Labels:    map[string]string{"app": "web", "version": "0.0.1"},
}

此时mypod的 lable从web变成了app-server,reflector就会创建一个这样的delta对象放入FIFO队列中。

&Delta{
    Type:   "Updated",
    Object: &Pod{
            Name:      "mypod",
            Namespace: "default",
            Labels:    map[string]string{"app": "app-server"},
        },
}

索引 key

deltaFIFO队列中,存储的是delta的key值,通过key值可以在items map中获取到对应的delta对象。

这个key值在初始化FIFO时通过KeyFunction进行定义,使用者没有指定时,都会使用自带的命名函数 MetaNamespaceKeyFunc 进行命名,命名规则是

  • namespace不为空,key为/
  • namespace为空,key为

这里的name是在yaml资源配置中的matadata.name,比如上面的mypod。在同一个资源下,name在所有api version都一定是唯一的。

func MetaNamespaceKeyFunc(obj interface{}) (string, error) {
	if key, ok := obj.(ExplicitKey); ok {
		return string(key), nil
	}
	meta, err := meta.Accessor(obj)

	if len(meta.GetNamespace()) > 0 {
		return meta.GetNamespace() + "/" + meta.GetName(), nil
	}
	return meta.GetName(), nil
}

queue push操作

watcher监控的资源变更时,会调用deltaFIFO中Added、Updated、Deleted、Replaced、Sync方法,最终它们都会通过queueActionLocked 方法往deltaFIFO队列中加入对应类型的delta对象。

queueActionLocked 也就是deltaFIFO的入队操作。

和一般的入队不同的是,新加入的delta不是直接加入到队尾,队列queue数组中保存的是delta的key。所以入队的操作是这样的

  1. 获取delta对应的key值(还记得keyfunc吗,又是它)
  2. 如果delta所属的资源key已经在队列中,直接将delta添加到key对应到deltas数组末尾。更新已存在的资源delta并不会影响他的key在队列中的位置。
  3. 如果delta所属的资源key不在队列中,就将key添加到队列末尾,并在items中关联key和delta
func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
	id, err := f.KeyOf(obj)

	// 自定义的转换函数。可以在delta事件被处理之前完成一些预处理
    // 常见的用法是用来过滤一些处理程序不关注的资源对象、以及处理数据格式等
	if f.transformer != nil {
		obj, err = f.transformer(obj)
	}

    // 将新的delta放入资源key对应的delta数组末尾
    // 如果原本的key不存在,就是创建了一个新的数组,并将新的delta放入其中
	oldDeltas := f.items[id]
	newDeltas := append(oldDeltas, Delta{actionType, obj})

    // 对delta数组中的delta去重
	newDeltas = dedupDeltas(newDeltas)

    // 判断key是否已经在队列中,并且更新key对应的delta数组
	if len(newDeltas) > 0 {
		if _, exists := f.items[id]; !exists {
			f.queue = append(f.queue, id)
		}
		f.items[id] = newDeltas
		f.cond.Broadcast()
    }
    
	return nil
}

delta push 去重

上一节提到,delta进行push操作时,会对加入的delta进行去重。去重逻辑目前只针对两个delete类型的delta有效:当delta数组中倒数第一个和第二个delta都是delete类型时,将会去掉其中一个

func dedupDeltas(deltas Deltas) Deltas {
	n := len(deltas)
	if n < 2 {
		return deltas
	}
	a := &deltas[n-1]
	b := &deltas[n-2]
	if out := isDup(a, b); out != nil {
		deltas[n-2] = *out
		return deltas[:n-1]
	}
	return deltas
}

// 判断a、b两个delta是否重复
// 目前暂时只有两个delete类型的delta会被判定为重复。
func isDup(a, b *Delta) *Delta {
	if out := isDeletionDup(a, b); out != nil {
		return out
	}
	return nil
}

// 判定两个delta是否都是deleted类型
func isDeletionDup(a, b *Delta) *Delta {
	if b.Type != Deleted || a.Type != Deleted {
		return nil
	}
    
	if _, ok := b.Object.(DeletedFinalStateUnknown); ok {
		return a
	}
	return b
}

举个小小的例子来回顾一下delta push操作。假设queue中有3个pod对象,对应了不同的变更事件,如下所示。
深入源码分析kubernetes informer机制(四)DeltaFIFO,kubernetes,容器,云原生,deltaFIFO,informer

此时watcher监听到资源发生变化:

  1. pod2收到了updated事件
  2. pod1收到了deleted事件
  3. pod3收到了deleted事件

于是,三个delta入队成功后的队列图如下
深入源码分析kubernetes informer机制(四)DeltaFIFO,kubernetes,容器,云原生,deltaFIFO,informer

pod1已有一个deleted事件,再次收到deleted后,经过dedupDeltas去重,最终只保留一个deleted。

pod3虽然有两个deleted事件,但是他们并不是连续的事件,不会被去重

queue pop操作

deltaFIFO出队的操作和普通的队列出队类似,从队头取出一个资源对象key,并删除items中key对应的deltas数组。

pop出队时,会调用传参PopProcessFunc对出队元素进行处理。

func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
	f.lock.Lock()
	defer f.lock.Unlock()
	for {
		for len(f.queue) == 0 {
            // 队列为空时阻塞
			if f.closed {
				return nil, ErrFIFOClosed
			}
			f.cond.Wait()
		}

        // 取出队首的资源对象key
		id := f.queue[0]
		f.queue = f.queue[1:]

        // 获取key对应的deltas数组
		item, ok := f.items[id]

        // 执行pop处理函数,处理delta事件,如果处理失败了,资源对象会被重新加入到队列中。
        // 但是如果队列中存在相同的对象,资源对象会被丢弃。
		err := process(item, isInInitialList)
		if e, ok := err.(ErrRequeue); ok {
			f.addIfNotPresent(id, item)
			err = e.Err
		}

		return item, err
	}
}

这里一开始有个小疑问,如果资源的delta处理失败了,并且队列中又出现了同样的资源key,这部分delta数据不就丢失了吗?

但是仔细看出队入队公用一个锁,pop处理对象时不会有新的对象入队,所以理论上不会出现在addIfNotPresent时,key是persent的情况。而deltaFIFO入队的逻辑,也不会存在一个队列有两个相同的key的情况,所以不会有丢失的问题,addIfNotPresent应该只是加多一层保障。如果理解有问题,欢迎大佬们指正。

回顾一下pop的调用方processLoop,调用pop时传入PopProcessFunc(c.config.Process))。

系列第一篇介绍informer时提到过,c.config.Process最终调用的是processDeltas函数,它包含了数据同步到存储,以及调用注册的用户函数两个操作。

func (c *controller) processLoop() {
	for {
		obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
		if err != nil {
			...
		}
	}
}

// 数据处理函数
func processDeltas(
	handler ResourceEventHandler,
	clientState Store,
	deltas Deltas,
	isInInitialList bool,
) error {
	// from oldest to newest
	for _, d := range deltas {
		obj := d.Object

        // 区分事件类型进行处理
		switch d.Type {
		case Sync, Replaced, Added, Updated:
            // 同步存储数据
			if old, exists, err := clientState.Get(obj); err == nil && exists {
				if err := clientState.Update(obj); err != nil {
					return err
				}
                // 回调用户函数
				handler.OnUpdate(old, obj)
			} else {
                // 同步存储数据
				if err := clientState.Add(obj); err != nil {
					return err
				}
                // 回调用户函数
				handler.OnAdd(obj, isInInitialList)
			}
		case Deleted:
            // 同步存储数据
			if err := clientState.Delete(obj); err != nil {
				return err
			}
            // 回调用户函数
			handler.OnDelete(obj)
		}
	}
	return nil
}

总结

还是用上一节的例子,小结回顾一下整体的流程
深入源码分析kubernetes informer机制(四)DeltaFIFO,kubernetes,容器,云原生,deltaFIFO,informer文章来源地址https://www.toymoban.com/news/detail-649745.html

到了这里,关于深入源码分析kubernetes informer机制(四)DeltaFIFO的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 深入分析Spring的IoC容器:从底层源码探索

    前言: 博主在最近的几次面试中,大中小厂都问到了Spring的ioc容器相关问题,这块知识确实是面试中的重点内容,因此结合所看的书籍,在这篇文章中总结下。该专栏比较适合刚入坑Java的小白以及准备秋招的大佬阅读,感谢大佬的关注。 如果文章有什么需要改进的地方欢迎

    2024年02月12日
    浏览(51)
  • 【云原生-深入理解 Kubernetes 系列 3】深入理解容器进程的文件系统

    【云原生-深入理解Kubernetes-1】容器的本质是进程 【云原生-深入理解Kubernetes-2】容器 Linux Cgroups 限制 大家好,我是秋意零。 😈 CSDN作者主页 😎 博客主页 👿 简介 👻 普通本科生在读 在校期间参与众多计算机相关比赛,如:🌟 “省赛”、“国赛” ,斩获多项奖项荣誉证书

    2024年02月06日
    浏览(50)
  • Kubernetes日志查看指南:深入了解容器日志管理技术

    简介: Kubernetes(简称K8s)已成为现代容器化应用程序管理的主要平台之一。了解如何有效地查看和管理Kubernetes集群中的容器日志对于故障排除、性能优化和安全监控至关重要。本文将向您介绍一些基本的技术和工具,帮助您在Kubernetes环境中查看和分析容器日志。 查看单个

    2024年02月12日
    浏览(42)
  • 【云原生-深入理解Kubernetes-1】容器的本质是进程

    大家好,我是秋意零。 😈 CSDN作者主页 😎 博客主页 👿 简介 👻 普通本科生在读 在校期间参与众多计算机相关比赛,如:🌟 “省赛”、“国赛” ,斩获多项奖项荣誉证书 🔥 各个平台, 秋意零/秋意临 账号创作者 🔥 云社区 创建者 点赞、收藏+关注下次不迷路! 欢迎加

    2024年02月02日
    浏览(58)
  • 云原生之深入解析Kubernetes中如何使用临时容器进行故障排查

    容器及其周围的生态系统改变了工程师部署、维护和排查工作负载故障的方式。但是,在 Kubernetes 集群上调试应用程序有时可能会很困难,因为可能在容器中找不到所需的调试工具。许多工程师使用基于精简、发行版构建无发行版的基础镜像,其中甚至没有包管理器或shell,

    2024年02月05日
    浏览(48)
  • 云原生之深入解析如何正确计算Kubernetes容器CPU使用率

    使用 Prometheus 配置 kubernetes 环境中 Container 的 CPU 使用率时,会经常遇到 CPU 使用超出 100%,现在来分析一下: container_spec_cpu_period:当对容器进行 CPU 限制时,CFS 调度的时间窗口,又称容器 CPU 的时钟周期通常是 100000 微秒 container_spec_cpu_quota:是指容器的使用 CPU 时间周期总量

    2024年02月10日
    浏览(71)
  • kubernetes--分析容器系统调用:Sysdig

    目录 Sysdig介绍: sysdig工作流程 安装Sysdig sysdig常用参数: sysdig过滤: sysdig之Chisels(工具箱): 其他常用命令        Sysdig是一个非常强大的系统监控分析和故障排查工具。汇聚strace+tcpdump+iftop+lsof工具功能为一身。        sysdig除了能获取系统资源利用率、进程、网络连

    2024年02月05日
    浏览(43)
  • 源码分析——ArrayList源码+扩容机制分析

    ArrayList 的底层是数组队列,相当于动态数组。与 Java 中的数组相比,它的容量能动态增长。在添加大量元素前,应用程序可以使用 ensureCapacity 操作来增加 ArrayList 实例的容量。这可以减少递增式再分配的数量。 ArrayList 继承于 AbstractList ,实现了 List , RandomAccess , Cloneable , ja

    2024年02月14日
    浏览(46)
  • 深入源码解析ArrayList:探秘Java动态数组的机制与性能

    1.1 介绍ArrayList的基本概念和作用 在Java中,ArrayList是一个实现了List接口的动态数组。它可以根据需要自动增加大小,因此可以存储任意数量的元素。 基本概念: ArrayList是Java中常用的集合类之一,它可以存储对象,并且可以根据索引访问和操作这些对象。 ArrayList是基于数组

    2024年02月04日
    浏览(41)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包