You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by av...@apache.org on 2018/01/15 11:39:06 UTC
ignite git commit: IGNITE-6923 Cache metrics are updated in
timeout-worker potentially delaying critical code execution due to current
implementation issues. (cherry picked from commit 83f104a)
Repository: ignite
Updated Branches:
refs/heads/ignite-2.4 31c60bd6b -> a9ad299c3
IGNITE-6923 Cache metrics are updated in timeout-worker potentially delaying critical code execution due to current implementation issues.
(cherry picked from commit 83f104a)
Signed-off-by: Anton Vinogradov <av...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/a9ad299c
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/a9ad299c
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/a9ad299c
Branch: refs/heads/ignite-2.4
Commit: a9ad299c3ec62000fb85777f1c4c2d2c0ed07940
Parents: 31c60bd
Author: Aleksey Plekhanov <pl...@gmail.com>
Authored: Mon Jan 15 14:35:17 2018 +0300
Committer: Anton Vinogradov <av...@apache.org>
Committed: Mon Jan 15 14:38:57 2018 +0300
----------------------------------------------------------------------
.../managers/discovery/ClusterMetricsImpl.java | 369 +++++++++++++++++++
.../discovery/GridDiscoveryManager.java | 98 +----
.../processors/cache/CacheMetricsImpl.java | 316 +++++++++++++---
.../processors/cache/CacheMetricsSnapshot.java | 26 +-
.../cache/CacheMetricsEntitiesCountTest.java | 172 +++++++++
.../IgniteCacheMetricsSelfTestSuite.java | 4 +-
6 files changed, 821 insertions(+), 164 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/a9ad299c/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/ClusterMetricsImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/ClusterMetricsImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/ClusterMetricsImpl.java
new file mode 100644
index 0000000..4138e641
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/ClusterMetricsImpl.java
@@ -0,0 +1,369 @@
+/*
+ * 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.managers.discovery;
+
+import java.util.Collection;
+import org.apache.ignite.cluster.ClusterMetrics;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
+import org.apache.ignite.internal.processors.jobmetrics.GridJobMetrics;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+/**
+ * Cluster metrics proxy
+ */
+public class ClusterMetricsImpl implements ClusterMetrics {
+ /** Job metrics. */
+ private volatile GridJobMetrics jobMetrics;
+
+ /** Job metrics update time. */
+ private volatile long jobMetricsUpdateTime;
+
+ /** Job metrics mutex. */
+ private final Object jobMetricsMux = new Object();
+
+ /** Kernel context. */
+ private final GridKernalContext ctx;
+
+ /** VM Metrics. */
+ private final GridLocalMetrics vmMetrics;
+
+ /** Node start time. */
+ private final long nodeStartTime;
+
+ /**
+ * @param ctx Kernel context.
+ * @param vmMetrics VM metrics.
+ * @param nodeStartTime Node start time;
+ */
+ public ClusterMetricsImpl(GridKernalContext ctx, GridLocalMetrics vmMetrics, long nodeStartTime) {
+ this.ctx = ctx;
+ this.vmMetrics = vmMetrics;
+ this.nodeStartTime = nodeStartTime;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getLastUpdateTime() {
+ return jobMetricsUpdateTime == 0 ? U.currentTimeMillis() : jobMetricsUpdateTime;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getMaximumActiveJobs() {
+ return jobMetrics().getMaximumActiveJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getCurrentActiveJobs() {
+ return jobMetrics().getCurrentActiveJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public float getAverageActiveJobs() {
+ return jobMetrics().getAverageActiveJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getMaximumWaitingJobs() {
+ return jobMetrics().getMaximumWaitingJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getCurrentWaitingJobs() {
+ return jobMetrics().getCurrentWaitingJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public float getAverageWaitingJobs() {
+ return jobMetrics().getAverageWaitingJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getMaximumRejectedJobs() {
+ return jobMetrics().getMaximumRejectedJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getCurrentRejectedJobs() {
+ return jobMetrics().getCurrentRejectedJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public float getAverageRejectedJobs() {
+ return jobMetrics().getAverageRejectedJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getTotalRejectedJobs() {
+ return jobMetrics().getTotalRejectedJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getMaximumCancelledJobs() {
+ return jobMetrics().getMaximumCancelledJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getCurrentCancelledJobs() {
+ return jobMetrics().getCurrentCancelledJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public float getAverageCancelledJobs() {
+ return jobMetrics().getAverageCancelledJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getTotalCancelledJobs() {
+ return jobMetrics().getTotalCancelledJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getTotalExecutedJobs() {
+ return jobMetrics().getTotalExecutedJobs();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getTotalJobsExecutionTime() {
+ return jobMetrics().getTotalJobsExecutionTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getMaximumJobWaitTime() {
+ return jobMetrics().getMaximumJobWaitTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getCurrentJobWaitTime() {
+ return jobMetrics().getCurrentJobWaitTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public double getAverageJobWaitTime() {
+ return jobMetrics().getAverageJobWaitTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getMaximumJobExecuteTime() {
+ return jobMetrics().getMaximumJobExecuteTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getCurrentJobExecuteTime() {
+ return jobMetrics().getCurrentJobExecuteTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public double getAverageJobExecuteTime() {
+ return jobMetrics().getAverageJobExecuteTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getTotalExecutedTasks() {
+ return ctx.task().getTotalExecutedTasks();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getTotalBusyTime() {
+ return getUpTime() - getTotalIdleTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getTotalIdleTime() {
+ return jobMetrics().getTotalIdleTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getCurrentIdleTime() {
+ return jobMetrics().getCurrentIdleTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public float getBusyTimePercentage() {
+ return 1 - getIdleTimePercentage();
+ }
+
+ /** {@inheritDoc} */
+ @Override public float getIdleTimePercentage() {
+ return getTotalIdleTime() / (float)getUpTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getTotalCpus() {
+ return vmMetrics.getAvailableProcessors();
+ }
+
+ /** {@inheritDoc} */
+ @Override public double getCurrentCpuLoad() {
+ return vmMetrics.getCurrentCpuLoad();
+ }
+
+ /** {@inheritDoc} */
+ @Override public double getAverageCpuLoad() {
+ return jobMetrics().getAverageCpuLoad();
+ }
+
+ /** {@inheritDoc} */
+ @Override public double getCurrentGcCpuLoad() {
+ return vmMetrics.getCurrentGcCpuLoad();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getHeapMemoryInitialized() {
+ return vmMetrics.getHeapMemoryInitialized();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getHeapMemoryUsed() {
+ return vmMetrics.getHeapMemoryUsed();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getHeapMemoryCommitted() {
+ return vmMetrics.getHeapMemoryCommitted();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getHeapMemoryMaximum() {
+ return vmMetrics.getHeapMemoryMaximum();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getHeapMemoryTotal() {
+ return vmMetrics.getHeapMemoryMaximum();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getNonHeapMemoryInitialized() {
+ return vmMetrics.getNonHeapMemoryInitialized();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getNonHeapMemoryUsed() {
+ Collection<GridCacheAdapter<?, ?>> caches = ctx.cache().internalCaches();
+
+ long nonHeapUsed = vmMetrics.getNonHeapMemoryUsed();
+
+ for (GridCacheAdapter<?, ?> cache : caches)
+ if (cache.context().statisticsEnabled() && cache.context().started()
+ && cache.context().affinity().affinityTopologyVersion().topologyVersion() > 0)
+ nonHeapUsed += cache.metrics0().getOffHeapAllocatedSize();
+
+ return nonHeapUsed;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getNonHeapMemoryCommitted() {
+ return vmMetrics.getNonHeapMemoryCommitted();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getNonHeapMemoryMaximum() {
+ return vmMetrics.getNonHeapMemoryMaximum();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getNonHeapMemoryTotal() {
+ return vmMetrics.getNonHeapMemoryMaximum();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getUpTime() {
+ return vmMetrics.getUptime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getStartTime() {
+ return vmMetrics.getStartTime();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getNodeStartTime() {
+ return nodeStartTime;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getCurrentThreadCount() {
+ return vmMetrics.getThreadCount();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getMaximumThreadCount() {
+ return vmMetrics.getPeakThreadCount();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getTotalStartedThreadCount() {
+ return vmMetrics.getTotalStartedThreadCount();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getCurrentDaemonThreadCount() {
+ return vmMetrics.getDaemonThreadCount();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getLastDataVersion() {
+ return ctx.cache().lastDataVersion();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getSentMessagesCount() {
+ return ctx.io().getSentMessagesCount();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getSentBytesCount() {
+ return ctx.io().getSentBytesCount();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getReceivedMessagesCount() {
+ return ctx.io().getReceivedMessagesCount();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long getReceivedBytesCount() {
+ return ctx.io().getReceivedBytesCount();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getOutboundMessagesQueueSize() {
+ return ctx.io().getOutboundMessagesQueueSize();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getTotalNodes() {
+ return 1;
+ }
+
+ /**
+ * Job metrics
+ */
+ public GridJobMetrics jobMetrics() {
+ if (jobMetrics == null)
+ synchronized (jobMetricsMux) {
+ if (jobMetrics == null) {
+ jobMetricsUpdateTime = U.currentTimeMillis();
+
+ jobMetrics = ctx.jobMetric().getJobMetrics();
+ }
+ }
+
+ return jobMetrics;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a9ad299c/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index faf222f..088c9ce 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -53,11 +53,10 @@ import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cluster.ClusterMetrics;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
-import org.apache.ignite.internal.ClusterMetricsSnapshot;
import org.apache.ignite.internal.GridComponent;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.GridNodeOrderComparator;
@@ -66,7 +65,6 @@ import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.events.DiscoveryCustomEvent;
import org.apache.ignite.internal.managers.GridManagerAdapter;
-import org.apache.ignite.internal.managers.communication.GridIoManager;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
@@ -82,7 +80,6 @@ import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMess
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
import org.apache.ignite.internal.processors.cluster.GridClusterStateProcessor;
-import org.apache.ignite.internal.processors.jobmetrics.GridJobMetrics;
import org.apache.ignite.internal.processors.security.SecurityContext;
import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import org.apache.ignite.internal.util.GridAtomicLong;
@@ -1048,98 +1045,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
/** {@inheritDoc} */
@Override public ClusterMetrics metrics() {
- GridJobMetrics jm = ctx.jobMetric().getJobMetrics();
-
- ClusterMetricsSnapshot nm = new ClusterMetricsSnapshot();
-
- nm.setLastUpdateTime(U.currentTimeMillis());
-
- // Job metrics.
- nm.setMaximumActiveJobs(jm.getMaximumActiveJobs());
- nm.setCurrentActiveJobs(jm.getCurrentActiveJobs());
- nm.setAverageActiveJobs(jm.getAverageActiveJobs());
- nm.setMaximumWaitingJobs(jm.getMaximumWaitingJobs());
- nm.setCurrentWaitingJobs(jm.getCurrentWaitingJobs());
- nm.setAverageWaitingJobs(jm.getAverageWaitingJobs());
- nm.setMaximumRejectedJobs(jm.getMaximumRejectedJobs());
- nm.setCurrentRejectedJobs(jm.getCurrentRejectedJobs());
- nm.setAverageRejectedJobs(jm.getAverageRejectedJobs());
- nm.setMaximumCancelledJobs(jm.getMaximumCancelledJobs());
- nm.setCurrentCancelledJobs(jm.getCurrentCancelledJobs());
- nm.setAverageCancelledJobs(jm.getAverageCancelledJobs());
- nm.setTotalRejectedJobs(jm.getTotalRejectedJobs());
- nm.setTotalCancelledJobs(jm.getTotalCancelledJobs());
- nm.setTotalExecutedJobs(jm.getTotalExecutedJobs());
- nm.setTotalJobsExecutionTime(jm.getTotalJobsExecutionTime());
- nm.setMaximumJobWaitTime(jm.getMaximumJobWaitTime());
- nm.setCurrentJobWaitTime(jm.getCurrentJobWaitTime());
- nm.setAverageJobWaitTime(jm.getAverageJobWaitTime());
- nm.setMaximumJobExecuteTime(jm.getMaximumJobExecuteTime());
- nm.setCurrentJobExecuteTime(jm.getCurrentJobExecuteTime());
- nm.setAverageJobExecuteTime(jm.getAverageJobExecuteTime());
- nm.setCurrentIdleTime(jm.getCurrentIdleTime());
- nm.setTotalIdleTime(jm.getTotalIdleTime());
- nm.setAverageCpuLoad(jm.getAverageCpuLoad());
-
- // Job metrics.
- nm.setTotalExecutedTasks(ctx.task().getTotalExecutedTasks());
-
- // VM metrics.
- nm.setAvailableProcessors(metrics.getAvailableProcessors());
- nm.setCurrentCpuLoad(metrics.getCurrentCpuLoad());
- nm.setCurrentGcCpuLoad(metrics.getCurrentGcCpuLoad());
- nm.setHeapMemoryInitialized(metrics.getHeapMemoryInitialized());
- nm.setHeapMemoryUsed(metrics.getHeapMemoryUsed());
- nm.setHeapMemoryCommitted(metrics.getHeapMemoryCommitted());
- nm.setHeapMemoryMaximum(metrics.getHeapMemoryMaximum());
- nm.setHeapMemoryTotal(metrics.getHeapMemoryMaximum());
- nm.setNonHeapMemoryInitialized(metrics.getNonHeapMemoryInitialized());
- nonHeapMemoryUsed(nm);
- nm.setNonHeapMemoryCommitted(metrics.getNonHeapMemoryCommitted());
- nm.setNonHeapMemoryMaximum(metrics.getNonHeapMemoryMaximum());
- nm.setNonHeapMemoryTotal(metrics.getNonHeapMemoryMaximum());
- nm.setUpTime(metrics.getUptime());
- nm.setStartTime(metrics.getStartTime());
- nm.setNodeStartTime(startTime);
- nm.setCurrentThreadCount(metrics.getThreadCount());
- nm.setMaximumThreadCount(metrics.getPeakThreadCount());
- nm.setTotalStartedThreadCount(metrics.getTotalStartedThreadCount());
- nm.setCurrentDaemonThreadCount(metrics.getDaemonThreadCount());
- nm.setTotalNodes(1);
-
- // Data metrics.
- nm.setLastDataVersion(ctx.cache().lastDataVersion());
-
- GridIoManager io = ctx.io();
-
- // IO metrics.
- nm.setSentMessagesCount(io.getSentMessagesCount());
- nm.setSentBytesCount(io.getSentBytesCount());
- nm.setReceivedMessagesCount(io.getReceivedMessagesCount());
- nm.setReceivedBytesCount(io.getReceivedBytesCount());
- nm.setOutboundMessagesQueueSize(io.getOutboundMessagesQueueSize());
-
- return nm;
- }
-
- /**
- * @param nm Initializing metrics snapshot.
- */
- private void nonHeapMemoryUsed(ClusterMetricsSnapshot nm) {
- long nonHeapUsed = metrics.getNonHeapMemoryUsed();
-
- Map<Integer, CacheMetrics> nodeCacheMetrics = cacheMetrics();
-
- if (nodeCacheMetrics != null) {
- for (Map.Entry<Integer, CacheMetrics> entry : nodeCacheMetrics.entrySet()) {
- CacheMetrics e = entry.getValue();
-
- if (e != null)
- nonHeapUsed += e.getOffHeapAllocatedSize();
- }
- }
-
- nm.setNonHeapMemoryUsed(nonHeapUsed);
+ return new ClusterMetricsImpl(ctx, metrics, startTime);
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/a9ad299c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsImpl.java
----------------------------------------------------------------------
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 8d46a21..1603ecb 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
@@ -17,13 +17,14 @@
package org.apache.ignite.internal.processors.cache;
-import java.util.Map;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
-import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.CacheMetrics;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.ratemetrics.HitRateMetrics;
@@ -220,45 +221,22 @@ public class CacheMetricsImpl implements CacheMetrics {
/** {@inheritDoc} */
@Override public long getOffHeapEntriesCount() {
- GridCacheAdapter<?, ?> cache = cctx.cache();
-
- return cache != null ? cache.offHeapEntriesCount() : -1;
+ return getEntriesStat().offHeapEntriesCount();
}
/** {@inheritDoc} */
@Override public long getHeapEntriesCount() {
- try {
- return cctx.cache().localSizeLong(ONHEAP_PEEK_MODES);
- }
- catch (IgniteCheckedException ignored) {
- return 0;
- }
+ return getEntriesStat().heapEntriesCount();
}
/** {@inheritDoc} */
@Override public long getOffHeapPrimaryEntriesCount() {
- try {
- return cctx.offheap().cacheEntriesCount(cctx.cacheId(),
- true,
- false,
- cctx.affinity().affinityTopologyVersion());
- }
- catch (IgniteCheckedException ignored) {
- return 0;
- }
+ return getEntriesStat().offHeapPrimaryEntriesCount();
}
/** {@inheritDoc} */
@Override public long getOffHeapBackupEntriesCount() {
- try {
- return cctx.offheap().cacheEntriesCount(cctx.cacheId(),
- false,
- true,
- cctx.affinity().affinityTopologyVersion());
- }
- catch (IgniteCheckedException ignored) {
- return 0;
- }
+ return getEntriesStat().offHeapBackupEntriesCount();
}
/** {@inheritDoc} */
@@ -270,21 +248,17 @@ public class CacheMetricsImpl implements CacheMetrics {
/** {@inheritDoc} */
@Override public int getSize() {
- GridCacheAdapter<?, ?> cache = cctx.cache();
-
- return cache != null ? cache.size() : 0;
+ return getEntriesStat().size();
}
/** {@inheritDoc} */
@Override public int getKeySize() {
- return getSize();
+ return getEntriesStat().keySize();
}
/** {@inheritDoc} */
@Override public boolean isEmpty() {
- GridCacheAdapter<?, ?> cache = cctx.cache();
-
- return cache == null || cache.isEmpty();
+ return getEntriesStat().isEmpty();
}
/** {@inheritDoc} */
@@ -766,34 +740,106 @@ public class CacheMetricsImpl implements CacheMetrics {
return ccfg != null && ccfg.isManagementEnabled();
}
- /** {@inheritDoc} */
- @Override public int getTotalPartitionsCount() {
- int res = 0;
+ /**
+ * Calculates entries count/partitions count metrics using one iteration over local partitions for all metrics
+ */
+ public EntriesStatMetrics getEntriesStat() {
+ int owningPartCnt = 0;
+ int movingPartCnt = 0;
+ long offHeapEntriesCnt = 0L;
+ long offHeapPrimaryEntriesCnt = 0L;
+ long offHeapBackupEntriesCnt = 0L;
+ long heapEntriesCnt = 0L;
+ int size = 0;
+ boolean isEmpty;
- if (cctx.isLocal())
- return res;
+ try {
+ if (cctx.isLocal()) {
+ if (cctx.cache() != null) {
+ offHeapEntriesCnt = cctx.cache().offHeapEntriesCount();
- for (Map.Entry<Integer, GridDhtPartitionState> e : cctx.topology().localPartitionMap().entrySet()) {
- if (e.getValue() == GridDhtPartitionState.OWNING || e.getValue() == GridDhtPartitionState.MOVING)
- res++;
- }
+ offHeapPrimaryEntriesCnt = offHeapEntriesCnt;
+ offHeapBackupEntriesCnt = offHeapEntriesCnt;
- return res;
- }
+ size = cctx.cache().size();
- /** {@inheritDoc} */
- @Override public int getRebalancingPartitionsCount() {
- int res = 0;
+ heapEntriesCnt = size;
+ }
+ }
+ else {
+ AffinityTopologyVersion topVer = cctx.affinity().affinityTopologyVersion();
- if (cctx.isLocal())
- return res;
+ Set<Integer> primaries = cctx.affinity().primaryPartitions(cctx.localNodeId(), topVer);
+ Set<Integer> backups = cctx.affinity().backupPartitions(cctx.localNodeId(), topVer);
+
+ if (cctx.isNear() && cctx.cache() != null)
+ heapEntriesCnt = cctx.cache().nearSize();
+
+ for (GridDhtLocalPartition part : cctx.topology().currentLocalPartitions()) {
+ // Partitions count.
+ GridDhtPartitionState partState = part.state();
+
+ if (partState == GridDhtPartitionState.OWNING)
+ owningPartCnt++;
+
+ if (partState == GridDhtPartitionState.MOVING)
+ movingPartCnt++;
+
+ // Offheap entries count
+ if (cctx.cache() == null)
+ continue;
+
+ int cacheSize = part.dataStore().cacheSize(cctx.cacheId());
+
+ offHeapEntriesCnt += cacheSize;
+
+ if (primaries.contains(part.id()))
+ offHeapPrimaryEntriesCnt += cacheSize;
+
+ if (backups.contains(part.id()))
+ offHeapBackupEntriesCnt += cacheSize;
- for (Map.Entry<Integer, GridDhtPartitionState> e : cctx.topology().localPartitionMap().entrySet()) {
- if (e.getValue() == GridDhtPartitionState.MOVING)
- res++;
+ size = (int)offHeapEntriesCnt;
+
+ heapEntriesCnt += part.publicSize(cctx.cacheId());
+ }
+ }
+ }
+ catch (Exception e) {
+ owningPartCnt = -1;
+ movingPartCnt = 0;
+ offHeapEntriesCnt = -1L;
+ offHeapPrimaryEntriesCnt = -1L;
+ offHeapBackupEntriesCnt = -1L;
+ heapEntriesCnt = -1L;
+ size = -1;
}
- return res;
+ isEmpty = (offHeapEntriesCnt == 0);
+
+ EntriesStatMetrics stat = new EntriesStatMetrics();
+
+ stat.offHeapEntriesCount(offHeapEntriesCnt);
+ stat.offHeapPrimaryEntriesCount(offHeapPrimaryEntriesCnt);
+ stat.offHeapBackupEntriesCount(offHeapBackupEntriesCnt);
+ stat.heapEntriesCount(heapEntriesCnt);
+ stat.size(size);
+ stat.keySize(size);
+ stat.isEmpty(isEmpty);
+ stat.totalPartitionsCount(owningPartCnt + movingPartCnt);
+ stat.rebalancingPartitionsCount(movingPartCnt);
+
+ return stat;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getTotalPartitionsCount() {
+ return getEntriesStat().totalPartitionsCount();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int getRebalancingPartitionsCount() {
+ return getEntriesStat().rebalancingPartitionsCount();
}
/** {@inheritDoc} */
@@ -951,4 +997,162 @@ public class CacheMetricsImpl implements CacheMetrics {
@Override public String toString() {
return S.toString(CacheMetricsImpl.class, this);
}
+
+ /**
+ * Entries and partitions metrics holder class.
+ */
+ public static class EntriesStatMetrics {
+ /** Total partitions count. */
+ private int totalPartsCnt;
+
+ /** Rebalancing partitions count. */
+ private int rebalancingPartsCnt;
+
+ /** Offheap entries count. */
+ private long offHeapEntriesCnt;
+
+ /** Offheap primary entries count. */
+ private long offHeapPrimaryEntriesCnt;
+
+ /** Offheap backup entries count. */
+ private long offHeapBackupEntriesCnt;
+
+ /** Onheap entries count. */
+ private long heapEntriesCnt;
+
+ /** Size. */
+ private int size;
+
+ /** Key size. */
+ private int keySize;
+
+ /** Is empty. */
+ private boolean isEmpty;
+
+ /**
+ * @return Total partitions count.
+ */
+ public int totalPartitionsCount() {
+ return totalPartsCnt;
+ }
+
+ /**
+ * @param totalPartsCnt Total partitions count.
+ */
+ public void totalPartitionsCount(int totalPartsCnt) {
+ this.totalPartsCnt = totalPartsCnt;
+ }
+
+ /**
+ * @return Rebalancing partitions count.
+ */
+ public int rebalancingPartitionsCount() {
+ return rebalancingPartsCnt;
+ }
+
+ /**
+ * @param rebalancingPartsCnt Rebalancing partitions count.
+ */
+ public void rebalancingPartitionsCount(int rebalancingPartsCnt) {
+ this.rebalancingPartsCnt = rebalancingPartsCnt;
+ }
+
+ /**
+ * @return Offheap entries count.
+ */
+ public long offHeapEntriesCount() {
+ return offHeapEntriesCnt;
+ }
+
+ /**
+ * @param offHeapEntriesCnt Offheap entries count.
+ */
+ public void offHeapEntriesCount(long offHeapEntriesCnt) {
+ this.offHeapEntriesCnt = offHeapEntriesCnt;
+ }
+
+ /**
+ * @return Offheap primary entries count.
+ */
+ public long offHeapPrimaryEntriesCount() {
+ return offHeapPrimaryEntriesCnt;
+ }
+
+ /**
+ * @param offHeapPrimaryEntriesCnt Offheap primary entries count.
+ */
+ public void offHeapPrimaryEntriesCount(long offHeapPrimaryEntriesCnt) {
+ this.offHeapPrimaryEntriesCnt = offHeapPrimaryEntriesCnt;
+ }
+
+ /**
+ * @return Offheap backup entries count.
+ */
+ public long offHeapBackupEntriesCount() {
+ return offHeapBackupEntriesCnt;
+ }
+
+ /**
+ * @param offHeapBackupEntriesCnt Offheap backup entries count.
+ */
+ public void offHeapBackupEntriesCount(long offHeapBackupEntriesCnt) {
+ this.offHeapBackupEntriesCnt = offHeapBackupEntriesCnt;
+ }
+
+ /**
+ * @return Heap entries count.
+ */
+ public long heapEntriesCount() {
+ return heapEntriesCnt;
+ }
+
+ /**
+ * @param heapEntriesCnt Onheap entries count.
+ */
+ public void heapEntriesCount(long heapEntriesCnt) {
+ this.heapEntriesCnt = heapEntriesCnt;
+ }
+
+ /**
+ * @return Size.
+ */
+ public int size() {
+ return size;
+ }
+
+ /**
+ * @param size Size.
+ */
+ public void size(int size) {
+ this.size = size;
+ }
+
+ /**
+ * @return Key size.
+ */
+ public int keySize() {
+ return keySize;
+ }
+
+ /**
+ * @param keySize Key size.
+ */
+ public void keySize(int keySize) {
+ this.keySize = keySize;
+ }
+
+ /**
+ * @return Is empty.
+ */
+ public boolean isEmpty() {
+ return isEmpty;
+ }
+
+ /**
+ * @param isEmpty Is empty flag.
+ */
+ public void isEmpty(boolean isEmpty) {
+ this.isEmpty = isEmpty;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a9ad299c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsSnapshot.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsSnapshot.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsSnapshot.java
index 08fc571..af6a174 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsSnapshot.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheMetricsSnapshot.java
@@ -248,7 +248,7 @@ public class CacheMetricsSnapshot implements CacheMetrics, Externalizable {
*
* @param m Cache metrics.
*/
- public CacheMetricsSnapshot(CacheMetrics m) {
+ public CacheMetricsSnapshot(CacheMetricsImpl m) {
reads = m.getCacheGets();
puts = m.getCachePuts();
hits = m.getCacheHits();
@@ -272,15 +272,20 @@ public class CacheMetricsSnapshot implements CacheMetrics, Externalizable {
offHeapEvicts = m.getOffHeapEvictions();
offHeapHits = m.getOffHeapHits();
offHeapMisses = m.getOffHeapMisses();
- offHeapEntriesCnt = m.getOffHeapEntriesCount();
- heapEntriesCnt = m.getHeapEntriesCount();
- offHeapPrimaryEntriesCnt = m.getOffHeapPrimaryEntriesCount();
- offHeapBackupEntriesCnt = m.getOffHeapBackupEntriesCount();
+
+ CacheMetricsImpl.EntriesStatMetrics entriesStat = m.getEntriesStat();
+
+ offHeapEntriesCnt = entriesStat.offHeapEntriesCount();
+ heapEntriesCnt = entriesStat.heapEntriesCount();
+ offHeapPrimaryEntriesCnt = entriesStat.offHeapPrimaryEntriesCount();
+ offHeapBackupEntriesCnt = entriesStat.offHeapBackupEntriesCount();
+
offHeapAllocatedSize = m.getOffHeapAllocatedSize();
- size = m.getSize();
- keySize = m.getKeySize();
- isEmpty = m.isEmpty();
+ size = entriesStat.size();
+ keySize = entriesStat.keySize();
+ isEmpty = entriesStat.isEmpty();
+
dhtEvictQueueCurrSize = m.getDhtEvictQueueCurrentSize();
txThreadMapSize = m.getTxThreadMapSize();
txXidMapSize = m.getTxXidMapSize();
@@ -316,8 +321,9 @@ public class CacheMetricsSnapshot implements CacheMetrics, Externalizable {
isValidForReading = m.isValidForReading();
isValidForWriting = m.isValidForWriting();
- totalPartitionsCnt = m.getTotalPartitionsCount();
- rebalancingPartitionsCnt = m.getRebalancingPartitionsCount();
+ totalPartitionsCnt = entriesStat.totalPartitionsCount();
+ rebalancingPartitionsCnt = entriesStat.rebalancingPartitionsCount();
+
keysToRebalanceLeft = m.getKeysToRebalanceLeft();
rebalancingBytesRate = m.getRebalancingBytesRate();
rebalancingKeysRate = m.getRebalancingKeysRate();
http://git-wip-us.apache.org/repos/asf/ignite/blob/a9ad299c/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheMetricsEntitiesCountTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheMetricsEntitiesCountTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheMetricsEntitiesCountTest.java
new file mode 100644
index 0000000..d3490f4
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheMetricsEntitiesCountTest.java
@@ -0,0 +1,172 @@
+/*
+ * 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 org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CachePeekMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ * This test checks that entries count metrics, calculated by method
+ * {@link org.apache.ignite.internal.processors.cache.CacheMetricsImpl#getEntriesStat()} (which uses one iteration
+ * over local partitions to get all set of metrics), have the same values as metrics, calculated by individual methods
+ * (which use iteration over local partition per each method call).
+ */
+public class CacheMetricsEntitiesCountTest extends GridCommonAbstractTest {
+ /** Grid count. */
+ private static final int GRID_CNT = 3;
+
+ /** Cache prefix. */
+ private static final String CACHE_PREFIX = "CACHE";
+
+ /** Entities cnt. */
+ private static final int ENTITIES_CNT = 100;
+
+ /** Onheap peek modes. */
+ private static final CachePeekMode[] ONHEAP_PEEK_MODES = new CachePeekMode[] {
+ CachePeekMode.ONHEAP, CachePeekMode.PRIMARY, CachePeekMode.BACKUP, CachePeekMode.NEAR};
+
+ /** Cache count. */
+ private static final int CACHE_CNT = 4;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ CacheConfiguration<?, ?> ccfg0 = new CacheConfiguration<>()
+ .setName(CACHE_PREFIX + 0)
+ .setCacheMode(CacheMode.LOCAL);
+
+ CacheConfiguration<?, ?> ccfg1 = new CacheConfiguration<>()
+ .setName(CACHE_PREFIX + 1)
+ .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
+ .setCacheMode(CacheMode.REPLICATED);
+
+ CacheConfiguration<?, ?> ccfg2 = new CacheConfiguration<>()
+ .setName(CACHE_PREFIX + 2)
+ .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
+ .setCacheMode(CacheMode.PARTITIONED)
+ .setBackups(1);
+
+ CacheConfiguration<?, ?> ccfg3 = new CacheConfiguration<>()
+ .setName(CACHE_PREFIX + 3)
+ .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
+ .setCacheMode(CacheMode.PARTITIONED)
+ .setBackups(1)
+ .setNearConfiguration(new NearCacheConfiguration<>());
+
+ cfg.setCacheConfiguration(ccfg0, ccfg1, ccfg2, ccfg3);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ startGrids(GRID_CNT);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * Test entities count, calculated by different implementations.
+ */
+ public void testEnitiesCount() throws Exception {
+ awaitPartitionMapExchange();
+
+ for (int igniteIdx = 0; igniteIdx < GRID_CNT; igniteIdx++)
+ for (int cacheIdx = 0; cacheIdx < CACHE_CNT; cacheIdx++)
+ fillCache(igniteIdx, cacheIdx);
+
+ for (int igniteIdx = 0; igniteIdx < GRID_CNT; igniteIdx++)
+ for (int cacheIdx = 0; cacheIdx < CACHE_CNT; cacheIdx++)
+ checkCache(igniteIdx, cacheIdx);
+ }
+
+ /**
+ * @param igniteIdx Ignite index.
+ * @param cacheIdx Cache index.
+ */
+ private void fillCache(int igniteIdx, int cacheIdx) {
+ log.info("Filling cache, igniteIdx=" + igniteIdx + ", cacheIdx=" + cacheIdx);
+
+ IgniteCache cache = grid(igniteIdx).cache(CACHE_PREFIX + cacheIdx);
+
+ for (int i = 0; i < ENTITIES_CNT; i++)
+ cache.put("key" + igniteIdx + "-" + i, i);
+ }
+
+ /**
+ * @param igniteIdx Ignite index.
+ * @param cacheIdx Cache index.
+ */
+ private void checkCache(int igniteIdx, int cacheIdx) throws IgniteCheckedException {
+ IgniteInternalCache internalCache = grid(igniteIdx).cachex(CACHE_PREFIX + cacheIdx);
+
+ GridCacheContext cctx = internalCache.context();
+
+ GridCacheAdapter cache = cctx.cache();
+
+ CacheMetricsImpl metrics = cache.metrics0();
+
+ CacheMetricsImpl.EntriesStatMetrics entriesStatMetrics = metrics.getEntriesStat();
+
+ long offHeapEntriesCnt = cache.offHeapEntriesCount();
+
+ long offHeapPrimaryEntriesCnt = cctx.offheap().cacheEntriesCount(cctx.cacheId(),
+ true,
+ false,
+ cctx.affinity().affinityTopologyVersion());
+
+ long offHeapBackupEntriesCnt = cctx.offheap().cacheEntriesCount(cctx.cacheId(),
+ false,
+ true,
+ cctx.affinity().affinityTopologyVersion());
+
+ long heapEntriesCnt = cache.localSizeLong(ONHEAP_PEEK_MODES);
+
+ int size = cache.size();
+
+ int keySize = size;
+
+ boolean isEmpty = cache.isEmpty();
+
+ String cacheInfo = "igniteIdx=" + igniteIdx + ", cacheIdx=" + cacheIdx + " ";
+
+ log.info("Checking cache, " + cacheInfo);
+
+ assertEquals(cacheInfo + " offHeapEntriesCnt", offHeapEntriesCnt,
+ entriesStatMetrics.offHeapEntriesCount());
+ assertEquals(cacheInfo + " offHeapBackupEntriesCnt", offHeapBackupEntriesCnt,
+ entriesStatMetrics.offHeapBackupEntriesCount());
+ assertEquals(cacheInfo + " offHeapPrimaryEntriesCnt", offHeapPrimaryEntriesCnt,
+ entriesStatMetrics.offHeapPrimaryEntriesCount());
+ assertEquals(cacheInfo + " heapEntriesCnt", heapEntriesCnt, entriesStatMetrics.heapEntriesCount());
+ assertEquals(cacheInfo + " size", size, entriesStatMetrics.size());
+ assertEquals(cacheInfo + " keySize", keySize, entriesStatMetrics.keySize());
+ assertEquals(cacheInfo + " isEmpty", isEmpty, entriesStatMetrics.isEmpty());
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a9ad299c/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMetricsSelfTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMetricsSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMetricsSelfTestSuite.java
index b7e65e5..ac7dca2 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMetricsSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheMetricsSelfTestSuite.java
@@ -21,6 +21,7 @@ import junit.framework.TestSuite;
import org.apache.ignite.internal.processors.cache.CacheGroupMetricsMBeanTest;
import org.apache.ignite.internal.processors.cache.CacheGroupsMetricsRebalanceTest;
import org.apache.ignite.internal.processors.cache.CacheMetricsEnableRuntimeTest;
+import org.apache.ignite.internal.processors.cache.CacheMetricsEntitiesCountTest;
import org.apache.ignite.internal.processors.cache.CacheMetricsForClusterGroupSelfTest;
import org.apache.ignite.internal.processors.cache.CacheValidatorMetricsTest;
import org.apache.ignite.internal.processors.cache.OffheapCacheMetricsForClusterGroupSelfTest;
@@ -64,11 +65,10 @@ public class IgniteCacheMetricsSelfTestSuite extends TestSuite {
suite.addTestSuite(GridCacheAtomicLocalTckMetricsSelfTestImpl.class);
suite.addTestSuite(CacheGroupsMetricsRebalanceTest.class);
-
suite.addTestSuite(CacheGroupMetricsMBeanTest.class);
suite.addTestSuite(CacheValidatorMetricsTest.class);
-
suite.addTestSuite(CacheMetricsEnableRuntimeTest.class);
+ suite.addTestSuite(CacheMetricsEntitiesCountTest.class);
// Cluster wide metrics.
suite.addTestSuite(CacheMetricsForClusterGroupSelfTest.class);