datasync copy files from s3 to

This first-stage Lambda workflow prepares a DataSync manifest from MongoDB metadata, uploads the manifest to S3, and starts an AWS DataSync task for S3 to Azure Blob transfer. It also tracks incremental execution state via last_run.txt.

Input and Environment Configuration

# Datasync files from S3 to Azure Blob Storage
# 1st Part prepare manifest file and store it in S3
# Connect to MongoDB and query
# DataSync is created with Terraform

# Lambda ENV Variables:
DATASYNC_SOURCE_S3_BUCKET: {S3_BUCKET_NAME}
DATASYNC_TASK_ARN: arn:aws:datasync:{region}:{account_id}:task/{task_id}
DELETE_LAST_RUN_TIMESTAMP: false
LOOKBACK_DAYS: 7
MANIFEST_S3_BUCKET: {S3_BUCKET_NAME}
MONGODB_COLLECTION: {Collection_NAME}
MONGODB_DB: {MongoDB_NAME}
MONGODB_SECRET_ARN: arn:aws:secretsmanager:{region}:{account_id}:secret:{secret_name}
S3_MANIFEST_KEY: manifests/latest.csv

Lambda Handler (Python)

import os
import boto3
from pymongo import MongoClient
import json
import logging
from datetime import datetime, timedelta, timezone
import time

# logs
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def get_mongo_uri():
    """
    Retrieve MongoDB URI from AWS Secrets Manager (preferred) or environment variable (fallback).
    """
    secret_arn = os.environ.get('MONGODB_SECRET_ARN')
    if secret_arn:
        try:
            secrets_client = boto3.client('secretsmanager')
            secret_value = secrets_client.get_secret_value(SecretId=secret_arn)
            secret = json.loads(secret_value['SecretString'])
            return secret.get('MONGODB_URI')
        except Exception as e:
            logger.error(f"Failed to retrieve secret from Secrets Manager: {e}")
            raise
    # Fallback to environment variable
    return os.environ.get('MONGODB_URI')

def lambda_handler(event, context):
    s3_client = boto3.client('s3')

    try:
        # config
        mongo_uri = get_mongo_uri()
        mongo_db = os.environ.get('MONGODB_DB')
        mongo_collection = os.environ.get('MONGODB_COLLECTION')
        datasync_source_s3_bucket = os.environ.get('DATASYNC_SOURCE_S3_BUCKET')
        manifest_s3_bucket = os.environ.get('MANIFEST_S3_BUCKET')
        s3_manifest_key = os.environ.get('S3_MANIFEST_KEY')
        datasync_task_arn = os.environ.get('DATASYNC_TASK_ARN')
        lookback_days = int(os.environ.get('LOOKBACK_DAYS', '7'))
        delete_last_run = os.environ.get('DELETE_LAST_RUN_TIMESTAMP', 'False').lower() == 'true'
        logger.info(f"All environment variables: {os.environ}")
        if not all([mongo_uri, mongo_db, mongo_collection, datasync_source_s3_bucket, manifest_s3_bucket, s3_manifest_key, datasync_task_arn]):
            raise ValueError('Missing one or more required environment variables.')

        # delete last_run.txt if DELETE_LAST_RUN_TIMESTAMP is True
        last_run_key = 'last_run.txt'
        if delete_last_run:
            try:
                s3_client.delete_object(Bucket=manifest_s3_bucket, Key=last_run_key)
                logger.info(f"Deleted {last_run_key} from {manifest_s3_bucket} as requested by DELETE_LAST_RUN_TIMESTAMP.")
            except s3_client.exceptions.NoSuchKey:
                logger.info(f"No {last_run_key} found to delete in {manifest_s3_bucket}.")
            except Exception as e:
                logger.error(f"Error deleting {last_run_key}: {e}")

        # get lookback window from env (default 7 days)
        try:
            last_run_obj = s3_client.get_object(Bucket=manifest_s3_bucket, Key=last_run_key)
            last_run_time = last_run_obj['Body'].read().decode('utf-8')
            lookback_ago = datetime.fromisoformat(last_run_time)
            lookback_ago_ts_ms = int(lookback_ago.timestamp() * 1000)
            logger.info(f"Using last run time from S3: {last_run_time}")
        except s3_client.exceptions.NoSuchKey:
            lookback_ago = datetime.now(timezone.utc) - timedelta(days=lookback_days)
            lookback_ago_ts_ms = int(lookback_ago.timestamp() * 1000)
            logger.info(f"No last run time found, using lookback window: {lookback_days} days")
        except Exception as e:
            logger.error(f"Error reading last_run.txt: {e}")
            lookback_ago = datetime.now(timezone.utc) - timedelta(days=lookback_days)
            lookback_ago_ts_ms = int(lookback_ago.timestamp() * 1000)

        # connect to mongodb
        logger.info("Connecting to MongoDB...")
        client = MongoClient(mongo_uri)
        db = client[mongo_db]
        collection = db[mongo_collection]
        logger.info("Successfully connected to MongoDB.")

        # find files from the last N days
        logger.info(f"Finding files from the last {lookback_days} days...")
        logger.info(f"Current time ms: {int(datetime.now(timezone.utc).timestamp() * 1000)}")
        logger.info(f"lookback_ago_ts_ms: {lookback_ago_ts_ms}")
        logger.info(f"Total docs: {collection.count_documents({})}")
        logger.info(f"Docs with fileType 'recent': {collection.count_documents({'fileType': 'recent'})}")
        logger.info(f"Docs with type 'footage': {collection.count_documents({'type': 'footage'})}")
        logger.info(f"Docs with createdDate >= lookback_ago_ts_ms: {collection.count_documents({'createdDate': {'$gte': lookback_ago_ts_ms}})}")

        # query for recent footage files from the last N days
        logger.info("Querying for recent footage files (fileType: 'recent', type: 'footage').")
        footage_query = {
            'metadata.fileType': 'recent',
            'type': 'footage',
            'status': 'validated',
            'createdDate': {'$gte': lookback_ago_ts_ms}
        }
        logger.info(f"footage_query: {footage_query}")
        logger.info(f"Docs with all filters: {collection.count_documents(footage_query)}")
        recent_footage_files = list(collection.find(footage_query))

        if not recent_footage_files:
            logger.info("No recent footage files found from the last {lookback_days} days. Exiting.")
            return {'statusCode': 200, 'body': json.dumps({'message': 'No new files to process.'})}

        logger.info(f"Found {len(recent_footage_files)} recent footage files.")

        # collect session IDs and file names from footage files
        session_ids = [doc['sessionId'] for doc in recent_footage_files if 'sessionId' in doc]
        footage_file_keys = [
            f"{doc['userId']}/{doc['sessionId']}/{doc['metadata']['fileName']}"
            for doc in recent_footage_files
            if 'userId' in doc and 'sessionId' in doc and 'metadata' in doc and 'fileName' in doc['metadata']
        ]
        gps_file_keys = []
        if not session_ids:
            logger.warning("No session IDs found in footage files, so cannot look for corresponding GPS files.")
        else:
            # query for corresponding gps files using the session IDs
            logger.info(f"Querying for corresponding GPS files for {len(session_ids)} sessions.")
            gps_query = {'metadata.fileType': 'gps', 'type': 'footage', 'status': 'validated', 'sessionId': {'$in': list(set(session_ids))}}
            gps_files = list(collection.find(gps_query))
            logger.info(f"Found {len(gps_files)} corresponding GPS files.")
            gps_file_keys = [
                f"{doc['userId']}/{doc['sessionId']}/{doc['metadata']['fileName']}"
                for doc in gps_files
                if 'userId' in doc and 'sessionId' in doc and 'metadata' in doc and 'fileName' in doc['metadata']
            ]

        # combine file keys for the manifest
        all_file_keys = footage_file_keys + gps_file_keys

        if not all_file_keys:
            logger.info("No files to include in the manifest after processing. Exiting.")
            return {'statusCode': 200, 'body': json.dumps({'message': 'No files to process.'})}

        logger.info(f"Total files for manifest: {len(all_file_keys)} ({len(footage_file_keys)} footage, {len(gps_file_keys)} GPS).")

        # create and upload DataSync manifest
        logger.info("Creating DataSync manifest content.")
        manifest_content = "\n".join(all_file_keys)

        logger.info(f"Uploading manifest to s3://{manifest_s3_bucket}/{s3_manifest_key}")
        s3_client.put_object(Bucket=manifest_s3_bucket, Key=s3_manifest_key, Body=manifest_content)
        logger.info("Manifest file uploaded successfully.")

        # DataSync task execution start
        datasync_client = boto3.client('datasync')
        response = datasync_client.start_task_execution(TaskArn=datasync_task_arn)
        task_execution_arn = response['TaskExecutionArn']
        logger.info(f"Successfully started task execution: {task_execution_arn}")

        # save current time to last_run.txt
        now_iso = datetime.now(timezone.utc).isoformat()
        if not delete_last_run:
            s3_client.put_object(Bucket=manifest_s3_bucket, Key=last_run_key, Body=now_iso)
            logger.info(f"Saved last run time to {last_run_key}: {now_iso}")
        else:
            logger.info("Not saving last run time because DELETE_LAST_RUN_TIMESTAMP is True.")

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Manifest created, DataSync task started, and last run time saved.',
                'files_in_manifest': len(all_file_keys),
                'lookback_start': lookback_ago.isoformat(),
                'lookback_days': lookback_days,
                'taskExecutionArn': task_execution_arn
            })
        }
    except Exception as e:
        logger.error(f"An error occurred: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

Operational Notes

Store MongoDB credentials in Secrets Manager and avoid plain env URIs where possible.
Validate manifest path and DataSync task ARN before production runs.
Use CloudWatch alarms for task execution failures and zero-file manifest outcomes.
Ensure MongoDB query filters match your production status/fileType conventions.