Insert and bulk insert actions on timeseries

Author: alebellu

code/src/sixsq/nuvla/db/binding.clj

@@ -105,31 +105,6 @@
      then a 400 (bad-request) response must be returned.  Other appropriate
      error codes can also be thrown.")
 
-  (add-metric
-    [this collection-id data options]
-    "This function adds the given metric to the database.  The metric
-     must not already exist in the database.
-
-     On success, the function must return a 201 ring response with the
-     relative URL of the new metric as the Location.
-
-     On failure, the function must throw an ex-info containing the error
-     ring response.  The error must be 409 (conflict) if the metric
-     exists already.  Other appropriate error codes can also be thrown.")
-
-  (bulk-insert-metrics
-    [this collection-id data options]
-    "This function insert the given metrics in the database where the
-    collection-id corresponds to the name of a metrics Collection.
-
-     On success, the function must return the summary map of what was done
-     on the db.
-
-     On failure, the function must throw an ex-info containing the error
-     ring response.  If the resource-id does not correspond to a Collection,
-     then a 400 (bad-request) response must be returned.  Other appropriate
-     error codes can also be thrown.")
-
   (bulk-delete
     [this collection-id options]
     "This function removes the given resources in the database where the
@@ -157,15 +132,39 @@
      error codes can also be thrown.")
 
   (create-timeseries
-    [this timeseries-id options]
-    "This function creates the given timeseries in the database.")
+    [this index options]
+    "This function creates a timeseries with the given index name in the database.")
 
   (retrieve-timeseries
-    [this timeseries-id]
+    [this index]
     "This function retrieves the identified timeseries from the database.
 
      On success, this returns the clojure map representation of the
      timeseries.  The response must not be embedded in a ring response.
 
      On failure, this function must throw an ex-info containing the error
-     ring response. If the resource doesn't exist, use a 404 status."))
+     ring response. If the resource doesn't exist, use a 404 status.")
+
+  (add-timeseries-datapoint
+    [this index data options]
+    "This function adds the given timeseries datapoint to the database.
+     The datapoint with the given timestamp and dimensions must not already exist in the database.
+
+     On success, the function must return a 201 ring response with the
+     relative URL of the new metric as the Location.
+
+     On failure, the function must throw an ex-info containing the error
+     ring response.  The error must be 409 (conflict) if the metric
+     exists already.  Other appropriate error codes can also be thrown.")
+
+  (bulk-insert-timeseries-datapoints
+    [this index data options]
+    "This function insert the given timeseries datapoints in the database.
+
+     On success, the function must return the summary map of what was done
+     on the db.
+
+     On failure, the function must throw an ex-info containing the error
+     ring response.  If the resource-id does not correspond to a Collection,
+     then a 400 (bad-request) response must be returned.  Other appropriate
+     error codes can also be thrown."))

code/src/sixsq/nuvla/db/es/binding.clj

@@ -1,7 +1,8 @@
 (ns sixsq.nuvla.db.es.binding
   "Binding protocol implemented for an Elasticsearch database that makes use
    of the Elasticsearch REST API."
-  (:require [clojure.tools.logging :as log]
+  (:require [clojure.data.json :as json]
+            [clojure.tools.logging :as log]
             [qbits.spandex :as spandex]
             [sixsq.nuvla.auth.utils.acl :as acl-utils]
             [sixsq.nuvla.db.binding :refer [Binding]]
@@ -234,13 +235,12 @@
             msg   (str "unexpected exception querying: " (or error e))]
         (throw (r/ex-response msg 500))))))
 
-(defn add-metric-data
-  [client collection-id data {:keys [refresh]
-                              :or   {refresh true}
-                              :as   _options}]
+(defn add-timeseries-datapoint
+  [client index data {:keys [refresh]
+                      :or   {refresh true}
+                      :as   _options}]
   (try
-    (let [index        (escu/collection-id->index collection-id)
-          updated-data (-> data
+    (let [updated-data (-> data
                            (dissoc :timestamp)
                            (assoc "@timestamp" (:timestamp data)))
           response     (spandex/request client {:url          [index :_doc]
@@ -251,20 +251,19 @@
       (if success?
         {:status 201
          :body   {:status  201
-                  :message (str collection-id " metric added")}}
-        (r/response-conflict collection-id)))
+                  :message (str index " metric added")}}
+        (r/response-conflict index)))
     (catch Exception e
       (let [{:keys [status body] :as _response} (ex-data e)
             error (:error body)]
         (if (= 409 status)
-          (r/response-conflict collection-id)
+          (r/response-conflict index)
           (r/response-error (str "unexpected exception: " (or error e))))))))
 
-(defn bulk-insert-metrics
-  [client collection-id data _options]
+(defn bulk-insert-timeseries-datapoints
+  [client index data _options]
   (try
-    (let [index          (escu/collection-id->index collection-id)
-          data-transform (fn [{:keys [timestamp] :as doc}]
+    (let [data-transform (fn [{:keys [timestamp] :as doc}]
                            (-> doc
                                (dissoc :timestamp)
                                (assoc "@timestamp" timestamp)))
@@ -517,12 +516,6 @@
   (query-native [_ collection-id query]
     (query-data-native client collection-id query))
 
-  (add-metric [_ collection-id data options]
-    (add-metric-data client collection-id data options))
-
-  (bulk-insert-metrics [_ collection-id data options]
-    (bulk-insert-metrics client collection-id data options))
-
   (bulk-delete [_ collection-id options]
     (bulk-delete-data client collection-id options))
 
@@ -535,6 +528,13 @@
   (retrieve-timeseries [_ timeseries-id]
     (retrieve-timeseries-impl client timeseries-id))
 
+  (add-timeseries-datapoint [_ index data options]
+    (add-timeseries-datapoint client index data options))
+
+  (bulk-insert-timeseries-datapoints [_ index data options]
+    (bulk-insert-timeseries-datapoints client index data options))
+
+
   Closeable
   (close [_]
     (when sniffer

code/src/sixsq/nuvla/db/impl.clj

@@ -45,24 +45,24 @@
 (defn query-native [collection-id query]
   (p/query-native *impl* collection-id query))
 
-(defn add-metric [collection-id data & [options]]
-  (p/add-metric *impl* collection-id data options))
-
-(defn bulk-insert-metrics [collection-id data & [options]]
-  (p/bulk-insert-metrics *impl* collection-id data options))
-
 (defn bulk-delete [collection-id & [options]]
   (p/bulk-delete *impl* collection-id options))
 
 (defn bulk-edit [collection-id & [options]]
   (p/bulk-edit *impl* collection-id options))
 
-(defn create-timeseries [timeseries-id & [options]]
-  (p/create-timeseries *impl* timeseries-id options))
+(defn create-timeseries [index & [options]]
+  (p/create-timeseries *impl* index options))
 
 (defn retrieve-timeseries [timeseries-id]
   (p/retrieve-timeseries *impl* timeseries-id))
 
+(defn add-timeseries-datapoint [index data & [options]]
+  (p/add-timeseries-datapoint *impl* index data options))
+
+(defn bulk-insert-timeseries-datapoints [index data & [options]]
+  (p/bulk-insert-timeseries-datapoints *impl* index data options))
+
 (defn close []
   (when-let [^Closeable impl *impl*]
     (try

code/src/sixsq/nuvla/server/resources/common/std_crud.clj

@@ -8,6 +8,7 @@
     [clojure.walk :as w]
     [sixsq.nuvla.auth.acl-resource :as a]
     [sixsq.nuvla.auth.utils :as auth]
+    [sixsq.nuvla.db.es.common.utils :as escu]
     [sixsq.nuvla.db.impl :as db]
     [sixsq.nuvla.server.middleware.cimi-params.impl :as impl]
     [sixsq.nuvla.server.resources.common.crud :as crud]
@@ -213,23 +214,24 @@
                           (str "_" resource-name))]
       (create-bulk-job action-name resource-name authn-info acl body))))
 
-(defn add-metric-fn
+(defn add-timeseries-datapoint-fn
   [resource-name collection-acl _resource-uri & {:keys [validate-fn options]}]
   (validate-collection-acl collection-acl)
   (fn [{:keys [body] :as request}]
     (a/throw-cannot-add collection-acl request)
     (validate-fn body)
-    (db/add-metric resource-name body options)))
+    (db/add-timeseries-datapoint (escu/collection-id->index resource-name)
+                                 body options)))
 
-(defn bulk-insert-metrics-fn
-  [resource-name collection-acl _collection-uri]
+(defn bulk-insert-timeseries-datapoints-fn
+  [index collection-acl _collection-uri]
   (validate-collection-acl collection-acl)
   (fn [{:keys [body] :as request}]
     (throw-bulk-header-missing request)
     (a/throw-cannot-add collection-acl request)
     (a/throw-cannot-bulk-action collection-acl request)
     (let [options    (select-keys request [:nuvla/authn :body])
-          response   (db/bulk-insert-metrics resource-name body options)]
+          response   (db/bulk-insert-timeseries-datapoints index body options)]
       (r/json-response response))))
 
 (defn generic-bulk-operation-fn

code/src/sixsq/nuvla/server/resources/spec/timeseries.cljc

@@ -14,6 +14,7 @@
 
 (s/def ::field-type
   (assoc (st/spec field-types)
+    :json-schema/type "string"
     :json-schema/description "Timeseries field name"))
 
 (s/def ::dimension
@@ -31,13 +32,21 @@
 
 (s/def ::metric-type
   (assoc (st/spec metric-types)
+    :json-schema/type "string"
     :json-schema/description "Timeseries metric type"))
 
+(s/def ::optional
+  (-> (st/spec boolean?)
+      (assoc :name "optional"
+             :json-schema/type "boolean"
+             :json-schema/description "optional value ? (default false)")))
+
 (s/def ::metric
   (assoc (st/spec (su/only-keys
                     :req-un [::field-name
                              ::field-type
-                             ::metric-type]))
+                             ::metric-type]
+                    :opt-un [::optional]))
     :json-schema/type "map"
     :json-schema/description "Timeseries metric"))