Datasync: Copy files from S3 to Azure Blob Storage (1st Part Prepare manifest and run Datasync Task)

sensor
devops-incident-response

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.

Official Documentation

Related Posts

Leave a Reply

Your email address will not be published. Required fields are marked *

☕ Support us · 💳 Monobank