Merge pull request #1558 from emqtt/develop

Version 2.3.7

Makefile

@@ -9,7 +9,7 @@ dep_gproc        = git https://github.com/uwiger/gproc
 dep_getopt       = git https://github.com/jcomellas/getopt v0.8.2
 dep_lager        = git https://github.com/basho/lager master
 dep_esockd       = git https://github.com/emqtt/esockd v5.2.1
-dep_ekka         = git https://github.com/emqtt/ekka v0.2.2
+dep_ekka         = git https://github.com/emqtt/ekka v0.2.3
 dep_mochiweb     = git https://github.com/emqtt/mochiweb v4.2.2
 dep_pbkdf2       = git https://github.com/emqtt/pbkdf2 2.0.1
 dep_lager_syslog = git https://github.com/basho/lager_syslog

etc/emq.conf

@@ -135,6 +135,12 @@ cluster.autoclean = 5m
 ## Value: String
 ## cluster.k8s.app_name = emq
 
+## Kubernates Namespace
+##
+## Value: String
+## cluster.k8s.namespace = default
+
+
 ##--------------------------------------------------------------------
 ## Node Args
 ##--------------------------------------------------------------------

priv/emq.schema

@@ -124,6 +124,10 @@
   {datatype, string}
 ]}.
 
+{mapping, "cluster.k8s.namespace", "ekka.cluster_discovery", [
+  {datatype, string}
+]}.
+
 {translation, "ekka.cluster_discovery", fun(Conf) ->
   Strategy = cuttlefish:conf_get("cluster.discovery", Conf),
   Filter  = fun(Opts) -> [{K, V} || {K, V} <- Opts, V =/= undefined] end,
@@ -152,7 +156,8 @@
                  [{apiserver, cuttlefish:conf_get("cluster.k8s.apiserver", Conf)},
                   {service_name, cuttlefish:conf_get("cluster.k8s.service_name", Conf)},
                   {address_type, cuttlefish:conf_get("cluster.k8s.address_type", Conf, ip)},
-                  {app_name, cuttlefish:conf_get("cluster.k8s.app_name", Conf)}];
+                  {app_name, cuttlefish:conf_get("cluster.k8s.app_name", Conf)},
+                  {namespace, cuttlefish:conf_get("cluster.k8s.namespace", Conf)}];
                (manual) ->
                  [ ]
             end,
@@ -434,7 +439,7 @@ end}.
     ConsoleLogLevel = cuttlefish:conf_get("log.console.level", Conf),
     ConsoleLogFile = cuttlefish:conf_get("log.console.file", Conf),
 
-    ConsoleHandler = {lager_console_backend, ConsoleLogLevel},
+    ConsoleHandler = {lager_console_backend, [ConsoleLogLevel]},
     ConsoleFileHandler = {lager_file_backend, [{file, ConsoleLogFile},
                                                {level, ConsoleLogLevel},
                                                {size, cuttlefish:conf_get("log.console.size", Conf)},

src/emqttd.app.src

@@ -1,6 +1,6 @@
 {application,emqttd,
              [{description,"Erlang MQTT Broker"},
-              {vsn,"2.3.6"},
+              {vsn,"2.3.7"},
               {modules,[]},
               {registered,[emqttd_sup]},
               {applications,[kernel,stdlib,gproc,lager,esockd,mochiweb,

src/emqttd_packet.erl

@@ -56,7 +56,7 @@ format_header(#mqtt_packet_header{type = Type,
                                   dup = Dup,
                                   qos = QoS,
                                   retain = Retain}, S) ->
-    S1 = if 
+    S1 = if
              S == undefined -> <<>>;
              true           -> [", ", S]
          end,
@@ -78,13 +78,13 @@ format_variable(#mqtt_packet_connect{
                  clean_sess    = CleanSess,
                  keep_alive    = KeepAlive,
                  client_id     = ClientId,
-                 will_topic    = WillTopic, 
+                 will_topic    = WillTopic,
-                 will_msg      = WillMsg, 
+                 will_msg      = WillMsg,
-                 username      = Username, 
+                 username      = Username,
                  password      = Password}) ->
     Format = "ClientId=~s, ProtoName=~s, ProtoVsn=~p, CleanSess=~s, KeepAlive=~p, Username=~s, Password=~s",
     Args = [ClientId, ProtoName, ProtoVer, CleanSess, KeepAlive, Username, format_password(Password)],
-    {Format1, Args1} = if 
+    {Format1, Args1} = if
                         WillFlag -> { Format ++ ", Will(Q~p, R~p, Topic=~s, Msg=~s)",
                                       Args ++ [WillQoS, i(WillRetain), WillTopic, WillMsg] };
                         true -> {Format, Args}
@@ -93,7 +93,7 @@ format_variable(#mqtt_packet_connect{
 
 format_variable(#mqtt_packet_connack{ack_flags   = AckFlags,
                                      return_code = ReturnCode}) ->
-    io_lib:format("AckFlags=~p, RetainCode=~p", [AckFlags, ReturnCode]);
+    io_lib:format("AckFlags=~p, ReturnCode=~p", [AckFlags, ReturnCode]);
 
 format_variable(#mqtt_packet_publish{topic_name = TopicName,
                                      packet_id  = PacketId}) ->

src/emqttd_protocol.erl

@@ -137,7 +137,7 @@ session(#proto_state{session = Session}) ->
 
 %% CONNECT – Client requests a connection to a Server
 
-%% A Client can only send the CONNECT Packet once over a Network Connection. 
+%% A Client can only send the CONNECT Packet once over a Network Connection.
 -spec(received(mqtt_packet(), proto_state()) -> {ok, proto_state()} | {error, term()}).
 received(Packet = ?PACKET(?CONNECT),
          State = #proto_state{connected = false, stats_data = Stats}) ->
@@ -162,11 +162,12 @@ received(Packet = ?PACKET(Type), State = #proto_state{stats_data = Stats}) ->
 
 subscribe(RawTopicTable, ProtoState = #proto_state{client_id = ClientId,
                                                    username  = Username,
-                                                   session   = Session}) ->
+                                                   session   = Session,
+                                                   mountpoint = MountPoint}) ->
     TopicTable = parse_topic_table(RawTopicTable),
     case emqttd_hooks:run('client.subscribe', [ClientId, Username], TopicTable) of
         {ok, TopicTable1} ->
-            emqttd_session:subscribe(Session, TopicTable1);
+            emqttd_session:subscribe(Session, mount(MountPoint, TopicTable1));
         {stop, _} ->
             ok
     end,
@@ -174,10 +175,11 @@ subscribe(RawTopicTable, ProtoState = #proto_state{client_id = ClientId,
 
 unsubscribe(RawTopics, ProtoState = #proto_state{client_id = ClientId,
                                                  username  = Username,
-                                                 session   = Session}) ->
+                                                 session   = Session,
+                                                 mountpoint = MountPoint}) ->
     case emqttd_hooks:run('client.unsubscribe', [ClientId, Username], parse_topics(RawTopics)) of
         {ok, TopicTable} ->
-            emqttd_session:unsubscribe(Session, TopicTable);
+            emqttd_session:unsubscribe(Session, mount(MountPoint, TopicTable));
         {stop, _} ->
             ok
     end,
@@ -228,7 +230,8 @@ process(?CONNECT_PACKET(Var), State0) ->
                             %% ACCEPT
                             {?CONNACK_ACCEPT, SP, State2#proto_state{session = Session, is_superuser = IsSuperuser}};
                         {error, Error} ->
-                            {stop, {shutdown, Error}, State2}
+                            ?LOG(error, "Username '~s' login failed for ~p", [Username, Error], State2),
+                            {?CONNACK_SERVER, false, State2}
                     end;
                 {error, Reason}->
                     ?LOG(error, "Username '~s' login failed for ~p", [Username, Reason], State1),
@@ -449,14 +452,14 @@ start_keepalive(Sec, #proto_state{keepalive_backoff = Backoff}) when Sec > 0 ->
 
 validate_connect(Connect = #mqtt_packet_connect{}, ProtoState) ->
     case validate_protocol(Connect) of
-        true -> 
+        true ->
             case validate_clientid(Connect, ProtoState) of
-                true -> 
+                true ->
                     ?CONNACK_ACCEPT;
-                false -> 
+                false ->
                     ?CONNACK_INVALID_ID
             end;
-        false -> 
+        false ->
             ?CONNACK_PROTO_VER
     end.
 
@@ -498,7 +501,7 @@ validate_packet(?SUBSCRIBE_PACKET(_PacketId, TopicTable)) ->
 validate_packet(?UNSUBSCRIBE_PACKET(_PacketId, Topics)) ->
     validate_topics(filter, Topics);
 
-validate_packet(_Packet) -> 
+validate_packet(_Packet) ->
     ok.
 
 validate_topics(_Type, []) ->
@@ -593,4 +596,3 @@ unmount(MountPoint, Msg = #mqtt_message{topic = Topic}) ->
         {MountPoint, Topic0} -> Msg#mqtt_message{topic = Topic0};
         _ -> Msg
     end.
-

src/emqttd_server.erl

@@ -92,7 +92,7 @@ publish(Msg = #mqtt_message{from = From}) ->
         {ok, Msg1 = #mqtt_message{topic = Topic}} ->
             emqttd_pubsub:publish(Topic, Msg1);
         {stop, Msg1} ->
-            lager:warning("Stop publishing: ~s", [emqttd_message:format(Msg1)]),
+            lager:info("Stop publishing: ~s", [emqttd_message:format(Msg1)]),
             ignore
     end.
 

src/emqttd_session.erl

@@ -32,7 +32,7 @@
 %%
 %% QoS 1 and QoS 2 messages pending transmission to the Client.
 %%
-%% QoS 2 messages which have been received from the Client, but have not 
+%% QoS 2 messages which have been received from the Client, but have not
 %% been completely acknowledged.
 %%
 %% Optionally, QoS 0 messages pending transmission to the Client.
@@ -720,7 +720,7 @@ enqueue_msg(Msg, State = #state{mqueue = Q}) ->
 %%--------------------------------------------------------------------
 
 redeliver(Msg = #mqtt_message{qos = QoS}, State) ->
-    deliver(Msg#mqtt_message{dup = if QoS =:= ?QOS2 -> false; true -> true end}, State);
+    deliver(Msg#mqtt_message{dup = if QoS =:= ?QOS0 -> false; true -> true end}, State);
 
 redeliver({pubrel, PacketId}, #state{client_pid = Pid}) ->
     Pid ! {redeliver, {?PUBREL, PacketId}}.
@@ -759,7 +759,7 @@ acked(pubrec, PacketId, State = #state{client_id = ClientId,
     case Inflight:lookup(PacketId) of
         {publish, Msg, _Ts} ->
             emqttd_hooks:run('message.acked', [ClientId, Username], Msg),
-            State#state{inflight = Inflight:update(PacketId, {pubrel, PacketId, os:timestamp()})}; 
+            State#state{inflight = Inflight:update(PacketId, {pubrel, PacketId, os:timestamp()})};
         {pubrel, PacketId, _Ts} ->
             ?LOG(warning, "Duplicated PUBREC Packet: ~p", [PacketId], State),
             State
@@ -853,4 +853,3 @@ shutdown(Reason, State) ->
 
 gc(State) ->
     emqttd_gc:maybe_force_gc(#state.force_gc_count, State).
-

src/emqttd_sm.erl

@@ -183,15 +183,16 @@ handle_cast(Msg, State) ->
 handle_info({'DOWN', MRef, process, DownPid, _Reason}, State) ->
     case dict:find(MRef, State#state.monitors) of
         {ok, ClientId} ->
-            case mnesia:dirty_read({mqtt_session, ClientId}) of
+            NewState =
-                [] ->
+              case mnesia:dirty_read({mqtt_session, ClientId}) of
-                    ok;
+                  [] -> State;
-                [Sess = #mqtt_session{sess_pid = DownPid}] ->
+                  [Sess = #mqtt_session{sess_pid = DownPid}] ->
-                    mnesia:dirty_delete_object(Sess);
+                      mnesia:dirty_delete_object(Sess),
-                [_Sess] ->
+                      erase_monitor(MRef, State);
-                    ok
+                  [_Sess] ->
-            end,
+                      State
-            {noreply, erase_monitor(MRef, State), hibernate};
+              end,
+            {noreply, NewState, hibernate};
         error ->
             lager:error("MRef of session ~p not found", [DownPid]),
             {noreply, State}
@@ -256,6 +257,7 @@ resume_session(Session = #mqtt_session{client_id = ClientId, sess_pid = SessPid}
             {ok, SessPid};
         false ->
             ?LOG(error, "Cannot resume ~p which seems already dead!", [SessPid], Session),
+            remove_session(Session),
             {error, session_died}
     end;
 
@@ -305,4 +307,3 @@ monitor_session(ClientId, SessPid, State = #state{monitors = Monitors}) ->
 erase_monitor(MRef, State = #state{monitors = Monitors}) ->
     erlang:demonitor(MRef, [flush]),
     State#state{monitors = dict:erase(MRef, Monitors)}.
-

src/emqttd_stats.erl

@@ -143,7 +143,7 @@ setstat(Stat, Val) ->
     ets:update_element(?STATS_TAB, Stat, {2, Val}).
 
 %% @doc Set stats with max
--spec(setstats(Stat :: atom(), MaxStat :: atom(), Val :: pos_integer()) -> boolean()).
+-spec(setstats(Stat :: atom(), MaxStat :: atom(), Val :: pos_integer()) -> ok).
 setstats(Stat, MaxStat, Val) ->
     gen_server:cast(?MODULE, {setstats, Stat, MaxStat, Val}).
 

src/emqttd_vm.erl

@@ -156,9 +156,9 @@ microsecs() ->
     (Mega * 1000000 + Sec) * 1000000 + Micro.
 
 loads() ->
-    [{load1,  ftos(cpu_sup:avg1()/256)},
+    [{load1,  ftos(avg1()/256)},
-     {load5,  ftos(cpu_sup:avg5()/256)},
+     {load5,  ftos(avg5()/256)},
-     {load15, ftos(cpu_sup:avg15()/256)}].
+     {load15, ftos(avg15()/256)}].
 
 get_system_info() ->
     [{Key, format_system_info(Key, get_system_info(Key))} || Key <- ?SYSTEM_INFO].
@@ -427,3 +427,27 @@ mapping([{owner, V}|Entries], Acc) when is_pid(V) ->
 mapping([{Key, Value}|Entries], Acc) ->
     mapping(Entries, [{Key, Value}|Acc]).
 
+avg1() ->
+   case cpu_sup:avg1() of
+        SystemLoad when is_integer(SystemLoad) ->
+            SystemLoad;
+        {error, Reason} ->
+            lager:error("Get the average system load in the last minute fail for ~p~n", [Reason]),
+            0.00
+    end.
+avg5() ->
+    case cpu_sup:avg5() of
+        SystemLoad when is_integer(SystemLoad) ->
+            SystemLoad;
+        {error, Reason} ->
+            lager:error("Get the average system load in the last 5 minutes fail for ~p~n", [Reason]),
+            0.00
+    end.
+avg15() ->
+    case cpu_sup:avg15() of
+        SystemLoad when is_integer(SystemLoad) ->
+            SystemLoad;
+        {error, Reason} ->
+            lager:error("Get the average system load in the last 15 minutes fail for ~p~n", [Reason]),
+            0.00
+    end.