diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index f60cd3507..53fed3dee 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -2444,3 +2444,23 @@ provider supplies that more specific evidence. `internal/monitoring/truenas_poller_test.go`, and `internal/monitoring/ceph_test.go` are the focused collection and identity proofs. + +### Cluster-endpoint discovery policy stays off the resolver on repeat polls + +The cluster-endpoint discovery-policy check (`clusterEndpointRuntimeURL` → +`clusterEndpointAllowedByDiscoveryPolicy` in +`internal/monitoring/monitor_cluster_helpers.go`) is a function of +configuration, not poll state, and must not generate per-poll DNS load. The +effective default policy — only the `NormalizeDiscoveryConfig`-injected +link-local blocklist `169.254.0.0/16` — is evaluated against literal endpoint +IPs only and never touches the resolver, because link-local addresses are not +legitimately served through DNS. Custom allowlist/blocklist policies memoize +their per-endpoint verdict for the shared 5-minute DNS-cache TTL, so hostname +endpoints resolve at most once per TTL window across poll cycles instead of per +node per cycle. SSH-based collectors in the same runtime follow the equivalent +rule for process spawning: `knownhosts` caches keyscan failures with doubling +backoff instead of re-executing `ssh-keyscan` every cycle, and the temperature +collector backs off per host after failed SSH collection instead of re-running +its two SSH probes every 10-second cycle. +`internal/monitoring/issue1638_dns_cache_test.go` is the registered proof that +repeat polls do not reach the raw resolver or re-exec the SSH probes. diff --git a/docs/release-control/v6/internal/subsystems/registry.json b/docs/release-control/v6/internal/subsystems/registry.json index 210b658a6..39429ddd8 100644 --- a/docs/release-control/v6/internal/subsystems/registry.json +++ b/docs/release-control/v6/internal/subsystems/registry.json @@ -1797,12 +1797,12 @@ "frontend-modern/src/utils/aiCostPresentation.ts", "frontend-modern/src/utils/aiProviderHealthPresentation.ts", "frontend-modern/src/utils/aiProviderPresentation.ts", - "internal/api/ai_handler.go", - "internal/api/ai_handlers.go", - "internal/api/ai_intelligence_handlers.go", - "internal/config/ai.go", - "internal/config/patrol_autopilot_persistence.go", - "pkg/aicontracts/action_broker.go", + "internal/api/ai_handler.go", + "internal/api/ai_handlers.go", + "internal/api/ai_intelligence_handlers.go", + "internal/config/ai.go", + "internal/config/patrol_autopilot_persistence.go", + "pkg/aicontracts/action_broker.go", "pkg/aicontracts/fix_execution.go", "pkg/aicontracts/investigation.go", "pkg/aicontracts/orchestrator_deps.go", @@ -2257,7 +2257,7 @@ "frontend-modern/src/features/alerts/__tests__/OverviewTab.total24h.test.tsx", "frontend-modern/src/features/alerts/__tests__/ThresholdsTab.test.tsx", "frontend-modern/src/features/alerts/__tests__/useAlertDestinationsTabState.test.tsx", - "frontend-modern/src/features/alerts/__tests__/useAlertOverridesState.test.tsx", + "frontend-modern/src/features/alerts/__tests__/useAlertOverridesState.test.tsx", "frontend-modern/src/features/alerts/AlertDeliveryHealthCard.test.tsx", "frontend-modern/src/features/alerts/identity.test.ts", "frontend-modern/src/features/alerts/thresholds/__tests__/helpers.test.ts", @@ -5633,6 +5633,7 @@ "internal/monitoring/issue1485_unraid_lifecycle_test.go", "internal/monitoring/issue1595_collection_trust_test.go", "internal/monitoring/issue1613_contract_test.go", + "internal/monitoring/issue1638_dns_cache_test.go", "internal/monitoring/monitor_additional_test.go", "internal/monitoring/monitor_alert_intent_test.go", "internal/monitoring/monitor_alert_override_migration_test.go", diff --git a/internal/monitoring/issue1638_dns_cache_test.go b/internal/monitoring/issue1638_dns_cache_test.go new file mode 100644 index 000000000..ecd39585a --- /dev/null +++ b/internal/monitoring/issue1638_dns_cache_test.go @@ -0,0 +1,194 @@ +package monitoring + +import ( + "context" + "errors" + "net" + "os" + "path/filepath" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" +) + +// Regression tests for discussion #1638: the cluster-endpoint discovery-policy +// check ran raw DNS lookups per node per poll cycle. The default policy (only +// the injected 169.254.0.0/16 blocklist) must never touch the resolver, and +// custom policies must memoize their verdict so repeat polls stay off DNS. + +func issue1638CountingLookup(t *testing.T, calls *int, ips map[string][]net.IP) { + t.Helper() + oldLookup := lookupIPFunc + lookupIPFunc = func(host string) ([]net.IP, error) { + *calls++ + if resolved, ok := ips[host]; ok { + return resolved, nil + } + return nil, errors.New("no such host") + } + t.Cleanup(func() { + lookupIPFunc = oldLookup + }) + resetDiscoveryPolicyDecisionCache() + t.Cleanup(resetDiscoveryPolicyDecisionCache) +} + +func TestIssue1638DefaultPolicySkipsDNSResolution(t *testing.T) { + calls := 0 + issue1638CountingLookup(t, &calls, map[string][]net.IP{ + "node-a.example.com": {net.ParseIP("192.168.1.5")}, + }) + + // NormalizeDiscoveryConfig injects the default link-local blocklist, so + // this is what every install without an explicit policy runs with. + discoveryCfg := config.NormalizeDiscoveryConfig(config.DiscoveryConfig{}) + endpoint := config.ClusterEndpoint{NodeName: "node-a", Host: "node-a.example.com"} + + for poll := 0; poll < 5; poll++ { + got := clusterEndpointRuntimeURL(endpoint, true, false, discoveryCfg) + if got != "https://node-a.example.com:8006" { + t.Fatalf("poll %d: runtime URL = %q, want %q", poll, got, "https://node-a.example.com:8006") + } + } + + if calls != 0 { + t.Fatalf("default link-local-only policy hit the resolver %d times, want 0", calls) + } +} + +func TestIssue1638DefaultPolicyStillBlocksLiteralLinkLocal(t *testing.T) { + calls := 0 + issue1638CountingLookup(t, &calls, nil) + + discoveryCfg := config.NormalizeDiscoveryConfig(config.DiscoveryConfig{}) + endpoint := config.ClusterEndpoint{NodeName: "node-b", IP: "169.254.10.20"} + + if got := clusterEndpointRuntimeURL(endpoint, false, false, discoveryCfg); got != "" { + t.Fatalf("literal link-local endpoint allowed through default blocklist: %q", got) + } + if calls != 0 { + t.Fatalf("literal IP evaluation hit the resolver %d times, want 0", calls) + } +} + +func TestIssue1638CustomPolicyResolvesOncePerEndpointAcrossPolls(t *testing.T) { + calls := 0 + issue1638CountingLookup(t, &calls, map[string][]net.IP{ + "allowed.local": {net.ParseIP("10.0.0.10")}, + "blocked.local": {net.ParseIP("192.168.1.10")}, + }) + + discoveryCfg := config.NormalizeDiscoveryConfig(config.DiscoveryConfig{ + SubnetAllowlist: []string{"10.0.0.0/8"}, + }) + allowed := config.ClusterEndpoint{NodeName: "node-a", Host: "allowed.local"} + blocked := config.ClusterEndpoint{NodeName: "node-b", Host: "blocked.local"} + + for poll := 0; poll < 20; poll++ { + if got := clusterEndpointRuntimeURL(allowed, true, false, discoveryCfg); got != "https://allowed.local:8006" { + t.Fatalf("poll %d: allowed endpoint URL = %q", poll, got) + } + if got := clusterEndpointRuntimeURL(blocked, true, false, discoveryCfg); got != "" { + t.Fatalf("poll %d: blocked endpoint unexpectedly allowed: %q", poll, got) + } + } + + if calls != 2 { + t.Fatalf("repeat polls hit the resolver %d times, want exactly 2 (one per endpoint host)", calls) + } +} + +func TestIssue1638CustomPolicyDecisionExpiresAfterTTL(t *testing.T) { + calls := 0 + issue1638CountingLookup(t, &calls, map[string][]net.IP{ + "allowed.local": {net.ParseIP("10.0.0.10")}, + }) + + baseTime := time.Now() + oldNow := discoveryPolicyTimeNow + discoveryPolicyTimeNow = func() time.Time { return baseTime } + t.Cleanup(func() { + discoveryPolicyTimeNow = oldNow + }) + + discoveryCfg := config.NormalizeDiscoveryConfig(config.DiscoveryConfig{ + SubnetAllowlist: []string{"10.0.0.0/8"}, + }) + endpoint := config.ClusterEndpoint{NodeName: "node-a", Host: "allowed.local"} + + clusterEndpointRuntimeURL(endpoint, true, false, discoveryCfg) + clusterEndpointRuntimeURL(endpoint, true, false, discoveryCfg) + if calls != 1 { + t.Fatalf("resolver hit %d times inside TTL, want 1", calls) + } + + discoveryPolicyTimeNow = func() time.Time { return baseTime.Add(discoveryPolicyDecisionTTL + time.Second) } + clusterEndpointRuntimeURL(endpoint, true, false, discoveryCfg) + if calls != 2 { + t.Fatalf("resolver hit %d times after TTL expiry, want 2", calls) + } +} + +func TestIssue1638KeyscanFailureIsNotRetriedEveryCycle(t *testing.T) { + scans := 0 + manager, err := NewKnownHostsManager( + filepath.Join(t.TempDir(), "known_hosts"), + WithKeyscanFunc(func(ctx context.Context, host string, port int, timeout time.Duration) ([]byte, error) { + scans++ + return nil, errors.New("connection refused") + }), + ) + if err != nil { + t.Fatalf("NewKnownHostsManager: %v", err) + } + + for cycle := 0; cycle < 5; cycle++ { + if err := manager.Ensure(context.Background(), "unreachable.local"); err == nil { + t.Fatalf("cycle %d: expected error from failing keyscan", cycle) + } + } + + if scans != 1 { + t.Fatalf("failing host was keyscanned %d times within backoff window, want 1", scans) + } +} + +type issue1638CountingRunner struct { + runs int +} + +func (r *issue1638CountingRunner) Run(ctx context.Context, name string, args ...string) ([]byte, error) { + r.runs++ + return nil, errors.New("connection refused") +} + +func TestIssue1638TemperatureSSHFailureBacksOff(t *testing.T) { + keyPath := filepath.Join(t.TempDir(), "id_ed25519_test") + if err := os.WriteFile(keyPath, []byte("dummy"), 0o600); err != nil { + t.Fatalf("write key: %v", err) + } + + tc := NewTemperatureCollectorWithPort("root", keyPath, 22) + tc.hostKeys = nil + runner := &issue1638CountingRunner{} + tc.runner = runner + + // First cycle attempts both the sensors and the RPi fallback command. + if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { + t.Fatalf("CollectTemperature: %v", err) + } + if runner.runs != 2 { + t.Fatalf("first cycle ran %d ssh commands, want 2", runner.runs) + } + + // Subsequent cycles inside the backoff window must not exec ssh again. + for cycle := 0; cycle < 5; cycle++ { + if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { + t.Fatalf("cycle %d: CollectTemperature: %v", cycle, err) + } + } + if runner.runs != 2 { + t.Fatalf("backoff window still ran ssh, total %d commands, want 2", runner.runs) + } +} diff --git a/internal/monitoring/knownhosts.go b/internal/monitoring/knownhosts.go index b42ac9bd6..c8608b2da 100644 --- a/internal/monitoring/knownhosts.go +++ b/internal/monitoring/knownhosts.go @@ -33,15 +33,27 @@ type KnownHostsManager interface { type knownHostsManager struct { path string cache map[string]struct{} + failures map[string]*keyscanFailure mu sync.Mutex keyscanFn keyscanFunc keyscanTimeout time.Duration } +// keyscanFailure remembers a failed ssh-keyscan so poll cycles don't re-exec +// it every pass against a host that keeps refusing (#1638). Backoff doubles +// per consecutive failure and any success clears the entry. +type keyscanFailure struct { + retryAt time.Time + backoff time.Duration + err error +} + type keyscanFunc func(ctx context.Context, host string, port int, timeout time.Duration) ([]byte, error) const ( - defaultKeyscanTimeout = 5 * time.Second + defaultKeyscanTimeout = 5 * time.Second + keyscanFailureInitialBackoff = 30 * time.Second + keyscanFailureMaxBackoff = 15 * time.Minute ) var ( @@ -117,6 +129,7 @@ func NewKnownHostsManager(path string, opts ...KnownHostsOption) (KnownHostsMana m := &knownHostsManager{ path: path, cache: make(map[string]struct{}), + failures: make(map[string]*keyscanFailure), keyscanFn: defaultKeyscan, keyscanTimeout: defaultKeyscanTimeout, } @@ -150,24 +163,54 @@ func (m *knownHostsManager) EnsureWithPort(ctx context.Context, host string, por cacheKey := fmt.Sprintf("%s:%d", host, port) m.mu.Lock() _, cached := m.cache[cacheKey] - m.mu.Unlock() if cached { + m.mu.Unlock() return nil } + if failure := m.failures[cacheKey]; failure != nil && time.Now().Before(failure.retryAt) { + err := failure.err + m.mu.Unlock() + return fmt.Errorf("knownhosts: ssh-keyscan for %s:%d suppressed until backoff expires: %w", host, port, err) + } + m.mu.Unlock() keyData, err := m.keyscanFn(ctx, host, port, m.keyscanTimeout) if err != nil { - return fmt.Errorf("knownhosts: ssh-keyscan failed for %s:%d: %w", host, port, err) + wrapped := fmt.Errorf("knownhosts: ssh-keyscan failed for %s:%d: %w", host, port, err) + m.recordKeyscanFailure(cacheKey, wrapped) + return wrapped } entries := sanitizeKeyscanOutput(hostSpec, keyData) if len(entries) == 0 { - return fmt.Errorf("%w for %s:%d", ErrNoHostKeys, host, port) + wrapped := fmt.Errorf("%w for %s:%d", ErrNoHostKeys, host, port) + m.recordKeyscanFailure(cacheKey, wrapped) + return wrapped } return m.EnsureWithEntries(ctx, host, port, entries) } +func (m *knownHostsManager) recordKeyscanFailure(cacheKey string, err error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.failures == nil { + m.failures = make(map[string]*keyscanFailure) + } + backoff := keyscanFailureInitialBackoff + if existing := m.failures[cacheKey]; existing != nil { + backoff = existing.backoff * 2 + if backoff > keyscanFailureMaxBackoff { + backoff = keyscanFailureMaxBackoff + } + } + m.failures[cacheKey] = &keyscanFailure{ + retryAt: time.Now().Add(backoff), + backoff: backoff, + err: err, + } +} + // EnsureWithEntries installs the provided host key entries for host:port. func (m *knownHostsManager) EnsureWithEntries(ctx context.Context, host string, port int, entries [][]byte) error { if strings.TrimSpace(host) == "" { @@ -226,6 +269,7 @@ func (m *knownHostsManager) EnsureWithEntries(ctx context.Context, host string, } m.cache[cacheKey] = struct{}{} + delete(m.failures, cacheKey) return nil } diff --git a/internal/monitoring/monitor_additional_test.go b/internal/monitoring/monitor_additional_test.go index dded09cf9..468e1cf25 100644 --- a/internal/monitoring/monitor_additional_test.go +++ b/internal/monitoring/monitor_additional_test.go @@ -293,6 +293,8 @@ func TestClusterEndpointEffectiveURL(t *testing.T) { } func TestBuildClusterEndpointsForInit_RespectsDiscoveryPolicy(t *testing.T) { + resetDiscoveryPolicyDecisionCache() + t.Cleanup(resetDiscoveryPolicyDecisionCache) oldLookup := lookupIPFunc lookupIPFunc = func(host string) ([]net.IP, error) { switch host { diff --git a/internal/monitoring/monitor_cluster_helpers.go b/internal/monitoring/monitor_cluster_helpers.go index a98cc486f..19379d1ee 100644 --- a/internal/monitoring/monitor_cluster_helpers.go +++ b/internal/monitoring/monitor_cluster_helpers.go @@ -4,6 +4,8 @@ import ( "net" "net/url" "strings" + "sync" + "time" "github.com/rcourtman/pulse-go-rewrite/internal/config" ) @@ -225,20 +227,93 @@ func discoveryPolicyIPsForEndpointHost(candidateURL string) []net.IP { return filtered } -func clusterEndpointAllowedByDiscoveryPolicy(endpoint config.ClusterEndpoint, candidateURL string, discoveryCfg config.DiscoveryConfig) bool { - if len(discoveryCfg.SubnetAllowlist) == 0 && len(discoveryCfg.SubnetBlocklist) == 0 && len(discoveryCfg.IPBlocklist) == 0 { - return true - } +// discoveryPolicyDecisionTTL bounds how long a cached discovery-policy verdict +// (and the DNS answer behind it) is reused before re-evaluating. It matches the +// tlsutil DNS cache refresh interval so policy decisions never trail the +// resolver view used for actual connections by more than one refresh. +const discoveryPolicyDecisionTTL = 5 * time.Minute +// discoveryPolicyDecisionCacheLimit caps the decision cache. Keys derive from +// configured endpoints and the discovery policy, so the map stays tiny in +// practice; the cap only guards against pathological configs. +const discoveryPolicyDecisionCacheLimit = 1024 + +type discoveryPolicyDecision struct { + allowed bool + expiresAt time.Time +} + +var ( + discoveryPolicyDecisionMu sync.Mutex + discoveryPolicyDecisionCache = map[string]discoveryPolicyDecision{} + discoveryPolicyTimeNow = time.Now +) + +func resetDiscoveryPolicyDecisionCache() { + discoveryPolicyDecisionMu.Lock() + discoveryPolicyDecisionCache = map[string]discoveryPolicyDecision{} + discoveryPolicyDecisionMu.Unlock() +} + +// discoveryPolicyIsDefaultOnly reports whether the effective policy consists +// solely of the injected default link-local blocklist. NormalizeDiscoveryConfig +// always adds 169.254.0.0/16, so every install has a non-empty policy and the +// zero-policy fast path above never fires (#1638). Link-local addresses are +// not routable across segments, so a hostname is never legitimately served by +// one; only literal link-local IPs need blocking, which requires no DNS. +func discoveryPolicyIsDefaultOnly(cfg config.DiscoveryConfig) bool { + if len(cfg.SubnetAllowlist) != 0 || len(cfg.IPBlocklist) != 0 { + return false + } + defaults := config.DefaultDiscoveryConfig().SubnetBlocklist + defaultSet := make(map[string]struct{}, len(defaults)) + for _, cidr := range defaults { + defaultSet[strings.TrimSpace(cidr)] = struct{}{} + } + for _, cidr := range cfg.SubnetBlocklist { + if _, ok := defaultSet[strings.TrimSpace(cidr)]; !ok { + return false + } + } + return true +} + +// discoveryPolicyLiteralIPs returns the IPs knowable for an endpoint without +// touching the resolver: a literal IP in the candidate URL, else the +// endpoint's recorded effective IP. +func discoveryPolicyLiteralIPs(endpoint config.ClusterEndpoint, candidateURL string) []net.IP { + if host := normalizeEndpointHost(candidateURL); host != "" { + if ip := net.ParseIP(host); ip != nil { + return []net.IP{ip} + } + } + if ip := net.ParseIP(strings.TrimSpace(endpoint.EffectiveIP())); ip != nil { + return []net.IP{ip} + } + return nil +} + +func discoveryPolicyDecisionKey(endpoint config.ClusterEndpoint, candidateURL string, cfg config.DiscoveryConfig) string { + parts := []string{ + candidateURL, + strings.TrimSpace(endpoint.EffectiveIP()), + strings.Join(cfg.SubnetAllowlist, ","), + strings.Join(cfg.SubnetBlocklist, ","), + strings.Join(cfg.IPBlocklist, ","), + } + return strings.Join(parts, "|") +} + +// evaluateClusterEndpointDiscoveryPolicy is the uncached policy check, +// including DNS resolution of hostname endpoints. +func evaluateClusterEndpointDiscoveryPolicy(endpoint config.ClusterEndpoint, candidateURL string, discoveryCfg config.DiscoveryConfig) bool { allowlist := discoveryPolicyCIDRs(discoveryCfg.SubnetAllowlist) blocklist := discoveryPolicyCIDRs(discoveryCfg.SubnetBlocklist) blockedIPs := discoveryPolicyBlockedIPs(discoveryCfg.IPBlocklist) resolvedIPs := discoveryPolicyIPsForEndpointHost(candidateURL) if len(resolvedIPs) == 0 { - if ip := net.ParseIP(strings.TrimSpace(endpoint.EffectiveIP())); ip != nil { - resolvedIPs = []net.IP{ip} - } + resolvedIPs = discoveryPolicyLiteralIPs(endpoint, candidateURL) } if len(resolvedIPs) == 0 { return true @@ -253,6 +328,51 @@ func clusterEndpointAllowedByDiscoveryPolicy(endpoint config.ClusterEndpoint, ca return true } +func clusterEndpointAllowedByDiscoveryPolicy(endpoint config.ClusterEndpoint, candidateURL string, discoveryCfg config.DiscoveryConfig) bool { + if len(discoveryCfg.SubnetAllowlist) == 0 && len(discoveryCfg.SubnetBlocklist) == 0 && len(discoveryCfg.IPBlocklist) == 0 { + return true + } + + // The policy is a function of configuration, not poll state, but since + // c5f5af7ab it is re-evaluated per node per poll cycle. With the default + // link-local-only blocklist no resolution is needed at all, and custom + // policies memoize their verdict so repeat polls stay off the resolver + // (#1638). + if discoveryPolicyIsDefaultOnly(discoveryCfg) { + blocklist := discoveryPolicyCIDRs(discoveryCfg.SubnetBlocklist) + for _, ip := range discoveryPolicyLiteralIPs(endpoint, candidateURL) { + if !discoveryPolicyAllowsIP(ip, nil, blocklist, nil) { + return false + } + } + return true + } + + key := discoveryPolicyDecisionKey(endpoint, candidateURL, discoveryCfg) + now := discoveryPolicyTimeNow() + + discoveryPolicyDecisionMu.Lock() + if cached, ok := discoveryPolicyDecisionCache[key]; ok && now.Before(cached.expiresAt) { + discoveryPolicyDecisionMu.Unlock() + return cached.allowed + } + discoveryPolicyDecisionMu.Unlock() + + allowed := evaluateClusterEndpointDiscoveryPolicy(endpoint, candidateURL, discoveryCfg) + + discoveryPolicyDecisionMu.Lock() + if len(discoveryPolicyDecisionCache) >= discoveryPolicyDecisionCacheLimit { + discoveryPolicyDecisionCache = map[string]discoveryPolicyDecision{} + } + discoveryPolicyDecisionCache[key] = discoveryPolicyDecision{ + allowed: allowed, + expiresAt: now.Add(discoveryPolicyDecisionTTL), + } + discoveryPolicyDecisionMu.Unlock() + + return allowed +} + func clusterEndpointRuntimeURL(endpoint config.ClusterEndpoint, verifySSL bool, hasFingerprint bool, discoveryCfg config.DiscoveryConfig) string { candidateURL := clusterEndpointEffectiveURL(endpoint, verifySSL, hasFingerprint) if candidateURL == "" { diff --git a/internal/monitoring/temperature.go b/internal/monitoring/temperature.go index 624600674..de4f282fc 100644 --- a/internal/monitoring/temperature.go +++ b/internal/monitoring/temperature.go @@ -12,6 +12,7 @@ import ( "path/filepath" "strconv" "strings" + "sync" "sync/atomic" "time" @@ -87,6 +88,54 @@ type TemperatureCollector struct { hostKeys KnownHostsManager missingKeyWarned atomic.Bool runner CommandRunner + sshFailureMu sync.Mutex + sshFailures map[string]*temperatureSSHFailure +} + +// temperatureSSHFailure remembers a host whose SSH collection failed so +// subsequent 10s poll cycles don't re-exec ssh (twice, with the RPi fallback) +// against a host that keeps failing identically (#1638). Backoff doubles per +// consecutive failure; any successful collection clears it. +type temperatureSSHFailure struct { + retryAt time.Time + backoff time.Duration +} + +const ( + temperatureSSHFailureInitialBackoff = 30 * time.Second + temperatureSSHFailureMaxBackoff = 15 * time.Minute +) + +func (tc *TemperatureCollector) inSSHFailureBackoff(host string) bool { + tc.sshFailureMu.Lock() + defer tc.sshFailureMu.Unlock() + failure := tc.sshFailures[host] + return failure != nil && time.Now().Before(failure.retryAt) +} + +func (tc *TemperatureCollector) recordSSHFailure(host string) { + tc.sshFailureMu.Lock() + defer tc.sshFailureMu.Unlock() + if tc.sshFailures == nil { + tc.sshFailures = make(map[string]*temperatureSSHFailure) + } + backoff := temperatureSSHFailureInitialBackoff + if existing := tc.sshFailures[host]; existing != nil { + backoff = existing.backoff * 2 + if backoff > temperatureSSHFailureMaxBackoff { + backoff = temperatureSSHFailureMaxBackoff + } + } + tc.sshFailures[host] = &temperatureSSHFailure{ + retryAt: time.Now().Add(backoff), + backoff: backoff, + } +} + +func (tc *TemperatureCollector) clearSSHFailure(host string) { + tc.sshFailureMu.Lock() + defer tc.sshFailureMu.Unlock() + delete(tc.sshFailures, host) } // NewTemperatureCollectorWithPort creates a new temperature collector with custom SSH port @@ -151,6 +200,14 @@ func (tc *TemperatureCollector) CollectTemperature(ctx context.Context, nodeHost return &models.Temperature{Available: false}, nil } + if tc.inSSHFailureBackoff(host) { + log.Debug(). + Str("node", nodeName). + Str("host", host). + Msg("Skipping SSH temperature collection while failure backoff is active") + return &models.Temperature{Available: false}, nil + } + // Direct SSH (legacy method for non-containerized deployments). // New setup scripts restrict the key to /usr/local/sbin/pulse-sensors, which emits // the canonical {sensors, smart} payload. Older keys ignore the requested command @@ -158,6 +215,7 @@ func (tc *TemperatureCollector) CollectTemperature(ctx context.Context, nodeHost output, err := tc.runSSHCommand(ctx, host, pulseSensorsSSHCommand) if err != nil || strings.TrimSpace(output) == "" { if tc.disableLegacySSHOnAuthFailure(err, nodeName, host) { + tc.recordSSHFailure(host) return &models.Temperature{Available: false}, nil } @@ -167,11 +225,13 @@ func (tc *TemperatureCollector) CollectTemperature(ctx context.Context, nodeHost // Parse RPi temperature format temp, parseErr := tc.parseRPiTemperature(output) if parseErr == nil { + tc.clearSSHFailure(host) return temp, nil } } if tc.disableLegacySSHOnAuthFailure(err, nodeName, host) { + tc.recordSSHFailure(host) return &models.Temperature{Available: false}, nil } @@ -180,9 +240,12 @@ func (tc *TemperatureCollector) CollectTemperature(ctx context.Context, nodeHost Str("host", host). Err(err). Msg("Failed to collect temperature data via SSH (tried both lm-sensors and RPi methods)") + tc.recordSSHFailure(host) return &models.Temperature{Available: false}, nil } + tc.clearSSHFailure(host) + // Parse sensors JSON output temp, err := tc.parseSensorsJSON(output) if err != nil {