Merge pull request #202 from rabbitmq/md/pt-field-specs

Store `seshat:field_specs/0` in persistent terms

Author: michaelklishin

src/osiris_counters.erl

@@ -20,11 +20,13 @@
 -spec init() -> ok.
 init() ->
     _ = seshat:new_group(osiris),
+    ok = osiris_writer:init_fields_spec(),
+    ok = osiris_replica_reader:init_fields_spec(),
+    ok = osiris_replica:init_fields_spec(),
     ok.
 
--spec new(name(), [{Name :: atom(), Position :: non_neg_integer(),
-                    Type :: atom(), Description :: term()}]) ->
-                 counters:counters_ref().
+-spec new(name(), seshat:fields_spec()) ->
+    counters:counters_ref().
 new(Name, Fields) ->
     seshat:new(osiris, Name, Fields).
 

src/osiris_log.erl

@@ -53,6 +53,7 @@
          evaluate_retention/2,
          directory/1,
          delete_directory/1,
+         counter_fields/0,
          make_counter/1,
          generate_log/4]).
 
@@ -360,7 +361,7 @@
 -type offset() :: osiris:offset().
 -type epoch() :: osiris:epoch().
 -type range() :: empty | {From :: offset(), To :: offset()}.
--type counter_spec() :: {Tag :: term(), Fields :: [atom()]}.
+-type counter_spec() :: {Tag :: term(), Fields :: seshat:fields_spec()}.
 -type chunk_type() ::
     ?CHNK_USER |
     ?CHNK_TRK_DELTA |
@@ -2848,13 +2849,23 @@ validate_crc(ChunkId, Crc, IOData) ->
             exit({crc_validation_failure, {chunk_id, ChunkId}})
     end.
 
+-spec counter_fields() -> seshat:fields_spec().
+counter_fields() ->
+    ?COUNTER_FIELDS.
+
 -spec make_counter(osiris_log:config()) ->
     counters:counters_ref().
 make_counter(#{counter := Counter}) ->
     Counter;
-make_counter(#{counter_spec := {Name, Fields}}) ->
+make_counter(#{counter_spec := {Name, Spec0}}) ->
+    Spec = case Spec0 of
+               {persistent_term, _Key} ->
+                   Spec0;
+               _ when is_list(Spec0) ->
+                   ?COUNTER_FIELDS ++ Spec0
+           end,
     %% create a registered counter
-    osiris_counters:new(Name, ?COUNTER_FIELDS ++ Fields);
+    osiris_counters:new(Name, Spec);
 make_counter(_) ->
     %% if no spec is provided we create a local counter only
     counters:new(?C_NUM_LOG_FIELDS, []).

src/osiris_replica.erl

@@ -33,6 +33,8 @@
          terminate/2,
          code_change/3,
          format_status/1]).
+%% for internal use
+-export([init_fields_spec/0]).
 
 %% holds static or rarely changing fields
 -record(cfg,
@@ -79,6 +81,7 @@
          {packets, ?C_PACKETS, counter, "Number of packets"},
          {readers, ?C_READERS, counter, "Number of readers"},
          {epoch, ?C_EPOCH, counter, "Current epoch"}]).
+-define(FIELDSPEC_KEY, osiris_replica_seshat_fields_spec).
 
 -define(DEFAULT_ONE_TIME_TOKEN_TIMEOUT, 30000).
 -define(TOKEN_SIZE, 32).
@@ -179,12 +182,12 @@ handle_continue(#{name := Name0,
             Transport = application:get_env(osiris, replication_transport, tcp),
             Self = self(),
             CntName = {?MODULE, ExtRef},
+            CntSpec = {CntName, {persistent_term, ?FIELDSPEC_KEY}},
 
             Dir = osiris_log:directory(Config),
             Log = osiris_log:init_acceptor(LeaderRange, LeaderEpochOffs,
                                            Config#{dir => Dir,
-                                                   counter_spec =>
-                                                   {CntName, ?ADD_COUNTER_FIELDS}}),
+                                                   counter_spec => CntSpec}),
             CntRef = osiris_log:counters_ref(Log),
             {NextOffset, LastChunk} = TailInfo = osiris_log:tail_info(Log),
 
@@ -789,3 +792,7 @@ listen(tcp, Port, Options) ->
     gen_tcp:listen(Port, Options);
 listen(ssl, Port, Options) ->
     ssl:listen(Port, Options).
+
+init_fields_spec() ->
+    persistent_term:put(?FIELDSPEC_KEY,
+                        ?ADD_COUNTER_FIELDS ++ osiris_log:counter_fields()).

src/osiris_replica_reader.erl

@@ -27,7 +27,8 @@
          handle_info/2,
          terminate/2,
          code_change/3]).
--export([formatter/1]).
+-export([init_fields_spec/0,
+         formatter/1]).
 
 -record(state,
         {log :: osiris_log:state(),
@@ -48,6 +49,7 @@
          {offset_listeners, ?C_OFFSET_LISTENERS, counter, "Number of offset listeners"}
         ]
        ).
+-define(FIELDSPEC_KEY, osiris_replica_reader_seshat_fields_spec).
 
 %%%===================================================================
 %%% API functions
@@ -122,7 +124,7 @@ init(#{hosts := Hosts,
             ?INFO_(Name, "replica reader successfully connected to host ~0p port ~b",
                    [Host, Port]),
             CntId = {?MODULE, ExtRef, Host, Port},
-            CntSpec = {CntId, ?COUNTER_FIELDS},
+            CntSpec = {CntId, {persistent_term, ?FIELDSPEC_KEY}},
             Config = #{counter_spec => CntSpec, transport => Transport},
             %% send token to replica to complete connection setup
             ok = send(Transport, Sock, Token),
@@ -433,3 +435,6 @@ maybe_add_sni_option(H) when is_list(H) ->
 maybe_add_sni_option(_) ->
     [].
 
+init_fields_spec() ->
+    persistent_term:put(?FIELDSPEC_KEY,
+                        ?COUNTER_FIELDS ++ osiris_log:counter_fields()).

src/osiris_writer.erl

@@ -26,6 +26,7 @@
          read_tracking/1,
          query_writers/2,
          query_replication_state/1,
+         init_fields_spec/0,
          init/1,
          handle_continue/2,
          handle_batch/2,
@@ -49,6 +50,7 @@
          {epoch, ?C_EPOCH, counter, "Current epoch"}
         ]
        ).
+-define(FIELDSPEC_KEY, osiris_writer_seshat_fields_spec).
 
 %% primary osiris process
 %% batch writes incoming data
@@ -179,6 +181,10 @@ query_writers(Pid, QueryFun) ->
 query_replication_state(Pid) when is_pid(Pid) ->
     gen_batch_server:call(Pid, query_replication_state).
 
+init_fields_spec() ->
+    persistent_term:put(?FIELDSPEC_KEY,
+                        osiris_log:counter_fields() ++ ?ADD_COUNTER_FIELDS).
+
 -spec init(osiris:config()) ->
     {ok, undefined, {continue, osiris:config()}}.
 init(#{name := Name0,
@@ -188,10 +194,11 @@ init(#{name := Name0,
     Shared = osiris_log_shared:new(),
     Dir = osiris_log:directory(Config0),
     CntName = {?MODULE, ExtRef},
+    CntSpec = {CntName, {persistent_term, ?FIELDSPEC_KEY}},
     Config = Config0#{name => Name,
                       dir => Dir,
                       shared => Shared,
-                      counter_spec => {CntName, ?ADD_COUNTER_FIELDS}},
+                      counter_spec => CntSpec},
     CntRef = osiris_log:make_counter(Config),
     {ok, undefined, {continue, Config#{counter => CntRef}}}.