Report on any jobs currently in the "FAILED" state

kc-jobs-report.clj

(ns grumpybank.observability.kc
  (:require
   [aleph.http :as http]
   [manifold.deferred :as d]
   [clojure.data.json :as json]
   [byte-streams :as bs]))

(defn connectors
  [connect-url]
  (d/chain (http/get (format "%s/connectors" connect-url))
    #(update % :body bs/to-string)
    #(update % :body json/read-str)
    #(:body %)))

(defn connector-status
  [connect-url connector]
  (d/chain (http/get (format "%s/connectors/%s/status"
                             connect-url
                             connector))
    #(update % :body bs/to-string)
    #(update % :body json/read-str)
    #(:body %)))

(defn failed?
  [status]
  (some #(= "FAILED" (get % "state")) (get status "tasks")))

(defn traces
  [status]
  (->> (get status "tasks")
       (filter #(= "FAILED" (get % "state")))
       (map #(get % "trace"))))
         
(defn connector-report
  [connect-url]
  (let [task-failed? #(= "FAILED" (get % "state"))
        task-running? #(= "RUNNING" (get % "state"))
        task-paused? #(= "PAUSED" (get % "state"))]
    (d/chain (connectors connect-url)
      #(apply d/zip (map (partial connector-status connect-url) %))
      #(map (fn [s]
              {:connector (get s "name")
               :failed? (failed? s)
               :total-tasks (count (get s "tasks"))
               :failed-tasks (->> (get s "tasks")
                                  (filter task-failed?)
                                  count)
               :running-tasks (->> (get s "tasks")
                                   (filter task-running?)
                                   count)
               :paused-tasks (->> (get s "tasks")
                                  (filter task-paused?)
                                  count)
             :trace (when (failed? s)
                      (traces s))}) %))))

(defn failed-connectors
  [report]
  (->> report
       (filter #(:failed? %))))

(defn format-connector-report
  [result]
  (-> (format "Connector: %s
Trace: %s"
          (:connector result)
          (clojure.string/join "\n" (:trace result)))
      println))

(defn format-connector-stats
  [stats]
  (-> (format "Summary:
Total Jobs: %s
Total Tasks: %s
By Status
  RUNNING: %s
  FAILED: %s
  PAUSED: %s"
              (:jobs stats)
              (:tasks stats)
              (:running stats)
              (:failed stats)
              (:paused stats))
      println))

(defn report-stats
  [report]
  (let [init {:jobs 0
              :tasks 0
              :failed 0
              :running 0
              :paused 0}]
    (reduce (fn [m job]
              (-> m
                  (update :jobs inc)
                  (update :tasks + (:total-tasks job))
                  (update :failed + (:failed-tasks job))
                  (update :running + (:running-tasks job))
                  (update :paused + (:paused-tasks job))))
            init
            report)))

(defn -main [& args]
  (let [connect-url "https://connect.grumpybank.co.uk"
        report (connector-report connect-url)]
    (->> (failed-connectors @report)
         (map format-connector-report)
         doall)

    (let [stats (report-stats @report)]
      (format-connector-stats stats))))