本文主要向大家介绍了Python语言 实现端口连通性检测,通过具体的内容向大家展示,希望对大家学习Python语言有所帮助。
# -*- coding: utf-8 -*-
#!/bin/env python
#AUTHOR:karl
#DATE:2018-1-19
#VERSION:V1.0
######################
import time
import os
import paramiko
import datetime
import sys
import MySQLdb
import threading
#######################################
#首次插入数据是REG=0,更新数据REG为其他
#######################################
REG=1
private_key = paramiko.RSAKey.from_private_key_file(‘/home/appdeploy/.ssh/id_rsa’)
def Data_mysql(info):
try:
string=info
Pip=string.split(” “)[0]
Sip=string.split(” “)[1]
Dip=string.split(” “)[2]
Port=string.split(” “)[3]
Result=string.split(” “)[4].replace(‘\n’,”)
con = MySQLdb.connect(host=’localhost’,user=”root”,passwd=”******”,db=”zabbix”,port=3306,charset=”utf8″)
dt = datetime.datetime.now().strftime(“%Y-%m-%d %H:%M:%S”)
cursor=con.cursor()
valuse=(dt,Result,Pip,Sip,Dip,Port)
if REG == 0:
sql=”””INSERT INTO Balant_telnet
(create_time,telnet_result,telnet_physicalip,telnet_sourceip,telnet_desip,telnet_port)
VALUES (%s,%s,%s,%s,%s,%s) “””
result=cursor.execute(sql,valuse)
else:
sql=”””UPDATE Balant_telnet set
create_time=’%s’,telnet_result=%s where telnet_physicalip=’%s’ and
telnet_sourceip=’%s’ and telnet_desip=’%s’ and telnet_port=’%s’;”””
%(valuse)
result=cursor.execute(sql)
cursor.close()
con.commit()
con.close()
except MySQLdb.Error, e:
print “Error %d: %s” % (e.args[0], e.args[1])
sys.exit(1)
def check_port(value,content):
ager={}
ager[value]=content
ssh=paramiko.SSHClient()
try:
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(value,port=22,username=’appdeploy’,pkey=private_key,timeout=5)
cmd=”sh /home/appdeploy/monitor.sh {value}”.format(value=ager[ager.keys()[index]])
stdin,stdout,stderr = ssh.exec_command(cmd)
for file_d in stdout.readlines():
Data_mysql(file_d)
except:
result=cmd+’,’+’failed’+’\n’
ssh.close()
###################################
# 对配置文件进行参数解析
###################################
def get_parameter():
ager={}
value=[]
with open(“monitor_config”) as context:
for line in context:
while not “,” in list(line):
host=line.strip()[1:-1]
value=[]
break
line=line.strip().split(“,”)
value=value+line
ager[str(host)]=value
return ager
if __name__ == ‘__main__’:
threads=[]
parm=get_parameter()
files=range(len(parm.keys()))
for index,content in parm.items():
t = threading.Thread(target=check_port,args=(index,content))
threads.append(t)
for index in files:
threads[index].start()
for index in files:
threads[index].join()
配置文件:
[10.117.194.23]
10.117.194.77,10.116.41.82,9920
10.117.194.77,10.116.47.12,1080
10.117.194.77,10.116.45.56,1081
[10.117.194.24]
10.117.194.78,10.116.41.82,9920
10.117.194.78,10.116.47.12,1080
10.117.194.78,10.116.45.56,1081
10.117.194.78,10.116.145.33,8001
【】里面的IP 是物理IP,下面分别对应着应用的源IP,目标IP和测试端口号
agent 直接用shell 实现.
数据库中效果
在zabbix 中 加入对表中数据的监控。在zabbix 的mysql 模板中加入Balant 的监控项,也可以加入图像,我们最后的结果是要展示在grafana 上进行展示。为1就说明应用端口是正常运行。
本文由职坐标整理并发布,希望对同学们学习Python有所帮助,更多内容请关注职坐标编程语言Python频道!
今天的文章python 测试端口连通_Python语言 实现端口连通性检测分享到此就结束了,感谢您的阅读,如果确实帮到您,您可以动动手指转发给其他人。
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
如需转载请保留出处:https://bianchenghao.cn/24698.html