feat: 消耗语料优化：宿主真实日志挂载 + 合成语料扩容 + 游标去重

- docker-compose 只读挂载 /var/log:/host-var-log，LOG_SOURCES 追加宿主路径（真实日志优先）
- 合成语料 5000→30000 行、模板 8→28 种、主机 6→12、覆盖 24h（行间隔 ≈86400/count）
- generate_dynamic_payload 随机起点改游标连续窗口：单次运行内零重复（时间轴随请求推进），
  耗尽后环绕（跨请求重复可接受，跨天语料重建）
- 验证：模板特征 20/20 命中、12 主机、6 窗口零重叠、payload 长度合规

docker-compose.yml

@@ -11,6 +11,8 @@ services:
       - ./volc_limit_sync.py:/app/volc_limit_sync.py
       # 自动消耗任务（每日 23:00 由 volc_limit_sync 调度触发）
       - ./volc_consume.py:/app/volc_consume.py
+      # 消耗任务语料：宿主真实系统日志（只读）
+      - /var/log:/host-var-log:ro  
     command:
       - "--config=/app/config.yaml"
     ports:

volc_consume.py

@@ -67,14 +67,23 @@ TASK_PROMPTS = [
     {"sys": "你是性能优化顾问。", "user": "根据以下日志评估系统负载特征，指出瓶颈迹象。"},
 ]
 
-# 日志源候选（容器内大概率不存在，由合成兜底）
+# 日志源候选：宿主挂载（/host-var-log，docker-compose 只读挂载 /var/log）优先，
+# 容器内 /var/log 兜底；均不存在时由合成语料兜底
 LOG_SOURCES = [
+    "/host-var-log/syslog",
+    "/host-var-log/messages",
+    "/host-var-log/kern.log",
+    "/host-var-log/auth.log",
+    "/host-var-log/journal.log",
     "/var/log/syslog",
     "/var/log/messages",
     "/var/log/kern.log",
     "/var/log/auth.log",
 ]
 
+# 单次运行内的语料游标（来源 -> 下一窗口起点），避免连续请求取到重复片段
+_CURSOR = {}
+
 # 敏感信息脱敏规则：(正则, 替换值)
 _SENSITIVE_PATTERNS = [
     (re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"), "x.x.x.x"),
@@ -94,34 +103,81 @@ def _sanitize_line(line):
     return line
 
 
-def _synthetic_lines(count=5000):
+def _synthetic_lines(count=30000):
     """
     生成时间线连续、格式多样的合成日志（语料不足时的兜底）。
-    从约 2 小时前开始，每行间隔约 1.5 秒，模拟真实系统日志流。
+    覆盖最近 24 小时（count 行均匀分布），多主机多风格，模拟真实系统日志流。
     """
-    hosts = ["web-01", "web-02", "api-01", "db-01", "cache-01", "gateway-01"]
+    hosts = [
-    templates = [
+        "web-01",
-        "kernel: [TCP] Possible SYN flooding on port {port}. Sending cookies.",
+        "web-02",
-        "sshd: Accepted publickey for deploy from x.x.x.x port {port} ssh2",
+        "web-03",
-        "nginx: worker process, pid={pid}, GET /api/v2/status HTTP/1.1 200 0.00{ms}",
+        "api-01",
-        "systemd: Started Daily apt download and clean activities.",
+        "api-02",
-        "python3[{pid}]: heartbeat ok, workers={n}, queue_depth={q}",
+        "db-01",
-        "postgres[{pid}]: LOG: checkpoint complete: wrote {n} buffers",
+        "cache-01",
-        "redis-server[{pid}]: Saving DB to disk complete",
+        "gateway-01",
-        "kernel: EXT4-fs (sda1): mounted filesystem with ordered data mode",
+        "worker-01",
+        "k8s-node-01",
+        "mq-01",
+        "log-01",
     ]
-    start_ts = datetime.now().timestamp() - count * 1.5
+    templates = [
+        # 内核 / 网络
+        "kernel: [TCP] Possible SYN flooding on port {port}. Sending cookies.",
+        "kernel: EXT4-fs (sda1): mounted filesystem with ordered data mode",
+        "kernel: nf_conntrack: table full, dropping packet.",
+        "kernel: br0: port 2(veth{hash}) entered blocking state",
+        "NetworkManager[{pid}]: <info> device (eth0): state change: activated -> connected",
+        # 系统 / systemd / 安全
+        "systemd[{pid}]: Started Daily apt download and clean activities.",
+        "systemd[{pid}]: Stopped User Manager for UID 1000.",
+        "systemd[{pid}]: Reloading nginx.service.",
+        'cron[{pid}]: (root) CMD (find /var/log -name "*.gz" -mtime +7 -delete)',
+        "sudo: pam_unix(sudo:session): session opened for user deploy by (uid=0)",
+        "sshd: Accepted publickey for deploy from x.x.x.x port {port} ssh2",
+        "sshd[{pid}]: Failed password for invalid user admin from {ip} port {port} ssh2",
+        # 应用
+        "nginx: worker process, pid={pid}, GET /api/v2/status HTTP/1.1 200 0.00{ms}",
+        'nginx: {ip} - - ["GET /v1/users HTTP/1.1"] 200 512',
+        "python3[{pid}]: heartbeat ok, workers={n}, queue_depth={q}",
+        "gunicorn[{pid}]: POST /api/chat 200 {ms}ms",
+        "celery[{pid}]: Task mail.send[abc{hash}] succeeded in 12.3s",
+        "node[{pid}]: memory usage 62%, heap {heap}MB, gc paused {gc}ms",
+        "app[{pid}]: slow query detected: SELECT * FROM metrics WHERE ts > now() - interval '1h'",
+        # 数据库 / 中间件
+        "postgres[{pid}]: LOG: checkpoint complete: wrote {n} buffers ({p}%)",
+        "postgres[{pid}]: LOG: autovacuum: processing database postgres",
+        "redis-server[{pid}]: Saving DB to disk complete",
+        "redis-server[{pid}]: 100 changes in 60 seconds. Saving...",
+        "mysql[{pid}]: [Note] InnoDB: Buffer pool(s) load completed",
+        "rabbitmq[{pid}]: connection from {ip}:{port} opened",
+        # 日志轮转 / 采集
+        "logrotate: rotating /var/log/nginx/access.log, log->access.log.1",
+        "rsyslogd: action 'action 18' resumed",
+        "fluent-bit[{pid}]: [info] tail input plugin started",
+    ]
+    # 24 小时覆盖，行间隔 ≈ 86400/count 秒
+    step = 86400.0 / count
+    start_ts = datetime.now().timestamp() - 86400.0
     lines = []
     for i in range(count):
-        ts = datetime.fromtimestamp(start_ts + i * 1.5).strftime("%b %d %H:%M:%S")
+        ts = datetime.fromtimestamp(start_ts + i * step).strftime("%b %d %H:%M:%S")
         host = random.choice(hosts)
         tpl = random.choice(templates)
         msg = tpl.format(
             port=random.randint(1024, 65535),
-            pid=random.randint(1000, 9999),
+            pid=random.randint(500, 9999),
             n=random.randint(10, 500),
             q=random.randint(0, 200),
-            ms=random.randint(1, 9),
+            ms=random.randint(1, 99),
+            hash="%04x" % random.randint(0, 0xFFFF),
+            a=random.randint(0, 255),
+            b=random.randint(0, 255),
+            heap=random.randint(200, 900),
+            gc=random.randint(1, 60),
+            p=random.randint(1, 100),
+            ip=f"10.0.{random.randint(0, 255)}.{random.randint(1, 254)}",
         )
         lines.append(f"{ts} {host} {msg}")
     return lines
@@ -152,9 +208,12 @@ def load_base_logs():
     total_lines = sum(len(v) for v in source_groups.values())
     if total_lines < 1000:
         vls.log("📄 本地日志不足，使用合成日志补齐语料")
-        source_groups["synthetic"] = _synthetic_lines(5000)
+        source_groups["synthetic"] = _synthetic_lines(30000)
         order.append("synthetic")
 
+    # 每次运行重置语料游标（新语料从头取窗口，跨请求时间轴推进）
+    _CURSOR.clear()
+
     # 去除相邻重复行（日志轮转/重复打印的常见现象）
     for key in source_groups:
         deduped = []
@@ -179,8 +238,14 @@ def generate_dynamic_payload(source_groups, order):
     source_key = random.choice(available)
     source_lines = source_groups[source_key]
 
-    window_size = min(len(source_lines), 5000)
+    # 游标取窗口：单次运行内连续前进，避免重复片段，且时间轴随请求推进
-    start_idx = random.randint(0, max(0, len(source_lines) - window_size))
+    #（语料按时间排序；耗尽后环绕，跨请求重复可接受，跨天语料重建）
+    total = len(source_lines)
+    window_size = min(total, 5000)
+    start_idx = _CURSOR.get(source_key, 0)
+    if start_idx + window_size > total:
+        start_idx = 0
+    _CURSOR[source_key] = (start_idx + window_size) % total
     sampled = source_lines[start_idx : start_idx + window_size]
 
     content_lines = []