feat: add option to initialize default schema

Author: savonarola

docker-compose.yml

@@ -39,8 +39,6 @@ services:
       MYSQL_PASSWORD: public
     ports:
       - "3306:3306"
-    volumes:
-      - ./docker/mysql/docker-entrypoint-initdb.d:/docker-entrypoint-initdb.d
     networks:
       - emqx_network
     healthcheck:
@@ -68,7 +66,6 @@ services:
       - "3307:3306"
     volumes:
       - ./test/assets/certs:/certs:ro
-      - ./docker/mysql/docker-entrypoint-initdb.d:/docker-entrypoint-initdb.d
     networks:
       - emqx_network
     healthcheck:

docker/mysql/docker-entrypoint-initdb.d/init.sql

@@ -14,6 +14,7 @@ mysql {
     username = "emqx"
     password = "public"
     database = "emqx"
+    init_default_schema = false
     insert_message_sql = "insert into mqtt_msg(msgid, sender, topic, qos, retain, payload, arrived) values(${id}, ${from}, ${topic}, ${qos}, ${flags.retain}, ${payload}, FROM_UNIXTIME(${timestamp}/1000))"
     delete_message_sql = "delete from mqtt_msg where msgid = ${id}"
     select_message_sql = "select * from mqtt_msg where topic = ${topic}"

priv/config.hocon

@@ -15,6 +15,14 @@
         "en": "Topic filters for topics to persist, separated by commas",
         "zh": "要持久化的主题过滤器，用逗号分隔"
     },
+    "$mysql_init_default_schema_label": {
+        "en": "Initialize Default Schema",
+        "zh": "初始化默认架构"
+    },
+    "$mysql_init_default_schema_desc": {
+        "en": "Initialize the default schema for the database",
+        "zh": "初始化数据库的默认架构"
+    },
     "$mysql_insert_message_sql_label": {
         "en": "MySQL Insert Message SQL",
         "zh": "MySQL 插入消息 SQL"

priv/config_i18n.json

@@ -19,6 +19,17 @@
               "description": "$mysql_enable_desc"
             }
           },
+          {
+            "name": "init_default_schema",
+            "type": "boolean",
+            "default": false,
+            "$ui": {
+              "component": "switch",
+              "required": true,
+              "label": "$mysql_init_default_schema_label",
+              "description": "$mysql_init_default_schema_desc"
+            }
+          },
           {
             "name": "insert_message_sql",
             "type": "string",

priv/config_schema.avsc

@@ -23,33 +23,32 @@
 ]).
 
 -define(RESOURCE_ID, <<"omp_mysql">>).
+-define(RESOURCE_ID_INIT, <<"omp_mysql_init">>).
 -define(RESOURCE_GROUP, <<"omp">>).
 -define(TIMEOUT, 1000).
 
-% -define(INIT_SQL, [
-%     """
-%     CREATE TABLE IF NOT EXISTS `mqtt_msg` (
-%         `id` bigint unsigned NOT NULL AUTO_INCREMENT,
-%         `msgid` varchar(64) DEFAULT NULL,
-%         `topic` varchar(180) NOT NULL,
-%         `sender` varchar(64) DEFAULT NULL,
-%         `qos` tinyint(1) NOT NULL DEFAULT '0',
-%         `retain` tinyint(1) DEFAULT NULL,
-%         `payload` blob,
-%         `arrived` datetime NOT NULL,
-%         PRIMARY KEY (`id`),
-%         INDEX topic_index(`topic`)
-%     ) ENGINE=InnoDB DEFAULT CHARSET=utf8MB4;
-%     """,
-%     """
-%     CREATE TABLE IF NOT EXISTS `mqtt_sub` (
-%         `clientid` varchar(64) NOT NULL,
-%         `topic` varchar(180) NOT NULL,
-%         `qos` tinyint(1) NOT NULL DEFAULT '0',
-%         PRIMARY KEY (`clientid`, `topic`)
-%     ) ENGINE=InnoDB DEFAULT CHARSET=utf8MB4;
-%     """
-% ]).
+-define(INIT_SQL, [
+    <<"CREATE TABLE IF NOT EXISTS `mqtt_msg` ("
+        "`id` bigint unsigned NOT NULL AUTO_INCREMENT,"
+        "`msgid` varchar(64) DEFAULT NULL,"
+        "`topic` varchar(180) NOT NULL,"
+        "`sender` varchar(64) DEFAULT NULL,"
+        "`qos` tinyint(1) NOT NULL DEFAULT '0',"
+        "`retain` tinyint(1) DEFAULT NULL,"
+        "`payload` blob,"
+        "`arrived` datetime NOT NULL,"
+        "PRIMARY KEY (`id`),"
+        "INDEX topic_index(`topic`)"
+    ")"
+    "ENGINE=InnoDB DEFAULT CHARSET=utf8MB4;">>,
+
+    <<"CREATE TABLE IF NOT EXISTS `mqtt_sub` ("
+        "`clientid` varchar(64) NOT NULL,"
+        "`topic` varchar(180) NOT NULL,"
+        "`qos` tinyint(1) NOT NULL DEFAULT '0',"
+        "PRIMARY KEY (`clientid`, `topic`)"
+    ") ENGINE=InnoDB DEFAULT CHARSET=utf8MB4;">>
+]).
 
 -type statement() :: emqx_template_sql:statement().
 -type param_template() :: emqx_template_sql:row_template().
@@ -90,9 +89,11 @@ stop() ->
 
 -spec start(map()) -> ok.
 start(ConfigRaw) ->
+    ok = init_default_schema(ConfigRaw),
     {MysqlConfig, ResourceOpts} = make_mysql_resource_config(ConfigRaw),
     ok = start_resource(MysqlConfig, ResourceOpts),
 
+
     Statements = parse_statements(
         [delete_message_sql, select_message_sql, insert_subscription_sql, select_subscriptions_sql],
         ConfigRaw
@@ -392,6 +393,30 @@ make_mysql_resource_config(#{<<"insert_message_sql">> := InsertMessageStatement}
 
     {MysqlConfig, ResourceOpts}.
 
+init_default_schema(ConfigRaw) ->
+    {MysqlConfig0, ResourceOpts} = make_mysql_resource_config(ConfigRaw),
+    MysqlConfig = MysqlConfig0#{prepare_statement => #{}},
+    {ok, _} = emqx_resource:create_local(
+        ?RESOURCE_ID_INIT,
+        ?RESOURCE_GROUP,
+        emqx_bridge_mysql_connector,
+        MysqlConfig,
+        ResourceOpts
+    ),
+    ok = lists:foreach(fun(Sql) ->
+        case emqx_resource:simple_sync_query(?RESOURCE_ID_INIT, {sql, Sql, [], ?TIMEOUT}) of
+            {error, Reason} ->
+                ?SLOG(error, #{
+                    msg => "omp_mysql_init_default_schema_error",
+                    sql => Sql,
+                    reason => Reason
+                });
+            _ ->
+                ok
+        end
+    end, ?INIT_SQL),
+    ok = emqx_resource:remove_local(?RESOURCE_ID_INIT).
+
 sync_query(Sql, Params) ->
     emqx_resource:simple_sync_query(?RESOURCE_ID, {sql, Sql, Params, ?TIMEOUT}).
 

src/emqx_omp_mysql.erl

@@ -252,7 +252,7 @@ on_message_publish(Message, #{message_ttl := TTL, topic_filters := TopicFilters}
                     [<<"EXPIRE">>, msg_table(Context, MsgIDb62), TTL],
                     [<<"ZREMRANGEBYSCORE">>, msg_table(Context, Topic), 2, Now]
                 ],
-                case emqx_resource:query(?RESOURCE_ID, {cmds, Cmds}) of
+                case sync_cmds(Cmds) of
                     {ok, _} ->
                         ok;
                     {error, Reason} ->
@@ -276,7 +276,7 @@ on_message_acked(
         [<<"DEL">>, msg_table(Context, MsgIDb62)],
         [<<"ZREM">>, msg_table(Context, Topic), MsgIDb62]
     ],
-    case emqx_resource:simple_sync_query(?RESOURCE_ID, {cmds, Cmds}) of
+    case sync_cmds(Cmds) of
         {ok, _} ->
             emqx_metrics_worker:inc(?METRICS_WORKER, message_acked, success);
         {error, Reason} ->

src/emqx_omp_redis.erl

@@ -291,6 +291,7 @@ plugin_config() ->
             username => <<"emqx">>,
             pool_size => 8,
             database => <<"emqx">>,
+            init_default_schema => true,
             select_message_sql => <<"select * from mqtt_msg where topic = ${topic}">>,
             delete_message_sql => <<"delete from mqtt_msg where msgid = ${id}">>,
             insert_message_sql => <<