RTSP服务器的日志分析与故障定位方法

一、RTSP服务器日志分析的重要性

做流媒体开发的朋友应该都体会过，RTSP服务器一旦出问题，排查起来特别让人头大。视频流突然断了、客户端连不上、延迟飙升到几秒……这些问题的答案，往往就藏在日志文件里。但是很多时候，日志量巨大，密密麻麻的文本让人望而生畏，不知道从哪里入手。

这篇文章就是来解决这个痛点的。我会用大白话把RTSP服务器日志分析的方法讲清楚，从怎么收集日志、怎么读懂日志，到怎么定位具体问题，一步步带你走一遍。不管你之前是刚入门的新人，还是有一定经验的开发者，读完之后应该都能上手干活。

1.1 RTSP协议基础回顾

在深入日志分析之前，咱们先花两分钟回忆一下RTSP协议的基本概念。RTSP全称是Real Time Streaming Protocol，中文叫实时流媒体传输协议。你可以把它理解成一个"遥控器"——客户端通过RTSP跟服务器说"我要看这个摄像头"、"我要暂停"、"我要快进"，服务器收到指令后，通过RTP/UDP或TCP把视频流传过来。

RTSP的交互过程大致分这么几步：先发送OPTIONS查询服务器支持哪些能力，然后DESCRIBE获取流媒体的描述信息，接着SETUP建立会话通道，之后PLAY开始播放，中间可能还有PAUSE暂停、TEARDOWN断开等操作。每一个步骤都会产生对应的日志记录，这些日志就是咱们后续分析的原材料。

1.2 为什么需要关注RTSP日志

可能有人会说，"服务器出问题了不重启一下就好了吗？"短期来看确实能解决问题，但如果你不分析日志，同样的问题还会再犯。打个比方，家里水管漏水了，你把阀门关了当然不漏了，但漏水点还在，下次开阀门照样漏。

更实际地说，RTSP服务器在安防监控场景里通常是7x24小时运行的。如果你有几十个甚至上百个摄像头通过RTSP推流，偶尔断一下可能还能忍受，但如果频繁断流、延迟越来越高，最终影响的是整个监控系统的可靠性。通过日志分析，你可以提前发现异常趋势，在问题恶化之前就处理好。

二、常见RTSP日志格式与解析

2.1 主流RTSP服务器日志格式

市面上常见的RTSP服务器有好几种，不同服务器的日志格式差异不小。拿几个主流的来说：

Live555 (ffmpeg自带的RTSP服务)： Live555的日志比较简洁，通常只记录关键事件。比如客户端连接、会话建立、流传输开始和结束等信息。它的日志路径一般是标准输出，需要你用重定向保存到文件。

MediaMTX： 这是比较流行的开源RTSP/RTMP/HLS服务器，它的日志格式非常清晰，每条日志都带时间戳、日志级别和模块名。对于新手来说，MediaMTX的日志是最容易上手的。

商用NVR/ONVIF设备： 如果你用的是海康威视、大华这类商用设备的内置RTSP服务，日志格式会各不相同。但好消息是，核心字段（时间、事件类型、客户端IP、流名称）大部分都能对上。

2.2 日志字段解析详解

不管哪种RTSP服务器的日志，核心的字段基本逃不出这几个维度：

# 技术栈：Python 3.11
# 这是一个通用的RTSP日志字段解析示例

# 模拟一条典型的RTSP服务器日志记录
log_sample = {
    "timestamp": "2024-11-15 08:23:45.123",  # 事件发生的时间
    "level": "ERROR",                        # 日志级别：DEBUG/INFO/WARN/ERROR/FATAL
    "module": "rtsp_session",               # 产生日志的模块名称
    "client_ip": "192.168.1.105",           # 客户端的IP地址
    "session_id": "S-20241115-0042",        # 会话标识符，每个连接唯一
    "event": "SETUP_FAILED",                # RTSP事件类型
    "stream_path": "/live/camera01",        # 请求的流媒体路径
    "error_code": 503,                      # 错误码（如果有）
    "error_msg": "Transport not supported", # 错误描述信息
    "duration": 345,                        # 会话持续时间（毫秒）
}

# 这些字段的组合可以告诉你很多信息
# 比如：什么时候、哪个IP、请求哪个流、出了什么错、持续了多久
print(f"[{log_sample['timestamp']}] {log_sample['level']} "
      f"| {log_sample['module']} | Client: {log_sample['client_ip']} "
      f"| Session: {log_sample['session_id']} "
      f"| Event: {log_sample['event']} "
      f"| Error: {log_sample['error_code']} - {log_sample['error_msg']}")

看懂了这些字段，你对RTSP服务器发生的事情就有了基本判断。接下来就是怎么把这些信息用起来。

三、日志收集与预处理方法

3.1 日志收集工具选择

日志分析的第一步是收集。如果你只有少量服务器，直接SSH上去看文件就够了。但一旦服务器数量多了，你肯定需要更自动化的方式。

常用的日志收集方案有这么几种：

第一种是rsync或者scp，定时把日志文件拉到一台专门的收集服务器上。优点是简单，缺点是实时性差。

第二种是Filebeat，这是Elastic Stack里的日志收集代理。它能实时监控日志文件的变化，把新产生的日志行发送到Elasticsearch。对于需要历史搜索和可视化分析的团队来说，这是首选方案。

第三种是自建轻量级收集脚本。对于中小规模的部署，写一个简单的Python脚本就足够了，下面我给你写一个。

3.2 日志预处理脚本示例

# 技术栈：Python 3.11
# RTSP服务器日志预处理脚本
# 功能：读取原始日志文件，提取关键字段，输出结构化数据

import re
import json
from datetime import datetime
from collections import Counter

# RTSP标准事件类型列表
RTSP_EVENTS = [
    "OPTIONS", "DESCRIBE", "SETUP", "PLAY",
    "PAUSE", "TEARDOWN", "ANNOUNCE", "RECORD",
    "REDIRECT", "PATCH"
]

# RTSP状态码映射
STATUS_CODES = {
    100: "Continue",
    200: "OK",
    400: "Bad Request",
    401: "Unauthorized",
    403: "Forbidden",
    404: "Not Found",
    500: "Internal Server Error",
    503: "Service Unavailable",
}

def parse_rtsp_log_line(line: str) -> dict | None:
    """
    解析单行RTSP日志文本，返回结构化字典
    返回None表示该行无法解析（非日志行）
    """
    # 匹配常见日志格式：时间戳 级别 模块 消息
    pattern = (
        r'(?P<timestamp>\d{4}-\d{2}-\d{2}\s+'
        r'\d{2}:\d{2}:\d{2}(?:\.\d+)?)\s+'
        r'(?P<level>DEBUG|INFO|WARN|ERROR|FATAL)\s+'
        r'\[?(?P<module>[^\]\s]+)\]?\s*'
        r'(?P<message>.+)$'
    )
    match = re.match(pattern, line.strip())
    if not match:
        return None

    log_entry = {
        "timestamp": match.group("timestamp"),
        "level": match.group("level"),
        "module": match.group("module"),
        "message": match.group("message"),
    }

    # 从消息中提取RTSP事件类型
    for event in RTSP_EVENTS:
        if event in log_entry["message"].upper():
            log_entry["rtsp_event"] = event
            break

    # 从消息中提取客户端IP地址
    ip_pattern = r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'
    ip_match = re.search(ip_pattern, log_entry["message"])
    if ip_match:
        log_entry["client_ip"] = ip_match.group(1)

    # 从消息中提取HTTP状态码
    code_pattern = r'(?:status|code|HTTP)\s*[:=]?\s*(\d{3})'
    code_match = re.search(code_pattern, log_entry["message"], re.IGNORECASE)
    if code_match:
        code = int(code_match.group(1))
        log_entry["status_code"] = code
        log_entry["status_text"] = STATUS_CODES.get(code, "Unknown")

    return log_entry


def analyze_log_file(filepath: str) -> dict:
    """
    分析整个日志文件，返回统计摘要
    """
    stats = {
        "total_lines": 0,
        "parsed_lines": 0,
        "level_counts": Counter(),
        "event_counts": Counter(),
        "ip_counts": Counter(),
        "errors": [],
        "time_range": {"start": None, "end": None},
    }

    with open(filepath, "r", encoding="utf-8") as f:
        for line in f:
            stats["total_lines"] += 1
            entry = parse_rtsp_log_line(line)
            if entry is None:
                continue

            stats["parsed_lines"] += 1

            # 统计日志级别分布
            stats["level_counts"][entry["level"]] += 1

            # 统计RTSP事件分布
            if "rtsp_event" in entry:
                stats["event_counts"][entry["rtsp_event"]] += 1

            # 统计客户端IP分布
            if "client_ip" in entry:
                stats["ip_counts"][entry["client_ip"]] += 1

            # 收集错误级别的日志
            if entry["level"] in ("ERROR", "FATAL"):
                stats["errors"].append(entry)

            # 记录时间范围
            if stats["time_range"]["start"] is None:
                stats["time_range"]["start"] = entry["timestamp"]
            stats["time_range"]["end"] = entry["timestamp"]

    # 转换为可序列化的格式（Counter不可直接json序列化）
    stats["level_counts"] = dict(stats["level_counts"])
    stats["event_counts"] = dict(stats["event_counts"])
    stats["ip_counts"] = dict(stats["ip_counts"])

    return stats


# ===== 使用示例 =====
if __name__ == "__main__":
    # 这里假设你已经有一个日志文件路径
    # result = analyze_log_file("/var/log/rtsp_server.log")
    # print(json.dumps(result, indent=2, ensure_ascii=False))

    # 模拟几条日志行来演示解析效果
    sample_lines = [
        "2024-11-15 08:23:45.123 INFO  [rtsp_session] Client 192.168.1.105 DESCRIBE /live/camera01 status=200",
        "2024-11-15 08:23:46.456 ERROR [rtsp_session] Client 10.0.0.88 SETUP /live/camera02 failed status=503",
        "2024-11-15 08:23:47.789 WARN  [rtsp_stream] RTP packet loss detected for session S-0042",
        "2024-11-15 08:23:48.000 INFO  [rtsp_session] Client 192.168.1.105 TEARDOWN /live/camera01 duration=5000ms",
        "2024-11-15 08:23:49.111 FATAL [rtsp_server] Server crashed: out of memory",
    ]

    print("=" * 60)
    print("RTSP日志解析演示")
    print("=" * 60)

    for line in sample_lines:
        entry = parse_rtsp_log_line(line)
        if entry:
            print(json.dumps(entry, indent=2, ensure_ascii=False))
            print("-" * 60)
        else:
            print(f"[跳过] 无法解析: {line[:50]}...")
            print("-" * 60)

这段代码的作用是把你拿到手的原始日志文本，变成结构化的数据。结构化之后，后续的统计分析、筛选过滤、可视化展示才变得可行。

四、典型故障场景与定位方法

4.1 连接建立失败问题

这是最常见的RTSP故障之一。客户端发请求，服务器没反应或者返回错误码。这种情况在日志里通常表现为DESCRIBE或SETUP阶段返回非200状态码。

# 技术栈：Python 3.11
# 故障定位工具：分析连接失败问题

import re
from collections import defaultdict

def diagnose_connection_failures(log_entries: list[dict]) -> dict:
    """
    诊断连接建立失败的原因
    
    参数:
        log_entries: 解析后的日志条目列表
    返回:
        诊断结果字典
    """
    diagnosis = {
        "total_connections": 0,       # 总连接数
        "failed_connections": 0,      # 失败连接数
        "failure_by_code": defaultdict(int),    # 按错误码分类
        "failure_by_ip": defaultdict(int),      # 按客户端IP分类
        "failure_by_stream": defaultdict(int),  # 按流路径分类
        "failure_details": [],         # 失败详情
    }

    for entry in log_entries:
        msg = entry.get("message", "").upper()

        # 识别连接建立相关的事件
        is_connection_event = any(
            event in msg for event in ["DESCRIBE", "SETUP", "OPTIONS"]
        )

        if is_connection_event:
            diagnosis["total_connections"] += 1

        # 识别失败事件
        status_code = entry.get("status_code")
        is_failure = (
            status_code is not None and status_code >= 400
        ) or (entry.get("level") in ("ERROR", "FATAL"))

        if is_failure:
            diagnosis["failed_connections"] += 1

            # 按错误码统计
            if status_code:
                diagnosis["failure_by_code"][status_code] += 1

            # 按IP统计
            if "client_ip" in entry:
                diagnosis["failure_by_ip"][entry["client_ip"]] += 1

            # 按流路径统计（从消息中提取）
            stream_match = re.search(
                r'(/live/\S+|/cameras/\S+|/streams/\S+)',
                entry.get("message", "")
            )
            if stream_match:
                diagnosis["failure_by_stream"][stream_match.group(1)] += 1

            # 记录详情
            diagnosis["failure_details"].append({
                "time": entry["timestamp"],
                "ip": entry.get("client_ip", "unknown"),
                "code": status_code,
                "message": entry.get("message", "")[:200],
            })

    # 计算失败率
    if diagnosis["total_connections"] > 0:
        diagnosis["failure_rate"] = (
            diagnosis["failed_connections"] / diagnosis["total_connections"]
        ) * 100
    else:
        diagnosis["failure_rate"] = 0.0

    # 转换defaultdict为普通dict以便JSON序列化
    diagnosis["failure_by_code"] = dict(diagnosis["failure_by_code"])
    diagnosis["failure_by_ip"] = dict(diagnosis["failure_by_ip"])
    diagnosis["failure_by_stream"] = dict(diagnosis["failure_by_stream"])

    return diagnosis


# ===== 诊断结果解读示例 =====
# 假设诊断结果如下：
sample_result = {
    "total_connections": 1500,
    "failed_connections": 230,
    "failure_rate": 15.33,
    "failure_by_code": {"404": 80, "503": 120, "401": 30},
    "failure_by_ip": {"10.0.0.88": 100, "192.168.1.200": 90, "172.16.0.5": 40},
    "failure_by_stream": {"/live/camera03": 90, "/live/camera07": 80, "/live/camera01": 60},
}

print("=" * 60)
print("连接失败诊断报告")
print("=" * 60)
print(f"总连接数: {sample_result['total_connections']}")
print(f"失败连接数: {sample_result['failed_connections']}")
print(f"失败率: {sample_result['failure_rate']:.2f}%")
print(f"\n按错误码分布:")
for code, count in sample_result["failure_by_code"].items():
    print(f"  HTTP {code}: {count}次 ({count/sample_result['failed_connections']*100:.1f}%)")

print(f"\n按客户端IP分布（Top 3）:")
for ip, count in sorted(
    sample_result["failure_by_ip"].items(), key=lambda x: -x[1]
)[:3]:
    print(f"  {ip}: {count}次")

print(f"\n按流路径分布（Top 3）:")
for stream, count in sorted(
    sample_result["failure_by_stream"].items(), key=lambda x: -x[1]
)[:3]:
    print(f"  {stream}: {count}次")

# 解读：
# - 如果503错误占大头，说明服务器资源不足或后端流源不可用
# - 如果404错误集中，说明请求的流路径不存在（可能配置错误或流未发布）
# - 如果某个IP的失败次数异常高，可能是客户端配置有问题或遭受攻击
# - 如果某个流路径的失败次数异常高，可能是该摄像头/推流端出了问题

看这个诊断结果就能快速定位问题。比如503错误占了大头，说明服务器可能扛不住了，需要扩容或者优化资源配置。如果某个IP的失败次数特别高，那八成是这个客户端的问题，不是服务器的问题。

4.2 流媒体传输中断问题

连接建立成功了，但视频流传着传着就断了。这种问题更隐蔽，因为从连接层面看一切正常，但用户体验就是视频卡住了。

# 技术栈：Python 3.11
# 流媒体传输中断分析

import time
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime

@dataclass
class StreamSession:
    """表示一个流媒体会话"""
    session_id: str
    client_ip: str
    stream_path: str
    start_time: float
    end_time: float | None = None
    packet_loss_ratio: float = 0.0    # 丢包率
    bitrate_drop_events: int = 0      # 码率下降事件次数
    interruptions: list[dict] = None

    def __post_init__(self):
        if self.interruptions is None:
            self.interruptions = []


def detect_stream_interruptions(
    log_entries: list[dict],
    interruption_threshold_seconds: float = 3.0,
) -> list[dict]:
    """
    检测流媒体传输中断
    
    参数:
        log_entries: 日志条目列表（按时间排序）
        interruption_threshold_seconds: 中断判定阈值（秒）
    返回:
        中断事件列表
    """
    interruptions = []
    active_sessions = {}  # 跟踪活跃的流会话

    for entry in log_entries:
        msg = entry.get("message", "")
        level = entry.get("level", "INFO")

        # 检测丢包事件
        if "packet loss" in msg.lower() or "rtp loss" in msg.lower():
            # 提取丢包率
            loss_match = re.search(r'(\d+\.?\d*)%', msg)
            loss_ratio = float(loss_match.group(1)) / 100 if loss_match else 0

            session_id = msg.split()[0] if msg else "unknown"
            if session_id not in active_sessions:
                active_sessions[session_id] = StreamSession(
                    session_id=session_id,
                    client_ip=entry.get("client_ip", "unknown"),
                    stream_path=msg,
                    start_time=parse_timestamp(entry["timestamp"]),
                    packet_loss_ratio=loss_ratio,
                )
            else:
                active_sessions[session_id].packet_loss_ratio = max(
                    active_sessions[session_id].packet_loss_ratio, loss_ratio
                )
                active_sessions[session_id].bitrate_drop_events += 1

        # 检测超时或连接重置
        if "timeout" in msg.lower() or "reset" in msg.lower() or "broken pipe" in msg.lower():
            session_id = msg.split()[0] if msg else "unknown"
            if session_id in active_sessions:
                session = active_sessions.pop(session_id)
                session.end_time = parse_timestamp(entry["timestamp"])
                duration = session.end_time - session.start_time

                # 如果会话持续时间超过阈值且有关键事件，标记为异常中断
                if duration > interruption_threshold_seconds or level in ("ERROR", "FATAL"):
                    interruptions.append({
                        "session_id": session_id,
                        "client_ip": session.client_ip,
                        "duration_seconds": round(duration, 2),
                        "packet_loss_ratio": round(session.packet_loss_ratio, 3),
                        "bitrate_drop_events": session.bitrate_drop_events,
                        "reason": entry.get("message", "unknown")[:100],
                        "end_time": entry["timestamp"],
                    })

    return interruptions


def parse_timestamp(ts_str: str) -> float:
    """将时间戳字符串转为时间戳数值"""
    try:
        # 支持多种时间格式
        for fmt in ["%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S"]:
            try:
                return datetime.strptime(ts_str, fmt).timestamp()
            except ValueError:
                continue
    except Exception:
        pass
    return 0.0


# ===== 使用示例 =====
# 典型的中断分析报告解读：
# 1. 如果packet_loss_ratio持续高于5%，说明网络环境有问题
#    可能是带宽不足、网络拥塞或设备性能瓶颈
# 2. 如果同一个session_id频繁出现中断，说明该连接不稳定
#    可能是客户端网络问题或服务器端资源被抢占
# 3. 如果所有会话的丢包率同时升高，很可能是网络基础设施问题
#    （交换机故障、路由器带宽打满等）
# 4. 如果只在特定时段出现中断，可能是高峰期资源争抢导致的

丢包率是判断流媒体传输质量的关键指标。一般来说，RTP丢包率低于2%的视频基本看不出明显影响，超过5%就开始出现画面卡顿，超过10%基本就不可用了。

4.3 性能瓶颈与资源耗尽

RTSP服务器扛不住负载的时候，日志里会有一系列"前兆信号"。学会看这些信号，你就能在服务器真正崩掉之前做出响应。

# 技术栈：Python 3.11
# 性能瓶颈与资源监控

import json
from datetime import datetime, timedelta
from collections import deque

class RTSPPerformanceMonitor:
    """
    RTSP服务器性能监控器
    基于日志数据实时计算关键性能指标
    """

    def __init__(self, window_seconds: int = 60):
        # 滑动时间窗口（秒），用于计算短期指标
        self.window_seconds = window_seconds
        # 用双端队列存储时间窗口内的数据点
        self.conns_per_minute = deque()
        self.active_sessions = deque()
        self.error_rate_samples = deque()
        self.response_time_samples = deque()

    def update_from_log(self, entry: dict) -> None:
        """
        从一条日志条目中提取性能数据并更新监控状态
        """
        timestamp = parse_timestamp(entry.get("timestamp", ""))
        if timestamp == 0:
            return

        # 记录活跃会话数（从SETUP成功事件推断）
        if "setup" in entry.get("message", "").lower() and entry.get("status_code") == 200:
            self.active_sessions.append(timestamp)

        # 如果SETUP失败，从活跃列表移除
        if "setup" in entry.get("message", "").lower() and entry.get("status_code") and entry["status_code"] >= 400:
            pass  # 不记录失败连接

        # 记录错误率样本
        if entry.get("level") in ("ERROR", "FATAL"):
            self.error_rate_samples.append(timestamp)

    def get_health_report(self, current_time: float | None = None) -> dict:
        """
        生成当前健康状态报告
        """
        if current_time is None:
            current_time = datetime.now().timestamp()

        # 清理过期数据点
        cutoff = current_time - self.window_seconds
        while self.conns_per_minute and self.conns_per_minute[0] < cutoff:
            self.conns_per_minute.popleft()
        while self.active_sessions and self.active_sessions[0] < cutoff:
            self.active_sessions.popleft()
        while self.error_rate_samples and self.error_rate_samples[0] < cutoff:
            self.error_rate_samples.popleft()

        total_samples = len(self.conns_per_minute) + 1  # +1避免除零
        error_count = len(self.error_rate_samples)
        current_error_rate = (error_count / total_samples) * 100

        # 健康评分计算（0-100分，越高越好）
        health_score = 100.0
        if current_error_rate > 5:
            health_score -= 30
        if current_error_rate > 15:
            health_score -= 30
        if current_error_rate > 30:
            health_score -= 30

        health_score = max(0, min(100, health_score))

        # 健康等级判定
        if health_score >= 80:
            level = "正常"
        elif health_score >= 50:
            level = "警告"
        elif health_score >= 30:
            level = "严重"
        else:
            level = "危急"

        return {
            "timestamp": datetime.fromtimestamp(current_time).isoformat(),
            "health_score": round(health_score, 1),
            "health_level": level,
            "active_sessions": len(self.active_sessions),
            "error_count_in_window": error_count,
            "error_rate_percent": round(current_error_rate, 2),
            "recommendation": self._get_recommendation(health_score, current_error_rate),
        }

    def _get_recommendation(self, score: float, error_rate: float) -> str:
        """根据健康评分给出运维建议"""
        if score >= 80:
            return "系统运行正常，继续保持监控"
        elif score >= 50:
            return "错误率偏高，建议检查网络环境和客户端连接"
        elif score >= 30:
            return "系统压力较大，建议考虑扩容或限制并发连接数"
        else:
            return "系统濒临崩溃，立即介入！检查服务器资源、网络带宽和后端流源"


# ===== 性能监控使用示例 =====
if __name__ == "__main__":
    monitor = RTSPPerformanceMonitor(window_seconds=300)  # 5分钟窗口

    # 模拟一批日志数据
    simulated_logs = [
        {
            "timestamp": "2024-11-15 08:00:00",
            "level": "INFO",
            "message": "SETUP /live/camera01 status=200 session=S-001",
            "status_code": 200,
            "client_ip": "192.168.1.100",
        },
        {
            "timestamp": "2024-11-15 08:00:01",
            "level": "ERROR",
            "message": "SETUP /live/camera02 status=503 service unavailable",
            "status_code": 503,
            "client_ip": "192.168.1.101",
        },
        {
            "timestamp": "2024-11-15 08:00:02",
            "level": "INFO",
            "message": "SETUP /live/camera03 status=200 session=S-003",
            "status_code": 200,
            "client_ip": "192.168.1.102",
        },
    ]

    # 更新监控状态
    for log in simulated_logs:
        monitor.update_from_log(log)

    # 获取健康报告
    report = monitor.get_health_report()
    print(json.dumps(report, indent=2, ensure_ascii=False))

这个监控器帮你持续跟踪服务器的健康状况。你可以把它集成到CI/CD流程或者定时任务里，让它在后台默默运行，一旦健康评分低于阈值就触发告警。

五、自动化监控与告警方案

5.1 基于日志的实时监控

日志分析不应该是出事后才去翻记录。更高效的做法是实时监控——日志产生后立刻处理，发现异常立刻告警。

# 技术栈：Python 3.11
# 实时日志监控与告警脚本

import subprocess
import time
import json
from datetime import datetime
from collections import defaultdict
import threading

class RealtimeLogMonitor:
    """
    实时RTSP日志监控器
    使用tail -f方式实时读取日志并分析
    """

    def __init__(self, log_path: str, alert_thresholds: dict = None):
        self.log_path = log_path
        # 告警阈值配置
        self.thresholds = alert_thresholds or {
            "error_rate_per_minute": 10,      # 每分钟错误数阈值
            "consecutive_failures": 5,         # 连续失败次数阈值
            "max_session_duration_minutes": 60, # 最大会话时长阈值
        }
        self.error_timestamps = []
        self.consecutive_failures = defaultdict(int)
        self.alert_cooldown = {}  # 告警冷却时间，避免重复告警
        self.running = False
        self.monitor_thread = None

    def start(self):
        """启动实时监控"""
        self.running = True
        self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
        self.monitor_thread.start()
        print(f"[Monitor] 已启动，监控日志文件: {self.log_path}")

    def stop(self):
        """停止监控"""
        self.running = False
        if self.monitor_thread:
            self.monitor_thread.join(timeout=5)
        print("[Monitor] 已停止")

    def _monitor_loop(self):
        """主监控循环"""
        # 使用tail -f实时跟踪日志文件
        proc = subprocess.Popen(
            ["tail", "-f", "-n", "0", self.log_path],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )

        try:
            for line in proc.stdout:
                if not self.running:
                    break
                self._process_line(line.strip())
        except Exception as e:
            print(f"[Monitor Error] 监控异常: {e}")
        finally:
            proc.terminate()

    def _process_line(self, line: str):
        """处理单行日志"""
        if not line:
            return

        now = time.time()

        # 检查是否为错误级别日志
        if any(level in line for level in ["ERROR", "FATAL"]):
            self.error_timestamps.append(now)
            # 清理30秒前的错误记录
            self.error_timestamps = [
                t for t in self.error_timestamps if now - t < 60
            ]

            # 检查错误率是否超过阈值
            if len(self.error_timestamps) >= self.thresholds["error_rate_per_minute"]:
                self._trigger_alert(
                    "HIGH_ERROR_RATE",
                    f"每分钟错误数达到 {len(self.error_timestamps)} 条，"
                    f"超过阈值 {self.thresholds['error_rate_per_minute']} 条"
                )

        # 检查连续失败
        failure_pattern = "status=503" in line or "status=404" in line
        if failure_pattern:
            ip_match = __import__('re').search(r'(\d+\.\d+\.\d+\.\d+)', line)
            client_ip = ip_match.group(1) if ip_match else "unknown"
            self.consecutive_failures[client_ip] += 1

            if (self.consecutive_failures[client_ip] >= 
                self.thresholds["consecutive_failures"]):
                self._trigger_alert(
                    "CONSECUTIVE_FAILURES",
                    f"客户端 {client_ip} 连续失败 "
                    f"{self.consecutive_failures[client_ip]} 次"
                )
                # 重置计数器
                self.consecutive_failures[client_ip] = 0
        else:
            # 非失败事件，重置所有客户端的连续失败计数
            for ip in list(self.consecutive_failures.keys()):
                self.consecutive_failures[ip] = max(0, self.consecutive_failures[ip] - 1)

    def _trigger_alert(self, alert_type: str, message: str):
        """触发告警（带去重逻辑）"""
        key = f"{alert_type}:{message}"
        now = time.time()

        # 5分钟内同一告警不重复发送
        if key in self.alert_cooldown:
            if now - self.alert_cooldown[key] < 300:
                return
        self.alert_cooldown[key] = now

        alert_record = {
            "alert_type": alert_type,
            "message": message,
            "timestamp": datetime.now().isoformat(),
            "severity": "critical" if "FATAL" in message else "warning",
        }
        print(f"\n{'!'*60}")
        print(f"  [ALERT] {alert_record['severity'].upper()}")
        print(f"  类型: {alert_type}")
        print(f"  消息: {message}")
        print(f"  时间: {alert_record['timestamp']}")
        print(f"{'!'*60}\n")
        # 这里可以扩展为发送钉钉/企微/邮件通知


# ===== 使用方法 =====
# 在实际使用中：
# 1. 创建监控实例
# monitor = RealtimeLogMonitor("/var/log/mediamtx/mediamtx.log")
# 2. 可选：自定义告警阈值
# monitor.thresholds["error_rate_per_minute"] = 20
# 3. 启动监控
# monitor.start()
# 4. 程序需要持续运行时，可以等待或加入其他业务逻辑
# time.sleep(86400)  # 运行24小时
# 5. 停止监控
# monitor.stop()

这个监控器可以挂在你服务器的后台持续运行，一旦日志里出现异常模式，立刻给你发告警。比手动翻日志效率高太多了。

5.2 告警规则配置

不同环境对告警的需求不一样。生产环境可能要求零容忍——出现任何ERROR级别日志就告警。而开发环境可能只需要在大量报错时才通知。下面是一个灵活的告警规则配置示例：

{
    "alert_rules": [
        {
            "rule_id": "high_error_rate",
            "description": "错误率过高",
            "condition": {
                "metric": "error_count_per_minute",
                "operator": ">",
                "threshold": 10,
                "window_seconds": 60
            },
            "severity": "critical",
            "cooldown_seconds": 300,
            "notification_channels": ["dingtalk", "email"]
        },
        {
            "rule_id": "session_timeout",
            "description": "会话超时",
            "condition": {
                "metric": "session_duration_seconds",
                "operator": ">",
                "threshold": 3600,
                "window_seconds": 3600
            },
            "severity": "warning",
            "cooldown_seconds": 600,
            "notification_channels": ["dingtalk"]
        },
        {
            "rule_id": "packet_loss_critical",
            "description": "丢包率超过临界值",
            "condition": {
                "metric": "packet_loss_ratio",
                "operator": ">",
                "threshold": 0.15,
                "window_seconds": 30
            },
            "severity": "critical",
            "cooldown_seconds": 120,
            "notification_channels": ["dingtalk", "email", "sms"]
        },
        {
            "rule_id": "consecutive_setup_failures",
            "description": "连续SETUP失败",
            "condition": {
                "metric": "consecutive_failures_per_ip",
                "operator": ">=",
                "threshold": 5,
                "window_seconds": 60
            },
            "severity": "warning",
            "cooldown_seconds": 180,
            "notification_channels": ["dingtalk"]
        }
    ],
    "global_settings": {
        "max_alerts_per_minute": 20,
        "silent_period_minutes": 30,
        "default_cooldown_seconds": 300
    }
}

六、应用场景分析

6.1 安防监控系统

安防监控是RTSP最典型的应用场景。一个中型城市的安防系统可能有几千甚至上万个摄像头，每个摄像头通过RTSP把视频流送到监控中心。在这种场景下，日志分析的痛点在于：

第一，摄像头数量多，一旦出问题很难快速定位是哪个设备。通过日志里的session_id和client_ip，可以快速把问题锁定到具体的摄像头。

第二，安防系统要求7x24小时不间断运行，任何中断都可能影响安全事件的处理。实时监控和告警在这种场景下是刚需，而不是锦上添花。

第三，很多摄像头用的是不同的推流软件和不同的RTSP服务器版本，日志格式不统一。这就需要你的解析脚本有足够的兼容性。

6.2 视频会议系统

视频会议系统对实时性的要求更高。RTSP虽然不是视频会议的首选协议（WebRTC更常见），但很多专业视频会议系统仍然使用RTSP来传输音视频流。在这种场景下，丢包率和延迟是最敏感的指标。

七、技术优缺点分析

7.1 优点

日志分析最大的好处是事后追溯能力强。不管系统出了什么问题，只要日志保存得当，你就能还原出问题发生时的完整时间线。这对于排查间歇性故障特别有效。

第二个好处是不需要修改系统代码。所有分析方法都是基于已有日志进行的，对生产环境零侵入。

第三个好处是成本极低。写脚本解析日志不需要额外的硬件投入，用一台普通的服务器就能跑起来。

7.2 缺点

日志分析也有明显的局限性。第一是实时性有限。即使是tail -f方式实时读取，也比直接在代码里埋监控点要慢一拍。

第二是信息可能不完整。日志记录的内容取决于你在服务器上配置了什么级别的日志。如果之前只开了INFO级别，事后想查DEBUG信息就查不到了。

第三是日志量管理是个问题。RTSP服务器的日志可以非常大，一台服务器每天的日志轻松达到几个GB。如果不做好轮转和归档策略，磁盘很快就满了。

7.3 注意事项

做RTSP日志分析有几个容易踩的坑需要注意：

首先是时区问题。如果你的RTSP服务器和日志收集服务器不在同一个时区，日志时间对不上就会让你怀疑人生。建议在收集日志时统一转换为UTC时间，或者至少统一时区。

其次是日志轮转。大多数RTSP服务器都支持日志轮转，但你需要确认轮转后旧日志的归档策略。如果服务器重启了，重启前的日志可能丢失。

第三是隐私合规。RTSP日志里可能包含客户端IP地址等敏感信息。如果你的系统需要处理个人数据（比如涉及人脸识别的安防系统），日志中的IP信息可能需要脱敏处理。

八、文章总结

RTSP服务器日志分析虽然看起来是个"脏活累活"，但一旦掌握了方法论，就能大幅提升流媒体系统的可维护性和可靠性。核心思路就是三步走：

第一步，把原始日志变成结构化数据。用Python写解析脚本，把文本日志拆成JSON格式的键值对，后续的统计分析才有基础。

第二步，建立故障知识库。把常见的错误码、异常模式和对应的解决方案整理出来。比如503错误对应资源不足，404对应路径不存在，丢包率过高对应网络问题。这样遇到问题就能快速匹配原因。

第三步，实现自动化监控告警。不要让日志分析变成人工活，写一个实时监控系统挂在后台，让它在异常发生时第一时间通知你。

最后说一句掏心窝的话：写解析脚本的时候，别追求一步到位。先把你最痛的场景解决了——比如"每次断流后花一小时排查"——用脚本把这个时间缩短到五分钟，就已经是巨大的效率提升了。然后再慢慢迭代，加上更多分析维度。工具是为了服务人，不要为了追求工具的完美而忽略了实际问题。