feat(cdk-ops): prepend prefix to masked logs

- `lambda/mask-access-logs` prepends a prefix to the keys of masked
  access logs files. The prefix is "masked/".

issue codemonger-io#30

cdk-ops/lambda/delete-access-logs/index.py

@@ -4,10 +4,12 @@
 masked access logs file.
 
 You have to specify the following environment variables,
-* ``SOURCE_BUCKET_NAME``: bucket name of the original CloudFront access logs
-  files
-* ``DESTINATION_BUCKET_NAME``: bucket name of the masked CloudFront access logs
-  files
+* ``SOURCE_BUCKET_NAME``: name of the S3 bucket containing original CloudFront
+  access logs files
+* ``DESTINATION_BUCKET_NAME``: name of the S3 bucket containing transformed
+  CloudFront access logs files
+* ``DESTINATION_KEY_PREFIX``: prefix of S3 object keys, which corresponds to
+  masked access logs
 """
 
 import json
@@ -18,6 +20,7 @@
 
 SOURCE_BUCKET_NAME = os.environ['SOURCE_BUCKET_NAME']
 DESTINATION_BUCKET_NAME = os.environ['DESTINATION_BUCKET_NAME']
+DESTINATION_KEY_PREFIX = os.environ['DESTINATION_KEY_PREFIX']
 
 LOGGER = logging.getLogger(__name__)
 LOGGER.setLevel(logging.DEBUG)
@@ -31,6 +34,9 @@ def lambda_handler(event, _):
 
     ``event`` is supposed to be SQS events described at
     https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html
+
+    Each SQS event is supposed to be an object-creation notification from the
+    S3 bucket containing masked access logs.
     """
     for record in event['Records']:
         body = record.get('body')
@@ -89,6 +95,15 @@ def process_s3_object(s3_object):
     if key is None:
         LOGGER.error('no object key in S3 object event: %s', str(s3_object))
         return
+    if not key.startswith(DESTINATION_KEY_PREFIX):
+        LOGGER.warning(
+            '"%s" does not have the preifx "%s".'
+            ' please check the event source configuration',
+            key,
+            DESTINATION_KEY_PREFIX,
+        )
+        return
+    key = key[len(DESTINATION_KEY_PREFIX):]
     src = source_bucket.Object(key)
     res = src.delete()
     LOGGER.debug('deleted object "%s": %s', key, str(res))

cdk-ops/lambda/mask-access-logs/index.py

@@ -1,10 +1,14 @@
 # -*- coding: utf-8 -*-
 
-"""Masks information in CloudFront access logs.
+"""Masks information in CloudFront access logs files.
 
 You have to specify the following environment variables,
-* SOURCE_BUCKET_NAME: name of the S3 bucket containing access logs files to be
-  masked.
+* SOURCE_BUCKET_NAME: name of the S3 bucket containing CloudFront access logs
+  files to be masked.
+* DESTINATION_BUCKET_NAME: name of the S3 bucket where masked CloudFront access
+  logs files are to be written.
+* DESTINATION_KEY_PREFIX: prefix to be prepended to the keys of objects in the
+  destination bucket.
 """
 
 import array
@@ -21,8 +25,9 @@
 import boto3
 
 
-SOURCE_BUCKET_NAME = os.environ.get('SOURCE_BUCKET_NAME')
-DESTINATION_BUCKET_NAME = os.environ.get('DESTINATION_BUCKET_NAME')
+SOURCE_BUCKET_NAME = os.environ['SOURCE_BUCKET_NAME']
+DESTINATION_BUCKET_NAME = os.environ['DESTINATION_BUCKET_NAME']
+DESTINATION_KEY_PREFIX = os.environ['DESTINATION_KEY_PREFIX']
 
 LOGGER = logging.getLogger(__name__)
 LOGGER.setLevel(logging.DEBUG)
@@ -123,10 +128,17 @@ def process_logs(logs_in: Iterator[str], logs_out: TextIO):
 
 
 def lambda_handler(event, _):
-    """Masks information in a given CloudFront access logs file on S3.
+    """Masks information in given CloudFront access logs files on S3.
 
     ``event`` is supposed to be an SQS message event described at
     https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html
+
+    Each SQS message event is supposed to be an object-creation notification
+    from the S3 bucket specified by ``SOURCE_BUCKET_NAME``.
+
+    This handler masks information in the given S3 objects and stores masked
+    results into the S3 bucket specified by ``DESTINATION_BUCKET_NAME`` with
+    the same object key but with ``DESTINATION_KEY_PREFIX`` prefixed.
     """
     for record in event['Records']:
         try:
@@ -189,7 +201,7 @@ def process_s3_object(s3_object):
         return
     with open_body(results) as body:
         with gzip.open(body, mode='rt') as tsv_in:
-            dest = destination_bucket.Object(key)
+            dest = destination_bucket.Object(f'{DESTINATION_KEY_PREFIX}{key}')
             with S3OutputStream(dest) as masked_out:
                 with gzip.open(masked_out, mode='wt') as tsv_out:
                     process_logs(tsv_in, tsv_out)

cdk-ops/lib/access-logs-etl.ts

@@ -58,6 +58,7 @@ export class AccessLogsETL extends Construct {
 
     // masks newly created CloudFront access logs
     // - Lambda function
+    const maskedAccessLogsKeyPrefix = 'masked/';
     const maskAccessLogsLambdaTimeout = Duration.seconds(30);
     const maskAccessLogsLambda = new PythonFunction(
       this,
@@ -71,6 +72,7 @@ export class AccessLogsETL extends Construct {
         environment: {
           SOURCE_BUCKET_NAME: accessLogsBucket.bucketName,
           DESTINATION_BUCKET_NAME: this.maskedAccessLogsBucket.bucketName,
+          DESTINATION_KEY_PREFIX: maskedAccessLogsKeyPrefix,
         },
         timeout: maskAccessLogsLambdaTimeout,
       },
@@ -127,6 +129,7 @@ export class AccessLogsETL extends Construct {
           SOURCE_BUCKET_NAME: accessLogsBucket.bucketName,
           // bucket name for masked logs is necessary to verify input events.
           DESTINATION_BUCKET_NAME: this.maskedAccessLogsBucket.bucketName,
+          DESTINATION_KEY_PREFIX: maskedAccessLogsKeyPrefix,
         },
         timeout: deleteAccessLogsLambdaTimeout,
       },
@@ -143,6 +146,7 @@ export class AccessLogsETL extends Construct {
     this.maskedAccessLogsBucket.addEventNotification(
       s3.EventType.OBJECT_CREATED,
       new s3n.SqsDestination(maskedLogsQueue),
+      { prefix: maskedAccessLogsKeyPrefix },
     );
     deleteAccessLogsLambda.addEventSource(
       new lambda_event.SqsEventSource(maskedLogsQueue, {