diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java b/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java index 4ea6dd22..a4f85cc5 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java +++ b/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java @@ -18,7 +18,6 @@ import io.micrometer.core.instrument.Counter; import io.micrometer.core.instrument.Gauge; -import io.micrometer.core.instrument.LongTaskTimer; import io.micrometer.core.instrument.MeterRegistry; import io.netty.util.Timer; import org.slf4j.Logger; @@ -30,6 +29,7 @@ import io.tarantool.core.connection.ConnectionFactory; import io.tarantool.core.protocol.Handlers; import io.tarantool.core.protocol.IProtoResponse; +import io.tarantool.pool.PoolEntry.PoolEntryMetrics; import io.tarantool.pool.exceptions.PoolClosedException; import io.tarantool.pool.exceptions.PoolException; @@ -127,19 +127,8 @@ public class IProtoClientPoolImpl implements IProtoClientPool { /** Time in milliseconds after reconnect will be run. */ private long reconnectAfter; - /** Count of successful connection requests. */ - private Counter requestSuccess; - - /** Count of failed connection requests. */ - private Counter requestErrors; - - /** Count of connection requests for wrong tag or index. */ - private Counter invalidRequests; - - /** - * Count of connection requests for connections which were invalidated or in reconnecting state. - */ - private Counter lockedConnectionRequests; + /** Metrics of {@link IProtoClientPoolImpl}. */ + private final PoolMetrics metrics; /** * Handler for accepting ignored packets and processing them somehow. It is an instance of {@link @@ -354,7 +343,7 @@ public IProtoClientPoolImpl( this.useTupleExtension = useTupleExtension; this.poolEventListener = poolEventListener; - initMetrics(); + this.metrics = initMetrics(); } @Override @@ -554,7 +543,8 @@ private void expandGroup(List connects, InstanceConnectionGroup group ignoredPacketsHandler, handlers, useTupleExtension, - poolEventListener)); + poolEventListener, + this.metrics != null ? this.metrics.poolEntryMetrics : null)); } } @@ -601,62 +591,12 @@ private void initStringBuilder() { } /** Metrics initialization. */ - private void initMetrics() { + private PoolMetrics initMetrics() { if (this.metricsRegistry == null) { - return; + return null; } - Gauge.builder("pool.size", this.connectionStatistics::total) - .description("total number of connections at the moment") - .register(metricsRegistry); - Gauge.builder("pool.available", this.connectionStatistics::availableConnections) - .description("total number of connections available at the moment") - .register(metricsRegistry); - Gauge.builder("pool.unavailable", this.connectionStatistics::unavailable) - .description("total number of connection unavailable at the moment") - .register(metricsRegistry); - Gauge.builder("pool.invalidated", this.connectionStatistics::invalidatedConnections) - .description("number of connections invalidated at the moment") - .register(metricsRegistry); - Gauge.builder("pool.reconnecting", this.connectionStatistics::reconnecting) - .description("number of connections reconnecting at the moment") - .register(metricsRegistry); - - Counter.builder("pool.connect.success") - .description("count of succeeded connects") - .register(metricsRegistry); - Counter.builder("pool.connect.errors") - .description("count of connect failures") - .register(metricsRegistry); - LongTaskTimer.builder("pool.connect.time") - .description("time of connects") - .register(metricsRegistry); - - requestSuccess = - Counter.builder("pool.request.success") - .description("Count of successful connections requests from pool") - .register(metricsRegistry); - requestErrors = - Counter.builder("pool.request.errors") - .description("Count of failed connection requests from pool") - .register(metricsRegistry); - invalidRequests = - Counter.builder("pool.request.misses") - .description("Count of connection requests with wrong tag or index") - .register(metricsRegistry); - lockedConnectionRequests = - Counter.builder("pool.request.unavail") - .description("Count of connection request finished with locked connect") - .register(metricsRegistry); - - Counter.builder("pool.heartbeat.success") - .description("Count of successful heartbeat ping requests") - .register(metricsRegistry); - Counter.builder("pool.heartbeat.errors") - .description("Count of failed heartbeat ping requests") - .register(metricsRegistry); - LongTaskTimer.builder("pool.heartbeat.time") - .description("Time of ping request") - .register(metricsRegistry); + + return new PoolMetrics(metricsRegistry, this.connectionStatistics); } /** @@ -668,33 +608,29 @@ private void initMetrics() { * counter for errors. */ private void incPoolRequestCounters(Object r, Throwable e) { - if (requestSuccess == null || requestErrors == null) { + if (this.metrics == null) { return; } if (e == null) { - requestSuccess.increment(); + this.metrics.poolRequestSuccess.increment(); } else { - requestErrors.increment(); + this.metrics.poolRequestErrors.increment(); } } /** Increment misses counter when wrong tag or index requested. */ private void incPoolInvalidRequests() { - if (invalidRequests == null) { - return; + if (this.metrics != null) { + this.metrics.poolRequestMisses.increment(); } - - invalidRequests.increment(); } /** Increment counter when invalidated or closed connection was requested. */ private void incPoolLockedConnectionRequests() { - if (lockedConnectionRequests == null) { - return; + if (this.metrics != null) { + this.metrics.poolRequestUnavailable.increment(); } - - lockedConnectionRequests.increment(); } @Override @@ -854,4 +790,112 @@ public int total() { return this.total; } } + + static final class PoolMetrics { + + static final String POOL_SIZE_METRIC = "pool.size"; + static final String POOL_SIZE_METRIC_DESC = "Total number of connections at the moment"; + private final Gauge poolSize; + + static final String POOL_AVAILABLE_METRIC = "pool.available"; + static final String POOL_AVAILABLE_METRIC_DESC = + "Total number of connections available at the moment"; + private final Gauge poolAvailable; + + static final String POOL_UNAVAILABLE_METRIC = "pool.unavailable"; + static final String POOL_UNAVAILABLE_METRIC_DESC = + "Total number of connection unavailable at the moment"; + private final Gauge poolUnavailable; + + static final String POOL_INVALIDATED_METRIC = "pool.invalidated"; + static final String POOL_INVALIDATED_METRIC_DESC = + "Total number of invalidated connections at the moment"; + private final Gauge poolInvalidated; + + static final String POOL_RECONNECTING_METRIC = "pool.reconnecting"; + static final String POOL_RECONNECTING_METRIC_DESC = + "Total number of reconnecting connections at the moment"; + private final Gauge poolReconnecting; + + static final String POOL_REQUEST_SUCCESS_METRIC = "pool.request.success"; + static final String POOL_REQUEST_SUCCESS_METRIC_DESC = + "Total count of successful connections requests from pool"; + + /** Count of successful connection requests. */ + private final Counter poolRequestSuccess; + + static final String POOL_REQUEST_ERRORS_METRIC = "pool.request.errors"; + static final String POOL_REQUEST_ERRORS_METRIC_DESC = + "Total count of failed connection requests from pool"; + + /** Count of failed connection requests. */ + private final Counter poolRequestErrors; + + static final String POOL_REQUEST_MISSES_METRIC = "pool.request.misses"; + static final String POOL_REQUEST_MISSES_METRIC_DESC = + "Total count of connection requests with wrong tag or index"; + + /** Count of connection requests for wrong tag or index. */ + private final Counter poolRequestMisses; + + static final String POOL_REQUEST_UNAVAILABLE_METRIC = "pool.request.unavail"; + static final String POOL_REQUEST_UNAVAILABLE_METRIC_DESC = + "Total count of connection request finished with locked connect"; + + /** + * Count of connection requests for connections which were invalidated or in reconnecting state. + */ + private final Counter poolRequestUnavailable; + + private final PoolEntryMetrics poolEntryMetrics; + + PoolMetrics(MeterRegistry registry, ConnectionStatistics statistics) { + this.poolSize = + Gauge.builder(POOL_SIZE_METRIC, statistics::total) + .description(POOL_SIZE_METRIC_DESC) + .register(registry); + + this.poolAvailable = + Gauge.builder(POOL_AVAILABLE_METRIC, statistics::availableConnections) + .description(POOL_AVAILABLE_METRIC_DESC) + .register(registry); + + this.poolUnavailable = + Gauge.builder(POOL_UNAVAILABLE_METRIC, statistics::unavailable) + .description(POOL_UNAVAILABLE_METRIC_DESC) + .register(registry); + + this.poolInvalidated = + Gauge.builder(POOL_INVALIDATED_METRIC, statistics::invalidatedConnections) + .description(POOL_INVALIDATED_METRIC_DESC) + .register(registry); + + this.poolReconnecting = + Gauge.builder(POOL_RECONNECTING_METRIC, statistics::reconnecting) + .description(POOL_RECONNECTING_METRIC_DESC) + .register(registry); + + this.poolRequestSuccess = + Counter.builder(POOL_REQUEST_SUCCESS_METRIC) + .description(POOL_REQUEST_SUCCESS_METRIC_DESC) + .register(registry); + + this.poolRequestErrors = + Counter.builder(POOL_REQUEST_ERRORS_METRIC) + .description(POOL_REQUEST_ERRORS_METRIC_DESC) + .register(registry); + + this.poolRequestMisses = + Counter.builder(POOL_REQUEST_MISSES_METRIC) + .description(POOL_REQUEST_MISSES_METRIC_DESC) + .register(registry); + + this.poolRequestUnavailable = + Counter.builder(POOL_REQUEST_UNAVAILABLE_METRIC) + .description(POOL_REQUEST_UNAVAILABLE_METRIC_DESC) + .register(registry); + + this.poolEntryMetrics = new PoolEntryMetrics(registry); + } + } } diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java index dfc4d175..ef8b34ad 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java +++ b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java @@ -8,7 +8,6 @@ import java.util.ArrayDeque; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; -import java.util.function.BiFunction; import java.util.function.Consumer; import io.micrometer.core.instrument.Counter; @@ -94,28 +93,12 @@ final class PoolEntry { /** Flag for graceful shutdown handling. */ private final boolean gracefulShutdown; - /** Death threshold from {@link io.tarantool.pool.HeartbeatOpts}. */ - private final int deathThreshold; - - /** Count of ping failed, computed from window size and percent of failures. */ - private final int failedPingsThreshold; - /** Index of entry within group. */ private final int index; - /** Window size from {@link io.tarantool.pool.HeartbeatOpts}. */ - private final int windowSize; - /** Time in milliseconds after that reconnect task will be executed. */ private final long reconnectAfter; - /** - * Function that send IProto packet that will be sent to tarantool. The response of this packet - * will be considered as pong result. - */ - private final BiFunction> - pingFunction; - /** Parameter that will say tarantool not to use Tuple Extension. */ private final boolean useTupleExtension; @@ -169,23 +152,8 @@ final class PoolEntry { */ private long connectTimeout; - /** Count of successul connect attempts. */ - private Counter connectSuccess; - - /** Count of failed connect attempts. */ - private Counter connectErrors; - - /** Time of connect. */ - private LongTaskTimer connectTime; - - /** Count of successful heartbeat ping requests. */ - private Counter heartbeatSuccess; - - /** Count of failed heartbeat ping requests. */ - private Counter heartbeatErrors; - - /** Time of ping request. */ - private LongTaskTimer heartbeatTime; + /** Metrics of {@link PoolEntry}. */ + private final PoolEntryMetrics metrics; /** * Constructor for PoolEntry. @@ -223,7 +191,9 @@ public PoolEntry( TripleConsumer ignoredPacketsHandler, Handlers handlers, boolean useTupleExtension, - PoolEventListener poolEventListener) { + PoolEventListener poolEventListener, + PoolEntryMetrics metrics) { + this.metrics = metrics; this.metricsRegistry = registry; this.handlers = handlers; this.client = @@ -238,23 +208,14 @@ public PoolEntry( // heartbeat related this.isHeartbeatStarted = false; this.lastHeartbeatEvent = HeartbeatEvent.KILL; - if (heartbeatOpts == null) { - this.heartbeatOpts = null; - this.heartbeatPingOpts = null; - this.deathThreshold = 0; - this.failedPingsThreshold = 0; - this.windowSize = 0; - this.window = null; - this.pingFunction = null; - } else { - this.heartbeatOpts = heartbeatOpts; + this.heartbeatOpts = heartbeatOpts; + if (heartbeatOpts != null) { + this.window = new ArrayDeque<>(heartbeatOpts.getWindowSize()); this.heartbeatPingOpts = IProtoRequestOpts.empty().withRequestTimeout(heartbeatOpts.getPingInterval()); - this.deathThreshold = heartbeatOpts.getDeathThreshold(); - this.failedPingsThreshold = heartbeatOpts.getInvalidationThreshold(); - this.windowSize = heartbeatOpts.getWindowSize(); - this.window = new ArrayDeque<>(this.windowSize); - this.pingFunction = heartbeatOpts.getPingFunction(); + } else { + this.window = null; + this.heartbeatPingOpts = null; } this.connectTimeout = connectTimeout; this.gracefulShutdown = gracefulShutdown; @@ -273,8 +234,6 @@ public PoolEntry( if (ignoredPacketsHandler != null) { client.onIgnoredPacket(packet -> ignoredPacketsHandler.accept(this.tag, this.index, packet)); } - - initMetrics(); } /** @@ -368,9 +327,9 @@ public void startHeartbeat() { "heartbeat: start for {}/{}: failures = {}, interval = {}, death pings = {}", tag, index, - failedPingsThreshold, + heartbeatOpts.getInvalidationThreshold(), heartbeatOpts.getPingInterval(), - deathThreshold); + heartbeatOpts.getDeathThreshold()); isHeartbeatStarted = true; window.clear(); // heartbeat is being started immediately after first check so it is @@ -400,7 +359,8 @@ public void stopHeartbeat() { */ private CompletableFuture internalConnect() { log.info("connect {}/{}", tag, index); - LongTaskTimer.Sample timer = startTimer(connectTime); + final LongTaskTimer.Sample timer = + this.metrics != null ? this.metrics.poolConnectTime.start() : null; CompletableFuture future = client.connect(group.getAddress(), connectTimeout, gracefulShutdown); String user = group.getUser(); @@ -408,7 +368,7 @@ private CompletableFuture internalConnect() { future .thenCompose( greeting -> { - stopTimer(timer); + PoolEntryMetrics.stopTimer(timer); if (user != null) { return client.authorize(user, group.getPassword(), group.getAuthType()); } @@ -429,13 +389,10 @@ private CompletableFuture internalConnect() { * @param exc exception occurred during connection process */ private void onConnectComplete(Object r, Throwable exc) { - if (metricsRegistry != null) { - if (exc != null) { - connectErrors.increment(); - } else { - connectSuccess.increment(); - } + if (this.metrics != null) { + this.metrics.incrementPoolConnectMetrics(exc); } + if (exc != null) { handleConnectError(r, exc); return; @@ -484,12 +441,14 @@ private void connectAfter() { */ private void ping(Timeout handler) { lastPingTs = System.currentTimeMillis(); - LongTaskTimer.Sample timer = startTimer(heartbeatTime); - pingFunction - .apply(client, heartbeatPingOpts) + final LongTaskTimer.Sample timer = + this.metrics != null ? this.metrics.heartbeatTime.start() : null; + heartbeatOpts + .getPingFunction() + .apply(client, this.heartbeatPingOpts) .whenComplete( (r, exc) -> { - stopTimer(timer); + PoolEntryMetrics.stopTimer(timer); pong(r, exc); }); } @@ -528,20 +487,20 @@ private void pong(IProtoResponse result, Throwable exc) { && !result.getBodyArrayValue(IPROTO_DATA).get(0).equals(ImmutableBooleanValueImpl.TRUE)) { failure = 1; } - incHeartbeatCounters(failure); + PoolEntryMetrics.incrementHeartbeatMetrics(this.metrics, failure); window.addFirst(failure); currentFailedPings += failure; - if (window.size() > windowSize) { + if (window.size() > heartbeatOpts.getWindowSize()) { currentFailedPings -= window.removeLast(); - } else if (window.size() < windowSize) { + } else if (window.size() < heartbeatOpts.getWindowSize()) { return; } - if (currentFailedPings >= failedPingsThreshold) { + if (currentFailedPings >= heartbeatOpts.getInvalidationThreshold()) { fire(HeartbeatEvent.INVALIDATE); currentDeathPings++; - if (currentDeathPings > deathThreshold) { + if (currentDeathPings > heartbeatOpts.getDeathThreshold()) { fire(HeartbeatEvent.KILL); } return; @@ -580,70 +539,6 @@ private void fire(HeartbeatEvent event) { } } - /** - * Metrics initializations. - * - *

When registry is passed to {@link PoolEntry} constructor, metrics for heartbeats and connect - * are created and registered in this registry. - */ - private void initMetrics() { - if (metricsRegistry == null) { - return; - } - - heartbeatSuccess = metricsRegistry.get("pool.heartbeat.success").counter(); - heartbeatErrors = metricsRegistry.get("pool.heartbeat.errors").counter(); - heartbeatTime = metricsRegistry.get("pool.heartbeat.time").longTaskTimer(); - - connectSuccess = metricsRegistry.get("pool.connect.success").counter(); - connectErrors = metricsRegistry.get("pool.connect.errors").counter(); - connectTime = metricsRegistry.get("pool.connect.time").longTaskTimer(); - } - - /** - * Creates timer span for time measurement. - * - * @param timer an instance of {@code LongTaskTimer} - * @return instance of {@code LongTaskTimer.Sample} - */ - private LongTaskTimer.Sample startTimer(LongTaskTimer timer) { - if (metricsRegistry == null) { - return null; - } - - return timer.start(); - } - - /** - * If timer span is not null, stops time measurement. - * - * @param timer an instance of {@code LongTaskTimer.Sample} - */ - private void stopTimer(LongTaskTimer.Sample timer) { - if (timer == null) { - return; - } - - timer.stop(); - } - - /** - * Increments heartbeat related metrics. - * - * @param fail if ping is failed then this value is 1, otherwise this value is 0. - */ - private void incHeartbeatCounters(int fail) { - if (metricsRegistry == null || heartbeatSuccess == null || heartbeatErrors == null) { - return; - } - - if (fail != 0) { - heartbeatErrors.increment(); - } else { - heartbeatSuccess.increment(); - } - } - /** Stops reconnecting task if it is active. */ private void stopReconnectTask() { if (reconnectTask != null) { @@ -663,4 +558,119 @@ private void emit(Consumer action) { action.accept(poolEventListener); } } + + /** {@link PoolEntry} metrics. */ + static class PoolEntryMetrics { + + static final String POOL_HEARTBEAT_SUCCESS_METRIC = "pool.heartbeat.success"; + static final String POOL_HEARTBEAT_SUCCESS_METRIC_DESC = + "Total count of successful heartbeat ping requests"; + + /** Count of successful heartbeat ping requests. */ + private final Counter heartbeatSuccess; + + static final String POOL_HEARTBEAT_ERRORS_METRIC = "pool.heartbeat.errors"; + static final String POOL_HEARTBEAT_ERRORS_METRIC_DESC = + "Total count of failed heartbeat ping requests"; + + /** Count of failed heartbeat ping requests. */ + private final Counter heartbeatErrors; + + static final String POOL_HEARTBEAT_TIME_METRIC = "pool.heartbeat.time"; + static final String POOL_HEARTBEAT_TIME_METRIC_DESC = "Time of ping request"; + + /** Time of ping request. */ + private final LongTaskTimer heartbeatTime; + + static final String POOL_CONNECT_SUCCESS_METRIC = "pool.connect.success"; + static final String POOL_CONNECT_SUCCESS_METRIC_DESC = "Total count of succeeded connects"; + + /** Count of successful connect attempts. */ + private final Counter poolConnectSuccess; + + static final String POOL_CONNECT_ERRORS_METRIC = "pool.connect.errors"; + static final String POOL_CONNECT_ERRORS_METRIC_DESC = "Total count of connect failures"; + + /** Count of failed connect attempts. */ + private final Counter poolConnectErrors; + + static final String POOL_CONNECT_TIME_METRIC = "pool.connect.time"; + static final String POOL_CONNECT_TIME_METRIC_DESC = "Total time of connects"; + + /** Time of connect. */ + private final LongTaskTimer poolConnectTime; + + public PoolEntryMetrics(MeterRegistry registry) { + + this.heartbeatSuccess = + Counter.builder(POOL_HEARTBEAT_SUCCESS_METRIC) + .description(POOL_HEARTBEAT_SUCCESS_METRIC_DESC) + .register(registry); + + this.heartbeatErrors = + Counter.builder(POOL_HEARTBEAT_ERRORS_METRIC) + .description(POOL_HEARTBEAT_ERRORS_METRIC_DESC) + .register(registry); + + this.heartbeatTime = + LongTaskTimer.builder(POOL_HEARTBEAT_TIME_METRIC) + .description(POOL_HEARTBEAT_TIME_METRIC_DESC) + .register(registry); + + this.poolConnectSuccess = + Counter.builder(POOL_CONNECT_SUCCESS_METRIC) + .description(POOL_CONNECT_SUCCESS_METRIC_DESC) + .register(registry); + + this.poolConnectErrors = + Counter.builder(POOL_CONNECT_ERRORS_METRIC) + .description(POOL_CONNECT_ERRORS_METRIC_DESC) + .register(registry); + + this.poolConnectTime = + LongTaskTimer.builder(POOL_CONNECT_TIME_METRIC) + .description(POOL_CONNECT_TIME_METRIC_DESC) + .register(registry); + } + + /** + * Increment {@link #poolConnectSuccess} if exception is null and increment {@link + * #poolConnectErrors} if exception is not null. + */ + private void incrementPoolConnectMetrics(Throwable exception) { + if (exception != null) { + this.poolConnectErrors.increment(); + } else { + this.poolConnectSuccess.increment(); + } + } + + /** + * Increments heartbeat related metrics. + * + * @param fail if ping is failed then this value is 1, otherwise this value is 0. + */ + private static void incrementHeartbeatMetrics(PoolEntryMetrics metrics, int fail) { + if (metrics == null) { + return; + } + + if (fail != 0) { + metrics.heartbeatErrors.increment(); + } else { + metrics.heartbeatSuccess.increment(); + } + } + + /** + * If timer span is not null, stops time measurement. + * + * @param timer an instance of {@code LongTaskTimer.Sample} + */ + private static void stopTimer(LongTaskTimer.Sample timer) { + if (timer != null) { + timer.stop(); + } + } + } }