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

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS MySQL,高可用系列 2核4GB
简介:

    最近公司做数据库的单台服务器多实例,需要对各个数据库的进程进行监控,因为我们的资产数据库中已经有这个服务器跑多少个实例,端口是什么,用来做什么的,所以就拿资产数据库的端口和服务器本地运行的数据库端口进行对比,报出没有运行的mysql实例以及他的用途。一种方法是根据"ps auxww|grep mysqld|grep -v root|grep -v grep"这个命令抓取本地运行的数据库端口,不过你也可以根据netstat命令来获取本地数据库实例的所有端口;还有一种方法就是从资产中得到这个服务器应该运行的所有mysql端口,用python的socket模块来检测端口是否存活,这种方法比较简单一些。我用第一种方法是因为我的这段代码我已经写过用于其他用途,等于复用,图个省事。以下是代码内容:

 
  1. #!/bin/env python 
  2. # -*- coding: utf-8 -*- 
  3.  
  4. import os,sys,MySQLdb 
  5.  
  6. def center(sql):#连接数据库 
  7.     try
  8.         center_ip = '192.168.1.100' 
  9.         center_user = 'root' 
  10.         center_passwd = 'xxxxxx' 
  11.         conn = MySQLdb.connect(host = center_ip,user = center_user,passwd = center_passwd,charset='utf8',connect_timeout=20
  12.         cursor = conn.cursor()  
  13.         cursor.execute(sql) 
  14.         alldata = cursor.fetchall() 
  15.         cursor.close() 
  16.         conn.close() 
  17.         return alldata 
  18.     except
  19.         return 0 
  20.  
  21. class check_port():#走资产中获取本地IP中应该有多少个mysql实例端口 
  22.  
  23.     def __init__(self): 
  24.         conn = "ip a|grep glob|grep -v '192.168'|awk '{print $2}'" 
  25.         self.host = os.popen(conn).readlines()[0].split("/")[0
  26.  
  27.     def remot(self): 
  28.         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 
  29.         alldata = center(sql) 
  30.         cent_port = [] 
  31.         if alldata != 0
  32.             for i in alldata: 
  33.                 cent_port.append(str(i[0])) 
  34.             return cent_port 
  35.         else
  36.             return cent_port 
  37.  
  38.     def local(self):#获取本地mysql有多少个实例运行 
  39.         psinfo = os.popen("ps auxww|grep mysqld|grep -v root|grep -v grep").readlines() 
  40.         local_port = [] 
  41.         if not psinfo: 
  42.             return local_port 
  43.         for i in psinfo: 
  44.             for j in i.split("--"): 
  45.                 if j.find("port") != -1
  46.                     port = j.split("=")[1].strip() 
  47.                     local_port.append(port) 
  48.         return local_port 
  49.  
  50.     def main(self): 
  51.         local_port = self.local() 
  52.         cent_port = self.remot() 
  53.                 cent_port.sort() 
  54.                 local_port.sort() 
  55.         if local_port == cent_port and len(local_port) != 0 and len(cent_port) != 0
  56.             print 0 
  57.         else
  58.             error = "" 
  59.             diff_list = list(set(local_port) ^ set(cent_port)) 
  60.             for port in diff_list: 
  61.                 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) 
  62.                 alldata = center(sql) 
  63.                 if error == "": 
  64.                     error = error + alldata[0][0
  65.                 else
  66.                     error = error + ";" + alldata[0][0
  67.             print error 
  68.  
  69. if __name__ == "__main__"
  70.     boss = check_port() 
  71.     boss.main() 

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

 
  1. ipmort socket 
  2.  
  3. def test_port() 
  4.     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  5. s.settimeout(2) 
  6.     address = '127.0.0.1' 
  7.     port = 80 
  8.     try
  9.         s.connect((address,port)) 
  10.         return True 
  11.     except Exception,e: 
  12.         return False     
  13. s.close()   

 


本文转自 lover00751CTO博客,原文链接:http://blog.51cto.com/wangwei007/1118809,如需转载请自行联系原作者


相关实践学习
如何快速连接云数据库RDS MySQL
本场景介绍如何通过阿里云数据管理服务DMS快速连接云数据库RDS MySQL,然后进行数据表的CRUD操作。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助     相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
相关文章
|
7天前
|
数据挖掘 vr&ar C++
让UE自动运行Python脚本:实现与实例解析
本文介绍如何配置Unreal Engine(UE)以自动运行Python脚本,提高开发效率。通过安装Python、配置UE环境及使用第三方插件,实现Python与UE的集成。结合蓝图和C++示例,展示自动化任务处理、关卡生成及数据分析等应用场景。
53 5
|
22天前
|
存储 Oracle 关系型数据库
服务器数据恢复—华为S5300存储Oracle数据库恢复案例
服务器存储数据恢复环境: 华为S5300存储中有12块FC硬盘,其中11块硬盘作为数据盘组建了一组RAID5阵列,剩下的1块硬盘作为热备盘使用。基于RAID的LUN分配给linux操作系统使用,存放的数据主要是Oracle数据库。 服务器存储故障: RAID5阵列中1块硬盘出现故障离线,热备盘自动激活开始同步数据,在同步数据的过程中又一块硬盘离线,RAID5阵列瘫痪,上层LUN无法使用。
|
1月前
|
PHP 数据库 数据安全/隐私保护
布谷直播源码部署服务器关于数据库配置的详细说明
布谷直播系统源码搭建部署时数据库配置明细!
|
1月前
|
存储 关系型数据库 MySQL
查询服务器CPU、内存、磁盘、网络IO、队列、数据库占用空间等等信息
查询服务器CPU、内存、磁盘、网络IO、队列、数据库占用空间等等信息
829 2
|
1月前
|
测试技术 API 数据安全/隐私保护
Python连接到Jira实例、登录、查询、修改和创建bug
通过使用Python和Jira的REST API,可以方便地连接到Jira实例并进行各种操作,包括查询、修改和创建Bug。`jira`库提供了简洁的接口,使得这些操作变得简单易行。无论是自动化测试还是开发工作流的集成,这些方法都可以极大地提高效率和准确性。希望通过本文的介绍,您能够更好地理解和应用这些技术。
166 0
|
2月前
|
Python
Python中的push方法详解与实例
Python中的push方法详解与实例
41 3
|
2月前
|
应用服务中间件 PHP Apache
PbootCMS提示错误信息“未检测到您服务器环境的sqlite3数据库扩展...”
PbootCMS提示错误信息“未检测到您服务器环境的sqlite3数据库扩展...”
|
2月前
|
数据可视化 Python
Python绘制基频曲线——实例解析与应用探讨
Python绘制基频曲线——实例解析与应用探讨
25 0
|
2月前
|
Python 容器
AutoDL Python实现 自动续签 防止实例过期释放 小脚本 定时任务 apscheduler requests
AutoDL Python实现 自动续签 防止实例过期释放 小脚本 定时任务 apscheduler requests
45 0
|
2月前
|
SQL 数据库
SQL-serve数据库不能连接本地服务器的解决方案
SQL-serve数据库不能连接本地服务器的解决方案
305 0