[dhcpmon] Make relay disparity detection resilient

src/dhcp_device.cpp

@@ -7,8 +7,11 @@
 #include <unistd.h>
 #include <libexplain/ioctl.h>
 #include <ifaddrs.h>
+#include <mutex>
 #include <pcap.h>
 #include <syslog.h>
+#include <unordered_map>
+#include <unordered_set>
 
 #include "dhcp_device.h"
 
@@ -69,6 +72,39 @@ static const char *counter_desc[DHCP_COUNTERS_COUNT] = {
     [DHCP_COUNTERS_SNAPSHOT_V6] = "Snapshot_V6",
 };
 
+typedef struct
+{
+    uint64_t last_rx;
+    uint64_t last_tx;
+    uint32_t pending_windows;
+    uint8_t tx_credit;
+    bool initialized;
+} relay_flow_state_t;
+
+static std::unordered_map<int, std::unordered_map<std::string, std::unordered_map<int, relay_flow_state_t>>> relay_flow_states;
+static std::mutex relay_flow_state_mutex;
+
+static void initialize_relay_flow_states(const std::string &ifname, int rx_sock, int tx_sock,
+                                         const int *monitored_msgs, size_t monitored_msg_cnt)
+{
+    const counter_t &rx_counters = sock_mgr_get_sock_info(rx_sock).all_counters.at(ifname);
+    const counter_t &tx_counters = sock_mgr_get_sock_info(tx_sock).all_counters.at(ifname);
+    for (size_t i = 0; i < monitored_msg_cnt; i++) {
+        int msg_type = monitored_msgs[i];
+        relay_flow_states[rx_sock][ifname][msg_type] = {
+            rx_counters.at(msg_type), tx_counters.at(msg_type), 0, 0, true
+        };
+    }
+}
+
+void dhcp_device_reset_health_state(const std::string &ifname)
+{
+    std::lock_guard<std::mutex> lock(relay_flow_state_mutex);
+    relay_flow_states[rx_sock].erase(ifname);
+    initialize_relay_flow_states(ifname, rx_sock, tx_sock,
+                                 (const int *)monitored_msgs, monitored_msg_sz);
+}
+
 /**
  * @code check_counter_not_transmitted(ifname, rx_sock, tx_sock, monitored_msgs, monitored_msg_cnt);
  * @brief Check if there are received DHCP messages that are not transmitted out
@@ -78,30 +114,80 @@ static const char *counter_desc[DHCP_COUNTERS_COUNT] = {
  * @param tx_sock           tx socket
  * @param monitored_msgs    array of monitored message types
  * @param monitored_msg_cnt number of monitored message types
- * @return                  true if there are received messages not transmitted out, false otherwise
+ * @return                  DHCP relay health status
  */
 // these helpers use const int * to accept both dhcp_message_type_t and dhcpv6_message_type_t arrays
 // without duplicating the function for each enum type; safe on GCC/Linux where unscoped enums use int
-static bool check_counter_not_transmitted(const std::string &ifname, int rx_sock, int tx_sock, const int *monitored_msgs, size_t monitored_msg_cnt)
+static std::unordered_map<int, uint32_t> get_untransmitted_windows(const std::string &ifname,
+                                                                  int rx_sock, int tx_sock,
+                                                                  const int *monitored_msgs,
+                                                                  size_t monitored_msg_cnt)
 {
+    std::lock_guard<std::mutex> lock(relay_flow_state_mutex);
     const sock_info_t &rx_sock_info = sock_mgr_get_sock_info(rx_sock);
     const counter_t &rx_counters = rx_sock_info.all_counters.at(ifname);
-    const counter_t &rx_counters_snapshot = rx_sock_info.all_counters_snapshot.at(ifname);
 
     const sock_info_t &tx_sock_info = sock_mgr_get_sock_info(tx_sock);
     const counter_t &tx_counters = tx_sock_info.all_counters.at(ifname);
-    const counter_t &tx_counters_snapshot = tx_sock_info.all_counters_snapshot.at(ifname);
 
-    // when there is packet in, no packet out
+    std::unordered_map<int, uint32_t> result;
     for (size_t i = 0; i < monitored_msg_cnt; i++) {
-        if (rx_counters.at(monitored_msgs[i]) > rx_counters_snapshot.at(monitored_msgs[i]) &&
-            tx_counters.at(monitored_msgs[i]) <= tx_counters_snapshot.at(monitored_msgs[i])) {
-            return true;
+        int msg_type = monitored_msgs[i];
+        uint64_t current_rx = rx_counters.at(msg_type);
+        uint64_t current_tx = tx_counters.at(msg_type);
+        relay_flow_state_t &state = relay_flow_states[rx_sock][ifname][msg_type];
+
+        if (!state.initialized || current_rx < state.last_rx || current_tx < state.last_tx) {
+            state = {current_rx, current_tx, 0, 0, true};
+            result[msg_type] = 0;
+            continue;
         }
+
+        uint64_t rx_delta = current_rx - state.last_rx;
+        uint64_t tx_delta = current_tx - state.last_tx;
+        bool had_pending = state.pending_windows > 0;
+        bool previous_tx_credit = state.tx_credit > 0;
+        bool current_tx_activity = tx_delta > 0;
+        state.last_rx = current_rx;
+        state.last_tx = current_tx;
+
+        if (had_pending) {
+            if (previous_tx_credit || current_tx_activity) {
+                state.pending_windows = 0;
+                state.tx_credit = current_tx_activity ? 1 : 0;
+            } else {
+                state.pending_windows++;
+                state.tx_credit = 0;
+            }
+        } else if (rx_delta > 0) {
+            if (previous_tx_credit) {
+                state.pending_windows = 0;
+                state.tx_credit = current_tx_activity ? 1 : 0;
+            } else if (current_tx_activity) {
+                state.pending_windows = 0;
+                state.tx_credit = 1;
+            } else {
+                state.pending_windows = 1;
+                state.tx_credit = 0;
+            }
+        } else {
+            state.pending_windows = 0;
+            state.tx_credit = current_tx_activity ? 1 : 0;
+        }
+        result[msg_type] = state.pending_windows;
     }
-    return false;
+    return result;
+}
+
+std::unordered_map<int, uint32_t> dhcp_device_get_untransmitted_windows(const std::string &ifname)
+{
+    return get_untransmitted_windows(ifname, rx_sock, tx_sock,
+                                     (const int *)monitored_msgs, monitored_msg_sz);
 }
 
+static bool check_counter_increased(const std::string &ifname, int sock,
+                                    const int *monitored_msgs, size_t monitored_msg_cnt);
+
 /**
  * @code dhcp_device_check_positive_health(ifname);
  * @brief Check that DHCP relayed messages are being transmitted out of this interface/dev
@@ -112,8 +198,16 @@ static bool check_counter_not_transmitted(const std::string &ifname, int rx_sock
  */
 static dhcp_mon_status_t dhcp_device_check_positive_health(const std::string &ifname)
 {
-    return check_counter_not_transmitted(ifname, rx_sock, tx_sock, (const int *)monitored_msgs, monitored_msg_sz) ?
-           DHCP_MON_STATUS_UNHEALTHY : DHCP_MON_STATUS_HEALTHY;
+    bool has_activity = check_counter_increased(ifname, rx_sock,
+                                                (const int *)monitored_msgs, monitored_msg_sz) ||
+                        check_counter_increased(ifname, tx_sock,
+                                                (const int *)monitored_msgs, monitored_msg_sz);
+    for (const auto &[msg_type, windows] : dhcp_device_get_untransmitted_windows(ifname)) {
+        if (windows > 0) {
+            return DHCP_MON_STATUS_UNHEALTHY;
+        }
+    }
+    return has_activity ? DHCP_MON_STATUS_HEALTHY : DHCP_MON_STATUS_INDETERMINATE;
 }
 
 /**
@@ -124,10 +218,10 @@ static dhcp_mon_status_t dhcp_device_check_positive_health(const std::string &if
  * @param ifname           interface name
  * @return                 DHCP_MON_STATUS_HEALTHY, DHCP_MON_STATUS_UNHEALTHY, or DHCP_MON_STATUS_INDETERMINATE
  */
-static dhcp_mon_status_t dhcp_device_check_positive_health_v6(const std::string &ifname)
+static dhcp_mon_status_t dhcp_device_check_positive_health_v6(const std::string &)
 {
-   return  check_counter_not_transmitted(ifname, rx_sock_v6, tx_sock_v6, (const int *)monitored_v6_msgs, monitored_v6_msg_sz) ?
-           DHCP_MON_STATUS_UNHEALTHY : DHCP_MON_STATUS_HEALTHY;
+    // Client and relay DHCPv6 message types differ across the relay boundary.
+    return DHCP_MON_STATUS_INDETERMINATE;
 }
 
 /**
@@ -382,7 +476,9 @@ void dhcp_device_print_status_debug(const std::string &ifname, dhcp_counters_typ
 
 dhcp_mon_status_t dhcp_device_get_status(const std::string &ifname, dhcp_device_check_t check_type)
 {
-    if (sock_mgr_counters_unchanged(ifname, (const int *)monitored_msgs, monitored_msg_sz, (const int *)monitored_v6_msgs, monitored_v6_msg_sz)) {
+    if (check_type != DHCP_DEVICE_CHECK_POSITIVE && check_type != DHCP_DEVICE_CHECK_POSITIVE_V6 &&
+        sock_mgr_counters_unchanged(ifname, (const int *)monitored_msgs, monitored_msg_sz,
+                                    (const int *)monitored_v6_msgs, monitored_v6_msg_sz)) {
         return DHCP_MON_STATUS_INDETERMINATE;
     }
 

src/dhcp_device.h

@@ -18,6 +18,7 @@
 #include <netinet/in.h>
 #include <net/ethernet.h>
 #include <string>
+#include <unordered_map>
 
 /** DHCP message types */
 typedef enum
@@ -242,6 +243,30 @@ void dhcp_device_free(dhcp_device_context_t *context);
  */
 dhcp_mon_status_t dhcp_device_get_status(const std::string &ifname, dhcp_device_check_t check_type);
 
+/**
+ * @code dhcp_device_get_untransmitted_windows(ifname);
+ *
+ * @brief update and return unmatched DHCPv4 relay RX age in health windows per message type.
+ *        Caller must hold the counter-state write lock or otherwise quiesce packet handlers.
+ *
+ * @param ifname           interface name
+ *
+ * @return message type to unmatched-window count
+ */
+std::unordered_map<int, uint32_t> dhcp_device_get_untransmitted_windows(const std::string &ifname);
+
+/**
+ * @code dhcp_device_reset_health_state(ifname);
+ *
+ * @brief reset relay health watermarks to the current counters for an interface.
+ *        Caller must hold the counter-state write lock or otherwise quiesce packet handlers.
+ *
+ * @param ifname           interface name
+ *
+ * @return none
+ */
+void dhcp_device_reset_health_state(const std::string &ifname);
+
 /**
  * @code dhcp_device_print_status(ifname, type);
  *

src/dhcp_mon.cpp

@@ -5,6 +5,7 @@
  */
 
 #include <signal.h>
+#include <atomic>
 #include <errno.h>
 #include <stdlib.h>
 #include <unistd.h>
@@ -41,6 +42,7 @@ static constexpr int MINIMAL_CLEAR_COUNTER_TIMEOUT_SEC = 5;
 static constexpr int CLEAR_COUNTER_DELAY_AFTER_DB_UPDATE_SEC = 1;
 /** Mutex lock to modify write_counter_to_db for different threads */
 static std::mutex db_sync_mutex;
+static std::atomic<bool> health_reset_pending{false};
 /** tag for db_update event */
 static const char db_update_tag[] = "DB_UPDATE";
 /** Latest timestamp of writing cache counter to COUNTERS_DB */
@@ -373,6 +375,7 @@ static void update_cache_counter_callback(evutil_socket_t fd, short event, void
     // for discrepency in interface between cache counter and DB counter, we dont handle it in this function
     // we leave it to db updater to handle it
     if (sock_mgr_pause_write_cache_to_db_all_cleared()) {
+        health_reset_pending = true;
         syslog(LOG_INFO, "All sockets cleared pause_write_cache_to_db, start write back to DB counter from cache counter");
         main_event_mgr->activate_all_events(db_update_tag, EV_TIMEOUT);
     }
@@ -393,6 +396,10 @@ static void timeout_callback(evutil_socket_t fd, short event, void *arg)
 {
     syslog_debug(LOG_INFO, "Received timeout signal for DHCP relay health check");
 
+    if (health_reset_pending.exchange(false)) {
+        reset_dhcp_relay_health_state(agg_dev_all);
+    }
+
     dhcp_devman_print_all_status_debug(DHCP_COUNTERS_CURRENT);
     dhcp_devman_print_all_status_debug(DHCP_COUNTERS_SNAPSHOT);
     dhcp_devman_print_all_status_debug(DHCP_COUNTERS_CURRENT_V6);
@@ -520,6 +527,7 @@ int dhcp_mon_init(size_t snaplen, int window_sec, int max_count, int db_update_i
     // deinitialization of counters is not our responsibility
     // cache counter will be cleanup by sock_mgr_free and the initialized db we intend to keep
     initialize_all_intf_counters();
+    reset_dhcp_relay_health_state(agg_dev_all);
     syslog(LOG_INFO, "Initialized all counters for tracked interfaces");
 
     window_interval_sec = window_sec;