Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
3997dd6e
Commit
3997dd6e
authored
Oct 23, 2018
by
artix
Committed by
antirez
Nov 07, 2018
Browse files
Cluster Manager: better fix subcommand.
parent
bd80291c
Changes
1
Show whitespace changes
Inline
Side-by-side
src/redis-cli.c
View file @
3997dd6e
...
@@ -2726,8 +2726,7 @@ static int clusterManagerSetSlot(clusterManagerNode *node1,
...
@@ -2726,8 +2726,7 @@ static int clusterManagerSetSlot(clusterManagerNode *node1,
if (err != NULL) {
if (err != NULL) {
*err = zmalloc((reply->len + 1) * sizeof(char));
*err = zmalloc((reply->len + 1) * sizeof(char));
strcpy(*err, reply->str);
strcpy(*err, reply->str);
CLUSTER_MANAGER_PRINT_REPLY_ERROR(node1, err);
} else CLUSTER_MANAGER_PRINT_REPLY_ERROR(node1, reply->str);
}
goto cleanup;
goto cleanup;
}
}
cleanup:
cleanup:
...
@@ -2848,14 +2847,21 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
...
@@ -2848,14 +2847,21 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
if (migrate_reply == NULL) goto next;
if (migrate_reply == NULL) goto next;
if (migrate_reply->type == REDIS_REPLY_ERROR) {
if (migrate_reply->type == REDIS_REPLY_ERROR) {
if (do_fix && strstr(migrate_reply->str, "BUSYKEY")) {
if (do_fix && strstr(migrate_reply->str, "BUSYKEY")) {
/* If the key already exists, try to migrate keys
* adding REPLACE option.
* If the key's slot is not served, try to assign slot
* to the target node. */
int is_busy = (strstr(migrate_reply->str, "BUSYKEY") != NULL);
if (strstr(migrate_reply->str, "slot not served") != NULL)
clusterManagerSetSlot(source, target, slot, "node", NULL);
clusterManagerLogWarn("*** Target key exists. "
clusterManagerLogWarn("*** Target key exists. "
"Replacing it for FIX.\n");
"Replacing it for FIX.\n");
freeReplyObject(migrate_reply);
freeReplyObject(migrate_reply);
/* Try to migrate keys adding REPLACE option. */
migrate_reply = clusterManagerMigrateKeysInReply(source,
migrate_reply = clusterManagerMigrateKeysInReply(source,
target,
target,
reply,
reply,
1, timeout,
is_busy,
timeout,
NULL);
NULL);
success = (migrate_reply != NULL &&
success = (migrate_reply != NULL &&
migrate_reply->type != REDIS_REPLY_ERROR);
migrate_reply->type != REDIS_REPLY_ERROR);
...
@@ -3670,14 +3676,22 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
...
@@ -3670,14 +3676,22 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
while ((nln = listNext(&nli)) != NULL) {
while ((nln = listNext(&nli)) != NULL) {
clusterManagerNode *src = nln->value;
clusterManagerNode *src = nln->value;
if (src == target) continue;
if (src == target) continue;
/* Assign the slot to target node in the source node. */
redisReply *r = CLUSTER_MANAGER_COMMAND(src,
"CLUSTER SETSLOT %s %s %s", slot,
"NODE", target->name);
if (!clusterManagerCheckRedisReply(src, r, NULL))
fixed = -1;
if (r) freeReplyObject(r);
if (fixed < 0) goto cleanup;
/* Set the source node in 'importing' state
/* Set the source node in 'importing' state
* (even if we will actually migrate keys away)
* (even if we will actually migrate keys away)
* in order to avoid receiving redirections
* in order to avoid receiving redirections
* for MIGRATE. */
* for MIGRATE. */
redisReply *
r = CLUSTER_MANAGER_COMMAND(src,
r = CLUSTER_MANAGER_COMMAND(src,
"CLUSTER SETSLOT %s %s %s", slot,
"CLUSTER SETSLOT %s %s %s", slot,
"IMPORTING", target->name);
"IMPORTING", target->name);
if (!clusterManagerCheckRedisReply(
target
, r, NULL))
if (!clusterManagerCheckRedisReply(
src
, r, NULL))
fixed = -1;
fixed = -1;
if (r) freeReplyObject(r);
if (r) freeReplyObject(r);
if (fixed < 0) goto cleanup;
if (fixed < 0) goto cleanup;
...
@@ -3687,6 +3701,13 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
...
@@ -3687,6 +3701,13 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
fixed = -1;
fixed = -1;
goto cleanup;
goto cleanup;
}
}
r = CLUSTER_MANAGER_COMMAND(src,
"CLUSTER SETSLOT %s %s", slot,
"STABLE");
if (!clusterManagerCheckRedisReply(src, r, NULL))
fixed = -1;
if (r) freeReplyObject(r);
if (fixed < 0) goto cleanup;
}
}
fixed++;
fixed++;
}
}
...
@@ -3707,7 +3728,7 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -3707,7 +3728,7 @@ static int clusterManagerFixOpenSlot(int slot) {
clusterManagerLogInfo(">>> Fixing open slot %d\n", slot);
clusterManagerLogInfo(">>> Fixing open slot %d\n", slot);
/* Try to obtain the current slot owner, according to the current
/* Try to obtain the current slot owner, according to the current
* nodes configuration. */
* nodes configuration. */
int success = 1;
int success = 1
, keys_in_multiple_nodes = 0
;
list *owners = listCreate();
list *owners = listCreate();
list *migrating = listCreate();
list *migrating = listCreate();
list *importing = listCreate();
list *importing = listCreate();
...
@@ -3720,11 +3741,23 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -3720,11 +3741,23 @@ static int clusterManagerFixOpenSlot(int slot) {
while ((ln = listNext(&li)) != NULL) {
while ((ln = listNext(&li)) != NULL) {
clusterManagerNode *n = ln->value;
clusterManagerNode *n = ln->value;
if (n->flags & CLUSTER_MANAGER_FLAG_SLAVE) continue;
if (n->flags & CLUSTER_MANAGER_FLAG_SLAVE) continue;
if (n->slots[slot]) {
if (n->slots[slot]) listAddNodeTail(owners, n);
if (owner == NULL) owner = n;
else {
redisReply *r = CLUSTER_MANAGER_COMMAND(n,
"CLUSTER COUNTKEYSINSLOT %d", slot);
success = clusterManagerCheckRedisReply(n, r, NULL);
if (success && r->integer > 0) {
clusterManagerLogWarn("*** Found keys about slot %d "
"in non-owner node %s:%d!\n", slot,
n->ip, n->port);
listAddNodeTail(owners, n);
listAddNodeTail(owners, n);
keys_in_multiple_nodes = 1;
}
if (r) freeReplyObject(r);
if (!success) goto cleanup;
}
}
}
}
if (listLength(owners) == 1) owner = listFirst(owners)->value;
listRewind(cluster_manager.nodes, &li);
listRewind(cluster_manager.nodes, &li);
while ((ln = listNext(&li)) != NULL) {
while ((ln = listNext(&li)) != NULL) {
clusterManagerNode *n = ln->value;
clusterManagerNode *n = ln->value;
...
@@ -3767,6 +3800,9 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -3767,6 +3800,9 @@ static int clusterManagerFixOpenSlot(int slot) {
clusterManagerLogWarn("*** Found keys about slot %d "
clusterManagerLogWarn("*** Found keys about slot %d "
"in node %s:%d!\n", slot, n->ip,
"in node %s:%d!\n", slot, n->ip,
n->port);
n->port);
char *sep = (listLength(importing) == 0 ? "" : ",");
importing_str = sdscatfmt(importing_str, "%s%S:%u",
sep, n->ip, n->port);
listAddNodeTail(importing, n);
listAddNodeTail(importing, n);
}
}
if (r) freeReplyObject(r);
if (r) freeReplyObject(r);
...
@@ -3799,6 +3835,15 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -3799,6 +3835,15 @@ static int clusterManagerFixOpenSlot(int slot) {
success = clusterManagerCheckRedisReply(owner, reply, NULL);
success = clusterManagerCheckRedisReply(owner, reply, NULL);
if (reply) freeReplyObject(reply);
if (reply) freeReplyObject(reply);
if (!success) goto cleanup;
if (!success) goto cleanup;
/* Ensure that the slot is unassigned before assigning it to the
* owner. */
reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER DELSLOTS %d", slot);
success = clusterManagerCheckRedisReply(owner, reply, NULL);
/* Ignore "already unassigned" error. */
if (!success && reply && reply->type == REDIS_REPLY_ERROR &&
strstr(reply->str, "already unassigned") != NULL) success = 1;
if (reply) freeReplyObject(reply);
if (!success) goto cleanup;
reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER ADDSLOTS %d", slot);
reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER ADDSLOTS %d", slot);
success = clusterManagerCheckRedisReply(owner, reply, NULL);
success = clusterManagerCheckRedisReply(owner, reply, NULL);
if (reply) freeReplyObject(reply);
if (reply) freeReplyObject(reply);
...
@@ -3835,12 +3880,22 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -3835,12 +3880,22 @@ static int clusterManagerFixOpenSlot(int slot) {
if (n == owner) continue;
if (n == owner) continue;
reply = CLUSTER_MANAGER_COMMAND(n, "CLUSTER DELSLOTS %d", slot);
reply = CLUSTER_MANAGER_COMMAND(n, "CLUSTER DELSLOTS %d", slot);
success = clusterManagerCheckRedisReply(n, reply, NULL);
success = clusterManagerCheckRedisReply(n, reply, NULL);
/* Ignore "already unassigned" error. */
if (!success && reply && reply->type == REDIS_REPLY_ERROR &&
strstr(reply->str, "already unassigned") != NULL) success = 1;
if (reply) freeReplyObject(reply);
if (reply) freeReplyObject(reply);
if (!success) goto cleanup;
if (!success) goto cleanup;
n->slots[slot] = 0;
/* Assign the slot to the owner in the node 'n' configuration.' */
success = clusterManagerSetSlot(n, owner, slot, "node", NULL);
if (!success) goto cleanup;
success = clusterManagerSetSlot(n, owner, slot, "importing", NULL);
success = clusterManagerSetSlot(n, owner, slot, "importing", NULL);
if (!success) goto cleanup;
if (!success) goto cleanup;
clusterManagerRemoveNodeFromList(importing, n); //Avoid duplicates
/* Avoid duplicates. */
clusterManagerRemoveNodeFromList(importing, n);
listAddNodeTail(importing, n);
listAddNodeTail(importing, n);
/* Ensure that the node is not in the migrating list. */
clusterManagerRemoveNodeFromList(migrating, n);
}
}
reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER BUMPEPOCH");
reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER BUMPEPOCH");
success = clusterManagerCheckRedisReply(owner, reply, NULL);
success = clusterManagerCheckRedisReply(owner, reply, NULL);
...
@@ -3883,6 +3938,7 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -3883,6 +3938,7 @@ static int clusterManagerFixOpenSlot(int slot) {
listLength(migrating) == 1);
listLength(migrating) == 1);
if (try_to_close_slot) {
if (try_to_close_slot) {
clusterManagerNode *n = listFirst(migrating)->value;
clusterManagerNode *n = listFirst(migrating)->value;
if (!owner || owner != n) {
redisReply *r = CLUSTER_MANAGER_COMMAND(n,
redisReply *r = CLUSTER_MANAGER_COMMAND(n,
"CLUSTER GETKEYSINSLOT %d %d", slot, 10);
"CLUSTER GETKEYSINSLOT %d %d", slot, 10);
success = clusterManagerCheckRedisReply(n, r, NULL);
success = clusterManagerCheckRedisReply(n, r, NULL);
...
@@ -3892,9 +3948,11 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -3892,9 +3948,11 @@ static int clusterManagerFixOpenSlot(int slot) {
}
}
if (!success) goto cleanup;
if (!success) goto cleanup;
}
}
}
/* Case 3: There are no slots claiming to be in importing state, but
/* Case 3: There are no slots claiming to be in importing state, but
* there is a migrating node that actually don't have any key. We
* there is a migrating node that actually don't have any key or is the
* can just close the slot, probably a reshard interrupted in the middle. */
* slot owner. We can just close the slot, probably a reshard interrupted
* in the middle. */
if (try_to_close_slot) {
if (try_to_close_slot) {
clusterManagerNode *n = listFirst(migrating)->value;
clusterManagerNode *n = listFirst(migrating)->value;
redisReply *r = CLUSTER_MANAGER_COMMAND(n, "CLUSTER SETSLOT %d %s",
redisReply *r = CLUSTER_MANAGER_COMMAND(n, "CLUSTER SETSLOT %d %s",
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment