深入浅出 kubernetes 之 WorkQueue 详解

发布于: 8 小时前
深入浅出kubernetes之WorkQueue详解

WorkQueue称为工作队列,Kubernetes的WorkQueue队列与普通FIFO(先进先出,First-In, First-Out)队列相比,实现略显复杂,它的主要功能在于标记和去重,并支持如下特性。

  • 有序:按照添加顺序处理元素(item)。

  • 去重:相同元素在同一时间不会被重复处理,例如一个元素在处理之前被添加了多次,它只会被处理一次。

  • 并发性:多生产者和多消费者。

  • 标记机制:支持标记功能,标记一个元素是否被处理,也允许元素在处理时重新排队。

  • 通知机制:ShutDown方法通过信号量通知队列不再接收新的元素,并通知metric goroutine退出。

  • 延迟:支持延迟队列,延迟一段时间后再将元素存入队列。

  • 限速:支持限速队列,元素存入队列时进行速率限制。限制一个元素被重新排队(Reenqueued)的次数。

  • Metric:支持metric监控指标,可用于Prometheus监控。

WorkQueue支持3种队列,并提供了3种接口,不同队列实现可应对不同的使用场景,分别介绍如下。

  • Interface:FIFO队列接口,先进先出队列,并支持去重机制。

  • DelayingInterface:延迟队列接口,基于Interface接口封装,延迟一段时间后再将元素存入队列。

  • RateLimitingInterface:限速队列接口,基于DelayingInterface接口封装,支持元素存入队列时进行速率限制。

FIFO队列

FIFO队列支持最基本的队列方法,例如插入元素、获取元素、获取队列长度等。另外,WorkQueue中的限速及延迟队列都基于Interface接口实现,其提供如下方法:

代码路径:vendor/k8s.io/client-go/util/workqueue/queue.go

type Interface interface {
Add(item interface{})
Len() int
Get() (item interface{}, shutdown bool)
Done(item interface{})
ShutDown()
ShuttingDown() bool
}

FIFO队列Interface方法说明如下。

  • Add:给队列添加元素(item),可以是任意类型元素。

  • Len:返回当前队列的长度。

  • Get:获取队列头部的一个元素。

  • Done:标记队列中该元素已被处理。

  • ShutDown:关闭队列。

  • ShuttingDown:查询队列是否正在关闭。

FIFO队列数据结构如下:

type Type struct {
queue []t
dirty set
processing set
cond *sync.Cond
shuttingDown bool
metrics queueMetrics
unfinishedWorkUpdatePeriod time.Duration
clock clock.Clock
}

FIFO队列数据结构中最主要的字段有queue、dirty和processing。其中queue字段是实际存储元素的地方,它是slice结构的,用于保证元素有序;dirty字段非常关键,除了能保证去重,还能保证在处理一个元素之前哪怕其被添加了多次(并发情况下),但也只会被处理一次;processing 字段用于标记机制,标记一个元素是否正在被处理。应根据WorkQueue的特性理解源码的实现,FIFO存储过程如图5-9所示。

图5-9  FIFO存储过程

通过Add方法往FIFO队列中分别插入1、2、3这3个元素,此时队列中的queue和dirty字段分别存有1、2、3元素,processing字段为空。然后通过Get方法获取最先进入的元素(也就是1元素),此时队列中的queue和dirty字段分别存有2、3元素,而1元素会被放入processing字段中,表示该元素正在被处理。最后,当我们处理完1元素时,通过Done方法标记该元素已经被处理完成,此时队列中的processing字段中的1元素会被删除。

如图5-9所示,这是FIFO队列的存储流程,在正常的情况下,FIFO队列运行在并发场景下。高并发下如何保证在处理一个元素之前哪怕其被添加了多次,但也只会被处理一次?下面进行讲解,FIFO并发存储过程如图5-10所示。

 图5-10  FIFO并发存储过程

如图5-10所示,在并发场景下,假设goroutine A通过Get方法获取1元素,1元素被添加到processing字段中,同一时间,goroutine B通过Add方法插入另一个1元素,此时在processing字段中已经存在相同的元素,所以后面的1元素并不会被直接添加到queue字段中,当前FIFO队列中的dirty字段中存有1、2、3元素,processing字段存有1元素。在goroutine A通过Done方法标记处理完成后,如果dirty字段中存有1元素,则将1元素追加到queue字段中的尾部。需要注意的是,dirty和processing字段都是用Hash Map数据结构实现的,所以不需要考虑无序,只保证去重即可。

延迟队列

延迟队列,基于FIFO队列接口封装,在原有功能上增加了AddAfter方法,其原理是延迟一段时间后再将元素插入FIFO队列。延迟队列数据结构如下:

代码路径:vendor/k8s.io/client-go/util/workqueue/delaying_queue.go

type DelayingInterface interface {
Interface
AddAfter(item interface{}, duration time.Duration)
}
type delayingType struct {
Interface
clock clock.Clock
stopCh chan struct{}
heartbeat clock.Ticker
waitingForAddCh chan *waitFor
metrics retryMetrics
deprecatedMetrics retryMetrics
}

AddAfter方法会插入一个item(元素)参数,并附带一个duration(延迟时间)参数,该duration参数用于指定元素延迟插入FIFO队列的时间。如果duration小于或等于0,会直接将元素插入FIFO队列中。

delayingType结构中最主要的字段是waitingForAddCh,其默认初始大小为1000,通过AddAfter方法插入元素时,是非阻塞状态的,只有当插入的元素大于或等于1000时,延迟队列才会处于阻塞状态。waitingForAddCh字段中的数据通过goroutine运行的waitingLoop函数持久运行。延迟队列运行原理如图5-11所示。

图5-11  延迟队列运行原理

如图5-11所示,将元素1放入waitingForAddCh字段中,通过waitingLoop函数消费元素数据。当元素的延迟时间不大于当前时间时,说明还需要延迟将元素插入FIFO队列的时间,此时将该元素放入优先队列(waitForPriorityQueue)中。当元素的延迟时间大于当前时间时,则将该元素插入FIFO队列中。另外,还会遍历优先队列(waitForPriorityQueue)中的元素,按照上述逻辑验证时间。

限速队列

限速队列,基于延迟队列和FIFO队列接口封装,限速队列接口(RateLimitingInterface)在原有功能上增加了AddRateLimited、Forget、NumRequeues方法。限速队列的重点不在于RateLimitingInterface接口,而在于它提供的4种限速算法接口(RateLimiter)。其原理是,限速队列利用延迟队列的特性,延迟某个元素的插入时间,达到限速目的。RateLimiter数据结构如下:

代码路径:vendor/k8s.io/client-go/util/workqueue/default_rate_limiters.go

type RateLimiter interface {
When(item interface{}) time.Duration
Forget(item interface{})
NumRequeues(item interface{}) int
}

限速队列接口方法说明如下。

  • When:获取指定元素应该等待的时间。

  • Forget:释放指定元素,清空该元素的排队数。

  • NumRequeues:获取指定元素的排队数。

下面会分别详解WorkQueue提供的4种限速算法,应对不同的场景,这4种限速算法分别如下。

  • 令牌桶算法(BucketRateLimiter)。

  • 排队指数算法(ItemExponentialFailureRateLimiter)。

  • 计数器算法(ItemFastSlowRateLimiter)。

  • 混合模式(MaxOfRateLimiter),将多种限速算法混合使用。

1. 令牌桶算法

令牌桶算法是通过Go语言的第三方库golang.org/x/time/rate实现的。令牌桶算法内部实现了一个存放token(令牌)的“桶”,初始时“桶”是空的,token会以固定速率往“桶”里填充,直到将其填满为止,多余的token会被丢弃。每个元素都会从令牌桶得到一个token,只有得到token的元素才允许通过(accept),而没有得到token的元素处于等待状态。令牌桶算法通过控制发放token来达到限速目的。令牌桶算法原理如图5-12所示。

图5-12  令牌桶算法原理

WorkQueue在默认的情况下会实例化令牌桶,代码示例如下:

rate.NewLimiter(rate.Limit(10), 100)

在实例化rate.NewLimiter后,传入r和b两个参数,其中r参数表示每秒往“桶”里填充的token数量,b参数表示令牌桶的大小(即令牌桶最多存放的token数量)。我们假定r为10,b为100。假设在一个限速周期内插入了1000个元素,通过r.Limiter.Reserve().Delay函数返回指定元素应该等待的时间,那么前b(即100)个元素会被立刻处理,而后面元素的延迟时间分别为item100/100ms、item101/200ms、item102/300ms、item103/400ms,以此类推。

2. 排队指数算法

排队指数算法将相同元素的排队数作为指数,排队数增大,速率限制呈指数级增长,但其最大值不会超过maxDelay。元素的排队数统计是有限速周期的,一个限速周期是指从执行AddRateLimited方法到执行完Forget方法之间的时间。如果该元素被Forget方法处理完,则清空排队数。排队指数算法的核心实现代码示例如下:

代码路径:vendor/k8s.io/client-go/util/workqueue/default_rate_limiters.go

r.failures[item] = r.failures[item] + 1
backoff := float64(r.baseDelay.Nanoseconds()) * math.Pow(2, float64(exp))
if backoff > math.MaxInt64 {
return r.maxDelay
}

该算法提供了3个主要字段:failures、baseDelay、maxDelay。其中,failures字段用于统计元素排队数,每当AddRateLimited方法插入新元素时,会为该字段加1;另外,baseDelay字段是最初的限速单位(默认为5ms),maxDelay字段是最大限速单位(默认为1000s)。排队指数增长趋势如图5-13所示。

图5-13  排队指数增长趋势

限速队列利用延迟队列的特性,延迟多个相同元素的插入时间,达到限速目的。

我们假定baseDelay是1 * time.Millisecond,maxDelay是1000 * time.Second。假设在一个限速周期内通过AddRateLimited方法插入10个相同元素,那么第1个元素会通过延迟队列的AddAfter方法插入并设置延迟时间为1ms(即baseDelay),第2个相同元素的延迟时间为2ms,第3个相同元素的延迟时间为4ms,第4个相同元素的延迟时间为8ms,第5个相同元素的延迟时间为16ms……第10个相同元素的延迟时间为512ms,最长延迟时间不超过1000s(即maxDelay)。

3. 计数器算法

计数器算法是限速算法中最简单的一种,其原理是:限制一段时间内允许通过的元素数量,例如在1分钟内只允许通过100个元素,每插入一个元素,计数器自增1,当计数器数到100的阈值且还在限速周期内时,则不允许元素再通过。但WorkQueue在此基础上扩展了fast和slow速率。

计数器算法提供了4个主要字段:failures、fastDelay、slowDelay及maxFastAttempts。其中,failures字段用于统计元素排队数,每当AddRateLimited方法插入新元素时,会为该字段加1;而fastDelay和slowDelay字段是用于定义fast、slow速率的;另外,maxFastAttempts字段用于控制从fast速率转换到slow速率。计数器算法核心实现的代码示例如下:

r.failures[item] = r.failures[item] + 1
if r.failures[item] <= r.maxFastAttempts {
return r.fastDelay
}
return r.slowDelay

假设fastDelay是5 * time.Millisecond,slowDelay是10 * time.Second,maxFastAttempts是3。在一个限速周期内通过AddRateLimited方法插入4个相同的元素,那么前3个元素使用fastDelay定义的fast速率,当触发maxFastAttempts字段时,第4个元素使用slowDelay定义的slow速率。

4. 混合模式

混合模式是将多种限速算法混合使用,即多种限速算法同时生效。例如,同时使用排队指数算法和令牌桶算法,代码示例如下:

func DefaultControllerRateLimiter() RateLimiter {
return NewMaxOfRateLimiter(
NewItemExponentialFailureRateLimiter(5*time.Millisecond, 1000*time.Second),
&BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
)
}

本文选自《Kubernetes源码剖析》一书

郑东旭著

电子工业出版社出版

购买链接:https://u.jd.com/3K5DNZ

发布于: 8 小时前 阅读数: 29
用户头像

还未添加个人签名 2019.10.21 加入

还未添加个人简介

评论 (1 条评论)

发布
用户头像
感谢博文编辑同学分享,我们首页推荐这篇。
4 小时前
回复
没有更多了
深入浅出kubernetes之WorkQueue详解