diff --git a/src/ductile/conn.clj b/src/ductile/conn.clj index 6e91885..63f1e43 100644 --- a/src/ductile/conn.clj +++ b/src/ductile/conn.clj @@ -6,12 +6,11 @@ [ductile.schemas :refer [ConnectParams ESConn]] [schema.core :as s])) -(def default-timeout 30000) - -(defn cm-options [{:keys [timeout]}] - {:timeout timeout - :threads 100 - :default-per-route 100}) +(def default-connection-ttl 60) +(def default-validate-after-inactivity 5000) +(def default-threads 100) +(def default-per-route 100) +(def default-connection-timeout 10000) (def default-opts {:as :json @@ -19,7 +18,7 @@ :throw-exceptions false}) (s/defn make-http-opts :- (s/pred map?) - ([{:keys [cm auth]} :- (s/maybe ESConn) + ([{:keys [cm auth timeouts]} :- (s/maybe ESConn) opts :- (s/pred map?) query-params-keys :- (s/maybe (s/pred coll?)) form-params :- (s/maybe (s/pred map?)) @@ -29,35 +28,81 @@ body (assoc :body body) form-params (assoc :form-params form-params) cm (assoc :connection-manager cm) + (:connection-timeout timeouts) (assoc :connection-timeout (:connection-timeout timeouts)) + (:socket-timeout timeouts) (assoc :socket-timeout (:socket-timeout timeouts)) (seq query-params-keys) (assoc :query-params (select-keys opts query-params-keys)))) ([conn opts query-params-keys] (make-http-opts conn opts query-params-keys nil nil)) ([conn opts] (make-http-opts conn opts [] nil nil)) ([conn] (make-http-opts conn {} [] nil nil))) -(defn make-connection-manager [cm-options] - (make-reusable-conn-manager cm-options)) +(defn make-connection-manager + "Create a reusable connection manager with the given options. + + Options: + - :connection-ttl - how long connections live in the pool in seconds + - :threads - max total connections + - :default-per-route - max connections per route + - :insecure? - allow insecure SSL (self-signed certs) + - :validate-after-inactivity - check idle connections before reuse (in ms) + Helps prevent NoHttpResponseException from stale connections." + [{:keys [connection-ttl validate-after-inactivity] :as opts}] + (let [;; clj-http uses :timeout for TTL + cm-opts (-> opts + (dissoc :connection-ttl :validate-after-inactivity) + (assoc :timeout connection-ttl)) + conn-mgr (make-reusable-conn-manager cm-opts)] + (when (some? validate-after-inactivity) + (.setValidateAfterInactivity conn-mgr (int validate-after-inactivity))) + conn-mgr)) (s/defn connect :- ESConn "Instantiate an ES conn from ConnectParams props. To intercept all ES HTTP requests, set :request-fn to function with the same interface as the 1-argument - arity of `clj-http.client/request`." - [{:keys [protocol host port timeout version engine auth request-fn] + arity of `clj-http.client/request`. + + Connection pool options: + - :connection-ttl - how long connections live in the pool in seconds (default: 60) + - :validate-after-inactivity - check idle connections before reuse, in ms (default: 5000) + Prevents NoHttpResponseException from stale connections closed server-side. + - :threads - max total connections in pool (default: 100) + - :default-per-route - max connections per route (default: 100) + - :insecure? - allow insecure SSL connections, e.g. self-signed certs (default: false) + + Request timeout options (applied to every request): + - :connection-timeout - time to establish TCP connection in ms (default: 10000) + - :socket-timeout - time to wait for data in ms (default: none, for long-running operations)" + [{:keys [protocol host port connection-ttl version engine auth request-fn + validate-after-inactivity threads default-per-route insecure? + connection-timeout socket-timeout] :or {protocol :http request-fn client/request - timeout default-timeout + connection-ttl default-connection-ttl + validate-after-inactivity default-validate-after-inactivity + threads default-threads + default-per-route default-per-route + insecure? false + connection-timeout default-connection-timeout version 7 engine :elasticsearch}} :- ConnectParams] - (let [conn {:cm (make-connection-manager - (cm-options {:timeout timeout})) + (let [timeouts (cond-> {} + connection-timeout (assoc :connection-timeout connection-timeout) + socket-timeout (assoc :socket-timeout socket-timeout)) + conn {:cm (make-connection-manager + {:connection-ttl connection-ttl + :validate-after-inactivity validate-after-inactivity + :threads threads + :default-per-route default-per-route + :insecure? insecure?}) :request-fn request-fn :uri (format "%s://%s:%s" (name protocol) host port) :version version :engine engine}] (cond-> conn - auth (assoc :auth (auth/http-options auth))))) + auth (assoc :auth (auth/http-options auth)) + (seq timeouts) (assoc :timeouts timeouts)))) (s/defn close [conn :- ESConn] (-> conn :cm shutdown-manager)) diff --git a/src/ductile/schemas.clj b/src/ductile/schemas.clj index f53d000..b2524e0 100644 --- a/src/ductile/schemas.clj +++ b/src/ductile/schemas.clj @@ -2,8 +2,7 @@ "All ES related schemas should be defined here" (:require [schema.core :as s] [schema-tools.core :as st]) - (:import [org.apache.http.impl.conn PoolingClientConnectionManager - PoolingHttpClientConnectionManager])) + (:import [org.apache.http.impl.conn PoolingHttpClientConnectionManager])) (s/defschema RequestFn "A function implementing the 1-argument @@ -33,20 +32,29 @@ (s/optional-key :authorization) s/Str (s/optional-key :version) s/Int (s/optional-key :engine) (s/enum :elasticsearch :opensearch) - (s/optional-key :timeout) s/Int + ;; Connection pool options + (s/optional-key :connection-ttl) s/Int + (s/optional-key :validate-after-inactivity) s/Int + (s/optional-key :threads) s/Int + (s/optional-key :default-per-route) s/Int + (s/optional-key :insecure?) s/Bool + ;; Request timeout options + (s/optional-key :connection-timeout) s/Int + (s/optional-key :socket-timeout) s/Int (s/optional-key :auth) AuthParams (s/optional-key :request-fn) RequestFn})) (s/defschema ESConn "an ES conn is a map with a connection manager and an index name" - {:cm (s/either PoolingClientConnectionManager - PoolingHttpClientConnectionManager) + {:cm PoolingHttpClientConnectionManager :uri s/Str :version s/Int :engine (s/enum :elasticsearch :opensearch) :request-fn RequestFn - (s/optional-key :auth) (s/pred map?)}) + (s/optional-key :auth) (s/pred map?) + (s/optional-key :timeouts) {(s/optional-key :connection-timeout) s/Int + (s/optional-key :socket-timeout) s/Int}}) (s/defschema Refresh "ES refresh parameter, see diff --git a/test/ductile/conn_test.clj b/test/ductile/conn_test.clj index ea7968a..cea3779 100644 --- a/test/ductile/conn_test.clj +++ b/test/ductile/conn_test.clj @@ -1,7 +1,10 @@ (ns ductile.conn-test (:require [clojure.test :refer [deftest is testing use-fixtures]] [ductile.conn :as sut] - [schema.test :refer [validate-schemas]])) + [ductile.capabilities :as caps] + [ductile.test-helpers :as helpers] + [schema.test :refer [validate-schemas]]) + (:import [org.apache.http.impl.conn PoolingHttpClientConnectionManager])) (use-fixtures :once validate-schemas) @@ -26,7 +29,8 @@ :param-2 "param-2"} common {:as :json :content-type :json - :throw-exceptions false}] + :throw-exceptions false + :connection-timeout 10000}] (is (= (assoc common :connection-manager cm @@ -66,7 +70,7 @@ :connection-manager cm) (sut/make-http-opts conn-wo-auth opts []) (sut/make-http-opts conn-wo-auth opts nil))) - (is (= (assoc common + (is (= (assoc (dissoc common :connection-timeout) :query-params {:param-1 "param-1"}) (sut/make-http-opts nil opts [:param-1]))))) @@ -134,3 +138,159 @@ "OpenSearch version 2 should be set correctly") (is (= :opensearch (:engine conn)) "Engine should be :opensearch")))) + +(deftest validate-after-inactivity-test + (testing "validate-after-inactivity defaults to 5000ms" + (let [conn (sut/connect {:host "127.0.0.1" + :port 9200}) + ^PoolingHttpClientConnectionManager cm (:cm conn)] + (is (= 5000 (.getValidateAfterInactivity cm)) + "Default validate-after-inactivity should be 5000ms"))) + + (testing "validate-after-inactivity can be customized" + (let [conn (sut/connect {:host "127.0.0.1" + :port 9200 + :validate-after-inactivity 10000}) + ^PoolingHttpClientConnectionManager cm (:cm conn)] + (is (= 10000 (.getValidateAfterInactivity cm)) + "Custom validate-after-inactivity should be set"))) + + (testing "validate-after-inactivity can be disabled with 0" + (let [conn (sut/connect {:host "127.0.0.1" + :port 9200 + :validate-after-inactivity 0}) + ^PoolingHttpClientConnectionManager cm (:cm conn)] + ;; When 0 is passed, validation is effectively disabled + (is (= 0 (.getValidateAfterInactivity cm)) + "Zero validate-after-inactivity disables validation")))) + +(deftest connection-pool-options-test + (testing "pool size defaults" + (let [conn (sut/connect {:host "127.0.0.1" :port 9200}) + ^PoolingHttpClientConnectionManager cm (:cm conn)] + (is (= 100 (.getMaxTotal cm)) + "Default max total connections should be 100") + (is (= 100 (.getDefaultMaxPerRoute cm)) + "Default max per route should be 100"))) + + (testing "pool sizes can be customized" + (let [conn (sut/connect {:host "127.0.0.1" + :port 9200 + :threads 50 + :default-per-route 25}) + ^PoolingHttpClientConnectionManager cm (:cm conn)] + (is (= 50 (.getMaxTotal cm)) + "Custom max total connections") + (is (= 25 (.getDefaultMaxPerRoute cm)) + "Custom max per route"))) + + (testing "insecure? option is accepted" + ;; Just verify it doesn't throw - actual SSL behavior tested in integration + (is (some? (sut/connect {:host "127.0.0.1" + :port 9200 + :insecure? true})) + "Connection with insecure? should succeed"))) + +(deftest request-timeout-options-test + (testing "connection-timeout defaults to 10000ms" + (let [conn (sut/connect {:host "127.0.0.1" :port 9200})] + (is (= 10000 (get-in conn [:timeouts :connection-timeout])) + "Default connection-timeout should be 10000ms") + (is (nil? (get-in conn [:timeouts :socket-timeout])) + "socket-timeout should not be set by default"))) + + (testing "connection-timeout is stored in conn" + (let [conn (sut/connect {:host "127.0.0.1" + :port 9200 + :connection-timeout 5000})] + (is (= 5000 (get-in conn [:timeouts :connection-timeout])) + "connection-timeout should be stored"))) + + (testing "socket-timeout is stored in conn" + (let [conn (sut/connect {:host "127.0.0.1" + :port 9200 + :socket-timeout 10000})] + (is (= 10000 (get-in conn [:timeouts :socket-timeout])) + "socket-timeout should be stored"))) + + (testing "both timeouts can be set together" + (let [conn (sut/connect {:host "127.0.0.1" + :port 9200 + :connection-timeout 5000 + :socket-timeout 10000})] + (is (= {:connection-timeout 5000 :socket-timeout 10000} + (:timeouts conn)) + "Both timeouts should be stored"))) + + (testing "timeouts are applied to http options" + (let [conn (sut/connect {:host "127.0.0.1" + :port 9200 + :connection-timeout 5000 + :socket-timeout 10000}) + http-opts (sut/make-http-opts conn {})] + (is (= 5000 (:connection-timeout http-opts)) + "connection-timeout should be in http opts") + (is (= 10000 (:socket-timeout http-opts)) + "socket-timeout should be in http opts")))) + +(defn- connect-with-validation + "Connect to test ES/OpenSearch with specified validate-after-inactivity setting" + [engine version validate-after-inactivity] + (sut/connect + (cond-> {:host "localhost" + :port (helpers/engine-port engine version) + :version version + :engine engine + :validate-after-inactivity validate-after-inactivity} + (= engine :elasticsearch) (assoc :auth helpers/basic-auth-opts)))) + +(deftest ^:integration validate-after-inactivity-behavior-test + (testing "connection validation allows request to succeed after idle period" + (helpers/for-each-es-version + "requests succeed with validation enabled after idle" + nil + ;; Create a connection with validation enabled (short: 1ms) + (let [conn-with-validation (connect-with-validation engine version 1) + ^PoolingHttpClientConnectionManager cm (:cm conn-with-validation)] + (try + ;; Verify validation setting is applied + (is (= 1 (.getValidateAfterInactivity cm)) + "Validation should be set to 1ms") + + ;; Make initial request to establish a pooled connection + (is (map? (caps/get-cluster-info conn-with-validation)) + "Initial request should succeed") + + ;; Wait longer than validate-after-inactivity to trigger validation + (Thread/sleep 10) + + ;; Make another request - validation will check the connection + (is (map? (caps/get-cluster-info conn-with-validation)) + "Second request should succeed after validation") + + ;; Verify pool is working correctly + (let [stats (.getTotalStats cm)] + (is (>= (.getMax stats) 1) + "Connection pool should be configured")) + (finally + (sut/close conn-with-validation)))))) + + (testing "multiple requests work correctly with connection pooling and validation" + (helpers/for-each-es-version + "sequential requests with validation" + nil + (let [conn (connect-with-validation engine version 100) + ^PoolingHttpClientConnectionManager cm (:cm conn)] + (try + ;; Make multiple requests with delays to exercise pool behavior + (dotimes [_ 5] + (is (map? (caps/get-cluster-info conn)) + "Each request should succeed") + (Thread/sleep 50)) + + ;; Verify no connection leaks + (let [stats (.getTotalStats cm)] + (is (<= (.getLeased stats) 0) + "All connections should be returned to pool after requests")) + (finally + (sut/close conn)))))))