From 607f35fee1ad643f00920b88c8b1935eb8043701 Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Tue, 6 Oct 2026 13:34:24 +0300 Subject: [PATCH 01/12] 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 435b7741ae23c..a7ebc54368483 100644 --- a/docs/_docs/monitoring-metrics/new-metrics.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics.adoc @@ -486,6 +486,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 5b7548b0c91aec81062d53479542dfb49f100e9f Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Thu, 15 Oct 2026 12:05:34 +0300 Subject: [PATCH 02/12] 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 a7ebc54368483..837af75027644 100644 --- a/docs/_docs/monitoring-metrics/new-metrics.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics.adoc @@ -471,6 +471,7 @@ Register name: `client.connector` |AffinityKeyRequestsMisses | long | The number of affinity-aware cache key requests that were sent not to the primary node. |AffinityQueryRequestsHits | long | The number of affinity-aware query requests that were sent to the primary node. |AffinityQueryRequestsMisses | long | The number of affinity-aware query requests that were sent not to the primary node. +|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. @@ -486,7 +487,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 ea9cd2117ab37..e1356f7f5f8fd 100644 --- a/docs/_docs/monitoring-metrics/system-views.adoc +++ b/docs/_docs/monitoring-metrics/system-views.adoc @@ -444,6 +444,7 @@ This view exposes information about currently opened client connections: JDBC, O |USER | string | User name |VERSION | string | Protocol version |DATA_CENTER_ID | string | Effective data center ID reported by the client +|ACTIVE_COMPUTE_TASKS | int | Number of active 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 d4dd8d9400528..1d3686450a19a 100644 --- a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java +++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java @@ -951,6 +951,7 @@ public void testGetAllColumns() throws Exception { "SYS.CLIENT_CONNECTIONS.USER.null", "SYS.CLIENT_CONNECTIONS.VERSION.null", "SYS.CLIENT_CONNECTIONS.DATA_CENTER_ID.null", + "SYS.CLIENT_CONNECTIONS.ACTIVE_COMPUTE_TASKS.null", "SYS.CLIENT_CONNECTION_ATTRIBUTES.CONNECTION_ID.null", "SYS.CLIENT_CONNECTION_ATTRIBUTES.NAME.null", "SYS.CLIENT_CONNECTION_ATTRIBUTES.VALUE.null", diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/SystemViewCommandTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/SystemViewCommandTest.java index 2e950c2eb35e8..3d485c6fdd88b 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/util/SystemViewCommandTest.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/util/SystemViewCommandTest.java @@ -603,7 +603,9 @@ public void testViewColumns() { asList("USER", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", String.class.getName()), asList("VERSION", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", String.class.getName()), asList("DATA_CENTER_ID", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", - String.class.getName()) + String.class.getName()), + asList("ACTIVE_COMPUTE_TASKS", "CLIENT_CONNECTIONS", SCHEMA_SYS, "null", "true", "-1", "-1", + Integer.class.getName()) )); Set> sqlViewColumnsView = systemView(ignite0, SQL_VIEW_COLS_VIEW).stream() diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/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 a0c5e18641dc63be919e0de0eb57a952264a0f1d Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Thu, 15 Oct 2026 18:43:50 +0300 Subject: [PATCH 03/12] 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 837af75027644..b7f06f2c53ebc 100644 --- a/docs/_docs/monitoring-metrics/new-metrics.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics.adoc @@ -471,7 +471,7 @@ Register name: `client.connector` |AffinityKeyRequestsMisses | long | The number of affinity-aware cache key requests that were sent not to the primary node. |AffinityQueryRequestsHits | long | The number of affinity-aware query requests that were sent to the primary node. |AffinityQueryRequestsMisses | long | The number of affinity-aware query requests that were sent not to the primary node. -|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 e1356f7f5f8fd..236e3f2492b24 100644 --- a/docs/_docs/monitoring-metrics/system-views.adoc +++ b/docs/_docs/monitoring-metrics/system-views.adoc @@ -444,7 +444,7 @@ This view exposes information about currently opened client connections: JDBC, O |USER | string | User name |VERSION | string | Protocol version |DATA_CENTER_ID | string | Effective data center ID reported by the client -|ACTIVE_COMPUTE_TASKS | int | Number of active 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()); From 1f01429d75c248ebc88a996649626b4a5060db6e Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Fri, 23 Oct 2026 12:24:23 +0300 Subject: [PATCH 04/12] IGNITE-29155 Code review fixes + javadoc reduced --- .../new-metrics-system.adoc | 2 +- .../_docs/monitoring-metrics/new-metrics.adoc | 2 +- .../monitoring-metrics/system-views.adoc | 2 +- .../odbc/ClientListenerProcessor.java | 57 +++--- .../client/ClientConnectionContext.java | 6 +- .../systemview/view/ClientConnectionView.java | 7 +- .../odbc/ClientListenerMetricsTest.java | 170 ++++++++---------- 7 files changed, 106 insertions(+), 140 deletions(-) diff --git a/docs/_docs/monitoring-metrics/new-metrics-system.adoc b/docs/_docs/monitoring-metrics/new-metrics-system.adoc index c70a6466985da..26fcb55ac2152 100644 --- a/docs/_docs/monitoring-metrics/new-metrics-system.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics-system.adoc @@ -419,7 +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]). -* `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]). +* `MaxCurrentActiveComputeTasksPerConnection` - The maximum number of active user compute tasks per thin client connection over the configured time interval. 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 b7f06f2c53ebc..b0b98136440cb 100644 --- a/docs/_docs/monitoring-metrics/new-metrics.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics.adoc @@ -471,7 +471,7 @@ Register name: `client.connector` |AffinityKeyRequestsMisses | long | The number of affinity-aware cache key requests that were sent not to the primary node. |AffinityQueryRequestsHits | long | The number of affinity-aware query requests that were sent to the primary node. |AffinityQueryRequestsMisses | long | The number of affinity-aware query requests that were sent not to the primary node. -|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. +|MaxCurrentActiveComputeTasksPerConnection | long | Maximum number of currently active user compute tasks per thin client connection over the configured time interval (90 seconds by default). |RejectedSessionsAuthenticationFailed | integer | TCP sessions count that were rejected due to failed authentication. |RejectedSessionsTimeout | integer | TCP sessions count that were rejected due to handshake timeout. |RejectedSessionsTotal | integer | Total number of rejected TCP connections. diff --git a/docs/_docs/monitoring-metrics/system-views.adoc b/docs/_docs/monitoring-metrics/system-views.adoc index 236e3f2492b24..b17b6d78ebc38 100644 --- a/docs/_docs/monitoring-metrics/system-views.adoc +++ b/docs/_docs/monitoring-metrics/system-views.adoc @@ -444,7 +444,7 @@ This view exposes information about currently opened client connections: JDBC, O |USER | string | User name |VERSION | string | Protocol version |DATA_CENTER_ID | string | Effective data center ID reported by the client -|ACTIVE_COMPUTE_TASKS | int | Number of active user compute tasks of a thin client connection, not greater than the `ThinClientConfiguration.maxActiveComputeTasksPerConnection` configuration parameter. `null` for JDBC and ODBC connections +|ACTIVE_COMPUTE_TASKS | int | Number of active user compute tasks of a thin client connection, `null` for JDBC and ODBC connections |=== == CLIENT_CONNECTION_ATTRIBUTES diff --git a/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 07bf3b6372d51..c6c86d25deffb 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 @@ -104,11 +104,11 @@ 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"; + /** */ + public static final String METRIC_MAX_CURRENT_ACTIVE_COMPUTE_TASKS = "MaxCurrentActiveComputeTasksPerConnection"; - /** 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 time interval of the {@link #METRIC_MAX_CURRENT_ACTIVE_COMPUTE_TASKS} metric in milliseconds. */ + public static final long DFLT_MAX_CURRENT_ACTIVE_COMPUTE_TASKS_INTERVAL = 90_000L; /** Default client connector configuration. */ public static final ClientConnectorConfiguration DFLT_CLI_CFG = new ClientConnectorConfigurationEx(); @@ -131,8 +131,8 @@ public class ClientListenerProcessor extends GridProcessorAdapter { /** Metrics. */ private ClientListenerMetrics metrics; - /** Peak of active user compute tasks per thin client connection. */ - private PeakWithCurrentMetric peakActiveComputeTasks; + /** */ + private MaxWithCurrentValueMetric maxCurActiveComputeTasks; /** Executor service. */ private ExecutorService execSvc; @@ -371,22 +371,22 @@ private void registerClientMetrics(MetricRegistryImpl mreg) { : thinCfg.getMaxActiveComputeTasksPerConnection(); // Registered before the server is started, since connections update it. - peakActiveComputeTasks = new PeakWithCurrentMetric( - metricName(mreg.name(), METRIC_PEAK_ACTIVE_COMPUTE_TASKS), + maxCurActiveComputeTasks = new MaxWithCurrentValueMetric( + metricName(mreg.name(), METRIC_MAX_CURRENT_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_MAX_CURRENT_ACTIVE_COMPUTE_TASKS_INTERVAL, DFLT_SIZE, maxActiveComputeTasks, - this::peakActiveComputeTasks + this::currentMaxActiveComputeTasks ); - mreg.register(peakActiveComputeTasks); + mreg.register(maxCurActiveComputeTasks); } - /** @return Current peak number of active user compute tasks across thin client connections. */ - private long peakActiveComputeTasks() { + /** @return Maximum of active user compute tasks across thin client connections. */ + private long currentMaxActiveComputeTasks() { long max = 0; for (GridNioSession ses : srv.sessions()) { @@ -663,13 +663,9 @@ 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); + /** @param activeTasksCnt Number of active user compute tasks of a connection. */ + public void updateMaxCurrentActiveComputeTasks(int activeTasksCnt) { + maxCurActiveComputeTasks.update(activeTasksCnt); } /** @@ -912,22 +908,15 @@ 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 static class MaxWithCurrentValueMetric extends MaxValueMetric { + /** */ 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) { + /** */ + MaxWithCurrentValueMetric(String name, String desc, long timeInterval, int size, long limit, LongSupplier cur) { super(name, desc, timeInterval, size); this.limit = limit; @@ -936,10 +925,10 @@ private static class PeakWithCurrentMetric extends MaxValueMetric { /** {@inheritDoc} */ @Override public long value() { - long peak = super.value(); + long val = 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()); + return val >= limit ? val : Math.max(val, 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 3623d4613feb9..36b5b3703e165 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 @@ -355,7 +355,7 @@ public void incrementActiveTasksCount() { activeTasksCnt.decrementAndGet(); // A rejected task counts as the limit. - ctx.clientListener().updatePeakActiveComputeTasks(maxActiveComputeTasks); + ctx.clientListener().updateMaxCurrentActiveComputeTasks(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 " + @@ -363,7 +363,7 @@ public void incrementActiveTasksCount() { "ThinClientConfiguration.MaxActiveComputeTasksPerConnection property."); } - ctx.clientListener().updatePeakActiveComputeTasks(cnt); + ctx.clientListener().updateMaxCurrentActiveComputeTasks(cnt); } /** @@ -375,7 +375,7 @@ public void decrementActiveTasksCount() { assert cnt >= 0 : "Unexpected active tasks count: " + cnt; } - /** @return Number of active user compute tasks for this connection capped by the limit. */ + /** */ public int activeTasksCount() { // The counter briefly exceeds the limit while a task is being rejected, so the value is capped by the limit. return Math.min(activeTasksCnt.get(), maxActiveComputeTasks); diff --git a/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ClientConnectionView.java b/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ClientConnectionView.java index b849973352971..8644de773d18d 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,7 +18,6 @@ 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,11 +120,7 @@ public String version() { return ctx == null ? null : ctx.attributes().get(IGNITE_DATA_CENTER_ID); } - /** - * @return Number of active user compute tasks of the thin client connection, not greater than - * {@link ThinClientConfiguration#getMaxActiveComputeTasksPerConnection()}; {@code null} for other connection - * types. - */ + /** @return Number of active user compute tasks of the thin client connection, {@code null} for other connection types. */ @Order(7) @Nullable public Integer activeComputeTasks() { return ctx instanceof ClientConnectionContext ? ((ClientConnectionContext)ctx).activeTasksCount() : null; diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java index d4d5425296225..f30f9a9062421 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 @@ -24,7 +24,6 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.stream.IntStream; -import java.util.stream.StreamSupport; import org.apache.ignite.IgniteException; import org.apache.ignite.Ignition; import org.apache.ignite.client.ClientAuthenticationException; @@ -49,8 +48,8 @@ import org.apache.ignite.internal.IgniteInterruptedCheckedException; import org.apache.ignite.internal.processors.metric.impl.MaxValueMetric; import org.apache.ignite.internal.processors.metric.impl.MetricUtils; -import org.apache.ignite.internal.util.lang.GridAbsPredicate; import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.metric.MetricRegistry; import org.apache.ignite.spi.metric.IntMetric; import org.apache.ignite.spi.systemview.view.ClientConnectionView; @@ -67,7 +66,9 @@ 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.internal.processors.odbc.ClientListenerProcessor.METRIC_MAX_CURRENT_ACTIVE_COMPUTE_TASKS; +import static org.apache.ignite.internal.util.GridTestClockTimer.DFLT_TIME_SUPPLIER; +import static org.apache.ignite.internal.util.GridTestClockTimer.timeSupplier; import static org.apache.ignite.ssl.SslContextFactory.DFLT_STORE_TYPE; import static org.apache.ignite.testframework.GridTestUtils.assertThrowsWithCause; import static org.apache.ignite.testframework.GridTestUtils.runAsync; @@ -77,15 +78,12 @@ * Client listener metrics tests. */ public class ClientListenerMetricsTest extends GridCommonAbstractTest { - /** Maximum active tasks. */ + /** */ private static final int MAX_TASKS = 3; - /** Metric time interval long enough for the peak not to expire during a test. Expiration is emulated by a reset. */ + /** Metric time interval. Expiration is emulated by a time shift. */ private static final long LONG_INTERVAL = 10 * 60_000L; - /** Metric time interval to check the peak expiration without a long wait. */ - private static final long SHORT_INTERVAL = 2_000L; - /** {@inheritDoc} */ @Override protected void afterTest() throws Exception { TestComputeTask.reset(); @@ -205,7 +203,7 @@ public void testClientListenerMetricsRejectGeneral() throws Exception { } /** - * Check that the active compute tasks metric reports the peak of the busiest connection (not the sum), keeps it + * Check that the active compute tasks metric reports the maximum of the busiest connection (not the sum), keeps it * during the time interval and drops it after the time interval. */ @Test @@ -218,63 +216,64 @@ public void testComputeTasksMetrics() throws Exception { IgniteClient client1 = Ignition.startClient(getClientConfiguration()); IgniteClient client2 = Ignition.startClient(getClientConfiguration()) ) { - checkPeakComputeTasks(0); + checkMaxComputeTasks(0); assertEquals(0, activeComputeTasksCount()); // Tasks 0, 1. futs.addAll(startTasks(client1, 2)); - checkPeakComputeTasks(2); + checkMaxComputeTasks(2); assertEquals(2, activeComputeTasksCount()); // Run tasks 2, 3. Metric should not change. futs.addAll(startTasks(client2, 2)); - checkPeakComputeTasks(2); + checkMaxComputeTasks(2); assertEquals(4, activeComputeTasksCount()); // Task 4. Metric should increase. futs.addAll(startTasks(client2, 1)); - checkPeakComputeTasks(MAX_TASKS); + checkMaxComputeTasks(MAX_TASKS); assertEquals(5, activeComputeTasksCount()); assertComputeTaskRejected(client2); // A rejected task counts as the 'MaxActiveComputeTasksPerConnection' limit. - checkPeakComputeTasks(MAX_TASKS); + checkMaxComputeTasks(MAX_TASKS); assertEquals(5, activeComputeTasksCount()); - // Finished tasks are released by the connections, but the metric keeps the peak during the time interval. + // Finished tasks are released by the connections, but the metric keeps the maximum during the time interval. finishTasks(2); - checkPeakComputeTasks(MAX_TASKS); + checkMaxComputeTasks(MAX_TASKS); finishTasks(0, 1); - checkPeakComputeTasks(MAX_TASKS); + checkMaxComputeTasks(MAX_TASKS); finishTasks(3, 4); - checkPeakComputeTasks(MAX_TASKS); + checkMaxComputeTasks(MAX_TASKS); - // Clear the peak, as if it expired. No tasks are active, so the metric reports zero. - computeTasksMetric().reset(); + // The maximum expires after the time interval. + shiftTime(LONG_INTERVAL); assertEquals(0, computeTasksMetric().value()); - // Shorten the time interval to check the peak expiration without a long wait. - computeTasksMetric().reset(SHORT_INTERVAL); - // Task 5. futs.addAll(startTasks(client1, 1)); - checkPeakComputeTasks(1); + checkMaxComputeTasks(1); - // The peak expires after the time interval. finishTasks(5); - checkPeakComputeTasks(0); + + shiftTime(2 * LONG_INTERVAL); + checkMaxComputeTasks(0); // Wait for all results before clients are closed. for (IgniteInternalFuture fut : futs) fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); } + finally { + timeSupplier(DFLT_TIME_SUPPLIER); + } } /** - * Check that the metric is not less than the current value, so it reports long-running tasks after their peak has + * Check that the metric is not less than the current value, so it reports long-running tasks after their maximum has * expired. */ @Test @@ -284,21 +283,24 @@ public void testComputeTasksMetricOnLongRunningTasks() throws Exception { try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { List> futs = startTasks(client, MAX_TASKS); - checkPeakComputeTasks(MAX_TASKS); + checkMaxComputeTasks(MAX_TASKS); - // Clear the peak, as if it expired while the tasks are active. The metric reports the current value. - computeTasksMetric().reset(); + // The maximum expires after the time interval. + shiftTime(LONG_INTERVAL); 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. + // Finished tasks do not update the maximum, 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); } + finally { + timeSupplier(DFLT_TIME_SUPPLIER); + } } /** @@ -312,16 +314,16 @@ public void testComputeTasksMetricOnRejectedTask() throws Exception { try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { List> futs = startTasks(client, MAX_TASKS); - checkPeakComputeTasks(MAX_TASKS); + checkMaxComputeTasks(MAX_TASKS); - // Clear the peak, as if it expired while the tasks are active. - computeTasksMetric().reset(); + // The maximum expires after the time interval. The time stays shifted, so the expired maximum is not counted. + shiftTime(LONG_INTERVAL); assertComputeTaskRejected(client); // The metric keeps the limit from the rejected task, though the current value is less. finishTasks(0); - checkPeakComputeTasks(MAX_TASKS); + checkMaxComputeTasks(MAX_TASKS); finishTasks(IntStream.range(1, MAX_TASKS).toArray()); @@ -329,6 +331,9 @@ public void testComputeTasksMetricOnRejectedTask() throws Exception { for (IgniteInternalFuture fut : futs) fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); } + finally { + timeSupplier(DFLT_TIME_SUPPLIER); + } } /** Check that closing the connection with unfinished tasks releases its tasks. */ @@ -339,28 +344,27 @@ public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { try (IgniteClient client1 = Ignition.startClient(getClientConfiguration())) { List> futs1 = startTasks(client1, 1); - List> futs2; + assertEquals(1, activeComputeTasksCount()); - ClientConnectionView connView2; + List> futs2; try (IgniteClient client2 = Ignition.startClient(getClientConfiguration())) { futs2 = startTasks(client2, 2); - checkPeakComputeTasks(2); + checkMaxComputeTasks(2); - connView2 = thinConnections().stream().filter(v -> v.activeComputeTasks() == 2) - .findFirst().orElseThrow(); + assertEquals(3, activeComputeTasksCount()); } // Tasks of the closed connection are released on disconnect. - assertTrue(waitForCondition(() -> connView2.activeComputeTasks() == 0, getTestTimeout())); + assertEquals(1, activeComputeTasksCount()); // 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); + // The metric keeps the maximum during the time interval. + checkMaxComputeTasks(2); // Unfinished tasks fail on the client side. The error depends on whether the client received the execute // response before the close. @@ -369,7 +373,7 @@ public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { // Last task of 'client1'. finishTasks(0); - checkPeakComputeTasks(2); + checkMaxComputeTasks(2); // Wait for the result before the client is closed. for (IgniteInternalFuture fut : futs1) @@ -377,13 +381,13 @@ public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { } } - /** Check the active compute tasks metric when compute is disabled for thin clients. */ + /** */ @Test public void testComputeTasksMetricsComputeDisabled() throws Exception { startGrid(0); try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { - checkPeakComputeTasks(0); + checkMaxComputeTasks(0); GridTestUtils.assertThrowsAnyCause( log, @@ -392,7 +396,7 @@ public void testComputeTasksMetricsComputeDisabled() throws Exception { "Compute grid functionality is disabled for thin clients" ); - checkPeakComputeTasks(0); + checkMaxComputeTasks(0); } } @@ -408,11 +412,7 @@ private void startGridWithCompute() throws Exception { computeTasksMetric().reset(LONG_INTERVAL); } - /** - * Checks that a task is rejected because of the limit. - * - * @param client Ignite client. - */ + /** */ private void assertComputeTaskRejected(IgniteClient client) { GridTestUtils.assertThrowsAnyCause( log, @@ -422,13 +422,7 @@ private void assertComputeTaskRejected(IgniteClient client) { ); } - /** - * Starts tasks via client and waits for them to be started on the node. - * - * @param client Ignite client. - * @param cnt Tasks count. - * @return Task futures. - */ + /** Starts tasks and waits for them to be started on the node. */ private List> startTasks(IgniteClient client, int cnt) throws Exception { int expTasks = TestComputeTask.TASKS.size() + cnt; @@ -436,17 +430,13 @@ private List> startTasks(IgniteClient client, int c .mapToObj(i -> runAsync(() -> client.compute().execute(TestComputeTask.class.getName(), null))) .toList(); - // A task instance is created after the counter is incremented. Wait for instances to keep the task order. + // A task is registered in 'map()' after the counter is incremented. Wait for the tasks to keep their order. assertTrue(waitForCondition(() -> TestComputeTask.TASKS.size() == expTasks, getTestTimeout())); return futs; } - /** - * Finishes tasks and waits for them to be released by the connections. - * - * @param idxs Indexes of the tasks. - */ + /** Finishes tasks and waits for them to be released by the connections. */ private void finishTasks(int... idxs) throws Exception { int expTasks = activeComputeTasksCount() - idxs.length; @@ -457,30 +447,32 @@ private void finishTasks(int... idxs) throws Exception { assertTrue(waitForCondition(() -> activeComputeTasksCount() == expTasks, getTestTimeout())); } - /** @return Total active compute tasks of the thin client connections from the system view. */ + /** @return Total active compute tasks of the thin client connections. */ private int activeComputeTasksCount() { return thinConnections().stream().mapToInt(ClientConnectionView::activeComputeTasks).sum(); } - /** - * Waits for the expected value of the active compute tasks metric. - * - * @param expMetric Expected value of the metric. - */ - private void checkPeakComputeTasks(long expMetric) throws Exception { + /** */ + private void checkMaxComputeTasks(long expMetric) throws Exception { assertTrue(waitForCondition(() -> computeTasksMetric().value() == expMetric, getTestTimeout())); } - /** @return Active compute tasks metric. */ + /** */ private MaxValueMetric computeTasksMetric() { - return grid(0).context().metric().registry(CLIENT_CONNECTOR_METRICS).findMetric(METRIC_PEAK_ACTIVE_COMPUTE_TASKS); + return grid(0).context().metric().registry(CLIENT_CONNECTOR_METRICS) + .findMetric(METRIC_MAX_CURRENT_ACTIVE_COMPUTE_TASKS); + } + + /** Shifts the time returned by {@link U#currentTimeMillis()} relative to the system time. */ + private static void shiftTime(long offset) { + timeSupplier(() -> System.currentTimeMillis() + offset); } - /** @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(); + return U.arrayList(view.iterator(), view.size(), v -> "THIN".equals(v.type())); } /** */ @@ -502,11 +494,8 @@ private static ClientConfiguration getClientConfiguration() { */ private void waitForMetricValue(MetricRegistry mreg, String metric, long value, long timeout) throws IgniteInterruptedCheckedException { - waitForCondition(new GridAbsPredicate() { - @Override public boolean apply() { - return mreg.findMetric(metric).value() == value; - } - }, timeout); + waitForCondition(() -> mreg.findMetric(metric).value() == value, timeout); + assertEquals(mreg.findMetric(metric).value(), value); } @@ -543,19 +532,16 @@ private void checkConnectionsMetrics(MetricRegistry mreg, int accepted, int acti /** 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) { + TASKS.add(this); + return F.asMap(new NoopJob(), subgrid.get(0)); } @@ -571,11 +557,7 @@ public TestComputeTask() { return null; } - /** - * Finishes task with a specified index. - * - * @param idx Index of a task. - */ + /** @param idx Index of a task. */ public static void finish(int idx) { TASKS.get(idx).finishLatch.countDown(); } @@ -588,7 +570,7 @@ public static void reset() { } } - /** No-op job. */ + /** */ private static class NoopJob extends ComputeJobAdapter { /** {@inheritDoc} */ @Override public Object execute() { From 665a244b9ae67dfd48dd7df95669b45fe1e4b257 Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Fri, 23 Oct 2026 15:00:58 +0300 Subject: [PATCH 05/12] IGNITE-29155 Fix test, JavaDoc --- .../odbc/ClientListenerMetricsTest.java | 40 ++++++------------- 1 file changed, 13 insertions(+), 27 deletions(-) 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 f30f9a9062421..510ba8537e38c 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 @@ -203,8 +203,8 @@ public void testClientListenerMetricsRejectGeneral() throws Exception { } /** - * Check that the active compute tasks metric reports the maximum of the busiest connection (not the sum), keeps it - * during the time interval and drops it after the time interval. + * Check that the active compute tasks metric reports the maximum among all connections and keeps it + * during the time interval. */ @Test public void testComputeTasksMetrics() throws Exception { @@ -234,23 +234,16 @@ public void testComputeTasksMetrics() throws Exception { checkMaxComputeTasks(MAX_TASKS); assertEquals(5, activeComputeTasksCount()); + // Metric should not exceed limit after a rejected task. assertComputeTaskRejected(client2); - - // A rejected task counts as the 'MaxActiveComputeTasksPerConnection' limit. checkMaxComputeTasks(MAX_TASKS); assertEquals(5, activeComputeTasksCount()); - // Finished tasks are released by the connections, but the metric keeps the maximum during the time interval. - finishTasks(2); - checkMaxComputeTasks(MAX_TASKS); - - finishTasks(0, 1); - checkMaxComputeTasks(MAX_TASKS); - - finishTasks(3, 4); + // Finished tasks should not reset the maximum. + finishTasks(0, 1, 2, 3, 4); checkMaxComputeTasks(MAX_TASKS); - // The maximum expires after the time interval. + // "Wait" for expiration. shiftTime(LONG_INTERVAL); assertEquals(0, computeTasksMetric().value()); @@ -263,7 +256,6 @@ public void testComputeTasksMetrics() throws Exception { shiftTime(2 * LONG_INTERVAL); checkMaxComputeTasks(0); - // Wait for all results before clients are closed. for (IgniteInternalFuture fut : futs) fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); } @@ -272,10 +264,7 @@ public void testComputeTasksMetrics() throws Exception { } } - /** - * Check that the metric is not less than the current value, so it reports long-running tasks after their maximum has - * expired. - */ + /** Check that the metric is not less than the current maximum, when metric has expired. */ @Test public void testComputeTasksMetricOnLongRunningTasks() throws Exception { startGridWithCompute(); @@ -285,16 +274,14 @@ public void testComputeTasksMetricOnLongRunningTasks() throws Exception { checkMaxComputeTasks(MAX_TASKS); - // The maximum expires after the time interval. + // "Wait" for expiration. shiftTime(LONG_INTERVAL); assertEquals(MAX_TASKS, computeTasksMetric().value()); finishTasks(IntStream.range(0, MAX_TASKS).toArray()); - // Finished tasks do not update the maximum, 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); } @@ -316,18 +303,18 @@ public void testComputeTasksMetricOnRejectedTask() throws Exception { checkMaxComputeTasks(MAX_TASKS); - // The maximum expires after the time interval. The time stays shifted, so the expired maximum is not counted. + // "Wait" for expiration. shiftTime(LONG_INTERVAL); assertComputeTaskRejected(client); - // The metric keeps the limit from the rejected task, though the current value is less. finishTasks(0); + + // The metric keeps the max limit, though the current value is less. checkMaxComputeTasks(MAX_TASKS); finishTasks(IntStream.range(1, MAX_TASKS).toArray()); - // Wait for all results before the client is closed. for (IgniteInternalFuture fut : futs) fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); } @@ -356,10 +343,9 @@ public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { assertEquals(3, activeComputeTasksCount()); } - // Tasks of the closed connection are released on disconnect. - assertEquals(1, activeComputeTasksCount()); + // Tasks of the closed client should be released on disconnect + assertTrue(waitForCondition(() -> activeComputeTasksCount() == 1, getTestTimeout())); - // Cancellation does not interrupt 'reduce()', so unblock the tasks of 'client2'. TestComputeTask.finish(1); TestComputeTask.finish(2); From 0049de7861583c4f051de1ac925570b184f74a65 Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Mon, 26 Oct 2026 09:32:19 +0300 Subject: [PATCH 06/12] WIP --- .../new-metrics-system.adoc | 2 +- .../_docs/monitoring-metrics/new-metrics.adoc | 2 +- .../odbc/ClientListenerProcessor.java | 44 ++++++++----------- .../odbc/ClientListenerMetricsTest.java | 6 +-- 4 files changed, 23 insertions(+), 31 deletions(-) diff --git a/docs/_docs/monitoring-metrics/new-metrics-system.adoc b/docs/_docs/monitoring-metrics/new-metrics-system.adoc index 26fcb55ac2152..094400c9f5b22 100644 --- a/docs/_docs/monitoring-metrics/new-metrics-system.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics-system.adoc @@ -419,7 +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]). -* `MaxCurrentActiveComputeTasksPerConnection` - The maximum number of active user compute tasks per thin client connection over the configured time interval. 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]). +* `MaxRunningComputeTasksPerConnection` - The maximum number of running user compute tasks per thin client connection over the configured time interval. The interval is 60 seconds by default and can be changed with `control.sh --metric --configure-max-value` (see link:tools/control-script#metric-configure-command[Metric configure command]). Set it longer than the polling period of your monitoring system, so peaks between polls are not missed (`client.connector` register, see link:monitoring-metrics/new-metrics#ignite-thin-client-connector[Ignite Thin Client Connector]). * The link:monitoring-metrics/system-views#client_connections[CLIENT_CONNECTIONS] system view lists every connection with its user, addresses and the current number of active user compute tasks. * `Connections` - Attribute of the `ClientProcessorMXBean` MBean, not available in the metric registers. A list of strings, each string containing information about a connection: diff --git a/docs/_docs/monitoring-metrics/new-metrics.adoc b/docs/_docs/monitoring-metrics/new-metrics.adoc index b0b98136440cb..b055dc4069544 100644 --- a/docs/_docs/monitoring-metrics/new-metrics.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics.adoc @@ -471,7 +471,7 @@ Register name: `client.connector` |AffinityKeyRequestsMisses | long | The number of affinity-aware cache key requests that were sent not to the primary node. |AffinityQueryRequestsHits | long | The number of affinity-aware query requests that were sent to the primary node. |AffinityQueryRequestsMisses | long | The number of affinity-aware query requests that were sent not to the primary node. -|MaxCurrentActiveComputeTasksPerConnection | long | Maximum number of currently active user compute tasks per thin client connection over the configured time interval (90 seconds by default). +|MaxRunningComputeTasksPerConnection | long | Maximum number of running user compute tasks per thin client connection over the configured time interval (60 seconds by default). |RejectedSessionsAuthenticationFailed | integer | TCP sessions count that were rejected due to failed authentication. |RejectedSessionsTimeout | integer | TCP sessions count that were rejected due to handshake timeout. |RejectedSessionsTotal | integer | Total number of rejected TCP connections. diff --git a/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 c6c86d25deffb..ab62147e5b260 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 @@ -105,10 +105,10 @@ public class ClientListenerProcessor extends GridProcessorAdapter { public static final String METRIC_ACTIVE = "ActiveSessions"; /** */ - public static final String METRIC_MAX_CURRENT_ACTIVE_COMPUTE_TASKS = "MaxCurrentActiveComputeTasksPerConnection"; + public static final String METRIC_MAX_RUNNING_COMPUTE_TASKS = "MaxRunningComputeTasksPerConnection"; - /** Default time interval of the {@link #METRIC_MAX_CURRENT_ACTIVE_COMPUTE_TASKS} metric in milliseconds. */ - public static final long DFLT_MAX_CURRENT_ACTIVE_COMPUTE_TASKS_INTERVAL = 90_000L; + /** Default time interval of the {@link #METRIC_MAX_RUNNING_COMPUTE_TASKS} metric in milliseconds. */ + public static final long DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL = 60_000L; /** Default client connector configuration. */ public static final ClientConnectorConfiguration DFLT_CLI_CFG = new ClientConnectorConfigurationEx(); @@ -132,7 +132,7 @@ public class ClientListenerProcessor extends GridProcessorAdapter { private ClientListenerMetrics metrics; /** */ - private MaxWithCurrentValueMetric maxCurActiveComputeTasks; + private MaxWithCurrentValueMetric maxRunningComputeTasksMetric; /** Executor service. */ private ExecutorService execSvc; @@ -370,23 +370,20 @@ private void registerClientMetrics(MetricRegistryImpl mreg) { ? ThinClientConfiguration.DFLT_MAX_ACTIVE_COMPUTE_TASKS_PER_CONNECTION : thinCfg.getMaxActiveComputeTasksPerConnection(); - // Registered before the server is started, since connections update it. - maxCurActiveComputeTasks = new MaxWithCurrentValueMetric( - metricName(mreg.name(), METRIC_MAX_CURRENT_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_MAX_CURRENT_ACTIVE_COMPUTE_TASKS_INTERVAL, + maxRunningComputeTasksMetric = new MaxWithCurrentValueMetric( + metricName(mreg.name(), METRIC_MAX_RUNNING_COMPUTE_TASKS), + "Maximum number of running user compute tasks per thin client connection over the configured " + + "time interval", + DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL, DFLT_SIZE, - maxActiveComputeTasks, - this::currentMaxActiveComputeTasks + maxActiveComputeTasks > 0 ? this::maxRunningComputeTasks : null ); - mreg.register(maxCurActiveComputeTasks); + mreg.register(maxRunningComputeTasksMetric); } - /** @return Maximum of active user compute tasks across thin client connections. */ - private long currentMaxActiveComputeTasks() { + /** @return Maximum of running user compute tasks across thin client connections. */ + private long maxRunningComputeTasks() { long max = 0; for (GridNioSession ses : srv.sessions()) { @@ -665,7 +662,7 @@ public ClientListenerMetrics metrics() { /** @param activeTasksCnt Number of active user compute tasks of a connection. */ public void updateMaxCurrentActiveComputeTasks(int activeTasksCnt) { - maxCurActiveComputeTasks.update(activeTasksCnt); + maxRunningComputeTasksMetric.update(activeTasksCnt); } /** @@ -909,26 +906,21 @@ public class ClientProcessorMXBeanImpl implements ClientProcessorMXBean { /** Maximum value over the time interval, but not less than the current value. */ private static class MaxWithCurrentValueMetric extends MaxValueMetric { - /** */ - private final long limit; - /** */ private final LongSupplier cur; /** */ - MaxWithCurrentValueMetric(String name, String desc, long timeInterval, int size, long limit, LongSupplier cur) { + MaxWithCurrentValueMetric(String name, String desc, long timeInterval, int size, @Nullable LongSupplier cur) { super(name, desc, timeInterval, size); - this.limit = limit; this.cur = cur; } /** {@inheritDoc} */ @Override public long value() { - long val = super.value(); - - // The current value can not exceed the limit, so there is no need to calculate it. - return val >= limit ? val : Math.max(val, cur.getAsLong()); + return cur == null ? + 0 : + Math.max(super.value(), cur.getAsLong()); } } } 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 510ba8537e38c..d21895a83e706 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 @@ -66,7 +66,7 @@ import static org.apache.ignite.internal.processors.odbc.ClientListenerMetrics.METRIC_REJECTED_TOTAL; import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.CLI_CONN_VIEW; import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.METRIC_ACTIVE; -import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.METRIC_MAX_CURRENT_ACTIVE_COMPUTE_TASKS; +import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.METRIC_MAX_RUNNING_COMPUTE_TASKS; import static org.apache.ignite.internal.util.GridTestClockTimer.DFLT_TIME_SUPPLIER; import static org.apache.ignite.internal.util.GridTestClockTimer.timeSupplier; import static org.apache.ignite.ssl.SslContextFactory.DFLT_STORE_TYPE; @@ -203,7 +203,7 @@ public void testClientListenerMetricsRejectGeneral() throws Exception { } /** - * Check that the active compute tasks metric reports the maximum among all connections and keeps it + * Check that the running compute tasks metric reports the maximum among all connections and keeps it * during the time interval. */ @Test @@ -446,7 +446,7 @@ private void checkMaxComputeTasks(long expMetric) throws Exception { /** */ private MaxValueMetric computeTasksMetric() { return grid(0).context().metric().registry(CLIENT_CONNECTOR_METRICS) - .findMetric(METRIC_MAX_CURRENT_ACTIVE_COMPUTE_TASKS); + .findMetric(METRIC_MAX_RUNNING_COMPUTE_TASKS); } /** Shifts the time returned by {@link U#currentTimeMillis()} relative to the system time. */ From cb4b661496c8ef07f4a79d3c3c92b14b21ccd7be Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Mon, 26 Oct 2026 10:13:06 +0300 Subject: [PATCH 07/12] WIP --- .../odbc/ClientListenerProcessor.java | 30 +++++++++---------- .../client/ClientConnectionContext.java | 4 +-- 2 files changed, 16 insertions(+), 18 deletions(-) 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 ab62147e5b260..87470dd06d4b5 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 @@ -132,7 +132,7 @@ public class ClientListenerProcessor extends GridProcessorAdapter { private ClientListenerMetrics metrics; /** */ - private MaxWithCurrentValueMetric maxRunningComputeTasksMetric; + private MaxValueMetric maxRunningComputeTasksMetric; /** Executor service. */ private ExecutorService execSvc; @@ -370,14 +370,14 @@ private void registerClientMetrics(MetricRegistryImpl mreg) { ? ThinClientConfiguration.DFLT_MAX_ACTIVE_COMPUTE_TASKS_PER_CONNECTION : thinCfg.getMaxActiveComputeTasksPerConnection(); - maxRunningComputeTasksMetric = new MaxWithCurrentValueMetric( - metricName(mreg.name(), METRIC_MAX_RUNNING_COMPUTE_TASKS), - "Maximum number of running user compute tasks per thin client connection over the configured " + - "time interval", - DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL, - DFLT_SIZE, - maxActiveComputeTasks > 0 ? this::maxRunningComputeTasks : null - ); + String name = metricName(mreg.name(), METRIC_MAX_RUNNING_COMPUTE_TASKS); + String desc = "Maximum number of running user compute tasks per thin client connection over the configured " + + "time interval"; + + maxRunningComputeTasksMetric = maxActiveComputeTasks > 0 + ? new MaxWithCurrentValueMetric(name, desc, DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL, DFLT_SIZE, + this::maxRunningComputeTasks) + : new MaxValueMetric(name, desc, DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL, DFLT_SIZE); mreg.register(maxRunningComputeTasksMetric); } @@ -660,9 +660,9 @@ public ClientListenerMetrics metrics() { return metrics; } - /** @param activeTasksCnt Number of active user compute tasks of a connection. */ - public void updateMaxCurrentActiveComputeTasks(int activeTasksCnt) { - maxRunningComputeTasksMetric.update(activeTasksCnt); + /** @param runningTasksCnt Number of running user compute tasks of a connection. */ + public void updateMaxRunningComputeTasks(int runningTasksCnt) { + maxRunningComputeTasksMetric.update(runningTasksCnt); } /** @@ -910,7 +910,7 @@ private static class MaxWithCurrentValueMetric extends MaxValueMetric { private final LongSupplier cur; /** */ - MaxWithCurrentValueMetric(String name, String desc, long timeInterval, int size, @Nullable LongSupplier cur) { + MaxWithCurrentValueMetric(String name, String desc, long timeInterval, int size, LongSupplier cur) { super(name, desc, timeInterval, size); this.cur = cur; @@ -918,9 +918,7 @@ private static class MaxWithCurrentValueMetric extends MaxValueMetric { /** {@inheritDoc} */ @Override public long value() { - return cur == null ? - 0 : - Math.max(super.value(), cur.getAsLong()); + return Math.max(super.value(), cur.getAsLong()); } } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java index 36b5b3703e165..15daf40f10286 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java @@ -355,7 +355,7 @@ public void incrementActiveTasksCount() { activeTasksCnt.decrementAndGet(); // A rejected task counts as the limit. - ctx.clientListener().updateMaxCurrentActiveComputeTasks(maxActiveComputeTasks); + ctx.clientListener().updateMaxRunningComputeTasks(maxActiveComputeTasks); throw new IgniteClientException(ClientStatus.TOO_MANY_COMPUTE_TASKS, "Active compute tasks per connection " + "limit (" + maxActiveComputeTasks + ") exceeded. To start a new task you need to wait for some of " + @@ -363,7 +363,7 @@ public void incrementActiveTasksCount() { "ThinClientConfiguration.MaxActiveComputeTasksPerConnection property."); } - ctx.clientListener().updateMaxCurrentActiveComputeTasks(cnt); + ctx.clientListener().updateMaxRunningComputeTasks(cnt); } /** From 38bac0754139f537ba585b936da72cf9c0c17c2f Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Mon, 26 Oct 2026 13:09:52 +0300 Subject: [PATCH 08/12] WIP --- .../odbc/ClientListenerProcessor.java | 39 ++++++++++--------- 1 file changed, 20 insertions(+), 19 deletions(-) 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 87470dd06d4b5..3000a3a8604eb 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,7 +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.LongUnaryOperator; import java.util.function.Predicate; import javax.cache.configuration.Factory; import javax.management.JMException; @@ -370,21 +370,22 @@ private void registerClientMetrics(MetricRegistryImpl mreg) { ? ThinClientConfiguration.DFLT_MAX_ACTIVE_COMPUTE_TASKS_PER_CONNECTION : thinCfg.getMaxActiveComputeTasksPerConnection(); - String name = metricName(mreg.name(), METRIC_MAX_RUNNING_COMPUTE_TASKS); - String desc = "Maximum number of running user compute tasks per thin client connection over the configured " + - "time interval"; - - maxRunningComputeTasksMetric = maxActiveComputeTasks > 0 - ? new MaxWithCurrentValueMetric(name, desc, DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL, DFLT_SIZE, - this::maxRunningComputeTasks) - : new MaxValueMetric(name, desc, DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL, DFLT_SIZE); + maxRunningComputeTasksMetric = new MaxWithValueUpdateMetric( + metricName(mreg.name(), METRIC_MAX_RUNNING_COMPUTE_TASKS), + "Maximum number of running user compute tasks per thin client connection over the configured time interval", + DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL, + DFLT_SIZE, + maxActiveComputeTasks > 0 ? this::maxRunningComputeTasks : l -> 0L); mreg.register(maxRunningComputeTasksMetric); } - /** @return Maximum of running user compute tasks across thin client connections. */ - private long maxRunningComputeTasks() { - long max = 0; + /** + * @param val Metric value. + * @return Maximum value derived from metric value and calculated one. + */ + private long maxRunningComputeTasks(long val) { + long max = val; for (GridNioSession ses : srv.sessions()) { ClientListenerConnectionContext connCtx = ses.meta(CONN_CTX_META_KEY); @@ -904,21 +905,21 @@ public class ClientProcessorMXBeanImpl implements ClientProcessorMXBean { } } - /** Maximum value over the time interval, but not less than the current value. */ - private static class MaxWithCurrentValueMetric extends MaxValueMetric { - /** */ - private final LongSupplier cur; + /** Maximum value over the time interval updated by the given operator on each read. */ + private static class MaxWithValueUpdateMetric extends MaxValueMetric { + /** Operator applied to the maximum value accumulated over the time interval. */ + private final LongUnaryOperator updater; /** */ - MaxWithCurrentValueMetric(String name, String desc, long timeInterval, int size, LongSupplier cur) { + MaxWithValueUpdateMetric(String name, String desc, long timeInterval, int size, LongUnaryOperator updater) { super(name, desc, timeInterval, size); - this.cur = cur; + this.updater = updater; } /** {@inheritDoc} */ @Override public long value() { - return Math.max(super.value(), cur.getAsLong()); + return updater.applyAsLong(super.value()); } } } From 0e02da1685eadc78d2bb4ef18153de8d7f1c58ed Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Mon, 26 Oct 2026 15:02:47 +0300 Subject: [PATCH 09/12] WIP --- .../internal/processors/odbc/ClientListenerProcessor.java | 2 +- .../ignite/spi/systemview/view/ClientConnectionView.java | 5 ++++- 2 files changed, 5 insertions(+), 2 deletions(-) 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 3000a3a8604eb..0b61a97f77bd8 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 @@ -907,7 +907,7 @@ public class ClientProcessorMXBeanImpl implements ClientProcessorMXBean { /** Maximum value over the time interval updated by the given operator on each read. */ private static class MaxWithValueUpdateMetric extends MaxValueMetric { - /** Operator applied to the maximum value accumulated over the time interval. */ + /** */ private final LongUnaryOperator updater; /** */ 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 8644de773d18d..e7c0a248b745f 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ClientConnectionView.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ClientConnectionView.java @@ -120,7 +120,10 @@ public String version() { return ctx == null ? null : ctx.attributes().get(IGNITE_DATA_CENTER_ID); } - /** @return Number of active user compute tasks of the thin client connection, {@code null} for other connection types. */ + /** + * @return Number of active user compute tasks of the thin client connection, + * {@code null} for other connection types. + */ @Order(7) @Nullable public Integer activeComputeTasks() { return ctx instanceof ClientConnectionContext ? ((ClientConnectionContext)ctx).activeTasksCount() : null; From 80ac4af9dc253bb2835ca334c2698d301cfd83c6 Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Mon, 26 Oct 2026 16:28:41 +0300 Subject: [PATCH 10/12] Fix test --- .../odbc/ClientListenerMetricsTest.java | 38 ++++++++++--------- 1 file changed, 21 insertions(+), 17 deletions(-) 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 d21895a83e706..4ca157f2fd05f 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 @@ -79,7 +79,7 @@ */ public class ClientListenerMetricsTest extends GridCommonAbstractTest { /** */ - private static final int MAX_TASKS = 3; + private static final int MAX_TASKS = 5; /** Metric time interval. Expiration is emulated by a time shift. */ private static final long LONG_INTERVAL = 10 * 60_000L; @@ -219,39 +219,43 @@ public void testComputeTasksMetrics() throws Exception { checkMaxComputeTasks(0); assertEquals(0, activeComputeTasksCount()); - // Tasks 0, 1. - futs.addAll(startTasks(client1, 2)); - checkMaxComputeTasks(2); - assertEquals(2, activeComputeTasksCount()); + int startCnt = 2; - // Run tasks 2, 3. Metric should not change. - futs.addAll(startTasks(client2, 2)); - checkMaxComputeTasks(2); - assertEquals(4, activeComputeTasksCount()); + // Run tasks on the first connection. + futs.addAll(startTasks(client1, startCnt)); + checkMaxComputeTasks(startCnt); + assertEquals(startCnt, activeComputeTasksCount()); - // Task 4. Metric should increase. - futs.addAll(startTasks(client2, 1)); + // The same number of tasks on the second one. Max metric should not change. + futs.addAll(startTasks(client2, startCnt)); + checkMaxComputeTasks(startCnt); + assertEquals(startCnt * 2, activeComputeTasksCount()); + + // Run tasks on the second connection up to the limit. + futs.addAll(startTasks(client2, MAX_TASKS - startCnt)); checkMaxComputeTasks(MAX_TASKS); - assertEquals(5, activeComputeTasksCount()); - // Metric should not exceed limit after a rejected task. + int allTasksCnt = MAX_TASKS + startCnt; + assertEquals(allTasksCnt, activeComputeTasksCount()); + + // Max metric should not exceed limit after a rejected task. assertComputeTaskRejected(client2); checkMaxComputeTasks(MAX_TASKS); - assertEquals(5, activeComputeTasksCount()); + assertEquals(allTasksCnt, activeComputeTasksCount()); // Finished tasks should not reset the maximum. - finishTasks(0, 1, 2, 3, 4); + finishTasks(IntStream.range(0, allTasksCnt).toArray()); checkMaxComputeTasks(MAX_TASKS); // "Wait" for expiration. shiftTime(LONG_INTERVAL); assertEquals(0, computeTasksMetric().value()); - // Task 5. + // Run one more task. futs.addAll(startTasks(client1, 1)); checkMaxComputeTasks(1); - finishTasks(5); + finishTasks(allTasksCnt); shiftTime(2 * LONG_INTERVAL); checkMaxComputeTasks(0); From 97aa6cafc5e7d2a7416a7309ca31211564775e37 Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Mon, 26 Oct 2026 17:47:01 +0300 Subject: [PATCH 11/12] Fix test --- .../odbc/ClientListenerMetricsTest.java | 59 +++++++++---------- 1 file changed, 27 insertions(+), 32 deletions(-) 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 4ca157f2fd05f..bec202842560c 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java @@ -30,6 +30,7 @@ import org.apache.ignite.client.ClientException; import org.apache.ignite.client.Config; import org.apache.ignite.client.IgniteClient; +import org.apache.ignite.client.IgniteClientFuture; import org.apache.ignite.client.SslMode; import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.cluster.ClusterState; @@ -44,7 +45,6 @@ import org.apache.ignite.configuration.IgniteConfiguration; import org.apache.ignite.configuration.ThinClientConfiguration; import org.apache.ignite.internal.IgniteEx; -import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.IgniteInterruptedCheckedException; import org.apache.ignite.internal.processors.metric.impl.MaxValueMetric; import org.apache.ignite.internal.processors.metric.impl.MetricUtils; @@ -71,7 +71,6 @@ import static org.apache.ignite.internal.util.GridTestClockTimer.timeSupplier; import static org.apache.ignite.ssl.SslContextFactory.DFLT_STORE_TYPE; import static org.apache.ignite.testframework.GridTestUtils.assertThrowsWithCause; -import static org.apache.ignite.testframework.GridTestUtils.runAsync; import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; /** @@ -210,7 +209,7 @@ public void testClientListenerMetricsRejectGeneral() throws Exception { public void testComputeTasksMetrics() throws Exception { startGridWithCompute(); - List> futs = new ArrayList<>(); + List> futs = new ArrayList<>(); try ( IgniteClient client1 = Ignition.startClient(getClientConfiguration()); @@ -226,7 +225,7 @@ public void testComputeTasksMetrics() throws Exception { checkMaxComputeTasks(startCnt); assertEquals(startCnt, activeComputeTasksCount()); - // The same number of tasks on the second one. Max metric should not change. + // The same number of tasks on the second connection. Max metric should not change. futs.addAll(startTasks(client2, startCnt)); checkMaxComputeTasks(startCnt); assertEquals(startCnt * 2, activeComputeTasksCount()); @@ -260,7 +259,7 @@ public void testComputeTasksMetrics() throws Exception { shiftTime(2 * LONG_INTERVAL); checkMaxComputeTasks(0); - for (IgniteInternalFuture fut : futs) + for (IgniteClientFuture fut : futs) fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); } finally { @@ -274,7 +273,7 @@ public void testComputeTasksMetricOnLongRunningTasks() throws Exception { startGridWithCompute(); try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { - List> futs = startTasks(client, MAX_TASKS); + List> futs = startTasks(client, MAX_TASKS); checkMaxComputeTasks(MAX_TASKS); @@ -286,7 +285,7 @@ public void testComputeTasksMetricOnLongRunningTasks() throws Exception { assertEquals(0, computeTasksMetric().value()); - for (IgniteInternalFuture fut : futs) + for (IgniteClientFuture fut : futs) fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); } finally { @@ -294,16 +293,13 @@ public void testComputeTasksMetricOnLongRunningTasks() throws Exception { } } - /** - * Check that a task rejected by the limit updates the metric with the limit, even if the active tasks were started - * before the time interval. - */ + /** */ @Test public void testComputeTasksMetricOnRejectedTask() throws Exception { startGridWithCompute(); try (IgniteClient client = Ignition.startClient(getClientConfiguration())) { - List> futs = startTasks(client, MAX_TASKS); + List> futs = startTasks(client, MAX_TASKS); checkMaxComputeTasks(MAX_TASKS); @@ -319,7 +315,7 @@ public void testComputeTasksMetricOnRejectedTask() throws Exception { finishTasks(IntStream.range(1, MAX_TASKS).toArray()); - for (IgniteInternalFuture fut : futs) + for (IgniteClientFuture fut : futs) fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); } finally { @@ -327,17 +323,17 @@ public void testComputeTasksMetricOnRejectedTask() throws Exception { } } - /** Check that closing the connection with unfinished tasks releases its tasks. */ + /** */ @Test public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { startGridWithCompute(); try (IgniteClient client1 = Ignition.startClient(getClientConfiguration())) { - List> futs1 = startTasks(client1, 1); + List> futs1 = startTasks(client1, 1); assertEquals(1, activeComputeTasksCount()); - List> futs2; + List> futs2; try (IgniteClient client2 = Ignition.startClient(getClientConfiguration())) { futs2 = startTasks(client2, 2); @@ -347,26 +343,24 @@ public void testComputeTasksMetricsAfterClientDisconnect() throws Exception { assertEquals(3, activeComputeTasksCount()); } - // Tasks of the closed client should be released on disconnect + // Tasks of the closed client should not be counted in system views. assertTrue(waitForCondition(() -> activeComputeTasksCount() == 1, getTestTimeout())); + // Finish tasks of the closed client. TestComputeTask.finish(1); TestComputeTask.finish(2); - // The metric keeps the maximum during the time interval. - checkMaxComputeTasks(2); - - // Unfinished tasks fail on the client side. The error depends on whether the client received the execute - // response before the close. - for (IgniteInternalFuture fut : futs2) + // Unfinished tasks of client2 fail. + for (IgniteClientFuture fut : futs2) assertThrowsWithCause(() -> fut.get(getTestTimeout(), TimeUnit.MILLISECONDS), ClientException.class); - // Last task of 'client1'. + // Finish task of alive client. finishTasks(0); + + // The metric keeps the maximum during the time interval. checkMaxComputeTasks(2); - // Wait for the result before the client is closed. - for (IgniteInternalFuture fut : futs1) + for (IgniteClientFuture fut : futs1) fut.get(getTestTimeout(), TimeUnit.MILLISECONDS); } } @@ -412,15 +406,16 @@ private void assertComputeTaskRejected(IgniteClient client) { ); } - /** Starts tasks and waits for them to be started on the node. */ - private List> startTasks(IgniteClient client, int cnt) throws Exception { + /** */ + private List> startTasks(IgniteClient client, int cnt) throws Exception { int expTasks = TestComputeTask.TASKS.size() + cnt; - List> futs = IntStream.range(0, cnt) - .mapToObj(i -> runAsync(() -> client.compute().execute(TestComputeTask.class.getName(), null))) - .toList(); + List> futs = new ArrayList<>(); + + for (int i = 0; i < cnt; i++) + futs.add(client.compute().executeAsync2(TestComputeTask.class.getName(), null)); - // A task is registered in 'map()' after the counter is incremented. Wait for the tasks to keep their order. + // Waits for tasks to be started on the nodes. assertTrue(waitForCondition(() -> TestComputeTask.TASKS.size() == expTasks, getTestTimeout())); return futs; From 310a8ef8e3c13e27e03a38c6eea7064b0058fe02 Mon Sep 17 00:00:00 2001 From: Ilya Shishkov Date: Mon, 26 Oct 2026 18:38:36 +0300 Subject: [PATCH 12/12] Fix comments --- .../odbc/ClientListenerProcessor.java | 27 +++++++++---------- 1 file changed, 12 insertions(+), 15 deletions(-) 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 0b61a97f77bd8..1186bc3427c2b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java @@ -29,7 +29,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Function; -import java.util.function.LongUnaryOperator; +import java.util.function.LongSupplier; import java.util.function.Predicate; import javax.cache.configuration.Factory; import javax.management.JMException; @@ -370,22 +370,19 @@ private void registerClientMetrics(MetricRegistryImpl mreg) { ? ThinClientConfiguration.DFLT_MAX_ACTIVE_COMPUTE_TASKS_PER_CONNECTION : thinCfg.getMaxActiveComputeTasksPerConnection(); - maxRunningComputeTasksMetric = new MaxWithValueUpdateMetric( + maxRunningComputeTasksMetric = new MaxWithCurrentValueMetric( metricName(mreg.name(), METRIC_MAX_RUNNING_COMPUTE_TASKS), "Maximum number of running user compute tasks per thin client connection over the configured time interval", DFLT_MAX_RUNNING_COMPUTE_TASKS_INTERVAL, DFLT_SIZE, - maxActiveComputeTasks > 0 ? this::maxRunningComputeTasks : l -> 0L); + maxActiveComputeTasks > 0 ? this::maxRunningComputeTasks : () -> 0L); mreg.register(maxRunningComputeTasksMetric); } - /** - * @param val Metric value. - * @return Maximum value derived from metric value and calculated one. - */ - private long maxRunningComputeTasks(long val) { - long max = val; + /** @return Maximum number of currently running user compute tasks among all thin client connections. */ + private long maxRunningComputeTasks() { + long max = 0; for (GridNioSession ses : srv.sessions()) { ClientListenerConnectionContext connCtx = ses.meta(CONN_CTX_META_KEY); @@ -905,21 +902,21 @@ public class ClientProcessorMXBeanImpl implements ClientProcessorMXBean { } } - /** Maximum value over the time interval updated by the given operator on each read. */ - private static class MaxWithValueUpdateMetric extends MaxValueMetric { + /** Maximum value over the time interval, or the current value if it is greater. */ + private static class MaxWithCurrentValueMetric extends MaxValueMetric { /** */ - private final LongUnaryOperator updater; + private final LongSupplier cur; /** */ - MaxWithValueUpdateMetric(String name, String desc, long timeInterval, int size, LongUnaryOperator updater) { + MaxWithCurrentValueMetric(String name, String desc, long timeInterval, int size, LongSupplier cur) { super(name, desc, timeInterval, size); - this.updater = updater; + this.cur = cur; } /** {@inheritDoc} */ @Override public long value() { - return updater.applyAsLong(super.value()); + return Math.max(super.value(), cur.getAsLong()); } } }