Add optional worker metrics

Signed-off-by: Greg Haskins <greg@manetu.com>

Author: ghaskins

project.clj

@@ -16,6 +16,10 @@
                  [org.slf4j/jul-to-slf4j "2.0.16"]
                  [org.slf4j/jcl-over-slf4j "2.0.16"]
                  [org.slf4j/log4j-over-slf4j "2.0.16"]
+                 [clj-commons/iapetos "0.1.14"]
+                 [io.prometheus/simpleclient_hotspot "0.16.0"]
+                 [com.uber.m3/tally-core "0.14.2"]
+                 [com.uber.m3/tally-prometheus "0.14.2"]
                  [progrock "0.1.2"]
                  [doric "0.9.0"]
                  [kixi/stats "0.5.7"]

src/manetu/temporal_benchmark/commands/worker.clj

@@ -8,6 +8,7 @@
             [temporal.activity :refer [defactivity] :as a]
             [temporal.promise :as tp]
             [temporal.client.worker :as worker]
+            [manetu.temporal-benchmark.metrics :as metrics]
             [manetu.temporal-benchmark.utils :refer [exec-command]]))
 
 (def command "worker")
@@ -49,28 +50,38 @@
     :default 1000
     :parse-fn parse-long
     :validate [pos? "Must be a positive integer"]]
+   [nil "--[no-]metrics-enabled" "Enable metrics endpoint (see --metrics-port)"
+    :default false]
+   [nil "--metrics-port NUM" "The HTTP port for metrics, when enabled"
+    :default 8080
+    :parse-fn parse-long
+    :validate [pos? "Must be a positive integer"]]
    [nil "--[no-]using-virtual-workflow-threads" "Use Virtual Threads for workflow threads (requires JDK 21+)"
     :default p.exec/vthreads-supported?]])
 
-(defn start [options client]
-  (worker/start client
-                (-> (select-keys options [:temporal-taskqueue
-                                          :max-concurrent-activity-task-pollers
-                                          :max-concurrent-activity-execution-size
-                                          :max-concurrent-workflow-task-pollers
-                                          :max-concurrent-workflow-task-execution-size])
-                    (rename-keys {:temporal-taskqueue :task-queue}))
-                (select-keys options [:max-workflow-thread-count
-                                      :workflow-cache-size
-                                      :using-virtual-workflow-threads])))
+(defn start [{:keys [metrics-enabled metrics-port] :as options} client]
+  (let [mctx (when metrics-enabled (metrics/start metrics-port))
+        wctx (worker/start client
+                           (-> (select-keys options [:temporal-taskqueue
+                                                     :max-concurrent-activity-task-pollers
+                                                     :max-concurrent-activity-execution-size
+                                                     :max-concurrent-workflow-task-pollers
+                                                     :max-concurrent-workflow-task-execution-size])
+                               (rename-keys {:temporal-taskqueue :task-queue}))
+                           (select-keys options [:max-workflow-thread-count
+                                                 :workflow-cache-size
+                                                 :using-virtual-workflow-threads]))]
+    {:mctx mctx :wctx wctx}))
 
-(def stop worker/stop)
+(defn stop [{:keys [mctx wctx]}]
+  (some-> mctx metrics/stop)
+  (some-> wctx worker/stop))
 
 (defn exec [options client]
-  (let [w (start options client)]
+  (let [ctx (start options client)]
     (println "Worker running.  Press CNTL-C to exit")
     (deref (promise))
-    (worker/stop w)))
+    (stop ctx)))
 
 (def spec {:description description
            :fn (partial exec-command {:command        command

src/manetu/temporal_benchmark/metrics.clj

@@ -0,0 +1,27 @@
+; Copyright © Manetu, Inc.  All rights reserved
+
+(ns manetu.temporal-benchmark.metrics
+  (:require [iapetos.core :as prometheus]
+            [iapetos.standalone :as standalone])
+  (:import [com.uber.m3.tally.experimental.prometheus PrometheusReporter PrometheusReporter$Builder TimerType]
+           [io.prometheus.client CollectorRegistry]
+           [com.uber.m3.tally RootScopeBuilder]
+           [com.uber.m3.util Duration]))
+
+(defn create-reporter
+  ^PrometheusReporter []
+  (-> (PrometheusReporter$Builder.)
+      (.timerType TimerType/HISTOGRAM)
+      (.registry CollectorRegistry/defaultRegistry)
+      (.build)))
+
+(def root-scope
+  (delay (-> (RootScopeBuilder.)
+             (.reporter (create-reporter))
+             (.reportEvery (Duration/ofSeconds 1)))))
+
+(defn start [port]
+  (standalone/metrics-server prometheus/default-registry {:port port}))
+
+(defn stop [ctx]
+  (.close ctx))

src/manetu/temporal_benchmark/utils.clj

@@ -5,7 +5,8 @@
             [clojure.tools.cli :refer [parse-opts]]
             [slingshot.slingshot :refer [throw+ try+]]
             [temporal.client.core :as c]
-            [temporal.tls :as tls]))
+            [temporal.tls :as tls]
+            [manetu.temporal-benchmark.metrics :as metrics]))
 
 (def tool-name "temporal-benchmark")
 
@@ -49,7 +50,8 @@
 
 (defn create-client [{:keys [temporal-target temporal-namespace tls] :as global-options}]
   (c/create-client (-> {:target temporal-target
-                        :namespace temporal-namespace}
+                        :namespace temporal-namespace
+                        :metrics-scope @metrics/root-scope}
                        (cond-> tls (assoc :ssl-context (new-ssl-context global-options))))))
 
 (defn exec-command