feat(config) add options for keyspace replication strategy

Possibility to configure the replication strategy used by the created
keyspace and its options.

Implements Kong#543 and Kong#350

kong.yml

@@ -1,3 +1,4 @@
+<<<<<<< HEAD
 ######
 ## Kong configuration file. All commented values are default values.
 ## Uncomment and update a value to configure Kong to your needs.
@@ -73,6 +74,29 @@
       ## for being eventually re-used before being closed.
       # keepalive: 60000
 
+      ######
+      ## Keyspace options. Set those before running Kong or any migration.
+      ## Those settings will be used to create a keyspace with the desired options
+      ## when first running the migrations.
+      ## See http://docs.datastax.com/en/cql/3.1/cql/cql_reference/create_keyspace_r.html
+
+      ######
+      ## The name of the replica placement strategy class for the new keyspace.
+      ## Can be "SimpleStrategy" or "NetworkTopologyStrategy".
+      # replication_strategy: SimpleStrategy
+
+      ######
+      ## For SimpleStrategy only.
+      ## The number of replicas of data on multiple nodes.
+      # replication_factor: 1
+
+      ######
+      ## For NetworkTopologyStrategy only.
+      ## The number of replicas of data on multiple nodes in each data center.
+      # data_centers:
+      #   dc1: 2
+      #   dc2: 3
+
       ######
       ## If true, will enable client-to-node encryption.
       # ssl: false

kong/dao/cassandra/schema/migrations.lua

@@ -1,14 +1,57 @@
+local DEFAULTS = {
+  ["SimpleStrategy"] = {
+    replication_factor = 1
+  },
+  ["NetworkTopologyStrategy"] = {
+    data_centers = {}
+  }
+}
+
 local Migrations = {
-  -- skeleton
   {
     init = true,
     name = "2015-01-12-175310_skeleton",
     up = function(options)
-      return [[
-        CREATE KEYSPACE IF NOT EXISTS "]]..options.keyspace..[["
-          WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1};
+      if not options.replication_strategy then options.replication_strategy = "SimpleStrategy" end
+      local keyspace_name = options.keyspace
+      local strategy, strategy_properties = "", ""
+
+      -- Find strategy and retrieve default options
+      for strategy_name, strategy_defaults in pairs(DEFAULTS) do
+        if options.replication_strategy == strategy_name then
+          strategy = strategy_name
+          for opt_name, opt_value in pairs(strategy_defaults) do
+            if not options[opt_name] then
+              options[opt_name] = opt_value
+            end
+          end
+        end
+      end
+
+      -- Format strategy options
+      if strategy == "SimpleStrategy" then
+        strategy_properties = string.format(", 'replication_factor': %s", options.replication_factor)
+      elseif strategy == "NetworkTopologyStrategy" then
+        local dcs = {}
+        for dc_name, dc_repl in pairs(options.data_centers) do
+          table.insert(dcs, string.format("'%s': %s", dc_name, dc_repl))
+        end
+        if #dcs > 0 then
+          strategy_properties = string.format(", %s", table.concat(dcs, ", "))
+        end
+      else
+        -- Strategy unknown
+        return nil, "invalid replication_strategy class"
+      end
+
+      -- Format final keyspace creation query
+      local keyspace_str = string.format([[
+        CREATE KEYSPACE IF NOT EXISTS "%s"
+          WITH REPLICATION = {'class': '%s'%s};
+      ]], keyspace_name, strategy, strategy_properties)
 
-        USE "]]..options.keyspace..[[";
+      return keyspace_str..[[
+        USE "]]..keyspace_name..[[";
 
         CREATE TABLE IF NOT EXISTS schema_migrations(
           id text PRIMARY KEY,

kong/tools/migrations.lua

@@ -16,7 +16,7 @@ function Migrations:new(dao, core_migrations, plugins_namespace)
   end
 
   self.dao = dao
-  self.options = {keyspace = dao._properties.keyspace}
+  self.dao_properties = dao._properties
   self.plugins_namespace = plugins_namespace and plugins_namespace or "kong.plugins"
 end
 
@@ -92,11 +92,14 @@ function Migrations:run_migrations(migrations, identifier, callback)
     diff_migrations = migrations
   end
 
-  local err
+  local up_query, err
   -- Execute all new migrations, in order
   for _, migration in ipairs(diff_migrations) do
     -- Generate UP query from string + options parameter
-    local up_query = migration.up(self.options)
+    up_query, err = migration.up(self.dao_properties)
+    if not up_query then
+      break
+    end
     err = self.dao:execute_queries(up_query, migration.init)
     if err then
       err = "Error executing migration for "..identifier..": "..err
@@ -145,7 +148,7 @@ function Migrations:run_rollback(migrations, identifier)
   end
 
   -- Generate DOWN query from string + options
-  local down_query = migration_to_rollback.down(self.options)
+  local down_query = migration_to_rollback.down(self.dao_properties)
   local err = self.dao:execute_queries(down_query)
   if err then
     return nil, err

spec/integration/dao/cassandra/migrations_spec.lua

@@ -61,9 +61,9 @@ local CORE_MIGRATIONS_STUB = {
       ]]
     end,
     down = function()
-       return [[
-         DROP TABLE users2;
-       ]]
+      return [[
+        DROP TABLE users2;
+      ]]
     end
   }
 
@@ -99,6 +99,37 @@ say:set("assertion.has_table.positive", "Expected keyspace to have table %s")
 say:set("assertion.has_table.negative", "Expected keyspace not to have table %s")
 assert:register("assertion", "has_table", has_table, "assertion.has_table.positive", "assertion.has_table.negative")
 
+local function has_keyspace(state, arguments)
+  local rows, err = session:execute("SELECT * FROM system.schema_keyspaces WHERE keyspace_name = ?", {arguments[1]})
+  if err then
+    error(err)
+  end
+
+  return #rows > 0
+end
+
+say:set("assertion.has_keyspace.positive", "Expected keyspace %s to exist")
+say:set("assertion.has_keyspace.negative", "Expected keyspace %s to not exist")
+assert:register("assertion", "has_keyspace", has_keyspace, "assertion.has_keyspace.positive", "assertion.has_keyspace.negative")
+
+local function has_replication_options(state, arguments)
+  local rows, err = session:execute("SELECT * FROM system.schema_keyspaces WHERE keyspace_name = ?", {arguments[1]})
+  if err then
+    error(err)
+  end
+
+  if #rows > 0 then
+    local keyspace = rows[1]
+    assert.equal("org.apache.cassandra.locator."..arguments[2], keyspace.strategy_class)
+    assert.equal(arguments[3], keyspace.strategy_options)
+    return true
+  end
+end
+
+say:set("assertion.has_replication_options.positive", "Expected keyspace %s to have given replication options")
+say:set("assertion.has_replication_options.negative", "Expected keyspace %s to not have given replication options")
+assert:register("assertion", "has_replication_options", has_replication_options, "assertion.has_replication_options.positive", "assertion.has_replication_options.negative")
+
 local function has_migration(state, arguments)
   local identifier = arguments[1]
   local migration = arguments[2]
@@ -121,7 +152,6 @@ local function has_migration(state, arguments)
   return false
 end
 
-local say = require "say"
 say:set("assertion.has_migration.positive", "Expected keyspace to have migration %s record")
 say:set("assertion.has_migration.negative", "Expected keyspace not to have migration %s recorded")
 assert:register("assertion", "has_migration", has_migration, "assertion.has_migration.positive", "assertion.has_migration.negative")
@@ -232,4 +262,29 @@ describe("Migrations", function()
       assert.has_migration("fixtures", "stub_fixture_mig2")
     end)
   end)
+  describe("keyspace replication strategy", function()
+    local KEYSPACE_NAME = "kong_replication_strategy_tests"
+
+    setup(function()
+      migrations = Migrations(test_dao)
+      migrations.dao_properties.keyspace = KEYSPACE_NAME
+    end)
+
+    after_each(function()
+      session:execute("DROP KEYSPACE "..KEYSPACE_NAME)
+    end)
+
+    it("should create a keyspace with SimpleStrategy by default", function()
+      local err = migrations:migrate("core")
+      assert.falsy(err)
+      assert.has_keyspace(KEYSPACE_NAME)
+      assert.has_replication_options(KEYSPACE_NAME, "SimpleStrategy", "{\"replication_factor\":\"1\"}")
+    end)
+    it("should catch an invalid replication strategy", function()
+      migrations.dao_properties.replication_strategy = "foo"
+      local err = migrations:migrate("core")
+      assert.truthy(err)
+      assert.equal("invalid replication_strategy class", err)
+    end)
+  end)
 end)

spec/unit/dao/cassandra/migrations_spec.lua

@@ -0,0 +1,52 @@
+local stringy = require "stringy"
+local migrations = require "kong.dao.cassandra.schema.migrations"
+local first_migration = migrations[1]
+
+local function strip_query(str)
+  str = stringy.split(str, ";")[1]
+  str = str:gsub("\n", " "):gsub("%s+", " ")
+  return stringy.strip(str)
+end
+
+describe("Cassandra migrations", function()
+  describe("Keyspace options", function()
+    it("should default to SimpleStrategy class with replication_factor of 1", function()
+      local queries = first_migration.up({keyspace = "kong"})
+      local keyspace_query = strip_query(queries)
+      assert.equal("CREATE KEYSPACE IF NOT EXISTS \"kong\" WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1}", keyspace_query)
+    end)
+    it("should be possible to set a custom replication_factor", function()
+      local queries = first_migration.up({keyspace = "kong", replication_factor = 2})
+      local keyspace_query = strip_query(queries)
+      assert.equal("CREATE KEYSPACE IF NOT EXISTS \"kong\" WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 2}", keyspace_query)
+    end)
+    it("should accept NetworkTopologyStrategy", function()
+      local queries = first_migration.up({
+        keyspace = "kong",
+        replication_strategy = "NetworkTopologyStrategy"
+      })
+      local keyspace_query = strip_query(queries)
+      assert.equal("CREATE KEYSPACE IF NOT EXISTS \"kong\" WITH REPLICATION = {'class': 'NetworkTopologyStrategy'}", keyspace_query)
+    end)
+    it("should be possible to set data centers for NetworkTopologyStrategy", function()
+      local queries = first_migration.up({
+        keyspace = "kong",
+        replication_strategy = "NetworkTopologyStrategy",
+        data_centers = {
+          dc1 = 2,
+          dc2 = 3
+        }
+      })
+      local keyspace_query = strip_query(queries)
+      assert.equal("CREATE KEYSPACE IF NOT EXISTS \"kong\" WITH REPLICATION = {'class': 'NetworkTopologyStrategy', 'dc1': 2, 'dc2': 3}", keyspace_query)
+    end)
+    it("should return an error if an invalid replication_strategy is given", function()
+      local queries, err = first_migration.up({
+        keyspace = "kong",
+        replication_strategy = "foo"
+      })
+      assert.falsy(queries)
+      assert.equal("invalid replication_strategy class", err)
+    end)
+  end)
+end)