fix(ds): use configured data dir for site storage

apps/emqx_durable_storage/src/emqx_ds.erl

@@ -23,6 +23,7 @@
 
 %% Management API:
 -export([
+    base_dir/0,
     open_db/2,
     update_db_config/2,
     add_generation/1,
@@ -71,6 +72,8 @@
 %% Type declarations
 %%================================================================================
 
+-define(APP, emqx_durable_storage).
+
 -type db() :: atom().
 
 %% Parsed topic.
@@ -189,6 +192,10 @@
 %% API funcions
 %%================================================================================
 
+-spec base_dir() -> file:filename().
+base_dir() ->
+    application:get_env(?APP, db_data_dir, emqx:data_dir()).
+
 %% @doc Different DBs are completely independent from each other. They
 %% could represent something like different tenants.
 -spec open_db(db(), create_db_opts()) -> ok.

apps/emqx_durable_storage/src/emqx_ds_replication_layer_meta.erl

@@ -417,7 +417,7 @@ ensure_tables() ->
     ok = mria:wait_for_tables([?META_TAB, ?NODE_TAB, ?SHARD_TAB]).
 
 ensure_site() ->
-    Filename = filename:join(emqx:data_dir(), "emqx_ds_builtin_site.eterm"),
+    Filename = filename:join(emqx_ds:base_dir(), "emqx_ds_builtin_site.eterm"),
     case file:consult(Filename) of
         {ok, [Site]} ->
             ok;

apps/emqx_durable_storage/src/emqx_ds_storage_layer.erl

@@ -52,7 +52,6 @@
 
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
--define(APP, emqx_durable_storage).
 -define(REF(ShardId), {via, gproc, {n, l, {?MODULE, ShardId}}}).
 
 %%================================================================================
@@ -608,11 +607,7 @@ rocksdb_open(Shard, Options) ->
 
 -spec db_dir(shard_id()) -> file:filename().
 db_dir({DB, ShardId}) ->
-    filename:join([base_dir(), atom_to_list(DB), binary_to_list(ShardId)]).
-
--spec base_dir() -> file:filename().
-base_dir() ->
-    application:get_env(?APP, db_data_dir, emqx:data_dir()).
+    filename:join([emqx_ds:base_dir(), atom_to_list(DB), binary_to_list(ShardId)]).
 
 -spec update_last_until(Schema, emqx_ds:time()) -> Schema when Schema :: shard_schema() | shard().
 update_last_until(Schema, Until) ->