euca_8999.py

#!/usr/bin/env python

from __future__ import division
import time
import os
import hashlib
import tempfile
from math import ceil
from cStringIO import StringIO
from concurrent.futures import ThreadPoolExecutor
from eucaops import Eucaops
from eucaops import S3ops
from eutester.eutestcase import EutesterTestCase


class Euca8999(EutesterTestCase):

    def __init__(self):
        self.setuptestcase()
        self.setup_parser()
        self.parser.add_argument("-T", "--threads", type=int, default=5)
        self.parser.add_argument("-b", "--buckets", type=int, default=1)
        self.parser.add_argument("-o", "--objects", type=int, default=5)
        self.parser.add_argument("-S", "--object-size", type=int, default=5, help="Object size in MB")
        self.get_args()
        # Setup basic eutester object
        if self.args.region:
            self.tester = S3ops( credpath=self.args.credpath, region=self.args.region)
        else:
            self.tester = Eucaops(credpath=self.args.credpath, config_file=self.args.config, password=self.args.password)
        self.start = time.time()

        self.bucket_names = []
        self.bucket_name = "concurrency-" + str(int(self.start)) + "-"

        for i in xrange(self.args.buckets):
            bucket_name = self.bucket_name + str(i)
            self.bucket_names.append(bucket_name)
            self.tester.create_bucket(bucket_name)

        self.temp_files = []

    def clean_method(self):
        with ThreadPoolExecutor(max_workers=self.args.threads) as executor:
            for i in xrange(self.args.buckets):
                executor.submit(self.tester.clear_bucket(self.bucket_names[i]))
        for tf in self.temp_files:
            tf.close()

    def create_file(self, size_in_mb, file_name="eutester-object"):
        temp_file = tempfile.NamedTemporaryFile(mode='w+b', prefix=file_name)
        self.temp_files.append(temp_file)
        temp_file.write(os.urandom(1024 * 1024 * size_in_mb))
        return temp_file.name

    def put_get_check(self, bucket_name, key_name, eu_file):
        bucket = self.tester.get_bucket_by_name(bucket_name)
        try:
            self.single_upload(bucket, key_name, eu_file.name)
        except Exception as e:
            self.fail(e.message)
        return True

    def single_upload(self, bucket, key_name, file_path):
        key = bucket.new_key(key_name)
        try:
            key.set_contents_from_filename(file_path)
        except:
            self.fail("Failed to upload object: " + file_path)
        self.debug("Uploaded key '" + key_name + "' to bucket '" + bucket.name + "'")
        return key

    def concurrent_upload(self):
        self.debug("Creating object of " + str(self.args.object_size) + "MB")
        eu_file = open(self.create_file(self.args.object_size))
        thread_pool = []
        with ThreadPoolExecutor(max_workers=self.args.threads) as executor:
            for i in xrange(self.args.buckets):
                for j in xrange(self.args.objects):
                    thread_pool.append(executor.submit(self.put_get_check, bucket_name=self.bucket_names[i],
                                                       key_name=eu_file.name + str(j), eu_file=eu_file))

        for tp in thread_pool:
            try:
                if not tp.result():
                    self.fail("[CRITICAL] failed upload in thread")
            except Exception as e:
                self.fail("Found exception in thread-pool: " + e.message)

if __name__ == "__main__":
    testcase = Euca8999()
    ### Use the list of tests passed from config/command line to determine what subset of tests to run
    ### or use a predefined list
    list = testcase.args.tests or ["concurrent_upload"]

    ### Convert test suite methods to EutesterUnitTest objects
    unit_list = [ ]
    for test in list:
        unit_list.append( testcase.create_testunit_by_name(test))

    ### Run the EutesterUnitTest objects
    result = testcase.run_test_case_list(unit_list)
    exit(result)

# this test passes with 10mb objects
#testcases/stress_tests/euca_8999.py --config /root/input/test --password foobar --threads 100 --objects 200 --object-size 10

# and 100mb files to reproduce the issue
# testcases/stress_tests/euca_8999.py --config /root/input/test --password foobar --threads 100 --objects 200 --object-size 100