| Адрес | Состояние | Валидатор | Попытка | Назначено |
|---|---|---|---|---|
| {{.IPAddress}} | +{{$b.Label}} | +{{deref .OwnerValidatorID}} | +{{.AttemptNumber}} | +{{fmtTime .AssignedAt}} | +
Сейчас нет адресов в обработке.
+{{end}} + ++pass: {{index .Breakdown "pass"}} · +partial: {{index .Breakdown "partial"}} · +fail: {{index .Breakdown "fail"}} · +cancelled: {{index .Breakdown "cancelled"}} +
+{{if .LastCompleted}} +| Адрес | Итог | Завершено |
|---|---|---|
| {{.IPAddress}} | +{{$b.Label}} | +{{fmtTime .AggregatedAt}} | +
Пока ничего не завершено.
+{{end}} +{{end}} diff --git a/internal/dashboard/templates/sites.html b/internal/dashboard/templates/sites.html new file mode 100644 index 0000000..6ced45b --- /dev/null +++ b/internal/dashboard/templates/sites.html @@ -0,0 +1,47 @@ +{{define "sites_page"}} + + +{{template "html_head" .}} + +{{template "page_header" .}} +Ровно три слота (1, 2, 3) — жёсткое ограничение схемы БД control-api. +Пустой слот — площадка не используется, входящие проверки для неё не ожидаются.
+ +| Слот | site_id | Действия |
|---|---|---|
| {{.Index}} | ++ + | ++{{if .SiteID}} + +{{end}} + | +
Группа целей — именованный список URL/адресов, на который затем ссылаются типы проверок.
+ + + +| Группа | Цели | Действия |
|---|---|---|
| {{.Name}} | ++ + | ++ + | +
Групп целей нет.
{{end}} +{{end}} diff --git a/internal/dashboard/templates/validators.html b/internal/dashboard/templates/validators.html new file mode 100644 index 0000000..03c7092 --- /dev/null +++ b/internal/dashboard/templates/validators.html @@ -0,0 +1,54 @@ +{{define "validators_page"}} + + +{{template "html_head" .}} + +{{template "page_header" .}} +| ID | OS Port ID | Состояние | Действия |
|---|---|---|---|
| {{.ValidatorID}} | ++ + | +{{$b.Label}} | ++ + | +
Валидаторов нет.
{{end}} +{{end}} diff --git a/internal/dashboard/util.go b/internal/dashboard/util.go new file mode 100644 index 0000000..8aad98c --- /dev/null +++ b/internal/dashboard/util.go @@ -0,0 +1,21 @@ +package dashboard + +import "strings" + +// splitList parses a textarea/hidden-input value into a clean list, +// accepting entries separated by newlines and/or commas interchangeably +// (different forms in this package use one or the other) and dropping +// blank entries from stray whitespace. +func splitList(raw string) []string { + fields := strings.FieldsFunc(raw, func(r rune) bool { + return r == '\n' || r == '\r' || r == ',' + }) + out := make([]string, 0, len(fields)) + for _, f := range fields { + f = strings.TrimSpace(f) + if f != "" { + out = append(out, f) + } + } + return out +} diff --git a/internal/db/bootstrap.go b/internal/db/bootstrap.go new file mode 100644 index 0000000..d15cac9 --- /dev/null +++ b/internal/db/bootstrap.go @@ -0,0 +1,104 @@ +package db + +import ( + "context" + "fmt" + + "cloudipvalidator/internal/config" +) + +// BootstrapFromConfig ensures the database's baseline state matches the +// YAML config on a fresh install, then gets out of the way. +// +// - ip_addresses -> SeedQueue: unchanged, always additive. This is a +// separate concept from the runtime admin queue API (SubmitIPs) — every +// control-api startup re-adds any address from cfg.IPAddresses that +// isn't already in the queue. +// - validators / sites / target_groups / check_types: applied from YAML +// only if the corresponding table is currently empty. Once any row +// exists (via this bootstrap or the admin API), YAML for that section +// is ignored on every subsequent startup — the database is the source +// of truth from that point on. This is what makes admin API changes to +// these four entities survive a restart. +func (d *DB) BootstrapFromConfig(ctx context.Context, cfg *config.ControlAPI) error { + if err := d.bootstrapValidators(ctx, cfg.Validators); err != nil { + return fmt.Errorf("bootstrap validators: %w", err) + } + if err := d.bootstrapSites(ctx, cfg.Sites); err != nil { + return fmt.Errorf("bootstrap sites: %w", err) + } + if err := d.bootstrapTargetGroups(ctx, cfg.Targets); err != nil { + return fmt.Errorf("bootstrap target groups: %w", err) + } + if err := d.bootstrapCheckTypes(ctx, cfg.CheckTypes); err != nil { + return fmt.Errorf("bootstrap check types: %w", err) + } + if err := d.SeedQueue(ctx, cfg.IPAddresses); err != nil { + return fmt.Errorf("seed ip queue: %w", err) + } + return nil +} + +func (d *DB) bootstrapValidators(ctx context.Context, validators []config.ValidatorConfig) error { + var count int + if err := d.QueryRowContext(ctx, `SELECT COUNT(*) FROM validators`).Scan(&count); err != nil { + return err + } + if count > 0 { + return nil + } + for _, v := range validators { + if err := d.RegisterValidator(ctx, v.ValidatorID, "", v.OSPortID, ""); err != nil { + return fmt.Errorf("seed validator %s: %w", v.ValidatorID, err) + } + } + return nil +} + +func (d *DB) bootstrapSites(ctx context.Context, sites []config.SiteConfig) error { + var count int + if err := d.QueryRowContext(ctx, `SELECT COUNT(*) FROM sites`).Scan(&count); err != nil { + return err + } + if count > 0 { + return nil + } + for _, s := range sites { + if err := d.UpsertSite(ctx, s.Index, s.SiteID); err != nil { + return fmt.Errorf("seed site %s: %w", s.SiteID, err) + } + } + return nil +} + +func (d *DB) bootstrapTargetGroups(ctx context.Context, targets map[string][]string) error { + var count int + if err := d.QueryRowContext(ctx, `SELECT COUNT(*) FROM target_groups`).Scan(&count); err != nil { + return err + } + if count > 0 { + return nil + } + for name, addrs := range targets { + if err := d.UpsertTargetGroup(ctx, name, addrs); err != nil { + return fmt.Errorf("seed target group %s: %w", name, err) + } + } + return nil +} + +func (d *DB) bootstrapCheckTypes(ctx context.Context, checkTypes []config.CheckTypeConfig) error { + var count int + if err := d.QueryRowContext(ctx, `SELECT COUNT(*) FROM check_types`).Scan(&count); err != nil { + return err + } + if count > 0 { + return nil + } + for _, ct := range checkTypes { + if err := d.UpsertCheckType(ctx, ct.Name, ct.Enabled, ct.Targets); err != nil { + return fmt.Errorf("seed check type %s: %w", ct.Name, err) + } + } + return nil +} diff --git a/internal/db/db.go b/internal/db/db.go index de3eceb..163390c 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -16,6 +16,21 @@ import ( //go:embed migrations/0001_init.sql var initSchema string +//go:embed migrations/0002_dynamic_config.sql +var dynamicConfigSchema 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 +// existing one only picks up what's new. +var migrations = []struct { + version int + sql string +}{ + {1, initSchema}, + {2, dynamicConfigSchema}, +} + type DB struct { *sql.DB } @@ -52,30 +67,36 @@ func Open(ctx context.Context, path string) (*DB, error) { return d, nil } -// migrate applies the embedded schema exactly once, tracked via -// PRAGMA user_version so repeated startups are no-ops. +// migrate applies every pending migration in order, tracked via +// PRAGMA user_version so repeated startups only apply what's new (and a +// fresh database walks the whole list once). func (d *DB) migrate(ctx context.Context) error { var version int if err := d.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil { return fmt.Errorf("read user_version: %w", err) } - if version >= 1 { - return nil - } - tx, err := d.BeginTx(ctx, nil) - if err != nil { - return err + for _, m := range migrations { + if m.version <= version { + continue + } + tx, err := d.BeginTx(ctx, nil) + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, m.sql); err != nil { + tx.Rollback() + return fmt.Errorf("apply migration %d: %w", m.version, err) + } + if _, err := tx.ExecContext(ctx, fmt.Sprintf("PRAGMA user_version=%d", m.version)); err != nil { + tx.Rollback() + return fmt.Errorf("set user_version=%d: %w", m.version, err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit migration %d: %w", m.version, err) + } } - defer tx.Rollback() - - if _, err := tx.ExecContext(ctx, initSchema); err != nil { - return fmt.Errorf("apply 0001_init.sql: %w", err) - } - if _, err := tx.ExecContext(ctx, "PRAGMA user_version=1"); err != nil { - return fmt.Errorf("set user_version: %w", err) - } - return tx.Commit() + return nil } // Now returns the current time truncated to millisecond precision, the diff --git a/internal/db/errors.go b/internal/db/errors.go new file mode 100644 index 0000000..e854dfd --- /dev/null +++ b/internal/db/errors.go @@ -0,0 +1,26 @@ +package db + +import "errors" + +// Sentinel errors returned by the dynamic-config and IP-queue admin +// mutations. Callers (internal/httpapi) map these to HTTP status codes via +// errors.Is rather than defaulting everything to 500. +var ( + // ErrNotFound: the referenced entity does not exist. + ErrNotFound = errors.New("not found") + // ErrConflict: the entity already exists, or a unique slot/key is + // already taken by something else. + ErrConflict = errors.New("conflict") + // ErrBusy: the entity can't be removed because it's currently in use + // (e.g. a validator that owns an in-progress IP). + ErrBusy = errors.New("busy") + // ErrInUse: the entity can't be removed because something else + // references it (e.g. a target group referenced by a check type). + ErrInUse = errors.New("in use") + // ErrValidation: the request is well-formed but refers to something + // invalid (e.g. a check type naming a target group that doesn't exist). + ErrValidation = errors.New("validation failed") + // ErrInvalidState: the requested state transition isn't valid from the + // entity's current state (e.g. cancelling an already-finished IP). + ErrInvalidState = errors.New("invalid state") +) diff --git a/internal/db/migrations/0002_dynamic_config.sql b/internal/db/migrations/0002_dynamic_config.sql new file mode 100644 index 0000000..fc3bdf1 --- /dev/null +++ b/internal/db/migrations/0002_dynamic_config.sql @@ -0,0 +1,26 @@ +-- Dynamic configuration tables: sites, target_groups, check_types. +-- validators already exists (0001_init.sql) and needs no new columns. +-- Bootstrap semantics (apply YAML only when the table is empty) live in +-- internal/db/bootstrap.go, not in this migration. + +CREATE TABLE sites ( + idx INTEGER PRIMARY KEY, -- 1, 2 or 3 -- fixed slot + site_id TEXT NOT NULL UNIQUE, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL +); + +CREATE TABLE target_groups ( + group_name TEXT PRIMARY KEY, + targets TEXT NOT NULL, -- JSON array of strings + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL +); + +CREATE TABLE check_types ( + name TEXT PRIMARY KEY, + enabled BOOLEAN NOT NULL DEFAULT 1, + target_groups TEXT NOT NULL, -- JSON array of group names + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL +); diff --git a/internal/db/models.go b/internal/db/models.go index 75565ce..e5eb09b 100644 --- a/internal/db/models.go +++ b/internal/db/models.go @@ -20,9 +20,10 @@ const ( IPDone = "done" IPFailed = "failed" - ResultPass = "pass" - ResultPartial = "partial" - ResultFail = "fail" + ResultPass = "pass" + ResultPartial = "partial" + ResultFail = "fail" + ResultCancelled = "cancelled" SourceEgress = "egress" ) @@ -115,3 +116,41 @@ type Event struct { OccurredAt time.Time CreatedAt time.Time } + +type Site struct { + Index int + SiteID string + CreatedAt time.Time + UpdatedAt time.Time +} + +type TargetGroup struct { + Name string + Targets []string + CreatedAt time.Time + UpdatedAt time.Time +} + +type CheckType struct { + Name string + Enabled bool + TargetGroups []string + CreatedAt time.Time + UpdatedAt time.Time +} + +// ResolvedCheckType is a check type with its target groups already expanded +// into a flat target list — what AssignmentForValidator hands to an agent. +type ResolvedCheckType struct { + Type string + Targets []string +} + +// SubmitIPsResult categorizes how each address in a SubmitIPs call was +// handled. +type SubmitIPsResult struct { + Added []string + Requeued []string + Reordered []string + SkippedInProgress []string +} diff --git a/internal/db/queries_checktypes.go b/internal/db/queries_checktypes.go new file mode 100644 index 0000000..e805ead --- /dev/null +++ b/internal/db/queries_checktypes.go @@ -0,0 +1,141 @@ +package db + +import ( + "context" + "encoding/json" + "fmt" +) + +// ListCheckTypes returns all configured check types, ordered by name. +func (d *DB) ListCheckTypes(ctx context.Context) ([]CheckType, error) { + rows, err := d.QueryContext(ctx, ` + SELECT name, enabled, target_groups, created_at, updated_at FROM check_types ORDER BY name + `) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []CheckType + for rows.Next() { + ct, err := scanCheckType(rows) + if err != nil { + return nil, err + } + out = append(out, *ct) + } + return out, rows.Err() +} + +// ListResolvedCheckTypes returns every enabled check type with its target +// groups already expanded into a flat target list — the egress +// check-type/target configuration handed to a validator-agent once its IP +// is ready to be worked on. This is the dynamic equivalent of what +// orchestrator.New used to compute once from static YAML config. +func (d *DB) ListResolvedCheckTypes(ctx context.Context) ([]ResolvedCheckType, error) { + groups, err := d.loadTargetGroupsMap(ctx) + if err != nil { + return nil, err + } + + rows, err := d.QueryContext(ctx, ` + SELECT name, target_groups FROM check_types WHERE enabled=1 ORDER BY name + `) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []ResolvedCheckType + for rows.Next() { + var name, groupsJSON string + if err := rows.Scan(&name, &groupsJSON); err != nil { + return nil, err + } + var groupNames []string + if err := json.Unmarshal([]byte(groupsJSON), &groupNames); err != nil { + return nil, fmt.Errorf("decode target_groups for check type %q: %w", name, err) + } + var targets []string + for _, g := range groupNames { + targets = append(targets, groups[g]...) + } + out = append(out, ResolvedCheckType{Type: name, Targets: targets}) + } + return out, rows.Err() +} + +func (d *DB) loadTargetGroupsMap(ctx context.Context) (map[string][]string, error) { + groups, err := d.ListTargetGroups(ctx) + if err != nil { + return nil, err + } + out := make(map[string][]string, len(groups)) + for _, g := range groups { + out[g.Name] = g.Targets + } + return out, nil +} + +// UpsertCheckType creates or replaces a check type. All named target groups +// must already exist. +func (d *DB) UpsertCheckType(ctx context.Context, name string, enabled bool, targetGroups []string) error { + if name == "" { + return fmt.Errorf("check type name must not be empty: %w", ErrValidation) + } + existing, err := d.loadTargetGroupsMap(ctx) + if err != nil { + return err + } + for _, g := range targetGroups { + if _, ok := existing[g]; !ok { + return fmt.Errorf("check type %q references unknown target group %q: %w", name, g, ErrValidation) + } + } + payload, err := json.Marshal(targetGroups) + if err != nil { + return err + } + now := timeToDB(Now()) + _, err = d.ExecContext(ctx, ` + INSERT INTO check_types (name, enabled, target_groups, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(name) DO UPDATE SET + enabled=excluded.enabled, target_groups=excluded.target_groups, updated_at=excluded.updated_at + `, name, enabled, string(payload), now, now) + if err != nil { + return fmt.Errorf("upsert check type: %w", err) + } + return nil +} + +// DeleteCheckType removes a check type. +func (d *DB) DeleteCheckType(ctx context.Context, name string) error { + res, err := d.ExecContext(ctx, `DELETE FROM check_types WHERE name=?`, name) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return fmt.Errorf("check type %q: %w", name, ErrNotFound) + } + return nil +} + +func scanCheckType(row rowScanner) (*CheckType, error) { + var ct CheckType + var groupsJSON, createdAt, updatedAt string + if err := row.Scan(&ct.Name, &ct.Enabled, &groupsJSON, &createdAt, &updatedAt); err != nil { + return nil, err + } + if err := json.Unmarshal([]byte(groupsJSON), &ct.TargetGroups); err != nil { + return nil, fmt.Errorf("decode target_groups for check type %q: %w", ct.Name, err) + } + var err error + if ct.CreatedAt, err = dbToTime(createdAt); err != nil { + return nil, err + } + if ct.UpdatedAt, err = dbToTime(updatedAt); err != nil { + return nil, err + } + return &ct, nil +} diff --git a/internal/db/queries_dynconfig_test.go b/internal/db/queries_dynconfig_test.go new file mode 100644 index 0000000..647bf38 --- /dev/null +++ b/internal/db/queries_dynconfig_test.go @@ -0,0 +1,332 @@ +package db + +import ( + "context" + "errors" + "path/filepath" + "testing" + "time" + + "cloudipvalidator/internal/config" +) + +func newTestDB(t *testing.T) (*DB, context.Context) { + t.Helper() + ctx := context.Background() + d, err := Open(ctx, filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatalf("open db: %v", err) + } + t.Cleanup(func() { d.Close() }) + return d, ctx +} + +func TestUpsertSiteValidation(t *testing.T) { + d, ctx := newTestDB(t) + + if err := d.UpsertSite(ctx, 0, "site-1"); !errors.Is(err, ErrValidation) { + t.Fatalf("expected ErrValidation for idx=0, got %v", err) + } + if err := d.UpsertSite(ctx, 4, "site-1"); !errors.Is(err, ErrValidation) { + t.Fatalf("expected ErrValidation for idx=4, got %v", err) + } + if err := d.UpsertSite(ctx, 1, ""); !errors.Is(err, ErrValidation) { + t.Fatalf("expected ErrValidation for empty site_id, got %v", err) + } + + if err := d.UpsertSite(ctx, 1, "site-1"); err != nil { + t.Fatalf("upsert site 1: %v", err) + } + if err := d.UpsertSite(ctx, 2, "site-1"); !errors.Is(err, ErrConflict) { + t.Fatalf("expected ErrConflict assigning site-1 to slot 2, got %v", err) + } + // Renaming the same slot is fine (not a conflict with itself). + if err := d.UpsertSite(ctx, 1, "site-1-renamed"); err != nil { + t.Fatalf("rename slot 1: %v", err) + } + + if err := d.DeleteSite(ctx, 1); err != nil { + t.Fatalf("delete site 1: %v", err) + } + if err := d.DeleteSite(ctx, 1); !errors.Is(err, ErrNotFound) { + t.Fatalf("expected ErrNotFound deleting already-gone slot, got %v", err) + } +} + +func TestTargetGroupInUseCannotBeDeleted(t *testing.T) { + d, ctx := newTestDB(t) + + if err := d.UpsertTargetGroup(ctx, "web", []string{"https://example.test"}); err != nil { + t.Fatalf("upsert target group: %v", err) + } + if err := d.UpsertTargetGroup(ctx, "empty", nil); !errors.Is(err, ErrValidation) { + t.Fatalf("expected ErrValidation for empty targets, got %v", err) + } + if err := d.UpsertCheckType(ctx, "https", true, []string{"web"}); err != nil { + t.Fatalf("upsert check type: %v", err) + } + + if err := d.DeleteTargetGroup(ctx, "web"); !errors.Is(err, ErrInUse) { + t.Fatalf("expected ErrInUse deleting group referenced by check type, got %v", err) + } + + if err := d.DeleteCheckType(ctx, "https"); err != nil { + t.Fatalf("delete check type: %v", err) + } + if err := d.DeleteTargetGroup(ctx, "web"); err != nil { + t.Fatalf("delete target group after check type removed: %v", err) + } +} + +func TestUpsertCheckTypeUnknownGroup(t *testing.T) { + d, ctx := newTestDB(t) + + err := d.UpsertCheckType(ctx, "https", true, []string{"does-not-exist"}) + if !errors.Is(err, ErrValidation) { + t.Fatalf("expected ErrValidation for unknown target group, got %v", err) + } +} + +func TestListResolvedCheckTypesExpandsGroups(t *testing.T) { + d, ctx := newTestDB(t) + + if err := d.UpsertTargetGroup(ctx, "web", []string{"https://a.test", "https://b.test"}); err != nil { + t.Fatalf("upsert web group: %v", err) + } + if err := d.UpsertTargetGroup(ctx, "extra", []string{"https://c.test"}); err != nil { + t.Fatalf("upsert extra group: %v", err) + } + if err := d.UpsertCheckType(ctx, "https", true, []string{"web", "extra"}); err != nil { + t.Fatalf("upsert https check type: %v", err) + } + if err := d.UpsertCheckType(ctx, "ssh", false, []string{"web"}); err != nil { + t.Fatalf("upsert disabled ssh check type: %v", err) + } + + resolved, err := d.ListResolvedCheckTypes(ctx) + if err != nil { + t.Fatalf("list resolved check types: %v", err) + } + if len(resolved) != 1 { + t.Fatalf("expected only the enabled check type, got %d entries: %+v", len(resolved), resolved) + } + if resolved[0].Type != "https" || len(resolved[0].Targets) != 3 { + t.Fatalf("expected https with 3 expanded targets, got %+v", resolved[0]) + } +} + +func TestAdminValidatorCRUD(t *testing.T) { + d, ctx := newTestDB(t) + + if err := d.AdminCreateValidator(ctx, "validator-1", "port-1"); err != nil { + t.Fatalf("create validator: %v", err) + } + if err := d.AdminCreateValidator(ctx, "validator-1", "port-2"); !errors.Is(err, ErrConflict) { + t.Fatalf("expected ErrConflict creating duplicate validator, got %v", err) + } + if err := d.AdminUpdateValidatorPort(ctx, "validator-1", "port-3"); err != nil { + t.Fatalf("update validator port: %v", err) + } + v, err := d.GetValidator(ctx, "validator-1") + if err != nil || v.OSPortID != "port-3" { + t.Fatalf("expected port-3 after update, got %+v (err=%v)", v, err) + } + if err := d.AdminUpdateValidatorPort(ctx, "no-such-validator", "port-9"); !errors.Is(err, ErrNotFound) { + t.Fatalf("expected ErrNotFound updating unknown validator, got %v", err) + } + + // Claim it onto an IP, then deletion should be refused as busy. + if err := d.SeedQueue(ctx, []string{"1.2.3.4"}); err != nil { + t.Fatalf("seed queue: %v", err) + } + if _, err := d.ClaimNextQueued(ctx, "validator-1", time.Minute); err != nil { + t.Fatalf("claim next queued: %v", err) + } + if err := d.DeleteValidator(ctx, "validator-1"); !errors.Is(err, ErrBusy) { + t.Fatalf("expected ErrBusy deleting validator that owns an ip, got %v", err) + } + + if err := d.ReleaseFIP(ctx, 1, "validator-1"); err != nil { + t.Fatalf("release fip: %v", err) + } + if err := d.DeleteValidator(ctx, "validator-1"); err != nil { + t.Fatalf("delete validator after release: %v", err) + } + if err := d.DeleteValidator(ctx, "validator-1"); !errors.Is(err, ErrNotFound) { + t.Fatalf("expected ErrNotFound deleting already-gone validator, got %v", err) + } +} + +func TestBootstrapFromConfigIgnoresYAMLOnNonEmptyTables(t *testing.T) { + d, ctx := newTestDB(t) + + cfg := &config.ControlAPI{ + Validators: []config.ValidatorConfig{{ValidatorID: "validator-1", OSPortID: "port-1"}}, + Sites: []config.SiteConfig{{SiteID: "site-1", Index: 1}}, + Targets: map[string][]string{"web": {"https://example.test"}}, + CheckTypes: []config.CheckTypeConfig{{Name: "https", Enabled: true, Targets: []string{"web"}}}, + } + if err := d.BootstrapFromConfig(ctx, cfg); err != nil { + t.Fatalf("first bootstrap: %v", err) + } + + // Simulate an admin API change that should survive a second bootstrap + // pass (e.g. a restart) even though the YAML still names the old port. + if err := d.AdminUpdateValidatorPort(ctx, "validator-1", "port-changed-via-api"); err != nil { + t.Fatalf("update validator port via api: %v", err) + } + if err := d.UpsertSite(ctx, 1, "site-changed-via-api"); err != nil { + t.Fatalf("update site via api: %v", err) + } + + if err := d.BootstrapFromConfig(ctx, cfg); err != nil { + t.Fatalf("second bootstrap: %v", err) + } + + v, err := d.GetValidator(ctx, "validator-1") + if err != nil { + t.Fatalf("get validator: %v", err) + } + if v.OSPortID != "port-changed-via-api" { + t.Fatalf("expected YAML to be ignored on non-empty table, got os_port_id=%q", v.OSPortID) + } + sites, err := d.ListSites(ctx) + if err != nil { + t.Fatalf("list sites: %v", err) + } + if len(sites) != 1 || sites[0].SiteID != "site-changed-via-api" { + t.Fatalf("expected YAML to be ignored on non-empty sites table, got %+v", sites) + } +} + +func TestSubmitIPsMixedBatch(t *testing.T) { + d, ctx := newTestDB(t) + + // Seed: 1.1.1.1 done, 2.2.2.2 already queued, 3.3.3.3 mid-check. + if err := d.SeedQueue(ctx, []string{"1.1.1.1", "2.2.2.2", "3.3.3.3"}); err != nil { + t.Fatalf("seed queue: %v", err) + } + ip1, err := d.GetIPByAddress(ctx, "1.1.1.1") + if err != nil { + t.Fatalf("get ip1: %v", err) + } + if err := d.FinishIP(ctx, ip1.ID, ResultPass); err != nil { + t.Fatalf("finish ip1: %v", err) + } + if err := d.AdminCreateValidator(ctx, "validator-1", "port-1"); err != nil { + t.Fatalf("create validator: %v", err) + } + claimed, err := d.ClaimNextQueued(ctx, "validator-1", time.Minute) + if err != nil || claimed == nil { + t.Fatalf("claim next queued: item=%+v err=%v", claimed, err) + } + if claimed.IPAddress != "2.2.2.2" { + t.Fatalf("expected to claim 2.2.2.2 (lowest sequence still queued), got %s", claimed.IPAddress) + } + // 2.2.2.2 is now assigning_fip (mid-check); 3.3.3.3 is still queued. + + result, err := d.SubmitIPs(ctx, []string{"4.4.4.4", "1.1.1.1", "2.2.2.2", "3.3.3.3"}) + if err != nil { + t.Fatalf("submit ips: %v", err) + } + if len(result.Added) != 1 || result.Added[0] != "4.4.4.4" { + t.Fatalf("expected 4.4.4.4 added, got %+v", result.Added) + } + if len(result.Requeued) != 1 || result.Requeued[0] != "1.1.1.1" { + t.Fatalf("expected 1.1.1.1 requeued (was done), got %+v", result.Requeued) + } + if len(result.Reordered) != 1 || result.Reordered[0] != "3.3.3.3" { + t.Fatalf("expected 3.3.3.3 reordered (was queued), got %+v", result.Reordered) + } + if len(result.SkippedInProgress) != 1 || result.SkippedInProgress[0] != "2.2.2.2" { + t.Fatalf("expected 2.2.2.2 skipped (mid-check), got %+v", result.SkippedInProgress) + } + + // 2.2.2.2 must be untouched: still assigning_fip, same owner. + ip2, err := d.GetIPByAddress(ctx, "2.2.2.2") + if err != nil { + t.Fatalf("get ip2: %v", err) + } + if ip2.State != IPAssigningFIP || ip2.OwnerValidatorID == nil || *ip2.OwnerValidatorID != "validator-1" { + t.Fatalf("expected 2.2.2.2 untouched mid-check, got %+v", ip2) + } + + // 1.1.1.1 must be freshly queued again with a bumped attempt number and + // cleared result. + ip1, err = d.GetIPByAddress(ctx, "1.1.1.1") + if err != nil { + t.Fatalf("get ip1 after requeue: %v", err) + } + if ip1.State != IPQueued || ip1.AttemptNumber != 2 || ip1.OverallResult != "" { + t.Fatalf("expected 1.1.1.1 requeued fresh, got %+v", ip1) + } + + // Queue processing order should follow the submitted list order for the + // touched/new addresses: 4.4.4.4, 1.1.1.1, 3.3.3.3 (2.2.2.2 excluded, + // already claimed by validator-1, which is now busy — use a second + // idle validator to observe what's claimed next). + if err := d.AdminCreateValidator(ctx, "validator-2", "port-2"); err != nil { + t.Fatalf("create validator-2: %v", err) + } + claimed, err = d.ClaimNextQueued(ctx, "validator-2", time.Minute) + if err != nil { + t.Fatalf("claim after submit: %v", err) + } + if claimed == nil || claimed.IPAddress != "4.4.4.4" { + t.Fatalf("expected 4.4.4.4 to be claimed first per batch order, got %+v", claimed) + } +} + +func TestSubmitIPsEmptyListRejected(t *testing.T) { + d, ctx := newTestDB(t) + if _, err := d.SubmitIPs(ctx, nil); !errors.Is(err, ErrValidation) { + t.Fatalf("expected ErrValidation for empty address list, got %v", err) + } +} + +func TestCancelIP(t *testing.T) { + d, ctx := newTestDB(t) + + if err := d.SeedQueue(ctx, []string{"1.2.3.4"}); err != nil { + t.Fatalf("seed queue: %v", err) + } + ip, err := d.GetIPByAddress(ctx, "1.2.3.4") + if err != nil { + t.Fatalf("get ip: %v", err) + } + + // Cancel straight from queued. + if err := d.CancelIP(ctx, ip.ID); err != nil { + t.Fatalf("cancel queued ip: %v", err) + } + ip, _ = d.GetIP(ctx, ip.ID) + if ip.State != IPFailed || ip.OverallResult != ResultCancelled { + t.Fatalf("expected failed/cancelled, got state=%s result=%s", ip.State, ip.OverallResult) + } + + // Cancelling an already-terminal IP is rejected. + if err := d.CancelIP(ctx, ip.ID); !errors.Is(err, ErrInvalidState) { + t.Fatalf("expected ErrInvalidState cancelling an already-finished ip, got %v", err) + } + + // Cancel mid-check: claim, associate, then cancel — owner/lease/fip + // should be cleared by CancelIP itself (disassociation is the + // orchestrator's job, not this method's). + if err := d.SeedQueue(ctx, []string{"5.6.7.8"}); err != nil { + t.Fatalf("seed queue 2: %v", err) + } + if err := d.AdminCreateValidator(ctx, "validator-1", "port-1"); err != nil { + t.Fatalf("create validator: %v", err) + } + claimed, err := d.ClaimNextQueued(ctx, "validator-1", time.Minute) + if err != nil || claimed == nil { + t.Fatalf("claim: item=%+v err=%v", claimed, err) + } + if err := d.CancelIP(ctx, claimed.ID); err != nil { + t.Fatalf("cancel mid-check ip: %v", err) + } + got, _ := d.GetIP(ctx, claimed.ID) + if got.State != IPFailed || got.OverallResult != ResultCancelled || got.OwnerValidatorID != nil { + t.Fatalf("expected cancelled + owner cleared, got %+v", got) + } +} diff --git a/internal/db/queries_ipqueue.go b/internal/db/queries_ipqueue.go index 9567722..ed8d5a6 100644 --- a/internal/db/queries_ipqueue.go +++ b/internal/db/queries_ipqueue.go @@ -222,6 +222,119 @@ func (d *DB) RequeueOrFail(ctx context.Context, ipID int64, validatorID string, return tx.Commit() } +// SubmitIPs is the single admin entry point for both "add new addresses to +// the queue" and "force a re-check of an already-finished address" — the +// same list can freely mix both. Addresses are processed in one +// transaction, in the order given: +// +// - unknown address: inserted as a new queued row. +// - address currently done/failed: reset to queued (new attempt, +// retry_count cleared — this is a deliberate admin-triggered restart, +// not a system retry). +// - address currently queued (not yet claimed): left in state=queued, +// only its sequence is updated. +// - address currently mid-check (assigning_fip / awaiting_self_check / +// checking / aggregating): left untouched entirely — never start a +// second concurrent check for the same address. +// +// Every touched/inserted address (new, requeued, or merely reordered) gets +// a sequence assigned in list order, continuing after the current max +// sequence, so a batch's relative order is preserved and, critically, +// resubmitting the same list later reproduces the same relative order. +func (d *DB) SubmitIPs(ctx context.Context, addresses []string) (SubmitIPsResult, error) { + var result SubmitIPsResult + if len(addresses) == 0 { + return result, fmt.Errorf("addresses must not be empty: %w", ErrValidation) + } + + tx, err := d.BeginTx(ctx, nil) + if err != nil { + return result, err + } + defer tx.Rollback() + + var base int + if err := tx.QueryRowContext(ctx, `SELECT COALESCE(MAX(sequence), -1) + 1 FROM ip_queue`).Scan(&base); err != nil { + return result, err + } + + now := timeToDB(Now()) + for i, addr := range addresses { + seq := base + i + + var state string + err := tx.QueryRowContext(ctx, `SELECT state FROM ip_queue WHERE ip_address=?`, addr).Scan(&state) + switch { + case err == sql.ErrNoRows: + 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 { + return result, fmt.Errorf("insert %s: %w", addr, err) + } + result.Added = append(result.Added, addr) + + case err != nil: + return result, err + + case state == IPDone || state == IPFailed: + 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, + site1_complete=0, site2_complete=0, site3_complete=0, overall_result='', + assigned_at=NULL, aggregated_at=NULL, fip_released_at=NULL, updated_at=? + WHERE ip_address=? + `, IPQueued, seq, now, addr); err != nil { + return result, fmt.Errorf("requeue %s: %w", addr, err) + } + result.Requeued = append(result.Requeued, addr) + + case state == IPQueued: + if _, err := tx.ExecContext(ctx, ` + UPDATE ip_queue SET sequence=?, updated_at=? WHERE ip_address=? + `, seq, now, addr); err != nil { + return result, fmt.Errorf("reorder %s: %w", addr, err) + } + result.Reordered = append(result.Reordered, addr) + + default: + // Actively being processed (assigning_fip / awaiting_self_check + // / checking / aggregating) — leave it alone, don't duplicate. + result.SkippedInProgress = append(result.SkippedInProgress, addr) + } + } + + if err := tx.Commit(); err != nil { + return result, err + } + return result, nil +} + +// CancelIP force-stops a non-terminal IP: marks it failed with +// overall_result=cancelled. It does not disassociate the floating IP or +// free the owning validator — that requires the OpenStack client, so it's +// the caller's (orchestrator's) job to do that before/after calling this. +// Returns ErrInvalidState if the IP is already done/failed (including the +// race where aggregation finishes between the caller's read and this call — +// closed by the single-connection transactional UPDATE below). +func (d *DB) CancelIP(ctx context.Context, ipID int64) error { + now := timeToDB(Now()) + res, err := d.ExecContext(ctx, ` + UPDATE ip_queue SET + state=?, overall_result=?, aggregated_at=?, owner_validator_id=NULL, fip_id='', + lease_expires_at=NULL, updated_at=? + WHERE id=? AND state NOT IN (?, ?) + `, IPFailed, ResultCancelled, now, now, ipID, IPDone, IPFailed) + if err != nil { + return fmt.Errorf("cancel ip: %w", err) + } + if n, _ := res.RowsAffected(); n == 0 { + return fmt.Errorf("ip_id %d already finished: %w", ipID, ErrInvalidState) + } + return nil +} + func (d *DB) SetEgressComplete(ctx context.Context, ipID int64) error { _, err := d.ExecContext(ctx, `UPDATE ip_queue SET egress_complete=1, updated_at=? WHERE id=?`, timeToDB(Now()), ipID) return err diff --git a/internal/db/queries_sites.go b/internal/db/queries_sites.go new file mode 100644 index 0000000..0d9c74f --- /dev/null +++ b/internal/db/queries_sites.go @@ -0,0 +1,97 @@ +package db + +import ( + "context" + "database/sql" + "fmt" +) + +// ListSites returns all configured prober sites, ordered by slot index. +func (d *DB) ListSites(ctx context.Context) ([]Site, error) { + rows, err := d.QueryContext(ctx, ` + SELECT idx, site_id, created_at, updated_at FROM sites ORDER BY idx + `) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []Site + for rows.Next() { + var s Site + var createdAt, updatedAt string + if err := rows.Scan(&s.Index, &s.SiteID, &createdAt, &updatedAt); err != nil { + return nil, err + } + var err error + if s.CreatedAt, err = dbToTime(createdAt); err != nil { + return nil, err + } + if s.UpdatedAt, err = dbToTime(updatedAt); err != nil { + return nil, err + } + out = append(out, s) + } + return out, rows.Err() +} + +// GetSiteIndex resolves a configured site_id to its 1/2/3 slot index. It +// returns (0, nil) — not an error — if no site with that ID is configured, +// matching the "0 means unconfigured" convention used throughout the +// prober-facing handlers. +func (d *DB) GetSiteIndex(ctx context.Context, siteID string) (int, error) { + var idx int + err := d.QueryRowContext(ctx, `SELECT idx FROM sites WHERE site_id=?`, siteID).Scan(&idx) + if err == sql.ErrNoRows { + return 0, nil + } + if err != nil { + return 0, err + } + return idx, nil +} + +// UpsertSite assigns (or renames) the site occupying the given slot. idx +// must be 1, 2, or 3 — the schema hard-caps the number of prober slots at +// three (see ip_queue.site{1,2,3}_complete). site_id must be unique across +// slots. +func (d *DB) UpsertSite(ctx context.Context, idx int, siteID string) error { + if idx < 1 || idx > 3 { + return fmt.Errorf("site index must be 1, 2, or 3, got %d: %w", idx, ErrValidation) + } + if siteID == "" { + return fmt.Errorf("site_id must not be empty: %w", ErrValidation) + } + + var existingIdx int + err := d.QueryRowContext(ctx, `SELECT idx FROM sites WHERE site_id=? AND idx!=?`, siteID, idx).Scan(&existingIdx) + if err != nil && err != sql.ErrNoRows { + return err + } + if err == nil { + return fmt.Errorf("site_id %q already assigned to slot %d: %w", siteID, existingIdx, ErrConflict) + } + + now := timeToDB(Now()) + _, err = d.ExecContext(ctx, ` + INSERT INTO sites (idx, site_id, created_at, updated_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(idx) DO UPDATE SET site_id=excluded.site_id, updated_at=excluded.updated_at + `, idx, siteID, now, now) + if err != nil { + return fmt.Errorf("upsert site: %w", err) + } + return nil +} + +// DeleteSite frees the given slot. +func (d *DB) DeleteSite(ctx context.Context, idx int) error { + res, err := d.ExecContext(ctx, `DELETE FROM sites WHERE idx=?`, idx) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return fmt.Errorf("site slot %d: %w", idx, ErrNotFound) + } + return nil +} diff --git a/internal/db/queries_targetgroups.go b/internal/db/queries_targetgroups.go new file mode 100644 index 0000000..f40e489 --- /dev/null +++ b/internal/db/queries_targetgroups.go @@ -0,0 +1,109 @@ +package db + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" +) + +// ListTargetGroups returns all configured target groups, ordered by name. +func (d *DB) ListTargetGroups(ctx context.Context) ([]TargetGroup, error) { + rows, err := d.QueryContext(ctx, ` + SELECT group_name, targets, created_at, updated_at FROM target_groups ORDER BY group_name + `) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []TargetGroup + for rows.Next() { + g, err := scanTargetGroup(rows) + if err != nil { + return nil, err + } + out = append(out, *g) + } + return out, rows.Err() +} + +// GetTargetGroup returns a single target group by name. +func (d *DB) GetTargetGroup(ctx context.Context, name string) (*TargetGroup, error) { + row := d.QueryRowContext(ctx, ` + SELECT group_name, targets, created_at, updated_at FROM target_groups WHERE group_name=? + `, name) + g, err := scanTargetGroup(row) + if err == sql.ErrNoRows { + return nil, fmt.Errorf("target group %q: %w", name, ErrNotFound) + } + return g, err +} + +// UpsertTargetGroup creates or replaces a target group's target list. +func (d *DB) UpsertTargetGroup(ctx context.Context, name string, targets []string) error { + if name == "" { + return fmt.Errorf("group name must not be empty: %w", ErrValidation) + } + if len(targets) == 0 { + return fmt.Errorf("target group %q must have at least one target: %w", name, ErrValidation) + } + payload, err := json.Marshal(targets) + if err != nil { + return err + } + now := timeToDB(Now()) + _, err = d.ExecContext(ctx, ` + INSERT INTO target_groups (group_name, targets, created_at, updated_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(group_name) DO UPDATE SET targets=excluded.targets, updated_at=excluded.updated_at + `, name, string(payload), now, now) + if err != nil { + return fmt.Errorf("upsert target group: %w", err) + } + return nil +} + +// DeleteTargetGroup removes a target group, refusing if any check type +// still references it. +func (d *DB) DeleteTargetGroup(ctx context.Context, name string) error { + checkTypes, err := d.ListCheckTypes(ctx) + if err != nil { + return err + } + for _, ct := range checkTypes { + for _, g := range ct.TargetGroups { + if g == name { + return fmt.Errorf("target group %q is used by check type %q: %w", name, ct.Name, ErrInUse) + } + } + } + + res, err := d.ExecContext(ctx, `DELETE FROM target_groups WHERE group_name=?`, name) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return fmt.Errorf("target group %q: %w", name, ErrNotFound) + } + return nil +} + +func scanTargetGroup(row rowScanner) (*TargetGroup, error) { + var g TargetGroup + var targetsJSON, createdAt, updatedAt string + if err := row.Scan(&g.Name, &targetsJSON, &createdAt, &updatedAt); err != nil { + return nil, err + } + if err := json.Unmarshal([]byte(targetsJSON), &g.Targets); err != nil { + return nil, fmt.Errorf("decode targets for group %q: %w", g.Name, err) + } + var err error + if g.CreatedAt, err = dbToTime(createdAt); err != nil { + return nil, err + } + if g.UpdatedAt, err = dbToTime(updatedAt); err != nil { + return nil, err + } + return &g, nil +} diff --git a/internal/db/queries_validators.go b/internal/db/queries_validators.go index ea494e6..8b94a29 100644 --- a/internal/db/queries_validators.go +++ b/internal/db/queries_validators.go @@ -148,6 +148,79 @@ func (d *DB) FreeValidator(ctx context.Context, validatorID string) error { return err } +// AdminCreateValidator registers a brand-new validator via the admin API. +// Unlike RegisterValidator (used by the agent's self-registration call), +// this refuses to upsert over an existing row. +func (d *DB) AdminCreateValidator(ctx context.Context, validatorID, osPortID string) error { + var exists int + err := d.QueryRowContext(ctx, `SELECT 1 FROM validators WHERE validator_id=?`, validatorID).Scan(&exists) + if err != nil && err != sql.ErrNoRows { + return err + } + if err == nil { + return fmt.Errorf("validator %q: %w", validatorID, ErrConflict) + } + + now := timeToDB(Now()) + _, err = d.ExecContext(ctx, ` + INSERT INTO validators (validator_id, hostname, os_port_id, agent_version, state, created_at, updated_at) + VALUES (?, '', ?, '', ?, ?, ?) + `, validatorID, osPortID, ValidatorIdle, now, now) + if err != nil { + return fmt.Errorf("create validator: %w", err) + } + return nil +} + +// AdminUpdateValidatorPort updates an existing validator's Neutron port ID. +func (d *DB) AdminUpdateValidatorPort(ctx context.Context, validatorID, osPortID string) error { + res, err := d.ExecContext(ctx, ` + UPDATE validators SET os_port_id=?, updated_at=? WHERE validator_id=? + `, osPortID, timeToDB(Now()), validatorID) + if err != nil { + return fmt.Errorf("update validator port: %w", err) + } + if n, _ := res.RowsAffected(); n == 0 { + return fmt.Errorf("validator %q: %w", validatorID, ErrNotFound) + } + return nil +} + +// DeleteValidator removes a validator, refusing if it currently owns an IP. +// ip_queue.owner_validator_id is a permanent historical record (set once an +// IP is claimed, never cleared on completion — see RequeueOrFail/ +// ReleaseFIP), so any validator that has ever processed an IP would +// otherwise violate the FK constraint on delete; those historical +// references are cleared in the same transaction once we've confirmed the +// validator isn't currently busy. +func (d *DB) DeleteValidator(ctx context.Context, validatorID string) error { + tx, err := d.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + + var currentIPID sql.NullInt64 + err = tx.QueryRowContext(ctx, `SELECT current_ip_id FROM validators WHERE validator_id=?`, validatorID).Scan(¤tIPID) + if err == sql.ErrNoRows { + return fmt.Errorf("validator %q: %w", validatorID, ErrNotFound) + } + if err != nil { + return err + } + if currentIPID.Valid { + return fmt.Errorf("validator %q owns ip_id %d: %w", validatorID, currentIPID.Int64, ErrBusy) + } + + if _, err := tx.ExecContext(ctx, `UPDATE ip_queue SET owner_validator_id=NULL WHERE owner_validator_id=?`, validatorID); err != nil { + return fmt.Errorf("clear historical ip_queue references: %w", err) + } + if _, err := tx.ExecContext(ctx, `DELETE FROM validators WHERE validator_id=?`, validatorID); err != nil { + return fmt.Errorf("delete validator: %w", err) + } + return tx.Commit() +} + type rowScanner interface { Scan(dest ...interface{}) error } diff --git a/internal/httpapi/dto_admin.go b/internal/httpapi/dto_admin.go new file mode 100644 index 0000000..2adbd88 --- /dev/null +++ b/internal/httpapi/dto_admin.go @@ -0,0 +1,62 @@ +package httpapi + +// DTOs for the admin queue-management and dynamic-config endpoints +// (/api/v1/admin/ips, /api/v1/admin/config/*). Unlike the older read-only +// admin endpoints (which marshal internal/db model structs directly, in +// PascalCase), these use explicit snake_case JSON tags to match the +// agent/prober DTO convention. + +type submitIPsRequest struct { + Addresses []string `json:"addresses"` +} + +type submitIPsResponse struct { + Added []string `json:"added"` + Requeued []string `json:"requeued"` + Reordered []string `json:"reordered"` + SkippedInProgress []string `json:"skipped_in_progress"` +} + +type validatorDTO struct { + ValidatorID string `json:"validator_id"` + OSPortID string `json:"os_port_id"` + State string `json:"state"` +} + +type createValidatorRequest struct { + ValidatorID string `json:"validator_id"` + OSPortID string `json:"os_port_id"` +} + +type updateValidatorRequest struct { + OSPortID string `json:"os_port_id"` +} + +type siteDTO struct { + Index int `json:"index"` + SiteID string `json:"site_id"` +} + +type putSiteRequest struct { + SiteID string `json:"site_id"` +} + +type targetGroupDTO struct { + Name string `json:"name"` + Targets []string `json:"targets"` +} + +type putTargetGroupRequest struct { + Targets []string `json:"targets"` +} + +type checkTypeDTO struct { + Name string `json:"name"` + Enabled bool `json:"enabled"` + Targets []string `json:"targets"` +} + +type putCheckTypeRequest struct { + Enabled bool `json:"enabled"` + Targets []string `json:"targets"` +} diff --git a/internal/httpapi/handlers_admin.go b/internal/httpapi/handlers_admin.go index cbba00e..167cce0 100644 --- a/internal/httpapi/handlers_admin.go +++ b/internal/httpapi/handlers_admin.go @@ -73,3 +73,49 @@ func (s *Server) handleAdminValidators(w http.ResponseWriter, r *http.Request) { } writeJSON(w, http.StatusOK, validators) } + +// handleAdminSubmitIPs is the single entry point for both adding new +// addresses to the queue and forcing a re-check of already-finished ones — +// see db.SubmitIPs for the exact per-address rules. It's a direct DB call +// (no OpenStack interaction is needed to merely queue work), matching the +// existing admin handlers above which also bypass the orchestrator for +// reads. +func (s *Server) handleAdminSubmitIPs(w http.ResponseWriter, r *http.Request) { + var req submitIPsRequest + if err := readJSON(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "invalid body: "+err.Error()) + return + } + result, err := s.DB.SubmitIPs(r.Context(), req.Addresses) + if err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusOK, submitIPsResponse{ + 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. +func (s *Server) handleAdminCancelIP(w http.ResponseWriter, r *http.Request) { + address := r.PathValue("ip") + if err := s.Orch.ForceCancel(r.Context(), address); err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusOK, okResponse{OK: true}) +} + +// emptyIfNil turns a nil slice into an empty one so these fields always +// marshal as `[]` rather than `null`. +func emptyIfNil(s []string) []string { + if s == nil { + return []string{} + } + return s +} diff --git a/internal/httpapi/handlers_config.go b/internal/httpapi/handlers_config.go new file mode 100644 index 0000000..4352d73 --- /dev/null +++ b/internal/httpapi/handlers_config.go @@ -0,0 +1,183 @@ +package httpapi + +import ( + "net/http" + "strconv" +) + +// --- validators --- + +func (s *Server) handleConfigListValidators(w http.ResponseWriter, r *http.Request) { + validators, err := s.DB.ListValidators(r.Context()) + if err != nil { + writeDBError(w, err) + return + } + out := make([]validatorDTO, len(validators)) + for i, v := range validators { + out[i] = validatorDTO{ValidatorID: v.ValidatorID, OSPortID: v.OSPortID, State: v.State} + } + writeJSON(w, http.StatusOK, out) +} + +func (s *Server) handleConfigCreateValidator(w http.ResponseWriter, r *http.Request) { + var req createValidatorRequest + if err := readJSON(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "invalid body: "+err.Error()) + return + } + if req.ValidatorID == "" { + writeError(w, http.StatusBadRequest, "validator_id must not be empty") + return + } + if err := s.DB.AdminCreateValidator(r.Context(), req.ValidatorID, req.OSPortID); err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusCreated, validatorDTO{ValidatorID: req.ValidatorID, OSPortID: req.OSPortID, State: "idle"}) +} + +func (s *Server) handleConfigUpdateValidator(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + var req updateValidatorRequest + if err := readJSON(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "invalid body: "+err.Error()) + return + } + if err := s.DB.AdminUpdateValidatorPort(r.Context(), id, req.OSPortID); err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusOK, okResponse{OK: true}) +} + +func (s *Server) handleConfigDeleteValidator(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + if err := s.DB.DeleteValidator(r.Context(), id); err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusOK, okResponse{OK: true}) +} + +// --- sites --- + +func (s *Server) handleConfigListSites(w http.ResponseWriter, r *http.Request) { + sites, err := s.DB.ListSites(r.Context()) + if err != nil { + writeDBError(w, err) + return + } + out := make([]siteDTO, len(sites)) + for i, site := range sites { + out[i] = siteDTO{Index: site.Index, SiteID: site.SiteID} + } + writeJSON(w, http.StatusOK, out) +} + +func (s *Server) handleConfigPutSite(w http.ResponseWriter, r *http.Request) { + idx, err := strconv.Atoi(r.PathValue("index")) + if err != nil { + writeError(w, http.StatusBadRequest, "index must be an integer") + return + } + var req putSiteRequest + if err := readJSON(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "invalid body: "+err.Error()) + return + } + if err := s.DB.UpsertSite(r.Context(), idx, req.SiteID); err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusOK, siteDTO{Index: idx, SiteID: req.SiteID}) +} + +func (s *Server) handleConfigDeleteSite(w http.ResponseWriter, r *http.Request) { + idx, err := strconv.Atoi(r.PathValue("index")) + if err != nil { + writeError(w, http.StatusBadRequest, "index must be an integer") + return + } + if err := s.DB.DeleteSite(r.Context(), idx); err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusOK, okResponse{OK: true}) +} + +// --- target groups --- + +func (s *Server) handleConfigListTargets(w http.ResponseWriter, r *http.Request) { + groups, err := s.DB.ListTargetGroups(r.Context()) + if err != nil { + writeDBError(w, err) + return + } + out := make([]targetGroupDTO, len(groups)) + for i, g := range groups { + out[i] = targetGroupDTO{Name: g.Name, Targets: g.Targets} + } + writeJSON(w, http.StatusOK, out) +} + +func (s *Server) handleConfigPutTargetGroup(w http.ResponseWriter, r *http.Request) { + name := r.PathValue("group") + var req putTargetGroupRequest + if err := readJSON(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "invalid body: "+err.Error()) + return + } + if err := s.DB.UpsertTargetGroup(r.Context(), name, req.Targets); err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusOK, targetGroupDTO{Name: name, Targets: req.Targets}) +} + +func (s *Server) handleConfigDeleteTargetGroup(w http.ResponseWriter, r *http.Request) { + name := r.PathValue("group") + if err := s.DB.DeleteTargetGroup(r.Context(), name); err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusOK, okResponse{OK: true}) +} + +// --- check types --- + +func (s *Server) handleConfigListCheckTypes(w http.ResponseWriter, r *http.Request) { + checkTypes, err := s.DB.ListCheckTypes(r.Context()) + if err != nil { + writeDBError(w, err) + return + } + out := make([]checkTypeDTO, len(checkTypes)) + for i, ct := range checkTypes { + out[i] = checkTypeDTO{Name: ct.Name, Enabled: ct.Enabled, Targets: ct.TargetGroups} + } + writeJSON(w, http.StatusOK, out) +} + +func (s *Server) handleConfigPutCheckType(w http.ResponseWriter, r *http.Request) { + name := r.PathValue("name") + var req putCheckTypeRequest + if err := readJSON(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "invalid body: "+err.Error()) + return + } + if err := s.DB.UpsertCheckType(r.Context(), name, req.Enabled, req.Targets); err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusOK, checkTypeDTO{Name: name, Enabled: req.Enabled, Targets: req.Targets}) +} + +func (s *Server) handleConfigDeleteCheckType(w http.ResponseWriter, r *http.Request) { + name := r.PathValue("name") + if err := s.DB.DeleteCheckType(r.Context(), name); err != nil { + writeDBError(w, err) + return + } + writeJSON(w, http.StatusOK, okResponse{OK: true}) +} diff --git a/internal/httpapi/handlers_config_test.go b/internal/httpapi/handlers_config_test.go new file mode 100644 index 0000000..ccc83b1 --- /dev/null +++ b/internal/httpapi/handlers_config_test.go @@ -0,0 +1,291 @@ +package httpapi + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" + + "cloudipvalidator/internal/config" + "cloudipvalidator/internal/db" + "cloudipvalidator/internal/openstack" + "cloudipvalidator/internal/orchestrator" +) + +// newConfigTestHarness sets up a control-api stack with an *empty* +// control-api.yaml (no validators/sites/targets/check_types) — everything +// in this test is created purely through the admin API, to prove the +// dynamic-config path works with no YAML at all. +func newConfigTestHarness(t *testing.T) (*fakeClient, *db.DB, *orchestrator.Orchestrator, *openstack.MockClient) { + t.Helper() + ctx := context.Background() + d, err := db.Open(ctx, filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatalf("open db: %v", err) + } + t.Cleanup(func() { d.Close() }) + + mock := openstack.NewMockClient() + + cfg := &config.ControlAPI{ + Orchestrator: config.OrchestratorConfig{ + PollIntervalSeconds: 1, SelfCheckTimeoutSeconds: 10, MaxSelfCheckRetries: 3, + CheckingWindowSeconds: 120, MaxRetries: 3, LeaseTTLSeconds: 180, HeartbeatTimeoutSeconds: 30, + }, + Aggregation: config.AggregationConfig{MissingCountsAsFail: true}, + Inbound: config.InboundConfig{Ports: []int{22, 80}, ICMP: true}, + } + if err := d.BootstrapFromConfig(ctx, cfg); err != nil { + t.Fatalf("bootstrap from (empty) config: %v", err) + } + + log := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + orch := orchestrator.New(d, mock, cfg, log) + srv := New(d, orch, log) + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(ts.Close) + + return &fakeClient{t: t, base: ts.URL, client: ts.Client()}, d, orch, mock +} + +// TestConfigManagedEntirelyViaAPI proves an operator can stand up a working +// validator/site/target-group/check-type configuration using only the +// admin API — no YAML at all — and that an IP submitted afterwards passes +// through the full checking cycle to `done`. +func TestConfigManagedEntirelyViaAPI(t *testing.T) { + fc, d, orch, mock := newConfigTestHarness(t) + ctx := context.Background() + + mock.Seed("fip-1", "9.9.9.9", "svc-project") + + resp, body := fc.do(http.MethodPost, "/api/v1/admin/config/validators", createValidatorRequest{ + ValidatorID: "validator-1", OSPortID: "port-1", + }) + if resp.StatusCode != http.StatusCreated { + t.Fatalf("create validator: status=%d body=%s", resp.StatusCode, body) + } + + resp, body = fc.do(http.MethodPut, "/api/v1/admin/config/targets/web", putTargetGroupRequest{ + Targets: []string{"https://example.test"}, + }) + if resp.StatusCode != http.StatusOK { + t.Fatalf("put target group: status=%d body=%s", resp.StatusCode, body) + } + + resp, body = fc.do(http.MethodPut, "/api/v1/admin/config/check-types/https", putCheckTypeRequest{ + Enabled: true, Targets: []string{"web"}, + }) + if resp.StatusCode != http.StatusOK { + t.Fatalf("put check type: status=%d body=%s", resp.StatusCode, body) + } + + // No sites configured -> inbound checks stay opt-out; egress alone + // should be enough to reach `done`. + + resp, body = fc.do(http.MethodPost, "/api/v1/admin/ips", submitIPsRequest{Addresses: []string{"9.9.9.9"}}) + if resp.StatusCode != http.StatusOK { + t.Fatalf("submit ips: status=%d body=%s", resp.StatusCode, body) + } + var submitResp submitIPsResponse + if err := json.Unmarshal(body, &submitResp); err != nil { + t.Fatalf("unmarshal submit response: %v", err) + } + if len(submitResp.Added) != 1 || submitResp.Added[0] != "9.9.9.9" { + t.Fatalf("expected 9.9.9.9 added, got %+v", submitResp) + } + + resp, body = fc.do(http.MethodPost, "/api/v1/agents/register", registerAgentRequest{ValidatorID: "validator-1"}) + if resp.StatusCode != http.StatusOK { + t.Fatalf("register agent: status=%d body=%s", resp.StatusCode, body) + } + + orch.Tick(ctx) + + resp, body = fc.do(http.MethodGet, "/api/v1/agents/validator-1/assignment", nil) + if resp.StatusCode != http.StatusOK { + t.Fatalf("assignment: status=%d body=%s", resp.StatusCode, body) + } + var assignment assignmentResponse + if err := json.Unmarshal(body, &assignment); err != nil { + t.Fatalf("unmarshal assignment: %v", err) + } + if len(assignment.CheckConfig) != 1 || assignment.CheckConfig[0].Type != "https" { + t.Fatalf("expected the API-created https check type in the assignment, got %+v", assignment.CheckConfig) + } + + resp, body = fc.do(http.MethodPost, "/api/v1/agents/validator-1/self-check", selfCheckRequest{ + IPID: assignment.IPID, DetectedEgress: "9.9.9.9", Success: true, + }) + if resp.StatusCode != http.StatusOK { + t.Fatalf("self-check: status=%d body=%s", resp.StatusCode, body) + } + resp, body = 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), + }}, + }) + if resp.StatusCode != http.StatusOK { + t.Fatalf("results: status=%d body=%s", resp.StatusCode, body) + } + resp, body = fc.do(http.MethodPost, "/api/v1/agents/validator-1/complete", agentCompleteRequest{IPID: assignment.IPID}) + if resp.StatusCode != http.StatusOK { + t.Fatalf("complete: status=%d body=%s", resp.StatusCode, body) + } + + orch.Tick(ctx) + + item, err := d.GetIPByAddress(ctx, "9.9.9.9") + if err != nil { + t.Fatalf("get ip: %v", err) + } + if item.State != db.IPDone || item.OverallResult != db.ResultPass { + t.Fatalf("expected done/pass, got state=%s result=%s", item.State, item.OverallResult) + } +} + +// TestSubmitIPsForcesRecheckOfFinishedAddress proves that resubmitting an +// address that already reached `done` starts a brand-new checking cycle +// rather than being ignored. +func TestSubmitIPsForcesRecheckOfFinishedAddress(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"}}) + + 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/agents/register", registerAgentRequest{ValidatorID: "validator-1"}) + 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 || item.AttemptNumber != 1 { + t.Fatalf("expected done after first cycle with attempt_number=1, got state=%s attempt=%d", item.State, item.AttemptNumber) + } + + // Force a recheck of the same, already-finished address. The mock FIP + // was disassociated at the end of the first cycle; associateFIP will + // simply re-associate it during the second cycle. + resp, body := fc.do(http.MethodPost, "/api/v1/admin/ips", submitIPsRequest{Addresses: []string{"9.9.9.9"}}) + if resp.StatusCode != http.StatusOK { + t.Fatalf("submit ips (recheck): status=%d body=%s", resp.StatusCode, body) + } + var submitResp submitIPsResponse + if err := json.Unmarshal(body, &submitResp); err != nil { + t.Fatalf("unmarshal submit response: %v", err) + } + if len(submitResp.Requeued) != 1 || submitResp.Requeued[0] != "9.9.9.9" { + t.Fatalf("expected 9.9.9.9 to be requeued, got %+v", submitResp) + } + + item, err = d.GetIPByAddress(ctx, "9.9.9.9") + if err != nil { + t.Fatalf("get ip after resubmit: %v", err) + } + if item.State != db.IPQueued || item.AttemptNumber != 2 || item.OverallResult != "" { + t.Fatalf("expected freshly queued with attempt_number=2, got %+v", item) + } + + runOneCycle() + item, err = d.GetIPByAddress(ctx, "9.9.9.9") + if err != nil { + t.Fatalf("get ip after second cycle: %v", err) + } + if item.State != db.IPDone || item.OverallResult != db.ResultPass || item.AttemptNumber != 2 { + t.Fatalf("expected done/pass on second attempt, got %+v", item) + } +} + +// TestForceCancelMidCheck proves POST /admin/ips/{ip}/cancel stops an +// in-progress check, disassociates its floating IP, and frees the +// validator, without waiting for the checking window to elapse. +func TestForceCancelMidCheck(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"}) + fc.do(http.MethodPost, "/api/v1/admin/ips", submitIPsRequest{Addresses: []string{"9.9.9.9"}}) + + orch.Tick(ctx) // claim + associate FIP -> awaiting_self_check + + item, err := d.GetIPByAddress(ctx, "9.9.9.9") + if err != nil { + t.Fatalf("get ip: %v", err) + } + if item.State != db.IPAwaitingSelfCheck { + t.Fatalf("expected awaiting_self_check before cancel, got %s", item.State) + } + if fip, _ := mock.GetFloatingIPByAddress(ctx, "9.9.9.9"); fip.PortID == "" { + t.Fatalf("expected fip associated before cancel") + } + + resp, body := fc.do(http.MethodPost, "/api/v1/admin/ips/9.9.9.9/cancel", nil) + if resp.StatusCode != http.StatusOK { + t.Fatalf("cancel: status=%d body=%s", resp.StatusCode, body) + } + + item, err = d.GetIPByAddress(ctx, "9.9.9.9") + if err != nil { + t.Fatalf("get ip after cancel: %v", err) + } + if item.State != db.IPFailed || item.OverallResult != db.ResultCancelled { + t.Fatalf("expected failed/cancelled, got state=%s result=%s", item.State, item.OverallResult) + } + if fip, _ := mock.GetFloatingIPByAddress(ctx, "9.9.9.9"); fip.PortID != "" { + t.Fatalf("expected fip disassociated after cancel, still on port %q", fip.PortID) + } + + v, err := d.GetValidator(ctx, "validator-1") + if err != nil { + t.Fatalf("get validator: %v", err) + } + if v.State != db.ValidatorIdle || v.CurrentIPID != nil { + t.Fatalf("expected validator freed, got state=%s current_ip=%v", v.State, v.CurrentIPID) + } + + // Cancelling again is rejected — nothing left to cancel. + resp, body = fc.do(http.MethodPost, "/api/v1/admin/ips/9.9.9.9/cancel", nil) + if resp.StatusCode != http.StatusConflict { + t.Fatalf("expected 409 cancelling an already-finished ip, status=%d body=%s", resp.StatusCode, body) + } + + // Cancelling an unknown address is a 404. + resp, body = fc.do(http.MethodPost, "/api/v1/admin/ips/1.1.1.1/cancel", nil) + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("expected 404 cancelling unknown ip, status=%d body=%s", resp.StatusCode, body) + } +} diff --git a/internal/httpapi/handlers_prober.go b/internal/httpapi/handlers_prober.go index 577433f..b0febcd 100644 --- a/internal/httpapi/handlers_prober.go +++ b/internal/httpapi/handlers_prober.go @@ -13,7 +13,12 @@ func (s *Server) handleProberRegister(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "invalid body: "+err.Error()) return } - if s.Orch.SiteIndexForID(req.SiteID) == 0 { + idx, err := s.Orch.SiteIndexForID(r.Context(), req.SiteID) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + if idx == 0 { writeError(w, http.StatusBadRequest, "unknown site_id: "+req.SiteID) return } @@ -26,7 +31,12 @@ func (s *Server) handleProberRegister(w http.ResponseWriter, r *http.Request) { // time, since multiple validators run in parallel. func (s *Server) handleProberAssignments(w http.ResponseWriter, r *http.Request) { siteID := r.PathValue("site_id") - if s.Orch.SiteIndexForID(siteID) == 0 { + idx, err := s.Orch.SiteIndexForID(r.Context(), siteID) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + if idx == 0 { writeError(w, http.StatusNotFound, "unknown site_id: "+siteID) return } @@ -47,7 +57,11 @@ func (s *Server) handleProberAssignments(w http.ResponseWriter, r *http.Request) func (s *Server) handleProberResults(w http.ResponseWriter, r *http.Request) { siteID := r.PathValue("site_id") - siteIndex := s.Orch.SiteIndexForID(siteID) + siteIndex, err := s.Orch.SiteIndexForID(r.Context(), siteID) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } if siteIndex == 0 { writeError(w, http.StatusNotFound, "unknown site_id: "+siteID) return diff --git a/internal/httpapi/httpapi_test.go b/internal/httpapi/httpapi_test.go index 9fd18de..bdd509e 100644 --- a/internal/httpapi/httpapi_test.go +++ b/internal/httpapi/httpapi_test.go @@ -78,6 +78,10 @@ func TestEndToEndHTTPFlow(t *testing.T) { Targets: map[string][]string{"web": {"https://example.test"}}, Inbound: config.InboundConfig{Ports: []int{22, 80}, ICMP: true}, } + if err := d.BootstrapFromConfig(ctx, cfg); err != nil { + t.Fatalf("bootstrap from config: %v", err) + } + log := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) orch := orchestrator.New(d, mock, cfg, log) diff --git a/internal/httpapi/routes.go b/internal/httpapi/routes.go index 4ca82fc..b5fadd6 100644 --- a/internal/httpapi/routes.go +++ b/internal/httpapi/routes.go @@ -19,6 +19,25 @@ 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("GET /api/v1/admin/ips/{ip}", s.handleAdminIPDetail) + mux.HandleFunc("POST /api/v1/admin/ips/{ip}/cancel", s.handleAdminCancelIP) mux.HandleFunc("GET /api/v1/admin/validators", s.handleAdminValidators) + + 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) + mux.HandleFunc("DELETE /api/v1/admin/config/validators/{id}", s.handleConfigDeleteValidator) + + mux.HandleFunc("GET /api/v1/admin/config/sites", s.handleConfigListSites) + mux.HandleFunc("PUT /api/v1/admin/config/sites/{index}", s.handleConfigPutSite) + mux.HandleFunc("DELETE /api/v1/admin/config/sites/{index}", s.handleConfigDeleteSite) + + mux.HandleFunc("GET /api/v1/admin/config/targets", s.handleConfigListTargets) + mux.HandleFunc("PUT /api/v1/admin/config/targets/{group}", s.handleConfigPutTargetGroup) + mux.HandleFunc("DELETE /api/v1/admin/config/targets/{group}", s.handleConfigDeleteTargetGroup) + + mux.HandleFunc("GET /api/v1/admin/config/check-types", s.handleConfigListCheckTypes) + mux.HandleFunc("PUT /api/v1/admin/config/check-types/{name}", s.handleConfigPutCheckType) + mux.HandleFunc("DELETE /api/v1/admin/config/check-types/{name}", s.handleConfigDeleteCheckType) } diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index fb43180..18c684c 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -7,6 +7,7 @@ package httpapi import ( "encoding/json" + "errors" "log/slog" "net/http" @@ -56,3 +57,19 @@ func readJSON(r *http.Request, v interface{}) error { dec := json.NewDecoder(r.Body) return dec.Decode(v) } + +// writeDBError maps the typed sentinel errors returned by internal/db's +// admin mutation methods to the appropriate HTTP status, instead of +// defaulting everything to 500 like the older read-only admin handlers do. +func writeDBError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, db.ErrNotFound): + writeError(w, http.StatusNotFound, err.Error()) + case errors.Is(err, db.ErrConflict), errors.Is(err, db.ErrBusy), errors.Is(err, db.ErrInUse), errors.Is(err, db.ErrInvalidState): + writeError(w, http.StatusConflict, err.Error()) + case errors.Is(err, db.ErrValidation): + writeError(w, http.StatusBadRequest, err.Error()) + default: + writeError(w, http.StatusInternalServerError, err.Error()) + } +} diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 0b48ca3..62b48ad 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -9,6 +9,8 @@ package orchestrator import ( "context" + "database/sql" + "errors" "fmt" "log/slog" "time" @@ -20,9 +22,8 @@ import ( // CheckConfig is the check-type/target configuration handed to a // validator-agent once its IP has passed self-check. It mirrors -// config.CheckTypeConfig + config.ControlAPI.Targets, pre-resolved into a -// flat list so the agent doesn't need its own copy of the target-group -// mapping. +// db.ResolvedCheckType, kept as a distinct type so httpapi's DTO layer +// doesn't need to import internal/db just for this shape. type CheckConfig struct { Type string `json:"type"` Targets []string `json:"targets"` @@ -33,31 +34,22 @@ type Orchestrator struct { OS openstack.FloatingIPClient Cfg config.OrchestratorConfig Agg config.AggregationConfig - Checks []CheckConfig - Sites []config.SiteConfig Inbound config.InboundConfig Log *slog.Logger } +// New constructs an Orchestrator. Egress check types/targets and prober +// sites are no longer taken from cfg — they're read from the database on +// every use (see AssignmentForValidator, expectedCheckCount, +// isReadyToAggregate) so admin API changes to them take effect without a +// restart. cfg.Validators/.Sites/.CheckTypes/.Targets/.IPAddresses are only +// consulted once, at process startup, by db.BootstrapFromConfig. func New(d *db.DB, osClient openstack.FloatingIPClient, cfg *config.ControlAPI, log *slog.Logger) *Orchestrator { - var checks []CheckConfig - for _, ct := range cfg.CheckTypes { - if !ct.Enabled { - continue - } - var targets []string - for _, group := range ct.Targets { - targets = append(targets, cfg.Targets[group]...) - } - checks = append(checks, CheckConfig{Type: ct.Name, Targets: targets}) - } return &Orchestrator{ DB: d, OS: osClient, Cfg: cfg.Orchestrator, Agg: cfg.Aggregation, - Checks: checks, - Sites: cfg.Sites, Inbound: cfg.Inbound, Log: log, } @@ -159,7 +151,9 @@ func (o *Orchestrator) SelfCheckResult(ctx context.Context, validatorID string, // AssignmentForValidator returns the check config for a validator's current // IP if it's ready to be worked on (awaiting_self_check or checking), -// or nil if the validator has nothing to do right now. +// or nil if the validator has nothing to do right now. The check config is +// read fresh from the database on every call, so admin changes to +// check_types/targets apply to the very next assignment. func (o *Orchestrator) AssignmentForValidator(ctx context.Context, validatorID string) (*db.IPQueueItem, []CheckConfig, error) { v, err := o.DB.GetValidator(ctx, validatorID) if err != nil { @@ -175,17 +169,21 @@ func (o *Orchestrator) AssignmentForValidator(ctx context.Context, validatorID s if item.State != db.IPAwaitingSelfCheck && item.State != db.IPChecking { return nil, nil, nil } - return item, o.Checks, nil + resolved, err := o.DB.ListResolvedCheckTypes(ctx) + if err != nil { + return nil, nil, err + } + checks := make([]CheckConfig, len(resolved)) + for i, r := range resolved { + checks[i] = CheckConfig{Type: r.Type, Targets: r.Targets} + } + return item, checks, nil } -// SiteIndexForID resolves a configured site_id to its 1/2/3 index. -func (o *Orchestrator) SiteIndexForID(siteID string) int { - for _, s := range o.Sites { - if s.SiteID == siteID { - return s.Index - } - } - return 0 +// SiteIndexForID resolves a configured site_id to its 1/2/3 index, or +// (0, nil) if unconfigured. +func (o *Orchestrator) SiteIndexForID(ctx context.Context, siteID string) (int, error) { + return o.DB.GetSiteIndex(ctx, siteID) } // RecordCheck upserts a single check result and, if it represents a @@ -203,6 +201,43 @@ func (o *Orchestrator) MarkSiteComplete(ctx context.Context, ipID int64, siteInd return o.DB.SetSiteComplete(ctx, ipID, siteIndex) } +// ForceCancel stops an in-progress (or still-queued) check for the given +// address on admin request, even though it was never going to finish on +// its own within the checking window. If a floating IP is currently +// associated, it's disassociated best-effort (same fallthrough-on-error +// behavior as aggregateAndRelease/sweepExpiredLeases: the DB/validator +// state must still be freed even if Neutron hiccups). Returns +// db.ErrNotFound if the address is unknown, or db.ErrInvalidState if it has +// already reached done/failed. +func (o *Orchestrator) ForceCancel(ctx context.Context, ipAddress string) error { + item, err := o.DB.GetIPByAddress(ctx, ipAddress) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("ip %q: %w", ipAddress, db.ErrNotFound) + } + return err + } + + if item.FIPID != "" { + if err := o.OS.DisassociateFloatingIP(ctx, item.FIPID); err != nil { + o.Log.Error("disassociate fip on force cancel", "ip_id", item.ID, "fip_id", item.FIPID, "err", err) + } + } + + if err := o.DB.CancelIP(ctx, item.ID); err != nil { + return err + } + + if item.OwnerValidatorID != nil { + if err := o.DB.FreeValidator(ctx, *item.OwnerValidatorID); err != nil { + return fmt.Errorf("free validator: %w", err) + } + } + + o.event(ctx, "control-api", "", &item.ID, "force_cancel", "") + return nil +} + // sweepCheckingWindow moves IPs that have either finished reporting from // every source, or hit the checking-window deadline, into aggregation. func (o *Orchestrator) sweepCheckingWindow(ctx context.Context) error { @@ -211,8 +246,12 @@ func (o *Orchestrator) sweepCheckingWindow(ctx context.Context) error { if err != nil { return fmt.Errorf("list checking: %w", err) } + sites, err := o.DB.ListSites(ctx) + if err != nil { + return fmt.Errorf("list sites: %w", err) + } for _, item := range checking { - if !o.isReadyToAggregate(item, deadline) { + if !o.isReadyToAggregate(item, deadline, sites) { continue } if err := o.aggregateAndRelease(ctx, item); err != nil { @@ -225,20 +264,20 @@ func (o *Orchestrator) sweepCheckingWindow(ctx context.Context) error { // isReadyToAggregate reports whether an in-progress IP has either finished // reporting from every source it's actually expecting, or hit the // checking-window deadline. Which inbound sources it's expecting is driven -// entirely by o.Sites — inbound checks are optional: an empty (or -// partial) `sites` config in control-api.yaml means this IP is ready as -// soon as egress completes (or after the corresponding subset of -// siteN_complete flags), with no need to wait on a prober that will never -// exist. This is what makes inbound checks genuinely opt-in rather than a -// hardcoded expectation of exactly three sites. -func (o *Orchestrator) isReadyToAggregate(item db.IPQueueItem, deadline time.Time) bool { +// entirely by the currently configured sites — inbound checks are optional: +// an empty (or partial) sites configuration means this IP is ready as soon +// as egress completes (or after the corresponding subset of siteN_complete +// flags), with no need to wait on a prober that will never exist. This is +// what makes inbound checks genuinely opt-in rather than a hardcoded +// expectation of exactly three sites. +func (o *Orchestrator) isReadyToAggregate(item db.IPQueueItem, deadline time.Time, sites []db.Site) bool { if item.AssignedAt != nil && item.AssignedAt.Before(deadline) { return true } if !item.EgressComplete { return false } - for _, s := range o.Sites { + for _, s := range sites { switch s.Index { case 1: if !item.Site1Complete { @@ -266,7 +305,10 @@ func (o *Orchestrator) aggregateAndRelease(ctx context.Context, item db.IPQueueI return err } - expected := o.expectedCheckCount() + expected, err := o.expectedCheckCount(ctx) + if err != nil { + return fmt.Errorf("expected check count: %w", err) + } passCount := 0 for _, c := range checks { if c.Success { @@ -317,17 +359,28 @@ func (o *Orchestrator) aggregateAndRelease(ctx context.Context, item db.IPQueueI // expectedCheckCount is the number of check rows a fully-reported IP should // have: one per (egress check-type x target) plus one per (site x inbound -// port/icmp probe). -func (o *Orchestrator) expectedCheckCount() int { +// port/icmp probe). Reads the current check_types/targets/sites from the +// database, so a config change between assignment and aggregation is +// reflected in this specific aggregation (see the "accepted tradeoff" note +// in docs/PLAN_API_CONFIG_MANAGEMENT.md). +func (o *Orchestrator) expectedCheckCount(ctx context.Context) (int, error) { + resolved, err := o.DB.ListResolvedCheckTypes(ctx) + if err != nil { + return 0, err + } egress := 0 - for _, c := range o.Checks { + for _, c := range resolved { egress += len(c.Targets) } + sites, err := o.DB.ListSites(ctx) + if err != nil { + return 0, err + } inboundPerSite := len(o.Inbound.Ports) if o.Inbound.ICMP { inboundPerSite++ } - return egress + inboundPerSite*len(o.Sites) + return egress + inboundPerSite*len(sites), nil } // sweepExpiredLeases reclaims non-terminal IPs whose lease has passed — diff --git a/internal/orchestrator/orchestrator_test.go b/internal/orchestrator/orchestrator_test.go index 2dd6e5e..97063e3 100644 --- a/internal/orchestrator/orchestrator_test.go +++ b/internal/orchestrator/orchestrator_test.go @@ -58,6 +58,10 @@ func newTestOrchestratorWithSites(t *testing.T, leaseTTLSeconds int, sites []con Inbound: config.InboundConfig{Ports: []int{22, 80}, ICMP: true}, } + if err := d.BootstrapFromConfig(ctx, cfg); err != nil { + t.Fatalf("bootstrap from config: %v", err) + } + log := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) return New(d, mock, cfg, log), d, mock } diff --git a/scripts/run-local-e2e.sh b/scripts/run-local-e2e.sh index 8428bf7..6d25466 100755 --- a/scripts/run-local-e2e.sh +++ b/scripts/run-local-e2e.sh @@ -210,6 +210,22 @@ status | python3 -m json.tool echo "--- final ip results ---" ips | python3 -m json.tool +echo "--- demonstrating admin API: force a re-check of the already-done address ---" +curl -fs -X POST "http://127.0.0.1:28080/api/v1/admin/ips" \ + -H 'Content-Type: application/json' -d '{"addresses":["127.0.0.1"]}' | python3 -m json.tool + +echo "--- waiting for the forced re-check to drain (up to 30s) ---" +for i in $(seq 1 60); do + sleep 0.5 + remaining=$(status | python3 -c "import json,sys; d=json.load(sys.stdin); s=d['ips_by_state']; print(sum(v for k,v in s.items() if k not in ('done','failed')))" 2>/dev/null || echo "?") + if [ "$remaining" = "0" ]; then + echo "re-check drained after ~$((i / 2))s" + break + fi +done +echo "--- ip results after forced re-check (attempt_number should have advanced) ---" +ips | python3 -m json.tool + echo "--- logs are in $WORK_DIR (kept for inspection; workdir NOT auto-deleted) ---" trap - EXIT cleanup