Robot 監(jiān)控服務(wù)器資源

文章目錄

• Robot 監(jiān)控服務(wù)器資源
• • 一 psutils 監(jiān)控資源信息
  • 二 數(shù)據(jù)可視化
  • • 1. deque雙端隊(duì)列的應(yīng)用
    • 2. 平滑曲線
    • 3. 圖像數(shù)據(jù)流轉(zhuǎn)換與加密
    • 4. 給不同級別的數(shù)據(jù)指定不同的顏色
  • 三 企業(yè)微信機(jī)器人推送
  • • 1. 配置機(jī)器人
    • 2. 推送信息編輯
  • 四 腳本完整代碼

python腳本監(jiān)控服務(wù)器的cpu， memory， swap等資源信息，可視化數(shù)據(jù)
通過微信機(jī)器人推送到企業(yè)微信，進(jìn)行日報(bào)統(tǒng)計(jì)或性能預(yù)警

對于此監(jiān)控腳本，我的設(shè)想是每隔10秒去獲取一次系統(tǒng)信息，每隔6小時(shí)統(tǒng)計(jì)一次6小時(shí)內(nèi)的歷史數(shù)據(jù)，并可視化6小時(shí)歷史監(jiān)控信息。 如果服務(wù)器超出預(yù)警閾值，可視化5分鐘內(nèi)的歷史監(jiān)控信息。以下是腳本的配置信息：

# ------------ config ------------- host_name ='Semacare-uat' # 服務(wù)器名 warning_level = 30 # 預(yù)警閾值 interal_time = 10 # 監(jiān)測間隔時(shí)間 warning_time = 5 * 60 # 5 minute 5分鐘預(yù)警一次 note_time = 6 * 3600 # 6 hour匯報(bào)一次服務(wù)器資源信息

一 psutils 監(jiān)控資源信息

關(guān)于python是如何監(jiān)控服務(wù)器cpu，memory等信息的，可查看一下幾篇文章：

linux資源監(jiān)控——計(jì)算CPU利用率

linux資源監(jiān)控——獲取GPU信息

linux資源監(jiān)控——獲取Memory與Swap的使用率

這次我們將使用一個(gè)python的第三方庫：psutil 來獲取服務(wù)器的各種信息。

psutil(process and system utilities)是一個(gè)跨平臺庫， 用于在Python中檢索有關(guān)運(yùn)行進(jìn)程和系統(tǒng)利用率(CPU、內(nèi)存、磁盤、網(wǎng)絡(luò)、傳感器)的信息。

它主要用于系統(tǒng)監(jiān)視、分析和限制進(jìn)程資源以及管理正在運(yùn)行的進(jìn)程。它實(shí)現(xiàn)了ps、top、iotop、lsof、netstat、ifconfig、free等經(jīng)典UNIX命令行工具所提供的許多功能。 psutil使用非常簡單，具體教程可參考此鏈接：https://pypi.org/project/psutil/

以下代碼是使用psutil獲取cpu， memory， swap：

# 受用psutils包 獲取cpu， memory信息 def get_CpuInfo():try:cpu_percent = psutil.cpu_percent(0.1)except:cpu_percent = -1print('cpu獲取失敗')return cpu_percentdef get_MemoryInfo():try:mem_percent = psutil.virtual_memory().percentswap_percent = psutil.swap_memory().percentexcept:mem_percent, swap_percent = -1, -1print('獲取內(nèi)存信息失敗')return mem_percent, swap_percent

二 數(shù)據(jù)可視化

1. deque雙端隊(duì)列的應(yīng)用

我們需要這樣一種數(shù)據(jù)結(jié)構(gòu)：長度大小固定，可以不停的壓入數(shù)據(jù)，當(dāng)長度超出設(shè)定值時(shí)會丟棄最老的數(shù)據(jù)，壓入最新的數(shù)據(jù)。

python的雙端隊(duì)列deque可以滿足要求。

from collections import dequedq = deque(maxlen=5) # 設(shè)定最大長度為5 for i in range(5):dq.append(i)dq Out[5]: deque([0, 1, 2, 3, 4])dq.append('a') dq Out[7]: deque([1, 2, 3, 4, 'a']) # 壓入'a'，最老的0會被丟棄。

2. 平滑曲線

在繪制5分鐘的預(yù)警信息圖時(shí)，大概只有30個(gè)點(diǎn)的數(shù)據(jù)，繪制出來的是折線圖，不美觀。如下圖

以下是插值法平滑曲線的代碼：

# 插值法平滑曲線 new_x = np.linspace(min(x_list), max(x_list), 300) cpu_smooth = make_interp_spline(x_list, cpu_list,)(new_x) mem_smooth = make_interp_spline(x_list, mem_list)(new_x)

可以看到折線圖被平滑成了曲線。

3. 圖像數(shù)據(jù)流轉(zhuǎn)換與加密

如果要將繪制的圖片通過微信機(jī)器人推送，就必須把圖片進(jìn)行md5加密，并且轉(zhuǎn)化成base64編碼的數(shù)據(jù)。一般來說我們需要先將圖片保存到磁盤，然后在讀文件，轉(zhuǎn)化成md5值和base64值。這樣我們需要給需要保存的圖片開辟臨時(shí)的磁盤空間，之后我們還需要在刪除圖片，維護(hù)起來比較麻煩。

更好的選擇是，使用io.BytesIO() 直接跳過磁盤io，將圖片轉(zhuǎn)成數(shù)據(jù)流，保存在內(nèi)存里，然在進(jìn)行md5加密和base64轉(zhuǎn)換。代碼如下：

# 圖片md5加密， 轉(zhuǎn)base64 def img_encryption(plt):pic_IObytes = io.BytesIO()plt.savefig(pic_IObytes, format='png')pic_IObytes.seek(0)content = pic_IObytes.read()pic_md5 = hashlib.md5(content).hexdigest() # md5pic_hash = base64.b64encode(content).decode() # base64return pic_md5, pic_hash

4. 給不同級別的數(shù)據(jù)指定不同的顏色

例如 cpu 性能在0 ～ 30%時(shí) 我們使用藍(lán)色繪制，30% ～ 80% 時(shí)使用橘黃色繪制， 超過80% 的使用紅色繪制。

實(shí)現(xiàn)很簡單，實(shí)現(xiàn)一個(gè)數(shù)據(jù)與顏色對應(yīng)的映射表。代碼如下：

cpu_colors = [] for i in cpu_list:if i >= 80:cpu_colors.append('red')elif 30 < i <= 80:cpu_colors.append('orange')else:cpu_colors.append('blue')mem_colors = []for i in mem_list:if i >= 80:mem_colors.append('red')elif 30 < i <= 80:mem_colors.append('orange')else:mem_colors.append('blue')plt.sca(ax1) plt.ylim(0, 100) plt.vlines(range(len(cpu_list)), 0, cpu_list, colors=cpu_colors) # 指定顏色映射plt.sca(ax2) plt.ylim(0, 100) plt.vlines(range(len(mem_list)), 0, mem_list, colors=mem_colors)

下圖是實(shí)現(xiàn)效果：

三 企業(yè)微信機(jī)器人推送

1. 配置機(jī)器人

添加一個(gè)群聊機(jī)器人，右鍵選擇的群組，點(diǎn)擊添加群機(jī)器人，即可添加一個(gè)群機(jī)器人。

點(diǎn)擊機(jī)器人，查看機(jī)器人信息。

點(diǎn)擊Webhook，可以查看機(jī)器人的各種配置，以及消息推送的不同方式及模版。

2. 推送信息編輯

本腳本使用的是markdown推送文本信息，使用圖片格式推送繪圖信息。

以下是markdown推送的模版：

代碼如下：

def post_info(cpu, memory, swap, host, type='warning'):webhook_url = 'https://qyapi.weixin.qq.com/xxxxeqwrfwetgerqw54361fsdagfas23435xxx'colors = []for i in [cpu, memory, swap]:if i <= 30:colors.append('info')elif i < 80:colors.append('warning')else:colors.append('red')warning_info = f'服務(wù)器:**<font color=\"blue\">{host}</font>**性能預(yù)警報(bào)告：\n' \f'>CPU: <font color=\"{colors[0]}\">{cpu}</font> % \n' \f'>MEM: <font color=\"{colors[1]}\">{memory}</font> % \n' \f'>SWP: <font color=\"{colors[2]}\">{swap}</font> % \n\n' \f'--------\n' \f'<font color=\"#FF99CC">粉粉</font>會持續(xù)為主人監(jiān)測服務(wù)器的性能～'note_info = f'服務(wù)器:**<font color=\"blue\">{host}</font>**監(jiān)測時(shí)報(bào)：\n' \f'>CPU: <font color=\"{colors[0]}\">{cpu}</font> % \n' \f'>MEM: <font color=\"{colors[1]}\">{memory}</font> % \n' \f'>SWP: <font color=\"{colors[2]}\">{swap}</font> % \n\n' \f'--------\n' \f'<font color=\"#FF99CC">粉粉</font>會持續(xù)為主人監(jiān)測服務(wù)器的性能～'content = {'msgtype': 'markdown','markdown': {'content': warning_info if type == 'warning' else note_info},}r = requests.post(webhook_url, json.dumps(content))print(f'{type} event post status code: {r.status_code}')

以下是圖片推送的模版：

圖片推送的代碼如下：

def post_img(pic_md5, pic_hash):if not pic_md5:returnwebhook_url = 'https://qyapi.weixxxxxxxxxx234235dfgdfsgdxxxxxxxx'content = {'msgtype': 'image','image': {'base64': pic_hash,'md5': pic_md5}}r = requests.post(webhook_url, json.dumps(content))

下面是推送的真實(shí)效果：

四 腳本完整代碼

# -*- coding: utf-8 -*- """ -------------------------------------------------File Name： reboot_push_messageDescription : 企業(yè)微信機(jī)器人，推送服務(wù)器資源：cpu， memory， swap， disk預(yù)警信息到企業(yè)微信群。Author : Gray-rpldate： 2020/12/18 -------------------------------------------------Change Activity:2020/12/18: v12020/12/25: v2 添加CPU，Mem的小時(shí)統(tǒng)計(jì)圖， 分鐘統(tǒng)計(jì)圖，并推送圖片。使用psutil獲取系統(tǒng)信息 ------------------------------------------------- """ import os import io import time import json import base64 import hashlib import psutil import requestsfrom threading import Timer from collections import deque import matplotlib.pyplot as pltimport numpy as np from scipy.interpolate import make_interp_spline# ------------ config ------------- host_name ='Semacare-uat' # 服務(wù)器名 warning_level = 30 # 預(yù)警閾值 interal_time = 10 # 監(jiān)測間隔時(shí)間 warning_time = 5 * 60 # 5 minute 5分鐘預(yù)警一次 note_time = 2 * 3600 # 2 hour匯報(bào)一次服務(wù)器資源信息 # --------------------------------pre_warning_time = 0 pre_note_time = 0cpu_dq = deque(maxlen=int(note_time // interal_time)) mem_dq = deque(maxlen=int(note_time // interal_time))def post_info(cpu, memory, swap, host, type='warning'):webhook_url = 'https://qyapi.weixin.qq.com/cgi-bin/webgsdfgw543623dsf123233254'colors = []for i in [cpu, memory, swap]:if i <= 30:colors.append('info')elif i < 80:colors.append('warning')else:colors.append('red')warning_info = f'服務(wù)器:**<font color=\"blue\">{host}</font>**性能預(yù)警報(bào)告：\n' \f'>CPU: <font color=\"{colors[0]}\">{cpu}</font> % \n' \f'>MEM: <font color=\"{colors[1]}\">{memory}</font> % \n' \f'>SWP: <font color=\"{colors[2]}\">{swap}</font> % \n\n' \f'--------\n' \f'<font color=\"#FF99CC">粉粉</font>會持續(xù)為主人監(jiān)測服務(wù)器的性能～'note_info = f'服務(wù)器:**<font color=\"blue\">{host}</font>**監(jiān)測時(shí)報(bào)：\n' \f'>CPU: <font color=\"{colors[0]}\">{cpu}</font> % \n' \f'>MEM: <font color=\"{colors[1]}\">{memory}</font> % \n' \f'>SWP: <font color=\"{colors[2]}\">{swap}</font> % \n\n' \f'--------\n' \f'<font color=\"#FF99CC">粉粉</font>會持續(xù)為主人監(jiān)測服務(wù)器的性能～'content = {'msgtype': 'markdown','markdown': {'content': warning_info if type == 'warning' else note_info},}r = requests.post(webhook_url, json.dumps(content))print(f'{type} event post status code: {r.status_code}')def post_img(pic_md5, pic_hash):if not pic_md5:returnwebhook_url = 'https://qyapi.weixin.frwqerqwr234264324123qesfsadcwxxxxxxx'content = {'msgtype': 'image','image': {'base64': pic_hash,'md5': pic_md5}}r = requests.post(webhook_url, json.dumps(content))# 獲取cpu信息 def get_CpuInfo():total_list = []idle_list = []# 連續(xù)取2次文件for k in range(2):data = []# 這里是使用ssh命令獲， 去服務(wù)器上獲取信息# grep -w cpu 只過濾含cpu字段的行，剛好也就是我們需要的那一行信息cpu_cmd = 'cat /proc/stat |grep -w cpu'res = os.popen(cpu_cmd, ).read().split()if not res:breakfor i in res:try:if isinstance(eval(i), int):data.append(i)except:continuetime.sleep(0.01)# print('cpu：' + str(data))total_cpu_time = sum([int(i) for i in data])total_list.append(total_cpu_time)idle_list.append(int(data[3]))if len(total_list) == 2:total = total_list[1] - total_list[0]idle = idle_list[1] - idle_list[0]pcpu = str(round(100 * (total - idle) / total, 2))else:print('%s:獲取cpu信息失敗')pcpu = '-1'return float(pcpu)# 獲取內(nèi)存信息 def get_MemoryInfo():timeout_seconds = 30cmd = 'free'# res = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)memory_utilization = -1swap_utilization = -1res = os.popen(cmd,)if res:stdout = res.readlines()if not stdout:print('獲取memory信息失敗')return [memory_utilization, swap_utilization]memory_info = stdout[1].split()swap_info = stdout[2].split()totoal_memory = int(memory_info[1])available_memory = int(memory_info[-1])# used_memory = float(memory_info[2])# memory_utilization = str(round((used_memory / totoal_memory) *100, 2))memory_utilization = ((totoal_memory-available_memory)/totoal_memory)*100memory_utilization = str(round(memory_utilization, 2))totoal_swap = int(swap_info[1])if totoal_swap == 0:print(',交換空間為0')swap_utilization = 0else:available_swap = int(swap_info[-1])swap_utilization = ((totoal_swap - available_swap) / totoal_swap) * 100swap_utilization = round(swap_utilization, 2)else:print('timeout > {}s, 獲取memory信息失敗'.format(timeout_seconds))return [float(memory_utilization), float(swap_utilization)]# 受用psutils包 獲取cpu， memory信息 def get_CpuInfo2():try:cpu_percent = psutil.cpu_percent(0.1)except:cpu_percent = -1print('cpu獲取失敗')return cpu_percentdef get_MemoryInfo2():try:mem_percent = psutil.virtual_memory().percentswap_percent = psutil.swap_memory().percentexcept:mem_percent, swap_percent = -1, -1print('獲取內(nèi)存信息失敗')return mem_percent, swap_percent# 繪制每6小時(shí)一次的資源通知圖 def draw_note_figure():# note_count = note_time // interal_timecpu_list = list(cpu_dq)mem_list = list(mem_dq)if len(cpu_list) < 30:print('Note statistic not enough data, drop img...')return (None, None)# import random# cpu_list = [random.randrange(0, 101) for i in range(2160)]# mem_list = [random.randrange(0, 101) for i in range(2160)]cpu_colors = []for i in cpu_list:if i >= 80:cpu_colors.append('red')elif 30 < i <= 80:cpu_colors.append('orange')else:cpu_colors.append('blue')mem_colors = []for i in mem_list:if i >= 80:mem_colors.append('red')elif 30 < i <= 80:mem_colors.append('orange')else:mem_colors.append('blue')plt.figure(figsize=(9, 4))ax1 = plt.subplot(2, 1, 1)ax2 = plt.subplot(2, 1, 2)# 添加子圖的標(biāo)題信息ax1.set_title(f'{host_name}: {note_time // 3600} hours CPU statistic')ax2.set_title(f'{host_name}: {note_time // 3600} hours Memory statistic')# 設(shè)置背景網(wǎng)格類型，顏色，透明度ax1.grid(color='#FF0000', linestyle='-.', linewidth=1, alpha=0.2)ax2.grid(color='#FF0000', linestyle='-.', linewidth=1, alpha=0.2)plt.sca(ax1) # 選擇子圖1plt.ylim(0, 100) # 限制y軸高度plt.vlines(range(len(cpu_list)), 0, cpu_list, colors=cpu_colors) # 指定顏色映射plt.sca(ax2) # 選擇子圖2plt.ylim(0, 100)plt.vlines(range(len(mem_list)), 0, mem_list, colors=mem_colors)plt.tight_layout() # 嚴(yán)謹(jǐn)布局# 將圖片直接轉(zhuǎn)base64， 跳過磁盤IOreturn img_encryption(plt)# 圖片md5加密， 轉(zhuǎn)base64 def img_encryption(plt):pic_IObytes = io.BytesIO()plt.savefig(pic_IObytes, format='png')pic_IObytes.seek(0)content = pic_IObytes.read()pic_md5 = hashlib.md5(content).hexdigest() # md5pic_hash = base64.b64encode(content).decode() # base64return pic_md5, pic_hash# 繪制5分鐘的預(yù)警信息圖 def draw_warning_figure():count = warning_time // interal_timecpu_list = list(cpu_dq)[-count:]mem_list = list(mem_dq)[-len(cpu_list):]if len(cpu_list) < 5:return (None, None)import randomcpu_list = [random.randrange(30, 101) for i in range(30)]mem_list = [random.randrange(0, 30) for i in range(30)]x_list = list(range(len(cpu_list)))# 插值法平滑曲線new_x = np.linspace(min(x_list), max(x_list), 300)cpu_smooth = make_interp_spline(x_list, cpu_list,)(new_x)mem_smooth = make_interp_spline(x_list, mem_list)(new_x)plt.figure(figsize=(9, 3))plt.plot(new_x, cpu_smooth,)plt.plot(new_x, mem_smooth,)# plt.plot(x_list, cpu_list, mem_list)plt.ylim(0, 100)plt.title(f'{host_name}: {warning_time // 60} minutes statistic')plt.legend(['Cpu', 'Memory'])plt.grid(color='#FF0000', linestyle='-.', linewidth=1, alpha=0.1)# ax = plt.gca()# 去頂部邊框# ax.spines['top'].set_color('none')# ax.spines['right'].set_color('none')# 坐標(biāo)原點(diǎn)相交# ax.xaxis.set_ticks_position('bottom')# ax.spines['bottom'].set_position(('data', 0))# ax.yaxis.set_ticks_position('left')# ax.spines['left'].set_position(('data', 0))return img_encryption(plt)# 監(jiān)控資源 def get_server_info():global pre_note_timeglobal pre_warning_timecpu = get_CpuInfo2()mem, swp = get_MemoryInfo2()global cpu_dq, mem_dqcpu_dq.append(cpu)mem_dq.append(mem)cpu_avg = sum(list(cpu_dq)[-5:]) // len(list(cpu_dq)[-5:])mem_avg = sum(list(mem_dq)[-5:]) // len(list(mem_dq)[-5:])print(f'cpu: {cpu} %, mem: {mem} %, swap: {swp} %')if cpu_avg >= warning_level or mem_avg >= warning_level:if time.time() - pre_warning_time >= warning_time:img_md5, img_hash = draw_warning_figure()post_info(cpu, mem, swp, host_name, )post_img(img_md5, img_hash)pre_warning_time = time.time()if time.time() - pre_note_time >= note_time:img_md5, img_hash = draw_note_figure()post_info(cpu, mem, swp, host_name, type='note')post_img(img_md5, img_hash)pre_note_time = time.time()print()def main():while True:get_server_info()time.sleep(interal_time)if __name__ == "__main__":main()

總結(jié)

以上是生活随笔為你收集整理的Robot 监控服务器资源的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。