IGNITE-29107 Add ClientConnector metric for current peak of running compute tasks

docs/_docs/monitoring-metrics/new-metrics-system.adoc

@@ -419,7 +419,8 @@ The metrics belong to the `snapshot` register, see link:monitoring-metrics/new-m
 Metrics related to JDBC/ODBC or thin client connections.
 
 * `ActiveSessionsCount`, `thin.ActiveSessions`, `jdbc.ActiveSessions`, `odbc.ActiveSessions` - The number of active connections, total and per client type (`client.connector` register, see link:monitoring-metrics/new-metrics#ignite-thin-client-connector[Ignite Thin Client Connector]).
-* The link:monitoring-metrics/system-views#client_connections[CLIENT_CONNECTIONS] system view lists every connection with its user and addresses.
+* `PeakActiveComputeTasksPerConnection` - The maximum number of active user compute tasks per thin client connection over the configured time interval, but not less than the current value. The upper limit is the `ThinClientConfiguration.maxActiveComputeTasksPerConnection` configuration parameter. A task rejected by the limit counts as the limit. The interval is 90 seconds by default and can be changed with `control.sh --metric --configure-max-value` (see link:tools/control-script#metric-configure-command[Metric configure command]). Set it longer than the polling period of your monitoring system, so peaks between polls are not missed (`client.connector` register, see link:monitoring-metrics/new-metrics#ignite-thin-client-connector[Ignite Thin Client Connector]).
+* The link:monitoring-metrics/system-views#client_connections[CLIENT_CONNECTIONS] system view lists every connection with its user, addresses and the current number of active user compute tasks.
 * `Connections` - Attribute of the `ClientProcessorMXBean` MBean, not available in the metric registers. A list of strings, each string containing information about a connection:
 
 ....

docs/_docs/monitoring-metrics/new-metrics.adoc

@@ -466,6 +466,7 @@ Register name: `client.connector`
 |AffinityKeyRequestsMisses | long | The number of affinity-aware cache key requests that were sent not to the primary node.
 |AffinityQueryRequestsHits | long | The number of affinity-aware query requests that were sent to the primary node.
 |AffinityQueryRequestsMisses | long | The number of affinity-aware query requests that were sent not to the primary node.
+|PeakActiveComputeTasksPerConnection | long | Maximum number of active user compute tasks per thin client connection over the configured time interval (90 seconds by default), but not less than the current value. The upper limit is the `ThinClientConfiguration.maxActiveComputeTasksPerConnection` configuration parameter. A task rejected by the limit counts as the limit.
 |RejectedSessionsAuthenticationFailed | integer | TCP sessions count that were rejected due to failed authentication.
 |RejectedSessionsTimeout | integer | TCP sessions count that were rejected due to handshake timeout.
 |RejectedSessionsTotal | integer | Total number of rejected TCP connections.

docs/_docs/monitoring-metrics/system-views.adoc

@@ -443,6 +443,7 @@ This view exposes information about currently opened client connections: JDBC, O
 |USER | string |  User name
 |VERSION | string |  Protocol version
 |DATA_CENTER_ID | string | Effective data center ID reported by the client
+|ACTIVE_COMPUTE_TASKS | int | Number of active user compute tasks of a thin client connection, not greater than the `ThinClientConfiguration.maxActiveComputeTasksPerConnection` configuration parameter. `null` for JDBC and ODBC connections
 |===
 
 == CLIENT_CONNECTION_ATTRIBUTES

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
 ----
 --
 

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",

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<List<String>> sqlViewColumnsView = systemView(ignite0, SQL_VIEW_COLS_VIEW).stream()

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<ClientConnectionAttributeView> 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());
+        }
+    }
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java

@@ -349,14 +349,21 @@ public void incrementActiveTasksCount() {
                     "To enable it set up the ThinClientConfiguration.MaxActiveComputeTasksPerConnection property.");
         }
 
-        if (activeTasksCnt.incrementAndGet() > maxActiveComputeTasks) {
+        int cnt = activeTasksCnt.incrementAndGet();
+
+        if (cnt > maxActiveComputeTasks) {
             activeTasksCnt.decrementAndGet();
 
+            // A rejected task counts as the limit.
+            ctx.clientListener().updatePeakActiveComputeTasks(maxActiveComputeTasks);
+
             throw new IgniteClientException(ClientStatus.TOO_MANY_COMPUTE_TASKS, "Active compute tasks per connection " +
                 "limit (" + maxActiveComputeTasks + ") exceeded. To start a new task you need to wait for some of " +
                 "currently active tasks complete. To change the limit set up the " +
                 "ThinClientConfiguration.MaxActiveComputeTasksPerConnection property.");
         }
+
+        ctx.clientListener().updatePeakActiveComputeTasks(cnt);
     }
 
     /**
@@ -367,4 +374,10 @@ public void decrementActiveTasksCount() {
 
         assert cnt >= 0 : "Unexpected active tasks count: " + cnt;
     }
+
+    /** @return Number of active user compute tasks for this connection capped by the limit. */
+    public int activeTasksCount() {
+        // The counter briefly exceeds the limit while a task is being rejected, so the value is capped by the limit.
+        return Math.min(activeTasksCnt.get(), maxActiveComputeTasks);
+    }
 }

modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ClientConnectionView.java

@@ -18,6 +18,7 @@
 package org.apache.ignite.spi.systemview.view;
 
 import java.net.InetSocketAddress;
+import org.apache.ignite.configuration.ThinClientConfiguration;
 import org.apache.ignite.internal.processors.odbc.ClientListenerConnectionContext;
 import org.apache.ignite.internal.processors.odbc.ClientListenerProtocolVersion;
 import org.apache.ignite.internal.processors.odbc.ClientListenerRequestHandler;
@@ -119,4 +120,14 @@ public String version() {
     @Nullable public String dataCenterId() {
         return ctx == null ? null : ctx.attributes().get(IGNITE_DATA_CENTER_ID);
     }
+
+    /**
+     * @return Number of active user compute tasks of the thin client connection, not greater than
+     * {@link ThinClientConfiguration#getMaxActiveComputeTasksPerConnection()}; {@code null} for other connection
+     * types.
+     */
+    @Order(7)
+    @Nullable public Integer activeComputeTasks() {
+        return ctx instanceof ClientConnectionContext ? ((ClientConnectionContext)ctx).activeTasksCount() : null;
+    }
 }