Add access logging to tcp_proxy

fixes #1159

Signed-off-by: Greg Greenway <ggreenway@apple.com>

include/envoy/http/access_log.h

@@ -69,7 +69,7 @@ class RequestInfo {
    * @return duration from request start to when the entire request was received from the
    * downstream client in microseconds. Note: if unset, will return 0 microseconds.
    */
-  virtual std::chrono::microseconds requestReceivedDuration() const PURE;
+  virtual Optional<std::chrono::microseconds> requestReceivedDuration() const PURE;
 
   /**
    * Set the duration from request start to when the entire request was received from the
@@ -82,7 +82,7 @@ class RequestInfo {
    * @return the duration from request start to when the entire response was received from the
    * upstream host in microseconds. Note: if unset, will return 0 microseconds.
    */
-  virtual std::chrono::microseconds responseReceivedDuration() const PURE;
+  virtual Optional<std::chrono::microseconds> responseReceivedDuration() const PURE;
 
   /**
    * Set the duration from request start to when the entire response was received from the
@@ -99,7 +99,7 @@ class RequestInfo {
   /**
    * @return the protocol of the request.
    */
-  virtual Protocol protocol() const PURE;
+  virtual Optional<Protocol> protocol() const PURE;
 
   /**
    * Set the request's protocol.

source/common/filter/BUILD

@@ -45,9 +45,11 @@ envoy_cc_library(
         "//include/envoy/buffer:buffer_interface",
         "//include/envoy/event:dispatcher_interface",
         "//include/envoy/event:timer_interface",
+        "//include/envoy/http:access_log_interface",
         "//include/envoy/network:connection_interface",
         "//include/envoy/network:filter_interface",
         "//include/envoy/router:router_interface",
+        "//include/envoy/server:filter_config_interface",
         "//include/envoy/stats:stats_interface",
         "//include/envoy/stats:stats_macros",
         "//include/envoy/stats:timespan",
@@ -56,6 +58,9 @@ envoy_cc_library(
         "//source/common/common:assert_lib",
         "//source/common/common:empty_string",
         "//source/common/common:logger_lib",
+        "//source/common/config:filter_json_lib",
+        "//source/common/http/access_log:access_log_lib",
+        "//source/common/http/access_log:request_info_lib",
         "//source/common/json:config_schemas_lib",
         "//source/common/json:json_loader_lib",
         "//source/common/network:cidr_range_lib",

source/common/filter/tcp_proxy.cc

@@ -12,9 +12,12 @@
 
 #include "common/common/assert.h"
 #include "common/common/empty_string.h"
+#include "common/config/filter_json.h"
+#include "common/http/access_log/access_log_impl.h"
 #include "common/json/config_schemas.h"
 #include "common/json/json_loader.h"
 
+#include "api/filter/http_connection_manager.pb.h"
 #include "fmt/format.h"
 
 namespace Envoy {
@@ -43,19 +46,25 @@ TcpProxyConfig::Route::Route(const Json::Object& config) {
 }
 
 TcpProxyConfig::TcpProxyConfig(const Json::Object& config,
-                               Upstream::ClusterManager& cluster_manager, Stats::Scope& scope)
-    : stats_(generateStats(config.getString("stat_prefix"), scope)) {
+                               Server::Configuration::FactoryContext& context)
+    : stats_(generateStats(config.getString("stat_prefix"), context.scope())) {
   config.validateSchema(Json::Schema::TCP_PROXY_NETWORK_FILTER_SCHEMA);
 
   for (const Json::ObjectSharedPtr& route_desc :
        config.getObject("route_config")->getObjectArray("routes")) {
     routes_.emplace_back(Route(*route_desc));
 
-    if (!cluster_manager.get(route_desc->getString("cluster"))) {
+    if (!context.clusterManager().get(route_desc->getString("cluster"))) {
       throw EnvoyException(fmt::format("tcp proxy: unknown cluster '{}' in TCP route",
                                        route_desc->getString("cluster")));
     }
   }
+
+  for (const Json::ObjectSharedPtr& json_access_log : config.getObjectArray("access_log", true)) {
+    envoy::api::v2::filter::AccessLog v2_access_log;
+    Config::FilterJson::translateAccessLog(*json_access_log, v2_access_log);
+    access_logs_.emplace_back(Http::AccessLog::AccessLogFactory::fromProto(v2_access_log, context));
+  }
 }
 
 const std::string& TcpProxyConfig::getRouteFromEntries(Network::Connection& connection) {
@@ -93,6 +102,12 @@ TcpProxy::TcpProxy(TcpProxyConfigSharedPtr config, Upstream::ClusterManager& clu
       upstream_callbacks_(new UpstreamCallbacks(*this)) {}
 
 TcpProxy::~TcpProxy() {
+  if (config_ != nullptr) {
+    for (const auto& access_log : config_->accessLogs()) {
+      access_log->log(nullptr, nullptr, request_info_);
+    }
+  }
+
   if (upstream_connection_) {
     read_callbacks_->upstreamHost()->cluster().stats().upstream_cx_destroy_.inc();
     read_callbacks_->upstreamHost()->cluster().stats().upstream_cx_active_.dec();
@@ -193,12 +208,14 @@ Network::FilterStatus TcpProxy::initializeUpstreamConnection() {
     if (config_) {
       config_->stats().downstream_cx_no_route_.inc();
     }
+    request_info_.setResponseFlag(Http::AccessLog::ResponseFlag::NoRouteFound);
     onInitFailure();
     return Network::FilterStatus::StopIteration;
   }
 
   Upstream::ClusterInfoConstSharedPtr cluster = thread_local_cluster->info();
   if (!cluster->resourceManager(Upstream::ResourcePriority::Default).connections().canCreate()) {
+    request_info_.setResponseFlag(Http::AccessLog::ResponseFlag::UpstreamOverflow);
     cluster->stats().upstream_cx_overflow_.inc();
     onInitFailure();
     return Network::FilterStatus::StopIteration;
@@ -209,10 +226,12 @@ Network::FilterStatus TcpProxy::initializeUpstreamConnection() {
   upstream_connection_ = std::move(conn_info.connection_);
   read_callbacks_->upstreamHost(conn_info.host_description_);
   if (!upstream_connection_) {
+    request_info_.setResponseFlag(Http::AccessLog::ResponseFlag::NoHealthyUpstream);
     onInitFailure();
     return Network::FilterStatus::StopIteration;
   }
 
+  request_info_.onUpstreamHostSelected(conn_info.host_description_);
   onUpstreamHostReady();
   cluster->resourceManager(Upstream::ResourcePriority::Default).connections().inc();
   upstream_connection_->addReadFilter(upstream_callbacks_);
@@ -252,6 +271,7 @@ void TcpProxy::onConnectTimeout() {
 
 Network::FilterStatus TcpProxy::onData(Buffer::Instance& data) {
   ENVOY_CONN_LOG(trace, "received {} bytes", read_callbacks_->connection(), data.length());
+  request_info_.bytes_received_ += data.length();
   upstream_connection_->write(data);
   ASSERT(0 == data.length());
   return Network::FilterStatus::StopIteration;
@@ -270,6 +290,7 @@ void TcpProxy::onDownstreamEvent(Network::ConnectionEvent event) {
 }
 
 void TcpProxy::onUpstreamData(Buffer::Instance& data) {
+  request_info_.bytes_sent_ += data.length();
   read_callbacks_->connection().write(data);
   ASSERT(0 == data.length());
 }
@@ -285,6 +306,7 @@ void TcpProxy::onUpstreamEvent(Network::ConnectionEvent event) {
 
   if (event == Network::ConnectionEvent::RemoteClose) {
     if (connect_timeout_timer_) {
+      request_info_.setResponseFlag(Http::AccessLog::ResponseFlag::UpstreamConnectionFailure);
       read_callbacks_->upstreamHost()->cluster().stats().upstream_cx_connect_fail_.inc();
       read_callbacks_->upstreamHost()->stats().cx_connect_fail_.inc();
     }

source/common/filter/tcp_proxy.h

@@ -6,14 +6,17 @@
 #include <vector>
 
 #include "envoy/event/timer.h"
+#include "envoy/http/access_log.h"
 #include "envoy/network/connection.h"
 #include "envoy/network/filter.h"
+#include "envoy/server/filter_config.h"
 #include "envoy/stats/stats_macros.h"
 #include "envoy/stats/timespan.h"
 #include "envoy/upstream/cluster_manager.h"
 #include "envoy/upstream/upstream.h"
 
 #include "common/common/logger.h"
+#include "common/http/access_log/request_info_impl.h"
 #include "common/json/json_loader.h"
 #include "common/network/cidr_range.h"
 #include "common/network/filter_impl.h"
@@ -49,8 +52,7 @@ struct TcpProxyStats {
  */
 class TcpProxyConfig {
 public:
-  TcpProxyConfig(const Json::Object& config, Upstream::ClusterManager& cluster_manager,
-                 Stats::Scope& scope);
+  TcpProxyConfig(const Json::Object& config, Server::Configuration::FactoryContext& context);
 
   /**
    * Find out which cluster an upstream connection should be opened to based on the
@@ -63,6 +65,7 @@ class TcpProxyConfig {
   const std::string& getRouteFromEntries(Network::Connection& connection);
 
   const TcpProxyStats& stats() { return stats_; }
+  const std::vector<Http::AccessLog::InstanceSharedPtr>& accessLogs() { return access_logs_; }
 
 private:
   struct Route {
@@ -79,6 +82,7 @@ class TcpProxyConfig {
 
   std::vector<Route> routes_;
   const TcpProxyStats stats_;
+  std::vector<Http::AccessLog::InstanceSharedPtr> access_logs_;
 };
 
 typedef std::shared_ptr<TcpProxyConfig> TcpProxyConfigSharedPtr;
@@ -180,6 +184,7 @@ class TcpProxy : public Network::ReadFilter,
   Stats::TimespanPtr connected_timespan_;
   std::shared_ptr<UpstreamCallbacks> upstream_callbacks_; // shared_ptr required for passing as a
                                                           // read filter.
+  Http::AccessLog::RequestInfoImpl request_info_;
 };
 
 } // Filter

source/common/grpc/http1_bridge_filter.cc

@@ -28,7 +28,10 @@ Http::FilterHeadersStatus Http1BridgeFilter::decodeHeaders(Http::HeaderMap& head
     setupStatTracking(headers);
   }
 
-  if (decoder_callbacks_->requestInfo().protocol() != Http::Protocol::Http2 && grpc_request) {
+  Optional<Http::Protocol> protocol = decoder_callbacks_->requestInfo().protocol();
+  ASSERT(protocol.valid());
+  if (decoder_callbacks_->requestInfo().protocol().value() != Http::Protocol::Http2 &&
+      grpc_request) {
     do_bridging_ = true;
   }
 

source/common/http/access_log/access_log_formatter.cc

@@ -13,7 +13,9 @@ namespace Envoy {
 namespace Http {
 namespace AccessLog {
 
-const std::string ResponseFlagUtils::NONE = "-";
+static const std::string UnspecifiedValueString = "-";
+
+const std::string ResponseFlagUtils::NONE = UnspecifiedValueString;
 const std::string ResponseFlagUtils::FAILED_LOCAL_HEALTH_CHECK = "LH";
 const std::string ResponseFlagUtils::NO_HEALTHY_UPSTREAM = "UH";
 const std::string ResponseFlagUtils::UPSTREAM_REQUEST_TIMEOUT = "UT";
@@ -104,14 +106,18 @@ static const std::string Http10String = "HTTP/1.0";
 static const std::string Http11String = "HTTP/1.1";
 static const std::string Http2String = "HTTP/2";
 
-const std::string& AccessLogFormatUtils::protocolToString(Protocol protocol) {
-  switch (protocol) {
-  case Protocol::Http10:
-    return Http10String;
-  case Protocol::Http11:
-    return Http11String;
-  case Protocol::Http2:
-    return Http2String;
+const std::string& AccessLogFormatUtils::protocolToString(Optional<Protocol> protocol) {
+  if (protocol.valid()) {
+    switch (protocol.value()) {
+    case Protocol::Http10:
+      return Http10String;
+    case Protocol::Http11:
+      return Http11String;
+    case Protocol::Http2:
+      return Http2String;
+    }
+  } else {
+    return UnspecifiedValueString;
   }
 
   NOT_REACHED;
@@ -233,15 +239,23 @@ RequestInfoFormatter::RequestInfoFormatter(const std::string& field_name) {
     };
   } else if (field_name == "REQUEST_DURATION") {
     field_extractor_ = [](const RequestInfo& request_info) {
-      return std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(
-                                request_info.requestReceivedDuration())
-                                .count());
+      Optional<std::chrono::microseconds> duration = request_info.requestReceivedDuration();
+      if (duration.valid()) {
+        return std::to_string(
+            std::chrono::duration_cast<std::chrono::milliseconds>(duration.value()).count());
+      } else {
+        return UnspecifiedValueString;
+      }
     };
   } else if (field_name == "RESPONSE_DURATION") {
     field_extractor_ = [](const RequestInfo& request_info) {
-      return std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(
-                                request_info.responseReceivedDuration())
-                                .count());
+      Optional<std::chrono::microseconds> duration = request_info.responseReceivedDuration();
+      if (duration.valid()) {
+        return std::to_string(
+            std::chrono::duration_cast<std::chrono::milliseconds>(duration.value()).count());
+      } else {
+        return UnspecifiedValueString;
+      }
     };
   } else if (field_name == "BYTES_RECEIVED") {
     field_extractor_ = [](const RequestInfo& request_info) {
@@ -275,7 +289,7 @@ RequestInfoFormatter::RequestInfoFormatter(const std::string& field_name) {
       if (request_info.upstreamHost()) {
         return request_info.upstreamHost()->address()->asString();
       } else {
-        return std::string("-");
+        return UnspecifiedValueString;
       }
     };
   } else if (field_name == "UPSTREAM_CLUSTER") {
@@ -285,7 +299,7 @@ RequestInfoFormatter::RequestInfoFormatter(const std::string& field_name) {
         upstream_cluster_name = request_info.upstreamHost()->cluster().name();
       }
 
-      return upstream_cluster_name.empty() ? "-" : upstream_cluster_name;
+      return upstream_cluster_name.empty() ? UnspecifiedValueString : upstream_cluster_name;
     };
   } else {
     throw EnvoyException(fmt::format("Not supported field in RequestInfo: {}", field_name));
@@ -318,7 +332,7 @@ std::string HeaderFormatter::format(const HeaderMap& headers) const {
 
   std::string header_value_string;
   if (!header) {
-    header_value_string = "-";
+    header_value_string = UnspecifiedValueString;
   } else {
     header_value_string = header->value().c_str();
   }

source/common/http/access_log/access_log_formatter.h

@@ -54,7 +54,7 @@ class AccessLogFormatParser {
 class AccessLogFormatUtils {
 public:
   static FormatterPtr defaultAccessLogFormatter();
-  static const std::string& protocolToString(Protocol protocol);
+  static const std::string& protocolToString(Optional<Protocol> protocol);
 
 private:
   AccessLogFormatUtils();

source/common/http/access_log/request_info_impl.h

@@ -10,22 +10,24 @@ namespace Http {
 namespace AccessLog {
 
 struct RequestInfoImpl : public RequestInfo {
-  RequestInfoImpl(Protocol protocol)
-      : protocol_(protocol), start_time_(std::chrono::system_clock::now()),
+  RequestInfoImpl()
+      : start_time_(std::chrono::system_clock::now()),
         start_time_monotonic_(std::chrono::steady_clock::now()) {}
 
+  RequestInfoImpl(Protocol protocol) : RequestInfoImpl() { protocol_ = protocol; }
+
   // Http::AccessLog::RequestInfo
   SystemTime startTime() const override { return start_time_; }
 
-  std::chrono::microseconds requestReceivedDuration() const override {
+  Optional<std::chrono::microseconds> requestReceivedDuration() const override {
     return request_received_duration_;
   }
   void requestReceivedDuration(MonotonicTime time) override {
     request_received_duration_ =
         std::chrono::duration_cast<std::chrono::microseconds>(time - start_time_monotonic_);
   }
 
-  std::chrono::microseconds responseReceivedDuration() const override {
+  Optional<std::chrono::microseconds> responseReceivedDuration() const override {
     return request_received_duration_;
   }
   void responseReceivedDuration(MonotonicTime time) override {
@@ -35,7 +37,7 @@ struct RequestInfoImpl : public RequestInfo {
 
   uint64_t bytesReceived() const override { return bytes_received_; }
 
-  Protocol protocol() const override { return protocol_; }
+  Optional<Protocol> protocol() const override { return protocol_; }
   void protocol(Protocol protocol) override { protocol_ = protocol; }
 
   const Optional<uint32_t>& responseCode() const override { return response_code_; }
@@ -65,7 +67,7 @@ struct RequestInfoImpl : public RequestInfo {
 
   const std::string& getDownstreamAddress() const override { return downstream_address_; };
 
-  Protocol protocol_;
+  Optional<Protocol> protocol_;
   const SystemTime start_time_;
   const MonotonicTime start_time_monotonic_;
   std::chrono::microseconds request_received_duration_{};