feature: master presence timout for get space

Added timeout condition for the validation of master presence in
replicaset and for the master connection to the `utils.get_space`
method.

Closes #95

Author: GRISHNOV

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ## [Unreleased]
 
 ### Added
+* Added timeout condition for the validation of master presence in 
+  replicaset and for the master connection (#95).
 
 ### Changed
 

crud/borders.lua

@@ -73,8 +73,7 @@ local function call_get_border_on_router(vshard_router, border_name, space_name,
         vshard_router = '?string|table',
     })
 
-    local replicasets = vshard_router:routeall()
-    local space, err = utils.get_space(space_name, replicasets)
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, BorderError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end
@@ -100,6 +99,10 @@ local function call_get_border_on_router(vshard_router, border_name, space_name,
     local cmp_key_parts = utils.merge_primary_key_parts(index.parts, primary_index.parts)
     local field_names = utils.enrich_field_names_with_cmp_key(opts.fields, cmp_key_parts, space:format())
 
+    local replicasets, err = vshard_router:routeall()
+    if err ~= nil then
+        return nil, BorderError:new("Failed to get router replicasets: %s", err)
+    end
     local call_opts = {
         mode = 'read',
         replicasets = replicasets,

crud/common/const.lua

@@ -5,6 +5,8 @@ const.RELOAD_SCHEMA_TIMEOUT = 3 -- 3 seconds
 const.FETCH_SHARDING_METADATA_TIMEOUT = 3 -- 3 seconds
 const.SHARDING_RELOAD_RETRIES_NUM = 1
 
+const.GET_SPACE_TIMEOUT = 3 -- 3 seconds
+
 const.NEED_SCHEMA_RELOAD   = 0x0001000
 const.NEED_SHARDING_RELOAD = 0x0001001
 

crud/common/utils.lua

@@ -21,7 +21,7 @@ local FilterFieldsError = errors.new_class('FilterFieldsError', {capture_stack =
 local NotInitializedError = errors.new_class('NotInitialized')
 local StorageInfoError = errors.new_class('StorageInfoError')
 local VshardRouterError = errors.new_class('VshardRouterError', {capture_stack = false})
-local fiber_clock = require('fiber').clock
+local fiber = require('fiber')
 
 local utils = {}
 
@@ -96,8 +96,26 @@ function utils.format_replicaset_error(replicaset_uuid, msg, ...)
     )
 end
 
-function utils.get_space(space_name, replicasets)
-    local replicaset = select(2, next(replicasets))
+function utils.get_space(space_name, vshard_router)
+    local replicasets, replicaset
+    local deadline = fiber.clock() + const.GET_SPACE_TIMEOUT
+    while (
+        -- Break if the deadline condition is exceeded.
+        -- Handling for deadline errors are below in the code.
+        fiber.clock() < deadline
+    ) do
+        -- Try to get master with timeout.
+        fiber.yield()
+        replicasets = vshard_router:routeall()
+        replicaset = select(2, next(replicasets))
+        if replicaset ~= nil and
+           replicaset.master ~= nil and
+           replicaset.master.conn.error == nil then
+            -- Break if there is master in replicaset and
+            -- connection to master is established.
+            break
+        end
+    end
 
     if replicaset == nil then
         return nil, GetSpaceError:new(
@@ -119,13 +137,14 @@ function utils.get_space(space_name, replicasets)
              replicaset.uuid, replicaset.master.conn.error)
         return nil, GetSpaceError:new(error_msg)
     end
+
     local space = replicaset.master.conn.space[space_name]
 
     return space
 end
 
-function utils.get_space_format(space_name, replicasets)
-    local space, err = utils.get_space(space_name, replicasets)
+function utils.get_space_format(space_name, vshard_router)
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, GetSpaceFormatError:new("An error occurred during the operation: %s", err)
     end
@@ -664,7 +683,7 @@ function utils.cut_rows(rows, metadata, field_names)
 end
 
 local function flatten_obj(vshard_router, space_name, obj, skip_nullability_check)
-    local space_format, err = utils.get_space_format(space_name, vshard_router:routeall())
+    local space_format, err = utils.get_space_format(space_name, vshard_router)
     if err ~= nil then
         return nil, FlattenError:new("Failed to get space format: %s", err), const.NEED_SCHEMA_RELOAD
     end
@@ -835,9 +854,9 @@ function utils.storage_info(opts)
         end
     end
 
-    local deadline = fiber_clock() + timeout
+    local deadline = fiber.clock() + timeout
     for replica_uuid, future in pairs(futures_by_replicas) do
-        local wait_timeout = deadline - fiber_clock()
+        local wait_timeout = deadline - fiber.clock()
         if wait_timeout < 0 then
             wait_timeout = 0
         end

crud/count.lua

@@ -132,12 +132,7 @@ local function call_count_on_router(vshard_router, space_name, user_conditions,
         return nil, CountError:new("Failed to parse conditions: %s", err)
     end
 
-    local replicasets, err = vshard_router:routeall()
-    if err ~= nil then
-        return nil, CountError:new("Failed to get router replicasets: %s", err)
-    end
-
-    local space, err = utils.get_space(space_name, replicasets)
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, CountError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end
@@ -169,7 +164,10 @@ local function call_count_on_router(vshard_router, space_name, user_conditions,
     check_count_safety(space_name, plan, opts)
 
     -- set replicasets to count from
-    local replicasets_to_count = replicasets
+    local replicasets_to_count, err = vshard_router:routeall()
+    if err ~= nil then
+        return nil, CountError:new("Failed to get router replicasets: %s", err)
+    end
 
     -- Whether to call one storage replicaset or perform
     -- map-reduce?

crud/delete.lua

@@ -62,7 +62,7 @@ local function call_delete_on_router(vshard_router, space_name, key, opts)
         vshard_router = '?string|table',
     })
 
-    local space, err = utils.get_space(space_name, vshard_router:routeall())
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, DeleteError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end

crud/get.lua

@@ -65,7 +65,7 @@ local function call_get_on_router(vshard_router, space_name, key, opts)
         vshard_router = '?string|table',
     })
 
-    local space, err = utils.get_space(space_name, vshard_router:routeall())
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, GetError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end

crud/insert.lua

@@ -64,7 +64,7 @@ local function call_insert_on_router(vshard_router, space_name, original_tuple,
         vshard_router = '?string|table',
     })
 
-    local space, err = utils.get_space(space_name, vshard_router:routeall())
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, InsertError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end

crud/insert_many.lua

@@ -131,7 +131,7 @@ local function call_insert_many_on_router(vshard_router, space_name, original_tu
         vshard_router = '?string|table',
     })
 
-    local space, err = utils.get_space(space_name, vshard_router:routeall())
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, {
             InsertManyError:new("An error occurred during the operation: %s", err)

crud/len.lua

@@ -64,7 +64,7 @@ function len.call(space_name, opts)
         return nil, LenError:new(err)
     end
 
-    local space, err = utils.get_space(space_name, vshard_router:routeall())
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, LenError:new("An error occurred during the operation: %s", err)
     end

crud/replace.lua

@@ -64,7 +64,7 @@ local function call_replace_on_router(vshard_router, space_name, original_tuple,
         vshard_router = '?string|table',
     })
 
-    local space, err = utils.get_space(space_name, vshard_router:routeall())
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, ReplaceError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end

crud/replace_many.lua

@@ -133,7 +133,7 @@ local function call_replace_many_on_router(vshard_router, space_name, original_t
         vshard_router = '?string|table',
     })
 
-    local space, err = utils.get_space(space_name, vshard_router:routeall())
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, {
             ReplaceManyError:new("An error occurred during the operation: %s", err)

crud/select/compat/select.lua

@@ -42,12 +42,7 @@ local function build_select_iterator(vshard_router, space_name, user_conditions,
         return nil, SelectError:new("Failed to parse conditions: %s", err)
     end
 
-    local replicasets, err = vshard_router:routeall()
-    if err ~= nil then
-        return nil, SelectError:new("Failed to get router replicasets: %s", err)
-    end
-
-    local space, err = utils.get_space(space_name, replicasets)
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, SelectError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end
@@ -84,7 +79,10 @@ local function build_select_iterator(vshard_router, space_name, user_conditions,
     end
 
     -- set replicasets to select from
-    local replicasets_to_select = replicasets
+    local replicasets_to_select, err = vshard_router:routeall()
+    if err ~= nil then
+        return nil, SelectError:new("Failed to get router replicasets: %s", err)
+    end
 
     -- Whether to call one storage replicaset or perform
     -- map-reduce?

crud/select/compat/select_old.lua

@@ -107,12 +107,7 @@ local function build_select_iterator(vshard_router, space_name, user_conditions,
         return nil, SelectError:new("Failed to parse conditions: %s", err)
     end
 
-    local replicasets, err = vshard_router:routeall()
-    if err ~= nil then
-        return nil, SelectError:new("Failed to get router replicasets: %s", err)
-    end
-
-    local space, err = utils.get_space(space_name, replicasets)
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, SelectError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end
@@ -151,7 +146,10 @@ local function build_select_iterator(vshard_router, space_name, user_conditions,
     end
 
     -- set replicasets to select from
-    local replicasets_to_select = replicasets
+    local replicasets_to_select, err = vshard_router:routeall()
+    if err ~= nil then
+        return nil, SelectError:new("Failed to get router replicasets: %s", err)
+    end
 
     -- See explanation of this logic in
     -- crud/select/compat/select.lua.

crud/stats/init.lua

@@ -264,7 +264,7 @@ local function resolve_space_name(space_id)
         return nil
     end
 
-    local space, err = utils.get_space(space_id, replicasets)
+    local space, err = utils.get_space(space_id, vshard_router)
     if err ~= nil then
         log.warn("An error occurred during getting space: %s", err)
         return nil

crud/update.lua

@@ -84,7 +84,7 @@ local function call_update_on_router(vshard_router, space_name, key, user_operat
         vshard_router = '?string|table',
     })
 
-    local space, err = utils.get_space(space_name, vshard_router:routeall())
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, UpdateError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end

crud/upsert.lua

@@ -62,7 +62,7 @@ local function call_upsert_on_router(vshard_router, space_name, original_tuple,
         vshard_router = '?string|table',
     })
 
-    local space, err = utils.get_space(space_name, vshard_router:routeall())
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, UpsertError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
     end

crud/upsert_many.lua

@@ -129,7 +129,7 @@ local function call_upsert_many_on_router(vshard_router, space_name, original_tu
         vshard_router = '?string|table',
     })
 
-    local space, err = utils.get_space(space_name, vshard_router:routeall())
+    local space, err = utils.get_space(space_name, vshard_router)
     if err ~= nil then
         return nil, {
             UpsertManyError:new("An error occurred during the operation: %s", err)

test/helper.lua

@@ -452,7 +452,7 @@ function helpers.is_space_exist(router, space_name)
         local vshard = require('vshard')
         local utils = require('crud.common.utils')
 
-        local space, err = utils.get_space(..., vshard.router.routeall())
+        local space, err = utils.get_space(..., vshard.router)
         if err ~= nil then
             return nil, err
         end