Epic/process tracing (#196)

* WIP: add handle functions

* add tracing endpoints

* fixed

* bump progressor

* cleanup

* fix typo

* fix timers

* bump progressor

* cleanup

* bump progressor

* fix issues

---------

Co-authored-by: ttt161 <losto@nix>
Co-authored-by: Артем <WWW_cool@inbox.ru>

Author: 3 people

apps/hellgate/src/hellgate.erl

@@ -64,6 +64,7 @@ get_api_child_spec(MachineHandlers, Opts) ->
             construct_health_routes(readiness, genlib_app:env(?MODULE, health_check_readiness, #{})),
     EventHandlerOpts = genlib_app:env(?MODULE, scoper_event_handler_options, #{}),
     PrometeusRoute = get_prometheus_route(),
+    ProcessTracingRoute = hg_progressor_handler:get_routes(),
     woody_server:child_spec(
         ?MODULE,
         #{
@@ -78,7 +79,7 @@ get_api_child_spec(MachineHandlers, Opts) ->
                     construct_service_handler(invoice_templating, hg_invoice_template, Opts),
                     construct_service_handler(proxy_host_provider, hg_proxy_host_provider, Opts)
                 ],
-            additional_routes => [PrometeusRoute | HealthRoutes],
+            additional_routes => [PrometeusRoute | HealthRoutes] ++ ProcessTracingRoute,
             shutdown_timeout => genlib_app:env(?MODULE, shutdown_timeout, 0)
         }
     ).

apps/hellgate/src/hg_invoice.erl

@@ -40,6 +40,7 @@
 -export([get_payment_opts/1]).
 -export([create/6]).
 -export([marshal_invoice/1]).
+-export([unmarshal_invoice/1]).
 -export([unmarshal_history/1]).
 -export([collapse_history/1]).
 

apps/hellgate/src/hg_invoice_template.erl

@@ -26,6 +26,7 @@
 %% API
 
 -export([get/1]).
+-export([unmarshal_invoice_template_params/1]).
 
 -type tpl_id() :: dmsl_domain_thrift:'InvoiceTemplateID'().
 -type tpl() :: dmsl_domain_thrift:'InvoiceTemplate'().

apps/hellgate/test/hg_invoice_lite_tests_SUITE.erl

@@ -27,6 +27,7 @@
 -export([payment_success_empty_cvv/1]).
 -export([payment_has_optional_fields/1]).
 -export([payment_last_trx_correct/1]).
+-export([payment_success_trace/1]).
 
 -type config() :: hg_ct_helper:config().
 -type test_case_name() :: hg_ct_helper:test_case_name().
@@ -63,6 +64,7 @@ groups() ->
         {payments, [parallel], [
             payment_start_idempotency,
             payment_success,
+            payment_success_trace,
             payment_w_first_blacklisted_success,
             payment_w_all_blacklisted,
             register_payment_success,
@@ -241,6 +243,7 @@ payment_success(C) ->
         },
         Payment
     ),
+
     ?assertMatch(
         #domain_TransactionInfo{
             extra = #{
@@ -250,6 +253,158 @@ payment_success(C) ->
         Trx
     ).
 
+-spec payment_success_trace(config()) -> test_return().
+payment_success_trace(C) ->
+    Client = cfg(client, C),
+    InvoiceID = start_invoice(<<"rubberduck">>, make_due_date(10), 42000, C),
+    Context = #base_Content{
+        type = <<"application/x-erlang-binary">>,
+        data = erlang:term_to_binary({you, 643, "not", [<<"welcome">>, here]})
+    },
+    PayerSessionInfo = #domain_PayerSessionInfo{
+        redirect_url = <<"https://redirectly.io/merchant">>
+    },
+    PaymentParams = (make_payment_params(?pmt_sys(<<"visa-ref">>)))#payproc_InvoicePaymentParams{
+        payer_session_info = PayerSessionInfo,
+        context = Context
+    },
+    PaymentID = process_payment(InvoiceID, PaymentParams, Client),
+    PaymentID = await_payment_capture(InvoiceID, PaymentID, Client),
+
+    RootUrl = unicode:characters_to_binary(cfg(root_url, C)),
+    UrlInternal = <<RootUrl/binary, "/traces/internal/invoice/", InvoiceID/binary>>,
+    UrlJaeger = <<RootUrl/binary, "/traces/jaeger/invoice/", InvoiceID/binary>>,
+    {ok, _Status, _Headers, RefInternal} = hackney:get(UrlInternal),
+    {ok, BodyInternal} = hackney:body(RefInternal),
+    [
+        #{
+            <<"args">> := #{
+                <<"content_type">> := <<"thrift_call">>,
+                <<"content">> := #{
+                    <<"call">> := #{
+                        <<"function">> := <<"Create">>,
+                        <<"service">> := <<"Invoicing">>
+                    },
+                    <<"params">> := _
+                }
+            },
+            <<"error">> := null,
+            <<"events">> := [
+                #{
+                    <<"event_id">> := 1,
+                    <<"event_payload">> := _,
+                    <<"event_timestamp">> := _
+                }
+            ],
+            <<"finished">> := _,
+            <<"otel_trace_id">> := _,
+            <<"retry_attempts">> := 0,
+            <<"retry_interval">> := 0,
+            <<"running">> := _,
+            <<"scheduled">> := _,
+            <<"task_id">> := _,
+            <<"task_metadata">> := #{<<"range">> := #{}},
+            <<"task_status">> := <<"finished">>,
+            <<"task_type">> := <<"init">>
+        },
+        #{<<"task_type">> := <<"call">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"finished">>},
+        #{<<"task_type">> := <<"timeout">>, <<"task_status">> := <<"cancelled">>}
+    ] = json:decode(BodyInternal),
+    {ok, _Status2, _Headers2, RefJaeger} = hackney:get(UrlJaeger),
+    {ok, BodyJaeger} = hackney:body(RefJaeger),
+    #{
+        <<"data">> := [
+            #{
+                <<"traceId">> := _,
+                <<"processes">> := #{
+                    InvoiceID := #{
+                        <<"service_name">> := <<"hellgate_invoice">>,
+                        <<"tags">> := []
+                    }
+                },
+                <<"spans">> := [
+                    #{
+                        <<"operationName">> := <<"init">>,
+                        <<"process">> := #{
+                            <<"service_name">> := <<"hellgate_invoice">>,
+                            <<"tags">> := []
+                        },
+                        <<"processID">> := InvoiceID,
+                        <<"spanId">> := _,
+                        <<"traceId">> := _,
+                        <<"startTime">> := _,
+                        <<"duration">> := _,
+                        <<"tags">> := [
+                            #{
+                                <<"key">> := <<"task.status">>,
+                                <<"type">> := <<"string">>,
+                                <<"value">> := <<"finished">>
+                            },
+                            #{
+                                <<"key">> := <<"task.retries">>,
+                                <<"type">> := <<"int64">>,
+                                <<"value">> := 0
+                            },
+                            #{
+                                <<"key">> := <<"task.input">>,
+                                <<"type">> := <<"string">>,
+                                <<"value">> := _NestedJsonArgs
+                            }
+                        ],
+                        <<"logs">> := [
+                            #{
+                                <<"timestamp">> := _,
+                                <<"fields">> := [
+                                    #{
+                                        <<"key">> := <<"event.id">>,
+                                        <<"type">> := <<"int64">>,
+                                        <<"value">> := 1
+                                    },
+                                    #{
+                                        <<"key">> := <<"event.payload">>,
+                                        <<"type">> := <<"string">>,
+                                        <<"value">> := _NestedJsonEvent
+                                    }
+                                ]
+                            }
+                        ]
+                    },
+                    #{<<"operationName">> := <<"call">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>},
+                    #{<<"operationName">> := <<"timeout">>}
+                ]
+            }
+        ]
+    } = json:decode(BodyJaeger),
+    BadInvoiceUrl = <<RootUrl/binary, "/traces/internal/invoice/UnknownInvoice">>,
+    {ok, 404, _, _} = hackney:get(BadInvoiceUrl),
+    BadFormatUrl = <<RootUrl/binary, "/traces/external/invoice/", InvoiceID/binary>>,
+    {ok, 400, _, _} = hackney:get(BadFormatUrl),
+    ok.
+
 -spec payment_w_first_blacklisted_success(config()) -> test_return().
 payment_w_first_blacklisted_success(C) ->
     Client = cfg(client, C),

apps/hg_progressor/src/hg_progressor.erl

@@ -273,7 +273,7 @@ marshal(status, {<<"running">>, _Detail}) ->
 marshal(status, {<<"error">>, Detail}) ->
     {'failed', #mg_stateproc_MachineStatusFailed{reason = Detail}};
 marshal(timestamp, Timestamp) ->
-    unicode:characters_to_binary(calendar:system_time_to_rfc3339(Timestamp, [{offset, "Z"}]));
+    unicode:characters_to_binary(calendar:system_time_to_rfc3339(Timestamp, [{offset, "Z"}, {unit, microsecond}]));
 marshal(term, Term) ->
     binary_to_term(Term);
 marshal(function, init) ->
@@ -315,7 +315,7 @@ unmarshal(events, {undefined, _ID}) ->
 unmarshal(events, {[], _}) ->
     [];
 unmarshal(events, {Events, LastEventID}) ->
-    Ts = erlang:system_time(second),
+    Ts = erlang:system_time(microsecond),
     lists:foldl(
         fun(#mg_stateproc_Content{format_version = Ver, data = Payload}, Acc) ->
             PrevID =
@@ -349,7 +349,7 @@ unmarshal(action, #mg_stateproc_ComplexAction{
     remove = RemoveAction
 }) when Timeout =/= undefined ->
     genlib_map:compact(#{
-        set_timer => erlang:system_time(second) + Timeout,
+        set_timer => erlang:system_time(microsecond) + (Timeout * 1000000),
         remove => maybe_unmarshal(remove_action, RemoveAction)
     });
 unmarshal(action, #mg_stateproc_ComplexAction{timer = {unset_timer, #'mg_stateproc_UnsetTimerAction'{}}}) ->
@@ -359,9 +359,9 @@ unmarshal(action, #mg_stateproc_ComplexAction{remove = #mg_stateproc_RemoveActio
 unmarshal(action, #mg_stateproc_ComplexAction{remove = undefined}) ->
     undefined;
 unmarshal(timer, {deadline, DateTimeRFC3339}) ->
-    calendar:rfc3339_to_system_time(unicode:characters_to_list(DateTimeRFC3339), [{unit, second}]);
+    calendar:rfc3339_to_system_time(unicode:characters_to_list(DateTimeRFC3339), [{unit, microsecond}]);
 unmarshal(timer, {timeout, Timeout}) ->
-    erlang:system_time(second) + Timeout;
+    erlang:system_time(microsecond) + (Timeout * 1000000);
 unmarshal(term, Term) ->
     erlang:term_to_binary(Term);
 unmarshal(remove_action, #mg_stateproc_RemoveAction{}) ->