From ace8653105a18f3055d2658ea01b027fd8221626 Mon Sep 17 00:00:00 2001 From: silverwind <2021+silverwind@noreply.gitea.com> Date: Thu, 24 Sep 2026 18:56:07 +0000 Subject: [PATCH] feat!: serve the actions cache from the runner (#1229) Make cache requests go through the runner and remove outbound traffic from the cache server. The runner forwards them to `cache.external_server` when one is set, and attaches itself to each job network so jobs can reach it. Benefits: 1. Jobs need no http proxy rules for the cache server, which `@actions/http-client` mishandles for CIDR ranges in `no_proxy` (https://github.com/actions/toolkit/pull/2495). 2. Only runners reach a shared cache server, which needs no route to Gitea. 3. A runner container sharing the host Docker socket now caches, where jobs previously could not reach it. Breaks: 1. Jobs that reach `external_server` but not their runner, such as a remote `DOCKER_HOST`. The built-in cache has always needed that reachability, so both modes now agree. Set `cache.host` and `cache.port` to an address those jobs reach. 2. Jobs on `container.network: bridge`, where Docker rejects the alias, so the runner cannot attach and their cache fails as it does today. Leave `container.network` empty, or name a user-defined network. *Written by Claude.* --------- Co-authored-by: bircni Reviewed-on: https://gitea.com/gitea/runner/pulls/1229 Reviewed-by: bircni Co-authored-by: silverwind <2021+silverwind@noreply.gitea.com> --- README.md | 4 +- act/artifactcache/handler.go | 88 ++++++++--- act/artifactcache/handler_test.go | 55 ++++++- act/artifactcache/handler_v2.go | 3 + act/artifactcache/results.go | 8 +- act/artifactcache/results_test.go | 12 +- act/container/docker_network.go | 76 ++++++++-- act/container/docker_network_test.go | 67 +++++++-- act/container/docker_stub.go | 16 +- act/runner/run_context.go | 4 + act/runner/run_context_test.go | 4 +- act/runner/runner.go | 1 + internal/app/run/runner.go | 150 +++++++++---------- internal/app/run/runner_cache_test.go | 133 +++++++--------- internal/app/run/runner_idle_cleanup_test.go | 7 +- internal/app/run/runner_test.go | 36 ++--- internal/pkg/config/config.example.yaml | 16 +- 17 files changed, 428 insertions(+), 252 deletions(-) diff --git a/README.md b/README.md index 6b24473a..d0140a4c 100644 --- a/README.md +++ b/README.md @@ -312,7 +312,7 @@ Use `uses: builtin:checkout` to check out a repository without downloading `acti #### Caching (`actions/cache`) -Each runner starts its own cache server, so runners do not share cached entries. When the runner itself runs in Docker, set `cache.host` to an address job containers can reach and `cache.port` to a fixed published port, or put jobs on a shared `container.network`. +Each runner starts its own cache server, so runners do not share cached entries. A runner in a container on the jobs' Docker daemon attaches itself to each job network, so jobs reach its cache without extra setup. **Sharing a cache between runners** @@ -337,7 +337,7 @@ cache: external_secret: "" ``` -Jobs connect to `external_server` too, so point it at the reverse proxy if one fronts the server. `--dir`, `--host` and `--port` override the matching `cache` keys. Eviction settings take effect on the cache server, not on the runners. +`--dir`, `--host` and `--port` override the matching `cache` keys. Eviction settings take effect on the cache server, not on the runners. Runners can also share one `cache.dir` on a file system with working file locks, at the cost of slower cache requests. diff --git a/act/artifactcache/handler.go b/act/artifactcache/handler.go index b54884a8..c4c75732 100644 --- a/act/artifactcache/handler.go +++ b/act/artifactcache/handler.go @@ -17,6 +17,7 @@ import ( "io" "net" "net/http" + "net/http/httputil" "net/url" "os" "path/filepath" @@ -57,10 +58,9 @@ type JobCredential struct { Repo string `json:"repo"` // Results is the instance whose artifact service this server forwards for the job, and - // InsecureTLS how the runner reaches it; see results.go. The tags are the wire format a - // remote runner registers with. - Results string `json:"results"` - InsecureTLS bool `json:"insecure_tls"` + // InsecureTLS how the runner reaches it; see results.go. + Results string `json:"-"` + InsecureTLS bool `json:"-"` // PublicURL is this server as a reverse proxy makes the job reach it, not the listen address. PublicURL string `json:"public_url"` @@ -117,6 +117,7 @@ type Options struct { Dir string OutboundIP string Port uint16 + Upstream string // InternalSecret, when non-empty, enables a control-plane API at // /_internal/{register,revoke} that lets a remote runner pre-register the @@ -129,7 +130,7 @@ type Options struct { Logger logrus.FieldLogger } -// StartHandler opens the on-disk cache store and starts the HTTP server. +// StartHandler starts an embedded cache or forwards cache requests to Upstream. func StartHandler(opts Options) (*Handler, error) { dir, logger := opts.Dir, opts.Logger h := &Handler{ @@ -147,6 +148,25 @@ func StartHandler(opts Options) (*Handler, error) { logger = logger.WithField("module", "artifactcache") h.logger = logger + if opts.OutboundIP != "" { + h.outboundIP = opts.OutboundIP + } else if ip := common.GetOutboundIP(); ip == nil { + return nil, errors.New("unable to determine outbound IP address") + } else { + h.outboundIP = ip.String() + } + + if opts.Upstream != "" { + handler, err := h.cacheForwarder(opts.Upstream) + if err != nil { + return nil, err + } + if err := h.serve(opts.Port, handler); err != nil { + return nil, err + } + return h, nil + } + if dir == "" { home, err := os.UserHomeDir() if err != nil { @@ -166,14 +186,6 @@ func StartHandler(opts Options) (*Handler, error) { } h.storage = storage - if opts.OutboundIP != "" { - h.outboundIP = opts.OutboundIP - } else if ip := common.GetOutboundIP(); ip == nil { - return nil, errors.New("unable to determine outbound IP address") - } else { - h.outboundIP = ip.String() - } - secret, err := loadOrCreateSecret(dir) if err != nil { return nil, err @@ -201,34 +213,69 @@ func StartHandler(opts Options) (*Handler, error) { h.gcCache() + if err := h.serve(opts.Port, router); err != nil { + return nil, err + } + return h, nil +} + +func (h *Handler) serve(port uint16, handler http.Handler) error { // Listen on all interfaces. Binding to outboundIP only would give no real // security benefit (it is the LAN/internet-facing address either way) and // can break Docker Desktop variants where the host's outbound IP is not // routable from inside the container network. Authentication is enforced // by the bearer middleware and per-repo scoping, not by reachability. - listener, err := net.Listen("tcp", fmt.Sprintf(":%d", opts.Port)) + listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port)) if err != nil { - return nil, err + return 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()) + return fmt.Errorf("cache server listens on %T, want a TCP address", listener.Addr()) } h.port = addr.Port server := &http.Server{ ReadHeaderTimeout: 2 * time.Second, - Handler: router, + Handler: handler, } go func() { if err := server.Serve(listener); err != nil && errors.Is(err, net.ErrClosed) { - logger.Errorf("http serve: %v", err) + h.logger.Errorf("http serve: %v", err) } }() h.listener = listener h.server = server - return h, nil + return nil +} + +func (h *Handler) cacheForwarder(upstream string) (http.Handler, error) { + target, err := url.Parse(strings.TrimRight(upstream, "/")) + if err != nil { + return nil, fmt.Errorf("parse cache upstream: %w", err) + } + transport, _ := http.DefaultTransport.(*http.Transport) + transport = transport.Clone() + transport.MaxIdleConnsPerHost = 64 // parallel block uploads would otherwise redial past the default 2 + proxy := &httputil.ReverseProxy{ + Transport: transport, + Rewrite: func(r *httputil.ProxyRequest) { + r.SetURL(target) + r.Out.URL.RawQuery = r.In.URL.RawQuery // Rewrite drops unparsable query params, and signed URLs must pass verbatim. + }, + ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) { + h.logger.Warnf("cache forward %s %s: %v", r.Method, r.URL.Path, err) + h.twirpError(w, r, twirpUnavailable, errors.New("cache upstream is unavailable")) + }, + } + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasPrefix(r.URL.Path, apiPath+"/") || strings.HasPrefix(r.URL.Path, cacheServiceV2Path+"/") { + proxy.ServeHTTP(w, r) + } else { + h.forwardOrNotFound(w, r) + } + }), nil } func (h *Handler) ExternalURL() string { @@ -708,8 +755,7 @@ func (h *Handler) internalRegister(w http.ResponseWriter, r *http.Request, _ htt return } h.RegisterJob(body.Token, body.JobCredential) - // A server too old to forward answers without this, which is how the caller knows. - h.responseJSON(w, r, http.StatusOK, map[string]any{"results_url": h.ResultsURL(body.JobCredential)}) + h.responseJSON(w, r, http.StatusOK) } // POST /_internal/revoke diff --git a/act/artifactcache/handler_test.go b/act/artifactcache/handler_test.go index 5e3e8e5f..366c6805 100644 --- a/act/artifactcache/handler_test.go +++ b/act/artifactcache/handler_test.go @@ -12,6 +12,7 @@ import ( "fmt" "io" "net/http" + "net/http/httptest" "os" "path/filepath" "strings" @@ -56,6 +57,55 @@ func TestHandler_ExternalURL(t *testing.T) { assert.Equal(t, "http://[2001:db8::1]:8080", (&Handler{outboundIP: "2001:db8::1", port: 8080}).ExternalURL()) } +func TestHandler_Upstream(t *testing.T) { + var upstream *httptest.Server + upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, http.MethodPut, r.Method) + assert.Equal(t, "/cache%20server"+blobPath+"/%34%32", r.URL.EscapedPath()) + assert.Equal(t, "sig=opaque%2Fvalue&raw=a;b", r.URL.RawQuery) + assert.Equal(t, strings.TrimPrefix(upstream.URL, "http://"), r.Host) + assert.Equal(t, "Bearer unregistered", r.Header.Get("Authorization")) + _, err := io.Copy(w, r.Body) + assert.NoError(t, err) + })) + t.Cleanup(upstream.Close) + dir := filepath.Join(t.TempDir(), "artifactcache") + handler, err := StartHandler(Options{ + Dir: dir, + OutboundIP: "127.0.0.1", + Upstream: upstream.URL + "/cache%20server//", + }) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, handler.Close()) }) + + req, err := http.NewRequestWithContext(t.Context(), http.MethodPut, + handler.ExternalURL()+blobPath+"/%34%32?sig=opaque%2Fvalue&raw=a;b", strings.NewReader("cache content")) + require.NoError(t, err) + req.Header.Set("Authorization", "Bearer unregistered") + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + forwarded, err := io.ReadAll(resp.Body) + resp.Body.Close() + require.NoError(t, err) + assert.Equal(t, "cache content", string(forwarded)) + + resp, err = http.Post(handler.ExternalURL()+internalPath+"/register", "application/json", nil) + require.NoError(t, err) + resp.Body.Close() + assert.Equal(t, http.StatusNotFound, resp.StatusCode) + + upstream.Close() + resp, err = http.Post(handler.ExternalURL()+cacheServiceV2Path+"/CreateCacheEntry", "application/json", strings.NewReader(`{}`)) + require.NoError(t, err) + var failure map[string]string + err = json.UnmarshalRead(resp.Body, &failure) + resp.Body.Close() + require.NoError(t, err) + assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) + assert.Equal(t, map[string]string{"code": "unavailable", "msg": "cache upstream is unavailable"}, failure) + assert.NoDirExists(t, dir) +} + func TestHandler(t *testing.T) { dir := filepath.Join(t.TempDir(), "artifactcache") handler, err := StartHandler(Options{Dir: dir}) @@ -1522,7 +1572,10 @@ func TestHandler_InternalAPI_AuthAndUsage(t *testing.T) { } assert.Equal(t, http.StatusUnauthorized, probe("via-internal-api")) - assert.Equal(t, http.StatusOK, post("/_internal/register", secret, `{"token":"via-internal-api","repo":"owner/repo"}`)) + assert.Equal(t, http.StatusOK, post("/_internal/register", secret, `{"token":"via-internal-api","repo":"owner/repo","public_url":"http://runner.example","results":"https://gitea.example","insecure_tls":true}`)) + cred, ok := handler.lookupCredential("via-internal-api") + require.True(t, ok) + assert.Equal(t, JobCredential{Repo: testRepo, PublicURL: "http://runner.example"}, cred) assert.NotEqual(t, http.StatusUnauthorized, probe("via-internal-api")) assert.Equal(t, http.StatusOK, post("/_internal/revoke", secret, `{"token":"via-internal-api"}`)) assert.Equal(t, http.StatusUnauthorized, probe("via-internal-api")) diff --git a/act/artifactcache/handler_v2.go b/act/artifactcache/handler_v2.go index 1c232b44..5f88c7f9 100644 --- a/act/artifactcache/handler_v2.go +++ b/act/artifactcache/handler_v2.go @@ -40,6 +40,7 @@ const ( blobUploadURLTTL = time.Hour twirpInternal = "internal" + twirpUnavailable = "unavailable" twirpUnauthenticated = "unauthenticated" ) @@ -261,6 +262,8 @@ func (h *Handler) twirpError(w http.ResponseWriter, r *http.Request, code string switch code { case twirpInternal: status = http.StatusInternalServerError + case twirpUnavailable: + status = http.StatusServiceUnavailable case twirpUnauthenticated: status = http.StatusUnauthorized } diff --git a/act/artifactcache/results.go b/act/artifactcache/results.go index fff703d7..9f6418cd 100644 --- a/act/artifactcache/results.go +++ b/act/artifactcache/results.go @@ -17,9 +17,6 @@ import ( // Gitea implements only the artifact half of it. Forwarding that half from here makes this origin // the whole service, so ACTIONS_RESULTS_URL can point at it truthfully, which is what the clients // this runner cannot patch need, docker buildx among them. -// -// The instance to forward to travels with the job registration rather than with configuration, so -// a cache server shared between runners serves each of their instances. const artifactServicePath = "/twirp/github.actions.results.api.v1.ArtifactService/" // forwardOrNotFound is the router's fallback: the artifact service of the instance the job @@ -81,9 +78,6 @@ func (h *Handler) mustProxy(r *http.Request, cred JobCredential, target *url.URL return true } scheme := r.Header.Get("X-Forwarded-Proto") - if base, err := url.Parse(cred.PublicURL); err == nil && base.Scheme != "" { - scheme = base.Scheme - } if scheme == "" { scheme = "http" } @@ -98,4 +92,4 @@ func mergeQuery(target, request string) string { } // insecureTransport is shared, because a transport per request would pool no connections. -var insecureTransport = &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}} //nolint:gosec // the runner reaches its instance on the operator's say-so +var insecureTransport = &http.Transport{Proxy: http.ProxyFromEnvironment, TLSClientConfig: &tls.Config{InsecureSkipVerify: true}} //nolint:gosec // the runner reaches its instance on the operator's say-so diff --git a/act/artifactcache/results_test.go b/act/artifactcache/results_test.go index df7547de..993e7982 100644 --- a/act/artifactcache/results_test.go +++ b/act/artifactcache/results_test.go @@ -54,11 +54,11 @@ func TestFrontResultsService(t *testing.T) { assert.Contains(t, body["msg"], "bearer") assert.NotEmpty(t, body["msg"]) - defer handler.RegisterJob(token, JobCredential{Repo: "owner/repo", Results: gitea.URL, PublicURL: "https://cache.example"})() + defer handler.RegisterJob(token, JobCredential{Repo: "owner/repo", Results: selfSigned.URL, InsecureTLS: true})() status, _ = post(rpc) - require.Equal(t, http.StatusOK, status, "a registered https public URL onto an http instance must proxy, not redirect") - assert.Equal(t, strings.TrimPrefix(gitea.URL, "http://"), gotHost, "Gitea must see the host it mints its URLs from") + require.Equal(t, http.StatusOK, status, "an http job reaching an https instance must proxy") + assert.Equal(t, strings.TrimPrefix(selfSigned.URL, "https://"), gotHost, "Gitea must see the host it mints its URLs from") assert.Empty(t, gotProto, "a forwarded scheme would make an https Gitea mint http URLs") assert.Equal(t, rpc, gotPath) @@ -87,15 +87,15 @@ func TestFrontResultsService(t *testing.T) { }, { name: "a trusted https instance is reached by the job itself", - cred: JobCredential{Results: selfSigned.URL, PublicURL: "https://cache.example"}, want: http.StatusTemporaryRedirect, + cred: JobCredential{Results: selfSigned.URL}, forwardedProto: "https", want: http.StatusTemporaryRedirect, }, { name: "the job does not share this server's disregard for the certificate", - cred: JobCredential{Results: selfSigned.URL, PublicURL: "https://cache.example", InsecureTLS: true}, want: http.StatusOK, + cred: JobCredential{Results: selfSigned.URL, InsecureTLS: true}, forwardedProto: "https", want: http.StatusOK, }, {name: "an http job is not redirected to an https instance", cred: JobCredential{Results: selfSigned.URL}, want: http.StatusInternalServerError}, { - name: "a runner too old to send its public URL leaves the terminator to say so", + name: "a forwarded https scheme onto an http instance is proxied", cred: JobCredential{Results: gitea.URL}, forwardedProto: "https", want: http.StatusOK, }, { diff --git a/act/container/docker_network.go b/act/container/docker_network.go index e6155d38..f61a476b 100644 --- a/act/container/docker_network.go +++ b/act/container/docker_network.go @@ -16,6 +16,8 @@ import ( "gitea.com/gitea/runner/act/common" + cerrdefs "github.com/containerd/errdefs" + "github.com/moby/moby/api/types/network" "github.com/moby/moby/client" ) @@ -33,17 +35,17 @@ const ( // had an endpoint on it at the time. Each one holds a subnet of the daemon's address pool // until it is removed. Networks created after createdBefore are left alone, so a job starting // while this runs cannot lose the network it has created but not yet attached a container to. -func RemoveOrphanNetworks(ctx context.Context, runnerUUID string, createdBefore time.Time) error { +func RemoveOrphanNetworks(ctx context.Context, runnerUUID, detach string, createdBefore time.Time) error { cli, err := GetDockerClient(ctx) if err != nil { return fmt.Errorf("failed to connect to the docker daemon: %w", err) } defer cli.Close() - return removeOrphanNetworks(ctx, cli, runnerUUID, createdBefore) + return removeOrphanNetworks(ctx, cli, runnerUUID, detach, createdBefore) } -func removeOrphanNetworks(ctx context.Context, cli client.APIClient, runnerUUID string, createdBefore time.Time) error { +func removeOrphanNetworks(ctx context.Context, cli client.APIClient, runnerUUID, detach string, createdBefore time.Time) error { networks, err := cli.NetworkList(ctx, client.NetworkListOptions{ Filters: make(client.Filters).Add("label", runnerUUIDLabel+"="+runnerUUID), }) @@ -52,6 +54,7 @@ func removeOrphanNetworks(ctx context.Context, cli client.APIClient, runnerUUID } var errs []error +networks: for _, n := range networks.Items { result, err := cli.NetworkInspect(ctx, n.ID, client.NetworkInspectOptions{}) if err != nil { @@ -60,9 +63,18 @@ func removeOrphanNetworks(ctx context.Context, cli client.APIClient, runnerUUID } // the emptiness check, not the label, is what keeps a live job of another process // sharing this registration safe - if len(result.Network.Containers) != 0 || result.Network.Created.After(createdBefore) { + if len(result.Network.Containers) > 1 || result.Network.Created.After(createdBefore) { continue } + for containerID := range result.Network.Containers { + if detach == "" || !strings.HasPrefix(containerID, detach) { + continue networks + } + if err := disconnectNetwork(ctx, cli, n.ID, containerID); err != nil { + errs = append(errs, fmt.Errorf("failed to disconnect container from network %s: %w", n.Name, err)) + continue networks + } + } if _, err := cli.NetworkRemove(ctx, n.ID, client.NetworkRemoveOptions{}); err != nil { errs = append(errs, fmt.Errorf("failed to remove network %s: %w", n.Name, err)) continue @@ -135,6 +147,50 @@ func isAddressPoolExhausted(err error) bool { strings.Contains(msg, "could not find an available, non-overlapping IPv4 address pool among the defaults") // docker 24 and older } +func NewDockerNetworkConnectExecutor(networkName, container string) common.Executor { + return func(ctx context.Context) error { + cli, err := GetDockerClient(ctx) + if err != nil { + return err + } + defer cli.Close() + if err := connectNetwork(ctx, cli, networkName, container); err != nil { + common.Logger(ctx).Warnf("cannot connect %s to network %s, so jobs on it cannot reach the cache: %v", container, networkName, err) + } + return nil + } +} + +func connectNetwork(ctx context.Context, cli client.APIClient, networkName, container string) error { + _, err := cli.NetworkConnect(ctx, networkName, client.NetworkConnectOptions{ + Container: container, + EndpointConfig: &network.EndpointSettings{Aliases: []string{container}}, + }) + if err != nil && (strings.Contains(err.Error(), "already exists in network") || strings.Contains(err.Error(), "already connected")) { + return nil + } + return err +} + +func NewDockerNetworkDisconnectExecutor(networkName, container string) common.Executor { + return func(ctx context.Context) error { + cli, err := GetDockerClient(ctx) + if err != nil { + return err + } + defer cli.Close() + return disconnectNetwork(ctx, cli, networkName, container) + } +} + +func disconnectNetwork(ctx context.Context, cli client.APIClient, networkName, container string) error { + _, err := cli.NetworkDisconnect(ctx, networkName, client.NetworkDisconnectOptions{Container: container, Force: true}) + if cerrdefs.IsNotFound(err) || (err != nil && strings.Contains(err.Error(), "is not connected")) { + return nil + } + return err +} + func NewDockerNetworkRemoveExecutor(name string) common.Executor { return func(ctx context.Context) error { cli, err := GetDockerClient(ctx) @@ -174,16 +230,17 @@ func NewDockerNetworkRemoveExecutor(name string) common.Executor { } } -func IsolatedNetwork(ctx context.Context, addr netip.Addr, jobNetwork string) (string, error) { +// IsolatedCacheContainer returns the container holding addr, when jobs on jobNetwork cannot reach it there. +func IsolatedCacheContainer(ctx context.Context, addr netip.Addr, jobNetwork string) (string, error) { cli, err := GetDockerClient(ctx) if err != nil { return "", err } defer cli.Close() - return isolatedNetwork(ctx, cli, addr, jobNetwork) + return isolatedCacheContainer(ctx, cli, addr, jobNetwork) } -func isolatedNetwork(ctx context.Context, cli client.APIClient, addr netip.Addr, jobNetwork string) (string, error) { +func isolatedCacheContainer(ctx context.Context, cli client.APIClient, addr netip.Addr, jobNetwork string) (string, error) { if jobNetwork == "host" || strings.HasPrefix(jobNetwork, "container:") { return "", nil } @@ -191,7 +248,7 @@ func isolatedNetwork(ctx context.Context, cli client.APIClient, addr netip.Addr, if err != nil { return "", err } - var holderID string + var holderID, holderContainer string for _, summary := range containers.Items { if summary.NetworkSettings == nil { continue @@ -199,6 +256,7 @@ func isolatedNetwork(ctx context.Context, cli client.APIClient, addr netip.Addr, for _, endpoint := range summary.NetworkSettings.Networks { if endpoint != nil && (endpoint.IPAddress == addr || endpoint.GlobalIPv6Address == addr) { holderID = endpoint.NetworkID + holderContainer = summary.ID[:12] } } } @@ -215,5 +273,5 @@ func isolatedNetwork(ctx context.Context, cli client.APIClient, addr netip.Addr, return "", err } } - return holder.Network.Name, nil + return holderContainer, nil } diff --git a/act/container/docker_network_test.go b/act/container/docker_network_test.go index 04a8e6fb..382f26f9 100644 --- a/act/container/docker_network_test.go +++ b/act/container/docker_network_test.go @@ -18,11 +18,12 @@ import ( "github.com/stretchr/testify/require" ) -func TestIsolatedNetwork(t *testing.T) { +func TestIsolatedCacheContainer(t *testing.T) { ctx := context.Background() cache, lan := netip.MustParseAddr("172.18.0.3"), netip.MustParseAddr("192.168.1.20") client := &mockDockerClient{} client.On("ContainerList", ctx, mobyclient.ContainerListOptions{}).Return(mobyclient.ContainerListResult{Items: []container.Summary{{}, { + ID: "0123456789abcdef", NetworkSettings: &container.NetworkSettingsSummary{Networks: map[string]*network.EndpointSettings{ "compose": {NetworkID: "c0ffee", IPAddress: cache}, "lan": {NetworkID: "beef", IPAddress: lan}, @@ -38,12 +39,12 @@ func TestIsolatedNetwork(t *testing.T) { Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Network: inspected}}, nil) } check := func(addr netip.Addr, jobNetwork, want string) { - got, err := isolatedNetwork(ctx, client, addr, jobNetwork) + got, err := isolatedCacheContainer(ctx, client, addr, jobNetwork) require.NoError(t, err) assert.Equal(t, want, got, jobNetwork) } - check(cache, "", "compose") - check(cache, "jobs", "compose") + check(cache, "", "0123456789ab") + check(cache, "jobs", "0123456789ab") check(cache, "c0f", "") check(cache, "host", "") check(lan, "", "") @@ -56,9 +57,44 @@ func TestIsAddressPoolExhausted(t *testing.T) { assert.False(t, isAddressPoolExhausted(cerrdefs.ErrInvalidArgument.WithMessage("invalid subnet 10.0.0.0/8: it overlaps with an existing network"))) } -// Of this runner's networks, only the ones nothing is attached to and old enough to predate -// any job now starting are the runner's to reclaim. An unexpected NetworkRemove fails the -// test on its own, since testify has no expectation to match it against. +func (m *mockDockerClient) NetworkConnect(ctx context.Context, id string, opts mobyclient.NetworkConnectOptions) (mobyclient.NetworkConnectResult, error) { + return mobyclient.NetworkConnectResult{}, m.Called(ctx, id, opts).Error(0) +} + +func (m *mockDockerClient) NetworkDisconnect(ctx context.Context, id string, opts mobyclient.NetworkDisconnectOptions) (mobyclient.NetworkDisconnectResult, error) { + return mobyclient.NetworkDisconnectResult{}, m.Called(ctx, id, opts).Error(0) +} + +func TestConnectAndDisconnectNetwork(t *testing.T) { + ctx := context.Background() + connect := func(answer error) error { + client := &mockDockerClient{} + client.On("NetworkConnect", ctx, "jobs", mobyclient.NetworkConnectOptions{ + Container: "0123456789ab", + EndpointConfig: &network.EndpointSettings{Aliases: []string{"0123456789ab"}}, + }).Return(answer).Once() + return connectNetwork(ctx, client, "jobs", "0123456789ab") + } + require.NoError(t, connect(nil)) + require.NoError(t, connect(errors.New("endpoint with name runner already exists in network jobs"))) + require.NoError(t, connect(errors.New("container runner already connected to network jobs"))) + require.ErrorIs(t, connect(context.Canceled), context.Canceled) + + disconnect := func(answer error) error { + client := &mockDockerClient{} + client.On("NetworkDisconnect", ctx, "jobs", mobyclient.NetworkDisconnectOptions{ + Container: "0123456789ab", Force: true, + }).Return(answer).Once() + return disconnectNetwork(ctx, client, "jobs", "0123456789ab") + } + require.NoError(t, disconnect(cerrdefs.ErrNotFound)) + require.NoError(t, disconnect(errors.New("container runner is not connected to the network jobs"))) + require.ErrorIs(t, disconnect(context.Canceled), context.Canceled) +} + +// Of this runner's networks, only the ones nothing but its own cache container is attached to +// and old enough to predate any job now starting are the runner's to reclaim. An unexpected +// NetworkRemove fails the test on its own, since testify has no expectation to match it against. func TestRemoveOrphanNetworks(t *testing.T) { ctx := context.Background() cutoff := time.Date(2026, time.April, 29, 20, 0, 0, 0, time.UTC) @@ -69,16 +105,29 @@ func TestRemoveOrphanNetworks(t *testing.T) { {ID: "orphan"}, {ID: "busy"}, {ID: "starting"}, + {ID: "cached"}, + {ID: "shared"}, }}, nil) client.On("NetworkInspect", ctx, "orphan", mobyclient.NetworkInspectOptions{}). Return(mobyclient.NetworkInspectResult{}, nil) client.On("NetworkInspect", ctx, "busy", mobyclient.NetworkInspectOptions{}). Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"c": {}}}}, nil) client.On("NetworkInspect", ctx, "starting", mobyclient.NetworkInspectOptions{}). - Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Network: network.Network{Created: cutoff.Add(time.Second)}}}, nil) + Return(mobyclient.NetworkInspectResult{Network: network.Inspect{ + Network: network.Network{Created: cutoff.Add(time.Second)}, + Containers: map[string]network.EndpointResource{"0123456789abcdef": {}}, + }}, nil) + client.On("NetworkInspect", ctx, "cached", mobyclient.NetworkInspectOptions{}). + Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"0123456789abcdef": {}}}}, nil) + client.On("NetworkInspect", ctx, "shared", mobyclient.NetworkInspectOptions{}). + Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"0123456789abcdef": {}, "job": {}}}}, nil) client.On("NetworkRemove", ctx, "orphan", mobyclient.NetworkRemoveOptions{}). Return(mobyclient.NetworkRemoveResult{}, nil) + client.On("NetworkDisconnect", ctx, "cached", mobyclient.NetworkDisconnectOptions{Container: "0123456789abcdef", Force: true}). + Return(nil).Once() + client.On("NetworkRemove", ctx, "cached", mobyclient.NetworkRemoveOptions{}). + Return(mobyclient.NetworkRemoveResult{}, nil).Once() - require.NoError(t, removeOrphanNetworks(ctx, client, "runner-1", cutoff)) + require.NoError(t, removeOrphanNetworks(ctx, client, "runner-1", "0123456789ab", cutoff)) client.AssertExpectations(t) } diff --git a/act/container/docker_stub.go b/act/container/docker_stub.go index 45fd6962..915c6740 100644 --- a/act/container/docker_stub.go +++ b/act/container/docker_stub.go @@ -18,7 +18,7 @@ import ( "github.com/moby/moby/api/types/system" ) -func IsolatedNetwork(context.Context, netip.Addr, string) (string, error) { +func IsolatedCacheContainer(context.Context, netip.Addr, string) (string, error) { return "", nil } @@ -91,7 +91,19 @@ func NewDockerNetworkRemoveExecutor(name string) common.Executor { } } -func RemoveOrphanNetworks(ctx context.Context, runnerUUID string, createdBefore time.Time) error { +func NewDockerNetworkConnectExecutor(_, _ string) common.Executor { + return func(_ context.Context) error { + return nil + } +} + +func NewDockerNetworkDisconnectExecutor(_, _ string) common.Executor { + return func(_ context.Context) error { + return nil + } +} + +func RemoveOrphanNetworks(_ context.Context, _, _ string, _ time.Time) error { return nil } diff --git a/act/runner/run_context.go b/act/runner/run_context.go index 44cd95a0..2b6e0f6c 100644 --- a/act/runner/run_context.go +++ b/act/runner/run_context.go @@ -610,6 +610,8 @@ func (rc *RunContext) startJobContainer() common.Executor { return common.NewPipelineExecutor( container.NewDockerNetworkCreateExecutor(networkName, rc.Config.ContainerNetworkCreateOptions). IfBool(createAndDeleteNetwork), + container.NewDockerNetworkConnectExecutor(networkName, rc.Config.CacheContainer). + IfBool(rc.Config.CacheContainer != ""), rc.startServiceContainers(), rc.reportUnstartedServices(), func(ctx context.Context) error { return rc.createJobVolumes(ctx, containerInput.Mounts) }, @@ -659,6 +661,8 @@ func (rc *RunContext) cleanupJobResources(networkName string, createAndDeleteNet } if createAndDeleteNetwork { logger.Infof("Cleaning up network for job %s, and network name is: %s", rc.JobName, networkName) + errs = append(errs, container.NewDockerNetworkDisconnectExecutor(networkName, rc.Config.CacheContainer). + IfBool(rc.Config.CacheContainer != "")(ctx)) errs = append(errs, container.NewDockerNetworkRemoveExecutor(networkName)(ctx)) } return errors.Join(errs...) diff --git a/act/runner/run_context_test.go b/act/runner/run_context_test.go index 9abd8651..405ad29a 100644 --- a/act/runner/run_context_test.go +++ b/act/runner/run_context_test.go @@ -743,7 +743,7 @@ func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) { service.On("Remove").Return(func(context.Context) error { return removeError }).Once() service.On("Close").Return(func(context.Context) error { return closeError }).Once() rc := &RunContext{ - Config: &Config{}, + Config: &Config{CacheContainer: "cache-container"}, Run: &model.Run{Workflow: &model.Workflow{Name: "wf"}, JobID: "job"}, JobContainer: job, serviceContainers: []*serviceContainer{{name: "svc", container: service}}, @@ -794,7 +794,7 @@ func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) { if name == "canceled" { require.ErrorIs(t, err, context.Canceled) } else { - for _, operation := range []string{"DELETE /volumes/" + rc.jobContainerName(), "DELETE /volumes/" + rc.jobContainerName() + "-env", "GET /networks"} { + for _, operation := range []string{"DELETE /volumes/" + rc.jobContainerName(), "DELETE /volumes/" + rc.jobContainerName() + "-env", "POST /networks/job-network/disconnect", "GET /networks"} { require.ErrorContains(t, err, operation) } assert.Equal(t, 2, volumeRemovals) diff --git a/act/runner/runner.go b/act/runner/runner.go index 11a88d8d..a2b8870d 100644 --- a/act/runner/runner.go +++ b/act/runner/runner.go @@ -57,6 +57,7 @@ type Config struct { DisableActEnv bool // do not inject the ACT=true environment variable into jobs ContainerNetworkMode docker_container.NetworkMode // the network mode of job containers (the value of --network) ContainerNetworkCreateOptions container.NewDockerNetworkCreateExecutorInput // the default network create options + CacheContainer string // let jobs reach an otherwise isolated cache ProxyEnv map[string]string // the proxy variables the job runs with, also given to service containers and image builds NoActionPatch bool // run actions exactly as published, applying no compatibility patches, see patch_actions.go diff --git a/internal/app/run/runner.go b/internal/app/run/runner.go index 2d4a2d37..97aabd43 100644 --- a/internal/app/run/runner.go +++ b/internal/app/run/runner.go @@ -11,6 +11,7 @@ import ( "errors" "fmt" "maps" + "net" "net/http" "net/netip" "net/url" @@ -71,7 +72,7 @@ type Runner struct { cacheHandler *artifactcache.Handler capabilities string - isolatedCacheNetwork func() string + isolatedCacheContainer func() string runningTasks sync.Map runningCount atomic.Int64 @@ -99,24 +100,22 @@ func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client) if cacheEnabled(cfg) { if cfg.Cache.ExternalServer != "" { warnIgnoredCachePolicy(cfg) - // The v1 client appends its path to this without a separator, so the slash is required. - envs["ACTIONS_CACHE_URL"] = strings.TrimRight(cfg.Cache.ExternalServer, "/") + "/" + } + warnIgnoredCacheSecret(cfg) + handler, err := artifactcache.StartHandler(artifactcache.Options{ + Dir: cfg.Cache.Dir, + OutboundIP: cfg.Cache.Host, + Port: cfg.Cache.Port, + Upstream: cfg.Cache.ExternalServer, + Policy: CachePolicy(cfg), + Logger: log.StandardLogger().WithField("module", "cache_request"), + }) + if err != nil { + log.Errorf("cannot init cache server, it will be disabled: %v", err) + // go on } else { - warnIgnoredCacheSecret(cfg) - handler, err := artifactcache.StartHandler(artifactcache.Options{ - Dir: cfg.Cache.Dir, - OutboundIP: cfg.Cache.Host, - Port: cfg.Cache.Port, - Policy: CachePolicy(cfg), - Logger: log.StandardLogger().WithField("module", "cache_request"), - }) - if err != nil { - log.Errorf("cannot init cache server, it will be disabled: %v", err) - // go on - } else { - cacheHandler = handler - envs["ACTIONS_CACHE_URL"] = handler.ExternalURL() + "/" - } + cacheHandler = handler + envs["ACTIONS_CACHE_URL"] = handler.ExternalURL() + "/" } } @@ -140,7 +139,7 @@ func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client) now: time.Now, runHealthCheck: executeHealthCheck, } - runner.isolatedCacheNetwork = sync.OnceValue(runner.detectIsolatedCacheNetwork) + runner.isolatedCacheContainer = sync.OnceValue(runner.detectIsolatedCacheContainer) return runner } @@ -183,7 +182,7 @@ func (r *Runner) cleanupOrphanDockerResources(ctx context.Context) { return } cutoff := r.now().Add(-r.cfg.Runner.WorkdirCleanupAge) - if err := removeOrphanNetworks(ctx, r.uuid, cutoff); err != nil { + if err := removeOrphanNetworks(ctx, r.uuid, r.isolatedCacheContainer(), cutoff); err != nil { log.Warnf("failed to clean up networks left behind by earlier jobs: %v", err) } if err := removeOrphanJobVolumes(ctx, r.uuid, cutoff); err != nil { @@ -418,11 +417,24 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report. taskContext := task.Context.Fields envs := r.cloneEnvs() + // act asks for the platform once per step, so resolve the fallback at most once per task. + fallbackPlatform := sync.OnceValue(func() string { return r.fallbackPlatform(ctx) }) + platformPicker := func(runsOn []string) string { + if platform := r.labels.PickPlatform(runsOn); platform != "" { + return platform + } + return fallbackPlatform() + } + cacheURL, cacheContainer := r.cacheForJob(job, platformPicker) + if cacheURL != "" { + envs["ACTIONS_CACHE_URL"] = cacheURL + "/" + } + // Added per task because this job's service containers must be reached directly, and // act reaches them by their workflow key. var services map[string]yaml.Node _ = job.RawServices.Decode(&services) // a whole-value expression names no service before evaluation - proxyEnv := JobProxyEnv(envs, r.builtInCacheURL(), slices.Sorted(maps.Keys(services))) + proxyEnv := JobProxyEnv(envs, cacheURL, slices.Sorted(maps.Keys(services))) maps.Copy(envs, proxyEnv) if r.capabilities != "" { @@ -467,10 +479,8 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report. // cache requests from the job container can authenticate. The credential // is removed when the task finishes, so a leaked token stops working as // soon as the job ends rather than remaining valid for the runner's - // lifetime. Only applies to the embedded cache server; when the operator - // points the runner at an external cache via cfg.Cache.ExternalServer, it - // is that server's responsibility to authenticate requests. - revokeCache, resultsURL := r.registerCacheForTask(giteaRuntimeToken, preset.Repository, reporter) + // lifetime. + revokeCache, resultsURL := r.registerCacheForTask(giteaRuntimeToken, preset.Repository, cacheURL, reporter) defer revokeCache() eventJSON, err := json.Marshal(preset.Event) @@ -503,20 +513,6 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report. // Without bind_workdir, the workspace path omits the task id; concurrent host-mode jobs // for the same repository would share this directory and can race with per-job cleanup. - // act asks for the platform once per step, so resolve the fallback at most once per task. - fallbackPlatform := sync.OnceValue(func() string { return r.fallbackPlatform(ctx) }) - platformPicker := func(runsOn []string) string { - if platform := r.labels.PickPlatform(runsOn); platform != "" { - return platform - } - return fallbackPlatform() - } - - if resultsURL != "" && r.cacheIsolatedFrom(job, platformPicker) { - reporter.Logf("::warning::%s", runner.EscapeCommandData(fmt.Sprintf("jobs cannot reach the cache server at %s on docker network %q, so caching fails and artifacts go to Gitea directly, set cache.host and cache.port to an address jobs reach, or container.network to %[2]q", - r.cacheHandler.ExternalURL(), r.isolatedCacheNetwork()))) - resultsURL = "" - } r.setResultsService(envs, resultsURL) runnerConfig := &runner.Config{ @@ -546,6 +542,7 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report. ContainerMaxLifetime: maxLifetime, CleanWorkdir: true, ContainerNetworkMode: docker_container.NetworkMode(r.cfg.Container.Network), + CacheContainer: cacheContainer, ContainerNetworkCreateOptions: container.NewDockerNetworkCreateExecutorInput{ EnableIPv4: r.cfg.Container.NetworkCreateOptions.EnableIPv4, EnableIPv6: r.cfg.Container.NetworkCreateOptions.EnableIPv6, @@ -643,32 +640,24 @@ func (r *Runner) instanceOutOfReach(instance string) bool { // function the caller must invoke (typically via defer) to revoke the // credential when the task finishes. // -// Two modes: -// - Embedded handler: register in-process via RegisterJob. -// - external_server: POST to the remote server's /_internal/register, defer a -// POST to /_internal/revoke. This is what enables full per-job auth and -// repo scoping over the network. -// // Safe with an empty token (older Gitea did not issue one). -// It also returns the URL to advertise as ACTIONS_RESULTS_URL, which is the cache server itself -// when it agreed to forward this instance's artifact service, and "" when it did not. -func (r *Runner) registerCacheForTask(token, repo string, reporter *report.Reporter) (func(), string) { - if token == "" { +// It also returns the URL to advertise as ACTIONS_RESULTS_URL. +func (r *Runner) registerCacheForTask(token, repo, publicURL string, reporter *report.Reporter) (func(), string) { + if token == "" || r.cacheHandler == nil { return func() {}, "" } cred := artifactcache.JobCredential{ Repo: repo, + PublicURL: publicURL, Results: r.envs["ACTIONS_RESULTS_URL"], // the instance as the job would reach it InsecureTLS: r.cfg.Runner.Insecure, } - if r.cacheHandler != nil { - return r.cacheHandler.RegisterJob(token, cred), r.cacheHandler.ResultsURL(cred) + revoke := r.cacheHandler.RegisterJob(token, cred) + if r.cfg.Cache.ExternalServer != "" { + revokeLocal, revokeExternal := revoke, r.registerExternalCacheJob(token, repo, publicURL, reporter) + revoke = func() { revokeLocal(); revokeExternal() } } - if cacheEnabled(r.cfg) && r.cfg.Cache.ExternalServer != "" && r.cfg.Cache.ExternalSecret != "" { - return r.registerExternalCacheJob(token, cred, reporter) - } - // No cache server to register against: caching is disabled, or the built-in server failed to start. - return func() {}, "" + return revoke, r.cacheHandler.ResultsURL(cred) } // registerExternalCacheJob POSTs to the remote cache-server's control-plane. @@ -676,23 +665,19 @@ func (r *Runner) registerCacheForTask(token, repo string, reporter *report.Repor // 401 the job's requests — better than failing the whole task for a cache // outage. The warning is mirrored to the job log so users can see why their // cache calls 401, instead of having to read the runner daemon's stderr. -func (r *Runner) registerExternalCacheJob(token string, cred artifactcache.JobCredential, reporter *report.Reporter) (func(), string) { +func (r *Runner) registerExternalCacheJob(token, repo, publicURL string, reporter *report.Reporter) func() { base := strings.TrimRight(r.cfg.Cache.ExternalServer, "/") - resultsURL := "" - if body, err := postInternalCache(base+"/_internal/register", r.cfg.Cache.ExternalSecret, map[string]any{ - "token": token, "repo": cred.Repo, "results": cred.Results, "insecure_tls": cred.InsecureTLS, - "public_url": base, + if err := postInternalCache(base+"/_internal/register", r.cfg.Cache.ExternalSecret, map[string]any{ + "token": token, "repo": repo, "public_url": publicURL, }); err != nil { log.Warnf("cache external_server register failed (%s): %v", base, err) if reporter != nil { reporter.Logf("::warning::%s", runner.EscapeCommandData(fmt.Sprintf( "cache external_server register failed (%s): %v — cache requests from this job will be unauthenticated and likely return 401", base, err))) } - } else if forwarded, _ := body["results_url"].(string); forwarded != "" { - resultsURL = base // the answer only says it forwards, its own address need not be the job's } return func() { - if _, err := postInternalCache(base+"/_internal/revoke", r.cfg.Cache.ExternalSecret, + if err := postInternalCache(base+"/_internal/revoke", r.cfg.Cache.ExternalSecret, map[string]any{"token": token}); err != nil { log.Warnf("cache external_server revoke failed (%s): %v", base, err) if reporter != nil { @@ -700,33 +685,30 @@ func (r *Runner) registerExternalCacheJob(token string, cred artifactcache.JobCr "cache external_server revoke failed (%s): %v", base, err))) } } - }, resultsURL + } } -func postInternalCache(url, secret string, body map[string]any) (map[string]any, error) { +func postInternalCache(url, secret string, body map[string]any) error { buf, err := json.Marshal(body) if err != nil { - return nil, err + return err } req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(buf)) if err != nil { - return nil, err + return err } req.Header.Set("Authorization", "Bearer "+secret) req.Header.Set("Content-Type", "application/json") client := &http.Client{Timeout: 5 * time.Second} resp, err := client.Do(req) if err != nil { - return nil, err + return err } defer resp.Body.Close() if resp.StatusCode/100 != 2 { - return nil, fmt.Errorf("status %d", resp.StatusCode) + return fmt.Errorf("status %d", resp.StatusCode) } - answer := map[string]any{} - // A server too old to answer with a body is not an error, it simply tells us nothing. - _ = json.UnmarshalRead(resp.Body, &answer) - return answer, nil + return nil } func (r *Runner) RunningCount() int64 { @@ -848,12 +830,22 @@ func warnIgnoredCacheSecret(cfg *config.Config) { log.Warnf("%s is set but cache.external_server is not; the built-in cache server does not use a shared secret, so the value is ignored", key) } -func (r *Runner) cacheIsolatedFrom(job *model.Job, pickPlatform func([]string) string) bool { +// cacheForJob returns the cache address for the job, and the container to attach to its network to reach it there. +func (r *Runner) cacheForJob(job *model.Job, pickPlatform func([]string) string) (string, string) { jobContainer := job.Container() - return (jobContainer != nil && jobContainer.Image != "" || pickPlatform(job.RunsOn()) != labels.SelfHostedPlatform) && r.isolatedCacheNetwork() != "" + cacheContainer := "" + if jobContainer != nil && jobContainer.Image != "" || pickPlatform(job.RunsOn()) != labels.SelfHostedPlatform { + cacheContainer = r.isolatedCacheContainer() + } + if cacheContainer == "" { + return strings.TrimSuffix(r.builtInCacheURL(), "/"), "" + } + cacheURL, _ := url.Parse(r.cacheHandler.ExternalURL()) + cacheURL.Host = net.JoinHostPort(cacheContainer, cacheURL.Port()) + return cacheURL.String(), cacheContainer } -func (r *Runner) detectIsolatedCacheNetwork() string { +func (r *Runner) detectIsolatedCacheContainer() string { if r.cacheHandler == nil { return "" } @@ -863,9 +855,9 @@ func (r *Runner) detectIsolatedCacheNetwork() string { } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - network, err := container.IsolatedNetwork(ctx, addr, r.cfg.Container.Network) + cacheContainer, err := container.IsolatedCacheContainer(ctx, addr, r.cfg.Container.Network) if err != nil { log.Warnf("cannot check whether jobs reach the cache server: %v", err) } - return network + return cacheContainer } diff --git a/internal/app/run/runner_cache_test.go b/internal/app/run/runner_cache_test.go index c6715bcb..74aca3f3 100644 --- a/internal/app/run/runner_cache_test.go +++ b/internal/app/run/runner_cache_test.go @@ -15,6 +15,7 @@ import ( "testing" "gitea.com/gitea/runner/act/artifactcache" + clientmocks "gitea.com/gitea/runner/internal/pkg/client/mocks" "gitea.com/gitea/runner/internal/pkg/config" "github.com/stretchr/testify/assert" @@ -23,54 +24,13 @@ import ( func emptyCfg() *config.Config { return &config.Config{} } -func TestRunner_registerCacheForTask(t *testing.T) { - dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"}) - require.NoError(t, err) - defer handler.Close() - - r := &Runner{cfg: emptyCfg(), cacheHandler: handler} - token := "run-token-123" - unregister, _ := r.registerCacheForTask(token, "owner/repo", nil) - - base := handler.ExternalURL() + "/_apis/artifactcache" - probe := func() int { - req, err := http.NewRequest(http.MethodGet, base+"/cache?keys=x&version=v", nil) - require.NoError(t, err) - req.Header.Set("Authorization", "Bearer "+token) - resp, err := http.DefaultClient.Do(req) - require.NoError(t, err) - resp.Body.Close() - return resp.StatusCode - } - - assert.NotEqual(t, http.StatusUnauthorized, probe(), - "token should be accepted while task is registered") - - unregister() - assert.Equal(t, http.StatusUnauthorized, probe(), - "token must be rejected after the revoker runs") -} - func TestRunner_registerCacheForTask_NoOps(t *testing.T) { - t.Run("nil cacheHandler", func(t *testing.T) { - r := &Runner{cfg: emptyCfg()} - unregister, _ := r.registerCacheForTask("tok", "owner/repo", nil) - require.NotNil(t, unregister) - unregister() - }) - - t.Run("empty token", func(t *testing.T) { - dir := filepath.Join(t.TempDir(), "artifactcache") - handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"}) - require.NoError(t, err) - defer handler.Close() - + for token, handler := range map[string]*artifactcache.Handler{"tok": nil, "": {}} { r := &Runner{cfg: emptyCfg(), cacheHandler: handler} - unregister, _ := r.registerCacheForTask("", "owner/repo", nil) - require.NotNil(t, unregister) + unregister, _ := r.registerCacheForTask(token, "owner/repo", "", nil) + require.NotNil(t, unregister, token) unregister() - }) + } } // Locks in @actions/cache's wire protocol: bearer on reserve/upload/commit @@ -81,10 +41,11 @@ func TestRunner_CacheFullFlow_MatchesToolkit(t *testing.T) { require.NoError(t, err) defer handler.Close() - r := &Runner{cfg: emptyCfg(), cacheHandler: handler} + r := &Runner{cfg: emptyCfg(), cacheHandler: handler, envs: map[string]string{"ACTIONS_RESULTS_URL": "https://gitea.example"}} + const publicURL = "http://a1b2c3d4e5f6:8088" token := "full-flow-token" - unregister, _ := r.registerCacheForTask(token, "owner/repo", nil) - defer unregister() + unregister, resultsURL := r.registerCacheForTask(token, "owner/repo", publicURL, nil) + assert.Equal(t, publicURL, resultsURL) base := handler.ExternalURL() + "/_apis/artifactcache" do := func(method, url, contentType, contentRange, body string) *http.Response { @@ -137,17 +98,22 @@ func TestRunner_CacheFullFlow_MatchesToolkit(t *testing.T) { } require.NoError(t, decodeJSON(resp, &hit)) require.Equal(t, key, hit.CacheKey) - require.NotEmpty(t, hit.ArchiveLocation) + require.True(t, strings.HasPrefix(hit.ArchiveLocation, publicURL+"/"), hit.ArchiveLocation) // download — toolkit does NOT attach Authorization here; the signature // in the URL must be enough. - dl, err := http.Get(hit.ArchiveLocation) + dl, err := http.Get(strings.Replace(hit.ArchiveLocation, publicURL, handler.ExternalURL(), 1)) require.NoError(t, err) defer dl.Body.Close() require.Equal(t, http.StatusOK, dl.StatusCode) got := make([]byte, 64) n, _ := dl.Body.Read(got) assert.Equal(t, body, string(got[:n])) + + unregister() + resp = do(http.MethodGet, fmt.Sprintf("%s/cache?keys=%s&version=%s", base, key, version), "", "", "") + resp.Body.Close() + assert.Equal(t, http.StatusUnauthorized, resp.StatusCode) } func decodeJSON(resp *http.Response, v any) error { @@ -155,33 +121,38 @@ func decodeJSON(resp *http.Response, v any) error { return json.UnmarshalRead(resp.Body, v) } -// End-to-end against a remote cache-server: token unknown → 401, register → -// reserve/upload/commit/find/download all OK, revoke → 401 again. Registering also names the -// instance and the address its jobs reach the server at, which makes a shared cache server the -// whole results service, as the built-in one is. +// End-to-end through a runner to a remote cache-server: token unknown → 401, register → +// reserve/upload/commit/find/download all OK, revoke → 401 again. func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) { - dir := filepath.Join(t.TempDir(), "remote-cache") const secret = "shared-secret-for-tests" - remote, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.2", InternalSecret: secret}) // advertised, never dialled + remote, err := artifactcache.StartHandler(artifactcache.Options{Dir: t.TempDir(), OutboundIP: "127.0.0.1", InternalSecret: secret}) require.NoError(t, err) defer remote.Close() - external := strings.Replace(remote.ExternalURL(), "127.0.0.2", "127.0.0.1", 1) - gitea := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + gitea := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + assert.Equal(t, "/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact", req.URL.Path) + assert.Equal(t, "Bearer external-task-token", req.Header.Get("Authorization")) _, _ = io.WriteString(w, `{"ok":true}`) })) defer gitea.Close() - r := &Runner{ - cfg: &config.Config{Cache: config.Cache{ - ExternalServer: external, + cli := clientmocks.NewClient(t) + cli.AddressValue = gitea.URL + r := NewRunner(&config.Config{ + Runner: config.Runner{Insecure: true}, + Cache: config.Cache{ + Host: "127.0.0.1", + ExternalServer: remote.ExternalURL() + "//", ExternalSecret: secret, - }}, - envs: map[string]string{"ACTIONS_RESULTS_URL": gitea.URL}, - } + }, + }, &config.Registration{}, cli) + t.Cleanup(func() { _ = r.Close() }) + require.NotNil(t, r.cacheHandler) + require.Equal(t, r.cacheHandler.ExternalURL()+"/", r.envs["ACTIONS_CACHE_URL"]) + const publicURL = "http://a1b2c3d4e5f6:8088" token := "external-task-token" repo := "owner/repoX" - base := external + "/_apis/artifactcache" + base := r.envs["ACTIONS_CACHE_URL"] + "_apis/artifactcache" probe := func() int { req, _ := http.NewRequest(http.MethodGet, base+"/cache?keys=k&version=v", nil) req.Header.Set("Authorization", "Bearer "+token) @@ -194,30 +165,24 @@ func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) { require.Equal(t, http.StatusUnauthorized, probe(), "token must be unknown to the remote server before registration") - disabled := &Runner{cfg: &config.Config{Cache: config.Cache{ - Enabled: new(bool), ExternalServer: external, ExternalSecret: secret, - }}} - disabled.registerCacheForTask(token, repo, nil) - require.Equal(t, http.StatusUnauthorized, probe(), - "a disabled cache registers nothing, whatever external server is left configured") - - unregister, resultsURL := r.registerCacheForTask(token, repo, nil) + unregister, resultsURL := r.registerCacheForTask(token, repo, publicURL, nil) require.NotEqual(t, http.StatusUnauthorized, probe(), "token must be accepted after registerCacheForTask") - // The server took the results service over, so the artifact half reaches Gitea through it. - require.Equal(t, external, resultsURL) + require.Equal(t, publicURL, resultsURL) artifact, err := http.NewRequestWithContext(t.Context(), http.MethodPost, - resultsURL+"/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact", nil) + r.cacheHandler.ExternalURL()+"/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact", nil) require.NoError(t, err) artifact.Header.Set("Authorization", "Bearer "+token) forwarded, err := http.DefaultClient.Do(artifact) require.NoError(t, err) - forwarded.Body.Close() require.Equal(t, http.StatusOK, forwarded.StatusCode) + var artifactResult map[string]bool + require.NoError(t, decodeJSON(forwarded, &artifactResult)) + assert.True(t, artifactResult["ok"]) // Full reserve→upload→commit→find→download cycle, identical to what - // @actions/cache does, against the remote (external) server. + // @actions/cache does, through the runner to the remote server. body := []byte("payload-from-task") reserveBody, _ := json.Marshal(&artifactcache.Request{Key: "ext-key", Version: "v", Size: int64(len(body))}) req, _ := http.NewRequest(http.MethodPost, base+"/caches", bytes.NewReader(reserveBody)) @@ -256,14 +221,22 @@ func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) { ArchiveLocation string `json:"archiveLocation"` } require.NoError(t, decodeJSON(resp, &hit)) - require.True(t, strings.HasPrefix(hit.ArchiveLocation, external), hit.ArchiveLocation) + require.True(t, strings.HasPrefix(hit.ArchiveLocation, publicURL+"/"), hit.ArchiveLocation) - dl, err := http.Get(hit.ArchiveLocation) + dl, err := http.Get(strings.Replace(hit.ArchiveLocation, publicURL, r.cacheHandler.ExternalURL(), 1)) require.NoError(t, err) defer dl.Body.Close() require.Equal(t, http.StatusOK, dl.StatusCode) + downloaded, err := io.ReadAll(dl.Body) + require.NoError(t, err) + assert.Equal(t, body, downloaded) unregister() assert.Equal(t, http.StatusUnauthorized, probe(), "token must be rejected after the revoker runs") + + forwarded, err = http.DefaultClient.Do(artifact) + require.NoError(t, err) + forwarded.Body.Close() + assert.Equal(t, http.StatusUnauthorized, forwarded.StatusCode) } diff --git a/internal/app/run/runner_idle_cleanup_test.go b/internal/app/run/runner_idle_cleanup_test.go index e5451f04..2bb73d4a 100644 --- a/internal/app/run/runner_idle_cleanup_test.go +++ b/internal/app/run/runner_idle_cleanup_test.go @@ -311,11 +311,13 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { } var swept []string + var detached []string var sweptVolumes []string var sweptCutoff time.Time origRemoveOrphanNetworks := removeOrphanNetworks - removeOrphanNetworks = func(_ context.Context, runnerUUID string, createdBefore time.Time) error { + removeOrphanNetworks = func(_ context.Context, runnerUUID, detach string, createdBefore time.Time) error { swept = append(swept, runnerUUID) + detached = append(detached, detach) sweptCutoff = createdBefore return nil } @@ -329,6 +331,7 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { t.Cleanup(func() { removeOrphanJobVolumes = origRemoveOrphanJobVolumes }) r := &Runner{uuid: "runner-1", cfg: cfg, now: func() time.Time { return now }} + r.isolatedCacheContainer = func() string { return "a1b2c3d4e5f6" } r.OnIdle(context.Background()) assert.Equal(t, []string{"runner-1"}, swept) assert.Equal(t, swept, sweptVolumes) @@ -340,6 +343,7 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { dockerReachable = func(context.Context) bool { return false } t.Cleanup(func() { dockerReachable = origDockerReachable }) hostOnly := &Runner{uuid: "runner-2", cfg: &config.Config{Runner: cfg.Runner}, now: func() time.Time { return now }} + hostOnly.isolatedCacheContainer = func() string { return "" } hostOnly.OnIdle(context.Background()) assert.Equal(t, []string{"runner-1"}, swept) assert.Equal(t, swept, sweptVolumes) @@ -348,4 +352,5 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { now = now.Add(time.Minute) hostOnly.OnIdle(context.Background()) assert.Equal(t, []string{"runner-1", "runner-2"}, sweptVolumes) + assert.Equal(t, []string{"a1b2c3d4e5f6", ""}, detached) } diff --git a/internal/app/run/runner_test.go b/internal/app/run/runner_test.go index 8d1b3e8b..87f57757 100644 --- a/internal/app/run/runner_test.go +++ b/internal/app/run/runner_test.go @@ -251,15 +251,19 @@ func TestNewRunnerLeavesProxyToTheTask(t *testing.T) { cfg := &config.Config{} cfg.Cache.ExternalServer = "http://cache.local:8088/" + cfg.Cache.Host = "runner.local" reg := &config.Registration{Name: "runner"} cli := clientmocks.NewClient(t) cli.AddressValue = "https://gitea.example/" r := NewRunner(cfg, reg, cli) + t.Cleanup(func() { _ = r.Close() }) require.NotContains(t, r.envs, "http_proxy") require.NotContains(t, r.envs, "no_proxy") - assert.Empty(t, r.builtInCacheURL(), "an external cache server is the operator's to exempt, not ours") + proxyEnv := JobProxyEnv(r.envs, r.builtInCacheURL(), nil) + assert.Contains(t, proxyEnv["no_proxy"], "runner.local") + assert.NotContains(t, proxyEnv["no_proxy"], "cache.local") } func taskWithDefaultActionsURL(url string) *runnerv1.Task { @@ -289,7 +293,7 @@ func TestNewRunnerCacheServiceV2(t *testing.T) { // The registration is what makes it true: the cache server takes the results service over, // having been told which instance to forward the artifact half to. - revoke, resultsURL := r.registerCacheForTask(token, "owner/repo", nil) + revoke, resultsURL := r.registerCacheForTask(token, "owner/repo", r.cacheHandler.ExternalURL(), nil) defer revoke() require.Equal(t, r.cacheHandler.ExternalURL(), resultsURL) @@ -341,26 +345,16 @@ func TestNewRunnerCacheServiceV2(t *testing.T) { workflow, err := model.ReadWorkflow(strings.NewReader(`jobs: {native: {runs-on: native}, linux: {runs-on: linux}, containerized: {runs-on: native, container: alpine}, empty: {runs-on: native, container: ""}}`)) require.NoError(t, err) - r.isolatedCacheNetwork = func() string { return "compose" } + r.isolatedCacheContainer = func() string { return "a1b2c3d4e5f6" } pickPlatform := func(runsOn []string) string { return map[string]string{"native": labels.SelfHostedPlatform}[runsOn[0]] } for job, isolated := range map[string]bool{"native": false, "linux": true, "containerized": true, "empty": false} { - assert.Equal(t, isolated, r.cacheIsolatedFrom(workflow.GetJob(job), pickPlatform), job) + cacheURL, cacheContainer := r.cacheForJob(workflow.GetJob(job), pickPlatform) + if isolated { + assert.Equal(t, "a1b2c3d4e5f6", cacheContainer, job) + assert.Equal(t, strings.Replace(r.cacheHandler.ExternalURL(), "127.0.0.1", cacheContainer, 1), cacheURL, job) + } else { + assert.Empty(t, cacheContainer, job) + assert.Equal(t, r.cacheHandler.ExternalURL(), cacheURL, job) + } } } - -// The v1 cache client appends its path to ACTIONS_CACHE_URL without a separator, so a configured -// server that is missing the slash would send it to a URL whose port swallows the path. -func TestNewRunnerNormalizesTheExternalCacheServer(t *testing.T) { - cfg := &config.Config{} - cfg.Cache.ExternalServer = "http://cache.local:8088//" - cli := clientmocks.NewClient(t) - cli.AddressValue = "https://gitea.example/" - - r := NewRunner(cfg, &config.Registration{Name: "runner"}, cli) - - assert.Equal(t, "http://cache.local:8088/", r.envs["ACTIONS_CACHE_URL"]) - // Nothing to front the results service with, so the variable stays unset and the client keeps - // to v1, which reads the cache URL first. - assert.Equal(t, "https://gitea.example", r.envs["ACTIONS_RESULTS_URL"]) - assert.Empty(t, r.envs[runner.CacheServiceV2Env]) -} diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index 3b7bcb25..8e4229e1 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -148,19 +148,12 @@ cache: #dir: "" # Outbound IP or hostname that job containers use to reach this runner's cache server. # Leave empty to detect automatically. 0.0.0.0 is not valid here. - # If the runner itself runs in Docker, automatic detection can choose an - # address on the runner container's network that job containers cannot reach - # when the runner creates a separate per-job network. In that case, set this - # to a hostname/IP reachable from job containers, and set port to a fixed - # published port or put the job containers on a shared Docker network. - # Ignored when external_server is set. + # A runner container on the jobs' Docker daemon attaches to each job network automatically. #host: "" # Port for the built-in cache server. 0 picks a random free port. - # Ignored when external_server is set. #port: 0 - # URL of a shared `gitea-runner cache-server` to use instead of starting a local one. + # URL of a shared `gitea-runner cache-server` to forward cache calls to. # Set on every runner that should share a cache pool. A trailing slash is optional. - # Jobs reach the server at this URL too, so set it to the reverse proxy when one fronts the server. # Example: "http://cache-host:8088/" # Requires external_secret (below) to match the value on the cache-server. #external_server: "" @@ -181,7 +174,7 @@ cache: # they do not take for GitHub, so reaching it means editing that check out of their own bundle, # put back after the copy into the job. That edit is made either way, this only governs the API # advertised. A bundle that does not match is left alone. With v2, artifact uploads go via the - # cache server unless container jobs cannot reach its Docker network. + # cache server. #v2: true # How the cache server discards entries, ignored when external_server is set since that # server applies its own. Leave a setting out for its default; 0s or 0 turns the three @@ -201,9 +194,8 @@ cache: container: # Specifies the network to which the container will connect. # Could be host, bridge or the name of a custom network. + # A runner container outside bridge cannot join it, so jobs on bridge cannot reach its cache. # If it's empty, runner will create a network automatically. - # For dockerized runners using the built-in cache server, a custom shared - # network can be required so job containers can reach cache.host/cache.port. # Deprecated: `network_mode` is still accepted for old configs; use `network` instead. #network: "" # network_create_options only apply when `network` is left empty and the runner