<commit_before>(ns cyanite-fetcher.core
  (:require [clojure.string :as str]
            [clojure.tools.cli :as cli]
            [clj-time.local :as tl]
            [clojure.core.reducers :as r]
            [clojurewerkz.elastisch.rest :as esr]
            [clojurewerkz.elastisch.rest.document :as esrd]
            [qbits.alia :as alia])
  (:gen-class))

;;------------------------------------------------------------------------------
;; Aggregation
;;------------------------------------------------------------------------------

(defmulti aggregate-with
  "This transforms a raw list of points according to the provided aggregation
   method. Each point is stored as a list of data points, so multiple
   methods make sense (max, min, mean). Additionally, a raw method is
   provided"
  (comp first list))

(defmethod aggregate-with :mean
  [_ {:keys [data] :as metric}]
  (if (seq data)
    (-> metric
        (dissoc :data)
        (assoc :metric (/ (reduce + 0.0 data) (count data))))
    metric))

(defmethod aggregate-with :sum
  [_ {:keys [data] :as metric}]
  (-> metric
      (dissoc :data)
      (assoc :metric (reduce + 0.0 data))))

(defmethod aggregate-with :max
  [_ {:keys [data] :as metric}]
  (-> metric
      (dissoc :data)
      (assoc :metric (apply max data))))

(defmethod aggregate-with :min
  [_ {:keys [data] :as metric}]
  (-> metric
      (dissoc :data)
      (assoc :metric (apply min data))))

(defmethod aggregate-with :raw
  [_ {:keys [data] :as metric}]
  (-> metric
      (dissoc :data)
      (assoc :metric data)))

(defmethod aggregate-with :avg
           [_ {:keys [data] :as metric}]
  (if (seq data)
    (-> metric
        (dissoc :data)
        (assoc :metric (/ (reduce + 0.0 data) (count data))))
    metric))

;;
;; if no method given parse metric name and select aggregation function
;;
(defn detect-aggregate
  [{:keys [path] :as metric}]
  (if-let [[_ m] (re-find #"^(sum|avg|mean|min|max|raw)\..*" path)]
    (aggregate-with (keyword m) metric)
    (aggregate-with :mean metric)))

;;------------------------------------------------------------------------------
;; Cassandra
;;------------------------------------------------------------------------------

(def ^:const keyspace "metric")

(defn fetchq
  "Yields a cassandra prepared statement of 7 arguments:

* `paths`: list of paths
* `tenant`: tenant identifier
* `rollup`: interval between points at this resolution
* `period`: rollup multiplier which determines the time to keep points for
* `min`: return points starting from this timestamp
* `max`: return points up to this timestamp
* `limit`: maximum number of points to return"
  [session]
  (alia/prepare
   session
   (str
    "SELECT path,data,time FROM metric WHERE "
    "path = ? AND tenant = ? AND rollup = ? AND period = ? "
    "AND time >= ? AND time <= ? ORDER BY time ASC;")))

(defn par-fetch
  "Fetch data in parallel fashion."
  [session fetch! paths tenant rollup period from to]
  (let [futures
        (doall (map #(future
                       (try
                         (->> (alia/execute
                               session fetch!
                               {:values [% tenant (int rollup)
                                         (int period)
                                         from to]
                                :fetch-size Integer/MAX_VALUE})
                              (map detect-aggregate)
                              (doall)
                              (seq))
                         (catch Exception e
                           (shutdown-agents)
                           (throw e))))
                    paths))]
    (map deref futures)))

(defn par-fetch-pmap
  "Fetch data in parallel fashion using pmap."
  [session fetch! paths tenant rollup period from to]
  (pmap (fn [path] (->> (alia/execute
                         session fetch!
                         {:values [path tenant (int rollup)
                                   (int period)
                                   from to]
                          :fetch-size Integer/MAX_VALUE})
                        (map detect-aggregate)
                        (doall)
                        (seq))) paths))

(defn par-fetch-async
  "Fetch data in parallel fashion using execute-async."
  [session fetch! paths tenant rollup period from to]
  (let [data (atom [])
        channels (doall (map #(alia/execute-async session fetch!
                                                  {:values [% tenant (int rollup)
                                                            (int period)
                                                            from to]
                                                   :fetch-size Integer/MAX_VALUE
                                                   :success (fn [rows]
                                                              (->> rows
                                                                   (map detect-aggregate)
                                                                   (doall)
                                                                   (seq)
                                                                   (swap! data conj)))}) paths))]
    (map deref channels)
    @data))

(defn c-get-data
  "Get data from C*."
  [host paths tenant rollup period from to]
  (println "Connecting to Cassandra...")
  (let [session (-> (alia/cluster {:contact-points [host]})
                    (alia/connect keyspace))
        fetch! (fetchq session)]
    (println "Getting data form Cassandra...")
    (let [data (time (doall (par-fetch session fetch! paths tenant
                                       rollup period from to)))]
      (newline)
      data)))

;;------------------------------------------------------------------------------
;; ElasticSearch<commit_msg>Add alia/shutdown<commit_after>(ns cyanite-fetcher.core
  (:require [clojure.string :as str]
            [clojure.tools.cli :as cli]
            [clj-time.local :as tl]
            [clojure.core.reducers :as r]
            [clojurewerkz.elastisch.rest :as esr]
            [clojurewerkz.elastisch.rest.document :as esrd]
            [qbits.alia :as alia])
  (:gen-class))

;;------------------------------------------------------------------------------
;; Aggregation
;;------------------------------------------------------------------------------

(defmulti aggregate-with
  "This transforms a raw list of points according to the provided aggregation
   method. Each point is stored as a list of data points, so multiple
   methods make sense (max, min, mean). Additionally, a raw method is
   provided"
  (comp first list))

(defmethod aggregate-with :mean
  [_ {:keys [data] :as metric}]
  (if (seq data)
    (-> metric
        (dissoc :data)
        (assoc :metric (/ (reduce + 0.0 data) (count data))))
    metric))

(defmethod aggregate-with :sum
  [_ {:keys [data] :as metric}]
  (-> metric
      (dissoc :data)
      (assoc :metric (reduce + 0.0 data))))

(defmethod aggregate-with :max
  [_ {:keys [data] :as metric}]
  (-> metric
      (dissoc :data)
      (assoc :metric (apply max data))))

(defmethod aggregate-with :min
  [_ {:keys [data] :as metric}]
  (-> metric
      (dissoc :data)
      (assoc :metric (apply min data))))

(defmethod aggregate-with :raw
  [_ {:keys [data] :as metric}]
  (-> metric
      (dissoc :data)
      (assoc :metric data)))

(defmethod aggregate-with :avg
           [_ {:keys [data] :as metric}]
  (if (seq data)
    (-> metric
        (dissoc :data)
        (assoc :metric (/ (reduce + 0.0 data) (count data))))
    metric))

;;
;; if no method given parse metric name and select aggregation function
;;
(defn detect-aggregate
  [{:keys [path] :as metric}]
  (if-let [[_ m] (re-find #"^(sum|avg|mean|min|max|raw)\..*" path)]
    (aggregate-with (keyword m) metric)
    (aggregate-with :mean metric)))

;;------------------------------------------------------------------------------
;; Cassandra
;;------------------------------------------------------------------------------

(def ^:const keyspace "metric")

(defn fetchq
  "Yields a cassandra prepared statement of 7 arguments:

* `paths`: list of paths
* `tenant`: tenant identifier
* `rollup`: interval between points at this resolution
* `period`: rollup multiplier which determines the time to keep points for
* `min`: return points starting from this timestamp
* `max`: return points up to this timestamp
* `limit`: maximum number of points to return"
  [session]
  (alia/prepare
   session
   (str
    "SELECT path,data,time FROM metric WHERE "
    "path = ? AND tenant = ? AND rollup = ? AND period = ? "
    "AND time >= ? AND time <= ? ORDER BY time ASC;")))

(defn par-fetch
  "Fetch data in parallel fashion."
  [session fetch! paths tenant rollup period from to]
  (let [futures
        (doall (map #(future
                       (try
                         (->> (alia/execute
                               session fetch!
                               {:values [% tenant (int rollup)
                                         (int period)
                                         from to]
                                :fetch-size Integer/MAX_VALUE})
                              (map detect-aggregate)
                              (doall)
                              (seq))
                         (catch Exception e
                           (shutdown-agents)
                           (throw e))))
                    paths))]
    (map deref futures)))

(defn par-fetch-pmap
  "Fetch data in parallel fashion using pmap."
  [session fetch! paths tenant rollup period from to]
  (pmap (fn [path] (->> (alia/execute
                         session fetch!
                         {:values [path tenant (int rollup)
                                   (int period)
                                   from to]
                          :fetch-size Integer/MAX_VALUE})
                        (map detect-aggregate)
                        (doall)
                        (seq))) paths))

(defn par-fetch-async
  "Fetch data in parallel fashion using execute-async."
  [session fetch! paths tenant rollup period from to]
  (let [data (atom [])
        channels (doall (map #(alia/execute-async session fetch!
                                                  {:values [% tenant (int rollup)
                                                            (int period)
                                                            from to]
                                                   :fetch-size Integer/MAX_VALUE
                                                   :success (fn [rows]
                                                              (->> rows
                                                                   (map detect-aggregate)
                                                                   (doall)
                                                                   (seq)
                                                                   (swap! data conj)))}) paths))]
    (map deref channels)
    @data))

(defn c-get-data
  "Get data from C*."
  [host paths tenant rollup period from to]
  (println "Connecting to Cassandra...")
  (let [cluster (alia/cluster {:contact-points [host]})
        session (alia/connect keyspace cluster)
        fetch! (fetchq session)]
    (println "Getting data form Cassandra...")
    (try
      (let [data (time (doall (par-fetch session fetch! paths tenant
                                         rollup period from to)))]
        (newline)
        data)
      (catch Exception e
        (alia/shutdown cluster)
        (throw e)))))

;;------------------------------------------------------------------------------
;; ElasticSearch
{'fraction_non_alphanumeric': 0.16835647950924426, 'fraction_numerical': 0.0011928090653488966, 'mean_word_length': 1.20390536988359, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 55, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '638869', 'n_tokens_mistral': 2956, 'n_tokens_neox': 2816, 'n_words': 948}