Compare commits
3 Commits
445ef57ada
...
af88877245
| Author | SHA1 | Date | |
|---|---|---|---|
| af88877245 | |||
| a13e5b7ded | |||
| bb30b3ad64 |
74
README.md
74
README.md
@@ -107,14 +107,60 @@ TTY detection, and security headers are always applied.
|
||||
| `DATA_DIR` | Directory for all SQLite databases | `/var/lib/webhooker` |
|
||||
| `DEBUG` | Enable debug logging | `false` |
|
||||
| `MAINTENANCE_MODE` | Report `maintenanceMode: true` in the healthcheck JSON. It does not change how any request is served — no maintenance page exists | `false` |
|
||||
| `METRICS_USERNAME` | Basic auth username for `/metrics` | `""` |
|
||||
| `METRICS_PASSWORD` | Basic auth password for `/metrics` | `""` |
|
||||
| `METRICS_USERNAME` | Basic auth username for `/metrics`. Must be set together with `METRICS_PASSWORD`; one without the other fails startup | `""` |
|
||||
| `METRICS_PASSWORD` | Basic auth password for `/metrics`. Must be set together with `METRICS_USERNAME`; one without the other fails startup | `""` |
|
||||
| `SENTRY_DSN` | Sentry error reporting DSN | `""` |
|
||||
| `RETENTION_SWEEP_INTERVAL` | How often the retention reaper and archive sweeper run (Go duration, must be positive) | `1h` |
|
||||
| `SESSION_IDLE_TIMEOUT` | Idle session timeout (Go duration) | `24h` |
|
||||
| `RECEIVER_RATE_LIMIT` | Receiver requests/minute per IP per entrypoint (10x that per IP across the route) | `120` |
|
||||
| `TRUSTED_PROXIES` | CIDRs whose forwarded headers are trusted (unset: all clients behind a proxy share one rate-limit bucket; a correct login password is never throttled either way) | `""` (none) |
|
||||
|
||||
#### Metrics credentials
|
||||
|
||||
`METRICS_USERNAME` and `METRICS_PASSWORD` are set together or not at
|
||||
all. With both set, `/metrics` is served behind basic auth. With
|
||||
neither set, the route is not registered and returns 404. With one set
|
||||
and the other empty or unset, the process refuses to start and exits
|
||||
non-zero with an error naming both variables — mounting the endpoint
|
||||
on the username alone would publish it behind a password that is the
|
||||
empty string, and quietly withholding it would deny an endpoint that
|
||||
was asked for. The `hasMetricsAuth` field in the startup log and the
|
||||
existence of the route are the same value, so they cannot disagree.
|
||||
|
||||
#### Single-instance lock
|
||||
|
||||
Exactly one webhooker process may use a `DATA_DIR` at a time. Two
|
||||
processes sharing one open the same databases and each run delivery
|
||||
recovery over the same rows, so every pending delivery goes out twice —
|
||||
duplicate delivery to your endpoints, from nothing worse than an
|
||||
overlapping deploy or a double start.
|
||||
|
||||
At startup, before anything opens a database, the process takes an
|
||||
exclusive advisory lock (`flock(2)`) on `{DATA_DIR}/webhooker.lock` and
|
||||
holds it for its lifetime. A second process pointed at the same
|
||||
directory prints a message naming it and exits non-zero:
|
||||
|
||||
```
|
||||
webhooker: data directory is already in use by another instance: /var/lib/webhooker (/var/lib/webhooker/webhooker.lock). Only one webhooker may use a data directory: two both run delivery recovery over the same rows and both deliver
|
||||
```
|
||||
|
||||
The lock is the kernel's, not the file's: it is released when the
|
||||
process exits, including `kill -9`, so a leftover `webhooker.lock`
|
||||
never blocks a restart and must not be deleted by hand. The file is
|
||||
also left in place on a clean shutdown, deliberately — unlinking it
|
||||
would let the next process lock a fresh inode while a third still held
|
||||
the old one.
|
||||
|
||||
To run two webhookers on one host, give each its own `DATA_DIR`.
|
||||
|
||||
`flock(2)` is host-local and per-inode: it arbitrates between processes
|
||||
and containers sharing a volume or bind mount on one machine, but not
|
||||
between hosts on a network filesystem, and a `DATA_DIR` inside a
|
||||
container's own writable layer is not shared with anything. On a
|
||||
filesystem that refuses `flock` outright, startup fails closed — the
|
||||
process reports the error and refuses to start rather than running
|
||||
unlocked.
|
||||
|
||||
#### Trusted proxies
|
||||
|
||||
`TRUSTED_PROXIES` is a comma-separated list of CIDR blocks (a bare
|
||||
@@ -283,10 +329,12 @@ is both the simplest and the only complete rule:
|
||||
|
||||
`{webhook_uuid}` is the webhook's UUID primary key in its canonical
|
||||
36-character hyphenated form, so a real filename looks like
|
||||
`events-3f2a1c9e-....db`. Nothing else is written to `DATA_DIR`, and no
|
||||
`-wal` or `-shm` files are produced (see below); a transient
|
||||
`{name}.db-journal` may exist beside a database while a write is in
|
||||
flight and is not part of the backup set.
|
||||
`events-3f2a1c9e-....db`. The only other file is `webhooker.lock`, the
|
||||
always-empty [single-instance lock](#single-instance-lock); it holds no
|
||||
state and is not part of the backup set — a copied one is stale and
|
||||
blocks nothing. No `-wal` or `-shm` files are produced (see below); a
|
||||
transient `{name}.db-journal` may exist beside a database while a write
|
||||
is in flight and is not part of the backup set either.
|
||||
|
||||
Configuration is **not** in `DATA_DIR` — it comes from the environment
|
||||
and from a `.env` file read out of the process working directory. Back
|
||||
@@ -1682,7 +1730,7 @@ abuse limit later; they are tracked as future work.
|
||||
|
||||
| Method | Path | Description |
|
||||
| ------ | ---------- | ----------- |
|
||||
| `GET` | `/metrics` | Prometheus metrics, behind basic auth. The route is registered only when `METRICS_USERNAME` is set; otherwise it does not exist and returns 404 |
|
||||
| `GET` | `/metrics` | Prometheus metrics, behind basic auth. The route is registered only when `METRICS_USERNAME` and `METRICS_PASSWORD` are both set; with neither set it does not exist and returns 404, and with only one set the process refuses to start |
|
||||
|
||||
#### API (Planned)
|
||||
|
||||
@@ -1709,7 +1757,7 @@ imports. The entry point is `cmd/webhooker/main.go`.
|
||||
```
|
||||
webhooker/
|
||||
├── cmd/webhooker/
|
||||
│ └── main.go # Entry point: sets globals, wires fx
|
||||
│ └── main.go # Entry point: sets globals, locks DATA_DIR, wires fx
|
||||
├── internal/
|
||||
│ ├── config/
|
||||
│ │ └── config.go # Configuration loading from environment variables
|
||||
@@ -1730,6 +1778,8 @@ webhooker/
|
||||
│ │ ├── retention.go # Retention reaper (per-webhook event expiry)
|
||||
│ │ ├── testing.go # NewTestDatabase: wrapper for tests, no fx lifecycle
|
||||
│ │ └── webhook_db_manager.go # Per-webhook DB lifecycle manager
|
||||
│ ├── datadir/
|
||||
│ │ └── lock.go # Exclusive advisory lock on DATA_DIR (one instance)
|
||||
│ ├── globals/
|
||||
│ │ └── globals.go # Build-time variables (appname, version, arch)
|
||||
│ ├── gormlog/
|
||||
@@ -1837,7 +1887,8 @@ Applied to all routes in this order:
|
||||
Permissions-Policy)
|
||||
3. **Logging** — Structured request logging (method, URL, status,
|
||||
latency, remote IP, user agent, request ID)
|
||||
4. **Metrics** — Prometheus HTTP metrics (if `METRICS_USERNAME` is set)
|
||||
4. **Metrics** — Prometheus HTTP metrics (if `METRICS_USERNAME` and
|
||||
`METRICS_PASSWORD` are both set)
|
||||
5. **CORS** — Cross-origin resource sharing headers
|
||||
6. **Timeout** — 60-second request timeout
|
||||
7. **Recoverer** — Panic recovery: one `ERROR` record through
|
||||
@@ -1869,8 +1920,9 @@ being read and without reaching CSRF, the route group's remaining
|
||||
middleware, or the handler. It is not rejected before *any* other
|
||||
middleware, though: the global entries listed above all run first, so
|
||||
such a request is still logged and given the security headers — and
|
||||
counted in the metrics, on a deployment where `METRICS_USERNAME` is
|
||||
set and the Metrics middleware is therefore registered at all. The
|
||||
counted in the metrics, on a deployment where the `/metrics`
|
||||
credentials are set and the Metrics middleware is therefore registered
|
||||
at all. The
|
||||
rejection itself is logged at `WARN` with the method, path and
|
||||
declared length. A chunked request, or
|
||||
one that lies about its length, is hard-capped by
|
||||
|
||||
@@ -2,11 +2,15 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"go.uber.org/fx"
|
||||
"sneak.berlin/go/webhooker/internal/config"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
"sneak.berlin/go/webhooker/internal/datadir"
|
||||
"sneak.berlin/go/webhooker/internal/delivery"
|
||||
"sneak.berlin/go/webhooker/internal/globals"
|
||||
"sneak.berlin/go/webhooker/internal/handlers"
|
||||
@@ -56,7 +60,32 @@ func main() {
|
||||
globals.Appname = appname
|
||||
globals.Version = version
|
||||
|
||||
os.Exit(run(os.Stderr))
|
||||
}
|
||||
|
||||
// run takes the exclusive DATA_DIR lock, then runs the application
|
||||
// under it, and returns the process exit status.
|
||||
//
|
||||
// The lock is taken here rather than inside the fx graph because it has
|
||||
// to be held before anything opens a database, and because a refusal
|
||||
// has to reach the operator as a plain line on standard error rather
|
||||
// than as one entry in an fx failure dump. It is released by the defer
|
||||
// on a clean shutdown, and by the kernel closing the descriptor on any
|
||||
// other exit — including the one fx performs itself when a start or
|
||||
// stop hook fails, which skips deferred calls.
|
||||
func run(stderr io.Writer) int {
|
||||
lock, err := datadir.Acquire(config.DataDir())
|
||||
if err != nil {
|
||||
_, _ = fmt.Fprintf(stderr, "%s: %v\n", appname, err)
|
||||
|
||||
return 1
|
||||
}
|
||||
|
||||
defer func() { _ = lock.Release() }()
|
||||
|
||||
newApp().Run()
|
||||
|
||||
return 0
|
||||
}
|
||||
|
||||
// newApp builds the application graph. It is separate from main so
|
||||
|
||||
@@ -1,10 +1,13 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/webhooker/internal/datadir"
|
||||
"sneak.berlin/go/webhooker/internal/server"
|
||||
)
|
||||
|
||||
@@ -33,6 +36,38 @@ func TestNewApp_StopTimeout(t *testing.T) {
|
||||
require.Less(t, got, dockerStopGrace)
|
||||
}
|
||||
|
||||
// TestRunRefusesLockedDataDir pins what an operator's second start
|
||||
// does. The entry point must refuse before it builds the fx graph —
|
||||
// nothing may open a database in a DATA_DIR another process holds —
|
||||
// and must exit non-zero with a message naming the directory rather
|
||||
// than starting a second delivery engine over the same rows.
|
||||
//
|
||||
// flock(2) locks descriptors independently, so holding the lock here
|
||||
// is the same denial a separate process gets; internal/datadir pins
|
||||
// that property and covers the real two-process case.
|
||||
func TestRunRefusesLockedDataDir(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("DATA_DIR", dir)
|
||||
|
||||
lock, err := datadir.Acquire(dir)
|
||||
require.NoError(t, err)
|
||||
|
||||
defer func() { _ = lock.Release() }()
|
||||
|
||||
var stderr bytes.Buffer
|
||||
|
||||
code := run(&stderr)
|
||||
|
||||
require.Equal(
|
||||
t, 1, code, "a second instance must exit non-zero",
|
||||
)
|
||||
assert.Contains(
|
||||
t, stderr.String(), dir,
|
||||
"the refusal must name the directory",
|
||||
)
|
||||
assert.Contains(t, stderr.String(), "another instance")
|
||||
}
|
||||
|
||||
// tailHeadroom is the slack the fx stop budget must keep beyond the
|
||||
// server stop hook. The hooks that run after the server — the
|
||||
// delivery engine, the healthcheck, the webhook DB manager and the
|
||||
|
||||
5
go.mod
5
go.mod
@@ -8,13 +8,14 @@ require (
|
||||
github.com/go-chi/chi v1.5.5
|
||||
github.com/go-chi/cors v1.2.1
|
||||
github.com/go-chi/httprate v0.15.0
|
||||
github.com/gofrs/flock v0.13.0
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/gorilla/csrf v1.7.3
|
||||
github.com/gorilla/sessions v1.4.0
|
||||
github.com/joho/godotenv v1.5.1
|
||||
github.com/prometheus/client_golang v1.18.0
|
||||
github.com/slok/go-http-metrics v0.11.0
|
||||
github.com/stretchr/testify v1.8.4
|
||||
github.com/stretchr/testify v1.11.1
|
||||
go.uber.org/fx v1.20.1
|
||||
golang.org/x/crypto v0.38.0
|
||||
gopkg.in/yaml.v3 v3.0.1
|
||||
@@ -49,7 +50,7 @@ require (
|
||||
go.uber.org/zap v1.23.0 // indirect
|
||||
golang.org/x/mod v0.17.0 // indirect
|
||||
golang.org/x/sync v0.14.0 // indirect
|
||||
golang.org/x/sys v0.33.0 // indirect
|
||||
golang.org/x/sys v0.37.0 // indirect
|
||||
golang.org/x/text v0.25.0 // indirect
|
||||
golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
|
||||
google.golang.org/protobuf v1.31.0 // indirect
|
||||
|
||||
14
go.sum
14
go.sum
@@ -23,6 +23,8 @@ github.com/go-chi/httprate v0.15.0 h1:j54xcWV9KGmPf/X4H32/aTH+wBlrvxL7P+SdnRqxh5
|
||||
github.com/go-chi/httprate v0.15.0/go.mod h1:rzGHhVrsBn3IMLYDOZQsSU4fJNWcjui4fWKJcCId1R4=
|
||||
github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA=
|
||||
github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og=
|
||||
github.com/gofrs/flock v0.13.0 h1:95JolYOvGMqeH31+FC7D2+uULf6mG61mEZ/A8dRYMzw=
|
||||
github.com/gofrs/flock v0.13.0/go.mod h1:jxeyy9R1auM5S6JYDBhDt+E2TCo7DkratH4Pgi8P+Z0=
|
||||
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
|
||||
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
|
||||
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
|
||||
@@ -81,11 +83,11 @@ github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncj
|
||||
github.com/slok/go-http-metrics v0.11.0 h1:ABJUpekCZSkQT1wQrFvS4kGbhea/w6ndFJaWJeh3zL0=
|
||||
github.com/slok/go-http-metrics v0.11.0/go.mod h1:ZGKeYG1ET6TEJpQx18BqAJAvxw9jBAZXCHU7bWQqqAc=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.5.1 h1:4VhoImhV/Bm0ToFkXFi8hXNXwpDRZ/ynw3amt82mzq0=
|
||||
github.com/stretchr/objx v0.5.1/go.mod h1:/iHQpkQwBD6DLUmQ4pE+s1TXdob1mORJ4/UFdrifcy0=
|
||||
github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
|
||||
github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
|
||||
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
|
||||
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
|
||||
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
|
||||
github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ=
|
||||
github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0=
|
||||
github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0=
|
||||
@@ -109,8 +111,8 @@ golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
|
||||
golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ=
|
||||
golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
|
||||
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
|
||||
golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
|
||||
golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
|
||||
golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
|
||||
golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4=
|
||||
golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA=
|
||||
golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg=
|
||||
|
||||
@@ -26,6 +26,10 @@ const (
|
||||
// EnvironmentProd represents production environment.
|
||||
EnvironmentProd = "prod"
|
||||
|
||||
// DefaultDataDir is where all SQLite databases live when DATA_DIR
|
||||
// is unset. The same default applies in every environment.
|
||||
DefaultDataDir = "/var/lib/webhooker"
|
||||
|
||||
// defaultPort is the default HTTP listen port.
|
||||
defaultPort = 8080
|
||||
|
||||
@@ -71,6 +75,16 @@ var ErrInvalidPort = errors.New("invalid port")
|
||||
// nor a bare IP address.
|
||||
var ErrInvalidCIDR = errors.New("invalid CIDR")
|
||||
|
||||
// ErrIncompleteMetricsAuth is returned when exactly one of
|
||||
// METRICS_USERNAME and METRICS_PASSWORD carries a value. Neither
|
||||
// fallback is acceptable: serving /metrics on the username alone
|
||||
// publishes an endpoint whose password is the empty string, and
|
||||
// silently leaving it unmounted withholds an endpoint the operator
|
||||
// asked for. Half-set is a configuration error, so startup fails.
|
||||
var ErrIncompleteMetricsAuth = errors.New(
|
||||
"incomplete metrics credentials",
|
||||
)
|
||||
|
||||
//nolint:revive // ConfigParams is a standard fx naming convention.
|
||||
type ConfigParams struct {
|
||||
fx.In
|
||||
@@ -128,12 +142,40 @@ func (c *Config) IsProd() bool {
|
||||
return c.Environment == EnvironmentProd
|
||||
}
|
||||
|
||||
// MetricsAuthEnabled reports whether /metrics is served behind basic
|
||||
// auth. It is the only answer to that question in the codebase: the
|
||||
// route mount, the Prometheus recording middleware and the startup
|
||||
// log's hasMetricsAuth field all read this one method, so the log
|
||||
// cannot report auth as off while the route is mounted.
|
||||
//
|
||||
// It requires both credentials rather than the username alone.
|
||||
// loadFromEnv already rejects a half-set pair, but a Config built in
|
||||
// code bypasses that, and the failure mode this guards is an endpoint
|
||||
// mounted with a credential map whose only password is the empty
|
||||
// string.
|
||||
func (c *Config) MetricsAuthEnabled() bool {
|
||||
return c.MetricsUsername != "" && c.MetricsPassword != ""
|
||||
}
|
||||
|
||||
// envString returns the value of the named environment variable,
|
||||
// or an empty string if not set.
|
||||
func envString(key string) string {
|
||||
return os.Getenv(key)
|
||||
}
|
||||
|
||||
// DataDir resolves DATA_DIR, applying DefaultDataDir when it is unset
|
||||
// or empty. It is exported so that entry points which must act on the
|
||||
// data directory before the fx graph exists — taking the exclusive
|
||||
// directory lock, above all — resolve it exactly as Config does.
|
||||
func DataDir() string {
|
||||
dir := envString("DATA_DIR")
|
||||
if dir == "" {
|
||||
return DefaultDataDir
|
||||
}
|
||||
|
||||
return dir
|
||||
}
|
||||
|
||||
// envBool returns the value of the named environment variable
|
||||
// parsed as a boolean. Returns defaultValue if not set. If the
|
||||
// variable is set but cannot be parsed, it returns a wrapped error
|
||||
@@ -329,6 +371,30 @@ func envPrefixList(key string) ([]netip.Prefix, error) {
|
||||
return prefixes, nil
|
||||
}
|
||||
|
||||
// resolveMetricsAuth reads the /metrics basic-auth credentials and
|
||||
// rejects a half-set pair, naming both variables either way. The
|
||||
// error carries neither value: the password is a secret.
|
||||
func resolveMetricsAuth() (string, string, error) {
|
||||
username := envString("METRICS_USERNAME")
|
||||
password := envString("METRICS_PASSWORD")
|
||||
|
||||
if (username == "") == (password == "") {
|
||||
return username, password, nil
|
||||
}
|
||||
|
||||
set, empty := "METRICS_USERNAME", "METRICS_PASSWORD"
|
||||
if username == "" {
|
||||
set, empty = empty, set
|
||||
}
|
||||
|
||||
return "", "", fmt.Errorf(
|
||||
"%w: %s is set but %s is empty; METRICS_USERNAME and "+
|
||||
"METRICS_PASSWORD must both be set to serve /metrics, "+
|
||||
"or both be empty to leave it unmounted",
|
||||
ErrIncompleteMetricsAuth, set, empty,
|
||||
)
|
||||
}
|
||||
|
||||
// resolveEnvironment reads WEBHOOKER_ENVIRONMENT, defaulting to
|
||||
// dev, and rejects unrecognised values.
|
||||
func resolveEnvironment() (string, error) {
|
||||
@@ -406,13 +472,18 @@ func loadFromEnv() (*Config, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
metricsUsername, metricsPassword, err := resolveMetricsAuth()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &Config{
|
||||
DataDir: envString("DATA_DIR"),
|
||||
DataDir: DataDir(),
|
||||
Debug: debug,
|
||||
MaintenanceMode: maintenanceMode,
|
||||
Environment: environment,
|
||||
MetricsUsername: envString("METRICS_USERNAME"),
|
||||
MetricsPassword: envString("METRICS_PASSWORD"),
|
||||
MetricsUsername: metricsUsername,
|
||||
MetricsPassword: metricsPassword,
|
||||
Port: port,
|
||||
SentryDSN: envString("SENTRY_DSN"),
|
||||
RetentionSweepInterval: retentionSweepInterval,
|
||||
@@ -485,14 +556,6 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
|
||||
s.log = log
|
||||
s.params = ¶ms
|
||||
|
||||
// Set default DataDir. All SQLite databases (main application
|
||||
// DB and per-webhook event DBs) live here. The same default is
|
||||
// used regardless of environment; override with DATA_DIR if
|
||||
// needed.
|
||||
if s.DataDir == "" {
|
||||
s.DataDir = "/var/lib/webhooker"
|
||||
}
|
||||
|
||||
if s.Debug {
|
||||
params.Logger.EnableDebugLogging()
|
||||
}
|
||||
@@ -512,8 +575,7 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
|
||||
"receiverRateLimit", s.ReceiverRateLimit,
|
||||
"trustedProxies", len(s.TrustedProxies),
|
||||
"hasSentryDSN", s.SentryDSN != "",
|
||||
"hasMetricsAuth",
|
||||
s.MetricsUsername != "" && s.MetricsPassword != "",
|
||||
"hasMetricsAuth", s.MetricsAuthEnabled(),
|
||||
)
|
||||
|
||||
s.warnSharedRateLimitBucket(log)
|
||||
|
||||
@@ -26,6 +26,12 @@ const (
|
||||
// cidrPrivateV4 is the sample trusted-proxy block the
|
||||
// TRUSTED_PROXIES cases are built from.
|
||||
cidrPrivateV4 = "10.0.0.0/8"
|
||||
|
||||
// metricsAuthValue is the sample METRICS_PASSWORD the metrics
|
||||
// credential cases are built from. It is asserted absent from
|
||||
// the startup error, so it must not be a substring of either
|
||||
// variable name that error prints.
|
||||
metricsAuthValue = "s3cret"
|
||||
)
|
||||
|
||||
func TestEnvironmentConfig(t *testing.T) {
|
||||
@@ -418,6 +424,36 @@ func TestDefaultDataDir(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestDataDirHelper pins the exported resolver against the value
|
||||
// Config carries. The entry point takes the DATA_DIR lock through the
|
||||
// helper before the fx graph exists, so the two disagreeing would mean
|
||||
// locking one directory and writing to another.
|
||||
func TestDataDirHelper(t *testing.T) {
|
||||
for _, set := range []string{"", "/tmp/webhooker-datadir-helper"} {
|
||||
name := "set"
|
||||
if set == "" {
|
||||
name = "unset"
|
||||
}
|
||||
|
||||
t.Run(name, func(t *testing.T) {
|
||||
// Cannot use t.Parallel() here because t.Setenv
|
||||
// is incompatible with parallel subtests.
|
||||
if set == "" {
|
||||
require.NoError(t, os.Unsetenv("DATA_DIR"))
|
||||
} else {
|
||||
t.Setenv("DATA_DIR", set)
|
||||
}
|
||||
|
||||
expected := set
|
||||
if expected == "" {
|
||||
expected = config.DefaultDataDir
|
||||
}
|
||||
|
||||
assert.Equal(t, expected, config.DataDir())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestReceiverRateLimit(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
@@ -726,3 +762,168 @@ func TestSharedRateLimitBucketWarning(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// metricsEnv describes what one subtest below puts in the
|
||||
// environment for a single METRICS_ variable. A variable that is
|
||||
// set to the empty string and one that is not set at all are
|
||||
// distinct inputs here, because the reported bug arrived through
|
||||
// the first of them.
|
||||
type metricsEnv struct {
|
||||
set bool
|
||||
value string
|
||||
}
|
||||
|
||||
// unset leaves the variable out of the environment entirely.
|
||||
func unset() metricsEnv {
|
||||
return metricsEnv{set: false, value: ""}
|
||||
}
|
||||
|
||||
// setTo sets the variable, including to the empty string.
|
||||
func setTo(value string) metricsEnv {
|
||||
return metricsEnv{set: true, value: value}
|
||||
}
|
||||
|
||||
// metricsAuthCase is one row of the table in TestMetricsAuthConfig,
|
||||
// named so the table can live in its own function and keep the test
|
||||
// itself short.
|
||||
type metricsAuthCase struct {
|
||||
name string
|
||||
username metricsEnv
|
||||
password metricsEnv
|
||||
expectError bool
|
||||
expectAuth bool
|
||||
}
|
||||
|
||||
// metricsAuthCases enumerates every combination of the two
|
||||
// credentials, counting "set to the empty string" and "not set at
|
||||
// all" as separate inputs on each side.
|
||||
func metricsAuthCases() []metricsAuthCase {
|
||||
return []metricsAuthCase{
|
||||
{
|
||||
name: "both unset leaves metrics unmounted",
|
||||
username: unset(),
|
||||
password: unset(),
|
||||
},
|
||||
{
|
||||
name: "both empty leaves metrics unmounted",
|
||||
username: setTo(""),
|
||||
password: setTo(""),
|
||||
},
|
||||
{
|
||||
name: "both set enables metrics auth",
|
||||
username: setTo("metrics"),
|
||||
password: setTo(metricsAuthValue),
|
||||
expectAuth: true,
|
||||
},
|
||||
{
|
||||
name: "username with unset password fails",
|
||||
username: setTo("metrics"),
|
||||
password: unset(),
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "username with empty password fails",
|
||||
username: setTo("metrics"),
|
||||
password: setTo(""),
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "password with unset username fails",
|
||||
username: unset(),
|
||||
password: setTo(metricsAuthValue),
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "password with empty username fails",
|
||||
username: setTo(""),
|
||||
password: setTo(metricsAuthValue),
|
||||
expectError: true,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// TestMetricsAuthConfig covers every combination of METRICS_USERNAME
|
||||
// and METRICS_PASSWORD. Either both carry a value, in which case
|
||||
// /metrics is served behind basic auth, or neither does, in which
|
||||
// case the route is never mounted. One without the other is a
|
||||
// startup error rather than a fallback: mounting on the username
|
||||
// alone published /metrics behind a credential map that accepted an
|
||||
// empty password, which is the defect this test exists to pin. See
|
||||
// https://git.eeqj.de/sneak/webhooker/issues/205.
|
||||
func TestMetricsAuthConfig(t *testing.T) {
|
||||
for _, tt := range metricsAuthCases() {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// Cannot use t.Parallel() here because t.Setenv
|
||||
// is incompatible with parallel subtests.
|
||||
if tt.username.set {
|
||||
t.Setenv("METRICS_USERNAME", tt.username.value)
|
||||
} else {
|
||||
require.NoError(
|
||||
t, os.Unsetenv("METRICS_USERNAME"),
|
||||
)
|
||||
}
|
||||
|
||||
if tt.password.set {
|
||||
t.Setenv("METRICS_PASSWORD", tt.password.value)
|
||||
} else {
|
||||
require.NoError(
|
||||
t, os.Unsetenv("METRICS_PASSWORD"),
|
||||
)
|
||||
}
|
||||
|
||||
if tt.expectError {
|
||||
assertMetricsAuthRejected(t)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
assertMetricsAuthAccepted(t, tt.expectAuth)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// assertMetricsAuthRejected requires that fx refused to build the
|
||||
// graph, that the failure is ErrIncompleteMetricsAuth, and that the
|
||||
// operator is told both variable names — the point of failing here
|
||||
// rather than degrading is that the message says what to fix.
|
||||
func assertMetricsAuthRejected(t *testing.T) {
|
||||
t.Helper()
|
||||
|
||||
var cfg *config.Config
|
||||
|
||||
app := fx.New(
|
||||
fx.NopLogger,
|
||||
fx.Provide(globals.New, logger.New, config.New),
|
||||
fx.Populate(&cfg),
|
||||
)
|
||||
|
||||
err := app.Err()
|
||||
require.Error(t, err)
|
||||
require.ErrorIs(t, err, config.ErrIncompleteMetricsAuth)
|
||||
assert.Contains(t, err.Error(), "METRICS_USERNAME")
|
||||
assert.Contains(t, err.Error(), "METRICS_PASSWORD")
|
||||
// The password is a secret and must not reach a startup error.
|
||||
assert.NotContains(t, err.Error(), metricsAuthValue)
|
||||
}
|
||||
|
||||
// assertMetricsAuthAccepted requires that startup succeeded and that
|
||||
// MetricsAuthEnabled — the single value the /metrics mount and the
|
||||
// startup log both read — reports what the environment asked for.
|
||||
func assertMetricsAuthAccepted(t *testing.T, expectAuth bool) {
|
||||
t.Helper()
|
||||
|
||||
var cfg *config.Config
|
||||
|
||||
app := fxtest.New(
|
||||
t,
|
||||
fx.Provide(globals.New, logger.New, config.New),
|
||||
fx.Populate(&cfg),
|
||||
)
|
||||
require.NoError(t, app.Err())
|
||||
|
||||
app.RequireStart()
|
||||
|
||||
defer app.RequireStop()
|
||||
|
||||
assert.Equal(t, expectAuth, cfg.MetricsAuthEnabled())
|
||||
}
|
||||
|
||||
126
internal/datadir/lock.go
Normal file
126
internal/datadir/lock.go
Normal file
@@ -0,0 +1,126 @@
|
||||
// Package datadir guards exclusive access to the directory holding
|
||||
// every SQLite database webhooker writes.
|
||||
//
|
||||
// Two processes sharing a DATA_DIR each open the same per-webhook
|
||||
// event databases and each run delivery recovery over the same rows,
|
||||
// so every pending delivery goes out twice. SQLite's own locking does
|
||||
// not prevent that: both writers are serialised correctly and both
|
||||
// deliver. The only thing that prevents it is refusing to be the
|
||||
// second process.
|
||||
//
|
||||
// The lock lives here rather than in the server's fx graph so that any
|
||||
// entry point which touches DATA_DIR — the server, or a CLI
|
||||
// subcommand that must not operate on a live deployment's data — takes
|
||||
// it the same way.
|
||||
package datadir
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"github.com/gofrs/flock"
|
||||
)
|
||||
|
||||
// LockFileName is the advisory lock file created inside DATA_DIR. Its
|
||||
// contents are never read: the lock is the flock(2) held on the open
|
||||
// descriptor, not the file's existence, so a leftover file from a
|
||||
// process that was killed with SIGKILL blocks nothing.
|
||||
const LockFileName = "webhooker.lock"
|
||||
|
||||
// dirPerm is the mode Acquire creates DATA_DIR with. It matches what
|
||||
// internal/database uses, since whichever runs first creates it.
|
||||
const dirPerm = 0o750
|
||||
|
||||
// ErrLocked reports that another live process holds the data
|
||||
// directory. Callers that need to know whether a deployment is running
|
||||
// — rather than merely failing to start — test for this with
|
||||
// errors.Is.
|
||||
var ErrLocked = errors.New(
|
||||
"data directory is already in use by another instance",
|
||||
)
|
||||
|
||||
// ErrNoDir reports that Acquire was given an empty directory.
|
||||
var ErrNoDir = errors.New("no data directory given")
|
||||
|
||||
// Lock is a held exclusive advisory lock on a data directory. It is
|
||||
// valid only while the process that took it lives: the kernel drops it
|
||||
// when the descriptor closes, whether that is Release, a normal exit,
|
||||
// or a SIGKILL.
|
||||
type Lock struct {
|
||||
dir string
|
||||
file *flock.Flock
|
||||
}
|
||||
|
||||
// Acquire takes the exclusive advisory lock on dir, creating dir if it
|
||||
// does not exist. It never waits: if another process holds the lock it
|
||||
// returns an error wrapping ErrLocked and naming dir.
|
||||
//
|
||||
// The returned Lock must be held for as long as the caller intends to
|
||||
// use dir.
|
||||
func Acquire(dir string) (*Lock, error) {
|
||||
if dir == "" {
|
||||
return nil, ErrNoDir
|
||||
}
|
||||
|
||||
err := os.MkdirAll(dir, dirPerm)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"creating data directory %s: %w", dir, err,
|
||||
)
|
||||
}
|
||||
|
||||
path := filepath.Join(dir, LockFileName)
|
||||
fl := flock.New(path)
|
||||
|
||||
held, err := fl.TryLock()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"locking data directory %s: %w", dir, err,
|
||||
)
|
||||
}
|
||||
|
||||
if !held {
|
||||
// A no-op on flock v0.13.0, which closes its own descriptor on
|
||||
// a failed TryLock; kept so no version can leak one.
|
||||
_ = fl.Close()
|
||||
|
||||
return nil, fmt.Errorf(
|
||||
"%w: %s (%s). Only one webhooker may use a data "+
|
||||
"directory: two both run delivery recovery over the "+
|
||||
"same rows and both deliver",
|
||||
ErrLocked, dir, path,
|
||||
)
|
||||
}
|
||||
|
||||
return &Lock{dir: dir, file: fl}, nil
|
||||
}
|
||||
|
||||
// Dir returns the locked directory.
|
||||
func (l *Lock) Dir() string {
|
||||
return l.dir
|
||||
}
|
||||
|
||||
// Path returns the lock file backing the lock.
|
||||
func (l *Lock) Path() string {
|
||||
return l.file.Path()
|
||||
}
|
||||
|
||||
// Release drops the lock and closes the descriptor. It is safe to call
|
||||
// more than once.
|
||||
//
|
||||
// The lock file is deliberately left on disk. Unlinking it would let
|
||||
// the next process create and lock a fresh inode while a third still
|
||||
// holds the old one, which is the one outcome this package exists to
|
||||
// prevent.
|
||||
func (l *Lock) Release() error {
|
||||
err := l.file.Unlock()
|
||||
if err != nil {
|
||||
return fmt.Errorf(
|
||||
"releasing lock on data directory %s: %w", l.dir, err,
|
||||
)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
250
internal/datadir/lock_test.go
Normal file
250
internal/datadir/lock_test.go
Normal file
@@ -0,0 +1,250 @@
|
||||
package datadir_test
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/webhooker/internal/datadir"
|
||||
)
|
||||
|
||||
// holderEnv names the directory the re-executed test binary should
|
||||
// lock and hold. When it is unset the child test does nothing, so an
|
||||
// ordinary run is unaffected.
|
||||
const holderEnv = "WEBHOOKER_DATADIR_LOCK_HOLDER"
|
||||
|
||||
// holderReadyPrefix labels the child's one-line report that it holds
|
||||
// the lock, so the parent can find it among the testing package's own
|
||||
// output on the same descriptor.
|
||||
const holderReadyPrefix = "DATADIR-LOCK-HELD "
|
||||
|
||||
// holderReadyTimeout bounds the wait for the child to take the lock.
|
||||
// It only has to cover process start on a loaded shared host.
|
||||
const holderReadyTimeout = 60 * time.Second
|
||||
|
||||
// holderHold is how long the child keeps the lock if nothing kills it.
|
||||
// A sleep rather than a bare block, so the runtime's deadlock detector
|
||||
// has a pending timer and the child cannot outlive a killed test run
|
||||
// by more than this.
|
||||
const holderHold = 10 * time.Minute
|
||||
|
||||
// TestLockHolder is the child half of the two-process tests below. It
|
||||
// takes the lock on the directory named by holderEnv, reports the lock
|
||||
// file on standard output, and then holds it until it is killed.
|
||||
func TestLockHolder(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := os.Getenv(holderEnv)
|
||||
if dir == "" {
|
||||
return
|
||||
}
|
||||
|
||||
lock, err := datadir.Acquire(dir)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Written to the descriptor directly: the parent reads fd 1, not
|
||||
// the testing package's buffered report.
|
||||
_, err = fmt.Fprintf(
|
||||
os.Stdout, "%s%s\n", holderReadyPrefix, lock.Path(),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
time.Sleep(holderHold)
|
||||
}
|
||||
|
||||
// startHolder re-executes this test binary as a separate process that
|
||||
// takes and holds the lock on dir, and returns once that process
|
||||
// actually holds it. The child is killed when the test ends.
|
||||
func startHolder(t *testing.T, dir string) *exec.Cmd {
|
||||
t.Helper()
|
||||
|
||||
//nolint:gosec // Re-executing this test binary, with a fixed arg.
|
||||
cmd := exec.CommandContext(
|
||||
t.Context(), os.Args[0], "-test.run", "^TestLockHolder$",
|
||||
)
|
||||
|
||||
cmd.Env = append(os.Environ(), holderEnv+"="+dir)
|
||||
cmd.Stderr = os.Stderr
|
||||
|
||||
stdout, err := cmd.StdoutPipe()
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, cmd.Start())
|
||||
|
||||
t.Cleanup(func() {
|
||||
_ = cmd.Process.Kill()
|
||||
_ = cmd.Wait()
|
||||
})
|
||||
|
||||
ready := make(chan string, 1)
|
||||
|
||||
go func() {
|
||||
scanner := bufio.NewScanner(stdout)
|
||||
for scanner.Scan() {
|
||||
after, found := strings.CutPrefix(
|
||||
scanner.Text(), holderReadyPrefix,
|
||||
)
|
||||
if found {
|
||||
ready <- after
|
||||
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
close(ready)
|
||||
|
||||
// Keep draining so the child never blocks on a full pipe.
|
||||
_, _ = io.Copy(io.Discard, stdout)
|
||||
}()
|
||||
|
||||
select {
|
||||
case path, ok := <-ready:
|
||||
require.True(
|
||||
t, ok, "holder exited without taking the lock",
|
||||
)
|
||||
require.Equal(t, filepath.Join(dir, datadir.LockFileName), path)
|
||||
case <-time.After(holderReadyTimeout):
|
||||
t.Fatal("timed out waiting for the holder to take the lock")
|
||||
}
|
||||
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TestSecondInstanceRefused is the regression test for the duplicate
|
||||
// delivery this package exists to prevent: a real second process
|
||||
// pointed at a data directory a live process already holds must be
|
||||
// refused, with an error that names the directory.
|
||||
func TestSecondInstanceRefused(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
|
||||
startHolder(t, dir)
|
||||
|
||||
lock, err := datadir.Acquire(dir)
|
||||
|
||||
require.Error(t, err, "the second instance took the lock too")
|
||||
require.Nil(t, lock)
|
||||
require.ErrorIs(
|
||||
t, err, datadir.ErrLocked,
|
||||
"the refusal must be distinguishable from any other failure",
|
||||
)
|
||||
assert.Contains(
|
||||
t, err.Error(), dir,
|
||||
"the refusal must name the directory it is about",
|
||||
)
|
||||
}
|
||||
|
||||
// TestRestartAfterHardKill is the other half of the regression: a
|
||||
// process killed with SIGKILL runs no cleanup and leaves its lock file
|
||||
// behind, and the next start must not be blocked by it. This is what a
|
||||
// pidfile would get wrong; the kernel drops a flock when the
|
||||
// descriptor closes, however the process died.
|
||||
func TestRestartAfterHardKill(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
|
||||
holder := startHolder(t, dir)
|
||||
|
||||
require.NoError(t, holder.Process.Kill())
|
||||
|
||||
// Wait for the kill to have actually happened. Re-acquiring while
|
||||
// the corpse still holds a descriptor would be a race, and would
|
||||
// make this test pass or fail on scheduling.
|
||||
_ = holder.Wait()
|
||||
|
||||
require.FileExists(
|
||||
t, filepath.Join(dir, datadir.LockFileName),
|
||||
"the stale lock file is what must not block the restart",
|
||||
)
|
||||
|
||||
lock, err := datadir.Acquire(dir)
|
||||
require.NoError(
|
||||
t, err, "a hard-killed instance must not block the next start",
|
||||
)
|
||||
|
||||
require.NoError(t, lock.Release())
|
||||
}
|
||||
|
||||
// TestSecondFdInSameProcessRefused pins the flock(2) property the
|
||||
// tests in cmd/webhooker rely on: descriptors are locked
|
||||
// independently, so a second acquisition is denied even when it comes
|
||||
// from the process that already holds the lock.
|
||||
func TestSecondFdInSameProcessRefused(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
|
||||
first, err := datadir.Acquire(dir)
|
||||
require.NoError(t, err)
|
||||
|
||||
defer func() { _ = first.Release() }()
|
||||
|
||||
_, err = datadir.Acquire(dir)
|
||||
require.ErrorIs(t, err, datadir.ErrLocked)
|
||||
}
|
||||
|
||||
// TestReleaseAllowsReacquire covers the clean-shutdown path: the lock
|
||||
// is released on exit, so a restart is not blocked by the previous
|
||||
// run.
|
||||
func TestReleaseAllowsReacquire(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
|
||||
first, err := datadir.Acquire(dir)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, first.Release())
|
||||
|
||||
second, err := datadir.Acquire(dir)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, second.Release())
|
||||
}
|
||||
|
||||
// TestAcquireCreatesDataDir covers a first start against a DATA_DIR
|
||||
// that does not exist yet, which is the normal case for a fresh
|
||||
// deployment: the lock is taken before anything else creates it.
|
||||
func TestAcquireCreatesDataDir(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := filepath.Join(t.TempDir(), "nested", "data")
|
||||
|
||||
lock, err := datadir.Acquire(dir)
|
||||
require.NoError(t, err)
|
||||
|
||||
defer func() { _ = lock.Release() }()
|
||||
|
||||
assert.Equal(t, dir, lock.Dir())
|
||||
assert.FileExists(t, filepath.Join(dir, datadir.LockFileName))
|
||||
}
|
||||
|
||||
// TestAcquireEmptyDir rejects an empty directory rather than locking
|
||||
// the process's working directory.
|
||||
func TestAcquireEmptyDir(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
_, err := datadir.Acquire("")
|
||||
require.ErrorIs(t, err, datadir.ErrNoDir)
|
||||
}
|
||||
|
||||
// TestAcquireUnusableDir reports an unusable DATA_DIR clearly, naming
|
||||
// it, instead of failing later and deeper.
|
||||
func TestAcquireUnusableDir(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
file := filepath.Join(t.TempDir(), "not-a-directory")
|
||||
require.NoError(t, os.WriteFile(file, nil, 0o600))
|
||||
|
||||
_, err := datadir.Acquire(file)
|
||||
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), file)
|
||||
}
|
||||
@@ -43,10 +43,7 @@ func (s *Server) serveUntilShutdown() {
|
||||
err := s.httpServer.ListenAndServe()
|
||||
if err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
s.log.Error("listen error", "error", err)
|
||||
|
||||
if s.cancelFunc != nil {
|
||||
s.cancelFunc()
|
||||
}
|
||||
s.shutdownOnListenFailure()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
95
internal/server/listen_failure_test.go
Normal file
95
internal/server/listen_failure_test.go
Normal file
@@ -0,0 +1,95 @@
|
||||
package server_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/fx"
|
||||
"sneak.berlin/go/webhooker/internal/globals"
|
||||
"sneak.berlin/go/webhooker/internal/server"
|
||||
)
|
||||
|
||||
// listenFailureDeadline is how long the app gets to give up after a
|
||||
// listen it cannot satisfy. The defect this pins left the process
|
||||
// reporting RUNNING for 183 seconds with nothing bound; a bind error
|
||||
// is known instantly, so anything past a moment here is that defect
|
||||
// back.
|
||||
const listenFailureDeadline = 2 * time.Second
|
||||
|
||||
// lifecycleTimeout bounds the app's start and stop sequences so a
|
||||
// wedged hook fails the test instead of hanging it.
|
||||
const lifecycleTimeout = 15 * time.Second
|
||||
|
||||
// TestListenFailure_ShutsDownTheApp pins that a listener the server
|
||||
// cannot bind terminates the application with a non-zero status.
|
||||
//
|
||||
// The fx OnStart hook returns as soon as the serving goroutine is
|
||||
// spawned, so a bind failure is discovered after fx has already
|
||||
// reported RUNNING. Nothing else in the graph observes it, and the
|
||||
// process used to stay alive with no listener: down, but indis-
|
||||
// tinguishable from healthy to systemd's Restart=on-failure and to
|
||||
// Docker's restart policies, which is the state this test exists to
|
||||
// keep from returning.
|
||||
//
|
||||
// The port is occupied by a listener this test holds open, on a
|
||||
// kernel-chosen port, so the failure is the real EADDRINUSE the
|
||||
// operator hits when a second instance starts. Loopback is enough to
|
||||
// collide with the server's wildcard bind: a listening socket on a
|
||||
// specific address blocks the wildcard from claiming the same port.
|
||||
func TestListenFailure_ShutsDownTheApp(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var listenCfg net.ListenConfig
|
||||
|
||||
occupied, err := listenCfg.Listen(
|
||||
t.Context(), "tcp", "127.0.0.1:0",
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
t.Cleanup(func() { _ = occupied.Close() })
|
||||
|
||||
addr, ok := occupied.Addr().(*net.TCPAddr)
|
||||
require.True(t, ok, "listener is not TCP")
|
||||
|
||||
// The collaborators come from the wired graph rather than stubs,
|
||||
// so the Server under test is the one that ships. Only the port
|
||||
// is test-specific.
|
||||
env := newTestEnv(t)
|
||||
env.cfg.Port = addr.Port
|
||||
|
||||
app := fx.New(
|
||||
fx.NopLogger,
|
||||
fx.Supply(env.log, env.cfg, env.mw, env.hnd),
|
||||
fx.Provide(globals.New, server.New),
|
||||
fx.Invoke(func(*server.Server) {}),
|
||||
)
|
||||
|
||||
startCtx, cancelStart := context.WithTimeout(
|
||||
context.Background(), lifecycleTimeout,
|
||||
)
|
||||
defer cancelStart()
|
||||
|
||||
require.NoError(t, app.Start(startCtx))
|
||||
|
||||
select {
|
||||
case sig := <-app.Wait():
|
||||
require.Equal(
|
||||
t, server.ListenFailureExitCode, sig.ExitCode,
|
||||
"listen failure must exit non-zero",
|
||||
)
|
||||
case <-time.After(listenFailureDeadline):
|
||||
t.Fatal("listen failure left the app running")
|
||||
}
|
||||
|
||||
// The stop sequence still has to complete: the fix must reach
|
||||
// shutdown through fx rather than around it.
|
||||
stopCtx, cancelStop := context.WithTimeout(
|
||||
context.Background(), lifecycleTimeout,
|
||||
)
|
||||
defer cancelStop()
|
||||
|
||||
require.NoError(t, app.Stop(stopCtx))
|
||||
}
|
||||
@@ -55,8 +55,11 @@ func (s *Server) setupGlobalMiddleware() {
|
||||
s.router.Use(s.mw.SecurityHeaders())
|
||||
s.router.Use(s.mw.Logging())
|
||||
|
||||
// Metrics middleware (only if credentials are configured)
|
||||
if s.params.Config.MetricsUsername != "" {
|
||||
// Metrics recording middleware, registered only when the
|
||||
// endpoint that exposes what it records is served. The
|
||||
// condition is the same MetricsAuthEnabled the /metrics mount
|
||||
// in setupRoutes reads.
|
||||
if s.params.Config.MetricsAuthEnabled() {
|
||||
s.router.Use(s.mw.Metrics())
|
||||
}
|
||||
|
||||
@@ -103,8 +106,14 @@ func (s *Server) setupRoutes() {
|
||||
s.h.HandleHealthCheck(),
|
||||
)
|
||||
|
||||
// set up authenticated /metrics route:
|
||||
if s.params.Config.MetricsUsername != "" {
|
||||
// Authenticated /metrics route. The condition is
|
||||
// Config.MetricsAuthEnabled and never the username alone: a
|
||||
// username with an empty password would otherwise mount the
|
||||
// endpoint behind a credential map that accepts an empty
|
||||
// password. Config rejects that combination at startup, and
|
||||
// this reads the same value the startup log reports, so the
|
||||
// two cannot disagree about whether the route exists.
|
||||
if s.params.Config.MetricsAuthEnabled() {
|
||||
s.router.Group(func(r chi.Router) {
|
||||
r.Use(s.mw.MetricsAuth())
|
||||
r.Get(
|
||||
|
||||
@@ -34,6 +34,13 @@ import (
|
||||
// the CSRF middleware executed.
|
||||
const csrfCookieName = "_gorilla_csrf"
|
||||
|
||||
const (
|
||||
// metricsUser and metricsAuthValue are the /metrics basic-auth
|
||||
// credentials the metrics routing tests below configure.
|
||||
metricsUser = "metrics"
|
||||
metricsAuthValue = "s3cret"
|
||||
)
|
||||
|
||||
type noopNotifier struct{}
|
||||
|
||||
func (n *noopNotifier) Notify([]delivery.Task) {}
|
||||
@@ -69,9 +76,23 @@ type testEnv struct {
|
||||
func newTestEnv(t *testing.T) *testEnv {
|
||||
t.Helper()
|
||||
|
||||
return newTestEnvWithConfig(t, &config.Config{
|
||||
DataDir: t.TempDir(),
|
||||
Environment: config.EnvironmentDev,
|
||||
})
|
||||
}
|
||||
|
||||
// newTestEnvWithConfig is newTestEnv over a caller-supplied Config,
|
||||
// for the routes whose existence the configuration decides. The same
|
||||
// pointer reaches the router and every middleware, so a test cannot
|
||||
// accidentally configure one and not the other.
|
||||
func newTestEnvWithConfig(
|
||||
t *testing.T, cfg *config.Config,
|
||||
) *testEnv {
|
||||
t.Helper()
|
||||
|
||||
var (
|
||||
log *logger.Logger
|
||||
cfg *config.Config
|
||||
mw *middleware.Middleware
|
||||
hnd *handlers.Handlers
|
||||
sess *session.Session
|
||||
@@ -84,12 +105,7 @@ func newTestEnv(t *testing.T) *testEnv {
|
||||
fx.Provide(
|
||||
globals.New,
|
||||
logger.New,
|
||||
func() *config.Config {
|
||||
return &config.Config{
|
||||
DataDir: t.TempDir(),
|
||||
Environment: config.EnvironmentDev,
|
||||
}
|
||||
},
|
||||
func() *config.Config { return cfg },
|
||||
database.New,
|
||||
database.NewWebhookDBManager,
|
||||
healthcheck.New,
|
||||
@@ -99,7 +115,7 @@ func newTestEnv(t *testing.T) *testEnv {
|
||||
middleware.New,
|
||||
handlers.New,
|
||||
),
|
||||
fx.Populate(&log, &cfg, &mw, &hnd, &sess, &db, &dbMgr),
|
||||
fx.Populate(&log, &mw, &hnd, &sess, &db, &dbMgr),
|
||||
)
|
||||
app.RequireStart()
|
||||
t.Cleanup(app.RequireStop)
|
||||
@@ -657,3 +673,119 @@ func TestSourceLogsBody_OtherUser404s(t *testing.T) {
|
||||
assert.Equal(t, http.StatusSeeOther, anon.Code)
|
||||
assert.Equal(t, "/pages/login", anon.Header().Get("Location"))
|
||||
}
|
||||
|
||||
// metricsConfig is a Config differing from the routing default only
|
||||
// in the two /metrics credentials.
|
||||
func metricsConfig(
|
||||
t *testing.T, username, password string,
|
||||
) *config.Config {
|
||||
t.Helper()
|
||||
|
||||
return &config.Config{
|
||||
DataDir: t.TempDir(),
|
||||
Environment: config.EnvironmentDev,
|
||||
MetricsUsername: username,
|
||||
MetricsPassword: password,
|
||||
}
|
||||
}
|
||||
|
||||
// metricsRequest asks the real router for /metrics with the given
|
||||
// basic-auth credentials, or with no Authorization header when
|
||||
// username is empty.
|
||||
func (e *testEnv) metricsRequest(
|
||||
username, password string,
|
||||
) *httptest.ResponseRecorder {
|
||||
req := httptest.NewRequestWithContext(
|
||||
context.Background(), http.MethodGet, "/metrics", nil,
|
||||
)
|
||||
|
||||
if username != "" {
|
||||
req.SetBasicAuth(username, password)
|
||||
}
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
e.router.ServeHTTP(w, req)
|
||||
|
||||
return w
|
||||
}
|
||||
|
||||
// TestMetricsRouteUnmountedWithoutCredentials pins that with neither
|
||||
// credential configured the route does not exist, which is the
|
||||
// documented behaviour and the only valid way for /metrics to be
|
||||
// absent.
|
||||
func TestMetricsRouteUnmountedWithoutCredentials(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
env := newTestEnvWithConfig(t, metricsConfig(t, "", ""))
|
||||
|
||||
assert.Equal(
|
||||
t, http.StatusNotFound,
|
||||
env.metricsRequest("", "").Code,
|
||||
)
|
||||
}
|
||||
|
||||
// TestMetricsRouteRequiresCredentials pins that with both credentials
|
||||
// configured the route exists and every request that does not carry
|
||||
// the configured pair is refused — including the empty password that
|
||||
// a half-set configuration used to make sufficient.
|
||||
func TestMetricsRouteRequiresCredentials(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
env := newTestEnvWithConfig(
|
||||
t, metricsConfig(t, metricsUser, metricsAuthValue),
|
||||
)
|
||||
|
||||
assert.Equal(
|
||||
t, http.StatusUnauthorized,
|
||||
env.metricsRequest("", "").Code,
|
||||
"no credentials must not reach the metrics handler",
|
||||
)
|
||||
assert.Equal(
|
||||
t, http.StatusUnauthorized,
|
||||
env.metricsRequest(metricsUser, "").Code,
|
||||
"an empty password must not reach the metrics handler",
|
||||
)
|
||||
assert.Equal(
|
||||
t, http.StatusUnauthorized,
|
||||
env.metricsRequest(metricsUser, "wrong").Code,
|
||||
)
|
||||
|
||||
ok := env.metricsRequest(metricsUser, metricsAuthValue)
|
||||
assert.Equal(t, http.StatusOK, ok.Code)
|
||||
assert.Contains(t, ok.Body.String(), "go_goroutines")
|
||||
}
|
||||
|
||||
// TestMetricsRouteUnmountedOnHalfSetConfig pins the defect from
|
||||
// https://git.eeqj.de/sneak/webhooker/issues/205 at the routing
|
||||
// layer. Config rejects a half-set pair at startup, so this Config
|
||||
// cannot be reached from the environment; the assertion is that the
|
||||
// route tree does not publish an endpoint accepting an empty
|
||||
// password even when handed one anyway, because the mount and the
|
||||
// startup log's hasMetricsAuth read the same value.
|
||||
func TestMetricsRouteUnmountedOnHalfSetConfig(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
username string
|
||||
password string
|
||||
}{
|
||||
{name: "username only", username: metricsUser},
|
||||
{name: "password only", password: metricsAuthValue},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cfg := metricsConfig(t, tc.username, tc.password)
|
||||
env := newTestEnvWithConfig(t, cfg)
|
||||
|
||||
assert.False(t, cfg.MetricsAuthEnabled())
|
||||
assert.Equal(
|
||||
t, http.StatusNotFound,
|
||||
env.metricsRequest(
|
||||
tc.username, tc.password,
|
||||
).Code,
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -50,6 +50,13 @@ const (
|
||||
minSentryFlush = 250 * time.Millisecond
|
||||
)
|
||||
|
||||
// ListenFailureExitCode is the status the process exits with when the
|
||||
// HTTP listener cannot be established, or dies for a reason other
|
||||
// than a requested shutdown. It must stay non-zero: systemd
|
||||
// `Restart=on-failure` and Docker's restart policies key off it, and a
|
||||
// zero exit would read as a deliberate stop.
|
||||
const ListenFailureExitCode = 1
|
||||
|
||||
// SentryFlushBudget reports how long the Sentry flush may run when
|
||||
// remaining is the time left on the fx stop context after the HTTP
|
||||
// drain. sentry.Flush takes a bare duration and honours no context,
|
||||
@@ -75,13 +82,13 @@ type ServerParams struct {
|
||||
Config *config.Config
|
||||
Middleware *middleware.Middleware
|
||||
Handlers *handlers.Handlers
|
||||
Shutdowner fx.Shutdowner
|
||||
}
|
||||
|
||||
// Server is the main HTTP server that wires up routes and manages
|
||||
// graceful shutdown.
|
||||
type Server struct {
|
||||
startupTime time.Time
|
||||
exitCode int
|
||||
sentryEnabled bool
|
||||
log *slog.Logger
|
||||
cancelFunc context.CancelFunc
|
||||
@@ -159,7 +166,12 @@ func (s *Server) enableSentry() {
|
||||
s.sentryEnabled = true
|
||||
}
|
||||
|
||||
func (s *Server) serve() int {
|
||||
// serve installs the signal watcher, starts the listener and blocks
|
||||
// until the server's context is cancelled. The process exit status is
|
||||
// fx's to decide — from a signal, or from the code
|
||||
// shutdownOnListenFailure hands the Shutdowner — so this reports
|
||||
// nothing back to its caller.
|
||||
func (s *Server) serve() {
|
||||
ctx, cancelFunc := context.WithCancel(context.Background())
|
||||
s.cancelFunc = cancelFunc
|
||||
|
||||
@@ -185,7 +197,30 @@ func (s *Server) serve() int {
|
||||
<-ctx.Done()
|
||||
// Shutdown is handled by the fx OnStop hook (cleanShutdown).
|
||||
// Do not call cleanShutdown() here to avoid double invocation.
|
||||
return s.exitCode
|
||||
}
|
||||
|
||||
// shutdownOnListenFailure ends the application after the HTTP
|
||||
// listener failed. The fx OnStart hook returns as soon as the serving
|
||||
// goroutine is spawned, so nothing downstream of it ever learns that
|
||||
// the listen failed: fx reports RUNNING and the process sits alive
|
||||
// with nothing bound, which is invisible to systemd and Docker
|
||||
// restart policies. Asking the Shutdowner to stop the app with a
|
||||
// non-zero code is what turns that into a visible failure.
|
||||
//
|
||||
// The context cancel that follows only unwinds serve()'s own wait.
|
||||
// The shutdown itself runs through fx's normal stop sequence, so the
|
||||
// clean-shutdown drain in cleanShutdown is reached unchanged.
|
||||
func (s *Server) shutdownOnListenFailure() {
|
||||
err := s.params.Shutdowner.Shutdown(
|
||||
fx.ExitCode(ListenFailureExitCode),
|
||||
)
|
||||
if err != nil {
|
||||
s.log.Error("shutdown request failed", "error", err)
|
||||
}
|
||||
|
||||
if s.cancelFunc != nil {
|
||||
s.cancelFunc()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) cleanupForExit() {
|
||||
@@ -193,9 +228,6 @@ func (s *Server) cleanupForExit() {
|
||||
}
|
||||
|
||||
func (s *Server) cleanShutdown(ctx context.Context) {
|
||||
// initiate clean shutdown
|
||||
s.exitCode = 0
|
||||
|
||||
ctxShutdown, shutdownCancel := context.WithTimeout(
|
||||
ctx, ShutdownTimeout,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user