defmodule Instream.Series do
@moduledoc """
## Series Definition
Series definitions can be used to have a fixed structured usable for
reading and writing data to an InfluxDB server:
defmodule MySeries.CPULoad do
use Instream.Series
series do
measurement "cpu_load"
tag :host, default: "www"
tag :core
field :value, default: 100
field :value_desc
end
end
The macros `tag/2` and `field/2` both accept a keyword tuple with a
`:default` entry. This value will be pre-assigned when using the data
struct with all other fields or tags being set to `nil`.
### Structs
Each of your series definitions will register three separate structs.
Based on the aforementioned `MySeries.CPULoad` you will have access
to the following structs:
%MySeries.CPULoad{
fields: %MySeries.CPULoad.Fields{value: 100, value_desc: nil},
tags: %MySeries.CPULoad.Tags{host: "www", core: nil},
timestamp: nil
}
`:timestamp` is expected to be either a
unix nanosecond or an RFC3339 timestamp.
### Compile-Time Series Validation
Defining a series triggers a validation function during compilation.
This validation for example prevents the usage of a field and tag sharing
the same name. Some internal keys like `:time` will also raise an
`ArgumentError` during compilation.
You can deactivate this compile time validation by passing
`skip_validation: true` in your series module:
defmodule MySeries.ConflictButAccepted do
use Instream.Series, skip_validation: true
series do
tag :conflict
field :conflict
end
end
Validations performed:
- having `use Instream.Series` requires also calling `series do .. end`
- a measurement must be defined
- at least one field must be defined
- fields and tags must not share a name
- the names `:_field`, `:_measurement`, and `:time` are not allowed to be
used for fields or tags
## Reading Series Points (Hydration)
Whenever you want to convert a plain map or a query result into a specific
series you can use the built-in hydration methods:
MySeries.from_map(%{
timestamp: 1_234_567_890,
some_tag: "hydrate",
some_field: 123
})
~S(SELECT * FROM "my_measurement")
|> MyConnection.query()
|> MySeries.from_result()
The timestamp itself is kept "as is" for integer values, timestamps in
RFC3339 format (e.g. `"1970-01-01T01:00:00.000+01:00"`) will be converted
to `:nanosecond` integer values.
Please be aware that when using an `OTP` release prior to `21.0` the time
will be truncated to `:microsecond` precision due to
`:calendar.rfc3339_to_system_time/2` not being available and
`DateTime.from_iso8601/1` only supporting microseconds.
### Hydrating Results with Multiple Fields (InfluxDB v2.x Pivoting)
If you are connecting to an InfluxDB v2.x instance and have more than one
field in your series you will need to "pivot" your query results to be able
to fully hydrate your structs:
MyConnection.query(~s[
from(bucket: "\#{MyConnection.config(:bucket)}")
|> range(start: -5m)
|> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
])
Without using the `pivot()` function in your `:flux` query you will receive
one result struct for each field (other fields containing default values)
instead of one struct with all fields.
See https://www.influxdata.com/blog/how-to-pivot-data-flux-columnar-data/ for
more details and examples.
## Writing Series Points
You can then use your series module to assemble a data point (one at a time)
for writing:
data = %MySeries{}
data = %{data | fields: %{data.fields | value: 17}}
data = %{data | tags: %{data.tags | bar: "bar", foo: "foo"}}
And then write one or many at once:
MyConnection.write(point)
MyConnection.write([point_1, point_2, point_3])
If you want to pass an explicit timestamp you can use the key `:timestamp`:
data = %MySeries{}
data = %{data | timestamp: 1_439_587_926_000_000_000}
The timestamp is (by default) expected to be a nanosecond unix timestamp.
To use different precision (for all points in this write operation!) you can
change this value by modifying your write call:
data = %MySeries{}
data = %{data | timestamp: 1_439_587_926}
MyConnection.write(data, precision: :second)
Supported precision types are:
- `:hour`
- `:minute`
- `:second`
- `:millisecond`
- `:microsecond`
- `:nanosecond`
- `:rfc3339`
Please be aware that the UDP protocol writer (`Instream.Writer.UDP`) does
not support custom timestamp precisions. All UDP timestamps are implicitly
expected to already be at nanosecond precision.
"""
alias Instream.Series.Hydrator
alias Instream.Series.Validator
defmacro __using__(opts) do
quote location: :keep do
unless unquote(opts[:skip_validation]) do
@after_compile unquote(__MODULE__)
end
import unquote(__MODULE__), only: [series: 1]
end
end
defmacro __after_compile__(%{module: module}, _bytecode) do
Validator.proper_series?(module)
end
@doc """
Defines the series.
"""
defmacro series(do: block) do
quote location: :keep do
@behaviour unquote(__MODULE__)
@measurement nil
Module.register_attribute(__MODULE__, :fields_raw, accumulate: true)
Module.register_attribute(__MODULE__, :tags_raw, accumulate: true)
try do
# scoped import
import unquote(__MODULE__)
unquote(block)
after
:ok
end
@fields_struct Enum.sort(@fields_raw, &unquote(__MODULE__).__sort_fields__/2)
@tags_struct Enum.sort(@tags_raw, &unquote(__MODULE__).__sort_tags__/2)
@impl unquote(__MODULE__)
def __meta__(:fields), do: Keyword.keys(@fields_struct)
def __meta__(:measurement), do: @measurement
def __meta__(:tags), do: Keyword.keys(@tags_struct)
Module.eval_quoted(__ENV__, [
unquote(__MODULE__).__struct_fields__(@fields_struct),
unquote(__MODULE__).__struct_tags__(@tags_struct)
])
Module.eval_quoted(__ENV__, [
unquote(__MODULE__).__struct__(__MODULE__)
])
@impl unquote(__MODULE__)
def from_map(data), do: Hydrator.from_map(__MODULE__, data)
@impl unquote(__MODULE__)
def from_result(data), do: Hydrator.from_result(__MODULE__, data)
end
end
@doc """
Provides additional metadata for a series.
## Available Information
- `:fields`: the fields in the series
- `:measurement`: the measurement of the series
- `:tags`: the available tags defining the series
"""
@callback __meta__(:field | :measurement | :tags) :: any
@doc """
Creates a series dataset from any given map.
Keys not defined in the series are silently dropped.
"""
@callback from_map(map) :: struct
@doc """
Creates a list of series datasets from a query result.
Keys not defined in the series are silently dropped.
"""
@callback from_result(map | [map]) :: [struct]
@doc """
Defines a field in the series.
"""
defmacro field(name, opts \\ []) do
quote do
@fields_raw {unquote(name), unquote(opts[:default])}
end
end
@doc """
Defines the measurement of the series.
"""
defmacro measurement(name) do
quote do
@measurement unquote(name)
end
end
@doc """
Defines a tag in the series.
"""
defmacro tag(name, opts \\ []) do
quote do
@tags_raw {unquote(name), unquote(opts[:default])}
end
end
@doc false
def __sort_fields__({left, _}, {right, _}), do: left < right
@doc false
def __sort_tags__({left, _}, {right, _}), do: left < right
@doc false
def __struct__(series) do
quote do
@type t :: %unquote(series){
fields: %unquote(series).Fields{},
tags: %unquote(series).Tags{},
timestamp: non_neg_integer | binary | nil
}
defstruct fields: %unquote(series).Fields{},
tags: %unquote(series).Tags{},
timestamp: nil
end
end
@doc false
def __struct_fields__(fields) do
quote do
defmodule Fields do
@moduledoc false
defstruct unquote(Macro.escape(fields))
end
end
end
@doc false
def __struct_tags__(tags) do
quote do
defmodule Tags do
@moduledoc false
defstruct unquote(Macro.escape(tags))
end
end
end
end
