From c1476d0397333e4a321d5a37693eae9198575415 Mon Sep 17 00:00:00 2001
From: Andrey Antukh <niwi@niwi.nz>
Date: Mon, 15 Feb 2021 13:14:05 +0100
Subject: [PATCH] :tada: Add optional loki integration.

And refactor internal error reporting.
---
 CHANGES.md                                    |   2 +
 backend/deps.edn                              |   3 +-
 backend/dev/user.clj                          |   8 +-
 backend/resources/error-report.tmpl           |  38 ++++--
 backend/resources/log4j2.xml                  |  17 ++-
 backend/scripts/repl                          |   7 +-
 backend/src/app/config.clj                    |  13 +-
 backend/src/app/http/errors.clj               |   7 --
 backend/src/app/loggers/loki.clj              |  92 ++++++++++++++
 .../mattermost.clj}                           | 118 +++++++++---------
 backend/src/app/loggers/zmq.clj               |  92 ++++++++++++++
 backend/src/app/main.clj                      |  15 ++-
 backend/src/app/util/json.clj                 |   8 ++
 backend/src/app/util/time.clj                 |   4 +
 backend/src/app/worker.clj                    |   5 -
 15 files changed, 331 insertions(+), 98 deletions(-)
 create mode 100644 backend/src/app/loggers/loki.clj
 rename backend/src/app/{error_reporter.clj => loggers/mattermost.clj} (57%)
 create mode 100644 backend/src/app/loggers/zmq.clj

diff --git a/CHANGES.md b/CHANGES.md
index 0de1f25a4..326b99e2c 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -4,10 +4,12 @@
 
 ### New features
 
+- Add optional loki integration.
 - Bounce & Complaint handling.
 - Disable groups interactions when holding "Ctrl" key (deep selection)
 - New action in context menu to "edit" some shapes (binded to key "Enter")
 
+
 ### Bugs fixed
 
 - Properly handle errors on github, gitlab and ldap auth backends.
diff --git a/backend/deps.edn b/backend/deps.edn
index c178b10a8..2452df5d9 100644
--- a/backend/deps.edn
+++ b/backend/deps.edn
@@ -16,6 +16,8 @@
   org.apache.logging.log4j/log4j-jul {:mvn/version "2.14.0"}
   org.apache.logging.log4j/log4j-slf4j-impl {:mvn/version "2.14.0"}
   org.slf4j/slf4j-api {:mvn/version "1.7.30"}
+  org.zeromq/jeromq {:mvn/version "0.5.2"}
+
 
   org.graalvm.js/js {:mvn/version "20.3.0"}
   com.taoensso/nippy {:mvn/version "3.1.1"}
@@ -43,7 +45,6 @@
   org.postgresql/postgresql {:mvn/version "42.2.18"}
   com.zaxxer/HikariCP {:mvn/version "3.4.5"}
 
-  funcool/log4j2-clojure {:mvn/version "2020.11.23-1"}
   funcool/datoteka {:mvn/version "1.2.0"}
   funcool/promesa {:mvn/version "6.0.0"}
   funcool/cuerdas {:mvn/version "2020.03.26-3"}
diff --git a/backend/dev/user.clj b/backend/dev/user.clj
index 6e66cd6c1..85b53afb5 100644
--- a/backend/dev/user.clj
+++ b/backend/dev/user.clj
@@ -9,24 +9,24 @@
 
 (ns user
   (:require
+   [app.common.exceptions :as ex]
    [app.config :as cfg]
    [app.main :as main]
    [app.util.time :as dt]
    [app.util.transit :as t]
-   [app.common.exceptions :as ex]
-   [taoensso.nippy :as nippy]
    [clojure.data.json :as json]
    [clojure.java.io :as io]
-   [clojure.test :as test]
    [clojure.pprint :refer [pprint]]
    [clojure.repl :refer :all]
    [clojure.spec.alpha :as s]
    [clojure.spec.gen.alpha :as sgen]
    [clojure.test :as test]
+   [clojure.test :as test]
    [clojure.tools.namespace.repl :as repl]
    [clojure.walk :refer [macroexpand-all]]
    [criterium.core :refer [quick-bench bench with-progress-reporting]]
-   [integrant.core :as ig]))
+   [integrant.core :as ig]
+   [taoensso.nippy :as nippy]))
 
 (repl/disable-reload! (find-ns 'integrant.core))
 
diff --git a/backend/resources/error-report.tmpl b/backend/resources/error-report.tmpl
index 3a420a60c..be4df1196 100644
--- a/backend/resources/error-report.tmpl
+++ b/backend/resources/error-report.tmpl
@@ -31,7 +31,7 @@
 
       .table-key {
         font-weight: 600;
-        width: 70px;
+        width: 60px;
         padding: 4px;
       }
 
@@ -70,27 +70,43 @@
 
       {% if user-agent %}
       <div class="table-row">
-        <div class="table-key">UAGENT: </div>
+        <div class="table-key">UAGT: </div>
         <div class="table-val">{{user-agent}}</div>
       </div>
       {% endif %}
 
       {% if frontend-version %}
       <div class="table-row">
-        <div class="table-key">FVERS: </div>
+        <div class="table-key">FVER: </div>
         <div class="table-val">{{frontend-version}}</div>
       </div>
       {% endif %}
 
       <div class="table-row">
-        <div class="table-key">BVERS: </div>
+        <div class="table-key">BVER: </div>
         <div class="table-val">{{version}}</div>
       </div>
 
+      {% if host %}
       <div class="table-row">
         <div class="table-key">HOST: </div>
         <div class="table-val">{{host}}</div>
       </div>
+      {% endif %}
+
+      {% if tenant %}
+      <div class="table-row">
+        <div class="table-key">ENV: </div>
+        <div class="table-val">{{tenant}}</div>
+      </div>
+      {% endif %}
+
+      {% if public-uri %}
+      <div class="table-row">
+        <div class="table-key">PURI: </div>
+        <div class="table-val">{{public-uri}}</div>
+      </div>
+      {% endif %}
 
       {% if type %}
       <div class="table-row">
@@ -106,15 +122,19 @@
       </div>
       {% endif %}
 
+      {% if error %}
       <div class="table-row">
-        <div class="table-key">CLASS: </div>
-        <div class="table-val">{{class}}</div>
+        <div class="table-key">CLSS: </div>
+        <div class="table-val">{{error.class}}</div>
       </div>
+      {% endif %}
 
+      {% if error %}
       <div class="table-row">
         <div class="table-key">HINT: </div>
-        <div class="table-val">{{hint}}</div>
+        <div class="table-val">{{error.message}}</div>
       </div>
+      {% endif %}
 
       {% if method %}
       <div class="table-row">
@@ -150,12 +170,14 @@
       </div>
       {% endif %}
 
+      {% if error %}
       <div class="table-row multiline">
         <div class="table-key">TRACE:</div>
         <div class="table-val">
-          <pre>{{message}}</pre>
+          <pre>{{error.trace}}</pre>
         </div>
       </div>
+      {% endif %}
     </div>
   </body>
 </html>
diff --git a/backend/resources/log4j2.xml b/backend/resources/log4j2.xml
index 0dff9f13d..8830df725 100644
--- a/backend/resources/log4j2.xml
+++ b/backend/resources/log4j2.xml
@@ -13,9 +13,10 @@
       <DefaultRolloverStrategy max="9"/>
     </RollingFile>
 
-    <CljFn name="error-reporter" ns="app.error-reporter" fn="queue-fn">
-      <PatternLayout pattern="[%d{YYYY-MM-dd HH:mm:ss.SSS}] [%t] %level{length=1} %logger{36} - %msg%n"/>
-    </CljFn>
+    <JeroMQ name="zmq">
+      <Property name="endpoint">tcp://localhost:45556</Property>
+      <JsonLayout complete="false" compact="true" includeTimeMillis="true" stacktraceAsString="true" properties="true" />
+    </JeroMQ>
   </Appenders>
 
   <Loggers>
@@ -27,13 +28,19 @@
       <AppenderRef ref="console"/>
     </Logger>
 
-    <Logger name="app.error-reporter" level="debug" additivity="false">
+    <Logger name="app.loggers" level="debug" additivity="false">
       <AppenderRef ref="main" level="debug" />
     </Logger>
 
     <Logger name="app" level="debug" additivity="false">
       <AppenderRef ref="main" level="debug" />
-      <AppenderRef ref="error-reporter" level="error" />
+      <AppenderRef ref="zmq" level="debug" />
+      <!-- <AppenderRef ref="error-reporter" level="error" /> -->
+    </Logger>
+
+    <Logger name="user" level="debug" additivity="false">
+      <AppenderRef ref="main" level="debug" />
+      <AppenderRef ref="zmq" level="debug" />
     </Logger>
 
     <Root level="info">
diff --git a/backend/scripts/repl b/backend/scripts/repl
index e3fa8b324..147661f35 100755
--- a/backend/scripts/repl
+++ b/backend/scripts/repl
@@ -2,6 +2,9 @@
 
 export PENPOT_ASSERTS_ENABLED=true
 
+export OPTIONS="-A:jmx-remote:dev -J-Dclojure.tools.logging.factory=clojure.tools.logging.impl/log4j2-factory -J-Xms512m -J-Xmx512m"
+export OPTIONS_EVAL="nil"
+# export OPTIONS_EVAL="(set! *warn-on-reflection* true)"
+
 set -ex
-# clojure -Ojmx-remote -A:dev -e "(set! *warn-on-reflection* true)" -m rebel-readline.main
-clojure -A:jmx-remote:dev -J-Xms512m -J-Xmx512m -M -m rebel-readline.main
+exec clojure $OPTIONS -M -e "$OPTIONS_EVAL" -m rebel-readline.main
diff --git a/backend/src/app/config.clj b/backend/src/app/config.clj
index 51741e329..00a296c13 100644
--- a/backend/src/app/config.clj
+++ b/backend/src/app/config.clj
@@ -21,7 +21,8 @@
 
 (def defaults
   {:http-server-port 6060
-
+   :host "devenv"
+   :tenant "dev"
    :database-uri "postgresql://127.0.0.1/penpot"
    :database-username "penpot"
    :database-password "penpot"
@@ -87,11 +88,17 @@
    })
 
 (s/def ::http-server-port ::us/integer)
+
+(s/def ::host ::us/string)
+(s/def ::tenant ::us/string)
+
 (s/def ::database-username (s/nilable ::us/string))
 (s/def ::database-password (s/nilable ::us/string))
 (s/def ::database-uri ::us/string)
 (s/def ::redis-uri ::us/string)
 
+(s/def ::loggers-loki-uri ::us/string)
+(s/def ::loggers-zmq-uri ::us/string)
 
 (s/def ::storage-backend ::us/keyword)
 (s/def ::storage-fs-directory ::us/string)
@@ -185,6 +192,7 @@
                    ::google-client-id
                    ::google-client-secret
                    ::http-server-port
+                   ::host
                    ::ldap-auth-avatar-attribute
                    ::ldap-auth-base-dn
                    ::ldap-auth-email-attribute
@@ -221,6 +229,8 @@
                    ::srepl-host
                    ::srepl-port
                    ::local-assets-uri
+                   ::loggers-loki-uri
+                   ::loggers-zmq-uri
                    ::storage-s3-bucket
                    ::storage-s3-region
                    ::telemetry-enabled
@@ -228,6 +238,7 @@
                    ::telemetry-server-enabled
                    ::telemetry-server-port
                    ::telemetry-uri
+                   ::tenant
                    ::initial-data-file
                    ::initial-data-project-name]))
 
diff --git a/backend/src/app/http/errors.clj b/backend/src/app/http/errors.clj
index af688d473..3758942f5 100644
--- a/backend/src/app/http/errors.clj
+++ b/backend/src/app/http/errors.clj
@@ -11,7 +11,6 @@
   "A errors handling for the http server."
   (:require
    [app.common.uuid :as uuid]
-   [app.config :as cfg]
    [app.util.log4j :refer [update-thread-context!]]
    [clojure.tools.logging :as log]
    [cuerdas.core :as str]
@@ -30,16 +29,10 @@
       :path    (:uri request)
       :method  (:request-method request)
       :params  (:params request)
-      :version (:full cfg/version)
-      :host    (:public-uri cfg/config)
-      :class   (.getCanonicalName ^java.lang.Class (class error))
-      :hint    (ex-message error)
       :data    edata}
-
      (let [headers (:headers request)]
        {:user-agent (get headers "user-agent")
         :frontend-version (get headers "x-frontend-version" "unknown")})
-
      (when (and (map? edata) (:data edata))
        {:explain (explain-error edata)}))))
 
diff --git a/backend/src/app/loggers/loki.clj b/backend/src/app/loggers/loki.clj
new file mode 100644
index 000000000..32813e5ec
--- /dev/null
+++ b/backend/src/app/loggers/loki.clj
@@ -0,0 +1,92 @@
+;; 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/.
+;;
+;; This Source Code Form is "Incompatible With Secondary Licenses", as
+;; defined by the Mozilla Public License, v. 2.0.
+;;
+;; Copyright (c) 2020-2021 UXBOX Labs SL
+
+(ns app.loggers.loki
+  "A Loki integration."
+  (:require
+   [app.common.spec :as us]
+   [app.config :as cfg]
+   [app.util.async :as aa]
+   [app.util.http :as http]
+   [app.util.json :as json]
+   [app.worker :as wrk]
+   [clojure.core.async :as a]
+   [clojure.spec.alpha :as s]
+   [clojure.tools.logging :as log]
+   [integrant.core :as ig]))
+
+(declare handle-event)
+
+(s/def ::uri ::us/string)
+(s/def ::receiver fn?)
+
+(defmethod ig/pre-init-spec ::reporter [_]
+  (s/keys :req-un [::wrk/executor ::receiver]
+          :opt-un [::uri]))
+
+(defmethod ig/init-key ::reporter
+  [_ {:keys [receiver uri] :as cfg}]
+  (when uri
+    (log/info "Intializing loki reporter.")
+    (let [output (a/chan (a/sliding-buffer 1024))]
+      (receiver :sub output)
+      (a/go-loop []
+        (let [msg (a/<! output)]
+          (if (nil? msg)
+            (log/info "Stoping error reporting loop.")
+            (do
+              (a/<! (handle-event cfg msg))
+              (recur)))))
+      output)))
+
+(defmethod ig/halt-key! ::reporter
+  [_ output]
+  (when output
+    (a/close! output)))
+
+(defn- prepare-payload
+  [event]
+  (let [labels {:host    (cfg/get :host)
+                :tenant  (cfg/get :tenant)
+                :version (:full cfg/version)
+                :logger  (:logger event)
+                :level   (:level event)}]
+    {:streams
+     [{:stream labels
+       :values [[(str (* (inst-ms (:created-at event)) 1000000))
+                 (str (:message event)
+                      (when-let [error (:error event)]
+                        (str "\n" (:trace error))))]]}]}))
+
+(defn- send-log
+  [uri payload i]
+  (try
+    (let [response (http/send! {:uri uri
+                                :timeout 6000
+                                :method :post
+                                :headers {"content-type" "application/json"}
+                                :body (json/encode payload)})]
+      (if (= (:status response) 204)
+        true
+        (do
+          (log/errorf "Error on sending log to loki (try %s).\n%s" i (pr-str response))
+          false)))
+    (catch Exception e
+      (log/errorf e "Error on sending message to loki (try %s)." i)
+      false)))
+
+(defn- handle-event
+  [{:keys [executor uri]} event]
+  (aa/with-thread executor
+    (let [payload (prepare-payload event)]
+      (loop [i 1]
+        (when (and (not (send-log uri payload i)) (< i 20))
+          (Thread/sleep (* i 2000))
+          (recur (inc i)))))))
+
diff --git a/backend/src/app/error_reporter.clj b/backend/src/app/loggers/mattermost.clj
similarity index 57%
rename from backend/src/app/error_reporter.clj
rename to backend/src/app/loggers/mattermost.clj
index a1f77f3ab..7f40a48bd 100644
--- a/backend/src/app/error_reporter.clj
+++ b/backend/src/app/loggers/mattermost.clj
@@ -5,9 +5,9 @@
 ;; This Source Code Form is "Incompatible With Secondary Licenses", as
 ;; defined by the Mozilla Public License, v. 2.0.
 ;;
-;; Copyright (c) 2020 Andrey Antukh <niwi@niwi.nz>
+;; Copyright (c) 2020-2021 UXBOX Labs SL
 
-(ns app.error-reporter
+(ns app.loggers.mattermost
   "A mattermost integration for error reporting."
   (:require
    [app.common.exceptions :as ex]
@@ -15,6 +15,7 @@
    [app.common.uuid :as uuid]
    [app.config :as cfg]
    [app.db :as db]
+   [app.util.async :as aa]
    [app.util.http :as http]
    [app.util.json :as json]
    [app.util.template :as tmpl]
@@ -24,11 +25,7 @@
    [clojure.spec.alpha :as s]
    [clojure.tools.logging :as log]
    [cuerdas.core :as str]
-   [integrant.core :as ig]
-   [promesa.exec :as px])
-  (:import
-   org.apache.logging.log4j.core.LogEvent
-   org.apache.logging.log4j.util.ReadOnlyStringMap))
+   [integrant.core :as ig]))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Error Listener
@@ -37,76 +34,51 @@
 (declare handle-event)
 
 (defonce enabled-mattermost (atom true))
-(defonce queue (a/chan (a/sliding-buffer 64)))
-(defonce queue-fn (fn [event] (a/>!! queue event)))
 
 (s/def ::uri ::us/string)
 
 (defmethod ig/pre-init-spec ::reporter [_]
-  (s/keys :req-un [::wrk/executor ::db/pool]
+  (s/keys :req-un [::wrk/executor ::db/pool ::receiver]
           :opt-un [::uri]))
 
 (defmethod ig/init-key ::reporter
-  [_ {:keys [executor] :as cfg}]
-  (log/info "Intializing error reporter.")
-  (let [close-ch (a/chan 1)]
+  [_ {:keys [receiver] :as cfg}]
+  (log/info "Intializing mattermost error reporter.")
+  (let [output (a/chan (a/sliding-buffer 128)
+                       (filter #(= (:level %) "error")))]
+    (receiver :sub output)
     (a/go-loop []
-      (let [[val port] (a/alts! [close-ch queue])]
-        (cond
-          (= port close-ch)
+      (let [msg (a/<! output)]
+        (if (nil? msg)
           (log/info "Stoping error reporting loop.")
-
-          (nil? val)
-          (log/info "Stoping error reporting loop.")
-
-          :else
           (do
-            (px/run! executor #(handle-event cfg val))
+            (a/<! (handle-event cfg msg))
             (recur)))))
-    close-ch))
+    output))
 
 (defmethod ig/halt-key! ::reporter
-  [_ close-ch]
-  (a/close! close-ch))
-
-(defn- get-context-data
-  [event]
-  (let [^LogEvent levent (deref event)
-        ^ReadOnlyStringMap rosm (.getContextData levent)]
-    (into {:message (str event)
-           :id      (uuid/next)} ; set default uuid for cases when it not comes.
-          (comp
-           (map (fn [[key val]]
-                  (cond
-                    (= "id" key)         [:id (uuid/uuid val)]
-                    (= "profile-id" key) [:profile-id (uuid/uuid val)]
-                    (str/blank? val)     nil
-                    (string? key)        [(keyword key) val]
-                    :else                [key val])))
-           (filter some?))
-
-          (.toMap rosm))))
+  [_ output]
+  (a/close! output))
 
 (defn- send-mattermost-notification!
-  [cfg {:keys [message host version id] :as cdata}]
+  [cfg {:keys [host version id error] :as cdata}]
   (try
-    (let [uri    (:uri cfg)
-          prefix (str "Unhandled exception (@channel):\n"
-                      "- detail: " (:public-uri cfg/config) "/dbg/error-by-id/" id "\n"
-                      "- host: `" host "`\n"
-                      "- version: `" version "`\n")
-          text   (str prefix "```\n" message "\n```")
+    (let [uri  (:uri cfg)
+          text (str "Unhandled exception (@channel):\n"
+                    "- detail: " (:public-uri cfg/config) "/dbg/error-by-id/" id "\n"
+                    "- host: `" host "`\n"
+                    "- version: `" version "`\n"
+                    (when error
+                      (str "```\n" (:trace error)  "\n```")))
           rsp    (http/send! {:uri uri
                               :method :post
                               :headers {"content-type" "application/json"}
                               :body (json/encode-str {:text text})})]
       (when (not= (:status rsp) 200)
-        (log/warnf "Error reporting webhook replying with unexpected status: %s\n%s"
-                   (:status rsp)
-                   (pr-str rsp))))
+        (log/errorf "Error on sending data to mattermost\n%s" (pr-str rsp))))
 
     (catch Exception e
-      (log/warnf e "Unexpected exception on error reporter."))))
+      (log/error e "Unexpected exception on error reporter."))))
 
 (defn- persist-on-database!
   [{:keys [pool] :as cfg} {:keys [id] :as cdata}]
@@ -114,15 +86,37 @@
     (db/insert! conn :server-error-report
                 {:id id :content (db/tjson cdata)})))
 
+(defn- parse-context
+  [event]
+  (reduce-kv
+   (fn [acc k v]
+     (cond
+       (= k :id)         (assoc acc k (uuid/uuid v))
+       (= k :profile-id) (assoc acc k (uuid/uuid v))
+       (str/blank? v)    acc
+       :else             (assoc acc k v)))
+   {}
+   (:context event)))
+
+(defn- parse-event
+  [event]
+  (-> (parse-context event)
+      (merge (dissoc event :context))
+      (assoc :tenant (cfg/get :tenant))
+      (assoc :host (cfg/get :host))
+      (assoc :public-uri (cfg/get :public-uri))
+      (assoc :version (:full cfg/version))))
+
 (defn handle-event
-  [cfg event]
-  (try
-    (let [cdata (get-context-data event)]
-      (when (and (:uri cfg) @enabled-mattermost)
-        (send-mattermost-notification! cfg cdata))
-      (persist-on-database! cfg cdata))
-    (catch Exception e
-      (log/warnf e "Unexpected exception on error reporter."))))
+  [{:keys [executor] :as cfg} event]
+  (aa/with-thread executor
+    (try
+      (let [cdata (parse-event event)]
+        (when (and (:uri cfg) @enabled-mattermost)
+          (send-mattermost-notification! cfg cdata))
+        (persist-on-database! cfg cdata))
+      (catch Exception e
+        (log/error e "Unexpected exception on error reporter.")))))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Http Handler
diff --git a/backend/src/app/loggers/zmq.clj b/backend/src/app/loggers/zmq.clj
new file mode 100644
index 000000000..834212cc8
--- /dev/null
+++ b/backend/src/app/loggers/zmq.clj
@@ -0,0 +1,92 @@
+;; 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/.
+;;
+;; This Source Code Form is "Incompatible With Secondary Licenses", as
+;; defined by the Mozilla Public License, v. 2.0.
+;;
+;; Copyright (c) 2020-2021 UXBOX Labs SL
+
+(ns app.loggers.zmq
+  "A generic ZMQ listener."
+  (:require
+   [app.common.data :as d]
+   [app.common.spec :as us]
+   [app.util.json :as json]
+   [app.util.time :as dt]
+   [clojure.core.async :as a]
+   [clojure.spec.alpha :as s]
+   [clojure.tools.logging :as log]
+   [cuerdas.core :as str]
+   [integrant.core :as ig])
+  (:import
+   org.zeromq.SocketType
+   org.zeromq.ZMQ$Socket
+   org.zeromq.ZContext))
+
+(declare prepare)
+(declare start-rcv-loop)
+
+(s/def ::endpoint ::us/string)
+
+(defmethod ig/pre-init-spec ::receiver [_]
+  (s/keys :opt-un [::endpoint]))
+
+(defmethod ig/init-key ::receiver
+  [_ {:keys [endpoint] :as cfg}]
+  (log/infof "Intializing ZMQ receiver on '%s'." endpoint)
+  (let [buffer (a/chan 1)
+        output (a/chan 1 (comp (filter map?)
+                               (map prepare)))
+        mult   (a/mult output)]
+    (when endpoint
+      (a/thread (start-rcv-loop {:out buffer :endpoint endpoint})))
+    (a/pipe buffer output)
+    (with-meta
+      (fn [cmd ch]
+        (case cmd
+          :sub (a/tap mult ch)
+          :unsub (a/untap mult ch))
+        ch)
+      {::output output
+       ::buffer buffer
+       ::mult mult})))
+
+(defmethod ig/halt-key! ::receiver
+  [_ f]
+  (a/close! (::buffer (meta f))))
+
+(defn- start-rcv-loop
+  ([] (start-rcv-loop nil))
+  ([{:keys [out endpoint] :or {endpoint "tcp://localhost:5556"}}]
+   (let [out    (or out (a/chan 1))
+         zctx   (ZContext.)
+         socket (.. zctx (createSocket SocketType/SUB))]
+     (.. socket (connect ^String endpoint))
+     (.. socket (subscribe ""))
+     (.. socket (setReceiveTimeOut 5000))
+     (loop []
+       (let [msg (.recv ^ZMQ$Socket socket)
+             msg (json/decode msg)
+             msg (if (nil? msg) :empty msg)]
+         (if (a/>!! out msg)
+           (recur)
+           (do
+             (.close ^java.lang.AutoCloseable socket)
+             (.close ^java.lang.AutoCloseable zctx))))))))
+
+(defn- prepare
+  [event]
+  (d/merge
+   {:logger     (:loggerName event)
+    :level      (str/lower (:level event))
+    :thread     (:thread event)
+    :created-at (dt/instant (:timeMillis event))
+    :message    (:message event)}
+   (when-let [ctx (:contextMap event)]
+     {:context ctx})
+   (when-let [thrown (:thrown event)]
+     {:error
+      {:class (:name thrown)
+       :message (:message thrown)
+       :trace (:extendedStackTrace thrown)}})))
diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj
index 6f2203445..ba8594798 100644
--- a/backend/src/app/main.clj
+++ b/backend/src/app/main.clj
@@ -95,7 +95,7 @@
      :svgparse    (ig/ref :app.svgparse/handler)
      :storage     (ig/ref :app.storage/storage)
      :sns-webhook (ig/ref :app.http.awsns/handler)
-     :error-report-handler (ig/ref :app.error-reporter/handler)}
+     :error-report-handler (ig/ref :app.loggers.mattermost/handler)}
 
     :app.http.assets/handlers
     {:metrics           (ig/ref :app.metrics/metrics)
@@ -280,12 +280,21 @@
     :app.sprops/props
     {:pool (ig/ref :app.db/pool)}
 
-    :app.error-reporter/reporter
+    :app.loggers.zmq/receiver
+    {:endpoint (:loggers-zmq-uri config)}
+
+    :app.loggers.loki/reporter
+    {:uri      (:loggers-loki-uri config)
+     :receiver (ig/ref :app.loggers.zmq/receiver)
+     :executor (ig/ref :app.worker/executor)}
+
+    :app.loggers.mattermost/reporter
     {:uri      (:error-report-webhook config)
+     :receiver (ig/ref :app.loggers.zmq/receiver)
      :pool     (ig/ref :app.db/pool)
      :executor (ig/ref :app.worker/executor)}
 
-    :app.error-reporter/handler
+    :app.loggers.mattermost/handler
     {:pool (ig/ref :app.db/pool)}
 
     :app.storage/storage
diff --git a/backend/src/app/util/json.clj b/backend/src/app/util/json.clj
index 7b3013a97..042517c62 100644
--- a/backend/src/app/util/json.clj
+++ b/backend/src/app/util/json.clj
@@ -16,10 +16,18 @@
   [v]
   (j/write-value-as-string v j/keyword-keys-object-mapper))
 
+(defn encode
+  [v]
+  (j/write-value-as-bytes v j/keyword-keys-object-mapper))
+
 (defn decode-str
   [v]
   (j/read-value v j/keyword-keys-object-mapper))
 
+(defn decode
+  [v]
+  (j/read-value v j/keyword-keys-object-mapper))
+
 (defn read
   [v]
   (j/read-value v j/keyword-keys-object-mapper))
diff --git a/backend/src/app/util/time.clj b/backend/src/app/util/time.clj
index be7012872..907098a77 100644
--- a/backend/src/app/util/time.clj
+++ b/backend/src/app/util/time.clj
@@ -93,6 +93,10 @@
   [t1 t2]
   (Duration/between t1 t2))
 
+(defn instant
+  [ms]
+  (Instant/ofEpochMilli ms))
+
 (defn parse-duration
   [s]
   (Duration/parse s))
diff --git a/backend/src/app/worker.clj b/backend/src/app/worker.clj
index 145e1ce58..306604b81 100644
--- a/backend/src/app/worker.clj
+++ b/backend/src/app/worker.clj
@@ -12,7 +12,6 @@
   (:require
    [app.common.spec :as us]
    [app.common.uuid :as uuid]
-   [app.config :as cfg]
    [app.db :as db]
    [app.util.async :as aa]
    [app.util.log4j :refer [update-thread-context!]]
@@ -210,10 +209,6 @@
   [error item]
   (let [edata (ex-data error)]
     {:id      (uuid/next)
-     :version (:full cfg/version)
-     :host    (:public-uri cfg/config)
-     :class   (.getCanonicalName ^java.lang.Class (class error))
-     :hint    (ex-message error)
      :data    edata
      :params  item}))