[8.x](backport #41206) [Filebeat] [AWS] add support to source logs from AWS linked source accounts when using log_group_name_prefix

CHANGELOG.next.asciidoc

@@ -330,6 +330,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Add support to source AWS cloudwatch logs from linked accounts. {pull}41188[41188]
 - Jounrald input now supports filtering by facilities {pull}41061[41061]
 - System module now supports reading from jounrald. {pull}41061[41061]
+- Add support to include AWS cloudwatch linked accounts when using log_group_name_prefix to define log group names. {pull}41206[41206]
 
 *Auditbeat*
 

x-pack/filebeat/_meta/config/filebeat.inputs.reference.xpack.yml.tmpl

@@ -144,10 +144,15 @@
   #log_group_name: test
 
   # The prefix for a group of log group names.
+  # You can include linked source accounts by using the property `include_linked_accounts_for_prefix_mode`.
   # Note: `region_name` is required when `log_group_name_prefix` is given.
   # `log_group_name` and `log_group_name_prefix` cannot be given at the same time.
   #log_group_name_prefix: /aws/
 
+  # State whether to include linked source accounts when obtaining log groups matching the prefix provided through `log_group_name_prefix`
+  # This property works together with `log_group_name_prefix` and default value (if unset) is false
+  #include_linked_accounts_for_prefix_mode: true
+
   # Region that the specified log group or log group prefix belongs to.
   #region_name: us-east-1
 

x-pack/filebeat/docs/inputs/input-aws-cloudwatch.asciidoc

@@ -56,13 +56,21 @@ Note: `region_name` is required when log_group_name is given.
 
 [float]
 ==== `log_group_name_prefix`
-The prefix for a group of log group names.
+The prefix for a group of log group names. See `include_linked_accounts_for_prefix_mode` option for linked source accounts behavior.
 
 Note: `region_name` is required when
 `log_group_name_prefix` is given. `log_group_name` and `log_group_name_prefix`
 cannot be given at the same time. The number of workers that will process the
 log groups under this prefix is set through the `number_of_workers` config.
 
+[float]
+==== `include_linked_accounts_for_prefix_mode`
+Configure whether to include linked source accounts that contains the prefix value defined through `log_group_name_prefix`.
+Accepts a boolean and this is by default disabled.
+
+Note: Utilize `log_group_arn` if you desire to obtain logs from a known log group (including linked source accounts)
+You can read more about AWS account linking and cross account observability from the https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Unified-Cross-Account.html[official documentation].
+
 [float]
 ==== `region_name`
 Region that the specified log group or log group prefix belongs to.

x-pack/filebeat/filebeat.reference.yml

@@ -3078,10 +3078,15 @@ filebeat.inputs:
   #log_group_name: test
 
   # The prefix for a group of log group names.
+  # You can include linked source accounts by using the property `include_linked_accounts_for_prefix_mode`.
   # Note: `region_name` is required when `log_group_name_prefix` is given.
   # `log_group_name` and `log_group_name_prefix` cannot be given at the same time.
   #log_group_name_prefix: /aws/
 
+  # State whether to include linked source accounts when obtaining log groups matching the prefix provided through `log_group_name_prefix`
+  # This property works together with `log_group_name_prefix` and default value (if unset) is false
+  #include_linked_accounts_for_prefix_mode: true
+
   # Region that the specified log group or log group prefix belongs to.
   #region_name: us-east-1
 

x-pack/filebeat/input/awscloudwatch/config.go

@@ -13,20 +13,21 @@ import (
 )
 
 type config struct {
-	harvester.ForwarderConfig `config:",inline"`
-	LogGroupARN               string              `config:"log_group_arn"`
-	LogGroupName              string              `config:"log_group_name"`
-	LogGroupNamePrefix        string              `config:"log_group_name_prefix"`
-	RegionName                string              `config:"region_name"`
-	LogStreams                []*string           `config:"log_streams"`
-	LogStreamPrefix           string              `config:"log_stream_prefix"`
-	StartPosition             string              `config:"start_position" default:"beginning"`
-	ScanFrequency             time.Duration       `config:"scan_frequency" validate:"min=0,nonzero"`
-	APITimeout                time.Duration       `config:"api_timeout" validate:"min=0,nonzero"`
-	APISleep                  time.Duration       `config:"api_sleep" validate:"min=0,nonzero"`
-	Latency                   time.Duration       `config:"latency"`
-	NumberOfWorkers           int                 `config:"number_of_workers"`
-	AWSConfig                 awscommon.ConfigAWS `config:",inline"`
+	harvester.ForwarderConfig          `config:",inline"`
+	LogGroupARN                        string              `config:"log_group_arn"`
+	LogGroupName                       string              `config:"log_group_name"`
+	LogGroupNamePrefix                 string              `config:"log_group_name_prefix"`
+	IncludeLinkedAccountsForPrefixMode bool                `config:"include_linked_accounts_for_prefix_mode"`
+	RegionName                         string              `config:"region_name"`
+	LogStreams                         []*string           `config:"log_streams"`
+	LogStreamPrefix                    string              `config:"log_stream_prefix"`
+	StartPosition                      string              `config:"start_position" default:"beginning"`
+	ScanFrequency                      time.Duration       `config:"scan_frequency" validate:"min=0,nonzero"`
+	APITimeout                         time.Duration       `config:"api_timeout" validate:"min=0,nonzero"`
+	APISleep                           time.Duration       `config:"api_sleep" validate:"min=0,nonzero"`
+	Latency                            time.Duration       `config:"latency"`
+	NumberOfWorkers                    int                 `config:"number_of_workers"`
+	AWSConfig                          awscommon.ConfigAWS `config:",inline"`
 }
 
 func defaultConfig() config {

x-pack/filebeat/input/awscloudwatch/input.go

@@ -105,8 +105,9 @@ func (in *cloudwatchInput) Run(inputContext v2.Context, pipeline beat.Pipeline)
 	})
 
 	if len(logGroupIDs) == 0 {
-		// fallback to LogGroupNamePrefix to derive group IDs
-		logGroupIDs, err = getLogGroupNames(svc, in.config.LogGroupNamePrefix)
+		// We haven't extracted group identifiers directly from the input configurations,
+		// now fallback to provided LogGroupNamePrefix and use derived service client to derive logGroupIDs
+		logGroupIDs, err = getLogGroupNames(svc, in.config.LogGroupNamePrefix, in.config.IncludeLinkedAccountsForPrefixMode)
 		if err != nil {
 			return fmt.Errorf("failed to get log group names from LogGroupNamePrefix: %w", err)
 		}
@@ -164,15 +165,16 @@ func fromConfig(cfg config, awsCfg awssdk.Config) (logGroupIDs []string, region
 	return logGroupIDs, region, nil
 }
 
-// getLogGroupNames uses DescribeLogGroups API to retrieve all log group names
-func getLogGroupNames(svc *cloudwatchlogs.Client, logGroupNamePrefix string) ([]string, error) {
+// getLogGroupNames uses DescribeLogGroups API to retrieve LogGroupArn entries that matches the provided logGroupNamePrefix
+func getLogGroupNames(svc *cloudwatchlogs.Client, logGroupNamePrefix string, withLinkedAccount bool) ([]string, error) {
 	// construct DescribeLogGroupsInput
 	describeLogGroupsInput := &cloudwatchlogs.DescribeLogGroupsInput{
-		LogGroupNamePrefix: awssdk.String(logGroupNamePrefix),
+		LogGroupNamePrefix:    awssdk.String(logGroupNamePrefix),
+		IncludeLinkedAccounts: awssdk.Bool(withLinkedAccount),
 	}
 
 	// make API request
-	var logGroupNames []string
+	var logGroupIDs []string
 	paginator := cloudwatchlogs.NewDescribeLogGroupsPaginator(svc, describeLogGroupsInput)
 	for paginator.HasMorePages() {
 		page, err := paginator.NextPage(context.TODO())
@@ -181,8 +183,8 @@ func getLogGroupNames(svc *cloudwatchlogs.Client, logGroupNamePrefix string) ([]
 		}
 
 		for _, lg := range page.LogGroups {
-			logGroupNames = append(logGroupNames, *lg.LogGroupName)
+			logGroupIDs = append(logGroupIDs, *lg.LogGroupArn)
 		}
 	}
-	return logGroupNames, nil
+	return logGroupIDs, nil
 }