用脚本监控网站是否宕机

wen 实用脚本 2

本文目录导读:

用脚本监控网站是否宕机

  1. 基础版 - 单URL监控脚本
  2. 进阶版 - 多网站监控脚本
  3. 极简版 - 单文件监控脚本
  4. 使用 Docker 部署监控服务
  5. 使用方法
  6. 告警功能集成
  7. 监控优化建议

基础版 - 单URL监控脚本

#!/bin/bash
# website_monitor.sh - 简单网站监控脚本
# 配置
URL="https://example.com"
TIMEOUT=10
CHECK_INTERVAL=60  # 秒
LOG_FILE="/var/log/website_monitor.log"
# 颜色输出
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color
# 记录日志函数
log_message() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" >> "$LOG_FILE"
    echo -e "$1"
}
# 监控函数
check_website() {
    local url=$1
    local timeout=$2
    # 使用curl检查网站状态
    local http_code=$(curl -s -o /dev/null -w "%{http_code}" \
                      --connect-timeout "$timeout" \
                      --max-time "$timeout" \
                      "$url" 2>/dev/null)
    local exit_code=$?
    if [ $exit_code -ne 0 ]; then
        log_message "${RED}[ERROR]${NC} $url - 无法连接 (curl exit code: $exit_code)"
        return 1
    fi
    if [[ "$http_code" =~ ^[23][0-9][0-9]$ ]]; then
        log_message "${GREEN}[OK]${NC} $url - HTTP $http_code"
        return 0
    else
        log_message "${RED}[ERROR]${NC} $url - HTTP $http_code"
        return 1
    fi
}
# 主循环
echo "开始监控 $URL (每 $CHECK_INTERVAL 秒检查一次)"
log_message "${YELLOW}[START]${NC} 开始监控 $URL"
while true; do
    if ! check_website "$URL" "$TIMEOUT"; then
        # 网站宕机,尝试多次确认
        fail_count=0
        for i in {1..3}; do
            sleep 5
            if ! check_website "$URL" "$TIMEOUT"; then
                ((fail_count++))
            fi
        done
        if [ $fail_count -ge 3 ]; then
            log_message "${RED}[ALERT]${NC} 网站 $URL 确认宕机!"
            # 这里可以添加发送邮件、短信等告警操作
        fi
    fi
    sleep "$CHECK_INTERVAL"
done

进阶版 - 多网站监控脚本

#!/usr/bin/env python3
# multi_website_monitor.py - 多网站监控脚本
import requests
import time
import logging
import json
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
import smtplib
from email.mime.text import MIMEText
from email.header import Header
import os
class WebsiteMonitor:
    def __init__(self, config_file='websites.json'):
        self.config_file = config_file
        self.websites = self.load_config()
        self.setup_logging()
    def setup_logging(self):
        """设置日志记录"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('website_monitor.log'),
                logging.StreamHandler()
            ]
        )
    def load_config(self):
        """加载网站配置"""
        with open(self.config_file, 'r', encoding='utf-8') as f:
            return json.load(f)
    def check_website(self, website):
        """检查单个网站状态"""
        url = website['url']
        timeout = website.get('timeout', 10)
        expected_keyword = website.get('expected_keyword', '')
        try:
            # 发送HTTP请求
            response = requests.get(
                url, 
                timeout=timeout,
                headers={'User-Agent': 'Website-Monitor/1.0'}
            )
            # 检查响应时间
            response_time = response.elapsed.total_seconds()
            # 检查状态码
            if response.status_code == 200 or 200 <= response.status_code < 300:
                # 检查关键词是否存在于页面内容中
                if expected_keyword and expected_keyword not in response.text:
                    logging.warning(f"{url} - 状态码正常但关键词'{expected_keyword}'未找到")
                    return False
                logging.info(f"{url} - 运行正常 (状态码: {response.status_code}, 响应时间: {response_time:.2f}s)")
                return True
            else:
                logging.warning(f"{url} - 异常状态码: {response.status_code}")
                return False
        except requests.exceptions.ConnectionError:
            logging.error(f"{url} - 连接失败")
            return False
        except requests.exceptions.Timeout:
            logging.error(f"{url} - 请求超时")
            return False
        except Exception as e:
            logging.error(f"{url} - 发生错误: {str(e)}")
            return False
    def check_all_websites(self):
        """并发检查所有网站"""
        results = {}
        failures = []
        with ThreadPoolExecutor(max_workers=5) as executor:
            futures = {executor.submit(self.check_website, site): site for site in self.websites}
            for future in futures:
                site = futures[future]
                try:
                    result = future.result()
                    results[site['url']] = result
                    if not result:
                        failures.append(site)
                except Exception as e:
                    logging.error(f"检查 {site['url']} 时发生错误: {e}")
                    failures.append(site)
        if failures:
            self.send_alert_email(failures)
        return results
    def send_alert_email(self, failures):
        """发送告警邮件"""
        # 邮件配置
        smtp_server = os.getenv('SMTP_SERVER', 'smtp.gmail.com')
        smtp_port = int(os.getenv('SMTP_PORT', 587))
        sender = os.getenv('ALERT_EMAIL')
        password = os.getenv('ALERT_EMAIL_PASSWORD')
        recipient = os.getenv('ALERT_RECIPIENT')
        if not all([sender, password, recipient]):
            logging.warning("未配置邮件告警参数,跳过邮件通知")
            return
        subject = f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 网站宕机告警"
        body = "以下网站出现异常:\n\n"
        for site in failures:
            body += f"- {site['url']}\n"
            body += f"  地址: {site.get('description', '无')}\n"
        msg = MIMEText(body, 'plain', 'utf-8')
        msg['Subject'] = Header(subject, 'utf-8')
        msg['From'] = sender
        msg['To'] = recipient
        try:
            with smtplib.SMTP(smtp_server, smtp_port) as server:
                server.starttls()
                server.login(sender, password)
                server.send_message(msg)
            logging.info(f"告警邮件已发送至 {recipient}")
        except Exception as e:
            logging.error(f"发送邮件失败: {e}")
    def run(self):
        """运行监控循环"""
        check_interval = self.websites[0].get('check_interval', 300) if self.websites else 300
        logging.info(f"开始监控 {len(self.websites)} 个网站")
        while True:
            try:
                self.check_all_websites()
                time.sleep(check_interval)
            except KeyboardInterrupt:
                logging.info("监控程序已停止")
                break
            except Exception as e:
                logging.error(f"监控循环发生错误: {e}")
                time.sleep(60)
# 配置文件示例 (websites.json)
config_example = """
[
    {
        "url": "https://example.com",
        "timeout": 10,
        "expected_keyword": "Example",
        "description": "主要网站",
        "check_interval": 60
    },
    {
        "url": "https://api.example.com/health",
        "timeout": 5,
        "expected_keyword": "ok",
        "description": "API服务",
        "check_interval": 30
    }
]
"""
if __name__ == "__main__":
    monitor = WebsiteMonitor()
    monitor.run()

极简版 - 单文件监控脚本

#!/bin/bash
# simple_monitor.sh - 简单快速监控
URL="https://your-website.com"
SCRIPT_LOG="/tmp/website_check.log"
# 检查直到成功或达到最大尝试次数
max_attempts=5
attempt=1
echo "开始监控: $URL"
while [ $attempt -le $max_attempts ]; do
    echo -n "尝试 $attempt/$max_attempts: "
    timeout 10 curl -s -o /dev/null -w "%{http_code}" "$URL" 2>/dev/null
    result=$?
    if [ $result -eq 0 ]; then
        echo " ✓ 网站正常"
        exit 0
    else
        echo " ✗ 请求失败"
        attempt=$((attempt+1))
        sleep 5
    fi
done
echo "网站确认无法访问!"
echo "时间: $(date)" >> "$SCRIPT_LOG"
# 在这里添加告警操作,如发送邮件等
exit 1

使用 Docker 部署监控服务

# docker-compose.yml
version: '3.8'
services:
  monitor:
    build: .
    container_name: website-monitor
    volumes:
      - ./logs:/var/log/monitor
      - ./websites.json:/app/websites.json
    environment:
      - SMTP_SERVER=smtp.gmail.com
      - SMTP_PORT=587
      - ALERT_EMAIL=your-email@gmail.com
      - ALERT_EMAIL_PASSWORD=your-app-password
      - ALERT_RECIPIENT=recipient@example.com
    restart: unless-stopped

使用方法

Bash 版本:

# 赋予执行权限
chmod +x website_monitor.sh
# 运行
./website_monitor.sh
# 后台运行
nohup ./website_monitor.sh &

Python 版本:

# 安装依赖
pip install requests
# 创建配置文件
cat > websites.json << EOF
[
    {"url": "https://example.com", "timeout": 10, "expected_keyword": "Example"}
]
EOF
# 运行
python3 multi_website_monitor.py

告警功能集成

# alert_functions.py - 可选的告警函数
import requests
import os
def send_webhook_alert(url, message):
    """发送Webhook告警"""
    payload = {
        "text": f"🚨 网站监控告警\n{message}",
        "timestamp": int(time.time())
    }
    try:
        requests.post(url, json=payload)
        print("Webhook通知已发送")
    except Exception as e:
        print(f"Webhook发送失败: {e}")
def send_sms_alert(phone, message, twilio_sid, twilio_token):
    """使用Twilio发送短信"""
    from twilio.rest import Client
    client = Client(twilio_sid, twilio_token)
    try:
        message = client.messages.create(
            body=message,
            from_='+1234567890',  # Twilio号码
            to=phone
        )
        print(f"SMS已发送: {message.sid}")
    except Exception as e:
        print(f"SMS发送失败: {e}")

监控优化建议

  1. 响应时间监控:加上curl -w '%{time_total}'测量响应时间验证**:检查页面是否包含特定关键词
  2. SSL证书监控:检查证书是否过期
  3. 多节点监控:从不同地域检查网站可用性
  4. 历史记录存储:将监控结果存入数据库

这个脚本会持续监控网站状态,并在出现问题时发出告警,您可以根据需要调整监控频率、超时时间和告警方式(邮件、短信、webhook等)。

抱歉,评论功能暂时关闭!