from __future__ import division

import time
from math import floor
from os.path import getsize

from dtest import Tester, debug


class TestSSTableSplit(Tester):

    def split_test(self):
        """
        Check that after running compaction, sstablessplit can succesfully split
        The resultant sstable.  Check that split is reversible and that data is readable
        after carrying out these operations.
        """
        cluster = self.cluster
        cluster.populate(1).start(wait_for_binary_proto=True)
        node = cluster.nodelist()[0]
        version = cluster.version()

        debug("Run stress to insert data")

        node.stress(['write', 'n=1000', 'no-warmup', '-rate', 'threads=50',
                     '-col', 'n=FIXED(10)', 'SIZE=FIXED(1024)'])

        self._do_compaction(node)
        self._do_split(node, version)
        self._do_compaction(node)
        self._do_split(node, version)

        debug("Run stress to ensure data is readable")
        node.stress(['read', 'n=1000', '-rate', 'threads=25',
                     '-col', 'n=FIXED(10)', 'SIZE=FIXED(1024)'])

    def _do_compaction(self, node):
        debug("Compact sstables.")
        node.flush()
        node.compact()
        keyspace = 'keyspace1'
        sstables = node.get_sstables(keyspace, '')
        debug("Number of sstables after compaction: %s" % len(sstables))

    def _do_split(self, node, version):
        debug("Run sstablesplit")
        time.sleep(5.0)
        node.stop()

        # default split size is 50MB
        splitmaxsize = 10
        expected_sstable_size = (10 * 1024 * 1024)
        keyspace = 'keyspace1'

        # get the initial sstables and their total size
        origsstables = node.get_sstables(keyspace, '')
        origsstable_size = sum([getsize(sstable) for sstable in origsstables])
        debug("Original sstable and sizes before split: {}".format([(name, getsize(name)) for name in origsstables]))

        # calculate the expected number of sstables post-split
        expected_num_sstables = floor(origsstable_size / expected_sstable_size)

        # split the sstables
        result = node.run_sstablesplit(keyspace=keyspace, size=splitmaxsize,
                                       no_snapshot=True, debug=True)

        for (out, error, rc) in result:
            debug("stdout: {}".format(out))
            debug("stderr: {}".format(error))
            debug("rc: {}".format(rc))

        # get the sstables post-split and their total size
        sstables = node.get_sstables(keyspace, '')
        debug("Number of sstables after split: %s. expected %s" % (len(sstables), expected_num_sstables))
        self.assertLessEqual(expected_num_sstables, len(sstables) + 1)
        self.assertLessEqual(1, len(sstables))

        # make sure none of the tables are bigger than the max expected size
        sstable_sizes = [getsize(sstable) for sstable in sstables]
        # add a bit extra for overhead
        self.assertLessEqual(max(sstable_sizes), expected_sstable_size + 512)
        # make sure node can start with changed sstables
        node.start(wait_for_binary_proto=True)

    def single_file_split_test(self):
        """
        Covers CASSANDRA-8623

        Check that sstablesplit doesn't crash when splitting a single sstable at the time.
        """
        cluster = self.cluster
        cluster.populate(1).start(wait_for_binary_proto=True)
        node = cluster.nodelist()[0]

        debug("Run stress to insert data")
        node.stress(['write', 'n=300', 'no-warmup', '-rate', 'threads=50',
                     '-col', 'n=FIXED(10)', 'SIZE=FIXED(1024)'])

        self._do_compaction(node)
        node.stop()
        result = node.run_sstablesplit(keyspace='keyspace1', size=1, no_snapshot=True)

        for (stdout, stderr, rc) in result:
            debug(stderr)
            failure = stderr.find("java.lang.AssertionError: Data component is missing")
            self.assertEqual(failure, -1, "Error during sstablesplit")

        sstables = node.get_sstables('keyspace1', '')
        self.assertGreaterEqual(len(sstables), 1, sstables)