<commit_before>;;
;; Copyright (c) Two Sigma Open Source, LLC
;;
;; Licensed under the Apache License, Version 2.0 (the "License");
;; you may not use this file except in compliance with the License.
;; You may obtain a copy of the License at
;;
;;  http://www.apache.org/licenses/LICENSE-2.0
;;
;; Unless required by applicable law or agreed to in writing, software
;; distributed under the License is distributed on an "AS IS" BASIS,
;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
;; See the License for the specific language governing permissions and
;; limitations under the License.
;;
(ns waiter.async-request
  (:require [clojure.core.async :as async]
            [clojure.string :as str]
            [clojure.tools.logging :as log]
            [metrics.counters :as counters]
            [metrics.timers :as timers]
            [waiter.correlation-id :as cid]
            [waiter.metrics :as metrics]
            [waiter.scheduler :as scheduler]
            [waiter.service :as service]
            [waiter.statsd :as statsd])
  (:import (java.net ConnectException SocketTimeoutException URI URLEncoder)
           (java.util.concurrent TimeoutException)))

(def daemon-counter (metrics/waiter-counter "async" "monitor" "daemon"))

(def in-flight-requests-counter (metrics/waiter-counter "async" "monitor" "request"))

(def status-check-timer (metrics/waiter-timer "async" "monitor" "status-check"))

(defn normalize-location-header
  "Uses the absolute url from the request to create a sanitized version of the location header.
   If the location has the same scheme, host and port as the request url, then a relative version is returned.
   Else the normalized absolute url of the location is returned.
   This function expects request-absolute-url to be a valid url including host and port."
  [request-absolute-url location-header]
  (if (str/blank? location-header)
    ""
    (let [request-uri (URI. (str request-absolute-url))
          location-uri (URI. (str location-header))
          location-absolute-url (str (.resolve (.normalize request-uri) (.normalize location-uri)))
          request-base-url (str (.getScheme request-uri) "://" (.getAuthority request-uri))]
      (if (str/starts-with? location-absolute-url request-base-url)
        (subs location-absolute-url (count request-base-url))
        location-absolute-url))))

(defn monitor-async-request
  "Launches a go-block that monitors the state of an async request at specified intervals.
   It makes calls to the backend instance and inspects the responses to decide when to treat the request as complete.
   A request is not complete as long as the backend keeps returning a 200 response.
   The request is forcefully completed at timeout."
  [make-http-request complete-async-request request-still-active? status-endpoint check-interval-ms request-timeout-ms
   correlation-id exit-chan]
  (cid/with-correlation-id
    (str correlation-id "|status-check")
    (async/go
      (try
        (counters/inc! daemon-counter)
        (log/info "monitoring async request at intervals of" check-interval-ms "ms")
        (loop [ttl request-timeout-ms]
          (if-not (pos? ttl)
            (do
              (log/info "request has timed out, releasing allocated instance")
              (complete-async-request :success)
              :monitor-timed-out)
            (let [timeout-chan (async/timeout (min check-interval-ms ttl))
                  [message trigger-chan] (async/alts! [exit-chan timeout-chan] :priority true)
                  continue-looping (if (= trigger-chan timeout-chan) (request-still-active?) (not= message :exit))]
              (if-not continue-looping
                (do
                  (log/info "request has been cleared from store, exiting monitoring loop")
                  (complete-async-request :success)
                  (if (= trigger-chan exit-chan) :request-terminated :request-no-longer-active))
                (let [{:keys [body headers error status]}
                      (metrics/with-counter
                        in-flight-requests-counter
                        (timers/start-stop-time!
                          status-check-timer
                          (async/<! (make-http-request))))]
                  (when body
                    (async/close! body))
                  (if error
                    (do
                      (condp instance? error
                        ConnectException (log/debug error "error in performing status check")
                        SocketTimeoutException (log/debug error "timeout in performing status check")
                        TimeoutException (log/debug error "timeout in performing status check")
                        Throwable (log/warn error "unexpected error in performing status check"))
                      (log/info (.getMessage error) "releasing allocated instance")
                      (complete-async-request :instance-error)
                      :make-request-error)
                    (case (int status)
                      200
                      (do
                        (log/debug "async request has not yet completed")
                        (recur (max 0 (- ttl check-interval-ms))))
                      303
                      (do
                        (log/info "async request has completed, result headers" headers)
                        (let [location-header (get headers "location")
                              location (normalize-location-header status-endpoint location-header)]
                          (if (str/starts-with? (str location) "/")
                            (recur (max 0 (- ttl check-interval-ms)))
                            (do
                              (log/info "completing async request as result location is not a relative path:" location)
                              (complete-async-request :success)
                              :status-see-other))))
                      410
                      (do
                        (log/info "async request has completed, result is no longer available!")
                        (complete-async-request :success)
                        :status-gone)
                      (do<commit_msg>adds log for async status with location header not starting with a slash (#1003)<commit_after>;;
;; Copyright (c) Two Sigma Open Source, LLC
;;
;; Licensed under the Apache License, Version 2.0 (the "License");
;; you may not use this file except in compliance with the License.
;; You may obtain a copy of the License at
;;
;;  http://www.apache.org/licenses/LICENSE-2.0
;;
;; Unless required by applicable law or agreed to in writing, software
;; distributed under the License is distributed on an "AS IS" BASIS,
;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
;; See the License for the specific language governing permissions and
;; limitations under the License.
;;
(ns waiter.async-request
  (:require [clojure.core.async :as async]
            [clojure.string :as str]
            [clojure.tools.logging :as log]
            [metrics.counters :as counters]
            [metrics.timers :as timers]
            [waiter.correlation-id :as cid]
            [waiter.metrics :as metrics]
            [waiter.scheduler :as scheduler]
            [waiter.service :as service]
            [waiter.statsd :as statsd])
  (:import (java.net ConnectException SocketTimeoutException URI URLEncoder)
           (java.util.concurrent TimeoutException)))

(def daemon-counter (metrics/waiter-counter "async" "monitor" "daemon"))

(def in-flight-requests-counter (metrics/waiter-counter "async" "monitor" "request"))

(def status-check-timer (metrics/waiter-timer "async" "monitor" "status-check"))

(defn normalize-location-header
  "Uses the absolute url from the request to create a sanitized version of the location header.
   If the location has the same scheme, host and port as the request url, then a relative version is returned.
   Else the normalized absolute url of the location is returned.
   This function expects request-absolute-url to be a valid url including host and port."
  [request-absolute-url location-header]
  (if (str/blank? location-header)
    ""
    (let [request-uri (URI. (str request-absolute-url))
          location-uri (URI. (str location-header))
          location-absolute-url (str (.resolve (.normalize request-uri) (.normalize location-uri)))
          request-base-url (str (.getScheme request-uri) "://" (.getAuthority request-uri))]
      (if (str/starts-with? location-absolute-url request-base-url)
        (subs location-absolute-url (count request-base-url))
        location-absolute-url))))

(defn monitor-async-request
  "Launches a go-block that monitors the state of an async request at specified intervals.
   It makes calls to the backend instance and inspects the responses to decide when to treat the request as complete.
   A request is not complete as long as the backend keeps returning a 200 response.
   The request is forcefully completed at timeout."
  [make-http-request complete-async-request request-still-active? status-endpoint check-interval-ms request-timeout-ms
   correlation-id exit-chan]
  (cid/with-correlation-id
    (str correlation-id "|status-check")
    (async/go
      (try
        (counters/inc! daemon-counter)
        (log/info "monitoring async request at intervals of" check-interval-ms "ms")
        (loop [ttl request-timeout-ms]
          (if-not (pos? ttl)
            (do
              (log/info "request has timed out, releasing allocated instance")
              (complete-async-request :success)
              :monitor-timed-out)
            (let [timeout-chan (async/timeout (min check-interval-ms ttl))
                  [message trigger-chan] (async/alts! [exit-chan timeout-chan] :priority true)
                  continue-looping (if (= trigger-chan timeout-chan) (request-still-active?) (not= message :exit))]
              (if-not continue-looping
                (do
                  (log/info "request has been cleared from store, exiting monitoring loop")
                  (complete-async-request :success)
                  (if (= trigger-chan exit-chan) :request-terminated :request-no-longer-active))
                (let [{:keys [body headers error status]}
                      (metrics/with-counter
                        in-flight-requests-counter
                        (timers/start-stop-time!
                          status-check-timer
                          (async/<! (make-http-request))))]
                  (when body
                    (async/close! body))
                  (if error
                    (do
                      (condp instance? error
                        ConnectException (log/debug error "error in performing status check")
                        SocketTimeoutException (log/debug error "timeout in performing status check")
                        TimeoutException (log/debug error "timeout in performing status check")
                        Throwable (log/warn error "unexpected error in performing status check"))
                      (log/info (.getMessage error) "releasing allocated instance")
                      (complete-async-request :instance-error)
                      :make-request-error)
                    (case (int status)
                      200
                      (do
                        (log/debug "async request has not yet completed")
                        (recur (max 0 (- ttl check-interval-ms))))
                      303
                      (do
                        (log/info "async request has completed, result headers" headers)
                        (let [location-header (get headers "location")
                              location (normalize-location-header status-endpoint location-header)]
                          (if (str/starts-with? (str location) "/")
                            (do
                              (log/info "continuing async request status checks as location is a relative path:" location)
                              (recur (max 0 (- ttl check-interval-ms))))
                            (do
                              (log/info "completing async request as result location is not a relative path:" location)
                              (complete-async-request :success)
                              :status-see-other))))
                      410
                      (do
                        (log/info "async request has completed, result is no longer available!")
                        (complete-async-request :success)
                        :status-gone)
                      (do
{'fraction_non_alphanumeric': 0.0936441679136804, 'fraction_numerical': 0.0031503504764905095, 'mean_word_length': 1.8971024412502853, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4315012', 'n_tokens_mistral': 3212, 'n_tokens_neox': 3050, 'n_words': 1078}