<commit_before>"""
pglookout - cluster monitoring component

Copyright (c) 2015 Ohmu Ltd
Copyright (c) 2014 F-Secure

This file is under the Apache License, Version 2.0.
See the file `LICENSE` for details.
"""

from . import logutil
from .common import get_iso_timestamp, parse_iso_datetime
from .pgutil import mask_connection_info
from concurrent.futures import as_completed, ThreadPoolExecutor
from email.utils import parsedate
from psycopg2.extras import RealDictCursor
from threading import Thread
import datetime
import errno
import logging
import psycopg2
import requests
import select
import time

try:
    from queue import Empty  # pylint: disable=import-error
except ImportError:
    from Queue import Empty  # pylint: disable=import-error


class PglookoutTimeout(Exception):
    pass


def wait_select(conn, timeout=5.0):
    end_time = time.time() + timeout
    while time.time() < end_time:
        time_left = end_time - time.time()
        state = conn.poll()
        try:
            if state == psycopg2.extensions.POLL_OK:
                return
            elif state == psycopg2.extensions.POLL_READ:
                select.select([conn.fileno()], [], [], min(timeout, time_left))
            elif state == psycopg2.extensions.POLL_WRITE:
                select.select([], [conn.fileno()], [], min(timeout, time_left))
            else:
                raise psycopg2.OperationalError("bad state from poll: %s" % state)
        except select.error as error:
            if error.args[0] != errno.EINTR:
                raise
    raise PglookoutTimeout("timed out in wait_select")


class ClusterMonitor(Thread):
    def __init__(self, config, cluster_state, observer_state, create_alert_file, trigger_check_queue,
                 stats):
        Thread.__init__(self)
        self.log = logging.getLogger("ClusterMonitor")
        self.stats = stats
        self.running = True
        self.cluster_state = cluster_state
        self.observer_state = observer_state
        self.config = config
        self.create_alert_file = create_alert_file
        self.db_conns = {}
        self.trigger_check_queue = trigger_check_queue
        self.session = requests.Session()
        if self.config.get("syslog"):
            self.syslog_handler = logutil.set_syslog_handler(
                address=self.config.get("syslog_address", "/dev/log"),
                facility=self.config.get("syslog_facility", "local2"),
                logger=self.log,
            )
        self.log.debug("Initialized ClusterMonitor with: %r", cluster_state)

    def _connect_to_db(self, instance, dsn):
        conn = self.db_conns.get(instance)
        if conn:
            return conn
        if not dsn:
            self.log.warning("Can't connect to %s, dsn is %r", instance, dsn)
            return None
        inst_info_str = "{0!r} ({1})".format(instance, mask_connection_info(dsn))
        try:
            self.log.info("Connecting to %s", inst_info_str)
            conn = psycopg2.connect(dsn=dsn, async=True)
            wait_select(conn)
            self.log.debug("Connected to %s", inst_info_str)
        except (PglookoutTimeout, psycopg2.OperationalError) as ex:
            self.log.warning("%s (%s) connecting to %s (%s)",
                             ex.__class__.__name__, str(ex).strip(),
                             instance, inst_info_str)
            if "password authentication" in getattr(ex, "message", ""):
                self.create_alert_file("authentication_error")
            conn = None  # make sure we don't try to use the connection if we timed out
        except Exception as ex:  # pylint: disable=broad-except
            self.log.exception("Failed to connect to %s (%s)",
                               instance, inst_info_str)
            self.stats.unexpected_exception(ex, where="_connect_to_db")
            conn = None
        self.db_conns[instance] = conn
        return conn

    def _fetch_observer_state(self, instance, uri):
        result = {"fetch_time": get_iso_timestamp(), "connection": True}<commit_msg>cluster_monitor: 'async' is a Python 3.7 keyword<commit_after>"""
pglookout - cluster monitoring component

Copyright (c) 2015 Ohmu Ltd
Copyright (c) 2014 F-Secure

This file is under the Apache License, Version 2.0.
See the file `LICENSE` for details.
"""

from . import logutil
from .common import get_iso_timestamp, parse_iso_datetime
from .pgutil import mask_connection_info
from concurrent.futures import as_completed, ThreadPoolExecutor
from email.utils import parsedate
from psycopg2.extras import RealDictCursor
from threading import Thread
import datetime
import errno
import logging
import psycopg2
import requests
import select
import time

try:
    from queue import Empty  # pylint: disable=import-error
except ImportError:
    from Queue import Empty  # pylint: disable=import-error


class PglookoutTimeout(Exception):
    pass


def wait_select(conn, timeout=5.0):
    end_time = time.time() + timeout
    while time.time() < end_time:
        time_left = end_time - time.time()
        state = conn.poll()
        try:
            if state == psycopg2.extensions.POLL_OK:
                return
            elif state == psycopg2.extensions.POLL_READ:
                select.select([conn.fileno()], [], [], min(timeout, time_left))
            elif state == psycopg2.extensions.POLL_WRITE:
                select.select([], [conn.fileno()], [], min(timeout, time_left))
            else:
                raise psycopg2.OperationalError("bad state from poll: %s" % state)
        except select.error as error:
            if error.args[0] != errno.EINTR:
                raise
    raise PglookoutTimeout("timed out in wait_select")


class ClusterMonitor(Thread):
    def __init__(self, config, cluster_state, observer_state, create_alert_file, trigger_check_queue,
                 stats):
        Thread.__init__(self)
        self.log = logging.getLogger("ClusterMonitor")
        self.stats = stats
        self.running = True
        self.cluster_state = cluster_state
        self.observer_state = observer_state
        self.config = config
        self.create_alert_file = create_alert_file
        self.db_conns = {}
        self.trigger_check_queue = trigger_check_queue
        self.session = requests.Session()
        if self.config.get("syslog"):
            self.syslog_handler = logutil.set_syslog_handler(
                address=self.config.get("syslog_address", "/dev/log"),
                facility=self.config.get("syslog_facility", "local2"),
                logger=self.log,
            )
        self.log.debug("Initialized ClusterMonitor with: %r", cluster_state)

    def _connect_to_db(self, instance, dsn):
        conn = self.db_conns.get(instance)
        if conn:
            return conn
        if not dsn:
            self.log.warning("Can't connect to %s, dsn is %r", instance, dsn)
            return None
        inst_info_str = "{0!r} ({1})".format(instance, mask_connection_info(dsn))
        try:
            self.log.info("Connecting to %s", inst_info_str)
            conn = psycopg2.connect(dsn=dsn, async_=True)
            wait_select(conn)
            self.log.debug("Connected to %s", inst_info_str)
        except (PglookoutTimeout, psycopg2.OperationalError) as ex:
            self.log.warning("%s (%s) connecting to %s (%s)",
                             ex.__class__.__name__, str(ex).strip(),
                             instance, inst_info_str)
            if "password authentication" in getattr(ex, "message", ""):
                self.create_alert_file("authentication_error")
            conn = None  # make sure we don't try to use the connection if we timed out
        except Exception as ex:  # pylint: disable=broad-except
            self.log.exception("Failed to connect to %s (%s)",
                               instance, inst_info_str)
            self.stats.unexpected_exception(ex, where="_connect_to_db")
            conn = None
        self.db_conns[instance] = conn
        return conn

    def _fetch_observer_state(self, instance, uri):
        result = {"fetch_time": get_iso_timestamp(), "connection": True}
{'fraction_non_alphanumeric': 0.09747734508939505, 'fraction_numerical': 0.00612294881214793, 'mean_word_length': 2.4576629974597797, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4200222', 'n_tokens_mistral': 2380, 'n_tokens_neox': 2303, 'n_words': 597}