import logging
import types
import time

from cassandra import ConsistencyLevel
from cassandra.query import SimpleStatement
from ccmlib.cluster import DEFAULT_CLUSTER_WAIT_TIMEOUT_IN_SECS
from ccmlib.node import Node
from ccmlib.node import NODE_WAIT_TIMEOUT_IN_SECS

from dtest import Tester
from tools.assertions import (assert_all)

from flaky import flaky

from cassandra.metadata import OrderedDict
import pytest
from itertools import chain
from tools.misc import new_node

since = pytest.mark.since

logger = logging.getLogger(__name__)


NODELOCAL = 11


def jmx_start(to_start, **kwargs):
    kwargs['jvm_args'] = kwargs.get('jvm_args', []) + ['-XX:-PerfDisableSharedMem']
    to_start.start(**kwargs)


def gen_expected(*values):
    return [["%05d" % i, i, i] for i in chain(*values)]


def repair_nodes(nodes):
    for node in nodes:
        node.nodetool('repair -pr')


def cleanup_nodes(nodes):
    for node in nodes:
        logger.debug("cleaning up %s" % node)
        node.nodetool('cleanup')


def patch_start(startable):
    old_start = startable.start

    def new_start(self, *args, **kwargs):
        kwargs['jvm_args'] = kwargs.get('jvm_args', []) + ['-XX:-PerfDisableSharedMem'
                                                           ' -Dcassandra.enable_nodelocal_queries=true']
        return old_start(*args, **kwargs)

    startable.start = types.MethodType(new_start, startable)


@since('4.0')
class TestTransientReplicationRing(Tester):

    keyspace = "ks"
    table = "tbl"

    def select(self):
        return "SELECT * from %s.%s" % (self.keyspace, self.table)

    def select_statement(self):
        return SimpleStatement(self.select(), consistency_level=NODELOCAL)

    def point_select(self):
        return "SELECT * from %s.%s where pk = %%s" % (self.keyspace, self.table)

    def point_select_statement(self):
        return SimpleStatement(self.point_select(), consistency_level=NODELOCAL)

    def check_expected(self, sessions, expected, node=None, cleanup=False):
        """Check that each node has the expected values present"""
        if node is None:
            node = list(range(1000))
        for idx, session, expect, node in zip(range(0, 1000), sessions, expected, node):
            logger.debug("Checking idx " + str(idx))
            logger.debug(str([row for row in session.execute(self.select_statement())]))
            if cleanup:
                node.nodetool('cleanup')
            assert_all(session,
                       self.select(),
                       expect,
                       cl=NODELOCAL)

    def check_replication(self, sessions, exactly=None, gte=None, lte=None):
        """Assert that the test values are replicated a required number of times"""
        for i in range(0, 40):
            count = 0
            for session in sessions:
                for _ in session.execute(self.point_select_statement(), ["%05d" % i]):
                    count += 1
            if exactly:
                assert count == exactly, "Wrong replication for %05d should be exactly %d" % (i, exactly)
            if gte:
                assert count >= gte, "Count for %05d should be >= %d" % (i, gte)
            if lte:
                assert count <= lte, "Count for %05d should be <= %d" % (i, lte)

    @pytest.fixture
    def cheap_quorums(self):
        return False

    @pytest.fixture(scope='function', autouse=True)
    def setup_cluster(self, fixture_dtest_setup):
        self.tokens = ['00010', '00020', '00030']

        patch_start(self.cluster)
        self.cluster.set_configuration_options(values={'hinted_handoff_enabled': False,
                                                       'num_tokens': 1,
                                                       'commitlog_sync_period_in_ms': 500,
                                                       'enable_transient_replication': True,
                                                       'partitioner': 'org.apache.cassandra.dht.OrderPreservingPartitioner'})
        logger.debug("CLUSTER INSTALL DIR: ")
        logger.debug(self.cluster.get_install_dir())
        self.cluster.populate(3, tokens=self.tokens, debug=True, install_byteman=True)
        # self.cluster.populate(3, debug=True, install_byteman=True)
        self.cluster.start(jvm_args=['-Dcassandra.enable_nodelocal_queries=true'],
                           timeout=DEFAULT_CLUSTER_WAIT_TIMEOUT_IN_SECS * 4)

        # enable shared memory
        for node in self.cluster.nodelist():
            patch_start(node)
            logger.debug(node.logfilename())

        self.nodes = self.cluster.nodelist()
        self.node1, self.node2, self.node3 = self.nodes
        session = self.exclusive_cql_connection(self.node3)

        replication_params = OrderedDict()
        replication_params['class'] = 'NetworkTopologyStrategy'
        replication_params['datacenter1'] = '3/1'
        replication_params = ', '.join("'%s': '%s'" % (k, v) for k, v in replication_params.items())

        session.execute("CREATE KEYSPACE %s WITH REPLICATION={%s}" % (self.keyspace, replication_params))
        logger.debug("CREATE KEYSPACE %s WITH REPLICATION={%s}" % (self.keyspace, replication_params))
        session.execute(
            "CREATE TABLE %s.%s (pk varchar, ck int, value int, PRIMARY KEY (pk, ck)) WITH speculative_retry = 'NEVER' AND additional_write_policy = 'NEVER' AND read_repair = 'NONE'" % (
                self.keyspace, self.table))

    def quorum(self, session, stmt_str):
        return session.execute(SimpleStatement(stmt_str, consistency_level=ConsistencyLevel.QUORUM))

    def insert_row(self, pk, ck, value, session=None, node=None):
        session = session or self.exclusive_cql_connection(node or self.node1)
        # token = BytesToken.from_key(pack('>i', pk)).value
        # assert token < BytesToken.from_string(self.tokens[0]).value or BytesToken.from_string(self.tokens[-1]).value < token   # primary replica should be node1
        # TODO Is quorum really right? I mean maybe we want ALL with retries since we really don't want to the data
        # not at a replica unless it is intentional
        self.quorum(session, "INSERT INTO %s.%s (pk, ck, value) VALUES ('%05d', %s, %s)" % (self.keyspace, self.table, pk, ck, value))

    @flaky(max_runs=1)
    @pytest.mark.no_vnodes
    def test_bootstrap_and_cleanup(self):
        """Test bootstrapping a new node across a mix of repaired and unrepaired data"""
        main_session = self.patient_cql_connection(self.node1)
        nodes = [self.node1, self.node2, self.node3]

        for i in range(0, 40, 2):
            self.insert_row(i, i, i, main_session)

        sessions = [self.exclusive_cql_connection(node) for node in [self.node1, self.node2, self.node3]]

        expected = [gen_expected(range(0, 11, 2), range(22, 40, 2)),
                    gen_expected(range(0, 22, 2), range(32, 40, 2)),
                    gen_expected(range(12, 31, 2))]
        self.check_expected(sessions, expected)

        # Make sure at least a little data is repaired, this shouldn't move data anywhere
        repair_nodes(nodes)

        self.check_expected(sessions, expected)

        # Ensure that there is at least some transient data around, because of this if it's missing after bootstrap
        # We know we failed to get it from the transient replica losing the range entirely
        nodes[1].stop(wait_other_notice=True)

        for i in range(1, 40, 2):
            self.insert_row(i, i, i, main_session)

        # we don't need to wait for other nodes, see CASSANDRA-18660
        nodes[1].start(wait_for_binary_proto=True, wait_other_notice=False)

        sessions = [self.exclusive_cql_connection(node) for node in [self.node1, self.node2, self.node3]]

        expected = [gen_expected(range(0, 11), range(11, 20, 2), range(21, 40)),
                    gen_expected(range(0, 21, 2), range(32, 40, 2)),
                    gen_expected(range(1, 11, 2), range(11, 31), range(31, 40, 2))]

        # Every node should have some of its fully replicated data and one and two should have some transient data
        self.check_expected(sessions, expected)

        node4 = new_node(self.cluster, bootstrap=True, token='00040')
        patch_start(node4)
        nodes.append(node4)
        node4.start(wait_for_binary_proto=True)

        expected.append(gen_expected(range(11, 20, 2), range(21, 40)))
        sessions.append(self.exclusive_cql_connection(node4))

        # Because repair was never run and nodes had transient data it will have data for transient ranges (node1, 11-20)
        assert_all(sessions[3],
                   self.select(),
                   expected[3],
                   cl=NODELOCAL)

        # Node1 no longer transiently replicates 11-20, so cleanup will clean it up
        # Node1 also now transiently replicates 21-30 and half the values in that range were repaired
        expected[0] = gen_expected(range(0, 11), range(21, 30, 2), range(31, 40))
        # Node2 still missing data since it was down during some insertions, it also lost some range (31-40)
        expected[1] = gen_expected(range(0, 21, 2))
        expected[2] = gen_expected(range(1, 11, 2), range(11, 31))

        # Cleanup should only impact if a node lost a range entirely or started to transiently replicate it and the data
        # was repaired
        self.check_expected(sessions, expected, nodes, cleanup=True)

        repair_nodes(nodes)

        expected = [gen_expected(range(0, 11), range(31, 40)),
                    gen_expected(range(0, 21)),
                    gen_expected(range(11, 31)),
                    gen_expected(range(21, 40))]

        self.check_expected(sessions, expected, nodes, cleanup=True)

        # Every value should be replicated exactly 2 times
        self.check_replication(sessions, exactly=2)

    @pytest.mark.no_vnodes
    def move_test(self, move_token, expected_after_move, expected_after_repair):
        """Helper method to run a move test cycle"""
        node4 = new_node(self.cluster, bootstrap=True, token='00040')
        patch_start(node4)
        node4.start(wait_for_binary_proto=NODE_WAIT_TIMEOUT_IN_SECS * 2)
        main_session = self.patient_cql_connection(self.node1)
        nodes = [self.node1, self.node2, self.node3, node4]

        for i in range(0, 40, 2):
            logger.debug("Inserting " + str(i))
            self.insert_row(i, i, i, main_session)

        # Make sure at least a little data is repaired
        repair_nodes(nodes)

        # Ensure that there is at least some transient data around, because of this if it's missing after bootstrap
        # We know we failed to get it from the transient replica losing the range entirely
        nodes[1].stop(wait_other_notice=True)

        for i in range(1, 40, 2):
            logger.debug("Inserting " + str(i))
            self.insert_row(i, i, i, main_session)

        nodes[1].start(wait_for_binary_proto=True)
        sessions = [self.exclusive_cql_connection(node) for node in [self.node1, self.node2, self.node3, node4]]

        expected = [gen_expected(range(0, 11), range(31, 40)),
                    gen_expected(range(0, 21, 2)),
                    gen_expected(range(1, 11, 2), range(11, 31)),
                    gen_expected(range(11, 20, 2), range(21, 40))]
        self.check_expected(sessions, expected)
        self.check_replication(sessions, exactly=2)

        nodes[0].move(move_token)
        # we need a small sleep for the remote side to finish closing the session, see CASSANDRA-18792
        time.sleep(1)
        cleanup_nodes(nodes)

        self.check_replication(sessions, gte=2, lte=3)
        self.check_expected(sessions, expected=expected_after_move)

        repair_nodes(nodes)

        self.check_expected(sessions, expected_after_repair, nodes, cleanup=True)
        self.check_replication(sessions, exactly=2)

    @flaky(max_runs=1)
    @pytest.mark.no_vnodes
    def test_move_forwards_between_and_cleanup(self):
        """Test moving a node forwards past a neighbor token"""
        move_token = '00025'
        expected_after_move = [gen_expected(range(0, 26), range(31, 40, 2)),
                               gen_expected(range(0, 21, 2), range(31, 40)),
                               gen_expected(range(1, 11, 2), range(11, 21, 2), range(21, 31)),
                               gen_expected(range(21, 26, 2), range(26, 40))]
        expected_after_repair = [gen_expected(range(0, 26)),
                                 gen_expected(range(0, 21), range(31, 40)),
                                 gen_expected(range(21, 31),),
                                 gen_expected(range(26, 40))]
        self.move_test(move_token, expected_after_move, expected_after_repair)

    @flaky(max_runs=1)
    @pytest.mark.no_vnodes
    def test_move_forwards_and_cleanup(self):
        """Test moving a node forwards without going past a neighbor token"""
        move_token = '00015'
        expected_after_move = [gen_expected(range(0, 16), range(31, 40)),
                               gen_expected(range(0, 21, 2)),
                               gen_expected(range(1, 16, 2), range(16, 31)),
                               gen_expected(range(17, 20, 2), range(21, 40))]
        expected_after_repair = [gen_expected(range(0, 16), range(31, 40)),
                                 gen_expected(range(0, 21)),
                                 gen_expected(range(16, 31)),
                                 gen_expected(range(21, 40))]
        self.move_test(move_token, expected_after_move, expected_after_repair)

    @flaky(max_runs=1)
    @pytest.mark.no_vnodes
    def test_move_backwards_between_and_cleanup(self):
        """Test moving a node backwards past it's preceding neighbor's token"""
        move_token = '00035'
        expected_after_move = [gen_expected(range(1, 21, 2), range(21, 36)),
                               gen_expected(range(0, 21, 2), range(36, 40)),
                               gen_expected(range(0, 31), range(37, 40, 2)),
                               gen_expected(range(21, 30, 2), range(31, 40))]
        expected_after_repair = [gen_expected(range(21, 36)),
                                 gen_expected(range(0, 21), range(36, 40)),
                                 gen_expected(range(0, 31)),
                                 gen_expected(range(31, 40))]
        self.move_test(move_token, expected_after_move, expected_after_repair)

    @flaky(max_runs=1)
    @pytest.mark.no_vnodes
    def test_move_backwards_and_cleanup(self):
        """Test moving a node backwards without moving past a neighbor token"""
        move_token = '00005'
        expected_after_move = [gen_expected(range(0, 6), range(31, 40)),
                               gen_expected(range(0, 21, 2)),
                               gen_expected(range(1, 6, 2), range(6, 31)),
                               gen_expected(range(7, 20, 2), range(21, 40))]
        expected_after_repair = [gen_expected(range(0, 6), range(31, 40)),
                                 gen_expected(range(0, 21)),
                                 gen_expected(range(6, 31)),
                                 gen_expected(range(21, 40))]
        self.move_test(move_token, expected_after_move, expected_after_repair)

    @flaky(max_runs=1)
    @pytest.mark.no_vnodes
    def test_decommission(self):
        """Test decommissioning a node correctly streams out all the data"""
        node4 = new_node(self.cluster, bootstrap=True, token='00040')
        patch_start(node4)
        node4.start(wait_for_binary_proto=True)
        main_session = self.patient_cql_connection(self.node1)
        nodes = [self.node1, self.node2, self.node3, node4]

        for i in range(0, 40, 2):
            logger.debug("Inserting " + str(i))
            self.insert_row(i, i, i, main_session)

        # Make sure at least a little data is repaired
        repair_nodes(nodes)

        # Ensure that there is at least some transient data around, because of this if it's missing after bootstrap
        # We know we failed to get it from the transient replica losing the range entirely
        nodes[1].stop(wait_other_notice=True)

        for i in range(1, 40, 2):
            logger.debug("Inserting " + str(i))
            self.insert_row(i, i, i, main_session)

        nodes[1].start(wait_for_binary_proto=True)
        sessions = [self.exclusive_cql_connection(node) for node in [self.node1, self.node2, self.node3, node4]]

        expected = [gen_expected(range(0, 11), range(31, 40)),
                    gen_expected(range(0, 21, 2)),
                    gen_expected(range(1, 11, 2), range(11, 31)),
                    gen_expected(range(11, 20, 2), range(21, 40))]

        self.check_expected(sessions, expected)

        # node1 has transient data we want to see streamed out on move
        nodes[3].nodetool('decommission')

        nodes = nodes[:-1]
        sessions = sessions[:-1]

        expected = [gen_expected(range(0, 11), range(11, 21, 2), range(21, 40)),
                    gen_expected(range(0, 21, 2), range(21, 30, 2), range(31, 40)),
                    gen_expected(range(1, 11, 2), range(11, 31), range(31, 40, 2))]

        cleanup_nodes(nodes)

        self.check_replication(sessions, gte=2, lte=3)
        self.check_expected(sessions, expected)

        repair_nodes(nodes)

        # There should be no transient data anywhere
        expected = [gen_expected(range(0, 11), range(21, 40)),
                    gen_expected(range(0, 21), range(31, 40)),
                    gen_expected(range(11, 31))]

        self.check_expected(sessions, expected, nodes, cleanup=True)
        self.check_replication(sessions, exactly=2)

    @pytest.mark.no_vnodes
    def test_remove(self):
        """Test  a mix of ring change operations across a mix of transient and repaired/unrepaired data"""
        node4 = new_node(self.cluster, bootstrap=True, token='00040')
        patch_start(node4)
        node4.start(wait_for_binary_proto=True)
        main_session = self.patient_cql_connection(self.node1)
        nodes = [self.node1, self.node2, self.node3]

        # We want the node being removed to have no data on it
        # so nodetool remove always gets all the necessary data from survivors
        node4_id = node4.nodetool('info').stdout[25:61]
        node4.stop(wait_other_notice=True)

        for i in range(0, 40):
            self.insert_row(i, i, i, main_session)

        sessions = [self.exclusive_cql_connection(node) for node in [self.node1, self.node2, self.node3]]

        expected = [gen_expected(range(0, 11), range(21, 40)),
                    gen_expected(range(0, 21), range(31, 40)),
                    gen_expected(range(11, 31))]

        # Every node should some of its fully replicated data and one and two should have some transient data
        self.check_expected(sessions, expected)

        nodes[0].nodetool('removenode ' + node4_id)

#        self._everyone_should_have_everything(sessions) # no they shouldn't

        repair_nodes(nodes)
        cleanup_nodes(nodes)

        self._nodes_have_proper_ranges_after_repair_and_cleanup(sessions)

    @pytest.mark.no_vnodes
    def test_replace(self):
        main_session = self.patient_cql_connection(self.node1)

        # We want the node being replaced to have no data on it so the replacement definitely fetches all the data
        self.node2.stop(wait_other_notice=True)

        for i in range(0, 40):
            self.insert_row(i, i, i, main_session)

        replacement_address = self.node2.address()
        self.cluster.remove(self.node2)
        self.node2 = Node('replacement', cluster=self.cluster, auto_bootstrap=True,
                          thrift_interface=None, storage_interface=(replacement_address, 7000),
                          jmx_port='7400', remote_debug_port='0', initial_token=None, binary_interface=(replacement_address, 9042))
        patch_start(self.node2)
        nodes = [self.node1, self.node2, self.node3]
        self.cluster.add(self.node2, False, data_center='datacenter1')
        jvm_args = ["-Dcassandra.replace_address=%s" % replacement_address,
                    "-Dcassandra.ring_delay_ms=10000",
                    "-Dcassandra.broadcast_interval_ms=10000"]
        self.node2.start(jvm_args=jvm_args, wait_for_binary_proto=True)

        sessions = [self.exclusive_cql_connection(node) for node in [self.node1, self.node2, self.node3]]

        self._everyone_should_have_everything(sessions)

        repair_nodes(nodes)
        cleanup_nodes(nodes)

        self._nodes_have_proper_ranges_after_repair_and_cleanup(sessions)

    def _everyone_should_have_everything(self, sessions):
        expected = [gen_expected(range(0, 40))] * 3
        self.check_replication(sessions, exactly=3)
        self.check_expected(sessions, expected)

    def _nodes_have_proper_ranges_after_repair_and_cleanup(self, sessions):
        expected = [gen_expected(range(0, 11), range(21, 40)),
                    gen_expected(range(0, 21), range(31, 40)),
                    gen_expected(range(11, 31))]
        self.check_replication(sessions, exactly=2)
        self.check_expected(sessions, expected)