Work on #61: Add support for ICMP

+ Update dependencies

vendor/k8s.io/apimachinery/pkg/watch/streamwatcher.go

@@ -17,13 +17,15 @@ limitations under the License.
 package watch
 
 import (
+	"fmt"
 	"io"
 	"sync"
 
+	"k8s.io/klog/v2"
+
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/net"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/klog"
 )
 
 // Decoder allows StreamWatcher to watch any stream for which a Decoder can be written.
@@ -39,19 +41,28 @@ type Decoder interface {
 	Close()
 }
 
+// Reporter hides the details of how an error is turned into a runtime.Object for
+// reporting on a watch stream since this package may not import a higher level report.
+type Reporter interface {
+	// AsObject must convert err into a valid runtime.Object for the watch stream.
+	AsObject(err error) runtime.Object
+}
+
 // StreamWatcher turns any stream for which you can write a Decoder interface
 // into a watch.Interface.
 type StreamWatcher struct {
 	sync.Mutex
-	source  Decoder
-	result  chan Event
-	stopped bool
+	source   Decoder
+	reporter Reporter
+	result   chan Event
+	stopped  bool
 }
 
 // NewStreamWatcher creates a StreamWatcher from the given decoder.
-func NewStreamWatcher(d Decoder) *StreamWatcher {
+func NewStreamWatcher(d Decoder, r Reporter) *StreamWatcher {
 	sw := &StreamWatcher{
-		source: d,
+		source:   d,
+		reporter: r,
 		// It's easy for a consumer to add buffering via an extra
 		// goroutine/channel, but impossible for them to remove it,
 		// so nonbuffered is better.
@@ -102,11 +113,13 @@ func (sw *StreamWatcher) receive() {
 			case io.ErrUnexpectedEOF:
 				klog.V(1).Infof("Unexpected EOF during watch stream event decoding: %v", err)
 			default:
-				msg := "Unable to decode an event from the watch stream: %v"
-				if net.IsProbableEOF(err) {
-					klog.V(5).Infof(msg, err)
+				if net.IsProbableEOF(err) || net.IsTimeout(err) {
+					klog.V(5).Infof("Unable to decode an event from the watch stream: %v", err)
 				} else {
-					klog.Errorf(msg, err)
+					sw.result <- Event{
+						Type:   Error,
+						Object: sw.reporter.AsObject(fmt.Errorf("unable to decode an event from the watch stream: %v", err)),
+					}
 				}
 			}
 			return

vendor/k8s.io/apimachinery/pkg/watch/watch.go

@@ -20,7 +20,7 @@ import (
 	"fmt"
 	"sync"
 
-	"k8s.io/klog"
+	"k8s.io/klog/v2"
 
 	"k8s.io/apimachinery/pkg/runtime"
 )
@@ -32,8 +32,8 @@ type Interface interface {
 	Stop()
 
 	// Returns a chan which will receive all the events. If an error occurs
-	// or Stop() is called, this channel will be closed, in which case the
-	// watch should be completely cleaned up.
+	// or Stop() is called, the implementation will close this channel and
+	// release any resources used by the watch.
 	ResultChan() <-chan Event
 }
 
@@ -44,8 +44,11 @@ const (
 	Added    EventType = "ADDED"
 	Modified EventType = "MODIFIED"
 	Deleted  EventType = "DELETED"
+	Bookmark EventType = "BOOKMARK"
 	Error    EventType = "ERROR"
+)
 
+var (
 	DefaultChanSize int32 = 100
 )
 
@@ -57,6 +60,10 @@ type Event struct {
 	// Object is:
 	//  * If Type is Added or Modified: the new state of the object.
 	//  * If Type is Deleted: the state of the object immediately before deletion.
+	//  * If Type is Bookmark: the object (instance of a type being watched) where
+	//    only ResourceVersion field is set. On successful restart of watch from a
+	//    bookmark resourceVersion, client is guaranteed to not get repeat event
+	//    nor miss any events.
 	//  * If Type is Error: *api.Status is recommended; other types may make sense
 	//    depending on context.
 	Object runtime.Object
@@ -85,7 +92,7 @@ func (w emptyWatch) ResultChan() <-chan Event {
 // FakeWatcher lets you test anything that consumes a watch.Interface; threadsafe.
 type FakeWatcher struct {
 	result  chan Event
-	Stopped bool
+	stopped bool
 	sync.Mutex
 }
 
@@ -105,24 +112,24 @@ func NewFakeWithChanSize(size int, blocking bool) *FakeWatcher {
 func (f *FakeWatcher) Stop() {
 	f.Lock()
 	defer f.Unlock()
-	if !f.Stopped {
+	if !f.stopped {
 		klog.V(4).Infof("Stopping fake watcher.")
 		close(f.result)
-		f.Stopped = true
+		f.stopped = true
 	}
 }
 
 func (f *FakeWatcher) IsStopped() bool {
 	f.Lock()
 	defer f.Unlock()
-	return f.Stopped
+	return f.stopped
 }
 
 // Reset prepares the watcher to be reused.
 func (f *FakeWatcher) Reset() {
 	f.Lock()
 	defer f.Unlock()
-	f.Stopped = false
+	f.stopped = false
 	f.result = make(chan Event)
 }