HTTP stream telemetry api (#433)

Author: TingDaoK

.github/workflows/ci.yml

@@ -158,7 +158,7 @@ jobs:
     - name: Build and test
       run: |
         python3 -c "from urllib.request import urlretrieve; urlretrieve('${{ env.BUILDER_HOST }}/${{ env.BUILDER_SOURCE }}/${{ env.BUILDER_VERSION }}/builder.pyz?run=${{ env.RUN }}', 'builder.pyz')"
-        python3 builder.pyz build -p aws-c-http --cmake-extra=-DENABLE_LOCALHOST_INTEGRATION_TESTS=ON
+        python3 builder.pyz build -p aws-c-http --cmake-extra=-DENABLE_LOCALHOST_INTEGRATION_TESTS=ON --config Debug
 
   localhost-test-mac:
     runs-on: macos-11 # latest
@@ -168,7 +168,7 @@ jobs:
     - name: Build and test
       run: |
         python3 -c "from urllib.request import urlretrieve; urlretrieve('${{ env.BUILDER_HOST }}/${{ env.BUILDER_SOURCE }}/${{ env.BUILDER_VERSION }}/builder.pyz?run=${{ env.RUN }}', 'builder.pyz')"
-        python3 builder.pyz build -p aws-c-http --cmake-extra=-DENABLE_LOCALHOST_INTEGRATION_TESTS=ON
+        python3 builder.pyz build -p aws-c-http --cmake-extra=-DENABLE_LOCALHOST_INTEGRATION_TESTS=ON --config Debug
 
   localhost-test-win:
     runs-on: windows-2022 # latest
@@ -178,4 +178,4 @@ jobs:
     - name: Build and test
       run: |
         python -c "from urllib.request import urlretrieve; urlretrieve('${{ env.BUILDER_HOST }}/${{ env.BUILDER_SOURCE }}/${{ env.BUILDER_VERSION }}/builder.pyz?run=${{ env.RUN }}', 'builder.pyz')"
-        python builder.pyz build -p aws-c-http --cmake-extra=-DENABLE_LOCALHOST_INTEGRATION_TESTS=ON
+        python builder.pyz build -p aws-c-http --cmake-extra=-DENABLE_LOCALHOST_INTEGRATION_TESTS=ON --config Debug

include/aws/http/connection.h

@@ -10,6 +10,7 @@
 
 struct aws_client_bootstrap;
 struct aws_socket_options;
+struct aws_socket_endpoint;
 struct aws_tls_connection_options;
 struct aws_http2_setting;
 struct proxy_env_var_settings;
@@ -512,6 +513,12 @@ enum aws_http_version aws_http_connection_get_version(const struct aws_http_conn
 AWS_HTTP_API
 struct aws_channel *aws_http_connection_get_channel(struct aws_http_connection *connection);
 
+/**
+ * Returns the remote endpoint of the HTTP connection.
+ */
+AWS_HTTP_API
+const struct aws_socket_endpoint *aws_http_connection_get_remote_endpoint(const struct aws_http_connection *connection);
+
 /**
  * Initialize an map copied from the *src map, which maps `struct aws_string *` to `enum aws_http_version`.
  */

include/aws/http/private/h2_connection.h

@@ -125,6 +125,7 @@ struct aws_h2_connection {
         uint64_t outgoing_timestamp_ns;
         /* Timestamp when connection has data to receive, which is when there is an active stream */
         uint64_t incoming_timestamp_ns;
+
     } thread_data;
 
     /* Any thread may touch this data, but the lock must be held (unless it's an atomic) */

include/aws/http/private/request_response_impl.h

@@ -43,11 +43,13 @@ struct aws_http_stream {
     aws_http_on_incoming_headers_fn *on_incoming_headers;
     aws_http_on_incoming_header_block_done_fn *on_incoming_header_block_done;
     aws_http_on_incoming_body_fn *on_incoming_body;
+    aws_http_on_stream_metrics_fn *on_metrics;
     aws_http_on_stream_complete_fn *on_complete;
     aws_http_on_stream_destroy_fn *on_destroy;
 
     struct aws_atomic_var refcount;
     enum aws_http_method request_method;
+    struct aws_http_stream_metrics metrics;
 
     union {
         struct aws_http_stream_client_data {

include/aws/http/request_response.h

@@ -194,6 +194,46 @@ typedef void(aws_http_on_stream_complete_fn)(struct aws_http_stream *stream, int
  */
 typedef void(aws_http_on_stream_destroy_fn)(void *user_data);
 
+/**
+ * Tracing metrics for aws_http_stream.
+ * Data maybe not be available if the data of stream was never sent/received before it completes.
+ */
+struct aws_http_stream_metrics {
+    /* The time stamp when the request started to be encoded. -1 means data not available. Timestamp
+     * are from `aws_high_res_clock_get_ticks` */
+    int64_t send_start_timestamp_ns;
+    /* The time stamp when the request finished to be encoded. -1 means data not available.
+     * Timestamp are from `aws_high_res_clock_get_ticks` */
+    int64_t send_end_timestamp_ns;
+    /* The time duration for the request from start encoding to finish encoding (send_end_timestamp_ns -
+     * send_start_timestamp_ns). -1 means data not available. */
+    int64_t sending_duration_ns;
+
+    /* The time stamp when the response started to be received from the network channel. -1 means data not available.
+     * Timestamp are from `aws_high_res_clock_get_ticks` */
+    int64_t receive_start_timestamp_ns;
+    /* The time stamp when the response finished to be received from the network channel. -1 means data not available.
+     * Timestamp are from `aws_high_res_clock_get_ticks` */
+    int64_t receive_end_timestamp_ns;
+    /* The time duration for the request from start receiving to finish receiving. receive_end_timestamp_ns -
+     * receive_start_timestamp_ns. -1 means data not available. */
+    int64_t receiving_duration_ns;
+
+    /* The stream-id on the connection when this stream was activated. */
+    uint32_t stream_id;
+};
+
+/**
+ * Invoked right before request/response stream is complete to report the tracing metrics for aws_http_stream.
+ * This may be invoked synchronously when aws_http_stream_release() is called.
+ * This is invoked even if the stream is never activated.
+ * See `aws_http_stream_metrics` for details.
+ */
+typedef void(aws_http_on_stream_metrics_fn)(
+    struct aws_http_stream *stream,
+    const struct aws_http_stream_metrics *metrics,
+    void *user_data);
+
 /**
  * Options for creating a stream which sends a request from the client and receives a response from the server.
  */
@@ -234,6 +274,13 @@ struct aws_http_make_request_options {
      */
     aws_http_on_incoming_body_fn *on_response_body;
 
+    /**
+     * Invoked right before stream is complete, whether successful or unsuccessful
+     * Optional.
+     * See `aws_http_on_stream_metrics_fn`
+     */
+    aws_http_on_stream_metrics_fn *on_metrics;
+
     /**
      * Invoked when request/response stream is complete, whether successful or unsuccessful
      * Optional.
@@ -972,6 +1019,12 @@ AWS_HTTP_API
 struct aws_http_stream *aws_http_stream_new_server_request_handler(
     const struct aws_http_request_handler_options *options);
 
+/**
+ * Acquire refcount on the stream to prevent it from being cleaned up until it is released.
+ */
+AWS_HTTP_API
+struct aws_http_stream *aws_http_stream_acquire(struct aws_http_stream *stream);
+
 /**
  * Users must release the stream when they are done with it, or its memory will never be cleaned up.
  * This will not cancel the stream, its callbacks will still fire if the stream is still in progress.

source/connection.c

@@ -18,6 +18,7 @@
 #include <aws/io/channel_bootstrap.h>
 #include <aws/io/logging.h>
 #include <aws/io/socket.h>
+#include <aws/io/socket_channel_handler.h>
 #include <aws/io/tls_channel_handler.h>
 
 #ifdef _MSC_VER
@@ -366,6 +367,16 @@ struct aws_channel *aws_http_connection_get_channel(struct aws_http_connection *
     return connection->channel_slot->channel;
 }
 
+const struct aws_socket_endpoint *aws_http_connection_get_remote_endpoint(
+    const struct aws_http_connection *connection) {
+    AWS_ASSERT(connection);
+    struct aws_channel *channel = connection->channel_slot->channel;
+    /* The first slot for an HTTP connection is always socket */
+    struct aws_channel_slot *socket_slot = aws_channel_get_first_slot(channel);
+    const struct aws_socket *socket = aws_socket_handler_get_socket(socket_slot->handler);
+    return &socket->remote_endpoint;
+}
+
 int aws_http_alpn_map_init(struct aws_allocator *allocator, struct aws_hash_table *map) {
     AWS_ASSERT(allocator);
     AWS_ASSERT(map);

source/h1_connection.c

@@ -371,6 +371,7 @@ int aws_h1_stream_activate(struct aws_http_stream *stream) {
 
     /* connection keeps activated stream alive until stream completes */
     aws_atomic_fetch_add(&stream->refcount, 1);
+    stream->metrics.stream_id = stream->id;
 
     if (should_schedule_task) {
         AWS_LOGF_TRACE(
@@ -633,6 +634,10 @@ static void s_stream_complete(struct aws_h1_stream *stream, int error_code) {
         aws_h1_chunk_complete_and_destroy(chunk, &stream->base, AWS_ERROR_HTTP_STREAM_HAS_COMPLETED);
     }
 
+    if (stream->base.on_metrics) {
+        stream->base.on_metrics(&stream->base, &stream->base.metrics, stream->base.user_data);
+    }
+
     /* Invoke callback and clean up stream. */
     if (stream->base.on_complete) {
         stream->base.on_complete(&stream->base, error_code, stream->base.user_data);
@@ -731,6 +736,12 @@ static struct aws_h1_stream *s_update_outgoing_stream_ptr(struct aws_h1_connecti
     /* If current stream is done sending data... */
     if (current && !aws_h1_encoder_is_message_in_progress(&connection->thread_data.encoder)) {
         current->is_outgoing_message_done = true;
+        AWS_ASSERT(current->base.metrics.send_end_timestamp_ns == -1);
+        aws_high_res_clock_get_ticks((uint64_t *)&current->base.metrics.send_end_timestamp_ns);
+        AWS_ASSERT(current->base.metrics.send_start_timestamp_ns != -1);
+        AWS_ASSERT(current->base.metrics.send_end_timestamp_ns >= current->base.metrics.send_start_timestamp_ns);
+        current->base.metrics.sending_duration_ns =
+            current->base.metrics.send_end_timestamp_ns - current->base.metrics.send_start_timestamp_ns;
 
         /* RFC-7230 section 6.6: Tear-down.
          * If this was the final stream, don't allows any further streams to be sent */
@@ -801,9 +812,13 @@ static struct aws_h1_stream *s_update_outgoing_stream_ptr(struct aws_h1_connecti
         s_set_outgoing_stream_ptr(connection, current);
 
         if (current) {
+            AWS_ASSERT(current->base.metrics.send_start_timestamp_ns == -1);
+            aws_high_res_clock_get_ticks((uint64_t *)&current->base.metrics.send_start_timestamp_ns);
+
             err = aws_h1_encoder_start_message(
                 &connection->thread_data.encoder, &current->encoder_message, &current->base);
             (void)err;
+            AWS_ASSERT(connection->thread_data.encoder.state == AWS_H1_ENCODER_STATE_INIT);
             AWS_ASSERT(!err);
         }
 
@@ -1110,6 +1125,15 @@ static int s_decoder_on_header(const struct aws_h1_decoded_header *header, void
                         "id=%p: Received 'Connection: close' header, no more request data will be sent.",
                         (void *)&incoming_stream->base);
                     incoming_stream->is_outgoing_message_done = true;
+                    AWS_ASSERT(incoming_stream->base.metrics.send_end_timestamp_ns == -1);
+                    aws_high_res_clock_get_ticks((uint64_t *)&incoming_stream->base.metrics.send_end_timestamp_ns);
+                    AWS_ASSERT(incoming_stream->base.metrics.send_start_timestamp_ns != -1);
+                    AWS_ASSERT(
+                        incoming_stream->base.metrics.send_end_timestamp_ns >=
+                        incoming_stream->base.metrics.send_start_timestamp_ns);
+                    incoming_stream->base.metrics.sending_duration_ns =
+                        incoming_stream->base.metrics.send_end_timestamp_ns -
+                        incoming_stream->base.metrics.send_start_timestamp_ns;
                 }
                 /* Stop writing right now.
                  * Shutdown will be scheduled after we finishing parsing the response */
@@ -1270,6 +1294,13 @@ static int s_decoder_on_done(void *user_data) {
 
     /* Otherwise the incoming stream is finished decoding and we will update it if needed */
     incoming_stream->is_incoming_message_done = true;
+    aws_high_res_clock_get_ticks((uint64_t *)&incoming_stream->base.metrics.receive_end_timestamp_ns);
+    AWS_ASSERT(incoming_stream->base.metrics.receive_start_timestamp_ns != -1);
+    AWS_ASSERT(
+        incoming_stream->base.metrics.receive_end_timestamp_ns >=
+        incoming_stream->base.metrics.receive_start_timestamp_ns);
+    incoming_stream->base.metrics.receiving_duration_ns = incoming_stream->base.metrics.receive_end_timestamp_ns -
+                                                          incoming_stream->base.metrics.receive_start_timestamp_ns;
 
     /* RFC-7230 section 6.6
      * After reading the final message, the connection must not read any more */
@@ -1822,6 +1853,11 @@ static int s_try_process_next_stream_read_message(struct aws_h1_connection *conn
     bool body_headers_ignored = incoming_stream->base.request_method == AWS_HTTP_METHOD_HEAD;
     aws_h1_decoder_set_body_headers_ignored(connection->thread_data.incoming_stream_decoder, body_headers_ignored);
 
+    if (incoming_stream->base.metrics.receive_start_timestamp_ns == -1) {
+        /* That's the first time for the stream receives any message */
+        aws_high_res_clock_get_ticks((uint64_t *)&incoming_stream->base.metrics.receive_start_timestamp_ns);
+    }
+
     /* As decoder runs, it invokes the internal s_decoder_X callbacks, which in turn invoke user callbacks.
      * The decoder will stop once it hits the end of the request/response OR the end of the message data. */
     if (aws_h1_decode(connection->thread_data.incoming_stream_decoder, &message_cursor)) {

source/h1_stream.c

@@ -361,6 +361,12 @@ static struct aws_h1_stream *s_stream_new_common(
     stream->base.on_incoming_body = on_incoming_body;
     stream->base.on_complete = on_complete;
     stream->base.on_destroy = on_destroy;
+    stream->base.metrics.send_start_timestamp_ns = -1;
+    stream->base.metrics.send_end_timestamp_ns = -1;
+    stream->base.metrics.sending_duration_ns = -1;
+    stream->base.metrics.receive_start_timestamp_ns = -1;
+    stream->base.metrics.receive_end_timestamp_ns = -1;
+    stream->base.metrics.receiving_duration_ns = -1;
 
     aws_channel_task_init(
         &stream->cross_thread_work_task, s_stream_cross_thread_work_task, stream, "http1_stream_cross_thread_work");
@@ -401,6 +407,7 @@ struct aws_h1_stream *aws_h1_stream_new_request(
 
     stream->base.client_data = &stream->base.client_or_server_data.client;
     stream->base.client_data->response_status = AWS_HTTP_STATUS_CODE_UNKNOWN;
+    stream->base.on_metrics = options->on_metrics;
 
     /* Validate request and cache info that the encoder will eventually need */
     if (aws_h1_encoder_message_init_from_request(

source/h2_connection.c

@@ -2057,6 +2057,7 @@ int aws_h2_stream_activate(struct aws_http_stream *stream) {
 
     /* connection keeps activated stream alive until stream completes */
     aws_atomic_fetch_add(&stream->refcount, 1);
+    stream->metrics.stream_id = stream->id;
 
     if (!was_cross_thread_work_scheduled) {
         CONNECTION_LOG(TRACE, connection, "Scheduling cross-thread work task");

source/h2_stream.c

@@ -5,6 +5,7 @@
 
 #include <aws/http/private/h2_stream.h>
 
+#include <aws/common/clock.h>
 #include <aws/http/private/h2_connection.h>
 #include <aws/http/private/strutil.h>
 #include <aws/http/status_code.h>
@@ -240,10 +241,17 @@ struct aws_h2_stream *aws_h2_stream_new_request(
     stream->base.on_incoming_headers = options->on_response_headers;
     stream->base.on_incoming_header_block_done = options->on_response_header_block_done;
     stream->base.on_incoming_body = options->on_response_body;
+    stream->base.on_metrics = options->on_metrics;
     stream->base.on_complete = options->on_complete;
     stream->base.on_destroy = options->on_destroy;
     stream->base.client_data = &stream->base.client_or_server_data.client;
     stream->base.client_data->response_status = AWS_HTTP_STATUS_CODE_UNKNOWN;
+    stream->base.metrics.send_start_timestamp_ns = -1;
+    stream->base.metrics.send_end_timestamp_ns = -1;
+    stream->base.metrics.sending_duration_ns = -1;
+    stream->base.metrics.receive_start_timestamp_ns = -1;
+    stream->base.metrics.receive_end_timestamp_ns = -1;
+    stream->base.metrics.receiving_duration_ns = -1;
     aws_linked_list_init(&stream->thread_data.outgoing_writes);
     aws_linked_list_init(&stream->synced_data.pending_write_list);
 
@@ -446,6 +454,9 @@ void aws_h2_stream_complete(struct aws_h2_stream *stream, int error_code) {
     s_h2_stream_destroy_pending_writes(stream);
 
     /* Invoke callback */
+    if (stream->base.on_metrics) {
+        stream->base.on_metrics(&stream->base, &stream->base.metrics, stream->base.user_data);
+    }
     if (stream->base.on_complete) {
         stream->base.on_complete(&stream->base, error_code, stream->base.user_data);
     }
@@ -706,7 +717,8 @@ int aws_h2_stream_on_activated(struct aws_h2_stream *stream, enum aws_h2_stream_
         AWS_H2_STREAM_LOGF(ERROR, stream, "Failed to create HEADERS frame: %s", aws_error_name(aws_last_error()));
         goto error;
     }
-
+    AWS_ASSERT(stream->base.metrics.send_start_timestamp_ns == -1);
+    aws_high_res_clock_get_ticks((uint64_t *)&stream->base.metrics.send_start_timestamp_ns);
     /* Initialize the flow-control window size */
     stream->thread_data.window_size_peer =
         connection->thread_data.settings_peer[AWS_HTTP2_SETTINGS_INITIAL_WINDOW_SIZE];
@@ -721,6 +733,11 @@ int aws_h2_stream_on_activated(struct aws_h2_stream *stream, enum aws_h2_stream_
         /* If stream has no body, then HEADERS frame marks the end of outgoing data */
         stream->thread_data.state = AWS_H2_STREAM_STATE_HALF_CLOSED_LOCAL;
         AWS_H2_STREAM_LOG(TRACE, stream, "Sending HEADERS with END_STREAM. State -> HALF_CLOSED_LOCAL");
+        /* There is no further frames to be sent, now is the end timestamp of sending. */
+        AWS_ASSERT(stream->base.metrics.send_end_timestamp_ns == -1);
+        aws_high_res_clock_get_ticks((uint64_t *)&stream->base.metrics.send_end_timestamp_ns);
+        stream->base.metrics.sending_duration_ns =
+            stream->base.metrics.send_end_timestamp_ns - stream->base.metrics.send_start_timestamp_ns;
     }
 
     if (s_h2_stream_has_outgoing_writes(stream)) {
@@ -798,6 +815,11 @@ int aws_h2_stream_encode_data_frame(
      */
     if (input_stream_complete && ends_stream) {
         /* Done sending data. No more data will be sent. */
+        AWS_ASSERT(stream->base.metrics.send_end_timestamp_ns == -1);
+        aws_high_res_clock_get_ticks((uint64_t *)&stream->base.metrics.send_end_timestamp_ns);
+        stream->base.metrics.sending_duration_ns =
+            stream->base.metrics.send_end_timestamp_ns - stream->base.metrics.send_start_timestamp_ns;
+
         if (stream->thread_data.state == AWS_H2_STREAM_STATE_HALF_CLOSED_REMOTE) {
             /* Both sides have sent END_STREAM */
             stream->thread_data.state = AWS_H2_STREAM_STATE_CLOSED;
@@ -841,6 +863,7 @@ struct aws_h2err aws_h2_stream_on_decoder_headers_begin(struct aws_h2_stream *st
     if (aws_h2err_failed(stream_err)) {
         return s_send_rst_and_close_stream(stream, stream_err);
     }
+    aws_high_res_clock_get_ticks((uint64_t *)&stream->base.metrics.receive_start_timestamp_ns);
 
     return AWS_H2ERR_SUCCESS;
 }
@@ -1150,6 +1173,13 @@ struct aws_h2err aws_h2_stream_on_decoder_end_stream(struct aws_h2_stream *strea
      * an actual frame type. It's a flag on DATA or HEADERS frames, and we
      * already checked the legality of those frames in their respective callbacks. */
 
+    AWS_ASSERT(stream->base.metrics.receive_start_timestamp_ns != -1);
+    AWS_ASSERT(stream->base.metrics.receive_end_timestamp_ns == -1);
+    aws_high_res_clock_get_ticks((uint64_t *)&stream->base.metrics.receive_end_timestamp_ns);
+    AWS_ASSERT(stream->base.metrics.receive_end_timestamp_ns >= stream->base.metrics.receive_start_timestamp_ns);
+    stream->base.metrics.receiving_duration_ns =
+        stream->base.metrics.receive_end_timestamp_ns - stream->base.metrics.receive_start_timestamp_ns;
+
     if (stream->thread_data.content_length_received) {
         if (stream->base.request_method != AWS_HTTP_METHOD_HEAD &&
             stream->base.client_data->response_status != AWS_HTTP_STATUS_CODE_304_NOT_MODIFIED) {

source/request_response.c

@@ -1107,6 +1107,15 @@ int aws_http_stream_send_response(struct aws_http_stream *stream, struct aws_htt
     return stream->owning_connection->vtable->stream_send_response(stream, response);
 }
 
+struct aws_http_stream *aws_http_stream_acquire(struct aws_http_stream *stream) {
+    AWS_PRECONDITION(stream);
+
+    size_t prev_refcount = aws_atomic_fetch_add(&stream->refcount, 1);
+    AWS_LOGF_TRACE(
+        AWS_LS_HTTP_STREAM, "id=%p: Stream refcount acquired, %zu remaining.", (void *)stream, prev_refcount + 1);
+    return stream;
+}
+
 void aws_http_stream_release(struct aws_http_stream *stream) {
     if (!stream) {
         return;