2

我有以下用于基本微审计系统的小型 MVWE，它运行良好但缺少功能：

DROP TABLE IF EXISTS audit CASCADE;
CREATE TABLE audit(
    Id          BIGSERIAL  NOT NULL
   ,TimeValue   TIMESTAMP  NOT NULL
   ,RoleName    NAME       NOT NULL
   ,Operation   NAME       NOT NULL
   ,SchemaName  NAME       NOT NULL
   ,TableName   NAME       NOT NULL
   ,Identifiers BIGINT[]
    ---
   ,PRIMARY KEY(Id)
);

-- Audit Trigger:
DROP FUNCTION IF EXISTS audit_trigger() CASCADE;
CREATE OR REPLACE FUNCTION audit_trigger()
RETURNS TRIGGER AS
$BODY$
BEGIN

INSERT INTO audit(TimeValue, RoleName, Operation, SchemaName, TableName) VALUES
(now()::TIMESTAMP, current_user, TG_OP, TG_TABLE_SCHEMA, TG_RELNAME);
RETURN NULL;

END;
$BODY$
LANGUAGE plpgsql SECURITY DEFINER;

-- Channels:
DROP TABLE IF EXISTS channels CASCADE;
CREATE TABLE channels(
    Id          INTEGER    NOT NULL
   ,UserKey     TEXT       NOT NULL
   ,Active      BOOLEAN    NOT NULL   DEFAULT(TRUE)
    ---
   ,PRIMARY KEY(Id)
   ,UNIQUE(UserKey)
);

CREATE TRIGGER channel_audit_trigger BEFORE INSERT OR UPDATE OR DELETE ON channels
FOR EACH STATEMENT EXECUTE PROCEDURE audit_trigger();

-- Perform some operations:
INSERT INTO channels(
SELECT C.Id, 'Channel-' || C.Id
FROM generate_series(1, 300, 10) AS C(Id)
);

DELETE FROM channels WHERE id < 10;

UPDATE channels
SET UserKey = 'wild channel'
WHERE id = 21;

我想在audit表的最后一列中添加已插入/更新/删除的行的标识符channels。

我使用STATEMENT了级别，因为我只需要在数组中收集标识符。但我找不到如何访问 DML 统计信息。相反，在ROW我必须处理的级别OLD和NEW案例中，我无法成功聚合所有接触的标识符。

audit为了用触摸的标识符填充表格的最后一列，我该如何继续？

更新

最后我达到了我的目标，但是这个解决方案可能无法扩展并且可能有一些不必要的缺点（我愿意接受任何建设性的反馈和建议）。

基本上，我是如何解决我的问题的：

  1. DMLROW级别的日志执行到表中；BEFORE audit_rowlevel
  2. 将新添加的内容聚合audit_rowlevel到DMLaudit_statementlevel级别；STATEMENTAFTER

最小的工作示例现在是：

DROP TABLE IF EXISTS audit_rowlevel CASCADE;
CREATE TABLE audit_rowlevel(
    Id          BIGSERIAL  NOT NULL
   ,Aggregated  BOOLEAN    NOT NULL    DEFAULT(FALSE)
   ,TimeValue   TIMESTAMP  NOT NULL
    -- https://www.postgresql.org/docs/current/static/functions-info.html
   ,RoleName    NAME       NOT NULL
   ,ClientIP    INET       NOT NULL
   ,ClientPid   INTEGER    NOT NULL
    -- https://www.postgresql.org/docs/current/static/plpgsql-trigger.html
   ,Operation   TEXT       NOT NULL
   ,SchemaName  NAME       NOT NULL
   ,TableName   NAME       NOT NULL
   ,RowId       BIGINT     NOT NULL
    -- https://www.postgresql.org/docs/current/static/functions-json.html
   ,OldValue    JSONB
   ,NewValue    JSONB
    ---
   ,PRIMARY KEY(Id)
);

-- Row Level Trigger:
DROP FUNCTION IF EXISTS audit_rowlevel_trigger() CASCADE;
CREATE OR REPLACE FUNCTION audit_rowlevel_trigger()
RETURNS TRIGGER AS
$BODY$
DECLARE

history  BOOLEAN := (TG_NARGS > 0) AND (TG_ARGV[0]::BOOLEAN);
rowid    BIGINT;
oldvalue JSONB;
newvalue JSONB;

BEGIN 

-- Handle NEW:
IF TG_OP = ANY('{INSERT,UPDATE}') THEN
    IF history THEN
        newvalue := to_jsonb(NEW);
    END IF;
    rowid := NEW.Id::BIGINT;
END IF;

-- Handle OLD:
IF TG_OP = ANY('{UPDATE,DELETE}') THEN
    IF history THEN
        oldvalue := to_jsonb(OLD);
    END IF;
    rowid := OLD.Id::BIGINT;
END IF;

-- INSERT:
INSERT INTO audit_rowlevel(TimeValue, RoleName, ClientIP, ClientPID, Operation, SchemaName, TableName, RowId, NewValue, OldValue) VALUES
(now()::TIMESTAMP, current_user, inet_client_addr(), pg_backend_pid(), TG_OP, TG_TABLE_SCHEMA, TG_RELNAME, RowId, NewValue, OldValue);

-- RETURN:
IF TG_OP = ANY('{INSERT,UPDATE}') THEN
    RETURN NEW;
ELSIF TG_OP = 'DELETE' THEN
    RETURN OLD;
ELSE
    RETURN NULL;
END IF; 

END;
$BODY$
LANGUAGE plpgsql SECURITY DEFINER;

-- Statement Level Trigger:
DROP TABLE IF EXISTS audit_statementlevel CASCADE;
CREATE TABLE audit_statementlevel(
    Id          BIGSERIAL  NOT NULL
   ,TimeValue   TIMESTAMP  NOT NULL
   ,RoleName    NAME       NOT NULL
   ,ClientIP    INET       NOT NULL
   ,ClientPid   INTEGER    NOT NULL
   ,Operation   TEXT       NOT NULL
   ,SchemaName  NAME       NOT NULL
   ,TableName   NAME       NOT NULL
   ,RowCount    BIGINT     NOT NULL
   ,RowIds      BIGINT[]   NOT NULL
   ,AuditIds    BIGINT[]   NOT NULL
    ---
   ,PRIMARY KEY(Id)
);


-- Row Level Trigger:
DROP FUNCTION IF EXISTS audit_statementlevel_trigger() CASCADE;
CREATE OR REPLACE FUNCTION audit_statementlevel_trigger()
RETURNS TRIGGER AS
$BODY$
DECLARE

rowcount BIGINT;

BEGIN  

WITH

A AS (
SELECT
    TimeValue, RoleName, ClientIP, ClientPid, Operation, SchemaName, TableName
   ,COUNT(*)
   ,array_agg(RowId)
   ,array_agg(Id)
FROM
    audit_rowlevel
WHERE
    NOT Aggregated
GROUP BY
    TimeValue, RoleName, ClientIP, ClientPid, Operation, SchemaName, TableName
ORDER BY
    TimeValue
),

B AS (
INSERT INTO audit_statementlevel(TimeValue, RoleName, ClientIP, ClientPid, Operation, SchemaName, TableName, RowCount, RowIds, AuditIds)
(SELECT * FROM A)
RETURNING AuditIds
),

C AS (
SELECT array_agg(DISTINCT T.id) AS Ids FROM B, unnest(B.AuditIds) AS T(id)
)

UPDATE
    audit_rowlevel
SET
    Aggregated = TRUE
FROM
    C
WHERE
    Id = ANY(C.Ids);

RETURN NULL;

END;
$BODY$
LANGUAGE plpgsql SECURITY DEFINER;


-- Channels:
DROP TABLE IF EXISTS channels CASCADE;
CREATE TABLE channels(
    Id          INTEGER    NOT NULL
   ,UserKey     TEXT       NOT NULL
   ,Active      BOOLEAN    NOT NULL   DEFAULT(TRUE)
    ---
   ,PRIMARY KEY(Id)
   ,UNIQUE(UserKey)
);

CREATE TRIGGER channel_audit_rowlevel_trigger BEFORE INSERT OR UPDATE OR DELETE ON channels
FOR EACH ROW EXECUTE PROCEDURE audit_rowlevel_trigger(TRUE);

CREATE TRIGGER channel_audit_statementlevel_trigger AFTER INSERT OR UPDATE OR DELETE ON channels
FOR EACH STATEMENT EXECUTE PROCEDURE audit_statementlevel_trigger();


-- Perform some operations:
INSERT INTO channels(
SELECT C.Id, 'Channel-' || C.Id
FROM generate_series(1, 300, 10) AS C(Id)
);

DELETE FROM channels WHERE id < 10;

UPDATE channels
SET UserKey = 'wild channel'
WHERE id = 21;

我很想知道这个解决方案是否适合专业开发人员。我是朝着好的方向发展还是这个解决方案是邪恶的？

4

0 回答 0