process tracing refactoring

elvis.config

@@ -13,7 +13,7 @@
                     {elvis_style, macro_module_names},
                     {elvis_style, operator_spaces, #{rules => [{right, ","}, {right, "++"}, {left, "++"}]}},
                     {elvis_style, nesting_level, #{level => 4}},
-                    {elvis_style, no_if_expression},
+                    {elvis_style, no_if_expression, #{ignore => [prg_pg_backend, prg_utils]}},
                     %% FIXME Implement appropriate behaviours
                     {elvis_style, invalid_dynamic_call, #{
                         ignore => [prg_storage, prg_worker_sidecar]

include/progressor.hrl

@@ -25,17 +25,17 @@
     last_event_id => event_id(),
     initialization => task_id(),
     previous_status => process_status(),
-    status_changed_at => timestamp_sec()
+    status_changed_at => timestamp_us()
 }.
 
 -type task() :: #{
     task_id => task_id(),
     process_id := id(),
     task_type := task_type(),
     status := task_status(),
-    scheduled_time := timestamp_sec(),
-    running_time => timestamp_sec(),
-    finished_time => timestamp_sec(),
+    scheduled_time := timestamp_us(),
+    running_time => timestamp_us(),
+    finished_time => timestamp_us(),
     args => binary(),
     metadata => map(),
     idempotency_key => binary(),
@@ -59,11 +59,12 @@
     task_id := task_id(),
     task_type := task_type(),
     task_status := task_status(),
-    scheduled := timestamp_sec(),
-    running => timestamp_sec(),
-    finished => timestamp_sec(),
+    scheduled := timestamp_us(),
+    running => timestamp_us(),
+    finished => timestamp_us(),
     args => binary(),
     metadata => map(),
+    context => binary(),
     idempotency_key => binary(),
     response => term(),
     retry_interval => non_neg_integer(),
@@ -190,10 +191,13 @@
 -type task_result() :: #{
     task_id := task_id(),
     status := task_status(),
-    finished_time => timestamp_sec(),
+    running_time => timestamp_us(),
+    finished_time => timestamp_us(),
     response => binary()
 }.
 
+%% microsecond
+-type timestamp_us() :: non_neg_integer().
 -type timestamp_ms() :: non_neg_integer().
 -type timestamp_sec() :: non_neg_integer().
 -type timeout_sec() :: non_neg_integer().
@@ -221,6 +225,6 @@
     process_id => ProcessId,
     status => <<"init">>,
     previous_status => <<"init">>,
-    created_at => erlang:system_time(second),
-    status_changed_at => erlang:system_time(second)
+    created_at => erlang:system_time(microsecond),
+    status_changed_at => erlang:system_time(microsecond)
 }).

src/prg_utils.erl

@@ -12,6 +12,20 @@
 -export([with_observe/5]).
 -export([with_span/2]).
 
+%% Time conversion
+-export([detect_unit/1]).
+-export([to_microseconds/1]).
+-export([to_seconds/1]).
+-export([split_timestamp/1]).
+-export([format_microseconds/1]).
+
+-type time_unit() :: second | millisecond | microsecond.
+
+%% Boundaries based on unix epoch (~5138 year)
+-define(MAX_SECONDS, 100000000000).
+-define(MAX_MILLISECONDS, 100000000000000).
+-define(MAX_MICROSECONDS, 100000000000000000).
+
 -spec registered_name(atom(), string()) -> atom().
 registered_name(BaseAtom, PostfixStr) ->
     erlang:list_to_atom(erlang:atom_to_list(BaseAtom) ++ PostfixStr).
@@ -91,3 +105,39 @@ collect(histogram, MetricKey, Labels, Value) ->
 %%collect(_, _MetricKey, _Labels, _Value) ->
 %%    %% TODO implement it
 %%    ok.
+
+-spec detect_unit(non_neg_integer()) -> time_unit() | no_return().
+detect_unit(Ts) when Ts < ?MAX_SECONDS -> second;
+detect_unit(Ts) when Ts < ?MAX_MILLISECONDS -> millisecond;
+detect_unit(Ts) when Ts < ?MAX_MICROSECONDS -> microsecond;
+detect_unit(Ts) -> error({unsupported_time_unit, Ts}).
+
+-spec to_microseconds(non_neg_integer()) -> non_neg_integer() | no_return().
+to_microseconds(Ts) ->
+    case detect_unit(Ts) of
+        second -> Ts * 1000000;
+        millisecond -> Ts * 1000;
+        microsecond -> Ts
+    end.
+
+-spec to_seconds(non_neg_integer()) -> non_neg_integer() | no_return().
+to_seconds(Ts) ->
+    case detect_unit(Ts) of
+        second -> Ts;
+        millisecond -> Ts div 1000;
+        microsecond -> Ts div 1000000
+    end.
+
+-spec split_timestamp(non_neg_integer()) -> {Seconds :: non_neg_integer(), MicroPart :: non_neg_integer()}.
+split_timestamp(Ts) ->
+    case detect_unit(Ts) of
+        second -> {Ts, 0};
+        millisecond -> {Ts div 1000, (Ts rem 1000) * 1000};
+        microsecond -> {Ts div 1000000, Ts rem 1000000}
+    end.
+
+-spec format_microseconds(non_neg_integer()) -> binary().
+format_microseconds(Val) ->
+    Bin = integer_to_binary(Val),
+    Pad = 6 - byte_size(Bin),
+    <<(binary:copy(<<"0">>, Pad))/binary, Bin/binary>>.