feat: add dogstatsd exporter for metrics

Author: hauleth

README.md

@@ -1,9 +1,55 @@
 oc_datadog
 =====
 
-An OTP library
+[Opencensus][oc] integration to [DataDog][dd] traces and metrics (via dogstatsd).
 
-Build
------
+## Installation
 
-    $ rebar3 compile
+Rebar3:
+
+```erlang
+{deps, [{oc_datadog, "~> 0.1.0"}]}.
+```
+
+Mix:
+
+```elixir
+def deps do
+  [
+    {:oc_datadog, "~> 0.1.0"}
+  ]
+end
+```
+
+## Configuration
+
+`sys.conf`:
+
+```erlang
+[
+    {opencensus, [
+        {reporter, {oc_reporter_datadog, []}},
+        {stat, [
+            {exporters, [
+                {oc_stat_exporter_datadog, []}
+            ]}
+        }]
+    ]}
+].
+```
+
+Elixir:
+
+```elixir
+config :opencensus, :reporter, {:oc_reporter_datadog, []}
+
+config :opencensus, :stat,
+  exporters: [{:oc_stat_exporter_datadog, []}]
+```
+
+## License
+
+See [LICENSE](LICENSE) file.
+
+[oc]: https://github.com/census-instrumentation/opencensus-erlang "Opencensus Erlang"
+[dd]: https://datadog.com "DataDog"

config/dev.conf

@@ -1,5 +1,8 @@
 [
-    {opencensus, [
-        {reporter, {oc_reporter_datadog, []}}
-        ]}
+ {opencensus, [
+               {reporter, {oc_reporter_datadog, []}},
+               {stat, [
+                       {exporters, [{oc_stat_exporter_datadog, []}]}
+                      ]}
+              ]}
 ].

src/oc_datadog.app.src

@@ -5,7 +5,8 @@
   {applications,
    [kernel,
     stdlib,
-    opencensus
+    opencensus,
+    inets
    ]},
   {env,[]},
   {modules, []},

src/oc_reporter_datadog.erl

@@ -1,3 +1,14 @@
+%% @author Łukasz Niemier <lukasz+opensource@niemier.pl>
+%%
+%% @doc Trace reporter for DataDog ([https://datadog.com]).
+%%
+%% == Configuration ==
+%%
+%% <ul>
+%%  <li>`host' - address where DataDog Agent lives (default `"localhost"')</li>
+%%  <li>`port' - port on which Agent listens for traces (default `8126')</li>
+%%  <li>`service' - service name (default `opencensus-app')</li>
+%% </ul>
 -module(oc_reporter_datadog).
 
 -behaviour(oc_reporter).
@@ -8,47 +19,69 @@
 -include_lib("opencensus/include/opencensus.hrl").
 -include_lib("kernel/include/logger.hrl").
 
+-define(TRACER_VERSION, "OC/0.1.0").
+
 -define(DEFAULT_HOST, "localhost").
 -define(DEFAULT_PORT, 8126).
 -define(DEFAULT_SERVICE, <<"opencensus-app">>).
 -define(DEFAULT_TYPE, <<"custom">>).
 
+-spec init(term()) -> oc_reporter:opts().
 init(Options) ->
     Host = proplists:get_value(host, Options, ?DEFAULT_HOST),
     Port = proplists:get_value(port, Options, ?DEFAULT_PORT),
     Service = proplists:get_value(service, Options, ?DEFAULT_SERVICE),
     Type = proplists:get_value(type, Options, ?DEFAULT_TYPE),
     #{host => Host, port => Port, service => Service, type => Type}.
 
-report(Spans, #{service := Service, host := Host, port := Port, type := Type}) ->
+-spec report(nonempty_list(opencensus:span()), oc_reporter:opts()) -> ok.
+report(Spans, #{
+         service := Service,
+         host := Host,
+         port := Port,
+         type := Type}) ->
     Sorted = lists:sort(fun(A, B) ->
                                 A#span.trace_id =< B#span.trace_id
                         end, Spans),
     Grouped = group(Sorted),
     DSpans = [[build_span(S, Service, Type) || S <- Trace] || Trace <- Grouped],
 
-
     try jsx:encode(DSpans) of
         JSON ->
             Address = io_lib:format('http://~s:~B/v0.3/traces', [Host, Port]),
+            Headers = [
+                       {"Datadog-Meta-Lang", "erlang"},
+                       {"Datadog-Meta-Lang-Version", lang_version()},
+                       {"Datadog-Meta-Lang-Interpreter", interpreter_version()},
+                       {"Datadog-Meta-Tracer-Version", ?TRACER_VERSION}
+                      ],
             case httpc:request(
                    put,
-                   {Address, [], "application/json", JSON},
+                   {Address, Headers, "application/json", JSON},
                    [],
                    []
                   ) of
                 {ok, {{_, Code, _}, _, _}} when Code >= 200; Code =< 299 ->
                     ok;
                 {ok, {{_, Code, _}, _, Message}} ->
-                    ?LOG_ERROR("DD: Unable to send spans, DD reported an error: ~p : ~p",
+                    ?LOG_ERROR("DD: Unable to send spans, DD reported an error: ~p: ~p",
                               [Code, Message]);
                 {error, Reason} ->
-                    ?LOG_ERROR("DD: Unable to send spans, client error: ~p", [Reason])
+                    ?LOG_ERROR("DD: Unable to send spans, client error: ~p",
+                               [Reason])
             end
     catch
-        error:_ -> throw(datadog_json_encode_error)
+        error:Error ->
+            ?LOG_ERROR("DD: Can't spans encode to json: ~p", [Error])
     end.
 
+lang_version() ->
+    erlang:system_info(otp_version).
+
+interpreter_version() ->
+    io_lib:format('~s-~s', [erlang:system_info(version),
+                            erlang:system_info(system_architecture)]).
+
 group([]) -> [];
 group([First | Spans]) -> group(Spans, [[First]]).
 
@@ -66,7 +99,8 @@ build_span(Span, Service, Type) ->
       <<"resource">> => Span#span.name,
       <<"service">> => to_tag(Service),
       <<"start">> => wts:to_absolute(Span#span.start_time) * 1000,
-      <<"duration">> => wts:duration(Span#span.start_time, Span#span.end_time) * 1000,
+      <<"duration">> =>
+        wts:duration(Span#span.start_time, Span#span.end_time) * 1000,
       <<"type">> => to_tag(Type),
       <<"meta">> => to_meta(Span#span.attributes)}.
 

src/oc_stat_exporter_datadog.erl

@@ -0,0 +1,86 @@
+%% @author Łukasz Niemier <lukasz+opensource@niemier.pl>
+%%
+%% @doc Trace reporter for DataDog ([https://datadog.com]).
+%%
+%% == Configuration ==
+%%
+%% <ul>
+%%  <li>`host' - address where DataDog Agent lives (default `"localhost"')</li>
+%%  <li>`port' - port on which Agent listens for statsd metrics (default `8125')</li>
+%% </ul>
+-module(oc_stat_exporter_datadog).
+
+-behaviour(oc_stat_exporter).
+
+-export([export/2]).
+
+-define(DEFAULT_HOST, "localhost").
+-define(DEFAULT_PORT, 8125).
+
+-include_lib("kernel/include/logger.hrl").
+
+export(ViewData, Options) ->
+    Host = proplists:get_value(host, Options, ?DEFAULT_HOST),
+    Port = proplists:get_value(port, Options, ?DEFAULT_PORT),
+    {ok, Socket} = gen_udp:open(0, [{active, false}]),
+    Packet = build_packet(ViewData),
+    ok = gen_udp:send(Socket, Host, Port, Packet),
+    ok = gen_udp:close(Socket),
+    ok.
+
+build_packet(Data) when is_list(Data) ->
+    List = [build_packet(Entry) || Entry <- Data],
+    lists:join($\n, List);
+build_packet(#{name := Name,
+               ctags := CTags,
+               tags := Tags,
+               data := #{type := Type,
+                         rows := Rows}}) ->
+    Key = to_key(Name),
+    List = [ build_rows(Key, Type, CTags, Tags, Row) || Row <- Rows ],
+    lists:join($\n, List).
+
+build_tags(Tags, TagsV, CTags) ->
+  build_tags(maps:merge(CTags, maps:from_list(lists:zip(Tags, TagsV)))).
+
+build_tags(Map) when is_map(Map) -> build_tags(maps:to_list(Map));
+build_tags([]) -> [];
+build_tags(Tags) ->
+    List = [[to_key(Key), $:, Value]
+      || {Key, Value} <- Tags],
+    ["|#", lists:join($,, List)].
+
+to_key(Atom) when is_atom(Atom) ->
+    erlang:atom_to_binary(Atom, utf8);
+to_key(Value) -> Value.
+
+build_rows(Name, Type, CTags, Tags, #{tags := TagsV, value := Value}) ->
+    TagList = build_tags(Tags, TagsV, CTags),
+    Metrics = [[Name, Stat] || Stat <- build_row(Type, Value, TagList) ],
+    lists:join($\n, Metrics).
+
+build_row(sum, #{count := Count,
+                 mean := Mean,
+                 sum := Sum}, Tags) ->
+    [
+     [".count:", format_num(Count), "|g", Tags],
+     [".mean:", format_num(Mean), "|g", Tags],
+     [".sum:", format_num(Sum), "|g", Tags]
+    ];
+build_row(distribution, #{buckets := Buckets} = Data, Tags) ->
+    BucketRows = [ bucket_row(Bucket, Tags) || Bucket <- Buckets ],
+    build_row(sum, Data, Tags) ++ BucketRows;
+build_row(_Type, Value, Tags) ->
+    [[$:, format_num(Value), "|g", Tags]].
+
+bucket_row({Bound, Count}, Tags) when is_integer(Count) ->
+    [$:, format_num(Count), "|g", tags_append(Tags, ["le:", format_num(Bound)])].
+
+tags_append([], Value) -> ["|#", Value];
+tags_append(List, Value) -> [List, $,, Value].
+
+format_num(infinity) -> <<"infinity">>;
+format_num(Integer) when is_integer(Integer) ->
+    erlang:integer_to_binary(Integer);
+format_num(Float) when is_float(Float) ->
+    erlang:float_to_binary(Float, [{decimal, 5}, compact]).