Skip to content

Commit

Permalink
Merge #3211 into 2.0.0-M4
Browse files Browse the repository at this point in the history
  • Loading branch information
@violetagg
violetagg committed Apr 30, 2024
2 parents c0b71af + 1c65384 commit 0af7df3
Show file tree
Hide file tree
Showing 3 changed files with 92 additions and 41 deletions.
102 changes: 75 additions & 27 deletions ...etty5-http/src/main/java/reactor/netty5/http/server/AbstractHttpServerMetricsHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@
import reactor.util.Logger;
import reactor.util.Loggers;
import reactor.util.annotation.Nullable;
import reactor.util.context.Context;
import reactor.util.context.ContextView;

import java.net.SocketAddress;
import java.time.Duration;
Expand All @@ -53,12 +55,21 @@ abstract class AbstractHttpServerMetricsHandler extends ChannelHandlerAdapter {
boolean channelActivated;
boolean channelOpened;

ContextView contextView;

long dataReceived;
long dataReceivedTime;

long dataSent;
long dataSentTime;

boolean initialized;

String method;
String path;
SocketAddress remoteSocketAddress;
String status;

final Function<String, String> methodTagValue;
final Function<String, String> uriTagValue;

Expand All @@ -72,10 +83,16 @@ protected AbstractHttpServerMetricsHandler(
protected AbstractHttpServerMetricsHandler(AbstractHttpServerMetricsHandler copy) {
this.channelActivated = copy.channelActivated;
this.channelOpened = copy.channelOpened;
this.contextView = copy.contextView;
this.dataReceived = copy.dataReceived;
this.dataReceivedTime = copy.dataReceivedTime;
this.dataSent = copy.dataSent;
this.dataSentTime = copy.dataSentTime;
this.initialized = copy.initialized;
this.method = copy.method;
this.path = copy.path;
this.remoteSocketAddress = copy.remoteSocketAddress;
this.status = copy.status;
this.methodTagValue = copy.methodTagValue;
this.uriTagValue = copy.uriTagValue;
}
Expand Down Expand Up @@ -134,8 +151,19 @@ public Future<Void> write(ChannelHandlerContext ctx, Object msg) {

ChannelOperations<?, ?> channelOps = ChannelOperations.get(ctx.channel());
if (channelOps instanceof HttpServerOperations ops) {
startWrite(ops, uriTagValue == null ? ops.path : uriTagValue.apply(ops.path),
methodTagValue.apply(ops.method().name()), ops.status().codeAsText().toString());
if (!initialized) {
method = methodTagValue.apply(ops.method().name());
path = uriTagValue == null ? ops.path : uriTagValue.apply(ops.path);
// Always take the remote address from the operations in order to consider proxy information
// Use remoteSocketAddress() in order to obtain UDS info
remoteSocketAddress = ops.remoteSocketAddress();
initialized = true;
}
if (contextView == null) {
contextView(ops);
}
status = ops.status().codeAsText().toString();
startWrite(ops);
}
}

Expand All @@ -150,8 +178,7 @@ public Future<Void> write(ChannelHandlerContext ctx, Object msg) {
.addListener(future -> {
if (channelOps instanceof HttpServerOperations ops) {
try {
recordWrite(ops, uriTagValue == null ? ops.path : uriTagValue.apply(ops.path),
methodTagValue.apply(ops.method().name()), ops.status().codeAsText().toString());
recordWrite(ops);
}
catch (RuntimeException e) {
// Allow request-response exchange to continue, unaffected by metrics problem
Expand All @@ -162,8 +189,6 @@ public Future<Void> write(ChannelHandlerContext ctx, Object msg) {
}

recordInactiveConnectionOrStream(ctx.channel());

dataSent = 0;
});
}
}
Expand All @@ -178,11 +203,20 @@ public Future<Void> write(ChannelHandlerContext ctx, Object msg) {

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
HttpServerOperations ops = null;
try {
if (msg instanceof HttpRequest) {
reset();
ChannelOperations<?, ?> channelOps = ChannelOperations.get(ctx.channel());
if (channelOps instanceof HttpServerOperations ops) {
startRead(ops, uriTagValue == null ? ops.path : uriTagValue.apply(ops.path), methodTagValue.apply(ops.method().name()));
if (channelOps instanceof HttpServerOperations) {
ops = (HttpServerOperations) channelOps;
method = methodTagValue.apply(ops.method().name());
path = uriTagValue == null ? ops.path : uriTagValue.apply(ops.path);
// Always take the remote address from the operations in order to consider proxy information
// Use remoteSocketAddress() in order to obtain UDS info
remoteSocketAddress = ops.remoteSocketAddress();
initialized = true;
startRead(ops);
}

channelActivated = true;
Expand All @@ -199,12 +233,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
dataReceived += extractProcessedDataFromBuffer(msg);

if (msg instanceof LastHttpContent) {
ChannelOperations<?, ?> channelOps = ChannelOperations.get(ctx.channel());
if (channelOps instanceof HttpServerOperations ops) {
recordRead(ops, uriTagValue == null ? ops.path : uriTagValue.apply(ops.path), methodTagValue.apply(ops.method().name()));
}

dataReceived = 0;
recordRead();
}
}
catch (RuntimeException e) {
Expand All @@ -215,16 +244,17 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
}

ctx.fireChannelRead(msg);

if (ops != null) {
// ContextView is available only when a subscription to the I/O Handler happens
contextView(ops);
}
}

@Override
public void channelExceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
try {
ChannelOperations<?, ?> channelOps = ChannelOperations.get(ctx.channel());
if (channelOps instanceof HttpServerOperations ops) {
// Always take the remote address from the operations in order to consider proxy information
recordException(ops, uriTagValue == null ? ops.path : uriTagValue.apply(ops.path));
}
recordException();
}
catch (RuntimeException e) {
// Allow request-response exchange to continue, unaffected by metrics problem
Expand All @@ -248,21 +278,25 @@ else if (msg instanceof HttpContent<?> httpContent) {

protected abstract HttpServerMetricsRecorder recorder();

protected void recordException(HttpServerOperations ops, String path) {
protected void contextView(HttpServerOperations ops) {
this.contextView = Context.empty();
}

protected void recordException() {
// Always take the remote address from the operations in order to consider proxy information
// Use remoteSocketAddress() in order to obtain UDS info
recorder().incrementErrorsCount(ops.remoteSocketAddress(), path);
recorder().incrementErrorsCount(remoteSocketAddress, path);
}

protected void recordRead(HttpServerOperations ops, String path, String method) {
protected void recordRead() {
recorder().recordDataReceivedTime(path, method, Duration.ofNanos(System.nanoTime() - dataReceivedTime));

// Always take the remote address from the operations in order to consider proxy information
// Use remoteSocketAddress() in order to obtain UDS info
recorder().recordDataReceived(ops.remoteSocketAddress(), path, dataReceived);
recorder().recordDataReceived(remoteSocketAddress, path, dataReceived);
}

protected void recordWrite(HttpServerOperations ops, String path, String method, String status) {
protected void recordWrite(HttpServerOperations ops) {
Duration dataSentTimeDuration = Duration.ofNanos(System.nanoTime() - dataSentTime);
recorder().recordDataSentTime(path, method, status, dataSentTimeDuration);

Expand All @@ -275,7 +309,7 @@ protected void recordWrite(HttpServerOperations ops, String path, String method,

// Always take the remote address from the operations in order to consider proxy information
// Use remoteSocketAddress() in order to obtain UDS info
recorder().recordDataSent(ops.remoteSocketAddress(), path, dataSent);
recorder().recordDataSent(remoteSocketAddress, path, dataSent);
}

protected void recordActiveConnection(SocketAddress localAddress) {
Expand All @@ -294,11 +328,11 @@ protected void recordClosedStream(SocketAddress localAddress) {
recorder().recordStreamClosed(localAddress);
}

protected void startRead(HttpServerOperations ops, String path, String method) {
protected void startRead(HttpServerOperations ops) {
dataReceivedTime = System.nanoTime();
}

protected void startWrite(HttpServerOperations ops, String path, String method, String status) {
protected void startWrite(HttpServerOperations ops) {
dataSentTime = System.nanoTime();
}

Expand All @@ -324,6 +358,20 @@ void recordInactiveConnectionOrStream(Channel channel) {
}
}

void reset() {
// There is no need to reset 'channelActivated' and 'channelOpened'
contextView = null;
dataReceived = 0;
dataReceivedTime = 0;
dataSent = 0;
dataSentTime = 0;
initialized = false;
method = null;
path = null;
remoteSocketAddress = null;
status = null;
}

static final Set<String> STANDARD_METHODS;
static {
Set<String> standardMethods = new HashSet<>();
Expand Down
19 changes: 11 additions & 8 deletions ...5-http/src/main/java/reactor/netty5/http/server/ContextAwareHttpServerMetricsHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,26 +50,29 @@ protected ContextAwareHttpServerMetricsRecorder recorder() {
}

@Override
protected void recordException(HttpServerOperations ops, String path) {
protected void contextView(HttpServerOperations ops) {
this.contextView = ops.currentContext();
}

@Override
protected void recordException() {
// Always take the remote address from the operations in order to consider proxy information
// Use remoteSocketAddress() in order to obtain UDS info
recorder().incrementErrorsCount(ops.currentContext(), ops.remoteSocketAddress(), path);
recorder().incrementErrorsCount(contextView, remoteSocketAddress, path);
}

@Override
protected void recordRead(HttpServerOperations ops, String path, String method) {
ContextView contextView = ops.currentContext();
protected void recordRead() {
recorder().recordDataReceivedTime(contextView, path, method,
Duration.ofNanos(System.nanoTime() - dataReceivedTime));

// Always take the remote address from the operations in order to consider proxy information
// Use remoteSocketAddress() in order to obtain UDS info
recorder().recordDataReceived(contextView, ops.remoteSocketAddress(), path, dataReceived);
recorder().recordDataReceived(contextView, remoteSocketAddress, path, dataReceived);
}

@Override
protected void recordWrite(HttpServerOperations ops, String path, String method, String status) {
ContextView contextView = ops.currentContext();
protected void recordWrite(HttpServerOperations ops) {
Duration dataSentTimeDuration = Duration.ofNanos(System.nanoTime() - dataSentTime);
recorder().recordDataSentTime(contextView, path, method, status, dataSentTimeDuration);

Expand All @@ -83,6 +86,6 @@ protected void recordWrite(HttpServerOperations ops, String path, String method,

// Always take the remote address from the operations in order to consider proxy information
// Use remoteSocketAddress() in order to obtain UDS info
recorder().recordDataSent(contextView, ops.remoteSocketAddress(), path, dataSent);
recorder().recordDataSent(contextView, remoteSocketAddress, path, dataSent);
}
}
12 changes: 6 additions & 6 deletions ...ty5-http/src/main/java/reactor/netty5/http/server/MicrometerHttpServerMetricsHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -85,13 +85,13 @@ protected HttpServerMetricsRecorder recorder() {
}

@Override
protected void recordWrite(HttpServerOperations ops, String path, String method, String status) {
protected void recordWrite(HttpServerOperations ops) {
Duration dataSentTimeDuration = Duration.ofNanos(System.nanoTime() - dataSentTime);
recorder().recordDataSentTime(path, method, status, dataSentTimeDuration);

// Always take the remote address from the operations in order to consider proxy information
// Use remoteSocketAddress() in order to obtain UDS info
recorder().recordDataSent(ops.remoteSocketAddress(), path, dataSent);
recorder().recordDataSent(remoteSocketAddress, path, dataSent);

// Cannot invoke the recorder anymore:
// 1. The recorder is one instance only, it is invoked for all requests that can happen
Expand All @@ -108,8 +108,8 @@ protected void recordWrite(HttpServerOperations ops, String path, String method,
}

@Override
protected void startRead(HttpServerOperations ops, String path, String method) {
super.startRead(ops, path, method);
protected void startRead(HttpServerOperations ops) {
super.startRead(ops);

responseTimeHandlerContext = new ResponseTimeHandlerContext(recorder, method, path, ops);
responseTimeObservation = Observation.createNotStarted(this.responseTimeName, responseTimeHandlerContext, OBSERVATION_REGISTRY);
Expand All @@ -119,8 +119,8 @@ protected void startRead(HttpServerOperations ops, String path, String method) {

// response
@Override
protected void startWrite(HttpServerOperations ops, String path, String method, String status) {
super.startWrite(ops, path, method, status);
protected void startWrite(HttpServerOperations ops) {
super.startWrite(ops);

if (responseTimeObservation == null) {
responseTimeHandlerContext = new ResponseTimeHandlerContext(recorder, method, path, ops);
Expand Down

0 comments on commit 0af7df3

Please sign in to comment.