This is an automated email from the ASF dual-hosted git repository.
nizhikov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 0af17321c89 IGNITE-28819 [3/N] Thin client implementation extraction
(#13338)
0af17321c89 is described below
commit 0af17321c89fc58a01320f9fe3201cd67ecbb1aa
Author: Nikolay <[email protected]>
AuthorDate: Wed Jul 8 17:56:02 2026 +0300
IGNITE-28819 [3/N] Thin client implementation extraction (#13338)
---
.../ignite/internal/processors/tracing/MTC.java | 0
.../internal/processors/tracing/NoopSpan.java | 0
.../processors/tracing/NoopSpanManager.java} | 30 +---
.../ignite/internal/processors/tracing/Span.java | 0
.../internal/processors/tracing/SpanManager.java | 7 +
.../internal/processors/tracing/SpanTags.java | 0
.../internal/processors/tracing/SpanType.java | 0
.../processors/tracing/messages/SpanTransport.java | 0
.../java/org/apache/ignite/spi/tracing/Scope.java | 0
.../org/apache/ignite/spi/tracing/SpanStatus.java | 0
.../GridNioClientConnectionMultiplexer.java | 3 +-
.../managers/tracing/GridTracingManager.java | 7 +
.../processors/odbc/ClientListenerProcessor.java | 12 +-
.../rest/protocols/tcp/GridTcpRestProtocol.java | 17 ++-
.../internal/processors/tracing/NoopTracing.java | 29 +---
.../apache/ignite/internal/util/IgniteUtils.java | 70 +++++++++
.../ignite/internal/util/nio/GridNioServer.java | 164 ++++++++++++---------
.../internal/util/nio/GridNioTracerFilter.java | 10 +-
.../util/nio/GridSelectorNioSessionImpl.java | 60 ++++----
.../internal/util/nio/ssl/GridNioSslFilter.java | 31 ++--
.../tcp/TcpCommunicationMetricsListener.java | 19 +++
.../spi/communication/tcp/TcpCommunicationSpi.java | 5 +-
.../tcp/internal/GridNioServerWrapper.java | 20 ++-
.../internal/util/nio/GridNioSslSelfTest.java | 2 +-
24 files changed, 288 insertions(+), 198 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/MTC.java
b/modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/MTC.java
similarity index 100%
rename from
modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/MTC.java
rename to
modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/MTC.java
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/NoopSpan.java
b/modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/NoopSpan.java
similarity index 100%
rename from
modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/NoopSpan.java
rename to
modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/NoopSpan.java
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/NoopTracing.java
b/modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/NoopSpanManager.java
similarity index 65%
copy from
modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/NoopTracing.java
copy to
modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/NoopSpanManager.java
index 77563757c57..0d948b5d41f 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/NoopTracing.java
+++
b/modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/NoopSpanManager.java
@@ -17,35 +17,17 @@
package org.apache.ignite.internal.processors.tracing;
-import
org.apache.ignite.internal.processors.tracing.configuration.NoopTracingConfigurationManager;
-import
org.apache.ignite.internal.processors.tracing.messages.TraceableMessagesHandler;
-import org.apache.ignite.logger.NullLogger;
-import org.apache.ignite.spi.tracing.TracingConfigurationManager;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
- * Noop implementation of {@link Tracing}.
+ * Noop implementation of {@link SpanManager}.
*/
-public class NoopTracing implements Tracing {
+public class NoopSpanManager implements SpanManager {
/** Noop serialized span. */
public static final byte[] NOOP_SERIALIZED_SPAN = new byte[0];
- /** Traceable messages handler. */
- private final TraceableMessagesHandler msgHnd;
-
- /**
- * Constructor.
- */
- public NoopTracing() {
- msgHnd = new TraceableMessagesHandler(this, new NullLogger());
- }
-
- /** {@inheritDoc} */
- @Override public TraceableMessagesHandler messages() {
- return msgHnd;
- }
-
/** {@inheritDoc} */
@Override public Span create(@NotNull SpanType spanType, @Nullable Span
parentSpan) {
return NoopSpan.INSTANCE;
@@ -60,7 +42,7 @@ public class NoopTracing implements Tracing {
@Override public @NotNull Span create(
@NotNull SpanType spanType,
@Nullable Span parentSpan,
- @Nullable String label) {
+ @Nullable String lb) {
return NoopSpan.INSTANCE;
}
@@ -70,7 +52,7 @@ public class NoopTracing implements Tracing {
}
/** {@inheritDoc} */
- @Override public @NotNull TracingConfigurationManager configuration() {
- return NoopTracingConfigurationManager.INSTANCE;
+ @Override public String traceName(Message msg) {
+ return null;
}
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/Span.java
b/modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/Span.java
similarity index 100%
rename from
modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/Span.java
rename to
modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/Span.java
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanManager.java
b/modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/SpanManager.java
similarity index 89%
rename from
modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanManager.java
rename to
modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/SpanManager.java
index 010506dca39..be43ec4cabb 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanManager.java
+++
b/modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/SpanManager.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.tracing;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -70,4 +71,10 @@ public interface SpanManager {
* @param span Span.
*/
byte[] serialize(@NotNull Span span);
+
+ /**
+ * @param msg Message to resolve a trace name for.
+ * @return Trace name of the message, or {@code null} if tracing is
disabled.
+ */
+ @Nullable String traceName(Message msg);
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanTags.java
b/modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/SpanTags.java
similarity index 100%
rename from
modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanTags.java
rename to
modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/SpanTags.java
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanType.java
b/modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/SpanType.java
similarity index 100%
rename from
modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanType.java
rename to
modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/SpanType.java
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/messages/SpanTransport.java
b/modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/messages/SpanTransport.java
similarity index 100%
rename from
modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/messages/SpanTransport.java
rename to
modules/commons/src/main/java/org/apache/ignite/internal/processors/tracing/messages/SpanTransport.java
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/tracing/Scope.java
b/modules/commons/src/main/java/org/apache/ignite/spi/tracing/Scope.java
similarity index 100%
rename from modules/core/src/main/java/org/apache/ignite/spi/tracing/Scope.java
rename to modules/commons/src/main/java/org/apache/ignite/spi/tracing/Scope.java
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/tracing/SpanStatus.java
b/modules/commons/src/main/java/org/apache/ignite/spi/tracing/SpanStatus.java
similarity index 100%
rename from
modules/core/src/main/java/org/apache/ignite/spi/tracing/SpanStatus.java
rename to
modules/commons/src/main/java/org/apache/ignite/spi/tracing/SpanStatus.java
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/io/gridnioserver/GridNioClientConnectionMultiplexer.java
b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/io/gridnioserver/GridNioClientConnectionMultiplexer.java
index a26a6f2dac7..a1505e9c7f1 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/io/gridnioserver/GridNioClientConnectionMultiplexer.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/io/gridnioserver/GridNioClientConnectionMultiplexer.java
@@ -43,6 +43,7 @@ import org.apache.ignite.internal.util.nio.GridNioFilter;
import org.apache.ignite.internal.util.nio.GridNioServer;
import org.apache.ignite.internal.util.nio.GridNioSession;
import org.apache.ignite.internal.util.nio.ssl.GridNioSslFilter;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.logger.NullLogger;
/**
@@ -82,7 +83,7 @@ public class GridNioClientConnectionMultiplexer implements
ClientConnectionMulti
sslCtx = ClientSslUtils.getSslContext(cfg);
if (sslCtx != null) {
- GridNioSslFilter sslFilter = new GridNioSslFilter(sslCtx, true,
ByteOrder.nativeOrder(), gridLog, null);
+ GridNioSslFilter sslFilter = U.sslFilter(sslCtx, true,
ByteOrder.nativeOrder(), gridLog, null);
sslFilter.directMode(false);
filters = new GridNioFilter[] {codecFilter, sslFilter};
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/managers/tracing/GridTracingManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/managers/tracing/GridTracingManager.java
index 4e5e66e6924..4e087102488 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/managers/tracing/GridTracingManager.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/managers/tracing/GridTracingManager.java
@@ -34,8 +34,10 @@ import
org.apache.ignite.internal.processors.tracing.SpanType;
import org.apache.ignite.internal.processors.tracing.Tracing;
import
org.apache.ignite.internal.processors.tracing.configuration.GridTracingConfigurationManager;
import
org.apache.ignite.internal.processors.tracing.messages.TraceableMessagesHandler;
+import
org.apache.ignite.internal.processors.tracing.messages.TraceableMessagesTable;
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.logger.NullLogger;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.IgniteSpiException;
import org.apache.ignite.spi.tracing.NoopTracingSpi;
import org.apache.ignite.spi.tracing.Scope;
@@ -507,6 +509,11 @@ public class GridTracingManager extends
GridManagerAdapter<TracingSpi> implement
}
}
+ /** {@inheritDoc} */
+ @Override public String traceName(Message msg) {
+ return TraceableMessagesTable.traceName(msg);
+ }
+
/** {@inheritDoc} */
@Override public TraceableMessagesHandler messages() {
// Optimization for noop spi.
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 68af9dfe14a..808437eb171 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
@@ -194,7 +194,7 @@ public class ClientListenerProcessor extends
GridProcessorAdapter {
for (int port = cliConnCfg.getPort(); port <= portTo && port
<= 65535; port++) {
try {
- srv = GridNioServer.<ClientMessage>builder()
+ GridNioServer.Builder<ClientMessage> builder =
GridNioServer.<ClientMessage>builder()
.address(hostAddr)
.port(port)
.listener(new ClientListenerNioListener(ctx,
busyLock, cliConnCfg, metrics, newConnEnabled))
@@ -210,9 +210,11 @@ public class ClientListenerProcessor extends
GridProcessorAdapter {
.filters(filters)
.directMode(true)
.idleTimeout(idleTimeout > 0 ? idleTimeout :
Long.MAX_VALUE)
- .metricRegistry(mreg)
- .messageQueueSizeListener(msgQueueSizeLsnr)
- .build();
+ .messageQueueSizeListener(msgQueueSizeLsnr);
+
+ srv = U.setNioServerMetrics(builder, mreg).build();
+
+ U.registerNioServerMetrics(srv, filters, mreg);
ctx.ports().registerPort(port, IgnitePortProtocol.TCP,
getClass());
@@ -492,7 +494,7 @@ public class ClientListenerProcessor extends
GridProcessorAdapter {
throw new IgniteCheckedException("Failed to create client
listener " +
"(SSL is enabled but factory is null). Check the
ClientConnectorConfiguration");
- GridNioSslFilter sslFilter = new
GridNioSslFilter(sslCtxFactory.create(),
+ GridNioSslFilter sslFilter = U.sslFilter(sslCtxFactory.create(),
true, ByteOrder.nativeOrder(), log,
ctx.metric().registry(CLIENT_CONNECTOR_METRICS));
sslFilter.directMode(true);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestProtocol.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestProtocol.java
index a1aaa0910dd..d8825397806 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestProtocol.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/protocols/tcp/GridTcpRestProtocol.java
@@ -36,6 +36,7 @@ import
org.apache.ignite.internal.client.marshaller.GridClientMarshaller;
import
org.apache.ignite.internal.client.marshaller.jdk.GridClientJdkMarshaller;
import
org.apache.ignite.internal.client.marshaller.optimized.GridClientOptimizedMarshaller;
import
org.apache.ignite.internal.client.marshaller.optimized.GridClientZipOptimizedMarshaller;
+import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
import org.apache.ignite.internal.processors.rest.GridRestProtocolHandler;
import
org.apache.ignite.internal.processors.rest.client.message.GridClientMessage;
import
org.apache.ignite.internal.processors.rest.protocols.GridRestProtocolAdapter;
@@ -208,13 +209,15 @@ public class GridTcpRestProtocol extends
GridRestProtocolAdapter {
GridNioFilter[] filters;
+ MetricRegistryImpl mreg =
ctx.metric().registry(REST_CONNECTOR_METRIC_REGISTRY_NAME);
+
if (sslCtx != null) {
- GridNioSslFilter sslFilter = new GridNioSslFilter(
+ GridNioSslFilter sslFilter = U.sslFilter(
sslCtx,
cfg.isDirectBuffer(),
ByteOrder.nativeOrder(),
log,
-
ctx.metric().registry(REST_CONNECTOR_METRIC_REGISTRY_NAME));
+ mreg);
sslFilter.directMode(false);
@@ -232,7 +235,7 @@ public class GridTcpRestProtocol extends
GridRestProtocolAdapter {
else
filters = new GridNioFilter[] { codec };
- srv = GridNioServer.<GridClientMessage>builder()
+ GridNioServer.Builder<GridClientMessage> builder =
GridNioServer.<GridClientMessage>builder()
.address(hostAddr)
.port(port)
.listener(lsnr)
@@ -247,9 +250,11 @@ public class GridTcpRestProtocol extends
GridRestProtocolAdapter {
.socketReceiveBufferSize(cfg.getReceiveBufferSize())
.sendQueueLimit(cfg.getSendQueueLimit())
.filters(filters)
- .directMode(false)
-
.metricRegistry(ctx.metric().registry(REST_CONNECTOR_METRIC_REGISTRY_NAME))
- .build();
+ .directMode(false);
+
+ srv = U.setNioServerMetrics(builder, mreg).build();
+
+ U.registerNioServerMetrics(srv, filters, mreg);
srv.idleTimeout(cfg.getIdleTimeout());
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/NoopTracing.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/NoopTracing.java
index 77563757c57..c7cd10f5626 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/NoopTracing.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/NoopTracing.java
@@ -22,15 +22,11 @@ import
org.apache.ignite.internal.processors.tracing.messages.TraceableMessagesH
import org.apache.ignite.logger.NullLogger;
import org.apache.ignite.spi.tracing.TracingConfigurationManager;
import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
/**
* Noop implementation of {@link Tracing}.
*/
-public class NoopTracing implements Tracing {
- /** Noop serialized span. */
- public static final byte[] NOOP_SERIALIZED_SPAN = new byte[0];
-
+public class NoopTracing extends NoopSpanManager implements Tracing {
/** Traceable messages handler. */
private final TraceableMessagesHandler msgHnd;
@@ -46,29 +42,6 @@ public class NoopTracing implements Tracing {
return msgHnd;
}
- /** {@inheritDoc} */
- @Override public Span create(@NotNull SpanType spanType, @Nullable Span
parentSpan) {
- return NoopSpan.INSTANCE;
- }
-
- /** {@inheritDoc} */
- @Override public Span create(@NotNull SpanType spanType, @Nullable byte[]
serializedParentSpan) {
- return NoopSpan.INSTANCE;
- }
-
- /** {@inheritDoc} */
- @Override public @NotNull Span create(
- @NotNull SpanType spanType,
- @Nullable Span parentSpan,
- @Nullable String label) {
- return NoopSpan.INSTANCE;
- }
-
- /** {@inheritDoc} */
- @Override public byte[] serialize(@NotNull Span span) {
- return NOOP_SERIALIZED_SPAN;
- }
-
/** {@inheritDoc} */
@Override public @NotNull TracingConfigurationManager configuration() {
return NoopTracingConfigurationManager.INSTANCE;
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
index 8c2f934aa4f..2223082eca0 100755
---
a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
@@ -114,6 +114,7 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.function.Consumer;
+import java.util.function.LongConsumer;
import java.util.jar.JarFile;
import java.util.logging.ConsoleHandler;
import java.util.logging.Handler;
@@ -134,6 +135,7 @@ import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.HttpsURLConnection;
+import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSession;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
@@ -186,6 +188,7 @@ import
org.apache.ignite.internal.processors.cache.CacheDefaultBinaryAffinityKey
import org.apache.ignite.internal.processors.cache.CacheObjectContext;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import
org.apache.ignite.internal.processors.cache.IgnitePeerToPeerClassLoadingException;
+import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
import
org.apache.ignite.internal.transactions.IgniteTxHeuristicCheckedException;
import
org.apache.ignite.internal.transactions.IgniteTxOptimisticCheckedException;
import
org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
@@ -194,6 +197,9 @@ import
org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.future.IgniteFutureImpl;
import org.apache.ignite.internal.util.lang.GridPeerDeployAware;
import org.apache.ignite.internal.util.lang.IgniteThrowableFunction;
+import org.apache.ignite.internal.util.nio.GridNioFilter;
+import org.apache.ignite.internal.util.nio.GridNioServer;
+import org.apache.ignite.internal.util.nio.ssl.GridNioSslFilter;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
@@ -7880,4 +7886,68 @@ public abstract class IgniteUtils extends CommonUtils {
return msg instanceof SecurityAwareCustomMessageWrapper ?
((SecurityAwareCustomMessageWrapper)msg).delegate() :
(DiscoveryCustomMessage)msg;
}
+
+ /**
+ * Sets the received/sent bytes and per-session queue-size metric
consumers on the given NIO server builder,
+ * creating the underlying metrics in the provided registry.
+ *
+ * @param builder NIO server builder.
+ * @param mreg Metric registry.
+ * @return The given builder for chaining.
+ */
+ public static <T> GridNioServer.Builder<T>
setNioServerMetrics(GridNioServer.Builder<T> builder, MetricRegistryImpl mreg) {
+ return builder
+ .receivedBytesMetric(mreg.longAdderMetric(
+ GridNioServer.RECEIVED_BYTES_METRIC_NAME,
GridNioServer.RECEIVED_BYTES_METRIC_DESC)::add)
+ .sentBytesMetric(mreg.longAdderMetric(
+ GridNioServer.SENT_BYTES_METRIC_NAME,
GridNioServer.SENT_BYTES_METRIC_DESC)::add)
+ .outboundMessagesQueueSizeMetric(mreg.longAdderMetric(
+ GridNioServer.OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_NAME,
+ GridNioServer.OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_DESC)::add)
+ .maxMessagesQueueSizeMetric(mreg.maxValueMetric(
+ GridNioServer.MAX_MESSAGES_QUEUE_SIZE_METRIC_NAME,
+ GridNioServer.MAX_MESSAGES_QUEUE_SIZE_METRIC_DESC, 60_000,
5)::update);
+ }
+
+ /**
+ * Registers the active TCP sessions count metric in the given registry,
backed by the NIO server.
+ *
+ * @param srv NIO server.
+ * @param mreg Metric registry.
+ */
+ public static void registerNioServerMetrics(GridNioServer<?> srv,
GridNioFilter[] filters, MetricRegistryImpl mreg) {
+ boolean sslEnabled = Arrays.stream(filters).anyMatch(filter -> filter
instanceof GridNioSslFilter);
+
+ mreg.register(GridNioServer.SSL_ENABLED_METRIC_NAME, () -> sslEnabled,
"Whether SSL is enabled");
+ mreg.register(GridNioServer.SESSIONS_CNT_METRIC_NAME,
srv::activeTcpSessionsCount, "Active TCP sessions count.");
+ }
+
+ /**
+ * Creates an SSL NIO filter, wiring its metrics from the given registry.
+ *
+ * @param sslCtx SSL context.
+ * @param directBuf Direct buffer flag.
+ * @param order Byte order.
+ * @param log Logger to use.
+ * @param mreg Optional metric registry; if {@code null}, the filter is
created without metrics.
+ * @return SSL NIO filter.
+ */
+ public static GridNioSslFilter sslFilter(
+ SSLContext sslCtx,
+ boolean directBuf,
+ ByteOrder order,
+ IgniteLogger log,
+ @Nullable MetricRegistryImpl mreg
+ ) {
+ LongConsumer handshakeDuration = mreg == null ? null : mreg.histogram(
+ GridNioSslFilter.SSL_HANDSHAKE_DURATION_HISTOGRAM_METRIC_NAME,
+ new long[] {250, 500, 1000},
+ "SSL handshake duration in milliseconds.")::value;
+
+ Runnable rejectedSesCnt = mreg == null ? null : mreg.intMetric(
+ GridNioSslFilter.SSL_REJECTED_SESSIONS_CNT_METRIC_NAME,
+ "TCP sessions count that were rejected due to SSL
errors.")::increment;
+
+ return new GridNioSslFilter(sslCtx, directBuf, order, log,
handshakeDuration, rejectedSesCnt);
+ }
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
index 797290677db..2668427f424 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
@@ -47,6 +47,7 @@ import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongConsumer;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
@@ -55,17 +56,15 @@ import
org.apache.ignite.configuration.ConnectorConfiguration;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
-import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
-import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
import org.apache.ignite.internal.processors.odbc.ClientMessage;
import org.apache.ignite.internal.processors.tracing.MTC;
import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
import org.apache.ignite.internal.processors.tracing.NoopSpan;
-import org.apache.ignite.internal.processors.tracing.NoopTracing;
+import org.apache.ignite.internal.processors.tracing.NoopSpanManager;
import org.apache.ignite.internal.processors.tracing.Span;
+import org.apache.ignite.internal.processors.tracing.SpanManager;
import org.apache.ignite.internal.processors.tracing.SpanTags;
import org.apache.ignite.internal.processors.tracing.SpanType;
-import org.apache.ignite.internal.processors.tracing.Tracing;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
import org.apache.ignite.internal.util.GridUnsafe;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
@@ -97,7 +96,6 @@ import static
org.apache.ignite.failure.FailureType.CRITICAL_ERROR;
import static org.apache.ignite.failure.FailureType.SYSTEM_WORKER_TERMINATION;
import static
org.apache.ignite.internal.processors.tracing.SpanTags.SOCKET_WRITE_BYTES;
import static
org.apache.ignite.internal.processors.tracing.SpanType.COMMUNICATION_SOCKET_WRITE;
-import static
org.apache.ignite.internal.processors.tracing.messages.TraceableMessagesTable.traceName;
import static
org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.MSG_WRITER;
import static
org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.NIO_OPERATION;
@@ -251,17 +249,17 @@ public class GridNioServer<T> {
/** Whether direct mode is used. */
private final boolean directMode;
- /** */
- @Nullable private final MetricRegistryImpl mreg;
-
/** Received bytes count metric. */
- @Nullable private final LongAdderMetric rcvdBytesCntMetric;
+ @Nullable private final LongConsumer rcvdBytesCntMetric;
/** Sent bytes count metric. */
- @Nullable private final LongAdderMetric sentBytesCntMetric;
+ @Nullable private final LongConsumer sentBytesCntMetric;
/** Outbound messages queue size. */
- @Nullable private final LongAdderMetric outboundMessagesQueueSizeMetric;
+ @Nullable private final LongConsumer outboundMessagesQueueSizeMetric;
+
+ /** Per-session maximum outbound messages queue size metric. */
+ @Nullable private final LongConsumer maxMessagesQueueSizeMetric;
/** Sessions. */
private final GridConcurrentHashSet<GridSelectorNioSessionImpl> sessions =
new GridConcurrentHashSet<>();
@@ -295,8 +293,8 @@ public class GridNioServer<T> {
*/
private final boolean readWriteSelectorsAssign;
- /** Tracing processor. */
- private Tracing tracing;
+ /** Span manager. */
+ private SpanManager tracing;
/** Message factory. */
private final MessageFactory msgFactory;
@@ -325,7 +323,10 @@ public class GridNioServer<T> {
* @param msgQueueLsnr Message queue size listener.
* @param readWriteSelectorsAssign If {@code true} then in/out connections
are assigned to even/odd workers.
* @param workerLsnr Worker lifecycle listener.
- * @param mreg Metrics registry.
+ * @param rcvdBytesCntMetric Received bytes count metric, or {@code null}
if metrics disabled.
+ * @param sentBytesCntMetric Sent bytes count metric, or {@code null} if
metrics disabled.
+ * @param outboundMessagesQueueSizeMetric Per-session outbound messages
queue size metric, or {@code null} if metrics disabled.
+ * @param maxMessagesQueueSizeMetric Per-session maximum outbound messages
queue size metric, or {@code null} if metrics disabled.
* @param filters Filters for this server.
* @throws IgniteCheckedException If failed.
*/
@@ -351,8 +352,11 @@ public class GridNioServer<T> {
IgniteBiInClosure<GridNioSession, Integer> msgQueueLsnr,
boolean readWriteSelectorsAssign,
@Nullable GridWorkerListener workerLsnr,
- @Nullable MetricRegistryImpl mreg,
- Tracing tracing,
+ @Nullable LongConsumer rcvdBytesCntMetric,
+ @Nullable LongConsumer sentBytesCntMetric,
+ @Nullable LongConsumer outboundMessagesQueueSizeMetric,
+ @Nullable LongConsumer maxMessagesQueueSizeMetric,
+ SpanManager tracing,
MessageFactory msgFactory,
GridNioFilter... filters
) throws IgniteCheckedException {
@@ -380,7 +384,11 @@ public class GridNioServer<T> {
this.selectorSpins = selectorSpins;
this.readWriteSelectorsAssign = readWriteSelectorsAssign;
this.lsnr = lsnr;
- this.tracing = tracing == null ? new NoopTracing() : tracing;
+ this.rcvdBytesCntMetric = rcvdBytesCntMetric;
+ this.sentBytesCntMetric = sentBytesCntMetric;
+ this.outboundMessagesQueueSizeMetric = outboundMessagesQueueSizeMetric;
+ this.maxMessagesQueueSizeMetric = maxMessagesQueueSizeMetric;
+ this.tracing = tracing == null ? new NoopSpanManager() : tracing;
this.msgFactory = msgFactory;
filterChain = new GridNioFilterChain<>(log, lsnr, new HeadFilter(),
filters);
@@ -461,27 +469,13 @@ public class GridNioServer<T> {
}
this.balancer = balancer0;
+ }
- this.mreg = mreg;
-
- rcvdBytesCntMetric = mreg == null ?
- null : mreg.longAdderMetric(RECEIVED_BYTES_METRIC_NAME,
RECEIVED_BYTES_METRIC_DESC);
-
- sentBytesCntMetric = mreg == null ?
- null : mreg.longAdderMetric(SENT_BYTES_METRIC_NAME,
SENT_BYTES_METRIC_DESC);
-
- outboundMessagesQueueSizeMetric = mreg == null ? null :
mreg.longAdderMetric(
- OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_NAME,
- OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_DESC
- );
-
- if (mreg != null) {
- mreg.register(SESSIONS_CNT_METRIC_NAME, sessions::size, "Active
TCP sessions count.");
-
- boolean sslEnabled = Arrays.stream(filters).anyMatch(filter ->
filter instanceof GridNioSslFilter);
-
- mreg.register(SSL_ENABLED_METRIC_NAME, () -> sslEnabled, "Whether
SSL is enabled");
- }
+ /**
+ * @return Number of active TCP sessions.
+ */
+ public int activeTcpSessionsCount() {
+ return sessions.size();
}
/**
@@ -1249,7 +1243,7 @@ public class GridNioServer<T> {
log.trace("Bytes received [sockCh=" + sockCh + ", cnt=" + cnt
+ ']');
if (rcvdBytesCntMetric != null)
- rcvdBytesCntMetric.add(cnt);
+ rcvdBytesCntMetric.accept(cnt);
ses.bytesReceived(cnt);
@@ -1316,7 +1310,7 @@ public class GridNioServer<T> {
span.addTag(SOCKET_WRITE_BYTES, () ->
Integer.toString(cnt));
if (sentBytesCntMetric != null)
- sentBytesCntMetric.add(cnt);
+ sentBytesCntMetric.accept(cnt);
ses.bytesSent(cnt);
}
@@ -1418,7 +1412,7 @@ public class GridNioServer<T> {
return;
if (rcvdBytesCntMetric != null)
- rcvdBytesCntMetric.add(cnt);
+ rcvdBytesCntMetric.accept(cnt);
ses.bytesReceived(cnt);
onRead(cnt);
@@ -1491,7 +1485,7 @@ public class GridNioServer<T> {
int cnt = sockCh.write(sslNetBuf);
if (sentBytesCntMetric != null)
- sentBytesCntMetric.add(cnt);
+ sentBytesCntMetric.accept(cnt);
ses.bytesSent(cnt);
@@ -1578,7 +1572,7 @@ public class GridNioServer<T> {
log.trace("Bytes sent [sockCh=" + sockCh + ",
cnt=" + cnt + ']');
if (sentBytesCntMetric != null)
- sentBytesCntMetric.add(cnt);
+ sentBytesCntMetric.accept(cnt);
ses.bytesSent(cnt);
}
@@ -1636,7 +1630,7 @@ public class GridNioServer<T> {
Span span = tracing.create(SpanType.COMMUNICATION_SOCKET_WRITE,
req.span());
try (TraceSurroundings ignore = span.equals(NoopSpan.INSTANCE) ?
null : MTC.support(span)) {
- span.addTag(SpanTags.MESSAGE, () -> traceName(msg));
+ span.addTag(SpanTags.MESSAGE, () -> tracing.traceName(msg));
assert msg != null;
@@ -1687,7 +1681,7 @@ public class GridNioServer<T> {
int cnt = sockCh.write(buf);
if (sentBytesCntMetric != null)
- sentBytesCntMetric.add(cnt);
+ sentBytesCntMetric.accept(cnt);
ses.bytesSent(cnt);
@@ -1778,7 +1772,7 @@ public class GridNioServer<T> {
log.trace("Bytes sent [sockCh=" + sockCh + ", cnt=" + cnt
+ ']');
if (sentBytesCntMetric != null)
- sentBytesCntMetric.add(cnt);
+ sentBytesCntMetric.accept(cnt);
ses.bytesSent(cnt);
onWrite(cnt);
@@ -1839,7 +1833,7 @@ public class GridNioServer<T> {
Span span = tracing.create(SpanType.COMMUNICATION_SOCKET_WRITE,
req.span());
try (TraceSurroundings ignore = span.equals(NoopSpan.INSTANCE) ?
null : MTC.support(span)) {
- span.addTag(SpanTags.MESSAGE, () -> traceName(msg));
+ span.addTag(SpanTags.MESSAGE, () -> tracing.traceName(msg));
int startPos = buf.position();
@@ -2782,7 +2776,9 @@ public class GridNioServer<T> {
(InetSocketAddress)sockCh.getRemoteAddress(),
fut.accepted(),
sndQueueLimit,
- mreg,
+ outboundMessagesQueueSizeMetric,
+ maxMessagesQueueSizeMetric,
+ tracing,
writeBuf,
readBuf);
@@ -3057,18 +3053,6 @@ public class GridNioServer<T> {
}
}
- /**
- * Gets outbound messages queue size.
- *
- * @return Write queue size.
- */
- public int outboundMessagesQueueSize() {
- if (outboundMessagesQueueSizeMetric == null)
- return -1;
-
- return (int)outboundMessagesQueueSizeMetric.value();
- }
-
/**
* A separate thread that will accept incoming connections and schedule
read to some worker.
*/
@@ -3946,11 +3930,20 @@ public class GridNioServer<T> {
/** Worker lifecycle listener to be used by server's worker threads. */
private GridWorkerListener workerLsnr;
- /** Metrics registry. */
- private MetricRegistryImpl mreg;
+ /** Received bytes count metric. */
+ private LongConsumer rcvdBytesCntMetric;
+
+ /** Sent bytes count metric. */
+ private LongConsumer sentBytesCntMetric;
+
+ /** Per-session outbound messages queue size metric. */
+ private LongConsumer outboundMessagesQueueSizeMetric;
+
+ /** Per-session maximum outbound messages queue size metric. */
+ private LongConsumer maxMessagesQueueSizeMetric;
- /** Tracing processor */
- private Tracing tracing;
+ /** Span manager */
+ private SpanManager tracing;
/** Message factory. */
private MessageFactory msgFactory;
@@ -3984,7 +3977,10 @@ public class GridNioServer<T> {
msgQueueLsnr,
readWriteSelectorsAssign,
workerLsnr,
- mreg,
+ rcvdBytesCntMetric,
+ sentBytesCntMetric,
+ outboundMessagesQueueSizeMetric,
+ maxMessagesQueueSizeMetric,
tracing,
msgFactory,
filters != null ? Arrays.copyOf(filters, filters.length) :
EMPTY_FILTERS
@@ -4010,10 +4006,10 @@ public class GridNioServer<T> {
}
/**
- * @param tracing Tracing processor.
+ * @param tracing Span manager.
* @return This for chaining.
*/
- public Builder<T> tracing(Tracing tracing) {
+ public Builder<T> tracing(SpanManager tracing) {
this.tracing = tracing;
return this;
@@ -4253,11 +4249,41 @@ public class GridNioServer<T> {
}
/**
- * @param mreg Metrics registry.
+ * @param rcvdBytesCntMetric Received bytes count metric.
+ * @return This for chaining.
+ */
+ public Builder<T> receivedBytesMetric(LongConsumer rcvdBytesCntMetric)
{
+ this.rcvdBytesCntMetric = rcvdBytesCntMetric;
+
+ return this;
+ }
+
+ /**
+ * @param sentBytesCntMetric Sent bytes count metric.
+ * @return This for chaining.
+ */
+ public Builder<T> sentBytesMetric(LongConsumer sentBytesCntMetric) {
+ this.sentBytesCntMetric = sentBytesCntMetric;
+
+ return this;
+ }
+
+ /**
+ * @param outboundMessagesQueueSizeMetric Per-session outbound
messages queue size metric.
+ * @return This for chaining.
+ */
+ public Builder<T> outboundMessagesQueueSizeMetric(LongConsumer
outboundMessagesQueueSizeMetric) {
+ this.outboundMessagesQueueSizeMetric =
outboundMessagesQueueSizeMetric;
+
+ return this;
+ }
+
+ /**
+ * @param maxMessagesQueueSizeMetric Per-session maximum outbound
messages queue size metric.
* @return This for chaining.
*/
- public Builder<T> metricRegistry(MetricRegistryImpl mreg) {
- this.mreg = mreg;
+ public Builder<T> maxMessagesQueueSizeMetric(LongConsumer
maxMessagesQueueSizeMetric) {
+ this.maxMessagesQueueSizeMetric = maxMessagesQueueSizeMetric;
return this;
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioTracerFilter.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioTracerFilter.java
index 650aeb2cc7c..47644636def 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioTracerFilter.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioTracerFilter.java
@@ -23,9 +23,9 @@ import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.tracing.MTC;
import org.apache.ignite.internal.processors.tracing.NoopSpan;
-import org.apache.ignite.internal.processors.tracing.NoopTracing;
+import org.apache.ignite.internal.processors.tracing.NoopSpanManager;
import org.apache.ignite.internal.processors.tracing.Span;
-import org.apache.ignite.internal.processors.tracing.Tracing;
+import org.apache.ignite.internal.processors.tracing.SpanManager;
import org.apache.ignite.internal.processors.tracing.messages.SpanTransport;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -42,7 +42,7 @@ public class GridNioTracerFilter extends GridNioFilterAdapter
{
private IgniteLogger log;
/** Tracing processor. */
- private final Tracing tracer;
+ private final SpanManager tracer;
/**
* Creates a tracer filter.
@@ -50,11 +50,11 @@ public class GridNioTracerFilter extends
GridNioFilterAdapter {
* @param log Log instance to use.
* @param tracer Tracing processor.
*/
- public GridNioTracerFilter(IgniteLogger log, Tracing tracer) {
+ public GridNioTracerFilter(IgniteLogger log, SpanManager tracer) {
super("GridNioTracerFilter");
this.log = log;
- this.tracer = tracer == null ? new NoopTracing() : tracer;
+ this.tracer = tracer == null ? new NoopSpanManager() : tracer;
}
/** {@inheritDoc} */
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridSelectorNioSessionImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridSelectorNioSessionImpl.java
index 561941903b6..267851bac0f 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridSelectorNioSessionImpl.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridSelectorNioSessionImpl.java
@@ -26,25 +26,19 @@ import java.util.List;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongConsumer;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
-import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
-import org.apache.ignite.internal.processors.metric.impl.MaxValueMetric;
import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.SpanManager;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.util.deque.FastSizeDeque;
import org.jetbrains.annotations.Nullable;
-import static
org.apache.ignite.internal.processors.tracing.messages.TraceableMessagesTable.traceName;
-import static
org.apache.ignite.internal.util.nio.GridNioServer.MAX_MESSAGES_QUEUE_SIZE_METRIC_DESC;
-import static
org.apache.ignite.internal.util.nio.GridNioServer.MAX_MESSAGES_QUEUE_SIZE_METRIC_NAME;
-import static
org.apache.ignite.internal.util.nio.GridNioServer.OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_DESC;
-import static
org.apache.ignite.internal.util.nio.GridNioServer.OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_NAME;
-
/**
* Session implementation bound to selector API and socket API.
* Note that this implementation requires non-null values for local and remote
@@ -93,10 +87,13 @@ public class GridSelectorNioSessionImpl extends
GridNioSessionImpl implements Gr
private volatile boolean closeSocket = true;
/** Outbound messages queue size metric. */
- @Nullable private final LongAdderMetric outboundMessagesQueueSizeMetric;
+ @Nullable private final LongConsumer outboundMessagesQueueSizeMetric;
/** Maximum outbound messages queue size metric. */
- @Nullable private final MaxValueMetric maxMessagesQueueSizeMetric;
+ @Nullable private final LongConsumer maxMessagesQueueSizeMetric;
+
+ /** Span manager used to resolve trace names for logging. */
+ private final SpanManager tracing;
/**
* Creates session instance.
@@ -108,6 +105,9 @@ public class GridSelectorNioSessionImpl extends
GridNioSessionImpl implements Gr
* @param rmtAddr Remote address.
* @param accepted Accepted flag.
* @param sndQueueLimit Send queue limit.
+ * @param outboundMessagesQueueSizeMetric Outbound messages queue size
metric, or {@code null} if metrics disabled.
+ * @param maxMessagesQueueSizeMetric Maximum outbound messages queue size
metric, or {@code null} if metrics disabled.
+ * @param tracing Span manager used to resolve trace names for logging.
* @param writeBuf Write buffer.
* @param readBuf Read buffer.
*/
@@ -119,7 +119,9 @@ public class GridSelectorNioSessionImpl extends
GridNioSessionImpl implements Gr
InetSocketAddress rmtAddr,
boolean accepted,
int sndQueueLimit,
- @Nullable MetricRegistryImpl mreg,
+ @Nullable LongConsumer outboundMessagesQueueSizeMetric,
+ @Nullable LongConsumer maxMessagesQueueSizeMetric,
+ SpanManager tracing,
@Nullable ByteBuffer writeBuf,
@Nullable ByteBuffer readBuf
) {
@@ -151,17 +153,11 @@ public class GridSelectorNioSessionImpl extends
GridNioSessionImpl implements Gr
this.readBuf = readBuf;
}
- outboundMessagesQueueSizeMetric = mreg == null ? null :
mreg.longAdderMetric(
- OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_NAME,
- OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_DESC
- );
+ this.outboundMessagesQueueSizeMetric = outboundMessagesQueueSizeMetric;
+
+ this.maxMessagesQueueSizeMetric = maxMessagesQueueSizeMetric;
- maxMessagesQueueSizeMetric = mreg == null ? null : mreg.maxValueMetric(
- MAX_MESSAGES_QUEUE_SIZE_METRIC_NAME,
- MAX_MESSAGES_QUEUE_SIZE_METRIC_DESC,
- 60_000,
- 5
- );
+ this.tracing = tracing;
}
/** {@inheritDoc} */
@@ -323,17 +319,17 @@ public class GridSelectorNioSessionImpl extends
GridNioSessionImpl implements Gr
boolean res = queue.offerFirst(writeFut);
- MTC.span().addLog(() -> "Added to system queue - " +
traceName(writeFut.message()));
+ MTC.span().addLog(() -> "Added to system queue - " +
tracing.traceName((Message)writeFut.message()));
assert res : "Future was not added to queue";
if (outboundMessagesQueueSizeMetric != null)
- outboundMessagesQueueSizeMetric.increment();
+ outboundMessagesQueueSizeMetric.accept(1);
if (maxMessagesQueueSizeMetric != null) {
int queueSize = queue.sizex();
- maxMessagesQueueSizeMetric.update(queueSize);
+ maxMessagesQueueSizeMetric.accept(queueSize);
return queueSize;
}
@@ -361,17 +357,17 @@ public class GridSelectorNioSessionImpl extends
GridNioSessionImpl implements Gr
boolean res = queue.offer(writeFut);
- MTC.span().addLog(() -> "Added to queue - " +
traceName(writeFut.message()));
+ MTC.span().addLog(() -> "Added to queue - " +
tracing.traceName((Message)writeFut.message()));
assert res : "Future was not added to queue";
if (outboundMessagesQueueSizeMetric != null)
- outboundMessagesQueueSizeMetric.increment();
+ outboundMessagesQueueSizeMetric.accept(1);
if (maxMessagesQueueSizeMetric != null) {
int queueSize = queue.sizex();
- maxMessagesQueueSizeMetric.update(queueSize);
+ maxMessagesQueueSizeMetric.accept(queueSize);
return queueSize;
}
@@ -390,10 +386,10 @@ public class GridSelectorNioSessionImpl extends
GridNioSessionImpl implements Gr
assert add;
if (outboundMessagesQueueSizeMetric != null)
- outboundMessagesQueueSizeMetric.add(futs.size());
+ outboundMessagesQueueSizeMetric.accept(futs.size());
if (maxMessagesQueueSizeMetric != null)
- maxMessagesQueueSizeMetric.update(futs.size());
+ maxMessagesQueueSizeMetric.accept(futs.size());
}
/**
@@ -404,7 +400,7 @@ public class GridSelectorNioSessionImpl extends
GridNioSessionImpl implements Gr
if (last != null) {
if (outboundMessagesQueueSizeMetric != null)
- outboundMessagesQueueSizeMetric.decrement();
+ outboundMessagesQueueSizeMetric.accept(-1);
if (sem != null && !last.messageThread())
sem.release();
@@ -439,7 +435,7 @@ public class GridSelectorNioSessionImpl extends
GridNioSessionImpl implements Gr
boolean rmv = queue.removeLastOccurrence(fut);
if (rmv && outboundMessagesQueueSizeMetric != null)
- outboundMessagesQueueSizeMetric.decrement();
+ outboundMessagesQueueSizeMetric.accept(-1);
return rmv;
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
index ab9f2994699..c6b5981471d 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.util.nio.ssl;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
+import java.util.function.LongConsumer;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLException;
@@ -27,9 +28,6 @@ import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
-import org.apache.ignite.internal.processors.metric.impl.HistogramMetricImpl;
-import org.apache.ignite.internal.processors.metric.impl.IntMetricImpl;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.nio.GridNioException;
@@ -86,10 +84,10 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
@Nullable private Exception onSessionOpenedException;
/** Metric that indicates sessions count that were rejected due to SSL
errors. */
- @Nullable private final IntMetricImpl rejectedSesCnt;
+ @Nullable private final Runnable rejectedSesCnt;
/** Histogram that provides distribution of SSL handshake duration. */
- @Nullable private final HistogramMetricImpl handshakeDuration;
+ @Nullable private final LongConsumer handshakeDuration;
/**
* Creates SSL filter.
@@ -98,14 +96,16 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
* @param directBuf Direct buffer flag.
* @param order Byte order.
* @param log Logger to use.
- * @param mreg Optional metric registry.
+ * @param handshakeDuration Records SSL handshake duration (ms), or {@code
null} if metrics disabled.
+ * @param rejectedSesCnt Increments the rejected-sessions counter, or
{@code null} if metrics disabled.
*/
public GridNioSslFilter(
SSLContext sslCtx,
boolean directBuf,
ByteOrder order,
IgniteLogger log,
- @Nullable MetricRegistryImpl mreg
+ @Nullable LongConsumer handshakeDuration,
+ @Nullable Runnable rejectedSesCnt
) {
super("SSL filter");
@@ -113,17 +113,8 @@ public class GridNioSslFilter extends GridNioFilterAdapter
{
this.sslCtx = sslCtx;
this.directBuf = directBuf;
this.order = order;
-
- handshakeDuration = mreg == null ? null : mreg.histogram(
- SSL_HANDSHAKE_DURATION_HISTOGRAM_METRIC_NAME,
- new long[] {250, 500, 1000},
- "SSL handshake duration in milliseconds."
- );
-
- rejectedSesCnt = mreg == null ? null : mreg.intMetric(
- SSL_REJECTED_SESSIONS_CNT_METRIC_NAME,
- "TCP sessions count that were rejected due to SSL errors."
- );
+ this.handshakeDuration = handshakeDuration;
+ this.rejectedSesCnt = rejectedSesCnt;
}
/**
@@ -253,7 +244,7 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
long startTime = System.nanoTime();
- fut.listen(() ->
handshakeDuration.value(U.nanosToMillis(System.nanoTime() - startTime)));
+ fut.listen(() ->
handshakeDuration.accept(U.nanosToMillis(System.nanoTime() - startTime)));
}
hnd.handshake();
@@ -275,7 +266,7 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
if (fut != null) {
if (rejectedSesCnt != null)
- rejectedSesCnt.increment();
+ rejectedSesCnt.run();
fut.onDone(new IgniteCheckedException("SSL handshake failed
(connection closed).", onSessionOpenedException));
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationMetricsListener.java
b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationMetricsListener.java
index 2fb542ceb8b..4bf17f1a969 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationMetricsListener.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationMetricsListener.java
@@ -44,6 +44,8 @@ import org.apache.ignite.spi.metric.ReadOnlyMetricRegistry;
import static java.util.stream.Collectors.toMap;
import static
org.apache.ignite.internal.processors.metric.impl.MetricUtils.SEPARATOR;
import static
org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName;
+import static
org.apache.ignite.internal.util.nio.GridNioServer.OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_DESC;
+import static
org.apache.ignite.internal.util.nio.GridNioServer.OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_NAME;
import static
org.apache.ignite.internal.util.nio.GridNioServer.RECEIVED_BYTES_METRIC_DESC;
import static
org.apache.ignite.internal.util.nio.GridNioServer.RECEIVED_BYTES_METRIC_NAME;
import static
org.apache.ignite.internal.util.nio.GridNioServer.SENT_BYTES_METRIC_DESC;
@@ -105,6 +107,9 @@ public class TcpCommunicationMetricsListener {
/** Received messages count metric. */
private final LongAdderMetric rcvdMsgsMetric;
+ /** Outbound messages queue size metric. */
+ private final LongAdderMetric outboundMessagesQueueSizeMetric;
+
/** Counters of sent and received messages by direct type. */
private final IntMap<IgniteBiTuple<LongAdderMetric, LongAdderMetric>>
msgCntrsByType;
@@ -146,6 +151,11 @@ public class TcpCommunicationMetricsListener {
sentMsgsMetric = mreg.longAdderMetric(SENT_MESSAGES_METRIC_NAME,
SENT_MESSAGES_METRIC_DESC);
rcvdMsgsMetric = mreg.longAdderMetric(RECEIVED_MESSAGES_METRIC_NAME,
RECEIVED_MESSAGES_METRIC_DESC);
+ outboundMessagesQueueSizeMetric = mreg.longAdderMetric(
+ OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_NAME,
+ OUTBOUND_MESSAGES_QUEUE_SIZE_METRIC_DESC
+ );
+
spiCtx.addMetricRegistryCreationListener(mreg -> {
// Metrics for the specific nodes or other communication metrics.
if (!TcpCommunicationSpi.isCommunicationMetrics(mreg.name()))
@@ -276,6 +286,15 @@ public class TcpCommunicationMetricsListener {
return rcvdBytesMetric.value();
}
+ /**
+ * Gets outbound messages queue size.
+ *
+ * @return Outbound messages queue size.
+ */
+ public int outboundMessagesQueueSize() {
+ return (int)outboundMessagesQueueSizeMetric.value();
+ }
+
/**
* Gets received messages counts (grouped by type).
*
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
index f9ab18727ea..4bcae5a26a2 100755
---
a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
@@ -411,9 +411,10 @@ public class TcpCommunicationSpi extends
TcpCommunicationConfigInitializer {
/** {@inheritDoc} */
@Override public int getOutboundMessagesQueueSize() {
- GridNioServer<Message> srv = nioSrvWrapper.nio();
+ if (metricsLsnr == null)
+ return 0;
- return srv != null ? srv.outboundMessagesQueueSize() : 0;
+ return metricsLsnr.outboundMessagesQueueSize();
}
/** {@inheritDoc} */
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/GridNioServerWrapper.java
b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/GridNioServerWrapper.java
index df7935c6c2f..eb90f002a24 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/GridNioServerWrapper.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/GridNioServerWrapper.java
@@ -56,6 +56,7 @@ import org.apache.ignite.internal.direct.DirectMessageWriter;
import org.apache.ignite.internal.managers.GridManager;
import org.apache.ignite.internal.managers.tracing.GridTracingManager;
import org.apache.ignite.internal.processors.metric.GridMetricManager;
+import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
import org.apache.ignite.internal.processors.tracing.Tracing;
import org.apache.ignite.internal.util.GridConcurrentFactory;
import org.apache.ignite.internal.util.IgniteExceptionRegistry;
@@ -909,7 +910,7 @@ public class GridNioServerWrapper {
filters.add(new GridConnectionBytesVerifyFilter(log));
if (stateProvider.isSslEnabled()) {
- GridNioSslFilter sslFilter = new GridNioSslFilter(
+ GridNioSslFilter sslFilter = U.sslFilter(
igniteCfg.getSslContextFactory().create(),
true,
ByteOrder.LITTLE_ENDIAN,
@@ -924,6 +925,11 @@ public class GridNioServerWrapper {
filters.add(sslFilter);
}
+ GridNioFilter[] filtersArr = filters.toArray(new
GridNioFilter[filters.size()]);
+
+ MetricRegistryImpl mreg = metricMgr != null ?
+ metricMgr.registry(COMMUNICATION_METRICS_GROUP_NAME) :
null;
+
GridNioServer.Builder<Message> builder =
GridNioServer.<Message>builder()
.address(cfg.localHost())
.port(port)
@@ -941,7 +947,7 @@ public class GridNioServerWrapper {
.directMode(true)
.writeTimeout(cfg.socketWriteTimeout())
.selectorSpins(cfg.selectorSpins())
- .filters(filters.toArray(new
GridNioFilter[filters.size()]))
+ .filters(filtersArr)
.writerFactory(writerFactory)
.skipRecoveryPredicate(skipRecoveryPred)
.messageQueueSizeListener(queueSizeMonitor)
@@ -949,13 +955,17 @@ public class GridNioServerWrapper {
.readWriteSelectorsAssign(cfg.usePairedConnections())
.messageFactory(msgFactory);
- if (metricMgr != null) {
- builder.workerListener(workersRegistry)
-
.metricRegistry(metricMgr.registry(COMMUNICATION_METRICS_GROUP_NAME));
+ if (mreg != null) {
+ builder.workerListener(workersRegistry);
+
+ U.setNioServerMetrics(builder, mreg);
}
GridNioServer<Message> srvr = builder.build();
+ if (mreg != null)
+ U.registerNioServerMetrics(srvr, filtersArr, mreg);
+
cfg.boundTcpPort(port);
// Ack Port the TCP server was bound to.
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSslSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSslSelfTest.java
index 79bc9f0ef82..9b2199f213e 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSslSelfTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioSslSelfTest.java
@@ -71,7 +71,7 @@ public class GridNioSslSelfTest extends GridNioSelfTest {
.sendQueueLimit(0)
.filters(
new GridNioCodecFilter(parser, log, false),
- new GridNioSslFilter(sslCtx, true, ByteOrder.nativeOrder(),
log, null));
+ new GridNioSslFilter(sslCtx, true, ByteOrder.nativeOrder(),
log, null, null));
}
/** {@inheritDoc} */
