How should a watcher be terminated? about k8s HOT 6 CLOSED

@F21 do you know what the retry behavior for the upstream client is? I've generally avoided adding higher level APIs because:

• It's hard to inject loggers into higher level interfaces (client-go just assumes glog everwhere).
• It's hard to get it right, and hard to produce a typed interface without generating a lot more code.
• client-go already implements these interfaces.

I think I'd rather have this client provide raw API capabilities and provide examples like the one above for users who want the added features.

from k8s.

watcher.Next is actually just de-serializing from the HTTP long poll. So when Next fails, it indicate the watcher is dead and shouldn't be called again. Maybe I should provide better examples?

It sounds like you want to actually watch forever, handling each event as it comes in, then attempting to re-watch when the connection fails. Maybe something like this? (I haven't actually tested this)

type podEvent struct {
	event *versioned.Event
	pod   *v1.Pod
}

// watchPods watches pods in the client's namespace, calling handler on each
// event received from the watch. It watches until context is canceled.
func watchPods(ctx context.Context, client *k8s.Client, handler func(e podEvent)) {
	// Last observed resource version.
	resourceVersion := ""

	// Channels used for each instance of the watch.
	eventCh := make(chan podEvent, 1)
	errCh := make(chan error, 1)
	for {
		opt := k8s.ResourceVersion(resourceVersion)

		watcher, err := client.CoreV1().WatchPods(ctx, client.Namespace, opt)
		if err != nil {
			log.Printf("watch pods: %v", err)
			// Implement backoff behavior here.
			continue
		}

		for {
			go func() {
				event, pod, err := watcher.Next()
				if err != nil {
					errCh <- err
					return
				}
				eventCh <- podEvent{event, pod}
			}()

			select {
			case e := <-eventCh:
				// Update last observed resource version.
				resourceVersion = *e.pod.Metadata.ResourceVersion
				handler(e)
			case err := <-errCh:
				log.Printf("watch err: %v", err)
				break
			case <-ctx.Done():
				watcher.Close()
				return
			}
		}
	}
}

from k8s.

I think you bring up a good point though. Watches are a little different than client-go and that should be called out in the docs.

from k8s.

Hey, thanks! Your example makes a lot of sense.

It would be great if a higher level api is implemented, smilar to the one in the official k8s.io client: https://godoc.org/k8s.io/apimachinery/pkg/watch#Interface

It would then be pretty easy to use the watcher in a for select loop and read events over a channel.

from k8s.

@ericchiang Unfortunately, no. However, your reasoning makes a lot of sense!

from k8s.

watcher.Next is actually just de-serializing from the HTTP long poll. So when Next fails, it indicate the watcher is dead and shouldn't be called again.

In the above snippet, the code:

			case err := <-errCh:
				log.Printf("watch err: %v", err)
				break

Would just break from the select statement, causing us to restart the inner for-loop and re-using the existing watcher in a new goroutine with watcher.Next(), without reconnecting. Am I missing something?

from k8s.