221 lines
11 KiB
C#
221 lines
11 KiB
C#
using Fleet.Core;
|
|
using Microsoft.Data.Sqlite;
|
|
using MoShiro.Backend.Analysis;
|
|
using MoShiro.Backend.Application;
|
|
|
|
namespace MoShiro.Backend.Storage.Sqlite;
|
|
|
|
/// <summary>指标窗口的通用存储：每个桶一行，内容是 <c>指标 id → 累加器状态</c>（对本层不透明，只在合并和补丁时解码）。</summary>
|
|
public sealed class SqliteMetricStore : IMetricStore
|
|
{
|
|
private readonly SqliteDatabase _db;
|
|
|
|
public SqliteMetricStore(SqliteDatabase db)
|
|
{
|
|
_db = db;
|
|
}
|
|
|
|
public ValueTask UpsertAsync(MetricLevel level, IReadOnlyList<MetricRow> rows, CancellationToken ct) =>
|
|
_db.ExecuteAsync(c =>
|
|
{
|
|
foreach (MetricRow row in rows)
|
|
{
|
|
Upsert(c, level, row, row.Metrics);
|
|
}
|
|
}, ct);
|
|
|
|
public ValueTask MergeAsync(MetricLevel level, IReadOnlyList<MetricRow> rows, CancellationToken ct) =>
|
|
_db.ExecuteAsync(c =>
|
|
{
|
|
foreach (MetricRow row in rows)
|
|
{
|
|
byte[]? existing = Load(c, level, row);
|
|
if (existing is null || !AccumulatorCodec.Decode(existing).TryGet(out MetricSet? merged, out _))
|
|
{
|
|
Upsert(c, level, row, row.Metrics);
|
|
continue;
|
|
}
|
|
|
|
if (AccumulatorCodec.Decode(row.Metrics).TryGet(out MetricSet? incoming, out _))
|
|
{
|
|
merged.Merge(incoming);
|
|
}
|
|
|
|
Upsert(c, level, row with { Partial = row.Partial || IsPartial(c, level, row) }, AccumulatorCodec.Encode(merged));
|
|
}
|
|
}, ct);
|
|
|
|
public ValueTask PatchAsync(MetricLevel level, IReadOnlyList<MetricRow> rows, IReadOnlySet<string> replaceMetricIds, CancellationToken ct) =>
|
|
_db.ExecuteAsync(c =>
|
|
{
|
|
foreach (MetricRow row in rows)
|
|
{
|
|
byte[]? existing = Load(c, level, row);
|
|
MetricSet target = new();
|
|
if (existing is not null && AccumulatorCodec.Decode(existing).TryGet(out MetricSet? current, out _))
|
|
{
|
|
foreach ((string id, Accumulator acc) in current.Items)
|
|
{
|
|
if (!replaceMetricIds.Contains(id))
|
|
{
|
|
target.Set(id, acc);
|
|
}
|
|
}
|
|
}
|
|
|
|
if (AccumulatorCodec.Decode(row.Metrics).TryGet(out MetricSet? fresh, out _))
|
|
{
|
|
foreach ((string id, Accumulator acc) in fresh.Items)
|
|
{
|
|
if (replaceMetricIds.Contains(id))
|
|
{
|
|
target.Set(id, acc);
|
|
}
|
|
}
|
|
}
|
|
|
|
Upsert(c, level, row, AccumulatorCodec.Encode(target));
|
|
}
|
|
}, ct);
|
|
|
|
public ValueTask<IReadOnlyList<MetricRow>> ReadAsync(MetricLevel level, NodeId agentId, string series, long fromMs, long toMs, CancellationToken ct) =>
|
|
_db.ReadAsync<IReadOnlyList<MetricRow>>(
|
|
c => c.Query(
|
|
$"SELECT agent_id, series, window_start, partial, metrics FROM {level.TableName()} WHERE agent_id = @a AND series = @s AND window_start >= @from AND window_start < @to ORDER BY window_start",
|
|
ReadRow, ("@a", agentId.Value), ("@s", series), ("@from", fromMs), ("@to", toMs)),
|
|
ct);
|
|
|
|
public ValueTask<long?> EarliestAsync(MetricLevel level, NodeId agentId, string series, CancellationToken ct) =>
|
|
_db.ReadAsync(
|
|
c => c.Query($"SELECT MIN(window_start) FROM {level.TableName()} WHERE agent_id = @a AND series = @s", r => r.NullableLong(0), ("@a", agentId.Value), ("@s", series)).FirstOrDefault(),
|
|
ct);
|
|
|
|
public ValueTask<long?> LatestAsync(MetricLevel level, NodeId agentId, string series, CancellationToken ct) =>
|
|
_db.ReadAsync(
|
|
c => c.Query($"SELECT MAX(window_start) FROM {level.TableName()} WHERE agent_id = @a AND series = @s", r => r.NullableLong(0), ("@a", agentId.Value), ("@s", series)).FirstOrDefault(),
|
|
ct);
|
|
|
|
public ValueTask<IReadOnlyList<MetricRow>> ReadAgentsAsync(MetricLevel level, IReadOnlyCollection<NodeId> agentIds, string series, long fromMs, long toMs, CancellationToken ct) =>
|
|
_db.ReadAsync<IReadOnlyList<MetricRow>>(
|
|
c =>
|
|
{
|
|
if (agentIds.Count == 0)
|
|
{
|
|
return [];
|
|
}
|
|
|
|
// 代理 ID 经参数传入：占位符是固定格式的 @a0、@a1…，不拼接任何外部值
|
|
var parameters = new List<(string, object?)> { ("@s", series), ("@from", fromMs), ("@to", toMs) };
|
|
var placeholders = new List<string>();
|
|
int i = 0;
|
|
foreach (NodeId agent in agentIds)
|
|
{
|
|
placeholders.Add("@a" + i);
|
|
parameters.Add(("@a" + i, agent.Value));
|
|
i++;
|
|
}
|
|
|
|
return c.Query(
|
|
$"SELECT agent_id, series, window_start, partial, metrics FROM {level.TableName()} WHERE series = @s AND window_start >= @from AND window_start < @to " +
|
|
$"AND agent_id IN ({string.Join(',', placeholders)}) ORDER BY agent_id, window_start",
|
|
ReadRow, [.. parameters]);
|
|
},
|
|
ct);
|
|
|
|
public ValueTask<int> DeleteOlderThanAsync(MetricLevel level, long cutoffMs, CancellationToken ct) =>
|
|
level == MetricLevel.H1
|
|
? ValueTask.FromResult(0)
|
|
: _db.WriteAsync(c => c.Exec($"DELETE FROM {level.TableName()} WHERE window_start < @cutoff", ("@cutoff", cutoffMs)), ct);
|
|
|
|
public ValueTask EnqueueAsync(IReadOnlyList<QueueItem> items, CancellationToken ct) =>
|
|
_db.ExecuteAsync(c =>
|
|
{
|
|
foreach (QueueItem item in items)
|
|
{
|
|
c.Exec(
|
|
"INSERT INTO analysis_queue (agent_id, series, window_start, not_before, reason, seq) VALUES (@a, @s, @w, @n, @r, 1) " +
|
|
"ON CONFLICT(agent_id, series, window_start) DO UPDATE SET not_before = MIN(not_before, excluded.not_before), reason = excluded.reason, seq = seq + 1",
|
|
("@a", item.AgentId.Value), ("@s", item.Series), ("@w", item.WindowStart), ("@n", item.NotBefore), ("@r", item.Reason));
|
|
}
|
|
}, ct);
|
|
|
|
public ValueTask<IReadOnlyList<QueueItem>> DequeueDueAsync(long nowMs, int limit, CancellationToken ct) =>
|
|
_db.ReadAsync<IReadOnlyList<QueueItem>>(
|
|
c => c.Query(
|
|
"SELECT agent_id, series, window_start, not_before, reason, seq FROM analysis_queue WHERE not_before <= @now ORDER BY not_before LIMIT @limit",
|
|
r => new QueueItem(new NodeId(r.GetString(0)), r.GetString(1), r.GetInt64(2), r.GetInt64(3), r.GetString(4), r.GetInt64(5)),
|
|
("@now", nowMs), ("@limit", limit)),
|
|
ct);
|
|
|
|
public ValueTask CompleteAsync(IReadOnlyList<QueueItem> items, CancellationToken ct) =>
|
|
_db.ExecuteAsync(c =>
|
|
{
|
|
foreach (QueueItem item in items)
|
|
{
|
|
// 处理期间又有迟到的块入队时序号已经变了，这一项会保留到下一轮
|
|
c.Exec(
|
|
"DELETE FROM analysis_queue WHERE agent_id = @a AND series = @s AND window_start = @w AND seq = @seq",
|
|
("@a", item.AgentId.Value), ("@s", item.Series), ("@w", item.WindowStart), ("@seq", item.Seq));
|
|
}
|
|
}, ct);
|
|
|
|
public ValueTask<int> QueueLengthAsync(CancellationToken ct) =>
|
|
_db.ReadAsync(c => (int)c.Scalar<long>("SELECT COUNT(*) FROM analysis_queue"), ct);
|
|
|
|
public ValueTask MarkRollupDirtyAsync(MetricLevel level, NodeId agentId, string series, long bucketStart, CancellationToken ct) =>
|
|
_db.ExecuteAsync(
|
|
c => c.Exec(
|
|
"INSERT OR IGNORE INTO rollup_dirty (level, agent_id, series, bucket_start) VALUES (@l, @a, @s, @b)",
|
|
("@l", (int)level), ("@a", agentId.Value), ("@s", series), ("@b", bucketStart)),
|
|
ct);
|
|
|
|
public ValueTask<IReadOnlyList<(MetricLevel Level, NodeId AgentId, string Series, long BucketStart)>> TakeRollupDirtyAsync(int limit, CancellationToken ct) =>
|
|
_db.WriteAsync<IReadOnlyList<(MetricLevel, NodeId, string, long)>>(c =>
|
|
{
|
|
// 先低层后高层：5 分钟层先于 1 小时层，保证 1 小时层读到的是最新的 5 分钟层
|
|
List<(MetricLevel, NodeId, string, long)> taken = c.Query(
|
|
"SELECT level, agent_id, series, bucket_start FROM rollup_dirty ORDER BY level, bucket_start LIMIT @limit",
|
|
r => ((MetricLevel)r.GetInt32(0), new NodeId(r.GetString(1)), r.GetString(2), r.GetInt64(3)),
|
|
("@limit", limit));
|
|
foreach ((MetricLevel level, NodeId agent, string series, long bucket) in taken)
|
|
{
|
|
c.Exec(
|
|
"DELETE FROM rollup_dirty WHERE level = @l AND agent_id = @a AND series = @s AND bucket_start = @b",
|
|
("@l", (int)level), ("@a", agent.Value), ("@s", series), ("@b", bucket));
|
|
}
|
|
|
|
return taken;
|
|
}, ct);
|
|
|
|
public ValueTask<IReadOnlyList<MetricVersionRecord>> GetVersionsAsync(CancellationToken ct) =>
|
|
_db.ReadAsync<IReadOnlyList<MetricVersionRecord>>(
|
|
c => c.Query("SELECT metric_id, fingerprint, recompute_from, recompute_cursor FROM metric_versions", r => new MetricVersionRecord(r.GetString(0), r.GetString(1), r.NullableLong(2), r.NullableLong(3))),
|
|
ct);
|
|
|
|
public ValueTask SetVersionAsync(MetricVersionRecord version, CancellationToken ct) =>
|
|
_db.ExecuteAsync(
|
|
c => c.Exec(
|
|
"INSERT INTO metric_versions (metric_id, fingerprint, recompute_from, recompute_cursor) VALUES (@id, @f, @from, @cur) " +
|
|
"ON CONFLICT(metric_id) DO UPDATE SET fingerprint = excluded.fingerprint, recompute_from = excluded.recompute_from, recompute_cursor = excluded.recompute_cursor",
|
|
("@id", version.MetricId), ("@f", version.Fingerprint), ("@from", version.RecomputeFrom), ("@cur", version.RecomputeCursor)),
|
|
ct);
|
|
|
|
private static byte[]? Load(SqliteConnection c, MetricLevel level, MetricRow row) =>
|
|
c.Scalar<byte[]>(
|
|
$"SELECT metrics FROM {level.TableName()} WHERE agent_id = @a AND series = @s AND window_start = @w",
|
|
("@a", row.AgentId.Value), ("@s", row.Series), ("@w", row.WindowStart));
|
|
|
|
private static bool IsPartial(SqliteConnection c, MetricLevel level, MetricRow row) =>
|
|
c.Scalar<long>(
|
|
$"SELECT partial FROM {level.TableName()} WHERE agent_id = @a AND series = @s AND window_start = @w",
|
|
("@a", row.AgentId.Value), ("@s", row.Series), ("@w", row.WindowStart)) != 0;
|
|
|
|
private static void Upsert(SqliteConnection c, MetricLevel level, MetricRow row, byte[] metrics) =>
|
|
c.Exec(
|
|
$"INSERT INTO {level.TableName()} (agent_id, series, window_start, partial, metrics) VALUES (@a, @s, @w, @p, @m) " +
|
|
"ON CONFLICT(agent_id, series, window_start) DO UPDATE SET partial = excluded.partial, metrics = excluded.metrics",
|
|
("@a", row.AgentId.Value), ("@s", row.Series), ("@w", row.WindowStart), ("@p", row.Partial ? 1 : 0), ("@m", metrics));
|
|
|
|
private static MetricRow ReadRow(SqliteDataReader r) => new(new NodeId(r.GetString(0)), r.GetString(1), r.GetInt64(2), r.GetInt64(3) != 0, r.Bytes(4));
|
|
}
|