From 3d7577a8244af1d55cf865953c7e7b3a6c894e12 Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Tue, 6 Oct 2026 13:34:24 +0300 Subject: [PATCH 1/3] IGNITE-29107 Add ClientConnector metric for active compute tasks on the busiest connection --- .../new-metrics-system.adoc | 1 + .../_docs/monitoring-metrics/new-metrics.adoc | 1 + .../odbc/ClientListenerProcessor.java | 26 ++ .../client/ClientConnectionContext.java | 5 + .../odbc/ClientListenerMetricsTest.java | 297 +++++++++++++++++- 5 files changed, 329 insertions(+), 1 deletion(-) diff --git a/docs/_docs/monitoring-metrics/new-metrics-system.adoc b/docs/_docs/monitoring-metrics/new-metrics-system.adoc index 4bfd340f23877..0d8ffc5d5b574 100644 --- a/docs/_docs/monitoring-metrics/new-metrics-system.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics-system.adoc @@ -419,6 +419,7 @@ 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]). +* `thin.ActiveComputeTasksOnBusiestConnection` - The maximum number of active compute tasks per connection among open thin client connections; system tasks are not counted (`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. * `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..50bada4e47b3d 100644 --- a/docs/_docs/monitoring-metrics/new-metrics.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics.adoc @@ -481,6 +481,7 @@ Register name: `client.connector` |receivedBytes | long | Total number of bytes received by current node. |sentBytes | long | Total number of bytes sent by current node. |thin.AcceptedSessions | integer | Number of successfully established sessions for the client type. +|thin.ActiveComputeTasksOnBusiestConnection | integer | Maximum number of active compute tasks per connection among open thin client connections. System tasks are not counted. |thin.ActiveSessions | integer | Number of active sessions for the thin client. |=== 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..eb763b1cb003b 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 @@ -47,6 +47,7 @@ import org.apache.ignite.internal.processors.metric.MetricRegistryImpl; 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; @@ -100,6 +101,13 @@ public class ClientListenerProcessor extends GridProcessorAdapter { /** */ public static final String METRIC_ACTIVE = "ActiveSessions"; + /** + * Name suffix of the metric of the maximum number of active compute tasks per connection among open thin client + * connections. System tasks are not counted. The full metric name is prefixed with the client type label, + * see {@link ClientListenerMetrics#clientTypeLabel(byte)}. + */ + public static final String METRIC_ACTIVE_COMPUTE_TASKS = "ActiveComputeTasksOnBusiestConnection"; + /** Default client connector configuration. */ public static final ClientConnectorConfiguration DFLT_CLI_CFG = new ClientConnectorConfigurationEx(); @@ -350,6 +358,24 @@ private void registerClientMetrics(MetricRegistry mreg) { "Number of active sessions for the " + cliTypeName + " client." ); } + + mreg.register( + metricName(clientTypeLabel(THIN_CLIENT), METRIC_ACTIVE_COMPUTE_TASKS), + () -> { + int res = 0; + + for (GridNioSession ses : srv.sessions()) { + ClientListenerConnectionContext connCtx = ses.meta(CONN_CTX_META_KEY); + + if (connCtx instanceof ClientConnectionContext) + res = Math.max(res, ((ClientConnectionContext)connCtx).activeTasksCount()); + } + + return res; + }, + "Maximum number of active compute tasks per connection among open thin client connections. " + + "System tasks are not counted." + ); } /** 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..ca0ebbf85ba0f 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 @@ -367,4 +367,9 @@ public void decrementActiveTasksCount() { assert cnt >= 0 : "Unexpected active tasks count: " + cnt; } + + /** @return Number of active compute tasks for this connection. System tasks are not counted. */ + public int activeTasksCount() { + return Math.min(activeTasksCnt.get(), maxActiveComputeTasks); + } } 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..002161fc3e630 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,48 @@ 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.concurrent.atomic.AtomicInteger; +import java.util.stream.IntStream; +import javax.management.DynamicMBean; +import org.apache.ignite.IgniteException; import org.apache.ignite.Ignition; import org.apache.ignite.client.ClientAuthenticationException; +import org.apache.ignite.client.ClientConnectionException; 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.MetricUtils; +import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext; import org.apache.ignite.internal.util.lang.GridAbsPredicate; +import org.apache.ignite.internal.util.nio.GridNioServer; +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.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 +66,31 @@ 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.ClientListenerMetrics.clientTypeLabel; +import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.CONN_CTX_META_KEY; +import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.THIN_CLIENT; import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.METRIC_ACTIVE; +import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.METRIC_ACTIVE_COMPUTE_TASKS; import static org.apache.ignite.ssl.SslContextFactory.DFLT_STORE_TYPE; +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; + + /** {@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 +200,206 @@ public void testClientListenerMetricsRejectGeneral() throws Exception { } } + /** Check active compute tasks on the busiest connection metric. */ + @Test + public void testComputeTasksMetrics() throws Exception { + IgniteConfiguration nodeCfg = getConfigurationWithCompute(); + + startGrid(nodeCfg); + + List> futs = new ArrayList<>(); + + try ( + IgniteClient client1 = Ignition.startClient(getClientConfiguration()); + IgniteClient client2 = Ignition.startClient(getClientConfiguration()) + ) { + checkComputeTasksMetrics(0); + + futs.addAll(startTasksAndCheckMetrics(client1, 1, 1)); + + ClientConnectionContext connCtx1 = thinClientConnectionContexts().stream() + .filter(c -> c.activeTasksCount() == 1) + .findFirst() + .orElseThrow(); + + futs.addAll(startTasksAndCheckMetrics(client2, 2, 2)); + + futs.addAll(startTasksAndCheckMetrics(client2, 1, MAX_TASKS)); + + GridTestUtils.assertThrowsAnyCause( + log, + () -> client2.compute().execute(TestComputeTask.class.getName(), null), + ClientException.class, + "Active compute tasks per connection limit (" + MAX_TASKS + ") exceeded" + ); + + checkComputeTasksMetrics(MAX_TASKS); + + // Finish on 'client1' does not affect busiest connection. + TestComputeTask.finish(0); + + // Counter is decremented after the result is sent to the client, so wait for the counter itself. + assertTrue(waitForCondition(() -> connCtx1.activeTasksCount() == 0, getTestTimeout())); + + checkComputeTasksMetrics(MAX_TASKS); + + // Finish some tasks on 'client2' affects busiest connection. + TestComputeTask.finish(1); + checkComputeTasksMetrics(MAX_TASKS - 1); + + TestComputeTask.finish(2); + checkComputeTasksMetrics(MAX_TASKS - 2); + + // Last task of 'client2'. + TestComputeTask.finish(3); + checkComputeTasksMetrics(0); + + // Wait for all results before clients are closed. + for (IgniteInternalFuture fut : futs) + fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); + } + } + + /** + * Check that closing the busiest connection with unfinished tasks releases its tasks, and the compute tasks metric + * switches to the next busiest connection. + */ + @Test + public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { + startGrid(getConfigurationWithCompute()); + + try (IgniteClient client1 = Ignition.startClient(getClientConfiguration())) { + List> futs1 = startTasksAndCheckMetrics(client1, 1, 1); + + List> futs2; + + ClientConnectionContext connCtx2; + + try (IgniteClient client2 = Ignition.startClient(getClientConfiguration())) { + futs2 = startTasksAndCheckMetrics(client2, 2, 2); + + List connCtxs = thinClientConnectionContexts(); + + assertEquals(2, connCtxs.size()); + + connCtx2 = connCtxs.stream().filter(c -> c.activeTasksCount() == 2).findFirst().orElseThrow(); + + // Wait until the client receives execute responses, otherwise requests fail with another error on close. + AtomicInteger clientTasksCnt = GridTestUtils.getFieldValue(client2.compute(), "tasksCnt"); + + assertTrue(waitForCondition(() -> clientTasksCnt.get() == 2, getTestTimeout())); + } + + // Task resources are released on disconnect, so the counter of the closed connection is decremented. + assertTrue(waitForCondition(() -> connCtx2.activeTasksCount() == 0, getTestTimeout())); + + // Tasks of the closed connection are not counted, the task of 'client1' is still counted. + checkComputeTasksMetrics(1); + + // Unfinished task execution fails on the client side, since the connection is closed. + for (IgniteInternalFuture fut : futs2) + GridTestUtils.assertThrowsAnyCause( + log, + () -> fut.get(getTestTimeout(), TimeUnit.MILLISECONDS), + ClientConnectionException.class, + "Connection to server is closed" + ); + + // Last task of 'client1'. + TestComputeTask.finish(0); + checkComputeTasksMetrics(0); + + // Wait for the result before the client is closed. + for (IgniteInternalFuture fut : futs1) + fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); + } + } + + /** + * Check active compute tasks on the busiest connection metric when compute is disabled for thin clients. + */ + @Test + public void testComputeTasksMetricsComputeDisabled() throws Exception { + startGrid(0); + + try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { + checkComputeTasksMetrics(0); + + GridTestUtils.assertThrowsAnyCause( + log, + () -> client.compute().execute(TestComputeTask.class.getName(), null), + ClientException.class, + "Compute grid functionality is disabled for thin clients" + ); + + checkComputeTasksMetrics(0); + } + } + + /** Get configuration with enabled compute on clients. */ + private IgniteConfiguration getConfigurationWithCompute() throws Exception { + return getConfiguration(getTestIgniteInstanceName(0)) + .setClientConnectorConfiguration(new ClientConnectorConfiguration() + .setThinClientConfiguration(new ThinClientConfiguration() + .setMaxActiveComputeTasksPerConnection(MAX_TASKS))); + } + + /** + * Starts tasks via client and checks compute tasks metrics. + * + * @param client Ignite client. + * @param cnt Tasks count. + * @param expActive Expected active compute tasks on the busiest connection. + * @return Task futures. + */ + private List> startTasksAndCheckMetrics( + IgniteClient client, + int cnt, + int expActive) 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(); + + // Task instances are created after the active tasks counter is incremented, so wait for them to keep the order. + assertTrue(waitForCondition(() -> TestComputeTask.TASKS.size() == expTasks, getTestTimeout())); + + checkComputeTasksMetrics(expActive); + + return futs; + } + + /** + * Check active compute tasks on the busiest connection metric via the metric registry and JMX. + * + * @param expActive Expected active compute tasks on the busiest connection. + */ + private void checkComputeTasksMetrics(int expActive) throws Exception { + MetricRegistry mreg = grid(0).context().metric().registry(CLIENT_CONNECTOR_METRICS); + + String metric = MetricUtils.metricName(clientTypeLabel(THIN_CLIENT), METRIC_ACTIVE_COMPUTE_TASKS); + + waitForMetricValue(mreg, metric, expActive, getTestTimeout()); + + assertEquals(expActive, mreg.findMetric(metric).value()); + + DynamicMBean mbean = metricRegistry(grid(0).name(), "client", "connector"); + + assertEquals(expActive, mbean.getAttribute(metric)); + } + + /** @return Connection contexts of the active thin client sessions. */ + private List thinClientConnectionContexts() { + GridNioServer srv = GridTestUtils.getFieldValue(grid(0).context().clientListener(), "srv"); + + return srv.sessions().stream() + .map(ses -> ses.meta(CONN_CTX_META_KEY)) + .filter(ClientConnectionContext.class::isInstance) + .map(ClientConnectionContext.class::cast) + .toList(); + } + /** */ private static ClientConfiguration getClientConfiguration() { return new ClientConfiguration() @@ -179,7 +419,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 +457,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(); + } + + /** Clears static state. */ + 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; + } + } } From 050f56ba2f8bb7459e1d1d99da583c61707ccef1 Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Thu, 15 Oct 2026 12:05:34 +0300 Subject: [PATCH 2/3] IGNITE-29107 Rework: add active tasks to a system view, refactor metric to a MaxValueMetric --- .../new-metrics-system.adoc | 4 +- .../_docs/monitoring-metrics/new-metrics.adoc | 2 +- .../monitoring-metrics/system-views.adoc | 1 + .../jdbc/thin/JdbcThinMetadataSelfTest.java | 1 + .../ignite/util/SystemViewCommandTest.java | 4 +- .../odbc/ClientListenerMetrics.java | 38 +++ .../odbc/ClientListenerProcessor.java | 26 -- .../client/ClientConnectionContext.java | 9 +- .../systemview/view/ClientConnectionView.java | 9 + .../odbc/ClientListenerMetricsTest.java | 275 +++++++++++------- .../cache/metric/SqlViewExporterSpiTest.java | 12 +- 11 files changed, 247 insertions(+), 134 deletions(-) diff --git a/docs/_docs/monitoring-metrics/new-metrics-system.adoc b/docs/_docs/monitoring-metrics/new-metrics-system.adoc index 0d8ffc5d5b574..849d94939b34a 100644 --- a/docs/_docs/monitoring-metrics/new-metrics-system.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics-system.adoc @@ -419,8 +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]). -* `thin.ActiveComputeTasksOnBusiestConnection` - The maximum number of active compute tasks per connection among open thin client connections; system tasks are not counted (`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 compute tasks per thin client connection during the last time interval (90 seconds by default, can be changed with `control.sh --metric --configure-max-value`). The value is updated when a task is started; system tasks are not counted, a task rejected because of the limit is counted as the limit. To not miss peaks between polls, set the interval longer than the polling period of your monitoring system (`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 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 50bada4e47b3d..ea88b9a379b2a 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 compute tasks per thin client connection during the last time interval (90 seconds by default). The value is updated when a task is started. System tasks are not counted. |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. @@ -481,7 +482,6 @@ Register name: `client.connector` |receivedBytes | long | Total number of bytes received by current node. |sentBytes | long | Total number of bytes sent by current node. |thin.AcceptedSessions | integer | Number of successfully established sessions for the client type. -|thin.ActiveComputeTasksOnBusiestConnection | integer | Maximum number of active compute tasks per connection among open thin client connections. System tasks are not counted. |thin.ActiveSessions | integer | Number of active sessions for the thin client. |=== diff --git a/docs/_docs/monitoring-metrics/system-views.adoc b/docs/_docs/monitoring-metrics/system-views.adoc index 969b9f2061944..e5a716f550169 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 compute tasks for thin client connections, system tasks are not counted. `null` for JDBC and ODBC connections |=== == CLIENT_CONNECTION_ATTRIBUTES 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/ClientListenerMetrics.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetrics.java index fecba7c65ee68..3647cbd614b33 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetrics.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetrics.java @@ -20,8 +20,11 @@ import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric; import org.apache.ignite.internal.processors.metric.impl.IntMetricImpl; import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric; +import org.apache.ignite.internal.processors.metric.impl.MaxValueMetric; import org.apache.ignite.internal.processors.metric.impl.MetricUtils; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.internal.processors.metric.impl.AbstractIntervalMetric.DFLT_SIZE; import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.CLI_TYPES; import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.JDBC_CLIENT; import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.ODBC_CLIENT; @@ -43,6 +46,18 @@ public class ClientListenerMetrics { /** Number of successfully established sessions. */ public static final String METRIC_ACEPTED = "AcceptedSessions"; + /** + * Maximum number of active compute tasks per thin client connection during the last time interval. System tasks are + * not counted. + */ + public static final String METRIC_PEAK_ACTIVE_COMPUTE_TASKS = "PeakActiveComputeTasksPerConnection"; + + /** + * Default time interval of the {@link #METRIC_PEAK_ACTIVE_COMPUTE_TASKS} metric in milliseconds. It is longer than the + * typical 1 minute polling period of monitoring systems to not miss peaks between polls. + */ + public static final long DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL = 90_000L; + /** */ public static final String AFF_KEY_HITS = "AffinityKeyRequestsHits"; @@ -79,6 +94,9 @@ public class ClientListenerMetrics { /** */ private final AtomicLongMetric affQryMisses; + /** Maximum number of active compute tasks per thin client connection. */ + private final MaxValueMetric peakActiveComputeTasks; + /** * @param mreg Metrics registry. */ @@ -105,6 +123,16 @@ public ClientListenerMetrics(MetricRegistryImpl mreg) { affQryMisses = mreg.longMetric(AFF_QRY_MISSES, "The number of affinity-aware query requests that were sent not to the primary node"); + peakActiveComputeTasks = mreg.maxValueMetric( + METRIC_PEAK_ACTIVE_COMPUTE_TASKS, + "Maximum number of active compute tasks per thin client connection during the last time interval " + + "(" + MILLISECONDS.toSeconds(DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL) + " seconds by default). " + + "The value is updated when a task is started. System tasks are not counted. A task rejected because " + + "of the limit is counted as the limit.", + DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL, + DFLT_SIZE + ); + for (byte clientType : CLI_TYPES) { String clientLbl = clientTypeLabel(clientType); @@ -146,6 +174,16 @@ public void onHandshakeAccept(byte clientType) { accepted[clientType].increment(); } + /** + * Callback invoked when a compute task is attempted to start by a thin client. + * + * @param activeTasksCnt Number of active compute tasks of the connection, including the invoked task, capped by + * the limit. + */ + public void onComputeTaskCountUpdate(int activeTasksCnt) { + peakActiveComputeTasks.update(activeTasksCnt); + } + /** */ public void onAffinityKeyHit() { affKeyHits.increment(); 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 eb763b1cb003b..808437eb1719a 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 @@ -47,7 +47,6 @@ import org.apache.ignite.internal.processors.metric.MetricRegistryImpl; 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; @@ -101,13 +100,6 @@ public class ClientListenerProcessor extends GridProcessorAdapter { /** */ public static final String METRIC_ACTIVE = "ActiveSessions"; - /** - * Name suffix of the metric of the maximum number of active compute tasks per connection among open thin client - * connections. System tasks are not counted. The full metric name is prefixed with the client type label, - * see {@link ClientListenerMetrics#clientTypeLabel(byte)}. - */ - public static final String METRIC_ACTIVE_COMPUTE_TASKS = "ActiveComputeTasksOnBusiestConnection"; - /** Default client connector configuration. */ public static final ClientConnectorConfiguration DFLT_CLI_CFG = new ClientConnectorConfigurationEx(); @@ -358,24 +350,6 @@ private void registerClientMetrics(MetricRegistry mreg) { "Number of active sessions for the " + cliTypeName + " client." ); } - - mreg.register( - metricName(clientTypeLabel(THIN_CLIENT), METRIC_ACTIVE_COMPUTE_TASKS), - () -> { - int res = 0; - - for (GridNioSession ses : srv.sessions()) { - ClientListenerConnectionContext connCtx = ses.meta(CONN_CTX_META_KEY); - - if (connCtx instanceof ClientConnectionContext) - res = Math.max(res, ((ClientConnectionContext)connCtx).activeTasksCount()); - } - - return res; - }, - "Maximum number of active compute tasks per connection among open thin client connections. " + - "System tasks are not counted." - ); } /** 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 ca0ebbf85ba0f..48264cc4aee87 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,7 +349,12 @@ public void incrementActiveTasksCount() { "To enable it set up the ThinClientConfiguration.MaxActiveComputeTasksPerConnection property."); } - if (activeTasksCnt.incrementAndGet() > maxActiveComputeTasks) { + int cnt = activeTasksCnt.incrementAndGet(); + + // Rejected task is counted as the limit, the counter may momentarily exceed it. + ctx.clientListener().metrics().onComputeTaskCountUpdate(Math.min(cnt, maxActiveComputeTasks)); + + if (cnt > maxActiveComputeTasks) { activeTasksCnt.decrementAndGet(); throw new IgniteClientException(ClientStatus.TOO_MANY_COMPUTE_TASKS, "Active compute tasks per connection " + @@ -366,6 +371,8 @@ public void decrementActiveTasksCount() { int cnt = activeTasksCnt.decrementAndGet(); assert cnt >= 0 : "Unexpected active tasks count: " + cnt; + + ctx.clientListener().metrics().onComputeTaskCountUpdate(cnt); } /** @return Number of active compute tasks for this connection. System tasks are not counted. */ 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..79b725d0105ce 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 compute tasks for the thin client connection, system tasks are not counted. + * {@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 002161fc3e630..ea8543e787785 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 @@ -23,13 +23,12 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; +import java.util.stream.StreamSupport; import javax.management.DynamicMBean; import org.apache.ignite.IgniteException; import org.apache.ignite.Ignition; import org.apache.ignite.client.ClientAuthenticationException; -import org.apache.ignite.client.ClientConnectionException; import org.apache.ignite.client.ClientException; import org.apache.ignite.client.Config; import org.apache.ignite.client.IgniteClient; @@ -49,28 +48,30 @@ import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.IgniteInterruptedCheckedException; +import org.apache.ignite.internal.management.api.CommandMBean; +import org.apache.ignite.internal.processors.metric.impl.MaxValueMetric; import org.apache.ignite.internal.processors.metric.impl.MetricUtils; -import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext; import org.apache.ignite.internal.util.lang.GridAbsPredicate; -import org.apache.ignite.internal.util.nio.GridNioServer; 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 java.util.Collections.singletonList; import static org.apache.ignite.internal.processors.metric.GridMetricManager.CLIENT_CONNECTOR_METRICS; +import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL; import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.METRIC_ACEPTED; +import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.METRIC_PEAK_ACTIVE_COMPUTE_TASKS; 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.ClientListenerMetrics.clientTypeLabel; -import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.CONN_CTX_META_KEY; -import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.THIN_CLIENT; +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_ACTIVE_COMPUTE_TASKS; import static org.apache.ignite.ssl.SslContextFactory.DFLT_STORE_TYPE; import static org.apache.ignite.testframework.GridTestUtils.runAsync; import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; @@ -82,6 +83,12 @@ public class ClientListenerMetricsTest extends GridCommonAbstractTest { /** Maximum active tasks. */ private static final int MAX_TASKS = 3; + /** Time interval of the active compute tasks metric, so the peaks do not expire during the test. */ + private static final long LONG_INTERVAL = 10 * 60_000L; + + /** Time interval of the active compute tasks metric to check that the peak expires, long enough to check the peak. */ + private static final long SHORT_INTERVAL = 10_000L; + /** {@inheritDoc} */ @Override protected void afterTest() throws Exception { TestComputeTask.reset(); @@ -200,12 +207,14 @@ public void testClientListenerMetricsRejectGeneral() throws Exception { } } - /** Check active compute tasks on the busiest connection metric. */ + /** + * Check that the active compute tasks metric reports the peak of the busiest connection (not the sum) and keeps it + * during the time interval, while the {@link ClientListenerProcessor#CLI_CONN_VIEW} system view reports current + * values. + */ @Test public void testComputeTasksMetrics() throws Exception { - IgniteConfiguration nodeCfg = getConfigurationWithCompute(); - - startGrid(nodeCfg); + startGridWithCompute(); List> futs = new ArrayList<>(); @@ -213,46 +222,48 @@ public void testComputeTasksMetrics() throws Exception { IgniteClient client1 = Ignition.startClient(getClientConfiguration()); IgniteClient client2 = Ignition.startClient(getClientConfiguration()) ) { - checkComputeTasksMetrics(0); + checkComputeTasks(0, 0, 0); - futs.addAll(startTasksAndCheckMetrics(client1, 1, 1)); + // Tasks 0, 1. + futs.addAll(startTasks(client1, 2)); + checkComputeTasks(2, 2, 0); - ClientConnectionContext connCtx1 = thinClientConnectionContexts().stream() - .filter(c -> c.activeTasksCount() == 1) - .findFirst() - .orElseThrow(); + // Tasks 2, 3. The sum of the connections (4) exceeds even the limit, the metric reports the busiest one. + futs.addAll(startTasks(client2, 2)); + checkComputeTasks(2, 2, 2); - futs.addAll(startTasksAndCheckMetrics(client2, 2, 2)); + // Task 4. + futs.addAll(startTasks(client2, 1)); + checkComputeTasks(MAX_TASKS, 2, MAX_TASKS); - futs.addAll(startTasksAndCheckMetrics(client2, 1, MAX_TASKS)); + assertComputeTaskRejected(client2); - GridTestUtils.assertThrowsAnyCause( - log, - () -> client2.compute().execute(TestComputeTask.class.getName(), null), - ClientException.class, - "Active compute tasks per connection limit (" + MAX_TASKS + ") exceeded" - ); + // Rejected task is not counted in the view. + checkComputeTasks(MAX_TASKS, 2, MAX_TASKS); - checkComputeTasksMetrics(MAX_TASKS); + // Finished tasks are not counted in the view, but the metric keeps the peak during the time interval. + TestComputeTask.finish(2); + checkComputeTasks(MAX_TASKS, 2, MAX_TASKS - 1); - // Finish on 'client1' does not affect busiest connection. TestComputeTask.finish(0); + TestComputeTask.finish(1); + checkComputeTasks(MAX_TASKS, 0, MAX_TASKS - 1); - // Counter is decremented after the result is sent to the client, so wait for the counter itself. - assertTrue(waitForCondition(() -> connCtx1.activeTasksCount() == 0, getTestTimeout())); - - checkComputeTasksMetrics(MAX_TASKS); + TestComputeTask.finish(3); + TestComputeTask.finish(4); + checkComputeTasks(MAX_TASKS, 0, 0); - // Finish some tasks on 'client2' affects busiest connection. - TestComputeTask.finish(1); - checkComputeTasksMetrics(MAX_TASKS - 1); + // Shorten the time interval to check that the peak expires. Reconfiguration also clears the peak. + configureComputeTasksInterval(SHORT_INTERVAL); + checkComputeTasks(0, 0, 0); - TestComputeTask.finish(2); - checkComputeTasksMetrics(MAX_TASKS - 2); + // Task 5. + futs.addAll(startTasks(client1, 1)); + checkComputeTasks(1, 1, 0); - // Last task of 'client2'. - TestComputeTask.finish(3); - checkComputeTasksMetrics(0); + // The peak expires after the time interval. + TestComputeTask.finish(5); + checkComputeTasks(0, 0, 0); // Wait for all results before clients are closed. for (IgniteInternalFuture fut : futs) @@ -261,53 +272,69 @@ public void testComputeTasksMetrics() throws Exception { } /** - * Check that closing the busiest connection with unfinished tasks releases its tasks, and the compute tasks metric - * switches to the next busiest connection. + * Check that a task rejected because of the limit updates the metric, so it reports the saturated connection even + * if its tasks were started before the current time interval. */ @Test + public void testComputeTasksMetricOnRejectedTask() throws Exception { + startGridWithCompute(); + + try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { + List> futs = startTasks(client, MAX_TASKS); + + checkComputeTasks(MAX_TASKS, MAX_TASKS); + + // Clear the accumulated peak, as if it expired while the tasks are still active. + computeTasksMetric().reset(); + checkComputeTasks(0, MAX_TASKS); + + assertComputeTaskRejected(client); + + checkComputeTasks(MAX_TASKS, MAX_TASKS); + + for (int i = 0; i < MAX_TASKS; i++) + TestComputeTask.finish(i); + + // 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 { - startGrid(getConfigurationWithCompute()); + startGridWithCompute(); try (IgniteClient client1 = Ignition.startClient(getClientConfiguration())) { - List> futs1 = startTasksAndCheckMetrics(client1, 1, 1); + List> futs1 = startTasks(client1, 1); List> futs2; - ClientConnectionContext connCtx2; + ClientConnectionView connView2; try (IgniteClient client2 = Ignition.startClient(getClientConfiguration())) { - futs2 = startTasksAndCheckMetrics(client2, 2, 2); + futs2 = startTasks(client2, 2); - List connCtxs = thinClientConnectionContexts(); + checkComputeTasks(2, 1, 2); - assertEquals(2, connCtxs.size()); - - connCtx2 = connCtxs.stream().filter(c -> c.activeTasksCount() == 2).findFirst().orElseThrow(); - - // Wait until the client receives execute responses, otherwise requests fail with another error on close. - AtomicInteger clientTasksCnt = GridTestUtils.getFieldValue(client2.compute(), "tasksCnt"); - - assertTrue(waitForCondition(() -> clientTasksCnt.get() == 2, getTestTimeout())); + connView2 = thinConnections().stream().filter(v -> v.activeComputeTasks() == 2).findFirst().orElseThrow(); } // Task resources are released on disconnect, so the counter of the closed connection is decremented. - assertTrue(waitForCondition(() -> connCtx2.activeTasksCount() == 0, getTestTimeout())); + assertTrue(waitForCondition(() -> connView2.activeComputeTasks() == 0, getTestTimeout())); - // Tasks of the closed connection are not counted, the task of 'client1' is still counted. - checkComputeTasksMetrics(1); + // Closed connection is not shown, the metric keeps the peak during the time interval. + checkComputeTasks(2, 1); - // Unfinished task execution fails on the client side, since the connection is closed. + // Unfinished task execution fails on the client side, since the connection is closed. The error depends on + // whether the client has received the execute response before the close. for (IgniteInternalFuture fut : futs2) - GridTestUtils.assertThrowsAnyCause( - log, - () -> fut.get(getTestTimeout(), TimeUnit.MILLISECONDS), - ClientConnectionException.class, - "Connection to server is closed" - ); + GridTestUtils.assertThrowsWithCause(() -> fut.get(getTestTimeout(), TimeUnit.MILLISECONDS), ClientException.class); // Last task of 'client1'. TestComputeTask.finish(0); - checkComputeTasksMetrics(0); + checkComputeTasks(2, 0); // Wait for the result before the client is closed. for (IgniteInternalFuture fut : futs1) @@ -315,15 +342,13 @@ public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { } } - /** - * Check active compute tasks on the busiest connection metric when compute is disabled for thin clients. - */ + /** Check the active compute tasks metric and system view when compute is disabled for thin clients. */ @Test public void testComputeTasksMetricsComputeDisabled() throws Exception { startGrid(0); try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { - checkComputeTasksMetrics(0); + checkComputeTasks(0, 0); GridTestUtils.assertThrowsAnyCause( log, @@ -332,30 +357,67 @@ public void testComputeTasksMetricsComputeDisabled() throws Exception { "Compute grid functionality is disabled for thin clients" ); - checkComputeTasksMetrics(0); + checkComputeTasks(0, 0); } } - /** Get configuration with enabled compute on clients. */ - private IgniteConfiguration getConfigurationWithCompute() throws Exception { - return getConfiguration(getTestIgniteInstanceName(0)) + /** + * Starts a node with enabled compute for thin clients. The time interval of the active compute tasks metric is + * changed to {@link #LONG_INTERVAL}, so the peaks do not expire during the test. + */ + private void startGridWithCompute() throws Exception { + startGrid(getConfiguration(getTestIgniteInstanceName(0)) .setClientConnectorConfiguration(new ClientConnectorConfiguration() .setThinClientConfiguration(new ThinClientConfiguration() - .setMaxActiveComputeTasksPerConnection(MAX_TASKS))); + .setMaxActiveComputeTasksPerConnection(MAX_TASKS)))); + + assertEquals(DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL, computeTasksMetric().timeInterval()); + + configureComputeTasksInterval(LONG_INTERVAL); } /** - * Starts tasks via client and checks compute tasks metrics. + * Changes the time interval of the active compute tasks metric the same way as + * {@code control.sh --metric --configure-max-value} does. + * + * @param interval Time interval in milliseconds. + */ + private void configureComputeTasksInterval(long interval) throws Exception { + DynamicMBean mbean = getMxBean(grid(0).name(), "management", singletonList("Metric"), "ConfigureMaxValue", + DynamicMBean.class); + + mbean.invoke( + CommandMBean.INVOKE, + new Object[] {MetricUtils.metricName(CLIENT_CONNECTOR_METRICS, METRIC_PEAK_ACTIVE_COMPUTE_TASKS), + Long.toString(interval), null}, + new String[] {String.class.getName(), String.class.getName(), String.class.getName()} + ); + + assertTrue(waitForCondition(() -> computeTasksMetric().timeInterval() == interval, getTestTimeout())); + } + + /** + * 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. - * @param expActive Expected active compute tasks on the busiest connection. * @return Task futures. */ - private List> startTasksAndCheckMetrics( - IgniteClient client, - int cnt, - int expActive) throws Exception { + private List> startTasks(IgniteClient client, int cnt) throws Exception { int expTasks = TestComputeTask.TASKS.size() + cnt; List> futs = IntStream.range(0, cnt) @@ -365,39 +427,54 @@ private List> startTasksAndCheckMetrics( // Task instances are created after the active tasks counter is incremented, so wait for them to keep the order. assertTrue(waitForCondition(() -> TestComputeTask.TASKS.size() == expTasks, getTestTimeout())); - checkComputeTasksMetrics(expActive); - return futs; } /** - * Check active compute tasks on the busiest connection metric via the metric registry and JMX. + * Checks the active compute tasks metric via the metric registry and JMX, and the current active compute tasks of + * the thin client connections via the system view. * - * @param expActive Expected active compute tasks on the busiest connection. + * @param expMetric Expected value of the metric. + * @param expConnTasks Expected active compute tasks of the thin client connections, in any order. */ - private void checkComputeTasksMetrics(int expActive) throws Exception { - MetricRegistry mreg = grid(0).context().metric().registry(CLIENT_CONNECTOR_METRICS); + private void checkComputeTasks(long expMetric, int... expConnTasks) throws Exception { + List exp = IntStream.of(expConnTasks).sorted().boxed().toList(); + + // Counter is decremented after the result is sent to the client, so wait for the values. + assertTrue(waitForCondition(() -> exp.equals(thinConnectionsComputeTasks()), getTestTimeout())); - String metric = MetricUtils.metricName(clientTypeLabel(THIN_CLIENT), METRIC_ACTIVE_COMPUTE_TASKS); + assertTrue(waitForCondition(() -> computeTasksMetric().value() == expMetric, getTestTimeout())); - waitForMetricValue(mreg, metric, expActive, getTestTimeout()); + assertTrue(waitForCondition(() -> jmxComputeTasksMetric() == expMetric, getTestTimeout())); + } - assertEquals(expActive, mreg.findMetric(metric).value()); + /** @return Active compute tasks metric. */ + private MaxValueMetric computeTasksMetric() { + return grid(0).context().metric().registry(CLIENT_CONNECTOR_METRICS).findMetric(METRIC_PEAK_ACTIVE_COMPUTE_TASKS); + } + /** @return Active compute tasks metric value read via JMX. */ + private long jmxComputeTasksMetric() { DynamicMBean mbean = metricRegistry(grid(0).name(), "client", "connector"); - assertEquals(expActive, mbean.getAttribute(metric)); + try { + return (Long)mbean.getAttribute(METRIC_PEAK_ACTIVE_COMPUTE_TASKS); + } + catch (Exception e) { + throw new IgniteException(e); + } } - /** @return Connection contexts of the active thin client sessions. */ - private List thinClientConnectionContexts() { - GridNioServer srv = GridTestUtils.getFieldValue(grid(0).context().clientListener(), "srv"); + /** @return Sorted active compute tasks of the thin client connections from the system view. */ + private List thinConnectionsComputeTasks() { + return thinConnections().stream().map(ClientConnectionView::activeComputeTasks).sorted().toList(); + } - return srv.sessions().stream() - .map(ses -> ses.meta(CONN_CTX_META_KEY)) - .filter(ClientConnectionContext.class::isInstance) - .map(ClientConnectionContext.class::cast) - .toList(); + /** @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(); } /** */ 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'"); From 164399cf95fcdd72ad7c513997821402cdf0aea5 Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Thu, 15 Oct 2026 18:43:50 +0300 Subject: [PATCH 3/3] IGNITE-29107 Rework: extended max metric. --- .../new-metrics-system.adoc | 4 +- .../_docs/monitoring-metrics/new-metrics.adoc | 2 +- .../monitoring-metrics/system-views.adoc | 2 +- docs/_docs/tools/control-script.adoc | 5 +- .../odbc/ClientListenerMetrics.java | 38 --- .../odbc/ClientListenerProcessor.java | 93 +++++++- .../client/ClientConnectionContext.java | 13 +- .../systemview/view/ClientConnectionView.java | 6 +- .../odbc/ClientListenerMetricsTest.java | 224 +++++++++--------- 9 files changed, 225 insertions(+), 162 deletions(-) diff --git a/docs/_docs/monitoring-metrics/new-metrics-system.adoc b/docs/_docs/monitoring-metrics/new-metrics-system.adoc index 849d94939b34a..c70a6466985da 100644 --- a/docs/_docs/monitoring-metrics/new-metrics-system.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics-system.adoc @@ -419,8 +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]). -* `PeakActiveComputeTasksPerConnection` - The maximum number of active compute tasks per thin client connection during the last time interval (90 seconds by default, can be changed with `control.sh --metric --configure-max-value`). The value is updated when a task is started; system tasks are not counted, a task rejected because of the limit is counted as the limit. To not miss peaks between polls, set the interval longer than the polling period of your monitoring system (`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 compute tasks. +* `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 ea88b9a379b2a..5e8e2a128715f 100644 --- a/docs/_docs/monitoring-metrics/new-metrics.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics.adoc @@ -466,7 +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 compute tasks per thin client connection during the last time interval (90 seconds by default). The value is updated when a task is started. System tasks are not counted. +|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 e5a716f550169..9b2a41db6bc19 100644 --- a/docs/_docs/monitoring-metrics/system-views.adoc +++ b/docs/_docs/monitoring-metrics/system-views.adoc @@ -443,7 +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 compute tasks for thin client connections, system tasks are not counted. `null` for JDBC and ODBC connections +|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/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetrics.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetrics.java index 3647cbd614b33..fecba7c65ee68 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetrics.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetrics.java @@ -20,11 +20,8 @@ import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric; import org.apache.ignite.internal.processors.metric.impl.IntMetricImpl; import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric; -import org.apache.ignite.internal.processors.metric.impl.MaxValueMetric; import org.apache.ignite.internal.processors.metric.impl.MetricUtils; -import static java.util.concurrent.TimeUnit.MILLISECONDS; -import static org.apache.ignite.internal.processors.metric.impl.AbstractIntervalMetric.DFLT_SIZE; import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.CLI_TYPES; import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.JDBC_CLIENT; import static org.apache.ignite.internal.processors.odbc.ClientListenerNioListener.ODBC_CLIENT; @@ -46,18 +43,6 @@ public class ClientListenerMetrics { /** Number of successfully established sessions. */ public static final String METRIC_ACEPTED = "AcceptedSessions"; - /** - * Maximum number of active compute tasks per thin client connection during the last time interval. System tasks are - * not counted. - */ - public static final String METRIC_PEAK_ACTIVE_COMPUTE_TASKS = "PeakActiveComputeTasksPerConnection"; - - /** - * Default time interval of the {@link #METRIC_PEAK_ACTIVE_COMPUTE_TASKS} metric in milliseconds. It is longer than the - * typical 1 minute polling period of monitoring systems to not miss peaks between polls. - */ - public static final long DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL = 90_000L; - /** */ public static final String AFF_KEY_HITS = "AffinityKeyRequestsHits"; @@ -94,9 +79,6 @@ public class ClientListenerMetrics { /** */ private final AtomicLongMetric affQryMisses; - /** Maximum number of active compute tasks per thin client connection. */ - private final MaxValueMetric peakActiveComputeTasks; - /** * @param mreg Metrics registry. */ @@ -123,16 +105,6 @@ public ClientListenerMetrics(MetricRegistryImpl mreg) { affQryMisses = mreg.longMetric(AFF_QRY_MISSES, "The number of affinity-aware query requests that were sent not to the primary node"); - peakActiveComputeTasks = mreg.maxValueMetric( - METRIC_PEAK_ACTIVE_COMPUTE_TASKS, - "Maximum number of active compute tasks per thin client connection during the last time interval " + - "(" + MILLISECONDS.toSeconds(DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL) + " seconds by default). " + - "The value is updated when a task is started. System tasks are not counted. A task rejected because " + - "of the limit is counted as the limit.", - DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL, - DFLT_SIZE - ); - for (byte clientType : CLI_TYPES) { String clientLbl = clientTypeLabel(clientType); @@ -174,16 +146,6 @@ public void onHandshakeAccept(byte clientType) { accepted[clientType].increment(); } - /** - * Callback invoked when a compute task is attempted to start by a thin client. - * - * @param activeTasksCnt Number of active compute tasks of the connection, including the invoked task, capped by - * the limit. - */ - public void onComputeTaskCountUpdate(int activeTasksCnt) { - peakActiveComputeTasks.update(activeTasksCnt); - } - /** */ public void onAffinityKeyHit() { affKeyHits.increment(); 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 48264cc4aee87..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 @@ -351,17 +351,19 @@ public void incrementActiveTasksCount() { int cnt = activeTasksCnt.incrementAndGet(); - // Rejected task is counted as the limit, the counter may momentarily exceed it. - ctx.clientListener().metrics().onComputeTaskCountUpdate(Math.min(cnt, maxActiveComputeTasks)); - 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); } /** @@ -371,12 +373,11 @@ public void decrementActiveTasksCount() { int cnt = activeTasksCnt.decrementAndGet(); assert cnt >= 0 : "Unexpected active tasks count: " + cnt; - - ctx.clientListener().metrics().onComputeTaskCountUpdate(cnt); } - /** @return Number of active compute tasks for this connection. System tasks are not counted. */ + /** @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 79b725d0105ce..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; @@ -121,8 +122,9 @@ public String version() { } /** - * @return Number of active compute tasks for the thin client connection, system tasks are not counted. - * {@code null} for other connection types. + * @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() { 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 ea8543e787785..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 @@ -25,7 +25,6 @@ import java.util.concurrent.TimeUnit; import java.util.stream.IntStream; import java.util.stream.StreamSupport; -import javax.management.DynamicMBean; import org.apache.ignite.IgniteException; import org.apache.ignite.Ignition; import org.apache.ignite.client.ClientAuthenticationException; @@ -48,7 +47,6 @@ import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.IgniteInterruptedCheckedException; -import org.apache.ignite.internal.management.api.CommandMBean; 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; @@ -62,17 +60,16 @@ import org.jetbrains.annotations.Nullable; import org.junit.Test; -import static java.util.Collections.singletonList; import static org.apache.ignite.internal.processors.metric.GridMetricManager.CLIENT_CONNECTOR_METRICS; -import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL; import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.METRIC_ACEPTED; -import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.METRIC_PEAK_ACTIVE_COMPUTE_TASKS; 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; @@ -83,11 +80,11 @@ public class ClientListenerMetricsTest extends GridCommonAbstractTest { /** Maximum active tasks. */ private static final int MAX_TASKS = 3; - /** Time interval of the active compute tasks metric, so the peaks do not expire during the test. */ + /** 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; - /** Time interval of the active compute tasks metric to check that the peak expires, long enough to check the peak. */ - private static final long SHORT_INTERVAL = 10_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 { @@ -208,9 +205,8 @@ public void testClientListenerMetricsRejectGeneral() throws Exception { } /** - * Check that the active compute tasks metric reports the peak of the busiest connection (not the sum) and keeps it - * during the time interval, while the {@link ClientListenerProcessor#CLI_CONN_VIEW} system view reports current - * values. + * 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 { @@ -222,48 +218,54 @@ public void testComputeTasksMetrics() throws Exception { IgniteClient client1 = Ignition.startClient(getClientConfiguration()); IgniteClient client2 = Ignition.startClient(getClientConfiguration()) ) { - checkComputeTasks(0, 0, 0); + checkPeakComputeTasks(0); + assertEquals(0, activeComputeTasksCount()); // Tasks 0, 1. futs.addAll(startTasks(client1, 2)); - checkComputeTasks(2, 2, 0); + checkPeakComputeTasks(2); + assertEquals(2, activeComputeTasksCount()); - // Tasks 2, 3. The sum of the connections (4) exceeds even the limit, the metric reports the busiest one. + // Run tasks 2, 3. Metric should not change. futs.addAll(startTasks(client2, 2)); - checkComputeTasks(2, 2, 2); + checkPeakComputeTasks(2); + assertEquals(4, activeComputeTasksCount()); - // Task 4. + // Task 4. Metric should increase. futs.addAll(startTasks(client2, 1)); - checkComputeTasks(MAX_TASKS, 2, MAX_TASKS); + checkPeakComputeTasks(MAX_TASKS); + assertEquals(5, activeComputeTasksCount()); assertComputeTaskRejected(client2); - // Rejected task is not counted in the view. - checkComputeTasks(MAX_TASKS, 2, MAX_TASKS); + // A rejected task counts as the 'MaxActiveComputeTasksPerConnection' limit. + checkPeakComputeTasks(MAX_TASKS); + assertEquals(5, activeComputeTasksCount()); - // Finished tasks are not counted in the view, but the metric keeps the peak during the time interval. - TestComputeTask.finish(2); - checkComputeTasks(MAX_TASKS, 2, MAX_TASKS - 1); + // Finished tasks are released by the connections, but the metric keeps the peak during the time interval. + finishTasks(2); + checkPeakComputeTasks(MAX_TASKS); - TestComputeTask.finish(0); - TestComputeTask.finish(1); - checkComputeTasks(MAX_TASKS, 0, MAX_TASKS - 1); + finishTasks(0, 1); + checkPeakComputeTasks(MAX_TASKS); - TestComputeTask.finish(3); - TestComputeTask.finish(4); - checkComputeTasks(MAX_TASKS, 0, 0); + 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 that the peak expires. Reconfiguration also clears the peak. - configureComputeTasksInterval(SHORT_INTERVAL); - checkComputeTasks(0, 0, 0); + // Shorten the time interval to check the peak expiration without a long wait. + computeTasksMetric().reset(SHORT_INTERVAL); // Task 5. futs.addAll(startTasks(client1, 1)); - checkComputeTasks(1, 1, 0); + checkPeakComputeTasks(1); // The peak expires after the time interval. - TestComputeTask.finish(5); - checkComputeTasks(0, 0, 0); + finishTasks(5); + checkPeakComputeTasks(0); // Wait for all results before clients are closed. for (IgniteInternalFuture fut : futs) @@ -272,8 +274,36 @@ public void testComputeTasksMetrics() throws Exception { } /** - * Check that a task rejected because of the limit updates the metric, so it reports the saturated connection even - * if its tasks were started before the current time interval. + * 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 { @@ -282,18 +312,18 @@ public void testComputeTasksMetricOnRejectedTask() throws Exception { try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { List> futs = startTasks(client, MAX_TASKS); - checkComputeTasks(MAX_TASKS, MAX_TASKS); + checkPeakComputeTasks(MAX_TASKS); - // Clear the accumulated peak, as if it expired while the tasks are still active. + // Clear the peak, as if it expired while the tasks are active. computeTasksMetric().reset(); - checkComputeTasks(0, MAX_TASKS); assertComputeTaskRejected(client); - checkComputeTasks(MAX_TASKS, MAX_TASKS); + // The metric keeps the limit from the rejected task, though the current value is less. + finishTasks(0); + checkPeakComputeTasks(MAX_TASKS); - for (int i = 0; i < MAX_TASKS; i++) - TestComputeTask.finish(i); + finishTasks(IntStream.range(1, MAX_TASKS).toArray()); // Wait for all results before the client is closed. for (IgniteInternalFuture fut : futs) @@ -316,25 +346,30 @@ public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { try (IgniteClient client2 = Ignition.startClient(getClientConfiguration())) { futs2 = startTasks(client2, 2); - checkComputeTasks(2, 1, 2); + checkPeakComputeTasks(2); - connView2 = thinConnections().stream().filter(v -> v.activeComputeTasks() == 2).findFirst().orElseThrow(); + connView2 = thinConnections().stream().filter(v -> v.activeComputeTasks() == 2) + .findFirst().orElseThrow(); } - // Task resources are released on disconnect, so the counter of the closed connection is decremented. + // Tasks of the closed connection are released on disconnect. assertTrue(waitForCondition(() -> connView2.activeComputeTasks() == 0, getTestTimeout())); - // Closed connection is not shown, the metric keeps the peak during the time interval. - checkComputeTasks(2, 1); + // 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 task execution fails on the client side, since the connection is closed. The error depends on - // whether the client has received the execute response before the close. + // 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) - GridTestUtils.assertThrowsWithCause(() -> fut.get(getTestTimeout(), TimeUnit.MILLISECONDS), ClientException.class); + assertThrowsWithCause(() -> fut.get(getTestTimeout(), TimeUnit.MILLISECONDS), ClientException.class); // Last task of 'client1'. - TestComputeTask.finish(0); - checkComputeTasks(2, 0); + finishTasks(0); + checkPeakComputeTasks(2); // Wait for the result before the client is closed. for (IgniteInternalFuture fut : futs1) @@ -342,13 +377,13 @@ public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { } } - /** Check the active compute tasks metric and system view when compute is disabled for thin clients. */ + /** 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())) { - checkComputeTasks(0, 0); + checkPeakComputeTasks(0); GridTestUtils.assertThrowsAnyCause( log, @@ -357,13 +392,12 @@ public void testComputeTasksMetricsComputeDisabled() throws Exception { "Compute grid functionality is disabled for thin clients" ); - checkComputeTasks(0, 0); + checkPeakComputeTasks(0); } } /** - * Starts a node with enabled compute for thin clients. The time interval of the active compute tasks metric is - * changed to {@link #LONG_INTERVAL}, so the peaks do not expire during the test. + * 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)) @@ -371,29 +405,7 @@ private void startGridWithCompute() throws Exception { .setThinClientConfiguration(new ThinClientConfiguration() .setMaxActiveComputeTasksPerConnection(MAX_TASKS)))); - assertEquals(DFLT_PEAK_ACTIVE_COMPUTE_TASKS_INTERVAL, computeTasksMetric().timeInterval()); - - configureComputeTasksInterval(LONG_INTERVAL); - } - - /** - * Changes the time interval of the active compute tasks metric the same way as - * {@code control.sh --metric --configure-max-value} does. - * - * @param interval Time interval in milliseconds. - */ - private void configureComputeTasksInterval(long interval) throws Exception { - DynamicMBean mbean = getMxBean(grid(0).name(), "management", singletonList("Metric"), "ConfigureMaxValue", - DynamicMBean.class); - - mbean.invoke( - CommandMBean.INVOKE, - new Object[] {MetricUtils.metricName(CLIENT_CONNECTOR_METRICS, METRIC_PEAK_ACTIVE_COMPUTE_TASKS), - Long.toString(interval), null}, - new String[] {String.class.getName(), String.class.getName(), String.class.getName()} - ); - - assertTrue(waitForCondition(() -> computeTasksMetric().timeInterval() == interval, getTestTimeout())); + computeTasksMetric().reset(LONG_INTERVAL); } /** @@ -424,50 +436,44 @@ private List> startTasks(IgniteClient client, int c .mapToObj(i -> runAsync(() -> client.compute().execute(TestComputeTask.class.getName(), null))) .toList(); - // Task instances are created after the active tasks counter is incremented, so wait for them to keep the order. + // 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; } /** - * Checks the active compute tasks metric via the metric registry and JMX, and the current active compute tasks of - * the thin client connections via the system view. + * Finishes tasks and waits for them to be released by the connections. * - * @param expMetric Expected value of the metric. - * @param expConnTasks Expected active compute tasks of the thin client connections, in any order. + * @param idxs Indexes of the tasks. */ - private void checkComputeTasks(long expMetric, int... expConnTasks) throws Exception { - List exp = IntStream.of(expConnTasks).sorted().boxed().toList(); - - // Counter is decremented after the result is sent to the client, so wait for the values. - assertTrue(waitForCondition(() -> exp.equals(thinConnectionsComputeTasks()), getTestTimeout())); + private void finishTasks(int... idxs) throws Exception { + int expTasks = activeComputeTasksCount() - idxs.length; - assertTrue(waitForCondition(() -> computeTasksMetric().value() == expMetric, getTestTimeout())); + for (int idx : idxs) + TestComputeTask.finish(idx); - assertTrue(waitForCondition(() -> jmxComputeTasksMetric() == expMetric, getTestTimeout())); + // Counter is decremented after the result is sent to the client. + assertTrue(waitForCondition(() -> activeComputeTasksCount() == expTasks, 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 Total active compute tasks of the thin client connections from the system view. */ + private int activeComputeTasksCount() { + return thinConnections().stream().mapToInt(ClientConnectionView::activeComputeTasks).sum(); } - /** @return Active compute tasks metric value read via JMX. */ - private long jmxComputeTasksMetric() { - DynamicMBean mbean = metricRegistry(grid(0).name(), "client", "connector"); - - try { - return (Long)mbean.getAttribute(METRIC_PEAK_ACTIVE_COMPUTE_TASKS); - } - catch (Exception e) { - throw new IgniteException(e); - } + /** + * 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 Sorted active compute tasks of the thin client connections from the system view. */ - private List thinConnectionsComputeTasks() { - return thinConnections().stream().map(ClientConnectionView::activeComputeTasks).sorted().toList(); + /** @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. */ @@ -574,7 +580,7 @@ public static void finish(int idx) { TASKS.get(idx).finishLatch.countDown(); } - /** Clears static state. */ + /** Finishes all tasks and clears the task list. */ public static void reset() { TASKS.forEach(t -> t.finishLatch.countDown());