python监控单台多实例数据库服务器的数据库端口

数据库的服务器端口是什么，用来做什么的，拿资产数据库的端口和服务器本地运行的数据库端口进行对比，报出没有运行的mysql实例以及他的用途。一种方法是根据"ps auxww|grep mysqld|grep -v root|grep -v grep"这个命令抓取本地运行的数据库端口，也可以根据netstat命令来获取本地数据库实例的所有端口；还有一种方法就是从资产中得到这个服务器应该运行的所有mysql端口，用python的socket模块来检测端口是否存活，这种方法比较简单一些。笔者使用第一种方法是因为这段代码已经写过并用于其他用途，等于复用，图个省事。以下是代码内容：

#!/bin/env python
# -- coding: utf-8 --

import os,sys,MySQLdb

def center(sql):#连接数据库
try:
centerip = '192.168.1.100'
centeruser = 'root'
centerpasswd = 'xxxxxx'
conn = MySQLdb.connect(host = center_ip,user = center_user,passwd = center_passwd,charset='utf8',connect_timeout=20)
cursor = conn.cursor()
cursor.execute(sql)
alldata = cursor.fetchall()
cursor.close()
conn.close()
return alldata
except:
return 0

class check_port():#在资产中获取本地IP中应该有多少个mysql实例端口

def __init(self):
conn = "ip a|grep glob|grep -v '192.168'|awk '{print $2}'"
self.host = os.popen(conn).readlines()[0].split("/")[0]

def remot(self):
sql = "SELECT PORT FROM center.host_info WHERE flag='1' AND del_info!=0 AND ip='%s' AND host_name IN ('a','b','c')" % self.host
alldata = center(sql)
cent_port = []
if alldata != 0:
for i in alldata:
cent_port.append(str(i[0]))
return cent_port
else:
return cent_port

def local(self):#获取本地mysql有多少个实例运行
psinfo = os.popen("ps auxww|grep mysqld|grep -v root|grep -v grep").readlines()
local_port = []
if not psinfo:
return local_port
for i in psinfo:
for j in i.split("—"):
if j.find("port") != -1:
port = j.split("=")[1].strip()
local_port.append(port)
return local_port

def main(self):
local_port = self.local()
cent_port = self.remot()
cent_port.sort()
local_port.sort()
if local_port == cent_port and len(local_port) != 0 and len(cent_port) != 0:
print 0
else:
error = ""
diff_list = list(set(local_port) ^ set(cent_port))
for port in diff_list:
sql = "SELECT CONCAT(a.main_name,'',b.subname,'',c.appname,'',c.port) AS used FROM centerapp.maincategory a, centerapp.subcategory b, center_app.app_info c WHERE a.id = b.main_id AND b.dist_id = c.dist_id AND b.main_id = c.main_id AND b.main_id='2' AND c.flag='1' AND c.del_info!=0 AND c.ip='%s' and c.port='%s'" % (self.host,port)
alldata = center(sql)
if error == "":
error = error + alldata[0][0]
else:
error = error + ";" + alldata[0][0]
print error

if __name == "__main":
boss = check_port()
boss.main()

如果用第二种方法的话，很简单，用下面的函数可以实现这个端口测试：

ipmort socket

def test_port()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(2)
address = '127.0.0.1'
port = 80
try:
s.connect((address,port))
return True
except Exception,e:
return False
s.close()