mirror of
https://gitea.com/gitea/act_runner
synced 2026-09-21 19:37:07 +02:00
feat: add GITEA_DOCKER_WORKSPACE and container cleanup (#1204)
1. Add `GITEA_DOCKER_WORKSPACE` which holds the workspace path as the daemon sees it, enabling `${GITEA_DOCKER_WORKSPACE:-.}/data:/app/data` in a compose file without having to resort to `bind_workdir` (which causes much more problems like breaking `actions/cache` because of unstable workspace paths).
2. Add container/network/volume cleanup for containers started within jobs, for example via `docker compose` inside a job. It works by running a lightweight docker socket proxy and injecting a `com.gitea.runner.job` label into every container creation and that label is used to remove containers started by that job at the end. Perf impact of this is near-zero.
Docs: https://gitea.com/gitea/docs/pulls/535
Assisted by Claude (Fable 5.1).
Co-authored-by: bircni <bircni@icloud.com>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1204
Reviewed-by: bircni <bircni@icloud.com>
Co-authored-by: silverwind <me@silverwind.io>
This commit is contained in:
committed by
bircni
co-authored by
bircni
parent
c158ac5472
commit
9e3647395a
@@ -266,6 +266,17 @@ A job in a container reaches a service by its id on the job network, on the port
|
||||
|
||||
Unlike GitHub, a job whose steps run on the host (a `host` label without `container:`) starts no service containers, so `job.services` and `job.container` stay empty. Give such a job a `container:` when it needs services.
|
||||
|
||||
#### Docker from a job (`GITEA_DOCKER_WORKSPACE`)
|
||||
|
||||
A container a job starts through the Docker socket cannot bind-mount the workspace by the job's own path, the daemon does not have it. `GITEA_DOCKER_WORKSPACE` holds the path the daemon sees. Use it as the prefix of workspace binds, with `.` as the fallback for local use, here in a `docker-compose.yaml`:
|
||||
|
||||
```yaml
|
||||
volumes:
|
||||
- ${GITEA_DOCKER_WORKSPACE:-.}/data:/app/data
|
||||
```
|
||||
|
||||
Containers, networks and volumes a job creates through the socket are removed when the job ends. Both apply to jobs in containers, on the host `.` already works.
|
||||
|
||||
#### Proxy
|
||||
|
||||
Set these variables in the runner's environment, with systemd `Environment=`, `docker run -e`, or Kubernetes `env:`:
|
||||
|
||||
@@ -75,6 +75,17 @@ const (
|
||||
// fragment, missingContainerError composes it into the message every operation shares.
|
||||
var ErrContainerNotFound = errors.New("does not exist")
|
||||
|
||||
// DockerProxy is a job's docker socket, fronting the daemon's for the job's lifetime.
|
||||
type DockerProxy struct {
|
||||
Socket string
|
||||
close func(context.Context) error
|
||||
}
|
||||
|
||||
// Close removes what the job created through the socket, then stops serving it.
|
||||
func (p *DockerProxy) Close(ctx context.Context) error {
|
||||
return p.close(ctx)
|
||||
}
|
||||
|
||||
// Info is a snapshot of a container, as of one inspect.
|
||||
type Info struct {
|
||||
ID string
|
||||
@@ -84,6 +95,7 @@ type Info struct {
|
||||
// HealthOutput is the last healthcheck probe's output.
|
||||
HealthOutput string
|
||||
Ports map[string]string // container port ("5432") to the host port it is published on
|
||||
Mounts map[string]string // container path to its source on the daemon
|
||||
}
|
||||
|
||||
// Container for managing docker run containers
|
||||
|
||||
@@ -0,0 +1,276 @@
|
||||
// Copyright 2026 The Gitea Authors. All rights reserved.
|
||||
// SPDX-License-Identifier: MIT
|
||||
|
||||
//go:build !(WITHOUT_DOCKER || !(linux || darwin || windows || netbsd))
|
||||
|
||||
package container
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httputil"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"gitea.com/gitea/runner/act/common"
|
||||
|
||||
cerrdefs "github.com/containerd/errdefs"
|
||||
"github.com/moby/moby/api/types/container"
|
||||
"github.com/moby/moby/api/types/mount"
|
||||
"github.com/moby/moby/client"
|
||||
)
|
||||
|
||||
const (
|
||||
jobLabel = "com.gitea.runner.job"
|
||||
maxCreateBody = 8 << 20
|
||||
)
|
||||
|
||||
var (
|
||||
createPath = regexp.MustCompile(`^(/v[0-9.]+)?/(containers|networks|volumes)/create$`)
|
||||
rawStreamPath = regexp.MustCompile(`^(/v[0-9.]+)?/(containers/[^/]+/attach|exec/[^/]+/start)$`)
|
||||
proxyProbe struct {
|
||||
sync.Mutex
|
||||
decided bool
|
||||
dir string
|
||||
}
|
||||
)
|
||||
|
||||
// DockerProxyDir returns where job proxy sockets live, "" while undecided or when the daemon cannot open the runner's files.
|
||||
func DockerProxyDir(ctx context.Context) string {
|
||||
proxyProbe.Lock()
|
||||
defer proxyProbe.Unlock()
|
||||
if proxyProbe.decided {
|
||||
return proxyProbe.dir
|
||||
}
|
||||
dir := filepath.Join(os.TempDir(), "gitea-runner-docker")
|
||||
ok, err := daemonSeesDir(ctx, dir)
|
||||
if err != nil {
|
||||
common.Logger(ctx).Debugf("docker proxy probe postponed: %v", err)
|
||||
return ""
|
||||
}
|
||||
proxyProbe.decided = true
|
||||
if ok {
|
||||
proxyProbe.dir = dir
|
||||
} else {
|
||||
common.Logger(ctx).Infof("the docker daemon cannot reach the runner's filesystem, jobs get the daemon socket directly")
|
||||
}
|
||||
return proxyProbe.dir
|
||||
}
|
||||
|
||||
func daemonSeesDir(ctx context.Context, dir string) (bool, error) {
|
||||
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||
return false, err
|
||||
}
|
||||
probe := filepath.Join(dir, "probe")
|
||||
if err := os.WriteFile(probe, nil, 0o600); err != nil {
|
||||
return false, err
|
||||
}
|
||||
cli, err := GetDockerClient(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
defer cli.Close()
|
||||
images, err := cli.ImageList(ctx, client.ImageListOptions{})
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if len(images.Items) == 0 {
|
||||
return false, errors.New("no image to probe with yet")
|
||||
}
|
||||
// creating validates that a bind source exists on the daemon's side, nothing is started
|
||||
created, err := cli.ContainerCreate(ctx, client.ContainerCreateOptions{
|
||||
Config: &container.Config{Image: images.Items[0].ID, Cmd: []string{"true"}},
|
||||
HostConfig: &container.HostConfig{Mounts: []mount.Mount{{Type: mount.TypeBind, Source: probe, Target: "/gitea-runner-probe"}}},
|
||||
})
|
||||
if cerrdefs.IsInvalidArgument(err) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
_, err = cli.ContainerRemove(ctx, created.ID, client.ContainerRemoveOptions{Force: true})
|
||||
return true, err
|
||||
}
|
||||
|
||||
// StartDockerProxy serves a job's docker socket in dir, labelling what the job creates through it.
|
||||
func StartDockerProxy(daemonSocket, dir, job string) (*DockerProxy, error) {
|
||||
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
digest := sha256.Sum256([]byte(job))
|
||||
socket := filepath.Join(dir, hex.EncodeToString(digest[:8])+".sock")
|
||||
_ = os.Remove(socket)
|
||||
listener, err := net.Listen("unix", socket)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if info, err := os.Stat(daemonSocket); err == nil {
|
||||
_ = os.Chmod(socket, info.Mode().Perm())
|
||||
}
|
||||
dial := func(ctx context.Context, _, _ string) (net.Conn, error) {
|
||||
return (&net.Dialer{}).DialContext(ctx, "unix", daemonSocket)
|
||||
}
|
||||
transport := &http.Transport{DialContext: dial}
|
||||
forward := &httputil.ReverseProxy{
|
||||
Rewrite: func(r *httputil.ProxyRequest) {
|
||||
r.Out.URL.Scheme = "http"
|
||||
r.Out.URL.Host = "docker"
|
||||
},
|
||||
Transport: transport,
|
||||
}
|
||||
server := &http.Server{ReadHeaderTimeout: 30 * time.Second, Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case r.Method != http.MethodPost:
|
||||
case createPath.MatchString(r.URL.Path):
|
||||
if err := addLabel(r, job); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
case rawStreamPath.MatchString(r.URL.Path):
|
||||
tunnel(w, r, dial)
|
||||
return
|
||||
}
|
||||
forward.ServeHTTP(w, r)
|
||||
})}
|
||||
go func() { _ = server.Serve(listener) }()
|
||||
return &DockerProxy{Socket: socket, close: func(ctx context.Context) error {
|
||||
err := removeJobResources(ctx, job)
|
||||
_ = server.Close()
|
||||
transport.CloseIdleConnections()
|
||||
_ = os.Remove(socket)
|
||||
return err
|
||||
}}, nil
|
||||
}
|
||||
|
||||
func addLabel(r *http.Request, job string) error {
|
||||
body, err := io.ReadAll(io.LimitReader(r.Body, maxCreateBody+1))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(body) > maxCreateBody {
|
||||
return errors.New("create request too large")
|
||||
}
|
||||
if len(bytes.TrimSpace(body)) == 0 {
|
||||
r.Body = io.NopCloser(bytes.NewReader(body))
|
||||
return nil
|
||||
}
|
||||
var fields map[string]json.RawMessage
|
||||
if err := json.Unmarshal(body, &fields); err != nil {
|
||||
return fmt.Errorf("invalid create request: %w", err)
|
||||
}
|
||||
key := "Labels"
|
||||
for name := range fields {
|
||||
if strings.EqualFold(name, key) {
|
||||
key = name
|
||||
break
|
||||
}
|
||||
}
|
||||
labels := map[string]string{}
|
||||
if raw := fields[key]; len(raw) > 0 && string(raw) != "null" {
|
||||
if err := json.Unmarshal(raw, &labels); err != nil {
|
||||
return fmt.Errorf("invalid create request: %w", err)
|
||||
}
|
||||
}
|
||||
labels[jobLabel] = job
|
||||
if fields[key], err = json.Marshal(labels); err != nil {
|
||||
return err
|
||||
}
|
||||
if body, err = json.Marshal(fields); err != nil {
|
||||
return err
|
||||
}
|
||||
r.Body = io.NopCloser(bytes.NewReader(body))
|
||||
r.ContentLength = int64(len(body))
|
||||
r.TransferEncoding = nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// tunnel splices attach and exec streams, which the daemon hijacks with or without an HTTP upgrade
|
||||
func tunnel(w http.ResponseWriter, r *http.Request, dial func(context.Context, string, string) (net.Conn, error)) {
|
||||
hijacker, ok := w.(http.Hijacker)
|
||||
if !ok {
|
||||
http.Error(w, "connection cannot be hijacked", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
upstream, err := dial(r.Context(), "", "")
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
defer upstream.Close()
|
||||
if err := r.Write(upstream); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
downstream, buffered, err := hijacker.Hijack()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer downstream.Close()
|
||||
if _, err := io.CopyN(upstream, buffered, int64(buffered.Reader.Buffered())); err != nil {
|
||||
return
|
||||
}
|
||||
done := make(chan struct{}, 2)
|
||||
go func() {
|
||||
_, _ = io.Copy(upstream, downstream)
|
||||
done <- struct{}{}
|
||||
}()
|
||||
go func() {
|
||||
_, _ = io.Copy(downstream, upstream)
|
||||
done <- struct{}{}
|
||||
}()
|
||||
<-done
|
||||
}
|
||||
|
||||
func removeJobResources(ctx context.Context, job string) error {
|
||||
cli, err := GetDockerClient(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer cli.Close()
|
||||
return removeLabelled(ctx, cli, job)
|
||||
}
|
||||
|
||||
func removeLabelled(ctx context.Context, cli client.APIClient, job string) error {
|
||||
logger := common.Logger(ctx)
|
||||
filters := make(client.Filters).Add("label", jobLabel+"="+job)
|
||||
containers, err := cli.ContainerList(ctx, client.ContainerListOptions{All: true, Filters: filters})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var errs []error
|
||||
for _, c := range containers.Items {
|
||||
logger.Infof("removing container %s the job left behind", strings.TrimPrefix(strings.Join(c.Names, ","), "/"))
|
||||
errs = append(errs, (&containerReference{cli: cli, id: c.ID}).remove()(ctx))
|
||||
}
|
||||
networks, err := cli.NetworkList(ctx, client.NetworkListOptions{Filters: filters})
|
||||
if err != nil {
|
||||
return errors.Join(append(errs, err)...)
|
||||
}
|
||||
for _, n := range networks.Items {
|
||||
if _, err := cli.NetworkRemove(ctx, n.ID, client.NetworkRemoveOptions{}); err != nil && !cerrdefs.IsNotFound(err) {
|
||||
errs = append(errs, fmt.Errorf("failed to remove network %s: %w", n.Name, err))
|
||||
}
|
||||
}
|
||||
volumes, err := cli.VolumeList(ctx, client.VolumeListOptions{Filters: filters})
|
||||
if err != nil {
|
||||
return errors.Join(append(errs, err)...)
|
||||
}
|
||||
for _, v := range volumes.Items {
|
||||
if _, err := cli.VolumeRemove(ctx, v.Name, client.VolumeRemoveOptions{}); err != nil && !cerrdefs.IsNotFound(err) {
|
||||
errs = append(errs, fmt.Errorf("failed to remove volume %s: %w", v.Name, err))
|
||||
}
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
@@ -0,0 +1,270 @@
|
||||
// Copyright 2026 The Gitea Authors. All rights reserved.
|
||||
// SPDX-License-Identifier: MIT
|
||||
|
||||
//go:build !(WITHOUT_DOCKER || !(linux || darwin || windows || netbsd))
|
||||
|
||||
package container
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
cerrdefs "github.com/containerd/errdefs"
|
||||
"github.com/moby/moby/api/types/container"
|
||||
"github.com/moby/moby/api/types/network"
|
||||
"github.com/moby/moby/api/types/volume"
|
||||
mobyclient "github.com/moby/moby/client"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/mock"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func (m *mockDockerClient) VolumeList(ctx context.Context, opts mobyclient.VolumeListOptions) (mobyclient.VolumeListResult, error) {
|
||||
args := m.Called(ctx, opts)
|
||||
return args.Get(0).(mobyclient.VolumeListResult), args.Error(1)
|
||||
}
|
||||
|
||||
func (m *mockDockerClient) VolumeRemove(ctx context.Context, id string, opts mobyclient.VolumeRemoveOptions) (mobyclient.VolumeRemoveResult, error) {
|
||||
args := m.Called(ctx, id, opts)
|
||||
return args.Get(0).(mobyclient.VolumeRemoveResult), args.Error(1)
|
||||
}
|
||||
|
||||
// unix socket paths are limited to about a hundred bytes, which the default macOS TMPDIR exceeds
|
||||
func shortTempDir(t testing.TB) string {
|
||||
t.Helper()
|
||||
t.Setenv("TMPDIR", "/tmp")
|
||||
return t.TempDir()
|
||||
}
|
||||
|
||||
func daemonSocketPath(t testing.TB, cli mobyclient.APIClient) string {
|
||||
t.Helper()
|
||||
host := cli.DaemonHost()
|
||||
if !strings.HasPrefix(host, "unix://") {
|
||||
t.Skipf("skipping: daemon at %s is not a unix socket", host)
|
||||
}
|
||||
return strings.TrimPrefix(host, "unix://")
|
||||
}
|
||||
|
||||
func TestDockerProxy(t *testing.T) {
|
||||
bodies := make(chan []byte, 8)
|
||||
daemonSocket := filepath.Join(shortTempDir(t), "d.sock")
|
||||
listener, err := net.Listen("unix", daemonSocket)
|
||||
require.NoError(t, err)
|
||||
daemon := &http.Server{ReadHeaderTimeout: time.Second, Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case strings.HasSuffix(r.URL.Path, "/create"):
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
bodies <- body
|
||||
w.WriteHeader(http.StatusCreated)
|
||||
case strings.HasSuffix(r.URL.Path, "/start"):
|
||||
conn, buffered, err := w.(http.Hijacker).Hijack()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer conn.Close()
|
||||
_, _ = conn.Write([]byte("HTTP/1.1 200 OK\r\nContent-Type: application/vnd.docker.raw-stream\r\n\r\n"))
|
||||
_, _ = io.Copy(conn, buffered)
|
||||
default:
|
||||
w.Header().Set("Api-Version", "1.47")
|
||||
_, _ = w.Write([]byte("OK " + r.Method + " " + r.URL.Path))
|
||||
}
|
||||
})}
|
||||
go func() { _ = daemon.Serve(listener) }()
|
||||
t.Cleanup(func() { _ = daemon.Close() })
|
||||
proxy, err := StartDockerProxy(daemonSocket, shortTempDir(t), "job-1")
|
||||
require.NoError(t, err)
|
||||
client := &http.Client{Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
|
||||
return (&net.Dialer{}).DialContext(ctx, "unix", proxy.Socket)
|
||||
}}}
|
||||
|
||||
t.Run("labels creates", func(t *testing.T) {
|
||||
for path, body := range map[string]string{
|
||||
"/v1.47/containers/create": `{"Image":"alpine","Labels":{"own":"1"}}`,
|
||||
"/networks/create": `{"Name":"n"}`,
|
||||
"/volumes/create": `{"Name":"v","labels":null}`,
|
||||
} {
|
||||
resp, err := client.Post("http://docker"+path, "application/json", strings.NewReader(body))
|
||||
require.NoError(t, err)
|
||||
resp.Body.Close()
|
||||
assert.Equal(t, http.StatusCreated, resp.StatusCode)
|
||||
|
||||
var got map[string]json.RawMessage
|
||||
require.NoError(t, json.Unmarshal(<-bodies, &got))
|
||||
key := "Labels"
|
||||
if _, ok := got["labels"]; ok {
|
||||
key = "labels"
|
||||
}
|
||||
var labels map[string]string
|
||||
require.NoError(t, json.Unmarshal(got[key], &labels))
|
||||
assert.Equal(t, "job-1", labels[jobLabel], path)
|
||||
if strings.Contains(body, "own") {
|
||||
assert.Equal(t, "1", labels["own"])
|
||||
assert.JSONEq(t, `"alpine"`, string(got["Image"]))
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("passes other requests through", func(t *testing.T) {
|
||||
resp, err := client.Get("http://docker/v1.47/_ping")
|
||||
require.NoError(t, err)
|
||||
defer resp.Body.Close()
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
assert.Equal(t, "1.47", resp.Header.Get("Api-Version"))
|
||||
assert.Equal(t, "OK GET /v1.47/_ping", string(body))
|
||||
})
|
||||
|
||||
t.Run("tunnels raw streams", func(t *testing.T) {
|
||||
conn, err := net.Dial("unix", proxy.Socket)
|
||||
require.NoError(t, err)
|
||||
defer conn.Close()
|
||||
_, err = conn.Write([]byte("POST /v1.47/exec/abc/start HTTP/1.1\r\nHost: docker\r\nContent-Length: 2\r\n\r\n{}ping\n"))
|
||||
require.NoError(t, err)
|
||||
reader := bufio.NewReader(conn)
|
||||
resp, err := http.ReadResponse(reader, nil)
|
||||
require.NoError(t, err)
|
||||
defer resp.Body.Close()
|
||||
assert.Equal(t, "application/vnd.docker.raw-stream", resp.Header.Get("Content-Type"))
|
||||
echoed, err := reader.ReadString('\n')
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "{}ping\n", echoed)
|
||||
})
|
||||
}
|
||||
|
||||
func TestRemoveLabelledRemovesContainersNetworksAndVolumes(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
filters := make(mobyclient.Filters).Add("label", jobLabel+"=job-1")
|
||||
cli := &mockDockerClient{}
|
||||
cli.On("ContainerList", ctx, mobyclient.ContainerListOptions{All: true, Filters: filters}).
|
||||
Return(mobyclient.ContainerListResult{Items: []container.Summary{{ID: "c1", Names: []string{"/app"}}}}, nil)
|
||||
cli.On("ContainerKill", ctx, "c1", mock.Anything).Return(mobyclient.ContainerKillResult{}, nil)
|
||||
cli.On("ContainerRemove", ctx, "c1", mobyclient.ContainerRemoveOptions{RemoveVolumes: true, Force: true}).
|
||||
Return(mobyclient.ContainerRemoveResult{}, nil)
|
||||
cli.On("NetworkList", ctx, mobyclient.NetworkListOptions{Filters: filters}).
|
||||
Return(mobyclient.NetworkListResult{Items: []network.Summary{{ID: "n1", Name: "app_default"}}}, nil)
|
||||
cli.On("NetworkRemove", ctx, "n1", mock.Anything).Return(mobyclient.NetworkRemoveResult{}, nil)
|
||||
cli.On("VolumeList", ctx, mobyclient.VolumeListOptions{Filters: filters}).
|
||||
Return(mobyclient.VolumeListResult{Items: []volume.Volume{{Name: "app_data"}}}, nil)
|
||||
cli.On("VolumeRemove", ctx, "app_data", mobyclient.VolumeRemoveOptions{}).Return(mobyclient.VolumeRemoveResult{}, nil)
|
||||
|
||||
require.NoError(t, removeLabelled(ctx, cli, "job-1"))
|
||||
cli.AssertExpectations(t)
|
||||
}
|
||||
|
||||
func TestDockerProxyWithDaemon(t *testing.T) {
|
||||
requireDocker(t)
|
||||
ctx := context.Background()
|
||||
require.NoError(t, NewDockerPullExecutor(NewDockerPullExecutorInput{Image: "alpine"})(ctx))
|
||||
direct, err := GetDockerClient(ctx)
|
||||
require.NoError(t, err)
|
||||
defer direct.Close()
|
||||
dir := shortTempDir(t)
|
||||
seen, err := daemonSeesDir(ctx, dir)
|
||||
require.NoError(t, err)
|
||||
t.Logf("daemon sees the runner's filesystem: %v", seen)
|
||||
|
||||
job := "proxy-test-" + t.Name()
|
||||
proxy, err := StartDockerProxy(daemonSocketPath(t, direct), dir, job)
|
||||
require.NoError(t, err)
|
||||
viaProxy, err := mobyclient.New(mobyclient.WithHost("unix://" + proxy.Socket))
|
||||
require.NoError(t, err)
|
||||
defer viaProxy.Close()
|
||||
|
||||
created, err := viaProxy.ContainerCreate(ctx, mobyclient.ContainerCreateOptions{
|
||||
Config: &container.Config{Image: "alpine", Cmd: []string{"sleep", "300"}, Labels: map[string]string{"own": "1"}},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _, _ = direct.ContainerRemove(ctx, created.ID, mobyclient.ContainerRemoveOptions{Force: true}) })
|
||||
_, err = viaProxy.ContainerStart(ctx, created.ID, mobyclient.ContainerStartOptions{})
|
||||
require.NoError(t, err)
|
||||
net, err := viaProxy.NetworkCreate(ctx, job, mobyclient.NetworkCreateOptions{})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _, _ = direct.NetworkRemove(ctx, net.ID, mobyclient.NetworkRemoveOptions{}) })
|
||||
_, err = viaProxy.VolumeCreate(ctx, mobyclient.VolumeCreateOptions{Name: job})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _, _ = direct.VolumeRemove(ctx, job, mobyclient.VolumeRemoveOptions{Force: true}) })
|
||||
|
||||
inspected, err := direct.ContainerInspect(ctx, created.ID, mobyclient.ContainerInspectOptions{})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, map[string]string{"own": "1", jobLabel: job}, inspected.Container.Config.Labels)
|
||||
vol, err := direct.VolumeInspect(ctx, job, mobyclient.VolumeInspectOptions{})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, job, vol.Volume.Labels[jobLabel])
|
||||
|
||||
exec, err := viaProxy.ExecCreate(ctx, created.ID, mobyclient.ExecCreateOptions{Cmd: []string{"cat"}, AttachStdin: true, AttachStdout: true, TTY: true})
|
||||
require.NoError(t, err)
|
||||
attached, err := viaProxy.ExecAttach(ctx, exec.ID, mobyclient.ExecAttachOptions{TTY: true})
|
||||
require.NoError(t, err)
|
||||
_, err = attached.Conn.Write([]byte("hello\n"))
|
||||
require.NoError(t, err)
|
||||
echoed, err := attached.Reader.ReadString('\n')
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "hello\r\n", echoed)
|
||||
attached.Close()
|
||||
|
||||
require.NoError(t, proxy.Close(ctx))
|
||||
_, err = direct.ContainerInspect(ctx, created.ID, mobyclient.ContainerInspectOptions{})
|
||||
assert.True(t, cerrdefs.IsNotFound(err))
|
||||
_, err = direct.NetworkInspect(ctx, net.ID, mobyclient.NetworkInspectOptions{})
|
||||
assert.True(t, cerrdefs.IsNotFound(err))
|
||||
_, err = direct.VolumeInspect(ctx, job, mobyclient.VolumeInspectOptions{})
|
||||
assert.True(t, cerrdefs.IsNotFound(err))
|
||||
}
|
||||
|
||||
func BenchmarkDockerProxy(b *testing.B) {
|
||||
ctx := context.Background()
|
||||
direct, err := GetDockerClient(ctx)
|
||||
require.NoError(b, err)
|
||||
defer direct.Close()
|
||||
if _, err := direct.Ping(ctx, mobyclient.PingOptions{}); err != nil {
|
||||
b.Skipf("docker daemon unreachable: %v", err)
|
||||
}
|
||||
proxy, err := StartDockerProxy(daemonSocketPath(b, direct), shortTempDir(b), "bench")
|
||||
require.NoError(b, err)
|
||||
defer func() { _ = proxy.Close(ctx) }()
|
||||
viaProxy, err := mobyclient.New(mobyclient.WithHost("unix://" + proxy.Socket))
|
||||
require.NoError(b, err)
|
||||
defer viaProxy.Close()
|
||||
|
||||
require.NoError(b, NewDockerPullExecutor(NewDockerPullExecutorInput{Image: "alpine"})(ctx))
|
||||
created, err := direct.ContainerCreate(ctx, mobyclient.ContainerCreateOptions{Config: &container.Config{Image: "alpine", Cmd: []string{"sleep", "600"}}})
|
||||
require.NoError(b, err)
|
||||
defer func() { _, _ = direct.ContainerRemove(ctx, created.ID, mobyclient.ContainerRemoveOptions{Force: true}) }()
|
||||
_, err = direct.ContainerStart(ctx, created.ID, mobyclient.ContainerStartOptions{})
|
||||
require.NoError(b, err)
|
||||
|
||||
var archive bytes.Buffer
|
||||
writer := tar.NewWriter(&archive)
|
||||
payload := make([]byte, 64<<20)
|
||||
require.NoError(b, writer.WriteHeader(&tar.Header{Name: "blob", Mode: 0o600, Size: int64(len(payload))}))
|
||||
_, _ = writer.Write(payload)
|
||||
require.NoError(b, writer.Close())
|
||||
|
||||
for name, cli := range map[string]mobyclient.APIClient{"direct": direct, "proxy": viaProxy} {
|
||||
b.Run("ping/"+name, func(b *testing.B) {
|
||||
for b.Loop() {
|
||||
if _, err := cli.Ping(ctx, mobyclient.PingOptions{}); err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
}
|
||||
})
|
||||
b.Run("copy64MiB/"+name, func(b *testing.B) {
|
||||
b.SetBytes(int64(archive.Len()))
|
||||
for b.Loop() {
|
||||
_, err := cli.CopyToContainer(ctx, created.ID, mobyclient.CopyToContainerOptions{DestinationPath: "/tmp", Content: bytes.NewReader(archive.Bytes())})
|
||||
if err != nil {
|
||||
b.Fatal(err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -252,6 +252,10 @@ func containerInfoFromInspect(inspect container.InspectResponse) *Info {
|
||||
ID: inspect.ID,
|
||||
Health: HealthNone,
|
||||
Ports: map[string]string{}, // an empty map, never null, in the expression context
|
||||
Mounts: make(map[string]string, len(inspect.Mounts)),
|
||||
}
|
||||
for _, mountPoint := range inspect.Mounts {
|
||||
info.Mounts[mountPoint.Destination] = mountPoint.Source
|
||||
}
|
||||
|
||||
if state := inspect.State; state != nil {
|
||||
|
||||
@@ -801,6 +801,20 @@ func TestContainerInfoFromInspect(t *testing.T) {
|
||||
assert.Equal(t, "abc123", info.ID)
|
||||
assert.Equal(t, HealthNone, info.Health)
|
||||
})
|
||||
|
||||
t.Run("reports the mount sources by container path", func(t *testing.T) {
|
||||
info := containerInfoFromInspect(container.InspectResponse{
|
||||
Mounts: []container.MountPoint{
|
||||
{Type: mount.TypeVolume, Name: "job", Source: "/var/lib/docker/volumes/job/_data", Destination: "/workspace/owner/repo"},
|
||||
{Type: mount.TypeBind, Source: "/var/run/docker.sock", Destination: "/var/run/docker.sock"},
|
||||
},
|
||||
})
|
||||
|
||||
assert.Equal(t, map[string]string{
|
||||
"/workspace/owner/repo": "/var/lib/docker/volumes/job/_data",
|
||||
"/var/run/docker.sock": "/var/run/docker.sock",
|
||||
}, info.Mounts)
|
||||
})
|
||||
}
|
||||
|
||||
func TestMergeContainerConfigsVolumesReplaceRunnerMounts(t *testing.T) {
|
||||
|
||||
@@ -29,6 +29,14 @@ func RemoveImage(ctx context.Context, imageName string, force, pruneChildren boo
|
||||
return false, errors.New("Unsupported Operation")
|
||||
}
|
||||
|
||||
func DockerProxyDir(ctx context.Context) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func StartDockerProxy(daemonSocket, dir, job string) (*DockerProxy, error) {
|
||||
return nil, errors.New("Unsupported Operation")
|
||||
}
|
||||
|
||||
// NewDockerBuildExecutor function to create a run executor for the container
|
||||
func NewDockerBuildExecutor(input NewDockerBuildExecutorInput) common.Executor {
|
||||
return func(ctx context.Context) error {
|
||||
|
||||
@@ -96,6 +96,7 @@ type RunContext struct {
|
||||
jobContainerID string
|
||||
hasBash *bool // memoized implicit-shell probe, only set on the top-level RunContext
|
||||
jobNetworkName string
|
||||
dockerProxy *container.DockerProxy
|
||||
// stepEnv is a copy of the running step's environment, so that workflow commands parsed out
|
||||
// of the container's output can be judged against it. Written by runStepExecutor and read on
|
||||
// the log-writer goroutine, hence unsecureCommandMu, which also guards unsecureCommandErr.
|
||||
@@ -253,11 +254,38 @@ func (rc *RunContext) validVolumes() []string {
|
||||
if rc.Config.BindWorkdir {
|
||||
volumes = append(volumes, rc.Config.Workdir)
|
||||
}
|
||||
if rc.dockerProxy != nil {
|
||||
volumes = append(volumes, rc.dockerProxy.Socket)
|
||||
}
|
||||
// TODO: add a new configuration to control whether the docker daemon can be mounted
|
||||
return append(volumes, name, name+"-env",
|
||||
getDockerDaemonSocketMountPath(rc.containerDaemonSocket()))
|
||||
}
|
||||
|
||||
func (rc *RunContext) jobDockerSocket() string {
|
||||
if rc.dockerProxy != nil {
|
||||
return rc.dockerProxy.Socket
|
||||
}
|
||||
return getDockerDaemonSocketMountPath(rc.containerDaemonSocket())
|
||||
}
|
||||
|
||||
func (rc *RunContext) startDockerProxy(ctx context.Context) {
|
||||
daemonSocket := rc.containerDaemonSocket()
|
||||
if daemonSocket == "-" || strings.HasPrefix(strings.ToLower(daemonSocket), "npipe://") {
|
||||
return
|
||||
}
|
||||
dir := container.DockerProxyDir(ctx)
|
||||
if dir == "" {
|
||||
return
|
||||
}
|
||||
proxy, err := container.StartDockerProxy(getDockerDaemonSocketMountPath(daemonSocket), dir, rc.jobContainerName())
|
||||
if err != nil {
|
||||
common.Logger(ctx).Warnf("docker proxy not started, the job gets the daemon socket directly: %v", err)
|
||||
return
|
||||
}
|
||||
rc.dockerProxy = proxy
|
||||
}
|
||||
|
||||
// toolCache returns the tool cache path the job sees, relocatable through RUNNER_TOOL_CACHE.
|
||||
func (rc *RunContext) toolCache(fallback string) string {
|
||||
if path := rc.GetEnv()["RUNNER_TOOL_CACHE"]; path != "" {
|
||||
@@ -327,8 +355,8 @@ func (rc *RunContext) GetBindsAndMounts() ([]string, map[string]string, error) {
|
||||
// the runner's own mounts below yield to the targets the job claims
|
||||
binds, mounts, claimed := splitVolumes(volumes)
|
||||
|
||||
if daemonSocket := rc.containerDaemonSocket(); daemonSocket != "-" && !claimed["/var/run/docker.sock"] {
|
||||
binds = append(binds, getDockerDaemonSocketMountPath(daemonSocket)+":/var/run/docker.sock")
|
||||
if rc.containerDaemonSocket() != "-" && !claimed["/var/run/docker.sock"] {
|
||||
binds = append(binds, rc.jobDockerSocket()+":/var/run/docker.sock")
|
||||
}
|
||||
if rc.Config.SharedToolCache {
|
||||
if toolCache := rc.toolCache(container.DefaultToolCache); !claimed[toolCache] {
|
||||
@@ -455,6 +483,7 @@ func (rc *RunContext) startJobContainer() common.Executor {
|
||||
// For gitea, to support --volumes-from <container_name_or_id> in options.
|
||||
// We need to set the container name to the environment variable.
|
||||
rc.Env["JOB_CONTAINER_NAME"] = name
|
||||
rc.startDockerProxy(ctx)
|
||||
|
||||
envList := make([]string, 0)
|
||||
|
||||
@@ -651,6 +680,12 @@ func (rc *RunContext) cleanupJobResources(networkName string, createAndDeleteNet
|
||||
logger.Errorf("Error while cleaning services: %v", err)
|
||||
}
|
||||
}
|
||||
if rc.dockerProxy != nil {
|
||||
if err := rc.dockerProxy.Close(ctx); err != nil {
|
||||
logger.Errorf("Error while removing what the job created: %v", err)
|
||||
}
|
||||
rc.dockerProxy = nil
|
||||
}
|
||||
if removeJobContainer {
|
||||
// after the containers using them, services can hold these via `--volumes-from`
|
||||
name := rc.jobContainerName()
|
||||
@@ -914,6 +949,9 @@ func (rc *RunContext) captureJobContainerInfo() common.Executor {
|
||||
return nil
|
||||
}
|
||||
rc.jobContainerID = info.ID
|
||||
if source := info.Mounts[rc.githubWorkspace()]; source != "" {
|
||||
rc.Env["GITEA_DOCKER_WORKSPACE"] = source
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1294,6 +1294,29 @@ func TestGetJobContextReportsContainers(t *testing.T) {
|
||||
}, jobContext.Services)
|
||||
}
|
||||
|
||||
func TestCaptureJobContainerInfoExportsDockerWorkspace(t *testing.T) {
|
||||
job := &containerMock{}
|
||||
job.On("Inspect", mock.Anything).Return(&container.Info{
|
||||
ID: "job-container-id",
|
||||
Mounts: map[string]string{"/workspace/owner/repo": "/var/lib/docker/volumes/job/_data"},
|
||||
}, nil)
|
||||
rc := &RunContext{
|
||||
Config: &Config{Workdir: "/workspace/owner/repo/"},
|
||||
Env: map[string]string{},
|
||||
JobContainer: job,
|
||||
}
|
||||
|
||||
require.NoError(t, rc.captureJobContainerInfo()(context.Background()))
|
||||
|
||||
assert.Equal(t, "job-container-id", rc.jobContainerID)
|
||||
assert.Equal(t, "/var/lib/docker/volumes/job/_data", rc.Env["GITEA_DOCKER_WORKSPACE"])
|
||||
|
||||
rc.Config.Workdir = "/elsewhere"
|
||||
rc.Env = map[string]string{}
|
||||
require.NoError(t, rc.captureJobContainerInfo()(context.Background()))
|
||||
assert.NotContains(t, rc.Env, "GITEA_DOCKER_WORKSPACE")
|
||||
}
|
||||
|
||||
// A job that never started a container reports an empty context, not a placeholder.
|
||||
func TestGetJobContextWithoutContainer(t *testing.T) {
|
||||
jobContext := (&RunContext{}).getJobContext()
|
||||
|
||||
@@ -256,11 +256,10 @@ container:
|
||||
#require_docker: false
|
||||
# Timeout to wait for the docker daemon to be reachable, if docker is required by require_docker or runner
|
||||
#docker_timeout: 0s
|
||||
# Bind the workspace to the host filesystem instead of using Docker volumes.
|
||||
# This is required for Docker-in-Docker (DinD) setups when jobs use docker compose
|
||||
# with bind mounts (e.g., ".:/app"), as volume-based workspaces are not accessible
|
||||
# from the DinD daemon's filesystem. When enabled, ensure the workspace parent
|
||||
# directory is also mounted into the runner container.
|
||||
# Mount the workspace from a host directory instead of a Docker volume, so jobs
|
||||
# can bind-mount it by its own path into sibling containers (".:/app" in docker
|
||||
# compose). Not needed when workflows use the GITEA_DOCKER_WORKSPACE variable.
|
||||
# The workspace parent directory must be mounted into the runner container.
|
||||
#bind_workdir: false
|
||||
# How long a job waits for a service container that declares a healthcheck to become
|
||||
# healthy. A negative value (e.g. -1s) starts the steps without waiting.
|
||||
|
||||
@@ -157,7 +157,7 @@ type Container struct {
|
||||
ForceRebuild bool `yaml:"force_rebuild"` // Rebuild docker image(s) even if already present
|
||||
RequireDocker bool `yaml:"require_docker"` // Always require a reachable docker daemon, even if not required by runner
|
||||
DockerTimeout time.Duration `yaml:"docker_timeout"` // Timeout to wait for the docker daemon to be reachable, if docker is required by require_docker or runner
|
||||
BindWorkdir bool `yaml:"bind_workdir"` // BindWorkdir binds the workspace to the host filesystem instead of using Docker volumes. Required for DinD when jobs use docker compose with bind mounts.
|
||||
BindWorkdir bool `yaml:"bind_workdir"` // BindWorkdir mounts the workspace from a host directory instead of a Docker volume.
|
||||
ServiceReadyTimeout time.Duration `yaml:"service_ready_timeout"` // ServiceReadyTimeout bounds how long a job waits for a service container that declares a healthcheck to report healthy. Negative disables waiting.
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user