Merge pull request #201 from Mashape/refactor/migrations

[refactor] separate migrations from the Factory

kong-0.2.1-1.rockspec

@@ -71,11 +71,12 @@ build = {
     ["kong.dao.cassandra.factory"] = "kong/dao/cassandra/factory.lua",
     ["kong.dao.cassandra.base_dao"] = "kong/dao/cassandra/base_dao.lua",
     ["kong.dao.cassandra.apis"] = "kong/dao/cassandra/apis.lua",
+    ["kong.dao.cassandra.consumers"] = "kong/dao/cassandra/consumers.lua",
+    ["kong.dao.cassandra.plugins_configurations"] = "kong/dao/cassandra/plugins_configurations.lua",
+    ["kong.dao.cassandra.migrations"] = "kong/dao/cassandra/migrations.lua",
     ["kong.dao.cassandra.ratelimiting_metrics"] = "kong/dao/cassandra/ratelimiting_metrics.lua",
     ["kong.dao.cassandra.basicauth_credentials"] = "kong/dao/cassandra/basicauth_credentials.lua",
     ["kong.dao.cassandra.keyauth_credentials"] = "kong/dao/cassandra/keyauth_credentials.lua",
-    ["kong.dao.cassandra.plugins_configurations"] = "kong/dao/cassandra/plugins_configurations.lua",
-    ["kong.dao.cassandra.consumers"] = "kong/dao/cassandra/consumers.lua",
 
     ["kong.plugins.base_plugin"] = "kong/plugins/base_plugin.lua",
 

kong/cli/migrations.lua

@@ -29,7 +29,7 @@ local migrations = Migrations(dao_factory, cutils.get_luarocks_install_dir())
 
 if args.command == "list" then
 
-  local migrations, err = dao_factory:get_migrations()
+  local migrations, err = dao_factory.migrations:get_migrations()
   if err then
     cutils.logger:error_exit(err)
   elseif migrations then

kong/cli/utils/signal.lua

@@ -137,7 +137,7 @@ local function prepare_database(args_config)
   local _, _, dao_factory = get_kong_config(args_config)
 
   -- Migrate the DB if needed and possible
-  local keyspace, err = dao_factory:get_migrations()
+  local keyspace, err = dao_factory.migrations:get_migrations()
   if err then
     cutils.logger:error_exit(err)
   elseif keyspace == nil then

kong/dao/cassandra/base_dao.lua

@@ -138,9 +138,10 @@ function BaseDao:_check_all_unique(t, is_updating)
 end
 
 -- Open a Cassandra session on the configured keyspace.
+-- @param `keyspace` (Optional) Override the keyspace for this session if specified.
 -- @return `session` Opened session
 -- @return `error`   Error if any
-function BaseDao:_open_session()
+function BaseDao:_open_session(keyspace)
   local ok, err
 
   -- Start cassandra session
@@ -158,7 +159,7 @@ function BaseDao:_open_session()
   end
 
   if times == 0 or not times then
-    ok, err = session:set_keyspace(self._properties.keyspace)
+    ok, err = session:set_keyspace(keyspace and keyspace or self._properties.keyspace)
     if not ok then
       return nil, DaoError(err, error_types.DATABASE)
     end
@@ -259,10 +260,11 @@ end
 -- @param `statement` Prepared statement, plain string query or BatchStatement.
 -- @param `args`      (Optional) Arguments to the query, simply passed to lua-resty-cassandra's :execute()
 -- @param `options`   (Optional) Options to give to lua-resty-cassandra's :execute()
+-- @param `keyspace`  (Optional) Override the keyspace for this query if specified.
 -- @return `results`  If results set are ROWS, a table with an array of unmarshalled rows and a `next_page` property if the results have a paging_state.
 -- @return `error`    An error if any during the whole execution (sockets/query execution)
-function BaseDao:_execute(statement, args, options)
-  local session, err = self:_open_session()
+function BaseDao:_execute(statement, args, options, keyspace)
+  local session, err = self:_open_session(keyspace)
   if err then
     return nil, err
   end

kong/dao/cassandra/factory.lua

@@ -13,6 +13,7 @@ local Object = require "classic"
 local Apis = require "kong.dao.cassandra.apis"
 local Consumers = require "kong.dao.cassandra.consumers"
 local PluginsConfigurations = require "kong.dao.cassandra.plugins_configurations"
+local Migrations = require "kong.dao.cassandra.migrations"
 local BasicAuthCredentials = require "kong.dao.cassandra.basicauth_credentials"
 local RateLimitingMetrics = require "kong.dao.cassandra.ratelimiting_metrics"
 local KeyAuthCredentials = require "kong.dao.cassandra.keyauth_credentials"
@@ -54,6 +55,8 @@ function CassandraFactory:new(properties)
   self.basicauth_credentials = BasicAuthCredentials(properties)
   self.ratelimiting_metrics = RateLimitingMetrics(properties)
   self.keyauth_credentials = KeyAuthCredentials(properties)
+
+  self.migrations = Migrations(properties)
 end
 
 function CassandraFactory:drop()
@@ -142,81 +145,4 @@ function CassandraFactory:execute_queries(queries, no_keyspace)
   session:close()
 end
 
---
--- Migrations
---
-
-local MIGRATION_IDENTIFIER = "migrations"
-
--- Create a cassandra session and execute a query on given keyspace or default one (from properties).
--- @param query Query or prepared statement given to session:execute
--- @param args List of arguments given to session:execute
--- @param keyspace Optional: overrides properties keyspace if specified
--- @return query result
--- @return error if any
-function CassandraFactory:execute(query, args, keyspace)
-  local ok, err
-  local session = cassandra.new()
-  session:set_timeout(self._properties.timeout)
-
-  ok, err = session:connect(self._properties.hosts, self._properties.port)
-  if not ok then
-    return nil, DaoError(err, constants.DATABASE_ERROR_TYPES.DATABASE)
-  end
-
-  ok, err = session:set_keyspace(keyspace and keyspace or self._properties.keyspace)
-  if not ok then
-    return nil, DaoError(err, constants.DATABASE_ERROR_TYPES.DATABASE)
-  end
-
-  ok, err = session:execute(query, args)
-
-  session:close()
-
-  if not ok then
-    return nil, DaoError(err, constants.DATABASE_ERROR_TYPES.DATABASE)
-  end
-
-  return ok
-end
-
--- Log (add) given migration to schema_migrations table.
--- @param migration_name Name of the migration to log
--- @return query result
--- @return error if any
-function CassandraFactory:add_migration(migration_name)
-  return self:execute("UPDATE schema_migrations SET migrations = migrations + ? WHERE id = ?",
-                      { cassandra.list({ migration_name }), MIGRATION_IDENTIFIER })
-end
-
--- Return all logged migrations if any. Check if keyspace exists before to avoid error during the first migration.
--- @return A list of previously executed migration (as strings)
--- @return error if any
-function CassandraFactory:get_migrations()
-  local rows, err
-
-  rows, err = self:execute("SELECT * FROM schema_keyspaces WHERE keyspace_name = ?", { self._properties.keyspace }, "system")
-  if err then
-    return nil, err
-  elseif #rows == 0 then
-    -- keyspace is not yet created, this is the first migration
-    return nil
-  end
-
-  rows, err = self:execute("SELECT migrations FROM schema_migrations WHERE id = ?", { MIGRATION_IDENTIFIER })
-  if err then
-    return nil, err
-  elseif rows and #rows > 0 then
-    return rows[1].migrations
-  end
-end
-
--- Unlog (delete) given migration from the schema_migrations table.
--- @return query result
--- @return error if any
-function CassandraFactory:delete_migration(migration_name)
-  return self:execute("UPDATE schema_migrations SET migrations = migrations - ? WHERE id = ?",
-                      { cassandra.list({ migration_name }), MIGRATION_IDENTIFIER })
-end
-
 return CassandraFactory

kong/dao/cassandra/migrations.lua

@@ -0,0 +1,65 @@
+local cassandra = require "cassandra"
+local BaseDao = require "kong.dao.cassandra.base_dao"
+
+local Migrations = BaseDao:extend()
+
+function Migrations:new(properties)
+  self._queries = {
+    add_migration = [[
+      UPDATE schema_migrations SET migrations = migrations + ? WHERE id = 'migrations';
+    ]],
+    get_keyspace = [[
+      SELECT * FROM system.schema_keyspaces WHERE keyspace_name = ?;
+    ]],
+    get_migrations = [[
+      SELECT migrations FROM schema_migrations WHERE id = 'migrations';
+    ]],
+    delete_migration = [[
+      UPDATE schema_migrations SET migrations = migrations - ? WHERE id = 'migrations';
+    ]]
+  }
+
+  Migrations.super.new(self, properties)
+end
+
+-- Log (add) given migration to schema_migrations table.
+-- @param migration_name Name of the migration to log
+-- @return query result
+-- @return error if any
+function Migrations:add_migration(migration_name)
+  return Migrations.super._execute(self, self._queries.add_migration,
+    { cassandra.list({ migration_name }) })
+end
+
+-- Return all logged migrations if any. Check if keyspace exists before to avoid error during the first migration.
+-- @return A list of previously executed migration (as strings)
+-- @return error if any
+function Migrations:get_migrations()
+  local rows, err
+
+  rows, err = Migrations.super._execute(self, self._queries.get_keyspace,
+    { self._properties.keyspace }, nil, "system")
+  if err then
+    return nil, err
+  elseif #rows == 0 then
+    -- keyspace is not yet created, this is the first migration
+    return nil
+  end
+
+  rows, err = Migrations.super._execute(self, self._queries.get_migrations)
+  if err then
+    return nil, err
+  elseif rows and #rows > 0 then
+    return rows[1].migrations
+  end
+end
+
+-- Unlog (delete) given migration from the schema_migrations table.
+-- @return query result
+-- @return error if any
+function Migrations:delete_migration(migration_name)
+  return Migrations.super._execute(self, self._queries.delete_migration,
+    { cassandra.list({ migration_name }) })
+end
+
+return Migrations

kong/tools/migrations.lua

@@ -53,7 +53,7 @@ end
 -- Execute all migrations UP
 -- @param callback A function to execute on each migration (ie: for logging)
 function Migrations:migrate(callback)
-  local old_migrations, err = self.dao:get_migrations()
+  local old_migrations, err = self.dao.migrations:get_migrations()
   if err then
     callback(nil, err)
     return
@@ -92,7 +92,7 @@ function Migrations:migrate(callback)
     end
 
     -- Record migration in db
-    local _, err = self.dao:add_migration(migration.name)
+    local _, err = self.dao.migrations:add_migration(migration.name)
     if err then
       err = "Cannot record migration "..migration.name..": "..err
     end
@@ -107,7 +107,7 @@ end
 -- Take the latest executed migration and DOWN it
 -- @param callback A function to execute (for consistency with other functions of this module)
 function Migrations:rollback(callback)
-  local old_migrations, err = self.dao:get_migrations()
+  local old_migrations, err = self.dao.migrations:get_migrations()
   if err then
     callback(nil, err)
     return
@@ -133,7 +133,7 @@ function Migrations:rollback(callback)
   -- delete migration from schema changes records if it's not the first one
   -- (otherwise the schema_migrations table doesn't exist anymore)
   if not migration_to_rollback.init then
-    local _, err = self.dao:delete_migration(migration_to_rollback.name)
+    local _, err = self.dao.migrations:delete_migration(migration_to_rollback.name)
     if err then
       callback(migration_to_rollback, "Cannot delete migration "..migration_to_rollback.name..": "..err)
       return