Update cloudwatch logs receiver with latest upstream changes (#3909)

Starting with Alloy 1.9.x the cloudwatch logs receiver starts reading
from the start which causes duplicate logs in case of restarts.

* Add `start_from` arg
* Add storage capabilities

Relates-To: open-telemetry/opentelemetry-collector-contrib#39007

Author: boernd

CHANGELOG.md

@@ -41,6 +41,7 @@ Main (unreleased)
 
 - Add support for `conditions` and statement-specific `error_mode` in `otelcol.processor.transform`. (@ptodev)
 
+- Add `storage` and `start_from` args to cloudwatch logs receiver. (@boernd)
 
 ### Bugfixes
 

docs/sources/reference/components/otelcol/otelcol.receiver.awscloudwatch.md

@@ -39,11 +39,12 @@ otelcol.receiver.awscloudwatch "<LABEL>" {
 
 You can use the following arguments with `otelcol.receiver.awscloudwatch`:
 
-| Name            | Type     | Description                      | Default | Required |
-| --------------- | -------- | -------------------------------- | ------- | -------- |
-| `region`        | `string` | AWS region to collect logs from. |         | yes      |
-| `imds_endpoint` | `string` | Custom EC2 IMDS endpoint to use. |         | no       |
-| `profile`       | `string` | AWS credentials profile to use.  |         | no       |
+| Name            | Type                       | Description                                                               | Default | Required |
+| --------------- | -------------------------- | ------------------------------------------------------------------------- | ------- | -------- |
+| `region`        | `string`                   | AWS region to collect logs from.                                          |         | yes      |
+| `imds_endpoint` | `string`                   | Custom EC2 IMDS endpoint to use.                                          |         | no       |
+| `profile`       | `string`                   | AWS credentials profile to use.                                           |         | no       |
+| `storage`       | `capsule(otelcol.Handler)` | Handler from an `otelcol.storage` component to use for persisting state.  |         | no       |
 
 If `imds_endpoint` isn't specified, and the environment variable `AWS_EC2_METADATA_SERVICE_ENDPOINT` has a value, it will be used as the IMDS endpoint.
 
@@ -81,6 +82,7 @@ The following arguments are supported:
 | ------------------------ | ---------- | -------------------------------------------------------------- | ------- | -------- |
 | `max_events_per_request` | `int`      | Maximum number of events to process per request to CloudWatch. | `1000`  | no       |
 | `poll_interval`          | `duration` | How frequently to poll for new log entries.                    | `"1m"`  | no       |
+| `start_from`             | `string`   | Timestamp in RFC3339 format where to start reading logs.       | `""`    | no       |
 
 The `logs` block supports the following blocks:
 

internal/component/otelcol/receiver/awscloudwatch/awscloudwatch.go

@@ -2,9 +2,12 @@
 package awscloudwatch
 
 import (
+	"fmt"
+
 	"github.com/grafana/alloy/internal/component"
 	"github.com/grafana/alloy/internal/component/otelcol"
 	otelcolCfg "github.com/grafana/alloy/internal/component/otelcol/config"
+	"github.com/grafana/alloy/internal/component/otelcol/extension"
 	"github.com/grafana/alloy/internal/component/otelcol/receiver"
 	"github.com/grafana/alloy/internal/featuregate"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscloudwatchreceiver"
@@ -29,10 +32,11 @@ func init() {
 
 // Arguments configures the otelcol.receiver.awscloudwatch component.
 type Arguments struct {
-	Region       string     `alloy:"region,attr"`
-	Profile      string     `alloy:"profile,attr,optional"`
-	IMDSEndpoint string     `alloy:"imds_endpoint,attr,optional"`
-	Logs         LogsConfig `alloy:"logs,block,optional"`
+	Region       string                      `alloy:"region,attr"`
+	Profile      string                      `alloy:"profile,attr,optional"`
+	IMDSEndpoint string                      `alloy:"imds_endpoint,attr,optional"`
+	Logs         LogsConfig                  `alloy:"logs,block,optional"`
+	Storage      *extension.ExtensionHandler `alloy:"storage,attr,optional"`
 
 	DebugMetrics otelcolCfg.DebugMetricsArguments `alloy:"debug_metrics,block,optional"`
 
@@ -79,12 +83,25 @@ func (args Arguments) Convert() (otelcomponent.Config, error) {
 		otelConfig.Logs.Groups.AutodiscoverConfig.Limit = defaultLogGroupLimit
 	}
 
+	// Configure storage if args.Storage is set.
+	if args.Storage != nil {
+		if args.Storage.Extension == nil {
+			return nil, fmt.Errorf("missing storage extension")
+		}
+
+		otelConfig.StorageID = &args.Storage.ID
+	}
+
 	return otelConfig, nil
 }
 
 // Extensions implements receiver.Arguments.
 func (args Arguments) Extensions() map[otelcomponent.ID]otelcomponent.Component {
-	return nil
+	m := make(map[otelcomponent.ID]otelcomponent.Component)
+	if args.Storage != nil {
+		m[args.Storage.ID] = args.Storage.Extension
+	}
+	return m
 }
 
 // Exporters implements receiver.Arguments.

internal/component/otelcol/receiver/awscloudwatch/awscloudwatch_test.go

@@ -25,6 +25,7 @@ func TestArguments_UnmarshalAlloy(t *testing.T) {
 			expected: awscloudwatchreceiver.Config{
 				Region: "us-west-2",
 				Logs: &awscloudwatchreceiver.LogsConfig{
+					StartFrom:           "",
 					PollInterval:        time.Minute,
 					MaxEventsPerRequest: 1000,
 					Groups: awscloudwatchreceiver.GroupConfig{
@@ -156,6 +157,32 @@ func TestArguments_UnmarshalAlloy(t *testing.T) {
 				},
 			},
 		},
+		{
+			testName: "start_from configuration set",
+			cfg: `
+				region = "us-west-2"
+				logs {
+					poll_interval = "1m"
+					start_from = "2025-06-25T00:00:00Z"
+				}
+				output {}
+			`,
+			expected: awscloudwatchreceiver.Config{
+				Region: "us-west-2",
+				Logs: &awscloudwatchreceiver.LogsConfig{
+					StartFrom:           "2025-06-25T00:00:00Z",
+					PollInterval:        time.Minute,
+					MaxEventsPerRequest: 1000,
+					Groups: awscloudwatchreceiver.GroupConfig{
+						AutodiscoverConfig: &awscloudwatchreceiver.AutodiscoverConfig{
+							Limit:   50,
+							Streams: awscloudwatchreceiver.StreamConfig{},
+						},
+						NamedConfigs: map[string]awscloudwatchreceiver.StreamConfig{},
+					},
+				},
+			},
+		},
 	}
 
 	for _, tc := range tests {
@@ -233,6 +260,18 @@ func TestArguments_Validate(t *testing.T) {
 			`,
 			expectedError: "both autodiscover and named configs are configured, Only one or the other is permitted",
 		},
+		{
+			testName: "invalid start_from configuration set",
+			cfg: `
+				region = "us-west-2"
+				logs {
+					poll_interval = "1m"
+					start_from = "earliest"
+				}
+				output {}
+			`,
+			expectedError: "invalid start_from time format",
+		},
 	}
 
 	for _, tc := range tests {

internal/component/otelcol/receiver/awscloudwatch/config_awscloudwatch.go

@@ -13,6 +13,7 @@ type LogsConfig struct {
 	PollInterval        time.Duration `alloy:"poll_interval,attr,optional"`
 	MaxEventsPerRequest int           `alloy:"max_events_per_request,attr,optional"`
 	Groups              GroupConfig   `alloy:"groups,block,optional"`
+	StartFrom           string        `alloy:"start_from,attr,optional"`
 }
 
 func (args *LogsConfig) Convert() *awscloudwatchreceiver.LogsConfig {
@@ -23,6 +24,7 @@ func (args *LogsConfig) Convert() *awscloudwatchreceiver.LogsConfig {
 		PollInterval:        args.PollInterval,
 		MaxEventsPerRequest: args.MaxEventsPerRequest,
 		Groups:              args.Groups.Convert(),
+		StartFrom:           args.StartFrom,
 	}
 }