Fix lambda-manager subscriptions when LogGroupClass is omitted (#193)

* Fix lambda-manager when LogGroupClass is omitted

* Avoid PyYAML in lambda-manager template test

Author: natnayr

src/lambda-manager/CHANGELOG.md

@@ -3,6 +3,11 @@
 All notable changes to this project will be documented in this file.
 This format is based on Keep a Changelog.
 
+## [2.0.12] - 2026-06-15
+### Fixed
+- Match CloudFormation `CreateLogGroup` events when `LogGroupClass` is omitted and the log group still defaults to `STANDARD`.
+- Continue skipping non-standard `INFREQUENT_ACCESS` and `DELIVERY` log groups during create-event processing.
+
 ## [2.0.11] - 2026-06-13
 ### Fixed
 - Ignore failed CloudTrail `CreateLogGroup` events so they do not trigger duplicate subscription work.

src/lambda-manager/lambda_function.py

@@ -26,6 +26,8 @@
 cloudwatch_logs = boto3.client('logs', config=config)
 lambda_client = boto3.client('lambda', config=config)
 
+SUPPORTED_LOG_GROUP_CLASS = "STANDARD"
+
 def lambda_handler(event: Dict[str, Any], context) -> None:
     """
     Main Lambda function handler that manages CloudWatch log group subscriptions.
@@ -94,6 +96,9 @@ def lambda_handler(event: Dict[str, Any], context) -> None:
                 return
 
             log_group_to_subscribe = event_detail['requestParameters']['logGroupName']
+
+            if not should_process_log_group_class(event_detail):
+                return
             found_log_group_in_regex_pattern = False
 
             for regex_pattern in regex_pattern_list:
@@ -250,6 +255,21 @@ def should_create_subscription(cloudwatch_logs, log_group_name: str, destination
 
     return True
 
+def should_process_log_group_class(event_detail: Dict[str, Any]) -> bool:
+    """
+    Allow standard log groups, including events where CloudTrail omits logGroupClass.
+
+    CloudFormation-created log groups default to STANDARD even when the property is
+    not present in requestParameters. Explicit non-standard classes should still be
+    ignored to preserve the existing behavior.
+    """
+    log_group_class = event_detail.get('requestParameters', {}).get('logGroupClass', SUPPORTED_LOG_GROUP_CLASS)
+    if log_group_class == SUPPORTED_LOG_GROUP_CLASS:
+        return True
+
+    logger.info(f"Skipping log group because logGroupClass {log_group_class} is not supported")
+    return False
+
 def add_subscription(
     filter_name: str, 
     logs_filter: str, 

src/lambda-manager/template.yaml

@@ -16,7 +16,7 @@ Metadata:
       - cloudwatch
       - lambda
     HomePageUrl: https://coralogix.com
-    SemanticVersion: 2.0.11
+    SemanticVersion: 2.0.12
     SourceCodeUrl: https://github.com/coralogix/coralogix-aws-serverless
   AWS::CloudFormation::Interface:
     ParameterGroups:
@@ -215,6 +215,7 @@ Resources:
                 requestParameters:
                   logGroupClass:
                     - "STANDARD"
+                    - exists: false
             Target:
                 Id: cx-loggroup-target
 

src/lambda-manager/tests/test_lambda_function.py

@@ -107,7 +107,7 @@ def test_existing_destination_subscription_is_not_duplicated(self):
         add_permission.assert_not_called()
         add_subscription.assert_not_called()
 
-    def test_new_destination_subscription_still_gets_created(self):
+    def test_event_without_log_group_class_still_gets_subscribed(self):
         event = {"detail": {"requestParameters": {"logGroupName": "/aws/lambda/example"}}}
         self.module.cloudwatch_logs.describe_subscription_filters.return_value = {"subscriptionFilters": []}
 
@@ -124,6 +124,25 @@ def test_new_destination_subscription_still_gets_created(self):
         )
         add_subscription.assert_called_once()
 
+    def test_non_standard_log_group_class_is_ignored(self):
+        for log_group_class in ("INFREQUENT_ACCESS", "DELIVERY"):
+            event = {
+                "detail": {
+                    "requestParameters": {
+                        "logGroupName": "/aws/lambda/example",
+                        "logGroupClass": log_group_class,
+                    }
+                }
+            }
+
+            with self.subTest(log_group_class=log_group_class):
+                with patch.object(self.module, "add_permission_to_lambda") as add_permission, patch.object(
+                    self.module, "add_subscription"
+                ) as add_subscription:
+                    self.invoke_handler(event)
+
+                add_permission.assert_not_called()
+                add_subscription.assert_not_called()
 
 if __name__ == "__main__":
     unittest.main()

src/lambda-manager/tests/test_template.py

@@ -0,0 +1,28 @@
+import re
+import unittest
+from pathlib import Path
+
+
+TEMPLATE_PATH = Path(__file__).resolve().parents[1] / "template.yaml"
+
+
+class LambdaManagerTemplateTests(unittest.TestCase):
+    def test_eventbridge_pattern_matches_standard_or_missing_log_group_class(self):
+        template = TEMPLATE_PATH.read_text()
+        match = re.search(r"logGroupClass:\n((?:\s+-.*\n)+)", template)
+
+        self.assertIsNotNone(match)
+
+        log_group_class_pattern = [line.strip() for line in match.group(1).splitlines()]
+
+        self.assertEqual(
+            [
+                '- "STANDARD"',
+                "- exists: false",
+            ],
+            log_group_class_pattern,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()