[8.0](backport #29695) Support running multiple log groups in cloudwatch input

CHANGELOG.next.asciidoc

@@ -108,6 +108,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Undo deletion of endpoint config from cloudtrail fileset in {pull}29415[29415]. {pull}29450[29450]
 - Make Cisco ASA and FTD modules conform to the ECS definition for event.outcome and event.type. {issue}29581[29581] {pull}29698[29698]
 - ibmmq: Fixed `@timestamp` not being populated with correct values. {pull}29773[29773]
+- Fix using log_group_name_prefix in aws-cloudwatch input. {pull}29695[29695]
 - aws-s3: Improve gzip detection to avoid false negatives. {issue}29968[29968]
 
 *Heartbeat*

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

@@ -50,12 +50,18 @@ log_group_name is given.
 ==== `log_group_name_prefix`
 The prefix for a group of log group names. 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.
+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]
 ==== `region_name`
 Region that the specified log group or log group prefix belongs to.
 
+[float]
+==== `number_of_workers`
+Number of workers that will process the log groups with the given `log_group_name_prefix`.
+Default value is 1.
+
 [float]
 ==== `log_streams`
 A list of strings of log streams names that Filebeat collect log events from.

x-pack/filebeat/input/awscloudwatch/_meta/terraform/.gitignore

@@ -0,0 +1,3 @@
+terraform/
+outputs.yml
+*.tfstate*

x-pack/filebeat/input/awscloudwatch/_meta/terraform/README.md

@@ -0,0 +1,46 @@
+# Terraform setup for AWS CloudWatch Input Integration Tests
+
+This directory contains a Terraform module that creates the AWS resources needed
+for executing the integration tests for the `aws-cloudwatch` Filebeat input. It
+creates two CloudWatch log groups, and one log stream under each log group.
+
+It outputs configuration information that is consumed by the tests to
+`outputs.yml`. The AWS resources are randomly named to prevent name collisions
+between multiple users.
+
+### Usage
+
+You must have the appropriate AWS environment variables for authentication set
+before running Terraform or the integration tests. The AWS key must be
+authorized to create and destroy AWS CloudWatch log groups.
+
+1. Initialize a working directory containing Terraform configuration files.
+
+ `terraform init`
+
+2. Execute terraform in this directory to create the resources. This will also
+ write the `outputs.yml`. You can use `export TF_VAR_aws_region=NNNNN` in order
+ to match the AWS region of the profile you are using.
+
+ `terraform apply`
+
+
+2. (Optional) View the output configuration.
+
+ ```yaml
+ "aws_region": "us-east-1"
+ "log_group_name_1": "filebeat-cloudwatch-integtest-1-417koa"
+ "log_group_name_2": "filebeat-cloudwatch-integtest-2-417koa"
+ ```
+
+3. Execute the integration test.
+
+ ```
+ cd x-pack/filebeat/input/awss3
+ go test -tags aws,integration -run TestInputRun.+ -v .
+ ```
+
+4. Cleanup AWS resources. Execute terraform to delete the log groups created for
+testing.
+
+ `terraform destroy`

x-pack/filebeat/input/awscloudwatch/_meta/terraform/main.tf

@@ -0,0 +1,44 @@
+terraform {
+ required_providers {
+ aws = {
+ source = "hashicorp/aws"
+ version = "~> 3.52"
+ }
+ }
+}
+
+provider "aws" {
+ region = var.aws_region
+}
+
+resource "random_string" "random" {
+ length = 6
+ special = false
+ upper = false
+}
+
+resource "aws_cloudwatch_log_group" "filebeat-integtest-1" {
+ name = "filebeat-log-group-integtest-1-${random_string.random.result}"
+
+ tags = {
+ Environment = "test"
+ }
+}
+
+resource "aws_cloudwatch_log_group" "filebeat-integtest-2" {
+ name = "filebeat-log-group-integtest-2-${random_string.random.result}"
+
+ tags = {
+ Environment = "test"
+ }
+}
+
+resource "aws_cloudwatch_log_stream" "filebeat-integtest-1" {
+ name = "filebeat-log-stream-integtest-1-${random_string.random.result}"
+ log_group_name = aws_cloudwatch_log_group.filebeat-integtest-1.name
+}
+
+resource "aws_cloudwatch_log_stream" "filebeat-integtest-2" {
+ name = "filebeat-log-stream-integtest-2-${random_string.random.result}"
+ log_group_name = aws_cloudwatch_log_group.filebeat-integtest-2.name
+}

x-pack/filebeat/input/awscloudwatch/_meta/terraform/outputs.tf

@@ -0,0 +1,11 @@
+resource "local_file" "secrets" {
+ content = yamlencode({
+ "log_group_name_1" : aws_cloudwatch_log_group.filebeat-integtest-1.name
+ "log_group_name_2" : aws_cloudwatch_log_group.filebeat-integtest-2.name
+ "log_stream_name_1" : aws_cloudwatch_log_stream.filebeat-integtest-1.name
+ "log_stream_name_2" : aws_cloudwatch_log_stream.filebeat-integtest-2.name
+ "aws_region" : var.aws_region
+ })
+ filename = "${path.module}/outputs.yml"
+ file_permission = "0644"
+}

x-pack/filebeat/input/awscloudwatch/_meta/terraform/variables.tf

@@ -0,0 +1,5 @@
+variable "aws_region" {
+ description = "AWS Region"
+ type = string
+ default = "us-east-1"
+}

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

@@ -0,0 +1,126 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package awscloudwatch
+
+import (
+ "context"
+ "sync"
+ "time"
+
+ awssdk "github.com/aws/aws-sdk-go-v2/aws"
+ "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
+ "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/cloudwatchlogsiface"
+ "github.com/pkg/errors"
+
+ "github.com/elastic/beats/v7/libbeat/logp"
+ "github.com/elastic/beats/v7/libbeat/monitoring"
+ "github.com/elastic/beats/v7/libbeat/statestore"
+ awscommon "github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
+)
+
+type cloudwatchPoller struct {
+ numberOfWorkers int
+ apiSleep time.Duration
+ region string
+ logStreams []string
+ logStreamPrefix string
+ startTime int64
+ endTime int64
+ prevEndTime int64
+ workerSem *awscommon.Sem
+ log *logp.Logger
+ metrics *inputMetrics
+ store *statestore.Store
+ workersListingMap *sync.Map
+ workersProcessingMap *sync.Map
+}
+
+func newCloudwatchPoller(log *logp.Logger, metrics *inputMetrics,
+ store *statestore.Store,
+ awsRegion string, apiSleep time.Duration,
+ numberOfWorkers int, logStreams []string, logStreamPrefix string) *cloudwatchPoller {
+ if metrics == nil {
+ metrics = newInputMetrics(monitoring.NewRegistry(), "")
+ }
+
+ return &cloudwatchPoller{
+ numberOfWorkers: numberOfWorkers,
+ apiSleep: apiSleep,
+ region: awsRegion,
+ logStreams: logStreams,
+ logStreamPrefix: logStreamPrefix,
+ startTime: int64(0),
+ endTime: int64(0),
+ workerSem: awscommon.NewSem(numberOfWorkers),
+ log: log,
+ metrics: metrics,
+ store: store,
+ workersListingMap: new(sync.Map),
+ workersProcessingMap: new(sync.Map),
+ }
+}
+
+func (p *cloudwatchPoller) run(svc cloudwatchlogsiface.ClientAPI, logGroup string, startTime int64, endTime int64, logProcessor *logProcessor) {
+ err := p.getLogEventsFromCloudWatch(svc, logGroup, startTime, endTime, logProcessor)
+ if err != nil {
+ var err *awssdk.RequestCanceledError
+ if errors.As(err, &err) {
+ p.log.Error("getLogEventsFromCloudWatch failed with RequestCanceledError: ", err)
+ }
+ p.log.Error("getLogEventsFromCloudWatch failed: ", err)
+ }
+}
+
+// getLogEventsFromCloudWatch uses FilterLogEvents API to collect logs from CloudWatch
+func (p *cloudwatchPoller) getLogEventsFromCloudWatch(svc cloudwatchlogsiface.ClientAPI, logGroup string, startTime int64, endTime int64, logProcessor *logProcessor) error {
+ // construct FilterLogEventsInput
+ filterLogEventsInput := p.constructFilterLogEventsInput(startTime, endTime, logGroup)
+
+ // make API request
+ req := svc.FilterLogEventsRequest(filterLogEventsInput)
+ paginator := cloudwatchlogs.NewFilterLogEventsPaginator(req)
+ for paginator.Next(context.TODO()) {
+ page := paginator.CurrentPage()
+ p.metrics.apiCallsTotal.Inc()
+
+ logEvents := page.Events
+ p.metrics.logEventsReceivedTotal.Add(uint64(len(logEvents)))
+
+ // This sleep is to avoid hitting the FilterLogEvents API limit(5 transactions per second (TPS)/account/Region).
+ p.log.Debugf("sleeping for %v before making FilterLogEvents API call again", p.apiSleep)
+ time.Sleep(p.apiSleep)
+ p.log.Debug("done sleeping")
+
+ p.log.Debugf("Processing #%v events", len(logEvents))
+ err := logProcessor.processLogEvents(logEvents, logGroup, p.region)
+ if err != nil {
+ err = errors.Wrap(err, "processLogEvents failed")
+ p.log.Error(err)
+ }
+ }
+
+ if err := paginator.Err(); err != nil {
+ return errors.Wrap(err, "error FilterLogEvents with Paginator")
+ }
+ return nil
+}
+
+func (p *cloudwatchPoller) constructFilterLogEventsInput(startTime int64, endTime int64, logGroup string) *cloudwatchlogs.FilterLogEventsInput {
+ filterLogEventsInput := &cloudwatchlogs.FilterLogEventsInput{
+ LogGroupName: awssdk.String(logGroup),
+ StartTime: awssdk.Int64(startTime),
+ EndTime: awssdk.Int64(endTime),
+ Limit: awssdk.Int64(100),
+ }
+
+ if len(p.logStreams) > 0 {
+ filterLogEventsInput.LogStreamNames = p.logStreams
+ }
+
+ if p.logStreamPrefix != "" {
+ filterLogEventsInput.LogStreamNamePrefix = awssdk.String(p.logStreamPrefix)
+ }
+ return filterLogEventsInput
+}

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

@@ -25,18 +25,20 @@ type config struct {
  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"`
- AwsConfig awscommon.ConfigAWS `config:",inline"`
+ NumberOfWorkers int `config:"number_of_workers"`
+ AWSConfig awscommon.ConfigAWS `config:",inline"`
 }
 
 func defaultConfig() config {
  return config{
  ForwarderConfig: harvester.ForwarderConfig{
  Type: "aws-cloudwatch",
  },
- StartPosition: "beginning",
- ScanFrequency: 10 * time.Second,
- APITimeout: 120 * time.Second,
- APISleep: 200 * time.Millisecond, // FilterLogEvents has a limit of 5 transactions per second (TPS)/account/Region: 1s / 5 = 200 ms
+ StartPosition: "beginning",
+ ScanFrequency: 10 * time.Second,
+ APITimeout: 120 * time.Second,
+ APISleep: 200 * time.Millisecond, // FilterLogEvents has a limit of 5 transactions per second (TPS)/account/Region: 1s / 5 = 200 ms
+ NumberOfWorkers: 1,
  }
 }