From c43cbe87ca95efd12eee221b5640646f24e6a511 Mon Sep 17 00:00:00 2001 From: bircni Date: Wed, 22 Jul 2026 15:10:45 +0000 Subject: [PATCH] feat: add runner health admission checks (#1090) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Reviewed-on: https://gitea.com/gitea/runner/pulls/1090 Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com> --- internal/app/cmd/daemon.go | 8 +- internal/app/poll/poller.go | 89 +++++++++++++- internal/app/poll/poller_test.go | 75 ++++++++++++ internal/app/run/disk_other.go | 12 ++ internal/app/run/disk_test.go | 53 +++++++++ internal/app/run/disk_unix.go | 16 +++ internal/app/run/disk_windows.go | 20 ++++ internal/app/run/health_check.go | 106 +++++++++++++++++ internal/app/run/health_check_test.go | 148 ++++++++++++++++++++++++ internal/app/run/runner.go | 83 +++++++++++-- internal/pkg/config/config.example.yaml | 20 +++- internal/pkg/config/config.go | 40 +++++-- internal/pkg/config/config_test.go | 35 ++++++ internal/pkg/metrics/metrics_test.go | 2 +- internal/pkg/metrics/server.go | 35 ++++-- internal/pkg/metrics/server_test.go | 43 +++++++ 16 files changed, 753 insertions(+), 32 deletions(-) create mode 100644 internal/app/run/disk_other.go create mode 100644 internal/app/run/disk_test.go create mode 100644 internal/app/run/disk_unix.go create mode 100644 internal/app/run/disk_windows.go create mode 100644 internal/app/run/health_check.go create mode 100644 internal/app/run/health_check_test.go create mode 100644 internal/pkg/metrics/server_test.go diff --git a/internal/app/cmd/daemon.go b/internal/app/cmd/daemon.go index a4b5a0a6..2ed6f70e 100644 --- a/internal/app/cmd/daemon.go +++ b/internal/app/cmd/daemon.go @@ -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() { diff --git a/internal/app/poll/poller.go b/internal/app/poll/poller.go index 37057afd..b41d3ceb 100644 --- a/internal/app/poll/poller.go +++ b/internal/app/poll/poller.go @@ -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) +} diff --git a/internal/app/poll/poller_test.go b/internal/app/poll/poller_test.go index 92ca8fad..cb28ba67 100644 --- a/internal/app/poll/poller_test.go +++ b/internal/app/poll/poller_test.go @@ -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 { diff --git a/internal/app/run/disk_other.go b/internal/app/run/disk_other.go new file mode 100644 index 00000000..2fa0d06b --- /dev/null +++ b/internal/app/run/disk_other.go @@ -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) +} diff --git a/internal/app/run/disk_test.go b/internal/app/run/disk_test.go new file mode 100644 index 00000000..07851a9c --- /dev/null +++ b/internal/app/run/disk_test.go @@ -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") +} diff --git a/internal/app/run/disk_unix.go b/internal/app/run/disk_unix.go new file mode 100644 index 00000000..7c7294c8 --- /dev/null +++ b/internal/app/run/disk_unix.go @@ -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 +} diff --git a/internal/app/run/disk_windows.go b/internal/app/run/disk_windows.go new file mode 100644 index 00000000..e3a08a68 --- /dev/null +++ b/internal/app/run/disk_windows.go @@ -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 +} diff --git a/internal/app/run/health_check.go b/internal/app/run/health_check.go new file mode 100644 index 00000000..5b00ee50 --- /dev/null +++ b/internal/app/run/health_check.go @@ -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 +} diff --git a/internal/app/run/health_check_test.go b/internal/app/run/health_check_test.go new file mode 100644 index 00000000..f6e09673 --- /dev/null +++ b/internal/app/run/health_check_test.go @@ -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) +} diff --git a/internal/app/run/runner.go b/internal/app/run/runner.go index f2ed4f7a..7dabc3a4 100644 --- a/internal/app/run/runner.go +++ b/internal/app/run/runner.go @@ -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(), diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index 95412e3b..0abee234 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -199,11 +199,29 @@ host: # If it's empty, $HOME/.cache/act/ will be used. workdir_parent: +# Optional local task-admission checks. Disabled by default. When enabled, low +# disk space or a failing script pauses new task fetching; existing jobs continue. +# No health checks run while any job is active; the last result is reused until idle. +health_check: + enabled: false + # Minimum free space required on the filesystem holding runner workspaces. + # Defaults to 1024 MiB when omitted or set to zero. + min_free_disk_space_mb: 1024 + # Optional additional executable. A non-zero exit, timeout, or startup failure + # marks the runner unavailable. + script: '' + # How long a script result is cached and its maximum execution time. + interval: 30s + timeout: 10s + metrics: # Enable the Prometheus metrics endpoint. - # When enabled, metrics are served at http:///metrics and a liveness check at /healthz. + # When enabled, metrics are served at /metrics, liveness at /healthz, and + # task-admission readiness at /readyz. enabled: false # The address for the metrics HTTP server to listen on. # Defaults to localhost only. Set to ":9101" to allow external access, # but ensure the port is firewall-protected as there is no authentication. addr: "127.0.0.1:9101" + # Consecutive polling failures may last this long before /readyz returns 503. + readiness_grace: 30s diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index 423ee9ed..c56b244e 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -95,18 +95,30 @@ type Host struct { // Metrics represents the configuration for the Prometheus metrics endpoint. type Metrics struct { - Enabled bool `yaml:"enabled"` // Enabled indicates whether the metrics endpoint is exposed. - Addr string `yaml:"addr"` // Addr specifies the listen address for the metrics HTTP server (e.g., ":9101"). + Enabled bool `yaml:"enabled"` // Enabled indicates whether the metrics endpoint is exposed. + Addr string `yaml:"addr"` // Addr specifies the listen address for the metrics HTTP server (e.g., ":9101"). + ReadinessGrace time.Duration `yaml:"readiness_grace"` // ReadinessGrace permits transient polling errors before /readyz becomes unhealthy. +} + +// HealthCheck represents local checks that control whether the runner accepts +// new tasks. The entire feature is opt-in through Enabled. +type HealthCheck struct { + Enabled bool `yaml:"enabled"` // Enabled activates local task-admission health checks. + MinFreeDiskSpaceMB int64 `yaml:"min_free_disk_space_mb"` // MinFreeDiskSpaceMB is the minimum free space required on the work volume. + Script string `yaml:"script"` // Script is an optional executable used as an additional health check. + Interval time.Duration `yaml:"interval"` // Interval controls how long a script result is cached. + Timeout time.Duration `yaml:"timeout"` // Timeout caps one health-check script invocation. } // Config represents the overall configuration. type Config struct { - Log Log `yaml:"log"` // Log represents the configuration for logging. - Runner Runner `yaml:"runner"` // Runner represents the configuration for the runner. - Cache Cache `yaml:"cache"` // Cache represents the configuration for caching. - Container Container `yaml:"container"` // Container represents the configuration for the container. - Host Host `yaml:"host"` // Host represents the configuration for the host. - Metrics Metrics `yaml:"metrics"` // Metrics represents the configuration for the Prometheus metrics endpoint. + Log Log `yaml:"log"` // Log represents the configuration for logging. + Runner Runner `yaml:"runner"` // Runner represents the configuration for the runner. + Cache Cache `yaml:"cache"` // Cache represents the configuration for caching. + Container Container `yaml:"container"` // Container represents the configuration for the container. + Host Host `yaml:"host"` // Host represents the configuration for the host. + Metrics Metrics `yaml:"metrics"` // Metrics represents the configuration for the Prometheus metrics endpoint. + HealthCheck HealthCheck `yaml:"health_check"` // HealthCheck controls opt-in local task-admission checks. } // LoadDefault returns the default configuration. @@ -220,9 +232,21 @@ func LoadDefault(file string) (*Config, error) { if cfg.Runner.PostTaskScript != "" && cfg.Runner.PostTaskScriptTimeout <= 0 { cfg.Runner.PostTaskScriptTimeout = DefaultPostTaskScriptTimeout } + if cfg.HealthCheck.MinFreeDiskSpaceMB <= 0 { + cfg.HealthCheck.MinFreeDiskSpaceMB = 1024 + } + if cfg.HealthCheck.Interval <= 0 { + cfg.HealthCheck.Interval = 30 * time.Second + } + if cfg.HealthCheck.Timeout <= 0 { + cfg.HealthCheck.Timeout = 10 * time.Second + } if cfg.Metrics.Addr == "" { cfg.Metrics.Addr = "127.0.0.1:9101" } + if cfg.Metrics.ReadinessGrace <= 0 { + cfg.Metrics.ReadinessGrace = 30 * time.Second + } // Validate and fix invalid config combinations to prevent confusing behavior. if cfg.Runner.FetchIntervalMax < cfg.Runner.FetchInterval { diff --git a/internal/pkg/config/config_test.go b/internal/pkg/config/config_test.go index 382aad48..f5aaa758 100644 --- a/internal/pkg/config/config_test.go +++ b/internal/pkg/config/config_test.go @@ -48,6 +48,41 @@ func TestLoadDefault_DefaultsWorkdirCleanupAge(t *testing.T) { assert.Equal(t, 10*time.Minute, cfg.Runner.IdleCleanupInterval) } +func TestLoadDefault_HealthChecksAreOptIn(t *testing.T) { + cfg, err := LoadDefault("") + require.NoError(t, err) + assert.False(t, cfg.HealthCheck.Enabled) + assert.Equal(t, int64(1024), cfg.HealthCheck.MinFreeDiskSpaceMB) + assert.Empty(t, cfg.HealthCheck.Script) + assert.Equal(t, 30*time.Second, cfg.HealthCheck.Interval) + assert.Equal(t, 10*time.Second, cfg.HealthCheck.Timeout) + assert.False(t, cfg.Metrics.Enabled) +} + +func TestLoadDefault_DiskAndReadinessSettings(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.yaml") + require.NoError(t, os.WriteFile(path, []byte(` +health_check: + enabled: true + min_free_disk_space_mb: 4096 + script: /usr/local/bin/runner-health + interval: 15s + timeout: 3s +metrics: + readiness_grace: 45s +`), 0o600)) + + cfg, err := LoadDefault(path) + require.NoError(t, err) + assert.True(t, cfg.HealthCheck.Enabled) + assert.Equal(t, int64(4096), cfg.HealthCheck.MinFreeDiskSpaceMB) + assert.Equal(t, "/usr/local/bin/runner-health", cfg.HealthCheck.Script) + assert.Equal(t, 15*time.Second, cfg.HealthCheck.Interval) + assert.Equal(t, 3*time.Second, cfg.HealthCheck.Timeout) + assert.Equal(t, 45*time.Second, cfg.Metrics.ReadinessGrace) +} + func TestLoadDefault_UsesConfiguredWorkdirCleanupAge(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "config.yaml") diff --git a/internal/pkg/metrics/metrics_test.go b/internal/pkg/metrics/metrics_test.go index 470c185c..cde3ece9 100644 --- a/internal/pkg/metrics/metrics_test.go +++ b/internal/pkg/metrics/metrics_test.go @@ -81,7 +81,7 @@ func TestRegisterRunningJobsFuncZeroCapacity(t *testing.T) { func TestStartServerCanBeCancelled(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) - StartServer(ctx, "127.0.0.1:0") + StartServer(ctx, "127.0.0.1:0", nil) cancel() } diff --git a/internal/pkg/metrics/server.go b/internal/pkg/metrics/server.go index 44a58bcd..9876e081 100644 --- a/internal/pkg/metrics/server.go +++ b/internal/pkg/metrics/server.go @@ -13,19 +13,12 @@ import ( ) // StartServer starts an HTTP server that serves Prometheus metrics on /metrics -// and a liveness check on /healthz. The server shuts down when ctx is cancelled. +// and health checks. The server shuts down when ctx is cancelled. // Call Init() before StartServer to register metrics with the Registry. -func StartServer(ctx context.Context, addr string) { - mux := http.NewServeMux() - mux.Handle("/metrics", promhttp.HandlerFor(Registry, promhttp.HandlerOpts{})) - mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { - w.WriteHeader(http.StatusOK) - _, _ = w.Write([]byte("ok")) - }) - +func StartServer(ctx context.Context, addr string, readiness func() (bool, string)) { srv := &http.Server{ Addr: addr, - Handler: mux, + Handler: NewHTTPHandler(readiness), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: 10 * time.Second, @@ -48,3 +41,25 @@ func StartServer(ctx context.Context, addr string) { } }() } + +// NewHTTPHandler returns the metrics and health endpoints used by StartServer. +func NewHTTPHandler(readiness func() (bool, string)) http.Handler { + mux := http.NewServeMux() + mux.Handle("/metrics", promhttp.HandlerFor(Registry, promhttp.HandlerOpts{})) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }) + mux.HandleFunc("/readyz", func(w http.ResponseWriter, _ *http.Request) { + ready, reason := true, "ok" + if readiness != nil { + ready, reason = readiness() + } + if !ready { + w.WriteHeader(http.StatusServiceUnavailable) + } + _, _ = w.Write([]byte(reason)) + }) + + return mux +} diff --git a/internal/pkg/metrics/server_test.go b/internal/pkg/metrics/server_test.go new file mode 100644 index 00000000..d3fa8a66 --- /dev/null +++ b/internal/pkg/metrics/server_test.go @@ -0,0 +1,43 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package metrics + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestReadyEndpoint(t *testing.T) { + tests := []struct { + name string + ready bool + reason string + status int + }{ + {"ready", true, "ok", http.StatusOK}, + {"not ready", false, "low disk space", http.StatusServiceUnavailable}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + handler := NewHTTPHandler(func() (bool, string) { return test.ready, test.reason }) + request := httptest.NewRequest(http.MethodGet, "/readyz", nil) + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + assert.Equal(t, test.status, response.Code) + assert.Equal(t, test.reason, response.Body.String()) + }) + } +} + +func TestHealthEndpointStaysLiveWhenNotReady(t *testing.T) { + handler := NewHTTPHandler(func() (bool, string) { return false, "low disk space" }) + request := httptest.NewRequest(http.MethodGet, "/healthz", nil) + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + assert.Equal(t, http.StatusOK, response.Code) + assert.Equal(t, "ok", response.Body.String()) +}