support ingestion timestamp returned by cassandra (#294)

* support ingestion timestamp returned by cassandra

* revert it test

Author: ramazanyich

src/it/resources/conf/auth/auth.props

@@ -1 +1 @@
-test: test
+test: test

src/it/scala/integration/AddAndQueryDataPointsIntegrationSpec.scala

@@ -43,7 +43,7 @@ class AddAndQueryDataPointsIntegrationSpec extends IntegrationSpec {
                   "my.new.metric",
                   Seq(GroupBy.GroupByType("number")),
                   Seq(TagResult("aoeu", Seq("snth"))),
-                  Seq((instant, KNumber(555)))
+                  Seq((instant, KNumber(555), None))
                 )
               )
             )
@@ -85,9 +85,9 @@ class AddAndQueryDataPointsIntegrationSpec extends IntegrationSpec {
                   Seq(GroupBy.GroupByType("number")),
                   List(TagResult("aoeu", List("123", "456")), TagResult("snth", List("321"))),
                   Seq(
-                    instant.plusMillis(1) -> KNumber(111),
-                    instant.plusMillis(2) -> KNumber(222),
-                    instant.plusMillis(3) -> KNumber(333)
+                    (instant.plusMillis(1), KNumber(111), None),
+                    (instant.plusMillis(2), KNumber(222), None),
+                    (instant.plusMillis(3), KNumber(333), None)
                   )
                 )
               )
@@ -149,9 +149,9 @@ class AddAndQueryDataPointsIntegrationSpec extends IntegrationSpec {
                       Seq(GroupBy.GroupByType("number")),
                       Seq(TagResult("aoeu", Seq("snth"))),
                       Seq(
-                        (Instant.parse("1970-01-02T00:00:00Z"), KNumber(555)),
-                        (Instant.parse("1970-01-03T00:00:00Z"), KNull),
-                        (Instant.parse("1970-01-04T00:00:00Z"), KNumber(555))
+                        (Instant.parse("1970-01-02T00:00:00Z"), KNumber(555), None),
+                        (Instant.parse("1970-01-03T00:00:00Z"), KNull, None),
+                        (Instant.parse("1970-01-04T00:00:00Z"), KNumber(555), None)
                       )
                     )
                   )
@@ -171,10 +171,10 @@ class AddAndQueryDataPointsIntegrationSpec extends IntegrationSpec {
                       Seq(GroupBy.GroupByType("number")),
                       Seq(TagResult("aoeu", Seq("snth"))),
                       Seq(
-                        (Instant.parse("1970-01-01T00:00:00Z"), KNull),
-                        (Instant.parse("1970-01-02T00:00:00Z"), KNumber(555)),
-                        (Instant.parse("1970-01-03T00:00:00Z"), KNull),
-                        (Instant.parse("1970-01-04T00:00:00Z"), KNumber(555))
+                        (Instant.parse("1970-01-01T00:00:00Z"), KNull, None),
+                        (Instant.parse("1970-01-02T00:00:00Z"), KNumber(555), None),
+                        (Instant.parse("1970-01-03T00:00:00Z"), KNull, None),
+                        (Instant.parse("1970-01-04T00:00:00Z"), KNumber(555), None)
                       )
                     )
                   )
@@ -194,11 +194,11 @@ class AddAndQueryDataPointsIntegrationSpec extends IntegrationSpec {
                       Seq(GroupBy.GroupByType("number")),
                       Seq(TagResult("aoeu", Seq("snth"))),
                       Seq(
-                        (Instant.parse("1970-01-01T00:00:00Z"), KNull),
-                        (Instant.parse("1970-01-02T00:00:00Z"), KNumber(555)),
-                        (Instant.parse("1970-01-03T00:00:00Z"), KNull),
-                        (Instant.parse("1970-01-04T00:00:00Z"), KNumber(555)),
-                        (Instant.parse("1970-01-05T00:00:00Z"), KNull)
+                        (Instant.parse("1970-01-01T00:00:00Z"), KNull, None),
+                        (Instant.parse("1970-01-02T00:00:00Z"), KNumber(555), None),
+                        (Instant.parse("1970-01-03T00:00:00Z"), KNull, None),
+                        (Instant.parse("1970-01-04T00:00:00Z"), KNumber(555), None),
+                        (Instant.parse("1970-01-05T00:00:00Z"), KNull, None)
                       )
                     )
                   )

src/it/scala/integration/DeleteDataPointsByTagIntegrationSpec.scala

@@ -61,9 +61,9 @@ class DeleteDataPointsByTagIntegrationSpec extends IntegrationSpec {
                     TagResult("snth", Seq("321"))
                   ), // not sure if the order is deteministic. Convert to Set?
                   Seq(
-                    instant.plusMillis(2) -> KNumber(222),
-                    instant.plusMillis(4) -> KNumber(444),
-                    instant.plusMillis(5) -> KNumber(123)
+                    (instant.plusMillis(2), KNumber(222), None),
+                    (instant.plusMillis(4), KNumber(444), None),
+                    (instant.plusMillis(5), KNumber(123), None)
                   )
                 )
               )

src/it/scala/integration/StringDataPointsIntegrationSpec.scala

@@ -46,7 +46,7 @@ class StringDataPointsIntegrationSpec extends IntegrationSpec {
                   "my.new.metric",
                   Seq(GroupBy.GroupByType("text")),
                   Seq(TagResult("aoeu", Seq("snth"))),
-                  Seq((instant, KString("my test string")))
+                  Seq((instant, KString("my test string"), None))
                 )
               )
             )
@@ -83,7 +83,7 @@ class StringDataPointsIntegrationSpec extends IntegrationSpec {
                   "my.new.metric",
                   Seq(GroupBy.GroupByType("text")),
                   Seq(TagResult("aoeu", Seq("snth"))),
-                  Seq((instant, KString("12345")))
+                  Seq((instant, KString("12345"), None))
                 )
               )
             )

src/main/scala/io/waylay/kairosdb/driver/models/Models.scala

@@ -8,21 +8,21 @@ import io.waylay.kairosdb.driver.models.TimeRange.KairosTimeUnit
 import java.time.Instant
 
 /**
-  * Metric names are case sensitive and can only contain the following characters: alphanumeric characters, period ”.”,
-  * slash “/”, dash “-”, and underscore “_” (not enforced)
-  */
+ * Metric names are case sensitive and can only contain the following characters: alphanumeric characters, period ”.”,
+ * slash “/”, dash “-”, and underscore “_” (not enforced)
+ */
 case class MetricName(name: String) extends AnyVal
 
 object TimeRange {
   sealed trait KairosTimeUnit
   case object MILLISECONDS extends KairosTimeUnit
-  case object SECONDS extends KairosTimeUnit
-  case object MINUTES extends KairosTimeUnit
-  case object HOURS extends KairosTimeUnit
-  case object DAYS extends KairosTimeUnit
-  case object WEEKS extends KairosTimeUnit
-  case object MONTHS extends KairosTimeUnit
-  case object YEARS extends KairosTimeUnit
+  case object SECONDS      extends KairosTimeUnit
+  case object MINUTES      extends KairosTimeUnit
+  case object HOURS        extends KairosTimeUnit
+  case object DAYS         extends KairosTimeUnit
+  case object WEEKS        extends KairosTimeUnit
+  case object MONTHS       extends KairosTimeUnit
+  case object YEARS        extends KairosTimeUnit
 }
 case class TimeRange(amount: Long, unit: KairosTimeUnit)
 
@@ -33,34 +33,46 @@ sealed trait DataPoint {
 }
 
 object DataPoint {
-  def apply(metricName: MetricName, value: KairosCompatibleType, timestamp: Instant = Instant.now, tags: Seq[Tag],
-    ttl: Option[TimeRange] = None) = DataPointWithSingleValue(metricName, value, timestamp, tags, ttl)
+  def apply(
+    metricName: MetricName,
+    value: KairosCompatibleType,
+    timestamp: Instant = Instant.now,
+    tags: Seq[Tag],
+    ttl: Option[TimeRange] = None
+  ) = DataPointWithSingleValue(metricName, value, timestamp, tags, ttl)
 }
 
 /**
-  * A data point has with metric name, a single value, a timestamp, and a list of one or more tags
-  *
-  * @param ttl Sets the Cassandra ttl for the data points. None or Some(0.seconds) will not set a TTL
-  *
-  */
-case class DataPointWithSingleValue(metricName: MetricName, value: KairosCompatibleType, timestamp: Instant,
-  tags: Seq[Tag], ttl: Option[TimeRange] = None) extends DataPoint
+ * A data point has with metric name, a single value, a timestamp, and a list of one or more tags
+ *
+ * @param ttl Sets the Cassandra ttl for the data points. None or Some(0.seconds) will not set a TTL
+ */
+case class DataPointWithSingleValue(
+  metricName: MetricName,
+  value: KairosCompatibleType,
+  timestamp: Instant,
+  tags: Seq[Tag],
+  ttl: Option[TimeRange] = None
+) extends DataPoint
 
 /**
-  * A data point has with metric name, a single value, a timestamp, and a list of one or more tags
-  *
-  * @param ttl Sets the Cassandra ttl for the data points. None or Some(0.seconds) will not set a TTL
-  */
-case class DataPointWithMultipleValues(metricName: MetricName, values: Seq[(Instant, KairosCompatibleType)],
-  tags: Seq[Tag] = Seq.empty, ttl: Option[TimeRange] = None) extends DataPoint
-
+ * A data point has with metric name, a single value, a timestamp, and a list of one or more tags
+ *
+ * @param ttl Sets the Cassandra ttl for the data points. None or Some(0.seconds) will not set a TTL
+ */
+case class DataPointWithMultipleValues(
+  metricName: MetricName,
+  values: Seq[(Instant, KairosCompatibleType)],
+  tags: Seq[Tag] = Seq.empty,
+  ttl: Option[TimeRange] = None
+) extends DataPoint
 
 /**
-  * Tags are named properties that identify the data, such as its type and where it comes from
-  *
-  * Tag names and values are case sensitive and can only contain the following characters: alphanumeric characters,
-  * period ”.”, slash “/”, dash “-”, and underscore “_” (not enforced)
-  */
+ * Tags are named properties that identify the data, such as its type and where it comes from
+ *
+ * Tag names and values are case sensitive and can only contain the following characters: alphanumeric characters,
+ * period ”.”, slash “/”, dash “-”, and underscore “_” (not enforced)
+ */
 case class Tag(name: String, value: String)
 
 /** Kairos base URL */
@@ -82,9 +94,7 @@ object KairosDBConfig {
     val uriWithCredentialsOpt = for {
       user <- username
       pass <- password
-    } yield {
-      baseUri.toAbsoluteUrl.withUser(user).withPassword(pass)
-    }
+    } yield baseUri.toAbsoluteUrl.withUser(user).withPassword(pass)
 
     val serverUri = uriWithCredentialsOpt getOrElse baseUri
 
@@ -95,22 +105,22 @@ object KairosDBConfig {
 }
 
 /**
-  * KairosDB provides REST APIs that show the health of the system.
-  *
-  * There are currently two health checks executed for each API.
-  *
-  *  - The JVM thread deadlock check verifies that no deadlocks exist in the KairosDB JVM.
-  *  - The Datastore query check performs a query on the data store to ensure that the data store is responding.
-  *
-  *  Example value: HealthStatusResults(Seq("JVM-Thread-Deadlock: OK", "Datastore-Query: OK"))
-  */
+ * KairosDB provides REST APIs that show the health of the system.
+ *
+ * There are currently two health checks executed for each API.
+ *
+ *  - The JVM thread deadlock check verifies that no deadlocks exist in the KairosDB JVM.
+ *  - The Datastore query check performs a query on the data store to ensure that the data store is responding.
+ *
+ *  Example value: HealthStatusResults(Seq("JVM-Thread-Deadlock: OK", "Datastore-Query: OK"))
+ */
 case class HealthStatusResults(results: Seq[String])
 
 sealed trait HealthCheckResult
 
 object HealthCheckResult {
   case object AllHealthy extends HealthCheckResult
-  case object Unhealthy extends HealthCheckResult
+  case object Unhealthy  extends HealthCheckResult
 }
 
 /** KairosDB only supports numbers and strings. Custom types can be defined */
@@ -124,24 +134,28 @@ object KairosCompatibleType {
   }
 
   case class KNumber(value: BigDecimal) extends KairosCompatibleType {
-    def kairosType:String = {
-      //TODO check if this is 100% correct
+    def kairosType: String =
+      // TODO check if this is 100% correct
       if (value.scale > 0) {
         "double"
       } else {
         "long"
       }
-    }
   }
   case class KString(value: String) extends KairosCompatibleType {
-    val kairosType:String = "string"
+    val kairosType: String = "string"
   }
 }
 
 object QueryResponse {
   case class Response(queries: Seq[ResponseQuery])
   case class ResponseQuery(sampleSize: Int, results: Seq[Result])
-  case class Result(name: MetricName, groupBy: Seq[GroupBy], tags: Seq[TagResult], values: Seq[(Instant, KairosCompatibleType)])
+  case class Result(
+    name: MetricName,
+    groupBy: Seq[GroupBy],
+    tags: Seq[TagResult],
+    values: Seq[(Instant, KairosCompatibleType, Option[Instant])]
+  )
   case class TagResult(name: String, values: Seq[String]) // not sure if this is the correct meaning
 }
 
@@ -152,40 +166,44 @@ object QueryMetricTagsResponse {
 }
 
 object KairosQuery {
+
   /**
-    * Used to filter a query. With `QueryTag("aoeu", Seq("snth", "htns"))` you would only get results where the value of
-    * tag `aoeu` is `snth` or `htns`
-    */
+   * Used to filter a query. With `QueryTag("aoeu", Seq("snth", "htns"))` you would only get results where the value of
+   * tag `aoeu` is `snth` or `htns`
+   */
   case class QueryTag(name: String, allowedValues: Seq[String])
 
   object QueryTag {
-    //def apply(name: String, allowedValue: String): QueryTag = QueryTag(name, Seq(allowedValue))
-    def apply(tuple: (String, String)): QueryTag = QueryTag(tuple._1, Seq(tuple._2))
+    // def apply(name: String, allowedValue: String): QueryTag = QueryTag(name, Seq(allowedValue))
+    def apply(tuple: (String, String)): QueryTag       = QueryTag(tuple._1, Seq(tuple._2))
     def apply(tuple: (String, String)*): Seq[QueryTag] = tuple.map(tup => QueryTag(tup)).to(Seq)
   }
 
   sealed trait Order { val value: String }
 
   object Order {
-    case object Ascending extends Order { override val value = "asc" }
-    case object Descending extends Order { override val value = "desc"}
+    case object Ascending  extends Order { override val value = "asc"  }
+    case object Descending extends Order { override val value = "desc" }
     val defaultOrder = Ascending
   }
 }
 
-/** @param tags Tags narrow down the search. Only metrics that include the tag and matches one of the values are
-  *             returned. Tags are optional.
-  * @param groupBys The resulting data points can be grouped by one or more tags, a time range, or by value, or by a
-  *                 combination of the three.
-  * @param aggregators An ordered array of aggregators. They are processed in the order specified. The output of an
-  *                    aggregator is passed to the input of the next until all have been processed.
-  * @param limit Limits the number of data points returned from the data store.
-  *              The limit is applied before any aggregator is executed.
-  * @param order Orders the returned data points. This sorting is done before any aggregators are executed.
-  * @param excludeTags By default, the result of the query includes tags and tag values associated with the data points.
-  *                    If `excludeTags` is set to true, the tags will be excluded from the response.
-  * @param plugins optional plugin references to customize the behavior of the query on this metric
-  */
+/**
+ * @param tags Tags narrow down the search. Only metrics that include the tag and matches one of the values are
+ *             returned. Tags are optional.
+ * @param groupBys The resulting data points can be grouped by one or more tags, a time range, or by value, or by a
+ *                 combination of the three.
+ * @param aggregators An ordered array of aggregators. They are processed in the order specified. The output of an
+ *                    aggregator is passed to the input of the next until all have been processed.
+ * @param limit Limits the number of data points returned from the data store.
+ *              The limit is applied before any aggregator is executed.
+ * @param order Orders the returned data points. This sorting is done before any aggregators are executed.
+ * @param excludeTags By default, the result of the query includes tags and tag values associated with the data points.
+ *                    If `excludeTags` is set to true, the tags will be excluded from the response.
+ * @param plugins optional plugin references to customize the behavior of the query on this metric
+ * @param returnIngestionTimestamp If set to true, KairosDB will return ingestion timestamps for the data points.
+ *                                 This requires KairosDB server support for the feature.
+ */
 case class Query(
   metricName: MetricName,
   tags: Seq[QueryTag] = Seq.empty,
@@ -194,29 +212,30 @@ case class Query(
   limit: Option[Int] = None,
   order: Order = Order.defaultOrder,
   excludeTags: Boolean = false,
-  plugins: Seq[QueryPlugin] = Seq.empty)
-
-/** @param timeZone The time zone for the time range of the query. If not specified, UTC is used. tz format, e.g. "Europe/Brussels"
-  * @param cacheTime The amount of time in seconds to re use the cache from a previous query. When a query is made,
-  *                  Kairos looks for the cache file for the query. If a cache file is found and the timestamp of the
-  *                  cache file is within cache_time seconds from the current query, the cache is used.
-  *                  Sending a query with a cacheTime set to 0 will always refresh the cache with new data from Cassandra.
-  * @param plugins optional plugin references to custom the behavior of this query
-  */
+  plugins: Seq[QueryPlugin] = Seq.empty,
+  returnIngestionTimestamp: Boolean = false
+)
+
+/**
+ * @param timeZone The time zone for the time range of the query. If not specified, UTC is used. tz format, e.g. "Europe/Brussels"
+ * @param cacheTime The amount of time in seconds to re use the cache from a previous query. When a query is made,
+ *                  Kairos looks for the cache file for the query. If a cache file is found and the timestamp of the
+ *                  cache file is within cache_time seconds from the current query, the cache is used.
+ *                  Sending a query with a cacheTime set to 0 will always refresh the cache with new data from Cassandra.
+ * @param plugins optional plugin references to custom the behavior of this query
+ */
 case class QueryMetrics(
   metrics: Seq[Query],
   timeSpan: TimeSpan,
   timeZone: Option[String] = None,
   cacheTime: Option[Int] = None,
-  plugins: Seq[QueryPlugin] = Seq.empty)
-
+  plugins: Seq[QueryPlugin] = Seq.empty
+)
 
 /**
-  * Reference to a plugin which can customize the behavior of a query.
-  *
-  * @param name published name of the plugin
-  * @param properties properties for the plugin within the query invocation
-  */
-case class QueryPlugin(
-  name: String,
-  properties: Map[String,Any] = Map.empty)
+ * Reference to a plugin which can customize the behavior of a query.
+ *
+ * @param name published name of the plugin
+ * @param properties properties for the plugin within the query invocation
+ */
+case class QueryPlugin(name: String, properties: Map[String, Any] = Map.empty)