上篇扒了 HPA 的源码,但是没深入细节,今天往细节深入。
Kubernetes 中的持久化数据保存在 etcd中,各个组件并不会直接访问 etcd,而是通过 api-server暴露的 RESTful 接口对集群进行访问和控制。
资源的控制器(图中右侧灰色的部分)读取数据也并不会直接从 api-server 中获取资源信息(这样会增加 api-server 的压力),而是从其“本地缓存”中读取。这个“本地缓存”只是表象的存在,加上缓存的同步逻辑就是今天要是说的Informer
(灰色区域中的第一个蓝色块)所提供的功能。
从图中可以看到 Informer 的几个组件:
api-server
交互,监听资源的变更。上篇 提到了水平自动伸缩的控制器HorizontalController
,其构造方法就需要提供 Informer
。
//pkg/controller/podautoscaler/horizontal.go
type HorizontalController struct {
scaleNamespacer scaleclient.ScalesGetter
hpaNamespacer autoscalingclient.HorizontalPodAutoscalersGetter
mapper apimeta.RESTMapper
replicaCalc *ReplicaCalculator
eventRecorder record.EventRecorder
downscaleStabilisationWindow time.Duration
hpaLister autoscalinglisters.HorizontalPodAutoscalerLister
hpaListerSynced cache.InformerSynced
podLister corelisters.PodLister
podListerSynced cache.InformerSynced
queue workqueue.RateLimitingInterface
recommendations map[string][]timestampedRecommendation
}
func NewHorizontalController(
evtNamespacer v1core.EventsGetter,
scaleNamespacer scaleclient.ScalesGetter,
hpaNamespacer autoscalingclient.HorizontalPodAutoscalersGetter,
mapper apimeta.RESTMapper,
metricsClient metricsclient.MetricsClient,
//从HorizontalPodAutoscalerInformer 获取hpa 实例信息
hpaInformer autoscalinginformers.HorizontalPodAutoscalerInformer,
//从PodInformer 中获取 pod 信息
podInformer coreinformers.PodInformer,
resyncPeriod time.Duration,
downscaleStabilisationWindow time.Duration,
tolerance float64,
cpuInitializationPeriod,
delayOfInitialReadinessStatus time.Duration,
) *HorizontalController {
......
hpaInformer.Informer().AddEventHandlerWithResyncPeriod( //添加事件处理器
cache.ResourceEventHandlerFuncs{
AddFunc: hpaController.enqueueHPA,
UpdateFunc: hpaController.updateHPA,
DeleteFunc: hpaController.deleteHPA,
},
resyncPeriod,
)
......
}
type HorizontalPodAutoscalerInformer interface {
Informer() cache.SharedIndexInformer
Lister() v1.HorizontalPodAutoscalerLister
}
HorizontalPodAutoscalerInformer
的实例化方法中就出现了今天的正主cache.NewSharedIndexInformer()
。
//staging/src/k8s.io/client-go/informers/autoscaling/v1/horizontalpodautoscaler.go
func NewFilteredHorizontalPodAutoscalerInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
return cache.NewSharedIndexInformer(
//用于 list 和 watch api-server 中的资源。比如用来创建 Reflector
&cache.ListWatch{
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
if tweakListOptions != nil {
tweakListOptions(&options)
}
//使用 HPA API 获取 HPA资源
return client.AutoscalingV1().HorizontalPodAutoscalers(namespace).List(options)
},
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
if tweakListOptions != nil {
tweakListOptions(&options)
}
//使用 HPA API 监控 HPA资源
return client.AutoscalingV1().HorizontalPodAutoscalers(namespace).Watch(options)
},
},
&autoscalingv1.HorizontalPodAutoscaler{},
resyncPeriod,
indexers,
)
}
//staging/src/k8s.io/client-go/tools/cache/index.go
type Indexers map[string]IndexFunc
type IndexFunc func(obj interface{}) ([]string, error)
实例化 Indexers cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}
//staging/src/k8s.io/client-go/tools/cache/shared_informer.go
// ListerWatcher 用于 list 和watch api-server 上的资源
//runtime.Object要监控的资源的运行时对象
//time.Duration同步的间隔时间
//Indexers 提供不同资源的索引数据的信息查询方法,如 namespace => MetaNamespaceIndexFunc
func NewSharedIndexInformer(lw ListerWatcher, objType runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer {
realClock := &clock.RealClock{}
sharedIndexInformer := &sharedIndexInformer{
processor: &sharedProcessor{clock: realClock},
indexer: NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers), //初始化 Indexer
listerWatcher: lw,
objectType: objType,
resyncCheckPeriod: defaultEventHandlerResyncPeriod,
defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod,
cacheMutationDetector: NewCacheMutationDetector(fmt.Sprintf("%T", objType)),
clock: realClock,
}
return sharedIndexInformer
}
Indexer
提供了本地缓存的实现:计算 key 和对数据进行控制(通过调用ThreadSafeStore
的接口)
type Indexer interface {
Store
Index(indexName string, obj interface{}) ([]interface{}, error)
IndexKeys(indexName, indexedValue string) ([]string, error)
ListIndexFuncValues(indexName string) []string
ByIndex(indexName, indexedValue string) ([]interface{}, error)
GetIndexers() Indexers
AddIndexers(newIndexers Indexers) error
}
Indexer
的创建
//staging/src/k8s.io/client-go/tools/cache/store.go
//keyFunc:key 的生成规则
//indexers:提供了索引资源的不同信息的访问方法,如用于查询命名空间的 namespace => MetaNamespaceIndexFunc
func NewIndexer(keyFunc KeyFunc, indexers Indexers) Indexer {
return &cache{
cacheStorage: NewThreadSafeStore(indexers, Indices{}),
keyFunc: keyFunc,
}
}
ThreadSafeStore
ThreadSafeStore提供了对存储的并发访问接口
注意事项:不能修改Get或List返回的任何内容,因为它不仅会破坏线程安全,还会破坏索引功能。
//staging/src/k8s.io/client-go/tools/cache/thread_safe_store.go
func NewThreadSafeStore(indexers Indexers, indices Indices) ThreadSafeStore {
return &threadSafeMap{
items: map[string]interface{}{},
indexers: indexers,
indices: indices,
}
}
type threadSafeMap struct {
lock sync.RWMutex
items map[string]interface{} //key => value
indexers Indexers //value 的信息的访问方法
indices Indices //索引
}
Reflector
通过ListerWatcher
(API)与api-server
交互,对资源进行监控。将资源实例的创建、更新、删除等时间封装后保存在Informer
的FIFO 队列中。
//staging/src/k8s.io/client-go/tools/cache/reflector.go
func NewReflector(lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
return NewNamedReflector(naming.GetNameFromCallsite(internalPackages...), lw, expectedType, store, resyncPeriod)
}
// NewNamedReflector same as NewReflector, but with a specified name for logging
func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
r := &Reflector{
name: name,
listerWatcher: lw,
store: store, //FIFO队列
period: time.Second,
resyncPeriod: resyncPeriod,
clock: &clock.RealClock{},
}
r.setExpectedType(expectedType)
return r
}
通过sharedIndexInformer#AddEventHandlerWithResyncPeriod()
注册事件监听器。
以前面的 HorizontalController为例,创建 informer 的时候添加了三个处理方法:AddFunc
、UpdateFunc
、DeleteFunc
。这三个方法的实现是将对应的元素的 key(固定格式 namespace/name
)从workequeue
中进行入队、出队的操作。(资源控制器监听了该 workqueue
)
controller-manager
在通过InformerFactory
创建Informer
完成后,都会将新建的Informer
加入到InformerFactory
的一个map
中。
在controller-manager
在完成所有的控制器(各种Controller
,包括 CRD)后,会调用InformerFactory#Start()
来启动InformerFactory
的map
中的所有Informer
(调用Informer#Run()
方法)
sharedIndexInformer#Run()
//staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
defer utilruntime.HandleCrash()
//创建一个增量的 FIFO队列:DeltaFIFO
fifo := NewDeltaFIFO(MetaNamespaceKeyFunc, s.indexer)
cfg := &Config{
Queue: fifo,
ListerWatcher: s.listerWatcher,
ObjectType: s.objectType,
FullResyncPeriod: s.resyncCheckPeriod,
RetryOnError: false,
ShouldResync: s.processor.shouldResync,
Process: s.HandleDeltas,
}
//启动前的初始化,创建 Controller
func() {
s.startedLock.Lock()
defer s.startedLock.Unlock()
s.controller = New(cfg)
s.controller.(*controller).clock = s.clock
s.started = true
}()
processorStopCh := make(chan struct{})
var wg wait.Group
defer wg.Wait() // Wait for Processor to stop
defer close(processorStopCh) // Tell Processor to stop
wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
wg.StartWithChannel(processorStopCh, s.processor.run)
//退出时的状态清理
defer func() {
s.startedLock.Lock()
defer s.startedLock.Unlock()
s.stopped = true // Don't want any new listeners
}()
//实行控制逻辑
s.controller.Run(stopCh)
}
controller#Run()
//staging/src/k8s.io/client-go/tools/cache/controller.go
func (c *controller) Run(stopCh <-chan struct{}) {
defer utilruntime.HandleCrash()
go func() {
<-stopCh
c.config.Queue.Close()
}()
//创建一个 Reflector,用于从 api-server list 和 watch 资源
r := NewReflector(
c.config.ListerWatcher,
c.config.ObjectType,
c.config.Queue,
c.config.FullResyncPeriod,
)
r.ShouldResync = c.config.ShouldResync
r.clock = c.clock
//为 controller 指定 Reflector
c.reflectorMutex.Lock()
c.reflector = r
c.reflectorMutex.Unlock()
var wg wait.Group
defer wg.Wait()
//执行Reflector#Run():会启动一个goroutine开始监控资源,将 watch 到的数据写入到queue(FIFO 队列)中
wg.StartWithChannel(stopCh, r.Run)
//持续从 queue(FIFO 队列) 获取数据并进行处理,处理的逻辑在sharedIndexInformer#HandleDeltas()
wait.Until(c.processLoop, time.Second, stopCh)
}
sharedIndexInformer#HandleDeltas()
//staging/src/k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
s.blockDeltas.Lock()
defer s.blockDeltas.Unlock()
// from oldest to newest
for _, d := range obj.(Deltas) { //循环处理 FIFO 队列中取出的资源实例
switch d.Type {
case Sync, Added, Updated: //同步(后面详细解读)、新增、更新事件
isSync := d.Type == Sync
s.cacheMutationDetector.AddObject(d.Object)
if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
if err := s.indexer.Update(d.Object); err != nil { //如果 indexer 中已经存在,更掉用 update 方法进行更新
return err
}
//更新成功后发送“更新”通知:包含了新、旧资源实例
s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
} else {
//如果 indexer 中没有该资源实例,则放入 indexer 中
if err := s.indexer.Add(d.Object); err != nil {
return err
}
//添加成功后,发送“新增”通知:包含了新加的资源实例
s.processor.distribute(addNotification{newObj: d.Object}, isSync)
}
case Deleted: //删除事件
if err := s.indexer.Delete(d.Object); err != nil {//从 indexer 中删除
return err
}
//删除成功后,发送“删除通知”:包含了删除的资源实例
s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
}
}
return nil
}
Informer 的实现不算复杂,却在 Kubernetes 中很常见,每种资源的控制也都通过 Informer 来获取api-server
的资源实例的变更。
文章统一发布在公众号
云原生指北
文章浏览阅读645次。这个肯定是末尾的IDAT了,因为IDAT必须要满了才会开始一下个IDAT,这个明显就是末尾的IDAT了。,对应下面的create_head()代码。,对应下面的create_tail()代码。不要考虑爆破,我已经试了一下,太多情况了。题目来源:UNCTF。_攻防世界困难模式攻略图文
文章浏览阅读2.9k次,点赞3次,收藏10次。偶尔会用到,记录、分享。1. 数据库导出1.1 切换到dmdba用户su - dmdba1.2 进入达梦数据库安装路径的bin目录,执行导库操作 导出语句:./dexp cwy_init/[email protected]:5236 file=cwy_init.dmp log=cwy_init_exp.log 注释: cwy_init/init_123..._达梦数据库导入导出
文章浏览阅读1.9k次。1. 在官网上下载KindEditor文件,可以删掉不需要要到的jsp,asp,asp.net和php文件夹。接着把文件夹放到项目文件目录下。2. 修改html文件,在页面引入js文件:<script type="text/javascript" src="./kindeditor/kindeditor-all.js"></script><script type="text/javascript" src="./kindeditor/lang/zh-CN.js"_kindeditor.js
文章浏览阅读2.3k次,点赞6次,收藏14次。SPI的详情简介不必赘述。假设我们通过SPI发送0xAA,我们的数据线就会变为10101010,通过修改不同的内容,即可修改SPI中0和1的持续时间。比如0xF0即为前半周期为高电平,后半周期为低电平的状态。在SPI的通信模式中,CPHA配置会影响该实验,下图展示了不同采样位置的SPI时序图[1]。CPOL = 0,CPHA = 1:CLK空闲状态 = 低电平,数据在下降沿采样,并在上升沿移出CPOL = 0,CPHA = 0:CLK空闲状态 = 低电平,数据在上升沿采样,并在下降沿移出。_stm32g431cbu6
文章浏览阅读1.2k次,点赞2次,收藏8次。数据链路层习题自测问题1.数据链路(即逻辑链路)与链路(即物理链路)有何区别?“电路接通了”与”数据链路接通了”的区别何在?2.数据链路层中的链路控制包括哪些功能?试讨论数据链路层做成可靠的链路层有哪些优点和缺点。3.网络适配器的作用是什么?网络适配器工作在哪一层?4.数据链路层的三个基本问题(帧定界、透明传输和差错检测)为什么都必须加以解决?5.如果在数据链路层不进行帧定界,会发生什么问题?6.PPP协议的主要特点是什么?为什么PPP不使用帧的编号?PPP适用于什么情况?为什么PPP协议不_接收方收到链路层数据后,使用crc检验后,余数为0,说明链路层的传输时可靠传输
文章浏览阅读587次。软件测试工程师移民加拿大 无证移民,未受过软件工程师的教育(第1部分) (Undocumented Immigrant With No Education to Software Engineer(Part 1))Before I start, I want you to please bear with me on the way I write, I have very little gen...
文章浏览阅读304次。Thinkpad X250笔记本电脑,装的是FreeBSD,进入BIOS修改虚拟化配置(其后可能是误设置了安全开机),保存退出后系统无法启动,显示:secure boot failed ,把自己惊出一身冷汗,因为这台笔记本刚好还没开始做备份.....根据错误提示,到bios里面去找相关配置,在Security里面找到了Secure Boot选项,发现果然被设置为Enabled,将其修改为Disabled ,再开机,终于正常启动了。_安装完系统提示secureboot failure
文章浏览阅读10w+次,点赞93次,收藏352次。1、用strtok函数进行字符串分割原型: char *strtok(char *str, const char *delim);功能:分解字符串为一组字符串。参数说明:str为要分解的字符串,delim为分隔符字符串。返回值:从str开头开始的一个个被分割的串。当没有被分割的串时则返回NULL。其它:strtok函数线程不安全,可以使用strtok_r替代。示例://借助strtok实现split#include <string.h>#include <stdio.h&_c++ 字符串分割
文章浏览阅读2.3k次。1 .高斯日记 大数学家高斯有个好习惯:无论如何都要记日记。他的日记有个与众不同的地方,他从不注明年月日,而是用一个整数代替,比如:4210后来人们知道,那个整数就是日期,它表示那一天是高斯出生后的第几天。这或许也是个好习惯,它时时刻刻提醒着主人:日子又过去一天,还有多少时光可以用于浪费呢?高斯出生于:1777年4月30日。在高斯发现的一个重要定理的日记_2013年第四届c a组蓝桥杯省赛真题解答
文章浏览阅读851次,点赞17次,收藏22次。摘要:本文利用供需算法对核极限学习机(KELM)进行优化,并用于分类。
文章浏览阅读1.1k次。一、系统弱密码登录1、在kali上执行命令行telnet 192.168.26.1292、Login和password都输入msfadmin3、登录成功,进入系统4、测试如下:二、MySQL弱密码登录:1、在kali上执行mysql –h 192.168.26.129 –u root2、登录成功,进入MySQL系统3、测试效果:三、PostgreSQL弱密码登录1、在Kali上执行psql -h 192.168.26.129 –U post..._metasploitable2怎么进入
文章浏览阅读257次。本文将为初学者提供Python学习的详细指南,从Python的历史、基础语法和数据类型到面向对象编程、模块和库的使用。通过本文,您将能够掌握Python编程的核心概念,为今后的编程学习和实践打下坚实基础。_python人工智能开发从入门到精通pdf