488 lines
22 KiB
C#
488 lines
22 KiB
C#
using System.Text.Json;
|
||
using Fleet.Core;
|
||
using MoShiro.Backend.Analysis;
|
||
using MoShiro.Protocol;
|
||
|
||
namespace MoShiro.Backend.Application;
|
||
|
||
/// <summary>
|
||
/// 展示数据查询：任意时间段、自动步长、数据分层与现场合并、精确汇总、列式结果。
|
||
/// 新增指标时这套代码不需要任何改动：它只通过注册表按 id 读取累加器并导出值。
|
||
/// </summary>
|
||
public sealed class QueryService
|
||
{
|
||
/// <summary>热力图的延迟桶边界（毫秒）。</summary>
|
||
public static readonly IReadOnlyList<double> HeatmapEdges =
|
||
[1, 2, 3, 5, 7, 10, 15, 20, 30, 50, 70, 100, 150, 200, 300, 500, 1000, 2000, 5000, 30000];
|
||
|
||
private readonly RangeResolver _resolver;
|
||
private readonly SourcePlanner _planner;
|
||
private readonly RowLoader _loader;
|
||
private readonly HotLayer _hot;
|
||
private readonly IEventStore _events;
|
||
private readonly ITestStore _tests;
|
||
private readonly BackendOptions _options;
|
||
private readonly TimeProvider _time;
|
||
private readonly AgentRuntime _runtime;
|
||
|
||
public QueryService(RangeResolver resolver, SourcePlanner planner, RowLoader loader, HotLayer hot, IEventStore events, ITestStore tests, BackendOptions options, TimeProvider time, AgentRuntime runtime)
|
||
{
|
||
_runtime = runtime;
|
||
_resolver = resolver;
|
||
_planner = planner;
|
||
_loader = loader;
|
||
_hot = hot;
|
||
_events = events;
|
||
_tests = tests;
|
||
_options = options;
|
||
_time = time;
|
||
}
|
||
|
||
public Result<(QueryRange Range, SourcePlan Plan)> Plan(RangeRequest request, string defaultRange)
|
||
{
|
||
long now = _time.GetUtcNow().ToUnixTimeMilliseconds();
|
||
Result<QueryRange> resolved = _resolver.Resolve(request, now, defaultRange);
|
||
if (!resolved.TryGet(out QueryRange? range, out FleetError? error))
|
||
{
|
||
return Result<(QueryRange, SourcePlan)>.Fail(error);
|
||
}
|
||
|
||
SourcePlan plan = _planner.Plan(range, now, _hot.Ready);
|
||
if (plan.StepMs != range.StepMs)
|
||
{
|
||
range = range.WithStep(plan.StepMs);
|
||
}
|
||
|
||
return Result<(QueryRange, SourcePlan)>.Ok((range, plan));
|
||
}
|
||
|
||
// ───────────────────────── 单条线路的时序 ─────────────────────────
|
||
|
||
/// <summary>这条线路现在是不是连着（监听方报告了这个拨号方的会话）：连着的线路窗口以当前时间为终点，不去查“最后一个样本”——Backend 刚重启、热层和持久层还没追上时，后者会落后好几个小时，页面就显示成旧数据。</summary>
|
||
private bool IsConnected(LineKey line) =>
|
||
_runtime.TryGet(line.Listener, out AgentRuntimeState? state) && state.Dialers.Any(p => p.Id == line.Dialer.Value && p.IpVersion == line.IpVersion);
|
||
|
||
public async ValueTask<Result<SeriesResult>> SeriesAsync(LineKey line, RangeRequest request, MetricSelection selection, CancellationToken ct)
|
||
{
|
||
// anchor=latest：离线或刚恢复的线路，窗口以它最后一个样本为终点（落在最近 5 秒之内就当作在线，仍以当前时间为终点）
|
||
if (string.Equals(request.Anchor, "latest", StringComparison.OrdinalIgnoreCase) && request.To is null && request.From is null && request.Range is not null
|
||
&& RangeResolver.TryParseSpan(request.Range, out long span)
|
||
&& !IsConnected(line))
|
||
{
|
||
long now = _time.GetUtcNow().ToUnixTimeMilliseconds();
|
||
if (await _loader.LatestSampleMsAsync(line.Listener, line.Series, now, ct) is { } latest && latest < now - 5000)
|
||
{
|
||
request = request with { Range = null, From = DateTimeOffset.FromUnixTimeMilliseconds(latest - span), To = DateTimeOffset.FromUnixTimeMilliseconds(latest) };
|
||
}
|
||
}
|
||
|
||
Result<(QueryRange Range, SourcePlan Plan)> planned = Plan(request, "15m");
|
||
if (!planned.TryGet(out (QueryRange Range, SourcePlan Plan) p, out FleetError? error))
|
||
{
|
||
return Result<SeriesResult>.Fail(error);
|
||
}
|
||
|
||
Result<IReadOnlyList<SourceRow>> loaded = await _loader.LoadAsync(p.Plan, p.Range, line.Listener, line.Series, ct);
|
||
if (!loaded.TryGet(out IReadOnlyList<SourceRow>? rows, out error))
|
||
{
|
||
return Result<SeriesResult>.Fail(error);
|
||
}
|
||
|
||
BucketedRows bucketed = BucketedRows.Create(rows, p.Range);
|
||
IReadOnlyDictionary<string, double> overlay = selection.WantsScores ? await TestOverlayAsync(line, ct) : new Dictionary<string, double>();
|
||
|
||
int n = p.Range.Points;
|
||
var values = selection.AllIds.ToDictionary(id => id, _ => new double?[n], StringComparer.Ordinal);
|
||
double widthSeconds = p.Plan.LevelWidthMs / 1000.0;
|
||
IEnumerable<string> evaluate = selection.WantsScores ? ExportEvaluator.AllExportIds() : selection.ExportIds;
|
||
for (int i = 0; i < n; i++)
|
||
{
|
||
MetricSet? set = bucketed.Sets[i];
|
||
if (set is null)
|
||
{
|
||
continue;
|
||
}
|
||
|
||
Dictionary<string, double> exported = ExportEvaluator.Evaluate(set, evaluate, bucketed.RowCounts[i] * widthSeconds);
|
||
foreach (string id in selection.ExportIds)
|
||
{
|
||
if (exported.TryGetValue(id, out double v))
|
||
{
|
||
values[id][i] = v;
|
||
}
|
||
}
|
||
|
||
if (selection.WantsScores)
|
||
{
|
||
Dictionary<string, ModelScore> scores = ModelRegistry.ScoreAll(BuildView(exported, overlay));
|
||
foreach (string model in selection.ScoreModelIds)
|
||
{
|
||
values["score." + model][i] = scores[model].Score;
|
||
}
|
||
}
|
||
}
|
||
|
||
SummaryResult summary = Summarize(bucketed.Merge(), bucketed.TotalRows, p.Plan, p.Range, selection, overlay);
|
||
long nowMs = _time.GetUtcNow().ToUnixTimeMilliseconds();
|
||
IReadOnlyList<EventRecord> events = await _events.ListAsync(line.Listener, null, p.Range.FromMs, p.Range.ToMs, null, 100, ct);
|
||
IReadOnlyList<GapInfo> gaps = FindGaps(bucketed, p.Range, nowMs, events);
|
||
var versions = new Dictionary<string, string>(StringComparer.Ordinal);
|
||
foreach (string id in selection.ExportIds)
|
||
{
|
||
if (MetricRegistry.TryFindExport(id, out MetricDef def, out _))
|
||
{
|
||
versions[id] = def.Fingerprint;
|
||
}
|
||
}
|
||
|
||
return Result<SeriesResult>.Ok(new SeriesResult(line, p.Range, p.Plan, bucketed.T, values, summary, gaps, events, versions));
|
||
}
|
||
|
||
/// <summary>某个 Agent 上一个单个 Agent 任务的时序（序列 <c>target:<任务 id></c>）。</summary>
|
||
public async ValueTask<Result<TargetSeriesResult>> TargetSeriesAsync(NodeId agent, TaskId task, RangeRequest request, MetricSelection selection, CancellationToken ct)
|
||
{
|
||
Result<(QueryRange Range, SourcePlan Plan)> planned = Plan(request, "3h");
|
||
if (!planned.TryGet(out (QueryRange Range, SourcePlan Plan) p, out FleetError? error))
|
||
{
|
||
return Result<TargetSeriesResult>.Fail(error);
|
||
}
|
||
|
||
Result<IReadOnlyList<SourceRow>> loaded = await _loader.LoadAsync(p.Plan, p.Range, agent, TargetSeries(task), ct);
|
||
if (!loaded.TryGet(out IReadOnlyList<SourceRow>? rows, out error))
|
||
{
|
||
return Result<TargetSeriesResult>.Fail(error);
|
||
}
|
||
|
||
BucketedRows bucketed = BucketedRows.Create(rows, p.Range);
|
||
int n = p.Range.Points;
|
||
var values = selection.ExportIds.ToDictionary(id => id, _ => new double?[n], StringComparer.Ordinal);
|
||
double widthSeconds = p.Plan.LevelWidthMs / 1000.0;
|
||
for (int i = 0; i < n; i++)
|
||
{
|
||
if (bucketed.Sets[i] is not { } set)
|
||
{
|
||
continue;
|
||
}
|
||
|
||
Dictionary<string, double> exported = ExportEvaluator.Evaluate(set, selection.ExportIds, bucketed.RowCounts[i] * widthSeconds);
|
||
foreach (string id in selection.ExportIds)
|
||
{
|
||
if (exported.TryGetValue(id, out double v))
|
||
{
|
||
values[id][i] = v;
|
||
}
|
||
}
|
||
}
|
||
|
||
var versions = new Dictionary<string, string>(StringComparer.Ordinal);
|
||
foreach (string id in selection.ExportIds)
|
||
{
|
||
if (MetricRegistry.TryFindExport(id, out MetricDef def, out _))
|
||
{
|
||
versions[id] = def.Fingerprint;
|
||
}
|
||
}
|
||
|
||
SummaryResult summary = Summarize(bucketed.Merge(), bucketed.TotalRows, p.Plan, p.Range, selection, new Dictionary<string, double>());
|
||
return Result<TargetSeriesResult>.Ok(new TargetSeriesResult(p.Range, p.Plan, bucketed.T, values, summary, versions));
|
||
}
|
||
|
||
/// <summary>这个单个 Agent 任务最后一个样本的时刻（Unix 毫秒）；还没有数据返回 null。</summary>
|
||
public ValueTask<long?> TargetLatestAsync(NodeId agent, TaskId task, CancellationToken ct) =>
|
||
_loader.LatestSampleMsAsync(agent, TargetSeries(task), _time.GetUtcNow().ToUnixTimeMilliseconds(), ct);
|
||
|
||
private static string TargetSeries(TaskId task) => "target:" + task.Value;
|
||
|
||
/// <summary>这条线路最早的数据（Unix 毫秒）；没有数据返回 null。</summary>
|
||
public ValueTask<long?> EarliestAsync(LineKey line, CancellationToken ct) =>
|
||
_loader.EarliestSampleMsAsync(line.Listener, line.Series, _time.GetUtcNow().ToUnixTimeMilliseconds(), ct);
|
||
|
||
// ───────────────────────── 首页排行 ─────────────────────────
|
||
|
||
public async ValueTask<Result<OverviewResult>> OverviewAsync(IReadOnlyList<LineView> lines, RangeRequest request, MetricSelection selection, CancellationToken ct)
|
||
{
|
||
Result<(QueryRange Range, SourcePlan Plan)> planned = Plan(request, "5m");
|
||
if (!planned.TryGet(out (QueryRange Range, SourcePlan Plan) p, out FleetError? error))
|
||
{
|
||
return Result<OverviewResult>.Fail(error);
|
||
}
|
||
|
||
Result<Dictionary<LineKey, IReadOnlyList<SourceRow>>> loaded = await _loader.LoadManyAsync(p.Plan, p.Range, [.. lines.Select(l => l.Key)], ct);
|
||
if (!loaded.TryGet(out Dictionary<LineKey, IReadOnlyList<SourceRow>>? byLine, out error))
|
||
{
|
||
return Result<OverviewResult>.Fail(error);
|
||
}
|
||
|
||
var result = new List<OverviewLine>(lines.Count);
|
||
foreach (LineView line in lines)
|
||
{
|
||
IReadOnlyList<SourceRow> rows = byLine.GetValueOrDefault(line.Key) ?? [];
|
||
var merged = new MetricSet();
|
||
int count = 0;
|
||
foreach (SourceRow row in rows)
|
||
{
|
||
if (row.Start >= p.Range.FromMs && row.Start < p.Range.ToMs)
|
||
{
|
||
merged.Merge(row.Set);
|
||
count++;
|
||
}
|
||
}
|
||
|
||
IReadOnlyDictionary<string, double> overlay = selection.WantsScores ? await TestOverlayAsync(line.Key, ct) : new Dictionary<string, double>();
|
||
IReadOnlyList<TestRecord> recent = await _tests.ListAsync(line.Key.Listener, line.Key.Dialer, 1, line.Key.IpVersion, ct);
|
||
result.Add(new OverviewLine(line, Summarize(merged, count, p.Plan, p.Range, selection, overlay), recent.Count > 0 ? recent[0] : null));
|
||
}
|
||
|
||
return Result<OverviewResult>.Ok(new OverviewResult(p.Range, result, p.Plan));
|
||
}
|
||
|
||
// ───────────────────────── 对比、分布、热力图 ─────────────────────────
|
||
|
||
public async ValueTask<Result<CompareResult>> CompareAsync(IReadOnlyList<LineKey> lines, string exportId, RangeRequest request, CancellationToken ct)
|
||
{
|
||
if (lines.Count == 0 || lines.Count > _options.MaxCompareAgents)
|
||
{
|
||
return ValidationError.Fail<CompareResult>("lines", $"between 1 and {_options.MaxCompareAgents} lines are required");
|
||
}
|
||
|
||
if (!MetricRegistry.TryFindExport(exportId, out _, out _))
|
||
{
|
||
return Result<CompareResult>.Fail(MoShiroErrors.UnknownMetric, $"unknown metric {exportId}");
|
||
}
|
||
|
||
Result<(QueryRange Range, SourcePlan Plan)> planned = Plan(request, "1h");
|
||
if (!planned.TryGet(out (QueryRange Range, SourcePlan Plan) p, out FleetError? error))
|
||
{
|
||
return Result<CompareResult>.Fail(error);
|
||
}
|
||
|
||
Result<Dictionary<LineKey, IReadOnlyList<SourceRow>>> loaded = await _loader.LoadManyAsync(p.Plan, p.Range, lines, ct);
|
||
if (!loaded.TryGet(out Dictionary<LineKey, IReadOnlyList<SourceRow>>? byLine, out error))
|
||
{
|
||
return Result<CompareResult>.Fail(error);
|
||
}
|
||
|
||
var values = new Dictionary<string, double?[]>(StringComparer.Ordinal);
|
||
var summary = new Dictionary<string, double?>(StringComparer.Ordinal);
|
||
long[] t = [];
|
||
double widthSeconds = p.Plan.LevelWidthMs / 1000.0;
|
||
foreach (LineKey line in lines)
|
||
{
|
||
BucketedRows bucketed = BucketedRows.Create(byLine[line], p.Range);
|
||
t = bucketed.T;
|
||
double?[] column = new double?[p.Range.Points];
|
||
for (int i = 0; i < column.Length; i++)
|
||
{
|
||
if (bucketed.Sets[i] is { } set && ExportEvaluator.Evaluate(set, [exportId], bucketed.RowCounts[i] * widthSeconds).TryGetValue(exportId, out double v))
|
||
{
|
||
column[i] = v;
|
||
}
|
||
}
|
||
|
||
values[line.Id] = column;
|
||
summary[line.Id] = ExportEvaluator.Evaluate(bucketed.Merge(), [exportId], bucketed.TotalRows * widthSeconds).TryGetValue(exportId, out double total) ? total : null;
|
||
}
|
||
|
||
return Result<CompareResult>.Ok(new CompareResult(t, values, summary, exportId, p.Range, p.Plan));
|
||
}
|
||
|
||
public async ValueTask<Result<DistributionResult>> DistributionAsync(LineKey line, string metricId, RangeRequest request, CancellationToken ct)
|
||
{
|
||
MetricDef? def = MetricRegistry.Find(metricId);
|
||
if (def is null || def.Kind != AccKind.Histogram)
|
||
{
|
||
return Result<DistributionResult>.Fail(MoShiroErrors.UnknownMetric, $"{metricId} is not a histogram metric");
|
||
}
|
||
|
||
Result<(QueryRange Range, SourcePlan Plan)> planned = Plan(request, "15m");
|
||
if (!planned.TryGet(out (QueryRange Range, SourcePlan Plan) p, out FleetError? error))
|
||
{
|
||
return Result<DistributionResult>.Fail(error);
|
||
}
|
||
|
||
Result<IReadOnlyList<SourceRow>> loaded = await _loader.LoadAsync(p.Plan, p.Range, line.Listener, line.Series, ct);
|
||
if (!loaded.TryGet(out IReadOnlyList<SourceRow>? rows, out error))
|
||
{
|
||
return Result<DistributionResult>.Fail(error);
|
||
}
|
||
|
||
var merged = new HistogramAcc();
|
||
foreach (SourceRow row in rows)
|
||
{
|
||
if (row.Set.TryGet(metricId, out Accumulator acc) && acc is HistogramAcc h)
|
||
{
|
||
merged.Merge(h);
|
||
}
|
||
}
|
||
|
||
var buckets = new List<DistributionBucket>();
|
||
if (merged.Count > 0)
|
||
{
|
||
Dictionary<int, ulong> counts = merged.Buckets().ToDictionary(b => b.Bucket, b => b.Count);
|
||
int first = counts.Keys.Min();
|
||
int last = counts.Keys.Max();
|
||
for (int b = first; b <= last; b++)
|
||
{
|
||
buckets.Add(new DistributionBucket(Math.Round(HistogramAcc.UpperBound(b), 3), counts.GetValueOrDefault(b)));
|
||
}
|
||
}
|
||
|
||
return Result<DistributionResult>.Ok(new DistributionResult(metricId, def.Exports[0].Unit.ToString().ToLowerInvariant(), buckets, merged.Quantile(0.5), merged.Quantile(0.95), merged.Quantile(0.99), merged.Count));
|
||
}
|
||
|
||
public async ValueTask<Result<HeatmapResult>> HeatmapAsync(LineKey line, string metricId, RangeRequest request, CancellationToken ct)
|
||
{
|
||
MetricDef? def = MetricRegistry.Find(metricId);
|
||
if (def is null || def.Kind != AccKind.Histogram)
|
||
{
|
||
return Result<HeatmapResult>.Fail(MoShiroErrors.UnknownMetric, $"{metricId} is not a histogram metric");
|
||
}
|
||
|
||
Result<(QueryRange Range, SourcePlan Plan)> planned = Plan(request, "6h");
|
||
if (!planned.TryGet(out (QueryRange Range, SourcePlan Plan) p, out FleetError? error))
|
||
{
|
||
return Result<HeatmapResult>.Fail(error);
|
||
}
|
||
|
||
Result<IReadOnlyList<SourceRow>> loaded = await _loader.LoadAsync(p.Plan, p.Range, line.Listener, line.Series, ct);
|
||
if (!loaded.TryGet(out IReadOnlyList<SourceRow>? rows, out error))
|
||
{
|
||
return Result<HeatmapResult>.Fail(error);
|
||
}
|
||
|
||
BucketedRows bucketed = BucketedRows.Create(rows, p.Range);
|
||
var counts = new List<IReadOnlyList<ulong>>(bucketed.T.Length);
|
||
foreach (MetricSet? set in bucketed.Sets)
|
||
{
|
||
ulong[] cells = new ulong[HeatmapEdges.Count];
|
||
if (set is not null && set.TryGet(metricId, out Accumulator acc) && acc is HistogramAcc h)
|
||
{
|
||
foreach ((int bucket, ulong count) in h.Buckets())
|
||
{
|
||
double upper = HistogramAcc.UpperBound(bucket);
|
||
int cell = HeatmapEdges.Count - 1;
|
||
for (int e = 0; e < HeatmapEdges.Count; e++)
|
||
{
|
||
if (upper <= HeatmapEdges[e])
|
||
{
|
||
cell = e;
|
||
break;
|
||
}
|
||
}
|
||
|
||
cells[cell] += count;
|
||
}
|
||
}
|
||
|
||
counts.Add(cells);
|
||
}
|
||
|
||
return Result<HeatmapResult>.Ok(new HeatmapResult(bucketed.T, HeatmapEdges, counts, p.Range, p.Plan));
|
||
}
|
||
|
||
// ───────────────────────── 汇总与评分 ─────────────────────────
|
||
|
||
private SummaryResult Summarize(MetricSet merged, int totalRows, SourcePlan plan, QueryRange range, MetricSelection selection, IReadOnlyDictionary<string, double> overlay)
|
||
{
|
||
double durationSeconds = totalRows * plan.LevelWidthMs / 1000.0;
|
||
IEnumerable<string> evaluate = selection.WantsScores ? ExportEvaluator.AllExportIds() : selection.ExportIds;
|
||
Dictionary<string, double> exported = ExportEvaluator.Evaluate(merged, evaluate, durationSeconds);
|
||
|
||
var values = new Dictionary<string, double?>(StringComparer.Ordinal);
|
||
foreach (string id in selection.ExportIds)
|
||
{
|
||
values[id] = exported.TryGetValue(id, out double v) ? v : null;
|
||
}
|
||
|
||
if (selection.WantsScores)
|
||
{
|
||
Dictionary<string, ModelScore> scores = ModelRegistry.ScoreAll(BuildView(exported, overlay));
|
||
foreach (string model in selection.ScoreModelIds)
|
||
{
|
||
values["score." + model] = scores[model].Score;
|
||
}
|
||
}
|
||
|
||
var samples = new Dictionary<string, ulong>(StringComparer.Ordinal);
|
||
foreach ((MetricDef def, MetricExport export) in MetricRegistry.AllExports().Where(e => e.Export.Id.StartsWith("n.", StringComparison.Ordinal)))
|
||
{
|
||
if (merged.TryGet(def.Id, out Accumulator acc) && acc is CounterAcc counter && counter.Value > 0)
|
||
{
|
||
samples[export.Id["n.".Length..]] = counter.Value;
|
||
}
|
||
}
|
||
|
||
long now = _time.GetUtcNow().ToUnixTimeMilliseconds();
|
||
long expected = Math.Max(1, (Math.Min(range.ToMs, Math.Max(now, range.FromMs + plan.LevelWidthMs)) - range.FromMs) / plan.LevelWidthMs);
|
||
return new SummaryResult(values, samples, Math.Round(Math.Min(1.0, (double)totalRows / expected), 4));
|
||
}
|
||
|
||
private static MetricView BuildView(Dictionary<string, double> exported, IReadOnlyDictionary<string, double> overlay)
|
||
{
|
||
// 最近一次测试的结果补充持续指标里没有的值（例如 hy2.* 与 connect.*）
|
||
var all = new Dictionary<string, double>(exported, StringComparer.Ordinal);
|
||
foreach ((string id, double value) in overlay)
|
||
{
|
||
all.TryAdd(id, value);
|
||
}
|
||
|
||
return new MetricView(all);
|
||
}
|
||
|
||
private async ValueTask<IReadOnlyDictionary<string, double>> TestOverlayAsync(LineKey line, CancellationToken ct)
|
||
{
|
||
long cutoff = _time.GetUtcNow().ToUnixTimeMilliseconds() - (7 * Steps.Day);
|
||
var overlay = new Dictionary<string, double>(StringComparer.Ordinal);
|
||
var seenPlans = new HashSet<string>();
|
||
foreach (TestRecord test in await _tests.ListAsync(line.Listener, line.Dialer, 20, line.IpVersion, ct))
|
||
{
|
||
if (test.Status != TestStatuses.Succeeded || test.FinishedAtMs is not { } finished || finished < cutoff || !seenPlans.Add(test.Plan))
|
||
{
|
||
continue;
|
||
}
|
||
|
||
if (await _tests.GetResultAsync(test.Id, ct) is { } stored)
|
||
{
|
||
foreach ((string id, double value) in TestService.ToView(stored).Metrics)
|
||
{
|
||
overlay.TryAdd(id, value);
|
||
}
|
||
}
|
||
}
|
||
|
||
return overlay;
|
||
}
|
||
|
||
/// <summary>连续 ≥ 2 个空桶视为缺口；原因取该缺口附近最近的事件，没有事件就是 no_data。</summary>
|
||
private static List<GapInfo> FindGaps(BucketedRows bucketed, QueryRange range, long nowMs, IReadOnlyList<EventRecord> events)
|
||
{
|
||
var gaps = new List<GapInfo>();
|
||
int i = 0;
|
||
int n = bucketed.Sets.Length;
|
||
while (i < n)
|
||
{
|
||
if (bucketed.Sets[i] is not null || bucketed.T[i] >= nowMs)
|
||
{
|
||
i++;
|
||
continue;
|
||
}
|
||
|
||
int start = i;
|
||
while (i < n && bucketed.Sets[i] is null && bucketed.T[i] < nowMs)
|
||
{
|
||
i++;
|
||
}
|
||
|
||
if (i - start >= 2)
|
||
{
|
||
long from = bucketed.T[start];
|
||
long to = bucketed.T[i - 1] + range.StepMs;
|
||
EventRecord? near = events.FirstOrDefault(e => e.AtMs >= from - range.StepMs && e.AtMs <= to && e.Kind is "dialer_disconnected" or "agent_offline" or "clock_step" or "gap" or "cpu_saturated");
|
||
gaps.Add(new GapInfo(from, to, near?.Kind ?? "no_data"));
|
||
}
|
||
}
|
||
|
||
return gaps;
|
||
}
|
||
}
|