Refactor oom watcher to allow greater test coverage

This diff contains a strict refactor; there are no behavioral changes.

Address a long standing TODO in `oom_watcher_linux_test.go` around test
coverage. We refactor our `oom.Watcher` so it takes in a struct
fulfulling the `streamer` interface (i.e. defines `StreamOoms` method).
In production, we will continue to use the `oomparser` from `cadvisor`.
However, for testing purposes, we can now create our own `fakeStreamer`,
and control how it streams `oomparser.OomInstance`. With this fake, we
can implement richer unit testing for the `oom.Watcher` itself.

Actually adding the additional unit tests will come in a later commit.

pkg/kubelet/kubelet.go

@@ -472,7 +472,10 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 
 	containerRefManager := kubecontainer.NewRefManager()
 
-	oomWatcher := oomwatcher.NewWatcher(kubeDeps.Recorder)
+	oomWatcher, err := oomwatcher.NewWatcher(kubeDeps.Recorder)
+	if err != nil {
+		return nil, err
+	}
 
 	clusterDNS := make([]net.IP, 0, len(kubeCfg.ClusterDNS))
 	for _, ipEntry := range kubeCfg.ClusterDNS {

pkg/kubelet/oom/BUILD

@@ -68,11 +68,13 @@ go_test(
         "@io_bazel_rules_go//go/platform:android": [
             "//staging/src/k8s.io/api/core/v1:go_default_library",
             "//staging/src/k8s.io/client-go/tools/record:go_default_library",
+            "//vendor/github.com/google/cadvisor/utils/oomparser:go_default_library",
             "//vendor/github.com/stretchr/testify/assert:go_default_library",
         ],
         "@io_bazel_rules_go//go/platform:linux": [
             "//staging/src/k8s.io/api/core/v1:go_default_library",
             "//staging/src/k8s.io/client-go/tools/record:go_default_library",
+            "//vendor/github.com/google/cadvisor/utils/oomparser:go_default_library",
             "//vendor/github.com/stretchr/testify/assert:go_default_library",
         ],
         "//conditions:default": [],

pkg/kubelet/oom/oom_watcher_linux.go

@@ -30,29 +30,41 @@ import (
 	"github.com/google/cadvisor/utils/oomparser"
 )
 
+type streamer interface {
+	StreamOoms(chan<- *oomparser.OomInstance)
+}
+
+var _ streamer = &oomparser.OomParser{}
+
 type realWatcher struct {
 	recorder    record.EventRecorder
+	oomStreamer streamer
 }
 
 var _ Watcher = &realWatcher{}
 
-// NewWatcher creates and initializes a OOMWatcher based on parameters.
+// NewWatcher creates and initializes a OOMWatcher backed by Cadvisor as
-func NewWatcher(recorder record.EventRecorder) Watcher {
+// the oom streamer.
-	return &realWatcher{
+func NewWatcher(recorder record.EventRecorder) (Watcher, error) {
-		recorder: recorder,
+	oomStreamer, err := oomparser.New()
+	if err != nil {
+		return nil, err
 	}
+
+	watcher := &realWatcher{
+		recorder:    recorder,
+		oomStreamer: oomStreamer,
+	}
+
+	return watcher, nil
 }
 
 const systemOOMEvent = "SystemOOM"
 
 // Start watches for system oom's and records an event for every system oom encountered.
 func (ow *realWatcher) Start(ref *v1.ObjectReference) error {
-	oomLog, err := oomparser.New()
-	if err != nil {
-		return err
-	}
 	outStream := make(chan *oomparser.OomInstance, 10)
-	go oomLog.StreamOoms(outStream)
+	go ow.oomStreamer.StreamOoms(outStream)
 
 	go func() {
 		defer runtime.HandleCrash()

pkg/kubelet/oom/oom_watcher_linux_test.go

@@ -19,19 +19,26 @@ package oom
 import (
 	"testing"
 
-	"github.com/stretchr/testify/assert"
-
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/record"
+
+	"github.com/google/cadvisor/utils/oomparser"
+	"github.com/stretchr/testify/assert"
 )
 
 // TestBasic verifies that the OOMWatch works without error.
 func TestBasic(t *testing.T) {
 	fakeRecorder := &record.FakeRecorder{}
 	node := &v1.ObjectReference{}
-	oomWatcher := NewWatcher(fakeRecorder)
-	assert.NoError(t, oomWatcher.Start(node))
 
-	// TODO: Improve this test once cadvisor exports events.EventChannel as an interface
+	// TODO: Substitute this `oomStreamer` out for a fake, and then write
-	// and thereby allow using a mock version of cadvisor.
+	// more comprehensive unit tests of the actual behavior.
+	oomStreamer, err := oomparser.New()
+	assert.NoError(t, err)
+
+	oomWatcher := &realWatcher{
+		recorder:    fakeRecorder,
+		oomStreamer: oomStreamer,
+	}
+	assert.NoError(t, oomWatcher.Start(node))
 }

pkg/kubelet/oom/oom_watcher_unsupported.go

@@ -28,8 +28,8 @@ type oomWatcherUnsupported struct{}
 var _ Watcher = new(oomWatcherUnsupported)
 
 // NewWatcher creates a fake one here
-func NewWatcher(_ record.EventRecorder) Watcher {
+func NewWatcher(_ record.EventRecorder) (Watcher, error) {
-	return &oomWatcherUnsupported{}
+	return &oomWatcherUnsupported{}, nil
 }
 
 func (ow *oomWatcherUnsupported) Start(_ *v1.ObjectReference) error {