test_dynamo.py

import time
from subprocess import run

import boto3
from cassandra import ConsistencyLevel
from cassandra.query import SimpleStatement

from dtest import Tester, debug
from scylla_tools import insert_c1c2


class Test(Tester):

    def test_dynamo_issue_5802(self):
        debug('Populate')
        cluster = self.cluster
        cluster.set_configuration_options({'start_native_transport': True})
        cluster.set_configuration_options({'experimental': True, 'alternator_port': 8080})
        cluster.populate([6])
        debug('start')
        cluster.start(wait_for_binary_proto=True, wait_other_notice=True)
        node1, node2, node3, node4, node5, node6 = cluster.nodelist()
        node1_ip_address = self.get_ip_from_node(node1)
        node5_ip_address = self.get_ip_from_node(node5)

        dynamodb = boto3.resource('dynamodb', endpoint_url=f'http://{node1_ip_address}:8080',
                                  region_name='None', aws_access_key_id='None', aws_secret_access_key='None')

        dynamodb2 = boto3.resource('dynamodb', endpoint_url=f'http://{node5_ip_address}:8080',
                                  region_name='None', aws_access_key_id='None', aws_secret_access_key='None')

        dynamodb.create_table(
            AttributeDefinitions=[
                {
                    'AttributeName': 'key',
                    'AttributeType': 'S'
                },
            ],
            BillingMode='PAY_PER_REQUEST',
            TableName='usertable',
            KeySchema=[
                {
                    'AttributeName': 'key',
                    'KeyType': 'HASH'
                },
            ])

        dynamodb.batch_write_item(RequestItems={
            'usertable': [
                {
                    'PutRequest': {
                        'Item': {
                            'key': 'test', 'x': {'hello': 'world'}
                        }
                    },
                },
                {
                    'PutRequest': {
                        'Item': {
                            'key': 'test1', 'x': {'hello': 'world'}
                        }
                    },
                },
                {
                    'PutRequest': {
                        'Item': {
                            'key': 'test2', 'x': {'hello': 'world'}
                        }
                    },
                }
            ]
        })

        table = dynamodb.Table('usertable')

        table2 = dynamodb2.Table('usertable')

        print(table.get_item(
            ConsistentRead=True,
            Key={'key': 'test'}))

        # node1.stop(wait_other_notice=True)
        # node1.start(wait_other_notice=True, wait_for_binary_proto=True)

        time.sleep(20)

        def remove_iptables():
            run(
                f'sudo iptables -t filter -D INPUT -p tcp --dport 7000 --destination {node1_ip_address}/32 -j REJECT --reject-with icmp-port-unreachable'.split(
                    ' '))
            run(
                f'sudo iptables -t filter -D INPUT -p tcp --dport 7001 --destination {node1_ip_address}/32 -j REJECT --reject-with icmp-port-unreachable'.split(
                    ' '))

        self.addCleanup(remove_iptables)

        run(
            f'sudo iptables -t filter -A INPUT -p tcp --dport 7000 --destination {node1_ip_address}/32 -j REJECT --reject-with icmp-port-unreachable'.split(
                ' '))
        run(
            f'sudo iptables -t filter -A INPUT -p tcp --dport 7001 --destination {node1_ip_address}/32 -j REJECT --reject-with icmp-port-unreachable'.split(
                ' '))

        time.sleep(25)

        for i in range(5):
            print(table2.get_item(
                ConsistentRead=False,
                Key={'key': 'test'}))

        for i in range(5):
            print(table.get_item(
                ConsistentRead=False,
                Key={'key': 'test2'}))

        debug('Test finished')

    def test_cql_issue_5802(self):
        debug('Populate')
        cluster = self.cluster
        cluster.set_configuration_options({'start_native_transport': True})
        cluster.set_configuration_options({'experimental': True, 'alternator_port': 8080})
        cluster.populate([6])
        debug('start')
        cluster.start(wait_for_binary_proto=True, wait_other_notice=True)
        node1, node2, node3, node4, node5, node6 = cluster.nodelist()
        node1_ip_address = self.get_ip_from_node(node1)

        session1 = self.patient_cql_connection(node1)
        session5 = self.patient_cql_connection(node5)
        self.create_ks(session1, 'ks', 3)
        self.create_cf(session1, 'cf', columns={'c1': 'text', 'c2': 'text'})

        num_keys = 10
        c1_values = list(map(lambda x: '{}'.format(x), range(num_keys)))
        c2_values = list(map(lambda x: '{}'.format(x), range(num_keys, 2 * num_keys)))
        keys = list(range(num_keys))

        insert_c1c2(session1, keys=keys, consistency=ConsistencyLevel.LOCAL_QUORUM,
                    c1_values=c1_values, c2_values=c2_values)

        print(list(session1.execute(SimpleStatement("SELECT * FROM ks.cf LIMIT 5"))))

        time.sleep(20)

        def remove_iptables():
            run(
                f'sudo iptables -t filter -D INPUT -p tcp --dport 7000 --destination {node1_ip_address}/32 -j REJECT --reject-with icmp-port-unreachable'.split(
                    ' '))
            run(
                f'sudo iptables -t filter -D INPUT -p tcp --dport 7001 --destination {node1_ip_address}/32 -j REJECT --reject-with icmp-port-unreachable'.split(
                    ' '))

        self.addCleanup(remove_iptables)

        run(
            f'sudo iptables -t filter -A INPUT -p tcp --dport 7000 --destination {node1_ip_address}/32 -j REJECT --reject-with icmp-port-unreachable'.split(
                ' '))
        run(
            f'sudo iptables -t filter -A INPUT -p tcp --dport 7001 --destination {node1_ip_address}/32 -j REJECT --reject-with icmp-port-unreachable'.split(
                ' '))

        time.sleep(25)

        for i in range(5):
            print(list(session5.execute(SimpleStatement("SELECT * FROM ks.cf WHERE key='k1'"))))

        for i in range(5):
            print(list(session1.execute(SimpleStatement("SELECT * FROM ks.cf WHERE key='k1'"))))

        debug('Test finished')