How We Migrated 10PB of Data from AWS S3 to Cloudflare R2 for 2026 Media Streaming

// Copyright 2026 StreamFlow Inc. All rights reserved.
// migrate-orchestrator is the core service coordinating 10PB S3 to R2 migration
package main

import (
    "context"
    "crypto/md5"
    "encoding/hex"
    "fmt"
    "log"
    "os"
    "sync"
    "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/cloudflare/cloudflare-go/v0.9.4/r2"
)

const (
    // Max concurrent workers per node to avoid rate limiting
    maxWorkers = 16
    // Chunk size for multipart uploads to R2 (100MB matches R2 recommended max)
    chunkSize = 100 * 1024 * 1024
    // Retry limit for failed transfers
    maxRetries = 3
)

// transferTask holds metadata for a single object transfer
type transferTask struct {
    Bucket    string
    Key       string
    Size      int64
    ETag      string
    Checksum  string
}

// orchestrator manages the migration workflow
type orchestrator struct {
    s3Client  *s3.Client
    r2Client  *r2.Client
    taskQueue chan transferTask
    wg        sync.WaitGroup
}

// newOrchestrator initializes AWS and R2 clients with retries
func newOrchestrator(ctx context.Context) (*orchestrator, error) {
    // Load AWS config with region fallback
    awsCfg, err := config.LoadDefaultConfig(ctx,
        config.WithRegion("us-east-1"),
        config.WithRetryMaxAttempts(maxRetries),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to load AWS config: %w", err)
    }

    // Initialize S3 client with request metrics
    s3Client := s3.NewFromConfig(awsCfg, func(o *s3.Options) {
        o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
    })

    // Load R2 credentials from environment
    r2AcctID := os.Getenv("CLOUDFLARE_ACCOUNT_ID")
    r2KeyID := os.Getenv("R2_ACCESS_KEY_ID")
    r2Secret := os.Getenv("R2_SECRET_ACCESS_KEY")
    if r2AcctID == "" || r2KeyID == "" || r2Secret == "" {
        return nil, fmt.Errorf("missing R2 credentials in environment")
    }

    // Initialize R2 client with Cloudflare Go SDK
    r2Client, err := r2.NewClient(r2AcctID, r2KeyID, r2Secret)
    if err != nil {
        return nil, fmt.Errorf("failed to initialize R2 client: %w", err)
    }

    return &orchestrator{
        s3Client:  s3Client,
        r2Client:  r2Client,
        taskQueue: make(chan transferTask, 1024),
    }, nil
}

// startWorkers launches concurrent transfer workers
func (o *orchestrator) startWorkers(ctx context.Context) {
    for i := 0; i < maxWorkers; i++ {
        o.wg.Add(1)
        go func(workerID int) {
            defer o.wg.Done()
            log.Printf("worker %d started", workerID)
            for task := range o.taskQueue {
                o.processTask(ctx, task, workerID)
            }
        }(i)
    }
}

// processTask handles single object transfer with retry logic
func (o *orchestrator) processTask(ctx context.Context, task transferTask, workerID int) {
    var err error
    for attempt := 1; attempt <= maxRetries; attempt++ {
        err = o.transferObject(ctx, task)
        if err == nil {
            log.Printf("worker %d: transferred %s (attempt %d)", workerID, task.Key, attempt)
            return
        }
        log.Printf("worker %d: transfer failed for %s (attempt %d/%d): %v",
            workerID, task.Key, attempt, maxRetries, err)
        time.Sleep(time.Duration(attempt*2) * time.Second)
    }
    log.Printf("worker %d: failed to transfer %s after %d attempts", workerID, task.Key, maxRetries)
}

// transferObject copies a single object from S3 to R2 with checksum validation
func (o *orchestrator) transferObject(ctx context.Context, task transferTask) error {
    // Fetch object from S3
    getResp, err := o.s3Client.GetObject(ctx, &s3.GetObjectInput{
        Bucket: aws.String(task.Bucket),
        Key:    aws.String(task.Key),
    })
    if err != nil {
        return fmt.Errorf("s3 get failed: %w", err)
    }
    defer getResp.Body.Close()

    // Calculate MD5 checksum of S3 object for validation
    hash := md5.New()
    if _, err := io.Copy(hash, getResp.Body); err != nil {
        return fmt.Errorf("checksum calculation failed: %w", err)
    }
    localChecksum := hex.EncodeToString(hash.Sum(nil))

    // Validate checksum matches S3 ETag (assuming ETag is MD5 for non-multipart)
    if localChecksum != task.Checksum {
        return fmt.Errorf("checksum mismatch: s3=%s local=%s", task.Checksum, localChecksum)
    }

    // Reset body reader (re-fetch since we consumed it for checksum)
    getResp, err = o.s3Client.GetObject(ctx, &s3.GetObjectInput{
        Bucket: aws.String(task.Bucket),
        Key:    aws.String(task.Key),
    })
    if err != nil {
        return fmt.Errorf("s3 re-get failed: %w", err)
    }
    defer getResp.Body.Close()

    // Upload to R2 with multipart if object is larger than chunk size
    if task.Size > chunkSize {
        return o.multipartUpload(ctx, task, getResp.Body)
    }
    return o.singleUpload(ctx, task, getResp.Body)
}

// singleUpload handles small objects (<100MB) in one request
func (o *orchestrator) singleUpload(ctx context.Context, task transferTask, body io.Reader) error {
    _, err := o.r2Client.PutObject(ctx, &r2.PutObjectInput{
        Bucket: aws.String("streamflow-media-2026"),
        Key:    aws.String(task.Key),
        Body:   body,
        Metadata: map[string]string{
            "migrated-at": time.Now().UTC().Format(time.RFC3339),
            "s3-etag":     task.ETag,
        },
    })
    if err != nil {
        return fmt.Errorf("r2 single upload failed: %w", err)
    }
    return nil
}

// multipartUpload handles large objects with R2 multipart API
func (o *orchestrator) multipartUpload(ctx context.Context, task transferTask, body io.Reader) error {
    // Initiate multipart upload
    initResp, err := o.r2Client.CreateMultipartUpload(ctx, &r2.CreateMultipartUploadInput{
        Bucket: aws.String("streamflow-media-2026"),
        Key:    aws.String(task.Key),
    })
    if err != nil {
        return fmt.Errorf("r2 multipart init failed: %w", err)
    }
    uploadID := initResp.UploadID

    // Upload parts with retry
    var parts []r2.CompletedPart
    partNum := 1
    buf := make([]byte, chunkSize)
    for {
        n, err := body.Read(buf)
        if n == 0 && err == io.EOF {
            break
        }
        if err != nil && err != io.EOF {
            o.abortMultipart(ctx, task.Key, uploadID)
            return fmt.Errorf("read chunk failed: %w", err)
        }

        // Upload part
        partResp, err := o.r2Client.UploadPart(ctx, &r2.UploadPartInput{
            Bucket:   aws.String("streamflow-media-2026"),
            Key:      aws.String(task.Key),
            UploadID: aws.String(uploadID),
            PartNumber: aws.Int32(int32(partNum)),
            Body:       bytes.NewReader(buf[:n]),
        })
        if err != nil {
            o.abortMultipart(ctx, task.Key, uploadID)
            return fmt.Errorf("upload part %d failed: %w", partNum, err)
        }

        parts = append(parts, r2.CompletedPart{
            ETag:       partResp.ETag,
            PartNumber: aws.Int32(int32(partNum)),
        })
        partNum++
    }

    // Complete multipart upload
    _, err = o.r2Client.CompleteMultipartUpload(ctx, &r2.CompleteMultipartUploadInput{
        Bucket:   aws.String("streamflow-media-2026"),
        Key:      aws.String(task.Key),
        UploadID: aws.String(uploadID),
        MultipartUpload: &r2.CompletedMultipartUpload{
            Parts: parts,
        },
    })
    if err != nil {
        o.abortMultipart(ctx, task.Key, uploadID)
        return fmt.Errorf("complete multipart failed: %w", err)
    }
    return nil
}

// abortMultipart cleans up failed multipart uploads
func (o *orchestrator) abortMultipart(ctx context.Context, key, uploadID string) {
    _, err := o.r2Client.AbortMultipartUpload(ctx, &r2.AbortMultipartUploadInput{
        Bucket:   aws.String("streamflow-media-2026"),
        Key:      aws.String(key),
        UploadID: aws.String(uploadID),
    })
    if err != nil {
        log.Printf("failed to abort multipart upload %s for %s: %v", uploadID, key, err)
    }
}

func main() {
    ctx := context.Background()
    orch, err := newOrchestrator(ctx)
    if err != nil {
        log.Fatalf("failed to initialize orchestrator: %v", err)
    }
    defer close(orch.taskQueue)

    // Start workers
    orch.startWorkers(ctx)

    // TODO: Add task queue population logic from S3 inventory
    log.Println("orchestrator started, waiting for tasks...")
    orch.wg.Wait()
}

// validate-integrity checks migrated objects for checksum and metadata consistency
package main

import (
    "context"
    "crypto/md5"
    "encoding/hex"
    "fmt"
    "log"
    "os"
    "sync"

    "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/cloudflare/cloudflare-go/v0.9.4/r2"
    "github.com/joho/godotenv"
)

const (
    // Batch size for listing objects to validate
    listBatchSize = 1000
    // Max concurrent validation workers
    validateWorkers = 32
)

// validationResult holds outcome of a single object check
type validationResult struct {
    Key        string
    Valid      bool
    Error      string
    S3Checksum string
    R2Checksum string
}

// validator coordinates integrity checks between S3 and R2
type validator struct {
    s3Client *s3.Client
    r2Client *r2.Client
    s3Bucket string
    r2Bucket string
    resultCh chan validationResult
    wg       sync.WaitGroup
}

// newValidator initializes clients for both storage providers
func newValidator(ctx context.Context) (*validator, error) {
    // Load .env for credentials (production uses IAM roles)
    if err := godotenv.Load(); err != nil {
        log.Printf("warning: no .env file found, using environment credentials")
    }

    // AWS S3 client
    awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion("us-east-1"))
    if err != nil {
        return nil, fmt.Errorf("aws config load failed: %w", err)
    }
    s3Client := s3.NewFromConfig(awsCfg)

    // R2 client
    r2AcctID := os.Getenv("CLOUDFLARE_ACCOUNT_ID")
    r2KeyID := os.Getenv("R2_ACCESS_KEY_ID")
    r2Secret := os.Getenv("R2_SECRET_ACCESS_KEY")
    r2Client, err := r2.NewClient(r2AcctID, r2KeyID, r2Secret)
    if err != nil {
        return nil, fmt.Errorf("r2 client init failed: %w", err)
    }

    return &validator{
        s3Client:  s3Client,
        r2Client:  r2Client,
        s3Bucket:  "streamflow-s3-media-legacy",
        r2Bucket:  "streamflow-media-2026",
        resultCh:  make(chan validationResult, 2048),
    }, nil
}

// startValidationWorkers launches concurrent validation goroutines
func (v *validator) startValidationWorkers(ctx context.Context) {
    for i := 0; i < validateWorkers; i++ {
        v.wg.Add(1)
        go func(workerID int) {
            defer v.wg.Done()
            log.Printf("validation worker %d started", workerID)
            for res := range v.resultCh {
                v.processValidation(ctx, res, workerID)
            }
        }(i)
    }
}

// processValidation compares S3 and R2 object checksums
func (v *validator) processValidation(ctx context.Context, res validationResult, workerID int) {
    // Fetch S3 object head for checksum
    s3Head, err := v.s3Client.HeadObject(ctx, &s3.HeadObjectInput{
        Bucket: aws.String(v.s3Bucket),
        Key:    aws.String(res.Key),
    })
    if err != nil {
        log.Printf("worker %d: s3 head failed for %s: %v", workerID, res.Key, err)
        return
    }
    s3Checksum := aws.ToString(s3Head.ETag)
    // Strip quotes from ETag if present
    if len(s3Checksum) >= 2 && s3Checksum[0] == '"' && s3Checksum[len(s3Checksum)-1] == '"' {
        s3Checksum = s3Checksum[1 : len(s3Checksum)-1]
    }

    // Fetch R2 object head for checksum
    r2Head, err := v.r2Client.HeadObject(ctx, &r2.HeadObjectInput{
        Bucket: aws.String(v.r2Bucket),
        Key:    aws.String(res.Key),
    })
    if err != nil {
        log.Printf("worker %d: r2 head failed for %s: %v", workerID, res.Key, err)
        return
    }
    r2Checksum := r2Head.ETag
    if len(r2Checksum) >= 2 && r2Checksum[0] == '"' && r2Checksum[len(r2Checksum)-1] == '"' {
        r2Checksum = r2Checksum[1 : len(r2Checksum)-1]
    }

    // Compare checksums
    if s3Checksum != r2Checksum {
        log.Printf("worker %d: CHECKSUM MISMATCH for %s: s3=%s r2=%s",
            workerID, res.Key, s3Checksum, r2Checksum)
        // Write mismatch to audit log
        auditLog := fmt.Sprintf("mismatch,%s,%s,%s\n", res.Key, s3Checksum, r2Checksum)
        if err := os.WriteFile("mismatch_audit.log", []byte(auditLog), 0644); err != nil {
            log.Printf("failed to write audit log: %v", err)
        }
    } else {
        log.Printf("worker %d: validated %s (checksum match)", workerID, res.Key)
    }
}

// listAndQueueObjects lists all S3 objects and queues them for validation
func (v *validator) listAndQueueObjects(ctx context.Context) error {
    var continuationToken *string
    totalObjects := 0

    for {
        listResp, err := v.s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
            Bucket:            aws.String(v.s3Bucket),
            MaxKeys:           aws.Int32(listBatchSize),
            ContinuationToken: continuationToken,
        })
        if err != nil {
            return fmt.Errorf("s3 list failed: %w", err)
        }

        // Queue each object for validation
        for _, obj := range listResp.Contents {
            v.resultCh <- validationResult{
                Key: aws.ToString(obj.Key),
            }
            totalObjects++
        }

        // Check if we have more objects to list
        if !aws.ToBool(listResp.IsTruncated) {
            break
        }
        continuationToken = listResp.NextContinuationToken
    }

    log.Printf("queued %d objects for validation", totalObjects)
    return nil
}

func main() {
    ctx := context.Background()
    v, err := newValidator(ctx)
    if err != nil {
        log.Fatalf("validator init failed: %v", err)
    }
    defer close(v.resultCh)

    // Start workers
    v.startValidationWorkers(ctx)

    // List and queue all objects
    if err := v.listAndQueueObjects(ctx); err != nil {
        log.Fatalf("object listing failed: %v", err)
    }

    // Wait for all validations to complete
    v.wg.Wait()
    log.Println("validation complete, check mismatch_audit.log for errors")
}

// traffic-shift gradually moves read traffic from S3 to R2 with automatic rollback
package main

import (
    "context"
    "fmt"
    "io"
    "log"
    "math/rand"
    "net/http"
    "os"
    "sync"
    "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/cloudflare/cloudflare-go/v0.9.4/r2"
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
)

const (
    // Initial traffic percentage to R2 (starts at 1% for safety)
    initialR2Percent = 1
    // Step size to increase R2 traffic every 10 minutes
    trafficStep = 5
    // Max consecutive errors before rollback
    maxErrors = 10
    // Metrics port
    metricsPort = ":9090"
)

// trafficController manages read traffic splitting between S3 and R2
type trafficController struct {
    s3Client     *s3.Client
    r2Client     *r2.Client
    s3Bucket     string
    r2Bucket     string
    r2Percent    int
    mu           sync.RWMutex
    errorCount   int
    totalRequests prometheus.Counter
    r2Requests   prometheus.Counter
    errorTotal   prometheus.Counter
}

// newTrafficController initializes the controller with metrics
func newTrafficController(ctx context.Context) (*trafficController, error) {
    // Initialize Prometheus metrics
    totalRequests := prometheus.NewCounter(prometheus.CounterOpts{
        Name: "media_requests_total",
        Help: "Total media read requests",
    })
    r2Requests := prometheus.NewCounter(prometheus.CounterOpts{
        Name: "media_requests_r2_total",
        Help: "Total media read requests served by R2",
    })
    errorTotal := prometheus.NewCounter(prometheus.CounterOpts{
        Name: "media_requests_error_total",
        Help: "Total media read request errors",
    })
    prometheus.MustRegister(totalRequests, r2Requests, errorTotal)

    // AWS S3 client
    awsCfg, err := config.LoadDefaultConfig(ctx, config.WithRegion("us-east-1"))
    if err != nil {
        return nil, fmt.Errorf("aws config failed: %w", err)
    }
    s3Client := s3.NewFromConfig(awsCfg)

    // R2 client
    r2AcctID := os.Getenv("CLOUDFLARE_ACCOUNT_ID")
    r2KeyID := os.Getenv("R2_ACCESS_KEY_ID")
    r2Secret := os.Getenv("R2_SECRET_ACCESS_KEY")
    r2Client, err := r2.NewClient(r2AcctID, r2KeyID, r2Secret)
    if err != nil {
        return nil, fmt.Errorf("r2 client failed: %w", err)
    }

    return &trafficController{
        s3Client:     s3Client,
        r2Client:     r2Client,
        s3Bucket:     "streamflow-s3-media-legacy",
        r2Bucket:     "streamflow-media-2026",
        r2Percent:    initialR2Percent,
        totalRequests: totalRequests,
        r2Requests:   r2Requests,
        errorTotal:   errorTotal,
    }, nil
}

// getObject routes read requests to S3 or R2 based on current traffic split
func (tc *trafficController) getObject(ctx context.Context, key string) (io.ReadCloser, error) {
    tc.totalRequests.Inc()
    tc.mu.RLock()
    currentPercent := tc.r2Percent
    tc.mu.RUnlock()

    // Route to R2 if random number is within current percentage
    useR2 := rand.Intn(100) < currentPercent
    if useR2 {
        tc.r2Requests.Inc()
        resp, err := tc.r2Client.GetObject(ctx, &r2.GetObjectInput{
            Bucket: aws.String(tc.r2Bucket),
            Key:    aws.String(key),
        })
        if err != nil {
            tc.recordError()
            // Fallback to S3 on R2 error
            return tc.s3Client.GetObject(ctx, &s3.GetObjectInput{
                Bucket: aws.String(tc.s3Bucket),
                Key:    aws.String(key),
            })
        }
        return resp.Body, nil
    }

    // Route to S3
    resp, err := tc.s3Client.GetObject(ctx, &s3.GetObjectInput{
        Bucket: aws.String(tc.s3Bucket),
        Key:    aws.String(key),
    })
    if err != nil {
        tc.recordError()
        return nil, fmt.Errorf("s3 get failed: %w", err)
    }
    return resp.Body, nil
}

// recordError increments error count and triggers rollback if threshold exceeded
func (tc *trafficController) recordError() {
    tc.errorTotal.Inc()
    tc.mu.Lock()
    defer tc.mu.Unlock()
    tc.errorCount++
    if tc.errorCount >= maxErrors {
        log.Printf("ERROR threshold exceeded (%d errors), rolling back to 1% R2 traffic", tc.errorCount)
        tc.r2Percent = initialR2Percent
        tc.errorCount = 0
    }
}

// adjustTrafficPeriodically increases R2 traffic gradually
func (tc *trafficController) adjustTrafficPeriodically(ctx context.Context) {
    ticker := time.NewTicker(10 * time.Minute)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            tc.mu.Lock()
            if tc.r2Percent+trafficStep <= 100 {
                tc.r2Percent += trafficStep
                log.Printf("increased R2 traffic to %d%%", tc.r2Percent)
            } else {
                tc.r2Percent = 100
                log.Printf("all traffic now served by R2")
            }
            // Reset error count on traffic adjustment
            tc.errorCount = 0
            tc.mu.Unlock()
        case <-ctx.Done():
            return
        }
    }
}

func main() {
    ctx := context.Background()
    rand.Seed(time.Now().UnixNano())

    tc, err := newTrafficController(ctx)
    if err != nil {
        log.Fatalf("controller init failed: %v", err)
    }

    // Start traffic adjustment goroutine
    go tc.adjustTrafficPeriodically(ctx)

    // Start metrics server
    http.Handle("/metrics", promhttp.Handler())
    go func() {
        log.Printf("metrics server listening on %s", metricsPort)
        if err := http.ListenAndServe(metricsPort, nil); err != nil {
            log.Fatalf("metrics server failed: %v", err)
        }
    }()

    // Example: handle media request (simplified)
    http.HandleFunc("/media/", func(w http.ResponseWriter, r *http.Request) {
        key := r.URL.Path[len("/media/"):]
        body, err := tc.getObject(ctx, key)
        if err != nil {
            http.Error(w, "failed to fetch media", http.StatusInternalServerError)
            return
        }
        defer body.Close()
        w.Header().Set("Content-Type", "video/mp4")
        io.Copy(w, body)
    })

    log.Println("traffic shift controller started, R2 traffic at 1%")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

import csv
import boto3
import gzip

s3 = boto3.client('s3')
inventory_bucket = 'streamflow-s3-inventory'
inventory_key = '2026-03-01/streamflow-s3-media-legacy.csv.gz'

# Download and parse gzipped inventory CSV
response = s3.get_object(Bucket=inventory_bucket, Key=inventory_key)
with gzip.open(response['Body'], 'rt') as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"Key: {row['key']}, Size: {row['size']}, ETag: {row['etag']}")

import (
    "context"
    "time"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/cloudflare/cloudflare-go/v0.9.4/r2"
)

func initMultipartUpload(ctx context.Context, r2Client *r2.Client, key string) (string, error) {
    resp, err := r2Client.CreateMultipartUpload(ctx, &r2.CreateMultipartUploadInput{
        Bucket: aws.String("streamflow-media-2026"),
        Key:    aws.String(key),
        Metadata: map[string]string{
            "migrated-at": time.Now().UTC().Format(time.RFC3339),
        },
    })
    if err != nil {
        return "", fmt.Errorf("multipart init failed: %w", err)
    }
    return aws.ToString(resp.UploadID), nil
}