diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/dto/IndexedDataRecord.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/dto/IndexedDataRecord.java new file mode 100644 index 0000000000000..f479d663a31cf --- /dev/null +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/dto/IndexedDataRecord.java @@ -0,0 +1,96 @@ +/* + * 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.dto; + +import java.util.Arrays; +import java.util.List; +import java.util.Objects; + +/** */ +public class IndexedDataRecord { + /** The base index converted to a string. */ + private final String strRepresentation; + + /** The primitive index boxed as an Integer. */ + private final Integer boxedIdx; + + /** True if the base index is an even number, false otherwise. */ + private final Boolean isEven; + + /** The first character of the string representation. */ + private final char leadingCharacter; + + /** An array containing the index, its double, and its square. */ + private final Integer[] seqArr; + + /** A list view of the sequence array. */ + private final List seqList; + + /** + * Constructs an immutable record by generating derived properties from the provided index. + * + * @param idx the base integer used to compute all internal fields + */ + public IndexedDataRecord(int idx) { + this.strRepresentation = String.valueOf(idx); + this.boxedIdx = idx; + this.isEven = (idx % 2 == 0); + this.leadingCharacter = this.strRepresentation.charAt(0); + this.seqArr = new Integer[] {idx, idx * 2, idx * idx}; + this.seqList = Arrays.asList(this.seqArr); + } + + /** {@inheritDoc} */ + @Override public boolean equals(Object o) { + if (this == o) + return true; + + if (o == null || getClass() != o.getClass()) + return false; + + IndexedDataRecord that = (IndexedDataRecord)o; + + return leadingCharacter == that.leadingCharacter + && Objects.equals(strRepresentation, that.strRepresentation) + && Objects.equals(boxedIdx, that.boxedIdx) + && Objects.equals(isEven, that.isEven) + && Arrays.equals(seqArr, that.seqArr) + && Objects.equals(seqList, that.seqList); + } + + /** {@inheritDoc} */ + @Override public int hashCode() { + int result = Objects.hash(strRepresentation, boxedIdx, isEven, leadingCharacter, seqList); + + result = 31 * result + Arrays.hashCode(seqArr); + + return result; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return "IndexedDataRecord{" + + "stringRepresentation='" + strRepresentation + '\'' + + ", boxedIndex=" + boxedIdx + + ", isEven=" + isEven + + ", leadingCharacter=" + leadingCharacter + + ", sequenceArray=" + Arrays.toString(seqArr) + + ", sequenceList=" + seqList + + '}'; + } +} diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/LoadMode.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/LoadMode.java new file mode 100644 index 0000000000000..a6f4bc2b546be --- /dev/null +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/LoadMode.java @@ -0,0 +1,46 @@ +/* + * 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; + +/** Operation modes shared by the MDC load applications. */ +public enum LoadMode { + /** Cache API reads with value verification. */ + GET, + + /** Cache API writes. */ + PUT, + + /** Transactional cache API writes (requires a TRANSACTIONAL cache). */ + TX_PUT, + + /** SQL reads with value verification (requires an SQL-enabled cache). */ + SQL_SELECT, + + /** SQL DML writes (requires an SQL-enabled cache). */ + SQL_PUT; + + /** @return Whether the mode writes to the cache. */ + public boolean isWrite() { + return this == PUT || this == TX_PUT || this == SQL_PUT; + } + + /** @return Whether the mode operates through SQL. */ + public boolean isSql() { + return this == SQL_SELECT || this == SQL_PUT; + } +} 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 new file mode 100644 index 0000000000000..447486faa9b75 --- /dev/null +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcCacheAwareApplication.java @@ -0,0 +1,177 @@ +/* + * 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. + *

+ * Supported cache parameters (all optional unless stated otherwise): + *

+ */ +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 mdcCache(JsonNode jNode) { + return ignite.getOrCreateCache(this.mdcCacheConfiguration(jNode)); + } + + /** + * Same MDC cache configuration with a {@link QueryEntity} on top, so the cache is + * queryable via SQL: {@code SELECT _VAL FROM "".LOAD WHERE _KEY = ?}. + * + * @param jNode Parameters. + * @return SQL-enabled cache configured with the MDC topology validator and backup filter. + */ + protected IgniteCache mdcSqlCache(JsonNode jNode) { + CacheConfiguration 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 CacheConfiguration 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 dcs = new HashSet<>(); + + jNode.get("datacenters").forEach(dc -> dcs.add(dc.asText())); + + topValidator.setDatacenters(dcs); + } + else { + String mainDc = jNode.hasNonNull("mainDc") ? jNode.get("mainDc").asText().trim() : ""; + + if (mainDc.isEmpty()) + throw new IllegalArgumentException("Either 'datacenters' or a non-empty 'mainDc' must be specified."); + + 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))); + } + + /** + * @param cacheName Cache name. + * @return Existing cache. The cache must have been created by the generator beforehand. + */ + protected IgniteCache existingCache(String cacheName) { + return ignite.cache(cacheName); + } + + /** + * @return Data center id this client belongs to (passed via {@link IgniteSystemProperties#IGNITE_DATA_CENTER_ID}). + */ + protected static String dcId() { + return IgniteSystemProperties.getString(IGNITE_DATA_CENTER_ID); + } +} diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcContinuousLoadApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcContinuousLoadApplication.java new file mode 100644 index 0000000000000..ae46919328b19 --- /dev/null +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcContinuousLoadApplication.java @@ -0,0 +1,304 @@ +/* + * 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). + *

+ * Parameters: + *

    + *
  • {@code mode} - one of {@code GET, PUT, TX_PUT, SQL_SELECT, SQL_PUT};
    • + *
  • {@code cacheName} - cache to operate on;
    • + *
  • {@code createCache} - if {@code true}, (re)creates the cache from the MDC + * configuration parameters (see {@link MdcCacheAwareApplication}); otherwise the + * cache must already exist;
    • + *
  • {@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;
    • + *
  • {@code iterations} - number of operations; {@code 0} means "run until terminated";
    • + *
  • {@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;
    • + *
  • {@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;
    • + *
  • {@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;
    • + *
  • {@code opPauseMs} - pause between operations, default 0;
    • + *
  • {@code resultPrefix} - prefix for recorded results, so that several runs reusing one + * service produce uniquely named results;
    • + *
  • {@code txConcurrency} / {@code txIsolation} - transaction parameters for {@code TX_PUT}.
    • + *
+ */ +public class MdcContinuousLoadApplication extends MdcCacheAwareApplication { + /** */ + public static final TransactionConcurrency DFLT_TX_CONCURRENCY = PESSIMISTIC; + + /** */ + public static final TransactionIsolation DFLT_TX_ISOLATION = REPEATABLE_READ; + + /** */ + public static final int DFLT_TX_TIMEOUT = 1_000; + + /** Cache for the cache API modes ({@code null} in SQL modes). */ + private IgniteCache cache; + + /** Cache for the SQL modes ({@code null} in cache API modes). */ + private IgniteCache 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; + + /** */ + private int txTimeout; + + /** {@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); + txTimeout = jNode.path("txTimeout").asInt(DFLT_TX_TIMEOUT); + + 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; + + // A read reaching here is a missed or corrupted entry, never a rejected write: + // 'inadmissible' must not excuse it. + if (mode.isWrite() && inadmissible) + log.info("Write rejected as expected [dc=" + dcId() + ", key=" + key + + ", msg=" + e.getMessage() + "]"); + else if (stopOnError) { + log.warn("Operation failed, cutting the load on first error [dc=" + dcId() + ", mode=" + mode + + ", key=" + key + ", succeeded=" + opsCnt + ", msg=" + e.getMessage() + "]", e); + + errCnt++; + stoppedOnError = true; + + break; + } + else if (continueOnError) + log.warn("Operation failed, tolerated [dc=" + dcId() + ", mode=" + mode + + ", key=" + key + ", msg=" + e.getMessage() + "]"); + else + throw new IllegalStateException("Operation failed [dc=" + dcId() + ", mode=" + mode + + ", key=" + key + "]", e); + } + + if (ok) { + opsCnt++; + + long now = System.currentTimeMillis(); + + maxStallMs = Math.max(maxStallMs, now - lastOkTs); + lastOkTs = now; + } + else + errCnt++; + + // Writes advance on success only, so [keyFrom, keyFrom + opsCnt) is guaranteed written. + // The inadmissible-probe mode advances always to probe distinct keys. Reads cycle the range. + if (ok || (mode.isWrite() && inadmissible)) { + key++; + + if (key >= keyTo) + key = keyFrom; + } + + if (opPauseMs > 0) + Thread.sleep(opPauseMs); + } + + long durationMs = System.currentTimeMillis() - startTs; + + if (mode.isWrite() && inadmissible && opsCnt > 0) { + throw new IllegalStateException("Write load is admissible while expected to be inadmissible [dc=" + + dcId() + ", mode=" + mode + ", succeeded=" + opsCnt + ", rejected=" + errCnt + "]"); + } + + recordResult(pfx + "OpsCnt", String.valueOf(opsCnt)); + recordResult(pfx + "ErrCnt", String.valueOf(errCnt)); + recordResult(pfx + "StoppedOnError", String.valueOf(stoppedOnError)); + recordResult(pfx + "DurationMs", String.valueOf(durationMs)); + + recordResult(pfx + "AvgOpMs", fmtMs(stats.avgNs())); + recordResult(pfx + "MinOpMs", fmtMs(stats.minNs())); + recordResult(pfx + "MaxOpMs", fmtMs(stats.maxNs())); + + recordResult(pfx + "DerivedTps", String.format(Locale.US, "%.1f", stats.tps())); + + recordResult(pfx + "MaxStallMs", String.valueOf(maxStallMs)); + + log.info("MDC load finished [dc=" + dcId() + ", mode=" + mode + ", ops=" + opsCnt + + ", errs=" + errCnt + ", stoppedOnError=" + stoppedOnError + ", durationMs=" + durationMs + + ", avgOpMs=" + fmtMs(stats.avgNs()) + ", maxOpMs=" + fmtMs(stats.maxNs()) + + ", maxStallMs=" + maxStallMs + "]"); + + markFinished(); + } + + /** + * Executes a single operation of the given mode against the given key. Throws a + * {@link CacheException} or {@link IgniteException} on operation failure (e.g. a write + * rejected by the topology validator). + */ + private void doOperation(LoadMode mode, int key) { + switch (mode) { + case GET: { + IndexedDataRecord val = timed(stats, () -> cache.get(key)); + + if (val == null || !val.equals(new IndexedDataRecord(key))) + throw new IgniteException("Read entry is missed or corrupted [dc=" + dcId() + ", key=" + key + + ", val=" + val + "]"); + + break; + } + + case PUT: { + IndexedDataRecord val = new IndexedDataRecord(key); + + timed(stats, () -> cache.put(key, val)); + + break; + } + + case TX_PUT: { + IndexedDataRecord val = new IndexedDataRecord(key); + + timed(stats, () -> { + try (Transaction tx = ignite.transactions().txStart(txConcurrency, txIsolation, txTimeout, 1)) { + cache.put(key, val); + + tx.commit(); + } + }); + + break; + } + + case SQL_PUT: { + SqlFieldsQuery qry = new SqlFieldsQuery(mergeSql).setArgs(key, key); + + timed(stats, () -> sqlCache.query(qry).getAll()); + + break; + } + + case SQL_SELECT: { + SqlFieldsQuery qry = new SqlFieldsQuery(selectSql).setArgs(key); + + List> rows = timed(stats, () -> sqlCache.query(qry).getAll()); + + if (rows.isEmpty() || !Objects.equals(rows.get(0).get(0), key)) + throw new IgniteException("SQL row is missed or corrupted [dc=" + dcId() + ", key=" + key + + ", rows=" + rows + "]"); + + break; + } + + default: + throw new IllegalArgumentException("Unknown mode: " + mode); + } + } +} diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcDataCheckerApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcDataCheckerApplication.java new file mode 100644 index 0000000000000..2232a5b59f51a --- /dev/null +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcDataCheckerApplication.java @@ -0,0 +1,72 @@ +/* + * 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 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; + +/** + * Verifies that all entries written by {@link MdcDataGeneratorApplication} are readable + * and hold expected values. Intended to be run from a client in each DC after the + * network partition: reads must succeed everywhere, even in the read-only DC. + */ +public class MdcDataCheckerApplication extends MdcCacheAwareApplication { + /** {@inheritDoc} */ + @Override public void run(JsonNode jNode) throws IgniteInterruptedCheckedException { + String cacheName = jNode.get("cacheName").asText(); + int from = jNode.path("from").asInt(0); + int to = jNode.path("to").asInt(10_000); + + markInitialized(); + waitForActivation(); + + IgniteCache cache = existingCache(cacheName); + + log.info("Data check started [dc=" + dcId() + ", cache=" + cache.getName() + + ", from=" + from + ", to=" + to + "]"); + + int missed = 0; + int corrupted = 0; + + for (int i = from; i < to && !terminated(); i++) { + IndexedDataRecord obj = cache.get(i); + + if (obj == null) { + missed++; + + log.error("Entry is missed [dc=" + dcId() + ", key=" + i + "]"); + } + else if (!obj.equals(new IndexedDataRecord(i))) { + corrupted++; + + log.error("Entry is corrupted [dc=" + dcId() + ", key=" + i + ", val=" + obj + "]"); + } + } + + if (missed > 0 || corrupted > 0) { + throw new IllegalStateException("Data check failed [dc=" + dcId() + + ", missed=" + missed + ", corrupted=" + corrupted + "]"); + } + + log.info("Data check passed [dc=" + dcId() + ", entries=" + (to - from) + "]"); + + markFinished(); + } +} diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcDataGeneratorApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcDataGeneratorApplication.java new file mode 100644 index 0000000000000..335efb52b8183 --- /dev/null +++ b/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}). + * Run it to completion before enabling the network partition. + */ +public class MdcDataGeneratorApplication extends MdcCacheAwareApplication { + /** {@inheritDoc} */ + @Override public void run(JsonNode jNode) throws IgniteInterruptedCheckedException { + int from = jNode.path("from").asInt(0); + int to = jNode.path("to").asInt(10_000); + int batchSize = jNode.path("batchSize").asInt(1_024); + boolean sqlMode = jNode.path("sqlMode").asBoolean(false); + + markInitialized(); + waitForActivation(); + + log.info("Data generation started [dc=" + dcId() + ", sqlMode=" + sqlMode + + ", from=" + from + ", to=" + to + "]"); + + if (sqlMode) + load(mdcSqlCache(jNode), from, to, batchSize, i -> i); + else + load(mdcCache(jNode), from, to, batchSize, IndexedDataRecord::new); + + log.info("Data generation finished [dc=" + dcId() + ", entries=" + (to - from) + "]"); + + markFinished(); + } + + /** + * Streams keys {@code [from, to)} into the cache in {@code putAll} batches, deriving + * each value from its key. + */ + private void load(IgniteCache cache, int from, int to, int batchSize, IntFunction valFn) { + Map batch = new TreeMap<>(); + + for (int i = from; i < to && !terminated(); i++) { + batch.put(i, valFn.apply(i)); + + if (batch.size() >= batchSize) { + cache.putAll(batch); + batch.clear(); + } + } + + if (!batch.isEmpty() && !terminated()) + cache.putAll(batch); + } +} diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcThinClientLoadApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcThinClientLoadApplication.java new file mode 100644 index 0000000000000..f2c6bbea5ac39 --- /dev/null +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/mdc/MdcThinClientLoadApplication.java @@ -0,0 +1,145 @@ +/* + * 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. + *

+ * 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. + *

+ * Parameters: {@code mode} ({@code GET}/{@code PUT}), {@code cacheName}, {@code keyFrom}, + * {@code keyTo}, {@code iterations}, {@code inadmissible} (PUT only), + * {@code resultPrefix}. + *

+ */ +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 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; + + // A read reaching here is a missed or corrupted entry, never a rejected write: + // 'inadmissible' must not excuse it. + if (put && inadmissible) + log.info("Put rejected as expected [key=" + key + ", msg=" + e.getMessage() + "]"); + else + throw new IllegalStateException("Operation failed [mode=" + mode + ", key=" + key + "]", e); + } + + if (ok) + opsCnt++; + else + errCnt++; + + // Writes advance always here: the inadmissible probe covers distinct keys, and for + // an admissible run any failure fails fast above, so success and attempt counts match. + key0++; + + if (key0 >= keyTo) + key0 = keyFrom; + } + + long durationMs = System.currentTimeMillis() - startTs; + + if (put && inadmissible && opsCnt > 0) { + throw new IllegalStateException("Put load is admissible while expected to be inadmissible " + + "[succeeded=" + opsCnt + ", rejected=" + errCnt + "]"); + } + + recordResult(pfx + "OpsCnt", String.valueOf(opsCnt)); + recordResult(pfx + "ErrCnt", String.valueOf(errCnt)); + + recordResult(pfx + "AvgOpMs", fmtMs(stats.avgNs())); + recordResult(pfx + "MinOpMs", fmtMs(stats.minNs())); + recordResult(pfx + "MaxOpMs", fmtMs(stats.maxNs())); + + log.info("MDC thin client load finished [mode=" + mode + ", ops=" + opsCnt + + ", errs=" + errCnt + ", durationMs=" + durationMs + + ", avgOpMs=" + fmtMs(stats.avgNs()) + "]"); + + markFinished(); + } +} diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/persistence_upgrade_test/DataLoaderAndCheckerApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/persistence_upgrade_test/DataLoaderAndCheckerApplication.java index 82108b3e5ef03..6a857b4505cb3 100644 --- a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/persistence_upgrade_test/DataLoaderAndCheckerApplication.java +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/persistence_upgrade_test/DataLoaderAndCheckerApplication.java @@ -17,13 +17,11 @@ package org.apache.ignite.internal.ducktest.tests.persistence_upgrade_test; -import java.util.Arrays; -import java.util.List; -import java.util.Objects; import com.fasterxml.jackson.databind.JsonNode; import org.apache.ignite.IgniteCache; import org.apache.ignite.configuration.CacheConfiguration; import org.apache.ignite.internal.IgniteInterruptedCheckedException; +import org.apache.ignite.internal.ducktest.tests.dto.IndexedDataRecord; import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication; import org.apache.ignite.internal.util.typedef.internal.U; @@ -40,16 +38,16 @@ public class DataLoaderAndCheckerApplication extends IgniteAwareApplication { markInitialized(); waitForActivation(); - CacheConfiguration cacheCfg = new CacheConfiguration<>("cache"); + CacheConfiguration cacheCfg = new CacheConfiguration<>("cache"); cacheCfg.setBackups(backups); - IgniteCache cache = ignite.getOrCreateCache(cacheCfg); + IgniteCache cache = ignite.getOrCreateCache(cacheCfg); log.info(check ? "Checking..." : " Preparing..."); for (int i = 0; i < entryCnt; i++) { - CustomObject obj = new CustomObject(i); + IndexedDataRecord obj = new IndexedDataRecord(i); if (!check) cache.put(i, obj); @@ -64,62 +62,4 @@ public class DataLoaderAndCheckerApplication extends IgniteAwareApplication { markFinished(); } - - /** - * - */ - private static class CustomObject { - /** String value. */ - private final String sVal; - - /** Integer value. */ - private final Integer iVal; - - /** Boolean value. */ - private final Boolean bVal; - - /** char value. */ - private final char cVal; - - /** Integer array value. */ - private final Integer[] iArVal; - - /** Integer List. */ - private final List iLiVal; - - /** - * @param idx Index. - */ - public CustomObject(int idx) { - sVal = String.valueOf(idx); - iVal = idx; - bVal = idx % 2 == 0; - cVal = sVal.charAt(0); - iArVal = new Integer[] {idx, idx * 2, idx * idx}; - iLiVal = Arrays.asList(iArVal); - } - - /** {@inheritDoc} */ - @Override public boolean equals(Object o) { - if (this == o) - return true; - if (o == null || getClass() != o.getClass()) - return false; - CustomObject obj = (CustomObject)o; - return cVal == obj.cVal - && Objects.equals(sVal, obj.sVal) - && Objects.equals(iVal, obj.iVal) - && Objects.equals(bVal, obj.bVal) - && Arrays.equals(iArVal, obj.iArVal) - && Objects.equals(iLiVal, obj.iLiVal); - } - - /** {@inheritDoc} */ - @Override public int hashCode() { - int result = Objects.hash(sVal, iVal, bVal, cVal, iLiVal); - result = 31 * result + Arrays.hashCode(iArVal); - return result; - } - } - } diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/utils/OpStats.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/utils/OpStats.java new file mode 100644 index 0000000000000..331cb5b915737 --- /dev/null +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/utils/OpStats.java @@ -0,0 +1,66 @@ +/* + * 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.utils; + +/** Accumulates latency of completed operations. */ +public class OpStats { + /** Sum of recorded latencies, nanoseconds. */ + private long totalNs; + + /** Minimum recorded latency, nanoseconds. */ + private long minNs = Long.MAX_VALUE; + + /** Maximum recorded latency, nanoseconds. */ + private long maxNs; + + /** Number of recorded operations. */ + private long cnt; + + /** Records a single operation latency. */ + public void record(long ns) { + totalNs += ns; + + minNs = Math.min(minNs, ns); + maxNs = Math.max(maxNs, ns); + + cnt++; + } + + /** @return Minimum latency in nanoseconds, or {@code -1} if nothing was recorded. */ + public long minNs() { + return cnt > 0 ? minNs : -1; + } + + /** @return Maximum latency in nanoseconds, or {@code -1} if nothing was recorded. */ + public long maxNs() { + return cnt > 0 ? maxNs : -1; + } + + /** @return Average latency in nanoseconds, or {@code -1} if nothing was recorded. */ + public double avgNs() { + return cnt > 0 ? totalNs / (double)cnt : -1; + } + + /** + * @return Derived throughput: recorded operations per second of pure operation time, + * or {@code -1} if nothing was recorded. + */ + public double tps() { + return totalNs > 0 ? cnt * 1e9 / totalNs : -1; + } +} diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/utils/Utils.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/utils/Utils.java new file mode 100644 index 0000000000000..25324348bdc4f --- /dev/null +++ b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/utils/Utils.java @@ -0,0 +1,84 @@ +/* + * 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.utils; + +import java.util.Locale; +import java.util.function.Supplier; +import com.fasterxml.jackson.databind.JsonNode; + +/** Small helpers shared by the ducktest applications. */ +public final class Utils { + /** */ + private Utils() { + // No-op. + } + + /** + * Runs the operation, records its latency into {@code stats} and returns its result. + * If the operation throws, nothing is recorded and the exception propagates. + */ + public static T timed(OpStats stats, Supplier op) { + long startNs = System.nanoTime(); + + T val = op.get(); + + stats.record(System.nanoTime() - startNs); + + return val; + } + + /** + * Runs the void operation and records its latency into {@code stats}. + * If the operation throws, nothing is recorded and the exception propagates. + */ + public static void timed(OpStats stats, Runnable op) { + timed(stats, () -> { + op.run(); + + return null; + }); + } + + /** Formats a nanosecond latency as milliseconds with 3 decimal places. */ + public static String fmtMs(double ns) { + return String.format(Locale.US, "%.3f", ns < 0 ? -1.0 : ns / 1e6); + } + + /** Parses an optional enum-valued field. Defaults only on missing/null. */ + public static > E getEnum(JsonNode jNode, String fieldName, E dfltVal) { + JsonNode field = jNode.path(fieldName); + + if (field.isMissingNode() || field.isNull()) + return dfltVal; + + return Enum.valueOf(dfltVal.getDeclaringClass(), field.asText().toUpperCase(Locale.ROOT)); + } + + /** + * Parses a required enum-valued field. Throws if the field is missing, null or not a valid + * constant of {@code cls} (case-insensitively) - a typo must fail the run, not pick a default. + */ + public static > E getEnum(JsonNode jNode, String fieldName, Class cls) { + JsonNode field = jNode.path(fieldName); + + if (field.isMissingNode() || field.isNull()) + throw new IllegalArgumentException("Missing required enum field '" + fieldName + "'"); + + return Enum.valueOf(cls, field.asText().toUpperCase(Locale.ROOT)); + } +} diff --git a/modules/ducktests/tests/checks/services/network_group/check_manager.py b/modules/ducktests/tests/checks/services/network_group/check_manager.py new file mode 100644 index 0000000000000..115c641d277cd --- /dev/null +++ b/modules/ducktests/tests/checks/services/network_group/check_manager.py @@ -0,0 +1,141 @@ +# 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 batched network probe NetworkGroupManager._log_network issues. + +The probe replaces three per-node SSH round-trips with a single command, so what +matters is that its combined output still splits back into exactly the three +sections the separate commands produced, and that an incomplete answer degrades +into a poorer overview instead of an exception - this runs inside an active +network partition, where nothing may fail the test around it. +""" + +import pytest + +from ignitetest.services.network_group.manager import NetworkGroupManager, PROBE_SECTION_SEPARATOR +from ignitetest.services.network_group.tc_rule_args import partition_chain_name + +INTERFACE = "eth0" + +CHAIN = partition_chain_name("DC1", "DC2") + +QDISC_SECTION = "qdisc netem 8001: root refcnt 2 limit 1000 delay 100ms loss 5%" + +# 'c0a80105' is 192.168.1.5 as tc reports it in a u32 destination match. +FILTER_SECTION = "filter parent 8001: protocol ip pref 1 u32 chain 0\n" \ + " match c0a80105/ffffffff at 16" + +IPTABLES_SECTION = "\n".join([ + "-P INPUT ACCEPT", + f"-N {CHAIN}", + f"-A {CHAIN} -s 192.168.1.5/32 -j DROP", + f"-A {CHAIN} -d 192.168.1.5/32 -j DROP" +]) + + +def _probe(*sections): + return f"\n{PROBE_SECTION_SEPARATOR}\n".join(sections) + + +# What each section renders as in the overview when the node did not answer with it. +NEUTRAL_RENDERINGS = [ + (NetworkGroupManager._parse_qdisc_constraints, "noqueue"), + (NetworkGroupManager._parse_filter_destinations, []), + (lambda lines: NetworkGroupManager._format_partition_drops( + NetworkGroupManager._parse_partition_drops(lines)), "") +] + + +class CheckNetworkProbe: + """ + Checks the composition and the parsing of the batched network probe. + """ + def check_probe_cmd_delimits_every_section(self, monkeypatch): + """The probe asks for all three dumps in one command, separator between each.""" + monkeypatch.setattr(NetworkGroupManager, "_get_default_network_interface", + lambda self, node: INTERFACE) + + cmd = NetworkGroupManager(logger=None, network_group_store=None, + network_group_registry={})._to_network_probe_cmd(node=None) + + assert cmd.count(PROBE_SECTION_SEPARATOR) == 2, "Three sections need exactly two separators" + + assert f"tc qdisc show dev {INTERFACE}" in cmd + assert f"tc filter show dev {INTERFACE}" in cmd + assert "iptables -S" in cmd + + # Unconditional chaining: one unavailable dump must not swallow the ones after it. + assert "&&" not in cmd + + def check_splits_into_the_three_sections(self): + """A complete probe yields exactly what the three separate commands produced.""" + qdisc, flt, iptables = NetworkGroupManager._split_probe_output( + _probe(QDISC_SECTION, FILTER_SECTION, IPTABLES_SECTION)) + + assert qdisc == QDISC_SECTION.splitlines() + assert flt == FILTER_SECTION.splitlines() + assert iptables == IPTABLES_SECTION.splitlines() + + def check_parses_an_active_partition(self): + """The deployed impairment and the fully cut peer both survive the round trip.""" + qdisc, flt, iptables = NetworkGroupManager._split_probe_output( + _probe(QDISC_SECTION, FILTER_SECTION, IPTABLES_SECTION)) + + assert NetworkGroupManager._parse_qdisc_constraints(qdisc) == "netem(delay: 100ms, loss: 5%)" + assert NetworkGroupManager._parse_filter_destinations(flt) == ["192.168.1.5"] + + drops = NetworkGroupManager._format_partition_drops( + NetworkGroupManager._parse_partition_drops(iptables)) + + assert drops == f" | partition: {CHAIN} <-X-> [192.168.1.5]" + + def check_parses_a_healed_link(self): + """After a heal the netem impairment is still reported, the drops are gone.""" + qdisc, _, iptables = NetworkGroupManager._split_probe_output( + _probe(QDISC_SECTION, FILTER_SECTION, "-P INPUT ACCEPT")) + + assert NetworkGroupManager._parse_qdisc_constraints(qdisc) == "netem(delay: 100ms, loss: 5%)" + assert NetworkGroupManager._format_partition_drops( + NetworkGroupManager._parse_partition_drops(iptables)) == "" + + def check_flags_a_half_applied_partition(self): + """A one-way drop is called out - the reason the iptables dump is worth keeping.""" + _, _, iptables = NetworkGroupManager._split_probe_output( + _probe(QDISC_SECTION, FILTER_SECTION, f"-A {CHAIN} -d 192.168.1.5/32 -j DROP")) + + drops = NetworkGroupManager._format_partition_drops( + NetworkGroupManager._parse_partition_drops(iptables)) + + assert drops == f" | partition: {CHAIN} out-X only [192.168.1.5]" + + @pytest.mark.parametrize(["probe_output", "unanswered"], [ + ("", [0, 1, 2]), + (QDISC_SECTION, [1, 2]), + (_probe(QDISC_SECTION, FILTER_SECTION), [2]), + (_probe("", "", ""), [0, 1, 2]) + ]) + def check_incomplete_probe_degrades(self, probe_output, unanswered): + """A node that answers partially yields a poorer overview, never an exception.""" + sections = NetworkGroupManager._split_probe_output(probe_output) + + assert len(sections) == 3, "Callers unpack three sections regardless of what came back" + + for idx in unanswered: + assert sections[idx] == [], f"Section {idx} went unanswered, it must come back empty" + + render, neutral = NEUTRAL_RENDERINGS[idx] + + assert render(sections[idx]) == neutral, f"Section {idx} must render as its neutral form" diff --git a/modules/ducktests/tests/checks/utils/check_cache_distribution.py b/modules/ducktests/tests/checks/utils/check_cache_distribution.py new file mode 100644 index 0000000000000..5eb580de60ca4 --- /dev/null +++ b/modules/ducktests/tests/checks/utils/check_cache_distribution.py @@ -0,0 +1,128 @@ +# 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 parsing of 'control.sh --cache distribution' output. + +The user attribute values in a row are emitted in the iteration order of the node's +attribute map on the control.sh side, which matches neither the --user-attributes +argument order nor alphabetical order. The header line is printed from the keys of +that same map, so these checks pin down that the names come from the header and not +from the requested list. +""" + +import pytest + +from ignitetest.services.utils.control_utility import ControlUtility + +HEADER = "[groupId,partition,nodeId,primary,state,updateCounter,partitionSize,nodeAddresses" + +GROUP = "[next group: id=-1368047377, name=mdc-cache]" + + +def _parse(output, user_attributes=None): + # pylint: disable=protected-access + return ControlUtility._ControlUtility__parse_cache_distribution(output, user_attributes) + + +def _copies(distribution, group="mdc-cache", partition=0): + return distribution.groups[group].partitions[partition] + + +class CheckCacheDistribution: + """ + Checks the distribution output parser. + """ + def check_no_user_attributes(self): + """Without --user-attributes the header has no trailing names and rows carry none.""" + distribution = _parse("\n".join([ + HEADER + "]", + GROUP, + "-1368047377,0,a1b2c3d4,P,OWNING,42,100,[127.0.0.1, 10.0.0.1]" + ])) + + copy = _copies(distribution)[0] + + assert copy.node_id == "a1b2c3d4" + assert copy.primary + assert copy.state == "OWNING" + assert copy.update_counter == 42 + assert copy.partition_size == 100 + assert copy.node_addresses == ["127.0.0.1", "10.0.0.1"] + assert copy.user_attributes == {} + + def check_single_user_attribute(self): + """The one-attribute case all current callers use.""" + distribution = _parse("\n".join([ + HEADER + ",IGNITE_DATA_CENTER_ID]", + GROUP, + "-1368047377,0,a1b2c3d4,P,OWNING,42,100,[127.0.0.1],DC1", + "-1368047377,0,e5f6a7b8,B,OWNING,42,100,[10.0.0.1],DC2" + ]), user_attributes=["IGNITE_DATA_CENTER_ID"]) + + primary, backup = _copies(distribution) + + assert primary.user_attributes == {"IGNITE_DATA_CENTER_ID": "DC1"} + assert backup.user_attributes == {"IGNITE_DATA_CENTER_ID": "DC2"} + + def check_attribute_names_follow_the_header_not_the_request(self): + """ + The regression the header-driven parse exists for: the map order on the control.sh + side is neither the requested order nor alphabetical, so zipping the values against + a sorted request list mis-assigns every value. + """ + distribution = _parse("\n".join([ + HEADER + ",ZONE,DC]", + GROUP, + "-1368047377,0,a1b2c3d4,P,OWNING,42,100,[127.0.0.1],east,DC1" + ]), user_attributes=["DC", "ZONE"]) + + assert _copies(distribution)[0].user_attributes == {"ZONE": "east", "DC": "DC1"} + + def check_missing_attribute_value_keeps_columns_aligned(self): + """A node lacking an attribute prints an empty field, it does not drop the column.""" + distribution = _parse("\n".join([ + HEADER + ",ZONE,DC]", + GROUP, + "-1368047377,0,a1b2c3d4,P,OWNING,42,100,[127.0.0.1],,DC1" + ]), user_attributes=["DC", "ZONE"]) + + assert _copies(distribution)[0].user_attributes == {"ZONE": "", "DC": "DC1"} + + def check_requested_attribute_absent_from_header_fails(self): + """A silently dropped --user-attributes argument must not parse as success.""" + with pytest.raises(AssertionError, match="missing from the distribution output"): + _parse("\n".join([ + HEADER + ",DC]", + GROUP, + "-1368047377,0,a1b2c3d4,P,OWNING,42,100,[127.0.0.1],DC1" + ]), user_attributes=["DC", "ZONE"]) + + def check_multiple_groups_and_partitions(self): + """Rows are filed under the group header that precedes them.""" + distribution = _parse("\n".join([ + HEADER + ",DC]", + GROUP, + "-1368047377,0,a1b2c3d4,P,OWNING,42,100,[127.0.0.1],DC1", + "-1368047377,1,e5f6a7b8,P,OWNING,7,50,[10.0.0.1],DC2", + "[next group: id=42, name=other-cache]", + "42,0,a1b2c3d4,P,MOVING,1,10,[127.0.0.1],DC1" + ]), user_attributes=["DC"]) + + assert sorted(distribution.groups) == ["mdc-cache", "other-cache"] + assert sorted(distribution.groups["mdc-cache"].partitions) == [0, 1] + + assert _copies(distribution, partition=1)[0].update_counter == 7 + assert _copies(distribution, "other-cache")[0].state == "MOVING" diff --git a/modules/ducktests/tests/docker/Dockerfile b/modules/ducktests/tests/docker/Dockerfile index 59ed67463d5de..db6f4222f0ec4 100644 --- a/modules/ducktests/tests/docker/Dockerfile +++ b/modules/ducktests/tests/docker/Dockerfile @@ -81,6 +81,7 @@ RUN apt-get update && apt-get install -y \ mc \ git \ build-essential \ + iproute2 \ && apt-get -y clean \ && rm -rf /var/lib/apt/lists/* diff --git a/modules/ducktests/tests/docker/requirements.txt b/modules/ducktests/tests/docker/requirements.txt index aba4f25a86d1d..6f96304e5b22c 100644 --- a/modules/ducktests/tests/docker/requirements.txt +++ b/modules/ducktests/tests/docker/requirements.txt @@ -16,3 +16,4 @@ filelock==3.8.2 ducktape==0.13.0 looseversion==1.3.0 +tcconfig==0.29.1 diff --git a/modules/ducktests/tests/docker/run_tests.sh b/modules/ducktests/tests/docker/run_tests.sh index d22d53e6f72bb..511c106701a29 100755 --- a/modules/ducktests/tests/docker/run_tests.sh +++ b/modules/ducktests/tests/docker/run_tests.sh @@ -127,6 +127,7 @@ while [[ $# -ge 1 ]]; do -t|--tc-paths) TC_PATHS="$2"; shift 2;; -n|--num-nodes) IGNITE_NUM_CONTAINERS="$2"; shift 2;; -j|--max-parallel) MAX_PARALLEL="$2"; shift 2;; + -r|--repeat) REPEAT="$2"; shift 2;; --subnet) SUBNET="--subnet $2"; shift 2;; --jdk) JDK_VERSION="$2"; shift 2;; --image) IMAGE_NAME="$2"; shift 2;; @@ -164,5 +165,9 @@ if [[ -n "$MAX_PARALLEL" ]]; then DUCKTAPE_OPTIONS="$DUCKTAPE_OPTIONS --max-parallel $MAX_PARALLEL" fi +if [[ -n "$REPEAT" ]]; then + DUCKTAPE_OPTIONS="$DUCKTAPE_OPTIONS --repeat $REPEAT" +fi + "$SCRIPT_DIR"/ducker-ignite test $TC_PATHS "$DUCKTAPE_OPTIONS" \ || die "ducker-ignite test failed" diff --git a/modules/ducktests/tests/ignitetest/services/ignite_app.py b/modules/ducktests/tests/ignitetest/services/ignite_app.py index 8ccd51e6f4e72..833cb2ed63034 100644 --- a/modules/ducktests/tests/ignitetest/services/ignite_app.py +++ b/modules/ducktests/tests/ignitetest/services/ignite_app.py @@ -34,7 +34,7 @@ class IgniteApplicationService(IgniteAwareService): APP_FINISH_EVT_MSG = "IGNITE_APPLICATION_FINISHED" APP_BROKEN_EVT_MSG = "IGNITE_APPLICATION_BROKEN" - def __init__(self, context, config, java_class_name, num_nodes=1, params="", startup_timeout_sec=60, + def __init__(self, context, config, java_class_name=None, num_nodes=1, params="", startup_timeout_sec=60, shutdown_timeout_sec=60, modules=None, main_java_class=SERVICE_JAVA_CLASS_NAME, jvm_opts=None, merge_with_default=True): super().__init__(context, config, num_nodes, startup_timeout_sec, shutdown_timeout_sec, main_java_class, @@ -43,27 +43,27 @@ def __init__(self, context, config, java_class_name, num_nodes=1, params="", sta self.java_class_name = java_class_name self.params = params - def await_started(self): - super().await_started() + def await_started(self, nodes=None): + super().await_started(nodes) - self.__check_status(self.APP_INIT_EVT_MSG, timeout=self.startup_timeout_sec) + self.__check_status(self.APP_INIT_EVT_MSG, timeout=self.startup_timeout_sec, nodes=nodes) def await_stopped(self): super().await_stopped() self.__check_status(self.APP_FINISH_EVT_MSG) - def __check_status(self, desired, timeout=1): - self.await_event("%s\\|%s" % (desired, self.APP_BROKEN_EVT_MSG), timeout, from_the_beginning=True) + def __check_status(self, desired, timeout=1, nodes=None): + self.await_event("%s\\|%s" % (desired, self.APP_BROKEN_EVT_MSG), timeout, nodes=nodes, from_the_beginning=True) try: - self.await_event(self.APP_BROKEN_EVT_MSG, 1, from_the_beginning=True) + self.await_event(self.APP_BROKEN_EVT_MSG, 1, nodes=nodes, from_the_beginning=True) raise IgniteExecutionException("Java application execution failed. %s" % self.extract_result("ERROR")) except TimeoutError: pass try: - self.await_event(desired, 1, from_the_beginning=True) + self.await_event(desired, 1, nodes=nodes, from_the_beginning=True) except Exception: raise Exception("Java application execution failed.") from None diff --git a/modules/ducktests/tests/ignitetest/services/mdc/__init__.py b/modules/ducktests/tests/ignitetest/services/mdc/__init__.py new file mode 100644 index 0000000000000..052a4803df6a2 --- /dev/null +++ b/modules/ducktests/tests/ignitetest/services/mdc/__init__.py @@ -0,0 +1,18 @@ +# 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. + +""" +Multi DC Cluster Service +""" diff --git a/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py b/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py new file mode 100644 index 0000000000000..d37b9c045fa1c --- /dev/null +++ b/modules/ducktests/tests/ignitetest/services/mdc/mdc_cluster.py @@ -0,0 +1,559 @@ +# 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" +ASSERTION_ERROR_PATTERN = "AssertionError" + +# Every log of a server node, including the ones rotated by a restart. +ALL_LOGS_GLOB = "ignite*.log*" + + +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, ASSERTION_ERROR_PATTERN): + for svc in self.servers.values(): + 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): + """ + Verifies that no active transactions are left on the cluster. + """ + txs = self.control(dc).tx() + + if isinstance(txs, list) and len(txs) > 0 and try_kill_hanging_tx: + for tx in txs: + self.control(dc).tx_kill(xid=tx.xid) + + txs = self.control(dc).tx() + + assert not isinstance(txs, list) or len(txs) == 0, f"No active transactions expected [txs={txs}]" + + @staticmethod + def result_int(svc: IgniteApplicationService, name: str) -> int: + """ + :return: Application-recorded integer result. + """ + return int(svc.extract_result(name)) + + @staticmethod + def result_float(svc: IgniteApplicationService, name: str) -> float: + """ + :return: Application-recorded float result. + """ + return float(svc.extract_result(name)) + + @staticmethod + def result_bool(svc: IgniteApplicationService, name: str) -> bool: + """ + :return: Application-recorded boolean result. + """ + val = svc.extract_result(name).strip().lower() + + return val == "true" + + +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 + enable/disable and the final network cleanup. + + :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) + + if not cfg.is_empty: + store.set_config(DC_1, DC_2, cfg) + + return NetworkGroupManager(logger, store, mdc.network_registry()) + + +def assert_cross_dc_distribution_by_attribute(distribution, dc_attr, expected_dcs, owning_only=True, + copies_per_dc=None): + """ + Asserts that every partition of every cache group has at least one copy in every DC, + using a node attribute (requested via --user-attributes) as the DC marker. + + :param distribution: CacheDistribution returned by ControlUtility.cache_distribution(), + requested with user_attributes=[dc_attr]. + :param dc_attr: Attribute name holding the DC id, e.g. "IGNITE_DATA_CENTER_ID". + :param expected_dcs: Collection of DC ids that must own a copy of every partition. + :param owning_only: Count only copies in OWNING state as present. + :param copies_per_dc: If set, each DC must hold exactly this many copies of every + partition - the MdcAffinityBackupFilter guarantee + ``(backups + 1) / dcsNum``. + """ + 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)}") + + +def _assert_cross_dc(distribution, expected_dcs, dc_of, owning_only, copies_per_dc, layout_hint): + 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} + + for copy in counted: + dc = dc_of(copy) + + if dc in per_dc: + per_dc[dc] += 1 + + missing = {dc for dc, cnt in per_dc.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} + + 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})" + for c in copies) + + problems = [] + + if missing: + problems.append(f"missing DCs={sorted(missing)}") + + if unbalanced: + problems.append(f"copies per DC != {copies_per_dc}: {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 diff --git a/modules/ducktests/tests/ignitetest/services/network_group/__init__.py b/modules/ducktests/tests/ignitetest/services/network_group/__init__.py new file mode 100644 index 0000000000000..867bd40a8353d --- /dev/null +++ b/modules/ducktests/tests/ignitetest/services/network_group/__init__.py @@ -0,0 +1,21 @@ +# 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. + +""" +Network Group Management Service + +This module provides tools for simulating complex network topologies by +defining traffic impairments between logical groups. +""" diff --git a/modules/ducktests/tests/ignitetest/services/network_group/configuration.py b/modules/ducktests/tests/ignitetest/services/network_group/configuration.py new file mode 100644 index 0000000000000..3fbf382b8c35f --- /dev/null +++ b/modules/ducktests/tests/ignitetest/services/network_group/configuration.py @@ -0,0 +1,64 @@ +# 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. + +from dataclasses import dataclass +from typing import Optional, Dict + + +@dataclass(frozen=True) +class CrossNetworkGroupConfiguration: + """ + Defines the network impairment profile between two network groups. + """ + delay: Optional[str] = None # tcset time expression, e.g. "100ms" + loss: Optional[float] = None # fraction in [0.0, 1.0], e.g. 0.1 (10%) + + def __post_init__(self): + if self.loss is not None and not 0.0 <= self.loss <= 1.0: + raise ValueError(f"loss must be within [0.0, 1.0], got {self.loss}") + + if self.delay is not None and not isinstance(self.delay, str): + raise TypeError(f"delay must be a tcset time expression string (e.g. '100ms'), got {self.delay!r}") + + @property + def is_empty(self) -> bool: + """True if the configuration defines no impairments.""" + return not self.delay and self.loss is None + + +class NetworkGroupStore: + """ + A registry for managing traffic impairments between different network groups. + """ + def __init__(self): + self.matrix: Dict[str, Dict[str, CrossNetworkGroupConfiguration]] = {} + + def set_config(self, group_a: str, group_b: str, impairment: CrossNetworkGroupConfiguration): + """ + Sets bidirectional rules between two network groups. + + Args: + group_a: The first network group identifier. + group_b: The second network group identifier. + impairment: The :class:`CrossNetworkGroupConfiguration` applied to all cross-group traffic directions. + """ + for src, dst in [(group_a, group_b), (group_b, group_a)]: + self.matrix.setdefault(src, {})[dst] = impairment + + def get_config(self, src_group: str, dst_group: str) -> Optional[CrossNetworkGroupConfiguration]: + """ + :return: :class:`CrossNetworkGroupConfiguration` for traffic from src to dst or None if not defined. + """ + return self.matrix.get(src_group, {}).get(dst_group) diff --git a/modules/ducktests/tests/ignitetest/services/network_group/manager.py b/modules/ducktests/tests/ignitetest/services/network_group/manager.py new file mode 100644 index 0000000000000..3f90b7f9450a1 --- /dev/null +++ b/modules/ducktests/tests/ignitetest/services/network_group/manager.py @@ -0,0 +1,476 @@ +# 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 + +# Separates the qdisc, filter and iptables sections in the output of the batched +# network probe issued by _log_network. +PROBE_SECTION_SEPARATOR = "=== ignitetest network probe section ===" + +# 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}") + + def disable_network_partition(self, group_a: str, group_b: str): + """ + Heals an active network partition between two groups by flushing the + pair's iptables DROP chain on every node - a single native call per + 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}]") + + 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)] + + self._ssh_parallel(tasks, tag="PARTITION_OFF") + + self._log_network(f"NET RESTORED {group_a} <-> {group_b}") + + def _resolve_group_ips(self, group: str) -> List[str]: + return [socket.gethostbyname(node.account.externally_routable_ip) + for node in self._iter_group_nodes(group)] + + def _apply_specs(self, specs: List[RuleSpec], tag: str): + """ + Compiles all rule specs into one batched command per source node and + executes them across nodes in parallel. + """ + node_cmds = self._collect_node_cmds(specs) + + # One SSH round-trip per node: all rules for a node take effect together. + tasks = [(node, " && ".join(cmds)) for node, cmds in node_cmds] + + self._ssh_parallel(tasks, tag=tag) + + def _collect_node_cmds(self, specs: List[RuleSpec]): + """ + Groups the tc commands produced by all specs by source node. + + :return: List of (node, [command, ...]) with insertion order preserved, + so that on deployment the first rule per node is an '--overwrite' and + every subsequent rule (across all destination groups) is an '--add'. + """ + per_node = {} + + for src_group, dst_group, action, cfg in specs: + dst_ips = [node.account.externally_routable_ip for node in self._iter_group_nodes(dst_group)] + + for src_node in self._iter_group_nodes(src_group): + interface = self._get_default_network_interface(src_node) + + _, cmds = per_node.setdefault(id(src_node), (src_node, [])) + + for dst_ip in dst_ips: + if action == ACTION_DEPLOY: + current_action = ACTION_OVERWRITE if not cmds else ACTION_ADD + else: + current_action = action + + cmd = to_tcset_cmd(interface=interface, dst_host_or_ip=dst_ip, config=cfg, + action=current_action) + + if cmd: + cmds.append(cmd) + elif action == ACTION_DEPLOY: + raise ValueError( + f"No network constraints defined from {src_node.account.hostname} to {dst_ip}." + ) + + return [entry for entry in per_node.values() if entry[1]] + + def _ssh_parallel(self, tasks: List[Tuple[object, str]], tag: str): + """ + Executes one command per node concurrently and propagates the first failure. + """ + if not tasks: + return + + started = monotonic() + + def run(task): + node, cmd = task + node.account.ssh(cmd) + + with ThreadPoolExecutor(max_workers=min(MAX_PARALLEL_SSH_SESSIONS, len(tasks))) as pool: + futures = [pool.submit(run, task) for task in tasks] + + for future in futures: + future.result() + + self.logger.debug(f"[{tag}] tc rules applied on {len(tasks)} node(s) in {monotonic() - started:.2f}s") + + def _ssh_output_parallel(self, tasks: List[Tuple[object, str]], tag: str) -> List[str]: + """ + Executes one command per node concurrently and returns the collected + outputs in task order. + """ + if not tasks: + return [] + + started = monotonic() + + def run(task): + node, cmd = task + return self._get_ssh_output(node, cmd) + + with ThreadPoolExecutor(max_workers=min(MAX_PARALLEL_SSH_SESSIONS, len(tasks))) as pool: + outputs = list(pool.map(run, tasks)) + + self.logger.debug(f"[{tag}] probed {len(tasks)} node(s) in {monotonic() - started:.2f}s") + + return outputs + + def _prefetch_network_interfaces(self): + """ + Warms up the per-node network interface cache in parallel, so that + command compilation later on requires no sequential SSH round-trips. + """ + nodes = list(self._iter_all_nodes()) + + if not nodes: + return + + with ThreadPoolExecutor(max_workers=min(MAX_PARALLEL_SSH_SESSIONS, len(nodes))) as pool: + list(pool.map(self._get_default_network_interface, nodes)) + + @staticmethod + def _bidirectional(group_a: str, group_b: str) -> List[Tuple[str, str]]: + return [(group_a, group_b), (group_b, group_a)] + + def _iter_group_nodes(self, group: str) -> Iterator: + for svc in self.network_group_registry[group]: + yield from svc.nodes + + def _iter_all_nodes(self) -> Iterator: + for group in self.network_group_registry: + yield from self._iter_group_nodes(group) + + def _log_network(self, log_tag: str): + """ + Logs a concise, structured overview of the active traffic control queuing + disciplines (qdiscs), routing filters and partition drops across all cluster nodes. + + Every node is probed with a single batched command and all nodes are probed + concurrently. This runs right after a partition is toggled, so it has to cost + one round-trip cluster-wide: probing node by node stretches a partition well + past the outage the test asked for and can push a short blip over the failure + detection timeout it is meant to stay under. + """ + self.logger.debug(f"Network State Overview: [START][{log_tag}]") + + entries = [(group, svc, node) + for group, services in self.network_group_registry.items() + for svc in services + for node in svc.nodes] + + probes = self._ssh_output_parallel([(node, self._to_network_probe_cmd(node)) for _, _, node in entries], + tag="PROBE") + + node_statuses = [] + + for (group, svc, node), probe in zip(entries, probes): + qdisc_lines, filter_lines, iptables_lines = self._split_probe_output(probe) + + dst_ips = self._parse_filter_destinations(filter_lines) + constraints = self._parse_qdisc_constraints(qdisc_lines) + + targets_str = f" -> to [{', '.join(dst_ips)}]" if dst_ips and constraints != "noqueue" else "" + node_ip = socket.gethostbyname(node.account.externally_routable_ip) + + partition_str = self._format_partition_drops( + self._parse_partition_drops(iptables_lines)) + + node_statuses.append(f"[{group:<4}] {svc.who_am_i(node):<45}[{node_ip}] : " + f"{constraints}{targets_str}{partition_str}") + + # The per-node SSH probes above flood the debug log with their own command output. + # Collect first, print contiguously after: the overview must stay readable as one block. + for node_status in node_statuses: + self.logger.debug(node_status) + + self.logger.debug(f"Network State Overview: [END][{log_tag}]") + + def _to_network_probe_cmd(self, node) -> str: + """ + Builds the single command that dumps everything the overview needs from a + node - the netem qdiscs, the u32 filters and the iptables rules - as three + PROBE_SECTION_SEPARATOR delimited sections. + + The sections are chained unconditionally: this is a debug dump, so a node + that cannot answer one of the probes degrades to an incomplete overview + instead of failing the test around it. + """ + interface = self._get_default_network_interface(node) + + return " ; ".join([ + f"sudo tc qdisc show dev {interface} 2>/dev/null", + f"echo '{PROBE_SECTION_SEPARATOR}'", + f"sudo tc filter show dev {interface} 2>/dev/null", + f"echo '{PROBE_SECTION_SEPARATOR}'", + "sudo iptables -S 2>/dev/null || true" + ]) + + @staticmethod + def _split_probe_output(probe_output: str) -> Tuple[List[str], List[str], List[str]]: + """ + Splits a combined probe output back into its qdisc, filter and iptables line + lists. Sections a node did not answer with come back empty, rendering as a + plain 'noqueue' with no partition suffix. + """ + sections = probe_output.split(PROBE_SECTION_SEPARATOR) + + sections += [""] * (3 - len(sections)) + + qdisc_lines, filter_lines, iptables_lines = (section.strip().splitlines() for section in sections[:3]) + + return qdisc_lines, filter_lines, iptables_lines + + @staticmethod + def _parse_filter_destinations(filter_lines: List[str]) -> List[str]: + """ + Parses raw 'tc filter' output lines to extract destination IPs. + Converts the internal u32 hexadecimal match filters back to human-readable strings. + """ + dst_ips = [] + + for line in filter_lines: + match = re.search(r"match\s+([0-9a-fA-F]{8})/ffffffff\s+at\s+16", line) + if match: + hex_ip = match.group(1) + try: + ip_bytes = struct.pack("!I", int(hex_ip, 16)) + dst_ips.append(socket.inet_ntoa(ip_bytes)) + except (ValueError, struct.error, OSError): + continue + + return dst_ips + + @staticmethod + def _parse_partition_drops(iptables_lines: List[str]) -> Dict[str, Dict[str, set]]: + """ + Parses 'iptables -S' output lines into per-partition-chain drop sets. + + :return: {chain_name: {'s': {inbound-dropped ips}, 'd': {outbound-dropped ips}}} + """ + pattern = re.compile( + rf"^-A\s+({re.escape(PARTITION_CHAIN_PREFIX)}\S+)\s+" + rf"-(s|d)\s+(\d{{1,3}}(?:\.\d{{1,3}}){{3}})(?:/32)?\s+-j\s+DROP$" + ) + + drops: Dict[str, Dict[str, set]] = {} + + for line in iptables_lines: + match = pattern.match(line.strip()) + + if match: + chain, direction, ip = match.groups() + + drops.setdefault(chain, {"s": set(), "d": set()})[direction].add(ip) + + return drops + + @staticmethod + def _format_partition_drops(drops: Dict[str, Dict[str, set]]) -> str: + """ + Renders parsed partition drops into a compact, human-readable suffix. + + Fully cut peers (both inbound and outbound DROP present) are shown as + '<-X->'. Peers with only a one-way drop are flagged explicitly as + 'in-X'/'out-X' — on a healthy partition these lists are empty, so any + occurrence pinpoints a node with partially applied rules. + """ + if not drops: + return "" + + def fmt(ips: set) -> str: + return ", ".join(sorted(ips, key=lambda ip: tuple(map(int, ip.split("."))))) + + chain_summaries = [] + + for chain in sorted(drops): + inbound, outbound = drops[chain]["s"], drops[chain]["d"] + + both, in_only, out_only = inbound & outbound, inbound - outbound, outbound - inbound + + details = [] + if both: + details.append(f"<-X-> [{fmt(both)}]") + if in_only: + details.append(f"in-X only [{fmt(in_only)}]") + if out_only: + details.append(f"out-X only [{fmt(out_only)}]") + + chain_summaries.append(f"{chain} {' '.join(details)}") + + return " | partition: " + "; ".join(chain_summaries) + + @staticmethod + def _parse_qdisc_constraints(qdisc_lines: List[str]) -> str: + """ + Parses raw 'tc qdisc' output lines to identify active traffic impairments. + Extracts active netem delay and loss parameters, ignoring verbose system handles. + """ + for line in qdisc_lines: + if "qdisc netem" in line: + delay_match = re.search(r"delay\s+(\d+\w+)", line) + loss_match = re.search(r"loss\s+(\d+%)", line) + + params = [] + if delay_match: + params.append(f"delay: {delay_match.group(1)}") + if loss_match: + params.append(f"loss: {loss_match.group(1)}") + + if params: + return f"netem({', '.join(params)})" + + return "noqueue" + + @memoize + def _get_default_network_interface(self, node): + return self._get_ssh_output(node, CMD_GET_NETWORK_INTERFACE) + + @staticmethod + def _get_ssh_output(node, cmd): + return node.account.ssh_output(cmd) \ + .decode(sys.getdefaultencoding()) \ + .strip() diff --git a/modules/ducktests/tests/ignitetest/services/network_group/tc_rule_args.py b/modules/ducktests/tests/ignitetest/services/network_group/tc_rule_args.py new file mode 100644 index 0000000000000..d5a8d7f0db753 --- /dev/null +++ b/modules/ducktests/tests/ignitetest/services/network_group/tc_rule_args.py @@ -0,0 +1,135 @@ +# 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 +from typing import Iterable, Optional + +from ignitetest.services.network_group.configuration import CrossNetworkGroupConfiguration + +# In non-interactive SSH sessions, 'sudo' enforces a strict 'secure_path' +# and ignores the user's environment. We explicitly pass common installation +# paths (global, virtualenv, and local user directories) via 'env PATH' so +# the system can locate the 'tcset' binary regardless of how tcconfig was installed. +SUDO_PREFIX = 'sudo env "PATH=$PATH:/home/ducker/.local/bin:/opt/venv/bin"' + +# tcset rule actions. +ACTION_OVERWRITE = "--overwrite" +ACTION_ADD = "--add" + +# -w: wait on the xtables lock instead of failing +IPTABLES = "sudo iptables -w" + +PARTITION_CHAIN_PREFIX = "NP_" + +# Extraction filter to find user-defined chains starting with our prefix +# - iptables -S outputs "-N CHAIN_NAME" for new chains +# - awk filters rows where column 1 is "-N" and column 2 matches our prefix +FIND_CUSTOM_CHAINS_AWK = f"awk '$1==\"-N\" && $2 ~ /^{PARTITION_CHAIN_PREFIX}/ {{print $2}}'" + +# iptables chain names are limited to 28 characters. +MAX_CHAIN_NAME_LEN = 28 + + +def to_tcset_cmd(interface: str, dst_host_or_ip: str, config: CrossNetworkGroupConfiguration, + action: str = ACTION_OVERWRITE) -> Optional[str]: + """ + Compiles the full 'tcset' command string. + Returns None if there are no network limitations configured. + """ + if config.is_empty: + return None + + dst_ip = socket.gethostbyname(dst_host_or_ip) + + args = [ + f"{SUDO_PREFIX} tcset {interface}", + f"--dst-network {dst_ip}/32", + action + ] + + if config.delay: + args.append(f"--delay {config.delay}") + + if config.loss is not None: + args.append(f"--loss {config.loss * 100:g}%") + + return " ".join(args) + + +def to_tcdel_all_cmd(interface: str) -> str: + """ + Compiles the absolute clear command for an interface. + """ + return f"{SUDO_PREFIX} tcdel {interface} --all" + + +def partition_chain_name(group_a: str, group_b: str) -> str: + """ + Deterministic, order-independent iptables chain name for a group pair. + """ + a, b = sorted((group_a, group_b)) + + raw = f"{PARTITION_CHAIN_PREFIX}{a}_{b}" + + return re.sub(r"[^A-Za-z0-9_]", "_", raw)[:MAX_CHAIN_NAME_LEN] + + +def to_partition_enable_cmd(chain: str, remote_ips: Iterable[str]) -> str: + """ + Compiles a single shell command that (idempotently) creates the partition + chain, hooks it into INPUT/OUTPUT, and drops all traffic to and from the + given remote IPs. + + Dropping both '-s' and '-d' on each side means the partition between a + node pair is effective as soon as *either* endpoint has applied its rules, + minimizing the window in which the partition is only half-visible. + """ + cmds = [ + f"{{ {IPTABLES} -N {chain} 2>/dev/null || true; }}", + f"{{ {IPTABLES} -C INPUT -j {chain} 2>/dev/null || {IPTABLES} -I INPUT 1 -j {chain}; }}", + f"{{ {IPTABLES} -C OUTPUT -j {chain} 2>/dev/null || {IPTABLES} -I OUTPUT 1 -j {chain}; }}", + ] + + for ip in remote_ips: + cmds.append(f"{IPTABLES} -A {chain} -s {ip}/32 -j DROP") + cmds.append(f"{IPTABLES} -A {chain} -d {ip}/32 -j DROP") + + return " && ".join(cmds) + + +def to_partition_disable_cmd(chain: str) -> str: + """ + Compiles the healing command: flushes the pair's DROP rules in one call. + The (now empty) chain and its INPUT/OUTPUT hooks are intentionally left in + place, so re-enabling the same partition later stays a pure append and the + final teardown remains the single owner of chain removal. + """ + return f"{IPTABLES} -F {chain} 2>/dev/null || true" + + +def to_partition_teardown_cmd() -> str: + """ + Compiles the full cleanup command: unhooks, flushes, and deletes + every partition chain on the node, restoring a pristine iptables state. + """ + find_chains_subshell = f"sudo iptables -S 2>/dev/null | {FIND_CUSTOM_CHAINS_AWK}" + + return ( + f"for c in $({find_chains_subshell}); do " + f"{IPTABLES} -D INPUT -j $c 2>/dev/null; " + f"{IPTABLES} -D OUTPUT -j $c 2>/dev/null; " + f"{IPTABLES} -F $c && {IPTABLES} -X $c; " + f"done; true" + ) diff --git a/modules/ducktests/tests/ignitetest/services/utils/cdc/kafka_to_ignite.py b/modules/ducktests/tests/ignitetest/services/utils/cdc/kafka_to_ignite.py index 1ffb501ecd1f5..e5cba2a0b834b 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/cdc/kafka_to_ignite.py +++ b/modules/ducktests/tests/ignitetest/services/utils/cdc/kafka_to_ignite.py @@ -121,7 +121,7 @@ def get_config(dst_cluster, client_type): version=dst_cluster.config.version ) - def await_started(self): + def await_started(self, nodes=None): """ Awaits kafka-to-ignite.sh started. """ diff --git a/modules/ducktests/tests/ignitetest/services/utils/control_utility.py b/modules/ducktests/tests/ignitetest/services/utils/control_utility.py index d95f97b0a1362..8ace03db5ae8d 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/control_utility.py +++ b/modules/ducktests/tests/ignitetest/services/utils/control_utility.py @@ -181,6 +181,105 @@ 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-?\d+), name=(?P[^\]]+)\]") + + # Column header, carrying the attribute names after nodeAddresses, e.g. + # [groupId,...,nodeAddresses,IGNITE_DATA_CENTER_ID]. + header_pattern = re.compile( + r"^\[groupId,partition,nodeId,primary,state,updateCounter,partitionSize,nodeAddresses" + r"(?P(?:,[^,\]]*)*)\]$") + + # Trailing attribute values appear after nodeAddresses, comma-separated, in the iteration + # order of the node's attribute map as deserialized in the control.sh JVM - which is + # neither the --user-attributes order nor alphabetical. CacheDistributionTaskResult#print() + # builds the header from the keys of that very map and each row from its values, so the + # header is the only reliable source of the names. Every requested attribute is always put + # (CacheDistributionTask), so a value missing on a node prints as an empty field rather + # than shifting the columns. + row_pattern = re.compile(r"(?P-?\d+)," + r"(?P\d+)," + r"(?P[0-9a-fA-F]+)," + r"(?P[PB])," + r"(?P[A-Z_]+)," + r"(?P\d+)," + r"(?P\d+)," + r"\[(?P[^\]]*)\]" + r"(?:,(?P.*))?$") + + groups = {} + cur_group = None + attr_names = [] + + for line in output.splitlines(): + line = line.strip() + + match = header_pattern.match(line) + if match: + attr_names = [name.strip() for name in match.group("attr_names").split(",") if name.strip()] + + assert not user_attributes or sorted(attr_names) == sorted(user_attributes), \ + f"Requested user attributes are missing from the distribution output " \ + f"[requested={sorted(user_attributes)}, reported={sorted(attr_names)}]" + + continue + + 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 attr_names and match.group("attr_values") is not None: + values = [v.strip() for v in match.group("attr_values").split(",")] + attrs = dict(zip(attr_names, values)) + + copy = PartitionCopy(node_id=match.group("node_id"), + primary=match.group("primary") == "P", + state=match.group("state"), + update_counter=int(match.group("update_counter")), + partition_size=int(match.group("partition_size")), + node_addresses=[a.strip() for a in match.group("node_addresses").split(",") if a], + user_attributes=attrs) + + cur_group.partitions.setdefault(int(match.group("partition")), []).append(copy) + + return CacheDistribution(groups=groups) + def check_consistency(self, args): """ Consistency check. @@ -390,6 +489,8 @@ def __parse_cluster_state(output): ",\\sS(tate|TATE)=(?P[^\\s,]+)" "(,\\sOrder=(?P\\d+))?") + coordinator_pattern = re.compile("\\(Coordinator: [^)]*Order=(?P\\d+)\\)") + match = state_pattern.search(output) state = match.group("cluster_state") if match else None @@ -404,7 +505,16 @@ def __parse_cluster_state(output): order=int(match.group("order")) if match.group("order") else None) baseline.append(node) - return ClusterState(state=state, topology_version=topology, baseline=baseline) + coordinator = None + + match = coordinator_pattern.search(output) + + if match: + order = int(match.group("order")) + + coordinator = next((node for node in baseline if node.order == order), None) + + return ClusterState(state=state, topology_version=topology, baseline=baseline, coordinator=coordinator) def __run(self, cmd, node=None): if node is None: @@ -481,6 +591,7 @@ class ClusterState(NamedTuple): state: str topology_version: int baseline: list + coordinator: BaselineNode = None class TxInfo(NamedTuple): @@ -520,6 +631,35 @@ class TxVerboseInfo(NamedTuple): states: list +class PartitionCopy(NamedTuple): + """ + Single copy (primary or backup) of a partition on a node. + """ + node_id: str + primary: bool + state: str + update_counter: int + partition_size: int + node_addresses: list + user_attributes: dict + + +class CacheGroupDistribution(NamedTuple): + """ + Distribution of a single cache group: partition id -> list of PartitionCopy. + """ + group_id: int + name: str + partitions: dict + + +class CacheDistribution(NamedTuple): + """ + Distribution info for all printed cache groups: group name -> CacheGroupDistribution. + """ + groups: dict + + class ControlUtilityError(RemoteCommandError): """ Error is raised when control utility failed. diff --git a/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py b/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py index 7d475844a2ed1..8bd758f339d04 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py +++ b/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py @@ -284,6 +284,28 @@ def await_event(self, evt_message, timeout_sec, nodes=None, from_the_beginning=F self.await_event_on_node(evt_message, node, timeout_sec, from_the_beginning=from_the_beginning, backoff_sec=backoff_sec, log_file=log_file) + def check_event_absent_on_node(self, evt_message, node, from_the_beginning=True, log_file=None): + """ + Verify that a specific event message is NOT present in a node's log file. + :param evt_message: Event message. + :param node: Ignite service node. + :param from_the_beginning: If True, search from the beginning of the log file + (this is usually what you want for an absence check). + :param log_file: Explicit log file. + """ + log = os.path.join(self.log_dir, log_file) if log_file else node.log_file + + with monitor_log(node, log, from_the_beginning) as monitor: + assert not monitor.found(evt_message), \ + "Event [%s] was unexpectedly found on '%s'" % (evt_message, node.name) + + def check_event_absent(self, evt_message, from_the_beginning=True, log_file=None): + """ + Verify that a specific event message is NOT present on any node of the service. + """ + for node in self.nodes: + self.check_event_absent_on_node(evt_message, node, from_the_beginning=from_the_beginning, log_file=log_file) + @staticmethod def event_time(evt_message, node): """ diff --git a/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/__init__.py b/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/__init__.py index 1fdec71e3b4cb..6c9ea5b03f686 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/__init__.py +++ b/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/__init__.py @@ -74,6 +74,7 @@ class IgniteConfiguration(NamedTuple): auto_activation_enabled: bool = None transaction_configuration: TransactionConfiguration = None sql_configuration: Bean = None + network_timeout: int = 5_000 def prepare_ssl(self, test_globals, shared_root): """ diff --git a/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/communication.py b/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/communication.py index 528c8f9bbca29..e00a261b58905 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/communication.py +++ b/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/communication.py @@ -53,7 +53,8 @@ def __init__(self, connections_per_node: int = None, use_paired_connections: bool = None, message_queue_limit: int = None, - unacknowledged_messages_buffer_size: int = None): + unacknowledged_messages_buffer_size: int = None, + connect_timeout: int = None): self.local_port = local_port self.local_port_range = local_port_range self.idle_connection_timeout: int = idle_connection_timeout @@ -63,6 +64,7 @@ def __init__(self, self.use_paired_connections: bool = use_paired_connections self.message_queue_limit: int = message_queue_limit self.unacknowledged_messages_buffer_size: int = unacknowledged_messages_buffer_size + self.connect_timeout: int = connect_timeout @property def class_name(self): diff --git a/modules/ducktests/tests/ignitetest/services/utils/ignite_spec.py b/modules/ducktests/tests/ignitetest/services/utils/ignite_spec.py index a99aa163026f4..1edad489ee363 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/ignite_spec.py +++ b/modules/ducktests/tests/ignitetest/services/utils/ignite_spec.py @@ -358,6 +358,8 @@ def __get_default_jvm_opts(self): ] def command(self, node): + assert self.service.java_class_name is not None, "Missing required 'java_class_name' to execute command." + args = [ str(self.service.config.service_type.name), self.service.java_class_name, diff --git a/modules/ducktests/tests/ignitetest/services/utils/log_utils.py b/modules/ducktests/tests/ignitetest/services/utils/log_utils.py index 09b1f9871e4d8..ba97e72c7ff7e 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/log_utils.py +++ b/modules/ducktests/tests/ignitetest/services/utils/log_utils.py @@ -22,6 +22,18 @@ from ducktape.cluster.remoteaccount import LogMonitor +class IgniteLogMonitor(LogMonitor): + """ + Extends ducktape's LogMonitor with a one-shot presence check. + """ + def found(self, pattern): + """ + Check once whether the pattern is present in the log after the initial + offset recorded when the monitor was created. + """ + return self.acct.ssh("tail -c +%d %s | grep '%s'" % (self.offset + 1, self.log, pattern), allow_fail=True) == 0 + + @contextmanager def monitor_log(node, log, from_the_beginning=False): """ @@ -38,4 +50,4 @@ def monitor_log(node, log, from_the_beginning=False): offset = 0 if from_the_beginning else int(node.account.ssh_output("wc -c %s" % log).split()[0]) except Exception: offset = 0 - yield LogMonitor(node.account, log, offset) + yield IgniteLogMonitor(node.account, log, offset) diff --git a/modules/ducktests/tests/ignitetest/services/utils/templates/ignite_configuration_macro.j2 b/modules/ducktests/tests/ignitetest/services/utils/templates/ignite_configuration_macro.j2 index fd1cbef5b60ec..c4953903afc38 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/templates/ignite_configuration_macro.j2 +++ b/modules/ducktests/tests/ignitetest/services/utils/templates/ignite_configuration_macro.j2 @@ -38,6 +38,7 @@ + diff --git a/modules/ducktests/tests/ignitetest/tests/control_utility/baseline_test.py b/modules/ducktests/tests/ignitetest/tests/control_utility/baseline_test.py index d9211786707ed..6bab99208caf7 100644 --- a/modules/ducktests/tests/ignitetest/tests/control_utility/baseline_test.py +++ b/modules/ducktests/tests/ignitetest/tests/control_utility/baseline_test.py @@ -63,8 +63,8 @@ def test_baseline_set(self, ignite_version): # Set baseline using topology version. new_node = self.__start_ignite_nodes(ignite_version, 1, join_cluster=servers) - _, version, _ = control_utility.cluster_state() - control_utility.set_baseline(version) + cluster_state = control_utility.cluster_state() + control_utility.set_baseline(cluster_state.topology_version) blt_size += 1 baseline = control_utility.baseline() @@ -125,15 +125,15 @@ def test_activate_deactivate(self, ignite_version): control_utility.activate() - state, _, _ = control_utility.cluster_state() + cluster_state = control_utility.cluster_state() - assert state.lower() == 'active', 'Unexpected state %s' % state + assert cluster_state.state.lower() == 'active', 'Unexpected state %s' % cluster_state.state control_utility.deactivate() - state, _, _ = control_utility.cluster_state() + cluster_state = control_utility.cluster_state() - assert state.lower() == 'inactive', 'Unexpected state %s' % state + assert cluster_state.state.lower() == 'inactive', 'Unexpected state %s' % cluster_state.state @cluster(num_nodes=NUM_NODES) @ignore_if(lambda version, globals: version < V_2_8_0) diff --git a/modules/ducktests/tests/ignitetest/tests/index_rebuild_test.py b/modules/ducktests/tests/ignitetest/tests/index_rebuild_test.py index 18b73446a4b80..d78ff64135a83 100644 --- a/modules/ducktests/tests/ignitetest/tests/index_rebuild_test.py +++ b/modules/ducktests/tests/ignitetest/tests/index_rebuild_test.py @@ -80,8 +80,8 @@ def test_index_bin_rebuild(self, ignite_version, backups, cache_count, entry_cou control_utility.disable_baseline_auto_adjust() - _, version, _ = control_utility.cluster_state() - control_utility.set_baseline(version) + cluster_state = control_utility.cluster_state() + control_utility.set_baseline(cluster_state.topology_version) preload_time = preload_data( self.test_context, diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/__init__.py b/modules/ducktests/tests/ignitetest/tests/mdc/__init__.py new file mode 100644 index 0000000000000..c3e67412c9c9f --- /dev/null +++ b/modules/ducktests/tests/ignitetest/tests/mdc/__init__.py @@ -0,0 +1,18 @@ +# 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. + +""" +Multi data center (MDC) tests. +""" diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/partition_resilience_test.py b/modules/ducktests/tests/ignitetest/tests/mdc/partition_resilience_test.py new file mode 100644 index 0000000000000..2ac3929805558 --- /dev/null +++ b/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 = 20 + +# A blip must stay well below the failure detection timeout (10s by default). +BLIP_SECS = 1 +BLIP_SETTLE_SECS = 5 +BLIPS = 10 + +BG_MAX_STALL_MS = 500 + + +class MdcPartitionResilienceTest(IgniteTest): + """ + Tests for cluster network partition resilience in MultiDC. + """ + @cluster(num_nodes=12) + @ignite_versions(str(DEV_BRANCH)) + @parametrize(cross_dc_latency_ms=100) + def test_mdc_cluster_partition_resilience(self, ignite_version, cross_dc_latency_ms): + """ + The canonical split-brain lifecycle: partition -> split into two healthy half-rings + (DC1 active, DC2 read-only) -> all data readable everywhere -> heal -> DC2 rejoins + via restart -> writes restored everywhere, distribution and consistency verified. + """ + mdc = MdcCluster(self, ignite_version, srv_per_dc=5, runners_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() + + mdc.generate_data(DC_1, CACHE_NAME, 0, 100, backups=BACKUPS) + mdc.generate_data(DC_2, CACHE_NAME, 100, 200, backups=BACKUPS) + + mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1) + + net.enable_network_partition(DC_1, DC_2) + + sleep(SPLIT_SETTLE_SECS) + + mdc.verify_split_brain() + + # All data written before the split is readable in both halves. + mdc.check_data(DC_1, CACHE_NAME, 0, 200) + mdc.check_data(DC_2, CACHE_NAME, 0, 200) + + # The half-ring holding the main DC accepts writes, the other one is read-only. + mdc.check_put_admissibility(DC_1, CACHE_NAME, True) + mdc.check_put_admissibility(DC_2, CACHE_NAME, False) + + net.disable_network_partition(DC_1, DC_2) + + # Split-brain does not self-heal: the read-only half rejoins via restart. + mdc.restart(DC_2) + + mdc.check_put_admissibility(DC_1, CACHE_NAME, True, key_offset=2_000_000) + mdc.check_put_admissibility(DC_2, CACHE_NAME, True, key_offset=3_000_000) + + mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1) + + mdc.control(DC_1).idle_verify(CACHE_NAME) + + mdc.verify_servers_log_clean() + + mdc.stop_servers() + + @cluster(num_nodes=6) + @ignite_versions(str(DEV_BRANCH)) + @parametrize(cross_dc_latency_ms=100) + def test_main_dc_loss_and_return(self, ignite_version, cross_dc_latency_ms): + """ + The inverse of the canonical scenario: the MAIN data center goes down entirely. + The surviving DC must stay readable but reject writes (the topology validator does + not see the main DC), and writes must resume in both DCs once the main DC returns. + No network impairments are involved - the main DC is stopped, not partitioned. + """ + mdc = MdcCluster(self, ignite_version, srv_per_dc=2, runners_per_dc=1) + + with cross_dc_network(self.logger, mdc, delay_ms=cross_dc_latency_ms): + mdc.start_servers() + + mdc.generate_data(DC_1, CACHE_NAME, 0, 100, backups=BACKUPS) + mdc.generate_data(DC_2, CACHE_NAME, 100, 200, backups=BACKUPS) + + mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1) + + mdc.servers[DC_1].stop() + + state = mdc.control(DC_2).cluster_state() + + assert "ACTIVE" == state.state, f"Surviving DC should remain ACTIVE [actual={state.state}]" + + # One copy of every partition lives in DC2, so all data stays readable... + mdc.check_data(DC_2, CACHE_NAME, 0, 200) + + # ...but the surviving DC is read-only while the main DC is invisible. + mdc.check_put_admissibility(DC_2, CACHE_NAME, False) + + # The shared ip finder memoized only DC1's (first started) addresses, so a + # restarted DC1 would seed off itself and form a separate cluster. Point + # discovery at both DCs so it rejoins through the surviving DC2. + mdc.sync_service_discovery() + + mdc.servers[DC_1].start(clean=False) + + mdc.servers[DC_1].await_rebalance() + + mdc.check_put_admissibility(DC_1, CACHE_NAME, True, key_offset=2_000_000) + mdc.check_put_admissibility(DC_2, CACHE_NAME, True, key_offset=3_000_000) + + mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1) + + mdc.control(DC_1).idle_verify(CACHE_NAME) + + mdc.verify_servers_log_clean() + + mdc.stop_servers() + + @cluster(num_nodes=10) + @ignite_versions(str(DEV_BRANCH)) + @parametrize(cross_dc_latency_ms=100) + def test_short_partition_blips_do_not_split(self, ignite_version, cross_dc_latency_ms): + """ + A flapping WAN link: several short (below the failure detection timeout) full + cross-DC connectivity drops. The cluster must NOT split: after the blips it is + still one ACTIVE cluster, all data is intact and both DCs accept writes. + """ + mdc = MdcCluster(self, ignite_version, srv_per_dc=3, 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() + + mdc.generate_data(DC_1, CACHE_NAME, 0, 100, backups=BACKUPS) + mdc.generate_data(DC_2, CACHE_NAME, 100, 200, backups=BACKUPS) + + for dc in (DC_1, DC_2): + mdc.start_loader(dc, {"mode": "GET", "cacheName": CACHE_NAME, + "keyFrom": 0, "keyTo": 200, + "continueOnError": True, "opPauseMs": 5, + "resultPrefix": f"bg{dc}"}) + + for i in range(BLIPS): + self.logger.info(f"Network blip {i + 1}/{BLIPS}") + + net.enable_network_partition(DC_1, DC_2) + + sleep(BLIP_SECS) + + net.disable_network_partition(DC_1, DC_2) + + sleep(BLIP_SETTLE_SECS) + + mdc.verify_whole_cluster_healthy() + + for dc in (DC_1, DC_2): + svc = mdc.stop_loader(dc) + + ops = mdc.result_int(svc, f"bg{dc}OpsCnt") + errs = mdc.result_int(svc, f"bg{dc}ErrCnt") + max_stall = mdc.result_int(svc, f"bg{dc}MaxStallMs") + + self.logger.info(f"Background GET load [dc={dc}, ops={ops}, errs={errs}, maxStallMs={max_stall}]") + + assert ops > 0, f"Background get load performed no operations [dc={dc}]" + assert errs == 0, \ + f"Background get load errors exceed the boundary tolerance [dc={dc}, ops={ops}, errs={errs}]" + assert max_stall < BG_MAX_STALL_MS, \ + f"Background get load stalled for too long [dc={dc}, maxStallMs={max_stall}]" + + mdc.check_data(DC_1, CACHE_NAME, 0, 200) + mdc.check_data(DC_2, CACHE_NAME, 0, 200) + + mdc.check_put_admissibility(DC_1, CACHE_NAME, True, key_offset=2_000_000) + mdc.check_put_admissibility(DC_2, CACHE_NAME, True, key_offset=3_000_000) + + mdc.control(DC_1).idle_verify(CACHE_NAME) + + mdc.verify_servers_log_clean() + + mdc.stop_servers() diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/thin_client_test.py b/modules/ducktests/tests/ignitetest/tests/mdc/thin_client_test.py new file mode 100644 index 0000000000000..1fce04e526c53 --- /dev/null +++ b/modules/ducktests/tests/ignitetest/tests/mdc/thin_client_test.py @@ -0,0 +1,156 @@ +# 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. This is asserted +from both DCs, so no routing that fixes on a single DC can satisfy it. + +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): + """ + Each 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) + + for dc, cli in [(DC_1, cli_dc1), (DC_2, cli_dc2)]: + svc = mdc.run_service(cli, {"mode": "GET", "cacheName": CACHE_NAME, "keyFrom": 0, + "keyTo": KEYS, "iterations": PINNED_GET_ITERS, + "resultPrefix": f"pinnedGet{dc}"}) + + avg_ms = mdc.result_float(svc, f"pinnedGet{dc}AvgOpMs") + err_cnt = mdc.result_int(svc, f"pinnedGet{dc}ErrCnt") + + self.logger.info(f"Thin client routing latency [dc={dc}, delayMs={cross_dc_latency_ms}, " + f"getAvgOpMs={avg_ms}, getErrCnt={err_cnt}]") + + assert avg_ms < cross_dc_latency_ms, \ + f"DC-pinned thin client reads should be served locally " \ + f"[dc={dc}, avgMs={avg_ms}, delayMs={cross_dc_latency_ms}]" + + assert err_cnt == 0, \ + f"Expected 0 errors for DC-pinned thin client reads [dc={dc}, errCnt={err_cnt}]" + + net.enable_network_partition(DC_1, DC_2) + + sleep(SPLIT_SETTLE_SECS) + + mdc.verify_split_brain() + + # The client pinned to the main half keeps writing... + mdc.run_service(cli_dc1, {"mode": "PUT", "cacheName": CACHE_NAME, "keyFrom": OFFSET_DURING, + "keyTo": OFFSET_DURING + PUT_ITERS, "iterations": PUT_ITERS, + "resultPrefix": "duringPut"}) + + # ...while the client pinned to the read-only half still reads cleanly + # (its DC1 addresses are unreachable, so it falls back to DC2 nodes)... + svc = mdc.run_service(cli_dc2, {"mode": "GET", "cacheName": CACHE_NAME, "keyFrom": 0, + "keyTo": KEYS, "iterations": PINNED_GET_ITERS, + "resultPrefix": "roGet"}) + + assert mdc.result_int(svc, "roGetErrCnt") == 0, \ + "Thin client reads in the read-only DC must be clean" + + # ...and has every write rejected by the topology validator. + mdc.run_service(cli_dc2, {"mode": "PUT", "cacheName": CACHE_NAME, + "keyFrom": OFFSET_REJECTED_PROBES, + "keyTo": OFFSET_REJECTED_PROBES + PUT_ITERS, + "iterations": PUT_ITERS, "inadmissible": True, + "resultPrefix": "roPut"}) + + net.disable_network_partition(DC_1, DC_2) + + mdc.restart(DC_2) + + mdc.run_service(cli_dc2, {"mode": "PUT", "cacheName": CACHE_NAME, "keyFrom": OFFSET_AFTER, + "keyTo": OFFSET_AFTER + PUT_ITERS, "iterations": PUT_ITERS, + "resultPrefix": "afterPut"}) + + mdc.check_data(DC_1, CACHE_NAME, OFFSET_DURING, OFFSET_DURING + PUT_ITERS) + mdc.check_data(DC_1, CACHE_NAME, OFFSET_AFTER, OFFSET_AFTER + PUT_ITERS) + + mdc.stop_servers() + + def _thin_client(self, mdc: MdcCluster, jvm_opts) -> IgniteApplicationService: + """ + Builds a single-node thin client application service with the addresses of all + server nodes of both DCs. + """ + return IgniteApplicationService( + self.test_context, + IgniteThinClientConfiguration(addresses=mdc.thin_client_addresses(), + version=mdc.ignite_config.version), + java_class_name=THIN_LOAD_APP, + num_nodes=1, + jvm_opts=jvm_opts) diff --git a/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py b/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_test.py new file mode 100644 index 0000000000000..4b1d2e3555cfe --- /dev/null +++ b/modules/ducktests/tests/ignitetest/tests/mdc/transactional_partition_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. + +""" +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 +explicit-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 backup +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 explicit 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_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=7) + @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 explicit-transaction insert load from the backup 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={DC_2: 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 explicit-transaction insert load from the backup DC. + # It stops on the very first exception (stopOnError) instead of failing the app, + # recording how many inserts had succeeded up to that point. + mdc.start_loader(DC_2, { + "mode": "TX_PUT", + "cacheName": CACHE_NAME, + "keyFrom": LOAD_KEY_FROM_DC_2, + "keyTo": LOAD_KEY_TO_DC_2, + "stopOnError": True, + "resultPrefix": f"txLoad{DC_2}", + "createCache": True, + "backups": BACKUPS, + "atomicity": "TRANSACTIONAL", + "mainDc": DC_1, + "txTimeout": 5_000 + }) + + 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. + svc = mdc.stop_loader(DC_2) + + inserts = mdc.result_int(svc, f"txLoad{DC_2}OpsCnt") + errors = mdc.result_int(svc, f"txLoad{DC_2}ErrCnt") + stop_on_error = mdc.result_bool(svc, f"txLoad{DC_2}StoppedOnError") + + self.logger.info(f"Transactional load cut by the partition [txLoad{DC_2}OpsCnt={inserts}, " + f"txLoad{DC_2}ErrCnt={errors}, txLoad{DC_2}StoppedOnError={stop_on_error}]") + + assert inserts > 0, "The transactional load performed no successful inserts" + assert errors > 0, "The transactional load should have failed on network partition" + assert stop_on_error, f"The load was expected to be cut off by the partition [dc={DC_2}]" + + # The point of the scenario: the aborted explicit transactions leave nothing + # hanging on either half-ring... + mdc.verify_no_hanging_txs(DC_1, try_kill_hanging_tx=True) + mdc.verify_no_hanging_txs(DC_2, try_kill_hanging_tx=True) + + # ...and neither half-ring logged a hung PME, a long running transaction or a + # lost partition. + mdc.verify_servers_log_clean() + + load_key_to = LOAD_KEY_TO_DC_2 if inserts > LOAD_KEY_TO_DC_2 - LOAD_KEY_FROM_DC_2 else inserts + + # All data written before the split is readable in both halves. + mdc.check_data(DC_1, CACHE_NAME, LOAD_KEY_FROM_DC_2, load_key_to) + mdc.check_data(DC_2, CACHE_NAME, LOAD_KEY_FROM_DC_2, load_key_to) + + # The half-ring holding the main DC accepts writes, the other one is read-only. + mdc.check_put_admissibility(DC_1, CACHE_NAME, True, key_offset=40_000_000) + mdc.check_put_admissibility(DC_2, CACHE_NAME, False, key_offset=41_000_000) + + net.disable_network_partition(DC_1, DC_2) + + mdc.restart(DC_2) + + mdc.check_put_admissibility(DC_1, CACHE_NAME, True, key_offset=100_000_000) + mdc.check_put_admissibility(DC_2, CACHE_NAME, True, key_offset=101_000_000) + + mdc.verify_cache_distribution(CACHE_NAME, copies_per_dc=1) + + mdc.control(DC_1).idle_verify(CACHE_NAME) + + mdc.verify_servers_log_clean() + + mdc.stop_servers()