Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat(ds): move session data from mnesia to emqx_ds #12675

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
+964 −53
Open

feat(ds): move session data from mnesia to emqx_ds #12675

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3666e5f
feat(ds): move session data from `mnesia` to `emqx_ds`
thalesmg Mar 7, 2024
5b31eff
feat(schema_hooks): introduce new API for union option injections
thalesmg Mar 8, 2024
ab20cae
test(ds): fix flaky test
thalesmg Mar 15, 2024
b8968f9
feat: allow defining fixed timestamp per message
thalesmg Mar 28, 2024
fe56326
test: use simpler cleanup
thalesmg Mar 28, 2024
f3d54c8
test: use better mocking decision
thalesmg Mar 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 30 additions & 4 deletions apps/emqx/integration_test/emqx_persistent_session_ds_SUITE.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ all() ->
emqx_common_test_helpers:all(?MODULE).

init_per_suite(Config) ->
emqx_common_test_helpers:clear_screen(),
TCApps = emqx_cth_suite:start(
app_specs(),
#{work_dir => emqx_cth_suite:work_dir(Config)}
Expand Down Expand Up @@ -199,6 +200,22 @@ force_last_alive_at(ClientId, Time) ->
_ = emqx_persistent_session_ds_state:commit(S),
ok.

%% Helper for determining if rpc faults should be injected.
%% When storing session data in DS, failing here could lead to a session being considered
%% new and not loaded, since we read non-atomically from the DB. If getting the streams
%% fail due to rpc problems, it leniently returns an empty list of streams, and that
%% results in a "new" session.
is_restoring_session() ->
{current_stacktrace, Stacktrace} = process_info(self(), current_stacktrace),
is_restoring_session(Stacktrace).

is_restoring_session([]) ->
false;
is_restoring_session([{emqx_persistent_session_ds, session_open, _, _} | _Rest]) ->
true;
is_restoring_session([_ | Rest]) ->
is_restoring_session(Rest).

%%------------------------------------------------------------------------------
%% Testcases
%%------------------------------------------------------------------------------
Expand Down Expand Up @@ -649,7 +666,7 @@ t_session_replay_retry(_Config) ->
ClientsPub
),

Pubs0 = emqx_common_test_helpers:wait_publishes(NClients, 5_000),
Pubs0 = emqx_common_test_helpers:wait_publishes(NClients, 6_000),
NPubs = length(Pubs0),
?assertEqual(NClients, NPubs, ?drainMailbox(1_500)),

Expand All @@ -658,9 +675,18 @@ t_session_replay_retry(_Config) ->
%% Make `emqx_ds` believe that roughly half of the shards are unavailable.
ok = emqx_ds_test_helpers:mock_rpc_result(
fun(_Node, emqx_ds_replication_layer, _Function, [_DB, Shard | _]) ->
case erlang:phash2(Shard) rem 2 of
0 -> unavailable;
1 -> passthrough
%% When storing session data in DS, failing here could lead to a session being
%% considered new and not loaded, since we read non-atomically from the DB.
%% If getting the streams fail due to rpc problems, it leniently returns an
%% empty list of streams, and that results in a "new" session.
case is_restoring_session() of
true ->
passthrough;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This sounds weird, though I have to admit I haven't yet finished reviewing.

I think that read failure should not be equivalent to having no session record. Is there any good reason to do that?

On the test code: wouldn't it be simpler to match on DB and simulate failure only for the message DB? Current test code looks both super-hacky and extemely fragile.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that read failure should not be equivalent to having no session record. Is there any good reason to do that?

Just that I'm not sure how we should handle (recoverable) failures in this case. 🤔
Sleep and retry?

On the test code: wouldn't it be simpler to match on DB and simulate failure only for the message DB? Current test code looks both super-hacky and extemely fragile.

Sounds like a much better idea, I'll fix it.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sleep and retry?

Good question. This warrants some discussion but as a safe bet I'd propagate this failure to the client, and let them decide if they want to retry.

Also wondering what would happen now if locking is unavailable in a mria transaction, probably something effectively similar?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good question. This warrants some discussion but as a safe bet I'd propagate this failure to the client, and let them decide if they want to retry.

This comes from get_streams being lenient with failures, so it could affect other portions of the code as well.

Also wondering what would happen now if locking is unavailable in a mria transaction, probably something effectively similar?

Not sure I understand the question, but indeed not having transactions will lead to potential inconsistencies (at least in the builtin backend; FDB could avoid them, at least if deletions are not involved).

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This comes from get_streams being lenient with failures

Oh I see now. TBH this is (again) feels like a hack: we probably don't want to hide failures on that level.

Not sure I understand the question

I mean, I'm trying to understand what would happen to the client which is trying to open an existing session when current storage (mria) is unavailable.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mean, I'm trying to understand what would happen to the client which is trying to open an existing session when current storage (mria) is unavailable.

Good question, not sure on the exact answer. Since it uses dirty reads and writes, my guess is that it'll work with the local copy of the data (maybe hang in replicants while they can't RPC a core). 🤔

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed the test code to match on the DB.

false ->
case erlang:phash2(Shard) rem 2 of
0 -> unavailable;
1 -> passthrough
end
end
end
),
Expand Down
44 changes: 40 additions & 4 deletions apps/emqx/src/emqx_ds_schema.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,13 +80,22 @@ namespace() ->
schema() ->
[
{messages,
ds_schema(#{
ds_schema(ref(builtin), 'durable_storage.messages', #{
default =>
#{
<<"backend">> => builtin
},
importance => ?IMPORTANCE_MEDIUM,
desc => ?DESC(messages)
})},
{sessions,
ds_schema(ref(builtin_session), 'durable_storage.sessions', #{
default =>
#{
<<"backend">> => builtin
},
importance => ?IMPORTANCE_MEDIUM,
desc => ?DESC(sessions)
})}
].

Expand Down Expand Up @@ -179,6 +188,19 @@ fields(builtin) ->
}
)}
];
fields(builtin_session) ->
lists:map(
fun
({layout = Key, Sc0}) ->
Override = #{type => hoconsc:union(builtin_layouts_session())},
{Key, hocon_schema:override(Sc0, Override)};
(Field) ->
Field
end,
fields(builtin)
);
fields(layout_builtin_wildcard_optimized_session) ->
fields(layout_builtin_wildcard_optimized);
fields(builtin_local_write_buffer) ->
[
{max_items,
Expand Down Expand Up @@ -253,36 +275,50 @@ fields(layout_builtin_reference) ->

desc(builtin) ->
?DESC(builtin);
desc(builtin_session) ->
?DESC(builtin);
desc(builtin_local_write_buffer) ->
?DESC(builtin_local_write_buffer);
desc(layout_builtin_wildcard_optimized) ->
?DESC(layout_builtin_wildcard_optimized);
desc(layout_builtin_wildcard_optimized_session) ->
?DESC(layout_builtin_wildcard_optimized);
desc(_) ->
undefined.

%%================================================================================
%% Internal functions
%%================================================================================

ds_schema(Options) ->
ds_schema(MainRef, InjectionPoint, Options) ->
sc(
hoconsc:union([
ref(builtin)
| emqx_schema_hooks:injection_point('durable_storage.backends', [])
MainRef
| emqx_schema_hooks:union_option_injections(InjectionPoint, [])
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This addresses this comment: https://github.com/emqx/emqx-platform/pull/135#discussion_r1514765105

]),
Options
).

-ifndef(TEST).
builtin_layouts() ->
[ref(layout_builtin_wildcard_optimized)].

builtin_layouts_session() ->
[ref(layout_builtin_wildcard_optimized_session)].
-else.
builtin_layouts() ->
%% Reference layout stores everything in one stream, so it's not
%% suitable for production use. However, it's very simple and
%% produces a very predictabale replay order, which can be useful
%% for testing and debugging:
[ref(layout_builtin_wildcard_optimized), ref(layout_builtin_reference)].

builtin_layouts_session() ->
%% Reference layout stores everything in one stream, so it's not
%% suitable for production use. However, it's very simple and
%% produces a very predictabale replay order, which can be useful
%% for testing and debugging:
[ref(layout_builtin_wildcard_optimized_session), ref(layout_builtin_reference)].
-endif.

sc(Type, Meta) -> hoconsc:mk(Type, Meta).
Expand Down
11 changes: 10 additions & 1 deletion apps/emqx/src/emqx_persistent_message.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ init() ->
Backend = storage_backend(),
ok = emqx_ds:open_db(?PERSISTENT_MESSAGE_DB, Backend),
ok = emqx_persistent_session_ds_router:init_tables(),
ok = emqx_persistent_session_ds:create_tables(),
ok = initialize_session_ds_state(),
ok
end).

Expand All @@ -72,6 +72,15 @@ is_persistence_enabled(Zone) ->
storage_backend() ->
storage_backend([durable_storage, messages]).

-ifdef(STORE_STATE_IN_DS).
initialize_session_ds_state() ->
ok = emqx_persistent_session_ds_state:open_db(storage_backend([durable_storage, sessions])).
-else.
initialize_session_ds_state() ->
ok = emqx_persistent_session_ds_state:create_tables().
%% -ifdef(STORE_STATE_IN_DS).
-endif.

%% Dev-only option: force all messages to go through
%% `emqx_persistent_session_ds':
-spec force_ds(emqx_types:zone()) -> boolean().
Expand Down
5 changes: 1 addition & 4 deletions apps/emqx/src/emqx_persistent_session_ds.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@
]).

%% session table operations
-export([create_tables/0, sync/1]).
-export([sync/1]).

%% internal export used by session GC process
-export([destroy_session/1]).
Expand Down Expand Up @@ -674,9 +674,6 @@ list_client_subscriptions(ClientId) ->
%% Session tables operations
%%--------------------------------------------------------------------

create_tables() ->
emqx_persistent_session_ds_state:create_tables().

%% @doc Force syncing of the transient state to persistent storage
sync(ClientId) ->
case emqx_cm:lookup_channels(ClientId) of
Expand Down
6 changes: 6 additions & 0 deletions apps/emqx/src/emqx_persistent_session_ds.hrl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,18 @@

-define(PERSISTENT_MESSAGE_DB, emqx_persistent_message).

-ifdef(STORE_STATE_IN_DS).
-define(PERSISTENT_SESSION_DB, emqx_persistent_session).
%% ELSE ifdef(STORE_STATE_IN_DS).
-else.
-define(SESSION_TAB, emqx_ds_session).
-define(SESSION_SUBSCRIPTIONS_TAB, emqx_ds_session_subscriptions).
-define(SESSION_STREAM_TAB, emqx_ds_stream_tab).
-define(SESSION_PUBRANGE_TAB, emqx_ds_pubrange_tab).
-define(SESSION_COMMITTED_OFFSET_TAB, emqx_ds_committed_offset_tab).
-define(DS_MRIA_SHARD, emqx_ds_session_shard).
%% END ifdef(STORE_STATE_IN_DS).
-endif.

%%%%% Session sequence numbers:

Expand Down
Loading
Loading