mirror of
https://gitea.com/gitea/act_runner.git
synced 2026-08-06 00:44:22 +02:00
Compare commits
2 Commits
0192861155
...
41c72216bf
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
41c72216bf | ||
|
|
3f7fd16ea1 |
@@ -11,6 +11,7 @@ linters:
|
||||
- dupl
|
||||
- errcheck
|
||||
- forbidigo
|
||||
- forcetypeassert
|
||||
- gocheckcompilerdirectives
|
||||
- gocritic
|
||||
- goheader
|
||||
@@ -102,6 +103,9 @@ linters:
|
||||
- linters:
|
||||
- forbidigo
|
||||
path: cmd
|
||||
- linters:
|
||||
- forcetypeassert
|
||||
path: _test\.go
|
||||
issues:
|
||||
max-issues-per-linter: 0
|
||||
max-same-issues: 0
|
||||
|
||||
29
README.md
29
README.md
@@ -209,6 +209,35 @@ Whenever the resulting labels differ from the ones in the registration file, the
|
||||
|
||||
> **Note:** A runner that only exposes `host` labels still needs access to a Docker daemon (e.g. a mounted `/var/run/docker.sock`) whenever a job uses a `docker://` action or a service container. `host` labels only change where the job's own steps run; container-based steps and actions are still executed with Docker.
|
||||
|
||||
#### Proxy
|
||||
|
||||
Set these variables in the runner's environment, with systemd `Environment=`, `docker run -e`, or Kubernetes `env:`:
|
||||
|
||||
```sh
|
||||
http_proxy=http://proxy.example:3128
|
||||
https_proxy=http://proxy.example:3128
|
||||
no_proxy=gitea.internal,.example.local
|
||||
```
|
||||
|
||||
The runner uses them for its own requests and gives them to every job, in lower and upper case.
|
||||
|
||||
These hosts are added to `no_proxy` for jobs, so they are always reached directly:
|
||||
|
||||
- the cache server
|
||||
- `localhost`, `127.0.0.1` and `::1`
|
||||
- the job's service containers
|
||||
- the Docker daemon, when it is reached over `tcp://`
|
||||
|
||||
Gitea is not added. Add it to `no_proxy` yourself if it should be reached directly.
|
||||
|
||||
To change a value for one job, set it in a step's `env:` or in the job's `container.env`. Setting it at workflow or job level has no effect. To change it for the whole runner, set it in `runner.envs`. A `no_proxy` set there is added to the list above instead of replacing it.
|
||||
|
||||
Images are pulled by the Docker daemon, which needs its own proxy setting. In the `dind` images the daemon runs in the same container and reads the variables above. For any other daemon, see [the Docker documentation](https://docs.docker.com/engine/daemon/proxy/). The runner logs a warning at startup if it has a proxy and the daemon does not.
|
||||
|
||||
Dockerfile actions are built with these variables as build arguments, so their `RUN` steps can reach the network.
|
||||
|
||||
A password in a proxy URL is hidden in job logs. Any step can still read it, because the step is given the proxy URL in its environment.
|
||||
|
||||
#### Caching (`actions/cache`)
|
||||
|
||||
Each runner starts its own cache server automatically. Cache entries are local to that runner — runners do not share a cache by default.
|
||||
|
||||
@@ -70,6 +70,7 @@ type Handler struct {
|
||||
storage *Storage
|
||||
router *httprouter.Router
|
||||
listener net.Listener
|
||||
port int
|
||||
server *http.Server
|
||||
logger logrus.FieldLogger
|
||||
|
||||
@@ -177,6 +178,12 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
addr, ok := listener.Addr().(*net.TCPAddr)
|
||||
if !ok {
|
||||
listener.Close()
|
||||
return nil, fmt.Errorf("cache server listens on %T, want a TCP address", listener.Addr())
|
||||
}
|
||||
h.port = addr.Port
|
||||
server := &http.Server{
|
||||
ReadHeaderTimeout: 2 * time.Second,
|
||||
Handler: router,
|
||||
@@ -194,9 +201,7 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
|
||||
|
||||
func (h *Handler) ExternalURL() string {
|
||||
// TODO: make the external url configurable if necessary
|
||||
return fmt.Sprintf("http://%s:%d",
|
||||
h.outboundIP,
|
||||
h.listener.Addr().(*net.TCPAddr).Port)
|
||||
return fmt.Sprintf("http://%s:%d", h.outboundIP, h.port)
|
||||
}
|
||||
|
||||
// RegisterJob makes token a valid bearer credential for cache requests from
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"sync"
|
||||
|
||||
"gitea.com/gitea/runner/act/common"
|
||||
"gitea.com/gitea/runner/internal/pkg/lock"
|
||||
|
||||
"github.com/go-git/go-git/v5"
|
||||
"github.com/go-git/go-git/v5/config"
|
||||
@@ -32,7 +33,7 @@ var (
|
||||
githubHTTPRegex = regexp.MustCompile(`^https?://.*github.com.*/(.+)/(.+?)(?:.git)?$`)
|
||||
githubSSHRegex = regexp.MustCompile(`github.com[:/](.+)/(.+?)(?:.git)?$`)
|
||||
|
||||
cloneLocks sync.Map // key: clone target directory; value: *sync.Mutex
|
||||
cloneLocks lock.Keyed[string] // key: clone target directory
|
||||
|
||||
ErrShortRef = errors.New("short SHA references are not supported")
|
||||
ErrNoRepo = errors.New("unable to find git repo")
|
||||
@@ -43,10 +44,7 @@ var (
|
||||
// Callers reading files inside dir (e.g. tarring a checked-out action into a job container) must hold this lock too,
|
||||
// otherwise a concurrent NewGitCloneExecutor on the same dir can mutate the worktree mid-read.
|
||||
func AcquireCloneLock(dir string) func() {
|
||||
v, _ := cloneLocks.LoadOrStore(dir, &sync.Mutex{})
|
||||
mu := v.(*sync.Mutex)
|
||||
mu.Lock()
|
||||
return mu.Unlock
|
||||
return cloneLocks.Lock(dir)
|
||||
}
|
||||
|
||||
type Error struct {
|
||||
|
||||
@@ -12,7 +12,6 @@ import (
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -610,12 +609,4 @@ func TestAcquireCloneLock(t *testing.T) {
|
||||
t.Fatal("acquire on a different directory must not block")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("same directory reuses the same mutex", func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
|
||||
v1, _ := cloneLocks.LoadOrStore(dir, &sync.Mutex{})
|
||||
v2, _ := cloneLocks.LoadOrStore(dir, &sync.Mutex{})
|
||||
require.Same(t, v1, v2)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -19,7 +19,9 @@ func GetOutboundIP() net.IP {
|
||||
conn, err := net.Dial("udp", "8.8.8.8:80")
|
||||
if err == nil {
|
||||
defer conn.Close()
|
||||
return conn.LocalAddr().(*net.UDPAddr).IP
|
||||
if addr, ok := conn.LocalAddr().(*net.UDPAddr); ok {
|
||||
return addr.IP
|
||||
}
|
||||
}
|
||||
|
||||
// So the machine cannot access the internet. Pick an IP address from network interfaces.
|
||||
|
||||
@@ -82,6 +82,7 @@ type NewDockerBuildExecutorInput struct {
|
||||
BuildContext io.Reader
|
||||
ImageTag string
|
||||
Platform string
|
||||
BuildArgs map[string]*string
|
||||
}
|
||||
|
||||
// NewDockerNetworkCreateExecutorInput the input for the NewDockerNetworkCreateExecutor function
|
||||
|
||||
@@ -49,6 +49,7 @@ func NewDockerBuildExecutor(input NewDockerBuildExecutorInput) common.Executor {
|
||||
Remove: true,
|
||||
AuthConfigs: LoadDockerAuthConfigs(ctx),
|
||||
Dockerfile: input.Dockerfile,
|
||||
BuildArgs: input.BuildArgs,
|
||||
}
|
||||
platform, err := parsePlatform(input.Platform)
|
||||
if err != nil {
|
||||
|
||||
@@ -51,7 +51,8 @@ func TestCreateFlagsValidate(t *testing.T) {
|
||||
|
||||
func TestNewContainerAppliesCreateFlags(t *testing.T) {
|
||||
input := &NewContainerInput{Platform: "linux/amd64", Options: "--platform linux/arm64 --pull never"}
|
||||
cr := NewContainer(input).(*containerReference)
|
||||
cr, ok := NewContainer(input).(*containerReference)
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, "linux/arm64", input.Platform)
|
||||
assert.Equal(t, pullPolicyNever, cr.pullPolicy)
|
||||
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"gitea.com/gitea/runner/act/model"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestLiterals(t *testing.T) {
|
||||
@@ -523,7 +524,9 @@ func TestOperatorsBooleanEvaluation(t *testing.T) {
|
||||
assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
|
||||
if expected, ok := tt.expected.(float64); ok && math.IsNaN(expected) {
|
||||
assert.True(t, math.IsNaN(output.(float64)))
|
||||
number, ok := output.(float64)
|
||||
require.True(t, ok, "want a number, got %T", output)
|
||||
assert.True(t, math.IsNaN(number))
|
||||
} else {
|
||||
assert.Equal(t, tt.expected, output)
|
||||
}
|
||||
|
||||
@@ -86,11 +86,12 @@ func (w *Workflow) OnSchedule() []string {
|
||||
case []any:
|
||||
allSchedules := []string{}
|
||||
for _, v := range val {
|
||||
for k, cron := range v.(map[string]any) {
|
||||
if k != "cron" {
|
||||
continue
|
||||
}
|
||||
allSchedules = append(allSchedules, cron.(string))
|
||||
entry, ok := v.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if cron, ok := entry["cron"].(string); ok {
|
||||
allSchedules = append(allSchedules, cron)
|
||||
}
|
||||
}
|
||||
return allSchedules
|
||||
@@ -443,9 +444,9 @@ func normalizeMatrixValue(key string, val any) ([]any, error) {
|
||||
// Scalar values are wrapped into single-element arrays automatically.
|
||||
// Template expressions are resolved by EvaluateYamlNode before this method is
|
||||
// called; if unresolved, the literal string is wrapped as a one-element fallback.
|
||||
func (j *Job) Matrix() map[string][]any {
|
||||
func (j *Job) Matrix() (map[string][]any, error) {
|
||||
if j.Strategy == nil || j.Strategy.RawMatrix.Kind != yaml.MappingNode {
|
||||
return nil
|
||||
return map[string][]any{}, nil
|
||||
}
|
||||
|
||||
// Decode to flexible map first so that scalar values don't cause a type error.
|
||||
@@ -455,9 +456,9 @@ func (j *Job) Matrix() map[string][]any {
|
||||
// Fall back to the strict array-only format for backward compatibility.
|
||||
var val map[string][]any
|
||||
if !decodeNode(j.Strategy.RawMatrix, &val) {
|
||||
return nil
|
||||
return map[string][]any{}, nil
|
||||
}
|
||||
return val
|
||||
return val, nil
|
||||
}
|
||||
|
||||
// Convert flexible format to expected format with validation
|
||||
@@ -465,12 +466,11 @@ func (j *Job) Matrix() map[string][]any {
|
||||
for k, v := range flexVal {
|
||||
normalized, err := normalizeMatrixValue(k, v)
|
||||
if err != nil {
|
||||
log.Errorf("matrix validation error: %v", err)
|
||||
return nil
|
||||
return nil, err
|
||||
}
|
||||
val[k] = normalized
|
||||
}
|
||||
return val
|
||||
return val, nil
|
||||
}
|
||||
|
||||
// GetMatrixes returns the matrix cross product
|
||||
@@ -482,38 +482,38 @@ func (j *Job) GetMatrixes() ([]map[string]any, error) {
|
||||
j.Strategy.FailFast = j.Strategy.GetFailFast()
|
||||
j.Strategy.MaxParallel = j.Strategy.GetMaxParallel()
|
||||
|
||||
if m := j.Matrix(); m != nil {
|
||||
m, err := j.Matrix()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(m) > 0 {
|
||||
includes := make([]map[string]any, 0)
|
||||
extraIncludes := make([]map[string]any, 0)
|
||||
addInclude := func(raw any) error {
|
||||
include, ok := raw.(map[string]any)
|
||||
if !ok {
|
||||
return fmt.Errorf("the workflow is not valid. Matrix include %v is not a map of matrix keys to values", raw)
|
||||
}
|
||||
for k := range include {
|
||||
if _, ok := m[k]; ok {
|
||||
includes = append(includes, include)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
extraIncludes = append(extraIncludes, include)
|
||||
return nil
|
||||
}
|
||||
for _, v := range m["include"] {
|
||||
switch t := v.(type) {
|
||||
case []any:
|
||||
for _, i := range t {
|
||||
i := i.(map[string]any)
|
||||
extraInclude := true
|
||||
for k := range i {
|
||||
if _, ok := m[k]; ok {
|
||||
includes = append(includes, i)
|
||||
extraInclude = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if extraInclude {
|
||||
extraIncludes = append(extraIncludes, i)
|
||||
if err := addInclude(i); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
case any:
|
||||
v := v.(map[string]any)
|
||||
extraInclude := true
|
||||
for k := range v {
|
||||
if _, ok := m[k]; ok {
|
||||
includes = append(includes, v)
|
||||
extraInclude = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if extraInclude {
|
||||
extraIncludes = append(extraIncludes, v)
|
||||
if err := addInclude(t); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -521,10 +521,13 @@ func (j *Job) GetMatrixes() ([]map[string]any, error) {
|
||||
|
||||
excludes := make([]map[string]any, 0)
|
||||
for _, e := range m["exclude"] {
|
||||
e := e.(map[string]any)
|
||||
for k := range e {
|
||||
exclude, ok := e.(map[string]any)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("the workflow is not valid. Matrix exclude %v is not a map of matrix keys to values", e)
|
||||
}
|
||||
for k := range exclude {
|
||||
if _, ok := m[k]; ok {
|
||||
excludes = append(excludes, e)
|
||||
excludes = append(excludes, exclude)
|
||||
} else {
|
||||
// We fail completely here because that's what GitHub does for non-existing matrix keys, fail on exclude, silent skip on include
|
||||
return nil, fmt.Errorf("the workflow is not valid. Matrix exclude key %q does not match any key within the matrix", k)
|
||||
|
||||
@@ -667,7 +667,9 @@ func TestReadWorkflow_Strategy(t *testing.T) {
|
||||
matrixes, err := job.GetMatrixes()
|
||||
assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
assert.Equal(t, matrixes, []map[string]any{{}}) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
assert.Equal(t, job.Matrix(), map[string][]any(nil))
|
||||
matrix, err := job.Matrix()
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, matrix)
|
||||
assert.Equal(t, job.Strategy.MaxParallel, 2) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
assert.Equal(t, job.Strategy.FailFast, true) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
|
||||
@@ -675,7 +677,9 @@ func TestReadWorkflow_Strategy(t *testing.T) {
|
||||
matrixes, err = job.GetMatrixes()
|
||||
assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
assert.Equal(t, matrixes, []map[string]any{{}}) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
assert.Equal(t, job.Matrix(), map[string][]any(nil))
|
||||
matrix, err = job.Matrix()
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, matrix)
|
||||
assert.Equal(t, job.Strategy.MaxParallel, 4) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
assert.Equal(t, job.Strategy.FailFast, false) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
|
||||
@@ -683,7 +687,9 @@ func TestReadWorkflow_Strategy(t *testing.T) {
|
||||
matrixes, err = job.GetMatrixes()
|
||||
assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
assert.Equal(t, matrixes, []map[string]any{{}}) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
assert.Equal(t, job.Matrix(), map[string][]any(nil))
|
||||
matrix, err = job.Matrix()
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, matrix)
|
||||
assert.Equal(t, job.Strategy.MaxParallel, 2) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
assert.Equal(t, job.Strategy.FailFast, false) //nolint:testifylint // pre-existing issue from nektos/act
|
||||
|
||||
@@ -700,7 +706,9 @@ func TestReadWorkflow_Strategy(t *testing.T) {
|
||||
{"datacenter": "site-b", "node-version": "12.x", "site": "dev"},
|
||||
},
|
||||
)
|
||||
assert.Equal(t, job.Matrix(), //nolint:testifylint // pre-existing issue from nektos/act
|
||||
matrix, err = job.Matrix()
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, matrix, //nolint:testifylint // pre-existing issue from nektos/act
|
||||
map[string][]any{
|
||||
"datacenter": {"site-c", "site-d"},
|
||||
"exclude": {
|
||||
@@ -1092,13 +1100,15 @@ jobs:
|
||||
t.Fatal("job not found")
|
||||
}
|
||||
|
||||
matrix := job.Matrix()
|
||||
matrix, err := job.Matrix()
|
||||
|
||||
if tt.wantErr {
|
||||
require.Error(t, err)
|
||||
assert.Nil(t, matrix, "matrix should be nil on error")
|
||||
} else {
|
||||
require.NoError(t, err)
|
||||
if tt.wantLen == 0 {
|
||||
assert.Nil(t, matrix, "matrix should be nil for jobs without strategy")
|
||||
assert.Empty(t, matrix, "no matrix for jobs without strategy")
|
||||
} else {
|
||||
assert.NotNil(t, matrix, "matrix should not be nil")
|
||||
assert.Len(t, matrix, tt.wantLen, "matrix should have expected number of keys")
|
||||
@@ -1130,11 +1140,9 @@ func TestJobMatrixValidation(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
// Attempt to get matrix
|
||||
matrix := job.Matrix()
|
||||
|
||||
// Should return nil due to validation error
|
||||
assert.Nil(t, matrix, "matrix with nested map should return nil")
|
||||
matrix, err := job.Matrix()
|
||||
require.ErrorContains(t, err, `matrix key "config" has invalid nested object value`)
|
||||
assert.Nil(t, matrix)
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -129,6 +129,16 @@ func readActionImpl(ctx context.Context, step *model.Step, actionDir, actionPath
|
||||
return action, err
|
||||
}
|
||||
|
||||
// cachedActionTar returns the action's tree from the action cache, which only a remote action
|
||||
// has an entry in.
|
||||
func cachedActionTar(ctx context.Context, step actionStep, name, includePrefix string) (io.ReadCloser, error) {
|
||||
remote, ok := step.(*stepActionRemote)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("action %q is a remote action but runs as %T", name, step)
|
||||
}
|
||||
return step.getRunContext().Config.ActionCache.GetTarArchive(ctx, remote.cacheDir, remote.resolvedSha, includePrefix)
|
||||
}
|
||||
|
||||
func maybeCopyToActionDir(ctx context.Context, step actionStep, actionDir, actionPath, containerActionDir string) error {
|
||||
logger := common.Logger(ctx)
|
||||
rc := step.getRunContext()
|
||||
@@ -147,8 +157,7 @@ func maybeCopyToActionDir(ctx context.Context, step actionStep, actionDir, actio
|
||||
}
|
||||
|
||||
if rc.Config != nil && rc.Config.ActionCache != nil {
|
||||
raction := step.(*stepActionRemote)
|
||||
ta, err := rc.Config.ActionCache.GetTarArchive(ctx, raction.cacheDir, raction.resolvedSha, "")
|
||||
ta, err := cachedActionTar(ctx, step, stepModel.Uses, "")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -351,8 +360,7 @@ func execAsDocker(ctx context.Context, step actionStep, actionName, actionDir, b
|
||||
}
|
||||
defer buildContext.Close()
|
||||
} else if rc.Config.ActionCache != nil {
|
||||
rstep := step.(*stepActionRemote)
|
||||
buildContext, err = rc.Config.ActionCache.GetTarArchive(ctx, rstep.cacheDir, rstep.resolvedSha, contextDir)
|
||||
buildContext, err = cachedActionTar(ctx, step, actionName, contextDir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -364,6 +372,7 @@ func execAsDocker(ctx context.Context, step actionStep, actionName, actionDir, b
|
||||
ImageTag: image,
|
||||
BuildContext: buildContext,
|
||||
Platform: rc.Config.ContainerArchitecture,
|
||||
BuildArgs: rc.proxyBuildArgs(),
|
||||
})
|
||||
if buildContext == nil {
|
||||
// Held across the whole build: the daemon drains contextDir lazily.
|
||||
|
||||
@@ -22,13 +22,13 @@ import (
|
||||
"runtime"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"gitea.com/gitea/runner/act/common"
|
||||
"gitea.com/gitea/runner/act/container"
|
||||
"gitea.com/gitea/runner/act/exprparser"
|
||||
"gitea.com/gitea/runner/act/model"
|
||||
"gitea.com/gitea/runner/internal/pkg/lock"
|
||||
|
||||
"github.com/docker/cli/cli/compose/loader"
|
||||
"github.com/docker/go-connections/nat"
|
||||
@@ -435,7 +435,9 @@ func (rc *RunContext) startJobContainer() common.Executor {
|
||||
continue
|
||||
}
|
||||
// interpolate env
|
||||
interpolatedEnvs := make(map[string]string, len(spec.Env))
|
||||
interpolatedEnvs := make(map[string]string, len(spec.Env)+len(rc.Config.ProxyEnv))
|
||||
// a service reaches the internet the way the job does; its own env still wins
|
||||
maps0.Copy(interpolatedEnvs, rc.Config.ProxyEnv)
|
||||
for k, v := range spec.Env {
|
||||
interpolatedEnvs[k] = rc.ExprEval.Interpolate(ctx, v)
|
||||
}
|
||||
@@ -715,13 +717,10 @@ func (rc *RunContext) ActionCacheDir() string {
|
||||
// jobMutexes serializes per-job result/output aggregation across the matrix combinations that
|
||||
// share one *model.Job and run in parallel. Keyed by the shared *model.Job (mirrors the
|
||||
// per-directory AcquireCloneLock pattern).
|
||||
var jobMutexes sync.Map // key: *model.Job; value: *sync.Mutex
|
||||
var jobMutexes lock.Keyed[*model.Job]
|
||||
|
||||
func lockJob(job *model.Job) func() {
|
||||
v, _ := jobMutexes.LoadOrStore(job, &sync.Mutex{})
|
||||
mu := v.(*sync.Mutex)
|
||||
mu.Lock()
|
||||
return mu.Unlock
|
||||
return jobMutexes.Lock(job)
|
||||
}
|
||||
|
||||
func (rc *RunContext) interpolateOutputs() common.Executor {
|
||||
@@ -972,6 +971,20 @@ func (rc *RunContext) isEnabled(ctx context.Context) (bool, error) {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// proxyBuildArgs returns the job's proxy variables as docker build args. The docker CLI
|
||||
// pre-populates these from its own client configuration, but act builds through the API,
|
||||
// so without them a Dockerfile action's RUN steps have no network behind a proxy.
|
||||
func (rc *RunContext) proxyBuildArgs() map[string]*string {
|
||||
if len(rc.Config.ProxyEnv) == 0 {
|
||||
return nil
|
||||
}
|
||||
args := make(map[string]*string, len(rc.Config.ProxyEnv))
|
||||
for name, value := range rc.Config.ProxyEnv {
|
||||
args[name] = &value
|
||||
}
|
||||
return args
|
||||
}
|
||||
|
||||
func mergeMaps(maps ...map[string]string) map[string]string {
|
||||
rtnMap := make(map[string]string)
|
||||
for _, m := range maps {
|
||||
|
||||
@@ -292,6 +292,83 @@ jobs:
|
||||
require.Equal(t, [2]string{"", ""}, credentials["redis:latest"])
|
||||
}
|
||||
|
||||
// A service container reaches the internet the same way the job does, so it inherits the
|
||||
// job's proxy; a service that sets the variable itself keeps its own value.
|
||||
func TestStartJobContainerGivesServicesTheJobProxy(t *testing.T) {
|
||||
workflow, err := model.ReadWorkflow(strings.NewReader(`
|
||||
name: test
|
||||
on: push
|
||||
jobs:
|
||||
job:
|
||||
runs-on: ubuntu-latest
|
||||
container:
|
||||
image: registry.example/job:latest
|
||||
services:
|
||||
redis:
|
||||
image: redis:latest
|
||||
db:
|
||||
image: postgres:latest
|
||||
env:
|
||||
no_proxy: db-only.example
|
||||
steps: []
|
||||
`))
|
||||
require.NoError(t, err)
|
||||
|
||||
var inputs []*container.NewContainerInput
|
||||
origNewContainer := newContainer
|
||||
newContainer = func(input *container.NewContainerInput) container.ExecutionsEnvironment {
|
||||
inputs = append(inputs, input)
|
||||
return fakeContainer{}
|
||||
}
|
||||
t.Cleanup(func() { newContainer = origNewContainer })
|
||||
|
||||
rc := &RunContext{
|
||||
Name: "test",
|
||||
Config: &Config{
|
||||
Workdir: "/tmp",
|
||||
ContainerNetworkMode: "host",
|
||||
ReuseContainers: true,
|
||||
Env: map[string]string{},
|
||||
ProxyEnv: map[string]string{"http_proxy": "http://proxy:3128", "no_proxy": "internal.example"},
|
||||
Secrets: map[string]string{},
|
||||
},
|
||||
Env: map[string]string{},
|
||||
Run: &model.Run{
|
||||
JobID: "job",
|
||||
Workflow: workflow,
|
||||
},
|
||||
}
|
||||
rc.ExprEval = rc.NewExpressionEvaluator(t.Context())
|
||||
|
||||
require.NoError(t, rc.startJobContainer()(t.Context()))
|
||||
|
||||
env := map[string][]string{}
|
||||
for _, in := range inputs {
|
||||
env[in.Image] = in.Env
|
||||
}
|
||||
|
||||
require.Contains(t, env["redis:latest"], "http_proxy=http://proxy:3128")
|
||||
require.Contains(t, env["redis:latest"], "no_proxy=internal.example")
|
||||
// the service's own env wins over what the runner injected, without dropping the rest
|
||||
require.Contains(t, env["postgres:latest"], "no_proxy=db-only.example")
|
||||
require.NotContains(t, env["postgres:latest"], "no_proxy=internal.example")
|
||||
require.Contains(t, env["postgres:latest"], "http_proxy=http://proxy:3128")
|
||||
}
|
||||
|
||||
// act builds Dockerfile actions through the API, which does not pre-populate the proxy
|
||||
// build args the docker CLI would, so the RUN steps would have no network behind a proxy.
|
||||
func TestProxyBuildArgs(t *testing.T) {
|
||||
rc := &RunContext{Config: &Config{ProxyEnv: map[string]string{"http_proxy": "http://proxy:3128"}}}
|
||||
|
||||
args := rc.proxyBuildArgs()
|
||||
|
||||
require.Len(t, args, 1)
|
||||
require.Equal(t, "http://proxy:3128", *args["http_proxy"])
|
||||
|
||||
// a job without a proxy builds exactly as it does today
|
||||
require.Nil(t, (&RunContext{Config: &Config{}}).proxyBuildArgs())
|
||||
}
|
||||
|
||||
func TestRunContext_GetBindsAndMounts(t *testing.T) {
|
||||
rctemplate := &RunContext{
|
||||
Name: "TestRCName",
|
||||
|
||||
@@ -73,6 +73,7 @@ type Config struct {
|
||||
ContainerNetworkMode docker_container.NetworkMode // the network mode of job containers (the value of --network)
|
||||
ContainerNetworkCreateOptions container.NewDockerNetworkCreateExecutorInput // the default network create options
|
||||
ActionCache ActionCache // Use a custom ActionCache Implementation
|
||||
ProxyEnv map[string]string // the proxy variables the job runs with, also given to service containers and image builds
|
||||
|
||||
PresetGitHubContext *model.GithubContext // the preset github context, overrides some fields like DefaultBranch, Env, Secrets etc.
|
||||
EventJSON string // the content of JSON file to use for event.json in containers, overrides EventPath
|
||||
|
||||
@@ -138,9 +138,10 @@ func (sar *stepActionRemote) prepareActionExecutor() common.Executor {
|
||||
})
|
||||
var ntErr common.Executor
|
||||
if err := gitClone(ctx); err != nil {
|
||||
if errors.Is(err, git.ErrShortRef) {
|
||||
var refErr *git.Error
|
||||
if errors.As(err, &refErr) && errors.Is(err, git.ErrShortRef) {
|
||||
return fmt.Errorf("Unable to resolve action `%s`, the provided ref `%s` is the shortened version of a commit SHA, which is not supported. Please use the full commit SHA `%s` instead",
|
||||
sar.Step.Uses, sar.remoteAction.Ref, err.(*git.Error).Commit())
|
||||
sar.Step.Uses, sar.remoteAction.Ref, refErr.Commit())
|
||||
} else if errors.Is(err, gogit.ErrForceNeeded) { // TODO: figure out if it will be easy to shadow/alias go-git err's
|
||||
ntErr = common.NewInfoExecutor("Non-terminating error while running 'git clone': %v", err)
|
||||
} else {
|
||||
|
||||
2
go.mod
2
go.mod
@@ -38,6 +38,7 @@ require (
|
||||
github.com/timshannon/bolthold v0.0.0-20240314194003-30aac6950928
|
||||
go.etcd.io/bbolt v1.5.0
|
||||
go.yaml.in/yaml/v4 v4.0.0-rc.3
|
||||
golang.org/x/net v0.56.0
|
||||
golang.org/x/sys v0.47.0
|
||||
golang.org/x/term v0.45.0
|
||||
golang.org/x/text v0.40.0
|
||||
@@ -105,7 +106,6 @@ require (
|
||||
go.opentelemetry.io/otel/trace v1.44.0 // indirect
|
||||
go.yaml.in/yaml/v3 v3.0.4 // indirect
|
||||
golang.org/x/crypto v0.53.0 // indirect
|
||||
golang.org/x/net v0.56.0 // indirect
|
||||
golang.org/x/sync v0.22.0 // indirect
|
||||
gopkg.in/warnings.v0 v0.1.2 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||
|
||||
@@ -64,6 +64,14 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
|
||||
log.Warn("no labels configured, runner may not be able to pick up jobs")
|
||||
}
|
||||
|
||||
// Before the first Docker API call: the standard library resolves the proxy
|
||||
// environment once. Ungated because host labels still reach the daemon.
|
||||
if dockerSocketPath, err := getDockerSocketPath(cfg.Container.DockerHost); err == nil {
|
||||
run.BypassProxyForDockerHost(dockerSocketPath)
|
||||
} else {
|
||||
log.Debugf("cannot resolve the docker socket path, so Docker API calls are not exempted from the proxy: %v", err)
|
||||
}
|
||||
|
||||
if ls.RequireDocker() || cfg.Container.RequireDocker {
|
||||
// Wait for dockerd be ready
|
||||
if timeout := cfg.Container.DockerTimeout; timeout > 0 {
|
||||
@@ -101,6 +109,7 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
|
||||
}
|
||||
// if dockerSocketPath passes the check, override DOCKER_HOST with dockerSocketPath
|
||||
os.Setenv("DOCKER_HOST", dockerSocketPath)
|
||||
run.WarnIfDaemonHasNoProxy(ctx)
|
||||
// empty cfg.Container.DockerHost means runner need to find an available docker host automatically
|
||||
// and assign the path to cfg.Container.DockerHost
|
||||
if cfg.Container.DockerHost == "" {
|
||||
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
"gitea.com/gitea/runner/act/common"
|
||||
"gitea.com/gitea/runner/act/model"
|
||||
"gitea.com/gitea/runner/act/runner"
|
||||
"gitea.com/gitea/runner/internal/app/run"
|
||||
|
||||
"github.com/joho/godotenv"
|
||||
"github.com/moby/moby/api/types/container"
|
||||
@@ -415,6 +416,11 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
|
||||
}
|
||||
handler.RegisterJob(actionsRuntimeToken, "__local/__exec")
|
||||
|
||||
// no service aliases: exec builds one config for the whole plan
|
||||
run.BypassProxyForDockerHost(os.Getenv("DOCKER_HOST"))
|
||||
proxyEnv := run.JobProxyEnv(env, env["ACTIONS_CACHE_URL"], nil)
|
||||
maps.Copy(env, proxyEnv)
|
||||
|
||||
// run the plan
|
||||
config := &runner.Config{
|
||||
Workdir: execArgs.Workdir(),
|
||||
@@ -425,6 +431,7 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
|
||||
LogOutput: true,
|
||||
JSONLogger: execArgs.jsonLogger,
|
||||
Env: env,
|
||||
ProxyEnv: proxyEnv,
|
||||
Vars: execArgs.LoadVars(),
|
||||
Secrets: execArgs.LoadSecrets(),
|
||||
InsecureSecrets: execArgs.insecureSecrets,
|
||||
|
||||
163
internal/app/run/proxy.go
Normal file
163
internal/app/run/proxy.go
Normal file
@@ -0,0 +1,163 @@
|
||||
// Copyright 2026 The Gitea Authors. All rights reserved.
|
||||
// SPDX-License-Identifier: MIT
|
||||
|
||||
package run
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/url"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"gitea.com/gitea/runner/act/container"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
"golang.org/x/net/http/httpproxy"
|
||||
)
|
||||
|
||||
// proxyFromEnv returns the runner's own proxy configuration, or nil when it has none.
|
||||
func proxyFromEnv() *httpproxy.Config {
|
||||
cfg := httpproxy.FromEnvironment()
|
||||
if cfg.HTTPProxy == "" && cfg.HTTPSProxy == "" {
|
||||
return nil
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
// JobProxyEnv returns the proxy variables a job runs with, given what runner.envs already
|
||||
// put in jobEnvs. Gitea is deliberately not made direct, so it stays reachable however the
|
||||
// runner reaches it.
|
||||
func JobProxyEnv(jobEnvs map[string]string, cacheURL string, serviceNames []string) map[string]string {
|
||||
cfg := proxyFromEnv()
|
||||
if cfg == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Go bypasses loopback on its own, curl and most other tools in a job do not.
|
||||
direct := append([]string{"localhost", "127.0.0.1", "::1"}, serviceNames...)
|
||||
direct = append(direct, hostOf(cacheURL))
|
||||
|
||||
proxyEnv := map[string]string{}
|
||||
setPair := func(lower, upper, value string) {
|
||||
if value == "" {
|
||||
return
|
||||
}
|
||||
// Either spelling in runner.envs takes over both, so the pair cannot disagree.
|
||||
if existing, ok := jobEnvs[lower]; ok {
|
||||
value = existing
|
||||
} else if existing, ok := jobEnvs[upper]; ok {
|
||||
value = existing
|
||||
}
|
||||
proxyEnv[lower], proxyEnv[upper] = value, value
|
||||
}
|
||||
setPair("http_proxy", "HTTP_PROXY", cfg.HTTPProxy)
|
||||
setPair("https_proxy", "HTTPS_PROXY", cfg.HTTPSProxy)
|
||||
|
||||
// no_proxy is merged rather than replaced: the hosts above are structural, and an
|
||||
// operator cannot list the cache server's startup-assigned address in advance.
|
||||
noProxy := appendNoProxy(cfg.NoProxy, direct...)
|
||||
for _, name := range []string{"no_proxy", "NO_PROXY"} {
|
||||
if existing, ok := jobEnvs[name]; ok {
|
||||
noProxy = appendNoProxy(existing, strings.Split(noProxy, ",")...)
|
||||
break
|
||||
}
|
||||
}
|
||||
proxyEnv["no_proxy"], proxyEnv["NO_PROXY"] = noProxy, noProxy
|
||||
return proxyEnv
|
||||
}
|
||||
|
||||
// BypassProxyForDockerHost keeps the runner's Docker API traffic off the proxy: the docker
|
||||
// client proxies every transport that is not a unix socket or a named pipe, so a tcp://
|
||||
// daemon would be reached through a proxy that cannot route to it.
|
||||
//
|
||||
// It must run before the first Docker API call, because the standard library resolves the
|
||||
// proxy environment once.
|
||||
func BypassProxyForDockerHost(dockerHost string) {
|
||||
cfg := proxyFromEnv()
|
||||
if cfg == nil {
|
||||
return
|
||||
}
|
||||
host := hostOf(dockerHost)
|
||||
if host == "" {
|
||||
// A unix socket or named pipe is never proxied.
|
||||
return
|
||||
}
|
||||
|
||||
noProxy := appendNoProxy(cfg.NoProxy, host)
|
||||
for _, name := range []string{"no_proxy", "NO_PROXY"} {
|
||||
if err := os.Setenv(name, noProxy); err != nil {
|
||||
log.Warnf("cannot set %s for the runner process: %v", name, err)
|
||||
}
|
||||
}
|
||||
log.Debugf("docker host %s is reached directly, no_proxy is now %q", host, noProxy)
|
||||
}
|
||||
|
||||
// WarnIfDaemonHasNoProxy points at the one part the runner cannot set: the docker daemon
|
||||
// pulls the images, and a daemon in its own container needs its own proxy.
|
||||
func WarnIfDaemonHasNoProxy(ctx context.Context) {
|
||||
if proxyFromEnv() == nil {
|
||||
return
|
||||
}
|
||||
info, err := container.GetHostInfo(ctx)
|
||||
if err != nil {
|
||||
log.Debugf("cannot read the docker daemon's proxy configuration: %v", err)
|
||||
return
|
||||
}
|
||||
if info.HTTPProxy == "" && info.HTTPSProxy == "" {
|
||||
log.Warn("the runner has a proxy but the docker daemon reports none, so image pulls will not use it: https://docs.docker.com/engine/daemon/proxy/")
|
||||
}
|
||||
}
|
||||
|
||||
// proxyPasswords returns the passwords embedded in the runner's proxy URLs, to mask before
|
||||
// a job echoes its environment.
|
||||
func proxyPasswords() []string {
|
||||
cfg := proxyFromEnv()
|
||||
if cfg == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
var passwords []string
|
||||
for _, raw := range []string{cfg.HTTPProxy, cfg.HTTPSProxy} {
|
||||
parsed, err := url.Parse(raw)
|
||||
if err != nil || parsed.User == nil {
|
||||
continue
|
||||
}
|
||||
if password, ok := parsed.User.Password(); ok && password != "" && !slices.Contains(passwords, password) {
|
||||
passwords = append(passwords, password)
|
||||
}
|
||||
}
|
||||
return passwords
|
||||
}
|
||||
|
||||
// appendNoProxy adds hosts to a no_proxy list, keeping the operator's entries and adding
|
||||
// none twice.
|
||||
func appendNoProxy(noProxy string, hosts ...string) string {
|
||||
entries := []string{}
|
||||
for entry := range strings.SplitSeq(noProxy, ",") {
|
||||
if entry = strings.TrimSpace(entry); entry != "" {
|
||||
entries = append(entries, entry)
|
||||
}
|
||||
}
|
||||
|
||||
for _, host := range hosts {
|
||||
if host == "" || slices.Contains(entries, host) {
|
||||
continue
|
||||
}
|
||||
entries = append(entries, host)
|
||||
}
|
||||
return strings.Join(entries, ",")
|
||||
}
|
||||
|
||||
// hostOf returns the host of a URL without its port, the form a no_proxy entry takes. It is
|
||||
// empty for anything without a network host, such as a unix socket.
|
||||
func hostOf(raw string) string {
|
||||
if raw == "" {
|
||||
return ""
|
||||
}
|
||||
parsed, err := url.Parse(strings.TrimSuffix(raw, "/"))
|
||||
if err != nil || parsed.Hostname() == "" {
|
||||
return ""
|
||||
}
|
||||
return parsed.Hostname()
|
||||
}
|
||||
192
internal/app/run/proxy_test.go
Normal file
192
internal/app/run/proxy_test.go
Normal file
@@ -0,0 +1,192 @@
|
||||
// Copyright 2026 The Gitea Authors. All rights reserved.
|
||||
// SPDX-License-Identifier: MIT
|
||||
|
||||
package run
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
// clearProxyEnv starts a test from a runner that has no proxy, whatever the developer's own
|
||||
// environment looks like.
|
||||
func clearProxyEnv(t *testing.T) {
|
||||
t.Helper()
|
||||
for _, name := range []string{"http_proxy", "HTTP_PROXY", "https_proxy", "HTTPS_PROXY", "no_proxy", "NO_PROXY"} {
|
||||
t.Setenv(name, "")
|
||||
}
|
||||
}
|
||||
|
||||
func TestJobProxyEnv(t *testing.T) {
|
||||
const loopback = "localhost,127.0.0.1,::1"
|
||||
const proxy = "http://proxy:3128"
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
runner map[string]string // the runner process environment
|
||||
envs map[string]string // what runner.envs already put in the job
|
||||
cacheURL string
|
||||
services []string
|
||||
want map[string]string
|
||||
}{
|
||||
{
|
||||
// The guarantee that makes this safe to ship: a runner without a proxy gives its
|
||||
// jobs nothing at all.
|
||||
name: "runner has no proxy",
|
||||
cacheURL: "http://cache.local:8088/",
|
||||
},
|
||||
{
|
||||
name: "a lone no_proxy is not a proxy",
|
||||
runner: map[string]string{"no_proxy": "example.com"},
|
||||
},
|
||||
{
|
||||
name: "both spellings of every variable",
|
||||
runner: map[string]string{"http_proxy": proxy, "https_proxy": proxy},
|
||||
want: map[string]string{
|
||||
"http_proxy": proxy, "HTTP_PROXY": proxy,
|
||||
"https_proxy": proxy, "HTTPS_PROXY": proxy,
|
||||
"no_proxy": loopback, "NO_PROXY": loopback,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "a variable the runner does not have is omitted",
|
||||
runner: map[string]string{"https_proxy": proxy},
|
||||
want: map[string]string{
|
||||
"https_proxy": proxy, "HTTPS_PROXY": proxy,
|
||||
"no_proxy": loopback, "NO_PROXY": loopback,
|
||||
},
|
||||
},
|
||||
{
|
||||
// The cache server and the service containers are on the local network; Gitea is
|
||||
// not added and keeps being reached through the proxy.
|
||||
name: "hosts the job must reach directly",
|
||||
runner: map[string]string{"http_proxy": proxy, "no_proxy": "internal.example"},
|
||||
cacheURL: "http://192.168.1.10:34567/",
|
||||
services: []string{"postgres", "redis"},
|
||||
want: map[string]string{
|
||||
"http_proxy": proxy, "HTTP_PROXY": proxy,
|
||||
"no_proxy": "internal.example," + loopback + ",postgres,redis,192.168.1.10",
|
||||
"NO_PROXY": "internal.example," + loopback + ",postgres,redis,192.168.1.10",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "runner.envs wins, for both spellings",
|
||||
runner: map[string]string{"http_proxy": "http://from-env:3128"},
|
||||
envs: map[string]string{"http_proxy": "http://from-config:3128"},
|
||||
want: map[string]string{
|
||||
"http_proxy": "http://from-config:3128", "HTTP_PROXY": "http://from-config:3128",
|
||||
"no_proxy": loopback, "NO_PROXY": loopback,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "runner.envs wins through the upper case spelling too",
|
||||
runner: map[string]string{"http_proxy": "http://from-env:3128"},
|
||||
envs: map[string]string{"HTTP_PROXY": "http://from-config:3128"},
|
||||
want: map[string]string{
|
||||
"http_proxy": "http://from-config:3128", "HTTP_PROXY": "http://from-config:3128",
|
||||
"no_proxy": loopback, "NO_PROXY": loopback,
|
||||
},
|
||||
},
|
||||
{
|
||||
// A runner.envs no_proxy adds to the hosts that must stay direct rather than
|
||||
// replacing them, which would send cache traffic through the proxy.
|
||||
name: "runner.envs no_proxy is merged, not substituted",
|
||||
runner: map[string]string{"http_proxy": proxy},
|
||||
envs: map[string]string{"no_proxy": "operator.example"},
|
||||
cacheURL: "http://192.168.1.10:34567/",
|
||||
want: map[string]string{
|
||||
"http_proxy": proxy, "HTTP_PROXY": proxy,
|
||||
"no_proxy": "operator.example," + loopback + ",192.168.1.10",
|
||||
"NO_PROXY": "operator.example," + loopback + ",192.168.1.10",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "runner.envs NO_PROXY is merged through the upper case spelling too",
|
||||
runner: map[string]string{"http_proxy": proxy},
|
||||
envs: map[string]string{"NO_PROXY": "operator.example"},
|
||||
cacheURL: "http://192.168.1.10:34567/",
|
||||
want: map[string]string{
|
||||
"http_proxy": proxy, "HTTP_PROXY": proxy,
|
||||
"no_proxy": "operator.example," + loopback + ",192.168.1.10",
|
||||
"NO_PROXY": "operator.example," + loopback + ",192.168.1.10",
|
||||
},
|
||||
},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
clearProxyEnv(t)
|
||||
for name, value := range tc.runner {
|
||||
t.Setenv(name, value)
|
||||
}
|
||||
assert.Equal(t, tc.want, JobProxyEnv(tc.envs, tc.cacheURL, tc.services))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// docker-in-docker over tcp: the docker client would otherwise send API calls to a proxy
|
||||
// that cannot route to the daemon.
|
||||
func TestBypassProxyForDockerHost(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
httpProxy string
|
||||
dockerHost string
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "tcp daemon is added",
|
||||
httpProxy: "http://proxy:3128",
|
||||
dockerHost: "tcp://docker:2375",
|
||||
want: "internal.example,docker",
|
||||
},
|
||||
{
|
||||
name: "unix socket is left alone",
|
||||
httpProxy: "http://proxy:3128",
|
||||
dockerHost: "unix:///var/run/docker.sock",
|
||||
want: "internal.example",
|
||||
},
|
||||
{
|
||||
name: "nothing happens without a proxy",
|
||||
dockerHost: "tcp://docker:2375",
|
||||
want: "internal.example",
|
||||
},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
clearProxyEnv(t)
|
||||
t.Setenv("no_proxy", "internal.example")
|
||||
if tc.httpProxy != "" {
|
||||
t.Setenv("http_proxy", tc.httpProxy)
|
||||
}
|
||||
|
||||
BypassProxyForDockerHost(tc.dockerHost)
|
||||
|
||||
assert.Equal(t, tc.want, os.Getenv("no_proxy"))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestProxyPasswords(t *testing.T) {
|
||||
clearProxyEnv(t)
|
||||
t.Setenv("http_proxy", "http://user:hunter2@proxy:3128")
|
||||
t.Setenv("https_proxy", "http://user:s3cret@proxy:3128")
|
||||
assert.Equal(t, []string{"hunter2", "s3cret"}, proxyPasswords())
|
||||
|
||||
t.Setenv("https_proxy", "http://proxy:3128")
|
||||
assert.Equal(t, []string{"hunter2"}, proxyPasswords())
|
||||
}
|
||||
|
||||
func TestAppendNoProxy(t *testing.T) {
|
||||
assert.Equal(t, "a.example,b.example", appendNoProxy(" a.example , b.example "))
|
||||
// a host already listed is not repeated
|
||||
assert.Equal(t, "cache.local", appendNoProxy("cache.local", "cache.local"))
|
||||
assert.Empty(t, appendNoProxy("", "", ""))
|
||||
}
|
||||
|
||||
func TestHostOf(t *testing.T) {
|
||||
assert.Equal(t, "cache.local", hostOf("http://cache.local:8088/"))
|
||||
assert.Equal(t, "192.168.1.10", hostOf("http://192.168.1.10:34567"))
|
||||
// nothing to bypass for a socket path or an unparseable URL
|
||||
assert.Empty(t, hostOf("unix:///var/run/docker.sock"))
|
||||
assert.Empty(t, hostOf("cache.local:8088"))
|
||||
assert.Empty(t, hostOf("://nope"))
|
||||
}
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -268,7 +269,8 @@ func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) error {
|
||||
|
||||
ctx, cancel := context.WithTimeout(ctx, r.cfg.Runner.Timeout)
|
||||
defer cancel()
|
||||
reporter := report.NewReporter(ctx, cancel, r.client, task, r.cfg)
|
||||
// A proxy URL may carry credentials, and every job is given it; keep them out of the log.
|
||||
reporter := report.NewReporter(ctx, cancel, r.client, task, r.cfg, proxyPasswords()...)
|
||||
var runErr error
|
||||
defer func() {
|
||||
r.runningCount.Add(-1)
|
||||
@@ -341,6 +343,11 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
|
||||
taskContext := task.Context.Fields
|
||||
envs := r.cloneEnvs()
|
||||
|
||||
// Added per task because this job's service containers must be reached directly, and
|
||||
// act reaches them by their workflow key.
|
||||
proxyEnv := JobProxyEnv(envs, envs["ACTIONS_CACHE_URL"], slices.Sorted(maps.Keys(job.Services)))
|
||||
maps.Copy(envs, proxyEnv)
|
||||
|
||||
if r.capabilities != "" {
|
||||
envs["GITEA_ACTIONS_CAPABILITIES"] = r.capabilities
|
||||
}
|
||||
@@ -450,6 +457,7 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
|
||||
LogOutput: true,
|
||||
JSONLogger: false,
|
||||
Env: envs,
|
||||
ProxyEnv: proxyEnv,
|
||||
Secrets: task.Secrets,
|
||||
GitHubInstance: strings.TrimSuffix(r.client.Address(), "/"),
|
||||
AutoRemove: true,
|
||||
|
||||
@@ -93,6 +93,24 @@ func TestNewRunnerInitializesLabelsAndEnvironment(t *testing.T) {
|
||||
require.Nil(t, r.cacheHandler)
|
||||
}
|
||||
|
||||
// Proxy variables are assembled per task, because a job's service containers have to be
|
||||
// reached directly and they are only known once the workflow is parsed.
|
||||
func TestNewRunnerLeavesProxyToTheTask(t *testing.T) {
|
||||
clearProxyEnv(t)
|
||||
t.Setenv("http_proxy", "http://proxy:3128")
|
||||
|
||||
cfg := &config.Config{}
|
||||
cfg.Cache.ExternalServer = "http://cache.local:8088/"
|
||||
reg := &config.Registration{Name: "runner"}
|
||||
cli := clientmocks.NewClient(t)
|
||||
cli.On("Address").Return("https://gitea.example/").Maybe()
|
||||
|
||||
r := NewRunner(cfg, reg, cli)
|
||||
|
||||
require.NotContains(t, r.envs, "http_proxy")
|
||||
require.NotContains(t, r.envs, "no_proxy")
|
||||
}
|
||||
|
||||
func taskWithDefaultActionsURL(url string) *runnerv1.Task {
|
||||
return &runnerv1.Task{
|
||||
Context: &structpb.Struct{
|
||||
|
||||
38
internal/pkg/lock/keyed.go
Normal file
38
internal/pkg/lock/keyed.go
Normal file
@@ -0,0 +1,38 @@
|
||||
// Copyright 2026 The Gitea Authors. All rights reserved.
|
||||
// SPDX-License-Identifier: MIT
|
||||
|
||||
package lock
|
||||
|
||||
import "sync"
|
||||
|
||||
// Keyed serializes the work done under one key while letting different keys run in
|
||||
// parallel. Its zero value is ready to use, and it keeps one mutex per key it has seen.
|
||||
type Keyed[K comparable] struct {
|
||||
mu sync.Mutex
|
||||
mutexes map[K]*sync.Mutex
|
||||
}
|
||||
|
||||
// Lock locks the mutex for key and returns its unlock function.
|
||||
func (kl *Keyed[K]) Lock(key K) func() {
|
||||
kl.mu.Lock()
|
||||
if kl.mutexes == nil {
|
||||
kl.mutexes = map[K]*sync.Mutex{}
|
||||
}
|
||||
lock := kl.mutexes[key]
|
||||
if lock == nil {
|
||||
lock = &sync.Mutex{}
|
||||
kl.mutexes[key] = lock
|
||||
}
|
||||
kl.mu.Unlock()
|
||||
|
||||
lock.Lock()
|
||||
return lock.Unlock
|
||||
}
|
||||
|
||||
// Delete drops the mutex for key. Holders of it keep working, they just no longer share it
|
||||
// with a later Lock of the same key.
|
||||
func (kl *Keyed[K]) Delete(key K) {
|
||||
kl.mu.Lock()
|
||||
delete(kl.mutexes, key)
|
||||
kl.mu.Unlock()
|
||||
}
|
||||
@@ -32,6 +32,12 @@ const (
|
||||
maxOutputValueLen = 1024 * 1024 // 1 MiB
|
||||
)
|
||||
|
||||
// jobOutput is a job output on its way to the server, sent once the server has acknowledged it.
|
||||
type jobOutput struct {
|
||||
value string
|
||||
sent bool
|
||||
}
|
||||
|
||||
type Reporter struct {
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
@@ -53,7 +59,8 @@ type Reporter struct {
|
||||
state *runnerv1.TaskState
|
||||
stateChanged bool
|
||||
stateMu sync.RWMutex
|
||||
outputs sync.Map
|
||||
outputsMu sync.Mutex
|
||||
outputs map[string]jobOutput
|
||||
daemon chan struct{}
|
||||
heartbeatStop chan struct{}
|
||||
heartbeatStopOnce sync.Once
|
||||
@@ -79,8 +86,13 @@ type Reporter struct {
|
||||
stopCommandEndToken string
|
||||
}
|
||||
|
||||
func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.Client, task *runnerv1.Task, cfg *config.Config) *Reporter {
|
||||
// extraMasks are values known before the job starts that are not among its secrets, such as
|
||||
// the password in the runner's proxy URL.
|
||||
func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.Client, task *runnerv1.Task, cfg *config.Config, extraMasks ...string) *Reporter {
|
||||
var oldnew []string
|
||||
for _, v := range extraMasks {
|
||||
oldnew = runner.AppendSecretMasker(oldnew, v)
|
||||
}
|
||||
if v := task.Context.Fields["token"].GetStringValue(); v != "" {
|
||||
oldnew = runner.AppendSecretMasker(oldnew, v)
|
||||
}
|
||||
@@ -394,7 +406,12 @@ func (r *Reporter) logf(format string, a ...any) {
|
||||
func (r *Reporter) SetOutputs(outputs map[string]string) {
|
||||
r.stateMu.Lock()
|
||||
defer r.stateMu.Unlock()
|
||||
r.outputsMu.Lock()
|
||||
defer r.outputsMu.Unlock()
|
||||
|
||||
if r.outputs == nil {
|
||||
r.outputs = map[string]jobOutput{}
|
||||
}
|
||||
for k, v := range outputs {
|
||||
if l := len(k); l > maxOutputKeyLen {
|
||||
log.Warnf("ignore output %q because the key is too long: %d > %d", k, l, maxOutputKeyLen)
|
||||
@@ -406,10 +423,9 @@ func (r *Reporter) SetOutputs(outputs map[string]string) {
|
||||
r.logf("ignore output %q because the value is too long: %d > %d", k, l, maxOutputValueLen)
|
||||
continue
|
||||
}
|
||||
if _, ok := r.outputs.Load(k); ok {
|
||||
continue
|
||||
if _, ok := r.outputs[k]; !ok {
|
||||
r.outputs[k] = jobOutput{value: v}
|
||||
}
|
||||
r.outputs.Store(k, v)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -574,14 +590,14 @@ func (r *Reporter) ReportState(reportResult bool) error {
|
||||
r.clientM.Lock()
|
||||
defer r.clientM.Unlock()
|
||||
|
||||
// Build the outputs map first (single Range pass instead of two).
|
||||
outputs := make(map[string]string)
|
||||
r.outputs.Range(func(k, v any) bool {
|
||||
if val, ok := v.(string); ok {
|
||||
outputs[k.(string)] = val
|
||||
r.outputsMu.Lock()
|
||||
for key, out := range r.outputs {
|
||||
if !out.sent {
|
||||
outputs[key] = out.value
|
||||
}
|
||||
return true
|
||||
})
|
||||
}
|
||||
r.outputsMu.Unlock()
|
||||
|
||||
// Consume stateChanged atomically with the snapshot; restored on error
|
||||
// below so a concurrent Fire() during UpdateTask isn't silently lost.
|
||||
@@ -594,7 +610,8 @@ func (r *Reporter) ReportState(reportResult bool) error {
|
||||
r.stateMu.Unlock()
|
||||
return nil
|
||||
}
|
||||
state := proto.Clone(r.state).(*runnerv1.TaskState)
|
||||
state := &runnerv1.TaskState{}
|
||||
proto.Merge(state, r.state)
|
||||
r.stateChanged = false
|
||||
r.stateMu.Unlock()
|
||||
|
||||
@@ -622,21 +639,23 @@ func (r *Reporter) ReportState(reportResult bool) error {
|
||||
metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultSuccess).Inc()
|
||||
r.lastReportedAtNanos.Store(time.Now().UnixNano())
|
||||
|
||||
var noSent []string
|
||||
r.outputsMu.Lock()
|
||||
for _, k := range resp.Msg.SentOutputs {
|
||||
r.outputs.Store(k, struct{}{})
|
||||
if _, ok := r.outputs[k]; ok {
|
||||
r.outputs[k] = jobOutput{sent: true}
|
||||
}
|
||||
}
|
||||
for key, out := range r.outputs {
|
||||
if !out.sent {
|
||||
noSent = append(noSent, key)
|
||||
}
|
||||
}
|
||||
r.outputsMu.Unlock()
|
||||
|
||||
if resp.Msg.State != nil && resp.Msg.State.Result == runnerv1.Result_RESULT_CANCELLED {
|
||||
r.cancel()
|
||||
}
|
||||
|
||||
var noSent []string
|
||||
r.outputs.Range(func(k, v any) bool {
|
||||
if _, ok := v.(string); ok {
|
||||
noSent = append(noSent, k.(string))
|
||||
}
|
||||
return true
|
||||
})
|
||||
if len(noSent) > 0 {
|
||||
return fmt.Errorf("there are still outputs that have not been sent: %v", noSent)
|
||||
}
|
||||
|
||||
@@ -7,6 +7,8 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
@@ -1011,33 +1013,59 @@ func TestReporter_SetOutputs(t *testing.T) {
|
||||
r := &Reporter{state: &runnerv1.TaskState{}}
|
||||
|
||||
r.SetOutputs(map[string]string{"foo": "bar"})
|
||||
got, ok := r.outputs.Load("foo")
|
||||
got, ok := r.outputs["foo"]
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, "bar", got)
|
||||
assert.Equal(t, "bar", got.value)
|
||||
|
||||
// first value wins: a later write to the same key is ignored
|
||||
r.SetOutputs(map[string]string{"foo": "baz"})
|
||||
got, _ = r.outputs.Load("foo")
|
||||
assert.Equal(t, "bar", got)
|
||||
assert.Equal(t, "bar", r.outputs["foo"].value)
|
||||
|
||||
// keys longer than maxOutputKeyLen are dropped
|
||||
longKey := strings.Repeat("k", maxOutputKeyLen+1)
|
||||
r.SetOutputs(map[string]string{longKey: "v"})
|
||||
_, ok = r.outputs.Load(longKey)
|
||||
_, ok = r.outputs[longKey]
|
||||
assert.False(t, ok)
|
||||
|
||||
// values longer than maxOutputValueLen are dropped
|
||||
longValue := strings.Repeat("v", maxOutputValueLen+1)
|
||||
r.SetOutputs(map[string]string{"big": longValue})
|
||||
_, ok = r.outputs.Load("big")
|
||||
_, ok = r.outputs["big"]
|
||||
assert.False(t, ok)
|
||||
|
||||
// a value at exactly the limit is still stored
|
||||
maxValue := strings.Repeat("v", maxOutputValueLen)
|
||||
r.SetOutputs(map[string]string{"atlimit": maxValue})
|
||||
got, ok = r.outputs.Load("atlimit")
|
||||
got, ok = r.outputs["atlimit"]
|
||||
require.True(t, ok)
|
||||
assert.Len(t, got, maxOutputValueLen)
|
||||
assert.Len(t, got.value, maxOutputValueLen)
|
||||
}
|
||||
|
||||
// An output the server acknowledged is not reported again.
|
||||
func TestReporter_OutputsSentOnce(t *testing.T) {
|
||||
client := mocks.NewClient(t)
|
||||
var reported []map[string]string
|
||||
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
|
||||
func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
|
||||
reported = append(reported, req.Msg.Outputs)
|
||||
return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{SentOutputs: slices.Collect(maps.Keys(req.Msg.Outputs))}), nil
|
||||
})
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
taskCtx, err := structpb.NewStruct(map[string]any{})
|
||||
require.NoError(t, err)
|
||||
cfg, _ := config.LoadDefault("")
|
||||
r := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
|
||||
|
||||
r.SetOutputs(map[string]string{"foo": "bar"})
|
||||
require.NoError(t, r.ReportState(false))
|
||||
assert.True(t, r.outputs["foo"].sent)
|
||||
|
||||
require.NoError(t, r.ReportState(true))
|
||||
require.Len(t, reported, 2)
|
||||
assert.Equal(t, map[string]string{"foo": "bar"}, reported[0])
|
||||
assert.Empty(t, reported[1])
|
||||
}
|
||||
|
||||
func TestReporter_EffectiveCloseTimeout(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user