Fix high memory cpu (#34)

- close all the logger and streams eventually when error happens to prevent the program stuck in memory/cpu loop
- retry for errors and make retry number configurable through env variables except for non-retriable errors: EOF and "file already closed"

Author: sharonx

Makefile

@@ -1,5 +1,5 @@
 PLUGIN_NAME=splunk/docker-logging-plugin
-PLUGIN_TAG=2.0.0-rc1
+PLUGIN_TAG=2.0.0-rc2
 PLUGIN_DIR=./splunk-logging-plugin
 
 all: clean docker rootfs create

driver.go

@@ -51,6 +51,12 @@ type logPair struct {
 	info    logger.Info
 }
 
+func (lf *logPair) Close() {
+	lf.stream.Close()
+	lf.splunkl.Close()
+	lf.jsonl.Close()
+}
+
 func newDriver() *driver {
 	return &driver{
 		logs: make(map[string]*logPair),
@@ -108,7 +114,9 @@ func (d *driver) StartLogging(file string, logCtx logger.Info) error {
 
 	// start to process the logs generated by docker
 	logrus.Debug("Start processing messages")
-	mg := &messageProcessor{}
+	mg := &messageProcessor{
+		retryNumber: getAdvancedOptionInt(envVarReadFifoErrorRetryNumber, defaultReadFifoErrorRetryNumber),
+	}
 	go mg.process(lf)
 	return nil
 }
@@ -118,7 +126,7 @@ func (d *driver) StopLogging(file string) error {
 	d.mu.Lock()
 	lf, ok := d.logs[file]
 	if ok {
-		lf.stream.Close()
+		lf.Close()
 		delete(d.logs, file)
 	}
 	d.mu.Unlock()

message_processor.go

@@ -20,6 +20,8 @@ import (
 	"bytes"
 	"encoding/binary"
 	"io"
+	"os"
+	"strings"
 	"time"
 	"unicode/utf8"
 
@@ -30,6 +32,7 @@ import (
 )
 
 type messageProcessor struct {
+	retryNumber int
 }
 
 func (mg messageProcessor) process(lf *logPair) {
@@ -49,21 +52,32 @@ func (mg messageProcessor) consumeLog(lf *logPair) {
 	// create a protobuf reader for the log stream
 	dec := protoio.NewUint32DelimitedReader(lf.stream, binary.BigEndian, 1e6)
 	defer dec.Close()
+	defer lf.Close()
 	// a temp buffer for each log entry
 	var buf logdriver.LogEntry
+	curRetryNumber := 0
 	for {
-		// reads a message from the log stream and put it in a buffer until the EOF
-		// if there is any other error, recreate the stream reader
+		// reads a message from the log stream and put it in a buffer
 		if err := dec.ReadMsg(&buf); err != nil {
-			if err == io.EOF {
-				logrus.WithField("id", lf.info.ContainerID).WithError(err).Debug("shutting down log logger")
-				lf.stream.Close()
+			// exit the loop if reader reaches EOF or the fifo is closed by the writer
+			if err == io.EOF || err == os.ErrClosed || strings.Contains(err.Error(), "file already closed") {
+				logrus.WithField("id", lf.info.ContainerID).WithError(err).Info("shutting down loggers")
 				return
 			}
 
-			logrus.WithField("id", lf.info.ContainerID).WithError(err).Debug("Ignoring error")
+			// exit the loop if retry number reaches the specified number
+			if mg.retryNumber != -1 && curRetryNumber > mg.retryNumber {
+				logrus.WithField("id", lf.info.ContainerID).WithField("curRetryNumber", curRetryNumber).WithField("retryNumber", mg.retryNumber).WithError(err).Error("Stop retrying. Shutting down loggers")
+				return
+			}
+
+			// if there is any other error, retry for robustness. If retryNumber is -1, retry forever
+			curRetryNumber++
+			logrus.WithField("id", lf.info.ContainerID).WithField("curRetryNumber", curRetryNumber).WithField("retryNumber", mg.retryNumber).WithError(err).Error("Encountered error and retrying")
+			time.Sleep(500 * time.Millisecond)
 			dec = protoio.NewUint32DelimitedReader(lf.stream, binary.BigEndian, 1e6)
 		}
+		curRetryNumber = 0
 
 		if mg.shouldSendMessage(buf.Line) {
 			// Append to temp buffer

splunk_logger.go

@@ -72,6 +72,9 @@ const (
 	defaultPartialMsgBufferHoldDuration = 100 * time.Millisecond
 	// Maximum buffer size for partial logging
 	defaultPartialMsgBufferMaximum = 1024 * 1024
+	// Number of retry if error happens while reading logs from docker provided fifo
+	// -1 means retry forever
+	defaultReadFifoErrorRetryNumber = 3
 )
 
 const (
@@ -81,6 +84,7 @@ const (
 	envVarStreamChannelSize            = "SPLUNK_LOGGING_DRIVER_CHANNEL_SIZE"
 	envVarPartialMsgBufferHoldDuration = "SPLUNK_LOGGING_DRIVER_TEMP_MESSAGES_HOLD_DURATION"
 	envVarPartialMsgBufferMaximum      = "SPLUNK_LOGGING_DRIVER_TEMP_MESSAGES_BUFFER_SIZE"
+	envVarReadFifoErrorRetryNumber     = "SPLUNK_LOGGING_DRIVER_FIFO_ERROR_RETRY_TIME"
 )
 
 type splunkLoggerInterface interface {