pythonは、単一のマルチインスタンス・データベース・サーバのデータベース・ポートを監視します.

最近会社はデータベースの1台のサーバーの多実例をして、各データベースのプロセスに対してモニタリングを行う必要があります.私たちの資産データベースにはすでにこのサーバーが何個の実例を走っているのか、ポートが何なのか、何に使うのか、資産データベースのポートとサーバーがローカルに運行しているデータベースのポートを比較します.実行されていないmysqlインスタンスと彼の用途を報告します.1つの方法は、「ps auxww|grep mysqld|grep-v root|grep-v grep」というコマンドに基づいて、ローカルで実行されているデータベースポートをキャプチャすることですが、netstatコマンドに基づいてローカルデータベースインスタンスのすべてのポートを取得することもできます.もう1つの方法は、このサーバが実行すべきすべてのmysqlポートを資産から取得し、pythonのsocketモジュールでポートが生存しているかどうかを検出する方法です.この方法は簡単です.私が最初の方法を使ったのは、私のこのコードが他の用途に使われているためです.多重化に等しく、手間が省けます.コードの内容は次のとおりです.

  
  
  
  

   
   
   
   
#!/bin/env python 
   
   
   
   
# -*- coding: utf-8 -*- 
   
   
   
   
 
   
   
   
   
import os,sys,MySQLdb 
   
   
   
   
 
   
   
   
   
def center(sql):#      
   
   
   
   
    try: 
   
   
   
   
        center_ip = '192.168.1.100' 
   
   
   
   
        center_user = 'root' 
   
   
   
   
        center_passwd = '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.sub_name,'_',c.app_name,'_',c.port) AS used FROM center_app.main_category a, center_app.sub_category 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() 
  
  
  
  

2つ目の方法を使用すると、簡単です.次の関数でこのポートテストを実現できます.

  
  
  
  

   
   
   
   
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()