Greenplum集群监控小试

Greenplum集群监控小试

这是学习笔记的第 2235篇文章读完需要9分钟速读仅需7分钟

对于PostgreSQL的监控，行业里多多少少还是有不少的开源方案可用的，基本上拿过来修修补补就能跑起来。但是对于Greenplum集群的监控方案比较少，有的同学会说GPCC也可以啊，我有以下的几点考虑。

1）GPCC的监控还是缺少一些系统层的监控细节，比如缺少每个数据segment节点网卡的流量，比如每个数据segment节点的IO负载历史，监控历史的区间范围过大等，同时一般的监控我们可能会跑偏方向，更多是在Master上面监控，但是显然这样是缺少一些真实信息的。

2）GPCC的功能会有独立的数据档案库gpperfmon,程序交互是一个独立的客户端

3）GPCC偏重于集群管理，还有防火墙变更等系统管理功能

4）对于监控选项不能够灵活扩展

我的核心需求是关于Greenplum的监控，在监控方面可以实现扩展，所以也算是做了一个尝试。

这个过程也碰到几个问题。

1）数据字典的选取，PG和GP的大部分数据字典是兼容的，但是GP体系有自己独有的数据字典，比如重要的gp_segment_configuration,整体的信息主要是通过专有的数据字典完成的

2）关于psycopg2版本的兼容性，驱动psycopg2的版本对于Python的版本比较敏感，导致会有一些比较尴尬的情况，比如GP自带的Python驱动是3.6.9，而操作系统层默认的是3.6.3左右，结果不同的用户调用就会有完全不同的结果。补充操作是对.bash_profile刻意做区分，做不同的配置选项。

3）监控脚本的补充和完善，目前这块的脚本是比较少的，所以也就自己在PG的基础上开始做一些改动。

整个改造的过程整体还是比较顺利的，能够基本实现集群信息概览，系统资源概览。

对数据segment节点的系统资源使用取平均值，去除毛刺，这样能够看到一个整体的使用情况。

对网卡流量进行聚合，可以看到整个集群的吞吐量情况。

关于监控的脚本，是在TCollector中PostgreSQL的脚本基础上改进的。对于单机多实例，同事也帮忙在统一接入的地方做了整合，在一个独立的配置文件中，使用的是一个JSON配置，

{"ip":"xxxxx","port":5432,"idc":"xxxx","service":"xxxxx"}

监控脚本内容如下：

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://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))

剩下的就是不断的调试和修正，直到看起来还像回事。

QQ群号：763628645

QQ群二维码如下， 添加请注明：姓名+地区+职位，否则不予通过

7

近期热文

我眼中的《庆余年》使用Python分析北京积分落户数据，分析完我陷入了深思MySQL的主键命名挺任性，就这么定了华裔教授发现二次方程极简解法,我默默的做了下验算回答:我不小心把公司的数据库给删了，该不该离职？迁移到MySQL的业务架构演进实战数据库修改密码风险高，如何保证业务持续，这几种密码双活方案可以参考MySQL业务双活的初步设计方案如何优化MySQL千万级大表，我写了6000字的解读一道经典的MySQL面试题，答案出现三次反转业务双活的数据切换思路设计（下）业务双活的数据切换思路设计（一）MySQL中的主键和rowid,看似简单,其实有一些使用陷阱需要注意小白学MySQL要多久?我整理了10多个问题的答案

8

去IOE or Not?拉里·佩奇（Larry Page）的伟大归来《吊打面试官》系列-Redis基础唯一ID生成算法剖析，看看这篇就够了关于大数据运维能力的一些思考DBA菜鸟的进化简史：不忘初心，记工作中踩过的三个坑美女主持直播,被突发意外打断!湾区网友却高喊: 我懂!超甜