IGNITE-28692 Add cache metrics for MDC-related checks

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsImpl.java

@@ -19,16 +19,20 @@
 
 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 void onOffHeapEvict() {
         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 void registerResolverMetrics() {
             "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);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java

@@ -217,6 +217,7 @@
 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 @@ private void onKernalStart(GridCacheAdapter<?, ?> cache) throws IgniteCheckedExc
 
         cache.onKernalStart();
 
+        registerMdcMetricsIfNeeded(cache);
+
         if (ctx.events().isRecordable(EventType.EVT_CACHE_STARTED))
             ctx.events().addEvent(EventType.EVT_CACHE_STARTED);
 
@@ -1144,6 +1147,19 @@ private void onKernalStart(GridCacheAdapter<?, ?> cache) throws IgniteCheckedExc
                 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.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java

@@ -35,6 +35,10 @@
 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.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.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 @@ private static void checkMemoryConfiguration(ClusterNode rmt, GridKernalContext
         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.

modules/core/src/main/java/org/apache/ignite/internal/processors/cluster/BaselineTopology.java

@@ -39,10 +39,12 @@
 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 @@
  *         [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 @@
  *     <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 @@
  *         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 Map<String, Object> attributes(Object consId) {
         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;
+    }
+
     /**
      *
      */

modules/core/src/test/java/org/apache/ignite/cache/affinity/rendezvous/MdcAffinityBackupFilterSelfTest.java

@@ -314,6 +314,8 @@ private IgniteEx startClusterAcrossDataCenters(String[] dcIds, int nodesPerDc) t
                 lastNode = startGrid(nodeIdx++);
         }
 
+        System.clearProperty(IgniteSystemProperties.IGNITE_DATA_CENTER_ID);
+
         return lastNode;
     }