Robot 監控服務器資源

文章目錄

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

python腳本監控服務器的cpu， memory， swap等資源信息，可視化數據
通過微信機器人推送到企業微信，進行日報統計或性能預警

對于此監控腳本，我的設想是每隔10秒去獲取一次系統信息，每隔6小時統計一次6小時內的歷史數據，并可視化6小時歷史監控信息。 如果服務器超出預警閾值，可視化5分鐘內的歷史監控信息。以下是腳本的配置信息：

# ------------ config ------------- host_name ='Semacare-uat' # 服務器名 warning_level = 30 # 預警閾值 interal_time = 10 # 監測間隔時間 warning_time = 5 * 60 # 5 minute 5分鐘預警一次 note_time = 6 * 3600 # 6 hour匯報一次服務器資源信息

一 psutils 監控資源信息

關于python是如何監控服務器cpu，memory等信息的，可查看一下幾篇文章：

linux資源監控——計算CPU利用率

linux資源監控——獲取GPU信息

linux資源監控——獲取Memory與Swap的使用率

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

psutil(process and system utilities)是一個跨平臺庫， 用于在Python中檢索有關運行進程和系統利用率(CPU、內存、磁盤、網絡、傳感器)的信息。

它主要用于系統監視、分析和限制進程資源以及管理正在運行的進程。它實現了ps、top、iotop、lsof、netstat、ifconfig、free等經典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('獲取內存信息失敗')return mem_percent, swap_percent

二 數據可視化

1. deque雙端隊列的應用

我們需要這樣一種數據結構：長度大小固定，可以不停的壓入數據，當長度超出設定值時會丟棄最老的數據，壓入最新的數據。

python的雙端隊列deque可以滿足要求。

from collections import dequedq = deque(maxlen=5) # 設定最大長度為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分鐘的預警信息圖時，大概只有30個點的數據，繪制出來的是折線圖，不美觀。如下圖

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

# 插值法平滑曲線 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. 圖像數據流轉換與加密

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

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

# 圖片md5加密， 轉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. 給不同級別的數據指定不同的顏色

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

實現很簡單，實現一個數據與顏色對應的映射表。代碼如下：

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)

下圖是實現效果：

三 企業微信機器人推送

1. 配置機器人

添加一個群聊機器人，右鍵選擇的群組，點擊添加群機器人，即可添加一個群機器人。

點擊機器人，查看機器人信息。

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

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'服務器:**<font color=\"blue\">{host}</font>**性能預警報告：\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>會持續為主人監測服務器的性能～'note_info = f'服務器:**<font color=\"blue\">{host}</font>**監測時報：\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>會持續為主人監測服務器的性能～'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))

下面是推送的真實效果：

四 腳本完整代碼

# -*- coding: utf-8 -*- """ -------------------------------------------------File Name： reboot_push_messageDescription : 企業微信機器人，推送服務器資源：cpu， memory， swap， disk預警信息到企業微信群。Author : Gray-rpldate： 2020/12/18 -------------------------------------------------Change Activity:2020/12/18: v12020/12/25: v2 添加CPU，Mem的小時統計圖， 分鐘統計圖，并推送圖片。使用psutil獲取系統信息 ------------------------------------------------- """ 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' # 服務器名 warning_level = 30 # 預警閾值 interal_time = 10 # 監測間隔時間 warning_time = 5 * 60 # 5 minute 5分鐘預警一次 note_time = 2 * 3600 # 2 hour匯報一次服務器資源信息 # --------------------------------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'服務器:**<font color=\"blue\">{host}</font>**性能預警報告：\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>會持續為主人監測服務器的性能～'note_info = f'服務器:**<font color=\"blue\">{host}</font>**監測時報：\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>會持續為主人監測服務器的性能～'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 = []# 連續取2次文件for k in range(2):data = []# 這里是使用ssh命令獲， 去服務器上獲取信息# 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)# 獲取內存信息 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('獲取內存信息失敗')return mem_percent, swap_percent# 繪制每6小時一次的資源通知圖 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)# 添加子圖的標題信息ax1.set_title(f'{host_name}: {note_time // 3600} hours CPU statistic')ax2.set_title(f'{host_name}: {note_time // 3600} hours Memory statistic')# 設置背景網格類型，顏色，透明度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() # 嚴謹布局# 將圖片直接轉base64， 跳過磁盤IOreturn img_encryption(plt)# 圖片md5加密， 轉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分鐘的預警信息圖 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')# 坐標原點相交# 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)# 監控資源 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()

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。