Merge pull request #42132 from xia-wu/add-create-log-stream

Add an option to skip create log stream for awslogs driver

daemon/logger/awslogs/cloudwatchlogs.go

@@ -35,6 +35,7 @@ const (
 	logGroupKey            = "awslogs-group"
 	logStreamKey           = "awslogs-stream"
 	logCreateGroupKey      = "awslogs-create-group"
+	logCreateStreamKey     = "awslogs-create-stream"
 	tagKey                 = "tag"
 	datetimeFormatKey      = "awslogs-datetime-format"
 	multilinePatternKey    = "awslogs-multiline-pattern"
@@ -71,6 +72,7 @@ type logStream struct {
 	logStreamName      string
 	logGroupName       string
 	logCreateGroup     bool
+	logCreateStream    bool
 	logNonBlocking     bool
 	forceFlushInterval time.Duration
 	multilinePattern   *regexp.Regexp
@@ -85,6 +87,7 @@ type logStreamConfig struct {
 	logStreamName      string
 	logGroupName       string
 	logCreateGroup     bool
+	logCreateStream    bool
 	logNonBlocking     bool
 	forceFlushInterval time.Duration
 	maxBufferedEvents  int
@@ -151,6 +154,7 @@ func New(info logger.Info) (logger.Logger, error) {
 		logStreamName:      containerStreamConfig.logStreamName,
 		logGroupName:       containerStreamConfig.logGroupName,
 		logCreateGroup:     containerStreamConfig.logCreateGroup,
+		logCreateStream:    containerStreamConfig.logCreateStream,
 		logNonBlocking:     containerStreamConfig.logNonBlocking,
 		forceFlushInterval: containerStreamConfig.forceFlushInterval,
 		multilinePattern:   containerStreamConfig.multilinePattern,
@@ -237,6 +241,13 @@ func newStreamConfig(info logger.Info) (*logStreamConfig, error) {
 	if info.Config[logStreamKey] != "" {
 		logStreamName = info.Config[logStreamKey]
 	}
+	logCreateStream := true
+	if info.Config[logCreateStreamKey] != "" {
+		logCreateStream, err = strconv.ParseBool(info.Config[logCreateStreamKey])
+		if err != nil {
+			return nil, err
+		}
+	}
 
 	multilinePattern, err := parseMultilineOptions(info)
 	if err != nil {
@@ -247,6 +258,7 @@ func newStreamConfig(info logger.Info) (*logStreamConfig, error) {
 		logStreamName:      logStreamName,
 		logGroupName:       logGroupName,
 		logCreateGroup:     logCreateGroup,
+		logCreateStream:    logCreateStream,
 		logNonBlocking:     logNonBlocking,
 		forceFlushInterval: forceFlushInterval,
 		maxBufferedEvents:  maxBufferedEvents,
@@ -480,6 +492,16 @@ func (l *logStream) createLogGroup() error {
 
 // createLogStream creates a log stream for the instance of the awslogs logging driver
 func (l *logStream) createLogStream() error {
+	// Directly return if we do not want to create log stream.
+	if !l.logCreateStream {
+		logrus.WithFields(logrus.Fields{
+			"logGroupName":    l.logGroupName,
+			"logStreamName":   l.logStreamName,
+			"logCreateStream": l.logCreateStream,
+		}).Info("Skipping creating log stream")
+		return nil
+	}
+
 	input := &cloudwatchlogs.CreateLogStreamInput{
 		LogGroupName:  aws.String(l.logGroupName),
 		LogStreamName: aws.String(l.logStreamName),

daemon/logger/awslogs/cloudwatchlogs_test.go

@@ -65,6 +65,7 @@ func TestNewStreamConfig(t *testing.T) {
 		logStreamName      string
 		logGroupName       string
 		logCreateGroup     string
+		logCreateStream    string
 		logNonBlocking     string
 		forceFlushInterval string
 		maxBufferedEvents  string
@@ -73,13 +74,13 @@ func TestNewStreamConfig(t *testing.T) {
 		shouldErr          bool
 		testName           string
 	}{
-		{"", groupName, "", "", "", "", "", "", false, "defaults"},
+		{"", groupName, "", "", "", "", "", "", "", false, "defaults"},
-		{"", groupName, "invalid create group", "", "", "", "", "", true, "invalid create group"},
+		{"", groupName, "invalid create group", "", "", "", "", "", "", true, "invalid create group"},
-		{"", groupName, "", "", "invalid flush interval", "", "", "", true, "invalid flush interval"},
+		{"", groupName, "", "", "", "invalid flush interval", "", "", "", true, "invalid flush interval"},
-		{"", groupName, "", "", "", "invalid max buffered events", "", "", true, "invalid max buffered events"},
+		{"", groupName, "", "", "", "", "invalid max buffered events", "", "", true, "invalid max buffered events"},
-		{"", groupName, "", "", "", "", "", "n{1001}", true, "invalid multiline pattern"},
+		{"", groupName, "", "", "", "", "", "", "n{1001}", true, "invalid multiline pattern"},
-		{"", groupName, "", "", "15", "", "", "", false, "flush interval at 15"},
+		{"", groupName, "", "", "", "15", "", "", "", false, "flush interval at 15"},
-		{"", groupName, "", "", "", "1024", "", "", false, "max buffered events at 1024"},
+		{"", groupName, "", "", "", "", "1024", "", "", false, "max buffered events at 1024"},
 	}
 
 	for _, tc := range tests {
@@ -91,6 +92,7 @@ func TestNewStreamConfig(t *testing.T) {
 				forceFlushIntervalKey: tc.forceFlushInterval,
 				maxBufferedEventsKey:  tc.maxBufferedEvents,
 				logStreamKey:          tc.logStreamName,
+				logCreateStreamKey:    tc.logCreateStream,
 				datetimeFormatKey:     tc.datetimeFormat,
 				multilinePatternKey:   tc.multilinePattern,
 			}
@@ -189,6 +191,7 @@ func TestCreateSuccess(t *testing.T) {
 		client:          mockClient,
 		logGroupName:    groupName,
 		logStreamName:   streamName,
+		logCreateStream: true,
 	}
 	mockClient.createLogStreamResult <- &createLogStreamResult{}
 
@@ -212,6 +215,20 @@ func TestCreateSuccess(t *testing.T) {
 	}
 }
 
+func TestCreateStreamSkipped(t *testing.T) {
+	stream := &logStream{
+		logGroupName:    groupName,
+		logStreamName:   streamName,
+		logCreateStream: false,
+	}
+
+	err := stream.create()
+
+	if err != nil {
+		t.Errorf("Received unexpected err: %v\n", err)
+	}
+}
+
 func TestCreateLogGroupSuccess(t *testing.T) {
 	mockClient := newMockClient()
 	stream := &logStream{
@@ -219,6 +236,7 @@ func TestCreateLogGroupSuccess(t *testing.T) {
 		logGroupName:    groupName,
 		logStreamName:   streamName,
 		logCreateGroup:  true,
+		logCreateStream: true,
 	}
 	mockClient.createLogGroupResult <- &createLogGroupResult{}
 	mockClient.createLogStreamResult <- &createLogStreamResult{}
@@ -247,6 +265,7 @@ func TestCreateError(t *testing.T) {
 	mockClient := newMockClient()
 	stream := &logStream{
 		client:          mockClient,
+		logCreateStream: true,
 	}
 	mockClient.createLogStreamResult <- &createLogStreamResult{
 		errorResult: errors.New("Error"),
@@ -263,6 +282,7 @@ func TestCreateAlreadyExists(t *testing.T) {
 	mockClient := newMockClient()
 	stream := &logStream{
 		client:          mockClient,
+		logCreateStream: true,
 	}
 	mockClient.createLogStreamResult <- &createLogStreamResult{
 		errorResult: awserr.New(resourceAlreadyExistsCode, "", nil),
@@ -1555,6 +1575,7 @@ func TestCreateTagSuccess(t *testing.T) {
 		client:          mockClient,
 		logGroupName:    groupName,
 		logStreamName:   logStreamName,
+		logCreateStream: true,
 	}
 	mockClient.createLogStreamResult <- &createLogStreamResult{}