alex-plekhanov commented on code in PR #13661:
URL: https://github.com/apache/ignite/pull/13661#discussion_r4352191192
##########
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
Review Comment:
`not greater than ...` is redundant
##########
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]).
Review Comment:
See comment about metric description
##########
docs/_docs/monitoring-metrics/new-metrics.adoc:
##########
@@ -469,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.
Review Comment:
Why `peak`, not `max`? Other metrics with the same functionality (for
example `maxOutboundMessagesQueueSize`) use `max` prefix. Description also
contains `Maximum number`.
Why time interval is 90 second by default? Again, other metrics use 1 minute
time interval.
`but not less than the current value` - it's implementation detail, for user
it's confusing.
`The upper limit ...` is also confusing in metric description, and also
implementation detail.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java:
##########
@@ -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());
Review Comment:
`cur` can be more than max only when metric is not updated more than time
interval, in this case value is 0 and only in this case we need to calculate
`cur`. `limit` filed is not required
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java:
##########
@@ -160,6 +204,285 @@ public void testClientListenerMetricsRejectGeneral()
throws Exception {
}
}
+ /**
+ * Check that the active compute tasks metric reports the peak of the
busiest connection (not the sum), keeps it
+ * during the time interval and drops it after the time interval.
+ */
+ @Test
+ public void testComputeTasksMetrics() throws Exception {
+ startGridWithCompute();
+
+ List<IgniteInternalFuture<Object>> futs = new ArrayList<>();
+
+ try (
+ IgniteClient client1 =
Ignition.startClient(getClientConfiguration());
+ IgniteClient client2 =
Ignition.startClient(getClientConfiguration())
+ ) {
+ checkPeakComputeTasks(0);
+ assertEquals(0, activeComputeTasksCount());
+
+ // Tasks 0, 1.
+ futs.addAll(startTasks(client1, 2));
+ checkPeakComputeTasks(2);
+ assertEquals(2, activeComputeTasksCount());
+
+ // Run tasks 2, 3. Metric should not change.
+ futs.addAll(startTasks(client2, 2));
+ checkPeakComputeTasks(2);
+ assertEquals(4, activeComputeTasksCount());
+
+ // Task 4. Metric should increase.
+ futs.addAll(startTasks(client2, 1));
+ checkPeakComputeTasks(MAX_TASKS);
Review Comment:
You should start `MAX_TASKS - 2` tasks (with corresponding changes in code
below) or check for `3` tasks, but not `MAX_TASKS` (with corresponding changes
in code below)
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java:
##########
@@ -160,6 +204,285 @@ public void testClientListenerMetricsRejectGeneral()
throws Exception {
}
}
+ /**
+ * Check that the active compute tasks metric reports the peak of the
busiest connection (not the sum), keeps it
+ * during the time interval and drops it after the time interval.
+ */
+ @Test
+ public void testComputeTasksMetrics() throws Exception {
+ startGridWithCompute();
+
+ List<IgniteInternalFuture<Object>> futs = new ArrayList<>();
+
+ try (
+ IgniteClient client1 =
Ignition.startClient(getClientConfiguration());
+ IgniteClient client2 =
Ignition.startClient(getClientConfiguration())
+ ) {
+ checkPeakComputeTasks(0);
+ assertEquals(0, activeComputeTasksCount());
+
+ // Tasks 0, 1.
+ futs.addAll(startTasks(client1, 2));
+ checkPeakComputeTasks(2);
+ assertEquals(2, activeComputeTasksCount());
+
+ // Run tasks 2, 3. Metric should not change.
+ futs.addAll(startTasks(client2, 2));
+ checkPeakComputeTasks(2);
+ assertEquals(4, activeComputeTasksCount());
+
+ // Task 4. Metric should increase.
+ futs.addAll(startTasks(client2, 1));
+ checkPeakComputeTasks(MAX_TASKS);
+ assertEquals(5, activeComputeTasksCount());
+
+ assertComputeTaskRejected(client2);
+
+ // A rejected task counts as the
'MaxActiveComputeTasksPerConnection' limit.
+ checkPeakComputeTasks(MAX_TASKS);
+ assertEquals(5, activeComputeTasksCount());
+
+ // Finished tasks are released by the connections, but the metric
keeps the peak during the time interval.
+ finishTasks(2);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ finishTasks(0, 1);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ finishTasks(3, 4);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ // Clear the peak, as if it expired. No tasks are active, so the
metric reports zero.
+ computeTasksMetric().reset();
+ assertEquals(0, computeTasksMetric().value());
+
+ // Shorten the time interval to check the peak expiration without
a long wait.
+ computeTasksMetric().reset(SHORT_INTERVAL);
+
+ // Task 5.
+ futs.addAll(startTasks(client1, 1));
+ checkPeakComputeTasks(1);
+
+ // The peak expires after the time interval.
+ finishTasks(5);
+ checkPeakComputeTasks(0);
+
+ // Wait for all results before clients are closed.
+ for (IgniteInternalFuture<Object> fut : futs)
+ fut.get(getTestTimeout(), TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /**
+ * Check that the metric is not less than the current value, so it reports
long-running tasks after their peak has
+ * expired.
+ */
+ @Test
+ public void testComputeTasksMetricOnLongRunningTasks() throws Exception {
+ startGridWithCompute();
+
+ try (IgniteClient client =
Ignition.startClient(getClientConfiguration())) {
+ List<IgniteInternalFuture<Object>> 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<Object> fut : futs)
+ fut.get(getTestTimeout(), TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /**
+ * Check that a task rejected by the limit updates the metric with the
limit, even if the active tasks were started
+ * before the time interval.
+ */
+ @Test
+ public void testComputeTasksMetricOnRejectedTask() throws Exception {
+ startGridWithCompute();
+
+ try (IgniteClient client =
Ignition.startClient(getClientConfiguration())) {
+ List<IgniteInternalFuture<Object>> futs = startTasks(client,
MAX_TASKS);
+
+ checkPeakComputeTasks(MAX_TASKS);
+
+ // Clear the peak, as if it expired while the tasks are active.
+ computeTasksMetric().reset();
+
+ assertComputeTaskRejected(client);
+
+ // The metric keeps the limit from the rejected task, though the
current value is less.
+ finishTasks(0);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ finishTasks(IntStream.range(1, MAX_TASKS).toArray());
+
+ // Wait for all results before the client is closed.
+ for (IgniteInternalFuture<Object> fut : futs)
+ fut.get(getTestTimeout(), TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /** Check that closing the connection with unfinished tasks releases its
tasks. */
+ @Test
+ public void testComputeTasksMetricsAfterClientDisconnect() throws
Exception {
+ startGridWithCompute();
+
+ try (IgniteClient client1 =
Ignition.startClient(getClientConfiguration())) {
+ List<IgniteInternalFuture<Object>> futs1 = startTasks(client1, 1);
+
+ List<IgniteInternalFuture<Object>> futs2;
+
+ ClientConnectionView connView2;
+
+ try (IgniteClient client2 =
Ignition.startClient(getClientConfiguration())) {
+ futs2 = startTasks(client2, 2);
+
+ checkPeakComputeTasks(2);
+
+ connView2 = thinConnections().stream().filter(v ->
v.activeComputeTasks() == 2)
+ .findFirst().orElseThrow();
+ }
+
+ // Tasks of the closed connection are released on disconnect.
+ assertTrue(waitForCondition(() -> connView2.activeComputeTasks()
== 0, getTestTimeout()));
+
+ // Cancellation does not interrupt 'reduce()', so unblock the
tasks of 'client2'.
+ TestComputeTask.finish(1);
+ TestComputeTask.finish(2);
+
+ // The metric keeps the peak during the time interval.
+ checkPeakComputeTasks(2);
+
+ // Unfinished tasks fail on the client side. The error depends on
whether the client received the execute
+ // response before the close.
+ for (IgniteInternalFuture<Object> fut : futs2)
+ assertThrowsWithCause(() -> fut.get(getTestTimeout(),
TimeUnit.MILLISECONDS), ClientException.class);
+
+ // Last task of 'client1'.
+ finishTasks(0);
+ checkPeakComputeTasks(2);
+
+ // Wait for the result before the client is closed.
+ for (IgniteInternalFuture<Object> fut : futs1)
+ fut.get(getTestTimeout(), TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /** Check the active compute tasks metric when compute is disabled for
thin clients. */
+ @Test
+ public void testComputeTasksMetricsComputeDisabled() throws Exception {
+ startGrid(0);
+
+ try (IgniteClient client =
Ignition.startClient(getClientConfiguration())) {
+ checkPeakComputeTasks(0);
+
+ GridTestUtils.assertThrowsAnyCause(
+ log,
+ () ->
client.compute().execute(TestComputeTask.class.getName(), null),
+ ClientException.class,
+ "Compute grid functionality is disabled for thin clients"
+ );
+
+ checkPeakComputeTasks(0);
+ }
+ }
+
+ /**
+ * Starts a node with compute enabled for thin clients and sets the metric
time interval to {@link #LONG_INTERVAL}.
+ */
+ private void startGridWithCompute() throws Exception {
+ startGrid(getConfiguration(getTestIgniteInstanceName(0))
+ .setClientConnectorConfiguration(new ClientConnectorConfiguration()
+ .setThinClientConfiguration(new ThinClientConfiguration()
+ .setMaxActiveComputeTasksPerConnection(MAX_TASKS))));
+
+ computeTasksMetric().reset(LONG_INTERVAL);
+ }
+
+ /**
+ * Checks that a task is rejected because of the limit.
+ *
+ * @param client Ignite client.
+ */
+ private void assertComputeTaskRejected(IgniteClient client) {
+ GridTestUtils.assertThrowsAnyCause(
+ log,
+ () -> client.compute().execute(TestComputeTask.class.getName(),
null),
+ ClientException.class,
+ "Active compute tasks per connection limit (" + MAX_TASKS + ")
exceeded"
+ );
+ }
+
+ /**
+ * Starts tasks via client and waits for them to be started on the node.
+ *
+ * @param client Ignite client.
+ * @param cnt Tasks count.
+ * @return Task futures.
+ */
+ private List<IgniteInternalFuture<Object>> startTasks(IgniteClient client,
int cnt) throws Exception {
+ int expTasks = TestComputeTask.TASKS.size() + cnt;
+
+ List<IgniteInternalFuture<Object>> futs = IntStream.range(0, cnt)
+ .mapToObj(i -> runAsync(() ->
client.compute().execute(TestComputeTask.class.getName(), null)))
+ .toList();
+
+ // A task instance is created after the counter is incremented. Wait
for instances to keep the task order.
+ assertTrue(waitForCondition(() -> TestComputeTask.TASKS.size() ==
expTasks, getTestTimeout()));
+
+ return futs;
+ }
+
+ /**
+ * Finishes tasks and waits for them to be released by the connections.
+ *
+ * @param idxs Indexes of the tasks.
+ */
+ private void finishTasks(int... idxs) throws Exception {
+ int expTasks = activeComputeTasksCount() - idxs.length;
+
+ for (int idx : idxs)
+ TestComputeTask.finish(idx);
+
+ // Counter is decremented after the result is sent to the client.
+ assertTrue(waitForCondition(() -> activeComputeTasksCount() ==
expTasks, getTestTimeout()));
+ }
+
+ /** @return Total active compute tasks of the thin client connections from
the system view. */
+ private int activeComputeTasksCount() {
+ return
thinConnections().stream().mapToInt(ClientConnectionView::activeComputeTasks).sum();
+ }
+
+ /**
+ * Waits for the expected value of the active compute tasks metric.
+ *
+ * @param expMetric Expected value of the metric.
+ */
+ private void checkPeakComputeTasks(long expMetric) throws Exception {
+ assertTrue(waitForCondition(() -> computeTasksMetric().value() ==
expMetric, getTestTimeout()));
+ }
+
+ /** @return Active compute tasks metric. */
+ private MaxValueMetric computeTasksMetric() {
+ return
grid(0).context().metric().registry(CLIENT_CONNECTOR_METRICS).findMetric(METRIC_PEAK_ACTIVE_COMPUTE_TASKS);
+ }
+
+ /** @return Thin client connections from the system view. */
+ private List<ClientConnectionView> thinConnections() {
+ SystemView<ClientConnectionView> view =
grid(0).context().systemView().view(CLI_CONN_VIEW);
+
+ return StreamSupport.stream(view.spliterator(), false).filter(v ->
"THIN".equals(v.type())).toList();
Review Comment:
Simplify a little `U.arrayList(view.iterator(), view.size(), v ->
"THIN".equals(v.type()));` ?
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java:
##########
@@ -179,7 +502,7 @@ private static ClientConfiguration getClientConfiguration()
{
*/
private void waitForMetricValue(MetricRegistry mreg, String metric, long
value, long timeout)
throws IgniteInterruptedCheckedException {
- GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ waitForCondition(new GridAbsPredicate() {
Review Comment:
`waitForCondition(() -> mreg.<IntMetric>findMetric(metric).value() == value,
timeout);`?
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java:
##########
@@ -160,6 +204,285 @@ public void testClientListenerMetricsRejectGeneral()
throws Exception {
}
}
+ /**
+ * Check that the active compute tasks metric reports the peak of the
busiest connection (not the sum), keeps it
+ * during the time interval and drops it after the time interval.
+ */
+ @Test
+ public void testComputeTasksMetrics() throws Exception {
+ startGridWithCompute();
+
+ List<IgniteInternalFuture<Object>> futs = new ArrayList<>();
+
+ try (
+ IgniteClient client1 =
Ignition.startClient(getClientConfiguration());
+ IgniteClient client2 =
Ignition.startClient(getClientConfiguration())
+ ) {
+ checkPeakComputeTasks(0);
+ assertEquals(0, activeComputeTasksCount());
+
+ // Tasks 0, 1.
+ futs.addAll(startTasks(client1, 2));
+ checkPeakComputeTasks(2);
+ assertEquals(2, activeComputeTasksCount());
+
+ // Run tasks 2, 3. Metric should not change.
+ futs.addAll(startTasks(client2, 2));
+ checkPeakComputeTasks(2);
+ assertEquals(4, activeComputeTasksCount());
+
+ // Task 4. Metric should increase.
+ futs.addAll(startTasks(client2, 1));
+ checkPeakComputeTasks(MAX_TASKS);
+ assertEquals(5, activeComputeTasksCount());
+
+ assertComputeTaskRejected(client2);
+
+ // A rejected task counts as the
'MaxActiveComputeTasksPerConnection' limit.
+ checkPeakComputeTasks(MAX_TASKS);
+ assertEquals(5, activeComputeTasksCount());
+
+ // Finished tasks are released by the connections, but the metric
keeps the peak during the time interval.
+ finishTasks(2);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ finishTasks(0, 1);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ finishTasks(3, 4);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ // Clear the peak, as if it expired. No tasks are active, so the
metric reports zero.
+ computeTasksMetric().reset();
+ assertEquals(0, computeTasksMetric().value());
+
+ // Shorten the time interval to check the peak expiration without
a long wait.
+ computeTasksMetric().reset(SHORT_INTERVAL);
+
+ // Task 5.
+ futs.addAll(startTasks(client1, 1));
+ checkPeakComputeTasks(1);
+
+ // The peak expires after the time interval.
+ finishTasks(5);
+ checkPeakComputeTasks(0);
+
+ // Wait for all results before clients are closed.
+ for (IgniteInternalFuture<Object> fut : futs)
+ fut.get(getTestTimeout(), TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /**
+ * Check that the metric is not less than the current value, so it reports
long-running tasks after their peak has
+ * expired.
+ */
+ @Test
+ public void testComputeTasksMetricOnLongRunningTasks() throws Exception {
+ startGridWithCompute();
+
+ try (IgniteClient client =
Ignition.startClient(getClientConfiguration())) {
+ List<IgniteInternalFuture<Object>> 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<Object> fut : futs)
+ fut.get(getTestTimeout(), TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /**
+ * Check that a task rejected by the limit updates the metric with the
limit, even if the active tasks were started
+ * before the time interval.
+ */
+ @Test
+ public void testComputeTasksMetricOnRejectedTask() throws Exception {
+ startGridWithCompute();
+
+ try (IgniteClient client =
Ignition.startClient(getClientConfiguration())) {
+ List<IgniteInternalFuture<Object>> futs = startTasks(client,
MAX_TASKS);
+
+ checkPeakComputeTasks(MAX_TASKS);
+
+ // Clear the peak, as if it expired while the tasks are active.
+ computeTasksMetric().reset();
+
+ assertComputeTaskRejected(client);
+
+ // The metric keeps the limit from the rejected task, though the
current value is less.
+ finishTasks(0);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ finishTasks(IntStream.range(1, MAX_TASKS).toArray());
+
+ // Wait for all results before the client is closed.
+ for (IgniteInternalFuture<Object> fut : futs)
+ fut.get(getTestTimeout(), TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /** Check that closing the connection with unfinished tasks releases its
tasks. */
+ @Test
+ public void testComputeTasksMetricsAfterClientDisconnect() throws
Exception {
+ startGridWithCompute();
+
+ try (IgniteClient client1 =
Ignition.startClient(getClientConfiguration())) {
+ List<IgniteInternalFuture<Object>> futs1 = startTasks(client1, 1);
+
+ List<IgniteInternalFuture<Object>> futs2;
+
+ ClientConnectionView connView2;
+
+ try (IgniteClient client2 =
Ignition.startClient(getClientConfiguration())) {
+ futs2 = startTasks(client2, 2);
+
+ checkPeakComputeTasks(2);
+
+ connView2 = thinConnections().stream().filter(v ->
v.activeComputeTasks() == 2)
+ .findFirst().orElseThrow();
+ }
+
+ // Tasks of the closed connection are released on disconnect.
+ assertTrue(waitForCondition(() -> connView2.activeComputeTasks()
== 0, getTestTimeout()));
Review Comment:
It's dangerous to access some stale view row, let's recheck the view again
instead.
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java:
##########
@@ -160,6 +204,285 @@ public void testClientListenerMetricsRejectGeneral()
throws Exception {
}
}
+ /**
+ * Check that the active compute tasks metric reports the peak of the
busiest connection (not the sum), keeps it
+ * during the time interval and drops it after the time interval.
+ */
+ @Test
+ public void testComputeTasksMetrics() throws Exception {
+ startGridWithCompute();
+
+ List<IgniteInternalFuture<Object>> futs = new ArrayList<>();
+
+ try (
+ IgniteClient client1 =
Ignition.startClient(getClientConfiguration());
+ IgniteClient client2 =
Ignition.startClient(getClientConfiguration())
+ ) {
+ checkPeakComputeTasks(0);
+ assertEquals(0, activeComputeTasksCount());
+
+ // Tasks 0, 1.
+ futs.addAll(startTasks(client1, 2));
+ checkPeakComputeTasks(2);
+ assertEquals(2, activeComputeTasksCount());
+
+ // Run tasks 2, 3. Metric should not change.
+ futs.addAll(startTasks(client2, 2));
+ checkPeakComputeTasks(2);
+ assertEquals(4, activeComputeTasksCount());
+
+ // Task 4. Metric should increase.
+ futs.addAll(startTasks(client2, 1));
+ checkPeakComputeTasks(MAX_TASKS);
+ assertEquals(5, activeComputeTasksCount());
+
+ assertComputeTaskRejected(client2);
+
+ // A rejected task counts as the
'MaxActiveComputeTasksPerConnection' limit.
+ checkPeakComputeTasks(MAX_TASKS);
+ assertEquals(5, activeComputeTasksCount());
+
+ // Finished tasks are released by the connections, but the metric
keeps the peak during the time interval.
+ finishTasks(2);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ finishTasks(0, 1);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ finishTasks(3, 4);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ // Clear the peak, as if it expired. No tasks are active, so the
metric reports zero.
+ computeTasksMetric().reset();
+ assertEquals(0, computeTasksMetric().value());
+
+ // Shorten the time interval to check the peak expiration without
a long wait.
+ computeTasksMetric().reset(SHORT_INTERVAL);
+
+ // Task 5.
+ futs.addAll(startTasks(client1, 1));
+ checkPeakComputeTasks(1);
+
+ // The peak expires after the time interval.
+ finishTasks(5);
+ checkPeakComputeTasks(0);
+
+ // Wait for all results before clients are closed.
+ for (IgniteInternalFuture<Object> fut : futs)
+ fut.get(getTestTimeout(), TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /**
+ * Check that the metric is not less than the current value, so it reports
long-running tasks after their peak has
+ * expired.
+ */
+ @Test
+ public void testComputeTasksMetricOnLongRunningTasks() throws Exception {
+ startGridWithCompute();
+
+ try (IgniteClient client =
Ignition.startClient(getClientConfiguration())) {
+ List<IgniteInternalFuture<Object>> 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.
Review Comment:
This statement is based on internal implementation knowledge, again, it's
better to rely on timer and fair expiration instead of reset.
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java:
##########
@@ -160,6 +204,285 @@ public void testClientListenerMetricsRejectGeneral()
throws Exception {
}
}
+ /**
+ * Check that the active compute tasks metric reports the peak of the
busiest connection (not the sum), keeps it
+ * during the time interval and drops it after the time interval.
+ */
+ @Test
+ public void testComputeTasksMetrics() throws Exception {
+ startGridWithCompute();
+
+ List<IgniteInternalFuture<Object>> futs = new ArrayList<>();
+
+ try (
+ IgniteClient client1 =
Ignition.startClient(getClientConfiguration());
+ IgniteClient client2 =
Ignition.startClient(getClientConfiguration())
+ ) {
+ checkPeakComputeTasks(0);
+ assertEquals(0, activeComputeTasksCount());
+
+ // Tasks 0, 1.
+ futs.addAll(startTasks(client1, 2));
+ checkPeakComputeTasks(2);
+ assertEquals(2, activeComputeTasksCount());
+
+ // Run tasks 2, 3. Metric should not change.
+ futs.addAll(startTasks(client2, 2));
+ checkPeakComputeTasks(2);
+ assertEquals(4, activeComputeTasksCount());
+
+ // Task 4. Metric should increase.
+ futs.addAll(startTasks(client2, 1));
+ checkPeakComputeTasks(MAX_TASKS);
+ assertEquals(5, activeComputeTasksCount());
+
+ assertComputeTaskRejected(client2);
+
+ // A rejected task counts as the
'MaxActiveComputeTasksPerConnection' limit.
+ checkPeakComputeTasks(MAX_TASKS);
+ assertEquals(5, activeComputeTasksCount());
+
+ // Finished tasks are released by the connections, but the metric
keeps the peak during the time interval.
+ finishTasks(2);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ finishTasks(0, 1);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ finishTasks(3, 4);
+ checkPeakComputeTasks(MAX_TASKS);
+
+ // Clear the peak, as if it expired. No tasks are active, so the
metric reports zero.
+ computeTasksMetric().reset();
Review Comment:
Hint: You can use something like `GridTestClockTimer.timeSupplier(() ->
U.currentTimeMillis() + LONG_INTERVAL);` to emulate time shift and metric
expiration without explicit `reset` (But don't forget to return to default
supplier after test, or it can fail other tests in the same JVM)
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/odbc/ClientListenerMetricsTest.java:
##########
@@ -217,4 +540,59 @@ private void checkConnectionsMetrics(MetricRegistry mreg,
int accepted, int acti
assertEquals(0,
mreg.<IntMetric>findMetric(METRIC_REJECTED_AUTHENTICATION).value());
assertEquals(0,
mreg.<IntMetric>findMetric(METRIC_REJECTED_TOTAL).value());
}
+
+ /** Task that waits for the latch on reduce. */
+ public static class TestComputeTask extends ComputeTaskAdapter<Object,
Object> {
+ /** Tasks instances. */
+ private static final List<TestComputeTask> TASKS = new
CopyOnWriteArrayList<>();
+
+ /** Finish latch. */
+ private final CountDownLatch finishLatch = new CountDownLatch(1);
+
+ /** Default constructor. */
+ public TestComputeTask() {
+ TASKS.add(this);
Review Comment:
`this` instance leakage in constructor, this is bad according to JMM. Yes,
it's a test, but still can have some effects and can make test flaky. Let's
move this to `map`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
