Replicate from AWS S3 to E2E Object Storage

View as Markdown

An AWS Lambda function that runs on S3 event notifications can copy objects from a folder in an AWS S3 bucket into an E2E Object Storage bucket as they appear there. Use it for live replication from S3 into Object Storage; to move objects between two E2E buckets, see Replicate a bucket to another region instead.

The pieces: an IAM role that lets the function read the S3 folder, a Lambda function that copies each new object to Object Storage with the MinIO SDK, and an S3 event notification that triggers it when an object is created or removed.

Before you begin

A bucket in E2E Object Storage, and an access key with a role on it; see Generate an access key.
The AWS account that owns the S3 bucket, with permission to create an IAM role, a Lambda function, and an event notification.

Create an IAM role that can read the S3 folder

In the AWS IAM console, create a role for Lambda and attach a policy that lets it list and read objects in the S3 bucket you are replicating from:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket", "s3:GetObject"],
      "Resource": ["arn:aws:s3:::your-bucket-name", "arn:aws:s3:::your-bucket-name/*"]
    }
  ]
}

Name the role (for example LambdaS3FullAccess) and attach it to the Lambda function you create next.

Create the Lambda function

In the AWS Lambda console, create a function from scratch, set the runtime to Python 3.10, and select the IAM role you created earlier.

The function needs the MinIO SDK installed alongside its code, so package it together. On your own machine:

mkdir -p lambda_package
pip install minio -t lambda_package/

Add the function code as lambda_function.py inside that folder. The function reads the S3 bucket’s environment variables, lists the folder, and copies anything missing into Object Storage, removing anything that no longer exists in S3:

import os
import io
import json
import boto3
from minio import Minio
from minio.error import S3Error

s3_client = boto3.client("s3")

MINIO_ENDPOINT = os.getenv("MINIO_ENDPOINT", "objectstore.e2enetworks.net")
MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY")
MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY")
MINIO_BUCKET = os.getenv("MINIO_BUCKET")
S3_BUCKET = os.getenv("S3_BUCKET")
S3_FOLDER = os.getenv("S3_FOLDER")
MINIO_FOLDER = os.getenv("MINIO_FOLDER")

minio_client = Minio(
    MINIO_ENDPOINT,
    access_key=MINIO_ACCESS_KEY,
    secret_key=MINIO_SECRET_KEY,
    secure=True,
)


def list_minio_objects():
    objects = minio_client.list_objects(MINIO_BUCKET, prefix=f"{MINIO_FOLDER}/", recursive=True)
    return {obj.object_name for obj in objects}


def sync_folders():
    s3_objects = s3_client.list_objects_v2(Bucket=S3_BUCKET, Prefix=S3_FOLDER)
    s3_keys = set()

    while True:
        if "Contents" in s3_objects:
            for obj in s3_objects["Contents"]:
                relative_path = obj["Key"][len(S3_FOLDER):].lstrip("/")
                s3_keys.add(relative_path)
                minio_object_path = f"{MINIO_FOLDER}/{relative_path}"

                try:
                    minio_client.stat_object(MINIO_BUCKET, minio_object_path)
                    continue
                except S3Error:
                    pass

                response = s3_client.get_object(Bucket=S3_BUCKET, Key=obj["Key"])
                data = io.BytesIO(response["Body"].read())
                minio_client.put_object(MINIO_BUCKET, minio_object_path, data, length=len(data.getvalue()))

        if not s3_objects.get("IsTruncated"):
            break
        s3_objects = s3_client.list_objects_v2(
            Bucket=S3_BUCKET, Prefix=S3_FOLDER, ContinuationToken=s3_objects.get("NextContinuationToken")
        )

    minio_objects = list_minio_objects()
    s3_keys_with_prefix = {f"{MINIO_FOLDER}/{key}" for key in s3_keys}
    objects_to_delete = minio_objects - s3_keys_with_prefix

    for obj in objects_to_delete:
        minio_client.remove_object(MINIO_BUCKET, obj)


def lambda_handler(event, context):
    try:
        sync_folders()
    except Exception as e:
        print(f"Error: {e}")
        return {"statusCode": 500, "body": json.dumps("Replication Failed")}

    return {"statusCode": 200, "body": json.dumps("Replication Successful")}

Zip the folder’s contents and upload it to the function:

cd lambda_package
zip -r ../lambda_function.zip .

In the Lambda console, choose Upload from .zip file, select the zip, and deploy.

Set the function’s environment variables

In the function’s Configuration > Environment Variables, add:

Variable	Value
MINIO_ACCESS_KEY	your E2E access key
MINIO_SECRET_KEY	your E2E secret key
MINIO_ENDPOINT	`objectstore.e2enetworks.net` (Delhi)
MINIO_BUCKET	your E2E bucket name
S3_BUCKET	your AWS S3 bucket name
S3_FOLDER	the folder inside it to replicate from
MINIO_FOLDER	the folder inside your E2E bucket to copy to

Use the Chennai endpoint from S3 compatibility if your E2E bucket is there.

Trigger the function from S3 events

In the S3 console, open the source bucket, then Properties > Event Notifications > Create Event Notification. Name it, select the event types, and set the destination to the Lambda function you created. Save the notification; from then on, objects created or removed in the folder trigger the function.

Test the Lambda function

Before wiring up real traffic, confirm the function works from a test event. In the Lambda console, create a test event shaped like an S3 event, with your own bucket and object key in place of the ones below:

{
  "Records": [
    {
      "eventSource": "aws:s3",
      "eventName": "ObjectCreated:Put",
      "s3": {
        "bucket": {
          "name": "your-bucket-name"
        },
        "object": {
          "key": "your-folder-name/example-object"
        }
      }
    }
  ]
}

Save the event, then invoke the function; it returns Replication Successful when the copy works. Test the whole setup in a test environment first; see Which objects the function changes for what it removes.

Which objects the function changes

The function works on two folders only: the S3 folder you chose and the matching Object Storage folder. It copies from S3 to Object Storage, one way, and keeps the two folders the same: anything in the Object Storage folder that is not in the S3 folder is removed, including objects that were there before the first run. Everything else in both buckets is left as it is. To keep existing objects, upload them to the S3 folder first, so the function tracks them too.

To check it end to end, upload a file to the S3 folder and list the E2E folder; see Connect with the mc client.

The function copies each object only if it is not already in the Object Storage folder. If you overwrite an object in S3 with a newer version of the same file, that change is not re-copied, because the object already exists at the destination. To pick up an overwritten file, delete its copy in the Object Storage folder first so the function copies it again.