From 8d3a35fbf80a55157c4e44a254e099d19c15cee1 Mon Sep 17 00:00:00 2001 From: Maksim Davydov Date: Thu, 17 Sep 2026 14:12:17 +0300 Subject: [PATCH 1/8] IGNITE-29049 [ducktests] Restart-safe JMX client and exact metric registry lookup Metric registry MBeans were looked up with ad-hoc patterns at every call site, and the node's JMX client was memoized for the whole life of the node object. * metric_registry_pattern(registry) builds the exact MBean name pattern of a registry: split at the first dot like JmxMetricExporterSpi, values spelled like IgniteUtils.escapeObjectNameValue(), escaped for grep -E, anchored at the end of the line ('name' sorts last). A registry without a group is filtered from the system view of the same name. The old 'group=cacheGroups.*name="%s"' pattern could not find an alphanumeric (unquoted) name at all. Names with a quote, a backslash or '?' are rejected, as the jmxterm command quoting cannot carry them. * The node's JmxClient is rebuilt on every start_node(). Before, a restarted node kept the pid and the clsLdr-bearing MBean names of its previous JVM, so callers had to build a client of their own. JmxClient takes an explicit pid for the CDC JVM. * node.metric_registry_mbean() and JmxMBean.value()/bool_value() replace the hand-built patterns and next(mbean.X) reads in rebalance, dump, snapshot and CDC. check_jmx_utils.py runs every pattern case through Python re and grep -E. --- .../tests/checks/utils/check_jmx_utils.py | 124 +++++++++++++ .../services/utils/cdc/cdc_helper.py | 8 +- .../services/utils/control_utility.py | 13 +- .../ignitetest/services/utils/ignite_aware.py | 6 +- .../ignitetest/services/utils/jmx_utils.py | 166 +++++++++++++++--- .../tests/ignitetest/tests/dump_test.py | 4 +- .../tests/ignitetest/tests/rebalance/util.py | 8 +- 7 files changed, 282 insertions(+), 47 deletions(-) create mode 100644 modules/ducktests/tests/checks/utils/check_jmx_utils.py diff --git a/modules/ducktests/tests/checks/utils/check_jmx_utils.py b/modules/ducktests/tests/checks/utils/check_jmx_utils.py new file mode 100644 index 0000000000000..42c24ee3798e1 --- /dev/null +++ b/modules/ducktests/tests/checks/utils/check_jmx_utils.py @@ -0,0 +1,124 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Checks the MBean name pattern of a metric registry. + +The pattern is grepped against the MBean list of a node, so it is checked here the same way: +the match has to cover the WHOLE object name, since that name is what the following +'get -b ' is issued against. Every case runs through Python's re and, where a grep is +installed, through 'grep -E' itself - the pattern is an ERE, not a Python regular expression. +""" +import re +import shutil +import subprocess + +import pytest + +from ignitetest.services.utils.jmx_utils import metric_registry_pattern + +# MBean names as a node prints them: the properties are ordered alphabetically, which puts +# the instance name between the group and the name. +PREFIX = "org.apache:clsLdr=3e2a56ab,group=%s,igniteInstanceName=ducker01,name=%s" +NO_GROUP_PREFIX = "org.apache:clsLdr=3e2a56ab,igniteInstanceName=ducker01,name=%s" + +GREP = shutil.which("grep") + + +def python_grep(pattern, negative_pattern, mbean_name): + """ + :return: The part of the name the patterns leave, as 'grep -E -o | grep -E -v' would return it. + """ + match = re.search(pattern, mbean_name) + + if not match or (negative_pattern and re.search(negative_pattern, match.group())): + return None + + return match.group() + + +def real_grep(tmp_path, pattern, negative_pattern, mbean_name): + """ + :return: The same as :func:`python_grep`, but produced by 'grep -E' itself. The patterns go + through files, since a quote in a command line argument does not survive Windows. + """ + def grep(flag, grep_pattern, text): + pattern_file = tmp_path / f"pattern{flag}" + pattern_file.write_bytes((grep_pattern + "\n").encode()) + + return subprocess.run([GREP, "-E", flag, "-f", str(pattern_file)], input=text.encode(), + capture_output=True).stdout.decode() + + out = grep("-o", pattern, mbean_name + "\n") + + if out and negative_pattern: + out = grep("-v", negative_pattern, out) + + return out.strip() or None + + +# (registry, MBean name, whether the registry's pattern must match that name) +CASES = [ + # A purely alphanumeric name is registered without quotes. + ("cache.myCache", PREFIX % ("cache", "myCache"), True), + ("cache.cache_1", PREFIX % ("cache", "cache_1"), True), + # Anything else - a dash is enough - is registered quoted, closing quote included. + ("cache.my-cache", PREFIX % ("cache", '"my-cache"'), True), + ("cache.mdc-demo-backup-filter", PREFIX % ("cache", '"mdc-demo-backup-filter"'), True), + # Only the FIRST dot splits the registry, so the rest of it is the MBean name. + ("io.dataregion.default", PREFIX % ("io", '"dataregion.default"'), True), + # 'cache' must not pick up 'cacheGroups': the two hold different metrics of the same cache. + ("cache.my-cache", PREFIX % ("cacheGroups", '"my-cache"'), False), + ("cacheGroups.my-cache", PREFIX % ("cacheGroups", '"my-cache"'), True), + # 'name' sorts last, so the pattern ends at the end of the line and a longer name does not answer. + ("cache.myCache", PREFIX % ("cache", "myCacheV2"), False), + ("cache.my-cache", PREFIX % ("cache", '"my-cache-2"'), False), + # The name is matched literally, ERE metacharacters included. + ("cache.my.cache", PREFIX % ("cache", '"myXcache"'), False), + ("cache.my.cache", PREFIX % ("cache", '"my.cache"'), True), + ("cache.a(b)+c", PREFIX % ("cache", '"a(b)+c"'), True), + ("cache.a(b)+c", PREFIX % ("cache", '"abbc"'), False), + # Characters that are special only right after an opening '[' or '{' stay unescaped. + ("cache.a]b}c", PREFIX % ("cache", '"a]b}c"'), True), + ("cache.a[b]{2}", PREFIX % ("cache", '"a[b]{2}"'), True), + ("cache.a[b]{2}", PREFIX % ("cache", '"abb"'), False), + # The exporter escapes '*' inside the quotes. + ("cache.a*b", PREFIX % ("cache", r'"a\*b"'), True), + # A registry without a dot has no group, and must not pick up the system view of the same name. + ("snapshot", NO_GROUP_PREFIX % "snapshot", True), + ("snapshot", PREFIX % ("views", "snapshot"), False), + ("cdc", PREFIX % ("cdc", "consumer"), False), +] + + +class CheckMetricRegistryPattern: + """ + Checks that the pattern covers how the JMX exporter really names a registry. + """ + @pytest.mark.parametrize(["registry", "mbean_name", "expected"], CASES) + def check_the_pattern_matches_exactly_the_registry(self, registry, mbean_name, expected): + assert python_grep(*metric_registry_pattern(registry), mbean_name) == (mbean_name if expected else None) + + @pytest.mark.skipif(GREP is None, reason="grep is not installed") + @pytest.mark.parametrize(["registry", "mbean_name", "expected"], CASES) + def check_grep_agrees(self, tmp_path, registry, mbean_name, expected): + assert real_grep(tmp_path, *metric_registry_pattern(registry), mbean_name) == \ + (mbean_name if expected else None) + + @pytest.mark.parametrize("registry", ["cache.it's", 'cache.a"b', "cache.a\\b", "cache.a?b"]) + def check_a_name_the_jmxterm_commands_cannot_carry_is_rejected(self, registry): + """The MBean name goes through the shell quoting of the grep and jmxterm commands.""" + with pytest.raises(ValueError): + metric_registry_pattern(registry) diff --git a/modules/ducktests/tests/ignitetest/services/utils/cdc/cdc_helper.py b/modules/ducktests/tests/ignitetest/services/utils/cdc/cdc_helper.py index 9965dbde21f30..0b8f83fb93a08 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/cdc/cdc_helper.py +++ b/modules/ducktests/tests/ignitetest/services/utils/cdc/cdc_helper.py @@ -207,8 +207,6 @@ def last_ignite_cdc_event_time(ignite_cdc): :return: Timestamp of the last CDC event (unix time in seconds). """ def last_event_time_on(node): - jmx_client = JmxClient(node) - if isinstance(ignite_cdc, IgniteCdcUtility): main_java_class = ignite_cdc.APP_SERVICE_CLASS else: @@ -219,12 +217,12 @@ def last_event_time_on(node): if len(pids) == 0: raise AssertionError("ignite_cdc java process is not found on node: " + node.account.hostname) - jmx_client.pid = pids[0] + jmx_client = JmxClient(node, pids[0]) try: - mbean = jmx_client.find_mbean('.*name=cdc.*') + mbean = jmx_client.find_metric_registry('cdc') - return int(next(mbean.LastEventTime).strip()) + return int(mbean.value("LastEventTime")) except (StopIteration, RemoteCommandError): ignite_cdc.logger.warn("Filed to read LastEventTime metric from ignite_cdc, node: " + node.account.hostname) diff --git a/modules/ducktests/tests/ignitetest/services/utils/control_utility.py b/modules/ducktests/tests/ignitetest/services/utils/control_utility.py index 8ace03db5ae8d..329e5321974e5 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/control_utility.py +++ b/modules/ducktests/tests/ignitetest/services/utils/control_utility.py @@ -28,7 +28,6 @@ from ignitetest.services.utils.auth import get_credentials, is_auth_enabled from ignitetest.services.utils.ignite_spec import envs_to_exports from ignitetest.services.utils.ssl.ssl_params import get_ssl_params, is_ssl_enabled, IGNITE_ADMIN_ALIAS -from ignitetest.services.utils.jmx_utils import JmxClient from ignitetest.utils.version import V_2_11_0 @@ -303,17 +302,17 @@ def snapshot_create(self, snapshot_name: str, timeout_sec: int = 60): while datetime.now() < delta_time: for node in self._cluster.nodes: - mbean = JmxClient(node).find_mbean('.*name=snapshot.*', negative_pattern='group=views') + mbean = node.metric_registry_mbean('snapshot') - if snapshot_name != next(mbean.LastSnapshotName, ""): + if snapshot_name != mbean.value("LastSnapshotName", ""): continue - start_time = int(next(mbean.LastSnapshotStartTime)) - end_time = int(next(mbean.LastSnapshotEndTime)) - err_msg = next(mbean.LastSnapshotErrorMessage) + start_time = int(mbean.value("LastSnapshotStartTime")) + end_time = int(mbean.value("LastSnapshotEndTime")) + err_msg = mbean.value("LastSnapshotErrorMessage") if (start_time < end_time) and (err_msg == ''): - assert snapshot_name == next(mbean.LastSnapshotName) + assert snapshot_name == mbean.value("LastSnapshotName") return raise TimeoutError(f'Failed to wait for the snapshot operation to complete: ' diff --git a/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py b/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py index 8bd758f339d04..0f7079d75b157 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py +++ b/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py @@ -37,7 +37,7 @@ from ignitetest.services.utils.background_thread import BackgroundThreadService from ignitetest.services.utils.concurrent import CountDownLatch, AtomicValue from ignitetest.services.utils.ignite_spec import resolve_spec, SHARED_PREPARED_FILE -from ignitetest.services.utils.jmx_utils import ignite_jmx_mixin, JmxClient +from ignitetest.services.utils.jmx_utils import ignite_jmx_mixin from ignitetest.services.utils.jvm_utils import JvmProcessMixin, JvmVersionMixin from ignitetest.services.utils.log_utils import monitor_log from ignitetest.services.utils.path import IgnitePathAware @@ -600,10 +600,10 @@ def await_rebalance(self, timeout_sec=600): node = random.choice(self.alive_nodes) rebalanced = False - mbean = JmxClient(node).find_mbean('.*name=cluster') + mbean = node.metric_registry_mbean('cluster') while datetime.now() < delta_time and not rebalanced: - rebalanced = next(mbean.Rebalanced) == 'true' + rebalanced = mbean.bool_value("Rebalanced") if rebalanced: return diff --git a/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py b/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py index 48cfe00003f53..60bfb14a2e443 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py +++ b/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py @@ -20,21 +20,87 @@ import os import re -from ignitetest.services.utils.decorators import memoize from ignitetest.services.utils.jvm_utils import java_version, java_major_version +_NO_DEFAULT = object() + +# Not allowed in a metric registry name: the single quote would break the quoted grep argument, and the +# $'...' quoting of the jmxterm commands turns the exporter's \" and \? escapes, and a lone backslash, into +# something else. +_UNSUPPORTED_REGISTRY_CHARS = frozenset("'\"\\?") + + +def _object_name_value(value): + """ + Spells a property value the way IgniteUtils.escapeObjectNameValue() does: as is when it is + purely alphanumeric or underscore, otherwise quoted, with backslash, quote, '?' and '*' escaped. + """ + if re.fullmatch(r'[A-Za-z0-9_]+', value): + return value + + return '"' + re.sub(r'([\\"?*])', r'\\\1', value) + '"' + + +def _ere_escape(value): + """ + Escapes a literal for a POSIX extended regular expression, which is what 'grep -E' reads - + re.escape() escapes for the Python dialect instead. + """ + return re.sub(r'([[\\.^$*+?(){|])', r'\\\1', value) + + +def metric_registry_pattern(registry): + """ + Builds the MBean name pattern of a single metric registry. + + The JMX metric exporter splits a registry name at its FIRST dot and makes the head the + MBean group and the tail its name: ``cache.myCache`` becomes ``group=cache`` plus + ``name=myCache``, ``io.dataregion.default`` becomes ``group=io`` plus + ``name="dataregion.default"``. A registry without a dot, ``snapshot`` say, gets no group at all. + + Both values are spelled as the exporter spells them - quoted and escaped unless purely + alphanumeric - and matched literally. The properties of an MBean name are ordered + alphabetically, so the group always comes before the name but never right next to it. And + ``name`` sorts last of them all, so the pattern ends at the end of the line: without that + anchor ``myCache`` would also match the registry of ``myCacheV2``. + + A registry without a group needs the negative pattern as well - an ERE has no lookahead, and + the system view of the same name (``group=views,...,name=snapshot``) matches otherwise. + + The MBean name of a registry carries no pid, but it does carry the class loader hash, so it + may differ between two runs of a node. + + :param registry: Metric registry name. It must not contain a single or double quote, a backslash + or a question mark - see :data:`_UNSUPPORTED_REGISTRY_CHARS`. + :return: Tuple of the pattern and the negative pattern to pass to :meth:`JmxClient.find_mbean`. + """ + if _UNSUPPORTED_REGISTRY_CHARS.intersection(registry): + raise ValueError(f"A metric registry name with a quote, a backslash or '?' is not supported: {registry}") + + group, dot, name = registry.partition('.') + + if not dot: + return rf'.*name={_ere_escape(_object_name_value(registry))}\s*$', 'group=' + + return rf'.*group={_ere_escape(_object_name_value(group))},' \ + rf'.*name={_ere_escape(_object_name_value(name))}\s*$', None + def ignite_jmx_mixin(node, service): """ - Dynamically mixin JMX attributes to Ignite service node. + Dynamically mixin JMX attributes to Ignite service node. Called on every start of the node, + which is what hands a restarted node a JMX client of its new JVM. :param node: Ignite service node. :param service: Ignite service. """ setattr(node, 'pids', service.pids(node, service.main_java_class)) setattr(node, 'install_root', service.install_root) - base_cls = node.__class__ - base_cls_name = node.__class__.__name__ - node.__class__ = type(base_cls_name, (base_cls, IgniteJmxMixin), {}) + setattr(node, '_jmx_client', None) + + if not isinstance(node, IgniteJmxMixin): + base_cls = node.__class__ + base_cls_name = node.__class__.__name__ + node.__class__ = type(base_cls_name, (base_cls, IgniteJmxMixin), {}) class JmxMBean: @@ -53,6 +119,30 @@ def __getattr__(self, attr): """ return self.client.mbean_attribute(self.name, attr) + def value(self, attr, default=_NO_DEFAULT): + """ + Reads a single valued attribute - which is what a metric read almost always wants, + as opposed to the raw line iterator the attribute access itself returns. + + :param attr: Attribute name. + :param default: Value to return if the attribute reads nothing; StopIteration is raised if not passed. + :return: Attribute value, whitespace stripped. + """ + try: + return next(self.client.mbean_attribute(self.name, attr)).strip() + except StopIteration: + if default is _NO_DEFAULT: + raise + + return default + + def bool_value(self, attr): + """ + :param attr: Attribute name. + :return: Attribute value as a boolean; anything but "true" is False. + """ + return self.value(attr).lower() == "true" + def run(self, operation, params): """" Runs through JMX client MBean operation. @@ -66,11 +156,16 @@ def run(self, operation, params): class JmxClient: """JMX client, invokes jmxterm on node locally. """ - def __init__(self, node): + def __init__(self, node, pid=None): + """ + :param node: Node to run jmxterm on. + :param pid: JVM to connect to; the first of the node's pids if not passed. + """ self.node = node self.install_root = node.install_root - self.pid = node.pids[0] + self.pid = pid if pid is not None else node.pids[0] self.java_major = java_major_version(java_version(self.node)) + self._mbeans = {} @property def jmx_util_cmd(self): @@ -81,24 +176,34 @@ def jmx_util_cmd(self): return os.path.join(f"java {extra_flag} -jar {self.install_root}/jmxterm.jar -v silent -n") - @memoize def find_mbean(self, pattern, negative_pattern=None, domain='org.apache'): """ - Find mbean by specified pattern and domain on node. + Find mbean by specified pattern and domain on node. The lookup is cached by this client, which + is safe as long as the client does not outlive the JVM it was built for. :param pattern: MBean name pattern. :param negative_pattern: if passed used to filter out some MBeans :param domain: Domain of MBean :return: JmxMBean instance """ - cmd = "echo $'open %s\\n beans -d %s \\n close' | %s | grep -E -o '%s'" \ - % (self.pid, domain, self.jmx_util_cmd, pattern) + key = (pattern, negative_pattern, domain) + + if key not in self._mbeans: + cmd = "echo $'open %s\\n beans -d %s \\n close' | %s | grep -E -o '%s'" \ + % (self.pid, domain, self.jmx_util_cmd, pattern) - if negative_pattern: - cmd += " | grep -E -v '%s'" % negative_pattern + if negative_pattern: + cmd += " | grep -E -v '%s'" % negative_pattern - name = next(self.__run_cmd(cmd)).strip() + self._mbeans[key] = JmxMBean(self, next(self.__run_cmd(cmd)).strip()) - return JmxMBean(self, name) + return self._mbeans[key] + + def find_metric_registry(self, registry): + """ + :param registry: Metric registry name, e.g. ``cache.myCache`` - see :func:`metric_registry_pattern`. + :return: JmxMBean of the metric registry. + """ + return self.find_mbean(*metric_registry_pattern(registry)) def mbean_attribute(self, mbean, attr): """ @@ -188,31 +293,34 @@ def __find__(self, pattern): class IgniteJmxMixin: """ Mixin to IgniteService node, exposing useful properties, obtained from JMX. + + Everything here is read through the JMX client of the node's CURRENT JVM: a client holds the + pid and caches the MBean names of the JVM it was built for, so ignite_jmx_mixin() drops it on + every start of the node and the next access builds a new one. """ - @memoize def jmx_client(self): """ - :return: JmxClient instance. + :return: JmxClient instance of the running node. """ - # noinspection PyTypeChecker - return JmxClient(self) + if self._jmx_client is None: + # noinspection PyTypeChecker + self._jmx_client = JmxClient(self) + + return self._jmx_client - @memoize def node_id(self): """ :return: Local node id. """ - return next(self.kernal_mbean().LocalNodeId).strip() + return self.kernal_mbean().value("LocalNodeId") def discovery_info(self): """ :return: DiscoveryInfo instance. """ disco_mbean = self.disco_mbean() - crd = next(disco_mbean.Coordinator).strip() - local = next(disco_mbean.LocalNodeFormatted).strip() - return DiscoveryInfo(crd, local) + return DiscoveryInfo(disco_mbean.value("Coordinator"), disco_mbean.value("LocalNodeFormatted")) def kernal_mbean(self): """ @@ -220,12 +328,18 @@ def kernal_mbean(self): """ return self.jmx_client().find_mbean('.*group=Kernal.*name=IgniteKernal') - @memoize + def metric_registry_mbean(self, registry): + """ + :param registry: Metric registry name, e.g. ``cache.myCache`` or ``cacheGroups.myGroup``. + :return: MBean of the metric registry. + """ + return self.jmx_client().find_metric_registry(registry) + def disco_mbean(self): """ :return: DiscoverySpi MBean. """ - disco_spi = next(self.kernal_mbean().DiscoverySpiFormatted).strip() + disco_spi = self.kernal_mbean().value("DiscoverySpiFormatted") if 'ZookeeperDiscoverySpi' in disco_spi: return self.jmx_client().find_mbean('.*group=SPIs.*name=ZookeeperDiscoverySpi') diff --git a/modules/ducktests/tests/ignitetest/tests/dump_test.py b/modules/ducktests/tests/ignitetest/tests/dump_test.py index db58d5064441a..0a8bfcd4d6964 100644 --- a/modules/ducktests/tests/ignitetest/tests/dump_test.py +++ b/modules/ducktests/tests/ignitetest/tests/dump_test.py @@ -115,8 +115,8 @@ def get_data_region_size(ignite): data_region_size = {} for node in ignite.nodes: - mbean = node.jmx_client().find_mbean('.*group=io.*name="dataregion.default"') - data_region_size[node.consistent_id] = int(next(mbean.TotalUsedSize)) + mbean = node.metric_registry_mbean('io.dataregion.default') + data_region_size[node.consistent_id] = int(mbean.value("TotalUsedSize")) return { "data_region_size": data_region_size diff --git a/modules/ducktests/tests/ignitetest/tests/rebalance/util.py b/modules/ducktests/tests/ignitetest/tests/rebalance/util.py index 9a5351dd2388e..282a8f4ce786b 100644 --- a/modules/ducktests/tests/ignitetest/tests/rebalance/util.py +++ b/modules/ducktests/tests/ignitetest/tests/rebalance/util.py @@ -190,12 +190,12 @@ def get_rebalance_metrics(node, cache_group): :param cache_group: Cache group. :return: RebalanceMetrics instance. """ - mbean = node.jmx_client().find_mbean('.*group=cacheGroups.*name="%s"' % cache_group) - start_time = int(next(mbean.RebalancingStartTime)) - end_time = int(next(mbean.RebalancingEndTime)) + mbean = node.metric_registry_mbean(f'cacheGroups.{cache_group}') + start_time = int(mbean.value("RebalancingStartTime")) + end_time = int(mbean.value("RebalancingEndTime")) return RebalanceMetrics( - received_bytes=int(next(mbean.RebalancingReceivedBytes)), + received_bytes=int(mbean.value("RebalancingReceivedBytes")), start_time=start_time, end_time=end_time, duration=(end_time - start_time) if start_time != -1 and end_time != -1 else 0, From d34d2656d73d549284283bd1ef872d3a50825a5d Mon Sep 17 00:00:00 2001 From: Maksim Davydov Date: Thu, 17 Sep 2026 16:45:17 +0300 Subject: [PATCH 2/8] IGNITE-29049 [ducktests] Escape '[' in the ERE escaping character class An unescaped '[' right after the class opener made Python emit 'FutureWarning: Possible nested set' on every JMX metric lookup. --- modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py b/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py index 60bfb14a2e443..73ca0abdc46ec 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py +++ b/modules/ducktests/tests/ignitetest/services/utils/jmx_utils.py @@ -46,7 +46,7 @@ def _ere_escape(value): Escapes a literal for a POSIX extended regular expression, which is what 'grep -E' reads - re.escape() escapes for the Python dialect instead. """ - return re.sub(r'([[\\.^$*+?(){|])', r'\\\1', value) + return re.sub(r'([\[\\.^$*+?(){|])', r'\\\1', value) def metric_registry_pattern(registry): From f7f7487962010209ae100ecfbadf066f5940de01 Mon Sep 17 00:00:00 2001 From: Maksim Davydov Date: Thu, 10 Sep 2026 16:48:33 +0300 Subject: [PATCH 3/8] IGNITE-28952 [ducktests] MDC fixture spanning an arbitrary number of data centers MdcTopologyValidator has two modes and the number of data centers picks one: with an EVEN DC count a segment stays writable while it sees the main DC, with an ODD one while it sees a majority of the DC set. The fixture only ever built a two DC cluster, so only the first mode could be tested. MdcCluster now takes the DC set it spans (dcs=, two by default) and compiles the cache parameters from it: * mdc_topology_params() emits mainDc or datacenters, never both - MdcTopologyValidator.checkConfiguration() rejects the pair; * min_backups() is the smallest backup count giving every DC one copy of every partition, and is what generate_data() uses by default; * _with_cache_params() is the single point an application that creates the cache is handed all of it, so no call site can configure a cache that disagrees with the DC set. That is what lets the transactional test drop its hand written mainDc. verify_half_ring_healthy()/verify_split_brain() generalize into verify_segment_healthy()/verify_segments(): a segment is now a DC or a group of DCs that still see each other, so a three DC cluster with one DC cut off is expressed as verify_segments((DC_1, DC_2), DC_3). NetworkGroupManager gains enable/disable_network_partitions(*pairs). A cluster of three or more groups is cut apart along several links at once, and every chain a node takes part in has to be installed by the same single SSH round-trip: rolling the links out one after the other would present the cluster with intermediate segmentations it would legitimately react to. The cache level MdcTopologyValidator becomes optional, through the cache parameter topologyValidator and the mdc_cache_topology_validator global, for a fork whose validator is configured elsewhere. The affinity backup filter moves into a protected MdcCacheAwareApplication.backupFilter() for the same reason: a fork that spreads the copies by something finer than the data center overrides one method rather than repeating the cache configuration. check_mdc_cluster.py and check_partition.py cover all of the above without a cluster. --- modules/ducktests/README.md | 1 + .../tests/mdc/MdcCacheAwareApplication.java | 67 ++- .../tests/checks/services/mdc/__init__.py | 14 + .../checks/services/mdc/check_mdc_cluster.py | 192 ++++++ .../services/network_group/check_partition.py | 157 +++++ .../ignitetest/services/mdc/mdc_cluster.py | 559 ++++++++++++++---- .../services/network_group/manager.py | 63 +- .../tests/mdc/transactional_partition_test.py | 3 +- 8 files changed, 900 insertions(+), 156 deletions(-) create mode 100644 modules/ducktests/tests/checks/services/mdc/__init__.py create mode 100644 modules/ducktests/tests/checks/services/mdc/check_mdc_cluster.py create mode 100644 modules/ducktests/tests/checks/services/network_group/check_partition.py diff --git a/modules/ducktests/README.md b/modules/ducktests/README.md index b52c22c8a0296..0712f86c88acc 100644 --- a/modules/ducktests/README.md +++ b/modules/ducktests/README.md @@ -200,6 +200,7 @@ You can modify test environments at execution time using global flags injected t | **AppSpec** | Specifies the class to use for application specifications in Ignite applications. Controls how Ignite applications are configured and started. | ```{"AppSpec": "myapp.services.MyAppSpec"}``` | | **IgniteTestContext** | Class name for the test context implementation. Allows customization of test context behavior. | ```{"IgniteTestContext": "myapp.context.CustomTestContext"}``` | | **project** | Project/fork name for version handling (e.g., "ignite", "fork"). Used to distinguish between different Ignite variants. Default is "ignite". | ```{"project": "fork"}``` | +| **mdc_cache_topology_validator** | Whether the MDC tests create their caches with the cache level `MdcTopologyValidator`. Default is True. | ```{"mdc_cache_topology_validator": false}``` | #### Paths & Directories diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java index 447486faa9b75..17a823a650c4a 100644 --- a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java @@ -19,6 +19,7 @@ import java.util.Collections; import java.util.HashSet; +import java.util.List; import java.util.Set; import com.fasterxml.jackson.databind.JsonNode; import org.apache.ignite.IgniteCache; @@ -29,9 +30,11 @@ import org.apache.ignite.cache.QueryEntity; import org.apache.ignite.cache.affinity.rendezvous.MdcAffinityBackupFilter; import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.configuration.CacheConfiguration; import org.apache.ignite.internal.ducktest.tests.dto.IndexedDataRecord; import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication; +import org.apache.ignite.lang.IgniteBiPredicate; import org.apache.ignite.topology.MdcTopologyValidator; import static org.apache.ignite.IgniteSystemProperties.IGNITE_DATA_CENTER_ID; @@ -50,8 +53,10 @@ *
    *
  • {@code cacheName} - cache name;
    • *
  • {@code backups} - number of backups; {@code (backups + 1)} must be divisible by {@code dcsNum};
    • + *
  • {@code topologyValidator} - whether to set the cache level {@link MdcTopologyValidator}, default + * {@code true};
    • *
  • {@code mainDc} - main data center for the topology validator (2 DC mode); required, and must be - * non-empty, unless {@code datacenters} is given;
    • + * non-empty, unless {@code datacenters} is given or the cache level validator is disabled; *
  • {@code datacenters} - full DC set for majority-based validation (odd DC count mode), * takes precedence over {@code mainDc};
    • *
  • {@code dcsNum} - number of data centers, default 2;
    • @@ -82,6 +87,9 @@ public abstract class MdcCacheAwareApplication extends IgniteAwareApplication { /** */ protected static final int DFLT_PARTITIONS = 512; + /** The cache level topology validator is set unless the parameters say otherwise. */ + protected static final boolean DFLT_CACHE_TOP_VALIDATOR = true; + /** */ protected static final CacheAtomicityMode DFLT_ATOMICITY_MODE = ATOMIC; @@ -129,6 +137,51 @@ protected CacheConfiguration mdcCacheConfiguration(JsonNode jNod int dcsNum = jNode.path("dcsNum").asInt(DFLT_DCS_NUM); + RendezvousAffinityFunction affinity = new RendezvousAffinityFunction().setPartitions(partitions); + + IgniteBiPredicate> backupFilter = backupFilter(jNode, dcsNum, backups); + + if (backupFilter != null) + affinity.setAffinityBackupFilter(backupFilter); + + CacheConfiguration cacheCfg = new CacheConfiguration() + .setName(cacheName) + .setCacheMode(cacheMode) + .setAtomicityMode(atomicity) + .setWriteSynchronizationMode(writeSync) + .setBackups(backups) + .setReadFromBackup(readFromBackup) + .setAffinity(affinity); + + if (jNode.path("topologyValidator").asBoolean(DFLT_CACHE_TOP_VALIDATOR)) + cacheCfg.setTopologyValidator(mdcTopologyValidator(jNode)); + else + log.info("Cache level topology validator is disabled [cache=" + cacheName + "]"); + + return cacheCfg; + } + + /** + * The affinity backup filter the cache is configured with. A {@link RendezvousAffinityFunction} + * holds exactly one, so an override replaces the MDC filter rather than complementing it - which + * is the point: a fork that spreads the copies by something finer than the data center (a cell, + * an availability zone) says so here instead of repeating the rest of the cache configuration. + * + * @param jNode Parameters. + * @param dcsNum Number of data centers. + * @param backups Number of backups. + * @return Affinity backup filter to set, or {@code null} for plain rendezvous affinity. + */ + protected IgniteBiPredicate> backupFilter(JsonNode jNode, int dcsNum, + int backups) { + return new MdcAffinityBackupFilter(dcsNum, backups); + } + + /** + * @param jNode Parameters. + * @return Cache level topology validator compiled from the application parameters. + */ + private MdcTopologyValidator mdcTopologyValidator(JsonNode jNode) { MdcTopologyValidator topValidator = new MdcTopologyValidator(); if (jNode.hasNonNull("datacenters")) { @@ -147,17 +200,7 @@ protected CacheConfiguration mdcCacheConfiguration(JsonNode jNod topValidator.setMainDatacenter(mainDc); } - return new CacheConfiguration() - .setName(cacheName) - .setTopologyValidator(topValidator) - .setCacheMode(cacheMode) - .setAtomicityMode(atomicity) - .setWriteSynchronizationMode(writeSync) - .setBackups(backups) - .setReadFromBackup(readFromBackup) - .setAffinity(new RendezvousAffinityFunction() - .setPartitions(partitions) - .setAffinityBackupFilter(new MdcAffinityBackupFilter(dcsNum, backups))); + return topValidator; } /** diff --git a/modules/ducktests/tests/checks/services/mdc/__init__.py b/modules/ducktests/tests/checks/services/mdc/__init__.py new file mode 100644 index 0000000000000..ec2014340d78f --- /dev/null +++ b/modules/ducktests/tests/checks/services/mdc/__init__.py @@ -0,0 +1,14 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/modules/ducktests/tests/checks/services/mdc/check_mdc_cluster.py b/modules/ducktests/tests/checks/services/mdc/check_mdc_cluster.py new file mode 100644 index 0000000000000..57b408df9d2d0 --- /dev/null +++ b/modules/ducktests/tests/checks/services/mdc/check_mdc_cluster.py @@ -0,0 +1,192 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Checks the DC-count dependent parts of the MDC fixture. + +The number of data centers decides which MdcTopologyValidator mode the caches are +configured for, how many backups spread one copy per DC, and which links a partition has +to cut. All of that is compiled without a cluster, so it is checked without one. +""" + +import pytest + +from ignitetest.services.mdc.mdc_cluster import MdcCluster, mdc_topology_params, min_backups, all_pairs, \ + isolation_pairs, cross_dc_network, per_dc, CACHE_TOP_VALIDATOR_GLOBAL, DCS_2, DCS_3, DC_1, DC_2, DC_3 + +DELAY_MS = 100 + +DFLT_DELAY = f"{DELAY_MS}ms" + + +class FakeMdcCluster: + """ + The two members cross_dc_network() reads off an MdcCluster. + """ + def __init__(self, dcs): + self.dcs = dcs + + def network_registry(self): + """Every DC group is non-empty; the services themselves are irrelevant here.""" + return {dc: [] for dc in self.dcs} + + +class CheckMdcTopologyParams: + """ + Checks the cache parameters that select the topology validator mode. + """ + def check_even_dc_count_uses_a_main_dc(self): + """An even DC set is validated against a main DC, the first one by default.""" + assert mdc_topology_params(DCS_2) == {"dcsNum": 2, "mainDc": DC_1} + + assert mdc_topology_params(DCS_2, main_dc=DC_2) == {"dcsNum": 2, "mainDc": DC_2} + + def check_odd_dc_count_uses_the_dc_set(self): + """An odd DC set is validated by majority, so it carries the DC set instead.""" + assert mdc_topology_params(DCS_3) == {"dcsNum": 3, "datacenters": [DC_1, DC_2, DC_3]} + + def check_the_two_modes_are_never_mixed(self): + """ + MdcTopologyValidator.checkConfiguration() rejects a main DC alongside an odd DC + set, so a main DC must not leak into the majority mode even when one is asked for. + """ + params = mdc_topology_params(DCS_3, main_dc=DC_1) + + assert "mainDc" not in params, "A main DC alongside an odd DC set fails cache startup" + + @pytest.mark.parametrize(["dcs", "expected"], [(DCS_2, 1), (DCS_3, 2)]) + def check_min_backups_gives_one_copy_per_dc(self, dcs, expected): + """(backups + 1) must divide by the DC count - the MdcAffinityBackupFilter contract.""" + assert min_backups(dcs) == expected + + assert (min_backups(dcs) + 1) % len(dcs) == 0 + + +class CheckMdcNetworkLayout: + """ + Checks the DC pairings a partition is expressed in, and the impairment mesh. + """ + def check_all_pairs_covers_the_mesh(self): + """Every cross-DC link appears exactly once, in a stable order.""" + assert all_pairs(DCS_2) == [(DC_1, DC_2)] + + assert all_pairs(DCS_3) == [(DC_1, DC_2), (DC_1, DC_3), (DC_2, DC_3)] + + def check_isolation_pairs_cut_one_dc_only(self): + """Isolating a DC cuts its own links and leaves the rest of the mesh intact.""" + cut = isolation_pairs(DC_3, DCS_3) + + assert cut == [(DC_3, DC_1), (DC_3, DC_2)] + + assert (DC_1, DC_2) not in [tuple(sorted(pair)) for pair in cut], \ + "The DCs left behind must keep seeing each other" + + def check_symmetric_impairment_reaches_every_pair(self): + """One delay argument impairs the whole mesh, not just the first pair.""" + net = cross_dc_network(None, FakeMdcCluster(DCS_3), delay_ms=DELAY_MS) + + for dc_a, dc_b in all_pairs(DCS_3): + cfg = net.network_group_store.get_config(dc_a, dc_b) + + assert cfg is not None and cfg.delay == DFLT_DELAY, f"{dc_a} -> {dc_b} is unimpaired" + + assert net.network_group_store.get_config(dc_b, dc_a) == cfg, "Impairments are bidirectional" + + def check_no_impairment_leaves_the_store_empty(self): + """Without delay or loss the manager still owns partitions, but deploys no netem.""" + net = cross_dc_network(None, FakeMdcCluster(DCS_3)) + + assert net.network_group_store.matrix == {} + + +class CheckMdcPerDcCounts: + """ + Checks how the per-DC service counts are spread over the DC set. + """ + def check_a_scalar_count_covers_every_dc(self): + """One number means that number of nodes in every DC the cluster spans.""" + assert per_dc(2, DCS_3) == {DC_1: 2, DC_2: 2, DC_3: 2} + + def check_a_dict_count_is_taken_as_is(self): + """An asymmetric layout names only the DCs it populates.""" + assert per_dc({DC_1: 3}, DCS_3) == {DC_1: 3} + + def check_a_dict_naming_a_foreign_dc_is_rejected(self): + """ + A DC outside the cluster's own set is skipped by network_registry(), so its nodes + would run with no impairments and no partition rules - and nothing else would say + so. It has to fail where it is declared. + """ + with pytest.raises(AssertionError, match=DC_3): + per_dc({DC_1: 1, DC_3: 1}, DCS_2) + + +def _fixture(dcs, top_validator=True): + """ + An MdcCluster carrying only what _with_cache_params() reads - no services are built, so + no ducktape cluster is needed. Bypassing the constructor is the point: it pins down how + little of the fixture the cache parameter compilation actually depends on. + """ + mdc = MdcCluster.__new__(MdcCluster) + + mdc.dcs = tuple(dcs) + mdc.main_dc = dcs[0] + mdc.cache_defaults = {"topologyValidator": top_validator} + + return mdc + + +class CheckMdcCacheParams: + """ + Checks the single point every cache of an MDC test is configured from. + """ + def check_an_app_that_creates_the_cache_is_handed_the_dc_set(self): + """A cache created by a scenario must agree with the DC set the cluster spans.""" + params = _fixture(DCS_3)._with_cache_params({"cacheName": "c", "createCache": True}) + + assert params["dcsNum"] == 3 + + assert params["datacenters"] == [DC_1, DC_2, DC_3] + + assert params["topologyValidator"] is True + + def check_an_app_that_only_uses_the_cache_is_handed_nothing(self): + """Cache parameters an application would only ignore must not reach it at all.""" + params = {"cacheName": "c", "mode": "GET"} + + assert _fixture(DCS_3)._with_cache_params(params) == params + + def check_an_app_that_always_creates_the_cache_needs_no_flag(self): + """The generator carries no createCache parameter, so its call site says so instead.""" + params = _fixture(DCS_2)._with_cache_params({"cacheName": "c"}, creates_cache=True) + + assert params["mainDc"] == DC_1 + + def check_an_explicit_parameter_wins(self): + """A scenario stays able to override what the fixture injects.""" + params = _fixture(DCS_2)._with_cache_params({"createCache": True, "mainDc": DC_2}) + + assert params["mainDc"] == DC_2 + + def check_the_global_reaches_the_cache(self): + """ + The global is only ever read into cache_defaults, so this covers the whole path from + --global-json to the application parameters. Its name is part of the README. + """ + assert CACHE_TOP_VALIDATOR_GLOBAL == "mdc_cache_topology_validator" + + params = _fixture(DCS_3, top_validator=False)._with_cache_params({"createCache": True}) + + assert params["topologyValidator"] is False diff --git a/modules/ducktests/tests/checks/services/network_group/check_partition.py b/modules/ducktests/tests/checks/services/network_group/check_partition.py new file mode 100644 index 0000000000000..434c832533922 --- /dev/null +++ b/modules/ducktests/tests/checks/services/network_group/check_partition.py @@ -0,0 +1,157 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Checks how NetworkGroupManager compiles a partition into per-node commands. + +A cluster of three or more groups can be cut apart along several links at once, and the +intermediate topologies of a link-by-link rollout are themselves valid segmentations the +cluster would react to. So what matters is that every node is configured by exactly one +SSH round-trip carrying every chain that node takes part in - and no chain it does not. +""" + +import logging + +import pytest + +from ignitetest.services.network_group.manager import NetworkGroupManager +from ignitetest.services.network_group.tc_rule_args import partition_chain_name + +DC_1, DC_2, DC_3 = "DC1", "DC2", "DC3" + +DCS = (DC_1, DC_2, DC_3) + +NODES_PER_DC = 2 + + +class FakeNode: + """A node is only ever an identity and an account here.""" + def __init__(self, name): + self.name = name + + def __repr__(self): + return self.name + + +class FakeService: + """The single member NetworkGroupManager reads off a registered service.""" + def __init__(self, nodes): + self.nodes = nodes + + +@pytest.fixture(name="manager") +def _manager(monkeypatch): + """ + A manager over three groups whose SSH layer is replaced by a recorder: yields the + manager and the list of (node, command) tasks it submits. + """ + registry = {dc: [FakeService([FakeNode(f"{dc}-{i}") for i in range(NODES_PER_DC)])] for dc in DCS} + + tasks = [] + + monkeypatch.setattr(NetworkGroupManager, "_resolve_group_ips", lambda self, group: [f"{group}-ip"]) + monkeypatch.setattr(NetworkGroupManager, "_ssh_parallel", lambda self, submitted, tag: tasks.extend(submitted)) + monkeypatch.setattr(NetworkGroupManager, "_log_network", lambda self, log_tag: None) + + yield NetworkGroupManager(logging.getLogger(__name__), None, registry), tasks + + +def _commands_by_node(tasks): + return {node.name: cmd for node, cmd in tasks} + + +class CheckNetworkPartition: + """ + Checks the per-node command compilation of single and multi-link partitions. + """ + def check_one_round_trip_per_node(self, manager): + """A three way split configures each of the six nodes exactly once.""" + mgr, tasks = manager + + mgr.enable_network_partitions((DC_1, DC_2), (DC_1, DC_3), (DC_2, DC_3)) + + assert len(tasks) == len(DCS) * NODES_PER_DC, "A node must be configured by a single SSH round-trip" + + assert len(_commands_by_node(tasks)) == len(tasks), "Node commands must not be split across tasks" + + def check_a_node_carries_every_chain_it_is_in(self, manager): + """Each node gets the chains of its own links, and none of the link it sits out.""" + mgr, tasks = manager + + mgr.enable_network_partitions((DC_1, DC_2), (DC_1, DC_3), (DC_2, DC_3)) + + cmds = _commands_by_node(tasks) + + for dc in DCS: + own_chains = {partition_chain_name(dc, other) for other in DCS if other != dc} + + foreign_chain = partition_chain_name(*[other for other in DCS if other != dc]) + + for i in range(NODES_PER_DC): + cmd = cmds[f"{dc}-{i}"] + + for chain in own_chains: + assert chain in cmd, f"{dc}-{i} is missing chain {chain}" + + assert foreign_chain not in cmd, f"{dc}-{i} took part in the foreign chain {foreign_chain}" + + def check_isolating_one_group_leaves_the_others_connected(self, manager): + """Cutting DC3 off touches DC1 and DC2 only through their links to DC3.""" + mgr, tasks = manager + + mgr.enable_network_partitions((DC_3, DC_1), (DC_3, DC_2)) + + cmds = _commands_by_node(tasks) + + assert len(cmds) == len(DCS) * NODES_PER_DC, "Every node of every group takes part in the cut" + + assert partition_chain_name(DC_1, DC_2) not in cmds[f"{DC_1}-0"], \ + "The groups left behind must keep seeing each other" + + # The isolated group holds both chains; the ones left behind hold only their own. + assert partition_chain_name(DC_1, DC_3) in cmds[f"{DC_3}-0"] + assert partition_chain_name(DC_2, DC_3) in cmds[f"{DC_3}-0"] + + assert partition_chain_name(DC_2, DC_3) not in cmds[f"{DC_1}-0"] + + def check_single_pair_partition_is_a_multi_partition_of_one(self, manager): + """The pairwise entry points stay the one-link case of the batched ones.""" + mgr, tasks = manager + + mgr.enable_network_partition(DC_1, DC_2) + + cmds = _commands_by_node(tasks) + + assert set(cmds) == {f"{DC_1}-0", f"{DC_1}-1", f"{DC_2}-0", f"{DC_2}-1"}, \ + "Only the two groups of the cut link are configured" + + chain = partition_chain_name(DC_1, DC_2) + + assert all(chain in cmd for cmd in cmds.values()) + + def check_heal_flushes_every_chain_on_every_node(self, manager): + """Healing a multi-link split flushes each node's chains in one round-trip.""" + mgr, tasks = manager + + mgr.disable_network_partitions((DC_1, DC_2), (DC_1, DC_3), (DC_2, DC_3)) + + cmds = _commands_by_node(tasks) + + assert len(cmds) == len(DCS) * NODES_PER_DC + + for dc in DCS: + for other in DCS: + if other != dc: + assert partition_chain_name(dc, other) in cmds[f"{dc}-0"] diff --git a/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py b/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py index a9ccbe4496d3c..4eb57406df08a 100644 --- a/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py +++ b/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py @@ -14,19 +14,22 @@ # limitations under the License. """ -MDC test fixture. +The fixture spans an arbitrary number of data centers, which selects the +:class:`MdcTopologyValidator` mode (see :func:`mdc_topology_params`): - mdc = MdcCluster(self, ignite_version, srv_per_dc=3, runners_per_dc=1) + mdc = MdcCluster(self, ignite_version, dcs=DCS_3, srv_per_dc=2, runners_per_dc=1) with cross_dc_network(self.logger, mdc, delay_ms=20) as net: - mdc.start_servers() - mdc.generate_data(DC_1, CACHE, 0, 1000, backups=1) - mdc.verify_cache_distribution(CACHE, copies_per_dc=1) + net.enable_network_partitions(*isolation_pairs(DC_3, mdc.dcs)) + mdc.verify_segments((DC_1, DC_2), DC_3) - net.enable_network_partition(DC_1, DC_2) - ... +Globals: + + mdc_cache_topology_validator - whether the MDC caches are created with the cache level + MdcTopologyValidator, default true. """ -from typing import Dict, List, Optional, Union +from itertools import combinations +from typing import Dict, List, Optional, Sequence, Tuple, Union from ignitetest.services.ignite import IgniteService from ignitetest.services.ignite_app import IgniteApplicationService @@ -41,10 +44,18 @@ DC_1 = "DC1" DC_2 = "DC2" -DCS = (DC_1, DC_2) +DC_3 = "DC3" + +# The two DC layouts the MDC topology validator distinguishes: an even DC count is +# validated against a main DC, an odd one by a majority of visible DCs. +DCS_2 = (DC_1, DC_2) +DCS_3 = (DC_1, DC_2, DC_3) IGNITE_STARTUP_TIMEOUT_SEC = 90 +# Global: set to false to create the MDC caches without the cache level topology validator. +CACHE_TOP_VALIDATOR_GLOBAL = "mdc_cache_topology_validator" + DATA_CENTER_ATTR = "IGNITE_DATA_CENTER_ID" IGNITE_SQL_RETRY_TIMEOUT_ATTR = "IGNITE_SQL_RETRY_TIMEOUT" @@ -67,6 +78,24 @@ # Every log of a server node, including the ones rotated by a restart. ALL_LOGS_GLOB = "ignite*.log*" +# Needed by everything that reads a node metric over JMX - await_rebalance(), the snapshot +# commands and the MDC safety metrics below among them. +JMX_METRIC_EXPORTER = "org.apache.ignite.spi.metric.jmx.JmxMetricExporterSpi" + +# Per-cache metrics holding the cluster's own verdict on the MDC guarantees. Registered on +# every server node that carries a DC id, for every cache. +# +# The two are not the same statement. The affinity one is about CONFIGURATION - whether the +# cache is set up to keep a copy of every partition in every DC at all, which is what the +# MdcAffinityBackupFilter provides. The distribution one is about the CURRENT assignment +# actually doing so, which a correctly configured cache still fails while some DC has no +# nodes to place a copy on. +MDC_SAFE_AFFINITY_METRIC = "IsCacheAffinityConfigurationMdcSafe" +MDC_SAFE_DISTRIBUTION_METRIC = "IsCachePartitionDistributionSafe" + +# A segment of a partitioned cluster: one DC or a group of DCs that still see each other. +Segment = Union[str, Sequence[str]] + def dc_jvm_opts(dc: str) -> List[str]: """ @@ -75,11 +104,86 @@ def dc_jvm_opts(dc: str) -> List[str]: return [f"-D{DATA_CENTER_ATTR}={dc}", f"-D{IGNITE_SQL_RETRY_TIMEOUT_ATTR}={IGNITE_SQL_RETRY_TIMEOUT_MS}"] -def _per_dc(value: Union[int, Dict[str, int]]) -> Dict[str, int]: +def mdc_topology_params(dcs: Sequence[str], main_dc: Optional[str] = None) -> dict: + """ + Compiles the cache parameters that pin ``MdcTopologyValidator`` and + ``MdcAffinityBackupFilter`` to the given DC set. + + The validator has two modes and the DC count picks one: with an EVEN number of DCs a + segment stays writable while it sees the main DC (``mainDc``), with an ODD number + while it sees a majority of the DC set (``datacenters``). Passing both is rejected by + ``MdcTopologyValidator.checkConfiguration()``, so exactly one is emitted here. + + :param dcs: All data centers the cluster spans. + :param main_dc: Main DC for the even-count mode, defaults to the first DC. Ignored for + an odd DC count, where the validator is majority based. + """ + params = {"dcsNum": len(dcs)} + + if len(dcs) % 2 == 1: + params["datacenters"] = list(dcs) + else: + params["mainDc"] = main_dc if main_dc is not None else dcs[0] + + return params + + +def min_backups(dcs: Sequence[str]) -> int: + """ + :return: Smallest backup count that gives every DC exactly one copy of every partition. + ``MdcAffinityBackupFilter`` requires ``(backups + 1)`` to be divisible by the + number of DCs, so this is the smallest admissible value at all. + """ + return len(dcs) - 1 + + +def all_pairs(dcs: Sequence[str]) -> List[Tuple[str, str]]: + """ + :return: Every unordered DC pair - the full cross-DC mesh. + """ + return list(combinations(dcs, 2)) + + +def isolation_pairs(dc: str, dcs: Sequence[str]) -> List[Tuple[str, str]]: + """ + :return: The DC pairs that cut ``dc`` off from every other DC, leaving the rest + connected. Feed to :meth:`NetworkGroupManager.enable_network_partitions`. + """ + return [(dc, other) for other in dcs if other != dc] + + +def per_dc(value: Union[int, Dict[str, int]], dcs: Sequence[str]) -> Dict[str, int]: """ Normalizes an int-or-dict per-DC count into a dict, e.g. 3 -> {DC1: 3, DC2: 3}. + + A dict naming a DC the cluster does not span is rejected here rather than left to fail + later: such a DC is skipped by :meth:`MdcCluster.network_registry`, so its nodes would + run with no impairments and no partition rules while every other call site kept working. + """ + if not isinstance(value, dict): + return {dc: value for dc in dcs} + + unknown = sorted(dc for dc in value if dc not in dcs) + + assert not unknown, \ + f"Per-DC counts name data centers the cluster does not span [unknown={unknown}, dcs={list(dcs)}]" + + return dict(value) + + +def _as_segment(segment: Segment) -> Tuple[str, ...]: """ - return dict(value) if isinstance(value, dict) else {dc: value for dc in DCS} + Normalizes a single DC name or a collection of DC names into a tuple of DC names. + """ + return (segment,) if isinstance(segment, str) else tuple(segment) + + +def _fmt_segment(segment: Tuple[str, ...]) -> str: + """ + :return: Segment rendered for an assertion message, e.g. "DC1+DC2" - a Python tuple + reads poorly in the middle of one, and a single DC renders as itself. + """ + return "+".join(segment) class MdcCluster: @@ -89,28 +193,42 @@ class MdcCluster: :param test: The ducktape test instance. :param ignite_version: Ignite version string. + :param dcs: Data centers the cluster spans, two by default. The count selects the + topology validator mode - see :func:`mdc_topology_params`. + :param main_dc: Main DC for an even-sized DC set, defaults to the first DC. :param srv_per_dc: Servers per DC, an int or a per-DC dict (asymmetric DCs). :param runners_per_dc: Reusable run-to-completion app services per DC (generator, checkers, load bursts). An int or a per-DC dict. :param loaders_per_dc: Dedicated background load app services per DC. They run concurrently with runner apps, hence separate containers. :param client_connector: Whether to expose the thin client connector on servers. + :param jmx_metrics: Whether to export the node metrics over JMX. Required by everything + that reads one - see :meth:`cache_mdc_metrics`. """ - def __init__(self, test, ignite_version: str, srv_per_dc: Union[int, Dict[str, int]] = 3, + def __init__(self, test, ignite_version: str, dcs: Sequence[str] = DCS_2, + main_dc: Optional[str] = None, + srv_per_dc: Union[int, Dict[str, int]] = 3, runners_per_dc: Union[int, Dict[str, int]] = 1, loaders_per_dc: Union[int, Dict[str, int]] = 0, client_connector: bool = False, + jmx_metrics: bool = False, network_timeout: int = 5_000, tcp_connect_timeout: int = 5_000): self.test_context = test.test_context self.logger = test.logger - # A single discovery SPI (hence a single ip finder) shared by both DCs' server - # services is what makes the two DCs form ONE cluster: prepare_on_start() - # memoizes the addresses of the first started DC into the shared ip finder, so - # the second DC discovers through the first DC's nodes, and restart() re-joins - # the same way. Restarting the first started DC itself is the one case this - # breaks - see sync_service_discovery(). + self.dcs = tuple(dcs) + + assert len(self.dcs) >= 2, f"An MDC cluster spans at least two data centers [dcs={self.dcs}]" + + self.main_dc = main_dc if main_dc is not None else self.dcs[0] + + # A single discovery SPI (hence a single ip finder) shared by all DCs' server + # services is what makes the DCs form ONE cluster: prepare_on_start() memoizes the + # addresses of the first started DC into the shared ip finder, so every later DC + # discovers through the first DC's nodes, and restart() re-joins the same way. + # Restarting the first started DC itself is the one case this breaks - see + # sync_service_discovery(). cfg_kwargs = { "version": IgniteVersion(ignite_version), "discovery_spi": TcpDiscoverySpi(), @@ -121,25 +239,27 @@ def __init__(self, test, ignite_version: str, srv_per_dc: Union[int, Dict[str, i if client_connector: cfg_kwargs["client_connector_configuration"] = ClientConnectorConfiguration() + if jmx_metrics: + # A fresh set, never the shared mutable default of IgniteConfiguration. + cfg_kwargs["metric_exporters"] = {JMX_METRIC_EXPORTER} + self.ignite_config = IgniteConfiguration(**cfg_kwargs) - self.srv_per_dc = _per_dc(srv_per_dc) + self.srv_per_dc = per_dc(srv_per_dc, self.dcs) self.servers: Dict[str, IgniteService] = { - dc: IgniteService(self.test_context, self.ignite_config, num_nodes=num, jvm_opts=dc_jvm_opts(dc), - startup_timeout_sec=IGNITE_STARTUP_TIMEOUT_SEC) - for dc, num in self.srv_per_dc.items() if num > 0} + dc: self._server_service(dc, num) for dc, num in self.srv_per_dc.items() if num > 0} self.runners: Dict[str, List[IgniteApplicationService]] = { dc: [self._app_service(dc) for _ in range(num)] - for dc, num in _per_dc(runners_per_dc).items()} + for dc, num in per_dc(runners_per_dc, self.dcs).items()} self.loaders: Dict[str, List[IgniteApplicationService]] = { dc: [self._app_service(dc) for _ in range(num)] - for dc, num in _per_dc(loaders_per_dc).items()} + for dc, num in per_dc(loaders_per_dc, self.dcs).items()} # Extra services (e.g. thin clients) registered into a DC's network group. - self.extras: Dict[str, List] = {dc: [] for dc in DCS} + self.extras: Dict[str, List] = {dc: [] for dc in self.dcs} # App services that have been started at least once: the first start is clean, # subsequent ones preserve work dirs (and logs - hence unique result prefixes). @@ -148,21 +268,66 @@ def __init__(self, test, ignite_version: str, srv_per_dc: Union[int, Dict[str, i # Admissibility checks run on reusable services, so each check needs a unique result prefix. self._adm_checks = 0 + # Cache parameters applied to every cache this fixture creates, unless a call overrides them. + self.cache_defaults = { + "topologyValidator": self.test_context.globals.get(CACHE_TOP_VALIDATOR_GLOBAL, True) + } + + self.logger.info(f"MDC cache defaults [{self.cache_defaults}]") + + def _server_service(self, dc: str, num_nodes: int) -> IgniteService: + """ + Builds the server service of one DC. The single place a server command line is put + together, so a fork that starts its servers differently overrides just this. + """ + return IgniteService(self.test_context, self.ignite_config, num_nodes=num_nodes, + jvm_opts=dc_jvm_opts(dc), startup_timeout_sec=IGNITE_STARTUP_TIMEOUT_SEC) + + def dc_servers(self, dc: str) -> List[IgniteService]: + """ + :return: All server services of the given DC - one, unless a subclass splits a DC's + servers into several services (node groups, cells, availability zones). + """ + return [self.servers[dc]] if dc in self.servers else [] + + def all_servers(self) -> List[IgniteService]: + """ + :return: Every server service of the cluster, DCs in order. + """ + return [svc for dc in sorted(self.servers) for svc in self.dc_servers(dc)] + + @property + def min_backups(self) -> int: + """ + :return: Smallest backup count giving every DC one copy of every partition + (2 for a three DC cluster, 1 for a two DC one). + """ + return min_backups(self.dcs) + + def topology_params(self) -> dict: + """ + :return: Cache parameters pinning the topology validator and the affinity backup + filter to this cluster's DC set. + """ + return mdc_topology_params(self.dcs, self.main_dc) + def sync_service_discovery(self): """ Points every server service at a discovery SPI covering all DCs. Required before restarting the FIRST started DC: the shared ip finder holds only that DC's addresses, so after a full stop its nodes would seed off themselves and - form a separate cluster instead of rejoining the surviving DC. + form a separate cluster instead of rejoining the surviving DCs. """ - discovery_spi = from_ignite_services(list(self.servers.values())) + discovery_spi = from_ignite_services(self.all_servers()) - for service in self.servers.values(): + for service in self.all_servers(): service.config = service.config._replace(discovery_spi=discovery_spi) def _app_service(self, dc: str) -> IgniteApplicationService: - client_cfg = self.ignite_config._replace(client_mode=True, discovery_spi=from_ignite_cluster(self.servers[dc])) + # Seeding off the DC's first server service is enough: all of them are one cluster. + client_cfg = self.ignite_config._replace(client_mode=True, + discovery_spi=from_ignite_cluster(self.dc_servers(dc)[0])) return IgniteApplicationService(self.test_context, client_cfg, jvm_opts=dc_jvm_opts(dc)) @@ -180,11 +345,8 @@ def network_registry(self) -> Dict[str, List]: """ registry = {} - for dc in DCS: - services = [] - - if dc in self.servers: - services.append(self.servers[dc]) + for dc in self.dcs: + services = list(self.dc_servers(dc)) services += self.runners.get(dc, []) services += self.loaders.get(dc, []) @@ -208,9 +370,9 @@ def describe(self) -> List[str]: """ lines = ["DATA CENTERS"] - for dc in DCS: + for dc in self.dcs: roles = [(label, [node.account.hostname for svc in services for node in svc.nodes]) - for label, services in (("server", [self.servers[dc]] if dc in self.servers else []), + for label, services in (("server", self.dc_servers(dc)), ("runner", self.runners.get(dc, [])), ("loader", self.loaders.get(dc, [])), ("extra", self.extras.get(dc, [])))] @@ -231,33 +393,58 @@ def thin_client_addresses(self) -> List[str]: """ port = self.ignite_config.client_connector_configuration.port - return [f"{node.account.hostname}:{port}" - for dc in sorted(self.servers) for node in self.servers[dc].nodes] + return [f"{node.account.hostname}:{port}" for svc in self.all_servers() for node in svc.nodes] def start_servers(self): """ Starts all server services. """ - for dc in sorted(self.servers): - self.servers[dc].start() + for svc in self.all_servers(): + svc.start() def stop_servers(self): """ Stops all server services. """ - for dc in sorted(self.servers): - self.servers[dc].stop() + for svc in self.all_servers(): + svc.stop() + + def stop_dcs(self, *dcs: str): + """ + Stops the server services of the given DCs, in the order given - a data center + outage, as opposed to the network partition :func:`cross_dc_network` produces. + """ + for dc in dcs: + for svc in self.dc_servers(dc): + svc.stop() + + def start_dcs(self, *dcs: str, clean: bool = False, await_rebalance: bool = True): + """ + Starts the given DCs back, by default preserving their persistence, and waits until + the cluster has rebalanced onto them. + + Every DC is started before the first wait, so that their joins are not serialized + behind each other's rebalance. + + Restarting the FIRST started DC needs :meth:`sync_service_discovery` beforehand. + """ + for dc in dcs: + for svc in self.dc_servers(dc): + svc.start(clean=clean) + + if await_rebalance: + for dc in dcs: + for svc in self.dc_servers(dc): + svc.await_rebalance() def restart(self, dc: str, clean: bool = False, await_rebalance: bool = True): """ Restarts a whole DC preserving its persistence (the pattern used to rejoin a - read-only half-ring back into the main cluster after a partition heals). + read-only segment back into the main cluster after a partition heals). """ - self.servers[dc].stop() - self.servers[dc].start(clean=clean) + self.stop_dcs(dc) - if await_rebalance: - self.servers[dc].await_rebalance() + self.start_dcs(dc, clean=clean, await_rebalance=await_rebalance) def run_app(self, dc: str, java_class: str, params: dict, runner: int = 0) -> IgniteApplicationService: """ @@ -276,7 +463,7 @@ def run_service(self, svc: IgniteApplicationService, params: dict, if java_class is not None: svc.java_class_name = java_class - svc.params = params + svc.params = self._with_cache_params(params) svc.start(clean=self._first_start(svc)) svc.wait() @@ -288,12 +475,14 @@ def start_loader(self, dc: str, params: dict, loader: int = 0, java_class: str = LOAD_APP) -> IgniteApplicationService: """ Starts a background load application (runs until stopped). Any exception raised - by the application surfaces in :meth:`stop_loader`. + by the application surfaces in :meth:`stop_loader`. A load that creates the cache + (``createCache``) has the MDC cache parameters injected - see + :meth:`_with_cache_params`. """ svc = self.loaders[dc][loader] svc.java_class_name = java_class - svc.params = params + svc.params = self._with_cache_params(params) svc.start(clean=self._first_start(svc)) @@ -310,6 +499,24 @@ def stop_loader(self, dc: str, loader: int = 0) -> IgniteApplicationService: return svc + def _with_cache_params(self, params: dict, creates_cache: bool = False) -> dict: + """ + Injects everything the MDC cache is configured from - the topology validator mode, + the DC count the affinity backup filter needs, and :attr:`cache_defaults` - into + the parameters of an application that creates it, so no call site can configure a + cache that disagrees with the DC set. Explicit parameters still win. + + The single injection point for all of it: an application that does not create the + cache is handed none of it, since it would only ever be ignored. + + :param creates_cache: Whether the application always creates the cache. The ones + that decide at run time say so with a ``createCache`` parameter instead. + """ + if not (creates_cache or params.get("createCache")): + return params + + return {**self.topology_params(), **self.cache_defaults, **params} + def _first_start(self, svc) -> bool: first = id(svc) not in self._started_apps @@ -317,17 +524,25 @@ def _first_start(self, svc) -> bool: return first - def generate_data(self, dc: str, cache_name: str, from_idx: int, to_idx: int, backups: int, - main_dc: str = DC_1, sql_mode: bool = False, **cache_params) -> IgniteApplicationService: + def generate_data(self, dc: str, cache_name: str, from_idx: int, to_idx: int, backups: Optional[int] = None, + sql_mode: bool = False, **cache_params) -> IgniteApplicationService: """ Creates the MDC cache (if absent) and populates keys ``[from_idx, to_idx)``. Extra cache parameters (``atomicity``, ``writeSync``, ``readFromBackup``, - ``partitions``, ...) are passed through to the cache configuration builder. + ``partitions``, ...) are passed through to the cache configuration builder, on top + of the MDC cache parameters - see :meth:`_with_cache_params`. + + :param backups: Backup count, by default the smallest one that gives every DC a + single copy of every partition (see :attr:`min_backups`). """ - params = {"cacheName": cache_name, "backups": backups, "mainDc": main_dc, - "from": from_idx, "to": to_idx, "sqlMode": sql_mode, **cache_params} + params = {"cacheName": cache_name, + "backups": self.min_backups if backups is None else backups, + "from": from_idx, "to": to_idx, "sqlMode": sql_mode, + **cache_params} - return self.run_app(dc, GENERATOR_APP, params) + # The generator always creates the cache, so it carries no createCache parameter + # for _with_cache_params() to key off. + return self.run_app(dc, GENERATOR_APP, self._with_cache_params(params, creates_cache=True)) def check_data(self, dc: str, cache_name: str, from_idx: int, to_idx: int) -> Optional[IgniteApplicationService]: """ @@ -347,8 +562,8 @@ def check_data(self, dc: str, cache_name: str, from_idx: int, to_idx: int) -> Op def check_put_admissibility(self, dc: str, cache_name: str, admissible: bool, key_offset: int = 1_000_000, probes: int = 100) -> IgniteApplicationService: """ - Verifies that put load from the given DC is admissible (primary DC visible) or - rejected by the topology validator (read-only DC). A PUT burst of the load + Verifies that put load from the given DC is admissible (the segment passes the + topology validator) or rejected by it (read-only segment). A PUT burst of the load application: an admissible check fails fast on the first rejected put, an inadmissible check fails if any of the probe puts succeeds. @@ -366,18 +581,66 @@ def run_load(self, dc: str, mode: str, cache_name: str, result_prefix: str, """ Runs a load burst (see ``MdcContinuousLoadApplication``) and returns the service. ``result_prefix`` must be unique per burst because runner services are reused. + A burst that creates the cache (``createCache``) has the MDC cache parameters + injected - see :meth:`_with_cache_params`. """ load_params = {"mode": mode, "cacheName": cache_name, "resultPrefix": result_prefix, **params} return self.run_app(dc, LOAD_APP, load_params, runner=runner) - def control(self, dc: str = DC_1) -> ControlUtility: + def control(self, dc: Optional[str] = None) -> ControlUtility: + """ + :return: Control utility bound to the given DC's servers, the first DC by default. """ - :return: Control utility bound to the given DC's servers. + return ControlUtility(self.dc_servers(dc if dc is not None else self.dcs[0])[0]) + + def cache_mdc_metrics(self, cache_name: str, dc: Optional[str] = None) -> Dict[str, bool]: """ - return ControlUtility(self.servers[dc]) + Reads the cache's MDC safety metrics off a server node over JMX - see + :data:`MDC_SAFE_AFFINITY_METRIC` and :data:`MDC_SAFE_DISTRIBUTION_METRIC` for what + each of them claims. + + Requires the cluster to have been built with ``jmx_metrics=True``, since the metrics + are only exposed by the JMX metric exporter. - def verify_cache_distribution(self, cache_name: str, copies_per_dc: Optional[int] = None, dc: str = DC_1): + :param cache_name: Cache to read the metrics of. + :param dc: DC whose node answers, the first one by default. Every server node reports + the same verdict, so this only matters for a partitioned cluster - where each + segment answers about the topology IT can see. + :return: Metric name -> value. + """ + node = next(node for svc in self.dc_servers(dc if dc is not None else self.dcs[0]) + for node in svc.alive_nodes) + + mbean = node.cache_mbean(cache_name) + + return {name: mbean.bool_value(name) + for name in (MDC_SAFE_AFFINITY_METRIC, MDC_SAFE_DISTRIBUTION_METRIC)} + + def verify_cache_mdc_metrics(self, cache_name: str, affinity_safe: Optional[bool] = None, + distribution_safe: Optional[bool] = None, dc: Optional[str] = None): + """ + Verifies the MDC safety metrics of a cache against what the scenario expects. Both + expectations are optional: a metric left as None is only reported, which is how a + scenario reads out a value whose verdict depends on the state of the topology rather + than on the point being made. + + :return: The metrics that were read. + """ + metrics = self.cache_mdc_metrics(cache_name, dc) + + self.logger.info(f"MDC safety metrics [cache={cache_name}, dc={dc}, {metrics}]") + + for name, expected in ((MDC_SAFE_AFFINITY_METRIC, affinity_safe), + (MDC_SAFE_DISTRIBUTION_METRIC, distribution_safe)): + if expected is not None: + assert metrics[name] == expected, \ + f"{name} should be {expected} [cache={cache_name}, actual={metrics[name]}]" + + return metrics + + def verify_cache_distribution(self, cache_name: str, copies_per_dc: Optional[int] = None, + dc: Optional[str] = None): """ Verifies that every partition of the cache has an OWNING copy in every DC, and optionally that each DC holds exactly ``copies_per_dc`` copies. @@ -387,77 +650,98 @@ def verify_cache_distribution(self, cache_name: str, copies_per_dc: Optional[int distribution = self.control(dc).cache_distribution(cache_names=cache_name, user_attributes=DATA_CENTER_ATTR) assert_cross_dc_distribution_by_attribute(distribution, dc_attr=DATA_CENTER_ATTR, - expected_dcs=DCS, copies_per_dc=copies_per_dc) + expected_dcs=self.dcs, copies_per_dc=copies_per_dc) return distribution def verify_split_brain(self): """ - Verifies that after the network partition the cluster has split into two independent - half-rings: their baselines don't intersect and each half elected its own coordinator. + Verifies that the network partition split the cluster into as many independent + segments as there are DCs, i.e. every DC ended up on its own. + """ + self.verify_segments(*self.dcs) + + def verify_segments(self, *segments: Segment): + """ + Verifies that the cluster has split into exactly the given independent segments: + every segment is healthy on its own, no two segments share a baseline node, and + each segment elected its own coordinator. + + A segment is a DC name or a collection of DC names that still see each other, e.g. + ``verify_segments((DC_1, DC_2), DC_3)`` for a cluster with DC3 cut off. """ - for dc in DCS: - self.verify_half_ring_healthy(dc) + normalized = [_as_segment(segment) for segment in segments] - state = {dc: self.control(dc).cluster_state() for dc in DCS} + # The state each segment is checked healthy against is the same one its baseline and + # coordinator are read from: a partitioned segment answers control.sh over the very + # links the test just cut, so it is fetched once per segment and passed around. + states = {segment: self.verify_segment_healthy(segment) for segment in normalized} - baselines = {dc: {node.consistent_id for node in state[dc].baseline} for dc in DCS} + baselines = {segment: {node.consistent_id for node in states[segment].baseline} for segment in normalized} - common_nodes = baselines[DC_1] & baselines[DC_2] + for seg_a, seg_b in combinations(normalized, 2): + common_nodes = baselines[seg_a] & baselines[seg_b] - assert not common_nodes, \ - f"Half-ring baselines should not intersect " \ - f"[common={sorted(common_nodes)}, dc1={sorted(baselines[DC_1])}, dc2={sorted(baselines[DC_2])}]" + assert not common_nodes, \ + f"Segment baselines should not intersect [common={sorted(common_nodes)}, " \ + f"{_fmt_segment(seg_a)}={sorted(baselines[seg_a])}, " \ + f"{_fmt_segment(seg_b)}={sorted(baselines[seg_b])}]" - for dc in DCS: - coordinator = state[dc].coordinator + coordinators = {} - assert coordinator, f"Coordinator is not found in {dc} half-ring baseline output!" + for segment in normalized: + coordinator = states[segment].coordinator - assert coordinator.consistent_id in baselines[dc], \ - f"{dc} coordinator should belong to its own half-ring baseline " \ - f"[coordinator={coordinator.consistent_id}, baseline={sorted(baselines[dc])}]" + assert coordinator, \ + f"Coordinator is not found in the {_fmt_segment(segment)} segment baseline output!" - assert state[DC_1].coordinator.consistent_id != state[DC_2].coordinator.consistent_id, \ - f"Half-rings should have different coordinators " \ - f"[coordinator={state[DC_1].coordinator.consistent_id}]" + assert coordinator.consistent_id in baselines[segment], \ + f"{_fmt_segment(segment)} coordinator should belong to its own segment baseline " \ + f"[coordinator={coordinator.consistent_id}, baseline={sorted(baselines[segment])}]" - def verify_half_ring_healthy(self, dc: str): + coordinators[_fmt_segment(segment)] = coordinator.consistent_id + + assert len(set(coordinators.values())) == len(normalized), \ + f"Every segment should have elected its own coordinator [coordinators={coordinators}]" + + def verify_segment_healthy(self, segment: Segment): """ - Verifies that a half-ring is fully alive, ACTIVE, and its baseline matches its size. + Verifies that a segment is fully alive, ACTIVE, and its baseline covers exactly + the servers of the DCs it consists of - and nothing else. + + :return: The ClusterState the segment was verified against, so that a caller + asserting further on it (see :meth:`verify_segments`) needs no second + control.sh round-trip into a segment that may be cut off. """ - exp_alive_nodes = self.srv_per_dc[dc] - act_alive_nodes = len(self.servers[dc].alive_nodes) + dcs = _as_segment(segment) + + name = _fmt_segment(dcs) + + # get(): a per-DC dict is allowed to name only the DCs it populates, and + # verify_whole_cluster_healthy() asks about every DC the cluster spans. + exp_alive_nodes = sum(self.srv_per_dc.get(dc, 0) for dc in dcs) + act_alive_nodes = sum(len(svc.alive_nodes) for dc in dcs for svc in self.dc_servers(dc)) assert act_alive_nodes == exp_alive_nodes, \ - f"{exp_alive_nodes} nodes should be alive in {dc}! [actual={act_alive_nodes}]" + f"{exp_alive_nodes} nodes should be alive in {name}! [actual={act_alive_nodes}]" - cluster_state = self.control(dc).cluster_state() + cluster_state = self.control(dcs[0]).cluster_state() assert "ACTIVE" == cluster_state.state, \ - f"{dc} half-ring state should remain ACTIVE [actual={cluster_state.state}]" + f"{name} segment state should remain ACTIVE [actual={cluster_state.state}]" assert len(cluster_state.baseline) == exp_alive_nodes, \ - f"{dc} half-ring baseline is not expected " \ + f"{name} segment baseline is not expected " \ f"[exp={exp_alive_nodes}, actual_baseline={cluster_state.baseline}]" + return cluster_state + def verify_whole_cluster_healthy(self): """ - Verifies that both DCs form a single ACTIVE cluster: every server node is alive - and the baseline seen from DC1 covers all servers of both DCs. + Verifies that all DCs form a single ACTIVE cluster: every server node is alive + and the baseline seen from the first DC covers all servers of every DC. """ - exp_total = sum(self.srv_per_dc.values()) - - act_alive = sum(len(self.servers[dc].alive_nodes) for dc in self.servers) - - assert act_alive == exp_total, f"All {exp_total} server nodes should be alive [actual={act_alive}]" - - cluster_state = self.control(DC_1).cluster_state() - - assert "ACTIVE" == cluster_state.state, f"Cluster should be ACTIVE [actual={cluster_state.state}]" - - assert len(cluster_state.baseline) == exp_total, \ - f"Cluster baseline should cover both DCs [exp={exp_total}, actual={cluster_state.baseline}]" + self.verify_segment_healthy(self.dcs) def verify_servers_log_clean(self): """ @@ -465,10 +749,10 @@ def verify_servers_log_clean(self): were detected, no PME hang and no lost partitions were reported. """ for pattern in (LRT_PATTERN, PME_FREEZE_PATTERN, LOST_PARTITIONS_PATTERN, ASSERTION_ERROR_PATTERN): - for svc in self.servers.values(): + for svc in self.all_servers(): svc.check_event_absent(pattern, log_file=ALL_LOGS_GLOB) - def verify_no_hanging_txs(self, dc: str = DC_1, try_kill_hanging_tx: bool = False): + def verify_no_hanging_txs(self, dc: Optional[str] = None, try_kill_hanging_tx: bool = False): """ Verifies that no active transactions are left on the cluster. """ @@ -509,20 +793,25 @@ def result_bool(svc: IgniteApplicationService, name: str) -> bool: def cross_dc_network(logger, mdc: MdcCluster, delay_ms: Optional[int] = None, loss: Optional[float] = None) -> NetworkGroupManager: """ - Builds a :class:`NetworkGroupManager` (context manager) for the cluster with symmetric - DC1 <-> DC2 impairments. With no impairments the manager still owns partition + Builds a :class:`NetworkGroupManager` (context manager) for the cluster, applying the + same impairment to every DC pair. With no impairments the manager still owns partition enable/disable and the final network cleanup. + A cluster whose links are not all alike needs no fixture support: build the + :class:`NetworkGroupStore` and construct the manager directly, the registry is all it + takes from here - ``NetworkGroupManager(logger, store, mdc.network_registry())``. + :param delay_ms: One-way cross-DC latency in milliseconds (the effective RTT is twice that, since netem delay is applied on egress in both directions). :param loss: Cross-DC packet loss fraction in [0.0, 1.0]. """ - store = NetworkGroupStore() - cfg = CrossNetworkGroupConfiguration(delay=f"{delay_ms}ms" if delay_ms is not None else None, loss=loss) + store = NetworkGroupStore() + if not cfg.is_empty: - store.set_config(DC_1, DC_2, cfg) + for dc_a, dc_b in all_pairs(mdc.dcs): + store.set_config(dc_a, dc_b, cfg) return NetworkGroupManager(logger, store, mdc.network_registry()) @@ -545,45 +834,65 @@ def assert_cross_dc_distribution_by_attribute(distribution, dc_attr, expected_dc def dc_of(copy): return copy.user_attributes.get(dc_attr) - _assert_cross_dc(distribution, set(expected_dcs), dc_of, owning_only, copies_per_dc, - layout_hint=f"DC attribute: {dc_attr}, expected DCs: {sorted(expected_dcs)}") + assert_spread(distribution, set(expected_dcs), dc_of, owning_only, copies_per_dc, label="DC", + layout_hint=f"DC attribute: {dc_attr}, expected DCs: {sorted(expected_dcs)}") -def _assert_cross_dc(distribution, expected_dcs, dc_of, owning_only, copies_per_dc, layout_hint): +def assert_spread(distribution, expected_groups, group_of, owning_only, copies_per_group, label, + layout_hint): + """ + Asserts that every partition of every cache group has a copy in every expected group, + where a group is whatever ``group_of(copy)`` returns. + + Public because the grouping is the only thing that varies: the DC spread above is one + ``group_of``, and an affinity backup filter that spreads copies by something finer - a + cell, an availability zone, a tuple of node attributes - is another. Such a check reuses + this rather than walking the distribution again. + + :param distribution: CacheDistribution requested with the attributes ``group_of`` reads. + :param expected_groups: Group keys that must each own a copy of every partition. + :param group_of: Copy -> its group key. + :param owning_only: Count only copies in OWNING state as present. + :param copies_per_group: If set, each group must hold exactly this many copies. + :param label: What a group is called in the assertion message, e.g. "DC". + :param layout_hint: Line appended to the message, naming the layout that was expected. + """ violations = [] for group in distribution.groups.values(): for part, copies in sorted(group.partitions.items()): counted = [c for c in copies if not owning_only or c.state == "OWNING"] - per_dc = {dc: 0 for dc in expected_dcs} + per_group = {key: 0 for key in expected_groups} for copy in counted: - dc = dc_of(copy) + key = group_of(copy) - if dc in per_dc: - per_dc[dc] += 1 + if key in per_group: + per_group[key] += 1 - missing = {dc for dc, cnt in per_dc.items() if cnt == 0} + missing = {key for key, cnt in per_group.items() if cnt == 0} - unbalanced = {} if copies_per_dc is None else \ - {dc: cnt for dc, cnt in per_dc.items() if cnt != copies_per_dc} + unbalanced = {} if copies_per_group is None else \ + {key: cnt for key, cnt in per_group.items() if cnt != copies_per_group} if missing or unbalanced: copies_dump = ", ".join( - f"{c.node_id}({'P' if c.primary else 'B'},{c.state},dc={dc_of(c)},{c.node_addresses})" + f"{c.node_id}({'P' if c.primary else 'B'},{c.state},{label}={group_of(c)}," + f"{c.node_addresses})" for c in copies) problems = [] if missing: - problems.append(f"missing DCs={sorted(missing)}") + problems.append(f"missing {label}s={sorted(missing)}") if unbalanced: - problems.append(f"copies per DC != {copies_per_dc}: {unbalanced}") + problems.append(f"copies per {label} != {copies_per_group}: {unbalanced}") violations.append(f"group={group.name}(id={group.group_id}), partition={part}, " f"{', '.join(problems)}, copies=[{copies_dump}]") assert not violations, \ - "Partition distribution is not cross-DC:\n " + "\n ".join(violations) + "\n" + layout_hint + f"Partition distribution does not cover every {label}:\n " + "\n ".join(violations) + \ + "\n" + layout_hint diff --git a/modules/ducktests/tests/ignitetest/services/network_group/manager.py b/modules/ducktests/tests/ignitetest/services/network_group/manager.py index 819212456bde7..493185d2d5a9d 100644 --- a/modules/ducktests/tests/ignitetest/services/network_group/manager.py +++ b/modules/ducktests/tests/ignitetest/services/network_group/manager.py @@ -44,7 +44,7 @@ MAX_PARALLEL_SSH_SESSIONS = 16 # Separates the qdisc, filter and iptables sections in the output of the batched -# network probe issued by _log_network. +# network probe issued by _probe_network. PROBE_SECTION_SEPARATOR = "=== ignitetest network probe section ===" # A rule spec: (src_group, dst_group, action, config). @@ -128,23 +128,36 @@ def enable_network_partition(self, group_a: str, group_b: str): simulating a split-brain. The netem impairments deployed via tcset are left untouched underneath. """ - self.logger.info(f"Enabling network partition between [{group_a}] <---> [{group_b}]") + self.enable_network_partitions((group_a, group_b)) - chain = partition_chain_name(group_a, group_b) + def enable_network_partitions(self, *group_pairs: Tuple[str, str]): + """ + Cuts several group pairs apart at once, e.g. to isolate one group from + every other one or to split a three group cluster three ways. - tasks = [] + Every chain a node takes part in is installed by the same single SSH + round-trip, so a multi-way split takes effect near-atomically instead of + rolling out pair by pair - the intermediate topologies of a pair-by-pair + rollout are themselves valid segmentations the cluster would react to. + """ + pairs_str = ", ".join(f"[{a}] <---> [{b}]" for a, b in group_pairs) + + self.logger.info(f"Enabling network partition between {pairs_str}") + + per_node = {} - for src_group, dst_group in self._bidirectional(group_a, group_b): - remote_ips = self._resolve_group_ips(dst_group) + for group_a, group_b in group_pairs: + chain = partition_chain_name(group_a, group_b) - cmd = to_partition_enable_cmd(chain, remote_ips) + for src_group, dst_group in self._bidirectional(group_a, group_b): + cmd = to_partition_enable_cmd(chain, self._resolve_group_ips(dst_group)) - for node in self._iter_group_nodes(src_group): - tasks.append((node, cmd)) + for node in self._iter_group_nodes(src_group): + per_node.setdefault(id(node), (node, []))[1].append(cmd) - self._ssh_parallel(tasks, tag="PARTITION_ON") + self._ssh_parallel([(node, " && ".join(cmds)) for node, cmds in per_node.values()], tag="PARTITION_ON") - self._log_network(f"PARTITION {group_a} <-> {group_b}") + self._log_network(f"PARTITION {pairs_str}") def disable_network_partition(self, group_a: str, group_b: str): """ @@ -153,17 +166,31 @@ def disable_network_partition(self, group_a: str, group_b: str): node. The originally deployed tcset impairments were never modified, so they are back in effect immediately without any re-application. """ - self.logger.info(f"Disabling network partition between [{group_a}] <---> [{group_b}]") + self.disable_network_partitions((group_a, group_b)) + + def disable_network_partitions(self, *group_pairs: Tuple[str, str]): + """ + Heals several partitions at once, one SSH round-trip per node - the + counterpart of :meth:`enable_network_partitions`. + """ + pairs_str = ", ".join(f"[{a}] <---> [{b}]" for a, b in group_pairs) + + self.logger.info(f"Disabling network partition between {pairs_str}") + + per_node = {} - cmd = to_partition_disable_cmd(partition_chain_name(group_a, group_b)) + for group_a, group_b in group_pairs: + cmd = to_partition_disable_cmd(partition_chain_name(group_a, group_b)) - tasks = [(node, cmd) - for group in (group_a, group_b) - for node in self._iter_group_nodes(group)] + for group in (group_a, group_b): + for node in self._iter_group_nodes(group): + per_node.setdefault(id(node), (node, []))[1].append(cmd) - self._ssh_parallel(tasks, tag="PARTITION_OFF") + # ';' rather than '&&': healing is best effort, and a chain that was never installed + # on this node must not stop the rest of its chains from being flushed. + self._ssh_parallel([(node, " ; ".join(cmds)) for node, cmds in per_node.values()], tag="PARTITION_OFF") - self._log_network(f"NET RESTORED {group_a} <-> {group_b}") + self._log_network(f"NET RESTORED {pairs_str}") def _resolve_group_ips(self, group: str) -> List[str]: return [socket.gethostbyname(node.account.externally_routable_ip) diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py b/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py index 4b1d2e3555cfe..6af2969f09ad0 100644 --- a/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py +++ b/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py @@ -88,10 +88,11 @@ def test_transactional_load_cut_on_partition(self, ignite_version, cross_dc_late "keyTo": LOAD_KEY_TO_DC_2, "stopOnError": True, "resultPrefix": f"txLoad{DC_2}", + # The MDC topology parameters (validator mode, DC count) are injected by + # the fixture, so the cache this load creates matches the cluster's DC set. "createCache": True, "backups": BACKUPS, "atomicity": "TRANSACTIONAL", - "mainDc": DC_1, "txTimeout": 5_000 }) From 880715fdb9e68ef553320ee3c1cb05342eff16cd Mon Sep 17 00:00:00 2001 From: Maksim Davydov Date: Thu, 10 Sep 2026 16:50:02 +0300 Subject: [PATCH 4/8] IGNITE-28952 [ducktests] Three data center MDC tests for majority based validation Three scenarios over a three DC cluster, each asserting something two DC mode cannot show: * test_minority_dc_isolation - one DC is cut off. The two DCs left keep writing, the isolated one goes read-only while still serving every read. Parametrized over the isolated DC, because no DC is privileged here: cutting off DC1 - the main DC of the equivalent two DC cluster, and the DC the others discovered through - is just as survivable as cutting off DC3. * test_three_way_split_blocks_all_writes - every cross-DC link drops at once. No segment holds a majority, so unlike the two DC case, where one half always survives as writable, the whole cluster goes read-only. * test_writes_survive_single_dc_loss - the same guarantee without any network impairment: losing one DC leaves a majority, losing a second one does not. Every partition owns exactly one copy per DC (backups = 2), so every segment down to a single isolated DC still serves every read - which is what separates the read assertions from the write ones throughout. test_minority_dc_isolation also reads the cluster's own verdict on the MDC guarantees off a node over JMX. IsCacheAffinityConfigurationMdcSafe is about the cache CONFIGURATION and holds in every segment; IsCachePartitionDistributionSafe is about the current assignment, and is only reported while the cluster is split until the expected value is confirmed by a run. --- .../tests/mdc/majority_partition_test.py | 255 ++++++++++++++++++ 1 file changed, 255 insertions(+) create mode 100644 modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py b/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py new file mode 100644 index 0000000000000..5b97a441bbc2b --- /dev/null +++ b/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py @@ -0,0 +1,255 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Three data center MDC resilience: majority based topology validation. + +With an ODD number of data centers MdcTopologyValidator switches from "the main DC must +be visible" to "a majority of the DC set must be visible". That is a different guarantee +and not a bigger version of the two DC one: + + - no data center is privileged - the cluster keeps accepting writes through the loss of + ANY single DC, including the first one, which two DC mode cannot do; + - a segment holding a minority of DCs goes read-only even though it is perfectly + healthy internally - so a three way split leaves NO writable segment at all. + +Every partition owns exactly one copy per DC (backups = 2), so every segment - down to a +single isolated DC - can still serve every read. That is what separates the read +assertions from the write ones throughout these tests. +""" +from time import sleep + +from ducktape.mark import parametrize + +from ignitetest.services.mdc.mdc_cluster import MdcCluster, cross_dc_network, all_pairs, isolation_pairs, \ + DCS_3, DC_1, DC_2, DC_3 +from ignitetest.utils import cluster, ignite_versions +from ignitetest.utils.ignite_test import IgniteTest +from ignitetest.utils.version import DEV_BRANCH + +CACHE_NAME = "mdc-majority" + +# Keys generated from each DC; the whole data set is [0, KEYS_PER_DC * len(DCS_3)). +KEYS_PER_DC = 100 + +# Time for discovery to detect the partition and for every segment to complete PME. +SPLIT_SETTLE_SECS = 20 + +# Probe key ranges of the put admissibility checks. Well above the data set, and one range +# per check, so that the writes an admissible check performs never collide. +PROBE_BASE = 1_000_000 +PROBE_STRIDE = 1_000_000 + + +def _probe_offset(idx: int) -> int: + return PROBE_BASE + idx * PROBE_STRIDE + + +class MdcMajorityPartitionTest(IgniteTest): + """ + Tests for majority based MDC topology validation across three data centers. + """ + @cluster(num_nodes=9) + @ignite_versions(str(DEV_BRANCH)) + @parametrize(cross_dc_latency_ms=100, isolated_dc=DC_3) + @parametrize(cross_dc_latency_ms=100, isolated_dc=DC_1) + def test_minority_dc_isolation(self, ignite_version, cross_dc_latency_ms, isolated_dc): + """ + One DC is cut off from the two others. The majority segment must stay writable and + the isolated one must go read-only while still serving every read. + + Parametrized over the isolated DC to assert that no DC is privileged: isolating DC1 + - the main DC of the equivalent two DC cluster, and the DC the others discovered + through - is just as survivable as isolating DC3. + """ + mdc = MdcCluster(self, ignite_version, dcs=DCS_3, srv_per_dc=2, runners_per_dc=1, jmx_metrics=True, + network_timeout=20_000, tcp_connect_timeout=10_000) + + majority = tuple(dc for dc in mdc.dcs if dc != isolated_dc) + + cut = isolation_pairs(isolated_dc, mdc.dcs) + + with cross_dc_network(self.logger, mdc, delay_ms=cross_dc_latency_ms) as net: + mdc.start_servers() + + total_keys = self._generate_data(mdc) + + mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1) + + # The cluster's own verdict on the guarantees, next to the test's: the affinity + # metric reports the cache CONFIGURATION, the distribution one the assignment + # verify_cache_distribution() has just walked partition by partition. + mdc.verify_cache_mdc_metrics(CACHE_NAME, affinity_safe=True, distribution_safe=True) + + # Both cuts land in the same SSH round-trip: rolling them out one after the + # other would briefly present the cluster with a two segment topology it would + # legitimately react to, which is not the scenario under test. + net.enable_network_partitions(*cut) + + sleep(SPLIT_SETTLE_SECS) + + mdc.verify_segments(majority, isolated_dc) + + # One copy of every partition lives in every DC, so both segments read everything. + for dc in mdc.dcs: + mdc.check_data(dc, CACHE_NAME, 0, total_keys) + + # Two DCs out of three are a majority - the segment keeps writing... + for idx, dc in enumerate(majority): + mdc.check_put_admissibility(dc, CACHE_NAME, True, key_offset=_probe_offset(idx)) + + # ...while the isolated DC sees one DC out of three and is rejected by the validator. + mdc.check_put_admissibility(isolated_dc, CACHE_NAME, False, key_offset=_probe_offset(2)) + + # The cache is configured for three DCs no matter what the segment can see, so the + # affinity metric holds. The distribution one is about what the segment CAN place + # a copy on and is only reported here - see the note on the constants. + mdc.verify_cache_mdc_metrics(CACHE_NAME, affinity_safe=True, dc=isolated_dc) + + net.disable_network_partitions(*cut) + + # The isolated DC may be the one the others discovered through, in which case the + # shared ip finder holds only its own addresses and a restart would seed it off + # itself. Point discovery at every DC so it rejoins the majority ring either way. + mdc.sync_service_discovery() + + # Split-brain does not self-heal: the read-only segment rejoins via restart. + mdc.restart(isolated_dc) + + self._verify_recovered(mdc, total_keys, probe_idx=3) + + # Whole again, so the assignment is back to one copy per DC and the cluster says so. + mdc.verify_cache_mdc_metrics(CACHE_NAME, affinity_safe=True, distribution_safe=True) + + mdc.stop_servers() + + @cluster(num_nodes=9) + @ignite_versions(str(DEV_BRANCH)) + @parametrize(cross_dc_latency_ms=100) + def test_three_way_split_blocks_all_writes(self, ignite_version, cross_dc_latency_ms): + """ + Every cross-DC link drops at once, leaving three single-DC segments. No segment + holds a majority, so - unlike the two DC case, where one half always survives as + writable - the whole cluster goes read-only until the links come back. + """ + mdc = MdcCluster(self, ignite_version, dcs=DCS_3, srv_per_dc=2, runners_per_dc=1, + network_timeout=20_000, tcp_connect_timeout=10_000) + + mesh = all_pairs(mdc.dcs) + + with cross_dc_network(self.logger, mdc, delay_ms=cross_dc_latency_ms) as net: + mdc.start_servers() + + total_keys = self._generate_data(mdc) + + mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1) + + net.enable_network_partitions(*mesh) + + sleep(SPLIT_SETTLE_SECS) + + # Every DC ended up on its own, each a healthy single-DC segment. + mdc.verify_split_brain() + + for idx, dc in enumerate(mdc.dcs): + mdc.check_data(dc, CACHE_NAME, 0, total_keys) + + mdc.check_put_admissibility(dc, CACHE_NAME, False, key_offset=_probe_offset(idx)) + + net.disable_network_partitions(*mesh) + + # The first DC keeps the ring, every other segment rejoins it via restart. + for dc in mdc.dcs[1:]: + mdc.restart(dc) + + self._verify_recovered(mdc, total_keys, probe_idx=3) + + mdc.stop_servers() + + @cluster(num_nodes=9) + @ignite_versions(str(DEV_BRANCH)) + def test_writes_survive_single_dc_loss(self, ignite_version): + """ + Data centers go down one by one, no network impairments involved. Losing the first + one leaves a majority and writes continue; losing the second one drops the survivor + into a minority and it goes read-only; both returning restores writes everywhere. + """ + mdc = MdcCluster(self, ignite_version, dcs=DCS_3, srv_per_dc=2, runners_per_dc=1) + + with cross_dc_network(self.logger, mdc): + mdc.start_servers() + + total_keys = self._generate_data(mdc) + + mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1) + + mdc.stop_dcs(DC_3) + + # Two DCs out of three remain visible - still a majority, still writable. + mdc.verify_segment_healthy((DC_1, DC_2)) + + mdc.check_data(DC_1, CACHE_NAME, 0, total_keys) + + mdc.check_put_admissibility(DC_1, CACHE_NAME, True, key_offset=_probe_offset(0)) + mdc.check_put_admissibility(DC_2, CACHE_NAME, True, key_offset=_probe_offset(1)) + + mdc.stop_dcs(DC_2) + + # One DC out of three is a minority: everything is still readable... + mdc.check_data(DC_1, CACHE_NAME, 0, total_keys) + + # ...but the validator rejects every write. + mdc.check_put_admissibility(DC_1, CACHE_NAME, False, key_offset=_probe_offset(2)) + + # DC1 never left, so the shared ip finder still points at a live ring. Both DCs + # are started before the first wait, so their joins are not serialized behind + # each other's rebalance. + mdc.start_dcs(DC_2, DC_3) + + self._verify_recovered(mdc, total_keys, probe_idx=3) + + mdc.stop_servers() + + @staticmethod + def _generate_data(mdc: MdcCluster) -> int: + """ + Populates the cache from every DC in turn, each with its own key range. + + :return: The size of the whole data set, i.e. the exclusive upper key bound. + """ + for idx, dc in enumerate(mdc.dcs): + mdc.generate_data(dc, CACHE_NAME, idx * KEYS_PER_DC, (idx + 1) * KEYS_PER_DC) + + return len(mdc.dcs) * KEYS_PER_DC + + @staticmethod + def _verify_recovered(mdc: MdcCluster, total_keys: int, probe_idx: int): + """ + Verifies that the cluster is whole again: one ACTIVE cluster spanning every DC, + writes accepted everywhere, the whole data set intact and evenly spread, and no + segment left a suspicious trace in its logs. + """ + mdc.verify_whole_cluster_healthy() + + for idx, dc in enumerate(mdc.dcs): + mdc.check_put_admissibility(dc, CACHE_NAME, True, key_offset=_probe_offset(probe_idx + idx)) + + mdc.check_data(mdc.dcs[0], CACHE_NAME, 0, total_keys) + + mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1) + + mdc.control().idle_verify(CACHE_NAME) + + mdc.verify_servers_log_clean() From 95e50affb88d5edddb8ecf9d75f9ad53d526e9e4 Mon Sep 17 00:00:00 2001 From: Maksim Davydov Date: Fri, 11 Sep 2026 13:05:35 +0300 Subject: [PATCH 5/8] IGNITE-28952 [ducktests] Cache metrics lookup fixed --- modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py b/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py index 4eb57406df08a..c4b126832d79b 100644 --- a/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py +++ b/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py @@ -37,6 +37,7 @@ from ignitetest.services.network_group.manager import NetworkGroupManager from ignitetest.services.utils.control_utility import ControlUtility from ignitetest.services.utils.ignite_configuration import IgniteConfiguration, TcpCommunicationSpi +from ignitetest.services.utils.jmx_utils import JmxClient, metric_registry_pattern from ignitetest.services.utils.ignite_configuration.discovery import TcpDiscoverySpi, from_ignite_cluster, \ from_ignite_services from ignitetest.services.utils.ssl.client_connector_configuration import ClientConnectorConfiguration @@ -612,7 +613,7 @@ def cache_mdc_metrics(self, cache_name: str, dc: Optional[str] = None) -> Dict[s node = next(node for svc in self.dc_servers(dc if dc is not None else self.dcs[0]) for node in svc.alive_nodes) - mbean = node.cache_mbean(cache_name) + mbean = JmxClient(node).find_mbean(metric_registry_pattern('cache', cache_name)) return {name: mbean.bool_value(name) for name in (MDC_SAFE_AFFINITY_METRIC, MDC_SAFE_DISTRIBUTION_METRIC)} From 7126596d11d55bbcc376b266b878ab18cf46a761 Mon Sep 17 00:00:00 2001 From: Maksim Davydov Date: Fri, 11 Sep 2026 14:44:36 +0300 Subject: [PATCH 6/8] IGNITE-28952 [ducktests] lower cross_dc_latency --- .../tests/ignitetest/tests/mdc/majority_partition_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py b/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py index 5b97a441bbc2b..7c3beb8edbde9 100644 --- a/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py +++ b/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py @@ -63,7 +63,7 @@ class MdcMajorityPartitionTest(IgniteTest): """ @cluster(num_nodes=9) @ignite_versions(str(DEV_BRANCH)) - @parametrize(cross_dc_latency_ms=100, isolated_dc=DC_3) + @parametrize(cross_dc_latency_ms=60, isolated_dc=DC_3) @parametrize(cross_dc_latency_ms=100, isolated_dc=DC_1) def test_minority_dc_isolation(self, ignite_version, cross_dc_latency_ms, isolated_dc): """ From a0c52cd1b57a16407bf77f58aadd25aac234497f Mon Sep 17 00:00:00 2001 From: Maksim Davydov Date: Thu, 17 Sep 2026 13:49:19 +0300 Subject: [PATCH 7/8] IGNITE-28952 [ducktests] Review fixes: JMX accessor, negative MDC metric case * cache_mdc_metrics() reads through node.metric_registry_mbean(): the node's JMX client is restart-safe since IGNITE-29049, so the manual JmxClient workaround is gone. * The MDC safety metrics are also checked to say False: a cache created with backupFilter=False (new MdcCacheAwareApplication parameter) is neither affinity- nor distribution-safe. It is destroyed before the network cut, since the isolated DC would lose its partitions. Adds ControlUtility.cache_destroy(). * Docstrings of backupFilter(), _server_service() and dc_servers() reworded. --- .../ducktest/tests/mdc/MdcCacheAwareApplication.java | 10 ++++++---- .../tests/ignitetest/services/mdc/mdc_cluster.py | 8 +++----- .../ignitetest/services/utils/control_utility.py | 10 ++++++++++ .../ignitetest/tests/mdc/majority_partition_test.py | 12 ++++++++++++ 4 files changed, 31 insertions(+), 9 deletions(-) diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java index 17a823a650c4a..86e168562146e 100644 --- a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java @@ -65,7 +65,8 @@ *
  • {@code writeSync} - {@link CacheWriteSynchronizationMode}, default {@code FULL_SYNC}. * Note: MDC-aware local reads require a mode other than {@code PRIMARY_SYNC};
    • *
  • {@code readFromBackup} - default {@code true}, required for DC-local reads;
    • - *
  • {@code partitions} - affinity partitions number, default 512.
    • + *
  • {@code partitions} - affinity partitions number, default 512;
    • + *
  • {@code backupFilter} - whether to set the MDC affinity backup filter, default {@code true}.
    • *
*/ public abstract class MdcCacheAwareApplication extends IgniteAwareApplication { @@ -163,9 +164,7 @@ protected CacheConfiguration mdcCacheConfiguration(JsonNode jNod /** * The affinity backup filter the cache is configured with. A {@link RendezvousAffinityFunction} - * holds exactly one, so an override replaces the MDC filter rather than complementing it - which - * is the point: a fork that spreads the copies by something finer than the data center (a cell, - * an availability zone) says so here instead of repeating the rest of the cache configuration. + * holds exactly one, so an override replaces the MDC filter rather than complementing it. * * @param jNode Parameters. * @param dcsNum Number of data centers. @@ -174,6 +173,9 @@ protected CacheConfiguration mdcCacheConfiguration(JsonNode jNod */ protected IgniteBiPredicate> backupFilter(JsonNode jNode, int dcsNum, int backups) { + if (!jNode.path("backupFilter").asBoolean(true)) + return null; + return new MdcAffinityBackupFilter(dcsNum, backups); } diff --git a/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py b/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py index c4b126832d79b..a262f956a2012 100644 --- a/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py +++ b/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py @@ -37,7 +37,6 @@ from ignitetest.services.network_group.manager import NetworkGroupManager from ignitetest.services.utils.control_utility import ControlUtility from ignitetest.services.utils.ignite_configuration import IgniteConfiguration, TcpCommunicationSpi -from ignitetest.services.utils.jmx_utils import JmxClient, metric_registry_pattern from ignitetest.services.utils.ignite_configuration.discovery import TcpDiscoverySpi, from_ignite_cluster, \ from_ignite_services from ignitetest.services.utils.ssl.client_connector_configuration import ClientConnectorConfiguration @@ -279,15 +278,14 @@ def __init__(self, test, ignite_version: str, dcs: Sequence[str] = DCS_2, def _server_service(self, dc: str, num_nodes: int) -> IgniteService: """ Builds the server service of one DC. The single place a server command line is put - together, so a fork that starts its servers differently overrides just this. + together. """ return IgniteService(self.test_context, self.ignite_config, num_nodes=num_nodes, jvm_opts=dc_jvm_opts(dc), startup_timeout_sec=IGNITE_STARTUP_TIMEOUT_SEC) def dc_servers(self, dc: str) -> List[IgniteService]: """ - :return: All server services of the given DC - one, unless a subclass splits a DC's - servers into several services (node groups, cells, availability zones). + :return: All server services of the given DC, one per DC in this fixture. """ return [self.servers[dc]] if dc in self.servers else [] @@ -613,7 +611,7 @@ def cache_mdc_metrics(self, cache_name: str, dc: Optional[str] = None) -> Dict[s node = next(node for svc in self.dc_servers(dc if dc is not None else self.dcs[0]) for node in svc.alive_nodes) - mbean = JmxClient(node).find_mbean(metric_registry_pattern('cache', cache_name)) + mbean = node.metric_registry_mbean(f'cache.{cache_name}') return {name: mbean.bool_value(name) for name in (MDC_SAFE_AFFINITY_METRIC, MDC_SAFE_DISTRIBUTION_METRIC)} diff --git a/modules/ducktests/tests/ignitetest/services/utils/control_utility.py b/modules/ducktests/tests/ignitetest/services/utils/control_utility.py index 329e5321974e5..d1de07ed98327 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/control_utility.py +++ b/modules/ducktests/tests/ignitetest/services/utils/control_utility.py @@ -180,6 +180,16 @@ def idle_verify_dump(self, node=None): return re.search(r'/.*.txt', data).group(0) + def cache_destroy(self, cache_names): + """ + Destroys caches. + :param cache_names: Cache name, or list of cache names. + """ + if isinstance(cache_names, str): + cache_names = [cache_names] + + return self.__run(f"--cache destroy --caches {','.join(cache_names)} --yes") + def cache_distribution(self, node_id=None, cache_names=None, user_attributes=None): """ Prints partition distribution. diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py b/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py index 7c3beb8edbde9..fe72b4027f595 100644 --- a/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py +++ b/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py @@ -41,6 +41,9 @@ CACHE_NAME = "mdc-majority" +# A cache left without the MDC affinity backup filter, to see the safety metrics say no. +UNSAFE_CACHE_NAME = "mdc-majority-unsafe" + # Keys generated from each DC; the whole data set is [0, KEYS_PER_DC * len(DCS_3)). KEYS_PER_DC = 100 @@ -93,6 +96,15 @@ def test_minority_dc_isolation(self, ignite_version, cross_dc_latency_ms, isolat # verify_cache_distribution() has just walked partition by partition. mdc.verify_cache_mdc_metrics(CACHE_NAME, affinity_safe=True, distribution_safe=True) + # ...and the verdict is not a constant. Without the MDC backup filter the cache is + # unsafe by configuration, and plain rendezvous leaves some partition without a copy + # in some DC. Destroyed before the cut, as the isolated DC would lose its partitions. + mdc.generate_data(mdc.dcs[0], UNSAFE_CACHE_NAME, 0, KEYS_PER_DC, backupFilter=False) + + mdc.verify_cache_mdc_metrics(UNSAFE_CACHE_NAME, affinity_safe=False, distribution_safe=False) + + mdc.control().cache_destroy(UNSAFE_CACHE_NAME) + # Both cuts land in the same SSH round-trip: rolling them out one after the # other would briefly present the cluster with a two segment topology it would # legitimately react to, which is not the scenario under test. From 19049346e28ed3d5c521301f2665780d15658de1 Mon Sep 17 00:00:00 2001 From: Maksim Davydov Date: Thu, 17 Sep 2026 17:48:13 +0300 Subject: [PATCH 8/8] IGNITE-28952 [ducktests] Drop the negative MDC metric case The unsafe cache check is not a functional scenario: the metric verdict for a cache without the MDC backup filter is already covered by MdcCacheMetricsTest. Creating and destroying an extra cache right before the network cut only prolongs the test and adds discovery load at the cut. Removes the backupFilter application parameter and ControlUtility.cache_destroy() along with it. --- .../ducktest/tests/mdc/MdcCacheAwareApplication.java | 6 +----- .../ignitetest/services/utils/control_utility.py | 10 ---------- .../ignitetest/tests/mdc/majority_partition_test.py | 12 ------------ 3 files changed, 1 insertion(+), 27 deletions(-) diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java index 86e168562146e..3e14d187c80de 100644 --- a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java @@ -65,8 +65,7 @@ *
  • {@code writeSync} - {@link CacheWriteSynchronizationMode}, default {@code FULL_SYNC}. * Note: MDC-aware local reads require a mode other than {@code PRIMARY_SYNC};
    • *
  • {@code readFromBackup} - default {@code true}, required for DC-local reads;
    • - *
  • {@code partitions} - affinity partitions number, default 512;
    • - *
  • {@code backupFilter} - whether to set the MDC affinity backup filter, default {@code true}.
    • + *
  • {@code partitions} - affinity partitions number, default 512.
    • * */ public abstract class MdcCacheAwareApplication extends IgniteAwareApplication { @@ -173,9 +172,6 @@ protected CacheConfiguration mdcCacheConfiguration(JsonNode jNod */ protected IgniteBiPredicate> backupFilter(JsonNode jNode, int dcsNum, int backups) { - if (!jNode.path("backupFilter").asBoolean(true)) - return null; - return new MdcAffinityBackupFilter(dcsNum, backups); } diff --git a/modules/ducktests/tests/ignitetest/services/utils/control_utility.py b/modules/ducktests/tests/ignitetest/services/utils/control_utility.py index d1de07ed98327..329e5321974e5 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/control_utility.py +++ b/modules/ducktests/tests/ignitetest/services/utils/control_utility.py @@ -180,16 +180,6 @@ def idle_verify_dump(self, node=None): return re.search(r'/.*.txt', data).group(0) - def cache_destroy(self, cache_names): - """ - Destroys caches. - :param cache_names: Cache name, or list of cache names. - """ - if isinstance(cache_names, str): - cache_names = [cache_names] - - return self.__run(f"--cache destroy --caches {','.join(cache_names)} --yes") - def cache_distribution(self, node_id=None, cache_names=None, user_attributes=None): """ Prints partition distribution. diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py b/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py index fe72b4027f595..7c3beb8edbde9 100644 --- a/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py +++ b/modules/ducktests/tests/ignitetest/tests/mdc/majority_partition_test.py @@ -41,9 +41,6 @@ CACHE_NAME = "mdc-majority" -# A cache left without the MDC affinity backup filter, to see the safety metrics say no. -UNSAFE_CACHE_NAME = "mdc-majority-unsafe" - # Keys generated from each DC; the whole data set is [0, KEYS_PER_DC * len(DCS_3)). KEYS_PER_DC = 100 @@ -96,15 +93,6 @@ def test_minority_dc_isolation(self, ignite_version, cross_dc_latency_ms, isolat # verify_cache_distribution() has just walked partition by partition. mdc.verify_cache_mdc_metrics(CACHE_NAME, affinity_safe=True, distribution_safe=True) - # ...and the verdict is not a constant. Without the MDC backup filter the cache is - # unsafe by configuration, and plain rendezvous leaves some partition without a copy - # in some DC. Destroyed before the cut, as the isolated DC would lose its partitions. - mdc.generate_data(mdc.dcs[0], UNSAFE_CACHE_NAME, 0, KEYS_PER_DC, backupFilter=False) - - mdc.verify_cache_mdc_metrics(UNSAFE_CACHE_NAME, affinity_safe=False, distribution_safe=False) - - mdc.control().cache_destroy(UNSAFE_CACHE_NAME) - # Both cuts land in the same SSH round-trip: rolling them out one after the # other would briefly present the cluster with a two segment topology it would # legitimately react to, which is not the scenario under test.