Reapply "Add default log publishing concurrency (#1770)" (#1819)

translator/tocwconfig/sampleConfig/compass_linux_config.json

@@ -15,6 +15,7 @@
     "deployment.environment": "agent-level-environment"
   },
   "logs": {
+    "concurrency": 1,
     "metrics_collected": {
       "otlp": {
         "service.name": "otlp-level-service",

translator/tocwconfig/sampleConfig/complete_darwin_config.json

@@ -143,6 +143,7 @@
     "endpoint_override": "https://monitoring-fips.us-west-2.amazonaws.com"
   },
   "logs": {
+    "concurrency": 1,
     "metrics_collected": {
       "emf": {
         "service_address": "udp://127.0.0.1:25888"

translator/tocwconfig/sampleConfig/complete_linux_config.json

@@ -246,6 +246,7 @@
     "endpoint_override": "https://monitoring-fips.us-west-2.amazonaws.com"
   },
   "logs": {
+    "concurrency": 1,
     "metrics_collected": {
       "emf": {
         "service_address": "udp://127.0.0.1:25888"

translator/tocwconfig/sampleConfig/complete_windows_config.json

@@ -150,6 +150,7 @@
     "endpoint_override": "https://monitoring-fips.us-west-2.amazonaws.com"
   },
   "logs": {
+    "concurrency": 1,
     "metrics_collected": {
       "emf": {
         "service_address": "udp://127.0.0.1:25888"

translator/tocwconfig/sampleConfig/config_with_env.json

@@ -3,6 +3,7 @@
     "region": "${ENV_REGION}"
   },
   "logs": {
+    "concurrency": 1,
     "credentials": {
       "role_arn": "${ENV_CREDENTIALS_ROLE_ARN}"
     },

translator/tocwconfig/sampleConfig/log_filter.json

@@ -3,6 +3,7 @@
     "region": "us-east-1"
   },
   "logs": {
+    "concurrency": 10,
     "logs_collected": {
       "files": {
         "collect_list": [
@@ -31,7 +32,6 @@
         ]
       }
     },
-    "concurrency": 10,
     "log_stream_name": "LOG_STREAM_NAME"
   }
 }

translator/tocwconfig/sampleConfig/log_only_config_windows.json

@@ -1,5 +1,6 @@
 {
   "logs": {
+    "concurrency": 10,
     "logs_collected": {
       "files": {
         "collect_list": [
@@ -36,7 +37,6 @@
         ]
       }
     },
-    "concurrency": 10,
     "log_stream_name": "LOG_STREAM_NAME"
   }
 }

translator/tocwconfig/sampleConfig/logs_and_kubernetes_config.json

@@ -3,6 +3,7 @@
     "region": "us-east-1"
   },
   "logs": {
+    "concurrency": 1,
     "metrics_collected": {
       "emf": {
       },

translator/tocwconfig/sampleConfig/no_skip_log_timestamp.json

@@ -1,5 +1,6 @@
 {
   "logs": {
+    "concurrency": 1,
     "logs_collected": {
       "files": {
         "collect_list": [

translator/tocwconfig/sampleConfig/no_skip_log_timestamp_windows.json

@@ -1,6 +1,6 @@
 {
-
   "logs": {
+    "concurrency": 1,
     "logs_collected": {
       "files": {
         "collect_list": [

translator/tocwconfig/sampleConfig/skip_log_timestamp.json

@@ -3,6 +3,7 @@
     "logfile": "/opt/tmp/a.log"
   },
   "logs": {
+    "concurrency": 1,
     "logs_collected": {
       "files": {
         "collect_list": [

translator/tocwconfig/sampleConfig/skip_log_timestamp_default.json

@@ -1,5 +1,6 @@
 {
   "logs": {
+    "concurrency": 1,
     "logs_collected": {
       "files": {
         "collect_list": [

translator/tocwconfig/sampleConfig/skip_log_timestamp_default_windows.json

@@ -1,5 +1,6 @@
 {
   "logs": {
+    "concurrency": 1,
     "logs_collected": {
       "files": {
         "collect_list": [

translator/tocwconfig/sampleConfig/skip_log_timestamp_windows.json

@@ -3,6 +3,7 @@
     "logfile": "c:\\tmp\\am.log"
   },
   "logs": {
+    "concurrency": 1,
     "logs_collected": {
       "files": {
         "collect_list": [

translator/tocwconfig/sampleConfig/windows_eventlog_only_config.json

@@ -1,5 +1,6 @@
 {
   "logs": {
+    "concurrency": 1,
     "logs_collected": {
       "windows_events": {
         "collect_list": [

translator/translate/logs/constants/constants.go

@@ -0,0 +1,24 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package constants
+
+const (
+	SectionKeyLogsCollected = "logs_collected"
+)
+
+// children of logs_collected
+const (
+	SectionKeyFiles = "files"
+)
+
+// child of files
+const (
+	SectionKeyCollectList = "collect_list"
+)
+
+// children of collect_list
+const (
+	SectionKeyFilePath        = "file_path"
+	SectionKeyTimestampFormat = "timestamp_format"
+)

translator/translate/logs/logs_collected/files/collect_list/collect_list.go

@@ -14,19 +14,19 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/translator/jsonconfig/mergeJsonRule"
 	"github.com/aws/amazon-cloudwatch-agent/translator/jsonconfig/mergeJsonUtil"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/agent"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/logs/constants"
 	parent "github.com/aws/amazon-cloudwatch-agent/translator/translate/logs/logs_collected/files"
 	logUtil "github.com/aws/amazon-cloudwatch-agent/translator/translate/logs/util"
 )
 
 type Rule translator.Rule
 
 const (
-	SectionKey              = "collect_list"
 	logConfigOutputFileName = "log-config.json"
 )
 
 func GetCurPath() string {
-	curPath := parent.GetCurPath() + SectionKey + "/"
+	curPath := parent.GetCurPath() + constants.SectionKeyCollectList + "/"
 	return curPath
 }
 
@@ -48,8 +48,8 @@ type FileConfig struct {
 func (f *FileConfig) ApplyRule(input interface{}) (returnKey string, returnVal interface{}) {
 	m := input.(map[string]interface{})
 	res := []interface{}{}
-	if translator.IsValid(input, SectionKey, GetCurPath()) {
-		configArr := m[SectionKey].([]interface{})
+	if translator.IsValid(input, constants.SectionKeyCollectList, GetCurPath()) {
+		configArr := m[constants.SectionKeyCollectList].([]interface{})
 		for i := 0; i < len(configArr); i++ {
 			Index += 1
 			result := map[string]interface{}{}
@@ -122,11 +122,11 @@ func outputLogConfig(logConfigs []interface{}) {
 var MergeRuleMap = map[string]mergeJsonRule.MergeRule{}
 
 func (f *FileConfig) Merge(source map[string]interface{}, result map[string]interface{}) {
-	mergeJsonUtil.MergeList(source, result, SectionKey)
+	mergeJsonUtil.MergeList(source, result, constants.SectionKeyCollectList)
 }
 
 func init() {
 	f := new(FileConfig)
-	parent.RegisterRule(SectionKey, f)
-	parent.MergeRuleMap[SectionKey] = f
+	parent.RegisterRule(constants.SectionKeyCollectList, f)
+	parent.MergeRuleMap[constants.SectionKeyCollectList] = f
 }

translator/translate/logs/logs_collected/files/collect_list/ruleTimestampFormat.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/aws/amazon-cloudwatch-agent/translator"
 	"github.com/aws/amazon-cloudwatch-agent/translator/context"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/logs/constants"
 )
 
 /*
@@ -145,9 +146,9 @@ func (t *TimestampRegex) ApplyRule(input interface{}) (returnKey string, returnV
 	//Convert the input string into []rune and iterate the map and build the output []rune
 	m := input.(map[string]interface{})
 	//If user not specify the timestamp_format, then no config entry for "timestamp_layout" in TOML
-	if val, ok := m["timestamp_format"]; !ok {
+	if val, ok := m[constants.SectionKeyTimestampFormat]; !ok {
 		return "", ""
-	} else if m["file_path"] == context.CurrentContext().GetAgentLogFile() {
+	} else if m[constants.SectionKeyFilePath] == context.CurrentContext().GetAgentLogFile() {
 		fmt.Printf("timestamp_format set file_path : %s is the same as agent log file %s thus do not use timestamp_regex \n", m["file_path"], context.CurrentContext().GetAgentLogFile())
 		return "", ""
 	} else {
@@ -164,7 +165,7 @@ func (t *TimestampRegex) ApplyRule(input interface{}) (returnKey string, returnV
 		res = "(" + res + ")"
 		returnKey = "timestamp_regex"
 		if _, err := regexp.Compile(res); err != nil {
-			translator.AddErrorMessages(GetCurPath()+"timestamp_format", fmt.Sprintf("Timestamp format %s is invalid", val))
+			translator.AddErrorMessages(GetCurPath()+constants.SectionKeyTimestampFormat, fmt.Sprintf("Timestamp format %s is invalid", val))
 			return
 		}
 		returnVal = res
@@ -181,9 +182,9 @@ func (t *TimestampLayout) ApplyRule(input interface{}) (returnKey string, return
 	//Convert the input string into []rune and iterate the map and build the output []rune
 	m := input.(map[string]interface{})
 	//If user not specify the timestamp_format, then no config entry for "timestamp_layout" in TOML
-	if val, ok := m["timestamp_format"]; !ok {
+	if val, ok := m[constants.SectionKeyTimestampFormat]; !ok {
 		return "", ""
-	} else if m["file_path"] == context.CurrentContext().GetAgentLogFile() {
+	} else if m[constants.SectionKeyFilePath] == context.CurrentContext().GetAgentLogFile() {
 		fmt.Printf("timestamp_format set file_path : %s is the same as agent log file %s thus do not use timestamp_layout \n", m["file_path"], context.CurrentContext().GetAgentLogFile())
 		return "", ""
 	} else {
@@ -232,5 +233,5 @@ func init() {
 	t2 := new(TimestampRegex)
 	t3 := new(Timezone)
 	r := []Rule{t1, t2, t3}
-	RegisterRule("timestamp_format", r)
+	RegisterRule(constants.SectionKeyTimestampFormat, r)
 }

translator/translate/logs/logs_collected/files/files.go

@@ -7,6 +7,7 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/translator"
 	"github.com/aws/amazon-cloudwatch-agent/translator/jsonconfig/mergeJsonRule"
 	"github.com/aws/amazon-cloudwatch-agent/translator/jsonconfig/mergeJsonUtil"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/logs/constants"
 	parent "github.com/aws/amazon-cloudwatch-agent/translator/translate/logs/logs_collected"
 )
 
@@ -16,7 +17,7 @@ type Files struct {
 }
 
 const (
-	SectionKey       = "files"
+	SectionKey       = constants.SectionKeyFiles
 	SectionMappedKey = "logfile"
 )
 

translator/translate/logs/logs_collected/logs_collected.go

@@ -11,6 +11,7 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/translator/jsonconfig/mergeJsonRule"
 	"github.com/aws/amazon-cloudwatch-agent/translator/jsonconfig/mergeJsonUtil"
 	parent "github.com/aws/amazon-cloudwatch-agent/translator/translate/logs"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/logs/constants"
 )
 
 type Rule translator.Rule
@@ -25,10 +26,8 @@ var (
 	darwinMetricCollectRule  = map[string]Rule{}
 )
 
-const SectionKey = "logs_collected"
-
 func GetCurPath() string {
-	curPath := parent.GetCurPath() + SectionKey + "/"
+	curPath := parent.GetCurPath() + constants.SectionKeyLogsCollected + "/"
 	return curPath
 }
 
@@ -60,12 +59,12 @@ func (l *LogsCollected) ApplyRule(input interface{}) (returnKey string, returnVa
 		log.Panicf("E! Unknown target platform: %s ", translator.GetTargetPlatform())
 	}
 
-	if _, ok := im[SectionKey]; !ok {
+	if _, ok := im[constants.SectionKeyLogsCollected]; !ok {
 		returnKey = ""
 		returnVal = ""
 	} else {
 		for _, rule := range targetRuleMap {
-			key, val := rule.ApplyRule(im[SectionKey])
+			key, val := rule.ApplyRule(im[constants.SectionKeyLogsCollected])
 			if key == "inputs" {
 				result = translator.MergeTwoUniqueMaps(result, val.(map[string]interface{}))
 			}
@@ -79,11 +78,11 @@ func (l *LogsCollected) ApplyRule(input interface{}) (returnKey string, returnVa
 var MergeRuleMap = map[string]mergeJsonRule.MergeRule{}
 
 func (l *LogsCollected) Merge(source map[string]interface{}, result map[string]interface{}) {
-	mergeJsonUtil.MergeMap(source, result, SectionKey, MergeRuleMap, GetCurPath())
+	mergeJsonUtil.MergeMap(source, result, constants.SectionKeyLogsCollected, MergeRuleMap, GetCurPath())
 }
 
 func init() {
 	obj := new(LogsCollected)
-	parent.RegisterRule(SectionKey, obj)
-	parent.MergeRuleMap[SectionKey] = obj
+	parent.RegisterRule(constants.SectionKeyLogsCollected, obj)
+	parent.MergeRuleMap[constants.SectionKeyLogsCollected] = obj
 }

translator/translate/logs/logs_test.go

@@ -6,10 +6,12 @@ package logs
 import (
 	"encoding/json"
 	"os"
+	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 
+	"github.com/aws/amazon-cloudwatch-agent/internal/util/testutil"
 	"github.com/aws/amazon-cloudwatch-agent/translator/config"
 	"github.com/aws/amazon-cloudwatch-agent/translator/context"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/agent"
@@ -239,3 +241,59 @@ func TestLogs_ServiceAndEnvironmentMissing(t *testing.T) {
 	assert.Equal(t, "my-service", GlobalLogConfig.ServiceName)
 	assert.Equal(t, "ec2:group", GlobalLogConfig.DeploymentEnvironment)
 }
+
+func TestLogs_Concurrency(t *testing.T) {
+	l := new(Logs)
+	agent.Global_Config.Region = "us-east-1"
+	agent.Global_Config.RegionType = "any"
+	context.ResetContext()
+
+	var input interface{}
+	err := json.Unmarshal([]byte(`{"logs":{"concurrency":10}}`), &input)
+	if err != nil {
+		assert.Fail(t, err.Error())
+	}
+	_, _ = l.ApplyRule(input)
+
+	assert.Equal(t, 10, GlobalLogConfig.Concurrency)
+}
+
+func TestLogs_Concurrency_Default(t *testing.T) {
+	l := new(Logs)
+	agent.Global_Config.Region = "us-east-1"
+	agent.Global_Config.RegionType = "any"
+	context.ResetContext()
+	context.CurrentContext().SetAgentLogFile("/tmp/amazon-cloudwatch-agent.log")
+
+	testCases := map[string]struct {
+		input map[string]any
+		want  int
+	}{
+		"WithMissingLogsCollected": {
+			input: map[string]any{
+				"logs": map[string]any{},
+			},
+			want: -1,
+		},
+		"WithLogFileOnly": {
+			input: testutil.GetJson(t, filepath.Join("testdata", "default_concurrency", "logfile_only.json")),
+			want:  defaultConcurrency,
+		},
+		"WithWindowsEventsOnly": {
+			input: testutil.GetJson(t, filepath.Join("testdata", "default_concurrency", "windows_events_only.json")),
+			want:  defaultConcurrency,
+		},
+		"WithMissingTimestampFormat/LogFile": {
+			input: testutil.GetJson(t, filepath.Join("testdata", "default_concurrency", "missing_timestamp.json")),
+			want:  -1,
+		},
+	}
+
+	for name, testCase := range testCases {
+		t.Run(name, func(t *testing.T) {
+			_, _ = l.ApplyRule(testCase.input)
+
+			assert.Equal(t, testCase.want, GlobalLogConfig.Concurrency)
+		})
+	}
+}