import os
import time
import pytest
import parse
import re
import logging

import ccmlib.common
from ccmlib.node import ToolError

from distutils.version import LooseVersion

from dtest import Tester, create_ks
from tools.jmxutils import (JolokiaAgent, enable_jmx_ssl, make_mbean)
                            
from tools.misc import generate_ssl_stores
from tools.data import create_c1c2_table, insert_c1c2

since = pytest.mark.since
logger = logging.getLogger(__name__)


class TestJMX(Tester):

    @pytest.fixture(autouse=True)
    def fixture_add_additional_log_patterns(self, fixture_dtest_setup):
        fixture_dtest_setup.ignore_log_patterns = (
            r'Failed to properly handshake with peer.* Closing the channel'
        )

    def test_netstats(self):
        """
        Check functioning of nodetool netstats, especially with restarts.
        @jira_ticket CASSANDRA-8122, CASSANDRA-6577
        """
        #
        cluster = self.cluster
        cluster.populate(3).start()
        node1, node2, node3 = cluster.nodelist()

        node1.stress(['write', 'n=500K', 'no-warmup', '-schema', 'replication(factor=3)'])
        node1.flush()
        node1.stop(gently=False)

        with pytest.raises(ToolError, message=r"ConnectException: 'Connection refused( \(Connection refused\))?'."):
            node1.nodetool('netstats')

        # don't wait; we're testing for when nodetool is called on a node mid-startup
        node1.start(wait_for_binary_proto=False)

        # until the binary interface is available, try `nodetool netstats`
        binary_interface = node1.network_interfaces['binary']
        time_out_at = time.time() + 30
        running = False
        while (not running and time.time() <= time_out_at):
            running = ccmlib.common.check_socket_listening(binary_interface, timeout=0.5)
            try:
                node1.nodetool('netstats')
            except Exception as e:
                assert 'java.lang.reflect.UndeclaredThrowableException' not in str(e), \
                    'Netstats failed with UndeclaredThrowableException (CASSANDRA-8122)'
                if not isinstance(e, ToolError):
                    raise
                else:
                    assert re.search("ConnectException: 'Connection refused( \(Connection refused\))?'.", repr(e))

        assert running, 'node1 never started'

    def test_table_metric_mbeans(self):
        """
        Test some basic table metric mbeans with simple writes.
        """
        cluster = self.cluster
        cluster.populate(3)
        node1, node2, node3 = cluster.nodelist()
        cluster.start()

        version = cluster.version()
        node1.stress(['write', 'n=10K', 'no-warmup', '-schema', 'replication(factor=3)'])

        typeName = "ColumnFamily" if version < '3.0' else 'Table'
        logger.debug('Version {} typeName {}'.format(version, typeName))

        # TODO the keyspace and table name are capitalized in 2.0
        memtable_size = make_mbean('metrics', type=typeName, keyspace='keyspace1', scope='standard1',
                                   name='AllMemtablesHeapSize')
        disk_size = make_mbean('metrics', type=typeName, keyspace='keyspace1', scope='standard1',
                               name='LiveDiskSpaceUsed')
        sstable_count = make_mbean('metrics', type=typeName, keyspace='keyspace1', scope='standard1',
                                   name='LiveSSTableCount')

        with JolokiaAgent(node1) as jmx:
            mem_size = jmx.read_attribute(memtable_size, "Value")
            assert int(mem_size) > 10000

            on_disk_size = jmx.read_attribute(disk_size, "Count")
            assert int(on_disk_size) == 0

            node1.flush()

            on_disk_size = jmx.read_attribute(disk_size, "Count")
            assert int(on_disk_size) > 10000

            sstables = jmx.read_attribute(sstable_count, "Value")
            assert int(sstables) >= 1

    @since('3.0')
    def test_mv_metric_mbeans_release(self):
        """
        Test that the right mbeans are created and released when creating mvs
        """
        cluster = self.cluster
        cluster.set_configuration_options({'enable_materialized_views': 'true'})
        cluster.populate(1)
        node = cluster.nodelist()[0]
        cluster.start()

        node.run_cqlsh(cmds="""
            CREATE KEYSPACE mvtest WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor': 1 };
            CREATE TABLE mvtest.testtable (
                foo int,
                bar text,
                baz text,
                PRIMARY KEY (foo, bar)
            );

            CREATE MATERIALIZED VIEW mvtest.testmv AS
                SELECT foo, bar, baz FROM mvtest.testtable WHERE
                foo IS NOT NULL AND bar IS NOT NULL AND baz IS NOT NULL
            PRIMARY KEY (foo, bar, baz);""")

        table_memtable_size = make_mbean('metrics', type='Table', keyspace='mvtest', scope='testtable',
                                         name='AllMemtablesHeapSize')
        table_view_read_time = make_mbean('metrics', type='Table', keyspace='mvtest', scope='testtable',
                                          name='ViewReadTime')
        table_view_lock_time = make_mbean('metrics', type='Table', keyspace='mvtest', scope='testtable',
                                          name='ViewLockAcquireTime')
        mv_memtable_size = make_mbean('metrics', type='Table', keyspace='mvtest', scope='testmv',
                                      name='AllMemtablesHeapSize')
        mv_view_read_time = make_mbean('metrics', type='Table', keyspace='mvtest', scope='testmv',
                                       name='ViewReadTime')
        mv_view_lock_time = make_mbean('metrics', type='Table', keyspace='mvtest', scope='testmv',
                                       name='ViewLockAcquireTime')

        missing_metric_message = "Table metric %s should have been registered after creating table %s" \
                                 "but wasn't!"

        with JolokiaAgent(node) as jmx:
            assert jmx.read_attribute(table_memtable_size, "Value") is not None, \
                missing_metric_message.format("AllMemtablesHeapSize", "testtable")
            assert jmx.read_attribute(table_view_read_time, "Count") is not None, \
                missing_metric_message.format("ViewReadTime", "testtable")
            assert jmx.read_attribute(table_view_lock_time, "Count") is not None, \
                missing_metric_message.format("ViewLockAcquireTime", "testtable")
            assert jmx.read_attribute(mv_memtable_size, "Value") is not None, \
                missing_metric_message.format("AllMemtablesHeapSize", "testmv")
            with pytest.raises(Exception, match=".*InstanceNotFoundException.*"):
                jmx.read_attribute(mbean=mv_view_read_time, attribute="Count", verbose=False)
            with pytest.raises(Exception, match=".*InstanceNotFoundException.*"):
                jmx.read_attribute(mbean=mv_view_lock_time, attribute="Count", verbose=False)

        node.run_cqlsh(cmds="DROP KEYSPACE mvtest;")
        with JolokiaAgent(node) as jmx:
            with pytest.raises(Exception, match=".*InstanceNotFoundException.*"):
                jmx.read_attribute(mbean=table_memtable_size, attribute="Value", verbose=False)
            with pytest.raises(Exception, match=".*InstanceNotFoundException.*"):
                jmx.read_attribute(mbean=table_view_lock_time, attribute="Count", verbose=False)
            with pytest.raises(Exception, match=".*InstanceNotFoundException.*"):
                jmx.read_attribute(mbean=table_view_read_time, attribute="Count", verbose=False)
            with pytest.raises(Exception, match=".*InstanceNotFoundException.*"):
                jmx.read_attribute(mbean=mv_memtable_size, attribute="Value", verbose=False)
            with pytest.raises(Exception, match=".*InstanceNotFoundException.*"):
                jmx.read_attribute(mbean=mv_view_lock_time, attribute="Count", verbose=False)
            with pytest.raises(Exception, match=".*InstanceNotFoundException.*"):
                jmx.read_attribute(mbean=mv_view_read_time, attribute="Count", verbose=False)

    def test_compactionstats(self):
        """
        @jira_ticket CASSANDRA-10504
        @jira_ticket CASSANDRA-10427

        Test that jmx MBean used by nodetool compactionstats
        properly updates the progress of a compaction
        """

        cluster = self.cluster
        cluster.populate(1)
        node = cluster.nodelist()[0]
        cluster.start()

        # Run a quick stress command to create the keyspace and table
        node.stress(['write', 'n=1', 'no-warmup'])
        # Disable compaction on the table
        node.nodetool('disableautocompaction keyspace1 standard1')
        node.nodetool('setcompactionthroughput 1')
        node.stress(['write', 'n=150K', 'no-warmup'])
        node.flush()
        # Run a major compaction. This will be the compaction whose
        # progress we track.
        node.nodetool_process('compact')
        # We need to sleep here to give compaction time to start
        # Why not do something smarter? Because if the bug regresses,
        # we can't rely on jmx to tell us that compaction started.
        time.sleep(5)

        compaction_manager = make_mbean('db', type='CompactionManager')
        with JolokiaAgent(node) as jmx:
            progress_string = jmx.read_attribute(compaction_manager, 'CompactionSummary')[0]

            # Pause in between reads
            # to allow compaction to move forward
            time.sleep(2)

            updated_progress_string = jmx.read_attribute(compaction_manager, 'CompactionSummary')[0]
            var = 'Compaction@{uuid}(keyspace1, standard1, {progress}/{total})bytes'
            if self.cluster.version() >= LooseVersion('4.0'): # CASSANDRA-15954
                var = 'Compaction({taskUuid}, {progress} / {total} bytes)@{uuid}(keyspace1, standard1)'
            progress = int(parse.search(var, progress_string).named['progress'])
            updated_progress = int(parse.search(var, updated_progress_string).named['progress'])

            logger.debug(progress_string)
            logger.debug(updated_progress_string)

            # We want to make sure that the progress is increasing,
            # and that values other than zero are displayed.
            assert updated_progress > progress
            assert progress >= 0
            assert updated_progress > 0

            # Block until the major compaction is complete
            # Otherwise nodetool will throw an exception
            # Give a timeout, in case compaction is broken
            # and never ends.
            start = time.time()
            max_query_timeout = 600
            logger.debug("Waiting for compaction to finish:")
            while (len(jmx.read_attribute(compaction_manager, 'CompactionSummary')) > 0) and (
                    time.time() - start < max_query_timeout):
                logger.debug(jmx.read_attribute(compaction_manager, 'CompactionSummary'))
                time.sleep(2)

    @since('2.2')
    def test_phi(self):
        """
        Check functioning of nodetool failuredetector.
        @jira_ticket CASSANDRA-9526
        """
        cluster = self.cluster
        cluster.populate(3).start()
        node1, node2, node3 = cluster.nodelist()

        stdout = node1.nodetool("failuredetector").stdout
        phivalues = stdout.splitlines()
        endpoint1values = phivalues[1].split()
        endpoint2values = phivalues[2].split()

        endpoint1 = endpoint1values[0][1:-1]
        endpoint2 = endpoint2values[0][1:-1]

        assert '127.0.0.2' in [endpoint1, endpoint2]
        assert '127.0.0.3' in [endpoint1, endpoint2]

        endpoint1phi = float(endpoint1values[1])
        endpoint2phi = float(endpoint2values[1])

        max_phi = 2.0
        assert endpoint1phi > 0.0
        assert endpoint1phi < max_phi

        assert endpoint2phi > 0.0
        assert endpoint2phi < max_phi

    @since('4.0')
    def test_set_get_batchlog_replay_throttle(self):
        """
        @jira_ticket CASSANDRA-13614

        Test that batchlog replay throttle can be set and get through JMX
        """
        cluster = self.cluster
        cluster.populate(2)
        node = cluster.nodelist()[0]
        cluster.start()

        # Set and get throttle with JMX, ensuring that the rate change is logged
        with JolokiaAgent(node) as jmx:
            mbean = make_mbean('db', 'StorageService')
            jmx.write_attribute(mbean, 'BatchlogReplayThrottleInKB', 4096)
            assert len(node.grep_log('Updating batchlog replay throttle to 4096 KB/s, 2048 KB/s per endpoint',
                                     filename='debug.log')) > 0
            assert 4096 == jmx.read_attribute(mbean, 'BatchlogReplayThrottleInKB')

    def test_bloom_filter_false_ratio(self):
        """
        Test for CASSANDRA-15834

        Verifies if BloomFilterFalseRatio takes into account true negatives. Without this fix, the following
        scenario (many reads for non-existing rows) would yield BloomFilterFalseRatio=1.0. With the fix we assume
        it should be less then the default bloom_filter_fp_chance.
        """
        cluster = self.cluster
        cluster.populate(1)
        node = cluster.nodelist()[0]
        cluster.start(wait_for_binary_proto=True)

        session = self.patient_exclusive_cql_connection(node)

        keyspace = 'bloom_ratio_test_ks'
        create_ks(session, keyspace, 1)
        create_c1c2_table(self, session)
        insert_c1c2(session, n=10)
        node.nodetool("flush " + keyspace)

        for key in range(10000):
            session.execute("SELECT * from cf where key = '{0}'".format(key))

        bloom_filter_false_ratios = [
            make_mbean('metrics', type='Table', name='RecentBloomFilterFalseRatio'),
            make_mbean('metrics', type='Table', keyspace=keyspace, scope='cf', name='BloomFilterFalseRatio'),
            make_mbean('metrics', type='Table', name='BloomFilterFalseRatio'),
            make_mbean('metrics', type='Table', keyspace=keyspace, scope='cf', name='RecentBloomFilterFalseRatio'),
        ]

        with JolokiaAgent(node) as jmx:
            for metric in bloom_filter_false_ratios:
                ratio = jmx.read_attribute(metric, "Value")
                # Bloom filter false positive ratio should not be greater than the default bloom_filter_fp_chance.
                assert ratio < 0.01


@since('3.9')
class TestJMXSSL(Tester):
    keystore_password = 'cassandra'
    truststore_password = 'cassandra'

    def truststore(self):
        return os.path.join(self.fixture_dtest_setup.test_path, 'truststore.jks')

    def keystore(self):
        return os.path.join(self.fixture_dtest_setup.test_path, 'keystore.jks')

    def test_jmx_connection(self):
        """
        Check connecting with a JMX client (via nodetool) where SSL is enabled for JMX
        @jira_ticket CASSANDRA-12109
        """
        cluster = self._populateCluster(require_client_auth=False)
        node = cluster.nodelist()[0]
        cluster.start()

        self.assert_insecure_connection_rejected(node)

        node.nodetool("info --ssl -Djavax.net.ssl.trustStore={ts} -Djavax.net.ssl.trustStorePassword={ts_pwd}"
                      .format(ts=self.truststore(), ts_pwd=self.truststore_password))

    def test_require_client_auth(self):
        """
        Check connecting with a JMX client (via nodetool) where SSL is enabled and
        client certificate auth is also configured
        @jira_ticket CASSANDRA-12109
        """
        cluster = self._populateCluster(require_client_auth=True)
        node = cluster.nodelist()[0]
        cluster.start()

        self.assert_insecure_connection_rejected(node)

        # specifying only the truststore containing the server cert should fail
        with pytest.raises(ToolError):
            node.nodetool("info --ssl -Djavax.net.ssl.trustStore={ts} -Djavax.net.ssl.trustStorePassword={ts_pwd}"
                          .format(ts=self.truststore(), ts_pwd=self.truststore_password))

        # when both truststore and a keystore containing the client key are supplied, connection should succeed
        node.nodetool(
            "info --ssl -Djavax.net.ssl.trustStore={ts} -Djavax.net.ssl.trustStorePassword={ts_pwd} -Djavax.net.ssl.keyStore={ks} -Djavax.net.ssl.keyStorePassword={ks_pwd}"
            .format(ts=self.truststore(), ts_pwd=self.truststore_password, ks=self.keystore(),
                    ks_pwd=self.keystore_password))

    def assert_insecure_connection_rejected(self, node):
        """
        Attempts to connect to JMX (via nodetool) without any client side ssl parameters, expecting failure
        """
        with pytest.raises(ToolError):
            node.nodetool("info")

    def _populateCluster(self, require_client_auth=False):
        cluster = self.cluster
        cluster.populate(1)

        generate_ssl_stores(self.fixture_dtest_setup.test_path)
        if require_client_auth:
            ts = self.truststore()
            ts_pwd = self.truststore_password
        else:
            ts = None
            ts_pwd = None

        enable_jmx_ssl(cluster.nodelist()[0],
                       require_client_auth=require_client_auth,
                       keystore=self.keystore(),
                       keystore_password=self.keystore_password,
                       truststore=ts,
                       truststore_password=ts_pwd)
        return cluster