listener: log on start of connections

api/envoy/config/listener/v3/listener.proto

@@ -54,7 +54,7 @@ message ListenerCollection {
   repeated xds.core.v3.CollectionEntry entries = 1;
 }
 
-// [#next-free-field: 37]
+// [#next-free-field: 38]
 message Listener {
   option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.Listener";
 
@@ -130,6 +130,11 @@ message Listener {
     core.v3.ConfigSource config_source = 2;
   }
 
+  message AccessLogOptions {
+    // If set to true, the access log is flushed when a TCP connection starts.
+    bool flush_access_log_on_start = 1;
+  }
+
   reserved 14, 23;
 
   // The unique name by which this listener is known. If no name is provided,
@@ -357,10 +362,16 @@ message Listener {
   //   is warned similar to macOS. It is left enabled for UDP with undefined behavior currently.
   google.protobuf.BoolValue enable_reuse_port = 29;
 
-  // Configuration for :ref:`access logs <arch_overview_access_logs>`
-  // emitted by this listener.
+  // Configuration for :ref:`access logs <arch_overview_access_logs>` emitted by this listener. By
+  // default, the log is flushed at the end of the TCP connections, or when the TCP connection is
+  // not established. See :ref:`access_log_options
+  // <envoy_v3_api_field_config.listener.v3.Listener.access_log_options>` for additional
+  // configuration.
   repeated accesslog.v3.AccessLog access_log = 22;
 
+  // Additional access log options for the listener.
+  AccessLogOptions access_log_options = 37;
+
   // The maximum length a tcp listener's pending connections queue can grow to. If no value is
   // provided net.core.somaxconn will be used on Linux and 128 otherwise.
   google.protobuf.UInt32Value tcp_backlog_size = 24;

api/envoy/data/accesslog/v3/accesslog.proto

@@ -36,6 +36,7 @@ enum AccessLogType {
   NotSet = 0;
   TcpUpstreamConnected = 1;
   TcpPeriodic = 2;
+  TcpConnectionStart = 14;
   TcpConnectionEnd = 3;
   DownstreamStart = 4;
   DownstreamPeriodic = 5;
@@ -47,6 +48,7 @@ enum AccessLogType {
   UdpTunnelUpstreamConnected = 11;
   UdpPeriodic = 12;
   UdpSessionEnd = 13;
+  NotConnected = 15;
 }
 
 message TCPAccessLogEntry {

changelogs/current.yaml

@@ -41,6 +41,9 @@ minor_behavior_changes:
   change: |
     Use mobile specific network observer registries to propagate network change signals. This behavior can be reverted by
     setting the runtime guard ``envoy.reloadable_features.mobile_use_network_observer_registry``.
+- area: logging
+  change: |
+    Listener access log entries now have their entry type set to ``TcpConnectionEnd`` or ``NotConnected``.
 
 bug_fixes:
 # *Changes expected to improve the state of the world and are unlikely to have negative effects*
@@ -105,6 +108,9 @@ new_features:
   change: |
     Added support for not-equal operator for access log filter rules, in
     :ref:`ComparisonFilter <envoy_v3_api_msg_config.accesslog.v3.ComparisonFilter>`.
+- area: logging
+  change: |
+    Added support to flush access logs in the listener on the start of the connections.
 - area: cel
   change: |
     Added per-expression configuration options for CEL evaluator to control string conversion, concatenation,

docs/root/configuration/observability/access_log/usage.rst

@@ -8,8 +8,9 @@ Configuration
 
 Access logs are configured as part of the :ref:`HTTP connection manager config
 <config_http_conn_man>`, :ref:`TCP Proxy <config_network_filters_tcp_proxy>`,
-:ref:`UDP Proxy <config_udp_listener_filters_udp_proxy>` or
-:ref:`Thrift Proxy <config_network_filters_thrift_proxy>`.
+:ref:`UDP Proxy <config_udp_listener_filters_udp_proxy>`,
+:ref:`Thrift Proxy <config_network_filters_thrift_proxy>` or
+:ref:`listener <envoy_v3_api_field_config.listener.v3.Listener.access_log>`
 
 * :ref:`v3 API reference <envoy_v3_api_msg_config.accesslog.v3.AccessLog>`
 
@@ -1409,19 +1410,21 @@ UDP
 
   * TcpUpstreamConnected - When TCP Proxy filter has successfully established an upstream connection.
   * TcpPeriodic - On any TCP Proxy filter periodic log record.
-  * TcpConnectionEnd - When a TCP connection is ended on TCP Proxy filter.
+  * TcpConnectionStart - When a TCP connection is started by a listener.
+  * TcpConnectionEnd - When a TCP connection is ended on TCP Proxy filter or a listener.
   * DownstreamStart - When HTTP Connection Manager filter receives a new HTTP request.
   * DownstreamTunnelSuccessfullyEstablished - When the HTTP Connection Manager sends response headers
-                                              indicating a successful HTTP tunnel.
+    indicating a successful HTTP tunnel.
   * DownstreamPeriodic - On any HTTP Connection Manager periodic log record.
   * DownstreamEnd - When an HTTP stream is ended on HTTP Connection Manager filter.
   * UpstreamPoolReady - When a new HTTP request is received by the HTTP Router filter.
   * UpstreamPeriodic - On any HTTP Router filter periodic log record.
   * UpstreamEnd - When an HTTP request is finished on the HTTP Router filter.
   * UdpTunnelUpstreamConnected - When UDP Proxy filter has successfully established an upstream connection.
-                                 Note: It is only relevant for UDP tunneling over HTTP.
+    Note: It is only relevant for UDP tunneling over HTTP.
   * UdpPeriodic - On any UDP Proxy filter periodic log record.
   * UdpSessionEnd - When a UDP session is ended on UDP Proxy filter.
+  * NotConnected - When a TCP connection fails to be established by a listener.
 
 %UNIQUE_ID%
    A unique identifier (UUID) that is generated dynamically.

envoy/network/listener.h

@@ -277,6 +277,11 @@ class ListenerConfig {
    */
   virtual const AccessLog::InstanceSharedPtrVector& accessLogs() const PURE;
 
+  /**
+   * @return bool whether the access logs are flushed when the TCP connection starts.
+   */
+  virtual bool flushAccessLogsOnStart() const PURE;
+
   /**
    * @return pending connection backlog for TCP listeners.
    */

source/common/listener_manager/active_stream_listener_base.cc

@@ -17,10 +17,15 @@ ActiveStreamListenerBase::ActiveStreamListenerBase(Network::ConnectionHandler& p
       listener_(std::move(listener)), dispatcher_(dispatcher) {}
 
 void ActiveStreamListenerBase::emitLogs(Network::ListenerConfig& config,
-                                        StreamInfo::StreamInfo& stream_info) {
-  stream_info.onRequestComplete();
+                                        StreamInfo::StreamInfo& stream_info,
+                                        AccessLog::AccessLogType access_log_type) {
+  if (access_log_type == AccessLog::AccessLogType::NotConnected ||
+      access_log_type == AccessLog::AccessLogType::TcpConnectionEnd) {
+    stream_info.onRequestComplete();
+  }
+  const Formatter::Context log_context{nullptr, nullptr, nullptr, {}, access_log_type};
   for (const auto& access_log : config.accessLogs()) {
-    access_log->log({}, stream_info);
+    access_log->log(log_context, stream_info);
   }
 }
 
@@ -35,7 +40,7 @@ void ActiveStreamListenerBase::newConnection(Network::ConnectionSocketPtr&& sock
     stats_.no_filter_chain_match_.inc();
     stream_info->setResponseFlag(StreamInfo::CoreResponseFlag::NoRouteFound);
     stream_info->setResponseCodeDetails(StreamInfo::ResponseCodeDetails::get().FilterChainNotFound);
-    emitLogs(*config_, *stream_info);
+    emitLogs(*config_, *stream_info, AccessLog::AccessLogType::NotConnected);
     socket->close();
     return;
   }
@@ -93,10 +98,15 @@ ActiveTcpConnection::ActiveTcpConnection(ActiveConnections& active_connections,
   // been incremented at this point either via the connection balancer or in the socket accept
   // path if there is no configured balancer.
   listener.parent_.incNumConnections();
+  if (listener.config_->flushAccessLogsOnStart()) {
+    ActiveStreamListenerBase::emitLogs(*listener.config_, *stream_info_,
+                                       AccessLog::AccessLogType::TcpConnectionStart);
+  }
 }
 
 ActiveTcpConnection::~ActiveTcpConnection() {
-  ActiveStreamListenerBase::emitLogs(*active_connections_.listener_.config_, *stream_info_);
+  ActiveStreamListenerBase::emitLogs(*active_connections_.listener_.config_, *stream_info_,
+                                     AccessLog::AccessLogType::TcpConnectionEnd);
   auto& listener = active_connections_.listener_;
   listener.stats_.downstream_cx_active_.dec();
   listener.stats_.downstream_cx_destroy_.inc();

source/common/listener_manager/active_stream_listener_base.h

@@ -5,6 +5,7 @@
 #include <list>
 #include <memory>
 
+#include "envoy/access_log/access_log.h"
 #include "envoy/common/time.h"
 #include "envoy/event/dispatcher.h"
 #include "envoy/network/connection.h"
@@ -30,7 +31,8 @@ class ActiveStreamListenerBase : public ActiveListenerImplBase,
 public:
   ActiveStreamListenerBase(Network::ConnectionHandler& parent, Event::Dispatcher& dispatcher,
                            Network::ListenerPtr&& listener, Network::ListenerConfig& config);
-  static void emitLogs(Network::ListenerConfig& config, StreamInfo::StreamInfo& stream_info);
+  static void emitLogs(Network::ListenerConfig& config, StreamInfo::StreamInfo& stream_info,
+                       AccessLog::AccessLogType access_log_type);
 
   Event::Dispatcher& dispatcher() { return dispatcher_; }
 
@@ -102,7 +104,8 @@ class ActiveStreamListenerBase : public ActiveListenerImplBase,
       if (!active_socket->connected()) {
         // If active_socket is about to be destructed, emit logs if a connection is not created.
         if (active_socket->streamInfo() != nullptr) {
-          emitLogs(*config_, *active_socket->streamInfo());
+          emitLogs(*config_, *active_socket->streamInfo(), AccessLog::AccessLogType::NotConnected);
+          ;
         } else {
           // If the active_socket is not connected, this socket is not promoted to active
           // connection. Thus the stream_info_ is owned by this active socket.

source/common/listener_manager/active_tcp_socket.cc

@@ -65,7 +65,8 @@ void ActiveTcpSocket::unlink() {
   }
   // Emit logs if a connection is not established.
   if (!connected_ && stream_info_ != nullptr) {
-    ActiveStreamListenerBase::emitLogs(*listener_.config_, *stream_info_);
+    ActiveStreamListenerBase::emitLogs(*listener_.config_, *stream_info_,
+                                       AccessLog::AccessLogType::NotConnected);
   }
   listener_.dispatcher().deferredDelete(std::move(removed));
 }

source/common/listener_manager/listener_impl.cc

@@ -318,6 +318,7 @@ ListenerImpl::ListenerImpl(const envoy::config::listener::v3::Listener& config,
           parent.server_, validation_visitor_, config, *this,
           parent_.factory_->createDrainManager(config.drain_type()))),
       reuse_port_(getReusePortOrDefault(parent_.server_, config, socket_type_)),
+      flush_logs_on_start_(config.access_log_options().flush_access_log_on_start()),
       cx_limit_runtime_key_("envoy.resource_limits.listener." + config.name() +
                             ".connection_limit"),
       open_connections_(std::make_shared<BasicResourceLimitImpl>(
@@ -462,7 +463,7 @@ ListenerImpl::ListenerImpl(ListenerImpl& origin,
       filter_chain_manager_(std::make_unique<FilterChainManagerImpl>(
           addresses_, origin.listener_factory_context_->parentFactoryContext(), initManager(),
           *origin.filter_chain_manager_)),
-      reuse_port_(origin.reuse_port_),
+      reuse_port_(origin.reuse_port_), flush_logs_on_start_(origin.flush_logs_on_start_),
       local_init_watcher_(fmt::format("Listener-local-init-watcher {}", name),
                           [this] {
                             ASSERT(workers_started_);

source/common/listener_manager/listener_impl.h

@@ -320,6 +320,7 @@ class ListenerImpl final : public Network::ListenerConfig,
   }
   ResourceLimit& openConnections() override { return *open_connections_; }
   const AccessLog::InstanceSharedPtrVector& accessLogs() const override { return access_logs_; }
+  bool flushAccessLogsOnStart() const override { return flush_logs_on_start_; }
   uint32_t tcpBacklogSize() const override { return tcp_backlog_size_; }
   uint32_t maxConnectionsToAcceptPerSocketEvent() const override {
     return max_connections_to_accept_per_socket_event_;
@@ -484,6 +485,7 @@ class ListenerImpl final : public Network::ListenerConfig,
   std::shared_ptr<PerListenerFactoryContextImpl> listener_factory_context_;
   std::unique_ptr<FilterChainManagerImpl> filter_chain_manager_;
   const bool reuse_port_;
+  const bool flush_logs_on_start_;
 
   // Per-listener connection limits are only specified via runtime.
   //

source/server/admin/admin.h

@@ -401,6 +401,7 @@ class AdminImpl : public Admin,
     const AccessLog::InstanceSharedPtrVector& accessLogs() const override {
       return empty_access_logs_;
     }
+    bool flushAccessLogsOnStart() const override { return false; }
     uint32_t tcpBacklogSize() const override { return ENVOY_TCP_BACKLOG_SIZE; }
     uint32_t maxConnectionsToAcceptPerSocketEvent() const override {
       return Network::DefaultMaxConnectionsToAcceptPerSocketEvent;

test/extensions/access_loggers/grpc/tcp_grpc_access_log_integration_test.cc

@@ -333,7 +333,7 @@ TEST_P(TcpGrpcAccessLogIntegrationTest, BasicAccessLogFlow) {
       downstream_direct_remote_address:
         socket_address:
           address: {}
-      access_log_type: NotSet
+      access_log_type: TcpConnectionEnd
     connection_properties:
       received_bytes: 3
       sent_bytes: 5
@@ -489,7 +489,7 @@ name: envoy.filters.network.rbac
       downstream_direct_remote_address:
         socket_address:
           address: {}
-      access_log_type: NotSet
+      access_log_type: TcpConnectionEnd
     connection_properties:
       received_bytes: 3
       sent_bytes: 5
@@ -558,7 +558,7 @@ TEST_P(TcpGrpcAccessLogIntegrationTest, SslTerminatedNoJA3) {
       downstream_direct_remote_address:
         socket_address:
           address: {}
-      access_log_type: NotSet
+      access_log_type: TcpConnectionEnd
     connection_properties:
 )EOF",
                                           Network::Test::getLoopbackAddressString(ipVersion()),
@@ -623,7 +623,7 @@ TEST_P(TcpGrpcAccessLogIntegrationTest, SslTerminatedWithJA3) {
       downstream_direct_remote_address:
         socket_address:
           address: {}
-      access_log_type: NotSet
+      access_log_type: TcpConnectionEnd
     connection_properties:
 )EOF",
                                           Network::Test::getLoopbackAddressString(ipVersion()),
@@ -676,6 +676,7 @@ TEST_P(TcpGrpcAccessLogIntegrationTest, SslNotTerminated) {
           address: {}
       tls_properties:
         tls_sni_hostname: sni
+      access_log_type: TcpConnectionEnd
     connection_properties:
       received_bytes: 138
       sent_bytes: 138
@@ -731,6 +732,7 @@ TEST_P(TcpGrpcAccessLogIntegrationTest, SslNotTerminatedWithJA3) {
       tls_properties:
         tls_sni_hostname: sni
         ja3_fingerprint: "ecaf91d232e224038f510cb81aa08b94"
+      access_log_type: TcpConnectionEnd
     connection_properties:
       received_bytes: 138
       sent_bytes: 138
@@ -784,6 +786,7 @@ TEST_P(TcpGrpcAccessLogIntegrationTest, SslNotTerminatedWithJA3NoSNI) {
           address: {}
       tls_properties:
         ja3_fingerprint: "71d1f47d1125ac53c3c6a4863c087cfe"
+      access_log_type: TcpConnectionEnd
     connection_properties:
       received_bytes: 126
       sent_bytes: 126
@@ -855,6 +858,7 @@ TEST_P(TcpGrpcAccessLogIntegrationTest, TlsHandshakeFailure_VerifyFailed) {
         socket_address: {{}}
       upstream_local_address:
         socket_address: {{}}
+      access_log_type: TcpConnectionEnd
     connection_properties: {{}}
 )EOF",
                                           Network::Test::getLoopbackAddressString(ipVersion()))));
@@ -913,6 +917,7 @@ TEST_P(TcpGrpcAccessLogIntegrationTest, TlsHandshakeFailure_NoSharedCipher) {
         socket_address: {{}}
       upstream_local_address:
         socket_address: {{}}
+      access_log_type: TcpConnectionEnd
     connection_properties: {{}}
 )EOF",
                                           Network::Test::getLoopbackAddressString(ipVersion()))));
@@ -970,6 +975,7 @@ TEST_P(TcpGrpcAccessLogIntegrationTest, SslHandshakeFailure_UnsupportedProtocol)
         socket_address: {{}}
       upstream_local_address:
         socket_address: {{}}
+      access_log_type: TcpConnectionEnd
     connection_properties: {{}}
 )EOF",
                                           Network::Test::getLoopbackAddressString(ipVersion()))));

test/extensions/bootstrap/internal_listener/active_internal_listener_test.cc

@@ -176,6 +176,7 @@ TEST_F(ActiveInternalListenerTest, AcceptSocketAndCreateNetworkFilter) {
   EXPECT_CALL(conn_handler_, incNumConnections());
   EXPECT_CALL(filter_chain_factory_, createNetworkFilterChain(_, _)).WillOnce(Return(true));
   EXPECT_CALL(listener_config_, perConnectionBufferLimitBytes());
+  EXPECT_CALL(listener_config_, flushAccessLogsOnStart());
   internal_listener_->onAccept(Network::ConnectionSocketPtr{accepted_socket});
   EXPECT_CALL(conn_handler_, decNumConnections());
   connection->close(Network::ConnectionCloseType::NoFlush);
@@ -224,6 +225,7 @@ TEST_F(ActiveInternalListenerTest, DestroyListenerCloseAllConnections) {
   EXPECT_CALL(conn_handler_, incNumConnections());
   EXPECT_CALL(filter_chain_factory_, createNetworkFilterChain(_, _)).WillOnce(Return(true));
   EXPECT_CALL(listener_config_, perConnectionBufferLimitBytes());
+  EXPECT_CALL(listener_config_, flushAccessLogsOnStart());
   internal_listener_->onAccept(Network::ConnectionSocketPtr{accepted_socket});
 
   EXPECT_CALL(conn_handler_, decNumConnections());
@@ -331,6 +333,7 @@ class ConnectionHandlerTest : public testing::Test, protected Logger::Loggable<L
       return *connection_balancer_;
     }
     const AccessLog::InstanceSharedPtrVector& accessLogs() const override { return access_logs_; }
+    bool flushAccessLogsOnStart() const override { return false; }
     ResourceLimit& openConnections() override { return open_connections_; }
     uint32_t tcpBacklogSize() const override { return tcp_backlog_size_; }
     uint32_t maxConnectionsToAcceptPerSocketEvent() const override {

test/extensions/common/proxy_protocol/proxy_protocol_regression_test.cc

@@ -95,6 +95,7 @@ class ProxyProtocolRegressionTest : public testing::TestWithParam<Network::Addre
   const AccessLog::InstanceSharedPtrVector& accessLogs() const override {
     return empty_access_logs_;
   }
+  bool flushAccessLogsOnStart() const override { return false; }
   uint32_t tcpBacklogSize() const override { return ENVOY_TCP_BACKLOG_SIZE; }
   Init::Manager& initManager() override { return *init_manager_; }
   bool ignoreGlobalConnLimit() const override { return false; }

test/extensions/filters/listener/common/fuzz/listener_filter_fuzzer.h

@@ -72,6 +72,7 @@ class ListenerFilterWithDataFuzzer : public Network::ListenerConfig,
   const AccessLog::InstanceSharedPtrVector& accessLogs() const override {
     return empty_access_logs_;
   }
+  bool flushAccessLogsOnStart() const override { return false; }
   uint32_t tcpBacklogSize() const override { return ENVOY_TCP_BACKLOG_SIZE; }
   uint32_t maxConnectionsToAcceptPerSocketEvent() const override {
     return Network::DefaultMaxConnectionsToAcceptPerSocketEvent;

test/extensions/filters/listener/proxy_protocol/proxy_protocol_test.cc

@@ -126,6 +126,7 @@ class ProxyProtocolTest : public testing::TestWithParam<Network::Address::IpVers
   const AccessLog::InstanceSharedPtrVector& accessLogs() const override {
     return empty_access_logs_;
   }
+  bool flushAccessLogsOnStart() const override { return false; }
   uint32_t tcpBacklogSize() const override { return ENVOY_TCP_BACKLOG_SIZE; }
   uint32_t maxConnectionsToAcceptPerSocketEvent() const override {
     return Network::DefaultMaxConnectionsToAcceptPerSocketEvent;
@@ -2606,6 +2607,7 @@ class WildcardProxyProtocolTest : public testing::TestWithParam<Network::Address
   const AccessLog::InstanceSharedPtrVector& accessLogs() const override {
     return empty_access_logs_;
   }
+  bool flushAccessLogsOnStart() const override { return false; }
   uint32_t tcpBacklogSize() const override { return ENVOY_TCP_BACKLOG_SIZE; }
   uint32_t maxConnectionsToAcceptPerSocketEvent() const override {
     return Network::DefaultMaxConnectionsToAcceptPerSocketEvent;

test/integration/fake_upstream.h

@@ -966,6 +966,7 @@ class FakeUpstream : Logger::Loggable<Logger::Id::testing>,
     const AccessLog::InstanceSharedPtrVector& accessLogs() const override {
       return empty_access_logs_;
     }
+    bool flushAccessLogsOnStart() const override { return false; }
     const Network::ListenerInfoConstSharedPtr& listenerInfo() const override {
       return listener_info_;
     }