feat(core) upstreams (#1735)

* adds loadbalancing on specified targets
* adds service registry
* implements #157 
* adds entities: upstreams and targets
* modifies timestamps to millisecond precision (except for the non-related tables when using postgres)
* adds collecting health-data on a per-request basis (unused for now)

config.ld

@@ -11,5 +11,5 @@ dir='doc'
 --readme='readme.md'
 sort=true
 sort_modules=true
-not_luadoc=true
+--not_luadoc=true
 all=false

kong-0.9.7-0.rockspec

@@ -75,6 +75,7 @@ build = {
     ["kong.api.routes.plugins"] = "kong/api/routes/plugins.lua",
     ["kong.api.routes.cache"] = "kong/api/routes/cache.lua",
     ["kong.api.routes.cluster"] = "kong/api/routes/cluster.lua",
+    ["kong.api.routes.upstreams"] = "kong/api/routes/upstreams.lua",
 
     ["kong.tools.dns"] = "kong/tools/dns.lua",
     ["kong.tools.utils"] = "kong/tools/utils.lua",
@@ -101,6 +102,8 @@ build = {
     ["kong.dao.schemas.nodes"] = "kong/dao/schemas/nodes.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",
     ["kong.dao.db"] = "kong/dao/db/init.lua",
     ["kong.dao.db.cassandra"] = "kong/dao/db/cassandra.lua",
     ["kong.dao.db.postgres"] = "kong/dao/db/postgres.lua",

kong/api/crud_helpers.lua

@@ -41,6 +41,24 @@ function _M.find_consumer_by_username_or_id(self, dao_factory, helpers)
   end
 end
 
+function _M.find_upstream_by_name_or_id(self, dao_factory, helpers)
+  local filter_keys = {
+    [utils.is_valid_uuid(self.params.name_or_id) and "id" or "name"] = self.params.name_or_id
+  }
+  self.params.name_or_id = nil
+
+  local rows, err = dao_factory.upstreams:find_all(filter_keys)
+  if err then
+    return helpers.yield_error(err)
+  end
+
+  -- We know name and id are unique, so if we have a row, it must be the only one
+  self.upstream = rows[1]
+  if not self.upstream then
+    return helpers.responses.send_HTTP_NOT_FOUND()
+  end
+end
+
 function _M.paginated_set(self, dao_collection)
   local size = self.params.size and tonumber(self.params.size) or 100
   local offset = self.params.offset and ngx.decode_base64(self.params.offset) or nil

kong/api/init.lua

@@ -110,7 +110,7 @@ end
 ngx.log(ngx.DEBUG, "Loading Admin API endpoints")
 
 -- Load core routes
-for _, v in ipairs({"kong", "apis", "consumers", "plugins", "cache", "cluster" }) do
+for _, v in ipairs({"kong", "apis", "consumers", "plugins", "cache", "cluster", "upstreams" }) do
   local routes = require("kong.api.routes."..v)
   attach_routes(insert_405(routes))
 end

kong/api/routes/upstreams.lua

@@ -0,0 +1,110 @@
+local crud = require "kong.api.crud_helpers"
+
+return {
+  ["/upstreams/"] = {
+    GET = function(self, dao_factory)
+      crud.paginated_set(self, dao_factory.upstreams)
+    end,
+
+    PUT = function(self, dao_factory)
+      crud.put(self.params, dao_factory.upstreams)
+    end,
+
+    POST = function(self, dao_factory, helpers)
+      crud.post(self.params, dao_factory.upstreams)
+    end
+  },
+
+  ["/upstreams/:name_or_id"] = {
+    before = function(self, dao_factory, helpers)
+      crud.find_upstream_by_name_or_id(self, dao_factory, helpers)
+    end,
+
+    GET = function(self, dao_factory, helpers)
+      return helpers.responses.send_HTTP_OK(self.upstream)
+    end,
+
+    PATCH = function(self, dao_factory)
+      crud.patch(self.params, dao_factory.upstreams, self.upstream)
+    end,
+
+    DELETE = function(self, dao_factory)
+      crud.delete(self.upstream, dao_factory.upstreams)
+    end
+  },
+
+  ["/upstreams/:name_or_id/targets/"] = {
+    before = function(self, dao_factory, helpers)
+      crud.find_upstream_by_name_or_id(self, dao_factory, helpers)
+      self.params.upstream_id = self.upstream.id
+    end,
+
+    GET = function(self, dao_factory)
+      crud.paginated_set(self, dao_factory.targets)
+    end,
+
+    POST = function(self, dao_factory, helpers)
+      -- when to cleanup: invalid-entries > (valid-ones * cleanup_factor)
+      local cleanup_factor = 10
+
+      --cleaning up history, check if it's necessary...
+      local target_history = dao_factory.targets:find_all(
+            { upstream_id = self.params.upstream_id })
+
+      if target_history then --ignoring errors here, will be caught when posting below
+        -- sort the targets
+        for _,target in ipairs(target_history) do
+          target.order = target.created_at..":"..target.id
+        end
+
+        -- sort table in reverse order
+        table.sort(target_history, function(a,b) return a.order>b.order end)
+        -- do clean up
+        local cleaned = {} 
+        local delete = {}
+
+        for _, entry in ipairs(target_history) do
+          if cleaned[entry.target] then
+            -- we got a newer entry for this target than this, so this one can go
+            delete[#delete+1] = entry
+
+          else
+            -- haven't got this one, so this is the last one for this target
+            cleaned[entry.target] = true
+            cleaned[#cleaned+1] = entry
+            if entry.weight == 0 then
+              delete[#delete+1] = entry
+            end
+          end
+        end
+
+        -- do we need to cleanup?
+        -- either nothing left, or when 10x more outdated than active entries
+        if (#cleaned == 0 and #delete > 0) or
+           (#delete >= (math.max(#cleaned,1)*cleanup_factor)) then
+
+          ngx.log(ngx.INFO, "[admin api] Starting cleanup of target table for upstream ", 
+                            tostring(self.params.upstream_id))
+          local cnt = 0
+          for _, entry in ipairs(delete) do
+            -- not sending update events, one event at the end, based on the
+            -- post of the new entry should suffice to reload only once
+            dao_factory.targets:delete(
+              { id = entry.id }, 
+              { quiet = true }
+            )
+            -- ignoring errors here, deleted by id, so should not matter
+            -- in case another kong-node does the same cleanup simultaneously
+            cnt = cnt + 1 
+          end
+
+          ngx.log(ngx.INFO, "[admin api] Finished cleanup of target table",
+            " for upstream ", tostring(self.params.upstream_id),
+            " removed ", tostring(cnt), " target entries")
+        end
+      end
+
+      crud.post(self.params, dao_factory.targets)
+    end,
+  },
+}