Extend benchmark tool (replikativ#259)

* Extend benchmark tool

* Switch to deps

* Fix dependency format

* Revert to original alignment

* Make benchmark configurations unique

* Reset config values

benchmark/src/benchmark/config.clj

@@ -28,12 +28,10 @@
    {:s1 (format "%15d" (rand-int max-int))
     :i1 (rand-int max-int)})
 
-
 (defn q1 [string-val]
   (conj '[:find ?e :where]
         (conj '[?e :s1] string-val)))
 
 (defn q2 [int-val]
   (conj '[:find ?a :where [?e :s1 ?a]]
         (conj '[?e :i1] int-val)))
-

benchmark/src/benchmark/core.clj

@@ -1,20 +1,72 @@
 (ns benchmark.core
-  (:require [benchmark.measure :as m]
+  (:require [clojure.tools.cli :as cli]
+            [benchmark.measure :as m]
             [benchmark.config :as c]
-            [clojure.pprint :as pp]))
-
-
-(defn -main [& _]
-  (let [measurements (vec (for [config       c/db-configs
-                                initial-size c/initial-datoms
-                                n            c/datom-counts
-                                _            (range c/iterations)]
-                              (m/measure-performance-full initial-size n config)))]
-    (->> measurements
-         (apply concat)
-         (group-by :context)
-         (mapv (fn [[context group]]
-                 {:context context
-                  :mean-time (/ (reduce (fn [x y] (+ x (:time y))) 0 group)
-                                (count group))}))
-         pp/pprint)))
+            [benchmark.store :refer [transact-missing-schema transact-results ->RemoteDB]]
+            [clojure.pprint :refer [pprint]]))
+
+
+(def cli-options
+  [["-u" "--db-server-url URL" "Base URL for datahike server, e.g. http://localhost:3000"
+    :default nil]
+   ["-n" "--db-name DBNAME" "Database name for datahike server" :default nil]
+   ["-g" "--db-token TOKEN" "Token for datahike server" :default nil]
+   ["-t" "--tag TAG" "Add tag to measurements"
+    :default #{}
+    :assoc-fn (fn [m k v] (assoc m k (conj (get m k) v)))]
+
+   ["-h" "--help"]])
+
+(defn print-usage-info [summary]
+  (println (str "Usage: clj -M:benchmark [options] \n\n  Options:\n" summary)))
+
+(defn full-server-description? [server-description]
+  (every? #(not (nil? %)) server-description))
+
+(defn partial-server-description? [server-description]
+  (and (some (comp not nil?) server-description)
+       (not (full-server-description? server-description))))
+
+(defn -main [& args]
+  (let [{:keys [options errors summary]} (cli/parse-opts args cli-options)
+        server-info-keys [:db-server-url :db-token :db-name]
+        server-description (map options server-info-keys)
+        tags (:tag options)]
+
+    (cond
+      (some? errors)
+      (do (println "Errors:" errors)
+          (print-usage-info summary))
+
+      (:help options)
+      (print-usage-info summary)
+
+      (partial-server-description? server-description)
+      (do (println (str "Only partial information for remote connection has been given: "
+                        (select-keys options server-info-keys)))
+          (println "Please, define URL, database name, and token to save the data on a remote datahike server.")
+          (print-usage-info summary))
+
+      :else
+      (let [measurements (vec (for [config c/db-configs
+                                    initial-size c/initial-datoms
+                                    n c/datom-counts
+                                    _ (range c/iterations)]
+                                (m/measure-performance-full initial-size n config)))
+            processed (->> measurements
+                           (apply concat)
+                           (group-by :context)
+                           (map (fn [[context group]]
+                                  (assoc context :mean-time (/ (reduce (fn [x y] (+ x (:time y))) 0 group)
+                                                               (count group))))))
+            tagged (if (empty? tags)
+                     (vec processed)
+                     (mapv (fn [entity] (assoc entity :tag (vec tags))) processed))]
+        (if (full-server-description? server-description)
+          (let [rdb (apply ->RemoteDB server-description)]
+            (println "Database used:" rdb)
+            (transact-missing-schema rdb)
+            (transact-results rdb tagged))
+          (pprint tagged)))))
+
+  (shutdown-agents))

benchmark/src/benchmark/measure.clj

@@ -41,10 +41,13 @@
         rand-i-val (rand-nth (mapv :i1 tx))
         t-query2-n   (:t (timed (d/q (c/q2 rand-i-val) @conn)))
 
-        final-size (+ initial-size n-datoms)]
+        final-size (+ initial-size n-datoms)
+        simple-config (-> config
+                          (dissoc :store)
+                          (assoc :dh-backend (get-in config [:store :backend])))]
     (d/release conn)
-    [{:time t-connection-0  :context {:db config :function :connection  :db-size initial-size}}
-     {:time t-transaction-n :context {:db config :function :transaction :db-size initial-size :tx-size n-datoms}}
-     {:time t-connection-n  :context {:db config :function :connection  :db-size final-size}}
-     {:time t-query1-n      :context {:db config :function :query1      :db-size final-size}}
-     {:time t-query2-n      :context {:db config :function :query2      :db-size final-size}}]))
+    [{:time t-connection-0  :context {:db-config simple-config :function :connection  :db-size initial-size}}
+     {:time t-transaction-n :context {:db-config simple-config :function :transaction :db-size initial-size :tx-size n-datoms}}
+     {:time t-connection-n  :context {:db-config simple-config :function :connection  :db-size final-size}}
+     {:time t-query1-n      :context {:db-config simple-config :function :query1      :db-size final-size}}
+     {:time t-query2-n      :context {:db-config simple-config :function :query2      :db-size final-size}}]))

benchmark/src/benchmark/store.clj

@@ -0,0 +1,104 @@
+(ns benchmark.store
+  (:require [clojure.edn :as edn]
+            [clj-http.client :as client]))
+
+(defrecord RemoteDB [baseurl token dbname])
+
+(def schema
+  [{:db/ident :db-config
+    :db/valueType :db.type/ref
+    :db/cardinality :db.cardinality/one}
+   {:db/ident :mean-time
+    :db/valueType :db.type/double
+    :db/cardinality :db.cardinality/one}
+   {:db/ident :function
+    :db/valueType :db.type/keyword
+    :db/cardinality :db.cardinality/one}
+   {:db/ident :db-size
+    :db/valueType :db.type/long
+    :db/cardinality :db.cardinality/one}
+   {:db/ident :tx-size
+    :db/valueType :db.type/long
+    :db/cardinality :db.cardinality/one}
+   {:db/ident :tag                                          ;; for branch identifier
+    :db/valueType :db.type/string
+    :db/cardinality :db.cardinality/many}
+   {:db/ident :index
+    :db/valueType :db.type/keyword
+    :db/cardinality :db.cardinality/one}
+   {:db/ident :keep-history?
+    :db/valueType :db.type/boolean
+    :db/cardinality :db.cardinality/one}
+   {:db/ident :schema-flexibility
+    :db/valueType :db.type/keyword
+    :db/cardinality :db.cardinality/one}
+   {:db/ident :dh-backend
+    :db/valueType :db.type/keyword
+    :db/cardinality :db.cardinality/one}])
+
+
+(defn parse-body [{:keys [body] :as response}]
+  (if-not (empty? body)
+    (edn/read-string body)
+    ""))
+
+(defn db-request
+  ([db method route] (db-request db method route nil))
+  ([db method route body]
+   (-> (client/request (merge {:url (str (:baseurl db) "/" route)
+                               :method method
+                               :content-type "application/edn"
+                               :accept "application/edn"}
+                              (when (or (= method :post) body)
+                                {:body (str body)})
+                              {:headers {:authorization (str "token " (:token db))
+                                         :db-name (:dbname db)}}))
+       parse-body)))
+
+(defn transact-data [db tx-data]
+  (db-request db :post "transact" {:tx-data tx-data}))
+
+(defn list-databases [db]
+  (db-request db :get "databases"))
+
+(defn get-datoms [db]
+  (db-request db :post "datoms" {:index :eavt}))
+
+(defn request-data [db q]
+  (let [query (if (map? q) q {:query q})]
+    (db-request db :post "q" query)))
+
+(defn get-schema [db]
+  (db-request db :get "schema"))
+
+(defn db-config-eid
+  "Get existing entity ID for database configuration or transact config and get ID from new entry"
+  [db {:keys [dh-backend index keep-history? schema-flexibility] :as db-config}]
+  (let [query {:query '[:find ?e
+                        :in $ ?b ?i ?h ?s
+                        :where
+                        [?e :dh-backend ?b]
+                        [?e :index ?i]
+                        [?e :keep-history? ?h]
+                        [?e :schema-flexibility ?s]]
+               :args [dh-backend index keep-history? schema-flexibility]}
+        existing-eid (ffirst (request-data db query))
+        eid (if (nil? existing-eid)
+              (first (second (:tx-data (transact-data db [db-config])))) ;; get new eid
+              existing-eid)]
+    eid))
+
+(defn transact-results [db results]
+  (let [config-mapping (memoize db-config-eid)
+        tx-data (time (map #(update % :db-config (partial config-mapping db))
+                           results))]
+    (transact-data db tx-data)))
+
+(defn transact-missing-schema [db]
+  (let [current-schema (get-schema db)
+        defined-attribs (vals current-schema)
+        missing-schema (filterv (fn [entity] (not-any? #(= % (:db/ident entity))
+                                                       defined-attribs))
+                                schema)]
+    (when (not-empty missing-schema)
+      (transact-data db missing-schema))))

deps.edn

@@ -16,10 +16,12 @@
 
            :1.10 {:override-deps {org.clojure/clojure {:mvn/version "1.10.0"}}}
 
-           :dev {:extra-paths ["dev"]
+           :dev {:extra-paths ["dev" "benchmark/src"]
                  :extra-deps {org.clojure/tools.namespace {:mvn/version "0.2.11"}
                               cider/cider-nrepl           {:mvn/version "0.19.0"}
-                              nrepl/nrepl                 {:mvn/version "0.5.3"}}}
+                              nrepl/nrepl                 {:mvn/version "0.5.3"}
+                              clj-http/clj-http           {:mvn/version "3.11.0"}
+                              org.clojure/tools.cli       {:mvn/version "1.0.194"}}}
 
            :test {:extra-paths ["test"]
                   :extra-deps {org.clojure/clojurescript {:mvn/version "1.10.516"}
@@ -34,9 +36,11 @@
                   :main-opts ["-m" "nrepl.cmdline" "--middleware" "[cider.nrepl/cider-middleware]"]}
 
            :benchmark {:main-opts ["-m" "benchmark.core"]
-                       :extra-paths ["benchmark/src" "src"]}
+                       :extra-paths ["benchmark/src"]
+                       :extra-deps {clj-http/clj-http     {:mvn/version "3.11.0"}
+                                    org.clojure/tools.cli {:mvn/version "1.0.194"}}}
 
-           :datomic {:extra-deps {com.datomic/datomic-free {:mvn/version  "0.9.5703"}}}
+           :datomic {:extra-deps {com.datomic/datomic-free {:mvn/version "0.9.5703"}}}
 
            :deploy {:extra-deps {deps-deploy/deps-deploy {:mvn/version "0.0.9"}}
                     :main-opts ["-m" "deps-deploy.deps-deploy" "deploy" "replikativ-datahike.jar"]}

dev/sandbox_benchmarks.clj

@@ -0,0 +1,27 @@
+(ns sandbox-benchmarks
+  (:require [benchmark.core :as b]
+            [benchmark.store :as s]))
+
+;(b/-main)                                                   ;; TIMBRE_LEVEL=':fatal' clj -M:benchmark
+
+;(b/-main "-t" "test-id")                                    ;; TIMBRE_LEVEL=':fatal' clj -M:benchmark  -t test-id
+
+;(b/-main "-t" "test-id" "-t" "test-id2")                    ;; TIMBRE_LEVEL=':fatal' clj -M:benchmark -t test-id -t test-id2
+
+;(b/-main "-t" "test-id" "-u" "http://localhost:3001" "-n" "benchmarks" "-g" "test-token") ;; TIMBRE_LEVEL=':fatal' clj -M:benchmark -t test-id -u http://localhost:3001 -n benchmarks -g test-token
+
+
+;; docker run -d --name datahike-server -p 3001:3000 -e DATAHIKE_SERVER_TOKEN=test-token -e DATAHIKE_SCHEMA_FLEXIBILITY=write -e DATAHIKE_STORE_BACKEND=file -e DATAHIKE_NAME=benchmarks -e DATAHIKE_STORE_PATH=/opt/datahike-server/benchmarks replikativ/datahike-server:snapshot
+
+;(def db (s/->RemoteDB "http://localhost:3001" "test-token" "benchmarks"))
+
+;(s/list-databases db)
+
+;(s/transact-missing-schema db)
+;(s/get-datoms db)
+;(s/request-data db '[:find ?e ?a ?v :where [?e ?a ?v]])
+;(s/request-data db '[:find ?e ?a ?v :where [1 ?a ?v]])      ;; -> error status 500
+;(s/request-data db '[:find ?e ?a ?v :where [?e :db ?v]])
+;(s/request-data db '[:find ?e ?v ?t :where [?e :dh-backend ?v ?t]])
+;(s/get-schema db)
+