Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Increase performance of Elasticsearch bulk forwarder. #1033

Merged
merged 1 commit into from
Dec 27, 2022
Merged

Increase performance of Elasticsearch bulk forwarder. #1033

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
21 changes: 11 additions & 10 deletions src/riemann/elasticsearch.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,8 @@
(:require [clj-http.client :as http]
[cheshire.core :as json]
[clj-time.coerce :as time-coerce]
[clj-time.format :as time-format]))
[clj-time.format :as time-format]
[clojure.string :as string]))

(defn- datetime-from-event
"Returns the datetime from event correcting (secs -> millisecs) before conversion."
Expand Down Expand Up @@ -90,19 +91,19 @@

(defn gen-request-bulk-body-reduce
"Reduction fn used in `gen-request-bulk-body` to generate the body request"
[result elem]
(str
result
;;action and metadata
(json/generate-string {(:es-action elem) (:es-metadata elem)}) "\n"
;; source (optional)
(when (:es-source elem)
(str (json/generate-string (:es-source elem)) "\n"))))
[elem]
(concat
[;;action and metadata
(json/generate-string {(:es-action elem) (:es-metadata elem)})]
;; source (optional)
(when (:es-source elem)
[(json/generate-string (:es-source elem))])))

(defn gen-request-bulk-body
"Takes a list of events, generates the body request for Elasticsearch"
[events]
(reduce gen-request-bulk-body-reduce "" events))
(when (not-empty events)
(str (string/join "\n" (mapcat gen-request-bulk-body-reduce events)) "\n")))

(defn default-bulk-formatter
"Returns a function which accepts an event and formats it for the Elasticsearch bulk API.
Expand Down
98 changes: 49 additions & 49 deletions test/riemann/elasticsearch_test.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,50 +115,50 @@
:throw-entire-message? true}]))))))

(deftest ^:elasticsearch gen-request-bulk-body-reduce-test
(is (= "{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n"
(gen-request-bulk-body-reduce "" {:es-action "index"
:es-metadata {:_index "test"
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}})))
(is (= "{\"delete\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"2\"}}\n"
(gen-request-bulk-body-reduce "" {:es-action "delete"
:es-metadata {:_index "test"
:_type "type1"
:_id "2"}}))))
(is (= ["{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}" "{\"field1\":\"value1\"}"]
(gen-request-bulk-body-reduce {:es-action "index"
:es-metadata {:_index "test"
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}})))
(is (= ["{\"delete\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"2\"}}"]
(gen-request-bulk-body-reduce {:es-action "delete"
:es-metadata {:_index "test"
:_type "type1"
:_id "2"}}))))

(deftest ^:elasticsearch gen-request-bulk-body-test
(is (= "{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n"
(gen-request-bulk-body [{:es-action "index"
:es-metadata {:_index "test"
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}}])))
:es-metadata {:_index "test"
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}}])))
(is (= "{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n{\"delete\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"2\"}}\n"
(gen-request-bulk-body [{:es-action "index"
:es-metadata {:_index "test"
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}}
{:es-action "delete"
:es-metadata {:_index "test"
:_type "type1"
:_id "2"}}])))
(is (= "{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n{\"delete\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"2\"}}\n{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n"
:es-metadata {:_index "test"
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}}
{:es-action "delete"
:es-metadata {:_index "test"
:_type "type1"
:_id "2"}}])))
(is (= "{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n{\"delete\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"2\"}}\n{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n"
(gen-request-bulk-body [{:es-action "index"
:es-metadata {:_index "test"
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}}
{:es-action "delete"
:es-metadata {:_index "test"
:_type "type1"
:_id "2"}}
{:es-action "index"
:es-metadata {:_index "test"
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}}]))))
:es-metadata {:_index "test"
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}}
{:es-action "delete"
:es-metadata {:_index "test"
:_type "type1"
:_id "2"}}
{:es-action "index"
:es-metadata {:_index "test"
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}}]))))

(deftest ^:elasticsearch elasticsearch-bulk-test
(with-mock [calls clj-http.client/post]
Expand All @@ -169,13 +169,13 @@
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}})
(is (= (last @calls)
["http://127.0.0.1:9200/_bulk"
(is (= ["http://127.0.0.1:9200/_bulk"
{:body "{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n"
:content-type "application/x-ndjson"
:conn-timeout 5000
:socket-timeout 5000
:throw-entire-message? true}]))
:throw-entire-message? true}]
(last @calls)))
(elastic [{:es-action "index"
:es-metadata {:_index "test"
:_type "type1"
Expand All @@ -190,13 +190,13 @@
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}}])
(is (= (last @calls)
["http://127.0.0.1:9200/_bulk"
(is (= ["http://127.0.0.1:9200/_bulk"
{:body "{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n{\"delete\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"2\"}}\n{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n"
:content-type "application/x-ndjson"
:conn-timeout 5000
:socket-timeout 5000
:throw-entire-message? true}]))))
:throw-entire-message? true}]
(last @calls)))))
(testing "with auth"
(let [elastic (elasticsearch-bulk {:username "elastic"
:password "changeme"})]
Expand All @@ -205,14 +205,14 @@
:_type "type1"
:_id "1"}
:es-source {:field1 "value1"}})
(is (= (last @calls)
["http://127.0.0.1:9200/_bulk"
(is (= ["http://127.0.0.1:9200/_bulk"
{:body "{\"index\":{\"_index\":\"test\",\"_type\":\"type1\",\"_id\":\"1\"}}\n{\"field1\":\"value1\"}\n"
:content-type "application/x-ndjson"
:conn-timeout 5000
:socket-timeout 5000
:basic-auth ["elastic" "changeme"]
:throw-entire-message? true}]))))
:throw-entire-message? true}]
(last @calls)))))
(testing "with formatter"
(let [formatter (default-bulk-formatter {:es-index "riemann"
:type "foo"
Expand All @@ -229,14 +229,14 @@
:tags ["t1"]
:es-id "3"
:ttl 30})
(is (= (last @calls)
["http://127.0.0.1:9300/_bulk"
(is (= ["http://127.0.0.1:9300/_bulk"
{:body "{\"index\":{\"_index\":\"riemann-2017.04.07\",\"_type\":\"foo\",\"_id\":\"3\"}}\n{\"host\":\"foo\",\"metric\":10,\"tags\":[\"t1\"],\"ttl\":30,\"@timestamp\":\"2017-04-07T12:16:22.000Z\"}\n"
:content-type "application/x-ndjson"
:conn-timeout 1000
:socket-timeout 5000
:basic-auth ["elastic" "changeme"]
:throw-entire-message? true}]))))))
:throw-entire-message? true}]
(last @calls)))))))

(deftest ^:elasticsearch default-bulk-formatter-test
(let [formatter (default-bulk-formatter {:es-index "riemann"
Expand Down