Datasync: Copy files from S3 to Azure Blob Storage (2nd Report)

sensor
devops-incident-response

This second-stage Lambda processes the DataSync execution report from S3, estimates transfer/request costs, builds a readable summary, and publishes the report to SNS for operational visibility.

Input and Environment Configuration

# Report Lambda
# Generate Report and send it to SNS
# Lambda Env Variable
DATASYNC_COST_PER_GB: 0.015
DATASYNC_TASK_EXECUTION_COST: 0.55
DATA_TRANSFER_OUT_FIRST_10TB_PER_GB: 0.09
DATA_TRANSFER_OUT_GREATER_150TB_PER_GB: 0.05
DATA_TRANSFER_OUT_NEXT_100TB_PER_GB: 0.07
DATA_TRANSFER_OUT_NEXT_40TB_PER_GB: 0.085
S3_GET_COST_PER_1K: 0.0004
S3_HEAD_COST_PER_1K: 0.0004
S3_LIST_COST_PER_1K: 0.005
S3_PUT_COST_PER_1K: 0.005
SNS_TOPIC_ARN: arn:aws:sns:{region}:{account_id}:{SNS_TOPIC_NAME}

Report Lambda (Python)

import os
import json
import boto3
import logging
import math

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

s3_client = boto3.client('s3')
sns_client = boto3.client('sns')

# Constants (as of July 2025, Enhanced mode)
DATASYNC_COST_PER_GB = float(os.environ.get('DATASYNC_COST_PER_GB', '0.015'))
DATASYNC_TASK_EXECUTION_COST = float(os.environ.get('DATASYNC_TASK_EXECUTION_COST', '0.55'))

# S3 request costs
S3_LIST_COST_PER_1K = float(os.environ.get('S3_LIST_COST_PER_1K', '0.005'))
S3_HEAD_COST_PER_1K = float(os.environ.get('S3_HEAD_COST_PER_1K', '0.0004'))
S3_GET_COST_PER_1K = float(os.environ.get('S3_GET_COST_PER_1K', '0.0004'))
S3_PUT_COST_PER_1K = float(os.environ.get('S3_PUT_COST_PER_1K', '0.005'))
S3_COPY_COST_PER_1K = float(os.environ.get('S3_COPY_COST_PER_1K', '0.005'))

# Data transfer out costs
DATA_TRANSFER_OUT_FIRST_10TB_PER_GB = float(os.environ.get('DATA_TRANSFER_OUT_FIRST_10TB_PER_GB', '0.09'))
DATA_TRANSFER_OUT_NEXT_40TB_PER_GB = float(os.environ.get('DATA_TRANSFER_OUT_NEXT_40TB_PER_GB', '0.085'))
DATA_TRANSFER_OUT_NEXT_100TB_PER_GB = float(os.environ.get('DATA_TRANSFER_OUT_NEXT_100TB_PER_GB', '0.07'))
DATA_TRANSFER_OUT_GREATER_150TB_PER_GB = float(os.environ.get('DATA_TRANSFER_OUT_GREATER_150TB_PER_GB', '0.05'))

def calculate_s3_request_cost(report):
    """Calculate S3 request costs based on DataSync report"""
    result = report.get('Result', {})
    files_listed = result.get('FilesListed', {})
    files_transferred = result.get('FilesTransferred', 0)
    source_is_s3 = report.get('SourceLocation', {}).get('LocationType') == 'Amazon S3'
    dest_is_s3 = report.get('DestinationLocation', {}).get('LocationType') == 'Amazon S3'

    list_requests = 0
    head_requests = 0
    get_requests = 0
    put_requests = 0
    copy_requests = 0

    if source_is_s3:
        list_requests += math.ceil(files_listed.get('AtSource', 0) / 1000.0)
        head_requests += files_listed.get('AtSource', 0)
        get_requests += files_transferred

    if dest_is_s3:
        list_requests += math.ceil(files_listed.get('AtDestinationForDelete', 0) / 1000.0)
        head_requests += files_listed.get('AtDestinationForDelete', 0)
        put_requests += files_transferred
        copy_requests += files_transferred

    costs = {
        'LIST': (list_requests / 1000.0) * S3_LIST_COST_PER_1K,
        'HEAD': (head_requests / 1000.0) * S3_HEAD_COST_PER_1K,
        'GET': (get_requests / 1000.0) * S3_GET_COST_PER_1K,
        'PUT': (put_requests / 1000.0) * S3_PUT_COST_PER_1K,
        'COPY': (copy_requests / 1000.0) * S3_COPY_COST_PER_1K,
    }

    return costs, {
        'LIST': list_requests,
        'HEAD': head_requests,
        'GET': get_requests,
        'PUT': put_requests,
        'COPY': copy_requests,
    }

def calculate_data_transfer_out_cost(gb_transferred):
    if gb_transferred <= 10240:
        return gb_transferred * DATA_TRANSFER_OUT_FIRST_10TB_PER_GB
    elif gb_transferred <= 51200:
        return (10240 * DATA_TRANSFER_OUT_FIRST_10TB_PER_GB) + ((gb_transferred - 10240) * DATA_TRANSFER_OUT_NEXT_40TB_PER_GB)
    elif gb_transferred <= 151200:
        return (
            (10240 * DATA_TRANSFER_OUT_FIRST_10TB_PER_GB)
            + (40960 * DATA_TRANSFER_OUT_NEXT_40TB_PER_GB)
            + ((gb_transferred - 51200) * DATA_TRANSFER_OUT_NEXT_100TB_PER_GB)
        )
    else:
        return (
            (10240 * DATA_TRANSFER_OUT_FIRST_10TB_PER_GB)
            + (40960 * DATA_TRANSFER_OUT_NEXT_40TB_PER_GB)
            + (100000 * DATA_TRANSFER_OUT_NEXT_100TB_PER_GB)
            + ((gb_transferred - 151200) * DATA_TRANSFER_OUT_GREATER_150TB_PER_GB)
        )

def lambda_handler(event, context):
    try:
        record = event['Records'][0]
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        logger.info(f"Triggered by S3 object: s3://{bucket}/{key}")
    except Exception as e:
        logger.error(f"Could not parse S3 event: {e}")
        return {"statusCode": 400, "body": "Invalid S3 event"}

    try:
        response = s3_client.get_object(Bucket=bucket, Key=key)
        report = json.loads(response['Body'].read())
    except Exception as e:
        logger.error(f"Could not read or parse report: {e}")
        return {"statusCode": 500, "body": "Report processing failed"}

    try:
        account_id = report.get('AccountId')
        task_execution_id = report.get('TaskExecutionId')
        source_location = report.get('SourceLocation', {})
        destination_location = report.get('DestinationLocation', {})
        start_time = report.get('StartTime')
        end_time = report.get('EndTime')
        total_time = report.get('TotalTime')
        overall_status = report.get('OverallStatus')
        result = report.get('Result', {})
        files_transferred = result.get('FilesTransferred', 0)
        bytes_transferred = result.get('BytesTransferred', 0)
        source_location_type = source_location.get('LocationType', 'Unknown')
        destination_location_type = destination_location.get('LocationType', 'Unknown')
        files_deleted = result.get('FilesDeleted', 0)
        files_verified = result.get('FilesVerified', 0)
        files_skipped = result.get('FilesSkipped', 0)
        error_code = result.get('ErrorCode')
        error_detail = result.get('ErrorDetail')
        files_listed = result.get('FilesListed', {})
    except Exception as e:
        logger.error(f"Could not extract fields: {e}")
        return {"statusCode": 500, "body": "Field extraction failed"}

    try:
        gb_transferred = bytes_transferred / (1024 ** 3)
        datasync_cost = gb_transferred * DATASYNC_COST_PER_GB
        task_execution_cost = DATASYNC_TASK_EXECUTION_COST
        s3_costs, request_counts = calculate_s3_request_cost(report)
        s3_request_cost = sum(s3_costs.values())
        data_transfer_out_cost = calculate_data_transfer_out_cost(gb_transferred)
        total_cost = datasync_cost + task_execution_cost + s3_request_cost + data_transfer_out_cost
    except Exception as e:
        logger.error(f"Could not calculate cost: {e}")
        return {"statusCode": 500, "body": "Cost calculation failed"}

    s3_cost_details = ""
    for req_type, cost in s3_costs.items():
        if cost > 0:
            count = request_counts[req_type]
            s3_cost_details += f"  {req_type} requests ({count}): \${cost:.8f}\\n"

    summary = (
        f"AccountId: {account_id}\\n"
        f"TaskExecutionId: {task_execution_id}\\n"
        f"SourceLocation: {source_location_type} ({source_location.get('LocationId', 'N/A')})\\n"
        f"DestinationLocation: {destination_location_type} ({destination_location.get('LocationId', 'N/A')})\\n"
        f"StartTime: {start_time}\\n"
        f"EndTime: {end_time}\\n"
        f"TotalTime: {total_time} ms\\n"
        f"OverallStatus: {overall_status}\\n"
    )

    if error_code:
        summary += f"ErrorCode: {error_code}\\n"
    if error_detail:
        summary += f"ErrorDetail: {error_detail}\\n"

    summary += (
        f"\\nTransfer Results:\\n"
        f"  FilesTransferred: {files_transferred}\\n"
        f"  BytesTransferred: {bytes_transferred} bytes ({gb_transferred:.2f} GB)\\n"
        f"  FilesDeleted: {files_deleted}\\n"
        f"  FilesVerified: {files_verified}\\n"
        f"  FilesSkipped: {files_skipped}\\n"
        f"  FilesListed (Source): {files_listed.get('AtSource', 0)}\\n"
        f"  FilesListed (Destination): {files_listed.get('AtDestinationForDelete', 0)}\\n"
        f"\\nEstimated Cost Breakdown:\\n"
        f"  DataSync Costs:\\n"
        f"    Data transferred: \${datasync_cost:.6f} ({gb_transferred:.2f} GB × \${DATASYNC_COST_PER_GB}/GB)\\n"
        f"    Task executions: \${task_execution_cost:.2f}\\n"
    )

    if s3_request_cost > 0:
        summary += f"  S3 Request Charges:\\n{s3_cost_details}"
    else:
        summary += "  S3 Request Charges: \$0.00\\n"

    summary += (
        f"  Data transfer out (to Internet): \${data_transfer_out_cost:.6f}\\n"
        f"  TOTAL ESTIMATED COST: \${total_cost:.6f}\\n\\n"
    )

    logger.info(f"Summary:\\n{summary}")

    try:
        sns_topic_arn = os.environ['SNS_TOPIC_ARN']
        status_indicator = "SUCCESS" if overall_status == "SUCCESS" else "ERROR"
        sns_client.publish(
            TopicArn=sns_topic_arn,
            Subject=f"AWS DataSync Task Report - {task_execution_id} [{status_indicator}]",
            Message=summary,
        )
        logger.info("Report sent to SNS.")
    except Exception as e:
        logger.error(f"Could not send SNS notification: {e}")
        return {"statusCode": 500, "body": "SNS notification failed"}

    return {"statusCode": 200, "body": "Report processed and sent."}

Operational Notes

  • Cost model values are configurable via env vars and should be reviewed when pricing changes.
  • Ensure S3 trigger points only to validated DataSync report objects.
  • SNS topic policy must allow Lambda publish in your account/region.
  • Consider adding CloudWatch metrics for cost anomalies and repeated failures.

Official Documentation

Related Posts

Leave a Reply

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

☕ Support us · 💳 Monobank