gitea/act_runner
1
0
Fork 0
mirror of https://gitea.com/gitea/act_runner.git synced 2026-04-25 05:10:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

Compare commits

base: gitea:v0.3.1
Branches Tags
gitea:main gitea:chore/upgrade-golangci-lint-v2.11.4 gitea:act-runner-merged gitea:act-runner-actions-oss-act gitea:fix-sec gitea:lunny/upgrade_act
gitea:v0.4.1 gitea:v0.4.0 gitea:v0.3.1 gitea:v0.3.0 gitea:v0.2.13 gitea:v0.2.12 gitea:v0.2.11 gitea:v0.2.10 gitea:v0.2.9 gitea:v0.2.8 gitea:v0.2.7 gitea:v0.2.6 gitea:v0.2.5 gitea:v0.2.4 gitea:v0.2.3 gitea:v0.2.2 gitea:v0.2.1 gitea:v0.2.0 gitea:v0.1.8 gitea:v0.1.7 gitea:v0.1.6 gitea:v0.1.5 gitea:v0.1.4 gitea:v0.1.3 gitea:v0.1.2 gitea:v0.1.1 gitea:v0.1.0 gitea:v0.0.1
..
compare: gitea:v0.4.0
Branches Tags
gitea:main gitea:chore/upgrade-golangci-lint-v2.11.4 gitea:act-runner-merged gitea:act-runner-actions-oss-act gitea:fix-sec gitea:lunny/upgrade_act
gitea:v0.4.1 gitea:v0.4.0 gitea:v0.3.1 gitea:v0.3.0 gitea:v0.2.13 gitea:v0.2.12 gitea:v0.2.11 gitea:v0.2.10 gitea:v0.2.9 gitea:v0.2.8 gitea:v0.2.7 gitea:v0.2.6 gitea:v0.2.5 gitea:v0.2.4 gitea:v0.2.3 gitea:v0.2.2 gitea:v0.2.1 gitea:v0.2.0 gitea:v0.1.8 gitea:v0.1.7 gitea:v0.1.6 gitea:v0.1.5 gitea:v0.1.4 gitea:v0.1.3 gitea:v0.1.2 gitea:v0.1.1 gitea:v0.1.0 gitea:v0.0.1

4 Commits
v0.3.1 ... v0.4.0

Author SHA1 Message Date
Bo-Yi Wu
40dcee0991 chore(deps): upgrade golangci-lint from v2.10.1 to v2.11.4 (#821) 
## Summary
- Bump golangci-lint from v2.10.1 to v2.11.4
- Remove unused `//nolint:revive` directive on metrics package declaration (detected by stricter nolintlint in new version)

## Changes between v2.10.1 and v2.11.4
- **v2.11.0** — Multiple linter dependency upgrades, Go 1.26 support
- **v2.11.2** — Bug fix for `fmt` with path
- **v2.11.3** — gosec update
- **v2.11.4** — Dependency updates (sqlclosecheck, noctx, etc.)

No breaking changes.

Reviewed-on: https://gitea.com/gitea/act_runner/pulls/821
Co-authored-by: Bo-Yi Wu <appleboy.tw@gmail.com>
Co-committed-by: Bo-Yi Wu <appleboy.tw@gmail.com>
 2026-04-15 03:56:34 +00:00
Bo-Yi Wu
f33e5a6245 feat: add Prometheus metrics endpoint for runner observability (#820) 
## What

Add an optional Prometheus `/metrics` HTTP endpoint to `act_runner` so operators can observe runner health, polling behavior, job outcomes, and RPC latency without scraping logs.

New surface:

- `internal/pkg/metrics/metrics.go` — metric definitions, custom `Registry`, static Go/process collectors, label constants, `ResultToStatusLabel` helper.
- `internal/pkg/metrics/server.go` — hardened `http.Server` serving `/metrics` and `/healthz` with Slowloris-safe timeouts (`ReadHeaderTimeout` 5s, `ReadTimeout`/`WriteTimeout` 10s, `IdleTimeout` 60s) and a 5s graceful shutdown.
- `daemon.go` wires it up behind `cfg.Metrics.Enabled` (disabled by default).
- `poller.go` / `reporter.go` / `runner.go` instrument their existing hot paths with counters/histograms/gauges — no behavior change.

Metrics exported (namespace `act_runner_`):

| Subsystem | Metric | Type | Labels |
|---|---|---|---|
| — | `info` | Gauge | `version`, `name` |
| — | `capacity`, `uptime_seconds` | Gauge | — |
| `poll` | `fetch_total`, `client_errors_total` | Counter | `result` / `method` |
| `poll` | `fetch_duration_seconds`, `backoff_seconds` | Histogram / Gauge | — |
| `job` | `total` | Counter | `status` |
| `job` | `duration_seconds`, `running`, `capacity_utilization_ratio` | Histogram / GaugeFunc | — |
| `report` | `log_total`, `state_total` | Counter | `result` |
| `report` | `log_duration_seconds`, `state_duration_seconds` | Histogram | — |
| `report` | `log_buffer_rows` | Gauge | — |
| — | `go_*`, `process_*` | standard collectors | — |

All label values are predefined constants — **no high-cardinality labels** (no task IDs, repo URLs, branches, tokens, or secrets) so scraping is safe and bounded.

## Why

Teams self-hosting Gitea + `act_runner` at scale need to answer basic SRE questions that are currently invisible:

- How often are RPCs failing? Which RPC? (`act_runner_client_errors_total`)
- Are runners saturated? (`act_runner_job_capacity_utilization_ratio`, `act_runner_job_running`)
- How long do jobs take? (`act_runner_job_duration_seconds`)
- Is polling backing off? (`act_runner_poll_backoff_seconds`, `act_runner_poll_fetch_total{result=\"error\"}`)
- Are log/state reports slow? (`act_runner_report_{log,state}_duration_seconds`)
- Is the log buffer draining? (`act_runner_report_log_buffer_rows`)

Today operators have to grep logs. This PR makes all of the above first-class metrics so they can feed dashboards and alerts (`rate(act_runner_client_errors_total[5m]) > 0.1`, capacity saturation alerts, etc.).

The endpoint is **disabled by default** and binds to `127.0.0.1:9101` when enabled, so it's opt-in and safe for existing deployments.

## How

### Config

```yaml
metrics:
  enabled: false           # opt-in
  addr: 127.0.0.1:9101     # change to 0.0.0.0:9101 only behind a reverse proxy
```

`config.example.yaml` documents both fields plus a security note about binding externally without auth.

### Wiring

1. `daemon.go` calls `metrics.Init()` (guarded by `sync.Once`), sets `act_runner_info`, `act_runner_capacity`, registers uptime + running-jobs GaugeFuncs, then starts the server goroutine with the daemon context — it shuts down cleanly on `ctx.Done()`.
2. `poller.fetchTask` observes RPC latency / result / error counters. `DeadlineExceeded` (long-poll idle) is treated as an empty result and **not** observed into the histogram so the 5s timeout doesn't swamp the buckets.
3. `poller.pollOnce` reports `poll_backoff_seconds` using the pre-jitter base interval (the true backoff level), and only when it changes — prevents noisy no-op gauge updates at the `FetchIntervalMax` plateau.
4. `reporter.ReportLog` / `ReportState` record duration histograms and success/error counters; `log_buffer_rows` is updated only when the value changes, guarded by the already-held `clientM`.
5. `runner.Run` observes `job_duration_seconds` and increments `job_total` by outcome via `metrics.ResultToStatusLabel`.

### Safety / security review

- All timeouts set; Slowloris-safe.
- Custom `prometheus.NewRegistry()` — no global registration side-effects.
- No sensitive data in labels (reviewed every instrumentation site).
- Single new dependency: `github.com/prometheus/client_golang v1.23.2`.
- Endpoint is unauthenticated by design and documented as such; default localhost bind mitigates exposure. Operators exposing externally should front it with a reverse proxy.

## Verification

### Unit tests

\`\`\`bash
go build ./...
go vet ./...
go test ./...
\`\`\`

### Manual smoke test

1. Enable metrics in `config.yaml`:
   \`\`\`yaml
   metrics:
     enabled: true
     addr: 127.0.0.1:9101
   \`\`\`
2. Start the runner against a Gitea instance: \`./act_runner daemon\`.
3. Scrape the endpoint:
   \`\`\`bash
   curl -s http://127.0.0.1:9101/metrics | grep '^act_runner_'
   curl -s http://127.0.0.1:9101/healthz   # → ok
   \`\`\`
4. Confirm the static series appear immediately: \`act_runner_info\`, \`act_runner_capacity\`, \`act_runner_uptime_seconds\`, \`act_runner_job_running\`, \`act_runner_job_capacity_utilization_ratio\`.
5. Trigger a workflow and confirm counters increment: \`act_runner_poll_fetch_total{result=\"task\"}\`, \`act_runner_job_total{status=\"success\"}\`, \`act_runner_report_log_total{result=\"success\"}\`.
6. Leave the runner idle and confirm \`act_runner_poll_backoff_seconds\` settles (and does **not** churn on every poll).
7. Ctrl-C and confirm a clean \"metrics server shutdown\" log line (no port-in-use error on restart within 5s).

### Prometheus integration

Add to \`prometheus.yml\`:

\`\`\`yaml
scrape_configs:
  - job_name: act_runner
    static_configs:
      - targets: ['127.0.0.1:9101']
\`\`\`

Sample alert to try:

\`\`\`
sum(rate(act_runner_client_errors_total[5m])) by (method) > 0.1
\`\`\`

## Out of scope (follow-ups)

- TLS and auth on the metrics endpoint (mitigated today by localhost default; add when operators need external scraping).
- Per-task labels (intentionally avoided for cardinality safety).

---

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Reviewed-on: https://gitea.com/gitea/act_runner/pulls/820
Reviewed-by: Lunny Xiao <xiaolunwen@gmail.com>
Co-authored-by: Bo-Yi Wu <appleboy.tw@gmail.com>
Co-committed-by: Bo-Yi Wu <appleboy.tw@gmail.com>
 2026-04-15 01:27:34 +00:00
Bo-Yi Wu
f2d545565f perf: reduce runner-to-server connection load with adaptive reporting and polling (#819) 
## Summary

Many teams self-host Gitea + Act Runner at scale. The current runner design causes excessive HTTP requests to the Gitea server, leading to high server load. This PR addresses three root causes: aggressive fixed-interval polling, per-task status reporting every 1 second regardless of activity, and unoptimized HTTP client configuration.

## Problem

The original architecture has these issues:

**1. Fixed 1-second reporting interval (RunDaemon)**

- Every running task calls ReportLog + ReportState every 1 second (2 HTTP requests/sec/task)
- These requests are sent even when there are no new log rows or state changes
- With 200 runners × 3 tasks each = **1,200 req/sec just for status reporting**

**2. Fixed 2-second polling interval (no backoff)**

- Idle runners poll FetchTask every 2 seconds forever, even when no jobs are queued
- No exponential backoff or jitter — all runners can synchronize after network recovery (thundering herd)
- 200 idle runners = **100 req/sec doing nothing useful**

**3. HTTP client not tuned**

- Uses http.DefaultClient with MaxIdleConnsPerHost=2, causing frequent TCP/TLS reconnects
- Creates two separate http.Client instances (one for Ping, one for Runner service) instead of sharing

**Total: ~1,300 req/sec for 200 runners with 3 tasks each**

## Solution

### Adaptive Event-Driven Log Reporting

Replace the recursive `time.AfterFunc(1s)` pattern in RunDaemon with a goroutine-based select event loop using three trigger mechanisms:

| Trigger | Default | Purpose |
|---------|---------|---------|
| `log_report_max_latency` | 3s | Guarantee even a single log line is delivered within this time |
| `log_report_interval` | 5s | Periodic sweep — steady-state cadence |
| `log_report_batch_size` | 100 rows | Immediate flush during bursty output (e.g., npm install) |

**Key design**: `log_report_max_latency` (3s) must be less than `log_report_interval` (5s) so the max-latency timer fires before the periodic ticker for single-line scenarios.

State reporting is decoupled to its own `state_report_interval` (default 5s), with immediate flush on step transitions (start/stop) via a stateNotify channel for responsive frontend UX.

Additionally:
- Skip ReportLog when `len(rows) == 0` (no pending log rows)
- Skip ReportState when `stateChanged == false && len(outputs) == 0` (nothing changed)
- Move expensive `proto.Clone` after the early-return check to avoid deep copies on no-op paths

### Polling Backoff with Jitter

Replace fixed `rate.Limiter` with adaptive exponential backoff:
- Track `consecutiveEmpty` and `consecutiveErrors` counters
- Interval doubles with each empty/error response: `base × 2^(n-1)`, capped at `fetch_interval_max` (default 60s)
- Add ±20% random jitter to prevent thundering herd
- Fetch first, sleep after ��� preserves burst=1 behavior for immediate first fetch on startup and after task completion

### HTTP Client Tuning

- Configure custom `http.Transport` with `MaxIdleConnsPerHost=10` (was 2)
- Share a single `http.Client` between PingService and RunnerService
- Add `IdleConnTimeout=90s` for clean connection lifecycle

## Load Reduction

For 200 runners × 3 tasks (70% with active log output):

| Component | Before | After | Reduction |
|-----------|--------|-------|-----------|
| Polling (idle) | 100 req/s | ~3.4 req/s | 97% |
| Log reporting | 420 req/s | ~84 req/s | 80% |
| State reporting | 126 req/s | ~25 req/s | 80% |
| **Total** | **~1,300 req/s** | **~113 req/s** | **~91%** |

## Frontend UX Impact

| Scenario | Before | After | Notes |
|----------|--------|-------|-------|
| Continuous output (npm install) | ~1s | ~5s | Periodic ticker sweep |
| Single line then silence | ~1s | ≤3s | maxLatencyTimer guarantee |
| Bursty output (100+ lines) | ~1s | <1s | Batch size immediate flush |
| Step start/stop | ~1s | <1s | stateNotify immediate flush |
| Job completion | ~1s | ~1s | Close() retry unchanged |

## New Configuration Options

All have safe defaults — existing config files need no changes:

```yaml
runner:
  fetch_interval_max: 60s        # Max backoff interval when idle
  log_report_interval: 5s        # Periodic log flush interval
  log_report_max_latency: 3s     # Max time a log row waits (must be < log_report_interval)
  log_report_batch_size: 100     # Immediate flush threshold
  state_report_interval: 5s      # State flush interval (step transitions are always immediate)
```

Config validation warns on invalid combinations:
- `fetch_interval_max < fetch_interval` → auto-corrected
- `log_report_max_latency >= log_report_interval` → warning (timer would be redundant)

## Test Plan

- [x] `go build ./...` passes
- [x] `go test ./...` passes (all existing + 3 new tests)
- [x] `golangci-lint run` — 0 issues
- [x] TestReporter_MaxLatencyTimer — verifies single log line flushed by maxLatencyTimer before logTicker
- [x] TestReporter_BatchSizeFlush — verifies batch size threshold triggers immediate flush
- [x] TestReporter_StateNotifyFlush — verifies step transition triggers immediate state flush
- [x] TestReporter_EphemeralRunnerDeletion — verifies Close/RunDaemon race safety
- [x] TestReporter_RunDaemonClose_Race — verifies concurrent Close safety

Reviewed-on: https://gitea.com/gitea/act_runner/pulls/819
Reviewed-by: Nicolas <173651+bircni@noreply.gitea.com>
Co-authored-by: Bo-Yi Wu <appleboy.tw@gmail.com>
Co-committed-by: Bo-Yi Wu <appleboy.tw@gmail.com>
 2026-04-14 11:29:25 +00:00
Lunny Xiao
90c1275f0e Upgrade yaml (#816) 
~wait https://gitea.com/gitea/act/pulls/157~

Reviewed-on: https://gitea.com/gitea/act_runner/pulls/816
Reviewed-by: Zettat123 <39446+zettat123@noreply.gitea.com>
 2026-03-28 16:18:47 +00:00
14 changed files with 1091 additions and 93 deletions
Expand all files Collapse all files

2
Makefile
View File

@@ -20,7 +20,7 @@ DOCKER_TAG ?= nightly
DOCKER_REF := $(DOCKER_IMAGE):$(DOCKER_TAG) DOCKER_REF := $(DOCKER_IMAGE):$(DOCKER_TAG)
DOCKER_ROOTLESS_REF := $(DOCKER_IMAGE):$(DOCKER_TAG)-dind-rootless DOCKER_ROOTLESS_REF := $(DOCKER_IMAGE):$(DOCKER_TAG)-dind-rootless
GOLANGCI_LINT_PACKAGE ?= github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.10.1 GOLANGCI_LINT_PACKAGE ?= github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.11.4
GOVULNCHECK_PACKAGE ?= golang.org/x/vuln/cmd/govulncheck@v1 GOVULNCHECK_PACKAGE ?= golang.org/x/vuln/cmd/govulncheck@v1
ifneq ($(shell uname), Darwin) ifneq ($(shell uname), Darwin)

13
go.mod
View File

@@ -14,14 +14,15 @@ require (
github.com/sirupsen/logrus v1.9.4 github.com/sirupsen/logrus v1.9.4
github.com/spf13/cobra v1.10.2 github.com/spf13/cobra v1.10.2
github.com/stretchr/testify v1.11.1 github.com/stretchr/testify v1.11.1
go.yaml.in/yaml/v4 v4.0.0-rc.3
golang.org/x/term v0.40.0 golang.org/x/term v0.40.0
golang.org/x/time v0.14.0 golang.org/x/time v0.14.0 // indirect
google.golang.org/protobuf v1.36.11 google.golang.org/protobuf v1.36.11
gopkg.in/yaml.v3 v3.0.1 gopkg.in/yaml.v3 v3.0.1
gotest.tools/v3 v3.5.2 gotest.tools/v3 v3.5.2
) )
require go.yaml.in/yaml/v4 v4.0.0-rc.3 require github.com/prometheus/client_golang v1.23.2
require ( require (
cyphar.com/go-pathrs v0.2.3 // indirect cyphar.com/go-pathrs v0.2.3 // indirect
@@ -30,6 +31,7 @@ require (
github.com/Masterminds/semver v1.5.0 // indirect github.com/Masterminds/semver v1.5.0 // indirect
github.com/Microsoft/go-winio v0.6.2 // indirect github.com/Microsoft/go-winio v0.6.2 // indirect
github.com/ProtonMail/go-crypto v1.3.0 // indirect github.com/ProtonMail/go-crypto v1.3.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/bmatcuk/doublestar/v4 v4.10.0 // indirect github.com/bmatcuk/doublestar/v4 v4.10.0 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/clipperhouse/uax29/v2 v2.7.0 // indirect github.com/clipperhouse/uax29/v2 v2.7.0 // indirect
@@ -76,12 +78,16 @@ require (
github.com/moby/sys/user v0.4.0 // indirect github.com/moby/sys/user v0.4.0 // indirect
github.com/moby/sys/userns v0.1.0 // indirect github.com/moby/sys/userns v0.1.0 // indirect
github.com/moby/term v0.5.2 // indirect github.com/moby/term v0.5.2 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect
github.com/opencontainers/image-spec v1.1.1 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect
github.com/opencontainers/selinux v1.13.1 // indirect github.com/opencontainers/selinux v1.13.1 // indirect
github.com/pjbgf/sha1cd v0.5.0 // indirect github.com/pjbgf/sha1cd v0.5.0 // indirect
github.com/pkg/errors v0.9.1 // indirect github.com/pkg/errors v0.9.1 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/prometheus/client_model v0.6.2 // indirect
github.com/prometheus/common v0.66.1 // indirect
github.com/prometheus/procfs v0.16.1 // indirect
github.com/rhysd/actionlint v1.7.11 // indirect github.com/rhysd/actionlint v1.7.11 // indirect
github.com/robfig/cron/v3 v3.0.1 // indirect github.com/robfig/cron/v3 v3.0.1 // indirect
github.com/sergi/go-diff v1.4.0 // indirect github.com/sergi/go-diff v1.4.0 // indirect
@@ -99,6 +105,7 @@ require (
go.opentelemetry.io/otel v1.40.0 // indirect go.opentelemetry.io/otel v1.40.0 // indirect
go.opentelemetry.io/otel/metric v1.40.0 // indirect go.opentelemetry.io/otel/metric v1.40.0 // indirect
go.opentelemetry.io/otel/trace v1.40.0 // indirect go.opentelemetry.io/otel/trace v1.40.0 // indirect
go.yaml.in/yaml/v2 v2.4.2 // indirect
golang.org/x/crypto v0.48.0 // indirect golang.org/x/crypto v0.48.0 // indirect
golang.org/x/net v0.50.0 // indirect golang.org/x/net v0.50.0 // indirect
golang.org/x/sync v0.19.0 // indirect golang.org/x/sync v0.19.0 // indirect
@@ -110,7 +117,7 @@ require (
gopkg.in/yaml.v2 v2.4.0 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect
) )
replace github.com/nektos/act => gitea.com/gitea/act v0.261.8 replace github.com/nektos/act => gitea.com/gitea/act v0.261.10
// Remove after github.com/docker/distribution is updated to support distribution/reference v0.6.0 // Remove after github.com/docker/distribution is updated to support distribution/reference v0.6.0
// (pulled in via moby/buildkit, breaks on undefined: reference.SplitHostname) // (pulled in via moby/buildkit, breaks on undefined: reference.SplitHostname)

22
go.sum
View File

@@ -8,8 +8,8 @@ cyphar.com/go-pathrs v0.2.3 h1:0pH8gep37wB0BgaXrEaN1OtZhUMeS7VvaejSr6i822o=
cyphar.com/go-pathrs v0.2.3/go.mod h1:y8f1EMG7r+hCuFf/rXsKqMJrJAUoADZGNh5/vZPKcGc= cyphar.com/go-pathrs v0.2.3/go.mod h1:y8f1EMG7r+hCuFf/rXsKqMJrJAUoADZGNh5/vZPKcGc=
dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8=
dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA=
gitea.com/gitea/act v0.261.8 h1:rUWB5GOZOubfe2VteKb7XP3HRIbcW3UUmfh7bVAgQcA= gitea.com/gitea/act v0.261.10 h1:ndwbtuMXXz1dpYF2iwY1/PkgKNETo4jmPXfinTZt8cs=
gitea.com/gitea/act v0.261.8/go.mod h1:lTp4136rwbZiZS3ZVQeHCvd4qRAZ7LYeiRBqOSdMY/4= gitea.com/gitea/act v0.261.10/go.mod h1:oIkqQHvU0lfuIWwcpqa4FmU+t3prA89tgkuHUTsrI2c=
github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk=
github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8=
github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg= github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg=
@@ -29,6 +29,8 @@ github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPd
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs= github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
github.com/avast/retry-go/v4 v4.7.0 h1:yjDs35SlGvKwRNSykujfjdMxMhMQQM0TnIjJaHB+Zio= github.com/avast/retry-go/v4 v4.7.0 h1:yjDs35SlGvKwRNSykujfjdMxMhMQQM0TnIjJaHB+Zio=
github.com/avast/retry-go/v4 v4.7.0/go.mod h1:ZMPDa3sY2bKgpLtap9JRUgk2yTAba7cgiFhqxY2Sg6Q= github.com/avast/retry-go/v4 v4.7.0/go.mod h1:ZMPDa3sY2bKgpLtap9JRUgk2yTAba7cgiFhqxY2Sg6Q=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/bmatcuk/doublestar/v4 v4.10.0 h1:zU9WiOla1YA122oLM6i4EXvGW62DvKZVxIe6TYWexEs= github.com/bmatcuk/doublestar/v4 v4.10.0 h1:zU9WiOla1YA122oLM6i4EXvGW62DvKZVxIe6TYWexEs=
github.com/bmatcuk/doublestar/v4 v4.10.0/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc= github.com/bmatcuk/doublestar/v4 v4.10.0/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc=
github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM= github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
@@ -129,6 +131,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE=
github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
@@ -153,6 +157,8 @@ github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ=
github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc=
github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A=
github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/onsi/gomega v1.34.1 h1:EUMJIKUjM8sKjYbtxQI9A4z2o+rruxnzNvpknOXie6k= github.com/onsi/gomega v1.34.1 h1:EUMJIKUjM8sKjYbtxQI9A4z2o+rruxnzNvpknOXie6k=
github.com/onsi/gomega v1.34.1/go.mod h1:kU1QgUvBDLXBJq618Xvm2LUX6rSAfRaFRTcdOeDLwwY= github.com/onsi/gomega v1.34.1/go.mod h1:kU1QgUvBDLXBJq618Xvm2LUX6rSAfRaFRTcdOeDLwwY=
github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
@@ -167,6 +173,14 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs=
github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
github.com/rhysd/actionlint v1.7.11 h1:m+aSuCpCIClS8X02xMG4Z8s87fCHPsAtYkAoWGQZgEE= github.com/rhysd/actionlint v1.7.11 h1:m+aSuCpCIClS8X02xMG4Z8s87fCHPsAtYkAoWGQZgEE=
github.com/rhysd/actionlint v1.7.11/go.mod h1:8n50YougV9+50niD7oxgDTZ1KbN/ZnKiQ2xpLFeVhsI= github.com/rhysd/actionlint v1.7.11/go.mod h1:8n50YougV9+50niD7oxgDTZ1KbN/ZnKiQ2xpLFeVhsI=
github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
@@ -237,6 +251,10 @@ go.opentelemetry.io/otel/trace v1.40.0 h1:WA4etStDttCSYuhwvEa8OP8I5EWu24lkOzp+ZY
go.opentelemetry.io/otel/trace v1.40.0/go.mod h1:zeAhriXecNGP/s2SEG3+Y8X9ujcJOTqQ5RgdEJcawiA= go.opentelemetry.io/otel/trace v1.40.0/go.mod h1:zeAhriXecNGP/s2SEG3+Y8X9ujcJOTqQ5RgdEJcawiA=
go.opentelemetry.io/proto/otlp v1.0.0 h1:T0TX0tmXU8a3CbNXzEKGeU5mIVOdf0oykP+u2lIVU/I= go.opentelemetry.io/proto/otlp v1.0.0 h1:T0TX0tmXU8a3CbNXzEKGeU5mIVOdf0oykP+u2lIVU/I=
go.opentelemetry.io/proto/otlp v1.0.0/go.mod h1:Sy6pihPLfYHkr3NkUbEhGHFhINUSI/v80hjKIs5JXpM= go.opentelemetry.io/proto/otlp v1.0.0/go.mod h1:Sy6pihPLfYHkr3NkUbEhGHFhINUSI/v80hjKIs5JXpM=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
go.yaml.in/yaml/v4 v4.0.0-rc.3 h1:3h1fjsh1CTAPjW7q/EMe+C8shx5d8ctzZTrLcs/j8Go= go.yaml.in/yaml/v4 v4.0.0-rc.3 h1:3h1fjsh1CTAPjW7q/EMe+C8shx5d8ctzZTrLcs/j8Go=
go.yaml.in/yaml/v4 v4.0.0-rc.3/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0= go.yaml.in/yaml/v4 v4.0.0-rc.3/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0=

10
internal/app/cmd/daemon.go
View File

@@ -27,6 +27,7 @@ import (
"gitea.com/gitea/act_runner/internal/pkg/config" "gitea.com/gitea/act_runner/internal/pkg/config"
"gitea.com/gitea/act_runner/internal/pkg/envcheck" "gitea.com/gitea/act_runner/internal/pkg/envcheck"
"gitea.com/gitea/act_runner/internal/pkg/labels" "gitea.com/gitea/act_runner/internal/pkg/labels"
"gitea.com/gitea/act_runner/internal/pkg/metrics"
"gitea.com/gitea/act_runner/internal/pkg/ver" "gitea.com/gitea/act_runner/internal/pkg/ver"
) )
@@ -149,6 +150,15 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
resp.Msg.Runner.Name, resp.Msg.Runner.Version, resp.Msg.Runner.Labels) resp.Msg.Runner.Name, resp.Msg.Runner.Version, resp.Msg.Runner.Labels)
} }
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)
}
poller := poll.New(cfg, cli, runner) poller := poll.New(cfg, cli, runner)
if daemArgs.Once || reg.Ephemeral { if daemArgs.Once || reg.Ephemeral {

114
internal/app/poll/poller.go
View File

@@ -7,17 +7,19 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"math/rand/v2"
"sync" "sync"
"sync/atomic" "sync/atomic"
"time"
runnerv1 "code.gitea.io/actions-proto-go/runner/v1" runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
"connectrpc.com/connect" "connectrpc.com/connect"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"golang.org/x/time/rate"
"gitea.com/gitea/act_runner/internal/app/run" "gitea.com/gitea/act_runner/internal/app/run"
"gitea.com/gitea/act_runner/internal/pkg/client" "gitea.com/gitea/act_runner/internal/pkg/client"
"gitea.com/gitea/act_runner/internal/pkg/config" "gitea.com/gitea/act_runner/internal/pkg/config"
"gitea.com/gitea/act_runner/internal/pkg/metrics"
) )
type Poller struct { type Poller struct {
@@ -35,6 +37,19 @@ type Poller struct {
done chan struct{} done chan struct{}
} }
// workerState holds per-goroutine polling state. Backoff counters are
// per-worker so that with Capacity > 1, N workers each seeing one empty
// response don't combine into a "consecutive N empty" reading on a shared
// counter and trigger an unnecessarily long backoff.
type workerState struct {
consecutiveEmpty int64
consecutiveErrors int64
// lastBackoff is the last interval reported to the PollBackoffSeconds gauge
// from this worker; used to suppress redundant no-op Set calls when the
// backoff plateaus (e.g. at FetchIntervalMax).
lastBackoff time.Duration
}
func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller { func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller {
pollingCtx, shutdownPolling := context.WithCancel(context.Background()) pollingCtx, shutdownPolling := context.WithCancel(context.Background())
@@ -58,11 +73,10 @@ func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller {
} }
func (p *Poller) Poll() { func (p *Poller) Poll() {
limiter := rate.NewLimiter(rate.Every(p.cfg.Runner.FetchInterval), 1)
wg := &sync.WaitGroup{} wg := &sync.WaitGroup{}
for i := 0; i < p.cfg.Runner.Capacity; i++ { for i := 0; i < p.cfg.Runner.Capacity; i++ {
wg.Add(1) wg.Add(1)
go p.poll(wg, limiter) go p.poll(wg)
} }
wg.Wait() wg.Wait()
@@ -71,9 +85,7 @@ func (p *Poller) Poll() {
} }
func (p *Poller) PollOnce() { func (p *Poller) PollOnce() {
limiter := rate.NewLimiter(rate.Every(p.cfg.Runner.FetchInterval), 1) p.pollOnce(&workerState{})
p.pollOnce(limiter)
// signal that we're done // signal that we're done
close(p.done) close(p.done)
@@ -108,10 +120,11 @@ func (p *Poller) Shutdown(ctx context.Context) error {
} }
} }
func (p *Poller) poll(wg *sync.WaitGroup, limiter *rate.Limiter) { func (p *Poller) poll(wg *sync.WaitGroup) {
defer wg.Done() defer wg.Done()
s := &workerState{}
for { for {
p.pollOnce(limiter) p.pollOnce(s)
select { select {
case <-p.pollingCtx.Done(): case <-p.pollingCtx.Done():
@@ -122,19 +135,61 @@ func (p *Poller) poll(wg *sync.WaitGroup, limiter *rate.Limiter) {
} }
} }
func (p *Poller) pollOnce(limiter *rate.Limiter) { // calculateInterval returns the polling interval with exponential backoff based on
// consecutive empty or error responses. The interval starts at FetchInterval and
// doubles with each consecutive empty/error, capped at FetchIntervalMax.
func (p *Poller) calculateInterval(s *workerState) time.Duration {
base := p.cfg.Runner.FetchInterval
maxInterval := p.cfg.Runner.FetchIntervalMax
n := max(s.consecutiveEmpty, s.consecutiveErrors)
if n <= 1 {
return base
}
// Capped exponential backoff: base * 2^(n-1), max shift=5 so multiplier <= 32
shift := min(n-1, 5)
interval := base * time.Duration(int64(1)<<shift)
return min(interval, maxInterval)
}
// addJitter adds +/- 20% random jitter to the given duration to avoid thundering herd.
func addJitter(d time.Duration) time.Duration {
if d <= 0 {
return d
}
// jitter range: [-20%, +20%] of d
jitterRange := int64(d) * 2 / 5 // 40% total range
if jitterRange <= 0 {
return d
}
jitter := rand.Int64N(jitterRange) - jitterRange/2
return d + time.Duration(jitter)
}
func (p *Poller) pollOnce(s *workerState) {
for { for {
if err := limiter.Wait(p.pollingCtx); err != nil { task, ok := p.fetchTask(p.pollingCtx, s)
if p.pollingCtx.Err() != nil {
log.WithError(err).Debug("limiter wait failed")
}
return
}
task, ok := p.fetchTask(p.pollingCtx)
if !ok { if !ok {
base := p.calculateInterval(s)
if base != s.lastBackoff {
metrics.PollBackoffSeconds.Set(base.Seconds())
s.lastBackoff = base
}
timer := time.NewTimer(addJitter(base))
select {
case <-timer.C:
case <-p.pollingCtx.Done():
timer.Stop()
return
}
continue continue
} }
// Got a task — reset backoff counters for fast subsequent polling.
s.consecutiveEmpty = 0
s.consecutiveErrors = 0
p.runTaskWithRecover(p.jobsCtx, task) p.runTaskWithRecover(p.jobsCtx, task)
return return
} }
@@ -153,24 +208,42 @@ func (p *Poller) runTaskWithRecover(ctx context.Context, task *runnerv1.Task) {
} }
} }
func (p *Poller) fetchTask(ctx context.Context) (*runnerv1.Task, bool) { func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task, bool) {
reqCtx, cancel := context.WithTimeout(ctx, p.cfg.Runner.FetchTimeout) reqCtx, cancel := context.WithTimeout(ctx, p.cfg.Runner.FetchTimeout)
defer cancel() defer cancel()
// Load the version value that was in the cache when the request was sent. // Load the version value that was in the cache when the request was sent.
v := p.tasksVersion.Load() v := p.tasksVersion.Load()
start := time.Now()
resp, err := p.client.FetchTask(reqCtx, connect.NewRequest(&runnerv1.FetchTaskRequest{ resp, err := p.client.FetchTask(reqCtx, connect.NewRequest(&runnerv1.FetchTaskRequest{
TasksVersion: v, TasksVersion: v,
})) }))
// DeadlineExceeded is the designed idle path for a long-poll: the server
// 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) { if errors.Is(err, context.DeadlineExceeded) {
err = nil s.consecutiveEmpty++
s.consecutiveErrors = 0 // timeout is a healthy idle response
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc()
return nil, false
} }
metrics.PollFetchDuration.Observe(time.Since(start).Seconds())
if err != nil { if err != nil {
log.WithError(err).Error("failed to fetch task") log.WithError(err).Error("failed to fetch task")
s.consecutiveErrors++
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultError).Inc()
metrics.ClientErrors.WithLabelValues(metrics.LabelMethodFetchTask).Inc()
return nil, false return nil, false
} }
// Successful response — reset error counter.
s.consecutiveErrors = 0
if resp == nil || resp.Msg == nil { if resp == nil || resp.Msg == nil {
s.consecutiveEmpty++
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc()
return nil, false return nil, false
} }
@@ -179,11 +252,14 @@ func (p *Poller) fetchTask(ctx context.Context) (*runnerv1.Task, bool) {
} }
if resp.Msg.Task == nil { if resp.Msg.Task == nil {
s.consecutiveEmpty++
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc()
return nil, false return nil, false
} }
// got a task, set `tasksVersion` to zero to focre query db in next request. // got a task, set `tasksVersion` to zero to force query db in next request.
p.tasksVersion.CompareAndSwap(resp.Msg.TasksVersion, 0) p.tasksVersion.CompareAndSwap(resp.Msg.TasksVersion, 0)
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultTask).Inc()
return resp.Msg.Task, true return resp.Msg.Task, true
} }

108
internal/app/poll/poller_test.go Normal file
View File

@@ -0,0 +1,108 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package poll
import (
"context"
"errors"
"testing"
"time"
runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
connect_go "connectrpc.com/connect"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
"gitea.com/gitea/act_runner/internal/pkg/client/mocks"
"gitea.com/gitea/act_runner/internal/pkg/config"
)
// TestPoller_PerWorkerCounters verifies that each worker maintains its own
// backoff counters. With a shared counter, N workers each seeing one empty
// response would inflate the counter to N and trigger an unnecessarily long
// backoff. With per-worker state, each worker only sees its own count.
func TestPoller_PerWorkerCounters(t *testing.T) {
client := mocks.NewClient(t)
client.On("FetchTask", mock.Anything, mock.Anything).Return(
func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) {
// Always return an empty response.
return connect_go.NewResponse(&runnerv1.FetchTaskResponse{}), nil
},
)
cfg, err := config.LoadDefault("")
require.NoError(t, err)
p := &Poller{client: client, cfg: cfg}
ctx := context.Background()
s1 := &workerState{}
s2 := &workerState{}
// Each worker independently observes one empty response.
_, ok := p.fetchTask(ctx, s1)
require.False(t, ok)
_, ok = p.fetchTask(ctx, s2)
require.False(t, ok)
assert.Equal(t, int64(1), s1.consecutiveEmpty, "worker 1 should only count its own empty response")
assert.Equal(t, int64(1), s2.consecutiveEmpty, "worker 2 should only count its own empty response")
// Worker 1 sees a second empty; worker 2 stays at 1.
_, ok = p.fetchTask(ctx, s1)
require.False(t, ok)
assert.Equal(t, int64(2), s1.consecutiveEmpty)
assert.Equal(t, int64(1), s2.consecutiveEmpty, "worker 2's counter must not be affected by worker 1's empty fetches")
}
// TestPoller_FetchErrorIncrementsErrorsOnly verifies that a fetch error
// increments only the per-worker error counter, not the empty counter.
func TestPoller_FetchErrorIncrementsErrorsOnly(t *testing.T) {
client := mocks.NewClient(t)
client.On("FetchTask", mock.Anything, mock.Anything).Return(
func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) {
return nil, errors.New("network unreachable")
},
)
cfg, err := config.LoadDefault("")
require.NoError(t, err)
p := &Poller{client: client, cfg: cfg}
s := &workerState{}
_, ok := p.fetchTask(context.Background(), s)
require.False(t, ok)
assert.Equal(t, int64(1), s.consecutiveErrors)
assert.Equal(t, int64(0), s.consecutiveEmpty)
}
// TestPoller_CalculateInterval verifies the per-worker exponential backoff
// math is correctly driven by the worker's own counters.
func TestPoller_CalculateInterval(t *testing.T) {
cfg, err := config.LoadDefault("")
require.NoError(t, err)
cfg.Runner.FetchInterval = 2 * time.Second
cfg.Runner.FetchIntervalMax = 60 * time.Second
p := &Poller{cfg: cfg}
cases := []struct {
name string
empty, errs int64
wantInterval time.Duration
}{
{"first poll, no backoff", 0, 0, 2 * time.Second},
{"single empty, still base", 1, 0, 2 * time.Second},
{"two empties, doubled", 2, 0, 4 * time.Second},
{"five empties, capped path", 5, 0, 32 * time.Second},
{"many empties, capped at max", 20, 0, 60 * time.Second},
{"errors drive backoff too", 0, 3, 8 * time.Second},
{"max(empty, errors) wins", 2, 4, 16 * time.Second},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
s := &workerState{consecutiveEmpty: tc.empty, consecutiveErrors: tc.errs}
assert.Equal(t, tc.wantInterval, p.calculateInterval(s))
})
}
}

18
internal/app/run/runner.go
View File

@@ -12,6 +12,7 @@ import (
"path/filepath" "path/filepath"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"time" "time"
runnerv1 "code.gitea.io/actions-proto-go/runner/v1" runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
@@ -26,6 +27,7 @@ import (
"gitea.com/gitea/act_runner/internal/pkg/client" "gitea.com/gitea/act_runner/internal/pkg/client"
"gitea.com/gitea/act_runner/internal/pkg/config" "gitea.com/gitea/act_runner/internal/pkg/config"
"gitea.com/gitea/act_runner/internal/pkg/labels" "gitea.com/gitea/act_runner/internal/pkg/labels"
"gitea.com/gitea/act_runner/internal/pkg/metrics"
"gitea.com/gitea/act_runner/internal/pkg/report" "gitea.com/gitea/act_runner/internal/pkg/report"
"gitea.com/gitea/act_runner/internal/pkg/ver" "gitea.com/gitea/act_runner/internal/pkg/ver"
) )
@@ -41,6 +43,7 @@ type Runner struct {
envs map[string]string envs map[string]string
runningTasks sync.Map runningTasks sync.Map
runningCount atomic.Int64
} }
func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client) *Runner { func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client) *Runner {
@@ -96,16 +99,25 @@ func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) error {
r.runningTasks.Store(task.Id, struct{}{}) r.runningTasks.Store(task.Id, struct{}{})
defer r.runningTasks.Delete(task.Id) defer r.runningTasks.Delete(task.Id)
r.runningCount.Add(1)
start := time.Now()
ctx, cancel := context.WithTimeout(ctx, r.cfg.Runner.Timeout) ctx, cancel := context.WithTimeout(ctx, r.cfg.Runner.Timeout)
defer cancel() defer cancel()
reporter := report.NewReporter(ctx, cancel, r.client, task) reporter := report.NewReporter(ctx, cancel, r.client, task, r.cfg)
var runErr error var runErr error
defer func() { defer func() {
r.runningCount.Add(-1)
lastWords := "" lastWords := ""
if runErr != nil { if runErr != nil {
lastWords = runErr.Error() lastWords = runErr.Error()
} }
_ = reporter.Close(lastWords) _ = reporter.Close(lastWords)
metrics.JobDuration.Observe(time.Since(start).Seconds())
metrics.JobsTotal.WithLabelValues(metrics.ResultToStatusLabel(reporter.Result())).Inc()
}() }()
reporter.RunDaemon() reporter.RunDaemon()
runErr = r.run(ctx, task, reporter) runErr = r.run(ctx, task, reporter)
@@ -266,6 +278,10 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
return execErr return execErr
} }
func (r *Runner) RunningCount() int64 {
return r.runningCount.Load()
}
func (r *Runner) Declare(ctx context.Context, labels []string) (*connect.Response[runnerv1.DeclareResponse], error) { func (r *Runner) Declare(ctx context.Context, labels []string) (*connect.Response[runnerv1.DeclareResponse], error) {
return r.client.Declare(ctx, connect.NewRequest(&runnerv1.DeclareRequest{ return r.client.Declare(ctx, connect.NewRequest(&runnerv1.DeclareRequest{
Version: ver.Version(), Version: ver.Version(),

21
internal/pkg/client/http.go
View File

@@ -8,6 +8,7 @@ import (
"crypto/tls" "crypto/tls"
"net/http" "net/http"
"strings" "strings"
"time"
"code.gitea.io/actions-proto-go/ping/v1/pingv1connect" "code.gitea.io/actions-proto-go/ping/v1/pingv1connect"
"code.gitea.io/actions-proto-go/runner/v1/runnerv1connect" "code.gitea.io/actions-proto-go/runner/v1/runnerv1connect"
@@ -15,16 +16,17 @@ import (
) )
func getHTTPClient(endpoint string, insecure bool) *http.Client { func getHTTPClient(endpoint string, insecure bool) *http.Client {
transport := &http.Transport{
MaxIdleConns: 10,
MaxIdleConnsPerHost: 10, // All requests go to one host; default is 2 which causes frequent reconnects.
IdleConnTimeout: 90 * time.Second,
}
if strings.HasPrefix(endpoint, "https://") && insecure { if strings.HasPrefix(endpoint, "https://") && insecure {
return &http.Client{ transport.TLSClientConfig = &tls.Config{
Transport: &http.Transport{ InsecureSkipVerify: true,
TLSClientConfig: &tls.Config{
InsecureSkipVerify: true,
},
},
} }
} }
return http.DefaultClient return &http.Client{Transport: transport}
} }
// New returns a new runner client. // New returns a new runner client.
@@ -47,14 +49,15 @@ func New(endpoint string, insecure bool, uuid, token, version string, opts ...co
} }
}))) })))
httpClient := getHTTPClient(endpoint, insecure)
return &HTTPClient{ return &HTTPClient{
PingServiceClient: pingv1connect.NewPingServiceClient( PingServiceClient: pingv1connect.NewPingServiceClient(
getHTTPClient(endpoint, insecure), httpClient,
baseURL, baseURL,
opts..., opts...,
), ),
RunnerServiceClient: runnerv1connect.NewRunnerServiceClient( RunnerServiceClient: runnerv1connect.NewRunnerServiceClient(
getHTTPClient(endpoint, insecure), httpClient,
baseURL, baseURL,
opts..., opts...,
), ),

27
internal/pkg/config/config.example.yaml
View File

@@ -32,6 +32,24 @@ runner:
fetch_timeout: 5s fetch_timeout: 5s
# The interval for fetching the job from the Gitea instance. # The interval for fetching the job from the Gitea instance.
fetch_interval: 2s fetch_interval: 2s
# The maximum interval for fetching the job from the Gitea instance.
# The runner uses exponential backoff when idle, increasing the interval up to this maximum.
# Set to 0 or same as fetch_interval to disable backoff.
fetch_interval_max: 60s
# The base interval for periodic log flush to the Gitea instance.
# Logs may be sent earlier if the buffer reaches log_report_batch_size
# or if log_report_max_latency expires after the first buffered row.
log_report_interval: 5s
# The maximum time a log row can wait before being sent.
# This ensures even a single log line appears on the frontend within this duration.
# Must be less than log_report_interval to have any effect.
log_report_max_latency: 3s
# Flush logs immediately when the buffer reaches this many rows.
# This ensures bursty output (e.g., npm install) is delivered promptly.
log_report_batch_size: 100
# The interval for reporting task state (step status, timing) to the Gitea instance.
# State is also reported immediately on step transitions (start/stop).
state_report_interval: 5s
# The github_mirror of a runner is used to specify the mirror address of the github that pulls the action repository. # The github_mirror of a runner is used to specify the mirror address of the github that pulls the action repository.
# It works when something like `uses: actions/checkout@v4` is used and DEFAULT_ACTIONS_URL is set to github, # It works when something like `uses: actions/checkout@v4` is used and DEFAULT_ACTIONS_URL is set to github,
# and github_mirror is not empty. In this case, # and github_mirror is not empty. In this case,
@@ -114,3 +132,12 @@ host:
# The parent directory of a job's working directory. # The parent directory of a job's working directory.
# If it's empty, $HOME/.cache/act/ will be used. # If it's empty, $HOME/.cache/act/ will be used.
workdir_parent: workdir_parent:
metrics:
# Enable the Prometheus metrics endpoint.
# When enabled, metrics are served at http://<addr>/metrics and a liveness check at /healthz.
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"

63
internal/pkg/config/config.go
View File

@@ -22,17 +22,22 @@ type Log struct {
// Runner represents the configuration for the runner. // Runner represents the configuration for the runner.
type Runner struct { type Runner struct {
File string `yaml:"file"` // File specifies the file path for the runner. File string `yaml:"file"` // File specifies the file path for the runner.
Capacity int `yaml:"capacity"` // Capacity specifies the capacity of the runner. Capacity int `yaml:"capacity"` // Capacity specifies the capacity of the runner.
Envs map[string]string `yaml:"envs"` // Envs stores environment variables for the runner. Envs map[string]string `yaml:"envs"` // Envs stores environment variables for the runner.
EnvFile string `yaml:"env_file"` // EnvFile specifies the path to the file containing environment variables for the runner. EnvFile string `yaml:"env_file"` // EnvFile specifies the path to the file containing environment variables for the runner.
Timeout time.Duration `yaml:"timeout"` // Timeout specifies the duration for runner timeout. Timeout time.Duration `yaml:"timeout"` // Timeout specifies the duration for runner timeout.
ShutdownTimeout time.Duration `yaml:"shutdown_timeout"` // ShutdownTimeout specifies the duration to wait for running jobs to complete during a shutdown of the runner. ShutdownTimeout time.Duration `yaml:"shutdown_timeout"` // ShutdownTimeout specifies the duration to wait for running jobs to complete during a shutdown of the runner.
Insecure bool `yaml:"insecure"` // Insecure indicates whether the runner operates in an insecure mode. Insecure bool `yaml:"insecure"` // Insecure indicates whether the runner operates in an insecure mode.
FetchTimeout time.Duration `yaml:"fetch_timeout"` // FetchTimeout specifies the timeout duration for fetching resources. FetchTimeout time.Duration `yaml:"fetch_timeout"` // FetchTimeout specifies the timeout duration for fetching resources.
FetchInterval time.Duration `yaml:"fetch_interval"` // FetchInterval specifies the interval duration for fetching resources. FetchInterval time.Duration `yaml:"fetch_interval"` // FetchInterval specifies the interval duration for fetching resources.
Labels []string `yaml:"labels"` // Labels specify the labels of the runner. Labels are declared on each startup FetchIntervalMax time.Duration `yaml:"fetch_interval_max"` // FetchIntervalMax specifies the maximum backoff interval when idle.
GithubMirror string `yaml:"github_mirror"` // GithubMirror defines what mirrors should be used when using github LogReportInterval time.Duration `yaml:"log_report_interval"` // LogReportInterval specifies the base interval for periodic log flush.
LogReportMaxLatency time.Duration `yaml:"log_report_max_latency"` // LogReportMaxLatency specifies the max time a log row can wait before being sent.
LogReportBatchSize int `yaml:"log_report_batch_size"` // LogReportBatchSize triggers immediate log flush when buffer reaches this size.
StateReportInterval time.Duration `yaml:"state_report_interval"` // StateReportInterval specifies the interval for state reporting.
Labels []string `yaml:"labels"` // Labels specify the labels of the runner. Labels are declared on each startup
GithubMirror string `yaml:"github_mirror"` // GithubMirror defines what mirrors should be used when using github
} }
// Cache represents the configuration for caching. // Cache represents the configuration for caching.
@@ -65,6 +70,12 @@ type Host struct {
WorkdirParent string `yaml:"workdir_parent"` // WorkdirParent specifies the parent directory for the host's working directory. WorkdirParent string `yaml:"workdir_parent"` // WorkdirParent specifies the parent directory for the host's working directory.
} }
// 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").
}
// Config represents the overall configuration. // Config represents the overall configuration.
type Config struct { type Config struct {
Log Log `yaml:"log"` // Log represents the configuration for logging. Log Log `yaml:"log"` // Log represents the configuration for logging.
@@ -72,6 +83,7 @@ type Config struct {
Cache Cache `yaml:"cache"` // Cache represents the configuration for caching. Cache Cache `yaml:"cache"` // Cache represents the configuration for caching.
Container Container `yaml:"container"` // Container represents the configuration for the container. Container Container `yaml:"container"` // Container represents the configuration for the container.
Host Host `yaml:"host"` // Host represents the configuration for the host. Host Host `yaml:"host"` // Host represents the configuration for the host.
Metrics Metrics `yaml:"metrics"` // Metrics represents the configuration for the Prometheus metrics endpoint.
} }
// LoadDefault returns the default configuration. // LoadDefault returns the default configuration.
@@ -137,6 +149,35 @@ func LoadDefault(file string) (*Config, error) {
if cfg.Runner.FetchInterval <= 0 { if cfg.Runner.FetchInterval <= 0 {
cfg.Runner.FetchInterval = 2 * time.Second cfg.Runner.FetchInterval = 2 * time.Second
} }
if cfg.Runner.FetchIntervalMax <= 0 {
cfg.Runner.FetchIntervalMax = 60 * time.Second
}
if cfg.Runner.LogReportInterval <= 0 {
cfg.Runner.LogReportInterval = 5 * time.Second
}
if cfg.Runner.LogReportMaxLatency <= 0 {
cfg.Runner.LogReportMaxLatency = 3 * time.Second
}
if cfg.Runner.LogReportBatchSize <= 0 {
cfg.Runner.LogReportBatchSize = 100
}
if cfg.Runner.StateReportInterval <= 0 {
cfg.Runner.StateReportInterval = 5 * time.Second
}
if cfg.Metrics.Addr == "" {
cfg.Metrics.Addr = "127.0.0.1:9101"
}
// Validate and fix invalid config combinations to prevent confusing behavior.
if cfg.Runner.FetchIntervalMax < cfg.Runner.FetchInterval {
log.Warnf("fetch_interval_max (%v) is less than fetch_interval (%v), setting fetch_interval_max to fetch_interval",
cfg.Runner.FetchIntervalMax, cfg.Runner.FetchInterval)
cfg.Runner.FetchIntervalMax = cfg.Runner.FetchInterval
}
if cfg.Runner.LogReportMaxLatency >= cfg.Runner.LogReportInterval {
log.Warnf("log_report_max_latency (%v) >= log_report_interval (%v), the max-latency timer will never fire before the periodic ticker; consider lowering log_report_max_latency",
cfg.Runner.LogReportMaxLatency, cfg.Runner.LogReportInterval)
}
// although `container.network_mode` will be deprecated, but we have to be compatible with it for now. // although `container.network_mode` will be deprecated, but we have to be compatible with it for now.
if cfg.Container.NetworkMode != "" && cfg.Container.Network == "" { if cfg.Container.NetworkMode != "" && cfg.Container.Network == "" {

216
internal/pkg/metrics/metrics.go Normal file
View File

@@ -0,0 +1,216 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package metrics
import (
"sync"
"time"
runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/collectors"
)
// Namespace is the Prometheus namespace for all act_runner metrics.
const Namespace = "act_runner"
// Label value constants for Prometheus metrics.
// Using constants prevents typos from silently creating new time-series.
//
// LabelResult* values are used on metrics with label key "result" (RPC outcomes).
// LabelStatus* values are used on metrics with label key "status" (job outcomes).
const (
LabelResultTask = "task"
LabelResultEmpty = "empty"
LabelResultError = "error"
LabelResultSuccess = "success"
LabelMethodFetchTask = "FetchTask"
LabelMethodUpdateLog = "UpdateLog"
LabelMethodUpdateTask = "UpdateTask"
LabelStatusSuccess = "success"
LabelStatusFailure = "failure"
LabelStatusCancelled = "cancelled"
LabelStatusSkipped = "skipped"
LabelStatusUnknown = "unknown"
)
// rpcDurationBuckets covers the expected latency range for short-running
// UpdateLog / UpdateTask RPCs. FetchTask uses its own buckets (it has a 10s tail).
var rpcDurationBuckets = []float64{0.01, 0.05, 0.1, 0.25, 0.5, 1, 2, 5}
// ResultToStatusLabel maps a runnerv1.Result to the "status" label value used on job metrics.
func ResultToStatusLabel(r runnerv1.Result) string {
switch r {
case runnerv1.Result_RESULT_SUCCESS:
return LabelStatusSuccess
case runnerv1.Result_RESULT_FAILURE:
return LabelStatusFailure
case runnerv1.Result_RESULT_CANCELLED:
return LabelStatusCancelled
case runnerv1.Result_RESULT_SKIPPED:
return LabelStatusSkipped
default:
return LabelStatusUnknown
}
}
var (
RunnerInfo = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Namespace: Namespace,
Name: "info",
Help: "Runner metadata. Always 1. Labels carry version and name.",
}, []string{"version", "name"})
RunnerCapacity = prometheus.NewGauge(prometheus.GaugeOpts{
Namespace: Namespace,
Name: "capacity",
Help: "Configured maximum concurrent jobs.",
})
PollFetchTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: Namespace,
Subsystem: "poll",
Name: "fetch_total",
Help: "Total number of FetchTask RPCs by result (task, empty, error).",
}, []string{"result"})
PollFetchDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
Namespace: Namespace,
Subsystem: "poll",
Name: "fetch_duration_seconds",
Help: "Latency of FetchTask RPCs, excluding expected long-poll timeouts.",
Buckets: []float64{0.01, 0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10},
})
PollBackoffSeconds = prometheus.NewGauge(prometheus.GaugeOpts{
Namespace: Namespace,
Subsystem: "poll",
Name: "backoff_seconds",
Help: "Last observed polling backoff interval. With Capacity > 1, reflects whichever worker wrote last.",
})
JobsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: Namespace,
Subsystem: "job",
Name: "total",
Help: "Total jobs processed by status (success, failure, cancelled, skipped, unknown).",
}, []string{"status"})
JobDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
Namespace: Namespace,
Subsystem: "job",
Name: "duration_seconds",
Help: "Duration of job execution from start to finish.",
Buckets: prometheus.ExponentialBuckets(1, 2, 14), // 1s to ~4.5h
})
ReportLogTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: Namespace,
Subsystem: "report",
Name: "log_total",
Help: "Total UpdateLog RPCs by result (success, error).",
}, []string{"result"})
ReportLogDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
Namespace: Namespace,
Subsystem: "report",
Name: "log_duration_seconds",
Help: "Latency of UpdateLog RPCs.",
Buckets: rpcDurationBuckets,
})
ReportStateTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: Namespace,
Subsystem: "report",
Name: "state_total",
Help: "Total UpdateTask (state) RPCs by result (success, error).",
}, []string{"result"})
ReportStateDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
Namespace: Namespace,
Subsystem: "report",
Name: "state_duration_seconds",
Help: "Latency of UpdateTask RPCs.",
Buckets: rpcDurationBuckets,
})
ReportLogBufferRows = prometheus.NewGauge(prometheus.GaugeOpts{
Namespace: Namespace,
Subsystem: "report",
Name: "log_buffer_rows",
Help: "Current number of buffered log rows awaiting send.",
})
ClientErrors = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: Namespace,
Subsystem: "client",
Name: "errors_total",
Help: "Total client RPC errors by method.",
}, []string{"method"})
)
// Registry is the custom Prometheus registry used by the runner.
var Registry = prometheus.NewRegistry()
var initOnce sync.Once
// Init registers all static metrics and the standard Go/process collectors.
// Safe to call multiple times; only the first call has effect.
func Init() {
initOnce.Do(func() {
Registry.MustRegister(
collectors.NewGoCollector(),
collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}),
RunnerInfo, RunnerCapacity,
PollFetchTotal, PollFetchDuration, PollBackoffSeconds,
JobsTotal, JobDuration,
ReportLogTotal, ReportLogDuration,
ReportStateTotal, ReportStateDuration, ReportLogBufferRows,
ClientErrors,
)
})
}
// RegisterUptimeFunc registers a GaugeFunc that reports seconds since startTime.
func RegisterUptimeFunc(startTime time.Time) {
Registry.MustRegister(prometheus.NewGaugeFunc(
prometheus.GaugeOpts{
Namespace: Namespace,
Name: "uptime_seconds",
Help: "Seconds since the runner daemon started.",
},
func() float64 { return time.Since(startTime).Seconds() },
))
}
// RegisterRunningJobsFunc registers GaugeFuncs for the running job count and
// capacity utilisation ratio, evaluated lazily at Prometheus scrape time.
func RegisterRunningJobsFunc(countFn func() int64, capacity int) {
capF := float64(capacity)
Registry.MustRegister(prometheus.NewGaugeFunc(
prometheus.GaugeOpts{
Namespace: Namespace,
Subsystem: "job",
Name: "running",
Help: "Number of jobs currently executing.",
},
func() float64 { return float64(countFn()) },
))
Registry.MustRegister(prometheus.NewGaugeFunc(
prometheus.GaugeOpts{
Namespace: Namespace,
Subsystem: "job",
Name: "capacity_utilization_ratio",
Help: "Ratio of running jobs to configured capacity (0.0-1.0).",
},
func() float64 {
if capF <= 0 {
return 0
}
return float64(countFn()) / capF
},
))
}

50
internal/pkg/metrics/server.go Normal file
View File

@@ -0,0 +1,50 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package metrics
import (
"context"
"net/http"
"time"
"github.com/prometheus/client_golang/prometheus/promhttp"
log "github.com/sirupsen/logrus"
)
// 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.
// 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"))
})
srv := &http.Server{
Addr: addr,
Handler: mux,
ReadHeaderTimeout: 5 * time.Second,
ReadTimeout: 10 * time.Second,
WriteTimeout: 10 * time.Second,
IdleTimeout: 60 * time.Second,
}
go func() {
log.Infof("metrics server listening on %s", addr)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.WithError(err).Error("metrics server failed")
}
}()
go func() {
<-ctx.Done()
shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := srv.Shutdown(shutCtx); err != nil {
log.WithError(err).Warn("metrics server shutdown error")
}
}()
}

230
internal/pkg/report/reporter.go
View File

@@ -20,6 +20,8 @@ import (
"google.golang.org/protobuf/types/known/timestamppb" "google.golang.org/protobuf/types/known/timestamppb"
"gitea.com/gitea/act_runner/internal/pkg/client" "gitea.com/gitea/act_runner/internal/pkg/client"
"gitea.com/gitea/act_runner/internal/pkg/config"
"gitea.com/gitea/act_runner/internal/pkg/metrics"
) )
type Reporter struct { type Reporter struct {
@@ -35,16 +37,32 @@ type Reporter struct {
logReplacer *strings.Replacer logReplacer *strings.Replacer
oldnew []string oldnew []string
state *runnerv1.TaskState // lastLogBufferRows is the last value written to the ReportLogBufferRows
stateMu sync.RWMutex // gauge; guarded by clientM (the same lock held around each ReportLog call)
outputs sync.Map // so the gauge skips no-op Set calls when the buffer size is unchanged.
daemon chan struct{} lastLogBufferRows int
state *runnerv1.TaskState
stateChanged bool
stateMu sync.RWMutex
outputs sync.Map
daemon chan struct{}
// Adaptive batching control
logReportInterval time.Duration
logReportMaxLatency time.Duration
logBatchSize int
stateReportInterval time.Duration
// Event notification channels (non-blocking, buffered 1)
logNotify chan struct{} // signal: new log rows arrived
stateNotify chan struct{} // signal: step transition (start/stop)
debugOutputEnabled bool debugOutputEnabled bool
stopCommandEndToken string stopCommandEndToken string
} }
func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.Client, task *runnerv1.Task) *Reporter { func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.Client, task *runnerv1.Task, cfg *config.Config) *Reporter {
var oldnew []string var oldnew []string
if v := task.Context.Fields["token"].GetStringValue(); v != "" { if v := task.Context.Fields["token"].GetStringValue(); v != "" {
oldnew = append(oldnew, v, "***") oldnew = append(oldnew, v, "***")
@@ -57,11 +75,17 @@ func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.C
} }
rv := &Reporter{ rv := &Reporter{
ctx: ctx, ctx: ctx,
cancel: cancel, cancel: cancel,
client: client, client: client,
oldnew: oldnew, oldnew: oldnew,
logReplacer: strings.NewReplacer(oldnew...), logReplacer: strings.NewReplacer(oldnew...),
logReportInterval: cfg.Runner.LogReportInterval,
logReportMaxLatency: cfg.Runner.LogReportMaxLatency,
logBatchSize: cfg.Runner.LogReportBatchSize,
stateReportInterval: cfg.Runner.StateReportInterval,
logNotify: make(chan struct{}, 1),
stateNotify: make(chan struct{}, 1),
state: &runnerv1.TaskState{ state: &runnerv1.TaskState{
Id: task.Id, Id: task.Id,
}, },
@@ -75,6 +99,13 @@ func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.C
return rv return rv
} }
// Result returns the final job result. Safe to call after Close() returns.
func (r *Reporter) Result() runnerv1.Result {
r.stateMu.RLock()
defer r.stateMu.RUnlock()
return r.state.Result
}
func (r *Reporter) ResetSteps(l int) { func (r *Reporter) ResetSteps(l int) {
r.stateMu.Lock() r.stateMu.Lock()
defer r.stateMu.Unlock() defer r.stateMu.Unlock()
@@ -108,11 +139,42 @@ func isJobStepEntry(entry *log.Entry) bool {
return true return true
} }
func (r *Reporter) Fire(entry *log.Entry) error { // notifyLog sends a non-blocking signal that new log rows are available.
r.stateMu.Lock() func (r *Reporter) notifyLog() {
defer r.stateMu.Unlock() select {
case r.logNotify <- struct{}{}:
default:
}
}
log.WithFields(entry.Data).Trace(entry.Message) // notifyState sends a non-blocking signal that a UX-critical state change occurred (step start/stop, job result).
func (r *Reporter) notifyState() {
select {
case r.stateNotify <- struct{}{}:
default:
}
}
// unlockAndNotify releases stateMu and sends channel notifications.
// Must be called with stateMu held.
func (r *Reporter) unlockAndNotify(urgentState bool) {
r.stateMu.Unlock()
r.notifyLog()
if urgentState {
r.notifyState()
}
}
func (r *Reporter) Fire(entry *log.Entry) error {
urgentState := false
r.stateMu.Lock()
r.stateChanged = true
if log.IsLevelEnabled(log.TraceLevel) {
log.WithFields(entry.Data).Trace(entry.Message)
}
timestamp := entry.Time timestamp := entry.Time
if r.state.StartedAt == nil { if r.state.StartedAt == nil {
@@ -135,11 +197,13 @@ func (r *Reporter) Fire(entry *log.Entry) error {
} }
} }
} }
urgentState = true
} }
} }
if !r.duringSteps() { if !r.duringSteps() {
r.logRows = appendIfNotNil(r.logRows, r.parseLogRow(entry)) r.logRows = appendIfNotNil(r.logRows, r.parseLogRow(entry))
} }
r.unlockAndNotify(urgentState)
return nil return nil
} }
@@ -153,11 +217,13 @@ func (r *Reporter) Fire(entry *log.Entry) error {
if !r.duringSteps() { if !r.duringSteps() {
r.logRows = appendIfNotNil(r.logRows, r.parseLogRow(entry)) r.logRows = appendIfNotNil(r.logRows, r.parseLogRow(entry))
} }
r.unlockAndNotify(false)
return nil return nil
} }
if step.StartedAt == nil { if step.StartedAt == nil {
step.StartedAt = timestamppb.New(timestamp) step.StartedAt = timestamppb.New(timestamp)
urgentState = true
} }
// Force reporting log errors as raw output to prevent silent failures // Force reporting log errors as raw output to prevent silent failures
@@ -185,26 +251,91 @@ func (r *Reporter) Fire(entry *log.Entry) error {
} }
step.Result = stepResult step.Result = stepResult
step.StoppedAt = timestamppb.New(timestamp) step.StoppedAt = timestamppb.New(timestamp)
urgentState = true
} }
} }
r.unlockAndNotify(urgentState)
return nil return nil
} }
func (r *Reporter) RunDaemon() { func (r *Reporter) RunDaemon() {
r.stateMu.RLock() go r.runDaemonLoop()
closed := r.closed }
r.stateMu.RUnlock()
if closed || r.ctx.Err() != nil { func (r *Reporter) stopLatencyTimer(active *bool, timer *time.Timer) {
// Acknowledge close if *active {
close(r.daemon) if !timer.Stop() {
return select {
case <-timer.C:
default:
}
}
*active = false
} }
}
_ = r.ReportLog(false) func (r *Reporter) runDaemonLoop() {
_ = r.ReportState(false) logTicker := time.NewTicker(r.logReportInterval)
stateTicker := time.NewTicker(r.stateReportInterval)
time.AfterFunc(time.Second, r.RunDaemon) // maxLatencyTimer ensures the first buffered log row is sent within logReportMaxLatency.
// Start inactive — it is armed when the first log row arrives in an empty buffer.
maxLatencyTimer := time.NewTimer(0)
if !maxLatencyTimer.Stop() {
<-maxLatencyTimer.C
}
maxLatencyActive := false
defer logTicker.Stop()
defer stateTicker.Stop()
defer maxLatencyTimer.Stop()
for {
select {
case <-logTicker.C:
_ = r.ReportLog(false)
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
case <-stateTicker.C:
_ = r.ReportState(false)
case <-r.logNotify:
r.stateMu.RLock()
n := len(r.logRows)
r.stateMu.RUnlock()
if n >= r.logBatchSize {
_ = r.ReportLog(false)
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
} else if !maxLatencyActive && n > 0 {
maxLatencyTimer.Reset(r.logReportMaxLatency)
maxLatencyActive = true
}
case <-r.stateNotify:
// Step transition or job result — flush both immediately for frontend UX.
_ = r.ReportLog(false)
_ = r.ReportState(false)
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
case <-maxLatencyTimer.C:
maxLatencyActive = false
_ = r.ReportLog(false)
case <-r.ctx.Done():
close(r.daemon)
return
}
r.stateMu.RLock()
closed := r.closed
r.stateMu.RUnlock()
if closed {
close(r.daemon)
return
}
}
} }
func (r *Reporter) Logf(format string, a ...any) { func (r *Reporter) Logf(format string, a ...any) {
@@ -268,6 +399,10 @@ func (r *Reporter) Close(lastWords string) error {
}) })
} }
r.stateMu.Unlock() r.stateMu.Unlock()
// Wake up the daemon loop so it detects closed promptly.
r.notifyLog()
// Wait for Acknowledge // Wait for Acknowledge
select { select {
case <-r.daemon: case <-r.daemon:
@@ -295,15 +430,24 @@ func (r *Reporter) ReportLog(noMore bool) error {
rows := r.logRows rows := r.logRows
r.stateMu.RUnlock() r.stateMu.RUnlock()
if !noMore && len(rows) == 0 {
return nil
}
start := time.Now()
resp, err := r.client.UpdateLog(r.ctx, connect.NewRequest(&runnerv1.UpdateLogRequest{ resp, err := r.client.UpdateLog(r.ctx, connect.NewRequest(&runnerv1.UpdateLogRequest{
TaskId: r.state.Id, TaskId: r.state.Id,
Index: int64(r.logOffset), Index: int64(r.logOffset),
Rows: rows, Rows: rows,
NoMore: noMore, NoMore: noMore,
})) }))
metrics.ReportLogDuration.Observe(time.Since(start).Seconds())
if err != nil { if err != nil {
metrics.ReportLogTotal.WithLabelValues(metrics.LabelResultError).Inc()
metrics.ClientErrors.WithLabelValues(metrics.LabelMethodUpdateLog).Inc()
return err return err
} }
metrics.ReportLogTotal.WithLabelValues(metrics.LabelResultSuccess).Inc()
ack := int(resp.Msg.AckIndex) ack := int(resp.Msg.AckIndex)
if ack < r.logOffset { if ack < r.logOffset {
@@ -314,7 +458,12 @@ func (r *Reporter) ReportLog(noMore bool) error {
r.logRows = r.logRows[ack-r.logOffset:] r.logRows = r.logRows[ack-r.logOffset:]
submitted := r.logOffset + len(rows) submitted := r.logOffset + len(rows)
r.logOffset = ack r.logOffset = ack
remaining := len(r.logRows)
r.stateMu.Unlock() r.stateMu.Unlock()
if remaining != r.lastLogBufferRows {
metrics.ReportLogBufferRows.Set(float64(remaining))
r.lastLogBufferRows = remaining
}
if noMore && ack < submitted { if noMore && ack < submitted {
return errors.New("not all logs are submitted") return errors.New("not all logs are submitted")
@@ -329,15 +478,7 @@ func (r *Reporter) ReportState(reportResult bool) error {
r.clientM.Lock() r.clientM.Lock()
defer r.clientM.Unlock() defer r.clientM.Unlock()
r.stateMu.RLock() // Build the outputs map first (single Range pass instead of two).
state := proto.Clone(r.state).(*runnerv1.TaskState)
r.stateMu.RUnlock()
// Only report result from Close to reliable sent logs
if !reportResult {
state.Result = runnerv1.Result_RESULT_UNSPECIFIED
}
outputs := make(map[string]string) outputs := make(map[string]string)
r.outputs.Range(func(k, v any) bool { r.outputs.Range(func(k, v any) bool {
if val, ok := v.(string); ok { if val, ok := v.(string); ok {
@@ -346,13 +487,36 @@ func (r *Reporter) ReportState(reportResult bool) error {
return true return true
}) })
// Consume stateChanged atomically with the snapshot; restored on error
// below so a concurrent Fire() during UpdateTask isn't silently lost.
r.stateMu.Lock()
if !reportResult && !r.stateChanged && len(outputs) == 0 {
r.stateMu.Unlock()
return nil
}
state := proto.Clone(r.state).(*runnerv1.TaskState)
r.stateChanged = false
r.stateMu.Unlock()
if !reportResult {
state.Result = runnerv1.Result_RESULT_UNSPECIFIED
}
start := time.Now()
resp, err := r.client.UpdateTask(r.ctx, connect.NewRequest(&runnerv1.UpdateTaskRequest{ resp, err := r.client.UpdateTask(r.ctx, connect.NewRequest(&runnerv1.UpdateTaskRequest{
State: state, State: state,
Outputs: outputs, Outputs: outputs,
})) }))
metrics.ReportStateDuration.Observe(time.Since(start).Seconds())
if err != nil { if err != nil {
metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultError).Inc()
metrics.ClientErrors.WithLabelValues(metrics.LabelMethodUpdateTask).Inc()
r.stateMu.Lock()
r.stateChanged = true
r.stateMu.Unlock()
return err return err
} }
metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultSuccess).Inc()
for _, k := range resp.Msg.SentOutputs { for _, k := range resp.Msg.SentOutputs {
r.outputs.Store(k, struct{}{}) r.outputs.Store(k, struct{}{})

290
internal/pkg/report/reporter_test.go
View File

@@ -6,8 +6,9 @@ package report
import ( import (
"context" "context"
"errors" "errors"
"fmt"
"strings" "strings"
"sync" "sync/atomic"
"testing" "testing"
"time" "time"
@@ -21,6 +22,7 @@ import (
"google.golang.org/protobuf/types/known/timestamppb" "google.golang.org/protobuf/types/known/timestamppb"
"gitea.com/gitea/act_runner/internal/pkg/client/mocks" "gitea.com/gitea/act_runner/internal/pkg/client/mocks"
"gitea.com/gitea/act_runner/internal/pkg/config"
) )
func TestReporter_parseLogRow(t *testing.T) { func TestReporter_parseLogRow(t *testing.T) {
@@ -175,9 +177,10 @@ func TestReporter_Fire(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
taskCtx, err := structpb.NewStruct(map[string]any{}) taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err) require.NoError(t, err)
cfg, _ := config.LoadDefault("")
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{ reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{
Context: taskCtx, Context: taskCtx,
}) }, cfg)
reporter.RunDaemon() reporter.RunDaemon()
defer func() { defer func() {
require.NoError(t, reporter.Close("")) require.NoError(t, reporter.Close(""))
@@ -252,7 +255,8 @@ func TestReporter_EphemeralRunnerDeletion(t *testing.T) {
defer cancel() defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{}) taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err) require.NoError(t, err)
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}) cfg, _ := config.LoadDefault("")
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(1) reporter.ResetSteps(1)
// Fire a log entry to create pending data // Fire a log entry to create pending data
@@ -315,23 +319,281 @@ func TestReporter_RunDaemonClose_Race(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
taskCtx, err := structpb.NewStruct(map[string]any{}) taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err) require.NoError(t, err)
cfg, _ := config.LoadDefault("")
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{ reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{
Context: taskCtx, Context: taskCtx,
}) }, cfg)
reporter.ResetSteps(1) reporter.ResetSteps(1)
// Start the daemon loop in a separate goroutine. // Start the daemon loop — RunDaemon spawns a goroutine internally.
// RunDaemon reads r.closed and reschedules itself via time.AfterFunc. reporter.RunDaemon()
var wg sync.WaitGroup
wg.Go(func() {
reporter.RunDaemon()
})
// Close concurrently — this races with RunDaemon on r.closed. // Close concurrently — this races with the daemon goroutine on r.closed.
require.NoError(t, reporter.Close("")) require.NoError(t, reporter.Close(""))
// Cancel context so pending AfterFunc callbacks exit quickly. // Cancel context so the daemon goroutine exits cleanly.
cancel() cancel()
wg.Wait() }
time.Sleep(2 * time.Second)
// TestReporter_MaxLatencyTimer verifies that the maxLatencyTimer flushes a
// single buffered log row before the periodic logTicker fires.
//
// Setup: logReportInterval=10s (effectively never), maxLatency=100ms.
// Fire one log line, then assert UpdateLog is called within 500ms.
func TestReporter_MaxLatencyTimer(t *testing.T) {
var updateLogCalls atomic.Int64
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Return(
func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
updateLogCalls.Add(1)
return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)),
}), nil
},
)
client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(
func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
},
)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
// Custom config: logTicker=10s (won't fire during test), maxLatency=100ms
cfg, _ := config.LoadDefault("")
cfg.Runner.LogReportInterval = 10 * time.Second
cfg.Runner.LogReportMaxLatency = 100 * time.Millisecond
cfg.Runner.LogReportBatchSize = 1000 // won't trigger batch flush
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(1)
reporter.RunDaemon()
defer func() {
_ = reporter.Close("")
}()
// Fire a single log line — not enough to trigger batch flush
require.NoError(t, reporter.Fire(&log.Entry{
Message: "single log line",
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
// maxLatencyTimer should flush within ~100ms. Wait up to 500ms.
assert.Eventually(t, func() bool {
return updateLogCalls.Load() > 0
}, 500*time.Millisecond, 10*time.Millisecond,
"maxLatencyTimer should have flushed the log before logTicker (10s)")
}
// TestReporter_BatchSizeFlush verifies that reaching logBatchSize triggers
// an immediate log flush without waiting for any timer.
func TestReporter_BatchSizeFlush(t *testing.T) {
var updateLogCalls atomic.Int64
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Return(
func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
updateLogCalls.Add(1)
return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)),
}), nil
},
)
client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(
func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
},
)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
// Custom config: large timers, small batch size
cfg, _ := config.LoadDefault("")
cfg.Runner.LogReportInterval = 10 * time.Second
cfg.Runner.LogReportMaxLatency = 10 * time.Second
cfg.Runner.LogReportBatchSize = 5
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(1)
reporter.RunDaemon()
defer func() {
_ = reporter.Close("")
}()
// Fire exactly batchSize log lines
for i := range 5 {
require.NoError(t, reporter.Fire(&log.Entry{
Message: fmt.Sprintf("log line %d", i),
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
}
// Batch threshold should trigger immediate flush
assert.Eventually(t, func() bool {
return updateLogCalls.Load() > 0
}, 500*time.Millisecond, 10*time.Millisecond,
"batch size threshold should have triggered immediate flush")
}
// TestReporter_StateChangedNotLostDuringReport asserts that a Fire() arriving
// mid-UpdateTask re-dirties the flag so the change is picked up by the next report.
func TestReporter_StateChangedNotLostDuringReport(t *testing.T) {
var updateTaskCalls atomic.Int64
inFlight := make(chan struct{})
release := make(chan struct{})
client := mocks.NewClient(t)
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
n := updateTaskCalls.Add(1)
if n == 1 {
// Signal that the first UpdateTask is in flight, then block until released.
close(inFlight)
<-release
}
return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
},
)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
cfg, _ := config.LoadDefault("")
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(2)
// Mark stateChanged=true so the first ReportState proceeds to UpdateTask.
reporter.stateMu.Lock()
reporter.stateChanged = true
reporter.stateMu.Unlock()
// Kick off the first ReportState in a goroutine — it will block in UpdateTask.
done := make(chan error, 1)
go func() {
done <- reporter.ReportState(false)
}()
// Wait until UpdateTask is in flight (snapshot taken, flag consumed).
<-inFlight
// Concurrent Fire() modifies state — must re-flip stateChanged so the
// change is not lost when the in-flight ReportState finishes.
require.NoError(t, reporter.Fire(&log.Entry{
Message: "step starts",
Data: log.Fields{"stage": "Main", "stepNumber": 1, "raw_output": true},
}))
// Release the in-flight UpdateTask and wait for it to return.
close(release)
require.NoError(t, <-done)
// stateChanged must still be true so the next ReportState picks up the
// concurrent Fire()'s change instead of skipping via the early-return path.
reporter.stateMu.RLock()
changed := reporter.stateChanged
reporter.stateMu.RUnlock()
assert.True(t, changed, "stateChanged must remain true after a concurrent Fire() during in-flight ReportState")
// And the next ReportState must actually send a second UpdateTask.
require.NoError(t, reporter.ReportState(false))
assert.Equal(t, int64(2), updateTaskCalls.Load(), "concurrent Fire() change must trigger a second UpdateTask, not be silently lost")
}
// TestReporter_StateChangedRestoredOnError verifies that when UpdateTask fails,
// the dirty flag is restored so the snapshotted change isn't silently lost.
func TestReporter_StateChangedRestoredOnError(t *testing.T) {
var updateTaskCalls atomic.Int64
client := mocks.NewClient(t)
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
n := updateTaskCalls.Add(1)
if n == 1 {
return nil, errors.New("transient network error")
}
return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
},
)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
cfg, _ := config.LoadDefault("")
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(1)
reporter.stateMu.Lock()
reporter.stateChanged = true
reporter.stateMu.Unlock()
// First ReportState fails — flag must be restored to true.
require.Error(t, reporter.ReportState(false))
reporter.stateMu.RLock()
changed := reporter.stateChanged
reporter.stateMu.RUnlock()
assert.True(t, changed, "stateChanged must be restored to true after UpdateTask error so the change is retried")
// The next ReportState should still issue a request because the flag was restored.
require.NoError(t, reporter.ReportState(false))
assert.Equal(t, int64(2), updateTaskCalls.Load())
}
// TestReporter_StateNotifyFlush verifies that step transitions trigger
// an immediate state flush via the stateNotify channel.
func TestReporter_StateNotifyFlush(t *testing.T) {
var updateTaskCalls atomic.Int64
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return(
func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)),
}), nil
},
)
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
updateTaskCalls.Add(1)
return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
},
)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
// Custom config: large state interval so only stateNotify can trigger
cfg, _ := config.LoadDefault("")
cfg.Runner.StateReportInterval = 10 * time.Second
cfg.Runner.LogReportInterval = 10 * time.Second
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(1)
reporter.RunDaemon()
defer func() {
_ = reporter.Close("")
}()
// Fire a log entry that starts a step — this triggers stateNotify
require.NoError(t, reporter.Fire(&log.Entry{
Message: "step starting",
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
// stateNotify should trigger immediate UpdateTask call
assert.Eventually(t, func() bool {
return updateTaskCalls.Load() > 0
}, 500*time.Millisecond, 10*time.Millisecond,
"step transition should have triggered immediate state flush via stateNotify")
} }