test: test zero-token nodes

We add tests to verify the basic properties of zero-token nodes.

`test_zero_token_nodes_no_replication` and
`test_not_enough_token_owners` are more or less deterministic tests.
Running them only in the dev mode is sufficient.

`test_zero_token_nodes_topology_ops` is quite slow, as expected,
considering parameterization and the number of topology operations.
In the future we can think of making it faster or skipping in the
debug mode. For now, our priority is to test zero-token nodes
thoroughly.

test/topology_custom/suite.yaml

@@ -10,6 +10,7 @@ run_first:
   - test_raft_recovery_basic
   - test_group0_schema_versioning
   - test_tablets_migration
+  - test_zero_token_nodes_topology_ops
 skip_in_debug:
   - test_shutdown_hang
   - test_replace
@@ -19,3 +20,5 @@ run_in_dev:
   - test_group0_schema_versioning
   - test_different_group0_ids
   - test_replace_ignore_nodes
+  - test_zero_token_nodes_no_replication
+  - test_not_enough_token_owners

test/topology_custom/test_zero_token_nodes_topology_ops.py

@@ -0,0 +1,110 @@
+#
+# Copyright (C) 2024-present ScyllaDB
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+import pytest
+import logging
+import time
+
+from cassandra.cluster import ConsistencyLevel
+
+from test.pylib.manager_client import ManagerClient
+from test.pylib.scylla_cluster import ReplaceConfig
+from test.pylib.util import wait_for_cql_and_get_hosts
+from test.topology.util import check_node_log_for_failed_mutations, start_writes
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('tablets_enabled', [True, False])
+async def test_zero_token_nodes_topology_ops(manager: ManagerClient, tablets_enabled: bool):
+    """
+    Test that:
+    - adding a zero-token node in the gossip-based topology fails
+    - topology operations in the Raft-based topology involving zero-token nodes succeed
+    - client requests to normal nodes in the presence of zero-token nodes (2 normal nodes, RF=2, CL=2) succeed
+    """
+    logging.info('Trying to add a zero-token server in the gossip-based topology')
+    await manager.server_add(config={'join_ring': False, 'force_gossip_topology_changes': True},
+                             expected_error='the raft-based topology is disabled')
+
+    normal_cfg = {'enable_tablets': tablets_enabled}
+    zero_token_cfg = {'enable_tablets': tablets_enabled, 'join_ring': False}
+
+    logging.info('Adding the first server')
+    server_a = await manager.server_add(config=normal_cfg)
+
+    logging.info('Adding the second server as zero-token')
+    server_b = await manager.server_add(config=zero_token_cfg)
+
+    logging.info('Adding the third server')
+    server_c = await manager.server_add(config=normal_cfg)
+
+    await wait_for_cql_and_get_hosts(manager.cql, [server_a, server_c], time.time() + 60)
+    finish_writes = await start_writes(manager.cql, 2, ConsistencyLevel.TWO)
+
+    logging.info('Adding the fourth server as zero-token')
+    await manager.server_add(config=zero_token_cfg)  # Necessary to preserve the Raft majority.
+
+    logging.info(f'Restarting {server_b}')
+    await manager.server_stop_gracefully(server_b.server_id)
+    await manager.server_start(server_b.server_id)
+
+    logging.info(f'Stopping {server_b}')
+    await manager.server_stop_gracefully(server_b.server_id)
+
+    replace_cfg_b = ReplaceConfig(replaced_id=server_b.server_id, reuse_ip_addr=False, use_host_id=False)
+    logging.info(f'Trying to replace {server_b} with a token-owing server')
+    await manager.server_add(replace_cfg_b, config=normal_cfg, expected_error='Cannot replace the zero-token node')
+
+    logging.info(f'Replacing {server_b}')
+    server_b = await manager.server_add(replace_cfg_b, config=zero_token_cfg)
+
+    logging.info(f'Stopping {server_b}')
+    await manager.server_stop_gracefully(server_b.server_id)
+
+    replace_cfg_b = ReplaceConfig(replaced_id=server_b.server_id, reuse_ip_addr=True, use_host_id=False)
+    logging.info(f'Replacing {server_b} with the same IP')
+    server_b = await manager.server_add(replace_cfg_b, config=zero_token_cfg)
+
+    logging.info(f'Decommissioning {server_b}')
+    await manager.decommission_node(server_b.server_id)
+
+    logging.info('Adding two zero-token servers')
+    [server_b, server_d] = await manager.servers_add(2, config=zero_token_cfg)
+
+    logging.info(f'Rebuilding {server_b}')
+    await manager.rebuild_node(server_b.server_id)
+
+    logging.info(f'Stopping {server_b}')
+    await manager.server_stop_gracefully(server_b.server_id)
+
+    logging.info(f'Stopping {server_d}')
+    await manager.server_stop_gracefully(server_d.server_id)
+
+    logging.info(f'Initiating removenode of {server_b} by {server_a}, ignore_dead={[server_d.ip_addr]}')
+    await manager.remove_node(server_a.server_id, server_b.server_id, [server_d.ip_addr])
+
+    logging.info(f'Initiating removenode of {server_d} by {server_a}')
+    await manager.remove_node(server_a.server_id, server_d.server_id)
+
+    logging.info('Adding a zero-token server')
+    await manager.server_add(config=zero_token_cfg)
+
+    # FIXME: Finish writes after the last server_add call once scylladb/scylladb#19737 is fixed.
+    logging.info('Checking results of the background writes')
+    await finish_writes()
+
+    logging.info('Adding a normal server')
+    server_e = await manager.server_add(config=normal_cfg)
+
+    logging.info(f'Stopping {server_e}')
+    await manager.server_stop_gracefully(server_e.server_id)
+
+    replace_cfg_e = ReplaceConfig(replaced_id=server_e.server_id, reuse_ip_addr=False, use_host_id=False)
+    logging.info(f'Trying to replace {server_e} with a zero-token server')
+    await manager.server_add(replace_cfg_e, config=zero_token_cfg,
+                             expected_error='Cannot replace the token-owning node')
+
+    await check_node_log_for_failed_mutations(manager, server_a)
+    await check_node_log_for_failed_mutations(manager, server_c)

test/topology_experimental_raft/test_not_enough_token_owners.py

@@ -0,0 +1,71 @@
+#
+# Copyright (C) 2024-present ScyllaDB
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+import pytest
+import logging
+import time
+
+from test.pylib.manager_client import ManagerClient
+from test.pylib.util import unique_name, wait_for_cql_and_get_hosts
+
+
+@pytest.mark.asyncio
+async def test_not_enough_token_owners(manager: ManagerClient):
+    """
+    Test that:
+    - the first node in the cluster cannot be a zero-token node
+    - removenode and decommission of the only token owner fail in the presence of zero-token nodes
+    - removenode and decommission of a token owner fail in the presence of zero-token nodes if the number of token
+      owners would fall below the RF of some keyspace using tablets
+    """
+    logging.info('Trying to add a zero-token server as the first server in the cluster')
+    await manager.server_add(config={'join_ring': False},
+                             expected_error='Cannot start the first node in the cluster as zero-token')
+
+    logging.info('Adding the first server')
+    server_a = await manager.server_add()
+
+    logging.info('Adding two zero-token servers')
+    # The second server is needed only to preserve the Raft majority.
+    server_b = (await manager.servers_add(2, config={'join_ring': False}))[0]
+
+    logging.info(f'Trying to decommission the only token owner {server_a}')
+    await manager.decommission_node(server_a.server_id,
+                                    expected_error='Cannot decommission the last token-owning node in the cluster')
+
+    logging.info(f'Stopping {server_a}')
+    await manager.server_stop_gracefully(server_a.server_id)
+
+    logging.info(f'Trying to remove the only token owner {server_a} by {server_b}')
+    await manager.remove_node(server_b.server_id, server_a.server_id,
+                              expected_error='cannot be removed because it is the last token-owning node in the cluster')
+
+    logging.info(f'Starting {server_a}')
+    await manager.server_start(server_a.server_id)
+
+    logging.info('Adding a normal server')
+    await manager.server_add()
+
+    cql = manager.get_cql()
+
+    await wait_for_cql_and_get_hosts(cql, [server_a], time.time() + 60)
+
+    ks_name = unique_name()
+    await cql.run_async(f"""CREATE KEYSPACE {ks_name} WITH replication = {{'class': 'NetworkTopologyStrategy',
+                             'replication_factor': 2}} AND tablets = {{ 'enabled': true }}""")
+    await cql.run_async(f'CREATE TABLE {ks_name}.tbl (pk int PRIMARY KEY, v int)')
+    await cql.run_async(f'INSERT INTO {ks_name}.tbl (pk, v) VALUES (1, 1)')
+
+    # FIXME: Once scylladb/scylladb#16195 is fixed, we will have to replace the expected error message.
+    # A similar change may be needed for remove_node below.
+    logging.info(f'Trying to decommission {server_a} - one of the two token owners')
+    await manager.decommission_node(server_a.server_id, expected_error='Unable to find new replica for tablet')
+
+    logging.info(f'Stopping {server_a}')
+    await manager.server_stop_gracefully(server_a.server_id)
+
+    logging.info(f'Trying to remove {server_a}, one of the two token owners, by {server_b}')
+    await manager.remove_node(server_b.server_id, server_a.server_id,
+                              expected_error='Unable to find new replica for tablet')

test/topology_experimental_raft/test_zero_token_nodes_multidc.py

@@ -0,0 +1,87 @@
+#
+# Copyright (C) 2024-present ScyllaDB
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+import logging
+import pytest
+
+from cassandra import ConsistencyLevel
+from cassandra.policies import WhiteListRoundRobinPolicy
+from cassandra.query import SimpleStatement
+from test.pylib.manager_client import ManagerClient
+
+from test.pylib.util import unique_name
+from test.topology.conftest import cluster_con
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('zero_token_nodes', [1, 2])
+async def test_zero_token_nodes_multidc_basic(manager: ManagerClient, zero_token_nodes: int):
+    """
+    Test the basic functionality of a DC with zero-token nodes:
+    - adding zero-token nodes to a new DC succeeds
+    - with tablets, ensuring enough replicas for tables depends on the number of token-owners in a DC, not all nodes
+    - client requests in the presence of zero-token nodes succeed (also when zero-token nodes coordinate)
+    """
+    normal_cfg = {'endpoint_snitch': 'GossipingPropertyFileSnitch'}
+    zero_token_cfg = {'endpoint_snitch': 'GossipingPropertyFileSnitch', 'join_ring': False}
+    property_file_dc1 = {'dc': 'dc1', 'rack': 'rack'}
+    property_file_dc2 = {'dc': 'dc2', 'rack': 'rack'}
+
+    logging.info('Creating dc1 with 2 token-owning nodes')
+    servers = await manager.servers_add(2, config=normal_cfg, property_file=property_file_dc1)
+
+    normal_nodes_in_dc2 = 2 - zero_token_nodes
+    logging.info(f'Creating dc2 with {normal_nodes_in_dc2} token-owning and {zero_token_nodes} zero-token nodes')
+    servers += await manager.servers_add(zero_token_nodes, config=zero_token_cfg, property_file=property_file_dc2)
+    if zero_token_nodes == 1:
+        servers.append(await manager.server_add(config=normal_cfg, property_file=property_file_dc2))
+
+    logging.info('Creating connections to dc1 and dc2')
+    dc1_cql = cluster_con([servers[0].ip_addr], 9042, False,
+                          load_balancing_policy=WhiteListRoundRobinPolicy([servers[0].ip_addr])).connect()
+    dc2_cql = cluster_con([servers[2].ip_addr], 9042, False,
+                          load_balancing_policy=WhiteListRoundRobinPolicy([servers[2].ip_addr])).connect()
+
+    ks_names = list[str]()
+    logging.info('Trying to create tables for different replication factors')
+    for rf in range(3):
+        ks_names.append(unique_name())
+        failed = False
+        await dc2_cql.run_async(f"""CREATE KEYSPACE {ks_names[rf]} WITH replication =
+                                     {{'class': 'NetworkTopologyStrategy', 'replication_factor': 2, 'dc2': {rf}}}
+                                     AND tablets = {{ 'enabled': true }}""")
+        try:
+            await dc2_cql.run_async(f'CREATE TABLE {ks_names[rf]}.tbl (pk int PRIMARY KEY, v int)')
+        except Exception:
+            failed = True
+        assert failed == (rf > normal_nodes_in_dc2)
+
+    logging.info('Sending requests with different consistency levels')
+    for rf in range(normal_nodes_in_dc2 + 1):
+        # FIXME: we may add LOCAL_QUORUM to the list below once scylladb/scylladb#20028 is fixed.
+        cls = [
+            ConsistencyLevel.ONE,
+            ConsistencyLevel.TWO,
+            ConsistencyLevel.QUORUM,
+            ConsistencyLevel.EACH_QUORUM,
+            ConsistencyLevel.ALL
+        ]
+        if rf == 1:
+            cls.append(ConsistencyLevel.LOCAL_ONE)
+
+        for cl in cls:
+            insert_query = SimpleStatement(f'INSERT INTO {ks_names[rf]}.tbl (pk, v) VALUES (1, 1)',
+                                           consistency_level=cl)
+            await dc1_cql.run_async(insert_query)
+            await dc2_cql.run_async(insert_query)
+
+            if cl == ConsistencyLevel.EACH_QUORUM:
+                continue  # EACH_QUORUM is supported only for writes
+
+            select_query = SimpleStatement(f'SELECT * FROM {ks_names[rf]}.tbl', consistency_level=cl)
+            dc1_result = list((await dc1_cql.run_async(select_query))[0])
+            dc2_result = list((await dc2_cql.run_async(select_query))[0])
+            assert dc1_result == [1, 1]
+            assert dc2_result == [1, 1]

test/topology_experimental_raft/test_zero_token_nodes_no_replication.py

@@ -0,0 +1,71 @@
+#
+# Copyright (C) 2024-present ScyllaDB
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+import pytest
+import logging
+
+from cassandra.cluster import ConsistencyLevel
+from cassandra.policies import WhiteListRoundRobinPolicy
+from cassandra.query import SimpleStatement
+
+from test.pylib.manager_client import ManagerClient
+from test.pylib.util import unique_name
+from test.topology.conftest import cluster_con
+
+
+@pytest.mark.asyncio
+async def test_zero_token_nodes_no_replication(manager: ManagerClient):
+    """
+    Test that zero-token nodes aren't replicas in all non-local replication strategies with and without tablets.
+    """
+    logging.info('Adding the first server')
+    server_a = await manager.server_add()
+    logging.info('Adding the second server as zero-token')
+    server_b = await manager.server_add(config={'join_ring': False})
+    logging.info('Adding the third server')
+    await manager.server_add()
+
+    logging.info(f'Initiating connections to {server_a} and {server_b}')
+    cql_a = cluster_con([server_a.ip_addr], 9042, False,
+                        load_balancing_policy=WhiteListRoundRobinPolicy([server_a.ip_addr])).connect()
+    cql_b = cluster_con([server_b.ip_addr], 9042, False,
+                        load_balancing_policy=WhiteListRoundRobinPolicy([server_b.ip_addr])).connect()
+
+    logging.info('Creating tables for each replication strategy and tablets combination')
+    ks_names = list[str]()
+    for replication_strategy in ['EverywhereStrategy', 'SimpleStrategy', 'NetworkTopologyStrategy']:
+        for tablets_enabled in [True, False]:
+            if tablets_enabled and replication_strategy != 'NetworkTopologyStrategy':
+                continue
+
+            ks_name = unique_name()
+            ks_names.append(ks_name)
+            await cql_b.run_async(f"""CREATE KEYSPACE {ks_name} WITH replication =
+                                    {{'class': '{replication_strategy}', 'replication_factor': 2}}
+                                    AND tablets = {{ 'enabled': {str(tablets_enabled).lower()} }}""")
+            await cql_b.run_async(f'CREATE TABLE {ks_name}.tbl (pk int PRIMARY KEY, v int)')
+            for i in range(100):
+                insert_query = f'INSERT INTO {ks_name}.tbl (pk, v) VALUES ({i}, {i})'
+                if i % 2 == 0:
+                    await cql_a.run_async(insert_query)
+                else:
+                    await cql_b.run_async(insert_query)  # Zero-token nodes should be able to coordinate requests.
+
+    select_queries = {ks_name: SimpleStatement(f'SELECT * FROM {ks_name}.tbl', consistency_level=ConsistencyLevel.ALL)
+                      for ks_name in ks_names}
+
+    for ks_name in ks_names:
+        result1 = [(row.pk, row.v) for row in await cql_b.run_async(select_queries[ks_name])]
+        result1.sort()
+        assert result1 == [(i, i) for i in range(100)]
+
+    logging.info(f'Stopping {server_b}')
+    await manager.server_stop_gracefully(server_b.server_id)
+
+    # If server_b was a replica of some token range or tablet, reads with CL=ALL would fail.
+    for ks_name in ks_names:
+        result2 = [(row.pk, row.v) for row in await cql_a.run_async(select_queries[ks_name])]
+        result2.sort()
+        assert result2 == [(i, i) for i in range(100)]