ayurishchev/cloud-ip-validator
1
0 Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Registry and Analytics: run, subnet, direction and protocol filters, successes-by-target chart

Browse Source 
Registry (/registry):
- filters by run (slice by the address's cycle in that run), subnet
  (drop-down of configured subnets), direction (egress/ingress) and
  protocol (icmp, tcp, ssh, https, tls); status in scope is computed over
  the narrowed checks
- chart "successful checks per target (egress) / site (ingress)" when both
  direction and protocol are chosen; a row opens the list of addresses
  (dialog, CSV)
- API: direction/protocol parameters and run in GET /admin/registry,
  GET /admin/registry/breakdown and /breakdown/list
- subnet filter passes ids as one JSON parameter (SQLite variable limit)

Analytics (/analytics):
- subnet filter recomputes the whole page over the addresses of the run
  inside the subnet; only their checks are read; cache per run and subnet
- direction and protocol focus the page; with both set the registry chart
  is shown
- subnet parameter in GET /admin/analytics/runs/{id} and lists (JSON, CSV)

Docs: plans and summaries in docs/changes, README, API, USAGE.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
ayurishchevandClaude Sonnet 5.5 committed 2026-10-06 14:23:48 +03:00
1 parent 068c10ea1c
commit ded196ec8d
40 files changed
+2545 -188
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/db/queries_registry.go
+188 -39
View File
@@ -3,8 +3,10 @@ package db
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"net/netip"
"slices"
"sort"
"strings"
"time"
@@ -106,28 +108,97 @@ func (d *DB) ListRegistry(ctx context.Context) ([]RegistrySummary, error) {
out := make([]RegistrySummary, len(items))
for i, item := range items {
s := RegistrySummary{RegistryItem: item}
if err := d.fillRegistrySummary(ctx, &s); err != nil {
if err := d.fillRegistrySummary(ctx, &s, registrySlice{}); err != nil {
return nil, err
}
out[i] = s
}
if err := d.fillRegistryLevels(ctx, out); err != nil {
if err := d.fillRegistryLevels(ctx, out, registrySlice{}); err != nil {
return nil, err
}
return out, nil
}
// RegistryFamilies lists the check families RegistryFilter.Family accepts.
// tls is an ingress-only family (the prober's TLS handshake on 443).
var RegistryFamilies = []string{"icmp", "tcp", "ssh", "https", "tls"}
// IsValidRegistryLevel reports whether s is a RegistryFilter.Level value.
func IsValidRegistryLevel(s string) bool { return s == LevelEgress || s == LevelIngress }
// IsValidRegistryFamily reports whether s is one of RegistryFamilies.
func IsValidRegistryFamily(s string) bool { return slices.Contains(RegistryFamilies, s) }
// RegistryFilter narrows ListRegistryPage. The zero value matches everything.
//
// RunID, Level and Family also choose which data of an address is read (the
// "slice"): with RunID, its cycle in that run (run_results.cycle_id) and the
// run's verdict; without, its newest cycle. Level and Family narrow the checks
// of that cycle: the address must have such checks, and LastResult then
// classifies just them (all succeeded = pass, none = fail, else partial).
type RegistryFilter struct {
Query string // substring of ip_address
LastResult string // pass|partial|fail|cancelled — same meaning as RegistrySummary.LastResult
RunID int64 // only addresses that have a result in this run
Subnet string // only addresses inside this CIDR
Level string // egress|ingress — only checks of this level (see CheckLevel)
Family string // one of RegistryFamilies — only checks of this family (see CheckFamily)
}
// registrySlice is the part of RegistryFilter that selects what is read of an
// address; the zero value is "newest cycle, all checks".
type registrySlice struct {
RunID int64
Level, Family string
}
func (sl registrySlice) scoped() bool { return sl.Level != "" || sl.Family != "" }
// validate rejects a Level or Family outside the accepted values.
func (sl registrySlice) validate() error {
if sl.Level != "" && !IsValidRegistryLevel(sl.Level) {
return fmt.Errorf("level %q: %w", sl.Level, ErrValidation)
}
if sl.Family != "" && !IsValidRegistryFamily(sl.Family) {
return fmt.Errorf("family %q: %w", sl.Family, ErrValidation)
}
return nil
}
// scopeCond is the SQL condition of Level and Family over `checks c`, "1 = 1"
// without them. Level and Family are validated, so the LIKE patterns are safe.
func (sl registrySlice) scopeCond() (string, []any) {
conds := []string{"1 = 1"}
var args []any
switch sl.Level {
case LevelEgress:
conds = append(conds, "c.source = 'egress'")
case LevelIngress:
conds = append(conds, "c.source LIKE 'inbound-site-%'")
}
if sl.Family != "" {
conds = append(conds, "(c.check_type = ? OR c.check_type LIKE ? || '-%')")
args = append(args, sl.Family, sl.Family)
}
return strings.Join(conds, " AND "), args
}
// scopeFrom is `FROM checks c WHERE …` over the checks of the slice of
// registry row r (with the run's run_results row as rr when RunID is set):
// the run's cycle or the newest one, narrowed by Level and Family.
func (sl registrySlice) scopeFrom() (string, []any) {
sc, args := sl.scopeCond()
if sl.RunID > 0 {
return `FROM checks c WHERE c.run_id = ? AND c.registry_id = r.id AND c.cycle_id = rr.cycle_id AND ` + sc,
append([]any{sl.RunID}, args...)
}
return `FROM checks c WHERE c.registry_id = r.id
AND c.cycle_id = (SELECT MAX(c2.cycle_id) FROM checks c2 WHERE c2.registry_id = r.id) AND ` + sc, args
}
// subnetIDs returns the registry ids of the addresses inside prefix. SQLite
// has no CIDR operators, so the registry's addresses are filtered here.
func (d *DB) subnetIDs(ctx context.Context, cidr string) ([]any, error) {
func (d *DB) subnetIDs(ctx context.Context, cidr string) ([]int64, error) {
p, err := netip.ParsePrefix(cidr)
if err != nil {
return nil, fmt.Errorf("subnet %q: %v: %w", cidr, err, ErrValidation)
@@ -137,7 +208,7 @@ func (d *DB) subnetIDs(ctx context.Context, cidr string) ([]any, error) {
return nil, err
}
defer rows.Close()
var ids []any
var ids []int64
for rows.Next() {
var id int64
var ip string
@@ -169,42 +240,88 @@ const lastResultCond = `(
) = ?)
)`
// ListRegistryPage returns one page (limit/offset) of the registry in the same
// order as ListRegistry, filtered by f, plus the total number of matching
// rows. LIMIT/OFFSET are applied in SQL before the per-row summary queries,
// so only the rows of the page pay for them. limit <= 0 means no limit.
func (d *DB) ListRegistryPage(ctx context.Context, f RegistryFilter, limit, offset int) ([]RegistrySummary, int, error) {
// registryFilterSQL is the FROM and the WHERE condition (without the keyword;
// "1 = 1" for no filter) of the addresses f selects, over `ip_registry r LEFT
// JOIN ip_queue q`, with the run's run_results row as rr when sl.RunID is set,
// and its arguments in order. ListRegistryPage and RegistryBreakdown share it, so
// the list and the chart over it always cover the same addresses.
func (d *DB) registryFilterSQL(ctx context.Context, f RegistryFilter, sl registrySlice) (from, cond string, args []any, err error) {
var conds []string
var args []any
if f.Query != "" {
conds = append(conds, "instr(r.ip_address, ?) > 0")
args = append(args, f.Query)
}
if f.LastResult != "" {
switch {
case f.LastResult != "" && sl.scoped():
// The verdict covers all checks, so a narrowed scope has none; cancelled
// leaves no checks at all.
if f.LastResult == ResultCancelled {
conds = append(conds, "0 = 1")
break
}
sfrom, fargs := sl.scopeFrom()
conds = append(conds, `(SELECT CASE WHEN SUM(c.success) = 0 THEN 'fail'
WHEN SUM(c.success) = COUNT(*) THEN 'pass' ELSE 'partial' END `+sfrom+` HAVING COUNT(*) > 0) = ?`)
args = append(args, fargs...)
args = append(args, f.LastResult)
case f.LastResult != "" && sl.RunID > 0:
conds = append(conds, "rr.verdict = ?")
args = append(args, f.LastResult)
case f.LastResult != "":
conds = append(conds, lastResultCond)
args = append(args, f.LastResult, f.LastResult)
}
if f.RunID > 0 {
conds = append(conds, "r.id IN (SELECT registry_id FROM run_results WHERE run_id = ?)")
args = append(args, f.RunID)
case sl.scoped():
sfrom, fargs := sl.scopeFrom()
conds = append(conds, "EXISTS (SELECT 1 "+sfrom+")")
args = append(args, fargs...)
}
if f.Subnet != "" {
ids, err := d.subnetIDs(ctx, f.Subnet)
if err != nil {
return nil, 0, err
return "", "", nil, err
}
if len(ids) == 0 {
conds = append(conds, "0 = 1")
} else {
conds = append(conds, "r.id IN ("+strings.TrimSuffix(strings.Repeat("?,", len(ids)), ",")+")")
args = append(args, ids...)
// One JSON parameter instead of an id per "?", which would hit
// SQLite's bound-variable limit on a large subnet.
b, err := json.Marshal(ids)
if err != nil {
return "", "", nil, err
}
conds = append(conds, "r.id IN (SELECT value FROM json_each(?))")
args = append(args, string(b))
}
}
from := ` FROM ip_registry r LEFT JOIN ip_queue q ON q.registry_id = r.id `
where := ""
if len(conds) > 0 {
where = "WHERE " + strings.Join(conds, " AND ") + " "
from = ` FROM ip_registry r LEFT JOIN ip_queue q ON q.registry_id = r.id `
if sl.RunID > 0 {
from += `JOIN run_results rr ON rr.registry_id = r.id AND rr.run_id = ? `
args = append([]any{sl.RunID}, args...)
}
cond = "1 = 1"
if len(conds) > 0 {
cond = strings.Join(conds, " AND ")
}
return from, cond, args, nil
}
// ListRegistryPage returns one page (limit/offset) of the registry in the same
// order as ListRegistry, filtered by f, plus the total number of matching
// rows. LIMIT/OFFSET are applied in SQL before the per-row summary queries,
// so only the rows of the page pay for them. limit <= 0 means no limit. With
// f.RunID the rows are the run's addresses and LastResult, LastCycleID, Egress
// and Ingress describe the address in that run; an unknown run is an empty
// list. Level and Family narrow Egress and Ingress to the matching checks.
func (d *DB) ListRegistryPage(ctx context.Context, f RegistryFilter, limit, offset int) ([]RegistrySummary, int, error) {
sl := registrySlice{RunID: f.RunID, Level: f.Level, Family: f.Family}
if err := sl.validate(); err != nil {
return nil, 0, err
}
from, cond, args, err := d.registryFilterSQL(ctx, f, sl)
if err != nil {
return nil, 0, err
}
where := "WHERE " + cond + " "
var total int
if err := d.QueryRowContext(ctx, `SELECT COUNT(*)`+from+where, args...).Scan(&total); err != nil {
@@ -231,12 +348,12 @@ func (d *DB) ListRegistryPage(ctx context.Context, f RegistryFilter, limit, offs
out := make([]RegistrySummary, len(items))
for i, item := range items {
s := RegistrySummary{RegistryItem: item}
if err := d.fillRegistrySummary(ctx, &s); err != nil {
if err := d.fillRegistrySummary(ctx, &s, sl); err != nil {
return nil, 0, err
}
out[i] = s
}
if err := d.fillRegistryLevels(ctx, out); err != nil {
if err := d.fillRegistryLevels(ctx, out, sl); err != nil {
return nil, 0, err
}
return out, total, nil
@@ -257,11 +374,11 @@ func (d *DB) GetRegistryByAddress(ctx context.Context, address string) (*Registr
return nil, err
}
s := &RegistrySummary{RegistryItem: *item}
if err := d.fillRegistrySummary(ctx, s); err != nil {
if err := d.fillRegistrySummary(ctx, s, registrySlice{}); err != nil {
return nil, err
}
one := []RegistrySummary{*s}
if err := d.fillRegistryLevels(ctx, one); err != nil {
if err := d.fillRegistryLevels(ctx, one, registrySlice{}); err != nil {
return nil, err
}
*s = one[0]
@@ -275,7 +392,10 @@ func (d *DB) GetRegistryByAddress(ctx context.Context, address string) (*Registr
// checked_at — a cycle with a mix of passing and failing checks (e.g. one
// egress target timed out while the rest succeeded) is "partial", even
// though the chronologically-last check to report in might have passed.
func (d *DB) fillRegistrySummary(ctx context.Context, s *RegistrySummary) error {
//
// With sl.RunID, LastResult and LastCycleID are instead the address's verdict
// and cycle in that run (from run_results); the rest is unchanged.
func (d *DB) fillRegistrySummary(ctx context.Context, s *RegistrySummary, sl registrySlice) error {
if err := d.QueryRowContext(ctx, `
SELECT COUNT(DISTINCT cycle_id) FROM checks WHERE registry_id=?
`, s.ID).Scan(&s.TotalCycles); err != nil {
@@ -305,6 +425,18 @@ func (d *DB) fillRegistrySummary(ctx context.Context, s *RegistrySummary) error
s.CurrentState = state.String
}
if sl.RunID > 0 {
var cycle sql.NullInt64
var verdict sql.NullString
err := d.QueryRowContext(ctx, `SELECT cycle_id, verdict FROM run_results WHERE run_id=? AND registry_id=?`, sl.RunID, s.ID).
Scan(&cycle, &verdict)
if err != nil && err != sql.ErrNoRows {
return err
}
s.LastResult, s.LastCycleID = verdict.String, int(cycle.Int64)
return nil
}
switch {
case overallResult.Valid && overallResult.String != "":
// The address has a live ip_queue row with a finished cycle
@@ -336,30 +468,46 @@ func (d *DB) fillRegistrySummary(ctx context.Context, s *RegistrySummary) error
// SQLite's bound-variable limit.
const registryLevelsChunk = 500
// registryLevelsQuery is the grouped query of fillRegistryLevels for n
// registry ids: per address, the counts of its newest cycle by source and
// check type.
func registryLevelsQuery(n int) string {
// registryLevelsQuery is the grouped query of fillRegistryLevels for the given
// registry ids and its arguments: per address, the counts of its cycle by
// source and check type — the newest one, or the run's with sl.RunID — over
// the checks that match sl.Level and sl.Family.
func registryLevelsQuery(ids []any, sl registrySlice) (string, []any) {
in := strings.TrimSuffix(strings.Repeat("?,", len(ids)), ",")
sc, scArgs := sl.scopeCond()
if sl.RunID > 0 {
args := append([]any{sl.RunID, sl.RunID}, ids...)
return `
SELECT c.registry_id, rr.cycle_id, c.source, c.check_type, COUNT(*), COALESCE(SUM(c.success), 0)
FROM checks c
JOIN run_results rr ON rr.run_id = ? AND rr.registry_id = c.registry_id AND rr.cycle_id = c.cycle_id
WHERE c.run_id = ? AND c.registry_id IN (` + in + `) AND ` + sc + `
GROUP BY c.registry_id, rr.cycle_id, c.source, c.check_type
`, append(args, scArgs...)
}
return `
SELECT c.registry_id, m.cid, c.source, c.check_type, COUNT(*), COALESCE(SUM(c.success), 0)
FROM checks c
JOIN (SELECT registry_id, MAX(cycle_id) AS cid FROM checks
WHERE registry_id IN (` + strings.TrimSuffix(strings.Repeat("?,", n), ",") + `)
WHERE registry_id IN (` + in + `)
GROUP BY registry_id) m
ON m.registry_id = c.registry_id AND m.cid = c.cycle_id
WHERE ` + sc + `
GROUP BY c.registry_id, m.cid, c.source, c.check_type
`
`, append(ids, scArgs...)
}
// fillRegistryLevels sets LastCycleID, Egress and Ingress on every summary in
// sums: the recorded checks of each address's newest cycle (the same cycle
// whose time is LastCheckedAt), counted per level and per check family. It
// runs one grouped query per chunk of addresses, not one per address, over
// idx_checks_registry_cycle. Checks whose source is neither egress nor an
// inbound site are not counted. The counts follow the recorded rows only, so
// they can differ from LastResult, which also treats missing results as
// failures.
func (d *DB) fillRegistryLevels(ctx context.Context, sums []RegistrySummary) error {
// idx_checks_registry_cycle. With sl.RunID the cycle is the address's one in
// that run (over idx_checks_run) and sl.Level and sl.Family leave only the
// matching checks, so the levels and types outside them stay empty. Checks
// whose source is neither egress nor an inbound site are not counted. The
// counts follow the recorded rows only, so they can differ from LastResult,
// which also treats missing results as failures.
func (d *DB) fillRegistryLevels(ctx context.Context, sums []RegistrySummary, sl registrySlice) error {
pos := make(map[int64]int, len(sums))
for i := range sums {
pos[sums[i].ID] = i
@@ -374,7 +522,8 @@ func (d *DB) fillRegistryLevels(ctx context.Context, sums []RegistrySummary) err
for _, s := range sums[start:end] {
args = append(args, s.ID)
}
rows, err := d.QueryContext(ctx, registryLevelsQuery(len(args)), args...)
q, qargs := registryLevelsQuery(args, sl)
rows, err := d.QueryContext(ctx, q, qargs...)
if err != nil {
return err
}