04.refactor
介绍
我们接下来看看 refactor.go 这个文件
Reflector 是保证 Informer 可靠性的核心组件,在丢失事件,收到异常事件,处理事件失败等多种异常情况下需要考虑的细节很多。单独的listwatcher缺少重新连接和重新同步机制,有可能出现数据不一致问题。其对事件响应是同步的,如果执行复杂的操作会引起阻塞,需要引入队列。
Reflector可以成为反射器,将etcd中的数据反射到存储(DeltaFIFO)中。Reflector通过其内部的List操作获取所有资源对象数据,保存到本地存储,之后Watch监视资源变化,触发对应事件处理,例如Add、Update、Delete等。
结构体
// Reflector watches a specified resource and causes all changes to be reflected in the given store.
type Reflector struct {
// 标识此 Reflector 的名称,默认为文件名和行号
name string
// 我们期望存储在存储中的类型的名称。如果提供了 expectedGVK,则名称将是其字符串化形式,否则为 expectedType 的字符串化形式。仅用于显示,不应用于解析或比较
expectedTypeName string
// 我们期望存储在存储中的类型的示例对象。只有 type 需要正确,但当类型为 unstructured.Unstructured 时,对象的 "apiVersion" 和 "kind" 也必须正确。
expectedType reflect.Type
// 如果对象类型为 unstructured,我们期望存储在存储中的对象的 GroupVersionKind
expectedGVK *schema.GroupVersionKind
// 与监视源同步的目标
store Store
// 用于执行列表和监视操作的对象。
listerWatcher ListerWatcher
// 管理 ListWatch 的退避
backoffManager wait.BackoffManager
// 管理与 ListAndWatch 的 Watch 调用的初始连接的退避。
initConnBackoffManager wait.BackoffManager
//重新同步周期。
resyncPeriod time.Duration
// 定期调用此函数,当返回 true 时,将调用 Store 的 Resync 操作。
ShouldResync func() bool
// 允许测试操纵时间的时钟对象。
clock clock.Clock
// 定义是否应为列表调用强制分页。根据初始列表调用的结果进行设置。
paginatedResult bool
// 上次同步时观察到的资源版本令牌,与底层存储同步,但是线程安全的
lastSyncResourceVersion string
// 如果上次使用 lastSyncResourceVersion 的列表或监视请求由于“过期”或“资源版本过大”错误而失败,则为 true
isLastSyncResourceVersionUnavailable bool
// 保护对 lastSyncResourceVersion 的读/写访问的互斥锁。
lastSyncResourceVersionMutex sync.RWMutex
// 初始和重新同步监视列表的请求块大小。如果未设置,对于一致性读取(RV = "")或选择任意旧数据的读取(RV = "0"),它将默认为 pager.PageSize,对于其他情况(RV != "" && RV != "0"),它将关闭分页以允许从监视缓存中提供服务。
WatchListPageSize int64
//当 ListAndWatch 因错误而中断连接时调用的函数
watchErrorHandler WatchErrorHandler
Run流程图
详细
// ResourceVersionUpdater is an interface that allows store implementation to
// track the current resource version of the reflector. This is especially
// important if storage bookmarks are enabled.
type ResourceVersionUpdater interface {
// UpdateResourceVersion is called each time current resource version of the reflector
// is updated.
UpdateResourceVersion(resourceVersion string)
}
ResourceVersionUpdater 接口:允许存储实现跟踪 reflector 的当前资源版本。当启用存储书签时,这尤为重要。接口定义了一个方法 UpdateResourceVersion,每次 reflector 的当前资源版本更新时都会调用它。
// The WatchErrorHandler is called whenever ListAndWatch drops the
// connection with an error. After calling this handler, the informer
// will backoff and retry.
//
// The default implementation looks at the error type and tries to log
// the error message at an appropriate level.
//
// Implementations of this handler may display the error message in other
// ways. Implementations should return quickly - any expensive processing
// should be offloaded.
type WatchErrorHandler func(r *Reflector, err error)
// DefaultWatchErrorHandler is the default implementation of WatchErrorHandler
func DefaultWatchErrorHandler(r *Reflector, err error) {
switch {
case isExpiredError(err):
// Don't set LastSyncResourceVersionUnavailable - LIST call with ResourceVersion=RV already
// has a semantic that it returns data at least as fresh as provided RV.
// So first try to LIST with setting RV to resource version of last observed object.
klog.V(4).Infof("%s: watch of %v closed with: %v", r.name, r.expectedTypeName, err)
case err == io.EOF:
// watch closed normally
case err == io.ErrUnexpectedEOF:
klog.V(1).Infof("%s: Watch for %v closed with unexpected EOF: %v", r.name, r.expectedTypeName, err)
default:
utilruntime.HandleError(fmt.Errorf("%s: Failed to watch %v: %v", r.name, r.expectedTypeName, err))
}
}
WatchErrorHandler 类型:一个函数类型,当 ListAndWatch 由于错误而断开连接时调用。调用此处理程序后,informer 将退避并重试。默认实现根据错误类型尝试以适当的级别记录错误消息。此处理程序的实现可以以其他方式显示错误消息。实现应快速返回 - 任何昂贵的处理都应卸载。
DefaultWatchErrorHandler 函数:是 WatchErrorHandler 的默认实现。
// NewNamespaceKeyedIndexerAndReflector creates an Indexer and a Reflector
// The indexer is configured to key on namespace
func NewNamespaceKeyedIndexerAndReflector(lw ListerWatcher, expectedType interface{}, resyncPeriod time.Duration) (indexer Indexer, reflector *Reflector) {
indexer = NewIndexer(MetaNamespaceKeyFunc, Indexers{NamespaceIndex: MetaNamespaceIndexFunc})
reflector = NewReflector(lw, expectedType, indexer, resyncPeriod)
return indexer, reflector
}
// NewReflector creates a new Reflector object which will keep the
// given store up to date with the server's contents for the given
// resource. Reflector promises to only put things in the store that
// have the type of expectedType, unless expectedType is nil. If
// resyncPeriod is non-zero, then the reflector will periodically
// consult its ShouldResync function to determine whether to invoke
// the Store's Resync operation; `ShouldResync==nil` means always
// "yes". This enables you to use reflectors to periodically process
// everything as well as incrementally processing the things that
// change.
func NewReflector(lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
return NewNamedReflector(naming.GetNameFromCallsite(internalPackages...), lw, expectedType, store, resyncPeriod)
}
NewNamespaceKeyedIndexerAndReflector 函数:创建一个配置为基于命名空间键的 Indexer 和一个 Reflector。
NewReflector 函数:创建一个新的 Reflector 对象,该对象将保持给定的存储与服务器的内容保持一致。Reflector 保证只将具有 expectedType 类型的对象放入存储,除非 expectedType 为 nil。如果 resyncPeriod 不为零,则 reflector 将定期调用其 ShouldResync 函数,以确定是否调用 Store 的 Resync 操作;ShouldResync == nil 表示始终为 "yes"。这使您能够使用 reflector 来定期处理所有内容以及增量处理发生更改的内容。
NewNamedReflector 函数:与 NewReflector 相同,但具有指定的用于记录的名称。
// Run repeatedly uses the reflector's ListAndWatch to fetch all the
// objects and subsequent deltas.
// Run will exit when stopCh is closed.
func (r *Reflector) Run(stopCh <-chan struct{}) {
klog.V(3).Infof("Starting reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
wait.BackoffUntil(func() {
if err := r.ListAndWatch(stopCh); err != nil {
r.watchErrorHandler(r, err)
}
}, r.backoffManager, true, stopCh)
klog.V(3).Infof("Stopping reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
}
这段代码定义了一个名为 Run 的方法,它属于 Reflector 结构体。该方法使用反射器的 ListAndWatch 方法重复获取所有对象及其后续增量。当 stopCh 被关闭时,Run 将退出。
以下是此方法的详细解释:
在开始运行 reflector 时,会使用 klog.V(3).Infof() 打印日志,记录 reflector 的开始、期望的类型名和反射器名称。
wait.BackoffUntil() 是一个带退避策略的无限循环,直到 stopCh 被关闭。该函数接受一个函数、一个退避管理器、一个布尔值(指示是否在启动时立即执行函数),以及一个表示停止信号的通道。在这个例子中,它会重复执行以下操作:
调用 r.ListAndWatch(stopCh),尝试获取对象及其后续增量。如果执行过程中发生错误,返回该错误。 调用 r.watchErrorHandler(r, err) 来处理 ListAndWatch 返回的错误。错误处理器可能会记录错误或采取其他措施。 当 stopCh 被关闭时,wait.BackoffUntil() 函数将退出,最后使用 klog.V(3).Infof() 打印日志,记录 reflector 停止的信息。
总之,Run 方法用于启动 reflector,它会持续运行,直到接收到停止信号。在运行过程中,它将尝试获取对象及其后续增量,并处理可能出现的错误
核心函数 ListAndWatch
这段代码实现了 Reflector 结构体的 ListAndWatch 方法,该方法首先列出所有对象及其资源版本,然后使用资源版本进行监听。以下是代码的主要部分及其功能:
ListAndWatch 方法接收一个 stopCh 通道作为参数,用于在需要停止的时候通知方法终止。
定义一个名为 resourceVersion 的变量,用于存储资源版本。
使用当前资源版本创建一个 metav1.ListOptions 对象。
使用一个匿名函数并立即执行来列出对象。在此过程中,会尝试分页列出对象,并处理可能出现的错误,例如资源版本不可用或过大。成功列出对象后,会提取对象和资源版本,然后将提取到的对象与存储中的数据同步。
创建一个 goroutine 来处理重新同步操作。当接收到重新同步信号或停止信号时,根据 r.ShouldResync 的值强制进行重新同步,并调用 r.store.Resync()。
创建一个无限循环来监听对象更改。在循环中,首先创建一个带有超时时间和允许使用书签的 ListOptions 对象。然后调用 r.listerWatcher.Watch(options) 进行监听。如果监听失败,根据错误类型采取不同的措施,例如重新连接、等待退避时间等。接着调用 r.watchHandler() 进行监听处理,如果监听处理失败,根据错误类型采取不同的措施,例如重新连接、等待退避时间等。
ListAndWatch 方法的主要目的是列出并监听对象的更改,以便在运行过程中同步数据并处理可能出现的错误。通过这种方式,Reflector 可以确保其缓存数据始终与 API 服务器上的数据保持一致。