vrr/ntopng
2
0
Fork 0
mirror of https://github.com/ntop/ntopng.git synced 2026-04-29 15:39:33 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Adds Clickhouse support to delete flows upon deleting interface data

Browse source
This commit is contained in:
Simone Mainardi 2021-11-10 14:29:06 +01:00
parent 62ee3437af
commit 94a970f626
1 changed files with 2 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

20
scripts/lua/modules/db_utils.lua
View file

@ -815,24 +815,8 @@ end
-- ########################################################
local function _harvest_expired_clickhouse_flows(ifname, mysql_retention, verbose)
local day_aligned_retention = mysql_retention - (mysql_retention % 86400)
-- Create a string that identifies the PARTITIONs name of the most recent partition that will be deleted
local retention_yyyymmdd = os.date("%Y%m%d", day_aligned_retention)
-- Deletion is done directly on partitions (Clickhouse database has daily partitions)
-- Query the partitions that need to be deleted. Convert YYYYMMDD strings into integers so that
-- only relevant partitions can be queried and deleted
-- The last condition > 999999 prevents old partitions created as YYYMM to be deleted
local partitions_q = string.format("SELECT DISTINCT database, table, toUInt32(partition) drop_part FROM system.parts WHERE active AND table='flows' AND database='ntopng' AND drop_part <= %u AND drop_part > 999999", retention_yyyymmdd)
local partitions_res = interface.execSQLQuery(partitions_q)
-- Iterate queried partitions and delete them
for _, partition_info in ipairs(partitions_res) do
local delete_partition_q = string.format("ALTER TABLE %s.%s DROP PARTITION '%s'",
partition_info["database"], partition_info["table"], partition_info["drop_part"])
local delete_partition_res = interface.execSQLQuery(delete_partition_q)
end
local sql = string.format("ALTER TABLE ntopng.flows DELETE WHERE toUnixTimestamp(FIRST_SEEN) < %u AND INTERFACE_ID = %d", mysql_retention, getInterfaceId(ifname))
interface.execSQLQuery(sql)
end
-- ########################################################