fetch schema to be applied from replicaset leader

Author: d.sharonov

migrator/init.lua

@@ -33,6 +33,22 @@ local function get_diff()
     return names, migrations_map
 end
 
+local function get_schema()
+    return ddl.get_schema()
+end
+
+-- since migrations might be triggered on a replica, we should fetch ddl schema from actual master
+-- see https://github.com/tarantool/migrations/issues/56 for details
+local function fetch_schema()
+    if vars.use_cartridge_ddl ~= true then return nil end
+    local schema, err = rpc.call('migrator', 'get_schema', nil, {prefer_local = true, leader_only = true})
+        if err ~= nil then
+        log.error(err)
+        error(err)
+    end
+    return schema
+end
+
 --- Run migrations on all nodes in the cluster
 -- Throws an exception in case of any problems
 -- @function up
@@ -97,7 +113,7 @@ local function up()
 
     local patch = {
         migrations = config,
-        ['schema.yml'] = vars.use_cartridge_ddl and ddl.get_schema() or nil
+        ['schema.yml'] = fetch_schema()
     }
     log.info('Migrations applied on all storages, changing clusterwide configuration...')
     log.verbose('All migrations applied successfully, changing cluster-wide configuration with a patch: %s', json.encode(patch))
@@ -158,5 +174,7 @@ return {
     up = up,
 
     set_loader = set_loader,
-    set_use_cartridge_ddl = set_use_cartridge_ddl
+    set_use_cartridge_ddl = set_use_cartridge_ddl,
+
+    get_schema = get_schema
 }

test/integration/fockups_test.lua

@@ -30,6 +30,7 @@ g.cluster = cartridge_helpers.Cluster:new({
             roles = { 'vshard-storage' },
             servers = {
                 { instance_uuid = cartridge_helpers.uuid('b', 1), },
+                { instance_uuid = cartridge_helpers.uuid('b', 2), },
             },
         },
     },
@@ -167,3 +168,62 @@ g.test_reload = function()
     end
 end
 
+-- https://github.com/tarantool/migrations/issues/56
+g.test_up_on_replica = function()
+    for _, server in pairs(g.cluster.servers) do
+        server.net_box:eval([[
+                require('migrator').set_loader(
+                    require('migrator.config-loader').new()
+                )
+            ]])
+    end
+
+    -- create some space
+    g.cluster.main_server:http_request('post', '/migrations/up', { json = {} })
+    utils.set_sections(g, { { filename = "migrations/source/100_create_space.lua", content = [[
+        return {
+            up = function()
+                local f = box.schema.create_space('somespace', {
+                    format = {
+                        { name = 'key', type = 'string' },
+                        { name = 'value', type = 'string', is_nullable = true }
+                    },
+                    if_not_exists = true,
+                })
+                f:create_index('primary', {
+                    parts = { 'key' },
+                    if_not_exists = true,
+                })
+            end
+        }
+    ]] } })
+    g.cluster.main_server:http_request('post', '/migrations/up', { json = {} })
+
+    fiber.sleep(0.5)
+
+    -- inject schema replication delay
+    g.cluster:server('storage-1-2').net_box:eval([[
+        box.space._space:before_replace(function(old, new) os.execute('sleep 0.5'); return new end)
+    ]])
+
+    -- change space format to make ddl schema incompatible
+    utils.set_sections(g, { { filename = "migrations/source/101_alter_space.lua", content = [[
+        return {
+            up = function()
+                box.space.somespace:format({
+                        { name = 'key', type = 'string' },
+                        { name = 'value', type = 'string', is_nullable = true },
+                        { name = 'secondvalue', type = 'string', is_nullable = true }
+                    })
+            end
+        }
+    ]] } })
+    g.cluster:server('storage-1-2'):http_request('post', '/migrations/up', { json = {} })
+end
+
+g.after_each(function()
+    g.cluster:server('storage-1-2').net_box:eval([[
+        local f = box.space._space:before_replace()
+         box.space._space:before_replace(nil, f[1])
+    ]])
+end)