Hallo,
ich hab eine collectd Instanz welche Daten in eine PG Datenbank loggt. Mein Problem ist, das der Prozess der Connection fast immer bei 100% hängt.
Die Probleme die ich habe:
- Die timestamp Werte kommen nicht 100% gleich rein bei zusammengehörigen Werten. Dadurch benötige ich ein Epsilon um zusammengehörige Werte zu gruppieren. Das macht das Anti Join in der Funktion zu einem Bottleneck.
- Die Werte müssen normalisiert werden und auf die richtigen Spalten gemappt werden.
- Die Statements müssen häufig zusammengebastelt werden und dann für jeden Wert neu geparst und ausgeführt werden.
Mit pg_stat_statements hab ich die durchschnittliche Ausführungsdauer von 110 ms auf 69 ms gedrückt. Ein reiner Insert in eine Tabelle ohne Irgendein Mapping ist bei 0.05 ms.
SELECT calls, total_exec_time, total_exec_time/calls as avg_time, query
FROM pg_stat_statements
ORDER BY total_exec_time DESC LIMIT 1;
18194 1277731.538421991 70.22818173144944 "CALL collectd_new_insert_v6($1, $2, $3, $4, $5, $6, $7, $8, $9)"
Vielleicht bringt es was nur sekundengenaue Timestamp Werte einzutragen? Wäre dann der Antijoin schneller?
Gibt es eine Möglichkeit diese Statements dynamisch vorzubereiten und dann immer wieder auszuführen? Ich habe leider keine Möglichkeit das in der Anwendung zu machen…
Hierzu nutze ich folgende Funktion:
CREATE OR REPLACE PROCEDURE public.collectd_new_insert_v6(
IN timestamp with time zone,
IN character varying,
IN character varying,
IN character varying,
IN character varying,
IN character varying,
IN character varying[],
IN character varying[],
IN double precision[])
LANGUAGE 'plpgsql'
AS $BODY$
DECLARE
p_time alias for $1;
p_host alias for $2;
p_plugin alias for $3;
p_plugin_inst alias for $4;
p_type alias for $5;
p_type_inst alias for $6;
p_value_names alias for $7;
-- don't use the type info; for 'StoreRates true' it's 'gauge' anyway
-- p_type_names alias for $8;
p_values alias for $9;
-- working variables
i integer;
stmt text;
cnt integer;
line RECORD;
ds_id integer;
epsilon interval;
tstamp_l timestamp with time zone;
tstamp_h timestamp with time zone;
BEGIN
----------------------------------------------------------------------
-- create the temporary table
----------------------------------------------------------------------
CREATE temporary TABLE IF NOT EXISTS temp_incoming
(
tstamp timestamp with time zone,
host text,
plugin text,
plugin_inst text,
tbl text,
tbl_col text,
val double precision
);
----------------------------------------------------------------------
-- Insert data
----------------------------------------------------------------------
i := 1;
LOOP
EXIT WHEN i > array_upper(p_value_names, 1);
SELECT * INTO line FROM impl_location(p_plugin, p_plugin_inst, p_type, p_type_inst, p_value_names[i]);
insert into temp_incoming (tstamp, host, plugin, plugin_inst, tbl, tbl_col, val)
VALUES (p_time, p_host, p_plugin, line.corr_plugin_inst, line.tbl, line.tbl_col, p_values[i]);
-- continue the loop
i := i + 1;
END LOOP;
-- do a early commit as we might not finish the rest
COMMIT;
----------------------------------------------------------------------
-- Determine if we need to move to the tables
----------------------------------------------------------------------
SELECT COUNT(*) INTO cnt FROM temp_incoming;
IF cnt >= 20 THEN
----------------------------------------------------------------------
-- create the tables
----------------------------------------------------------------------
--RAISE INFO 'Create the plugin table and tstamp idx';
FOR line IN
SELECT tbl
FROM temp_incoming
GROUP BY tbl
ORDER BY tbl
LOOP
-- Create the plugin table
stmt = format(
'
CREATE TABLE IF NOT EXISTS %I
(
id smallint NOT NULL,
tstamp timestamp with time zone NOT NULL,
FOREIGN KEY (id) REFERENCES instance (id)
);
', line.tbl);
EXECUTE stmt;
-- Create the tstamp_id_idx
stmt = format(
'
CREATE INDEX IF NOT EXISTS %s_tstamp_id_idx
ON %I USING brin
(tstamp,id);
',line.tbl, line.tbl);
EXECUTE stmt;
END LOOP;
----------------------------------------------------------------------
-- Create the columns
----------------------------------------------------------------------
--RAISE INFO 'Create the columns';
FOR line IN
SELECT tbl,tbl_col
FROM temp_incoming
GROUP BY tbl,tbl_col
ORDER BY tbl,tbl_col
LOOP
stmt = format(
'ALTER table %I ADD COLUMN IF NOT EXISTS %I real DEFAULT NULL;',
line.tbl, line.tbl_col);
EXECUTE stmt;
END LOOP;
----------------------------------------------------------------------
-- Create the instances
----------------------------------------------------------------------
--RAISE INFO 'Create the instances';
INSERT INTO instance (host,plugin,plugin_inst)
SELECT host,plugin, plugin_inst
FROM temp_incoming
WHERE impl_instance_id(host, plugin, plugin_inst) IS NULL
GROUP BY host,plugin, plugin_inst
ORDER BY host,plugin, plugin_inst;
----------------------------------------------------------------------
-- Move the info -- Anti Join TAOP book
----------------------------------------------------------------------
--RAISE INFO 'Move incoming to tables';
epsilon = '0.5 seconds'::interval;
FOR line IN
SELECT *
FROM temp_incoming
ORDER BY tstamp, host, plugin
LOOP
SELECT impl_instance_id(line.host, line.plugin, line.plugin_inst) INTO ds_id;
SELECT line.tstamp - epsilon INTO tstamp_l;
SELECT line.tstamp + epsilon INTO tstamp_h;
stmt = format(
'
WITH upd AS
(
UPDATE %I
SET %I=%L
WHERE %L <= tstamp AND tstamp < %L
AND id=%L
returning 1
),
ins AS (
INSERT INTO %I (tstamp,id,%I)
SELECT %L,%L,%L
WHERE NOT EXISTS
(
SELECT 1 FROM %I
WHERE %L <= tstamp AND tstamp < %L
AND id=%L
)
returning 1
)
select (select count(*) from upd) as updates,
(select count(*) from ins) as inserts;
', line.tbl,
line.tbl_col, line.val,
tstamp_l, tstamp_h,
ds_id,
line.tbl, line.tbl_col,
line.tstamp, ds_id, line.val,
line.tbl,
tstamp_l, tstamp_h, ds_id
);
EXECUTE stmt;
END LOOP;
-- We are done, clear the incoming table
DELETE FROM temp_incoming;
COMMIT;
END IF;
END;
$BODY$;