--- /dev/null
+[queue_loader]
+
+job_name = loader_dst
+db = dbname=loaderdst
+queue_name = loaderq
+
+logfile = log/%(job_name)s.log
+pidfile = pid/%(job_name)s.pid
+
+rename_tables =
+
+[data.simple_tbl]
+table_mode = direct
+
+[data.bulk_tbl]
+table_mode = direct
+row_mode = bulk
+
+[data.keep_all_tbl]
+table_mode = split
+row_mode = bulk
+split_mode = by-date-field:tstamp
+
+### Non-inherited partitions
+split_part_template =
+ create table %%(part)s (like %%(parent)s);
+ alter table only %%(part)s add primary key (%%(pkey)s);
+
+[data.keep_latest_tbl]
+table_mode = split
+row_mode = bulk
+split_mode = by-date-field:tstamp
+
+### Inherited partitions
+split_part_template =
+ create table %%(part)s () inherits (%%(parent)s);
+ alter table only %%(part)s add primary key (%%(pkey)s);
+
+
+[DEFAULT]
+
+# fields - which fields to send through
+#fields = col1, col2, col3:renamed3
+#fields = *
+
+# table_mode - how to handle a table
+#
+# ignore - ignore this table
+# direct - update table directly
+# split - split data into partitions
+#table_mode = ignore
+
+# split_mode - how to split, if requested
+#
+# by-batch-time: use batch time for splitting
+# by-event-time: use event time for splitting
+# by-date-field:fld - use fld for splitting
+#split_mode = by-batch-time
+
+# split_part - partition name format
+#
+# %(table_name)s %(year)s %(month)s %(day)s %(hour)s
+#split_part = %(table_name)s_%(year)s_%(month)s_%(day)s
+
+# split_part_template - How to create new partition tables
+#
+# Available fields:
+# %(part)s
+# %(parent)s
+# %(pkey)s
+#
+### Non-inherited partitions
+#split_part_template =
+# create table %%(part)s (like %%(parent)s);
+# alter table only %%(part)s add primary key (%%(pkey)s);
+#
+### Inherited partitions
+#split_part_template =
+# create table %%(part)s () inherits (%%(parent)s);
+# alter table only %%(part)s add primary key (%%(pkey)s);
+
+
+# row_mode - How to apply the events
+#
+# plain - each event creates SQL statement to run
+# keep_latest - change updates to DELETE + INSERT
+# keep_all - change updates to inserts, ignore deletes
+# bulk - instead of statement-per-row, do bulk updates
+#row_mode = plain
+
+
+# bulk_mode - How to do the bulk update
+#
+# correct - inserts as COPY into table,
+# update as COPY into temp table and single UPDATE from there
+# delete as COPY into temp table and single DELETE from there
+# delete - as 'correct', but do update as DELETE + COPY
+# merged - as 'delete', but merge insert rows with update rows
+#bulk_mode=correct
+
--- /dev/null
+[queue_loader]
+
+job_name = loader_src
+db = dbname=loadersrc
+queue_name = loaderq
+
+logfile = log/%(job_name)s.log
+pidfile = pid/%(job_name)s.pid
--- /dev/null
+[cascade_admin]
+
+job_name = setadm_loaderq
+
+node_db = dbname=loadersrc
+
+queue_name = loaderq
+
+
+logfile = log/%(job_name)s.log
+pidfile = pid/%(job_name)s.pid
--- /dev/null
+[pgqadm]
+job_name = ticker_loadersrc
+db = dbname=loadersrc
+loop_delay = 0.5
+logfile = log/%(job_name)s.log
+pidfile = pid/%(job_name)s.pid
--- /dev/null
+#! /bin/sh
+
+. ../env.sh
+
+lst="loadersrc loaderdst"
+
+for db in $lst; do
+ echo dropdb $db
+ dropdb $db
+done
+for db in $lst; do
+ echo createdb $db
+ createdb $db
+done
--- /dev/null
+#! /bin/sh
+
+. ../env.sh
+
+mkdir -p log pid conf
+
+./zstop.sh
+
+v=
+v=-q
+v=-v
+
+(cd ../..; make -s python-install )
+
+echo ""
+
+cleardb() {
+ echo "Clearing database $1"
+ psql -q -d $1 -c '
+ set client_min_messages=warning;
+ drop schema if exists londiste cascade;
+ drop schema if exists pgq_node cascade;
+ drop schema if exists pgq cascade;
+ drop schema if exists data cascade;
+ '
+}
+
+run() {
+ echo "$ $*"
+ "$@"
+}
+
+db_list="loadersrc loaderdst"
+
+for db in $db_list; do
+ cleardb $db
+done
+
+echo "clean logs"
+rm -f log/*.log
+
+set -e
+
+run setadm $v conf/setadm_loaderq.ini create-root ldr-src 'dbname=loadersrc' --worker=loader_src
+run setadm $v conf/setadm_loaderq.ini create-leaf ldr-dst 'dbname=loaderdst' --worker=loader_dst --provider="dbname=loadersrc"
+
+run pgqadm $v conf/ticker_loadersrc.ini -d ticker
+
+run queue_loader $v -d conf/loader_src.ini
+run queue_loader $v -d conf/loader_dst.ini
+
+run psql -d loadersrc -f tables.sql
+run psql -d loadersrc -f triggers.sql
+
+run psql -d loaderdst -f tables.sql
+
+run psql -d loadersrc -f send.data.sql
+run psql -d loadersrc -f send.data.sql
+run psql -d loadersrc -f send.data.sql
+
+run sleep 2
+
+run setadm $v conf/setadm_loaderq.ini status
+
+./zcheck.sh
+
--- /dev/null
+
+insert into data.simple_tbl (username, contactname, data)
+values ('randuser'||random()::text, 'randcontact'||random()::text, 'link');
+
+/*
+insert into data.simple_tbl (username, contactname, data)
+values ('sameuser', 'samecontact', 'link');
+update data.simple_tbl
+*/
+
+insert into data.bulk_tbl (data) values ('newdata');
+
+
+insert into data.keep_all_tbl (username, data) values ('sameuser', 'newdata');
+
+insert into data.keep_latest_tbl (username, data) values ('sameuser', 'newdata');
+
+insert into data.random_tbl (username, data) values ('sameuser', 'newdata');
+
+
+
--- /dev/null
+
+set client_min_messages = 'warning';
+
+create schema data;
+
+create table data.simple_tbl (
+ username text not null,
+ contactname text not null,
+ data text,
+ primary key (username, contactname)
+);
+
+create table data.bulk_tbl (
+ id serial primary key,
+ data text
+);
+
+create table data.keep_all_tbl (
+ id serial primary key,
+ username text not null,
+ tstamp timestamptz not null default now(),
+ data text
+);
+
+create table data.keep_latest_tbl (
+ id serial primary key,
+ username text not null,
+ tstamp timestamptz not null default now(),
+ data text
+);
+
+create table data.random_tbl (
+ id serial primary key,
+ username text not null,
+ tstamp timestamptz not null default now(),
+ data text
+);
+
--- /dev/null
+
+create trigger logger after insert or update or delete on data.simple_tbl
+for each row execute procedure pgq.logutriga('loaderq');
+
+create trigger logger after insert or update or delete on data.bulk_tbl
+for each row execute procedure pgq.logutriga('loaderq');
+
+create trigger logger after insert or update or delete on data.keep_all_tbl
+for each row execute procedure pgq.logutriga('loaderq');
+
+create trigger logger after insert or update or delete on data.keep_latest_tbl
+for each row execute procedure pgq.logutriga('loaderq');
+
+create trigger logger after insert or update or delete on data.random_tbl
+for each row execute procedure pgq.logutriga('loaderq');
+
--- /dev/null
+#! /bin/sh
+
+grep -E 'ERR|WARN|CRIT' log/*.log || echo "All OK"
+
--- /dev/null
+#! /bin/sh
+
+#. ../env.sh
+
+for p in pid/*.pid*; do
+ test -f "$p" || continue
+ pid=`cat "$p"`
+ test -d "/proc/$pid" || {
+ rm -f "$p"
+ continue
+ }
+ kill "$pid"
+done
+
projects / skytools.git / commitdiff