A RollupDay job folds each finished day's InteractionEvents into one InstanceDay per server:
- Counters for the admin, keyed channel:activity:object:outcome:reason, plus signature
schemes, audiences, local kinds and features;
- PublicCounters, everything a public page may ever read:
- inbound and outbound activities from an allowlist, on public, unlisted or unaddressed
traffic only, with the outcome collapsed (accepted or dropped, delivered or failed);
- no reasons, no Flag or Block;
- features of public objects;
- health:ok or health:failed from reachability (a 4xx means the server answered);
- latency, wait and byte histograms, and the number of distinct accounts from that day's
hashes.
Re-running a day replaces it and keeps the live Reads counters. The day's salt is then
deleted, so its hashes can never be recomputed, and the next day is queued.
StatisticsSchedule plans today's rollup every hour and catches up any of the last seven
days that have events but no rollup. Waits round up into their bucket.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ELjqpznMFMNrJoJUj6K5p2
133 lines
5.2 KiB
C#
133 lines
5.2 KiB
C#
using PrivaPub.Models.Statistics;
|
|
|
|
namespace PrivaPub.Infrastructure.Statistics
|
|
{
|
|
public sealed class HostFold
|
|
{
|
|
public Dictionary<string, long> Counters { get; } = new();
|
|
public Dictionary<string, long> PublicCounters { get; } = new();
|
|
public Dictionary<string, long> Latency { get; } = new();
|
|
public Dictionary<string, long> Waits { get; } = new();
|
|
public Dictionary<string, long> Bytes { get; } = new();
|
|
public HashSet<string> Accounts { get; } = new(StringComparer.Ordinal);
|
|
}
|
|
|
|
public sealed class DayFold
|
|
{
|
|
public Dictionary<string, HostFold> Hosts { get; } = new(StringComparer.Ordinal);
|
|
public HashSet<string> Accounts { get; } = new(StringComparer.Ordinal);
|
|
}
|
|
|
|
public static class Rollup
|
|
{
|
|
static readonly HashSet<string> PublicActivities = new(StringComparer.Ordinal)
|
|
{
|
|
"Create", "Update", "Delete", "Announce", "Like", "EmojiReact", "Dislike", "Follow", "Accept", "Reject", "Undo", "Move", "Add", "Remove",
|
|
"Join", "Leave", "QuoteRequest"
|
|
};
|
|
|
|
static readonly HashSet<string> PublicObjects = new(StringComparer.Ordinal)
|
|
{
|
|
"Note", "Article", "Page", "Question", "Video", "Audio", "Image", "Event", "Document", "Tombstone", "Person", "Group", "Service",
|
|
"Application", "Organization", "Create", "Update", "Delete", "Announce", "Like", "Dislike", "Follow", "Undo", "EmojiReact"
|
|
};
|
|
|
|
static readonly HashSet<string> HealthPurposes = new(StringComparer.Ordinal) { "actor", "key", "object" };
|
|
|
|
public static DayFold Fold(IEnumerable<InteractionEvent> events)
|
|
{
|
|
var day = new DayFold();
|
|
foreach (var e in events)
|
|
{
|
|
var host = e.Host ?? Interactions.Unknown;
|
|
if (!day.Hosts.TryGetValue(host, out var fold))
|
|
day.Hosts[host] = fold = new HostFold();
|
|
Count(fold, e);
|
|
if (e.ActorHash != default)
|
|
{
|
|
fold.Accounts.Add(e.ActorHash);
|
|
day.Accounts.Add(e.ActorHash);
|
|
}
|
|
}
|
|
return day;
|
|
}
|
|
|
|
static void Count(HostFold fold, InteractionEvent e)
|
|
{
|
|
var channel = e.Channel ?? Interactions.Other;
|
|
switch (channel)
|
|
{
|
|
case Interactions.Receive:
|
|
Add(fold.Counters, Key(channel, e.Activity, e.Status?.ToString(System.Globalization.CultureInfo.InvariantCulture), e.Reason));
|
|
if (e.Signature != default)
|
|
Add(fold.Counters, Key("sig", e.Signature));
|
|
break;
|
|
case Interactions.In or Interactions.Out:
|
|
Add(fold.Counters, Key(channel, e.Activity, e.Object, e.Outcome, e.Reason));
|
|
if (e.Audience != default)
|
|
Add(fold.Counters, Key("aud", channel, e.Audience));
|
|
if (e.LocalKind != default)
|
|
Add(fold.Counters, Key("kind", channel, e.LocalKind));
|
|
PublicActivity(fold, e, channel);
|
|
break;
|
|
case Interactions.Http:
|
|
Add(fold.Counters, Key(channel, e.Purpose, e.Outcome, e.Reason));
|
|
break;
|
|
default:
|
|
Add(fold.Counters, Key(channel, e.Outcome, e.Reason));
|
|
break;
|
|
}
|
|
if (Health(e) is { } health)
|
|
Add(fold.PublicCounters, Key("health", health));
|
|
foreach (var feature in e.Features ?? new List<string>())
|
|
{
|
|
Add(fold.Counters, Key("feat", feature));
|
|
if (e.Audience == Interactions.Public)
|
|
Add(fold.PublicCounters, Key("feat", feature));
|
|
}
|
|
var timing = channel == Interactions.Http ? Key(channel, e.Purpose) : channel;
|
|
if (e.LatencyMs is { } latency)
|
|
Add(fold.Latency, Key(timing, Interactions.Latency(latency)));
|
|
if (e.WaitMs is { } wait)
|
|
Add(fold.Waits, Key(channel, Interactions.Wait(wait)));
|
|
if (e.Bytes is { } bytes && bytes > 0)
|
|
Add(fold.Bytes, timing, bytes);
|
|
}
|
|
|
|
static void PublicActivity(HostFold fold, InteractionEvent e, string channel)
|
|
{
|
|
if (e.Audience is not (Interactions.Public or Interactions.Unlisted or Interactions.None)
|
|
|| e.Activity == default || !PublicActivities.Contains(e.Activity))
|
|
return;
|
|
var outcome = channel == Interactions.Out
|
|
? e.Outcome == Interactions.Ok ? "delivered" : e.Outcome is Interactions.Retry or Interactions.Dead or Interactions.Failed ? "failed" : default
|
|
: e.Outcome == Interactions.Accepted ? "accepted" : e.Outcome is Interactions.Dropped or Interactions.Rejected ? "dropped" : default;
|
|
if (outcome == default)
|
|
return;
|
|
var type = e.Object == default ? Interactions.Unknown : PublicObjects.Contains(e.Object) ? e.Object : Interactions.Other;
|
|
Add(fold.PublicCounters, Key(channel, e.Activity, type, outcome));
|
|
}
|
|
|
|
static string Health(InteractionEvent e) => e.Channel switch
|
|
{
|
|
Interactions.Out when e.Outcome == Interactions.Ok || e.Outcome == Interactions.Dead && e.Status is >= 400 and < 500 => "ok",
|
|
Interactions.Out when e.Outcome is Interactions.Retry or Interactions.Failed
|
|
|| e.Outcome == Interactions.Dead && (e.Status >= 500 || e.Reason is "timeout" or "network") => "failed",
|
|
Interactions.Http when HealthPurposes.Contains(e.Purpose ?? string.Empty) =>
|
|
e.Outcome is Interactions.Ok or Interactions.Refused ? "ok" : e.Outcome == Interactions.Failed ? "failed" : default,
|
|
_ => default
|
|
};
|
|
|
|
public static string Key(params string[] segments)
|
|
{
|
|
var count = segments.Length;
|
|
while (count > 1 && segments[count - 1] == default)
|
|
count--;
|
|
return string.Join(':', segments.Take(count).Select(s => (s ?? Interactions.Unknown).Replace('.', '_').Replace('$', '_')));
|
|
}
|
|
|
|
static void Add(Dictionary<string, long> counters, string key, long by = 1) =>
|
|
counters[key] = counters.GetValueOrDefault(key) + by;
|
|
}
|
|
}
|