Rsyslog output to postgresql

README

A simple configuration to store syslog log data into postgresql using rsyslog.

This works incredible well and uses "stable" packages from Debian Trixie.

Logs are happily stored as received from other Linux distributions like CentOS and OpenWRT, Switches from various vendors (Dell, Cisco, Netgear, Juniper), FreeBSD and Opnsense.

You can easily query logs using your favourite laguage, or grafana, pgadmin4, pgcli, psql etc.

Resource usage is trivial. The rsyslog service uses nominal cpu and memory with very modest disk consumption.

rsyslog.conf

# /etc/rsyslog.conf configuration file for collecting logs via TCP/UDP and writing to PostgreSQL
#
# Thoughts:
# I recommend running a discrete rsyslog just recieving and storing in to postgresql.
# You may have additional input modules but again, I encourage functional seperation between
# local collection on the server and storage. 
# Then run your "normal" rsyslog instance on the same machine configured identically to everything
# else in your network to transmit to it. Or whatever log daemon you like.
# When possible please use TLS.

# Load required modules
module(load="imudp") # UDP syslog reception
module(load="imtcp") # TCP syslog reception
module(load="imrelp") # RELP syslog reception

# Define input for UDP and TCP and RELP
# In all cases please review documentation for their wide range of tunable options
input(type="imudp" port="514")
input(type="imtcp" port="514")
input(type="imrelp" port="2514" maxDataSize="10k")

# This may give higher performance on linux, but no TLS.
# input(type="imptcp" port="514")

# Optional: Also write logs to a file for testing
# *.* /var/log/test

# PostgreSQL output module
module(load="ompgsql")
# FYI. String template example. 
# I have found Property Replacers arent as good as in list templates though.
#template(
#    name="LogToPgSQL" type="string" option.stdsql="on"
#    string="INSERT INTO system_events (ReceivedAt, DeviceReportedTime, Facility, Priority, FromHost, FromIpAddress, Message, EventSource, EventID, SysLogTag, EventLogType) VALUES ('%timereported:::date-rfc3339%', '%timegenerated:::date-rfc3339%', %syslogfacility%, %syslogpriority%, '%fromhost%', '%fromhost-ip%', '%msg%', '%programname%', '%MSGID%', '%syslogtag%', 'syslog')" 
#)

template(
    name="LogToPgSQL" type="list" option.stdsql="on" ) {
    constant(value="INSERT INTO system_events (ReceivedAt, DeviceReportedTime, Facility, Priority, FromHost, FromIpAddress, Message, EventSource, EventID, SysLogTag, EventLogType) VALUES ('")
    property(name="timereported" dateFormat="pgsql" date.inUTC="on")
    constant(value="', '")
    property(name="timegenerated" dateFormat="pgsql" date.inUTC="on")
    constant(value="', ")
    property(name="syslogfacility")
    constant(value=", ")
    property(name="syslogpriority")
    constant(value=", '")
    property(name="fromhost")
    constant(value="', '")
    property(name="fromhost-ip")
    constant(value="', '")
    property(name="msg")
    constant(value="', '")
    property(name="programname")
    constant(value="', '")
    property(name="msgid")
    constant(value="', '")
    property(name="syslogtag")
    constant(value="', 'syslog')")
}

action(
    type="ompgsql" server="localhost"
    user="rsyslog" pass="YOUR PASSWORD HERE"
    db="syslog" template="LogToPgSQL"
)

# CREATE ROLE rsyslog WITH LOGIN NOSUPERUSER INHERIT NOCREATEDB NOCREATEROLE NOREPLICATION NOBYPASSRLS
#  ENCRYPTED PASSWORD 'XXXXXXXX'

syslog.sql

-- Here is a suggested schema.
-- It is based upon the sample schema https://github.com/rsyslog/rsyslog/blob/main/plugins/ompgsql/createDB.sql
-- I dont know what all at the fields are for, I have added some and mapped what I could.
-- 
-- Design choices:
-- I am running this on Postgresql 17 with pg_cron 1.6.7 (the normal packages in Debian Trixie)
-- Uses partitions on a per month basis. These are largely transparent when inserting and selecting.
-- Functions are provided to create and remove partitions
-- Uses pg_cron to manage automatically manage the partitions. This requires 1.6 for "schedule_in_database" function.
-- Paritions are created 3 months in advance
-- You could schedule equivalent calls using system cron if that works better for you.
-- More indexes may be needed
--
-- Install pg_cron (rough) steps:
-- apt-get install postgresql-17-cron
-- add to postgresql.conf: shared_preload_libraries = 'pg_cron"
-- restart postgresql
-- echo "CREATE EXTENSION pg_cron" | sudo -u postgres psql postgres

CREATE DATABASE "syslog" WITH ENCODING 'SQL_ASCII' TEMPLATE template0;

\connect syslog;

CREATE TABLE events_properties
(
    ID serial not null primary key,
    SystemEventID int NULL ,
    ParamName varchar(255) NULL ,
    ParamValue text NULL
);

CREATE TABLE system_events
(
    ID SERIAL NOT NULL,
    CustomerID BIGINT,
    ReceivedAt TIMESTAMP WITH TIME ZONE NOT NULL, -- Partitioning key, NOT NULL
    DeviceReportedTime TIMESTAMP WITH TIME ZONE NULL,
    Facility SMALLINT NULL,
    Priority SMALLINT NULL,
    FromHost VARCHAR(255) NULL,
    FromIpAddress INET NULL,
    Message TEXT,
    NTSeverity INT NULL,
    Importance INT NULL,
    EventSource VARCHAR(60),
    EventUser VARCHAR(60) NULL,
    EventCategory INT NULL,
    EventID VARCHAR(60) NULL,
    EventBinaryData TEXT NULL,
    MaxAvailable INT NULL,
    CurrUsage INT NULL,
    MinUsage INT NULL,
    MaxUsage INT NULL,
    InfoUnitID INT NULL,
    SysLogTag VARCHAR(60),
    EventLogType VARCHAR(60),
    GenericFileName VARCHAR(60),
    SystemID INT NULL,
    PRIMARY KEY (ID, ReceivedAt) -- Include ReceivedAt in the primary key
) PARTITION BY RANGE (ReceivedAt);

CREATE INDEX idx_system_events_facility ON system_events (Facility);
CREATE INDEX idx_system_events_priority ON system_events (Priority);

-- Create table for Facility mappings
CREATE TABLE facilities (
    facility_id SMALLINT PRIMARY KEY,
    facility_name VARCHAR(100) NOT NULL
);

-- Insert common syslog facility values (example mappings, adjust as needed)
INSERT INTO facilities (facility_id, facility_name) VALUES
(0, 'kern'),    -- Kernel messages
(1, 'user'),    -- User-level messages
(2, 'mail'),    -- Mail system
(3, 'daemon'),  -- System daemons
(4, 'auth'),    -- Security/authorization messages
(5, 'syslog'),  -- Messages generated internally by syslogd
(6, 'lpr'),     -- Line printer subsystem
(7, 'news'),    -- Network news subsystem
(8, 'uucp'),    -- UUCP subsystem
(9, 'cron'),    -- Clock daemon
(10, 'authpriv'), -- Security/authorization messages (private)
(11, 'ftp'),     -- FTP daemon
(12, 'ntp'),     -- NTP subsystem
(13, 'security'), -- Log audit
(14, 'console'), -- Log alert
(15, 'solaris-cron'), -- Scheduling daemon (Solaris)
(16, 'local0'),  -- Local use 0
(17, 'local1'),  -- Local use 1
(18, 'local2'),  -- Local use 2
(19, 'local3'),  -- Local use 3
(20, 'local4'),  -- Local use 4
(21, 'local5'),  -- Local use 5
(22, 'local6'),  -- Local use 6
(23, 'local7');  -- Local use 7

-- Create table for Priority mappings
CREATE TABLE priorities (
    priority_id SMALLINT PRIMARY KEY,
    priority_name VARCHAR(100) NOT NULL
);

-- Insert common syslog priority values (example mappings, adjust as needed)
INSERT INTO priorities (priority_id, priority_name) VALUES
(0, 'emerg'),   -- System is unusable
(1, 'alert'),   -- Action must be taken immediately
(2, 'crit'),    -- Critical conditions
(3, 'err'),     -- Error conditions
(4, 'warning'), -- Warning conditions
(5, 'notice'),  -- Normal but significant condition
(6, 'info'),    -- Informational messages
(7, 'debug');   -- Debug-level messages

-- Add foreign key constraints to system_events
ALTER TABLE system_events
ADD CONSTRAINT fk_facility
FOREIGN KEY (Facility) REFERENCES facilities (facility_id);

ALTER TABLE system_events
ADD CONSTRAINT fk_priority
FOREIGN KEY (Priority) REFERENCES priorities (priority_id);

-- A convenient view
CREATE VIEW v_system_events AS
    SELECT se.*, f.facility_name, p.priority_name
    FROM system_events se
    LEFT JOIN facilities f ON se.Facility = f.facility_id
    LEFT JOIN priorities p ON se.Priority = p.priority_id;

-- Create partitions for a few months (e.g., January to March 2025)
CREATE TABLE system_events_2025_01 PARTITION OF system_events
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');

CREATE TABLE system_events_2025_02 PARTITION OF system_events
    FOR VALUES FROM ('2025-02-01') TO ('2025-03-01');

CREATE TABLE system_events_2025_03 PARTITION OF system_events
    FOR VALUES FROM ('2025-03-01') TO ('2025-04-01');

-- Procedure to create a new monthly partition
CREATE OR REPLACE PROCEDURE public.create_monthly_partition(p_table_name TEXT, year INT, month INT)
LANGUAGE plpgsql AS $$
DECLARE
    partition_name TEXT;
    start_date DATE;
    end_date DATE;
BEGIN
    start_date := make_date(year, month, 1);
    end_date := (start_date + INTERVAL '1 month')::DATE;
    partition_name := format('%s_%s_%s', p_table_name, year, lpad(month::TEXT, 2, '0'));
    
    IF NOT EXISTS (
        SELECT FROM pg_tables WHERE tablename = partition_name AND schemaname = 'public'
    ) THEN
        EXECUTE format(
            'CREATE TABLE public.%I PARTITION OF public.%I FOR VALUES FROM (%L) TO (%L)',
            partition_name, p_table_name, start_date, end_date
        );
        RAISE NOTICE 'Created partition public.%', partition_name;
    ELSE
        RAISE NOTICE 'Partition public.% already exists, skipping', partition_name;
    END IF;
EXCEPTION
    WHEN OTHERS THEN
        RAISE WARNING 'Error creating partition public.%: %', partition_name, SQLERRM;
        RAISE;
END;
$$;

-- Procedure to create partitions for the next N months
CREATE OR REPLACE PROCEDURE public.create_future_partitions(p_table_name TEXT, months_ahead INT)
LANGUAGE plpgsql AS $$
DECLARE
    current_date DATE := CURRENT_DATE;
    target_date DATE;
    target_year INT;
    target_month INT;
    i INT;
BEGIN
    FOR i IN 0..months_ahead LOOP
        target_date := current_date + (i * INTERVAL '1 month');
        target_year := EXTRACT(YEAR FROM target_date);
        target_month := EXTRACT(MONTH FROM target_date);
        CALL public.create_monthly_partition(p_table_name, target_year, target_month);
    END LOOP;
EXCEPTION
    WHEN OTHERS THEN
        RAISE WARNING 'Error in create_future_partitions for %: %', p_table_name, SQLERRM;
        RAISE;
END;
$$;

-- Procedure to drop partitions older than a specified interval
CREATE OR REPLACE PROCEDURE public.drop_old_partitions(p_table_name TEXT, retention_interval INTERVAL)
LANGUAGE plpgsql AS $$
DECLARE
    partition_name TEXT;
    dropped_count INT := 0;
BEGIN
    FOR partition_name IN (
        SELECT tables.table_name
        FROM information_schema.tables
        WHERE tables.table_name LIKE p_table_name || '_%'
        AND tables.table_name < p_table_name || '_' || to_char(CURRENT_DATE - retention_interval, 'YYYY_MM')
        AND tables.table_schema = 'public'
    )
    LOOP
        EXECUTE 'DROP TABLE IF EXISTS public.' || quote_ident(partition_name);
        RAISE NOTICE 'Dropped partition public.%', partition_name;
        dropped_count := dropped_count + 1;
    END LOOP;
    IF dropped_count = 0 THEN
        RAISE NOTICE 'No partitions dropped for % older than %', p_table_name, retention_interval;
    END IF;
EXCEPTION
    WHEN OTHERS THEN
        RAISE WARNING 'Error dropping partitions for %: %', p_table_name, SQLERRM;
        RAISE;
END;
$$;

-- Create partitions in advance and clear up old ones (created above as an example)
-- Also checks /now/ that functions are working rather that finding out when your disk fills.
CALL public.create_future_partitions('system_events', 4);
CALL public.drop_old_partitions('system_events', INTERVAL '3 months');

GRANT ALL ON SCHEMA public TO postgres;
GRANT ALL ON TABLE public.system_events TO postgres;
GRANT EXECUTE ON PROCEDURE public.create_monthly_partition(TEXT, INT, INT) TO postgres;
GRANT EXECUTE ON PROCEDURE public.create_future_partitions(TEXT, INT) TO postgres;
GRANT EXECUTE ON PROCEDURE public.drop_old_partitions(TEXT, INTERVAL) TO postgres;

GRANT USAGE ON SCHEMA public TO rsyslog;
GRANT USAGE, SELECT ON SEQUENCE public.system_events_id_seq TO rsyslog;
GRANT INSERT ON TABLE public.system_events TO rsyslog;

-- Run this in the postgres database where pg_cron is installed
\connect postgres

CREATE EXTENSION IF NOT EXISTS pg_cron WITH SCHEMA public;

-- pg_cron job to delete partitions older than 3 months
SELECT cron.unschedule('drop_old_system_events_partitions');
SELECT cron.schedule_in_database(
    'drop_old_system_events_partitions',
    '0 14 1 * *',
    $$CALL public.drop_old_partitions('system_events', INTERVAL '3 months');$$,
    'syslog'
);

-- pg_cron job to create partitions for the next 3 months
SELECT cron.unschedule('create_future_system_events_partitions');
SELECT cron.schedule_in_database(
    'create_future_system_events_partitions',
    '5 14 1 * *',
    $$CALL public.create_future_partitions('system_events', 3);$$,
    'syslog'
);