s3-bulk-acl.go

package main

import (
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
)

const (
	WORKERS    = 100
	AWSPROFILE = "test-account"
	KEYSFILE   = "keys.txt"
)

var (
	AWSREGION = "eu-central-1"
	S3BUCKET  = "prod-videos-bucket"
	S3ACL     = "bucket-owner-full-control"
	FINISHED  = 0
)

func getSvc() (*s3.S3, error) {
	credsFile := os.Getenv("HOME") + "/.aws/credentials"
	creds := credentials.NewSharedCredentials(credsFile, AWSPROFILE)
	sess, err := session.NewSession(&aws.Config{
		Region:      &AWSREGION,
		Credentials: creds,
	})
	if err != nil {
		return &s3.S3{}, err
	}
	return s3.New(sess), nil
}

func worker(id int, wg *sync.WaitGroup, keys <-chan string) {
	svc, err := getSvc()
	if err != nil {
		panic(err)
	}

	for k := range keys {
		_, err := svc.PutObjectAcl(&s3.PutObjectAclInput{
			ACL:    &S3ACL,
			Bucket: &S3BUCKET,
			Key:    &k,
		})
		if err != nil {
			fmt.Fprintln(os.Stderr, err)
			fmt.Println(k)
			continue
		}
		FINISHED++
	}
	wg.Done()
}

func main() {
	bs, err := ioutil.ReadFile(KEYSFILE)
	if err != nil {
		log.Fatal(err)
	}

	keys := strings.Split(string(bs), "\n")

	keyChan := make(chan string, 100)

	var wg sync.WaitGroup
	wg.Add(WORKERS)
	for i := 0; i < WORKERS; i++ {
		go worker(i, &wg, keyChan)
	}
	fmt.Fprintf(os.Stderr, "started %d workers\n", WORKERS)

	go func() {
		for _, key := range keys {
			if key == "" {
				continue
			}
			keyChan <- key
		}
		close(keyChan)
		fmt.Fprintf(os.Stderr, "added %d keys to the channel\n", len(keys))
	}()

	go func() {
		for {
			fmt.Fprintln(os.Stderr, FINISHED)
			time.Sleep(time.Second * 1)
		}
	}()

	wg.Wait()
	fmt.Fprintf(os.Stderr, "processed %d keys\n", FINISHED)
}