mirror of
https://github.com/penpot/penpot.git
synced 2026-11-06 01:50:39 +00:00
5c476b4d9c
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
278 lines
9.2 KiB
Clojure
278 lines
9.2 KiB
Clojure
;; This Source Code Form is subject to the terms of the Mozilla Public
|
|
;; License, v. 2.0. If a copy of the MPL was not distributed with this
|
|
;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
|
|
;;
|
|
;; Copyright (c) KALEIDOS INC Sucursal en España SL
|
|
|
|
(ns app.http
|
|
(:require
|
|
[app.auth.oidc :as-alias oidc]
|
|
[app.common.data :as d]
|
|
[app.common.logging :as l]
|
|
[app.common.schema :as sm]
|
|
[app.common.transit :as t]
|
|
[app.db :as-alias db]
|
|
[app.http.access-token :as actoken]
|
|
[app.http.assets :as-alias assets]
|
|
[app.http.awsns :as-alias awsns]
|
|
[app.http.debug :as-alias debug]
|
|
[app.http.errors :as errors]
|
|
[app.http.management :as mgmt]
|
|
[app.http.middleware :as mw]
|
|
[app.http.security :as sec]
|
|
[app.http.session :as session]
|
|
[app.http.websocket :as-alias ws]
|
|
[app.main :as-alias main]
|
|
[app.metrics :as mtx]
|
|
[app.rpc :as-alias rpc]
|
|
[app.setup :as-alias setup]
|
|
[integrant.core :as ig]
|
|
[promesa.exec :as px]
|
|
[reitit.core :as r]
|
|
[reitit.middleware :as rr]
|
|
[yetti.adapter :as yt]
|
|
[yetti.request :as yreq]
|
|
[yetti.response :as-alias yres])
|
|
(:import
|
|
io.undertow.server.ConnectorStatistics
|
|
io.undertow.Undertow
|
|
org.xnio.management.XnioWorkerMXBean))
|
|
|
|
(declare router-handler)
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; HTTP SERVER
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(def default-params
|
|
{::port 6060
|
|
::host "0.0.0.0"
|
|
::max-body-size 367001600 ; default 350 MiB
|
|
})
|
|
|
|
(def ^:private metrics-sample-interval-ms 15000)
|
|
|
|
(defn sample-worker-metrics
|
|
"Publishes the current state of the xnio worker thread pool (the
|
|
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)
|
|
pool-size (.getWorkerPoolSize mxbean)
|
|
max-size (.getMaxWorkerPoolSize mxbean)]
|
|
|
|
;; 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 (>= pool-size 0)
|
|
(mtx/run! metrics :id :http-worker-pool-size :val pool-size))
|
|
|
|
(when (>= max-size 0)
|
|
(mtx/run! metrics :id :http-worker-max-pool-size :val max-size))))
|
|
|
|
(defn sample-connector-metrics
|
|
"Publishes the current state of the http listener connection
|
|
statistics. Undertow exposes absolute totals, so counters are
|
|
published as deltas of the last seen values (the atom state holds the
|
|
last observed totals). When a delta comes back negative (mainly
|
|
because the underlying counters were reset) the counter is skipped
|
|
and the reference updated."
|
|
[metrics state ^ConnectorStatistics cs]
|
|
(let [{:keys [last-requests last-errors]} (deref state)
|
|
total-requests (.getRequestCount cs)
|
|
total-errors (.getErrorCount cs)
|
|
delta-requests (max 0 (- total-requests last-requests))
|
|
delta-errors (max 0 (- total-errors last-errors))]
|
|
|
|
(when (pos? delta-requests)
|
|
(mtx/run! metrics :id :http-connector-requests-total :inc delta-requests))
|
|
|
|
(when (pos? delta-errors)
|
|
(mtx/run! metrics :id :http-connector-errors-total :inc delta-errors))
|
|
|
|
(mtx/run! metrics
|
|
:id :http-connector-active-connections
|
|
:val (.getActiveConnections cs))
|
|
|
|
(swap! state merge {:last-requests total-requests
|
|
:last-errors total-errors})))
|
|
|
|
(defn sample-http-metrics
|
|
"Samples the current state of the http server: worker thread pool
|
|
state and listener connection statistics. Called periodically by a
|
|
sampler that starts together with the server."
|
|
[metrics state ^Undertow server]
|
|
(try
|
|
(when-let [mxbean (some-> server (.getWorker) (.getMXBean))]
|
|
(sample-worker-metrics metrics mxbean))
|
|
|
|
(when-let [cs (some-> server (.getListenerInfo) (first) (.getConnectorStatistics))]
|
|
(sample-connector-metrics metrics state cs))
|
|
|
|
(catch Exception cause
|
|
(l/warn :msg "unexpected error on http metrics sampling"
|
|
:cause cause))))
|
|
|
|
(defn create-metrics-sampler
|
|
"Creates a daemon scheduler that periodically samples the state of
|
|
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})]
|
|
(mtx/create-sampler metrics-sample-interval-ms
|
|
(fn [] (sample-http-metrics metrics state server))
|
|
{:name "http-metrics"})))
|
|
|
|
(defmethod ig/expand-key ::server
|
|
[k v]
|
|
{k (merge default-params (d/without-nils v))})
|
|
|
|
(def ^:private schema:server-params
|
|
[:map
|
|
[::port ::sm/int]
|
|
[::host ::sm/text]
|
|
[::io-threads {:optional true} ::sm/int]
|
|
[::max-worker-threads {:optional true} ::sm/int]
|
|
[::max-body-size {:optional true} ::sm/int]
|
|
[::router {:optional true} [:fn r/router?]]
|
|
[::handler {:optional true} ::sm/fn]])
|
|
|
|
(defmethod ig/assert-key ::server
|
|
[_ params]
|
|
(assert (sm/check schema:server-params params)))
|
|
|
|
(defmethod ig/init-key ::server
|
|
[_ {:keys [::handler ::router ::host ::port ::mtx/metrics] :as cfg}]
|
|
(l/info :hint "starting http server" :port port :host host)
|
|
(let [on-dispatch
|
|
(fn [_ start-at-ns]
|
|
(let [timing (- (System/nanoTime) start-at-ns)
|
|
timing (int (/ timing 1000000))]
|
|
(mtx/run! metrics
|
|
:id :http-server-dispatch-timing
|
|
:val timing)))
|
|
|
|
options
|
|
{:http/port port
|
|
:http/host host
|
|
:http/max-body-size (::max-body-size cfg)
|
|
:http/max-multipart-body-size (::max-body-size cfg)
|
|
:xnio/direct-buffers false
|
|
:xnio/io-threads (::io-threads cfg)
|
|
:xnio/max-worker-threads (::max-worker-threads cfg)
|
|
:ring/compat :ring2
|
|
:server/statistics true
|
|
:events/on-dispatch on-dispatch
|
|
:socket/backlog 4069}
|
|
|
|
handler
|
|
(cond
|
|
(some? router)
|
|
(router-handler router)
|
|
|
|
(some? handler)
|
|
handler
|
|
|
|
:else
|
|
(throw (UnsupportedOperationException. "handler or router are required")))
|
|
|
|
server
|
|
(yt/start! (yt/server handler (d/without-nils options)))
|
|
|
|
sampler
|
|
(create-metrics-sampler server metrics)]
|
|
|
|
(assoc cfg ::server server ::metrics-sampler sampler)))
|
|
|
|
(defmethod ig/halt-key! ::server
|
|
[_ {:keys [::metrics-sampler ::server ::port] :as cfg}]
|
|
(l/info :msg "stopping http server" :port port)
|
|
(px/shutdown-now metrics-sampler)
|
|
(yt/stop! server))
|
|
|
|
(defn- not-found-handler
|
|
[_]
|
|
{::yres/status 404})
|
|
|
|
(defn- router-handler
|
|
[router]
|
|
(letfn [(resolve-handler [request]
|
|
(if-let [match (r/match-by-path router (yreq/path request))]
|
|
(let [params (:path-params match)
|
|
result (:result match)
|
|
handler (or (:handler result) not-found-handler)
|
|
request (assoc request :path-params params)]
|
|
(partial handler request))
|
|
(partial not-found-handler request)))
|
|
|
|
(on-error [cause request]
|
|
(let [{:keys [::yres/body] :as response} (errors/handle cause request)]
|
|
(cond-> response
|
|
(map? body)
|
|
(-> (update ::yres/headers assoc "content-type" "application/transit+json")
|
|
(assoc ::yres/body (t/encode-str body {:type :json-verbose}))))))]
|
|
|
|
(fn [request]
|
|
(let [handler (resolve-handler request)]
|
|
(try
|
|
(handler)
|
|
(catch Throwable cause
|
|
(on-error cause request)))))))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; HTTP ROUTER
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(def ^:private schema:routes
|
|
[:vector :any])
|
|
|
|
(def ^:private schema:router-params
|
|
[:map
|
|
[::ws/routes schema:routes]
|
|
[::rpc/routes schema:routes]
|
|
[::oidc/routes schema:routes]
|
|
[::assets/routes schema:routes]
|
|
[::debug/routes schema:routes]
|
|
[::mtx/routes schema:routes]
|
|
[::awsns/routes schema:routes]
|
|
[::mgmt/routes schema:routes]
|
|
::session/manager
|
|
::setup/props
|
|
::db/pool])
|
|
|
|
(defmethod ig/assert-key ::router
|
|
[_ params]
|
|
(assert (sm/check schema:router-params params)))
|
|
|
|
(defmethod ig/init-key ::router
|
|
[_ cfg]
|
|
(rr/router
|
|
[["" {:middleware [[mw/server-timing]
|
|
[sec/sec-fetch-metadata]
|
|
[mw/params]
|
|
[mw/format-response]
|
|
[mw/auth {:bearer (partial session/decode-token cfg)
|
|
:cookie (partial session/decode-token cfg)
|
|
:token (partial actoken/decode-token cfg)}]
|
|
[mw/parse-request]
|
|
[mw/errors errors/handle]
|
|
[mw/restrict-methods]]}
|
|
|
|
(::mtx/routes cfg)
|
|
(::assets/routes cfg)
|
|
(::debug/routes cfg)
|
|
|
|
["/webhooks"
|
|
(::awsns/routes cfg)]
|
|
|
|
["/management"
|
|
(::mgmt/routes cfg)]
|
|
|
|
(::ws/routes cfg)
|
|
(::oidc/routes cfg)
|
|
(::rpc/routes cfg)]]))
|