From b45bd66b944fbad4911c58eb053f79375409355d Mon Sep 17 00:00:00 2001 From: "courtmanr@gmail.com" Date: Tue, 28 Jul 2026 12:07:19 +0100 Subject: [PATCH] Route discovery policy DNS through the cached resolver (#1638) The discovery-policy check resolved endpoint hostnames with a bare net.LookupIP while the actual dials went through pkg/tlsutil's process-global cached resolver, so the policy and the connection reasoned about two different DNS views. That split is why 108aa4e20 had to skip resolution entirely for the default policy, which left the injected 169.254.0.0/16 blocklist enforced only against literal IPs: a hostname endpoint pointed at the metadata range walked straight through. Resolve through tlsutil.LookupHostCached instead. The shared resolver caches answers and lookup failures alike until its next refresh, so repeat poll cycles cost a cache hit rather than a query and the per-poll DNS volume that opened #1638 stays gone. With that in place the default-policy skip is removed and the blocklist applies to resolved addresses again, and the five-minute decision cache is dropped rather than kept: it bought nothing on top of the resolver cache, made the verdict trail the configuration, and memoized the fail-open "resolution failed, allow" outcome for minutes even with an explicit allowlist configured. Its claim to match a DNS refresh interval that operators configure through DNS_CACHE_TIMEOUT goes with it. The SSH backoffs now only escalate for work that ran. A knownhosts manager suppressing a call inside its own window reports ErrKeyscanSuppressed, and the temperature layer neither records a failure nor pays for the RPi fallback in that case. An expired collection deadline is our own budget rather than evidence about the host, so it holds the window at the floor. Both backoffs decay once a retry deadline is more than one window past, and replacing the temperature SSH key on disk clears both maps so a repaired key is tried on the next cycle instead of after fifteen minutes. Refs discussion #1638. Co-Authored-By: Claude Fable 5 --- .../v6/internal/subsystems/monitoring.md | 49 ++- .../monitoring/issue1638_dns_cache_test.go | 399 ++++++++++++++++-- internal/monitoring/knownhosts.go | 72 +++- .../monitoring/monitor_additional_test.go | 2 - .../monitoring/monitor_cluster_helpers.go | 143 ++----- internal/monitoring/temperature.go | 148 ++++++- pkg/tlsutil/dnscache.go | 33 ++ 7 files changed, 659 insertions(+), 187 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 0b045e3d5..8bf0dae11 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -2477,17 +2477,40 @@ proofs. 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. +configuration, not poll state, and must not generate per-poll DNS load. It +resolves hostname endpoints through the process-global cached resolver that +`pkg/tlsutil` dials with (`tlsutil.LookupHostCached`), never through a bare +`net.LookupIP`. That gives the policy and the connection one DNS view, and +because the resolver caches lookup failures as well as answers, repeat poll +cycles cost a cache hit rather than a query — roughly one query per endpoint +host per DNS cache refresh, whose interval operators set through +`DNS_CACHE_TIMEOUT`. There is deliberately no second, policy-level verdict +cache: it would buy nothing on top of the resolver cache, would make the +verdict trail the configuration, and would freeze a fail-open +resolution-failed verdict in place for the length of its window. The effective +default policy — the `NormalizeDiscoveryConfig`-injected link-local blocklist +`169.254.0.0/16` — is therefore enforced against resolved addresses too, so a +hostname endpoint pointed into the link-local range is rejected rather than +allowed through unresolved. Literal-IP endpoints are still evaluated without +any resolution. + +SSH-based collectors in the same runtime follow the equivalent rule for +process spawning, and only escalate for work that actually ran. `knownhosts` +caches keyscan failures with doubling backoff instead of re-executing +`ssh-keyscan` every cycle, and reports a suppressed call as +`ErrKeyscanSuppressed` so callers can tell it apart from a refusal. The +temperature collector backs off per host after failed SSH collection instead of +re-running its two SSH probes every 10-second cycle, but leaves its backoff +untouched when the host key scan was suppressed (no ssh was executed) and holds +it at the floor when the collection deadline expired rather than compounding on +evidence about Pulse's own budget rather than the host. Both backoffs decay: a +failure whose retry deadline passed more than one window ago restarts at the +floor instead of resuming the ceiling. Neither backoff may be a trap the +operator cannot leave: replacing the temperature SSH key on disk clears both +maps (`TemperatureCollector.ResetSSHFailures`, triggered from the per-cycle key +change check), so repairing the key is retried on the next cycle rather than +after a window that has compounded to fifteen minutes. `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. +repeat polls stay on the DNS cache, that the link-local blocklist still rejects +hostname endpoints resolving into it, and that the SSH backoffs suppress, +decay, and reset as described. diff --git a/internal/monitoring/issue1638_dns_cache_test.go b/internal/monitoring/issue1638_dns_cache_test.go index ecd39585a..0b6bae512 100644 --- a/internal/monitoring/issue1638_dns_cache_test.go +++ b/internal/monitoring/issue1638_dns_cache_test.go @@ -3,19 +3,25 @@ package monitoring import ( "context" "errors" + "fmt" "net" "os" "path/filepath" + "sync/atomic" "testing" "time" "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/pkg/tlsutil" ) // 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. +// check ran raw DNS lookups per node per poll cycle, and the SSH collectors +// re-executed ssh against hosts that kept failing. The policy check must now go +// through the shared cached resolver — which keeps DNS at roughly one query per +// host per cache refresh while still enforcing the blocklist against resolved +// addresses — and the SSH backoffs must only escalate for work that actually +// ran. func issue1638CountingLookup(t *testing.T, calls *int, ips map[string][]net.IP) { t.Helper() @@ -30,34 +36,57 @@ func issue1638CountingLookup(t *testing.T, calls *int, ips map[string][]net.IP) t.Cleanup(func() { lookupIPFunc = oldLookup }) - resetDiscoveryPolicyDecisionCache() - t.Cleanup(resetDiscoveryPolicyDecisionCache) } -func TestIssue1638DefaultPolicySkipsDNSResolution(t *testing.T) { +// issue1638FakeClock pins the SSH backoff clock so compounding, decay, and +// expiry are observable without sleeping. +type issue1638FakeClock struct { + now time.Time +} + +func (c *issue1638FakeClock) advance(d time.Duration) { + c.now = c.now.Add(d) +} + +func issue1638UseFakeClock(t *testing.T) *issue1638FakeClock { + t.Helper() + clock := &issue1638FakeClock{now: time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC)} + old := sshBackoffNow + sshBackoffNow = func() time.Time { return clock.now } + t.Cleanup(func() { + sshBackoffNow = old + }) + return clock +} + +// TestIssue1638DefaultPolicyBlocksHostnameResolvingToLinkLocal pins the +// blocklist enforcement that the first fix traded away. The default policy is +// just the injected 169.254.0.0/16 blocklist, and it has to apply to whatever a +// hostname endpoint resolves to, not only to literal link-local IPs — otherwise +// a name pointed at the metadata range walks straight through. +func TestIssue1638DefaultPolicyBlocksHostnameResolvingToLinkLocal(t *testing.T) { calls := 0 issue1638CountingLookup(t, &calls, map[string][]net.IP{ - "node-a.example.com": {net.ParseIP("192.168.1.5")}, + "imds.example.com": {net.ParseIP("169.254.169.254")}, + "node-a.local": {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") - } + blocked := config.ClusterEndpoint{NodeName: "node-imds", Host: "imds.example.com"} + if got := clusterEndpointRuntimeURL(blocked, true, false, discoveryCfg); got != "" { + t.Fatalf("hostname resolving into 169.254.0.0/16 was allowed: %q", got) } - if calls != 0 { - t.Fatalf("default link-local-only policy hit the resolver %d times, want 0", calls) + allowed := config.ClusterEndpoint{NodeName: "node-a", Host: "node-a.local"} + if got := clusterEndpointRuntimeURL(allowed, true, false, discoveryCfg); got != "https://node-a.local:8006" { + t.Fatalf("ordinary hostname endpoint URL = %q, want %q", got, "https://node-a.local:8006") } } -func TestIssue1638DefaultPolicyStillBlocksLiteralLinkLocal(t *testing.T) { +func TestIssue1638DefaultPolicyStillBlocksLiteralLinkLocalWithoutDNS(t *testing.T) { calls := 0 issue1638CountingLookup(t, &calls, nil) @@ -72,7 +101,7 @@ func TestIssue1638DefaultPolicyStillBlocksLiteralLinkLocal(t *testing.T) { } } -func TestIssue1638CustomPolicyResolvesOncePerEndpointAcrossPolls(t *testing.T) { +func TestIssue1638CustomPolicyEnforcedPerPoll(t *testing.T) { calls := 0 issue1638CountingLookup(t, &calls, map[string][]net.IP{ "allowed.local": {net.ParseIP("10.0.0.10")}, @@ -93,44 +122,73 @@ func TestIssue1638CustomPolicyResolvesOncePerEndpointAcrossPolls(t *testing.T) { 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")}, - }) +// issue1638CountingResolver counts the DNS queries the shared cached resolver +// actually issues, as opposed to the lookups the policy check asks for. +type issue1638CountingResolver struct { + queries int32 + answers map[string][]string +} - baseTime := time.Now() - oldNow := discoveryPolicyTimeNow - discoveryPolicyTimeNow = func() time.Time { return baseTime } +func (r *issue1638CountingResolver) LookupHost(_ context.Context, host string) ([]string, error) { + atomic.AddInt32(&r.queries, 1) + if answer, ok := r.answers[host]; ok { + return answer, nil + } + return nil, &net.DNSError{Err: "no such host", Name: host, IsNotFound: true} +} + +func (r *issue1638CountingResolver) LookupAddr(_ context.Context, addr string) ([]string, error) { + atomic.AddInt32(&r.queries, 1) + return nil, &net.DNSError{Err: "no such host", Name: addr, IsNotFound: true} +} + +// TestIssue1638PolicyLookupsStayCachedAcrossPolls is the #1638 win itself: +// with the decision cache gone, repeat poll cycles must still not generate DNS +// traffic, because the policy check resolves through the same process-global +// cached resolver that pkg/tlsutil dials with. +func TestIssue1638PolicyLookupsStayCachedAcrossPolls(t *testing.T) { + // Names unique to this test so the process-global cache cannot be warmed + // or poisoned by another test in the package. + const ( + allowedHost = "issue1638-allowed.invalid" + blockedHost = "issue1638-blocked.invalid" + ) + + counting := &issue1638CountingResolver{answers: map[string][]string{ + allowedHost: {"10.0.0.10"}, + blockedHost: {"169.254.169.254"}, + }} + + resolver := tlsutil.GetDNSResolver() + previous := resolver.Resolver + resolver.Resolver = counting t.Cleanup(func() { - discoveryPolicyTimeNow = oldNow + resolver.Resolver = previous }) - discoveryCfg := config.NormalizeDiscoveryConfig(config.DiscoveryConfig{ - SubnetAllowlist: []string{"10.0.0.0/8"}, - }) - endpoint := config.ClusterEndpoint{NodeName: "node-a", Host: "allowed.local"} + discoveryCfg := config.NormalizeDiscoveryConfig(config.DiscoveryConfig{}) + allowed := config.ClusterEndpoint{NodeName: "node-a", Host: allowedHost} + blocked := config.ClusterEndpoint{NodeName: "node-b", Host: blockedHost} - clusterEndpointRuntimeURL(endpoint, true, false, discoveryCfg) - clusterEndpointRuntimeURL(endpoint, true, false, discoveryCfg) - if calls != 1 { - t.Fatalf("resolver hit %d times inside TTL, want 1", calls) + for poll := 0; poll < 20; poll++ { + if got := clusterEndpointRuntimeURL(allowed, true, false, discoveryCfg); got != "https://"+allowedHost+":8006" { + t.Fatalf("poll %d: allowed endpoint URL = %q", poll, got) + } + if got := clusterEndpointRuntimeURL(blocked, true, false, discoveryCfg); got != "" { + t.Fatalf("poll %d: link-local endpoint unexpectedly allowed: %q", poll, got) + } } - 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) + if queries := atomic.LoadInt32(&counting.queries); queries != 2 { + t.Fatalf("20 poll cycles issued %d DNS queries, want exactly 2 (one per endpoint host)", queries) } } func TestIssue1638KeyscanFailureIsNotRetriedEveryCycle(t *testing.T) { + clock := issue1638UseFakeClock(t) + scans := 0 manager, err := NewKnownHostsManager( filepath.Join(t.TempDir(), "known_hosts"), @@ -152,18 +210,92 @@ func TestIssue1638KeyscanFailureIsNotRetriedEveryCycle(t *testing.T) { if scans != 1 { t.Fatalf("failing host was keyscanned %d times within backoff window, want 1", scans) } + + // Suppressed calls must be distinguishable from a real refusal so callers + // don't escalate their own backoff for work that never ran. + suppressed := manager.Ensure(context.Background(), "unreachable.local") + if !errors.Is(suppressed, ErrKeyscanSuppressed) { + t.Fatalf("suppressed Ensure error = %v, want it to wrap ErrKeyscanSuppressed", suppressed) + } + + // Resetting the manager (as applying settings does) retries immediately + // rather than waiting out the window. + manager.ResetFailures() + if err := manager.Ensure(context.Background(), "unreachable.local"); err == nil { + t.Fatal("expected error from failing keyscan after reset") + } + if scans != 2 { + t.Fatalf("reset did not retry the keyscan, scans = %d, want 2", scans) + } + + // Backoff compounds only while failures keep arriving inside the window. + clock.advance(keyscanFailureInitialBackoff + time.Second) + if err := manager.Ensure(context.Background(), "unreachable.local"); err == nil { + t.Fatal("expected error once the first backoff window expired") + } + if scans != 3 { + t.Fatalf("expired backoff did not retry, scans = %d, want 3", scans) + } + clock.advance(keyscanFailureInitialBackoff + time.Second) + if err := manager.Ensure(context.Background(), "unreachable.local"); !errors.Is(err, ErrKeyscanSuppressed) { + t.Fatalf("second failure should have doubled the window, got %v", err) + } + if scans != 3 { + t.Fatalf("doubled backoff window still re-scanned, scans = %d, want 3", scans) + } +} + +func TestIssue1638KeyscanTimeoutDoesNotCompound(t *testing.T) { + clock := issue1638UseFakeClock(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, fmt.Errorf("ssh-keyscan gave up: %w", context.DeadlineExceeded) + }), + ) + if err != nil { + t.Fatalf("NewKnownHostsManager: %v", err) + } + + // Three consecutive timeouts, each one window apart. A compounding backoff + // would have suppressed the third; a timeout is not evidence about the + // host, so the window stays at the floor. + for cycle := 0; cycle < 3; cycle++ { + if err := manager.Ensure(context.Background(), "slow.local"); err == nil { + t.Fatalf("cycle %d: expected timeout error", cycle) + } + clock.advance(keyscanFailureInitialBackoff + time.Second) + } + + if scans != 3 { + t.Fatalf("timeouts compounded the keyscan backoff, scans = %d, want 3", scans) + } } type issue1638CountingRunner struct { runs int + err error } func (r *issue1638CountingRunner) Run(ctx context.Context, name string, args ...string) ([]byte, error) { r.runs++ + if r.err != nil { + return nil, r.err + } return nil, errors.New("connection refused") } -func TestIssue1638TemperatureSSHFailureBacksOff(t *testing.T) { +func issue1638TemperatureCollector(t *testing.T) (*TemperatureCollector, *issue1638CountingRunner) { + t.Helper() + tc, runner, _ := issue1638TemperatureCollectorWithKey(t) + return tc, runner +} + +func issue1638TemperatureCollectorWithKey(t *testing.T) (*TemperatureCollector, *issue1638CountingRunner, string) { + t.Helper() keyPath := filepath.Join(t.TempDir(), "id_ed25519_test") if err := os.WriteFile(keyPath, []byte("dummy"), 0o600); err != nil { t.Fatalf("write key: %v", err) @@ -173,6 +305,49 @@ func TestIssue1638TemperatureSSHFailureBacksOff(t *testing.T) { tc.hostKeys = nil runner := &issue1638CountingRunner{} tc.runner = runner + return tc, runner, keyPath +} + +// TestIssue1638RepairedSSHKeyClearsBackoff pins the escape hatch from the +// backoff: replacing the SSH key is the usual fix for whatever opened the +// window, so the next cycle must retry instead of waiting the window out. +func TestIssue1638RepairedSSHKeyClearsBackoff(t *testing.T) { + issue1638UseFakeClock(t) + tc, runner, keyPath := issue1638TemperatureCollectorWithKey(t) + + 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) + } + + if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { + t.Fatalf("CollectTemperature inside backoff: %v", err) + } + if runner.runs != 2 { + t.Fatalf("backoff window still ran ssh, total %d commands, want 2", runner.runs) + } + + // Operator installs a working key. + if err := os.WriteFile(keyPath, []byte("repaired"), 0o600); err != nil { + t.Fatalf("rewrite key: %v", err) + } + if err := os.Chtimes(keyPath, time.Now().Add(time.Minute), time.Now().Add(time.Minute)); err != nil { + t.Fatalf("touch key: %v", err) + } + + if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { + t.Fatalf("CollectTemperature after key repair: %v", err) + } + if runner.runs != 4 { + t.Fatalf("repaired key did not clear the backoff, total %d commands, want 4", runner.runs) + } +} + +func TestIssue1638TemperatureSSHFailureBacksOff(t *testing.T) { + clock := issue1638UseFakeClock(t) + tc, runner := issue1638TemperatureCollector(t) // First cycle attempts both the sensors and the RPi fallback command. if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { @@ -191,4 +366,142 @@ func TestIssue1638TemperatureSSHFailureBacksOff(t *testing.T) { if runner.runs != 2 { t.Fatalf("backoff window still ran ssh, total %d commands, want 2", runner.runs) } + + // Once the first window expires the host is retried, and that second hard + // failure doubles the window: a poll one initial window later is suppressed. + clock.advance(temperatureSSHFailureInitialBackoff + time.Second) + if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { + t.Fatalf("CollectTemperature after expiry: %v", err) + } + if runner.runs != 4 { + t.Fatalf("expired backoff did not retry, total %d commands, want 4", runner.runs) + } + + clock.advance(temperatureSSHFailureInitialBackoff + time.Second) + if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { + t.Fatalf("CollectTemperature inside doubled window: %v", err) + } + if runner.runs != 4 { + t.Fatalf("backoff did not compound, total %d commands, want 4", runner.runs) + } + + // Applying settings clears the backoff so a repaired key is picked up on + // the next cycle instead of waiting the window out. + tc.ResetSSHFailures() + if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { + t.Fatalf("CollectTemperature after reset: %v", err) + } + if runner.runs != 6 { + t.Fatalf("reset did not retry ssh, total %d commands, want 6", runner.runs) + } +} + +// TestIssue1638TemperatureBackoffDecaysAfterQuietPeriod pins that a host whose +// backoff expired long ago starts over at the floor rather than resuming the +// compounding it had reached. +func TestIssue1638TemperatureBackoffDecaysAfterQuietPeriod(t *testing.T) { + clock := issue1638UseFakeClock(t) + tc, runner := issue1638TemperatureCollector(t) + + 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) + } + + // Sit out the window and one further window on top of it, then fail again. + clock.advance(2*temperatureSSHFailureInitialBackoff + time.Second) + if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { + t.Fatalf("CollectTemperature after quiet period: %v", err) + } + if runner.runs != 4 { + t.Fatalf("quiet-period retry did not run, total %d commands, want 4", runner.runs) + } + + // The decayed backoff is back at the floor, so one initial window is enough + // to retry rather than the doubled window a compounding-only policy keeps. + clock.advance(temperatureSSHFailureInitialBackoff + time.Second) + if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { + t.Fatalf("CollectTemperature after decayed window: %v", err) + } + if runner.runs != 6 { + t.Fatalf("backoff did not decay to the floor, total %d commands, want 6", runner.runs) + } +} + +// TestIssue1638TemperatureDoesNotEscalateWhenNoSSHRan covers the case where the +// knownhosts manager suppresses the call inside its own backoff. Nothing was +// executed, so the temperature layer must not record a failure of its own — +// otherwise two independent backoffs compound against a single event. +func TestIssue1638TemperatureDoesNotEscalateWhenNoSSHRan(t *testing.T) { + issue1638UseFakeClock(t) + tc, runner := issue1638TemperatureCollector(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) + } + tc.hostKeys = manager + + // Put the knownhosts manager into its backoff window first. + if err := manager.Ensure(context.Background(), "node1.local"); err == nil { + t.Fatal("expected keyscan failure") + } + if scans != 1 { + t.Fatalf("keyscan ran %d times, want 1", scans) + } + + if _, err := tc.CollectTemperature(context.Background(), "node1.local", "node1"); err != nil { + t.Fatalf("CollectTemperature: %v", err) + } + + if runner.runs != 0 { + t.Fatalf("ssh was executed %d times despite a suppressed host key scan, want 0", runner.runs) + } + if scans != 1 { + t.Fatalf("ssh-keyscan re-ran inside its backoff, scans = %d, want 1", scans) + } + tc.sshFailureMu.Lock() + recorded := len(tc.sshFailures) + tc.sshFailureMu.Unlock() + if recorded != 0 { + t.Fatalf("temperature layer recorded %d failures for an attempt that never ran, want 0", recorded) + } +} + +// TestIssue1638TemperatureDeadlineDoesNotCompound pins the lenient treatment of +// our own collection deadline: it is not evidence that the host is broken, so +// the window holds at the floor and the wasted second probe is skipped. +func TestIssue1638TemperatureDeadlineDoesNotCompound(t *testing.T) { + clock := issue1638UseFakeClock(t) + tc, runner := issue1638TemperatureCollector(t) + runner.err = errors.New("signal: killed") + + expired := func() (context.Context, context.CancelFunc) { + return context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + } + + for cycle := 0; cycle < 3; cycle++ { + ctx, cancel := expired() + if _, err := tc.CollectTemperature(ctx, "node1.local", "node1"); err != nil { + cancel() + t.Fatalf("cycle %d: CollectTemperature: %v", cycle, err) + } + cancel() + clock.advance(temperatureSSHFailureInitialBackoff + time.Second) + } + + // One probe per cycle (the RPi fallback is pointless once the deadline has + // passed) and no compounding, so all three cycles ran. + if runner.runs != 3 { + t.Fatalf("deadline handling ran %d ssh commands, want 3 (one per cycle, no fallback, no compounding)", runner.runs) + } } diff --git a/internal/monitoring/knownhosts.go b/internal/monitoring/knownhosts.go index c8608b2da..429d925db 100644 --- a/internal/monitoring/knownhosts.go +++ b/internal/monitoring/knownhosts.go @@ -28,6 +28,9 @@ type KnownHostsManager interface { EnsureWithEntries(ctx context.Context, host string, port int, entries [][]byte) error // Path returns the absolute path to the managed known_hosts file. Path() string + // ResetFailures drops all recorded keyscan failure backoff so the next + // Ensure call retries immediately. + ResetFailures() } type knownHostsManager struct { @@ -56,6 +59,35 @@ const ( keyscanFailureMaxBackoff = 15 * time.Minute ) +// sshBackoffNow is the clock the SSH-related backoff maps in this package read. +// Tests replace it to exercise expiry and compounding without sleeping. +var sshBackoffNow = time.Now + +// sshBackoffDecayed reports whether a recorded failure is stale enough that the +// next failure should restart from the floor instead of compounding. Backoff +// that only ever doubles keeps a host pinned at the ceiling after a long quiet +// period, so an entry whose retry deadline passed more than one backoff window +// ago is treated as fresh (#1638). +func sshBackoffDecayed(now, retryAt time.Time, backoff time.Duration) bool { + if backoff <= 0 { + return true + } + return now.After(retryAt.Add(backoff)) +} + +// nextSSHBackoff returns the backoff to record for a failure, given whatever +// was recorded for the same target before. +func nextSSHBackoff(now time.Time, previousRetryAt time.Time, previousBackoff, initial, ceiling time.Duration, escalate bool) time.Duration { + if !escalate || previousBackoff <= 0 || sshBackoffDecayed(now, previousRetryAt, previousBackoff) { + return initial + } + backoff := previousBackoff * 2 + if backoff > ceiling { + backoff = ceiling + } + return backoff +} + var ( mkdirAllFn = os.MkdirAll statFn = os.Stat @@ -71,6 +103,11 @@ var ( // ErrNoHostKeys is returned when ssh-keyscan yields no usable entries. ErrNoHostKeys = errors.New("knownhosts: no host keys discovered") + // ErrKeyscanSuppressed signals that no ssh-keyscan was executed because the + // previous failure's backoff window has not expired yet. Callers use it to + // tell "the host refused us" apart from "we did not ask", so they do not + // escalate their own backoff for work that never ran. + ErrKeyscanSuppressed = errors.New("knownhosts: ssh-keyscan suppressed by backoff") // ErrHostKeyChanged signals that a host key already exists with a different fingerprint. ErrHostKeyChanged = errors.New("knownhosts: host key changed") ) @@ -167,10 +204,10 @@ func (m *knownHostsManager) EnsureWithPort(ctx context.Context, host string, por m.mu.Unlock() return nil } - if failure := m.failures[cacheKey]; failure != nil && time.Now().Before(failure.retryAt) { + if failure := m.failures[cacheKey]; failure != nil && sshBackoffNow().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) + return fmt.Errorf("%w for %s:%d until it expires: %w", ErrKeyscanSuppressed, host, port, err) } m.mu.Unlock() @@ -197,20 +234,34 @@ func (m *knownHostsManager) recordKeyscanFailure(cacheKey string, err error) { if m.failures == nil { m.failures = make(map[string]*keyscanFailure) } - backoff := keyscanFailureInitialBackoff + + // A keyscan that ran out of time says nothing about the host refusing us, + // so hold such a failure at the floor rather than compounding it. + escalate := !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) + + now := sshBackoffNow() + var previousRetryAt time.Time + var previousBackoff time.Duration if existing := m.failures[cacheKey]; existing != nil { - backoff = existing.backoff * 2 - if backoff > keyscanFailureMaxBackoff { - backoff = keyscanFailureMaxBackoff - } + previousRetryAt = existing.retryAt + previousBackoff = existing.backoff } + backoff := nextSSHBackoff(now, previousRetryAt, previousBackoff, keyscanFailureInitialBackoff, keyscanFailureMaxBackoff, escalate) + m.failures[cacheKey] = &keyscanFailure{ - retryAt: time.Now().Add(backoff), + retryAt: now.Add(backoff), backoff: backoff, err: err, } } +// ResetFailures implements KnownHostsManager.ResetFailures. +func (m *knownHostsManager) ResetFailures() { + m.mu.Lock() + defer m.mu.Unlock() + m.failures = make(map[string]*keyscanFailure) +} + // 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) == "" { @@ -455,6 +506,11 @@ func defaultKeyscan(ctx context.Context, host string, port int, timeout time.Dur output, err := keyscanCmdRunner(scanCtx, args...) if err != nil { + // Surface our own timeout as a context error so the backoff can tell it + // apart from a host actively refusing the scan. + if ctxErr := scanCtx.Err(); ctxErr != nil { + return nil, fmt.Errorf("%w (output: %s)", ctxErr, strings.TrimSpace(string(output))) + } return nil, fmt.Errorf("%w (output: %s)", err, strings.TrimSpace(string(output))) } return output, nil diff --git a/internal/monitoring/monitor_additional_test.go b/internal/monitoring/monitor_additional_test.go index 468e1cf25..dded09cf9 100644 --- a/internal/monitoring/monitor_additional_test.go +++ b/internal/monitoring/monitor_additional_test.go @@ -293,8 +293,6 @@ 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 19379d1ee..9ad5164fc 100644 --- a/internal/monitoring/monitor_cluster_helpers.go +++ b/internal/monitoring/monitor_cluster_helpers.go @@ -1,16 +1,33 @@ package monitoring import ( + "context" "net" "net/url" "strings" - "sync" "time" "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/pkg/tlsutil" ) -var lookupIPFunc = net.LookupIP +// discoveryPolicyLookupTimeout bounds a policy resolution so a poll cycle can +// never stall on an unresponsive resolver. +const discoveryPolicyLookupTimeout = 5 * time.Second + +// lookupIPFunc resolves an endpoint hostname for the discovery-policy check. +// +// It goes through the process-global cached resolver that pkg/tlsutil dials +// with, rather than a bare net.LookupIP, for two reasons: the policy verdict is +// then made against the same addresses the connection will actually reach, and +// repeat poll cycles cost a cache hit instead of a DNS query. The resolver +// caches failures as well as answers, so an endpoint whose name does not +// resolve costs one query per cache refresh too (#1638). Tests replace it. +var lookupIPFunc = func(host string) ([]net.IP, error) { + ctx, cancel := context.WithTimeout(context.Background(), discoveryPolicyLookupTimeout) + defer cancel() + return tlsutil.LookupHostCached(ctx, host) +} func lookupClusterEndpointLabel(instance *config.PVEInstance, nodeName string) string { if instance == nil { @@ -227,57 +244,6 @@ func discoveryPolicyIPsForEndpointHost(candidateURL string) []net.IP { return filtered } -// 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. @@ -293,20 +259,22 @@ func discoveryPolicyLiteralIPs(endpoint config.ClusterEndpoint, candidateURL str 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, ","), +// clusterEndpointAllowedByDiscoveryPolicy evaluates the configured discovery +// policy against every address the endpoint would be dialled at. +// +// There is deliberately no memoized verdict here. Resolution goes through the +// shared cached resolver (see lookupIPFunc), which already collapses repeat +// poll cycles to a cache hit, so a second decision cache would buy nothing +// while making the policy trail the configuration and, worse, freezing a +// fail-open "resolution failed, allow" verdict in place for minutes at a time. +// Every policy — including the default link-local blocklist that +// NormalizeDiscoveryConfig injects — is therefore enforced against resolved +// addresses, not just literal ones (#1638). +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 } - 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) @@ -328,51 +296,6 @@ func evaluateClusterEndpointDiscoveryPolicy(endpoint config.ClusterEndpoint, can 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 de4f282fc..3d0546018 100644 --- a/internal/monitoring/temperature.go +++ b/internal/monitoring/temperature.go @@ -90,6 +90,7 @@ type TemperatureCollector struct { runner CommandRunner sshFailureMu sync.Mutex sshFailures map[string]*temperatureSSHFailure + sshKeyIdentity string } // temperatureSSHFailure remembers a host whose SSH collection failed so @@ -106,28 +107,66 @@ const ( temperatureSSHFailureMaxBackoff = 15 * time.Minute ) +// sshFailureOutcome describes what a failed collection attempt tells us about +// the host, which decides how the per-host backoff moves. +type sshFailureOutcome int + +const ( + // sshFailureHard is a refusal, auth failure, or unusable output: the host + // answered (or actively did not) and the backoff should compound. + sshFailureHard sshFailureOutcome = iota + // sshFailureTransient is our own deadline expiring. That says nothing about + // the host, so hold the backoff at the floor instead of doubling it. + sshFailureTransient + // sshFailureNotAttempted means no ssh ran at all — the knownhosts manager + // suppressed the call inside its own backoff. Nothing was learned and + // nothing was spent, so the backoff must not move. + sshFailureNotAttempted +) + +// classifySSHFailure maps a failed attempt onto its backoff treatment (#1638). +func classifySSHFailure(err error) sshFailureOutcome { + switch { + case err == nil: + return sshFailureHard + case errors.Is(err, ErrKeyscanSuppressed): + return sshFailureNotAttempted + case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled): + return sshFailureTransient + default: + return sshFailureHard + } +} + 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) + return failure != nil && sshBackoffNow().Before(failure.retryAt) } -func (tc *TemperatureCollector) recordSSHFailure(host string) { +func (tc *TemperatureCollector) recordSSHFailure(host string, outcome sshFailureOutcome) { + if outcome == sshFailureNotAttempted { + return + } + tc.sshFailureMu.Lock() defer tc.sshFailureMu.Unlock() if tc.sshFailures == nil { tc.sshFailures = make(map[string]*temperatureSSHFailure) } - backoff := temperatureSSHFailureInitialBackoff + + now := sshBackoffNow() + var previousRetryAt time.Time + var previousBackoff time.Duration if existing := tc.sshFailures[host]; existing != nil { - backoff = existing.backoff * 2 - if backoff > temperatureSSHFailureMaxBackoff { - backoff = temperatureSSHFailureMaxBackoff - } + previousRetryAt = existing.retryAt + previousBackoff = existing.backoff } + backoff := nextSSHBackoff(now, previousRetryAt, previousBackoff, temperatureSSHFailureInitialBackoff, temperatureSSHFailureMaxBackoff, outcome == sshFailureHard) + tc.sshFailures[host] = &temperatureSSHFailure{ - retryAt: time.Now().Add(backoff), + retryAt: now.Add(backoff), backoff: backoff, } } @@ -138,6 +177,59 @@ func (tc *TemperatureCollector) clearSSHFailure(host string) { delete(tc.sshFailures, host) } +// ResetSSHFailures drops every recorded SSH backoff, including the knownhosts +// manager's keyscan backoff, so an operator who has just repaired an SSH key or +// host key sees the next poll cycle retry rather than waiting out a window that +// has compounded to a quarter of an hour (#1638). +func (tc *TemperatureCollector) ResetSSHFailures() { + if tc == nil { + return + } + + tc.sshFailureMu.Lock() + tc.sshFailures = nil + hostKeys := tc.hostKeys + tc.sshFailureMu.Unlock() + + if hostKeys != nil { + hostKeys.ResetFailures() + } +} + +// resetSSHFailuresOnKeyChange clears the backoff maps when the SSH key on disk +// has been replaced since the last cycle. +// +// Repairing that key is the usual fix for the failures that opened these +// backoff windows in the first place, and nothing else notices it happening, so +// without this the operator waits out a window that has already compounded +// toward fifteen minutes before Pulse tries the repaired key (#1638). +func (tc *TemperatureCollector) resetSSHFailuresOnKeyChange() { + identity := tc.currentSSHKeyIdentity() + + tc.sshFailureMu.Lock() + changed := tc.sshKeyIdentity != "" && tc.sshKeyIdentity != identity + tc.sshKeyIdentity = identity + tc.sshFailureMu.Unlock() + + if changed { + log.Info(). + Str("sshKeyPath", tc.sshKeyPath). + Msg("Temperature SSH key changed on disk; clearing per-host SSH backoff") + tc.ResetSSHFailures() + } +} + +// currentSSHKeyIdentity fingerprints the key file cheaply enough to run every +// poll cycle. Content is deliberately not read: this is change detection, not +// validation. +func (tc *TemperatureCollector) currentSSHKeyIdentity() string { + info, err := os.Stat(strings.TrimSpace(tc.sshKeyPath)) + if err != nil { + return "" + } + return fmt.Sprintf("%d:%d", info.ModTime().UnixNano(), info.Size()) +} + // NewTemperatureCollectorWithPort creates a new temperature collector with custom SSH port func NewTemperatureCollectorWithPort(sshUser, sshKeyPath string, sshPort int) *TemperatureCollector { if sshPort <= 0 { @@ -200,6 +292,8 @@ func (tc *TemperatureCollector) CollectTemperature(ctx context.Context, nodeHost return &models.Temperature{Available: false}, nil } + tc.resetSSHFailuresOnKeyChange() + if tc.inSSHFailureBackoff(host) { log.Debug(). Str("node", nodeName). @@ -214,8 +308,17 @@ func (tc *TemperatureCollector) CollectTemperature(ctx context.Context, nodeHost // and still force sensors -j, so keep the parser backward-compatible. output, err := tc.runSSHCommand(ctx, host, pulseSensorsSSHCommand) if err != nil || strings.TrimSpace(output) == "" { + // A suppressed keyscan or an expired deadline would fail the RPi + // fallback identically, so stop here rather than paying for a second + // probe, and do not compound a backoff the host has not earned (#1638). + if outcome := classifySSHFailure(err); outcome != sshFailureHard { + tc.logSkippedSSHAttempt(outcome, nodeName, host, err) + tc.recordSSHFailure(host, outcome) + return &models.Temperature{Available: false}, nil + } + if tc.disableLegacySSHOnAuthFailure(err, nodeName, host) { - tc.recordSSHFailure(host) + tc.recordSSHFailure(host, sshFailureHard) return &models.Temperature{Available: false}, nil } @@ -230,8 +333,14 @@ func (tc *TemperatureCollector) CollectTemperature(ctx context.Context, nodeHost } } + if outcome := classifySSHFailure(err); outcome != sshFailureHard { + tc.logSkippedSSHAttempt(outcome, nodeName, host, err) + tc.recordSSHFailure(host, outcome) + return &models.Temperature{Available: false}, nil + } + if tc.disableLegacySSHOnAuthFailure(err, nodeName, host) { - tc.recordSSHFailure(host) + tc.recordSSHFailure(host, sshFailureHard) return &models.Temperature{Available: false}, nil } @@ -240,7 +349,7 @@ 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) + tc.recordSSHFailure(host, sshFailureHard) return &models.Temperature{Available: false}, nil } @@ -265,6 +374,17 @@ func (tc *TemperatureCollector) CollectTemperature(ctx context.Context, nodeHost return temp, nil } +// logSkippedSSHAttempt records why an attempt is being abandoned without +// escalating the per-host backoff. +func (tc *TemperatureCollector) logSkippedSSHAttempt(outcome sshFailureOutcome, nodeName, host string, err error) { + event := log.Debug().Str("node", nodeName).Str("host", host).Err(err) + if outcome == sshFailureNotAttempted { + event.Msg("Skipping SSH temperature collection; host key scan is in its own backoff window") + return + } + event.Msg("Abandoning SSH temperature collection after the collection deadline expired") +} + func (tc *TemperatureCollector) runSSHCommand(ctx context.Context, host, command string) (string, error) { if err := tc.validateSSHKeyPath(); err != nil { return "", err @@ -316,6 +436,12 @@ func (tc *TemperatureCollector) runSSHCommand(ctx context.Context, host, command if errors.Is(err, errTemperatureCommandOutputTooLarge) { return "", fmt.Errorf("ssh command output exceeded %d bytes", maxTemperatureCommandOutputSize) } + // Our own budget running out is not evidence about the host, and + // sanitizeSSHCommandError would flatten the signal into an exit status, + // so report the context error directly (#1638). + if ctxErr := runCtx.Err(); ctxErr != nil { + return "", fmt.Errorf("ssh command aborted: %w", ctxErr) + } return "", sanitizeSSHCommandError(err) } diff --git a/pkg/tlsutil/dnscache.go b/pkg/tlsutil/dnscache.go index f52b49cc9..e17ed6556 100644 --- a/pkg/tlsutil/dnscache.go +++ b/pkg/tlsutil/dnscache.go @@ -67,6 +67,39 @@ func SetDNSCacheTTL(ttl time.Duration) { Msg("DNS cache TTL configured") } +// LookupHostCached resolves host through the same process-global cached +// resolver that DialContextWithCache uses, returning parsed addresses. +// +// Callers that reason about which addresses a connection will actually reach +// (endpoint policy checks, for example) must go through this rather than +// net.LookupIP, so the decision and the dial share one DNS view. The resolver +// caches both answers and lookup errors until the next refresh tick, so repeat +// callers cost a map lookup rather than a query (#1638). +func LookupHostCached(ctx context.Context, host string) ([]net.IP, error) { + if ctx == nil { + ctx = context.Background() + } + + addrs, err := GetDNSResolver().LookupHost(ctx, host) + if err != nil { + return nil, err + } + + ips := make([]net.IP, 0, len(addrs)) + for _, addr := range addrs { + if ip := net.ParseIP(addr); ip != nil { + ips = append(ips, ip) + } + } + if len(ips) == 0 { + return nil, &net.DNSError{ + Err: "no IP addresses found", + Name: host, + } + } + return ips, nil +} + // DialContextWithCache is a DialContext function that uses the DNS cache. // On macOS, connections to RFC 1918 addresses are routed through a subprocess // (nc) to bypass VPN/NECP routing captures that affect the host process.