12235783303

Committed 09 Dec 2024 12:36PM UTC coverage: 82.037%. First build

Build # 12235783303

Build Type

Pull #14362

github

Committed by

web-flow

Commit Message

Merge 4819ded51 into 83154d24b

Pull Request Pull Request #14362: refactor(resource): forbid changing resource state from `on_get_status` return

Run Details

62 of 82 new or added lines in 27 files covered. (75.61%)

56457 of 68819 relevant lines covered (82.04%)

15149.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.73

/apps/emqx_mysql/src/emqx_mysql.erl

%%--------------------------------------------------------------------
%% Copyright (c) 2020-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_mysql).

-include_lib("emqx_resource/include/emqx_resource.hrl").
-include_lib("emqx_connector/include/emqx_connector.hrl").
-include_lib("typerefl/include/types.hrl").
-include_lib("hocon/include/hoconsc.hrl").
-include_lib("emqx/include/logger.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").

-behaviour(emqx_resource).

%% callbacks of behaviour emqx_resource
-export([
    resource_type/0,
    callback_mode/0,
    on_start/2,
    on_stop/2,
    on_query/3,
    on_batch_query/4,
    on_get_status/2,
    on_format_query_result/1
]).

%% ecpool connect & reconnect
-export([connect/1, prepare_sql_to_conn/2, get_reconnect_callback_signature/1]).

-export([
    init_prepare/1,
    prepare_sql/2,
    parse_prepare_sql/1,
    parse_prepare_sql/2,
    unprepare_sql/2
]).

-export([roots/0, fields/1, namespace/0]).

-export([do_get_status/1]).

-define(MYSQL_HOST_OPTIONS, #{
    default_port => ?MYSQL_DEFAULT_PORT
}).

-type template() :: {unicode:chardata(), emqx_template:str()}.
-type state() ::
    #{
        pool_name := binary(),
        prepares := ok | {error, _},
        templates := #{{atom(), batch | prepstmt} => template()},
        query_templates := map()
    }.
-export_type([state/0]).
%%=====================================================================
%% Hocon schema

namespace() -> mysql.

roots() ->
    [{config, #{type => hoconsc:ref(?MODULE, config)}}].

fields(config) ->
    [{server, server()}] ++
        add_default_username(emqx_connector_schema_lib:relational_db_fields(), []) ++
        emqx_connector_schema_lib:ssl_fields().

add_default_username([{username, OrigUsernameFn} | Tail], Head) ->
    Head ++ [{username, add_default_fn(OrigUsernameFn, <<"root">>)} | Tail];
add_default_username([Field | Tail], Head) ->
    add_default_username(Tail, Head ++ [Field]).

add_default_fn(OrigFn, Default) ->
    fun
        (default) -> Default;
        (Field) -> OrigFn(Field)
    end.

server() ->
    Meta = #{desc => ?DESC("server")},
    emqx_schema:servers_sc(Meta, ?MYSQL_HOST_OPTIONS).

%% ===================================================================
resource_type() -> mysql.

callback_mode() -> always_sync.

-spec on_start(binary(), hocon:config()) -> {ok, state()} | {error, _}.
on_start(
    InstId,
    #{
        server := Server,
        database := DB,
        username := Username,
        pool_size := PoolSize,
        ssl := SSL
    } = Config
) ->
    #{hostname := Host, port := Port} = emqx_schema:parse_server(Server, ?MYSQL_HOST_OPTIONS),
    ?SLOG(info, #{
        msg => "starting_mysql_connector",
        connector => InstId,
        config => emqx_utils:redact(Config)
    }),
    SslOpts =
        case maps:get(enable, SSL) of
            true ->
                [{ssl, emqx_tls_lib:to_client_opts(SSL)}];
            false ->
                []
        end,
    Options =
        maybe_add_password_opt(
            maps:get(password, Config, undefined),
            [
                {host, Host},
                {port, Port},
                {user, Username},
                {database, DB},
                {auto_reconnect, ?AUTO_RECONNECT_INTERVAL},
                {pool_size, PoolSize}
            ]
        ),
    State = parse_prepare_sql(Config),
    case emqx_resource_pool:start(InstId, ?MODULE, Options ++ SslOpts) of
        ok ->
            {ok, init_prepare(State#{pool_name => InstId})};
        {error, Reason} ->
            ?tp(
                mysql_connector_start_failed,
                #{error => Reason}
            ),
            {error, Reason}
    end.

maybe_add_password_opt(undefined, Options) ->
    Options;
maybe_add_password_opt(Password, Options) ->
    [{password, Password} | Options].

on_stop(InstId, _State) ->
    ?SLOG(info, #{
        msg => "stopping_mysql_connector",
        connector => InstId
    }),
    emqx_resource_pool:stop(InstId).

on_query(InstId, {TypeOrKey, SQLOrKey}, State) ->
    on_query(InstId, {TypeOrKey, SQLOrKey, [], default_timeout}, State);
on_query(InstId, {TypeOrKey, SQLOrKey, Params}, State) ->
    on_query(InstId, {TypeOrKey, SQLOrKey, Params, default_timeout}, State);
on_query(
    InstId,
    {TypeOrKey, SQLOrKey, Params, Timeout},
    State
) ->
    MySqlFunction = mysql_function(TypeOrKey),
    {SQLOrKey2, Data} = proc_sql_params(TypeOrKey, SQLOrKey, Params, State),
    case on_sql_query(InstId, MySqlFunction, SQLOrKey2, Data, Timeout, State) of
        {error, not_prepared} ->
            case maybe_prepare_sql(SQLOrKey2, State) of
                ok ->
                    ?tp(
                        mysql_connector_on_query_prepared_sql,
                        #{type_or_key => TypeOrKey, sql_or_key => SQLOrKey, params => Params}
                    ),
                    %% not return result, next loop will try again
                    on_query(InstId, {TypeOrKey, SQLOrKey, Params, Timeout}, State);
                {error, Reason} ->
                    ?tp(
                        error,
                        "mysql_connector_do_prepare_failed",
                        #{
                            connector => InstId,
                            sql => SQLOrKey,
                            state => State,
                            reason => Reason
                        }
                    ),
                    {error, Reason}
            end;
        Result ->
            Result
    end.

on_batch_query(
    InstId,
    BatchReq = [{Key, _} | _],
    #{query_templates := Templates} = State,
    ChannelConfig
) ->
    case maps:get({Key, batch}, Templates, undefined) of
        undefined ->
            {error, {unrecoverable_error, batch_select_not_implemented}};
        Template ->
            on_batch_insert(InstId, BatchReq, Template, State, ChannelConfig)
    end;
on_batch_query(
    InstId,
    BatchReq,
    State,
    _
) ->
    ?SLOG(error, #{
        msg => "invalid request",
        connector => InstId,
        request => BatchReq,
        state => State
    }),
    {error, {unrecoverable_error, invalid_request}}.

on_format_query_result({ok, ColumnNames, Rows}) ->
    #{result => ok, column_names => ColumnNames, rows => Rows};
on_format_query_result({ok, DataList}) ->
    #{result => ok, column_names_rows_list => DataList};
on_format_query_result(Result) ->
    Result.

mysql_function(sql) ->
    query;
mysql_function(prepared_query) ->
    execute;
%% for bridge
mysql_function(_) ->
    mysql_function(prepared_query).

on_get_status(_InstId, #{pool_name := PoolName} = State) ->
    case emqx_resource_pool:health_check_workers(PoolName, fun ?MODULE:do_get_status/1) of
        true ->
            case do_check_prepares(State) of
                ok ->
                    ?status_connected;
                {error, undefined_table} ->
                    {?status_disconnected, unhealthy_target};
                {error, _Reason} ->
                    %% do not log error, it is logged in prepare_sql_to_conn
                    ?status_connecting
            end;
        false ->
            ?status_connecting
    end.

do_get_status(Conn) ->
    ok == element(1, mysql:query(Conn, <<"SELECT count(1) AS T">>)).

do_check_prepares(
    #{
        pool_name := PoolName,
        templates := #{{send_message, prepstmt} := SQL}
    }
) ->
    % it's already connected. Verify if target table still exists
    Workers = [Worker || {_WorkerName, Worker} <- ecpool:workers(PoolName)],
    lists:foldl(
        fun
            (WorkerPid, ok) ->
                case ecpool_worker:client(WorkerPid) of
                    {ok, Conn} ->
                        case mysql:prepare(Conn, get_status, SQL) of
                            {error, {1146, _, _}} ->
                                {error, undefined_table};
                            {ok, Statement} ->
                                mysql:unprepare(Conn, Statement);
                            _ ->
                                ok
                        end;
                    _ ->
                        ok
                end;
            (_, Acc) ->
                Acc
        end,
        ok,
        Workers
    );
do_check_prepares(_NoTemplates) ->
    ok.

%% ===================================================================

connect(Options) ->
    %% TODO: teach `tdengine` to accept 0-arity closures as passwords.
    NOptions = init_connect_opts(Options),
    mysql:start_link(NOptions).

init_connect_opts(Options) ->
    case lists:keytake(password, 1, Options) of
        {value, {password, Secret}, Rest} ->
            [{password, emqx_secret:unwrap(Secret)} | Rest];
        false ->
            Options
    end.

init_prepare(State = #{query_templates := Templates}) ->
    case maps:size(Templates) of
        0 ->
            State#{prepares => ok};
        _ ->
            case prepare_sql(State) of
                ok ->
                    State#{prepares => ok};
                {error, Reason} ->
                    ?SLOG(error, #{
                        msg => <<"MySQL init prepare statement failed">>,
                        reason => Reason
                    }),
                    %% mark the prepare_statement as failed
                    State#{prepares => {error, Reason}}
            end
    end.

maybe_prepare_sql(SQLOrKey, State = #{query_templates := Templates}) ->
    case maps:is_key({SQLOrKey, prepstmt}, Templates) of
        true -> prepare_sql(State);
        false -> {error, {unrecoverable_error, prepared_statement_invalid}}
    end.

prepare_sql(#{query_templates := Templates, pool_name := PoolName}) ->
    prepare_sql(maps:to_list(Templates), PoolName).

prepare_sql(Templates, PoolName) ->
    case do_prepare_sql(Templates, PoolName) of
        ok ->
            %% prepare for reconnect
            ecpool:add_reconnect_callback(PoolName, {?MODULE, prepare_sql_to_conn, [Templates]}),
            ok;
        {error, R} ->
            {error, R}
    end.

do_prepare_sql(Templates, PoolName) ->
    Conns = get_connections_from_pool(PoolName),
    prepare_sql_to_conn_list(Conns, Templates).

get_connections_from_pool(PoolName) ->
    lists:map(
        fun(Worker) ->
            {ok, Conn} = ecpool_worker:client(Worker),
            Conn
        end,
        pool_workers(PoolName)
    ).

pool_workers(PoolName) ->
    lists:map(fun({_Name, Worker}) -> Worker end, ecpool:workers(PoolName)).

prepare_sql_to_conn_list([], _Templates) ->
    ok;
prepare_sql_to_conn_list([Conn | ConnList], Templates) ->
    case prepare_sql_to_conn(Conn, Templates) of
        ok ->
            prepare_sql_to_conn_list(ConnList, Templates);
        {error, R} ->
            %% rollback
            _ = [unprepare_sql_to_conn(Conn, Template) || Template <- Templates],
            {error, R}
    end.

%% this callback accepts the arg list provided to
%% ecpool:add_reconnect_callback(PoolName, {?MODULE, prepare_sql_to_conn, [Templates]})
%% so ecpool_worker can de-duplicate the callbacks based on the signature.
get_reconnect_callback_signature([Templates]) ->
    [{{ChannelID, _}, _}] = lists:filter(
        fun
            ({{_, prepstmt}, _}) ->
                true;
            (_) ->
                false
        end,
        Templates
    ),
    ChannelID.

prepare_sql_to_conn(_Conn, []) ->
    ok;
prepare_sql_to_conn(Conn, [{{Key, prepstmt}, {SQL, _RowTemplate}} | Rest]) ->
    LogMeta = #{msg => "MySQL Prepare Statement", name => Key, prepare_sql => SQL},
    ?SLOG(info, LogMeta),
    _ = unprepare_sql_to_conn(Conn, Key),
    case mysql:prepare(Conn, Key, SQL) of
        {ok, _Key} ->
            ?SLOG(info, LogMeta#{result => success}),
            prepare_sql_to_conn(Conn, Rest);
        {error, {1146, _, _} = Reason} ->
            %% Target table is not created
            ?tp(mysql_undefined_table, #{}),
            ?SLOG(error, LogMeta#{result => failed, reason => Reason}),
            {error, undefined_table};
        {error, Reason} ->
            % FIXME: we should try to differ on transient failures and
            % syntax failures. Retrying syntax failures is not very productive.
            ?SLOG(error, LogMeta#{result => failed, reason => Reason}),
            {error, Reason}
    end;
prepare_sql_to_conn(Conn, [{_Key, _Template} | Rest]) ->
    prepare_sql_to_conn(Conn, Rest).

unprepare_sql(ChannelID, #{query_templates := Templates, pool_name := PoolName}) ->
    lists:foreach(
        fun(Worker) ->
            ok = ecpool_worker:remove_reconnect_callback_by_signature(Worker, ChannelID),
            case ecpool_worker:client(Worker) of
                {ok, Conn} ->
                    lists:foreach(
                        fun(Template) -> unprepare_sql_to_conn(Conn, Template) end,
                        maps:to_list(Templates)
                    );
                _ ->
                    ok
            end
        end,
        pool_workers(PoolName)
    ).

unprepare_sql_to_conn(Conn, {{Key, prepstmt}, _}) ->
    mysql:unprepare(Conn, Key);
unprepare_sql_to_conn(Conn, Key) when is_atom(Key) ->
    mysql:unprepare(Conn, Key);
unprepare_sql_to_conn(_Conn, _) ->
    ok.

parse_prepare_sql(Config) ->
    parse_prepare_sql(send_message, Config).

parse_prepare_sql(Key, Config) ->
    Queries =
        case Config of
            #{prepare_statement := Qs} ->
                Qs;
            #{sql := Query} ->
                #{Key => Query};
            _ ->
                #{}
        end,
    Templates = maps:fold(fun parse_prepare_sql/3, #{}, Queries),
    #{query_templates => Templates}.

parse_prepare_sql(Key, Query, Acc) ->
    Template = emqx_template_sql:parse_prepstmt(Query, #{parameters => '?'}),
    AccNext = Acc#{{Key, prepstmt} => Template},
    parse_batch_sql(Key, Query, AccNext).

parse_batch_sql(Key, Query, Acc) ->
    case emqx_utils_sql:get_statement_type(Query) of
        insert ->
            case emqx_utils_sql:parse_insert(Query) of
                {ok, {Insert, Params}} ->
                    RowTemplate = emqx_template_sql:parse(Params),
                    Acc#{{Key, batch} => {Insert, RowTemplate}};
                {error, Reason} ->
                    ?SLOG(error, #{
                        msg => "parse insert sql statement failed",
                        sql => Query,
                        reason => Reason
                    }),
                    Acc
            end;
        select ->
            Acc;
        Type ->
            ?SLOG(error, #{
                msg => "invalid sql statement type",
                sql => Query,
                type => Type
            }),
            Acc
    end.

proc_sql_params(query, SQLOrKey, Params, _State) ->
    {SQLOrKey, Params};
proc_sql_params(prepared_query, SQLOrKey, Params, _State) ->
    {SQLOrKey, Params};
proc_sql_params(TypeOrKey, SQLOrData, Params, #{query_templates := Templates}) ->
    case maps:get({TypeOrKey, prepstmt}, Templates, undefined) of
        undefined ->
            {SQLOrData, Params};
        {_InsertPart, RowTemplate} ->
            % NOTE
            % Ignoring errors here, missing variables are set to `null`.
            {Row, _Errors} = emqx_template_sql:render_prepstmt(
                RowTemplate,
                {emqx_jsonish, SQLOrData}
            ),
            {TypeOrKey, Row}
    end;
proc_sql_params(_TypeOrKey, SQLOrData, Params, _State) ->
    {SQLOrData, Params}.

on_batch_insert(InstId, BatchReqs, {InsertPart, RowTemplate}, State, ChannelConfig) ->
    Rows = [render_row(RowTemplate, Msg, ChannelConfig) || {_, Msg} <- BatchReqs],
    Query = [InsertPart, <<" values ">> | lists:join($,, Rows)],
    on_sql_query(InstId, query, Query, no_params, default_timeout, State).

render_row(RowTemplate, Data, ChannelConfig) ->
    RenderOpts =
        case maps:get(undefined_vars_as_null, ChannelConfig, false) of
            % NOTE:
            %  Ignoring errors here, missing variables are set to "'undefined'" due to backward
            %  compatibility requirements.
            false -> #{escaping => mysql, undefined => <<"undefined">>};
            true -> #{escaping => mysql}
        end,
    {Row, _Errors} = emqx_template_sql:render(RowTemplate, {emqx_jsonish, Data}, RenderOpts),
    Row.

on_sql_query(
    InstId,
    SQLFunc,
    SQLOrKey,
    Params,
    Timeout,
    #{pool_name := PoolName} = State
) ->
    LogMeta = #{connector => InstId, sql => SQLOrKey, state => State},
    ?TRACE("QUERY", "mysql_connector_received", LogMeta),
    ChannelID = maps:get(channel_id, State, no_channel),
    emqx_trace:rendered_action_template(
        ChannelID,
        #{
            sql_or_key => SQLOrKey,
            parameters => Params
        }
    ),
    Worker = ecpool:get_client(PoolName),
    case ecpool_worker:client(Worker) of
        {ok, Conn} ->
            ?tp(
                mysql_connector_send_query,
                #{sql_func => SQLFunc, sql_or_key => SQLOrKey, data => Params}
            ),
            do_sql_query(SQLFunc, Conn, SQLOrKey, Params, Timeout, LogMeta);
        {error, disconnected} ->
            ?tp(
                error,
                "mysql_connector_do_sql_query_failed",
                LogMeta#{reason => worker_is_disconnected}
            ),
            {error, {recoverable_error, disconnected}}
    end.

do_sql_query(SQLFunc, Conn, SQLOrKey, Params, Timeout, LogMeta) ->
    try mysql:SQLFunc(Conn, SQLOrKey, Params, no_filtermap_fun, Timeout) of
        {error, disconnected} ->
            ?SLOG(
                error,
                LogMeta#{msg => "mysql_connector_do_sql_query_failed", reason => disconnected}
            ),
            %% kill the pool worker to trigger reconnection
            _ = exit(Conn, restart),
            {error, {recoverable_error, disconnected}};
        {error, not_prepared} = Error ->
            ?tp(
                mysql_connector_prepare_query_failed,
                #{error => not_prepared}
            ),
            ?SLOG(
                warning,
                LogMeta#{msg => "mysql_connector_prepare_query_failed", reason => not_prepared}
            ),
            Error;
        {error, {1053, <<"08S01">>, Reason}} ->
            %% mysql sql server shutdown in progress
            ?SLOG(
                error,
                LogMeta#{msg => "mysql_connector_do_sql_query_failed", reason => Reason}
            ),
            {error, {recoverable_error, Reason}};
        {error, Reason} ->
            ?SLOG(
                error,
                LogMeta#{msg => "mysql_connector_do_sql_query_failed", reason => Reason}
            ),
            {error, {unrecoverable_error, Reason}};
        Result ->
            ?tp(
                mysql_connector_query_return,
                #{result => Result}
            ),
            Result
    catch
        error:badarg ->
            ?SLOG(
                error,
                LogMeta#{msg => "mysql_connector_invalid_params", params => Params}
            ),
            {error, {unrecoverable_error, {invalid_params, Params}}}
    end.

1	%%--------------------------------------------------------------------
2	%% Copyright (c) 2020-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
3	%%
4	%% Licensed under the Apache License, Version 2.0 (the "License");
5	%% you may not use this file except in compliance with the License.
6	%% You may obtain a copy of the License at
7	%%
8	%% http://www.apache.org/licenses/LICENSE-2.0
9	%%
10	%% Unless required by applicable law or agreed to in writing, software
11	%% distributed under the License is distributed on an "AS IS" BASIS,
12	%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	%% See the License for the specific language governing permissions and
14	%% limitations under the License.
15	%%--------------------------------------------------------------------
16	-module(emqx_mysql).
17
18	-include_lib("emqx_resource/include/emqx_resource.hrl").
19	-include_lib("emqx_connector/include/emqx_connector.hrl").
20	-include_lib("typerefl/include/types.hrl").
21	-include_lib("hocon/include/hoconsc.hrl").
22	-include_lib("emqx/include/logger.hrl").
23	-include_lib("snabbkaffe/include/snabbkaffe.hrl").
24
25	-behaviour(emqx_resource).
26
27	%% callbacks of behaviour emqx_resource
28	-export([
29	resource_type/0,
30	callback_mode/0,
31	on_start/2,
32	on_stop/2,
33	on_query/3,
34	on_batch_query/4,
35	on_get_status/2,
36	on_format_query_result/1
37	]).
38
39	%% ecpool connect & reconnect
40	-export([connect/1, prepare_sql_to_conn/2, get_reconnect_callback_signature/1]).
41
42	-export([
43	init_prepare/1,
44	prepare_sql/2,
45	parse_prepare_sql/1,
46	parse_prepare_sql/2,
47	unprepare_sql/2
48	]).
49
50	-export([roots/0, fields/1, namespace/0]).
51
52	-export([do_get_status/1]).
53
54	-define(MYSQL_HOST_OPTIONS, #{
55	default_port => ?MYSQL_DEFAULT_PORT
56	}).
57
58	-type template() :: {unicode:chardata(), emqx_template:str()}.
59	-type state() ::
60	#{
61	pool_name := binary(),
62	prepares := ok \| {error, _},
63	templates := #{{atom(), batch \| prepstmt} => template()},
64	query_templates := map()
65	}.
66	-export_type([state/0]).
67	%%=====================================================================
68	%% Hocon schema
69
70	namespace() -> mysql.	×
71
72	roots() ->
73	[{config, #{type => hoconsc:ref(?MODULE, config)}}].	4✔
74
75	fields(config) ->
76	[{server, server()}] ++	14,565✔
77	add_default_username(emqx_connector_schema_lib:relational_db_fields(), []) ++
78	emqx_connector_schema_lib:ssl_fields().
79
80	add_default_username([{username, OrigUsernameFn} \| Tail], Head) ->
81	Head ++ [{username, add_default_fn(OrigUsernameFn, <<"root">>)} \| Tail];	14,565✔
82	add_default_username([Field \| Tail], Head) ->
83	add_default_username(Tail, Head ++ [Field]).	29,130✔
84
85	add_default_fn(OrigFn, Default) ->
86	fun	14,565✔
87	(default) -> Default;	2,351✔
88	(Field) -> OrigFn(Field)	32,743✔
89	end.
90
91	server() ->
92	Meta = #{desc => ?DESC("server")},	14,565✔
93	emqx_schema:servers_sc(Meta, ?MYSQL_HOST_OPTIONS).	14,565✔
94
95	%% ===================================================================
96	resource_type() -> mysql.	209✔
97
98	callback_mode() -> always_sync.	209✔
99
100	-spec on_start(binary(), hocon:config()) -> {ok, state()} \| {error, _}.
101	on_start(
102	InstId,
103	#{
104	server := Server,
105	database := DB,
106	username := Username,
107	pool_size := PoolSize,
108	ssl := SSL
109	} = Config
110	) ->
111	#{hostname := Host, port := Port} = emqx_schema:parse_server(Server, ?MYSQL_HOST_OPTIONS),	211✔
112	?SLOG(info, #{	211✔
113	msg => "starting_mysql_connector",
114	connector => InstId,
115	config => emqx_utils:redact(Config)
116	}),	211✔
117	SslOpts =	211✔
118	case maps:get(enable, SSL) of
119	true ->
120	[{ssl, emqx_tls_lib:to_client_opts(SSL)}];	93✔
121	false ->
122	[]	118✔
123	end,
124	Options =	211✔
125	maybe_add_password_opt(
126	maps:get(password, Config, undefined),
127	[
128	{host, Host},
129	{port, Port},
130	{user, Username},
131	{database, DB},
132	{auto_reconnect, ?AUTO_RECONNECT_INTERVAL},
133	{pool_size, PoolSize}
134	]
135	),
136	State = parse_prepare_sql(Config),	211✔
137	case emqx_resource_pool:start(InstId, ?MODULE, Options ++ SslOpts) of	211✔
138	ok ->
139	{ok, init_prepare(State#{pool_name => InstId})};	183✔
140	{error, Reason} ->
141	?tp(	28✔
142	mysql_connector_start_failed,
143	#{error => Reason}
144	),
145	{error, Reason}	28✔
146	end.
147
148	maybe_add_password_opt(undefined, Options) ->
149	Options;	15✔
150	maybe_add_password_opt(Password, Options) ->
151	[{password, Password} \| Options].	196✔
152
153	on_stop(InstId, _State) ->
154	?SLOG(info, #{	184✔
155	msg => "stopping_mysql_connector",
156	connector => InstId
157	}),	184✔
158	emqx_resource_pool:stop(InstId).	184✔
159
160	on_query(InstId, {TypeOrKey, SQLOrKey}, State) ->
161	on_query(InstId, {TypeOrKey, SQLOrKey, [], default_timeout}, State);	54✔
162	on_query(InstId, {TypeOrKey, SQLOrKey, Params}, State) ->
163	on_query(InstId, {TypeOrKey, SQLOrKey, Params, default_timeout}, State);	61✔
164	on_query(
165	InstId,
166	{TypeOrKey, SQLOrKey, Params, Timeout},
167	State
168	) ->
169	MySqlFunction = mysql_function(TypeOrKey),	1,887✔
170	{SQLOrKey2, Data} = proc_sql_params(TypeOrKey, SQLOrKey, Params, State),	1,887✔
171	case on_sql_query(InstId, MySqlFunction, SQLOrKey2, Data, Timeout, State) of	1,887✔
172	{error, not_prepared} ->
173	case maybe_prepare_sql(SQLOrKey2, State) of	10✔
174	ok ->
175	?tp(	4✔
176	mysql_connector_on_query_prepared_sql,
177	#{type_or_key => TypeOrKey, sql_or_key => SQLOrKey, params => Params}
178	),
179	%% not return result, next loop will try again
180	on_query(InstId, {TypeOrKey, SQLOrKey, Params, Timeout}, State);	4✔
181	{error, Reason} ->
182	?tp(	6✔
183	error,
184	"mysql_connector_do_prepare_failed",
185	#{
186	connector => InstId,
187	sql => SQLOrKey,
188	state => State,
189	reason => Reason
190	}
191	),
192	{error, Reason}	6✔
193	end;
194	Result ->
195	Result	1,868✔
196	end.
197
198	on_batch_query(
199	InstId,
200	BatchReq = [{Key, _} \| _],
201	#{query_templates := Templates} = State,
202	ChannelConfig
203	) ->
204	case maps:get({Key, batch}, Templates, undefined) of	746✔
205	undefined ->
206	{error, {unrecoverable_error, batch_select_not_implemented}};	4✔
207	Template ->
208	on_batch_insert(InstId, BatchReq, Template, State, ChannelConfig)	742✔
209	end;
210	on_batch_query(
211	InstId,
212	BatchReq,
213	State,
214	_
215	) ->
216	?SLOG(error, #{	8✔
217	msg => "invalid request",
218	connector => InstId,
219	request => BatchReq,
220	state => State
221	}),	×
222	{error, {unrecoverable_error, invalid_request}}.	8✔
223
224	on_format_query_result({ok, ColumnNames, Rows}) ->
225	#{result => ok, column_names => ColumnNames, rows => Rows};	×
226	on_format_query_result({ok, DataList}) ->
227	#{result => ok, column_names_rows_list => DataList};	×
228	on_format_query_result(Result) ->
229	Result.	×
230
231	mysql_function(sql) ->
232	query;	80✔
233	mysql_function(prepared_query) ->
234	execute;	1,807✔
235	%% for bridge
236	mysql_function(_) ->
237	mysql_function(prepared_query).	1,743✔
238
239	on_get_status(_InstId, #{pool_name := PoolName} = State) ->
240	case emqx_resource_pool:health_check_workers(PoolName, fun ?MODULE:do_get_status/1) of	388✔
241	true ->
242	case do_check_prepares(State) of	366✔
243	ok ->
244	?status_connected;	366✔
245	{error, undefined_table} ->
NEW 246	{?status_disconnected, unhealthy_target};	×
247	{error, _Reason} ->
248	%% do not log error, it is logged in prepare_sql_to_conn
NEW 249	?status_connecting	×
250	end;
251	false ->
252	?status_connecting	21✔
253	end.
254
255	do_get_status(Conn) ->
256	ok == element(1, mysql:query(Conn, <<"SELECT count(1) AS T">>)).	1,597✔
257
258	do_check_prepares(
259	#{
260	pool_name := PoolName,
261	templates := #{{send_message, prepstmt} := SQL}
262	}
263	) ->
264	% it's already connected. Verify if target table still exists
265	Workers = [Worker \|\| {_WorkerName, Worker} <- ecpool:workers(PoolName)],	×
266	lists:foldl(	×
267	fun
268	(WorkerPid, ok) ->
269	case ecpool_worker:client(WorkerPid) of	×
270	{ok, Conn} ->
271	case mysql:prepare(Conn, get_status, SQL) of	×
272	{error, {1146, _, _}} ->
273	{error, undefined_table};	×
274	{ok, Statement} ->
275	mysql:unprepare(Conn, Statement);	×
276	_ ->
277	ok	×
278	end;
279	_ ->
280	ok	×
281	end;
282	(_, Acc) ->
283	Acc	×
284	end,
285	ok,
286	Workers
287	);
288	do_check_prepares(_NoTemplates) ->
289	ok.	366✔
290
291	%% ===================================================================
292
293	connect(Options) ->
294	%% TODO: teach `tdengine` to accept 0-arity closures as passwords.
295	NOptions = init_connect_opts(Options),	998✔
296	mysql:start_link(NOptions).	998✔
297
298	init_connect_opts(Options) ->
299	case lists:keytake(password, 1, Options) of	998✔
300	{value, {password, Secret}, Rest} ->
301	[{password, emqx_secret:unwrap(Secret)} \| Rest];	878✔
302	false ->
303	Options	120✔
304	end.
305
306	init_prepare(State = #{query_templates := Templates}) ->
307	case maps:size(Templates) of	328✔
308	0 ->
309	State#{prepares => ok};	156✔
310	_ ->
311	case prepare_sql(State) of	172✔
312	ok ->
313	State#{prepares => ok};	151✔
314	{error, Reason} ->
315	?SLOG(error, #{	21✔
316	msg => <<"MySQL init prepare statement failed">>,
317	reason => Reason
318	}),	×
319	%% mark the prepare_statement as failed
320	State#{prepares => {error, Reason}}	21✔
321	end
322	end.
323
324	maybe_prepare_sql(SQLOrKey, State = #{query_templates := Templates}) ->
325	case maps:is_key({SQLOrKey, prepstmt}, Templates) of	10✔
326	true -> prepare_sql(State);	6✔
327	false -> {error, {unrecoverable_error, prepared_statement_invalid}}	4✔
328	end.
329
330	prepare_sql(#{query_templates := Templates, pool_name := PoolName}) ->
331	prepare_sql(maps:to_list(Templates), PoolName).	178✔
332
333	prepare_sql(Templates, PoolName) ->
334	case do_prepare_sql(Templates, PoolName) of	178✔
335	ok ->
336	%% prepare for reconnect
337	ecpool:add_reconnect_callback(PoolName, {?MODULE, prepare_sql_to_conn, [Templates]}),	155✔
338	ok;	155✔
339	{error, R} ->
340	{error, R}	23✔
341	end.
342
343	do_prepare_sql(Templates, PoolName) ->
344	Conns = get_connections_from_pool(PoolName),	178✔
345	prepare_sql_to_conn_list(Conns, Templates).	178✔
346
347	get_connections_from_pool(PoolName) ->
348	lists:map(	178✔
349	fun(Worker) ->
350	{ok, Conn} = ecpool_worker:client(Worker),	773✔
351	Conn	773✔
352	end,
353	pool_workers(PoolName)
354	).
355
356	pool_workers(PoolName) ->
357	lists:map(fun({_Name, Worker}) -> Worker end, ecpool:workers(PoolName)).	305✔
358
359	prepare_sql_to_conn_list([], _Templates) ->
360	ok;	155✔
361	prepare_sql_to_conn_list([Conn \| ConnList], Templates) ->
362	case prepare_sql_to_conn(Conn, Templates) of	690✔
363	ok ->
364	prepare_sql_to_conn_list(ConnList, Templates);	667✔
365	{error, R} ->
366	%% rollback
367	_ = [unprepare_sql_to_conn(Conn, Template) \|\| Template <- Templates],	23✔
368	{error, R}	23✔
369	end.
370
371	%% this callback accepts the arg list provided to
372	%% ecpool:add_reconnect_callback(PoolName, {?MODULE, prepare_sql_to_conn, [Templates]})
373	%% so ecpool_worker can de-duplicate the callbacks based on the signature.
374	get_reconnect_callback_signature([Templates]) ->
375	[{{ChannelID, _}, _}] = lists:filter(	1,227✔
376	fun
377	({{_, prepstmt}, _}) ->
378	true;	1,227✔
379	(_) ->
380	false	1,056✔
381	end,
382	Templates
383	),
384	ChannelID.	1,227✔
385
386	prepare_sql_to_conn(_Conn, []) ->
387	ok;	675✔
388	prepare_sql_to_conn(Conn, [{{Key, prepstmt}, {SQL, _RowTemplate}} \| Rest]) ->
389	LogMeta = #{msg => "MySQL Prepare Statement", name => Key, prepare_sql => SQL},	698✔
390	?SLOG(info, LogMeta),	698✔
391	_ = unprepare_sql_to_conn(Conn, Key),	698✔
392	case mysql:prepare(Conn, Key, SQL) of	698✔
393	{ok, _Key} ->
394	?SLOG(info, LogMeta#{result => success}),	675✔
395	prepare_sql_to_conn(Conn, Rest);	675✔
396	{error, {1146, _, _} = Reason} ->
397	%% Target table is not created
398	?tp(mysql_undefined_table, #{}),	12✔
399	?SLOG(error, LogMeta#{result => failed, reason => Reason}),	12✔
400	{error, undefined_table};	12✔
401	{error, Reason} ->
402	% FIXME: we should try to differ on transient failures and
403	% syntax failures. Retrying syntax failures is not very productive.
404	?SLOG(error, LogMeta#{result => failed, reason => Reason}),	11✔
405	{error, Reason}	11✔
406	end;
407	prepare_sql_to_conn(Conn, [{_Key, _Template} \| Rest]) ->
408	prepare_sql_to_conn(Conn, Rest).	553✔
409
410	unprepare_sql(ChannelID, #{query_templates := Templates, pool_name := PoolName}) ->
411	lists:foreach(	127✔
412	fun(Worker) ->
413	ok = ecpool_worker:remove_reconnect_callback_by_signature(Worker, ChannelID),	536✔
414	case ecpool_worker:client(Worker) of	536✔
415	{ok, Conn} ->
416	lists:foreach(	456✔
417	fun(Template) -> unprepare_sql_to_conn(Conn, Template) end,	×
418	maps:to_list(Templates)
419	);
420	_ ->
421	ok	80✔
422	end
423	end,
424	pool_workers(PoolName)
425	).
426
427	unprepare_sql_to_conn(Conn, {{Key, prepstmt}, _}) ->
428	mysql:unprepare(Conn, Key);	23✔
429	unprepare_sql_to_conn(Conn, Key) when is_atom(Key) ->
430	mysql:unprepare(Conn, Key);	113✔
431	unprepare_sql_to_conn(_Conn, _) ->
432	ok.	602✔
433
434	parse_prepare_sql(Config) ->
435	parse_prepare_sql(send_message, Config).	211✔
436
437	parse_prepare_sql(Key, Config) ->
438	Queries =	364✔
439	case Config of
440	#{prepare_statement := Qs} ->
441	Qs;	47✔
442	#{sql := Query} ->
443	#{Key => Query};	153✔
444	_ ->
445	#{}	164✔
446	end,
447	Templates = maps:fold(fun parse_prepare_sql/3, #{}, Queries),	364✔
448	#{query_templates => Templates}.	364✔
449
450	parse_prepare_sql(Key, Query, Acc) ->
451	Template = emqx_template_sql:parse_prepstmt(Query, #{parameters => '?'}),	200✔
452	AccNext = Acc#{{Key, prepstmt} => Template},	200✔
453	parse_batch_sql(Key, Query, AccNext).	200✔
454
455	parse_batch_sql(Key, Query, Acc) ->
456	case emqx_utils_sql:get_statement_type(Query) of	200✔
457	insert ->
458	case emqx_utils_sql:parse_insert(Query) of	137✔
459	{ok, {Insert, Params}} ->
460	RowTemplate = emqx_template_sql:parse(Params),	137✔
461	Acc#{{Key, batch} => {Insert, RowTemplate}};	137✔
462	{error, Reason} ->
463	?SLOG(error, #{	×
464	msg => "parse insert sql statement failed",
465	sql => Query,
466	reason => Reason
467	}),	×
468	Acc	×
469	end;
470	select ->
471	Acc;	47✔
472	Type ->
473	?SLOG(error, #{	16✔
474	msg => "invalid sql statement type",
475	sql => Query,
476	type => Type
477	}),	×
478	Acc	16✔
479	end.
480
481	proc_sql_params(query, SQLOrKey, Params, _State) ->
482	{SQLOrKey, Params};	×
483	proc_sql_params(prepared_query, SQLOrKey, Params, _State) ->
484	{SQLOrKey, Params};	64✔
485	proc_sql_params(TypeOrKey, SQLOrData, Params, #{query_templates := Templates}) ->
486	case maps:get({TypeOrKey, prepstmt}, Templates, undefined) of	1,823✔
487	undefined ->
488	{SQLOrData, Params};	80✔
489	{_InsertPart, RowTemplate} ->
490	% NOTE
491	% Ignoring errors here, missing variables are set to `null`.
492	{Row, _Errors} = emqx_template_sql:render_prepstmt(	1,743✔
493	RowTemplate,
494	{emqx_jsonish, SQLOrData}
495	),
496	{TypeOrKey, Row}	1,743✔
497	end;
498	proc_sql_params(_TypeOrKey, SQLOrData, Params, _State) ->
499	{SQLOrData, Params}.	×
500
501	on_batch_insert(InstId, BatchReqs, {InsertPart, RowTemplate}, State, ChannelConfig) ->
502	Rows = [render_row(RowTemplate, Msg, ChannelConfig) \|\| {_, Msg} <- BatchReqs],	742✔
503	Query = [InsertPart, <<" values ">> \| lists:join($,, Rows)],	742✔
504	on_sql_query(InstId, query, Query, no_params, default_timeout, State).	742✔
505
506	render_row(RowTemplate, Data, ChannelConfig) ->
507	RenderOpts =	3,232✔
508	case maps:get(undefined_vars_as_null, ChannelConfig, false) of
509	% NOTE:
510	% Ignoring errors here, missing variables are set to "'undefined'" due to backward
511	% compatibility requirements.
512	false -> #{escaping => mysql, undefined => <<"undefined">>};	3,228✔
513	true -> #{escaping => mysql}	4✔
514	end,
515	{Row, _Errors} = emqx_template_sql:render(RowTemplate, {emqx_jsonish, Data}, RenderOpts),	3,232✔
516	Row.	3,232✔
517
518	on_sql_query(
519	InstId,
520	SQLFunc,
521	SQLOrKey,
522	Params,
523	Timeout,
524	#{pool_name := PoolName} = State
525	) ->
526	LogMeta = #{connector => InstId, sql => SQLOrKey, state => State},	2,629✔
527	?TRACE("QUERY", "mysql_connector_received", LogMeta),	2,629✔
528	ChannelID = maps:get(channel_id, State, no_channel),	2,629✔
529	emqx_trace:rendered_action_template(	2,629✔
530	ChannelID,
531	#{
532	sql_or_key => SQLOrKey,
533	parameters => Params
534	}
535	),
536	Worker = ecpool:get_client(PoolName),	2,629✔
537	case ecpool_worker:client(Worker) of	2,629✔
538	{ok, Conn} ->
539	?tp(	2,613✔
540	mysql_connector_send_query,
541	#{sql_func => SQLFunc, sql_or_key => SQLOrKey, data => Params}
542	),
543	do_sql_query(SQLFunc, Conn, SQLOrKey, Params, Timeout, LogMeta);	2,613✔
544	{error, disconnected} ->
545	?tp(	16✔
546	error,
547	"mysql_connector_do_sql_query_failed",
548	LogMeta#{reason => worker_is_disconnected}
549	),
550	{error, {recoverable_error, disconnected}}	16✔
551	end.
552
553	do_sql_query(SQLFunc, Conn, SQLOrKey, Params, Timeout, LogMeta) ->
554	try mysql:SQLFunc(Conn, SQLOrKey, Params, no_filtermap_fun, Timeout) of	2,613✔
555	{error, disconnected} ->
556	?SLOG(	×
557	error,	×
558	LogMeta#{msg => "mysql_connector_do_sql_query_failed", reason => disconnected}	×
559	),
560	%% kill the pool worker to trigger reconnection
561	_ = exit(Conn, restart),	×
562	{error, {recoverable_error, disconnected}};	×
563	{error, not_prepared} = Error ->
564	?tp(	10✔
565	mysql_connector_prepare_query_failed,
566	#{error => not_prepared}
567	),
568	?SLOG(	10✔
569	warning,	10✔
570	LogMeta#{msg => "mysql_connector_prepare_query_failed", reason => not_prepared}	×
571	),
572	Error;	10✔
573	{error, {1053, <<"08S01">>, Reason}} ->
574	%% mysql sql server shutdown in progress
575	?SLOG(	×
576	error,	×
577	LogMeta#{msg => "mysql_connector_do_sql_query_failed", reason => Reason}	×
578	),
579	{error, {recoverable_error, Reason}};	×
580	{error, Reason} ->
581	?SLOG(	36✔
582	error,	36✔
583	LogMeta#{msg => "mysql_connector_do_sql_query_failed", reason => Reason}	×
584	),
585	{error, {unrecoverable_error, Reason}};	36✔
586	Result ->
587	?tp(	2,554✔
588	mysql_connector_query_return,
589	#{result => Result}
590	),
591	Result	2,554✔
592	catch
593	error:badarg ->
594	?SLOG(	4✔
595	error,	4✔
596	LogMeta#{msg => "mysql_connector_invalid_params", params => Params}	×
597	),
598	{error, {unrecoverable_error, {invalid_params, Params}}}	4✔
599	end.

emqx / emqx / 12235783303

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous