Add caught time stamp for http tap filter (envoyproxy#28472)

Signed-off-by: fchen7 <[email protected]>

api/envoy/data/tap/v3/http.proto

@@ -5,6 +5,8 @@ package envoy.data.tap.v3;
 import "envoy/config/core/v3/base.proto";
 import "envoy/data/tap/v3/common.proto";
 
+import "google/protobuf/timestamp.proto";
+
 import "udpa/annotations/status.proto";
 import "udpa/annotations/versioning.proto";
 
@@ -34,6 +36,9 @@ message HttpBufferedTrace {
 
     // Message trailers.
     repeated config.core.v3.HeaderValue trailers = 3;
+
+    // The timestamp after receiving the message headers.
+    google.protobuf.Timestamp headers_received_time = 4;
   }
 
   // Request message.

api/envoy/extensions/filters/http/tap/v3/tap.proto

@@ -26,4 +26,9 @@ message Tap {
   // Common configuration for the HTTP tap filter.
   common.tap.v3.CommonExtensionConfig common_config = 1
       [(validate.rules).message = {required: true}];
+
+  // Indicates whether HTTP tap filter records the time stamp for request/response headers.
+  // Request headers time stamp is stored after receiving request headers.
+  // Response headers time stamp is stored after receiving response headers.
+  bool record_headers_received_time = 2;
 }

changelogs/current.yaml

@@ -109,6 +109,10 @@ new_features:
   change: |
     added :ref:`custom_sink <envoy_v3_api_field_config.tap.v3.OutputSink.custom_sink>` type to enable writing tap data
     out to a custom sink extension.
+- area: tap
+  change: |
+    added :ref:`record_headers_received_time <envoy_v3_api_field_extensions.filters.http.tap.v3.Tap.record_headers_received_time>`
+    to control writing request and response headers received time in trace output.
 
 deprecated:
 - area: tracing

source/extensions/common/tap/BUILD

@@ -17,6 +17,7 @@ envoy_cc_library(
         "//source/extensions/common/matcher:matcher_lib",
         "@envoy_api//envoy/config/tap/v3:pkg_cc_proto",
         "@envoy_api//envoy/data/tap/v3:pkg_cc_proto",
+        "@envoy_api//envoy/extensions/filters/http/tap/v3:pkg_cc_proto",
     ],
 )
 

source/extensions/common/tap/tap.h

@@ -3,6 +3,7 @@
 #include "envoy/common/pure.h"
 #include "envoy/config/tap/v3/common.pb.h"
 #include "envoy/data/tap/v3/wrapper.pb.h"
+#include "envoy/extensions/filters/http/tap/v3/tap.pb.h"
 #include "envoy/http/header_map.h"
 
 #include "source/extensions/common/matcher/matcher.h"

source/extensions/filters/http/tap/BUILD

@@ -20,6 +20,7 @@ envoy_cc_library(
         "//envoy/buffer:buffer_interface",
         "//envoy/http:header_map_interface",
         "//source/extensions/common/tap:tap_interface",
+        "@envoy_api//envoy/extensions/filters/http/tap/v3:pkg_cc_proto",
     ],
 )
 
@@ -33,6 +34,7 @@ envoy_cc_library(
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
         "@envoy_api//envoy/config/tap/v3:pkg_cc_proto",
         "@envoy_api//envoy/data/tap/v3:pkg_cc_proto",
+        "@envoy_api//envoy/extensions/filters/http/tap/v3:pkg_cc_proto",
     ],
 )
 

source/extensions/filters/http/tap/tap_config.h

@@ -2,6 +2,7 @@
 
 #include "envoy/buffer/buffer.h"
 #include "envoy/common/pure.h"
+#include "envoy/extensions/filters/http/tap/v3/tap.pb.h"
 #include "envoy/http/header_map.h"
 
 #include "source/extensions/common/tap/tap.h"
@@ -65,9 +66,17 @@ class HttpTapConfig : public virtual Extensions::Common::Tap::TapConfig {
 public:
   /**
    * @return a new per-request HTTP tapper which is used to handle tapping of a discrete request.
+   * @param tap_config provides http tap config
    * @param stream_id supplies the owning HTTP stream ID.
    */
-  virtual HttpPerRequestTapperPtr createPerRequestTapper(uint64_t stream_id) PURE;
+  virtual HttpPerRequestTapperPtr
+  createPerRequestTapper(const envoy::extensions::filters::http::tap::v3::Tap& tap_config,
+                         uint64_t stream_id) PURE;
+
+  /**
+   * @return time source to use for timestamp
+   */
+  virtual TimeSource& timeSource() const PURE;
 };
 
 using HttpTapConfigSharedPtr = std::shared_ptr<HttpTapConfig>;

source/extensions/filters/http/tap/tap_config_impl.cc

@@ -31,10 +31,12 @@ fillHeaderList(Protobuf::RepeatedPtrField<envoy::config::core::v3::HeaderValue>*
 HttpTapConfigImpl::HttpTapConfigImpl(const envoy::config::tap::v3::TapConfig& proto_config,
                                      Common::Tap::Sink* admin_streamer,
                                      Server::Configuration::FactoryContext& context)
-    : TapCommon::TapConfigBaseImpl(std::move(proto_config), admin_streamer, context) {}
+    : TapCommon::TapConfigBaseImpl(std::move(proto_config), admin_streamer, context),
+      time_source_(context.mainThreadDispatcher().timeSource()) {}
 
-HttpPerRequestTapperPtr HttpTapConfigImpl::createPerRequestTapper(uint64_t stream_id) {
-  return std::make_unique<HttpPerRequestTapperImpl>(shared_from_this(), stream_id);
+HttpPerRequestTapperPtr HttpTapConfigImpl::createPerRequestTapper(
+    const envoy::extensions::filters::http::tap::v3::Tap& tap_config, uint64_t stream_id) {
+  return std::make_unique<HttpPerRequestTapperImpl>(shared_from_this(), tap_config, stream_id);
 }
 
 void HttpPerRequestTapperImpl::streamRequestHeaders() {
@@ -46,6 +48,10 @@ void HttpPerRequestTapperImpl::streamRequestHeaders() {
 
 void HttpPerRequestTapperImpl::onRequestHeaders(const Http::RequestHeaderMap& headers) {
   request_headers_ = &headers;
+  if (should_record_headers_received_time_) {
+    setTimeStamp(request_headers_received_time_);
+  }
+
   config_->rootMatcher().onHttpRequestHeaders(headers, statuses_);
   if (config_->streaming() && config_->rootMatcher().matchStatus(statuses_).matches_) {
     ASSERT(!started_streaming_trace_);
@@ -101,6 +107,10 @@ void HttpPerRequestTapperImpl::streamResponseHeaders() {
 
 void HttpPerRequestTapperImpl::onResponseHeaders(const Http::ResponseHeaderMap& headers) {
   response_headers_ = &headers;
+  if (should_record_headers_received_time_) {
+    setTimeStamp(response_headers_received_time_);
+  }
+
   config_->rootMatcher().onHttpResponseHeaders(headers, statuses_);
   if (config_->streaming() && config_->rootMatcher().matchStatus(statuses_).matches_) {
     if (!started_streaming_trace_) {
@@ -159,12 +169,20 @@ bool HttpPerRequestTapperImpl::onDestroyLog() {
   auto& http_trace = *buffered_full_trace_->mutable_http_buffered_trace();
   if (request_headers_ != nullptr) {
     request_headers_->iterate(fillHeaderList(http_trace.mutable_request()->mutable_headers()));
+    if (should_record_headers_received_time_) {
+      http_trace.mutable_request()->mutable_headers_received_time()->MergeFrom(
+          Protobuf::util::TimeUtil::NanosecondsToTimestamp(request_headers_received_time_));
+    }
   }
   if (request_trailers_ != nullptr) {
     request_trailers_->iterate(fillHeaderList(http_trace.mutable_request()->mutable_trailers()));
   }
   if (response_headers_ != nullptr) {
     response_headers_->iterate(fillHeaderList(http_trace.mutable_response()->mutable_headers()));
+    if (should_record_headers_received_time_) {
+      http_trace.mutable_response()->mutable_headers_received_time()->MergeFrom(
+          Protobuf::util::TimeUtil::NanosecondsToTimestamp(response_headers_received_time_));
+    }
   }
   if (response_trailers_ != nullptr) {
     response_trailers_->iterate(fillHeaderList(http_trace.mutable_response()->mutable_trailers()));

source/extensions/filters/http/tap/tap_config_impl.h

@@ -3,6 +3,7 @@
 #include "envoy/config/tap/v3/common.pb.h"
 #include "envoy/data/tap/v3/common.pb.h"
 #include "envoy/data/tap/v3/http.pb.h"
+#include "envoy/extensions/filters/http/tap/v3/tap.pb.h"
 #include "envoy/http/header_map.h"
 
 #include "source/common/common/logger.h"
@@ -23,14 +24,24 @@ class HttpTapConfigImpl : public Extensions::Common::Tap::TapConfigBaseImpl,
                     Server::Configuration::FactoryContext& context);
 
   // TapFilter::HttpTapConfig
-  HttpPerRequestTapperPtr createPerRequestTapper(uint64_t stream_id) override;
+  HttpPerRequestTapperPtr
+  createPerRequestTapper(const envoy::extensions::filters::http::tap::v3::Tap& tap_config,
+                         uint64_t stream_id) override;
+
+  TimeSource& timeSource() const override { return time_source_; }
+
+private:
+  TimeSource& time_source_;
 };
 
 class HttpPerRequestTapperImpl : public HttpPerRequestTapper, Logger::Loggable<Logger::Id::tap> {
 public:
-  HttpPerRequestTapperImpl(HttpTapConfigSharedPtr config, uint64_t stream_id)
-      : config_(std::move(config)), stream_id_(stream_id),
-        sink_handle_(config_->createPerTapSinkHandleManager(stream_id)),
+  HttpPerRequestTapperImpl(HttpTapConfigSharedPtr config,
+                           const envoy::extensions::filters::http::tap::v3::Tap& tap_config,
+                           uint64_t stream_id)
+      : config_(std::move(config)),
+        should_record_headers_received_time_(tap_config.record_headers_received_time()),
+        stream_id_(stream_id), sink_handle_(config_->createPerTapSinkHandleManager(stream_id)),
         statuses_(config_->createMatchStatusVector()) {
     config_->rootMatcher().onNewStream(statuses_);
   }
@@ -73,11 +84,21 @@ class HttpPerRequestTapperImpl : public HttpPerRequestTapper, Logger::Loggable<L
   void streamResponseHeaders();
   void streamBufferedResponseBody();
 
+  // Functions for request/response caught time stamp
+  void setTimeStamp(long& pTimeStamp) {
+    pTimeStamp = std::chrono::duration_cast<std::chrono::nanoseconds>(
+                     config_->timeSource().systemTime().time_since_epoch())
+                     .count();
+  }
+
   HttpTapConfigSharedPtr config_;
+  const bool should_record_headers_received_time_;
   const uint64_t stream_id_;
   Extensions::Common::Tap::PerTapSinkHandleManagerPtr sink_handle_;
   Extensions::Common::Tap::Matcher::MatchStatusVector statuses_;
   bool started_streaming_trace_{};
+  long request_headers_received_time_;
+  long response_headers_received_time_;
   const Http::RequestHeaderMap* request_headers_{};
   const Http::HeaderMap* request_trailers_{};
   const Http::ResponseHeaderMap* response_headers_{};

source/extensions/filters/http/tap/tap_filter.cc

@@ -14,7 +14,7 @@ FilterConfigImpl::FilterConfigImpl(
     ThreadLocal::SlotAllocator& tls, Event::Dispatcher& main_thread_dispatcher)
     : ExtensionConfigBase(proto_config.common_config(), std::move(config_factory), admin,
                           singleton_manager, tls, main_thread_dispatcher),
-      stats_(Filter::generateStats(stats_prefix, scope)) {}
+      stats_(Filter::generateStats(stats_prefix, scope)), tap_config_(proto_config) {}
 
 HttpTapConfigSharedPtr FilterConfigImpl::currentConfig() {
   return currentConfigHelper<HttpTapConfig>();

source/extensions/filters/http/tap/tap_filter.h

@@ -43,6 +43,11 @@ class FilterConfig {
    */
   virtual HttpTapConfigSharedPtr currentConfig() PURE;
 
+  /**
+   * @return the http tap config.
+   */
+  virtual const envoy::extensions::filters::http::tap::v3::Tap& getTapConfig() const PURE;
+
   /**
    * @return the filter stats.
    */
@@ -66,9 +71,13 @@ class FilterConfigImpl : public FilterConfig, public Extensions::Common::Tap::Ex
   // FilterConfig
   HttpTapConfigSharedPtr currentConfig() override;
   FilterStats& stats() override { return stats_; }
+  const envoy::extensions::filters::http::tap::v3::Tap& getTapConfig() const override {
+    return tap_config_;
+  }
 
 private:
   FilterStats stats_;
+  const envoy::extensions::filters::http::tap::v3::Tap tap_config_;
 };
 
 /**
@@ -90,7 +99,8 @@ class Filter : public Http::StreamFilter, public AccessLog::Instance {
   Http::FilterTrailersStatus decodeTrailers(Http::RequestTrailerMap& trailers) override;
   void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override {
     HttpTapConfigSharedPtr config = config_->currentConfig();
-    tapper_ = config ? config->createPerRequestTapper(callbacks.streamId()) : nullptr;
+    tapper_ = config ? config->createPerRequestTapper(config_->getTapConfig(), callbacks.streamId())
+                     : nullptr;
   }
 
   // Http::StreamEncoderFilter

test/extensions/filters/http/tap/BUILD

@@ -45,6 +45,7 @@ envoy_extension_cc_test(
         "//source/extensions/filters/http/tap:tap_config_impl",
         "//test/extensions/common/tap:common",
         "//test/mocks:common_lib",
+        "//test/test_common:simulated_time_system_lib",
         "//test/test_common:utility_lib",
     ],
 )

test/extensions/filters/http/tap/common.h

@@ -11,8 +11,10 @@ namespace TapFilter {
 
 class MockHttpTapConfig : public HttpTapConfig {
 public:
-  HttpPerRequestTapperPtr createPerRequestTapper(uint64_t stream_id) override {
-    return HttpPerRequestTapperPtr{createPerRequestTapper_(stream_id)};
+  HttpPerRequestTapperPtr
+  createPerRequestTapper(const envoy::extensions::filters::http::tap::v3::Tap& tap_config,
+                         uint64_t stream_id) override {
+    return HttpPerRequestTapperPtr{createPerRequestTapper_(tap_config, stream_id)};
   }
 
   Extensions::Common::Tap::PerTapSinkHandleManagerPtr
@@ -21,7 +23,9 @@ class MockHttpTapConfig : public HttpTapConfig {
         createPerTapSinkHandleManager_(trace_id)};
   }
 
-  MOCK_METHOD(HttpPerRequestTapper*, createPerRequestTapper_, (uint64_t stream_id));
+  MOCK_METHOD(HttpPerRequestTapper*, createPerRequestTapper_,
+              (const envoy::extensions::filters::http::tap::v3::Tap& tap_config,
+               uint64_t stream_id));
   MOCK_METHOD(Extensions::Common::Tap::PerTapSinkHandleManager*, createPerTapSinkHandleManager_,
               (uint64_t trace_id));
   MOCK_METHOD(uint32_t, maxBufferedRxBytes, (), (const));
@@ -30,6 +34,7 @@ class MockHttpTapConfig : public HttpTapConfig {
               (const));
   MOCK_METHOD(const Extensions::Common::Tap::Matcher&, rootMatcher, (), (const));
   MOCK_METHOD(bool, streaming, (), (const));
+  MOCK_METHOD(TimeSource&, timeSource, (), (const));
 };
 
 } // namespace TapFilter