| |
| |
|
|
| import time |
| import queue |
| import threading |
|
|
| class LegacyFlowController: |
| """ |
| The Organism: Adaptive Legacy Flow Controller [LK] |
| Throttles and paces telemetry handoffs to match the exact ingestion |
| speed of legacy hard drives and SIEM systems, preventing backpressure. |
| """ |
| def __init__(self, max_batch_size: int = 50, flush_interval_seconds: float = 5.0): |
| self._lock = threading.Lock() |
| self._outbox_queue = queue.Queue() |
| self._max_batch_size = max_batch_size |
| self._flush_interval = flush_interval_seconds |
| self._last_flush_time = time.time() |
|
|
| def enqueue_for_legacy_accountant(self, legacy_formatted_msg: str, priority: int = 2) -> None: |
| """Enqueues formatted legacy messages (Syslog/CEF) for paced delivery.""" |
| with self._lock: |
| self._outbox_queue.put((priority, time.time(), legacy_formatted_msg)) |
|
|
| def get_next_batch(self) -> list: |
| """ |
| Releases data strictly at a controlled pace matching legacy speeds, |
| protecting slow hard drives and traditional log collectors from flooding. |
| """ |
| current_time = time.time() |
| batch = [] |
| |
| with self._lock: |
| if (current_time - self._last_flush_time >= self._flush_interval) or self._outbox_queue.qsize() >= self._max_batch_size: |
| while not self._outbox_queue.empty() and len(batch) < self._max_batch_size: |
| priority, timestamp, msg = self._outbox_queue.get() |
| batch.append(msg) |
| self._last_flush_time = current_time |
|
|
| return batch |
|
|
| if __name__ == "__main__": |
| controller = LegacyFlowController(max_batch_size=3, flush_interval_seconds=1.0) |
| print("=== THE ORGANISM: LEGACY FLOW CONTROLLER INITIALIZED [LK] ===") |
| |
| |
| controller.enqueue_for_legacy_accountant("CEF:0|Digitview|TheOrganism|1.0|SIG01|PROBE|5|src=127.0.0.1") |
| controller.enqueue_for_legacy_accountant("CEF:0|Digitview|TheOrganism|1.0|SIG02|PROBE|5|src=127.0.0.1") |
| controller.enqueue_for_legacy_accountant("CEF:0|Digitview|TheOrganism|1.0|SIG03|CRITICAL|10|src=127.0.0.1") |
| |
| print("Batch 1 (Immediate check):", controller.get_next_batch()) |
| time.sleep(1.1) |
| print("Batch 2 (After flow interval):", controller.get_next_batch()) |
|
|
