抓取监控看板 - 爬虫系统实时监控与告警详解

📂 所属阶段：第六阶段 — 运维与监控（工程化篇）
🔗 相关章节：Scrapyd与ScrapydWeb · Docker容器化爬虫 · Scrapy-Redis分布式架构

目录

监控系统概述

爬虫监控系统是保障工程化爬虫稳定运行的「眼睛」和「耳朵」——它通过指标收集、日志聚合、可视化展示、自动化告警，把抽象的爬虫/系统状态转化为直观的可观测信息。

为什么工程化爬虫必须有监控？

小爬虫跑在本地无所谓，但一旦分布式化、长期稳定服务化：

  • 防失联：及时发现爬虫进程/容器/集群宕机
  • 控质量：监控抓取成功率、去重率、数据完整性
  • 降成本：避免资源浪费（比如长期跑空任务、内存泄漏撑爆服务器）
  • 速排错：不用翻几百G的原始日志，靠指标定位问题范围

极简可落地架构

我们推荐轻量化三剑客架构——比全套ELK+Prom+Grafana更易上手，适合中小规模爬虫：

"""
极简爬虫监控架构：

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│  Scrapy/Scrapyd │────│  指标/日志上传  │────│  存储+展示+告警 │
│  (数据源)       │    │  (轻量SDK/File) │    │  (Grafana Cloud)│
└─────────────────┘    └─────────────────┘    └─────────────────┘

Grafana Cloud 提供免费版的 Prometheus、Loki、Grafana 托管，不用自己搭服务器，就能跑通全流程。

核心组件快速实践

第一步：用Prometheus Client暴露核心指标

不需要复杂的自定义收集器，Scrapy本身可以加个中间件暴露基本指标：

# scrapy_project/middlewares.py
from prometheus_client import start_http_server, Counter, Gauge, Histogram
import time

# 全局指标定义
REQUESTS = Counter('scrapy_requests_total', '总请求数', ['spider', 'status'])
ERRORS = Counter('scrapy_errors_total', '总错误数', ['spider', 'error_type'])
ACTIVE_CONCURRENCY = Gauge('scrapy_active_concurrency', '当前活跃请求数', ['spider'])
RESPONSE_TIME = Histogram('scrapy_response_time_seconds', '响应时间分布', ['spider'], buckets=[0.1, 0.5, 1, 3, 10])

class PrometheusMetricsMiddleware:
    def __init__(self, settings):
        self.port = settings.getint('PROMETHEUS_METRICS_PORT', 8000)
        start_http_server(self.port)  # 爬虫启动时暴露指标端口
    
    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler.settings)
    
    def process_request(self, request, spider):
        ACTIVE_CONCURRENCY.labels(spider=spider.name).inc()
        request.meta['start_time'] = time.time()
    
    def process_response(self, request, response, spider):
        if 'start_time' in request.meta:
            RESPONSE_TIME.labels(spider=spider.name).observe(time.time() - request.meta['start_time'])
        ACTIVE_CONCURRENCY.labels(spider=spider.name).dec()
        REQUESTS.labels(spider=spider.name, status=str(response.status)).inc()
        return response
    
    def process_exception(self, request, exception, spider):
        ACTIVE_CONCURRENCY.labels(spider=spider.name).dec()
        # 简单分类错误类型
        error_type = type(exception).__name__
        ERRORS.labels(spider=spider.name, error_type=error_type).inc()

记得在 settings.py 中启用中间件：

DOWNLOADER_MIDDLEWARES = {
    'scrapy_project.middlewares.PrometheusMetricsMiddleware': 100,
}
PROMETHEUS_METRICS_PORT = 8001  # 每个爬虫实例用不同端口

第二步：托管版Grafana Cloud快速配置

  1. 注册 Grafana Cloud 免费账号
  2. 进入「Connections」→「Add new connection」→ 搜索「Prometheus」
  3. 复制远程Prometheus的「scrape config」，修改爬虫的 targets 为你的公网IP+端口（或用ngrok穿透本地端口）
  4. 回到「Dashboards」，搜索「Scrapy」官方模板，一键导入即可看到基本监控面板

故障排查与诊断

快速补全诊断工具

前面的代码断了，先把核心诊断工具补好：

# diagnostic_tools.py
import psutil
import requests
import socket
import time
import os
from datetime import datetime
import logging
from typing import Dict, List, Optional

class DiagnosticTools:
    """故障诊断工具集合（轻量版）"""
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    def check_system_resources(self) -> Dict:
        """检查系统资源阈值（超85%报警示）"""
        res = {
            'cpu_percent': psutil.cpu_percent(interval=1),
            'memory_percent': psutil.virtual_memory().percent,
            'disk_percent': psutil.disk_usage('/').percent,
            'process_count': len(psutil.pids()),
        }
        # 快速标记异常
        res['warnings'] = []
        if res['cpu_percent'] > 85:
            res['warnings'].append('CPU使用率过高')
        if res['memory_percent'] > 85:
            res['warnings'].append('内存使用率过高')
        if res['disk_percent'] > 85:
            res['warnings'].append('磁盘空间不足')
        return res
    
    def check_network_connectivity(self, urls: List[str]) -> Dict:
        """检查爬虫目标站/中间件的网络连通性"""
        results = {}
        for url in urls:
            try:
                start = time.time()
                response = requests.head(url, timeout=5, allow_redirects=True)
                results[url] = {
                    'status': 'success',
                    'code': response.status_code,
                    'latency': round(time.time() - start, 3)
                }
            except Exception as e:
                results[url] = {'status': 'failed', 'reason': str(e)}
        return results
    
    def check_port_availability(self, host: str, ports: List[int]) -> Dict:
        """检查Scrapyd/Redis/Prometheus等关键端口"""
        results = {}
        for port in ports:
            sock = None
            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout(2)
                sock.connect((host, port))
                results[port] = 'open'
            except Exception:
                results[port] = 'closed'
            finally:
                if sock:
                    sock.close()
        return results
    
    def run_full_diagnosis(self, spider_hosts: List[str] = None) -> str:
        """一键生成诊断报告（简化版）"""
        report = [f"# 爬虫诊断报告 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"]
        report.append("\n## 系统资源")
        sys_res = self.check_system_resources()
        report.append(json.dumps(sys_res, indent=2, ensure_ascii=False))
        report.append("\n## 关键端口（本地）")
        port_res = self.check_port_availability('localhost', [6800, 6379, 8001])
        report.append(json.dumps(port_res, indent=2))
        return "\n".join(report)

# 使用示例
if __name__ == "__main__":
    import json
    diag = DiagnosticTools()
    print(diag.run_full_diagnosis())

常见故障速查表

现象排查步骤
抓取成功率骤降1. 查Prometheus scrapy_requests_total 的5xx/4xx状态；2. 查网络连通性；3. 查目标站是否反爬（验证码/IP封禁）
爬虫运行一段时间后停止1. 查内存/CPU是否撑爆；2. 查日志是否有内存泄漏；3. 查Scrapyd的任务日志
Grafana无数据1. 查本地Prometheus指标端口是否可访问；2. 查Grafana Cloud的scrape config是否正确；3. 查ngrok是否稳定

监控最佳实践

  1. 别暴露公网指标端口：用内网穿透（仅限本地测试）或VPC网络，指标端口只对Prometheus开放
  2. 设置合理的告警阈值：比如错误率连续5分钟>5%才告警，避免误报
  3. 分阶段监控：开发阶段用轻量本地Grafana，测试阶段加告警，生产阶段用托管或高可用架构
  4. 定期清理日志/指标：Loki和Prometheus都有保留策略，免费版别超过存储限制
  5. 监控面板分层：总览层（老板/运维看）、详情层（开发看）、单任务层（排查用）

（全文完，约2800字）