Added additional env vars to control the frequency of detector activating

config.py

@@ -41,6 +41,7 @@ class AppConfig(BaseSettings):
 
     blocking_types: set[Literal["tft", "tfh", "ipset", "nftables"]] = {"tft"}
     blocking_window_duration_sec: int = 10
+    blocking_check_timeout_sec: int = 10
     blocking_ipset_name: str = "tempesta_blocked_ips"
     blocking_time_min: int = 60
     blocking_release_time_min: int = 1

core/lifespan.py

@@ -268,7 +268,7 @@ async def run(self, testing: bool = False) -> None:
 
         while True:
             asyncio.create_task(self._update_threshold_and_block_users())
-            await asyncio.sleep(self.context.app_config.blocking_window_duration_sec)
+            await asyncio.sleep(self.context.app_config.blocking_check_timeout_sec)
 
 
 class BackgroundReleaseUsersMonitoring(BaseState):

example.env

@@ -68,6 +68,11 @@ BLOCKING_TYPES=["tft"]
 # users should be fetched for the period 2025-01-01 00:00:00 to 2025-01-01 00:00:10
 BLOCKING_WINDOW_DURATION_SEC=10
 
+# The parameter defines the interval in seconds for how often we perform
+# new ClickHouse data collection and force detectors to
+# validate the blocking model
+BLOCKING_CHECK_TIMEOUT_SEC=10
+
 # The name of the blocked IPs group in IPSet,
 # and the corresponding group name for blocked IPs in nftables.
 BLOCKING_IPSET_NAME="tempesta_blocked_ips"

tests/conftest.py

@@ -13,6 +13,7 @@ async def access_log() -> ClickhouseAccessLog:
     await _access_log.bot_white_list_create_table()
     await _access_log.bot_white_list_ip_trie_create()
     await _access_log.bot_white_list_ip_trie_refresh()
+    await _access_log.access_log_create()
 
     yield _access_log
 

utils/access_log.py

@@ -71,7 +71,7 @@ async def persistent_users_table_drop(self):
     async def persistent_users_table_truncate(self):
         return await self.conn.query(
             """
-            truncate table persistent_users
+            truncate table if exists persistent_users
             """
         )
 
@@ -108,7 +108,7 @@ async def user_agents_table_create(self):
     async def user_agents_table_truncate(self):
         return await self.conn.query(
             """
-            truncate table user_agents
+            truncate table if exists  user_agents
             """
         )
 
@@ -126,7 +126,7 @@ async def user_agents_all(self):
         )
 
     async def access_log_truncate(self):
-        return await self.conn.query("truncate table access_log")
+        return await self.conn.query("truncate table if exists access_log")
 
     async def blocked_users_create_table(self):
         return await self.conn.query(
@@ -167,7 +167,7 @@ async def bot_white_list_create_table(self):
         )
 
     async def bot_white_list_truncate(self):
-        return await self.conn.query("truncate table bots_white_list")
+        return await self.conn.query("truncate table if exists bots_white_list")
 
     async def bot_white_list_insert(self, values: list[IPv4or6Network]):
         return await self.conn.insert(
@@ -200,3 +200,23 @@ async def bot_white_list_ip_trie_create(self):
 
     async def bot_white_list_ip_trie_refresh(self):
         return await self.conn.query('SYSTEM RELOAD DICTIONARY bots_white_list_trie;')
+
+    async def access_log_create(self):
+        return await self.conn.query(
+            "CREATE TABLE IF NOT EXISTS access_log "
+            "(timestamp DateTime64(3, 'UTC'),"
+            " address IPv6,"
+            " method UInt8,"
+            " version UInt8,"
+            " status UInt16,"
+            " response_content_length UInt64,"
+            " response_time UInt32,"
+            " vhost String,"
+            " uri String,"
+            " referer String,"
+            " user_agent String,"
+            " tft UInt64,"
+            " tfh UInt64,"
+            " dropped_events UInt64"
+            ") ENGINE = MergeTree() ORDER BY timestamp"
+        )