OpenClaw日志与调试技巧的入门到精通指南

import logging
# Gateway 服务日志器
gateway_logger = logging.getLogger('openclaw.gateway')
# Agent 核心日志器
agent_logger = logging.getLogger('openclaw.agent')
# Skill 模块日志器
skill_logger = logging.getLogger('openclaw.skill')
# Channel 通道日志器
channel_logger = logging.getLogger('openclaw.channel')

import logging
from logging.handlers import RotatingFileHandler
import os
def setup_logging(log_level: str = "INFO", log_dir: str = "/var/log/openclaw"):
    """
    配置 OpenClaw 日志系统
    Args:
        log_level: 日志级别，支持 DEBUG/INFO/WARNING/ERROR/CRITICAL
        log_dir: 日志文件存储目录
    """
    # 确保日志目录存在
    os.makedirs(log_dir, exist_ok=True)
    # 获取根日志器
    root_logger = logging.getLogger('openclaw')
    root_logger.setLevel(getattr(logging, log_level.upper()))
    # 创建格式化器
    formatter = logging.Formatter(
        fmt='%(asctime)s | %(levelname)-8s | %(name)s | %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    # 控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    console_handler.setFormatter(formatter)
    root_logger.addHandler(console_handler)
    # 文件处理器（轮转）
    file_handler = RotatingFileHandler(
        filename=os.path.join(log_dir, 'openclaw.log'),
        maxBytes=10 * 1024 * 1024,  # 10MB
        backupCount=5,
        encoding='utf-8'
    )
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(formatter)
    root_logger.addHandler(file_handler)
    return root_logger
# 初始化日志
logger = setup_logging(log_level="DEBUG")

# openclaw-logging.yaml
version: 1
disable_existing_loggers: false
formatters:
  standard:
    format: '%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
    datefmt: '%Y-%m-%d %H:%M:%S'
  json:
    format: '{"timestamp": "%(asctime)s", "level": "%(levelname)s", "logger": "%(name)s", "message": "%(message)s"}'
    datefmt: '%Y-%m-%dT%H:%M:%S'
handlers:
  console:
    class: logging.StreamHandler
    level: DEBUG
    formatter: standard
    stream: ext://sys.stdout
  file:
    class: logging.handlers.RotatingFileHandler
    level: INFO
    formatter: standard
    filename: /var/log/openclaw/openclaw.log
    maxBytes: 10485760  # 10MB
    backupCount: 5
    encoding: utf-8
  json_file:
    class: logging.handlers.RotatingFileHandler
    level: INFO
    formatter: json
    filename: /var/log/openclaw/openclaw.json
    maxBytes: 10485760
    backupCount: 5
    encoding: utf-8
loggers:
  openclaw:
    level: INFO
    handlers: [console, file]
    propagate: false
  openclaw.gateway:
    level: DEBUG
    handlers: [console, file]
    propagate: false
  openclaw.agent:
    level: INFO
    handlers: [console, file, json_file]
    propagate: false
root:
  level: WARNING
  handlers: [console]

2024-03-19 10:30:45 | INFO     | openclaw.gateway | Gateway service started on port 18789
2024-03-19 10:30:46 | DEBUG    | openclaw.agent | Loading skill: weather
2024-03-19 10:30:47 | INFO     | openclaw.agent | Skill loaded successfully: weather v1.0.0
2024-03-19 10:31:02 | WARNING  | openclaw.channel | Telegram API rate limit approaching
2024-03-19 10:31:15 | ERROR    | openclaw.gateway | Failed to process request: connection timeout

{"timestamp": "2024-03-19T10:30:45", "level": "INFO", "logger": "openclaw.gateway", "message": "Gateway service started on port 18789", "service": "openclaw", "version": "1.0.0", "host": "openclaw-server-01"}
{"timestamp": "2024-03-19T10:30:46", "level": "DEBUG", "logger": "openclaw.agent", "message": "Loading skill: weather", "skill_name": "weather", "skill_path": "/app/skills/weather"}
{"timestamp": "2024-03-19T10:30:47", "level": "INFO", "logger": "openclaw.agent", "message": "Skill loaded successfully", "skill": {"name": "weather", "version": "1.0.0", "author": "openclaw"}}
{"timestamp": "2024-03-19T10:31:02", "level": "WARNING", "logger": "openclaw.channel", "message": "Telegram API rate limit approaching", "channel": "telegram", "rate_limit": {"remaining": 5, "reset_at": "2024-03-19T10:32:00"}}
{"timestamp": "2024-03-19T10:31:15", "level": "ERROR", "logger": "openclaw.gateway", "message": "Failed to process request", "error": {"type": "TimeoutError", "message": "connection timeout", "stack_trace": "..."}}

import logging
import json
from datetime import datetime
class JsonFormatter(logging.Formatter):
    """
    JSON 格式化器，将日志输出为结构化 JSON 格式
    """
    def __init__(self, service_name: str = "openclaw", version: str = "1.0.0"):
        super().__init__()
        self.service_name = service_name
        self.version = version
        self.hostname = os.environ.get('HOSTNAME', 'localhost')
    def format(self, record: logging.LogRecord) -> str:
        """
        格式化日志记录为 JSON 字符串
        Args:
            record: 日志记录对象
        Returns:
            JSON 格式的日志字符串
        """
        log_data = {
            "timestamp": datetime.fromtimestamp(record.created).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "service": self.service_name,
            "version": self.version,
            "host": self.hostname,
            "file": record.filename,
            "line": record.lineno,
            "function": record.funcName
        }
        # 添加额外字段
        if hasattr(record, 'extra_data'):
            log_data['extra'] = record.extra_data
        # 添加异常信息
        if record.exc_info:
            log_data['exception'] = {
                "type": record.exc_info[0].__name__,
                "message": str(record.exc_info[1]),
                "stack_trace": self.formatException(record.exc_info)
            }
        return json.dumps(log_data, ensure_ascii=False)

from logging.handlers import RotatingFileHandler
# 创建按大小轮转的文件处理器
rotating_handler = RotatingFileHandler(
    filename='/var/log/openclaw/openclaw.log',
    maxBytes=10 * 1024 * 1024,  # 单文件最大 10MB
    backupCount=10,              # 保留 10 个备份文件
    encoding='utf-8'
)
# 轮转后的文件命名规则：
# openclaw.log       (当前日志)
# openclaw.log.1     (最近的备份)
# openclaw.log.2     (次近的备份)
# ...
# openclaw.log.10    (最老的备份)

from logging.handlers import TimedRotatingFileHandler
# 创建按时间轮转的文件处理器
timed_handler = TimedRotatingFileHandler(
    filename='/var/log/openclaw/openclaw.log',
    when='midnight',    # 每天午夜轮转
    interval=1,         # 间隔 1 天
    backupCount=30,     # 保留 30 天日志
    encoding='utf-8'
)
# 设置文件名后缀格式
timed_handler.suffix = '%Y-%m-%d'
# 轮转后的文件命名规则：
# openclaw.log              (当前日志)
# openclaw.log.2024-03-18   (昨天)
# openclaw.log.2024-03-17   (前天)
# ...

import logging
import re
from typing import List, Pattern

class SensitiveDataFilter(logging.Filter):
    """
    敏感信息脱敏过滤器
    
    自动检测并脱敏日志中的敏感信息，包括：
    - API 密钥和令牌
    - 手机号码
    - 电子邮箱
    - IP 地址
    - 自定义敏感词
    """
    
    # 预定义的敏感信息匹配模式
    PATTERNS = {
        'api_key': re.compile(r'(sk-[a-zA-Z0-9]{20,}|token-[a-zA-Z0-9]{10,})'),
        'phone': re.compile(r'1[3-9]\d{9}'),
        'email': re.compile(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'),
        'ip_address': re.compile(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}'),
        'password': re.compile(r'(password|passwd|pwd)["\']?\s*[:=]\s*["\']?[^\s"\']+', re.IGNORECASE),
    }
    
    def __init__(self, custom_patterns: List[str] = None):
        """
        初始化脱敏过滤器
        
        Args:
            custom_patterns: 自定义敏感词列表
        """
        super().__init__()
        self.custom_patterns = custom_patterns or []
    
    def filter(self, record: logging.LogRecord) -> bool:
        """
        过滤日志记录，对敏感信息进行脱敏
        
        Args:
            record: 日志记录对象
        
        Returns:
            始终返回 True，允许日志通过，但会修改消息内容
        """
        # 对日志消息进行脱敏处理
        record.msg = self._sanitize(str(record.msg))
        
        # 对参数进行脱敏处理
        if record.args:
            record.args = tuple(
                self._sanitize(str(arg)) if isinstance(arg, str) else arg
                for arg in record.args
            )
        
        return True
    
    def _sanitize(self, text: str) -> str:
        """
        对文本中的敏感信息进行脱敏
        
        Args:
            text: 原始文本
        
        Returns:
            脱敏后的文本
        """
        result = text
        
        # 应用预定义模式
        for pattern_name, pattern in self.PATTERNS.items():
            result = pattern.sub(self._get_replacement(pattern_name), result)
        
        # 应用自定义敏感词
        for word in self.custom_patterns:
            result = result.replace(word, '***REDACTED***')
        
        return result
    
    def _get_replacement(self, pattern_name: str) -> str:
        """
        根据模式类型返回脱敏后的替换文本
        
        Args:
            pattern_name: 模式名称
        
        Returns:
            脱敏后的替换文本
        """
        replacements = {
            'api_key': '***API_KEY_REDACTED***',
            'phone': '***PHONE_REDACTED***',
            'email': '***EMAIL_REDACTED***',
            'ip_address': '***IP_REDACTED***',
            'password': '***PASSWORD_REDACTED***',
        }
        return replacements.get(pattern_name, '***REDACTED***')

# 使用示例
logger = logging.getLogger('openclaw')
logger.addFilter(SensitiveDataFilter(
    custom_patterns=['my-secret-key', 'internal-token']
))

# 测试脱敏效果
logger.info("User login with token: sk-abcdefghijklmnopqrstuvwxyz123456")
# 输出: User login with token: ***API_KEY_REDACTED***

logger.info("User phone: 13800138000, email: user@example.com")
# 输出: User phone: ***PHONE_REDACTED***, email: ***EMAIL_REDACTED***

# openclaw-config.yaml
openclaw:
  debug:
    enabled: true              # 开启调试模式
    log_level: DEBUG           # 日志级别
    trace_requests: true       # 追踪请求
    profile_performance: true  # 性能分析
    memory_debug: false        # 内存调试（影响性能）
  logging:
    level: DEBUG
    format: detailed           # detailed / json / compact
    output:
      - console
      - file
    file:
      path: /var/log/openclaw/debug.log
      max_size: 50MB
      backup_count: 3

# 开启调试模式
export OPENCLAW_DEBUG=true
export OPENCLAW_LOG_LEVEL=DEBUG

# 开启请求追踪
export OPENCLAW_TRACE_REQUESTS=true

# 开启性能分析
export OPENCLAW_PROFILE=true

# 启动 OpenClaw
openclaw gateway start

# 开启调试模式启动 Gateway
openclaw gateway start --debug --log-level DEBUG
# 开启性能分析模式
openclaw gateway start --profile
# 开启详细请求追踪
openclaw gateway start --trace-requests

2024-03-19 10:35:00.123 | DEBUG    | openclaw.gateway | Incoming request: POST /api/v1/chat
2024-03-19 10:35:00.124 | DEBUG    | openclaw.gateway | Request headers: {"Content-Type": "application/json", "Authorization": "***REDACTED***"}
2024-03-19 10:35:00.125 | DEBUG    | openclaw.gateway | Request body: {"message": "今天天气怎么样", "channel": "feishu"}
2024-03-19 10:35:00.126 | DEBUG    | openclaw.agent | Routing to skill: weather
2024-03-19 10:35:00.127 | DEBUG    | openclaw.skill.weather | Executing weather query for: 北京
2024-03-19 10:35:00.245 | DEBUG    | openclaw.skill.weather | API response: {"temp": 18, "condition": "晴", "humidity": 45}
2024-03-19 10:35:00.246 | DEBUG    | openclaw.agent | Generating response with model: gpt-4o-mini
2024-03-19 10:35:00.892 | DEBUG    | openclaw.gateway | Response: 北京今天天气晴朗，气温18°C，湿度45%，适合外出活动。
2024-03-19 10:35:00.893 | INFO     | openclaw.gateway | Request completed in 770ms

# 1. 检查配置文件语法
openclaw config validate
# 2. 检查端口是否被占用
lsof -i :18789
# 3. 检查日志文件权限
ls -la /var/log/openclaw/
# 4. 以调试模式启动查看详细错误
openclaw gateway start --debug

# 1. 查看最近的错误日志
tail -f /var/log/openclaw/openclaw.log | grep ERROR
# 2. 检查 AI 模型 API 响应时间
curl -w "Time: %{time_total}s\n" -X POST https://api.openai.com/v1/chat/completions \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "test"}]}'
# 3. 检查并发连接数
netstat -an | grep 18789 | wc -l
# 4. 开启性能分析
export OPENCLAW_PROFILE=true
openclaw gateway restart

# 1. 查看进程内存使用
ps aux | grep openclaw
# 2. 开启内存调试
export OPENCLAW_MEMORY_DEBUG=true
openclaw gateway restart
# 3. 分析内存日志
grep "Memory usage" /var/log/openclaw/openclaw.log
# 4. 检查是否有内存泄漏
valgrind --leak-check=full openclaw gateway start

import logging
from logging.handlers import QueueHandler, QueueListener
from queue import Queue
def setup_async_logging():
    """
    配置异步日志系统，提升性能
    使用队列将日志写入操作从主线程分离，
    避免日志 I/O 阻塞业务处理。
    """
    # 创建日志队列
    log_queue = Queue(maxsize=10000)
    # 创建队列处理器（主线程使用）
    queue_handler = QueueHandler(log_queue)
    # 创建实际的文件处理器（后台线程使用）
    file_handler = logging.FileHandler('/var/log/openclaw/openclaw.log')
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
    ))
    # 创建队列监听器
    listener = QueueListener(
        log_queue,
        file_handler,
        respect_handler_level=True
    )
    listener.start()
    # 配置日志器使用队列处理器
    logger = logging.getLogger('openclaw')
    logger.addHandler(queue_handler)
    return listener
# 使用示例
listener = setup_async_logging()
# 应用退出时停止监听器
# listener.stop()

import logging
from typing import Optional
class DynamicLogLevelManager:
    """
    动态日志级别管理器
    支持运行时动态调整日志级别，无需重启服务。
    可通过 API 或配置文件触发调整。
    """
    def __init__(self):
        self._loggers = {}
        self._original_levels = {}
    def register(self, logger_name: str) -> None:
        """
        注册需要管理的日志器
        Args:
            logger_name: 日志器名称
        """
        logger = logging.getLogger(logger_name)
        self._loggers[logger_name] = logger
        self._original_levels[logger_name] = logger.level
    def set_level(self, logger_name: str, level: str) -> bool:
        """
        设置日志级别
        Args:
            logger_name: 日志器名称，支持通配符
            level: 目标级别 (DEBUG/INFO/WARNING/ERROR/CRITICAL)
        Returns:
            是否设置成功
        """
        try:
            level_value = getattr(logging, level.upper())
            # 支持通配符匹配
            if logger_name.endswith('*'):
                prefix = logger_name[:-1]
                for name, logger in self._loggers.items():
                    if name.startswith(prefix):
                        logger.setLevel(level_value)
                        logging.info(f"Set {name} log level to {level}")
            else:
                if logger_name in self._loggers:
                    self._loggers[logger_name].setLevel(level_value)
                    logging.info(f"Set {logger_name} log level to {level}")
            return True
        except AttributeError:
            logging.error(f"Invalid log level: {level}")
            return False
    def reset(self, logger_name: Optional[str] = None) -> None:
        """
        重置日志级别到原始值
        Args:
            logger_name: 日志器名称，None 表示重置所有
        """
        if logger_name:
            if logger_name in self._original_levels:
                self._loggers[logger_name].setLevel(self._original_levels[logger_name])
        else:
            for name, original_level in self._original_levels.items():
                self._loggers[name].setLevel(original_level)
# 使用示例
manager = DynamicLogLevelManager()
manager.register('openclaw.gateway')
manager.register('openclaw.agent')
manager.register('openclaw.channel')
# 临时开启调试模式
manager.set_level('openclaw.*', 'DEBUG')
# 排查完成后恢复
manager.reset()

# openclaw-alerts.yaml
groups:
  - name: openclaw_alerts
    rules:
      # 错误率告警
      - alert: HighErrorRate
        expr: |
          rate(openclaw_requests_total{status="error"}[5m]) 
          / 
          rate(openclaw_requests_total[5m]) > 0.1
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "OpenClaw 错误率过高"
          description: "过去 5 分钟错误率超过 10%，当前值: {{ $value | humanizePercentage }}"
      # 响应时间告警
      - alert: SlowResponse
        expr: |
          histogram_quantile(0.95, 
            rate(openclaw_request_duration_seconds_bucket[5m])
          ) > 5
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "OpenClaw 响应时间过长"
          description: "P95 响应时间超过 5 秒，当前值: {{ $value | humanizeDuration }}"
      # 服务不可用告警
      - alert: ServiceDown
        expr: up{job="openclaw"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "OpenClaw 服务不可用"
          description: "OpenClaw 服务已停止响应超过 1 分钟"
      # 内存使用告警
      - alert: HighMemoryUsage
        expr: |
          process_resident_memory_bytes{job="openclaw"} 
          / 
          (1024 * 1024 * 1024) > 4
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "OpenClaw 内存使用过高"
          description: "内存使用超过 4GB，当前值: {{ $value | humanize }}B"

import logging
import requests
from typing import Optional

class AlertHandler(logging.Handler):
    """
    告警日志处理器
    
    当出现 ERROR 及以上级别的日志时，
    自动发送告警到指定渠道（飞书/钉钉/企业微信等）。
    """
    
    def __init__(
        self,
        webhook_url: str,
        level: int = logging.ERROR,
        cooldown_seconds: int = 300
    ):
        """
        初始化告警处理器
        
        Args:
            webhook_url: 告警 Webhook URL
            level: 触发告警的最低日志级别
            cooldown_seconds: 告警冷却时间（秒），避免重复告警
        """
        super().__init__(level=level)
        self.webhook_url = webhook_url
        self.cooldown_seconds = cooldown_seconds
        self._last_alert_time: dict = {}
    
    def emit(self, record: logging.LogRecord) -> None:
        """
        发送告警
        
        Args:
            record: 日志记录对象
        """
        import time
        
        # 检查冷却时间
        alert_key = f"{record.name}:{record.getMessage()[:50]}"
        now = time.time()
        
        if alert_key in self._last_alert_time:
            if now - self._last_alert_time[alert_key] < self.cooldown_seconds:
                return  # 在冷却期内，跳过
        
        self._last_alert_time[alert_key] = now
        
        # 构建告警消息
        alert_data = {
            "msg_type": "text",
            "content": {
                "text": f"【OpenClaw 告警】\n"
                        f"级别: {record.levelname}\n"
                        f"模块: {record.name}\n"
                        f"时间: {self.format_time(record)}\n"
                        f"消息: {record.getMessage()}"
            }
        }
        
        # 发送告警
        try:
            requests.post(
                self.webhook_url,
                json=alert_data,
                timeout=5
            )
        except Exception as e:
            # 告警发送失败，记录到标准错误
            import sys
            print(f"Failed to send alert: {e}", file=sys.stderr)
    
    def format_time(self, record: logging.LogRecord) -> str:
        """格式化时间"""
        from datetime import datetime
        return datetime.fromtimestamp(record.created).strftime('%Y-%m-%d %H:%M:%S')

# 使用示例
logger = logging.getLogger('openclaw')
logger.addHandler(AlertHandler(
    webhook_url='https://open.feishu.cn/open-apis/bot/v2/hook/xxx',
    level=logging.ERROR,
    cooldown_seconds=300
))