其他服务配置

grafana 系列服务配置指南

• 之前想用一种方法监听小米球和 napcat 的日志
• 当时完全采用一个命令行运行小米球，一个命令行运行 napcat，一个命令行运行主程序，一个命令行运行连接，程序间的日志不互通
• 之后把四个命令全部集中到主程序里
• 需要一种方法来捕获并统计日志和统计数据
• 先找到了 Prometheus 和 Granfana，然后顺便用 Windows Exporter 统计电脑运行情况，最后又加上了 loki 和 promtail
• 之后觉得这些东西占电脑运行内存太大了（32GB 总运存的 50%）故全部删掉了
• 如果涉及到跨平台等，又不想自己写，可以试着用一用

• 从 grafana 官网下载 grafana，可以修改 conf/defaults.ini 的 http_port = 5930 来更改运行端口
• 从 prometheus 官网下载 prometheus
• 配置 prometheus.yml

global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: "prometheus"
    static_configs:
      - targets: ["localhost:5926"]
  - job_name: "windows_exporter"
    static_configs:
        - targets: ["localhost:5924"]
  - job_name: "application_metrics"
    static_configs:
        - targets: [ "localhost:5925" ]

• 新建 prometheus.bat： prometheus.exe --config.file=prometheus.yml --web.listen-address=":5926"
• 需要在本地的 5923 配置一个 ws 连接，定时发送 SERVICE_STATUS（定义的 prometheus 指标），接收 grafana 的后端返回结果
• 从 windows-exporter 官网下载 windows-exporter
• 配置 windows-exporter.yml

collectors:
  enabled: cpu,cs,logical_disk,net,os,service,system
collector:
  service:
    include: "windows_exporter"
  scheduled_task:
    include: /Microsoft/.+
log:
  level: debug
scrape:
  timeout-margin: 0.5
telemetry:
  path: /metrics
  max-requests: 5
web:
  listen-address: ":5924"

• 新建 windows-exporter.bat： windows_exporter-0.30.0-beta.4-amd64.exe --config.file="windows_exporter.yml"
• 访问 http://localhost:5926 和 http://localhost:5924/metrics 确认 windows-exporter 启动，配置 rate(interface_tx_bytes_total)
• 从 loki 官网下载 loki
• 配置 loki.yml

auth_enabled: false

server:
  http_listen_port: 5927
  log_level: info

common:
  ring:
    instance_addr: 127.0.0.1
    kvstore:
      store: inmemory
  replication_factor: 1
  path_prefix: loki

schema_config:
  configs:
  - from: 2020-05-15
    store: tsdb
    object_store: filesystem
    schema: v13
    index:
      prefix: index_
      period: 24h

storage_config:
  filesystem:
    directory: chunks

• 新建 loki.bat： @echo off \n chcp 65001 \n loki-windows-amd64.exe -config.file=loki.yml
• 从 loki 官网下载 promtail
• 配置 promtail.yml

server:
  http_listen_port: 5928
  grpc_listen_port: 0
positions:
  filename: positions.yaml
clients:
  - url: http://localhost:5927/loki/api/v1/push
scrape_configs:
- job_name: service_logs
  static_configs:
  - targets:
      - localhost
    labels:
      job: service_logs
      __path__: ../service.log
  pipeline_stages:
    - json:
        expressions:
          level: level
          time: time
          service: service  # 从日志内容中提取 service 字段作为标签
          message: message
    - labels:
        level: level
        time: time
        service: service  # 从日志内容中提取 service 字段作为标签
        message: message
- job_name: robot_logs
  static_configs:
  - targets:
      - localhost
    labels:
      job: robot_logs
      __path__: ../robot.log
  pipeline_stages:
    - json:
        expressions:
          level: level
          time: time
          robot: robot
          event: event
          command: command
          target: target
          message: message
    - labels:
        level: level
        time: time
        robot: robot
        event: event
        command: command
        target: target
        message: message

• 新建 promtail.bat： @echo off \n chcp 65001 \n promtail-windows-amd64.exe -config.file=promtail.yml
• 安装包 pip install prometheus_client，将下方的 start_services(),monitor_services() 和 stop_services()分别放在 main 里面的 start、tasks、stop 中，将最后面的 filter 移动至 log.py，运行
• 以上所有文件夹都放置在 lrobot 文件夹同级位置
• 访问 http://localhost:3000 ，settings-data sources-add data source，选择 Prometheus 和 Loki，地址分别为 http://localhost:5926 和 http://localhost:5927
• 在仪表盘中根据 windows-exporter 配置上下行网速；设置 loki 查询面板，根据 level 和 robot 筛选；使用 web 组件与 5923 端口通信，实现指标的展示与反馈（几个服务是否在运行），可以配置节点图根据运行与否亮灭
• 图片给丢了，还挺高级的

日志格式

• 以下为自启动与代码启动（格式化后）的日志对比，日志格式转化可以在下方中找到

napcat

• 自启动
• 代码启动
• 代码启动会多出很多可能是 node.js 里面出现而在控制台里不出现的内容，所以把这些以"[]"开头的内容简单地识别进debug里（包括会在控制台里出现的[NapCat Backend] Main Process ID:22820）启动的 qq 进程 id
• 断网后会显示:12-05 14:51:56 [info] LR5921 | 账号状态变更为离线
• 可以输出相同的日志，除了从[NapCat Backend] 到NapCat Shell App Loading...中间的8行为控制台独有
• 可以识别中文，忽略了 active 那一句

xiaomiqiu

• 自启动
• 自启动会显示一个状态表，正常是 connect，断网后是 connecting
• 正常启动的输出只有几行
• 即使用跟 loki 一样的重定向方法，内部输出语句的优先级仍然高于控制台，会输出到最新日志的文件夹里
• 原因在于未在执行命令里显示将日志输出到控制台，修改后的日志输出为：
• 修改后代码启动为：
• 但当断网时不会把日志等级设为 error，需要在 log 里增加判断
• 断网时启动会直接设为 error
• 另外，这个也是 go 语言
• 简化了正则表达式，支持识别中文

windows_exporter

• 日志较少，自启动时仅仅有几条日志，然后访问 /metrics 或 /health 时只会有几条debug
• 当端口被占用时程序会直接终止且没有任何日志，需要靠 monitor 去发现
• 并且这个也是 go，创建文件比 loki 慢，loki 启动非常快导致日志文件必须从开头读，这个启动非常慢导致：
• 并且这个也是 go，创建文件比 loki 慢，loki 启动非常快导致日志文件必须从开头读，这个启动非常慢导致：
  • 必须结束后删除日志文件
  • 必须循环等待日志文件读取
• 调试完成后日志（控制台相同）

prometheus

• 自启动
• 代码启动
• 在处理 Prometheus 输出的时候发现只需要 asyncio.create_subprocess_shell 里面不加 pipe 捕获，就可以正常输出（但不能捕获） windows_exporter 的所有日志，故删除了之前的逻辑
• 然后发现 napcat 输出不加 pipe 捕获，就捕获不了日志而是直接输出到控制台
• 然后换 ai 重构了代码

grafana

• 自启动
• 代码启动
• 基本一样，但自启动有特殊格式

loki

• 自启动(比较慢)
• 代码启动
• 一样的，启动比较慢，可以使用 netstat -ano | findstr :5927和tasklist | findstr 19528来查看 loki 是否正常关闭（会过一会关闭），http://localhost:5927/metrics 和 http://localhost:5927/ready 都可以访问
• 在日志归类时发现了 loki 存在一些不含 msg 的日志：level=info ts=2024-12-09T08:03:21.0071912Z caller=metrics.go:386 org_id=fake traceID=400f7ba8831fd4c3 latency=fast query_type=stats start=2024-12-09T09:58:19+08:00 end=2024-12-09T09:59:30+08:00 start_delta=6h5m2.0071912s end_delta=6h3m51.0071912s length=1m11s duration=1.0007ms status=200 query=\"{job=\\\"logs\\\", service=~\\\"grafana\\\", level=~\\\".*\\\"}\" query_hash=3442096169 total_entries=1
• 故把 caller 后面的全部作为 msg 消息提取

promtail

• 自启动
• 代码启动
• 单独启动 promtail，不启动 loki 时会一直尝试往loki发送消息

相关代码

• 服务启动，用于启动、停止、监控（在grafana中监控）服务启停

服务相关代码

import asyncio
import subprocess
from prometheus_client import Gauge, Counter, Summary, start_http_server
from config import path, config
from log import loggers

server_logger = loggers["server"]

# exe与bat服务运行指标
SERVICE_STATUS = Gauge("service_status", "Status of the service", ["service_name"])

running_processes = {}  # 已启动服务的进程
SERVICES = {
    "xiaomiqiu": str(path.parent / "xiaomiqiu" / "xiaomiqiu.bat"),
    "windows_exporter": str(path.parent / "prometheus" / "windows_exporter.bat"),
    "prometheus": str(path.parent / "prometheus" / "prometheus.bat"),
    "promtail": str(path.parent / "loki" / "promtail.bat"),
    "loki": str(path.parent / "loki" / "loki.bat"),
    "grafana": str(path.parent / "grafana" / "grafana.bat"),
    "napcat": str(path.parent / "napcat" / "napcat.quick.bat"),
}


# 初始化 Prometheus 服务
async def Prometheus_start():
    start_http_server(5925)  # 在主线程启动 Prometheus 服务


# 启动服务
async def start_services():
    for name, path in SERVICES.items():
        if (
            name not in running_processes
            or running_processes[name].returncode is not None
        ):  # 服务未启动或已停止
            try:
                working_dir = path.parent
                process = await asyncio.create_subprocess_shell(
                    path,
                    cwd=working_dir,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                )
                running_processes[name] = process
                SERVICE_STATUS.labels(service_name=name).set(1)
                server_logger.info(
                    f"Started {name}",
                    extra={"event": "运行日志"},
                )
                asyncio.create_task(log_service(name, process))
            except Exception as e:
                SERVICE_STATUS.labels(service_name=name).set(0)
                server_logger.error(
                    f"Failed to start {name}: {e}",
                    extra={"event": "运行日志"},
                )


# 启动日志记录
async def log_service(service_name, process):
    while True:
        line = await process.stdout.readline()
        if not line:
            break
        server_logger.info(
            line.decode("utf-8").strip(),
            extra={"event": service_name},
        )
    await process.wait()
    server_logger.error(
        f"Service {service_name} has stopped.",
        extra={"event": "运行日志"},
    )


# 监控服务函数
async def monitor_services():
    while True:
        for name, process in list(running_processes.items()):
            if process.returncode is not None:  # 服务已停止
                SERVICE_STATUS.labels(service_name=name).set(0)
                running_processes.pop(name)
                server_logger.error(
                    f"Service {name} has stopped.",
                    extra={"event": "运行日志"},
                )
            else:
                SERVICE_STATUS.labels(service_name=name).set(1)
        await asyncio.sleep(60)


# 停止服务函数
async def stop_services():
    for name, process in running_processes.items():
        try:
            pid = process.pid
            await asyncio.create_subprocess_exec("taskkill", "/F", "/PID", str(pid))
            SERVICE_STATUS.labels(service_name=name).set(0)
            server_logger.info(
                f"Stopped {name}",
                extra={"event": "运行日志"},
            )
        except Exception as e:
            error_message = f"Failed to stop service: {e}"
            SERVICE_STATUS.labels(service_name=name).set(0)
            server_logger.error(
                f"Failed to stop {name}: {error_message}",
                extra={"event": "运行日志"},
            )

服务日志代码

# 需要把下面的部分整合至 log.py,输出符合 loki 格式的日志
# 正则表达式定义
LOG_FORMATS = {
    "xiaomiqiu": r"\[(?P<level>.*?)\]\t(?P<file_path>.*?)\t(?P<info>.*)",
    "windows_exporter": r"level=(?P<level>[^ ]+).*?msg=(?P<info>.*)",
    "prometheus": r"level=(?P<level>[^ ]+).*?msg=(?P<info>.*)",
    "promtail": r"level=(?P<level>[^ ]+).*?msg=(?P<info>.*)",
    "loki": r"level=(?P<level>[^ ]+).*?caller=[^ ]+ (?P<info>.*)",
    "grafana": r"level=(?P<level>[^ ]+).*?msg=(?P<info>.*)",
    "napcat": r"\[(\u001b\[\d+m(?P<level>[^\u001b]+)\u001b\[39m)\] (?P<info>.*)",
}


class ServerFilter(logging.Filter):
    """各 server 的日志过滤器"""

    def filter(self, record):
        if not record.getMessage():
            return False
        service = getattr(record, "event", "")
        if "debug1" in record.getMessage():
            record.levelname = "DEBUG"  # 设置为 DEBUG 级别
            record.msg = record.msg.replace("debug1:", "").strip()  # 去掉 debug1: 前缀
        # 更改掉xiaomiqiu以error开头的日志
        if service == "xiaomiqiu" and record.getMessage().startswith("Error"):
            record.levelname = "ERROR"
        # 设置bat中激活utf-8的日志，以及napcat以[]开头的日志为DEBUG
        if record.getMessage() == "Active code page: 65001" or (
            service == "napcat"
            and record.getMessage().startswith("[")
            and not record.getMessage().startswith("[NapCat Backend]")
        ):
            record.levelname = "DEBUG"  # 设置为 DEBUG 级别
        if service in LOG_FORMATS:
            pattern = LOG_FORMATS[service]
            match = re.search(pattern, record.getMessage())
            if match:
                record.levelname = match.group("level").upper()  # 设置为 DEBUG 级别
                record.msg = match.group("info")
        record.levelno = getattr(logging, record.levelname, logging.INFO)

        record.msg = json.dumps(
            {
                "level": record.levelname,
                "time": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
                "service": service,
                "message": record.msg,
            },
            ensure_ascii=False,
        )

        return True


class AdapterFilter(logging.Filter):
    """机器人日志过滤器"""

    EVENT_TO_LEVEL = {
        # "事件名称": "日志级别"
        # 例如： "error_event": "ERROR"
    }

    def filter(self, record):
        # 处理事件日志级别
        event = getattr(record, "event", "")
        log_level = self.EVENT_TO_LEVEL.get(event, "INFO").upper()

        record.levelname = log_level
        record.levelno = getattr(logging, log_level, logging.INFO)

        # 统一日志格式
        record.msg = json.dumps(
            {
                "level": record.levelname,
                "time": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
                "robot": getattr(record, "robot", ""),
                "event": event,
                "command": getattr(record, "command", ""),
                "target": getattr(record, "target", ""),
                "message": record.getMessage(),
            },
            ensure_ascii=False,
        )

        return True