diff --git a/docs/_docs/monitoring-metrics/new-metrics-system.adoc b/docs/_docs/monitoring-metrics/new-metrics-system.adoc index 4bfd340f23877..c70a6466985da 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. +* `PeakActiveComputeTasksPerConnection` - The maximum number of active user compute tasks per thin client connection over the configured time interval, but not less than the current value. The upper limit is the `ThinClientConfiguration.maxActiveComputeTasksPerConnection` configuration parameter. A task rejected by the limit counts as the limit. The interval is 90 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 dfc54ceb1312b..5e8e2a128715f 100644 --- a/docs/_docs/monitoring-metrics/new-metrics.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics.adoc @@ -466,6 +466,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. +|PeakActiveComputeTasksPerConnection | long | Maximum number of active user compute tasks per thin client connection over the configured time interval (90 seconds by default), but not less than the current value. The upper limit is the `ThinClientConfiguration.maxActiveComputeTasksPerConnection` configuration parameter. A task rejected by the limit counts as the limit. |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 969b9f2061944..9b2a41db6bc19 100644 --- a/docs/_docs/monitoring-metrics/system-views.adoc +++ b/docs/_docs/monitoring-metrics/system-views.adoc @@ -443,6 +443,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, not greater than the `ThinClientConfiguration.maxActiveComputeTasksPerConnection` configuration parameter. `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 13e5438399a60..aab72e872b129 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..07bf3b6372d51 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"; + /** Peak number of active user compute tasks per thin client connection. */ + public static final String METRIC_PEAK_ACTIVE_COMPUTE_TASKS = "PeakActiveComputeTasksPerConnection"; + + /** Default time interval of the {@link #METRIC_PEAK_ACTIVE_COMPUTE_TASKS} metric in milliseconds. */ + public static final long DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL = 90_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; + /** Peak of active user compute tasks per thin client connection. */ + private PeakWithCurrentMetric peakActiveComputeTasks; + /** 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,41 @@ 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(); + + // Registered before the server is started, since connections update it. + peakActiveComputeTasks = new PeakWithCurrentMetric( + metricName(mreg.name(), METRIC_PEAK_ACTIVE_COMPUTE_TASKS), + "Maximum number of active user compute tasks per thin client connection over the configured time " + + "interval, but not less than the current value. The upper limit is the " + + "ThinClientConfiguration.maxActiveComputeTasksPerConnection configuration parameter.", + DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL, + DFLT_SIZE, + maxActiveComputeTasks, + this::peakActiveComputeTasks + ); + + mreg.register(peakActiveComputeTasks); + } + + /** @return Current peak number of active user compute tasks across thin client connections. */ + private long peakActiveComputeTasks() { + 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 +663,15 @@ public ClientListenerMetrics metrics() { return metrics; } + /** + * Updates the peak of active user compute tasks per thin client connection. + * + * @param activeTasksCnt Number of active user compute tasks of the connection, capped by the limit. + */ + public void updatePeakActiveComputeTasks(int activeTasksCnt) { + peakActiveComputeTasks.update(activeTasksCnt); + } + /** * Prepare connector configuration. * @@ -853,4 +910,36 @@ public class ClientProcessorMXBeanImpl implements ClientProcessorMXBean { } } } + + /** Maximum value over the time interval, but not less than the current value. */ + private static class PeakWithCurrentMetric extends MaxValueMetric { + /** Upper limit of the value. */ + private final long limit; + + /** Current value supplier. */ + private final LongSupplier cur; + + /** + * @param name Name. + * @param desc Description. + * @param timeInterval Time interval in milliseconds. + * @param size Number of buckets. + * @param limit Upper limit of the value. + * @param cur Current value supplier. + */ + PeakWithCurrentMetric(String name, String desc, long timeInterval, int size, long limit, LongSupplier cur) { + super(name, desc, timeInterval, size); + + this.limit = limit; + this.cur = cur; + } + + /** {@inheritDoc} */ + @Override public long value() { + long peak = super.value(); + + // The current value can not exceed the limit, so there is no need to calculate it. + return peak >= limit ? peak : Math.max(peak, 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..3623d4613feb9 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().updatePeakActiveComputeTasks(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().updatePeakActiveComputeTasks(cnt); } /** @@ -367,4 +374,10 @@ public void decrementActiveTasksCount() { assert cnt >= 0 : "Unexpected active tasks count: " + cnt; } + + /** @return Number of active user compute tasks for this connection capped by the limit. */ + 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..b849973352971 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 @@ -18,6 +18,7 @@ package org.apache.ignite.spi.systemview.view; import java.net.InetSocketAddress; +import org.apache.ignite.configuration.ThinClientConfiguration; import org.apache.ignite.internal.processors.odbc.ClientListenerConnectionContext; import org.apache.ignite.internal.processors.odbc.ClientListenerProtocolVersion; import org.apache.ignite.internal.processors.odbc.ClientListenerRequestHandler; @@ -119,4 +120,14 @@ 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, not greater than + * {@link ThinClientConfiguration#getMaxActiveComputeTasksPerConnection()}; {@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..d4d5425296225 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,47 @@ 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 java.util.stream.StreamSupport; +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.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.IgniteInternalFuture; 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.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 +65,36 @@ 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_PEAK_ACTIVE_COMPUTE_TASKS; 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.runAsync; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; /** * Client listener metrics tests. */ public class ClientListenerMetricsTest extends GridCommonAbstractTest { + /** Maximum active tasks. */ + private static final int MAX_TASKS = 3; + + /** Metric time interval long enough for the peak not to expire during a test. Expiration is emulated by a reset. */ + private static final long LONG_INTERVAL = 10 * 60_000L; + + /** Metric time interval to check the peak expiration without a long wait. */ + private static final long SHORT_INTERVAL = 2_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 +204,285 @@ public void testClientListenerMetricsRejectGeneral() throws Exception { } } + /** + * Check that the active compute tasks metric reports the peak of the busiest connection (not the sum), keeps it + * during the time interval and drops it after 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()) + ) { + checkPeakComputeTasks(0); + assertEquals(0, activeComputeTasksCount()); + + // Tasks 0, 1. + futs.addAll(startTasks(client1, 2)); + checkPeakComputeTasks(2); + assertEquals(2, activeComputeTasksCount()); + + // Run tasks 2, 3. Metric should not change. + futs.addAll(startTasks(client2, 2)); + checkPeakComputeTasks(2); + assertEquals(4, activeComputeTasksCount()); + + // Task 4. Metric should increase. + futs.addAll(startTasks(client2, 1)); + checkPeakComputeTasks(MAX_TASKS); + assertEquals(5, activeComputeTasksCount()); + + assertComputeTaskRejected(client2); + + // A rejected task counts as the 'MaxActiveComputeTasksPerConnection' limit. + checkPeakComputeTasks(MAX_TASKS); + assertEquals(5, activeComputeTasksCount()); + + // Finished tasks are released by the connections, but the metric keeps the peak during the time interval. + finishTasks(2); + checkPeakComputeTasks(MAX_TASKS); + + finishTasks(0, 1); + checkPeakComputeTasks(MAX_TASKS); + + finishTasks(3, 4); + checkPeakComputeTasks(MAX_TASKS); + + // Clear the peak, as if it expired. No tasks are active, so the metric reports zero. + computeTasksMetric().reset(); + assertEquals(0, computeTasksMetric().value()); + + // Shorten the time interval to check the peak expiration without a long wait. + computeTasksMetric().reset(SHORT_INTERVAL); + + // Task 5. + futs.addAll(startTasks(client1, 1)); + checkPeakComputeTasks(1); + + // The peak expires after the time interval. + finishTasks(5); + checkPeakComputeTasks(0); + + // Wait for all results before clients are closed. + for (IgniteInternalFuture fut : futs) + fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); + } + } + + /** + * Check that the metric is not less than the current value, so it reports long-running tasks after their peak has + * expired. + */ + @Test + public void testComputeTasksMetricOnLongRunningTasks() throws Exception { + startGridWithCompute(); + + try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { + List> futs = startTasks(client, MAX_TASKS); + + checkPeakComputeTasks(MAX_TASKS); + + // Clear the peak, as if it expired while the tasks are active. The metric reports the current value. + computeTasksMetric().reset(); + assertEquals(MAX_TASKS, computeTasksMetric().value()); + + finishTasks(IntStream.range(0, MAX_TASKS).toArray()); + + // Finished tasks do not update the peak, so the metric drops to the current value at once. + assertEquals(0, computeTasksMetric().value()); + + // Wait for all results before the client is closed. + for (IgniteInternalFuture fut : futs) + fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); + } + } + + /** + * Check that a task rejected by the limit updates the metric with the limit, even if the active tasks were started + * before the time interval. + */ + @Test + public void testComputeTasksMetricOnRejectedTask() throws Exception { + startGridWithCompute(); + + try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { + List> futs = startTasks(client, MAX_TASKS); + + checkPeakComputeTasks(MAX_TASKS); + + // Clear the peak, as if it expired while the tasks are active. + computeTasksMetric().reset(); + + assertComputeTaskRejected(client); + + // The metric keeps the limit from the rejected task, though the current value is less. + finishTasks(0); + checkPeakComputeTasks(MAX_TASKS); + + finishTasks(IntStream.range(1, MAX_TASKS).toArray()); + + // Wait for all results before the client is closed. + for (IgniteInternalFuture fut : futs) + fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); + } + } + + /** Check that closing the connection with unfinished tasks releases its tasks. */ + @Test + public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { + startGridWithCompute(); + + try (IgniteClient client1 = Ignition.startClient(getClientConfiguration())) { + List> futs1 = startTasks(client1, 1); + + List> futs2; + + ClientConnectionView connView2; + + try (IgniteClient client2 = Ignition.startClient(getClientConfiguration())) { + futs2 = startTasks(client2, 2); + + checkPeakComputeTasks(2); + + connView2 = thinConnections().stream().filter(v -> v.activeComputeTasks() == 2) + .findFirst().orElseThrow(); + } + + // Tasks of the closed connection are released on disconnect. + assertTrue(waitForCondition(() -> connView2.activeComputeTasks() == 0, getTestTimeout())); + + // Cancellation does not interrupt 'reduce()', so unblock the tasks of 'client2'. + TestComputeTask.finish(1); + TestComputeTask.finish(2); + + // The metric keeps the peak during the time interval. + checkPeakComputeTasks(2); + + // Unfinished tasks fail on the client side. The error depends on whether the client received the execute + // response before the close. + for (IgniteInternalFuture fut : futs2) + assertThrowsWithCause(() -> fut.get(getTestTimeout(), TimeUnit.MILLISECONDS), ClientException.class); + + // Last task of 'client1'. + finishTasks(0); + checkPeakComputeTasks(2); + + // Wait for the result before the client is closed. + for (IgniteInternalFuture fut : futs1) + fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); + } + } + + /** Check the active compute tasks metric when compute is disabled for thin clients. */ + @Test + public void testComputeTasksMetricsComputeDisabled() throws Exception { + startGrid(0); + + try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { + checkPeakComputeTasks(0); + + GridTestUtils.assertThrowsAnyCause( + log, + () -> client.compute().execute(TestComputeTask.class.getName(), null), + ClientException.class, + "Compute grid functionality is disabled for thin clients" + ); + + checkPeakComputeTasks(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); + } + + /** + * Checks that a task is rejected because of the limit. + * + * @param client Ignite client. + */ + 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" + ); + } + + /** + * Starts tasks via client and waits for them to be started on the node. + * + * @param client Ignite client. + * @param cnt Tasks count. + * @return Task futures. + */ + private List> startTasks(IgniteClient client, int cnt) throws Exception { + int expTasks = TestComputeTask.TASKS.size() + cnt; + + List> futs = IntStream.range(0, cnt) + .mapToObj(i -> runAsync(() -> client.compute().execute(TestComputeTask.class.getName(), null))) + .toList(); + + // A task instance is created after the counter is incremented. Wait for instances to keep the task order. + assertTrue(waitForCondition(() -> TestComputeTask.TASKS.size() == expTasks, getTestTimeout())); + + return futs; + } + + /** + * Finishes tasks and waits for them to be released by the connections. + * + * @param idxs Indexes of the tasks. + */ + 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 from the system view. */ + private int activeComputeTasksCount() { + return thinConnections().stream().mapToInt(ClientConnectionView::activeComputeTasks).sum(); + } + + /** + * Waits for the expected value of the active compute tasks metric. + * + * @param expMetric Expected value of the metric. + */ + private void checkPeakComputeTasks(long expMetric) throws Exception { + assertTrue(waitForCondition(() -> computeTasksMetric().value() == expMetric, getTestTimeout())); + } + + /** @return Active compute tasks metric. */ + private MaxValueMetric computeTasksMetric() { + return grid(0).context().metric().registry(CLIENT_CONNECTOR_METRICS).findMetric(METRIC_PEAK_ACTIVE_COMPUTE_TASKS); + } + + /** @return Thin client connections from the system view. */ + private List thinConnections() { + SystemView view = grid(0).context().systemView().view(CLI_CONN_VIEW); + + return StreamSupport.stream(view.spliterator(), false).filter(v -> "THIN".equals(v.type())).toList(); + } + /** */ private static ClientConfiguration getClientConfiguration() { return new ClientConfiguration() @@ -179,7 +502,7 @@ private static ClientConfiguration getClientConfiguration() { */ private void waitForMetricValue(MetricRegistry mreg, String metric, long value, long timeout) throws IgniteInterruptedCheckedException { - GridTestUtils.waitForCondition(new GridAbsPredicate() { + waitForCondition(new GridAbsPredicate() { @Override public boolean apply() { return mreg.findMetric(metric).value() == value; } @@ -217,4 +540,59 @@ 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 { + /** Tasks instances. */ + private static final List TASKS = new CopyOnWriteArrayList<>(); + + /** Finish latch. */ + private final CountDownLatch finishLatch = new CountDownLatch(1); + + /** Default constructor. */ + public TestComputeTask() { + TASKS.add(this); + } + + /** {@inheritDoc} */ + @Override public Map map(List subgrid, @Nullable Object arg) { + 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; + } + + /** + * Finishes task with a specified index. + * + * @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(); + } + } + + /** No-op job. */ + 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'");