Londiste 3 replicate case - 1 下节

postgres@db5-> vi add_column.sql
begin;
alter table digoal_01.user_info add column c1 int;
alter table digoal_01.user_session add column c1 int;
end;

postgres@db5-> londiste3 /home/postgres/londiste3/src_digoal_01.ini execute ./add_column.sql 
2012-05-31 10:11:09,230 482 INFO Executing: add_column.sql
2012-05-31 10:11:09,253 482 INFO Execute finished: add_column.sql

postgres@db5-> londiste3 /home/postgres/londiste3/dst2_digoal_02.ini execute ./add_column.sql 
2012-05-31 10:14:52,014 584 ERROR Node is not root node: replika

2012-05-31 10:11:09,230 482 INFO Executing: add_column.sql
2012-05-31 10:11:09,253 482 INFO Execute finished: add_column.sql

dst1_digoal_01.log
2012-05-31 10:11:11,203 31872 INFO Executing: add_column.sql
2012-05-31 10:11:11,224 31872 INFO Execute finished: add_column.sql

pg92@db-172-16-3-33-> psql digoal_01 digoal_01
psql (9.2beta1)
Type "help" for help.

digoal_01=> \d user_info 
             Table "digoal_01.user_info"
   Column   |            Type             | Modifiers 
------------+-----------------------------+-----------
 userid     | integer                     | not null
 engname    | text                        | 
 cnname     | text                        | 
 occupation | text                        | 
 birthday   | date                        | 
 signname   | text                        | 
 email      | text                        | 
 qq         | numeric                     | 
 crt_time   | timestamp without time zone | 
 mod_time   | timestamp without time zone | 
 c1         | integer                     | 
Indexes:
    "pk_user_info" PRIMARY KEY, btree (userid)
Triggers:
    _londiste_replika AFTER INSERT OR DELETE OR UPDATE ON user_info FOR EACH ROW EXECUTE PROCEDURE pgq.logutriga('replika', 'deny')
    _londiste_replika_truncate AFTER TRUNCATE ON user_info FOR EACH STATEMENT EXECUTE PROCEDURE pgq.sqltriga('replika', 'deny')

digoal_01=> \d user_session 
                         Table "digoal_01.user_session"
     Column      |              Type              |          Modifiers           
-----------------+--------------------------------+------------------------------
 userid          | integer                        | not null
 logintime       | timestamp(0) without time zone | 
 login_count     | bigint                         | default 0
 logouttime      | timestamp(0) without time zone | 
 online_interval | interval                       | default '00:00:00'::interval
 c1              | integer                        | 
Indexes:
    "pk_user_session" PRIMARY KEY, btree (userid)
Triggers:
    _londiste_replika AFTER INSERT OR DELETE OR UPDATE ON user_session FOR EACH ROW EXECUTE PROCEDURE pgq.logutriga('replika', 'deny')
    _londiste_replika_truncate AFTER TRUNCATE ON user_session FOR EACH STATEMENT EXECUTE PROCEDURE pgq.sqltriga('replika', 'deny')

2012-05-31 10:16:50,921 32303 INFO Executing: add_column.sql
2012-05-31 10:16:50,928 32303 ERROR Job dst2_digoal_02 got error on connection 'db': schema "digoal_01" does not exist.   Query: alt
er table digoal_01.user_info add column c1 int;
Traceback (most recent call last):
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/skytools/scripting.py", line 565, in run_func_safely
    return func()
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/pgq/cascade/consumer.py", line 199, in work
    return Consumer.work(self)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/pgq/consumer.py", line 257, in work
    self._launch_process_batch(db, batch_id, ev_list)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/pgq/consumer.py", line 286, in _launch_process_batch
    self.process_batch(db, batch_id, list)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/pgq/cascade/consumer.py", line 172, in process_batch
    self.process_remote_batch(src_db, tick_id, event_list, dst_db)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/londiste/playback.py", line 368, in process_remote_batch
    CascadedWorker.process_remote_batch(self, src_db, tick_id, ev_list, dst_db)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/pgq/cascade/worker.py", line 155, in process_remote_batch
    self.process_remote_event(src_curs, dst_curs, ev)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/londiste/playback.py", line 597, in process_remote_event
    self.handle_execute_event(ev, dst_curs)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/londiste/playback.py", line 669, in handle_execute_event
    dst_curs.execute(stmt)
  File "/opt/python2.7.3/lib/python2.7/site-packages/psycopg2/extras.py", line 123, in execute
    return _cursor.execute(self, query, vars)
ProgrammingError: schema "digoal_01" does not exist

pg92@db-172-16-3-33-> psql digoal_02 digoal_02
psql (9.2beta1)
Type "help" for help.

digoal_02=> begin;
BEGIN
digoal_02=> alter table digoal_02.user_info1 add column c1 int;
ALTER TABLE
digoal_02=> alter table digoal_02.user_session1 add column c1 int;
ALTER TABLE
digoal_02=> end;
COMMIT

postgres@db5-> londiste3 /home/postgres/londiste3/dst2_digoal_02.ini status
Queue: replika   Local node: dst2_digoal_02

src_digoal_01 (root)
  |                           Tables: 4/0/0
  |                           Lag: 18s, Tick: 2238
  +--dst1_digoal_01 (leaf)
  |                           Tables: 4/0/0
  |                           Lag: 18s, Tick: 2238
  +--dst2_digoal_02 (leaf)
                              Tables: 2/0/2
                              Lag: 2h10m42s, Tick: 2021
                              ERR: dst2_digoal_02: schema "digoal_01" does not exist

2012-05-31 12:28:44,500 32303 INFO Executing: add_column.sql
2012-05-31 12:28:44,507 32303 ERROR Job dst2_digoal_02 got error on connection 'db': schema "digoal_01" does not exist.   Query: alt
er table digoal_01.user_info add column c1 int;
Traceback (most recent call last):
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/skytools/scripting.py", line 565, in run_func_safely
    return func()
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/pgq/cascade/consumer.py", line 199, in work
    return Consumer.work(self)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/pgq/consumer.py", line 257, in work
    self._launch_process_batch(db, batch_id, ev_list)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/pgq/consumer.py", line 286, in _launch_process_batch
    self.process_batch(db, batch_id, list)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/pgq/cascade/consumer.py", line 172, in process_batch
    self.process_remote_batch(src_db, tick_id, event_list, dst_db)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/londiste/playback.py", line 368, in process_remote_batch
    CascadedWorker.process_remote_batch(self, src_db, tick_id, ev_list, dst_db)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/pgq/cascade/worker.py", line 155, in process_remote_batch
    self.process_remote_event(src_curs, dst_curs, ev)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/londiste/playback.py", line 597, in process_remote_event
    self.handle_execute_event(ev, dst_curs)
  File "/opt/skytools3.0.2/lib/python2.7/site-packages/londiste/playback.py", line 669, in handle_execute_event
    dst_curs.execute(stmt)
  File "/opt/python2.7.3/lib/python2.7/site-packages/psycopg2/extras.py", line 123, in execute
    return _cursor.execute(self, query, vars)
ProgrammingError: schema "digoal_01" does not exist

pg92@db-172-16-3-33-> psql digoal_02 digoal_02
psql (9.2beta1)
Type "help" for help.
digoal_02=> create schema digoal_01;
CREATE SCHEMA
digoal_02=> create table digoal_01.user_info(id int);
CREATE TABLE
digoal_02=> create table digoal_01.user_session(id int);
CREATE TABLE

postgres@db5-> londiste3 /home/postgres/londiste3/dst2_digoal_02.ini status
Queue: replika   Local node: dst2_digoal_02

src_digoal_01 (root)
  |                           Tables: 4/0/0
  |                           Lag: 19s, Tick: 2266
  +--dst1_digoal_01 (leaf)
  |                           Tables: 4/0/0
  |                           Lag: 19s, Tick: 2266
  +--dst2_digoal_02 (leaf)
                              Tables: 2/0/2
                              Lag: 19s, Tick: 2266

digoal_02=> select * from digoal_01.user_info ;
 id | c1 
----+----
(0 rows)
digoal_02=> select * from digoal_01.user_session ;
 id | c1 
----+----
(0 rows)
digoal_02=> drop schema digoal_01 cascade;
NOTICE:  drop cascades to 2 other objects
DETAIL:  drop cascades to table digoal_01.user_info
drop cascades to table digoal_01.user_session
DROP SCHEMA

postgres@db5-> londiste3 --ini
[londiste3]

## Parameters for Londiste ##

# target database
db = dbname=somedb host=127.0.0.1

# how many tables can be copied in parallel
#parallel_copies = 1

# accept only events for locally present tables
#local_only = true

## compare/repair
# max amount of time table can be locked
#lock_timeout = 10
# compare: sql to use
#compare_sql = select count(1) as cnt, sum(hashtext(t.*::text)) as chksum from only _TABLE_ t
#compare_fmt = %(cnt)d rows, checksum=%(chksum)s


## Parameters for pgq.CascadedWorker ##

# how often the root node should push wm downstream (seconds)
#global_wm_publish_period = 300

# how often the nodes should report their wm upstream (seconds)
#local_wm_publish_period = 300


## Parameters for pgq.Consumer ##

# queue name to read from
queue_name =

# override consumer name
#consumer_name = %(job_name)s

# whether to use cursor to fetch events (0 disables)
#pgq_lazy_fetch = 300

# whether to read from source size in autocommmit mode
# not compatible with pgq_lazy_fetch
# the actual user script on top of pgq.Consumer must also support it
#pgq_autocommit = 0

# whether to wait for specified number of events, before
# assigning a batch (0 disables)
#pgq_batch_collect_events = 0

# whether to wait specified amount of time,
# before assigning a batch (postgres interval)
#pgq_batch_collect_interval =

# whether to stay behind queue top (postgres interval)
#pgq_keep_lag =


## Parameters for skytools.DBScript ##

# default lifetime for database connections (in seconds)
#connection_lifetime = 1200


## Parameters for skytools.DBScript ##

# how many seconds to sleep between work loops
# if missing or 0, then instead sleeping, the script will exit
loop_delay = 1.0

# where to log
logfile = ~/log/%(job_name)s.log

# where to write pidfile
pidfile = ~/pid/%(job_name)s.pid

# per-process name to use in logging
#job_name = %(config_name)s

# whether centralized logging should be used
# search-path [ ./skylog.ini, ~/.skylog.ini, /etc/skylog.ini ]
#   0 - disabled
#   1 - enabled, unless non-daemon on console (os.isatty())
#   2 - always enabled
#use_skylog = 0

# how many seconds to sleep after catching a exception
#exception_sleep = 20

postgres@db5-> pgqd --ini
[pgqd]

# where to log
logfile = ~/log/pgqd.log

# pidfile
pidfile = ~/pid/pgqd.pid

## optional parameters ##

# libpq connect string without dbname=
#base_connstr =

# startup db to query other databases
#initial_database = template1

# limit ticker to specific databases
#database_list =

# log into syslog
#syslog = 1
#syslog_ident = pgqd

## optional timeouts ##

# how often to check for new databases
#check_period = 60

# how often to flush retry queue
#retry_period = 30

# how often to do maintentance
#maint_period = 120

# how often to run ticker
#ticker_period = 1

pg92@db-172-16-3-33-> pwd
/opt/skytools3.0.2/lib/python2.7/site-packages/londiste
pg92@db-172-16-3-33-> ll
total 220K
-rw-r--r-- 1 root root 2.8K May 11 02:40 bublin.py
-rw-r--r-- 1 root root 3.5K May 30 14:47 bublin.pyc
-rw-r--r-- 1 root root 1.6K May 11 02:40 compare.py
-rw-r--r-- 1 root root 2.0K May 30 14:47 compare.pyc
-rw-r--r-- 1 root root 6.6K May 11 02:40 handler.py
-rw-r--r-- 1 root root 8.6K May 30 14:47 handler.pyc
drwxr-xr-x 2 root root 4.0K May 30 14:47 handlers
-rw-r--r-- 1 root root  601 May 11 02:40 __init__.py
-rw-r--r-- 1 root root  708 May 30 14:47 __init__.pyc
-rw-r--r-- 1 root root  32K May 11 02:40 playback.py
-rw-r--r-- 1 root root  28K May 30 14:47 playback.pyc
-rw-r--r-- 1 root root 9.8K May 11 02:40 repair.py
-rw-r--r-- 1 root root  11K May 30 14:47 repair.pyc
-rw-r--r-- 1 root root  21K May 11 02:40 setup.py
-rw-r--r-- 1 root root  20K May 30 14:47 setup.pyc
-rw-r--r-- 1 root root 8.5K May 11 02:40 syncer.py
-rw-r--r-- 1 root root 8.4K May 30 14:47 syncer.pyc
-rw-r--r-- 1 root root 8.6K May 11 02:40 table_copy.py
-rw-r--r-- 1 root root 7.1K May 30 14:47 table_copy.pyc

-- Adding a column to a replicated table
This case is handled in a simple process:
add the column on all the subscribers
BEGIN; -- on the provider
add the column on the provider
SELECT londiste.provider_refresh_trigger('queue_name', 'tablename');
COMMIT;
-- Removing a column to a replicated table
drop column from provider and change trigger in same transaction
look at the lag when londiste has passed the moment of the drop
drop column on subscribers
The trick here is to drop the column on the subscribers only when there's no more event in the queue referencing it.
-- Adding custom triggers on subscriber side
By default, londiste will consider that the triggers which exist on the subscriber tables are there because you just restore the provider schema there, being as lazy as possible. If you intend to run custom triggers on the subscriber, you have to tell londiste about them as follows:
create the custom trigger on the subscriber
londiste.py p-to-s.ini subscriber add public.T1
londiste.py p-to-s.ini subscriber restore-triggers public.T1
CAUTION: When londiste stops, it will remove the trigger again.
You can even give the restore trigger a specific trigger name, and only this one will get reinstalled.