feat(aws): Add scheduled invocation for retry event

aws/logs_monitoring/README.md

@@ -92,10 +92,14 @@ If you can't install the Forwarder using the provided CloudFormation template, y
 5. Some AWS accounts are configured such that triggers will not automatically create resource-based policies allowing Cloudwatch log groups to invoke the forwarder. Reference the [CloudWatchLogPermissions][103] to see which permissions are required for the forwarder to be invoked by Cloudwatch Log Events.
 6. [Configure triggers][104].
 7. Create an S3 bucket, and set environment variable `DD_S3_BUCKET_NAME` to the bucket name. Also provide `s3:GetObject`, `s3:PutObject`, `s3:ListBucket`, and `s3:DeleteObject` permissions on this bucket to the Lambda execution role. This bucket is used to store the different tags cache i.e. Lambda, S3, Step Function and Log Group. Additionally, this bucket will be used to store unforwarded events incase of forwarding exceptions.
-8. Set environment variable `DD_STORE_FAILED_EVENTS` to `true` to enable the forwarder to also store event data in the S3 bucket. In case of exceptions when sending logs, metrics or traces to intake, the forwarder will store relevant data in the S3 bucket. On custom invocations i.e. on receiving an event with the `retry` keyword set to a non empty string (which can be manually triggered - see below), the forwarder will retry sending the stored events. When successful it will clear up the storage in the bucket.
+8. Set environment variable `DD_STORE_FAILED_EVENTS` to `true` to enable the forwarder to also store event data in the S3 bucket. In case of exceptions when sending logs, metrics or traces to intake, the forwarder will store relevant data in the S3 bucket. On custom invocations i.e. on receiving an event with the `retry` keyword explicitly set to `true`, the forwarder will retry sending the stored events. Stored logs will be cleanup upon a successful forwarding.
 
 ```bash
-aws lambda invoke --function-name <function-name> --payload '{"retry":"true"}' --cli-binary-format raw-in-base64-out --log-type Tail /dev/stdout
+aws lambda invoke --function-name <function-name> \
+    --payload '{"retry":true}' \
+    --cli-binary-format raw-in-base64-out \
+    --log-type Tail /dev/stdout |
+        jq -r 'select(.LogResult) | .LogResult' | base64 -d | xargs -0 printf "%s"
 ```
 
 <div class="alert alert-warning">

aws/logs_monitoring/lambda_function.py

@@ -62,6 +62,17 @@ def datadog_forwarder(event, context):
     init_cache_layer(function_prefix)
     init_forwarder(function_prefix)
 
+    if len(event) == 1 and str(event.get(DD_RETRY_KEYWORD, "false")).lower() == "true":
+        logger.info("Retry-only invocation")
+
+        try:
+            forwarder.retry()
+        except Exception as e:
+            if logger.isEnabledFor(logging.DEBUG):
+                logger.debug(f"Failed to retry forwarding {e}")
+
+        return
+
     parsed = parse(event, context, cache_layer)
     enriched = enrich(parsed, cache_layer)
     transformed = transform(enriched)
@@ -71,12 +82,11 @@ def datadog_forwarder(event, context):
     parse_and_submit_enhanced_metrics(logs, cache_layer)
 
     try:
-        if bool(event.get(DD_RETRY_KEYWORD, False)) is True:
+        if str(event.get(DD_RETRY_KEYWORD, "false")).lower() == "true":
             forwarder.retry()
     except Exception as e:
         if logger.isEnabledFor(logging.DEBUG):
             logger.debug(f"Failed to retry forwarding {e}")
-        pass
 
 
 def init_cache_layer(function_prefix):

aws/logs_monitoring/template.yaml

@@ -261,6 +261,17 @@ Parameters:
       - true
       - false
     Description: Set to true to enable the forwarder to store events that failed to send to Datadog.
+  DdScheduleRetryFailedEvents:
+    Type: String
+    Default: false
+    AllowedValues:
+      - true
+      - false
+    Description: Set to true to enable a scheduled forwarder invocation (via AWS EventBridge) to process stored failed events.
+  DdScheduleRetryInterval:
+    Type: Number
+    Default: 6
+    Description: Interval in hours for scheduled forwarder invocation (via AWS EventBridge).
   DdForwarderExistingBucketName:
     Type: String
     Default: ""
@@ -292,7 +303,7 @@ Parameters:
   KmsKeyList:
     Type: CommaDelimitedList
     Default: ""
-    Description: List of KMS Key ARNs the Lambda forwarder function can use to decrypt, seperated by comma  
+    Description: List of KMS Key ARNs the Lambda forwarder function can use to decrypt, seperated by comma
 Conditions:
   IsAWSChina: !Equals [!Ref "AWS::Partition", aws-cn]
   IsGovCloud: !Equals [!Ref "AWS::Partition", aws-us-gov]
@@ -348,7 +359,8 @@ Conditions:
   SetLayerARN: !Not
     - !Equals [!Ref LayerARN, ""]
   SetDdForwardLog: !Equals [!Ref DdForwardLog, false]
-  SetDdStepFunctionsTraceEnabled: !Equals [!Ref DdStepFunctionsTraceEnabled, true]
+  SetDdStepFunctionsTraceEnabled:
+    !Equals [!Ref DdStepFunctionsTraceEnabled, true]
   SetDdUseCompression: !Equals [!Ref DdUseCompression, false]
   SetDdCompressionLevel: !Not
     - !Equals [!Ref DdCompressionLevel, 6]
@@ -384,6 +396,7 @@ Conditions:
     - !Equals [!Ref DdLogLevel, ""]
   SetDdForwarderDecryptKeys: !Not
     - !Equals [!Join ["", !Ref KmsKeyList], ""]
+  CreateRetryScheduler: !Equals [!Ref DdScheduleRetryFailedEvents, true]
 Rules:
   MustSetDdApiKey:
     Assertions:
@@ -431,7 +444,10 @@ Resources:
             - !Ref DdForwarderExistingBucketName
           S3Key: !Sub
             - "aws-dd-forwarder-${DdForwarderVersion}.zip"
-            - {DdForwarderVersion: !FindInMap [Constants, DdForwarder, Version]}
+            - {
+                DdForwarderVersion:
+                  !FindInMap [Constants, DdForwarder, Version],
+              }
         - ZipFile: " "
       MemorySize: !Ref MemorySize
       Runtime: python3.13
@@ -831,7 +847,7 @@ Resources:
         - !Ref SourceZipUrl
         - !Sub
           - "https://github.com/DataDog/datadog-serverless-functions/releases/download/aws-dd-forwarder-${DdForwarderVersion}/aws-dd-forwarder-${DdForwarderVersion}.zip"
-          - {DdForwarderVersion: !FindInMap [Constants, DdForwarder, Version]}
+          - { DdForwarderVersion: !FindInMap [Constants, DdForwarder, Version] }
   # The Forwarder's source code is too big to fit the inline code size limit for CloudFormation. In most of AWS
   # partitions and regions, the Forwarder is able to load its source code from a Lambda layer attached to it.
   # In places where Datadog can't/doesn't yet publish Lambda layers, use another Lambda to copy the source code
@@ -970,6 +986,50 @@ Resources:
                       - - "arn:*:s3:::"
                         - !Select [1, !Split ["s3://", !Ref SourceZipUrl]]
                 - !Ref AWS::NoValue
+  SchedulerRole:
+    Type: AWS::IAM::Role
+    Condition: CreateRetryScheduler
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: "2012-10-17"
+        Statement:
+          - Action:
+              - sts:AssumeRole
+            Effect: Allow
+            Principal:
+              Service: !If
+                - IsAWSChina
+                - "scheduler.amazonaws.com.cn"
+                - "scheduler.amazonaws.com"
+      PermissionsBoundary: !If
+        - SetPermissionsBoundary
+        - !Ref PermissionsBoundaryArn
+        - !Ref AWS::NoValue
+      Policies:
+        - PolicyName: ForwarderZipCopierRolePolicy0
+          PolicyDocument:
+            Version: "2012-10-17"
+            Statement:
+              - Effect: Allow
+                Action:
+                  - lambda:InvokeFunction
+                Resource:
+                  - !GetAtt
+                    - Forwarder
+                    - Arn
+  Scheduler:
+    Type: AWS::Scheduler::Schedule
+    Condition: CreateRetryScheduler
+    Properties:
+      Name: !Sub "${AWS::StackName}-retry"
+      Description: Retry the failed events from the Datadog Lambda Forwarder
+      ScheduleExpression: !Sub "rate(${DdScheduleRetryInterval} hours)"
+      FlexibleTimeWindow:
+        Mode: "OFF"
+      Target:
+        Arn: !GetAtt "Forwarder.Arn"
+        RoleArn: !GetAtt "SchedulerRole.Arn"
+        Input: '{"retry": true}'
 Outputs:
   DatadogForwarderArn:
     Description: Datadog Forwarder Lambda Function ARN