第一種方法
使用k8s.io/client-go/tools/cache
func StartWatchingServices(c *Client) {
watchlist := cache.NewListWatchFromClient(
c.K8sClientset.CoreV1().RESTClient(),
string(v1.ResourceServices),
v1.NamespaceAll,
fields.Everything(),
)
_, controller := cache.NewInformer(
watchlist,
&v1.Service{},
0, //Duration is int64
cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
fmt.Printf("added: %s \n", obj)
},
DeleteFunc: func(obj interface{}) {
fmt.Printf(deleted: %s \n", obj)
},
UpdateFunc: func(oldObj, newObj interface{}) {
fmt.Printf("changed \n")
},
},
)
stop := make(chan struct{})
defer close(stop)
go controller.Run(stop)
for {
time.Sleep(time.Second)
}
}
第二種方法
使用k8s.io/client-go/informers
骆膝,可以watch CRD資源但需要使用code-generator生成clientset和informer(指CRD資源)祭衩。
func StartWatchingServices2(c *Client) {
kubeInformerFactory := informers.NewSharedInformerFactory(c.K8sClientset, time.Second*30)
svcInformer := kubeInformerFactory.Core().V1().Services().Informer()
svcInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
fmt.Printf("added: %s \n", obj)
},
DeleteFunc: func(obj interface{}) {
fmt.Printf("deleted: %s \n", obj)
},
UpdateFunc: func(oldObj, newObj interface{}) {
fmt.Printf("changed: %s \n", newObj)
},
})
stop := make(chan struct{})
defer close(stop)
kubeInformerFactory.Start(stop)
for {
time.Sleep(time.Second)
}
}
第三種方法
使用runtimeCache "sigs.k8s.io/controller-runtime/pkg/cache"
, 可以watch CRD資源,無需特殊處理阅签。下例v1alpha1.Component
是CRD資源掐暮。
func StartWatchingComponents(c *Client) {
informerCache, err := runtimeCache.New(c.RestConfig, runtimeCache.Options{})
if err != nil {
panic(err)
}
informer, err := informerCache.GetInformer(&v1alpha1.Component{})
if err != nil {
panic(err)
}
informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
fmt.Printf("added: %s \n", obj)
},
DeleteFunc: func(obj interface{}) {
fmt.Printf("deleted: %s \n", obj)
},
UpdateFunc: func(oldObj, newObj interface{}) {
fmt.Printf("changed: %s \n", newObj)
},
})
stop := make(chan struct{})
defer close(stop)
informerCache.Start(stop)
// for {
// time.Sleep(time.Second)
// }
}