From ece70bea1087ec12ba7d98cd83e4ca3ad4bafa86 Mon Sep 17 00:00:00 2001 From: Guillaume Buisson Date: Wed, 10 Dec 2025 22:23:21 +0100 Subject: [PATCH] Add update-policy! and auto-update on conflict for OpenSearch ISM - Add get-policy-raw function to retrieve full API response including _seq_no and _primary_term metadata needed for OpenSearch ISM updates - Add update-policy! function that handles optimistic locking for OpenSearch ISM (requires seq_no and primary_term query params) - Modify create-policy! to catch 409 Conflict errors and automatically retry with update-policy!, making policy creation idempotent - Add integration tests for policy update functionality OpenSearch ISM requires version control parameters for updates, unlike Elasticsearch ILM which allows simple PUT overwrites. This change makes the API consistent across both engines. --- src/ductile/lifecycle.clj | 133 +++++++++++++++++++++++++++++------- test/ductile/index_test.clj | 64 +++++++++++++++++ 2 files changed, 171 insertions(+), 26 deletions(-) diff --git a/src/ductile/lifecycle.clj b/src/ductile/lifecycle.clj index eea78dd..3c3f281 100644 --- a/src/ductile/lifecycle.clj +++ b/src/ductile/lifecycle.clj @@ -277,13 +277,90 @@ ;; Backward compatibility: default to Elasticsearch ILM (policy-uri uri policy-name :elasticsearch))) +(s/defn get-policy-raw + "Get a lifecycle management policy with full response metadata. + Returns the raw response including _seq_no and _primary_term for OpenSearch. + Use this when you need version info for updates." + [{:keys [uri version engine request-fn] :as conn} :- ESConn + policy-name :- s/Str] + (when-not (feat/lifecycle-management-type conn) + (throw (ex-info "Lifecycle management not supported" + {:engine engine :version version}))) + (-> (make-http-opts conn) + (assoc :method :get + :url (policy-uri uri policy-name engine)) + request-fn + safe-es-read)) + +(s/defn update-policy! + "Update an existing lifecycle management policy. + - For Elasticsearch: Updates an ILM policy (simple PUT, no version control needed) + - For OpenSearch: Updates an ISM policy using seq_no and primary_term for optimistic locking + + The policy parameter should be in ILM format. It will be automatically + transformed to ISM format if connecting to OpenSearch. + + For OpenSearch, this function first retrieves the current policy to get + _seq_no and _primary_term, then performs a conditional update." + [{:keys [uri version engine request-fn] :as conn} :- ESConn + policy-name :- s/Str + policy :- Policy] + (when-not (feat/lifecycle-management-type conn) + (throw (ex-info "Lifecycle management not supported" + {:engine engine :version version}))) + + (case engine + :elasticsearch + ;; Elasticsearch ILM allows simple PUT to update + (let [response (-> (make-http-opts conn {} [] {:policy policy} nil) + (assoc :method :put + :url (policy-uri uri policy-name engine)) + request-fn + safe-es-read)] + response) + + :opensearch + ;; OpenSearch ISM requires seq_no and primary_term for updates + (let [existing (get-policy-raw conn policy-name)] + (when-not existing + (throw (ex-info "Policy not found for update" + {:policy-name policy-name + :engine engine}))) + (let [seq-no (:_seq_no existing) + primary-term (:_primary_term existing) + normalized-policy (normalize-policy policy engine) + update-url (str (policy-uri uri policy-name engine) + "?if_seq_no=" seq-no + "&if_primary_term=" primary-term) + response (-> (make-http-opts conn {} [] {:policy normalized-policy} nil) + (assoc :method :put + :url update-url) + request-fn + safe-es-read)] + (if (:_id response) + {:acknowledged true} + response))) + + ;; Default fallback + (throw (ex-info "Unknown engine type" + {:engine engine})))) + +(defn- policy-conflict? + "Check if an exception is a 409 conflict error (policy already exists)" + [ex] + (when-let [es-res (-> ex ex-data :es-http-res)] + (= 409 (:status es-res)))) + (s/defn create-policy! "Create a lifecycle management policy. - For Elasticsearch: Creates an ILM policy - For OpenSearch: Creates an ISM policy (automatically transforms ILM if needed) The policy parameter should be in ILM format. It will be automatically - transformed to ISM format if connecting to OpenSearch." + transformed to ISM format if connecting to OpenSearch. + + If the policy already exists (409 conflict), this function will automatically + update the existing policy using update-policy!." [{:keys [uri version engine request-fn] :as conn} :- ESConn policy-name :- s/Str policy :- Policy] @@ -299,21 +376,32 @@ :elasticsearch {:policy policy} :opensearch {:policy normalized-policy} {:policy policy}) - response (-> (make-http-opts conn - {} - [] - request-body - nil) - (assoc :method :put - :url (policy-uri uri policy-name engine)) - request-fn - safe-es-read)] - ;; Normalize OpenSearch response to match Elasticsearch format - (case engine - :opensearch (if (:_id response) - {:acknowledged true} - response) - response))) + do-create (fn [] + (-> (make-http-opts conn + {} + [] + request-body + nil) + (assoc :method :put + :url (policy-uri uri policy-name engine)) + request-fn + safe-es-read))] + (try + (let [response (do-create)] + ;; Normalize OpenSearch response to match Elasticsearch format + (case engine + :opensearch (if (:_id response) + {:acknowledged true} + response) + response)) + (catch Exception e + (if (policy-conflict? e) + ;; Policy already exists, update it instead + (do + (log/info "Policy" policy-name "already exists, updating...") + (update-policy! conn policy-name policy)) + ;; Re-throw other exceptions + (throw e)))))) (s/defn delete-policy! "Delete a lifecycle management policy. @@ -339,19 +427,12 @@ "Get a lifecycle management policy. Works with both Elasticsearch ILM and OpenSearch ISM. Returns the policy in its native format (ILM or ISM)." - [{:keys [uri version engine request-fn] :as conn} :- ESConn + [{:keys [engine] :as conn} :- ESConn policy-name :- s/Str] - (when-not (feat/lifecycle-management-type conn) - (throw (ex-info "Lifecycle management not supported" - {:engine engine :version version}))) - (let [response (-> (make-http-opts conn) - (assoc :method :get - :url (policy-uri uri policy-name engine)) - request-fn - safe-es-read)] + (let [response (get-policy-raw conn policy-name)] ;; Normalize OpenSearch response to match Elasticsearch format ;; ES: {:policy-name {:policy {:phases {...}}}} - ;; OS GET: {:_id "policy-name" :policy {:policy_id ... :states [...] ...}} + ;; OS GET: {:_id "policy-name" :_seq_no N :_primary_term N :policy {...}} ;; We need to wrap the policy in the same structure as Elasticsearch (case engine :opensearch (when (:_id response) diff --git a/test/ductile/index_test.clj b/test/ductile/index_test.clj index ccca9c8..f6db2a1 100644 --- a/test/ductile/index_test.clj +++ b/test/ductile/index_test.clj @@ -97,6 +97,70 @@ ;; Verify deletion (is (= nil (lifecycle/get-policy conn policy-name)))))) +(deftest ^:integration policy-update-test + (let [policy-name "test-update-policy" + policy-v1 {:phases + {:hot + {:min_age "0ms" + :actions + {:rollover {:max_docs 100000}}}}} + policy-v2 {:phases + {:hot + {:min_age "0ms" + :actions + {:rollover {:max_docs 200000}}}}}] + (for-each-es-version + "Policy update operations" + ;; Clean up any existing policy + (try (lifecycle/delete-policy! conn policy-name) (catch Exception _)) + + ;; Create initial policy + (is (= {:acknowledged true} + (lifecycle/create-policy! conn policy-name policy-v1))) + + ;; Test update-policy! + (is (= {:acknowledged true} + (lifecycle/update-policy! conn policy-name policy-v2))) + + ;; Verify update (OpenSearch transforms to ISM, so check differently) + (let [retrieved (lifecycle/get-policy conn policy-name)] + (case engine + :elasticsearch + (is (= 200000 + (get-in retrieved [(keyword policy-name) :policy :phases :hot :actions :rollover :max_docs]))) + + :opensearch + ;; For OpenSearch, verify the ISM format has the updated value + (let [ism-policy (get-in retrieved [(keyword policy-name) :policy]) + hot-state (first (:states ism-policy)) + rollover-action (first (filter :rollover (:actions hot-state)))] + (is (= 200000 (get-in rollover-action [:rollover :min_doc_count])))))) + + ;; Test create-policy! with existing policy (should auto-update) + (let [policy-v3 {:phases + {:hot + {:min_age "0ms" + :actions + {:rollover {:max_docs 300000}}}}}] + (is (= {:acknowledged true} + (lifecycle/create-policy! conn policy-name policy-v3))) + + ;; Verify the auto-update worked + (let [retrieved (lifecycle/get-policy conn policy-name)] + (case engine + :elasticsearch + (is (= 300000 + (get-in retrieved [(keyword policy-name) :policy :phases :hot :actions :rollover :max_docs]))) + + :opensearch + (let [ism-policy (get-in retrieved [(keyword policy-name) :policy]) + hot-state (first (:states ism-policy)) + rollover-action (first (filter :rollover (:actions hot-state)))] + (is (= 300000 (get-in rollover-action [:rollover :min_doc_count]))))))) + + ;; Clean up + (lifecycle/delete-policy! conn policy-name)))) + (deftest ^:integration index-crud-ops (let [indexname "test_index" indexkw (keyword indexname)]