Merge branch 'release/2.0.0' into develop

Author: rxie

Makefile

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

config.json

@@ -15,6 +15,48 @@
 			"description": "Set log level to output for plugin logs",
 			"value": "info",
 			"settable": ["value"]
+		},
+		{
+			"name": "SPLUNK_LOGGING_DRIVER_FIFO_ERROR_RETRY_TIME",
+			"description": "Set number of retry when reading fifo from docker failed. -1 means retry forever",
+			"value": "3",
+			"settable": ["value"]
+		},
+		{
+			"name": "SPLUNK_LOGGING_DRIVER_POST_MESSAGES_FREQUENCY",
+			"description": "Set how often do we send messages (if we are not reaching batch size)",
+			"value": "5s",
+			"settable": ["value"]
+		},
+		{
+			"name": "SPLUNK_LOGGING_DRIVER_POST_MESSAGES_BATCH_SIZE",
+			"description": "Set number of messages to batch before buffer timeout",
+			"value": "1000",
+			"settable": ["value"]
+		},
+		{
+			"name": "SPLUNK_LOGGING_DRIVER_BUFFER_MAX",
+			"description": "Set maximum number of messages wait in the buffer before sent to Splunk",
+			"value": "10000",
+			"settable": ["value"]
+		},
+		{
+			"name": "SPLUNK_LOGGING_DRIVER_CHANNEL_SIZE",
+			"description": "Set number of messages allowed to be queued in the channel when reading from the docker provided FIFO",
+			"value": "4000",
+			"settable": ["value"]
+		},
+		{
+			"name": "SPLUNK_LOGGING_DRIVER_TEMP_MESSAGES_HOLD_DURATION",
+			"description": "Used when logs that are chunked by docker with 16kb limit. Set how long the system can wait for the next message to come.",
+			"value": "100ms",
+			"settable": ["value"]
+		},
+		{
+			"name": "SPLUNK_LOGGING_DRIVER_TEMP_MESSAGES_BUFFER_SIZE",
+			"description": "Used when logs that are chunked by docker with 16kb limit. Set the biggest message that the system can reassemble.",
+			"value": "1048576",
+			"settable": ["value"]
 		}
 	]
 }

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
@@ -85,8 +99,6 @@ func (mg messageProcessor) sendMessage(l logger.Logger, buf *logdriver.LogEntry,
 	// Only send if partial bit is not set or temp buffer size reached max or temp buffer timer expired
 	// Check for temp buffer timer expiration
 	if !buf.Partial || t.shouldFlush(time.Now()) {
-		logrus.WithField("id", containerid).WithField("Buffer partial flag should be false:",
-			buf.Partial).WithField("Temp buffer Length:", t.tBuf.Len()).Debug("Buffer details")
 		msg.Line = t.tBuf.Bytes()
 		msg.Source = buf.Source
 		msg.Partial = buf.Partial

partial_message_buffer.go

@@ -50,7 +50,6 @@ func (b *partialMsgBuffer) append(l *logdriver.LogEntry) (err error) {
 
 func (b *partialMsgBuffer) reset() {
 	if b.bufferReset {
-		logrus.Debug("Resetting temp buffer")
 		b.tBuf.Reset()
 		b.bufferTimer = time.Now()
 	}
@@ -67,4 +66,4 @@ func (b *partialMsgBuffer) hasLengthExceeded() bool {
 
 func (b *partialMsgBuffer) shouldFlush(t time.Time) bool {
 	return b.hasLengthExceeded() || b.hasHoldDurationExpired(t)
-}
+}

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 {