Code review changes

Signed-off-by: Justin Menga <justin.menga@gmail.com>

daemon/logger/awslogs/cloudwatchlogs.go

@@ -93,29 +93,6 @@ func init() {
 	}
 }
 
-// Parses awslogs-multiline-pattern and awslogs-datetime-format options
-// If awslogs-datetime-format is present, convert the format from strftime
-// to regexp and return.
-// If awslogs-multiline-pattern is present, compile regexp and return
-func parseMultilineOptions(info logger.Info) (*regexp.Regexp, error) {
-	dateTimeFormat := info.Config[datetimeFormatKey]
-	multilinePatternKey := info.Config[multilinePatternKey]
-	if dateTimeFormat != "" {
-		r := regexp.MustCompile("%.")
-		multilinePatternKey = r.ReplaceAllStringFunc(dateTimeFormat, func(s string) string {
-			return strftimeToRegex[s]
-		})
-	}
-	if multilinePatternKey != "" {
-		multilinePattern, err := regexp.Compile(multilinePatternKey)
-		if err != nil {
-			return nil, errors.Wrapf(err, "awslogs could not parse multiline pattern key %q", multilinePatternKey)
-		}
-		return multilinePattern, nil
-	}
-	return nil, nil
-}
-
 // New creates an awslogs logger using the configuration passed in on the
 // context.  Supported context configuration variables are awslogs-region,
 // awslogs-group, awslogs-stream, awslogs-create-group, awslogs-multiline-pattern
@@ -167,6 +144,56 @@ func New(info logger.Info) (logger.Logger, error) {
 	return containerStream, nil
 }
 
+// Parses awslogs-multiline-pattern and awslogs-datetime-format options
+// If awslogs-datetime-format is present, convert the format from strftime
+// to regexp and return.
+// If awslogs-multiline-pattern is present, compile regexp and return
+func parseMultilineOptions(info logger.Info) (*regexp.Regexp, error) {
+	dateTimeFormat := info.Config[datetimeFormatKey]
+	multilinePatternKey := info.Config[multilinePatternKey]
+	// strftime input is parsed into a regular expression
+	if dateTimeFormat != "" {
+		// %. matches each strftime format sequence and ReplaceAllStringFunc
+		// looks up each format sequence in the conversion table strftimeToRegex
+		// to replace with a defined regular expression
+		r := regexp.MustCompile("%.")
+		multilinePatternKey = r.ReplaceAllStringFunc(dateTimeFormat, func(s string) string {
+			return strftimeToRegex[s]
+		})
+	}
+	if multilinePatternKey != "" {
+		multilinePattern, err := regexp.Compile(multilinePatternKey)
+		if err != nil {
+			return nil, errors.Wrapf(err, "awslogs could not parse multiline pattern key %q", multilinePatternKey)
+		}
+		return multilinePattern, nil
+	}
+	return nil, nil
+}
+
+// Maps strftime format strings to regex
+var strftimeToRegex = map[string]string{
+	/*weekdayShort          */ `%a`: `(?:Mon|Tue|Wed|Thu|Fri|Sat|Sun)`,
+	/*weekdayFull           */ `%A`: `(?:Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday)`,
+	/*weekdayZeroIndex      */ `%w`: `[0-6]`,
+	/*dayZeroPadded         */ `%d`: `(?:0[1-9]|[1,2][0-9]|3[0,1])`,
+	/*monthShort            */ `%b`: `(?:Jan|Feb|Mar|Apr|Jun|Jul|Aug|Sep|Oct|Nov|Dec)`,
+	/*monthFull             */ `%B`: `(?:January|February|March|April|June|July|August|September|October|November|December)`,
+	/*monthZeroPadded       */ `%m`: `(?:0[1-9]|1[0-2])`,
+	/*yearCentury           */ `%Y`: `\d{4}`,
+	/*yearZeroPadded        */ `%y`: `\d{2}`,
+	/*hour24ZeroPadded      */ `%H`: `(?:[0,1][0-9]|2[0-3])`,
+	/*hour12ZeroPadded      */ `%I`: `(?:0[0-9]|1[0-2])`,
+	/*AM or PM              */ `%p`: "[A,P]M",
+	/*minuteZeroPadded      */ `%M`: `[0-5][0-9]`,
+	/*secondZeroPadded      */ `%S`: `[0-5][0-9]`,
+	/*microsecondZeroPadded */ `%f`: `\d{6}`,
+	/*utcOffset             */ `%z`: `[+-]\d{4}`,
+	/*tzName                */ `%Z`: `[A-Z]{1,4}T`,
+	/*dayOfYearZeroPadded   */ `%j`: `(?:0[0-9][1-9]|[1,2][0-9][0-9]|3[0-5][0-9]|36[0-6])`,
+	/*milliseconds          */ `%L`: `\.\d{3}`,
+}
+
 func parseLogGroup(info logger.Info, groupTemplate string) (string, error) {
 	tmpl, err := templates.NewParse("log-group", groupTemplate)
 	if err != nil {
@@ -352,7 +379,9 @@ func (l *logStream) collectBatch() {
 			// If event buffer is older than batch publish frequency flush the event buffer
 			if eventBufferTimestamp > 0 && len(eventBuffer) > 0 {
 				eventBufferAge := t.UnixNano()/int64(time.Millisecond) - eventBufferTimestamp
-				if eventBufferAge > int64(batchPublishFrequency)/int64(time.Millisecond) {
+				eventBufferExpired := eventBufferAge > int64(batchPublishFrequency)/int64(time.Millisecond)
+				eventBufferNegative := eventBufferAge < 0
+				if eventBufferExpired || eventBufferNegative {
 					events = l.processEvent(events, eventBuffer, eventBufferTimestamp)
 				}
 			}
@@ -376,12 +405,12 @@ func (l *logStream) collectBatch() {
 					eventBufferTimestamp = msg.Timestamp.UnixNano() / int64(time.Millisecond)
 					eventBuffer = eventBuffer[:0]
 				}
-				eventBuffer = append(eventBuffer, unprocessedLine...)
-				// If we have exceeded max bytes per event flush the event buffer up to max bytes
-				if len(eventBuffer) > maximumBytesPerEvent {
-					events = l.processEvent(events, eventBuffer[:maximumBytesPerEvent], eventBufferTimestamp)
-					eventBuffer = eventBuffer[maximumBytesPerEvent:]
+				// If we will exceed max bytes per event flush the current event buffer before appending
+				if len(eventBuffer)+len(unprocessedLine) > maximumBytesPerEvent {
+					events = l.processEvent(events, eventBuffer, eventBufferTimestamp)
+					eventBuffer = eventBuffer[:0]
 				}
+				eventBuffer = append(eventBuffer, unprocessedLine...)
 				logger.PutMessage(msg)
 				continue
 			}
@@ -464,29 +493,6 @@ func (l *logStream) publishBatch(events []wrappedEvent) {
 	}
 }
 
-// Maps strftime format strings to regex
-var strftimeToRegex = map[string]string{
-	/*weekdayShort          */ `%a`: `(?:Mon|Tue|Wed|Thu|Fri|Sat|Sun)`,
-	/*weekdayFull           */ `%A`: `(?:Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday)`,
-	/*weekdayZeroIndex      */ `%w`: `[0-6]`,
-	/*dayZeroPadded         */ `%d`: `(?:0[1-9]|[1,2][0-9]|3[0,1])`,
-	/*monthShort            */ `%b`: `(?:Jan|Feb|Mar|Apr|Jun|Jul|Aug|Sep|Oct|Nov|Dec)`,
-	/*monthFull             */ `%B`: `(?:January|February|March|April|June|July|August|September|October|November|December)`,
-	/*monthZeroPadded       */ `%m`: `(?:0[1-9]|1[0-2])`,
-	/*yearCentury           */ `%Y`: `\d{4}`,
-	/*yearZeroPadded        */ `%y`: `\d{2}`,
-	/*hour24ZeroPadded      */ `%H`: `(?:[0,1][0-9]|2[0-3])`,
-	/*hour12ZeroPadded      */ `%I`: `(?:0[0-9]|1[0-2])`,
-	/*AM or PM              */ `%p`: "[A,P]M",
-	/*minuteZeroPadded      */ `%M`: `[0-5][0-9]`,
-	/*secondZeroPadded      */ `%S`: `[0-5][0-9]`,
-	/*microsecondZeroPadded */ `%f`: `\d{6}`,
-	/*utcOffset             */ `%z`: `[+-]\d{4}`,
-	/*tzName                */ `%Z`: `[A-Z]{1,4}T`,
-	/*dayOfYearZeroPadded   */ `%j`: `(?:0[0-9][1-9]|[1,2][0-9][0-9]|3[0-5][0-9]|36[0-6])`,
-	/*milliseconds          */ `%L`: `\.\d{3}`,
-}
-
 // putLogEvents wraps the PutLogEvents API
 func (l *logStream) putLogEvents(events []*cloudwatchlogs.InputLogEvent, sequenceToken *string) (*string, error) {
 	input := &cloudwatchlogs.PutLogEventsInput{
@@ -512,7 +518,8 @@ func (l *logStream) putLogEvents(events []*cloudwatchlogs.InputLogEvent, sequenc
 }
 
 // ValidateLogOpt looks for awslogs-specific log options awslogs-region,
-// awslogs-group, awslogs-stream, awslogs-create-group
+// awslogs-group, awslogs-stream, awslogs-create-group, awslogs-datetime-format,
+// awslogs-multiline-pattern
 func ValidateLogOpt(cfg map[string]string) error {
 	for key := range cfg {
 		switch key {
@@ -535,6 +542,11 @@ func ValidateLogOpt(cfg map[string]string) error {
 			return fmt.Errorf("must specify valid value for log opt '%s': %v", logCreateGroupKey, err)
 		}
 	}
+	_, datetimeFormatKeyExists := cfg[datetimeFormatKey]
+	_, multilinePatternKeyExists := cfg[multilinePatternKey]
+	if datetimeFormatKeyExists && multilinePatternKeyExists {
+		return fmt.Errorf("you cannot configure log opt '%s' and '%s' at the same time", datetimeFormatKey, multilinePatternKey)
+	}
 	return nil
 }
 

daemon/logger/awslogs/cloudwatchlogs_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/docker/docker/daemon/logger"
 	"github.com/docker/docker/daemon/logger/loggerutils"
 	"github.com/docker/docker/dockerversion"
+	"github.com/stretchr/testify/assert"
 )
 
 const (
@@ -527,33 +528,21 @@ func TestCollectBatchMultilinePattern(t *testing.T) {
 		Timestamp: time.Now(),
 	})
 
-	ticks <- time.Time{}
+	ticks <- time.Now()
 
 	// Verify single multiline event
 	argument := <-mockClient.putLogEventsArgument
-	if argument == nil {
-		t.Fatal("Expected non-nil PutLogEventsInput")
-	}
-	if len(argument.LogEvents) != 1 {
-		t.Errorf("Expected LogEvents to contain 1 elements, but contains %d", len(argument.LogEvents))
-	}
-	if *argument.LogEvents[0].Message != logline+logline {
-		t.Errorf("Expected message to be %s but was %s", logline+logline, *argument.LogEvents[0].Message)
-	}
+	assert.NotNil(t, argument, "Expected non-nil PutLogEventsInput")
+	assert.Equal(t, 1, len(argument.LogEvents), "Expected LogEvents to contain 1 elements, but contains %d", len(argument.LogEvents))
+	assert.Equal(t, logline+logline, *argument.LogEvents[0].Message, "Expected message to be %s but was %s", logline+logline, *argument.LogEvents[0].Message)
 
 	stream.Close()
 
 	// Verify single event
 	argument = <-mockClient.putLogEventsArgument
-	if argument == nil {
-		t.Fatal("Expected non-nil PutLogEventsInput")
-	}
-	if len(argument.LogEvents) != 1 {
-		t.Errorf("Expected LogEvents to contain 1 elements, but contains %d", len(argument.LogEvents))
-	}
-	if *argument.LogEvents[0].Message != "xxxx "+logline {
-		t.Errorf("Expected message to be %s but was %s", "xxxx "+logline, *argument.LogEvents[0].Message)
-	}
+	assert.NotNil(t, argument, "Expected non-nil PutLogEventsInput")
+	assert.Equal(t, 1, len(argument.LogEvents), "Expected LogEvents to contain 1 elements, but contains %d", len(argument.LogEvents))
+	assert.Equal(t, "xxxx "+logline, *argument.LogEvents[0].Message, "Expected message to be %s but was %s", "xxxx "+logline, *argument.LogEvents[0].Message)
 }
 
 func BenchmarkCollectBatch(b *testing.B) {
@@ -656,16 +645,58 @@ func TestCollectBatchMultilinePatternMaxEventAge(t *testing.T) {
 
 	// Verify single multiline event is flushed after maximum event buffer age (batchPublishFrequency)
 	argument := <-mockClient.putLogEventsArgument
-	if argument == nil {
-		t.Fatal("Expected non-nil PutLogEventsInput")
+	assert.NotNil(t, argument, "Expected non-nil PutLogEventsInput")
+	assert.Equal(t, 1, len(argument.LogEvents), "Expected LogEvents to contain 1 elements, but contains %d", len(argument.LogEvents))
+	assert.Equal(t, logline+logline, *argument.LogEvents[0].Message, "Expected message to be %s but was %s", logline+logline, *argument.LogEvents[0].Message)
+
+	stream.Close()
+}
+
+func TestCollectBatchMultilinePatternNegativeEventAge(t *testing.T) {
+	mockClient := newMockClient()
+	multilinePattern := regexp.MustCompile("xxxx")
+	stream := &logStream{
+		client:           mockClient,
+		logGroupName:     groupName,
+		logStreamName:    streamName,
+		multilinePattern: multilinePattern,
+		sequenceToken:    aws.String(sequenceToken),
+		messages:         make(chan *logger.Message),
 	}
-	if len(argument.LogEvents) != 1 {
-		t.Errorf("Expected LogEvents to contain 1 elements, but contains %d", len(argument.LogEvents))
+	mockClient.putLogEventsResult <- &putLogEventsResult{
+		successResult: &cloudwatchlogs.PutLogEventsOutput{
+			NextSequenceToken: aws.String(nextSequenceToken),
+		},
 	}
-	if *argument.LogEvents[0].Message != logline+logline {
-		t.Errorf("Expected message to be %s but was %s", logline+logline, *argument.LogEvents[0].Message)
+	ticks := make(chan time.Time)
+	newTicker = func(_ time.Duration) *time.Ticker {
+		return &time.Ticker{
+			C: ticks,
+		}
 	}
 
+	go stream.collectBatch()
+
+	stream.Log(&logger.Message{
+		Line:      []byte(logline),
+		Timestamp: time.Now(),
+	})
+
+	// Log an event 1 second later
+	stream.Log(&logger.Message{
+		Line:      []byte(logline),
+		Timestamp: time.Now().Add(time.Second),
+	})
+
+	// Fire ticker in past to simulate negative event buffer age
+	ticks <- time.Now().Add(-time.Second)
+
+	// Verify single multiline event is flushed with a negative event buffer age
+	argument := <-mockClient.putLogEventsArgument
+	assert.NotNil(t, argument, "Expected non-nil PutLogEventsInput")
+	assert.Equal(t, 1, len(argument.LogEvents), "Expected LogEvents to contain 1 elements, but contains %d", len(argument.LogEvents))
+	assert.Equal(t, logline+logline, *argument.LogEvents[0].Message, "Expected message to be %s but was %s", logline+logline, *argument.LogEvents[0].Message)
+
 	stream.Close()
 }
 
@@ -930,32 +961,8 @@ func TestParseLogOptionsMultilinePattern(t *testing.T) {
 	}
 
 	multilinePattern, err := parseMultilineOptions(info)
-	if err != nil {
-		t.Errorf("Received unexpected err: %v\n", err)
-	}
-	if !multilinePattern.MatchString("xxxx") {
-		t.Errorf("Expected multilinePattern to match string xxxx but no match found")
-	}
-}
-
-func TestParseLogOptionsDatetimeFormatSupersedesMultilinePattern(t *testing.T) {
-	info := logger.Info{
-		Config: map[string]string{
-			multilinePatternKey: "^xxxx",
-			datetimeFormatKey:   "%Y-%m-%d",
-		},
-	}
-
-	multilinePattern, err := parseMultilineOptions(info)
-	if err != nil {
-		t.Errorf("Received unexpected err: %v\n", err)
-	}
-	if multilinePattern.MatchString("xxxx") {
-		t.Errorf("Expected multilinePattern to NOT match string xxxx but match was made")
-	}
-	if !multilinePattern.MatchString("2017-01-01") {
-		t.Errorf("Expected multilinePattern to match string 2017-01-01 but no match found")
-	}
+	assert.Nil(t, err, "Received unexpected err: %v\n", err)
+	assert.True(t, multilinePattern.MatchString("xxxx"), "Expected multilinePattern to match string xxxx but no match found")
 }
 
 func TestParseLogOptionsDatetimeFormat(t *testing.T) {
@@ -972,21 +979,32 @@ func TestParseLogOptionsDatetimeFormat(t *testing.T) {
 		{"Day of the week: %w, Day of the year: %j", "Day of the week: 4, Day of the year: 091"},
 	}
 	for _, dt := range datetimeFormatTests {
-		info := logger.Info{
-			Config: map[string]string{
-				datetimeFormatKey: dt.format,
-			},
-		}
-		multilinePattern, err := parseMultilineOptions(info)
-		if err != nil {
-			t.Errorf("Received unexpected err: %v\n", err)
-		}
-		if !multilinePattern.MatchString(dt.match) {
-			t.Errorf("Expected multilinePattern %s to match string %s but no match found", dt.format, dt.match)
-		}
+		t.Run(dt.match, func(t *testing.T) {
+			info := logger.Info{
+				Config: map[string]string{
+					datetimeFormatKey: dt.format,
+				},
+			}
+			multilinePattern, err := parseMultilineOptions(info)
+			assert.Nil(t, err, "Received unexpected err: %v\n", err)
+			assert.True(t, multilinePattern.MatchString(dt.match), "Expected multilinePattern %s to match string %s but no match found", dt.format, dt.match)
+		})
 	}
 }
 
+func TestValidateLogOptionsDatetimeFormatAndMultilinePattern(t *testing.T) {
+	cfg := map[string]string{
+		multilinePatternKey: "^xxxx",
+		datetimeFormatKey:   "%Y-%m-%d",
+		logGroupKey:         groupName,
+	}
+	conflictingLogOptionsError := "you cannot configure log opt 'awslogs-datetime-format' and 'awslogs-multiline-pattern' at the same time"
+
+	err := ValidateLogOpt(cfg)
+	assert.NotNil(t, err, "Expected an error but received nil")
+	assert.Equal(t, err.Error(), conflictingLogOptionsError, "Received incorrect error: %v\n", err)
+}
+
 func TestCreateTagSuccess(t *testing.T) {
 	mockClient := newMockClient()
 	info := logger.Info{