mirror of
https://github.com/penpot/penpot.git
synced 2026-10-28 21:40:34 +00:00
✨ Make the CLIMIT queue visible with an accumulated gauge
The queue gauge was written by whichever limiter of a name ran last, so a per-profile pile-up never showed: a healthy profile writing 0 overwrote the 15 of the profile that was queueing (4 of 1441 samples in production), and a limiter whose requests never come back read 0 while 72 threads stayed parked. A queue is an entry and an exit, so the new gauge is accumulated: one counter per limiter name and instance, up when a call starts waiting for a permit, down when it stops, so a call that never leaves stays visible. It is always written with `:val`, and a daemon sampler republishes every count every 15s so a name whose traffic died does not go stale in the scrape; halting the component stops that thread. Rejections had no metric at all: penpot_rpc_climit_rejected_total counts them by name and reason (queue-full or timeout), and the log lines now carry queue-max, permits, code and reason, so an operator can tell whether the limiter did its job or was saturated. The gauges that could not tell the truth are gone: queue and permits (same last-write-wins problem, and waiting plus the wait histogram replace them) and http_worker_busy_threads (reads 0 on an instance with 72 blocked threads; pool_size is the signal). No other dashboard or alert rule reads them. AI-assisted-by: deepseek-v4.1-flash
This commit is contained in:
Andrey Antukh
parent
dd6cc835ef
commit
5c476b4d9c
@ -4,7 +4,7 @@ Backend: JVM Clojure; Integrant; PostgreSQL; Redis/Valkey; RPC; HTTP; storage; m
|
||||
|
||||
## Focused memories
|
||||
|
||||
- RPC, DB helpers, workers, cron: `mem:backend/rpc-db-worker-subtleties`
|
||||
- RPC, DB helpers, workers, cron, CLIMIT: `mem:backend/subtleties`
|
||||
- Storage abstraction, logical buckets, object lifecycle, deduplication, access, and garbage collection: `mem:backend/storage`.
|
||||
- HTTP sessions, config, media processing, and file data persistence: `mem:backend/http-storage-filedata-subtleties`.
|
||||
- Auth flows, permission model, teams, projects, invitations, comments, webhooks, audit: `mem:backend/auth-permissions-product-domains`
|
||||
@ -45,13 +45,13 @@ Database migrations live in `backend/src/app/migrations/`; pure SQL migrations a
|
||||
For interactive PostgreSQL access with correct dev defaults, use `scripts/psql`; to dump
|
||||
the current DDL schema, use `scripts/db-schema` (see `mem:scripts/psql`).
|
||||
|
||||
For deeper details on transaction semantics, advisory locks, Transit vs JSON helpers, and dev/test DB URLs: `mem:backend/rpc-db-worker-subtleties`.
|
||||
For deeper details on transaction semantics, advisory locks, Transit vs JSON helpers, and dev/test DB URLs: `mem:backend/subtleties`.
|
||||
|
||||
## Background tasks
|
||||
|
||||
A task handler is an Integrant component with `ig/assert-key`, `ig/expand-key`, and `ig/init-key`, returning the function run by the worker. New tasks also need wiring in `app.main`: handler config, worker registry entry, and cron entry if scheduled.
|
||||
|
||||
For worker dispatch, cron, retry semantics, deduplication, and queue internals: `mem:backend/rpc-db-worker-subtleties`.
|
||||
For worker dispatch, cron, retry semantics, deduplication, and queue internals: `mem:backend/subtleties`.
|
||||
|
||||
## REPL
|
||||
|
||||
|
||||
@ -1,32 +1,5 @@
|
||||
# Backend RPC/DB/Worker Subtleties
|
||||
# Backend RPC/DB/Worker Subtleties (superseded)
|
||||
|
||||
## RPC exposure and wrappers
|
||||
This memory was superseded by `mem:backend/subtleties`, which holds everything it had (RPC wrappers, DB helpers, workers and cron) plus the CLIMIT notes, config/HTTP middleware, http self-metrics, storage and file data. Read that one.
|
||||
|
||||
- RPC commands are discovered from vars created by `app.util.services/defmethod`; adding a command namespace is not enough unless `backend/src/app/rpc.clj` includes it in `resolve-methods`.
|
||||
- `GET`/`HEAD` RPC calls are only allowed for method names starting with `get-`. Other methods are method-not-allowed even if they are read-only internally.
|
||||
- RPC auth defaults to enabled. Public endpoints must set `::auth false` metadata explicitly.
|
||||
- The wrapper stack does auth before params validation, then auditing/rate/concurrency/metrics/retry/condition handling, with DB transaction handling inside that stack. `::db/transaction` metadata controls transaction wrapping.
|
||||
- Params with `::sm/params` are decoded/conformed through the JSON transformer and successful IObj results get `:encode/json` metadata. Legacy spec conforming only applies when no Malli params schema exists.
|
||||
- Nil RPC bodies become HTTP 204 unless explicit status metadata is present. Stream bodies default to `application/octet-stream` when no content type is set.
|
||||
- `::climit/id` is either a plain keyword, which gives one shared limiter per process, or the vector form `[[id key-fn]]`, which gives one limiter per key. `climit/wrap` takes the key only from that positional `key-fn` (`climit/get-limits` maps a plain keyword to a single-element vector), so `::climit/key-fn` is metadata that nothing reads: a command declaring `::climit/id :x/by-profile` next to it ends up with a single shared limiter, and the missing key is visible in the `app.rpc.climit` log lines, whose `id` carries no key suffix. That is what happened to `push-audit-events`, the last user of the dead metadata, from `a5c6d78ee5` (2024-01-27, "fix some fundamental bugs on climit module", which moved the key to the vector form without updating it) until it was fixed; nothing declares `::climit/key-fn` any more, so keep using the vector form for anything that must be per profile.
|
||||
- The `penpot_rpc_climit_queue`/`permits` gauges carry only the limiter name, never the key, so every per-profile limiter of the same name collapses into one series and the last call wins: a per-key pile-up is invisible there and a healthy looking `cola=0` proves nothing. A stuck limiter does show up in `penpot_rpc_climit_timing_count` (the acquired path), which is a counter: compare the same limiter name across instances and the stuck one reads 0 acquisitions while its peer reads thousands. In thread dumps it reads as threads parked in `psm/acquire` whose CPU time does not grow between two dumps, on an instance that is otherwise idle.
|
||||
- Nothing in the limiter path has a timeout unless the config sets one, and `psm/acquire` then calls `Semaphore.acquire`, which parks the calling thread forever. A parked thread still counts as in-system for every outer limiter of its chain, so the outermost one (the method's own by-profile limiter) stops granting permits and answers 429 to everything. A limiter without `:queue` gets an unlimited queue and therefore never rejects: its pile-up is silent, with no 429 and no log line except the WARN `enqueued`/`acquired` lines of the keyed id.
|
||||
- Deadlock shape between a command's own limits and the root ones (fixed): a command declaring its own limits held `process-image/by-profile` before `root/by-profile`, while a command with no limits of its own takes the roots first and then re-enters with the same keyed limits from inside its handler (`create-file-media-object-from-url`, `upload-photo`) and waits for them holding a root permit. Two calls of one profile then held the same two limiters inverted and parked for good: 69 threads at 0 CPU, an audit limiter with 0 acquisitions in 24h and 100% 429s, on an instance that never served a 204.
|
||||
- `climit/wrap` now appends the root limiters after the command's own, so `root/global`/`root/by-profile` end up outermost and every path meets them first; the nested `climit/invoke!` chains carry only the command's own limits and run inside them, which puts both paths in the same order. `impl-invoke` also keeps, per thread, the set of limiters that thread is inside of and runs the handler inline when a call asks for one it already holds, instead of queueing behind itself. That check relies on `pbh/invoke!` running the handler on the calling thread; the mark is removed in a `finally`. Commands declare the by-profile limit before the global one, which makes it the outermost of the pair: keep that declaration order, the reverse one would put two paths back in inverted order. The two nested call sites (`create-file-media-object-from-url`, `upload-photo`) still share `process-image` with the enclosing call instead of getting a limiter of their own, so the comment there about keeping the download out of that limit no longer holds.
|
||||
- The shipped `backend/resources/climit.edn` sets a timeout on every limiter, 60s for operations that can be slow and 30s otherwise, with a test that fails when an entry ships without one, and `init-key` logs a WARN for every entry without a timeout. The production file lives outside the repository: mirror it there, and remember that a parked thread only goes away with a restart or a deploy.
|
||||
|
||||
## DB helpers
|
||||
|
||||
- Most `app.db` helpers accept a pool, connection, or map containing `::db/pool` / `::db/conn`; preserve that convention in shared code.
|
||||
- `db/tx-run!` uses `next.jdbc.transaction/*nested-tx* :ignore`: nested transaction calls reuse the outer transaction, not a savepoint. Use explicit savepoints when nested rollback semantics matter.
|
||||
- `db/run!` opens/reuses one connection but does not create a transaction.
|
||||
- `db/tjson` is Transit JSON for jsonb storage; `db/json` is plain JSON. Worker task props use Transit and are decoded with `decode-transit-pgobject`.
|
||||
- Advisory transaction locks accept UUIDs or ints. UUID locks are hashed using a zero-UUID seeded siphash.
|
||||
|
||||
## Workers and cron
|
||||
|
||||
- Task queues are tenant-prefixed. Submit dedupe only removes not-yet-due `new` tasks with the same name/queue/label; it does not dedupe due, scheduled, retry, running, or completed work.
|
||||
- The dispatcher selects `new`/`retry` tasks with `FOR UPDATE SKIP LOCKED`, marks them `scheduled`, and publishes Redis payload `[id scheduled-at]`. The runner skips Redis messages whose scheduled timestamp no longer matches DB state.
|
||||
- Lost `scheduled` tasks are rescheduled after about 5 minutes; `running` tasks older than about 24 hours are marked failed as orphans.
|
||||
- A task handler that is missing or returns an invalid result currently defaults to completed after warning. Throwing with `ex-data :type ::retry` controls retry behavior; `:strategy ::noop` retries without incrementing retry count.
|
||||
- Cron jobs lock their `scheduled_task` row with `FOR UPDATE SKIP LOCKED`, disable statement/idle-in-transaction timeouts locally, and reschedule themselves in `finally` unless interrupted. Worker, dispatcher, and cron components do not start when the DB pool is read-only.
|
||||
Nothing should point here any more.
|
||||
|
||||
@ -10,6 +10,17 @@
|
||||
- Params schemas are open by default, so undeclared client keys reach the handler unless the map is `:closed true`. Creation commands (`create-file`, `create-project`, `create-team`, `create-team-with-invitations`, `upload-file-media-object`, `create-file-media-object-from-url`, `assemble-file-media-object`) use closed schemas: a client-provided `:id` fails with `:params-validation`. Their internal creation functions still accept an optional explicit `:id` for imports, duplicates and deterministic test fixtures.
|
||||
- Nil RPC bodies become HTTP 204 unless explicit status metadata is present. Stream bodies default to `application/octet-stream` when no content type is set.
|
||||
|
||||
## Concurrency limiter (CLIMIT)
|
||||
|
||||
- `::climit/id` is either a plain keyword, which gives one shared limiter per process, or the vector form `[[id key-fn]]`, which gives one limiter per key. `climit/wrap` takes the key only from that positional `key-fn` (`climit/get-limits` maps a plain keyword to a single-element vector), so `::climit/key-fn` is metadata that nothing reads: a command declaring `::climit/id :x/by-profile` next to it ends up with a single shared limiter, and the missing key is visible in the `app.rpc.climit` log lines, whose `id` carries no key suffix. That is what happened to `push-audit-events`, the last user of the dead metadata, from `a5c6d78ee5` (2024-01-27, "fix some fundamental bugs on climit module", which moved the key to the vector form without updating it) until it was fixed; nothing declares `::climit/key-fn` any more, so keep using the vector form for anything that must be per profile.
|
||||
- A gauge written by whichever limiter of a name ran last cannot show a queue: every per-profile limiter of the name collapses into one series, so a healthy profile writing 0 overwrites the 15 of the one that is queueing (measured in production: 4 of 1441 samples). `penpot_rpc_climit_waiting` replaces it and is accumulated: one counter per name and instance, up when a call starts waiting for a permit, down when it stops, so a call that enters and never leaves keeps the value above zero for ever. `penpot_rpc_climit_rejected_total` counts rejections by name and `reason`: `queue-full` when the error carries `:size`, `timeout` when it carries `:timeout` (both carry `:code :capacity-limit-reached`, so the timeout field is the discriminator). The two gauges that could not tell the truth (`penpot_rpc_climit_queue`, `penpot_rpc_climit_permits`) were deleted, and so was `penpot_http_worker_busy_threads`, which reads 0 on an instance with 72 blocked threads: `penpot_http_worker_pool_size` is the signal for retained threads. A stuck limiter also shows up in `penpot_rpc_climit_timing_count` (the acquired path), which is a counter: compare the same limiter name across instances and the stuck one reads 0 acquisitions while its peer reads thousands. In thread dumps it reads as threads parked in `psm/acquire` whose CPU time does not grow between two dumps, on an instance that is otherwise idle.
|
||||
- The waiting counters are a field of the `ConcurrencyLimiter` component (a `ConcurrentHashMap` of name to `AtomicLong`), so they start at zero, die with the component, and the sampler that closes over that map republishes only its own names every 15 s; the sampler stops with the component in `halt-key!`. `impl-invoke` pairs one decrement with one increment through a per-call flag: the flag turns a second call into a no-op and the `finally` guarantees that there is one. The increment sits inside the `try` and the decrement lives only in the acquired callback and the `finally`, so a handler that throws, a rejection and a throwable from the publishing path all leave the count where it was and the thread without the limiter marked as held (a leftover mark would skip that limiter for that key for ever). Concurrent `inc`/`dec` publications of two threads can land out of order and the value can read low for a moment; the republish corrects it.
|
||||
- The `app.rpc.climit` log lines carry `queue`, `queue-max`, `permits`, `elapsed`, `code` and `reason`, and their `id` carries the key (`name/<profile or key>`). That is the only place where the profile behind a queue is visible: a pile-up is chased there, on demand, not in the metrics. The `queue` of the line is the calls inside the limiter minus its permits, so it reads 0 for a lone caller that times out: that line goes out at TRACE instead of WARN, and the rejection count has to come from the metric.
|
||||
- Nothing in the limiter path has a timeout unless the config sets one, and `psm/acquire` then calls `Semaphore.acquire`, which parks the calling thread forever. A parked thread still counts as in-system for every outer limiter of its chain, so the outermost one (the method's own by-profile limiter) stops granting permits and answers 429 to everything. A limiter without `:queue` gets an unlimited queue and therefore never rejects: its pile-up is silent, with no 429 and no log line except the WARN `enqueued`/`acquired` lines of the keyed id.
|
||||
- Deadlock shape between a command's own limits and the root ones (fixed): a command declaring its own limits held `process-image/by-profile` before `root/by-profile`, while a command with no limits of its own takes the roots first and then re-enters with the same keyed limits from inside its handler (`create-file-media-object-from-url`, `upload-photo`) and waits for them holding a root permit. Two calls of one profile then held the same two limiters inverted and parked for good: 69 threads at 0 CPU, an audit limiter with 0 acquisitions in 24h and 100% 429s, on an instance that never served a 204.
|
||||
- `climit/wrap` now appends the root limiters after the command's own, so `root/global`/`root/by-profile` end up outermost and every path meets them first; the nested `climit/invoke!` chains carry only the command's own limits and run inside them, which puts both paths in the same order. `impl-invoke` also keeps, per thread, the set of limiters that thread is inside of and runs the handler inline when a call asks for one it already holds, instead of queueing behind itself. That check relies on `pbh/invoke!` running the handler on the calling thread; the mark is removed in a `finally`. Commands declare the by-profile limit before the global one, which makes it the outermost of the pair: keep that declaration order, the reverse one would put two paths back in inverted order. The two nested call sites (`create-file-media-object-from-url`, `upload-photo`) still share `process-image` with the enclosing call instead of getting a limiter of their own, so the comment there about keeping the download out of that limit no longer holds.
|
||||
- The shipped `backend/resources/climit.edn` sets a timeout on every limiter, 60s for operations that can be slow and 30s otherwise, with a test that fails when an entry ships without one, and `init-key` logs a WARN for every entry without a timeout. The production file lives outside the repository: mirror it there, and remember that a parked thread only goes away with a restart or a deploy.
|
||||
|
||||
## DB helpers
|
||||
|
||||
- Most `app.db` helpers accept a pool, connection, or map containing `::db/pool` / `::db/conn`; preserve that convention in shared code.
|
||||
@ -41,8 +52,9 @@
|
||||
## HTTP server self-metrics
|
||||
|
||||
- `app.http` enables Undertow connection statistics via the yetti option `:server/statistics` (requires yetti ≥ v11.11, which exposes it; before the patch `ListenerInfo#getConnectorStatistics` returned `nil`).
|
||||
- A daemon sampler built on `promesa.exec` (`px/scheduled-executor` plus a self-rescheduling `px/schedule` chain, so a failing sample never cancels the next one; 15 s, started with the server in `ig/init-key` and stopped with `px/shutdown-now` in `halt-key!`) publishes worker and listener state: `penpot_http_worker_queue_size`, `busy_threads`, `pool_size`, `max_pool_size`, `penpot_http_connector_active_connections`, `requests_total`, `errors_total`. Definitions live in `app.main/default-metrics`.
|
||||
- The xnio worker MXBean can return transient `-1` (e.g. busy-thread count); negative samples are discarded (gauge keeps its previous value). Undertow exposes absolute request/error totals, so the sampler keeps a watermark atom and publishes deltas; a counter reset (decreasing totals) skips the negative delta and moves the watermark forward.
|
||||
- `app.metrics/create-sampler` holds the sampler shape: a single daemon `px/scheduled-executor` thread with a self-rescheduling `px/schedule` chain, so a failing sample never cancels the next one; the owner starts it in `ig/init-key` and stops it with `px/shutdown-now` in its `halt-key!`. Http and CLIMIT both sample every 15 s.
|
||||
- `app.http` uses it to publish worker and listener state: `penpot_http_worker_queue_size`, `penpot_http_worker_pool_size`, `penpot_http_worker_max_pool_size`, `penpot_http_connector_active_connections`, `penpot_http_connector_requests_total`, `penpot_http_connector_errors_total`. Definitions live in `app.main/default-metrics`. The xnio busy-thread count is not published: it reads 0 with threads blocked inside a limiter, which is why it was deleted.
|
||||
- The xnio worker MXBean can return transient `-1` (e.g. on the queue or the pool size); negative samples are discarded (gauge keeps its previous value). Undertow exposes absolute request/error totals, so the sampler keeps a watermark atom and publishes deltas; a counter reset (decreasing totals) skips the negative delta and moves the watermark forward.
|
||||
- The `process_*` families (`process_open_fds`, `process_max_fds`, `process_cpu_seconds_total`, …) come from the prometheus client `StandardExports`, registered by `app.metrics/create-registry`. They read the OS MXBean reflectively and need the `jdk.management` module: on a pruned `jlink` JRE the MXBean is `sun.management.BaseOperatingSystemImpl`, the getters throw `NoSuchMethodException` and `StandardExports#collect` swallows it, so those families silently vanish from `/metrics`. `docker/images/Dockerfile.backend` keeps `jdk.management` in the `--add-modules` list, and `backend-tests.metrics-test` pins the contract.
|
||||
|
||||
## Storage and media
|
||||
|
||||
@ -9,4 +9,4 @@ JVM `clojure.test` (kaocha runner) under `backend/test/backend_tests/`.
|
||||
- Isolated run: `clojure -M:dev:test --focus backend-tests.my-ns-test` for a specific test namespace, or `clojure -M:dev:test --focus backend-tests.my-ns-test/my-test-var` for a specific test var.
|
||||
- Regression run: `clojure -M:dev:test` to ensure no regressions in related functional areas.
|
||||
- If you need to filter output, tee to a temp file first: `clojure -M:dev:test 2>&1 | tee /tmp/penpot-test-output.txt`.
|
||||
- `backend-tests.helpers` has `metrics-stub` (a metrics instance with no collectors, where every `run!` is a no-op) and `init-climit` (a concurrency limiter built from an inline config map), for tests that need a limiter without starting the system. Log lines are capturable with `add-watch` on `app.common.logging/log-record` (it holds the last record; the watcher sees every one).
|
||||
- `backend-tests.helpers` has `metrics-stub` (a metrics instance with no collectors, where every `run!` is a no-op), `init-climit` (a concurrency limiter built from an inline config map, with an optional second argument for the metrics instance to publish to) and `with-climit`, the macro that builds one, runs the body and halts it. `init-climit` starts the republisher daemon thread, so a test that builds a limiter without halting it leaks that thread for the rest of the JVM run: use `with-climit`. Log lines are capturable with `add-watch` on `app.common.logging/log-record` (it holds the last record; the watcher sees every one, but only of the levels the logger has enabled: a line the app writes at TRACE never gets there).
|
||||
@ -7,7 +7,7 @@ Keep this memory free of internal names: no host names, no deployed versions, no
|
||||
## Datasources
|
||||
|
||||
- Metrics — **Prometheus**, uid `xkRmyFBGk`. PromQL. Metric families: `penpot_*` (RPC, task timings, CLIMIT, HTTP, websocket, storage), `hikaricp_*`, `jvm_*` and `process_*` on the backend (`job="app"`); `pg_*` from the Postgres exporter (`job="postgres"`); `redis_*` from the Redis exporter (`job="redis"`); `node_*` and `container_*` from node-exporter and cAdvisor. Backend histograms are in **milliseconds**.
|
||||
- Logs — **Loki**, uid `h3CGKhEMk`. LogQL, and filter by the structured metadata `logger`/`level` (`{job="penpot"} | logger="..." | level="ERROR"`). Some signals exist only as logs: CLIMIT rejections have no metric (`logger="app.rpc.climit"`, `hint="rejected"`).
|
||||
- Logs — **Loki**, uid `h3CGKhEMk`. LogQL, and filter by the structured metadata `logger`/`level` (`{job="penpot"} | logger="..." | level="ERROR"`). Signals that only exist as logs: the profile behind a CLIMIT queue (`logger="app.rpc.climit"`, whose `id` carries the key). CLIMIT rejections have their own counters (`penpot_rpc_climit_rejected_total`, by `name` and `reason`), and their `hint="rejected"` line goes out at WARN only when something stays queued, so counting the lines is not counting the rejections.
|
||||
- SQL — **Postgres**, uid `y8ea9tfMk`. Panels need `rawQuery: true`, `editorMode: "code"` and `format: "table"`. The `task` table is the only truth about the queue depth.
|
||||
- **Postgres (telemetry)**, uid `9YAUuM_Gz`.
|
||||
- `$__rate_interval` and `$__interval` are substituted by Grafana; use them instead of fixed windows in graphs.
|
||||
@ -37,6 +37,6 @@ Keep this memory free of internal names: no host names, no deployed versions, no
|
||||
- Saving through the API normalizes the dashboard (rows may be flattened, `y` recomputed): submit the native form (rows with `panels: []`, panels top-level, contiguous `y`).
|
||||
- Panel images cannot be rendered (no image renderer installed) and there is no browser here: verify a dashboard by running its queries with the variables substituted.
|
||||
- Signals that exist only as logs are counted with Loki, and the count is then cross-checked against the nginx access log for the same window and the same status code (`job="frontend"`, e.g. a `429`): two independent sources must agree before a number is trusted.
|
||||
- The app writes its own logs at WARN only when the CLIMIT queue is not empty and at TRACE otherwise, so production Loki holds just the under-pressure lines: a count of `hint="enqueued"` or `hint="acquired"` is not the number of calls, while rejections always land at WARN and their count is the real one.
|
||||
- The app writes its own logs at WARN only when something is left queued behind the call and at TRACE otherwise, so production Loki holds just the under-pressure lines: a count of `hint="enqueued"` or `hint="acquired"` is not the number of calls, and neither is a count of `hint="rejected"` lines, because a rejection that leaves the queue empty (a lone caller that times out) is written at TRACE. The rejection counters are the real number.
|
||||
- The MCP Prometheus tool can answer empty for every query, `up` included, while the datasource itself is fine: an empty result from it is not proof that a panel has no data, and blaming the datasource for it is wrong. Run PromQL by hand with `grafana_api_request` against `/api/datasources/uid/<uid>/resources/api/v1/query?query=...` (or `POST /api/ds/query`, the path dashboards use), and check when a series is expected but the tool says nothing.
|
||||
- **Check `list_dashboard_versions` before writing.** The user edits dashboards by hand and a full-JSON overwrite silently reverts that work; prefer `operations` patches per panel.
|
||||
|
||||
@ -176,7 +176,6 @@
|
||||
[:audit-log-http-handler-concurrency {:optional true} ::sm/int]
|
||||
|
||||
[:default-executor-parallelism {:optional true} ::sm/int] ;; REVIEW
|
||||
[:scheduled-executor-parallelism {:optional true} ::sm/int] ;; REVIEW
|
||||
[:worker-default-parallelism {:optional true} ::sm/int]
|
||||
[:worker-webhook-parallelism {:optional true} ::sm/int]
|
||||
|
||||
|
||||
@ -54,21 +54,19 @@
|
||||
|
||||
(defn sample-worker-metrics
|
||||
"Publishes the current state of the xnio worker thread pool (the
|
||||
request dispatch queue and its threads) as gauges."
|
||||
request dispatch queue and its threads) as gauges. The pool size is the
|
||||
signal to watch: the busy thread count of this MXBean stays at zero with
|
||||
threads blocked inside a concurrency limiter, so it is not published."
|
||||
[metrics ^XnioWorkerMXBean mxbean]
|
||||
(let [queue-size (.getWorkerQueueSize mxbean)
|
||||
busy-count (.getBusyWorkerThreadCount mxbean)
|
||||
pool-size (.getWorkerPoolSize mxbean)
|
||||
max-size (.getMaxWorkerPoolSize mxbean)]
|
||||
|
||||
;; negative values are missing measurements, not zeros: the xnio
|
||||
;; MXBean may transiently report -1 on the busy thread count.
|
||||
;; negative values are missing measurements, not zeros: the xnio MXBean
|
||||
;; may transiently report -1.
|
||||
(when (>= queue-size 0)
|
||||
(mtx/run! metrics :id :http-worker-queue-size :val queue-size))
|
||||
|
||||
(when (>= busy-count 0)
|
||||
(mtx/run! metrics :id :http-worker-busy-threads :val busy-count))
|
||||
|
||||
(when (>= pool-size 0)
|
||||
(mtx/run! metrics :id :http-worker-pool-size :val pool-size))
|
||||
|
||||
@ -120,26 +118,14 @@
|
||||
|
||||
(defn create-metrics-sampler
|
||||
"Creates a daemon scheduler that periodically samples the state of
|
||||
the http server and publishes it as metrics. A single thread is used,
|
||||
and an unexpected error on a single sample does NOT cancel the
|
||||
subsequent runs."
|
||||
the http server and publishes it as metrics, in the shape shared by
|
||||
`app.metrics/create-sampler`: a single thread, and an unexpected error
|
||||
on one sample does not cancel the following runs."
|
||||
[^Undertow server metrics]
|
||||
(let [state (atom {:last-requests 0 :last-errors 0})
|
||||
scheduler (px/scheduled-executor
|
||||
:parallelism 1
|
||||
:factory (px/thread-factory :prefix "penpot/http-metrics/"
|
||||
:daemon true))
|
||||
sample (fn sample []
|
||||
(try
|
||||
(sample-http-metrics metrics state server)
|
||||
(finally
|
||||
;; reschedule even if a single sample fails, so
|
||||
;; an unexpected error does not cancel the
|
||||
;; following runs.
|
||||
(px/schedule scheduler metrics-sample-interval-ms sample))))]
|
||||
|
||||
(px/schedule scheduler 0 sample)
|
||||
scheduler))
|
||||
(let [state (atom {:last-requests 0 :last-errors 0})]
|
||||
(mtx/create-sampler metrics-sample-interval-ms
|
||||
(fn [] (sample-http-metrics metrics state server))
|
||||
{:name "http-metrics"})))
|
||||
|
||||
(defmethod ig/expand-key ::server
|
||||
[k v]
|
||||
|
||||
@ -105,17 +105,17 @@
|
||||
::mdef/labels ["name"]
|
||||
::mdef/type :histogram}
|
||||
|
||||
:rpc-climit-queue
|
||||
{::mdef/name "penpot_rpc_climit_queue"
|
||||
::mdef/help "Current number of queued submissions."
|
||||
:rpc-climit-waiting
|
||||
{::mdef/name "penpot_rpc_climit_waiting"
|
||||
::mdef/help "Calls of this limiter name waiting for a permit, on this instance."
|
||||
::mdef/labels ["name"]
|
||||
::mdef/type :gauge}
|
||||
|
||||
:rpc-climit-permits
|
||||
{::mdef/name "penpot_rpc_climit_permits"
|
||||
::mdef/help "Current number of available permits"
|
||||
::mdef/labels ["name"]
|
||||
::mdef/type :gauge}
|
||||
:rpc-climit-rejected
|
||||
{::mdef/name "penpot_rpc_climit_rejected_total"
|
||||
::mdef/help "Calls rejected by this limiter, by name and reason."
|
||||
::mdef/labels ["name" "reason"]
|
||||
::mdef/type :counter}
|
||||
|
||||
:rpc-climit-timing
|
||||
{::mdef/name "penpot_rpc_climit_timing"
|
||||
@ -152,11 +152,6 @@
|
||||
::mdef/help "Current number of queued tasks in the http server xnio worker."
|
||||
::mdef/type :gauge}
|
||||
|
||||
:http-worker-busy-threads
|
||||
{::mdef/name "penpot_http_worker_busy_threads"
|
||||
::mdef/help "Current number of busy threads in the http server xnio worker."
|
||||
::mdef/type :gauge}
|
||||
|
||||
:http-worker-pool-size
|
||||
{::mdef/name "penpot_http_worker_pool_size"
|
||||
::mdef/help "Current number of threads in the http server xnio worker pool."
|
||||
|
||||
@ -10,7 +10,8 @@
|
||||
[app.common.logging :as l]
|
||||
[app.common.schema :as sm]
|
||||
[app.metrics.definition :as-alias mdef]
|
||||
[integrant.core :as ig])
|
||||
[integrant.core :as ig]
|
||||
[promesa.exec :as px])
|
||||
(:import
|
||||
io.prometheus.client.CollectorRegistry
|
||||
io.prometheus.client.Counter
|
||||
@ -141,6 +142,33 @@
|
||||
(run-collector! mobj params)
|
||||
true))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; SAMPLER
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
|
||||
(defn create-sampler
|
||||
"Create a daemon scheduler that runs `sample` every `interval-ms`,
|
||||
starting right away. A single thread is used, named after `name`
|
||||
(default `metrics`), and an unexpected error on a single sample does NOT
|
||||
cancel the following runs: the next run is scheduled in a `finally`.
|
||||
The caller owns the returned scheduler and has to shut it down
|
||||
(`px/shutdown-now`) when its component halts."
|
||||
([interval-ms sample]
|
||||
(create-sampler interval-ms sample {}))
|
||||
([interval-ms sample {:keys [name] :or {name "metrics"}}]
|
||||
(let [prefix (str "penpot/" name "/")
|
||||
scheduler (px/scheduled-executor
|
||||
:parallelism 1
|
||||
:factory (px/thread-factory :prefix prefix :daemon true))
|
||||
task (fn task []
|
||||
(try
|
||||
(sample)
|
||||
(finally
|
||||
(px/schedule scheduler interval-ms task))))]
|
||||
|
||||
(px/schedule scheduler 0 task)
|
||||
scheduler)))
|
||||
|
||||
(defn- create-registry
|
||||
[]
|
||||
(let [registry (CollectorRegistry.)]
|
||||
|
||||
@ -21,10 +21,12 @@
|
||||
[clojure.set :as set]
|
||||
[datoteka.fs :as fs]
|
||||
[integrant.core :as ig]
|
||||
[promesa.exec :as px]
|
||||
[promesa.exec.bulkhead :as pbh])
|
||||
(:import
|
||||
clojure.lang.ExceptionInfo
|
||||
java.util.concurrent.atomic.AtomicLong))
|
||||
java.util.concurrent.atomic.AtomicLong
|
||||
java.util.concurrent.ConcurrentHashMap))
|
||||
|
||||
(set! *warn-on-reflection* true)
|
||||
|
||||
@ -54,6 +56,71 @@
|
||||
(def ^:private check-config
|
||||
(sm/check-fn schema:config))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; WAITING COUNTER
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
|
||||
(def ^:private waiting-sample-interval-ms 15000)
|
||||
|
||||
(defn- create-waiting-counters
|
||||
"Per limiter name, how many calls on this instance are waiting for a
|
||||
permit. Every limiter of a name adds to the same entry, so the value does
|
||||
not depend on which of them ran last the way a sampled gauge does: a
|
||||
request that enters and never leaves stays visible for ever. The map
|
||||
belongs to the component, so its counts die with it instead of surviving
|
||||
into the next system built in the same JVM."
|
||||
[]
|
||||
(ConcurrentHashMap.))
|
||||
|
||||
(defn- waiting-counter
|
||||
^AtomicLong
|
||||
[^ConcurrentHashMap counters name]
|
||||
(.computeIfAbsent counters name
|
||||
(reify java.util.function.Function
|
||||
(apply [_ _] (AtomicLong. 0)))))
|
||||
|
||||
(defn- publish-waiting
|
||||
[metrics name value]
|
||||
(mtx/run! metrics
|
||||
:id :rpc-climit-waiting
|
||||
:val value
|
||||
:labels (into-array String [name])))
|
||||
|
||||
(defn- waiting-inc
|
||||
[metrics counters name]
|
||||
(publish-waiting metrics name (.incrementAndGet (waiting-counter counters name))))
|
||||
|
||||
(defn- waiting-dec
|
||||
[metrics counters name]
|
||||
(publish-waiting metrics name (.decrementAndGet (waiting-counter counters name))))
|
||||
|
||||
(defn- republish-waiting
|
||||
"Publish the waiting count of every name again. A count that stops changing
|
||||
(a limiter whose requests never come back and whose traffic died) has to keep
|
||||
being written, or the series goes stale and the pile-up disappears."
|
||||
[metrics counters]
|
||||
(doseq [[name counter] counters]
|
||||
(publish-waiting metrics name (.get ^AtomicLong counter))))
|
||||
|
||||
(defn- create-waiting-sampler
|
||||
"A daemon sampler that republishes the waiting counts of this component."
|
||||
[metrics counters]
|
||||
(mtx/create-sampler waiting-sample-interval-ms
|
||||
(fn [] (republish-waiting metrics counters))
|
||||
{:name "climit-metrics"}))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; COMPONENT
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
|
||||
(defrecord ConcurrencyLimiter [params cache metrics counters sampler]
|
||||
IConcurrencyLimiter
|
||||
(get-config [_ id]
|
||||
(get params id))
|
||||
|
||||
(invoke [_ config handler]
|
||||
(impl-invoke metrics counters cache config handler)))
|
||||
|
||||
(def ^:private schema:climit-params
|
||||
[:map
|
||||
::mtx/metrics
|
||||
@ -70,22 +137,24 @@
|
||||
(when enabled
|
||||
(when-let [params (some->> config slurp edn/read-string check-config)]
|
||||
(l/inf :hint "initializing concurrency limit" :config (str config))
|
||||
(let [params (reduce-kv (fn [result k v]
|
||||
(when-not (:timeout v)
|
||||
(l/wrn :hint "limiter without timeout, a saturated one parks its threads forever"
|
||||
:id k))
|
||||
(assoc result k (assoc v ::id k)))
|
||||
params
|
||||
params)
|
||||
cache (create-cache cfg)]
|
||||
(let [params (reduce-kv (fn [result k v]
|
||||
(when-not (:timeout v)
|
||||
(l/wrn :hint "limiter without timeout, a saturated one parks its threads forever"
|
||||
:id k))
|
||||
(assoc result k (assoc v ::id k)))
|
||||
params
|
||||
params)
|
||||
cache (create-cache cfg)
|
||||
counters (create-waiting-counters)]
|
||||
|
||||
(reify
|
||||
IConcurrencyLimiter
|
||||
(get-config [_ id]
|
||||
(get params id))
|
||||
(->ConcurrencyLimiter params cache metrics counters
|
||||
(create-waiting-sampler metrics counters))))))
|
||||
|
||||
(invoke [_ config handler]
|
||||
(impl-invoke metrics cache config handler)))))))
|
||||
(defmethod ig/halt-key! ::rpc/climit
|
||||
[_ instance]
|
||||
(when-let [sampler (:sampler instance)]
|
||||
(l/inf :hint "stopping concurrency limit metrics")
|
||||
(px/shutdown-now sampler)))
|
||||
|
||||
(defn- id->str
|
||||
([id]
|
||||
@ -113,28 +182,15 @@
|
||||
:keepalive "5m")))
|
||||
|
||||
(defn- measure
|
||||
[metrics mlabels stats elapsed]
|
||||
(let [mpermits (:max-permits stats)
|
||||
permits (:permits stats)
|
||||
queue (:queue stats)
|
||||
queue (- queue mpermits)
|
||||
queue (if (neg? queue) 0 queue)]
|
||||
|
||||
(mtx/run! metrics
|
||||
:id :rpc-climit-queue
|
||||
:val queue
|
||||
:labels mlabels)
|
||||
|
||||
(mtx/run! metrics
|
||||
:id :rpc-climit-permits
|
||||
:val permits
|
||||
:labels mlabels)
|
||||
|
||||
(when elapsed
|
||||
(mtx/run! metrics
|
||||
:id :rpc-climit-timing
|
||||
:val (inst-ms elapsed)
|
||||
:labels mlabels))))
|
||||
"Publish the wait time of a call that got its permit, in milliseconds. The
|
||||
queue is not published here: a gauge written by whichever limiter of the name
|
||||
ran last cannot show it, and the `:rpc-climit-waiting` gauge accumulates the
|
||||
entries and the exits instead."
|
||||
[metrics mlabels elapsed]
|
||||
(mtx/run! metrics
|
||||
:id :rpc-climit-timing
|
||||
:val (inst-ms elapsed)
|
||||
:labels mlabels))
|
||||
|
||||
(defn- prepare-params-for-debug
|
||||
[params]
|
||||
@ -143,7 +199,7 @@
|
||||
(update-vals str)))
|
||||
|
||||
(defn- log
|
||||
[action req-id stats limit-id limit-label limit-params elapsed]
|
||||
[{:keys [id label params permits queue-max]} action req-id stats elapsed code reason]
|
||||
(let [mpermits (:max-permits stats)
|
||||
queue (:queue stats)
|
||||
queue (- queue mpermits)
|
||||
@ -153,11 +209,15 @@
|
||||
(l/log level
|
||||
:hint action
|
||||
:req req-id
|
||||
:id limit-id
|
||||
:label limit-label
|
||||
:id id
|
||||
:label label
|
||||
:queue queue
|
||||
:queue-max queue-max
|
||||
:permits permits
|
||||
:elapsed (some-> elapsed ct/format-duration)
|
||||
:params @limit-params)))
|
||||
:code code
|
||||
:reason reason
|
||||
:params @params)))
|
||||
|
||||
(def ^:private idseq (AtomicLong. 0))
|
||||
|
||||
@ -172,18 +232,24 @@
|
||||
(get [_] (java.util.HashSet.)))))
|
||||
|
||||
(defn- impl-invoke
|
||||
[metrics cache config handler]
|
||||
[metrics counters cache config handler]
|
||||
(let [limit-id (::id config)
|
||||
limit-key (::key config)
|
||||
limit-label (::label config)
|
||||
limit-name (id->str limit-id)
|
||||
limit-params (delay
|
||||
(prepare-params-for-debug
|
||||
(::params config)))
|
||||
|
||||
mlabels (into-array String [(id->str limit-id)])
|
||||
limit-id (id->str limit-id limit-key)
|
||||
limiter (cache/get cache limit-id (partial create-limiter config))
|
||||
held ^java.util.Set (.get ^ThreadLocal held-limiters)]
|
||||
mlabels (into-array String [limit-name])
|
||||
limit-id (id->str limit-id limit-key)
|
||||
limit {:id limit-id
|
||||
:label limit-label
|
||||
:params limit-params
|
||||
:permits (or (:permits config) (:concurrency config) 1)
|
||||
:queue-max (or (:queue config) (:queue-size config))}
|
||||
limiter (cache/get cache limit-id (partial create-limiter config))
|
||||
held ^java.util.Set (.get ^ThreadLocal held-limiters)]
|
||||
|
||||
(if (.contains held limiter)
|
||||
;; The calling thread already holds this limiter for this key: run the
|
||||
@ -192,28 +258,47 @@
|
||||
(l/trc :hint "reentrant invocation" :id limit-id :label limit-label)
|
||||
(handler))
|
||||
|
||||
(let [tpoint (ct/tpoint)
|
||||
req-id (.incrementAndGet ^AtomicLong idseq)]
|
||||
(let [tpoint (ct/tpoint)
|
||||
req-id (.incrementAndGet ^AtomicLong idseq)
|
||||
waiting (volatile! true)
|
||||
stop-waiting (fn []
|
||||
;; Exactly one decrement per increment: the flag
|
||||
;; makes a second call a no-op and the `finally`
|
||||
;; guarantees that there is one.
|
||||
(when @waiting
|
||||
(vreset! waiting false)
|
||||
(waiting-dec metrics counters limit-name)))]
|
||||
|
||||
(.add held limiter)
|
||||
|
||||
(try
|
||||
;; The increment lives inside the `try`, so a throwable from the
|
||||
;; publishing path unwinds through the `finally`: the thread drops the
|
||||
;; mark and the count comes back down instead of staying up for good.
|
||||
(waiting-inc metrics counters limit-name)
|
||||
|
||||
(let [stats (pbh/get-stats limiter)]
|
||||
(measure metrics mlabels stats nil)
|
||||
(log "enqueued" req-id stats limit-id limit-label limit-params nil))
|
||||
(log limit "enqueued" req-id stats nil nil nil))
|
||||
|
||||
(pbh/invoke! limiter (fn []
|
||||
(stop-waiting)
|
||||
(let [elapsed (tpoint)
|
||||
stats (pbh/get-stats limiter)]
|
||||
(measure metrics mlabels stats elapsed)
|
||||
(log "acquired" req-id stats limit-id limit-label limit-params elapsed)
|
||||
(measure metrics mlabels elapsed)
|
||||
(log limit "acquired" req-id stats elapsed nil nil)
|
||||
(handler))))
|
||||
|
||||
(catch ExceptionInfo cause
|
||||
(let [{:keys [type code]} (ex-data cause)]
|
||||
(let [{:keys [type code] :as data} (ex-data cause)]
|
||||
(if (= :bulkhead-error type)
|
||||
(let [elapsed (tpoint)
|
||||
stats (pbh/get-stats limiter)]
|
||||
(log "rejected" req-id stats limit-id limit-label limit-params elapsed)
|
||||
stats (pbh/get-stats limiter)
|
||||
reason (if (some? (:timeout data)) "timeout" "queue-full")]
|
||||
(mtx/run! metrics
|
||||
:id :rpc-climit-rejected
|
||||
:inc 1
|
||||
:labels (into-array String [limit-name reason]))
|
||||
(log limit "rejected" req-id stats elapsed code reason)
|
||||
(ex/raise :type :concurrency-limit
|
||||
:code code
|
||||
:hint "concurrency limit reached"
|
||||
@ -222,11 +307,11 @@
|
||||
|
||||
(finally
|
||||
(.remove held limiter)
|
||||
(stop-waiting)
|
||||
(let [elapsed (tpoint)
|
||||
stats (pbh/get-stats limiter)]
|
||||
stats (pbh/get-stats limiter)]
|
||||
|
||||
(measure metrics mlabels stats nil)
|
||||
(log "finished" req-id stats limit-id limit-label limit-params elapsed))))))))
|
||||
(log limit "finished" req-id stats elapsed nil nil))))))))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; MIDDLEWARE
|
||||
|
||||
@ -8,6 +8,7 @@
|
||||
(:require
|
||||
[app.common.logging :as l]
|
||||
[app.common.uuid :as uuid]
|
||||
[app.main :as main]
|
||||
[app.metrics :as mtx]
|
||||
[app.rpc :as-alias rpc]
|
||||
[app.rpc.climit :as climit]
|
||||
@ -15,23 +16,24 @@
|
||||
[backend-tests.helpers :as th]
|
||||
[clojure.edn :as edn]
|
||||
[clojure.java.io :as io]
|
||||
[clojure.test :as t]))
|
||||
[clojure.test :as t]
|
||||
[integrant.core :as ig]))
|
||||
|
||||
(def ^:private !calls (atom []))
|
||||
(def ^:private calls (atom []))
|
||||
|
||||
(defn- record-metrics!
|
||||
(defn- record-metrics
|
||||
"Run `f` recording every metrics call it makes."
|
||||
[f]
|
||||
(reset! !calls [])
|
||||
(reset! calls [])
|
||||
(with-redefs [mtx/run!
|
||||
(fn [_ & {:keys [id labels]}]
|
||||
(swap! !calls conj {:id id :labels (vec labels)}))]
|
||||
(fn [_ & {:keys [id labels val]}]
|
||||
(swap! calls conj {:id id :labels (vec labels) :val val}))]
|
||||
(f)))
|
||||
|
||||
(defn- acquired-in-order
|
||||
"Limiter names acquired so far, in acquisition order."
|
||||
[]
|
||||
(->> @!calls
|
||||
(->> @calls
|
||||
(filter #(= :rpc-climit-timing (:id %)))
|
||||
(mapcat :labels)
|
||||
(vec)))
|
||||
@ -42,7 +44,7 @@
|
||||
(let [deadline (+ (System/currentTimeMillis) timeout)]
|
||||
(loop []
|
||||
(cond
|
||||
(some (fn [call] (some #{label} (:labels call))) @!calls)
|
||||
(some (fn [call] (some #{label} (:labels call))) @calls)
|
||||
true
|
||||
|
||||
(> (System/currentTimeMillis) deadline)
|
||||
@ -104,18 +106,18 @@
|
||||
;; The nested `climit/invoke!` calls take only the command's own limits and
|
||||
;; run inside the root ones, so any other order lets two calls of the same
|
||||
;; profile hold the same limiters in opposite order and park for good.
|
||||
(record-metrics!
|
||||
(record-metrics
|
||||
(fn []
|
||||
(let [climit (th/init-climit test-config)
|
||||
mdata {::sv/name "test-command" ::climit/id image-limits}
|
||||
wrapped (climit/wrap {::rpc/climit climit} (fn [_ _] :ok) mdata)
|
||||
result (wrapped {::rpc/climit climit} {::rpc/profile-id (uuid/next)})]
|
||||
(t/is (= :ok result))
|
||||
(t/is (= ["root/by-profile"
|
||||
"root/global"
|
||||
"process-image/by-profile"
|
||||
"process-image/global"]
|
||||
(acquired-in-order)))))))
|
||||
(th/with-climit [climit test-config]
|
||||
(let [mdata {::sv/name "test-command" ::climit/id image-limits}
|
||||
wrapped (climit/wrap {::rpc/climit climit} (fn [_ _] :ok) mdata)
|
||||
result (wrapped {::rpc/climit climit} {::rpc/profile-id (uuid/next)})]
|
||||
(t/is (= :ok result))
|
||||
(t/is (= ["root/by-profile"
|
||||
"root/global"
|
||||
"process-image/by-profile"
|
||||
"process-image/global"]
|
||||
(acquired-in-order))))))))
|
||||
|
||||
(t/deftest nested-limits-do-not-deadlock-with-the-command-own-limits
|
||||
;; Two paths over one profile: a command declaring its own limits (like
|
||||
@ -123,49 +125,49 @@
|
||||
;; then asks for the same limits from inside its handler (like
|
||||
;; `create-file-media-object-from-url`). Each path holds what the other one
|
||||
;; waits for the moment the order differs, and nothing releases.
|
||||
(record-metrics!
|
||||
(record-metrics
|
||||
(fn []
|
||||
(let [climit (th/init-climit test-config)
|
||||
pid (uuid/next)
|
||||
gate (promise)
|
||||
entered (promise)
|
||||
(th/with-climit [climit test-config]
|
||||
(let [pid (uuid/next)
|
||||
gate (promise)
|
||||
entered (promise)
|
||||
|
||||
handler-b (fn [cfg params]
|
||||
(deliver entered true)
|
||||
@gate
|
||||
(-> cfg
|
||||
(assoc ::climit/id [[:process-image/by-profile pid]
|
||||
[:process-image/global]])
|
||||
(assoc ::climit/label "path-b")
|
||||
(climit/invoke! (fn [_ _] :nested-ok) params)))
|
||||
handler-b (fn [cfg params]
|
||||
(deliver entered true)
|
||||
@gate
|
||||
(-> cfg
|
||||
(assoc ::climit/id [[:process-image/by-profile pid]
|
||||
[:process-image/global]])
|
||||
(assoc ::climit/label "path-b")
|
||||
(climit/invoke! (fn [_ _] :nested-ok) params)))
|
||||
|
||||
wrapped-a (climit/wrap {::rpc/climit climit}
|
||||
(fn [_ _] :ok)
|
||||
{::sv/name "path-a" ::climit/id image-limits})
|
||||
wrapped-a (climit/wrap {::rpc/climit climit}
|
||||
(fn [_ _] :ok)
|
||||
{::sv/name "path-a" ::climit/id image-limits})
|
||||
|
||||
wrapped-b (climit/wrap {::rpc/climit climit}
|
||||
handler-b
|
||||
{::sv/name "path-b"})]
|
||||
wrapped-b (climit/wrap {::rpc/climit climit}
|
||||
handler-b
|
||||
{::sv/name "path-b"})]
|
||||
|
||||
;; Path B takes the root limiters and waits inside its handler.
|
||||
(let [fb (future (failed-type #(wrapped-b {::rpc/climit climit}
|
||||
{::rpc/profile-id pid})))]
|
||||
(t/is (true? (deref entered 5000 false)))
|
||||
|
||||
;; Path A, started now, takes its own limiters and then waits for the
|
||||
;; root ones, which B holds. The recorded metrics are how the test
|
||||
;; knows A got that far instead of guessing with a sleep.
|
||||
(reset! !calls [])
|
||||
(let [fa (future (failed-type #(wrapped-a {::rpc/climit climit}
|
||||
;; Path B takes the root limiters and waits inside its handler.
|
||||
(let [fb (future (failed-type #(wrapped-b {::rpc/climit climit}
|
||||
{::rpc/profile-id pid})))]
|
||||
(t/is (true? (saw-call? "root/by-profile" 5000)))
|
||||
(t/is (true? (deref entered 5000 false)))
|
||||
|
||||
;; Let B ask for the limits A holds: if the order differs, each path
|
||||
;; now waits for a permit the other one holds.
|
||||
(deliver gate true)
|
||||
;; Path A, started now, takes its own limiters and then waits for the
|
||||
;; root ones, which B holds. The recorded metrics are how the test
|
||||
;; knows A got that far instead of guessing with a sleep.
|
||||
(reset! calls [])
|
||||
(let [fa (future (failed-type #(wrapped-a {::rpc/climit climit}
|
||||
{::rpc/profile-id pid})))]
|
||||
(t/is (true? (saw-call? "root/by-profile" 5000)))
|
||||
|
||||
(t/is (= :ok (deref fa 15000 :timeout)))
|
||||
(t/is (= :nested-ok (deref fb 15000 :timeout)))))))))
|
||||
;; Let B ask for the limits A holds: if the order differs, each path
|
||||
;; now waits for a permit the other one holds.
|
||||
(deliver gate true)
|
||||
|
||||
(t/is (= :ok (deref fa 15000 :timeout)))
|
||||
(t/is (= :nested-ok (deref fb 15000 :timeout))))))))))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; REENTRANCY
|
||||
@ -175,60 +177,60 @@
|
||||
;; A handler entering climit for a limit it already holds must not queue
|
||||
;; behind itself: it runs inline and takes no second permit. Another thread
|
||||
;; asking for the same limiter still waits, so the limit stays in place.
|
||||
(let [climit (th/init-climit {:process-image/by-profile
|
||||
{:permits 1 :timeout 2000}})
|
||||
pid (uuid/next)
|
||||
cfg {::rpc/climit climit
|
||||
::climit/id [[:process-image/by-profile pid]]}
|
||||
inner (promise)
|
||||
other (promise)
|
||||
handler (fn [cfg params]
|
||||
(deliver inner
|
||||
(failed-type #(climit/invoke! cfg (fn [_ _] :inner-ok) params)))
|
||||
(deliver other (future (climit/invoke! cfg (fn [_ _] :other-ok) params)))
|
||||
(t/is (= :waiting (deref @other 300 :waiting)))
|
||||
:outer-ok)
|
||||
(th/with-climit [climit {:process-image/by-profile
|
||||
{:permits 1 :timeout 2000}}]
|
||||
(let [pid (uuid/next)
|
||||
cfg {::rpc/climit climit
|
||||
::climit/id [[:process-image/by-profile pid]]}
|
||||
inner (promise)
|
||||
other (promise)
|
||||
handler (fn [cfg params]
|
||||
(deliver inner
|
||||
(failed-type #(climit/invoke! cfg (fn [_ _] :inner-ok) params)))
|
||||
(deliver other (future (climit/invoke! cfg (fn [_ _] :other-ok) params)))
|
||||
(t/is (= :waiting (deref @other 300 :waiting)))
|
||||
:outer-ok)
|
||||
|
||||
outcome (failed-type #(climit/invoke! cfg handler {}))]
|
||||
outcome (failed-type #(climit/invoke! cfg handler {}))]
|
||||
|
||||
;; With no reentrancy handling the inner call waits for the permit this very
|
||||
;; thread holds and ends failing by timeout.
|
||||
(t/is (= :inner-ok (deref inner 5000 :timeout)))
|
||||
(t/is (= :outer-ok outcome))
|
||||
;; With no reentrancy handling the inner call waits for the permit this very
|
||||
;; thread holds and ends failing by timeout.
|
||||
(t/is (= :inner-ok (deref inner 5000 :timeout)))
|
||||
(t/is (= :outer-ok outcome))
|
||||
|
||||
;; once the outer call releases, the other thread gets its permit
|
||||
(t/is (= :other-ok (deref @other 5000 :timeout)))))
|
||||
;; once the outer call releases, the other thread gets its permit
|
||||
(t/is (= :other-ok (deref @other 5000 :timeout))))))
|
||||
|
||||
(t/deftest reentrant-tracking-is-cleaned-up-when-the-handler-fails
|
||||
;; If the mark is not removed, the thread keeps believing it holds the
|
||||
;; limiter and every later call of that key would skip the limit.
|
||||
(let [climit (th/init-climit {:process-image/by-profile
|
||||
{:permits 1 :timeout 1000}})
|
||||
pid (uuid/next)
|
||||
cfg {::rpc/climit climit
|
||||
::climit/id [[:process-image/by-profile pid]]}
|
||||
holding (promise)
|
||||
gate (promise)]
|
||||
(th/with-climit [climit {:process-image/by-profile
|
||||
{:permits 1 :timeout 1000}}]
|
||||
(let [pid (uuid/next)
|
||||
cfg {::rpc/climit climit
|
||||
::climit/id [[:process-image/by-profile pid]]}
|
||||
holding (promise)
|
||||
gate (promise)]
|
||||
|
||||
;; The failing call runs on this thread, so this thread is the one that
|
||||
;; must forget it held the limiter.
|
||||
(t/is (thrown? clojure.lang.ExceptionInfo
|
||||
(climit/invoke! cfg (fn [_ _] (throw (ex-info "boom" {}))) {})))
|
||||
;; The failing call runs on this thread, so this thread is the one that
|
||||
;; must forget it held the limiter.
|
||||
(t/is (thrown? clojure.lang.ExceptionInfo
|
||||
(climit/invoke! cfg (fn [_ _] (throw (ex-info "boom" {}))) {})))
|
||||
|
||||
;; another thread takes the only permit
|
||||
(let [fb (future (climit/invoke! cfg (fn [_ _]
|
||||
(deliver holding true)
|
||||
@gate
|
||||
:held)
|
||||
{}))]
|
||||
(t/is (true? (deref holding 5000 false)))
|
||||
;; another thread takes the only permit
|
||||
(let [fb (future (climit/invoke! cfg (fn [_ _]
|
||||
(deliver holding true)
|
||||
@gate
|
||||
:held)
|
||||
{}))]
|
||||
(t/is (true? (deref holding 5000 false)))
|
||||
|
||||
;; this thread must wait for it, not skip the limiter
|
||||
(t/is (= :concurrency-limit
|
||||
(failed-type #(climit/invoke! cfg (fn [_ _] :inner-ok) {}))))
|
||||
;; this thread must wait for it, not skip the limiter
|
||||
(t/is (= :concurrency-limit
|
||||
(failed-type #(climit/invoke! cfg (fn [_ _] :inner-ok) {}))))
|
||||
|
||||
(deliver gate true)
|
||||
(t/is (= :held (deref fb 5000 :timeout))))))
|
||||
(deliver gate true)
|
||||
(t/is (= :held (deref fb 5000 :timeout)))))))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; SHIPPED CONFIG
|
||||
@ -250,15 +252,441 @@
|
||||
;; The config that production loads lives outside the repository and is
|
||||
;; maintained by hand, so a limiter without a timeout has to say so at
|
||||
;; startup instead of parking threads in silence months later.
|
||||
(let [!records (atom [])]
|
||||
(let [records (atom [])]
|
||||
(add-watch l/log-record ::capture
|
||||
(fn [_ _ _ record] (swap! !records conj record)))
|
||||
(fn [_ _ _ record] (swap! records conj record)))
|
||||
(try
|
||||
(th/init-climit {:process-image/by-profile {:permits 1 :timeout 1000}
|
||||
:process-image/global {:permits 8}})
|
||||
|
||||
(t/is (true? (warned? !records 2000)))
|
||||
(t/is (= #{:process-image/global} (warned-ids @!records)))
|
||||
(th/with-climit [_climit {:process-image/by-profile {:permits 1 :timeout 1000}
|
||||
:process-image/global {:permits 8}}]
|
||||
(t/is (true? (warned? records 2000)))
|
||||
(t/is (= #{:process-image/global} (warned-ids @records))))
|
||||
|
||||
(finally
|
||||
(remove-watch l/log-record ::capture)))))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; WAITING COUNTER
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
|
||||
(defn- waiting-value-of
|
||||
"Values published to the waiting gauge for one limiter name, in order."
|
||||
[name]
|
||||
(->> @calls
|
||||
(filter #(= :rpc-climit-waiting (:id %)))
|
||||
(filter #(= name (first (:labels %))))
|
||||
(map :val)
|
||||
(vec)))
|
||||
|
||||
(defn- waiting-last
|
||||
"Last value published to the waiting gauge for one limiter name. How many
|
||||
publications land inside a window is not part of the contract, the value is,
|
||||
so this is what the assertions read."
|
||||
[name]
|
||||
(last (waiting-value-of name)))
|
||||
|
||||
(defn- waiting-count
|
||||
"Current waiting count of a limiter name, read from its component."
|
||||
[climit name]
|
||||
(.get ^java.util.concurrent.atomic.AtomicLong
|
||||
(#'climit/waiting-counter (:counters climit) name)))
|
||||
|
||||
(defn- wait-until
|
||||
"Wait until `pred` holds, or give up."
|
||||
[pred timeout]
|
||||
(let [deadline (+ (System/currentTimeMillis) timeout)]
|
||||
(loop []
|
||||
(cond
|
||||
(pred) true
|
||||
(> (System/currentTimeMillis) deadline) false
|
||||
:else (do (Thread/sleep 10) (recur))))))
|
||||
|
||||
(defn- limiter-name
|
||||
"The name climit publishes for a limiter id: the keyword without the colon."
|
||||
[id]
|
||||
(subs (str id) 1))
|
||||
|
||||
(defn- limiter-cfg
|
||||
[climit name key]
|
||||
{::rpc/climit climit
|
||||
::climit/id [[name key]]})
|
||||
|
||||
(t/deftest waiting-gauge-follows-the-queue
|
||||
;; The value is the number of calls waiting for a permit: up when a call
|
||||
;; starts waiting, down when the wait ends. The gauges that sampled the
|
||||
;; limiter stats could not show this, because the value of whichever limiter
|
||||
;; of the name ran last was the one that survived.
|
||||
(record-metrics
|
||||
(fn []
|
||||
(th/with-climit [climit {:test-waiting/by-profile {:permits 1 :timeout 3000}}]
|
||||
(let [name (limiter-name :test-waiting/by-profile)
|
||||
entered (promise)
|
||||
gate (promise)
|
||||
holder (future (climit/invoke! (limiter-cfg climit :test-waiting/by-profile "a")
|
||||
(fn [_ _] (deliver entered true) @gate :held)
|
||||
{}))]
|
||||
|
||||
(t/is (true? (deref entered 5000 false)))
|
||||
|
||||
;; the holder is running, not waiting
|
||||
(reset! calls [])
|
||||
(let [waiter (future (climit/invoke! (limiter-cfg climit :test-waiting/by-profile "a")
|
||||
(fn [_ _] :waited)
|
||||
{}))]
|
||||
(t/is (true? (wait-until #(= 1 (waiting-last name)) 5000)))
|
||||
(t/is (= 1 (waiting-count climit name)))
|
||||
|
||||
(deliver gate true)
|
||||
(t/is (= :waited (deref waiter 5000 :timeout)))
|
||||
(t/is (= :held (deref holder 5000 :timeout)))
|
||||
(t/is (true? (wait-until #(= 0 (waiting-last name)) 5000)))
|
||||
(t/is (= 0 (waiting-count climit name)))))))))
|
||||
|
||||
(t/deftest waiting-gauge-sums-every-key-of-a-name
|
||||
;; Two profiles queueing on the same limiter name add up. That is the shape
|
||||
;; that stayed invisible in production: one series per name, counting the
|
||||
;; entries and the exits, instead of one value per limiter overwritten by the
|
||||
;; next one.
|
||||
(record-metrics
|
||||
(fn []
|
||||
(th/with-climit [climit {:test-waiting-keys/by-profile {:permits 1 :timeout 3000}}]
|
||||
(let [name (limiter-name :test-waiting-keys/by-profile)
|
||||
gates {:a (promise) :b (promise)}
|
||||
holders
|
||||
(into {}
|
||||
(map (fn [k]
|
||||
(let [entered (promise)]
|
||||
[k {:entered entered
|
||||
:future (future (climit/invoke! (limiter-cfg climit :test-waiting-keys/by-profile k)
|
||||
(fn [_ _]
|
||||
(deliver entered true)
|
||||
@(gates k)
|
||||
:held)
|
||||
{}))}])))
|
||||
[:a :b])]
|
||||
|
||||
(doseq [k [:a :b]]
|
||||
(t/is (true? (deref (:entered (holders k)) 5000 false))))
|
||||
|
||||
;; one call per key, waiting behind its own holder
|
||||
(let [waiters (into {}
|
||||
(map (fn [k]
|
||||
[k (future (climit/invoke! (limiter-cfg climit :test-waiting-keys/by-profile k)
|
||||
(fn [_ _] :waited)
|
||||
{}))]))
|
||||
[:a :b])]
|
||||
|
||||
(t/is (true? (wait-until #(= 2 (waiting-count climit name)) 5000)))
|
||||
;; the increment that reaches 2 always publishes 2, whichever waiter
|
||||
;; does it last: the order of two concurrent publications is not the
|
||||
;; contract, the value is.
|
||||
(t/is (some #(= 2 %) (waiting-value-of name)))
|
||||
|
||||
(doseq [k [:a :b]] (deliver (gates k) true))
|
||||
|
||||
(doseq [k [:a :b]]
|
||||
(t/is (= :waited (deref (waiters k) 5000 :timeout)))
|
||||
(t/is (= :held (deref (:future (holders k)) 5000 :timeout))))
|
||||
|
||||
(t/is (true? (wait-until #(= 0 (waiting-count climit name)) 5000)))
|
||||
(t/is (some #(= 0 %) (waiting-value-of name)))))))))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; REJECTIONS
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
|
||||
(defn- rejections
|
||||
"Rejections published, as [name reason] pairs."
|
||||
[]
|
||||
(->> @calls
|
||||
(filter #(= :rpc-climit-rejected (:id %)))
|
||||
(map (fn [call] [(first (:labels call)) (second (:labels call))]))
|
||||
(vec)))
|
||||
|
||||
(t/deftest rejection-counter-tells-why
|
||||
;; Rejections had no metric: the panel counted them from the logs. Now a call
|
||||
;; rejected because the queue is full and a call rejected because it waited
|
||||
;; too long are counted apart.
|
||||
(record-metrics
|
||||
(fn []
|
||||
(th/with-climit [climit {:test-rejected/by-profile {:permits 1 :queue 1 :timeout 60000}}]
|
||||
(let [name (limiter-name :test-rejected/by-profile)
|
||||
cfg (limiter-cfg climit :test-rejected/by-profile "a")
|
||||
entered (promise)
|
||||
gate (promise)
|
||||
holder (future (climit/invoke! cfg
|
||||
(fn [_ _] (deliver entered true) @gate :held)
|
||||
{}))]
|
||||
|
||||
(t/is (true? (deref entered 5000 false)))
|
||||
(reset! calls [])
|
||||
|
||||
;; the queue holds one waiter, so the next call is rejected right away
|
||||
(let [queued (future (failed-type #(climit/invoke! cfg (fn [_ _] :queued) {})))]
|
||||
(t/is (true? (wait-until #(= 1 (waiting-last name)) 5000)))
|
||||
(t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :extra) {}))))
|
||||
(t/is (= [[name "queue-full"]] (rejections)))
|
||||
|
||||
(deliver gate true)
|
||||
(t/is (= :held (deref holder 5000 :timeout)))
|
||||
(t/is (= :queued (deref queued 5000 :timeout))))))
|
||||
|
||||
;; a wait that times out is counted as such, not as a full queue
|
||||
(th/with-climit [climit {:test-timeout/by-profile {:permits 1 :timeout 300}}]
|
||||
(let [name (limiter-name :test-timeout/by-profile)
|
||||
cfg (limiter-cfg climit :test-timeout/by-profile "a")
|
||||
entered (promise)
|
||||
gate (promise)
|
||||
holder (future (climit/invoke! cfg
|
||||
(fn [_ _] (deliver entered true) @gate :held)
|
||||
{}))]
|
||||
|
||||
(t/is (true? (deref entered 5000 false)))
|
||||
(reset! calls [])
|
||||
(t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :late) {}))))
|
||||
(t/is (= [[name "timeout"]] (rejections)))
|
||||
|
||||
(deliver gate true)
|
||||
(t/is (= :held (deref holder 5000 :timeout))))))))
|
||||
|
||||
(t/deftest the-waiting-count-comes-back-after-a-rejection
|
||||
;; A rejected call increments and decrements exactly once: a limiter that
|
||||
;; rejects under pressure must not drift the panel up.
|
||||
(record-metrics
|
||||
(fn []
|
||||
(th/with-climit [climit {:test-rejected-baseline/by-profile {:permits 1 :queue 0 :timeout 3000}}]
|
||||
(let [name (limiter-name :test-rejected-baseline/by-profile)
|
||||
cfg (limiter-cfg climit :test-rejected-baseline/by-profile "a")
|
||||
entered (promise)
|
||||
gate (promise)
|
||||
holder (future (climit/invoke! cfg
|
||||
(fn [_ _] (deliver entered true) @gate :held)
|
||||
{}))]
|
||||
|
||||
(t/is (true? (deref entered 5000 false)))
|
||||
(reset! calls [])
|
||||
|
||||
(t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :extra) {}))))
|
||||
(t/is (= 0 (waiting-count climit name)))
|
||||
(t/is (true? (wait-until #(= 0 (waiting-last name)) 5000)))
|
||||
|
||||
(deliver gate true)
|
||||
(t/is (= :held (deref holder 5000 :timeout))))))))
|
||||
|
||||
(t/deftest the-waiting-count-comes-back-when-the-handler-throws
|
||||
;; A handler that throws still leaves the count where it was.
|
||||
(record-metrics
|
||||
(fn []
|
||||
(th/with-climit [climit {:test-handler-throws/by-profile {:permits 1 :timeout 3000}}]
|
||||
(let [name (limiter-name :test-handler-throws/by-profile)
|
||||
cfg (limiter-cfg climit :test-handler-throws/by-profile "a")]
|
||||
|
||||
(reset! calls [])
|
||||
(t/is (thrown? clojure.lang.ExceptionInfo
|
||||
(climit/invoke! cfg (fn [_ _] (throw (ex-info "boom" {}))) {})))
|
||||
|
||||
(t/is (= 0 (waiting-count climit name)))
|
||||
(t/is (true? (wait-until #(= 0 (waiting-last name)) 5000))))))))
|
||||
|
||||
(defn- failing-first-waiting-publish
|
||||
"A `mtx/run!` replacement that fails the first waiting publication and records
|
||||
the rest, so the decrement of the unwinding path still lands."
|
||||
[calls]
|
||||
(let [pending (atom true)]
|
||||
(fn [_ & {:keys [id labels val]}]
|
||||
(if (and @pending (= :rpc-climit-waiting id))
|
||||
(do (reset! pending false)
|
||||
(throw (ex-info "publishing is down" {})))
|
||||
(swap! calls conj {:id id :labels (vec labels) :val val})))))
|
||||
|
||||
(t/deftest a-throwable-while-waiting-leaves-no-mark-and-no-count
|
||||
;; `waiting-inc` sits inside the `try`: a throwable from the publishing path
|
||||
;; unwinds through the `finally`, so the thread neither keeps the limiter
|
||||
;; marked as held (it would skip that limiter for that key for ever) nor
|
||||
;; leaves the count up.
|
||||
(th/with-climit [climit {:test-throw-waiting/by-profile {:permits 1 :timeout 3000}}]
|
||||
(let [name (limiter-name :test-throw-waiting/by-profile)
|
||||
cfg (limiter-cfg climit :test-throw-waiting/by-profile "a")]
|
||||
|
||||
(reset! calls [])
|
||||
(with-redefs [mtx/run! (failing-first-waiting-publish calls)]
|
||||
(t/is (thrown? clojure.lang.ExceptionInfo
|
||||
(climit/invoke! cfg (fn [_ _] :never) {}))))
|
||||
|
||||
;; the count came back down instead of staying up for good
|
||||
(t/is (= 0 (waiting-count climit name)))
|
||||
(t/is (= [0] (waiting-value-of name)))
|
||||
|
||||
;; and the thread dropped the mark: with it left behind, this call would
|
||||
;; run inline instead of waiting for the permit another thread holds.
|
||||
(let [entered (promise)
|
||||
gate (promise)
|
||||
holder (future (climit/invoke! cfg
|
||||
(fn [_ _] (deliver entered true) @gate :held)
|
||||
{}))]
|
||||
(t/is (true? (deref entered 5000 false)))
|
||||
(t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :inline) {}))))
|
||||
(deliver gate true)
|
||||
(t/is (= :held (deref holder 5000 :timeout)))))))
|
||||
|
||||
(defn- gauge-value
|
||||
"Current value of the waiting gauge for one limiter name, read from a real
|
||||
prometheus collector."
|
||||
[metrics name]
|
||||
(let [collector (:app.metrics.definition/instance
|
||||
(mtx/get-collector metrics :rpc-climit-waiting))]
|
||||
(.get ^io.prometheus.client.Gauge$Child
|
||||
(.labels ^io.prometheus.client.Gauge collector (into-array String [name])))))
|
||||
|
||||
(t/deftest the-waiting-gauge-reaches-a-real-registry
|
||||
;; The recorded calls prove the arguments, not that a collector with that name
|
||||
;; and those labels exists: a typo in the metric id or a wrong label arity
|
||||
;; would only show up in production. Here the value is read back from a real
|
||||
;; prometheus registry.
|
||||
(let [metrics (ig/init-key :app.metrics/metrics {:default main/default-metrics})
|
||||
name (limiter-name :test-registry/by-profile)]
|
||||
|
||||
(th/with-climit [climit {:test-registry/by-profile {:permits 1 :timeout 3000}} metrics]
|
||||
(let [cfg (limiter-cfg climit :test-registry/by-profile "a")
|
||||
entered (promise)
|
||||
gate (promise)
|
||||
holder (future (climit/invoke! cfg
|
||||
(fn [_ _] (deliver entered true) @gate :held)
|
||||
{}))]
|
||||
|
||||
(t/is (true? (deref entered 5000 false)))
|
||||
|
||||
(let [waiter (future (climit/invoke! cfg (fn [_ _] :waited) {}))]
|
||||
(t/is (true? (wait-until #(= 1.0 (gauge-value metrics name)) 5000)))
|
||||
(deliver gate true)
|
||||
(t/is (= :waited (deref waiter 5000 :timeout)))
|
||||
(t/is (= :held (deref holder 5000 :timeout)))
|
||||
(t/is (true? (wait-until #(= 0.0 (gauge-value metrics name)) 5000))))))))
|
||||
|
||||
(defn- rejected-record
|
||||
"The first captured log record that reports a rejection with `reason`."
|
||||
[records reason]
|
||||
(let [deadline (+ (System/currentTimeMillis) 5000)]
|
||||
(loop []
|
||||
(let [record (first (filter (fn [record]
|
||||
(let [props (::l/props record)]
|
||||
(and (= "rejected" (get props :hint))
|
||||
(= reason (get props :reason)))))
|
||||
@records))]
|
||||
(cond
|
||||
record record
|
||||
(> (System/currentTimeMillis) deadline) nil
|
||||
:else (do (Thread/sleep 10) (recur)))))))
|
||||
|
||||
(t/deftest rejection-logs-say-why-and-how-big-the-limits-are
|
||||
;; The line has to say whether the queue was full or the wait timed out, and
|
||||
;; how many permits and how much queue the limiter has, or nobody can tell
|
||||
;; whether the limiter did its job.
|
||||
(let [records (atom [])]
|
||||
(add-watch l/log-record ::capture
|
||||
(fn [_ _ _ record] (swap! records conj record)))
|
||||
(try
|
||||
(record-metrics
|
||||
(fn []
|
||||
(th/with-climit [climit {:test-log/by-profile {:permits 1 :queue 2 :timeout 60000}}]
|
||||
(let [name (limiter-name :test-log/by-profile)
|
||||
cfg (limiter-cfg climit :test-log/by-profile "a")
|
||||
entered (promise)
|
||||
gate (promise)
|
||||
holder (future (climit/invoke! cfg
|
||||
(fn [_ _] (deliver entered true) @gate :held)
|
||||
{}))]
|
||||
|
||||
(t/is (true? (deref entered 5000 false)))
|
||||
(reset! calls [])
|
||||
|
||||
(let [q1 (future (climit/invoke! cfg (fn [_ _] :q1) {}))
|
||||
q2 (future (climit/invoke! cfg (fn [_ _] :q2) {}))]
|
||||
|
||||
;; the two of them fill the queue, so the next one is rejected
|
||||
(t/is (true? (wait-until #(= 2 (waiting-count climit name)) 5000)))
|
||||
(t/is (some #(= 2 %) (waiting-value-of name)))
|
||||
(t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :x) {}))))
|
||||
|
||||
(let [props (::l/props (rejected-record records "queue-full"))]
|
||||
(t/is (= 1 (get props :permits)))
|
||||
(t/is (= 2 (get props :queue-max)))
|
||||
(t/is (= :capacity-limit-reached (get props :code)))
|
||||
(t/is (= "queue-full" (get props :reason))))
|
||||
|
||||
(deliver gate true)
|
||||
(t/is (= :held (deref holder 5000 :timeout)))
|
||||
(deref q1 5000 nil)
|
||||
(deref q2 5000 nil))))
|
||||
|
||||
;; A wait that times out carries its own reason in the line, next to the
|
||||
;; code the limiter raised. The line is emitted at TRACE when nothing
|
||||
;; else is left queued behind it (the caller was the only one waiting),
|
||||
;; so the call to `log` is captured instead of the record, which the
|
||||
;; logger would drop.
|
||||
(th/with-climit [climit {:test-log-timeout/by-profile {:permits 1 :timeout 300}}]
|
||||
(let [cfg (limiter-cfg climit :test-log-timeout/by-profile "a")
|
||||
lines (atom [])
|
||||
entered (promise)
|
||||
gate (promise)
|
||||
capture (fn [_ action _req-id _stats _elapsed code reason]
|
||||
(swap! lines conj [action code reason]))
|
||||
holder (future (climit/invoke! cfg
|
||||
(fn [_ _] (deliver entered true) @gate :held)
|
||||
{}))]
|
||||
|
||||
(t/is (true? (deref entered 5000 false)))
|
||||
|
||||
(with-redefs [climit/log capture]
|
||||
(t/is (= :concurrency-limit (failed-type #(climit/invoke! cfg (fn [_ _] :late) {}))))
|
||||
(t/is (= [["rejected" :capacity-limit-reached "timeout"]]
|
||||
(filterv #(= "rejected" (first %)) @lines))))
|
||||
|
||||
(deliver gate true)
|
||||
(t/is (= :held (deref holder 5000 :timeout)))))))
|
||||
|
||||
(finally
|
||||
(remove-watch l/log-record ::capture)))))
|
||||
|
||||
(t/deftest republish-writes-the-current-waiting-count
|
||||
;; A call that never comes back keeps the count up, but if it is also the
|
||||
;; last thing that happened nothing rewrites that series and it goes stale in
|
||||
;; the scrape. The republish is what keeps it visible.
|
||||
(th/with-climit [climit {:test-republish/by-profile {:permits 1 :timeout 3000}}]
|
||||
(let [name (limiter-name :test-republish/by-profile)]
|
||||
|
||||
(.set ^java.util.concurrent.atomic.AtomicLong
|
||||
(#'climit/waiting-counter (:counters climit) name)
|
||||
3)
|
||||
|
||||
(record-metrics
|
||||
(fn []
|
||||
(#'climit/republish-waiting (th/metrics-stub) (:counters climit))
|
||||
(t/is (= [3] (waiting-value-of name))))))))
|
||||
|
||||
(t/deftest halting-the-component-stops-the-republisher
|
||||
;; The republisher is a daemon thread started with the component: halting it
|
||||
;; has to stop the thread, or every restart leaves another one running.
|
||||
(let [climit (th/init-climit {:test-halt/by-profile {:permits 1 :timeout 3000}})
|
||||
sampler ^java.util.concurrent.ScheduledThreadPoolExecutor (:sampler climit)]
|
||||
|
||||
(t/is (false? (.isShutdown sampler)))
|
||||
|
||||
(ig/halt-key! ::rpc/climit climit)
|
||||
|
||||
(t/is (true? (.isShutdown sampler)))))
|
||||
|
||||
(t/deftest the-climit-metrics-that-remain-are-defined
|
||||
;; A typo in a metric id, or a definition that is not there, makes the
|
||||
;; publishing silent, which is the failure mode this work removes. And the
|
||||
;; gauges that could not tell the truth must be gone.
|
||||
(let [metrics (ig/init-key :app.metrics/metrics {:default main/default-metrics})]
|
||||
|
||||
(t/is (= ["name"]
|
||||
(:app.metrics.definition/labels (mtx/get-collector metrics :rpc-climit-waiting))))
|
||||
(t/is (= ["name" "reason"]
|
||||
(:app.metrics.definition/labels (mtx/get-collector metrics :rpc-climit-rejected))))
|
||||
(t/is (some? (mtx/get-collector metrics :rpc-climit-timing)))
|
||||
|
||||
(t/is (nil? (mtx/get-collector metrics :rpc-climit-queue)))
|
||||
(t/is (nil? (mtx/get-collector metrics :rpc-climit-permits)))
|
||||
(t/is (nil? (mtx/get-collector metrics :http-worker-busy-threads)))))
|
||||
|
||||
@ -436,15 +436,30 @@
|
||||
(get-registry [_] nil)))
|
||||
|
||||
(defn init-climit
|
||||
"Build a concurrency limiter component from an inline config map."
|
||||
[config]
|
||||
(let [file (doto (java.io.File/createTempFile "climit" ".edn")
|
||||
(.deleteOnExit))]
|
||||
(spit file (pr-str config))
|
||||
(ig/init-key ::rpc/climit
|
||||
{::climit/enabled true
|
||||
::climit/config (.getAbsolutePath file)
|
||||
::metrics/metrics (metrics-stub)})))
|
||||
"Build a concurrency limiter component from an inline config map. The
|
||||
optional second argument is the metrics instance to publish to (a
|
||||
`metrics-stub` by default, where every `run!` is a no-op)."
|
||||
([config]
|
||||
(init-climit config (metrics-stub)))
|
||||
([config metrics]
|
||||
(let [file (doto (java.io.File/createTempFile "climit" ".edn")
|
||||
(.deleteOnExit))]
|
||||
(spit file (pr-str config))
|
||||
(ig/init-key ::rpc/climit
|
||||
{::climit/enabled true
|
||||
::climit/config (.getAbsolutePath file)
|
||||
::metrics/metrics metrics}))))
|
||||
|
||||
(defmacro with-climit
|
||||
"Build a concurrency limiter from an inline config map, run the body and
|
||||
halt it, so the republisher thread it starts never outlives the test. An
|
||||
optional third binding element is the metrics instance to publish to."
|
||||
[[sym config & [metrics]] & body]
|
||||
`(let [~sym (init-climit ~config ~@(when metrics [metrics]))]
|
||||
(try
|
||||
~@body
|
||||
(finally
|
||||
(ig/halt-key! ::rpc/climit ~sym)))))
|
||||
|
||||
;; --- UTILS
|
||||
|
||||
|
||||
@ -16,7 +16,6 @@
|
||||
io.prometheus.client.Counter
|
||||
io.prometheus.client.Gauge
|
||||
io.undertow.server.ConnectorStatistics
|
||||
java.util.concurrent.ScheduledThreadPoolExecutor
|
||||
org.xnio.management.XnioWorkerMXBean))
|
||||
|
||||
(t/use-fixtures :once th/state-init)
|
||||
@ -29,9 +28,6 @@
|
||||
{:http-worker-queue-size {:name "penpot_http_worker_queue_size"
|
||||
:help "test"
|
||||
:type :gauge}
|
||||
:http-worker-busy-threads {:name "penpot_http_worker_busy_threads"
|
||||
:help "test"
|
||||
:type :gauge}
|
||||
:http-worker-pool-size {:name "penpot_http_worker_pool_size"
|
||||
:help "test"
|
||||
:type :gauge}
|
||||
@ -132,23 +128,20 @@
|
||||
(http/sample-worker-metrics metrics mxbean)
|
||||
|
||||
(t/is (= 2.0 (gauge-value (collector-instance metrics :http-worker-queue-size))))
|
||||
(t/is (= 10.0 (gauge-value (collector-instance metrics :http-worker-busy-threads))))
|
||||
(t/is (= 40.0 (gauge-value (collector-instance metrics :http-worker-pool-size))))
|
||||
(t/is (= 512.0 (gauge-value (collector-instance metrics :http-worker-max-pool-size))))))
|
||||
|
||||
(t/deftest sample-worker-metrics-skips-negative-samples
|
||||
;; the xnio MXBean occasionally returns -1 on the busy thread count;
|
||||
;; a negative value is a missing measurement, not a zero.
|
||||
;; the xnio MXBean occasionally returns -1 on a count; a negative value is a
|
||||
;; missing measurement, not a zero.
|
||||
(let [metrics (fake-metrics)
|
||||
mxbean (fake-mxbean {:queue 0 :busy -1 :pool 4 :max 512})]
|
||||
mxbean (fake-mxbean {:queue -1 :busy -1 :pool 4 :max 512})]
|
||||
|
||||
(http/sample-worker-metrics metrics mxbean)
|
||||
|
||||
(t/is (= 0.0 (gauge-value (collector-instance metrics :http-worker-queue-size))))
|
||||
(t/is (= 4.0 (gauge-value (collector-instance metrics :http-worker-pool-size))))
|
||||
(t/is (= 512.0 (gauge-value (collector-instance metrics :http-worker-max-pool-size))))
|
||||
(t/is (= 0.0
|
||||
(gauge-value (collector-instance metrics :http-worker-busy-threads))))))
|
||||
(t/is (= 512.0 (gauge-value (collector-instance metrics :http-worker-max-pool-size))))))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; Test: connector metrics
|
||||
@ -230,28 +223,3 @@
|
||||
(finally
|
||||
(px/shutdown-now sampler)
|
||||
(t/is (px/shutdown? sampler))))))
|
||||
|
||||
(t/deftest create-metrics-sampler-reschedules-after-error
|
||||
;; the docstring promise: an unexpected error on a single sample must
|
||||
;; not cancel the following runs. The first sample runs immediately
|
||||
;; and throws; the next one must still be scheduled afterwards.
|
||||
(let [calls (atom 0)]
|
||||
(with-redefs [http/sample-http-metrics (fn [_ _ _]
|
||||
(swap! calls inc)
|
||||
(throw (ex-info "boom" {})))]
|
||||
(let [sampler (http/create-metrics-sampler nil (fake-metrics))
|
||||
queue (.getQueue ^ScheduledThreadPoolExecutor sampler)]
|
||||
(try
|
||||
(t/is (loop [i 0]
|
||||
(cond (pos? @calls) true
|
||||
(> i 200) false
|
||||
:else (do (Thread/sleep 10) (recur (inc i)))))
|
||||
"the first sample must run immediately")
|
||||
|
||||
(t/is (loop [i 0]
|
||||
(cond (= 1 (.size queue)) true
|
||||
(> i 200) false
|
||||
:else (do (Thread/sleep 10) (recur (inc i)))))
|
||||
"the next sample must be scheduled after the error")
|
||||
(finally
|
||||
(px/shutdown-now sampler)))))))
|
||||
|
||||
@ -8,10 +8,12 @@
|
||||
(:require
|
||||
[app.metrics :as mtx]
|
||||
[clojure.test :as t]
|
||||
[integrant.core :as ig])
|
||||
[integrant.core :as ig]
|
||||
[promesa.exec :as px])
|
||||
(:import
|
||||
io.prometheus.client.Collector$MetricFamilySamples
|
||||
io.prometheus.client.Collector$MetricFamilySamples$Sample))
|
||||
io.prometheus.client.Collector$MetricFamilySamples$Sample
|
||||
java.util.concurrent.ScheduledThreadPoolExecutor))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; Helpers
|
||||
@ -47,3 +49,28 @@
|
||||
(t/is (contains? names "process_open_fds"))
|
||||
(t/is (contains? names "process_max_fds"))
|
||||
(t/is (contains? names "process_cpu_seconds_total"))))
|
||||
|
||||
(t/deftest create-sampler-reschedules-after-error
|
||||
;; The docstring promise, tested once for both call sites: an unexpected error
|
||||
;; on a single sample must not cancel the following runs. The first sample
|
||||
;; runs immediately and throws; the next one must still be scheduled.
|
||||
(let [calls (atom 0)
|
||||
sampler (mtx/create-sampler 1000 (fn []
|
||||
(swap! calls inc)
|
||||
(throw (ex-info "boom" {}))))]
|
||||
(try
|
||||
(t/is (loop [i 0]
|
||||
(cond (pos? @calls) true
|
||||
(> i 200) false
|
||||
:else (do (Thread/sleep 10) (recur (inc i)))))
|
||||
"the first sample must run immediately")
|
||||
|
||||
(let [queue (.getQueue ^ScheduledThreadPoolExecutor sampler)]
|
||||
(t/is (loop [i 0]
|
||||
(cond (= 1 (.size queue)) true
|
||||
(> i 200) false
|
||||
:else (do (Thread/sleep 10) (recur (inc i)))))
|
||||
"the next sample must be scheduled after the error"))
|
||||
|
||||
(finally
|
||||
(px/shutdown-now sampler)))))
|
||||
|
||||
@ -43,41 +43,40 @@
|
||||
;; `key-fn` of the vector form. That silently turned the limit into a single
|
||||
;; shared one, so one stuck call ended up rejecting the audit batches of
|
||||
;; every profile on the instance. Two profiles must not share a limiter.
|
||||
(let [climit (th/init-climit {:submit-audit-events/by-profile
|
||||
{:permits 1 :queue 1 :timeout 30000}})
|
||||
(th/with-climit [climit {:submit-audit-events/by-profile
|
||||
{:permits 1 :queue 1 :timeout 30000}}]
|
||||
(let [;; The metadata declared by the command itself, that is what the test
|
||||
;; has to cover.
|
||||
mdata (meta (ns-resolve 'app.rpc.commands.audit 'sm$push-audit-events))
|
||||
|
||||
;; The metadata declared by the command itself, that is what the test
|
||||
;; has to cover.
|
||||
mdata (meta (ns-resolve 'app.rpc.commands.audit 'sm$push-audit-events))
|
||||
gate (promise)
|
||||
entered (promise)
|
||||
calls (atom 0)
|
||||
handler (fn [_ _]
|
||||
(when (= 1 (swap! calls inc))
|
||||
(deliver entered true)
|
||||
@gate)
|
||||
:ok)
|
||||
wrapped (climit/wrap {::rpc/climit climit} handler mdata)
|
||||
call (fn [profile-id]
|
||||
(wrapped {::rpc/climit climit} {::rpc/profile-id profile-id}))
|
||||
pid-a (uuid/next)
|
||||
pid-b (uuid/next)
|
||||
holder (future (call pid-a))]
|
||||
|
||||
gate (promise)
|
||||
entered (promise)
|
||||
calls (atom 0)
|
||||
handler (fn [_ _]
|
||||
(when (= 1 (swap! calls inc))
|
||||
(deliver entered true)
|
||||
@gate)
|
||||
:ok)
|
||||
wrapped (climit/wrap {::rpc/climit climit} handler mdata)
|
||||
call (fn [profile-id]
|
||||
(wrapped {::rpc/climit climit} {::rpc/profile-id profile-id}))
|
||||
pid-a (uuid/next)
|
||||
pid-b (uuid/next)
|
||||
holder (future (call pid-a))]
|
||||
;; The first call stays inside the handler holding the only permit.
|
||||
(t/is (true? (deref entered 2000 false)))
|
||||
|
||||
;; The first call stays inside the handler holding the only permit.
|
||||
(t/is (true? (deref entered 2000 false)))
|
||||
;; Another call from the SAME profile shares the limiter, so it waits.
|
||||
(let [same (future (call pid-a))]
|
||||
(t/is (= :waiting (deref same 300 :waiting))))
|
||||
|
||||
;; Another call from the SAME profile shares the limiter, so it waits.
|
||||
(let [same (future (call pid-a))]
|
||||
(t/is (= :waiting (deref same 300 :waiting))))
|
||||
;; A different profile owns a different limiter and is served right away.
|
||||
;; With the shared limiter (the bug) this one waits too and times out.
|
||||
(t/is (= :ok (deref (future (call pid-b)) 2000 :timeout)))
|
||||
|
||||
;; A different profile owns a different limiter and is served right away.
|
||||
;; With the shared limiter (the bug) this one waits too and times out.
|
||||
(t/is (= :ok (deref (future (call pid-b)) 2000 :timeout)))
|
||||
|
||||
(deliver gate true)
|
||||
(t/is (= :ok (deref holder 2000 :timeout)))))
|
||||
(deliver gate true)
|
||||
(t/is (= :ok (deref holder 2000 :timeout))))))
|
||||
|
||||
(t/deftest prepare-context-initiator-is-plain-string
|
||||
;; The initiator must always be a plain string, never a keyword: shared-key
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user