Benchmark: AWS S3 Express One Zone vs Cloudflare R2 for High-Throughput Object Storage

import asyncio
import boto3
import time
import os
import argparse
from botocore.config import Config
from botocore.exceptions import ClientError, EndpointConnectionError

# Benchmark configuration for AWS S3 Express One Zone
# Methodology: 100 concurrent workers, 1MB objects, 10k total objects per run
# Hardware: AWS EC2 c7g.16xlarge (64 vCPU, 128GB RAM, 100Gbps network)
# SDK Version: boto3 1.34.12, botocore 1.34.12
# Region: us-east-1 (S3 Express One Zone bucket in us-east-1a)

class S3ExpressBenchmark:
    def __init__(self, bucket: str, prefix: str, object_size: int = 1024*1024):
        self.bucket = bucket
        self.prefix = prefix
        self.object_size = object_size
        # S3 Express One Zone optimized config: higher max pool connections, no retries for benchmark
        self.config = Config(
            retries={'max_attempts': 0},
            max_pool_connections=100,
            s3={'payload_signing_enabled': False}  # Disable payload signing for higher throughput
        )
        self.client = boto3.client(
            's3',
            region_name='us-east-1',
            config=self.config
        )
        self.results = {'puts': [], 'gets': []}

    async def _put_object(self, object_id: int) -> float:
        '''Upload a single 1MB object, return latency in ms'''
        key = f'{self.prefix}/obj_{object_id}'
        data = os.urandom(self.object_size)
        start = time.perf_counter()
        try:
            self.client.put_object(
                Bucket=self.bucket,
                Key=key,
                Body=data,
                StorageClass='EXPRESS_ONEZONE'  # Explicitly set S3 Express storage class
            )
            end = time.perf_counter()
            return (end - start) * 1000  # ms
        except (ClientError, EndpointConnectionError) as e:
            print(f'PUT failed for {key}: {str(e)}')
            return -1  # Mark failed requests

    async def _get_object(self, object_id: int) -> float:
        '''Download a single 1MB object, return latency in ms'''
        key = f'{self.prefix}/obj_{object_id}'
        start = time.perf_counter()
        try:
            response = self.client.get_object(Bucket=self.bucket, Key=key)
            # Read entire body to ensure full download
            response['Body'].read()
            end = time.perf_counter()
            return (end - start) * 1000  # ms
        except (ClientError, EndpointConnectionError) as e:
            print(f'GET failed for {key}: {str(e)}')
            return -1

    async def run_put_benchmark(self, num_objects: int, concurrency: int = 100) -> None:
        '''Run PUT benchmark with specified concurrency'''
        print(f'Starting S3 Express PUT benchmark: {num_objects} objects, {concurrency} concurrency')
        sem = asyncio.Semaphore(concurrency)
        async def put_with_semaphore(obj_id):
            async with sem:
                return await self._put_object(obj_id)
        tasks = [put_with_semaphore(i) for i in range(num_objects)]
        self.results['puts'] = await asyncio.gather(*tasks)
        # Filter out failed requests
        self.results['puts'] = [lat for lat in self.results['puts'] if lat > 0]
        print(f'PUT benchmark complete: {len(self.results["puts"])} successful requests')

    async def run_get_benchmark(self, num_objects: int, concurrency: int = 100) -> None:
        '''Run GET benchmark with specified concurrency'''
        print(f'Starting S3 Express GET benchmark: {num_objects} objects, {concurrency} concurrency')
        sem = asyncio.Semaphore(concurrency)
        async def get_with_semaphore(obj_id):
            async with sem:
                return await self._get_object(obj_id)
        tasks = [get_with_semaphore(i) for i in range(num_objects)]
        self.results['gets'] = await asyncio.gather(*tasks)
        self.results['gets'] = [lat for lat in self.results['gets'] if lat > 0]
        print(f'GET benchmark complete: {len(self.results["gets"])} successful requests')

    def calculate_metrics(self) -> dict:
        '''Calculate p50, p99, avg latency and throughput'''
        metrics = {}
        for op in ['puts', 'gets']:
            lats = sorted(self.results[op])
            if not lats:
                continue
            metrics[op] = {
                'p50': lats[len(lats)//2],
                'p99': lats[int(len(lats)*0.99)],
                'avg': sum(lats)/len(lats),
                'throughput_gbps': (len(lats) * self.object_size * 8) / (sum(lats)/1000) / 1e9
            }
        return metrics

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='AWS S3 Express One Zone Benchmark')
    parser.add_argument('--bucket', required=True, help='S3 Express bucket name')
    parser.add_argument('--prefix', default='bench', help='Object key prefix')
    parser.add_argument('--num-objects', type=int, default=10000, help='Number of objects per benchmark')
    parser.add_argument('--concurrency', type=int, default=100, help='Concurrent requests')
    args = parser.parse_args()

    bench = S3ExpressBenchmark(args.bucket, args.prefix)
    loop = asyncio.get_event_loop()
    loop.run_until_complete(bench.run_put_benchmark(args.num_objects, args.concurrency))
    loop.run_until_complete(bench.run_get_benchmark(args.num_objects, args.concurrency))
    metrics = bench.calculate_metrics()
    print('\nS3 Express One Zone Benchmark Results:')
    for op, vals in metrics.items():
        print(f'{op.upper()}: p50={vals["p50"]:.2f}ms, p99={vals["p99"]:.2f}ms, throughput={vals["throughput_gbps"]:.2f}Gbps')

import asyncio
import boto3
import time
import os
import argparse
from botocore.config import Config
from botocore.exceptions import ClientError, EndpointConnectionError

# Benchmark configuration for Cloudflare R2
# Methodology: 100 concurrent workers, 1MB objects, 10k total objects per run
# Hardware: Cloudflare Workers (global edge, 1vCPU per worker, 128MB RAM) + local test client (100Gbps)
# SDK Version: boto3 1.34.12, botocore 1.34.12
# Region: auto (R2 is global, uses us-east-1 compatible region for S3 API)

class R2Benchmark:
    def __init__(self, bucket: str, prefix: str, account_id: str, access_key: str, secret_key: str, object_size: int = 1024*1024):
        self.bucket = bucket
        self.prefix = prefix
        self.object_size = object_size
        self.account_id = account_id
        # R2 S3-compatible endpoint configuration
        self.endpoint_url = f'https://{account_id}.r2.cloudflarestorage.com'
        self.config = Config(
            retries={'max_attempts': 0},
            max_pool_connections=100,
            s3={'payload_signing_enabled': False}
        )
        self.client = boto3.client(
            's3',
            region_name='us-east-1',  # R2 uses us-east-1 as compatible region
            endpoint_url=self.endpoint_url,
            aws_access_key_id=access_key,
            aws_secret_access_key=secret_key,
            config=self.config
        )
        self.results = {'puts': [], 'gets': []}

    async def _put_object(self, object_id: int) -> float:
        '''Upload a single 1MB object to R2, return latency in ms'''
        key = f'{self.prefix}/obj_{object_id}'
        data = os.urandom(self.object_size)
        start = time.perf_counter()
        try:
            self.client.put_object(
                Bucket=self.bucket,
                Key=key,
                Body=data
                # R2 does not support storage class parameters, uses standard global storage
            )
            end = time.perf_counter()
            return (end - start) * 1000  # ms
        except (ClientError, EndpointConnectionError) as e:
            print(f'R2 PUT failed for {key}: {str(e)}')
            return -1

    async def _get_object(self, object_id: int) -> float:
        '''Download a single 1MB object from R2, return latency in ms'''
        key = f'{self.prefix}/obj_{object_id}'
        start = time.perf_counter()
        try:
            response = self.client.get_object(Bucket=self.bucket, Key=key)
            response['Body'].read()
            end = time.perf_counter()
            return (end - start) * 1000  # ms
        except (ClientError, EndpointConnectionError) as e:
            print(f'R2 GET failed for {key}: {str(e)}')
            return -1

    async def run_put_benchmark(self, num_objects: int, concurrency: int = 100) -> None:
        '''Run PUT benchmark with specified concurrency'''
        print(f'Starting R2 PUT benchmark: {num_objects} objects, {concurrency} concurrency')
        sem = asyncio.Semaphore(concurrency)
        async def put_with_semaphore(obj_id):
            async with sem:
                return await self._put_object(obj_id)
        tasks = [put_with_semaphore(i) for i in range(num_objects)]
        self.results['puts'] = await asyncio.gather(*tasks)
        self.results['puts'] = [lat for lat in self.results['puts'] if lat > 0]
        print(f'R2 PUT benchmark complete: {len(self.results["puts"])} successful requests')

    async def run_get_benchmark(self, num_objects: int, concurrency: int = 100) -> None:
        '''Run GET benchmark with specified concurrency'''
        print(f'Starting R2 GET benchmark: {num_objects} objects, {concurrency} concurrency')
        sem = asyncio.Semaphore(concurrency)
        async def get_with_semaphore(obj_id):
            async with sem:
                return await self._get_object(obj_id)
        tasks = [get_with_semaphore(i) for i in range(num_objects)]
        self.results['gets'] = await asyncio.gather(*tasks)
        self.results['gets'] = [lat for lat in self.results['gets'] if lat > 0]
        print(f'R2 GET benchmark complete: {len(self.results["gets"])} successful requests')

    def calculate_metrics(self) -> dict:
        '''Calculate p50, p99, avg latency and throughput for R2'''
        metrics = {}
        for op in ['puts', 'gets']:
            lats = sorted(self.results[op])
            if not lats:
                continue
            metrics[op] = {
                'p50': lats[len(lats)//2],
                'p99': lats[int(len(lats)*0.99)],
                'avg': sum(lats)/len(lats),
                'throughput_gbps': (len(lats) * self.object_size * 8) / (sum(lats)/1000) / 1e9
            }
        return metrics

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Cloudflare R2 Benchmark')
    parser.add_argument('--bucket', required=True, help='R2 bucket name')
    parser.add_argument('--prefix', default='bench', help='Object key prefix')
    parser.add_argument('--account-id', required=True, help='Cloudflare account ID')
    parser.add_argument('--access-key', required=True, help='R2 access key ID')
    parser.add_argument('--secret-key', required=True, help='R2 secret access key')
    parser.add_argument('--num-objects', type=int, default=10000, help='Number of objects per benchmark')
    parser.add_argument('--concurrency', type=int, default=100, help='Concurrent requests')
    args = parser.parse_args()

    bench = R2Benchmark(
        args.bucket,
        args.prefix,
        args.account_id,
        args.access_key,
        args.secret_key
    )
    loop = asyncio.get_event_loop()
    loop.run_until_complete(bench.run_put_benchmark(args.num_objects, args.concurrency))
    loop.run_until_complete(bench.run_get_benchmark(args.num_objects, args.concurrency))
    metrics = bench.calculate_metrics()
    print('\nCloudflare R2 Benchmark Results:')
    for op, vals in metrics.items():
        print(f'{op.upper()}: p50={vals["p50"]:.2f}ms, p99={vals["p99"]:.2f}ms, throughput={vals["throughput_gbps"]:.2f}Gbps')

import argparse
from dataclasses import dataclass
from typing import Optional

# Cost comparison tool: AWS S3 Express One Zone vs Cloudflare R2
# Pricing as of June 2024:
# S3 Express One Zone (us-east-1):
#   Storage: $0.20/GB/month
#   Egress: $0.09/GB (first 10PB), $0.085/GB (next 40PB), $0.07/GB (over 50PB)
#   PUT: $0.0004 per 1k requests
#   GET: $0.0001 per 1k requests
#   Cross-AZ replication: $0.01/GB replicated
# Cloudflare R2 (global):
#   Storage: $0.015/GB/month
#   Egress: $0.00 (no egress fees)
#   PUT: $0.0004 per 1k requests (same as S3)
#   GET: $0.0001 per 1k requests (same as S3)
#   No replication fees (built-in global edge)

@dataclass
class Workload:
    storage_gb_month: float
    egress_gb_month: float
    put_requests: int  # per month
    get_requests: int  # per month
    cross_az_replication: bool = False  # Only for S3 Express

class S3ExpressCost:
    def __init__(self, workload: Workload):
        self.workload = workload
        self.storage_cost_per_gb = 0.20
        self.egress_tiers = [
            (10 * 1024 * 1024, 0.09),  # First 10PB (10*1024*1024 GB)
            (40 * 1024 * 1024, 0.085),  # Next 40PB
            (float('inf'), 0.07)  # Over 50PB
        ]
        self.put_cost_per_1k = 0.0004
        self.get_cost_per_1k = 0.0001
        self.cross_az_replication_cost_per_gb = 0.01

    def calculate_egress_cost(self) -> float:
        '''Calculate tiered egress cost for S3 Express'''
        remaining_egress = self.workload.egress_gb_month
        total_egress_cost = 0.0
        for tier_size, tier_rate in self.egress_tiers:
            if remaining_egress <= 0:
                break
            tier_usage = min(remaining_egress, tier_size)
            total_egress_cost += tier_usage * tier_rate
            remaining_egress -= tier_usage
        return total_egress_cost

    def calculate_total_cost(self) -> float:
        '''Calculate total monthly cost for S3 Express'''
        storage_cost = self.workload.storage_gb_month * self.storage_cost_per_gb
        egress_cost = self.calculate_egress_cost()
        put_cost = (self.workload.put_requests / 1000) * self.put_cost_per_1k
        get_cost = (self.workload.get_requests / 1000) * self.get_cost_per_1k
        cross_az_cost = 0.0
        if self.workload.cross_az_replication:
            cross_az_cost = self.workload.storage_gb_month * self.cross_az_replication_cost_per_gb
        return storage_cost + egress_cost + put_cost + get_cost + cross_az_cost

class R2Cost:
    def __init__(self, workload: Workload):
        self.workload = workload
        self.storage_cost_per_gb = 0.015
        self.egress_cost_per_gb = 0.00
        self.put_cost_per_1k = 0.0004
        self.get_cost_per_1k = 0.0001

    def calculate_total_cost(self) -> float:
        '''Calculate total monthly cost for R2'''
        storage_cost = self.workload.storage_gb_month * self.storage_cost_per_gb
        egress_cost = self.workload.egress_gb_month * self.egress_cost_per_gb
        put_cost = (self.workload.put_requests / 1000) * self.put_cost_per_1k
        get_cost = (self.workload.get_requests / 1000) * self.get_cost_per_1k
        return storage_cost + egress_cost + put_cost + get_cost

def print_comparison(s3_cost: float, r2_cost: float, workload: Workload) -> None:
    '''Print formatted cost comparison'''
    savings = s3_cost - r2_cost
    savings_pct = (savings / s3_cost) * 100 if s3_cost > 0 else 0
    print('\n' + '='*60)
    print('Cost Comparison: AWS S3 Express One Zone vs Cloudflare R2')
    print('='*60)
    print(f'Workload: {workload.storage_gb_month:.2f} GB storage, {workload.egress_gb_month:.2f} GB egress')
    print(f'Requests: {workload.put_requests:,} PUT, {workload.get_requests:,} GET')
    print('-'*60)
    print(f'AWS S3 Express One Zone Total: ${s3_cost:,.2f}/month')
    print(f'Cloudflare R2 Total:          ${r2_cost:,.2f}/month')
    print('-'*60)
    if savings > 0:
        print(f'R2 Savings: ${savings:,.2f}/month ({savings_pct:.1f}% less than S3)')
    else:
        print(f'S3 Express Savings: ${-savings:,.2f}/month (${-savings_pct:.1f}% less than R2)')
    print('='*60)

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='S3 Express vs R2 Cost Calculator')
    parser.add_argument('--storage-gb', type=float, required=True, help='Monthly storage in GB')
    parser.add_argument('--egress-gb', type=float, required=True, help='Monthly egress in GB')
    parser.add_argument('--puts', type=int, required=True, help='Monthly PUT requests')
    parser.add_argument('--gets', type=int, required=True, help='Monthly GET requests')
    parser.add_argument('--cross-az', action='store_true', help='Enable S3 Express cross-AZ replication')
    args = parser.parse_args()

    workload = Workload(
        storage_gb_month=args.storage_gb,
        egress_gb_month=args.egress_gb,
        put_requests=args.puts,
        get_requests=args.gets,
        cross_az_replication=args.cross_az
    )

    s3_calc = S3ExpressCost(workload)
    r2_calc = R2Cost(workload)

    s3_total = s3_calc.calculate_total_cost()
    r2_total = r2_calc.calculate_total_cost()

    print_comparison(s3_total, r2_total, workload)

import uuid
import boto3

def generate_s3_express_key(base_prefix: str, object_id: int) -> str:
    '''Generate a random 8-char hex prefix for S3 Express to avoid partition throttling'''
    random_prefix = uuid.uuid4().hex[:8]  # 8-char random hex
    return f'{base_prefix}/{random_prefix}/obj_{object_id}'

# Example usage with S3 Express client
s3 = boto3.client('s3', region_name='us-east-1')
key = generate_s3_express_key('video-segments', 12345)
s3.put_object(
    Bucket='streamingco-hot-segments',
    Key=key,
    Body=open('segment_12345.mp4', 'rb'),
    StorageClass='EXPRESS_ONEZONE'
)

import boto3
import datetime

def generate_r2_signed_url(bucket: str, key: str, account_id: str, access_key: str, secret_key: str) -> str:
    '''Generate a 15-minute signed URL for R2 edge caching'''
    s3 = boto3.client(
        's3',
        region_name='us-east-1',
        endpoint_url=f'https://{account_id}.r2.cloudflarestorage.com',
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key
    )
    # Generate signed URL with 15-minute expiry
    url = s3.generate_presigned_url(
        'get_object',
        Params={'Bucket': bucket, 'Key': key},
        ExpiresIn=900  # 15 minutes
    )
    return url

# Example usage
r2_url = generate_r2_signed_url(
    bucket='streamingco-cold-segments',
    key='2024/05/01/segment_12345.mp4',
    account_id='your-account-id',
    access_key='your-access-key',
    secret_key='your-secret-key'
)
print(f'Signed R2 URL: {r2_url}')

import (
    "context"
    "fmt"
    "time"
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/service/s3"
)

const (
    hotAgeThreshold = 7 * 24 * time.Hour  // 7 days
    s3ExpressBucket = "streamingco-hot-segments"
    r2Bucket        = "streamingco-cold-segments"
)

func routeGetRequest(ctx context.Context, objKey string, objAge time.Duration, s3Client *s3.Client) (string, error) {
    // Route objects younger than 7 days to S3 Express, older to R2
    if objAge < hotAgeThreshold {
        // Fetch from S3 Express One Zone
        _, err := s3Client.GetObject(ctx, &s3.GetObjectInput{
            Bucket: aws.String(s3ExpressBucket),
            Key:    aws.String(objKey),
        })
        if err != nil {
            return "", fmt.Errorf("S3 Express GET failed: %w", err)
        }
        return "s3-express", nil
    } else {
        // Fetch from R2 (pseudo-code, use R2 S3-compatible client)
        // r2Client.GetObject(ctx, &s3.GetObjectInput{Bucket: r2Bucket, Key: objKey})
        return "r2", nil
    }
}