ROX-28557: Improved networking delta when the config is updated (#2068)

Author: JoukoVirtanen

collector/lib/ConnTracker.cpp

@@ -71,7 +71,7 @@ void ConnectionTracker::Update(
   }
 }
 
-IPNet ConnectionTracker::NormalizeAddressNoLock(const Address& address) const {
+IPNet ConnectionTracker::NormalizeAddressNoLock(const Address& address, bool enable_external_ips) const {
   if (address.IsNull()) {
     return {};
   }
@@ -96,7 +96,7 @@ IPNet ConnectionTracker::NormalizeAddressNoLock(const Address& address) const {
     return network;
   }
 
-  if (enable_external_ips_) {
+  if (enable_external_ips) {
     return IPNet(address, address.length() * 8);
   }
 
@@ -111,6 +111,58 @@ IPNet ConnectionTracker::NormalizeAddressNoLock(const Address& address) const {
   }
 }
 
+bool ConnectionTracker::ShouldNormalizeConnection(const Connection* conn) const {
+  Endpoint local, remote = conn->remote();
+  IPNet ipnet = NormalizeAddressNoLock(remote.address(), false);
+
+  return Address::IsCanonicalExternalIp(ipnet.address());
+}
+
+/**
+ * Closes connections that match a predicate and moves them to a delta
+ */
+void ConnectionTracker::CloseConnections(ConnMap* old_conn_state, ConnMap* delta_conn, std::function<bool(const Connection*)> predicate) {
+  for (auto it = old_conn_state->begin(); it != old_conn_state->end();) {
+    auto& old_conn = *it;
+    if (predicate(&old_conn.first)) {
+      if (old_conn.second.IsActive()) {
+        old_conn.second.SetActive(false);
+      }
+      delta_conn->insert(old_conn);
+      it = old_conn_state->erase(it);
+    } else {
+      it++;
+    }
+  }
+}
+
+/**
+ * Closes connections that have the 255.255.255.255 external IP address
+ */
+void ConnectionTracker::CloseNormalizedConnections(ConnMap* old_conn_state, ConnMap* delta_conn) {
+  CloseConnections(old_conn_state, delta_conn, [](const Connection* conn) {
+    return Address::IsCanonicalExternalIp(conn->remote().address());
+  });
+}
+
+/**
+ * Closes unnormalized connections that would be normalized to the canonical external
+ * IP address if external IPs was enabled
+ */
+void ConnectionTracker::CloseExternalUnnormalizedConnections(ConnMap* old_conn_state, ConnMap* delta_conn) {
+  CloseConnections(old_conn_state, delta_conn, [this](const Connection* conn) {
+    return ShouldNormalizeConnection(conn) && !Address::IsCanonicalExternalIp(conn->remote().address());
+  });
+}
+
+void ConnectionTracker::CloseConnectionsOnRuntimeConfigChange(ConnMap* old_conn_state, ConnMap* delta_conn, bool enableExternalIPs) {
+  if (enableExternalIPs) {
+    CloseNormalizedConnections(old_conn_state, delta_conn);
+  } else {
+    CloseExternalUnnormalizedConnections(old_conn_state, delta_conn);
+  }
+}
+
 Connection ConnectionTracker::NormalizeConnectionNoLock(const Connection& conn) const {
   bool is_server = conn.is_server();
   if (conn.l4proto() == L4Proto::UDP) {
@@ -123,11 +175,11 @@ Connection ConnectionTracker::NormalizeConnectionNoLock(const Connection& conn)
   if (is_server) {
     // If this is the server, only the local port is relevant, while the remote port does not matter.
     local = Endpoint(IPNet(Address()), conn.local().port());
-    remote = Endpoint(NormalizeAddressNoLock(conn.remote().address()), 0);
+    remote = Endpoint(NormalizeAddressNoLock(conn.remote().address(), enable_external_ips_), 0);
   } else {
     // If this is the client, the local port and address are not relevant.
     local = Endpoint();
-    remote = Endpoint(NormalizeAddressNoLock(remote.address()), remote.port());
+    remote = Endpoint(NormalizeAddressNoLock(remote.address(), enable_external_ips_), remote.port());
   }
 
   return Connection(conn.container(), local, remote, conn.l4proto(), is_server);

collector/lib/ConnTracker.h

@@ -100,6 +100,11 @@ class ConnectionTracker {
   template <typename T>
   static void UpdateOldState(UnorderedMap<T, ConnStatus>* old_state, const UnorderedMap<T, ConnStatus>& new_state, int64_t time_micros, int64_t afterglow_period_micros);
 
+  void CloseConnections(ConnMap* old_conn_state, ConnMap* delta_conn, std::function<bool(const Connection*)> predicate);
+  void CloseNormalizedConnections(ConnMap* old_conn_state, ConnMap* delta_conn);
+  void CloseExternalUnnormalizedConnections(ConnMap* old_conn_state, ConnMap* delta_conn);
+  void CloseConnectionsOnRuntimeConfigChange(ConnMap* old_conn_state, ConnMap* delta_conn, bool enableExternalIPs);
+
   // ComputeDelta computes a diff between new_state and old_state
   template <typename T>
   static void ComputeDeltaAfterglow(const UnorderedMap<T, ConnStatus>& new_state, const UnorderedMap<T, ConnStatus>& old_state, UnorderedMap<T, ConnStatus>& delta, int64_t time_micros, int64_t time_at_last_scrape, int64_t afterglow_period_micros);
@@ -154,11 +159,12 @@ class ConnectionTracker {
   // Those counters are updated as new connections are reported by the system.
   Stats GetConnectionStats_NewConnectionCounters();
 
+  bool ShouldNormalizeConnection(const Connection* conn) const;
+
  private:
   // NormalizeConnection transforms a connection into a normalized form.
   Connection NormalizeConnectionNoLock(const Connection& conn) const;
-
-  IPNet NormalizeAddressNoLock(const Address& address) const;
+  IPNet NormalizeAddressNoLock(const Address& address, bool enable_external_ips) const;
 
   // Returns true if any connection filters are found.
   inline bool HasConnectionFilters() const {

collector/lib/NetworkConnection.h

@@ -136,6 +136,17 @@ class Address {
     }
   }
 
+  static bool IsCanonicalExternalIp(const Address& addr) {
+    switch (addr.family_) {
+      case Family::IPV4:
+        return addr.data_[0] == 0xffffffffULL;
+      case Family::IPV6:
+        return addr.data_[0] == 0xffffffffffffffffULL && addr.data_[1] == 0xffffffffffffffffULL;
+      default:
+        return false;
+    }
+  }
+
  private:
   friend std::ostream& operator<<(std::ostream& os, const Address& addr) {
     int af = (addr.family_ == Family::IPV4) ? AF_INET : AF_INET6;

collector/lib/NetworkStatusNotifier.cpp

@@ -225,6 +225,8 @@ void NetworkStatusNotifier::RunSingle(IDuplexClientWriter<sensor::NetworkConnect
   auto next_scrape = std::chrono::system_clock::now();
   int64_t time_at_last_scrape = NowMicros();
 
+  bool prevEnableExternalIPs = config_.EnableExternalIPs();
+
   while (writer->Sleep(next_scrape)) {
     CLOG(DEBUG) << "Starting network status notification";
     next_scrape = std::chrono::system_clock::now() + std::chrono::seconds(config_.ScrapeInterval());
@@ -240,12 +242,18 @@ void NetworkStatusNotifier::RunSingle(IDuplexClientWriter<sensor::NetworkConnect
     const sensor::NetworkConnectionInfoMessage* msg;
     ConnMap new_conn_state, delta_conn;
     AdvertisedEndpointMap new_cep_state;
+    bool enableExternalIPs = config_.EnableExternalIPs();
+
     WITH_TIMER(CollectorStats::net_fetch_state) {
-      conn_tracker_->EnableExternalIPs(config_.EnableExternalIPs());
+      conn_tracker_->EnableExternalIPs(enableExternalIPs);
 
       new_conn_state = conn_tracker_->FetchConnState(true, true);
       if (config_.EnableAfterglow()) {
         ConnectionTracker::ComputeDeltaAfterglow(new_conn_state, old_conn_state, delta_conn, time_micros, time_at_last_scrape, config_.AfterglowPeriod());
+        if (prevEnableExternalIPs != enableExternalIPs) {
+          conn_tracker_->CloseConnectionsOnRuntimeConfigChange(&old_conn_state, &delta_conn, enableExternalIPs);
+          prevEnableExternalIPs = enableExternalIPs;
+        }
       } else {
         ConnectionTracker::ComputeDelta(new_conn_state, &old_conn_state);
       }

collector/test/ConnTrackerTest.cpp

@@ -1651,6 +1651,104 @@ TEST(ConnTrackerTest, TestConnectionStats) {
   EXPECT_EQ(stats.outbound.public_, 4);
 }
 
+TEST(ConnTrackerTest, TestCloseNormalizedConnections) {
+  ConnectionTracker tracker;
+
+  Endpoint a(Address(192, 168, 0, 1), 80);
+  Endpoint b(Address(255, 255, 255, 255), 9999);
+
+  Connection conn("xyz", a, b, L4Proto::TCP, true);
+  int64_t connection_time = 990;
+
+  ConnMap old_state = {{conn, ConnStatus(connection_time, true)}};
+  ConnMap delta;
+  ConnMap expected_delta = {{conn, ConnStatus(connection_time, false)}};
+
+  tracker.CloseNormalizedConnections(&old_state, &delta);
+
+  EXPECT_THAT(old_state, IsEmpty());
+  EXPECT_THAT(delta, expected_delta);
+}
+
+TEST(CloseNormalizedConnectionsTest, UnnormalizedConnectionsAreKept) {
+  ConnectionTracker tracker;
+
+  Endpoint a(Address(192, 168, 0, 1), 80);
+  Endpoint b(Address(192, 168, 1, 10), 9999);
+
+  Connection conn("xyz", a, b, L4Proto::TCP, true);
+  int64_t connection_time = 990;
+
+  ConnMap old_state = {{conn, ConnStatus(connection_time, true)}};
+  ConnMap delta;
+  ConnMap expected_old_state = {{conn, ConnStatus(connection_time, true)}};
+
+  tracker.CloseNormalizedConnections(&old_state, &delta);
+
+  EXPECT_THAT(old_state, expected_old_state);
+  EXPECT_THAT(delta, IsEmpty());
+}
+
+TEST(ConnTrackerTest, TestCloseExternalUnnormalizedConnections) {
+  ConnectionTracker tracker;
+
+  Endpoint a(Address(192, 168, 0, 1), 80);
+  Endpoint b(Address(11, 168, 1, 10), 9999);
+
+  Connection conn("xyz", a, b, L4Proto::TCP, true);
+  int64_t connection_time = 990;
+
+  ConnMap old_state = {{conn, ConnStatus(connection_time, true)}};
+  ConnMap delta;
+  ConnMap expected_delta = {{conn, ConnStatus(connection_time, false)}};
+
+  tracker.CloseExternalUnnormalizedConnections(&old_state, &delta);
+
+  EXPECT_THAT(old_state, IsEmpty());
+  EXPECT_THAT(delta, expected_delta);
+}
+
+TEST(CloseExternalUnnormalizedConnectionsTest, InternalConnectionsAreKept) {
+  ConnectionTracker tracker;
+
+  Endpoint a(Address(192, 168, 0, 1), 80);
+  Endpoint b(Address(192, 168, 1, 10), 9999);
+
+  Connection conn("xyz", a, b, L4Proto::TCP, true);
+  int64_t connection_time = 990;
+
+  ConnMap old_state = {{conn, ConnStatus(connection_time, true)}};
+  ConnMap delta;
+  ConnMap expected_old_state = {{conn, ConnStatus(connection_time, true)}};
+
+  tracker.CloseExternalUnnormalizedConnections(&old_state, &delta);
+
+  EXPECT_THAT(old_state, expected_old_state);
+  EXPECT_THAT(delta, IsEmpty());
+}
+
+TEST(ConnTrackerTest, TestShouldNormalizeConnection) {
+  ConnectionTracker tracker;
+
+  Endpoint a(Address(192, 168, 0, 1), 80);
+  Endpoint b(Address(11, 168, 1, 10), 9999);
+
+  const Connection conn("xyz", a, b, L4Proto::TCP, true);
+
+  EXPECT_TRUE(tracker.ShouldNormalizeConnection(&conn));
+}
+
+TEST(ConnTrackerTest, TestShouldNormalizeConnectionFalse) {
+  ConnectionTracker tracker;
+
+  Endpoint a(Address(192, 168, 0, 1), 80);
+  Endpoint b(Address(192, 168, 1, 10), 9999);
+
+  const Connection conn("xyz", a, b, L4Proto::TCP, true);
+
+  EXPECT_FALSE(tracker.ShouldNormalizeConnection(&conn));
+}
+
 }  // namespace
 
 }  // namespace collector

collector/test/NetworkConnectionTest.cpp

@@ -162,6 +162,20 @@ TEST(TestIPNet, Parse) {
   EXPECT_FALSE(ip_net);
 }
 
+TEST(TestIPNet, TestIsCanonicalExternalIp) {
+  std::vector<std::pair<Address, bool>> tests = {
+      {{192, 168, 0, 1}, false},
+      {{192, 168, 1, 10}, false},
+      {{255, 255, 255, 255}, true},
+      {{0xdeadbeefULL, 0xffffffffffffffffULL}, false},
+      {{0xffffffffffffffffULL, 0xffffffffffffffffULL}, true},
+  };
+
+  for (const auto& [address, expected] : tests) {
+    EXPECT_EQ(Address::IsCanonicalExternalIp(address), expected);
+  }
+}
+
 }  // namespace
 
 }  // namespace collector