快捷導(dǎo)航

Python3監(jiān)控windows,linux系統(tǒng)的CPU、硬盤、內(nèi)存使用率和各個端口的開啟情況詳細代碼實例

更新時間：2020年03月18日 09:07:13 作者：沙振宇

這篇文章主要介紹了Python3監(jiān)控windows,linux系統(tǒng)的CPU、硬盤、內(nèi)存使用率和各個端口的開啟情況詳細代碼實例,需要的朋友可以參考下

由于項目的需要，需要做一個簡單監(jiān)控服務(wù)器的CPU利用率、CPU負載、硬盤使用率、內(nèi)存利用率和服務(wù)器的各個端口的開啟情況的程序，并把結(jié)果通知到監(jiān)控平臺，如果出現(xiàn)異常，監(jiān)控平臺打電話或者發(fā)短信通知給具體的運維人員

python版本要求：python3.0 以上

安裝 python 的 psutil 包和 requests 包

pip install psutil

pip install requests

Linux系統(tǒng)下運行效果

Windows系統(tǒng)下運行效果

代碼實例核心程序

# 獲取端口信息
	@classmethod
	def get_ports(cls, port):
		sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		result = sock.connect_ex(('127.0.0.1',int(port)))
		if result != 0:
			send_data = cls.g_web_ip+"服務(wù)器的"+port+'端口掛了，快去修復(fù)哈'
			cls.send_msg(send_data)
		else:
			print("端口："+port+"正常")

	# CPU利用率
	@classmethod
	def get_cpu_used(cls):
		if (sysstr == "Linux"):
			f = os.popen("top -bi -n 1| awk '{print $4}'").read().split('\n')[2]
			float_cpu_used = float(f)
			float_g_cpu_used = float(cls.g_cpu_used.split("%")[0])
			print("CPU利用率:",f,"%")
			if float(float_cpu_used) > float(float_g_cpu_used):
				cls.send_msg(cls.g_web_ip+"服務(wù)器的CPU利用率超過"+cls.g_cpu_used+"了，快去看看咋回事！")
		else:
			print(sysstr + " CPU Adoption rate Cannot read.")
		printL()

	# CPU平均負載
	@classmethod
	def aver_load(cls):
		if (sysstr == "Linux"):
			f = os.popen("uptime | sed 's/,//g' | awk '{print $8,$9,$10}'")
			str_aver_load = f.read().strip().split(":")[1].strip()
			print("CPU平均負載:",str_aver_load)
			if float(str_aver_load) > float(cls.g_aver_load):
				cls.send_msg(cls.g_web_ip+"服務(wù)器的CPU平均負載超過"+cls.g_aver_load+"了，快去看看咋回事！")
		else:
			print(sysstr + " CPU Load average Cannot read.")
		printL()

	#獲取硬盤使用率
	@classmethod
	def get_disk_used(cls):
		if (sysstr == "Linux"):
			disk_val = os.popen("df -h | head -2 | tail -1 |awk '{print $5}'").read().strip()
			int_disk_val = int(disk_val.split("%")[0])
			int_g_disk_val = int(cls.g_disk_used.split("%")[0])
			print("硬盤使用率:",disk_val)
			if int_disk_val > int_g_disk_val:
				cls.send_msg(cls.g_web_ip+"服務(wù)器的硬盤使用率超過"+cls.g_disk_used+"了，快去看看咋回事！")
		else:
			print(sysstr + " hard disk Cannot read.")
		printL()

	# 獲取內(nèi)存使用率
	@classmethod
	def get_mem_used(cls):
		if (sysstr == "Linux"):
			f = os.popen("free -m |grep Mem |awk '{print $3/$2}'")
			str_men = f.read().strip()
			print("內(nèi)存使用率:",str_men)
			if float(str_men) > float(cls.g_mem_used):
				cls.send_msg(cls.g_web_ip+"服務(wù)器的內(nèi)存使用率超過"+cls.g_mem_used+"了，快去看看咋回事！")
		else:
			print(sysstr + " RAM Cannot read.")
		printL()

	#調(diào)用報警函數(shù)
	@classmethod
	def send_msg(cls, content):
		cls.send_http(content)

	# 調(diào)用http接口
	@classmethod
	def send_http(cls,content):
		printL()
		print("send_http:",type(content),content)
		url_total = cls.g_php_url + "?msg=" + content
		print("url_total:",url_total)
		rp = requests.get(url_total)
		print("rp:",rp.text)
		printL()

	# 發(fā)微信預(yù)警消息
	@classmethod
	def send_wx_alarm(cls,content):
		post_url = cls.g_wx_url
		for id in cls.g_wx_id:
			try:
				post_data = '{"operSys":"MCS","content":"服務(wù)器監(jiān)控告警：%s\n%s","phones":"%s"}'%(cls.g_web_ip, content, id)
				print(post_data)
				# data = urllib.parse.urlencode(post_data)
				# data = data.encode('utf-8')

				req = requests.get(url=post_url,data=post_data)
				print("send_wx_alarm req:",req,type(req))
				result = json.loads(req.text())
				print(result)
			except Exception as e:
				print("send_wx_alarm:",e)

	# 發(fā)郵件預(yù)警消息
	@classmethod
	def send_email_alarm(cls,content):
		post_url = cls.g_email_url
		for id in cls.g_email_id:
			try:
				post_data = '{"subject":"%s服務(wù)器監(jiān)控告警","email":"%s","bccEmail":"","operSys":"LOG","content":"%s"}'%(cls.g_web_ip, id, content)
				print(post_data)
				# data = urllib.parse.urlencode(post_data)
				# data = data.encode('utf-8')

				req = requests.get(url=post_url,data=post_data)
				print("send_email_alarm req:",req,type(req))
				# req.add_header("Content-Type", "application/x-www-form-urlencoded;charset=utf-8")
				result = json.loads(req.text())
				print(result)
			except Exception as e:
				print("send_email_alarm:",e)

實例代碼配置文件

# 本機IP地址（這里之所以不自動獲取是因為有些機器只有內(nèi)網(wǎng)）
web_ip=***

# 檢測的端口
monitor_ports=3306, 8088, 6004 ,6379

# CPU利用率
cpu_used=100%

# CPU平均負載
aver_load=1

# 內(nèi)存使用率
mem_used=0.8

# 磁盤使用率
disk_used=80%

# 通知地址
php_url=http://***:**/TaskMonitor/action

# 微信地址
wecaht_url=http://***:**/wechat/sendWeChat

# 微信ID
wecaht_id=123456,13123

# email地址
email_url=http://***:**/email/sendEmail

# 郵件郵箱
email_id=test@mucfc.com,11223344@qq.com