CMR-11560: Fix Collection Delete on Large Collections

access-control-app/src/cmr/access_control/config.clj

@@ -4,8 +4,9 @@
             [cmr.message-queue.config :as rmq-conf]))
 
 (declare access-control-exchange-name provider-exchange-name provider-queue-name index-queue-name
-         concept-ingest-exchange-name index-queue-listener-count sync-entry-titles-concept-ids-collection-batch-size
-         enable-edl-groups enable-cmr-group-sids enable-cmr-groups enable-sids-get)
+         concept-ingest-exchange-name ingest-delete-exchange-name index-queue-listener-count
+         sync-entry-titles-concept-ids-collection-batch-size
+         enable-edl-groups enable-cmr-group-sids enable-cmr-groups enable-sids-get lockdown-acl-get-requests)
 
 (defconfig access-control-exchange-name
   "The access control exchange to which update/save messages are published for access control data."
@@ -27,6 +28,10 @@
   "The ingest exchange to which collection and granule change messages are published."
   {:default "cmr_ingest.exchange"})
 
+(defconfig ingest-delete-exchange-name
+  "The exchange to which ingest delete event messages are published."
+  {:default "cmr_ingest_delete.exchange"})
+
 (defconfig index-queue-listener-count
   "Number of worker threads to use for the queue listener"
   {:default 5
@@ -65,7 +70,9 @@
   (assoc (rmq-conf/default-config)
          :queues [(index-queue-name) (provider-queue-name)]
          :exchanges [(access-control-exchange-name) (provider-exchange-name)
-                     (concept-ingest-exchange-name)]
+                     (concept-ingest-exchange-name)
+                     (ingest-delete-exchange-name)]
          :queues-to-exchanges {(index-queue-name) [(access-control-exchange-name)
-                                                   (concept-ingest-exchange-name)]
+                                                   (concept-ingest-exchange-name)
+                                                   (ingest-delete-exchange-name)]
                                (provider-queue-name) [(provider-exchange-name)]}))

elastic-utils-lib/src/cmr/elastic_utils/config.clj

@@ -121,6 +121,20 @@
   {:default false
    :type Boolean})
 
+(declare default-index-queue-max-wait-secs)
+(defconfig default-index-queue-max-wait-secs
+  "The default maximum time (in seconds) the indexer will wait for an Elasticsearch
+   concept deletion to finish before cancelling it. This MUST be set to slightly
+   less than the SQS Delete Queue Visibility Timeout of the index queue."
+  {:default 270 ;; 4.5 mins on a 5 min sqs timeout
+   :type Long})
+
+(declare delete-by-query-error-retry-attempts)
+(defconfig delete-by-query-error-retry-attempts
+  "The amount of times the delete-by-query will attempt to retry elastic deletes based on specific errors before giving up"
+  {:default 3
+   :type Long})
+
 (defn gran-elastic-config
   "Returns the elastic config as a map"
   []

elastic-utils-lib/src/cmr/elastic_utils/es_helper.clj

@@ -130,21 +130,23 @@
 (defn- attempt-to-start-task
   "Makes a single attempt to start the delete-by-query task.
   Throws an exception on any failure, returns a task-id on success."
-  [conn index query]
-  (let [start-task-url (es-util/url-with-path conn index "_delete_by_query")
+  [conn index query options]
+  (let [slices (get options :slices 1)               ;; Default to 1
+        scroll-size (get options :scroll-size 500)   ;; Default to 500
+        start-task-url (es-util/url-with-path conn index "_delete_by_query")
         response (http/post start-task-url
                             (merge (:http-opts conn)
                                    {:headers {"Authorization" (es-config/elastic-admin-token)
                                               "Confirm-delete-action" "true"
                                               :client-id t-config/cmr-client-id}
                                     :content-type :json
                                     :query-params {:wait_for_completion false
-                                                   :slices 1
-                                                   :scroll_size 500
+                                                   :slices slices
+                                                   :scroll_size scroll-size
                                                    :conflicts "proceed"}
                                     :body (json/generate-string {:query query})
                                     :throw-exceptions false}))
-        _ (debug "Delete-By-Query: Response for starting delete query task is " response)
+        _ (debug (format "Delete-By-Query: Response for starting delete query task with query %s slices %s and scroll size %s for index %s is %s" query slices scroll-size index response))
         status (:status response)
         body (:body response)]
 
@@ -171,15 +173,33 @@
     (catch Exception e
       (warn (str "Delete-By-Query: Failed to cancel task " task-id ". It may keep running in the background. Exception: " (ex-message e))))))
 
+(defn- cleanup-task-record!
+  "Deletes the permanent record of the task from the ES .tasks index
+   to prevent it from silently eating up disk space over time."
+  [conn task-id]
+  (try
+    (debug (str "Delete-By-Query: Cleaning up task record for " task-id " from .tasks index."))
+    ;; Remove a stored task result
+    (let [delete-url (es-util/url-with-path conn (str ".tasks/_doc/" task-id))]
+      (http/delete delete-url
+                   (merge (:http-opts conn)
+                          {:headers {"Authorization" (es-config/elastic-admin-token)
+                                     :client-id t-config/cmr-client-id}
+                           :throw-exceptions false})))
+    (catch Exception e
+      (warn (str "Delete-By-Query: Failed to clean up task record " task-id ". Exception: " (ex-message e))))))
+
 (defn- poll-task-for-completion
   "Polls a given task-id until it completes, fails, or times out."
-  [conn task-id]
+  [conn task-id options]
   (let [polling-interval-ms 5000
-        ;; Hard limit set to 4.5 minutes to safely fit inside the 5-minute SQS window
-        max-wait-ms (* 4.5 60 1000)
+        ;; how long our app will wait for ES to finish deleting before giving up.
+        ;; This val is always < delete SQS timeout val to prevent SQS from picking up the same msg again and
+        ;; re-trying delete from the beginning causing 2+ parallel massive deletes on ES at the same time
+        max-wait-ms (get options :max-wait-ms (* (es-config/default-index-queue-max-wait-secs) 1000))
         start-time (System/currentTimeMillis)]
 
-    (info (str "Delete-By-Query: Polling task " task-id " for completion..."))
+    (debug (format "Delete-By-Query: Polling task %s for completion with max-wait-ms %s..." task-id max-wait-ms))
     (loop []
       (let [check-task-url (es-util/url-with-path conn (str "_tasks/" task-id))
             task-status-response (http/get check-task-url
@@ -218,7 +238,7 @@
               ;; Task ran into version conflicts
               (> conflicts 0)
               (do
-                (warn (format "Delete-By-Query: Task %s finished, but skipped %d documents due to version conflicts." task-id conflicts))
+                (warn (format "Delete-By-Query: Task %s finished, but skipped %s documents due to version conflicts." task-id conflicts))
                 {:status 200
                  :body final-response})
 
@@ -227,16 +247,25 @@
               (do
                 (info (format "Delete-By-Query: Task %s completed successfully. Final result: %s"
                               task-id final-response))
+
+                (cleanup-task-record! conn task-id)
+
                 {:status 200
                  :body final-response})))
 
           ;; Check for timeout
           (> (- (System/currentTimeMillis) start-time) max-wait-ms)
           (do
-            (warn (format "Delete-By-Query: Task %s exceeded the 4.5-minute limit. Cancelling the task on Elasticsearch to prevent overlap on the next SQS retry." task-id))
+            (let [timeout-minutes (/ max-wait-ms 60000.0)]
+              (warn (format (str "Delete-By-Query: Task %s exceeded the %.1f-minute limit. Cancelling the task on "
+                                 "Elasticsearch to prevent overlap on the next SQS retry.")
+                            task-id timeout-minutes)))
+
             ;; End the task on the cluster to prevent overlapping tasks when SQS retries this failed msg
             (cancel-task! conn task-id)
-            ;; Throw the error so the queue broker handles the retry
+
+            ;; Throw the error so SQS can pick up failed msg and re-try, picking up where the prev failed msg delete left off.
+            ;; Creates sequential deletes without duplicate delete attempts
             (throw (ex-info (str "Delete-By-Query: Timed out waiting for task " task-id) {:task-id task-id})))
 
           ;; Continue polling
@@ -248,40 +277,49 @@
             (recur)))))))
 
 (defn delete-by-query
-  "Performs a delete-by-query operation, blocking until completion, with retries for scroll contexts."
-  [conn index _mapping-type query]
-  (debug (format "Delete-By-Query: Delete-by-query started for index : %s with query %s" index query))
-
-  (loop [attempt 1]
-    (let [result (try
-                   (let [task-id (attempt-to-start-task conn index query)]
-                     [:ok (poll-task-for-completion conn task-id)])
-                   (catch Exception e
-                     [:error e]))]
-
-      ;; Attempt Retries if failed due to Elastic's scroll context errors
-      (if (= :ok (first result))
-        (second result)
-        (let [e (second result)
-              ex-data-map (ex-data e)]
-
-          (if (and (< attempt 3) (= :scroll-context-error (:type ex-data-map)))
-            (do
-              ;; Log the exact ex-info message and data that triggered the retry
-              (warn (format "Delete-By-Query: Caught exception: %s | Data: %s"
-                            (ex-message e)
-                            (pr-str ex-data-map)))
-              (info (format "Delete-By-Query: Scroll context error on attempt %d. Retrying in 1 second..." attempt))
-              (Thread/sleep 1000)
-              (recur (inc attempt)))
-
-            (do
-              ;; Log the exact ex-info message and data right before permanently failing
-              (error (format "Delete-By-Query: Task permanently failed or max retries reached. Exception: %s | Data: %s"
-                             (ex-message e)
-                             (pr-str ex-data-map)))
-              ;; Pass the exception up the chain
-              (throw e))))))))
+  "Performs a delete-by-query operation, blocking until completion, with retries for scroll contexts.
+  Accepts an optional map of tuning parameters: {:slices, :scroll-size, :max-wait-ms}"
+  ([conn index mapping-type query]
+   (delete-by-query conn index mapping-type query {}))
+
+  ([conn index _mapping-type query options]
+   (loop [attempt 1]
+     (let [max-attempts (es-config/delete-by-query-error-retry-attempts)
+           result (try
+                    (let [task-id (attempt-to-start-task conn index query options)]
+                      [:ok (poll-task-for-completion conn task-id options)])
+                    (catch Exception e
+                      [:error e]))]
+
+       ;; Attempt Retries if failed due to Elastic's scroll context errors
+       (if (= :ok (first result))
+         (second result)
+         (let [e (second result)
+               ex-data-map (ex-data e)]
+
+           (if (and (< attempt max-attempts) (= :scroll-context-error (:type ex-data-map)))
+             (do
+               ;; Exponential Backoff
+               (let [calculated-backoff (* (Math/pow 5 attempt) 1000)
+                     ;; never allow the backoff to be > 60 secs
+                     backoff-ms (long (min calculated-backoff 60000))]
+                 (info (format "Delete-By-Query: Ran into scroll context error on attempt %d/%d. Retrying in %d seconds..."
+                               attempt
+                               max-attempts
+                               (/ backoff-ms 1000)))
+                 (Thread/sleep backoff-ms))
+
+               (recur (inc attempt)))
+
+             (do
+               (error (format (str "Delete-By-Query: Task permanently failed due to max retries reached or some "
+                                   "exception that wasn't related to scroll context error. "
+                                   "Retries %d/%d | Exception: %s | Data: %s")
+                              attempt
+                              max-attempts
+                              (ex-message e)
+                              (pr-str ex-data-map)))
+               (throw e)))))))))
 
 (defn delete-index
   "Deletes an index from the elastic store"

indexer-app/src/cmr/indexer/config.clj

@@ -4,10 +4,11 @@
    [cmr.message-queue.config :as rmq-conf]))
 
 ;; This is here to remove linter unresolved variable error.
-(declare index-queue-name index-queue-listener-count all-revisions-index-queue-name
+(declare index-queue-name index-queue-listener-count index-delete-queue-name index-delete-queue-listener-count
+         index-delete-queue-visibility-timeout-secs index-delete-queue-max-wait-secs all-revisions-index-queue-name
          all-revisions-index-queue-listener-count deleted-granule-index-queue-name
          deleted-granules-index-queue-listener-count provider-queue-name provider-queue-listener-count
-         ingest-exchange-name provider-exchange-name deleted-concept-revision-exchange-name
+         ingest-exchange-name ingest-delete-exchange-name provider-exchange-name deleted-concept-revision-exchange-name
          deleted-granule-exchange-name indexer-nrepl-port ongoing-days reduced-indexer-log)
 
 
@@ -36,6 +37,29 @@
   {:default 5
    :type Long})
 
+(defconfig index-delete-queue-name
+  "The queue containing ingest delete events for the indexer"
+  {:default "cmr_index_delete.queue"})
+
+(defconfig index-delete-queue-listener-count
+  "Number of worker threads to use for the index delete queue listener"
+  {:default 2
+   :type Long})
+
+(defconfig index-delete-queue-visibility-timeout-secs
+  "The SQS visibility timeout for the index delete queue to allow enough time for
+  massive concept deletes before sqs retries"
+  {:default 3900 ;; 65 mins
+   :type Long})
+
+(defconfig index-delete-queue-max-wait-secs
+  "The maximum time (in seconds) the indexer will wait for an Elasticsearch
+   concept deletion to finish before cancelling it. This MUST be set to slightly
+   less than the SQS Delete Queue Visibility Timeout of the index delete queue set
+   by the param index-delete-queue-visibility-timeout-secs."
+  {:default 3600 ;; 60 mins on a 65 min sqs visibility timeout
+  :type Long})
+
 (defconfig all-revisions-index-queue-name
   "The queue containing ingest events for the indexer all revisions index. We use a
   separate index here because it allows independent retries of indexing failures. If
@@ -74,6 +98,10 @@
   "The ingest exchange to which ingest event messages are published."
   {:default "cmr_ingest.exchange"})
 
+(defconfig ingest-delete-exchange-name
+  "The exchange to which ingest delete event messages are published."
+  {:default "cmr_ingest_delete.exchange"})
+
 (defconfig provider-exchange-name
   "The ingest exchange to which provider change and non-ingest messages are published."
   {:default "cmr_ingest_provider.exchange"})
@@ -94,20 +122,23 @@
   []
   (assoc (rmq-conf/default-config)
          :queues [(index-queue-name)
+                  (index-delete-queue-name)
                   (all-revisions-index-queue-name)
                   (deleted-granule-index-queue-name)
                   (provider-queue-name)]
          :exchanges [(ingest-exchange-name)
+                     (ingest-delete-exchange-name)
                      (deleted-concept-revision-exchange-name)
                      (deleted-granule-exchange-name)
                      (provider-exchange-name)]
+         :queues-to-policies {(index-delete-queue-name) {:visibility-timeout-secs (index-delete-queue-visibility-timeout-secs)}}
          :queues-to-exchanges
          {(index-queue-name) [(ingest-exchange-name)]
+          (index-delete-queue-name) [(ingest-delete-exchange-name)]
           (provider-queue-name) [(provider-exchange-name)]
           (deleted-granule-index-queue-name) [(deleted-granule-exchange-name)]
-          ;; The all revisions index  queue will be bound to both the ingest exchange and the
-          ;; deleted concept revision exchange
           (all-revisions-index-queue-name) [(ingest-exchange-name)
+                                            (ingest-delete-exchange-name)
                                             (deleted-concept-revision-exchange-name)]}))
 
 (defconfig indexer-nrepl-port

indexer-app/src/cmr/indexer/services/event_handler.clj

@@ -2,7 +2,7 @@
   "Provides functions for subscribing to and handling events."
   (:require
    [cmr.common.concepts :as cc]
-   [cmr.common.log :refer [debug]]
+   [cmr.common.log :refer [debug error]]
    [cmr.common.services.errors :as errors]
    [cmr.common.util :as util]
    [cmr.indexer.config :as config]
@@ -128,6 +128,35 @@
                      (str *ns*) concept-id revision-id all-revisions-index? tm))
       result)))
 
+;; SEPARATE INGEST AND INGEST DELETES BUT EVENTUALLY ALL CONCEPT DELETES WILL GO TO THE SAME QUEUE AND BE HANDLED BY THIS FUNC
+(defmulti handle-ingest-delete-event
+  "Handle the concept deletes that can be requested via the index delete queue. This is particularly helpful in handling
+  large deletes that can go on for a long time."
+  (fn [_context _all-revisions-index? msg]
+    (keyword (:action msg))))
+
+(defmethod handle-ingest-delete-event :default
+  [_context _all-revisions-index? msg]
+  (let [action (:action msg)
+        err-msg (format "CRITICAL: The index-delete-queue received an unexpected message action: %s.
+        This queue is strictly for deletions! Especially, the large deletions. Message: %s"
+                        action (pr-str msg))]
+    (error err-msg)
+    ;; Throw the exception so the queue broker does NOT delete the message so it can go to the DLQ.
+    (throw (ex-info err-msg {:msg msg}))))
+
+(defmethod handle-ingest-delete-event :concept-delete
+  [context all-revisions-index? {:keys [concept-id revision-id]}]
+  (when-not (= :humanizer (cc/concept-id->type concept-id))
+    (let [[tm result] (util/time-execution
+                        (indexer/delete-concept
+                          context concept-id revision-id {:ignore-conflict? true
+                                                          :all-revisions-index? all-revisions-index?}))]
+      (debug (format (str "Timed function %s handle-ingest-event concept-delete for concept-id %s "
+                          "revision-id %s all-revisions-index? %s took %d ms.")
+                     (str *ns*) concept-id revision-id all-revisions-index? tm))
+      result)))
+
 (defn subscribe-to-events
   "Subscribe to event messages on various queues"
   [context]
@@ -136,6 +165,10 @@
       (queue-protocol/subscribe queue-broker
                                 (config/provider-queue-name)
                                 #(handle-provider-event context %)))
+    (dotimes [_ (config/index-delete-queue-listener-count)]
+      (queue-protocol/subscribe queue-broker
+                                (config/index-delete-queue-name)
+                                #(handle-ingest-delete-event context false %)))
     (dotimes [_ (config/index-queue-listener-count)]
       (queue-protocol/subscribe queue-broker
                                 (config/index-queue-name)