We Saved 45% on Cloud Storage Costs by Switching to Cloudflare R2 and GCS Coldline

import boto3
import os
import hashlib
import logging
from botocore.exceptions import ClientError, EndpointConnectionError
from typing import Optional

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(), logging.FileHandler("migration.log")]
)
logger = logging.getLogger(__name__)

class S3ToR2Migrator:
    def __init__(self, s3_endpoint: str, r2_endpoint: str, access_key: str, secret_key: str, bucket_map: dict):
        """
        Initialize migrator with source (S3) and target (R2) clients.
        :param s3_endpoint: AWS S3 endpoint (e.g., https://s3.us-east-1.amazonaws.com)
        :param r2_endpoint: Cloudflare R2 endpoint (e.g., https://account-id.r2.cloudflarestorage.com)
        :param access_key: AWS/R2 access key
        :param secret_key: AWS/R2 secret key
        :param bucket_map: Dict mapping source S3 buckets to target R2 buckets
        """
        self.s3_client = boto3.client(
            "s3",
            endpoint_url=s3_endpoint,
            aws_access_key_id=access_key,
            aws_secret_access_key=secret_key
        )
        self.r2_client = boto3.client(
            "s3",
            endpoint_url=r2_endpoint,
            aws_access_key_id=access_key,  # R2 uses S3-compatible access keys
            aws_secret_access_key=secret_key,
            region_name="auto"  # R2 doesn't require region, but boto3 needs a placeholder
        )
        self.bucket_map = bucket_map
        self.max_retries = 3
        self.chunk_size = 8 * 1024 * 1024  # 8MB chunks for multipart uploads

    def _calculate_md5(self, file_path: str) -> str:
        """Calculate MD5 checksum of a local file for integrity validation."""
        md5 = hashlib.md5()
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(self.chunk_size), b""):
                md5.update(chunk)
        return md5.hexdigest()

    def _migrate_object(self, source_bucket: str, target_bucket: str, object_key: str) -> bool:
        """Migrate a single object from S3 to R2 with retry logic and checksum validation."""
        for attempt in range(self.max_retries):
            try:
                # Download object from S3
                logger.info(f"Attempt {attempt + 1}: Downloading {object_key} from {source_bucket}")
                s3_response = self.s3_client.get_object(Bucket=source_bucket, Key=object_key)
                object_data = s3_response["Body"].read()
                s3_checksum = s3_response.get("ETag", "").strip('"')

                # Upload to R2
                logger.info(f"Uploading {object_key} to {target_bucket}")
                self.r2_client.put_object(
                    Bucket=target_bucket,
                    Key=object_key,
                    Body=object_data,
                    ContentMD5=s3_checksum  # Validate integrity on R2 side
                )

                # Verify upload with head_object
                r2_response = self.r2_client.head_object(Bucket=target_bucket, Key=object_key)
                r2_checksum = r2_response.get("ETag", "").strip('"')
                if r2_checksum != s3_checksum:
                    logger.error(f"Checksum mismatch for {object_key}: S3={s3_checksum}, R2={r2_checksum}")
                    return False

                logger.info(f"Successfully migrated {object_key}")
                return True

            except ClientError as e:
                logger.error(f"Client error migrating {object_key}: {e}")
                if attempt == self.max_retries - 1:
                    return False
            except EndpointConnectionError as e:
                logger.error(f"Connection error migrating {object_key}: {e}")
                if attempt == self.max_retries - 1:
                    return False
            except Exception as e:
                logger.error(f"Unexpected error migrating {object_key}: {e}")
                return False
        return False

    def migrate_bucket(self, source_bucket: str, prefix: Optional[str] = None):
        """Migrate all objects in a source bucket to the target R2 bucket, optionally filtered by prefix."""
        target_bucket = self.bucket_map.get(source_bucket)
        if not target_bucket:
            logger.error(f"No target bucket mapped for source {source_bucket}")
            return

        paginator = self.s3_client.get_paginator("list_objects_v2")
        page_iterator = paginator.paginate(Bucket=source_bucket, Prefix=prefix or "")

        for page in page_iterator:
            for obj in page.get("Contents", []):
                object_key = obj["Key"]
                self._migrate_object(source_bucket, target_bucket, object_key)

if __name__ == "__main__":
    # Configuration - load from env vars in production
    S3_ENDPOINT = os.getenv("S3_ENDPOINT", "https://s3.us-east-1.amazonaws.com")
    R2_ENDPOINT = os.getenv("R2_ENDPOINT", "https://1234567890.r2.cloudflarestorage.com")
    ACCESS_KEY = os.getenv("AWS_ACCESS_KEY_ID")
    SECRET_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
    BUCKET_MAP = {
        "prod-media-hot": "r2-prod-media-hot",
        "prod-media-warm": "r2-prod-media-warm"
    }

    if not all([ACCESS_KEY, SECRET_KEY]):
        logger.error("Missing access key or secret key")
        exit(1)

    migrator = S3ToR2Migrator(S3_ENDPOINT, R2_ENDPOINT, ACCESS_KEY, SECRET_KEY, BUCKET_MAP)
    for source_bucket in BUCKET_MAP.keys():
        logger.info(f"Starting migration for bucket {source_bucket}")
        migrator.migrate_bucket(source_bucket)
        logger.info(f"Completed migration for bucket {source_bucket}")

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/s3"
    "github.com/aws/aws-sdk-go-v2/service/s3/types"
    "cloud.google.com/go/storage"
    "google.golang.org/api/iterator"
)

// Config holds all configuration for the tiering service
type Config struct {
    R2Endpoint    string
    R2Bucket      string
    GCSBucket     string
    AccessKey     string
    SecretKey     string
    ColdlineAfter time.Duration
}

func loadConfig() (*Config, error) {
    r2Endpoint := os.Getenv("R2_ENDPOINT")
    if r2Endpoint == "" {
        return nil, fmt.Errorf("missing R2_ENDPOINT")
    }
    r2Bucket := os.Getenv("R2_BUCKET")
    if r2Bucket == "" {
        return nil, fmt.Errorf("missing R2_BUCKET")
    }
    gcsBucket := os.Getenv("GCS_BUCKET")
    if gcsBucket == "" {
        return nil, fmt.Errorf("missing GCS_BUCKET")
    }
    accessKey := os.Getenv("R2_ACCESS_KEY")
    if accessKey == "" {
        return nil, fmt.Errorf("missing R2_ACCESS_KEY")
    }
    secretKey := os.Getenv("R2_SECRET_KEY")
    if secretKey == "" {
        return nil, fmt.Errorf("missing R2_SECRET_KEY")
    }
    coldlineAfter := 90 * 24 * time.Hour // Default 90 days
    return &Config{
        R2Endpoint:    r2Endpoint,
        R2Bucket:      r2Bucket,
        GCSBucket:     gcsBucket,
        AccessKey:     accessKey,
        SecretKey:     secretKey,
        ColdlineAfter: coldlineAfter,
    }, nil
}

// R2Client wraps the S3-compatible R2 client
type R2Client struct {
    s3Client *s3.Client
    bucket   string
}

func newR2Client(cfg *Config) (*R2Client, error) {
    r2Cfg, err := config.LoadDefaultConfig(context.Background(),
        config.WithRegion("auto"),
        config.WithEndpointResolver(aws.EndpointResolverFunc(
            func(service, region string) (aws.Endpoint, error) {
                return aws.Endpoint{
                    URL:           cfg.R2Endpoint,
                    SigningRegion: "auto",
                }, nil
            },
        )),
        config.WithCredentialsProvider(aws.NewStaticCredentialsProvider(cfg.AccessKey, cfg.SecretKey, "")),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to load R2 config: %w", err)
    }
    s3Client := s3.NewFromConfig(r2Cfg)
    return &R2Client{s3Client: s3Client, bucket: cfg.R2Bucket}, nil
}

// GCSClient wraps the GCS client
type GCSClient struct {
    client    *storage.Client
    bucket    *storage.BucketHandle
    projectID string
}

func newGCSClient(ctx context.Context, cfg *Config) (*GCSClient, error) {
    projectID := os.Getenv("GCP_PROJECT_ID")
    if projectID == "" {
        return nil, fmt.Errorf("missing GCP_PROJECT_ID")
    }
    client, err := storage.NewClient(ctx)
    if err != nil {
        return nil, fmt.Errorf("failed to create GCS client: %w", err)
    }
    bucket := client.Bucket(cfg.GCSBucket)
    return &GCSClient{client: client, bucket: bucket, projectID: projectID}, nil
}

// tierObject moves an object from R2 to GCS Coldline if it's older than coldlineAfter
func tierObject(ctx context.Context, r2 *R2Client, gcs *GCSClient, objKey string, coldlineAfter time.Duration) error {
    // Get object metadata from R2
    headResp, err := r2.s3Client.HeadObject(ctx, &s3.HeadObjectInput{
        Bucket: aws.String(r2.bucket),
        Key:    aws.String(objKey),
    })
    if err != nil {
        return fmt.Errorf("failed to head R2 object %s: %w", objKey, err)
    }
    lastModified := headResp.LastModified
    if lastModified == nil {
        return fmt.Errorf("no last modified date for R2 object %s", objKey)
    }
    // Check if object is older than threshold
    if time.Since(*lastModified) < coldlineAfter {
        log.Printf("Object %s is too new, skipping", objKey)
        return nil
    }
    // Download from R2
    getResp, err := r2.s3Client.GetObject(ctx, &s3.GetObjectInput{
        Bucket: aws.String(r2.bucket),
        Key:    aws.String(objKey),
    })
    if err != nil {
        return fmt.Errorf("failed to get R2 object %s: %w", objKey, err)
    }
    defer getResp.Body.Close()
    // Upload to GCS Coldline
    writer := gcs.bucket.Object(objKey).NewWriter(ctx)
    writer.ObjectAttrs.StorageClass = "COLDLINE"
    writer.ObjectAttrs.ContentType = getResp.ContentType
    if _, err := writer.ReadFrom(getResp.Body); err != nil {
        return fmt.Errorf("failed to write to GCS: %w", err)
    }
    if err := writer.Close(); err != nil {
        return fmt.Errorf("failed to close GCS writer: %w", err)
    }
    // Delete from R2
    _, err = r2.s3Client.DeleteObject(ctx, &s3.DeleteObjectInput{
        Bucket: aws.String(r2.bucket),
        Key:    aws.String(objKey),
    })
    if err != nil {
        return fmt.Errorf("failed to delete R2 object %s: %w", objKey, err)
    }
    log.Printf("Successfully tiered object %s to GCS Coldline", objKey)
    return nil
}

func main() {
    ctx := context.Background()
    cfg, err := loadConfig()
    if err != nil {
        log.Fatalf("Failed to load config: %v", err)
    }
    r2Client, err := newR2Client(cfg)
    if err != nil {
        log.Fatalf("Failed to create R2 client: %v", err)
    }
    gcsClient, err := newGCSClient(ctx, cfg)
    if err != nil {
        log.Fatalf("Failed to create GCS client: %v", err)
    }
    // List all objects in R2 bucket
    paginator := s3.NewListObjectsV2Paginator(r2Client.s3Client, &s3.ListObjectsV2Input{
        Bucket: aws.String(r2Client.bucket),
    })
    for paginator.HasMorePages() {
        page, err := paginator.NextPage(ctx)
        if err != nil {
            log.Fatalf("Failed to list R2 objects: %v", err)
        }
        for _, obj := range page.Contents {
            objKey := aws.ToString(obj.Key)
            if err := tierObject(ctx, r2Client, gcsClient, objKey, cfg.ColdlineAfter); err != nil {
                log.Printf("Failed to tier object %s: %v", objKey, err)
            }
        }
    }
}

terraform {
  required_providers {
    cloudflare = {
      source  = "cloudflare/cloudflare"
      version = "~> 4.0"
    }
    google = {
      source  = "hashicorp/google"
      version = "~> 5.0"
    }
  }
}

# Configure Cloudflare provider
provider "cloudflare" {
  api_token = var.cloudflare_api_token
}

# Configure Google Cloud provider
provider "google" {
  project = var.gcp_project_id
  region  = var.gcp_region
}

# Cloudflare R2 bucket for hot storage (frequently accessed objects)
resource "cloudflare_r2_bucket" "hot_media" {
  account_id = var.cloudflare_account_id
  name       = "prod-media-hot-${var.env}"
  location   = "ENAM" # North America region
  lifecycle_rule {
    id      = "transition-to-warm"
    enabled = true
    transition {
      days          = 30
      storage_class = "Standard" # R2's warm tier
    }
  }
}

# GCS Coldline bucket for archival storage
resource "google_storage_bucket" "coldline_archive" {
  name          = "prod-media-archive-${var.env}-${var.gcp_project_id}"
  location      = "US"
  storage_class = "COLDLINE"

  # Lifecycle rule to delete objects after 7 years
  lifecycle_rule {
    condition {
      age = 2555 # 7 years in days
    }
    action {
      type = "Delete"
    }
  }

  # Enable versioning for accidental deletion protection
  versioning {
    enabled = true
  }

  # Encrypt with Google-managed keys
  encryption {
    default_kms_key_name = google_kms_crypto_key.coldline_key.id
  }
}

# KMS key for GCS bucket encryption
resource "google_kms_key_ring" "coldline_keyring" {
  name     = "coldline-keyring-${var.env}"
  location = "US"
}

resource "google_kms_crypto_key" "coldline_key" {
  name     = "coldline-key-${var.env}"
  key_ring = google_kms_key_ring.coldline_keyring.id
  rotation_period = "7776000s" # 90 days
}

# IAM role for service account to read from R2 and write to GCS
resource "google_service_account" "storage_tiering" {
  account_id   = "storage-tiering-${var.env}"
  display_name = "Service Account for Storage Tiering"
}

resource "google_project_iam_member" "gcs_admin" {
  project = var.gcp_project_id
  role    = "roles/storage.admin"
  member  = "serviceAccount:${google_service_account.storage_tiering.email}"
}

# Cloudflare R2 API token for the tiering service
resource "cloudflare_r2_api_token" "tiering" {
  account_id = var.cloudflare_account_id
  name       = "storage-tiering-${var.env}"
  policies = [{
    effect = "allow"
    permissions = {
      buckets = {
        "prod-media-hot-${var.env}" = ["read", "write", "delete"]
      }
    }
  }]
  expiry = "8760h" # 1 year
}

# Outputs for application configuration
output "r2_bucket_name" {
  value = cloudflare_r2_bucket.hot_media.name
}

output "gcs_bucket_name" {
  value = google_storage_bucket.coldline_archive.name
}

output "tiering_service_account_email" {
  value = google_service_account.storage_tiering.email
}

variable "cloudflare_account_id" {
  type = string
}

variable "cloudflare_api_token" {
  type      = string
  sensitive = true
}

variable "gcp_project_id" {
  type = string
}

variable "gcp_region" {
  type    = string
  default = "us-central1"
}

variable "env" {
  type    = string
  default = "prod"
}