Add WebSocket client metrics by LivingLikeKrillin · Pull Request #4118 · reactor/reactor-netty

LivingLikeKrillin · 2026-02-28T15:12:46Z

Summary

Add metrics support for WebSocket clients, including handshake time,
data received/sent, connection duration, and error counting.

Add constants, meter/observation documentation, and recorder interfaces
with Micrometer-based implementation
Add channel handlers that replace the HTTP metrics handler on WebSocket
upgrade and track connection duration via handler lifecycle
Exclude control frames (Close, Ping, Pong) from data metrics

Test plan

Handshake time timer
Data received time and distribution summary
Data sent time and distribution summary
Connection duration timer
Multiple connections aggregation
URI tag value normalization function
Handshake failure scenario
HTTP/1.1 and H2 protocol combinations

Related to #3820

violetagg · 2026-03-09T08:47:07Z

@LivingLikeKrillin On most classes you have @author raccoonback is that the correct author? Also the copyright is 2025 while it should be 2026, also change the @since 1.3.2 to 1.3.5

Add WEBSOCKET_CLIENT_PREFIX, HANDSHAKE_TIME, and CONNECTION_DURATION constants, meter/observation documentation enums, and recorder interfaces with Micrometer-based implementation. Related to reactor#3820 Signed-off-by: LivingLikeKrillin <143606756+LivingLikeKrillin@users.noreply.github.com>

Replace the HTTP metrics handler with a WebSocket-aware handler on upgrade. Connection duration is recorded when the handler is removed from the pipeline. Control frames (Close, Ping, Pong) are excluded from data metrics. Related to reactor#3820 Signed-off-by: LivingLikeKrillin <143606756+LivingLikeKrillin@users.noreply.github.com>

Test handshake time, data received/sent, connection duration, handshake failure, and protocol combinations (HTTP/1.1 and H2). Related to reactor#3820 Signed-off-by: LivingLikeKrillin <143606756+LivingLikeKrillin@users.noreply.github.com>

Signed-off-by: LivingLikeKrillin <143606756+LivingLikeKrillin@users.noreply.github.com>

violetagg

Thanks for the PR!
I started with the review. Can you please resolve the issues that I've identified till now?

...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

- Compute path and contextView eagerly in swapMetricsHandler() instead of lazy initialization via initMetrics() - Change extractProcessedDataFromBuffer parameter type to WebSocketFrame - Use GET for HTTP/1.1 and CONNECT for HTTP/2 via wsHttpMethod() - Restore missing @author raccoonback in WebsocketClientOperations

reactor-netty-core/src/main/java/reactor/netty/Metrics.java

...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

reactor-netty-http/src/main/java/reactor/netty/http/client/WebsocketClientOperations.java

- Remove @nullable from path and contextView fields and parameters, since they are eagerly initialized in swapMetricsHandler() - Remove dead code: null guards, ternary fallbacks, and else branches in ContextAwareWebSocketClientMetricsHandler that are unreachable after the @nullable removal - Remove unused copy constructors from all handler classes - Move resolvePath() from AbstractWebSocketClientMetricsHandler to WebsocketClientOperations where it is actually called - Remove unused URI parameter from swapMetricsHandler() - Remove unused status parameter from recordHandshakeFailure() and hardcode "ERROR" since no other value is used - Rename metric namespace from reactor.netty.http.client.websocket to reactor.netty.websocket.client

Accumulate data across fragmented WebSocket frames and record metrics only when isFinalFragment() is true, instead of recording per individual frame. - Set dataSentTime/dataReceivedTime only on the first fragment - Accumulate dataSent/dataReceived across continuation frames - Record and reset counters on final fragment - Use synchronous recording in write() since reactor-netty's SendManyInner does not support ChannelPromise.addListener()

- testWebSocketFragmentedDataSentMetrics: verify that sending 3 fragments records DATA_SENT once with accumulated total - testWebSocketFragmentedDataReceivedMetrics: verify that receiving 2 fragments records DATA_RECEIVED once with accumulated total - Add /ws-fragment server route that sends fragmented frames

...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

+
+	protected abstract WebSocketClientMetricsRecorder recorder();
+
+	protected void recordConnectionClosed(ChannelHandlerContext ctx) {


...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

+		}
+	}
+
+	protected void recordException(ChannelHandlerContext ctx) {


...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

+		}
+	}
+
+	protected void recordRead(io.netty.channel.Channel channel, SocketAddress address) {


...http/src/main/java/reactor/netty/http/client/ContextAwareWebSocketClientMetricsRecorder.java

reactor-netty-http/src/test/java/reactor/netty/http/WebSocketClientMetricsHandlerTests.java

violetagg

I have proposal for moving handshake timer to the abstract class so that it is available for all handlers. What do you think?

violetagg · 2026-03-24T12:17:01Z

...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

+	String path;
+
+	ContextView contextView;


Suggested change

String path;

ContextView contextView;

final String path;

final ContextView contextView;

violetagg · 2026-03-24T12:18:31Z

...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

+	long dataSentTime;
+
+	long connectionStartTime;
+


What do you think about moving the handshake timer so that it is available for all variants?

Suggested change

long handshakeStartTime;

violetagg · 2026-03-24T12:20:10Z

...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

+		super.handlerAdded(ctx);
+		connectionStartTime = System.nanoTime();
+	}
+


What do you think about moving the handshake timer so that it is available for all variants?

Suggested change

void startHandshake(Channel channel) {

handshakeStartTime = System.nanoTime();

}

void recordHandshakeComplete(Channel channel, String status) {

Duration time = Duration.ofNanos(System.nanoTime() - handshakeStartTime);

if (proxyAddress == null) {

recorder().recordWebSocketHandshakeTime(remoteAddress, path, status, time);

}

else {

recorder().recordWebSocketHandshakeTime(remoteAddress, proxyAddress, path, status, time);

}

}

void recordHandshakeFailure(Channel channel) {

Duration time = Duration.ofNanos(System.nanoTime() - handshakeStartTime);

if (proxyAddress == null) {

recorder().recordWebSocketHandshakeTime(remoteAddress, path, "ERROR", time);

}

else {

recorder().recordWebSocketHandshakeTime(remoteAddress, proxyAddress, path, "ERROR", time);

}

}

violetagg · 2026-03-24T12:21:56Z

...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

+		try {
+			if (msg instanceof WebSocketFrame) {
+				WebSocketFrame frame = (WebSocketFrame) msg;
+				if (isDataFrame(frame)) {
+					if (dataSentTime == 0) {
+						dataSentTime = System.nanoTime();
+					}
+					dataSent += extractProcessedDataFromBuffer(frame);
+
+					if (frame.isFinalFragment()) {
+						recordWrite(remoteAddress);
+						dataSentTime = 0;
+					}
+				}
+			}
+		}
+		catch (RuntimeException e) {
+			if (log.isWarnEnabled()) {
+				log.warn(format(ctx.channel(), "Exception caught while recording metrics."), e);
+			}
+		}


What do you think if we do the recording once we know that the write was successful?

Suggested change

try {

if (msg instanceof WebSocketFrame) {

WebSocketFrame frame = (WebSocketFrame) msg;

if (isDataFrame(frame)) {

if (dataSentTime == 0) {

dataSentTime = System.nanoTime();

}

dataSent += extractProcessedDataFromBuffer(frame);

if (frame.isFinalFragment()) {

recordWrite(remoteAddress);

dataSentTime = 0;

}

}

}

}

catch (RuntimeException e) {

if (log.isWarnEnabled()) {

log.warn(format(ctx.channel(), "Exception caught while recording metrics."), e);

}

}

if (msg instanceof WebSocketFrame) {

WebSocketFrame frame = (WebSocketFrame) msg;

if (isDataFrame(frame)) {

if (dataSentTime == 0) {

dataSentTime = System.nanoTime();

}

dataSent += extractProcessedDataFromBuffer(frame);

if (frame.isFinalFragment()) {

promise.addListener(f -> {

try {

recordWrite(remoteAddress);

dataSentTime = 0;

}

catch (RuntimeException e) {

if (log.isWarnEnabled()) {

log.warn(format(ctx.channel(), "Exception caught while recording metrics."), e);

}

}

});

}

}

}

violetagg · 2026-03-24T12:23:38Z

...-http/src/main/java/reactor/netty/http/client/ContextAwareWebSocketClientMetricsHandler.java

+	protected ContextAwareWebSocketClientMetricsRecorder recorder() {
+		return recorder;
+	}
+


Bring here an implementation for handshake

Suggested change

@Override

void recordHandshakeComplete(io.netty.channel.Channel channel, String status) {

Duration time = Duration.ofNanos(System.nanoTime() - handshakeStartTime);

if (proxyAddress == null) {

recorder.recordWebSocketHandshakeTime(contextView, remoteAddress, path, status, time);

}

else {

recorder.recordWebSocketHandshakeTime(contextView, remoteAddress, proxyAddress, path, status, time);

}

}

@Override

void recordHandshakeFailure(io.netty.channel.Channel channel) {

Duration time = Duration.ofNanos(System.nanoTime() - handshakeStartTime);

if (proxyAddress == null) {

recorder.recordWebSocketHandshakeTime(contextView, remoteAddress, path, "ERROR", time);

}

else {

recorder.recordWebSocketHandshakeTime(contextView, remoteAddress, proxyAddress, path, "ERROR", time);

}

}

violetagg · 2026-03-24T12:37:40Z

reactor-netty-http/src/main/java/reactor/netty/http/client/WebsocketClientOperations.java

 	final Sinks.One<WebSocketCloseStatus> onCloseState;
 	final boolean proxyPing;

+	@Nullable MicrometerWebSocketClientMetricsHandler micrometerWsHandler;


We have handshake timer for all

Suggested change

@Nullable MicrometerWebSocketClientMetricsHandler micrometerWsHandler;

@Nullable AbstractWebSocketClientMetricsHandler micrometerWsHandler;

violetagg · 2026-03-24T12:39:44Z

reactor-netty-http/src/main/java/reactor/netty/http/client/WebsocketClientOperations.java

+				MicrometerWebSocketClientMetricsHandler micrometerHandler = new MicrometerWebSocketClientMetricsHandler(
+						MicrometerWebSocketClientMetricsRecorder.INSTANCE,
+						httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);
+				micrometerHandler.startHandshake(channel);
+				this.micrometerWsHandler = micrometerHandler;
+				wsHandler = micrometerHandler;


Let's here only construct the new handler

Suggested change

MicrometerWebSocketClientMetricsHandler micrometerHandler = new MicrometerWebSocketClientMetricsHandler(

MicrometerWebSocketClientMetricsRecorder.INSTANCE,

httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);

micrometerHandler.startHandshake(channel);

this.micrometerWsHandler = micrometerHandler;

wsHandler = micrometerHandler;

wsHandler = new MicrometerWebSocketClientMetricsHandler(

MicrometerWebSocketClientMetricsRecorder.INSTANCE,

httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);

violetagg · 2026-03-24T12:41:24Z

reactor-netty-http/src/main/java/reactor/netty/http/client/WebsocketClientOperations.java

+				wsHandler = new ContextAwareWebSocketClientMetricsHandler(
+						new DefaultContextAwareWebSocketClientMetricsRecorder(ctxHandler.recorder),
+						httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);


Wrap the recorder only if it is not ContextAwareWebSocketClientMetricsRecorder

Suggested change

wsHandler = new ContextAwareWebSocketClientMetricsHandler(

new DefaultContextAwareWebSocketClientMetricsRecorder(ctxHandler.recorder),

httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);

ContextAwareWebSocketClientMetricsRecorder wsRecorder;

if (ctxHandler.recorder instanceof ContextAwareWebSocketClientMetricsRecorder) {

wsRecorder = (ContextAwareWebSocketClientMetricsRecorder) ctxHandler.recorder;

}

else {

wsRecorder = new DefaultContextAwareWebSocketClientMetricsRecorder(ctxHandler.recorder);

}

wsHandler = new ContextAwareWebSocketClientMetricsHandler(

wsRecorder,

httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);

violetagg · 2026-03-24T12:44:51Z

reactor-netty-http/src/main/java/reactor/netty/http/client/WebsocketClientOperations.java

+			else {
+				wsHandler = new WebSocketClientMetricsHandler(
+						MicrometerWebSocketClientMetricsRecorder.INSTANCE,
+						httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);
+			}


We will need a wrapper DefaultWebSocketClientMetricsRecorder similar to what we have to context aware. What do you think?

Suggested change

else {

wsHandler = new WebSocketClientMetricsHandler(

MicrometerWebSocketClientMetricsRecorder.INSTANCE,

httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);

}

else if (httpHandler instanceof HttpClientMetricsHandler) {

HttpClientMetricsHandler plainHandler = (HttpClientMetricsHandler) httpHandler;

WebSocketClientMetricsRecorder wsRecorder;

if (plainHandler.recorder instanceof WebSocketClientMetricsRecorder) {

wsRecorder = (WebSocketClientMetricsRecorder) plainHandler.recorder;

}

else {

wsRecorder = new DefaultWebSocketClientMetricsRecorder(plainHandler.recorder);

}

wsHandler = new WebSocketClientMetricsHandler(

wsRecorder,

httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);

}

else {

return;

}

violetagg · 2026-03-24T12:45:51Z

reactor-netty-http/src/main/java/reactor/netty/http/client/WebsocketClientOperations.java

+						httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);
+			}
+
+			channel.pipeline().addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.WsMetricsHandler, wsHandler);


Suggested change

channel.pipeline().addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.WsMetricsHandler, wsHandler);

wsHandler.startHandshake(channel);

this.micrometerWsHandler = wsHandler;

channel.pipeline().addBefore(NettyPipeline.ReactiveBridge, NettyPipeline.WsMetricsHandler, wsHandler);

violetagg · 2026-03-24T12:50:41Z

...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

+
+	protected abstract WebSocketClientMetricsRecorder recorder();
+
+	protected void recordConnectionClosed(ChannelHandlerContext ctx) {


Do we need the method param?

violetagg · 2026-03-24T12:51:16Z

...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

+		}
+	}
+
+	protected void recordException(ChannelHandlerContext ctx) {


Do we need the method param?

violetagg · 2026-03-24T12:52:11Z

...etty-http/src/main/java/reactor/netty/http/client/AbstractWebSocketClientMetricsHandler.java

+		}
+	}
+
+	protected void recordRead(io.netty.channel.Channel channel, SocketAddress address) {


Do we need the method param channel?

violetagg · 2026-03-24T13:05:43Z

Please also ensure you have Signed-off-by so that we have a happy DCO check

LivingLikeKrillin force-pushed the feature/websocket-client-metrics branch from d16260f to 49cdeda Compare February 28, 2026 15:19

violetagg linked an issue Mar 4, 2026 that may be closed by this pull request

Support for WebSocket client metrics #3820

Open

LivingLikeKrillin added 3 commits March 9, 2026 18:26

Add WebSocket client metrics tests

dcbecaa

Test handshake time, data received/sent, connection duration, handshake failure, and protocol combinations (HTTP/1.1 and H2). Related to reactor#3820 Signed-off-by: LivingLikeKrillin <143606756+LivingLikeKrillin@users.noreply.github.com>

LivingLikeKrillin force-pushed the feature/websocket-client-metrics branch from 49cdeda to dcbecaa Compare March 9, 2026 09:27

Merge branch 'main' into feature/websocket-client-metrics

159101e

Signed-off-by: LivingLikeKrillin <143606756+LivingLikeKrillin@users.noreply.github.com>

violetagg added the type/enhancement A general enhancement label Mar 9, 2026

violetagg modified the milestones: 1.3.4, 1.3.5 Mar 9, 2026

violetagg requested changes Mar 9, 2026

View reviewed changes

violetagg requested changes Mar 13, 2026

View reviewed changes

LivingLikeKrillin added 3 commits March 24, 2026 09:47

LivingLikeKrillin force-pushed the feature/websocket-client-metrics branch from 7161609 to 10db602 Compare March 24, 2026 00:48

github-advanced-security bot found potential problems Mar 24, 2026

View reviewed changes

violetagg requested changes Mar 24, 2026

View reviewed changes


		protected abstract WebSocketClientMetricsRecorder recorder();

		protected void recordConnectionClosed(ChannelHandlerContext ctx) {

+	void startHandshake(Channel channel) {
+		handshakeStartTime = System.nanoTime();
+	}
+	void recordHandshakeComplete(Channel channel, String status) {
+		Duration time = Duration.ofNanos(System.nanoTime() - handshakeStartTime);
+		if (proxyAddress == null) {
+			recorder().recordWebSocketHandshakeTime(remoteAddress, path, status, time);
+		}
+		else {
+			recorder().recordWebSocketHandshakeTime(remoteAddress, proxyAddress, path, status, time);
+		}
+	}
+	void recordHandshakeFailure(Channel channel) {
+		Duration time = Duration.ofNanos(System.nanoTime() - handshakeStartTime);
+		if (proxyAddress == null) {
+			recorder().recordWebSocketHandshakeTime(remoteAddress, path, "ERROR", time);
+		}
+		else {
+			recorder().recordWebSocketHandshakeTime(remoteAddress, proxyAddress, path, "ERROR", time);
+		}
+	}

+	@Override
+	void recordHandshakeComplete(io.netty.channel.Channel channel, String status) {
+		Duration time = Duration.ofNanos(System.nanoTime() - handshakeStartTime);
+		if (proxyAddress == null) {
+			recorder.recordWebSocketHandshakeTime(contextView, remoteAddress, path, status, time);
+		}
+		else {
+			recorder.recordWebSocketHandshakeTime(contextView, remoteAddress, proxyAddress, path, status, time);
+		}
+	}
+	@Override
+	void recordHandshakeFailure(io.netty.channel.Channel channel) {
+		Duration time = Duration.ofNanos(System.nanoTime() - handshakeStartTime);
+		if (proxyAddress == null) {
+			recorder.recordWebSocketHandshakeTime(contextView, remoteAddress, path, "ERROR", time);
+		}
+		else {
+			recorder.recordWebSocketHandshakeTime(contextView, remoteAddress, proxyAddress, path, "ERROR", time);
+		}
+	}

	@Nullable MicrometerWebSocketClientMetricsHandler micrometerWsHandler;
	@Nullable AbstractWebSocketClientMetricsHandler micrometerWsHandler;

-			else {
-				wsHandler = new WebSocketClientMetricsHandler(
-						MicrometerWebSocketClientMetricsRecorder.INSTANCE,
-						httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);
-			}
+			else if (httpHandler instanceof HttpClientMetricsHandler) {
+				HttpClientMetricsHandler plainHandler = (HttpClientMetricsHandler) httpHandler;
+				WebSocketClientMetricsRecorder wsRecorder;
+				if (plainHandler.recorder instanceof WebSocketClientMetricsRecorder) {
+					wsRecorder = (WebSocketClientMetricsRecorder) plainHandler.recorder;
+				}
+				else {
+					wsRecorder = new DefaultWebSocketClientMetricsRecorder(plainHandler.recorder);
+				}
+				wsHandler = new WebSocketClientMetricsHandler(
+						wsRecorder,
+						httpHandler.remoteAddress, httpHandler.proxyAddress, resolvedPath, ctxView, httpMethod);
+			}
+			else {
+				return;
+			}

Conversation

LivingLikeKrillin commented Feb 28, 2026

Summary

Test plan

Uh oh!

violetagg commented Mar 9, 2026

Uh oh!

violetagg left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Check notice

Check notice

Check notice

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

violetagg left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

violetagg commented Mar 24, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

violetagg left a comment •

edited

Loading