<commit_before>(ns cmr.ingest.data.indexer
  "Implements Ingest App datalayer access interface. Takes on the role of a proxy to indexer app."
  (:require [clj-http.client :as client]
            [cheshire.core :as json]
            [cmr.ingest.config :as config]
            [cmr.common.config :as cfg]
            [cmr.common.log :as log :refer (debug info warn error)]
            [cmr.common.services.errors :as errors]
            [cmr.common.services.health-helper :as hh]
            [cmr.system-trace.core :refer [deftracefn]]
            [cmr.system-trace.http :as ch]
            [cmr.transmit.config :as transmit-config]
            [cmr.transmit.connection :as transmit-conn]
            [cmr.message-queue.services.queue :as queue]
            [cmr.acl.core :as acl]
            [clojail.core :as timeout]))

(defn- get-headers
  "Gets the headers to use for communicating with the indexer."
  [context]
  (assoc (ch/context->http-headers context) acl/token-header (transmit-config/echo-system-token)))

(defn- delete-from-index-via-http
  "Execute http delete of the given url on the indexer"
  [context delete-url]
  (let [conn (transmit-config/context->app-connection context :indexer)
        indexer-root (transmit-conn/root-url conn)
        response (client/delete (format "%s/%s" indexer-root delete-url)
                                {:accept :json
                                 :throw-exceptions false
                                 :headers (get-headers context)
                                 :connection-manager (transmit-conn/conn-mgr conn)})
        status (:status response)]
    (when-not (some #{200, 204} [status])
      (errors/internal-error!
        (format "Delete %s operation failed. Indexer app response status code: %s %s"
                delete-url status response)))
    response))

(def ^:const publish-timeout-ms
  "Number of milliseconds to wait for a publish request to be confirmed before considering the
  request timed out."
  60000)

(defn- put-message-on-queue
  "Put an index operation on the message queue. Throws a service unavailable error if the message
  fails to be put on the queue.

  Requests to publish a message include a timeout value to handle error cases with the Rabbit MQ
  server. Otherwise when the memory limit within RabbitMQ is reached it could block indefinitely.
  Also when the server is down or unreachable, calls to publish will return a failure. Rather than
  raise an error to the caller immediately, the publication will be retried until the timeout
  period has expired. By retrying, routine maintenance such as restarting the RabbitMQ server
  will not result in any ingest errors returned to the provider."
  ([context msg]
   (put-message-on-queue context msg publish-timeout-ms))
  ([context msg timeout-ms]
   (let [queue-broker (get-in context [:system :queue-broker])
         queue-name (config/index-queue-name)
         start-time (System/currentTimeMillis)]
     (loop []
       (let [message-published? (try
                                  (let [response (timeout/thunk-timeout
                                                   #(queue/publish queue-broker queue-name msg)
                                                   timeout-ms)]
                                    (when-not response
                                      (if (< (- (System/currentTimeMillis) start-time) timeout-ms)
                                        (warn "Retrying publishing message: " msg)
                                        (errors/throw-service-error
                                          :service-unavailable
                                          (str "All retries exhausted to queue message: " msg))))
                                    response)
                                  (catch java.util.concurrent.TimeoutException e
                                    (errors/throw-service-error
                                      :service-unavailable
                                      (str "Request timed out when attempting to publish message: "
                                           msg)
                                      e)))]
         (when-not message-published?
           (Thread/sleep 2000)
           (recur)))))))

(defmulti index-concept-with-method
  "Index the concept using an http request or the indexing queue"
  (fn [context concept-id revision-id]
    (keyword (config/indexing-communication-method))))

(defmethod index-concept-with-method :http
  [context concept-id revision-id]
  (let [conn (transmit-config/context->app-connection context :indexer)
        indexer-url (transmit-conn/root-url conn)
        concept-attribs {:concept-id concept-id, :revision-id revision-id}
        response (client/post indexer-url {:body (json/generate-string concept-attribs)
                                           :content-type :json
                                           :throw-exceptions false
                                           :accept :json<commit_msg>CMR-1247: Make queue publication timeout a configuration parameter<commit_after>(ns cmr.ingest.data.indexer
  "Implements Ingest App datalayer access interface. Takes on the role of a proxy to indexer app."
  (:require [clj-http.client :as client]
            [cheshire.core :as json]
            [cmr.ingest.config :as config]
            [cmr.common.config :as cfg :refer [defconfig]]
            [cmr.common.log :as log :refer (debug info warn error)]
            [cmr.common.services.errors :as errors]
            [cmr.common.services.health-helper :as hh]
            [cmr.system-trace.core :refer [deftracefn]]
            [cmr.system-trace.http :as ch]
            [cmr.transmit.config :as transmit-config]
            [cmr.transmit.connection :as transmit-conn]
            [cmr.message-queue.services.queue :as queue]
            [cmr.acl.core :as acl]
            [clojail.core :as timeout]))

(defn- get-headers
  "Gets the headers to use for communicating with the indexer."
  [context]
  (assoc (ch/context->http-headers context) acl/token-header (transmit-config/echo-system-token)))

(defn- delete-from-index-via-http
  "Execute http delete of the given url on the indexer"
  [context delete-url]
  (let [conn (transmit-config/context->app-connection context :indexer)
        indexer-root (transmit-conn/root-url conn)
        response (client/delete (format "%s/%s" indexer-root delete-url)
                                {:accept :json
                                 :throw-exceptions false
                                 :headers (get-headers context)
                                 :connection-manager (transmit-conn/conn-mgr conn)})
        status (:status response)]
    (when-not (some #{200, 204} [status])
      (errors/internal-error!
        (format "Delete %s operation failed. Indexer app response status code: %s %s"
                delete-url status response)))
    response))

(defconfig publish-queue-timeout-ms
  "Number of milliseconds to wait for a publish request to be confirmed before considering the
  request timed out."
  {:default 60000 :type Long})

(defn- put-message-on-queue
  "Put an index operation on the message queue. Throws a service unavailable error if the message
  fails to be put on the queue.

  Requests to publish a message include a timeout value to handle error cases with the Rabbit MQ
  server. Otherwise when the memory limit within RabbitMQ is reached it could block indefinitely.
  Also when the server is down or unreachable, calls to publish will return a failure. Rather than
  raise an error to the caller immediately, the publication will be retried until the timeout
  period has expired. By retrying, routine maintenance such as restarting the RabbitMQ server
  will not result in any ingest errors returned to the provider."
  ([context msg]
   (put-message-on-queue context msg (publish-queue-timeout-ms)))
  ([context msg timeout-ms]
   (let [queue-broker (get-in context [:system :queue-broker])
         queue-name (config/index-queue-name)
         start-time (System/currentTimeMillis)]
     (loop []
       (let [message-published? (try
                                  (let [response (timeout/thunk-timeout
                                                   #(queue/publish queue-broker queue-name msg)
                                                   timeout-ms)]
                                    (when-not response
                                      (if (< (- (System/currentTimeMillis) start-time) timeout-ms)
                                        (warn "Attempt to queue message failed. Retrying: " msg)
                                        (errors/throw-service-error
                                          :service-unavailable
                                          (str "All retries exhausted to queue message: " msg))))
                                    response)
                                  (catch java.util.concurrent.TimeoutException e
                                    (errors/throw-service-error
                                      :service-unavailable
                                      (str "Request timed out when attempting to publish message: "
                                           msg)
                                      e)))]
         (when-not message-published?
           (Thread/sleep 2000)
           (recur)))))))

(defmulti index-concept-with-method
  "Index the concept using an http request or the indexing queue"
  (fn [context concept-id revision-id]
    (keyword (config/indexing-communication-method))))

(defmethod index-concept-with-method :http
  [context concept-id revision-id]
  (let [conn (transmit-config/context->app-connection context :indexer)
        indexer-url (transmit-conn/root-url conn)
        concept-attribs {:concept-id concept-id, :revision-id revision-id}
        response (client/post indexer-url {:body (json/generate-string concept-attribs)
                                           :content-type :json
                                           :throw-exceptions false
                                           :accept :json
{'fraction_non_alphanumeric': 0.08710250839423267, 'fraction_numerical': 0.003357693067351373, 'mean_word_length': 1.7289140393424953, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 64, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6602171', 'n_tokens_mistral': 2565, 'n_tokens_neox': 2477, 'n_words': 823}