Merge branch 'hotfix/cassandra-auth' into next

kong-0.6.0-1.rockspec

@@ -19,7 +19,7 @@ dependencies = {
   "yaml ~> 1.1.2-1",
   "lapis ~> 1.3.1-1",
   "stringy ~> 0.4-1",
-  "lua-cassandra ~> 0.4.2-0",
+  "lua-cassandra ~> 0.5.0",
   "multipart ~> 0.2-1",
   "lua-path ~> 0.2.3-1",
   "lua-cjson ~> 2.1.0-1",

kong.yml

@@ -103,6 +103,10 @@
   ## Name of the keyspace used by Kong. Will be created if it does not exist.
   # keyspace: kong
 
+  #####
+  ## Connection and reading timeout (in ms).
+  # timeout: 5000
+
   ######
   ## Keyspace options. Set those before running Kong or any migration.
   ## Those settings will be used to create a keyspace with the desired options

kong/dao/cassandra/factory.lua

@@ -11,12 +11,6 @@ local stringy = require "stringy"
 local Object = require "classic"
 local utils = require "kong.tools.utils"
 
-if ngx ~= nil and type(ngx.get_phase) == "function" and ngx.get_phase() == "init" and not ngx.stub then
-  cassandra.set_log_level("INFO")
-else
-  cassandra.set_log_level("QUIET")
-end
-
 local CassandraFactory = Object:extend()
 
 -- Shorthand for accessing one of the underlying DAOs
@@ -36,6 +30,10 @@ function CassandraFactory:new(properties, plugins, spawn_cluster, events_handler
   self.type = "cassandra"
   self.daos = {}
 
+  if properties.username and properties.password then
+    self.properties.auth = cassandra.auth.PlainTextProvider(properties.username, properties.password)
+  end
+
   if spawn_cluster then
     local ok, err = cassandra.spawn_cluster(self:get_session_options())
     if not ok then
@@ -119,13 +117,16 @@ function CassandraFactory:get_session_options()
     query_options = {
       prepare = true
     },
-    username = self.properties.username,
-    password = self.properties.password,
+    socket_options = {
+      connect_timeout = self.properties.timeout,
+      read_timeout = self.properties.timeout
+    },
     ssl_options = {
       enabled = self.properties.ssl.enabled,
       verify = self.properties.ssl.verify,
       ca = self.properties.ssl.certificate_authority
-    }
+    },
+    auth = self.properties.auth
   }
 end
 

kong/dao/cassandra/migrations.lua

@@ -27,8 +27,12 @@ function Migrations:new(properties, events_handler)
   Migrations.super.new(self, properties, events_handler)
 end
 
+function Migrations:execute(query, args, keyspace)
+  return Migrations.super.execute(self, query, args, {prepare = false}, keyspace)
+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 = self:execute(self.queries.get_keyspace, {self.properties.keyspace}, "system")
   if err then
     return nil, err
   else
@@ -41,7 +45,7 @@ end
 -- @return query result
 -- @return error if any
 function Migrations:add_migration(migration_name, identifier)
-  return Migrations.super.execute(self, self.queries.add_migration, {cassandra.list({migration_name}), identifier})
+  return self:execute(self.queries.add_migration, {cassandra.list({migration_name}), identifier})
 end
 
 -- Return all logged migrations with a filter by identifier optionally. Check if keyspace exists before to avoid error during the first migration.
@@ -59,9 +63,9 @@ function Migrations:get_migrations(identifier)
 
   local rows, err
   if identifier ~= nil then
-    rows, err = Migrations.super.execute(self, self.queries.get_migrations, {identifier})
+    rows, err = self:execute(self.queries.get_migrations, {identifier})
   else
-    rows, err = Migrations.super.execute(self, self.queries.get_all_migrations)
+    rows, err = self:execute(self.queries.get_all_migrations)
   end
 
   if err and stringy.find(err.message, "unconfigured columnfamily schema_migrations") ~= nil then
@@ -77,14 +81,14 @@ end
 -- @return query result
 -- @return error if any
 function Migrations:delete_migration(migration_name, identifier)
-  return Migrations.super.execute(self, self.queries.delete_migration, {cassandra.list({migration_name}), identifier})
+  return self:execute(self.queries.delete_migration, {cassandra.list({migration_name}), identifier})
 end
 
 -- Drop the entire keyspace
 -- @param `keyspace` Name of the keyspace to drop
 -- @return query result
 function Migrations:drop_keyspace(keyspace)
-  return Migrations.super.execute(self, string.format("DROP keyspace \"%s\"", keyspace))
+  return self:execute(string.format("DROP keyspace \"%s\"", keyspace))
 end
 
 function Migrations:drop()

kong/tools/config_defaults.lua

@@ -39,6 +39,7 @@ return {
     content = {
       ["contact_points"] = {type = "array", default = {"127.0.0.1:9042"}},
       ["keyspace"] = {type = "string", default = "kong"},
+      ["timeout"] = {type = "number", default = 5000},
       ["replication_strategy"] = {type = "string", default = "SimpleStrategy", enum = {"SimpleStrategy", "NetworkTopologyStrategy"}},
       ["replication_factor"] = {type = "number", default = 1},
       ["data_centers"] = {type = "table", default = {}},

spec/unit/dao/cassandra/factory_spec.lua

@@ -1,5 +1,6 @@
 local Factory = require "kong.dao.cassandra.factory"
 local utils = require "kong.tools.utils"
+local cassandra = require "cassandra"
 local spec_helpers = require "spec.spec_helpers"
 local env = spec_helpers.get_env()
 local default_dao_properties = env.configuration.cassandra
@@ -24,6 +25,10 @@ describe("Cassadra factory", function()
         query_options = {
           prepare = true
         },
+        socket_options = {
+          connect_timeout = 5000,
+          read_timeout = 5000
+        },
         ssl_options = {
           enabled = false,
           verify = false
@@ -46,6 +51,10 @@ describe("Cassadra factory", function()
         query_options = {
           prepare = true
         },
+        socket_options = {
+          connect_timeout = 5000,
+          read_timeout = 5000
+        },
         ssl_options = {
           enabled = false,
           verify = false
@@ -68,12 +77,15 @@ describe("Cassadra factory", function()
         query_options = {
           prepare = true
         },
+        socket_options = {
+          connect_timeout = 5000,
+          read_timeout = 5000
+        },
         ssl_options = {
           enabled = false,
           verify = false
         },
-        username = "cassie",
-        password = "cassiepwd"
+        auth = cassandra.auth.PlainTextProvider("cassie", "cassiepwd")
       }, options)
     end)
     it("should accept SSL properties", function()
@@ -93,6 +105,10 @@ describe("Cassadra factory", function()
         query_options = {
           prepare = true
         },
+        socket_options = {
+          connect_timeout = 5000,
+          read_timeout = 5000
+        },
         ssl_options = {
           enabled = false,
           verify = true
@@ -113,6 +129,10 @@ describe("Cassadra factory", function()
         query_options = {
           prepare = true
         },
+        socket_options = {
+          connect_timeout = 5000,
+          read_timeout = 5000
+        },
         ssl_options = {
           enabled = true,
           verify = true