PostgreSQL table partitioning and flow-export

To export data aggregated by flow-capture to Postgresql, similar table structure must be created:

After this, flow-capture must be launched as following:

#file: /etc/flow-tools/flow-capture.conf
-w /var/flowdata -n 95  -N 3 -R /usr/local/bin/flowdata-export.sh 0/0/555

This means that it will run /usr/local/bin/flowdata-export.sh script every time it rotates its log file. With the first argument as the flow file name after rotating. Here's is the simple script:

#!/bin/sh

LOGS=/var/flowdata
DB=dbuser:dbpass:dbhost:dbport:dbname:dbtable

if [ $# -lt 1 ]; then
    echo "USE: $0 <filename>"
    exit
fi

flow-export -f 5 -m UNIX_SECS,DOCTETS,SRCADDR,DSTADDR,SRCPORT,DSTPORT,PROT -u $DB < $LOGS/$1

There can be a lot of data in db, and it could be wise to split it into smaller chunks. This can be done using postgresql table partitioning. Here's how we can improve our database/table setup:

CREATE TABLE flowdata (
    unix_secs   bigint NOT NULL,
    doctets     bigint NOT NULL,
    srcaddr     inet NOT NULL,
    dstaddr     inet NOT NULL,
    srcport     integer NOT NULL,
    dstport     integer NOT NULL,
    prot        integer NOT NULL
);

CREATE TABLE flowdata_01 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 1 )
) INHERITS (flowdata);

CREATE TABLE flowdata_02 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 2 )
) INHERITS (flowdata);

CREATE TABLE flowdata_03 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 3 )
) INHERITS (flowdata);

CREATE TABLE flowdata_04 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 4 )
) INHERITS (flowdata);

CREATE TABLE flowdata_05 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 5 )
) INHERITS (flowdata);

CREATE TABLE flowdata_06 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 6 )
) INHERITS (flowdata);

CREATE TABLE flowdata_07 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 7 )
) INHERITS (flowdata);

CREATE TABLE flowdata_08 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 8 )
) INHERITS (flowdata);

CREATE TABLE flowdata_09 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 9 )
) INHERITS (flowdata);

CREATE TABLE flowdata_10 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 10 )
) INHERITS (flowdata);

CREATE TABLE flowdata_11 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 11 )
) INHERITS (flowdata);

CREATE TABLE flowdata_12 (
    CHECK ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 12 )
) INHERITS (flowdata);

CREATE INDEX flowdata_01_unix_secs ON flowdata_01 ( unix_secs );
CREATE INDEX flowdata_01_srcaddr ON flowdata_01 ( srcaddr );
CREATE INDEX flowdata_01_dstaddr ON flowdata_01 ( dstaddr );
CREATE INDEX flowdata_01_srcport ON flowdata_01 ( srcport );
CREATE INDEX flowdata_01_dstport ON flowdata_01 ( dstport );
CREATE INDEX flowdata_01_prot ON flowdata_01 ( prot );

CREATE INDEX flowdata_02_unix_secs ON flowdata_02 ( unix_secs );
CREATE INDEX flowdata_02_srcaddr ON flowdata_02 ( srcaddr );
CREATE INDEX flowdata_02_dstaddr ON flowdata_02 ( dstaddr );
CREATE INDEX flowdata_02_srcport ON flowdata_02 ( srcport );
CREATE INDEX flowdata_02_dstport ON flowdata_02 ( dstport );
CREATE INDEX flowdata_02_prot ON flowdata_02 ( prot );

CREATE INDEX flowdata_03_unix_secs ON flowdata_03 ( unix_secs );
CREATE INDEX flowdata_03_srcaddr ON flowdata_03 ( srcaddr );
CREATE INDEX flowdata_03_dstaddr ON flowdata_03 ( dstaddr );
CREATE INDEX flowdata_03_srcport ON flowdata_03 ( srcport );
CREATE INDEX flowdata_03_dstport ON flowdata_03 ( dstport );
CREATE INDEX flowdata_03_prot ON flowdata_03 ( prot );

CREATE INDEX flowdata_04_unix_secs ON flowdata_04 ( unix_secs );
CREATE INDEX flowdata_04_srcaddr ON flowdata_04 ( srcaddr );
CREATE INDEX flowdata_04_dstaddr ON flowdata_04 ( dstaddr );
CREATE INDEX flowdata_04_srcport ON flowdata_04 ( srcport );
CREATE INDEX flowdata_04_dstport ON flowdata_04 ( dstport );
CREATE INDEX flowdata_04_prot ON flowdata_04 ( prot );

CREATE INDEX flowdata_05_unix_secs ON flowdata_05 ( unix_secs );
CREATE INDEX flowdata_05_srcaddr ON flowdata_05 ( srcaddr );
CREATE INDEX flowdata_05_dstaddr ON flowdata_05 ( dstaddr );
CREATE INDEX flowdata_05_srcport ON flowdata_05 ( srcport );
CREATE INDEX flowdata_05_dstport ON flowdata_05 ( dstport );
CREATE INDEX flowdata_05_prot ON flowdata_05 ( prot );

CREATE INDEX flowdata_06_unix_secs ON flowdata_06 ( unix_secs );
CREATE INDEX flowdata_06_srcaddr ON flowdata_06 ( srcaddr );
CREATE INDEX flowdata_06_dstaddr ON flowdata_06 ( dstaddr );
CREATE INDEX flowdata_06_srcport ON flowdata_06 ( srcport );
CREATE INDEX flowdata_06_dstport ON flowdata_06 ( dstport );
CREATE INDEX flowdata_06_prot ON flowdata_06 ( prot );

CREATE INDEX flowdata_07_unix_secs ON flowdata_07 ( unix_secs );
CREATE INDEX flowdata_07_srcaddr ON flowdata_07 ( srcaddr );
CREATE INDEX flowdata_07_dstaddr ON flowdata_07 ( dstaddr );
CREATE INDEX flowdata_07_srcport ON flowdata_07 ( srcport );
CREATE INDEX flowdata_07_dstport ON flowdata_07 ( dstport );
CREATE INDEX flowdata_07_prot ON flowdata_07 ( prot );

CREATE INDEX flowdata_08_unix_secs ON flowdata_08 ( unix_secs );
CREATE INDEX flowdata_08_srcaddr ON flowdata_08 ( srcaddr );
CREATE INDEX flowdata_08_dstaddr ON flowdata_08 ( dstaddr );
CREATE INDEX flowdata_08_srcport ON flowdata_08 ( srcport );
CREATE INDEX flowdata_08_dstport ON flowdata_08 ( dstport );
CREATE INDEX flowdata_08_prot ON flowdata_08 ( prot );

CREATE INDEX flowdata_09_unix_secs ON flowdata_09 ( unix_secs );
CREATE INDEX flowdata_09_srcaddr ON flowdata_09 ( srcaddr );
CREATE INDEX flowdata_09_dstaddr ON flowdata_09 ( dstaddr );
CREATE INDEX flowdata_09_srcport ON flowdata_09 ( srcport );
CREATE INDEX flowdata_09_dstport ON flowdata_09 ( dstport );
CREATE INDEX flowdata_09_prot ON flowdata_09 ( prot );

CREATE INDEX flowdata_10_unix_secs ON flowdata_10 ( unix_secs );
CREATE INDEX flowdata_10_srcaddr ON flowdata_10 ( srcaddr );
CREATE INDEX flowdata_10_dstaddr ON flowdata_10 ( dstaddr );
CREATE INDEX flowdata_10_srcport ON flowdata_10 ( srcport );
CREATE INDEX flowdata_10_dstport ON flowdata_10 ( dstport );
CREATE INDEX flowdata_10_prot ON flowdata_10 ( prot );

CREATE INDEX flowdata_11_unix_secs ON flowdata_11 ( unix_secs );
CREATE INDEX flowdata_11_srcaddr ON flowdata_11 ( srcaddr );
CREATE INDEX flowdata_11_dstaddr ON flowdata_11 ( dstaddr );
CREATE INDEX flowdata_11_srcport ON flowdata_11 ( srcport );
CREATE INDEX flowdata_11_dstport ON flowdata_11 ( dstport );
CREATE INDEX flowdata_11_prot ON flowdata_11 ( prot );

CREATE INDEX flowdata_12_unix_secs ON flowdata_12 ( unix_secs );
CREATE INDEX flowdata_12_srcaddr ON flowdata_12 ( srcaddr );
CREATE INDEX flowdata_12_dstaddr ON flowdata_12 ( dstaddr );
CREATE INDEX flowdata_12_srcport ON flowdata_12 ( srcport );
CREATE INDEX flowdata_12_dstport ON flowdata_12 ( dstport );
CREATE INDEX flowdata_12_prot ON flowdata_12 ( prot );

CREATE OR REPLACE RULE flowdata_insert_01 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 1 )
DO INSTEAD
    INSERT INTO flowdata_01 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_02 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 2 )
DO INSTEAD
    INSERT INTO flowdata_02 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_03 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 3 )
DO INSTEAD
    INSERT INTO flowdata_03 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_04 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 4 )
DO INSTEAD
    INSERT INTO flowdata_04 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_05 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 5 )
DO INSTEAD
    INSERT INTO flowdata_05 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_06 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 6 )
DO INSTEAD
    INSERT INTO flowdata_06 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_07 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 7 )
DO INSTEAD
    INSERT INTO flowdata_07 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_08 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 8 )
DO INSTEAD
    INSERT INTO flowdata_08 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_09 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 9 )
DO INSTEAD
    INSERT INTO flowdata_09 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_10 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 10 )
DO INSTEAD
    INSERT INTO flowdata_10 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_11 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 11 )
DO INSTEAD
    INSERT INTO flowdata_11 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

CREATE OR REPLACE RULE flowdata_insert_12 AS
ON INSERT TO flowdata WHERE
    ( EXTRACT(MONTH FROM to_timestamp(unix_secs) ) = 12 )
DO INSTEAD
    INSERT INTO flowdata_12 VALUES ( NEW.unix_secs,
                                     NEW.doctets,
                                     NEW.srcaddr, 
                                     NEW.dstaddr,
                                     NEW.srcport,
                                     NEW.dstport,
                                                                         NEW.prot );

After this, there will be a dedicated table for every month. Data can be insterted into flowdata table as before (as well as selected), though in fact it will reside in its corresponding table. Such structure is good for keeping logs for the last year. All we need to do is to flush current month table every first day of the month. Thus we'll always have data for the last 12 monthes.

PostgreSQL table partitioning and flow-export

RSS Feed