init

Author: NeilMenne

.formatter.exs

@@ -0,0 +1,10 @@
+[
+  inputs: [
+    "test/**/*.{ex,exs}",
+    "lib/**/*.{ex,exs}",
+    "config/**/*.exs",
+    "rel/**/*.exs",
+    "mix.exs"
+  ],
+  line_length: 80,
+]

.gitignore

@@ -0,0 +1,20 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where 3rd-party dependencies like ExDoc output generated docs.
+/doc/
+
+# Ignore .fetch files in case you like to edit your project deps locally.
+/.fetch
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez

README.md

@@ -0,0 +1,21 @@
+# Inflex
+
+**TODO: Add description**
+
+## Installation
+
+If [available in Hex](https://hex.pm/docs/publish), the package can be installed
+by adding `inflex` to your list of dependencies in `mix.exs`:
+
+```elixir
+def deps do
+  [
+    {:inflex, "~> 0.1.0"}
+  ]
+end
+```
+
+Documentation can be generated with [ExDoc](https://github.com/elixir-lang/ex_doc)
+and published on [HexDocs](https://hexdocs.pm). Once published, the docs can
+be found at [https://hexdocs.pm/inflex](https://hexdocs.pm/inflex).
+

config/config.exs

@@ -0,0 +1,2 @@
+use Mix.Config
+import_config "#{Mix.env()}.exs"

config/dev.exs

@@ -0,0 +1 @@
+use Mix.Config

config/test.exs

@@ -0,0 +1,10 @@
+use Mix.Config
+
+config :logger, level: :warn
+
+config :inflex, Inflex.TestDatabase,
+  host: "localhost",
+  udp_port: 8089,
+  batch_size: 5,
+  max_queue_size: 100,
+  flush_interval: 1

lib/application.ex

@@ -0,0 +1,23 @@
+defmodule Inflex.Application do
+  @moduledoc """
+  Inflex's per-node registry w/in the library's exported application
+
+  Inflex uses `Registry` to support database-name-based dispatch as a primary
+  design goal was multi-tenancy or sharding support for non-enterprise use
+  cases.
+  """
+
+  use Application
+
+  def start(_type, _args) do
+    import Supervisor.Spec
+
+    children = [
+      supervisor(Registry, [:unique, Inflex.Registry])
+    ]
+
+    opts = [strategy: :one_for_one, name: Inflex.Application]
+
+    Supervisor.start_link(children, opts)
+  end
+end

lib/conn/udp.ex

@@ -0,0 +1,44 @@
+defmodule Inflex.Conn.UDP do
+  @moduledoc """
+  Light, opinionated wrapper around `:gen_udp`
+  """
+
+  @spec open(
+          port :: :inet.port_number() | integer(),
+          udp_opts :: [:gen_udp.option()]
+        ) :: :gen_udp.socket()
+  @doc """
+  open a port for UDP with all of the configuration afforded to you by
+  `:gen_udp.open/2`
+  """
+  def open(port \\ 0, udp_opts \\ [:binary, {:active, false}]) do
+    {:ok, socket} = :gen_udp.open(port, udp_opts)
+    socket
+  end
+
+  @spec close(socket :: :gen_udp.socket()) :: :ok
+  @doc """
+  when possible, sockets should be closed as a well-behaved program
+  """
+  def close(socket) do
+    :gen_udp.close(socket)
+  end
+
+  @spec write(
+          socket :: :gen_udp.socket(),
+          host :: :inet.ip_address(),
+          port :: :inet.port_number() | integer(),
+          String.t()
+        ) :: :ok | {:error, reason :: any()}
+  @doc """
+  convert the string to utf-8 binary and send via udp
+  """
+  def write(socket, host, port, payload) do
+    :gen_udp.send(
+      socket,
+      host,
+      port,
+      :unicode.characters_to_binary(payload)
+    )
+  end
+end

lib/database.ex

@@ -0,0 +1,64 @@
+defmodule Inflex.Database do
+  @moduledoc """
+  Configuration of a single database's connection pool, queue and supervision
+  tree. Since InfluxDB uses a single UDP port on the server per database, this
+  library's goal is to make it easy to do 1:1 mappings of a pool of UDP
+  connections to that destination port and batching of datapoints.
+
+  To get started, create your own module and do something like:
+
+      use Inflex.Database, otp_app: :your_app, database: "your_db_name"
+
+  In your application, simply add `YourApp.YourInflexDatabase` to the children:
+
+      children = [
+        ...,
+        YourApp.YourInflexDatabase
+      ]
+
+  The database name isn't used for the UDP connection in anyway, but it is used
+  to create database-specific workers. To support querying interfaces and HTTP
+  based sending, the database would need to be known. To support these features
+  in the future, the database name is held by all workers in their
+  configuration.
+
+  Where possible, `Inflex.Database.Supervisor` provides sane defaults for
+  configuration options to make it as close as working out of the box as
+  possible. A full list of configuration options and their types can be found
+  there.
+  """
+
+  defmacro __using__(opts) do
+    quote bind_quoted: [opts: opts] do
+      @otp_app Keyword.fetch!(opts, :otp_app)
+      @database Keyword.fetch!(opts, :database)
+
+      alias Inflex.Database.QueueWorker
+      alias Inflex.Database.Supervisor, as: DBSupervisor
+
+      def child_spec(opts) do
+        %{
+          id: __MODULE__,
+          start: {__MODULE__, :start_link, [opts]},
+          type: :supervisor
+        }
+      end
+
+      def start_link(opts \\ []) do
+        opts = Keyword.put(opts, :database, @database)
+        DBSupervisor.start_link(__MODULE__, @otp_app, opts)
+      end
+
+      def push(%{} = point) do
+        GenServer.cast(QueueWorker.via_tuple(@database), {:push, point})
+      end
+
+      defoverridable child_spec: 1
+    end
+  end
+
+  @doc """
+  queue a single data point (may trigger a batch being sent asynchronously)
+  """
+  @callback push(point :: Inflex.Point.t() | map()) :: :ok
+end

lib/database/pool_worker.ex

@@ -0,0 +1,133 @@
+defmodule Inflex.Database.PoolWorker do
+  @moduledoc """
+  single worker responsible for sending stats to the configured database
+  """
+
+  use GenServer
+
+  require Logger
+
+  alias Inflex.Conn.UDP
+  alias Inflex.LineProtocol
+
+  def child_spec(opts) do
+    :poolboy.child_spec(:worker, poolboy_config(opts), opts)
+  end
+
+  defp poolboy_config(opts) do
+    [
+      {:name, opts |> Map.fetch!(:database) |> via_tuple()},
+      {:worker_module, __MODULE__},
+      {:size, Map.get(opts, :pool_size)},
+      {:max_overflow, Map.get(opts, :pool_overflow)}
+    ]
+  end
+
+  @doc false
+  def start_link(opts) do
+    GenServer.start_link(__MODULE__, opts)
+  end
+
+  @doc """
+  single pool worker initialization with deferred UDP socket setup
+  """
+  def init(opts) do
+    Process.send(self(), {:init, opts}, [])
+    {:ok, %{}}
+  end
+
+  @spec worker_pid(database :: String.t()) :: pid() | :full
+  @doc """
+  Perform a non-blocking checkout of an available worker for sending data. If no
+  worker is available, `:poolboy.checkout/2` will return `:full`
+  """
+  def worker_pid(database) do
+    :poolboy.checkout(
+      via_tuple(database),
+      false
+    )
+  end
+
+  @spec send_points(pid(), points :: [map() | Inflex.Point.t()]) :: :ok
+  @doc """
+  Given an identified worker via `worker_pid/1` and points, asynchronously ship
+  the points to influx.
+  """
+  def send_points(pid, points), do: GenServer.cast(pid, {:send, points})
+
+  @doc false
+  def handle_cast({:send, points}, state) do
+    send_batch(points, state)
+    {:noreply, state}
+  end
+
+  defp send_batch(points, state) do
+    payload =
+      points
+      |> Stream.map(&LineProtocol.encode/1)
+      |> Enum.join("\n")
+
+    UDP.write(
+      state.udp_socket,
+      state.config.host,
+      state.config.udp_port,
+      payload
+    )
+  rescue
+    e ->
+      Logger.debug(fn ->
+        {
+          Exception.format(:error, e, System.stacktrace()),
+          [points: inspect(points)]
+        }
+      end)
+  after
+    # regardless of formatting errors, invalid types, etc, check the worker
+    # back into the pool
+    :poolboy.checkin(via_tuple(state.config.database), self())
+  end
+
+  def handle_info({:init, opts}, _state) do
+    udp_opts =
+      Map.get(
+        opts,
+        :udp_conn_opts
+      )
+
+    socket = UDP.open(0, udp_opts)
+
+    config =
+      opts
+      |> Map.new()
+      |> prepare_host()
+
+    {:noreply, %{config: config, udp_socket: socket}}
+  end
+
+  defp prepare_host(state),
+    do: Map.update(state, :host, "", &String.to_charlist/1)
+
+  @doc false
+  def via_tuple(database) do
+    {:via, Registry, {Inflex.Registry, database <> "_pool"}}
+  end
+
+  def create_point(input) do
+    %{
+      measurement: "measure",
+      fields: %{
+        input: input,
+        value: :rand.uniform(10)
+      },
+      tags: %{
+        tag0:
+          if :rand.uniform() > 0.5 do
+            "val0"
+          else
+            "val1"
+          end
+      },
+      timestamp: System.os_time(:nanosecond)
+    }
+  end
+end