diff --git a/agent/agent.go b/agent/agent.go index e9cce805c..48000c953 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -10,6 +10,7 @@ import ( "net" "strings" "sync" + "sync/atomic" "time" "github.com/gliderlabs/ssh" @@ -40,6 +41,8 @@ type Agent struct { netInterfaceDeltaTrackers map[uint16]*deltatracker.DeltaTracker[string, uint64] // Per-cache-time NIC delta trackers dockerManager *dockerManager // Manages Docker API requests sensorConfig *SensorConfig // Sensors config + tempsReading atomic.Bool // A temperature read is still running (possibly blocked in the kernel) + fansReading atomic.Bool // A fan read is still running (possibly blocked in the kernel) systemInfo system.Info // Host system info (dynamic) systemDetails system.Details // Host system details (static, once-per-connection) detailsDirty bool // Whether system details have changed and need to be resent diff --git a/agent/fans.go b/agent/fans.go index 7b312a583..52398619d 100644 --- a/agent/fans.go +++ b/agent/fans.go @@ -1,6 +1,7 @@ package agent import ( + "errors" "log/slog" "os" "path/filepath" @@ -11,6 +12,8 @@ import ( "github.com/henrygd/beszel/internal/entities/system" ) +var errFanFetchTimeout = errors.New("fan collection timed out") + type fanSensor struct { key, path, chip string } @@ -29,16 +32,15 @@ func (a *Agent) updateFans(systemStats *system.Stats) { if hwmonRoot == "" { return } - sensors, err := getFanSensors() + fans, err := a.readFansWithTimeout(getFanSensors) if err != nil { - slog.Debug("Error reading fans", "err", err) + if errors.Is(err, errFanFetchTimeout) { + slog.Warn("Error reading fans", "err", err) + } else { + slog.Debug("Error reading fans", "err", err) + } return } - // Filter before reading fan*_input: each read can wake an idle GPU. - if a.sensorConfig != nil && a.sensorConfig.skipGPU { - sensors = filterGpuFans(sensors) - } - fans := readFanSensors(sensors) if len(fans) == 0 { return } @@ -56,6 +58,27 @@ func (a *Agent) updateFans(systemStats *system.Stats) { // } } +// readFansWithTimeout discovers and reads fan sensors without letting a blocked +// sysfs read (e.g. an i915 fan1_input stuck resuming the GPU) hang collection, +// which holds the agent lock and would stall every later hub request. +func (a *Agent) readFansWithTimeout(getSensors func() ([]fanSensor, error)) (map[string]uint16, error) { + timeout := defaultSensorsTimeout + if a.sensorConfig != nil && a.sensorConfig.timeout > 0 { + timeout = a.sensorConfig.timeout + } + return readWithTimeout(&a.fansReading, timeout, errFanFetchTimeout, func() (map[string]uint16, error) { + sensors, err := getSensors() + if err != nil { + return nil, err + } + // Filter before reading fan*_input: each read can wake an idle GPU. + if a.sensorConfig != nil && a.sensorConfig.skipGPU { + sensors = filterGpuFans(sensors) + } + return readFanSensors(sensors), nil + }) +} + // readHwmonFans walks the given hwmon root (typically /sys/class/hwmon) and // returns a map of "_" → RPM for every fan*_input // file it finds. Zero RPM is retained because it can represent a real fan that diff --git a/agent/fans_test.go b/agent/fans_test.go index 113dc0095..2f1d28884 100644 --- a/agent/fans_test.go +++ b/agent/fans_test.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -120,3 +121,37 @@ func TestFilterGpuFans(t *testing.T) { assert.Equal(t, "nct6798_fan1", filtered[0].key) assert.Len(t, discovered, 2) } + +func TestReadFansWithTimeout(t *testing.T) { + agent := &Agent{sensorConfig: &SensorConfig{timeout: 10 * time.Millisecond}} + release := make(chan struct{}) + blocked := func() ([]fanSensor, error) { + <-release + return nil, nil + } + + fans, err := agent.readFansWithTimeout(blocked) + assert.ErrorIs(t, err, errFanFetchTimeout) + assert.Nil(t, fans) + + // a read still blocked in the kernel isn't retried on later polls + calls := 0 + fans, err = agent.readFansWithTimeout(func() ([]fanSensor, error) { + calls++ + return nil, nil + }) + assert.ErrorIs(t, err, errSensorReadInFlight) + assert.Nil(t, fans) + assert.Zero(t, calls) + + // once the blocked read returns, collection resumes + close(release) + require.Eventually(t, func() bool { return !agent.fansReading.Load() }, time.Second, time.Millisecond) + + root := t.TempDir() + writeFile(t, filepath.Join(root, "hwmon0", "name"), "chip\n") + writeFile(t, filepath.Join(root, "hwmon0", "fan1_input"), "1000\n") + fans, err = agent.readFansWithTimeout(newFanSensorCache(root)) + require.NoError(t, err) + assert.Equal(t, map[string]uint16{"chip_fan1": 1000}, fans) +} diff --git a/agent/sensors.go b/agent/sensors.go index 23428ea63..4265cc616 100644 --- a/agent/sensors.go +++ b/agent/sensors.go @@ -11,6 +11,7 @@ import ( "runtime" "strconv" "strings" + "sync/atomic" "time" "unicode/utf8" @@ -21,7 +22,12 @@ import ( "github.com/shirou/gopsutil/v4/sensors" ) -var errTemperatureFetchTimeout = errors.New("temperature collection timed out") +var ( + errTemperatureFetchTimeout = errors.New("temperature collection timed out") + errSensorReadInFlight = errors.New("previous sensor read still in progress") +) + +const defaultSensorsTimeout = 2 * time.Second // Matches sensors.TemperaturesWithContext to allow for panic recovery (gopsutil/issues/1832) type getTempsFn func(ctx context.Context) ([]sensors.TemperatureStat, error) @@ -53,7 +59,7 @@ func (a *Agent) newSensorConfig() *SensorConfig { // newSensorConfigWithEnv creates a SensorConfig with the provided environment variables // sensorsSet indicates if the SENSORS environment variable was explicitly set (even to empty string) func (a *Agent) newSensorConfigWithEnv(primarySensor, sysSensors, sensorsEnvVal, sensorsTimeout string, skipCollection, skipGPU bool) *SensorConfig { - timeout := 2 * time.Second + timeout := defaultSensorsTimeout if sensorsTimeout != "" { if d, err := time.ParseDuration(sensorsTimeout); err == nil { timeout = d @@ -126,7 +132,7 @@ func (a *Agent) updateTemperatures(systemStats *system.Stats) { temps, err := a.getTempsWithTimeout(getSensorTemps) if err != nil { // retry once on panic (gopsutil/issues/1832) - if !errors.Is(err, errTemperatureFetchTimeout) { + if !errors.Is(err, errTemperatureFetchTimeout) && !errors.Is(err, errSensorReadInFlight) { temps, err = a.getTempsWithTimeout(getSensorTemps) } if err != nil { @@ -195,11 +201,6 @@ func (a *Agent) getTempsWithPanicRecovery(getTemps getTempsFn) (temps []sensors. } func (a *Agent) getTempsWithTimeout(getTemps getTempsFn) ([]sensors.TemperatureStat, error) { - type result struct { - temps []sensors.TemperatureStat - err error - } - // Use a longer timeout on the first run to allow for initialization // (e.g. Windows LHM subprocess startup) timeout := a.sensorConfig.timeout @@ -208,17 +209,37 @@ func (a *Agent) getTempsWithTimeout(getTemps getTempsFn) ([]sensors.TemperatureS timeout = 10 * time.Second } + return readWithTimeout(&a.tempsReading, timeout, errTemperatureFetchTimeout, func() ([]sensors.TemperatureStat, error) { + return a.getTempsWithPanicRecovery(getTemps) + }) +} + +// readWithTimeout runs read in a goroutine and returns timeoutErr if it doesn't +// finish within timeout. A sysfs read blocked in the kernel can't be cancelled, +// so inflight stays set until it returns and later calls get errSensorReadInFlight +// instead of stacking up another blocked goroutine on every poll. +func readWithTimeout[T any](inflight *atomic.Bool, timeout time.Duration, timeoutErr error, read func() (T, error)) (T, error) { + var zero T + if !inflight.CompareAndSwap(false, true) { + return zero, errSensorReadInFlight + } + + type result struct { + value T + err error + } resultCh := make(chan result, 1) go func() { - temps, err := a.getTempsWithPanicRecovery(getTemps) - resultCh <- result{temps: temps, err: err} + value, err := read() + inflight.Store(false) + resultCh <- result{value, err} }() select { case res := <-resultCh: - return res.temps, res.err + return res.value, res.err case <-time.After(timeout): - return nil, errTemperatureFetchTimeout + return zero, timeoutErr } } diff --git a/agent/sensors_test.go b/agent/sensors_test.go index 0a3c91572..2a3ace60c 100644 --- a/agent/sensors_test.go +++ b/agent/sensors_test.go @@ -594,6 +594,31 @@ func TestGetTempsWithTimeout(t *testing.T) { assert.Nil(t, temps) assert.ErrorIs(t, err, errTemperatureFetchTimeout) }) + + t.Run("skips while a timed out read is still blocked", func(t *testing.T) { + agent := &Agent{sensorConfig: &SensorConfig{context: context.Background(), timeout: 10 * time.Millisecond}} + release := make(chan struct{}) + _, err := agent.getTempsWithTimeout(func(ctx context.Context) ([]sensors.TemperatureStat, error) { + <-release + return nil, nil + }) + assert.ErrorIs(t, err, errTemperatureFetchTimeout) + + called := false + _, err = agent.getTempsWithTimeout(func(ctx context.Context) ([]sensors.TemperatureStat, error) { + called = true + return nil, nil + }) + assert.ErrorIs(t, err, errSensorReadInFlight) + assert.False(t, called) + + close(release) + require.Eventually(t, func() bool { return !agent.tempsReading.Load() }, time.Second, time.Millisecond) + _, err = agent.getTempsWithTimeout(func(ctx context.Context) ([]sensors.TemperatureStat, error) { + return nil, nil + }) + assert.NoError(t, err) + }) } func TestUpdateTemperaturesSkipsOnTimeout(t *testing.T) {