python脚本对域名监控

服务器常对外提供tcp和http两种接口，我们判断服务和域名是否可用，有多种方法，比如在服务器上写个程序定期执行ps命令可以判断程序是否在运行，用访问域名的方式判断域名是否可行。

下面的python程序是我们自己常用的脚本程序，用来判断域名及服务是否可行，供各位小伙伴一起参考，如有问题，请多多指点，有不清楚也可以留言，我看到会回复

#!/usr/bin/python
# -*- coding: UTF-8 -*-
import thread
import threading
import httplib
import time
import Queue
import json
import os
import sys
import urllib
import socket
import math
import datetime# 线程列表
threads = []# 发往钉钉的消息队列
ddQueueLock = threading.Lock()
ddMsgQueue = Queue.Queue(1024)# 文件日志的消息队列
logQueueLock = threading.Lock()
logMsgQueue = Queue.Queue(1024)def current_dir():return os.path.dirname(os.path.abspath(sys.argv[0]))#将utc时间转化为北京时间
def beijin_time():timenow = (datetime.datetime.utcnow() + datetime.timedelta(hours=8)) return timenow.strftime('%Y-%m-%d %H:%M:%S')def in_dd_queue(msg):bjTime = beijin_time()msgFilter = "[北京时间:%s]%s" % (bjTime, msg)#ddQueueLock.acquire()ddMsgQueue.put(msgFilter)#ddQueueLock.release()def out_dd_queue():#ddQueueLock.acquire()return ddMsgQueue.get()#ddQueueLock.release()def in_log_queue(msg):bjTime = beijin_time()msgFilter = "[北京时间:%s]%s" % (bjTime, msg)#logQueueLock.acquire()logMsgQueue.put(msgFilter)#logQueueLock.release()def out_log_queue():#logQueueLock.acquire()return logMsgQueue.get()#logQueueLock.release()def httpGetRequest(protocol, host, port, path, reqTimeout, respStatus, respReason, respBody):returnCode = returnValue = Noneconn = Nonetry:if protocol == "https":conn = httplib.HTTPSConnection(host, port, timeout=reqTimeout)else:conn = httplib.HTTPConnection(host, port, timeout=reqTimeout)conn.request("GET", path, headers = {"User-Agent": "Mozilla/5.0 (Windows; U; Windows NT 5.1; zh-CN; rv:1.9.1) Gecko/20090624 Firefox/3.5"})result = conn.getresponse()status = result.statusreason = result.reasonbody = result.read().strip()# 这里的逻辑自定义：实际接收的返回值和期望的返回值的对比if (status == respStatus and reason == respReason and body == respBody):returnCode = returnValue = "ok"else:returnCode = "error"returnValue = "<请求>%s://%s:%d%s, \r\n <返回> status:%d, reason:%s, body:%s" \% (protocol, host, port, path, status, reason, body)except Exception as e:returnCode = "error"returnValue = "<请求>%s://%s:%d%s, \r\n <返回错误>exception:%s" \% (protocol, host, port, path, e)finally:if conn:conn.close()return returnCode, returnValuedef tcpRequest(host, port, reqTimeout, reqType):returnCode = returnValue = Nones = Nonetry:s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)s.settimeout(reqTimeout)s.connect((host, port))# 这里发送你自定义的消息格式。 这里假设发送字符串hellos.settimeout(reqTimeout)s.send("hello")# 这里接收以后，你自己的处理逻辑。 这里假设接收是字符是hi,i am xxxxxs.settimeout(reqTimeout)recvData=s.recv(1024)recvIndex=recvData.find('hi') # 查找返回结果里有没有hiif recvIndex == -1:returnCode = "error"returnValue = "<TCP连接>%s %d \r\n <错误> exception:%s" % (host, port, recvData)else:returnCode = returnValue = "ok"except Exception as e:returnCode = "error"returnValue = "<TCP连接>%s %d \r\n <错误> exception:%s" % (host, port, e)finally:if s:s.close()return returnCode, returnValue# 200, OK的情况下，判断一下钉钉的返回body是不是对的。
# 钉钉正常的返回值body是：{"errcode":0,"errmsg":"ok"}
def parse_dd_response(body):jsonDict = NonereturnErrCode = NonereturnErrMsg = Nonetry:jsonDict = json.loads(body)errCode = jsonDict['errcode'],returnErrCode = "%d" % (errCode)if returnErrCode == "0":returnErrMsg = "ok"else:returnErrMsg = jsonDict['errmsg']except Exception as e:returnErrCode = "-1"returnErrMsg = "%s" % (e)return returnErrCode, returnErrMsgdef send_dd_msg(msg, Timeout):formatMsg = "[KEY]%s" % (msg) #内容要有关键字，参见钉钉自定义机器人接入文档text = {"msgtype":"text", "text":{"content":formatMsg}}requrl = "https://oapi.dingtalk.com/robot/send?access_token=your_robot_access_token"conn = Nonetry:headerdata = {"Content-type": "application/json"}conn = httplib.HTTPSConnection("oapi.dingtalk.com", 443, timeout=Timeout)conn.request(method='POST',url=requrl, body=json.dumps(text), headers = headerdata)result = conn.getresponse()status = result.statusreason = result.reasonbody = result.read().strip()if (status == 200 and reason == "OK"):r1, r2 = parse_dd_response(body)if r1 != "0":in_log_queue("DingDing response status:%d, reason:%s, body:%s, send_msg:%s" % (status, reason, body, formatMsg))else:in_log_queue("DingDing response status:%d, reason:%s, body:%s, send_msg:%s" % (status, reason, body, formatMsg))except Exception as e:in_log_queue("DingDing send error:%s, send_msg:%s" % (e, formatMsg))finally:if conn:conn.close()def write_log(filePath, msg):fd = NoneisExist = os.path.exists(filePath)if isExist == True:filePath1 = unicode(filePath,'utf8')fsize = os.path.getsize(filePath1)fsize = fsize/float(1024*1024)fsize1 = math.floor(fsize)if fsize1 > 10:#大于10MB时，清空重写openMode = "w"else:openMode = "a+"else:openMode = "w"try:fd = open(filePath, mode=openMode)fd.write("%s\n" % (msg))except Exception as e:# send to dd.Error = "[write file error]:%s" % (e)in_dd_queue(Error)finally:if fd:fd.close()class NoticeThread (threading.Thread):def __init__(self, threadID, name):threading.Thread.__init__(self)self.threadID = threadIDself.name = namedef run(self):fileName = os.path.basename(sys.argv[0]).split(".")[0]print "Starting " + self.namein_log_queue("Starting %s" % (self.name))while 1:if not ddMsgQueue.empty():send_dd_msg("[%s]%s" % (fileName, out_dd_queue()), 5)time.sleep(1)else:time.sleep(1)class LogThread (threading.Thread):def __init__(self, threadID, name):threading.Thread.__init__(self)self.threadID = threadIDself.name = namedef run(self):curDir = current_dir()filePath = "%s/monitor.log" % curDirwhile 1:if not logMsgQueue.empty():write_log(filePath, out_log_queue())time.sleep(1)else:time.sleep(2)class HttpReqThread (threading.Thread):def __init__(self, threadID, name, protocol, hosts, port, path, reqTimeout, respStatus, respReason, respBody, intervalTimeout, startAfterTime):threading.Thread.__init__(self)self.threadID = threadIDself.name = nameself.protocol = protocolself.hosts = hostsself.port = portself.path = pathself.reqTimeout = reqTimeoutself.respStatus = respStatusself.respReason = respReasonself.respBody = respBodyself.intervalTimeout = intervalTimeoutself.startAfterTime = startAfterTimedef run(self):                  print "Starting " + self.namein_log_queue("Starting %s" % (self.name))time.sleep(self.startAfterTime)while 1:for host in self.hosts:status, reason = httpGetRequest(self.protocol, host, self.port, self.path, self.reqTimeout, self.respStatus, self.respReason, self.respBody)if status == "error":in_dd_queue(reason)in_log_queue(reason)time.sleep(self.intervalTimeout)else:time.sleep(self.intervalTimeout)print "Exiting " + self.nameclass TcpReqThread (threading.Thread):def __init__(self, threadID, name, hosts, port, reqTimeout, intervalTimeout, startAfterTime):threading.Thread.__init__(self)self.threadID = threadIDself.name = nameself.hosts = hostsself.port = portself.reqTimeout = reqTimeoutself.intervalTimeout = intervalTimeoutself.startAfterTime = startAfterTimedef run(self):                  in_log_queue("Starting %s" % (self.name))time.sleep(self.startAfterTime)while 1:for host in self.hosts:status, reason = tcpRequest(host, self.port, self.reqTimeout)if status == "error":in_dd_queue(reason)in_log_queue(reason)time.sleep(self.intervalTimeout)else:time.sleep(self.intervalTimeout)print "Exiting " + self.namedef start_process(threadName, addInThreadQueue, startAfterSeconds):threadObj = Noneif threadName == "log_thread":thread = LogThread(100, "log_thread")thread.start()threadObj = threadelif threadName == "dd_thread":thread = NoticeThread(101, "notice_thread")thread.start()threadObj = threadelif threadName == "tcp_server_1":hosts = ["hello1.abc.com", "hello2.abc.com"] # 一个服务可以有多个域名，如果只有一个就写一个thread = TcpReqThread(2, threadName, hosts, 10000, 5, 60, startAfterSeconds)thread.start()threadObj = threadelif threadName == "http_server_1":hosts = ["hi1.abc.com"]thread = HttpReqThread(3, threadName, "https", hosts, 443, "/status", 5, 200, "OK", "ok", 60, startAfterSeconds)thread.start()threadObj = threadelse:print "unkonw server"if (addInThreadQueue == True and threadObj != None):threads.append(threadObj)def loop():while 1:for t in threads:isAlive = t.is_alive()if isAlive == False:in_dd_queue("%s 异常中止，自动尝试重启它" % t.name)in_log_queue("%s 异常中止，自动尝试重启它" % t.name)threads.remove(t)start_process(t.name, True, 0)time.sleep(5)time.sleep(5)if __name__ == "__main__":# 启动文件日志和发给钉钉消息的线程。start_process("log_thread", True, 0)start_process("dd_thread", True, 0)time.sleep(1)# 启动其它各类线程。start_process("tcp_server_1", True, 5)start_process("http_server_1", True, 10)loop()

python脚本对域名监控相关推荐

python自动发邮件 foxmail_使用 python 脚本实现自动监控网站并发送邮件告警
今天中午的时候,网站莫名奇妙地出现无法访问的现象,持续了两个多小时,等发现问题的时候立刻重启了服务器才恢复正常.为防止同样的事情再次发生,所以就想编写一个自动化脚本来监控网站服务,今天为大家分享一下使 ...
写了个Python脚本监控nginx进程
写了个Python脚本监控nginx进程 « Xiaoxia[PG] 写了个Python脚本监控nginx进程接上一文用iptables让SSH服务对陌生人说不.还是有点担心这个学期内,nginx可 ...
python脚本监控网站状态 - 赵海华_运维之路 - 51CTO技术博客
python脚本监控网站状态 - 赵海华_运维之路 - 51CTO技术博客 python脚本监控网站状态 2013-01-09 09:21:02 标签:监控 python 原创作品,允许转载,转载时请 ...
python连接oracle进行监控_使用Python脚本zabbix自定义key监控oracle连接状态
目的:此次实验目的是为了zabbix服务端能够实时监控某服务器上oracle实例能否正常连接环境:1.zabbix_server 2.zabbix_agent(含有oracle) 主要知识点: 1. ...
bat脚本中获取上级目录_使用Python写一个可以监控Tomcat 运行的脚本,并且把.py文件转换成.exe文件...
使用Python写一个可以监控Tomcat 运行的脚本,并且把.py文件转换成.exe文件文章来源与博主本人的CSDN博客,博客地址:https://blog.csdn.net/weixin_435 ...
利用python脚本程序监控文件被修改
需求: 利用python编写监控程序,监控一个文件目录,当目录下的文件发生改变时,实现有修改就发报警邮件邮件使用QQ邮箱,需要开启smtp,使用手机发生短信,腾讯会给你发邮箱密码.如下所示: 把这个 ...
python漏洞利用脚本_利用Python脚本实现漏洞情报监控与通知的经验分享
原标题:利用Python脚本实现漏洞情报监控与通知的经验分享前言本文主要介绍了笔者利用一个简单的Python脚本实现漏洞情报的监控以及自动通报的相关经验. 一.背景笔者所在公司某一个业务系统用到 ...
指令脚本redis线上环境监控脚本（python脚本）
在改章节中,我们要主介绍指令脚本的内容,自我觉感有个不错的议建和大家分享下近来一个月没啥新更,边身生发太多事,结业几年来霉运太多,虽然不信命,但我信有些性命的确好,有些性命的确差,其它不说也罢.(大 ...
python脚本监控docker容器
脚本功能: 监控CPU使用率监控内存使用状况监控网络流量 #!/usr/bin/env python # --*-- coding:UTF-8 --*-- import sys import ta ...

python脚本对域名监控

python脚本对域名监控相关推荐

最新文章

热门文章