feat(dao) cluster spawning in init_by_lua

kong/cli/migrations.lua

@@ -49,7 +49,7 @@ if args.command == "list" then
   elseif migrations then
     cutils.logger:info(string.format(
       "Executed migrations for keyspace %s (%s):",
-      cutils.colors.yellow(dao_factory._properties.keyspace),
+      cutils.colors.yellow(dao_factory.properties.keyspace),
       dao_factory.type
     ))
 
@@ -63,7 +63,7 @@ if args.command == "list" then
     cutils.logger:info(string.format(
       "No migrations have been run yet for %s on keyspace: %s",
       cutils.colors.yellow(dao_factory.type),
-      cutils.colors.yellow(dao_factory._properties.keyspace)
+      cutils.colors.yellow(dao_factory.properties.keyspace)
     ))
   end
 
@@ -73,7 +73,7 @@ elseif args.command == "up" then
     cutils.logger:info(string.format(
       "Migrating %s on keyspace \"%s\" (%s)",
       cutils.colors.yellow(identifier),
-      cutils.colors.yellow(dao_factory._properties.keyspace),
+      cutils.colors.yellow(dao_factory.properties.keyspace),
       dao_factory.type
     ))
   end
@@ -110,7 +110,7 @@ elseif args.command == "down" then
     cutils.logger:info(string.format(
       "Rollbacking %s in keyspace \"%s\" (%s)",
       cutils.colors.yellow(identifier),
-      cutils.colors.yellow(dao_factory._properties.keyspace),
+      cutils.colors.yellow(dao_factory.properties.keyspace),
       dao_factory.type
     ))
   end
@@ -130,7 +130,7 @@ elseif args.command == "down" then
 
 elseif args.command == "reset" then
 
-  local keyspace = dao_factory._properties.keyspace
+  local keyspace = dao_factory.properties.keyspace
 
   cutils.logger:info(string.format(
     "Resetting \"%s\" keyspace (%s)",

kong/cli/utils/signal.lua

@@ -163,7 +163,7 @@ local function prepare_database(args_config)
     cutils.logger:info(string.format(
       "Migrating %s on keyspace \"%s\" (%s)",
       cutils.colors.yellow(identifier),
-      cutils.colors.yellow(dao_factory._properties.keyspace),
+      cutils.colors.yellow(dao_factory.properties.keyspace),
       dao_factory.type
     ))
   end
@@ -230,7 +230,7 @@ end
 -- Checks whether a port is available. Exits the application if not available.
 -- @param port The port to check
 -- @param name Functional name the port is used for (display name)
--- @param timeout (optional) Timeout in seconds after which a failure is logged 
+-- @param timeout (optional) Timeout in seconds after which a failure is logged
 -- and application exit is performed, if not provided then it will fail at once without retries.
 local function check_port(port, name, timeout)
   local expire = socket.gettime() + (timeout or 0)
@@ -267,9 +267,9 @@ function _M.send_signal(args_config, signal)
   if not signal then signal = START end
 
   if signal == START then
-    local ports = { 
-      ["Kong proxy"] = kong_config.proxy_port, 
-      ["Kong proxy ssl"] = kong_config.proxy_ssl_port, 
+    local ports = {
+      ["Kong proxy"] = kong_config.proxy_port,
+      ["Kong proxy ssl"] = kong_config.proxy_ssl_port,
       ["Kong admin api"] = kong_config.admin_api_port
     }
     for name, port in pairs(ports) do

kong/dao/cassandra/base_dao.lua

@@ -371,7 +371,7 @@ function BaseDao:delete(where_t)
 
   -- Delete successful, trigger cascade delete hooks if any.
   local foreign_err
-  for _, hook in ipairs(self._cascade_delete_hooks) do
+  for _, hook in ipairs(self.cascade_delete_hooks) do
     foreign_err = select(2, hook(t_primary_key))
     if foreign_err then
       return false, foreign_err
@@ -418,8 +418,8 @@ function BaseDao:new(properties)
     }
   end
 
-  self._properties = properties
-  self._cascade_delete_hooks = {}
+  self.properties = properties
+  self.cascade_delete_hooks = {}
 end
 
 ---
@@ -613,7 +613,7 @@ function BaseDao:add_delete_hook(foreign_dao_name, foreign_column, parent_column
     return true
   end
 
-  table.insert(self._cascade_delete_hooks, delete_hook)
+  table.insert(self.cascade_delete_hooks, delete_hook)
 end
 
 return BaseDao

kong/dao/cassandra/factory.lua

@@ -11,8 +11,11 @@ local stringy = require "stringy"
 local Object = require "classic"
 local utils = require "kong.tools.utils"
 
--- Silent outside of ngx_lua logging
-cassandra.set_log_level("QUIET")
+if ngx ~= nil and type(ngx.get_phase) == "function" and ngx.get_phase() == "init" then
+  cassandra.set_log_level("INFO")
+else
+  cassandra.set_log_level("QUIET")
+end
 
 local CassandraFactory = Object:extend()
 
@@ -28,7 +31,16 @@ end
 -- Instantiate a Cassandra Factory and all its DAOs for various entities
 -- @param `properties` Cassandra properties
 function CassandraFactory:new(properties, plugins)
-  self._properties = properties
+  local ok, err = cassandra.spawn_cluster {
+    shm = "cassandra",
+    prepared_shm = "cassandra_prepared",
+    contact_points = properties.contact_points
+  }
+  if not ok then
+    error(err)
+  end
+
+  self.properties = properties
   self.type = "cassandra"
   self.daos = {}
 
@@ -66,7 +78,7 @@ end
 function CassandraFactory:load_daos(plugin_daos)
   local dao
   for name, plugin_dao in pairs(plugin_daos) do
-    dao = plugin_dao(self._properties)
+    dao = plugin_dao(self.properties)
     dao._factory = self
     self.daos[name] = dao
     if dao._schema then
@@ -103,15 +115,15 @@ function CassandraFactory:get_session_options()
   return {
     shm = "cassandra",
     prepared_shm = "cassandra_prepared",
-    contact_points = self._properties.contact_points,
-    keyspace = self._properties.keyspace,
+    contact_points = self.properties.contact_points,
+    keyspace = self.properties.keyspace,
     query_options = {
       prepare = true
     },
     ssl_options = {
-      enabled = self._properties.ssl.enabled,
-      verify = self._properties.ssl.verify,
-      ca = self._properties.ssl.certificate_authority
+      enabled = self.properties.ssl.enabled,
+      verify = self.properties.ssl.verify,
+      ca = self.properties.ssl.certificate_authority
     }
   }
 end

kong/dao/cassandra/migrations.lua

@@ -28,7 +28,7 @@ function Migrations:new(properties)
 end
 
 function Migrations:keyspace_exists(keyspace)
-  local rows, err = Migrations.super.execute(self, self.queries.get_keyspace, {self._properties.keyspace}, nil, "system")
+  local rows, err = Migrations.super.execute(self, self.queries.get_keyspace, {self.properties.keyspace}, nil, "system")
   if err then
     return nil, err
   else

kong/tools/migrations.lua

@@ -14,7 +14,7 @@ function Migrations:new(dao, kong_config, core_migrations_module, plugins_namesp
   dao:load_daos(require("kong.dao."..dao.type..".migrations"))
 
   self.dao = dao
-  self.dao_properties = dao._properties
+  self.dao_properties = dao.properties
   self.migrations = {
     [_CORE_MIGRATIONS_IDENTIFIER] = require(core_migrations_module)
   }