<commit_before>(ns riemann.influxdb
  "Forwards events to InfluxDB"
  (:require [capacitor.core :as influx])
  (:use [clojure.string :only [join split]]))

(defn influxdb-series
  "Constructs a series name for an event."
  [opts event]
  ((:series opts) event))
 
(defn influxdb
  "Returns a function which accepts an event and sends it to InfluxDB.

  ;; For giving series name as the concatenation of :host and :service fields with dot separator.

  (influxdb {:host \"play.influxdb.org\" :port 8086 :series #(str (:host %) \".\" (:service %))})

  Options:

  :name           Name of the metric which is same as the series name.

  :db             Name of the DB of InfluxDB to push riemann-events.

  :username       Name of the user who is allowed to push metrics to this DB.

  :password       <PASSWORD>.
 
  :series         Name of the InfluxDB's time-series. Default value is :service field of event,
                  incase that is nil, \"riemann-events\" will be the default name."
  [opts]
  (let [opts (merge {:host "127.0.0.1"
                     :port 8086
                     :scheme "http"
                     :username "root"
                     :password "<PASSWORD>"
                     :db "riemann"
                     :series :service} opts)
        client (influx/make-client opts)]
    (fn [event]
      (when (:metric event)
        (when (:service event)
          (influx/post-points client (if-let [series (influxdb-series opts event)] series "riemann-events") [{ :name (:service event)
                                                                                                               :host (if-let [hostname (:host event)] hostname "") 
                                                                                                               :state (:state event)
                                                                                                               :value (:metric event) }]))))))<commit_msg>Refactor influxdb client to accept sequences of events.<commit_after>(ns riemann.influxdb
  "Forwards events to InfluxDB"
  (:require [capacitor.core :as influx])
  (:use [clojure.string :only [join split]]))

(defn event->point
  "Transform a Riemann event to an InfluxDB point, or nil if the event is
  missing a metric or service."
  [event]
  (when (and (:metric event) (:service event))
    {:name (:service event)
     :host (or (:host event) "")
     :state (:state event)
     :value (:metric event)}))

(defn events->points
  "Takes a series fn that finds the series for a given event, and a sequence of
  events, and emits a map of series names to vectors of points for that series."
  [series-fn events]
  (persistent!
    (reduce (fn [m event]
              (let [series (or (series-fn event) "riemann-events")
                    point  (event->point event)]
                (if point
                  (assoc! m series (conj (get m series []) point))
                  m)))
            (transient {})
            events)))

(defn influxdb
  "Returns a function which accepts an event, or sequence of events, and sends
  it to InfluxDB.

  ; For giving series name as the concatenation of :host and :service fields
  ; with dot separator.

  (influxdb {:host   \"play.influxdb.org\"
             :port   8086
             :series #(str (:host %) \".\" (:service %))})

  Options:

  :name           Name of the metric which is same as the series name.

  :db             Name of the DB of InfluxDB to push riemann-events.

  :username       Name of the user who is allowed to push metrics to this DB.

  :password       <PASSWORD> the <PASSWORD>.

  :series         Function which takes an event and returns the InfluxDB series
                  name to use. Defaults to :service. If this function returns
                  nil, series names default to \"riemann-events\"."
  [opts]
  (let [opts (merge {:host "127.0.0.1"
                     :port 8086
                     :scheme "http"
                     :username "root"
                     :password "<PASSWORD>"
                     :db "riemann"
                     :series :service}
                    opts)
        series (:series opts)
        client (influx/make-client opts)]

    (fn stream [events]
      (let [events (if (sequential? events) events (list events))
            points (events->points series events)]
        (when-not (empty? points)
          (doseq [[series points] points]
            (influx/post-points client series points)))))))
{'fraction_non_alphanumeric': 0.10070984915705412, 'fraction_numerical': 0.006211180124223602, 'mean_word_length': 1.6871275327771156, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 15, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4379689', 'n_tokens_mistral': 1269, 'n_tokens_neox': 1155, 'n_words': 469}