(PDB-5712) benchmark: use JDK http client instead of puppet's

Some previous investigations have suggested that the puppet client has
performance issues (at least the sync client), and recent pdb
benchmarking has implied something similar -- it could only emulate
about 80k hosts with a 30s run interval, no matter how many senders or
simulation threads it had (on a 60-core, non-hyperthreaded server).

We suspect the puppet client may be adding extra complexity by
(if nothing else) emulating sync via its async client (rather than
just using the Apache sync client directly), and since benchmark only
needs to make synchronous POSTs from existing threads (the senders),
switch to the straightforward JDK client.

Preliminary testing was promising, and after fully implementing this,
benchmark was able to emulate over 140k hosts on the server mentioned
above, using the same randomization percentage of 100.  Guessing that
we might now be hitting a CPU constraint, we changed the randomization
to 10 percent, and were able to emulate at least 200k hosts.

Fixes: puppetlabs#3886

documentation/release_notes_8.markdown

@@ -17,14 +17,22 @@ Release date undetermined, and contributors pending
 
 ### New features and improvements
 
+* The [`benchmark` command][benchmark] should be able to reach notably
+ higher maximum output rates. On one 60 core (non-hyperthreaded)
+ host where previously it could only simulate about 80k nodes with a
+ 30 minute runinterval, it can now simulate over 140k nodes, more if
+ the randomization percentage is reduced from 100.
+ ([GitHub #3886](https://github.com/puppetlabs/puppetdb/issues/3886))
+ (PDB-5712)
+
 * The [`benchmark` command][benchmark] command will now space out the
  factset, catalog, and report for each host more realistically.
  ([GitHub #3880](https://github.com/puppetlabs/puppetdb/pull/3880))
  (PDB-5691)
 
 ### Contributors
 
-Austin Blatt, and ...
+Austin Blatt, Rob Browning, and ...
 
 ## PuppetDB 8.1.1
 

project.clj

@@ -173,6 +173,7 @@
  [puppetlabs/dujour-version-check]
  [puppetlabs/i18n]
  [puppetlabs/kitchensink]
+ [puppetlabs/ssl-utils]
  [puppetlabs/stockpile "0.0.4"]
  [puppetlabs/structured-logging]
  [puppetlabs/trapperkeeper]

src/puppetlabs/puppetdb/client.clj

@@ -1,70 +1,122 @@
 (ns puppetlabs.puppetdb.client
  (:require
+ [clojure.java.io :as io]
  [clojure.tools.logging :as log]
- [puppetlabs.http.client.sync :as http-client]
- [puppetlabs.puppetdb.cheshire :as json]
+ [puppetlabs.http.client.sync :as pclient]
  [puppetlabs.puppetdb.command.constants :refer [command-names]]
+ [puppetlabs.puppetdb.cheshire :as json]
  [puppetlabs.puppetdb.schema :refer [defn-validated]]
  [puppetlabs.puppetdb.time :as t]
  [puppetlabs.puppetdb.utils :as utils]
  [schema.core :as s])
  (:import
- (java.net HttpURLConnection)))
+ (com.puppetlabs.ssl_utils SSLUtils)
+ (java.net HttpURLConnection URI)
+ (java.net.http HttpClient
+ HttpClient$Builder
+ HttpClient$Version
+ HttpRequest
+ HttpRequest$Builder
+ HttpRequest$BodyPublishers
+ HttpResponse$BodyHandlers)
+ (java.util.function Supplier)))
+
+(def ^:private warn-on-reflection-orig *warn-on-reflection*)
+(set! *warn-on-reflection* true)
+
+(defn- ssl-info->context
+ [& {:keys [ssl-cert ssl-key ssl-ca-cert]}]
+ (try
+ ;; REVIEW: looks like this may throw any of these; pass them on for now?
+ ;; KeyStoreException
+ ;; CertificateException
+ ;; NoSuchAlgorithmException
+ ;; KeyManagementException
+ ;; UnrecoverableKeyException
+ ;; NoSuchProviderException
+ (SSLUtils/pemsToSSLContext (io/reader ssl-cert)
+ (io/reader ssl-key)
+ (io/reader ssl-ca-cert))))
+
+(defn- build-http-client [& {:keys [ssl-cert ssl-key ssl-ca-cert] :as opts}]
+ (cond-> (HttpClient/newBuilder)
+ ;; REVIEW: just examples, perhaps remove before we merge
+ ;; true (.version HttpClient$Version/HTTP_1_1)
+ ;; true (.followRedirects HttpClient$Redirect/NORMAL)
+ ssl-cert (.sslContext (ssl-info->context opts))
+ true .build))
+
+;; Until we require requests to provide the client (perhaps we should)
+(def ^:private http-client (memoize build-http-client))
+
+(defn- json-request-generator
+ ([uri] (.uri ^java.net.http.HttpRequest$Builder (json-request-generator) uri))
+ ([] (-> (HttpRequest/newBuilder)
+ ;; REVIEW: just examples, perhaps remove before we merge
+ ;;(.version HttpClient$Version/HTTP_1_1)
+ ;;(.followRedirects HttpClient$Redirect/NORMAL)
+ (.header "Content-Type" "application/json; charset=UTF-8")
+ (.header "Accept" "application/json"))))
+
+;; REVIEW: just examples, perhaps remove before we merge. I think
+;; maybe this is a one-shot because it's required to be, but don't
+;; recall, and we don't currently use it.
+(defn- stream-publisher [s]
+ (let [s-once (atom s)]
+ (-> (reify Supplier
+ (get [this]
+ (let [s @s-once]
+ (reset! s-once nil)
+ s)))
+ HttpRequest$BodyPublishers/ofInputStream)))
+
+(defn- string-publisher [s] (HttpRequest$BodyPublishers/ofString s))
+(defn- string-handler [] (HttpResponse$BodyHandlers/ofString))
+(defn- stream-handler [] (HttpResponse$BodyHandlers/ofInputStream))
+
+(defn- post-body
+ [^HttpClient client
+ ^HttpRequest$Builder req-generator
+ body-publisher
+ response-body-handler]
+ (let [res (.send client (-> req-generator (.POST body-publisher) .build)
+ response-body-handler)]
+ ;; REVIEW: is :status this enough for current purposes?
+ {::jdk-response res
+ :status (.statusCode res)}))
 
 (defn get-metric [base-url metric-name]
  (let [url (str (utils/base-url->str base-url)
  "/mbeans/"
- (java.net.URLEncoder/encode metric-name "UTF-8"))]
- (:body
- (http-client/get url {:throw-exceptions false
- :content-type :json
- :character-encoding "UTF-8"
- :accept :json}))) )
+ (java.net.URLEncoder/encode ^String metric-name "UTF-8"))]
 
-(defn-validated submit-command-via-http!
+ (:body (pclient/get url {:throw-exceptions false
+ :content-type :json
+ :character-encoding "UTF-8"
+ :accept :json}))) )
+
+(defn submit-command-via-http!
  "Submits `payload` as a valid command of type `command` and
  `version` to the PuppetDB instance specified by `host` and
  `port`. The `payload` will be converted to JSON before
  submission. Alternately accepts a command-map object (such as those
  returned by `parse-command`). Returns the server response."
- ([base-url
- certname :- s/Str
- command :- s/Str
- version :- s/Int
- payload]
+ ([base-url certname command version payload]
  (submit-command-via-http! base-url certname command version payload nil {}))
- ([base-url
- certname :- s/Str
- command :- s/Str
- version :- s/Int
- payload
- timeout]
+ ([base-url certname command version payload timeout]
  (submit-command-via-http! base-url certname command version payload timeout {}))
- ([base-url
- certname :- s/Str
- command :- s/Str
- version :- s/Int
- payload :- {s/Any s/Any}
- timeout
- ssl-opts :- {s/Keyword s/Str}]
- (let [body (json/generate-string payload)
- url-params (utils/cmd-url-params {:command command
- :version version
- :certname certname
- :producer-timestamp (-> payload
- :producer_timestamp
- t/to-string)
- :timeout timeout})
- url (str (utils/base-url->str base-url) url-params)
- post-opts (merge {:body body
- :as :text
- :headers {"Content-Type" "application/json"}}
- ; :throw-exceptions false
- ; :content-type :json
- ; :character-encoding "UTF-8"
- ; :accept :json}
- (select-keys ssl-opts [:ssl-cert :ssl-key :ssl-ca-cert]))]
- (http-client/post url post-opts))))
+ ([base-url certname command version payload timeout ssl-opts]
+ (let [stamp (-> payload :producer_timestamp t/to-string)
+ url (str (utils/base-url->str base-url)
+ (utils/cmd-url-params {:command command
+ :version version
+ :certname certname
+ :producer-timestamp stamp
+ :timeout timeout}))]
+ (post-body (http-client (select-keys ssl-opts [:ssl-cert :ssl-key :ssl-ca-cert]))
+ (json-request-generator (URI. url))
+ (-> payload json/generate-string string-publisher)
+ (string-handler)))))
 
 (defn-validated submit-catalog
  "Send the given wire-format `catalog` (associated with `host`) to a
@@ -127,3 +179,5 @@
  ssl-opts)]
  (when-not (= HttpURLConnection/HTTP_OK (:status result))
  (log/error result)))))
+
+(set! *warn-on-reflection* warn-on-reflection-orig)