(ns com.github.ivarref.yoltq (:require [clojure.tools.logging :as log] [com.github.ivarref.yoltq.error-poller :as errpoller] [com.github.ivarref.yoltq.impl :as i] [com.github.ivarref.yoltq.migrate :as migrate] [com.github.ivarref.yoltq.poller :as poller] [com.github.ivarref.yoltq.report-queue :as rq] [com.github.ivarref.yoltq.slow-executor-detector :as slow-executor] [com.github.ivarref.yoltq.utils :as u] [datomic.api :as d]) (:import (datomic Connection) (java.lang.management ManagementFactory) (java.time Duration Instant ZoneOffset ZonedDateTime) (java.util.concurrent BlockingQueue ExecutorService Executors ScheduledExecutorService TimeUnit))) (defonce ^:dynamic *config* (atom nil)) (defonce threadpool (atom nil)) (defonce ^:dynamic *running?* (atom false)) (defonce ^:dynamic *test-mode* false) (def default-opts (-> {; Default number of times a queue job will be retried before giving up ; Can be overridden on a per-consumer basis with ; (yq/add-consumer! :q (fn [payload] ...) {:max-retries 200}) ; If you want no limit on the number of retries, specify ; the value `0`. That will set the effective retry limit to ; 9223372036854775807 times. :max-retries 9223372036854775807 ; Minimum amount of time to wait before a failed queue job is retried :error-backoff-time (Duration/ofSeconds 5) ; Max time a queue job can execute before an error is logged :max-execute-time (Duration/ofMinutes 5) ; Amount of time an in progress queue job can run before it is considered failed ; and will be marked as such. :hung-backoff-time (Duration/ofMinutes 30) ; Most queue jobs in init state will be consumed by the tx-report-queue listener. ; However, in the case where an init job was added right before the application ; was shut down and did not have time to be processed by the tx-report-queue listener, ; it will be consumer by the init poller. This init poller backs off by ; :init-backoff-time to avoid unnecessary compare-and-swap lock failures that could ; otherwise occur if competing with the tx-report-queue listener. :init-backoff-time (Duration/ofSeconds 60) ; If you are dealing with a flaky downstream service, you may not want ; yoltq to mark itself as unhealthy on the first failure encounter with ; the downstream service. Change this setting to let yoltq mark itself ; as healthy even though a queue item has been failing for some time. :healthy-allowed-error-time (Duration/ofMinutes 15) ; How frequent polling for init, error and hung jobs should be done. :poll-delay (Duration/ofSeconds 10) ; Specifies the number of threads available for executing queue and polling jobs. ; The final thread pool will be this size + 2. ; ; One thread is permanently allocated for listening to the ; tx-report-queue. ; ; Another thread is permanently allocated for checking :max-execute-time. ; This means that if all executing queue jobs are stuck and the thread pool is unavailable ; as such, at least an error will be logged about this. The log entry will ; contain the stacktrace of the stuck threads. :pool-size 4 :capture-bindings (if-let [s (resolve (symbol "taoensso.timbre/*context*"))] [s] []) ; How often should the system be polled for failed queue jobs :system-error-poll-delay (Duration/ofMinutes 1) ; How often should the system invoke :system-error-callback-backoff (Duration/ofHours 1) ; Should old, possibly stalled jobs be automatically be migrated ; as part of `start!`? :auto-migrate? true} u/duration->millis)) (defn init! [{:keys [conn tx-report-queue] :as cfg}] (assert (instance? Connection conn) (str "Expected :conn to be of type datomic Connection. Was: " (or (some-> conn class str) "nil"))) (when (some? tx-report-queue) (assert (instance? BlockingQueue tx-report-queue) (str "Expected :tx-report-queue to be of type java.util.concurrent.BlockingQueue"))) (locking threadpool @(d/transact conn i/schema) (let [new-cfg (swap! *config* (fn [old-conf] (-> (merge-with (fn [_ b] b) {:running-queues (atom #{}) :start-execute-time (atom {}) :system-error (atom {}) :healthy? (atom nil) :slow? (atom nil) :slow-thread-watcher-done? (promise)} default-opts (if *test-mode* old-conf (select-keys old-conf [:handlers])) cfg) u/duration->millis)))] new-cfg))) (defn get-queue-id [queue-id-or-var] (cond (and (var? queue-id-or-var) (keyword? (:yoltq/queue-id (meta queue-id-or-var)))) (:yoltq/queue-id (meta queue-id-or-var)) (keyword? queue-id-or-var) queue-id-or-var :else (throw (ex-info (str "Could not get queue-id for " queue-id-or-var) {:queue-id queue-id-or-var})))) (defn add-consumer! ([queue-id f] (add-consumer! queue-id f {})) ([queue-id f opts] (swap! *config* (fn [old-config] (assoc-in old-config [:handlers (get-queue-id queue-id)] (merge opts {:f f})))))) (defn put ([queue-id payload] (put queue-id payload {})) ([queue-id payload opts] (let [{:keys [bootstrap-poller! conn] :as cfg} @*config*] (when (and *test-mode* bootstrap-poller!) (bootstrap-poller! conn)) (i/put cfg (get-queue-id queue-id) payload opts)))) (defn- do-start! [] (let [{:keys [poll-delay pool-size system-error-poll-delay auto-migrate? slow-thread-watcher-done?] :as cfg} @*config*] (when auto-migrate? (future (migrate/migrate! cfg))) (let [pool (reset! threadpool (Executors/newScheduledThreadPool (+ 1 pool-size))) queue-listener-ready (promise)] (reset! *running?* true) (.scheduleAtFixedRate ^ScheduledExecutorService pool (fn [] (poller/poll-all-queues! *running?* *config* pool)) 0 poll-delay TimeUnit/MILLISECONDS) (.scheduleAtFixedRate ^ScheduledExecutorService pool (fn [] (errpoller/poll-errors *running?* *config*)) 0 system-error-poll-delay TimeUnit/MILLISECONDS) (.execute ^ScheduledExecutorService pool (fn [] (try (log/debug "report-queue-listener starting") (rq/report-queue-listener *running?* queue-listener-ready pool *config*) (finally (log/debug "report-queue-listener exiting") (deliver queue-listener-ready :finally))))) (future (try (slow-executor/show-slow-threads pool *config*) (finally (deliver slow-thread-watcher-done? :done)))) (let [q-listener-retval (deref queue-listener-ready 30000 :timeout)] (cond (= :timeout q-listener-retval) (do (log/error "Timed out waiting for report-queue-listener to start") (throw (IllegalStateException. "Timed out waiting for report-queue-listener to start"))) (= :finally q-listener-retval) (do (log/error "report-queue-listener did not start") (throw (IllegalStateException. "report-queue-listener did not start"))) (= :ready q-listener-retval) (do (log/debug "report-queue-listener is ready")) :else (do (log/error (str "Unexpected queue-listener-retval: " (pr-str q-listener-retval))) (throw (IllegalStateException. (str "Unexpected queue-listener-retval: " (pr-str q-listener-retval)))))))))) (defn start! [] (locking threadpool (cond (true? *test-mode*) (log/info "test mode enabled, doing nothing for start!") (true? @*running?*) nil (false? @*running?*) (do-start!)))) (defn stop! [] (locking threadpool (cond (true? *test-mode*) (log/info "test mode enabled, doing nothing for stop!") (false? @*running?*) nil (true? @*running?*) (do (reset! *running?* false) (when-let [^ExecutorService tp @threadpool] (log/debug "shutting down threadpool") (.shutdown tp) (while (not (.awaitTermination tp 1 TimeUnit/SECONDS)) (log/trace "waiting for threadpool to stop")) (log/debug "stopped!") (reset! threadpool nil)) (when-let [wait-slow-threads (some->> *config* deref :slow-thread-watcher-done?)] (log/debug "waiting for slow-thread-watcher to stop ...") @wait-slow-threads (log/debug "waiting for slow-thread-watcher to stop ... OK")))))) (defn healthy? [] (cond (< (.toMinutes (Duration/ofMillis (.getUptime (ManagementFactory/getRuntimeMXBean)))) 10) true (false? (some->> @*config* :healthy? (deref))) false (true? (some->> @*config* :slow? (deref))) false :else true)) (defn unhealthy? "Returns `true` if there are queues in error or a thread is slow, and the application has been up for over 10 minutes, otherwise `false`." [] (false? (healthy?))) (defn queue-stats [] (let [{:keys [conn]} @*config* db (d/db conn)] (->> (d/q '[:find ?e ?qname ?status :in $ :where [?e :com.github.ivarref.yoltq/queue-name ?qname] [?e :com.github.ivarref.yoltq/status ?status]] db) (mapv (partial zipmap [:e :qname :status])) (mapv #(select-keys % [:qname :status])) (mapv (fn [qitem] {qitem 1})) (reduce (partial merge-with +) {}) (mapv (fn [[{:keys [qname status]} v]] (array-map :qname qname :status status :count v))) (sort-by (juxt :qname :status)) (vec)))) (defn get-errors [qname] (let [{:keys [conn]} @*config* db (d/db conn)] (->> (d/q '[:find [?id ...] :in $ ?qname ?status :where [?e :com.github.ivarref.yoltq/queue-name ?qname] [?e :com.github.ivarref.yoltq/status ?status] [?e :com.github.ivarref.yoltq/id ?id]] db qname :error) (mapv (partial u/get-queue-item db))))) (defn retry-one-error! [qname] (let [{:keys [handlers] :as cfg} @*config* _ (assert (contains? handlers qname) "Queue not found") cfg (assoc-in cfg [:handlers qname :max-retries] Long/MAX_VALUE)] (poller/poll-once! cfg qname :error))) (defn retry-stats "Gather retry statistics. Optional keyword arguments: * :age-days — last number of days to look at data from. Defaults to 30. * :queue-name — only gather statistics for this queue name. Defaults to nil, meaning all queues. Example return value: {:queue-a {:ok 100, :retries 2, :retry-percentage 2.0} :queue-b {:ok 100, :retries 75, :retry-percentage 75.0}} From the example value above, we can see that :queue-b fails at a much higher rate than :queue-a. Assuming that the queue consumers are correctly implemented, this means that the service representing :queue-b is much more unstable than the one representing :queue-a. This again implies that you will probably want to fix the downstream service of :queue-b, if that is possible. " [{:keys [age-days queue-name now db] :or {age-days 30 now (ZonedDateTime/now ZoneOffset/UTC)}}] (let [{:keys [conn]} @*config* db (or db (d/db conn))] (->> (d/query {:query {:find '[?qname ?status ?tries ?init-time] :in (into '[$] (when queue-name '[?qname])) :where '[[?e :com.github.ivarref.yoltq/queue-name ?qname] [?e :com.github.ivarref.yoltq/status ?status] [?e :com.github.ivarref.yoltq/tries ?tries] [?e :com.github.ivarref.yoltq/init-time ?init-time]]} :args (remove nil? [db queue-name])}) (mapv (partial zipmap [:qname :status :tries :init-time])) (mapv #(update % :init-time (fn [init-time] (.atZone (Instant/ofEpochMilli init-time) ZoneOffset/UTC)))) (mapv #(assoc % :age-days (.toDays (Duration/between (:init-time %) now)))) (filter #(<= (:age-days %) age-days)) (group-by :qname) (mapv (fn [[q values]] {q (let [{:keys [ok retries] :as m} (->> values (mapv (fn [{:keys [tries status]}] (condp = status u/status-init {} u/status-processing {:processing 1 :retries (dec tries)} u/status-done {:ok 1 :retries (dec tries)} u/status-error {:error 1 :retries (dec tries)}))) (reduce (partial merge-with +) {}))] (into (sorted-map) (merge m (when (pos-int? ok) {:retry-percentage (double (* 100 (/ retries ok)))}))))})) (into (sorted-map))))) (defn- percentile [n values] (let [idx (int (Math/floor (* (count values) (/ n 100))))] (nth values idx))) (defn processing-time-stats "Gather processing time statistics. Default unit is seconds. Optional keyword arguments: * :age-days — last number of days to look at data from. Defaults to 30. Use nil to have no limit. * :queue-name — only gather statistics for this queue name. Defaults to nil, meaning all queues. * :duration->long - Specify what unit should be used for values. Must take a java.time.Duration as input and return a long. Defaults to (fn [duration] (.toSeconds duration). I.e. the default unit is seconds. Example return value: {:queue-a {:avg 1 :max 10 :min 0 :p50 ... :p90 ... :p95 ... :p99 ...}}" [{:keys [age-days queue-name now db duration->long] :or {age-days 30 now (ZonedDateTime/now ZoneOffset/UTC) duration->long (fn [duration] (.toSeconds ^Duration duration))}}] (let [{:keys [conn]} @*config* db (or db (d/db conn)) ->zdt #(.atZone (Instant/ofEpochMilli %) ZoneOffset/UTC)] (->> (d/query {:query {:find '[?qname ?status ?init-time ?done-time] :in (into '[$ ?status] (when queue-name '[?qname])) :where '[[?e :com.github.ivarref.yoltq/queue-name ?qname] [?e :com.github.ivarref.yoltq/status ?status] [?e :com.github.ivarref.yoltq/init-time ?init-time] [?e :com.github.ivarref.yoltq/done-time ?done-time]]} :args (vec (remove nil? [db u/status-done queue-name]))}) (mapv (partial zipmap [:qname :status :init-time :done-time])) (mapv #(update % :init-time ->zdt)) (mapv #(update % :done-time ->zdt)) (mapv #(assoc % :age-days (.toDays (Duration/between (:init-time %) now)))) (mapv #(assoc % :spent-time (duration->long (Duration/between (:init-time %) (:done-time %))))) (filter #(or (nil? age-days) (<= (:age-days %) age-days))) (group-by :qname) (mapv (fn [[q values]] (let [values (vec (sort (mapv :spent-time values)))] {q (sorted-map :max (apply max values) :avg (int (Math/floor (/ (reduce + 0 values) (count values)))) :p50 (percentile 50 values) :p90 (percentile 90 values) :p95 (percentile 95 values) :p99 (percentile 99 values) :min (apply min values))}))) (into (sorted-map))))) (defn get-tx-report-queue-multicast! "Multicast the datomic.api/tx-report-queue to different consumers. A multicaster thread is started on demand per connection. `conn` and `id` identifies the consumer. Repeated calls using the same `conn` and `id` returns the same queue. The optional third parameter, `send-end-token?`, if set to `true`, instructs the multicaster thread to send `:end` if the queue is stopped. The default value for `send-end-token?` is `false`. A queue may be stopped using `stop-multicaster-id!`, `stop-multicaster!` or `stop-all-multicasters!`. Returns a `java.util.concurrent.BlockingQueue` like `datomic.api/tx-report-queue`." ([conn id] (get-tx-report-queue-multicast! conn id false)) ([conn id send-end-token?] (assert (instance? Connection conn)) (assert (boolean? send-end-token?)) (rq/get-tx-report-queue-multicast! conn id send-end-token?))) (defn stop-multicast-consumer-id! "Stop forwarding reports from datomic.api/tx-report-queue to the queue identified by `conn` and `id`. If this is the last report destination for the given `conn`, the multicaster thread will exit. Repeated calls are no-op. The multicaster thread will send `:end` if `send-end-token?` was `true` when `get-tx-report-queue-multicast!` was called. Returns `true` if the queue was stopped. Return `false` if the queue does not exist." [conn id] (assert (instance? Connection conn)) (rq/stop-multicast-consumer-id! conn id)) (defn stop-multicaster! "Stop forwarding reports from datomic.api/tx-report-queue to any queues belonging to `conn`. The multicaster thread will exit. Repeated calls are no-op. The multicaster thread will send `:end` if `send-end-token?` was `true` when `get-tx-report-queue-multicast!` was called. Returns `true` if any queue belonging to `conn` was stopped. Returns `false` is `conn` did not have any associated queues." [conn] (assert (instance? Connection conn)) (rq/stop-multicaster! conn)) (defn stop-all-multicasters! "Stop forwarding all reports from datomic.api/tx-report-queue for any `conn`. All multicaster threads will exit. Repeated calls are no-op. The multicaster thread will send `:end` if `send-end-token?` was `true` when `get-tx-report-queue-multicast!` was called. Returns `true` if any queue was stopped. Returns `false` if no queues existed." [] (rq/stop-all-multicasters!)) (comment (do (require 'com.github.ivarref.yoltq.log-init) (com.github.ivarref.yoltq.log-init/init-logging! [[#{"datomic.*" "com.datomic.*" "org.apache.*"} :warn] [#{"ivarref.yoltq.report-queue"} :info] [#{"ivarref.yoltq.poller"} :info] [#{"com.github.ivarref.yoltq"} :debug] ;[#{"ivarref.yoltq*"} :info] [#{"*"} :info]]) (stop!) (future (let [received (atom []) uri (str "datomic:mem://demo")] (d/delete-database uri) (d/create-database uri) (let [conn (d/connect uri) started-consuming? (promise) n 1] (init! {:conn conn :error-backoff-time (Duration/ofSeconds 1) :poll-delay (Duration/ofSeconds 1) :max-execute-time (Duration/ofSeconds 3) :slow-thread-show-stacktrace? false}) (add-consumer! :q (fn [_] (deliver started-consuming? true) (log/info "sleeping...") (Thread/sleep (.toMillis (Duration/ofSeconds 60))) (log/info "done sleeping"))) (start!) @(d/transact conn [(put :q {:work 123})]) @started-consuming? (stop!) nil))))) (comment (do (require 'com.github.ivarref.yoltq.log-init) (com.github.ivarref.yoltq.log-init/init-logging! [[#{"datomic.*" "com.datomic.*" "org.apache.*"} :warn] [#{"com.github.ivarref.yoltq.report-queue"} :debug] [#{"com.github.ivarref.yoltq.poller"} :info] [#{"com.github.ivarref.yoltq.migrate"} :warn] [#{"com.github.ivarref.yoltq"} :debug] ;[#{"ivarref.yoltq*"} :info] [#{"*"} :info]]) (stop!) (let [received (atom []) uri (str "datomic:mem://demo")] (d/delete-database uri) (d/create-database uri) (let [conn (d/connect uri) started-consuming? (promise) n 1] (init! {:conn conn :tx-report-queue (get-tx-report-queue-multicast! conn :yoltq true) :slow-thread-show-stacktrace? false}) (add-consumer! :q (fn [_] (deliver started-consuming? true))) (log/info "begin start! ...") (start!) (log/info "begin start! ... Done") (Thread/sleep 2000) (log/info "*******************************************") @(d/transact conn [(put :q {:work 123})]) @started-consuming? (stop-multicaster! conn) (log/info "*******************************************") (stop!) (log/info "stop! done") nil))))