#!/usr/bin/env python

import argparse
import boto
from boto.utils import parse_ts
from datetime import datetime
import logging
import os
import sys
import re


log = logging.getLogger('tablesnap-sunset')
stderr = logging.StreamHandler()
stderr.setFormatter(logging.Formatter(
    '%(name)s [%(asctime)s] %(levelname)s %(message)s'))
log.addHandler(stderr)
if os.environ.get('TDEBUG', False):
    log.setLevel(logging.DEBUG)
else:
    log.setLevel(logging.INFO)

# S3 limit for single file upload
s3_limit = 5 * 2**30

# Max file size to upload without doing multipart in MB
max_file_size = 5120

# Default chunk size for multipart uploads in B
chunk_size = 1000000000

class SunsetHandler(object):

    def __init__(self, args):
        self.key = args.aws_key
        self.secret = args.aws_secret
        self.token = args.aws_token
        self.prefix = args.prefix
        self.bucket_name = args.bucket
        self.sunset_bucket_name = args.sunset_bucket
        self.dry_run = args.dry_run

        self.max_size = max_file_size * 2**20
        self.chunk_size = chunk_size

        self.get_buckets()


    def get_buckets(self):
        log.info('Connecting to s3')
        conn = boto.connect_s3(aws_access_key_id=self.key,
                               aws_secret_access_key=self.secret,
                               security_token=self.token)
        log.info('Connected to s3')
        self.bucket = conn.get_bucket(self.bucket_name)
        log.info('Have backup bucket')
        self.sunset_bucket = conn.get_bucket(self.sunset_bucket_name)
        log.info('Have sunset bucket')


    def run(self):
        count = 0

        for key in self.bucket.list(prefix=self.prefix):
            count += 1
            if (count % 1000 == 0):
                log.info('Iterated over %s keys' % (count))
            filename = key.name.split(':')[1]

            if re.search('/system/compactions_in_progress/', filename):
                # We don't need these files in S3, delete them
                log.info('Removing key %s from active bucket' % (key.name))
                if not self.dry_run:
                    key.delete()
                continue

            if not os.path.isfile(filename):
                # check if timestamp too new, might still be uploading
                # https://forums.aws.amazon.com/thread.jspa?threadID=21634&tstart=0
                delta = datetime.utcnow() - parse_ts(key.last_modified)
                if delta.days <= 0 and delta.seconds < 3600:
                    log.info('key is too new, keeping it: %s: %s' % (key.name, delta))
                    continue

                # check for file in sunset bucket already
                sunset_key = self.sunset_bucket.get_key(key.name)

                if key.storage_class == 'GLACIER':
                    log.critical('Glacier file in manifest, unexpected, quiting')
                    sys.exit(1)
                    return
                elif sunset_key == None:
                    log.info('Moving key %s to sunset bucket' % (key.name))
                    if not self.dry_run:
                        self.copy(key)
                elif key.size != sunset_key.size:
                    log.info('Key size mismatch (%d != %d), moving key %s to sunset bucket' % (key.size, sunset_key.size, key.name))
                    if not self.dry_run:
                        self.copy(key)
                else:
                    log.info('Key %s already present in sunset bucket. copying over to update timestamp' % (key.name))
                    if not self.dry_run:
                        try:
                            self.copy(key)
                        except Exception as e:
                            log.error('key failed to copy, maybe because its already been moved. Skip for now: %s: %s' % (key.name, e))
                            continue

                log.info('Removing key %s from active bucket' % (key.name))
                if not self.dry_run:
                    key.delete()
            else:
                log.info('Keeping key %s' % (key.name))

    def copy(self, key):
        log.debug('File size check: %s > %s ? : %s' %
            (key.size, self.max_size,
            (key.size > self.max_size),))
        if key.size > self.max_size:
            log.info('Performing multipart copy for %s' %
                         (key.name))
            # refetch key to make sure we have metadata
            key = self.bucket.get_key(key.name)
            # work around bug https://github.com/boto/boto/issues/2536
            metadata = {}
            for k, v in list(key.metadata.items()):
                metadata[k] = v.encode('utf8')

            mp = self.sunset_bucket.initiate_multipart_upload(key.name,
                metadata=metadata)
            part = 1
            chunk = None
            try:
                for chunk_start in range(0, key.size, self.chunk_size):
                    chunk_end = chunk_start + self.chunk_size -1
                    if chunk_end >= key.size:
                        chunk_end = key.size-1
                    chunk_size = chunk_end - chunk_start + 1
                    log.debug('Uploading part #%d '
                                   '(size: %d)' %
                                   (part, chunk_size,))
                    mp.copy_part_from_key(self.bucket_name, key.name, part, chunk_start, chunk_end)
                    part += 1
                part -= 1
            except Exception as e:
                log.debug(e)
                log.info('Error uploading part %d' % (part,))
                mp.cancel_upload()
                raise
            log.debug('Uploaded %d parts, '
                           'completing upload' % (part,))
            mp.complete_upload()
        else:
            log.debug('Performing monolithic copy')
            self.sunset_bucket.copy_key(key.name, self.bucket_name, key.name)

def main():
    parser = argparse.ArgumentParser(description='tablesnap_sunset is a script that '
        'iterates over all files in an s3 bucket and moves files to cheaper longer term '
        'storage options given some meta-data conditions have been met.'
        'e.g. File is no longer being used by C* on local file system, File was uploaded '
        'more than X days ago, etc.')
    parser.add_argument('-k', '--aws-key',
        default=os.environ.get('AWS_ACCESS_KEY_ID'),
        help='Amazon S3 Key (default from AWS_ACCESS_KEY_ID in environment)')
    parser.add_argument('-s', '--aws-secret',
        default=os.environ.get('AWS_SECRET_ACCESS_KEY'),
        help='Amazon S3 Secret (default from AWS_SECRET_ACCESS_KEY in environment)')
    parser.add_argument('--aws-token',
        default=os.environ.get('AWS_SECURITY_TOKEN'),
        help='Amazon S3 Token (default from AWS_SECURITY_TOKEN in environment)')
    parser.add_argument('--dry-run', action='store_true',
        help='output which files would be moved')

    parser.add_argument('prefix', help='prefix to key path within bucket, e.g. `production/prodcassandra01:')
    parser.add_argument('bucket', help='S3 bucket containing backup sstables')
    parser.add_argument('sunset_bucket',
        help='S3 bucket where old, unused files are taken before they die')

    args = parser.parse_args()

    sh = SunsetHandler(args)
    sh.run()

if __name__ == '__main__':
    sys.exit(main())

c3733ab9ca7ce9d19e8210c6691e691bda3fc86f
<mask token>


class SunsetHandler(object):

    def __init__(self, args):
        self.key = args.aws_key
        self.secret = args.aws_secret
        self.token = args.aws_token
        self.prefix = args.prefix
        self.bucket_name = args.bucket
        self.sunset_bucket_name = args.sunset_bucket
        self.dry_run = args.dry_run
        self.max_size = max_file_size * 2 ** 20
        self.chunk_size = chunk_size
        self.get_buckets()

    def get_buckets(self):
        log.info('Connecting to s3')
        conn = boto.connect_s3(aws_access_key_id=self.key,
            aws_secret_access_key=self.secret, security_token=self.token)
        log.info('Connected to s3')
        self.bucket = conn.get_bucket(self.bucket_name)
        log.info('Have backup bucket')
        self.sunset_bucket = conn.get_bucket(self.sunset_bucket_name)
        log.info('Have sunset bucket')

    def run(self):
        count = 0
        for key in self.bucket.list(prefix=self.prefix):
            count += 1
            if count % 1000 == 0:
                log.info('Iterated over %s keys' % count)
            filename = key.name.split(':')[1]
            if re.search('/system/compactions_in_progress/', filename):
                log.info('Removing key %s from active bucket' % key.name)
                if not self.dry_run:
                    key.delete()
                continue
            if not os.path.isfile(filename):
                delta = datetime.utcnow() - parse_ts(key.last_modified)
                if delta.days <= 0 and delta.seconds < 3600:
                    log.info('key is too new, keeping it: %s: %s' % (key.
                        name, delta))
                    continue
                sunset_key = self.sunset_bucket.get_key(key.name)
                if key.storage_class == 'GLACIER':
                    log.critical(
                        'Glacier file in manifest, unexpected, quiting')
                    sys.exit(1)
                    return
                elif sunset_key == None:
                    log.info('Moving key %s to sunset bucket' % key.name)
                    if not self.dry_run:
                        self.copy(key)
                elif key.size != sunset_key.size:
                    log.info(
                        'Key size mismatch (%d != %d), moving key %s to sunset bucket'
                         % (key.size, sunset_key.size, key.name))
                    if not self.dry_run:
                        self.copy(key)
                else:
                    log.info(
                        'Key %s already present in sunset bucket. copying over to update timestamp'
                         % key.name)
                    if not self.dry_run:
                        try:
                            self.copy(key)
                        except Exception as e:
                            log.error(
                                'key failed to copy, maybe because its already been moved. Skip for now: %s: %s'
                                 % (key.name, e))
                            continue
                log.info('Removing key %s from active bucket' % key.name)
                if not self.dry_run:
                    key.delete()
            else:
                log.info('Keeping key %s' % key.name)

    def copy(self, key):
        log.debug('File size check: %s > %s ? : %s' % (key.size, self.
            max_size, key.size > self.max_size))
        if key.size > self.max_size:
            log.info('Performing multipart copy for %s' % key.name)
            key = self.bucket.get_key(key.name)
            metadata = {}
            for k, v in list(key.metadata.items()):
                metadata[k] = v.encode('utf8')
            mp = self.sunset_bucket.initiate_multipart_upload(key.name,
                metadata=metadata)
            part = 1
            chunk = None
            try:
                for chunk_start in range(0, key.size, self.chunk_size):
                    chunk_end = chunk_start + self.chunk_size - 1
                    if chunk_end >= key.size:
                        chunk_end = key.size - 1
                    chunk_size = chunk_end - chunk_start + 1
                    log.debug('Uploading part #%d (size: %d)' % (part,
                        chunk_size))
                    mp.copy_part_from_key(self.bucket_name, key.name, part,
                        chunk_start, chunk_end)
                    part += 1
                part -= 1
            except Exception as e:
                log.debug(e)
                log.info('Error uploading part %d' % (part,))
                mp.cancel_upload()
                raise
            log.debug('Uploaded %d parts, completing upload' % (part,))
            mp.complete_upload()
        else:
            log.debug('Performing monolithic copy')
            self.sunset_bucket.copy_key(key.name, self.bucket_name, key.name)


<mask token>

<mask token>


class SunsetHandler(object):

    def __init__(self, args):
        self.key = args.aws_key
        self.secret = args.aws_secret
        self.token = args.aws_token
        self.prefix = args.prefix
        self.bucket_name = args.bucket
        self.sunset_bucket_name = args.sunset_bucket
        self.dry_run = args.dry_run
        self.max_size = max_file_size * 2 ** 20
        self.chunk_size = chunk_size
        self.get_buckets()

    def get_buckets(self):
        log.info('Connecting to s3')
        conn = boto.connect_s3(aws_access_key_id=self.key,
            aws_secret_access_key=self.secret, security_token=self.token)
        log.info('Connected to s3')
        self.bucket = conn.get_bucket(self.bucket_name)
        log.info('Have backup bucket')
        self.sunset_bucket = conn.get_bucket(self.sunset_bucket_name)
        log.info('Have sunset bucket')

    def run(self):
        count = 0
        for key in self.bucket.list(prefix=self.prefix):
            count += 1
            if count % 1000 == 0:
                log.info('Iterated over %s keys' % count)
            filename = key.name.split(':')[1]
            if re.search('/system/compactions_in_progress/', filename):
                log.info('Removing key %s from active bucket' % key.name)
                if not self.dry_run:
                    key.delete()
                continue
            if not os.path.isfile(filename):
                delta = datetime.utcnow() - parse_ts(key.last_modified)
                if delta.days <= 0 and delta.seconds < 3600:
                    log.info('key is too new, keeping it: %s: %s' % (key.
                        name, delta))
                    continue
                sunset_key = self.sunset_bucket.get_key(key.name)
                if key.storage_class == 'GLACIER':
                    log.critical(
                        'Glacier file in manifest, unexpected, quiting')
                    sys.exit(1)
                    return
                elif sunset_key == None:
                    log.info('Moving key %s to sunset bucket' % key.name)
                    if not self.dry_run:
                        self.copy(key)
                elif key.size != sunset_key.size:
                    log.info(
                        'Key size mismatch (%d != %d), moving key %s to sunset bucket'
                         % (key.size, sunset_key.size, key.name))
                    if not self.dry_run:
                        self.copy(key)
                else:
                    log.info(
                        'Key %s already present in sunset bucket. copying over to update timestamp'
                         % key.name)
                    if not self.dry_run:
                        try:
                            self.copy(key)
                        except Exception as e:
                            log.error(
                                'key failed to copy, maybe because its already been moved. Skip for now: %s: %s'
                                 % (key.name, e))
                            continue
                log.info('Removing key %s from active bucket' % key.name)
                if not self.dry_run:
                    key.delete()
            else:
                log.info('Keeping key %s' % key.name)

    def copy(self, key):
        log.debug('File size check: %s > %s ? : %s' % (key.size, self.
            max_size, key.size > self.max_size))
        if key.size > self.max_size:
            log.info('Performing multipart copy for %s' % key.name)
            key = self.bucket.get_key(key.name)
            metadata = {}
            for k, v in list(key.metadata.items()):
                metadata[k] = v.encode('utf8')
            mp = self.sunset_bucket.initiate_multipart_upload(key.name,
                metadata=metadata)
            part = 1
            chunk = None
            try:
                for chunk_start in range(0, key.size, self.chunk_size):
                    chunk_end = chunk_start + self.chunk_size - 1
                    if chunk_end >= key.size:
                        chunk_end = key.size - 1
                    chunk_size = chunk_end - chunk_start + 1
                    log.debug('Uploading part #%d (size: %d)' % (part,
                        chunk_size))
                    mp.copy_part_from_key(self.bucket_name, key.name, part,
                        chunk_start, chunk_end)
                    part += 1
                part -= 1
            except Exception as e:
                log.debug(e)
                log.info('Error uploading part %d' % (part,))
                mp.cancel_upload()
                raise
            log.debug('Uploaded %d parts, completing upload' % (part,))
            mp.complete_upload()
        else:
            log.debug('Performing monolithic copy')
            self.sunset_bucket.copy_key(key.name, self.bucket_name, key.name)


def main():
    parser = argparse.ArgumentParser(description=
        'tablesnap_sunset is a script that iterates over all files in an s3 bucket and moves files to cheaper longer term storage options given some meta-data conditions have been met.e.g. File is no longer being used by C* on local file system, File was uploaded more than X days ago, etc.'
        )
    parser.add_argument('-k', '--aws-key', default=os.environ.get(
        'AWS_ACCESS_KEY_ID'), help=
        'Amazon S3 Key (default from AWS_ACCESS_KEY_ID in environment)')
    parser.add_argument('-s', '--aws-secret', default=os.environ.get(
        'AWS_SECRET_ACCESS_KEY'), help=
        'Amazon S3 Secret (default from AWS_SECRET_ACCESS_KEY in environment)')
    parser.add_argument('--aws-token', default=os.environ.get(
        'AWS_SECURITY_TOKEN'), help=
        'Amazon S3 Token (default from AWS_SECURITY_TOKEN in environment)')
    parser.add_argument('--dry-run', action='store_true', help=
        'output which files would be moved')
    parser.add_argument('prefix', help=
        'prefix to key path within bucket, e.g. `production/prodcassandra01:')
    parser.add_argument('bucket', help='S3 bucket containing backup sstables')
    parser.add_argument('sunset_bucket', help=
        'S3 bucket where old, unused files are taken before they die')
    args = parser.parse_args()
    sh = SunsetHandler(args)
    sh.run()


<mask token>

<mask token>
log = logging.getLogger('tablesnap-sunset')
stderr = logging.StreamHandler()
stderr.setFormatter(logging.Formatter(
    '%(name)s [%(asctime)s] %(levelname)s %(message)s'))
log.addHandler(stderr)
if os.environ.get('TDEBUG', False):
    log.setLevel(logging.DEBUG)
else:
    log.setLevel(logging.INFO)
s3_limit = 5 * 2 ** 30
max_file_size = 5120
chunk_size = 1000000000


class SunsetHandler(object):

    def __init__(self, args):
        self.key = args.aws_key
        self.secret = args.aws_secret
        self.token = args.aws_token
        self.prefix = args.prefix
        self.bucket_name = args.bucket
        self.sunset_bucket_name = args.sunset_bucket
        self.dry_run = args.dry_run
        self.max_size = max_file_size * 2 ** 20
        self.chunk_size = chunk_size
        self.get_buckets()

    def get_buckets(self):
        log.info('Connecting to s3')
        conn = boto.connect_s3(aws_access_key_id=self.key,
            aws_secret_access_key=self.secret, security_token=self.token)
        log.info('Connected to s3')
        self.bucket = conn.get_bucket(self.bucket_name)
        log.info('Have backup bucket')
        self.sunset_bucket = conn.get_bucket(self.sunset_bucket_name)
        log.info('Have sunset bucket')

    def run(self):
        count = 0
        for key in self.bucket.list(prefix=self.prefix):
            count += 1
            if count % 1000 == 0:
                log.info('Iterated over %s keys' % count)
            filename = key.name.split(':')[1]
            if re.search('/system/compactions_in_progress/', filename):
                log.info('Removing key %s from active bucket' % key.name)
                if not self.dry_run:
                    key.delete()
                continue
            if not os.path.isfile(filename):
                delta = datetime.utcnow() - parse_ts(key.last_modified)
                if delta.days <= 0 and delta.seconds < 3600:
                    log.info('key is too new, keeping it: %s: %s' % (key.
                        name, delta))
                    continue
                sunset_key = self.sunset_bucket.get_key(key.name)
                if key.storage_class == 'GLACIER':
                    log.critical(
                        'Glacier file in manifest, unexpected, quiting')
                    sys.exit(1)
                    return
                elif sunset_key == None:
                    log.info('Moving key %s to sunset bucket' % key.name)
                    if not self.dry_run:
                        self.copy(key)
                elif key.size != sunset_key.size:
                    log.info(
                        'Key size mismatch (%d != %d), moving key %s to sunset bucket'
                         % (key.size, sunset_key.size, key.name))
                    if not self.dry_run:
                        self.copy(key)
                else:
                    log.info(
                        'Key %s already present in sunset bucket. copying over to update timestamp'
                         % key.name)
                    if not self.dry_run:
                        try:
                            self.copy(key)
                        except Exception as e:
                            log.error(
                                'key failed to copy, maybe because its already been moved. Skip for now: %s: %s'
                                 % (key.name, e))
                            continue
                log.info('Removing key %s from active bucket' % key.name)
                if not self.dry_run:
                    key.delete()
            else:
                log.info('Keeping key %s' % key.name)

    def copy(self, key):
        log.debug('File size check: %s > %s ? : %s' % (key.size, self.
            max_size, key.size > self.max_size))
        if key.size > self.max_size:
            log.info('Performing multipart copy for %s' % key.name)
            key = self.bucket.get_key(key.name)
            metadata = {}
            for k, v in list(key.metadata.items()):
                metadata[k] = v.encode('utf8')
            mp = self.sunset_bucket.initiate_multipart_upload(key.name,
                metadata=metadata)
            part = 1
            chunk = None
            try:
                for chunk_start in range(0, key.size, self.chunk_size):
                    chunk_end = chunk_start + self.chunk_size - 1
                    if chunk_end >= key.size:
                        chunk_end = key.size - 1
                    chunk_size = chunk_end - chunk_start + 1
                    log.debug('Uploading part #%d (size: %d)' % (part,
                        chunk_size))
                    mp.copy_part_from_key(self.bucket_name, key.name, part,
                        chunk_start, chunk_end)
                    part += 1
                part -= 1
            except Exception as e:
                log.debug(e)
                log.info('Error uploading part %d' % (part,))
                mp.cancel_upload()
                raise
            log.debug('Uploaded %d parts, completing upload' % (part,))
            mp.complete_upload()
        else:
            log.debug('Performing monolithic copy')
            self.sunset_bucket.copy_key(key.name, self.bucket_name, key.name)


def main():
    parser = argparse.ArgumentParser(description=
        'tablesnap_sunset is a script that iterates over all files in an s3 bucket and moves files to cheaper longer term storage options given some meta-data conditions have been met.e.g. File is no longer being used by C* on local file system, File was uploaded more than X days ago, etc.'
        )
    parser.add_argument('-k', '--aws-key', default=os.environ.get(
        'AWS_ACCESS_KEY_ID'), help=
        'Amazon S3 Key (default from AWS_ACCESS_KEY_ID in environment)')
    parser.add_argument('-s', '--aws-secret', default=os.environ.get(
        'AWS_SECRET_ACCESS_KEY'), help=
        'Amazon S3 Secret (default from AWS_SECRET_ACCESS_KEY in environment)')
    parser.add_argument('--aws-token', default=os.environ.get(
        'AWS_SECURITY_TOKEN'), help=
        'Amazon S3 Token (default from AWS_SECURITY_TOKEN in environment)')
    parser.add_argument('--dry-run', action='store_true', help=
        'output which files would be moved')
    parser.add_argument('prefix', help=
        'prefix to key path within bucket, e.g. `production/prodcassandra01:')
    parser.add_argument('bucket', help='S3 bucket containing backup sstables')
    parser.add_argument('sunset_bucket', help=
        'S3 bucket where old, unused files are taken before they die')
    args = parser.parse_args()
    sh = SunsetHandler(args)
    sh.run()


if __name__ == '__main__':
    sys.exit(main())

import argparse
import boto
from boto.utils import parse_ts
from datetime import datetime
import logging
import os
import sys
import re
log = logging.getLogger('tablesnap-sunset')
stderr = logging.StreamHandler()
stderr.setFormatter(logging.Formatter(
    '%(name)s [%(asctime)s] %(levelname)s %(message)s'))
log.addHandler(stderr)
if os.environ.get('TDEBUG', False):
    log.setLevel(logging.DEBUG)
else:
    log.setLevel(logging.INFO)
s3_limit = 5 * 2 ** 30
max_file_size = 5120
chunk_size = 1000000000


class SunsetHandler(object):

    def __init__(self, args):
        self.key = args.aws_key
        self.secret = args.aws_secret
        self.token = args.aws_token
        self.prefix = args.prefix
        self.bucket_name = args.bucket
        self.sunset_bucket_name = args.sunset_bucket
        self.dry_run = args.dry_run
        self.max_size = max_file_size * 2 ** 20
        self.chunk_size = chunk_size
        self.get_buckets()

    def get_buckets(self):
        log.info('Connecting to s3')
        conn = boto.connect_s3(aws_access_key_id=self.key,
            aws_secret_access_key=self.secret, security_token=self.token)
        log.info('Connected to s3')
        self.bucket = conn.get_bucket(self.bucket_name)
        log.info('Have backup bucket')
        self.sunset_bucket = conn.get_bucket(self.sunset_bucket_name)
        log.info('Have sunset bucket')

    def run(self):
        count = 0
        for key in self.bucket.list(prefix=self.prefix):
            count += 1
            if count % 1000 == 0:
                log.info('Iterated over %s keys' % count)
            filename = key.name.split(':')[1]
            if re.search('/system/compactions_in_progress/', filename):
                log.info('Removing key %s from active bucket' % key.name)
                if not self.dry_run:
                    key.delete()
                continue
            if not os.path.isfile(filename):
                delta = datetime.utcnow() - parse_ts(key.last_modified)
                if delta.days <= 0 and delta.seconds < 3600:
                    log.info('key is too new, keeping it: %s: %s' % (key.
                        name, delta))
                    continue
                sunset_key = self.sunset_bucket.get_key(key.name)
                if key.storage_class == 'GLACIER':
                    log.critical(
                        'Glacier file in manifest, unexpected, quiting')
                    sys.exit(1)
                    return
                elif sunset_key == None:
                    log.info('Moving key %s to sunset bucket' % key.name)
                    if not self.dry_run:
                        self.copy(key)
                elif key.size != sunset_key.size:
                    log.info(
                        'Key size mismatch (%d != %d), moving key %s to sunset bucket'
                         % (key.size, sunset_key.size, key.name))
                    if not self.dry_run:
                        self.copy(key)
                else:
                    log.info(
                        'Key %s already present in sunset bucket. copying over to update timestamp'
                         % key.name)
                    if not self.dry_run:
                        try:
                            self.copy(key)
                        except Exception as e:
                            log.error(
                                'key failed to copy, maybe because its already been moved. Skip for now: %s: %s'
                                 % (key.name, e))
                            continue
                log.info('Removing key %s from active bucket' % key.name)
                if not self.dry_run:
                    key.delete()
            else:
                log.info('Keeping key %s' % key.name)

    def copy(self, key):
        log.debug('File size check: %s > %s ? : %s' % (key.size, self.
            max_size, key.size > self.max_size))
        if key.size > self.max_size:
            log.info('Performing multipart copy for %s' % key.name)
            key = self.bucket.get_key(key.name)
            metadata = {}
            for k, v in list(key.metadata.items()):
                metadata[k] = v.encode('utf8')
            mp = self.sunset_bucket.initiate_multipart_upload(key.name,
                metadata=metadata)
            part = 1
            chunk = None
            try:
                for chunk_start in range(0, key.size, self.chunk_size):
                    chunk_end = chunk_start + self.chunk_size - 1
                    if chunk_end >= key.size:
                        chunk_end = key.size - 1
                    chunk_size = chunk_end - chunk_start + 1
                    log.debug('Uploading part #%d (size: %d)' % (part,
                        chunk_size))
                    mp.copy_part_from_key(self.bucket_name, key.name, part,
                        chunk_start, chunk_end)
                    part += 1
                part -= 1
            except Exception as e:
                log.debug(e)
                log.info('Error uploading part %d' % (part,))
                mp.cancel_upload()
                raise
            log.debug('Uploaded %d parts, completing upload' % (part,))
            mp.complete_upload()
        else:
            log.debug('Performing monolithic copy')
            self.sunset_bucket.copy_key(key.name, self.bucket_name, key.name)


def main():
    parser = argparse.ArgumentParser(description=
        'tablesnap_sunset is a script that iterates over all files in an s3 bucket and moves files to cheaper longer term storage options given some meta-data conditions have been met.e.g. File is no longer being used by C* on local file system, File was uploaded more than X days ago, etc.'
        )
    parser.add_argument('-k', '--aws-key', default=os.environ.get(
        'AWS_ACCESS_KEY_ID'), help=
        'Amazon S3 Key (default from AWS_ACCESS_KEY_ID in environment)')
    parser.add_argument('-s', '--aws-secret', default=os.environ.get(
        'AWS_SECRET_ACCESS_KEY'), help=
        'Amazon S3 Secret (default from AWS_SECRET_ACCESS_KEY in environment)')
    parser.add_argument('--aws-token', default=os.environ.get(
        'AWS_SECURITY_TOKEN'), help=
        'Amazon S3 Token (default from AWS_SECURITY_TOKEN in environment)')
    parser.add_argument('--dry-run', action='store_true', help=
        'output which files would be moved')
    parser.add_argument('prefix', help=
        'prefix to key path within bucket, e.g. `production/prodcassandra01:')
    parser.add_argument('bucket', help='S3 bucket containing backup sstables')
    parser.add_argument('sunset_bucket', help=
        'S3 bucket where old, unused files are taken before they die')
    args = parser.parse_args()
    sh = SunsetHandler(args)
    sh.run()


if __name__ == '__main__':
    sys.exit(main())

#!/usr/bin/env python

import argparse
import boto
from boto.utils import parse_ts
from datetime import datetime
import logging
import os
import sys
import re


log = logging.getLogger('tablesnap-sunset')
stderr = logging.StreamHandler()
stderr.setFormatter(logging.Formatter(
    '%(name)s [%(asctime)s] %(levelname)s %(message)s'))
log.addHandler(stderr)
if os.environ.get('TDEBUG', False):
    log.setLevel(logging.DEBUG)
else:
    log.setLevel(logging.INFO)

# S3 limit for single file upload
s3_limit = 5 * 2**30

# Max file size to upload without doing multipart in MB
max_file_size = 5120

# Default chunk size for multipart uploads in B
chunk_size = 1000000000

class SunsetHandler(object):

    def __init__(self, args):
        self.key = args.aws_key
        self.secret = args.aws_secret
        self.token = args.aws_token
        self.prefix = args.prefix
        self.bucket_name = args.bucket
        self.sunset_bucket_name = args.sunset_bucket
        self.dry_run = args.dry_run

        self.max_size = max_file_size * 2**20
        self.chunk_size = chunk_size

        self.get_buckets()


    def get_buckets(self):
        log.info('Connecting to s3')
        conn = boto.connect_s3(aws_access_key_id=self.key,
                               aws_secret_access_key=self.secret,
                               security_token=self.token)
        log.info('Connected to s3')
        self.bucket = conn.get_bucket(self.bucket_name)
        log.info('Have backup bucket')
        self.sunset_bucket = conn.get_bucket(self.sunset_bucket_name)
        log.info('Have sunset bucket')


    def run(self):
        count = 0

        for key in self.bucket.list(prefix=self.prefix):
            count += 1
            if (count % 1000 == 0):
                log.info('Iterated over %s keys' % (count))
            filename = key.name.split(':')[1]

            if re.search('/system/compactions_in_progress/', filename):
                # We don't need these files in S3, delete them
                log.info('Removing key %s from active bucket' % (key.name))
                if not self.dry_run:
                    key.delete()
                continue

            if not os.path.isfile(filename):
                # check if timestamp too new, might still be uploading
                # https://forums.aws.amazon.com/thread.jspa?threadID=21634&tstart=0
                delta = datetime.utcnow() - parse_ts(key.last_modified)
                if delta.days <= 0 and delta.seconds < 3600:
                    log.info('key is too new, keeping it: %s: %s' % (key.name, delta))
                    continue

                # check for file in sunset bucket already
                sunset_key = self.sunset_bucket.get_key(key.name)

                if key.storage_class == 'GLACIER':
                    log.critical('Glacier file in manifest, unexpected, quiting')
                    sys.exit(1)
                    return
                elif sunset_key == None:
                    log.info('Moving key %s to sunset bucket' % (key.name))
                    if not self.dry_run:
                        self.copy(key)
                elif key.size != sunset_key.size:
                    log.info('Key size mismatch (%d != %d), moving key %s to sunset bucket' % (key.size, sunset_key.size, key.name))
                    if not self.dry_run:
                        self.copy(key)
                else:
                    log.info('Key %s already present in sunset bucket. copying over to update timestamp' % (key.name))
                    if not self.dry_run:
                        try:
                            self.copy(key)
                        except Exception as e:
                            log.error('key failed to copy, maybe because its already been moved. Skip for now: %s: %s' % (key.name, e))
                            continue

                log.info('Removing key %s from active bucket' % (key.name))
                if not self.dry_run:
                    key.delete()
            else:
                log.info('Keeping key %s' % (key.name))

    def copy(self, key):
        log.debug('File size check: %s > %s ? : %s' %
            (key.size, self.max_size,
            (key.size > self.max_size),))
        if key.size > self.max_size:
            log.info('Performing multipart copy for %s' %
                         (key.name))
            # refetch key to make sure we have metadata
            key = self.bucket.get_key(key.name)
            # work around bug https://github.com/boto/boto/issues/2536
            metadata = {}
            for k, v in list(key.metadata.items()):
                metadata[k] = v.encode('utf8')

            mp = self.sunset_bucket.initiate_multipart_upload(key.name,
                metadata=metadata)
            part = 1
            chunk = None
            try:
                for chunk_start in range(0, key.size, self.chunk_size):
                    chunk_end = chunk_start + self.chunk_size -1
                    if chunk_end >= key.size:
                        chunk_end = key.size-1
                    chunk_size = chunk_end - chunk_start + 1
                    log.debug('Uploading part #%d '
                                   '(size: %d)' %
                                   (part, chunk_size,))
                    mp.copy_part_from_key(self.bucket_name, key.name, part, chunk_start, chunk_end)
                    part += 1
                part -= 1
            except Exception as e:
                log.debug(e)
                log.info('Error uploading part %d' % (part,))
                mp.cancel_upload()
                raise
            log.debug('Uploaded %d parts, '
                           'completing upload' % (part,))
            mp.complete_upload()
        else:
            log.debug('Performing monolithic copy')
            self.sunset_bucket.copy_key(key.name, self.bucket_name, key.name)

def main():
    parser = argparse.ArgumentParser(description='tablesnap_sunset is a script that '
        'iterates over all files in an s3 bucket and moves files to cheaper longer term '
        'storage options given some meta-data conditions have been met.'
        'e.g. File is no longer being used by C* on local file system, File was uploaded '
        'more than X days ago, etc.')
    parser.add_argument('-k', '--aws-key',
        default=os.environ.get('AWS_ACCESS_KEY_ID'),
        help='Amazon S3 Key (default from AWS_ACCESS_KEY_ID in environment)')
    parser.add_argument('-s', '--aws-secret',
        default=os.environ.get('AWS_SECRET_ACCESS_KEY'),
        help='Amazon S3 Secret (default from AWS_SECRET_ACCESS_KEY in environment)')
    parser.add_argument('--aws-token',
        default=os.environ.get('AWS_SECURITY_TOKEN'),
        help='Amazon S3 Token (default from AWS_SECURITY_TOKEN in environment)')
    parser.add_argument('--dry-run', action='store_true',
        help='output which files would be moved')

    parser.add_argument('prefix', help='prefix to key path within bucket, e.g. `production/prodcassandra01:')
    parser.add_argument('bucket', help='S3 bucket containing backup sstables')
    parser.add_argument('sunset_bucket',
        help='S3 bucket where old, unused files are taken before they die')

    args = parser.parse_args()

    sh = SunsetHandler(args)
    sh.run()

if __name__ == '__main__':
    sys.exit(main())
