Listener: reset the file event when destroying listener filters

envoy/network/io_handle.h

@@ -299,6 +299,12 @@ class IoHandle {
   virtual void initializeFileEvent(Event::Dispatcher& dispatcher, Event::FileReadyCb cb,
                                    Event::FileTriggerType trigger, uint32_t events) PURE;
 
+  /**
+   * Check whether the file event is initialized or not.
+   * @return return true if the file event is initialize, otherwise return false.
+   */
+  virtual bool isFileEventInitialized() PURE;
+
   /**
    * Activates file events for the current underlying fd.
    * @param events events that will be activated.

source/common/network/io_socket_handle_impl.h

@@ -70,7 +70,7 @@ class IoSocketHandleImpl : public IoHandle, protected Logger::Loggable<Logger::I
   Address::InstanceConstSharedPtr peerAddress() override;
   void initializeFileEvent(Event::Dispatcher& dispatcher, Event::FileReadyCb cb,
                            Event::FileTriggerType trigger, uint32_t events) override;
-
+  bool isFileEventInitialized() override { return file_event_ != nullptr; }
   IoHandlePtr duplicate() override;
 
   void activateFileEvents(uint32_t events) override;

source/common/quic/quic_io_handle_wrapper.h

@@ -133,6 +133,8 @@ class QuicIoHandleWrapper : public Network::IoHandle {
     io_handle_.initializeFileEvent(dispatcher, cb, trigger, events);
   }
 
+  bool isFileEventInitialized() override { return io_handle_.isFileEventInitialized(); }
+
   Network::IoHandlePtr duplicate() override { return io_handle_.duplicate(); }
 
   void activateFileEvents(uint32_t events) override { io_handle_.activateFileEvents(events); }

source/extensions/filters/listener/http_inspector/http_inspector.h

@@ -66,6 +66,11 @@ using ConfigSharedPtr = std::shared_ptr<Config>;
 class Filter : public Network::ListenerFilter, Logger::Loggable<Logger::Id::filter> {
 public:
   Filter(const ConfigSharedPtr config);
+  ~Filter() override {
+    if (cb_) {
+      cb_->socket().ioHandle().resetFileEvents();
+    }
+  }
 
   // Network::ListenerFilter
   Network::FilterStatus onAccept(Network::ListenerFilterCallbacks& cb) override;

source/extensions/filters/listener/proxy_protocol/proxy_protocol.h

@@ -83,7 +83,11 @@ enum class ReadOrParseState { Done, TryAgainLater, Error };
 class Filter : public Network::ListenerFilter, Logger::Loggable<Logger::Id::filter> {
 public:
   Filter(const ConfigSharedPtr& config) : config_(config) {}
-
+  ~Filter() override {
+    if (cb_) {
+      cb_->socket().ioHandle().resetFileEvents();
+    }
+  }
   // Network::ListenerFilter
   Network::FilterStatus onAccept(Network::ListenerFilterCallbacks& cb) override;
 
@@ -118,7 +122,7 @@ class Filter : public Network::ListenerFilter, Logger::Loggable<Logger::Id::filt
   bool parseV2Header(char* buf);
   absl::optional<size_t> lenV2Address(char* buf);
 
-  Network::ListenerFilterCallbacks* cb_{};
+  Network::ListenerFilterCallbacks* cb_{nullptr};
 
   // The offset in buf_ that has been fully read
   size_t buf_off_{};

source/extensions/filters/listener/tls_inspector/tls_inspector.h

@@ -73,6 +73,11 @@ using ConfigSharedPtr = std::shared_ptr<Config>;
 class Filter : public Network::ListenerFilter, Logger::Loggable<Logger::Id::filter> {
 public:
   Filter(const ConfigSharedPtr config);
+  ~Filter() override {
+    if (cb_) {
+      cb_->socket().ioHandle().resetFileEvents();
+    }
+  }
 
   // Network::ListenerFilter
   Network::FilterStatus onAccept(Network::ListenerFilterCallbacks& cb) override;
@@ -85,7 +90,7 @@ class Filter : public Network::ListenerFilter, Logger::Loggable<Logger::Id::filt
   void onServername(absl::string_view name);
 
   ConfigSharedPtr config_;
-  Network::ListenerFilterCallbacks* cb_;
+  Network::ListenerFilterCallbacks* cb_{nullptr};
 
   bssl::UniquePtr<SSL> ssl_;
   uint64_t read_{0};

source/extensions/io_socket/user_space/io_handle_impl.h

@@ -81,6 +81,8 @@ class IoHandleImpl final : public Network::IoHandle,
 
   void initializeFileEvent(Event::Dispatcher& dispatcher, Event::FileReadyCb cb,
                            Event::FileTriggerType trigger, uint32_t events) override;
+  bool isFileEventInitialized() override { return user_file_event_ != nullptr; }
+
   Network::IoHandlePtr duplicate() override;
   void activateFileEvents(uint32_t events) override;
   void enableFileEvents(uint32_t events) override;

source/server/active_tcp_listener.cc

@@ -199,10 +199,8 @@ void ActiveTcpSocket::newConnection() {
     if (socket_->detectedTransportProtocol().empty()) {
       socket_->setDetectedTransportProtocol("raw_buffer");
     }
-    // TODO(lambdai): add integration test
-    // TODO: Address issues in wider scope. See https://github.com/envoyproxy/envoy/issues/8925
-    // Erase accept filter states because accept filters may not get the opportunity to clean up.
-    // Particularly the assigned events need to reset before assigning new events in the follow up.
+    // Clear the listener filter to ensure the file event registered by
+    // listener filter to be removed. reference PR #8922.
     accept_filters_.clear();
     // Create a new connection on this listener.
     listener_.newConnection(std::move(socket_), std::move(stream_info_));

test/extensions/filters/listener/http_inspector/http_inspector_test.cc

@@ -31,7 +31,11 @@ class HttpInspectorTest : public testing::Test {
   HttpInspectorTest()
       : cfg_(std::make_shared<Config>(store_)),
         io_handle_(std::make_unique<Network::IoSocketHandleImpl>(42)) {}
-  ~HttpInspectorTest() override { io_handle_->close(); }
+  ~HttpInspectorTest() override {
+    io_handle_->close();
+    filter_.reset();
+    EXPECT_EQ(false, io_handle_->isFileEventInitialized());
+  }
 
   void init(bool include_inline_recv = true) {
     filter_ = std::make_unique<Filter>(cfg_);

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

@@ -79,8 +79,10 @@ class ProxyProtocolTest : public testing::TestWithParam<Network::Address::IpVers
   bool bindToPort() override { return true; }
   bool handOffRestoredDestinationConnections() const override { return false; }
   uint32_t perConnectionBufferLimitBytes() const override { return 0; }
-  std::chrono::milliseconds listenerFiltersTimeout() const override { return {}; }
-  bool continueOnListenerFiltersTimeout() const override { return false; }
+  std::chrono::milliseconds listenerFiltersTimeout() const override {
+    return std::chrono::milliseconds(1000);
+  }
+  bool continueOnListenerFiltersTimeout() const override { return true; }
   Stats::Scope& listenerScope() override { return stats_store_; }
   uint64_t listenerTag() const override { return 1; }
   ResourceLimit& openConnections() override { return open_connections_; }
@@ -210,6 +212,23 @@ INSTANTIATE_TEST_SUITE_P(IpVersions, ProxyProtocolTest,
                          testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),
                          TestUtility::ipTestParamsToString);
 
+// This test ensures the socket file event was reset after timeout, otherwise
+// the assertion which avoid to create file event duplicated will be triggered.
+TEST_P(ProxyProtocolTest, Timeout) {
+  connect();
+  absl::SleepFor(absl::Seconds(5));
+  dispatcher_->run(Event::Dispatcher::RunType::NonBlock);
+  if (GetParam() == Envoy::Network::Address::IpVersion::v4) {
+    EXPECT_EQ(server_connection_->addressProvider().remoteAddress()->ip()->addressAsString(),
+              "127.0.0.1");
+  } else {
+    EXPECT_EQ(server_connection_->addressProvider().remoteAddress()->ip()->addressAsString(),
+              "::1");
+  }
+
+  disconnect();
+}
+
 TEST_P(ProxyProtocolTest, V1Basic) {
   connect();
   write("PROXY TCP4 1.2.3.4 253.253.253.253 65535 1234\r\nmore data");

test/extensions/filters/listener/tls_inspector/tls_inspector_test.cc

@@ -32,7 +32,11 @@ class TlsInspectorTest : public testing::TestWithParam<std::tuple<uint16_t, uint
   TlsInspectorTest()
       : cfg_(std::make_shared<Config>(store_)),
         io_handle_(std::make_unique<Network::IoSocketHandleImpl>(42)) {}
-  ~TlsInspectorTest() override { io_handle_->close(); }
+  ~TlsInspectorTest() override {
+    io_handle_->close();
+    filter_.reset();
+    EXPECT_EQ(false, io_handle_->isFileEventInitialized());
+  }
 
   void init() {
     filter_ = std::make_unique<Filter>(cfg_);

test/integration/listener_filter_integration_test.cc

@@ -48,44 +48,64 @@ class ListenerFilterIntegrationTest : public testing::TestWithParam<Network::Add
     }
   }
 
-  void initializeWithListenerFilter(absl::optional<bool> listener_filter_disabled = absl::nullopt) {
+  void initializeWithListenerFilter(bool ssl_client,
+                                    absl::optional<bool> listener_filter_disabled = absl::nullopt) {
     config_helper_.renameListener("echo");
     std::string tls_inspector_config = ConfigHelper::tlsInspectorFilter();
     if (listener_filter_disabled.has_value()) {
       tls_inspector_config = appendMatcher(tls_inspector_config, listener_filter_disabled.value());
     }
     config_helper_.addListenerFilter(tls_inspector_config);
-    config_helper_.addConfigModifier([](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
-      auto* filter_chain =
-          bootstrap.mutable_static_resources()->mutable_listeners(0)->mutable_filter_chains(0);
-      auto* alpn = filter_chain->mutable_filter_chain_match()->add_application_protocols();
-      *alpn = "envoyalpn";
+
+    config_helper_.addConfigModifier([ssl_client](
+                                         envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+      if (ssl_client) {
+        auto* filter_chain =
+            bootstrap.mutable_static_resources()->mutable_listeners(0)->mutable_filter_chains(0);
+        auto* alpn = filter_chain->mutable_filter_chain_match()->add_application_protocols();
+        *alpn = "envoyalpn";
+      }
+      auto* timeout = bootstrap.mutable_static_resources()
+                          ->mutable_listeners(0)
+                          ->mutable_listener_filters_timeout();
+      timeout->MergeFrom(ProtobufUtil::TimeUtil::MillisecondsToDuration(1000));
+      bootstrap.mutable_static_resources()
+          ->mutable_listeners(0)
+          ->set_continue_on_listener_filters_timeout(true);
     });
-    config_helper_.addSslConfig();
+    if (ssl_client) {
+      config_helper_.addSslConfig();
+    }
+
     useListenerAccessLog("%RESPONSE_CODE_DETAILS%");
     BaseIntegrationTest::initialize();
 
     context_manager_ =
         std::make_unique<Extensions::TransportSockets::Tls::ContextManagerImpl>(timeSystem());
   }
 
-  void setupConnections(bool listener_filter_disabled, bool expect_connection_open) {
-    initializeWithListenerFilter(listener_filter_disabled);
+  void setupConnections(bool listener_filter_disabled, bool expect_connection_open,
+                        bool ssl_client) {
+    initializeWithListenerFilter(ssl_client, listener_filter_disabled);
 
     // Set up the SSL client.
     Network::Address::InstanceConstSharedPtr address =
         Ssl::getSslAddress(version_, lookupPort("echo"));
     context_ = Ssl::createClientSslTransportSocketFactory({}, *context_manager_, *api_);
-    ssl_client_ = dispatcher_->createClientConnection(
-        address, Network::Address::InstanceConstSharedPtr(),
-        context_->createTransportSocket(
-            // nullptr
-            std::make_shared<Network::TransportSocketOptionsImpl>(
-                absl::string_view(""), std::vector<std::string>(),
-                std::vector<std::string>{"envoyalpn"})),
-        nullptr);
-    ssl_client_->addConnectionCallbacks(connect_callbacks_);
-    ssl_client_->connect();
+    Network::TransportSocketPtr transport_socket;
+    if (ssl_client) {
+      transport_socket =
+          context_->createTransportSocket(std::make_shared<Network::TransportSocketOptionsImpl>(
+              absl::string_view(""), std::vector<std::string>(),
+              std::vector<std::string>{"envoyalpn"}));
+    } else {
+      auto transport_socket_factory = std::make_unique<Network::RawBufferSocketFactory>();
+      transport_socket = transport_socket_factory->createTransportSocket(nullptr);
+    }
+    client_ = dispatcher_->createClientConnection(
+        address, Network::Address::InstanceConstSharedPtr(), std::move(transport_socket), nullptr);
+    client_->addConnectionCallbacks(connect_callbacks_);
+    client_->connect();
     while (!connect_callbacks_.connected() && !connect_callbacks_.closed()) {
       dispatcher_->run(Event::Dispatcher::RunType::NonBlock);
     }
@@ -98,27 +118,45 @@ class ListenerFilterIntegrationTest : public testing::TestWithParam<Network::Add
       ASSERT(connect_callbacks_.closed());
     }
   }
+
   std::unique_ptr<Ssl::ContextManager> context_manager_;
   Network::TransportSocketFactoryPtr context_;
   ConnectionStatusCallbacks connect_callbacks_;
   testing::NiceMock<Secret::MockSecretManager> secret_manager_;
-  Network::ClientConnectionPtr ssl_client_;
+  Network::ClientConnectionPtr client_;
 };
 
 // Each listener filter is enabled by default.
 TEST_P(ListenerFilterIntegrationTest, AllListenerFiltersAreEnabledByDefault) {
-  setupConnections(/*listener_filter_disabled=*/false, /*expect_connection_open=*/true);
-  ssl_client_->close(Network::ConnectionCloseType::NoFlush);
+  setupConnections(/*listener_filter_disabled=*/false, /*expect_connection_open=*/true,
+                   /*ssl_client=*/true);
+  client_->close(Network::ConnectionCloseType::NoFlush);
   EXPECT_THAT(waitForAccessLog(listener_access_log_name_), testing::Eq("-"));
 }
 
 // The tls_inspector is disabled. The ALPN won't be sniffed out and no filter chain is matched.
 TEST_P(ListenerFilterIntegrationTest, DisabledTlsInspectorFailsFilterChainFind) {
-  setupConnections(/*listener_filter_disabled=*/true, /*expect_connection_open=*/false);
+  setupConnections(/*listener_filter_disabled=*/true, /*expect_connection_open=*/false,
+                   /*ssl_client=*/true);
   EXPECT_THAT(waitForAccessLog(listener_access_log_name_),
               testing::Eq(StreamInfo::ResponseCodeDetails::get().FilterChainNotFound));
 }
 
+// trigger the tls inspect filter timeout, and continue create new connection after timeout
+TEST_P(ListenerFilterIntegrationTest, ContinueOnListenerTimeout) {
+  setupConnections(/*listener_filter_disabled=*/false, /*expect_connection_open=*/true,
+                   /*ssl_client=*/false);
+  // The length of tls hello message is defined as `TLS_MAX_CLIENT_HELLO = 64 * 1024`
+  // if tls inspect filter doesn't read the max length of hello message data, it
+  // will continue wait. Then the listener filter timeout timer will be triggered.
+  Buffer::OwnedImpl buffer("fake data");
+  client_->write(buffer, false);
+  // the timeout is set as one seconds, sleep 5 to trigger the timeout.
+  absl::SleepFor(absl::Seconds(5));
+  client_->close(Network::ConnectionCloseType::NoFlush);
+  EXPECT_THAT(waitForAccessLog(listener_access_log_name_), testing::Eq("-"));
+}
+
 INSTANTIATE_TEST_SUITE_P(IpVersions, ListenerFilterIntegrationTest,
                          testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),
                          TestUtility::ipTestParamsToString);

test/mocks/network/io_handle.h

@@ -63,6 +63,7 @@ class MockIoHandle : public IoHandle {
   MOCK_METHOD(absl::optional<std::chrono::milliseconds>, lastRoundTripTime, ());
   MOCK_METHOD(Api::SysCallIntResult, ioctl,
               (unsigned long, void*, unsigned long, void*, unsigned long, unsigned long*));
+  MOCK_METHOD(bool, isFileEventInitialized, ());
 };
 
 } // namespace Network