defmodule ExDataDog do use GenServer @default_port 8125 @default_host "127.0.0.1" @default_namespace nil @default_sink nil @timing_stub 1.234 @type statsd_port :: number @type host :: String.t @type sink :: String.t @type name :: String.t @type namespace :: String.t @type options :: [ port: statsd_port, host: host, namespace: namespace, sink: sink, name: name ] def start_link(options \\ []) do state = %{ port: Keyword.get(options, :port, Application.get_env(:ex_data_dog, :port, @default_port)), host: Keyword.get(options, :host, Application.get_env(:ex_data_dog, :host, @default_host)) |> parse_host(), namespace: Keyword.get(options, :namespace, Application.get_env(:ex_data_dog, :namespace, @default_namespace)), sink: Keyword.get(options, :sink, Application.get_env(:ex_data_dog, :sink, @default_sink)), socket: nil } GenServer.start_link(__MODULE__, state, Keyword.merge([name: __MODULE__], options)) end def stop(name \\__MODULE__), do: GenServer.stop(name) def flush(name \\__MODULE__), do: GenServer.call(name, :flush) defp parse_host(host) when is_binary(host) do case host |> to_char_list() |> :inet.parse_address() do {:error, _} -> host |> String.to_atom() {:ok, address} -> address end end def counter(amount, metric, options \\ default_options()) do sampling options, fn(decision) -> case decision do {:sample, rate} -> transmit({metric, amount, :c}, options, rate) amount _ -> amount end end end def count(collection, metric, options \\ default_options()) do sampling options, fn(decision) -> case decision do {:sample, rate} -> transmit({metric, Enum.count(collection), :c}, options, rate) collection _ -> collection end end end def increment(metric, options \\ default_options()), do: counter(1, metric, options) def decrement(metric, options \\ default_options()), do: counter(-1, metric, options) def gauge(amount, metric, options \\ [tags: []]) do transmit({metric, amount, :g}, options) amount end def set(member, metric, options \\ [tags: []]) do transmit({metric, member, :s}, options) member end def timer(amount, metric, options \\ default_options()) do sampling options, fn(decision) -> case decision do {:sample, rate} -> transmit({metric, amount, :ms}, options, rate) amount _ -> amount end end end def timing(metric, fun, options \\ default_options()) do sampling options, fn(decision) -> case decision do {:sample, rate} -> {time, value} = :timer.tc(fun) amount = if Application.get_env(:ex_data_dog, :test_mode, false), do: @timing_stub, else: time / 1000.0 transmit({metric, amount, :ms}, options, rate) value _ -> fun.() end end end @doc """ Record a histogram value (DogStatsD-only). * `sample_rate`: Limit how often the metric is collected * `tags`: Add tags to entry (DogStatsD-only) It returns the value given as the first argument, making it suitable for pipelining. """ def histogram(amount, metric, options \\ default_options()) do sampling options, fn(decision) -> case decision do {:sample, rate} -> transmit({metric, amount, :h}, options, rate) amount _ -> amount end end end @doc """ Time a function using a histogram metric (DogStatsD-only). * `sample_rate`: Limit how often the metric is collected * `tags`: Add tags to entry (DogStatsD-only) It returns the result of the function call, making it suitable for pipelining. """ def histogram_timing(metric, fun, options \\ default_options()) do sampling options, fn(decision) -> case decision do {:sample, rate} -> {time, value} = :timer.tc(fun) amount = if Application.get_env(:ex_data_dog, :test_mode, false), do: @timing_stub, else: time / 1000.0 transmit({metric, amount, :h}, options, rate) value _ -> fun.() end end end defp default_options do [ sample_rate: 1, tags: [], name: __MODULE__, ] end @doc """ Emit event. `text` supports line breaks, only first 4KB will be transmitted. Available options: * `tags`: Add tags to entry (DogStatsD-only) * `priority`: Can be *normal* or *low*, default *normal* * `alert_type`: Can be *error*, *warning*, *info* or *success*, default *info* * `aggregation_key`: Assign an aggregation key to the event, to group it with some others * `hostname`: Assign a hostname to the event * `source_type_name`: Assign a source type to the event * `date_happened`: Assign a timestamp to the event, default current time It returns the title of the event, making it suitable for pipelining. """ def event(title, text \\ "", options \\ [tags: []]) do transmit({:event, title, text, options}, options) title end defp sampling(options, fun) when is_list(options) do case Keyword.get(options, :sample_rate, 1) do 1 -> fun.({:sample, 1}) sample_rate -> sample(sample_rate, fun) end end defp sample(sample_rate, fun) do case :rand.uniform <= sample_rate do true -> fun.({:sample, sample_rate}) _ -> fun.(:no_sample) end end defp transmit(message, options), do: transmit(message, options, 1) defp transmit(message, options, sample_rate) do name = Keyword.get(options, :name, __MODULE__) GenServer.cast(name, {:transmit, message, options, sample_rate}) end defp packet({:event, title, text, opts}, namespace, tags, sample_rate) do text = text |> String.replace("\n","\\n") |> String.slice(0, 4096) [ stat_name(:event, namespace), ":#{title}|#{text}", opts[:priority] && "|p:#{opts[:priority]}" || "", opts[:alert_type] && "|t:#{opts[:alert_type]}" || "", opts[:source_type_name] && "|s:#{opts[:source_type_name]}" || "", opts[:aggregation_key] && "|k:#{opts[:aggregation_key]}" || "", opts[:hostname] && "|h:#{opts[:hostname]}" || "", opts[:date_happened] && "|d:#{opts[:date_happened]}" || "", sample_rate_suffix(sample_rate), tags_suffix(tags) ] end defp packet({key, value, type}, namespace, tags, sample_rate) do [ stat_name(key, namespace), ":#{value}|#{type}", sample_rate_suffix(sample_rate), tags_suffix(tags) ] end defp sample_rate_suffix(1), do: "" defp sample_rate_suffix(sample_rate) do [ "|@", :io_lib.format('~.2f', [sample_rate]), ] end defp tags_suffix([]), do: "" defp tags_suffix(tags) do [ "|#", Enum.join(tags, ",") ] end defp stat_name(key, nil), do: key defp stat_name(key, namespace), do: "#{namespace}.#{key}" def handle_cast({:transmit, message, options, sample_rate}, %{sink: sink} = state) when is_list(sink) do tags = Keyword.get(options, :tags, []) pkt = message |> packet(state.namespace, tags, sample_rate) |> IO.iodata_to_binary() {:noreply, %{state | sink: [pkt | sink]}} end def handle_cast({:transmit, message, options, sample_rate}, state) do tags = Keyword.get(options, :tags, []) pkt = message |> packet(state.namespace, tags, sample_rate) |> IO.iodata_to_binary() {:ok, socket} = :gen_udp.open(0, [:binary]) :gen_udp.send(socket, state.host, state.port, pkt) :gen_udp.close(socket) {:noreply, state} end def handle_call(:flush, _from, state), do: {:reply, :ok, state} end