perf: batch connection inserts in ingest_connections
Replace one INSERT per event (up to FETCH_LIMIT auto-committed round trips per node per poll) with multi-row INSERTs in chunks of 200, wrapped in a single transaction per node. Same data, one commit instead of hundreds, which was making fetch runs slower as the table grew.
This commit is contained in:
1 parent
849360e61c
commit
b0fdd99ad6
1 file changed
+33
-11
+33
-11
@@ -75,19 +75,16 @@ function fetch_node_api(array $node, string $path): mixed {
|
||||
* Ingest connection events from portspoof_py that are strictly newer than
|
||||
* $last_event_at (ISO 8601 string, or null to ingest everything).
|
||||
*
|
||||
* New rows are written with multi-row INSERTs in chunks, inside a single
|
||||
* transaction per call, to keep the number of round trips and commits low.
|
||||
*
|
||||
* Returns ['inserted' => int, 'new_max_ts' => string|null]
|
||||
* where new_max_ts is the DATETIME(6) of the newest row inserted, or null if
|
||||
* nothing new was inserted.
|
||||
*/
|
||||
function ingest_connections(int $node_id, array $events, ?string $last_event_at): array {
|
||||
$inserted = 0;
|
||||
$new_max_ts = null;
|
||||
|
||||
$s = db()->prepare(
|
||||
'INSERT INTO connections
|
||||
(node_id, occurred_at, src_ip, src_port, dst_port, banner_hex, banner_len)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)'
|
||||
);
|
||||
$rows = [];
|
||||
|
||||
foreach ($events as $ev) {
|
||||
$raw_ts = $ev['timestamp'] ?? null;
|
||||
@@ -103,7 +100,7 @@ function ingest_connections(int $node_id, array $events, ?string $last_event_at)
|
||||
continue;
|
||||
}
|
||||
|
||||
$s->execute([
|
||||
$rows[] = [
|
||||
$node_id,
|
||||
$ts,
|
||||
$ev['src_ip'] ?? '',
|
||||
@@ -111,15 +108,40 @@ function ingest_connections(int $node_id, array $events, ?string $last_event_at)
|
||||
(int)($ev['dst_port'] ?? 0),
|
||||
$ev['banner_hex'] ?? null,
|
||||
(int)($ev['banner_len'] ?? 0),
|
||||
]);
|
||||
$inserted++;
|
||||
];
|
||||
|
||||
if ($new_max_ts === null || $ts > $new_max_ts) {
|
||||
$new_max_ts = $ts;
|
||||
}
|
||||
}
|
||||
|
||||
return ['inserted' => $inserted, 'new_max_ts' => $new_max_ts];
|
||||
if ($rows === []) {
|
||||
return ['inserted' => 0, 'new_max_ts' => null];
|
||||
}
|
||||
|
||||
$pdo = db();
|
||||
$pdo->beginTransaction();
|
||||
try {
|
||||
foreach (array_chunk($rows, 200) as $chunk) {
|
||||
$values = implode(', ', array_fill(0, count($chunk), '(?, ?, ?, ?, ?, ?, ?)'));
|
||||
$params = [];
|
||||
foreach ($chunk as $r) {
|
||||
$params = array_merge($params, $r);
|
||||
}
|
||||
$s = $pdo->prepare(
|
||||
'INSERT INTO connections
|
||||
(node_id, occurred_at, src_ip, src_port, dst_port, banner_hex, banner_len)
|
||||
VALUES ' . $values
|
||||
);
|
||||
$s->execute($params);
|
||||
}
|
||||
$pdo->commit();
|
||||
} catch (PDOException $e) {
|
||||
$pdo->rollBack();
|
||||
throw $e;
|
||||
}
|
||||
|
||||
return ['inserted' => count($rows), 'new_max_ts' => $new_max_ts];
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in new issue
Block a user