gsadmin/AutopilotMonitor
1
0
Fork 0
mirror of https://github.com/okieselbach/AutopilotMonitor.git synced 2026-10-24 12:42:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(backend): fold do* fields from download_progress — DO data survives IME 1.104

Browse Source 
IME 1.104 removed the [DO TEL] log line, leaving the agent's DO collector as
the only do_telemetry source — and its gate only fires when a 3s poll itself
observes the completed download. Fast, interrupted or between-poll-completed
downloads never emit one, so DoDownloadMode stayed -1 and DoAggregator dropped
the whole row: DO bytes read as "practically null" on 1.104 fleets even though
download_progress events carried the full do* set on every poll.

- Shared ApplyDoFields for both writers: download_progress folds byte counters
  and file size via Math.Max (monotonic, replay-idempotent) and sets the mode
  only when >= 0; do_telemetry stays authoritative (last-write-wins) but never
  regresses DoDownloadMode to the -1 unset sentinel.
- DownloadBytes now prefers actually-transferred bytes over doFileSize (the
  file size used to inflate the transfer measure).

Intended effect after deploy: DO aggregates count previously-invisible
incomplete downloads, so app counts/bytes rise. No backfill of historic rows.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-08-20 01:25:28 +02:00
parent aab7ae12ff
commit ff8b847dc0
2 changed files with 255 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Backend/AutopilotMonitor.Functions.Tests/AppInstallDoFieldsFallbackTests.cs
+162
View File
@@ -0,0 +1,162 @@
using AutopilotMonitor.Functions.Services;
using AutopilotMonitor.Shared.Models;
using Xunit;
namespace AutopilotMonitor.Functions.Tests;
/// <summary>
/// DO ingest fallback (IME ≥ 1.104 removed the [DO TEL] log line; the agent's DO collector
/// only emits <c>do_telemetry</c> for downloads whose completion a poll itself observed).
/// <c>download_progress</c> events carry the full <c>do*</c> set on every poll and must fold
/// into the summary's Do* columns so interrupted / between-poll-completed downloads still
/// reach the Delivery Optimization aggregates:
/// - progress-only sessions populate Do* incl. a valid DoDownloadMode (DoAggregator gate),
/// - a later do_telemetry stays authoritative,
/// - replayed progress events are idempotent (Math.Max),
/// - DownloadBytes prefers transferred bytes over file size.
/// </summary>
public class AppInstallDoFieldsFallbackTests
{
private static readonly DateTime T0 = new(2026, 8, 19, 10, 0, 0, DateTimeKind.Utc);
private static Dictionary<string, AppInstallAggregationState> Aggregate(params EnrollmentEvent[] events)
{
var summaries = new Dictionary<string, AppInstallAggregationState>(StringComparer.OrdinalIgnoreCase);
foreach (var evt in events)
EventIngestProcessor.AggregateAppInstallEvent(evt, "tenant", "session", summaries);
return summaries;
}
private static EnrollmentEvent DownloadProgress(
long bytesDownloaded, long doTotal, long doPeers, long doHttp, int doMode,
long doFileSize = 0, int doPct = 0, long doCacheServer = 0, DateTime? ts = null)
=> new EnrollmentEvent
{
EventType = "download_progress",
Timestamp = ts ?? T0,
Data = new Dictionary<string, object>
{
["appName"] = "App",
["bytesDownloaded"] = bytesDownloaded,
["doFileSize"] = doFileSize,
["doTotalBytesDownloaded"] = doTotal,
["doBytesFromPeers"] = doPeers,
["doBytesFromHttp"] = doHttp,
["doPercentPeerCaching"] = doPct,
["doDownloadMode"] = doMode,
["doBytesFromCacheServer"] = doCacheServer,
},
};
private static EnrollmentEvent DoTelemetry(
long doFileSize, long doTotal, long doPeers, long doHttp, int doMode,
string? duration = null)
{
var data = new Dictionary<string, object>
{
["appName"] = "App",
["doFileSize"] = doFileSize,
["doTotalBytesDownloaded"] = doTotal,
["doBytesFromPeers"] = doPeers,
["doBytesFromHttp"] = doHttp,
["doPercentPeerCaching"] = 50,
["doDownloadMode"] = doMode,
};
if (duration != null) data["doDownloadDuration"] = duration;
return new EnrollmentEvent { EventType = "do_telemetry", Timestamp = T0, Data = data };
}
[Fact]
public void ProgressOnly_PopulatesDoColumns_AndSurvivesDoAggregator()
{
// An interrupted download: progress events observed, do_telemetry never fired.
var s = Aggregate(
DownloadProgress(bytesDownloaded: 1000, doTotal: 1000, doPeers: 600, doHttp: 400, doMode: 1, doFileSize: 5000, doPct: 60),
DownloadProgress(bytesDownloaded: 3000, doTotal: 3000, doPeers: 2000, doHttp: 1000, doMode: 1, doFileSize: 5000, doPct: 66)
)["App"].Summary;
Assert.Equal(3000, s.DoTotalBytesDownloaded);
Assert.Equal(2000, s.DoBytesFromPeers);
Assert.Equal(1000, s.DoBytesFromHttp);
Assert.Equal(5000, s.DoFileSize);
Assert.Equal(66, s.DoPercentPeerCaching);
Assert.Equal(1, s.DoDownloadMode);
Assert.Equal(3000, s.DownloadBytes); // from bytesDownloaded, not the file size
// The whole point: DoDownloadMode >= 0 keeps the row inside the DO aggregate.
var aggregate = DoAggregator.Compute(new[] { s });
Assert.Equal(1, aggregate.DoAppCount);
Assert.Equal(2000, aggregate.BytesFromPeers);
}
[Fact]
public void ProgressReplay_IsIdempotent_MaxFoldNeverRegresses()
{
// Out-of-order / replayed batch: the older (smaller) progress arrives again last.
var s = Aggregate(
DownloadProgress(bytesDownloaded: 3000, doTotal: 3000, doPeers: 2000, doHttp: 1000, doMode: 1),
DownloadProgress(bytesDownloaded: 1000, doTotal: 1000, doPeers: 600, doHttp: 400, doMode: 1)
)["App"].Summary;
Assert.Equal(3000, s.DoTotalBytesDownloaded);
Assert.Equal(2000, s.DoBytesFromPeers);
Assert.Equal(1000, s.DoBytesFromHttp);
}
[Fact]
public void DoTelemetry_StaysAuthoritative_OverEarlierProgress()
{
// Progress saw a partial state; the final telemetry read has the complete numbers
// (which can be LOWER per bucket than a transient progress max — telemetry wins).
var s = Aggregate(
DownloadProgress(bytesDownloaded: 4000, doTotal: 4000, doPeers: 3500, doHttp: 500, doMode: 1, doPct: 87),
DoTelemetry(doFileSize: 5000, doTotal: 5000, doPeers: 3000, doHttp: 2000, doMode: 1, duration: "00:01:40")
)["App"].Summary;
Assert.Equal(5000, s.DoTotalBytesDownloaded);
Assert.Equal(3000, s.DoBytesFromPeers); // telemetry overwrote the higher progress value
Assert.Equal(2000, s.DoBytesFromHttp);
Assert.Equal(50, s.DoPercentPeerCaching);
Assert.Equal("00:01:40", s.DoDownloadDuration);
Assert.Equal(100, s.DownloadDurationSeconds);
}
[Fact]
public void DoTelemetry_WithUnsetMode_DoesNotRegressAKnownMode()
{
// A telemetry event whose DownloadMode property was absent parses as -1; it must not
// re-hide the row from DoAggregator after progress already established the mode.
var s = Aggregate(
DownloadProgress(bytesDownloaded: 1000, doTotal: 1000, doPeers: 600, doHttp: 400, doMode: 3),
DoTelemetry(doFileSize: 1000, doTotal: 1000, doPeers: 600, doHttp: 400, doMode: -1)
)["App"].Summary;
Assert.Equal(3, s.DoDownloadMode);
}
[Fact]
public void DownloadBytes_PrefersTransferredBytes_OverFileSize()
{
// Historic bug: DownloadBytes was max-ed with doFileSize, inflating the transfer
// measure. With a real transfer total present the file size must not win.
var s = Aggregate(
DoTelemetry(doFileSize: 9000, doTotal: 5000, doPeers: 3000, doHttp: 2000, doMode: 1)
)["App"].Summary;
Assert.Equal(5000, s.DownloadBytes);
Assert.Equal(9000, s.DoFileSize); // the file size column itself keeps the real value
}
[Fact]
public void DownloadBytes_FallsBackToFileSize_WhenNoTransferTotal()
{
var evt = new EnrollmentEvent
{
EventType = "do_telemetry",
Timestamp = T0,
Data = new Dictionary<string, object> { ["appName"] = "App", ["doFileSize"] = 7000L, ["doDownloadMode"] = 1 },
};
var s = Aggregate(evt)["App"].Summary;
Assert.Equal(7000, s.DownloadBytes);
}
}
src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.AppInstall.cs
+93 -36
View File
@@ -184,52 +184,109 @@ namespace AutopilotMonitor.Functions.Services
: evt.Data?.ContainsKey("bytes_downloaded") == true ? "bytes_downloaded" : null;
if (bytesKey != null && long.TryParse(evt.Data![bytesKey]?.ToString(), out var bytes))
summary.DownloadBytes = Math.Max(summary.DownloadBytes, bytes);
// DO fallback: IME >= 1.104 removed the [DO TEL] log line, and the agent's
// DO collector only emits do_telemetry for downloads whose completion a poll
// itself observed — fast, interrupted or between-poll-completed downloads
// never get one. The progress events carry the full do* set on every poll,
// so fold them in monotonically; a do_telemetry (below) stays authoritative.
if (evt.Data != null)
ApplyDoFields(summary, evt.Data, isAuthoritative: false);
break;
case "do_telemetry":
if (evt.Data != null)
{
if (evt.Data.ContainsKey("doFileSize") && long.TryParse(evt.Data["doFileSize"]?.ToString(), out var doFs))
{
summary.DownloadBytes = Math.Max(summary.DownloadBytes, doFs);
summary.DoFileSize = doFs;
}
if (evt.Data.ContainsKey("doTotalBytesDownloaded") && long.TryParse(evt.Data["doTotalBytesDownloaded"]?.ToString(), out var doTotalDl))
summary.DoTotalBytesDownloaded = doTotalDl;
if (evt.Data.ContainsKey("doBytesFromPeers") && long.TryParse(evt.Data["doBytesFromPeers"]?.ToString(), out var doPeers))
summary.DoBytesFromPeers = doPeers;
if (evt.Data.ContainsKey("doBytesFromHttp") && long.TryParse(evt.Data["doBytesFromHttp"]?.ToString(), out var doHttp))
summary.DoBytesFromHttp = doHttp;
if (evt.Data.ContainsKey("doPercentPeerCaching") && int.TryParse(evt.Data["doPercentPeerCaching"]?.ToString(), out var doPct))
summary.DoPercentPeerCaching = doPct;
if (evt.Data.ContainsKey("doDownloadMode") && int.TryParse(evt.Data["doDownloadMode"]?.ToString(), out var doMode))
summary.DoDownloadMode = doMode;
if (evt.Data.ContainsKey("doDownloadDuration"))
{
var doDurStr = evt.Data["doDownloadDuration"]?.ToString() ?? string.Empty;
summary.DoDownloadDuration = doDurStr;
if (TimeSpan.TryParse(doDurStr, out var doDurTs) && doDurTs.TotalSeconds >= 1)
summary.DownloadDurationSeconds = Math.Max(summary.DownloadDurationSeconds, (int)doDurTs.TotalSeconds);
}
if (evt.Data.ContainsKey("doBytesFromLanPeers") && long.TryParse(evt.Data["doBytesFromLanPeers"]?.ToString(), out var doLan))
summary.DoBytesFromLanPeers = doLan;
if (evt.Data.ContainsKey("doBytesFromGroupPeers") && long.TryParse(evt.Data["doBytesFromGroupPeers"]?.ToString(), out var doGroup))
summary.DoBytesFromGroupPeers = doGroup;
if (evt.Data.ContainsKey("doBytesFromInternetPeers") && long.TryParse(evt.Data["doBytesFromInternetPeers"]?.ToString(), out var doInet))
summary.DoBytesFromInternetPeers = doInet;
if (evt.Data.ContainsKey("doBytesFromLinkLocalPeers") && long.TryParse(evt.Data["doBytesFromLinkLocalPeers"]?.ToString(), out var doLinkLocal))
summary.DoBytesFromLinkLocalPeers = doLinkLocal;
if (evt.Data.ContainsKey("doBytesFromCacheServer") && long.TryParse(evt.Data["doBytesFromCacheServer"]?.ToString(), out var doCache))
summary.DoBytesFromCacheServer = doCache;
if (evt.Data.ContainsKey("doCacheHost"))
summary.DoCacheHost = evt.Data["doCacheHost"]?.ToString() ?? string.Empty;
}
ApplyDoFields(summary, evt.Data, isAuthoritative: true);
break;
}
RecalculateAppDurations(state);
}
/// <summary>
/// Folds the <c>do*</c> Delivery Optimization fields of one event into the summary.
/// Two writers share this so they cannot drift:
/// <list type="bullet">
/// <item><c>do_telemetry</c> (<paramref name="isAuthoritative"/> = true) — the collector's
/// final per-app read: last-write-wins per present key (pre-existing semantics), except
/// <c>DoDownloadMode</c> never regresses to the -1 "unset" sentinel — that would re-hide
/// the row from <c>DoAggregator</c> after a progress event already established a mode.</item>
/// <item><c>download_progress</c> (false) — per-poll observations: byte counters and file
/// size fold via <c>Math.Max</c> (monotonic per download, so replays and out-of-order
/// batches stay idempotent); mode/percent/cacheHost are latest-observation writes.</item>
/// </list>
/// <c>DownloadBytes</c> prefers actually-transferred bytes (<c>doTotalBytesDownloaded</c>)
/// over <c>doFileSize</c> — the file size is only the fallback when no transfer total exists
/// (it used to unconditionally inflate the transfer measure).
/// </summary>
internal static void ApplyDoFields(AppInstallSummary summary, Dictionary<string, object> data, bool isAuthoritative)
{
static bool TryGetLong(Dictionary<string, object> d, string key, out long value)
{
value = 0;
return d.TryGetValue(key, out var raw) && long.TryParse(raw?.ToString(), out value);
}
static bool TryGetInt(Dictionary<string, object> d, string key, out int value)
{
value = 0;
return d.TryGetValue(key, out var raw) && int.TryParse(raw?.ToString(), out value);
}
static long Fold(bool authoritative, long current, long incoming)
=> authoritative ? incoming : Math.Max(current, incoming);
var hasFileSize = TryGetLong(data, "doFileSize", out var doFs);
var hasTotalDl = TryGetLong(data, "doTotalBytesDownloaded", out var doTotalDl);
if (hasFileSize)
summary.DoFileSize = Fold(isAuthoritative, summary.DoFileSize, doFs);
if (hasTotalDl)
summary.DoTotalBytesDownloaded = Fold(isAuthoritative, summary.DoTotalBytesDownloaded, doTotalDl);
// Transfer measure: real transferred bytes when known, file size only as fallback.
// Progress events already fed DownloadBytes via their bytesDownloaded field, so only
// the authoritative telemetry contributes here.
if (isAuthoritative && (hasFileSize || hasTotalDl))
summary.DownloadBytes = Math.Max(summary.DownloadBytes, hasTotalDl && doTotalDl > 0 ? doTotalDl : doFs);
if (TryGetLong(data, "doBytesFromPeers", out var doPeers))
summary.DoBytesFromPeers = Fold(isAuthoritative, summary.DoBytesFromPeers, doPeers);
if (TryGetLong(data, "doBytesFromHttp", out var doHttp))
summary.DoBytesFromHttp = Fold(isAuthoritative, summary.DoBytesFromHttp, doHttp);
if (TryGetLong(data, "doBytesFromLanPeers", out var doLan))
summary.DoBytesFromLanPeers = Fold(isAuthoritative, summary.DoBytesFromLanPeers, doLan);
if (TryGetLong(data, "doBytesFromGroupPeers", out var doGroup))
summary.DoBytesFromGroupPeers = Fold(isAuthoritative, summary.DoBytesFromGroupPeers, doGroup);
if (TryGetLong(data, "doBytesFromInternetPeers", out var doInet))
summary.DoBytesFromInternetPeers = Fold(isAuthoritative, summary.DoBytesFromInternetPeers, doInet);
if (TryGetLong(data, "doBytesFromLinkLocalPeers", out var doLinkLocal))
summary.DoBytesFromLinkLocalPeers = Fold(isAuthoritative, summary.DoBytesFromLinkLocalPeers, doLinkLocal);
if (TryGetLong(data, "doBytesFromCacheServer", out var doCache))
summary.DoBytesFromCacheServer = Fold(isAuthoritative, summary.DoBytesFromCacheServer, doCache);
if (TryGetInt(data, "doPercentPeerCaching", out var doPct))
summary.DoPercentPeerCaching = doPct;
// -1 is the "unset" sentinel DoAggregator filters on — never let it clobber a
// known mode (a telemetry event without the DownloadMode property parses as -1).
if (TryGetInt(data, "doDownloadMode", out var doMode) && doMode >= 0)
summary.DoDownloadMode = doMode;
if (data.TryGetValue("doCacheHost", out var cacheHostRaw))
{
var cacheHost = cacheHostRaw?.ToString();
if (isAuthoritative || !string.IsNullOrEmpty(cacheHost))
summary.DoCacheHost = cacheHost ?? string.Empty;
}
// Only do_telemetry carries a duration; progress events never have the key.
if (data.ContainsKey("doDownloadDuration"))
{
var doDurStr = data["doDownloadDuration"]?.ToString() ?? string.Empty;
summary.DoDownloadDuration = doDurStr;
if (TimeSpan.TryParse(doDurStr, out var doDurTs) && doDurTs.TotalSeconds >= 1)
summary.DownloadDurationSeconds = Math.Max(summary.DownloadDurationSeconds, (int)doDurTs.TotalSeconds);
}
}
internal static void RecalculateAppDurations(AppInstallAggregationState state)
{
var summary = state.Summary;