Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d6602d5bf8 | ||
|
|
d0377ae3da | ||
|
|
fc5ffc91f1 |
@@ -126,6 +126,17 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
|
||||
- `PIXA_CACHE_MAX_BYTES`: disk cache limit in bytes; `0` disables it;
|
||||
default 75% of (free space + what the cache holds)
|
||||
- the rest are in the table under Configuration below
|
||||
- **Memory:** pixa can use up to `max_concurrent_processing` times 1.6 GiB for
|
||||
the images it processes at once, plus `upstream_connections` times
|
||||
`upstream_max_response_size` for the source images it has fetched that wait
|
||||
for processing. 1.6 GiB is the most pixad's resident memory was measured to
|
||||
grow while it processed one 8192x8192 source, the largest it accepts, into
|
||||
each output format in turn, one at a time; it grew most for a 40 MB JPEG of
|
||||
random pixels saved as AVIF. At the defaults that is half the CPUs pixa sees
|
||||
times 1.6 GiB, plus 64 times 50 MiB: on a host where pixa sees 48 CPUs, 24
|
||||
times 1.6 GiB plus 3.1 GiB, about 42 GiB. Set `PIXA_MAX_CONCURRENT_PROCESSING`
|
||||
and `PIXA_UPSTREAM_CONNECTIONS` so that this fits the memory the container may
|
||||
use.
|
||||
- **Health check:** the image's `HEALTHCHECK` requests
|
||||
`/.well-known/healthcheck.json`. upaas reads the container's health 60 seconds
|
||||
after a deploy and marks the deploy failed unless it is `healthy`. The probe
|
||||
@@ -198,11 +209,9 @@ path under `/v1/` answers 200, in maintenance mode too.
|
||||
redirects to, is `localhost`, ends in `.localhost` or `.local`, or has an
|
||||
address in a blocked network (see `blocked_networks`); 429 over
|
||||
`image_requests_per_minute` (below); 415 when the upstream answers with a
|
||||
`Content-Type` that is not one of the source formats (below), with a body
|
||||
whose first bytes are not those of the format its `Content-Type` names, or
|
||||
with an image whose first bytes libvips reads as another format; 502 when the
|
||||
upstream answered with an error status, and for 5 minutes after that, or after
|
||||
a body whose first bytes are not those of its `Content-Type`, for the same
|
||||
`Content-Type` that is not one of the source formats (below), or with an image
|
||||
whose first bytes libvips reads as another format; 502 when the upstream
|
||||
answered with an error status, and for 5 minutes after that for the same
|
||||
source URL; 503 when pixa is busy or in maintenance mode; 500 for any other
|
||||
failure.
|
||||
- `GET` or `HEAD` `/v1/e/<token>/<name>` — an image through an encrypted URL
|
||||
@@ -281,12 +290,8 @@ above 8192.
|
||||
The source image may be JPEG, PNG, GIF, WebP, AVIF or JPEG XL. The upstream's
|
||||
`Content-Type` must name its format, and the image's first bytes must match it.
|
||||
An upstream answering with any other `Content-Type`, SVG included, is refused
|
||||
with 415. So is a body whose first bytes are not those of the format its
|
||||
`Content-Type` names, such as a JPEG sent as `image/png`, text sent as an image,
|
||||
or an empty body. pixa then does not fetch that source URL again for 5 minutes,
|
||||
and answers 502 for it meanwhile, as after an upstream error status. An image
|
||||
whose first bytes libvips reads as another format, such as an SVG whose first
|
||||
bytes pass the check for AVIF, is refused with 415 too.
|
||||
with 415, and so is an image whose first bytes libvips reads as another format,
|
||||
such as an SVG whose first bytes pass the check for AVIF.
|
||||
|
||||
With the format `auto`, pixa chooses the format for each request from its
|
||||
`Accept` header, in this order:
|
||||
@@ -521,7 +526,7 @@ tell. With no file, pixa uses the environment and the defaults.
|
||||
| `PIXA_ALLOW_HTTP` | `allow_http` | Allow plain-HTTP upstreams, for testing only; default `false` |
|
||||
| `PIXA_UPSTREAM_CONNECTIONS_PER_HOST` | `upstream_connections_per_host` | Concurrent connections per upstream host; default `20` |
|
||||
| `PIXA_UPSTREAM_CONNECTIONS` | `upstream_connections` | Concurrent connections to all upstream hosts together; default `64` |
|
||||
| `PIXA_MAX_CONCURRENT_PROCESSING` | `max_concurrent_processing` | Images processed at once; default the number of CPUs |
|
||||
| `PIXA_MAX_CONCURRENT_PROCESSING` | `max_concurrent_processing` | Images processed at once; default half the CPUs, at least 1 |
|
||||
| `PIXA_IMAGE_REQUESTS_PER_MINUTE` | `image_requests_per_minute` | Image requests per client per minute; `0` turns it off; default `600` |
|
||||
| `PIXA_UPSTREAM_FETCH_TIMEOUT` | `upstream_fetch_timeout` | Time allowed for one fetch from an upstream host; default `30s` |
|
||||
| `PIXA_UPSTREAM_MAX_RESPONSE_SIZE` | `upstream_max_response_size` | Largest upstream response accepted, in bytes; default 50 MiB |
|
||||
@@ -607,10 +612,12 @@ Key settings in more detail:
|
||||
does, and `downstream_timeout` has not ended first, the request is answered
|
||||
503 with the error `server busy, try again later`
|
||||
- `max_concurrent_processing` — the most images decoded and encoded at once;
|
||||
default the number of CPUs pixa can use (`GOMAXPROCS`), which follows a
|
||||
container's CPU limit. A request that finds all of them in use waits up to 10
|
||||
seconds for one to free up; if none does, and `downstream_timeout` has not
|
||||
ended first, it is answered 503 the same way
|
||||
default half the CPUs pixa can use (`GOMAXPROCS`, which follows a container's
|
||||
CPU limit), rounded down, and at least 1. A value that is not a whole number
|
||||
of at least 1 aborts startup. A request that finds all of them in use waits up
|
||||
to 10 seconds for one to free up; if none does, and `downstream_timeout` has
|
||||
not ended first, it is answered 503 the same way. Each one can take a lot of
|
||||
memory: see "Running under upaas"
|
||||
- `image_requests_per_minute` — how many requests one client may send to the
|
||||
image routes per minute (see the limit on them under Routes); default `600`.
|
||||
`0` turns the limit off, and a negative or non-integer value aborts startup.
|
||||
|
||||
@@ -30,15 +30,16 @@ P2: security: per-origin rate limiting
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-15 a source whose first bytes do not match its `Content-Type` answers
|
||||
415 (closes #253): a body whose first bytes are not those of the format its
|
||||
`Content-Type` names, such as a JPEG sent as `image/png`, text, or a body
|
||||
under 12 bytes, answered 500 on both image routes and was logged as an error
|
||||
of pixa's, and each request for it fetched the whole body again. Both routes
|
||||
now answer 415, and the image service logs it at warning level as a refused
|
||||
source. It also gets a negative cache entry, as an upstream error status does,
|
||||
so the source URL is not fetched again for 5 minutes and answers 502
|
||||
meanwhile.
|
||||
- 2026-10-15 `max_concurrent_processing` defaults to half the CPUs pixa sees,
|
||||
rounded down, and at least 1 (closes #242): it defaulted to one image per CPU,
|
||||
and a container with no CPU limit sees every CPU of the host, so on a 48-CPU
|
||||
host pixa processed 48 images at once, each of which can take about 1.6 GiB.
|
||||
`imageprocessor.DefaultMaxConcurrentProcessing` gives the default to the
|
||||
config and to `imageprocessor.New`; `PIXA_MAX_CONCURRENT_PROCESSING` still
|
||||
overrides it. "Running under upaas" in `README.md` gives the worst-case
|
||||
memory, measured, and tells the operator to set
|
||||
`PIXA_MAX_CONCURRENT_PROCESSING` and `PIXA_UPSTREAM_CONNECTIONS` to fit the
|
||||
container's memory.
|
||||
- 2026-10-15 a `:` in the source path or query no longer moves where the path
|
||||
ends and the query begins (closes #247): the signed data and the cache key
|
||||
joined host, path and query with `:` and no escaping, so a URL signed for
|
||||
|
||||
@@ -108,10 +108,11 @@ upstream_connections_per_host: 20
|
||||
# downstream_timeout has ended first.
|
||||
upstream_connections: 64
|
||||
|
||||
# Maximum number of images decoded and encoded at once (default: the
|
||||
# number of CPUs pixa can use, which follows a container's CPU limit). A
|
||||
# request that finds none free waits up to 10 seconds for one, and if none
|
||||
# frees up it is answered 503, unless downstream_timeout has ended first.
|
||||
# Maximum number of images decoded and encoded at once (default: half the
|
||||
# number of CPUs pixa can use, at least 1; the number of CPUs follows a
|
||||
# container's CPU limit). A request that finds none free waits up to 10
|
||||
# seconds for one, and if none frees up it is answered 503, unless
|
||||
# downstream_timeout has ended first.
|
||||
# max_concurrent_processing: 4
|
||||
|
||||
# How many requests to the image routes (/v1/image/ and /v1/e/) one client
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"runtime"
|
||||
"testing"
|
||||
)
|
||||
@@ -12,8 +13,8 @@ const (
|
||||
)
|
||||
|
||||
// TestOmittedConcurrencyLimitsUseDefaults checks that an omitted
|
||||
// max_concurrent_processing is the number of CPUs Go uses and an omitted
|
||||
// upstream_connections is 64.
|
||||
// max_concurrent_processing is half the CPUs Go uses, and at least 1, and
|
||||
// an omitted upstream_connections is 64.
|
||||
func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -22,9 +23,9 @@ func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
|
||||
t.Fatalf("minimal config should be valid, got error: %v", err)
|
||||
}
|
||||
|
||||
if c.MaxConcurrentProcessing != runtime.GOMAXPROCS(0) {
|
||||
t.Errorf("MaxConcurrentProcessing = %d, want %d, one per CPU",
|
||||
c.MaxConcurrentProcessing, runtime.GOMAXPROCS(0))
|
||||
if c.MaxConcurrentProcessing != max(runtime.GOMAXPROCS(0)/2, 1) {
|
||||
t.Errorf("MaxConcurrentProcessing = %d, want %d, half the CPUs",
|
||||
c.MaxConcurrentProcessing, max(runtime.GOMAXPROCS(0)/2, 1))
|
||||
}
|
||||
|
||||
if c.UpstreamConnections != 64 {
|
||||
@@ -32,6 +33,36 @@ func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestOmittedMaxConcurrentProcessingIsHalfTheCPUsAndAtLeastOne checks
|
||||
// that an omitted max_concurrent_processing is half the CPUs Go uses,
|
||||
// rounded down, and 1 when Go uses one CPU.
|
||||
//
|
||||
//nolint:paralleltest // it sets GOMAXPROCS, which other tests read
|
||||
func TestOmittedMaxConcurrentProcessingIsHalfTheCPUsAndAtLeastOne(t *testing.T) {
|
||||
cpusBefore := runtime.GOMAXPROCS(0)
|
||||
|
||||
t.Cleanup(func() { runtime.GOMAXPROCS(cpusBefore) })
|
||||
|
||||
for _, tc := range []struct{ cpus, want int }{
|
||||
{1, 1}, {2, 1}, {3, 1}, {8, 4},
|
||||
} {
|
||||
runtime.GOMAXPROCS(tc.cpus)
|
||||
|
||||
c, err := configFromYAML(t, signingKeyLine)
|
||||
if err != nil {
|
||||
t.Fatalf("minimal config should be valid, got error: %v", err)
|
||||
}
|
||||
|
||||
t.Logf("GOMAXPROCS %d: MaxConcurrentProcessing %d",
|
||||
tc.cpus, c.MaxConcurrentProcessing)
|
||||
|
||||
if c.MaxConcurrentProcessing != tc.want {
|
||||
t.Errorf("GOMAXPROCS %d: MaxConcurrentProcessing = %d, want %d",
|
||||
tc.cpus, c.MaxConcurrentProcessing, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestExplicitConcurrencyLimitsAreUsed checks that valid values for the
|
||||
// two limits are used as given.
|
||||
func TestExplicitConcurrencyLimitsAreUsed(t *testing.T) {
|
||||
@@ -160,3 +191,19 @@ func TestConcurrencyLimitsFromEnvironment(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestInvalidMaxConcurrentProcessingFromEnvironmentAbortsStartup checks
|
||||
// that a PIXA_MAX_CONCURRENT_PROCESSING that is not a whole number of at
|
||||
// least 1, an empty one included, aborts startup naming the variable and
|
||||
// the key instead of taking the default.
|
||||
func TestInvalidMaxConcurrentProcessingFromEnvironmentAbortsStartup(t *testing.T) {
|
||||
for _, value := range []string{"", " ", "0.5", "half", "-1", "4 images"} {
|
||||
t.Run(fmt.Sprintf("value %q", value), func(t *testing.T) {
|
||||
t.Setenv(testMaxConcurrentProcessingVar, value)
|
||||
|
||||
_, err := configFromYAML(t, signingKeyLine)
|
||||
wantStartupError(t, err,
|
||||
testMaxConcurrentProcessingVar, keyMaxConcurrentProcessing)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,7 +12,6 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -21,6 +20,7 @@ import (
|
||||
|
||||
"git.eeqj.de/sneak/smartconfig"
|
||||
"go.uber.org/fx"
|
||||
"sneak.berlin/go/pixa/internal/imageprocessor"
|
||||
"sneak.berlin/go/pixa/internal/logger"
|
||||
)
|
||||
|
||||
@@ -297,10 +297,8 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
|
||||
keyUpstreamConnectionsPerHost, DefaultUpstreamConnectionsPerHost),
|
||||
UpstreamConnections: loader.intVal(
|
||||
keyUpstreamConnections, DefaultUpstreamConnections),
|
||||
// Decoding and encoding are CPU-bound, so the default is one image
|
||||
// per CPU Go uses, which follows a container's CPU limit.
|
||||
MaxConcurrentProcessing: loader.intVal(
|
||||
keyMaxConcurrentProcessing, runtime.GOMAXPROCS(0)),
|
||||
MaxConcurrentProcessing: loader.intVal(keyMaxConcurrentProcessing,
|
||||
imageprocessor.DefaultMaxConcurrentProcessing()),
|
||||
UpstreamFetchTimeout: loader.durationVal(
|
||||
keyUpstreamFetchTimeout, DefaultUpstreamFetchTimeout),
|
||||
UpstreamMaxResponseSize: loader.int64Val(
|
||||
|
||||
@@ -1,180 +0,0 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"image/color"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"slices"
|
||||
"testing"
|
||||
"testing/fstest"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"sneak.berlin/go/pixa/internal/encurl"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imgcache"
|
||||
)
|
||||
|
||||
// mismatchedSourcePath is the path of the source that
|
||||
// TestFirstBytesNotMatchingContentType serves on allowlistedHost. The mock
|
||||
// fetcher names a file's Content-Type from its extension, so the source is
|
||||
// sent as image/png.
|
||||
const mismatchedSourcePath = "/images/source.png"
|
||||
|
||||
// newRoutesFetchingThrough returns both image routes, and the Handlers behind
|
||||
// them, whose image service fetches through fetcher and has allowlistedHost on
|
||||
// its allowlist. The handlers and the image service log to log.
|
||||
func newRoutesFetchingThrough(
|
||||
t *testing.T, fetcher httpfetcher.Fetcher, log *slog.Logger,
|
||||
) (*Handlers, http.Handler) {
|
||||
t.Helper()
|
||||
|
||||
cache, err := imgcache.NewCache(setupTestDB(t), imgcache.CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
NegativeTTL: 5 * time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("imgcache.NewCache() error = %v", err)
|
||||
}
|
||||
|
||||
svc, err := imgcache.NewService(&imgcache.ServiceConfig{
|
||||
Cache: cache,
|
||||
Fetcher: fetcher,
|
||||
SigningKey: testSigningKey,
|
||||
Allowlist: []string{allowlistedHost},
|
||||
Logger: log,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("imgcache.NewService() error = %v", err)
|
||||
}
|
||||
|
||||
encGen, err := encurl.NewGenerator(testSigningKey)
|
||||
if err != nil {
|
||||
t.Fatalf("encurl.NewGenerator() error = %v", err)
|
||||
}
|
||||
|
||||
h := &Handlers{log: log, imgSvc: svc, encGen: encGen}
|
||||
|
||||
r := chi.NewRouter()
|
||||
r.Get("/v1/image/*", h.HandleImage())
|
||||
r.Get("/v1/e/{token}/*", h.HandleImageEnc())
|
||||
|
||||
return h, r
|
||||
}
|
||||
|
||||
// mismatchedSourcePlainURL returns the /v1/image/ URL of mismatchedSourcePath
|
||||
// on allowlistedHost at 50x50 as a JPEG.
|
||||
func mismatchedSourcePlainURL(*testing.T, *Handlers) string {
|
||||
return "/v1/image/" + allowlistedHost + mismatchedSourcePath + "/50x50.jpeg"
|
||||
}
|
||||
|
||||
// mismatchedSourceEncURL returns an encrypted /v1/e/ URL, made with h's
|
||||
// generator, of mismatchedSourcePath on allowlistedHost at 50x50 as a JPEG.
|
||||
func mismatchedSourceEncURL(t *testing.T, h *Handlers) string {
|
||||
t.Helper()
|
||||
|
||||
token, err := h.encGen.Generate(&encurl.Payload{
|
||||
SourceHost: allowlistedHost,
|
||||
SourcePath: mismatchedSourcePath,
|
||||
Width: 50,
|
||||
Height: 50,
|
||||
Format: imgcache.FormatJPEG,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Generate() error = %v", err)
|
||||
}
|
||||
|
||||
return "/v1/e/" + token + "/img.jpg"
|
||||
}
|
||||
|
||||
// loggedLevels reads the JSON log lines in logs and returns the messages
|
||||
// logged at each level.
|
||||
func loggedLevels(t *testing.T, logs *bytes.Buffer) map[string][]string {
|
||||
t.Helper()
|
||||
|
||||
t.Logf("logged:\n%s", logs.String())
|
||||
|
||||
messages := make(map[string][]string)
|
||||
|
||||
dec := json.NewDecoder(logs)
|
||||
for dec.More() {
|
||||
var line struct {
|
||||
Level string `json:"level"`
|
||||
Msg string `json:"msg"`
|
||||
}
|
||||
|
||||
err := dec.Decode(&line)
|
||||
if err != nil {
|
||||
t.Fatalf("decoding log line: %v", err)
|
||||
}
|
||||
|
||||
messages[line.Level] = append(messages[line.Level], line.Msg)
|
||||
}
|
||||
|
||||
return messages
|
||||
}
|
||||
|
||||
// TestFirstBytesNotMatchingContentType checks that both image routes answer
|
||||
// 415 for an upstream body whose first bytes are not those of the format its
|
||||
// Content-Type names, and log it as a refused source, not as an error of
|
||||
// pixa's. The next request for that source must answer 502 without fetching
|
||||
// it again, as after an upstream error status.
|
||||
func TestFirstBytesNotMatchingContentType(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
bodies := map[string][]byte{
|
||||
"JPEG sent as image/png": generateTestJPEG(t, 10, 10, color.Black),
|
||||
"text sent as image/png": []byte("This is just text, not a PNG"),
|
||||
"body too short to name a format": []byte("short"),
|
||||
}
|
||||
|
||||
// The source's URL on each image route.
|
||||
sourceURLs := map[string]func(*testing.T, *Handlers) string{
|
||||
"plain URL": mismatchedSourcePlainURL,
|
||||
"encrypted URL": mismatchedSourceEncURL,
|
||||
}
|
||||
|
||||
for bodyName, body := range bodies {
|
||||
for urlName, sourceURL := range sourceURLs {
|
||||
t.Run(urlName+", "+bodyName, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
fetcher := &countingFetcher{
|
||||
Fetcher: httpfetcher.NewMock(fstest.MapFS{
|
||||
allowlistedHost + mismatchedSourcePath: &fstest.MapFile{
|
||||
Data: body,
|
||||
},
|
||||
}),
|
||||
}
|
||||
|
||||
var logs bytes.Buffer
|
||||
|
||||
h, routes := newRoutesFetchingThrough(t, fetcher,
|
||||
slog.New(slog.NewJSONHandler(&logs, nil)))
|
||||
target := sourceURL(t, h)
|
||||
|
||||
checkErrorBody(t, sendGet(t, routes, target),
|
||||
http.StatusUnsupportedMediaType, "unsupported source format")
|
||||
|
||||
levels := loggedLevels(t, &logs)
|
||||
if len(levels[slog.LevelError.String()]) != 0 {
|
||||
t.Errorf("logged at error level: %q",
|
||||
levels[slog.LevelError.String()])
|
||||
}
|
||||
|
||||
if !slices.Contains(levels[slog.LevelWarn.String()], "source refused") {
|
||||
t.Error(`no "source refused" line logged at warning level`)
|
||||
}
|
||||
|
||||
checkErrorBody(t, sendGet(t, routes, target),
|
||||
http.StatusBadGateway, "upstream error")
|
||||
|
||||
if n := fetcher.fetches.Load(); n != 1 {
|
||||
t.Errorf("upstream fetched %d times, want 1", n)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -15,7 +15,6 @@ import (
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imageprocessor"
|
||||
"sneak.berlin/go/pixa/internal/imgcache"
|
||||
"sneak.berlin/go/pixa/internal/magic"
|
||||
)
|
||||
|
||||
// HandleImage handles the main image proxy route:
|
||||
@@ -221,18 +220,6 @@ func (s *Handlers) HandleTooManyRequests() http.HandlerFunc {
|
||||
func (s *Handlers) respondImageError(
|
||||
w http.ResponseWriter, req *imgcache.ImageRequest, err error,
|
||||
) {
|
||||
// An upstream body whose first bytes are not those of the format its
|
||||
// Content-Type names is the upstream's error, not pixa's: the image
|
||||
// service has logged it as a refused source.
|
||||
if errors.Is(err, magic.ErrMagicByteMismatch) ||
|
||||
errors.Is(err, magic.ErrUnknownFormat) ||
|
||||
errors.Is(err, magic.ErrNotEnoughData) {
|
||||
s.respondError(w, "unsupported source format",
|
||||
http.StatusUnsupportedMediaType)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
s.log.Error("failed to get image",
|
||||
"host", req.SourceHost,
|
||||
"path", req.SourcePath,
|
||||
|
||||
@@ -15,7 +15,6 @@ import (
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imageprocessor"
|
||||
"sneak.berlin/go/pixa/internal/imgcache"
|
||||
"sneak.berlin/go/pixa/internal/magic"
|
||||
)
|
||||
|
||||
// HandleImageEnc handles requests to /v1/e/{token}/* for encrypted
|
||||
@@ -163,10 +162,7 @@ func (s *Handlers) handleImageError(w http.ResponseWriter, err error) {
|
||||
s.respondError(w, "server busy, try again later",
|
||||
http.StatusServiceUnavailable)
|
||||
case errors.Is(err, httpfetcher.ErrInvalidContentType),
|
||||
errors.Is(err, imageprocessor.ErrUnsupportedInputFormat),
|
||||
errors.Is(err, magic.ErrMagicByteMismatch),
|
||||
errors.Is(err, magic.ErrUnknownFormat),
|
||||
errors.Is(err, magic.ErrNotEnoughData):
|
||||
errors.Is(err, imageprocessor.ErrUnsupportedInputFormat):
|
||||
s.respondError(w, "unsupported source format",
|
||||
http.StatusUnsupportedMediaType)
|
||||
case errors.Is(err, imageprocessor.ErrOutputTooLarge):
|
||||
|
||||
@@ -20,11 +20,11 @@ import (
|
||||
var vipsOnce sync.Once
|
||||
|
||||
// initVips initializes libvips with quiet logging, one worker thread per
|
||||
// image and no operation cache. Process already works on one image per CPU
|
||||
// by default, so more threads per image would only compete for the CPUs.
|
||||
// Each request decodes different source bytes, so the operation cache
|
||||
// would rarely be hit and would hold memory outside MaxConcurrentProcessing;
|
||||
// repeated requests are served from pixa's disk cache instead.
|
||||
// image and no operation cache. With one thread each, the images processed
|
||||
// at once use no more CPUs than MaxConcurrentProcessing. Each request
|
||||
// decodes different source bytes, so the operation cache would rarely be
|
||||
// hit and would hold memory outside MaxConcurrentProcessing; repeated
|
||||
// requests are served from pixa's disk cache instead.
|
||||
func initVips() {
|
||||
vipsOnce.Do(func() {
|
||||
vips.LoggingSettings(nil, vips.LogLevelError)
|
||||
@@ -179,10 +179,19 @@ type Params struct {
|
||||
// If <= 0, DefaultMaxInputBytes is used.
|
||||
MaxInputBytes int64
|
||||
// MaxConcurrentProcessing is the most images processed at once.
|
||||
// If <= 0, the number of CPUs Go uses (runtime.GOMAXPROCS(0)) is used.
|
||||
// If <= 0, DefaultMaxConcurrentProcessing() is used.
|
||||
MaxConcurrentProcessing int
|
||||
}
|
||||
|
||||
// DefaultMaxConcurrentProcessing returns the most images processed at once
|
||||
// when none is set: half the CPUs Go uses (runtime.GOMAXPROCS(0), which
|
||||
// follows a container's CPU limit), rounded down, and at least 1.
|
||||
func DefaultMaxConcurrentProcessing() int {
|
||||
const cpusPerImage = 2
|
||||
|
||||
return max(runtime.GOMAXPROCS(0)/cpusPerImage, 1)
|
||||
}
|
||||
|
||||
// New creates a new image processor with the given parameters.
|
||||
// A zero-value Params{} uses sensible defaults.
|
||||
func New(params Params) *ImageProcessor {
|
||||
@@ -195,7 +204,7 @@ func New(params Params) *ImageProcessor {
|
||||
|
||||
maxConcurrentProcessing := params.MaxConcurrentProcessing
|
||||
if maxConcurrentProcessing <= 0 {
|
||||
maxConcurrentProcessing = runtime.GOMAXPROCS(0)
|
||||
maxConcurrentProcessing = DefaultMaxConcurrentProcessing()
|
||||
}
|
||||
|
||||
return &ImageProcessor{
|
||||
|
||||
@@ -115,14 +115,14 @@ func waitForEntries(t *testing.T, entered <-chan struct{}, count int) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewDefaultsMaxConcurrentProcessingToCPUs(t *testing.T) {
|
||||
func TestNewDefaultsMaxConcurrentProcessingToHalfTheCPUs(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
for _, limit := range []int{0, -1} {
|
||||
proc := New(Params{MaxConcurrentProcessing: limit})
|
||||
if got := cap(proc.processingSemaphore); got != runtime.GOMAXPROCS(0) {
|
||||
t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, one per CPU",
|
||||
limit, got, runtime.GOMAXPROCS(0))
|
||||
if got := cap(proc.processingSemaphore); got != max(runtime.GOMAXPROCS(0)/2, 1) {
|
||||
t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, half the CPUs",
|
||||
limit, got, max(runtime.GOMAXPROCS(0)/2, 1))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -132,6 +132,30 @@ func TestNewDefaultsMaxConcurrentProcessingToCPUs(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestNewDefaultIsHalfTheCPUsAndAtLeastOne checks that New, given no
|
||||
// MaxConcurrentProcessing, makes one slot for every two CPUs Go uses,
|
||||
// rounded down, and one slot when Go uses one CPU.
|
||||
//
|
||||
//nolint:paralleltest // it sets GOMAXPROCS, which other tests read
|
||||
func TestNewDefaultIsHalfTheCPUsAndAtLeastOne(t *testing.T) {
|
||||
cpusBefore := runtime.GOMAXPROCS(0)
|
||||
|
||||
t.Cleanup(func() { runtime.GOMAXPROCS(cpusBefore) })
|
||||
|
||||
for _, tc := range []struct{ cpus, want int }{
|
||||
{1, 1}, {2, 1}, {3, 1}, {8, 4},
|
||||
} {
|
||||
runtime.GOMAXPROCS(tc.cpus)
|
||||
|
||||
got := cap(New(Params{}).processingSemaphore)
|
||||
t.Logf("GOMAXPROCS %d: %d slots", tc.cpus, got)
|
||||
|
||||
if got != tc.want {
|
||||
t.Errorf("GOMAXPROCS %d: %d slots, want %d", tc.cpus, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestProcessNeverExceedsMaxConcurrentProcessing starts more Process calls
|
||||
// than MaxConcurrentProcessing allows and holds the first ones inside
|
||||
// Process until the test lets them go. No more than the limit may be
|
||||
|
||||
@@ -52,7 +52,7 @@ type ServiceConfig struct {
|
||||
// Allowlist is the list of hosts that don't require signatures
|
||||
Allowlist []string
|
||||
// MaxConcurrentProcessing is the most images processed at once; zero
|
||||
// uses the image processor's default, one per CPU
|
||||
// uses the image processor's default, half the CPUs
|
||||
MaxConcurrentProcessing int
|
||||
// Logger for logging
|
||||
Logger *slog.Logger
|
||||
@@ -438,7 +438,13 @@ func (s *Service) fetchAndProcess(
|
||||
if err != nil {
|
||||
// Store negative cache for certain errors
|
||||
if isNegativeCacheable(err) {
|
||||
s.storeNegative(ctx, req, extractStatusCode(err), err)
|
||||
statusCode := extractStatusCode(err)
|
||||
|
||||
storeErr := s.cache.StoreNegative(ctx, req, statusCode, err.Error())
|
||||
if storeErr != nil {
|
||||
s.log.Warn("failed to store negative cache entry",
|
||||
"host", req.SourceHost, "path", req.SourcePath, "error", storeErr)
|
||||
}
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("upstream fetch failed: %w", err)
|
||||
@@ -484,16 +490,9 @@ func (s *Service) fetchAndProcess(
|
||||
"cipher", fetchResult.TLSCipherSuite,
|
||||
)
|
||||
|
||||
// Validate magic bytes match content type. A body that does not is the
|
||||
// upstream's error, so it gets a negative cache entry, as an upstream
|
||||
// error status does.
|
||||
// Validate magic bytes match content type
|
||||
err = magic.ValidateMagicBytes(sourceData, fetchResult.ContentType)
|
||||
if err != nil {
|
||||
s.log.Warn("source refused", "request_id", middleware.GetReqID(ctx),
|
||||
"host", req.SourceHost, "path", req.SourcePath,
|
||||
"content_type", fetchResult.ContentType, "error", err)
|
||||
s.storeNegative(ctx, req, httpStatusUnsupportedMediaType, err)
|
||||
|
||||
return nil, fmt.Errorf("content validation failed: %w", err)
|
||||
}
|
||||
|
||||
@@ -509,19 +508,6 @@ func (s *Service) fetchAndProcess(
|
||||
)
|
||||
}
|
||||
|
||||
// storeNegative stores a negative cache entry for req's source, whose fetch
|
||||
// failed with err, so the source is not fetched again until the entry
|
||||
// expires. A failure to store the entry is only logged.
|
||||
func (s *Service) storeNegative(
|
||||
ctx context.Context, req *ImageRequest, statusCode int, err error,
|
||||
) {
|
||||
storeErr := s.cache.StoreNegative(ctx, req, statusCode, err.Error())
|
||||
if storeErr != nil {
|
||||
s.log.Warn("failed to store negative cache entry",
|
||||
"host", req.SourceHost, "path", req.SourcePath, "error", storeErr)
|
||||
}
|
||||
}
|
||||
|
||||
// processedVariant is a variant as processAndStore made it. Each request that
|
||||
// shared its processing serves it through a reader of its own.
|
||||
type processedVariant struct {
|
||||
@@ -630,9 +616,8 @@ func signatureRequest(req *ImageRequest) *signature.Request {
|
||||
|
||||
// HTTP status codes for error responses.
|
||||
const (
|
||||
httpStatusBadGateway = 502
|
||||
httpStatusInternalError = 500
|
||||
httpStatusUnsupportedMediaType = 415
|
||||
httpStatusBadGateway = 502
|
||||
httpStatusInternalError = 500
|
||||
)
|
||||
|
||||
// isNegativeCacheable returns true if the error should be cached.
|
||||
|
||||
Reference in New Issue
Block a user