diff --git a/README.md b/README.md index ffbb5116..05da59ed 100644 --- a/README.md +++ b/README.md @@ -372,7 +372,7 @@ All carry `strategy` and `registry`. Registry values come from configured host/A The HTTP server metrics of these handlers carry `cachew.cache.result` (`hit`, `miss`, `coalesced` or `error`). Hits also carry `cachew.cache.source`, the backend that served them (`disk`, `s3`, `memory`, `remote`, or `other`), after stale and conditional probes. -Origin metrics count HTTP exchanges that reach an origin through `http.DefaultTransport`, including redirect hops and transport failures. Requests rejected during an upstream 429 cooldown are not counted. `status_code=none` means no HTTP response; `error_type` is `none`, `transport`, `timeout`, or `canceled`. Duration ends at response headers or transport failure. Registry identifies the logical artifact provider, including its redirects, rather than each physical destination. These metrics do not count every low-level connection retry, Git transport, Go-module fetch, GitHub metadata lookup, OCI tag-list or referrer pass-through, or S3 operation. OCI registry ping and token exchanges are counted, because they go to the registry for the artifact. Internal Hermit forwarding to another Cachew route is excluded from origin/fill counts; the receiving handler measures its own work. +Origin metrics count HTTP exchanges that reach an origin through the shared upstream transport, including redirect hops and transport failures. Requests rejected during an upstream 429 cooldown are not counted. `status_code=none` means no HTTP response; `error_type` is `none`, `transport`, `timeout`, or `canceled`. Duration ends at response headers or transport failure. Registry identifies the logical artifact provider, including its redirects, rather than each physical destination. These metrics do not count every low-level connection retry, Git transport, Go-module fetch, GitHub metadata lookup, OCI tag-list or referrer pass-through, or S3 operation. OCI registry ping and token exchanges are counted, because they go to the registry for the artifact. Internal Hermit forwarding to another Cachew route is excluded from origin/fill counts; the receiving handler measures its own work. Fills record a terminal outcome, rather than treating an origin 200 as successful publication. `purpose` separates `origin`, `backfill` and ranged `heal`. Results distinguish successful completion from creation, write, commit, transfer or read errors, cancellation, and superseded heals. Tier writes report backend outcomes separately: a failed S3 write can coexist with a successful disk write. Early close or canceled backfill is not a successful fill. Client responses continue normally when cache writes fail. diff --git a/client/client.go b/client/client.go index 1805d695..088a2e1c 100644 --- a/client/client.go +++ b/client/client.go @@ -65,8 +65,8 @@ type CacheWriter interface { Abort(err error) error } -// stdTransport is captured before cachewd wraps http.DefaultTransport with -// upstream 429 cooldowns, so cache clients keep their own tuned pool. +// stdTransport is the process default transport, kept so cache clients have +// their own tuned pool even if a test has swapped http.DefaultTransport. var stdTransport = http.DefaultTransport.(*http.Transport) //nolint:errcheck,gochecknoglobals // HeaderFunc returns headers to attach to each outgoing request. diff --git a/cmd/cachewd/main.go b/cmd/cachewd/main.go index 49b2a142..90fddee0 100644 --- a/cmd/cachewd/main.go +++ b/cmd/cachewd/main.go @@ -460,10 +460,13 @@ func configureProcess(logger *slog.Logger, mutexProfileFraction int, upstreamBac // side-effect import; OPA gates external access. runtime.SetMutexProfileFraction(mutexProfileFraction) - // Every strategy that uses http.DefaultTransport shares one set of + // Every strategy that uses httputil.UpstreamTransport shares one set of // per-hostname and per-credential 429 cooldowns. Origin metrics sit below the cooldowns so that // requests rejected during a cooldown are not counted as origin traffic. - http.DefaultTransport = httputil.NewBackoffTransport(metrics.NewOriginTransport(http.DefaultTransport), upstreamBackoff, upstreamBackoffMax, logger) //nolint:reassign + // http.DefaultTransport itself stays a *http.Transport: OPA's http.send + // asserts that type whenever a policy sets a TLS option, and panics on a + // wrapper. + httputil.SetUpstreamTransport(httputil.NewBackoffTransport(metrics.NewOriginTransport(http.DefaultTransport), upstreamBackoff, upstreamBackoffMax, logger)) } // loadGlobalConfig unmarshals the global config from HCL, using a two-pass diff --git a/cmd/cachewd/opa_transport_test.go b/cmd/cachewd/opa_transport_test.go new file mode 100644 index 00000000..d2596b76 --- /dev/null +++ b/cmd/cachewd/opa_transport_test.go @@ -0,0 +1,70 @@ +package main + +import ( + "encoding/pem" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + "github.com/open-policy-agent/opa/v1/rego" + + "github.com/block/cachew/internal/httputil" + "github.com/block/cachew/internal/logging" +) + +const opaTransportStartupChildEnv = "CACHEWD_OPA_TRANSPORT_STARTUP_CHILD" + +func TestOPAHTTPSendWithTLSOptionsAfterProcessSetup(t *testing.T) { + if os.Getenv(opaTransportStartupChildEnv) != "1" { + cmd := exec.CommandContext(t.Context(), os.Args[0], "-test.run=^"+t.Name()+"$", "-test.count=1") + cmd.Env = append(os.Environ(), opaTransportStartupChildEnv+"=1") + output, err := cmd.CombinedOutput() + assert.NoError(t, err, "%s", output) + return + } + logger, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelError}) + configureProcess(logger, 0, time.Second, time.Minute) + + _, ok := http.DefaultTransport.(*http.Transport) + assert.True(t, ok, "configureProcess must leave http.DefaultTransport a *http.Transport, got %T", http.DefaultTransport) + _, ok = httputil.UpstreamTransport().(*http.Transport) + assert.False(t, ok, "configureProcess must install the upstream transport") + + var upstreamHits int + upstream := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + upstreamHits++ + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(upstream.Close) + caFile := filepath.Join(t.TempDir(), "ca.crt") + assert.NoError(t, os.WriteFile(caFile, pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", Bytes: upstream.Certificate().Raw, + }), 0o600)) + + query, err := rego.New( + rego.Query("data.cachew.allow"), + rego.Module("policy.rego", ` +package cachew + +default allow := false + +allow if { + resp := http.send({"method": "HEAD", "url": input.url, "tls_ca_cert_file": input.ca_file}) + resp.status_code == 200 +} +`), + ).PrepareForEval(ctx) + assert.NoError(t, err) + + results, err := query.Eval(ctx, rego.EvalInput(map[string]any{"url": upstream.URL, "ca_file": caFile})) + assert.NoError(t, err) + assert.Equal(t, 1, len(results)) + assert.Equal(t, true, results[0].Expressions[0].Value) + assert.Equal(t, 1, upstreamHits) +} diff --git a/internal/githubapp/tokens.go b/internal/githubapp/tokens.go index fa3e37aa..73519d66 100644 --- a/internal/githubapp/tokens.go +++ b/internal/githubapp/tokens.go @@ -24,7 +24,7 @@ type TokenManagerProvider func() (*TokenManager, error) // from one or more GitHub App configurations. func NewTokenManagerProvider(configs []Config, logger *slog.Logger) TokenManagerProvider { return sync.OnceValues(func() (*TokenManager, error) { - return newTokenManager(configs, logger, githubAPIBase, http.DefaultClient) + return newTokenManager(configs, logger, githubAPIBase, httputil.UpstreamClient()) }) } diff --git a/internal/githubapp/transport.go b/internal/githubapp/transport.go index acb08437..3e77196d 100644 --- a/internal/githubapp/transport.go +++ b/internal/githubapp/transport.go @@ -33,7 +33,7 @@ func NewAPIClient(base *http.Client) *http.Client { client := *base transport := client.Transport if transport == nil { - transport = http.DefaultTransport + transport = httputil.UpstreamTransport() } client.Transport = &apiTransport{ base: transport, diff --git a/internal/httputil/upstream.go b/internal/httputil/upstream.go new file mode 100644 index 00000000..b9126587 --- /dev/null +++ b/internal/httputil/upstream.go @@ -0,0 +1,45 @@ +package httputil + +import ( + "net/http" + "sync/atomic" +) + +// upstream is the transport cachewd installs for outbound requests that would +// otherwise use http.DefaultTransport. It is kept here instead of being +// assigned to http.DefaultTransport because third-party code, such as OPA's +// http.send, asserts that the global is a *http.Transport and panics on any +// wrapper. +var upstream atomic.Pointer[http.RoundTripper] //nolint:gochecknoglobals + +// SetUpstreamTransport installs the shared outbound transport returned by +// [UpstreamTransport], leaving http.DefaultTransport unchanged. A nil +// transport restores the default of following http.DefaultTransport. +func SetUpstreamTransport(rt http.RoundTripper) { + if rt == nil { + upstream.Store(nil) + return + } + upstream.Store(&rt) +} + +// UpstreamTransport returns the shared outbound transport. Until one is +// installed, requests go to whatever http.DefaultTransport is at request time, +// so tests can still swap the global to intercept upstream traffic. +func UpstreamTransport() http.RoundTripper { + return upstreamTransport{} +} + +// UpstreamClient returns a client that sends through [UpstreamTransport]. +func UpstreamClient() *http.Client { + return &http.Client{Transport: UpstreamTransport()} +} + +type upstreamTransport struct{} + +func (upstreamTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if rt := upstream.Load(); rt != nil { + return (*rt).RoundTrip(req) //nolint:wrapcheck + } + return http.DefaultTransport.RoundTrip(req) //nolint:wrapcheck +} diff --git a/internal/httputil/upstream_test.go b/internal/httputil/upstream_test.go new file mode 100644 index 00000000..52909062 --- /dev/null +++ b/internal/httputil/upstream_test.go @@ -0,0 +1,38 @@ +package httputil_test + +import ( + "net/http" + "testing" + + "github.com/alecthomas/assert/v2" + + "github.com/block/cachew/internal/httputil" +) + +type staticTransport struct{ status int } + +func (s staticTransport) RoundTrip(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: s.status, Body: http.NoBody}, nil +} + +func TestUpstreamTransportFollowsDefaultUntilSet(t *testing.T) { + defaultTransport := http.DefaultTransport + t.Cleanup(func() { + http.DefaultTransport = defaultTransport //nolint:reassign + httputil.SetUpstreamTransport(nil) + }) + httputil.SetUpstreamTransport(nil) + http.DefaultTransport = staticTransport{status: http.StatusTeapot} //nolint:reassign + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://example.com/", nil) + assert.NoError(t, err) + + resp, err := httputil.UpstreamClient().Do(req) + assert.NoError(t, err) + assert.Equal(t, http.StatusTeapot, resp.StatusCode) + + httputil.SetUpstreamTransport(staticTransport{status: http.StatusAccepted}) + resp, err = httputil.UpstreamClient().Do(req) + assert.NoError(t, err) + assert.Equal(t, http.StatusAccepted, resp.StatusCode) + assert.Equal[http.RoundTripper](t, staticTransport{status: http.StatusTeapot}, http.DefaultTransport) +} diff --git a/internal/strategy/artifactory.go b/internal/strategy/artifactory.go index 28132463..80ffbcfa 100644 --- a/internal/strategy/artifactory.go +++ b/internal/strategy/artifactory.go @@ -70,7 +70,7 @@ func NewArtifactory(ctx context.Context, config ArtifactoryConfig, cache cache.C a := &Artifactory{ target: u, cache: cache, - client: &http.Client{}, + client: httputil.UpstreamClient(), logger: logging.FromContext(ctx), } diff --git a/internal/strategy/git/git.go b/internal/strategy/git/git.go index 63350800..c5c7b2ba 100644 --- a/internal/strategy/git/git.go +++ b/internal/strategy/git/git.go @@ -160,7 +160,7 @@ func New( config: config, cache: cache, cloneManager: cloneManager, - httpClient: http.DefaultClient, + httpClient: cachewhttputil.UpstreamClient(), ctx: ctx, scheduler: scheduler.WithQueuePrefix("git"), spoolRegistry: spools, diff --git a/internal/strategy/github_releases.go b/internal/strategy/github_releases.go index 79871b89..943e44cc 100644 --- a/internal/strategy/github_releases.go +++ b/internal/strategy/github_releases.go @@ -61,7 +61,7 @@ func NewGitHubReleases(ctx context.Context, config GitHubReleasesConfig, cache c s := &GitHubReleases{ config: config, cache: cache, - client: githubapp.NewAPIClient(http.DefaultClient), + client: githubapp.NewAPIClient(httputil.UpstreamClient()), tokenManager: tokenManager, } // eg. https://github.com/alecthomas/chroma/releases/download/v2.21.1/chroma-2.21.1-darwin-amd64.tar.gz diff --git a/internal/strategy/gomod/gomod.go b/internal/strategy/gomod/gomod.go index cc3a3d97..05a8d33a 100644 --- a/internal/strategy/gomod/gomod.go +++ b/internal/strategy/gomod/gomod.go @@ -12,6 +12,7 @@ import ( "github.com/block/cachew/internal/cache" "github.com/block/cachew/internal/gitclone" + "github.com/block/cachew/internal/httputil" "github.com/block/cachew/internal/logging" "github.com/block/cachew/internal/strategy" ) @@ -64,6 +65,7 @@ func New(ctx context.Context, config Config, cache cache.Cache, mux strategy.Mux } publicFetcher := &goproxy.GoFetcher{ + Transport: httputil.UpstreamTransport(), Env: []string{ "GOPROXY=" + config.Proxy, "GOSUMDB=off", // Disable checksum database validation in fetcher, to prevent unneccessary double validation @@ -81,8 +83,9 @@ func New(ctx context.Context, config Config, cache cache.Cache, mux strategy.Mux } s.goproxy = &goproxy.Goproxy{ - Logger: s.logger, - Fetcher: fetcher, + Logger: s.logger, + Fetcher: fetcher, + Transport: httputil.UpstreamTransport(), Cacher: &goproxyCacher{ cache: cache, }, diff --git a/internal/strategy/hermit.go b/internal/strategy/hermit.go index bcc524af..3c06a981 100644 --- a/internal/strategy/hermit.go +++ b/internal/strategy/hermit.go @@ -62,7 +62,7 @@ func NewHermit(ctx context.Context, config HermitConfig, _ jobscheduler.Schedule s := &Hermit{ config: config, cache: c, - client: http.DefaultClient, + client: httputil.UpstreamClient(), logger: logger, mux: mux, } diff --git a/internal/strategy/host.go b/internal/strategy/host.go index 38931e11..6010b175 100644 --- a/internal/strategy/host.go +++ b/internal/strategy/host.go @@ -69,13 +69,13 @@ func NewHost(_ context.Context, config HostConfig, cache cache.Cache, mux Mux, s h := &Host{ target: u, cache: cache, - client: &http.Client{}, + client: httputil.UpstreamClient(), prefix: prefix, headers: config.Headers, } if len(config.Fallbacks) > 0 { - transport := &hostFallbackTransport{client: &http.Client{Transport: http.DefaultTransport}} + transport := &hostFallbackTransport{client: httputil.UpstreamClient()} for _, config := range config.Fallbacks { target, err := url.Parse(config.Target) if err != nil || target.Host == "" || (target.Scheme != "http" && target.Scheme != "https") || diff --git a/internal/strategy/oci.go b/internal/strategy/oci.go index 0e8ca08e..449e755f 100644 --- a/internal/strategy/oci.go +++ b/internal/strategy/oci.go @@ -103,7 +103,7 @@ func NewOCI(ctx context.Context, config OCIConfig, c cache.Cache, mux Mux, spool if err != nil { return nil, err } - transport, err := newOCITransport(upstream, config.Username, config.Password, http.DefaultTransport) + transport, err := newOCITransport(upstream, config.Username, config.Password, httputil.UpstreamTransport()) if err != nil { return nil, err } diff --git a/internal/strategy/proxy.go b/internal/strategy/proxy.go index 8a16355b..6ba7bdb3 100644 --- a/internal/strategy/proxy.go +++ b/internal/strategy/proxy.go @@ -10,6 +10,7 @@ import ( "github.com/alecthomas/errors" "github.com/block/cachew/internal/cache" + "github.com/block/cachew/internal/httputil" "github.com/block/cachew/internal/logging" "github.com/block/cachew/internal/spool" "github.com/block/cachew/internal/strategy/handler" @@ -56,7 +57,7 @@ var ( func NewHTTPProxy(ctx context.Context, c cache.Cache, _ Mux, spools *spool.Registry) (*HTTPProxy, error) { logger := logging.FromContext(ctx) - client := &http.Client{} + client := httputil.UpstreamClient() p := &HTTPProxy{logger: logger} p.handler = handler.New(client, c, spools).Metrics("proxy", "other").