feat: add runner health admission checks (#1090)

Opt-in local task-admission checks under a `health_check` config section (disabled by default):

- pause new task fetching when free disk space on the workspace volume is below the configured minimum
- optional executable health-check script — a non-zero exit, timeout, or start failure marks the runner unavailable
- checks run only while the runner is idle; the last result is reused while a job is active, and polling resumes automatically on recovery
- `/readyz` reports task-admission readiness (reusing the poll loop's last check); `/healthz` stays a process-liveness endpoint

---------

Co-authored-by: silverwind <me@silverwind.io>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1090
Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com>
This commit is contained in:
bircni
2026-07-22 15:10:45 +00:00
parent 7bec310002
commit c43cbe87ca
16 changed files with 753 additions and 32 deletions

View File

@@ -147,17 +147,19 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
}
runner.SetCapabilitiesFromDeclare(resp)
poller := poll.New(cfg, cli, runner)
if cfg.Metrics.Enabled {
metrics.Init()
metrics.RunnerInfo.WithLabelValues(ver.Version(), resp.Msg.Runner.Name).Set(1)
metrics.RunnerCapacity.Set(float64(cfg.Runner.Capacity))
metrics.RegisterUptimeFunc(time.Now())
metrics.RegisterRunningJobsFunc(runner.RunningCount, cfg.Runner.Capacity)
metrics.StartServer(ctx, cfg.Metrics.Addr)
metrics.StartServer(ctx, cfg.Metrics.Addr, func() (bool, string) {
return poller.Ready(cfg.Metrics.ReadinessGrace)
})
}
poller := poll.New(cfg, cli, runner)
if daemArgs.Once || reg.Ephemeral {
done := make(chan struct{})
go func() {

View File

@@ -32,6 +32,12 @@ type IdleRunner interface {
OnIdle(ctx context.Context)
}
// AvailabilityRunner can temporarily pause task fetching for local resource
// conditions such as low disk space without changing server-side scheduling.
type AvailabilityRunner interface {
CanAcceptTask(ctx context.Context) (bool, string)
}
type Poller struct {
client client.Client
runner TaskRunner
@@ -48,7 +54,12 @@ type Poller struct {
// unregistered is set when the server rejects the runner with an
// Unauthenticated response, meaning the runner is no longer registered.
unregistered atomic.Bool
unregistered atomic.Bool
lastHealthyPoll atomic.Int64
lastPollFailed atomic.Bool
availabilityMu sync.Mutex
availabilityReady bool
availabilityReason string
}
// workerState holds the single poller's backoff state. Consecutive empty or
@@ -70,7 +81,7 @@ func New(cfg *config.Config, client client.Client, runner TaskRunner) *Poller {
done := make(chan struct{})
return &Poller{
p := &Poller{
client: client,
runner: runner,
cfg: cfg,
@@ -83,6 +94,10 @@ func New(cfg *config.Config, client client.Client, runner TaskRunner) *Poller {
done: done,
}
p.lastHealthyPoll.Store(time.Now().UnixNano())
p.availabilityReady = true
p.availabilityReason = "ok"
return p
}
func (p *Poller) Poll() {
@@ -102,6 +117,17 @@ func (p *Poller) Poll() {
return
}
ready, reason := p.localAvailability(p.pollingCtx)
p.reportAvailability(ready, reason)
if !ready {
p.runIdleMaintenance()
<-sem
if !p.waitBackoff(s) {
return
}
continue
}
task, ok := p.fetchTask(p.pollingCtx, s)
if !ok {
p.runIdleMaintenance()
@@ -127,6 +153,15 @@ func (p *Poller) PollOnce() {
defer close(p.done)
s := &workerState{}
for {
ready, reason := p.localAvailability(p.pollingCtx)
p.reportAvailability(ready, reason)
if !ready {
p.runIdleMaintenance()
if !p.waitBackoff(s) {
return
}
continue
}
task, ok := p.fetchTask(p.pollingCtx, s)
if !ok {
p.runIdleMaintenance()
@@ -154,6 +189,48 @@ func (p *Poller) Unregistered() bool {
return p.unregistered.Load()
}
// Ready reports whether the daemon can currently communicate with Gitea and
// accept work, reusing the availability the poll loop last observed rather than
// re-running the check. Transient transport failures are tolerated for grace.
func (p *Poller) Ready(grace time.Duration) (bool, string) {
if p.unregistered.Load() {
return false, "runner is no longer registered"
}
p.availabilityMu.Lock()
ready, reason := p.availabilityReady, p.availabilityReason
p.availabilityMu.Unlock()
if !ready {
return false, reason
}
if !p.lastPollFailed.Load() {
return true, "ok"
}
if time.Since(time.Unix(0, p.lastHealthyPoll.Load())) <= grace {
return true, "polling errors within grace period"
}
return false, "unable to poll Gitea"
}
func (p *Poller) localAvailability(ctx context.Context) (bool, string) {
if available, ok := p.runner.(AvailabilityRunner); ok {
return available.CanAcceptTask(ctx)
}
return true, "ok"
}
func (p *Poller) reportAvailability(ready bool, reason string) {
p.availabilityMu.Lock()
defer p.availabilityMu.Unlock()
switch {
case !ready && p.availabilityReady:
log.Warnf("runner temporarily unavailable: %s", reason)
case ready && !p.availabilityReady:
log.Info("runner local health recovered, resuming task polling")
}
p.availabilityReady = ready
p.availabilityReason = reason
}
func (p *Poller) runIdleMaintenance() {
if idleRunner, ok := p.runner.(IdleRunner); ok {
idleRunner.OnIdle(p.jobsCtx)
@@ -273,6 +350,7 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task,
// found no work within FetchTimeout. Treat it as an empty response and do
// not record the duration — the timeout value would swamp the histogram.
if errors.Is(err, context.DeadlineExceeded) {
p.markHealthyPoll()
s.consecutiveEmpty++
s.consecutiveErrors = 0 // timeout is a healthy idle response
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc()
@@ -291,11 +369,13 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task,
return nil, false
}
log.WithError(err).Error("failed to fetch task")
p.lastPollFailed.Store(true)
s.consecutiveErrors++
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultError).Inc()
metrics.ClientErrors.WithLabelValues(metrics.LabelMethodFetchTask).Inc()
return nil, false
}
p.markHealthyPoll()
// Successful response — reset error counter.
s.consecutiveErrors = 0
@@ -322,3 +402,8 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task,
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultTask).Inc()
return resp.Msg.Task, true
}
func (p *Poller) markHealthyPoll() {
p.lastHealthyPoll.Store(time.Now().UnixNano())
p.lastPollFailed.Store(false)
}

View File

@@ -159,6 +159,81 @@ type idleAwareRunner struct {
idleCalls atomic.Int64
}
type availabilityRunner struct {
mockRunner
ready atomic.Bool
reason string
}
func (r *availabilityRunner) CanAcceptTask(_ context.Context) (bool, string) {
if r.ready.Load() {
return true, "ok"
}
return false, r.reason
}
func TestPollerReady(t *testing.T) {
cfg, err := config.LoadDefault("")
require.NoError(t, err)
poller := New(cfg, nil, &availabilityRunner{})
// /readyz reuses the availability the poll loop last recorded.
ready, reason := poller.Ready(time.Second)
assert.True(t, ready)
assert.Equal(t, "ok", reason)
poller.reportAvailability(false, "low disk space")
ready, reason = poller.Ready(time.Second)
assert.False(t, ready)
assert.Equal(t, "low disk space", reason)
poller.reportAvailability(true, "ok")
poller.lastPollFailed.Store(true)
poller.lastHealthyPoll.Store(time.Now().UnixNano())
ready, _ = poller.Ready(time.Second)
assert.True(t, ready, "transient polling errors should remain ready during grace")
poller.lastHealthyPoll.Store(time.Now().Add(-2 * time.Second).UnixNano())
ready, reason = poller.Ready(time.Second)
assert.False(t, ready)
assert.Equal(t, "unable to poll Gitea", reason)
poller.unregistered.Store(true)
ready, reason = poller.Ready(time.Second)
assert.False(t, ready)
assert.Equal(t, "runner is no longer registered", reason)
}
func TestPollerPausesAndResumesForLocalAvailability(t *testing.T) {
var fetches atomic.Int64
cli := mocks.NewClient(t)
cli.On("FetchTask", mock.Anything, mock.Anything).Maybe().Run(func(mock.Arguments) {
fetches.Add(1)
}).Return(connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil)
cfg, err := config.LoadDefault("")
require.NoError(t, err)
cfg.Runner.FetchInterval = 10 * time.Millisecond
cfg.Runner.FetchIntervalMax = 10 * time.Millisecond
runner := &availabilityRunner{reason: "low disk space"}
poller := New(cfg, cli, runner)
var wg sync.WaitGroup
wg.Go(poller.Poll)
time.Sleep(40 * time.Millisecond)
assert.Zero(t, fetches.Load(), "an unavailable runner must not fetch a task")
runner.ready.Store(true)
require.Eventually(t, func() bool {
return fetches.Load() > 0
}, time.Second, 10*time.Millisecond, "polling should resume after local recovery")
shutdownCtx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
require.NoError(t, poller.Shutdown(shutdownCtx))
wg.Wait()
}
func (m *mockRunner) Run(ctx context.Context, _ *runnerv1.Task) error {
atomicMax(&m.maxConcurrent, m.running.Add(1))
select {

View File

@@ -0,0 +1,12 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
//go:build !aix && !darwin && !dragonfly && !freebsd && !linux && !netbsd && !openbsd && !solaris && !windows
package run
import "fmt"
func freeDiskBytes(path string) (uint64, error) {
return 0, fmt.Errorf("free disk space checks are not supported for %s", path)
}

View File

@@ -0,0 +1,53 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package run
import (
"testing"
"gitea.com/gitea/runner/internal/pkg/config"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestCanAcceptTaskDiskGuard(t *testing.T) {
cfg, err := config.LoadDefault("")
require.NoError(t, err)
cfg.Host.WorkdirParent = t.TempDir()
cfg.HealthCheck.Enabled = true
r := &Runner{cfg: cfg}
ready, _ := r.CanAcceptTask(t.Context())
assert.True(t, ready)
cfg.HealthCheck.MinFreeDiskSpaceMB = 1 << 40
ready, reason := r.CanAcceptTask(t.Context())
assert.False(t, ready)
assert.Contains(t, reason, "low disk space")
}
func TestDiskCheckDeferredWhileJobRuns(t *testing.T) {
cfg, err := config.LoadDefault("")
require.NoError(t, err)
cfg.HealthCheck.Enabled = true
cfg.HealthCheck.MinFreeDiskSpaceMB = 1
cfg.Host.WorkdirParent = t.TempDir()
r := &Runner{cfg: cfg}
ready, reason := r.CanAcceptTask(t.Context())
assert.True(t, ready)
assert.Equal(t, "ok", reason)
cfg.HealthCheck.MinFreeDiskSpaceMB = 1 << 40
r.runningCount.Store(1)
ready, reason = r.CanAcceptTask(t.Context())
assert.True(t, ready, "the disk check must not run while a job is active")
assert.Equal(t, "ok", reason)
r.runningCount.Store(0)
ready, reason = r.CanAcceptTask(t.Context())
assert.False(t, ready)
assert.Contains(t, reason, "low disk space")
}

View File

@@ -0,0 +1,16 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
package run
import "golang.org/x/sys/unix"
func freeDiskBytes(path string) (uint64, error) {
var stat unix.Statfs_t
if err := unix.Statfs(path, &stat); err != nil {
return 0, err
}
return stat.Bavail * uint64(stat.Bsize), nil
}

View File

@@ -0,0 +1,20 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
//go:build windows
package run
import "golang.org/x/sys/windows"
func freeDiskBytes(path string) (uint64, error) {
pathPtr, err := windows.UTF16PtrFromString(path)
if err != nil {
return 0, err
}
var available uint64
if err := windows.GetDiskFreeSpaceEx(pathPtr, &available, nil, nil); err != nil {
return 0, err
}
return available, nil
}

View File

@@ -0,0 +1,106 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package run
import (
"context"
"errors"
"fmt"
"maps"
"os"
"os/exec"
"strconv"
"strings"
"time"
"gitea.com/gitea/runner/act/common"
"gitea.com/gitea/runner/internal/pkg/process"
)
func (r *Runner) checkConfiguredHealth(ctx context.Context) (bool, string) {
script := r.cfg.HealthCheck.Script
if script == "" {
return true, "ok"
}
now := time.Now()
if r.now != nil {
now = r.now()
}
if !r.healthCheckLast.IsZero() && now.Sub(r.healthCheckLast) < r.cfg.HealthCheck.Interval {
return r.healthCheckReady, r.healthCheckReason
}
env := processEnvironment()
maps.Copy(env, r.cloneEnvs())
env["GITEA_RUNNER_HEALTH_CHECK"] = "true"
env["GITEA_RUNNER_NAME"] = r.name
if r.client != nil {
env["GITEA_INSTANCE_URL"] = r.client.Address()
}
env["GITEA_RUNNER_RUNNING_JOBS"] = strconv.FormatInt(r.RunningCount(), 10)
runner := r.runHealthCheck
if runner == nil {
runner = executeHealthCheck
}
err := runner(ctx, script, r.cfg.HealthCheck.Timeout, env)
r.healthCheckLast = now
r.healthCheckReady = err == nil
if err != nil {
r.healthCheckReason = "runner health check failed: " + err.Error()
} else {
r.healthCheckReason = "ok"
}
return r.healthCheckReady, r.healthCheckReason
}
func executeHealthCheck(ctx context.Context, script string, timeout time.Duration, env map[string]string) error {
checkCtx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
cmd := exec.CommandContext(checkCtx, script)
cmd.Env = envListFromMap(env)
cmd.SysProcAttr = process.SysProcAttr(script, false)
writer := common.NewLineWriter(func(line string) bool {
line = strings.TrimRight(line, "\r\n")
if line != "" {
common.Logger(ctx).Infof("health check: %s", line)
}
return true
})
cmd.Stdout = writer
cmd.Stderr = writer
treeKill := process.NewTreeKill(cmd)
if err := cmd.Start(); err != nil {
return fmt.Errorf("start: %w", err)
}
if killer, err := treeKill.Capture(cmd.Process); err == nil {
defer killer.Close()
}
err := cmd.Wait()
common.FlushWriter(writer)
if errors.Is(checkCtx.Err(), context.DeadlineExceeded) {
return fmt.Errorf("timed out after %s", timeout)
}
if err == nil {
return nil
}
var exitErr *exec.ExitError
if errors.As(err, &exitErr) {
return fmt.Errorf("exited with code %d", exitErr.ExitCode())
}
return err
}
func processEnvironment() map[string]string {
environ := os.Environ()
env := make(map[string]string, len(environ))
for _, value := range environ {
if key, item, ok := strings.Cut(value, "="); ok {
env[key] = item
}
}
return env
}

View File

@@ -0,0 +1,148 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package run
import (
"context"
"errors"
"testing"
"time"
"gitea.com/gitea/runner/internal/pkg/config"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestConfiguredHealthCheckCachesAndRecovers(t *testing.T) {
cfg, err := config.LoadDefault("")
require.NoError(t, err)
cfg.HealthCheck.Enabled = true
cfg.HealthCheck.Script = "/health-check"
cfg.HealthCheck.Interval = time.Minute
cfg.HealthCheck.Timeout = time.Second
now := time.Now()
calls := 0
fail := true
r := &Runner{
cfg: cfg,
name: "runner-1",
now: func() time.Time { return now },
runHealthCheck: func(_ context.Context, script string, timeout time.Duration, env map[string]string) error {
calls++
assert.Equal(t, "/health-check", script)
assert.Equal(t, time.Second, timeout)
assert.Equal(t, "true", env["GITEA_RUNNER_HEALTH_CHECK"])
assert.Equal(t, "runner-1", env["GITEA_RUNNER_NAME"])
if fail {
return errors.New("unhealthy")
}
return nil
},
}
ready, reason := r.CanAcceptTask(t.Context())
assert.False(t, ready)
assert.Contains(t, reason, "unhealthy")
assert.Equal(t, 1, calls)
fail = false
ready, _ = r.CanAcceptTask(t.Context())
assert.False(t, ready, "the failed result should remain cached")
assert.Equal(t, 1, calls)
now = now.Add(time.Minute)
ready, reason = r.CanAcceptTask(t.Context())
assert.True(t, ready)
assert.Equal(t, "ok", reason)
assert.Equal(t, 2, calls)
}
func TestConfiguredHealthCheckDisabled(t *testing.T) {
cfg, err := config.LoadDefault("")
require.NoError(t, err)
cfg.HealthCheck.MinFreeDiskSpaceMB = 1 << 40
cfg.HealthCheck.Script = "/must-not-run"
r := &Runner{
cfg: cfg,
runHealthCheck: func(_ context.Context, _ string, _ time.Duration, _ map[string]string) error {
t.Fatal("disabled health check executed")
return nil
},
}
ready, reason := r.CanAcceptTask(t.Context())
assert.True(t, ready)
assert.Equal(t, "ok", reason)
}
func TestConfiguredHealthCheckDeferredWhileJobRuns(t *testing.T) {
cfg, err := config.LoadDefault("")
require.NoError(t, err)
cfg.HealthCheck.Enabled = true
cfg.HealthCheck.Script = "/health-check"
cfg.HealthCheck.Interval = time.Minute
now := time.Now()
calls := 0
fail := false
r := &Runner{
cfg: cfg,
now: func() time.Time { return now },
runHealthCheck: func(_ context.Context, _ string, _ time.Duration, _ map[string]string) error {
calls++
if fail {
return errors.New("unhealthy")
}
return nil
},
}
ready, reason := r.CanAcceptTask(t.Context())
assert.True(t, ready)
assert.Equal(t, "ok", reason)
assert.Equal(t, 1, calls)
now = now.Add(time.Minute)
fail = true
r.runningCount.Store(1)
ready, reason = r.CanAcceptTask(t.Context())
assert.True(t, ready, "the last result should be reused while a job runs")
assert.Equal(t, "ok", reason)
assert.Equal(t, 1, calls)
r.runningCount.Store(0)
ready, reason = r.CanAcceptTask(t.Context())
assert.False(t, ready)
assert.Contains(t, reason, "unhealthy")
assert.Equal(t, 2, calls)
}
func TestConfiguredHealthCheckInitialRunDeferredWhileJobRuns(t *testing.T) {
cfg, err := config.LoadDefault("")
require.NoError(t, err)
cfg.HealthCheck.Enabled = true
cfg.HealthCheck.Script = "/health-check"
calls := 0
r := &Runner{
cfg: cfg,
runHealthCheck: func(_ context.Context, _ string, _ time.Duration, _ map[string]string) error {
calls++
return nil
},
}
r.runningCount.Store(1)
ready, reason := r.CanAcceptTask(t.Context())
assert.True(t, ready)
assert.Contains(t, reason, "deferred")
assert.Zero(t, calls)
r.runningCount.Store(0)
ready, reason = r.CanAcceptTask(t.Context())
assert.True(t, ready)
assert.Equal(t, "ok", reason)
assert.Equal(t, 1, calls)
}

View File

@@ -66,6 +66,13 @@ type Runner struct {
runningCount atomic.Int64
lastIdleCleanupUnixNano atomic.Int64
now func() time.Time
healthCheckLast time.Time
healthCheckReady bool
healthCheckReason string
healthStatusSet bool
healthStatusReady bool
healthStatusReason string
runHealthCheck func(context.Context, string, time.Duration, map[string]string) error
}
func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client) *Runner {
@@ -109,13 +116,14 @@ func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client)
envs["GITEA_ACTIONS_RUNNER_VERSION"] = ver.Version()
runner := &Runner{
name: reg.Name,
cfg: cfg,
client: cli,
labels: ls,
envs: envs,
cacheHandler: cacheHandler,
now: time.Now,
name: reg.Name,
cfg: cfg,
client: cli,
labels: ls,
envs: envs,
cacheHandler: cacheHandler,
now: time.Now,
runHealthCheck: executeHealthCheck,
}
return runner
}
@@ -579,6 +587,67 @@ func (r *Runner) RunningCount() int64 {
return r.runningCount.Load()
}
// CanAcceptTask checks local admission conditions without consuming a task. It is
// called only from the poll loop, so the cached health fields need no lock.
func (r *Runner) CanAcceptTask(ctx context.Context) (bool, string) {
if !r.cfg.HealthCheck.Enabled {
return true, "ok"
}
if r.RunningCount() > 0 {
if !r.healthStatusSet {
return true, "health checks deferred while jobs are running"
}
return r.healthStatusReady, r.healthStatusReason
}
if ready, reason := checkFreeDisk(r.cfg); !ready {
r.setHealthStatus(ready, reason)
return false, reason
}
ready, reason := r.checkConfiguredHealth(ctx)
r.setHealthStatus(ready, reason)
return ready, reason
}
func (r *Runner) setHealthStatus(ready bool, reason string) {
r.healthStatusSet = true
r.healthStatusReady = ready
r.healthStatusReason = reason
}
// checkFreeDisk evaluates the configured task-admission disk threshold.
func checkFreeDisk(cfg *config.Config) (bool, string) {
root := cfg.Host.WorkdirParent
if cfg.Container.BindWorkdir {
root = filepath.FromSlash("/" + strings.TrimLeft(cfg.Container.WorkdirParent, "/"))
}
root = nearestExistingPath(root)
available, err := freeDiskBytes(root)
if err != nil {
return false, fmt.Sprintf("cannot determine free disk space for %s: %v", root, err)
}
availableMB := available / (1024 * 1024)
if availableMB < uint64(cfg.HealthCheck.MinFreeDiskSpaceMB) {
return false, fmt.Sprintf("low disk space on %s: %d MiB available, %d MiB required", root, availableMB, cfg.HealthCheck.MinFreeDiskSpaceMB)
}
return true, "ok"
}
func nearestExistingPath(path string) string {
for path != "" {
if _, err := os.Stat(path); err == nil {
return path
}
parent := filepath.Dir(path)
if parent == path {
return parent
}
path = parent
}
return "."
}
func (r *Runner) Declare(ctx context.Context, labels []string) (*connect.Response[runnerv1.DeclareResponse], error) {
return r.client.Declare(ctx, connect.NewRequest(&runnerv1.DeclareRequest{
Version: ver.Version(),