194 lines
11 KiB
Markdown
194 lines
11 KiB
Markdown
# Backend 存储
|
||
|
||
实现位于 `MoShiro.Backend.Storage.Sqlite`，负责实现 `Backend.Application` 中定义的端口接口，**只做读写，不含业务逻辑**。降采样和分析由 Application 调用 Analysis 完成。
|
||
|
||
## 1. 目录布局
|
||
|
||
```text
|
||
data/
|
||
├─ moshiro.db # SQLite：配置、身份、原始块索引、指标、测试结果
|
||
└─ raw/
|
||
└─ 2026-09-27/ # 按 UTC 日期分目录，保留期满整目录删除
|
||
├─ agt_01J…/ # 每条线路或每个 Agent 一个目录
|
||
│ ├─ udp_echo.seg # 段文件：同一天、同一 Kind 的块依次追加
|
||
│ ├─ tcp_echo.seg
|
||
│ └─ tsk_…_icmp_echo.seg # 目标任务
|
||
└─ tests/tst_01J…/phase0.seg …
|
||
```
|
||
|
||
- **段文件**：把收到的 RawChunk 原样追加写入（块本身已经带 crc，并且自描述），SQLite 只记录 `(segment, offset, length)`。每 10 秒一块也不会产生海量小文件。
|
||
- **写入顺序**：追加写入 → `fsync`（每批合并执行一次）→ 插入索引 → 由收件箱确认（Fleet 回复 `fleet.outbox.ack`）。启动时截断没有索引记录的段文件尾部。
|
||
- **去重**：`chunk_id` 已存在时，直接回复 ACK。
|
||
|
||
## 2. SQLite 配置
|
||
|
||
- `journal_mode=WAL`、`synchronous=NORMAL`、`foreign_keys=ON`、`busy_timeout=5000`。
|
||
- **单写入者**：所有写操作经过一个有界 `Channel`，由一个后台循环执行；读操作使用独立的只读连接池。
|
||
- 时间统一存储为 Unix 毫秒 `INTEGER`（UTC）；表结构版本记录在 `PRAGMA user_version`，迁移脚本为内置 SQL。
|
||
|
||
## 3. 表结构
|
||
|
||
`agents`、`join_tokens` 以及公钥和状态版本字段，同时作为 Fleet 的 `IFleetStore` 存储（由 `MoShiro.Backend.Storage.Sqlite` 实现），和 MoShiro 的其他数据放在同一个 `moshiro.db` 中，由同一个单写入者写入。CI 签名的发布清单存在 `settings` 表里（`release.<组件>.manifest`、`release.<组件>.signature`），面板只原样保存和转发。
|
||
|
||
```sql
|
||
-- 身份与注册 -------------------------------------------------------------
|
||
CREATE TABLE agents (
|
||
id TEXT PRIMARY KEY, -- agt_<ULID>
|
||
name TEXT NOT NULL UNIQUE,
|
||
status TEXT NOT NULL, -- pending | active | disabled
|
||
pubkey BLOB NOT NULL, -- ECDSA P-256 身份公钥
|
||
tls_spki BLOB NOT NULL, -- 面向 Probe 的证书指纹（由 tls_pubkey 算出）
|
||
tls_pubkey BLOB, -- 面向 Probe 的证书公钥，Agent 在 hello 里上报；Probe 用它固定证书、派生敲门密钥
|
||
public_host TEXT NOT NULL, -- 自动识别（Agent 上报的公网地址），手填后标记 public_host_manual，不再被覆盖
|
||
probe_tcp_port INTEGER NOT NULL, probe_udp_ports TEXT NOT NULL,
|
||
labels TEXT NOT NULL DEFAULT '[]',
|
||
public_visible INTEGER NOT NULL DEFAULT 1, -- 是否在 public API 中展示
|
||
kinds TEXT, agent_version TEXT, local_limits TEXT, -- 来自 fleet.hello
|
||
task_version INTEGER NOT NULL DEFAULT 0, applied_task_version INTEGER NOT NULL DEFAULT 0,
|
||
created_at INTEGER NOT NULL, last_seen_at INTEGER, row_version INTEGER NOT NULL DEFAULT 0
|
||
);
|
||
|
||
CREATE TABLE join_tokens (
|
||
id TEXT PRIMARY KEY, token_sha256 BLOB NOT NULL UNIQUE, note TEXT,
|
||
labels TEXT NOT NULL DEFAULT '[]', auto_approve INTEGER NOT NULL,
|
||
max_uses INTEGER NOT NULL, uses INTEGER NOT NULL DEFAULT 0,
|
||
expires_at INTEGER NOT NULL, revoked_at INTEGER, created_at INTEGER NOT NULL
|
||
);
|
||
|
||
CREATE TABLE host_samples ( -- 主机资源占用，每个 Agent 每分钟一行（schema v3）
|
||
agent_id TEXT NOT NULL, at INTEGER NOT NULL,
|
||
cpu REAL, mem_used INTEGER, mem_total INTEGER, disk_used INTEGER, disk_total INTEGER,
|
||
net_rx_bps INTEGER, net_tx_bps INTEGER, uptime INTEGER,
|
||
PRIMARY KEY (agent_id, at)
|
||
) WITHOUT ROWID;
|
||
|
||
CREATE TABLE probes ( -- 每个 Probe 身份一行（吊销只设 revoked_at）
|
||
id TEXT PRIMARY KEY, name TEXT NOT NULL, pubkey BLOB NOT NULL,
|
||
created_at INTEGER NOT NULL, revoked_at INTEGER
|
||
);
|
||
|
||
CREATE TABLE users (
|
||
id TEXT PRIMARY KEY, username TEXT NOT NULL UNIQUE,
|
||
password_hash BLOB NOT NULL, password_salt BLOB NOT NULL, iterations INTEGER NOT NULL,
|
||
created_at INTEGER NOT NULL
|
||
);
|
||
CREATE TABLE sessions (
|
||
id_sha256 BLOB PRIMARY KEY, user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||
csrf_token BLOB NOT NULL, created_at INTEGER NOT NULL, expires_at INTEGER NOT NULL
|
||
);
|
||
CREATE TABLE api_tokens (
|
||
id TEXT PRIMARY KEY, name TEXT NOT NULL, scope TEXT NOT NULL, -- read | admin
|
||
token_sha256 BLOB NOT NULL UNIQUE, created_at INTEGER NOT NULL, last_used_at INTEGER, revoked_at INTEGER
|
||
);
|
||
CREATE TABLE settings (key TEXT PRIMARY KEY, value TEXT NOT NULL); -- 如 site.public = true
|
||
|
||
-- 任务 ------------------------------------------------------------------
|
||
CREATE TABLE tasks (
|
||
id TEXT PRIMARY KEY, -- tsk_<ULID>
|
||
scope TEXT NOT NULL, -- line | target
|
||
agent_id TEXT REFERENCES agents(id) ON DELETE CASCADE, -- 为 NULL 时，是适用于所有 Agent 的 line 模板
|
||
base_task_id TEXT REFERENCES tasks(id) ON DELETE CASCADE, -- 针对某个 Agent 覆盖模板时，指向模板
|
||
kind TEXT NOT NULL,
|
||
target TEXT, -- 仅 target 任务
|
||
params TEXT NOT NULL DEFAULT '{}',
|
||
schedule TEXT NOT NULL, -- continuous | every:<分钟>
|
||
enabled INTEGER NOT NULL DEFAULT 1,
|
||
updated_at INTEGER NOT NULL, row_version INTEGER NOT NULL DEFAULT 0
|
||
);
|
||
|
||
CREATE TABLE tests ( -- 按需测试（once）
|
||
id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, plan TEXT NOT NULL, params TEXT NOT NULL,
|
||
trigger TEXT NOT NULL, -- api | schedule:<task_id>
|
||
status TEXT NOT NULL, -- queued | running | succeeded | failed | aborted
|
||
error_code TEXT, requested_at INTEGER NOT NULL, started_at INTEGER, finished_at INTEGER
|
||
);
|
||
CREATE INDEX ix_tests_agent ON tests (agent_id, requested_at);
|
||
|
||
-- 原始数据索引 ------------------------------------------------------------
|
||
CREATE TABLE raw_segments (
|
||
id INTEGER PRIMARY KEY, day TEXT NOT NULL, path TEXT NOT NULL UNIQUE, bytes INTEGER NOT NULL
|
||
);
|
||
CREATE TABLE raw_chunks (
|
||
chunk_id TEXT PRIMARY KEY, segment_id INTEGER NOT NULL, offset INTEGER NOT NULL, length INTEGER NOT NULL,
|
||
agent_id TEXT NOT NULL, series TEXT NOT NULL, -- line | target:<task_id>
|
||
kind TEXT NOT NULL, task_id TEXT, test_id TEXT, phase INTEGER,
|
||
start_ts INTEGER NOT NULL, duration_ms INTEGER NOT NULL, partial INTEGER NOT NULL
|
||
) WITHOUT ROWID;
|
||
CREATE INDEX ix_raw_series ON raw_chunks (agent_id, series, start_ts);
|
||
CREATE INDEX ix_raw_test ON raw_chunks (test_id) WHERE test_id IS NOT NULL;
|
||
|
||
-- 指标（通用存储：每个桶一行，内容为 指标id → 累加器状态 的映射）----------------
|
||
-- 分层：metric_10s（48 小时）、metric_1m（30 天）、metric_5m（365 天）、metric_1h（永久）
|
||
-- 另有 1s 精度的内存热层（最近 1 小时，不落盘），见 query.md 第 3 节
|
||
CREATE TABLE metric_1m ( -- 60 s
|
||
agent_id TEXT NOT NULL, series TEXT NOT NULL, window_start INTEGER NOT NULL,
|
||
partial INTEGER NOT NULL,
|
||
metrics BLOB NOT NULL, -- AccumulatorCodec 编码的 {metric_id: state}
|
||
PRIMARY KEY (agent_id, series, window_start)
|
||
) WITHOUT ROWID;
|
||
-- metric_10s 结构相同（由流式路径直接写入，不经过降采样）
|
||
-- metric_5m / metric_1h 结构相同（window_start 换成 bucket_start，另加 windows 计数）
|
||
|
||
CREATE TABLE test_results (
|
||
test_id TEXT PRIMARY KEY, metrics BLOB NOT NULL, series BLOB -- series 为逐秒或逐档的曲线
|
||
) WITHOUT ROWID;
|
||
|
||
CREATE TABLE metric_versions (metric_id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL,
|
||
recompute_from INTEGER, recompute_cursor INTEGER);
|
||
|
||
CREATE TABLE analysis_queue (
|
||
agent_id TEXT NOT NULL, series TEXT NOT NULL, window_start INTEGER NOT NULL,
|
||
not_before INTEGER NOT NULL, reason TEXT NOT NULL, -- new | late_chunk | recompute
|
||
PRIMARY KEY (agent_id, series, window_start)
|
||
) WITHOUT ROWID;
|
||
CREATE TABLE rollup_dirty (level INTEGER NOT NULL, agent_id TEXT NOT NULL, series TEXT NOT NULL,
|
||
bucket_start INTEGER NOT NULL, PRIMARY KEY (level, agent_id, series, bucket_start)) WITHOUT ROWID;
|
||
|
||
-- 事件 ------------------------------------------------------------------
|
||
CREATE TABLE events (id INTEGER PRIMARY KEY, agent_id TEXT, kind TEXT NOT NULL,
|
||
at INTEGER NOT NULL, detail TEXT);
|
||
CREATE INDEX ix_events_time ON events (at);
|
||
```
|
||
|
||
**要点：新增 Kind 或指标都不需要改表结构。** 指标是 `metric_*.metrics` 里的一个键；Kind 只是 `raw_chunks.kind` 的一个取值。
|
||
|
||
## 4. 保留
|
||
|
||
| 数据 | 默认保留 |
|
||
| --- | --- |
|
||
| 持续任务的原始段文件 | **30 天** |
|
||
| 按需测试的原始段文件 | 90 天 |
|
||
| 热层（1s，内存） | 最近 1 小时 |
|
||
| `metric_10s` | 48 小时 |
|
||
| `metric_1m` | 30 天 |
|
||
| `metric_5m` | 365 天 |
|
||
| `metric_1h`、`tests`、`test_results` | 永久 |
|
||
| `events` | 365 天 |
|
||
|
||
保留期在管理端的设置里改，对应的键：`retention.raw_days`、`retention.test_raw_days`、`retention.metric_10s_hours`、`retention.metric_1m_days`、`retention.metric_5m_days`、`retention.events_days`。
|
||
|
||
降采样由 Application 中的 `Downsampler` 负责：读取下一级的累加器状态，用 `AccumulatorCodec.Merge` 合并后写回。所有保留期都可以配置。
|
||
|
||
## 5. 容量估算
|
||
|
||
以 15 个 Agent、默认模板（`udp_echo` 10 Hz + `tcp_echo` 20 Hz）为例：
|
||
|
||
| 数据 | 估算 |
|
||
| --- | --- |
|
||
| 原始数据 | 每条线路约 7 MB/天（udp_echo）+ 14 MB/天（tcp_echo 及内核采样）≈ 21 MB/天；15 条线路约 0.3 GB/天，30 天约 **9.5 GB** |
|
||
| 按需测试 | 一次 `hy2_ladder` 约 1.5 MB；每条线路每天 4 次，90 天约 8 GB |
|
||
| SQLite | 指标窗口和降采样约 1～2 GB/年 |
|
||
|
||
写入量约每秒 3 块，单写入者绰绰有余。M5 阶段会用模拟数据验证这些估算。
|
||
|
||
## 6. 查询
|
||
|
||
时间段、自动步长、数据源选择、现场合并和降级规则见 [query.md](query.md)。
|
||
|
||
## 7. 与实现的差异（以 `Schema.cs` 为准）
|
||
|
||
- `agents`：`name` 不再 `UNIQUE`，改为“已建立（非 pending）节点之间唯一”的部分唯一索引，pending 节点可以重名；新增 `task_hash`（有效任务集的哈希，用来判断版本是否要递增）、`rejected`（节点上报的拒绝原因）、`meta`（注册元数据）。
|
||
- `tests`：新增 `phase`、`phase_count`；`test_results` 的 `metrics` / `series` 保存 JSON。
|
||
- `analysis_queue`：新增 `seq`，每次入队递增；完成时只删除序号没变的项，处理期间到达的迟到块不会丢更新。
|
||
- `rollup_dirty`：先处理低层（5 分钟）再处理高层（1 小时）。
|