patroni 部分源码阅读

问题1

/usr/local/lib/python3.9/site-packages/patroni/postgresql/init.py

964     @contextmanager
965     def get_replication_connection_cursor(self, host=None, port=5432, **kwargs):
966         conn_kwargs = self.config.replication.copy()
967         conn_kwargs.update(host=host, port=int(port) if port else None, user=conn_kwargs.pop('username'),
968                            connect_timeout=3, replication=1, options='-c statement_timeout=2000')
969         with get_connection_cursor(**conn_kwargs) as cur:
970             yield cur
971
972     def get_replica_timeline(self):
973         try:
974             with self.get_replication_connection_cursor(**self.config.local_replication_address) as cur:
975                 cur.execute('IDENTIFY_SYSTEM')
976                 return cur.fetchone()[1]
977         except Exception:
978             logger.exception('Can not fetch local timeline and lsn from replication connection')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

/usr/local/lib/python3.9/site-packages/patroni/postgresql/connection.py

      1 import logging
      2
      3 from contextlib import contextmanager
      4 from threading import Lock
      5
      6 from .. import psycopg
      7
      8 logger = logging.getLogger(__name__)
      9
     10
     11 class Connection(object):
     12
     13     def __init__(self):
     14         self._lock = Lock()
     15         self._connection = None
     16         self._cursor_holder = None
     17
     18     def set_conn_kwargs(self, conn_kwargs):
     19         self._conn_kwargs = conn_kwargs
     20
     21     def get(self):
     22         with self._lock:
     23             if not self._connection or self._connection.closed != 0:
     24                 self._connection = psycopg.connect(**self._conn_kwargs)
     25                 self.server_version = self._connection.server_version
     26         return self._connection
     27
     28     def cursor(self):
     29         if not self._cursor_holder or self._cursor_holder.closed or self._cursor_holder.connection.closed != 0:
     30             logger.info("establishing a new patroni connection to the postgres cluster")
     31             self._cursor_holder = self.get().cursor()
     32         return self._cursor_holder
     33
     34     def close(self):
     35         if self._connection and self._connection.closed == 0:
     36             self._connection.close()
     37             logger.info("closed patroni connection to the postgresql cluster")
     38         self._cursor_holder = self._connection = None
     39
     40
     41 @contextmanager
     42 def get_connection_cursor(**kwargs):
     43     conn = psycopg.connect(**kwargs)
     44     with conn.cursor() as cur:
     45         yield cur
     46     conn.close()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

/usr/local/lib/python3.9/site-packages/patroni/psycopg.py

      1 __all__ = ['connect', 'quote_ident', 'quote_literal', 'DatabaseError', 'Error', 'OperationalError', 'ProgrammingError']
      2
      3 _legacy = False
      4 try:
      5     from psycopg2 import __version__
      6     from . import MIN_PSYCOPG2, parse_version
      7     if parse_version(__version__) < MIN_PSYCOPG2:
      8         raise ImportError
      9     from psycopg2 import connect as _connect, Error, DatabaseError, OperationalError, ProgrammingError
     10     from psycopg2.extensions import adapt
     11
     12     try:
     13         from psycopg2.extensions import quote_ident as _quote_ident
     14     except ImportError:
     15         _legacy = True
     16
     17     def quote_literal(value, conn=None):
     18         value = adapt(value)
     19         if conn:
     20             value.prepare(conn)
     21         return value.getquoted().decode('utf-8')
     22 except ImportError:
     23     from psycopg import connect as __connect, sql, Error, DatabaseError, OperationalError, ProgrammingError
     24
     25     def _connect(*args, **kwargs):
     26         ret = __connect(*args, **kwargs)
     27         ret.server_version = ret.pgconn.server_version  # compatibility with psycopg2
     28         return ret
     29
     30     def _quote_ident(value, conn):
     31         return sql.Identifier(value).as_string(conn)
     32
     33     def quote_literal(value, conn=None):
     34         return sql.Literal(value).as_string(conn)
     35
     36
     37 def connect(*args, **kwargs):
     38     if kwargs and 'replication' not in kwargs and kwargs.get('fallback_application_name') != 'Patroni ctl':
     39         options = [kwargs['options']] if 'options' in kwargs else []
     40         options.append('-c search_path=pg_catalog')
     41         kwargs['options'] = ' '.join(options)
     42     ret = _connect(*args, **kwargs)
     43     ret.autocommit = True
     44     return ret
     45
     46
     47 def quote_ident(value, conn=None):
     48     if _legacy or conn is None:
     49         return '"{0}"'.format(value.replace('"', '""'))
     50     return _quote_ident(value, conn)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

/usr/lib64/python3.9/site-packages/psycopg2/init.py

     80 def connect(dsn=None, connection_factory=None, cursor_factory=None, **kwargs):
     81     """
     82     Create a new database connection.
     83
     84     The connection parameters can be specified as a string:
     85
     86         conn = psycopg2.connect("dbname=test user=postgres password=secret")
     87
     88     or using a set of keyword arguments:
     89
     90         conn = psycopg2.connect(database="test", user="postgres", password="secret")
     91
     92     Or as a mix of both. The basic connection parameters are:
     93
     94     - *dbname*: the database name
     95     - *database*: the database name (only as keyword argument)
     96     - *user*: user name used to authenticate
     97     - *password*: password used to authenticate
     98     - *host*: database host address (defaults to UNIX socket if not provided)
     99     - *port*: connection port number (defaults to 5432 if not provided)
    100
    101     Using the *connection_factory* parameter a different class or connections
    102     factory can be specified. It should be a callable object taking a dsn
    103     argument.
    104
    105     Using the *cursor_factory* parameter, a new default cursor factory will be
    106     used by cursor().
    107
    108     Using *async*=True an asynchronous connection will be created. *async_* is
    109     a valid alias (for Python versions where ``async`` is a keyword).
    110
    111     Any other keyword parameter will be passed to the underlying client
    112     library: the list of supported parameters depends on the library version.
    113
    114     """
    115     kwasync = {}
    116     if 'async' in kwargs:
    117         kwasync['async'] = kwargs.pop('async')
    118     if 'async_' in kwargs:
    119         kwasync['async_'] = kwargs.pop('async_')
    120
    121     dsn = _ext.make_dsn(dsn, **kwargs)
    122     conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
    123     if cursor_factory is not None:
    124         conn.cursor_factory = cursor_factory
    125
    126     return conn

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

问题2

 0 [BACKEND] LOG:  shared memory 30603 Mbytes, memory context 133220 Mbytes, max process memory 163840 Mbytes
[2024-04-23 18:21:54.541][214][][gs_ctl]: gs_ctl status,datadir is /pgdata/data/opengauss-55634e8f 
gs_ctl: server is running (PID: 203)
/usr/local/opengauss/bin/gaussdb "-D" "/pgdata/data/opengauss-55634e8f" "--config-file=/pgdata/data/opengauss-55634e8f/postgresql.conf" "-M" "standby"
2024-04-23 18:21:54,551 INFO: is_update=true
2024-04-23 18:21:54,551 INFO: is_running
2024-04-23 18:21:54,551 INFO: cluster_member = ({'opengauss-55634e8f-0-0': '245.0.2.54'})
2024-04-23 18:21:54,552 INFO: configuration = ({'listen_addresses': '0.0.0.0', 'port': '5432', 'wal_level': 'hot_standby', 'hot_standby': 'on', 'max_connections': 8800, 'max_wal_senders': 10, 'max_prepared_transactions': 0, 'max_locks_per_transaction': 64, 'archive_command': 'sh /home/postgres/bin/opengauss_archive_push.sh %p %f opengauss-55634e8f qfusion-admin opengauss', 'archive_mode': True, 'datestyle': 'iso, mdy', 'enable_cbm_tracking': True, 'enable_page_lsn_check': True, 'log_destination': 'csvlog', 'log_directory': '/pglog', 'log_filename': 'postgresql-%a.log', 'log_min_duration_statement': -1, 'log_timezone': 'PRC', 'log_truncate_on_rotation': 'on', 'logging_collector': True, 'pgaudit.log': 'none', 'synchronous_commit': True, 'synchronous_standby_names': '', 'sysadmin_reserved_connections': 20, 'timezone': 'PRC', 'unix_socket_directory': '/tmp', 'application_name': 'opengauss-55634e8f-0-0', 'wal_keep_segments': 8})
[2024-04-23 18:21:54.564][218][][gs_ctl]: gs_ctl reload ,datadir is /pgdata/data/opengauss-55634e8f 
server signaled
2024-04-23 18:21:54,565 INFO: self._async_executor.busy =(False)
2024-04-23 18:21:54,566 INFO: establishing a new patroni connection to the postgres cluster
2024-04-23 18:21:54,641 ERROR: get_postgresql_status
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/patroni/api.py", line 743, in query
    with self.patroni.postgresql.connection().cursor() as cursor:
  File "/usr/local/lib/python3.9/site-packages/patroni/postgresql/__init__.py", line 321, in connection
    return self._connection.get()
  File "/usr/local/lib/python3.9/site-packages/patroni/postgresql/connection.py", line 24, in get
    self._connection = psycopg.connect(**self._conn_kwargs)
  File "/usr/local/lib/python3.9/site-packages/patroni/psycopg.py", line 42, in connect
    ret = _connect(*args, **kwargs)
  File "/usr/lib64/python3.9/site-packages/psycopg2/__init__.py", line 122, in connect
    conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
psycopg2.OperationalError: could not receive data from server, error: Connection reset by peer, remote datanode: (null)


During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/patroni/api.py", line 672, in get_postgresql_status
    row = self.query(stmt.format(postgresql.wal_name, postgresql.lsn_name), retry=retry)[0]
  File "/usr/local/lib/python3.9/site-packages/patroni/api.py", line 655, in query
    return self.server.query(sql, *params)
  File "/usr/local/lib/python3.9/site-packages/patroni/api.py", line 749, in query
    raise PostgresConnectionException('connection problems')
patroni.exceptions.PostgresConnectionException: 'connection problems'
2024-04-23 18:21:55,414 INFO: establishing a new patroni connection to the postgres cluster
2024-04-23 18:21:55,414 WARNING: Retry got exception: 'connection problems'
[2024-04-23 18:21:55.425][223][][gs_ctl]: gs_ctl status,datadir is /pgdata/data/opengauss-55634e8f 
no server running
2024-04-23 18:21:55,426 WARNING: Failed to determine PostgreSQL state from the connection, falling back to cached role
2024-04-23 18:21:55,427 WARNING: Failed to determine PostgreSQL state from the connection, falling back to cached role
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

问题3

2024-05-20 09:49:35,269 INFO: Local timeline=2 lsn=0/7000000
2024-05-20 09:49:35,271 ERROR: Exception when working with leader
Traceback (most recent call last):
  File "/usr/local/bin/patroni/patroni/postgresql/rewind.py", line 79, in check_leader_is_not_in_recovery
    with get_connection_cursor(connect_timeout=3, options='-c statement_timeout=2000', **conn_kwargs) as cur:
  File "/usr/lib64/python3.9/contextlib.py", line 119, in __enter__
    return next(self.gen)
  File "/usr/local/bin/patroni/patroni/postgresql/connection.py", line 48, in get_connection_cursor
    conn = psycopg.connect(**kwargs)
  File "/usr/local/bin/patroni/patroni/psycopg.py", line 103, in connect
    ret = _connect(*args, **kwargs)
  File "/usr/lib64/python3.9/site-packages/psycopg2/__init__.py", line 127, in connect
    conn = _connect(dsn, connection_factory=connection_factory, **kwasync)
psycopg2.OperationalError: FATAL:  the database system is shutting down
1
2
3
4
5
6
7
8
9
10
11
12
13
14

/pglog/xxx.csv

could not receive data from WAL stream: ERROR:  requested WAL segment 000000030000000000000007 has already been removed 
1

pip3 install cdiff
1