fix: add sns support

.gitignore

@@ -22,6 +22,8 @@ venv/
 env.json
 samconfig.toml
 packaged.yml
+package.json
+package-lock.json
 # Serverless directories
 .serverless
 # SAM

src/handler.py

@@ -90,7 +90,7 @@ class BadRequestException(Exception):
 
 def _is_ignore_log_file(key=None, regex_pattern=None):
     """
-    This functions checks whether this log file should be ignored based on regex pattern.
+    This function checks whether this log file should be ignored based on regex pattern.
     """
     if not regex_pattern:
         regex_pattern = _get_optional_env("S3_IGNORE_PATTERN", "$^")
@@ -100,7 +100,7 @@ def _is_ignore_log_file(key=None, regex_pattern=None):
 
 def _isCloudTrail(key=None, regex_pattern=None):
     """
-    This functions checks whether this log file is a CloudTrail log based on regex pattern.
+    This function checks whether this log file is a CloudTrail log based on regex pattern.
     """
     if not regex_pattern:
         regex_pattern = _get_optional_env(
@@ -110,7 +110,7 @@ def _isCloudTrail(key=None, regex_pattern=None):
 
 def _isCloudTrailDigest(key=None):
     """
-    This functions checks whether this log file is a CloudTrail-Digest based on regex pattern.
+    This function checks whether this log file is a CloudTrail-Digest based on regex pattern.
     """
     return bool(re.search(".*_CloudTrail-Digest_.*\.json.gz$", key))
 
@@ -123,15 +123,15 @@ def _convert_float(s):
 
 def _get_batch_size_factor(batch_size_factor=None):
     """
-    This functions gets BATCH_SIZE_FACTOR from env vars.
+    This function gets BATCH_SIZE_FACTOR from env vars.
     """
     if batch_size_factor:
         return batch_size_factor
     return _convert_float(_get_optional_env("BATCH_SIZE_FACTOR", BATCH_SIZE_FACTOR))
 
 def _get_license_key(license_key=None):
     """
-    This functions gets New Relic's license key from env vars.
+    This function gets New Relic's license key from env vars.
     """
     if license_key:
         return license_key
@@ -140,7 +140,7 @@ def _get_license_key(license_key=None):
 
 def _get_log_type(log_type=None):
     """
-    This functions gets the New Relic logtype from env vars.
+    This function gets the New Relic logtype from env vars.
     """
     return log_type or _get_optional_env("LOG_TYPE", "")
 
@@ -272,6 +272,21 @@ def create_log_payload_request(data, session):
     return send_log(session, req.get_full_url(), req.data, req.headers)
 
 
+def get_s3_event(event):
+    if "s3" in event["Records"][0]:
+        return event
+    elif "Sns" in event["Records"][0]:
+        sns_msg = event["Records"][0]["Sns"]["Message"]
+        try:
+            sns_msg_dict = json.loads(sns_msg)
+            if "Records" in sns_msg_dict and "s3" in sns_msg_dict["Records"][0]:
+                return sns_msg_dict
+        except Exception:
+            if logger.isEnableFor(logging.DEBUG):
+                logger.debug(f"No s3 event detected from SNS message: {sns_msg}")
+        raise Exception("Event type not supported")
+
+
 async def _fetch_data_from_s3(bucket, key, context):
     """
         Stream data from S3 bucket. Create batches of size MAX_PAYLOAD_SIZE
@@ -339,9 +354,11 @@ async def _fetch_data_from_s3(bucket, key, context):
 def lambda_handler(event, context):
     # Get bucket from s3 upload event
     _setting_console_logging_level()
-    bucket = event['Records'][0]['s3']['bucket']['name']
+
+    s3_event = get_s3_event(event)
+    bucket = s3_event['Records'][0]['s3']['bucket']['name']
     key = urllib.parse.unquote_plus(
-        event['Records'][0]['s3']['object']['key'], encoding='utf-8')
+        s3_event['Records'][0]['s3']['object']['key'], encoding='utf-8')
 
     # Allow user to skip log file using regex pattern set in env variable: S3_IGNORE_PATTERN 
     if _is_ignore_log_file(key):

src/requirements.txt

@@ -1,20 +1,20 @@
-aiohttp>=3.9.5
+aiohttp>=3.10.2
 async-timeout==4.0.2
 attrs==19.3.0
 boto==2.49.0
 boto3==1.34.31
 botocore==1.34.31
-certifi==2023.7.22
+certifi==2024.07.04
 chardet==3.0.4
 docutils==0.15.2
 idna>=3.7
 jmespath==0.10.0
-multidict==4.7.6
+multidict>=6.0.5
 python-dateutil==2.8.1
 requests>=2.32.0
 s3transfer==0.10.0
 six==1.15.0
-smart-open==2.1.0
+smart-open==7.0.5
 typing-extensions==3.7.4.2
 urllib3==2.0.7
 yarl==1.9.1

template.yml

@@ -105,8 +105,7 @@ Resources:
         BucketEvent1:
           Type: S3
           Properties:
-            Bucket:
-              Ref: SourceLogBucket
+            Bucket: !Ref SourceLogBucket
             Events:
               - 's3:ObjectCreated:*'
 
@@ -141,8 +140,7 @@ Resources:
         BucketEvent1:
           Type: S3
           Properties:
-            Bucket:
-              Ref: SourceLogBucket
+            Bucket: !Ref SourceLogBucket
             Events:
               - 's3:ObjectCreated:*'
 

test/mock.json → test/s3-mock.json

@@ -3,7 +3,7 @@
     {
       "eventVersion": "2.1",
       "eventSource": "aws:s3",
-      "awsRegion": "us-east-1",
+      "awsRegion": "us-east-2",
       "eventName": "ObjectCreated:Put",
       "s3": {
         "s3SchemaVersion": "1.0",

test/sns-mock.json

@@ -0,0 +1,17 @@
+{
+    "Records": [
+        {
+            "EventSource": "aws:sns",
+            "EventVersion": "1.0",
+            "EventSubscriptionArn": "arn:aws:sns:us-east-2:1111111:nr-logging-test",
+            "Sns": {
+                "Type": "Notification",
+                "MessageId": "1",
+                "TopicArn": "arn:aws:sns:us-east-2:1111111:nr-logging-test",
+                "Subject": "Amazon S3 Notification",
+                "Message": "{\"Records\":[{\"eventVersion\":\"2.1\",\"eventSource\":\"aws:s3\",\"awsRegion\":\"us-east-2\",\"eventName\":\"ObjectCreated:Put\",\"s3\":{\"s3SchemaVersion\":\"1.0\",\"bucket\":{\"name\":\"s3-nr-log-test-bucket\",\"arn\":\"arn:aws:s3:::s3-nr-log-test-bucket\"},\"object\":{\"key\":\"cloudflare-test-data.json.gz\"}}}]}",
+                "Timestamp": "2025-06-01T19:56:13.299Z"
+            }
+        }
+    ]
+}