Helper functions and views to use the log_fdw in AWS RDS

log_fdw_helpers.sql

-- Create the extension if it doesn't exist
CREATE EXTENSION IF NOT EXISTS log_fdw;


-- Create the server if it doesn't exist
DO $$
BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_foreign_server WHERE srvname = 'log_fdw_server'
    ) THEN
        CREATE SERVER log_fdw_server FOREIGN DATA WRAPPER log_fdw;
    END IF;
END
$$;


-- PostgreSQL Log Analysis Functions
-- This file contains approved functions for analyzing PostgreSQL log files.
-- Function to find log files for a specific date
CREATE OR REPLACE FUNCTION find_log_files (target_date DATE DEFAULT CURRENT_DATE) RETURNS TABLE (file_name TEXT, file_size_bytes BIGINT) AS $$
BEGIN
    RETURN QUERY
    SELECT
        l.file_name,
        l.file_size_bytes
    FROM list_postgres_log_files() l
    WHERE
        -- Match both possible formats:
        -- postgresql.log.YYYY-MM-DD* or postgresql-YYYY-MM-DD.*
        l.file_name LIKE 'postgresql%' || to_char(target_date, 'YYYY-MM-DD') || '%'
    ORDER BY
        -- Order by name which includes timestamp
        l.file_name DESC;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;


-- Function to query logs from all files for a specific date
CREATE OR REPLACE FUNCTION query_logs (target_date DATE DEFAULT CURRENT_DATE) RETURNS TABLE (
    log_time TIMESTAMP WITH TIME ZONE,
    severity TEXT,
    process_id INTEGER,
    database_name TEXT,
    user_name TEXT,
    message TEXT,
    detail TEXT,
    raw_log TEXT
) AS $$
DECLARE
    log_file RECORD;
BEGIN
    -- Create temporary table to hold results
    DROP TABLE IF EXISTS log_results;
    CREATE TEMP TABLE log_results (
        log_time TIMESTAMP WITH TIME ZONE,
        severity TEXT,
        process_id INTEGER,
        database_name TEXT,
        user_name TEXT,
        message TEXT,
        detail TEXT,
        raw_log TEXT
    );

    -- Process each log file
    FOR log_file IN SELECT * FROM find_log_files(target_date) ORDER BY file_name DESC
    LOOP
        BEGIN
            -- Create foreign table
            PERFORM create_foreign_table_for_log_file(
                'current_log_table',
                'log_fdw_server',
                log_file.file_name
            );

            -- Insert parsed results into our temp table
            WITH raw_logs AS (
                SELECT log_entry FROM current_log_table WHERE log_entry IS NOT NULL
            ),
            parsed AS (
                SELECT
                    log_entry,
                    -- Extract timestamp
                    (SUBSTRING(log_entry FROM '^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})'::text))::timestamp with time zone AS log_time,
                    -- Extract severity
                    COALESCE(SUBSTRING(log_entry FROM '(ERROR|WARNING|INFO|LOG|FATAL|PANIC|DEBUG)'), 'LOG') AS severity,
                    -- Extract process ID
                    (SUBSTRING(log_entry FROM '\[(\d+)\]'))::integer AS process_id,
                    -- Extract process info parts (after timestamp, before message)
                    SUBSTRING(log_entry FROM '\d{2}:\d{2}:\d{2}[^:]+:([^:]+)') AS process_info,
                    -- Extract message (everything after severity)
                    TRIM(SUBSTRING(log_entry FROM '(?:ERROR|WARNING|INFO|LOG|FATAL|PANIC|DEBUG):\s+(.+)$')) AS clean_message
                FROM raw_logs
            )
            INSERT INTO log_results
            SELECT
                p.log_time,
                p.severity,
                p.process_id,
                -- Extract database name (if exists)
                CASE
                    WHEN p.process_info ~ '@' THEN
                        NULLIF(TRIM(SUBSTRING(p.process_info FROM '([^@]+)@')), '')
                    ELSE NULL
                END AS database_name,
                -- Extract username (if exists)
                CASE
                    WHEN p.process_info ~ '@' THEN
                        NULLIF(TRIM(SUBSTRING(p.process_info FROM '@([^:]+)')), '')
                    ELSE NULL
                END AS user_name,
                -- Use clean message
                p.clean_message AS message,
                -- Extract detail
                CASE
                    WHEN p.log_entry LIKE '%DETAIL:%' THEN
                        SUBSTRING(p.log_entry FROM 'DETAIL:\s+([^\n]+)')
                    ELSE NULL
                END AS detail,
                -- Include raw log
                p.log_entry AS raw_log
            FROM parsed p
            WHERE p.log_time::date = target_date;

            -- Drop the temporary foreign table
            DROP FOREIGN TABLE IF EXISTS current_log_table;

            -- Log success
            RAISE NOTICE 'Successfully processed file: %', log_file.file_name;

        EXCEPTION WHEN OTHERS THEN
            -- Log error and continue with next file
            RAISE WARNING 'Error processing file % (size: % bytes): %',
                log_file.file_name,
                log_file.file_size_bytes,
                SQLERRM;
            -- Ensure table is dropped even on error
            DROP FOREIGN TABLE IF EXISTS current_log_table;
        END;
    END LOOP;

    -- Return results ordered by timestamp
    RETURN QUERY
    SELECT r.* FROM log_results r
    ORDER BY r.log_time DESC;

    -- Clean up
    DROP TABLE IF EXISTS log_results;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;


-- View for the last two days of logs
CREATE OR REPLACE VIEW recent_logs AS
SELECT
    *
FROM
    (
        -- Yesterday's logs
        SELECT
            *
        FROM
            query_logs (CURRENT_DATE - 1)
        UNION ALL
        -- Today's logs
        SELECT
            *
        FROM
            query_logs (CURRENT_DATE)
    ) combined_logs
ORDER BY
    log_time DESC;


-- Grant execute permissions
GRANT
EXECUTE ON FUNCTION find_log_files (DATE) TO PUBLIC;


GRANT
EXECUTE ON FUNCTION query_logs (DATE) TO PUBLIC;


GRANT
SELECT
    ON recent_logs TO PUBLIC;