rebalancer: refactoring of rebalancer_request_state

Before this patch the function `rebalancer_request_state` returned only
nil in case of errors (e.g. presence of SENDING, RECEIVING, GARBAGE
buckets, not active rebalancer). This makes it inconsistent compared to
other rebalancer functions.

Now, in case of errors we return `(nil, err)` instead of `nil`. It can
help us to propagate a meaningful error in rebalancer service. In
addition we introduce a new vshard error for cases when the storage has
some non-active buckets during rebalancing - `BUCKET_INVALID_STATE`.

Also we remove "Some buckets are not active" log from
`rebalancer_service_f` because a meaningful error about non-active
buckets is already returned from `rebalancer_download_states`.

NO_TEST=refactoring
NO_DOC=refactoring

Author: mrForza

test/rebalancer/rebalancer.result

@@ -318,12 +318,13 @@ _bucket:update({150}, {{'=', 2, vshard.consts.BUCKET.RECEIVING}})
 ---
 - [150, 'receiving']
 ...
-wait_rebalancer_state("Some buckets are not active", test_run)
----
-...
+-- We should not check the certain status of buckets (e.g. receiving) because
+-- in rare cases we accidentally can get the wrong one. For example we wait for
+-- "receiving" status, but get "garbage" due to some previous rebalancer error.
+wait_rebalancer_state('Error during downloading rebalancer states:.*' ..  \
+                      'BUCKET_INVALID_STATE', test_run)                   \
 _bucket:update({150}, {{'=', 2, vshard.consts.BUCKET.ACTIVE}})
 ---
-- [150, 'active']
 ...
 vshard.storage.sync()
 ---

test/rebalancer/rebalancer.test.lua

@@ -156,7 +156,11 @@ util.map_bucket_protection(test_run, {REPLICASET_1}, false)
 test_run:switch('box_1_a')
 vshard.storage.rebalancer_enable()
 _bucket:update({150}, {{'=', 2, vshard.consts.BUCKET.RECEIVING}})
-wait_rebalancer_state("Some buckets are not active", test_run)
+-- We should not check the certain status of buckets (e.g. receiving) because
+-- in rare cases we accidentally can get the wrong one. For example we wait for
+-- "receiving" status, but get "garbage" due to some previous rebalancer error.
+wait_rebalancer_state('Error during downloading rebalancer states:.*' ..  \
+                      'BUCKET_INVALID_STATE', test_run)                   \
 _bucket:update({150}, {{'=', 2, vshard.consts.BUCKET.ACTIVE}})
 vshard.storage.sync()
 

test/unit/rebalancer.result

@@ -290,9 +290,11 @@ build_routes(replicasets)
 vshard.storage.internal.is_master = true
 ---
 ...
-get_state = vshard.storage._rebalancer_request_state
----
-...
+get_state = function()                                          \
+    local res, err = vshard.storage._rebalancer_request_state() \
+    if res == nil then err.trace = nil end                      \
+    return res, err                                             \
+end                                                             \
 _bucket = box.schema.create_space('_bucket')
 ---
 ...
@@ -318,13 +320,21 @@ get_state()
 ---
 - bucket_active_count: 2
   bucket_pinned_count: 0
+- null
 ...
 _bucket:replace{1, consts.BUCKET.RECEIVING}
 ---
 - [1, 'receiving']
 ...
 get_state()
 ---
+- null
+- buckets_state: receiving
+  code: 42
+  replica_id: _
+  type: ShardingError
+  message: Replica _ has receiving buckets
+  name: BUCKET_INVALID_STATE
 ...
 vshard.storage.internal.is_master = false
 ---

test/unit/rebalancer.test.lua

@@ -76,7 +76,11 @@ build_routes(replicasets)
 -- Test rebalancer local state.
 --
 vshard.storage.internal.is_master = true
-get_state = vshard.storage._rebalancer_request_state
+get_state = function()                                          \
+    local res, err = vshard.storage._rebalancer_request_state() \
+    if res == nil then err.trace = nil end                      \
+    return res, err                                             \
+end                                                             \
 _bucket = box.schema.create_space('_bucket')
 pk = _bucket:create_index('pk')
 status = _bucket:create_index('status', {parts = {{2, 'string'}}, unique = false})

vshard/error.lua

@@ -207,6 +207,11 @@ local error_message_template = {
         msg = 'Mismatch server name: expected "%s", but got "%s"',
         args = {'expected_name', 'actual_name'},
     },
+    [42] = {
+        name = 'BUCKET_INVALID_STATE',
+        msg = 'Replica %s has %s buckets',
+        args = {'replica_id', 'buckets_state'}
+    }
 }
 
 --

vshard/storage/init.lua

@@ -2851,12 +2851,10 @@ local function rebalancer_service_f(service, limiter)
             return
         end
         if not status or replicasets == nil then
-            local err = status and total_bucket_active_count or replicasets
-            if err then
-                limiter:log_error(err, service:set_status_error(
-                    'Error during downloading rebalancer states: %s', err))
-            end
-            log.info('Some buckets are not active, retry rebalancing later')
+            local err = total_bucket_active_count
+            limiter:log_error(err, service:set_status_error(
+                'Error during downloading rebalancer states: %s, ' ..
+                'retry rebalancing later', err))
             service:set_activity('idling')
             lfiber.testcancel()
             lfiber.sleep(consts.REBALANCER_WORK_INTERVAL)
@@ -2945,18 +2943,22 @@ local function rebalancer_request_state()
         return nil, err
     end
     if not M.is_rebalancer_active or rebalancing_is_in_progress() then
-        return
+        return nil, lerror.make('Rebalancer is not active or is in progress')
     end
     local _bucket = box.space._bucket
     local status_index = _bucket.index.status
+    local repl_id = M.this_replica and M.this_replica.id or '_'
     if #status_index:select({BSENDING}, {limit = 1}) > 0 then
-        return
+        return nil, lerror.vshard(lerror.code.BUCKET_INVALID_STATE,
+                                  repl_id, 'sending')
     end
     if #status_index:select({BRECEIVING}, {limit = 1}) > 0 then
-        return
+        return nil, lerror.vshard(lerror.code.BUCKET_INVALID_STATE,
+                                  repl_id, 'receiving')
     end
     if #status_index:select({BGARBAGE}, {limit = 1}) > 0 then
-        return
+        return nil, lerror.vshard(lerror.code.BUCKET_INVALID_STATE,
+                                  repl_id, 'garbage')
     end
     return {
         bucket_active_count = status_index:count({BACTIVE}),