refactor: handle logs async with chans

Author: J0sueTM

deps.edn

@@ -1,5 +1,6 @@
 {:paths ["resources" "src"]
  :deps  {org.clojure/clojure                   {:mvn/version "1.11.1"}
+         org.clojure/core.async                {:mvn/version "1.6.681"}
          com.moclojer/moclojer                 {:mvn/version "0.3.2"}
          com.moclojer/rq                       {:mvn/version "0.1.4"}
          clj-http/clj-http                     {:mvn/version "3.12.3"}

src/com/moclojer/components/logs.clj

@@ -1,18 +1,17 @@
 (ns com.moclojer.components.logs
   (:require
+   [clojure.core.async :as async]
    [clojure.data.json :as json]
    [clj-http.client :as http-client]
-   [taoensso.telemere :as t]
-   [taoensso.telemere.timbre :as timbre])
-  (:import (java.util.concurrent TimeoutException TimeUnit)))
+   [taoensso.telemere :as t])
+  (:import [clojure.core.async.impl.channels ManyToManyChannel]))
 
 (defn build-opensearch-base-req
   [config]
   (let [{:keys [username password host port]} config
         url (str "https://" host ":" port "/_bulk")]
     {:method :post
      :url url
-     :async? true
      :basic-auth [username password]
      :content-type :json
      :body (json/write-str {:index {:_index "logs"}})}))
@@ -27,33 +26,46 @@
                     :else (pr-str v))))
    {} m))
 
-(defn send-opensearch-signal-req
-  [base-req signal]
-  (let [log (-> (select-keys signal [:level :ctx :data
-                                     :msg_ :error :thread
-                                     :uid :inst])
-                (->str-values)
-                (json/write-str))
-        future (http-client/request
-                (update
-                 base-req :body
-                 str \newline log \newline)
-                identity #(throw ^Exception %))]
-    (try
-      (.get future 1 TimeUnit/SECONDS)
-      (catch TimeoutException _
-        (.cancel future true)))))
+(defn send-opensearch-log-req
+  [base-req log]
+  (http-client/request
+   (update
+    base-req :body
+    str \newline (json/write-str (->str-values log)) \newline)
+   identity #(throw ^Exception %)))
+
+(defonce log-ch (atom (async/chan)))
 
 (defn setup [config level env]
-  (let [prod? (= env "prod")
-        os-cfg (when prod? (:opensearch config))
-        os-base-req (build-opensearch-base-req os-cfg)]
+  (let [prod? (= env :prod)
+        log-ch' (swap!
+                 log-ch
+                 (fn [ch]
+                   (when ch (async/close! ch))
+                   (async/chan)))]
+
     (t/set-min-level! level)
-    (t/set-ns-filter! {:disallow "*" :allow "com.moclojer.*"})
-    (when prod?
-      (telemere/add-signal!
-       :opensearch
-       (fn [signal])))))
+
+    (when (and prod? (instance? ManyToManyChannel log-ch'))
+      (let [os-cfg (when prod? (:opensearch config))
+            os-base-req (build-opensearch-base-req os-cfg)]
+
+        (t/set-ns-filter! {:disallow #{"*jetty*" "*hikari*"
+                                       "*pedestal*" "*migratus*"}})
+
+        (t/add-handler!
+         :opensearch
+         (fn [signal]
+           (async/go
+             (async/>!
+              log-ch'
+              (select-keys signal [:level :ctx :data :msg_ :error
+                                   :thread :uid :inst])))))
+
+        (async/go
+          (while true
+            (let [[log _] (async/alts! [log-ch'])]
+              (send-opensearch-log-req os-base-req log))))))))
 
 (comment
   (def my-signal
@@ -70,8 +82,6 @@
       :host "foobar"
       :port 25060}))
 
-  (send-opensearch-signal-req base-req my-signal)
-
   (http-client/request
    (update
     base-req :body
@@ -82,15 +92,31 @@
   ;;
   )
 
-(defmacro log [level & args]
-  `(timbre/log ~level ~@args))
+(defn log [level msg & [:as data]]
+  (t/log! {:level level
+           :data (first data)}
+          (str msg)))
+
+(comment
+  ;; DEPRECATED
+  (defmacro log [level & args]
+    `(timbre/log ~level ~@args))
+  ;;
+  )
 
 (defn gen-ctx-with-cid []
   {:cid (str "cid-" (random-uuid) "-" (System/currentTimeMillis))})
 
 (comment
-  (setup :dev)
+  log-ch
+  
+  (setup {:opensearch
+          {:username "foobar"
+           :password "foobar"
+           :host "foobar"
+           :port 25060}}
+         :info :prod)
 
-  (log :info :world)
+  (log :info "something happened" {:user "j0suetm"})
   ;;
   )