This is an automated email from the ASF dual-hosted git repository.
sergey-chugunov-1985 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new b6779b7805d IGNITE-28692 Add cache partition distribution metrics for
caches in MDC environments (#13238)
b6779b7805d is described below
commit b6779b7805d10d514b74dae8b48cf2c571946c69
Author: Sergey Chugunov <[email protected]>
AuthorDate: Mon Jul 6 12:40:22 2026 +0400
IGNITE-28692 Add cache partition distribution metrics for caches in MDC
environments (#13238)
---
.../processors/cache/CacheMetricsImpl.java | 92 ++++
.../processors/cache/GridCacheProcessor.java | 16 +
.../cache/ValidationOnNodeJoinUtils.java | 34 ++
.../processors/cluster/BaselineTopology.java | 38 +-
.../MdcAffinityBackupFilterSelfTest.java | 2 +
.../processors/cache/MdcCacheMetricsTest.java | 600 +++++++++++++++++++++
.../ignite/testsuites/IgniteCacheTestSuite2.java | 2 +
7 files changed, 773 insertions(+), 11 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsImpl.java
index a67d265dbf7..b483dd64904 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsImpl.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsImpl.java
@@ -19,16 +19,20 @@ package org.apache.ignite.internal.processors.cache;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.CacheMetrics;
+import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.affinity.AffinityAssignment;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import
org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import
org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import
org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import
org.apache.ignite.internal.processors.cache.store.GridCacheWriteBehindStore;
+import org.apache.ignite.internal.processors.cluster.BaselineTopology;
import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric;
import org.apache.ignite.internal.processors.metric.impl.HistogramMetricImpl;
@@ -246,6 +250,15 @@ public class CacheMetricsImpl implements CacheMetrics {
/** Conflict resolver merged entries count. */
private LongAdderMetric rslvrMergedCnt;
+ /** Affinity configuration MDC safe metric. */
+ private volatile Boolean affCfgMdcSafe;
+
+ /** Current partition distribution MDC safe metric. */
+ private volatile Boolean mdcSafePartDistrib;
+
+ /** Last affinity version when MdcSafe Partition Distribution metric was
recalculated. */
+ private final AtomicReference<AffinityTopologyVersion>
mdcSafeMetricLastVer = new AtomicReference<>();
+
/**
* Creates cache metrics.
*
@@ -1679,6 +1692,71 @@ public class CacheMetricsImpl implements CacheMetrics {
return fut != null && !fut.isDone();
}
+ /**
+ * Returns {@code true} if affinity configuration is aware of multiple
data centers, and
+ * it is able to spread partitions' copies across data centers.
+ */
+ private boolean isAffinityCfgMdcSafe() {
+ return affCfgMdcSafe != null && affCfgMdcSafe;
+ }
+
+ /**
+ * Returns {@code true} if current cache partition distribution maintains
guarantee 'at least one partition copy in each datacenter'.
+ */
+ private boolean isMdcSafePartitionDistribution() {
+ return updateMdcSafeDistributionMetricIfNeeded();
+ }
+
+ /**
+ * Updates calculated value of Partition Distribution MDC safe metric if
affinity has changed
+ * since last affinity change.
+ *
+ * @return Updated value of Partition Distribution MDC safe metric or
cached value if affinity has not changed.
+ */
+ private boolean updateMdcSafeDistributionMetricIfNeeded() {
+ GridCacheAffinityManager aff = cctx.affinity();
+
+ try {
+ AffinityTopologyVersion currVer = aff.affinityTopologyVersion();
+ AffinityTopologyVersion lastVer = mdcSafeMetricLastVer.get();
+
+ // Skip expensive recalculation if the aff topology has not
changed.
+ if (!currVer.equals(lastVer)) {
+ if (mdcSafeMetricLastVer.compareAndSet(lastVer, currVer))
+ mdcSafePartDistrib =
recalculateMdcSafeMetric(aff.assignment(currVer));
+ }
+ }
+ catch (Exception ignored) {
+ // Affinity manager could throw an exception if aff not found for
the cache.
+ // This should not break any code calling this method, so we just
ignore it.
+ }
+
+ return mdcSafePartDistrib == null || mdcSafePartDistrib;
+ }
+
+ /**
+ * Recalcalculates Partition Distribution MDC safe metric based on
provided assignment.
+ *
+ * @param assignment New assignment for the cache.
+ */
+ private Boolean recalculateMdcSafeMetric(AffinityAssignment assignment) {
+ BaselineTopology top =
cctx.discovery().discoCache().state().baselineTopology();
+ if (top != null && top.numberOfDatacenters() > 1) {
+ int numberOfDataCenters = top.numberOfDatacenters();
+
+ for (List<ClusterNode> nodes : assignment.assignment()) {
+ int dcsCnt =
(int)nodes.stream().map(ClusterNode::dataCenterId).distinct().count();
+
+ if (dcsCnt < numberOfDataCenters)
+ return false;
+ }
+
+ return true;
+ }
+
+ return mdcSafePartDistrib;
+ }
+
/** {@inheritDoc} */
@Override public long getIndexRebuildKeysProcessed() {
return idxRebuildKeyProcessed.value();
@@ -1745,6 +1823,20 @@ public class CacheMetricsImpl implements CacheMetrics {
"Conflict resolver merged entries count");
}
+ /** Registers metric for partition distribution. */
+ public void registerPartitionDistributionSafeMetric() {
+ mreg.register("IsCachePartitionDistributionSafe",
this::isMdcSafePartitionDistribution,
+ "True if current cache partition distribution maintains guarantee
'one partition copy in each datacenter'.");
+ }
+
+ /** Registers metric for cache configuration related to distributing
partitions across DCs. */
+ public void registerAffinityConfigurationSafeMetric(boolean affCfgMdcSafe)
{
+ mreg.register("IsCacheAffinityConfigurationMdcSafe",
this::isAffinityCfgMdcSafe,
+ "True if cache affinity guarantees having a copy of each partition
in each data center.");
+
+ this.affCfgMdcSafe = affCfgMdcSafe;
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(CacheMetricsImpl.class, this);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index a82a8a9179c..d38fc3db417 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -217,6 +217,7 @@ import static
org.apache.ignite.internal.GridComponent.DiscoveryDataExchangeType
import static org.apache.ignite.internal.IgniteComponentType.JTA;
import static
org.apache.ignite.internal.processors.cache.GridCacheUtils.isNearEnabled;
import static
org.apache.ignite.internal.processors.cache.GridCacheUtils.isPersistentCache;
+import static
org.apache.ignite.internal.processors.cache.ValidationOnNodeJoinUtils.isAffinityConfigurationMdcSafe;
import static
org.apache.ignite.internal.processors.cache.ValidationOnNodeJoinUtils.validateHashIdResolvers;
import static
org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition.DFLT_CACHE_REMOVE_ENTRIES_TTL;
import static
org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName;
@@ -1136,6 +1137,8 @@ public class GridCacheProcessor extends
GridProcessorAdapter {
cache.onKernalStart();
+ registerMdcMetricsIfNeeded(cache);
+
if (ctx.events().isRecordable(EventType.EVT_CACHE_STARTED))
ctx.events().addEvent(EventType.EVT_CACHE_STARTED);
@@ -1144,6 +1147,19 @@ public class GridCacheProcessor extends
GridProcessorAdapter {
cache.configuration().getCacheMode() + ']');
}
+ /** */
+ private void registerMdcMetricsIfNeeded(GridCacheAdapter<?, ?> cache) {
+ if (ctx.clientNode())
+ return;
+
+ if (ctx.discovery().localNode() == null ||
ctx.discovery().localNode().dataCenterId() == null)
+ return;
+
+
cache.metrics0().registerAffinityConfigurationSafeMetric(isAffinityConfigurationMdcSafe(cache.configuration()));
+
+ cache.metrics0().registerPartitionDistributionSafeMetric();
+ }
+
/**
* @param cache Cache to stop.
* @param cancel Cancel flag.
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java
index ac8d97b2a12..16ea96c3cf4 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java
@@ -35,6 +35,10 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cache.CacheRebalanceMode;
import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.cache.affinity.AffinityFunction;
+import
org.apache.ignite.cache.affinity.rendezvous.ClusterNodeAttributeAffinityBackupFilter;
+import
org.apache.ignite.cache.affinity.rendezvous.ClusterNodeAttributeColocatedBackupFilter;
+import org.apache.ignite.cache.affinity.rendezvous.MdcAffinityBackupFilter;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.cluster.ClusterNode;
@@ -56,6 +60,7 @@ import org.apache.ignite.internal.thread.context.Scope;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.plugin.security.SecurityException;
import org.apache.ignite.spi.IgniteNodeValidationResult;
@@ -70,6 +75,7 @@ import static org.apache.ignite.cache.CacheMode.REPLICATED;
import static org.apache.ignite.cache.CacheRebalanceMode.SYNC;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_ASYNC;
import static
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_CONSISTENCY_CHECK_SKIPPED;
+import static
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DATA_CENTER_ID;
import static
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_TX_AWARE_QUERIES_ENABLED;
import static
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_TX_SERIALIZABLE_ENABLED;
import static
org.apache.ignite.internal.processors.cache.GridCacheUtils.isDefaultDataRegionPersistent;
@@ -631,6 +637,34 @@ public class ValidationOnNodeJoinUtils {
return null;
}
+ /**
+ * Analyzes affinity settings of a provided {@link CacheConfiguration} to
inspect if it provides guarantees
+ * that partitions of the cache will be spread across all datacenters
presented in cluster.
+ *
+ * @return {@code true} if affinity settings guarantee spreading
partitions across all datacenters and {@code false} otherwise.
+ */
+ static boolean isAffinityConfigurationMdcSafe(CacheConfiguration cc) {
+ if (cc.getCacheMode() == REPLICATED)
+ return true;
+
+ AffinityFunction affFunc = cc.getAffinity();
+
+ if (affFunc instanceof RendezvousAffinityFunction) {
+ IgniteBiPredicate<ClusterNode, List<ClusterNode>> filter =
((RendezvousAffinityFunction)affFunc).getAffinityBackupFilter();
+
+ if (filter instanceof ClusterNodeAttributeAffinityBackupFilter
attrFilter) {
+ if
(!F.asList(attrFilter.getAttributeNames()).contains(ATTR_DATA_CENTER_ID))
+ return false;
+ }
+
+ return filter instanceof MdcAffinityBackupFilter
+ || filter instanceof
ClusterNodeAttributeColocatedBackupFilter
+ || filter instanceof
ClusterNodeAttributeAffinityBackupFilter;
+ }
+
+ return true;
+ }
+
/**
* @param rmtNode Remote node to check.
* @param ctx Context.
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/BaselineTopology.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/BaselineTopology.java
index 1f71cb192a2..0a573ff3cc9 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/BaselineTopology.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/BaselineTopology.java
@@ -39,10 +39,12 @@ import org.apache.ignite.lang.IgnitePredicate;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import static
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DATA_CENTER_ID;
+
/**
* BaselineTopology represents a set of "database nodes" - nodes responsible
for holding persistent-enabled caches
* and persisting their information to durable storage.
- *
+ * <p/>
* Two major features BaselineTopology allows are:
* <ol>
* <li>Protection from conflicting updates.</li>
@@ -61,10 +63,10 @@ import org.jetbrains.annotations.Nullable;
* [A,B] [C]
* | |
* (2)updates to both parts
- *
+ * <p/>
* After independent updates applied to both parts of cluster at point(2)
node C should not be allowed to join
* [A,B] part.
- *
+ * <p/>
* The following algorithm makes sure node C will never join [A,B] part
back:
* <ol>
* <li>
@@ -95,10 +97,10 @@ import org.jetbrains.annotations.Nullable;
* <li>
* When BaselineTopology is set (e.g. on first activation) or
recreated (e.g. with set-baseline command)
* its ID on all nodes is incremented by one.
- *
+ * <p/>
* So when cluster receives a join request with BaselineTopology it
firstly compares joining node BlT ID with
* local BlT ID.
- *
+ * <p/>
* If joining node has a BaselineTopology with ID greater than one in
cluster it means that BlT was changed
* more times there; therefore new node is not allowed to join the
cluster.
* </li>
@@ -107,27 +109,27 @@ import org.jetbrains.annotations.Nullable;
* Instead current set of online nodes from BaselineTopology is used
to update {@link BaselineTopology#branchingPntHash}
* property of current BaselineTopology.
* Old value of the property is moved to {@link
BaselineTopology#branchingHist} list.
- *
+ * <p/>
* If joining node and local BlT IDs are the same then cluster takes
<b>branchingPntHash</b> of joining node
* and verifies that its local <b>branchingHist</b> contains that hash.
- *
+ * <p/>
* If joining node hash is not presented in cluster branching history
list
* it means that joining node was activated independently of currently
running cluster;
* therefore new node is not allowed to join the cluster.
- *
+ * <p/>
* If joining node hash is presented in the history, that it is safe
to let the node join the cluster.
* </li>
* <li>
* When BaselineTopology is recreated (e.g. with set-baseline command)
previous BaselineTopology is moved
* to BaselineHistory (consult source code of {@link
GridClusterStateProcessor} for more details).
- *
+ * <p/>
* If cluster sees that joining node BlT ID is less than cluster BlT
ID it looks up for BaselineHistory item
* for new node ID.
* Having this BaselineHistory item cluster verifies that branching
history of the item contains
* branching point hash of joining node
* (similar check as in the case above with only difference that
joining node BlT is compared against
* BaselineHistory item instead of BaselineTopology).
- *
+ * <p/>
* If new node branching point hash is found in the history than node
is allowed to join;
* otherwise it is rejected.
* </li>
@@ -172,7 +174,7 @@ public class BaselineTopology implements Serializable {
private final List<Long> branchingHist;
/**
- * @param nodeMap Map of node consistent ID to it's attributes.
+ * @param nodeMap Map of node consistent ID to its attributes.
*/
private BaselineTopology(Map<Object, Map<String, Object>> nodeMap, int id)
{
this.id = id;
@@ -274,6 +276,20 @@ public class BaselineTopology implements Serializable {
return nodeMap.get(consId);
}
+ /**
+ * Calculates number of datacenters presented in current baseline.
+ *
+ * @return Number of datacenters presented in the baseline or {@code -1}
if unknown.
+ */
+ public int numberOfDatacenters() {
+ Collection<Map<String, Object>> allNodesAttrs = nodeMap.values();
+
+ if (!allNodesAttrs.isEmpty() &&
allNodesAttrs.iterator().next().get(ATTR_DATA_CENTER_ID) != null)
+ return (int)allNodesAttrs.stream().map(m ->
m.get(ATTR_DATA_CENTER_ID)).distinct().count();
+
+ return -1;
+ }
+
/**
*
*/
diff --git
a/modules/core/src/test/java/org/apache/ignite/cache/affinity/rendezvous/MdcAffinityBackupFilterSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/cache/affinity/rendezvous/MdcAffinityBackupFilterSelfTest.java
index a45f3ccd901..97c19494584 100644
---
a/modules/core/src/test/java/org/apache/ignite/cache/affinity/rendezvous/MdcAffinityBackupFilterSelfTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/cache/affinity/rendezvous/MdcAffinityBackupFilterSelfTest.java
@@ -314,6 +314,8 @@ public class MdcAffinityBackupFilterSelfTest extends
GridCommonAbstractTest {
lastNode = startGrid(nodeIdx++);
}
+ System.clearProperty(IgniteSystemProperties.IGNITE_DATA_CENTER_ID);
+
return lastNode;
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/MdcCacheMetricsTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/MdcCacheMetricsTest.java
new file mode 100644
index 00000000000..8f390775c52
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/MdcCacheMetricsTest.java
@@ -0,0 +1,600 @@
+/*
+ * 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.processors.cache;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.ignite.IgniteSystemProperties;
+import
org.apache.ignite.cache.affinity.rendezvous.ClusterNodeAttributeAffinityBackupFilter;
+import
org.apache.ignite.cache.affinity.rendezvous.ClusterNodeAttributeColocatedBackupFilter;
+import org.apache.ignite.cache.affinity.rendezvous.MdcAffinityBackupFilter;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.cluster.ClusterState;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.lang.IgniteBiPredicate;
+import org.apache.ignite.spi.metric.BooleanMetric;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DATA_CENTER_ID;
+import static
org.apache.ignite.internal.processors.metric.impl.MetricUtils.cacheMetricsRegistryName;
+
+/**
+ * Test for new cache metrics for highlighting two data safety issues in Multi
DataCenter environments:
+ * 1. If cache configuration doesn't specify an affinity backup filter that
could guarantee presence of data copy in each DC.
+ * 2. If cluster topology changed in such a way that partition copies are not
spread across all available DCs.
+ */
+public class MdcCacheMetricsTest extends GridCommonAbstractTest {
+ /** */
+ private static final int NODES_NUMBER = 5;
+
+ /** */
+ private static final String CACHE_WITH_MDC_FILTER = "mdcSafeCache0";
+
+ /** */
+ private static final String CACHE_WITH_COLOCATED_FILTER = "mdcSafeCache1";
+
+ /** */
+ private static final String CACHE_WITH_MDC_SAFE_ATTRIBUTE_FILTER =
"mdcSafeCache2";
+
+ /** */
+ private static final String MDC_UNSAFE_CACHE = "mdcUnsafeCache0";
+
+ /** */
+ private static final String CACHE_WITH_MDC_UNSAFE_ATTRIBUTE_FILTER =
"mdcUnsafeCache1";
+
+ /** */
+ private static final String STRETCHED_CELL_ATTR_NAME = "DC_CELL_ATTR";
+
+ /** */
+ private static final String ATTR_FOR_UNSAFE_ATTR_FILTER =
"MDC_UNAWARE_ATTR";
+
+ /** */
+ private static final String[] STRETCHED_CELL_IDS = {"CELL_0", "CELL_1"};
+
+ /** */
+ private static final String DC_ID_0 = "DC_0";
+
+ /** */
+ private static final String DC_ID_1 = "DC_1";
+
+ /** */
+ private static final String AFFINITY_CFG_MDC_SAFE_METRIC_NAME =
"IsCacheAffinityConfigurationMdcSafe";
+
+ /** */
+ private static final String PARTITION_DISTRIBUTION_SAFE_METRIC_NAME =
"IsCachePartitionDistributionSafe";
+
+ /** */
+ private String dcId;
+
+ /** */
+ private String cellId;
+
+ /** */
+ private boolean useStaticCaches;
+
+ /** */
+ private boolean persistenceEnabled;
+
+ /** */
+ private final Set<String> allCaches = new HashSet<>();
+
+ /** */
+ private final Set<String> mdcSafeCaches = new HashSet<>();
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ stopAllGrids();
+
+ allCaches.clear();
+ mdcSafeCaches.clear();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+
+ cleanPersistenceDir();
+
+ allCaches.clear();
+ mdcSafeCaches.clear();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setDataStorageConfiguration(new DataStorageConfiguration()
+ .setDefaultDataRegionConfiguration(new DataRegionConfiguration()
+ .setPersistenceEnabled(persistenceEnabled)
+ .setMaxSize(32 * 1024 * 1024)
+ ));
+
+ if (useStaticCaches) {
+ CacheConfiguration mdcSafeCacheCfg0 = prepareCacheCfg(
+ CACHE_WITH_MDC_FILTER,
+ new MdcAffinityBackupFilter(2, 1),
+ true);
+
+ CacheConfiguration mdcSafeCacheCfg1 = prepareCacheCfg(
+ CACHE_WITH_COLOCATED_FILTER,
+ new
ClusterNodeAttributeColocatedBackupFilter(STRETCHED_CELL_ATTR_NAME),
+ true);
+
+ CacheConfiguration mdcUnsafeCacheCfg0 =
prepareCacheCfg(MDC_UNSAFE_CACHE, null, false);
+
+ CacheConfiguration mdcUnsafeCacheCfg1 = prepareCacheCfg(
+ CACHE_WITH_MDC_UNSAFE_ATTRIBUTE_FILTER,
+ new
ClusterNodeAttributeAffinityBackupFilter(ATTR_FOR_UNSAFE_ATTR_FILTER),
+ false);
+
+ cfg.setCacheConfiguration(mdcSafeCacheCfg0, mdcSafeCacheCfg1,
mdcUnsafeCacheCfg0, mdcUnsafeCacheCfg1);
+ }
+
+ if (!cfg.isClientMode())
+ cfg.setUserAttributes(F.asMap(
+ STRETCHED_CELL_ATTR_NAME,
+ cellId,
+ IgniteSystemProperties.IGNITE_DATA_CENTER_ID,
+ dcId));
+
+ return cfg;
+ }
+
+ /** */
+ private CacheConfiguration prepareCacheCfg(
+ String cacheName,
+ IgniteBiPredicate<ClusterNode, List<ClusterNode>> affBackupFilter,
+ boolean affCfgMdcSafe) {
+ return prepareCacheCfg(cacheName, affBackupFilter, affCfgMdcSafe,
null);
+ }
+
+ /** */
+ private CacheConfiguration prepareCacheCfg(
+ String cacheName,
+ IgniteBiPredicate<ClusterNode, List<ClusterNode>> affBackupFilter,
+ boolean affCfgMdcSafe,
+ String cacheGroupName) {
+ CacheConfiguration cacheCfg = new CacheConfiguration(cacheName)
+ .setCacheMode(PARTITIONED)
+ .setBackups(1);
+
+ if (cacheGroupName != null)
+ cacheCfg.setGroupName(cacheGroupName);
+
+ cacheCfg.setAffinity(
+ new RendezvousAffinityFunction()
+ .setPartitions(32)
+ .setAffinityBackupFilter(affBackupFilter));
+
+ if (affCfgMdcSafe)
+ mdcSafeCaches.add(cacheName);
+
+ allCaches.add(cacheName);
+
+ return cacheCfg;
+ }
+
+ /**
+ * Test verifies correctness of metric for cache configuration related to
data distribution across DCs
+ * if caches are organized into groups.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testAffinityCfgMdcSafeMetricForCacheGroup() throws Exception {
+ startClusterAcrossDataCenters(new String[] {DC_ID_0, DC_ID_1}, 2);
+
+ IgniteEx client = startClientGrid(NODES_NUMBER - 1);
+
+ client.getOrCreateCache(
+ prepareCacheCfg(
+ CACHE_WITH_MDC_FILTER + "_0",
+ new MdcAffinityBackupFilter(2, 1),
+ true,
+ "mdcSafeCachesGroup"));
+
+ client.getOrCreateCache(
+ prepareCacheCfg(
+ CACHE_WITH_MDC_FILTER + "_1",
+ new MdcAffinityBackupFilter(2, 1),
+ true,
+ "mdcSafeCachesGroup"));
+
+ client.getOrCreateCache(
+ prepareCacheCfg(MDC_UNSAFE_CACHE + "_0", null, false,
"mdcUnsafeCachesGroup"));
+
+ client.getOrCreateCache(
+ prepareCacheCfg(MDC_UNSAFE_CACHE + "_1", null, false,
"mdcUnsafeCachesGroup"));
+
+ checkMdcReadyMetric();
+ }
+
+ /**
+ * Test verifies correctness of metric for partition copies distribution
across DCs
+ * if caches are organized into groups.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testPartitionDistributionMetricForCacheGroups() throws
Exception {
+ persistenceEnabled = true;
+
+ startClusterAcrossDataCenters(new String[] {DC_ID_0, DC_ID_1}, 2);
+
+ IgniteEx client = startClientGrid(NODES_NUMBER - 1);
+
+ client.cluster().state(ClusterState.ACTIVE);
+
+ client.getOrCreateCache(prepareCacheCfg(
+ CACHE_WITH_MDC_FILTER + "_0",
+ new MdcAffinityBackupFilter(2, 1),
+ true,
+ "mdcFilterCacheGroup"));
+ client.getOrCreateCache(prepareCacheCfg(
+ CACHE_WITH_MDC_FILTER + "_1",
+ new MdcAffinityBackupFilter(2, 1),
+ true,
+ "mdcFilterCacheGroup"));
+
+ awaitPartitionMapExchange();
+
+ BooleanMetric cache0DistributionSafeMetric = findMetricForCache(
+ grid(1),
+ CACHE_WITH_MDC_FILTER + "_0",
+ PARTITION_DISTRIBUTION_SAFE_METRIC_NAME);
+ BooleanMetric cache1DistributionSafeMetric = findMetricForCache(
+ grid(1),
+ CACHE_WITH_MDC_FILTER + "_1",
+ PARTITION_DISTRIBUTION_SAFE_METRIC_NAME);
+
+ assertNotNull(cache0DistributionSafeMetric);
+ assertNotNull(cache1DistributionSafeMetric);
+ assertTrue(cache0DistributionSafeMetric.value());
+ assertTrue(cache1DistributionSafeMetric.value());
+
+ stopGrid(0);
+
+ assertFalse(cache0DistributionSafeMetric.value());
+ assertFalse(cache1DistributionSafeMetric.value());
+
+
client.cluster().setBaselineTopology(client.cluster().topologyVersion());
+
+ assertTrue(cache0DistributionSafeMetric.value());
+ assertTrue(cache1DistributionSafeMetric.value());
+ }
+
+ /**
+ * Test verifies correctness of metric for cache configuration related to
data distribution across DCs for dynamically started caches.
+ * Metric should take a {@code false} value if cache configuration doesn't
guarantee presence of data copy in each DC
+ * and {@code true} otherwise.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testAffinityCfgMdcSafeMetricForDynamicCaches() throws
Exception {
+ startClusterAcrossDataCenters(new String[] {DC_ID_0, DC_ID_1}, 2);
+
+ IgniteEx client = startClientGrid(NODES_NUMBER - 1);
+
+ client.cluster().state(ClusterState.ACTIVE);
+
+ client.getOrCreateCache(
+ prepareCacheCfg(CACHE_WITH_MDC_FILTER, new
MdcAffinityBackupFilter(2, 1), true));
+
+ client.getOrCreateCache(
+ prepareCacheCfg(CACHE_WITH_COLOCATED_FILTER, new
ClusterNodeAttributeColocatedBackupFilter(STRETCHED_CELL_ATTR_NAME), true));
+
+ client.getOrCreateCache(
+ prepareCacheCfg(MDC_UNSAFE_CACHE, null, false));
+
+ client.getOrCreateCache(
+ prepareCacheCfg(CACHE_WITH_MDC_UNSAFE_ATTRIBUTE_FILTER,
+ new
ClusterNodeAttributeAffinityBackupFilter(ATTR_FOR_UNSAFE_ATTR_FILTER), false));
+
+ checkMdcReadyMetric();
+ }
+
+ /**
+ * Test verifies correctness of metric for cache configuration related to
data distribution across DCs for statically configured caches.
+ * Metric should take a {@code false} value if cache configuration doesn't
guarantee presence of data copy in each DC
+ * and {@code true} otherwise.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testAffinityCfgMdcSafeMetricForStaticCaches() throws Exception
{
+ useStaticCaches = true;
+
+ startClusterAcrossDataCenters(new String[] {DC_ID_0, DC_ID_1}, 2);
+
+ IgniteEx client = startClientGrid(NODES_NUMBER - 1);
+
+ client.cluster().state(ClusterState.ACTIVE);
+
+ checkMdcReadyMetric();
+ }
+
+ /**
+ * Test verifies correctness of metric for partition copies distribution
across DCs.
+ * Metric should take a {@code false} value if there is at least one
partition which doesn't have copies in all DCs
+ * and {@code true} otherwise.
+ * <p/>
+ * This test considers in-memory caches only.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testPartitionDistributionMetricInMemoryCaches() throws
Exception {
+ startClusterAcrossDataCenters(new String[] {DC_ID_0, DC_ID_1}, 2);
+
+ IgniteEx client = startClientGrid(NODES_NUMBER - 1);
+
+ client.getOrCreateCache(prepareCacheCfg(CACHE_WITH_MDC_FILTER, new
MdcAffinityBackupFilter(2, 1), true));
+ client.getOrCreateCache(prepareCacheCfg(CACHE_WITH_COLOCATED_FILTER,
+ new
ClusterNodeAttributeColocatedBackupFilter(STRETCHED_CELL_ATTR_NAME), true));
+
client.getOrCreateCache(prepareCacheCfg(CACHE_WITH_MDC_SAFE_ATTRIBUTE_FILTER,
+ new ClusterNodeAttributeAffinityBackupFilter(ATTR_DATA_CENTER_ID),
true));
+
+ awaitPartitionMapExchange();
+
+ BooleanMetric cacheWithMdcFilterDistributionSafeMetric =
findMetricForCache(
+ grid(1),
+ CACHE_WITH_MDC_FILTER,
+ PARTITION_DISTRIBUTION_SAFE_METRIC_NAME);
+ BooleanMetric cacheWithColocatedFilterDistributionSafeMetric =
findMetricForCache(
+ grid(1),
+ CACHE_WITH_COLOCATED_FILTER,
+ PARTITION_DISTRIBUTION_SAFE_METRIC_NAME);
+ BooleanMetric cacheWithMdcSafeAttrFilterDistributionSafeMetric =
findMetricForCache(
+ grid(1),
+ CACHE_WITH_MDC_SAFE_ATTRIBUTE_FILTER,
+ PARTITION_DISTRIBUTION_SAFE_METRIC_NAME
+ );
+
+ assertNotNull(cacheWithMdcFilterDistributionSafeMetric);
+ assertNotNull(cacheWithColocatedFilterDistributionSafeMetric);
+ assertNotNull(cacheWithMdcSafeAttrFilterDistributionSafeMetric);
+ assertTrue(cacheWithMdcFilterDistributionSafeMetric.value());
+ assertTrue(cacheWithColocatedFilterDistributionSafeMetric.value());
+ assertTrue(cacheWithMdcSafeAttrFilterDistributionSafeMetric.value());
+
+ stopGrid(0);
+
+ assertTrue(cacheWithMdcFilterDistributionSafeMetric.value());
+ assertTrue(cacheWithMdcSafeAttrFilterDistributionSafeMetric.value());
+ assertFalse(cacheWithColocatedFilterDistributionSafeMetric.value());
+ }
+
+ /**
+ * Test verifies correctness of metric for partition copies distribution
across DCs.
+ * Metric should take a {@code false} value if there is at least one
partition which doesn't have copies in all DCs
+ * and {@code true} otherwise.
+ * <p/>
+ * This test considers persistent caches only and takes into account
changes of BaselineTopology.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testPartitionDistributionMetricPersistentCaches() throws
Exception {
+ persistenceEnabled = true;
+
+ startClusterAcrossDataCenters(new String[] {DC_ID_0, DC_ID_1}, 2);
+
+ IgniteEx client = startClientGrid(NODES_NUMBER - 1);
+
+ client.cluster().state(ClusterState.ACTIVE);
+
+ client.getOrCreateCache(prepareCacheCfg(
+ CACHE_WITH_MDC_FILTER, new MdcAffinityBackupFilter(2, 1), true));
+ client.getOrCreateCache(prepareCacheCfg(CACHE_WITH_COLOCATED_FILTER,
+ new
ClusterNodeAttributeColocatedBackupFilter(STRETCHED_CELL_ATTR_NAME), true));
+
client.getOrCreateCache(prepareCacheCfg(CACHE_WITH_MDC_SAFE_ATTRIBUTE_FILTER,
+ new ClusterNodeAttributeAffinityBackupFilter(ATTR_DATA_CENTER_ID),
true));
+
+ awaitPartitionMapExchange();
+
+ BooleanMetric cacheWithMdcFilterDistributionSafeMetric =
findMetricForCache(
+ grid(1),
+ CACHE_WITH_MDC_FILTER,
+ PARTITION_DISTRIBUTION_SAFE_METRIC_NAME);
+ BooleanMetric cacheWithColocatedFilterDistributionSafeMetric =
findMetricForCache(
+ grid(1),
+ CACHE_WITH_COLOCATED_FILTER,
+ PARTITION_DISTRIBUTION_SAFE_METRIC_NAME);
+ BooleanMetric cacheWithMdcSafeAttrFilterDistributionSafeMetric =
findMetricForCache(
+ grid(1),
+ CACHE_WITH_MDC_SAFE_ATTRIBUTE_FILTER,
+ PARTITION_DISTRIBUTION_SAFE_METRIC_NAME
+ );
+
+ assertNotNull(cacheWithMdcFilterDistributionSafeMetric);
+ assertNotNull(cacheWithColocatedFilterDistributionSafeMetric);
+ assertNotNull(cacheWithMdcSafeAttrFilterDistributionSafeMetric);
+ assertTrue(cacheWithMdcFilterDistributionSafeMetric.value());
+ assertTrue(cacheWithColocatedFilterDistributionSafeMetric.value());
+ assertTrue(cacheWithMdcSafeAttrFilterDistributionSafeMetric.value());
+
+ stopGrid(0);
+
+ assertFalse(cacheWithMdcFilterDistributionSafeMetric.value());
+ assertFalse(cacheWithColocatedFilterDistributionSafeMetric.value());
+ assertFalse(cacheWithMdcSafeAttrFilterDistributionSafeMetric.value());
+
+
client.cluster().setBaselineTopology(client.cluster().topologyVersion());
+
+ // MdcAffinityBackupFilter and
ClusterNodeAttributeAffinityBackupFilter are able to reassign partitions
+ // after BaselineTopology change, thus distribution safe metric
restores to true.
+ assertTrue(cacheWithMdcFilterDistributionSafeMetric.value());
+ assertTrue(cacheWithMdcSafeAttrFilterDistributionSafeMetric.value());
+ // But ClusterNodeAttributeColocatedBackupFilter doesn't reassign
partitions after BaselineTopology change
+ // so distribution safe metric for this cache remains false.
+ assertFalse(cacheWithColocatedFilterDistributionSafeMetric.value());
+ }
+
+ /** */
+ private BooleanMetric findMetricForCache(IgniteEx grid, String cacheName,
String metricName) {
+ GridCacheContext<Object, Object> cacheCtx =
grid.cachex(cacheName).context();
+
+ return
cacheCtx.kernalContext().metric().registry(cacheMetricsRegistryName(
+ cacheCtx.name(),
cacheCtx.cache().isNear())).findMetric(metricName);
+ }
+
+ /** */
+ private void checkMdcReadyMetric() throws InterruptedException {
+ awaitPartitionMapExchange();
+
+ for (int i = 0; i < NODES_NUMBER; i++) {
+ IgniteEx ig = grid(i);
+
+ for (String cacheName : allCaches) {
+ BooleanMetric cacheMdcSafeMetric = findMetricForCache(ig,
cacheName, AFFINITY_CFG_MDC_SAFE_METRIC_NAME);
+
+ if (ig.localNode().isClient()) {
+ assertNull(cacheMdcSafeMetric);
+
+ continue;
+ }
+ else
+ assertNotNull("Grid: " + i + ", cache: " + cacheName,
cacheMdcSafeMetric);
+
+ boolean cacheMdcSafe = cacheMdcSafeMetric.value();
+
+ assertEquals(mdcSafeCaches.contains(cacheName), cacheMdcSafe);
+ }
+ }
+ }
+
+ /**
+ * Test verifies that IsCacheAffinityConfigurationMdcSafe metric is NOT
registered
+ * when MDC attribute (data center ID) is not configured in the cluster.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testAffinityCfgMdcSafeMetricNotRegisteredOnNonMdcCluster()
throws Exception {
+ startGrids(2);
+
+ IgniteEx client = startClientGrid();
+
+ client.cluster().state(ClusterState.ACTIVE);
+
+ client.getOrCreateCache(
+ prepareCacheCfg(CACHE_WITH_MDC_FILTER, new
MdcAffinityBackupFilter(2, 1), true));
+
+ client.getOrCreateCache(
+ prepareCacheCfg(MDC_UNSAFE_CACHE, null, false));
+
+ awaitPartitionMapExchange();
+
+ // Verify that IsCacheAffinityConfigurationMdcSafe metric is NOT
registered on server nodes
+ for (int i = 0; i < 2; i++) {
+ IgniteEx srv = grid(i);
+
+ for (String cacheName : allCaches) {
+ BooleanMetric cacheMdcSafeMetric = findMetricForCache(srv,
cacheName, AFFINITY_CFG_MDC_SAFE_METRIC_NAME);
+
+ assertNull("Metric " + AFFINITY_CFG_MDC_SAFE_METRIC_NAME +
+ " should not be registered when MDC attribute is not
configured [grid=" + i +
+ ", cache=" + cacheName + ']', cacheMdcSafeMetric);
+ }
+ }
+
+ // Verify that the metric is also NOT registered on client nodes
+ for (String cacheName : allCaches) {
+ BooleanMetric cacheMdcSafeMetric = findMetricForCache(client,
cacheName, AFFINITY_CFG_MDC_SAFE_METRIC_NAME);
+
+ assertNull("Metric " + AFFINITY_CFG_MDC_SAFE_METRIC_NAME +
+ " should not be registered on client node [cache=" + cacheName
+ ']', cacheMdcSafeMetric);
+ }
+ }
+
+ /**
+ * Test verifies that IsCachePartitionDistributionSafe metric is NOT
registered
+ * when MDC attribute (data center ID) is not configured in the cluster.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testPartitionDistributionMetricNotRegisteredOnNonMdcCluster()
throws Exception {
+ startGrids(2);
+
+ IgniteEx client = startClientGrid();
+
+ client.cluster().state(ClusterState.ACTIVE);
+
+ client.getOrCreateCache(
+ prepareCacheCfg(CACHE_WITH_MDC_FILTER, new
MdcAffinityBackupFilter(2, 1), true));
+
+ client.getOrCreateCache(
+ prepareCacheCfg(MDC_UNSAFE_CACHE, null, false));
+
+ awaitPartitionMapExchange();
+
+ // Verify that IsCachePartitionDistributionSafe metric is NOT
registered on server nodes
+ for (int i = 0; i < 2; i++) {
+ IgniteEx srv = grid(i);
+
+ for (String cacheName : allCaches) {
+ BooleanMetric partDistMetric = findMetricForCache(srv,
cacheName, PARTITION_DISTRIBUTION_SAFE_METRIC_NAME);
+
+ assertNull("Metric " + PARTITION_DISTRIBUTION_SAFE_METRIC_NAME
+
+ " should not be registered when MDC attribute is not
configured [grid=" + i +
+ ", cache=" + cacheName + ']', partDistMetric);
+ }
+ }
+
+ // Verify that the metric is also NOT registered on client nodes
+ for (String cacheName : allCaches) {
+ BooleanMetric partDistMetric = findMetricForCache(client,
cacheName, PARTITION_DISTRIBUTION_SAFE_METRIC_NAME);
+
+ assertNull("Metric " + PARTITION_DISTRIBUTION_SAFE_METRIC_NAME +
+ " should not be registered on client node [cache=" + cacheName
+ ']', partDistMetric);
+ }
+ }
+
+ /** */
+ private IgniteEx startClusterAcrossDataCenters(String[] dcIds, int
nodesPerDc) throws Exception {
+ int nodeIdx = 0;
+ IgniteEx lastNode = null;
+
+ for (String dcId : dcIds) {
+ this.dcId = dcId;
+
+ for (int i = 0; i < nodesPerDc; i++) {
+ cellId = STRETCHED_CELL_IDS[i];
+
+ lastNode = startGrid(nodeIdx++);
+ }
+ }
+
+ return lastNode;
+ }
+}
diff --git
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
index 58b1b84819a..6b6dd442dad 100644
---
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
+++
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite2.java
@@ -40,6 +40,7 @@ import
org.apache.ignite.internal.processors.cache.IgniteCacheIncrementTxTest;
import
org.apache.ignite.internal.processors.cache.IgniteCachePartitionMapUpdateSafeLossPolicyTest;
import
org.apache.ignite.internal.processors.cache.IgniteDynamicCacheAndNodeStop;
import
org.apache.ignite.internal.processors.cache.IgniteOnePhaseCommitNearReadersTest;
+import org.apache.ignite.internal.processors.cache.MdcCacheMetricsTest;
import
org.apache.ignite.internal.processors.cache.NonAffinityCoordinatorDynamicStartStopTest;
import
org.apache.ignite.internal.processors.cache.RebalanceIteratorLargeEntriesOOMTest;
import
org.apache.ignite.internal.processors.cache.distributed.CacheLoadingConcurrentGridStartSelfTestAllowOverwrite;
@@ -208,6 +209,7 @@ public class IgniteCacheTestSuite2 {
GridTestUtils.addTestIfNeeded(suite,
RendezvousAffinityFunctionBackupFilterSelfTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
ClusterNodeAttributeColocatedBackupFilterSelfTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
MdcAffinityBackupFilterSelfTest.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite, MdcCacheMetricsTest.class,
ignoredTests);
GridTestUtils.addTestIfNeeded(suite, CachePartitionStateTest.class,
ignoredTests);
GridTestUtils.addTestIfNeeded(suite, CacheComparatorTest.class,
ignoredTests);
