標簽

PostgreSQL , protocol , proxy , 通信協議

---

背景

理解PostgreSQL的通信協議可以更好的開發類似SQL代理，SQL中間件，SQL防火墻，連接池等軟件。

學習資料與軟件

《PostgreSQL 讀寫分離代理 - Crunchy Proxy(base on golang)》

Postgres on the wire - A look at the PostgreSQL wire protocol

PG通信協議介紹

https://www.postgresql.org/docs/current/static/protocol.html

https://www.postgresql.org/docs/current/static/protocol-message-formats.html

This section describes the detailed format of each message. Each is marked to indicate that it can be sent by a frontend (F), a backend (B), or both (F & B). Notice that although each message includes a byte count at the beginning, the message format is defined so that the message end can be found without reference to the byte count. This aids validity checking. (The CopyData message is an exception, because it forms part of a data stream; the contents of any individual CopyData message cannot be interpretable on their own.)

通信包格式樣例

Query (F)  
Byte1('Q')  
Identifies the message as a simple query.  Int32  
Length of message contents in bytes, including self.  String  
The query string itself.  CopyData (F & B)  
Byte1('d')  
Identifies the message as COPY data.  Int32  
Length of message contents in bytes, including self.  Byten  
Data that forms part of a COPY data stream.   
Messages sent from the backend will always correspond to single data rows,   
but messages sent by frontends might divide the data stream arbitrarily.  CopyDone (F & B)  
Byte1('c')  
Identifies the message as a COPY-complete indicator.  Int32(4)  
Length of message contents in bytes, including self.  

協議處理代碼樣例

src/interfaces/libpq/fe-protocol3.c

                                case 'C':               /* command complete */  if (pqGets(&conn->workBuffer, conn))  return;  if (conn->result == NULL)  {  conn->result = PQmakeEmptyPGresult(conn,  PGRES_COMMAND_OK);  if (!conn->result)  {  printfPQExpBuffer(&conn->errorMessage,  libpq_gettext("out of memory"));  pqSaveErrorResult(conn);  }  }  if (conn->result)  strlcpy(conn->result->cmdStatus, conn->workBuffer.data,  CMDSTATUS_LEN);  conn->asyncStatus = PGASYNC_READY;  break;  

某 PostgreSQL 代理軟件

https://github.com/CrunchyData/crunchy-proxy

https://github.com/wgliang/pgproxy

某 PostgreSQL 協議測試軟件

https://github.com/tatsuo-ishii/pgproto

http://dalibo.github.io/pgshark/

pgproto的使用簡介

yum install -y automake    
git clone https://github.com/tatsuo-ishii/pgproto    cd pgproto    
touch configure.ac Makefile Makefile.ac    
autoreconf -ivf    export PGPORT=1921    
export PGDATA=/data01/pg/pg_root$PGPORT      
export PGHOME=/usr/pgsql-10     
export LD_LIBRARY_PATH=$PGHOME/lib:/lib64:/usr/lib64:/usr/local/lib64:/lib:/usr/lib:/usr/local/lib:$LD_LIBRARY_PATH      
export PATH=$PGHOME/bin:$PATH:.      
export MANPATH=$PGHOME/share/man:$MANPATH      
export PGHOST=$PGDATA      
export PGUSER=postgres      
export PGDATABASE=postgres      ./configure --prefix=/var/lib/pgsql/pgp    
make    
make install    

psql    
create table aaa(id int);    
insert into aaa select generate_series(1,100);    

vi test.sql    #    
# Test data example    
#    
'Q'	"SELECT * FROM aaa"    
'Y'    
'P'	"S1"	"BEGIN"	0    
'B'	""	"S1"	0	0	0    
'E'	""	0    
'C'	'S'	"S1"    
'P'	"foo"	"SELECT 1"	0    
'B'	"myportal"	"foo"	0	0	0    
'E'	"myportal"	0    
'P'	"S2"	"COMMIT"	0    
'B'	""	"S2"	0	0	0    
'E'	""	0    
'C'	'S'	"S2"    
'S'    
'Y'    
'X'    

./pgp/bin/pgproto -h $PGDATA -p 1921 -u postgres -d postgres  -f ./test.sql    
FE=> Query(query="SELECT * FROM aaa")    
<= BE RowDescription    
<= BE DataRow    
<= BE DataRow    
<= BE DataRow    
<= BE DataRow    
<= BE DataRow    
<= BE DataRow    
<= BE DataRow    
<= BE DataRow    
<= BE DataRow    
<= BE DataRow    
<= BE DataRow    
.............. 100行    
<= BE CommandComplete(SELECT 1)    
<= BE ReadyForQuery(I)    
FE=> Parse(stmt="S1", query="BEGIN")    
FE=> Bind(stmt="S1", portal="")    
FE=> Execute(portal="")    
FE=> Close(stmt="S1")    
FE=> Parse(stmt="foo", query="SELECT 1")    
FE=> Bind(stmt="foo", portal="myportal")    
FE=> Execute(portal="myportal")    
FE=> Parse(stmt="S2", query="COMMIT")    
FE=> Bind(stmt="S2", portal="")    
FE=> Execute(portal="")    
FE=> Close(stmt="S2")    
FE=> Sync    
<= BE ParseComplete    
<= BE BindComplete    
<= BE CommandComplete(BEGIN)    
<= BE CloseComplete    
<= BE ParseComplete    
<= BE BindComplete    
<= BE DataRow    
<= BE CommandComplete(SELECT 1)    
<= BE ParseComplete    
<= BE BindComplete    
<= BE CommandComplete(COMMIT)    
<= BE CloseComplete    
<= BE ReadyForQuery(I)    
FE=> Terminate