caleb-brown/redict
1
0
Fork 0
mirror of https://codeberg.org/redict/redict.git synced 2025-01-23 08:38:27 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Slave removal: replication.c logs fixed.

Browse Source
This commit is contained in:
antirez 2018-09-10 17:07:33 +02:00
parent cff5f36d94
commit 61b7a176ef
1 changed files with 35 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
src/replication.c
View File

@ -48,7 +48,7 @@ int cancelReplicationHandshake(void);
/* Return the pointer to a string representing the slave ip:listening_port /* Return the pointer to a string representing the slave ip:listening_port
* pair. Mostly useful for logging, since we want to log a slave using its * pair. Mostly useful for logging, since we want to log a slave using its
* IP address and its listening port which is more clear for the user, for * IP address and its listening port which is more clear for the user, for
* example: "Closing connection with slave 10.1.2.3:6380". */ * example: "Closing connection with replica 10.1.2.3:6380". */
char *replicationGetSlaveName(client *c) { char *replicationGetSlaveName(client *c) {
static char buf[NET_PEER_ID_LEN]; static char buf[NET_PEER_ID_LEN];
char ip[NET_IP_STR_LEN]; char ip[NET_IP_STR_LEN];
@ -64,7 +64,7 @@ char *replicationGetSlaveName(client *c) {
if (c->slave_listening_port) if (c->slave_listening_port)
anetFormatAddr(buf,sizeof(buf),ip,c->slave_listening_port); anetFormatAddr(buf,sizeof(buf),ip,c->slave_listening_port);
else else
snprintf(buf,sizeof(buf),"%s:<unknown-slave-port>",ip); snprintf(buf,sizeof(buf),"%s:<unknown-replica-port>",ip);
} else { } else {
snprintf(buf,sizeof(buf),"client id #%llu", snprintf(buf,sizeof(buf),"client id #%llu",
(unsigned long long) c->id); (unsigned long long) c->id);
@ -344,7 +344,7 @@ void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv,
long long addReplyReplicationBacklog(client *c, long long offset) { long long addReplyReplicationBacklog(client *c, long long offset) {
long long j, skip, len; long long j, skip, len;
serverLog(LL_DEBUG, "[PSYNC] Slave request offset: %lld", offset); serverLog(LL_DEBUG, "[PSYNC] Replica request offset: %lld", offset);
if (server.repl_backlog_histlen == 0) { if (server.repl_backlog_histlen == 0) {
serverLog(LL_DEBUG, "[PSYNC] Backlog history len is zero"); serverLog(LL_DEBUG, "[PSYNC] Backlog history len is zero");
@ -472,7 +472,7 @@ int masterTryPartialResynchronization(client *c) {
strcasecmp(master_replid, server.replid2)) strcasecmp(master_replid, server.replid2))
{ {
serverLog(LL_NOTICE,"Partial resynchronization not accepted: " serverLog(LL_NOTICE,"Partial resynchronization not accepted: "
"Replication ID mismatch (Slave asked for '%s', my " "Replication ID mismatch (Replica asked for '%s', my "
"replication IDs are '%s' and '%s')", "replication IDs are '%s' and '%s')",
master_replid, server.replid, server.replid2); master_replid, server.replid, server.replid2);
} else { } else {
@ -481,7 +481,7 @@ int masterTryPartialResynchronization(client *c) {
"up to %lld", psync_offset, server.second_replid_offset); "up to %lld", psync_offset, server.second_replid_offset);
} }
} else { } else {
serverLog(LL_NOTICE,"Full resync requested by slave %s", serverLog(LL_NOTICE,"Full resync requested by replica %s",
replicationGetSlaveName(c)); replicationGetSlaveName(c));
} }
goto need_full_resync; goto need_full_resync;
@ -493,10 +493,10 @@ int masterTryPartialResynchronization(client *c) {
psync_offset > (server.repl_backlog_off + server.repl_backlog_histlen)) psync_offset > (server.repl_backlog_off + server.repl_backlog_histlen))
{ {
serverLog(LL_NOTICE, serverLog(LL_NOTICE,
"Unable to partial resync with slave %s for lack of backlog (Slave request was: %lld).", replicationGetSlaveName(c), psync_offset); "Unable to partial resync with replica %s for lack of backlog (Replica request was: %lld).", replicationGetSlaveName(c), psync_offset);
if (psync_offset > server.master_repl_offset) { if (psync_offset > server.master_repl_offset) {
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Warning: slave %s tried to PSYNC with an offset that is greater than the master replication offset.", replicationGetSlaveName(c)); "Warning: replica %s tried to PSYNC with an offset that is greater than the master replication offset.", replicationGetSlaveName(c));
} }
goto need_full_resync; goto need_full_resync;
} }
@ -567,7 +567,7 @@ int startBgsaveForReplication(int mincapa) {
listNode *ln; listNode *ln;
serverLog(LL_NOTICE,"Starting BGSAVE for SYNC with target: %s", serverLog(LL_NOTICE,"Starting BGSAVE for SYNC with target: %s",
socket_target ? "slaves sockets" : "disk"); socket_target ? "replicas sockets" : "disk");
rdbSaveInfo rsi, *rsiptr; rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi); rsiptr = rdbPopulateSaveInfo(&rsi);
@ -644,7 +644,7 @@ void syncCommand(client *c) {
return; return;
} }
serverLog(LL_NOTICE,"Slave %s asks for synchronization", serverLog(LL_NOTICE,"Replica %s asks for synchronization",
replicationGetSlaveName(c)); replicationGetSlaveName(c));
/* Try a partial resynchronization if this is a PSYNC command. /* Try a partial resynchronization if this is a PSYNC command.
@ -725,7 +725,7 @@ void syncCommand(client *c) {
} else { } else {
/* No way, we need to wait for the next BGSAVE in order to /* No way, we need to wait for the next BGSAVE in order to
* register differences. */ * register differences. */
serverLog(LL_NOTICE,"Can't attach the slave to the current BGSAVE. Waiting for next BGSAVE for SYNC"); serverLog(LL_NOTICE,"Can't attach the replica to the current BGSAVE. Waiting for next BGSAVE for SYNC");
} }
/* CASE 2: BGSAVE is in progress, with socket target. */ /* CASE 2: BGSAVE is in progress, with socket target. */
@ -798,7 +798,7 @@ void replconfCommand(client *c) {
memcpy(c->slave_ip,ip,sdslen(ip)+1); memcpy(c->slave_ip,ip,sdslen(ip)+1);
} else { } else {
addReplyErrorFormat(c,"REPLCONF ip-address provided by " addReplyErrorFormat(c,"REPLCONF ip-address provided by "
"slave instance is too long: %zd bytes", sdslen(ip)); "replica instance is too long: %zd bytes", sdslen(ip));
return; return;
} }
} else if (!strcasecmp(c->argv[j]->ptr,"capa")) { } else if (!strcasecmp(c->argv[j]->ptr,"capa")) {
@ -858,12 +858,12 @@ void putSlaveOnline(client *slave) {
slave->repl_ack_time = server.unixtime; /* Prevent false timeout. */ slave->repl_ack_time = server.unixtime; /* Prevent false timeout. */
if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
sendReplyToClient, slave) == AE_ERR) { sendReplyToClient, slave) == AE_ERR) {
serverLog(LL_WARNING,"Unable to register writable event for slave bulk transfer: %s", strerror(errno)); serverLog(LL_WARNING,"Unable to register writable event for replica bulk transfer: %s", strerror(errno));
freeClient(slave); freeClient(slave);
return; return;
} }
refreshGoodSlavesCount(); refreshGoodSlavesCount();
serverLog(LL_NOTICE,"Synchronization with slave %s succeeded", serverLog(LL_NOTICE,"Synchronization with replica %s succeeded",
replicationGetSlaveName(slave)); replicationGetSlaveName(slave));
} }
@ -880,7 +880,7 @@ void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
if (slave->replpreamble) { if (slave->replpreamble) {
nwritten = write(fd,slave->replpreamble,sdslen(slave->replpreamble)); nwritten = write(fd,slave->replpreamble,sdslen(slave->replpreamble));
if (nwritten == -1) { if (nwritten == -1) {
serverLog(LL_VERBOSE,"Write error sending RDB preamble to slave: %s", serverLog(LL_VERBOSE,"Write error sending RDB preamble to replica: %s",
strerror(errno)); strerror(errno));
freeClient(slave); freeClient(slave);
return; return;
@ -900,14 +900,14 @@ void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
lseek(slave->repldbfd,slave->repldboff,SEEK_SET); lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
buflen = read(slave->repldbfd,buf,PROTO_IOBUF_LEN); buflen = read(slave->repldbfd,buf,PROTO_IOBUF_LEN);
if (buflen <= 0) { if (buflen <= 0) {
serverLog(LL_WARNING,"Read error sending DB to slave: %s", serverLog(LL_WARNING,"Read error sending DB to replica: %s",
(buflen == 0) ? "premature EOF" : strerror(errno)); (buflen == 0) ? "premature EOF" : strerror(errno));
freeClient(slave); freeClient(slave);
return; return;
} }
if ((nwritten = write(fd,buf,buflen)) == -1) { if ((nwritten = write(fd,buf,buflen)) == -1) {
if (errno != EAGAIN) { if (errno != EAGAIN) {
serverLog(LL_WARNING,"Write error sending DB to slave: %s", serverLog(LL_WARNING,"Write error sending DB to replica: %s",
strerror(errno)); strerror(errno));
freeClient(slave); freeClient(slave);
} }
@ -961,7 +961,7 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
* the slave online. */ * the slave online. */
if (type == RDB_CHILD_TYPE_SOCKET) { if (type == RDB_CHILD_TYPE_SOCKET) {
serverLog(LL_NOTICE, serverLog(LL_NOTICE,
"Streamed RDB transfer with slave %s succeeded (socket). Waiting for REPLCONF ACK from slave to enable streaming", "Streamed RDB transfer with replica %s succeeded (socket). Waiting for REPLCONF ACK from slave to enable streaming",
replicationGetSlaveName(slave)); replicationGetSlaveName(slave));
/* Note: we wait for a REPLCONF ACK message from slave in /* Note: we wait for a REPLCONF ACK message from slave in
* order to really put it online (install the write handler * order to really put it online (install the write handler
@ -1096,7 +1096,7 @@ void restartAOF() {
sleep(1); sleep(1);
} }
if (!retry) { if (!retry) {
serverLog(LL_WARNING,"FATAL: this slave instance finished the synchronization with its master, but the AOF can't be turned on. Exiting now."); serverLog(LL_WARNING,"FATAL: this replica instance finished the synchronization with its master, but the AOF can't be turned on. Exiting now.");
exit(1); exit(1);
} }
} }
@ -1161,12 +1161,12 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
* at the next call. */ * at the next call. */
server.repl_transfer_size = 0; server.repl_transfer_size = 0;
serverLog(LL_NOTICE, serverLog(LL_NOTICE,
"MASTER <-> SLAVE sync: receiving streamed RDB from master"); "MASTER <-> REPLICA sync: receiving streamed RDB from master");
} else { } else {
usemark = 0; usemark = 0;
server.repl_transfer_size = strtol(buf+1,NULL,10); server.repl_transfer_size = strtol(buf+1,NULL,10);
serverLog(LL_NOTICE, serverLog(LL_NOTICE,
"MASTER <-> SLAVE sync: receiving %lld bytes from master", "MASTER <-> REPLICA sync: receiving %lld bytes from master",
(long long) server.repl_transfer_size); (long long) server.repl_transfer_size);
} }
return; return;
@ -1207,7 +1207,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
server.repl_transfer_lastio = server.unixtime; server.repl_transfer_lastio = server.unixtime;
if ((nwritten = write(server.repl_transfer_fd,buf,nread)) != nread) { if ((nwritten = write(server.repl_transfer_fd,buf,nread)) != nread) {
serverLog(LL_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchronization: %s", serverLog(LL_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> REPLICA synchronization: %s",
(nwritten == -1) ? strerror(errno) : "short write"); (nwritten == -1) ? strerror(errno) : "short write");
goto error; goto error;
} }
@ -1246,11 +1246,11 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
int aof_is_enabled = server.aof_state != AOF_OFF; int aof_is_enabled = server.aof_state != AOF_OFF;
if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) { if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
serverLog(LL_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno)); serverLog(LL_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> REPLICA synchronization: %s", strerror(errno));
cancelReplicationHandshake(); cancelReplicationHandshake();
return; return;
} }
serverLog(LL_NOTICE, "MASTER <-> SLAVE sync: Flushing old data"); serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Flushing old data");
/* We need to stop any AOFRW fork before flusing and parsing /* We need to stop any AOFRW fork before flusing and parsing
* RDB, otherwise we'll create a copy-on-write disaster. */ * RDB, otherwise we'll create a copy-on-write disaster. */
if(aof_is_enabled) stopAppendOnly(); if(aof_is_enabled) stopAppendOnly();
@ -1264,7 +1264,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
* rdbLoad() will call the event loop to process events from time to * rdbLoad() will call the event loop to process events from time to
* time for non blocking loading. */ * time for non blocking loading. */
aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE); aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
serverLog(LL_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory"); serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Loading DB in memory");
rdbSaveInfo rsi = RDB_SAVE_INFO_INIT; rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
if (rdbLoad(server.rdb_filename,&rsi) != C_OK) { if (rdbLoad(server.rdb_filename,&rsi) != C_OK) {
serverLog(LL_WARNING,"Failed trying to load the MASTER synchronization DB from disk"); serverLog(LL_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
@ -1292,7 +1292,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
* masters after a failover. */ * masters after a failover. */
if (server.repl_backlog == NULL) createReplicationBacklog(); if (server.repl_backlog == NULL) createReplicationBacklog();
serverLog(LL_NOTICE, "MASTER <-> SLAVE sync: Finished with success"); serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Finished with success");
/* Restart the AOF subsystem now that we finished the sync. This /* Restart the AOF subsystem now that we finished the sync. This
* will trigger an AOF rewrite, and when done will start appending * will trigger an AOF rewrite, and when done will start appending
* to the new file. */ * to the new file. */
@ -1791,7 +1791,7 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
* uninstalling the read handler from the file descriptor. */ * uninstalling the read handler from the file descriptor. */
if (psync_result == PSYNC_CONTINUE) { if (psync_result == PSYNC_CONTINUE) {
serverLog(LL_NOTICE, "MASTER <-> SLAVE sync: Master accepted a Partial Resynchronization."); serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Master accepted a Partial Resynchronization.");
return; return;
} }
@ -1823,7 +1823,7 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
sleep(1); sleep(1);
} }
if (dfd == -1) { if (dfd == -1) {
serverLog(LL_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno)); serverLog(LL_WARNING,"Opening the temp file needed for MASTER <-> REPLICA synchronization: %s",strerror(errno));
goto error; goto error;
} }
@ -2001,7 +2001,7 @@ void replicaofCommand(client *c) {
/* SLAVEOF is not allowed in cluster mode as replication is automatically /* SLAVEOF is not allowed in cluster mode as replication is automatically
* configured using the current address of the master node. */ * configured using the current address of the master node. */
if (server.cluster_enabled) { if (server.cluster_enabled) {
addReplyError(c,"SLAVEOF not allowed in cluster mode."); addReplyError(c,"REPLICAOF not allowed in cluster mode.");
return; return;
} }
@ -2025,7 +2025,7 @@ void replicaofCommand(client *c) {
/* Check if we are already attached to the specified slave */ /* Check if we are already attached to the specified slave */
if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr) if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr)
&& server.masterport == port) { && server.masterport == port) {
serverLog(LL_NOTICE,"SLAVE OF would result into synchronization with the master we are already connected with. No operation performed."); serverLog(LL_NOTICE,"REPLICAOF would result into synchronization with the master we are already connected with. No operation performed.");
addReplySds(c,sdsnew("+OK Already connected to specified master\r\n")); addReplySds(c,sdsnew("+OK Already connected to specified master\r\n"));
return; return;
} }
@ -2033,7 +2033,7 @@ void replicaofCommand(client *c) {
* we can continue. */ * we can continue. */
replicationSetMaster(c->argv[1]->ptr, port); replicationSetMaster(c->argv[1]->ptr, port);
sds client = catClientInfoString(sdsempty(),c); sds client = catClientInfoString(sdsempty(),c);
serverLog(LL_NOTICE,"SLAVE OF %s:%d enabled (user request from '%s')", serverLog(LL_NOTICE,"REPLICAOF %s:%d enabled (user request from '%s')",
server.masterhost, server.masterport, client); server.masterhost, server.masterport, client);
sdsfree(client); sdsfree(client);
} }
@ -2191,7 +2191,7 @@ void replicationCacheMasterUsingMyself(void) {
unlinkClient(server.master); unlinkClient(server.master);
server.cached_master = server.master; server.cached_master = server.master;
server.master = NULL; server.master = NULL;
serverLog(LL_NOTICE,"Before turning into a slave, using my master parameters to synthesize a cached master: I may be able to synchronize with the new master with just a partial transfer."); serverLog(LL_NOTICE,"Before turning into a replica, using my master parameters to synthesize a cached master: I may be able to synchronize with the new master with just a partial transfer.");
} }
/* Free a cached master, called when there are no longer the conditions for /* Free a cached master, called when there are no longer the conditions for
@ -2407,7 +2407,7 @@ void waitCommand(client *c) {
long long offset = c->woff; long long offset = c->woff;
if (server.masterhost) { if (server.masterhost) {
addReplyError(c,"WAIT cannot be used with slave instances. Please also note that since Redis 4.0 if a slave is configured to be writable (which is not the default) writes to slaves are just local and are not propagated."); addReplyError(c,"WAIT cannot be used with replica instances. Please also note that since Redis 4.0 if a replica is configured to be writable (which is not the default) writes to replicas are just local and are not propagated.");
return; return;
} }
@ -2539,7 +2539,7 @@ void replicationCron(void) {
serverLog(LL_NOTICE,"Connecting to MASTER %s:%d", serverLog(LL_NOTICE,"Connecting to MASTER %s:%d",
server.masterhost, server.masterport); server.masterhost, server.masterport);
if (connectWithMaster() == C_OK) { if (connectWithMaster() == C_OK) {
serverLog(LL_NOTICE,"MASTER <-> SLAVE sync started"); serverLog(LL_NOTICE,"MASTER <-> REPLICA sync started");
} }
} }
@ -2611,7 +2611,7 @@ void replicationCron(void) {
if (slave->flags & CLIENT_PRE_PSYNC) continue; if (slave->flags & CLIENT_PRE_PSYNC) continue;
if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout) if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout)
{ {
serverLog(LL_WARNING, "Disconnecting timedout slave: %s", serverLog(LL_WARNING, "Disconnecting timedout replica: %s",
replicationGetSlaveName(slave)); replicationGetSlaveName(slave));
freeClient(slave); freeClient(slave);
} }
@ -2650,7 +2650,7 @@ void replicationCron(void) {
freeReplicationBacklog(); freeReplicationBacklog();
serverLog(LL_NOTICE, serverLog(LL_NOTICE,
"Replication backlog freed after %d seconds " "Replication backlog freed after %d seconds "
"without connected slaves.", "without connected replicas.",
(int) server.repl_backlog_time_limit); (int) server.repl_backlog_time_limit);
} }
} }