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&＃61;0&＃xff0c;更新数据REG为其余

#######################################

REG&＃61;1

private_key &＃61; paramiko.RSAKey.from_private_key_file(&＃39;/home/appdeploy/.ssh/id_rsa&＃39;)

def Data_mysql(info):

try:

string&＃61;info

Pip&＃61;string.split(" ")[0]

Sip&＃61;string.split(" ")[1]

Dip&＃61;string.split(" ")[2]

Port&＃61;string.split(" ")[3]

Result&＃61;string.split(" ")[4].replace(&＃39;\n&＃39;,&＃39;&＃39;)

con &＃61; MySQLdb.connect(host&＃61;&＃39;localhost&＃39;,user&＃61;"root",passwd&＃61;"******",db&＃61;"zabbix",port&＃61;3306,charset&＃61;"utf8")

dt &＃61; datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

cursor&＃61;con.cursor()

valuse&＃61;(dt,Result,Pip,Sip,Dip,Port)

if REG &＃61;&＃61; 0:

sql&＃61;"""INSERT INTO Balant_telnet (create_time,telnet_result,telnet_physicalip,telnet_sourceip,telnet_desip,telnet_port) VALUES (%s,%s,%s,%s,%s,%s) """

result&＃61;cursor.execute(sql,valuse)

else:

sql&＃61;"""UPDATE Balant_telnet set create_time&＃61;&＃39;%s&＃39;,telnet_result&＃61;%s where telnet_physicalip&＃61;&＃39;%s&＃39; and telnet_sourceip&＃61;&＃39;%s&＃39; and telnet_desip&＃61;&＃39;%s&＃39; and telnet_port&＃61;&＃39;%s&＃39;;""" %(valuse)

result&＃61;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&＃61;{}

ager[value]&＃61;content

ssh&＃61;paramiko.SSHClient()

try:

ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

ssh.connect(value,port&＃61;22,username&＃61;&＃39;appdeploy&＃39;,pkey&＃61;private_key,timeout&＃61;5)

cmd&＃61;"sh /home/appdeploy/monitor.sh {value}".format(value&＃61;ager[ager.keys()[index]])

stdin,stdout,stderr &＃61; ssh.exec_command(cmd)

for file_d in stdout.readlines():

Data_mysql(file_d)

except:

result&＃61;cmd&＃43;&＃39;,&＃39;&＃43;&＃39;failed&＃39;&＃43;&＃39;\n&＃39;

ssh.close()

###################################

# 对配置文件进行参数解析

###################################

def get_parameter():

ager&＃61;{}

value&＃61;[]

with open("monitor_config") as context:

for line in context:

while not "," in list(line):

host&＃61;line.strip()[1:-1]

value&＃61;[]

break

line&＃61;line.strip().split(",")

value&＃61;value&＃43;line

ager[str(host)]&＃61;value

return ager

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

threads&＃61;[]

parm&＃61;get_parameter()

files&＃61;range(len(parm.keys()))

for index,content in parm.items():

t &＃61; threading.Thread(target&＃61;check_port,args&＃61;(index,content))

threads.append(t)

for index in files:

threads[index].start()

for index in files:

threads[index].join()

配置文件&＃xff1a;

python

[10.117.194.23]mysql

10.117.194.77,10.116.41.82,9920sql

10.117.194.77,10.116.47.12,1080shell

10.117.194.77,10.116.45.56,1081数据库

[10.117.194.24]app

10.117.194.78,10.116.41.82,9920ssh

10.117.194.78,10.116.47.12,1080ide

10.117.194.78,10.116.45.56,1081测试

10.117.194.78,10.116.145.33,8001orm

【】里面的IP 是物理IP&＃xff0c;下面分别对应着应用的源IP&＃xff0c;目标IP和测试端口号

agent 直接用shell 实现.

数据库中效果

在zabbix 中 加入对表中数据的监控。在zabbix 的mysql 模板中加入Balant 的监控项&＃xff0c;也能够加入图像&＃xff0c;咱们最后的结果是要展现在grafana 上进行展现。为1就说明应用端口是正常运行。