diff --git a/docker/data/test_pcaps/ctu-sample.pcap b/docker/data/test_pcaps/ctu-sample.pcap deleted file mode 100755 index b9b8d19c..00000000 Binary files a/docker/data/test_pcaps/ctu-sample.pcap and /dev/null differ diff --git a/src/alerter/alerter.py b/src/alerter/alerter.py index 54aed993..d0cd40ec 100644 --- a/src/alerter/alerter.py +++ b/src/alerter/alerter.py @@ -331,9 +331,7 @@ def bootstrap_alerter_instance(self): # 2. Executing Base Logging Actions self._log_to_file_action() self._log_to_kafka_action() - self._record_alerter_terminal_events( - server_message_ids - ) + self._record_alerter_terminal_events(server_message_ids) finally: self.clear_data() diff --git a/src/base/kafka/config.py b/src/base/kafka/config.py index 72124bc2..2988c56e 100644 --- a/src/base/kafka/config.py +++ b/src/base/kafka/config.py @@ -56,6 +56,5 @@ def bootstrap_servers() -> str: """Return the configured brokers in confluent-kafka format.""" return ",".join( - f"{broker['hostname']}:{broker['internal_port']}" - for broker in KAFKA_BROKERS + f"{broker['hostname']}:{broker['internal_port']}" for broker in KAFKA_BROKERS ) diff --git a/src/base/kafka/consumer.py b/src/base/kafka/consumer.py index 3a9a5829..60a1d103 100644 --- a/src/base/kafka/consumer.py +++ b/src/base/kafka/consumer.py @@ -124,15 +124,19 @@ def consume_batch( """Fetch a bounded group of records without committing offsets.""" batch_size = max( 1, - kafka_config.KAFKA_TRANSACTION_BATCH_SIZE - if max_messages is None - else int(max_messages), + ( + kafka_config.KAFKA_TRANSACTION_BATCH_SIZE + if max_messages is None + else int(max_messages) + ), ) batch_timeout_ms = max( 0, - kafka_config.KAFKA_TRANSACTION_BATCH_TIMEOUT_MS - if timeout_ms is None - else int(timeout_ms), + ( + kafka_config.KAFKA_TRANSACTION_BATCH_TIMEOUT_MS + if timeout_ms is None + else int(timeout_ms) + ), ) deadline = time.monotonic() + batch_timeout_ms / 1000 records = [] diff --git a/src/base/kafka/topics.py b/src/base/kafka/topics.py index c5f507c0..258b7e36 100644 --- a/src/base/kafka/topics.py +++ b/src/base/kafka/topics.py @@ -203,9 +203,7 @@ def _wait_for_admin_futures(futures: dict, operation: str) -> None: try: future.result() except KafkaException as exception: - if operation == "create topic" and _is_topic_already_created( - exception - ): + if operation == "create topic" and _is_topic_already_created(exception): logger.info("Kafka topic '%s' already exists.", topic) continue if ( diff --git a/src/base/retry.py b/src/base/retry.py index 55a2f844..aadaf1c7 100644 --- a/src/base/retry.py +++ b/src/base/retry.py @@ -47,9 +47,7 @@ def load_retry_settings(config: dict[str, Any]) -> RetrySettings: max_delay_seconds=max( initial_delay, _float_setting(merged, "max_delay_seconds") ), - backoff_multiplier=max( - 1.0, _float_setting(merged, "backoff_multiplier") - ), + backoff_multiplier=max(1.0, _float_setting(merged, "backoff_multiplier")), jitter_seconds=max(0.0, _float_setting(merged, "jitter_seconds")), log_every_attempts=max(1, _int_setting(merged, "log_every_attempts")), ) diff --git a/src/logserver/server.py b/src/logserver/server.py index 56878933..27e54d7f 100644 --- a/src/logserver/server.py +++ b/src/logserver/server.py @@ -124,7 +124,9 @@ def fetch_from_kafka(self) -> None: for source_message in source_messages: value = source_message.value if value is None: - raise ValueError("LogServer received a Kafka record without data.") + raise ValueError( + "LogServer received a Kafka record without data." + ) logger.debug(f"From Kafka: '{value}'") message_id = uuid.uuid4() diff --git a/src/monitoring/monitoring_agent.py b/src/monitoring/monitoring_agent.py index 693c35e0..0e648ea9 100644 --- a/src/monitoring/monitoring_agent.py +++ b/src/monitoring/monitoring_agent.py @@ -27,12 +27,8 @@ "hostname" ] MONITORING_CONSUMER_CONFIG = CONFIG["pipeline"]["monitoring"]["kafka_consumer"] -MONITORING_CONSUMER_BATCH_SIZE = max( - 1, int(MONITORING_CONSUMER_CONFIG["batch_size"]) -) -MONITORING_CONSUMER_TIMEOUT_MS = max( - 0, int(MONITORING_CONSUMER_CONFIG["timeout_ms"]) -) +MONITORING_CONSUMER_BATCH_SIZE = max(1, int(MONITORING_CONSUMER_CONFIG["batch_size"])) +MONITORING_CONSUMER_TIMEOUT_MS = max(0, int(MONITORING_CONSUMER_CONFIG["timeout_ms"])) def prepare_all_tables(): @@ -144,9 +140,7 @@ def run(self) -> None: ) for source_record in source_records: try: - table_name = source_record.topic.removeprefix( - "clickhouse_" - ) + table_name = source_record.topic.removeprefix("clickhouse_") data = self.data_schemas[table_name].loads( source_record.value ) @@ -177,9 +171,7 @@ def build_monitoring_worker(worker_id: str) -> MonitoringAgent: return MonitoringAgent(worker_id=worker_id) -def run_monitoring_worker_process( - process_index: int, threads_per_process: int -) -> None: +def run_monitoring_worker_process(process_index: int, threads_per_process: int) -> None: """Run all monitoring threads assigned to one process.""" run_thread_worker_pool( worker_factory=build_monitoring_worker, diff --git a/tests/logserver/test_server.py b/tests/logserver/test_server.py index eb2490ad..dc631b26 100644 --- a/tests/logserver/test_server.py +++ b/tests/logserver/test_server.py @@ -134,9 +134,7 @@ async def test_fetch_from_kafka( mock_uuid.return_value = mock_uuid_instance mock_uuid.uuid4.return_value = UUID("bd72ccb4-0ef2-4100-aa22-e787122d6875") mock_consume_handler = MagicMock() - source_message = ConsumedKafkaMessage( - "key1", "value1", "test-topic", 0, 1 - ) + source_message = ConsumedKafkaMessage("key1", "value1", "test-topic", 0, 1) mock_consume_handler.consume_batch.side_effect = [ [source_message], _StopFetching(),