Merge pull request #276 from gojek/feature/EV-37

[Feature/EV 37] Integrate Native Prometheus client in Ziggurat

.github/workflows/build.yml

@@ -2,7 +2,6 @@ name: Ziggurat CI
 
 on:
   push:
-    branches:
     tags:
   pull_request:
     branches:

.gitignore

@@ -17,3 +17,7 @@ pom.xml
 .factorypath
 .project
 .settings
+.clj-kondo/*
+.calva/*
+.lsp/*
+.vscode

project.clj

@@ -53,6 +53,7 @@
                  [ch.qos.logback.contrib/logback-json-classic "0.1.5"]
                  [ch.qos.logback.contrib/logback-jackson "0.1.5"]
                  [net.logstash.logback/logstash-logback-encoder "6.6"]
+                 [clj-commons/iapetos "0.1.9"]
                  [org.apache.commons/commons-pool2 "2.11.1"]]
   :deploy-repositories [["clojars" {:url           "https://clojars.org/repo"
                                     :username      :env/clojars_username

resources/config.test.edn

@@ -94,4 +94,6 @@
             :tracer               {:enabled         [false :bool]
                                    :custom-provider ""}
             :new-relic            {:report-errors false}
+            :prometheus           {:port 8002
+                                   :enabled true}
             :log-format           "text"}}

src/ziggurat/config.clj

@@ -46,6 +46,8 @@
               :http-server          {:middlewares  {:swagger {:enabled false}}
                                      :port         8080
                                      :thread-count 100}
+              :prometheus           {:port 8002
+                                     :enabled true}
               :new-relic            {:report-errors false}
               :log-format           "text"}})
 
@@ -96,6 +98,9 @@
 (defn rabbitmq-config []
   (get (ziggurat-config) :rabbit-mq))
 
+(defn prometheus-config []
+  (get (ziggurat-config) :prometheus))
+
 (defn statsd-config []
   (let [cfg (ziggurat-config)]
     (get cfg :statsd)))

src/ziggurat/init.clj

@@ -147,14 +147,14 @@
   (set-properties-for-structured-logging))
 
 (defn start-common-states []
-  (start* #{#'metrics/statsd-reporter
+  (start* #{#'metrics/metrics-reporter
             #'sentry-reporter
             #'nrepl-server/server}))
 
 (defn stop-common-states []
   (mount/stop #'config/config
               #'sentry-reporter
-              #'metrics/statsd-reporter
+              #'metrics/metrics-reporter
               #'nrepl-server/server))
 
 (defn start

src/ziggurat/metrics.clj

@@ -1,9 +1,10 @@
 (ns ziggurat.metrics
   (:require [clojure.string :as str]
             [clojure.tools.logging :as log]
-            [ziggurat.config :refer [statsd-config ziggurat-config]]
+            [ziggurat.config :refer [statsd-config ziggurat-config prometheus-config]]
             [ziggurat.util.java-util :as util]
             [mount.core :refer [defstate]]
+            [ziggurat.prometheus-exporter :as prometheus-exporter]
             [ziggurat.metrics-interface :as metrics-interface]
             [ziggurat.dropwizard-metrics-wrapper :refer [->DropwizardMetrics]])
   (:gen-class
@@ -33,12 +34,20 @@
   (let [metrics-impl-constructor (get-metrics-implementor-constructor)]
     (reset! metric-impl (metrics-impl-constructor))))
 
-(defstate statsd-reporter
+(def prometheus-enabled?
+  (:enabled (prometheus-config)))
+
+(defstate metrics-reporter
   :start (do (log/info "Initializing Metrics")
-             (initialise-metrics-library)
-             (metrics-interface/initialize @metric-impl (statsd-config)))
+             (if prometheus-enabled?
+               (prometheus-exporter/start-http-server)
+               (do
+                 (initialise-metrics-library)
+                 (metrics-interface/initialize @metric-impl (statsd-config)))))
   :stop (do (log/info "Terminating Metrics")
-            (metrics-interface/terminate @metric-impl)))
+            (if prometheus-enabled?
+              (prometheus-exporter/stop-http-server)
+              (metrics-interface/terminate @metric-impl))))
 
 (defn intercalate-dot
   [names]
@@ -93,7 +102,9 @@
          tags              (remove-topic-tag-for-old-namespace (get-all-tags (get-map additional-tags) metric-namespaces) metric-namespace)
          signed-int-value  (sign (get-int n))]
      (doseq [metric-ns metric-namespaces]
-       (metrics-interface/update-counter @metric-impl metric-ns metric tags signed-int-value)))))
+       (if prometheus-enabled?
+         (prometheus-exporter/update-counter metric-ns metric tags signed-int-value)
+         (metrics-interface/update-counter @metric-impl metric-ns metric tags signed-int-value))))))
 
 (def increment-count (partial inc-or-dec-count +))
 
@@ -112,7 +123,9 @@
          integer-value                  (get-int val)
          metric                          "all"]
      (doseq [metric-ns intercalated-metric-namespaces]
-       (metrics-interface/update-timing @metric-impl metric-ns metric tags integer-value)))))
+       (if prometheus-enabled?
+         (prometheus-exporter/report-histogram metric-ns metric tags integer-value)
+         (metrics-interface/update-timing @metric-impl metric-ns metric tags integer-value))))))
 
 (defn report-time
   "This function is an alias for `report-histogram`.

src/ziggurat/prometheus_exporter.clj

@@ -0,0 +1,45 @@
+(ns ziggurat.prometheus-exporter
+  (:require [iapetos.core :as prometheus]
+            [iapetos.standalone :as standalone]
+            [ziggurat.config :refer [prometheus-config]]))
+
+(def registry
+  (atom (prometheus/collector-registry)))
+
+(defn get-metric-value [kw-metric labels]
+  (prometheus/value (@registry kw-metric labels)))
+
+(defn register-collecter-if-not-exist [collector metric-name label-vec]
+  (let [counter (@registry metric-name {:labels label-vec})]
+    (when-not counter
+      (let [new-counter (collector metric-name {:labels label-vec})
+            new-registry (prometheus/register @registry new-counter)]
+        (reset! registry new-registry)
+        new-counter))))
+
+(defn update-counter [namespace metric tags value]
+  (let [metric-name {:name metric :namespace namespace}
+        label-vec (vec (keys tags))]
+    (register-collecter-if-not-exist prometheus/gauge metric-name label-vec)
+    (prometheus/inc (@registry metric-name tags) value)
+    metric-name))
+
+(defn report-histogram [namespace metric tags integer-value]
+  (let [metric-name {:name metric :namespace namespace}
+        label-vec (vec (keys tags))]
+    (register-collecter-if-not-exist prometheus/histogram metric-name label-vec)
+    (prometheus/observe (@registry metric-name tags) integer-value)))
+
+(defonce httpd (atom nil))
+
+(defn start-http-server []
+  (when-not @httpd
+    (reset! httpd (standalone/metrics-server @registry
+                                             {:port
+                                              (get (prometheus-config)
+                                                   :port 8002)}))))
+
+(defn stop-http-server []
+  (when-let [server @httpd]
+    (.close server)
+    (reset! httpd nil)))

test/ziggurat/fixtures.clj

@@ -61,9 +61,9 @@
       (mount/stop)))
 
 (defn mount-metrics [f]
-  (mount/start (mount/only [#'metrics/statsd-reporter]))
+  (mount/start (mount/only [#'metrics/metrics-reporter]))
   (f)
-  (mount/stop #'metrics/statsd-reporter))
+  (mount/stop #'metrics/metrics-reporter))
 
 (defn mount-config-with-tracer [f]
   (mount-config)

test/ziggurat/metrics_test.clj

@@ -19,7 +19,7 @@
 
 (deftest increment-count-test
   (with-redefs [config (assoc-in config [:ziggurat :metrics :constructor] "ziggurat.util.mock-metrics-implementation/->MockImpl")]
-    (mount/start (mount/only [#'metrics/statsd-reporter]))
+    (mount/start (mount/only [#'metrics/metrics-reporter]))
     (let [passed-metric-name         "metric3"
           expected-topic-entity-name "expected-topic-entity-name"
           input-tags                 {:topic_name expected-topic-entity-name}
@@ -114,11 +114,11 @@
                                                     (reset! increment-count-called? true)))]
             (metrics/-incrementCount metric-namespace passed-metric-name tags)
             (is (true? @increment-count-called?))))))
-    (mount/stop #'metrics/statsd-reporter)))
+    (mount/stop #'metrics/metrics-reporter)))
 
 (deftest decrement-count-test
   (with-redefs [config (assoc-in config [:ziggurat :metrics :constructor] "ziggurat.util.mock-metrics-implementation/->MockImpl")]
-    (mount/start (mount/only [#'metrics/statsd-reporter]))
+    (mount/start (mount/only [#'metrics/metrics-reporter]))
     (let [expected-topic-name "expected-topic-name"
           passed-metric-name  "metric3"
           input-tags          {:topic_name expected-topic-name}
@@ -195,11 +195,11 @@
                                                     (reset! decrement-count-called? true)))]
             (metrics/-decrementCount metric-namespace passed-metric-name tags)
             (is (true? @decrement-count-called?))))))
-    (mount/stop #'metrics/statsd-reporter)))
+    (mount/stop #'metrics/metrics-reporter)))
 
 (deftest report-histogram-test
   (with-redefs [config (assoc-in config [:ziggurat :metrics :constructor] "ziggurat.util.mock-metrics-implementation/->MockImpl")]
-    (mount/start (mount/only [#'metrics/statsd-reporter]))
+    (mount/start (mount/only [#'metrics/metrics-reporter]))
     (let [topic-entity-name "expected-topic-entity-name"
           input-tags        {:topic_name topic-entity-name}
           time-val          10
@@ -265,7 +265,7 @@
                                                      (reset! report-histogram-called? true)))]
             (metrics/-reportTime metric-namespace time-val tags)
             (is (true? @report-histogram-called?))))))
-    (mount/stop #'metrics/statsd-reporter)))
+    (mount/stop #'metrics/metrics-reporter)))
 
 (deftest report-time-test
   (let [metric-namespace "metric-namespace"

test/ziggurat/prometheus_exporter_test.clj

@@ -0,0 +1,83 @@
+(ns ziggurat.prometheus-exporter-test
+  (:require [clojure.test :refer [deftest is testing]]
+            [ziggurat.prometheus-exporter :as prometheus-exporter]
+            [iapetos.core :as prometheus]))
+
+(defn clear-prometheus-registry []
+  (reset! prometheus-exporter/registry (prometheus/collector-registry)))
+
+(deftest test-register-collecter-if-not-exist
+  (testing "Should register gauge if it does not exist"
+    (clear-prometheus-registry)
+    (let [tags {:actor-topic "food" :app_name "delivery"}
+          metric-name {:name "test-gauge" :namespace "bar"}
+          label-vec (vec (keys tags))]
+      (is (nil? (@prometheus-exporter/registry metric-name)) "Gauge should not exist initially")
+      (prometheus-exporter/register-collecter-if-not-exist prometheus/gauge metric-name label-vec)
+      (is (not (nil? (@prometheus-exporter/registry metric-name))) "Gauge should be registered in the registry"))
+
+    (testing "Gauge not registered if it exists"
+      (clear-prometheus-registry)
+      (let [tags {:actor-topic "food" :app_name "delivery"}
+            metric-name "test-gauge"
+            label-vec (vec (keys tags))
+            existing-gauge (prometheus-exporter/register-collecter-if-not-exist prometheus/gauge metric-name label-vec)]
+        (let [expected-nil (prometheus-exporter/register-collecter-if-not-exist prometheus/gauge metric-name label-vec)]
+          (is (not (nil? existing-gauge)))
+          (is (nil? expected-nil) "Should retrieve the existing gauge"))))))
+
+(deftest test-update-counter
+  (testing "Updating counter can increments its value"
+    (clear-prometheus-registry)
+    (let [namespace "test_namespace"
+          metric "test_metric"
+          old_value 5.0
+          increment 2
+          new-value (+ old_value increment)
+          tags {:some "label"}
+          _ (prometheus-exporter/update-counter namespace metric tags old_value)]
+      (prometheus-exporter/update-counter namespace metric tags increment)
+      (@prometheus-exporter/registry (assoc tags :name metric :namespace namespace))
+      (let [gauge-value (prometheus-exporter/get-metric-value (keyword (str namespace "/" metric)) tags)]
+        (is (= new-value gauge-value) "Gauge value should be incremented"))))
+
+  (testing "Updating counter can decrement its value"
+    (clear-prometheus-registry)
+    (let [namespace "test_namespace"
+          metric "test_metric"
+          old_value 5.0
+          decrement -2
+          new-value (+ old_value decrement)
+          tags {:some "label"}
+          _ (prometheus-exporter/update-counter namespace metric tags old_value)]
+      (prometheus-exporter/update-counter namespace metric tags decrement)
+      (@prometheus-exporter/registry (assoc tags :name metric :namespace namespace))
+      (let [gauge-value (prometheus-exporter/get-metric-value (keyword (str namespace "/" metric)) tags)]
+        (is (= new-value gauge-value) "Gauge value should be incremented")))))
+
+(deftest test-report-histogram
+  (testing "should register histogram when it does not exist"
+    (clear-prometheus-registry)
+    (let [namespace "test_namespace"
+          metric "test_metric"
+          value 5.0
+          tags {:some "label"}
+          label (assoc tags :name metric :namespace namespace)]
+      (is (nil? (@prometheus-exporter/registry label)))
+      (prometheus-exporter/report-histogram namespace metric tags value)
+      (is (not (nil? (@prometheus-exporter/registry label))))
+      (let [gauge-value (prometheus-exporter/get-metric-value (keyword (str namespace "/" metric)) tags)]
+        (is (= (gauge-value :sum) 5.0))
+        (is (= (gauge-value :count) 1.0)))))
+  (testing "should update histogram when it exist and increment the count"
+    (clear-prometheus-registry)
+    (let [namespace "test_namespace"
+          metric "test_metric"
+          value 5.0
+          tags {:some "label"}
+          label (assoc tags :name metric :namespace namespace)
+          _ (prometheus-exporter/report-histogram namespace metric tags value)]
+      (is (not (nil? (@prometheus-exporter/registry label))))
+      (prometheus-exporter/report-histogram namespace metric tags value)
+      (let [gauge-value (prometheus-exporter/get-metric-value (keyword (str namespace "/" metric)) tags)]
+        (is (= (gauge-value :count) 2.0))))))