opentelemetry

%%%------------------------------------------------------------------------
%% Copyright 2021, OpenTelemetry 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 This Span Processor synchronously exports each ended Span.
%%
%% Use this processor if ending a Span should block until it has been
%% exported. This is useful for cases like a serverless environment where
%% the application will possibly be suspended after handling a request.
%%
%% @end
%%%-----------------------------------------------------------------------
-module(otel_simple_processor).

-behaviour(gen_statem).
-behaviour(otel_span_processor).

-export([start_link/1,
         on_start/3,
         on_end/2,
         force_flush/1,
         report_cb/1,

         %% deprecated
         set_exporter/1,
         set_exporter/2,
         set_exporter/3]).

-export([init/1,
         callback_mode/0,
         idle/3,
         exporting/3,
         terminate/3]).

%% uncomment when OTP-23 becomes the minimum required version
%% -deprecated({set_exporter, 1, "set through the otel_tracer_provider instead"}).
%% -deprecated({set_exporter, 2, "set through the otel_tracer_provider instead"}).
%% -deprecated({set_exporter, 3, "set through the otel_tracer_provider instead"}).

-eqwalizer({nowarn_function, on_end/2}).

-include_lib("opentelemetry_api/include/opentelemetry.hrl").
-include_lib("kernel/include/logger.hrl").
-include("otel_span.hrl").

-record(data, {exporter             :: {module(), term()} | undefined,
               exporter_config      :: {module(), term()} | undefined | none,
               current_from         :: gen_statem:from() | undefined,
               resource             :: otel_resource:t(),
               handed_off_table     :: atom() | undefined,
               runner_pid           :: pid() | undefined,
               exporting_timeout_ms :: integer()}).

-define(DEFAULT_EXPORTER_TIMEOUT_MS, timer:minutes(5)).
-define(NAME_TO_ATOM(Name, Unique), list_to_atom(lists:concat([Name, "_", Unique]))).

%% require a unique name to distinguish multiple simple processors while
%% still having a single name, instead of a possibly changing pid, to
%% communicate with the processor
%% @doc Starts a Simple Span Processor.
%% @end
-spec start_link(#{name := atom() | list()}) -> {ok, pid(), map()}.
start_link(Config=#{name := Name}) ->
    RegisterName = ?NAME_TO_ATOM(?MODULE, Name),
    Config1 = Config#{reg_name => RegisterName},
    {ok, Pid} = gen_statem:start_link({local, RegisterName}, ?MODULE, [Config1], []),
    {ok, Pid, Config1}.

%% @deprecated Please use {@link otel_tracer_provider}
set_exporter(Exporter) ->
    set_exporter(global, Exporter, []).

%% @deprecated Please use {@link otel_tracer_provider}
-spec set_exporter(module(), term()) -> ok.
set_exporter(Exporter, Options) ->
    gen_statem:call(?REG_NAME(global), {set_exporter, {Exporter, Options}}).

%% @deprecated Please use {@link otel_tracer_provider}
-spec set_exporter(atom(), module(), term()) -> ok.
set_exporter(Name, Exporter, Options) ->
    gen_statem:call(?REG_NAME(Name), {set_exporter, {Exporter, Options}}).

%% @private
-spec on_start(otel_ctx:t(), opentelemetry:span(), otel_span_processor:processor_config())
              -> opentelemetry:span().
on_start(_Ctx, Span, _) ->
    Span.

%% @private
-spec on_end(opentelemetry:span(), otel_span_processor:processor_config())
            -> true | dropped | {error, invalid_span} | {error, no_export_buffer}.
on_end(#span{trace_flags=TraceFlags}, _) when not(?IS_SAMPLED(TraceFlags)) ->
    dropped;
on_end(Span=#span{}, #{reg_name := RegName}) ->
    gen_statem:call(RegName, {export, Span});
on_end(_Span, _) ->
    {error, invalid_span}.

%% @private
-spec force_flush(#{reg_name := gen_statem:server_ref()}) -> ok.
force_flush(#{reg_name := RegName}) ->
    gen_statem:cast(RegName, force_flush).

%% @private
init([Args]) ->
    process_flag(trap_exit, true),

    ExportingTimeout = maps:get(exporting_timeout_ms, Args, ?DEFAULT_EXPORTER_TIMEOUT_MS),

    %% TODO: this should be passed in from the tracer server
    Resource = case maps:find(resource, Args) of
                   {ok, R} ->
                       R;
                   error ->
                       otel_resource_detector:get_resource()
               end,
    %% Resource = otel_tracer_provider:resource(),

    {ok, idle, #data{exporter=undefined,
                     exporter_config=maps:get(exporter, Args, none),
                     resource = Resource,
                     handed_off_table=undefined,
                     exporting_timeout_ms=ExportingTimeout},
     [{next_event, internal, init_exporter}]}.

%% @private
callback_mode() ->
    state_functions.

%% @private
idle({call, From}, {export, _Span}, #data{exporter=undefined}) ->
    {keep_state_and_data, [{reply, From, dropped}]};
idle({call, From}, {export, Span}, Data) ->
    {next_state, exporting, Data, [{next_event, internal, {export, From, Span}}]};
idle(EventType, Event, Data) ->
    handle_event_(idle, EventType, Event, Data).

%% @private
exporting({call, _From}, {export, _}, _) ->
    {keep_state_and_data, [postpone]};
exporting(internal, {export, From, Span}, Data=#data{exporting_timeout_ms=ExportingTimeout}) ->
    {OldTableName, RunnerPid} = export_span(Span, Data),
    {keep_state, Data#data{runner_pid=RunnerPid,
                           current_from=From,
                           handed_off_table=OldTableName},
     [{state_timeout, ExportingTimeout, exporting_timeout}]};
exporting(state_timeout, exporting_timeout, Data=#data{current_from=From,
                                                       handed_off_table=_ExportingTable}) ->
    %% kill current exporting process because it is taking too long
    %% which deletes the exporting table, so create a new one and
    %% repeat the state to force another span exporting immediately
    Data1 = kill_runner(Data),
    {next_state, idle, Data1, [{reply, From, {error, timeout}}]};
%% important to verify runner_pid and FromPid are the same in case it was sent
%% after kill_runner was called but before it had done the unlink
exporting(info, {'EXIT', FromPid, _}, Data=#data{runner_pid=FromPid}) ->
    complete_exporting(Data);
%% important to verify runner_pid and FromPid are the same in case it was sent
%% after kill_runner was called but before it had done the unlink
exporting(info, {completed, FromPid}, Data=#data{runner_pid=FromPid}) ->
    complete_exporting(Data);
exporting(EventType, Event, Data) ->
    handle_event_(exporting, EventType, Event, Data).

handle_event_(_, internal, init_exporter, Data=#data{exporter=undefined,
                                                     exporter_config=ExporterConfig}) ->
    Exporter = otel_exporter:init(ExporterConfig),
    {keep_state, Data#data{exporter=Exporter}};
handle_event_(_, {call, From}, {set_exporter, Exporter}, Data=#data{exporter=OldExporter}) ->
    otel_exporter:shutdown(OldExporter),
    {keep_state, Data#data{exporter=otel_exporter:init(Exporter)}, [{reply, From, ok}]};
handle_event_(_, _, _, _) ->
    keep_state_and_data.

%% @private
terminate(_, _, _Data) ->
    ok.

%%

complete_exporting(Data=#data{current_from=From,
                              handed_off_table=ExportingTable})
  when ExportingTable =/= undefined ->
    {next_state, idle, Data#data{current_from=undefined,
                                 runner_pid=undefined,
                                 handed_off_table=undefined},
     [{reply, From, ok}]}.

kill_runner(Data=#data{runner_pid=RunnerPid}) when RunnerPid =/= undefined ->
    Mon = erlang:monitor(process, RunnerPid),
    erlang:unlink(RunnerPid),
    erlang:exit(RunnerPid, kill),
    %% Wait for the runner process termination to be sure that
    %% the export table is destroyed and can be safely recreated
    receive
        {'DOWN', Mon, process, RunnerPid, _} ->
            Data#data{runner_pid=undefined, handed_off_table=undefined}
    end.

new_export_table(Name) ->
     ets:new(Name, [public,
                    {write_concurrency, true},
                    duplicate_bag,
                    %% OpenTelemetry exporter protos group by the
                    %% instrumentation_scope. So using instrumentation_scope
                    %% as the key means we can easily lookup all spans for
                    %% for each instrumentation_scope and export together.
                    {keypos, #span.instrumentation_scope}]).

export_span(Span, #data{exporter=Exporter,
                        resource=Resource}) ->
    Table = new_export_table(otel_simple_processor_table),
    _ = ets:insert(Table, Span),
    Self = self(),
    RunnerPid = erlang:spawn_link(fun() -> send_spans(Self, Resource, Exporter) end),
    ets:give_away(Table, RunnerPid, export),
    {Table, RunnerPid}.

%% Additional benefit of using a separate process is calls to `register` won't
%% timeout if the actual exporting takes longer than the call timeout
send_spans(FromPid, Resource, Exporter) ->
    receive
        {'ETS-TRANSFER', Table, FromPid, export} ->
            export(Exporter, Resource, Table),
            ets:delete(Table),
            completed(FromPid)
    end.

completed(FromPid) ->
    FromPid ! {completed, self()}.

export(undefined, _, _) ->
    true;
export(Exporter, Resource, SpansTid) ->
    %% don't let a exporter exception crash us
    %% and return true if exporter failed
    try
        otel_exporter_traces:export(Exporter, SpansTid, Resource) =:= failed_not_retryable
    catch
        Kind:Reason:StackTrace ->
            ?LOG_INFO(#{source => exporter,
                        during => export,
                        kind => Kind,
                        reason => Reason,
                        exporter => Exporter,
                        stacktrace => StackTrace}, #{report_cb => fun ?MODULE:report_cb/1}),
            true
    end.

%% logger format functions
%% @private
report_cb(#{source := exporter,
            during := export,
            kind := Kind,
            reason := Reason,
            exporter := {ExporterModule, _},
            stacktrace := StackTrace}) ->
    {"exporter threw exception: exporter=~p ~ts",
     [ExporterModule, otel_utils:format_exception(Kind, Reason, StackTrace)]}.