阿里云ApsaraDB RDS for PostgreSQL 最佳实践 - 4 水平分库(plproxy) 之 节点扩展
背景
RDS现在还欠缺一个功能，就是数据库克隆，你可以这么理解，给现有的数据库创建STANDBY，然后将这个STANDBY激活，就完成了对数据库的克隆。
(目前可以使用阿里云的开源软件RDS_DBSYNC来克隆，虽然复杂一点。 rds_dbsync支持mysql或pgsql克隆到pgsql)
https://github.com/aliyun/rds_dbsync
为什么我们需要数据库克隆功能呢？
这会使得数据库的扩容变得非常简单，比如我们这里的应用场景，如果要将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