Add Floating IP scanning and a durable address registry with configurable history depth
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>
This commit is contained in:
1 parent
78b20fa5be
commit
582b44f314
47 files changed
+1781
-107
No files matched your search
@@ -78,6 +78,12 @@ type OrchestratorConfig struct {
|
||||
// field in this struct, it deliberately gets no nonzero default in
|
||||
// LoadControlAPI below.
|
||||
FIPSettleSeconds int `yaml:"fip_settle_seconds"`
|
||||
// FIPScanIntervalSeconds enables periodic Orchestrator.ScanFloatingIPs
|
||||
// runs on this interval when > 0. 0 (the default) disables periodic
|
||||
// scanning — an operator can still trigger a scan manually via
|
||||
// POST /api/v1/admin/ips/scan or the dashboard's "Scan Floating IPs"
|
||||
// button either way.
|
||||
FIPScanIntervalSeconds int `yaml:"fip_scan_interval_seconds"`
|
||||
}
|
||||
|
||||
type AggregationConfig struct {
|
||||
|
||||
@@ -136,6 +136,32 @@ func (c *client) ClearQueue(ctx context.Context) (clearQueueResponse, error) {
|
||||
return out, err
|
||||
}
|
||||
|
||||
// ScanFloatingIPs lists the OpenStack project's free (unassociated)
|
||||
// floating IPs and submits them to the check queue — see
|
||||
// orchestrator.ScanFloatingIPs.
|
||||
func (c *client) ScanFloatingIPs(ctx context.Context) (scanIPsResponse, error) {
|
||||
var out scanIPsResponse
|
||||
err := c.do(ctx, http.MethodPost, "/api/v1/admin/ips/scan", nil, &out)
|
||||
return out, err
|
||||
}
|
||||
|
||||
// ListRegistry returns every address ever submitted to the check queue,
|
||||
// each with a summary of its accumulated check history — survives an
|
||||
// address being deleted from the queue and later re-added.
|
||||
func (c *client) ListRegistry(ctx context.Context) ([]registryItem, error) {
|
||||
var out []registryItem
|
||||
err := c.do(ctx, http.MethodGet, "/api/v1/admin/registry", nil, &out)
|
||||
return out, err
|
||||
}
|
||||
|
||||
// GetRegistryHistory returns one address's registry record plus its full
|
||||
// retained check history across every cycle still kept.
|
||||
func (c *client) GetRegistryHistory(ctx context.Context, ip string) (registryHistoryResponse, error) {
|
||||
var out registryHistoryResponse
|
||||
err := c.do(ctx, http.MethodGet, "/api/v1/admin/registry/"+url.PathEscape(ip), nil, &out)
|
||||
return out, err
|
||||
}
|
||||
|
||||
func (c *client) ListValidators(ctx context.Context) ([]validatorDTO, error) {
|
||||
var out []validatorDTO
|
||||
err := c.do(ctx, http.MethodGet, "/api/v1/admin/config/validators", nil, &out)
|
||||
@@ -207,10 +233,10 @@ func (c *client) GetOrchestratorSettings(ctx context.Context) (orchestratorSetti
|
||||
return out, err
|
||||
}
|
||||
|
||||
func (c *client) PutOrchestratorSettings(ctx context.Context, fipSettleSeconds int) (orchestratorSettingsDTO, error) {
|
||||
func (c *client) PutOrchestratorSettings(ctx context.Context, fipSettleSeconds, historyRetentionCycles int) (orchestratorSettingsDTO, error) {
|
||||
var out orchestratorSettingsDTO
|
||||
err := c.do(ctx, http.MethodPut, "/api/v1/admin/config/orchestrator",
|
||||
orchestratorSettingsDTO{FIPSettleSeconds: fipSettleSeconds}, &out)
|
||||
orchestratorSettingsDTO{FIPSettleSeconds: fipSettleSeconds, HistoryRetentionCycles: historyRetentionCycles}, &out)
|
||||
return out, err
|
||||
}
|
||||
|
||||
|
||||
@@ -33,16 +33,26 @@ type fakeControlAPI struct {
|
||||
fipSettleSeconds int
|
||||
inboundPorts []int
|
||||
inboundICMP bool
|
||||
|
||||
historyRetentionCycles int
|
||||
// scanFreeAddresses is what POST /ips/scan "discovers" — tests set it
|
||||
// directly rather than this fake reimplementing OpenStack floating-IP
|
||||
// filtering (already covered by internal/orchestrator's own tests).
|
||||
scanFreeAddresses []string
|
||||
registry map[string]registryItem
|
||||
registryChecks map[string][]check
|
||||
}
|
||||
|
||||
func newFakeControlAPI(t *testing.T) (*fakeControlAPI, string) {
|
||||
t.Helper()
|
||||
f := &fakeControlAPI{
|
||||
sites: map[int]string{},
|
||||
siteState: map[int]string{},
|
||||
siteHostname: map[int]string{},
|
||||
groups: map[string][]string{},
|
||||
checkTypes: map[string]checkTypeDTO{},
|
||||
sites: map[int]string{},
|
||||
siteState: map[int]string{},
|
||||
siteHostname: map[int]string{},
|
||||
groups: map[string][]string{},
|
||||
checkTypes: map[string]checkTypeDTO{},
|
||||
registry: map[string]registryItem{},
|
||||
registryChecks: map[string][]check{},
|
||||
}
|
||||
ts := httptest.NewServer(f.handler())
|
||||
t.Cleanup(ts.Close)
|
||||
@@ -197,7 +207,10 @@ func (f *fakeControlAPI) handler() http.Handler {
|
||||
mux.HandleFunc("GET /api/v1/admin/config/orchestrator", func(w http.ResponseWriter, r *http.Request) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
writeJSON(w, http.StatusOK, orchestratorSettingsDTO{FIPSettleSeconds: f.fipSettleSeconds})
|
||||
writeJSON(w, http.StatusOK, orchestratorSettingsDTO{
|
||||
FIPSettleSeconds: f.fipSettleSeconds,
|
||||
HistoryRetentionCycles: f.historyRetentionCycles,
|
||||
})
|
||||
})
|
||||
mux.HandleFunc("PUT /api/v1/admin/config/orchestrator", func(w http.ResponseWriter, r *http.Request) {
|
||||
f.mu.Lock()
|
||||
@@ -208,8 +221,54 @@ func (f *fakeControlAPI) handler() http.Handler {
|
||||
writeAPIErr(w, http.StatusBadRequest, "fip_settle_seconds must be >= 0")
|
||||
return
|
||||
}
|
||||
if req.HistoryRetentionCycles < 0 {
|
||||
writeAPIErr(w, http.StatusBadRequest, "history_retention_cycles must be >= 0")
|
||||
return
|
||||
}
|
||||
f.fipSettleSeconds = req.FIPSettleSeconds
|
||||
writeJSON(w, http.StatusOK, orchestratorSettingsDTO{FIPSettleSeconds: f.fipSettleSeconds})
|
||||
f.historyRetentionCycles = req.HistoryRetentionCycles
|
||||
writeJSON(w, http.StatusOK, orchestratorSettingsDTO{
|
||||
FIPSettleSeconds: f.fipSettleSeconds,
|
||||
HistoryRetentionCycles: f.historyRetentionCycles,
|
||||
})
|
||||
})
|
||||
|
||||
mux.HandleFunc("POST /api/v1/admin/ips/scan", func(w http.ResponseWriter, r *http.Request) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
resp := scanIPsResponse{ScannedFree: len(f.scanFreeAddresses)}
|
||||
for _, addr := range f.scanFreeAddresses {
|
||||
idx := f.findIP(addr)
|
||||
if idx < 0 {
|
||||
now := time.Now()
|
||||
f.ips = append(f.ips, ipQueueItem{IPAddress: addr, State: "queued", CreatedAt: now, UpdatedAt: now})
|
||||
resp.Added = append(resp.Added, addr)
|
||||
continue
|
||||
}
|
||||
resp.Reordered = append(resp.Reordered, addr)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, resp)
|
||||
})
|
||||
|
||||
mux.HandleFunc("GET /api/v1/admin/registry", func(w http.ResponseWriter, r *http.Request) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
out := make([]registryItem, 0, len(f.registry))
|
||||
for _, item := range f.registry {
|
||||
out = append(out, item)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
})
|
||||
mux.HandleFunc("GET /api/v1/admin/registry/{ip}", func(w http.ResponseWriter, r *http.Request) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
addr := r.PathValue("ip")
|
||||
item, ok := f.registry[addr]
|
||||
if !ok {
|
||||
writeAPIErr(w, http.StatusNotFound, "unknown ip: "+addr)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, registryHistoryResponse{Registry: item, Checks: f.registryChecks[addr]})
|
||||
})
|
||||
|
||||
mux.HandleFunc("GET /api/v1/admin/config/inbound-checks", func(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
@@ -41,6 +41,8 @@ type ipQueueItem struct {
|
||||
|
||||
type check struct {
|
||||
ID int64
|
||||
RegistryID int64
|
||||
CycleID int
|
||||
IPID int64
|
||||
IPAddress string
|
||||
AttemptNumber int
|
||||
@@ -96,6 +98,33 @@ type clearQueueResponse struct {
|
||||
Deleted []string `json:"deleted"`
|
||||
}
|
||||
|
||||
type scanIPsResponse struct {
|
||||
ScannedFree int `json:"scanned_free"`
|
||||
Added []string `json:"added"`
|
||||
Requeued []string `json:"requeued"`
|
||||
Reordered []string `json:"reordered"`
|
||||
SkippedInProgress []string `json:"skipped_in_progress"`
|
||||
}
|
||||
|
||||
// registryItem is one row of the durable per-address registry — see
|
||||
// httpapi's registryDTO. Survives an address being deleted from the check
|
||||
// queue and later re-added, unlike ipQueueItem above.
|
||||
type registryItem struct {
|
||||
IPAddress string `json:"ip_address"`
|
||||
FirstSeenAt time.Time `json:"first_seen_at"`
|
||||
LastSeenAt time.Time `json:"last_seen_at"`
|
||||
TotalCycles int `json:"total_cycles"`
|
||||
LastResult string `json:"last_result"`
|
||||
LastCheckedAt *time.Time `json:"last_checked_at"`
|
||||
InQueue bool `json:"in_queue"`
|
||||
CurrentState string `json:"current_state"`
|
||||
}
|
||||
|
||||
type registryHistoryResponse struct {
|
||||
Registry registryItem `json:"registry"`
|
||||
Checks []check `json:"checks"`
|
||||
}
|
||||
|
||||
type validatorDTO struct {
|
||||
ValidatorID string `json:"validator_id"`
|
||||
Hostname string `json:"hostname"`
|
||||
@@ -128,7 +157,8 @@ type errorResponse struct {
|
||||
}
|
||||
|
||||
type orchestratorSettingsDTO struct {
|
||||
FIPSettleSeconds int `json:"fip_settle_seconds"`
|
||||
FIPSettleSeconds int `json:"fip_settle_seconds"`
|
||||
HistoryRetentionCycles int `json:"history_retention_cycles"`
|
||||
}
|
||||
|
||||
type inboundChecksDTO struct {
|
||||
|
||||
@@ -117,3 +117,11 @@ func (s *Server) handleIPsClear(w http.ResponseWriter, r *http.Request) {
|
||||
_, err := s.CA.ClearQueue(r.Context())
|
||||
s.renderIPsTable(w, r, err)
|
||||
}
|
||||
|
||||
// handleIPsScan lists the OpenStack project's free (unassociated) floating
|
||||
// IPs and submits them to the check queue in one step — see
|
||||
// client.ScanFloatingIPs.
|
||||
func (s *Server) handleIPsScan(w http.ResponseWriter, r *http.Request) {
|
||||
_, err := s.CA.ScanFloatingIPs(r.Context())
|
||||
s.renderIPsTable(w, r, err)
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package dashboard
|
||||
|
||||
import "net/http"
|
||||
|
||||
type registryPageData struct {
|
||||
PageData
|
||||
Items []registryItem
|
||||
}
|
||||
|
||||
type registryDetailData struct {
|
||||
PageData
|
||||
History registryHistoryResponse
|
||||
}
|
||||
|
||||
// handleRegistryPage lists every address ever submitted to the check
|
||||
// queue, with a summary of its accumulated check history — the durable
|
||||
// record that survives an address being deleted from /ips and later
|
||||
// re-added. See internal/db/migrations/0007_ip_registry.sql.
|
||||
func (s *Server) handleRegistryPage(w http.ResponseWriter, r *http.Request) {
|
||||
items, err := s.CA.ListRegistry(r.Context())
|
||||
data := registryPageData{Items: items}
|
||||
data.ActiveNav = "registry"
|
||||
data.Banner = bannerFor(err)
|
||||
s.renderPage(w, "registry_page", data)
|
||||
}
|
||||
|
||||
// handleRegistryDetail shows one address's full retained check history
|
||||
// across every cycle it has ever run, not just the current attempt — see
|
||||
// ip_detail_content in ip_detail.html for the attempt-scoped equivalent.
|
||||
func (s *Server) handleRegistryDetail(w http.ResponseWriter, r *http.Request) {
|
||||
ip := r.PathValue("ip")
|
||||
history, err := s.CA.GetRegistryHistory(r.Context(), ip)
|
||||
data := registryDetailData{History: history}
|
||||
data.ActiveNav = "registry"
|
||||
data.Banner = bannerFor(err)
|
||||
s.renderPage(w, "registry_detail_page", data)
|
||||
}
|
||||
@@ -51,7 +51,12 @@ func (s *Server) handleSettingsPut(w http.ResponseWriter, r *http.Request) {
|
||||
s.renderSettingsForm(w, r, &apiErr{Status: http.StatusBadRequest, Message: "пауза должна быть целым числом секунд"})
|
||||
return
|
||||
}
|
||||
_, err = s.CA.PutOrchestratorSettings(r.Context(), seconds)
|
||||
retentionCycles, err := strconv.Atoi(r.PostFormValue("history_retention_cycles"))
|
||||
if err != nil {
|
||||
s.renderSettingsForm(w, r, &apiErr{Status: http.StatusBadRequest, Message: "глубина истории должна быть целым числом циклов"})
|
||||
return
|
||||
}
|
||||
_, err = s.CA.PutOrchestratorSettings(r.Context(), seconds, retentionCycles)
|
||||
s.renderSettingsForm(w, r, err)
|
||||
}
|
||||
|
||||
|
||||
@@ -230,27 +230,44 @@ func TestSettingsGetAndPut(t *testing.T) {
|
||||
t.Fatalf("expected default 0 in the form, got:\n%s", page)
|
||||
}
|
||||
|
||||
body := postForm(t, ts, "PUT", "/settings", map[string][]string{"fip_settle_seconds": {"15"}})
|
||||
if !strings.Contains(body, `value="15"`) {
|
||||
t.Fatalf("expected updated value 15 in re-rendered form, got:\n%s", body)
|
||||
body := postForm(t, ts, "PUT", "/settings", map[string][]string{
|
||||
"fip_settle_seconds": {"15"}, "history_retention_cycles": {"10"},
|
||||
})
|
||||
if !strings.Contains(body, `value="15"`) || !strings.Contains(body, `value="10"`) {
|
||||
t.Fatalf("expected updated values 15/10 in re-rendered form, got:\n%s", body)
|
||||
}
|
||||
if fake.fipSettleSeconds != 15 {
|
||||
t.Fatalf("expected fake control-api settings updated, got %d", fake.fipSettleSeconds)
|
||||
}
|
||||
if fake.historyRetentionCycles != 10 {
|
||||
t.Fatalf("expected fake control-api history_retention_cycles updated, got %d", fake.historyRetentionCycles)
|
||||
}
|
||||
|
||||
// A control-api validation error (negative value here) surfaces via
|
||||
// the banner, not a crash.
|
||||
body = postForm(t, ts, "PUT", "/settings", map[string][]string{"fip_settle_seconds": {"-1"}})
|
||||
body = postForm(t, ts, "PUT", "/settings", map[string][]string{
|
||||
"fip_settle_seconds": {"-1"}, "history_retention_cycles": {"10"},
|
||||
})
|
||||
if !strings.Contains(body, "alert-warning") {
|
||||
t.Fatalf("expected client error banner for invalid value, got:\n%s", body)
|
||||
}
|
||||
|
||||
// A non-numeric value is caught by the dashboard itself before it ever
|
||||
// reaches control-api.
|
||||
body = postForm(t, ts, "PUT", "/settings", map[string][]string{"fip_settle_seconds": {"not-a-number"}})
|
||||
body = postForm(t, ts, "PUT", "/settings", map[string][]string{
|
||||
"fip_settle_seconds": {"not-a-number"}, "history_retention_cycles": {"10"},
|
||||
})
|
||||
if !strings.Contains(body, "alert-warning") {
|
||||
t.Fatalf("expected client error banner for non-numeric value, got:\n%s", body)
|
||||
}
|
||||
|
||||
// Same for a non-numeric retention value.
|
||||
body = postForm(t, ts, "PUT", "/settings", map[string][]string{
|
||||
"fip_settle_seconds": {"15"}, "history_retention_cycles": {"not-a-number"},
|
||||
})
|
||||
if !strings.Contains(body, "alert-warning") {
|
||||
t.Fatalf("expected client error banner for non-numeric retention value, got:\n%s", body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSettingsPageShowsInboundChecks(t *testing.T) {
|
||||
@@ -463,6 +480,52 @@ func TestTargetsAndCheckTypesRoundTrip(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestIPsScan(t *testing.T) {
|
||||
fake, caURL := newFakeControlAPI(t)
|
||||
fake.scanFreeAddresses = []string{"5.5.5.5"}
|
||||
ts := newTestServer(t, caURL)
|
||||
|
||||
body := postForm(t, ts, "POST", "/ips/scan", nil)
|
||||
if !strings.Contains(body, "5.5.5.5") {
|
||||
t.Fatalf("expected scanned address in re-rendered table, got:\n%s", body)
|
||||
}
|
||||
if len(fake.ips) != 1 || fake.ips[0].IPAddress != "5.5.5.5" {
|
||||
t.Fatalf("expected fake control-api queue to contain the scanned address, got %+v", fake.ips)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRegistryPageAndDetail(t *testing.T) {
|
||||
fake, caURL := newFakeControlAPI(t)
|
||||
now := time.Now()
|
||||
fake.registry["9.9.9.9"] = registryItem{
|
||||
IPAddress: "9.9.9.9", FirstSeenAt: now, LastSeenAt: now,
|
||||
TotalCycles: 2, LastResult: "pass", InQueue: false,
|
||||
}
|
||||
fake.registryChecks["9.9.9.9"] = []check{
|
||||
{CycleID: 2, Source: "egress", CheckType: "https", Target: "https://example.test", Success: true, CheckedAt: now},
|
||||
{CycleID: 1, Source: "egress", CheckType: "https", Target: "https://example.test", Success: false, CheckedAt: now},
|
||||
}
|
||||
ts := newTestServer(t, caURL)
|
||||
|
||||
page := get(t, ts, "/registry")
|
||||
if !strings.Contains(page, "9.9.9.9") || !strings.Contains(page, "/registry/9.9.9.9") {
|
||||
t.Fatalf("expected registry list to show the address with a detail link, got:\n%s", page)
|
||||
}
|
||||
|
||||
detail := get(t, ts, "/registry/9.9.9.9")
|
||||
if !strings.Contains(detail, "9.9.9.9") {
|
||||
t.Fatalf("expected detail page to show the address, got:\n%s", detail)
|
||||
}
|
||||
if !strings.Contains(detail, "https://example.test") {
|
||||
t.Fatalf("expected detail page to list retained checks, got:\n%s", detail)
|
||||
}
|
||||
|
||||
notFound := get(t, ts, "/registry/1.2.3.4")
|
||||
if !strings.Contains(notFound, "alert-warning") {
|
||||
t.Fatalf("expected client error banner for unknown registry address, got:\n%s", notFound)
|
||||
}
|
||||
}
|
||||
|
||||
func TestControlAPIUnreachable(t *testing.T) {
|
||||
// Point the dashboard at an address nothing listens on, rather than a
|
||||
// closed httptest.Server, to get a deterministic connection-refused
|
||||
|
||||
@@ -11,6 +11,7 @@ func (s *Server) routes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("GET /ips", s.handleIPsPage)
|
||||
mux.HandleFunc("GET /ips/{ip}", s.handleIPDetail)
|
||||
mux.HandleFunc("POST /ips", s.handleIPsSubmit)
|
||||
mux.HandleFunc("POST /ips/scan", s.handleIPsScan)
|
||||
mux.HandleFunc("POST /ips/{ip}/recheck", s.handleIPRecheck)
|
||||
mux.HandleFunc("POST /ips/{ip}/cancel", s.handleIPCancel)
|
||||
mux.HandleFunc("DELETE /ips/{ip}", s.handleIPDelete)
|
||||
@@ -18,6 +19,9 @@ func (s *Server) routes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("POST /ips/recheck", s.handleIPsRecheckSelected)
|
||||
mux.HandleFunc("POST /ips/clear", s.handleIPsClear)
|
||||
|
||||
mux.HandleFunc("GET /registry", s.handleRegistryPage)
|
||||
mux.HandleFunc("GET /registry/{ip}", s.handleRegistryDetail)
|
||||
|
||||
mux.HandleFunc("GET /validators", s.handleValidatorsPage)
|
||||
mux.HandleFunc("POST /validators", s.handleValidatorCreate)
|
||||
mux.HandleFunc("PUT /validators/{id}", s.handleValidatorUpdate)
|
||||
|
||||
@@ -25,6 +25,7 @@
|
||||
<div class="topbar">
|
||||
<h1 class="mono" style="display:flex;align-items:center;gap:10px">{{.Detail.IP.IPAddress}} <span class="pill {{$b.Class}}">{{$b.Label}}</span></h1>
|
||||
</div>
|
||||
<p class="muted"><a href="/registry/{{.Detail.IP.IPAddress}}">Полная история проверок этого адреса за всё время →</a></p>
|
||||
<p class="muted">
|
||||
попытка {{.Detail.IP.AttemptNumber}} (повторов: {{.Detail.IP.RetryCount}})
|
||||
· валидатор {{deref .Detail.IP.OwnerValidatorID}}
|
||||
|
||||
@@ -34,7 +34,10 @@
|
||||
</div>
|
||||
</form>
|
||||
<p class="muted" style="margin-top:8px">Новый адрес встаёт в очередь; уже завершённый (done/failed) запускается заново
|
||||
(тот же вызов); тот, что сейчас проверяется, не трогается.</p>
|
||||
(тот же вызов); тот, что сейчас проверяется, не трогается. Кнопка «Сканировать Floating IP» ниже делает то же самое
|
||||
автоматически: находит в проекте OpenStack все свободные (не привязанные к порту) Floating IP и сразу передаёт их
|
||||
в очередь на проверку. Полная история проверок по каждому адресу — включая уже удалённые из очереди — доступна в
|
||||
<a href="/registry">реестре</a>.</p>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@@ -47,6 +50,7 @@
|
||||
<form id="ips-form">
|
||||
<div class="panel">
|
||||
<div class="panel-body" style="display:flex; gap:8px; align-items:center;">
|
||||
<button type="button" class="btn btn-ghost btn-sm" hx-post="/ips/scan" hx-target="#ips-table-wrap" hx-swap="innerHTML" hx-sync="#ips-table-wrap:queue last">Сканировать Floating IP</button>
|
||||
<button type="button" class="btn btn-ghost btn-sm" hx-post="/ips/recheck" hx-target="#ips-table-wrap" hx-swap="innerHTML" hx-sync="#ips-table-wrap:queue last">Перепроверить выбранные</button>
|
||||
<button type="button" class="btn btn-danger-ghost btn-sm" hx-post="/ips/delete" hx-target="#ips-table-wrap" hx-swap="innerHTML" hx-sync="#ips-table-wrap:queue last" hx-confirm="Удалить выбранные адреса без возможности восстановления?">Удалить выбранные</button>
|
||||
<button type="button" class="btn btn-danger-ghost btn-sm" hx-post="/ips/clear" hx-target="#ips-table-wrap" hx-swap="innerHTML" hx-sync="#ips-table-wrap:queue last" hx-confirm="Удалить ВСЕ адреса из очереди, включая те, что сейчас проверяются? Действие необратимо.">Очистить всё</button>
|
||||
|
||||
@@ -52,6 +52,10 @@
|
||||
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><rect x="3" y="4" width="18" height="16" rx="2"/><path d="M3 10h18M8 4v16"/></svg>
|
||||
Очередь IP
|
||||
</a>
|
||||
<a class="nav-link{{if eq .ActiveNav "registry"}} active{{end}}" {{if eq .ActiveNav "registry"}}aria-current="page"{{end}} href="/registry">
|
||||
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M4 19.5A2.5 2.5 0 0 1 6.5 17H20"/><path d="M6.5 2H20v20H6.5A2.5 2.5 0 0 1 4 19.5v-15A2.5 2.5 0 0 1 6.5 2Z"/></svg>
|
||||
Реестр
|
||||
</a>
|
||||
<a class="nav-link{{if eq .ActiveNav "validators"}} active{{end}}" {{if eq .ActiveNav "validators"}}aria-current="page"{{end}} href="/validators">
|
||||
<svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><rect x="4" y="3" width="16" height="7" rx="1.5"/><rect x="4" y="14" width="16" height="7" rx="1.5"/><circle cx="8" cy="6.5" r="1"/><circle cx="8" cy="17.5" r="1"/></svg>
|
||||
Валидаторы
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
{{define "registry_page"}}
|
||||
<!doctype html>
|
||||
<html lang="ru">
|
||||
<head>{{template "html_head" .}}</head>
|
||||
<body>
|
||||
<div class="bg-grid"></div>
|
||||
<input type="checkbox" id="nav-toggle" class="nav-toggle">
|
||||
<div class="shell">
|
||||
{{template "sidebar_nav" .}}
|
||||
<div>
|
||||
{{template "topbar_mobile" .}}
|
||||
<main class="main">
|
||||
<div id="error-banner">{{template "banner_inner" .Banner}}</div>
|
||||
{{template "registry_content" .}}
|
||||
</main>
|
||||
</div>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
{{end}}
|
||||
|
||||
{{define "registry_content"}}
|
||||
<div class="topbar"><h1>Реестр IP</h1></div>
|
||||
<p class="muted" style="margin-bottom:16px">Все адреса, когда-либо поставленные на проверку — накопленная статистика
|
||||
сохраняется здесь даже после удаления адреса из <a href="/ips">очереди</a> и не теряется при повторном добавлении.
|
||||
Глубина хранимой истории на адрес настраивается на <a href="/settings">странице настроек</a>.</p>
|
||||
|
||||
{{if .Items}}
|
||||
<div class="panel">
|
||||
<div class="table-scroll">
|
||||
<table>
|
||||
<thead><tr><th>Адрес</th><th>Впервые замечен</th><th>Последний раз замечен</th><th>Циклов</th><th>Последний результат</th><th>Сейчас в очереди</th></tr></thead>
|
||||
<tbody>
|
||||
{{range .Items}}
|
||||
<tr>
|
||||
<td class="addr" data-label="Адрес"><a href="/registry/{{.IPAddress}}">{{.IPAddress}}</a></td>
|
||||
<td data-label="Впервые замечен">{{fmtTime .FirstSeenAt}}</td>
|
||||
<td data-label="Последний раз замечен">{{fmtTime .LastSeenAt}}</td>
|
||||
<td class="num" data-label="Циклов">{{.TotalCycles}}</td>
|
||||
<td data-label="Последний результат">
|
||||
{{if eq .LastResult "pass"}}<span class="pill pill-success">pass</span>
|
||||
{{else if eq .LastResult "fail"}}<span class="pill pill-danger">fail</span>
|
||||
{{else}}<span class="pill pill-neutral">—</span>{{end}}
|
||||
</td>
|
||||
<td data-label="Сейчас в очереди">
|
||||
{{if .InQueue}}<span class="pill pill-info">{{.CurrentState}}</span>{{else}}<span class="pill pill-neutral">нет</span>{{end}}
|
||||
</td>
|
||||
</tr>
|
||||
{{end}}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</div>
|
||||
{{else}}<p class="muted">Реестр пуст — ни один адрес ещё не ставился на проверку.</p>{{end}}
|
||||
{{end}}
|
||||
@@ -0,0 +1,56 @@
|
||||
{{define "registry_detail_page"}}
|
||||
<!doctype html>
|
||||
<html lang="ru">
|
||||
<head>{{template "html_head" .}}</head>
|
||||
<body>
|
||||
<div class="bg-grid"></div>
|
||||
<input type="checkbox" id="nav-toggle" class="nav-toggle">
|
||||
<div class="shell">
|
||||
{{template "sidebar_nav" .}}
|
||||
<div>
|
||||
{{template "topbar_mobile" .}}
|
||||
<main class="main">
|
||||
<div id="error-banner">{{template "banner_inner" .Banner}}</div>
|
||||
{{template "registry_detail_content" .}}
|
||||
</main>
|
||||
</div>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
{{end}}
|
||||
|
||||
{{define "registry_detail_content"}}
|
||||
<p><a href="/registry">← К реестру</a></p>
|
||||
<div class="topbar">
|
||||
<h1 class="mono" style="display:flex;align-items:center;gap:10px">{{.History.Registry.IPAddress}}
|
||||
{{if .History.Registry.InQueue}}<a class="pill pill-info" href="/ips/{{.History.Registry.IPAddress}}">{{.History.Registry.CurrentState}} · в очереди</a>{{else}}<span class="pill pill-neutral">не в очереди</span>{{end}}
|
||||
</h1>
|
||||
</div>
|
||||
<p class="muted">
|
||||
впервые замечен: {{fmtTime .History.Registry.FirstSeenAt}} · последний раз замечен: {{fmtTime .History.Registry.LastSeenAt}}
|
||||
· всего циклов проверки: {{.History.Registry.TotalCycles}}
|
||||
</p>
|
||||
|
||||
<h2 class="section-title">История проверок (все сохранённые циклы)</h2>
|
||||
{{if .History.Checks}}
|
||||
<div class="panel">
|
||||
<div class="table-scroll">
|
||||
<table>
|
||||
<thead><tr><th>Цикл</th><th>Источник</th><th>Тип</th><th>Цель</th><th>Успех</th><th>Задержка, мс</th><th>Время</th><th>Детали</th></tr></thead>
|
||||
<tbody>
|
||||
{{range .History.Checks}}
|
||||
<tr>
|
||||
<td class="num" data-label="Цикл">{{.CycleID}}</td>
|
||||
<td data-label="Источник">{{.Source}}</td><td data-label="Тип">{{.CheckType}}</td><td class="addr" data-label="Цель">{{.Target}}</td>
|
||||
<td data-label="Успех">{{if .Success}}<span class="pill pill-success">ok</span>{{else}}<span class="pill pill-danger">fail</span>{{end}}</td>
|
||||
<td class="num" data-label="Задержка, мс">{{.LatencyMS}}</td><td class="num" data-label="Время">{{fmtTime .CheckedAt}}</td><td data-label="Детали">{{.Detail}}</td>
|
||||
</tr>
|
||||
{{end}}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</div>
|
||||
<p class="muted" style="margin-top:8px">Если глубина истории (см. <a href="/settings">настройки</a>) ограничена, здесь
|
||||
показаны только сохранённые циклы — более старые уже вытеснены.</p>
|
||||
{{else}}<p class="muted">Проверок пока нет.</p>{{end}}
|
||||
{{end}}
|
||||
@@ -35,12 +35,20 @@
|
||||
проверять его. <code>0</code> — без паузы (поведение по умолчанию). Значение должно оставлять запас внутри лизинга
|
||||
адреса: <code>fip_settle_seconds + self_check_timeout_seconds < lease_ttl_seconds</code> — иначе адрес не успеет
|
||||
пройти self-check до истечения лизинга и будет возвращён в очередь.</p>
|
||||
<p class="muted" style="margin-bottom:16px">Сколько последних циклов проверки хранить в
|
||||
<a href="/registry">реестре</a> для каждого адреса. Адрес и его накопленная статистика остаются в реестре даже
|
||||
после удаления из очереди — эта настройка ограничивает только глубину истории конкретных проверок, не сам реестр.
|
||||
<code>0</code> — хранить без ограничения.</p>
|
||||
<form hx-put="/settings" hx-target="#settings-form-wrap" hx-swap="innerHTML">
|
||||
<div class="field-row">
|
||||
<div class="field">
|
||||
<label for="fip_settle_seconds">Пауза перед self-check (сек)</label>
|
||||
<input type="number" id="fip_settle_seconds" name="fip_settle_seconds" min="0" step="1" value="{{.Settings.FIPSettleSeconds}}" required>
|
||||
</div>
|
||||
<div class="field">
|
||||
<label for="history_retention_cycles">Глубина истории проверок (циклов на адрес, 0 = не ограничено)</label>
|
||||
<input type="number" id="history_retention_cycles" name="history_retention_cycles" min="0" step="1" value="{{.Settings.HistoryRetentionCycles}}" required>
|
||||
</div>
|
||||
<button type="submit" class="btn btn-primary">Сохранить</button>
|
||||
</div>
|
||||
</form>
|
||||
|
||||
@@ -31,6 +31,9 @@ var unboundedSitesSchema string
|
||||
//go:embed migrations/0006_prober_heartbeat.sql
|
||||
var proberHeartbeatSchema string
|
||||
|
||||
//go:embed migrations/0007_ip_registry.sql
|
||||
var ipRegistrySchema string
|
||||
|
||||
// migrations is the ordered list of schema versions. Each entry's SQL is
|
||||
// applied, in order, for any version greater than the database's current
|
||||
// PRAGMA user_version — so a fresh database walks the whole list and an
|
||||
@@ -45,6 +48,7 @@ var migrations = []struct {
|
||||
{4, inboundChecksAdminSchema},
|
||||
{5, unboundedSitesSchema},
|
||||
{6, proberHeartbeatSchema},
|
||||
{7, ipRegistrySchema},
|
||||
}
|
||||
|
||||
type DB struct {
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
-- Durable per-address registry, decoupled from ip_queue's lifecycle: today
|
||||
-- deleting an address from ip_queue (DeleteIP/DeleteIPs/ClearQueue) cascades
|
||||
-- to a hard DELETE of its checks/events, so history is lost forever if an
|
||||
-- address is removed and later re-added. ip_registry gives every address
|
||||
-- ever submitted a durable identity that check/event history attaches to
|
||||
-- instead, surviving ip_queue row deletion and recreation.
|
||||
|
||||
CREATE TABLE ip_registry (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
ip_address TEXT NOT NULL UNIQUE,
|
||||
first_seen_at TIMESTAMP NOT NULL,
|
||||
last_seen_at TIMESTAMP NOT NULL,
|
||||
next_cycle INTEGER NOT NULL DEFAULT 1,
|
||||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
-- Backfill: one registry row per address currently (or ever) in ip_queue.
|
||||
-- ip_queue.ip_address is already UNIQUE, so this is a straight 1:1 copy.
|
||||
-- next_cycle starts past the current attempt_number so the first future
|
||||
-- resubmission of an address gets a cycle_id that has never been used
|
||||
-- before, even though it's seeded from attempt_number here.
|
||||
INSERT INTO ip_registry (ip_address, first_seen_at, last_seen_at, next_cycle, created_at, updated_at)
|
||||
SELECT ip_address, created_at, updated_at, attempt_number + 1, created_at, updated_at FROM ip_queue;
|
||||
|
||||
ALTER TABLE ip_queue ADD COLUMN registry_id INTEGER REFERENCES ip_registry(id);
|
||||
ALTER TABLE ip_queue ADD COLUMN cycle_id INTEGER NOT NULL DEFAULT 1;
|
||||
UPDATE ip_queue SET
|
||||
registry_id = (SELECT id FROM ip_registry r WHERE r.ip_address = ip_queue.ip_address),
|
||||
cycle_id = attempt_number;
|
||||
|
||||
-- checks.ip_id is today NOT NULL + REFERENCES ip_queue(id), which is exactly
|
||||
-- what forces the cascading DELETE on ip_queue row removal (foreign_keys=ON
|
||||
-- would otherwise block the delete once ip_queue's row disappears out from
|
||||
-- under a referencing row). To let history outlive its ip_queue row, ip_id
|
||||
-- must become nullable and checks must carry their own durable registry_id.
|
||||
-- SQLite has no ALTER to relax a column's NOT NULL/REFERENCES, so the table
|
||||
-- is rebuilt.
|
||||
CREATE TABLE checks_new (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
registry_id INTEGER NOT NULL REFERENCES ip_registry(id),
|
||||
cycle_id INTEGER NOT NULL,
|
||||
ip_id INTEGER REFERENCES ip_queue(id),
|
||||
ip_address TEXT NOT NULL,
|
||||
attempt_number INTEGER NOT NULL,
|
||||
validator_id TEXT NOT NULL DEFAULT '',
|
||||
source TEXT NOT NULL,
|
||||
check_type TEXT NOT NULL,
|
||||
target TEXT NOT NULL DEFAULT '',
|
||||
success BOOLEAN NOT NULL,
|
||||
latency_ms INTEGER NOT NULL DEFAULT 0,
|
||||
detail TEXT NOT NULL DEFAULT '',
|
||||
checked_at TIMESTAMP NOT NULL,
|
||||
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
UNIQUE(registry_id, cycle_id, source, check_type, target)
|
||||
);
|
||||
INSERT INTO checks_new (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)
|
||||
SELECT c.id, iq.registry_id, iq.cycle_id, c.ip_id, c.ip_address, c.attempt_number, c.validator_id,
|
||||
c.source, c.check_type, c.target, c.success, c.latency_ms, c.detail, c.checked_at, c.created_at
|
||||
FROM checks c JOIN ip_queue iq ON iq.id = c.ip_id;
|
||||
DROP TABLE checks;
|
||||
ALTER TABLE checks_new RENAME TO checks;
|
||||
CREATE INDEX idx_checks_registry_cycle ON checks(registry_id, cycle_id);
|
||||
CREATE INDEX idx_checks_ip_attempt ON checks(ip_id, attempt_number);
|
||||
|
||||
-- events.ip_id is already nullable, so no rebuild is needed there — just
|
||||
-- add the durable registry_id (plus cycle_id, so retention pruning can cut
|
||||
-- events at the same cycle boundary as checks) alongside it.
|
||||
ALTER TABLE events ADD COLUMN registry_id INTEGER REFERENCES ip_registry(id);
|
||||
ALTER TABLE events ADD COLUMN cycle_id INTEGER NOT NULL DEFAULT 0;
|
||||
UPDATE events SET
|
||||
registry_id = (SELECT registry_id FROM ip_queue WHERE ip_queue.id = events.ip_id),
|
||||
cycle_id = (SELECT cycle_id FROM ip_queue WHERE ip_queue.id = events.ip_id)
|
||||
WHERE ip_id IS NOT NULL;
|
||||
CREATE INDEX idx_events_registry ON events(registry_id, cycle_id);
|
||||
|
||||
-- Configurable history retention depth, in check cycles per address. 0 (the
|
||||
-- default, matching today's unbounded behavior) means keep everything.
|
||||
ALTER TABLE settings ADD COLUMN history_retention_cycles INTEGER NOT NULL DEFAULT 0;
|
||||
+37
-3
@@ -95,12 +95,21 @@ type IPQueueItem struct {
|
||||
FIPAssociatedAt *time.Time
|
||||
AggregatedAt *time.Time
|
||||
FIPReleasedAt *time.Time
|
||||
RegistryID int64
|
||||
CycleID int
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
type Check struct {
|
||||
ID int64
|
||||
ID int64
|
||||
RegistryID int64
|
||||
CycleID int
|
||||
// IPID is the ip_queue row this check was originally recorded against.
|
||||
// It's cleared to 0 (SQL NULL) if that row was later deleted — history
|
||||
// stays reachable via RegistryID/CycleID regardless (see
|
||||
// migrations/0007_ip_registry.sql). 0 is never a valid ip_queue id
|
||||
// (AUTOINCREMENT starts at 1), so it unambiguously means "orphaned."
|
||||
IPID int64
|
||||
IPAddress string
|
||||
AttemptNumber int
|
||||
@@ -115,11 +124,33 @@ type Check struct {
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// RegistryItem is a durable per-address record that survives an address
|
||||
// being removed from ip_queue and later re-added — see
|
||||
// migrations/0007_ip_registry.sql. NextCycle is the cycle_id that will be
|
||||
// assigned the next time this address is (re)submitted; it only ever
|
||||
// increases, so cycle_id stays unique for this address even across
|
||||
// ip_queue row deletion/recreation.
|
||||
type RegistryItem struct {
|
||||
ID int64
|
||||
IPAddress string
|
||||
FirstSeenAt time.Time
|
||||
LastSeenAt time.Time
|
||||
NextCycle int
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
type Event struct {
|
||||
ID int64
|
||||
SourceType string
|
||||
SourceID string
|
||||
IPID *int64
|
||||
// RegistryID/CycleID are resolved from IPID at insert time (see
|
||||
// InsertEvent) and stay set even after the ip_queue row IPID pointed to
|
||||
// is later deleted, so retention pruning can cut events at the same
|
||||
// cycle boundary as checks — see migrations/0007_ip_registry.sql.
|
||||
RegistryID int64
|
||||
CycleID int
|
||||
EventType string
|
||||
Payload string
|
||||
OccurredAt time.Time
|
||||
@@ -179,8 +210,11 @@ type DeleteIPsResult struct {
|
||||
// admin-configurable at runtime (see queries_settings.go).
|
||||
type Settings struct {
|
||||
FIPSettleSeconds int
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
// HistoryRetentionCycles caps how many recent check cycles are kept per
|
||||
// registry address (see PruneRegistryHistory); 0 means unlimited.
|
||||
HistoryRetentionCycles int
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
// InboundChecksSettings is the singleton row describing what the prober
|
||||
|
||||
@@ -2,50 +2,88 @@ package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
)
|
||||
|
||||
// UpsertCheck records (or, on retry, overwrites) a single check result. The
|
||||
// UNIQUE(ip_id, attempt_number, source, check_type, target) constraint plus
|
||||
// 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.
|
||||
// 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 (ip_id, ip_address, attempt_number, validator_id, source, check_type, target,
|
||||
success, latency_ms, detail, checked_at, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(ip_id, attempt_number, source, check_type, target) DO UPDATE SET
|
||||
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.IPAddress, c.AttemptNumber, c.ValidatorID, c.Source, c.CheckType, c.Target,
|
||||
`, 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, `
|
||||
SELECT id, ip_id, ip_address, attempt_number, validator_id, source, check_type, target,
|
||||
success, latency_ms, detail, checked_at, created_at
|
||||
FROM checks WHERE ip_id=? AND attempt_number=?
|
||||
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.IPID, &c.IPAddress, &c.AttemptNumber, &c.ValidatorID, &c.Source,
|
||||
&c.CheckType, &c.Target, &c.Success, &c.LatencyMS, &c.Detail, &checkedAt, &createdAt); err != nil {
|
||||
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
|
||||
}
|
||||
|
||||
@@ -1,12 +1,19 @@
|
||||
package db
|
||||
|
||||
import "context"
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
)
|
||||
|
||||
// InsertEvent records an audit-trail row. When e.IPID is set, registry_id
|
||||
// and cycle_id are resolved from that ip_queue row's current values, so the
|
||||
// event stays traceable to its address (via registry_id) and prunable at
|
||||
// the right cycle boundary even after the ip_queue row itself is deleted.
|
||||
func (d *DB) InsertEvent(ctx context.Context, e Event) error {
|
||||
_, err := d.ExecContext(ctx, `
|
||||
INSERT INTO events (source_type, source_id, ip_id, event_type, payload, occurred_at, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
`, e.SourceType, e.SourceID, e.IPID, e.EventType, e.Payload, timeToDB(e.OccurredAt), timeToDB(Now()))
|
||||
INSERT INTO events (source_type, source_id, ip_id, registry_id, cycle_id, event_type, payload, occurred_at, created_at)
|
||||
VALUES (?, ?, ?, (SELECT registry_id FROM ip_queue WHERE id=?), COALESCE((SELECT cycle_id FROM ip_queue WHERE id=?), 0), ?, ?, ?, ?)
|
||||
`, e.SourceType, e.SourceID, e.IPID, e.IPID, e.IPID, e.EventType, e.Payload, timeToDB(e.OccurredAt), timeToDB(Now()))
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -14,7 +21,7 @@ func (d *DB) InsertEvent(ctx context.Context, e Event) error {
|
||||
// first — used by the admin detail endpoint.
|
||||
func (d *DB) ListEventsForIP(ctx context.Context, ipID int64) ([]Event, error) {
|
||||
rows, err := d.QueryContext(ctx, `
|
||||
SELECT id, source_type, source_id, ip_id, event_type, payload, occurred_at, created_at
|
||||
SELECT id, source_type, source_id, ip_id, registry_id, cycle_id, event_type, payload, occurred_at, created_at
|
||||
FROM events WHERE ip_id=? ORDER BY occurred_at DESC
|
||||
`, ipID)
|
||||
if err != nil {
|
||||
@@ -26,7 +33,7 @@ func (d *DB) ListEventsForIP(ctx context.Context, ipID int64) ([]Event, error) {
|
||||
|
||||
func (d *DB) ListRecentEvents(ctx context.Context, limit int) ([]Event, error) {
|
||||
rows, err := d.QueryContext(ctx, `
|
||||
SELECT id, source_type, source_id, ip_id, event_type, payload, occurred_at, created_at
|
||||
SELECT id, source_type, source_id, ip_id, registry_id, cycle_id, event_type, payload, occurred_at, created_at
|
||||
FROM events ORDER BY id DESC LIMIT ?
|
||||
`, limit)
|
||||
if err != nil {
|
||||
@@ -45,11 +52,16 @@ func scanEvents(rows interface {
|
||||
for rows.Next() {
|
||||
var e Event
|
||||
var ipID *int64
|
||||
var registryID sql.NullInt64
|
||||
var occurredAt, createdAt string
|
||||
if err := rows.Scan(&e.ID, &e.SourceType, &e.SourceID, &ipID, &e.EventType, &e.Payload, &occurredAt, &createdAt); err != nil {
|
||||
if err := rows.Scan(&e.ID, &e.SourceType, &e.SourceID, &ipID, ®istryID, &e.CycleID,
|
||||
&e.EventType, &e.Payload, &occurredAt, &createdAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
e.IPID = ipID
|
||||
if registryID.Valid {
|
||||
e.RegistryID = registryID.Int64
|
||||
}
|
||||
var err error
|
||||
if e.OccurredAt, err = dbToTime(occurredAt); err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -21,12 +21,26 @@ func (d *DB) SeedQueue(ctx context.Context, addresses []string) error {
|
||||
|
||||
now := timeToDB(Now())
|
||||
for i, addr := range addresses {
|
||||
_, err := tx.ExecContext(ctx, `
|
||||
INSERT INTO ip_queue (ip_address, sequence, state, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT(ip_address) DO NOTHING
|
||||
`, addr, i, IPQueued, now, now)
|
||||
var exists bool
|
||||
if err := tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM ip_queue WHERE ip_address=?)`, addr).Scan(&exists); err != nil {
|
||||
return fmt.Errorf("check existing %s: %w", addr, err)
|
||||
}
|
||||
if exists {
|
||||
continue
|
||||
}
|
||||
registryID, err := findOrCreateRegistryTx(ctx, tx, addr, now)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cycle, err := nextRegistryCycleTx(ctx, tx, registryID, now)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
INSERT INTO ip_queue (ip_address, sequence, state, registry_id, cycle_id, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(ip_address) DO NOTHING
|
||||
`, addr, i, IPQueued, registryID, cycle, now, now); err != nil {
|
||||
return fmt.Errorf("seed %s: %w", addr, err)
|
||||
}
|
||||
}
|
||||
@@ -227,13 +241,21 @@ func (d *DB) RequeueOrFail(ctx context.Context, ipID int64, validatorID string,
|
||||
}
|
||||
|
||||
if nextState == IPQueued {
|
||||
var registryID int64
|
||||
if err := tx.QueryRowContext(ctx, `SELECT registry_id FROM ip_queue WHERE id=?`, ipID).Scan(®istryID); err != nil {
|
||||
return err
|
||||
}
|
||||
cycle, cErr := nextRegistryCycleTx(ctx, tx, registryID, now)
|
||||
if cErr != nil {
|
||||
return cErr
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
UPDATE ip_queue SET
|
||||
state=?, owner_validator_id=NULL, fip_id='', retry_count=?, attempt_number=attempt_number+1,
|
||||
lease_expires_at=NULL, egress_complete=0,
|
||||
cycle_id=?, lease_expires_at=NULL, egress_complete=0,
|
||||
overall_result='', assigned_at=NULL, fip_associated_at=NULL, updated_at=?
|
||||
WHERE id=?
|
||||
`, nextState, retryCount, now, ipID)
|
||||
`, nextState, retryCount, cycle, now, ipID)
|
||||
} else {
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
UPDATE ip_queue SET
|
||||
@@ -300,10 +322,18 @@ func (d *DB) SubmitIPs(ctx context.Context, addresses []string) (SubmitIPsResult
|
||||
err := tx.QueryRowContext(ctx, `SELECT state FROM ip_queue WHERE ip_address=?`, addr).Scan(&state)
|
||||
switch {
|
||||
case err == sql.ErrNoRows:
|
||||
registryID, rErr := findOrCreateRegistryTx(ctx, tx, addr, now)
|
||||
if rErr != nil {
|
||||
return result, rErr
|
||||
}
|
||||
cycle, cErr := nextRegistryCycleTx(ctx, tx, registryID, now)
|
||||
if cErr != nil {
|
||||
return result, cErr
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
INSERT INTO ip_queue (ip_address, sequence, state, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
`, addr, seq, IPQueued, now, now); err != nil {
|
||||
INSERT INTO ip_queue (ip_address, sequence, state, registry_id, cycle_id, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
`, addr, seq, IPQueued, registryID, cycle, now, now); err != nil {
|
||||
return result, fmt.Errorf("insert %s: %w", addr, err)
|
||||
}
|
||||
result.Added = append(result.Added, addr)
|
||||
@@ -312,14 +342,22 @@ func (d *DB) SubmitIPs(ctx context.Context, addresses []string) (SubmitIPsResult
|
||||
return result, err
|
||||
|
||||
case state == IPDone || state == IPFailed || state == IPOccupied:
|
||||
var registryID int64
|
||||
if err := tx.QueryRowContext(ctx, `SELECT registry_id FROM ip_queue WHERE ip_address=?`, addr).Scan(®istryID); err != nil {
|
||||
return result, err
|
||||
}
|
||||
cycle, cErr := nextRegistryCycleTx(ctx, tx, registryID, now)
|
||||
if cErr != nil {
|
||||
return result, cErr
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
UPDATE ip_queue SET
|
||||
state=?, sequence=?, owner_validator_id=NULL, fip_id='', retry_count=0,
|
||||
attempt_number=attempt_number+1, lease_expires_at=NULL, egress_complete=0,
|
||||
attempt_number=attempt_number+1, cycle_id=?, lease_expires_at=NULL, egress_complete=0,
|
||||
overall_result='',
|
||||
assigned_at=NULL, fip_associated_at=NULL, aggregated_at=NULL, fip_released_at=NULL, updated_at=?
|
||||
WHERE ip_address=?
|
||||
`, IPQueued, seq, now, addr); err != nil {
|
||||
`, IPQueued, seq, cycle, now, addr); err != nil {
|
||||
return result, fmt.Errorf("requeue %s: %w", addr, err)
|
||||
}
|
||||
result.Requeued = append(result.Requeued, addr)
|
||||
@@ -427,8 +465,11 @@ func (d *DB) DeleteIPs(ctx context.Context, addresses []string) (DeleteIPsResult
|
||||
}
|
||||
|
||||
// deleteIPTx is the shared body of DeleteIP/DeleteIPs: free the owning
|
||||
// validator, delete dependent checks/events, then the ip_queue row itself
|
||||
// — the same FK-clearing order DeleteValidator uses for owner_validator_id.
|
||||
// validator, detach dependent checks/events from the doomed ip_queue row
|
||||
// (their registry_id already anchors them permanently — see
|
||||
// migrations/0007_ip_registry.sql — so this is a hand-off, not a loss),
|
||||
// delete the ephemeral per-attempt ip_site_checks progress flags, then the
|
||||
// ip_queue row itself.
|
||||
func deleteIPTx(ctx context.Context, tx *sql.Tx, ipID int64) error {
|
||||
now := timeToDB(Now())
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
@@ -437,11 +478,11 @@ func deleteIPTx(ctx context.Context, tx *sql.Tx, ipID int64) error {
|
||||
`, ValidatorIdle, now, ipID); err != nil {
|
||||
return fmt.Errorf("free owning validator: %w", err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `DELETE FROM checks WHERE ip_id=?`, ipID); err != nil {
|
||||
return fmt.Errorf("delete checks: %w", err)
|
||||
if _, err := tx.ExecContext(ctx, `UPDATE checks SET ip_id=NULL WHERE ip_id=?`, ipID); err != nil {
|
||||
return fmt.Errorf("detach checks: %w", err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `DELETE FROM events WHERE ip_id=?`, ipID); err != nil {
|
||||
return fmt.Errorf("delete events: %w", err)
|
||||
if _, err := tx.ExecContext(ctx, `UPDATE events SET ip_id=NULL WHERE ip_id=?`, ipID); err != nil {
|
||||
return fmt.Errorf("detach events: %w", err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `DELETE FROM ip_site_checks WHERE ip_id=?`, ipID); err != nil {
|
||||
return fmt.Errorf("delete ip_site_checks: %w", err)
|
||||
@@ -507,7 +548,8 @@ func (d *DB) ListExpiredLeases(ctx context.Context, now time.Time) ([]IPQueueIte
|
||||
const ipQueueSelect = `
|
||||
SELECT id, ip_address, sequence, state, owner_validator_id, fip_id, attempt_number, retry_count,
|
||||
lease_expires_at, egress_complete, overall_result,
|
||||
assigned_at, fip_associated_at, aggregated_at, fip_released_at, created_at, updated_at
|
||||
assigned_at, fip_associated_at, aggregated_at, fip_released_at,
|
||||
registry_id, cycle_id, created_at, updated_at
|
||||
FROM ip_queue
|
||||
`
|
||||
|
||||
@@ -527,18 +569,23 @@ func scanIPQueueItem(row rowScanner) (*IPQueueItem, error) {
|
||||
var item IPQueueItem
|
||||
var owner sql.NullString
|
||||
var leaseExpires, assignedAt, fipAssociatedAt, aggregatedAt, fipReleasedAt sql.NullString
|
||||
var registryID sql.NullInt64
|
||||
var createdAt, updatedAt string
|
||||
if err := row.Scan(
|
||||
&item.ID, &item.IPAddress, &item.Sequence, &item.State, &owner, &item.FIPID,
|
||||
&item.AttemptNumber, &item.RetryCount, &leaseExpires,
|
||||
&item.EgressComplete,
|
||||
&item.OverallResult, &assignedAt, &fipAssociatedAt, &aggregatedAt, &fipReleasedAt, &createdAt, &updatedAt,
|
||||
&item.OverallResult, &assignedAt, &fipAssociatedAt, &aggregatedAt, &fipReleasedAt,
|
||||
®istryID, &item.CycleID, &createdAt, &updatedAt,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if owner.Valid {
|
||||
item.OwnerValidatorID = &owner.String
|
||||
}
|
||||
if registryID.Valid {
|
||||
item.RegistryID = registryID.Int64
|
||||
}
|
||||
var err error
|
||||
if item.LeaseExpiresAt, err = nullStringToTimePtr(leaseExpires); err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
package db
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// findOrCreateRegistryTx returns the ip_registry row id for addr, creating
|
||||
// it (next_cycle starting at 1) the first time this address is ever seen.
|
||||
// Safe to call on every (re)submission of an address — ip_address is
|
||||
// UNIQUE, so a second call for the same address is a no-op lookup.
|
||||
func findOrCreateRegistryTx(ctx context.Context, tx *sql.Tx, addr string, now string) (int64, error) {
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
INSERT INTO ip_registry (ip_address, first_seen_at, last_seen_at, next_cycle, created_at, updated_at)
|
||||
VALUES (?, ?, ?, 1, ?, ?)
|
||||
ON CONFLICT(ip_address) DO NOTHING
|
||||
`, addr, now, now, now, now); err != nil {
|
||||
return 0, fmt.Errorf("find or create registry for %s: %w", addr, err)
|
||||
}
|
||||
var id int64
|
||||
if err := tx.QueryRowContext(ctx, `SELECT id FROM ip_registry WHERE ip_address=?`, addr).Scan(&id); err != nil {
|
||||
return 0, fmt.Errorf("lookup registry id for %s: %w", addr, err)
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// nextRegistryCycleTx allocates the next cycle_id for registryID and
|
||||
// advances last_seen_at. Call once per fresh generation of check history for
|
||||
// an address — a brand new ip_queue row, an admin-triggered requeue, or a
|
||||
// system retry that bumps attempt_number — so every generation's checks get
|
||||
// a cycle_id no other generation, past or future (even across ip_queue row
|
||||
// deletion and recreation), will ever reuse.
|
||||
func nextRegistryCycleTx(ctx context.Context, tx *sql.Tx, registryID int64, now string) (int, error) {
|
||||
var cycle int
|
||||
if err := tx.QueryRowContext(ctx, `SELECT next_cycle FROM ip_registry WHERE id=?`, registryID).Scan(&cycle); err != nil {
|
||||
return 0, fmt.Errorf("read next_cycle for registry %d: %w", registryID, err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
UPDATE ip_registry SET next_cycle=next_cycle+1, last_seen_at=?, updated_at=? WHERE id=?
|
||||
`, now, now, registryID); err != nil {
|
||||
return 0, fmt.Errorf("advance next_cycle for registry %d: %w", registryID, err)
|
||||
}
|
||||
return cycle, nil
|
||||
}
|
||||
|
||||
// RegistrySummary is one row of the full IP registry, combining the durable
|
||||
// ip_registry record with a rollup of its check history and, if the address
|
||||
// currently has a live ip_queue row, that row's state.
|
||||
type RegistrySummary struct {
|
||||
RegistryItem
|
||||
TotalCycles int
|
||||
LastResult string
|
||||
LastCheckedAt *time.Time
|
||||
InQueue bool
|
||||
CurrentState string
|
||||
}
|
||||
|
||||
// ListRegistry returns every address ever submitted, newest first-seen
|
||||
// last, each with a summary of its accumulated check history. Reads two
|
||||
// simple queries plus one aggregate rather than a single large join, since
|
||||
// the registry is expected to stay small enough (one row per distinct
|
||||
// address ever seen) that this is simpler to reason about than a
|
||||
// multi-way correlated subquery.
|
||||
func (d *DB) ListRegistry(ctx context.Context) ([]RegistrySummary, error) {
|
||||
rows, err := d.QueryContext(ctx, `
|
||||
SELECT id, ip_address, first_seen_at, last_seen_at, next_cycle, created_at, updated_at
|
||||
FROM ip_registry ORDER BY first_seen_at
|
||||
`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
items, err := scanRegistryItems(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
out := make([]RegistrySummary, len(items))
|
||||
for i, item := range items {
|
||||
s := RegistrySummary{RegistryItem: item}
|
||||
if err := d.fillRegistrySummary(ctx, &s); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[i] = s
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// GetRegistryByAddress returns the registry row (with summary) for a single
|
||||
// address, or ErrNotFound if it has never been submitted.
|
||||
func (d *DB) GetRegistryByAddress(ctx context.Context, address string) (*RegistrySummary, error) {
|
||||
row := d.QueryRowContext(ctx, `
|
||||
SELECT id, ip_address, first_seen_at, last_seen_at, next_cycle, created_at, updated_at
|
||||
FROM ip_registry WHERE ip_address=?
|
||||
`, address)
|
||||
item, err := scanRegistryItem(row)
|
||||
if err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, fmt.Errorf("ip %q: %w", address, ErrNotFound)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
s := &RegistrySummary{RegistryItem: *item}
|
||||
if err := d.fillRegistrySummary(ctx, s); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (d *DB) fillRegistrySummary(ctx context.Context, s *RegistrySummary) error {
|
||||
if err := d.QueryRowContext(ctx, `
|
||||
SELECT COUNT(DISTINCT cycle_id) FROM checks WHERE registry_id=?
|
||||
`, s.ID).Scan(&s.TotalCycles); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var lastResult sql.NullBool
|
||||
var lastCheckedAt sql.NullString
|
||||
if err := d.QueryRowContext(ctx, `
|
||||
SELECT success, checked_at FROM checks WHERE registry_id=? ORDER BY cycle_id DESC, checked_at DESC LIMIT 1
|
||||
`, s.ID).Scan(&lastResult, &lastCheckedAt); err != nil && err != sql.ErrNoRows {
|
||||
return err
|
||||
}
|
||||
if lastResult.Valid {
|
||||
if lastResult.Bool {
|
||||
s.LastResult = "pass"
|
||||
} else {
|
||||
s.LastResult = "fail"
|
||||
}
|
||||
}
|
||||
t, err := nullStringToTimePtr(lastCheckedAt)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.LastCheckedAt = t
|
||||
|
||||
var state sql.NullString
|
||||
if err := d.QueryRowContext(ctx, `SELECT state FROM ip_queue WHERE registry_id=?`, s.ID).Scan(&state); err != nil && err != sql.ErrNoRows {
|
||||
return err
|
||||
}
|
||||
if state.Valid {
|
||||
s.InQueue = true
|
||||
s.CurrentState = state.String
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func scanRegistryItems(rows *sql.Rows) ([]RegistryItem, error) {
|
||||
var out []RegistryItem
|
||||
for rows.Next() {
|
||||
item, err := scanRegistryItem(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, *item)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func scanRegistryItem(row rowScanner) (*RegistryItem, error) {
|
||||
var item RegistryItem
|
||||
var firstSeenAt, lastSeenAt, createdAt, updatedAt string
|
||||
if err := row.Scan(&item.ID, &item.IPAddress, &firstSeenAt, &lastSeenAt, &item.NextCycle, &createdAt, &updatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var err error
|
||||
if item.FirstSeenAt, err = dbToTime(firstSeenAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if item.LastSeenAt, err = dbToTime(lastSeenAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if item.CreatedAt, err = dbToTime(createdAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if item.UpdatedAt, err = dbToTime(updatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &item, nil
|
||||
}
|
||||
|
||||
// PruneRegistryHistory deletes all but the newest keepCycles cycles' worth
|
||||
// of check history for registryID — a no-op if keepCycles <= 0 (unlimited
|
||||
// retention) or the address has that many cycles or fewer. events rows tied
|
||||
// to this registry are pruned to the same cutoff; ip_registry itself and its
|
||||
// next_cycle counter are never touched, so pruning never risks a future
|
||||
// cycle_id collision.
|
||||
func (d *DB) PruneRegistryHistory(ctx context.Context, registryID int64, keepCycles int) error {
|
||||
if keepCycles <= 0 {
|
||||
return nil
|
||||
}
|
||||
var cutoff sql.NullInt64
|
||||
err := d.QueryRowContext(ctx, `
|
||||
SELECT MIN(cycle_id) FROM (
|
||||
SELECT DISTINCT cycle_id FROM checks WHERE registry_id=? ORDER BY cycle_id DESC LIMIT ?
|
||||
)
|
||||
`, registryID, keepCycles).Scan(&cutoff)
|
||||
if err != nil {
|
||||
return fmt.Errorf("find prune cutoff for registry %d: %w", registryID, err)
|
||||
}
|
||||
if !cutoff.Valid {
|
||||
return nil // fewer than keepCycles cycles recorded — nothing to prune
|
||||
}
|
||||
if _, err := d.ExecContext(ctx, `DELETE FROM checks WHERE registry_id=? AND cycle_id<?`, registryID, cutoff.Int64); err != nil {
|
||||
return fmt.Errorf("prune checks for registry %d: %w", registryID, err)
|
||||
}
|
||||
// Cut events at the same cycle boundary as checks, so an address's audit
|
||||
// trail and its check history are retained (and pruned) in lockstep.
|
||||
if _, err := d.ExecContext(ctx, `DELETE FROM events WHERE registry_id=? AND cycle_id<?`, registryID, cutoff.Int64); err != nil {
|
||||
return fmt.Errorf("prune events for registry %d: %w", registryID, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,242 @@
|
||||
package db
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"cloudipvalidator/internal/config"
|
||||
)
|
||||
|
||||
// TestRegistryHistorySurvivesIPDeletion proves that deleting an address
|
||||
// from ip_queue no longer destroys its check history — the whole point of
|
||||
// ip_registry (see migrations/0007_ip_registry.sql).
|
||||
func TestRegistryHistorySurvivesIPDeletion(t *testing.T) {
|
||||
d, ctx := newTestDB(t)
|
||||
|
||||
if _, err := d.SubmitIPs(ctx, []string{"1.2.3.4"}); err != nil {
|
||||
t.Fatalf("submit ips: %v", err)
|
||||
}
|
||||
ip, err := d.GetIPByAddress(ctx, "1.2.3.4")
|
||||
if err != nil {
|
||||
t.Fatalf("get ip: %v", err)
|
||||
}
|
||||
if err := d.UpsertCheck(ctx, Check{
|
||||
IPID: ip.ID, IPAddress: ip.IPAddress, AttemptNumber: ip.AttemptNumber,
|
||||
Source: SourceEgress, CheckType: "https", Target: "https://example.test",
|
||||
Success: true, CheckedAt: Now(),
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert check: %v", err)
|
||||
}
|
||||
|
||||
if err := d.DeleteIP(ctx, ip.ID); err != nil {
|
||||
t.Fatalf("delete ip: %v", err)
|
||||
}
|
||||
|
||||
// The live queue no longer knows about the address...
|
||||
if _, err := d.GetIPByAddress(ctx, "1.2.3.4"); err == nil {
|
||||
t.Fatalf("expected ip_queue row gone after delete")
|
||||
}
|
||||
|
||||
// ...but the registry and its check history are untouched.
|
||||
summary, err := d.GetRegistryByAddress(ctx, "1.2.3.4")
|
||||
if err != nil {
|
||||
t.Fatalf("get registry: %v", err)
|
||||
}
|
||||
if summary.TotalCycles != 1 {
|
||||
t.Fatalf("expected 1 cycle retained, got %d", summary.TotalCycles)
|
||||
}
|
||||
checks, err := d.ListChecksForRegistry(ctx, summary.ID, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("list checks for registry: %v", err)
|
||||
}
|
||||
if len(checks) != 1 {
|
||||
t.Fatalf("expected 1 retained check, got %+v", checks)
|
||||
}
|
||||
if checks[0].IPID != 0 {
|
||||
t.Fatalf("expected orphaned check's ip_id cleared to 0, got %d", checks[0].IPID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRegistryCycleSurvivesReaddWithoutCollision proves that deleting an
|
||||
// address and resubmitting it later gives the new generation of checks a
|
||||
// cycle_id distinct from the old one, so both generations' history is kept
|
||||
// as separate rows rather than colliding on the UNIQUE constraint.
|
||||
func TestRegistryCycleSurvivesReaddWithoutCollision(t *testing.T) {
|
||||
d, ctx := newTestDB(t)
|
||||
|
||||
if _, err := d.SubmitIPs(ctx, []string{"1.2.3.4"}); err != nil {
|
||||
t.Fatalf("submit ips (1st time): %v", err)
|
||||
}
|
||||
ip1, err := d.GetIPByAddress(ctx, "1.2.3.4")
|
||||
if err != nil {
|
||||
t.Fatalf("get ip: %v", err)
|
||||
}
|
||||
if err := d.UpsertCheck(ctx, Check{
|
||||
IPID: ip1.ID, IPAddress: ip1.IPAddress, AttemptNumber: ip1.AttemptNumber,
|
||||
Source: SourceEgress, CheckType: "https", Target: "https://example.test",
|
||||
Success: true, CheckedAt: Now(),
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert check (1st time): %v", err)
|
||||
}
|
||||
if err := d.DeleteIP(ctx, ip1.ID); err != nil {
|
||||
t.Fatalf("delete ip: %v", err)
|
||||
}
|
||||
|
||||
if _, err := d.SubmitIPs(ctx, []string{"1.2.3.4"}); err != nil {
|
||||
t.Fatalf("submit ips (2nd time): %v", err)
|
||||
}
|
||||
ip2, err := d.GetIPByAddress(ctx, "1.2.3.4")
|
||||
if err != nil {
|
||||
t.Fatalf("get ip (2nd time): %v", err)
|
||||
}
|
||||
if ip2.ID == ip1.ID {
|
||||
t.Fatalf("expected a fresh ip_queue row, got the same id %d", ip1.ID)
|
||||
}
|
||||
if ip2.CycleID == ip1.CycleID {
|
||||
t.Fatalf("expected a fresh cycle_id, got the same value %d twice", ip1.CycleID)
|
||||
}
|
||||
// Same check_type/target/source as the first cycle — would collide on
|
||||
// the old UNIQUE(ip_id, attempt_number, ...) key structure.
|
||||
if err := d.UpsertCheck(ctx, Check{
|
||||
IPID: ip2.ID, IPAddress: ip2.IPAddress, AttemptNumber: ip2.AttemptNumber,
|
||||
Source: SourceEgress, CheckType: "https", Target: "https://example.test",
|
||||
Success: false, CheckedAt: Now(),
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert check (2nd time): %v", err)
|
||||
}
|
||||
|
||||
summary, err := d.GetRegistryByAddress(ctx, "1.2.3.4")
|
||||
if err != nil {
|
||||
t.Fatalf("get registry: %v", err)
|
||||
}
|
||||
if summary.TotalCycles != 2 {
|
||||
t.Fatalf("expected 2 distinct cycles retained, got %d", summary.TotalCycles)
|
||||
}
|
||||
checks, err := d.ListChecksForRegistry(ctx, summary.ID, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("list checks for registry: %v", err)
|
||||
}
|
||||
if len(checks) != 2 {
|
||||
t.Fatalf("expected 2 separate check rows (one per cycle), got %+v", checks)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPruneRegistryHistoryKeepsOnlyNewestCycles proves the retention-depth
|
||||
// setting actually deletes older cycles' checks once an address has
|
||||
// accumulated more cycles than the configured depth.
|
||||
func TestPruneRegistryHistoryKeepsOnlyNewestCycles(t *testing.T) {
|
||||
d, ctx := newTestDB(t)
|
||||
|
||||
var registryID int64
|
||||
for i := 0; i < 3; i++ {
|
||||
if _, err := d.SubmitIPs(ctx, []string{"1.2.3.4"}); err != nil {
|
||||
t.Fatalf("submit ips (cycle %d): %v", i, err)
|
||||
}
|
||||
ip, err := d.GetIPByAddress(ctx, "1.2.3.4")
|
||||
if err != nil {
|
||||
t.Fatalf("get ip (cycle %d): %v", i, err)
|
||||
}
|
||||
registryID = ip.RegistryID
|
||||
if err := d.UpsertCheck(ctx, Check{
|
||||
IPID: ip.ID, IPAddress: ip.IPAddress, AttemptNumber: ip.AttemptNumber,
|
||||
Source: SourceEgress, CheckType: "https", Target: "https://example.test",
|
||||
Success: true, CheckedAt: Now(),
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert check (cycle %d): %v", i, err)
|
||||
}
|
||||
if i < 2 {
|
||||
if err := d.DeleteIP(ctx, ip.ID); err != nil {
|
||||
t.Fatalf("delete ip (cycle %d): %v", i, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
summary, err := d.GetRegistryByAddress(ctx, "1.2.3.4")
|
||||
if err != nil {
|
||||
t.Fatalf("get registry: %v", err)
|
||||
}
|
||||
if summary.TotalCycles != 3 {
|
||||
t.Fatalf("expected 3 cycles before pruning, got %d", summary.TotalCycles)
|
||||
}
|
||||
|
||||
if err := d.PruneRegistryHistory(ctx, registryID, 1); err != nil {
|
||||
t.Fatalf("prune registry history: %v", err)
|
||||
}
|
||||
|
||||
summary, err = d.GetRegistryByAddress(ctx, "1.2.3.4")
|
||||
if err != nil {
|
||||
t.Fatalf("get registry after prune: %v", err)
|
||||
}
|
||||
if summary.TotalCycles != 1 {
|
||||
t.Fatalf("expected 1 cycle after pruning to depth 1, got %d", summary.TotalCycles)
|
||||
}
|
||||
|
||||
// Pruning must never touch next_cycle — otherwise a future cycle_id
|
||||
// could collide with one that was just pruned away.
|
||||
var nextCycle int
|
||||
if err := d.QueryRowContext(ctx, `SELECT next_cycle FROM ip_registry WHERE id=?`, registryID).Scan(&nextCycle); err != nil {
|
||||
t.Fatalf("read next_cycle: %v", err)
|
||||
}
|
||||
if nextCycle != 4 {
|
||||
t.Fatalf("expected next_cycle unaffected by pruning (still 4), got %d", nextCycle)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPruneRegistryHistoryNoopWhenUnderDepth proves pruning is a no-op when
|
||||
// an address has fewer cycles than the configured retention depth.
|
||||
func TestPruneRegistryHistoryNoopWhenUnderDepth(t *testing.T) {
|
||||
d, ctx := newTestDB(t)
|
||||
|
||||
if _, err := d.SubmitIPs(ctx, []string{"1.2.3.4"}); err != nil {
|
||||
t.Fatalf("submit ips: %v", err)
|
||||
}
|
||||
ip, err := d.GetIPByAddress(ctx, "1.2.3.4")
|
||||
if err != nil {
|
||||
t.Fatalf("get ip: %v", err)
|
||||
}
|
||||
if err := d.UpsertCheck(ctx, Check{
|
||||
IPID: ip.ID, IPAddress: ip.IPAddress, AttemptNumber: ip.AttemptNumber,
|
||||
Source: SourceEgress, CheckType: "https", Target: "https://example.test",
|
||||
Success: true, CheckedAt: Now(),
|
||||
}); err != nil {
|
||||
t.Fatalf("upsert check: %v", err)
|
||||
}
|
||||
|
||||
if err := d.PruneRegistryHistory(ctx, ip.RegistryID, 10); err != nil {
|
||||
t.Fatalf("prune registry history: %v", err)
|
||||
}
|
||||
summary, err := d.GetRegistryByAddress(ctx, "1.2.3.4")
|
||||
if err != nil {
|
||||
t.Fatalf("get registry: %v", err)
|
||||
}
|
||||
if summary.TotalCycles != 1 {
|
||||
t.Fatalf("expected the single cycle to survive a no-op prune, got %d", summary.TotalCycles)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetHistoryRetentionCyclesRejectsNegative(t *testing.T) {
|
||||
d, ctx := newTestDB(t)
|
||||
if err := d.BootstrapFromConfig(ctx, &config.ControlAPI{}); err != nil {
|
||||
t.Fatalf("bootstrap: %v", err)
|
||||
}
|
||||
if err := d.SetHistoryRetentionCycles(ctx, -1); !errors.Is(err, ErrValidation) {
|
||||
t.Fatalf("expected ErrValidation, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetSetHistoryRetentionCyclesRoundTrip(t *testing.T) {
|
||||
d, ctx := newTestDB(t)
|
||||
if err := d.BootstrapFromConfig(ctx, &config.ControlAPI{}); err != nil {
|
||||
t.Fatalf("bootstrap: %v", err)
|
||||
}
|
||||
if err := d.SetHistoryRetentionCycles(ctx, 5); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
s, err := d.GetSettings(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("get settings: %v", err)
|
||||
}
|
||||
if s.HistoryRetentionCycles != 5 {
|
||||
t.Fatalf("expected 5, got %d", s.HistoryRetentionCycles)
|
||||
}
|
||||
}
|
||||
@@ -13,8 +13,8 @@ func (d *DB) GetSettings(ctx context.Context) (Settings, error) {
|
||||
var s Settings
|
||||
var createdAt, updatedAt string
|
||||
err := d.QueryRowContext(ctx, `
|
||||
SELECT fip_settle_seconds, created_at, updated_at FROM settings WHERE id=1
|
||||
`).Scan(&s.FIPSettleSeconds, &createdAt, &updatedAt)
|
||||
SELECT fip_settle_seconds, history_retention_cycles, created_at, updated_at FROM settings WHERE id=1
|
||||
`).Scan(&s.FIPSettleSeconds, &s.HistoryRetentionCycles, &createdAt, &updatedAt)
|
||||
if err != nil {
|
||||
return Settings{}, err
|
||||
}
|
||||
@@ -42,3 +42,19 @@ func (d *DB) SetFIPSettleSeconds(ctx context.Context, seconds int) error {
|
||||
`, seconds, now)
|
||||
return err
|
||||
}
|
||||
|
||||
// SetHistoryRetentionCycles persists how many recent check cycles to keep
|
||||
// per registry address; 0 means unlimited (the default, matching behavior
|
||||
// before this setting existed). Does not retroactively prune anything
|
||||
// itself — pruning happens per-address as each check cycle finishes (see
|
||||
// orchestrator.aggregateAndRelease).
|
||||
func (d *DB) SetHistoryRetentionCycles(ctx context.Context, cycles int) error {
|
||||
if cycles < 0 {
|
||||
return fmt.Errorf("history_retention_cycles must be >= 0: %w", ErrValidation)
|
||||
}
|
||||
now := timeToDB(Now())
|
||||
_, err := d.ExecContext(ctx, `
|
||||
UPDATE settings SET history_retention_cycles=?, updated_at=? WHERE id=1
|
||||
`, cycles, now)
|
||||
return err
|
||||
}
|
||||
@@ -32,6 +32,27 @@ type clearQueueResponse struct {
|
||||
Deleted []string `json:"deleted"`
|
||||
}
|
||||
|
||||
type scanIPsResponse struct {
|
||||
ScannedFree int `json:"scanned_free"`
|
||||
Added []string `json:"added"`
|
||||
Requeued []string `json:"requeued"`
|
||||
Reordered []string `json:"reordered"`
|
||||
SkippedInProgress []string `json:"skipped_in_progress"`
|
||||
}
|
||||
|
||||
// registryDTO is one row of the durable per-address registry — see
|
||||
// db.RegistrySummary.
|
||||
type registryDTO struct {
|
||||
IPAddress string `json:"ip_address"`
|
||||
FirstSeenAt time.Time `json:"first_seen_at"`
|
||||
LastSeenAt time.Time `json:"last_seen_at"`
|
||||
TotalCycles int `json:"total_cycles"`
|
||||
LastResult string `json:"last_result"`
|
||||
LastCheckedAt *time.Time `json:"last_checked_at"`
|
||||
InQueue bool `json:"in_queue"`
|
||||
CurrentState string `json:"current_state"`
|
||||
}
|
||||
|
||||
type validatorDTO struct {
|
||||
ValidatorID string `json:"validator_id"`
|
||||
Hostname string `json:"hostname"`
|
||||
@@ -85,7 +106,8 @@ type putCheckTypeRequest struct {
|
||||
// request body for /api/v1/admin/config/orchestrator — a single-field DTO,
|
||||
// same shape both ways, like putSiteRequest/siteDTO.
|
||||
type orchestratorSettingsDTO struct {
|
||||
FIPSettleSeconds int `json:"fip_settle_seconds"`
|
||||
FIPSettleSeconds int `json:"fip_settle_seconds"`
|
||||
HistoryRetentionCycles int `json:"history_retention_cycles"`
|
||||
}
|
||||
|
||||
// inboundChecksDTO doubles as both the GET response and the PUT request
|
||||
|
||||
@@ -99,6 +99,26 @@ func (s *Server) handleAdminSubmitIPs(w http.ResponseWriter, r *http.Request) {
|
||||
})
|
||||
}
|
||||
|
||||
// handleAdminScanFloatingIPs lists every floating IP in the configured
|
||||
// OpenStack project, filters to the free (unassociated) pool, and submits
|
||||
// that address list to the check queue — see orchestrator.ScanFloatingIPs.
|
||||
// Takes no body; POST is used (rather than GET) because it mutates the
|
||||
// queue, matching handleAdminSubmitIPs.
|
||||
func (s *Server) handleAdminScanFloatingIPs(w http.ResponseWriter, r *http.Request) {
|
||||
result, scannedFree, err := s.Orch.ScanFloatingIPs(r.Context())
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadGateway, err.Error())
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, scanIPsResponse{
|
||||
ScannedFree: scannedFree,
|
||||
Added: emptyIfNil(result.Added),
|
||||
Requeued: emptyIfNil(result.Requeued),
|
||||
Reordered: emptyIfNil(result.Reordered),
|
||||
SkippedInProgress: emptyIfNil(result.SkippedInProgress),
|
||||
})
|
||||
}
|
||||
|
||||
// handleAdminCancelIP force-stops a check in progress (or still-queued) for
|
||||
// the given address. Requires the orchestrator, since a floating IP may
|
||||
// need to be disassociated in OpenStack.
|
||||
|
||||
@@ -200,7 +200,10 @@ func (s *Server) handleConfigGetOrchestratorSettings(w http.ResponseWriter, r *h
|
||||
writeDBError(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, orchestratorSettingsDTO{FIPSettleSeconds: settings.FIPSettleSeconds})
|
||||
writeJSON(w, http.StatusOK, orchestratorSettingsDTO{
|
||||
FIPSettleSeconds: settings.FIPSettleSeconds,
|
||||
HistoryRetentionCycles: settings.HistoryRetentionCycles,
|
||||
})
|
||||
}
|
||||
|
||||
// handleConfigPutOrchestratorSettings goes through the orchestrator (not a
|
||||
@@ -217,7 +220,14 @@ func (s *Server) handleConfigPutOrchestratorSettings(w http.ResponseWriter, r *h
|
||||
writeDBError(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, orchestratorSettingsDTO{FIPSettleSeconds: req.FIPSettleSeconds})
|
||||
if err := s.DB.SetHistoryRetentionCycles(r.Context(), req.HistoryRetentionCycles); err != nil {
|
||||
writeDBError(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, orchestratorSettingsDTO{
|
||||
FIPSettleSeconds: req.FIPSettleSeconds,
|
||||
HistoryRetentionCycles: req.HistoryRetentionCycles,
|
||||
})
|
||||
}
|
||||
|
||||
// --- prober inbound checks ---
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"cloudipvalidator/internal/db"
|
||||
)
|
||||
|
||||
// handleAdminRegistry lists every address ever submitted to the check
|
||||
// queue, each with a summary of its accumulated check history — the
|
||||
// durable record that survives an address being deleted from ip_queue and
|
||||
// later re-added. See migrations/0007_ip_registry.sql.
|
||||
func (s *Server) handleAdminRegistry(w http.ResponseWriter, r *http.Request) {
|
||||
items, err := s.DB.ListRegistry(r.Context())
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
out := make([]registryDTO, len(items))
|
||||
for i, it := range items {
|
||||
out[i] = registrySummaryToDTO(it)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
}
|
||||
|
||||
// handleAdminRegistryHistory returns one address's registry record plus its
|
||||
// full retained check history (across every cycle still kept — see
|
||||
// db.PruneRegistryHistory / settings.history_retention_cycles), newest
|
||||
// cycle first.
|
||||
func (s *Server) handleAdminRegistryHistory(w http.ResponseWriter, r *http.Request) {
|
||||
address := r.PathValue("ip")
|
||||
summary, err := s.DB.GetRegistryByAddress(r.Context(), address)
|
||||
if err != nil {
|
||||
writeDBError(w, err)
|
||||
return
|
||||
}
|
||||
checks, err := s.DB.ListChecksForRegistry(r.Context(), summary.ID, nil)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, struct {
|
||||
Registry registryDTO `json:"registry"`
|
||||
Checks []db.Check `json:"checks"`
|
||||
}{registrySummaryToDTO(*summary), checks})
|
||||
}
|
||||
|
||||
func registrySummaryToDTO(s db.RegistrySummary) registryDTO {
|
||||
return registryDTO{
|
||||
IPAddress: s.IPAddress,
|
||||
FirstSeenAt: s.FirstSeenAt,
|
||||
LastSeenAt: s.LastSeenAt,
|
||||
TotalCycles: s.TotalCycles,
|
||||
LastResult: s.LastResult,
|
||||
LastCheckedAt: s.LastCheckedAt,
|
||||
InQueue: s.InQueue,
|
||||
CurrentState: s.CurrentState,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,150 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"cloudipvalidator/internal/db"
|
||||
)
|
||||
|
||||
// TestScanFloatingIPsEndpoint proves POST /api/v1/admin/ips/scan only
|
||||
// queues floating IPs that are currently unassociated in OpenStack.
|
||||
func TestScanFloatingIPsEndpoint(t *testing.T) {
|
||||
fc, _, _, mock := newConfigTestHarness(t)
|
||||
mock.Seed("fip-free", "5.5.5.5", "svc-project")
|
||||
mock.SeedWithPort("fip-occupied", "6.6.6.6", "svc-project", "some-port")
|
||||
|
||||
resp, body := fc.do(http.MethodPost, "/api/v1/admin/ips/scan", nil)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("scan: status=%d body=%s", resp.StatusCode, body)
|
||||
}
|
||||
var scanResp scanIPsResponse
|
||||
if err := json.Unmarshal(body, &scanResp); err != nil {
|
||||
t.Fatalf("unmarshal scan response: %v", err)
|
||||
}
|
||||
if scanResp.ScannedFree != 1 {
|
||||
t.Fatalf("expected 1 free fip scanned, got %+v", scanResp)
|
||||
}
|
||||
if len(scanResp.Added) != 1 || scanResp.Added[0] != "5.5.5.5" {
|
||||
t.Fatalf("expected only 5.5.5.5 added, got %+v", scanResp)
|
||||
}
|
||||
|
||||
_, body = fc.do(http.MethodGet, "/api/v1/admin/ips", nil)
|
||||
var ips []db.IPQueueItem
|
||||
if err := json.Unmarshal(body, &ips); err != nil {
|
||||
t.Fatalf("unmarshal ips: %v", err)
|
||||
}
|
||||
if len(ips) != 1 || ips[0].IPAddress != "5.5.5.5" {
|
||||
t.Fatalf("expected only the free address queued, got %+v", ips)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRegistryHistoryOutlivesIPDeletion proves that after an address
|
||||
// completes a full check cycle and is then deleted from the queue, its
|
||||
// history is still reachable via the registry endpoints (though the plain
|
||||
// /ips/{ip} endpoint now 404s), and that resubmitting the same address adds
|
||||
// a second, distinct cycle to the same registry entry.
|
||||
func TestRegistryHistoryOutlivesIPDeletion(t *testing.T) {
|
||||
fc, d, orch, mock := newConfigTestHarness(t)
|
||||
ctx := context.Background()
|
||||
mock.Seed("fip-1", "9.9.9.9", "svc-project")
|
||||
|
||||
fc.do(http.MethodPost, "/api/v1/admin/config/validators", createValidatorRequest{ValidatorID: "validator-1", OSPortID: "port-1"})
|
||||
fc.do(http.MethodPut, "/api/v1/admin/config/targets/web", putTargetGroupRequest{Targets: []string{"https://example.test"}})
|
||||
fc.do(http.MethodPut, "/api/v1/admin/config/check-types/https", putCheckTypeRequest{Enabled: true, Targets: []string{"web"}})
|
||||
fc.do(http.MethodPost, "/api/v1/agents/register", registerAgentRequest{ValidatorID: "validator-1"})
|
||||
|
||||
runOneCycle := func() {
|
||||
orch.Tick(ctx)
|
||||
_, body := fc.do(http.MethodGet, "/api/v1/agents/validator-1/assignment", nil)
|
||||
var assignment assignmentResponse
|
||||
if err := json.Unmarshal(body, &assignment); err != nil {
|
||||
t.Fatalf("unmarshal assignment: %v", err)
|
||||
}
|
||||
fc.do(http.MethodPost, "/api/v1/agents/validator-1/self-check", selfCheckRequest{
|
||||
IPID: assignment.IPID, DetectedEgress: "9.9.9.9", Success: true,
|
||||
})
|
||||
fc.do(http.MethodPost, "/api/v1/agents/validator-1/results", agentResultsRequest{
|
||||
Results: []checkResultDTO{{
|
||||
IPID: assignment.IPID, CheckType: "https", Target: "https://example.test",
|
||||
Success: true, CheckedAt: time.Now().Format(time.RFC3339Nano),
|
||||
}},
|
||||
})
|
||||
fc.do(http.MethodPost, "/api/v1/agents/validator-1/complete", agentCompleteRequest{IPID: assignment.IPID})
|
||||
orch.Tick(ctx)
|
||||
}
|
||||
|
||||
fc.do(http.MethodPost, "/api/v1/admin/ips", submitIPsRequest{Addresses: []string{"9.9.9.9"}})
|
||||
runOneCycle()
|
||||
|
||||
item, err := d.GetIPByAddress(ctx, "9.9.9.9")
|
||||
if err != nil {
|
||||
t.Fatalf("get ip: %v", err)
|
||||
}
|
||||
if item.State != db.IPDone {
|
||||
t.Fatalf("expected done after first cycle, got %s", item.State)
|
||||
}
|
||||
|
||||
resp, _ := fc.do(http.MethodDelete, "/api/v1/admin/ips/9.9.9.9", nil)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("delete ip: status=%d", resp.StatusCode)
|
||||
}
|
||||
|
||||
resp, _ = fc.do(http.MethodGet, "/api/v1/admin/ips/9.9.9.9", nil)
|
||||
if resp.StatusCode != http.StatusNotFound {
|
||||
t.Fatalf("expected 404 for deleted address on /ips, got %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
resp, body := fc.do(http.MethodGet, "/api/v1/admin/registry/9.9.9.9", nil)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("registry history: status=%d body=%s", resp.StatusCode, body)
|
||||
}
|
||||
var hist struct {
|
||||
Registry registryDTO `json:"registry"`
|
||||
Checks []db.Check `json:"checks"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &hist); err != nil {
|
||||
t.Fatalf("unmarshal registry history: %v", err)
|
||||
}
|
||||
if hist.Registry.TotalCycles != 1 {
|
||||
t.Fatalf("expected 1 retained cycle, got %+v", hist.Registry)
|
||||
}
|
||||
if len(hist.Checks) == 0 {
|
||||
t.Fatalf("expected retained check history, got none")
|
||||
}
|
||||
if hist.Registry.InQueue {
|
||||
t.Fatalf("expected registry entry to report not-in-queue after delete, got %+v", hist.Registry)
|
||||
}
|
||||
|
||||
// Resubmitting starts a second, distinct cycle on the same registry
|
||||
// entry rather than colliding with the first.
|
||||
mock.Seed("fip-1", "9.9.9.9", "svc-project")
|
||||
fc.do(http.MethodPost, "/api/v1/admin/ips", submitIPsRequest{Addresses: []string{"9.9.9.9"}})
|
||||
runOneCycle()
|
||||
|
||||
resp, body = fc.do(http.MethodGet, "/api/v1/admin/registry/9.9.9.9", nil)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("registry history (2nd): status=%d body=%s", resp.StatusCode, body)
|
||||
}
|
||||
if err := json.Unmarshal(body, &hist); err != nil {
|
||||
t.Fatalf("unmarshal registry history (2nd): %v", err)
|
||||
}
|
||||
if hist.Registry.TotalCycles != 2 {
|
||||
t.Fatalf("expected 2 retained cycles after resubmission, got %+v", hist.Registry)
|
||||
}
|
||||
if !hist.Registry.InQueue {
|
||||
t.Fatalf("expected registry entry to report in-queue again, got %+v", hist.Registry)
|
||||
}
|
||||
|
||||
_, body = fc.do(http.MethodGet, "/api/v1/admin/registry", nil)
|
||||
var all []registryDTO
|
||||
if err := json.Unmarshal(body, &all); err != nil {
|
||||
t.Fatalf("unmarshal registry list: %v", err)
|
||||
}
|
||||
if len(all) != 1 || all[0].IPAddress != "9.9.9.9" {
|
||||
t.Fatalf("expected a single registry entry for the address, got %+v", all)
|
||||
}
|
||||
}
|
||||
@@ -21,6 +21,7 @@ func (s *Server) routes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("GET /api/v1/admin/status", s.handleAdminStatus)
|
||||
mux.HandleFunc("GET /api/v1/admin/ips", s.handleAdminIPs)
|
||||
mux.HandleFunc("POST /api/v1/admin/ips", s.handleAdminSubmitIPs)
|
||||
mux.HandleFunc("POST /api/v1/admin/ips/scan", s.handleAdminScanFloatingIPs)
|
||||
mux.HandleFunc("GET /api/v1/admin/ips/{ip}", s.handleAdminIPDetail)
|
||||
mux.HandleFunc("POST /api/v1/admin/ips/{ip}/cancel", s.handleAdminCancelIP)
|
||||
mux.HandleFunc("DELETE /api/v1/admin/ips/{ip}", s.handleAdminDeleteIP)
|
||||
@@ -28,6 +29,9 @@ func (s *Server) routes(mux *http.ServeMux) {
|
||||
mux.HandleFunc("POST /api/v1/admin/ips/clear", s.handleAdminClearQueue)
|
||||
mux.HandleFunc("GET /api/v1/admin/validators", s.handleAdminValidators)
|
||||
|
||||
mux.HandleFunc("GET /api/v1/admin/registry", s.handleAdminRegistry)
|
||||
mux.HandleFunc("GET /api/v1/admin/registry/{ip}", s.handleAdminRegistryHistory)
|
||||
|
||||
mux.HandleFunc("GET /api/v1/admin/config/validators", s.handleConfigListValidators)
|
||||
mux.HandleFunc("POST /api/v1/admin/config/validators", s.handleConfigCreateValidator)
|
||||
mux.HandleFunc("PUT /api/v1/admin/config/validators/{id}", s.handleConfigUpdateValidator)
|
||||
|
||||
@@ -150,6 +150,22 @@ func (c *Client) GetFloatingIPByAddress(ctx context.Context, address string) (*F
|
||||
return &FloatingIP{ID: f.ID, Address: f.FloatingIP, PortID: f.PortID, ProjectID: f.TenantID}, nil
|
||||
}
|
||||
|
||||
func (c *Client) ListFloatingIPs(ctx context.Context) ([]FloatingIP, error) {
|
||||
pages, err := floatingips.List(c.networking, floatingips.ListOpts{}).AllPages(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("openstack: list floating ips: %w", err)
|
||||
}
|
||||
list, err := floatingips.ExtractFloatingIPs(pages)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("openstack: extract floating ips: %w", err)
|
||||
}
|
||||
out := make([]FloatingIP, 0, len(list))
|
||||
for _, f := range list {
|
||||
out = append(out, FloatingIP{ID: f.ID, Address: f.FloatingIP, PortID: f.PortID, ProjectID: f.TenantID})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (c *Client) AssociateFloatingIP(ctx context.Context, fipID, portID string) error {
|
||||
_, err := floatingips.Update(ctx, c.networking, fipID, floatingips.UpdateOpts{
|
||||
PortID: &portID,
|
||||
|
||||
@@ -24,6 +24,12 @@ type FloatingIPClient interface {
|
||||
// is registered in the service project.
|
||||
GetFloatingIPByAddress(ctx context.Context, address string) (*FloatingIP, error)
|
||||
|
||||
// ListFloatingIPs returns every floating IP registered in the service
|
||||
// project, both associated and free — filtering to the free pool (empty
|
||||
// PortID) is the caller's job. Used to discover addresses to feed into
|
||||
// the check queue without an admin having to enumerate them by hand.
|
||||
ListFloatingIPs(ctx context.Context) ([]FloatingIP, error)
|
||||
|
||||
// AssociateFloatingIP attaches the floating IP to the given Neutron
|
||||
// port (the validator's primary NIC port).
|
||||
AssociateFloatingIP(ctx context.Context, fipID, portID string) error
|
||||
|
||||
@@ -56,6 +56,16 @@ func (m *MockClient) GetFloatingIPByAddress(ctx context.Context, address string)
|
||||
return &f, nil
|
||||
}
|
||||
|
||||
func (m *MockClient) ListFloatingIPs(ctx context.Context) ([]FloatingIP, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
out := make([]FloatingIP, 0, len(m.fips))
|
||||
for _, f := range m.fips {
|
||||
out = append(out, *f)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (m *MockClient) AssociateFloatingIP(ctx context.Context, fipID, portID string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
@@ -395,6 +395,43 @@ func (o *Orchestrator) ClearQueue(ctx context.Context) (db.DeleteIPsResult, erro
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// ScanFloatingIPs lists every floating IP in the OpenStack project, filters
|
||||
// to the ones not currently associated to any port (the free pool awaiting
|
||||
// validation before reissue), and submits that address list to the check
|
||||
// queue via db.SubmitIPs — the same entry point the admin API's "add
|
||||
// addresses" call uses, so add/requeue/reorder semantics are identical
|
||||
// whether the address list came from an operator or from this scan. Returns
|
||||
// the SubmitIPs outcome plus how many free floating IPs were found in total
|
||||
// (which can be larger than the sum of the SubmitIPsResult slices, since
|
||||
// addresses already mid-check are silently skipped — see db.SubmitIPs).
|
||||
func (o *Orchestrator) ScanFloatingIPs(ctx context.Context) (db.SubmitIPsResult, int, error) {
|
||||
fips, err := o.OS.ListFloatingIPs(ctx)
|
||||
if err != nil {
|
||||
return db.SubmitIPsResult{}, 0, fmt.Errorf("list floating ips: %w", err)
|
||||
}
|
||||
|
||||
var free []string
|
||||
for _, f := range fips {
|
||||
if f.PortID == "" {
|
||||
free = append(free, f.Address)
|
||||
}
|
||||
}
|
||||
|
||||
if len(free) == 0 {
|
||||
o.event(ctx, "control-api", "", nil, "fip_scan", `{"scanned_free":0}`)
|
||||
return db.SubmitIPsResult{}, 0, nil
|
||||
}
|
||||
|
||||
result, err := o.DB.SubmitIPs(ctx, free)
|
||||
if err != nil {
|
||||
return result, len(free), fmt.Errorf("submit scanned ips: %w", err)
|
||||
}
|
||||
o.event(ctx, "control-api", "", nil, "fip_scan", fmt.Sprintf(
|
||||
`{"scanned_free":%d,"added":%d,"requeued":%d,"reordered":%d,"skipped_in_progress":%d}`,
|
||||
len(free), len(result.Added), len(result.Requeued), len(result.Reordered), len(result.SkippedInProgress)))
|
||||
return result, len(free), nil
|
||||
}
|
||||
|
||||
// deletedAddressesPayload builds the event payload for the batch delete
|
||||
// operations — a proper JSON array via encoding/json rather than fmt's %q
|
||||
// slice formatting (which produces space-separated quoted strings, not
|
||||
@@ -508,6 +545,14 @@ func (o *Orchestrator) aggregateAndRelease(ctx context.Context, item db.IPQueueI
|
||||
o.event(ctx, "control-api", "", &item.ID, "aggregated",
|
||||
fmt.Sprintf(`{"result":%q,"checks":%d,"passed":%d,"missing":%d}`, result, len(checks), passCount, missing))
|
||||
|
||||
if settings, err := o.DB.GetSettings(ctx); err != nil {
|
||||
o.Log.Error("get settings for history retention", "ip_id", item.ID, "err", err)
|
||||
} else if settings.HistoryRetentionCycles > 0 {
|
||||
if err := o.DB.PruneRegistryHistory(ctx, item.RegistryID, settings.HistoryRetentionCycles); err != nil {
|
||||
o.Log.Error("prune registry history", "ip_id", item.ID, "registry_id", item.RegistryID, "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
if item.FIPID != "" {
|
||||
if err := o.OS.DisassociateFloatingIP(ctx, item.FIPID); err != nil {
|
||||
o.Log.Error("disassociate fip", "ip_id", item.ID, "fip_id", item.FIPID, "err", err)
|
||||
|
||||
@@ -882,3 +882,60 @@ func TestAggregationWaitsForTLSAndSSHResults(t *testing.T) {
|
||||
t.Fatalf("expected partial (missing ssh/tls-443 counted against it), got %s", ip.OverallResult)
|
||||
}
|
||||
}
|
||||
|
||||
// TestScanFloatingIPsSubmitsOnlyFreeAddresses proves ScanFloatingIPs filters
|
||||
// out floating IPs already associated to a port and only submits the free
|
||||
// pool to the check queue.
|
||||
func TestScanFloatingIPsSubmitsOnlyFreeAddresses(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
o, d, mock := newTestOrchestrator(t, 180)
|
||||
mock.Seed("fip-free-1", "1.1.1.1", "svc-project")
|
||||
mock.Seed("fip-free-2", "2.2.2.2", "svc-project")
|
||||
mock.SeedWithPort("fip-occupied", "3.3.3.3", "svc-project", "some-other-port")
|
||||
|
||||
result, scanned, err := o.ScanFloatingIPs(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("scan floating ips: %v", err)
|
||||
}
|
||||
if scanned != 2 {
|
||||
t.Fatalf("expected 2 free fips scanned, got %d", scanned)
|
||||
}
|
||||
if len(result.Added) != 2 {
|
||||
t.Fatalf("expected 2 addresses added, got %+v", result)
|
||||
}
|
||||
|
||||
ips, err := d.ListIPs(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("list ips: %v", err)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, ip := range ips {
|
||||
seen[ip.IPAddress] = true
|
||||
}
|
||||
if !seen["1.1.1.1"] || !seen["2.2.2.2"] {
|
||||
t.Fatalf("expected free addresses queued, got %+v", ips)
|
||||
}
|
||||
if seen["3.3.3.3"] {
|
||||
t.Fatalf("expected occupied address not queued, got %+v", ips)
|
||||
}
|
||||
}
|
||||
|
||||
// TestScanFloatingIPsNoFreeAddressesIsNotAnError proves scanning a project
|
||||
// with no free floating IPs (or none at all) succeeds with an empty result
|
||||
// rather than hitting SubmitIPs' "addresses must not be empty" validation.
|
||||
func TestScanFloatingIPsNoFreeAddressesIsNotAnError(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
o, _, mock := newTestOrchestrator(t, 180)
|
||||
mock.SeedWithPort("fip-occupied", "3.3.3.3", "svc-project", "some-other-port")
|
||||
|
||||
result, scanned, err := o.ScanFloatingIPs(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("scan floating ips: %v", err)
|
||||
}
|
||||
if scanned != 0 {
|
||||
t.Fatalf("expected 0 free fips scanned, got %d", scanned)
|
||||
}
|
||||
if len(result.Added) != 0 {
|
||||
t.Fatalf("expected nothing added, got %+v", result)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user