ayurishchevandClaude Sonnet 5
582b44f314
Adds POST /api/v1/admin/ips/scan (plus an optional periodic ticker) to
discover free Floating IPs in the OpenStack project and feed them straight
into the check queue. More importantly, decouples check/event history from
ip_queue's lifecycle: a new ip_registry table (migration 0007) gives every
address ever submitted a durable identity, so deleting it from the queue no
longer destroys its history — it's still reachable via the new
GET /api/v1/admin/registry[/{ip}] endpoints and the dashboard's /registry
pages, with retention depth configurable in check cycles per address
(history_retention_cycles, 0 = unlimited).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
97 lines
3.5 KiB
Go
97 lines
3.5 KiB
Go
package db
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
)
|
|
|
|
// UpsertCheck records (or, on retry, overwrites) a single check result.
|
|
// registry_id/cycle_id are resolved from c.IPID's current ip_queue row at
|
|
// write time, so callers (agentcore/probercore) never need to know about
|
|
// the registry — see migrations/0007_ip_registry.sql. The
|
|
// UNIQUE(registry_id, cycle_id, source, check_type, target) constraint plus
|
|
// this upsert is what makes agent/prober result submission safely
|
|
// retryable without producing duplicate rows, now scoped to the durable
|
|
// per-address cycle rather than the ip_queue row's attempt_number, so it
|
|
// survives that row being deleted and the address later resubmitted.
|
|
func (d *DB) UpsertCheck(ctx context.Context, c Check) error {
|
|
_, err := d.ExecContext(ctx, `
|
|
INSERT INTO checks (registry_id, cycle_id, ip_id, ip_address, attempt_number, validator_id,
|
|
source, check_type, target, success, latency_ms, detail, checked_at, created_at)
|
|
VALUES ((SELECT registry_id FROM ip_queue WHERE id=?), (SELECT cycle_id FROM ip_queue WHERE id=?),
|
|
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(registry_id, cycle_id, source, check_type, target) DO UPDATE SET
|
|
validator_id=excluded.validator_id,
|
|
success=excluded.success,
|
|
latency_ms=excluded.latency_ms,
|
|
detail=excluded.detail,
|
|
checked_at=excluded.checked_at
|
|
`, c.IPID, c.IPID, c.IPID, c.IPAddress, c.AttemptNumber, c.ValidatorID, c.Source, c.CheckType, c.Target,
|
|
c.Success, c.LatencyMS, c.Detail, timeToDB(c.CheckedAt), timeToDB(Now()))
|
|
return err
|
|
}
|
|
|
|
const checksSelect = `
|
|
SELECT id, registry_id, cycle_id, ip_id, ip_address, attempt_number, validator_id, source, check_type, target,
|
|
success, latency_ms, detail, checked_at, created_at
|
|
FROM checks
|
|
`
|
|
|
|
// ListChecksForAttempt returns every check recorded for an IP's current
|
|
// attempt — the input to overall-result aggregation.
|
|
func (d *DB) ListChecksForAttempt(ctx context.Context, ipID int64, attemptNumber int) ([]Check, error) {
|
|
rows, err := d.QueryContext(ctx, checksSelect+`
|
|
WHERE ip_id=? AND attempt_number=?
|
|
ORDER BY source, check_type, target
|
|
`, ipID, attemptNumber)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
return scanChecks(rows)
|
|
}
|
|
|
|
// ListChecksForRegistry returns an address's check history across every
|
|
// cycle still retained (see PruneRegistryHistory), newest cycle first. A
|
|
// nil limit returns everything currently retained.
|
|
func (d *DB) ListChecksForRegistry(ctx context.Context, registryID int64, limit *int) ([]Check, error) {
|
|
query := checksSelect + `WHERE registry_id=? ORDER BY cycle_id DESC, source, check_type, target`
|
|
args := []interface{}{registryID}
|
|
if limit != nil {
|
|
query += ` LIMIT ?`
|
|
args = append(args, *limit)
|
|
}
|
|
rows, err := d.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
return scanChecks(rows)
|
|
}
|
|
|
|
func scanChecks(rows *sql.Rows) ([]Check, error) {
|
|
var out []Check
|
|
for rows.Next() {
|
|
var c Check
|
|
var ipID sql.NullInt64
|
|
var checkedAt, createdAt string
|
|
if err := rows.Scan(&c.ID, &c.RegistryID, &c.CycleID, &ipID, &c.IPAddress, &c.AttemptNumber,
|
|
&c.ValidatorID, &c.Source, &c.CheckType, &c.Target, &c.Success, &c.LatencyMS, &c.Detail,
|
|
&checkedAt, &createdAt); err != nil {
|
|
return nil, err
|
|
}
|
|
if ipID.Valid {
|
|
c.IPID = ipID.Int64
|
|
}
|
|
var err error
|
|
if c.CheckedAt, err = dbToTime(checkedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
if c.CreatedAt, err = dbToTime(createdAt); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, c)
|
|
}
|
|
return out, rows.Err()
|
|
}
|