IGNITE-8554 Cache metrics: expose metrics with rebalance info about k…

…eys - Fixes #4094.

Signed-off-by: Ivan Rakov <irakov@apache.org>

modules/core/src/main/java/org/apache/ignite/cache/CacheMetrics.java

@@ -508,6 +508,16 @@ public interface CacheMetrics {
      */
     public int getRebalancingPartitionsCount();
 
+    /**
+     * @return Number of already rebalanced keys.
+     */
+    public long getRebalancedKeys();
+
+    /**
+     * @return Number estimated to rebalance keys.
+     */
+    public long getEstimatedRebalancingKeys();
+
     /**
      * @return Estimated number of keys to be rebalanced on current node.
      */

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

@@ -472,6 +472,7 @@ void onCacheGroupCreated(CacheGroupContext grp) {
                                     clientTop.partitionMap(true),
                                     clientTop.fullUpdateCounters(),
                                     Collections.<Integer>emptySet(),
+                                    null,
                                     null);
                             }
 
@@ -530,7 +531,7 @@ else if (!fetchFuts.containsKey(grp.groupId())) {
 
                 grp.topology().updateTopologyVersion(topFut, discoCache, -1, false);
 
-                grp.topology().update(topVer, partMap, null, Collections.<Integer>emptySet(), null);
+                grp.topology().update(topVer, partMap, null, Collections.<Integer>emptySet(), null, null);
 
                 topFut.validate(grp, discoCache.allNodes());
             }

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

@@ -369,6 +369,16 @@ class CacheClusterMetricsMXBeanImpl implements CacheMetricsMXBean {
         return cache.clusterMetrics().getTotalPartitionsCount();
     }
 
+    /** {@inheritDoc} */
+    @Override public long getRebalancedKeys() {
+        return cache.clusterMetrics().getRebalancedKeys();
+    }
+
+    /** {@inheritDoc} */
+    @Override public long getEstimatedRebalancingKeys() {
+        return cache.clusterMetrics().getEstimatedRebalancingKeys();
+    }
+
     /** {@inheritDoc} */
     @Override public int getRebalancingPartitionsCount() {
         return cache.clusterMetrics().getRebalancingPartitionsCount();

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

@@ -370,6 +370,14 @@ class CacheLocalMetricsMXBeanImpl implements CacheMetricsMXBean {
         return cache.metrics0().getTotalPartitionsCount();
     }
 
+    @Override public long getRebalancedKeys() {
+        return cache.metrics0().getRebalancedKeys();
+    }
+
+    @Override public long getEstimatedRebalancingKeys() {
+        return cache.metrics0().getEstimatedRebalancingKeys();
+    }
+
     /** {@inheritDoc} */
     @Override public int getRebalancingPartitionsCount() {
         return cache.metrics0().getRebalancingPartitionsCount();

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

@@ -856,6 +856,16 @@ public EntriesStatMetrics getEntriesStat() {
         return getEntriesStat().rebalancingPartitionsCount();
     }
 
+    /** {@inheritDoc} */
+    @Override public long getRebalancedKeys() {
+        return rebalancedKeys.get();
+    }
+
+    /** {@inheritDoc} */
+    @Override public long getEstimatedRebalancingKeys() {
+        return estimatedRebalancingKeys.get();
+    }
+
     /** {@inheritDoc} */
     @Override public long getKeysToRebalanceLeft() {
         return Math.max(0, estimatedRebalancingKeys.get() - rebalancedKeys.get());
@@ -935,7 +945,10 @@ public void rebalanceClearingPartitions(int partitions) {
      * First rebalance supply message callback.
      * @param keysCnt Estimated number of keys.
      */
-    public void onRebalancingKeysCountEstimateReceived(long keysCnt) {
+    public void onRebalancingKeysCountEstimateReceived(Long keysCnt) {
+        if (keysCnt == null)
+            return;
+
         estimatedRebalancingKeys.addAndGet(keysCnt);
     }
 

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

@@ -197,6 +197,12 @@ public class CacheMetricsSnapshot implements CacheMetrics, Externalizable {
     /** Rebalancing partitions count. */
     private int rebalancingPartitionsCnt;
 
+    /** Number of already rebalanced keys. */
+    private long rebalancedKeys;
+
+    /** Number estimated to rebalance keys. */
+    private long estimatedRebalancingKeys;
+
     /** Keys to rebalance left. */
     private long keysToRebalanceLeft;
 
@@ -331,6 +337,8 @@ public CacheMetricsSnapshot(CacheMetricsImpl m) {
         totalPartitionsCnt = entriesStat.totalPartitionsCount();
         rebalancingPartitionsCnt = entriesStat.rebalancingPartitionsCount();
 
+        rebalancedKeys = m.getRebalancedKeys();
+        estimatedRebalancingKeys = m.getEstimatedRebalancingKeys();
         keysToRebalanceLeft = m.getKeysToRebalanceLeft();
         rebalancingBytesRate = m.getRebalancingBytesRate();
         rebalancingKeysRate = m.getRebalancingKeysRate();
@@ -459,6 +467,8 @@ public CacheMetricsSnapshot(CacheMetrics loc, Collection<CacheMetrics> metrics)
             else
                 writeBehindErrorRetryCnt = -1;
 
+            rebalancedKeys += e.getRebalancedKeys();
+            estimatedRebalancingKeys += e.getEstimatedRebalancingKeys();
             totalPartitionsCnt += e.getTotalPartitionsCount();
             rebalancingPartitionsCnt += e.getRebalancingPartitionsCount();
             keysToRebalanceLeft += e.getKeysToRebalanceLeft();
@@ -733,6 +743,14 @@ public CacheMetricsSnapshot(CacheMetrics loc, Collection<CacheMetrics> metrics)
         return totalPartitionsCnt;
     }
 
+    @Override public long getRebalancedKeys() {
+        return rebalancedKeys;
+    }
+
+    @Override public long getEstimatedRebalancingKeys() {
+        return estimatedRebalancingKeys;
+    }
+
     /** {@inheritDoc} */
     @Override public int getRebalancingPartitionsCount() {
         return rebalancingPartitionsCnt;
@@ -926,6 +944,12 @@ public CacheMetricsSnapshot(CacheMetrics loc, Collection<CacheMetrics> metrics)
         out.writeLong(keysToRebalanceLeft);
         out.writeLong(rebalancingBytesRate);
         out.writeLong(rebalancingKeysRate);
+
+        out.writeLong(rebalancedKeys);
+        out.writeLong(estimatedRebalancingKeys);
+        out.writeLong(rebalanceStartTime);
+        out.writeLong(rebalanceFinishTime);
+        out.writeLong(rebalanceClearingPartitionsLeft);
     }
 
     /** {@inheritDoc} */
@@ -981,5 +1005,11 @@ public CacheMetricsSnapshot(CacheMetrics loc, Collection<CacheMetrics> metrics)
         keysToRebalanceLeft = in.readLong();
         rebalancingBytesRate = in.readLong();
         rebalancingKeysRate = in.readLong();
+
+        rebalancedKeys = in.readLong();
+        estimatedRebalancingKeys = in.readLong();
+        rebalanceStartTime = in.readLong();
+        rebalanceFinishTime = in.readLong();
+        rebalanceClearingPartitionsLeft = in.readLong();
     }
 }

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

@@ -1121,6 +1121,8 @@ public GridDhtPartitionsFullMessage createPartitionsFullMessage(
                         affCache.similarAffinityKey());
                 }
 
+                m.addPartitionSizes(grp.groupId(), grp.topology().globalPartSizes());
+
                 if (exchId != null) {
                     CachePartitionFullCountersMap cntrsMap = grp.topology().fullUpdateCounters();
 
@@ -1154,6 +1156,8 @@ public GridDhtPartitionsFullMessage createPartitionsFullMessage(
                     m.addPartitionUpdateCounters(top.groupId(), cntrsMap);
                 else
                     m.addPartitionUpdateCounters(top.groupId(), CachePartitionFullCountersMap.toCountersMap(cntrsMap));
+
+                m.addPartitionSizes(top.groupId(), top.globalPartSizes());
             }
         }
 
@@ -1264,9 +1268,9 @@ public GridDhtPartitionsSingleMessage createPartitionsSingleMessage(
 
                     m.addPartitionUpdateCounters(grp.groupId(),
                         newCntrMap ? cntrsMap : CachePartitionPartialCountersMap.toCountersMap(cntrsMap));
-
-                    m.addPartitionSizes(grp.groupId(), grp.topology().partitionSizes());
                 }
+
+                m.addPartitionSizes(grp.groupId(), grp.topology().partitionSizes());
             }
         }
 
@@ -1288,9 +1292,9 @@ public GridDhtPartitionsSingleMessage createPartitionsSingleMessage(
 
                 m.addPartitionUpdateCounters(top.groupId(),
                     newCntrMap ? cntrsMap : CachePartitionPartialCountersMap.toCountersMap(cntrsMap));
-
-                m.addPartitionSizes(top.groupId(), top.partitionSizes());
             }
+
+            m.addPartitionSizes(top.groupId(), top.partitionSizes());
         }
 
         return m;
@@ -1482,6 +1486,7 @@ else if (!grp.isLocal())
                             entry.getValue(),
                             null,
                             msg.partsToReload(cctx.localNodeId(), grpId),
+                            msg.partitionSizes(grpId),
                             msg.topologyVersion());
                     }
                 }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java

@@ -119,6 +119,9 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
     /** */
     private final int parts;
 
+    /** */
+    private volatile Map<Integer, Long> globalPartSizes;
+
     /**
      * @param cctx Context.
      * @param discoCache Discovery data cache.
@@ -707,6 +710,7 @@ private boolean shouldOverridePartitionMap(GridDhtPartitionMap currentMap, GridD
         GridDhtPartitionFullMap partMap,
         @Nullable CachePartitionFullCountersMap cntrMap,
         Set<Integer> partsToReload,
+        @Nullable Map<Integer, Long> partSizes,
         @Nullable AffinityTopologyVersion msgTopVer) {
         if (log.isDebugEnabled())
             log.debug("Updating full partition map [exchVer=" + exchangeVer + ", parts=" + fullMapString() + ']');
@@ -810,6 +814,9 @@ private boolean shouldOverridePartitionMap(GridDhtPartitionMap currentMap, GridD
             if (cntrMap != null)
                 this.cntrMap = new CachePartitionFullCountersMap(cntrMap);
 
+            if (partSizes != null)
+                this.globalPartSizes = partSizes;
+
             consistencyCheck();
 
             if (log.isDebugEnabled())
@@ -1222,6 +1229,34 @@ private void removeNode(UUID nodeId) {
         return Collections.emptyMap();
     }
 
+    /** {@inheritDoc} */
+    @Override @Nullable public Map<Integer, Long> globalPartSizes() {
+        lock.readLock().lock();
+
+        try {
+            if (globalPartSizes == null)
+                return Collections.emptyMap();
+
+            return Collections.unmodifiableMap(globalPartSizes);
+        }
+        finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void globalPartSizes(@Nullable Map<Integer, Long> partSizes) {
+        lock.writeLock().lock();
+
+        try {
+            this.globalPartSizes = partSizes;
+        }
+        finally {
+            lock.writeLock().unlock();
+        }
+    }
+
+
     /** {@inheritDoc} */
     @Override public boolean rebalanceFinished(AffinityTopologyVersion topVer) {
         assert false : "Should not be called on non-affinity node";

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopology.java

@@ -287,6 +287,7 @@ public boolean update(
         GridDhtPartitionFullMap partMap,
         @Nullable CachePartitionFullCountersMap cntrMap,
         Set<Integer> partsToReload,
+        @Nullable Map<Integer, Long> partSizes,
         @Nullable AffinityTopologyVersion msgTopVer);
 
     /**
@@ -382,6 +383,16 @@ public boolean update(@Nullable GridDhtPartitionExchangeId exchId,
      */
     public void printMemoryStats(int threshold);
 
+    /**
+     * @return Sizes of up-to-date partition versions in topology.
+     */
+    Map<Integer, Long> globalPartSizes();
+
+    /**
+     * @param partSizes Sizes of up-to-date partition versions in topology.
+     */
+    void globalPartSizes(@Nullable Map<Integer, Long> partSizes);
+
     /**
      * @param topVer Topology version.
      * @return {@code True} if rebalance process finished.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java

@@ -137,6 +137,9 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
     /** Partition update counter. */
     private final CachePartitionFullCountersMap cntrMap;
 
+    /** */
+    private volatile Map<Integer, Long> globalPartSizes;
+
     /** */
     private volatile AffinityTopologyVersion rebalancedTopVer = AffinityTopologyVersion.NONE;
 
@@ -1329,6 +1332,7 @@ private boolean shouldOverridePartitionMap(GridDhtPartitionMap currentMap, GridD
         GridDhtPartitionFullMap partMap,
         @Nullable CachePartitionFullCountersMap incomeCntrMap,
         Set<Integer> partsToReload,
+        @Nullable Map<Integer, Long> partSizes,
         @Nullable AffinityTopologyVersion msgTopVer) {
         if (log.isDebugEnabled()) {
             log.debug("Updating full partition map [grp=" + grp.cacheOrGroupName() + ", exchVer=" + exchangeVer +
@@ -1544,6 +1548,9 @@ else if (state == MOVING) {
                     updateRebalanceVersion(aff.assignment());
                 }
 
+                if (partSizes != null)
+                    this.globalPartSizes = partSizes;
+
                 consistencyCheck();
 
                 if (log.isDebugEnabled()) {
@@ -2624,6 +2631,33 @@ private void removeNode(UUID nodeId) {
         }
     }
 
+    /** {@inheritDoc} */
+    @Override public Map<Integer, Long> globalPartSizes() {
+        lock.readLock().lock();
+
+        try {
+            if (globalPartSizes == null)
+                return Collections.emptyMap();
+
+            return Collections.unmodifiableMap(globalPartSizes);
+        }
+        finally {
+            lock.readLock().unlock();
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void globalPartSizes(@Nullable Map<Integer, Long> partSizes) {
+        lock.writeLock().lock();
+
+        try {
+            this.globalPartSizes = partSizes;
+        }
+        finally {
+            lock.writeLock().unlock();
+        }
+    }
+
     /** {@inheritDoc} */
     @Override public boolean rebalanceFinished(AffinityTopologyVersion topVer) {
         AffinityTopologyVersion curTopVer = this.readyTopVer;