a repeatable client-go watcher with timeout

client-go-watcher.go

package main

import (
	"flag"
	"fmt"
	"github.com/golang/glog"
	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/watch"
	"strings"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
)

var (
	kubeconfig = flag.String("kubeconfig", "./config", "absolute path to the kubeconfig file")
)

func main() {
	flag.Parse()
	config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
	if err != nil {
		panic(err.Error())
	}
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		panic(err.Error())
	}
	timeout := int64(40)
	options := metav1.ListOptions{
		LabelSelector:  "nylabel=test",
		TimeoutSeconds: &timeout,
	}
	//watchlist := cache.NewListWatchFromClient(clientset.CoreV1().RESTClient(), "pods", "antstack",
	//	fields.Everything())
	//watcher, err := watchlist.Watch(options)
	// set timeout
	timer := time.NewTimer(2 * time.Minute)
	glog.Infof("starting watcher")
	for {
		watcher, err := clientset.CoreV1().Pods("antstack").Watch(options)
		if err != nil {
			panic("failed to start a new watcher")
		}
		err = watchOnce(watcher, timer)
		if strings.Contains(err.Error(), "timeout") {
			glog.Error(err.Error())
			return
		}
		watcher.Stop()
	}
}

func watchOnce(w watch.Interface, timer *time.Timer) error {
	var err error
loop:
	for {
		select {
		case pod, ok := <-w.ResultChan():
			if !ok {
				glog.Infof("alb watcher closed , breaking out.")
				err = fmt.Errorf("unexpected close")
				break loop
			}
			newPod, ok := pod.Object.(*v1.Pod)
			if !ok {
				glog.Errorf("invalid object watched: %#v", newPod)
				continue
			}
			glog.Infof("watch event for pod: %s", newPod.Name)
		case <-timer.C:
			glog.Errorf("timeout")

			return fmt.Errorf("timeout error")
		}
	}
	return err
}