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

make inbound check optional

Browse Source
This commit is contained in:
ayurishchev committed 2026-08-21 11:25:52 +03:00
1 parent 67adc6670e
commit c630f13c57
11 files changed
+203 -45
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/orchestrator/orchestrator.go
+41 -3
View File
@@ -207,11 +207,14 @@ func (o *Orchestrator) MarkSiteComplete(ctx context.Context, ipID int64, siteInd
// every source, or hit the checking-window deadline, into aggregation.
func (o *Orchestrator) sweepCheckingWindow(ctx context.Context) error {
deadline := db.Now().Add(-time.Duration(o.Cfg.CheckingWindowSeconds) * time.Second)
ready, err := o.DB.ListReadyToAggregate(ctx, deadline)
checking, err := o.DB.ListChecking(ctx)
if err != nil {
return fmt.Errorf("list ready to aggregate: %w", err)
return fmt.Errorf("list checking: %w", err)
}
for _, item := range ready {
for _, item := range checking {
if !o.isReadyToAggregate(item, deadline) {
continue
}
if err := o.aggregateAndRelease(ctx, item); err != nil {
o.Log.Error("aggregate and release", "ip_id", item.ID, "err", err)
}
@@ -219,6 +222,41 @@ func (o *Orchestrator) sweepCheckingWindow(ctx context.Context) error {
return nil
}
// 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 {
if item.AssignedAt != nil && item.AssignedAt.Before(deadline) {
return true
}
if !item.EgressComplete {
return false
}
for _, s := range o.Sites {
switch s.Index {
case 1:
if !item.Site1Complete {
return false
}
case 2:
if !item.Site2Complete {
return false
}
case 3:
if !item.Site3Complete {
return false
}
}
}
return true
}
func (o *Orchestrator) aggregateAndRelease(ctx context.Context, item db.IPQueueItem) error {
if err := o.DB.SetAggregating(ctx, item.ID); err != nil {
return err
internal/orchestrator/orchestrator_test.go
+110 -8
View File
@@ -13,7 +13,18 @@ import (
"cloudipvalidator/internal/openstack"
)
var threeSites = []config.SiteConfig{
{SiteID: "site-1", Index: 1},
{SiteID: "site-2", Index: 2},
{SiteID: "site-3", Index: 3},
}
func newTestOrchestrator(t *testing.T, leaseTTLSeconds int) (*Orchestrator, *db.DB, *openstack.MockClient) {
t.Helper()
return newTestOrchestratorWithSites(t, leaseTTLSeconds, threeSites)
}
func newTestOrchestratorWithSites(t *testing.T, leaseTTLSeconds int, sites []config.SiteConfig) (*Orchestrator, *db.DB, *openstack.MockClient) {
t.Helper()
ctx := context.Background()
dbPath := filepath.Join(t.TempDir(), "test.db")
@@ -36,11 +47,7 @@ func newTestOrchestrator(t *testing.T, leaseTTLSeconds int) (*Orchestrator, *db.
HeartbeatTimeoutSeconds: 30,
},
Aggregation: config.AggregationConfig{MissingCountsAsFail: true},
Sites: []config.SiteConfig{
{SiteID: "site-1", Index: 1},
{SiteID: "site-2", Index: 2},
{SiteID: "site-3", Index: 3},
},
Sites: sites,
CheckTypes: []config.CheckTypeConfig{
{Name: "https", Enabled: true, Targets: []string{"web"}},
{Name: "ssh", Enabled: false, Targets: []string{"web"}},
@@ -259,15 +266,24 @@ func TestLeaseReclaim(t *testing.T) {
func TestMaxRetriesExhausted(t *testing.T) {
ctx := context.Background()
o, d, mock := newTestOrchestrator(t, 0)
// A 1s lease (rather than 0) avoids the same intra-tick race noted in
// TestLeaseReclaim: with a 0s TTL, whether a freshly claimed item is
// reclaimed within the very same Tick (making each iteration's timing
// unpredictable) depends on how much wall-clock time claim+associate
// happened to take, which made this test flaky under load.
o, d, mock := newTestOrchestrator(t, 1)
o.Cfg.MaxRetries = 1
mock.Seed("fip-1", "1.2.3.4", "svc-project")
_ = d.RegisterValidator(ctx, "validator-1", "host-1", "port-1", "v0.1")
_ = d.SeedQueue(ctx, []string{"1.2.3.4"})
for i := 0; i < 3; i++ {
// Each (claim, sleep past 1s lease) pair reclaims once, bumping
// retry_count by 1; with MaxRetries=1 that needs two full reclaim
// cycles (retry_count 0->1 requeues, 1->2 fails) — four ticks total:
// claim, reclaim+requeue, re-claim, reclaim+fail.
for i := 0; i < 4; i++ {
o.Tick(ctx)
time.Sleep(5 * time.Millisecond)
time.Sleep(1100 * time.Millisecond)
}
ip, _ := d.GetIPByAddress(ctx, "1.2.3.4")
@@ -275,3 +291,89 @@ func TestMaxRetriesExhausted(t *testing.T) {
t.Fatalf("expected failed after exhausting retries, got %s (retry_count=%d)", ip.State, ip.RetryCount)
}
}
// TestInboundChecksDisabled confirms inbound (prober) checks are genuinely
// optional: with no sites configured, an IP must aggregate as soon as
// egress completes, without ever waiting on siteN_complete flags that
// nothing will ever set — and, critically, without waiting out the full
// checking_window_seconds timeout to get there (newTestOrchestrator uses
// 120s; this test never sleeps, so a pass here proves the "all required
// sources complete" path fired, not the timeout fallback).
func TestInboundChecksDisabled(t *testing.T) {
ctx := context.Background()
o, d, mock := newTestOrchestratorWithSites(t, 180, nil)
mock.Seed("fip-1", "1.2.3.4", "svc-project")
_ = d.RegisterValidator(ctx, "validator-1", "host-1", "port-1", "v0.1")
_ = d.SeedQueue(ctx, []string{"1.2.3.4"})
o.Tick(ctx)
ip, _ := d.GetIPByAddress(ctx, "1.2.3.4")
_ = o.SelfCheckResult(ctx, "validator-1", ip.ID, true, "ok")
ip, _ = d.GetIP(ctx, ip.ID)
_ = o.RecordCheck(ctx, db.Check{
IPID: ip.ID, IPAddress: ip.IPAddress, AttemptNumber: ip.AttemptNumber,
ValidatorID: "validator-1", Source: db.SourceEgress, CheckType: "https",
Target: "https://example.test", Success: true, CheckedAt: db.Now(),
})
_ = o.MarkEgressComplete(ctx, ip.ID)
o.Tick(ctx)
ip, _ = d.GetIP(ctx, ip.ID)
if ip.State != db.IPDone {
t.Fatalf("expected done immediately after egress completed with no sites configured, got %s", ip.State)
}
if ip.OverallResult != db.ResultPass {
t.Fatalf("expected pass, got %s", ip.OverallResult)
}
}
// TestInboundChecksPartialSites confirms a partially-configured sites list
// (fewer than 3 slots assigned) only waits on the sites actually
// configured — the two unassigned slots are never expected.
func TestInboundChecksPartialSites(t *testing.T) {
ctx := context.Background()
sites := []config.SiteConfig{{SiteID: "site-1", Index: 1}}
o, d, mock := newTestOrchestratorWithSites(t, 180, sites)
mock.Seed("fip-1", "1.2.3.4", "svc-project")
_ = d.RegisterValidator(ctx, "validator-1", "host-1", "port-1", "v0.1")
_ = d.SeedQueue(ctx, []string{"1.2.3.4"})
o.Tick(ctx)
ip, _ := d.GetIPByAddress(ctx, "1.2.3.4")
_ = o.SelfCheckResult(ctx, "validator-1", ip.ID, true, "ok")
ip, _ = d.GetIP(ctx, ip.ID)
_ = o.RecordCheck(ctx, db.Check{
IPID: ip.ID, IPAddress: ip.IPAddress, AttemptNumber: ip.AttemptNumber,
ValidatorID: "validator-1", Source: db.SourceEgress, CheckType: "https",
Target: "https://example.test", Success: true, CheckedAt: db.Now(),
})
_ = o.MarkEgressComplete(ctx, ip.ID)
// Egress is done but the one configured site (site-1) hasn't reported
// yet — must not aggregate.
o.Tick(ctx)
ip, _ = d.GetIP(ctx, ip.ID)
if ip.State != db.IPChecking {
t.Fatalf("expected still checking (site-1 pending), got %s", ip.State)
}
for _, ct := range []string{"tcp-22", "tcp-80", "icmp"} {
_ = o.RecordCheck(ctx, db.Check{
IPID: ip.ID, IPAddress: ip.IPAddress, AttemptNumber: ip.AttemptNumber,
Source: db.InboundSource(1), CheckType: ct, Target: ip.IPAddress, Success: true, CheckedAt: db.Now(),
})
}
_ = o.MarkSiteComplete(ctx, ip.ID, 1)
o.Tick(ctx)
ip, _ = d.GetIP(ctx, ip.ID)
if ip.State != db.IPDone {
t.Fatalf("expected done once the single configured site reported, got %s", ip.State)
}
if ip.OverallResult != db.ResultPass {
t.Fatalf("expected pass, got %s", ip.OverallResult)
}
}