Greenplum集群监控小试

import sys
import os
import time
import socket
import errno
try:
    import psycopg2
except ImportError:
    psycopg2 = None  # handled in main()
COLLECTION_INTERVAL = 15  # seconds
CONNECT_TIMEOUT = 2  # seconds
SCAN_INTERVAL = 300
from collectors.lib import utils
from collectors.etc import greenplumconf
# Directories under which to search socket files
SEARCH_DIRS = frozenset([
    "/var/run/postgresql",  # Debian default
    "/var/pgsql_socket",  # MacOS default
    "/usr/local/var/postgres",  # custom compilation
    "/tmp",  # custom compilation
])
def now():
    return int(time.time())
def find_sockdir():
    """Returns a path to PostgreSQL socket file to monitor."""
    for dir in SEARCH_DIRS:
        for dirpath, dirnames, dirfiles in os.walk(dir, followlinks=True):
            for name in dirfiles:
                # ensure selection of PostgreSQL socket only
                if (utils.is_sockfile(os.path.join(dirpath, name))
                        and "PGSQL" in name):
                    return (dirpath)
def postgres_connect(host, port):
    """Connects to the PostgreSQL server using the specified socket file."""
    user, password = greenplumconf.get_user_password()
    try:
        return psycopg2.connect("host='%s' port='%s' user='%s' password='%s' "
                                "connect_timeout='%s' dbname=postgres"
                                % (host, port, user, password,
                                   CONNECT_TIMEOUT))
    except (EnvironmentError, EOFError, RuntimeError), e:
        utils.err("Couldn't connect to DB :%s" % (e))
def collect(db, tags_str):
    """
    Collects and prints stats.
    Here we collect only general info, for full list of data for collection
    see http://www.postgresql.org/docs/9.2/static/monitoring-stats.html
    """
    try:
        cursor = db.cursor()
        # general statics
        cursor.execute("SELECT pg_stat_database.*"
                       " FROM pg_database JOIN"
                       " pg_stat_database ON pg_database.datname ="
                       " pg_stat_database.datname WHERE pg_stat_database.datname"
                       " NOT IN ('template0', 'template1', 'postgres')")
        ts = time.time()
        stats = cursor.fetchall()
        #  datid |  datname   | numbackends | xact_commit | xact_rollback | blks_read  |  blks_hit   | tup_returned | tup_fetched | tup_inserted | tup_updated | tup_deleted | conflicts | temp_files |  temp_bytes  | deadlocks | blk_read_time | blk_write_time |          stats_reset          |     size
        result = {}
        for stat in stats:
            database = stat[1]
            result[database] = stat
        for database in result:
            for i in range(2, len(cursor.description)):
                metric = cursor.description[i][0]
                value = result[database][i]
                try:
                    if metric in ("stats_reset"):
                        continue
                    print ("greenplum.%s %i %s database=%s %s"
                           % (metric, ts, value, database, tags_str))
                except:
                    utils.err("got here")
                    continue
        # DB connections
        cursor.execute("SELECT datname, count(datname) FROM pg_stat_activity"
                       " GROUP BY datname")
        ts = time.time()
        dbconnections = cursor.fetchall()
        for database, connection in dbconnections:
            print ("greenplum.dbconnections %i %s database=%s %s"
                   % (ts, connection, database, tags_str))
        # DB segment status
        cursor.execute("select role,count(*) from gp_segment_configuration "
                       "  group by role")
        ts = time.time()
        segment_roles = cursor.fetchall()
        for segment_role, role_no in segment_roles:
            print ("greenplum.segment_roles %i %s segment_role=%s %s"
                   % (ts, role_no, segment_role, tags_str))
        cursor.execute("select status,count(*) from gp_segment_configuration "
                       "  group by status")
        ts = time.time()
        segments_status = cursor.fetchall()
        for segment_status, status_no in segments_status:
            print ("greenplum.segment_status %i %s segment_role=%s %s"
                   % (ts, status_no, segment_status, tags_str))
        cursor.execute("select count(*) from gp_segment_configuration where status='d' ")
        ts = time.time()
        dead_segments_no = cursor.fetchone()
        for dead_segemnt_no in dead_segments_no:
            print ("greenplum.dead_segement_no %i %s %s"
                   % (ts, dead_segemnt_no, tags_str))
    except (EnvironmentError, EOFError, RuntimeError,), e:
        if isinstance(e, IOError) and e[0] == errno.EPIPE:
            # exit on a broken pipe. There is no point in continuing
            # because no one will read our stdout anyway.
            return 2
        utils.err("error: failed to collect data: %s" % e)
def scan_for_instances():
    try:
        f = open("/etc/tcollector/greenplum_tags.conf")
    except IOError, e:
        tils.err("error: can't open /etc/tcollector/tags.conf : %s" % e)
        return None
    f.seek(0)
    out = {}
    for each_data in f:
        tags = eval(each_data)
        port = tags["port"]
        # tags_str = "endpoint=%s port=%s idc=%s service=%s cluster_type=%s db_soft=%s db_engins=%s" % (tags["ip"], tags["port"], tags["idc"], tags["service"], tags["cluster_type"], tags["db_soft"], tags["db_engine"])
        tags_str = "endpoint=%s port=%s idc=%s service=%s" % (tags["ip"], tags["port"], tags["idc"], tags["service"])
        out[port] = {}
        out[port]["tags_map"] = tags
        out[port]["tags_str"] = tags_str
    return out
def main(args):
    """Collects and dumps stats from a PostgreSQL server."""
    if psycopg2 is None:
        utils.err("error: Python module 'psycopg2' is missing")
        return 13  # Ask tcollector to not respawn us
    # sockdir = find_sockdir()
    # if not sockdir: # Nothing to monitor
    #  utils.err("error: Can't find postgresql socket file")
    #  return 13 # Ask tcollector to not respawn us
    last_scan = now()
    instances = scan_for_instances()
    if not len(instances):
        return 13
    while True:
        ts = now()
        if ts - last_scan >= SCAN_INTERVAL:
            instances = scan_for_instances()
            last_scan = ts
        for port in instances:
            tags_str = instances[port]["tags_str"]
            tags_map = instances[port]["tags_map"]
            db = postgres_connect(tags_map["ip"], tags_map["port"])
            collect(db, tags_str)
        sys.stdout.flush()
        time.sleep(COLLECTION_INTERVAL)
if __name__ == "__main__":
    sys.stdin.close()
    sys.exit(main(sys.argv))