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
|
- dupl
|
||||||
- errcheck
|
- errcheck
|
||||||
- forbidigo
|
- forbidigo
|
||||||
|
- forcetypeassert
|
||||||
- gocheckcompilerdirectives
|
- gocheckcompilerdirectives
|
||||||
- gocritic
|
- gocritic
|
||||||
- goheader
|
- goheader
|
||||||
@@ -102,6 +103,9 @@ linters:
|
|||||||
- linters:
|
- linters:
|
||||||
- forbidigo
|
- forbidigo
|
||||||
path: cmd
|
path: cmd
|
||||||
|
- linters:
|
||||||
|
- forcetypeassert
|
||||||
|
path: _test\.go
|
||||||
issues:
|
issues:
|
||||||
max-issues-per-linter: 0
|
max-issues-per-linter: 0
|
||||||
max-same-issues: 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.
|
> **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`)
|
#### 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.
|
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
|
storage *Storage
|
||||||
router *httprouter.Router
|
router *httprouter.Router
|
||||||
listener net.Listener
|
listener net.Listener
|
||||||
|
port int
|
||||||
server *http.Server
|
server *http.Server
|
||||||
logger logrus.FieldLogger
|
logger logrus.FieldLogger
|
||||||
|
|
||||||
@@ -177,6 +178,12 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
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{
|
server := &http.Server{
|
||||||
ReadHeaderTimeout: 2 * time.Second,
|
ReadHeaderTimeout: 2 * time.Second,
|
||||||
Handler: router,
|
Handler: router,
|
||||||
@@ -194,9 +201,7 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
|
|||||||
|
|
||||||
func (h *Handler) ExternalURL() string {
|
func (h *Handler) ExternalURL() string {
|
||||||
// TODO: make the external url configurable if necessary
|
// TODO: make the external url configurable if necessary
|
||||||
return fmt.Sprintf("http://%s:%d",
|
return fmt.Sprintf("http://%s:%d", h.outboundIP, h.port)
|
||||||
h.outboundIP,
|
|
||||||
h.listener.Addr().(*net.TCPAddr).Port)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// RegisterJob makes token a valid bearer credential for cache requests from
|
// RegisterJob makes token a valid bearer credential for cache requests from
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"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"
|
||||||
"github.com/go-git/go-git/v5/config"
|
"github.com/go-git/go-git/v5/config"
|
||||||
@@ -32,7 +33,7 @@ var (
|
|||||||
githubHTTPRegex = regexp.MustCompile(`^https?://.*github.com.*/(.+)/(.+?)(?:.git)?$`)
|
githubHTTPRegex = regexp.MustCompile(`^https?://.*github.com.*/(.+)/(.+?)(?:.git)?$`)
|
||||||
githubSSHRegex = regexp.MustCompile(`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")
|
ErrShortRef = errors.New("short SHA references are not supported")
|
||||||
ErrNoRepo = errors.New("unable to find git repo")
|
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,
|
// 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.
|
// otherwise a concurrent NewGitCloneExecutor on the same dir can mutate the worktree mid-read.
|
||||||
func AcquireCloneLock(dir string) func() {
|
func AcquireCloneLock(dir string) func() {
|
||||||
v, _ := cloneLocks.LoadOrStore(dir, &sync.Mutex{})
|
return cloneLocks.Lock(dir)
|
||||||
mu := v.(*sync.Mutex)
|
|
||||||
mu.Lock()
|
|
||||||
return mu.Unlock
|
|
||||||
}
|
}
|
||||||
|
|
||||||
type Error struct {
|
type Error struct {
|
||||||
|
|||||||
@@ -12,7 +12,6 @@ import (
|
|||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
|
||||||
"syscall"
|
"syscall"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -610,12 +609,4 @@ func TestAcquireCloneLock(t *testing.T) {
|
|||||||
t.Fatal("acquire on a different directory must not block")
|
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")
|
conn, err := net.Dial("udp", "8.8.8.8:80")
|
||||||
if err == nil {
|
if err == nil {
|
||||||
defer conn.Close()
|
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.
|
// So the machine cannot access the internet. Pick an IP address from network interfaces.
|
||||||
|
|||||||
@@ -82,6 +82,7 @@ type NewDockerBuildExecutorInput struct {
|
|||||||
BuildContext io.Reader
|
BuildContext io.Reader
|
||||||
ImageTag string
|
ImageTag string
|
||||||
Platform string
|
Platform string
|
||||||
|
BuildArgs map[string]*string
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewDockerNetworkCreateExecutorInput the input for the NewDockerNetworkCreateExecutor function
|
// NewDockerNetworkCreateExecutorInput the input for the NewDockerNetworkCreateExecutor function
|
||||||
|
|||||||
@@ -49,6 +49,7 @@ func NewDockerBuildExecutor(input NewDockerBuildExecutorInput) common.Executor {
|
|||||||
Remove: true,
|
Remove: true,
|
||||||
AuthConfigs: LoadDockerAuthConfigs(ctx),
|
AuthConfigs: LoadDockerAuthConfigs(ctx),
|
||||||
Dockerfile: input.Dockerfile,
|
Dockerfile: input.Dockerfile,
|
||||||
|
BuildArgs: input.BuildArgs,
|
||||||
}
|
}
|
||||||
platform, err := parsePlatform(input.Platform)
|
platform, err := parsePlatform(input.Platform)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -51,7 +51,8 @@ func TestCreateFlagsValidate(t *testing.T) {
|
|||||||
|
|
||||||
func TestNewContainerAppliesCreateFlags(t *testing.T) {
|
func TestNewContainerAppliesCreateFlags(t *testing.T) {
|
||||||
input := &NewContainerInput{Platform: "linux/amd64", Options: "--platform linux/arm64 --pull never"}
|
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, "linux/arm64", input.Platform)
|
||||||
assert.Equal(t, pullPolicyNever, cr.pullPolicy)
|
assert.Equal(t, pullPolicyNever, cr.pullPolicy)
|
||||||
|
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ import (
|
|||||||
"gitea.com/gitea/runner/act/model"
|
"gitea.com/gitea/runner/act/model"
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestLiterals(t *testing.T) {
|
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
|
assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act
|
||||||
|
|
||||||
if expected, ok := tt.expected.(float64); ok && math.IsNaN(expected) {
|
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 {
|
} else {
|
||||||
assert.Equal(t, tt.expected, output)
|
assert.Equal(t, tt.expected, output)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -86,11 +86,12 @@ func (w *Workflow) OnSchedule() []string {
|
|||||||
case []any:
|
case []any:
|
||||||
allSchedules := []string{}
|
allSchedules := []string{}
|
||||||
for _, v := range val {
|
for _, v := range val {
|
||||||
for k, cron := range v.(map[string]any) {
|
entry, ok := v.(map[string]any)
|
||||||
if k != "cron" {
|
if !ok {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
allSchedules = append(allSchedules, cron.(string))
|
if cron, ok := entry["cron"].(string); ok {
|
||||||
|
allSchedules = append(allSchedules, cron)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return allSchedules
|
return allSchedules
|
||||||
@@ -443,9 +444,9 @@ func normalizeMatrixValue(key string, val any) ([]any, error) {
|
|||||||
// Scalar values are wrapped into single-element arrays automatically.
|
// Scalar values are wrapped into single-element arrays automatically.
|
||||||
// Template expressions are resolved by EvaluateYamlNode before this method is
|
// Template expressions are resolved by EvaluateYamlNode before this method is
|
||||||
// called; if unresolved, the literal string is wrapped as a one-element fallback.
|
// 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 {
|
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.
|
// 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.
|
// Fall back to the strict array-only format for backward compatibility.
|
||||||
var val map[string][]any
|
var val map[string][]any
|
||||||
if !decodeNode(j.Strategy.RawMatrix, &val) {
|
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
|
// Convert flexible format to expected format with validation
|
||||||
@@ -465,12 +466,11 @@ func (j *Job) Matrix() map[string][]any {
|
|||||||
for k, v := range flexVal {
|
for k, v := range flexVal {
|
||||||
normalized, err := normalizeMatrixValue(k, v)
|
normalized, err := normalizeMatrixValue(k, v)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("matrix validation error: %v", err)
|
return nil, err
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
val[k] = normalized
|
val[k] = normalized
|
||||||
}
|
}
|
||||||
return val
|
return val, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetMatrixes returns the matrix cross product
|
// 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.FailFast = j.Strategy.GetFailFast()
|
||||||
j.Strategy.MaxParallel = j.Strategy.GetMaxParallel()
|
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)
|
includes := make([]map[string]any, 0)
|
||||||
extraIncludes := 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"] {
|
for _, v := range m["include"] {
|
||||||
switch t := v.(type) {
|
switch t := v.(type) {
|
||||||
case []any:
|
case []any:
|
||||||
for _, i := range t {
|
for _, i := range t {
|
||||||
i := i.(map[string]any)
|
if err := addInclude(i); err != nil {
|
||||||
extraInclude := true
|
return nil, err
|
||||||
for k := range i {
|
|
||||||
if _, ok := m[k]; ok {
|
|
||||||
includes = append(includes, i)
|
|
||||||
extraInclude = false
|
|
||||||
break
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if extraInclude {
|
|
||||||
extraIncludes = append(extraIncludes, i)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
case any:
|
case any:
|
||||||
v := v.(map[string]any)
|
if err := addInclude(t); err != nil {
|
||||||
extraInclude := true
|
return nil, err
|
||||||
for k := range v {
|
|
||||||
if _, ok := m[k]; ok {
|
|
||||||
includes = append(includes, v)
|
|
||||||
extraInclude = false
|
|
||||||
break
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if extraInclude {
|
|
||||||
extraIncludes = append(extraIncludes, v)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -521,10 +521,13 @@ func (j *Job) GetMatrixes() ([]map[string]any, error) {
|
|||||||
|
|
||||||
excludes := make([]map[string]any, 0)
|
excludes := make([]map[string]any, 0)
|
||||||
for _, e := range m["exclude"] {
|
for _, e := range m["exclude"] {
|
||||||
e := e.(map[string]any)
|
exclude, ok := e.(map[string]any)
|
||||||
for k := range e {
|
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 {
|
if _, ok := m[k]; ok {
|
||||||
excludes = append(excludes, e)
|
excludes = append(excludes, exclude)
|
||||||
} else {
|
} else {
|
||||||
// We fail completely here because that's what GitHub does for non-existing matrix keys, fail on exclude, silent skip on include
|
// 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)
|
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()
|
matrixes, err := job.GetMatrixes()
|
||||||
assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act
|
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, 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.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
|
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()
|
matrixes, err = job.GetMatrixes()
|
||||||
assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act
|
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, 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.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
|
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()
|
matrixes, err = job.GetMatrixes()
|
||||||
assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act
|
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, 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.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
|
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"},
|
{"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{
|
map[string][]any{
|
||||||
"datacenter": {"site-c", "site-d"},
|
"datacenter": {"site-c", "site-d"},
|
||||||
"exclude": {
|
"exclude": {
|
||||||
@@ -1092,13 +1100,15 @@ jobs:
|
|||||||
t.Fatal("job not found")
|
t.Fatal("job not found")
|
||||||
}
|
}
|
||||||
|
|
||||||
matrix := job.Matrix()
|
matrix, err := job.Matrix()
|
||||||
|
|
||||||
if tt.wantErr {
|
if tt.wantErr {
|
||||||
|
require.Error(t, err)
|
||||||
assert.Nil(t, matrix, "matrix should be nil on error")
|
assert.Nil(t, matrix, "matrix should be nil on error")
|
||||||
} else {
|
} else {
|
||||||
|
require.NoError(t, err)
|
||||||
if tt.wantLen == 0 {
|
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 {
|
} else {
|
||||||
assert.NotNil(t, matrix, "matrix should not be nil")
|
assert.NotNil(t, matrix, "matrix should not be nil")
|
||||||
assert.Len(t, matrix, tt.wantLen, "matrix should have expected number of keys")
|
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, err := job.Matrix()
|
||||||
matrix := job.Matrix()
|
require.ErrorContains(t, err, `matrix key "config" has invalid nested object value`)
|
||||||
|
assert.Nil(t, matrix)
|
||||||
// Should return nil due to validation error
|
|
||||||
assert.Nil(t, matrix, "matrix with nested map should return nil")
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -129,6 +129,16 @@ func readActionImpl(ctx context.Context, step *model.Step, actionDir, actionPath
|
|||||||
return action, err
|
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 {
|
func maybeCopyToActionDir(ctx context.Context, step actionStep, actionDir, actionPath, containerActionDir string) error {
|
||||||
logger := common.Logger(ctx)
|
logger := common.Logger(ctx)
|
||||||
rc := step.getRunContext()
|
rc := step.getRunContext()
|
||||||
@@ -147,8 +157,7 @@ func maybeCopyToActionDir(ctx context.Context, step actionStep, actionDir, actio
|
|||||||
}
|
}
|
||||||
|
|
||||||
if rc.Config != nil && rc.Config.ActionCache != nil {
|
if rc.Config != nil && rc.Config.ActionCache != nil {
|
||||||
raction := step.(*stepActionRemote)
|
ta, err := cachedActionTar(ctx, step, stepModel.Uses, "")
|
||||||
ta, err := rc.Config.ActionCache.GetTarArchive(ctx, raction.cacheDir, raction.resolvedSha, "")
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -351,8 +360,7 @@ func execAsDocker(ctx context.Context, step actionStep, actionName, actionDir, b
|
|||||||
}
|
}
|
||||||
defer buildContext.Close()
|
defer buildContext.Close()
|
||||||
} else if rc.Config.ActionCache != nil {
|
} else if rc.Config.ActionCache != nil {
|
||||||
rstep := step.(*stepActionRemote)
|
buildContext, err = cachedActionTar(ctx, step, actionName, contextDir)
|
||||||
buildContext, err = rc.Config.ActionCache.GetTarArchive(ctx, rstep.cacheDir, rstep.resolvedSha, contextDir)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -364,6 +372,7 @@ func execAsDocker(ctx context.Context, step actionStep, actionName, actionDir, b
|
|||||||
ImageTag: image,
|
ImageTag: image,
|
||||||
BuildContext: buildContext,
|
BuildContext: buildContext,
|
||||||
Platform: rc.Config.ContainerArchitecture,
|
Platform: rc.Config.ContainerArchitecture,
|
||||||
|
BuildArgs: rc.proxyBuildArgs(),
|
||||||
})
|
})
|
||||||
if buildContext == nil {
|
if buildContext == nil {
|
||||||
// Held across the whole build: the daemon drains contextDir lazily.
|
// Held across the whole build: the daemon drains contextDir lazily.
|
||||||
|
|||||||
@@ -22,13 +22,13 @@ import (
|
|||||||
"runtime"
|
"runtime"
|
||||||
"slices"
|
"slices"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/act/common"
|
||||||
"gitea.com/gitea/runner/act/container"
|
"gitea.com/gitea/runner/act/container"
|
||||||
"gitea.com/gitea/runner/act/exprparser"
|
"gitea.com/gitea/runner/act/exprparser"
|
||||||
"gitea.com/gitea/runner/act/model"
|
"gitea.com/gitea/runner/act/model"
|
||||||
|
"gitea.com/gitea/runner/internal/pkg/lock"
|
||||||
|
|
||||||
"github.com/docker/cli/cli/compose/loader"
|
"github.com/docker/cli/cli/compose/loader"
|
||||||
"github.com/docker/go-connections/nat"
|
"github.com/docker/go-connections/nat"
|
||||||
@@ -435,7 +435,9 @@ func (rc *RunContext) startJobContainer() common.Executor {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
// interpolate env
|
// 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 {
|
for k, v := range spec.Env {
|
||||||
interpolatedEnvs[k] = rc.ExprEval.Interpolate(ctx, v)
|
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
|
// 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
|
// share one *model.Job and run in parallel. Keyed by the shared *model.Job (mirrors the
|
||||||
// per-directory AcquireCloneLock pattern).
|
// 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() {
|
func lockJob(job *model.Job) func() {
|
||||||
v, _ := jobMutexes.LoadOrStore(job, &sync.Mutex{})
|
return jobMutexes.Lock(job)
|
||||||
mu := v.(*sync.Mutex)
|
|
||||||
mu.Lock()
|
|
||||||
return mu.Unlock
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (rc *RunContext) interpolateOutputs() common.Executor {
|
func (rc *RunContext) interpolateOutputs() common.Executor {
|
||||||
@@ -972,6 +971,20 @@ func (rc *RunContext) isEnabled(ctx context.Context) (bool, error) {
|
|||||||
return true, nil
|
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 {
|
func mergeMaps(maps ...map[string]string) map[string]string {
|
||||||
rtnMap := make(map[string]string)
|
rtnMap := make(map[string]string)
|
||||||
for _, m := range maps {
|
for _, m := range maps {
|
||||||
|
|||||||
@@ -292,6 +292,83 @@ jobs:
|
|||||||
require.Equal(t, [2]string{"", ""}, credentials["redis:latest"])
|
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) {
|
func TestRunContext_GetBindsAndMounts(t *testing.T) {
|
||||||
rctemplate := &RunContext{
|
rctemplate := &RunContext{
|
||||||
Name: "TestRCName",
|
Name: "TestRCName",
|
||||||
|
|||||||
@@ -73,6 +73,7 @@ type Config struct {
|
|||||||
ContainerNetworkMode docker_container.NetworkMode // the network mode of job containers (the value of --network)
|
ContainerNetworkMode docker_container.NetworkMode // the network mode of job containers (the value of --network)
|
||||||
ContainerNetworkCreateOptions container.NewDockerNetworkCreateExecutorInput // the default network create options
|
ContainerNetworkCreateOptions container.NewDockerNetworkCreateExecutorInput // the default network create options
|
||||||
ActionCache ActionCache // Use a custom ActionCache Implementation
|
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.
|
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
|
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
|
var ntErr common.Executor
|
||||||
if err := gitClone(ctx); err != nil {
|
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",
|
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
|
} 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)
|
ntErr = common.NewInfoExecutor("Non-terminating error while running 'git clone': %v", err)
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
2
go.mod
2
go.mod
@@ -38,6 +38,7 @@ require (
|
|||||||
github.com/timshannon/bolthold v0.0.0-20240314194003-30aac6950928
|
github.com/timshannon/bolthold v0.0.0-20240314194003-30aac6950928
|
||||||
go.etcd.io/bbolt v1.5.0
|
go.etcd.io/bbolt v1.5.0
|
||||||
go.yaml.in/yaml/v4 v4.0.0-rc.3
|
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/sys v0.47.0
|
||||||
golang.org/x/term v0.45.0
|
golang.org/x/term v0.45.0
|
||||||
golang.org/x/text v0.40.0
|
golang.org/x/text v0.40.0
|
||||||
@@ -105,7 +106,6 @@ require (
|
|||||||
go.opentelemetry.io/otel/trace v1.44.0 // indirect
|
go.opentelemetry.io/otel/trace v1.44.0 // indirect
|
||||||
go.yaml.in/yaml/v3 v3.0.4 // indirect
|
go.yaml.in/yaml/v3 v3.0.4 // indirect
|
||||||
golang.org/x/crypto v0.53.0 // 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
|
golang.org/x/sync v0.22.0 // indirect
|
||||||
gopkg.in/warnings.v0 v0.1.2 // indirect
|
gopkg.in/warnings.v0 v0.1.2 // indirect
|
||||||
gopkg.in/yaml.v3 v3.0.1 // 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")
|
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 {
|
if ls.RequireDocker() || cfg.Container.RequireDocker {
|
||||||
// Wait for dockerd be ready
|
// Wait for dockerd be ready
|
||||||
if timeout := cfg.Container.DockerTimeout; timeout > 0 {
|
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
|
// if dockerSocketPath passes the check, override DOCKER_HOST with dockerSocketPath
|
||||||
os.Setenv("DOCKER_HOST", dockerSocketPath)
|
os.Setenv("DOCKER_HOST", dockerSocketPath)
|
||||||
|
run.WarnIfDaemonHasNoProxy(ctx)
|
||||||
// empty cfg.Container.DockerHost means runner need to find an available docker host automatically
|
// empty cfg.Container.DockerHost means runner need to find an available docker host automatically
|
||||||
// and assign the path to cfg.Container.DockerHost
|
// and assign the path to cfg.Container.DockerHost
|
||||||
if cfg.Container.DockerHost == "" {
|
if cfg.Container.DockerHost == "" {
|
||||||
|
|||||||
@@ -22,6 +22,7 @@ import (
|
|||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/act/common"
|
||||||
"gitea.com/gitea/runner/act/model"
|
"gitea.com/gitea/runner/act/model"
|
||||||
"gitea.com/gitea/runner/act/runner"
|
"gitea.com/gitea/runner/act/runner"
|
||||||
|
"gitea.com/gitea/runner/internal/app/run"
|
||||||
|
|
||||||
"github.com/joho/godotenv"
|
"github.com/joho/godotenv"
|
||||||
"github.com/moby/moby/api/types/container"
|
"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")
|
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
|
// run the plan
|
||||||
config := &runner.Config{
|
config := &runner.Config{
|
||||||
Workdir: execArgs.Workdir(),
|
Workdir: execArgs.Workdir(),
|
||||||
@@ -425,6 +431,7 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
|
|||||||
LogOutput: true,
|
LogOutput: true,
|
||||||
JSONLogger: execArgs.jsonLogger,
|
JSONLogger: execArgs.jsonLogger,
|
||||||
Env: env,
|
Env: env,
|
||||||
|
ProxyEnv: proxyEnv,
|
||||||
Vars: execArgs.LoadVars(),
|
Vars: execArgs.LoadVars(),
|
||||||
Secrets: execArgs.LoadSecrets(),
|
Secrets: execArgs.LoadSecrets(),
|
||||||
InsecureSecrets: execArgs.insecureSecrets,
|
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"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"runtime"
|
"runtime"
|
||||||
|
"slices"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"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)
|
ctx, cancel := context.WithTimeout(ctx, r.cfg.Runner.Timeout)
|
||||||
defer cancel()
|
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
|
var runErr error
|
||||||
defer func() {
|
defer func() {
|
||||||
r.runningCount.Add(-1)
|
r.runningCount.Add(-1)
|
||||||
@@ -341,6 +343,11 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
|
|||||||
taskContext := task.Context.Fields
|
taskContext := task.Context.Fields
|
||||||
envs := r.cloneEnvs()
|
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 != "" {
|
if r.capabilities != "" {
|
||||||
envs["GITEA_ACTIONS_CAPABILITIES"] = 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,
|
LogOutput: true,
|
||||||
JSONLogger: false,
|
JSONLogger: false,
|
||||||
Env: envs,
|
Env: envs,
|
||||||
|
ProxyEnv: proxyEnv,
|
||||||
Secrets: task.Secrets,
|
Secrets: task.Secrets,
|
||||||
GitHubInstance: strings.TrimSuffix(r.client.Address(), "/"),
|
GitHubInstance: strings.TrimSuffix(r.client.Address(), "/"),
|
||||||
AutoRemove: true,
|
AutoRemove: true,
|
||||||
|
|||||||
@@ -93,6 +93,24 @@ func TestNewRunnerInitializesLabelsAndEnvironment(t *testing.T) {
|
|||||||
require.Nil(t, r.cacheHandler)
|
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 {
|
func taskWithDefaultActionsURL(url string) *runnerv1.Task {
|
||||||
return &runnerv1.Task{
|
return &runnerv1.Task{
|
||||||
Context: &structpb.Struct{
|
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
|
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 {
|
type Reporter struct {
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
@@ -53,7 +59,8 @@ type Reporter struct {
|
|||||||
state *runnerv1.TaskState
|
state *runnerv1.TaskState
|
||||||
stateChanged bool
|
stateChanged bool
|
||||||
stateMu sync.RWMutex
|
stateMu sync.RWMutex
|
||||||
outputs sync.Map
|
outputsMu sync.Mutex
|
||||||
|
outputs map[string]jobOutput
|
||||||
daemon chan struct{}
|
daemon chan struct{}
|
||||||
heartbeatStop chan struct{}
|
heartbeatStop chan struct{}
|
||||||
heartbeatStopOnce sync.Once
|
heartbeatStopOnce sync.Once
|
||||||
@@ -79,8 +86,13 @@ type Reporter struct {
|
|||||||
stopCommandEndToken string
|
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
|
var oldnew []string
|
||||||
|
for _, v := range extraMasks {
|
||||||
|
oldnew = runner.AppendSecretMasker(oldnew, v)
|
||||||
|
}
|
||||||
if v := task.Context.Fields["token"].GetStringValue(); v != "" {
|
if v := task.Context.Fields["token"].GetStringValue(); v != "" {
|
||||||
oldnew = runner.AppendSecretMasker(oldnew, 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) {
|
func (r *Reporter) SetOutputs(outputs map[string]string) {
|
||||||
r.stateMu.Lock()
|
r.stateMu.Lock()
|
||||||
defer r.stateMu.Unlock()
|
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 {
|
for k, v := range outputs {
|
||||||
if l := len(k); l > maxOutputKeyLen {
|
if l := len(k); l > maxOutputKeyLen {
|
||||||
log.Warnf("ignore output %q because the key is too long: %d > %d", 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)
|
r.logf("ignore output %q because the value is too long: %d > %d", k, l, maxOutputValueLen)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if _, ok := r.outputs.Load(k); ok {
|
if _, ok := r.outputs[k]; !ok {
|
||||||
continue
|
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()
|
r.clientM.Lock()
|
||||||
defer r.clientM.Unlock()
|
defer r.clientM.Unlock()
|
||||||
|
|
||||||
// Build the outputs map first (single Range pass instead of two).
|
|
||||||
outputs := make(map[string]string)
|
outputs := make(map[string]string)
|
||||||
r.outputs.Range(func(k, v any) bool {
|
r.outputsMu.Lock()
|
||||||
if val, ok := v.(string); ok {
|
for key, out := range r.outputs {
|
||||||
outputs[k.(string)] = val
|
if !out.sent {
|
||||||
|
outputs[key] = out.value
|
||||||
}
|
}
|
||||||
return true
|
}
|
||||||
})
|
r.outputsMu.Unlock()
|
||||||
|
|
||||||
// Consume stateChanged atomically with the snapshot; restored on error
|
// Consume stateChanged atomically with the snapshot; restored on error
|
||||||
// below so a concurrent Fire() during UpdateTask isn't silently lost.
|
// 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()
|
r.stateMu.Unlock()
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
state := proto.Clone(r.state).(*runnerv1.TaskState)
|
state := &runnerv1.TaskState{}
|
||||||
|
proto.Merge(state, r.state)
|
||||||
r.stateChanged = false
|
r.stateChanged = false
|
||||||
r.stateMu.Unlock()
|
r.stateMu.Unlock()
|
||||||
|
|
||||||
@@ -622,21 +639,23 @@ func (r *Reporter) ReportState(reportResult bool) error {
|
|||||||
metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultSuccess).Inc()
|
metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultSuccess).Inc()
|
||||||
r.lastReportedAtNanos.Store(time.Now().UnixNano())
|
r.lastReportedAtNanos.Store(time.Now().UnixNano())
|
||||||
|
|
||||||
|
var noSent []string
|
||||||
|
r.outputsMu.Lock()
|
||||||
for _, k := range resp.Msg.SentOutputs {
|
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 {
|
if resp.Msg.State != nil && resp.Msg.State.Result == runnerv1.Result_RESULT_CANCELLED {
|
||||||
r.cancel()
|
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 {
|
if len(noSent) > 0 {
|
||||||
return fmt.Errorf("there are still outputs that have not been sent: %v", noSent)
|
return fmt.Errorf("there are still outputs that have not been sent: %v", noSent)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,6 +7,8 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"maps"
|
||||||
|
"slices"
|
||||||
"strings"
|
"strings"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
@@ -1011,33 +1013,59 @@ func TestReporter_SetOutputs(t *testing.T) {
|
|||||||
r := &Reporter{state: &runnerv1.TaskState{}}
|
r := &Reporter{state: &runnerv1.TaskState{}}
|
||||||
|
|
||||||
r.SetOutputs(map[string]string{"foo": "bar"})
|
r.SetOutputs(map[string]string{"foo": "bar"})
|
||||||
got, ok := r.outputs.Load("foo")
|
got, ok := r.outputs["foo"]
|
||||||
require.True(t, ok)
|
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
|
// first value wins: a later write to the same key is ignored
|
||||||
r.SetOutputs(map[string]string{"foo": "baz"})
|
r.SetOutputs(map[string]string{"foo": "baz"})
|
||||||
got, _ = r.outputs.Load("foo")
|
assert.Equal(t, "bar", r.outputs["foo"].value)
|
||||||
assert.Equal(t, "bar", got)
|
|
||||||
|
|
||||||
// keys longer than maxOutputKeyLen are dropped
|
// keys longer than maxOutputKeyLen are dropped
|
||||||
longKey := strings.Repeat("k", maxOutputKeyLen+1)
|
longKey := strings.Repeat("k", maxOutputKeyLen+1)
|
||||||
r.SetOutputs(map[string]string{longKey: "v"})
|
r.SetOutputs(map[string]string{longKey: "v"})
|
||||||
_, ok = r.outputs.Load(longKey)
|
_, ok = r.outputs[longKey]
|
||||||
assert.False(t, ok)
|
assert.False(t, ok)
|
||||||
|
|
||||||
// values longer than maxOutputValueLen are dropped
|
// values longer than maxOutputValueLen are dropped
|
||||||
longValue := strings.Repeat("v", maxOutputValueLen+1)
|
longValue := strings.Repeat("v", maxOutputValueLen+1)
|
||||||
r.SetOutputs(map[string]string{"big": longValue})
|
r.SetOutputs(map[string]string{"big": longValue})
|
||||||
_, ok = r.outputs.Load("big")
|
_, ok = r.outputs["big"]
|
||||||
assert.False(t, ok)
|
assert.False(t, ok)
|
||||||
|
|
||||||
// a value at exactly the limit is still stored
|
// a value at exactly the limit is still stored
|
||||||
maxValue := strings.Repeat("v", maxOutputValueLen)
|
maxValue := strings.Repeat("v", maxOutputValueLen)
|
||||||
r.SetOutputs(map[string]string{"atlimit": maxValue})
|
r.SetOutputs(map[string]string{"atlimit": maxValue})
|
||||||
got, ok = r.outputs.Load("atlimit")
|
got, ok = r.outputs["atlimit"]
|
||||||
require.True(t, ok)
|
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) {
|
func TestReporter_EffectiveCloseTimeout(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user