opencensus_zipkin

%%%------------------------------------------------------------------------
%% Copyright 2018, OpenCensus Authors
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%
%% @doc Exports spans to Zipkin
%% @end
%%%------------------------------------------------------------------------


-module(oc_reporter_zipkin).

-include_lib("opencensus/include/opencensus.hrl").

-behaviour(oc_reporter).

-ifdef(OTP_RELEASE).
-include_lib("kernel/include/logger.hrl").
-else.
-define(LOG_INFO(Format, Args), error_logger:info_msg(Format, Args)).
-define(LOG_ERROR(Format, Args), error_logger:error_msg(Format, Args)).
-endif.

-define(DEFAULT_ZIPKIN_ADDRESS, "http://localhost:9411/api/v2/spans").
-define(DEFAULT_LOCAL_ENDPOINT, #{<<"serviceName">> => node()}).

-export([init/1,
         report/2]).

init(Opts) ->
    Address = zipkin_address(Opts),
    LocalEndpoint = local_endpoint(Opts),
    {Address, LocalEndpoint}.

report(Spans, {Address, LocalEndpoint}) ->
    ZSpans = [zipkin_span(Span, LocalEndpoint) || Span <- Spans],

    try jsx:encode(ZSpans) of
        JSON ->
            case httpc:request(post, {Address, [], "application/json", JSON}, [], []) of
                {ok, {{_, 202, _}, _, _}} ->
                    ok;
                {ok, {{_, 200, _}, _, _}} ->
                    ok;
                {ok, {{_, Code, _}, _, Message}} ->
                    ?LOG_ERROR("Zipkin: Unable to send spans, Zipkin reported an error: ~p : ~p",
                              [Code, Message]);

                {error, Reason} ->
                    ?LOG_ERROR("Zipkin: Unable to send spans, client error: ~p", [Reason])
            end
    catch
        error:Error ->
            ?LOG_ERROR("Zipkin: Can't spans encode to json: ~p", [Error])
    end.

zipkin_span(Span, LocalEndpoint) ->
    (optional_fields(Span))#{
       <<"traceId">> => iolist_to_binary(io_lib:format("~32.16.0b", [Span#span.trace_id])),
       <<"name">> => iolist_to_binary(Span#span.name),
       <<"id">> => iolist_to_binary(io_lib:format("~16.16.0b", [Span#span.span_id])),
       <<"timestamp">> => wts:to_absolute(Span#span.start_time),
       <<"duration">> => wts:duration(Span#span.start_time, Span#span.end_time),
       <<"debug">> => false, %% TODO: get from attributes?
       <<"shared">> => false, %% TODO: get from attributes?
       <<"localEndpoint">> => LocalEndpoint,
       %% <<"remoteEndpoint">> =>  %% TODO: get from attributes?
       <<"annotations">> => to_annotations(Span#span.time_events),
       <<"tags">> => to_tags(Span#span.attributes) %% TODO: merge with oc_tags?
     }.

to_annotations(TimeEvents) ->
    to_annotations(TimeEvents, []).

to_annotations([], Annotations) ->
    Annotations;
to_annotations([{Timestamp, #annotation{description=Description,
                                        attributes=Attributes}} | Rest], Annotations) ->
    to_annotations(Rest, [#{<<"timestamp">> => wts:to_absolute(Timestamp),
                            <<"value">> => annotation_value(Description, Attributes)} | Annotations]);
to_annotations([{Timestamp, MessageEvent=#message_event{}} | Rest], Annotations) ->
    to_annotations(Rest, [#{<<"timestamp">> => wts:to_absolute(Timestamp),
                            <<"value">> => annotation_value(MessageEvent)} | Annotations]).

annotation_value(Description, Attributes) ->
    AttrString = lists:join(", ", [[Key, "=", to_string(Value)] ||
                                      {Key, Value} <- maps:to_list(Attributes)]),
    iolist_to_binary([Description, " Attributes:{", AttrString, "}"]).

annotation_value(#message_event{type=Type,
                                id=Id,
                                uncompressed_size=UncompressedSize,
                                compressed_size=CompressedSize}) ->
    iolist_to_binary(["MessageEvent:{type=", atom_to_binary(Type, utf8),
                      ", id=", integer_to_binary(Id),
                      ", uncompressed_size=", integer_to_binary(UncompressedSize),
                      ", compressed_size=", integer_to_binary(CompressedSize), "}"]).


to_string(Value) when is_function(Value) ->
    to_string(Value());
to_string(Value) when is_list(Value) ;
                      is_binary(Value) ->
    Value;
to_string(Value) ->
    io_lib:format("~p", [Value]).

to_binary_string(Name, Value) when is_function(Value) ->
    to_binary_string(Name, Value());
to_binary_string(_Name, Value) when is_list(Value) ->
    list_to_binary(Value);
to_binary_string(_Name, Value) when is_integer(Value) ->
    integer_to_binary(Value);
to_binary_string(_Name, Value) when is_float(Value) ->
    float_to_binary(Value);
to_binary_string(_Name, Value) when is_atom(Value) ->
    atom_to_binary(Value, utf8);
to_binary_string(_Name, Value) ->
    Value.

to_tags(Attributes) ->
    maps:map(fun(Name, Value) ->
                     to_binary_string(Name, Value)
             end, Attributes).

zipkin_address(Options) ->
    proplists:get_value(address, Options, ?DEFAULT_ZIPKIN_ADDRESS).

local_endpoint(Options) ->
    proplists:get_value(local_endpoint, Options, ?DEFAULT_LOCAL_ENDPOINT).

optional_fields(Span) ->
    lists:foldl(fun(Field, Acc) ->
                        case span_field(Field, Span) of
                            undefined ->
                                Acc;
                            Value ->
                                maps:put(Field, Value, Acc)
                        end
                end, #{}, [<<"kind">>, <<"parentId">>]).

span_field(<<"parentId">>, #span{parent_span_id=undefined}) ->
    undefined;
span_field(<<"parentId">>, #span{parent_span_id=ParentId}) ->
    iolist_to_binary(io_lib:format("~16.16.0b", [ParentId]));
span_field(<<"kind">>, #span{kind=?SPAN_KIND_UNSPECIFIED}) ->
    undefined;
span_field(<<"kind">>, #span{kind=?SPAN_KIND_SERVER}) ->
    <<"SERVER">>;
span_field(<<"kind">>, #span{kind=?SPAN_KIND_CLIENT}) ->
    <<"CLIENT">>.