Repository: ignite Updated Branches: refs/heads/master b10ba044d -> fd3c50e0d
IGNITE-9366: SQL: Added IGNITE.NODE_METRICS system view. This closes #4615. Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/fd3c50e0 Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/fd3c50e0 Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/fd3c50e0 Branch: refs/heads/master Commit: fd3c50e0df916cd636b98849d02d3a730fcac73d Parents: b10ba04 Author: Aleksey Plekhanov <[email protected]> Authored: Fri Sep 7 14:22:48 2018 +0300 Committer: devozerov <[email protected]> Committed: Fri Sep 7 14:22:48 2018 +0300 ---------------------------------------------------------------------- .../processors/query/h2/IgniteH2Indexing.java | 2 + .../h2/sys/view/SqlAbstractLocalSystemView.java | 27 +++ .../h2/sys/view/SqlSystemViewNodeMetrics.java | 210 +++++++++++++++++ .../query/SqlSystemViewsSelfTest.java | 226 +++++++++++++++++-- 4 files changed, 448 insertions(+), 17 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/fd3c50e0/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java ---------------------------------------------------------------------- diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java index 9b71f37..a795552 100644 --- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java +++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java @@ -130,6 +130,7 @@ import org.apache.ignite.internal.processors.query.h2.sql.GridSqlTable; import org.apache.ignite.internal.processors.query.h2.sys.SqlSystemTableEngine; import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewBaselineNodes; import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewNodeAttributes; +import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewNodeMetrics; import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewNodes; import org.apache.ignite.internal.processors.query.h2.twostep.GridMapQueryExecutor; import org.apache.ignite.internal.processors.query.h2.twostep.GridReduceQueryExecutor; @@ -3202,6 +3203,7 @@ public class IgniteH2Indexing implements GridQueryIndexing { views.add(new SqlSystemViewNodes(ctx)); views.add(new SqlSystemViewNodeAttributes(ctx)); views.add(new SqlSystemViewBaselineNodes(ctx)); + views.add(new SqlSystemViewNodeMetrics(ctx)); return views; } http://git-wip-us.apache.org/repos/asf/ignite/blob/fd3c50e0/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlAbstractLocalSystemView.java ---------------------------------------------------------------------- diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlAbstractLocalSystemView.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlAbstractLocalSystemView.java index ac90b63..d692dba 100644 --- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlAbstractLocalSystemView.java +++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlAbstractLocalSystemView.java @@ -26,6 +26,8 @@ import org.h2.table.Column; import org.h2.value.Value; import org.h2.value.ValueNull; import org.h2.value.ValueString; +import org.h2.value.ValueTime; +import org.h2.value.ValueTimestamp; /** * Local system view base class (which uses only local node data). @@ -125,4 +127,29 @@ public abstract class SqlAbstractLocalSystemView extends SqlAbstractSystemView { return null; } } + + /** + * Converts millis to ValueTime + * + * @param millis Millis. + */ + protected static Value valueTimeFromMillis(long millis) { + if (millis == -1L || millis == Long.MAX_VALUE) + return ValueNull.INSTANCE; + else + // Note: ValueTime.fromMillis(long) method trying to convert time using timezone and return wrong result. + return ValueTime.fromNanos(millis * 1_000_000L); + } + + /** + * Converts millis to ValueTimestamp + * + * @param millis Millis. + */ + protected static Value valueTimestampFromMillis(long millis) { + if (millis <= 0L || millis == Long.MAX_VALUE) + return ValueNull.INSTANCE; + else + return ValueTimestamp.fromMillis(millis); + } } http://git-wip-us.apache.org/repos/asf/ignite/blob/fd3c50e0/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlSystemViewNodeMetrics.java ---------------------------------------------------------------------- diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlSystemViewNodeMetrics.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlSystemViewNodeMetrics.java new file mode 100644 index 0000000..01b4e97 --- /dev/null +++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sys/view/SqlSystemViewNodeMetrics.java @@ -0,0 +1,210 @@ +/* + * 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.query.h2.sys.view; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.UUID; +import org.apache.ignite.cluster.ClusterMetrics; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.util.typedef.F; +import org.h2.engine.Session; +import org.h2.result.Row; +import org.h2.result.SearchRow; +import org.h2.value.Value; + +/** + * System view: node metrics. + */ +public class SqlSystemViewNodeMetrics extends SqlAbstractLocalSystemView { + /** + * @param ctx Grid context. + */ + public SqlSystemViewNodeMetrics(GridKernalContext ctx) { + super("NODE_METRICS", "Node metrics", ctx, new String[] {"NODE_ID"}, + newColumn("NODE_ID", Value.UUID), + newColumn("LAST_UPDATE_TIME", Value.TIMESTAMP), + newColumn("MAX_ACTIVE_JOBS", Value.INT), + newColumn("CUR_ACTIVE_JOBS", Value.INT), + newColumn("AVG_ACTIVE_JOBS", Value.FLOAT), + newColumn("MAX_WAITING_JOBS", Value.INT), + newColumn("CUR_WAITING_JOBS", Value.INT), + newColumn("AVG_WAITING_JOBS", Value.FLOAT), + newColumn("MAX_REJECTED_JOBS", Value.INT), + newColumn("CUR_REJECTED_JOBS", Value.INT), + newColumn("AVG_REJECTED_JOBS", Value.FLOAT), + newColumn("TOTAL_REJECTED_JOBS", Value.INT), + newColumn("MAX_CANCELED_JOBS", Value.INT), + newColumn("CUR_CANCELED_JOBS", Value.INT), + newColumn("AVG_CANCELED_JOBS", Value.FLOAT), + newColumn("TOTAL_CANCELED_JOBS", Value.INT), + newColumn("MAX_JOBS_WAIT_TIME", Value.TIME), + newColumn("CUR_JOBS_WAIT_TIME", Value.TIME), + newColumn("AVG_JOBS_WAIT_TIME", Value.TIME), + newColumn("MAX_JOBS_EXECUTE_TIME", Value.TIME), + newColumn("CUR_JOBS_EXECUTE_TIME", Value.TIME), + newColumn("AVG_JOBS_EXECUTE_TIME", Value.TIME), + newColumn("TOTAL_JOBS_EXECUTE_TIME", Value.TIME), + newColumn("TOTAL_EXECUTED_JOBS", Value.INT), + newColumn("TOTAL_EXECUTED_TASKS", Value.INT), + newColumn("TOTAL_BUSY_TIME", Value.TIME), + newColumn("TOTAL_IDLE_TIME", Value.TIME), + newColumn("CUR_IDLE_TIME", Value.TIME), + newColumn("BUSY_TIME_PERCENTAGE", Value.FLOAT), + newColumn("IDLE_TIME_PERCENTAGE", Value.FLOAT), + newColumn("TOTAL_CPU", Value.INT), + newColumn("CUR_CPU_LOAD", Value.DOUBLE), + newColumn("AVG_CPU_LOAD", Value.DOUBLE), + newColumn("CUR_GC_CPU_LOAD", Value.DOUBLE), + newColumn("HEAP_MEMORY_INIT", Value.LONG), + newColumn("HEAP_MEMORY_USED", Value.LONG), + newColumn("HEAP_MEMORY_COMMITED", Value.LONG), + newColumn("HEAP_MEMORY_MAX", Value.LONG), + newColumn("HEAP_MEMORY_TOTAL", Value.LONG), + newColumn("NONHEAP_MEMORY_INIT", Value.LONG), + newColumn("NONHEAP_MEMORY_USED", Value.LONG), + newColumn("NONHEAP_MEMORY_COMMITED", Value.LONG), + newColumn("NONHEAP_MEMORY_MAX", Value.LONG), + newColumn("NONHEAP_MEMORY_TOTAL", Value.LONG), + newColumn("UPTIME", Value.TIME), + newColumn("JVM_START_TIME", Value.TIMESTAMP), + newColumn("NODE_START_TIME", Value.TIMESTAMP), + newColumn("LAST_DATA_VERSION", Value.LONG), + newColumn("CUR_THREAD_COUNT", Value.INT), + newColumn("MAX_THREAD_COUNT", Value.INT), + newColumn("TOTAL_THREAD_COUNT", Value.LONG), + newColumn("CUR_DAEMON_THREAD_COUNT", Value.INT), + newColumn("SENT_MESSAGES_COUNT", Value.INT), + newColumn("SENT_BYTES_COUNT", Value.LONG), + newColumn("RECEIVED_MESSAGES_COUNT", Value.INT), + newColumn("RECEIVED_BYTES_COUNT", Value.LONG), + newColumn("OUTBOUND_MESSAGES_QUEUE", Value.INT) + ); + } + + /** {@inheritDoc} */ + @Override public Iterator<Row> getRows(Session ses, SearchRow first, SearchRow last) { + List<Row> rows = new ArrayList<>(); + + Collection<ClusterNode> nodes; + + SqlSystemViewColumnCondition idCond = conditionForColumn("NODE_ID", first, last); + + if (idCond.isEquality()) { + try { + UUID nodeId = uuidFromValue(idCond.valueForEquality()); + + ClusterNode node = nodeId == null ? null : ctx.discovery().node(nodeId); + + if (node != null) + nodes = Collections.singleton(node); + else + nodes = Collections.emptySet(); + } + catch (Exception e) { + nodes = Collections.emptySet(); + } + } + else + nodes = F.concat(false, ctx.discovery().allNodes(), ctx.discovery().daemonNodes()); + + for (ClusterNode node : nodes) { + if (node != null) { + ClusterMetrics metrics = node.metrics(); + + rows.add( + createRow(ses, rows.size(), + node.id(), + valueTimestampFromMillis(metrics.getLastUpdateTime()), + metrics.getMaximumActiveJobs(), + metrics.getCurrentActiveJobs(), + metrics.getAverageActiveJobs(), + metrics.getMaximumWaitingJobs(), + metrics.getCurrentWaitingJobs(), + metrics.getAverageWaitingJobs(), + metrics.getMaximumRejectedJobs(), + metrics.getCurrentRejectedJobs(), + metrics.getAverageRejectedJobs(), + metrics.getTotalRejectedJobs(), + metrics.getMaximumCancelledJobs(), + metrics.getCurrentCancelledJobs(), + metrics.getAverageCancelledJobs(), + metrics.getTotalCancelledJobs(), + valueTimeFromMillis(metrics.getMaximumJobWaitTime()), + valueTimeFromMillis(metrics.getCurrentJobWaitTime()), + valueTimeFromMillis((long)metrics.getAverageJobWaitTime()), + valueTimeFromMillis(metrics.getMaximumJobExecuteTime()), + valueTimeFromMillis(metrics.getCurrentJobExecuteTime()), + valueTimeFromMillis((long)metrics.getAverageJobExecuteTime()), + valueTimeFromMillis(metrics.getTotalJobsExecutionTime()), + metrics.getTotalExecutedJobs(), + metrics.getTotalExecutedTasks(), + valueTimeFromMillis(metrics.getTotalBusyTime()), + valueTimeFromMillis(metrics.getTotalIdleTime()), + valueTimeFromMillis(metrics.getCurrentIdleTime()), + metrics.getBusyTimePercentage(), + metrics.getIdleTimePercentage(), + metrics.getTotalCpus(), + metrics.getCurrentCpuLoad(), + metrics.getAverageCpuLoad(), + metrics.getCurrentGcCpuLoad(), + metrics.getHeapMemoryInitialized(), + metrics.getHeapMemoryUsed(), + metrics.getHeapMemoryCommitted(), + metrics.getHeapMemoryMaximum(), + metrics.getHeapMemoryTotal(), + metrics.getNonHeapMemoryInitialized(), + metrics.getNonHeapMemoryUsed(), + metrics.getNonHeapMemoryCommitted(), + metrics.getNonHeapMemoryMaximum(), + metrics.getNonHeapMemoryTotal(), + valueTimeFromMillis(metrics.getUpTime()), + valueTimestampFromMillis(metrics.getStartTime()), + valueTimestampFromMillis(metrics.getNodeStartTime()), + metrics.getLastDataVersion(), + metrics.getCurrentThreadCount(), + metrics.getMaximumThreadCount(), + metrics.getTotalStartedThreadCount(), + metrics.getCurrentDaemonThreadCount(), + metrics.getSentMessagesCount(), + metrics.getSentBytesCount(), + metrics.getReceivedMessagesCount(), + metrics.getReceivedBytesCount(), + metrics.getOutboundMessagesQueueSize() + ) + ); + } + } + + return rows.iterator(); + } + + /** {@inheritDoc} */ + @Override public boolean canGetRowCount() { + return true; + } + + /** {@inheritDoc} */ + @Override public long getRowCount() { + return F.concat(false, ctx.discovery().allNodes(), ctx.discovery().daemonNodes()).size(); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/fd3c50e0/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/SqlSystemViewsSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/SqlSystemViewsSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/SqlSystemViewsSelfTest.java index c8374a7..1a4dae7 100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/SqlSystemViewsSelfTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/SqlSystemViewsSelfTest.java @@ -17,21 +17,31 @@ package org.apache.ignite.internal.processors.query; +import java.sql.Time; +import java.sql.Timestamp; import java.util.Collections; import java.util.List; +import java.util.Random; +import java.util.TimeZone; import java.util.UUID; import java.util.concurrent.Callable; import org.apache.ignite.Ignite; import org.apache.ignite.IgniteCache; import org.apache.ignite.cache.QueryEntity; import org.apache.ignite.cache.query.SqlFieldsQuery; +import org.apache.ignite.cluster.ClusterMetrics; 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.ClusterMetricsSnapshot; import org.apache.ignite.internal.IgniteNodeAttributes; import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.G; import org.apache.ignite.internal.util.typedef.X; +import org.apache.ignite.lang.IgniteFuture; +import org.apache.ignite.lang.IgniteRunnable; import org.apache.ignite.testframework.GridTestUtils; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; @@ -39,6 +49,9 @@ import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; * Tests for ignite SQL system views. */ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { + /** Metrics check attempts. */ + private static final int METRICS_CHECK_ATTEMPTS = 10; + /** {@inheritDoc} */ @Override protected void beforeTest() throws Exception { super.beforeTest(); @@ -171,7 +184,7 @@ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { private void assertColumnTypes(List<?> rowData, Class<?> ... colTypes) { for (int i = 0; i < colTypes.length; i++) { if (rowData.get(i) != null) - assertEquals("Column " + i + " type", rowData.get(i).getClass(), colTypes[i]); + assertEquals("Column " + i + " type", colTypes[i], rowData.get(i).getClass()); } } @@ -181,9 +194,14 @@ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { * @throws Exception If failed. */ public void testNodesViews() throws Exception { - Ignite ignite1 = startGrid(getTestIgniteInstanceName(), getConfiguration()); - Ignite ignite2 = startGrid(getTestIgniteInstanceName(1), getConfiguration().setClientMode(true)); - Ignite ignite3 = startGrid(getTestIgniteInstanceName(2), getConfiguration().setDaemon(true)); + Ignite igniteSrv = startGrid(getTestIgniteInstanceName(), getConfiguration().setMetricsUpdateFrequency(500L)); + + Ignite igniteCli = startGrid(getTestIgniteInstanceName(1), getConfiguration().setMetricsUpdateFrequency(500L) + .setClientMode(true)); + + startGrid(getTestIgniteInstanceName(2), getConfiguration().setMetricsUpdateFrequency(500L).setDaemon(true)); + + UUID nodeId0 = igniteSrv.cluster().localNode().id(); awaitPartitionMapExchange(); @@ -201,7 +219,7 @@ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { assertEquals(1, resSrv.size()); - assertEquals(ignite1.cluster().localNode().id(), resSrv.get(0).get(0)); + assertEquals(nodeId0, resSrv.get(0).get(0)); assertEquals(1, resSrv.get(0).get(1)); @@ -210,7 +228,7 @@ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { assertEquals(1, resCli.size()); - assertEquals(ignite2.cluster().localNode().id(), resCli.get(0).get(0)); + assertEquals(nodeId(1), resCli.get(0).get(0)); assertEquals(2, resCli.get(0).get(1)); @@ -219,7 +237,7 @@ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { assertEquals(1, resDaemon.size()); - assertEquals(ignite3.cluster().localNode().id(), resDaemon.get(0).get(0)); + assertEquals(nodeId(2), resDaemon.get(0).get(0)); assertEquals(3, resDaemon.get(0).get(1)); @@ -227,32 +245,30 @@ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { assertEquals(0, execSql("SELECT ID FROM IGNITE.NODES WHERE ID = '-'").size()); assertEquals(1, execSql("SELECT ID FROM IGNITE.NODES WHERE ID = ?", - ignite1.cluster().localNode().id()).size()); + nodeId0).size()); assertEquals(1, execSql("SELECT ID FROM IGNITE.NODES WHERE ID = ?", - ignite3.cluster().localNode().id()).size()); + nodeId(2)).size()); // Check index on ID column with disjunction. assertEquals(3, execSql("SELECT ID FROM IGNITE.NODES WHERE ID = ? " + - "OR node_order=1 OR node_order=2 OR node_order=3", ignite1.cluster().localNode().id()).size()); + "OR node_order=1 OR node_order=2 OR node_order=3", nodeId0).size()); // Check quick-count. assertEquals(3L, execSql("SELECT COUNT(*) FROM IGNITE.NODES").get(0).get(0)); // Check joins - assertEquals(ignite1.cluster().localNode().id(), execSql("SELECT N1.ID FROM IGNITE.NODES N1 JOIN " + + assertEquals(nodeId0, execSql("SELECT N1.ID FROM IGNITE.NODES N1 JOIN " + "IGNITE.NODES N2 ON N1.NODE_ORDER = N2.NODE_ORDER JOIN IGNITE.NODES N3 ON N2.ID = N3.ID " + "WHERE N3.NODE_ORDER = 1") .get(0).get(0)); // Check sub-query - assertEquals(ignite1.cluster().localNode().id(), execSql("SELECT N1.ID FROM IGNITE.NODES N1 " + + assertEquals(nodeId0, execSql("SELECT N1.ID FROM IGNITE.NODES N1 " + "WHERE NOT EXISTS (SELECT 1 FROM IGNITE.NODES N2 WHERE N2.ID = N1.ID AND N2.NODE_ORDER <> 1)") .get(0).get(0)); // Check node attributes view - UUID cliNodeId = ignite2.cluster().localNode().id(); - String cliAttrName = IgniteNodeAttributes.ATTR_CLIENT_MODE; assertColumnTypes(execSql("SELECT NODE_ID, NAME, VALUE FROM IGNITE.NODE_ATTRIBUTES").get(0), @@ -267,7 +283,7 @@ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { assertEquals(1, execSql("SELECT NODE_ID FROM IGNITE.NODE_ATTRIBUTES WHERE NODE_ID = ? AND NAME = ? AND VALUE = 'true'", - cliNodeId, cliAttrName).size()); + nodeId(1), cliAttrName).size()); assertEquals(0, execSql("SELECT NODE_ID FROM IGNITE.NODE_ATTRIBUTES WHERE NODE_ID = '-' AND NAME = ?", @@ -275,7 +291,169 @@ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { assertEquals(0, execSql("SELECT NODE_ID FROM IGNITE.NODE_ATTRIBUTES WHERE NODE_ID = ? AND NAME = '-'", - cliNodeId).size()); + nodeId(1)).size()); + + // Check node metrics view. + String sqlAllMetrics = "SELECT NODE_ID, LAST_UPDATE_TIME, " + + "MAX_ACTIVE_JOBS, CUR_ACTIVE_JOBS, AVG_ACTIVE_JOBS, " + + "MAX_WAITING_JOBS, CUR_WAITING_JOBS, AVG_WAITING_JOBS, " + + "MAX_REJECTED_JOBS, CUR_REJECTED_JOBS, AVG_REJECTED_JOBS, TOTAL_REJECTED_JOBS, " + + "MAX_CANCELED_JOBS, CUR_CANCELED_JOBS, AVG_CANCELED_JOBS, TOTAL_CANCELED_JOBS, " + + "MAX_JOBS_WAIT_TIME, CUR_JOBS_WAIT_TIME, AVG_JOBS_WAIT_TIME, " + + "MAX_JOBS_EXECUTE_TIME, CUR_JOBS_EXECUTE_TIME, AVG_JOBS_EXECUTE_TIME, TOTAL_JOBS_EXECUTE_TIME, " + + "TOTAL_EXECUTED_JOBS, TOTAL_EXECUTED_TASKS, " + + "TOTAL_BUSY_TIME, TOTAL_IDLE_TIME, CUR_IDLE_TIME, BUSY_TIME_PERCENTAGE, IDLE_TIME_PERCENTAGE, " + + "TOTAL_CPU, CUR_CPU_LOAD, AVG_CPU_LOAD, CUR_GC_CPU_LOAD, " + + "HEAP_MEMORY_INIT, HEAP_MEMORY_USED, HEAP_MEMORY_COMMITED, HEAP_MEMORY_MAX, HEAP_MEMORY_TOTAL, " + + "NONHEAP_MEMORY_INIT, NONHEAP_MEMORY_USED, NONHEAP_MEMORY_COMMITED, NONHEAP_MEMORY_MAX, NONHEAP_MEMORY_TOTAL, " + + "UPTIME, JVM_START_TIME, NODE_START_TIME, LAST_DATA_VERSION, " + + "CUR_THREAD_COUNT, MAX_THREAD_COUNT, TOTAL_THREAD_COUNT, CUR_DAEMON_THREAD_COUNT, " + + "SENT_MESSAGES_COUNT, SENT_BYTES_COUNT, RECEIVED_MESSAGES_COUNT, RECEIVED_BYTES_COUNT, " + + "OUTBOUND_MESSAGES_QUEUE FROM IGNITE.NODE_METRICS"; + + List<List<?>> resMetrics = execSql(sqlAllMetrics); + + assertColumnTypes(resMetrics.get(0), UUID.class, Timestamp.class, + Integer.class, Integer.class, Float.class, // Active jobs. + Integer.class, Integer.class, Float.class, // Waiting jobs. + Integer.class, Integer.class, Float.class, Integer.class, // Rejected jobs. + Integer.class, Integer.class, Float.class, Integer.class, // Canceled jobs. + Time.class, Time.class, Time.class, // Jobs wait time. + Time.class, Time.class, Time.class, Time.class, // Jobs execute time. + Integer.class, Integer.class, // Executed jobs/task. + Time.class, Time.class, Time.class, Float.class, Float.class, // Busy/idle time. + Integer.class, Double.class, Double.class, Double.class, // CPU. + Long.class, Long.class, Long.class, Long.class, Long.class, // Heap memory. + Long.class, Long.class, Long.class, Long.class, Long.class, // Nonheap memory. + Time.class, Timestamp.class, Timestamp.class, Long.class, // Uptime. + Integer.class, Integer.class, Long.class, Integer.class, // Threads. + Integer.class, Long.class, Integer.class, Long.class, // Sent/received messages. + Integer.class); // Outbound message queue. + + assertEquals(3, resAll.size()); + + // Check join with nodes. + assertEquals(3, execSql("SELECT NM.LAST_UPDATE_TIME FROM IGNITE.NODES N " + + "JOIN IGNITE.NODE_METRICS NM ON N.ID = NM.NODE_ID").size()); + + // Check index on NODE_ID column. + assertEquals(1, execSql("SELECT LAST_UPDATE_TIME FROM IGNITE.NODE_METRICS WHERE NODE_ID = ?", + nodeId(1)).size()); + + // Check malformed value for indexed column. + assertEquals(0, execSql("SELECT LAST_UPDATE_TIME FROM IGNITE.NODE_METRICS WHERE NODE_ID = ?", + "-").size()); + + // Check quick-count. + assertEquals(3L, execSql("SELECT COUNT(*) FROM IGNITE.NODE_METRICS").get(0).get(0)); + + // Check metric values. + + // Broadcast jobs to server and client nodes to get non zero metric values. + for (int i = 0; i < 100; i++) { + IgniteFuture<Void > fut = igniteSrv.compute(igniteSrv.cluster().forNodeId(nodeId0, nodeId(1))) + .broadcastAsync( + new IgniteRunnable() { + @Override public void run() { + Random rnd = new Random(); + + try { + doSleep(rnd.nextInt(100)); + } + catch (Throwable ignore) { + // No-op. + } + } + }); + + if (i % 10 == 0) + fut.cancel(); + } + + doSleep(igniteSrv.configuration().getMetricsUpdateFrequency() * 3L); + + for (Ignite grid : G.allGrids()) { + UUID nodeId = grid.cluster().localNode().id(); + + // Metrics for node must be collected from another node to avoid race and get consistent metrics snapshot. + Ignite ignite = F.eq(nodeId, nodeId0) ? igniteCli : igniteSrv; + + for (int i = 0; i < METRICS_CHECK_ATTEMPTS; i++) { + ClusterMetrics metrics = ignite.cluster().node(nodeId).metrics(); + + assertTrue(metrics instanceof ClusterMetricsSnapshot); + + resMetrics = execSql(ignite, sqlAllMetrics + " WHERE NODE_ID = ?", nodeId); + + log.info("Check metrics for node " + grid.name() + ", attempt " + (i + 1)); + + if (metrics.getLastUpdateTime() == ((Timestamp)resMetrics.get(0).get(1)).getTime()) { + assertEquals(metrics.getMaximumActiveJobs(), resMetrics.get(0).get(2)); + assertEquals(metrics.getCurrentActiveJobs(), resMetrics.get(0).get(3)); + assertEquals(metrics.getAverageActiveJobs(), resMetrics.get(0).get(4)); + assertEquals(metrics.getMaximumWaitingJobs(), resMetrics.get(0).get(5)); + assertEquals(metrics.getCurrentWaitingJobs(), resMetrics.get(0).get(6)); + assertEquals(metrics.getAverageWaitingJobs(), resMetrics.get(0).get(7)); + assertEquals(metrics.getMaximumRejectedJobs(), resMetrics.get(0).get(8)); + assertEquals(metrics.getCurrentRejectedJobs(), resMetrics.get(0).get(9)); + assertEquals(metrics.getAverageRejectedJobs(), resMetrics.get(0).get(10)); + assertEquals(metrics.getTotalRejectedJobs(), resMetrics.get(0).get(11)); + assertEquals(metrics.getMaximumCancelledJobs(), resMetrics.get(0).get(12)); + assertEquals(metrics.getCurrentCancelledJobs(), resMetrics.get(0).get(13)); + assertEquals(metrics.getAverageCancelledJobs(), resMetrics.get(0).get(14)); + assertEquals(metrics.getTotalCancelledJobs(), resMetrics.get(0).get(15)); + assertEquals(metrics.getMaximumJobWaitTime(), convertToMilliseconds(resMetrics.get(0).get(16))); + assertEquals(metrics.getCurrentJobWaitTime(), convertToMilliseconds(resMetrics.get(0).get(17))); + assertEquals((long)metrics.getAverageJobWaitTime(), convertToMilliseconds(resMetrics.get(0).get(18))); + assertEquals(metrics.getMaximumJobExecuteTime(), convertToMilliseconds(resMetrics.get(0).get(19))); + assertEquals(metrics.getCurrentJobExecuteTime(), convertToMilliseconds(resMetrics.get(0).get(20))); + assertEquals((long)metrics.getAverageJobExecuteTime(), convertToMilliseconds(resMetrics.get(0).get(21))); + assertEquals(metrics.getTotalJobsExecutionTime(), convertToMilliseconds(resMetrics.get(0).get(22))); + assertEquals(metrics.getTotalExecutedJobs(), resMetrics.get(0).get(23)); + assertEquals(metrics.getTotalExecutedTasks(), resMetrics.get(0).get(24)); + assertEquals(metrics.getTotalBusyTime(), convertToMilliseconds(resMetrics.get(0).get(25))); + assertEquals(metrics.getTotalIdleTime(), convertToMilliseconds(resMetrics.get(0).get(26))); + assertEquals(metrics.getCurrentIdleTime(), convertToMilliseconds(resMetrics.get(0).get(27))); + assertEquals(metrics.getBusyTimePercentage(), resMetrics.get(0).get(28)); + assertEquals(metrics.getIdleTimePercentage(), resMetrics.get(0).get(29)); + assertEquals(metrics.getTotalCpus(), resMetrics.get(0).get(30)); + assertEquals(metrics.getCurrentCpuLoad(), resMetrics.get(0).get(31)); + assertEquals(metrics.getAverageCpuLoad(), resMetrics.get(0).get(32)); + assertEquals(metrics.getCurrentGcCpuLoad(), resMetrics.get(0).get(33)); + assertEquals(metrics.getHeapMemoryInitialized(), resMetrics.get(0).get(34)); + assertEquals(metrics.getHeapMemoryUsed(), resMetrics.get(0).get(35)); + assertEquals(metrics.getHeapMemoryCommitted(), resMetrics.get(0).get(36)); + assertEquals(metrics.getHeapMemoryMaximum(), resMetrics.get(0).get(37)); + assertEquals(metrics.getHeapMemoryTotal(), resMetrics.get(0).get(38)); + assertEquals(metrics.getNonHeapMemoryInitialized(), resMetrics.get(0).get(39)); + assertEquals(metrics.getNonHeapMemoryUsed(), resMetrics.get(0).get(40)); + assertEquals(metrics.getNonHeapMemoryCommitted(), resMetrics.get(0).get(41)); + assertEquals(metrics.getNonHeapMemoryMaximum(), resMetrics.get(0).get(42)); + assertEquals(metrics.getNonHeapMemoryTotal(), resMetrics.get(0).get(43)); + assertEquals(metrics.getUpTime(), convertToMilliseconds(resMetrics.get(0).get(44))); + assertEquals(metrics.getStartTime(), ((Timestamp)resMetrics.get(0).get(45)).getTime()); + assertEquals(metrics.getNodeStartTime(), ((Timestamp)resMetrics.get(0).get(46)).getTime()); + assertEquals(metrics.getLastDataVersion(), resMetrics.get(0).get(47)); + assertEquals(metrics.getCurrentThreadCount(), resMetrics.get(0).get(48)); + assertEquals(metrics.getMaximumThreadCount(), resMetrics.get(0).get(49)); + assertEquals(metrics.getTotalStartedThreadCount(), resMetrics.get(0).get(50)); + assertEquals(metrics.getCurrentDaemonThreadCount(), resMetrics.get(0).get(51)); + assertEquals(metrics.getSentMessagesCount(), resMetrics.get(0).get(52)); + assertEquals(metrics.getSentBytesCount(), resMetrics.get(0).get(53)); + assertEquals(metrics.getReceivedMessagesCount(), resMetrics.get(0).get(54)); + assertEquals(metrics.getReceivedBytesCount(), resMetrics.get(0).get(55)); + assertEquals(metrics.getOutboundMessagesQueueSize(), resMetrics.get(0).get(56)); + + break; + } + else { + log.info("Metrics was updated in background, will retry check"); + + if (i == METRICS_CHECK_ATTEMPTS - 1) + fail("Failed to check metrics, attempts limit reached (" + METRICS_CHECK_ATTEMPTS + ')'); + } + } + } } /** @@ -322,7 +500,7 @@ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { } /** - * Gets ignite configuration with persistance enabled. + * Gets ignite configuration with persistence enabled. */ private IgniteConfiguration getPdsConfiguration(String consistentId) throws Exception { IgniteConfiguration cfg = getConfiguration(); @@ -336,4 +514,18 @@ public class SqlSystemViewsSelfTest extends GridCommonAbstractTest { return cfg; } + + /** + * Convert Time to milliseconds. + * + * Note: Returned Time values from SQL it's milliseconds since January 1, 1970, 00:00:00 GMT. To get right interval + * in milliseconds this value must be adjusted to current time zone. + * + * @param sqlTime Time value returned from SQL. + */ + private long convertToMilliseconds(Object sqlTime) { + Time time0 = (Time)sqlTime; + + return time0.getTime() + TimeZone.getDefault().getOffset(time0.getTime()); + } }
