Provide the context of the interaction when reporting client requests and HTTP requests.

Motivation:

The metrics SPI before the tracing SPI, it does not take in account the request context unlike the tracing SPI. We can definitely introduce its support by passing the context as part of the metrics SPI event signal.

Changes:

- Overload appropriate metrics SPI events with a signature augmented with the Vert.x context
- Deprecate for removal in Vertx 6.0 the original versions
- Implement and test the behavior

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/impl/http1x/Http1xClientConnection.java

@@ -255,7 +255,7 @@ private void beginRequest(Stream stream, io.vertx.core.http.impl.HttpRequestHead
       inflight.addLast(stream);
       this.isConnect = connect;
       if (this.metrics != null) {
-        stream.metric = this.metrics.requestBegin(request.uri, new ObservableRequest(request));
+        stream.metric = this.metrics.requestBegin(stream.context, request.uri, new ObservableRequest(request));
       }
       VertxTracer tracer = stream.context.tracer();
       if (tracer != null) {

vertx-core/src/main/java/io/vertx/core/http/impl/http1x/Http1xServerRequest.java

@@ -609,7 +609,7 @@ private void reportRequestComplete() {
   private void reportRequestBegin() {
     HttpServerMetrics metrics = conn.metrics;
     if (metrics != null) {
-      metric = metrics.requestBegin(conn.metric(), this);
+      metric = metrics.requestBegin(context, conn.metric(), this);
     }
     VertxTracer tracer = context.tracer();
     if (tracer != null) {

vertx-core/src/main/java/io/vertx/core/http/impl/http2/DefaultHttp2ClientStream.java

@@ -197,7 +197,7 @@ public void onPush(DefaultHttp2ClientStream pushStream, int promisedStreamId, Ht
     HttpClientPush push = new HttpClientPush(new HttpRequestHead(headers.scheme(), headers.method(), headers.path(), headers, headers.authority(), null, null), pushStream);
     pushStream.init(promisedStreamId, writable);
     if (pushStream.observable != null) {
-      pushStream.observable.observePush(headers);
+      pushStream.observable.observePush(pushStream.context, headers);
     }
     context.execute(push, this::handlePush);
   }

vertx-core/src/main/java/io/vertx/core/http/impl/http2/codec/Http2ServerConnectionImpl.java

@@ -32,6 +32,7 @@
 import io.vertx.core.http.impl.http2.Http2ServerConnection;
 import io.vertx.core.http.impl.http2.Http2ServerStream;
 import io.vertx.core.internal.ContextInternal;
+import io.vertx.core.internal.PromiseInternal;
 import io.vertx.core.net.HostAndPort;
 import io.vertx.core.spi.metrics.HttpServerMetrics;
 import io.vertx.core.spi.metrics.TransportMetrics;
@@ -229,7 +230,7 @@ private synchronized void doSendPush(int streamId, HostAndPort authority, HttpMe
             this,
             metrics,
             metric(),
-            context,
+            ((PromiseInternal)promise).context(),
             new HttpRequestHeaders(headers_),
             method,
             path,

vertx-core/src/main/java/io/vertx/core/http/impl/observability/ClientStreamObserver.java

@@ -40,9 +40,9 @@ public ClientStreamObserver(ContextInternal context, TracingPolicy tracingPolicy
     this.clientMetrics = clientMetrics;
   }
 
-  public void observePush(HttpRequestHeaders headers) {
+  public void observePush(ContextInternal context,  HttpRequestHeaders headers) {
     if (clientMetrics != null) {
-      Object metric = clientMetrics.requestBegin(headers.path().toString(), observableRequest(headers, remoteAddress));
+      Object metric = clientMetrics.requestBegin(context, headers.path().toString(), observableRequest(headers, remoteAddress));
       this.metric = metric;
       clientMetrics.requestEnd(metric, 0L);
     }
@@ -56,7 +56,7 @@ public void observeUpgrade(Object metric, Object trace) {
   public void observeOutboundHeaders(HttpHeaders headers) {
     HttpRequestHeaders r = (HttpRequestHeaders) headers;
     if (clientMetrics != null) {
-      metric = clientMetrics.requestBegin(r.path(), observableRequest(r, remoteAddress));
+      metric = clientMetrics.requestBegin(context, r.path(), observableRequest(r, remoteAddress));
     }
     VertxTracer tracer = context.tracer();
     if (tracer != null) {

vertx-core/src/main/java/io/vertx/core/http/impl/observability/ServerStreamObserver.java

@@ -51,7 +51,7 @@ public void observeInboundTrailers(long bytesRead) {
 
   public void observeInboundHeaders(HttpHeaders headers) {
     if (serverMetrics != null) {
-      metric = serverMetrics.requestBegin(socketMetric, observableRequest((HttpRequestHeaders) headers, remoteAddress));
+      metric = serverMetrics.requestBegin(context, socketMetric, observableRequest((HttpRequestHeaders) headers, remoteAddress));
     }
     VertxTracer tracer = context.tracer();
     if (tracer != null) {
@@ -85,7 +85,7 @@ public void observeRoute(String route) {
 
   public void observePush(HttpResponseHeaders headers, HttpMethod method, String uri) {
     if (serverMetrics != null) {
-      metric = serverMetrics.responsePushed(socketMetric, method, uri, observableResponse(headers, remoteAddress));
+      metric = serverMetrics.responsePushed(context, socketMetric, method, uri, observableResponse(headers, remoteAddress));
     }
   }
 }

vertx-core/src/main/java/io/vertx/core/spi/metrics/ClientMetrics.java

@@ -11,26 +11,37 @@
 
 package io.vertx.core.spi.metrics;
 
+import io.vertx.core.Context;
+
 /**
  * The client metrics SPI that Vert.x will use to call when client events occur.<p/>
  *
  * @author <a href="mailto:[email protected]">Julien Viet</a>
  */
 public interface ClientMetrics<M, Req, Resp> extends Metrics {
 
+  /**
+   * @deprecated instead override {@link #requestBegin(Context, String, Object)}, this will be removed in Vert.x 6
+   */
+  @Deprecated(forRemoval = true)
+  default M requestBegin(String uri, Req request) {
+    return null;
+  }
+
   /**
    * Called when a client request begins. Vert.x will invoke {@link #requestEnd} when the request
    * has ended or {@link #requestReset} if the request/response has failed before.
    *
    * <p>The request uri is an arbitrary URI that depends on the client, e.g an HTTP request uri,
    * a SQL query, etc...
    *
+   * @param context the vertx context associated with the request
    * @param uri an arbitrary uri
    * @param request the request object
    * @return the request metric
    */
-  default M requestBegin(String uri, Req request) {
-    return null;
+  default M requestBegin(Context context, String uri, Req request) {
+    return requestBegin(uri, request);
   }
 
   /**

vertx-core/src/main/java/io/vertx/core/spi/metrics/HttpServerMetrics.java

@@ -11,6 +11,7 @@
 
 package io.vertx.core.spi.metrics;
 
+import io.vertx.core.Context;
 import io.vertx.core.http.HttpMethod;
 import io.vertx.core.http.ServerWebSocket;
 import io.vertx.core.spi.observability.HttpRequest;
@@ -37,16 +38,25 @@
  */
 public interface HttpServerMetrics<R, W, S> extends TransportMetrics<S> {
 
+  /**
+   * @deprecated instead override {@link #requestBegin(Context, Object, HttpRequest)}, this will be removed in Vert.x 6
+   */
+  @Deprecated(forRemoval = true)
+  default R requestBegin(S socketMetric, HttpRequest request) {
+    return null;
+  }
+
   /**
    * Called when an http server request begins. Vert.x will invoke {@link #responseEnd} when the response has ended
    * or {@link #requestReset} if the request/response has failed before.
    *
+   * @param context the vertx context associated with the request
    * @param socketMetric the socket metric
    * @param request the http server reuqest
    * @return the request metric
    */
-  default R requestBegin(S socketMetric, HttpRequest request) {
-    return null;
+  default R requestBegin(Context context, S socketMetric, HttpRequest request) {
+    return requestBegin(socketMetric, request);
   }
 
   /**
@@ -76,16 +86,26 @@ default void requestReset(R requestMetric) {
   default void responseBegin(R requestMetric, HttpResponse response) {
   }
 
+  /**
+   * @deprecated instead override {@link #responsePushed(Context, Object, HttpMethod, String, HttpResponse)},
+   * this will be removed in Vert.x 6
+   */
+  @Deprecated(forRemoval = true)
+  default R responsePushed(S socketMetric, HttpMethod method, String uri, HttpResponse response) {
+    return null;
+  }
+
   /**
    * Called when an http server response is pushed.
    *
+   * @param context the vertx context associated with the request
    * @param socketMetric the socket metric
    * @param method the pushed response method
    * @param uri the pushed response uri
    * @param response the http server response  @return the request metric
    */
-  default R responsePushed(S socketMetric, HttpMethod method, String uri, HttpResponse response) {
-    return null;
+  default R responsePushed(Context context, S socketMetric, HttpMethod method, String uri, HttpResponse response) {
+    return responsePushed(socketMetric, method, uri, response);
   }
 
   /**

vertx-core/src/test/java/io/vertx/test/fakemetrics/EndpointMetric.java

@@ -11,6 +11,7 @@
 
 package io.vertx.test.fakemetrics;
 
+import io.vertx.core.Context;
 import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.core.spi.observability.HttpRequest;
 import io.vertx.core.spi.observability.HttpResponse;
@@ -32,9 +33,9 @@ public EndpointMetric() {
   }
 
   @Override
-  public HttpClientMetric requestBegin(String uri, HttpRequest request) {
+  public HttpClientMetric requestBegin(Context context, String uri, HttpRequest request) {
     requestCount.incrementAndGet();
-    HttpClientMetric metric = new HttpClientMetric(this, request);
+    HttpClientMetric metric = new HttpClientMetric(this, context, request);
     requests.put(request, metric);
     return metric;
   }

vertx-core/src/test/java/io/vertx/test/fakemetrics/FakeHttpServerMetrics.java

@@ -11,6 +11,7 @@
 
 package io.vertx.test.fakemetrics;
 
+import io.vertx.core.Context;
 import io.vertx.core.http.HttpMethod;
 import io.vertx.core.http.HttpServerRequest;
 import io.vertx.core.http.ServerWebSocket;
@@ -45,8 +46,8 @@ public HttpServerMetric getResponseMetric(String uri) {
   }
 
   @Override
-  public HttpServerMetric requestBegin(SocketMetric socketMetric, HttpRequest request) {
-    HttpServerMetric metric = new HttpServerMetric(request, socketMetric);
+  public HttpServerMetric requestBegin(Context context, SocketMetric socketMetric, HttpRequest request) {
+    HttpServerMetric metric = new HttpServerMetric(context, request, socketMetric);
     requests.add(metric);
     return metric;
   }
@@ -58,8 +59,8 @@ public void requestEnd(HttpServerMetric requestMetric, HttpRequest request, long
   }
 
   @Override
-  public HttpServerMetric responsePushed(SocketMetric socketMetric, HttpMethod method, String uri, HttpResponse response) {
-    HttpServerMetric requestMetric = new HttpServerMetric(uri, socketMetric);
+  public HttpServerMetric responsePushed(Context context, SocketMetric socketMetric, HttpMethod method, String uri, HttpResponse response) {
+    HttpServerMetric requestMetric = new HttpServerMetric(context, uri, socketMetric);
     requestMetric.response.set(response);
     requests.add(requestMetric);
     return requestMetric;