network_filter: Pass connection close from upstream to downstream

Monitor connection close events on the upstream filter and close the
downstream connection if both connections have the same 5-tuple.

This fixes the issue where a newly reopened upstream connection with the
same 5-tuple causes the downstream connection to become non-functional.

Signed-off-by: Jarno Rajahalme <[email protected]>

Author: jrajahalme

cilium/network_filter.cc

@@ -115,6 +115,9 @@ Config::Config(const ::cilium::NetworkFilter& config, bool is_upstream,
   if (config.proxylib().length() > 0) {
     proxylib_ = std::make_shared<Cilium::GoFilter>(config.proxylib(), config.proxylib_params());
   }
+
+  ENVOY_LOG(debug, "cilium.network: {} filter configured",
+            is_upstream_ ? "upstream" : "downstream");
 }
 
 void Config::log(Cilium::AccessLog::Entry& entry, ::cilium::EntryType type) {
@@ -123,6 +126,61 @@ void Config::log(Cilium::AccessLog::Entry& entry, ::cilium::EntryType type) {
   }
 }
 
+void Instance::onEvent(Network::ConnectionEvent event) {
+  auto& conn = callbacks_->connection();
+  switch (event) {
+  case Network::ConnectionEvent::LocalClose:
+  case Network::ConnectionEvent::RemoteClose:
+    // close downstream connection when upstream is closed, but only if the original source address
+    // and port are used on the upstream side.
+    if (config_->is_upstream_) {
+      auto& stream_info = conn.streamInfo();
+
+      // check if the downstream connection exists
+      const auto downstream_connection_fs =
+          stream_info.filterState()->getDataMutable<Network::Cilium::DownstreamConnection>(
+              Network::Cilium::DownstreamConnection::key());
+      if (downstream_connection_fs) {
+        Network::Connection* ds_conn = downstream_connection_fs->connection_;
+        if (ds_conn) {
+          // check if upstream and downstream connections have the same source and destination
+          // addresses, respectively
+          if (*conn.connectionInfoProvider().remoteAddress() ==
+                  *ds_conn->connectionInfoProvider().localAddress() &&
+              *conn.connectionInfoProvider().localAddress() ==
+                  *ds_conn->connectionInfoProvider().remoteAddress()) {
+            ENVOY_CONN_LOG(
+                debug,
+                "Upstream connection closed, closing downstream connection due to the same 5-tuple",
+                conn);
+            ds_conn->close(Network::ConnectionCloseType::FlushWrite);
+          } else {
+            ENVOY_CONN_LOG(debug,
+                           "Upstream connection closed, but it has different 5-tuple, downstream "
+                           "connection not closed (src: {}/{}, dst: {}/{})",
+                           conn, ds_conn->connectionInfoProvider().remoteAddress()->asStringView(),
+                           conn.connectionInfoProvider().localAddress()->asStringView(),
+                           ds_conn->connectionInfoProvider().localAddress()->asStringView(),
+                           conn.connectionInfoProvider().remoteAddress()->asStringView());
+          }
+        } else {
+          ENVOY_CONN_LOG(debug, "Upstream connection closed, downstream connection == nullptr",
+                         conn);
+        }
+      } else {
+        ENVOY_CONN_LOG(
+            debug, "Upstream connection closed, but no downstream connection filter state found",
+            conn);
+      }
+    } else {
+      ENVOY_CONN_LOG(debug, "Downstream filter ignoring connection close event", conn);
+    }
+    break;
+  default:
+    break;
+  }
+}
+
 bool Instance::enforceNetworkPolicy(const Cilium::CiliumPolicyFilterState* policy_fs,
                                     Cilium::CiliumDestinationFilterState* dest_fs,
                                     uint32_t destination_identity,
@@ -138,7 +196,7 @@ bool Instance::enforceNetworkPolicy(const Cilium::CiliumPolicyFilterState* polic
 
   // Is there a pod egress policy?
   bool use_proxy_lib = false;
-  if (policy_fs->pod_ip_.length() > 0) {
+  if (!policy_fs->pod_ip_.empty()) {
     if (!policy_fs->enforcePodNetworkPolicy(conn, destination_identity, destination_port_, sni,
                                             use_proxy_lib, l7proto_)) {
       log_entry_.initFromConnection(policy_fs->pod_ip_, policy_fs->proxy_id_, false,
@@ -152,7 +210,7 @@ bool Instance::enforceNetworkPolicy(const Cilium::CiliumPolicyFilterState* polic
   }
 
   // Is there an Ingress policy?
-  if (policy_fs->ingress_policy_name_.length() > 0) {
+  if (!policy_fs->ingress_policy_name_.empty()) {
     log_entry_.initFromConnection(policy_fs->ingress_policy_name_, policy_fs->proxy_id_, false,
                                   policy_fs->source_identity_,
                                   stream_info.downstreamAddressProvider().remoteAddress(),
@@ -186,16 +244,18 @@ bool Instance::enforceNetworkPolicy(const Cilium::CiliumPolicyFilterState* polic
 }
 
 Network::FilterStatus Instance::onNewConnection() {
-  // Upstream handling happens in onDestinationSelected() below.
-  if (config_->is_upstream_) {
-    return Network::FilterStatus::Continue;
-  }
-
   // If there is no upstream filter, onDestinationSelected for the upstream connection
   // will be called on the downstream filter instead, but after this call.
 
   auto& conn = callbacks_->connection();
-  ENVOY_CONN_LOG(debug, "cilium.network: onNewConnection (downstream)", conn);
+  ENVOY_CONN_LOG(debug, "cilium.network: onNewConnection ({})", conn,
+		 config_->is_upstream_ ? "upstream" : "downstream");
+
+  // Upstream handling happens in onDestinationSelected() below.
+  if (config_->is_upstream_) {
+    conn.addConnectionCallbacks(*this);
+    return Network::FilterStatus::Continue;
+  }
 
   // Buffer data until proxylib policy is available, if configured with proxylib
   if (config_->proxylib_.get() != nullptr) {

cilium/network_filter.h

@@ -53,7 +53,9 @@ using ConfigSharedPtr = std::shared_ptr<Config>;
 /**
  * Implementation of a Cilium network filter.
  */
-class Instance : public Network::Filter, Logger::Loggable<Logger::Id::filter> {
+class Instance : public Network::Filter,
+                 public Network::ConnectionCallbacks,
+                 Logger::Loggable<Logger::Id::filter> {
 public:
   Instance(const ConfigSharedPtr& config) : config_(config) {}
 
@@ -70,6 +72,11 @@ class Instance : public Network::Filter, Logger::Loggable<Logger::Id::filter> {
   // Network::WriteFilter
   Network::FilterStatus onWrite(Buffer::Instance&, bool end_stream) override;
 
+  // Network::ConnectionCallbacks
+  void onEvent(Network::ConnectionEvent event) override;
+  void onAboveWriteBufferHighWatermark() override {}
+  void onBelowWriteBufferLowWatermark() override {}
+
 private:
   // helper to be used either directly from onNewConnection (no L7 LB),
   // or from upstream callback (l7 lb)