diff --git a/docs/_docs/monitoring-metrics/new-metrics-system.adoc b/docs/_docs/monitoring-metrics/new-metrics-system.adoc index 4bfd340f23877..094400c9f5b22 100644 --- a/docs/_docs/monitoring-metrics/new-metrics-system.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics-system.adoc @@ -419,7 +419,8 @@ The metrics belong to the `snapshot` register, see link:monitoring-metrics/new-m Metrics related to JDBC/ODBC or thin client connections. * `ActiveSessionsCount`, `thin.ActiveSessions`, `jdbc.ActiveSessions`, `odbc.ActiveSessions` - The number of active connections, total and per client type (`client.connector` register, see link:monitoring-metrics/new-metrics#ignite-thin-client-connector[Ignite Thin Client Connector]). -* The link:monitoring-metrics/system-views#client_connections[CLIENT_CONNECTIONS] system view lists every connection with its user and addresses. +* `MaxRunningComputeTasksPerConnection` - The maximum number of running user compute tasks per thin client connection over the configured time interval. The interval is 60 seconds by default and can be changed with `control.sh --metric --configure-max-value` (see link:tools/control-script#metric-configure-command[Metric configure command]). Set it longer than the polling period of your monitoring system, so peaks between polls are not missed (`client.connector` register, see link:monitoring-metrics/new-metrics#ignite-thin-client-connector[Ignite Thin Client Connector]). +* The link:monitoring-metrics/system-views#client_connections[CLIENT_CONNECTIONS] system view lists every connection with its user, addresses and the current number of active user compute tasks. * `Connections` - Attribute of the `ClientProcessorMXBean` MBean, not available in the metric registers. A list of strings, each string containing information about a connection: .... diff --git a/docs/_docs/monitoring-metrics/new-metrics.adoc b/docs/_docs/monitoring-metrics/new-metrics.adoc index 435b7741ae23c..b055dc4069544 100644 --- a/docs/_docs/monitoring-metrics/new-metrics.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics.adoc @@ -471,6 +471,7 @@ Register name: `client.connector` |AffinityKeyRequestsMisses | long | The number of affinity-aware cache key requests that were sent not to the primary node. |AffinityQueryRequestsHits | long | The number of affinity-aware query requests that were sent to the primary node. |AffinityQueryRequestsMisses | long | The number of affinity-aware query requests that were sent not to the primary node. +|MaxRunningComputeTasksPerConnection | long | Maximum number of running user compute tasks per thin client connection over the configured time interval (60 seconds by default). |RejectedSessionsAuthenticationFailed | integer | TCP sessions count that were rejected due to failed authentication. |RejectedSessionsTimeout | integer | TCP sessions count that were rejected due to handshake timeout. |RejectedSessionsTotal | integer | Total number of rejected TCP connections. diff --git a/docs/_docs/monitoring-metrics/system-views.adoc b/docs/_docs/monitoring-metrics/system-views.adoc index ea9cd2117ab37..b17b6d78ebc38 100644 --- a/docs/_docs/monitoring-metrics/system-views.adoc +++ b/docs/_docs/monitoring-metrics/system-views.adoc @@ -444,6 +444,7 @@ This view exposes information about currently opened client connections: JDBC, O |USER | string | User name |VERSION | string | Protocol version |DATA_CENTER_ID | string | Effective data center ID reported by the client +|ACTIVE_COMPUTE_TASKS | int | Number of active user compute tasks of a thin client connection, `null` for JDBC and ODBC connections |=== == CLIENT_CONNECTION_ATTRIBUTES diff --git a/docs/_docs/tools/control-script.adoc b/docs/_docs/tools/control-script.adoc index f4ba219555317..6f7541afa5427 100644 --- a/docs/_docs/tools/control-script.adoc +++ b/docs/_docs/tools/control-script.adoc @@ -1095,7 +1095,8 @@ Command [METRIC] finished with code: 0 == Metric configure command -The metrics command configure bounds of histogram metrics or rate time interval of hitrate metric. +The metrics command configure bounds of histogram metrics, rate time interval of hitrate metric or time interval of max value metric. +Time intervals are set in milliseconds. [tabs] -- @@ -1104,12 +1105,14 @@ tab:Unix[] ---- control.sh --metric --configure-histogram histogram-metric-name 1,2,3 control.sh --metric --configure-hitrate hitrate-metric-name 1000 +control.sh --metric --configure-max-value max-value-metric-name 60000 ---- tab:Windows[] [source,shell,subs="verbatim,quotes"] ---- control.bat --metric --configure-histogram histogram-metric-name 1,2,3 control.bat --metric --configure-hitrate hitrate-metric-name 1000 +control.bat --metric --configure-max-value max-value-metric-name 60000 ---- -- diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java index d4dd8d9400528..1d3686450a19a 100644 --- a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java +++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java @@ -951,6 +951,7 @@ public void testGetAllColumns() throws Exception { "SYS.CLIENT_CONNECTIONS.USER.null", "SYS.CLIENT_CONNECTIONS.VERSION.null", "SYS.CLIENT_CONNECTIONS.DATA_CENTER_ID.null", + "SYS.CLIENT_CONNECTIONS.ACTIVE_COMPUTE_TASKS.null", "SYS.CLIENT_CONNECTION_ATTRIBUTES.CONNECTION_ID.null", "SYS.CLIENT_CONNECTION_ATTRIBUTES.NAME.null", "SYS.CLIENT_CONNECTION_ATTRIBUTES.VALUE.null", diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/SystemViewCommandTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/SystemViewCommandTest.java index 2e950c2eb35e8..3d485c6fdd88b 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/util/SystemViewCommandTest.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/util/SystemViewCommandTest.java @@ -603,7 +603,9 @@ public void testViewColumns() { asList("USER", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", String.class.getName()), asList("VERSION", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", String.class.getName()), asList("DATA_CENTER_ID", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", - String.class.getName()) + String.class.getName()), + asList("ACTIVE_COMPUTE_TASKS", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", + Integer.class.getName()) )); Set> sqlViewColumnsView = systemView(ignite0, SQL_VIEW_COLS_VIEW).stream() diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java index 808437eb1719a..1186bc3427c2b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java @@ -29,6 +29,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Function; +import java.util.function.LongSupplier; import java.util.function.Predicate; import javax.cache.configuration.Factory; import javax.management.JMException; @@ -40,13 +41,16 @@ import org.apache.ignite.configuration.IgniteConfiguration; import org.apache.ignite.configuration.OdbcConfiguration; import org.apache.ignite.configuration.SqlConnectorConfiguration; +import org.apache.ignite.configuration.ThinClientConfiguration; import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.processors.GridProcessorAdapter; import org.apache.ignite.internal.processors.configuration.distributed.DistributedBooleanProperty; import org.apache.ignite.internal.processors.configuration.distributed.DistributedThinClientConfiguration; import org.apache.ignite.internal.processors.metric.MetricRegistryImpl; +import org.apache.ignite.internal.processors.metric.impl.MaxValueMetric; import org.apache.ignite.internal.processors.odbc.jdbc.JdbcConnectionContext; import org.apache.ignite.internal.processors.odbc.odbc.OdbcConnectionContext; +import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext; import org.apache.ignite.internal.systemview.ClientConnectionAttributeViewWalker; import org.apache.ignite.internal.systemview.ClientConnectionViewWalker; import org.apache.ignite.internal.util.GridSpinBusyLock; @@ -62,7 +66,6 @@ import org.apache.ignite.internal.util.worker.GridWorker; import org.apache.ignite.internal.util.worker.GridWorkerPool; import org.apache.ignite.lang.IgniteBiInClosure; -import org.apache.ignite.metric.MetricRegistry; import org.apache.ignite.mxbean.ClientProcessorMXBean; import org.apache.ignite.plugin.security.SecurityPermission; import org.apache.ignite.spi.IgnitePortProtocol; @@ -73,6 +76,7 @@ import static org.apache.ignite.internal.cluster.DistributedConfigurationUtils.newConnectionEnabledProperty; import static org.apache.ignite.internal.processors.metric.GridMetricManager.CLIENT_CONNECTOR_METRICS; +import static org.apache.ignite.internal.processors.metric.impl.AbstractIntervalMetric.DFLT_SIZE; import static org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName; import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.clientTypeLabel; import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.CLI_TYPES; @@ -100,6 +104,12 @@ public class ClientListenerProcessor extends GridProcessorAdapter { /** */ public static final String METRIC_ACTIVE = "ActiveSessions"; + /** */ + public static final String METRIC_MAX_RUNNING_COMPUTE_TASKS = "MaxRunningComputeTasksPerConnection"; + + /** Default time interval of the {@link #METRIC_MAX_RUNNING_COMPUTE_TASKS} metric in milliseconds. */ + public static final long DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL = 60_000L; + /** Default client connector configuration. */ public static final ClientConnectorConfiguration DFLT_CLI_CFG = new ClientConnectorConfigurationEx(); @@ -121,6 +131,9 @@ public class ClientListenerProcessor extends GridProcessorAdapter { /** Metrics. */ private ClientListenerMetrics metrics; + /** */ + private MaxValueMetric maxRunningComputeTasksMetric; + /** Executor service. */ private ExecutorService execSvc; @@ -327,7 +340,7 @@ private Iterable connectionAttributeViewSupplier( } /** @param mreg Metric registry. */ - private void registerClientMetrics(MetricRegistry mreg) { + private void registerClientMetrics(MetricRegistryImpl mreg) { for (int i = 0; i < CLI_TYPES.length; i++) { byte cliType = CLI_TYPES[i]; @@ -350,6 +363,36 @@ private void registerClientMetrics(MetricRegistry mreg) { "Number of active sessions for the " + cliTypeName + " client." ); } + + ThinClientConfiguration thinCfg = cliConnCfg.getThinClientConfiguration(); + + int maxActiveComputeTasks = thinCfg == null + ? ThinClientConfiguration.DFLT_MAX_ACTIVE_COMPUTE_TASKS_PER_CONNECTION + : thinCfg.getMaxActiveComputeTasksPerConnection(); + + maxRunningComputeTasksMetric = new MaxWithCurrentValueMetric( + metricName(mreg.name(), METRIC_MAX_RUNNING_COMPUTE_TASKS), + "Maximum number of running user compute tasks per thin client connection over the configured time interval", + DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL, + DFLT_SIZE, + maxActiveComputeTasks > 0 ? this::maxRunningComputeTasks : () -> 0L); + + mreg.register(maxRunningComputeTasksMetric); + } + + /** @return Maximum number of currently running user compute tasks among all thin client connections. */ + private long maxRunningComputeTasks() { + long max = 0; + + for (GridNioSession ses : srv.sessions()) { + ClientListenerConnectionContext connCtx = ses.meta(CONN_CTX_META_KEY); + + // Context is null before the handshake. JDBC and ODBC connections are skipped. + if (connCtx instanceof ClientConnectionContext) + max = Math.max(max, ((ClientConnectionContext)connCtx).activeTasksCount()); + } + + return max; } /** @@ -615,6 +658,11 @@ public ClientListenerMetrics metrics() { return metrics; } + /** @param runningTasksCnt Number of running user compute tasks of a connection. */ + public void updateMaxRunningComputeTasks(int runningTasksCnt) { + maxRunningComputeTasksMetric.update(runningTasksCnt); + } + /** * Prepare connector configuration. * @@ -853,4 +901,22 @@ public class ClientProcessorMXBeanImpl implements ClientProcessorMXBean { } } } + + /** Maximum value over the time interval, or the current value if it is greater. */ + private static class MaxWithCurrentValueMetric extends MaxValueMetric { + /** */ + private final LongSupplier cur; + + /** */ + MaxWithCurrentValueMetric(String name, String desc, long timeInterval, int size, LongSupplier cur) { + super(name, desc, timeInterval, size); + + this.cur = cur; + } + + /** {@inheritDoc} */ + @Override public long value() { + return Math.max(super.value(), cur.getAsLong()); + } + } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java index 3225d52e53dd7..15daf40f10286 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java @@ -349,14 +349,21 @@ public void incrementActiveTasksCount() { "To enable it set up the ThinClientConfiguration.MaxActiveComputeTasksPerConnection property."); } - if (activeTasksCnt.incrementAndGet() > maxActiveComputeTasks) { + int cnt = activeTasksCnt.incrementAndGet(); + + if (cnt > maxActiveComputeTasks) { activeTasksCnt.decrementAndGet(); + // A rejected task counts as the limit. + ctx.clientListener().updateMaxRunningComputeTasks(maxActiveComputeTasks); + throw new IgniteClientException(ClientStatus.TOO_MANY_COMPUTE_TASKS, "Active compute tasks per connection " + "limit (" + maxActiveComputeTasks + ") exceeded. To start a new task you need to wait for some of " + "currently active tasks complete. To change the limit set up the " + "ThinClientConfiguration.MaxActiveComputeTasksPerConnection property."); } + + ctx.clientListener().updateMaxRunningComputeTasks(cnt); } /** @@ -367,4 +374,10 @@ public void decrementActiveTasksCount() { assert cnt >= 0 : "Unexpected active tasks count: " + cnt; } + + /** */ + public int activeTasksCount() { + // The counter briefly exceeds the limit while a task is being rejected, so the value is capped by the limit. + return Math.min(activeTasksCnt.get(), maxActiveComputeTasks); + } } diff --git a/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ClientConnectionView.java b/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ClientConnectionView.java index 6ad4d299f13d4..e7c0a248b745f 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ClientConnectionView.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ClientConnectionView.java @@ -119,4 +119,13 @@ public String version() { @Nullable public String dataCenterId() { return ctx == null ? null : ctx.attributes().get(IGNITE_DATA_CENTER_ID); } + + /** + * @return Number of active user compute tasks of the thin client connection, + * {@code null} for other connection types. + */ + @Order(7) + @Nullable public Integer activeComputeTasks() { + return ctx instanceof ClientConnectionContext ? ((ClientConnectionContext)ctx).activeTasksCount() : null; + } } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java index c2e998c303910..bec202842560c 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java @@ -17,26 +17,46 @@ package org.apache.ignite.internal.processors.odbc; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; +import org.apache.ignite.IgniteException; import org.apache.ignite.Ignition; import org.apache.ignite.client.ClientAuthenticationException; import org.apache.ignite.client.ClientException; import org.apache.ignite.client.Config; import org.apache.ignite.client.IgniteClient; +import org.apache.ignite.client.IgniteClientFuture; import org.apache.ignite.client.SslMode; +import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.cluster.ClusterState; +import org.apache.ignite.compute.ComputeJob; +import org.apache.ignite.compute.ComputeJobAdapter; +import org.apache.ignite.compute.ComputeJobResult; +import org.apache.ignite.compute.ComputeTaskAdapter; import org.apache.ignite.configuration.ClientConfiguration; import org.apache.ignite.configuration.ClientConnectorConfiguration; import org.apache.ignite.configuration.DataRegionConfiguration; import org.apache.ignite.configuration.DataStorageConfiguration; import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.configuration.ThinClientConfiguration; import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInterruptedCheckedException; +import org.apache.ignite.internal.processors.metric.impl.MaxValueMetric; import org.apache.ignite.internal.processors.metric.impl.MetricUtils; -import org.apache.ignite.internal.util.lang.GridAbsPredicate; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.metric.MetricRegistry; import org.apache.ignite.spi.metric.IntMetric; +import org.apache.ignite.spi.systemview.view.ClientConnectionView; +import org.apache.ignite.spi.systemview.view.SystemView; import org.apache.ignite.testframework.GridTestUtils; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.jetbrains.annotations.Nullable; import org.junit.Test; import static org.apache.ignite.internal.processors.metric.GridMetricManager.CLIENT_CONNECTOR_METRICS; @@ -44,13 +64,34 @@ import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.METRIC_REJECTED_AUTHENTICATION; import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.METRIC_REJECTED_TIMEOUT; import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.METRIC_REJECTED_TOTAL; +import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.CLI_CONN_VIEW; import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.METRIC_ACTIVE; +import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.METRIC_MAX_RUNNING_COMPUTE_TASKS; +import static org.apache.ignite.internal.util.GridTestClockTimer.DFLT_TIME_SUPPLIER; +import static org.apache.ignite.internal.util.GridTestClockTimer.timeSupplier; import static org.apache.ignite.ssl.SslContextFactory.DFLT_STORE_TYPE; +import static org.apache.ignite.testframework.GridTestUtils.assertThrowsWithCause; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; /** * Client listener metrics tests. */ public class ClientListenerMetricsTest extends GridCommonAbstractTest { + /** */ + private static final int MAX_TASKS = 5; + + /** Metric time interval. Expiration is emulated by a time shift. */ + private static final long LONG_INTERVAL = 10 * 60_000L; + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + TestComputeTask.reset(); + + stopAllGrids(); + + super.afterTest(); + } + /** * Check that valid connections and disconnections to the grid affect metrics. */ @@ -160,6 +201,265 @@ public void testClientListenerMetricsRejectGeneral() throws Exception { } } + /** + * Check that the running compute tasks metric reports the maximum among all connections and keeps it + * during the time interval. + */ + @Test + public void testComputeTasksMetrics() throws Exception { + startGridWithCompute(); + + List> futs = new ArrayList<>(); + + try ( + IgniteClient client1 = Ignition.startClient(getClientConfiguration()); + IgniteClient client2 = Ignition.startClient(getClientConfiguration()) + ) { + checkMaxComputeTasks(0); + assertEquals(0, activeComputeTasksCount()); + + int startCnt = 2; + + // Run tasks on the first connection. + futs.addAll(startTasks(client1, startCnt)); + checkMaxComputeTasks(startCnt); + assertEquals(startCnt, activeComputeTasksCount()); + + // The same number of tasks on the second connection. Max metric should not change. + futs.addAll(startTasks(client2, startCnt)); + checkMaxComputeTasks(startCnt); + assertEquals(startCnt * 2, activeComputeTasksCount()); + + // Run tasks on the second connection up to the limit. + futs.addAll(startTasks(client2, MAX_TASKS - startCnt)); + checkMaxComputeTasks(MAX_TASKS); + + int allTasksCnt = MAX_TASKS + startCnt; + assertEquals(allTasksCnt, activeComputeTasksCount()); + + // Max metric should not exceed limit after a rejected task. + assertComputeTaskRejected(client2); + checkMaxComputeTasks(MAX_TASKS); + assertEquals(allTasksCnt, activeComputeTasksCount()); + + // Finished tasks should not reset the maximum. + finishTasks(IntStream.range(0, allTasksCnt).toArray()); + checkMaxComputeTasks(MAX_TASKS); + + // "Wait" for expiration. + shiftTime(LONG_INTERVAL); + assertEquals(0, computeTasksMetric().value()); + + // Run one more task. + futs.addAll(startTasks(client1, 1)); + checkMaxComputeTasks(1); + + finishTasks(allTasksCnt); + + shiftTime(2 * LONG_INTERVAL); + checkMaxComputeTasks(0); + + for (IgniteClientFuture fut : futs) + fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); + } + finally { + timeSupplier(DFLT_TIME_SUPPLIER); + } + } + + /** Check that the metric is not less than the current maximum, when metric has expired. */ + @Test + public void testComputeTasksMetricOnLongRunningTasks() throws Exception { + startGridWithCompute(); + + try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { + List> futs = startTasks(client, MAX_TASKS); + + checkMaxComputeTasks(MAX_TASKS); + + // "Wait" for expiration. + shiftTime(LONG_INTERVAL); + assertEquals(MAX_TASKS, computeTasksMetric().value()); + + finishTasks(IntStream.range(0, MAX_TASKS).toArray()); + + assertEquals(0, computeTasksMetric().value()); + + for (IgniteClientFuture fut : futs) + fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); + } + finally { + timeSupplier(DFLT_TIME_SUPPLIER); + } + } + + /** */ + @Test + public void testComputeTasksMetricOnRejectedTask() throws Exception { + startGridWithCompute(); + + try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { + List> futs = startTasks(client, MAX_TASKS); + + checkMaxComputeTasks(MAX_TASKS); + + // "Wait" for expiration. + shiftTime(LONG_INTERVAL); + + assertComputeTaskRejected(client); + + finishTasks(0); + + // The metric keeps the max limit, though the current value is less. + checkMaxComputeTasks(MAX_TASKS); + + finishTasks(IntStream.range(1, MAX_TASKS).toArray()); + + for (IgniteClientFuture fut : futs) + fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); + } + finally { + timeSupplier(DFLT_TIME_SUPPLIER); + } + } + + /** */ + @Test + public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { + startGridWithCompute(); + + try (IgniteClient client1 = Ignition.startClient(getClientConfiguration())) { + List> futs1 = startTasks(client1, 1); + + assertEquals(1, activeComputeTasksCount()); + + List> futs2; + + try (IgniteClient client2 = Ignition.startClient(getClientConfiguration())) { + futs2 = startTasks(client2, 2); + + checkMaxComputeTasks(2); + + assertEquals(3, activeComputeTasksCount()); + } + + // Tasks of the closed client should not be counted in system views. + assertTrue(waitForCondition(() -> activeComputeTasksCount() == 1, getTestTimeout())); + + // Finish tasks of the closed client. + TestComputeTask.finish(1); + TestComputeTask.finish(2); + + // Unfinished tasks of client2 fail. + for (IgniteClientFuture fut : futs2) + assertThrowsWithCause(() -> fut.get(getTestTimeout(), TimeUnit.MILLISECONDS), ClientException.class); + + // Finish task of alive client. + finishTasks(0); + + // The metric keeps the maximum during the time interval. + checkMaxComputeTasks(2); + + for (IgniteClientFuture fut : futs1) + fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); + } + } + + /** */ + @Test + public void testComputeTasksMetricsComputeDisabled() throws Exception { + startGrid(0); + + try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { + checkMaxComputeTasks(0); + + GridTestUtils.assertThrowsAnyCause( + log, + () -> client.compute().execute(TestComputeTask.class.getName(), null), + ClientException.class, + "Compute grid functionality is disabled for thin clients" + ); + + checkMaxComputeTasks(0); + } + } + + /** + * Starts a node with compute enabled for thin clients and sets the metric time interval to {@link #LONG_INTERVAL}. + */ + private void startGridWithCompute() throws Exception { + startGrid(getConfiguration(getTestIgniteInstanceName(0)) + .setClientConnectorConfiguration(new ClientConnectorConfiguration() + .setThinClientConfiguration(new ThinClientConfiguration() + .setMaxActiveComputeTasksPerConnection(MAX_TASKS)))); + + computeTasksMetric().reset(LONG_INTERVAL); + } + + /** */ + private void assertComputeTaskRejected(IgniteClient client) { + GridTestUtils.assertThrowsAnyCause( + log, + () -> client.compute().execute(TestComputeTask.class.getName(), null), + ClientException.class, + "Active compute tasks per connection limit (" + MAX_TASKS + ") exceeded" + ); + } + + /** */ + private List> startTasks(IgniteClient client, int cnt) throws Exception { + int expTasks = TestComputeTask.TASKS.size() + cnt; + + List> futs = new ArrayList<>(); + + for (int i = 0; i < cnt; i++) + futs.add(client.compute().executeAsync2(TestComputeTask.class.getName(), null)); + + // Waits for tasks to be started on the nodes. + assertTrue(waitForCondition(() -> TestComputeTask.TASKS.size() == expTasks, getTestTimeout())); + + return futs; + } + + /** Finishes tasks and waits for them to be released by the connections. */ + private void finishTasks(int... idxs) throws Exception { + int expTasks = activeComputeTasksCount() - idxs.length; + + for (int idx : idxs) + TestComputeTask.finish(idx); + + // Counter is decremented after the result is sent to the client. + assertTrue(waitForCondition(() -> activeComputeTasksCount() == expTasks, getTestTimeout())); + } + + /** @return Total active compute tasks of the thin client connections. */ + private int activeComputeTasksCount() { + return thinConnections().stream().mapToInt(ClientConnectionView::activeComputeTasks).sum(); + } + + /** */ + private void checkMaxComputeTasks(long expMetric) throws Exception { + assertTrue(waitForCondition(() -> computeTasksMetric().value() == expMetric, getTestTimeout())); + } + + /** */ + private MaxValueMetric computeTasksMetric() { + return grid(0).context().metric().registry(CLIENT_CONNECTOR_METRICS) + .findMetric(METRIC_MAX_RUNNING_COMPUTE_TASKS); + } + + /** Shifts the time returned by {@link U#currentTimeMillis()} relative to the system time. */ + private static void shiftTime(long offset) { + timeSupplier(() -> System.currentTimeMillis() + offset); + } + + /** */ + private List thinConnections() { + SystemView view = grid(0).context().systemView().view(CLI_CONN_VIEW); + + return U.arrayList(view.iterator(), view.size(), v -> "THIN".equals(v.type())); + } + /** */ private static ClientConfiguration getClientConfiguration() { return new ClientConfiguration() @@ -179,11 +479,8 @@ private static ClientConfiguration getClientConfiguration() { */ private void waitForMetricValue(MetricRegistry mreg, String metric, long value, long timeout) throws IgniteInterruptedCheckedException { - GridTestUtils.waitForCondition(new GridAbsPredicate() { - @Override public boolean apply() { - return mreg.findMetric(metric).value() == value; - } - }, timeout); + waitForCondition(() -> mreg.findMetric(metric).value() == value, timeout); + assertEquals(mreg.findMetric(metric).value(), value); } @@ -217,4 +514,52 @@ private void checkConnectionsMetrics(MetricRegistry mreg, int accepted, int acti assertEquals(0, mreg.findMetric(METRIC_REJECTED_AUTHENTICATION).value()); assertEquals(0, mreg.findMetric(METRIC_REJECTED_TOTAL).value()); } + + /** Task that waits for the latch on reduce. */ + public static class TestComputeTask extends ComputeTaskAdapter { + /** */ + private static final List TASKS = new CopyOnWriteArrayList<>(); + + /** */ + private final CountDownLatch finishLatch = new CountDownLatch(1); + + /** {@inheritDoc} */ + @Override public Map map(List subgrid, @Nullable Object arg) { + TASKS.add(this); + + return F.asMap(new NoopJob(), subgrid.get(0)); + } + + /** {@inheritDoc} */ + @Override public @Nullable Object reduce(List results) { + try { + finishLatch.await(); + } + catch (InterruptedException e) { + throw new IgniteException(e); + } + + return null; + } + + /** @param idx Index of a task. */ + public static void finish(int idx) { + TASKS.get(idx).finishLatch.countDown(); + } + + /** Finishes all tasks and clears the task list. */ + public static void reset() { + TASKS.forEach(t -> t.finishLatch.countDown()); + + TASKS.clear(); + } + } + + /** */ + private static class NoopJob extends ComputeJobAdapter { + /** {@inheritDoc} */ + @Override public Object execute() { + return null; + } + } } diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/metric/SqlViewExporterSpiTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/metric/SqlViewExporterSpiTest.java index 848e733d18f0e..374cfdbb48b29 100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/metric/SqlViewExporterSpiTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/metric/SqlViewExporterSpiTest.java @@ -329,11 +329,14 @@ public void testClientsConnections() throws Exception { try (IgniteClient client = Ignition.startClient(new ClientConfiguration().setAddresses(host + ":" + port) .setUserAttributes(F.asMap(IGNITE_DATA_CENTER_ID, dcId)))) { try (Connection conn = new IgniteJdbcThinDriver().connect("jdbc:ignite:thin://" + host, new Properties())) { - List> conns = execute(ignite0, "SELECT TYPE, DATA_CENTER_ID FROM SYS.CLIENT_CONNECTIONS"); + List> conns = execute(ignite0, + "SELECT TYPE, DATA_CENTER_ID, ACTIVE_COMPUTE_TASKS FROM SYS.CLIENT_CONNECTIONS"); assertEquals(2, conns.size()); - assertTrue(conns.stream().anyMatch(row -> "THIN".equals(row.get(0)) && dcId.equals(row.get(1)))); - assertTrue(conns.stream().anyMatch(row -> "JDBC".equals(row.get(0)) && row.get(1) == null)); + assertTrue(conns.stream().anyMatch(row -> + "THIN".equals(row.get(0)) && dcId.equals(row.get(1)) && Integer.valueOf(0).equals(row.get(2)))); + assertTrue(conns.stream().anyMatch(row -> + "JDBC".equals(row.get(0)) && row.get(1) == null && row.get(2) == null)); } } } @@ -576,7 +579,8 @@ public void testViewColumns() { asList("TYPE", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, String.class.getName()), asList("USER", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, String.class.getName()), asList("VERSION", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, String.class.getName()), - asList("DATA_CENTER_ID", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, String.class.getName()) + asList("DATA_CENTER_ID", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, String.class.getName()), + asList("ACTIVE_COMPUTE_TASKS", "CLIENT_CONNECTIONS", SCHEMA_SYS, null, true, -1, -1, Integer.class.getName()) ); List> res = execute(ignite0, "SELECT * FROM SYS.VIEW_COLUMNS WHERE VIEW_NAME = 'CLIENT_CONNECTIONS'");