Merge pull request #94 from Random-Liu/add-multiple-log-monitor-support

Add multiple system log monitor support

Dockerfile.in

@@ -20,4 +20,4 @@ RUN test -h /etc/localtime && rm -f /etc/localtime && cp /usr/share/zoneinfo/UTC
 
 ADD ./bin/node-problem-detector /node-problem-detector
 ADD config /config
-ENTRYPOINT ["/node-problem-detector", "--system-log-monitor=/config/kernel-monitor.json"]
+ENTRYPOINT ["/node-problem-detector", "--system-log-monitors=/config/kernel-monitor.json"]

README.md

@@ -54,8 +54,10 @@ List of supported problem daemons:
 # Usage
 ## Flags
 * `--version`: Print current version of node-problem-detector.
-* `--system-log-monitor`: The configuration used by the system log monitor, e.g.
+* `--system-log-monitors`: List of paths to system log monitor configuration files, comma separated, e.g.
   [config/kernel-monitor.json](https://github.com/kubernetes/node-problem-detector/blob/master/config/kernel-monitor.json).
+  Node problem detector will start a separate log monitor for each configuration. You can
+  use different log monitors to monitor different system log.
 * `--apiserver-override`: A URI parameter used to customize how node-problem-detector
 connects the apiserver. The format is same as the
 [`source`](https://github.com/kubernetes/heapster/blob/master/docs/source-configuration.md#kubernetes)

cmd/node_problem_detector.go

@@ -67,9 +67,17 @@ func main() {
 		os.Exit(0)
 	}
 
-	l := systemlogmonitor.NewLogMonitorOrDie(npdo.SystemLogMonitorConfigPath)
+	monitors := make(map[string]systemlogmonitor.LogMonitor)
+	for _, config := range npdo.SystemLogMonitorConfigPaths {
+		if _, ok := monitors[config]; ok {
+			// Skip the config if it's duplictaed.
+			glog.Warningf("Duplicated log monitor configuration %q", config)
+			continue
+		}
+		monitors[config] = systemlogmonitor.NewLogMonitorOrDie(config)
+	}
 	c := problemclient.NewClientOrDie(npdo)
-	p := problemdetector.NewProblemDetector(l, c)
+	p := problemdetector.NewProblemDetector(monitors, c)
 
 	// Start http server.
 	if npdo.ServerPort > 0 {

pkg/options/options.go

@@ -30,8 +30,9 @@ import (
 type NodeProblemDetectorOptions struct {
 	// command line options
 
-	// SystemLogMonitorConfigPath specifies the path to system log monitor configuration file.
-	SystemLogMonitorConfigPath string
+	// SystemLogMonitorConfigPaths specifies the list of paths to system log monitor configuration
+	// files.
+	SystemLogMonitorConfigPaths []string
 	// ApiServerOverride is the custom URI used to connect to Kubernetes ApiServer.
 	ApiServerOverride string
 	// PrintVersion is the flag determining whether version information is printed.
@@ -55,8 +56,8 @@ func NewNodeProblemDetectorOptions() *NodeProblemDetectorOptions {
 
 // AddFlags adds node problem detector command line options to pflag.
 func (npdo *NodeProblemDetectorOptions) AddFlags(fs *pflag.FlagSet) {
-	fs.StringVar(&npdo.SystemLogMonitorConfigPath, "system-log-monitor",
-		"/config/kernel-monitor.json", "The path to the system log monitor config file")
+	fs.StringSliceVar(&npdo.SystemLogMonitorConfigPaths, "system-log-monitors",
+		[]string{}, "List of paths to system log monitor config files, comma separated.")
 	fs.StringVar(&npdo.ApiServerOverride, "apiserver-override",
 		"", "Custom URI used to connect to Kubernetes ApiServer")
 	fs.BoolVar(&npdo.PrintVersion, "version", false, "Print version information and quit")

pkg/problemdetector/problem_detector.go

@@ -17,6 +17,7 @@ limitations under the License.
 package problemdetector
 
 import (
+	"fmt"
 	"net/http"
 
 	"github.com/golang/glog"
@@ -26,6 +27,7 @@ import (
 	"k8s.io/node-problem-detector/pkg/condition"
 	"k8s.io/node-problem-detector/pkg/problemclient"
 	"k8s.io/node-problem-detector/pkg/systemlogmonitor"
+	"k8s.io/node-problem-detector/pkg/types"
 	"k8s.io/node-problem-detector/pkg/util"
 )
 
@@ -38,28 +40,39 @@ type ProblemDetector interface {
 type problemDetector struct {
 	client           problemclient.Client
 	conditionManager condition.ConditionManager
-	// TODO(random-liu): Use slices of problem daemons if multiple monitors are needed in the future
-	monitor systemlogmonitor.LogMonitor
+	monitors         map[string]systemlogmonitor.LogMonitor
 }
 
 // NewProblemDetector creates the problem detector. Currently we just directly passed in the problem daemons, but
 // in the future we may want to let the problem daemons register themselves.
-func NewProblemDetector(monitor systemlogmonitor.LogMonitor, client problemclient.Client) ProblemDetector {
+func NewProblemDetector(monitors map[string]systemlogmonitor.LogMonitor, client problemclient.Client) ProblemDetector {
 	return &problemDetector{
 		client:           client,
 		conditionManager: condition.NewConditionManager(client, clock.RealClock{}),
-		monitor:          monitor,
+		monitors:         monitors,
 	}
 }
 
 // Run starts the problem detector.
 func (p *problemDetector) Run() error {
 	p.conditionManager.Start()
-	ch, err := p.monitor.Start()
-	if err != nil {
-		return err
+	// Start the log monitors one by one.
+	var chans []<-chan *types.Status
+	for cfg, m := range p.monitors {
+		ch, err := m.Start()
+		if err != nil {
+			// Do not return error and keep on trying the following config files.
+			glog.Errorf("Failed to start log monitor %q: %v", cfg, err)
+			continue
+		}
+		chans = append(chans, ch)
+	}
+	if len(chans) == 0 {
+		return fmt.Errorf("no log montior is successfully setup")
 	}
+	ch := groupChannel(chans)
 	glog.Info("Problem detector started")
+
 	for {
 		select {
 		case status := <-ch:
@@ -80,3 +93,15 @@ func (p *problemDetector) RegisterHTTPHandlers() {
 		util.ReturnHTTPJson(w, p.conditionManager.GetConditions())
 	})
 }
+
+func groupChannel(chans []<-chan *types.Status) <-chan *types.Status {
+	statuses := make(chan *types.Status)
+	for _, ch := range chans {
+		go func(c <-chan *types.Status) {
+			for status := range c {
+				statuses <- status
+			}
+		}(ch)
+	}
+	return statuses
+}

pkg/systemlogmonitor/log_monitor_test.go

@@ -17,10 +17,15 @@ limitations under the License.
 package systemlogmonitor
 
 import (
+	"fmt"
 	"reflect"
+	"runtime"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
+	watchertest "k8s.io/node-problem-detector/pkg/systemlogmonitor/logwatchers/testing"
 	logtypes "k8s.io/node-problem-detector/pkg/systemlogmonitor/types"
 	"k8s.io/node-problem-detector/pkg/types"
 )
@@ -131,3 +136,13 @@ func TestGenerateStatus(t *testing.T) {
 		}
 	}
 }
+
+func TestGoroutineLeak(t *testing.T) {
+	orignal := runtime.NumGoroutine()
+	f := watchertest.NewFakeLogWatcher(10)
+	f.InjectError(fmt.Errorf("unexpected error"))
+	l := &logMonitor{watcher: f}
+	_, err := l.Start()
+	assert.Error(t, err)
+	assert.Equal(t, orignal, runtime.NumGoroutine())
+}

pkg/systemlogmonitor/logwatchers/filelog/log_watcher_test.go

@@ -19,6 +19,7 @@ package filelog
 import (
 	"io/ioutil"
 	"os"
+	"runtime"
 	"testing"
 	"time"
 
@@ -170,3 +171,16 @@ Jan  2 03:04:05 kernel: [2.000000] 3
 		}
 	}
 }
+
+func TestGoroutineLeak(t *testing.T) {
+	orignal := runtime.NumGoroutine()
+	w := NewSyslogWatcherOrDie(types.WatcherConfig{
+		Plugin:       "filelog",
+		PluginConfig: getTestPluginConfig(),
+		LogPath:      "/not/exist/path",
+		Lookback:     "10m",
+	})
+	_, err := w.Watch()
+	assert.Error(t, err)
+	assert.Equal(t, orignal, runtime.NumGoroutine())
+}

pkg/systemlogmonitor/logwatchers/journald/log_watcher.go

@@ -20,6 +20,7 @@ package journald
 
 import (
 	"fmt"
+	"os"
 	"strings"
 	"time"
 
@@ -134,6 +135,11 @@ func getJournal(cfg types.WatcherConfig) (*sdjournal.Journal, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse lookback duration %q: %v", cfg.Lookback, err)
 	}
+	// If the path doesn't present, NewJournalFromDir will create it instead of
+	// returning error. So check the path existence ourselves.
+	if _, err := os.Stat(path); err != nil {
+		return nil, fmt.Errorf("failed to stat the log path %q: %v", path, err)
+	}
 	// Get journal client from the log path.
 	journal, err := sdjournal.NewJournalFromDir(path)
 	if err != nil {

pkg/systemlogmonitor/logwatchers/journald/log_watcher_test.go

@@ -19,12 +19,14 @@ limitations under the License.
 package journald
 
 import (
+	"runtime"
 	"testing"
 	"time"
 
 	"github.com/coreos/go-systemd/sdjournal"
 	"github.com/stretchr/testify/assert"
 
+	"k8s.io/node-problem-detector/pkg/systemlogmonitor/logwatchers/types"
 	logtypes "k8s.io/node-problem-detector/pkg/systemlogmonitor/types"
 )
 
@@ -62,3 +64,16 @@ func TestTranslate(t *testing.T) {
 		assert.Equal(t, test.log, translate(test.entry))
 	}
 }
+
+func TestGoroutineLeak(t *testing.T) {
+	orignal := runtime.NumGoroutine()
+	w := NewJournaldWatcher(types.WatcherConfig{
+		Plugin:       "journald",
+		PluginConfig: map[string]string{"source": "not-exist-service"},
+		LogPath:      "/not/exist/path",
+		Lookback:     "10m",
+	})
+	_, err := w.Watch()
+	assert.Error(t, err)
+	assert.Equal(t, orignal, runtime.NumGoroutine())
+}

pkg/systemlogmonitor/logwatchers/testing/fake_log_watcher.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2017 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package testing
+
+import (
+	"sync"
+
+	"k8s.io/node-problem-detector/pkg/systemlogmonitor/logwatchers/types"
+	logtypes "k8s.io/node-problem-detector/pkg/systemlogmonitor/types"
+)
+
+// FakeLogWatcher is a fake mock of log watcher.
+type FakeLogWatcher struct {
+	sync.Mutex
+	buf chan *logtypes.Log
+	err error
+}
+
+var _ types.LogWatcher = &FakeLogWatcher{}
+
+func NewFakeLogWatcher(bufferSize int) *FakeLogWatcher {
+	return &FakeLogWatcher{buf: make(chan *logtypes.Log, bufferSize)}
+}
+
+// InjectLog injects a fake log into the watch channel
+func (f *FakeLogWatcher) InjectLog(log *logtypes.Log) {
+	f.buf <- log
+}
+
+// InjectError injects an error of Watch function.
+func (f *FakeLogWatcher) InjectError(err error) {
+	f.Lock()
+	defer f.Unlock()
+	f.err = err
+}
+
+// Watch is the fake watch function.
+func (f *FakeLogWatcher) Watch() (<-chan *logtypes.Log, error) {
+	return f.buf, f.err
+}
+
+// Stop is the fake stop function.
+func (f *FakeLogWatcher) Stop() {
+	close(f.buf)
+}