anton-vinogradov commented on code in PR #12740: URL: https://github.com/apache/ignite/pull/12740#discussion_r3621349387
########## modules/ducktests/tests/docker/Dockerfile: ########## @@ -125,31 +126,31 @@ RUN echo 'PermitUserEnvironment yes' >> /etc/ssh/sshd_config ARG APACHE_MIRROR="https://apache-mirror.rbc.ru/pub/apache/"; ARG APACHE_ARCHIVE="https://archive.apache.org/dist/"; -# Install Ignite -RUN for v in "2.7.6" "2.17.0"; \ - do cd /opt; \ - curl -O $APACHE_ARCHIVE/ignite/$v/apache-ignite-$v-bin.zip;\ - unzip apache-ignite-$v-bin.zip && mv /opt/apache-ignite-$v-bin /opt/ignite-$v; \ - done \ - && rm /opt/apache-ignite-*-bin.zip - -# Install Zookeeper -ARG ZOOKEEPER_VERSION="3.5.8" -ARG ZOOKEEPER_NAME="zookeeper-$ZOOKEEPER_VERSION" -ARG ZOOKEEPER_RELEASE_NAME="apache-$ZOOKEEPER_NAME-bin" -ARG ZOOKEEPER_RELEASE_ARTIFACT="$ZOOKEEPER_RELEASE_NAME.tar.gz" -RUN cd /opt && curl -O $APACHE_ARCHIVE/zookeeper/$ZOOKEEPER_NAME/$ZOOKEEPER_RELEASE_ARTIFACT \ - && tar xvf $ZOOKEEPER_RELEASE_ARTIFACT && rm $ZOOKEEPER_RELEASE_ARTIFACT \ - && mv /opt/$ZOOKEEPER_RELEASE_NAME /opt/$ZOOKEEPER_NAME - -# Install Kafka -ARG KAFKA_VERSION="3.9.1" -ARG KAFKA_NAME="kafka" -ARG KAFKA_RELEASE_NAME="${KAFKA_NAME}_2.13-$KAFKA_VERSION" -ARG KAFKA_RELEASE_ARTIFACT="$KAFKA_RELEASE_NAME.tgz" -RUN cd /opt && curl -O $APACHE_ARCHIVE/kafka/$KAFKA_VERSION/$KAFKA_RELEASE_ARTIFACT \ - && tar xvf $KAFKA_RELEASE_ARTIFACT && rm $KAFKA_RELEASE_ARTIFACT \ - && mv /opt/$KAFKA_RELEASE_NAME /opt/$KAFKA_NAME-$KAFKA_VERSION +## Install Ignite Review Comment: **[blocker]** This block (together with the Zookeeper and Kafka ones below) got commented out by the "codestyle" commit (69b6538) - looks like a local image-rebuild speed-up that slipped into the branch. With this image every existing ducktest that runs released versions (`@ignite_versions(str(LATEST))` - baseline, discovery, compatibility suites), the ZooKeeper discovery suites and the CDC-Kafka tests lose their binaries. Needs to be restored before merge. ########## modules/ducktests/tests/ignitetest/services/utils/control_utility.py: ########## @@ -481,6 +569,7 @@ class ClusterState(NamedTuple): state: str topology_version: int baseline: list + coordinator: BaselineNode = None Review Comment: **[blocker]** Adding a 4th field to `ClusterState` breaks positional unpacking in existing tests: `control_utility/baseline_test.py:66/128/134` and `index_rebuild_test.py:83` do `_, version, _ = control_utility.cluster_state()`. NamedTuple unpacking always yields **all** fields regardless of defaults, so these raise `ValueError: too many values to unpack (expected 3)` on the first call. Please update those four call sites within this PR (named access is future-proof: `control_utility.cluster_state().topology_version`). ########## modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py: ########## @@ -0,0 +1,146 @@ +# 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. + +""" +MDC transactional load through a cross-DC network partition. + +A TRANSACTIONAL cache spans both data centers: with backups=1 and the +MdcAffinityBackupFilter every partition owns exactly one copy per DC, so every +implicit-transaction write (a plain put on a transactional cache) must reach a +node in the other DC. A continuous single-threaded insert load runs from the main +DC; the instant the DCs are partitioned the next commit cannot reach all partition +copies and fails with a cache exception. The load cuts itself off on that first +exception and records how many inserts had succeeded. + +After the split settles the test asserts that the cluster really split-brained, +that the load stopped because of the partition (not because it ran out of work), +and - the point of the scenario - that the aborted implicit transactions left +nothing hanging on either half-ring and no suspicious entries in the server logs. + +Data accessibility during the split is deliberately NOT checked: a transactional +cache needs all partition copies available, which a split-brained half-ring cannot +offer. +""" +from time import sleep + +from ducktape.mark import parametrize + +from ignitetest.services.mdc.mdc_cluster import MdcCluster, cross_dc_network, DC_1, DC_2 +from ignitetest.utils import cluster, ignite_versions +from ignitetest.utils.ignite_test import IgniteTest +from ignitetest.utils.version import DEV_BRANCH + +CACHE_NAME = "mdc-tx-load" + +BACKUPS = 1 + +# Fresh, disjoint key range for the continuous insert load: the load advances the key +# on every success, so [LOAD_KEY_FROM, LOAD_KEY_FROM + successfulInserts) gets inserted. +LOAD_KEY_FROM_DC_1 = 1_000_000 +LOAD_KEY_TO_DC_1 = 10_000_000 + +LOAD_KEY_FROM_DC_2 = 10_000_000 +LOAD_KEY_TO_DC_2 = 20_000_000 + +# Let the load accumulate successful inserts before the DCs are cut apart. +LOAD_WARMUP_SECS = 15 + +# Time for discovery to detect the partition and for both half-rings to complete PME +# and account for the split. +SPLIT_SETTLE_SECS = 15 + + +class MdcTransactionalPartitionTest(IgniteTest): + """ + Transactional load resilience to a cross-DC network partition. + """ + @cluster(num_nodes=8) + @ignite_versions(str(DEV_BRANCH)) + @parametrize(cross_dc_latency_ms=100) + def test_transactional_load_cut_on_partition(self, ignite_version, cross_dc_latency_ms): + """ + Continuous implicit-transaction insert load from the main DC is cut off by the first + cache exception the cross-DC partition triggers; afterwards no transaction is left + hanging on either half-ring and the server logs are clean. + """ + mdc = MdcCluster(self, ignite_version, srv_per_dc=2, runners_per_dc=1, loaders_per_dc=1, + network_timeout=20_000, tcp_connect_timeout=10_000) + + with cross_dc_network(self.logger, mdc, delay_ms=cross_dc_latency_ms) as net: + mdc.start_servers() + + # Continuous single-threaded implicit-transaction insert load from the main DC. + # It stops on the very first exception (stopOnError) instead of failing the app, + # recording how many inserts had succeeded up to that point. + for dc, offset_from, to in [(DC_1, LOAD_KEY_FROM_DC_1, LOAD_KEY_TO_DC_1), + (DC_2, LOAD_KEY_FROM_DC_2, LOAD_KEY_TO_DC_2)]: + mdc.start_loader(dc, { + "mode": "TX_PUT", + "cacheName": CACHE_NAME, + "keyFrom": offset_from, + "keyTo": to, + "stopOnError": True, + "resultPrefix": f"txLoad{dc}", + "createCache": True, + "backups": BACKUPS, + "atomicity": "TRANSACTIONAL", + "mainDc": DC_1, + }) + + sleep(LOAD_WARMUP_SECS) + + net.enable_network_partition(DC_1, DC_2) + + # The load hits its first exception here and cuts itself off; give discovery and + # PME time to detect the split and settle into two independent half-rings. + sleep(SPLIT_SETTLE_SECS) + + mdc.verify_split_brain() + + # The load has already finished on its own - this just collects its results. + for dc in [DC_1, DC_2]: + svc = mdc.stop_loader(dc) + + inserts = mdc.result_int(svc, f"txLoad{dc}OpsCnt") + errors = mdc.result_int(svc, f"txLoad{dc}ErrCnt") + + self.logger.info(f"Transactional load cut by the partition " + f"[txLoad{dc}OpsCnt={inserts}, txLoad{dc}ErrCnt={errors}]") + + assert inserts > 0, "The transactional load performed no successful inserts" Review Comment: **[major]** The scenario's central claim - the load was cut off by the partition - is never asserted: `errors` is only logged, and `inserts > 0` also holds when nothing was cut at all (with `iterations=0` the loader runs until `stop_loader()`, so e.g. a broken/absent topology validator keeps this test green). The application already records the flag: ```suggestion assert inserts > 0, "The transactional load performed no successful inserts" assert svc.extract_result(f"txLoad{dc}StoppedOnError") == "true", \ f"The load was expected to be cut off by the partition [dc={dc}]" ``` ########## modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcContinuousLoadApplication.java: ########## @@ -0,0 +1,295 @@ +/* + * 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. + */ + +package org.apache.ignite.internal.ducktest.tests.mdc; + +import java.util.List; +import java.util.Locale; +import java.util.Objects; +import javax.cache.CacheException; +import com.fasterxml.jackson.databind.JsonNode; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteException; +import org.apache.ignite.cache.query.SqlFieldsQuery; +import org.apache.ignite.internal.ducktest.tests.dto.IndexedDataRecord; +import org.apache.ignite.internal.ducktest.utils.OpStats; +import org.apache.ignite.transactions.Transaction; +import org.apache.ignite.transactions.TransactionConcurrency; +import org.apache.ignite.transactions.TransactionIsolation; + +import static org.apache.ignite.internal.ducktest.utils.Utils.fmtMs; +import static org.apache.ignite.internal.ducktest.utils.Utils.getEnum; +import static org.apache.ignite.internal.ducktest.utils.Utils.timed; +import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC; +import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ; + +/** + * Universal MDC load application. Runs a single-threaded synchronous load of the + * requested {@link LoadMode} either for a fixed number of iterations (a "burst") or until + * externally terminated (a "background" load spanning several test phases, e.g. a + * network partition and its healing). + * <p> + * Parameters: + * <ul> + * <li>{@code mode} - one of {@code GET, PUT, TX_PUT, SQL_SELECT, SQL_PUT};</li> + * <li>{@code cacheName} - cache to operate on;</li> + * <li>{@code createCache} - if {@code true}, (re)creates the cache from the MDC + * configuration parameters (see {@link MdcCacheAwareApplication}); otherwise the + * cache must already exist;</li> + * <li>{@code keyFrom} / {@code keyTo} - key range. Read modes cycle within the range; + * write modes advance sequentially from {@code keyFrom} on every success, so on + * completion the keys {@code [keyFrom, keyFrom + opsCnt)} are guaranteed written;</li> + * <li>{@code iterations} - number of operations; {@code 0} means "run until terminated";</li> + * <li>{@code inadmissible} - write modes only. {@code false} (default): writes must + * succeed; {@code true}: every write must be rejected by the topology validator + * (read-only DC), any success fails the application;</li> + * <li>{@code stopOnError} - if {@code true}, the load stops gracefully on the very first + * operation failure (rather than failing the application), records the number of + * successful operations and a {@code StoppedOnError} flag, and finishes. Takes precedence + * over {@code continueOnError}. Intended for a load that must be cut off by the first + * exception a network partition triggers;</li> + * <li>{@code continueOnError} - if {@code true}, operation failures are counted instead of + * failing fast. Intended for background loads crossing a partition boundary, where a + * short transient error window is possible;</li> + * <li>{@code opPauseMs} - pause between operations, default 0;</li> + * <li>{@code resultPrefix} - prefix for recorded results, so that several runs reusing one + * service produce uniquely named results;</li> + * <li>{@code txConcurrency} / {@code txIsolation} - transaction parameters for {@code TX_PUT}.</li> + * </ul> + */ +public class MdcContinuousLoadApplication extends MdcCacheAwareApplication { + /** */ + public static final TransactionConcurrency DFLT_TX_CONCURRENCY = PESSIMISTIC; + + /** */ + public static final TransactionIsolation DFLT_TX_ISOLATION = REPEATABLE_READ; + + /** Cache for the cache API modes ({@code null} in SQL modes). */ + private IgniteCache<Integer, IndexedDataRecord> cache; + + /** Cache for the SQL modes ({@code null} in cache API modes). */ + private IgniteCache<Integer, Integer> sqlCache; + + /** Compiled DML statement for {@link LoadMode#SQL_PUT}. */ + private String mergeSql; + + /** Compiled query for {@link LoadMode#SQL_SELECT}. */ + private String selectSql; + + /** Latency of successful operations. */ + private final OpStats stats = new OpStats(); + + /** */ + private TransactionConcurrency txConcurrency; + + /** */ + private TransactionIsolation txIsolation; + + /** {@inheritDoc} */ + @Override public void run(JsonNode jNode) throws Exception { + LoadMode mode = getEnum(jNode, "mode", LoadMode.class); + + String cacheName = jNode.path("cacheName").asText(DFLT_CACHE_NAME); + boolean createCache = jNode.path("createCache").asBoolean(false); + + int keyFrom = jNode.path("keyFrom").asInt(0); + int keyTo = jNode.path("keyTo").asInt(Integer.MAX_VALUE); + long iterations = jNode.path("iterations").asLong(0); + + boolean inadmissible = jNode.path("inadmissible").asBoolean(false); + boolean continueOnError = jNode.path("continueOnError").asBoolean(false); + boolean stopOnError = jNode.path("stopOnError").asBoolean(false); + + long opPauseMs = jNode.path("opPauseMs").asLong(0); + + String pfx = jNode.path("resultPrefix").asText(""); + + txConcurrency = getEnum(jNode, "txConcurrency", DFLT_TX_CONCURRENCY); + txIsolation = getEnum(jNode, "txIsolation", DFLT_TX_ISOLATION); + + markInitialized(); + waitForActivation(); + + if (mode.isSql()) + sqlCache = createCache ? mdcSqlCache(jNode) : ignite.cache(cacheName); + else + cache = createCache ? mdcCache(jNode) : ignite.cache(cacheName); + + mergeSql = String.format("MERGE INTO \"%s\".%s(_KEY, _VAL) VALUES(?, ?)", cacheName, SQL_TABLE); + selectSql = String.format("SELECT _VAL FROM \"%s\".%s WHERE _KEY = ?", cacheName, SQL_TABLE); + + log.info("MDC load started [dc=" + dcId() + ", mode=" + mode + ", cache=" + cacheName + + ", keyFrom=" + keyFrom + ", keyTo=" + keyTo + ", iterations=" + iterations + + ", inadmissible=" + inadmissible + ", continueOnError=" + continueOnError + "]"); + + long opsCnt = 0; + long errCnt = 0; + + boolean stoppedOnError = false; + + long maxStallMs = 0; + + long startTs = System.currentTimeMillis(); + long lastOkTs = startTs; + + int key = keyFrom; + + while (!terminated() && (iterations == 0 || opsCnt + errCnt < iterations)) { + boolean ok = true; + + try { + doOperation(mode, key); + } + catch (CacheException | IgniteException e) { + ok = false; + + if (inadmissible) Review Comment: **[minor]** The rename dropped the `mode.isWrite()` guard the old `mode.isWrite() && !expectAdmissible` had (and the key-advance condition below still keeps): a read-mode run with a stray `inadmissible=true` now swallows every failure - including the missed/corrupted-read `IgniteException` - logging it as "Write rejected as expected", and finishes green, because the final admissibility check is write-only. Restoring the guard keeps the javadoc's "write modes only" honest: ```suggestion if (mode.isWrite() && inadmissible) ``` ########## modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java: ########## @@ -0,0 +1,170 @@ +/* + * 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. + */ + +package org.apache.ignite.internal.ducktest.tests.mdc; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; +import com.fasterxml.jackson.databind.JsonNode; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.cache.CacheAtomicityMode; +import org.apache.ignite.cache.CacheMode; +import org.apache.ignite.cache.CacheWriteSynchronizationMode; +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.configuration.CacheConfiguration; +import org.apache.ignite.internal.ducktest.tests.dto.IndexedDataRecord; +import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication; +import org.apache.ignite.topology.MdcTopologyValidator; + +import static org.apache.ignite.IgniteSystemProperties.IGNITE_DATA_CENTER_ID; +import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC; +import static org.apache.ignite.cache.CacheMode.PARTITIONED; +import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC; +import static org.apache.ignite.internal.ducktest.utils.Utils.getEnum; + +/** + * Base class for MDC test applications. + * Encapsulates the cache configuration (with {@link MdcTopologyValidator} and + * {@link MdcAffinityBackupFilter}) so that the generator, the checkers and the load + * applications always operate on an identically configured cache. + * <p> + * Supported cache parameters (all optional unless stated otherwise): + * <ul> + * <li>{@code cacheName} - cache name;</li> + * <li>{@code backups} - number of backups; {@code (backups + 1)} must be divisible by {@code dcsNum};</li> + * <li>{@code mainDc} - main data center for the topology validator (2 DC mode);</li> + * <li>{@code datacenters} - full DC set for majority-based validation (odd DC count mode), + * takes precedence over {@code mainDc};</li> + * <li>{@code dcsNum} - number of data centers, default 2;</li> + * <li>{@code cacheMode} - {@link CacheMode}, default {@code PARTITIONED};</li> + * <li>{@code atomicity} - {@link CacheAtomicityMode}, default {@code ATOMIC};</li> + * <li>{@code writeSync} - {@link CacheWriteSynchronizationMode}, default {@code FULL_SYNC}. + * Note: MDC-aware local reads require a mode other than {@code PRIMARY_SYNC};</li> + * <li>{@code readFromBackup} - default {@code true}, required for DC-local reads;</li> + * <li>{@code partitions} - affinity partitions number, default 512.</li> + * </ul> + */ +public abstract class MdcCacheAwareApplication extends IgniteAwareApplication { + /** Table name of the SQL-enabled MDC cache. The SQL schema is the quoted cache name. */ + protected static final String SQL_TABLE = "LOAD"; + + /** */ + protected static final String DFLT_CACHE_NAME = "default"; + + /** */ + protected static final CacheMode DFLT_CACHE_MODE = PARTITIONED; + + /** One backup: with the default 2 DCs, {@code (backups + 1) / dcsNum} = 1 copy per DC. */ + protected static final int DFLT_BACKUPS = 1; + + /** */ + protected static final int DFLT_DCS_NUM = 2; + + /** */ + protected static final int DFLT_PARTITIONS = 512; + + /** */ + protected static final CacheAtomicityMode DFLT_ATOMICITY_MODE = ATOMIC; + + /** */ + protected static final CacheWriteSynchronizationMode DFLT_WRITE_SYNC = FULL_SYNC; + + /** + * @param jNode Parameters. + * @return Cache configured with the MDC topology validator and backup filter. + */ + protected IgniteCache<Integer, IndexedDataRecord> mdcCache(JsonNode jNode) { + return ignite.getOrCreateCache(this.<IndexedDataRecord>mdcCacheConfiguration(jNode)); + } + + /** + * Same MDC cache configuration with a {@link QueryEntity} on top, so the cache is + * queryable via SQL: {@code SELECT _VAL FROM "<cacheName>".LOAD WHERE _KEY = ?}. + * + * @param jNode Parameters. + * @return SQL-enabled cache configured with the MDC topology validator and backup filter. + */ + protected IgniteCache<Integer, Integer> mdcSqlCache(JsonNode jNode) { + CacheConfiguration<Integer, Integer> cacheCfg = mdcCacheConfiguration(jNode); + + cacheCfg.setQueryEntities(Collections.singletonList( + new QueryEntity(Integer.class, Integer.class).setTableName(SQL_TABLE))); + + return ignite.getOrCreateCache(cacheCfg); + } + + /** + * @param jNode Parameters. + * @return MDC cache configuration compiled from the application parameters. + */ + protected <V> CacheConfiguration<Integer, V> mdcCacheConfiguration(JsonNode jNode) { + String cacheName = jNode.path("cacheName").asText(DFLT_CACHE_NAME); + int backups = jNode.path("backups").asInt(DFLT_BACKUPS); + int partitions = jNode.path("partitions").asInt(DFLT_PARTITIONS); + + CacheAtomicityMode atomicity = getEnum(jNode, "atomicity", DFLT_ATOMICITY_MODE); + CacheWriteSynchronizationMode writeSync = getEnum(jNode, "writeSync", DFLT_WRITE_SYNC); + CacheMode cacheMode = getEnum(jNode, "cacheMode", DFLT_CACHE_MODE); + + boolean readFromBackup = jNode.path("readFromBackup").asBoolean(true); + + int dcsNum = jNode.path("dcsNum").asInt(DFLT_DCS_NUM); + + MdcTopologyValidator topValidator = new MdcTopologyValidator(); + + if (jNode.hasNonNull("datacenters")) { + Set<String> dcs = new HashSet<>(); + + jNode.get("datacenters").forEach(dc -> dcs.add(dc.asText())); + + topValidator.setDatacenters(dcs); + } + else + topValidator.setMainDatacenter(jNode.path("mainDc").asText()); Review Comment: **[minor]** `path().asText()` turns a missing `mainDc` into `""`, which passes `MdcTopologyValidator.checkConfiguration()` (it only guards `null`) and then can never match any node's `dataCenterId` in `validate()`: a test that forgets the parameter gets a cache that silently rejects every write instead of a configuration error. Fail fast: ```suggestion else if (jNode.hasNonNull("mainDc")) topValidator.setMainDatacenter(jNode.get("mainDc").asText()); else throw new IllegalArgumentException("Either 'datacenters' or 'mainDc' must be specified."); ``` ########## modules/ducktests/tests/ignitetest/services/network_group/manager.py: ########## @@ -0,0 +1,420 @@ +# 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. +import re +import socket +import struct +import sys +from concurrent.futures import ThreadPoolExecutor +from itertools import permutations +from time import monotonic +from typing import Dict, Iterator, List, Tuple + +from ducktape.services.service import Service + +from ignitetest.services.network_group.configuration import NetworkGroupStore, CrossNetworkGroupConfiguration +from ignitetest.services.network_group.tc_rule_args import ( + ACTION_ADD, ACTION_OVERWRITE, + to_tcset_cmd, to_tcdel_all_cmd, + partition_chain_name, to_partition_enable_cmd, to_partition_disable_cmd, to_partition_teardown_cmd, + PARTITION_CHAIN_PREFIX +) +from ignitetest.services.utils.decorators import memoize + + +# Get the default network interface (e.g., eth0, ens3) +CMD_GET_NETWORK_INTERFACE = "ip route | grep default | awk -- '{printf $5}'" + +# Pseudo-action used only on initial deployment: the first rule per node +# overwrites any stale state, subsequent rules are appended. +ACTION_DEPLOY = "deploy" + +# Upper bound on concurrent SSH sessions used to apply tc rules cluster-wide. +MAX_PARALLEL_SSH_SESSIONS = 16 + +# A rule spec: (src_group, dst_group, action, config). +RuleSpec = Tuple[str, str, str, CrossNetworkGroupConfiguration] + + +class NetworkGroupManager: + """ + Deploys and tears down traffic-control rules between logical node groups, + and toggles full network partitions between them at test time. + + Baseline impairments (delay/loss) are deployed once via tcset. + Partitions are layered on top as per-pair iptables DROP chains. The netem + rules are never touched by a partition, so healing is a pure chain flush + that automatically restores the originally deployed impairments. + + All commands targeting a single node are batched into one SSH invocation, + and nodes are configured in parallel, so that a partition (or its removal) + takes effect near-atomically across the cluster instead of rolling out + node by node. + """ + def __init__(self, logger, network_group_store: NetworkGroupStore, + network_group_registry: Dict[str, List[Service]]): + self.logger = logger + + self.network_group_store = network_group_store + self.network_group_registry = network_group_registry + + def __enter__(self): + self.deploy() + + self._log_network("ON_DEPLOY") + + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.destroy() + + self._log_network("ON_EXIT") + + def deploy(self): + """ + Compiles routing maps and deploys cross-group network constraints. + """ + self._prefetch_network_interfaces() + + self.destroy() + + specs = [] + + for src_group, dst_group in permutations(self.network_group_registry.keys(), 2): + cfg = self.network_group_store.get_config(src_group, dst_group) + + if cfg is None: + # No impairment configured between these groups: traffic flows unconstrained. + self.logger.debug(f"No configuration for {src_group} -> {dst_group}, skipping.") + continue + + specs.append((src_group, dst_group, ACTION_DEPLOY, cfg)) + + self._apply_specs(specs, tag="DEPLOY") + + def destroy(self): + """ + Restores network interfaces back to their un-throttled state and + removes any leftover partition chains from iptables. + """ + tasks = [] + + for node in self._iter_all_nodes(): + interface = self._get_default_network_interface(node) + + tasks.append((node, f"{to_tcdel_all_cmd(interface)} && {to_partition_teardown_cmd()}")) + + self._ssh_parallel(tasks, tag="DESTROY") + + def enable_network_partition(self, group_a: str, group_b: str): + """ + Creates a complete, bidirectional network partition between two groups + by installing iptables DROP rules for all cross-group traffic, + 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}]") + + chain = partition_chain_name(group_a, group_b) + + tasks = [] + + for src_group, dst_group in self._bidirectional(group_a, group_b): + remote_ips = self._resolve_group_ips(dst_group) + + cmd = to_partition_enable_cmd(chain, remote_ips) + + for node in self._iter_group_nodes(src_group): + tasks.append((node, cmd)) + + self._ssh_parallel(tasks, tag="PARTITION_ON") + + self._log_network(f"PARTITION {group_a} <-> {group_b}") Review Comment: **[minor]** `_log_network` runs two sequential SSH round-trips (`tc qdisc show` / `tc filter show`) per node *inside* the partition window - after the DROP rules are installed but before the caller's `sleep(BLIP_SECS)` starts. With 8+ nodes that inflates every "1-second" blip by several seconds, uncomfortably close to the 10s failure detection timeout the blips scenario relies on staying under. The tc state doesn't change when toggling iptables, so the dump adds no information here - consider dropping it from `enable_network_partition`/`disable_network_partition` (or making it opt-in) so the blip lasts what the test asked for. ########## modules/ducktests/tests/ignitetest/services/utils/control_utility.py: ########## @@ -181,6 +181,83 @@ def idle_verify_dump(self, node=None): return re.search(r'/.*.txt', data).group(0) + def cache_distribution(self, node_id=None, cache_names=None, user_attributes=None): + """ + Prints partition distribution. + + :param node_id: Node id to get distribution for, all nodes if None. + :param cache_names: Cache name, or list of cache names, all caches if None. + :param user_attributes: Node attribute name or list of names to add to the output + (e.g. "IGNITE_DATA_CENTER_ID"). + :return: CacheDistribution. + """ + if isinstance(cache_names, str): + cache_names = [cache_names] + + if isinstance(user_attributes, str): + user_attributes = [user_attributes] + + cmd = f"--cache distribution {node_id if node_id else 'null'}" + + if cache_names: + cmd += f" {','.join(cache_names)}" + + if user_attributes: + cmd += f" --user-attributes {','.join(user_attributes)}" + + result = self.__run(cmd) + + return self.__parse_cache_distribution(result, user_attributes) + + @staticmethod + def __parse_cache_distribution(output, user_attributes=None): + group_pattern = re.compile(r"\[next group: id=(?P<group_id>-?\d+), name=(?P<name>[^\]]+)\]") + + # Trailing attribute values appear after nodeAddresses, comma-separated, + # in the same order they were passed to --user-attributes. + row_pattern = re.compile(r"(?P<group_id>-?\d+)," + r"(?P<partition>\d+)," + r"(?P<node_id>[0-9a-fA-F]+)," + r"(?P<primary>[PB])," + r"(?P<state>[A-Z_]+)," + r"(?P<update_counter>\d+)," + r"(?P<partition_size>\d+)," + r"\[(?P<node_addresses>[^\]]*)\]" + r"(?:,(?P<attr_values>.*))?$") + + groups = {} + cur_group = None + + for line in output.splitlines(): + line = line.strip() + + match = group_pattern.search(line) + if match: + cur_group = CacheGroupDistribution(group_id=int(match.group("group_id")), + name=match.group("name"), + partitions={}) + groups[cur_group.name] = cur_group + continue + + match = row_pattern.match(line) + if match and cur_group is not None: + attrs = {} + if user_attributes and match.group("attr_values") is not None: + values = [v.strip() for v in match.group("attr_values").split(",")] + attrs = dict(zip(sorted(user_attributes), values)) Review Comment: **[major]** My round-1 suggestion here was itself incomplete - sorry for the detour. The `TreeMap` only exists on the server side: `CacheDistributionNode.userAttrs` is an IDTO field serialized via `U.writeStringMap`/`U.readStringMap`, and `readStringMap` materializes a plain `HashMap`. `CacheDistributionTaskResult.print()` runs in the control.sh JVM **after** deserialization, so both the header and the per-row values iterate that `HashMap` - the real order is HashMap iteration order: deterministic, but neither the `--user-attributes` argument order nor alphabetical. With a single attribute (all current usages) any zip works, but for 2+ attributes `sorted()` can still mis-assign values. The output already carries the truth: the header line (`[groupId,partition,...,nodeAddresses,<attr names>]`) is printed by iterating the same map as the row values, so header order == value order always. Robust fix: capture the trailing attribute names from the header line and zip with those instead of `sorted(user_attributes)`. The comment above `row_pattern` ("in the same order they were passed") needs the same correction, otherwise the next editor will "fix" the code back to the comment. ########## modules/ducktests/tests/ignitetest/tests/mdc/partition_resilience_test.py: ########## @@ -0,0 +1,215 @@ +# 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. + +""" +MDC cluster resilience to network and data center failures. + +Covers the full split-brain lifecycle (partition -> active half + read-only half -> +heal -> read-only half rejoins via restart), the loss and return of the main DC, +and short network blips that must NOT split the cluster. +""" +from time import sleep + +from ducktape.mark import parametrize + +from ignitetest.services.mdc.mdc_cluster import MdcCluster, cross_dc_network, DC_1, DC_2 +from ignitetest.utils import cluster, ignite_versions +from ignitetest.utils.ignite_test import IgniteTest +from ignitetest.utils.version import DEV_BRANCH + +CACHE_NAME = "mdc-resilience" + +BACKUPS = 1 + +# Time for discovery to detect the partition and for both half-rings to complete PME. +SPLIT_SETTLE_SECS = 10 Review Comment: **[minor, flakiness]** `SPLIT_SETTLE_SECS = 10` equals the default `failure_detection_timeout=10_000` exactly. With iptables DROP the surviving ring needs the *full* FDT just to detect the loss, then process the node failures and finish PME - so `verify_split_brain()` fires right at the theoretical minimum, and on a slow CI runner this is a coin flip (false-red). Consider waiting deterministically (e.g. `await_event` for the node-failed message on the surviving DC before verifying) or at least a comfortable margin above FDT. ########## modules/ducktests/tests/ignitetest/tests/mdc/thin_client_test.py: ########## @@ -0,0 +1,154 @@ +# 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. + +""" +Thin client data-center-aware routing. + +Every thin client is configured with the addresses of ALL server nodes of BOTH DCs. +A client pinned to a data center must route partition-aware reads to nodes of its own DC: +with a large cross-DC netem delay its average read latency stays small. + +The partition part verifies the thin client experience of a split-brain: a client +pinned to the main-DC half keeps writing; a client pinned to the read-only half still +reads cleanly (falling back to the reachable nodes from its address list) but gets +every write rejected; and writes resume after the heal and rejoin. + +The thin client services are registered into their DC's network group, so both the +netem delay and the iptables partition apply to their traffic as well. +""" +from time import sleep + +from ducktape.mark import parametrize + +from ignitetest.services.ignite_app import IgniteApplicationService +from ignitetest.services.mdc.mdc_cluster import MdcCluster, dc_jvm_opts, DC_1, DC_2, cross_dc_network, THIN_LOAD_APP +from ignitetest.services.utils.ignite_configuration import IgniteThinClientConfiguration +from ignitetest.utils import cluster, ignite_versions +from ignitetest.utils.ignite_test import IgniteTest +from ignitetest.utils.version import DEV_BRANCH + +CACHE_NAME = "mdc-thin" + +KEYS = 100 + +PINNED_GET_ITERS = 200 +PUT_ITERS = 30 + +OFFSET_DURING = 1_000_000 +OFFSET_REJECTED_PROBES = 9_000_000 +OFFSET_AFTER = 2_000_000 + +SPLIT_SETTLE_SECS = 15 + + +class MdcThinClientTest(IgniteTest): + """ + Tests for thin client DC-aware routing and behavior through a partition. + """ + @cluster(num_nodes=7) + @ignite_versions(str(DEV_BRANCH)) + @parametrize(cross_dc_latency_ms=100) + def test_thin_client_dc_aware_routing_and_partition(self, ignite_version, cross_dc_latency_ms): + """ + DC-pinned thin client reads locally (latency far below the cross-DC delay); + through a partition the pinned clients behave like their half-ring: main half writes, read-only half reads + but rejects writes, and writes resume after the heal. + """ + mdc = MdcCluster(self, ignite_version, srv_per_dc=2, runners_per_dc={DC_1: 1}, + client_connector=True) + + # All thin clients get the full address list of both DCs: DC preference must come + # from routing, not from the address list. + cli_dc1 = self._thin_client(mdc, dc_jvm_opts(DC_1)) + cli_dc2 = self._thin_client(mdc, dc_jvm_opts(DC_2)) + + # Register the thin client hosts into the network groups, so netem and the partition apply to them. + mdc.register(DC_1, cli_dc1) + mdc.register(DC_2, cli_dc2) + + with cross_dc_network(self.logger, mdc, delay_ms=cross_dc_latency_ms) as net: + mdc.start_servers() + + mdc.generate_data(DC_1, CACHE_NAME, 0, KEYS, backups=1) + + svc = mdc.run_service(cli_dc1, {"mode": "GET", "cacheName": CACHE_NAME, "keyFrom": 0, + "keyTo": KEYS, "iterations": PINNED_GET_ITERS, + "resultPrefix": "pinnedGet"}) + + avg_cli_dc_1 = mdc.result_float(svc, "pinnedGetAvgOpMs") + err_cnt_cli_dc_1 = mdc.result_int(svc, "pinnedGetErrCnt") + + self.logger.info(f"Thin client routing latency [delayMs={cross_dc_latency_ms}, getAvgOpMs={avg_cli_dc_1}, " + f"getErrCnt={err_cnt_cli_dc_1}]") + + assert avg_cli_dc_1 < cross_dc_latency_ms, \ Review Comment: **[minor]** This proves DC-locality only for the DC1-pinned client, and DC1 happens to be the first/main DC - a routing regression of the form "always prefer the first configured DC" would still pass. `cli_dc2` is already registered: a symmetric `pinnedGet` burst from it before the partition, with the same `< cross_dc_latency_ms` assert, would pin down actual DC-awareness at the cost of a few seconds. ########## modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcThinClientLoadApplication.java: ########## @@ -0,0 +1,143 @@ +/* + * 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. + */ + +package org.apache.ignite.internal.ducktest.tests.mdc; + +import javax.cache.CacheException; +import com.fasterxml.jackson.databind.JsonNode; +import org.apache.ignite.IgniteException; +import org.apache.ignite.client.ClientCache; +import org.apache.ignite.client.ClientException; +import org.apache.ignite.internal.ducktest.tests.dto.IndexedDataRecord; +import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication; +import org.apache.ignite.internal.ducktest.utils.OpStats; + +import static org.apache.ignite.internal.ducktest.utils.Utils.fmtMs; +import static org.apache.ignite.internal.ducktest.utils.Utils.getEnum; +import static org.apache.ignite.internal.ducktest.utils.Utils.timed; + +/** + * Thin client MDC load application. + * <p> + * The application performs synchronous {@code GET} or {@code PUT} operations against an + * existing cache and records the average operation latency, which the python side uses as + * a crude proxy for "which data center served the request": with a high cross-DC netem + * delay, DC-local routing yields a small average, cross-DC routing a large one. + * <p> + * Parameters: {@code mode} ({@code GET}/{@code PUT}), {@code cacheName}, {@code keyFrom}, + * {@code keyTo}, {@code iterations}, {@code inadmissible} (PUT only), + * {@code resultPrefix}. + * <p> + */ +public class MdcThinClientLoadApplication extends IgniteAwareApplication { + /** {@inheritDoc} */ + @Override public void run(JsonNode jNode) throws Exception { + LoadMode mode = getEnum(jNode, "mode", LoadMode.class); + + if (mode != LoadMode.GET && mode != LoadMode.PUT) + throw new IllegalArgumentException("Unsupported thin client mode: " + mode); + + String cacheName = jNode.get("cacheName").asText(); + + int keyFrom = jNode.path("keyFrom").asInt(0); + int keyTo = jNode.path("keyTo").asInt(Integer.MAX_VALUE); + long iterations = jNode.path("iterations").asLong(100); + + boolean put = mode == LoadMode.PUT; + boolean inadmissible = jNode.path("inadmissible").asBoolean(false); + + String pfx = jNode.path("resultPrefix").asText(""); + + markInitialized(); + + ClientCache<Integer, IndexedDataRecord> cache = client.cache(cacheName); + + log.info("MDC thin client load started [mode=" + mode + ", cache=" + cacheName + + ", keyFrom=" + keyFrom + ", keyTo=" + keyTo + ", iterations=" + iterations + + ", inadmissible=" + inadmissible + "]"); + + long opsCnt = 0; + long errCnt = 0; + + OpStats stats = new OpStats(); + + long startTs = System.currentTimeMillis(); + + int key0 = keyFrom; + + for (long i = 0; i < iterations && !terminated(); i++) { + int key = key0; + + boolean ok = true; + + try { + if (put) { + IndexedDataRecord val = new IndexedDataRecord(key); + + timed(stats, () -> cache.put(key, val)); + } + else { + IndexedDataRecord val = timed(stats, () -> cache.get(key)); + + if (val == null || !val.equals(new IndexedDataRecord(key))) + throw new IgniteException("Read entry is missed or corrupted [key=" + key + + ", val=" + val + "]"); + } + } + catch (ClientException | CacheException | IgniteException e) { + ok = false; + + if (inadmissible) Review Comment: **[minor]** Same as in `MdcContinuousLoadApplication`: without the write guard a GET run with a stray `inadmissible=true` swallows corrupted reads as "Put rejected as expected". ```suggestion if (put && inadmissible) ``` ########## modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py: ########## @@ -0,0 +1,146 @@ +# 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. + +""" +MDC transactional load through a cross-DC network partition. + +A TRANSACTIONAL cache spans both data centers: with backups=1 and the +MdcAffinityBackupFilter every partition owns exactly one copy per DC, so every +implicit-transaction write (a plain put on a transactional cache) must reach a +node in the other DC. A continuous single-threaded insert load runs from the main +DC; the instant the DCs are partitioned the next commit cannot reach all partition +copies and fails with a cache exception. The load cuts itself off on that first +exception and records how many inserts had succeeded. + +After the split settles the test asserts that the cluster really split-brained, +that the load stopped because of the partition (not because it ran out of work), +and - the point of the scenario - that the aborted implicit transactions left +nothing hanging on either half-ring and no suspicious entries in the server logs. + +Data accessibility during the split is deliberately NOT checked: a transactional +cache needs all partition copies available, which a split-brained half-ring cannot +offer. +""" +from time import sleep + +from ducktape.mark import parametrize + +from ignitetest.services.mdc.mdc_cluster import MdcCluster, cross_dc_network, DC_1, DC_2 +from ignitetest.utils import cluster, ignite_versions +from ignitetest.utils.ignite_test import IgniteTest +from ignitetest.utils.version import DEV_BRANCH + +CACHE_NAME = "mdc-tx-load" + +BACKUPS = 1 + +# Fresh, disjoint key range for the continuous insert load: the load advances the key +# on every success, so [LOAD_KEY_FROM, LOAD_KEY_FROM + successfulInserts) gets inserted. +LOAD_KEY_FROM_DC_1 = 1_000_000 +LOAD_KEY_TO_DC_1 = 10_000_000 + +LOAD_KEY_FROM_DC_2 = 10_000_000 +LOAD_KEY_TO_DC_2 = 20_000_000 + +# Let the load accumulate successful inserts before the DCs are cut apart. +LOAD_WARMUP_SECS = 15 + +# Time for discovery to detect the partition and for both half-rings to complete PME +# and account for the split. +SPLIT_SETTLE_SECS = 15 + + +class MdcTransactionalPartitionTest(IgniteTest): + """ + Transactional load resilience to a cross-DC network partition. + """ + @cluster(num_nodes=8) + @ignite_versions(str(DEV_BRANCH)) + @parametrize(cross_dc_latency_ms=100) + def test_transactional_load_cut_on_partition(self, ignite_version, cross_dc_latency_ms): + """ + Continuous implicit-transaction insert load from the main DC is cut off by the first Review Comment: **[nit]** "implicit-transaction" (here, in the module docstring and in the comments below): `TX_PUT` actually runs *explicit* `txStart(PESSIMISTIC, REPEATABLE_READ)` transactions (`MdcContinuousLoadApplication`). Either the wording should say "explicit pessimistic transactions", or - if implicit txs (a plain `cache.put` on a transactional cache) were the intent - the load mode should change; the two abort differently mid-prepare, so for a recovery scenario the distinction isn't cosmetic. ########## modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcDataGeneratorApplication.java: ########## @@ -0,0 +1,76 @@ +/* + * 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. + */ + +package org.apache.ignite.internal.ducktest.tests.mdc; + +import java.util.Map; +import java.util.TreeMap; +import java.util.function.IntFunction; +import com.fasterxml.jackson.databind.JsonNode; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.internal.IgniteInterruptedCheckedException; +import org.apache.ignite.internal.ducktest.tests.dto.IndexedDataRecord; + +/** + * Populates the MDC cache with a deterministic data set: keys in {@code [from, to)}, + * values {@code IndexedDataRecord(key)} (or the key itself in {@code sqlMode}). + * Runs it until the network partition event. Review Comment: **[nit]** The new wording promises more than the code does: the generator writes `[from, to)` and finishes, and `run_app()` waits for that completion - in every scenario it is done before the partition is even enabled; nothing here runs "until the network partition event". ```suggestion * Run it to completion before enabling the network partition. ``` ########## modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py: ########## @@ -0,0 +1,542 @@ +# 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. + +""" +MDC test fixture. + + mdc = MdcCluster(self, ignite_version, srv_per_dc=3, 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_partition(DC_1, DC_2) + ... +""" +from typing import Dict, List, Optional, Union + +from ignitetest.services.ignite import IgniteService +from ignitetest.services.ignite_app import IgniteApplicationService +from ignitetest.services.network_group.configuration import NetworkGroupStore, CrossNetworkGroupConfiguration +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.ignite_configuration.discovery import TcpDiscoverySpi, from_ignite_cluster, \ + from_ignite_services +from ignitetest.services.utils.ssl.client_connector_configuration import ClientConnectorConfiguration +from ignitetest.utils.version import IgniteVersion + +DC_1 = "DC1" +DC_2 = "DC2" +DCS = (DC_1, DC_2) + +IGNITE_STARTUP_TIMEOUT_SEC = 90 + +DATA_CENTER_ATTR = "IGNITE_DATA_CENTER_ID" +IGNITE_SQL_RETRY_TIMEOUT_ATTR = "IGNITE_SQL_RETRY_TIMEOUT" + +IGNITE_SQL_RETRY_TIMEOUT_MS = 1_000 + +_APP_PKG = "org.apache.ignite.internal.ducktest.tests.mdc." + +GENERATOR_APP = _APP_PKG + "MdcDataGeneratorApplication" +DATA_CHECKER_APP = _APP_PKG + "MdcDataCheckerApplication" +LOAD_APP = _APP_PKG + "MdcContinuousLoadApplication" +THIN_LOAD_APP = _APP_PKG + "MdcThinClientLoadApplication" + +# Suspicious server log patterns: none of them is expected in any MDC scenario, +# partitioned or not. Matched against the node console capture. +LRT_PATTERN = "long running transactions" +PME_FREEZE_PATTERN = "Failed to wait for partition map exchange" +LOST_PARTITIONS_PATTERN = "Detected lost partitions" + + +def dc_jvm_opts(dc: str) -> List[str]: + """ + :return: JVM options assigning a node to the given data center. + """ + 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]: + """ + Normalizes an int-or-dict per-DC count into a dict, e.g. 3 -> {DC1: 3, DC2: 3}. + """ + return dict(value) if isinstance(value, dict) else {dc: value for dc in DCS} + + +class MdcCluster: + """ + Owns the per-DC Ignite services and reusable application services of an MDC test, + plus the MDC-specific verification helpers. + + :param test: The ducktape test instance. + :param ignite_version: Ignite version string. + :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. + """ + def __init__(self, test, ignite_version: str, 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, + 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(). + cfg_kwargs = { + "version": IgniteVersion(ignite_version), + "discovery_spi": TcpDiscoverySpi(), + "network_timeout": network_timeout, + "communication_spi": TcpCommunicationSpi(connect_timeout=tcp_connect_timeout) + } + + if client_connector: + cfg_kwargs["client_connector_configuration"] = ClientConnectorConfiguration() + + self.ignite_config = IgniteConfiguration(**cfg_kwargs) + + self.srv_per_dc = _per_dc(srv_per_dc) + + 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} + + 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()} + + 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()} + + # Extra services (e.g. thin clients) registered into a DC's network group. + self.extras: Dict[str, List] = {dc: [] for dc in 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). + self._started_apps = set() + + # Admissibility checks run on reusable services, so each check needs a unique result prefix. + self._adm_checks = 0 + + 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. + """ + discovery_spi = from_ignite_services(list(self.servers.values())) + + for service in self.servers.values(): + 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])) + + return IgniteApplicationService(self.test_context, client_cfg, jvm_opts=dc_jvm_opts(dc)) + + def register(self, dc: str, service): + """ + Registers an extra service (e.g. a thin client app) into a DC's network group, + so netem impairments and partitions apply to it. Must be called before + :func:`cross_dc_network` snapshots the registry into a :class:`NetworkGroupManager`. + """ + self.extras[dc].append(service) + + def network_registry(self) -> Dict[str, List]: + """ + :return: Network group registry: DC name -> all services belonging to that DC. + """ + registry = {} + + for dc in DCS: + services = [] + + if dc in self.servers: + services.append(self.servers[dc]) + + services += self.runners.get(dc, []) + services += self.loaders.get(dc, []) + services += self.extras.get(dc, []) + + if services: + registry[dc] = services + + return registry + + def thin_client_addresses(self) -> List[str]: + """ + :return: Thin client addresses of all server nodes across all DCs. + """ + 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] + + def start_servers(self): + """ + Starts all server services. + """ + for dc in sorted(self.servers): + self.servers[dc].start() + + def stop_servers(self): + """ + Stops all server services. + """ + for dc in sorted(self.servers): + self.servers[dc].stop() + + 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). + """ + self.servers[dc].stop() + self.servers[dc].start(clean=clean) + + if await_rebalance: + self.servers[dc].await_rebalance() + + def run_app(self, dc: str, java_class: str, params: dict, runner: int = 0) -> IgniteApplicationService: + """ + Runs a run-to-completion application on one of the DC's reusable runner services + and returns the service (for ``extract_result``). + """ + return self.run_service(self.runners[dc][runner], params, java_class=java_class) + + def run_service(self, svc: IgniteApplicationService, params: dict, + java_class: str = None) -> IgniteApplicationService: + """ + Runs any reusable run-to-completion application service (a runner, a registered + thin client, ...): the first start is clean, subsequent starts preserve work dirs. + Returns the service (for ``extract_result``). + """ + if java_class is not None: + svc.java_class_name = java_class + + svc.params = params + + svc.start(clean=self._first_start(svc)) + svc.wait() + svc.stop() + + return svc + + 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`. + """ + svc = self.loaders[dc][loader] + + svc.java_class_name = java_class + svc.params = params + + svc.start(clean=self._first_start(svc)) + + return svc + + def stop_loader(self, dc: str, loader: int = 0) -> IgniteApplicationService: + """ + Stops a background load application. The application finishes its loop, records + results and exits; a failed application fails the test here. + """ + svc = self.loaders[dc][loader] + + svc.stop() + + return svc + + def _first_start(self, svc) -> bool: + first = id(svc) not in self._started_apps + + self._started_apps.add(id(svc)) + + 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: + """ + 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. + """ + params = {"cacheName": cache_name, "backups": backups, "mainDc": main_dc, + "from": from_idx, "to": to_idx, "sqlMode": sql_mode, **cache_params} + + return self.run_app(dc, GENERATOR_APP, params) + + def check_data(self, dc: str, cache_name: str, from_idx: int, to_idx: int) -> Optional[IgniteApplicationService]: + """ + Verifies that every key in ``[from_idx, to_idx)`` is readable and holds the + expected value, from a client in the given DC. + + :return: The service that ran the check, or None for an empty range. + """ + if to_idx <= from_idx: + self.logger.debug(f"Nothing to check [cache={cache_name}, from={from_idx}, to={to_idx}]") + return None + + params = {"cacheName": cache_name, "from": from_idx, "to": to_idx} + + return self.run_app(dc, DATA_CHECKER_APP, params) + + 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 + application: an admissible check fails fast on the first rejected put, an + inadmissible check fails if any of the probe puts succeeds. + + Probe keys start at ``key_offset`` (defaults to 1_000_000) so they never intersect + with the data set verified by ``check_data``. + """ + self._adm_checks += 1 + + return self.run_load(dc, "PUT", cache_name, f"admCheck{self._adm_checks}", + keyFrom=key_offset, keyTo=key_offset + probes, + iterations=probes, inadmissible=not admissible) + + def run_load(self, dc: str, mode: str, cache_name: str, result_prefix: str, + runner: int = 0, **params) -> IgniteApplicationService: + """ + Runs a load burst (see ``MdcContinuousLoadApplication``) and returns the service. + ``result_prefix`` must be unique per burst because runner services are reused. + """ + 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: + """ + :return: Control utility bound to the given DC's servers. + """ + return ControlUtility(self.servers[dc]) + + def verify_cache_distribution(self, cache_name: str, copies_per_dc: Optional[int] = None, dc: str = DC_1): + """ + 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. + + :return: The CacheDistribution for further custom assertions. + """ + 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) + + 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. + """ + for dc in DCS: + self.verify_half_ring_healthy(dc) + + state = {dc: self.control(dc).cluster_state() for dc in DCS} + + baselines = {dc: {node.consistent_id for node in state[dc].baseline} for dc in DCS} + + common_nodes = baselines[DC_1] & baselines[DC_2] + + 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])}]" + + for dc in DCS: + coordinator = state[dc].coordinator + + assert coordinator, f"Coordinator is not found in {dc} half-ring baseline output!" + + 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 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}]" + + def verify_half_ring_healthy(self, dc: str): + """ + Verifies that a half-ring is fully alive, ACTIVE, and its baseline matches its size. + """ + exp_alive_nodes = self.srv_per_dc[dc] + act_alive_nodes = len(self.servers[dc].alive_nodes) + + assert act_alive_nodes == exp_alive_nodes, \ + f"{exp_alive_nodes} nodes should be alive in {dc}! [actual={act_alive_nodes}]" + + cluster_state = self.control(dc).cluster_state() + + assert "ACTIVE" == cluster_state.state, \ + f"{dc} half-ring 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"[exp={exp_alive_nodes}, actual_baseline={cluster_state.baseline}]" + + 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. + """ + 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}]" + + def verify_servers_log_clean(self): + """ + Verifies the negative invariants on all server nodes: no long running transactions + were detected, no PME hang and no lost partitions were reported. + """ + for pattern in (LRT_PATTERN, PME_FREEZE_PATTERN, LOST_PARTITIONS_PATTERN): Review Comment: **[minor]** After a `restart()` the previous incarnation's log is rotated to `ignite.log.N` (`IgniteAwareService.__update_node_log_file`), and the default `node.log_file` glob `ignite*.log` does not match it. So the final `verify_servers_log_clean()` in the transactional test never sees what DC2 logged between the pre-restart check and the restart - exactly the partition-heal PME window. An explicit glob covers both: `svc.check_event_absent(pattern, log_file="ignite.log*")`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
