path: root/src/com/github/ivarref/yoltq.clj

(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 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))))