listwatchext

README

介绍

这个包是对 listwatch 的一个补充让它更加适应我们的场景.

1. 初始化是时进行一次全量list, 然后同步到服务端
2. 然后开始监听增量, 并进行适当合并(比如key相同的只保留最后一条元数据), 然后批量同步到服务端
3. 一旦同步到服务端发生不可知错误, 需要重新开始listwatch, 这样才能保证元数据是不丢失的

直接使用k8s的cache包可以做到类似的效果, 但缺点是我们拦截不到第一次的全量list的结果. 它的第一次全量list对我们来说只是n次add.
而我们会使用全量list的结果来做全量数据同步, 包含删除已经不存在的pod.

Documentation

Index

• type BatchEventConsumer
• type EventMerger
  • func NewEventMerger(interval time.Duration, consumer BatchEventConsumer, stopCh <-chan struct{}) *EventMerger
  • func (m *EventMerger) Add(e watch.Event)
• type ListFunc
• type ListResult
• type ListWatch
  • func NewListWatchFromClient(c cache.Getter, resource string, namespace string) *ListWatch
• type ListWatchCallback
• type ListWatchHelper
  • func NewListWatchHelper(lw *ListWatch, callback ListWatchCallback) *ListWatchHelper
  • func (h *ListWatchHelper) Run(stop <-chan struct{})
• type WatchFunc

Types

type BatchEventConsumer

type BatchEventConsumer func(merged map[string]watch.Event)

type EventMerger

type EventMerger struct {
	// contains filtered or unexported fields
}

func NewEventMerger

func NewEventMerger(interval time.Duration, consumer BatchEventConsumer, stopCh <-chan struct{}) *EventMerger

func (*EventMerger) Add

func (m *EventMerger) Add(e watch.Event)

type ListFunc

type ListFunc func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error)

type ListResult

type ListResult struct {
	Result          runtime.Object
	ResourceVersion string
	PaginatedResult bool
	Err             error
}

type ListWatch

type ListWatch struct {
	ListFunc  ListFunc
	WatchFunc WatchFunc
}

func NewListWatchFromClient

func NewListWatchFromClient(c cache.Getter, resource string, namespace string) *ListWatch

type ListWatchCallback

type ListWatchCallback struct {
	// 用户的方法不允许返回异常, 如果想返回异常, 请自己处理并调用reset重新开始监听
	OnList  func(items []runtime.Object)
	OnEvent func(e watch.Event)
}

type ListWatchHelper

type ListWatchHelper struct {
	// contains filtered or unexported fields
}

func NewListWatchHelper

func NewListWatchHelper(lw *ListWatch, callback ListWatchCallback) *ListWatchHelper

func (*ListWatchHelper) Run

func (h *ListWatchHelper) Run(stop <-chan struct{})

type WatchFunc

type WatchFunc func(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error)