Adding missing serf commands and cluster CLI

kong/cmd/cluster.lua

@@ -0,0 +1,54 @@
+local pl_app = require "pl.lapp"
+local conf_loader = require "kong.conf_loader"
+local DAOFactory = require "kong.dao.factory"
+local Serf = require "kong.serf"
+local log = require "kong.cmd.utils.log"
+local fmt = string.format
+
+local function execute(args)
+  local conf = assert(conf_loader(args.conf, {
+    prefix = args.prefix
+  }))
+
+  local dao = DAOFactory(conf)
+  local serf = Serf.new(conf, conf.prefix, dao)
+
+  if args.command == "members" then
+    local members = assert(serf:members(true))
+    for _, v in ipairs(members) do
+      print(fmt("%s\t%s\t%s", v.name, v.addr, v.status))
+    end
+  elseif args.command == "keygen" then
+    print(assert(serf:keygen()))
+  elseif args.command == "reachability" then
+    log("Please wait..")
+    print(assert(serf:reachability()))
+  elseif args.command == "force-leave" then
+    local node_name = args[1]
+    if not node_name then
+      pl_app.quit("You need to specify the node name to leave")
+    end
+    log(fmt("Force-leaving %s", node_name))
+    assert(serf:force_leave(node_name))
+    log("Done")
+  end
+end
+
+local lapp = [[
+Usage: kong cluster COMMAND [OPTIONS]
+
+The available commands are:
+ members
+ force-leave <node_name>
+ keygen
+ reachability
+
+Options:
+ -c,--conf (optional string) configuration file
+]]
+
+return {
+  lapp = lapp,
+  execute = execute,
+  sub_commands = {members = true, keygen = true, reachability = true, ["force-leave"] = true}
+}

kong/cmd/init.lua

@@ -29,6 +29,7 @@ local cmds = {
   check = "check",
   compile = "compile",
   migrations = "migrations",
+  cluster = "cluster",
   version = "version",
   roar = "roar"
 }

kong/cmd/stop.lua

@@ -2,6 +2,7 @@ local nginx_signals = require "kong.cmd.utils.nginx_signals"
 local serf_signals = require "kong.cmd.utils.serf_signals"
 local dnsmasq_signals = require "kong.cmd.utils.dnsmasq_signals"
 local conf_loader = require "kong.conf_loader"
+local DAOFactory = require "kong.dao.factory"
 local log = require "kong.cmd.utils.log"
 
 local function execute(args)
@@ -11,9 +12,10 @@ local function execute(args)
     prefix = args.prefix
   }))
 
+  local dao = DAOFactory(conf)
+
   assert(nginx_signals.stop(conf.prefix))
-  assert(serf_signals.stop(conf.prefix))
-  assert(serf_signals.stop(conf.prefix))
+  assert(serf_signals.stop(conf, conf.prefix, dao))
   assert(dnsmasq_signals.stop(conf.prefix))
   log("Stopped")
 end

kong/cmd/utils/nginx_conf_compiler.lua

@@ -1,4 +1,5 @@
 local NGINX_VARS = {
+  prefix = true,
   plugins = true,
   cluster_listen = true,
   cluster_listen_rpc = true,

kong/cmd/utils/serf_signals.lua

@@ -10,10 +10,12 @@ local pl_path = require "pl.path"
 local pl_file = require "pl.file"
 local kill = require "kong.cmd.utils.kill"
 local log = require "kong.cmd.utils.log"
+local utils = require "kong.tools.utils"
 local fmt = string.format
 
 local serf_bin_name = "serf"
 local serf_pid_name = "serf.pid"
+local serf_node_id = "serf.id"
 local serf_event_name = "kong"
 local start_timeout = 2
 
@@ -55,11 +57,27 @@ client:request { \
 resty -e "$CMD"
 ]]
 
+local function prepare_identifier(kong_config, nginx_prefix)
+  local id_path = pl_path.join(nginx_prefix, serf_node_id)
+  if not pl_path.exists(id_path) then
+    local id = utils.get_hostname().."_"..kong_config.cluster_listen.."_"..utils.random_string()
+
+    log.verbose("saving Serf identifier in %s", id_path)
+    local ok, err = pl_file.write(id_path, id)
+    if not ok then return nil, err end
+  end
+  return true
+end
+
 local function prepare_prefix(kong_config, nginx_prefix, script_path)
+  local ok, err = prepare_identifier(kong_config, nginx_prefix)
+  if not ok then return nil, err end
+
   log.verbose("dumping Serf shell script handler in %s", script_path)
   local script = fmt(script_template, "127.0.0.1", kong_config.admin_port)
 
-  pl_file.write(script_path, script)
+  local ok, err = pl_file.write(script_path, script)
+  if not ok then return nil, err end
   local ok, _, _, stderr = pl_utils.executeex("chmod +x "..script_path)
   if not ok then return nil, stderr end
 
@@ -85,20 +103,20 @@ function _M.start(kong_config, nginx_prefix, dao)
     pl_file.delete(pid_path)
   end
 
+  -- prepare shell script
+  local script_path = pl_path.join(nginx_prefix, "serf_event.sh")
+  local ok, err = prepare_prefix(kong_config, nginx_prefix, script_path)
+  if not ok then return nil, err end
+
   -- make sure Serf is in PATH
   local ok, err = check_serf_bin()
   if not ok then return nil, err end
 
-  local serf = Serf.new(kong_config, dao)
+  local serf = Serf.new(kong_config, nginx_prefix, dao)
 
   local node_name = serf.node_name
-  local script_path = pl_path.join(nginx_prefix, "serf_event.sh")
   local log_path = pl_path.join(nginx_prefix, "serf.log")
 
-  -- prepare shell script
-  local ok, err = prepare_prefix(kong_config, nginx_prefix, script_path)
-  if not ok then return nil, err end
-
   local args = setmetatable({
     ["-bind"] = kong_config.cluster_listen,
     ["-rpc-addr"] = kong_config.cluster_listen_rpc,
@@ -152,7 +170,14 @@ function _M.start(kong_config, nginx_prefix, dao)
   return true
 end
 
-function _M.stop(nginx_prefix)
+function _M.stop(kong_config, nginx_prefix, dao)
+  log.info("Leaving cluster")
+
+  local serf = Serf.new(kong_config, nginx_prefix, dao)
+
+  local ok, err = serf:leave()
+  if not ok then return nil, err end
+
   local pid_path = pl_path.join(nginx_prefix, serf_pid_name)
   log.verbose("stopping Serf agent at %s", pid_path)
   return kill(pid_path, "-9")

kong/kong.lua

@@ -126,7 +126,7 @@ function Kong.init(config)
 
   -- populate singletons
   singletons.loaded_plugins = assert(load_plugins(config, dao, events))
-  singletons.serf = Serf.new(config, dao)
+  singletons.serf = Serf.new(config, config.prefix, dao)
   singletons.dao = dao
   singletons.events = events
   singletons.configuration = config

kong/serf.lua

@@ -3,13 +3,13 @@
 
 local pl_stringx = require "pl.stringx"
 local pl_utils = require "pl.utils"
+local pl_path = require "pl.path"
+local pl_file = require "pl.file"
 local cjson = require "cjson.safe"
 local log = require "kong.cmd.utils.log"
 local fmt = string.format
 
-local ok, _, stdout, stderr = pl_utils.executeex "/bin/hostname"
-if not ok then error(stderr) end
-local hostname = pl_stringx.strip(stdout)
+local serf_node_id = "serf.id"
 
 local Serf = {}
 Serf.__index = Serf
@@ -22,9 +22,9 @@ Serf.args_mt = {
   end
 }
 
-function Serf.new(kong_config, dao)
+function Serf.new(kong_config, nginx_prefix, dao)
   return setmetatable({
-    node_name = hostname.."_"..kong_config.cluster_listen,
+    node_name = assert(pl_file.read(pl_path.join(nginx_prefix, serf_node_id))),
     config = kong_config,
     dao = dao
   }, Serf)
@@ -40,7 +40,7 @@ function Serf:invoke_signal(signal, args, no_rpc)
   local rpc = no_rpc and "" or "-rpc-addr="..self.config.cluster_listen_rpc
   local cmd = fmt("serf %s %s %s", signal, rpc, tostring(args))
   local ok, code, stdout = pl_utils.executeex(cmd)
-  if not ok or code ~= 0 then return nil, stdout end
+  if not ok or code ~= 0 then return nil, pl_stringx.splitlines(stdout)[1] end -- always print the first error line of serf
 
   return stdout
 end
@@ -49,6 +49,23 @@ function Serf:join_node(address)
   return select(2, self:invoke_signal("join", address)) == nil
 end
 
+function Serf:leave()
+  local res, err = self:invoke_signal("leave")
+  if not res then return nil, err end
+
+  local _, err = self.dao.nodes:delete {name = self.node_name}
+  if err then return nil, err end
+
+  return true
+end
+
+function Serf:force_leave(node_name)
+  local res, err = self:invoke_signal("force-leave", node_name)
+  if not res then return nil, err end
+
+  return true
+end
+
 function Serf:members()
   local res, err = self:invoke_signal("members", {["-format"] = "json"})
   if not res then return nil, err end
@@ -59,6 +76,18 @@ function Serf:members()
   return json.members
 end
 
+function Serf:keygen()
+  local res, err = self:invoke_signal("keygen")
+  if not res then return nil, err end
+  return res
+end
+
+function Serf:reachability()
+  local res, err = self:invoke_signal("reachability")
+  if not res then return nil, err end
+  return res
+end
+
 function Serf:autojoin()
   -- Delete current node just in case it was there
   -- (due to an inconsistency caused by a crash)
@@ -110,7 +139,7 @@ function Serf:add_node()
   local _, err = self.dao.nodes:insert({
     name = self.node_name,
     cluster_listening_address = pl_stringx.strip(addr)
-  }, {ttl = 3600})
+  }, {ttl = self.config.cluster_ttl_on_failure})
   if err then return nil, tostring(err) end
 
   return true