From ee1245621d4b2651d0b15d446f6175c00cae982d Mon Sep 17 00:00:00 2001 From: vibe-code Date: Tue, 11 Aug 2026 15:17:52 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B6=88=E8=80=97=E8=AF=AD=E6=96=99?= =?UTF-8?q?=E4=BC=98=E5=8C=96=EF=BC=9A=E5=AE=BF=E4=B8=BB=E7=9C=9F=E5=AE=9E?= =?UTF-8?q?=E6=97=A5=E5=BF=97=E6=8C=82=E8=BD=BD=20+=20=E5=90=88=E6=88=90?= =?UTF-8?q?=E8=AF=AD=E6=96=99=E6=89=A9=E5=AE=B9=20+=20=E6=B8=B8=E6=A0=87?= =?UTF-8?q?=E5=8E=BB=E9=87=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 4 +- volc_consume.py | 105 ++++++++++++++++++++++++++++++++++++--------- 2 files changed, 88 insertions(+), 21 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 20a7677..71e7dba 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -10,7 +10,9 @@ services: # 【关键】挂载自定义脚本到容器根目录 - ./volc_limit_sync.py:/app/volc_limit_sync.py # 自动消耗任务（每日 23:00 由 volc_limit_sync 调度触发） - - ./volc_consume.py:/app/volc_consume.py + - ./volc_consume.py:/app/volc_consume.py + # 消耗任务语料：宿主真实系统日志（只读） + - /var/log:/host-var-log:ro command: - "--config=/app/config.yaml" ports: diff --git a/volc_consume.py b/volc_consume.py index 248303a..88f04b7 100644 --- a/volc_consume.py +++ b/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"] - templates = [ - "kernel: [TCP] Possible SYN flooding on port {port}. Sending cookies.", - "sshd: Accepted publickey for deploy from x.x.x.x port {port} ssh2", - "nginx: worker process, pid={pid}, GET /api/v2/status HTTP/1.1 200 0.00{ms}", - "systemd: Started Daily apt download and clean activities.", - "python3[{pid}]: heartbeat ok, workers={n}, queue_depth={q}", - "postgres[{pid}]: LOG: checkpoint complete: wrote {n} buffers", - "redis-server[{pid}]: Saving DB to disk complete", - "kernel: EXT4-fs (sda1): mounted filesystem with ordered data mode", + hosts = [ + "web-01", + "web-02", + "web-03", + "api-01", + "api-02", + "db-01", + "cache-01", + "gateway-01", + "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}]: 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 = []