feat(db) move consumers from kong.dao to kong.db

kong-0.13.1-0.rockspec

@@ -108,7 +108,6 @@ build = {
     ["kong.dao.errors"] = "kong/dao/errors.lua",
     ["kong.dao.schemas_validation"] = "kong/dao/schemas_validation.lua",
     ["kong.dao.schemas.apis"] = "kong/dao/schemas/apis.lua",
-    ["kong.dao.schemas.consumers"] = "kong/dao/schemas/consumers.lua",
     ["kong.dao.schemas.plugins"] = "kong/dao/schemas/plugins.lua",
     ["kong.dao.schemas.upstreams"] = "kong/dao/schemas/upstreams.lua",
     ["kong.dao.schemas.targets"] = "kong/dao/schemas/targets.lua",
@@ -127,7 +126,9 @@ build = {
     ["kong.db.dao"] = "kong/db/dao/init.lua",
     ["kong.db.dao.certificates"] = "kong/db/dao/certificates.lua",
     ["kong.db.dao.snis"] = "kong/db/dao/snis.lua",
+    ["kong.db.dao.consumers"] = "kong/db/dao/consumers.lua",
     ["kong.db.schema"] = "kong/db/schema/init.lua",
+    ["kong.db.schema.entities.consumers"] = "kong/db/schema/entities/consumers.lua",
     ["kong.db.schema.entities.routes"] = "kong/db/schema/entities/routes.lua",
     ["kong.db.schema.entities.services"] = "kong/db/schema/entities/services.lua",
     ["kong.db.schema.entities.certificates"] = "kong/db/schema/entities/certificates.lua",

kong/api/crud_helpers.lua

@@ -3,7 +3,6 @@ local utils         = require "kong.tools.utils"
 local responses     = require "kong.tools.responses"
 local app_helpers   = require "lapis.application"
 
-
 local decode_base64 = ngx.decode_base64
 local encode_base64 = ngx.encode_base64
 local encode_args   = ngx.encode_args
@@ -80,22 +79,37 @@ function _M.find_plugin_by_filter(self, dao_factory, filter, helpers)
   end
 end
 
+
 function _M.find_consumer_by_username_or_id(self, dao_factory, helpers)
-  local rows, err = _M.find_by_id_or_field(dao_factory.consumers, {},
-                                           self.params.username_or_id, "username")
+  local username_or_id = self.params.username_or_id
+  local db = assert(dao_factory.db.new_db)
+  local consumer, err
+  if utils.is_valid_uuid(username_or_id) then
+    consumer, err = db.consumers:select({ id = username_or_id })
 
-  if err then
-    return helpers.yield_error(err)
+    if err then
+      return helpers.yield_error(err)
+    end
   end
+
+  if not consumer then
+    consumer, err = db.consumers:select_by_username(username_or_id)
+
+    if err then
+      return helpers.yield_error(err)
+    end
+  end
+
   self.params.username_or_id = nil
 
   -- We know username and id are unique, so if we have a row, it must be the only one
-  self.consumer = rows[1]
+  self.consumer = consumer
   if not self.consumer then
     return helpers.responses.send_HTTP_NOT_FOUND()
   end
 end
 
+
 function _M.find_upstream_by_name_or_id(self, dao_factory, helpers)
   local rows, err = _M.find_by_id_or_field(dao_factory.upstreams, {},
                                            self.params.upstream_name_or_id, "name")

kong/api/routes/consumers.lua

@@ -1,42 +1,11 @@
 local crud = require "kong.api.crud_helpers"
 
 return {
-  ["/consumers/"] = {
-    GET = function(self, dao_factory)
-      crud.paginated_set(self, dao_factory.consumers)
-    end,
-
-    PUT = function(self, dao_factory)
-      crud.put(self.params, dao_factory.consumers)
-    end,
-
-    POST = function(self, dao_factory)
-      crud.post(self.params, dao_factory.consumers)
-    end
-  },
-
-  ["/consumers/:username_or_id"] = {
-    before = function(self, dao_factory, helpers)
-      self.params.username_or_id = ngx.unescape_uri(self.params.username_or_id)
-      crud.find_consumer_by_username_or_id(self, dao_factory, helpers)
-    end,
-
-    GET = function(self, dao_factory, helpers)
-      return helpers.responses.send_HTTP_OK(self.consumer)
-    end,
-
-    PATCH = function(self, dao_factory)
-      crud.patch(self.params, dao_factory.consumers, self.consumer)
-    end,
-
-    DELETE = function(self, dao_factory)
-      crud.delete(self.consumer, dao_factory.consumers)
-    end
-  },
 
-  ["/consumers/:username_or_id/plugins/"] = {
+  ["/consumers/:consumers/plugins"] = {
     before = function(self, dao_factory, helpers)
-      self.params.username_or_id = ngx.unescape_uri(self.params.username_or_id)
+      self.params.username_or_id = ngx.unescape_uri(self.params.consumers)
+      self.params.consumers = nil
       crud.find_consumer_by_username_or_id(self, dao_factory, helpers)
       self.params.consumer_id = self.consumer.id
     end,
@@ -54,9 +23,10 @@ return {
     end
   },
 
-  ["/consumers/:username_or_id/plugins/:id"] = {
+  ["/consumers/:consumers/plugins/:id"] = {
     before = function(self, dao_factory, helpers)
-      self.params.username_or_id = ngx.unescape_uri(self.params.username_or_id)
+      self.params.username_or_id = ngx.unescape_uri(self.params.consumers)
+      self.params.consumers = nil
       crud.find_consumer_by_username_or_id(self, dao_factory, helpers)
       crud.find_plugin_by_filter(self, dao_factory, {
         consumer_id = self.consumer.id,

kong/dao/factory.lua

@@ -9,7 +9,6 @@ local fmt = string.format
 
 local CORE_MODELS = {
   "apis",
-  "consumers",
   "plugins",
   "upstreams",
   "targets",
@@ -40,7 +39,9 @@ local function build_constraints(schemas)
       if type(field.foreign) == "string" then
         local f_entity, f_field = unpack(utils.split(field.foreign, ":"))
         if f_entity ~= nil and f_field ~= nil then
-          local f_schema = schemas[f_entity]
+          local f_schema = assert(schemas[f_entity], "could not find schema for " ..
+                                  f_entity .. " when parsing constraints for " ..
+                                  m_name)
           constraints.foreign[col] = {
             table = f_schema.table,
             schema = f_schema,

kong/dao/migrations/cassandra.lua

@@ -1,6 +1,7 @@
 local log = require "kong.cmd.utils.log"
 local cassandra = require "cassandra"
 local utils = require "kong.tools.utils"
+local helpers = require "kong.dao.migrations.helpers"
 
 local migration_helpers = require "kong.dao.migrations.helpers"
 
@@ -732,14 +733,41 @@ return {
         return err
       end
     end,
-    down = nil
+    down = nil,
   },
   { name = "2018-03-27-002500_drop_old_ssl_tables",
     up = [[
       DROP INDEX ssl_servers_names_ssl_certificate_id_idx;
       DROP TABLE ssl_certificates;
       DROP TABLE ssl_servers_names;
     ]],
-    down = nil
+    down = nil,
+  },
+  {
+    name = "2018-03-13-160000_partition_consumers",
+    up = function(db, kong_config, dao)
+      local consumers = {
+        name = "consumers",
+        columns = {
+          id = "uuid",
+          created_at = "timestamp",
+          custom_id = "text",
+          username = "text",
+        },
+        partition_keys = { "id" },
+      }
+      local _, err = helpers.cassandra.add_partition(dao, consumers)
+      if err then
+        return err
+      end
+    end,
+    down = nil,
+  },
+  {
+    name = "2018-03-16-160000_index_consumers",
+    up = [[
+      CREATE INDEX IF NOT EXISTS ON consumers(custom_id);
+      CREATE INDEX IF NOT EXISTS ON consumers(username);
+    ]]
   }
 }

kong/dao/migrations/postgres.lua

@@ -327,7 +327,12 @@ return {
   {
     name = "2016-01-25-103600_unique_custom_id",
     up = [[
-      ALTER TABLE consumers ADD CONSTRAINT consumers_custom_id_key UNIQUE(custom_id);
+      DO $$
+      BEGIN
+        ALTER TABLE consumers ADD CONSTRAINT consumers_custom_id_key UNIQUE(custom_id);
+      EXCEPTION WHEN duplicate_table THEN
+        -- Do nothing, accept existing state
+      END$$;
     ]],
     down = [[
       ALTER TABLE consumers DROP CONSTRAINT consumers_custom_id_key;

kong/dao/schemas/plugins.lua

@@ -1,6 +1,5 @@
 local utils = require "kong.tools.utils"
 local Errors = require "kong.dao.errors"
-local db_errors = require "kong.db.errors"
 local singletons = require "kong.singletons"
 local constants = require "kong.constants"
 
@@ -61,7 +60,7 @@ return {
     },
     consumer_id = {
       type = "id",
-      foreign = "consumers:id"
+      -- foreign = "consumers:id" -- manually tested in self_check
     },
     name = {
       type = "string",
@@ -84,39 +83,33 @@ return {
                                   "and one of route_id or service_id")
     end
 
-    if plugin_t.service_id ~= nil then
-      local service, err, err_t = dao.db.new_db.services:select({
-        id = plugin_t.service_id
-      })
-      if err then
-        if err_t.code == db_errors.codes.DATABASE_ERROR then
-          return false, Errors.db(err)
-        end
 
-        return false, Errors.schema(err_t)
-      end
+    local new_db = dao.db.new_db
 
-      if not service then
-        return false, Errors.foreign("no such Service (id=" ..
-                                     plugin_t.service_id .. ")")
+    local service_id = plugin_t.service_id
+    if service_id ~= nil then
+      local ok, err = new_db.services:check_foreign_key({ id = service_id },
+                                                        "Service")
+      if not ok then
+        return false, err
       end
     end
 
-    if plugin_t.route_id ~= nil then
-      local route, err, err_t = dao.db.new_db.routes:select({
-        id = plugin_t.route_id
-      })
-      if err then
-        if err_t.code == db_errors.codes.DATABASE_ERROR then
-          return false, Errors.db(err)
-        end
-
-        return false, Errors.schema(err_t)
+    local route_id = plugin_t.route_id
+    if route_id ~= nil then
+      local ok, err = new_db.routes:check_foreign_key({ id = route_id },
+                                                      "Route")
+      if not ok then
+        return false, err
       end
+    end
 
-      if not route then
-        return false, Errors.foreign("no such Route (id=" ..
-                                     plugin_t.route_id .. ")")
+    local consumer_id = plugin_t.consumer_id
+    if consumer_id ~= nil then
+      local ok, err = new_db.consumers:check_foreign_key({ id = consumer_id },
+                                                         "Consumer")
+      if not ok then
+        return false, err
       end
     end
 

kong/db/dao/consumers.lua

@@ -0,0 +1,57 @@
+local _Consumers = {}
+
+local function delete_cascade(self, table_name, fk)
+  local old_dao = self.db.old_dao
+  local rows, err = old_dao[table_name]:find_all(fk)
+  if err then
+    ngx.log(ngx.ERR, "could not gather associated entities for delete cascade: ", err)
+    return
+  end
+
+  for _, row in pairs(rows) do
+    local row_pk, _, _, err  = old_dao[table_name].model_mt(row):extract_keys()
+    if err then
+      ngx.log(ngx.ERR, "could not extract pk while delete-cascading entity: ", err)
+
+    else
+      local _, err = old_dao[table_name]:delete(row_pk)
+      if err then
+        ngx.log(ngx.ERR, "could not delete-cascade entity: ", err)
+      end
+    end
+  end
+end
+
+
+local function delete_cascade_all(self, consumer_id)
+  local fk = { consumer_id = consumer_id }
+  delete_cascade(self, "plugins", fk)
+  delete_cascade(self, "jwt_secrets", fk)
+  delete_cascade(self, "basicauth_credentials", fk)
+  delete_cascade(self, "oauth2_credentials", fk)
+  delete_cascade(self, "hmacauth_credentials", fk)
+  delete_cascade(self, "acls", fk)
+  delete_cascade(self, "keyauth_credentials", fk)
+end
+
+
+function _Consumers:delete(primary_key)
+  delete_cascade_all(self, primary_key.id)
+  return self.super.delete(self, primary_key)
+end
+
+
+function _Consumers:delete_by_username(username)
+  local entity, err, err_t = self:select_by_username(username)
+  if err then
+    return nil, err, err_t
+  end
+  if not entity then
+    return true
+  end
+  delete_cascade_all(self, entity.id)
+  return self.super.delete_by_username(self, username)
+end
+
+
+return _Consumers