pip install psutil watchdog matplotlib requests

import psutil
import time
from datetime import datetime

def monitor_cpu(interval=1, threshold=80, alarm_callback=None):
    """
    CPU監(jiān)控核心函數(shù)
    :param interval: 監(jiān)控間隔（秒）
    :param threshold: 告警閾值（使用率%）
    :param alarm_callback: 告警回調(diào)函數(shù)
    """
    while True:
        # 獲取CPU整體使用率（percpu=True可獲取每個核心的使用率）
        cpu_percent = psutil.cpu_percent(interval=interval, percpu=False)
        # 獲取CPU核心數(shù)、負(fù)載信息（交叉驗證：與top命令結(jié)果一致）
        cpu_count = psutil.cpu_count(logical=True)  # 邏輯核心數(shù)
        load_avg = psutil.getloadavg()  # 1/5/15分鐘負(fù)載
        
        # 構(gòu)造監(jiān)控數(shù)據(jù)
        monitor_data = {
            "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "cpu_percent": cpu_percent,
            "cpu_count": cpu_count,
            "load_avg_1min": load_avg[0],
            "load_avg_5min": load_avg[1]
        }
        print(f"CPU監(jiān)控數(shù)據(jù)：{monitor_data}")
        
        # 閾值判斷，觸發(fā)告警
        if cpu_percent > threshold:
            alarm_msg = f"CPU使用率告警！當(dāng)前使用率：{cpu_percent}%，閾值：{threshold}%，負(fù)載信息：{load_avg}"
            print(f"??  {alarm_msg}")
            if alarm_callback:
                alarm_callback(alarm_msg)
        
        time.sleep(interval)

# 示例：告警回調(diào)函數(shù)（推送至企業(yè)微信/郵件，此處簡化為打?。?
def alarm_push(msg):
    # 實際開發(fā)中可替換為requests.post調(diào)用企業(yè)微信API
    print(f"【告警推送】{msg}")

if __name__ == "__main__":
    # 啟動CPU監(jiān)控：間隔1秒，閾值80%
    monitor_cpu(interval=1, threshold=80, alarm_callback=alarm_push)
    

import psutil
import time
from datetime import datetime

def monitor_memory(interval=5, threshold=85, swap_threshold=70, alarm_callback=None):
    """
    內(nèi)存監(jiān)控核心函數(shù)
    :param interval: 監(jiān)控間隔（秒）
    :param threshold: 內(nèi)存使用率告警閾值（%）
    :param swap_threshold: 交換分區(qū)使用率告警閾值（%）
    :param alarm_callback: 告警回調(diào)函數(shù)
    """
    while True:
        # 獲取內(nèi)存詳細(xì)信息
        mem_info = psutil.virtual_memory()
        # 獲取交換分區(qū)信息
        swap_info = psutil.swap_memory()
        
        # 構(gòu)造監(jiān)控數(shù)據(jù)（單位：GB，保留2位小數(shù)）
        monitor_data = {
            "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "total_mem": round(mem_info.total / 1024**3, 2),
            "used_mem": round(mem_info.used / 1024**3, 2),
            "available_mem": round(mem_info.available / 1024**3, 2),
            "mem_percent": mem_info.percent,
            "swap_total": round(swap_info.total / 1024**3, 2),
            "swap_used": round(swap_info.used / 1024**3, 2),
            "swap_percent": swap_info.percent
        }
        print(f"內(nèi)存監(jiān)控數(shù)據(jù)：{monitor_data}")
        
        # 觸發(fā)告警
        if mem_info.percent > threshold:
            alarm_msg = f"內(nèi)存使用率告警！當(dāng)前使用率：{mem_info.percent}%，可用內(nèi)存：{monitor_data['available_mem']}GB"
            print(f"??  {alarm_msg}")
            if alarm_callback:
                alarm_callback(alarm_msg)
        if swap_info.percent > swap_threshold:
            alarm_msg = f"交換分區(qū)告警！當(dāng)前使用率：{swap_info.percent}%，交換分區(qū)頻繁使用會嚴(yán)重影響性能"
            print(f"??  {alarm_msg}")
            if alarm_callback:
                alarm_callback(alarm_msg)
        
        time.sleep(interval)

if __name__ == "__main__":
    # 啟動內(nèi)存監(jiān)控：間隔5秒，內(nèi)存閾值85%，交換分區(qū)閾值70%
    monitor_memory(interval=5, threshold=85, swap_threshold=70, alarm_callback=alarm_push)
    

import psutil
import time
from datetime import datetime

def monitor_disk(interval=10, threshold=90, alarm_callback=None):
    """
    磁盤監(jiān)控核心函數(shù)
    :param interval: 監(jiān)控間隔（秒）
    :param threshold: 分區(qū)使用率告警閾值（%）
    :param alarm_callback: 告警回調(diào)函數(shù)
    """
    # 獲取需要監(jiān)控的磁盤分區(qū)（過濾虛擬分區(qū)和光驅(qū)）
    partitions = [p.mountpoint for p in psutil.disk_partitions() if 'ext' in p.fstype or 'xfs' in p.fstype]
    
    while True:
        for mountpoint in partitions:
            # 獲取分區(qū)使用率
            disk_usage = psutil.disk_usage(mountpoint)
            # 獲取磁盤IO統(tǒng)計（首次讀取為基準(zhǔn)值，第二次讀取計算差值）
            disk_io = psutil.disk_io_counters(perdisk=True)
            time.sleep(1)  # 間隔1秒計算IO吞吐量
            disk_io_new = psutil.disk_io_counters(perdisk=True)
            
            # 計算IO吞吐量（MB/s）
            io_stats = {}
            for disk, stats in disk_io.items():
                if disk in disk_io_new:
                    read_mb = round((disk_io_new[disk].read_bytes - stats.read_bytes) / 1024**2, 2)
                    write_mb = round((disk_io_new[disk].write_bytes - stats.write_bytes) / 1024**2, 2)
                    io_stats[disk] = {"read_mb_per_sec": read_mb, "write_mb_per_sec": write_mb}
            
            # 構(gòu)造監(jiān)控數(shù)據(jù)
            monitor_data = {
                "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "mountpoint": mountpoint,
                "total_disk": round(disk_usage.total / 1024**3, 2),
                "used_disk": round(disk_usage.used / 1024**3, 2),
                "free_disk": round(disk_usage.free / 1024**3, 2),
                "disk_percent": disk_usage.percent,
                "io_stats": io_stats
            }
            print(f"磁盤監(jiān)控數(shù)據(jù)（{mountpoint}）：{monitor_data}")
            
            # 觸發(fā)告警
            if disk_usage.percent > threshold:
                alarm_msg = f"磁盤分區(qū)告警！{mountpoint}分區(qū)使用率：{disk_usage.percent}%，剩余空間：{monitor_data['free_disk']}GB"
                print(f"??  {alarm_msg}")
                if alarm_callback:
                    alarm_callback(alarm_msg)
        
        time.sleep(interval - 1)  # 減去IO統(tǒng)計的1秒間隔

if __name__ == "__main__":
    # 啟動磁盤監(jiān)控：間隔10秒，閾值90%
    monitor_disk(interval=10, threshold=90, alarm_callback=alarm_push)
    

import psutil
import time
from datetime import datetime
import re

def find_pid_by_name(process_name):
    """根據(jù)進(jìn)程名查找pid（支持模糊匹配）"""
    pids = []
    for proc in psutil.process_iter(['pid', 'name']):
        try:
            if re.search(process_name, proc.info['name']):
                pids.append(proc.info['pid'])
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return pids

def monitor_process(process_name, interval=5, cpu_threshold=50, mem_threshold=20, alarm_callback=None):
    """
    進(jìn)程監(jiān)控核心函數(shù)
    :param process_name: 進(jìn)程名（支持模糊匹配）
    :param interval: 監(jiān)控間隔（秒）
    :param cpu_threshold: 進(jìn)程CPU使用率告警閾值（%）
    :param mem_threshold: 進(jìn)程內(nèi)存使用率告警閾值（%，相對于總內(nèi)存）
    :param alarm_callback: 告警回調(diào)函數(shù)
    """
    while True:
        pids = find_pid_by_name(process_name)
        if not pids:
            alarm_msg = f"進(jìn)程告警！未找到名稱包含{process_name}的進(jìn)程，可能已異常退出"
            print(f"??  {alarm_msg}")
            if alarm_callback:
                alarm_callback(alarm_msg)
                time.sleep(interval)
                continue
        
        total_mem = psutil.virtual_memory().total / 1024**3  # 總內(nèi)存（GB）
        for pid in pids:
            try:
                proc = psutil.Process(pid)
                # 獲取進(jìn)程資源占用
                cpu_percent = proc.cpu_percent(interval=0.5)  # 0.5秒內(nèi)的CPU使用率
                mem_info = proc.memory_full_info()
                mem_used = mem_info.rss / 1024**3  # 進(jìn)程占用內(nèi)存（GB）
                mem_percent = (mem_used / total_mem) * 100  # 進(jìn)程內(nèi)存占比（相對于總內(nèi)存）
                # 獲取進(jìn)程啟動時間、狀態(tài)
                start_time = datetime.fromtimestamp(proc.create_time()).strftime("%Y-%m-%d %H:%M:%S")
                status = proc.status()
                
                # 構(gòu)造監(jiān)控數(shù)據(jù)
                monitor_data = {
                    "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                    "pid": pid,
                    "process_name": proc.name(),
                    "cpu_percent": cpu_percent,
                    "mem_used": round(mem_used, 2),
                    "mem_percent": round(mem_percent, 2),
                    "start_time": start_time,
                    "status": status
                }
                print(f"進(jìn)程監(jiān)控數(shù)據(jù)（pid={pid}）：{monitor_data}")
                
                # 觸發(fā)告警
                if cpu_percent > cpu_threshold:
                    alarm_msg = f"進(jìn)程CPU告警！pid={pid}，進(jìn)程名：{proc.name()}，當(dāng)前CPU使用率：{cpu_percent}%"
                    print(f"??  {alarm_msg}")
                    if alarm_callback:
                        alarm_callback(alarm_msg)
                if mem_percent > mem_threshold:
                    alarm_msg = f"進(jìn)程內(nèi)存告警！pid={pid}，進(jìn)程名：{proc.name()}，當(dāng)前內(nèi)存占比：{mem_percent}%，占用內(nèi)存：{mem_used}GB"
                    print(f"??  {alarm_msg}")
                    if alarm_callback:
                        alarm_callback(alarm_msg)
            except psutil.NoSuchProcess:
                alarm_msg = f"進(jìn)程告警！pid={pid}的進(jìn)程已退出"
                print(f"??  {alarm_msg}")
                if alarm_callback:
                    alarm_callback(alarm_msg)
            except psutil.AccessDenied:
                print(f"無權(quán)限訪問pid={pid}的進(jìn)程信息")
        
        time.sleep(interval - 0.5)  # 減去CPU統(tǒng)計的0.5秒間隔

if __name__ == "__main__":
    # 啟動進(jìn)程監(jiān)控：監(jiān)控名稱包含python的進(jìn)程，CPU閾值50%，內(nèi)存閾值20%
    monitor_process(process_name="python", interval=5, cpu_threshold=50, mem_threshold=20, alarm_callback=alarm_push)
    

import time
from datetime import datetime
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class LogMonitorHandler(FileSystemEventHandler):
    """日志監(jiān)控事件處理器"""
    def __init__(self, log_file, key_words, alarm_callback=None):
        super().__init__()
        self.log_file = log_file  # 要監(jiān)控的日志文件路徑
        self.key_words = key_words  # 要匹配的關(guān)鍵錯誤字（列表）
        self.alarm_callback = alarm_callback
        # 記錄上次讀取的文件位置（避免重復(fù)讀取歷史內(nèi)容）
        self.last_position = 0
    
    def on_modified(self, event):
        """文件被修改時觸發(fā)"""
        if event.src_path != self.log_file:
            return  # 只處理目標(biāo)日志文件
        
        # 讀取新增內(nèi)容
        with open(self.log_file, 'r', encoding='utf-8') as f:
            f.seek(self.last_position)  # 移動到上次讀取的位置
            new_content = f.read()
            self.last_position = f.tell()  # 更新上次讀取位置
        
        if not new_content:
            return
        
        # 匹配關(guān)鍵錯誤字
        for line in new_content.split('\n'):
            if line.strip() == '':
                continue
            for key in self.key_words:
                if key in line:
                    alarm_msg = f"日志錯誤告警！{self.log_file}，時間：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}，錯誤內(nèi)容：{line}"
                    print(f"??  {alarm_msg}")
                    if self.alarm_callback:
                        self.alarm_callback(alarm_msg)
                    break

def monitor_log(log_file, key_words, alarm_callback=None):
    """
    日志監(jiān)控核心函數(shù)
    :param log_file: 日志文件路徑
    :param key_words: 關(guān)鍵錯誤字列表（如['ERROR', 'Exception', '警告']）
    :param alarm_callback: 告警回調(diào)函數(shù)
    """
    # 初始化事件處理器
    event_handler = LogMonitorHandler(log_file, key_words, alarm_callback)
    # 初始化觀察者
    observer = Observer()
    # 監(jiān)聽日志文件所在目錄（只監(jiān)控目標(biāo)文件的修改）
    observer.schedule(event_handler, path='/'.join(log_file.split('/')[:-1]), recursive=False)
    # 啟動觀察者
    observer.start()
    print(f"日志監(jiān)控已啟動，監(jiān)控文件：{log_file}，關(guān)鍵錯誤字：{key_words}")
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        # 停止觀察者
        observer.stop()
    observer.join()

if __name__ == "__main__":
    # 啟動日志監(jiān)控：監(jiān)控test.log文件，匹配ERROR、Exception關(guān)鍵字
    monitor_log(
        log_file="/var/log/test.log",
        key_words=["ERROR", "Exception", "警告"],
        alarm_callback=alarm_push
    )
    

### 1. 客戶端：多模塊整合+數(shù)據(jù)推送
import psutil
import time
import requests
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor

# 服務(wù)端接口地址
SERVER_URL = "http://192.168.1.100:5000/monitor/data"
# 本服務(wù)器標(biāo)識（用于服務(wù)端區(qū)分）
SERVER_ID = "order-server-01"
# 要監(jiān)控的核心進(jìn)程
CORE_PROCESSES = ["user-service", "order-service", "pay-service"]

def collect_cpu_data():
    """采集CPU數(shù)據(jù)"""
    cpu_percent = psutil.cpu_percent(interval=0.5, percpu=False)
    load_avg = psutil.getloadavg()
    return {
        "type": "cpu",
        "server_id": SERVER_ID,
        "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "cpu_percent": cpu_percent,
        "load_avg_1min": load_avg[0],
        "load_avg_5min": load_avg[1]
    }

def collect_mem_data():
    """采集內(nèi)存數(shù)據(jù)"""
    mem_info = psutil.virtual_memory()
    swap_info = psutil.swap_memory()
    return {
        "type": "memory",
        "server_id": SERVER_ID,
        "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "mem_percent": mem_info.percent,
        "available_mem": round(mem_info.available / 1024**3, 2),
        "swap_percent": swap_info.percent
    }

def collect_process_data():
    """采集核心進(jìn)程數(shù)據(jù)"""
    process_data = []
    total_mem = psutil.virtual_memory().total / 1024**3
    for proc_name in CORE_PROCESSES:
        pids = [p.pid for p in psutil.process_iter(['pid', 'name']) if proc_name in p.info['name']]
        if not pids:
            process_data.append({
                "type": "process",
                "server_id": SERVER_ID,
                "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                "process_name": proc_name,
                "status": "not_running",
                "cpu_percent": 0,
                "mem_percent": 0
            })
            continue
        for pid in pids:
            try:
                proc = psutil.Process(pid)
                cpu_percent = proc.cpu_percent(interval=0.1)
                mem_used = proc.memory_full_info().rss / 1024**3
                mem_percent = (mem_used / total_mem) * 100
                process_data.append({
                    "type": "process",
                    "server_id": SERVER_ID,
                    "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                    "process_name": proc_name,
                    "pid": pid,
                    "status": proc.status(),
                    "cpu_percent": cpu_percent,
                    "mem_percent": round(mem_percent, 2)
                })
            except psutil.NoSuchProcess:
                process_data.append({
                    "type": "process",
                    "server_id": SERVER_ID,
                    "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                    "process_name": proc_name,
                    "status": "exited",
                    "cpu_percent": 0,
                    "mem_percent": 0
                })
    return process_data

def push_data(data):
    """推送數(shù)據(jù)到服務(wù)端"""
    try:
        response = requests.post(SERVER_URL, json=data, timeout=5)
        if response.status_code == 200:
            print(f"數(shù)據(jù)推送成功：{data}")
        else:
            print(f"數(shù)據(jù)推送失敗，狀態(tài)碼：{response.status_code}，數(shù)據(jù)：{data}")
    except Exception as e:
        print(f"數(shù)據(jù)推送異常：{str(e)}，數(shù)據(jù)：{data}")

def main():
    """客戶端主函數(shù)：多線程并行采集數(shù)據(jù)"""
    with ThreadPoolExecutor(max_workers=3) as executor:
        while True:
            # 提交采集任務(wù)
            future_cpu = executor.submit(collect_cpu_data)
            future_mem = executor.submit(collect_mem_data)
            future_proc = executor.submit(collect_process_data)
            
            # 獲取采集結(jié)果并推送
            cpu_data = future_cpu.result()
            push_data(cpu_data)
            
            mem_data = future_mem.result()
            push_data(mem_data)
            
            proc_data_list = future_proc.result()
            for proc_data in proc_data_list:
                push_data(proc_data)
            
            time.sleep(5)  # 每5秒采集一次

if __name__ == "__main__":
    main()

### 2. 服務(wù)端：Flask接收數(shù)據(jù)+存儲（簡化版）
from flask import Flask, request, jsonify
import pymysql
from datetime import datetime

app = Flask(__name__)

# 數(shù)據(jù)庫配置
DB_CONFIG = {
    "host": "localhost",
    "user": "root",
    "password": "123456",
    "database": "monitor_db"
}

def insert_db(data):
    """插入數(shù)據(jù)到數(shù)據(jù)庫"""
    conn = pymysql.connect(**DB_CONFIG)
    cursor = conn.cursor()
    try:
        if data["type"] == "cpu":
            sql = """INSERT INTO cpu_monitor (server_id, monitor_time, cpu_percent, load_avg_1min, load_avg_5min)
                     VALUES (%s, %s, %s, %s, %s)"""
            cursor.execute(sql, (data["server_id"], data["time"], data["cpu_percent"], data["load_avg_1min"], data["load_avg_5min"]))
        elif data["type"] == "memory":
            sql = """INSERT INTO mem_monitor (server_id, monitor_time, mem_percent, available_mem, swap_percent)
                     VALUES (%s, %s, %s, %s, %s)"""
            cursor.execute(sql, (data["server_id"], data["time"], data["mem_percent"], data["available_mem"], data["swap_percent"]))
        elif data["type"] == "process":
            sql = """INSERT INTO process_monitor (server_id, monitor_time, process_name, pid, status, cpu_percent, mem_percent)
                     VALUES (%s, %s, %s, %s, %s, %s, %s)"""
            cursor.execute(sql, (data["server_id"], data["time"], data["process_name"], data.get("pid", 0), data["status"], data["cpu_percent"], data["mem_percent"]))
        conn.commit()
    except Exception as e:
        print(f"數(shù)據(jù)庫插入異常：{str(e)}")
        conn.rollback()
    finally:
        cursor.close()
        conn.close()

@app.route("/monitor/data", methods=["POST"])
def receive_data():
    """接收客戶端推送的監(jiān)控數(shù)據(jù)"""
    data = request.get_json()
    if not data:
        return jsonify({"code": 400, "msg": "無效數(shù)據(jù)"}), 400
    # 插入數(shù)據(jù)庫
    insert_db(data)
    # 這里可添加告警判斷邏輯（如超過閾值觸發(fā)告警）
    return jsonify({"code": 200, "msg": "success"}), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=False)