2c884c5612
Некоторые проверки не удались
CI / test (push) Successful in 2m5s
Docker / Build and publish worker image (push) Failing after 31s
452 строки
16 KiB
Go
452 строки
16 KiB
Go
package models_test
|
||
|
||
import (
|
||
"sync"
|
||
"testing"
|
||
|
||
"github.com/stretchr/testify/assert"
|
||
"github.com/stretchr/testify/require"
|
||
"gorm.io/datatypes"
|
||
|
||
"rocketgit.ru/rsmon/worker/app/models"
|
||
"rocketgit.ru/rsmon/worker/config/database"
|
||
)
|
||
|
||
func init() {
|
||
database.Init()
|
||
}
|
||
|
||
// seedRegionRoutingWorld builds two regions and three groups (one per
|
||
// monitor) plus three workers. The monitors and their PreferredRegions /
|
||
// RegionMode are configured by the caller via a callback so each test can
|
||
// express the exact routing scenario it wants to assert on.
|
||
//
|
||
// Returns a routerTestWorld that knows how to look up each fixture by name
|
||
// for readable assertions.
|
||
type routerTestWorld struct {
|
||
groupAny *models.Group
|
||
groupSpec *models.Group
|
||
groupAll *models.Group
|
||
|
||
workerMsk *models.WorkerNode
|
||
workerUSEast *models.WorkerNode
|
||
workerLocal *models.WorkerNode
|
||
}
|
||
|
||
// seedWorker creates a worker bound to regionCode. AuthToken is unique per
|
||
// worker so the FOR UPDATE SKIP LOCKED path can exercise two distinct
|
||
// concurrent callers.
|
||
func seedWorker(t *testing.T, id, regionCode string) *models.WorkerNode {
|
||
t.Helper()
|
||
seedRegion(t, regionCode)
|
||
w := &models.WorkerNode{
|
||
WorkerID: id,
|
||
RegionCode: regionCode,
|
||
Status: "active",
|
||
AuthToken: "tok-" + id,
|
||
Concurrency: 4,
|
||
Capabilities: datatypes.JSON([]byte(`{"check_types":["http"]}`)),
|
||
}
|
||
require.NoError(t, models.DB().Create(w).Error)
|
||
return w
|
||
}
|
||
|
||
// seedRouterMonitor creates a Monitor with the given region routing
|
||
// attributes and one ready-to-run http Check. The check has Interval=60
|
||
// (minimum allowed) and LastStart=nil so ChecksForWorker picks it up
|
||
// immediately on the next poll.
|
||
func seedRouterMonitor(t *testing.T, group *models.Group, host, regionMode string, preferred []string) (models.Monitor, models.Check) {
|
||
t.Helper()
|
||
enTrue := true
|
||
mon := models.Monitor{
|
||
Name: stringPtrRouter(host),
|
||
Host: host,
|
||
GroupID: group.ID,
|
||
Enabled: true,
|
||
}
|
||
if regionMode != "" {
|
||
mon.RegionMode = regionMode
|
||
}
|
||
if preferred != nil {
|
||
mon.PreferredRegions = models.RegionCodesFromSlice(preferred)
|
||
}
|
||
require.NoError(t, models.DB().Create(&mon).Error)
|
||
|
||
check := models.Check{
|
||
MonitorID: mon.ID,
|
||
Kind: "http",
|
||
Interval: 60,
|
||
Enabled: &enTrue,
|
||
State: "UNK",
|
||
Settings: datatypes.JSON([]byte(`{}`)),
|
||
}
|
||
require.NoError(t, models.DB().Create(&check).Error)
|
||
return mon, check
|
||
}
|
||
|
||
func stringPtrRouter(s string) *string { return &s }
|
||
|
||
// seedRouterWorld is the common fixture for the TestRegionRouting_* table.
|
||
// It provisions two regions (ru-msk, us-east) and three monitors pinned to
|
||
// different routing modes; the workers are created lazily by the caller.
|
||
func seedRouterWorld(t *testing.T) routerTestWorld {
|
||
t.Helper()
|
||
models.Drop()
|
||
models.Migrate()
|
||
|
||
seedRegion(t, "ru-msk")
|
||
seedRegion(t, "us-east")
|
||
|
||
plan := models.Plan{Name: "router", Default: true}
|
||
require.NoError(t, models.DB().Create(&plan).Error)
|
||
acc := models.Account{Name: "router-acc", PlanID: &plan.ID}
|
||
require.NoError(t, models.DB().Create(&acc).Error)
|
||
|
||
groupAny := &models.Group{AccountID: acc.ID, Name: "any"}
|
||
groupSpec := &models.Group{AccountID: acc.ID, Name: "spec"}
|
||
groupAll := &models.Group{AccountID: acc.ID, Name: "all"}
|
||
require.NoError(t, models.DB().Create(groupAny).Error)
|
||
require.NoError(t, models.DB().Create(groupSpec).Error)
|
||
require.NoError(t, models.DB().Create(groupAll).Error)
|
||
|
||
seedRouterMonitor(t, groupAny, "any.test", models.RegionModeAny, nil)
|
||
seedRouterMonitor(t, groupSpec, "spec-msk.test", models.RegionModeSpecific, []string{"ru-msk"})
|
||
seedRouterMonitor(t, groupSpec, "spec-us.test", models.RegionModeSpecific, []string{"us-east"})
|
||
seedRouterMonitor(t, groupAll, "all.test", models.RegionModeAll, []string{"ru-msk", "us-east"})
|
||
|
||
return routerTestWorld{
|
||
groupAny: groupAny,
|
||
groupSpec: groupSpec,
|
||
groupAll: groupAll,
|
||
workerMsk: seedWorker(t, "w-msk", "ru-msk"),
|
||
workerUSEast: seedWorker(t, "w-us", "us-east"),
|
||
workerLocal: seedWorker(t, "w-local", "local"),
|
||
}
|
||
}
|
||
|
||
// idsOf returns a sorted list of monitor IDs assigned to the worker for
|
||
// easier assertions across checks.
|
||
func idsOf(checks []*models.Check) []int64 {
|
||
out := make([]int64, 0, len(checks))
|
||
for _, c := range checks {
|
||
if c.Monitor == nil {
|
||
continue
|
||
}
|
||
out = append(out, c.Monitor.ID)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// TestRegionRouting_AnyReturnsAll confirms the SQL filter preserves the
|
||
// legacy behavior for region_mode='any' monitors and the Phase 3
|
||
// placeholder 'all' monitors: a worker in a region nobody explicitly
|
||
// whitelisted must still see them, while monitors pinned to specific
|
||
// regions stay filtered out.
|
||
//
|
||
// seedRouterWorld configures four monitors:
|
||
// - any.test → region_mode='any', no PreferredRegions
|
||
// - spec-msk.test → region_mode='specific', preferred=[ru-msk]
|
||
// - spec-us.test → region_mode='specific', preferred=[us-east]
|
||
// - all.test → region_mode='all', preferred=[ru-msk, us-east]
|
||
//
|
||
// A worker in region "remote" (whitelisted by nobody) must see exactly
|
||
// {any.test, all.test} — the two monitors whose region_mode bypasses the
|
||
// whitelist — and nothing else.
|
||
func TestRegionRouting_AnyReturnsAll(t *testing.T) {
|
||
seedRouterWorld(t)
|
||
|
||
seedRegion(t, "remote")
|
||
w := seedWorker(t, "w-remote", "remote")
|
||
|
||
checks := models.ChecksForWorker(w, []string{"http"}, 50)
|
||
hosts := hostsOf(checks)
|
||
|
||
assert.ElementsMatch(t, []string{"any.test", "all.test"}, hosts,
|
||
"region_mode='any' and the Phase 3 'all' placeholder must bypass the whitelist")
|
||
assert.NotContains(t, hosts, "spec-msk.test",
|
||
"specific-mode monitor with whitelisted ru-msk must NOT reach a remote worker")
|
||
assert.NotContains(t, hosts, "spec-us.test",
|
||
"specific-mode monitor with whitelisted us-east must NOT reach a remote worker")
|
||
}
|
||
|
||
// TestRegionRouting_SpecificFiltersByRegion proves the core Phase 2 promise:
|
||
// workers in different regions never receive a monitor whose PreferredRegions
|
||
// does not include their region code. The check is run with parallel
|
||
// goroutines because ChecksForWorker stamps `last_start` on every row it
|
||
// leases — a sequential second poll would always see an already-claimed
|
||
// queue and the filter would have nothing to test against.
|
||
//
|
||
// Under the FOR UPDATE SKIP LOCKED race, whichever SELECT fires first grabs
|
||
// every matching row, so the *exact* per-worker host list is non-deterministic.
|
||
// The deterministic invariant the test asserts is the no-leak property: a
|
||
// worker in ru-msk must never see spec-us.test, and vice versa.
|
||
func TestRegionRouting_SpecificFiltersByRegion(t *testing.T) {
|
||
world := seedRouterWorld(t)
|
||
|
||
var (
|
||
wg sync.WaitGroup
|
||
mskChecks []*models.Check
|
||
usChecks []*models.Check
|
||
)
|
||
wg.Add(2)
|
||
go func() {
|
||
defer wg.Done()
|
||
mskChecks = models.ChecksForWorker(world.workerMsk, []string{"http"}, 50)
|
||
}()
|
||
go func() {
|
||
defer wg.Done()
|
||
usChecks = models.ChecksForWorker(world.workerUSEast, []string{"http"}, 50)
|
||
}()
|
||
wg.Wait()
|
||
|
||
mskHosts := hostsOf(mskChecks)
|
||
usHosts := hostsOf(usChecks)
|
||
|
||
// Aggregate coverage: together the two workers must see every check
|
||
// the routing layer would ever allow them — the four seeded monitors.
|
||
assert.ElementsMatch(t, []string{
|
||
"any.test",
|
||
"spec-msk.test",
|
||
"spec-us.test",
|
||
"all.test",
|
||
},
|
||
append(append([]string{}, mskHosts...), usHosts...),
|
||
"union of both workers' slices must cover every seeded monitor (any/specific/all × region)")
|
||
|
||
// Core Phase 2 invariant: regional filtering never leaks across
|
||
// PreferredRegions boundaries. This is the only assertion a
|
||
// concurrent SKIP LOCKED race lets us pin deterministically.
|
||
assert.NotContains(t, mskHosts, "spec-us.test",
|
||
"ru-msk worker must never see a monitor whitelisted for us-east only")
|
||
assert.NotContains(t, usHosts, "spec-msk.test",
|
||
"us-east worker must never see a monitor whitelisted for ru-msk only")
|
||
}
|
||
|
||
// TestRegionRouting_SpecificEmptyPreferredFallsBackToAny confirms the
|
||
// documented fall-back: a monitor in RegionModeSpecific with no
|
||
// PreferredRegions behaves like RegionModeAny so the field is safe to
|
||
// leave blank. We poll from the us-east worker — without the fall-back it
|
||
// would only see any.test + all.test + spec-us.test.
|
||
func TestRegionRouting_SpecificEmptyPreferredFallsBackToAny(t *testing.T) {
|
||
world := seedRouterWorld(t)
|
||
|
||
// Reset the spec-msk monitor to have an empty PreferredRegions list
|
||
// (the seed above gave it one). The Monitor row's RegionMode stays
|
||
// 'specific'.
|
||
require.NoError(t, models.DB().Model(&models.Monitor{}).
|
||
Where("host = ?", "spec-msk.test").
|
||
Update("preferred_regions", models.RegionCodesFromSlice(nil)).Error)
|
||
|
||
checks := models.ChecksForWorker(world.workerUSEast, []string{"http"}, 50)
|
||
hosts := hostsOf(checks)
|
||
assert.Contains(t, hosts, "spec-msk.test",
|
||
"empty PreferredRegions with region_mode=specific must fall back to 'any'")
|
||
}
|
||
|
||
// TestRegionRouting_AllDeferredToAny pins the Phase 3 placeholder behavior:
|
||
// region_mode='all' is logged and treated as 'any' today. The test asserts
|
||
// the monitor flows to a worker in any region (the TODO log marker is
|
||
// emitted from applyRegionRouting — pinned here as a code-grep contract).
|
||
func TestRegionRouting_AllDeferredToAny(t *testing.T) {
|
||
world := seedRouterWorld(t)
|
||
|
||
checks := models.ChecksForWorker(world.workerMsk, []string{"http"}, 50)
|
||
hosts := hostsOf(checks)
|
||
assert.Contains(t, hosts, "all.test",
|
||
"region_mode='all' must currently behave like 'any' so existing checks keep flowing")
|
||
}
|
||
|
||
// TestRegionRouting_LocalWorkerBypass ensures the historic "local" region
|
||
// still routes everything: the in-process scheduler handles those monitors
|
||
// and we don't want the Phase 2 filter to leak platform workers through it.
|
||
func TestRegionRouting_LocalWorkerBypass(t *testing.T) {
|
||
world := seedRouterWorld(t)
|
||
|
||
checks := models.ChecksForWorker(world.workerLocal, []string{"http"}, 50)
|
||
hosts := hostsOf(checks)
|
||
assert.ElementsMatch(t, []string{
|
||
"any.test",
|
||
"spec-msk.test",
|
||
"spec-us.test",
|
||
"all.test",
|
||
}, hosts, "worker in region 'local' must receive every check (bypass)")
|
||
}
|
||
|
||
// TestRegionRouting_NilWorkerReturnsAll asserts the diagnostic-friendly
|
||
// escape hatch: passing nil for the worker skips the routing filter and
|
||
// returns every check the kinds/limit envelope allows.
|
||
func TestRegionRouting_NilWorkerReturnsAll(t *testing.T) {
|
||
seedRouterWorld(t)
|
||
|
||
checks := models.ChecksForWorker(nil, []string{"http"}, 50)
|
||
hosts := hostsOf(checks)
|
||
assert.ElementsMatch(t, []string{
|
||
"any.test",
|
||
"spec-msk.test",
|
||
"spec-us.test",
|
||
"all.test",
|
||
}, hosts, "nil worker must bypass the routing filter")
|
||
}
|
||
|
||
// TestRegionRouting_LoadBalanceImplicit confirms the SKIP LOCKED implicit
|
||
// load-balancing story: when two workers in the same region race for a pool
|
||
// of pending checks, each of them receives a non-empty disjoint slice. The
|
||
// two polls run in parallel goroutines so the FOR UPDATE SKIP LOCKED race
|
||
// window is actually exercised.
|
||
//
|
||
// IMPORTANT: SKIP LOCKED with a large LIMIT is unfair — whichever
|
||
// transaction's SELECT fires first grabs everything. The test therefore
|
||
// uses LIMIT=4 with 10 pending rows so each worker is forced to leave some
|
||
// rows unlocked for the other worker to pick up. Together they must cover
|
||
// at most 8 rows (LIMIT × workers) without overlap; the remaining rows are
|
||
// intentionally left for a future poll cycle, which mirrors production
|
||
// behavior where workers continually drain a backlog.
|
||
func TestRegionRouting_LoadBalanceImplicit(t *testing.T) {
|
||
models.Drop()
|
||
models.Migrate()
|
||
|
||
seedRegion(t, "shared")
|
||
plan := models.Plan{Name: "lb-plan", Default: true}
|
||
require.NoError(t, models.DB().Create(&plan).Error)
|
||
acc := models.Account{Name: "lb", PlanID: &plan.ID}
|
||
require.NoError(t, models.DB().Create(&acc).Error)
|
||
group := &models.Group{AccountID: acc.ID, Name: "lb-g"}
|
||
require.NoError(t, models.DB().Create(group).Error)
|
||
|
||
enTrue := true
|
||
for i := 0; i < 10; i++ {
|
||
host := "lb-" + string(rune('a'+i)) + ".test"
|
||
mon := models.Monitor{
|
||
Name: stringPtrRouter(host),
|
||
Host: host,
|
||
GroupID: group.ID,
|
||
Enabled: true,
|
||
}
|
||
require.NoError(t, models.DB().Create(&mon).Error)
|
||
ck := models.Check{
|
||
MonitorID: mon.ID, Kind: "http", Interval: 60,
|
||
Enabled: &enTrue, State: "UNK",
|
||
Settings: datatypes.JSON([]byte(`{}`)),
|
||
}
|
||
require.NoError(t, models.DB().Create(&ck).Error)
|
||
}
|
||
|
||
w1 := seedWorker(t, "lb-w1", "shared")
|
||
w2 := seedWorker(t, "lb-w2", "shared")
|
||
|
||
const limitPerWorker = 4
|
||
var (
|
||
wg sync.WaitGroup
|
||
aChecks, bChecks []*models.Check
|
||
)
|
||
wg.Add(2)
|
||
go func() {
|
||
defer wg.Done()
|
||
aChecks = models.ChecksForWorker(w1, []string{"http"}, limitPerWorker)
|
||
}()
|
||
go func() {
|
||
defer wg.Done()
|
||
bChecks = models.ChecksForWorker(w2, []string{"http"}, limitPerWorker)
|
||
}()
|
||
wg.Wait()
|
||
|
||
assert.Greater(t, len(aChecks), 0, "worker 1 must receive at least one check")
|
||
assert.Greater(t, len(bChecks), 0, "worker 2 must receive at least one check")
|
||
assert.LessOrEqual(t, len(aChecks)+len(bChecks), 2*limitPerWorker,
|
||
"two concurrent workers with LIMIT each can lease at most LIMIT*2 rows per cycle")
|
||
assert.Empty(t, intersectHosts(aChecks, bChecks),
|
||
"the two slices must be disjoint (FOR UPDATE SKIP LOCKED must not double-lease)")
|
||
}
|
||
|
||
// TestMonitorValidateRegionMode exercises the documented enum on the
|
||
// Monitor type so the validator surface does not regress.
|
||
func TestMonitorValidateRegionMode(t *testing.T) {
|
||
cases := []struct {
|
||
mode string
|
||
wantErr bool
|
||
}{
|
||
{"", false},
|
||
{"any", false},
|
||
{"specific", false},
|
||
{"all", false},
|
||
{"round-robin", true},
|
||
{"RANDOM", true},
|
||
}
|
||
for _, c := range cases {
|
||
t.Run("mode="+c.mode, func(t *testing.T) {
|
||
m := models.Monitor{RegionMode: c.mode}
|
||
err := m.ValidateRegionMode()
|
||
if c.wantErr {
|
||
assert.Error(t, err)
|
||
} else {
|
||
assert.NoError(t, err)
|
||
}
|
||
})
|
||
}
|
||
}
|
||
|
||
// TestMonitorWantsRegion pins the public WantsRegion helper that powers the
|
||
// non-SQL callers (UI preview, plan validation). Phase 3 will swap the
|
||
// 'all' branch's behavior, so the table covers all three modes today.
|
||
func TestMonitorWantsRegion(t *testing.T) {
|
||
cases := []struct {
|
||
name string
|
||
mode string
|
||
regions []string
|
||
workerRC string
|
||
want bool
|
||
}{
|
||
{"any_always_true", "any", []string{"ru-msk"}, "us-east", true},
|
||
{"any_empty_pref_still_true", "any", nil, "us-east", true},
|
||
{"specific_match", "specific", []string{"ru-msk", "eu-west"}, "ru-msk", true},
|
||
{"specific_no_match", "specific", []string{"ru-msk", "eu-west"}, "us-east", false},
|
||
{"specific_empty_pref_fallback", "specific", nil, "us-east", true},
|
||
{"all_placeholder_true", "all", []string{"ru-msk", "us-east"}, "ru-msk", true},
|
||
{"all_placeholder_foreign_region", "all", []string{"ru-msk", "us-east"}, "eu-west", true},
|
||
{"empty_mode_defaults_to_any", "", nil, "us-east", true},
|
||
}
|
||
for _, c := range cases {
|
||
t.Run(c.name, func(t *testing.T) {
|
||
m := models.Monitor{
|
||
RegionMode: c.mode,
|
||
PreferredRegions: models.RegionCodesFromSlice(c.regions),
|
||
}
|
||
assert.Equal(t, c.want, m.WantsRegion(c.workerRC))
|
||
})
|
||
}
|
||
}
|
||
|
||
// hostsOf extracts the hostnames from the assigned checks for readable
|
||
// assertions in table-driven tests.
|
||
func hostsOf(checks []*models.Check) []string {
|
||
out := make([]string, 0, len(checks))
|
||
for _, c := range checks {
|
||
if c.Monitor == nil {
|
||
continue
|
||
}
|
||
out = append(out, c.Monitor.Host)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// intersectHosts returns the hostnames present in both slices — used to
|
||
// prove two concurrent workers did not lease the same check twice.
|
||
func intersectHosts(a, b []*models.Check) []string {
|
||
set := make(map[string]struct{}, len(a))
|
||
for _, c := range a {
|
||
if c.Monitor != nil {
|
||
set[c.Monitor.Host] = struct{}{}
|
||
}
|
||
}
|
||
var out []string
|
||
for _, c := range b {
|
||
if c.Monitor == nil {
|
||
continue
|
||
}
|
||
if _, ok := set[c.Monitor.Host]; ok {
|
||
out = append(out, c.Monitor.Host)
|
||
}
|
||
}
|
||
return out
|
||
}
|