diff --git a/.gitignore b/.gitignore index a71afd6..2d4067f 100644 --- a/.gitignore +++ b/.gitignore @@ -22,14 +22,15 @@ out/ *.db *.sqlite *.sqlite3 -client_state.json -*.tmp - # Live configuration with generated private keys & tokens (samples are tracked) server_config.json client_config.json +client_state.json +local_collector_state.json test_*.json test_*.db +test_pipeline_master_ca/ +test_pipeline_certs/ master_ca_certs/ .env *.env diff --git a/src/Server.py b/src/Server.py index accbd7e..eb5fa50 100644 --- a/src/Server.py +++ b/src/Server.py @@ -1879,20 +1879,12 @@ async def handle_socket_client(reader: asyncio.StreamReader, writer: asyncio.Str if client_id and "server" not in log_payload: log_payload["server"] = client_id - INGEST_QUEUE.push(log_payload) - if len(INGEST_QUEUE._buffer) >= INGEST_QUEUE.batch_size: - res = INGEST_QUEUE.flush( - db_path=SERVER_STATE["config"]["db_path"], - window_hours=SERVER_STATE["config"]["evaluation_window_hours"], - min_runs=SERVER_STATE["config"]["min_persistence_runs"] - ) - else: - res = { - "status": "success", - "buffered": True, - "processed": len(log_payload.get("logs", [])), - "queue_size": len(INGEST_QUEUE._buffer) - } + res = process_ingested_logs( + log_payload, + db_path=SERVER_STATE["config"]["db_path"], + window_hours=SERVER_STATE["config"]["evaluation_window_hours"], + min_runs=SERVER_STATE["config"]["min_persistence_runs"] + ) resp_bytes = json.dumps(res).encode("utf-8") writer.write(struct.pack(">I", len(resp_bytes)) + resp_bytes)