mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-19 10:33:29 +00:00
Scale connection staleness cutoff with the configured polling cadence
The connections ledger marked any PVE/PBS/PMG/VMware/TrueNAS connection stale once its last successful poll was more than a fixed 2 minutes old, so polling cadences above 120s read as permanently stale and latched the connection-degraded alert with no path back to active (#1620). Derive the active->stale cutoff as max(3 x polling interval, 2 minutes), mirroring how availability probes already scale. PVE/PBS/PMG cadences come from the server config; TrueNAS uses the per-instance interval and VMware the poller summary interval. The 3x multiplier absorbs the evaluation-order skew where connection alerts are checked concurrently with the poll on the same scheduler tick, leaving LastSuccess roughly one interval old even when healthy. Fixes #1620 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
courtmanr@gmail.com
parent
c9600eace1
commit
5a9e20d971
5 changed files with 179 additions and 41 deletions
|
|
@ -18,13 +18,29 @@ import (
|
|||
"github.com/rcourtman/pulse-go-rewrite/internal/remoteconfig"
|
||||
)
|
||||
|
||||
// connectionStaleThreshold is the baseline "haven't heard from this connection
|
||||
// recently" cutoff used to transition `active` → `stale`. Per-type poll
|
||||
// intervals vary (PVE 5-10s, TrueNAS 60s, agents 30-60s); 2 minutes sits
|
||||
// connectionStaleThreshold is the floor for the "haven't heard from this
|
||||
// connection recently" cutoff used to transition `active` → `stale`. Per-type
|
||||
// poll intervals vary (PVE 5-10s, TrueNAS 60s, agents 30-60s); 2 minutes sits
|
||||
// comfortably above 2× the slowest default so we don't flap on a single
|
||||
// dropped tick. Refined later once per-type intervals become first-class.
|
||||
// dropped tick. Connections polled slower than that get a scaled cutoff via
|
||||
// connectionStaleThresholdFor.
|
||||
const connectionStaleThreshold = 2 * time.Minute
|
||||
|
||||
// connectionStaleThresholdFor scales the active→stale cutoff with the
|
||||
// configured polling cadence: max(3 × interval, connectionStaleThreshold).
|
||||
// The multiplier is 3× (not the 2× availability probes use) because
|
||||
// connection-degraded alerts are evaluated concurrently with the poll on the
|
||||
// same scheduler tick, so a healthy connection's LastSuccess is routinely one
|
||||
// full interval old at evaluation time; 2× would flap on a single missed
|
||||
// tick. A non-positive interval (cadence unknown) keeps the floor.
|
||||
func connectionStaleThresholdFor(pollInterval time.Duration) time.Duration {
|
||||
threshold := 3 * pollInterval
|
||||
if threshold < connectionStaleThreshold {
|
||||
return connectionStaleThreshold
|
||||
}
|
||||
return threshold
|
||||
}
|
||||
|
||||
// connectionAuthErrorPattern matches the error strings pollers surface when a
|
||||
// credential is wrong or the token lacks scope. Kept centrally so
|
||||
// state-derivation stays the same across types.
|
||||
|
|
@ -106,6 +122,14 @@ type aggregatorInputs struct {
|
|||
instanceHealth map[string]monitoring.InstanceHealth
|
||||
expectedAgentVersion string
|
||||
now time.Time
|
||||
|
||||
// Configured poll cadences, used to scale the active→stale cutoff so slow
|
||||
// cadences (e.g. 5 minutes) don't read as permanently stale. Zero means
|
||||
// unknown and falls back to the connectionStaleThreshold floor. VMware and
|
||||
// TrueNAS cadences come from their summaries/instances instead.
|
||||
pvePollingInterval time.Duration
|
||||
pbsPollingInterval time.Duration
|
||||
pmgPollingInterval time.Duration
|
||||
}
|
||||
|
||||
type connectionAgentDesiredConfig struct {
|
||||
|
|
@ -127,13 +151,13 @@ func buildConnections(in aggregatorInputs) []Connection {
|
|||
len(in.vmwareInstances)+len(in.truenasInstances)+len(in.availabilityTargets)+len(in.hosts))
|
||||
|
||||
for _, pve := range in.pveInstances {
|
||||
out = append(out, buildPVEConnection(pve, in.instanceHealth, now))
|
||||
out = append(out, buildPVEConnection(pve, in.instanceHealth, now, in.pvePollingInterval))
|
||||
}
|
||||
for _, pbs := range in.pbsInstances {
|
||||
out = append(out, buildPBSConnection(pbs, in.instanceHealth, now))
|
||||
out = append(out, buildPBSConnection(pbs, in.instanceHealth, now, in.pbsPollingInterval))
|
||||
}
|
||||
for _, pmg := range in.pmgInstances {
|
||||
out = append(out, buildPMGConnection(pmg, in.instanceHealth, now))
|
||||
out = append(out, buildPMGConnection(pmg, in.instanceHealth, now, in.pmgPollingInterval))
|
||||
}
|
||||
for _, vmw := range in.vmwareInstances {
|
||||
out = append(out, buildVMwareConnection(vmw, in.instanceHealth, in.vmwareSummaries, now))
|
||||
|
|
@ -166,7 +190,7 @@ func buildConnections(in aggregatorInputs) []Connection {
|
|||
return out
|
||||
}
|
||||
|
||||
func buildPVEConnection(inst config.PVEInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection {
|
||||
func buildPVEConnection(inst config.PVEInstance, health map[string]monitoring.InstanceHealth, now time.Time, pollInterval time.Duration) Connection {
|
||||
enabled := !inst.Disabled
|
||||
surfaces := []string{"vms", "containers", "storage", "backups"}
|
||||
scope := map[string]bool{
|
||||
|
|
@ -176,7 +200,7 @@ func buildPVEConnection(inst config.PVEInstance, health map[string]monitoring.In
|
|||
"backups": inst.MonitorBackups,
|
||||
}
|
||||
h := health["pve::"+inst.Name]
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now)
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now, pollInterval)
|
||||
conn := withFleetGovernance(Connection{
|
||||
ID: "pve:" + inst.Name,
|
||||
Type: ConnectionTypePVE,
|
||||
|
|
@ -197,7 +221,7 @@ func buildPVEConnection(inst config.PVEInstance, health map[string]monitoring.In
|
|||
return conn
|
||||
}
|
||||
|
||||
func buildPBSConnection(inst config.PBSInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection {
|
||||
func buildPBSConnection(inst config.PBSInstance, health map[string]monitoring.InstanceHealth, now time.Time, pollInterval time.Duration) Connection {
|
||||
enabled := !inst.Disabled
|
||||
surfaces := []string{"backups", "datastores", "syncJobs", "verifyJobs", "pruneJobs", "garbageJobs"}
|
||||
scope := map[string]bool{
|
||||
|
|
@ -209,7 +233,7 @@ func buildPBSConnection(inst config.PBSInstance, health map[string]monitoring.In
|
|||
"garbageJobs": inst.MonitorGarbageJobs,
|
||||
}
|
||||
h := health["pbs::"+inst.Name]
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now)
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now, pollInterval)
|
||||
conn := withFleetGovernance(Connection{
|
||||
ID: "pbs:" + inst.Name,
|
||||
Type: ConnectionTypePBS,
|
||||
|
|
@ -230,7 +254,7 @@ func buildPBSConnection(inst config.PBSInstance, health map[string]monitoring.In
|
|||
return conn
|
||||
}
|
||||
|
||||
func buildPMGConnection(inst config.PMGInstance, health map[string]monitoring.InstanceHealth, now time.Time) Connection {
|
||||
func buildPMGConnection(inst config.PMGInstance, health map[string]monitoring.InstanceHealth, now time.Time, pollInterval time.Duration) Connection {
|
||||
enabled := !inst.Disabled
|
||||
surfaces := []string{"mailStats", "queues", "quarantine", "domainStats"}
|
||||
scope := map[string]bool{
|
||||
|
|
@ -240,7 +264,7 @@ func buildPMGConnection(inst config.PMGInstance, health map[string]monitoring.In
|
|||
"domainStats": inst.MonitorDomainStats,
|
||||
}
|
||||
h := health["pmg::"+inst.Name]
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now)
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now, pollInterval)
|
||||
conn := withFleetGovernance(Connection{
|
||||
ID: "pmg:" + inst.Name,
|
||||
Type: ConnectionTypePMG,
|
||||
|
|
@ -275,10 +299,14 @@ func buildVMwareConnection(
|
|||
"datastores": inst.MonitorDatastores,
|
||||
}
|
||||
h := health["vmware::"+inst.ID]
|
||||
var pollInterval time.Duration
|
||||
if summary, ok := summaries[strings.TrimSpace(inst.ID)]; ok {
|
||||
h = mergeVMwareSummaryHealth(h, summary)
|
||||
if summary.Poll != nil && summary.Poll.IntervalSeconds > 0 {
|
||||
pollInterval = time.Duration(summary.Poll.IntervalSeconds) * time.Second
|
||||
}
|
||||
}
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now)
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now, pollInterval)
|
||||
port := inst.Port
|
||||
if port == 0 {
|
||||
port = 443
|
||||
|
|
@ -320,7 +348,8 @@ func buildTrueNASConnection(
|
|||
if summary, ok := summaries[strings.TrimSpace(inst.ID)]; ok {
|
||||
h = mergeTrueNASSummaryHealth(h, summary)
|
||||
}
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now)
|
||||
pollInterval := time.Duration(inst.EffectivePollIntervalSecs()) * time.Second
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(enabled, h, now, pollInterval)
|
||||
scheme := "https"
|
||||
if !inst.UseHTTPS {
|
||||
scheme = "http"
|
||||
|
|
@ -1279,8 +1308,10 @@ func connectionHostAliasesForAgent(host models.Host, name, address string) []str
|
|||
|
||||
// deriveConnectionState maps (Enabled, InstanceHealth) onto the unified state
|
||||
// vocabulary. No new state is persisted — the inputs come from the existing
|
||||
// monitoring scheduler.
|
||||
func deriveConnectionState(enabled bool, h monitoring.InstanceHealth, now time.Time) (ConnectionState, string, *time.Time, *ConnectionError) {
|
||||
// monitoring scheduler. pollInterval is the configured cadence for this
|
||||
// connection (zero if unknown); it scales the active→stale cutoff via
|
||||
// connectionStaleThresholdFor.
|
||||
func deriveConnectionState(enabled bool, h monitoring.InstanceHealth, now time.Time, pollInterval time.Duration) (ConnectionState, string, *time.Time, *ConnectionError) {
|
||||
var lastSeen *time.Time
|
||||
if h.PollStatus.LastSuccess != nil && !h.PollStatus.LastSuccess.IsZero() {
|
||||
t := *h.PollStatus.LastSuccess
|
||||
|
|
@ -1324,7 +1355,7 @@ func deriveConnectionState(enabled bool, h monitoring.InstanceHealth, now time.T
|
|||
return ConnectionStateUnreachable, reason, lastSeen, lastError
|
||||
}
|
||||
|
||||
if lastSeen != nil && now.Sub(*lastSeen) > connectionStaleThreshold {
|
||||
if lastSeen != nil && now.Sub(*lastSeen) > connectionStaleThresholdFor(pollInterval) {
|
||||
return ConnectionStateStale, fmt.Sprintf("no successful poll in %s", now.Sub(*lastSeen).Round(time.Second)), lastSeen, lastError
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ var wantPMGSurfaces = []string{"mailStats", "queues", "quarantine", "domainStats
|
|||
func TestBranchcov0722PMGConnectionDisabledFlip(t *testing.T) {
|
||||
t.Run("disabled_instance_is_paused_and_not_enabled", func(t *testing.T) {
|
||||
inst := config.PMGInstance{Name: "mail-relay", Host: "mail.lan", Disabled: true}
|
||||
conn := buildPMGConnection(inst, nil, frozenNow)
|
||||
conn := buildPMGConnection(inst, nil, frozenNow, 0)
|
||||
|
||||
if conn.Enabled {
|
||||
t.Fatalf("Enabled = true, want false for Disabled instance")
|
||||
|
|
@ -104,7 +104,7 @@ func TestBranchcov0722PMGConnectionDisabledFlip(t *testing.T) {
|
|||
|
||||
t.Run("enabled_instance_is_not_paused", func(t *testing.T) {
|
||||
inst := config.PMGInstance{Name: "mail-relay", Host: "mail.lan"}
|
||||
conn := buildPMGConnection(inst, nil, frozenNow)
|
||||
conn := buildPMGConnection(inst, nil, frozenNow, 0)
|
||||
|
||||
if !conn.Enabled {
|
||||
t.Fatalf("Enabled = false, want true for non-Disabled instance")
|
||||
|
|
@ -177,7 +177,7 @@ func TestBranchcov0722PMGConnectionScopeFlags(t *testing.T) {
|
|||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
conn := buildPMGConnection(tc.inst, nil, frozenNow)
|
||||
conn := buildPMGConnection(tc.inst, nil, frozenNow, 0)
|
||||
|
||||
// Surfaces are always the full, ordered list regardless of scope.
|
||||
if !reflect.DeepEqual(conn.Surfaces, wantPMGSurfaces) {
|
||||
|
|
@ -216,7 +216,7 @@ func TestBranchcov0722PMGConnectionHealthLookup(t *testing.T) {
|
|||
health := map[string]monitoring.InstanceHealth{
|
||||
"pmg::mail-relay": pmgHealthEntry(&recentSuccess, nil, "", "", "closed"),
|
||||
}
|
||||
conn := buildPMGConnection(inst, health, frozenNow)
|
||||
conn := buildPMGConnection(inst, health, frozenNow, 0)
|
||||
if conn.State != ConnectionStateActive {
|
||||
t.Fatalf("State = %q, want %q", conn.State, ConnectionStateActive)
|
||||
}
|
||||
|
|
@ -236,7 +236,7 @@ func TestBranchcov0722PMGConnectionHealthLookup(t *testing.T) {
|
|||
"empty_map": {},
|
||||
"wrong_key": {"pmg::other-relay": pmgHealthEntry(&recentSuccess, nil, "", "", "closed")},
|
||||
} {
|
||||
conn := buildPMGConnection(inst, health, frozenNow)
|
||||
conn := buildPMGConnection(inst, health, frozenNow, 0)
|
||||
if conn.State != ConnectionStatePending {
|
||||
t.Fatalf("%s: State = %q, want %q", label, conn.State, ConnectionStatePending)
|
||||
}
|
||||
|
|
@ -271,7 +271,7 @@ func TestBranchcov0722PMGConnectionStateDerivation(t *testing.T) {
|
|||
health := map[string]monitoring.InstanceHealth{
|
||||
healthKey: pmgHealthEntry(ptrTime(frozenNow.Add(-30*time.Second)), &errAt, "401 Unauthorized", "auth", "closed"),
|
||||
}
|
||||
conn := buildPMGConnection(inst, health, frozenNow)
|
||||
conn := buildPMGConnection(inst, health, frozenNow, 0)
|
||||
if conn.State != ConnectionStatePaused {
|
||||
t.Fatalf("State = %q, want %q", conn.State, ConnectionStatePaused)
|
||||
}
|
||||
|
|
@ -283,7 +283,7 @@ func TestBranchcov0722PMGConnectionStateDerivation(t *testing.T) {
|
|||
t.Run("pending_with_no_poll_history", func(t *testing.T) {
|
||||
conn := buildPMGConnection(base, map[string]monitoring.InstanceHealth{
|
||||
healthKey: pmgHealthEntry(nil, nil, "", "", "closed"),
|
||||
}, frozenNow)
|
||||
}, frozenNow, 0)
|
||||
if conn.State != ConnectionStatePending {
|
||||
t.Fatalf("State = %q, want %q", conn.State, ConnectionStatePending)
|
||||
}
|
||||
|
|
@ -300,7 +300,7 @@ func TestBranchcov0722PMGConnectionStateDerivation(t *testing.T) {
|
|||
health := map[string]monitoring.InstanceHealth{
|
||||
healthKey: pmgHealthEntry(nil, &errAt, "403 Forbidden: token lacks scope", "auth", "closed"),
|
||||
}
|
||||
conn := buildPMGConnection(base, health, frozenNow)
|
||||
conn := buildPMGConnection(base, health, frozenNow, 0)
|
||||
if conn.State != ConnectionStateUnauthorized {
|
||||
t.Fatalf("State = %q, want %q", conn.State, ConnectionStateUnauthorized)
|
||||
}
|
||||
|
|
@ -329,7 +329,7 @@ func TestBranchcov0722PMGConnectionStateDerivation(t *testing.T) {
|
|||
health := map[string]monitoring.InstanceHealth{
|
||||
healthKey: pmgHealthEntry(nil, &errAt, "connection refused", "network", "closed"),
|
||||
}
|
||||
conn := buildPMGConnection(base, health, frozenNow)
|
||||
conn := buildPMGConnection(base, health, frozenNow, 0)
|
||||
if conn.State != ConnectionStateUnreachable {
|
||||
t.Fatalf("State = %q, want %q", conn.State, ConnectionStateUnreachable)
|
||||
}
|
||||
|
|
@ -348,7 +348,7 @@ func TestBranchcov0722PMGConnectionStateDerivation(t *testing.T) {
|
|||
health := map[string]monitoring.InstanceHealth{
|
||||
healthKey: pmgHealthEntry(&success, nil, "", "", "open"),
|
||||
}
|
||||
conn := buildPMGConnection(base, health, frozenNow)
|
||||
conn := buildPMGConnection(base, health, frozenNow, 0)
|
||||
if conn.State != ConnectionStateUnreachable {
|
||||
t.Fatalf("State = %q, want %q", conn.State, ConnectionStateUnreachable)
|
||||
}
|
||||
|
|
@ -361,12 +361,13 @@ func TestBranchcov0722PMGConnectionStateDerivation(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("stale_when_last_success_beyond_threshold", func(t *testing.T) {
|
||||
// connectionStaleThreshold is 2 minutes; 5 minutes ago must be stale.
|
||||
// With no configured interval the connectionStaleThreshold floor of
|
||||
// 2 minutes applies; 5 minutes ago must be stale.
|
||||
staleSuccess := frozenNow.Add(-5 * time.Minute)
|
||||
health := map[string]monitoring.InstanceHealth{
|
||||
healthKey: pmgHealthEntry(&staleSuccess, nil, "", "", "closed"),
|
||||
}
|
||||
conn := buildPMGConnection(base, health, frozenNow)
|
||||
conn := buildPMGConnection(base, health, frozenNow, 0)
|
||||
if conn.State != ConnectionStateStale {
|
||||
t.Fatalf("State = %q, want %q", conn.State, ConnectionStateStale)
|
||||
}
|
||||
|
|
@ -384,7 +385,7 @@ func TestBranchcov0722PMGConnectionStateDerivation(t *testing.T) {
|
|||
health := map[string]monitoring.InstanceHealth{
|
||||
healthKey: pmgHealthEntry(&recentSuccess, nil, "", "", "closed"),
|
||||
}
|
||||
conn := buildPMGConnection(base, health, frozenNow)
|
||||
conn := buildPMGConnection(base, health, frozenNow, 0)
|
||||
if conn.State != ConnectionStateActive {
|
||||
t.Fatalf("State = %q, want %q", conn.State, ConnectionStateActive)
|
||||
}
|
||||
|
|
@ -427,7 +428,7 @@ func TestBranchcov0722PMGConnectionCredentialKind(t *testing.T) {
|
|||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
conn := buildPMGConnection(tc.inst, nil, frozenNow)
|
||||
conn := buildPMGConnection(tc.inst, nil, frozenNow, 0)
|
||||
ch := conn.Fleet.CredentialHealth
|
||||
if ch == nil {
|
||||
t.Fatalf("CredentialHealth is nil")
|
||||
|
|
@ -457,7 +458,7 @@ func TestBranchcov0722PMGConnectionStaticFields(t *testing.T) {
|
|||
Name: "mail-relay",
|
||||
Host: "https://mail.lan:8006",
|
||||
}
|
||||
conn := buildPMGConnection(inst, nil, frozenNow)
|
||||
conn := buildPMGConnection(inst, nil, frozenNow, 0)
|
||||
|
||||
if conn.ID != "pmg:mail-relay" {
|
||||
t.Fatalf("ID = %q, want %q", conn.ID, "pmg:mail-relay")
|
||||
|
|
|
|||
|
|
@ -49,7 +49,7 @@ func desiredAgentConfig(t *testing.T, commandsEnabled *bool, settings map[string
|
|||
}
|
||||
|
||||
func TestDeriveConnectionState_Paused(t *testing.T) {
|
||||
state, reason, _, _ := deriveConnectionState(false, monitoring.InstanceHealth{}, time.Now())
|
||||
state, reason, _, _ := deriveConnectionState(false, monitoring.InstanceHealth{}, time.Now(), 0)
|
||||
if state != ConnectionStatePaused {
|
||||
t.Fatalf("got %q, want %q", state, ConnectionStatePaused)
|
||||
}
|
||||
|
|
@ -59,7 +59,7 @@ func TestDeriveConnectionState_Paused(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestDeriveConnectionState_Pending(t *testing.T) {
|
||||
state, _, lastSeen, lastError := deriveConnectionState(true, monitoring.InstanceHealth{}, time.Now())
|
||||
state, _, lastSeen, lastError := deriveConnectionState(true, monitoring.InstanceHealth{}, time.Now(), 0)
|
||||
if state != ConnectionStatePending {
|
||||
t.Fatalf("got %q, want %q", state, ConnectionStatePending)
|
||||
}
|
||||
|
|
@ -71,7 +71,7 @@ func TestDeriveConnectionState_Pending(t *testing.T) {
|
|||
func TestDeriveConnectionState_Unauthorized(t *testing.T) {
|
||||
now := time.Now()
|
||||
h := healthEntry(ptrTime(now.Add(-30*time.Second)), "401 Unauthorized: token invalid", "auth", "closed")
|
||||
state, reason, _, err := deriveConnectionState(true, h, now)
|
||||
state, reason, _, err := deriveConnectionState(true, h, now, 0)
|
||||
if state != ConnectionStateUnauthorized {
|
||||
t.Fatalf("got %q, want %q", state, ConnectionStateUnauthorized)
|
||||
}
|
||||
|
|
@ -86,7 +86,7 @@ func TestDeriveConnectionState_Unauthorized(t *testing.T) {
|
|||
func TestDeriveConnectionState_Unreachable(t *testing.T) {
|
||||
now := time.Now()
|
||||
h := healthEntry(ptrTime(now.Add(-30*time.Second)), "connection refused", "network", "open")
|
||||
state, _, _, _ := deriveConnectionState(true, h, now)
|
||||
state, _, _, _ := deriveConnectionState(true, h, now, 0)
|
||||
if state != ConnectionStateUnreachable {
|
||||
t.Fatalf("got %q, want %q", state, ConnectionStateUnreachable)
|
||||
}
|
||||
|
|
@ -96,7 +96,7 @@ func TestDeriveConnectionState_Stale(t *testing.T) {
|
|||
now := time.Now()
|
||||
stale := now.Add(-5 * time.Minute)
|
||||
h := healthEntry(ptrTime(stale), "", "", "closed")
|
||||
state, reason, _, _ := deriveConnectionState(true, h, now)
|
||||
state, reason, _, _ := deriveConnectionState(true, h, now, 0)
|
||||
if state != ConnectionStateStale {
|
||||
t.Fatalf("got %q, want %q", state, ConnectionStateStale)
|
||||
}
|
||||
|
|
@ -108,7 +108,7 @@ func TestDeriveConnectionState_Stale(t *testing.T) {
|
|||
func TestDeriveConnectionState_Active(t *testing.T) {
|
||||
now := time.Now()
|
||||
h := healthEntry(ptrTime(now.Add(-10*time.Second)), "", "", "closed")
|
||||
state, _, _, _ := deriveConnectionState(true, h, now)
|
||||
state, _, _, _ := deriveConnectionState(true, h, now, 0)
|
||||
if state != ConnectionStateActive {
|
||||
t.Fatalf("got %q, want %q", state, ConnectionStateActive)
|
||||
}
|
||||
|
|
@ -1024,7 +1024,7 @@ func TestBuildConnections_PlatformPollerSummariesDriveRuntimeState(t *testing.T)
|
|||
func TestDeriveConnectionState_FirstPollFailureIsUnreachable(t *testing.T) {
|
||||
now := time.Now()
|
||||
h := healthEntry(nil, "connection refused", "network", "closed")
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(true, h, now)
|
||||
state, reason, lastSeen, lastError := deriveConnectionState(true, h, now, 0)
|
||||
if state != ConnectionStateUnreachable {
|
||||
t.Fatalf("state = %q, want unreachable", state)
|
||||
}
|
||||
|
|
@ -1061,7 +1061,7 @@ func TestDeriveConnectionState_CurrentFailureOverridesRecentSuccess(t *testing.T
|
|||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
h := healthEntry(&lastSuccess, test.message, "transient", "closed")
|
||||
state, reason, gotLastSeen, lastError := deriveConnectionState(true, h, now)
|
||||
state, reason, gotLastSeen, lastError := deriveConnectionState(true, h, now, 0)
|
||||
if state != test.wantState {
|
||||
t.Fatalf("state = %q, want %q", state, test.wantState)
|
||||
}
|
||||
|
|
@ -1137,3 +1137,83 @@ func TestBuildConnections_UsesHealthLookup(t *testing.T) {
|
|||
t.Fatalf("lastSeen not propagated: %+v", got[0].LastSeen)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConnectionStaleThresholdFor(t *testing.T) {
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
interval time.Duration
|
||||
want time.Duration
|
||||
}{
|
||||
{name: "unknown cadence keeps floor", interval: 0, want: connectionStaleThreshold},
|
||||
{name: "negative cadence keeps floor", interval: -time.Minute, want: connectionStaleThreshold},
|
||||
{name: "fast cadence keeps floor", interval: 10 * time.Second, want: connectionStaleThreshold},
|
||||
{name: "floor boundary cadence keeps floor", interval: 40 * time.Second, want: connectionStaleThreshold},
|
||||
{name: "slow cadence scales to 3x", interval: 5 * time.Minute, want: 15 * time.Minute},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
if got := connectionStaleThresholdFor(test.interval); got != test.want {
|
||||
t.Fatalf("connectionStaleThresholdFor(%s) = %s, want %s", test.interval, got, test.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Regression for https://github.com/rcourtman/Pulse/issues/1620: with a
|
||||
// 5-minute polling cadence the previous fixed 2-minute cutoff marked every
|
||||
// connection permanently stale, latching the connection-degraded alert.
|
||||
func TestDeriveConnectionState_SlowCadenceIsNotPermanentlyStale(t *testing.T) {
|
||||
now := time.Now()
|
||||
interval := 5 * time.Minute
|
||||
|
||||
// A last success exactly one interval old is the steady state when the
|
||||
// connection check runs concurrently with the poll on the same tick; it
|
||||
// must derive active, not stale.
|
||||
h := healthEntry(ptrTime(now.Add(-interval)), "", "", "closed")
|
||||
state, _, _, _ := deriveConnectionState(true, h, now, interval)
|
||||
if state != ConnectionStateActive {
|
||||
t.Fatalf("state at 1x interval = %q, want %q", state, ConnectionStateActive)
|
||||
}
|
||||
|
||||
// Beyond 3x the interval the connection is genuinely stale.
|
||||
h = healthEntry(ptrTime(now.Add(-3*interval-time.Second)), "", "", "closed")
|
||||
state, _, _, _ = deriveConnectionState(true, h, now, interval)
|
||||
if state != ConnectionStateStale {
|
||||
t.Fatalf("state beyond 3x interval = %q, want %q", state, ConnectionStateStale)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildConnections_ScalesStaleThresholdWithPollingIntervals(t *testing.T) {
|
||||
now := time.Now()
|
||||
lastSuccess := now.Add(-5 * time.Minute)
|
||||
in := aggregatorInputs{
|
||||
pveInstances: []config.PVEInstance{{Name: "pve1", Host: "https://pve1.lan:8006"}},
|
||||
pbsInstances: []config.PBSInstance{{Name: "pbs1", Host: "https://pbs1.lan:8007"}},
|
||||
truenasInstances: []config.TrueNASInstance{
|
||||
{ID: "tn1", Name: "tn1", Host: "tn1.lan", Enabled: true, PollIntervalSecs: 300},
|
||||
},
|
||||
instanceHealth: map[string]monitoring.InstanceHealth{
|
||||
"pve::pve1": healthEntry(&lastSuccess, "", "", "closed"),
|
||||
"pbs::pbs1": healthEntry(&lastSuccess, "", "", "closed"),
|
||||
"truenas::tn1": healthEntry(&lastSuccess, "", "", "closed"),
|
||||
},
|
||||
pvePollingInterval: 5 * time.Minute,
|
||||
now: now,
|
||||
}
|
||||
got := buildConnections(in)
|
||||
byID := make(map[string]Connection, len(got))
|
||||
for _, conn := range got {
|
||||
byID[conn.ID] = conn
|
||||
}
|
||||
|
||||
if state := byID["pve:pve1"].State; state != ConnectionStateActive {
|
||||
t.Fatalf("pve state = %q, want active (threshold scaled by pvePollingInterval)", state)
|
||||
}
|
||||
if state := byID["truenas:tn1"].State; state != ConnectionStateActive {
|
||||
t.Fatalf("truenas state = %q, want active (threshold scaled by PollIntervalSecs)", state)
|
||||
}
|
||||
// PBS was left at the zero-value interval, so the 2-minute floor still
|
||||
// applies and the same 5-minute-old success reads as stale.
|
||||
if state := byID["pbs:pbs1"].State; state != ConnectionStateStale {
|
||||
t.Fatalf("pbs state = %q, want stale under the floor threshold", state)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -31,6 +31,9 @@ func buildAggregatorInputsWithRuntimeSources(
|
|||
inputs.pbsInstances = cfg.PBSInstances
|
||||
inputs.pmgInstances = cfg.PMGInstances
|
||||
inputs.apiTokens = append([]config.APITokenRecord(nil), cfg.APITokens...)
|
||||
inputs.pvePollingInterval = cfg.PVEPollingInterval
|
||||
inputs.pbsPollingInterval = cfg.PBSPollingInterval
|
||||
inputs.pmgPollingInterval = cfg.PMGPollingInterval
|
||||
}
|
||||
|
||||
if persistence != nil {
|
||||
|
|
|
|||
|
|
@ -1,11 +1,13 @@
|
|||
package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/config"
|
||||
)
|
||||
|
||||
// This file exercises the previously-uncovered branch arms of three helpers
|
||||
|
|
@ -272,3 +274,24 @@ func TestBranchcov0722UniqueMatch(t *testing.T) {
|
|||
}
|
||||
})
|
||||
}
|
||||
|
||||
// Regression for https://github.com/rcourtman/Pulse/issues/1620: the alert
|
||||
// snapshot path must thread the configured polling cadences into the
|
||||
// aggregator so slow cadences do not read as permanently stale.
|
||||
func TestBuildAggregatorInputsThreadsPollingIntervals(t *testing.T) {
|
||||
cfg := &config.Config{
|
||||
PVEPollingInterval: 5 * time.Minute,
|
||||
PBSPollingInterval: 4 * time.Minute,
|
||||
PMGPollingInterval: 3 * time.Minute,
|
||||
}
|
||||
inputs := buildAggregatorInputsWithRuntimeSources(context.Background(), cfg, nil, nil, aggregatorRuntimeSources{})
|
||||
if inputs.pvePollingInterval != 5*time.Minute {
|
||||
t.Fatalf("pvePollingInterval = %s, want 5m", inputs.pvePollingInterval)
|
||||
}
|
||||
if inputs.pbsPollingInterval != 4*time.Minute {
|
||||
t.Fatalf("pbsPollingInterval = %s, want 4m", inputs.pbsPollingInterval)
|
||||
}
|
||||
if inputs.pmgPollingInterval != 3*time.Minute {
|
||||
t.Fatalf("pmgPollingInterval = %s, want 3m", inputs.pmgPollingInterval)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue