Set audit sql

datamodel/app/create_app.py

@@ -57,9 +57,10 @@ def create_app(
 
     run_sql("CREATE SCHEMA tww_app;", pg_service)
 
-    run_sql_file("symbology_functions.sql", pg_service)
-    run_sql_file("reach_direction_change.sql", pg_service, variables)
-    run_sql_file("14_geometry_functions.sql", pg_service, variables)
+    run_sql_file("functions/symbology_functions.sql", pg_service)
+    run_sql_file("functions/reach_direction_change.sql", pg_service, variables)
+    run_sql_file("functions/geometry_functions.sql", pg_service, variables)
+    run_sql_file("functions/audit_functions.sql", pg_service, variables)
 
     # open YAML files
     if tww_reach_extra:

datamodel/app/functions/audit_functions.sql

@@ -0,0 +1,295 @@
+-----------------------------------------------
+-----------------------------------------------
+-- Provide audit functions for pg_history_viewer
+-----------------------------------------------
+-----------------------------------------------
+
+CREATE OR REPLACE FUNCTION tww_sys.if_modified_func() RETURNS TRIGGER AS $body$
+DECLARE
+    audit_row tww_sys.logged_actions;
+    include_values BOOLEAN;
+    log_diffs BOOLEAN;
+    h_old hstore;
+    h_new hstore;
+    excluded_cols text[] = ARRAY[]::text[];
+BEGIN
+
+    IF NOT (TG_WHEN IN ('AFTER' , 'INSTEAD OF')) THEN
+        RAISE EXCEPTION 'tww_sys.if_modified_func() may only run as an AFTER trigger';
+    END IF;
+
+    audit_row = ROW(
+        NEXTVAL('tww_sys.logged_actions_event_id_seq'), -- event_id
+        TG_TABLE_SCHEMA::text,                        -- schema_name
+        TG_TABLE_NAME::text,                          -- table_name
+        TG_RELID,                                     -- relation OID for much quicker searches
+        session_user::text,                           -- session_user_name
+        current_timestamp,                            -- action_tstamp_tx
+        statement_timestamp(),                        -- action_tstamp_stm
+        clock_timestamp(),                            -- action_tstamp_clk
+        txid_current(),                               -- transaction ID
+        (SELECT setting FROM pg_settings WHERE name = 'application_name'),
+        inet_client_addr(),                           -- client_addr
+        inet_client_port(),                           -- client_port
+        current_query(),                              -- top-level query or queries (if multistatement) from client
+        substring(TG_OP,1,1),                         -- action
+        NULL, NULL,                                   -- row_data, changed_fields
+        'f'                                           -- statement_only
+        );
+
+    IF NOT TG_ARGV[0]::BOOLEAN IS DISTINCT FROM 'f'::BOOLEAN THEN
+        audit_row.client_query = NULL;
+
+    END IF;
+
+    IF TG_ARGV[1] IS NOT NULL THEN
+        excluded_cols = TG_ARGV[1]::text[];
+    END IF;
+
+    IF (TG_OP = 'UPDATE' AND TG_LEVEL = 'ROW') THEN
+        h_old = hstore(OLD.*) - excluded_cols;
+        audit_row.row_data = h_old;
+        h_new = hstore(NEW.*)- excluded_cols;
+        audit_row.changed_fields =  h_new - h_old;
+
+        IF audit_row.changed_fields = hstore('') THEN
+            -- All changed fields are ignored. Skip this update.
+            RAISE WARNING '[tww_sys.if_modified_func] - Trigger detected NULL hstore. ending';
+            RETURN NULL;
+        END IF;
+  INSERT INTO tww_sys.logged_actions VALUES (audit_row.*);
+  RETURN NEW;
+
+    ELSIF (TG_OP = 'DELETE' AND TG_LEVEL = 'ROW') THEN
+        audit_row.row_data = hstore(OLD.*) - excluded_cols;
+  INSERT INTO tww_sys.logged_actions VALUES (audit_row.*);
+        RETURN OLD;
+
+    ELSIF (TG_OP = 'INSERT' AND TG_LEVEL = 'ROW') THEN
+        audit_row.row_data = hstore(NEW.*) - excluded_cols;
+  INSERT INTO tww_sys.logged_actions VALUES (audit_row.*);
+        RETURN NEW;
+
+    ELSIF (TG_LEVEL = 'STATEMENT' AND TG_OP IN ('INSERT','UPDATE','DELETE','TRUNCATE')) THEN
+        audit_row.statement_only = 't';
+        INSERT INTO tww_sys.logged_actions VALUES (audit_row.*);
+  RETURN NULL;
+
+    ELSE
+        RAISE EXCEPTION USING MESSAGE = '[tww_sys.if_modified_func] - Trigger func added as trigger for unhandled case: '||TG_OP||', '||TG_LEVEL;
+        RETURN NEW;
+    END IF;
+
+
+END;
+$body$
+LANGUAGE plpgsql
+SECURITY DEFINER
+SET search_path = pg_catalog, public;
+
+
+COMMENT ON FUNCTION tww_sys.if_modified_func() IS $body$
+Track changes TO a TABLE at the statement AND/OR row level.
+
+Optional parameters TO TRIGGER IN CREATE TRIGGER call:
+
+param 0: BOOLEAN, whether TO log the query text. default 't'.
+
+param 1: text[], COLUMNS TO IGNORE IN updates. default [].
+
+         Updates TO ignored cols are omitted FROM changed_fields.
+
+         Updates WITH only ignored cols changed are NOT inserted
+         INTO the audit log.
+
+         Almost ALL the processing work IS still done for updates
+         that ignored. If you need to save the load, you need to use
+         WHEN clause on the trigger instead.
+
+         No warning or error is issued if ignored_cols contains columns
+         that do not exist in the target table. This lets you specify
+         a standard set of ignored columns.
+
+There is no parameter to disable logging of values. Add this trigger as
+a 'FOR EACH STATEMENT' rather than 'FOR EACH ROW' trigger if you do not
+want to log row values.
+
+Note that the user name logged is the login role for the session. The audit trigger
+cannot obtain the active role because it is reset by the SECURITY DEFINER invocation
+of the audit trigger its self.
+$body$;
+
+
+
+CREATE OR REPLACE FUNCTION tww_sys.audit_table(target_table regclass, audit_rows BOOLEAN, audit_query_text BOOLEAN, ignored_cols text[]) RETURNS void AS $body$
+DECLARE
+  stm_targets text = 'INSERT OR UPDATE OR DELETE OR TRUNCATE';
+  _q_txt text;
+  _ignored_cols_snip text = '';
+BEGIN
+    EXECUTE 'DROP TRIGGER IF EXISTS audit_trigger_row ON ' || target_table::text;
+    EXECUTE 'DROP TRIGGER IF EXISTS audit_trigger_stm ON ' || target_table::text;
+
+    IF audit_rows THEN
+        IF array_length(ignored_cols,1) > 0 THEN
+            _ignored_cols_snip = ', ' || quote_literal(ignored_cols);
+        END IF;
+        _q_txt = 'CREATE TRIGGER audit_trigger_row AFTER INSERT OR UPDATE OR DELETE ON ' ||
+                 target_table::text ||
+                 ' FOR EACH ROW EXECUTE PROCEDURE tww_sys.if_modified_func(' ||
+                 quote_literal(audit_query_text) || _ignored_cols_snip || ');';
+        RAISE NOTICE '%',_q_txt;
+        EXECUTE _q_txt;
+        stm_targets = 'TRUNCATE';
+    ELSE
+    END IF;
+
+    _q_txt = 'CREATE TRIGGER audit_trigger_stm AFTER ' || stm_targets || ' ON ' ||
+             target_table ||
+             ' FOR EACH STATEMENT EXECUTE PROCEDURE tww_sys.if_modified_func('||
+             quote_literal(audit_query_text) || ');';
+    RAISE NOTICE '%',_q_txt;
+    EXECUTE _q_txt;
+
+    -- store primary key names
+    insert into tww_sys.logged_relations (relation_name, uid_column)
+         select target_table, a.attname
+           from pg_index i
+           join pg_attribute a on a.attrelid = i.indrelid
+                              and a.attnum = any(i.indkey)
+          where i.indrelid = target_table::regclass
+            and i.indisprimary
+          on conflict do nothing;
+END;
+$body$
+LANGUAGE plpgsql;
+
+COMMENT ON FUNCTION tww_sys.audit_table(regclass, BOOLEAN, BOOLEAN, text[]) IS $body$
+ADD auditing support TO a TABLE.
+
+Arguments:
+   target_table:     TABLE name, schema qualified IF NOT ON search_path
+   audit_rows:       Record each row CHANGE, OR only audit at a statement level
+   audit_query_text: Record the text of the client query that triggered the audit event?
+   ignored_cols:     COLUMNS TO exclude FROM UPDATE diffs, IGNORE updates that CHANGE only ignored cols.
+$body$;
+
+-- Pg doesn't allow variadic calls with 0 params, so provide a wrapper
+CREATE OR REPLACE FUNCTION tww_sys.audit_table(target_table regclass, audit_rows BOOLEAN, audit_query_text BOOLEAN) RETURNS void AS $body$
+SELECT tww_sys.audit_table($1, $2, $3, ARRAY[]::text[]);
+$body$ LANGUAGE SQL;
+
+-- And provide a convenience call wrapper for the simplest case
+-- of row-level logging with no excluded cols and query logging enabled.
+--
+CREATE OR REPLACE FUNCTION tww_sys.audit_table(target_table regclass) RETURNS void AS $body$
+SELECT tww_sys.audit_table($1, BOOLEAN 't', BOOLEAN 't');
+$body$ LANGUAGE 'sql';
+
+COMMENT ON FUNCTION tww_sys.audit_table(regclass) IS $body$
+Add auditing support to the given table. Row-level changes will be logged with full client query text. No cols are ignored.
+$body$;
+
+CREATE OR REPLACE FUNCTION tww_sys.unaudit_table(target_table regclass) RETURNS void AS $body$
+BEGIN
+    EXECUTE 'DROP TRIGGER IF EXISTS audit_trigger_row ON ' || target_table::text;
+    EXECUTE 'DROP TRIGGER IF EXISTS audit_trigger_stm ON ' || target_table::text;
+END;
+$body$
+LANGUAGE plpgsql;
+
+
+CREATE OR REPLACE FUNCTION tww_sys.replay_event(pevent_id int) RETURNS void AS $body$
+DECLARE
+  query text;
+BEGIN
+    with
+    event as (
+        select * from tww_sys.logged_actions where event_id = pevent_id
+    )
+    -- get primary key names
+    , where_pks as (
+        select array_to_string(array_agg(uid_column || '=' || quote_literal(row_data->uid_column)), ' AND ') as where_clause
+          from tww_sys.logged_relations r
+          join event on relation_name = (schema_name || '.' || table_name)
+    )
+    select into query
+        case
+            when action = 'I' then
+                'INSERT INTO ' || schema_name || '.' || table_name ||
+                ' ('||(select string_agg(key, ',') from each(row_data))||') VALUES ' ||
+                '('||(select string_agg(case when value is null then 'null' else quote_literal(value) end, ',') from each(row_data))||')'
+            when action = 'D' then
+                'DELETE FROM ' || schema_name || '.' || table_name ||
+                ' WHERE ' || where_clause
+            when action = 'U' then
+                'UPDATE ' || schema_name || '.' || table_name ||
+                ' SET ' || (select string_agg(key || '=' || case when value is null then 'null' else quote_literal(value) end, ',') from each(changed_fields)) ||
+                ' WHERE ' || where_clause
+        end
+    from
+        event, where_pks
+    ;
+
+    execute query;
+END;
+$body$
+LANGUAGE plpgsql;
+
+COMMENT ON FUNCTION tww_sys.replay_event(int) IS $body$
+Replay a logged event.
+
+Arguments:
+   pevent_id:  The event_id of the event in tww_sys.logged_actions to replay
+$body$;
+
+CREATE OR REPLACE FUNCTION tww_sys.audit_view(target_view regclass, audit_query_text BOOLEAN, ignored_cols text[], uid_cols text[]) RETURNS void AS $body$
+DECLARE
+  stm_targets text = 'INSERT OR UPDATE OR DELETE';
+  _q_txt text;
+  _ignored_cols_snip text = '';
+
+BEGIN
+    EXECUTE 'DROP TRIGGER IF EXISTS audit_trigger_row ON ' || target_view::text;
+    EXECUTE 'DROP TRIGGER IF EXISTS audit_trigger_stm ON ' || target_view::text;
+
+	IF array_length(ignored_cols,1) > 0 THEN
+	    _ignored_cols_snip = ', ' || quote_literal(ignored_cols);
+	END IF;
+	_q_txt = 'CREATE TRIGGER audit_trigger_row INSTEAD OF INSERT OR UPDATE OR DELETE ON ' ||
+		 target_view::TEXT ||
+		 ' FOR EACH ROW EXECUTE PROCEDURE tww_sys.if_modified_func(' ||
+		 quote_literal(audit_query_text) || _ignored_cols_snip || ');';
+	RAISE NOTICE '%',_q_txt;
+	EXECUTE _q_txt;
+
+    -- store uid columns if not already present
+  IF (select count(*) from tww_sys.logged_relations where relation_name = (select target_view)::text AND  uid_column= (select unnest(uid_cols))::text) = 0 THEN
+      insert into tww_sys.logged_relations (relation_name, uid_column)
+       select target_view, unnest(uid_cols);
+  END IF;
+
+END;
+$body$
+LANGUAGE plpgsql;
+
+COMMENT ON FUNCTION tww_sys.audit_view(regclass, BOOLEAN, text[], text[]) IS $body$
+ADD auditing support TO a VIEW.
+
+Arguments:
+   target_view:      TABLE name, schema qualified IF NOT ON search_path
+   audit_query_text: Record the text of the client query that triggered the audit event?
+   ignored_cols:     COLUMNS TO exclude FROM UPDATE diffs, IGNORE updates that CHANGE only ignored cols.
+   uid_cols:         MANDATORY COLUMNS to use to uniquely identify a row from the view (in order to replay UPDATE and DELETE)
+
+Example:
+  SELECT tww_sys.audit_view('tww_app.vw_tww_wastewater_structure', 'true'::BOOLEAN, 'field_to_ignore'::text[], 'obj_id'::text[])
+$body$;
+
+CREATE OR REPLACE FUNCTION tww_sys.unaudit_view(target_view regclass) RETURNS void AS $body$
+BEGIN
+    EXECUTE 'DROP TRIGGER IF EXISTS audit_trigger_row ON ' || target_view::text;
+    EXECUTE 'DROP TRIGGER IF EXISTS audit_trigger_stm ON ' || target_view::text;
+END;
+$body$
+LANGUAGE plpgsql;

datamodel/changelogs/0001/90_audit.sql

@@ -0,0 +1,79 @@
+--
+-- TODO : Is this comment still relevant :
+-- Audited data. Lots of information is available, it's just a matter of how much
+-- you really want to record. See:
+--
+--   http://www.postgresql.org/docs/9.1/static/functions-info.html
+--
+-- Remember, every column you add takes up more audit table space and slows audit
+-- inserts.
+--
+-- Every index you add has a big impact too, so avoid adding indexes to the
+-- audit table unless you REALLY need them. The hstore GIST indexes are
+-- particularly expensive.
+--
+-- It is sometimes worth copying the audit table, or a coarse subset of it that
+-- you're interested in, into a temporary table where you CREATE any useful
+-- indexes and do your analysis.
+--
+
+BEGIN;
+
+DROP TABLE IF EXISTS tww_sys.logged_actions;
+
+CREATE TABLE tww_sys.logged_actions (
+    event_id bigserial PRIMARY KEY,
+    schema_name text NOT NULL,
+    table_name text NOT NULL,
+    relid oid NOT NULL,
+    session_user_name text,
+    action_tstamp_tx TIMESTAMP WITH TIME ZONE NOT NULL,
+    action_tstamp_stm TIMESTAMP WITH TIME ZONE NOT NULL,
+    action_tstamp_clk TIMESTAMP WITH TIME ZONE NOT NULL,
+    transaction_id bigint,
+    application_name text,
+    client_addr inet,
+    client_port integer,
+    client_query text NOT NULL,
+    action TEXT NOT NULL CHECK (action IN ('I','D','U','T')),
+    row_data hstore,
+    changed_fields hstore,
+    statement_only BOOLEAN NOT NULL
+);
+
+REVOKE ALL ON tww_sys.logged_actions FROM public;
+
+COMMENT ON TABLE tww_sys.logged_actions IS 'History of auditable actions on audited tables, from tww_sys.if_modified_func()';
+COMMENT ON COLUMN tww_sys.logged_actions.event_id IS 'Unique identifier for each auditable event';
+COMMENT ON COLUMN tww_sys.logged_actions.schema_name IS 'Database schema audited table for this event is in';
+COMMENT ON COLUMN tww_sys.logged_actions.table_name IS 'Non-schema-qualified table name of table event occured in';
+COMMENT ON COLUMN tww_sys.logged_actions.relid IS 'Table OID. Changes with drop/create. Get with ''tablename''::regclass';
+COMMENT ON COLUMN tww_sys.logged_actions.session_user_name IS 'Login / session user whose statement caused the audited event';
+COMMENT ON COLUMN tww_sys.logged_actions.action_tstamp_tx IS 'Transaction start timestamp for tx in which audited event occurred';
+COMMENT ON COLUMN tww_sys.logged_actions.action_tstamp_stm IS 'Statement start timestamp for tx in which audited event occurred';
+COMMENT ON COLUMN tww_sys.logged_actions.action_tstamp_clk IS 'Wall clock time at which audited event''s trigger call occurred';
+COMMENT ON COLUMN tww_sys.logged_actions.transaction_id IS 'Identifier of transaction that made the change. May wrap, but unique paired with action_tstamp_tx.';
+COMMENT ON COLUMN tww_sys.logged_actions.client_addr IS 'IP address of client that issued query. Null for unix domain socket.';
+COMMENT ON COLUMN tww_sys.logged_actions.client_port IS 'Remote peer IP port address of client that issued query. Undefined for unix socket.';
+COMMENT ON COLUMN tww_sys.logged_actions.client_query IS 'Top-level query that caused this auditable event. May be more than one statement.';
+COMMENT ON COLUMN tww_sys.logged_actions.application_name IS 'Application name set when this audit event occurred. Can be changed in-session by client.';
+COMMENT ON COLUMN tww_sys.logged_actions.action IS 'Action type; I = insert, D = delete, U = update, T = truncate';
+COMMENT ON COLUMN tww_sys.logged_actions.row_data IS 'Record value. Null for statement-level trigger. For INSERT this is the new tuple. For DELETE and UPDATE it is the old tuple.';
+COMMENT ON COLUMN tww_sys.logged_actions.changed_fields IS 'New values of fields changed by UPDATE. Null except for row-level UPDATE events.';
+COMMENT ON COLUMN tww_sys.logged_actions.statement_only IS '''t'' if audit event is from an FOR EACH STATEMENT trigger, ''f'' for FOR EACH ROW';
+
+CREATE INDEX logged_actions_relid_idx ON tww_sys.logged_actions(relid);
+CREATE INDEX logged_actions_action_tstamp_tx_stm_idx ON tww_sys.logged_actions(action_tstamp_stm);
+CREATE INDEX logged_actions_action_idx ON tww_sys.logged_actions(action);
+
+CREATE TABLE tww_sys.logged_relations (
+    relation_name text not null,
+    uid_column text not null,
+    PRIMARY KEY (relation_name, uid_column)
+);
+
+COMMENT ON TABLE tww_sys.logged_relations IS 'Table used to store unique identifier columns for table or views, so that events can be replayed';
+COMMENT ON COLUMN tww_sys.logged_relations.relation_name IS 'Relation (table or view) name (with schema if needed)';
+COMMENT ON COLUMN tww_sys.logged_relations.uid_column IS 'Name of a column that is used to uniquely identify a row in the relation';
+
+COMMIT;