mirror of
https://gitea.com/gitea/runner.git
synced 2026-11-09 08:40:29 +00:00
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 <[email protected]> Reviewed-on: https://gitea.com/gitea/runner/pulls/1229 Reviewed-by: bircni <[email protected]> Co-authored-by: silverwind <[email protected]>
This commit is contained in:
1 parent
95daa50d9b
commit
ace8653105
17 files changed
+428
-252
No files matched your search
@@ -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
|
||||
|
||||
@@ -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"))
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
@@ -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,
|
||||
},
|
||||
{
|
||||
|
||||
Reference in new issue
Block a user