Аудит изменения данных PostgreSQL

Возникла необходимость вести аудит изменения данных в существующей системе.

Требования:

• Простота подключения/отключения логгирования отдельных таблиц.
• Сократить до минимума изменения в уже существующих функциях БД.
• Минимизировать деградацию производительности.

Первая мысль была добавить в логгируемые таблицы поля _user, _create_date, _delete_date.

На операции INSERT, UPDATE, DELETE повесить триггеры, работающие с этими полями.
При добавлении записи заполнять поля _user и _create_date.

Вместо обновления делать копию обновляемой строки (с измененными значениями), а в самой обновляемой строке заполнять поле _delete_date.

Вместо удаления записи заполнять поле _delete_date.

При обращении к такой таблице в блок WHERE необходимо добавлять _delete_date IS NULL.

Этот механизм мог бы сработать, если бы был заложен в архитектуру БД изначально, но у нас к моменту внедрения логгирования было написано более 3000 функций, в каждую из которых пришлось бы вносить изменения.

Затем появилась идея хранить логи отдельно от данных. Идея заключалась в следующем:
В схеме logs создается копия структуры таблицы плюс несколько служебных полей.

На каждую логгируемую таблицу вешается триггер, который выполняет всю грязную работу по сохранению изменившихся данных.

Код триггера

CREATE OR REPLACE FUNCTION logs.tf_log_table()
RETURNS trigger AS
$BODY$
DECLARE
query text;
safe_table_name text;
BEGIN
SELECT quote_ident(nspname||'.'||relname)
FROM pg_class cl INNER JOIN pg_namespace nsp ON (cl.relnamespace=nsp.oid)
WHERE cl.oid=TG_RELID INTO safe_table_name;

query='INSERT INTO logs.'||safe_table_name||' SELECT ($1).*, now(),$2,session_user;';

IF (TG_OP = 'DELETE')
THEN
	EXECUTE query using OLD,'D';
	RETURN OLD;
ELSIF (TG_OP = 'UPDATE') THEN
	EXECUTE query using OLD,'U';
	RETURN NEW;
ELSIF (TG_OP = 'INSERT') THEN
	EXECUTE query using NEW,'I';
	RETURN NEW;
END IF;

/*Если таблица для логов не создана или ее структура отличается от текущей, то пересоздаем ее, и снова пытаемся записать туда данные*/
EXCEPTION
WHEN SQLSTATE '42P01' OR SQLSTATE '42801' OR SQLSTATE '42804' THEN
	PERFORM logs.create_log_tables(TG_RELID::regclass);
	IF (TG_OP = 'DELETE') THEN
	EXECUTE query using OLD,'D';
	RETURN OLD;
	ELSIF (TG_OP = 'UPDATE') THEN
	EXECUTE query using OLD,'U';
	RETURN NEW;
	ELSIF (TG_OP = 'INSERT') THEN
	EXECUTE query using NEW,'I';
	RETURN NEW;
	END IF;
/* Если что-то другое, игнорируем ошибку и возвращаем стандартный ответ*/
WHEN OTHERS then
IF (TG_OP = 'DELETE') THEN RETURN OLD;
ELSE RETURN NEW;
END IF;
END;
$BODY$
LANGUAGE plpgsql VOLATILE SECURITY DEFINER;

В этом триггере есть несколько конструкций, свойственных только plpgsql, попробую расписать их более подробно.

SELECT quote_ident(nspname||'.'||relname)
FROM pg_class cl INNER JOIN pg_namespace nsp ON (cl.relnamespace=nsp.oid)
WHERE cl.oid=TG_RELID INTO safe_table_name;

TG_RELID специальная переменная, существующая только при срабатывании триггерной функции, в ней хранится идентификатор таблицы, которая вызвала триггер.
С ее помощью мы генерируем имя таблицы, в которую будут записаны логи.

query='INSERT INTO logs.'||safe_table_name||' SELECT ($1).*, now(),$2,session_user;';

Для вставки самих данных используется динамический SQL.
На место переменной $1 подставляются данные из строки, на которой сработал триггер (туда подставляется целиком вся строка, ее необходимо развернуть на отдельные поля — это делается конструкцией (ROW).*
now() — функция, возвращающая время начала транзакции.
session_user — имя пользователя текущего сеанса

IF (TG_OP = 'DELETE')
THEN
EXECUTE query USING OLD,'D';

TG_OP — еще одна переменная, существующая только в триггерных функциях, в ней хранится имя операции, от которой сработал триггер (INSERT, UPDATE, DELETE или TRUNCATE)
OLD, NEW — в этих переменных хранятся старая и новая версия строки.

Дальше, на тот случай, если что-то пойдет не так, предусмотрена достаточно простая обработка ошибок:

EXCEPTION
WHEN SQLSTATE '42P01' OR SQLSTATE '42801' OR SQLSTATE '42804' THEN
PERFORM logs.create_log_tables(TG_RELID::regclass);

Если была изменена структура таблицы, или же по какой-то причине таблица с логами не была создана, то она создается заново и выполняется попытка записать в нее лог.
Во всех остальных случаях при ошибке процедура логгирования игнорируется.

Код функции подключающей логгирование

CREATE OR REPLACE FUNCTION logs.create_log_tables(table_oid oid) RETURNS int AS $BODY$
DECLARE

log_namespace oid=(SELECT oid from pg_namespace WHERE nspname='logs');
p_relname text;
new_tbl_name text;
safe_table_name text;

BEGIN

SELECT relname FROM pg_class WHERE oid=table_oid INTO p_relname;
SELECT quote_ident(nspname||'.'||relname) FROM pg_class cl inner join pg_namespace nsp ON (cl.relnamespace=nsp.oid) where cl.oid=table_oid INTO safe_table_name;
/*Генерация нового имени для таблицы*/

SELECT safe_table_name||'_'||(now()::date)::text||'('||i||')' FROM generate_series(1,10) a(i)
WHERE safe_table_name||'_'||(now()::date)::text||'('||i||')' not in(select relname from pg_class where relnamespace=log_namespace and relpersistence='p')
ORDER BY i LIMIT 1 INTO new_tbl_name;

/*Переименовываем старую таблицу с логами*/
EXECUTE 'ALTER TABLE IF EXISTS logs.'||safe_table_name|| ' RENAME TO '||quote_ident(new_tbl_name)||';';

/*Создаем таблицу с такой же структурой, как логгируемая, плюс служебные поля*/
EXECUTE 'create table logs.'||safe_table_name||' (like '||table_oid::regclass||');';

EXECUTE 'ALTER TABLE logs.'||safe_table_name||' ADD COLUMN "'||p_relname||'_timestamp" timestamp with time zone;';
EXECUTE 'ALTER TABLE logs.'||safe_table_name||' ADD COLUMN "'||p_relname||'_operation" char;';
EXECUTE 'ALTER TABLE logs.'||safe_table_name||' ADD COLUMN "'||p_relname||'_user" text;';


/*Подключаем триггер*/
EXECUTE '
DROP TRIGGER IF exists tr_log_table ON '||table_oid::regclass::text||';
CREATE TRIGGER tr_log_table
BEFORE UPDATE OR DELETE OR INSERT
ON '||table_oid::regclass::text||'
FOR EACH ROW
EXECUTE PROCEDURE logs.tf_log_table();';

RETURN 0;
end;
$BODY$ LANGUAGE plpgsql VOLATILE SECURITY DEFINER;

Алгоритм подключения логгирования достаточно прост. Для начала на основе имени логгируемой таблицы создается ее клон (если таблица с таким именем уже существовала, старая таблица переименовывается), затем в этот клон добавляются необходимые служебные поля, и на логгируемую таблицу подключается триггер.

Плюсы такого варианта:

• Не нужно ничего менять в существующих функциях.
• Запросы на выборку никак не пострадают по производительности.
• При изменениях в структуре логгируемой таблицы таблица с логами будет автоматически пересоздана.
• Логи можно быстро очистить, удалив старые таблицы.