fluxter

defmodule Fluxter do
  @moduledoc """
  InfluxDB writer for Elixir that uses InfluxDB's line protocol over UDP.

  To get started with Fluxter, you have to create a module that calls `use
  Fluxter`, like this:

      defmodule MyApp.Fluxter do
        use Fluxter
      end

  This way, `MyApp.Fluxter` becomes an InfluxDB connection pool. Each Fluxter
  pool provides a `start_link/0` function that starts that pool and connects to
  InfluxDB; this function needs to be invoked before being able to send data to
  InfluxDB. Typically, you won't call `start_link/0` directly as you'll want to
  add Fluxter pools to your application's supervision tree; for example:

      def start(_type, _args) do
        import Supervisor.Spec

        children = [
          supervisor(MyApp.Fluxter, []),
          # ...
        ]
        Supervisor.start_link(children, strategy: :one_for_one)
      end

  Once a Fluxter pool is started, its `write/2,3` and `measure/2,3,4` functions
  can successfully be used to send points to the data store. A Fluxter pool
  implements the `Fluxter` behaviour, so you can read documentation for the
  callbacks the behaviour provides to know more about these functions.

  ## Configuration

  Fluxter can be configured either globally or on a per-pool basis.

  The global configuration will affect all Fluxter pools; it can be specified by
  configuring the `:fluxter` application:

      config :fluxter,
        host: "metrics.example.com",
        port: 1122

  The per-pool configuration can be specified by configuring the pool module
  under the `:fluxter` application:

      config :fluxter, MyApp.Fluxter,
        host: "metrics.example.com",
        port: 1122,
        pool_size: 10

  The following is a list of all the supported options:

    * `:host` - (binary) the host to send metrics to. Defaults to `"127.0.0.1"`.
    * `:port` - (integer) the port (on `:host`) to send the metrics to. Defaults
      to `8092`.
    * `:prefix` - (binary or `nil`) all metrics sent to the data store through
      the configured Fluxter pool will be prefixed by the value of this
      option. If `nil`, metrics will not be prefixed. Defaults to `nil`.
    * `:pool_size` - (integer) the size of the connection pool for the given
      Fluxter pool. **This option can only be configured on a per-pool basis**;
      configuring it globally for the `:fluxter` application has no
      effect. Defaults to `5`.

  ## Batching

  Fluxter supports "batching": a batch is a metric aggregator designed to
  locally aggregate an numeric value and flush the aggregated value only once to
  the storage, as a single metric. This is very useful when you have the need to
  write a high number of metrics in a very short amount of time. Doing so can
  have a negative impact on the speed of your code and can also cause network
  packet drops.

  For example, code like the following:

      for i <- 1_000_000 do
        my_operation(i)
        MyApp.Fluxter.write("my_operation_success", [host: "eu-west"], 1)
      end

  can take advantage of batching:

      {:ok, batch} = MyApp.Fluxter.start_batch("my_operation_success", [host: "eu-west"])
      for i <- 1_000_000 do
        my_operation(i)
        MyApp.Fluxter.write_to_batch(batch, 1)
      end
      MyApp.Fluxter.flush_batch(batch)

  """

  @type field_value :: number | boolean | binary
  @type tags :: [{String.Chars.t, String.Chars.t}]
  @type fields :: [{String.Chars.t, field_value}]

  @doc """
  Starts this Fluxter pool.

  A Fluxter pool is a set of processes supervised by a supervisor; this function
  starts all that processes and that supervisor.

  Usually, you'll want to use a Fluxter pool in the supervision tree of your
  application:

      def start(_type, _args) do
        import Supervisor.Spec

        children = [
          supervisor(MyApp.Fluxter, []),
          # ...
        ]
        Supervisor.start_link(children, strategy: :one_for_one)
      end

  """
  @callback start_link() :: Supervisor.on_start

  @doc """
  Writes a metric to the data store.

  `name` is the name of the metric to write. `tags` is a list of key-value pairs
  that specifies tags (as name and value) for the metric to write; note that tag
  values are converted to strings as InfluxDB only support string values for
  tags. `fields` can either be a list of key-value pairs, in which case it
  specifies a list of fields (as name and value), or a single value
  (specifically, a boolean, float, integer, or binary). In the latter case, the
  default field name of `value` will be used: calling `write("foo", [], 4.3)` is
  the same as calling `write("foo", [], value: 4.3)`.

  The return value is always `:ok` as writing is a *fire-and-forget* operation.

  ## Examples

  Assuming a `MyApp.Fluxter` Fluxter pool exists:

      iex> MyApp.Fluxter.write("cpu_temp", [host: "eu-west"], 68)
      :ok

  """
  @callback write(name :: String.Chars.t, tags, field_value | fields) :: :ok

  @doc """
  Should be the same as `write(name, [], fields)`.
  """
  @callback write(name :: String.Chars.t, field_value | fields) :: :ok

  @doc """
  Should be the same as `measure(name, [], [], fun)`.
  """
  @callback measure(name :: String.Chars.t, fun :: (() -> result)) ::
    result when result: any

  @doc """
  Should be the same as `measure(name, tags, [], fun)`.
  """
  @callback measure(name :: String.Chars.t, tags, fun :: (() -> result)) ::
    result when result: any

  @doc """
  Measures the execution time of `fun` and writes it as a metric named `name`.

  This function is just an utility function to measure the execution time of a
  given function `fun`. The `name` and `tags` arguments work in the same way as
  in `c:write/3`.

  `fun`'s execution time is prepended as a field called `value` to the already
  existing list of `fields`. This means that if there's already a field called
  `value` in `fields`, it will be overridden by the measurement. This also means
  that `fields` must be a list of key-value pairs (field name and value): simple
  floats, integers, booleans, and binaries as values for `fields` are not
  supported like they are in `c:write/3`.

  This function returns whatever `fun` returns.

  ## Examples

  Assuming a `MyApp.Fluxter` Fluxter pool exists:

      iex> MyApp.Fluxter.measure "task_exec_time", [host: "us-east"], fn ->
      ...>   1 + 1
      ...> end
      2

  """
  @callback measure(name :: String.Chars.t, tags, fields, fun :: (() -> result)) :: result
    when result: any

  @doc """
  Should be the same as `start_batch(name, [], [])`.
  """
  @callback start_batch(name :: String.Chars.t) :: {:ok, pid}

  @doc """
  Should be the same as `start_batch(name, tags, [])`.
  """
  @callback start_batch(name :: String.Chars.t, tags) :: {:ok, pid}

  @doc """
  Starts a batch for a metric named `name`.

  The purpose of this batch is to aggregate a numeric metric: values aggregated
  in the batch will only be written to the storage as a single metric when the
  batch is "flushed" (see `c:flush_batch/1`). `tags` and `fields` will be tags
  and fields attached to the metric when it's flushed. The aggregated value of
  the metric will be prepended to `fields` as a field called `value`; this means
  that if there's already a field called `value` in `fields`, it will be
  overridden.

  This function returns `{:ok, pid}` where `pid` is the pid of the new batch.

  See the "Batching" section in the documentation for `Fluxter` for more
  information on batches.

  ## Examples

  Assuming a `MyApp.Fluxter` Fluxter pool exists:

      iex> MyApp.Fluxter.start_batch("hits", [host: "us-west"])
      {:ok, #PID<...>}

  """
  @callback start_batch(name :: String.Chars.t, tags, fields) :: {:ok, pid}

  @doc """
  Adds the `extra` value to the given `batch`.

  This function adds the `extra` value (a number) to the current value of the
  given `batch`. To subtract, just use a negative number to add to the current
  value of `batch`.

  This function performs a *fire-and-forget* operation (a cast) on the given
  batch, hence it will always return `:ok`.

  See the "Batching" section in the documentation for `Fluxter` for more
  information on batches.

  ## Examples

  Assuming a `MyApp.Fluxter` Fluxter pool exists:

      iex> MyApp.Fluxter.write_to_batch(batch, 1)
      :ok

  """
  @callback write_to_batch(batch :: pid, extra :: number) :: :ok

  @doc """
  Flushes the given `batch` by writing its aggregated value as a single metric.

  This function performs a *fire-and-forget* operation (a cast) on the given
  batch, hence it will always return `:ok`.

  This function will also stop the `batch` process after the metric is flushed.

  See the "Batching" section in the documentation for `Fluxter` for more
  information on batches.

  ## Examples

  Assuming a `MyApp.Fluxter` Fluxter pool exists:

      iex> MyApp.Fluxter.flush_batch(batch)
      :ok

  """
  @callback flush_batch(batch :: pid) :: :ok

  @doc false
  defmacro __using__(_opts) do
    quote [unquote: false, location: :keep] do
      @behaviour Fluxter

      @pool_size Application.get_env(__MODULE__, :pool_size, 5)
      @worker_names Enum.map(0..(@pool_size - 1), &:'#{__MODULE__}-#{&1}')

      def start_link() do
        import Supervisor.Spec

        {host, port, prefix} = Fluxter.config_for(__MODULE__)
        conn = Fluxter.Conn.new(host, port)
        conn = %{conn | header: [conn.header | prefix]}

        Enum.map(@worker_names, &worker(Fluxter.Conn, [conn, &1], id: &1))
        |> Supervisor.start_link(strategy: :one_for_one)
      end

      @compile {:inline, worker_name: 1}
      for {name, index} <- Enum.with_index(@worker_names) do
        defp worker_name(unquote(index)) do
          unquote(name)
        end
      end

      def write(name, tags \\ [], fields)

      def write(name, tags, fields) when is_list(fields) do
        System.unique_integer([:positive])
        |> rem(@pool_size)
        |> worker_name()
        |> Fluxter.Conn.write(name, tags, fields)
      end

      def write(name, tags, value)
      when is_float(value) or is_integer(value)
      when is_boolean(value) or is_binary(value) do
        write(name, tags, [value: value])
      end

      def measure(name, tags \\ [], fields \\ [], fun)
      when is_function(fun, 0) do
        {elapsed, result} = :timer.tc(fun)
        write(name, tags, [value: elapsed] ++ fields)
        result
      end

      def start_batch(name, tags \\ [], fields \\ []) do
        Fluxter.Batch.start(__MODULE__, name, tags, fields)
      end

      defdelegate write_to_batch(batch, extra), to: Fluxter.Batch, as: :write
      defdelegate flush_batch(batch), to: Fluxter.Batch, as: :flush
    end
  end

  @doc false
  def config_for(module) do
    {loc_env, glob_env} =
      Application.get_all_env(:fluxter)
      |> Keyword.pop(module, [])

    host = loc_env[:host] || glob_env[:host]
    port = loc_env[:port] || glob_env[:port]
    prefix = make_prefix(glob_env[:prefix], loc_env[:prefix])

    {host, port, prefix}
  end

  defp make_prefix(global, local) do
    Enum.map_join([global, local], &(&1 && [&1, ?_]))
  end
end