Skip to content

support ingestion timestamp returned by cassandra #294

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Sep 29, 2025
Merged

support ingestion timestamp returned by cassandra #294

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
00f07e4
support ingestion timestamp returned by cassandra
ramazanyich Sep 26, 2025
bf1b494
revert it test
ramazanyich Sep 26, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/it/resources/conf/auth/auth.props
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1 +1 @@
test: test
test: test
32 changes: 16 additions & 16 deletions src/it/scala/integration/AddAndQueryDataPointsIntegrationSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))
)
)
)
Expand Down Expand Up @@ -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)
)
)
)
Expand Down Expand Up @@ -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)
)
)
)
Expand All @@ -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)
)
)
)
Expand All @@ -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)
)
)
)
Expand Down
6 changes: 3 additions & 3 deletions src/it/scala/integration/DeleteDataPointsByTagIntegrationSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
)
)
)
Expand Down
4 changes: 2 additions & 2 deletions src/it/scala/integration/StringDataPointsIntegrationSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))
)
)
)
Expand Down Expand Up @@ -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))
)
)
)
Expand Down
195 changes: 107 additions & 88 deletions src/main/scala/io/waylay/kairosdb/driver/models/Models.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand All @@ -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 */
Expand All @@ -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

Expand All @@ -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 */
Expand All @@ -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
}

Expand All @@ -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,
Expand All @@ -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)
Loading