import logging
# Gateway 服務(wù)日志器
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 日志系統(tǒng)
    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()))
    # 創(chuàng)建格式化器
    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)
    # 文件處理器（輪轉(zhuǎn)）
    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 格式化器，將日志輸出為結(jié)構(gòu)化 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
# 創(chuàng)建按大小輪轉(zhuǎn)的文件處理器
rotating_handler = RotatingFileHandler(
    filename='/var/log/openclaw/openclaw.log',
    maxBytes=10 * 1024 * 1024,  # 單文件最大 10MB
    backupCount=10,              # 保留 10 個備份文件
    encoding='utf-8'
)
# 輪轉(zhuǎn)后的文件命名規(guī)則：
# openclaw.log       (當前日志)
# openclaw.log.1     (最近的備份)
# openclaw.log.2     (次近的備份)
# ...
# openclaw.log.10    (最老的備份)

from logging.handlers import TimedRotatingFileHandler
# 創(chuàng)建按時間輪轉(zhuǎn)的文件處理器
timed_handler = TimedRotatingFileHandler(
    filename='/var/log/openclaw/openclaw.log',
    when='midnight',    # 每天午夜輪轉(zhuǎn)
    interval=1,         # 間隔 1 天
    backupCount=30,     # 保留 30 天日志
    encoding='utf-8'
)
# 設(shè)置文件名后綴格式
timed_handler.suffix = '%Y-%m-%d'
# 輪轉(zhuǎn)后的文件命名規(guī)則：
# 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 地址
    - 自定義敏感詞
    """
    
    # 預(yù)定義的敏感信息匹配模式
    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，允許日志通過，但會修改消息內(nèi)容
        """
        # 對日志消息進行脫敏處理
        record.msg = self._sanitize(str(record.msg))
        
        # 對參數(shù)進行脫敏處理
        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
        
        # 應(yīng)用預(yù)定義模式
        for pattern_name, pattern in self.PATTERNS.items():
            result = pattern.sub(self._get_replacement(pattern_name), result)
        
        # 應(yīng)用自定義敏感詞
        for word in self.custom_patterns:
            result = result.replace(word, '***REDACTED***')
        
        return result
    
    def _get_replacement(self, pattern_name: str) -> str:
        """
        根據(jù)模式類型返回脫敏后的替換文本
        
        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              # 開啟調(diào)試模式
    log_level: DEBUG           # 日志級別
    trace_requests: true       # 追蹤請求
    profile_performance: true  # 性能分析
    memory_debug: false        # 內(nèi)存調(diào)試（影響性能）
  logging:
    level: DEBUG
    format: detailed           # detailed / json / compact
    output:
      - console
      - file
    file:
      path: /var/log/openclaw/debug.log
      max_size: 50MB
      backup_count: 3

# 開啟調(diào)試模式
export OPENCLAW_DEBUG=true
export OPENCLAW_LOG_LEVEL=DEBUG

# 開啟請求追蹤
export OPENCLAW_TRACE_REQUESTS=true

# 開啟性能分析
export OPENCLAW_PROFILE=true

# 啟動 OpenClaw
openclaw gateway start

# 開啟調(diào)試模式啟動 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. 檢查日志文件權(quán)限
ls -la /var/log/openclaw/
# 4. 以調(diào)試模式啟動查看詳細錯誤
openclaw gateway start --debug

# 1. 查看最近的錯誤日志
tail -f /var/log/openclaw/openclaw.log | grep ERROR
# 2. 檢查 AI 模型 API 響應(yīng)時間
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. 檢查并發(fā)連接數(shù)
netstat -an | grep 18789 | wc -l
# 4. 開啟性能分析
export OPENCLAW_PROFILE=true
openclaw gateway restart

# 1. 查看進程內(nèi)存使用
ps aux | grep openclaw
# 2. 開啟內(nèi)存調(diào)試
export OPENCLAW_MEMORY_DEBUG=true
openclaw gateway restart
# 3. 分析內(nèi)存日志
grep "Memory usage" /var/log/openclaw/openclaw.log
# 4. 檢查是否有內(nèi)存泄漏
valgrind --leak-check=full openclaw gateway start

import logging
from logging.handlers import QueueHandler, QueueListener
from queue import Queue
def setup_async_logging():
    """
    配置異步日志系統(tǒng)，提升性能
    使用隊列將日志寫入操作從主線程分離，
    避免日志 I/O 阻塞業(yè)務(wù)處理。
    """
    # 創(chuàng)建日志隊列
    log_queue = Queue(maxsize=10000)
    # 創(chuàng)建隊列處理器（主線程使用）
    queue_handler = QueueHandler(log_queue)
    # 創(chuàng)建實際的文件處理器（后臺線程使用）
    file_handler = logging.FileHandler('/var/log/openclaw/openclaw.log')
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
    ))
    # 創(chuàng)建隊列監(jiān)聽器
    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()
# 應(yīng)用退出時停止監(jiān)聽器
# listener.stop()

import logging
from typing import Optional
class DynamicLogLevelManager:
    """
    動態(tài)日志級別管理器
    支持運行時動態(tài)調(diào)整日志級別，無需重啟服務(wù)。
    可通過 API 或配置文件觸發(fā)調(diào)整。
    """
    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:
        """
        設(shè)置日志級別
        Args:
            logger_name: 日志器名稱，支持通配符
            level: 目標級別 (DEBUG/INFO/WARNING/ERROR/CRITICAL)
        Returns:
            是否設(shè)置成功
        """
        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')
# 臨時開啟調(diào)試模式
manager.set_level('openclaw.*', 'DEBUG')
# 排查完成后恢復(fù)
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 }}"
      # 響應(yīng)時間告警
      - alert: SlowResponse
        expr: |
          histogram_quantile(0.95, 
            rate(openclaw_request_duration_seconds_bucket[5m])
          ) > 5
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "OpenClaw 響應(yīng)時間過長"
          description: "P95 響應(yīng)時間超過 5 秒，當前值: {{ $value | humanizeDuration }}"
      # 服務(wù)不可用告警
      - alert: ServiceDown
        expr: up{job="openclaw"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "OpenClaw 服務(wù)不可用"
          description: "OpenClaw 服務(wù)已停止響應(yīng)超過 1 分鐘"
      # 內(nèi)存使用告警
      - alert: HighMemoryUsage
        expr: |
          process_resident_memory_bytes{job="openclaw"} 
          / 
          (1024 * 1024 * 1024) > 4
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "OpenClaw 內(nèi)存使用過高"
          description: "內(nèi)存使用超過 4GB，當前值: {{ $value | humanize }}B"

import logging
import requests
from typing import Optional

class AlertHandler(logging.Handler):
    """
    告警日志處理器
    
    當出現(xiàn) ERROR 及以上級別的日志時，
    自動發(fā)送告警到指定渠道（飛書/釘釘/企業(yè)微信等）。
    """
    
    def __init__(
        self,
        webhook_url: str,
        level: int = logging.ERROR,
        cooldown_seconds: int = 300
    ):
        """
        初始化告警處理器
        
        Args:
            webhook_url: 告警 Webhook URL
            level: 觸發(fā)告警的最低日志級別
            cooldown_seconds: 告警冷卻時間（秒），避免重復(fù)告警
        """
        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:
        """
        發(fā)送告警
        
        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  # 在冷卻期內(nèi)，跳過
        
        self._last_alert_time[alert_key] = now
        
        # 構(gòu)建告警消息
        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()}"
            }
        }
        
        # 發(fā)送告警
        try:
            requests.post(
                self.webhook_url,
                json=alert_data,
                timeout=5
            )
        except Exception as e:
            # 告警發(fā)送失敗，記錄到標準錯誤
            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
))