ALICloudDB for PostgreSQL 试用报告 - 4 水平分库 之 节点扩展

RDS现在还欠缺一个功能，就是数据库克隆，你可以这么理解，给现有的数据库创建STANDBY，然后将这个STANDBY激活，就完成了对数据库的克隆。
为什么我们需要数据库克隆功能呢？
这会使得数据库的扩容变得非常简单，比如我们这里的应用场景，如果要将16个RDS，变成32个RDS，那么克隆无疑是最好的办法。因为不需要做逻辑数据迁移的事情，只需要删除不需要的数据库，以及调整plproxy的cluster配置即可。
我们先假设RDS有创建STANDBYD的功能（相信未来会增加），看看如何来实现RDS的扩容。

假设主RDS1包含db0,db16两个库，现在要拆分为两个RDS，RDS1(db0), RDS2(db16)，分别包含db0和db16。
1. 为需要拆分的主RDS创建standby RDS, 确认流复制正常，确认standby RDS到达catchup状态。

2. 配置密码文件~/.pgpass，包含克隆库的认证信息。

3. 调整plproxy cluster配置。使用只读事务连接需要迁移的数据库，避免写操作带来的主备不一致。
例如 : 
alter server rds_pg_cluster options (set p16 'host=old_rds_hostname dbname=db16 port=3433 user=digoal keepalives_idle=30 keepalives_interval=10 keepalives_count=10 application_name=16 options=''-c default_transaction_read_only=true'' ');

4. 确认主RDS需迁移的库(db16)没有连接，确认standby处于catchup状态。

5. 激活standby。

6. 调整plproxy cluster配置。原连接RDS1(db16)，修改为RDS2(db16)。
例如 : 
alter server rds_pg_cluster options (set p16 'host=new_rds_hostname dbname=db16 port=3433 user=digoal keepalives_idle=30 keepalives_interval=10 keepalives_count=10 application_name=16');

7. 删除主RDS节点已迁移的数据库(db16)，删除standby节点多余的数据库(db0)。

循环1-7，将所有RDS拆分。

回到现实，现实是目前RDS没有提供克隆功能。那么我们需要人工实现数据迁移，需迁移的东西还挺多，包括表，视图，函数，。。。。。可以用pg_dump，但是怎么实现增量呢？可以通过PostgreSQL的logical decoding来实现增量复制。现在阿里云RDS PG已经支持了逻辑复制的功能。

下面是人力扩容的例子：
源：
postgres=> select datname,pg_database_size(oid)/1024/1024 from pg_database;
  datname  | ?column? 
-----------+----------
 template1 |        6
 template0 |        6
 postgres  |     3618
 digoal    |        6
 db7       |      179
 db23      |      179
(6 rows)

目标，把db23迁移到以下RDS：
postgres=> select datname,pg_database_size(oid)/1024/1024 from pg_database;
  datname  | ?column? 
-----------+----------
 template1 |        6
 template0 |        6
 postgres  |        6
 digoal    |        6
 db8       |      179
 db24      |      179
(6 rows)

在目标RDS创建db23数据库：
postgres=> create database db23;
CREATE DATABASE
postgres=> \c db23
psql (9.4.3, server 9.4.1)
You are now connected to database "db23" as user "digoal".
创建schema和需要迁移的函数：
db23=> create schema digoal;
CREATE SCHEMA
db23=> CREATE OR REPLACE FUNCTION digoal.dy(sql text)
db23->      RETURNS SETOF record
db23->      LANGUAGE plpgsql
db23->      STRICT
db23->     AS $function$
db23$>       declare
db23$>         rec record;
db23$>       begin
db23$>         for rec in execute sql loop
db23$>           return next rec;
db23$>         end loop;
db23$>         return;
db23$>       end;
db23$>     $function$;
CREATE FUNCTION
db23=> CREATE OR REPLACE FUNCTION digoal.dy_ddl(sql text)
db23->      RETURNS VOID
db23->      LANGUAGE plpgsql
db23->      STRICT
db23->     AS $function$
db23$>       declare
db23$>       begin
db23$>         execute sql;
db23$>         return;
db23$>       exception when others then return;
db23$>       end;
db23$>     $function$;
CREATE FUNCTION
准备需要迁移的数据的外部表：
db23=> create extension postgres_fdw;
CREATE EXTENSION

CREATE SERVER db23
        FOREIGN DATA WRAPPER postgres_fdw
        OPTIONS (host 'old.pg.rds.aliyuncs.com', port '3433', dbname 'db23');

CREATE USER MAPPING FOR digoal  --  locale user
        SERVER db23
        OPTIONS (user 'digoal', password 'digoal');  --  remote user/password

CREATE FOREIGN TABLE digoal.ft_userinfo (
        dbid int default 23,
        userid int,
    info text
)
        SERVER db23
        OPTIONS (schema_name 'digoal', table_name 'userinfo');

CREATE FOREIGN TABLE digoal.ft_session (
        dbid int default 23,
        userid int,
    last_login timestamp without time zone
)
        SERVER db23
        OPTIONS (schema_name 'digoal', table_name 'session');

CREATE FOREIGN TABLE digoal.ft_tbl_small (
        userid int,
    info text
)
        SERVER db23
        OPTIONS (schema_name 'digoal', table_name 'tbl_small'); 

CREATE FOREIGN TABLE digoal.ft_login_log (
        dbid int default 23,
        userid int,
        db_user name,
        client_addr inet,
        client_port int,
        server_addr inet,
        server_port int,
        login_time timestamp without time zone
)
        SERVER db23
        OPTIONS (schema_name 'digoal', table_name 'login_log');

创建物化视图，日志表（日志数据不迁移）
CREATE MATERIALIZED VIEW digoal.userinfo (
        dbid ,
        userid ,
    info 
) as select * from digoal.ft_userinfo;
set maintenance_work_mem='10GB';  --  超出RDS内存限制，可能会被杀掉
create unique index pk_userinfo on digoal.userinfo (userid);

CREATE MATERIALIZED VIEW digoal.session (
        dbid ,
        userid ,
    last_login 
) as select * from digoal.ft_session;
set maintenance_work_mem='10GB';  --  超出RDS内存限制，可能会被杀掉
create unique index pk_session on digoal.session (userid);

CREATE MATERIALIZED VIEW digoal.tbl_small (
        userid ,
    info 
) as select * from digoal.ft_tbl_small;
set maintenance_work_mem='10GB';  --  超出RDS内存限制，可能会被杀掉
create unique index pk_tbl_small on digoal.tbl_small (userid);

CREATE TABLE digoal.login_log (
        dbid int default 23,
        userid int,
        db_user name,
        client_addr inet,
        client_port int,
        server_addr inet,
        server_port int,
        login_time timestamp without time zone
);

创建需要迁移的函数：
CREATE OR REPLACE FUNCTION digoal.dy_generate_test_ddl()
     RETURNS VOID
     LANGUAGE plpgsql
     STRICT
    AS $function$
      declare
        node int;
    sql text;
      begin
        select application_name::int into node from pg_stat_activity where pid=pg_backend_pid();
    sql := $a$insert into digoal.userinfo select $a$||node||$a$,generate_series($a$||node||$a$,32000000,32)$a$;
    execute sql;
    sql := $a$insert into digoal.session select dbid,userid from digoal.userinfo$a$;
    execute sql;
        return;
      exception when others then return;
      end;
    $function$;

CREATE OR REPLACE FUNCTION digoal.query_pk(IN i_userid int, OUT dbid int, OUT userid int, OUT info text)
     RETURNS record
     LANGUAGE plpgsql
     STRICT
    AS $function$
      declare
      begin
        select t.dbid,t.userid,t.info into dbid,userid,info from digoal.userinfo t where t.userid=i_userid;
        return;
      end;
    $function$;

CREATE OR REPLACE FUNCTION digoal.insert_log(IN i_userid int)
     RETURNS void
     LANGUAGE plpgsql
     STRICT
    AS $function$
      declare
      begin
        insert into digoal.login_log (userid,db_user,client_addr,client_port,server_addr,server_port,login_time)
        values (i_userid,current_user,inet_client_addr(),inet_client_port(),inet_server_addr(),inet_server_port(),now());
      end;
    $function$;

CREATE OR REPLACE FUNCTION digoal.query_insert(IN i_userid int, OUT dbid int, OUT userid int, OUT info text)
     RETURNS record
     LANGUAGE plpgsql
     STRICT
    AS $function$
      declare
      begin
        select t.dbid,t.userid,t.info into dbid,userid,info from digoal.userinfo t where t.userid=i_userid;
        insert into digoal.login_log (userid,db_user,client_addr,client_port,server_addr,server_port,login_time)
        values (i_userid,current_user,inet_client_addr(),inet_client_port(),inet_server_addr(),inet_server_port(),now());
        return;
      end;
    $function$;


CREATE OR REPLACE FUNCTION digoal.update_pk(IN i_userid int)
     RETURNS void
     LANGUAGE plpgsql
     STRICT
    AS $function$
      declare
      begin
        update digoal.session t set last_login=now() where t.userid=i_userid;
      end;
    $function$;

CREATE OR REPLACE FUNCTION digoal.query_update_insert(IN i_userid int, OUT dbid int, OUT userid int, OUT info text)
     RETURNS record
     LANGUAGE plpgsql
     STRICT
    AS $function$
      declare
      begin
        select t.dbid,t.userid,t.info into dbid,userid,info from digoal.userinfo t where t.userid=i_userid;
        insert into digoal.login_log (userid,db_user,client_addr,client_port,server_addr,server_port,login_time)
        values (i_userid,current_user,inet_client_addr(),inet_client_port(),inet_server_addr(),inet_server_port(),now());
        update digoal.session t set last_login=now() where t.userid=i_userid;
        return;
      end;
    $function$;

CREATE OR REPLACE FUNCTION digoal.query_smalltbl(IN i_userid int, OUT userid int, OUT info text)
     RETURNS record
     LANGUAGE plpgsql
     STRICT
    AS $function$
      declare
      begin
        select t.userid,t.info into userid,info from digoal.tbl_small t where t.userid=i_userid;
        return;
      end;
    $function$;


PL/Proxy节点操作如下：
配置.pgpass, 新增：
new.pg.rds.aliyuncs.com:3433:*:digoal:digoal
使用default_transaction_read_only默认读事务，屏蔽写操作.
在迁移时间段内，用户可以正常执行读请求，但是如果执行写请求会失败，这样确保数据迁移的一致性，同时降低迁移过程对业务的影响。
postgres# alter server rds_pg_cluster options (set p23 'host=old.pg.rds.aliyuncs.com dbname=db23 port=3433 user=digoal keepalives_idle=30 keepalives_interval=10 keepalives_count=10 application_name=23 options=''-c default_transaction_read_only=true''');    -- 注意里面是两个单引号
测试读正常，写失败：
postgres=# select query_pk(23);
 query_pk 
----------
 (23,23,)
(1 row)

postgres=# select insert_log(23);
ERROR:  public.insert_log(1): [db23] REMOTE ERROR: cannot execute INSERT in a read-only transaction
CONTEXT:  Remote context: SQL statement "insert into digoal.login_log (userid,db_user,client_addr,client_port,server_addr,server_port,login_time)
    values (i_userid,current_user,inet_client_addr(),inet_client_port(),inet_server_addr(),inet_server_port(),now())"
PL/pgSQL function insert_log(integer) line 4 at SQL statement

new RDS执行操作如下，刷新物化视图:
refresh materialized view CONCURRENTLY digoal.userinfo;
REFRESH MATERIALIZED VIEW
Time: 10953.220 ms
refresh materialized view CONCURRENTLY digoal.session;
REFRESH MATERIALIZED VIEW
Time: 11013.860 ms
refresh materialized view CONCURRENTLY digoal.tbl_small;
REFRESH MATERIALIZED VIEW
Time: 5084.118 ms

这里卡住，因为要修改数据字典需要超级用户。而RDS提供的用户是普通用户，无法修改数据字典（虽然有风险，这里只为演示）。所以这样迁移行不通。
如果是超级用户，那么操作请参考我前期写的BLOG
http://blog.163.com/digoal@126/blog/static/163877040201559105235803/

为了演示下去，我只能选择全量迁移。（其他增量方法也有，本文不演示）
db23=> drop materialized view digoal.session ;
DROP MATERIALIZED VIEW
Time: 16.528 ms
db23=> drop materialized view digoal.userinfo;
DROP MATERIALIZED VIEW
Time: 15.781 ms
db23=> drop materialized view digoal.tbl_small;
DROP MATERIALIZED VIEW
Time: 9.458 ms

为了提高迁移速度，用了一些手段。
set synchronous_commit=off;
set maintenance_work_mem='10GB';  --  超出RDS内存限制，可能会被杀掉
CREATE TABLE digoal.userinfo (
        dbid int default 23,
        userid int,
    info text
) with (autovacuum_enabled=off, toast.autovacuum_enabled=off);

CREATE TABLE digoal.session (
        dbid int default 23,
        userid int,
    last_login timestamp without time zone
) with (autovacuum_enabled=off, toast.autovacuum_enabled=off);

CREATE TABLE digoal.tbl_small (
        userid int,
    info text
) with (autovacuum_enabled=off, toast.autovacuum_enabled=off);

insert into digoal.userinfo select * from digoal.ft_userinfo;
INSERT 0 1000000
Time: 45290.701 ms
insert into digoal.session select * from digoal.ft_session;
INSERT 0 1000000
Time: 42212.278 ms
insert into digoal.tbl_small select * from digoal.ft_tbl_small;
INSERT 0 500000
Time: 22885.456 ms

alter table digoal.userinfo add constraint pk_userinfo primary key (userid);
ALTER TABLE
Time: 16962.174 ms
alter table digoal.session add constraint pk_session primary key (userid);
ALTER TABLE
Time: 20809.422 ms
alter table digoal.tbl_small add constraint pk_tbl_small primary key (userid);
ALTER TABLE
Time: 17484.201 ms

vacuum analyze digoal.userinfo;
Time: 65.790 ms
vacuum analyze digoal.session;
Time: 65.427 ms
vacuum analyze digoal.tbl_small;
Time: 45.453 ms

alter table digoal.userinfo set (autovacuum_enabled=on, toast.autovacuum_enabled=on);
alter table digoal.session set (autovacuum_enabled=on, toast.autovacuum_enabled=on);
alter table digoal.tbl_small set (autovacuum_enabled=on, toast.autovacuum_enabled=on);


PL/Proxy,  修改集群，db23的目标主机为新的RDS，并且开放读写权限：
postgres# alter server rds_pg_cluster options (set p23 'host=old.pg.rds.aliyuncs.com dbname=db23 port=3433 user=digoal keepalives_idle=30 keepalives_interval=10 keepalives_count=10 application_name=23');


删除老RDS上的db23.
psql -h old.pg.rds.aliyuncs.com -p 3433 -U digoal postgres
postgres=> drop database db23;
DROP DATABASE

测试plproxy分发是否正常分发到新的数据库：
postgres=# select * from query_pk(23);
 dbid | userid | info 
------+--------+------
   23 |     23 | 
(1 row)

vi test.sql
\setrandom id 1 32000000
select insert_log(:id);
pgbench -M prepared -n -r -f ./test.sql -P 1 -c 30 -j 30 -T 30
progress: 1.0 s, 7853.2 tps, lat 3.340 ms stddev 6.056
progress: 2.0 s, 10766.4 tps, lat 2.432 ms stddev 5.433
progress: 3.0 s, 11395.0 tps, lat 2.277 ms stddev 4.590
progress: 4.0 s, 11622.1 tps, lat 2.216 ms stddev 4.493
progress: 5.0 s, 10519.9 tps, lat 2.454 ms stddev 5.600
progress: 6.0 s, 11153.4 tps, lat 2.317 ms stddev 4.795
progress: 7.0 s, 11474.3 tps, lat 2.312 ms stddev 4.802
progress: 8.0 s, 11398.5 tps, lat 2.253 ms stddev 4.308
progress: 9.0 s, 12106.7 tps, lat 2.174 ms stddev 3.302
progress: 10.0 s, 12567.8 tps, lat 2.068 ms stddev 2.075

连接到新的db23：
psql -h new_rds -p 3433 -U digoal
postgres=> \c db23
psql (9.4.3, server 9.4.1)
You are now connected to database "db23" as user "digoal".
db23=> select count(*) from login_log ;
 count 
-------
 10547
(1 row)

[补充]
1. 如果实际使用内存超出了RDS内存限制，会被杀掉。
postgres=> set maintenance_work_mem='10GB';
SET
postgres=> alter table session add constraint pk_session primary key(userid);
server closed the connection unexpectedly
        This probably means the server terminated abnormally
        before or while processing the request.
The connection to the server was lost. Attempting reset: Failed.
!> \q

[参考]
1. http://blog.163.com/digoal@126/blog/static/163877040201559105235803/
2. http://www.postgresql.org/docs/9.4/static/libpq-connect.html