fix some bulk api and scrolling tests

Author: jmaeng72

access-control-app/src/cmr/access_control/data/bulk_index.clj

@@ -40,13 +40,13 @@
   * :force-version? - true indicates that we should overwrite whatever is in elasticsearch with the
   latest regardless of whether the version in the database is older than the _version in elastic.
   Returns a map with keys of :num-indexed and :max-revision-date."
-  ([context concept-batches]
-   (bulk-index-with-revision-date context concept-batches nil))
-  ([context concept-batches options]
+  ([context concept-batches es-cluster-name]
+   (bulk-index-with-revision-date context concept-batches es-cluster-name nil))
+  ([context concept-batches es-cluster-name options]
    (reduce (fn [{:keys [num-indexed max-revision-date]} batch]
              (let [max-revision-date (get-max-revision-date batch max-revision-date)
                    batch (prepare-batch context batch options)]
-               (es/bulk-index-documents context batch)
+               (es/bulk-index-documents context batch es-cluster-name)
                {:num-indexed (+ num-indexed (count batch))
                 :max-revision-date max-revision-date}))
            {:num-indexed 0 :max-revision-date nil}

bootstrap-app/src/cmr/bootstrap/api/bulk_index.clj

@@ -74,6 +74,7 @@
                    \"granule\", \"collection\", etc.
     concept_ids  - a vector of concept ids."
   [context request-details-map params]
+  (println "INSIDE index-concepts-by-id in api/bulk_index.clj")
   (let [dispatcher (api-util/get-dispatcher context params :index-concepts-by-id)
         provider-id (get request-details-map "provider_id")
         concept-type (keyword (get request-details-map "concept_type"))

bootstrap-app/src/cmr/bootstrap/api/messages.clj

@@ -88,6 +88,7 @@
 
 (defn index-concepts-by-id
   [params result]
+  (println "INSIDE index-concepts-by-id in api/messages.clj")
   (if (api-util/synchronous? params)
     (format "Processed %s concepts for bulk indexing." result)
     (str "Processing concepts for bulk indexing.")))

bootstrap-app/src/cmr/bootstrap/data/bulk_index.clj

@@ -8,6 +8,7 @@
    [cmr.common.concepts :as cc]
    [cmr.common.log :refer (info warn error)]
    [cmr.common.util :as util]
+   [cmr.elastic-utils.config :as es-config]
    [cmr.elastic-utils.es-helper :as es-helper]
    [cmr.indexer.indexer-util :as indexer-util]
    [cmr.indexer.data.elasticsearch :as es]
@@ -138,10 +139,10 @@
 ;; TODO CMR-10636 fix this. We need to separate out the concepts and go to each seperate cluster
 (defn- bulk-index-concept-batches
   "Bulk index the given concept batches in both regular index and all revisions index."
-  [system concept-batches]
+  [system concept-batches es-cluster-name]
   (let [indexer-context {:system (helper/get-indexer system)}]
-    (index/bulk-index indexer-context concept-batches nil {:all-revisions-index? true})
-    (index/bulk-index indexer-context concept-batches nil {})))
+    (index/bulk-index indexer-context concept-batches es-cluster-name {:all-revisions-index? true})
+    (index/bulk-index indexer-context concept-batches es-cluster-name {})))
 
 (defn- index-concepts-by-provider
   "Bulk index concepts for the given provider and concept-type."
@@ -158,7 +159,10 @@
                          db provider
                          params
                          (:db-batch-size system))
-        num-concepts (bulk-index-concept-batches system concept-batches)
+        es-cluster-name (if (= concept-type :granule)
+                          cmr.elastic-utils.config/gran-elastic-name
+                          cmr.elastic-utils.config/non-gran-elastic-name)
+        num-concepts (bulk-index-concept-batches system concept-batches es-cluster-name)
         msg (format "Indexing of %s %s revisions for provider %s completed."
                     num-concepts
                     (name concept-type)
@@ -184,14 +188,15 @@
   "Bulk index ACLs or access groups"
   [system concept-batches]
   (info "Indexing concepts")
-  (ac-bulk-index/bulk-index-with-revision-date {:system (helper/get-indexer system)} concept-batches))
+  (ac-bulk-index/bulk-index-with-revision-date {:system (helper/get-indexer system)} concept-batches cmr.elastic-utils.config/non-gran-elastic-name))
 
 (defn- index-concepts
   "Bulk index the given concepts using the indexer-app"
-  [system concept-batches]
+  [system concept-batches es-cluster-name]
   (info "Indexing concepts")
-  (index/bulk-index-with-revision-date {:system (helper/get-indexer system)} concept-batches))
+  (index/bulk-index-with-revision-date {:system (helper/get-indexer system)} concept-batches es-cluster-name))
 
+;; TODO CMR-10636 -- This whole series of index funcs need to be tested thoroughly.
 (defn- fetch-and-index-new-concepts
   "Get batches of concepts for a given provider/concept type that have a revision-date
   newer than the given date time and then index them."
@@ -206,9 +211,13 @@
                  params)
         concept-batches (db/find-concepts-in-batches
                           db provider params (:db-batch-size system))
+        es-cluster-name (if (= concept-type :granule)
+                          cmr.elastic-utils.config/gran-elastic-name
+                          cmr.elastic-utils.config/non-gran-elastic-name)
+        _ (println "INSIDE fetch-and-index-new-concepts with concept-type = " concept-type " and concept batches = " concept-batches " and es-cluster-name = " es-cluster-name)
         {:keys [max-revision-date num-indexed]} (if (contains? #{:acl :access-group} concept-type)
                                                  (index-access-control-concepts system concept-batches)
-                                                 (index-concepts system concept-batches))]
+                                                 (index-concepts system concept-batches es-cluster-name))]
 
     (info (format (str "Indexed %d %s(s) for provider %s with revision-date later than %s and max "
                        "revision date was %s.")
@@ -242,6 +251,7 @@
 (defn index-concepts-by-id
   "Index concepts of the given type for the given provider with the given concept-ids."
   [system provider-id concept-type concept-ids]
+  (println "INSIDE index-concepts-by-id in data/bulk_index.clj")
   (let [db (helper/get-metadata-db-db system)
         provider (helper/get-provider system provider-id)
         ;; Oracle only allows 1000 values in an 'in' clause, so we partition here
@@ -254,7 +264,11 @@
                                                                          {:concept-type concept-type :concept-id batch}
                                                                          (:db-batch-size system))]
                           concept-batch)
-        total (index/bulk-index {:system (helper/get-indexer system)} concept-batches nil)]
+        es-cluster-name (if (= :granule concept-type)
+                          cmr.elastic-utils.config/gran-elastic-name
+                          cmr.elastic-utils.config/non-gran-elastic-name)
+        _ (println "es-cluster-name selected = " es-cluster-name " for concept-batches in index-concepts-by-id = " concept-batches)
+        total (index/bulk-index {:system (helper/get-indexer system)} concept-batches es-cluster-name)]
 
     ;; for concept types that have all revisions index, also index the all revisions index
     (when-not (#{:tag :granule} concept-type)

bootstrap-app/src/cmr/bootstrap/services/bootstrap_service.clj

@@ -100,6 +100,7 @@
 (defn index-concepts-by-id
   "Bulk index the concepts given by the concept-ids"
   [context dispatcher provider-id concept-type concept-ids]
+  (println "INSIDE index-concepts-by-id in bootstrap_service.clj")
   (dispatch/index-concepts-by-id dispatcher context provider-id concept-type concept-ids))
 
 (defn index-variables

bootstrap-app/src/cmr/bootstrap/services/dispatch/impl/async.clj

@@ -55,6 +55,7 @@
 (defn index-concepts-by-id
   "Bulk index the concepts given by the concept-ids"
   [this context provider-id concept-type concept-ids]
+  (println "INSIDE index-concepts-by-id in async.clj")
   (let [channel (:concept-id-channel this)]
     (info "Adding bulk index request to concept-id channel.")
     (async/go (>! channel {:provider-id provider-id

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

@@ -120,3 +120,15 @@
    ;; to retry again
    :retry-handler nil
    :admin-token (elastic-admin-token)})
+
+;; TODO 10636 unit test
+(defn es-cluster-name-str->keyword
+  [es-cluster-name]
+  (let [es-cluster-name-keyword (if (keyword? es-cluster-name)
+                                  es-cluster-name
+                                  (keyword es-cluster-name))]
+    (if (or (= es-cluster-name-keyword (keyword cmr.elastic-utils.config/gran-elastic-name))
+            (= es-cluster-name-keyword (keyword cmr.elastic-utils.config/non-gran-elastic-name)))
+      es-cluster-name-keyword
+      (throw (Exception. (str "Expected es-cluster-name to be gran-elastic or non-gran-elastic, but got value of " es-cluster-name " instead.")))
+      )))

elastic-utils-lib/src/cmr/elastic_utils/search/ac_elasticsearch.clj

@@ -6,6 +6,7 @@
    [cmr.common.concepts :as cs]
    [cmr.common.log :refer [error]]
    [cmr.common.services.errors :as errors]
+   [cmr.elastic-utils.config]
    [cmr.elastic-utils.es-helper :as es-helper]))
 
 (def MAX_BULK_OPERATIONS_PER_REQUEST
@@ -70,11 +71,11 @@
 ;: TODO 10636 - need to fix this func.
 (defn bulk-index-documents
   "Save a batch of documents in Elasticsearch."
-  [context docs]
+  [context docs es-cluster-name]
   (doseq [docs-batch (partition-all MAX_BULK_OPERATIONS_PER_REQUEST docs)]
     (let [bulk-operations (cmr-bulk/create-bulk-index-operations docs-batch)
-          ;; TODO 10636- this conn will not work anymore there is no :db, it's either gran or non-gran cluster, so need to change
-          conn (get-in context [:system :db :conn])
+          ;; TODO 10636- technically this func to do this conversion already exists in context->es-store in indexer-util.clj, need to organize this later
+          conn (get-in context [:system (cmr.elastic-utils.config/es-cluster-name-str->keyword es-cluster-name)])
           response (es-helper/bulk conn bulk-operations)
           ;; we don't care about version conflicts or deletes that aren't found
           bad-errors (some (fn [item]

elastic-utils-lib/src/cmr/elastic_utils/search/es_index.clj

@@ -142,13 +142,13 @@
   [ex _scroll-id]
   (throw ex))
 
-;; TODO 10636 Fix
+;; TODO 10636 - Test
 (defn- scroll-search
   "Performs a scroll search, handling errors where possible."
-  [context scroll-id]
+  [context scroll-id es-cluster-name]
   (try
     (es-helper/scroll
-     (context->conn context cmr.elastic-utils.config/gran-elastic-name)
+     (context->conn context es-cluster-name)
      scroll-id
      {:scroll (es-config/elastic-scroll-timeout)})
     (catch ExceptionInfo e
@@ -189,7 +189,7 @@
   (try
     (if (pos? max-retries)
       (if-let [scroll-id (:scroll-id query)]
-        (scroll-search context scroll-id)
+        (scroll-search context scroll-id (get-es-cluster-name-by-index-info-type-name index-info))
         (es-helper/search
           (context->conn context (get-es-cluster-name-by-index-info-type-name index-info))
           (:index-name index-info)

indexer-app/src/cmr/indexer/data/index_set_elasticsearch.clj

@@ -6,8 +6,10 @@
    [cmr.common.log :as log :refer [info warn error]]
    [cmr.common.services.errors :as errors]
    [cmr.common.util :as util]
+   [cmr.elastic-utils.config]
    [cmr.elastic-utils.es-helper :as es-helper]
    [cmr.elastic-utils.es-index-helper :as esi-helper]
+   [cmr.elastic-utils.search.es-index :as es-index]
    [cmr.indexer.config :as config]
    [cmr.indexer.services.messages :as m]
    [cmr.indexer.indexer-util :as idx-util]
@@ -78,7 +80,7 @@
 (defn get-index-set
   "Fetch index-set associated with an id and a specific elastic cluster."
   [context es-cluster-name index-set-id]
-  (let [es-cluster-name-keyword (idx-util/es-cluster-name-str->keyword es-cluster-name)
+  (let [es-cluster-name-keyword (cmr.elastic-utils.config/es-cluster-name-str->keyword es-cluster-name)
         {:keys [index-name mapping]} (config/idx-cfg-for-index-sets es-cluster-name)
         idx-mapping-type (first (keys mapping))]
     (when-let [result (index-set-exists?
@@ -122,7 +124,7 @@
   "Save the document in Elasticsearch in specific elastic cluster, raise error on failure."
   [context es-index es-mapping-type doc-id es-doc]
   (try
-    (let [es-cluster-name (cmr.elastic-utils.search.es-index/get-es-cluster-name-from-index-name es-index)
+    (let [es-cluster-name (es-index/get-es-cluster-name-from-index-name es-index)
           conn (get-in context [:system (keyword es-cluster-name) :conn])
           result (es-helper/put conn es-index es-mapping-type doc-id es-doc)
           _ (esi-helper/refresh conn es-index)
@@ -139,7 +141,7 @@
 (defn delete-document
   "Delete the document from specific elastic cluster, raise error on failure."
   [context index-name _mapping-type id]
-  (let [es-cluster-name (cmr.elastic-utils.search.es-index/get-es-cluster-name-from-index-name index-name)
+  (let [es-cluster-name (es-index/get-es-cluster-name-from-index-name index-name)
         {:keys [host port admin-token]} (get-in context [:system (keyword es-cluster-name) :config])
         delete-doc-url (format "http://%s:%s/%s/_doc/%s?refresh=true" host port index-name id)
         result (client/delete delete-doc-url