diff --git a/cmd/main.go b/cmd/main.go index cf32fa5..acbae79 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -146,8 +146,9 @@ func main() { "Interval between provider health checks (can the cluster reach each hypervisor, API "+ "and account), published as ProviderHealth objects and metrics. 0 turns them off; "+ "otherwise at least 10s.") - flag.DurationVar(&providerCheckTimeout, "provider-check-timeout", 15*time.Second, - "Time limit for one round of provider health checks; at most the interval.") + flag.DurationVar(&providerCheckTimeout, "provider-check-timeout", 0, + "Time limit for one round of provider health checks: 15s, or the interval when that is "+ + "shorter. If set, at most the interval.") opts := zap.Options{ Development: true, diff --git a/config/manager/manager.yaml b/config/manager/manager.yaml index 1e21b85..8b3c5ee 100644 --- a/config/manager/manager.yaml +++ b/config/manager/manager.yaml @@ -68,7 +68,8 @@ spec: # (high volume: ~1/s per proxy). # Provider health checks (kubectl get providerhealth) run every # minute; tune with --provider-check-interval (0 turns them off, - # otherwise at least 10s) and --provider-check-timeout (15s). + # otherwise at least 10s) and --provider-check-timeout (15s, or + # the interval if that is shorter). env: # Bearer token for the discovery API. Optional: without the # Secret the API serves unauthenticated (with a loud warning). diff --git a/docs/architecture.md b/docs/architecture.md index fc3fded..c847d55 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -378,7 +378,8 @@ target and never fails as a whole. All three providers do: ```text Monitor.Start(ctx) ── leader-only; --provider-check-interval (1m; 0 = off, - │ otherwise ≥ 10s), --provider-check-timeout (15s) + │ otherwise ≥ 10s), --provider-check-timeout (15s, or + │ the interval if that is shorter) ├─ once: delete managed-by-labelled ProviderHealth objects whose provider │ is not configured (all of them when checks are off) — the providers │ config is read once per process @@ -598,8 +599,13 @@ they read as choices, not accidents. Chronological by build step. fresh dial would not pick up a rotated Secret anyway, since the SSH material is cached. It would also open an sshd session per host per minute. And a working cached connection is exactly what reconciles use. - The cost is that a check against a hung host drops the shared - connection, the same rule every cancelled reconcile already follows. + Sharing must not cost the reconciles anything, so the check calls + through `hostPool.probe`. A check that runs out of time abandons its + RPCs instead of closing the connection, which would kill an upload or a + Create running on it. At most one abandoned check per host is left + waiting, and until it returns, the next check reports `Timeout` at once. + Waiting for a host's connection slot also ends with the caller's + deadline, so a reconcile redialing a dead host cannot stall the round. - **Provider transitions are logs, not Events**, in line with the EventRecorder decision above. They log at Info with a `WARNING:` prefix, and each target's `lastTransitionTime` in the object records when it diff --git a/docs/kvm-provider.md b/docs/kvm-provider.md index 0f75c27..feab3f5 100644 --- a/docs/kvm-provider.md +++ b/docs/kvm-provider.md @@ -292,7 +292,7 @@ shows its SSH `user`. A failing target says where the check broke: | `Unreachable` | the TCP connect failed, or the link died mid-check | DNS, Tailscale or a firewall between the cluster and `address` | | `SSHHandshakeFailed` | the host key is missing from known_hosts or has changed, or `user` rejected the key | the known_hosts entry; `user`'s `authorized_keys` | | `LibvirtConnectFailed` | SSH worked, the libvirt socket did not | `user` in the `libvirt` group, `libvirtd` running, the `socket` path | -| `Timeout` | no answer within `--provider-check-timeout` (default 15s) | a hung `libvirtd` or a saturated link | +| `Timeout` | no answer within `--provider-check-timeout` (default 15s), or the previous check of the host is still waiting | a hung `libvirtd`, a saturated link, or a reconcile stuck redialing the host | | `PoolNotFound` | there is no pool named `pool` | `virsh pool-list --all` | | `PoolNotActive` | the pool exists but is not running | `virsh pool-start ` and `virsh pool-autostart ` | | `LibvirtError` | any other libvirt error | the message carries libvirt's own text | @@ -302,9 +302,14 @@ Sharing the connection has two consequences: - The leader now keeps one SSH connection per host open from startup, with a keepalive every 15s. Before, the first connection opened with the first KVM reconcile. -- A check that hits its timeout drops the host's connection, just as a - cancelled reconcile already does. Calls in flight on a hung link then fail - fast, and the next call redials. +- A check that runs out of time gives up on its RPCs but leaves the + connection alone, so an image upload or a Create running on it is not cut + off. At most one abandoned check per host keeps waiting; until it + returns, the next check reports `Timeout` straight away. A link that is + really dead is closed by the keepalive within about two keepalive + intervals. +- A check never waits longer than its timeout for the host's connection, + even while a reconcile holds it to redial a host that is down. ## Known gaps diff --git a/docs/plans-executions/2026-09-11-1526-provider-health-checks.md b/docs/plans-executions/2026-09-11-1526-provider-health-checks.md index 550e397..60b0a0c 100644 --- a/docs/plans-executions/2026-09-11-1526-provider-health-checks.md +++ b/docs/plans-executions/2026-09-11-1526-provider-health-checks.md @@ -312,3 +312,54 @@ contained `proxy_operator_provider_target_up{provider="kubernetes",target="apise That exercises the generated RBAC, the API-reader path and the `main.go` gauge wiring end to end. The two skipped specs are the Tempo-gated tracing specs; `TEMPO_URL` was not set. + +### Follow-up — the check must not cost the reconciles their connection + +Reviewing the shared-connection decision after the automated validation turned up two +ways the health check could hurt the work it shares the link with, both of which the +plan's "a check against a hung host drops the shared connection" line accepted too +easily: + +- A check that timed out went through `hostPool.call`, whose cancel path drops the + connection. A `libvirtd` slow to answer one `ConnectGetLibVersion` would therefore + cut off an image upload or a `Create` running on the same link. +- `hostEntry` was guarded by a `sync.Mutex`, so waiting for a host's slot could not be + interrupted. A reconcile redialing a host that is down holds the slot for as long as + the SSH dial takes — minutes on a blackholed address — and the whole check round + would wait behind it, past its own timeout. + +So the check now calls `hostPool.probe` instead of `call`: on `ctx` ending it abandons +the RPC and leaves the connection alone (a link that is really dead is closed by the +keepalive anyway, which ends the abandoned `fn` too). One abandoned probe per host is +allowed to stay waiting — `hostEntry.probing` is an `atomic.Bool`, not a counter, so a +daemon that never answers cannot pile up goroutines; further probes get the new +`errProbeBusy`, which `checkReason` maps to `Timeout`. `hostEntry.mu` became a one-slot +channel with an `acquire(ctx)` that only bounds *waiting*: an uncontended caller takes a +free slot even with an expired context, so `call`'s behaviour is unchanged where nothing +is contended. + +The second change came out of the same pass: `--provider-check-timeout` defaulted to a +flat 15s while the flag validation rejected a timeout longer than the interval, so +`--provider-check-interval=12s` alone failed to start. The flag now defaults to 0 and +`Monitor.setDefaults` derives `min(15s, Interval)`; `ValidateSchedule` accepts 0 and +rejects only negatives and timeouts past the interval. + +Re-verified with the same commands as above: + +```bash +go vet ./... +go test -race -count=1 ./internal/provider/kvm/ ./internal/providerhealth/ +make test +# every package ok; internal/provider/kvm 92.2 %, internal/providerhealth 93.7 % coverage +``` + +Worth noting: +- `make lint` still cannot run, for the toolchain reason recorded above; unchanged by + this follow-up. +- The two new `kvm` tests are timing-shaped by necessity — they assert that a caller + with a 50ms deadline does not wait out a dial that is blocked on a channel. The + assertions are one-sided (`> 1s` fails) so a slow machine does not make them flaky. +- `docs/architecture.md`, `docs/kvm-provider.md` and `docs/testing.md` were corrected in + the same commit: the old text promised the drop-on-timeout behaviour that is now gone. +- The real-hypervisor commands from the plan's Verification section are still the user's + to run; nothing here replaces them. diff --git a/docs/testing.md b/docs/testing.md index 2eb97d6..e7ac883 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -60,7 +60,7 @@ manual runs too. | | `errors_test.go`, `conn_test.go` | libvirt-code classification table; connection pool (reuse, redial, drop on transport error, cancel releases a blocked RPC, one dial per host under concurrency); SSH dial, host-key policy, keepalive | fake dialer; a **real in-process `x/crypto/ssh` server** forwarding a unix socket | | | `image_test.go`, `kvm_test.go` | image cache (fresh fetch, ready, marker missing, checksum change/mismatch, 4xx/5xx, concurrent callers); Create resume after every partial step, leftover truncated seed, state/IP mapping, every Delete already-gone case, ListByTag across hosts / one down / unreadable domain; logging tiers (V0 silent, V1 one line per RPC, V2 XML) with cloud-init never logged | `httptest` image server + in-memory `fakeAPI` behind the `libvirtAPI` seam | | | `trace_test.go`, `conn_test.go`, `errors_test.go` | every `libvirtAPI` method traced exactly once (guarded by reflection on the interface), results and errors pass through, libvirt error fields, detail keys only at V2, upload content never logged; pool connect/drop lines, no wrapping below V1 | `fakeAPI` + `funcr` capture | -| | `check_test.go`, `conn_test.go` | health check: details, per-host verdicts in config order, hosts checked concurrently, timeout, the reason table; an inactive or missing pool and daemon errors keep the connection; every dial stage tagged through the real SSH server, with classification unchanged | `fakeAPI`, fake dialers, in-process SSH server | +| | `check_test.go`, `conn_test.go` | health check: details, per-host verdicts in config order, hosts checked concurrently, a timeout that keeps the connection and refuses a second check while the first still waits, no waiting behind a reconcile's dial, the reason table; `probe` abandons without closing, but still drops on a transport error; waiting for a host's slot ends with the caller's deadline; an inactive or missing pool and daemon errors keep the connection; every dial stage tagged through the real SSH server, with classification unchanged | `fakeAPI`, fake dialers, in-process SSH server | | | `kvm_integration_test.go` (`-tags=integration`) | one real VM: image fetch, Create, idempotent re-Create, Running with IP, ListByTag, Delete, pool state afterwards; `TestKVMCheck_realHost` runs the health check against the host and creates nothing | a **real hypervisor** — see below | | `internal/health` | `probe_test.go` | probes through a **real CONNECT-capable proxy stub** to a real TLS server: tunnel success, refused CONNECT, wrong status, dead proxy, plain-http forward | `httptest` + hijacked bidirectional tunnel | | | `engine_test.go` | thresholds, first-verdict, latency suppression matrix, dropped-event retry, stale-probe UID guard, tick scheduling/seeding/pruning, `Start` end-to-end | fake `client.Reader` + injected `probeFn` | diff --git a/internal/provider/kvm/check.go b/internal/provider/kvm/check.go index a2d7f51..a700731 100644 --- a/internal/provider/kvm/check.go +++ b/internal/provider/kvm/check.go @@ -48,10 +48,12 @@ func (p *Provider) Check(ctx context.Context) []provider.TargetStatus { return out } -// checkHost returns the RPC errors from fn untouched: call drops the -// shared connection on anything that is not a libvirt error, and a pool -// that exists but is stopped says nothing about the link. facts is read -// only after call returned nil, since an abandoned fn may still write it. +// checkHost goes through probe, so a check that runs out of time abandons +// its RPCs rather than closing the connection the reconciles share. fn +// returns the RPC errors untouched, since a transport error still drops the +// link; a pool that exists but is stopped says nothing about the link, so +// that verdict is reached afterwards. facts is read only after probe +// returned nil, since an abandoned fn may still write it. func (p *Provider) checkHost(ctx context.Context, host provider.KVMHost) provider.TargetStatus { ts := provider.TargetStatus{ Name: host.Name, @@ -60,7 +62,7 @@ func (p *Provider) checkHost(ctx context.Context, host provider.KVMHost) provide } start := time.Now() var facts hostFacts - err := p.hosts.call(ctx, p.logger(ctx), host, func(api libvirtAPI) error { + err := p.hosts.probe(ctx, p.logger(ctx), host, func(api libvirtAPI) error { version, err := api.ConnectGetLibVersion() if err != nil { return err @@ -125,7 +127,7 @@ func checkReason(err error) string { } } switch { - case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled): + case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled), errors.Is(err, errProbeBusy): return provider.ReasonTimeout case hasCode(err, libvirt.ErrNoStoragePool): return reasonPoolNotFound diff --git a/internal/provider/kvm/check_test.go b/internal/provider/kvm/check_test.go index ca99639..022790d 100644 --- a/internal/provider/kvm/check_test.go +++ b/internal/provider/kvm/check_test.go @@ -7,6 +7,7 @@ import ( "io" "maps" "strings" + "sync" "sync/atomic" "testing" "time" @@ -174,14 +175,18 @@ func TestCheck_hostsCheckedConcurrently(t *testing.T) { } } -func TestCheck_timeout(t *testing.T) { +// A daemon that stopped answering times the check out, but the connection +// the reconciles share stays up, and the next check does not stack a +// second hung RPC on top of the first. +func TestCheck_timeoutKeepsTheConnection(t *testing.T) { t.Parallel() api := newFakeAPI("default") block := make(chan struct{}) t.Cleanup(func() { close(block) }) blocked := &blockingAPI{fakeAPI: api, block: block} + var closed atomic.Bool dial := func(context.Context, provider.KVMHost) (*hostClient, error) { - return &hostClient{api: blocked, close: func() {}}, nil + return &hostClient{api: blocked, close: func() { closed.Store(true) }}, nil } pc := testConfig("nas1") p := newWithDialer(pc.Name, *pc.KVM, dial, nil) @@ -193,6 +198,49 @@ func TestCheck_timeout(t *testing.T) { if got[0].Ready || got[0].Reason != provider.ReasonTimeout { t.Errorf("Check() = %+v, want Timeout", got[0]) } + if closed.Load() { + t.Error("the timed-out check closed the connection the reconciles share") + } + again := p.Check(context.Background()) + if again[0].Reason != provider.ReasonTimeout || !strings.Contains(again[0].Message, "still waiting") { + t.Errorf("second Check() = %+v, want an immediate Timeout while the first still waits", again[0]) + } +} + +// A reconcile redialing a dead host holds the host's slot for minutes; the +// check must give up at its own deadline rather than wait that out. +func TestCheck_doesNotWaitBehindAReconcileDial(t *testing.T) { + t.Parallel() + release := make(chan struct{}) + dialing := make(chan struct{}) + var once sync.Once + dial := func(context.Context, provider.KVMHost) (*hostClient, error) { + once.Do(func() { close(dialing) }) + <-release + return nil, errHostOff + } + pc := testConfig("nas1") + p := newWithDialer(pc.Name, *pc.KVM, dial, nil) + reconcile := make(chan error, 1) + go func() { + _, err := p.Get(context.Background(), formatProviderID("nas1", testName)) + reconcile <- err + }() + <-dialing + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + start := time.Now() + + got := p.Check(ctx) + + if waited := time.Since(start); waited > time.Second { + t.Errorf("Check() took %s behind the reconcile's dial, want about 50ms", waited) + } + if got[0].Reason != provider.ReasonTimeout { + t.Errorf("Check() = %+v, want Timeout", got[0]) + } + close(release) + <-reconcile } // blockingAPI hangs in ConnectGetLibVersion until block closes, like a @@ -222,6 +270,7 @@ func TestCheckReason(t *testing.T) { &dialError{stage: stageSSHHandshake, err: context.DeadlineExceeded}), reasonSSHHandshakeFailed}, {"deadline", context.DeadlineExceeded, provider.ReasonTimeout}, {"cancelled", context.Canceled, provider.ReasonTimeout}, + {"previous check still waiting", errProbeBusy, provider.ReasonTimeout}, {"no pool", lvError(libvirt.ErrNoStoragePool, "no pool"), reasonPoolNotFound}, {"other libvirt error", lvError(libvirt.ErrInternalError, "boom"), reasonLibvirtError}, {"link died mid-call", io.EOF, provider.ReasonUnreachable}, diff --git a/internal/provider/kvm/conn.go b/internal/provider/kvm/conn.go index da7ff1a..5b962c0 100644 --- a/internal/provider/kvm/conn.go +++ b/internal/provider/kvm/conn.go @@ -9,6 +9,7 @@ import ( "net" "os" "sync" + "sync/atomic" "time" libvirt "github.com/digitalocean/go-libvirt" @@ -102,9 +103,16 @@ type hostPool struct { entries map[string]*hostEntry } +// hostEntry is one host's slot in the pool. lock is a one-slot semaphore +// rather than a mutex so that waiting for it can end with the caller's +// context: whoever holds it may be dialing a dead host for as long as the +// dial takes. type hostEntry struct { - mu sync.Mutex + lock chan struct{} client *hostClient + // probing is set while a probe's fn runs, including one whose caller + // has already given up on it. + probing atomic.Bool } func newHostPool(dial hostDialer) *hostPool { @@ -116,16 +124,37 @@ func (p *hostPool) entry(name string) *hostEntry { defer p.mu.Unlock() e, ok := p.entries[name] if !ok { - e = &hostEntry{} + e = &hostEntry{lock: make(chan struct{}, 1)} p.entries[name] = e } return e } +// acquire takes e.lock. Only waiting is bounded by ctx: a free slot is +// taken even when ctx has already ended, so an uncontended caller behaves +// as it did with a mutex, and a contended one gives up with ctx's error. +func (e *hostEntry) acquire(ctx context.Context) error { + select { + case e.lock <- struct{}{}: + return nil + default: + } + select { + case e.lock <- struct{}{}: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (e *hostEntry) release() { <-e.lock } + func (p *hostPool) get(ctx context.Context, log logr.Logger, host provider.KVMHost) (*hostClient, error) { e := p.entry(host.Name) - e.mu.Lock() - defer e.mu.Unlock() + if err := e.acquire(ctx); err != nil { + return nil, fmt.Errorf("waiting for the connection to host %s (%s): %w", host.Name, host.Address, err) + } + defer e.release() if e.client != nil { if e.client.alive() { return e.client, nil @@ -149,8 +178,8 @@ func (p *hostPool) get(ctx context.Context, log logr.Logger, host provider.KVMHo // connection that was already replaced by a redial is left alone. func (p *hostPool) drop(host string, c *hostClient) { e := p.entry(host) - e.mu.Lock() - defer e.mu.Unlock() + e.lock <- struct{}{} + defer e.release() if e.client == c { e.client = nil } @@ -173,19 +202,61 @@ func (p *hostPool) call(ctx context.Context, log logr.Logger, host provider.KVMH if err != nil { return err } - api := c.api - if log.V(1).Enabled() { - api = &tracingAPI{inner: c.api, log: log} - } - err = awaitOrCancel(ctx, func() error { return fn(api) }, func() { + err = awaitOrCancel(ctx, func() error { return fn(traced(c, log)) }, func() { log.V(1).Info("libvirt connection dropped", "reason", "context cancelled", "error", ctx.Err().Error()) p.drop(host.Name, c) }) + p.dropOnTransportError(ctx, log, host, c, err) + return err +} + +// errProbeBusy refuses a probe while the previous one on the same host has +// not come back yet. +var errProbeBusy = errors.New("the previous check of this host is still waiting for libvirt to answer") + +// probe is call for the health check, which must never cost the reconciles +// their connection: when ctx ends first, fn is abandoned and the connection +// left alone. A link that is really dead is closed by the keepalive, which +// ends fn too. At most one probe per host runs at a time, an abandoned one +// included, so a daemon that stopped answering cannot pile up goroutines; +// until it answers, further probes get errProbeBusy. +func (p *hostPool) probe(ctx context.Context, log logr.Logger, host provider.KVMHost, fn func(api libvirtAPI) error) error { + log = log.WithName("libvirt") + e := p.entry(host.Name) + if !e.probing.CompareAndSwap(false, true) { + return errProbeBusy + } + c, err := p.get(ctx, log, host) + if err != nil { + e.probing.Store(false) + return err + } + err = awaitOrCancel(ctx, func() error { + defer e.probing.Store(false) + return fn(traced(c, log)) + }, func() { + log.V(1).Info("libvirt probe abandoned, connection kept", "error", ctx.Err().Error()) + }) + p.dropOnTransportError(ctx, log, host, c, err) + return err +} + +// traced returns c's API, logging every RPC when log has V(1) enabled. +func traced(c *hostClient, log logr.Logger) libvirtAPI { + if log.V(1).Enabled() { + return &tracingAPI{inner: c.api, log: log} + } + return c.api +} + +// dropOnTransportError drops c when err is evidence against the link: +// neither a libvirt error, nor a verdict already classified, nor the +// caller's own context ending. +func (p *hostPool) dropOnTransportError(ctx context.Context, log logr.Logger, host provider.KVMHost, c *hostClient, err error) { if err != nil && !isLibvirtError(err) && !isClassified(err) && !errors.Is(err, ctx.Err()) { log.V(1).Info("libvirt connection dropped", "reason", "transport error", "error", err.Error()) p.drop(host.Name, c) } - return err } // awaitOrCancel runs fn in the background and returns its result, unless ctx diff --git a/internal/provider/kvm/conn_test.go b/internal/provider/kvm/conn_test.go index 722c948..b3a777b 100644 --- a/internal/provider/kvm/conn_test.go +++ b/internal/provider/kvm/conn_test.go @@ -232,6 +232,97 @@ func TestHostPool_call_hostsAreIndependent(t *testing.T) { } } +// A dial to a dead host holds the host's slot for as long as it takes; a +// caller with a deadline must not wait past it. +func TestHostPool_get_waitingRespectsContext(t *testing.T) { + t.Parallel() + release := make(chan struct{}) + dialing := make(chan struct{}) + var once sync.Once + dial := func(context.Context, provider.KVMHost) (*hostClient, error) { + once.Do(func() { close(dialing) }) + <-release + return nil, errors.New("ssh dial: connection timed out") + } + p := newHostPool(dial) + slow := make(chan error, 1) + go func() { slow <- p.call(context.Background(), discard, testHost, noop) }() + <-dialing + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + start := time.Now() + + err := p.call(ctx, discard, testHost, noop) + + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("call() error = %v, want the caller's deadline", err) + } + if waited := time.Since(start); waited > time.Second { + t.Errorf("call() waited %s behind another caller's dial, want about 50ms", waited) + } + close(release) + if err := <-slow; err == nil { + t.Error("the blocked dial succeeded, want its error") + } +} + +func TestHostPool_probe_abandonsWithoutClosing(t *testing.T) { + t.Parallel() + d := &fakeDialer{} + p := newHostPool(d.dial) + unblock := make(chan struct{}) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + + err := p.probe(ctx, discard, testHost, func(libvirtAPI) error { + <-unblock // a daemon that stopped answering + return nil + }) + + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("probe() error = %v, want DeadlineExceeded", err) + } + if d.clients[0].closed.Load() { + t.Error("probe closed the shared connection when it gave up") + } + if err := p.probe(context.Background(), discard, testHost, noop); !errors.Is(err, errProbeBusy) { + t.Errorf("probe() while the abandoned one still waits = %v, want errProbeBusy", err) + } + if err := p.call(context.Background(), discard, testHost, noop); err != nil { + t.Errorf("call() on the kept connection = %v", err) + } + close(unblock) + deadline := time.Now().Add(time.Second) + for { + err := p.probe(context.Background(), discard, testHost, noop) + if err == nil { + break + } + if !errors.Is(err, errProbeBusy) || time.Now().After(deadline) { + t.Fatalf("probe() after the abandoned one returned = %v, want nil", err) + } + time.Sleep(time.Millisecond) + } + if d.dialCount() != 1 { + t.Errorf("dials = %d, want 1: nothing should have redialed", d.dialCount()) + } +} + +func TestHostPool_probe_dropsConnectionOnTransportError(t *testing.T) { + t.Parallel() + d := &fakeDialer{} + p := newHostPool(d.dial) + + err := p.probe(context.Background(), discard, testHost, func(libvirtAPI) error { return io.ErrUnexpectedEOF }) + + if !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("probe() error = %v, want io.ErrUnexpectedEOF passed through", err) + } + if !d.clients[0].closed.Load() { + t.Error("a transport error during a probe left the connection cached") + } +} + // --- SSH dial against an in-process server --- func newKeyPair(t *testing.T) (ssh.Signer, []byte) { diff --git a/internal/providerhealth/monitor.go b/internal/providerhealth/monitor.go index e001ea7..133a6bc 100644 --- a/internal/providerhealth/monitor.go +++ b/internal/providerhealth/monitor.go @@ -73,7 +73,8 @@ type Monitor struct { // round runs at start: nothing races it, and early visibility is the // point. Interval time.Duration - // Timeout bounds each round (default 15s). + // Timeout bounds each round; 0 means 15s, or the interval when that is + // shorter. Timeout time.Duration now func() time.Time @@ -92,31 +93,36 @@ func (m *Monitor) NeedLeaderElection() bool { return true } // ValidateSchedule checks --provider-check-interval and // --provider-check-timeout. The interval is 0 (off) or at least 10s, which -// is what keeps one status write per provider per round cheap; the timeout -// is positive and no longer than the interval. +// is what keeps one status write per provider per round cheap. The timeout +// is 0 (derived from the interval, see Monitor.Timeout) or positive and no +// longer than the interval. func ValidateSchedule(interval, timeout time.Duration) error { switch { case interval == 0: return nil case interval < minInterval: return fmt.Errorf("--provider-check-interval must be 0 (off) or at least %s, got %s", minInterval, interval) - case timeout <= 0 || timeout > interval: - return fmt.Errorf("--provider-check-timeout must be positive and at most the interval (%s), got %s", interval, timeout) + case timeout < 0 || timeout > interval: + return fmt.Errorf("--provider-check-timeout must be at most the interval (%s), got %s", interval, timeout) } return nil } +func (m *Monitor) setDefaults() { + if m.Timeout == 0 { + m.Timeout = min(defaultTimeout, m.Interval) + } + if m.now == nil { + m.now = time.Now + } +} + // Start deletes the objects of providers that are no longer configured, // then checks every Interval until ctx ends. The providers config is read // once per process, so cleaning up once is enough. With checking off, // every object goes, so none is left claiming a stale Ready. func (m *Monitor) Start(ctx context.Context) error { - if m.Timeout == 0 { - m.Timeout = defaultTimeout - } - if m.now == nil { - m.now = time.Now - } + m.setDefaults() log := logf.FromContext(ctx).WithName("provider-health") ctx = logf.IntoContext(ctx, log) diff --git a/internal/providerhealth/monitor_test.go b/internal/providerhealth/monitor_test.go index b2616d3..5e2d402 100644 --- a/internal/providerhealth/monitor_test.go +++ b/internal/providerhealth/monitor_test.go @@ -340,7 +340,8 @@ func TestValidateSchedule(t *testing.T) { {"shortest interval", 10 * time.Second, 10 * time.Second, false}, {"too frequent", 5 * time.Second, time.Second, true}, {"negative interval", -time.Minute, time.Second, true}, - {"no timeout", time.Minute, 0, true}, + {"timeout left to derive, short interval", 12 * time.Second, 0, false}, + {"negative timeout", time.Minute, -time.Second, true}, {"timeout longer than interval", time.Minute, 2 * time.Minute, true}, } for _, tc := range tests { @@ -352,3 +353,25 @@ func TestValidateSchedule(t *testing.T) { }) } } + +func TestMonitor_setDefaults_timeout(t *testing.T) { + t.Parallel() + tests := []struct { + name string + interval, timeout, want time.Duration + }{ + {"default", time.Minute, 0, 15 * time.Second}, + {"interval shorter than the default", 12 * time.Second, 0, 12 * time.Second}, + {"explicit", time.Minute, 20 * time.Second, 20 * time.Second}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + m := &Monitor{Interval: tc.interval, Timeout: tc.timeout} + m.setDefaults() + if m.Timeout != tc.want { + t.Errorf("Timeout = %s, want %s", m.Timeout, tc.want) + } + }) + } +}