Phase 7: telemetry and usage (daily rollup + week dashboard)
Build on Phases 1–6. Telemetry already arrived on
restrict/<device>/telemetry; this phase stores it in a daily rollup and
shows it.
- store: migration 003 adds telemetry_daily (device, day, app seconds,
site seconds). InsertTelemetry maintains the affected days in the same
transaction; RebuildRollups backfills and repairs it. UsageRange and
TopUsage serve the week grid and top lists. CapAudit enforces the
size-capped append-only audit log from AGENTS.md §7.
- web: GET /devices/{id}/usage — Monday-to-Sunday totals plus the top 10
apps and top 10 sites, using CSP-safe native <progress> bars.
- theme.js: light/dark from the OS preference, with a nav toggle, loaded
as an external script so the strict CSP still holds.
- cmd/control: `rollup` subcommand; rollup rebuild and audit cap at
start-up and every 15 minutes.
- Version 0.5.0-phase7 (compose, Makefile, docs).
Tests: go test ./... and go vet ./... pass in golang:1.25-alpine.
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
@@ -41,6 +41,14 @@ const republishInterval = 15 * time.Minute
|
||||
// the loop last looked.
|
||||
const republishPoll = 30 * time.Second
|
||||
|
||||
// Maintenance (Phase 7): the daily usage rollup is rebuilt periodically so it
|
||||
// is repaired from the detail table, and the append-only audit log is capped so
|
||||
// it cannot grow without bound (AGENTS.md §7 retention).
|
||||
const (
|
||||
maintenanceInterval = 15 * time.Minute
|
||||
auditMaxRows = 20000
|
||||
)
|
||||
|
||||
func main() {
|
||||
if len(os.Args) > 1 {
|
||||
switch os.Args[1] {
|
||||
@@ -50,6 +58,9 @@ func main() {
|
||||
case "mqtt-clients":
|
||||
mqttClients(os.Args[2:])
|
||||
return
|
||||
case "rollup":
|
||||
rollup(os.Args[2:])
|
||||
return
|
||||
case "version":
|
||||
fmt.Println(version)
|
||||
return
|
||||
@@ -83,6 +94,19 @@ func run() error {
|
||||
return err
|
||||
}
|
||||
|
||||
// Backfill the daily rollup from any telemetry written before Phase 7, then
|
||||
// cap the audit log. Both are idempotent.
|
||||
if n, err := st.RebuildRollups(context.Background()); err != nil {
|
||||
return err
|
||||
} else if n > 0 {
|
||||
log.Info("usage rollups rebuilt", "rows", n)
|
||||
}
|
||||
if n, err := st.CapAudit(context.Background(), auditMaxRows); err != nil {
|
||||
log.Error("audit cap", "err", err)
|
||||
} else if n > 0 {
|
||||
log.Info("audit log capped", "removed", n, "max", auditMaxRows)
|
||||
}
|
||||
|
||||
am, err := auth.New(cfg.SessionSecret, cfg.SessionTTL, cfg.CookieSecure,
|
||||
cfg.AdminUsername, cfg.AdminPasswordHash, cfg.AdminPassword)
|
||||
if err != nil {
|
||||
@@ -95,6 +119,8 @@ func run() error {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
go maintenanceLoop(ctx, log, st)
|
||||
|
||||
// The broker link (Phase 3). It is optional on purpose: the dashboard must
|
||||
// keep working while the broker is down, and the device keeps its last state
|
||||
// (AGENTS.md §10 rule 2).
|
||||
@@ -234,6 +260,66 @@ func republishLoop(ctx context.Context, log *slog.Logger, link *mqtt.Link, st *s
|
||||
}
|
||||
}
|
||||
|
||||
// maintenanceLoop keeps derived data healthy without blocking the service: it
|
||||
// rebuilds the daily rollups and caps the audit log on a fixed interval.
|
||||
func maintenanceLoop(ctx context.Context, log *slog.Logger, st *store.Store) {
|
||||
ticker := time.NewTicker(maintenanceInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
runMaintenance(ctx, log, st)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func runMaintenance(ctx context.Context, log *slog.Logger, st *store.Store) {
|
||||
mctx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
if n, err := st.RebuildRollups(mctx); err != nil {
|
||||
log.Error("usage rollups", "err", err)
|
||||
} else if n > 0 {
|
||||
log.Debug("usage rollups rebuilt", "rows", n)
|
||||
}
|
||||
if n, err := st.CapAudit(mctx, auditMaxRows); err != nil {
|
||||
log.Error("audit cap", "err", err)
|
||||
} else if n > 0 {
|
||||
log.Info("audit log capped", "removed", n, "max", auditMaxRows)
|
||||
}
|
||||
}
|
||||
|
||||
// rollup rebuilds the daily usage rollup from the telemetry detail table. It is
|
||||
// idempotent and safe to run while the service is up; it exits non-zero on a
|
||||
// failure so a script can tell.
|
||||
func rollup(args []string) {
|
||||
dbPath := env("CONTROL_DB_PATH", "/data/control.db")
|
||||
for i := 0; i < len(args); i++ {
|
||||
if (args[i] == "--db" || args[i] == "-db") && i+1 < len(args) {
|
||||
dbPath = args[i+1]
|
||||
i++
|
||||
}
|
||||
}
|
||||
|
||||
st, err := store.Open(dbPath)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "control:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer st.Close()
|
||||
if err := st.Migrate(context.Background()); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "control:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
n, err := st.RebuildRollups(context.Background())
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "control:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
fmt.Printf("rebuilt %d daily rollup rows from %s\n", n, dbPath)
|
||||
}
|
||||
|
||||
// mqttClients prints the MQTT identity of every registered device, one device
|
||||
// per line, as tab-separated fields:
|
||||
//
|
||||
|
||||
Reference in New Issue
Block a user