feat(db) Support for Cassandra 3.x - lua-cassandra 1.x

.ci/setup_env.sh

@@ -90,6 +90,8 @@ export PATH=$PATH:$OPENRESTY_INSTALL/nginx/sbin:$OPENRESTY_INSTALL/bin:$LUAROCKS
 
 eval `luarocks path`
 
+luarocks purge --tree=$LUAROCKS_INSTALL
+
 # -------------------------------------
 # Install ccm & setup Cassandra cluster
 # -------------------------------------

.travis.yml

@@ -1,9 +1,9 @@
 sudo: false
 
-language: c
+language: java
 
-compiler:
-  - gcc
+jdk:
+  - oraclejdk8
 
 notifications:
   email: false
@@ -22,7 +22,7 @@ env:
     - SERF=0.7.0
     - LUAROCKS=2.4.0
     - OPENSSL=1.0.2h
-    - CASSANDRA=2.2.7
+    - CASSANDRA=2.2.8
     - OPENRESTY_BASE=1.9.15.1
     - OPENRESTY_LATEST=1.11.2.1
     - OPENRESTY=$OPENRESTY_BASE
@@ -37,8 +37,10 @@ env:
       OPENRESTY=$OPENRESTY_BASE
     - TEST_SUITE=integration
       OPENRESTY=$OPENRESTY_LATEST
+      CASSANDRA=3.9
     - TEST_SUITE=plugins
       OPENRESTY=$OPENRESTY_LATEST
+      CASSANDRA=3.9
 
 before_install:
   - source .ci/setup_env.sh

bin/busted

@@ -5,5 +5,7 @@ require("kong.core.globalpatches")({
   rbusted = true
 })
 
+package.path = "?/init.lua;"..package.path
+
 -- Busted command-line runner
 require 'busted.runner'({ standalone = false })

bin/kong

@@ -4,4 +4,6 @@ require("kong.core.globalpatches")({
   cli = true
 })
 
+package.path = "?/init.lua;"..package.path
+
 require("kong.cmd.init")(arg)

kong-0.9.3-0.rockspec

@@ -20,7 +20,7 @@ dependencies = {
   "multipart == 0.4",
   "version == 0.2",
   "lapis == 1.5.1",
-  "lua-cassandra == 0.5.4",
+  "lua-cassandra == dev-0",
   "pgmoon-mashape == 2.0.1",
   "luatz == 0.3",
   "lua_system_constants == 0.1.1",
@@ -45,8 +45,8 @@ build = {
 
     ["kong.vendor.classic"] = "kong/vendor/classic.lua",
 
+    ["kong.cmd"] = "kong/cmd/init.lua",
     ["kong.cmd.roar"] = "kong/cmd/roar.lua",
-    ["kong.cmd.init"] = "kong/cmd/init.lua",
     ["kong.cmd.stop"] = "kong/cmd/stop.lua",
     ["kong.cmd.quit"] = "kong/cmd/quit.lua",
     ["kong.cmd.start"] = "kong/cmd/start.lua",
@@ -65,7 +65,7 @@ build = {
     ["kong.cmd.utils.prefix_handler"] = "kong/cmd/utils/prefix_handler.lua",
     ["kong.cmd.utils.dnsmasq_signals"] = "kong/cmd/utils/dnsmasq_signals.lua",
 
-    ["kong.api.init"] = "kong/api/init.lua",
+    ["kong.api"] = "kong/api/init.lua",
     ["kong.api.api_helpers"] = "kong/api/api_helpers.lua",
     ["kong.api.crud_helpers"] = "kong/api/crud_helpers.lua",
     ["kong.api.routes.kong"] = "kong/api/routes/kong.lua",
@@ -98,9 +98,9 @@ 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.base_db"] = "kong/dao/base_db.lua",
-    ["kong.dao.cassandra_db"] = "kong/dao/cassandra_db.lua",
-    ["kong.dao.postgres_db"] = "kong/dao/postgres_db.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.dao.dao"] = "kong/dao/dao.lua",
     ["kong.dao.factory"] = "kong/dao/factory.lua",
     ["kong.dao.model_factory"] = "kong/dao/model_factory.lua",
@@ -166,9 +166,9 @@ build = {
     ["kong.plugins.rate-limiting.migrations.postgres"] = "kong/plugins/rate-limiting/migrations/postgres.lua",
     ["kong.plugins.rate-limiting.handler"] = "kong/plugins/rate-limiting/handler.lua",
     ["kong.plugins.rate-limiting.schema"] = "kong/plugins/rate-limiting/schema.lua",
-    ["kong.plugins.rate-limiting.policies"] = "kong/plugins/rate-limiting/policies.lua",
-    ["kong.plugins.rate-limiting.dao.cassandra"] = "kong/plugins/rate-limiting/dao/cassandra.lua",
-    ["kong.plugins.rate-limiting.dao.postgres"] = "kong/plugins/rate-limiting/dao/postgres.lua",
+    ["kong.plugins.rate-limiting.daos"] = "kong/plugins/rate-limiting/daos.lua",
+    ["kong.plugins.rate-limiting.policies"] = "kong/plugins/rate-limiting/policies/init.lua",
+    ["kong.plugins.rate-limiting.policies.cluster"] = "kong/plugins/rate-limiting/policies/cluster.lua",
 
     ["kong.plugins.response-ratelimiting.migrations.cassandra"] = "kong/plugins/response-ratelimiting/migrations/cassandra.lua",
     ["kong.plugins.response-ratelimiting.migrations.postgres"] = "kong/plugins/response-ratelimiting/migrations/postgres.lua",
@@ -177,9 +177,9 @@ build = {
     ["kong.plugins.response-ratelimiting.header_filter"] = "kong/plugins/response-ratelimiting/header_filter.lua",
     ["kong.plugins.response-ratelimiting.log"] = "kong/plugins/response-ratelimiting/log.lua",
     ["kong.plugins.response-ratelimiting.schema"] = "kong/plugins/response-ratelimiting/schema.lua",
-    ["kong.plugins.response-ratelimiting.policies"] = "kong/plugins/response-ratelimiting/policies.lua",
-    ["kong.plugins.response-ratelimiting.dao.cassandra"] = "kong/plugins/response-ratelimiting/dao/cassandra.lua",
-    ["kong.plugins.response-ratelimiting.dao.postgres"] = "kong/plugins/response-ratelimiting/dao/postgres.lua",
+    ["kong.plugins.response-ratelimiting.daos"] = "kong/plugins/response-ratelimiting/daos.lua",
+    ["kong.plugins.response-ratelimiting.policies"] = "kong/plugins/response-ratelimiting/policies/init.lua",
+    ["kong.plugins.response-ratelimiting.policies.cluster"] = "kong/plugins/response-ratelimiting/policies/cluster.lua",
 
     ["kong.plugins.request-size-limiting.handler"] = "kong/plugins/request-size-limiting/handler.lua",
     ["kong.plugins.request-size-limiting.schema"] = "kong/plugins/request-size-limiting/schema.lua",

kong.conf.default

@@ -129,9 +129,6 @@
 
 #cassandra_keyspace = kong       # The keyspace to use in your cluster.
 
-#cassandra_consistency = ONE     # Consistency setting to use when reading/
-                                 # writing to the Cassandra cluster.
-
 #cassandra_timeout = 5000        # Defines the timeout (in ms), for reading
                                  # and writing.
 
@@ -149,6 +146,22 @@
 #cassandra_password = kong       # Password when using the
                                  # `PasswordAuthenticator` scheme.
 
+#cassandra_consistency = ONE     # Consistency setting to use when reading/
+                                 # writing to the Cassandra cluster.
+
+#cassandra_lb_policy = RoundRobin  # Load balancing policy to use when
+                                   # distributing queries across your Cassandra
+                                   # cluster.
+                                   # Accepted values are `RoundRobin` and
+                                   # `DCAwareRoundRobin`.
+                                   # Prefer the later if and only if you are
+                                   # using a multi-datacenter cluster.
+
+#cassandra_local_datacenter =    # When using the `DCAwareRoundRobin` load
+                                 # balancing policy, you must specify the name
+                                 # of the local (closest) datacenter for this
+                                 # Kong node.
+
 #cassandra_repl_strategy = SimpleStrategy  # When migrating for the first time,
                                            # Kong will use this setting to
                                            # create your keyspace.

kong/cmd/cluster.lua

@@ -7,7 +7,7 @@ local conf_loader = require "kong.conf_loader"
 local function execute(args)
   if args.command == "keygen" then
     local conf = assert(conf_loader(args.conf))
-    local dao = DAOFactory(conf)
+    local dao = assert(DAOFactory.new(conf))
     local serf = Serf.new(conf, dao)
     print(assert(serf:keygen()))
     return
@@ -21,7 +21,7 @@ local function execute(args)
   assert(pl_path.exists(default_conf.prefix),
          "no such prefix: "..default_conf.prefix)
   local conf = assert(conf_loader(default_conf.kong_conf))
-  local dao = DAOFactory(conf)
+  local dao = assert(DAOFactory.new(conf))
   local serf = Serf.new(conf, dao)
 
   if args.command == "members" then

kong/cmd/migrations.lua

@@ -28,7 +28,7 @@ end
 
 local function execute(args)
   local conf = assert(conf_loader(args.conf))
-  local dao = DAOFactory(conf, conf.plugins)
+  local dao = assert(DAOFactory.new(conf, conf.plugins))
 
   if args.command == "up" then
     assert(dao:run_migrations())

kong/cmd/quit.lua

@@ -37,7 +37,8 @@ local function execute(args)
     assert(nginx_signals.stop(conf))
   end
 
-  assert(serf_signals.stop(conf, DAOFactory(conf)))
+  local dao = assert(DAOFactory.new(conf))
+  assert(serf_signals.stop(conf, dao))
 
   if conf.dnsmasq then
     assert(dnsmasq_signals.stop(conf))

kong/cmd/reload.lua

@@ -25,7 +25,9 @@ local function execute(args)
   if conf.dnsmasq then
     assert(dnsmasq_signals.start(conf))
   end
-  assert(serf_signals.start(conf, DAOFactory(conf)))
+
+  local dao = assert(DAOFactory.new(conf))
+  assert(serf_signals.start(conf, dao))
   assert(nginx_signals.reload(conf))
   log("Kong reloaded")
 end

kong/cmd/start.lua

@@ -15,8 +15,8 @@ local function execute(args)
   assert(not kill.is_running(conf.nginx_pid),
          "Kong is already running in "..conf.prefix)
 
-  local dao = DAOFactory(conf)
   local err
+  local dao = assert(DAOFactory.new(conf))
   xpcall(function()
     assert(prefix_handler.prepare_prefix(conf, args.nginx_conf))
     assert(dao:run_migrations())

kong/cmd/stop.lua

@@ -18,8 +18,9 @@ local function execute(args)
 
   -- load <PREFIX>/kong.conf containing running node's config
   local conf = assert(conf_loader(default_conf.kong_conf))
+  local dao = assert(DAOFactory.new(conf))
   assert(nginx_signals.stop(conf))
-  assert(serf_signals.stop(conf, DAOFactory(conf)))
+  assert(serf_signals.stop(conf, dao))
   if conf.dnsmasq then
     assert(dnsmasq_signals.stop(conf))
   end

kong/conf_loader.lua

@@ -63,14 +63,16 @@ local CONF_INFERENCES = {
 
   cassandra_contact_points = {typ = "array"},
   cassandra_port = {typ = "number"},
-  cassandra_repl_strategy = {enum = {"SimpleStrategy", "NetworkTopologyStrategy"}},
-  cassandra_repl_factor = {typ = "number"},
-  cassandra_data_centers = {typ = "array"},
-  cassandra_consistency = {enum = {"ALL", "EACH_QUORUM", "QUORUM", "LOCAL_QUORUM", "ONE",
-                                   "TWO", "THREE", "LOCAL_ONE"}}, -- no ANY: this is R/W
   cassandra_timeout = {typ = "number"},
   cassandra_ssl = {typ = "boolean"},
   cassandra_ssl_verify = {typ = "boolean"},
+  cassandra_consistency = {enum = {"ALL", "EACH_QUORUM", "QUORUM", "LOCAL_QUORUM", "ONE",
+                                   "TWO", "THREE", "LOCAL_ONE"}}, -- no ANY: this is R/W
+  cassandra_lb_policy = {enum = {"RoundRobin", "DCAwareRoundRobin"}},
+  cassandra_local_datacenter = {typ = "string"},
+  cassandra_repl_strategy = {enum = {"SimpleStrategy", "NetworkTopologyStrategy"}},
+  cassandra_repl_factor = {typ = "number"},
+  cassandra_data_centers = {typ = "array"},
 
   cluster_profile = {enum = {"local", "lan", "wan"}},
   cluster_ttl_on_failure = {typ = "number"},
@@ -162,6 +164,12 @@ local function check_and_infer(conf)
   -- custom validations
   ---------------------
 
+  if conf.cassandra_lb_policy == "DCAwareRoundRobin" and
+     not conf.cassandra_local_datacenter then
+     errors[#errors+1] = "must specify 'cassandra_local_datacenter' when "..
+                        "DCAwareRoundRobin policy is in use"
+  end
+
   if conf.ssl then
     if conf.ssl_cert and not conf.ssl_cert_key then
       errors[#errors+1] = "ssl_cert_key must be specified"

kong/core/globalpatches.lua

@@ -13,6 +13,113 @@ return function(options)
       local socket = require(namespace .. ".socket")
       socket.force_luasocket(ngx.get_phase(), true)
     end
+
+    do
+      -- ngx.shared.DICT proxy
+      -- https://github.com/bsm/fakengx/blob/master/fakengx.lua
+      -- with minor fixes and addtions such as exptime
+      --
+      -- See https://github.com/openresty/resty-cli/pull/12
+      -- for a definitive solution ot using shms in CLI
+      local SharedDict = {}
+      local function set(data, key, value)
+        data[key] = {
+          value = value,
+          info = {expired = false}
+        }
+      end
+      function SharedDict:new()
+        return setmetatable({data = {}}, {__index = self})
+      end
+      function SharedDict:get(key)
+        return self.data[key] and self.data[key].value, nil
+      end
+      function SharedDict:set(key, value)
+        set(self.data, key, value)
+        return true, nil, false
+      end
+      SharedDict.safe_set = SharedDict.set
+      function SharedDict:add(key, value, exptime)
+        if self.data[key] ~= nil then
+          return false, "exists", false
+        end
+
+        if exptime then
+          ngx.timer.at(exptime, function()
+            self.data[key] = nil
+          end)
+        end
+
+        set(self.data, key, value)
+        return true, nil, false
+      end
+      function SharedDict:replace(key, value)
+        if self.data[key] == nil then
+          return false, "not found", false
+        end
+        set(self.data, key, value)
+        return true, nil, false
+      end
+      function SharedDict:delete(key)
+        self.data[key] = nil
+        return true
+      end
+      function SharedDict:incr(key, value)
+        if not self.data[key] then
+          return nil, "not found"
+        elseif type(self.data[key].value) ~= "number" then
+          return nil, "not a number"
+        end
+        self.data[key].value = self.data[key].value + value
+        return self.data[key].value, nil
+      end
+      function SharedDict:flush_all()
+        for _, item in pairs(self.data) do
+          item.info.expired = true
+        end
+      end
+      function SharedDict:flush_expired(n)
+        local data = self.data
+        local flushed = 0
+
+        for key, item in pairs(self.data) do
+          if item.info.expired then
+            data[key] = nil
+            flushed = flushed + 1
+            if n and flushed == n then
+              break
+            end
+          end
+        end
+        self.data = data
+        return flushed
+      end
+      function SharedDict:get_keys(n)
+        n = n or 1024
+        local i = 0
+        local keys = {}
+        for k in pairs(self.data) do
+          keys[#keys+1] = k
+          i = i + 1
+          if n ~= 0 and i == n then
+            break
+          end
+        end
+        return keys
+      end
+
+      -- hack
+      _G.ngx.shared = setmetatable({}, {
+        __index = function(self, key)
+          local shm = rawget(self, key)
+          if not shm then
+            shm = SharedDict:new()
+            rawset(self, key, SharedDict:new())
+          end
+          return shm
+        end
+      })
+    end
   end
 
   if options.rbusted then
@@ -93,6 +200,5 @@ return function(options)
         return seed
       end
     end
-
   end
 end