Change logging API from push to pull

CHANGELOG.md

@@ -21,6 +21,10 @@ road for namespaces. foo.bar.Buz is parsed into [foo, bar, <<"Buz">>] (if foo
 and bar are already existing atoms, but 'Buz' is not).
 - Upgrade grisp dependency to 2.8.0.
 - Add jittered exponential backoff for reconnection.
+- Changed logging API from push to pull. Instead of the client (grisp_connect)
+pushing batches of log event to the server (grisp.io), the server is now pulling
+them with the request `log.get`. In order to synchronize the client ring buffer,
+the server sends `log.sync` notifications.
 
 ## Fixed
 

docs/grisp_connect_api.md

@@ -137,6 +137,34 @@ This should only be called if the new software is functioning as expected.
 </p>
 </details>
 
+<details><summary><code>log.get</code> - retrieve a batch of log entry from the device </summary>
+<p>
+
+**`params`:**
+| key (required *)  | value    | description                            |
+| ----------------- | -------- | -------------------------------------- |
+| `"max_batch_size"`| integer  | Maximum number of events in the result |
+| `"max_byte_size"` | integer  | Maximum byte size of the result        |
+
+**`result`**:  JSON Object
+| key(required *) | value          | description                   |
+|-----------------|----------------|-------------------------------|
+| dropped *       | integer        | Number of dropped log entries |
+| events *        | list of Events | The list of log events        |
+
+**`event format`:**
+Each log events is a list of two elements, first the sequence number of the
+event, and then an object describing the log event with the following fields:
+ - `meta`: meta data of the log entry as an object:
+   - `time`: log time in microseconds.
+   - `file`: `null` or a filename as a string.
+   - `mfa`: `null` or the function the log is from as a list with module name
+            as a tring, function name as a string and arity as an integer.
+ - `msg`: the log entry message, either as a string, or as a json object if it
+          is a report entry.
+ - `level`: the log level as a string.
+
+
 ### Notifications
 
 <details><summary><code>update</code> <code>{"type":"software_update_event"}</code> - notify the current progess of grisp_updater </summary>
@@ -154,6 +182,16 @@ This should only be called if the new software is functioning as expected.
 </p>
 </details>
 
+<details><summary><code>log.sync</code> - synchronize the device log buffer, truncating the entries the server is aware of </summary>
+<p>
+
+**`params`:**
+| key (required *)  | value    | description                                      |
+| ----------------- | -------- | ------------------------------------------------ |
+| `"seq"` *         | integer  | The sequence number of the last stored log event |
+| `"dropped"` *     | integer  | The number of "confirmed dropped log events      |
+
+
 ## Error Codes
 
 ### Default error codes

docs/grisp_connect_architecture.md

@@ -6,10 +6,8 @@ graph TD
 
     subgraph GrispConnectApp[Grisp Connect Application]
         GrispConnectRootSup[Root Supervisor<br>grisp_connect_sup]
-        GrispConnectLogServer[Log Server<br>grisp_connect_log_server]
         GrispConnectClient[Client<br>grisp_connect_client]
 
-        GrispConnectRootSup --Supervise--> GrispConnectLogServer
         GrispConnectRootSup --Supervise--> GrispConnectClient
         GrispConnectClient --Spawn and Monitor--> JarlConnection
     end

src/grisp_connect.app.src

@@ -26,7 +26,6 @@
         {ws_request_timeout, 5_000},
         {ws_ping_timeout, 60_000},
         {ws_max_retries, infinity},
-        {logs_interval, 2_000},
         {logs_batch_size, 100},
         {logger, [
             % Enable our own default handler,

src/grisp_connect_api.erl

@@ -19,15 +19,19 @@
   when Msg :: {request, Method :: jarl:method(), Params :: map() | list(), ReqRef :: binary() | integer()}
             | {notification, jarl:method(), Params :: map() | list()}.
 handle_msg({notification, M, Params}) ->
-    ?LOG_ERROR("Received unexpected notification ~p: ~p", [M, Params]),
-    ok;
-handle_msg({request, M, Params, ID})
-  when M == [?method_post]; M == [?method_get] ->
+    handle_notification(M, Params);
+handle_msg({request, M, Params, ID}) ->
     handle_request(M, Params, ID).
 
 
 %--- Internal Funcitons --------------------------------------------------------
 
+handle_notification([log, sync], Params) ->
+    grisp_connect_log:sync(Params);
+handle_notification(Method, Params) ->
+    ?LOG_ERROR("Received unexpected notification ~p: ~p", [Method, Params]),
+    ok.
+
 handle_request([?method_get], #{type := <<"system_info">>} = _Params, ID) ->
     Info = grisp_connect_updater:system_info(),
     {reply, Info, ID};
@@ -73,5 +77,8 @@ handle_request([?method_post], #{type := <<"cancel">>}, ID) ->
         ok ->
             {reply, ok, ID}
     end;
-handle_request(_T, _P, ID) ->
+handle_request([log, get], Params, ID) ->
+    {reply, grisp_connect_log:get(Params), ID};
+handle_request(Method, Params, ID) ->
+    ?LOG_ERROR("Received unexpected request ~p: ~p", [Method, Params]),
     {error, method_not_found, undefined, undefined, ID}.

src/grisp_connect_client.erl

@@ -213,7 +213,6 @@ connecting(info, {jarl, Conn, {connected, _}}, Data = #data{conn = Conn}) ->
 connected(enter, _OldState, Data = #data{wait_calls = WaitCalls}) ->
     % When entering connected, we reply to all wait_connected calls with ok
     gen_statem:reply([{reply, F, ok} || F <- WaitCalls]),
-    grisp_connect_log_server:start(),
     {keep_state, Data#data{wait_calls = [], last_error = undefined}};
 connected({call, From}, is_connected, _) ->
     {keep_state_and_data, [{reply, From, true}]};
@@ -371,13 +370,11 @@ conn_start(Data = #data{conn = undefined,
 conn_close(Data = #data{conn = undefined}, _Reason) ->
     Data;
 conn_close(Data = #data{conn = Conn}, _Reason) ->
-    grisp_connect_log_server:stop(),
     jarl:disconnect(Conn),
     Data#data{conn = undefined}.
 
 % Safe to call in any state
 conn_died(Data) ->
-    grisp_connect_log_server:stop(),
     Data#data{conn = undefined}.
 
 -spec conn_request(data(), jarl:method(), atom(), map(),

src/grisp_connect_log.erl

@@ -0,0 +1,91 @@
+%% @doc Utility module to handle logs.
+%% @end
+-module(grisp_connect_log).
+
+-include_lib("kernel/include/logger.hrl").
+
+% API functions
+-export([get/1]).
+-export([sync/1]).
+
+%--- Macros --------------------------------------------------------------------
+
+% FixMe:
+% Sending over ~30_000 bytes over WS breaks rtems I/O driver.
+% We want avoid to return chunks that are bigger then that.
+-define(MAX_CHUNK_BYTES, 30_000).
+
+%--- Types ---------------------------------------------------------------------
+
+-type get_options() :: #{
+    max_batch_size => non_neg_integer(),
+    max_byte_size => non_neg_integer()
+}.
+
+-type sync_options() :: #{
+    seq := non_neg_integer(),
+    dropped => non_neg_integer()
+}.
+
+
+%--- API Functions -------------------------------------------------------------
+
+-spec get(Opts :: get_options()) ->
+    #{dropped := non_neg_integer(), events := list()}.
+get(Opts) ->
+    {ok, DefaultSize} = application:get_env(grisp_connect, logs_batch_size),
+    BatchSize = maps:get(max_batch_size, Opts, DefaultSize),
+    ByteSize = min(maps:get(max_byte_size, Opts, ?MAX_CHUNK_BYTES), ?MAX_CHUNK_BYTES),
+    {Events, Dropped} = grisp_connect_logger_bin:chunk(BatchSize, ByteSize),
+    #{events => [[Seq, jsonify(E)] || {Seq, E} <- Events],
+      dropped => Dropped}.
+
+-spec sync(Opts :: sync_options()) -> ok.
+sync(#{seq := Seq, dropped := Dropped}) ->
+    grisp_connect_logger_bin:sync(Seq, Dropped).
+
+
+%--- Internal Functions --------------------------------------------------------
+
+jsonify(Event) ->
+    jsonify_meta(jsonify_msg(binary_to_term(base64:decode(Event)))).
+
+jsonify_msg(#{msg := {string, String}} = Event) ->
+    maps:put(msg, unicode:characters_to_binary(String), Event);
+jsonify_msg(#{msg := {report, Report}} = Event) ->
+    case is_json_compatible(Report) of
+        true ->
+            maps:put(msg, Report, Event);
+        false ->
+            String = unicode:characters_to_binary(
+                       io_lib:format("[JSON incompatible term]~n~tp", [Report])
+                      ),
+            maps:put(msg, String, Event)
+    end;
+jsonify_msg(#{msg := {FormatString, Term}} = Event) ->
+    %FIXME: scan format and ensure unicode encoding
+    String = unicode:characters_to_binary(io_lib:format(FormatString, Term)),
+    maps:put(msg, String, Event).
+
+jsonify_meta(#{meta := Meta} = Event) ->
+    MFA = case maps:is_key(mfa, Meta) of
+              true ->
+                  {M, F, A} = maps:get(mfa, Meta),
+                  [M, F, A];
+              false ->
+                  null
+          end,
+    File = case maps:is_key(file, Meta) of
+               true  -> unicode:characters_to_binary(maps:get(file, Meta));
+               false -> null
+           end,
+    Default = #{mfa => MFA, file => File},
+    Optional = maps:without(maps:keys(Default), Meta),
+    FilterFun = fun(Key, Value) -> jsx:is_term(#{Key => Value}) end,
+    maps:put(meta, maps:merge(maps:filter(FilterFun, Optional), Default), Event).
+
+is_json_compatible(Term) ->
+    try jsx:is_term(Term)
+    catch error:_ ->
+        false
+    end.

src/grisp_connect_sup.erl

@@ -38,13 +38,7 @@ init([]) ->
         {ok, true} -> [worker(grisp_connect_ntp, [])];
         {ok, false} -> []
     end,
-%% Notes:
-%% grisp_connect_log_server is required to be running by grisp_connect_client
-%% that starts and stops the logging loop in grisp_connect_log_server asynchronous.
-%% Hence grisp_connect_log_server should be started before grisp_connect_client
-%% and a crash in grisp_connect_log_server should crash grisp_connect_client as well.
     ChildSpecs = NTP ++ [
-        worker(grisp_connect_log_server, []),
         worker(grisp_connect_client, [])
     ],
     {ok, {SupFlags, ChildSpecs}}.