mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Feike Steenbergen
a56b346295
For the postgresql helper, hardened the code to get "the" cursor of the postgresql instance. For the statuspage, a small status json is returned. To find out what status a PostgreSQL cluster is in we use the cursor (instead of the provided query() function), as the query function does some retrying etc. For the healthcheck we want to simple provide an answer to a simple query, if we have to reconnect, we are not *that* healthy anyway. Dropped catching exceptions in the do_GET block, as the HTTPServer will do that nicely for us anyway.
208 lines
7.8 KiB
Python
208 lines
7.8 KiB
Python
import os, psycopg2, re, time
|
|
import logging
|
|
|
|
from urlparse import urlparse
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Postgresql:
|
|
|
|
def __init__(self, config, aws_host_address=None):
|
|
self.name = config["name"]
|
|
self.host, self.port = config["listen"].split(":")
|
|
self.libpq_parameters = {
|
|
'host' : aws_host_address or self.host,
|
|
'port' : self.port,
|
|
'fallback_application_name' : 'Governor',
|
|
'connect_timeout' : 5,
|
|
'options' : '-c statement_timeout=2000'
|
|
}
|
|
self.data_dir = config["data_dir"]
|
|
self.replication = config["replication"]
|
|
self.superuser = config.get('superuser')
|
|
self.admin = config.get('admin')
|
|
|
|
self.config = config
|
|
|
|
self.cursor_holder = None
|
|
self.connection_string = "postgres://%s:%s@%s:%s/postgres" % (self.replication["username"], self.replication["password"], self.libpq_parameters['host'], self.port)
|
|
|
|
self.conn = None
|
|
|
|
def cursor(self):
|
|
if (self.cursor_holder is None) or self.cursor_holder.closed:
|
|
if (self.conn is None) or self.conn.closed:
|
|
self.conn = psycopg2.connect(**self.libpq_parameters)
|
|
self.conn.autocommit = True
|
|
self.cursor_holder = self.conn.cursor()
|
|
|
|
return self.cursor_holder
|
|
|
|
def disconnect(self):
|
|
try:
|
|
self.conn.close()
|
|
except Exception as e:
|
|
logger.error("Error disconnecting: %s" % e)
|
|
|
|
def query(self, sql):
|
|
max_attempts = 0
|
|
while True:
|
|
try:
|
|
self.cursor().execute(sql)
|
|
break
|
|
except psycopg2.OperationalError as e:
|
|
if self.conn:
|
|
self.disconnect()
|
|
self.cursor_holder = None
|
|
if max_attempts > 4:
|
|
raise e
|
|
max_attempts += 1
|
|
time.sleep(5)
|
|
return self.cursor()
|
|
|
|
def data_directory_empty(self):
|
|
return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == []
|
|
|
|
def initialize(self):
|
|
if os.system("initdb -D %s" % self.data_dir) == 0:
|
|
self.write_pg_hba()
|
|
|
|
return True
|
|
|
|
return False
|
|
|
|
def sync_from_leader(self, leader):
|
|
leader = urlparse(leader["address"])
|
|
|
|
f = open("./pgpass", "w")
|
|
f.write("%(hostname)s:%(port)s:*:%(username)s:%(password)s\n" %
|
|
{"hostname": leader.hostname, "port": leader.port, "username": leader.username, "password": leader.password})
|
|
f.close()
|
|
|
|
os.system("chmod 600 pgpass")
|
|
|
|
return os.system("PGPASSFILE=pgpass pg_basebackup -R -D %(data_dir)s --host=%(host)s --port=%(port)s -U %(username)s" %
|
|
{"data_dir": self.data_dir, "host": leader.hostname, "port": leader.port, "username": leader.username}) == 0
|
|
|
|
def is_leader(self):
|
|
return not self.query("SELECT pg_is_in_recovery();").fetchone()[0]
|
|
|
|
def is_running(self):
|
|
return os.system("pg_ctl status -D %s > /dev/null" % self.data_dir) == 0
|
|
|
|
def start(self):
|
|
if self.is_running():
|
|
logger.error("Cannot start PostgreSQL because one is already running.")
|
|
return False
|
|
|
|
pid_path = "%s/postmaster.pid" % self.data_dir
|
|
if os.path.exists(pid_path):
|
|
os.remove(pid_path)
|
|
logger.info("Removed %s" % pid_path)
|
|
|
|
command_code = os.system("postgres -D %s %s &" % (self.data_dir, self.server_options()))
|
|
time.sleep(5)
|
|
return command_code != 0
|
|
|
|
def stop(self):
|
|
return os.system("pg_ctl stop -w -D %s -m fast -w" % self.data_dir) != 0
|
|
|
|
def reload(self):
|
|
return os.system("pg_ctl reload -w -D %s" % self.data_dir) == 0
|
|
|
|
def restart(self):
|
|
return os.system("pg_ctl restart -w -D %s -m fast" % self.data_dir) == 0
|
|
|
|
def server_options(self):
|
|
options = "-c listen_addresses=%s -c port=%s" % (self.host, self.port)
|
|
for setting, value in self.config["parameters"].iteritems():
|
|
options += " -c \"%s=%s\"" % (setting, value)
|
|
return options
|
|
|
|
def is_healthy(self):
|
|
if not self.is_running():
|
|
logger.warning("Postgresql is not running.")
|
|
return False
|
|
|
|
return True
|
|
|
|
def is_healthiest_node(self, members):
|
|
for member in members:
|
|
if member["hostname"] == self.name:
|
|
continue
|
|
try:
|
|
member_conn = psycopg2.connect(member["address"])
|
|
member_conn.autocommit = True
|
|
member_cursor = member_conn.cursor()
|
|
member_cursor.execute("SELECT '%s'::pg_lsn - pg_last_xlog_replay_location() AS bytes;" % self.xlog_position())
|
|
xlog_diff = member_cursor.fetchone()[0]
|
|
logger.info([self.name, member["hostname"], xlog_diff])
|
|
if xlog_diff < 0:
|
|
member_cursor.close()
|
|
return False
|
|
member_cursor.close()
|
|
except psycopg2.OperationalError:
|
|
continue
|
|
return True
|
|
|
|
def replication_slot_name(self):
|
|
member = os.environ.get("MEMBER")
|
|
(member, _) = re.subn(r'[^a-z0-9]+', r'_', member)
|
|
return member
|
|
|
|
def write_pg_hba(self):
|
|
f = open("%s/pg_hba.conf" % self.data_dir, "a")
|
|
f.write("host replication %(username)s %(network)s md5" %
|
|
{"username": self.replication["username"], "network": self.replication["network"]})
|
|
# allow TCP connections from the host's own address
|
|
f.write("\nhost postgres postgres samehost trust\n")
|
|
# allow TCP connections from the rest of the world with a password
|
|
f.write("\nhost all all 0.0.0.0/0 md5\n")
|
|
f.close()
|
|
|
|
def write_recovery_conf(self, leader_hash):
|
|
leader = urlparse(leader_hash["address"])
|
|
|
|
f = open("%s/recovery.conf" % self.data_dir, "w")
|
|
f.write("""
|
|
standby_mode = 'on'
|
|
primary_slot_name = '%(recovery_slot)s'
|
|
primary_conninfo = 'user=%(user)s password=%(password)s host=%(hostname)s port=%(port)s sslmode=prefer sslcompression=1'
|
|
recovery_target_timeline = 'latest'
|
|
""" % {"recovery_slot": self.name, "user": leader.username, "password": leader.password, "hostname": leader.hostname, "port": leader.port})
|
|
if "recovery_conf" in self.config:
|
|
for name, value in self.config["recovery_conf"].iteritems():
|
|
f.write("%s = '%s'" % (name, value))
|
|
f.close()
|
|
|
|
def follow_the_leader(self, leader_hash):
|
|
leader = urlparse(leader_hash["address"])
|
|
if os.system("grep 'host=%(hostname)s port=%(port)s' %(data_dir)s/recovery.conf > /dev/null" % {"hostname": leader.hostname, "port": leader.port, "data_dir": self.data_dir}) != 0:
|
|
self.write_recovery_conf(leader_hash)
|
|
self.restart()
|
|
return True
|
|
|
|
def promote(self):
|
|
return os.system("pg_ctl promote -w -D %s" % self.data_dir) == 0
|
|
|
|
def demote(self, leader):
|
|
self.write_recovery_conf(leader)
|
|
self.restart()
|
|
|
|
def create_replication_user(self):
|
|
self.query("CREATE USER \"%s\" WITH REPLICATION ENCRYPTED PASSWORD '%s';" % (self.replication["username"], self.replication["password"]))
|
|
|
|
def create_connection_users(self):
|
|
if self.superuser:
|
|
if 'username' in self.superuser:
|
|
self.query("CREATE ROLE \"{0}\" LOGIN SUPERUSER PASSWORD '{1}';".format(self.superuser["username"], self.superuser["password"]))
|
|
else:
|
|
self.query("ALTER ROLE postgres PASSWORD '{0}';".format(self.superuser['password']))
|
|
if self.admin:
|
|
self.query("CREATE ROLE \"{0}\" LOGIN CREATEDB CREATEROLE PASSWORD '{1}';".format(self.admin["username"], self.admin["password"]))
|
|
|
|
def xlog_position(self):
|
|
return self.query("SELECT pg_last_xlog_replay_location();").fetchone()[0]
|