Add scheduling for config hash verficiation

forch/faucet_event_client.py

@@ -212,9 +212,11 @@ def _dispatch_faucet_event(self, target, target_event):
     def _should_log_event(self, event):
         return event and os.getenv('FAUCET_EVENT_DEBUG')
 
-    def next_event(self, blocking=False):
+    def next_event(self, get_config_hash_clashed, blocking=False):
         """Return the next event from the queue"""
         while self.event_socket_connected and self.has_event(blocking=blocking):
+            if get_config_hash_clashed():
+                raise Exception('Config hash clashed')
             with self._buffer_lock:
                 line, remainder = self.buffer.split('\n', 1)
                 self.buffer = remainder

forch/forchestrator.py

@@ -45,6 +45,7 @@
 _FAUCET_PROM_PORT_DEFAULT = 9302
 _GAUGE_PROM_HOST = '127.0.0.1'
 _GAUGE_PROM_PORT_DEFAULT = 9303
+_CONFIG_HASH_CLASH_TIMEOUT_SEC_DEFAULT = '60'
 
 _TARGET_FAUCET_METRICS = (
     'port_status',
@@ -113,6 +114,14 @@ def __init__(self, config):
         self._metrics = None
         self._varz_proxy = None
 
+        self._config_hash_clash_timer = None
+        self._config_hash_clashed = False
+        self._config_hash_clash_timeout_sec = (
+            self._config.event_client.config_hash_clash_timeout_sec or
+            int(os.getenv(
+                '_CONFIG_HASH_CLASH_TIMEOUT_SEC', _CONFIG_HASH_CLASH_TIMEOUT_SEC_DEFAULT))
+        )
+
         self._lock = threading.Lock()
 
     def initialize(self):
@@ -377,7 +386,7 @@ def _process_device_placement(self, eth_src, device_placement, static=False, exp
             self._authenticator.process_device_placement(eth_src, device_placement)
         else:
             LOGGER.info(
-                'Ignored vlan expiration for device %s with expired vlan %d', eth_src, expired_vlan)
+                'Ignored vlan expiration for device %s with expired vlan %s', eth_src, expired_vlan)
 
     def handle_auth_result(self, mac, access, segment, role):
         """Method passed as callback to authenticator to forward auth results"""
@@ -436,9 +445,43 @@ def _restore_states(self):
     def _restore_faucet_config(self, timestamp, config_hash):
         config_info, faucet_dps, _ = self._get_faucet_config()
         self._update_config_warning_varz()
-        assert config_hash == config_info['hashes'], 'config hash info does not match'
+
+        if config_hash == config_info['hashes']:
+            self._attempt_cancel_config_hash_clash_timer()
+        else:
+            LOGGER.warning('Config hash does not match')
+            self._attempt_start_config_hash_clash_timer()
+
         self._faucet_collector.process_dataplane_config_change(timestamp, faucet_dps)
 
+    def _attempt_start_config_hash_clash_timer(self):
+        if self._config_hash_clash_timer and self._config_hash_clash_timer.is_alive():
+            return
+        self._config_hash_clash_timer = threading.Timer(
+            interval=self._config_hash_clash_timeout_sec,
+            function=self._check_config_hash_clashed)
+        self._config_hash_clash_timer.start()
+        LOGGER.info(
+            'Config hash clash timer started with %s seconds', self._config_hash_clash_timeout_sec)
+
+    def _attempt_cancel_config_hash_clash_timer(self):
+        if not self._config_hash_clash_timer or not self._config_hash_clash_timer.is_alive():
+            return
+        self._config_hash_clash_timer.cancel()
+        LOGGER.info('Config hash clash timer cancelled')
+
+    def _check_config_hash_clashed(self):
+        _, varz_config_hashes = self._get_varz_config()
+        config_info, _, _ = self._get_faucet_config()
+
+        if varz_config_hashes != config_info['hashes']:
+            LOGGER.error(
+                'Config hash does not match after %s seconds', self._config_hash_clash_timeout_sec)
+            self._config_hash_clashed = True
+
+    def _get_config_hash_clashed(self):
+        return self._config_hash_clashed
+
     def _process_config_change(self, event):
         self._faucet_collector.process_dp_config_change(
             event.timestamp, event.dp_name, event.restart_type, event.dp_id)
@@ -465,7 +508,7 @@ def main_loop(self):
                     self._faucet_events_connect()
 
                 try:
-                    self._faucet_events.next_event(blocking=True)
+                    self._faucet_events.next_event(self._get_config_hash_clashed, blocking=True)
                 except FaucetEventOrderError as e:
                     LOGGER.error("Faucet event order error: %s", e)
                     if self._metrics: