<commit_before>import copy
import datetime
import errno
import logging
import logging.handlers
import os
import psycopg2
import re
import requests
import select
import signal
import socket
import subprocess
import sys
import time
from email.utils import parsedate
from psycopg2.extras import RealDictCursor
from threading import Thread

try:
    from SocketServer import ThreadingMixIn # pylint: disable=F0401
    from BaseHTTPServer import HTTPServer # pylint: disable=F0401
    from SimpleHTTPServer import SimpleHTTPRequestHandler # pylint: disable=F0401
    import simplejson as json # pylint: disable=F0401
except ImportError: # Support Py3k
    import json
    from socketserver import ThreadingMixIn # pylint: disable=F0401
    from http.server import HTTPServer, SimpleHTTPRequestHandler # pylint: disable=F0401

format_str = "%(asctime)s\t%(name)s\t%(levelname)s\t%(message)s"
syslog_format_str = '%(name)s %(levelname)s: %(message)s'

logging.basicConfig(level=logging.DEBUG, format=format_str)

class TimeoutError(Exception):
    pass

def get_iso_timestamp(fetch_time=None):
    if not fetch_time:
        fetch_time = datetime.datetime.utcnow()
    elif fetch_time.tzinfo:
        fetch_time = fetch_time.replace(tzinfo=None) - datetime.timedelta(seconds=fetch_time.utcoffset().seconds)
    return fetch_time.isoformat() + "Z"

def parse_iso_datetime(value):
    pattern_ext = r'(?P<year>\d{4})-(?P<month>\d\d)-(?P<day>\d\d)(T(?P<hour>\d\d):(?P<minute>\d\d)(:(?P<second>\d\d)(.(?P<microsecond>\d{6}))?)?Z)?$' # pylint: disable=C0301
    pattern_basic = r'(?P<year>\d{4})(?P<month>\d\d)(?P<day>\d\d)(T(?P<hour>\d\d)(?P<minute>\d\d)((?P<second>\d\d)((?P<microsecond>\d{6}))?)?Z)?$' # pylint: disable=C0301
    match = re.match(pattern_ext, value)
    if not match:
        match = re.match(pattern_basic, value)
    parts = dict((key, int(match.group(key) or '0'))
                 for key in ('year', 'month', 'day', 'hour', 'minute', 'second', 'microsecond'))
    return datetime.datetime(tzinfo=None, **parts) # pylint: disable=W0142

def convert_xlog_location_to_offset(xlog_location):
    log_id, offset = xlog_location.split("/")
    return int('ffffffff', 16) * int(log_id, 16) * int(offset, 16)

def set_syslog_handler(syslog_address, syslog_facility, logger):
    syslog_handler = logging.handlers.SysLogHandler(address=syslog_address, facility=syslog_facility)
    logger.addHandler(syslog_handler)
    formatter = logging.Formatter(syslog_format_str)
    syslog_handler.setFormatter(formatter)
    return syslog_handler

class PgLookout(object):
    def __init__(self, config_path):
        self.log = logging.getLogger("pglookout")
        self.running = True
        self.replication_lag_over_warning_limit = False

        self.config_path = config_path
        self.config = {}
        self.log_level = "DEBUG"

        self.connected_master_nodes = {}
        self.replication_lag_warning_boundary = None
        self.replication_lag_failover_timeout = None
        self.own_db = None
        self.current_master = None
        self.failover_command = None
        self.over_warning_limit_command = None
        self.never_promote_these_nodes = None
        self.cluster_monitor = None
        self.syslog_handler = None
        self.load_config()

        signal.signal(signal.SIGHUP, self.load_config)
        signal.signal(signal.SIGINT, self.quit)
        signal.signal(signal.SIGTERM, self.quit)

        self.cluster_state = {}
        self.observer_state = {}
        self.overall_state = {"db_nodes": self.cluster_state, "observer_nodes": self.observer_state,
                              "current_master": self.current_master,
                              "replication_lag_over_warning": self.replication_lag_over_warning_limit}

        self.cluster_monitor = ClusterMonitor(self.config, self.cluster_state,
                                              self.observer_state)
        # cluster_monitor doesn't exist at the time of reading the config initially
        self.cluster_monitor.log.setLevel(self.log_level)
        self.webserver = WebServer(self.config, self.cluster_state)

        self.log.info("PGLookout initialized, own_hostname: %r, own_db: %r, cwd: %r",
                      socket.gethostname(), self.own_db, os.getcwd())

    def quit(self, _signal=None, _frame=None):
        self.log.warning("Quitting, signal: %r, frame: %r", _signal, _frame)
        self.cluster_monitor.running = False
        self.running = False
        self.webserver.close()

    def load_config(self, _signal=None, _frame=None):
        self.log.debug("Loading JSON config from: %r, signal: %r, frame: %r",
                       self.config_path, _signal, _frame)
        try:
            self.config = json.load(open(self.config_path, "r"))
        except:
            self.log.exception("Invalid JSON config, exiting")
            sys.exit(0)

        if self.config.get("syslog") and not self.syslog_handler:
            self.syslog_handler = set_syslog_handler(self.config.get("syslog_address", "/dev/log"),
                                                     self.config.get("syslog_facility", "local2"),
                                                     self.log)
        self.own_db = self.config.get("own_db")
        # the levelNames hack is needed for Python2.6
        self.log_level = logging._levelNames[self.config.get("log_level", "DEBUG")] # pylint: disable=W0212
        try:
            self.log.setLevel(self.log_level)
            if self.cluster_monitor:
                self.cluster_monitor.log.setLevel(self.log_level)
        except ValueError:
            print("Problem setting log level %r" % self.log_level)
            self.log.exception("Problem with log_level: %r", self.log_level)
        self.never_promote_these_nodes = self.config.get("never_promote_these_nodes", [])
        # we need the failover_command to be converted into subprocess [] format
        self.failover_command = self.config.get("failover_command", "").split(" ")
        self.over_warning_limit_command = self.config.get("over_warning_limit_command")
        self.replication_lag_warning_boundary = self.config.get("warning_replication_time_lag", 30.0)
        self.replication_lag_failover_timeout = self.config.get("max_failover_replication_time_lag", 120.0)
        self.log.debug("Loaded config: %r from: %r", self.config, self.config_path)

    def write_cluster_state_to_json_file(self):
        """Periodically write a JSON state file to disk"""
        start_time = time.time()
        state_file_path = self.config.get("json_state_file_path", "/tmp/json_state_file")
        try:
            self.overall_state = {"db_nodes": self.cluster_state, "observer_nodes": self.observer_state,
                                  "current_master": self.current_master}
            json_to_dump = json.dumps(self.overall_state, indent=4)
            self.log.debug("Writing JSON state file to: %r, file_size: %r", state_file_path, len(json_to_dump))
            with open(state_file_path + ".tmp", "w") as fp:
                fp.write(json_to_dump)
            os.rename(state_file_path + ".tmp", state_file_path)
            self.log.debug("Wrote JSON state file to disk, took %.4fs", time.time() - start_time)
        except:
            self.log.exception("Problem in writing JSON: %r file to disk, took %.4fs",
                               self.overall_state, time.time() - start_time)

    def create_node_map(self, cluster_state, observer_state):
        standby_nodes, master_node, master_host = {}, None, None
        connected_master_nodes, disconnected_master_nodes = {}, {}
        self.log.debug("Creating node map out of cluster_state: %r and observer_state: %r",
                       cluster_state, observer_state)
        for host, state in cluster_state.items():
            if 'pg_is_in_recovery' in state:
                if state['pg_is_in_recovery']:
                    standby_nodes[host] = state
                elif state['connection']:
                    connected_master_nodes[host] = state
                elif not state['connection']:
                    disconnected_master_nodes[host] = state
            else:
                self.log.debug("No knowledge on host: %r state: %r of whether it's in recovery or not", host, state)

        for observer_name, state in observer_state.items():
            for host, db_state in state.items():
                if isinstance(db_state, dict):
                    own_fetch_time = parse_iso_datetime(cluster_state.get(host, {"fetch_time": datetime.datetime(year=2000, month=1, day=1)})['fetch_time']) # pylint: disable=C0301
                    observer_fetch_time = parse_iso_datetime(db_state['fetch_time'])
                    self.log.debug("observer_name: %r, dbname: %r, state: %r, observer_fetch_time: %r",
                                   observer_name, host, db_state, observer_fetch_time)
                    if 'pg_is_in_recovery' in db_state:
                        if db_state['pg_is_in_recovery']:
                            # we always trust ourselves the most for localhost, and
                            # in case we are actually connected to the other node
                            if observer_fetch_time >= own_fetch_time and host != self.own_db and standby_nodes[host]['connection'] == False: #pylint: disable=C0301
                                standby_nodes[host] = db_state
                        else:
                            master_node = connected_master_nodes.get(host, {})
                            connected = master_node.get("connection", False)
                            self.log.debug("Observer: %r sees %r as master, we see: %r, same_master: %r, connection: %r",
                                           observer_name, host, self.current_master, host == self.current_master,
                                           db_state.get('connection'))
                            if observer_fetch_time >= own_fetch_time and host != self.own_db:
                                if connected:
                                    connected_master_nodes[host] = db_state
                                else:
                                    disconnected_master_nodes[host] = db_state
                    else:
                        self.log.warning("No knowledge on if: %r %r from observer: %r is in recovery",
                                         host, db_state, observer_name)

        self.connected_master_nodes = connected_master_nodes
        if len(self.connected_master_nodes) == 0:
            self.log.warning("No known master node, disconnected masters: %r", list(disconnected_master_nodes.keys()))
            if len(disconnected_master_nodes) > 0:
                master_host, master_node = disconnected_master_nodes.keys()[0], disconnected_master_nodes.values()[0]
        elif len(self.connected_master_nodes) == 1:
            master_host, master_node = connected_master_nodes.keys()[0], connected_master_nodes.values()[0]
            if disconnected_master_nodes:
                self.log.warning("Picked %r as master since %r are in a disconnected state",
                                 master_host, disconnected_master_nodes)
        else:
            self.create_alert_file("multiple_master_warning")
            self.log.error("More than one master node connected_master_nodes: %r, disconnected_master_nodes: %r",
                           connected_master_nodes, disconnected_master_nodes)

        return master_host, master_node, standby_nodes

    def check_cluster_state(self):
        master_node = None
        cluster_state = copy.deepcopy(self.cluster_state)
        observer_state = copy.deepcopy(self.observer_state)
        if not cluster_state:
            self.log.warning("No cluster state, probably still starting up")
            return

        master_host, master_node, standby_nodes = self.create_node_map(cluster_state, observer_state) # pylint: disable=W0612

        if master_host != self.current_master:
            self.log.info("New master node detected: old: %r new: %r: %r", self.current_master, master_host, master_node)
            self.current_master = master_host

        own_state = self.cluster_state.get(self.own_db)

        self.log.debug("Cluster has %s standbys, %s observers and %s as master, own_db: %r, own_state: %r",
                       ','.join(standby_nodes.keys()) or 'no',
                       ','.join(observer_state.keys()) or 'no',
                       self.current_master,
                       self.own_db,
                       own_state)

        if self.own_db:
            if self.own_db == self.current_master:
                # We are the master of this cluster, nothing to do<commit_msg>pglookout: On observer nodes, log a bit nicer debug state message<commit_after>import copy
import datetime
import errno
import logging
import logging.handlers
import os
import psycopg2
import re
import requests
import select
import signal
import socket
import subprocess
import sys
import time
from email.utils import parsedate
from psycopg2.extras import RealDictCursor
from threading import Thread

try:
    from SocketServer import ThreadingMixIn # pylint: disable=F0401
    from BaseHTTPServer import HTTPServer # pylint: disable=F0401
    from SimpleHTTPServer import SimpleHTTPRequestHandler # pylint: disable=F0401
    import simplejson as json # pylint: disable=F0401
except ImportError: # Support Py3k
    import json
    from socketserver import ThreadingMixIn # pylint: disable=F0401
    from http.server import HTTPServer, SimpleHTTPRequestHandler # pylint: disable=F0401

format_str = "%(asctime)s\t%(name)s\t%(levelname)s\t%(message)s"
syslog_format_str = '%(name)s %(levelname)s: %(message)s'

logging.basicConfig(level=logging.DEBUG, format=format_str)

class TimeoutError(Exception):
    pass

def get_iso_timestamp(fetch_time=None):
    if not fetch_time:
        fetch_time = datetime.datetime.utcnow()
    elif fetch_time.tzinfo:
        fetch_time = fetch_time.replace(tzinfo=None) - datetime.timedelta(seconds=fetch_time.utcoffset().seconds)
    return fetch_time.isoformat() + "Z"

def parse_iso_datetime(value):
    pattern_ext = r'(?P<year>\d{4})-(?P<month>\d\d)-(?P<day>\d\d)(T(?P<hour>\d\d):(?P<minute>\d\d)(:(?P<second>\d\d)(.(?P<microsecond>\d{6}))?)?Z)?$' # pylint: disable=C0301
    pattern_basic = r'(?P<year>\d{4})(?P<month>\d\d)(?P<day>\d\d)(T(?P<hour>\d\d)(?P<minute>\d\d)((?P<second>\d\d)((?P<microsecond>\d{6}))?)?Z)?$' # pylint: disable=C0301
    match = re.match(pattern_ext, value)
    if not match:
        match = re.match(pattern_basic, value)
    parts = dict((key, int(match.group(key) or '0'))
                 for key in ('year', 'month', 'day', 'hour', 'minute', 'second', 'microsecond'))
    return datetime.datetime(tzinfo=None, **parts) # pylint: disable=W0142

def convert_xlog_location_to_offset(xlog_location):
    log_id, offset = xlog_location.split("/")
    return int('ffffffff', 16) * int(log_id, 16) * int(offset, 16)

def set_syslog_handler(syslog_address, syslog_facility, logger):
    syslog_handler = logging.handlers.SysLogHandler(address=syslog_address, facility=syslog_facility)
    logger.addHandler(syslog_handler)
    formatter = logging.Formatter(syslog_format_str)
    syslog_handler.setFormatter(formatter)
    return syslog_handler

class PgLookout(object):
    def __init__(self, config_path):
        self.log = logging.getLogger("pglookout")
        self.running = True
        self.replication_lag_over_warning_limit = False

        self.config_path = config_path
        self.config = {}
        self.log_level = "DEBUG"

        self.connected_master_nodes = {}
        self.replication_lag_warning_boundary = None
        self.replication_lag_failover_timeout = None
        self.own_db = None
        self.current_master = None
        self.failover_command = None
        self.over_warning_limit_command = None
        self.never_promote_these_nodes = None
        self.cluster_monitor = None
        self.syslog_handler = None
        self.load_config()

        signal.signal(signal.SIGHUP, self.load_config)
        signal.signal(signal.SIGINT, self.quit)
        signal.signal(signal.SIGTERM, self.quit)

        self.cluster_state = {}
        self.observer_state = {}
        self.overall_state = {"db_nodes": self.cluster_state, "observer_nodes": self.observer_state,
                              "current_master": self.current_master,
                              "replication_lag_over_warning": self.replication_lag_over_warning_limit}

        self.cluster_monitor = ClusterMonitor(self.config, self.cluster_state,
                                              self.observer_state)
        # cluster_monitor doesn't exist at the time of reading the config initially
        self.cluster_monitor.log.setLevel(self.log_level)
        self.webserver = WebServer(self.config, self.cluster_state)

        self.log.info("PGLookout initialized, own_hostname: %r, own_db: %r, cwd: %r",
                      socket.gethostname(), self.own_db, os.getcwd())

    def quit(self, _signal=None, _frame=None):
        self.log.warning("Quitting, signal: %r, frame: %r", _signal, _frame)
        self.cluster_monitor.running = False
        self.running = False
        self.webserver.close()

    def load_config(self, _signal=None, _frame=None):
        self.log.debug("Loading JSON config from: %r, signal: %r, frame: %r",
                       self.config_path, _signal, _frame)
        try:
            self.config = json.load(open(self.config_path, "r"))
        except:
            self.log.exception("Invalid JSON config, exiting")
            sys.exit(0)

        if self.config.get("syslog") and not self.syslog_handler:
            self.syslog_handler = set_syslog_handler(self.config.get("syslog_address", "/dev/log"),
                                                     self.config.get("syslog_facility", "local2"),
                                                     self.log)
        self.own_db = self.config.get("own_db")
        # the levelNames hack is needed for Python2.6
        self.log_level = logging._levelNames[self.config.get("log_level", "DEBUG")] # pylint: disable=W0212
        try:
            self.log.setLevel(self.log_level)
            if self.cluster_monitor:
                self.cluster_monitor.log.setLevel(self.log_level)
        except ValueError:
            print("Problem setting log level %r" % self.log_level)
            self.log.exception("Problem with log_level: %r", self.log_level)
        self.never_promote_these_nodes = self.config.get("never_promote_these_nodes", [])
        # we need the failover_command to be converted into subprocess [] format
        self.failover_command = self.config.get("failover_command", "").split(" ")
        self.over_warning_limit_command = self.config.get("over_warning_limit_command")
        self.replication_lag_warning_boundary = self.config.get("warning_replication_time_lag", 30.0)
        self.replication_lag_failover_timeout = self.config.get("max_failover_replication_time_lag", 120.0)
        self.log.debug("Loaded config: %r from: %r", self.config, self.config_path)

    def write_cluster_state_to_json_file(self):
        """Periodically write a JSON state file to disk"""
        start_time = time.time()
        state_file_path = self.config.get("json_state_file_path", "/tmp/json_state_file")
        try:
            self.overall_state = {"db_nodes": self.cluster_state, "observer_nodes": self.observer_state,
                                  "current_master": self.current_master}
            json_to_dump = json.dumps(self.overall_state, indent=4)
            self.log.debug("Writing JSON state file to: %r, file_size: %r", state_file_path, len(json_to_dump))
            with open(state_file_path + ".tmp", "w") as fp:
                fp.write(json_to_dump)
            os.rename(state_file_path + ".tmp", state_file_path)
            self.log.debug("Wrote JSON state file to disk, took %.4fs", time.time() - start_time)
        except:
            self.log.exception("Problem in writing JSON: %r file to disk, took %.4fs",
                               self.overall_state, time.time() - start_time)

    def create_node_map(self, cluster_state, observer_state):
        standby_nodes, master_node, master_host = {}, None, None
        connected_master_nodes, disconnected_master_nodes = {}, {}
        self.log.debug("Creating node map out of cluster_state: %r and observer_state: %r",
                       cluster_state, observer_state)
        for host, state in cluster_state.items():
            if 'pg_is_in_recovery' in state:
                if state['pg_is_in_recovery']:
                    standby_nodes[host] = state
                elif state['connection']:
                    connected_master_nodes[host] = state
                elif not state['connection']:
                    disconnected_master_nodes[host] = state
            else:
                self.log.debug("No knowledge on host: %r state: %r of whether it's in recovery or not", host, state)

        for observer_name, state in observer_state.items():
            for host, db_state in state.items():
                if isinstance(db_state, dict):
                    own_fetch_time = parse_iso_datetime(cluster_state.get(host, {"fetch_time": datetime.datetime(year=2000, month=1, day=1)})['fetch_time']) # pylint: disable=C0301
                    observer_fetch_time = parse_iso_datetime(db_state['fetch_time'])
                    self.log.debug("observer_name: %r, dbname: %r, state: %r, observer_fetch_time: %r",
                                   observer_name, host, db_state, observer_fetch_time)
                    if 'pg_is_in_recovery' in db_state:
                        if db_state['pg_is_in_recovery']:
                            # we always trust ourselves the most for localhost, and
                            # in case we are actually connected to the other node
                            if observer_fetch_time >= own_fetch_time and host != self.own_db and standby_nodes[host]['connection'] == False: #pylint: disable=C0301
                                standby_nodes[host] = db_state
                        else:
                            master_node = connected_master_nodes.get(host, {})
                            connected = master_node.get("connection", False)
                            self.log.debug("Observer: %r sees %r as master, we see: %r, same_master: %r, connection: %r",
                                           observer_name, host, self.current_master, host == self.current_master,
                                           db_state.get('connection'))
                            if observer_fetch_time >= own_fetch_time and host != self.own_db:
                                if connected:
                                    connected_master_nodes[host] = db_state
                                else:
                                    disconnected_master_nodes[host] = db_state
                    else:
                        self.log.warning("No knowledge on if: %r %r from observer: %r is in recovery",
                                         host, db_state, observer_name)

        self.connected_master_nodes = connected_master_nodes
        if len(self.connected_master_nodes) == 0:
            self.log.warning("No known master node, disconnected masters: %r", list(disconnected_master_nodes.keys()))
            if len(disconnected_master_nodes) > 0:
                master_host, master_node = disconnected_master_nodes.keys()[0], disconnected_master_nodes.values()[0]
        elif len(self.connected_master_nodes) == 1:
            master_host, master_node = connected_master_nodes.keys()[0], connected_master_nodes.values()[0]
            if disconnected_master_nodes:
                self.log.warning("Picked %r as master since %r are in a disconnected state",
                                 master_host, disconnected_master_nodes)
        else:
            self.create_alert_file("multiple_master_warning")
            self.log.error("More than one master node connected_master_nodes: %r, disconnected_master_nodes: %r",
                           connected_master_nodes, disconnected_master_nodes)

        return master_host, master_node, standby_nodes

    def check_cluster_state(self):
        master_node = None
        cluster_state = copy.deepcopy(self.cluster_state)
        observer_state = copy.deepcopy(self.observer_state)
        if not cluster_state:
            self.log.warning("No cluster state, probably still starting up")
            return

        master_host, master_node, standby_nodes = self.create_node_map(cluster_state, observer_state) # pylint: disable=W0612

        if master_host != self.current_master:
            self.log.info("New master node detected: old: %r new: %r: %r", self.current_master, master_host, master_node)
            self.current_master = master_host

        own_state = self.cluster_state.get(self.own_db)

        # If we're an observer ourselves, we'll grab the IP address from HTTP server address
        observer_info = ','.join(observer_state.keys()) or 'no'
        if not self.own_db:
            observer_info = self.config.get("http_address", observer_info)

        self.log.debug("Cluster has %s standbys, %s observers and %s as master, own_db: %r, own_state: %r",
                       ','.join(standby_nodes.keys()) or 'no',
                       observer_info,
                       self.current_master,
                       self.own_db,
                       own_state or "observer")

        if self.own_db:
            if self.own_db == self.current_master:
                # We are the master of this cluster, nothing to do
{'fraction_non_alphanumeric': 0.1030796048808832, 'fraction_numerical': 0.007205113306217315, 'mean_word_length': 2.4421333333333335, 'pattern_counts': {'":': 16, '<': 31, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 36, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5734550', 'n_tokens_mistral': 7646, 'n_tokens_neox': 7329, 'n_words': 1583}