1 /* Redis Sentinel implementation 2 * 3 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> 4 * All rights reserved. 5 * 6 * Redistribution and use in source and binary forms, with or without 7 * modification, are permitted provided that the following conditions are met: 8 * 9 * * Redistributions of source code must retain the above copyright notice, 10 * this list of conditions and the following disclaimer. 11 * * Redistributions in binary form must reproduce the above copyright 12 * notice, this list of conditions and the following disclaimer in the 13 * documentation and/or other materials provided with the distribution. 14 * * Neither the name of Redis nor the names of its contributors may be used 15 * to endorse or promote products derived from this software without 16 * specific prior written permission. 17 * 18 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 19 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 20 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 21 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 22 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 23 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 24 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 25 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 26 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 27 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 28 * POSSIBILITY OF SUCH DAMAGE. 29 */ 30 31 #include "redis.h" 32 #include "hiredis.h" 33 #include "async.h" 34 35 #include <ctype.h> 36 #include <arpa/inet.h> 37 #include <sys/socket.h> 38 #include <sys/wait.h> 39 #include <fcntl.h> 40 41 extern char **environ; 42 43 #define REDIS_SENTINEL_PORT 26379 44 45 /* ======================== Sentinel global state =========================== */ 46 47 /* Address object, used to describe an ip:port pair. */ 48 typedef struct sentinelAddr { 49 char *ip; 50 int port; 51 } sentinelAddr; 52 53 /* A Sentinel Redis Instance object is monitoring. */ 54 #define SRI_MASTER (1<<0) 55 #define SRI_SLAVE (1<<1) 56 #define SRI_SENTINEL (1<<2) 57 #define SRI_DISCONNECTED (1<<3) 58 #define SRI_S_DOWN (1<<4) /* Subjectively down (no quorum). */ 59 #define SRI_O_DOWN (1<<5) /* Objectively down (confirmed by others). */ 60 #define SRI_MASTER_DOWN (1<<6) /* A Sentinel with this flag set thinks that 61 its master is down. */ 62 #define SRI_FAILOVER_IN_PROGRESS (1<<7) /* Failover is in progress for 63 this master. */ 64 #define SRI_PROMOTED (1<<8) /* Slave selected for promotion. */ 65 #define SRI_RECONF_SENT (1<<9) /* SLAVEOF <newmaster> sent. */ 66 #define SRI_RECONF_INPROG (1<<10) /* Slave synchronization in progress. */ 67 #define SRI_RECONF_DONE (1<<11) /* Slave synchronized with new master. */ 68 #define SRI_FORCE_FAILOVER (1<<12) /* Force failover with master up. */ 69 #define SRI_SCRIPT_KILL_SENT (1<<13) /* SCRIPT KILL already sent on -BUSY */ 70 71 /* Note: times are in milliseconds. */ 72 #define SENTINEL_INFO_PERIOD 10000 73 #define SENTINEL_PING_PERIOD 1000 74 #define SENTINEL_ASK_PERIOD 1000 75 #define SENTINEL_PUBLISH_PERIOD 2000 76 #define SENTINEL_DEFAULT_DOWN_AFTER 30000 77 #define SENTINEL_HELLO_CHANNEL "__sentinel__:hello" 78 #define SENTINEL_TILT_TRIGGER 2000 79 #define SENTINEL_TILT_PERIOD (SENTINEL_PING_PERIOD*30) 80 #define SENTINEL_DEFAULT_SLAVE_PRIORITY 100 81 #define SENTINEL_SLAVE_RECONF_TIMEOUT 10000 82 #define SENTINEL_DEFAULT_PARALLEL_SYNCS 1 83 #define SENTINEL_MIN_LINK_RECONNECT_PERIOD 15000 84 #define SENTINEL_DEFAULT_FAILOVER_TIMEOUT (60*3*1000) 85 #define SENTINEL_MAX_PENDING_COMMANDS 100 86 #define SENTINEL_ELECTION_TIMEOUT 10000 87 #define SENTINEL_MAX_DESYNC 1000 88 89 /* Failover machine different states. */ 90 #define SENTINEL_FAILOVER_STATE_NONE 0 /* No failover in progress. */ 91 #define SENTINEL_FAILOVER_STATE_WAIT_START 1 /* Wait for failover_start_time*/ 92 #define SENTINEL_FAILOVER_STATE_SELECT_SLAVE 2 /* Select slave to promote */ 93 #define SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE 3 /* Slave -> Master */ 94 #define SENTINEL_FAILOVER_STATE_WAIT_PROMOTION 4 /* Wait slave to change role */ 95 #define SENTINEL_FAILOVER_STATE_RECONF_SLAVES 5 /* SLAVEOF newmaster */ 96 #define SENTINEL_FAILOVER_STATE_UPDATE_CONFIG 6 /* Monitor promoted slave. */ 97 98 #define SENTINEL_MASTER_LINK_STATUS_UP 0 99 #define SENTINEL_MASTER_LINK_STATUS_DOWN 1 100 101 /* Generic flags that can be used with different functions. 102 * They use higher bits to avoid colliding with the function specific 103 * flags. */ 104 #define SENTINEL_NO_FLAGS 0 105 #define SENTINEL_GENERATE_EVENT (1<<16) 106 #define SENTINEL_LEADER (1<<17) 107 #define SENTINEL_OBSERVER (1<<18) 108 109 /* Script execution flags and limits. */ 110 #define SENTINEL_SCRIPT_NONE 0 111 #define SENTINEL_SCRIPT_RUNNING 1 112 #define SENTINEL_SCRIPT_MAX_QUEUE 256 113 #define SENTINEL_SCRIPT_MAX_RUNNING 16 114 #define SENTINEL_SCRIPT_MAX_RUNTIME 60000 /* 60 seconds max exec time. */ 115 #define SENTINEL_SCRIPT_MAX_RETRY 10 116 #define SENTINEL_SCRIPT_RETRY_DELAY 30000 /* 30 seconds between retries. */ 117 118 typedef struct sentinelRedisInstance { 119 int flags; /* See SRI_... defines */ 120 char *name; /* Master name from the point of view of this sentinel. */ 121 char *runid; /* run ID of this instance. */ 122 uint64_t config_epoch; /* Configuration epoch. */ 123 sentinelAddr *addr; /* Master host. */ 124 redisAsyncContext *cc; /* Hiredis context for commands. */ 125 redisAsyncContext *pc; /* Hiredis context for Pub / Sub. */ 126 int pending_commands; /* Number of commands sent waiting for a reply. */ 127 mstime_t cc_conn_time; /* cc connection time. */ 128 mstime_t pc_conn_time; /* pc connection time. */ 129 mstime_t pc_last_activity; /* Last time we received any message. */ 130 mstime_t last_avail_time; /* Last time the instance replied to ping with 131 a reply we consider valid. */ 132 mstime_t last_ping_time; /* Last time a pending ping was sent in the 133 context of the current command connection 134 with the instance. 0 if still not sent or 135 if pong already received. */ 136 mstime_t last_pong_time; /* Last time the instance replied to ping, 137 whatever the reply was. That's used to check 138 if the link is idle and must be reconnected. */ 139 mstime_t last_pub_time; /* Last time we sent hello via Pub/Sub. */ 140 mstime_t last_hello_time; /* Only used if SRI_SENTINEL is set. Last time 141 we received a hello from this Sentinel 142 via Pub/Sub. */ 143 mstime_t last_master_down_reply_time; /* Time of last reply to 144 SENTINEL is-master-down command. */ 145 mstime_t s_down_since_time; /* Subjectively down since time. */ 146 mstime_t o_down_since_time; /* Objectively down since time. */ 147 mstime_t down_after_period; /* Consider it down after that period. */ 148 mstime_t info_refresh; /* Time at which we received INFO output from it. */ 149 150 /* Role and the first time we observed it. 151 * This is useful in order to delay replacing what the instance reports 152 * with our own configuration. We need to always wait some time in order 153 * to give a chance to the leader to report the new configuration before 154 * we do silly things. */ 155 int role_reported; 156 mstime_t role_reported_time; 157 mstime_t slave_conf_change_time; /* Last time slave master addr changed. */ 158 159 /* Master specific. */ 160 dict *sentinels; /* Other sentinels monitoring the same master. */ 161 dict *slaves; /* Slaves for this master instance. */ 162 int quorum; /* Number of sentinels that need to agree on failure. */ 163 int parallel_syncs; /* How many slaves to reconfigure at same time. */ 164 char *auth_pass; /* Password to use for AUTH against master & slaves. */ 165 166 /* Slave specific. */ 167 mstime_t master_link_down_time; /* Slave replication link down time. */ 168 int slave_priority; /* Slave priority according to its INFO output. */ 169 mstime_t slave_reconf_sent_time; /* Time at which we sent SLAVE OF <new> */ 170 struct sentinelRedisInstance *master; /* Master instance if it's slave. */ 171 char *slave_master_host; /* Master host as reported by INFO */ 172 int slave_master_port; /* Master port as reported by INFO */ 173 int slave_master_link_status; /* Master link status as reported by INFO */ 174 unsigned long long slave_repl_offset; /* Slave replication offset. */ 175 /* Failover */ 176 char *leader; /* If this is a master instance, this is the runid of 177 the Sentinel that should perform the failover. If 178 this is a Sentinel, this is the runid of the Sentinel 179 that this Sentinel voted as leader. */ 180 uint64_t leader_epoch; /* Epoch of the 'leader' field. */ 181 uint64_t failover_epoch; /* Epoch of the currently started failover. */ 182 int failover_state; /* See SENTINEL_FAILOVER_STATE_* defines. */ 183 mstime_t failover_state_change_time; 184 mstime_t failover_start_time; /* Last failover attempt start time. */ 185 mstime_t failover_timeout; /* Max time to refresh failover state. */ 186 struct sentinelRedisInstance *promoted_slave; /* Promoted slave instance. */ 187 /* Scripts executed to notify admin or reconfigure clients: when they 188 * are set to NULL no script is executed. */ 189 char *notification_script; 190 char *client_reconfig_script; 191 } sentinelRedisInstance; 192 193 /* Main state. */ 194 struct sentinelState { 195 uint64_t current_epoch; /* Current epoch. */ 196 dict *masters; /* Dictionary of master sentinelRedisInstances. 197 Key is the instance name, value is the 198 sentinelRedisInstance structure pointer. */ 199 int tilt; /* Are we in TILT mode? */ 200 int running_scripts; /* Number of scripts in execution right now. */ 201 mstime_t tilt_start_time; /* When TITL started. */ 202 mstime_t previous_time; /* Last time we ran the time handler. */ 203 list *scripts_queue; /* Queue of user scripts to execute. */ 204 } sentinel; 205 206 /* A script execution job. */ 207 typedef struct sentinelScriptJob { 208 int flags; /* Script job flags: SENTINEL_SCRIPT_* */ 209 int retry_num; /* Number of times we tried to execute it. */ 210 char **argv; /* Arguments to call the script. */ 211 mstime_t start_time; /* Script execution time if the script is running, 212 otherwise 0 if we are allowed to retry the 213 execution at any time. If the script is not 214 running and it's not 0, it means: do not run 215 before the specified time. */ 216 pid_t pid; /* Script execution pid. */ 217 } sentinelScriptJob; 218 219 /* ======================= hiredis ae.c adapters ============================= 220 * Note: this implementation is taken from hiredis/adapters/ae.h, however 221 * we have our modified copy for Sentinel in order to use our allocator 222 * and to have full control over how the adapter works. */ 223 224 typedef struct redisAeEvents { 225 redisAsyncContext *context; 226 aeEventLoop *loop; 227 int fd; 228 int reading, writing; 229 } redisAeEvents; 230 231 static void redisAeReadEvent(aeEventLoop *el, int fd, void *privdata, int mask) { 232 ((void)el); ((void)fd); ((void)mask); 233 234 redisAeEvents *e = (redisAeEvents*)privdata; 235 redisAsyncHandleRead(e->context); 236 } 237 238 static void redisAeWriteEvent(aeEventLoop *el, int fd, void *privdata, int mask) { 239 ((void)el); ((void)fd); ((void)mask); 240 241 redisAeEvents *e = (redisAeEvents*)privdata; 242 redisAsyncHandleWrite(e->context); 243 } 244 245 static void redisAeAddRead(void *privdata) { 246 redisAeEvents *e = (redisAeEvents*)privdata; 247 aeEventLoop *loop = e->loop; 248 if (!e->reading) { 249 e->reading = 1; 250 aeCreateFileEvent(loop,e->fd,AE_READABLE,redisAeReadEvent,e); 251 } 252 } 253 254 static void redisAeDelRead(void *privdata) { 255 redisAeEvents *e = (redisAeEvents*)privdata; 256 aeEventLoop *loop = e->loop; 257 if (e->reading) { 258 e->reading = 0; 259 aeDeleteFileEvent(loop,e->fd,AE_READABLE); 260 } 261 } 262 263 static void redisAeAddWrite(void *privdata) { 264 redisAeEvents *e = (redisAeEvents*)privdata; 265 aeEventLoop *loop = e->loop; 266 if (!e->writing) { 267 e->writing = 1; 268 aeCreateFileEvent(loop,e->fd,AE_WRITABLE,redisAeWriteEvent,e); 269 } 270 } 271 272 static void redisAeDelWrite(void *privdata) { 273 redisAeEvents *e = (redisAeEvents*)privdata; 274 aeEventLoop *loop = e->loop; 275 if (e->writing) { 276 e->writing = 0; 277 aeDeleteFileEvent(loop,e->fd,AE_WRITABLE); 278 } 279 } 280 281 static void redisAeCleanup(void *privdata) { 282 redisAeEvents *e = (redisAeEvents*)privdata; 283 redisAeDelRead(privdata); 284 redisAeDelWrite(privdata); 285 zfree(e); 286 } 287 288 static int redisAeAttach(aeEventLoop *loop, redisAsyncContext *ac) { 289 redisContext *c = &(ac->c); 290 redisAeEvents *e; 291 292 /* Nothing should be attached when something is already attached */ 293 if (ac->ev.data != NULL) 294 return REDIS_ERR; 295 296 /* Create container for context and r/w events */ 297 e = (redisAeEvents*)zmalloc(sizeof(*e)); 298 e->context = ac; 299 e->loop = loop; 300 e->fd = c->fd; 301 e->reading = e->writing = 0; 302 303 /* Register functions to start/stop listening for events */ 304 ac->ev.addRead = redisAeAddRead; 305 ac->ev.delRead = redisAeDelRead; 306 ac->ev.addWrite = redisAeAddWrite; 307 ac->ev.delWrite = redisAeDelWrite; 308 ac->ev.cleanup = redisAeCleanup; 309 ac->ev.data = e; 310 311 return REDIS_OK; 312 } 313 314 /* ============================= Prototypes ================================= */ 315 316 void sentinelLinkEstablishedCallback(const redisAsyncContext *c, int status); 317 void sentinelDisconnectCallback(const redisAsyncContext *c, int status); 318 void sentinelReceiveHelloMessages(redisAsyncContext *c, void *reply, void *privdata); 319 sentinelRedisInstance *sentinelGetMasterByName(char *name); 320 char *sentinelGetSubjectiveLeader(sentinelRedisInstance *master); 321 char *sentinelGetObjectiveLeader(sentinelRedisInstance *master); 322 int yesnotoi(char *s); 323 void sentinelDisconnectInstanceFromContext(const redisAsyncContext *c); 324 void sentinelKillLink(sentinelRedisInstance *ri, redisAsyncContext *c); 325 const char *sentinelRedisInstanceTypeStr(sentinelRedisInstance *ri); 326 void sentinelAbortFailover(sentinelRedisInstance *ri); 327 void sentinelEvent(int level, char *type, sentinelRedisInstance *ri, const char *fmt, ...); 328 sentinelRedisInstance *sentinelSelectSlave(sentinelRedisInstance *master); 329 void sentinelScheduleScriptExecution(char *path, ...); 330 void sentinelStartFailover(sentinelRedisInstance *master); 331 void sentinelDiscardReplyCallback(redisAsyncContext *c, void *reply, void *privdata); 332 int sentinelSendSlaveOf(sentinelRedisInstance *ri, char *host, int port); 333 char *sentinelVoteLeader(sentinelRedisInstance *master, uint64_t req_epoch, char *req_runid, uint64_t *leader_epoch); 334 void sentinelFlushConfig(void); 335 void sentinelGenerateInitialMonitorEvents(void); 336 int sentinelSendPing(sentinelRedisInstance *ri); 337 338 /* ========================= Dictionary types =============================== */ 339 340 unsigned int dictSdsHash(const void *key); 341 int dictSdsKeyCompare(void *privdata, const void *key1, const void *key2); 342 void releaseSentinelRedisInstance(sentinelRedisInstance *ri); 343 344 void dictInstancesValDestructor (void *privdata, void *obj) { 345 releaseSentinelRedisInstance(obj); 346 } 347 348 /* Instance name (sds) -> instance (sentinelRedisInstance pointer) 349 * 350 * also used for: sentinelRedisInstance->sentinels dictionary that maps 351 * sentinels ip:port to last seen time in Pub/Sub hello message. */ 352 dictType instancesDictType = { 353 dictSdsHash, /* hash function */ 354 NULL, /* key dup */ 355 NULL, /* val dup */ 356 dictSdsKeyCompare, /* key compare */ 357 NULL, /* key destructor */ 358 dictInstancesValDestructor /* val destructor */ 359 }; 360 361 /* Instance runid (sds) -> votes (long casted to void*) 362 * 363 * This is useful into sentinelGetObjectiveLeader() function in order to 364 * count the votes and understand who is the leader. */ 365 dictType leaderVotesDictType = { 366 dictSdsHash, /* hash function */ 367 NULL, /* key dup */ 368 NULL, /* val dup */ 369 dictSdsKeyCompare, /* key compare */ 370 NULL, /* key destructor */ 371 NULL /* val destructor */ 372 }; 373 374 /* =========================== Initialization =============================== */ 375 376 void sentinelCommand(redisClient *c); 377 void sentinelInfoCommand(redisClient *c); 378 void sentinelSetCommand(redisClient *c); 379 void sentinelPublishCommand(redisClient *c); 380 381 struct redisCommand sentinelcmds[] = { 382 {"ping",pingCommand,1,"",0,NULL,0,0,0,0,0}, 383 {"sentinel",sentinelCommand,-2,"",0,NULL,0,0,0,0,0}, 384 {"subscribe",subscribeCommand,-2,"",0,NULL,0,0,0,0,0}, 385 {"unsubscribe",unsubscribeCommand,-1,"",0,NULL,0,0,0,0,0}, 386 {"psubscribe",psubscribeCommand,-2,"",0,NULL,0,0,0,0,0}, 387 {"punsubscribe",punsubscribeCommand,-1,"",0,NULL,0,0,0,0,0}, 388 {"publish",sentinelPublishCommand,3,"",0,NULL,0,0,0,0,0}, 389 {"info",sentinelInfoCommand,-1,"",0,NULL,0,0,0,0,0}, 390 {"shutdown",shutdownCommand,-1,"",0,NULL,0,0,0,0,0} 391 }; 392 393 /* This function overwrites a few normal Redis config default with Sentinel 394 * specific defaults. */ 395 void initSentinelConfig(void) { 396 server.port = REDIS_SENTINEL_PORT; 397 } 398 399 /* Perform the Sentinel mode initialization. */ 400 void initSentinel(void) { 401 int j; 402 403 /* Remove usual Redis commands from the command table, then just add 404 * the SENTINEL command. */ 405 dictEmpty(server.commands,NULL); 406 for (j = 0; j < sizeof(sentinelcmds)/sizeof(sentinelcmds[0]); j++) { 407 int retval; 408 struct redisCommand *cmd = sentinelcmds+j; 409 410 retval = dictAdd(server.commands, sdsnew(cmd->name), cmd); 411 redisAssert(retval == DICT_OK); 412 } 413 414 /* Initialize various data structures. */ 415 sentinel.current_epoch = 0; 416 sentinel.masters = dictCreate(&instancesDictType,NULL); 417 sentinel.tilt = 0; 418 sentinel.tilt_start_time = 0; 419 sentinel.previous_time = mstime(); 420 sentinel.running_scripts = 0; 421 sentinel.scripts_queue = listCreate(); 422 } 423 424 /* This function gets called when the server is in Sentinel mode, started, 425 * loaded the configuration, and is ready for normal operations. */ 426 void sentinelIsRunning(void) { 427 redisLog(REDIS_WARNING,"Sentinel runid is %s", server.runid); 428 429 if (server.configfile == NULL) { 430 redisLog(REDIS_WARNING, 431 "Sentinel started without a config file. Exiting..."); 432 exit(1); 433 } else if (access(server.configfile,W_OK) == -1) { 434 redisLog(REDIS_WARNING, 435 "Sentinel config file %s is not writable: %s. Exiting...", 436 server.configfile,strerror(errno)); 437 exit(1); 438 } 439 440 /* We want to generate a +monitor event for every configured master 441 * at startup. */ 442 sentinelGenerateInitialMonitorEvents(); 443 } 444 445 /* ============================== sentinelAddr ============================== */ 446 447 /* Create a sentinelAddr object and return it on success. 448 * On error NULL is returned and errno is set to: 449 * ENOENT: Can't resolve the hostname. 450 * EINVAL: Invalid port number. 451 */ 452 sentinelAddr *createSentinelAddr(char *hostname, int port) { 453 char buf[32]; 454 sentinelAddr *sa; 455 456 if (port <= 0 || port > 65535) { 457 errno = EINVAL; 458 return NULL; 459 } 460 if (anetResolve(NULL,hostname,buf,sizeof(buf)) == ANET_ERR) { 461 errno = ENOENT; 462 return NULL; 463 } 464 sa = zmalloc(sizeof(*sa)); 465 sa->ip = sdsnew(buf); 466 sa->port = port; 467 return sa; 468 } 469 470 /* Return a duplicate of the source address. */ 471 sentinelAddr *dupSentinelAddr(sentinelAddr *src) { 472 sentinelAddr *sa; 473 474 sa = zmalloc(sizeof(*sa)); 475 sa->ip = sdsnew(src->ip); 476 sa->port = src->port; 477 return sa; 478 } 479 480 /* Free a Sentinel address. Can't fail. */ 481 void releaseSentinelAddr(sentinelAddr *sa) { 482 sdsfree(sa->ip); 483 zfree(sa); 484 } 485 486 /* Return non-zero if two addresses are equal. */ 487 int sentinelAddrIsEqual(sentinelAddr *a, sentinelAddr *b) { 488 return a->port == b->port && !strcasecmp(a->ip,b->ip); 489 } 490 491 /* =========================== Events notification ========================== */ 492 493 /* Send an event to log, pub/sub, user notification script. 494 * 495 * 'level' is the log level for logging. Only REDIS_WARNING events will trigger 496 * the execution of the user notification script. 497 * 498 * 'type' is the message type, also used as a pub/sub channel name. 499 * 500 * 'ri', is the redis instance target of this event if applicable, and is 501 * used to obtain the path of the notification script to execute. 502 * 503 * The remaining arguments are printf-alike. 504 * If the format specifier starts with the two characters "%@" then ri is 505 * not NULL, and the message is prefixed with an instance identifier in the 506 * following format: 507 * 508 * <instance type> <instance name> <ip> <port> 509 * 510 * If the instance type is not master, than the additional string is 511 * added to specify the originating master: 512 * 513 * @ <master name> <master ip> <master port> 514 * 515 * Any other specifier after "%@" is processed by printf itself. 516 */ 517 void sentinelEvent(int level, char *type, sentinelRedisInstance *ri, 518 const char *fmt, ...) { 519 va_list ap; 520 char msg[REDIS_MAX_LOGMSG_LEN]; 521 robj *channel, *payload; 522 523 /* Handle %@ */ 524 if (fmt[0] == '%' && fmt[1] == '@') { 525 sentinelRedisInstance *master = (ri->flags & SRI_MASTER) ? 526 NULL : ri->master; 527 528 if (master) { 529 snprintf(msg, sizeof(msg), "%s %s %s %d @ %s %s %d", 530 sentinelRedisInstanceTypeStr(ri), 531 ri->name, ri->addr->ip, ri->addr->port, 532 master->name, master->addr->ip, master->addr->port); 533 } else { 534 snprintf(msg, sizeof(msg), "%s %s %s %d", 535 sentinelRedisInstanceTypeStr(ri), 536 ri->name, ri->addr->ip, ri->addr->port); 537 } 538 fmt += 2; 539 } else { 540 msg[0] = '\0'; 541 } 542 543 /* Use vsprintf for the rest of the formatting if any. */ 544 if (fmt[0] != '\0') { 545 va_start(ap, fmt); 546 vsnprintf(msg+strlen(msg), sizeof(msg)-strlen(msg), fmt, ap); 547 va_end(ap); 548 } 549 550 /* Log the message if the log level allows it to be logged. */ 551 if (level >= server.verbosity) 552 redisLog(level,"%s %s",type,msg); 553 554 /* Publish the message via Pub/Sub if it's not a debugging one. */ 555 if (level != REDIS_DEBUG) { 556 channel = createStringObject(type,strlen(type)); 557 payload = createStringObject(msg,strlen(msg)); 558 pubsubPublishMessage(channel,payload); 559 decrRefCount(channel); 560 decrRefCount(payload); 561 } 562 563 /* Call the notification script if applicable. */ 564 if (level == REDIS_WARNING && ri != NULL) { 565 sentinelRedisInstance *master = (ri->flags & SRI_MASTER) ? 566 ri : ri->master; 567 if (master->notification_script) { 568 sentinelScheduleScriptExecution(master->notification_script, 569 type,msg,NULL); 570 } 571 } 572 } 573 574 /* This function is called only at startup and is used to generate a 575 * +monitor event for every configured master. The same events are also 576 * generated when a master to monitor is added at runtime via the 577 * SENTINEL MONITOR command. */ 578 void sentinelGenerateInitialMonitorEvents(void) { 579 dictIterator *di; 580 dictEntry *de; 581 582 di = dictGetIterator(sentinel.masters); 583 while((de = dictNext(di)) != NULL) { 584 sentinelRedisInstance *ri = dictGetVal(de); 585 sentinelEvent(REDIS_WARNING,"+monitor",ri,"%@ quorum %d",ri->quorum); 586 } 587 dictReleaseIterator(di); 588 } 589 590 /* ============================ script execution ============================ */ 591 592 /* Release a script job structure and all the associated data. */ 593 void sentinelReleaseScriptJob(sentinelScriptJob *sj) { 594 int j = 0; 595 596 while(sj->argv[j]) sdsfree(sj->argv[j++]); 597 zfree(sj->argv); 598 zfree(sj); 599 } 600 601 #define SENTINEL_SCRIPT_MAX_ARGS 16 602 void sentinelScheduleScriptExecution(char *path, ...) { 603 va_list ap; 604 char *argv[SENTINEL_SCRIPT_MAX_ARGS+1]; 605 int argc = 1; 606 sentinelScriptJob *sj; 607 608 va_start(ap, path); 609 while(argc < SENTINEL_SCRIPT_MAX_ARGS) { 610 argv[argc] = va_arg(ap,char*); 611 if (!argv[argc]) break; 612 argv[argc] = sdsnew(argv[argc]); /* Copy the string. */ 613 argc++; 614 } 615 va_end(ap); 616 argv[0] = sdsnew(path); 617 618 sj = zmalloc(sizeof(*sj)); 619 sj->flags = SENTINEL_SCRIPT_NONE; 620 sj->retry_num = 0; 621 sj->argv = zmalloc(sizeof(char*)*(argc+1)); 622 sj->start_time = 0; 623 sj->pid = 0; 624 memcpy(sj->argv,argv,sizeof(char*)*(argc+1)); 625 626 listAddNodeTail(sentinel.scripts_queue,sj); 627 628 /* Remove the oldest non running script if we already hit the limit. */ 629 if (listLength(sentinel.scripts_queue) > SENTINEL_SCRIPT_MAX_QUEUE) { 630 listNode *ln; 631 listIter li; 632 633 listRewind(sentinel.scripts_queue,&li); 634 while ((ln = listNext(&li)) != NULL) { 635 sj = ln->value; 636 637 if (sj->flags & SENTINEL_SCRIPT_RUNNING) continue; 638 /* The first node is the oldest as we add on tail. */ 639 listDelNode(sentinel.scripts_queue,ln); 640 sentinelReleaseScriptJob(sj); 641 break; 642 } 643 redisAssert(listLength(sentinel.scripts_queue) <= 644 SENTINEL_SCRIPT_MAX_QUEUE); 645 } 646 } 647 648 /* Lookup a script in the scripts queue via pid, and returns the list node 649 * (so that we can easily remove it from the queue if needed). */ 650 listNode *sentinelGetScriptListNodeByPid(pid_t pid) { 651 listNode *ln; 652 listIter li; 653 654 listRewind(sentinel.scripts_queue,&li); 655 while ((ln = listNext(&li)) != NULL) { 656 sentinelScriptJob *sj = ln->value; 657 658 if ((sj->flags & SENTINEL_SCRIPT_RUNNING) && sj->pid == pid) 659 return ln; 660 } 661 return NULL; 662 } 663 664 /* Run pending scripts if we are not already at max number of running 665 * scripts. */ 666 void sentinelRunPendingScripts(void) { 667 listNode *ln; 668 listIter li; 669 mstime_t now = mstime(); 670 671 /* Find jobs that are not running and run them, from the top to the 672 * tail of the queue, so we run older jobs first. */ 673 listRewind(sentinel.scripts_queue,&li); 674 while (sentinel.running_scripts < SENTINEL_SCRIPT_MAX_RUNNING && 675 (ln = listNext(&li)) != NULL) 676 { 677 sentinelScriptJob *sj = ln->value; 678 pid_t pid; 679 680 /* Skip if already running. */ 681 if (sj->flags & SENTINEL_SCRIPT_RUNNING) continue; 682 683 /* Skip if it's a retry, but not enough time has elapsed. */ 684 if (sj->start_time && sj->start_time > now) continue; 685 686 sj->flags |= SENTINEL_SCRIPT_RUNNING; 687 sj->start_time = mstime(); 688 sj->retry_num++; 689 pid = fork(); 690 691 if (pid == -1) { 692 /* Parent (fork error). 693 * We report fork errors as signal 99, in order to unify the 694 * reporting with other kind of errors. */ 695 sentinelEvent(REDIS_WARNING,"-script-error",NULL, 696 "%s %d %d", sj->argv[0], 99, 0); 697 sj->flags &= ~SENTINEL_SCRIPT_RUNNING; 698 sj->pid = 0; 699 } else if (pid == 0) { 700 /* Child */ 701 execve(sj->argv[0],sj->argv,environ); 702 /* If we are here an error occurred. */ 703 _exit(2); /* Don't retry execution. */ 704 } else { 705 sentinel.running_scripts++; 706 sj->pid = pid; 707 sentinelEvent(REDIS_DEBUG,"+script-child",NULL,"%ld",(long)pid); 708 } 709 } 710 } 711 712 /* How much to delay the execution of a script that we need to retry after 713 * an error? 714 * 715 * We double the retry delay for every further retry we do. So for instance 716 * if RETRY_DELAY is set to 30 seconds and the max number of retries is 10 717 * starting from the second attempt to execute the script the delays are: 718 * 30 sec, 60 sec, 2 min, 4 min, 8 min, 16 min, 32 min, 64 min, 128 min. */ 719 mstime_t sentinelScriptRetryDelay(int retry_num) { 720 mstime_t delay = SENTINEL_SCRIPT_RETRY_DELAY; 721 722 while (retry_num-- > 1) delay *= 2; 723 return delay; 724 } 725 726 /* Check for scripts that terminated, and remove them from the queue if the 727 * script terminated successfully. If instead the script was terminated by 728 * a signal, or returned exit code "1", it is scheduled to run again if 729 * the max number of retries did not already elapsed. */ 730 void sentinelCollectTerminatedScripts(void) { 731 int statloc; 732 pid_t pid; 733 734 while ((pid = wait3(&statloc,WNOHANG,NULL)) > 0) { 735 int exitcode = WEXITSTATUS(statloc); 736 int bysignal = 0; 737 listNode *ln; 738 sentinelScriptJob *sj; 739 740 if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc); 741 sentinelEvent(REDIS_DEBUG,"-script-child",NULL,"%ld %d %d", 742 (long)pid, exitcode, bysignal); 743 744 ln = sentinelGetScriptListNodeByPid(pid); 745 if (ln == NULL) { 746 redisLog(REDIS_WARNING,"wait3() returned a pid (%ld) we can't find in our scripts execution queue!", (long)pid); 747 continue; 748 } 749 sj = ln->value; 750 751 /* If the script was terminated by a signal or returns an 752 * exit code of "1" (that means: please retry), we reschedule it 753 * if the max number of retries is not already reached. */ 754 if ((bysignal || exitcode == 1) && 755 sj->retry_num != SENTINEL_SCRIPT_MAX_RETRY) 756 { 757 sj->flags &= ~SENTINEL_SCRIPT_RUNNING; 758 sj->pid = 0; 759 sj->start_time = mstime() + 760 sentinelScriptRetryDelay(sj->retry_num); 761 } else { 762 /* Otherwise let's remove the script, but log the event if the 763 * execution did not terminated in the best of the ways. */ 764 if (bysignal || exitcode != 0) { 765 sentinelEvent(REDIS_WARNING,"-script-error",NULL, 766 "%s %d %d", sj->argv[0], bysignal, exitcode); 767 } 768 listDelNode(sentinel.scripts_queue,ln); 769 sentinelReleaseScriptJob(sj); 770 sentinel.running_scripts--; 771 } 772 } 773 } 774 775 /* Kill scripts in timeout, they'll be collected by the 776 * sentinelCollectTerminatedScripts() function. */ 777 void sentinelKillTimedoutScripts(void) { 778 listNode *ln; 779 listIter li; 780 mstime_t now = mstime(); 781 782 listRewind(sentinel.scripts_queue,&li); 783 while ((ln = listNext(&li)) != NULL) { 784 sentinelScriptJob *sj = ln->value; 785 786 if (sj->flags & SENTINEL_SCRIPT_RUNNING && 787 (now - sj->start_time) > SENTINEL_SCRIPT_MAX_RUNTIME) 788 { 789 sentinelEvent(REDIS_WARNING,"-script-timeout",NULL,"%s %ld", 790 sj->argv[0], (long)sj->pid); 791 kill(sj->pid,SIGKILL); 792 } 793 } 794 } 795 796 /* Implements SENTINEL PENDING-SCRIPTS command. */ 797 void sentinelPendingScriptsCommand(redisClient *c) { 798 listNode *ln; 799 listIter li; 800 801 addReplyMultiBulkLen(c,listLength(sentinel.scripts_queue)); 802 listRewind(sentinel.scripts_queue,&li); 803 while ((ln = listNext(&li)) != NULL) { 804 sentinelScriptJob *sj = ln->value; 805 int j = 0; 806 807 addReplyMultiBulkLen(c,10); 808 809 addReplyBulkCString(c,"argv"); 810 while (sj->argv[j]) j++; 811 addReplyMultiBulkLen(c,j); 812 j = 0; 813 while (sj->argv[j]) addReplyBulkCString(c,sj->argv[j++]); 814 815 addReplyBulkCString(c,"flags"); 816 addReplyBulkCString(c, 817 (sj->flags & SENTINEL_SCRIPT_RUNNING) ? "running" : "scheduled"); 818 819 addReplyBulkCString(c,"pid"); 820 addReplyBulkLongLong(c,sj->pid); 821 822 if (sj->flags & SENTINEL_SCRIPT_RUNNING) { 823 addReplyBulkCString(c,"run-time"); 824 addReplyBulkLongLong(c,mstime() - sj->start_time); 825 } else { 826 mstime_t delay = sj->start_time ? (sj->start_time-mstime()) : 0; 827 if (delay < 0) delay = 0; 828 addReplyBulkCString(c,"run-delay"); 829 addReplyBulkLongLong(c,delay); 830 } 831 832 addReplyBulkCString(c,"retry-num"); 833 addReplyBulkLongLong(c,sj->retry_num); 834 } 835 } 836 837 /* This function calls, if any, the client reconfiguration script with the 838 * following parameters: 839 * 840 * <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port> 841 * 842 * It is called every time a failover is performed. 843 * 844 * <state> is currently always "failover". 845 * <role> is either "leader" or "observer". 846 * 847 * from/to fields are respectively master -> promoted slave addresses for 848 * "start" and "end". */ 849 void sentinelCallClientReconfScript(sentinelRedisInstance *master, int role, char *state, sentinelAddr *from, sentinelAddr *to) { 850 char fromport[32], toport[32]; 851 852 if (master->client_reconfig_script == NULL) return; 853 ll2string(fromport,sizeof(fromport),from->port); 854 ll2string(toport,sizeof(toport),to->port); 855 sentinelScheduleScriptExecution(master->client_reconfig_script, 856 master->name, 857 (role == SENTINEL_LEADER) ? "leader" : "observer", 858 state, from->ip, fromport, to->ip, toport, NULL); 859 } 860 861 /* ========================== sentinelRedisInstance ========================= */ 862 863 /* Create a redis instance, the following fields must be populated by the 864 * caller if needed: 865 * runid: set to NULL but will be populated once INFO output is received. 866 * info_refresh: is set to 0 to mean that we never received INFO so far. 867 * 868 * If SRI_MASTER is set into initial flags the instance is added to 869 * sentinel.masters table. 870 * 871 * if SRI_SLAVE or SRI_SENTINEL is set then 'master' must be not NULL and the 872 * instance is added into master->slaves or master->sentinels table. 873 * 874 * If the instance is a slave or sentinel, the name parameter is ignored and 875 * is created automatically as hostname:port. 876 * 877 * The function fails if hostname can't be resolved or port is out of range. 878 * When this happens NULL is returned and errno is set accordingly to the 879 * createSentinelAddr() function. 880 * 881 * The function may also fail and return NULL with errno set to EBUSY if 882 * a master or slave with the same name already exists. */ 883 sentinelRedisInstance *createSentinelRedisInstance(char *name, int flags, char *hostname, int port, int quorum, sentinelRedisInstance *master) { 884 sentinelRedisInstance *ri; 885 sentinelAddr *addr; 886 dict *table = NULL; 887 char slavename[128], *sdsname; 888 889 redisAssert(flags & (SRI_MASTER|SRI_SLAVE|SRI_SENTINEL)); 890 redisAssert((flags & SRI_MASTER) || master != NULL); 891 892 /* Check address validity. */ 893 addr = createSentinelAddr(hostname,port); 894 if (addr == NULL) return NULL; 895 896 /* For slaves and sentinel we use ip:port as name. */ 897 if (flags & (SRI_SLAVE|SRI_SENTINEL)) { 898 snprintf(slavename,sizeof(slavename), 899 strchr(hostname,':') ? "[%s]:%d" : "%s:%d", 900 hostname,port); 901 name = slavename; 902 } 903 904 /* Make sure the entry is not duplicated. This may happen when the same 905 * name for a master is used multiple times inside the configuration or 906 * if we try to add multiple times a slave or sentinel with same ip/port 907 * to a master. */ 908 if (flags & SRI_MASTER) table = sentinel.masters; 909 else if (flags & SRI_SLAVE) table = master->slaves; 910 else if (flags & SRI_SENTINEL) table = master->sentinels; 911 sdsname = sdsnew(name); 912 if (dictFind(table,sdsname)) { 913 sdsfree(sdsname); 914 errno = EBUSY; 915 return NULL; 916 } 917 918 /* Create the instance object. */ 919 ri = zmalloc(sizeof(*ri)); 920 /* Note that all the instances are started in the disconnected state, 921 * the event loop will take care of connecting them. */ 922 ri->flags = flags | SRI_DISCONNECTED; 923 ri->name = sdsname; 924 ri->runid = NULL; 925 ri->config_epoch = 0; 926 ri->addr = addr; 927 ri->cc = NULL; 928 ri->pc = NULL; 929 ri->pending_commands = 0; 930 ri->cc_conn_time = 0; 931 ri->pc_conn_time = 0; 932 ri->pc_last_activity = 0; 933 /* We set the last_ping_time to "now" even if we actually don't have yet 934 * a connection with the node, nor we sent a ping. 935 * This is useful to detect a timeout in case we'll not be able to connect 936 * with the node at all. */ 937 ri->last_ping_time = mstime(); 938 ri->last_avail_time = mstime(); 939 ri->last_pong_time = mstime(); 940 ri->last_pub_time = mstime(); 941 ri->last_hello_time = mstime(); 942 ri->last_master_down_reply_time = mstime(); 943 ri->s_down_since_time = 0; 944 ri->o_down_since_time = 0; 945 ri->down_after_period = master ? master->down_after_period : 946 SENTINEL_DEFAULT_DOWN_AFTER; 947 ri->master_link_down_time = 0; 948 ri->auth_pass = NULL; 949 ri->slave_priority = SENTINEL_DEFAULT_SLAVE_PRIORITY; 950 ri->slave_reconf_sent_time = 0; 951 ri->slave_master_host = NULL; 952 ri->slave_master_port = 0; 953 ri->slave_master_link_status = SENTINEL_MASTER_LINK_STATUS_DOWN; 954 ri->slave_repl_offset = 0; 955 ri->sentinels = dictCreate(&instancesDictType,NULL); 956 ri->quorum = quorum; 957 ri->parallel_syncs = SENTINEL_DEFAULT_PARALLEL_SYNCS; 958 ri->master = master; 959 ri->slaves = dictCreate(&instancesDictType,NULL); 960 ri->info_refresh = 0; 961 962 /* Failover state. */ 963 ri->leader = NULL; 964 ri->leader_epoch = 0; 965 ri->failover_epoch = 0; 966 ri->failover_state = SENTINEL_FAILOVER_STATE_NONE; 967 ri->failover_state_change_time = 0; 968 ri->failover_start_time = 0; 969 ri->failover_timeout = SENTINEL_DEFAULT_FAILOVER_TIMEOUT; 970 ri->promoted_slave = NULL; 971 ri->notification_script = NULL; 972 ri->client_reconfig_script = NULL; 973 974 /* Role */ 975 ri->role_reported = ri->flags & (SRI_MASTER|SRI_SLAVE); 976 ri->role_reported_time = mstime(); 977 ri->slave_conf_change_time = mstime(); 978 979 /* Add into the right table. */ 980 dictAdd(table, ri->name, ri); 981 return ri; 982 } 983 984 /* Release this instance and all its slaves, sentinels, hiredis connections. 985 * This function does not take care of unlinking the instance from the main 986 * masters table (if it is a master) or from its master sentinels/slaves table 987 * if it is a slave or sentinel. */ 988 void releaseSentinelRedisInstance(sentinelRedisInstance *ri) { 989 /* Release all its slaves or sentinels if any. */ 990 dictRelease(ri->sentinels); 991 dictRelease(ri->slaves); 992 993 /* Release hiredis connections. */ 994 if (ri->cc) sentinelKillLink(ri,ri->cc); 995 if (ri->pc) sentinelKillLink(ri,ri->pc); 996 997 /* Free other resources. */ 998 sdsfree(ri->name); 999 sdsfree(ri->runid); 1000 sdsfree(ri->notification_script); 1001 sdsfree(ri->client_reconfig_script); 1002 sdsfree(ri->slave_master_host); 1003 sdsfree(ri->leader); 1004 sdsfree(ri->auth_pass); 1005 releaseSentinelAddr(ri->addr); 1006 1007 /* Clear state into the master if needed. */ 1008 if ((ri->flags & SRI_SLAVE) && (ri->flags & SRI_PROMOTED) && ri->master) 1009 ri->master->promoted_slave = NULL; 1010 1011 zfree(ri); 1012 } 1013 1014 /* Lookup a slave in a master Redis instance, by ip and port. */ 1015 sentinelRedisInstance *sentinelRedisInstanceLookupSlave( 1016 sentinelRedisInstance *ri, char *ip, int port) 1017 { 1018 sds key; 1019 sentinelRedisInstance *slave; 1020 1021 redisAssert(ri->flags & SRI_MASTER); 1022 key = sdscatprintf(sdsempty(), 1023 strchr(ip,':') ? "[%s]:%d" : "%s:%d", 1024 ip,port); 1025 slave = dictFetchValue(ri->slaves,key); 1026 sdsfree(key); 1027 return slave; 1028 } 1029 1030 /* Return the name of the type of the instance as a string. */ 1031 const char *sentinelRedisInstanceTypeStr(sentinelRedisInstance *ri) { 1032 if (ri->flags & SRI_MASTER) return "master"; 1033 else if (ri->flags & SRI_SLAVE) return "slave"; 1034 else if (ri->flags & SRI_SENTINEL) return "sentinel"; 1035 else return "unknown"; 1036 } 1037 1038 /* This function removes all the instances found in the dictionary of 1039 * sentinels in the specified 'master', having either: 1040 * 1041 * 1) The same ip/port as specified. 1042 * 2) The same runid. 1043 * 1044 * "1" and "2" don't need to verify at the same time, just one is enough. 1045 * If "runid" is NULL it is not checked. 1046 * Similarly if "ip" is NULL it is not checked. 1047 * 1048 * This function is useful because every time we add a new Sentinel into 1049 * a master's Sentinels dictionary, we want to be very sure about not 1050 * having duplicated instances for any reason. This is important because 1051 * other sentinels are needed to reach ODOWN quorum, and later to get 1052 * voted for a given configuration epoch in order to perform the failover. 1053 * 1054 * The function returns the number of Sentinels removed. */ 1055 int removeMatchingSentinelsFromMaster(sentinelRedisInstance *master, char *ip, int port, char *runid) { 1056 dictIterator *di; 1057 dictEntry *de; 1058 int removed = 0; 1059 1060 di = dictGetSafeIterator(master->sentinels); 1061 while((de = dictNext(di)) != NULL) { 1062 sentinelRedisInstance *ri = dictGetVal(de); 1063 1064 if ((ri->runid && runid && strcmp(ri->runid,runid) == 0) || 1065 (ip && strcmp(ri->addr->ip,ip) == 0 && port == ri->addr->port)) 1066 { 1067 dictDelete(master->sentinels,ri->name); 1068 removed++; 1069 } 1070 } 1071 dictReleaseIterator(di); 1072 return removed; 1073 } 1074 1075 /* Search an instance with the same runid, ip and port into a dictionary 1076 * of instances. Return NULL if not found, otherwise return the instance 1077 * pointer. 1078 * 1079 * runid or ip can be NULL. In such a case the search is performed only 1080 * by the non-NULL field. */ 1081 sentinelRedisInstance *getSentinelRedisInstanceByAddrAndRunID(dict *instances, char *ip, int port, char *runid) { 1082 dictIterator *di; 1083 dictEntry *de; 1084 sentinelRedisInstance *instance = NULL; 1085 1086 redisAssert(ip || runid); /* User must pass at least one search param. */ 1087 di = dictGetIterator(instances); 1088 while((de = dictNext(di)) != NULL) { 1089 sentinelRedisInstance *ri = dictGetVal(de); 1090 1091 if (runid && !ri->runid) continue; 1092 if ((runid == NULL || strcmp(ri->runid, runid) == 0) && 1093 (ip == NULL || (strcmp(ri->addr->ip, ip) == 0 && 1094 ri->addr->port == port))) 1095 { 1096 instance = ri; 1097 break; 1098 } 1099 } 1100 dictReleaseIterator(di); 1101 return instance; 1102 } 1103 1104 /* Master lookup by name */ 1105 sentinelRedisInstance *sentinelGetMasterByName(char *name) { 1106 sentinelRedisInstance *ri; 1107 sds sdsname = sdsnew(name); 1108 1109 ri = dictFetchValue(sentinel.masters,sdsname); 1110 sdsfree(sdsname); 1111 return ri; 1112 } 1113 1114 /* Add the specified flags to all the instances in the specified dictionary. */ 1115 void sentinelAddFlagsToDictOfRedisInstances(dict *instances, int flags) { 1116 dictIterator *di; 1117 dictEntry *de; 1118 1119 di = dictGetIterator(instances); 1120 while((de = dictNext(di)) != NULL) { 1121 sentinelRedisInstance *ri = dictGetVal(de); 1122 ri->flags |= flags; 1123 } 1124 dictReleaseIterator(di); 1125 } 1126 1127 /* Remove the specified flags to all the instances in the specified 1128 * dictionary. */ 1129 void sentinelDelFlagsToDictOfRedisInstances(dict *instances, int flags) { 1130 dictIterator *di; 1131 dictEntry *de; 1132 1133 di = dictGetIterator(instances); 1134 while((de = dictNext(di)) != NULL) { 1135 sentinelRedisInstance *ri = dictGetVal(de); 1136 ri->flags &= ~flags; 1137 } 1138 dictReleaseIterator(di); 1139 } 1140 1141 /* Reset the state of a monitored master: 1142 * 1) Remove all slaves. 1143 * 2) Remove all sentinels. 1144 * 3) Remove most of the flags resulting from runtime operations. 1145 * 4) Reset timers to their default value. 1146 * 5) In the process of doing this undo the failover if in progress. 1147 * 6) Disconnect the connections with the master (will reconnect automatically). 1148 */ 1149 1150 #define SENTINEL_RESET_NO_SENTINELS (1<<0) 1151 void sentinelResetMaster(sentinelRedisInstance *ri, int flags) { 1152 redisAssert(ri->flags & SRI_MASTER); 1153 dictRelease(ri->slaves); 1154 ri->slaves = dictCreate(&instancesDictType,NULL); 1155 if (!(flags & SENTINEL_RESET_NO_SENTINELS)) { 1156 dictRelease(ri->sentinels); 1157 ri->sentinels = dictCreate(&instancesDictType,NULL); 1158 } 1159 if (ri->cc) sentinelKillLink(ri,ri->cc); 1160 if (ri->pc) sentinelKillLink(ri,ri->pc); 1161 ri->flags &= SRI_MASTER|SRI_DISCONNECTED; 1162 if (ri->leader) { 1163 sdsfree(ri->leader); 1164 ri->leader = NULL; 1165 } 1166 ri->failover_state = SENTINEL_FAILOVER_STATE_NONE; 1167 ri->failover_state_change_time = 0; 1168 ri->failover_start_time = 0; 1169 ri->promoted_slave = NULL; 1170 sdsfree(ri->runid); 1171 sdsfree(ri->slave_master_host); 1172 ri->runid = NULL; 1173 ri->slave_master_host = NULL; 1174 ri->last_ping_time = mstime(); 1175 ri->last_avail_time = mstime(); 1176 ri->last_pong_time = mstime(); 1177 ri->role_reported_time = mstime(); 1178 ri->role_reported = SRI_MASTER; 1179 if (flags & SENTINEL_GENERATE_EVENT) 1180 sentinelEvent(REDIS_WARNING,"+reset-master",ri,"%@"); 1181 } 1182 1183 /* Call sentinelResetMaster() on every master with a name matching the specified 1184 * pattern. */ 1185 int sentinelResetMastersByPattern(char *pattern, int flags) { 1186 dictIterator *di; 1187 dictEntry *de; 1188 int reset = 0; 1189 1190 di = dictGetIterator(sentinel.masters); 1191 while((de = dictNext(di)) != NULL) { 1192 sentinelRedisInstance *ri = dictGetVal(de); 1193 1194 if (ri->name) { 1195 if (stringmatch(pattern,ri->name,0)) { 1196 sentinelResetMaster(ri,flags); 1197 reset++; 1198 } 1199 } 1200 } 1201 dictReleaseIterator(di); 1202 return reset; 1203 } 1204 1205 /* Reset the specified master with sentinelResetMaster(), and also change 1206 * the ip:port address, but take the name of the instance unmodified. 1207 * 1208 * This is used to handle the +switch-master event. 1209 * 1210 * The function returns REDIS_ERR if the address can't be resolved for some 1211 * reason. Otherwise REDIS_OK is returned. */ 1212 int sentinelResetMasterAndChangeAddress(sentinelRedisInstance *master, char *ip, int port) { 1213 sentinelAddr *oldaddr, *newaddr; 1214 sentinelAddr **slaves = NULL; 1215 int numslaves = 0, j; 1216 dictIterator *di; 1217 dictEntry *de; 1218 1219 newaddr = createSentinelAddr(ip,port); 1220 if (newaddr == NULL) return REDIS_ERR; 1221 1222 /* Make a list of slaves to add back after the reset. 1223 * Don't include the one having the address we are switching to. */ 1224 di = dictGetIterator(master->slaves); 1225 while((de = dictNext(di)) != NULL) { 1226 sentinelRedisInstance *slave = dictGetVal(de); 1227 1228 if (sentinelAddrIsEqual(slave->addr,newaddr)) continue; 1229 slaves = zrealloc(slaves,sizeof(sentinelAddr*)*(numslaves+1)); 1230 slaves[numslaves++] = createSentinelAddr(slave->addr->ip, 1231 slave->addr->port); 1232 } 1233 dictReleaseIterator(di); 1234 1235 /* If we are switching to a different address, include the old address 1236 * as a slave as well, so that we'll be able to sense / reconfigure 1237 * the old master. */ 1238 if (!sentinelAddrIsEqual(newaddr,master->addr)) { 1239 slaves = zrealloc(slaves,sizeof(sentinelAddr*)*(numslaves+1)); 1240 slaves[numslaves++] = createSentinelAddr(master->addr->ip, 1241 master->addr->port); 1242 } 1243 1244 /* Reset and switch address. */ 1245 sentinelResetMaster(master,SENTINEL_RESET_NO_SENTINELS); 1246 oldaddr = master->addr; 1247 master->addr = newaddr; 1248 master->o_down_since_time = 0; 1249 master->s_down_since_time = 0; 1250 1251 /* Add slaves back. */ 1252 for (j = 0; j < numslaves; j++) { 1253 sentinelRedisInstance *slave; 1254 1255 slave = createSentinelRedisInstance(NULL,SRI_SLAVE,slaves[j]->ip, 1256 slaves[j]->port, master->quorum, master); 1257 releaseSentinelAddr(slaves[j]); 1258 if (slave) { 1259 sentinelEvent(REDIS_NOTICE,"+slave",slave,"%@"); 1260 sentinelFlushConfig(); 1261 } 1262 } 1263 zfree(slaves); 1264 1265 /* Release the old address at the end so we are safe even if the function 1266 * gets the master->addr->ip and master->addr->port as arguments. */ 1267 releaseSentinelAddr(oldaddr); 1268 sentinelFlushConfig(); 1269 return REDIS_OK; 1270 } 1271 1272 /* Return non-zero if there was no SDOWN or ODOWN error associated to this 1273 * instance in the latest 'ms' milliseconds. */ 1274 int sentinelRedisInstanceNoDownFor(sentinelRedisInstance *ri, mstime_t ms) { 1275 mstime_t most_recent; 1276 1277 most_recent = ri->s_down_since_time; 1278 if (ri->o_down_since_time > most_recent) 1279 most_recent = ri->o_down_since_time; 1280 return most_recent == 0 || (mstime() - most_recent) > ms; 1281 } 1282 1283 /* Return the current master address, that is, its address or the address 1284 * of the promoted slave if already operational. */ 1285 sentinelAddr *sentinelGetCurrentMasterAddress(sentinelRedisInstance *master) { 1286 /* If we are failing over the master, and the state is already 1287 * SENTINEL_FAILOVER_STATE_RECONF_SLAVES or greater, it means that we 1288 * already have the new configuration epoch in the master, and the 1289 * slave acknowledged the configuration switch. Advertise the new 1290 * address. */ 1291 if ((master->flags & SRI_FAILOVER_IN_PROGRESS) && 1292 master->promoted_slave && 1293 master->failover_state >= SENTINEL_FAILOVER_STATE_RECONF_SLAVES) 1294 { 1295 return master->promoted_slave->addr; 1296 } else { 1297 return master->addr; 1298 } 1299 } 1300 1301 /* This function sets the down_after_period field value in 'master' to all 1302 * the slaves and sentinel instances connected to this master. */ 1303 void sentinelPropagateDownAfterPeriod(sentinelRedisInstance *master) { 1304 dictIterator *di; 1305 dictEntry *de; 1306 int j; 1307 dict *d[] = {master->slaves, master->sentinels, NULL}; 1308 1309 for (j = 0; d[j]; j++) { 1310 di = dictGetIterator(d[j]); 1311 while((de = dictNext(di)) != NULL) { 1312 sentinelRedisInstance *ri = dictGetVal(de); 1313 ri->down_after_period = master->down_after_period; 1314 } 1315 dictReleaseIterator(di); 1316 } 1317 } 1318 1319 /* ============================ Config handling ============================= */ 1320 char *sentinelHandleConfiguration(char **argv, int argc) { 1321 sentinelRedisInstance *ri; 1322 1323 if (!strcasecmp(argv[0],"monitor") && argc == 5) { 1324 /* monitor <name> <host> <port> <quorum> */ 1325 int quorum = atoi(argv[4]); 1326 1327 if (quorum <= 0) return "Quorum must be 1 or greater."; 1328 if (createSentinelRedisInstance(argv[1],SRI_MASTER,argv[2], 1329 atoi(argv[3]),quorum,NULL) == NULL) 1330 { 1331 switch(errno) { 1332 case EBUSY: return "Duplicated master name."; 1333 case ENOENT: return "Can't resolve master instance hostname."; 1334 case EINVAL: return "Invalid port number"; 1335 } 1336 } 1337 } else if (!strcasecmp(argv[0],"down-after-milliseconds") && argc == 3) { 1338 /* down-after-milliseconds <name> <milliseconds> */ 1339 ri = sentinelGetMasterByName(argv[1]); 1340 if (!ri) return "No such master with specified name."; 1341 ri->down_after_period = atoi(argv[2]); 1342 if (ri->down_after_period <= 0) 1343 return "negative or zero time parameter."; 1344 sentinelPropagateDownAfterPeriod(ri); 1345 } else if (!strcasecmp(argv[0],"failover-timeout") && argc == 3) { 1346 /* failover-timeout <name> <milliseconds> */ 1347 ri = sentinelGetMasterByName(argv[1]); 1348 if (!ri) return "No such master with specified name."; 1349 ri->failover_timeout = atoi(argv[2]); 1350 if (ri->failover_timeout <= 0) 1351 return "negative or zero time parameter."; 1352 } else if (!strcasecmp(argv[0],"parallel-syncs") && argc == 3) { 1353 /* parallel-syncs <name> <milliseconds> */ 1354 ri = sentinelGetMasterByName(argv[1]); 1355 if (!ri) return "No such master with specified name."; 1356 ri->parallel_syncs = atoi(argv[2]); 1357 } else if (!strcasecmp(argv[0],"notification-script") && argc == 3) { 1358 /* notification-script <name> <path> */ 1359 ri = sentinelGetMasterByName(argv[1]); 1360 if (!ri) return "No such master with specified name."; 1361 if (access(argv[2],X_OK) == -1) 1362 return "Notification script seems non existing or non executable."; 1363 ri->notification_script = sdsnew(argv[2]); 1364 } else if (!strcasecmp(argv[0],"client-reconfig-script") && argc == 3) { 1365 /* client-reconfig-script <name> <path> */ 1366 ri = sentinelGetMasterByName(argv[1]); 1367 if (!ri) return "No such master with specified name."; 1368 if (access(argv[2],X_OK) == -1) 1369 return "Client reconfiguration script seems non existing or " 1370 "non executable."; 1371 ri->client_reconfig_script = sdsnew(argv[2]); 1372 } else if (!strcasecmp(argv[0],"auth-pass") && argc == 3) { 1373 /* auth-pass <name> <password> */ 1374 ri = sentinelGetMasterByName(argv[1]); 1375 if (!ri) return "No such master with specified name."; 1376 ri->auth_pass = sdsnew(argv[2]); 1377 } else if (!strcasecmp(argv[0],"current-epoch") && argc == 2) { 1378 /* current-epoch <epoch> */ 1379 unsigned long long current_epoch = strtoull(argv[1],NULL,10); 1380 if (current_epoch > sentinel.current_epoch) 1381 sentinel.current_epoch = current_epoch; 1382 } else if (!strcasecmp(argv[0],"config-epoch") && argc == 3) { 1383 /* config-epoch <name> <epoch> */ 1384 ri = sentinelGetMasterByName(argv[1]); 1385 if (!ri) return "No such master with specified name."; 1386 ri->config_epoch = strtoull(argv[2],NULL,10); 1387 /* The following update of current_epoch is not really useful as 1388 * now the current epoch is persisted on the config file, but 1389 * we leave this check here for redundancy. */ 1390 if (ri->config_epoch > sentinel.current_epoch) 1391 sentinel.current_epoch = ri->config_epoch; 1392 } else if (!strcasecmp(argv[0],"leader-epoch") && argc == 3) { 1393 /* leader-epoch <name> <epoch> */ 1394 ri = sentinelGetMasterByName(argv[1]); 1395 if (!ri) return "No such master with specified name."; 1396 ri->leader_epoch = strtoull(argv[2],NULL,10); 1397 } else if (!strcasecmp(argv[0],"known-slave") && argc == 4) { 1398 sentinelRedisInstance *slave; 1399 1400 /* known-slave <name> <ip> <port> */ 1401 ri = sentinelGetMasterByName(argv[1]); 1402 if (!ri) return "No such master with specified name."; 1403 if ((slave = createSentinelRedisInstance(NULL,SRI_SLAVE,argv[2], 1404 atoi(argv[3]), ri->quorum, ri)) == NULL) 1405 { 1406 return "Wrong hostname or port for slave."; 1407 } 1408 } else if (!strcasecmp(argv[0],"known-sentinel") && 1409 (argc == 4 || argc == 5)) { 1410 sentinelRedisInstance *si; 1411 1412 /* known-sentinel <name> <ip> <port> [runid] */ 1413 ri = sentinelGetMasterByName(argv[1]); 1414 if (!ri) return "No such master with specified name."; 1415 if ((si = createSentinelRedisInstance(NULL,SRI_SENTINEL,argv[2], 1416 atoi(argv[3]), ri->quorum, ri)) == NULL) 1417 { 1418 return "Wrong hostname or port for sentinel."; 1419 } 1420 if (argc == 5) si->runid = sdsnew(argv[4]); 1421 } else { 1422 return "Unrecognized sentinel configuration statement."; 1423 } 1424 return NULL; 1425 } 1426 1427 /* Implements CONFIG REWRITE for "sentinel" option. 1428 * This is used not just to rewrite the configuration given by the user 1429 * (the configured masters) but also in order to retain the state of 1430 * Sentinel across restarts: config epoch of masters, associated slaves 1431 * and sentinel instances, and so forth. */ 1432 void rewriteConfigSentinelOption(struct rewriteConfigState *state) { 1433 dictIterator *di, *di2; 1434 dictEntry *de; 1435 sds line; 1436 1437 /* For every master emit a "sentinel monitor" config entry. */ 1438 di = dictGetIterator(sentinel.masters); 1439 while((de = dictNext(di)) != NULL) { 1440 sentinelRedisInstance *master, *ri; 1441 sentinelAddr *master_addr; 1442 1443 /* sentinel monitor */ 1444 master = dictGetVal(de); 1445 master_addr = sentinelGetCurrentMasterAddress(master); 1446 line = sdscatprintf(sdsempty(),"sentinel monitor %s %s %d %d", 1447 master->name, master_addr->ip, master_addr->port, 1448 master->quorum); 1449 rewriteConfigRewriteLine(state,"sentinel",line,1); 1450 1451 /* sentinel down-after-milliseconds */ 1452 if (master->down_after_period != SENTINEL_DEFAULT_DOWN_AFTER) { 1453 line = sdscatprintf(sdsempty(), 1454 "sentinel down-after-milliseconds %s %ld", 1455 master->name, (long) master->down_after_period); 1456 rewriteConfigRewriteLine(state,"sentinel",line,1); 1457 } 1458 1459 /* sentinel failover-timeout */ 1460 if (master->failover_timeout != SENTINEL_DEFAULT_FAILOVER_TIMEOUT) { 1461 line = sdscatprintf(sdsempty(), 1462 "sentinel failover-timeout %s %ld", 1463 master->name, (long) master->failover_timeout); 1464 rewriteConfigRewriteLine(state,"sentinel",line,1); 1465 } 1466 1467 /* sentinel parallel-syncs */ 1468 if (master->parallel_syncs != SENTINEL_DEFAULT_PARALLEL_SYNCS) { 1469 line = sdscatprintf(sdsempty(), 1470 "sentinel parallel-syncs %s %d", 1471 master->name, master->parallel_syncs); 1472 rewriteConfigRewriteLine(state,"sentinel",line,1); 1473 } 1474 1475 /* sentinel notification-script */ 1476 if (master->notification_script) { 1477 line = sdscatprintf(sdsempty(), 1478 "sentinel notification-script %s %s", 1479 master->name, master->notification_script); 1480 rewriteConfigRewriteLine(state,"sentinel",line,1); 1481 } 1482 1483 /* sentinel client-reconfig-script */ 1484 if (master->client_reconfig_script) { 1485 line = sdscatprintf(sdsempty(), 1486 "sentinel client-reconfig-script %s %s", 1487 master->name, master->client_reconfig_script); 1488 rewriteConfigRewriteLine(state,"sentinel",line,1); 1489 } 1490 1491 /* sentinel auth-pass */ 1492 if (master->auth_pass) { 1493 line = sdscatprintf(sdsempty(), 1494 "sentinel auth-pass %s %s", 1495 master->name, master->auth_pass); 1496 rewriteConfigRewriteLine(state,"sentinel",line,1); 1497 } 1498 1499 /* sentinel config-epoch */ 1500 line = sdscatprintf(sdsempty(), 1501 "sentinel config-epoch %s %llu", 1502 master->name, (unsigned long long) master->config_epoch); 1503 rewriteConfigRewriteLine(state,"sentinel",line,1); 1504 1505 /* sentinel leader-epoch */ 1506 line = sdscatprintf(sdsempty(), 1507 "sentinel leader-epoch %s %llu", 1508 master->name, (unsigned long long) master->leader_epoch); 1509 rewriteConfigRewriteLine(state,"sentinel",line,1); 1510 1511 /* sentinel known-slave */ 1512 di2 = dictGetIterator(master->slaves); 1513 while((de = dictNext(di2)) != NULL) { 1514 sentinelAddr *slave_addr; 1515 1516 ri = dictGetVal(de); 1517 slave_addr = ri->addr; 1518 1519 /* If master_addr (obtained using sentinelGetCurrentMasterAddress() 1520 * so it may be the address of the promoted slave) is equal to this 1521 * slave's address, a failover is in progress and the slave was 1522 * already successfully promoted. So as the address of this slave 1523 * we use the old master address instead. */ 1524 if (sentinelAddrIsEqual(slave_addr,master_addr)) 1525 slave_addr = master->addr; 1526 line = sdscatprintf(sdsempty(), 1527 "sentinel known-slave %s %s %d", 1528 master->name, ri->addr->ip, ri->addr->port); 1529 rewriteConfigRewriteLine(state,"sentinel",line,1); 1530 } 1531 dictReleaseIterator(di2); 1532 1533 /* sentinel known-sentinel */ 1534 di2 = dictGetIterator(master->sentinels); 1535 while((de = dictNext(di2)) != NULL) { 1536 ri = dictGetVal(de); 1537 line = sdscatprintf(sdsempty(), 1538 "sentinel known-sentinel %s %s %d%s%s", 1539 master->name, ri->addr->ip, ri->addr->port, 1540 ri->runid ? " " : "", 1541 ri->runid ? ri->runid : ""); 1542 rewriteConfigRewriteLine(state,"sentinel",line,1); 1543 } 1544 dictReleaseIterator(di2); 1545 } 1546 1547 /* sentinel current-epoch is a global state valid for all the masters. */ 1548 line = sdscatprintf(sdsempty(), 1549 "sentinel current-epoch %llu", (unsigned long long) sentinel.current_epoch); 1550 rewriteConfigRewriteLine(state,"sentinel",line,1); 1551 1552 dictReleaseIterator(di); 1553 } 1554 1555 /* This function uses the config rewriting Redis engine in order to persist 1556 * the state of the Sentinel in the current configuration file. 1557 * 1558 * Before returning the function calls fsync() against the generated 1559 * configuration file to make sure changes are committed to disk. 1560 * 1561 * On failure the function logs a warning on the Redis log. */ 1562 void sentinelFlushConfig(void) { 1563 int fd; 1564 int saved_hz = server.hz; 1565 1566 server.hz = REDIS_DEFAULT_HZ; 1567 if (rewriteConfig(server.configfile) != -1) { 1568 /* Rewrite succeded, fsync it. */ 1569 if ((fd = open(server.configfile,O_RDONLY)) != -1) { 1570 fsync(fd); 1571 close(fd); 1572 } 1573 } else { 1574 redisLog(REDIS_WARNING,"WARNING: Sentinel was not able to save the new configuration on disk!!!: %s", strerror(errno)); 1575 } 1576 server.hz = saved_hz; 1577 return; 1578 } 1579 1580 /* ====================== hiredis connection handling ======================= */ 1581 1582 /* Completely disconnect a hiredis link from an instance. */ 1583 void sentinelKillLink(sentinelRedisInstance *ri, redisAsyncContext *c) { 1584 if (ri->cc == c) { 1585 ri->cc = NULL; 1586 ri->pending_commands = 0; 1587 } 1588 if (ri->pc == c) ri->pc = NULL; 1589 c->data = NULL; 1590 ri->flags |= SRI_DISCONNECTED; 1591 redisAsyncFree(c); 1592 } 1593 1594 /* This function takes a hiredis context that is in an error condition 1595 * and make sure to mark the instance as disconnected performing the 1596 * cleanup needed. 1597 * 1598 * Note: we don't free the hiredis context as hiredis will do it for us 1599 * for async connections. */ 1600 void sentinelDisconnectInstanceFromContext(const redisAsyncContext *c) { 1601 sentinelRedisInstance *ri = c->data; 1602 int pubsub; 1603 1604 if (ri == NULL) return; /* The instance no longer exists. */ 1605 1606 pubsub = (ri->pc == c); 1607 sentinelEvent(REDIS_DEBUG, pubsub ? "-pubsub-link" : "-cmd-link", ri, 1608 "%@ #%s", c->errstr); 1609 if (pubsub) 1610 ri->pc = NULL; 1611 else 1612 ri->cc = NULL; 1613 ri->flags |= SRI_DISCONNECTED; 1614 } 1615 1616 void sentinelLinkEstablishedCallback(const redisAsyncContext *c, int status) { 1617 if (status != REDIS_OK) { 1618 sentinelDisconnectInstanceFromContext(c); 1619 } else { 1620 sentinelRedisInstance *ri = c->data; 1621 int pubsub = (ri->pc == c); 1622 1623 sentinelEvent(REDIS_DEBUG, pubsub ? "+pubsub-link" : "+cmd-link", ri, 1624 "%@"); 1625 } 1626 } 1627 1628 void sentinelDisconnectCallback(const redisAsyncContext *c, int status) { 1629 sentinelDisconnectInstanceFromContext(c); 1630 } 1631 1632 /* Send the AUTH command with the specified master password if needed. 1633 * Note that for slaves the password set for the master is used. 1634 * 1635 * We don't check at all if the command was successfully transmitted 1636 * to the instance as if it fails Sentinel will detect the instance down, 1637 * will disconnect and reconnect the link and so forth. */ 1638 void sentinelSendAuthIfNeeded(sentinelRedisInstance *ri, redisAsyncContext *c) { 1639 char *auth_pass = (ri->flags & SRI_MASTER) ? ri->auth_pass : 1640 ri->master->auth_pass; 1641 1642 if (auth_pass) { 1643 if (redisAsyncCommand(c, sentinelDiscardReplyCallback, NULL, "AUTH %s", 1644 auth_pass) == REDIS_OK) ri->pending_commands++; 1645 } 1646 } 1647 1648 /* Use CLIENT SETNAME to name the connection in the Redis instance as 1649 * sentinel-<first_8_chars_of_runid>-<connection_type> 1650 * The connection type is "cmd" or "pubsub" as specified by 'type'. 1651 * 1652 * This makes it possible to list all the sentinel instances connected 1653 * to a Redis servewr with CLIENT LIST, grepping for a specific name format. */ 1654 void sentinelSetClientName(sentinelRedisInstance *ri, redisAsyncContext *c, char *type) { 1655 char name[64]; 1656 1657 snprintf(name,sizeof(name),"sentinel-%.8s-%s",server.runid,type); 1658 if (redisAsyncCommand(c, sentinelDiscardReplyCallback, NULL, 1659 "CLIENT SETNAME %s", name) == REDIS_OK) 1660 { 1661 ri->pending_commands++; 1662 } 1663 } 1664 1665 /* Create the async connections for the specified instance if the instance 1666 * is disconnected. Note that the SRI_DISCONNECTED flag is set even if just 1667 * one of the two links (commands and pub/sub) is missing. */ 1668 void sentinelReconnectInstance(sentinelRedisInstance *ri) { 1669 if (!(ri->flags & SRI_DISCONNECTED)) return; 1670 1671 /* Commands connection. */ 1672 if (ri->cc == NULL) { 1673 ri->cc = redisAsyncConnect(ri->addr->ip,ri->addr->port); 1674 if (ri->cc->err) { 1675 sentinelEvent(REDIS_DEBUG,"-cmd-link-reconnection",ri,"%@ #%s", 1676 ri->cc->errstr); 1677 sentinelKillLink(ri,ri->cc); 1678 } else { 1679 ri->cc_conn_time = mstime(); 1680 ri->cc->data = ri; 1681 redisAeAttach(server.el,ri->cc); 1682 redisAsyncSetConnectCallback(ri->cc, 1683 sentinelLinkEstablishedCallback); 1684 redisAsyncSetDisconnectCallback(ri->cc, 1685 sentinelDisconnectCallback); 1686 sentinelSendAuthIfNeeded(ri,ri->cc); 1687 sentinelSetClientName(ri,ri->cc,"cmd"); 1688 1689 /* Send a PING ASAP when reconnecting. */ 1690 sentinelSendPing(ri); 1691 } 1692 } 1693 /* Pub / Sub */ 1694 if ((ri->flags & (SRI_MASTER|SRI_SLAVE)) && ri->pc == NULL) { 1695 ri->pc = redisAsyncConnect(ri->addr->ip,ri->addr->port); 1696 if (ri->pc->err) { 1697 sentinelEvent(REDIS_DEBUG,"-pubsub-link-reconnection",ri,"%@ #%s", 1698 ri->pc->errstr); 1699 sentinelKillLink(ri,ri->pc); 1700 } else { 1701 int retval; 1702 1703 ri->pc_conn_time = mstime(); 1704 ri->pc->data = ri; 1705 redisAeAttach(server.el,ri->pc); 1706 redisAsyncSetConnectCallback(ri->pc, 1707 sentinelLinkEstablishedCallback); 1708 redisAsyncSetDisconnectCallback(ri->pc, 1709 sentinelDisconnectCallback); 1710 sentinelSendAuthIfNeeded(ri,ri->pc); 1711 sentinelSetClientName(ri,ri->pc,"pubsub"); 1712 /* Now we subscribe to the Sentinels "Hello" channel. */ 1713 retval = redisAsyncCommand(ri->pc, 1714 sentinelReceiveHelloMessages, NULL, "SUBSCRIBE %s", 1715 SENTINEL_HELLO_CHANNEL); 1716 if (retval != REDIS_OK) { 1717 /* If we can't subscribe, the Pub/Sub connection is useless 1718 * and we can simply disconnect it and try again. */ 1719 sentinelKillLink(ri,ri->pc); 1720 return; 1721 } 1722 } 1723 } 1724 /* Clear the DISCONNECTED flags only if we have both the connections 1725 * (or just the commands connection if this is a sentinel instance). */ 1726 if (ri->cc && (ri->flags & SRI_SENTINEL || ri->pc)) 1727 ri->flags &= ~SRI_DISCONNECTED; 1728 } 1729 1730 /* ======================== Redis instances pinging ======================== */ 1731 1732 /* Return true if master looks "sane", that is: 1733 * 1) It is actually a master in the current configuration. 1734 * 2) It reports itself as a master. 1735 * 3) It is not SDOWN or ODOWN. 1736 * 4) We obtained last INFO no more than two times the INFO period time ago. */ 1737 int sentinelMasterLooksSane(sentinelRedisInstance *master) { 1738 return 1739 master->flags & SRI_MASTER && 1740 master->role_reported == SRI_MASTER && 1741 (master->flags & (SRI_S_DOWN|SRI_O_DOWN)) == 0 && 1742 (mstime() - master->info_refresh) < SENTINEL_INFO_PERIOD*2; 1743 } 1744 1745 /* Process the INFO output from masters. */ 1746 void sentinelRefreshInstanceInfo(sentinelRedisInstance *ri, const char *info) { 1747 sds *lines; 1748 int numlines, j; 1749 int role = 0; 1750 1751 /* The following fields must be reset to a given value in the case they 1752 * are not found at all in the INFO output. */ 1753 ri->master_link_down_time = 0; 1754 1755 /* Process line by line. */ 1756 lines = sdssplitlen(info,strlen(info),"\r\n",2,&numlines); 1757 for (j = 0; j < numlines; j++) { 1758 sentinelRedisInstance *slave; 1759 sds l = lines[j]; 1760 1761 /* run_id:<40 hex chars>*/ 1762 if (sdslen(l) >= 47 && !memcmp(l,"run_id:",7)) { 1763 if (ri->runid == NULL) { 1764 ri->runid = sdsnewlen(l+7,40); 1765 } else { 1766 if (strncmp(ri->runid,l+7,40) != 0) { 1767 sentinelEvent(REDIS_NOTICE,"+reboot",ri,"%@"); 1768 sdsfree(ri->runid); 1769 ri->runid = sdsnewlen(l+7,40); 1770 } 1771 } 1772 } 1773 1774 /* old versions: slave0:<ip>,<port>,<state> 1775 * new versions: slave0:ip=127.0.0.1,port=9999,... */ 1776 if ((ri->flags & SRI_MASTER) && 1777 sdslen(l) >= 7 && 1778 !memcmp(l,"slave",5) && isdigit(l[5])) 1779 { 1780 char *ip, *port, *end; 1781 1782 if (strstr(l,"ip=") == NULL) { 1783 /* Old format. */ 1784 ip = strchr(l,':'); if (!ip) continue; 1785 ip++; /* Now ip points to start of ip address. */ 1786 port = strchr(ip,','); if (!port) continue; 1787 *port = '\0'; /* nul term for easy access. */ 1788 port++; /* Now port points to start of port number. */ 1789 end = strchr(port,','); if (!end) continue; 1790 *end = '\0'; /* nul term for easy access. */ 1791 } else { 1792 /* New format. */ 1793 ip = strstr(l,"ip="); if (!ip) continue; 1794 ip += 3; /* Now ip points to start of ip address. */ 1795 port = strstr(l,"port="); if (!port) continue; 1796 port += 5; /* Now port points to start of port number. */ 1797 /* Nul term both fields for easy access. */ 1798 end = strchr(ip,','); if (end) *end = '\0'; 1799 end = strchr(port,','); if (end) *end = '\0'; 1800 } 1801 1802 /* Check if we already have this slave into our table, 1803 * otherwise add it. */ 1804 if (sentinelRedisInstanceLookupSlave(ri,ip,atoi(port)) == NULL) { 1805 if ((slave = createSentinelRedisInstance(NULL,SRI_SLAVE,ip, 1806 atoi(port), ri->quorum, ri)) != NULL) 1807 { 1808 sentinelEvent(REDIS_NOTICE,"+slave",slave,"%@"); 1809 } 1810 } 1811 } 1812 1813 /* master_link_down_since_seconds:<seconds> */ 1814 if (sdslen(l) >= 32 && 1815 !memcmp(l,"master_link_down_since_seconds",30)) 1816 { 1817 ri->master_link_down_time = strtoll(l+31,NULL,10)*1000; 1818 } 1819 1820 /* role:<role> */ 1821 if (!memcmp(l,"role:master",11)) role = SRI_MASTER; 1822 else if (!memcmp(l,"role:slave",10)) role = SRI_SLAVE; 1823 1824 if (role == SRI_SLAVE) { 1825 /* master_host:<host> */ 1826 if (sdslen(l) >= 12 && !memcmp(l,"master_host:",12)) { 1827 if (ri->slave_master_host == NULL || 1828 strcasecmp(l+12,ri->slave_master_host)) 1829 { 1830 sdsfree(ri->slave_master_host); 1831 ri->slave_master_host = sdsnew(l+12); 1832 ri->slave_conf_change_time = mstime(); 1833 } 1834 } 1835 1836 /* master_port:<port> */ 1837 if (sdslen(l) >= 12 && !memcmp(l,"master_port:",12)) { 1838 int slave_master_port = atoi(l+12); 1839 1840 if (ri->slave_master_port != slave_master_port) { 1841 ri->slave_master_port = slave_master_port; 1842 ri->slave_conf_change_time = mstime(); 1843 } 1844 } 1845 1846 /* master_link_status:<status> */ 1847 if (sdslen(l) >= 19 && !memcmp(l,"master_link_status:",19)) { 1848 ri->slave_master_link_status = 1849 (strcasecmp(l+19,"up") == 0) ? 1850 SENTINEL_MASTER_LINK_STATUS_UP : 1851 SENTINEL_MASTER_LINK_STATUS_DOWN; 1852 } 1853 1854 /* slave_priority:<priority> */ 1855 if (sdslen(l) >= 15 && !memcmp(l,"slave_priority:",15)) 1856 ri->slave_priority = atoi(l+15); 1857 1858 /* slave_repl_offset:<offset> */ 1859 if (sdslen(l) >= 18 && !memcmp(l,"slave_repl_offset:",18)) 1860 ri->slave_repl_offset = strtoull(l+18,NULL,10); 1861 } 1862 } 1863 ri->info_refresh = mstime(); 1864 sdsfreesplitres(lines,numlines); 1865 1866 /* ---------------------------- Acting half ----------------------------- 1867 * Some things will not happen if sentinel.tilt is true, but some will 1868 * still be processed. */ 1869 1870 /* Remember when the role changed. */ 1871 if (role != ri->role_reported) { 1872 ri->role_reported_time = mstime(); 1873 ri->role_reported = role; 1874 if (role == SRI_SLAVE) ri->slave_conf_change_time = mstime(); 1875 /* Log the event with +role-change if the new role is coherent or 1876 * with -role-change if there is a mismatch with the current config. */ 1877 sentinelEvent(REDIS_VERBOSE, 1878 ((ri->flags & (SRI_MASTER|SRI_SLAVE)) == role) ? 1879 "+role-change" : "-role-change", 1880 ri, "%@ new reported role is %s", 1881 role == SRI_MASTER ? "master" : "slave", 1882 ri->flags & SRI_MASTER ? "master" : "slave"); 1883 } 1884 1885 /* None of the following conditions are processed when in tilt mode, so 1886 * return asap. */ 1887 if (sentinel.tilt) return; 1888 1889 /* Handle master -> slave role switch. */ 1890 if ((ri->flags & SRI_MASTER) && role == SRI_SLAVE) { 1891 /* Nothing to do, but masters claiming to be slaves are 1892 * considered to be unreachable by Sentinel, so eventually 1893 * a failover will be triggered. */ 1894 } 1895 1896 /* Handle slave -> master role switch. */ 1897 if ((ri->flags & SRI_SLAVE) && role == SRI_MASTER) { 1898 /* If this is a promoted slave we can change state to the 1899 * failover state machine. */ 1900 if ((ri->master->flags & SRI_FAILOVER_IN_PROGRESS) && 1901 (ri->master->failover_state == 1902 SENTINEL_FAILOVER_STATE_WAIT_PROMOTION)) 1903 { 1904 /* Now that we are sure the slave was reconfigured as a master 1905 * set the master configuration epoch to the epoch we won the 1906 * election to perform this failover. This will force the other 1907 * Sentinels to update their config (assuming there is not 1908 * a newer one already available). */ 1909 ri->master->config_epoch = ri->master->failover_epoch; 1910 ri->master->failover_state = SENTINEL_FAILOVER_STATE_RECONF_SLAVES; 1911 ri->master->failover_state_change_time = mstime(); 1912 sentinelFlushConfig(); 1913 sentinelEvent(REDIS_WARNING,"+promoted-slave",ri,"%@"); 1914 sentinelEvent(REDIS_WARNING,"+failover-state-reconf-slaves", 1915 ri->master,"%@"); 1916 sentinelCallClientReconfScript(ri->master,SENTINEL_LEADER, 1917 "start",ri->master->addr,ri->addr); 1918 } else { 1919 /* A slave turned into a master. We want to force our view and 1920 * reconfigure as slave. Wait some time after the change before 1921 * going forward, to receive new configs if any. */ 1922 mstime_t wait_time = SENTINEL_PUBLISH_PERIOD*4; 1923 1924 if (sentinelMasterLooksSane(ri->master) && 1925 sentinelRedisInstanceNoDownFor(ri,wait_time) && 1926 mstime() - ri->role_reported_time > wait_time) 1927 { 1928 int retval = sentinelSendSlaveOf(ri, 1929 ri->master->addr->ip, 1930 ri->master->addr->port); 1931 if (retval == REDIS_OK) 1932 sentinelEvent(REDIS_NOTICE,"+convert-to-slave",ri,"%@"); 1933 } 1934 } 1935 } 1936 1937 /* Handle slaves replicating to a different master address. */ 1938 if ((ri->flags & SRI_SLAVE) && 1939 role == SRI_SLAVE && 1940 (ri->slave_master_port != ri->master->addr->port || 1941 strcasecmp(ri->slave_master_host,ri->master->addr->ip))) 1942 { 1943 mstime_t wait_time = ri->master->failover_timeout; 1944 1945 /* Make sure the master is sane before reconfiguring this instance 1946 * into a slave. */ 1947 if (sentinelMasterLooksSane(ri->master) && 1948 sentinelRedisInstanceNoDownFor(ri,wait_time) && 1949 mstime() - ri->slave_conf_change_time > wait_time) 1950 { 1951 int retval = sentinelSendSlaveOf(ri, 1952 ri->master->addr->ip, 1953 ri->master->addr->port); 1954 if (retval == REDIS_OK) 1955 sentinelEvent(REDIS_NOTICE,"+fix-slave-config",ri,"%@"); 1956 } 1957 } 1958 1959 /* Detect if the slave that is in the process of being reconfigured 1960 * changed state. */ 1961 if ((ri->flags & SRI_SLAVE) && role == SRI_SLAVE && 1962 (ri->flags & (SRI_RECONF_SENT|SRI_RECONF_INPROG))) 1963 { 1964 /* SRI_RECONF_SENT -> SRI_RECONF_INPROG. */ 1965 if ((ri->flags & SRI_RECONF_SENT) && 1966 ri->slave_master_host && 1967 strcmp(ri->slave_master_host, 1968 ri->master->promoted_slave->addr->ip) == 0 && 1969 ri->slave_master_port == ri->master->promoted_slave->addr->port) 1970 { 1971 ri->flags &= ~SRI_RECONF_SENT; 1972 ri->flags |= SRI_RECONF_INPROG; 1973 sentinelEvent(REDIS_NOTICE,"+slave-reconf-inprog",ri,"%@"); 1974 } 1975 1976 /* SRI_RECONF_INPROG -> SRI_RECONF_DONE */ 1977 if ((ri->flags & SRI_RECONF_INPROG) && 1978 ri->slave_master_link_status == SENTINEL_MASTER_LINK_STATUS_UP) 1979 { 1980 ri->flags &= ~SRI_RECONF_INPROG; 1981 ri->flags |= SRI_RECONF_DONE; 1982 sentinelEvent(REDIS_NOTICE,"+slave-reconf-done",ri,"%@"); 1983 } 1984 } 1985 } 1986 1987 void sentinelInfoReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 1988 sentinelRedisInstance *ri = c->data; 1989 redisReply *r; 1990 1991 if (ri) ri->pending_commands--; 1992 if (!reply || !ri) return; 1993 r = reply; 1994 1995 if (r->type == REDIS_REPLY_STRING) { 1996 sentinelRefreshInstanceInfo(ri,r->str); 1997 } 1998 } 1999 2000 /* Just discard the reply. We use this when we are not monitoring the return 2001 * value of the command but its effects directly. */ 2002 void sentinelDiscardReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 2003 sentinelRedisInstance *ri = c->data; 2004 2005 if (ri) ri->pending_commands--; 2006 } 2007 2008 void sentinelPingReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 2009 sentinelRedisInstance *ri = c->data; 2010 redisReply *r; 2011 2012 if (ri) ri->pending_commands--; 2013 if (!reply || !ri) return; 2014 r = reply; 2015 2016 if (r->type == REDIS_REPLY_STATUS || 2017 r->type == REDIS_REPLY_ERROR) { 2018 /* Update the "instance available" field only if this is an 2019 * acceptable reply. */ 2020 if (strncmp(r->str,"PONG",4) == 0 || 2021 strncmp(r->str,"LOADING",7) == 0 || 2022 strncmp(r->str,"MASTERDOWN",10) == 0) 2023 { 2024 ri->last_avail_time = mstime(); 2025 ri->last_ping_time = 0; /* Flag the pong as received. */ 2026 } else { 2027 /* Send a SCRIPT KILL command if the instance appears to be 2028 * down because of a busy script. */ 2029 if (strncmp(r->str,"BUSY",4) == 0 && 2030 (ri->flags & SRI_S_DOWN) && 2031 !(ri->flags & SRI_SCRIPT_KILL_SENT)) 2032 { 2033 if (redisAsyncCommand(ri->cc, 2034 sentinelDiscardReplyCallback, NULL, 2035 "SCRIPT KILL") == REDIS_OK) 2036 ri->pending_commands++; 2037 ri->flags |= SRI_SCRIPT_KILL_SENT; 2038 } 2039 } 2040 } 2041 ri->last_pong_time = mstime(); 2042 } 2043 2044 /* This is called when we get the reply about the PUBLISH command we send 2045 * to the master to advertise this sentinel. */ 2046 void sentinelPublishReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 2047 sentinelRedisInstance *ri = c->data; 2048 redisReply *r; 2049 2050 if (ri) ri->pending_commands--; 2051 if (!reply || !ri) return; 2052 r = reply; 2053 2054 /* Only update pub_time if we actually published our message. Otherwise 2055 * we'll retry against in 100 milliseconds. */ 2056 if (r->type != REDIS_REPLY_ERROR) 2057 ri->last_pub_time = mstime(); 2058 } 2059 2060 /* Process an hello message received via Pub/Sub in master or slave instance, 2061 * or sent directly to this sentinel via the (fake) PUBLISH command of Sentinel. 2062 * 2063 * If the master name specified in the message is not known, the message is 2064 * discareded. */ 2065 void sentinelProcessHelloMessage(char *hello, int hello_len) { 2066 /* Format is composed of 8 tokens: 2067 * 0=ip,1=port,2=runid,3=current_epoch,4=master_name, 2068 * 5=master_ip,6=master_port,7=master_config_epoch. */ 2069 int numtokens, port, removed, master_port; 2070 uint64_t current_epoch, master_config_epoch; 2071 char **token = sdssplitlen(hello, hello_len, ",", 1, &numtokens); 2072 sentinelRedisInstance *si, *master; 2073 2074 if (numtokens == 8) { 2075 /* Obtain a reference to the master this hello message is about */ 2076 master = sentinelGetMasterByName(token[4]); 2077 if (!master) goto cleanup; /* Unknown master, skip the message. */ 2078 2079 /* First, try to see if we already have this sentinel. */ 2080 port = atoi(token[1]); 2081 master_port = atoi(token[6]); 2082 si = getSentinelRedisInstanceByAddrAndRunID( 2083 master->sentinels,token[0],port,token[2]); 2084 current_epoch = strtoull(token[3],NULL,10); 2085 master_config_epoch = strtoull(token[7],NULL,10); 2086 2087 if (!si) { 2088 /* If not, remove all the sentinels that have the same runid 2089 * OR the same ip/port, because it's either a restart or a 2090 * network topology change. */ 2091 removed = removeMatchingSentinelsFromMaster(master,token[0],port, 2092 token[2]); 2093 if (removed) { 2094 sentinelEvent(REDIS_NOTICE,"-dup-sentinel",master, 2095 "%@ #duplicate of %s:%d or %s", 2096 token[0],port,token[2]); 2097 } 2098 2099 /* Add the new sentinel. */ 2100 si = createSentinelRedisInstance(NULL,SRI_SENTINEL, 2101 token[0],port,master->quorum,master); 2102 if (si) { 2103 sentinelEvent(REDIS_NOTICE,"+sentinel",si,"%@"); 2104 /* The runid is NULL after a new instance creation and 2105 * for Sentinels we don't have a later chance to fill it, 2106 * so do it now. */ 2107 si->runid = sdsnew(token[2]); 2108 sentinelFlushConfig(); 2109 } 2110 } 2111 2112 /* Update local current_epoch if received current_epoch is greater.*/ 2113 if (current_epoch > sentinel.current_epoch) { 2114 sentinel.current_epoch = current_epoch; 2115 sentinelFlushConfig(); 2116 sentinelEvent(REDIS_WARNING,"+new-epoch",master,"%llu", 2117 (unsigned long long) sentinel.current_epoch); 2118 } 2119 2120 /* Update master info if received configuration is newer. */ 2121 if (master->config_epoch < master_config_epoch) { 2122 master->config_epoch = master_config_epoch; 2123 if (master_port != master->addr->port || 2124 strcmp(master->addr->ip, token[5])) 2125 { 2126 sentinelAddr *old_addr; 2127 2128 sentinelEvent(REDIS_WARNING,"+switch-master", 2129 master,"%s %s %d %s %d", 2130 master->name, 2131 master->addr->ip, master->addr->port, 2132 token[5], master_port); 2133 2134 old_addr = dupSentinelAddr(master->addr); 2135 sentinelResetMasterAndChangeAddress(master, token[5], master_port); 2136 sentinelCallClientReconfScript(master, 2137 SENTINEL_OBSERVER,"start", 2138 old_addr,master->addr); 2139 releaseSentinelAddr(old_addr); 2140 } 2141 } 2142 2143 /* Update the state of the Sentinel. */ 2144 if (si) si->last_hello_time = mstime(); 2145 } 2146 2147 cleanup: 2148 sdsfreesplitres(token,numtokens); 2149 } 2150 2151 2152 /* This is our Pub/Sub callback for the Hello channel. It's useful in order 2153 * to discover other sentinels attached at the same master. */ 2154 void sentinelReceiveHelloMessages(redisAsyncContext *c, void *reply, void *privdata) { 2155 sentinelRedisInstance *ri = c->data, *master; 2156 redisReply *r; 2157 2158 if (!reply || !ri) return; 2159 r = reply; 2160 2161 master = (ri->flags & SRI_MASTER) ? ri : ri->master; 2162 2163 /* Update the last activity in the pubsub channel. Note that since we 2164 * receive our messages as well this timestamp can be used to detect 2165 * if the link is probably disconnected even if it seems otherwise. */ 2166 ri->pc_last_activity = mstime(); 2167 2168 /* Sanity check in the reply we expect, so that the code that follows 2169 * can avoid to check for details. */ 2170 if (r->type != REDIS_REPLY_ARRAY || 2171 r->elements != 3 || 2172 r->element[0]->type != REDIS_REPLY_STRING || 2173 r->element[1]->type != REDIS_REPLY_STRING || 2174 r->element[2]->type != REDIS_REPLY_STRING || 2175 strcmp(r->element[0]->str,"message") != 0) return; 2176 2177 /* We are not interested in meeting ourselves */ 2178 if (strstr(r->element[2]->str,server.runid) != NULL) return; 2179 2180 sentinelProcessHelloMessage(r->element[2]->str, r->element[2]->len); 2181 } 2182 2183 /* Send an "Hello" message via Pub/Sub to the specified 'ri' Redis 2184 * instance in order to broadcast the current configuraiton for this 2185 * master, and to advertise the existence of this Sentinel at the same time. 2186 * 2187 * The message has the following format: 2188 * 2189 * sentinel_ip,sentinel_port,sentinel_runid,current_epoch, 2190 * master_name,master_ip,master_port,master_config_epoch. 2191 * 2192 * Returns REDIS_OK if the PUBLISH was queued correctly, otherwise 2193 * REDIS_ERR is returned. */ 2194 int sentinelSendHello(sentinelRedisInstance *ri) { 2195 char ip[REDIS_IP_STR_LEN]; 2196 char payload[REDIS_IP_STR_LEN+1024]; 2197 int retval; 2198 sentinelRedisInstance *master = (ri->flags & SRI_MASTER) ? ri : ri->master; 2199 sentinelAddr *master_addr = sentinelGetCurrentMasterAddress(master); 2200 2201 /* Try to obtain our own IP address. */ 2202 if (anetSockName(ri->cc->c.fd,ip,sizeof(ip),NULL) == -1) return REDIS_ERR; 2203 if (ri->flags & SRI_DISCONNECTED) return REDIS_ERR; 2204 2205 /* Format and send the Hello message. */ 2206 snprintf(payload,sizeof(payload), 2207 "%s,%d,%s,%llu," /* Info about this sentinel. */ 2208 "%s,%s,%d,%llu", /* Info about current master. */ 2209 ip, server.port, server.runid, 2210 (unsigned long long) sentinel.current_epoch, 2211 /* --- */ 2212 master->name,master_addr->ip,master_addr->port, 2213 (unsigned long long) master->config_epoch); 2214 retval = redisAsyncCommand(ri->cc, 2215 sentinelPublishReplyCallback, NULL, "PUBLISH %s %s", 2216 SENTINEL_HELLO_CHANNEL,payload); 2217 if (retval != REDIS_OK) return REDIS_ERR; 2218 ri->pending_commands++; 2219 return REDIS_OK; 2220 } 2221 2222 /* Send a PING to the specified instance and refresh the last_ping_time 2223 * if it is zero (that is, if we received a pong for the previous ping). 2224 * 2225 * On error zero is returned, and we can't consider the PING command 2226 * queued in the connection. */ 2227 int sentinelSendPing(sentinelRedisInstance *ri) { 2228 int retval = redisAsyncCommand(ri->cc, 2229 sentinelPingReplyCallback, NULL, "PING"); 2230 if (retval == REDIS_OK) { 2231 ri->pending_commands++; 2232 /* We update the ping time only if we received the pong for 2233 * the previous ping, otherwise we are technically waiting 2234 * since the first ping that did not received a reply. */ 2235 if (ri->last_ping_time == 0) ri->last_ping_time = mstime(); 2236 return 1; 2237 } else { 2238 return 0; 2239 } 2240 } 2241 2242 /* Send periodic PING, INFO, and PUBLISH to the Hello channel to 2243 * the specified master or slave instance. */ 2244 void sentinelSendPeriodicCommands(sentinelRedisInstance *ri) { 2245 mstime_t now = mstime(); 2246 mstime_t info_period, ping_period; 2247 int retval; 2248 2249 /* Return ASAP if we have already a PING or INFO already pending, or 2250 * in the case the instance is not properly connected. */ 2251 if (ri->flags & SRI_DISCONNECTED) return; 2252 2253 /* For INFO, PING, PUBLISH that are not critical commands to send we 2254 * also have a limit of SENTINEL_MAX_PENDING_COMMANDS. We don't 2255 * want to use a lot of memory just because a link is not working 2256 * properly (note that anyway there is a redundant protection about this, 2257 * that is, the link will be disconnected and reconnected if a long 2258 * timeout condition is detected. */ 2259 if (ri->pending_commands >= SENTINEL_MAX_PENDING_COMMANDS) return; 2260 2261 /* If this is a slave of a master in O_DOWN condition we start sending 2262 * it INFO every second, instead of the usual SENTINEL_INFO_PERIOD 2263 * period. In this state we want to closely monitor slaves in case they 2264 * are turned into masters by another Sentinel, or by the sysadmin. */ 2265 if ((ri->flags & SRI_SLAVE) && 2266 (ri->master->flags & (SRI_O_DOWN|SRI_FAILOVER_IN_PROGRESS))) { 2267 info_period = 1000; 2268 } else { 2269 info_period = SENTINEL_INFO_PERIOD; 2270 } 2271 2272 /* We ping instances every time the last received pong is older than 2273 * the configured 'down-after-milliseconds' time, but every second 2274 * anyway if 'down-after-milliseconds' is greater than 1 second. */ 2275 ping_period = ri->down_after_period; 2276 if (ping_period > SENTINEL_PING_PERIOD) ping_period = SENTINEL_PING_PERIOD; 2277 2278 if ((ri->flags & SRI_SENTINEL) == 0 && 2279 (ri->info_refresh == 0 || 2280 (now - ri->info_refresh) > info_period)) 2281 { 2282 /* Send INFO to masters and slaves, not sentinels. */ 2283 retval = redisAsyncCommand(ri->cc, 2284 sentinelInfoReplyCallback, NULL, "INFO"); 2285 if (retval == REDIS_OK) ri->pending_commands++; 2286 } else if ((now - ri->last_pong_time) > ping_period) { 2287 /* Send PING to all the three kinds of instances. */ 2288 sentinelSendPing(ri); 2289 } else if ((now - ri->last_pub_time) > SENTINEL_PUBLISH_PERIOD) { 2290 /* PUBLISH hello messages to all the three kinds of instances. */ 2291 sentinelSendHello(ri); 2292 } 2293 } 2294 2295 /* =========================== SENTINEL command ============================= */ 2296 2297 const char *sentinelFailoverStateStr(int state) { 2298 switch(state) { 2299 case SENTINEL_FAILOVER_STATE_NONE: return "none"; 2300 case SENTINEL_FAILOVER_STATE_WAIT_START: return "wait_start"; 2301 case SENTINEL_FAILOVER_STATE_SELECT_SLAVE: return "select_slave"; 2302 case SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE: return "send_slaveof_noone"; 2303 case SENTINEL_FAILOVER_STATE_WAIT_PROMOTION: return "wait_promotion"; 2304 case SENTINEL_FAILOVER_STATE_RECONF_SLAVES: return "reconf_slaves"; 2305 case SENTINEL_FAILOVER_STATE_UPDATE_CONFIG: return "update_config"; 2306 default: return "unknown"; 2307 } 2308 } 2309 2310 /* Redis instance to Redis protocol representation. */ 2311 void addReplySentinelRedisInstance(redisClient *c, sentinelRedisInstance *ri) { 2312 char *flags = sdsempty(); 2313 void *mbl; 2314 int fields = 0; 2315 2316 mbl = addDeferredMultiBulkLength(c); 2317 2318 addReplyBulkCString(c,"name"); 2319 addReplyBulkCString(c,ri->name); 2320 fields++; 2321 2322 addReplyBulkCString(c,"ip"); 2323 addReplyBulkCString(c,ri->addr->ip); 2324 fields++; 2325 2326 addReplyBulkCString(c,"port"); 2327 addReplyBulkLongLong(c,ri->addr->port); 2328 fields++; 2329 2330 addReplyBulkCString(c,"runid"); 2331 addReplyBulkCString(c,ri->runid ? ri->runid : ""); 2332 fields++; 2333 2334 addReplyBulkCString(c,"flags"); 2335 if (ri->flags & SRI_S_DOWN) flags = sdscat(flags,"s_down,"); 2336 if (ri->flags & SRI_O_DOWN) flags = sdscat(flags,"o_down,"); 2337 if (ri->flags & SRI_MASTER) flags = sdscat(flags,"master,"); 2338 if (ri->flags & SRI_SLAVE) flags = sdscat(flags,"slave,"); 2339 if (ri->flags & SRI_SENTINEL) flags = sdscat(flags,"sentinel,"); 2340 if (ri->flags & SRI_DISCONNECTED) flags = sdscat(flags,"disconnected,"); 2341 if (ri->flags & SRI_MASTER_DOWN) flags = sdscat(flags,"master_down,"); 2342 if (ri->flags & SRI_FAILOVER_IN_PROGRESS) 2343 flags = sdscat(flags,"failover_in_progress,"); 2344 if (ri->flags & SRI_PROMOTED) flags = sdscat(flags,"promoted,"); 2345 if (ri->flags & SRI_RECONF_SENT) flags = sdscat(flags,"reconf_sent,"); 2346 if (ri->flags & SRI_RECONF_INPROG) flags = sdscat(flags,"reconf_inprog,"); 2347 if (ri->flags & SRI_RECONF_DONE) flags = sdscat(flags,"reconf_done,"); 2348 2349 if (sdslen(flags) != 0) sdsrange(flags,0,-2); /* remove last "," */ 2350 addReplyBulkCString(c,flags); 2351 sdsfree(flags); 2352 fields++; 2353 2354 addReplyBulkCString(c,"pending-commands"); 2355 addReplyBulkLongLong(c,ri->pending_commands); 2356 fields++; 2357 2358 if (ri->flags & SRI_FAILOVER_IN_PROGRESS) { 2359 addReplyBulkCString(c,"failover-state"); 2360 addReplyBulkCString(c,(char*)sentinelFailoverStateStr(ri->failover_state)); 2361 fields++; 2362 } 2363 2364 addReplyBulkCString(c,"last-ping-sent"); 2365 addReplyBulkLongLong(c, 2366 ri->last_ping_time ? (mstime() - ri->last_ping_time) : 0); 2367 fields++; 2368 2369 addReplyBulkCString(c,"last-ok-ping-reply"); 2370 addReplyBulkLongLong(c,mstime() - ri->last_avail_time); 2371 fields++; 2372 2373 addReplyBulkCString(c,"last-ping-reply"); 2374 addReplyBulkLongLong(c,mstime() - ri->last_pong_time); 2375 fields++; 2376 2377 if (ri->flags & SRI_S_DOWN) { 2378 addReplyBulkCString(c,"s-down-time"); 2379 addReplyBulkLongLong(c,mstime()-ri->s_down_since_time); 2380 fields++; 2381 } 2382 2383 if (ri->flags & SRI_O_DOWN) { 2384 addReplyBulkCString(c,"o-down-time"); 2385 addReplyBulkLongLong(c,mstime()-ri->o_down_since_time); 2386 fields++; 2387 } 2388 2389 addReplyBulkCString(c,"down-after-milliseconds"); 2390 addReplyBulkLongLong(c,ri->down_after_period); 2391 fields++; 2392 2393 /* Masters and Slaves */ 2394 if (ri->flags & (SRI_MASTER|SRI_SLAVE)) { 2395 addReplyBulkCString(c,"info-refresh"); 2396 addReplyBulkLongLong(c,mstime() - ri->info_refresh); 2397 fields++; 2398 2399 addReplyBulkCString(c,"role-reported"); 2400 addReplyBulkCString(c, (ri->role_reported == SRI_MASTER) ? "master" : 2401 "slave"); 2402 fields++; 2403 2404 addReplyBulkCString(c,"role-reported-time"); 2405 addReplyBulkLongLong(c,mstime() - ri->role_reported_time); 2406 fields++; 2407 } 2408 2409 /* Only masters */ 2410 if (ri->flags & SRI_MASTER) { 2411 addReplyBulkCString(c,"config-epoch"); 2412 addReplyBulkLongLong(c,ri->config_epoch); 2413 fields++; 2414 2415 addReplyBulkCString(c,"num-slaves"); 2416 addReplyBulkLongLong(c,dictSize(ri->slaves)); 2417 fields++; 2418 2419 addReplyBulkCString(c,"num-other-sentinels"); 2420 addReplyBulkLongLong(c,dictSize(ri->sentinels)); 2421 fields++; 2422 2423 addReplyBulkCString(c,"quorum"); 2424 addReplyBulkLongLong(c,ri->quorum); 2425 fields++; 2426 2427 addReplyBulkCString(c,"failover-timeout"); 2428 addReplyBulkLongLong(c,ri->failover_timeout); 2429 fields++; 2430 2431 addReplyBulkCString(c,"parallel-syncs"); 2432 addReplyBulkLongLong(c,ri->parallel_syncs); 2433 fields++; 2434 2435 if (ri->notification_script) { 2436 addReplyBulkCString(c,"notification-script"); 2437 addReplyBulkCString(c,ri->notification_script); 2438 fields++; 2439 } 2440 2441 if (ri->client_reconfig_script) { 2442 addReplyBulkCString(c,"client-reconfig-script"); 2443 addReplyBulkCString(c,ri->client_reconfig_script); 2444 fields++; 2445 } 2446 } 2447 2448 /* Only slaves */ 2449 if (ri->flags & SRI_SLAVE) { 2450 addReplyBulkCString(c,"master-link-down-time"); 2451 addReplyBulkLongLong(c,ri->master_link_down_time); 2452 fields++; 2453 2454 addReplyBulkCString(c,"master-link-status"); 2455 addReplyBulkCString(c, 2456 (ri->slave_master_link_status == SENTINEL_MASTER_LINK_STATUS_UP) ? 2457 "ok" : "err"); 2458 fields++; 2459 2460 addReplyBulkCString(c,"master-host"); 2461 addReplyBulkCString(c, 2462 ri->slave_master_host ? ri->slave_master_host : "?"); 2463 fields++; 2464 2465 addReplyBulkCString(c,"master-port"); 2466 addReplyBulkLongLong(c,ri->slave_master_port); 2467 fields++; 2468 2469 addReplyBulkCString(c,"slave-priority"); 2470 addReplyBulkLongLong(c,ri->slave_priority); 2471 fields++; 2472 2473 addReplyBulkCString(c,"slave-repl-offset"); 2474 addReplyBulkLongLong(c,ri->slave_repl_offset); 2475 fields++; 2476 } 2477 2478 /* Only sentinels */ 2479 if (ri->flags & SRI_SENTINEL) { 2480 addReplyBulkCString(c,"last-hello-message"); 2481 addReplyBulkLongLong(c,mstime() - ri->last_hello_time); 2482 fields++; 2483 2484 addReplyBulkCString(c,"voted-leader"); 2485 addReplyBulkCString(c,ri->leader ? ri->leader : "?"); 2486 fields++; 2487 2488 addReplyBulkCString(c,"voted-leader-epoch"); 2489 addReplyBulkLongLong(c,ri->leader_epoch); 2490 fields++; 2491 } 2492 2493 setDeferredMultiBulkLength(c,mbl,fields*2); 2494 } 2495 2496 /* Output a number of instances contained inside a dictionary as 2497 * Redis protocol. */ 2498 void addReplyDictOfRedisInstances(redisClient *c, dict *instances) { 2499 dictIterator *di; 2500 dictEntry *de; 2501 2502 di = dictGetIterator(instances); 2503 addReplyMultiBulkLen(c,dictSize(instances)); 2504 while((de = dictNext(di)) != NULL) { 2505 sentinelRedisInstance *ri = dictGetVal(de); 2506 2507 addReplySentinelRedisInstance(c,ri); 2508 } 2509 dictReleaseIterator(di); 2510 } 2511 2512 /* Lookup the named master into sentinel.masters. 2513 * If the master is not found reply to the client with an error and returns 2514 * NULL. */ 2515 sentinelRedisInstance *sentinelGetMasterByNameOrReplyError(redisClient *c, 2516 robj *name) 2517 { 2518 sentinelRedisInstance *ri; 2519 2520 ri = dictFetchValue(sentinel.masters,c->argv[2]->ptr); 2521 if (!ri) { 2522 addReplyError(c,"No such master with that name"); 2523 return NULL; 2524 } 2525 return ri; 2526 } 2527 2528 void sentinelCommand(redisClient *c) { 2529 if (!strcasecmp(c->argv[1]->ptr,"masters")) { 2530 /* SENTINEL MASTERS */ 2531 if (c->argc != 2) goto numargserr; 2532 addReplyDictOfRedisInstances(c,sentinel.masters); 2533 } else if (!strcasecmp(c->argv[1]->ptr,"master")) { 2534 /* SENTINEL MASTER <name> */ 2535 sentinelRedisInstance *ri; 2536 2537 if (c->argc != 3) goto numargserr; 2538 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) 2539 == NULL) return; 2540 addReplySentinelRedisInstance(c,ri); 2541 } else if (!strcasecmp(c->argv[1]->ptr,"slaves")) { 2542 /* SENTINEL SLAVES <master-name> */ 2543 sentinelRedisInstance *ri; 2544 2545 if (c->argc != 3) goto numargserr; 2546 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) == NULL) 2547 return; 2548 addReplyDictOfRedisInstances(c,ri->slaves); 2549 } else if (!strcasecmp(c->argv[1]->ptr,"sentinels")) { 2550 /* SENTINEL SENTINELS <master-name> */ 2551 sentinelRedisInstance *ri; 2552 2553 if (c->argc != 3) goto numargserr; 2554 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) == NULL) 2555 return; 2556 addReplyDictOfRedisInstances(c,ri->sentinels); 2557 } else if (!strcasecmp(c->argv[1]->ptr,"is-master-down-by-addr")) { 2558 /* SENTINEL IS-MASTER-DOWN-BY-ADDR <ip> <port> <current-epoch> <runid>*/ 2559 sentinelRedisInstance *ri; 2560 long long req_epoch; 2561 uint64_t leader_epoch = 0; 2562 char *leader = NULL; 2563 long port; 2564 int isdown = 0; 2565 2566 if (c->argc != 6) goto numargserr; 2567 if (getLongFromObjectOrReply(c,c->argv[3],&port,NULL) != REDIS_OK || 2568 getLongLongFromObjectOrReply(c,c->argv[4],&req_epoch,NULL) 2569 != REDIS_OK) 2570 return; 2571 ri = getSentinelRedisInstanceByAddrAndRunID(sentinel.masters, 2572 c->argv[2]->ptr,port,NULL); 2573 2574 /* It exists? Is actually a master? Is subjectively down? It's down. 2575 * Note: if we are in tilt mode we always reply with "0". */ 2576 if (!sentinel.tilt && ri && (ri->flags & SRI_S_DOWN) && 2577 (ri->flags & SRI_MASTER)) 2578 isdown = 1; 2579 2580 /* Vote for the master (or fetch the previous vote) if the request 2581 * includes a runid, otherwise the sender is not seeking for a vote. */ 2582 if (ri && ri->flags & SRI_MASTER && strcasecmp(c->argv[5]->ptr,"*")) { 2583 leader = sentinelVoteLeader(ri,(uint64_t)req_epoch, 2584 c->argv[5]->ptr, 2585 &leader_epoch); 2586 } 2587 2588 /* Reply with a three-elements multi-bulk reply: 2589 * down state, leader, vote epoch. */ 2590 addReplyMultiBulkLen(c,3); 2591 addReply(c, isdown ? shared.cone : shared.czero); 2592 addReplyBulkCString(c, leader ? leader : "*"); 2593 addReplyLongLong(c, (long long)leader_epoch); 2594 if (leader) sdsfree(leader); 2595 } else if (!strcasecmp(c->argv[1]->ptr,"reset")) { 2596 /* SENTINEL RESET <pattern> */ 2597 if (c->argc != 3) goto numargserr; 2598 addReplyLongLong(c,sentinelResetMastersByPattern(c->argv[2]->ptr,SENTINEL_GENERATE_EVENT)); 2599 } else if (!strcasecmp(c->argv[1]->ptr,"get-master-addr-by-name")) { 2600 /* SENTINEL GET-MASTER-ADDR-BY-NAME <master-name> */ 2601 sentinelRedisInstance *ri; 2602 2603 if (c->argc != 3) goto numargserr; 2604 ri = sentinelGetMasterByName(c->argv[2]->ptr); 2605 if (ri == NULL) { 2606 addReply(c,shared.nullmultibulk); 2607 } else { 2608 sentinelAddr *addr = sentinelGetCurrentMasterAddress(ri); 2609 2610 addReplyMultiBulkLen(c,2); 2611 addReplyBulkCString(c,addr->ip); 2612 addReplyBulkLongLong(c,addr->port); 2613 } 2614 } else if (!strcasecmp(c->argv[1]->ptr,"failover")) { 2615 /* SENTINEL FAILOVER <master-name> */ 2616 sentinelRedisInstance *ri; 2617 2618 if (c->argc != 3) goto numargserr; 2619 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) == NULL) 2620 return; 2621 if (ri->flags & SRI_FAILOVER_IN_PROGRESS) { 2622 addReplySds(c,sdsnew("-INPROG Failover already in progress\r\n")); 2623 return; 2624 } 2625 if (sentinelSelectSlave(ri) == NULL) { 2626 addReplySds(c,sdsnew("-NOGOODSLAVE No suitable slave to promote\r\n")); 2627 return; 2628 } 2629 redisLog(REDIS_WARNING,"Executing user requested FAILOVER of '%s'", 2630 ri->name); 2631 sentinelStartFailover(ri); 2632 ri->flags |= SRI_FORCE_FAILOVER; 2633 addReply(c,shared.ok); 2634 } else if (!strcasecmp(c->argv[1]->ptr,"pending-scripts")) { 2635 /* SENTINEL PENDING-SCRIPTS */ 2636 2637 if (c->argc != 2) goto numargserr; 2638 sentinelPendingScriptsCommand(c); 2639 } else if (!strcasecmp(c->argv[1]->ptr,"monitor")) { 2640 /* SENTINEL MONITOR <name> <ip> <port> <quorum> */ 2641 sentinelRedisInstance *ri; 2642 long quorum, port; 2643 char buf[32]; 2644 2645 if (c->argc != 6) goto numargserr; 2646 if (getLongFromObjectOrReply(c,c->argv[5],&quorum,"Invalid quorum") 2647 != REDIS_OK) return; 2648 if (getLongFromObjectOrReply(c,c->argv[4],&port,"Invalid port") 2649 != REDIS_OK) return; 2650 /* Make sure the IP field is actually a valid IP before passing it 2651 * to createSentinelRedisInstance(), otherwise we may trigger a 2652 * DNS lookup at runtime. */ 2653 if (anetResolveIP(NULL,c->argv[3]->ptr,buf,sizeof(buf)) == ANET_ERR) { 2654 addReplyError(c,"Invalid IP address specified"); 2655 return; 2656 } 2657 2658 /* Parameters are valid. Try to create the master instance. */ 2659 ri = createSentinelRedisInstance(c->argv[2]->ptr,SRI_MASTER, 2660 c->argv[3]->ptr,port,quorum,NULL); 2661 if (ri == NULL) { 2662 switch(errno) { 2663 case EBUSY: 2664 addReplyError(c,"Duplicated master name"); 2665 break; 2666 case EINVAL: 2667 addReplyError(c,"Invalid port number"); 2668 break; 2669 default: 2670 addReplyError(c,"Unspecified error adding the instance"); 2671 break; 2672 } 2673 } else { 2674 sentinelFlushConfig(); 2675 sentinelEvent(REDIS_WARNING,"+monitor",ri,"%@ quorum %d",ri->quorum); 2676 addReply(c,shared.ok); 2677 } 2678 } else if (!strcasecmp(c->argv[1]->ptr,"remove")) { 2679 /* SENTINEL REMOVE <name> */ 2680 sentinelRedisInstance *ri; 2681 2682 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) 2683 == NULL) return; 2684 sentinelEvent(REDIS_WARNING,"-monitor",ri,"%@"); 2685 dictDelete(sentinel.masters,c->argv[2]->ptr); 2686 sentinelFlushConfig(); 2687 addReply(c,shared.ok); 2688 } else if (!strcasecmp(c->argv[1]->ptr,"set")) { 2689 if (c->argc < 3 || c->argc % 2 == 0) goto numargserr; 2690 sentinelSetCommand(c); 2691 } else { 2692 addReplyErrorFormat(c,"Unknown sentinel subcommand '%s'", 2693 (char*)c->argv[1]->ptr); 2694 } 2695 return; 2696 2697 numargserr: 2698 addReplyErrorFormat(c,"Wrong number of arguments for 'sentinel %s'", 2699 (char*)c->argv[1]->ptr); 2700 } 2701 2702 /* SENTINEL INFO [section] */ 2703 void sentinelInfoCommand(redisClient *c) { 2704 char *section = c->argc == 2 ? c->argv[1]->ptr : "default"; 2705 sds info = sdsempty(); 2706 int defsections = !strcasecmp(section,"default"); 2707 int sections = 0; 2708 2709 if (c->argc > 2) { 2710 addReply(c,shared.syntaxerr); 2711 return; 2712 } 2713 2714 if (!strcasecmp(section,"server") || defsections) { 2715 if (sections++) info = sdscat(info,"\r\n"); 2716 sds serversection = genRedisInfoString("server"); 2717 info = sdscatlen(info,serversection,sdslen(serversection)); 2718 sdsfree(serversection); 2719 } 2720 2721 if (!strcasecmp(section,"sentinel") || defsections) { 2722 dictIterator *di; 2723 dictEntry *de; 2724 int master_id = 0; 2725 2726 if (sections++) info = sdscat(info,"\r\n"); 2727 info = sdscatprintf(info, 2728 "# Sentinel\r\n" 2729 "sentinel_masters:%lu\r\n" 2730 "sentinel_tilt:%d\r\n" 2731 "sentinel_running_scripts:%d\r\n" 2732 "sentinel_scripts_queue_length:%ld\r\n", 2733 dictSize(sentinel.masters), 2734 sentinel.tilt, 2735 sentinel.running_scripts, 2736 listLength(sentinel.scripts_queue)); 2737 2738 di = dictGetIterator(sentinel.masters); 2739 while((de = dictNext(di)) != NULL) { 2740 sentinelRedisInstance *ri = dictGetVal(de); 2741 char *status = "ok"; 2742 2743 if (ri->flags & SRI_O_DOWN) status = "odown"; 2744 else if (ri->flags & SRI_S_DOWN) status = "sdown"; 2745 info = sdscatprintf(info, 2746 "master%d:name=%s,status=%s,address=%s:%d," 2747 "slaves=%lu,sentinels=%lu\r\n", 2748 master_id++, ri->name, status, 2749 ri->addr->ip, ri->addr->port, 2750 dictSize(ri->slaves), 2751 dictSize(ri->sentinels)+1); 2752 } 2753 dictReleaseIterator(di); 2754 } 2755 2756 addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n", 2757 (unsigned long)sdslen(info))); 2758 addReplySds(c,info); 2759 addReply(c,shared.crlf); 2760 } 2761 2762 /* SENTINEL SET <mastername> [<option> <value> ...] */ 2763 void sentinelSetCommand(redisClient *c) { 2764 sentinelRedisInstance *ri; 2765 int j, changes = 0; 2766 char *option, *value; 2767 2768 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) 2769 == NULL) return; 2770 2771 /* Process option - value pairs. */ 2772 for (j = 3; j < c->argc; j += 2) { 2773 option = c->argv[j]->ptr; 2774 value = c->argv[j+1]->ptr; 2775 robj *o = c->argv[j+1]; 2776 long long ll; 2777 2778 if (!strcasecmp(option,"down-after-milliseconds")) { 2779 /* down-after-millisecodns <milliseconds> */ 2780 if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll <= 0) 2781 goto badfmt; 2782 ri->down_after_period = ll; 2783 sentinelPropagateDownAfterPeriod(ri); 2784 changes++; 2785 } else if (!strcasecmp(option,"failover-timeout")) { 2786 /* failover-timeout <milliseconds> */ 2787 if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll <= 0) 2788 goto badfmt; 2789 ri->failover_timeout = ll; 2790 changes++; 2791 } else if (!strcasecmp(option,"parallel-syncs")) { 2792 /* parallel-syncs <milliseconds> */ 2793 if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll <= 0) 2794 goto badfmt; 2795 ri->parallel_syncs = ll; 2796 changes++; 2797 } else if (!strcasecmp(option,"notification-script")) { 2798 /* notification-script <path> */ 2799 if (strlen(value) && access(value,X_OK) == -1) { 2800 addReplyError(c, 2801 "Notification script seems non existing or non executable"); 2802 if (changes) sentinelFlushConfig(); 2803 return; 2804 } 2805 sdsfree(ri->notification_script); 2806 ri->notification_script = strlen(value) ? sdsnew(value) : NULL; 2807 changes++; 2808 } else if (!strcasecmp(option,"client-reconfig-script")) { 2809 /* client-reconfig-script <path> */ 2810 if (strlen(value) && access(value,X_OK) == -1) { 2811 addReplyError(c, 2812 "Client reconfiguration script seems non existing or " 2813 "non executable"); 2814 if (changes) sentinelFlushConfig(); 2815 return; 2816 } 2817 sdsfree(ri->client_reconfig_script); 2818 ri->client_reconfig_script = strlen(value) ? sdsnew(value) : NULL; 2819 changes++; 2820 } else if (!strcasecmp(option,"auth-pass")) { 2821 /* auth-pass <password> */ 2822 sdsfree(ri->auth_pass); 2823 ri->auth_pass = strlen(value) ? sdsnew(value) : NULL; 2824 changes++; 2825 } else if (!strcasecmp(option,"quorum")) { 2826 /* quorum <count> */ 2827 if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll <= 0) 2828 goto badfmt; 2829 ri->quorum = ll; 2830 changes++; 2831 } else { 2832 addReplyErrorFormat(c,"Unknown option '%s' for SENTINEL SET", 2833 option); 2834 if (changes) sentinelFlushConfig(); 2835 return; 2836 } 2837 sentinelEvent(REDIS_WARNING,"+set",ri,"%@ %s %s",option,value); 2838 } 2839 2840 if (changes) sentinelFlushConfig(); 2841 addReply(c,shared.ok); 2842 return; 2843 2844 badfmt: /* Bad format errors */ 2845 if (changes) sentinelFlushConfig(); 2846 addReplyErrorFormat(c,"Invalid argument '%s' for SENTINEL SET '%s'", 2847 value, option); 2848 } 2849 2850 /* Our fake PUBLISH command: it is actually useful only to receive hello messages 2851 * from the other sentinel instances, and publishing to a channel other than 2852 * SENTINEL_HELLO_CHANNEL is forbidden. 2853 * 2854 * Because we have a Sentinel PUBLISH, the code to send hello messages is the same 2855 * for all the three kind of instances: masters, slaves, sentinels. */ 2856 void sentinelPublishCommand(redisClient *c) { 2857 if (strcmp(c->argv[1]->ptr,SENTINEL_HELLO_CHANNEL)) { 2858 addReplyError(c, "Only HELLO messages are accepted by Sentinel instances."); 2859 return; 2860 } 2861 sentinelProcessHelloMessage(c->argv[2]->ptr,sdslen(c->argv[2]->ptr)); 2862 addReplyLongLong(c,1); 2863 } 2864 2865 /* ===================== SENTINEL availability checks ======================= */ 2866 2867 /* Is this instance down from our point of view? */ 2868 void sentinelCheckSubjectivelyDown(sentinelRedisInstance *ri) { 2869 mstime_t elapsed = 0; 2870 2871 if (ri->last_ping_time) 2872 elapsed = mstime() - ri->last_ping_time; 2873 2874 /* Check if we are in need for a reconnection of one of the 2875 * links, because we are detecting low activity. 2876 * 2877 * 1) Check if the command link seems connected, was connected not less 2878 * than SENTINEL_MIN_LINK_RECONNECT_PERIOD, but still we have a 2879 * pending ping for more than half the timeout. */ 2880 if (ri->cc && 2881 (mstime() - ri->cc_conn_time) > SENTINEL_MIN_LINK_RECONNECT_PERIOD && 2882 ri->last_ping_time != 0 && /* Ther is a pending ping... */ 2883 /* The pending ping is delayed, and we did not received 2884 * error replies as well. */ 2885 (mstime() - ri->last_ping_time) > (ri->down_after_period/2) && 2886 (mstime() - ri->last_pong_time) > (ri->down_after_period/2)) 2887 { 2888 sentinelKillLink(ri,ri->cc); 2889 } 2890 2891 /* 2) Check if the pubsub link seems connected, was connected not less 2892 * than SENTINEL_MIN_LINK_RECONNECT_PERIOD, but still we have no 2893 * activity in the Pub/Sub channel for more than 2894 * SENTINEL_PUBLISH_PERIOD * 3. 2895 */ 2896 if (ri->pc && 2897 (mstime() - ri->pc_conn_time) > SENTINEL_MIN_LINK_RECONNECT_PERIOD && 2898 (mstime() - ri->pc_last_activity) > (SENTINEL_PUBLISH_PERIOD*3)) 2899 { 2900 sentinelKillLink(ri,ri->pc); 2901 } 2902 2903 /* Update the SDOWN flag. We believe the instance is SDOWN if: 2904 * 2905 * 1) It is not replying. 2906 * 2) We believe it is a master, it reports to be a slave for enough time 2907 * to meet the down_after_period, plus enough time to get two times 2908 * INFO report from the instance. */ 2909 if (elapsed > ri->down_after_period || 2910 (ri->flags & SRI_MASTER && 2911 ri->role_reported == SRI_SLAVE && 2912 mstime() - ri->role_reported_time > 2913 (ri->down_after_period+SENTINEL_INFO_PERIOD*2))) 2914 { 2915 /* Is subjectively down */ 2916 if ((ri->flags & SRI_S_DOWN) == 0) { 2917 sentinelEvent(REDIS_WARNING,"+sdown",ri,"%@"); 2918 ri->s_down_since_time = mstime(); 2919 ri->flags |= SRI_S_DOWN; 2920 } 2921 } else { 2922 /* Is subjectively up */ 2923 if (ri->flags & SRI_S_DOWN) { 2924 sentinelEvent(REDIS_WARNING,"-sdown",ri,"%@"); 2925 ri->flags &= ~(SRI_S_DOWN|SRI_SCRIPT_KILL_SENT); 2926 } 2927 } 2928 } 2929 2930 /* Is this instance down according to the configured quorum? 2931 * 2932 * Note that ODOWN is a weak quorum, it only means that enough Sentinels 2933 * reported in a given time range that the instance was not reachable. 2934 * However messages can be delayed so there are no strong guarantees about 2935 * N instances agreeing at the same time about the down state. */ 2936 void sentinelCheckObjectivelyDown(sentinelRedisInstance *master) { 2937 dictIterator *di; 2938 dictEntry *de; 2939 int quorum = 0, odown = 0; 2940 2941 if (master->flags & SRI_S_DOWN) { 2942 /* Is down for enough sentinels? */ 2943 quorum = 1; /* the current sentinel. */ 2944 /* Count all the other sentinels. */ 2945 di = dictGetIterator(master->sentinels); 2946 while((de = dictNext(di)) != NULL) { 2947 sentinelRedisInstance *ri = dictGetVal(de); 2948 2949 if (ri->flags & SRI_MASTER_DOWN) quorum++; 2950 } 2951 dictReleaseIterator(di); 2952 if (quorum >= master->quorum) odown = 1; 2953 } 2954 2955 /* Set the flag accordingly to the outcome. */ 2956 if (odown) { 2957 if ((master->flags & SRI_O_DOWN) == 0) { 2958 sentinelEvent(REDIS_WARNING,"+odown",master,"%@ #quorum %d/%d", 2959 quorum, master->quorum); 2960 master->flags |= SRI_O_DOWN; 2961 master->o_down_since_time = mstime(); 2962 } 2963 } else { 2964 if (master->flags & SRI_O_DOWN) { 2965 sentinelEvent(REDIS_WARNING,"-odown",master,"%@"); 2966 master->flags &= ~SRI_O_DOWN; 2967 } 2968 } 2969 } 2970 2971 /* Receive the SENTINEL is-master-down-by-addr reply, see the 2972 * sentinelAskMasterStateToOtherSentinels() function for more information. */ 2973 void sentinelReceiveIsMasterDownReply(redisAsyncContext *c, void *reply, void *privdata) { 2974 sentinelRedisInstance *ri = c->data; 2975 redisReply *r; 2976 2977 if (ri) ri->pending_commands--; 2978 if (!reply || !ri) return; 2979 r = reply; 2980 2981 /* Ignore every error or unexpected reply. 2982 * Note that if the command returns an error for any reason we'll 2983 * end clearing the SRI_MASTER_DOWN flag for timeout anyway. */ 2984 if (r->type == REDIS_REPLY_ARRAY && r->elements == 3 && 2985 r->element[0]->type == REDIS_REPLY_INTEGER && 2986 r->element[1]->type == REDIS_REPLY_STRING && 2987 r->element[2]->type == REDIS_REPLY_INTEGER) 2988 { 2989 ri->last_master_down_reply_time = mstime(); 2990 if (r->element[0]->integer == 1) { 2991 ri->flags |= SRI_MASTER_DOWN; 2992 } else { 2993 ri->flags &= ~SRI_MASTER_DOWN; 2994 } 2995 if (strcmp(r->element[1]->str,"*")) { 2996 /* If the runid in the reply is not "*" the Sentinel actually 2997 * replied with a vote. */ 2998 sdsfree(ri->leader); 2999 if (ri->leader_epoch != r->element[2]->integer) 3000 redisLog(REDIS_WARNING, 3001 "%s voted for %s %llu", ri->name, 3002 r->element[1]->str, 3003 (unsigned long long) r->element[2]->integer); 3004 ri->leader = sdsnew(r->element[1]->str); 3005 ri->leader_epoch = r->element[2]->integer; 3006 } 3007 } 3008 } 3009 3010 /* If we think the master is down, we start sending 3011 * SENTINEL IS-MASTER-DOWN-BY-ADDR requests to other sentinels 3012 * in order to get the replies that allow to reach the quorum 3013 * needed to mark the master in ODOWN state and trigger a failover. */ 3014 #define SENTINEL_ASK_FORCED (1<<0) 3015 void sentinelAskMasterStateToOtherSentinels(sentinelRedisInstance *master, int flags) { 3016 dictIterator *di; 3017 dictEntry *de; 3018 3019 di = dictGetIterator(master->sentinels); 3020 while((de = dictNext(di)) != NULL) { 3021 sentinelRedisInstance *ri = dictGetVal(de); 3022 mstime_t elapsed = mstime() - ri->last_master_down_reply_time; 3023 char port[32]; 3024 int retval; 3025 3026 /* If the master state from other sentinel is too old, we clear it. */ 3027 if (elapsed > SENTINEL_ASK_PERIOD*5) { 3028 ri->flags &= ~SRI_MASTER_DOWN; 3029 sdsfree(ri->leader); 3030 ri->leader = NULL; 3031 } 3032 3033 /* Only ask if master is down to other sentinels if: 3034 * 3035 * 1) We believe it is down, or there is a failover in progress. 3036 * 2) Sentinel is connected. 3037 * 3) We did not received the info within SENTINEL_ASK_PERIOD ms. */ 3038 if ((master->flags & SRI_S_DOWN) == 0) continue; 3039 if (ri->flags & SRI_DISCONNECTED) continue; 3040 if (!(flags & SENTINEL_ASK_FORCED) && 3041 mstime() - ri->last_master_down_reply_time < SENTINEL_ASK_PERIOD) 3042 continue; 3043 3044 /* Ask */ 3045 ll2string(port,sizeof(port),master->addr->port); 3046 retval = redisAsyncCommand(ri->cc, 3047 sentinelReceiveIsMasterDownReply, NULL, 3048 "SENTINEL is-master-down-by-addr %s %s %llu %s", 3049 master->addr->ip, port, 3050 sentinel.current_epoch, 3051 (master->failover_state > SENTINEL_FAILOVER_STATE_NONE) ? 3052 server.runid : "*"); 3053 if (retval == REDIS_OK) ri->pending_commands++; 3054 } 3055 dictReleaseIterator(di); 3056 } 3057 3058 /* =============================== FAILOVER ================================= */ 3059 3060 /* Vote for the sentinel with 'req_runid' or return the old vote if already 3061 * voted for the specifed 'req_epoch' or one greater. 3062 * 3063 * If a vote is not available returns NULL, otherwise return the Sentinel 3064 * runid and populate the leader_epoch with the epoch of the vote. */ 3065 char *sentinelVoteLeader(sentinelRedisInstance *master, uint64_t req_epoch, char *req_runid, uint64_t *leader_epoch) { 3066 if (req_epoch > sentinel.current_epoch) { 3067 sentinel.current_epoch = req_epoch; 3068 sentinelFlushConfig(); 3069 sentinelEvent(REDIS_WARNING,"+new-epoch",master,"%llu", 3070 (unsigned long long) sentinel.current_epoch); 3071 } 3072 3073 if (master->leader_epoch < req_epoch && sentinel.current_epoch <= req_epoch) 3074 { 3075 sdsfree(master->leader); 3076 master->leader = sdsnew(req_runid); 3077 master->leader_epoch = sentinel.current_epoch; 3078 sentinelFlushConfig(); 3079 sentinelEvent(REDIS_WARNING,"+vote-for-leader",master,"%s %llu", 3080 master->leader, (unsigned long long) master->leader_epoch); 3081 /* If we did not voted for ourselves, set the master failover start 3082 * time to now, in order to force a delay before we can start a 3083 * failover for the same master. */ 3084 if (strcasecmp(master->leader,server.runid)) 3085 master->failover_start_time = mstime()+rand()%SENTINEL_MAX_DESYNC; 3086 } 3087 3088 *leader_epoch = master->leader_epoch; 3089 return master->leader ? sdsnew(master->leader) : NULL; 3090 } 3091 3092 struct sentinelLeader { 3093 char *runid; 3094 unsigned long votes; 3095 }; 3096 3097 /* Helper function for sentinelGetLeader, increment the counter 3098 * relative to the specified runid. */ 3099 int sentinelLeaderIncr(dict *counters, char *runid) { 3100 dictEntry *de = dictFind(counters,runid); 3101 uint64_t oldval; 3102 3103 if (de) { 3104 oldval = dictGetUnsignedIntegerVal(de); 3105 dictSetUnsignedIntegerVal(de,oldval+1); 3106 return oldval+1; 3107 } else { 3108 de = dictAddRaw(counters,runid); 3109 redisAssert(de != NULL); 3110 dictSetUnsignedIntegerVal(de,1); 3111 return 1; 3112 } 3113 } 3114 3115 /* Scan all the Sentinels attached to this master to check if there 3116 * is a leader for the specified epoch. 3117 * 3118 * To be a leader for a given epoch, we should have the majorify of 3119 * the Sentinels we know that reported the same instance as 3120 * leader for the same epoch. */ 3121 char *sentinelGetLeader(sentinelRedisInstance *master, uint64_t epoch) { 3122 dict *counters; 3123 dictIterator *di; 3124 dictEntry *de; 3125 unsigned int voters = 0, voters_quorum; 3126 char *myvote; 3127 char *winner = NULL; 3128 uint64_t leader_epoch; 3129 uint64_t max_votes = 0; 3130 3131 redisAssert(master->flags & (SRI_O_DOWN|SRI_FAILOVER_IN_PROGRESS)); 3132 counters = dictCreate(&leaderVotesDictType,NULL); 3133 3134 /* Count other sentinels votes */ 3135 di = dictGetIterator(master->sentinels); 3136 while((de = dictNext(di)) != NULL) { 3137 sentinelRedisInstance *ri = dictGetVal(de); 3138 if (ri->leader != NULL && ri->leader_epoch == sentinel.current_epoch) 3139 sentinelLeaderIncr(counters,ri->leader); 3140 voters++; 3141 } 3142 dictReleaseIterator(di); 3143 3144 /* Check what's the winner. For the winner to win, it needs two conditions: 3145 * 1) Absolute majority between voters (50% + 1). 3146 * 2) And anyway at least master->quorum votes. */ 3147 di = dictGetIterator(counters); 3148 while((de = dictNext(di)) != NULL) { 3149 uint64_t votes = dictGetUnsignedIntegerVal(de); 3150 3151 if (votes > max_votes) { 3152 max_votes = votes; 3153 winner = dictGetKey(de); 3154 } 3155 } 3156 dictReleaseIterator(di); 3157 3158 /* Count this Sentinel vote: 3159 * if this Sentinel did not voted yet, either vote for the most 3160 * common voted sentinel, or for itself if no vote exists at all. */ 3161 if (winner) 3162 myvote = sentinelVoteLeader(master,epoch,winner,&leader_epoch); 3163 else 3164 myvote = sentinelVoteLeader(master,epoch,server.runid,&leader_epoch); 3165 3166 if (myvote && leader_epoch == epoch) { 3167 uint64_t votes = sentinelLeaderIncr(counters,myvote); 3168 3169 if (votes > max_votes) { 3170 max_votes = votes; 3171 winner = myvote; 3172 } 3173 } 3174 voters++; /* Anyway, count me as one of the voters. */ 3175 3176 voters_quorum = voters/2+1; 3177 if (winner && (max_votes < voters_quorum || max_votes < master->quorum)) 3178 winner = NULL; 3179 3180 winner = winner ? sdsnew(winner) : NULL; 3181 sdsfree(myvote); 3182 dictRelease(counters); 3183 return winner; 3184 } 3185 3186 /* Send SLAVEOF to the specified instance, always followed by a 3187 * CONFIG REWRITE command in order to store the new configuration on disk 3188 * when possible (that is, if the Redis instance is recent enough to support 3189 * config rewriting, and if the server was started with a configuration file). 3190 * 3191 * If Host is NULL the function sends "SLAVEOF NO ONE". 3192 * 3193 * The command returns REDIS_OK if the SLAVEOF command was accepted for 3194 * (later) delivery otherwise REDIS_ERR. The command replies are just 3195 * discarded. */ 3196 int sentinelSendSlaveOf(sentinelRedisInstance *ri, char *host, int port) { 3197 char portstr[32]; 3198 int retval; 3199 3200 ll2string(portstr,sizeof(portstr),port); 3201 3202 if (host == NULL) { 3203 host = "NO"; 3204 memcpy(portstr,"ONE",4); 3205 } 3206 3207 retval = redisAsyncCommand(ri->cc, 3208 sentinelDiscardReplyCallback, NULL, "SLAVEOF %s %s", host, portstr); 3209 if (retval == REDIS_ERR) return retval; 3210 3211 ri->pending_commands++; 3212 if (redisAsyncCommand(ri->cc, 3213 sentinelDiscardReplyCallback, NULL, "CONFIG REWRITE") == REDIS_OK) 3214 { 3215 ri->pending_commands++; 3216 } 3217 return REDIS_OK; 3218 } 3219 3220 /* Setup the master state to start a failover. */ 3221 void sentinelStartFailover(sentinelRedisInstance *master) { 3222 redisAssert(master->flags & SRI_MASTER); 3223 3224 master->failover_state = SENTINEL_FAILOVER_STATE_WAIT_START; 3225 master->flags |= SRI_FAILOVER_IN_PROGRESS; 3226 master->failover_epoch = ++sentinel.current_epoch; 3227 sentinelEvent(REDIS_WARNING,"+new-epoch",master,"%llu", 3228 (unsigned long long) sentinel.current_epoch); 3229 sentinelEvent(REDIS_WARNING,"+try-failover",master,"%@"); 3230 master->failover_start_time = mstime()+rand()%SENTINEL_MAX_DESYNC; 3231 master->failover_state_change_time = mstime(); 3232 } 3233 3234 /* This function checks if there are the conditions to start the failover, 3235 * that is: 3236 * 3237 * 1) Master must be in ODOWN condition. 3238 * 2) No failover already in progress. 3239 * 3) No failover already attempted recently. 3240 * 3241 * We still don't know if we'll win the election so it is possible that we 3242 * start the failover but that we'll not be able to act. 3243 * 3244 * Return non-zero if a failover was started. */ 3245 int sentinelStartFailoverIfNeeded(sentinelRedisInstance *master) { 3246 /* We can't failover if the master is not in O_DOWN state. */ 3247 if (!(master->flags & SRI_O_DOWN)) return 0; 3248 3249 /* Failover already in progress? */ 3250 if (master->flags & SRI_FAILOVER_IN_PROGRESS) return 0; 3251 3252 /* Last failover attempt started too little time ago? */ 3253 if (mstime() - master->failover_start_time < 3254 master->failover_timeout*2) return 0; 3255 3256 sentinelStartFailover(master); 3257 return 1; 3258 } 3259 3260 /* Select a suitable slave to promote. The current algorithm only uses 3261 * the following parameters: 3262 * 3263 * 1) None of the following conditions: S_DOWN, O_DOWN, DISCONNECTED. 3264 * 2) Last time the slave replied to ping no more than 5 times the PING period. 3265 * 3) info_refresh not older than 3 times the INFO refresh period. 3266 * 4) master_link_down_time no more than: 3267 * (now - master->s_down_since_time) + (master->down_after_period * 10). 3268 * Basically since the master is down from our POV, the slave reports 3269 * to be disconnected no more than 10 times the configured down-after-period. 3270 * This is pretty much black magic but the idea is, the master was not 3271 * available so the slave may be lagging, but not over a certain time. 3272 * Anyway we'll select the best slave according to replication offset. 3273 * 5) Slave priority can't be zero, otherwise the slave is discarded. 3274 * 3275 * Among all the slaves matching the above conditions we select the slave 3276 * with, in order of sorting key: 3277 * 3278 * - lower slave_priority. 3279 * - bigger processed replication offset. 3280 * - lexicographically smaller runid. 3281 * 3282 * Basically if runid is the same, the slave that processed more commands 3283 * from the master is selected. 3284 * 3285 * The function returns the pointer to the selected slave, otherwise 3286 * NULL if no suitable slave was found. 3287 */ 3288 3289 /* Helper for sentinelSelectSlave(). This is used by qsort() in order to 3290 * sort suitable slaves in a "better first" order, to take the first of 3291 * the list. */ 3292 int compareSlavesForPromotion(const void *a, const void *b) { 3293 sentinelRedisInstance **sa = (sentinelRedisInstance **)a, 3294 **sb = (sentinelRedisInstance **)b; 3295 char *sa_runid, *sb_runid; 3296 3297 if ((*sa)->slave_priority != (*sb)->slave_priority) 3298 return (*sa)->slave_priority - (*sb)->slave_priority; 3299 3300 /* If priority is the same, select the slave with greater replication 3301 * offset (processed more data frmo the master). */ 3302 if ((*sa)->slave_repl_offset > (*sb)->slave_repl_offset) { 3303 return -1; /* a < b */ 3304 } else if ((*sa)->slave_repl_offset < (*sb)->slave_repl_offset) { 3305 return 1; /* b > a */ 3306 } 3307 3308 /* If the replication offset is the same select the slave with that has 3309 * the lexicographically smaller runid. Note that we try to handle runid 3310 * == NULL as there are old Redis versions that don't publish runid in 3311 * INFO. A NULL runid is considered bigger than any other runid. */ 3312 sa_runid = (*sa)->runid; 3313 sb_runid = (*sb)->runid; 3314 if (sa_runid == NULL && sb_runid == NULL) return 0; 3315 else if (sa_runid == NULL) return 1; /* a > b */ 3316 else if (sb_runid == NULL) return -1; /* a < b */ 3317 return strcasecmp(sa_runid, sb_runid); 3318 } 3319 3320 sentinelRedisInstance *sentinelSelectSlave(sentinelRedisInstance *master) { 3321 sentinelRedisInstance **instance = 3322 zmalloc(sizeof(instance[0])*dictSize(master->slaves)); 3323 sentinelRedisInstance *selected = NULL; 3324 int instances = 0; 3325 dictIterator *di; 3326 dictEntry *de; 3327 mstime_t max_master_down_time = 0; 3328 3329 if (master->flags & SRI_S_DOWN) 3330 max_master_down_time += mstime() - master->s_down_since_time; 3331 max_master_down_time += master->down_after_period * 10; 3332 3333 di = dictGetIterator(master->slaves); 3334 while((de = dictNext(di)) != NULL) { 3335 sentinelRedisInstance *slave = dictGetVal(de); 3336 mstime_t info_validity_time; 3337 3338 if (slave->flags & (SRI_S_DOWN|SRI_O_DOWN|SRI_DISCONNECTED)) continue; 3339 if (mstime() - slave->last_avail_time > SENTINEL_PING_PERIOD*5) continue; 3340 if (slave->slave_priority == 0) continue; 3341 3342 /* If the master is in SDOWN state we get INFO for slaves every second. 3343 * Otherwise we get it with the usual period so we need to account for 3344 * a larger delay. */ 3345 if (master->flags & SRI_S_DOWN) 3346 info_validity_time = SENTINEL_PING_PERIOD*5; 3347 else 3348 info_validity_time = SENTINEL_INFO_PERIOD*3; 3349 if (mstime() - slave->info_refresh > info_validity_time) continue; 3350 if (slave->master_link_down_time > max_master_down_time) continue; 3351 instance[instances++] = slave; 3352 } 3353 dictReleaseIterator(di); 3354 if (instances) { 3355 qsort(instance,instances,sizeof(sentinelRedisInstance*), 3356 compareSlavesForPromotion); 3357 selected = instance[0]; 3358 } 3359 zfree(instance); 3360 return selected; 3361 } 3362 3363 /* ---------------- Failover state machine implementation ------------------- */ 3364 void sentinelFailoverWaitStart(sentinelRedisInstance *ri) { 3365 char *leader; 3366 int isleader; 3367 3368 /* Check if we are the leader for the failover epoch. */ 3369 leader = sentinelGetLeader(ri, ri->failover_epoch); 3370 isleader = leader && strcasecmp(leader,server.runid) == 0; 3371 sdsfree(leader); 3372 3373 /* If I'm not the leader, and it is not a forced failover via 3374 * SENTINEL FAILOVER, then I can't continue with the failover. */ 3375 if (!isleader && !(ri->flags & SRI_FORCE_FAILOVER)) { 3376 int election_timeout = SENTINEL_ELECTION_TIMEOUT; 3377 3378 /* The election timeout is the MIN between SENTINEL_ELECTION_TIMEOUT 3379 * and the configured failover timeout. */ 3380 if (election_timeout > ri->failover_timeout) 3381 election_timeout = ri->failover_timeout; 3382 /* Abort the failover if I'm not the leader after some time. */ 3383 if (mstime() - ri->failover_start_time > election_timeout) { 3384 sentinelEvent(REDIS_WARNING,"-failover-abort-not-elected",ri,"%@"); 3385 sentinelAbortFailover(ri); 3386 } 3387 return; 3388 } 3389 sentinelEvent(REDIS_WARNING,"+elected-leader",ri,"%@"); 3390 ri->failover_state = SENTINEL_FAILOVER_STATE_SELECT_SLAVE; 3391 ri->failover_state_change_time = mstime(); 3392 sentinelEvent(REDIS_WARNING,"+failover-state-select-slave",ri,"%@"); 3393 } 3394 3395 void sentinelFailoverSelectSlave(sentinelRedisInstance *ri) { 3396 sentinelRedisInstance *slave = sentinelSelectSlave(ri); 3397 3398 /* We don't handle the timeout in this state as the function aborts 3399 * the failover or go forward in the next state. */ 3400 if (slave == NULL) { 3401 sentinelEvent(REDIS_WARNING,"-failover-abort-no-good-slave",ri,"%@"); 3402 sentinelAbortFailover(ri); 3403 } else { 3404 sentinelEvent(REDIS_WARNING,"+selected-slave",slave,"%@"); 3405 slave->flags |= SRI_PROMOTED; 3406 ri->promoted_slave = slave; 3407 ri->failover_state = SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE; 3408 ri->failover_state_change_time = mstime(); 3409 sentinelEvent(REDIS_NOTICE,"+failover-state-send-slaveof-noone", 3410 slave, "%@"); 3411 } 3412 } 3413 3414 void sentinelFailoverSendSlaveOfNoOne(sentinelRedisInstance *ri) { 3415 int retval; 3416 3417 /* We can't send the command to the promoted slave if it is now 3418 * disconnected. Retry again and again with this state until the timeout 3419 * is reached, then abort the failover. */ 3420 if (ri->promoted_slave->flags & SRI_DISCONNECTED) { 3421 if (mstime() - ri->failover_state_change_time > ri->failover_timeout) { 3422 sentinelEvent(REDIS_WARNING,"-failover-abort-slave-timeout",ri,"%@"); 3423 sentinelAbortFailover(ri); 3424 } 3425 return; 3426 } 3427 3428 /* Send SLAVEOF NO ONE command to turn the slave into a master. 3429 * We actually register a generic callback for this command as we don't 3430 * really care about the reply. We check if it worked indirectly observing 3431 * if INFO returns a different role (master instead of slave). */ 3432 retval = sentinelSendSlaveOf(ri->promoted_slave,NULL,0); 3433 if (retval != REDIS_OK) return; 3434 sentinelEvent(REDIS_NOTICE, "+failover-state-wait-promotion", 3435 ri->promoted_slave,"%@"); 3436 ri->failover_state = SENTINEL_FAILOVER_STATE_WAIT_PROMOTION; 3437 ri->failover_state_change_time = mstime(); 3438 } 3439 3440 /* We actually wait for promotion indirectly checking with INFO when the 3441 * slave turns into a master. */ 3442 void sentinelFailoverWaitPromotion(sentinelRedisInstance *ri) { 3443 /* Just handle the timeout. Switching to the next state is handled 3444 * by the function parsing the INFO command of the promoted slave. */ 3445 if (mstime() - ri->failover_state_change_time > ri->failover_timeout) { 3446 sentinelEvent(REDIS_WARNING,"-failover-abort-slave-timeout",ri,"%@"); 3447 sentinelAbortFailover(ri); 3448 } 3449 } 3450 3451 void sentinelFailoverDetectEnd(sentinelRedisInstance *master) { 3452 int not_reconfigured = 0, timeout = 0; 3453 dictIterator *di; 3454 dictEntry *de; 3455 mstime_t elapsed = mstime() - master->failover_state_change_time; 3456 3457 /* We can't consider failover finished if the promoted slave is 3458 * not reachable. */ 3459 if (master->promoted_slave == NULL || 3460 master->promoted_slave->flags & SRI_S_DOWN) return; 3461 3462 /* The failover terminates once all the reachable slaves are properly 3463 * configured. */ 3464 di = dictGetIterator(master->slaves); 3465 while((de = dictNext(di)) != NULL) { 3466 sentinelRedisInstance *slave = dictGetVal(de); 3467 3468 if (slave->flags & (SRI_PROMOTED|SRI_RECONF_DONE)) continue; 3469 if (slave->flags & SRI_S_DOWN) continue; 3470 not_reconfigured++; 3471 } 3472 dictReleaseIterator(di); 3473 3474 /* Force end of failover on timeout. */ 3475 if (elapsed > master->failover_timeout) { 3476 not_reconfigured = 0; 3477 timeout = 1; 3478 sentinelEvent(REDIS_WARNING,"+failover-end-for-timeout",master,"%@"); 3479 } 3480 3481 if (not_reconfigured == 0) { 3482 sentinelEvent(REDIS_WARNING,"+failover-end",master,"%@"); 3483 master->failover_state = SENTINEL_FAILOVER_STATE_UPDATE_CONFIG; 3484 master->failover_state_change_time = mstime(); 3485 } 3486 3487 /* If I'm the leader it is a good idea to send a best effort SLAVEOF 3488 * command to all the slaves still not reconfigured to replicate with 3489 * the new master. */ 3490 if (timeout) { 3491 dictIterator *di; 3492 dictEntry *de; 3493 3494 di = dictGetIterator(master->slaves); 3495 while((de = dictNext(di)) != NULL) { 3496 sentinelRedisInstance *slave = dictGetVal(de); 3497 int retval; 3498 3499 if (slave->flags & 3500 (SRI_RECONF_DONE|SRI_RECONF_SENT|SRI_DISCONNECTED)) continue; 3501 3502 retval = sentinelSendSlaveOf(slave, 3503 master->promoted_slave->addr->ip, 3504 master->promoted_slave->addr->port); 3505 if (retval == REDIS_OK) { 3506 sentinelEvent(REDIS_NOTICE,"+slave-reconf-sent-be",slave,"%@"); 3507 slave->flags |= SRI_RECONF_SENT; 3508 } 3509 } 3510 dictReleaseIterator(di); 3511 } 3512 } 3513 3514 /* Send SLAVE OF <new master address> to all the remaining slaves that 3515 * still don't appear to have the configuration updated. */ 3516 void sentinelFailoverReconfNextSlave(sentinelRedisInstance *master) { 3517 dictIterator *di; 3518 dictEntry *de; 3519 int in_progress = 0; 3520 3521 di = dictGetIterator(master->slaves); 3522 while((de = dictNext(di)) != NULL) { 3523 sentinelRedisInstance *slave = dictGetVal(de); 3524 3525 if (slave->flags & (SRI_RECONF_SENT|SRI_RECONF_INPROG)) 3526 in_progress++; 3527 } 3528 dictReleaseIterator(di); 3529 3530 di = dictGetIterator(master->slaves); 3531 while(in_progress < master->parallel_syncs && 3532 (de = dictNext(di)) != NULL) 3533 { 3534 sentinelRedisInstance *slave = dictGetVal(de); 3535 int retval; 3536 3537 /* Skip the promoted slave, and already configured slaves. */ 3538 if (slave->flags & (SRI_PROMOTED|SRI_RECONF_DONE)) continue; 3539 3540 /* If too much time elapsed without the slave moving forward to 3541 * the next state, consider it reconfigured even if it is not. 3542 * Sentinels will detect the slave as misconfigured and fix its 3543 * configuration later. */ 3544 if ((slave->flags & SRI_RECONF_SENT) && 3545 (mstime() - slave->slave_reconf_sent_time) > 3546 SENTINEL_SLAVE_RECONF_TIMEOUT) 3547 { 3548 sentinelEvent(REDIS_NOTICE,"-slave-reconf-sent-timeout",slave,"%@"); 3549 slave->flags &= ~SRI_RECONF_SENT; 3550 slave->flags |= SRI_RECONF_DONE; 3551 } 3552 3553 /* Nothing to do for instances that are disconnected or already 3554 * in RECONF_SENT state. */ 3555 if (slave->flags & (SRI_DISCONNECTED|SRI_RECONF_SENT|SRI_RECONF_INPROG)) 3556 continue; 3557 3558 /* Send SLAVEOF <new master>. */ 3559 retval = sentinelSendSlaveOf(slave, 3560 master->promoted_slave->addr->ip, 3561 master->promoted_slave->addr->port); 3562 if (retval == REDIS_OK) { 3563 slave->flags |= SRI_RECONF_SENT; 3564 slave->slave_reconf_sent_time = mstime(); 3565 sentinelEvent(REDIS_NOTICE,"+slave-reconf-sent",slave,"%@"); 3566 in_progress++; 3567 } 3568 } 3569 dictReleaseIterator(di); 3570 3571 /* Check if all the slaves are reconfigured and handle timeout. */ 3572 sentinelFailoverDetectEnd(master); 3573 } 3574 3575 /* This function is called when the slave is in 3576 * SENTINEL_FAILOVER_STATE_UPDATE_CONFIG state. In this state we need 3577 * to remove it from the master table and add the promoted slave instead. */ 3578 void sentinelFailoverSwitchToPromotedSlave(sentinelRedisInstance *master) { 3579 sentinelRedisInstance *ref = master->promoted_slave ? 3580 master->promoted_slave : master; 3581 3582 sentinelEvent(REDIS_WARNING,"+switch-master",master,"%s %s %d %s %d", 3583 master->name, master->addr->ip, master->addr->port, 3584 ref->addr->ip, ref->addr->port); 3585 3586 sentinelResetMasterAndChangeAddress(master,ref->addr->ip,ref->addr->port); 3587 } 3588 3589 void sentinelFailoverStateMachine(sentinelRedisInstance *ri) { 3590 redisAssert(ri->flags & SRI_MASTER); 3591 3592 if (!(ri->flags & SRI_FAILOVER_IN_PROGRESS)) return; 3593 3594 switch(ri->failover_state) { 3595 case SENTINEL_FAILOVER_STATE_WAIT_START: 3596 sentinelFailoverWaitStart(ri); 3597 break; 3598 case SENTINEL_FAILOVER_STATE_SELECT_SLAVE: 3599 sentinelFailoverSelectSlave(ri); 3600 break; 3601 case SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE: 3602 sentinelFailoverSendSlaveOfNoOne(ri); 3603 break; 3604 case SENTINEL_FAILOVER_STATE_WAIT_PROMOTION: 3605 sentinelFailoverWaitPromotion(ri); 3606 break; 3607 case SENTINEL_FAILOVER_STATE_RECONF_SLAVES: 3608 sentinelFailoverReconfNextSlave(ri); 3609 break; 3610 } 3611 } 3612 3613 /* Abort a failover in progress: 3614 * 3615 * This function can only be called before the promoted slave acknowledged 3616 * the slave -> master switch. Otherwise the failover can't be aborted and 3617 * will reach its end (possibly by timeout). */ 3618 void sentinelAbortFailover(sentinelRedisInstance *ri) { 3619 redisAssert(ri->flags & SRI_FAILOVER_IN_PROGRESS); 3620 redisAssert(ri->failover_state <= SENTINEL_FAILOVER_STATE_WAIT_PROMOTION); 3621 3622 ri->flags &= ~(SRI_FAILOVER_IN_PROGRESS|SRI_FORCE_FAILOVER); 3623 ri->failover_state = SENTINEL_FAILOVER_STATE_NONE; 3624 ri->failover_state_change_time = mstime(); 3625 if (ri->promoted_slave) { 3626 ri->promoted_slave->flags &= ~SRI_PROMOTED; 3627 ri->promoted_slave = NULL; 3628 } 3629 } 3630 3631 /* ======================== SENTINEL timer handler ========================== 3632 * This is the "main" our Sentinel, being sentinel completely non blocking 3633 * in design. The function is called every second. 3634 * -------------------------------------------------------------------------- */ 3635 3636 /* Perform scheduled operations for the specified Redis instance. */ 3637 void sentinelHandleRedisInstance(sentinelRedisInstance *ri) { 3638 /* ========== MONITORING HALF ============ */ 3639 /* Every kind of instance */ 3640 sentinelReconnectInstance(ri); 3641 sentinelSendPeriodicCommands(ri); 3642 3643 /* ============== ACTING HALF ============= */ 3644 /* We don't proceed with the acting half if we are in TILT mode. 3645 * TILT happens when we find something odd with the time, like a 3646 * sudden change in the clock. */ 3647 if (sentinel.tilt) { 3648 if (mstime()-sentinel.tilt_start_time < SENTINEL_TILT_PERIOD) return; 3649 sentinel.tilt = 0; 3650 sentinelEvent(REDIS_WARNING,"-tilt",NULL,"#tilt mode exited"); 3651 } 3652 3653 /* Every kind of instance */ 3654 sentinelCheckSubjectivelyDown(ri); 3655 3656 /* Masters and slaves */ 3657 if (ri->flags & (SRI_MASTER|SRI_SLAVE)) { 3658 /* Nothing so far. */ 3659 } 3660 3661 /* Only masters */ 3662 if (ri->flags & SRI_MASTER) { 3663 sentinelCheckObjectivelyDown(ri); 3664 if (sentinelStartFailoverIfNeeded(ri)) 3665 sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_ASK_FORCED); 3666 sentinelFailoverStateMachine(ri); 3667 sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_NO_FLAGS); 3668 } 3669 } 3670 3671 /* Perform scheduled operations for all the instances in the dictionary. 3672 * Recursively call the function against dictionaries of slaves. */ 3673 void sentinelHandleDictOfRedisInstances(dict *instances) { 3674 dictIterator *di; 3675 dictEntry *de; 3676 sentinelRedisInstance *switch_to_promoted = NULL; 3677 3678 /* There are a number of things we need to perform against every master. */ 3679 di = dictGetIterator(instances); 3680 while((de = dictNext(di)) != NULL) { 3681 sentinelRedisInstance *ri = dictGetVal(de); 3682 3683 sentinelHandleRedisInstance(ri); 3684 if (ri->flags & SRI_MASTER) { 3685 sentinelHandleDictOfRedisInstances(ri->slaves); 3686 sentinelHandleDictOfRedisInstances(ri->sentinels); 3687 if (ri->failover_state == SENTINEL_FAILOVER_STATE_UPDATE_CONFIG) { 3688 switch_to_promoted = ri; 3689 } 3690 } 3691 } 3692 if (switch_to_promoted) 3693 sentinelFailoverSwitchToPromotedSlave(switch_to_promoted); 3694 dictReleaseIterator(di); 3695 } 3696 3697 /* This function checks if we need to enter the TITL mode. 3698 * 3699 * The TILT mode is entered if we detect that between two invocations of the 3700 * timer interrupt, a negative amount of time, or too much time has passed. 3701 * Note that we expect that more or less just 100 milliseconds will pass 3702 * if everything is fine. However we'll see a negative number or a 3703 * difference bigger than SENTINEL_TILT_TRIGGER milliseconds if one of the 3704 * following conditions happen: 3705 * 3706 * 1) The Sentiel process for some time is blocked, for every kind of 3707 * random reason: the load is huge, the computer was frozen for some time 3708 * in I/O or alike, the process was stopped by a signal. Everything. 3709 * 2) The system clock was altered significantly. 3710 * 3711 * Under both this conditions we'll see everything as timed out and failing 3712 * without good reasons. Instead we enter the TILT mode and wait 3713 * for SENTINEL_TILT_PERIOD to elapse before starting to act again. 3714 * 3715 * During TILT time we still collect information, we just do not act. */ 3716 void sentinelCheckTiltCondition(void) { 3717 mstime_t now = mstime(); 3718 mstime_t delta = now - sentinel.previous_time; 3719 3720 if (delta < 0 || delta > SENTINEL_TILT_TRIGGER) { 3721 sentinel.tilt = 1; 3722 sentinel.tilt_start_time = mstime(); 3723 sentinelEvent(REDIS_WARNING,"+tilt",NULL,"#tilt mode entered"); 3724 } 3725 sentinel.previous_time = mstime(); 3726 } 3727 3728 void sentinelTimer(void) { 3729 sentinelCheckTiltCondition(); 3730 sentinelHandleDictOfRedisInstances(sentinel.masters); 3731 sentinelRunPendingScripts(); 3732 sentinelCollectTerminatedScripts(); 3733 sentinelKillTimedoutScripts(); 3734 3735 /* We continuously change the frequency of the Redis "timer interrupt" 3736 * in order to desynchronize every Sentinel from every other. 3737 * This non-determinism avoids that Sentinels started at the same time 3738 * exactly continue to stay synchronized asking to be voted at the 3739 * same time again and again (resulting in nobody likely winning the 3740 * election because of split brain voting). */ 3741 server.hz = REDIS_DEFAULT_HZ + rand() % REDIS_DEFAULT_HZ; 3742 } 3743 3744