Penpot/penpot
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2025-02-15 11:38:24 -05:00
Code Issues Activity

Merge pull request #683 from penpot/niwinz/session-updater

Browse source 
Enhacements
This commit is contained in:
Andrés Moya 2021-02-22 14:53:10 +01:00 • committed by GitHub
commit 4a61eba3b9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
30 changed files with 658 additions and 418 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
CHANGES.md
View file

@ -7,6 +7,7 @@
- Add major refactor of internal pubsub/redis code; improves scalability and performance [#640](https://github.com/penpot/penpot/pull/640) - Add major refactor of internal pubsub/redis code; improves scalability and performance [#640](https://github.com/penpot/penpot/pull/640)
- Add optional loki integration [#645](https://github.com/penpot/penpot/pull/645) - Add optional loki integration [#645](https://github.com/penpot/penpot/pull/645)
- Add emailcatcher and ldap test containers to devenv. [#506](https://github.com/penpot/penpot/pull/506) - Add emailcatcher and ldap test containers to devenv. [#506](https://github.com/penpot/penpot/pull/506)
- Add proper http session lifecycle handling.
- Add more presets for artboard [#654](https://github.com/penpot/penpot/pull/654) - Add more presets for artboard [#654](https://github.com/penpot/penpot/pull/654)
- Bounce & Complaint handling [#635](https://github.com/penpot/penpot/pull/635) - Bounce & Complaint handling [#635](https://github.com/penpot/penpot/pull/635)
- Disable groups interactions when holding "Ctrl" key (deep selection) - Disable groups interactions when holding "Ctrl" key (deep selection)
@ -21,6 +22,7 @@
- Fix corner cases on invitation/signup flows. - Fix corner cases on invitation/signup flows.
- Fix problem width handoff code generation [Taiga #1204](https://tree.taiga.io/project/penpot/issue/1204) - Fix problem width handoff code generation [Taiga #1204](https://tree.taiga.io/project/penpot/issue/1204)
- Fix problem with indices refreshing on page changes [#646](https://github.com/penpot/penpot/issues/646) - Fix problem with indices refreshing on page changes [#646](https://github.com/penpot/penpot/issues/646)
- Fix infinite recursion on logout.
- Have language change notification written in the new language [Taiga #1205](https://tree.taiga.io/project/penpot/issue/1205) - Have language change notification written in the new language [Taiga #1205](https://tree.taiga.io/project/penpot/issue/1205)
- Properly handle errors on github, gitlab and ldap auth backends. - Properly handle errors on github, gitlab and ldap auth backends.
- Properly mark profile auth backend (on first register/ auth with 3rd party auth provider). - Properly mark profile auth backend (on first register/ auth with 3rd party auth provider).

2
backend/src/app/cli/fixtures.clj
View file

@ -237,6 +237,6 @@
(try (try
(run-in-system system preset) (run-in-system system preset)
(catch Exception e (catch Exception e
(log/errorf e "Unhandled exception.")) (log/errorf e "unhandled exception"))
(finally (finally
(ig/halt! system))))) (ig/halt! system)))))

141
backend/src/app/config.clj
View file

@ -5,7 +5,7 @@
;; This Source Code Form is "Incompatible With Secondary Licenses", as ;; This Source Code Form is "Incompatible With Secondary Licenses", as
;; defined by the Mozilla Public License, v. 2.0. ;; defined by the Mozilla Public License, v. 2.0.
;; ;;
;; Copyright (c) 2020 UXBOX Labs SL ;; Copyright (c) 2020-2021 UXBOX Labs SL
(ns app.config (ns app.config
"A configuration management." "A configuration management."
@ -80,92 +80,78 @@
;; :initial-data-project-name "Penpot Oboarding" ;; :initial-data-project-name "Penpot Oboarding"
}) })
(s/def ::http-server-port ::us/integer) (s/def ::allow-demo-users ::us/boolean)
(s/def ::asserts-enabled ::us/boolean)
(s/def ::host ::us/string) (s/def ::assets-path ::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-password (s/nilable ::us/string))
(s/def ::database-uri ::us/string) (s/def ::database-uri ::us/string)
(s/def ::redis-uri ::us/string) (s/def ::database-username (s/nilable ::us/string))
(s/def ::default-blob-version ::us/integer)
(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)
(s/def ::assets-path ::us/string)
(s/def ::storage-s3-region ::us/keyword)
(s/def ::storage-s3-bucket ::us/string)
(s/def ::media-uri ::us/string)
(s/def ::media-directory ::us/string)
(s/def ::asserts-enabled ::us/boolean)
(s/def ::feedback-enabled ::us/boolean)
(s/def ::feedback-destination ::us/string)
(s/def ::profile-complaint-max-age ::dt/duration)
(s/def ::profile-complaint-threshold ::us/integer)
(s/def ::profile-bounce-max-age ::dt/duration)
(s/def ::profile-bounce-threshold ::us/integer)
(s/def ::error-report-webhook ::us/string) (s/def ::error-report-webhook ::us/string)
(s/def ::feedback-destination ::us/string)
(s/def ::smtp-enabled ::us/boolean) (s/def ::feedback-enabled ::us/boolean)
(s/def ::smtp-default-reply-to ::us/string)
(s/def ::smtp-default-from ::us/string)
(s/def ::smtp-host ::us/string)
(s/def ::smtp-port ::us/integer)
(s/def ::smtp-username (s/nilable ::us/string))
(s/def ::smtp-password (s/nilable ::us/string))
(s/def ::smtp-tls ::us/boolean)
(s/def ::smtp-ssl ::us/boolean)
(s/def ::allow-demo-users ::us/boolean)
(s/def ::registration-enabled ::us/boolean)
(s/def ::registration-domain-whitelist ::us/string)
(s/def ::public-uri ::us/string)
(s/def ::srepl-host ::us/string)
(s/def ::srepl-port ::us/integer)
(s/def ::rlimits-password ::us/integer)
(s/def ::rlimits-image ::us/integer)
(s/def ::google-client-id ::us/string)
(s/def ::google-client-secret ::us/string)
(s/def ::gitlab-client-id ::us/string)
(s/def ::gitlab-client-secret ::us/string)
(s/def ::gitlab-base-uri ::us/string)
(s/def ::github-client-id ::us/string) (s/def ::github-client-id ::us/string)
(s/def ::github-client-secret ::us/string) (s/def ::github-client-secret ::us/string)
(s/def ::gitlab-base-uri ::us/string)
(s/def ::ldap-host ::us/string) (s/def ::gitlab-client-id ::us/string)
(s/def ::ldap-port ::us/integer) (s/def ::gitlab-client-secret ::us/string)
(s/def ::ldap-bind-dn ::us/string) (s/def ::google-client-id ::us/string)
(s/def ::ldap-bind-password ::us/string) (s/def ::google-client-secret ::us/string)
(s/def ::ldap-ssl ::us/boolean) (s/def ::host ::us/string)
(s/def ::ldap-starttls ::us/boolean) (s/def ::http-server-port ::us/integer)
(s/def ::ldap-base-dn ::us/string) (s/def ::http-session-cookie-name ::us/string)
(s/def ::ldap-user-query ::us/string) (s/def ::http-session-idle-max-age ::dt/duration)
(s/def ::ldap-attrs-username ::us/string) (s/def ::http-session-updater-batch-max-age ::dt/duration)
(s/def ::http-session-updater-batch-max-size ::us/integer)
(s/def ::initial-data-file ::us/string)
(s/def ::initial-data-project-name ::us/string)
(s/def ::ldap-attrs-email ::us/string) (s/def ::ldap-attrs-email ::us/string)
(s/def ::ldap-attrs-fullname ::us/string) (s/def ::ldap-attrs-fullname ::us/string)
(s/def ::ldap-attrs-photo ::us/string) (s/def ::ldap-attrs-photo ::us/string)
(s/def ::ldap-attrs-username ::us/string)
(s/def ::ldap-base-dn ::us/string)
(s/def ::ldap-bind-dn ::us/string)
(s/def ::ldap-bind-password ::us/string)
(s/def ::ldap-host ::us/string)
(s/def ::ldap-port ::us/integer)
(s/def ::ldap-ssl ::us/boolean)
(s/def ::ldap-starttls ::us/boolean)
(s/def ::ldap-user-query ::us/string)
(s/def ::loggers-loki-uri ::us/string)
(s/def ::loggers-zmq-uri ::us/string)
(s/def ::media-directory ::us/string)
(s/def ::media-uri ::us/string)
(s/def ::profile-bounce-max-age ::dt/duration)
(s/def ::profile-bounce-threshold ::us/integer)
(s/def ::profile-complaint-max-age ::dt/duration)
(s/def ::profile-complaint-threshold ::us/integer)
(s/def ::public-uri ::us/string)
(s/def ::redis-uri ::us/string)
(s/def ::registration-domain-whitelist ::us/string)
(s/def ::registration-enabled ::us/boolean)
(s/def ::rlimits-image ::us/integer)
(s/def ::rlimits-password ::us/integer)
(s/def ::smtp-default-from ::us/string)
(s/def ::smtp-default-reply-to ::us/string)
(s/def ::smtp-enabled ::us/boolean)
(s/def ::smtp-host ::us/string)
(s/def ::smtp-password (s/nilable ::us/string))
(s/def ::smtp-port ::us/integer)
(s/def ::smtp-ssl ::us/boolean)
(s/def ::smtp-tls ::us/boolean)
(s/def ::smtp-username (s/nilable ::us/string))
(s/def ::srepl-host ::us/string)
(s/def ::srepl-port ::us/integer)
(s/def ::storage-backend ::us/keyword)
(s/def ::storage-fs-directory ::us/string)
(s/def ::storage-s3-bucket ::us/string)
(s/def ::storage-s3-region ::us/keyword)
(s/def ::telemetry-enabled ::us/boolean) (s/def ::telemetry-enabled ::us/boolean)
(s/def ::telemetry-with-taiga ::us/boolean)
(s/def ::telemetry-uri ::us/string)
(s/def ::telemetry-server-enabled ::us/boolean) (s/def ::telemetry-server-enabled ::us/boolean)
(s/def ::telemetry-server-port ::us/integer) (s/def ::telemetry-server-port ::us/integer)
(s/def ::telemetry-uri ::us/string)
(s/def ::initial-data-file ::us/string) (s/def ::telemetry-with-taiga ::us/boolean)
(s/def ::initial-data-project-name ::us/string) (s/def ::tenant ::us/string)
(s/def ::default-blob-version ::us/integer)
(s/def ::config (s/def ::config
(s/keys :opt-un [::allow-demo-users (s/keys :opt-un [::allow-demo-users
@ -185,6 +171,9 @@
::google-client-id ::google-client-id
::google-client-secret ::google-client-secret
::http-server-port ::http-server-port
::http-session-updater-batch-max-age
::http-session-updater-batch-max-size
::http-session-idle-max-age
::host ::host
::ldap-attrs-username ::ldap-attrs-username
::ldap-attrs-email ::ldap-attrs-email

23
backend/src/app/db.clj
View file

@ -13,6 +13,7 @@
[app.common.geom.point :as gpt] [app.common.geom.point :as gpt]
[app.common.spec :as us] [app.common.spec :as us]
[app.db.sql :as sql] [app.db.sql :as sql]
[app.metrics :as mtx]
[app.util.json :as json] [app.util.json :as json]
[app.util.migrations :as mg] [app.util.migrations :as mg]
[app.util.time :as dt] [app.util.time :as dt]
@ -45,19 +46,21 @@
;; Initialization ;; Initialization
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(declare instrument-jdbc!)
(s/def ::uri ::us/not-empty-string) (s/def ::uri ::us/not-empty-string)
(s/def ::name ::us/not-empty-string) (s/def ::name ::us/not-empty-string)
(s/def ::min-pool-size ::us/integer) (s/def ::min-pool-size ::us/integer)
(s/def ::max-pool-size ::us/integer) (s/def ::max-pool-size ::us/integer)
(s/def ::migrations map?) (s/def ::migrations map?)
(s/def ::metrics map?)
(defmethod ig/pre-init-spec ::pool [_] (defmethod ig/pre-init-spec ::pool [_]
(s/keys :req-un [::uri ::name ::min-pool-size ::max-pool-size ::migrations])) (s/keys :req-un [::uri ::name ::min-pool-size ::max-pool-size ::migrations ::mtx/metrics]))
(defmethod ig/init-key ::pool (defmethod ig/init-key ::pool
[_ {:keys [migrations] :as cfg}] [_ {:keys [migrations metrics] :as cfg}]
(log/debugf "initialize connection pool %s with uri %s" (:name cfg) (:uri cfg)) (log/infof "initialize connection pool '%s' with uri '%s'" (:name cfg) (:uri cfg))
(instrument-jdbc! (:registry metrics))
(let [pool (create-pool cfg)] (let [pool (create-pool cfg)]
(when (seq migrations) (when (seq migrations)
(with-open [conn ^AutoCloseable (open pool)] (with-open [conn ^AutoCloseable (open pool)]
@ -70,12 +73,22 @@
[_ pool] [_ pool]
(.close ^HikariDataSource pool)) (.close ^HikariDataSource pool))
(defn- instrument-jdbc!
[registry]
(mtx/instrument-vars!
[#'next.jdbc/execute-one!
#'next.jdbc/execute!]
{:registry registry
:type :counter
:name "database_query_count"
:help "An absolute counter of database queries."}))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; API & Impl ;; API & Impl
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(def initsql (def initsql
(str "SET statement_timeout = 60000;\n" (str "SET statement_timeout = 120000;\n"
"SET idle_in_transaction_session_timeout = 120000;")) "SET idle_in_transaction_session_timeout = 120000;"))
(defn- create-datasource-config (defn- create-datasource-config

10
backend/src/app/http/awsns.clj
View file

@ -42,7 +42,7 @@
(= mtype "SubscriptionConfirmation") (= mtype "SubscriptionConfirmation")
(let [surl (get body "SubscribeURL") (let [surl (get body "SubscribeURL")
stopic (get body "TopicArn")] stopic (get body "TopicArn")]
(log/infof "Subscription received (topic=%s, url=%s)" stopic surl) (log/infof "subscription received (topic=%s, url=%s)" stopic surl)
(http/send! {:uri surl :method :post :timeout 10000})) (http/send! {:uri surl :method :post :timeout 10000}))
(= mtype "Notification") (= mtype "Notification")
@ -52,7 +52,7 @@
(process-report cfg notification))) (process-report cfg notification)))
:else :else
(log/warn (str "Unexpected data received.\n" (log/warn (str "unexpected data received\n"
(pprint-report body)))) (pprint-report body))))
{:status 200 :body ""}))) {:status 200 :body ""})))
@ -184,14 +184,14 @@
(defn- process-report (defn- process-report
[cfg {:keys [type profile-id] :as report}] [cfg {:keys [type profile-id] :as report}]
(log/debug (str "Procesing report:\n" (pprint-report report))) (log/debug (str "procesing report:\n" (pprint-report report)))
(cond (cond
;; In this case we receive a bounce/complaint notification without ;; In this case we receive a bounce/complaint notification without
;; confirmed identity, we just emit a warning but do nothing about ;; confirmed identity, we just emit a warning but do nothing about
;; it because this is not a normal case. All notifications should ;; it because this is not a normal case. All notifications should
;; come with profile identity. ;; come with profile identity.
(nil? profile-id) (nil? profile-id)
(log/warn (str "A notification without identity recevied from AWS\n" (log/warn (str "a notification without identity recevied from AWS\n"
(pprint-report report))) (pprint-report report)))
(= "bounce" type) (= "bounce" type)
@ -201,7 +201,7 @@
(register-complaint-for-profile cfg report) (register-complaint-for-profile cfg report)
:else :else
(log/warn (str "Unrecognized report received from AWS\n" (log/warn (str "unrecognized report received from AWS\n"
(pprint-report report))))) (pprint-report report)))))

4
backend/src/app/http/errors.clj
View file

@ -73,7 +73,7 @@
(let [edata (ex-data error) (let [edata (ex-data error)
cdata (get-error-context request error)] cdata (get-error-context request error)]
(update-thread-context! cdata) (update-thread-context! cdata)
(log/errorf error "Internal error: assertion (id: %s)" (str (:id cdata))) (log/errorf error "internal error: assertion (id: %s)" (str (:id cdata)))
{:status 500 {:status 500
:body {:type :server-error :body {:type :server-error
:data (-> edata :data (-> edata
@ -88,7 +88,7 @@
[error request] [error request]
(let [cdata (get-error-context request error)] (let [cdata (get-error-context request error)]
(update-thread-context! cdata) (update-thread-context! cdata)
(log/errorf error "Internal error: %s (id: %s)" (log/errorf error "internal error: %s (id: %s)"
(ex-message error) (ex-message error)
(str (:id cdata))) (str (:id cdata)))
{:status 500 {:status 500

143
backend/src/app/http/session.clj
View file

@ -9,11 +9,20 @@
(ns app.http.session (ns app.http.session
(:require (:require
[app.common.data :as d]
[app.common.exceptions :as ex]
[app.config :as cfg]
[app.db :as db] [app.db :as db]
[app.metrics :as mtx]
[app.util.async :as aa]
[app.util.log4j :refer [update-thread-context!]] [app.util.log4j :refer [update-thread-context!]]
[app.util.time :as dt]
[app.worker :as wrk]
[buddy.core.codecs :as bc] [buddy.core.codecs :as bc]
[buddy.core.nonce :as bn] [buddy.core.nonce :as bn]
[clojure.core.async :as a]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log]
[integrant.core :as ig])) [integrant.core :as ig]))
;; --- IMPL ;; --- IMPL
@ -42,8 +51,7 @@
(defn- retrieve (defn- retrieve
[{:keys [conn] :as cfg} token] [{:keys [conn] :as cfg} token]
(when token (when token
(-> (db/exec-one! conn ["select profile_id from http_session where id = ?" token]) (db/exec-one! conn ["select id, profile_id from http_session where id = ?" token])))
(:profile-id))))
(defn- retrieve-from-request (defn- retrieve-from-request
[{:keys [cookie-name] :as cfg} {:keys [cookies] :as request}] [{:keys [cookie-name] :as cfg} {:keys [cookies] :as request}]
@ -57,24 +65,33 @@
(defn- middleware (defn- middleware
[cfg handler] [cfg handler]
(fn [request] (fn [request]
(if-let [profile-id (retrieve-from-request cfg request)] (if-let [{:keys [id profile-id] :as session} (retrieve-from-request cfg request)]
(do (let [ech (::events-ch cfg)]
(a/>!! ech id)
(update-thread-context! {:profile-id profile-id}) (update-thread-context! {:profile-id profile-id})
(handler (assoc request :profile-id profile-id))) (handler (assoc request :profile-id profile-id)))
(handler request)))) (handler request))))
;; --- STATE INIT ;; --- STATE INIT: SESSION
(s/def ::cookie-name ::cfg/http-session-cookie-name)
(defmethod ig/pre-init-spec ::session [_] (defmethod ig/pre-init-spec ::session [_]
(s/keys :req-un [::db/pool])) (s/keys :req-un [::db/pool]
:opt-un [::cookie-name]))
(defmethod ig/prep-key ::session (defmethod ig/prep-key ::session
[_ cfg] [_ cfg]
(merge {:cookie-name "auth-token"} cfg)) (merge {:cookie-name "auth-token"
:buffer-size 64}
(d/without-nils cfg)))
(defmethod ig/init-key ::session (defmethod ig/init-key ::session
[_ {:keys [pool] :as cfg}] [_ {:keys [pool] :as cfg}]
(let [cfg (assoc cfg :conn pool)] (let [events (a/chan (a/dropping-buffer (:buffer-size cfg)))
cfg (assoc cfg
:conn pool
::events-ch events)]
(-> cfg (-> cfg
(assoc :middleware #(middleware cfg %)) (assoc :middleware #(middleware cfg %))
(assoc :create (fn [profile-id] (assoc :create (fn [profile-id]
@ -89,3 +106,113 @@
:body "" :body ""
:cookies (cookies cfg {:value "" :max-age -1}))))))) :cookies (cookies cfg {:value "" :max-age -1})))))))
(defmethod ig/halt-key! ::session
[_ data]
(a/close! (::events-ch data)))
;; --- STATE INIT: SESSION UPDATER
(declare batch-events)
(declare update-sessions)
(s/def ::session map?)
(s/def ::max-batch-age ::cfg/http-session-updater-batch-max-age)
(s/def ::max-batch-size ::cfg/http-session-updater-batch-max-size)
(defmethod ig/pre-init-spec ::updater [_]
(s/keys :req-un [::db/pool ::wrk/executor ::mtx/metrics ::session]
:opt-un [::max-batch-age
::max-batch-size]))
(defmethod ig/prep-key ::updater
[_ cfg]
(merge {:max-batch-age (dt/duration {:minutes 5})
:max-batch-size 200}
(d/without-nils cfg)))
(defmethod ig/init-key ::updater
[_ {:keys [session metrics] :as cfg}]
(log/infof "initialize session updater (max-batch-age=%s, max-batch-size=%s)"
(str (:max-batch-age cfg))
(str (:max-batch-size cfg)))
(let [input (batch-events cfg (::events-ch session))
mcnt (mtx/create
{:name "http_session_updater_count"
:help "A counter of session update batch events."
:registry (:registry metrics)
:type :counter})]
(a/go-loop []
(when-let [[reason batch] (a/<! input)]
(let [result (a/<! (update-sessions cfg batch))]
(mcnt :inc)
(if (ex/exception? result)
(log/error result "updater: unexpected error on update sessions")
(log/debugf "updater: updated %s sessions (reason: %s)." result (name reason)))
(recur))))))
(defn- timeout-chan
[cfg]
(a/timeout (inst-ms (:max-batch-age cfg))))
(defn- batch-events
[cfg in]
(let [out (a/chan)]
(a/go-loop [tch (timeout-chan cfg)
buf #{}]
(let [[val port] (a/alts! [tch in])]
(cond
(identical? port tch)
(if (empty? buf)
(recur (timeout-chan cfg) buf)
(do
(a/>! out [:timeout buf])
(recur (timeout-chan cfg) #{})))
(nil? val)
(a/close! out)
(identical? port in)
(let [buf (conj buf val)]
(if (>= (count buf) (:max-batch-size cfg))
(do
(a/>! out [:size buf])
(recur (timeout-chan cfg) #{}))
(recur tch buf))))))
out))
(defn- update-sessions
[{:keys [pool executor]} ids]
(aa/with-thread executor
(db/exec-one! pool ["update http_session set updated_at=now() where id = ANY(?)"
(into-array String ids)])
(count ids)))
;; --- STATE INIT: SESSION GC
(declare sql:delete-expired)
(s/def ::max-age ::dt/duration)
(defmethod ig/pre-init-spec ::gc-task [_]
(s/keys :req-un [::db/pool]
:opt-un [::max-age]))
(defmethod ig/prep-key ::gc-task
[_ cfg]
(merge {:max-age (dt/duration {:days 2})}
(d/without-nils cfg)))
(defmethod ig/init-key ::gc-task
[_ {:keys [pool max-age] :as cfg}]
(fn [_]
(db/with-atomic [conn pool]
(let [interval (db/interval max-age)
result (db/exec-one! conn [sql:delete-expired interval])
result (:next.jdbc/update-count result)]
(log/debugf "gc-task: removed %s rows from http-session table" result)
result))))
(def ^:private
sql:delete-expired
"delete from http_session
where updated_at < now() - ?::interval")

8
backend/src/app/loggers/loki.clj
View file

@ -33,13 +33,13 @@
(defmethod ig/init-key ::reporter (defmethod ig/init-key ::reporter
[_ {:keys [receiver uri] :as cfg}] [_ {:keys [receiver uri] :as cfg}]
(when uri (when uri
(log/info "Intializing loki reporter.") (log/info "intializing loki reporter")
(let [output (a/chan (a/sliding-buffer 1024))] (let [output (a/chan (a/sliding-buffer 1024))]
(receiver :sub output) (receiver :sub output)
(a/go-loop [] (a/go-loop []
(let [msg (a/<! output)] (let [msg (a/<! output)]
(if (nil? msg) (if (nil? msg)
(log/info "Stoping error reporting loop.") (log/info "stoping error reporting loop")
(do (do
(a/<! (handle-event cfg msg)) (a/<! (handle-event cfg msg))
(recur))))) (recur)))))
@ -75,10 +75,10 @@
(if (= (:status response) 204) (if (= (:status response) 204)
true true
(do (do
(log/errorf "Error on sending log to loki (try %s).\n%s" i (pr-str response)) (log/errorf "error on sending log to loki (try %s)\n%s" i (pr-str response))
false))) false)))
(catch Exception e (catch Exception e
(log/errorf e "Error on sending message to loki (try %s)." i) (log/errorf e "error on sending message to loki (try %s)" i)
false))) false)))
(defn- handle-event (defn- handle-event

10
backend/src/app/loggers/mattermost.clj
View file

@ -43,14 +43,14 @@
(defmethod ig/init-key ::reporter (defmethod ig/init-key ::reporter
[_ {:keys [receiver] :as cfg}] [_ {:keys [receiver] :as cfg}]
(log/info "Intializing mattermost error reporter.") (log/info "intializing mattermost error reporter")
(let [output (a/chan (a/sliding-buffer 128) (let [output (a/chan (a/sliding-buffer 128)
(filter #(= (:level %) "error")))] (filter #(= (:level %) "error")))]
(receiver :sub output) (receiver :sub output)
(a/go-loop [] (a/go-loop []
(let [msg (a/<! output)] (let [msg (a/<! output)]
(if (nil? msg) (if (nil? msg)
(log/info "Stoping error reporting loop.") (log/info "stoping error reporting loop")
(do (do
(a/<! (handle-event cfg msg)) (a/<! (handle-event cfg msg))
(recur))))) (recur)))))
@ -75,10 +75,10 @@
:headers {"content-type" "application/json"} :headers {"content-type" "application/json"}
:body (json/encode-str {:text text})})] :body (json/encode-str {:text text})})]
(when (not= (:status rsp) 200) (when (not= (:status rsp) 200)
(log/errorf "Error on sending data to mattermost\n%s" (pr-str rsp)))) (log/errorf "error on sending data to mattermost\n%s" (pr-str rsp))))
(catch Exception e (catch Exception e
(log/error e "Unexpected exception on error reporter.")))) (log/error e "unexpected exception on error reporter"))))
(defn- persist-on-database! (defn- persist-on-database!
[{:keys [pool] :as cfg} {:keys [id] :as cdata}] [{:keys [pool] :as cfg} {:keys [id] :as cdata}]
@ -116,7 +116,7 @@
(send-mattermost-notification! cfg cdata)) (send-mattermost-notification! cfg cdata))
(persist-on-database! cfg cdata)) (persist-on-database! cfg cdata))
(catch Exception e (catch Exception e
(log/error e "Unexpected exception on error reporter."))))) (log/error e "unexpected exception on error reporter")))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Http Handler ;; Http Handler

2
backend/src/app/loggers/zmq.clj
View file

@ -34,7 +34,7 @@
(defmethod ig/init-key ::receiver (defmethod ig/init-key ::receiver
[_ {:keys [endpoint] :as cfg}] [_ {:keys [endpoint] :as cfg}]
(log/infof "Intializing ZMQ receiver on '%s'." endpoint) (log/infof "intializing ZMQ receiver on '%s'" endpoint)
(let [buffer (a/chan 1) (let [buffer (a/chan 1)
output (a/chan 1 (comp (filter map?) output (a/chan 1 (comp (filter map?)
(map prepare))) (map prepare)))

69
backend/src/app/main.clj
View file

@ -37,11 +37,19 @@
:max-pool-size 20} :max-pool-size 20}
:app.metrics/metrics :app.metrics/metrics
{} {:definitions
{:profile-register
{:name "actions_profile_register_count"
:help "A global counter of user registrations."
:type :counter}
:profile-activation
{:name "actions_profile_activation_count"
:help "A global counter of profile activations"
:type :counter}}}
:app.migrations/all :app.migrations/all
{:main (ig/ref :app.migrations/migrations) {:main (ig/ref :app.migrations/migrations)
:telemetry (ig/ref :app.telemetry/migrations)} :telemetry (ig/ref :app.telemetry/migrations)}
:app.migrations/migrations :app.migrations/migrations
{} {}
@ -69,7 +77,19 @@
:app.http.session/session :app.http.session/session
{:pool (ig/ref :app.db/pool) {:pool (ig/ref :app.db/pool)
:cookie-name "auth-token"} :cookie-name (:http-session-cookie-name config)}
:app.http.session/gc-task
{:pool (ig/ref :app.db/pool)
:max-age (:http-session-idle-max-age config)}
:app.http.session/updater
{:pool (ig/ref :app.db/pool)
:metrics (ig/ref :app.metrics/metrics)
:executor (ig/ref :app.worker/executor)
:session (ig/ref :app.http.session/session)
:max-batch-age (:http-session-updater-batch-max-age config)
:max-batch-size (:http-session-updater-batch-max-size config)}
:app.http.awsns/handler :app.http.awsns/handler
{:tokens (ig/ref :app.tokens/tokens) {:tokens (ig/ref :app.tokens/tokens)
@ -174,46 +194,61 @@
:app.worker/worker :app.worker/worker
{:executor (ig/ref :app.worker/executor) {:executor (ig/ref :app.worker/executor)
:pool (ig/ref :app.db/pool) :pool (ig/ref :app.db/pool)
:tasks (ig/ref :app.tasks/all)} :tasks (ig/ref :app.tasks/registry)}
:app.worker/scheduler :app.worker/scheduler
{:executor (ig/ref :app.worker/executor) {:executor (ig/ref :app.worker/executor)
:pool (ig/ref :app.db/pool) :pool (ig/ref :app.db/pool)
:tasks (ig/ref :app.tasks/registry)
:schedule :schedule
[{:id "file-media-gc" [{:id "file-media-gc"
:cron #app/cron "0 0 0 */1 * ? *" ;; daily :cron #app/cron "0 0 0 */1 * ? *" ;; daily
:fn (ig/ref :app.tasks.file-media-gc/handler)} :task :file-media-gc}
{:id "file-xlog-gc" {:id "file-xlog-gc"
:cron #app/cron "0 0 */1 * * ?" ;; hourly :cron #app/cron "0 0 */1 * * ?" ;; hourly
:fn (ig/ref :app.tasks.file-xlog-gc/handler)} :task :file-xlog-gc}
{:id "storage-deleted-gc" {:id "storage-deleted-gc"
:cron #app/cron "0 0 1 */1 * ?" ;; daily (1 hour shift) :cron #app/cron "0 0 1 */1 * ?" ;; daily (1 hour shift)
:fn (ig/ref :app.storage/gc-deleted-task)} :task :storage-deleted-gc}
{:id "storage-touched-gc" {:id "storage-touched-gc"
:cron #app/cron "0 0 2 */1 * ?" ;; daily (2 hour shift) :cron #app/cron "0 0 2 */1 * ?" ;; daily (2 hour shift)
:fn (ig/ref :app.storage/gc-touched-task)} :task :storage-touched-gc}
{:id "session-gc"
:cron #app/cron "0 0 3 */1 * ?" ;; daily (3 hour shift)
:task :session-gc}
{:id "storage-recheck" {:id "storage-recheck"
:cron #app/cron "0 0 */1 * * ?" ;; hourly :cron #app/cron "0 0 */1 * * ?" ;; hourly
:fn (ig/ref :app.storage/recheck-task)} :task :storage-recheck}
{:id "tasks-gc" {:id "tasks-gc"
:cron #app/cron "0 0 0 */1 * ?" ;; daily :cron #app/cron "0 0 0 */1 * ?" ;; daily
:fn (ig/ref :app.tasks.tasks-gc/handler)} :task :tasks-gc}
(when (:telemetry-enabled config) (when (:telemetry-enabled config)
{:id "telemetry" {:id "telemetry"
:cron #app/cron "0 0 */6 * * ?" ;; every 6h :cron #app/cron "0 0 */6 * * ?" ;; every 6h
:uri (:telemetry-uri config) :uri (:telemetry-uri config)
:fn (ig/ref :app.tasks.telemetry/handler)})]} :task :telemetry})]}
:app.tasks/all :app.tasks/registry
{"sendmail" (ig/ref :app.tasks.sendmail/handler) {:metrics (ig/ref :app.metrics/metrics)
"delete-object" (ig/ref :app.tasks.delete-object/handler) :tasks
"delete-profile" (ig/ref :app.tasks.delete-profile/handler)} {:sendmail (ig/ref :app.tasks.sendmail/handler)
:delete-object (ig/ref :app.tasks.delete-object/handler)
:delete-profile (ig/ref :app.tasks.delete-profile/handler)
:file-media-gc (ig/ref :app.tasks.file-media-gc/handler)
:file-xlog-gc (ig/ref :app.tasks.file-xlog-gc/handler)
:storage-deleted-gc (ig/ref :app.storage/gc-deleted-task)
:storage-touched-gc (ig/ref :app.storage/gc-touched-task)
:storage-recheck (ig/ref :app.storage/recheck-task)
:tasks-gc (ig/ref :app.tasks.tasks-gc/handler)
:telemetry (ig/ref :app.tasks.telemetry/handler)
:session-gc (ig/ref :app.http.session/gc-task)}}
:app.tasks.sendmail/handler :app.tasks.sendmail/handler
{:host (:smtp-host config) {:host (:smtp-host config)
@ -335,7 +370,7 @@
(-> system-config (-> system-config
(ig/prep) (ig/prep)
(ig/init)))) (ig/init))))
(log/infof "Welcome to penpot! Version: '%s'." (log/infof "welcome to penpot (version: '%s')"
(:full cfg/version)))) (:full cfg/version))))
(defn stop (defn stop

120
backend/src/app/metrics.clj
View file

@ -10,17 +10,15 @@
(ns app.metrics (ns app.metrics
(:require (:require
[app.common.exceptions :as ex] [app.common.exceptions :as ex]
[app.util.time :as dt]
[app.worker]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log] [clojure.tools.logging :as log]
[integrant.core :as ig] [integrant.core :as ig])
[next.jdbc :as jdbc])
(:import (:import
io.prometheus.client.CollectorRegistry io.prometheus.client.CollectorRegistry
io.prometheus.client.Counter io.prometheus.client.Counter
io.prometheus.client.Gauge io.prometheus.client.Gauge
io.prometheus.client.Summary io.prometheus.client.Summary
io.prometheus.client.Histogram
io.prometheus.client.exporter.common.TextFormat io.prometheus.client.exporter.common.TextFormat
io.prometheus.client.hotspot.DefaultExports io.prometheus.client.hotspot.DefaultExports
io.prometheus.client.jetty.JettyStatisticsCollector io.prometheus.client.jetty.JettyStatisticsCollector
@ -30,41 +28,12 @@
(declare instrument-vars!) (declare instrument-vars!)
(declare instrument) (declare instrument)
(declare create-registry) (declare create-registry)
(declare create)
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Entry Point ;; Entry Point
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn- instrument-jdbc!
[registry]
(instrument-vars!
[#'next.jdbc/execute-one!
#'next.jdbc/execute!]
{:registry registry
:type :counter
:name "database_query_counter"
:help "An absolute counter of database queries."}))
(defn- instrument-workers!
[registry]
(instrument-vars!
[#'app.worker/run-task]
{:registry registry
:type :summary
:name "worker_task_checkout_millis"
:help "Latency measured between scheduld_at and execution time."
:wrap (fn [rootf mobj]
(let [mdata (meta rootf)
origf (::original mdata rootf)]
(with-meta
(fn [tasks item]
(let [now (inst-ms (dt/now))
sat (inst-ms (:scheduled-at item))]
(mobj :observe (- now sat))
(origf tasks item)))
{::original origf})))}))
(defn- handler (defn- handler
[registry _request] [registry _request]
(let [samples (.metricFamilySamples ^CollectorRegistry registry) (let [samples (.metricFamilySamples ^CollectorRegistry registry)
@ -73,13 +42,24 @@
{:headers {"content-type" TextFormat/CONTENT_TYPE_004} {:headers {"content-type" TextFormat/CONTENT_TYPE_004}
:body (.toString writer)})) :body (.toString writer)}))
(s/def ::definitions
(s/map-of keyword? map?))
(defmethod ig/pre-init-spec ::metrics [_]
(s/keys :opt-un [::definitions]))
(defmethod ig/init-key ::metrics (defmethod ig/init-key ::metrics
[_ _cfg] [_ {:keys [definitions] :as cfg}]
(log/infof "Initializing prometheus registry and instrumentation.") (log/infof "Initializing prometheus registry and instrumentation.")
(let [registry (create-registry)] (let [registry (create-registry)
(instrument-workers! registry) definitions (reduce-kv (fn [res k v]
(instrument-jdbc! registry) (->> (assoc v :registry registry)
(create)
(assoc res k)))
{}
definitions)]
{:handler (partial handler registry) {:handler (partial handler registry)
:definitions definitions
:registry registry})) :registry registry}))
(s/def ::handler fn?) (s/def ::handler fn?)
@ -87,7 +67,6 @@
(s/def ::metrics (s/def ::metrics
(s/keys :req-un [::registry ::handler])) (s/keys :req-un [::registry ::handler]))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Implementation ;; Implementation
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@ -126,7 +105,7 @@
(invoke [_ cmd labels] (invoke [_ cmd labels]
(.. ^Counter instance (.. ^Counter instance
(labels labels) (labels (into-array String labels))
(inc)))))) (inc))))))
(defn make-gauge (defn make-gauge
@ -150,19 +129,27 @@
:dec (.dec ^Gauge instance))) :dec (.dec ^Gauge instance)))
(invoke [_ cmd labels] (invoke [_ cmd labels]
(case cmd (let [labels (into-array String [labels])]
:inc (.. ^Gauge instance (labels labels) (inc)) (case cmd
:dec (.. ^Gauge instance (labels labels) (dec))))))) :inc (.. ^Gauge instance (labels labels) (inc))
:dec (.. ^Gauge instance (labels labels) (dec))))))))
(def default-quantiles
[[0.75 0.02]
[0.99 0.001]])
(defn make-summary (defn make-summary
[{:keys [name help registry reg labels max-age] :or {max-age 3600} :as props}] [{:keys [name help registry reg labels max-age quantiles buckets]
:or {max-age 3600 buckets 6 quantiles default-quantiles} :as props}]
(let [registry (or registry reg) (let [registry (or registry reg)
instance (doto (Summary/build) instance (doto (Summary/build)
(.name name) (.name name)
(.help help) (.help help))
(.maxAgeSeconds max-age) _ (when (seq quantiles)
(.quantile 0.75 0.02) (.maxAgeSeconds ^Summary instance max-age)
(.quantile 0.99 0.001)) (.ageBuckets ^Summary instance buckets))
_ (doseq [[q e] quantiles]
(.quantile ^Summary instance q e))
_ (when (seq labels) _ (when (seq labels)
(.labelNames instance (into-array String labels))) (.labelNames instance (into-array String labels)))
instance (.register instance registry)] instance (.register instance registry)]
@ -176,7 +163,34 @@
(invoke [_ cmd val labels] (invoke [_ cmd val labels]
(.. ^Summary instance (.. ^Summary instance
(labels labels) (labels (into-array String labels))
(observe val))))))
(def default-histogram-buckets
[1 5 10 25 50 75 100 250 500 750 1000 2500 5000 7500])
(defn make-histogram
[{:keys [name help registry reg labels buckets]
:or {buckets default-histogram-buckets}}]
(let [registry (or registry reg)
instance (doto (Histogram/build)
(.name name)
(.help help)
(.buckets (into-array Double/TYPE buckets)))
_ (when (seq labels)
(.labelNames instance (into-array String labels)))
instance (.register instance registry)]
(reify
clojure.lang.IDeref
(deref [_] instance)
clojure.lang.IFn
(invoke [_ cmd val]
(.observe ^Histogram instance val))
(invoke [_ cmd val labels]
(.. ^Histogram instance
(labels (into-array String labels))
(observe val)))))) (observe val))))))
(defn create (defn create
@ -184,7 +198,8 @@
(case type (case type
:counter (make-counter props) :counter (make-counter props)
:gauge (make-gauge props) :gauge (make-gauge props)
:summary (make-summary props))) :summary (make-summary props)
:histogram (make-histogram props)))
(defn wrap-counter (defn wrap-counter
([rootf mobj] ([rootf mobj]
@ -204,7 +219,6 @@
(assoc mdata ::original origf)))) (assoc mdata ::original origf))))
([rootf mobj labels] ([rootf mobj labels]
(let [mdata (meta rootf) (let [mdata (meta rootf)
labels (into-array String labels)
origf (::original mdata rootf)] origf (::original mdata rootf)]
(with-meta (with-meta
(fn (fn
@ -241,7 +255,6 @@
([rootf mobj labels] ([rootf mobj labels]
(let [mdata (meta rootf) (let [mdata (meta rootf)
labels (into-array String labels)
origf (::original mdata rootf)] origf (::original mdata rootf)]
(with-meta (with-meta
(fn (fn
@ -284,6 +297,9 @@
(instance? Summary @obj) (instance? Summary @obj)
((or wrap wrap-summary) f obj) ((or wrap wrap-summary) f obj)
(instance? Histogram @obj)
((or wrap wrap-summary) f obj)
:else :else
(ex/raise :type :not-implemented)))) (ex/raise :type :not-implemented))))

2
backend/src/app/migrations.clj
View file

@ -158,6 +158,8 @@
{:name "0048-mod-storage-tables" {:name "0048-mod-storage-tables"
:fn (mg/resource "app/migrations/sql/0048-mod-storage-tables.sql")} :fn (mg/resource "app/migrations/sql/0048-mod-storage-tables.sql")}
{:name "0049-mod-http-session-table"
:fn (mg/resource "app/migrations/sql/0049-mod-http-session-table.sql")}
]) ])

6
backend/src/app/migrations/sql/0049-mod-http-session-table.sql Normal file
View file

@ -0,0 +1,6 @@
ALTER TABLE http_session
ADD COLUMN updated_at timestamptz NULL;
CREATE INDEX http_session__updated_at__idx
ON http_session (updated_at)
WHERE updated_at IS NOT NULL;

52
backend/src/app/notifications.clj
View file

@ -14,6 +14,7 @@
[app.db :as db] [app.db :as db]
[app.metrics :as mtx] [app.metrics :as mtx]
[app.util.async :as aa] [app.util.async :as aa]
[app.util.time :as dt]
[app.util.transit :as t] [app.util.transit :as t]
[clojure.core.async :as a] [clojure.core.async :as a]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
@ -46,29 +47,32 @@
mtx-active-connections mtx-active-connections
(mtx/create (mtx/create
{:name "websocket_notifications_active_connections" {:name "websocket_active_connections"
:registry (:registry metrics) :registry (:registry metrics)
:type :gauge :type :gauge
:help "Active websocket connections on notifications service."}) :help "Active websocket connections."})
mtx-message-recv mtx-messages
(mtx/create (mtx/create
{:name "websocket_notifications_message_recv_timing" {:name "websocket_message_count"
:registry (:registry metrics) :registry (:registry metrics)
:type :summary :labels ["op"]
:help "Message receive summary timing (ms)."}) :type :counter
:help "Counter of processed messages."})
mtx-message-send mtx-sessions
(mtx/create (mtx/create
{:name "websocket_notifications_message_send_timing" {:name "websocket_session_timing"
:registry (:registry metrics) :registry (:registry metrics)
:type :summary :quantiles []
:help "Message receive summary timing (ms)."}) :help "Websocket session timing (seconds)."
:type :summary})
cfg (assoc cfg cfg (assoc cfg
:mtx-active-connections mtx-active-connections :mtx-active-connections mtx-active-connections
:mtx-message-recv mtx-message-recv :mtx-messages mtx-messages
:mtx-message-send mtx-message-send)] :mtx-sessions mtx-sessions
)]
(-> #(handler cfg %) (-> #(handler cfg %)
(wrap-session) (wrap-session)
(wrap-keyword-params) (wrap-keyword-params)
@ -130,16 +134,17 @@
(defn websocket (defn websocket
[{:keys [file-id team-id msgbus] :as cfg}] [{:keys [file-id team-id msgbus] :as cfg}]
(let [in (a/chan 32) (let [in (a/chan 32)
out (a/chan 32) out (a/chan 32)
mtx-active-connections (:mtx-active-connections cfg) mtx-aconn (:mtx-active-connections cfg)
mtx-message-send (:mtx-message-send cfg) mtx-messages (:mtx-messages cfg)
mtx-message-recv (:mtx-message-recv cfg) mtx-sessions (:mtx-sessions cfg)
created-at (dt/now)
ws-send (mtx/wrap-summary ws-send mtx-message-send)] ws-send (mtx/wrap-counter ws-send mtx-messages ["send"])]
(letfn [(on-connect [conn] (letfn [(on-connect [conn]
(mtx-active-connections :inc) (mtx-aconn :inc)
(let [sub (a/chan) (let [sub (a/chan)
ws (WebSocket. conn in out sub nil cfg)] ws (WebSocket. conn in out sub nil cfg)]
@ -159,11 +164,14 @@
(a/close! sub)))) (a/close! sub))))
(on-error [_conn _e] (on-error [_conn _e]
(mtx-aconn :dec)
(mtx-sessions :observe (/ (inst-ms (dt/duration-between created-at (dt/now))) 1000.0))
(a/close! out) (a/close! out)
(a/close! in)) (a/close! in))
(on-close [_conn _status _reason] (on-close [_conn _status _reason]
(mtx-active-connections :dec) (mtx-aconn :dec)
(mtx-sessions :observe (/ (inst-ms (dt/duration-between created-at (dt/now))) 1000.0))
(a/close! out) (a/close! out)
(a/close! in)) (a/close! in))
@ -174,7 +182,7 @@
{:on-connect on-connect {:on-connect on-connect
:on-error on-error :on-error on-error
:on-close on-close :on-close on-close
:on-text (mtx/wrap-summary on-message mtx-message-recv) :on-text (mtx/wrap-counter on-message mtx-messages ["recv"])
:on-bytes (constantly nil)}))) :on-bytes (constantly nil)})))
(declare handle-message) (declare handle-message)
@ -188,7 +196,7 @@
(aa/<? (start-loop! ws)) (aa/<? (start-loop! ws))
(aa/<? (handle-message ws {:type :disconnect})) (aa/<? (handle-message ws {:type :disconnect}))
(catch Throwable err (catch Throwable err
(log/errorf err "Unexpected exception on websocket handler.") (log/errorf err "unexpected exception on websocket handler")
(let [session (.getSession ^WebSocketAdapter conn)] (let [session (.getSession ^WebSocketAdapter conn)]
(when session (when session
(.disconnect session))))))) (.disconnect session)))))))

19
backend/src/app/rpc.clj
View file

@ -25,6 +25,11 @@
[_] [_]
(ex/raise :type :not-found)) (ex/raise :type :not-found))
(defn- run-hook
[hook-fn response]
(ex/ignoring (hook-fn))
response)
(defn- rpc-query-handler (defn- rpc-query-handler
[methods {:keys [profile-id] :as request}] [methods {:keys [profile-id] :as request}]
(let [type (keyword (get-in request [:path-params :type])) (let [type (keyword (get-in request [:path-params :type]))
@ -50,7 +55,11 @@
result ((get methods type default-handler) data) result ((get methods type default-handler) data)
mdata (meta result)] mdata (meta result)]
(cond->> {:status 200 :body result} (cond->> {:status 200 :body result}
(fn? (:transform-response mdata)) ((:transform-response mdata) request)))) (fn? (:transform-response mdata))
((:transform-response mdata) request)
(fn? (:before-complete mdata))
(run-hook (:before-complete mdata)))))
(defn- wrap-with-metrics (defn- wrap-with-metrics
[cfg f mdata] [cfg f mdata]
@ -66,7 +75,7 @@
(ex/raise :type :internal (ex/raise :type :internal
:code :rlimit-not-configured :code :rlimit-not-configured
:hint (str/fmt "%s rlimit not configured" key))) :hint (str/fmt "%s rlimit not configured" key)))
(log/tracef "Adding rlimit to '%s' rpc handler." (::sv/name mdata)) (log/tracef "adding rlimit to '%s' rpc handler" (::sv/name mdata))
(fn [cfg params] (fn [cfg params]
(rlm/execute rlinst (f cfg params)))) (rlm/execute rlinst (f cfg params))))
f)) f))
@ -76,7 +85,7 @@
(let [f (wrap-with-rlimits cfg f mdata) (let [f (wrap-with-rlimits cfg f mdata)
f (wrap-with-metrics cfg f mdata) f (wrap-with-metrics cfg f mdata)
spec (or (::sv/spec mdata) (s/spec any?))] spec (or (::sv/spec mdata) (s/spec any?))]
(log/tracef "Registering '%s' command to rpc service." (::sv/name mdata)) (log/tracef "registering '%s' command to rpc service" (::sv/name mdata))
(fn [params] (fn [params]
(when (and (:auth mdata true) (not (uuid? (:profile-id params)))) (when (and (:auth mdata true) (not (uuid? (:profile-id params))))
(ex/raise :type :authentication (ex/raise :type :authentication
@ -96,7 +105,7 @@
{:name "rpc_query_timing" {:name "rpc_query_timing"
:labels ["name"] :labels ["name"]
:registry (get-in cfg [:metrics :registry]) :registry (get-in cfg [:metrics :registry])
:type :summary :type :histogram
:help "Timing of query services."}) :help "Timing of query services."})
cfg (assoc cfg ::mobj mobj)] cfg (assoc cfg ::mobj mobj)]
(->> (sv/scan-ns 'app.rpc.queries.projects (->> (sv/scan-ns 'app.rpc.queries.projects
@ -115,7 +124,7 @@
{:name "rpc_mutation_timing" {:name "rpc_mutation_timing"
:labels ["name"] :labels ["name"]
:registry (get-in cfg [:metrics :registry]) :registry (get-in cfg [:metrics :registry])
:type :summary :type :histogram
:help "Timing of mutation services."}) :help "Timing of mutation services."})
cfg (assoc cfg ::mobj mobj)] cfg (assoc cfg ::mobj mobj)]
(->> (sv/scan-ns 'app.rpc.mutations.demo (->> (sv/scan-ns 'app.rpc.mutations.demo

2
backend/src/app/rpc/mutations/ldap.clj
View file

@ -30,7 +30,7 @@
(try (try
(ldap/connect params) (ldap/connect params)
(catch Exception e (catch Exception e
(log/errorf e "Cannot connect to LDAP %s:%s" (log/errorf e "cannot connect to LDAP %s:%s"
(get-in params [:host :address]) (get-in params [:host :address])
(get-in params [:host :port]))))))) (get-in params [:host :port])))))))

108
backend/src/app/rpc/mutations/profile.clj
View file

@ -26,7 +26,6 @@
[app.util.time :as dt] [app.util.time :as dt]
[buddy.hashers :as hashers] [buddy.hashers :as hashers]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log]
[cuerdas.core :as str])) [cuerdas.core :as str]))
;; --- Helpers & Specs ;; --- Helpers & Specs
@ -42,10 +41,12 @@
;; --- Mutation: Register Profile ;; --- Mutation: Register Profile
(declare annotate-profile-register)
(declare check-profile-existence!) (declare check-profile-existence!)
(declare create-profile) (declare create-profile)
(declare create-profile-relations) (declare create-profile-relations)
(declare email-domain-in-whitelist?) (declare email-domain-in-whitelist?)
(declare register-profile)
(s/def ::invitation-token ::us/not-empty-string) (s/def ::invitation-token ::us/not-empty-string)
(s/def ::register-profile (s/def ::register-profile
@ -63,48 +64,64 @@
:code :email-domain-is-not-allowed)) :code :email-domain-is-not-allowed))
(db/with-atomic [conn pool] (db/with-atomic [conn pool]
(check-profile-existence! conn params) (let [cfg (assoc cfg :conn conn)]
(let [profile (->> (create-profile conn params) (register-profile cfg params))))
(create-profile-relations conn))]
(create-profile-initial-data conn profile)
(if-let [token (:invitation-token params)] (defn- annotate-profile-register
;; If invitation token comes in params, this is because the "A helper for properly increase the profile-register metric once the
;; user comes from team-invitation process; in this case, transaction is completed."
;; regenerate token and send back to the user a new invitation [metrics profile]
;; token (and mark current session as logged). (fn []
(let [claims (tokens :verify {:token token :iss :team-invitation}) (when (::created profile)
claims (assoc claims ((get-in metrics [:definitions :profile-register]) :inc))))
:member-id (:id profile)
:member-email (:email profile))
token (tokens :generate claims)]
(with-meta
{:invitation-token token}
{:transform-response ((:create session) (:id profile))}))
;; If no token is provided, send a verification email (defn- register-profile
(let [vtoken (tokens :generate [{:keys [conn tokens session metrics] :as cfg} params]
{:iss :verify-email (check-profile-existence! conn params)
:exp (dt/in-future "48h") (let [profile (->> (create-profile conn params)
:profile-id (:id profile) (create-profile-relations conn))
:email (:email profile)}) profile (assoc profile ::created true)]
ptoken (tokens :generate-predefined (create-profile-initial-data conn profile)
{:iss :profile-identity
:profile-id (:id profile)})]
;; Don't allow proceed in register page if the email is (if-let [token (:invitation-token params)]
;; already reported as permanent bounced ;; If invitation token comes in params, this is because the
(when (emails/has-bounce-reports? conn (:email profile)) ;; user comes from team-invitation process; in this case,
(ex/raise :type :validation ;; regenerate token and send back to the user a new invitation
:code :email-has-permanent-bounces ;; token (and mark current session as logged).
:hint "looks like the email has one or many bounces reported")) (let [claims (tokens :verify {:token token :iss :team-invitation})
claims (assoc claims
:member-id (:id profile)
:member-email (:email profile))
token (tokens :generate claims)
resp {:invitation-token token}]
(with-meta resp
{:transform-response ((:create session) (:id profile))
:before-complete (annotate-profile-register metrics profile)}))
(emails/send! conn emails/register ;; If no token is provided, send a verification email
{:to (:email profile) (let [vtoken (tokens :generate
:name (:fullname profile) {:iss :verify-email
:token vtoken :exp (dt/in-future "48h")
:extra-data ptoken}) :profile-id (:id profile)
profile))))) :email (:email profile)})
ptoken (tokens :generate-predefined
{:iss :profile-identity
:profile-id (:id profile)})]
;; Don't allow proceed in register page if the email is
;; already reported as permanent bounced
(when (emails/has-bounce-reports? conn (:email profile))
(ex/raise :type :validation
:code :email-has-permanent-bounces
:hint "looks like the email has one or many bounces reported"))
(emails/send! conn emails/register
{:to (:email profile)
:name (:fullname profile)
:token vtoken
:extra-data ptoken})
(with-meta profile
{:before-complete (annotate-profile-register metrics profile)})))))
(defn email-domain-in-whitelist? (defn email-domain-in-whitelist?
"Returns true if email's domain is in the given whitelist or if given "Returns true if email's domain is in the given whitelist or if given
@ -142,8 +159,7 @@
[attempt password] [attempt password]
(try (try
(hashers/verify attempt password) (hashers/verify attempt password)
(catch Exception e (catch Exception _e
(log/warnf e "Error on verify password (only informative, nothing affected to user).")
{:update false {:update false
:valid false}))) :valid false})))
@ -269,10 +285,12 @@
(s/keys :req-un [::email ::fullname ::backend])) (s/keys :req-un [::email ::fullname ::backend]))
(sv/defmethod ::login-or-register {:auth false} (sv/defmethod ::login-or-register {:auth false}
[{:keys [pool] :as cfg} params] [{:keys [pool metrics] :as cfg} params]
(db/with-atomic [conn pool] (db/with-atomic [conn pool]
(-> (assoc cfg :conn conn) (let [profile (-> (assoc cfg :conn conn)
(login-or-register params)))) (login-or-register params))]
(with-meta profile
{:before-complete (annotate-profile-register metrics profile)}))))
(defn login-or-register (defn login-or-register
[{:keys [conn] :as cfg} {:keys [email backend] :as params}] [{:keys [conn] :as cfg} {:keys [email backend] :as params}]
@ -294,7 +312,7 @@
(let [profile (->> (create-profile conn params) (let [profile (->> (create-profile conn params)
(create-profile-relations conn))] (create-profile-relations conn))]
(create-profile-initial-data conn profile) (create-profile-initial-data conn profile)
profile))] (assoc profile ::created true)))]
(let [profile (profile/retrieve-profile-data-by-email conn email) (let [profile (profile/retrieve-profile-data-by-email conn email)
profile (if profile profile (if profile

12
backend/src/app/rpc/mutations/verify_token.clj
View file

@ -40,8 +40,15 @@
{:id profile-id}) {:id profile-id})
claims) claims)
(defn- annotate-profile-activation
"A helper for properly increase the profile-activation metric once the
transaction is completed."
[metrics]
(fn []
((get-in metrics [:definitions :profile-activation]) :inc)))
(defmethod process-token :verify-email (defmethod process-token :verify-email
[{:keys [conn session] :as cfg} _params {:keys [profile-id] :as claims}] [{:keys [conn session metrics] :as cfg} _ {:keys [profile-id] :as claims}]
(let [profile (profile/retrieve-profile conn profile-id) (let [profile (profile/retrieve-profile conn profile-id)
claims (assoc claims :profile profile)] claims (assoc claims :profile profile)]
@ -56,7 +63,8 @@
{:id (:id profile)})) {:id (:id profile)}))
(with-meta claims (with-meta claims
{:transform-response ((:create session) profile-id)}))) {:transform-response ((:create session) profile-id)
:before-complete (annotate-profile-activation metrics)})))
(defmethod process-token :auth (defmethod process-token :auth
[{:keys [conn] :as cfg} _params {:keys [profile-id] :as claims}] [{:keys [conn] :as cfg} _params {:keys [profile-id] :as claims}]

75
backend/src/app/tasks.clj
View file

@ -5,17 +5,19 @@
;; This Source Code Form is "Incompatible With Secondary Licenses", as ;; This Source Code Form is "Incompatible With Secondary Licenses", as
;; defined by the Mozilla Public License, v. 2.0. ;; defined by the Mozilla Public License, v. 2.0.
;; ;;
;; Copyright (c) 2020 UXBOX Labs SL ;; Copyright (c) 2020-2021 UXBOX Labs SL
(ns app.tasks (ns app.tasks
(:require (:require
[app.common.spec :as us] [app.common.spec :as us]
[app.common.uuid :as uuid] [app.common.uuid :as uuid]
[app.db :as db] [app.db :as db]
;; [app.metrics :as mtx] [app.metrics :as mtx]
[app.util.time :as dt] [app.util.time :as dt]
[app.worker]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log])) [clojure.tools.logging :as log]
[integrant.core :as ig]))
(s/def ::name ::us/string) (s/def ::name ::us/string)
(s/def ::delay (s/def ::delay
@ -41,11 +43,68 @@
interval (db/interval duration) interval (db/interval duration)
props (db/tjson props) props (db/tjson props)
id (uuid/next)] id (uuid/next)]
(log/infof "Submit task '%s' to be executed in '%s'." name (str duration)) (log/debugf "submit task '%s' to be executed in '%s'" name (str duration))
(db/exec-one! conn [sql:insert-new-task id name props queue priority max-retries interval]) (db/exec-one! conn [sql:insert-new-task id name props queue priority max-retries interval])
id)) id))
;; (mtx/instrument-with-counter! (defn- instrument!
;; {:var #'submit! [registry]
;; :id "tasks__submit_counter" (mtx/instrument-vars!
;; :help "Absolute task submit counter."}) [#'submit!]
{:registry registry
:type :counter
:labels ["name"]
:name "tasks_submit_counter"
:help "An absolute counter of task submissions."
:wrap (fn [rootf mobj]
(let [mdata (meta rootf)
origf (::original mdata rootf)]
(with-meta
(fn [conn params]
(let [tname (:name params)]
(mobj :inc [tname])
(origf conn params)))
{::original origf})))})
(mtx/instrument-vars!
[#'app.worker/run-task]
{:registry registry
:type :summary
:quantiles []
:name "tasks_checkout_timing"
:help "Latency measured between scheduld_at and execution time."
:wrap (fn [rootf mobj]
(let [mdata (meta rootf)
origf (::original mdata rootf)]
(with-meta
(fn [tasks item]
(let [now (inst-ms (dt/now))
sat (inst-ms (:scheduled-at item))]
(mobj :observe (- now sat))
(origf tasks item)))
{::original origf})))}))
;; --- STATE INIT: REGISTRY
(s/def ::tasks
(s/map-of keyword? fn?))
(defmethod ig/pre-init-spec ::registry [_]
(s/keys :req-un [::mtx/metrics ::tasks]))
(defmethod ig/init-key ::registry
[_ {:keys [metrics tasks]}]
(instrument! (:registry metrics))
(let [mobj (mtx/create
{:registry (:registry metrics)
:type :summary
:labels ["name"]
:quantiles []
:name "tasks_timing"
:help "Background task execution timing."})]
(reduce-kv (fn [res k v]
(let [tname (name k)]
(log/debugf "registring task '%s'" tname)
(assoc res tname (mtx/wrap-summary v mobj [tname]))))
{}
tasks)))

24
backend/src/app/tasks/delete_object.clj
View file

@ -12,42 +12,32 @@
(:require (:require
[app.common.spec :as us] [app.common.spec :as us]
[app.db :as db] [app.db :as db]
[app.metrics :as mtx]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log] [clojure.tools.logging :as log]
[integrant.core :as ig])) [integrant.core :as ig]))
(declare handler)
(declare handle-deletion) (declare handle-deletion)
(defmethod ig/pre-init-spec ::handler [_] (defmethod ig/pre-init-spec ::handler [_]
(s/keys :req-un [::db/pool ::mtx/metrics])) (s/keys :req-un [::db/pool]))
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ {:keys [metrics] :as cfg}] [_ {:keys [pool] :as cfg}]
(let [handler #(handler cfg %)] (fn [{:keys [props] :as task}]
(->> {:registry (:registry metrics) (us/verify ::props props)
:type :summary (db/with-atomic [conn pool]
:name "task_delete_object_timing" (handle-deletion conn props))))
:help "delete object task timing"}
(mtx/instrument handler))))
(s/def ::type ::us/keyword) (s/def ::type ::us/keyword)
(s/def ::id ::us/uuid) (s/def ::id ::us/uuid)
(s/def ::props (s/keys :req-un [::id ::type])) (s/def ::props (s/keys :req-un [::id ::type]))
(defn- handler
[{:keys [pool]} {:keys [props] :as task}]
(us/verify ::props props)
(db/with-atomic [conn pool]
(handle-deletion conn props)))
(defmulti handle-deletion (defmulti handle-deletion
(fn [_ props] (:type props))) (fn [_ props] (:type props)))
(defmethod handle-deletion :default (defmethod handle-deletion :default
[_conn {:keys [type]}] [_conn {:keys [type]}]
(log/warnf "no handler found for %s" type)) (log/warnf "no handler found for '%s'" type))
(defmethod handle-deletion :file (defmethod handle-deletion :file
[conn {:keys [id] :as props}] [conn {:keys [id] :as props}]

36
backend/src/app/tasks/delete_profile.clj
View file

@ -13,27 +13,16 @@
[app.common.spec :as us] [app.common.spec :as us]
[app.db :as db] [app.db :as db]
[app.db.sql :as sql] [app.db.sql :as sql]
[app.metrics :as mtx]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log] [clojure.tools.logging :as log]
[integrant.core :as ig])) [integrant.core :as ig]))
(declare delete-profile-data) (declare delete-profile-data)
(declare handler)
;; --- INIT ;; --- INIT
(defmethod ig/pre-init-spec ::handler [_] (defmethod ig/pre-init-spec ::handler [_]
(s/keys :req-un [::db/pool ::mtx/metrics])) (s/keys :req-un [::db/pool]))
(defmethod ig/init-key ::handler
[_ {:keys [metrics] :as cfg}]
(let [handler #(handler cfg %)]
(->> {:registry (:registry metrics)
:type :summary
:name "task_delete_profile_timing"
:help "delete profile task timing"}
(mtx/instrument handler))))
;; This task is responsible to permanently delete a profile with all ;; This task is responsible to permanently delete a profile with all
;; the dependent data. As step (1) we delete all owned teams of the ;; the dependent data. As step (1) we delete all owned teams of the
@ -48,16 +37,17 @@
(s/def ::profile-id ::us/uuid) (s/def ::profile-id ::us/uuid)
(s/def ::props (s/keys :req-un [::profile-id])) (s/def ::props (s/keys :req-un [::profile-id]))
(defn handler (defmethod ig/init-key ::handler
[{:keys [pool]} {:keys [props] :as task}] [_ {:keys [pool] :as cfg}]
(us/verify ::props props) (fn [{:keys [props] :as task}]
(db/with-atomic [conn pool] (us/verify ::props props)
(let [id (:profile-id props) (db/with-atomic [conn pool]
profile (db/exec-one! conn (sql/select :profile {:id id} {:for-update true}))] (let [id (:profile-id props)
(if (or (:is-demo profile) profile (db/exec-one! conn (sql/select :profile {:id id} {:for-update true}))]
(:deleted-at profile)) (if (or (:is-demo profile)
(delete-profile-data conn id) (:deleted-at profile))
(log/warnf "Profile %s does not match constraints for deletion" id))))) (delete-profile-data conn id)
(log/warnf "profile '%s' does not match constraints for deletion" id))))))
;; --- IMPL ;; --- IMPL
@ -80,7 +70,7 @@
(defn- delete-profile-data (defn- delete-profile-data
[conn profile-id] [conn profile-id]
(log/infof "Proceding to delete all data related to profile id = %s" profile-id) (log/debugf "proceding to delete all data related to profile '%s'" profile-id)
(delete-teams conn profile-id) (delete-teams conn profile-id)
(delete-profile conn profile-id) (delete-profile conn profile-id)
true) true)

42
backend/src/app/tasks/file_media_gc.clj
View file

@ -5,7 +5,7 @@
;; This Source Code Form is "Incompatible With Secondary Licenses", as ;; This Source Code Form is "Incompatible With Secondary Licenses", as
;; defined by the Mozilla Public License, v. 2.0. ;; defined by the Mozilla Public License, v. 2.0.
;; ;;
;; Copyright (c) 2020 UXBOX Labs SL ;; Copyright (c) 2020-2021 UXBOX Labs SL
(ns app.tasks.file-media-gc (ns app.tasks.file-media-gc
"A maintenance task that is responsible to purge the unused media "A maintenance task that is responsible to purge the unused media
@ -14,44 +14,34 @@
(:require (:require
[app.common.pages.migrations :as pmg] [app.common.pages.migrations :as pmg]
[app.db :as db] [app.db :as db]
[app.metrics :as mtx]
[app.util.blob :as blob] [app.util.blob :as blob]
[app.util.time :as dt] [app.util.time :as dt]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log] [clojure.tools.logging :as log]
[integrant.core :as ig])) [integrant.core :as ig]))
(declare handler)
(declare process-file) (declare process-file)
(declare retrieve-candidates) (declare retrieve-candidates)
(s/def ::max-age ::dt/duration) (s/def ::max-age ::dt/duration)
(defmethod ig/pre-init-spec ::handler [_] (defmethod ig/pre-init-spec ::handler [_]
(s/keys :req-un [::db/pool ::mtx/metrics ::max-age])) (s/keys :req-un [::db/pool ::max-age]))
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ {:keys [metrics] :as cfg}] [_ {:keys [pool] :as cfg}]
(let [handler #(handler cfg %)] (fn [_]
(->> {:registry (:registry metrics) (db/with-atomic [conn pool]
:type :summary (let [cfg (assoc cfg :conn conn)]
:name "task_file_media_gc_timing" (loop [n 0]
:help "file media garbage collection task timing"} (let [files (retrieve-candidates cfg)]
(mtx/instrument handler)))) (if (seq files)
(do
(defn- handler (run! (partial process-file cfg) files)
[{:keys [pool] :as cfg} _] (recur (+ n (count files))))
(db/with-atomic [conn pool] (do
(let [cfg (assoc cfg :conn conn)] (log/debugf "finalized with total of %s processed files" n)
(loop [n 0] {:processed n}))))))))
(let [files (retrieve-candidates cfg)]
(if (seq files)
(do
(run! (partial process-file cfg) files)
(recur (+ n (count files))))
(do
(log/infof "finalized with total of %s processed files" n)
{:processed n})))))))
(def ^:private (def ^:private
sql:retrieve-candidates-chunk sql:retrieve-candidates-chunk
@ -98,7 +88,7 @@
unused (->> (db/query conn :file-media-object {:file-id id}) unused (->> (db/query conn :file-media-object {:file-id id})
(remove #(contains? used (:id %))))] (remove #(contains? used (:id %))))]
(log/infof "processing file: id='%s' age='%s' to-delete=%s" id age (count unused)) (log/debugf "processing file: id='%s' age='%s' to-delete=%s" id age (count unused))
;; Mark file as trimmed ;; Mark file as trimmed
(db/update! conn :file (db/update! conn :file

31
backend/src/app/tasks/file_xlog_gc.clj
View file

@ -5,45 +5,36 @@
;; This Source Code Form is "Incompatible With Secondary Licenses", as ;; This Source Code Form is "Incompatible With Secondary Licenses", as
;; defined by the Mozilla Public License, v. 2.0. ;; defined by the Mozilla Public License, v. 2.0.
;; ;;
;; Copyright (c) 2020 UXBOX Labs SL ;; Copyright (c) 2020-2021 UXBOX Labs SL
(ns app.tasks.file-xlog-gc (ns app.tasks.file-xlog-gc
"A maintenance task that performs a garbage collection of the file "A maintenance task that performs a garbage collection of the file
change (transaction) log." change (transaction) log."
(:require (:require
[app.db :as db] [app.db :as db]
[app.metrics :as mtx]
[app.util.time :as dt] [app.util.time :as dt]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log] [clojure.tools.logging :as log]
[integrant.core :as ig])) [integrant.core :as ig]))
(declare handler) (declare sql:delete-files-xlog)
(s/def ::max-age ::dt/duration) (s/def ::max-age ::dt/duration)
(defmethod ig/pre-init-spec ::handler [_] (defmethod ig/pre-init-spec ::handler [_]
(s/keys :req-un [::db/pool ::mtx/metrics ::max-age])) (s/keys :req-un [::db/pool ::max-age]))
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ {:keys [metrics] :as cfg}] [_ {:keys [pool max-age] :as cfg}]
(let [handler #(handler cfg %)] (fn [_]
(->> {:registry (:registry metrics) (db/with-atomic [conn pool]
:type :summary (let [interval (db/interval max-age)
:name "task_file_xlog_gc_timing" result (db/exec-one! conn [sql:delete-files-xlog interval])
:help "file changes garbage collection task timing"} result (:next.jdbc/update-count result)]
(mtx/instrument handler)))) (log/debugf "removed %s rows from file-change table" result)
result))))
(def ^:private (def ^:private
sql:delete-files-xlog sql:delete-files-xlog
"delete from file_change "delete from file_change
where created_at < now() - ?::interval") where created_at < now() - ?::interval")
(defn- handler
[{:keys [pool max-age]} _]
(db/with-atomic [conn pool]
(let [interval (db/interval max-age)
result (db/exec-one! conn [sql:delete-files-xlog interval])
result (:next.jdbc/update-count result)]
(log/infof "removed %s rows from file_change table" result)
nil)))

23
backend/src/app/tasks/sendmail.clj
View file

@ -10,13 +10,12 @@
(ns app.tasks.sendmail (ns app.tasks.sendmail
(:require (:require
[app.config :as cfg] [app.config :as cfg]
[app.metrics :as mtx]
[app.util.emails :as emails] [app.util.emails :as emails]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log] [clojure.tools.logging :as log]
[integrant.core :as ig])) [integrant.core :as ig]))
(declare handler) (declare send-console!)
(s/def ::username ::cfg/smtp-username) (s/def ::username ::cfg/smtp-username)
(s/def ::password ::cfg/smtp-password) (s/def ::password ::cfg/smtp-password)
@ -29,7 +28,7 @@
(s/def ::enabled ::cfg/smtp-enabled) (s/def ::enabled ::cfg/smtp-enabled)
(defmethod ig/pre-init-spec ::handler [_] (defmethod ig/pre-init-spec ::handler [_]
(s/keys :req-un [::enabled ::mtx/metrics] (s/keys :req-un [::enabled]
:opt-un [::username :opt-un [::username
::password ::password
::tls ::tls
@ -40,13 +39,11 @@
::default-reply-to])) ::default-reply-to]))
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ {:keys [metrics] :as cfg}] [_ cfg]
(let [handler #(handler cfg %)] (fn [{:keys [props] :as task}]
(->> {:registry (:registry metrics) (if (:enabled cfg)
:type :summary (emails/send! cfg props)
:name "task_sendmail_timing" (send-console! cfg props))))
:help "sendmail task timing"}
(mtx/instrument handler))))
(defn- send-console! (defn- send-console!
[cfg email] [cfg email]
@ -59,9 +56,3 @@
(println (.toString baos)) (println (.toString baos))
(println "******** end email "(:id email) "**********"))] (println "******** end email "(:id email) "**********"))]
(log/info out)))) (log/info out))))
(defn handler
[cfg {:keys [props] :as task}]
(if (:enabled cfg)
(emails/send! cfg props)
(send-console! cfg props)))

32
backend/src/app/tasks/tasks_gc.clj
View file

@ -5,46 +5,36 @@
;; This Source Code Form is "Incompatible With Secondary Licenses", as ;; This Source Code Form is "Incompatible With Secondary Licenses", as
;; defined by the Mozilla Public License, v. 2.0. ;; defined by the Mozilla Public License, v. 2.0.
;; ;;
;; Copyright (c) 2020 UXBOX Labs SL ;; Copyright (c) 2020-2021 UXBOX Labs SL
(ns app.tasks.tasks-gc (ns app.tasks.tasks-gc
"A maintenance task that performs a cleanup of already executed tasks "A maintenance task that performs a cleanup of already executed tasks
from the database table." from the database table."
(:require (:require
[app.db :as db] [app.db :as db]
[app.metrics :as mtx]
[app.util.time :as dt] [app.util.time :as dt]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log] [clojure.tools.logging :as log]
[integrant.core :as ig])) [integrant.core :as ig]))
(declare handler) (declare sql:delete-completed-tasks)
(s/def ::max-age ::dt/duration) (s/def ::max-age ::dt/duration)
(defmethod ig/pre-init-spec ::handler [_] (defmethod ig/pre-init-spec ::handler [_]
(s/keys :req-un [::db/pool ::mtx/metrics ::max-age])) (s/keys :req-un [::db/pool ::max-age]))
(defmethod ig/init-key ::handler (defmethod ig/init-key ::handler
[_ {:keys [metrics] :as cfg}] [_ {:keys [pool max-age] :as cfg}]
(let [handler #(handler cfg %)] (fn [_]
(->> {:registry (:registry metrics) (db/with-atomic [conn pool]
:type :summary (let [interval (db/interval max-age)
:name "task_tasks_gc_timing" result (db/exec-one! conn [sql:delete-completed-tasks interval])
:help "tasks garbage collection task timing"} result (:next.jdbc/update-count result)]
(mtx/instrument handler)))) (log/debugf "removed %s rows from tasks-completed table" result)
result))))
(def ^:private (def ^:private
sql:delete-completed-tasks sql:delete-completed-tasks
"delete from task_completed "delete from task_completed
where scheduled_at < now() - ?::interval") where scheduled_at < now() - ?::interval")
(defn- handler
[{:keys [pool max-age]} _]
(db/with-atomic [conn pool]
(let [interval (db/interval max-age)
result (db/exec-one! conn [sql:delete-completed-tasks interval])
result (:next.jdbc/update-count result)]
(log/infof "removed %s rows from tasks_completed table" result)
nil)))

4
backend/src/app/telemetry.clj
View file

@ -88,7 +88,7 @@
(catch Exception e (catch Exception e
;; We don't want notify user of a error, just log it for posible ;; We don't want notify user of a error, just log it for posible
;; future investigation. ;; future investigation.
(log/warn e (str "Unexpected error on telemetry:\n" (log/warn e (str "unexpected error on telemetry:\n"
(when-let [edata (ex-data e)] (when-let [edata (ex-data e)]
(str "ex-data: \n" (str "ex-data: \n"
(with-out-str (pprint edata)))) (with-out-str (pprint edata))))
@ -118,4 +118,4 @@
data data
data]))) data])))
(catch Exception e (catch Exception e
(log/errorf e "Error on procesing request.")))) (log/errorf e "error on procesing request"))))

69
backend/src/app/worker.clj
View file

@ -10,6 +10,7 @@
(ns app.worker (ns app.worker
"Async tasks abstraction (impl)." "Async tasks abstraction (impl)."
(:require (:require
[app.common.exceptions :as ex]
[app.common.spec :as us] [app.common.spec :as us]
[app.common.uuid :as uuid] [app.common.uuid :as uuid]
[app.db :as db] [app.db :as db]
@ -19,6 +20,7 @@
[clojure.core.async :as a] [clojure.core.async :as a]
[clojure.spec.alpha :as s] [clojure.spec.alpha :as s]
[clojure.tools.logging :as log] [clojure.tools.logging :as log]
[cuerdas.core :as str]
[integrant.core :as ig] [integrant.core :as ig]
[promesa.exec :as px]) [promesa.exec :as px])
(:import (:import
@ -72,7 +74,7 @@
(s/def ::queue ::us/string) (s/def ::queue ::us/string)
(s/def ::parallelism ::us/integer) (s/def ::parallelism ::us/integer)
(s/def ::batch-size ::us/integer) (s/def ::batch-size ::us/integer)
(s/def ::tasks (s/map-of string? ::us/fn)) (s/def ::tasks (s/map-of string? fn?))
(s/def ::poll-interval ::dt/duration) (s/def ::poll-interval ::dt/duration)
(defmethod ig/pre-init-spec ::worker [_] (defmethod ig/pre-init-spec ::worker [_]
@ -94,7 +96,7 @@
(defmethod ig/init-key ::worker (defmethod ig/init-key ::worker
[_ {:keys [pool poll-interval name queue] :as cfg}] [_ {:keys [pool poll-interval name queue] :as cfg}]
(log/infof "Starting worker '%s' on queue '%s'." name queue) (log/infof "starting worker '%s' on queue '%s'" name queue)
(let [cch (a/chan 1) (let [cch (a/chan 1)
poll-ms (inst-ms poll-interval)] poll-ms (inst-ms poll-interval)]
(a/go-loop [] (a/go-loop []
@ -103,30 +105,30 @@
;; Terminate the loop if close channel is closed or ;; Terminate the loop if close channel is closed or
;; event-loop-fn returns nil. ;; event-loop-fn returns nil.
(or (= port cch) (nil? val)) (or (= port cch) (nil? val))
(log/infof "Stop condition found. Shutdown worker: '%s'" name) (log/infof "stop condition found; shutdown worker: '%s'" name)
(db/pool-closed? pool) (db/pool-closed? pool)
(do (do
(log/info "Worker eventloop is aborted because pool is closed.") (log/info "worker eventloop is aborted because pool is closed")
(a/close! cch)) (a/close! cch))
(and (instance? java.sql.SQLException val) (and (instance? java.sql.SQLException val)
(contains? #{"08003" "08006" "08001" "08004"} (.getSQLState ^java.sql.SQLException val))) (contains? #{"08003" "08006" "08001" "08004"} (.getSQLState ^java.sql.SQLException val)))
(do (do
(log/error "Connection error, trying resume in some instants.") (log/error "connection error, trying resume in some instants")
(a/<! (a/timeout poll-interval)) (a/<! (a/timeout poll-interval))
(recur)) (recur))
(and (instance? java.sql.SQLException val) (and (instance? java.sql.SQLException val)
(= "40001" (.getSQLState ^java.sql.SQLException val))) (= "40001" (.getSQLState ^java.sql.SQLException val)))
(do (do
(log/debug "Serialization failure (retrying in some instants).") (log/debug "serialization failure (retrying in some instants)")
(a/<! (a/timeout poll-ms)) (a/<! (a/timeout poll-ms))
(recur)) (recur))
(instance? Exception val) (instance? Exception val)
(do (do
(log/errorf val "Unexpected error ocurried on polling the database (will resume in some instants).") (log/errorf val "unexpected error ocurried on polling the database (will resume in some instants)")
(a/<! (a/timeout poll-ms)) (a/<! (a/timeout poll-ms))
(recur)) (recur))
@ -202,7 +204,7 @@
(let [task-fn (get tasks name)] (let [task-fn (get tasks name)]
(if task-fn (if task-fn
(task-fn item) (task-fn item)
(log/warn "no task handler found for" (pr-str name))) (log/warnf "no task handler found for '%s'" (pr-str name)))
{:status :completed :task item})) {:status :completed :task item}))
(defn get-error-context (defn get-error-context
@ -227,7 +229,7 @@
(let [cdata (get-error-context error item)] (let [cdata (get-error-context error item)]
(update-thread-context! cdata) (update-thread-context! cdata)
(log/errorf error "Unhandled exception on task (id: %s)" (:id cdata)) (log/errorf error "unhandled exception on task (id: '%s')" (:id cdata))
(if (>= (:retry-num item) (:max-retries item)) (if (>= (:retry-num item) (:max-retries item))
{:status :failed :task item :error error} {:status :failed :task item :error error}
{:status :retry :task item :error error}))))) {:status :retry :task item :error error})))))
@ -235,12 +237,12 @@
(defn- run-task (defn- run-task
[{:keys [tasks]} item] [{:keys [tasks]} item]
(try (try
(log/debugf "Started task '%s/%s/%s'." (:name item) (:id item) (:retry-num item)) (log/debugf "started task '%s/%s/%s'" (:name item) (:id item) (:retry-num item))
(handle-task tasks item) (handle-task tasks item)
(catch Exception e (catch Exception e
(handle-exception e item)) (handle-exception e item))
(finally (finally
(log/debugf "Finished task '%s/%s/%s'." (:name item) (:id item) (:retry-num item))))) (log/debugf "finished task '%s/%s/%s'" (:name item) (:id item) (:retry-num item)))))
(def sql:select-next-tasks (def sql:select-next-tasks
"select * from task as t "select * from task as t
@ -289,21 +291,31 @@
(s/def ::id ::us/string) (s/def ::id ::us/string)
(s/def ::cron dt/cron?) (s/def ::cron dt/cron?)
(s/def ::props (s/nilable map?)) (s/def ::props (s/nilable map?))
(s/def ::task keyword?)
(s/def ::scheduled-task-spec (s/def ::scheduled-task-spec
(s/keys :req-un [::id ::cron ::fn] (s/keys :req-un [::id ::cron ::task]
:opt-un [::props])) :opt-un [::props]))
(s/def ::schedule (s/def ::schedule (s/coll-of (s/nilable ::scheduled-task-spec)))
(s/coll-of (s/nilable ::scheduled-task-spec)))
(defmethod ig/pre-init-spec ::scheduler [_] (defmethod ig/pre-init-spec ::scheduler [_]
(s/keys :req-un [::executor ::db/pool ::schedule])) (s/keys :req-un [::executor ::db/pool ::schedule ::tasks]))
(defmethod ig/init-key ::scheduler (defmethod ig/init-key ::scheduler
[_ {:keys [schedule] :as cfg}] [_ {:keys [schedule tasks] :as cfg}]
(let [scheduler (Executors/newScheduledThreadPool (int 1)) (let [scheduler (Executors/newScheduledThreadPool (int 1))
schedule (filter some? schedule) schedule (->> schedule
(filter some?)
(map (fn [{:keys [task] :as item}]
(let [f (get tasks (name task))]
(when-not f
(ex/raise :type :internal
:code :task-not-found
:hint (str/fmt "task %s not configured" task)))
(-> item
(dissoc :task)
(assoc :fn f))))))
cfg (assoc cfg cfg (assoc cfg
:scheduler scheduler :scheduler scheduler
:schedule schedule)] :schedule schedule)]
@ -330,7 +342,7 @@
(defn- synchronize-schedule-item (defn- synchronize-schedule-item
[conn {:keys [id cron]}] [conn {:keys [id cron]}]
(let [cron (str cron)] (let [cron (str cron)]
(log/debugf "initialize scheduled task '%s' (cron: '%s')." id cron) (log/infof "initialize scheduled task '%s' (cron: '%s')" id cron)
(db/exec-one! conn [sql:upsert-scheduled-task id cron cron]))) (db/exec-one! conn [sql:upsert-scheduled-task id cron cron])))
(defn- synchronize-schedule (defn- synchronize-schedule
@ -351,27 +363,16 @@
(letfn [(run-task [conn] (letfn [(run-task [conn]
(try (try
(when (db/exec-one! conn [sql:lock-scheduled-task id]) (when (db/exec-one! conn [sql:lock-scheduled-task id])
(log/info "Executing scheduled task" id) (log/debugf "executing scheduled task '%s'" id)
((:fn task) task)) ((:fn task) task))
(catch Exception e (catch Throwable e
e))) e)))
(handle-task* [conn]
(let [result (run-task conn)]
(if (instance? Throwable result)
(do
(log/warnf result "unhandled exception on scheduled task '%s'" id)
(db/insert! conn :scheduled-task-history
{:id (uuid/next)
:task-id id
:is-error true
:reason (exception->string result)}))
(db/insert! conn :scheduled-task-history
{:id (uuid/next)
:task-id id}))))
(handle-task [] (handle-task []
(db/with-atomic [conn pool] (db/with-atomic [conn pool]
(handle-task* conn)))] (let [result (run-task conn)]
(when (ex/exception? result)
(log/errorf result "unhandled exception on scheduled task '%s'" id)))))]
(try (try
(px/run! executor handle-task) (px/run! executor handle-task)

4
common/app/common/exceptions.cljc
View file

@ -52,3 +52,7 @@
(defn ex-info? (defn ex-info?
[v] [v]
(instance? #?(:clj clojure.lang.ExceptionInfo :cljs cljs.core.ExceptionInfo) v)) (instance? #?(:clj clojure.lang.ExceptionInfo :cljs cljs.core.ExceptionInfo) v))
(defn exception?
[v]
(instance? #?(:clj java.lang.Throwable :cljs js/Error) v))

1
frontend/src/app/main/data/auth.cljs
View file

@ -90,6 +90,7 @@
ptk/WatchEvent ptk/WatchEvent
(watch [_ state s] (watch [_ state s]
(->> (rp/mutation :logout) (->> (rp/mutation :logout)
(rx/catch (constantly (rx/empty)))
(rx/ignore))) (rx/ignore)))
ptk/EffectEvent ptk/EffectEvent