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 40 extern char **environ; 41 42 #define REDIS_SENTINEL_PORT 26379 43 44 /* ======================== Sentinel global state =========================== */ 45 46 typedef long long mstime_t; /* millisecond time type. */ 47 48 /* Address object, used to describe an ip:port pair. */ 49 typedef struct sentinelAddr { 50 char *ip; 51 int port; 52 } sentinelAddr; 53 54 /* A Sentinel Redis Instance object is monitoring. */ 55 #define SRI_MASTER (1<<0) 56 #define SRI_SLAVE (1<<1) 57 #define SRI_SENTINEL (1<<2) 58 #define SRI_DISCONNECTED (1<<3) 59 #define SRI_S_DOWN (1<<4) /* Subjectively down (no quorum). */ 60 #define SRI_O_DOWN (1<<5) /* Objectively down (quorum reached). */ 61 #define SRI_MASTER_DOWN (1<<6) /* A Sentinel with this flag set thinks that 62 its master is down. */ 63 /* SRI_CAN_FAILOVER when set in an SRI_MASTER instance means that we are 64 * allowed to perform the failover for this master. 65 * When set in a SRI_SENTINEL instance means that sentinel is allowed to 66 * perform the failover on its master. */ 67 #define SRI_CAN_FAILOVER (1<<7) 68 #define SRI_FAILOVER_IN_PROGRESS (1<<8) /* Failover is in progress for 69 this master. */ 70 #define SRI_I_AM_THE_LEADER (1<<9) /* We are the leader for this master. */ 71 #define SRI_PROMOTED (1<<10) /* Slave selected for promotion. */ 72 #define SRI_RECONF_SENT (1<<11) /* SLAVEOF <newmaster> sent. */ 73 #define SRI_RECONF_INPROG (1<<12) /* Slave synchronization in progress. */ 74 #define SRI_RECONF_DONE (1<<13) /* Slave synchronized with new master. */ 75 #define SRI_FORCE_FAILOVER (1<<14) /* Force failover with master up. */ 76 #define SRI_SCRIPT_KILL_SENT (1<<15) /* SCRIPT KILL already sent on -BUSY */ 77 78 #define SENTINEL_INFO_PERIOD 10000 79 #define SENTINEL_PING_PERIOD 1000 80 #define SENTINEL_ASK_PERIOD 1000 81 #define SENTINEL_PUBLISH_PERIOD 5000 82 #define SENTINEL_DOWN_AFTER_PERIOD 30000 83 #define SENTINEL_HELLO_CHANNEL "__sentinel__:hello" 84 #define SENTINEL_TILT_TRIGGER 2000 85 #define SENTINEL_TILT_PERIOD (SENTINEL_PING_PERIOD*30) 86 #define SENTINEL_DEFAULT_SLAVE_PRIORITY 100 87 #define SENTINEL_PROMOTION_RETRY_PERIOD 30000 88 #define SENTINEL_SLAVE_RECONF_RETRY_PERIOD 10000 89 #define SENTINEL_DEFAULT_PARALLEL_SYNCS 1 90 #define SENTINEL_MIN_LINK_RECONNECT_PERIOD 15000 91 #define SENTINEL_DEFAULT_FAILOVER_TIMEOUT (60*15*1000) 92 #define SENTINEL_MAX_PENDING_COMMANDS 100 93 #define SENTINEL_EXTENDED_SDOWN_MULTIPLIER 10 94 95 /* How many milliseconds is an information valid? This applies for instance 96 * to the reply to SENTINEL IS-MASTER-DOWN-BY-ADDR replies. */ 97 #define SENTINEL_INFO_VALIDITY_TIME 5000 98 #define SENTINEL_FAILOVER_FIXED_DELAY 5000 99 #define SENTINEL_FAILOVER_MAX_RANDOM_DELAY 10000 100 101 /* Failover machine different states. */ 102 #define SENTINEL_FAILOVER_STATE_NONE 0 /* No failover in progress. */ 103 #define SENTINEL_FAILOVER_STATE_WAIT_START 1 /* Wait for failover_start_time*/ 104 #define SENTINEL_FAILOVER_STATE_SELECT_SLAVE 2 /* Select slave to promote */ 105 #define SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE 3 /* Slave -> Master */ 106 #define SENTINEL_FAILOVER_STATE_WAIT_PROMOTION 4 /* Wait slave to change role */ 107 #define SENTINEL_FAILOVER_STATE_RECONF_SLAVES 5 /* SLAVEOF newmaster */ 108 #define SENTINEL_FAILOVER_STATE_WAIT_NEXT_SLAVE 6 /* wait replication */ 109 #define SENTINEL_FAILOVER_STATE_ALERT_CLIENTS 7 /* Run user script. */ 110 #define SENTINEL_FAILOVER_STATE_WAIT_ALERT_SCRIPT 8 /* Wait script exec. */ 111 #define SENTINEL_FAILOVER_STATE_DETECT_END 9 /* Check for failover end. */ 112 #define SENTINEL_FAILOVER_STATE_UPDATE_CONFIG 10 /* Monitor promoted slave. */ 113 114 #define SENTINEL_MASTER_LINK_STATUS_UP 0 115 #define SENTINEL_MASTER_LINK_STATUS_DOWN 1 116 117 /* Generic flags that can be used with different functions. */ 118 #define SENTINEL_NO_FLAGS 0 119 #define SENTINEL_GENERATE_EVENT 1 120 #define SENTINEL_LEADER 2 121 #define SENTINEL_OBSERVER 4 122 123 /* Script execution flags and limits. */ 124 #define SENTINEL_SCRIPT_NONE 0 125 #define SENTINEL_SCRIPT_RUNNING 1 126 #define SENTINEL_SCRIPT_MAX_QUEUE 256 127 #define SENTINEL_SCRIPT_MAX_RUNNING 16 128 #define SENTINEL_SCRIPT_MAX_RUNTIME 60000 /* 60 seconds max exec time. */ 129 #define SENTINEL_SCRIPT_MAX_RETRY 10 130 #define SENTINEL_SCRIPT_RETRY_DELAY 30000 /* 30 seconds between retries. */ 131 132 typedef struct sentinelRedisInstance { 133 int flags; /* See SRI_... defines */ 134 char *name; /* Master name from the point of view of this sentinel. */ 135 char *runid; /* run ID of this instance. */ 136 sentinelAddr *addr; /* Master host. */ 137 redisAsyncContext *cc; /* Hiredis context for commands. */ 138 redisAsyncContext *pc; /* Hiredis context for Pub / Sub. */ 139 int pending_commands; /* Number of commands sent waiting for a reply. */ 140 mstime_t cc_conn_time; /* cc connection time. */ 141 mstime_t pc_conn_time; /* pc connection time. */ 142 mstime_t pc_last_activity; /* Last time we received any message. */ 143 mstime_t last_avail_time; /* Last time the instance replied to ping with 144 a reply we consider valid. */ 145 mstime_t last_pong_time; /* Last time the instance replied to ping, 146 whatever the reply was. That's used to check 147 if the link is idle and must be reconnected. */ 148 mstime_t last_pub_time; /* Last time we sent hello via Pub/Sub. */ 149 mstime_t last_hello_time; /* Only used if SRI_SENTINEL is set. Last time 150 we received an hello from this Sentinel 151 via Pub/Sub. */ 152 mstime_t last_master_down_reply_time; /* Time of last reply to 153 SENTINEL is-master-down command. */ 154 mstime_t s_down_since_time; /* Subjectively down since time. */ 155 mstime_t o_down_since_time; /* Objectively down since time. */ 156 mstime_t down_after_period; /* Consider it down after that period. */ 157 mstime_t info_refresh; /* Time at which we received INFO output from it. */ 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 SRI_SLAVE is set. */ 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 /* Failover */ 175 char *leader; /* If this is a master instance, this is the runid of 176 the Sentinel that should perform the failover. If 177 this is a Sentinel, this is the runid of the Sentinel 178 that this other Sentinel is voting as leader. 179 This field is valid only if SRI_MASTER_DOWN is 180 set on the Sentinel instance. */ 181 int failover_state; /* See SENTINEL_FAILOVER_STATE_* defines. */ 182 mstime_t failover_state_change_time; 183 mstime_t failover_start_time; /* When to start to failover if leader. */ 184 mstime_t failover_timeout; /* Max time to refresh failover state. */ 185 struct sentinelRedisInstance *promoted_slave; /* Promoted slave instance. */ 186 /* Scripts executed to notify admin or reconfigure clients: when they 187 * are set to NULL no script is executed. */ 188 char *notification_script; 189 char *client_reconfig_script; 190 } sentinelRedisInstance; 191 192 /* Main state. */ 193 struct sentinelState { 194 dict *masters; /* Dictionary of master sentinelRedisInstances. 195 Key is the instance name, value is the 196 sentinelRedisInstance structure pointer. */ 197 int tilt; /* Are we in TILT mode? */ 198 int running_scripts; /* Number of scripts in execution right now. */ 199 mstime_t tilt_start_time; /* When TITL started. */ 200 mstime_t previous_time; /* Time last time we ran the time handler. */ 201 list *scripts_queue; /* Queue of user scripts to execute. */ 202 } sentinel; 203 204 /* A script execution job. */ 205 typedef struct sentinelScriptJob { 206 int flags; /* Script job flags: SENTINEL_SCRIPT_* */ 207 int retry_num; /* Number of times we tried to execute it. */ 208 char **argv; /* Arguments to call the script. */ 209 mstime_t start_time; /* Script execution time if the script is running, 210 otherwise 0 if we are allowed to retry the 211 execution at any time. If the script is not 212 running and it's not 0, it means: do not run 213 before the specified time. */ 214 pid_t pid; /* Script execution pid. */ 215 } sentinelScriptJob; 216 217 /* ======================= hiredis ae.c adapters ============================= 218 * Note: this implementation is taken from hiredis/adapters/ae.h, however 219 * we have our modified copy for Sentinel in order to use our allocator 220 * and to have full control over how the adapter works. */ 221 222 typedef struct redisAeEvents { 223 redisAsyncContext *context; 224 aeEventLoop *loop; 225 int fd; 226 int reading, writing; 227 } redisAeEvents; 228 229 static void redisAeReadEvent(aeEventLoop *el, int fd, void *privdata, int mask) { 230 ((void)el); ((void)fd); ((void)mask); 231 232 redisAeEvents *e = (redisAeEvents*)privdata; 233 redisAsyncHandleRead(e->context); 234 } 235 236 static void redisAeWriteEvent(aeEventLoop *el, int fd, void *privdata, int mask) { 237 ((void)el); ((void)fd); ((void)mask); 238 239 redisAeEvents *e = (redisAeEvents*)privdata; 240 redisAsyncHandleWrite(e->context); 241 } 242 243 static void redisAeAddRead(void *privdata) { 244 redisAeEvents *e = (redisAeEvents*)privdata; 245 aeEventLoop *loop = e->loop; 246 if (!e->reading) { 247 e->reading = 1; 248 aeCreateFileEvent(loop,e->fd,AE_READABLE,redisAeReadEvent,e); 249 } 250 } 251 252 static void redisAeDelRead(void *privdata) { 253 redisAeEvents *e = (redisAeEvents*)privdata; 254 aeEventLoop *loop = e->loop; 255 if (e->reading) { 256 e->reading = 0; 257 aeDeleteFileEvent(loop,e->fd,AE_READABLE); 258 } 259 } 260 261 static void redisAeAddWrite(void *privdata) { 262 redisAeEvents *e = (redisAeEvents*)privdata; 263 aeEventLoop *loop = e->loop; 264 if (!e->writing) { 265 e->writing = 1; 266 aeCreateFileEvent(loop,e->fd,AE_WRITABLE,redisAeWriteEvent,e); 267 } 268 } 269 270 static void redisAeDelWrite(void *privdata) { 271 redisAeEvents *e = (redisAeEvents*)privdata; 272 aeEventLoop *loop = e->loop; 273 if (e->writing) { 274 e->writing = 0; 275 aeDeleteFileEvent(loop,e->fd,AE_WRITABLE); 276 } 277 } 278 279 static void redisAeCleanup(void *privdata) { 280 redisAeEvents *e = (redisAeEvents*)privdata; 281 redisAeDelRead(privdata); 282 redisAeDelWrite(privdata); 283 zfree(e); 284 } 285 286 static int redisAeAttach(aeEventLoop *loop, redisAsyncContext *ac) { 287 redisContext *c = &(ac->c); 288 redisAeEvents *e; 289 290 /* Nothing should be attached when something is already attached */ 291 if (ac->ev.data != NULL) 292 return REDIS_ERR; 293 294 /* Create container for context and r/w events */ 295 e = (redisAeEvents*)zmalloc(sizeof(*e)); 296 e->context = ac; 297 e->loop = loop; 298 e->fd = c->fd; 299 e->reading = e->writing = 0; 300 301 /* Register functions to start/stop listening for events */ 302 ac->ev.addRead = redisAeAddRead; 303 ac->ev.delRead = redisAeDelRead; 304 ac->ev.addWrite = redisAeAddWrite; 305 ac->ev.delWrite = redisAeDelWrite; 306 ac->ev.cleanup = redisAeCleanup; 307 ac->ev.data = e; 308 309 return REDIS_OK; 310 } 311 312 /* ============================= Prototypes ================================= */ 313 314 void sentinelLinkEstablishedCallback(const redisAsyncContext *c, int status); 315 void sentinelDisconnectCallback(const redisAsyncContext *c, int status); 316 void sentinelReceiveHelloMessages(redisAsyncContext *c, void *reply, void *privdata); 317 sentinelRedisInstance *sentinelGetMasterByName(char *name); 318 char *sentinelGetSubjectiveLeader(sentinelRedisInstance *master); 319 char *sentinelGetObjectiveLeader(sentinelRedisInstance *master); 320 int yesnotoi(char *s); 321 void sentinelDisconnectInstanceFromContext(const redisAsyncContext *c); 322 void sentinelKillLink(sentinelRedisInstance *ri, redisAsyncContext *c); 323 const char *sentinelRedisInstanceTypeStr(sentinelRedisInstance *ri); 324 void sentinelAbortFailover(sentinelRedisInstance *ri); 325 void sentinelEvent(int level, char *type, sentinelRedisInstance *ri, const char *fmt, ...); 326 sentinelRedisInstance *sentinelSelectSlave(sentinelRedisInstance *master); 327 void sentinelScheduleScriptExecution(char *path, ...); 328 void sentinelStartFailover(sentinelRedisInstance *master, int state); 329 void sentinelDiscardReplyCallback(redisAsyncContext *c, void *reply, void *privdata); 330 331 /* ========================= Dictionary types =============================== */ 332 333 unsigned int dictSdsHash(const void *key); 334 int dictSdsKeyCompare(void *privdata, const void *key1, const void *key2); 335 void releaseSentinelRedisInstance(sentinelRedisInstance *ri); 336 337 void dictInstancesValDestructor (void *privdata, void *obj) { 338 releaseSentinelRedisInstance(obj); 339 } 340 341 /* Instance name (sds) -> instance (sentinelRedisInstance pointer) 342 * 343 * also used for: sentinelRedisInstance->sentinels dictionary that maps 344 * sentinels ip:port to last seen time in Pub/Sub hello message. */ 345 dictType instancesDictType = { 346 dictSdsHash, /* hash function */ 347 NULL, /* key dup */ 348 NULL, /* val dup */ 349 dictSdsKeyCompare, /* key compare */ 350 NULL, /* key destructor */ 351 dictInstancesValDestructor /* val destructor */ 352 }; 353 354 /* Instance runid (sds) -> votes (long casted to void*) 355 * 356 * This is useful into sentinelGetObjectiveLeader() function in order to 357 * count the votes and understand who is the leader. */ 358 dictType leaderVotesDictType = { 359 dictSdsHash, /* hash function */ 360 NULL, /* key dup */ 361 NULL, /* val dup */ 362 dictSdsKeyCompare, /* key compare */ 363 NULL, /* key destructor */ 364 NULL /* val destructor */ 365 }; 366 367 /* =========================== Initialization =============================== */ 368 369 void sentinelCommand(redisClient *c); 370 void sentinelInfoCommand(redisClient *c); 371 372 struct redisCommand sentinelcmds[] = { 373 {"ping",pingCommand,1,"",0,NULL,0,0,0,0,0}, 374 {"sentinel",sentinelCommand,-2,"",0,NULL,0,0,0,0,0}, 375 {"subscribe",subscribeCommand,-2,"",0,NULL,0,0,0,0,0}, 376 {"unsubscribe",unsubscribeCommand,-1,"",0,NULL,0,0,0,0,0}, 377 {"psubscribe",psubscribeCommand,-2,"",0,NULL,0,0,0,0,0}, 378 {"punsubscribe",punsubscribeCommand,-1,"",0,NULL,0,0,0,0,0}, 379 {"info",sentinelInfoCommand,-1,"",0,NULL,0,0,0,0,0} 380 }; 381 382 /* This function overwrites a few normal Redis config default with Sentinel 383 * specific defaults. */ 384 void initSentinelConfig(void) { 385 server.port = REDIS_SENTINEL_PORT; 386 } 387 388 /* Perform the Sentinel mode initialization. */ 389 void initSentinel(void) { 390 int j; 391 392 /* Remove usual Redis commands from the command table, then just add 393 * the SENTINEL command. */ 394 dictEmpty(server.commands); 395 for (j = 0; j < sizeof(sentinelcmds)/sizeof(sentinelcmds[0]); j++) { 396 int retval; 397 struct redisCommand *cmd = sentinelcmds+j; 398 399 retval = dictAdd(server.commands, sdsnew(cmd->name), cmd); 400 redisAssert(retval == DICT_OK); 401 } 402 403 /* Initialize various data structures. */ 404 sentinel.masters = dictCreate(&instancesDictType,NULL); 405 sentinel.tilt = 0; 406 sentinel.tilt_start_time = mstime(); 407 sentinel.previous_time = mstime(); 408 sentinel.running_scripts = 0; 409 sentinel.scripts_queue = listCreate(); 410 } 411 412 /* ============================== sentinelAddr ============================== */ 413 414 /* Create a sentinelAddr object and return it on success. 415 * On error NULL is returned and errno is set to: 416 * ENOENT: Can't resolve the hostname. 417 * EINVAL: Invalid port number. 418 */ 419 sentinelAddr *createSentinelAddr(char *hostname, int port) { 420 char buf[32]; 421 sentinelAddr *sa; 422 423 if (port <= 0 || port > 65535) { 424 errno = EINVAL; 425 return NULL; 426 } 427 if (anetResolve(NULL,hostname,buf) == ANET_ERR) { 428 errno = ENOENT; 429 return NULL; 430 } 431 sa = zmalloc(sizeof(*sa)); 432 sa->ip = sdsnew(buf); 433 sa->port = port; 434 return sa; 435 } 436 437 /* Free a Sentinel address. Can't fail. */ 438 void releaseSentinelAddr(sentinelAddr *sa) { 439 sdsfree(sa->ip); 440 zfree(sa); 441 } 442 443 /* =========================== Events notification ========================== */ 444 445 /* Send an event to log, pub/sub, user notification script. 446 * 447 * 'level' is the log level for logging. Only REDIS_WARNING events will trigger 448 * the execution of the user notification script. 449 * 450 * 'type' is the message type, also used as a pub/sub channel name. 451 * 452 * 'ri', is the redis instance target of this event if applicable, and is 453 * used to obtain the path of the notification script to execute. 454 * 455 * The remaining arguments are printf-alike. 456 * If the format specifier starts with the two characters "%@" then ri is 457 * not NULL, and the message is prefixed with an instance identifier in the 458 * following format: 459 * 460 * <instance type> <instance name> <ip> <port> 461 * 462 * If the instance type is not master, than the additional string is 463 * added to specify the originating master: 464 * 465 * @ <master name> <master ip> <master port> 466 * 467 * Any other specifier after "%@" is processed by printf itself. 468 */ 469 void sentinelEvent(int level, char *type, sentinelRedisInstance *ri, 470 const char *fmt, ...) { 471 va_list ap; 472 char msg[REDIS_MAX_LOGMSG_LEN]; 473 robj *channel, *payload; 474 475 /* Handle %@ */ 476 if (fmt[0] == '%' && fmt[1] == '@') { 477 sentinelRedisInstance *master = (ri->flags & SRI_MASTER) ? 478 NULL : ri->master; 479 480 if (master) { 481 snprintf(msg, sizeof(msg), "%s %s %s %d @ %s %s %d", 482 sentinelRedisInstanceTypeStr(ri), 483 ri->name, ri->addr->ip, ri->addr->port, 484 master->name, master->addr->ip, master->addr->port); 485 } else { 486 snprintf(msg, sizeof(msg), "%s %s %s %d", 487 sentinelRedisInstanceTypeStr(ri), 488 ri->name, ri->addr->ip, ri->addr->port); 489 } 490 fmt += 2; 491 } else { 492 msg[0] = '\0'; 493 } 494 495 /* Use vsprintf for the rest of the formatting if any. */ 496 if (fmt[0] != '\0') { 497 va_start(ap, fmt); 498 vsnprintf(msg+strlen(msg), sizeof(msg)-strlen(msg), fmt, ap); 499 va_end(ap); 500 } 501 502 /* Log the message if the log level allows it to be logged. */ 503 if (level >= server.verbosity) 504 redisLog(level,"%s %s",type,msg); 505 506 /* Publish the message via Pub/Sub if it's not a debugging one. */ 507 if (level != REDIS_DEBUG) { 508 channel = createStringObject(type,strlen(type)); 509 payload = createStringObject(msg,strlen(msg)); 510 pubsubPublishMessage(channel,payload); 511 decrRefCount(channel); 512 decrRefCount(payload); 513 } 514 515 /* Call the notification script if applicable. */ 516 if (level == REDIS_WARNING && ri != NULL) { 517 sentinelRedisInstance *master = (ri->flags & SRI_MASTER) ? 518 ri : ri->master; 519 if (master->notification_script) { 520 sentinelScheduleScriptExecution(master->notification_script, 521 type,msg,NULL); 522 } 523 } 524 } 525 526 /* ============================ script execution ============================ */ 527 528 /* Release a script job structure and all the associated data. */ 529 void sentinelReleaseScriptJob(sentinelScriptJob *sj) { 530 int j = 0; 531 532 while(sj->argv[j]) sdsfree(sj->argv[j++]); 533 zfree(sj->argv); 534 zfree(sj); 535 } 536 537 #define SENTINEL_SCRIPT_MAX_ARGS 16 538 void sentinelScheduleScriptExecution(char *path, ...) { 539 va_list ap; 540 char *argv[SENTINEL_SCRIPT_MAX_ARGS+1]; 541 int argc = 1; 542 sentinelScriptJob *sj; 543 544 va_start(ap, path); 545 while(argc < SENTINEL_SCRIPT_MAX_ARGS) { 546 argv[argc] = va_arg(ap,char*); 547 if (!argv[argc]) break; 548 argv[argc] = sdsnew(argv[argc]); /* Copy the string. */ 549 argc++; 550 } 551 va_end(ap); 552 argv[0] = sdsnew(path); 553 554 sj = zmalloc(sizeof(*sj)); 555 sj->flags = SENTINEL_SCRIPT_NONE; 556 sj->retry_num = 0; 557 sj->argv = zmalloc(sizeof(char*)*(argc+1)); 558 sj->start_time = 0; 559 sj->pid = 0; 560 memcpy(sj->argv,argv,sizeof(char*)*(argc+1)); 561 562 listAddNodeTail(sentinel.scripts_queue,sj); 563 564 /* Remove the oldest non running script if we already hit the limit. */ 565 if (listLength(sentinel.scripts_queue) > SENTINEL_SCRIPT_MAX_QUEUE) { 566 listNode *ln; 567 listIter li; 568 569 listRewind(sentinel.scripts_queue,&li); 570 while ((ln = listNext(&li)) != NULL) { 571 sj = ln->value; 572 573 if (sj->flags & SENTINEL_SCRIPT_RUNNING) continue; 574 /* The first node is the oldest as we add on tail. */ 575 listDelNode(sentinel.scripts_queue,ln); 576 sentinelReleaseScriptJob(sj); 577 break; 578 } 579 redisAssert(listLength(sentinel.scripts_queue) <= 580 SENTINEL_SCRIPT_MAX_QUEUE); 581 } 582 } 583 584 /* Lookup a script in the scripts queue via pid, and returns the list node 585 * (so that we can easily remove it from the queue if needed). */ 586 listNode *sentinelGetScriptListNodeByPid(pid_t pid) { 587 listNode *ln; 588 listIter li; 589 590 listRewind(sentinel.scripts_queue,&li); 591 while ((ln = listNext(&li)) != NULL) { 592 sentinelScriptJob *sj = ln->value; 593 594 if ((sj->flags & SENTINEL_SCRIPT_RUNNING) && sj->pid == pid) 595 return ln; 596 } 597 return NULL; 598 } 599 600 /* Run pending scripts if we are not already at max number of running 601 * scripts. */ 602 void sentinelRunPendingScripts(void) { 603 listNode *ln; 604 listIter li; 605 mstime_t now = mstime(); 606 607 /* Find jobs that are not running and run them, from the top to the 608 * tail of the queue, so we run older jobs first. */ 609 listRewind(sentinel.scripts_queue,&li); 610 while (sentinel.running_scripts < SENTINEL_SCRIPT_MAX_RUNNING && 611 (ln = listNext(&li)) != NULL) 612 { 613 sentinelScriptJob *sj = ln->value; 614 pid_t pid; 615 616 /* Skip if already running. */ 617 if (sj->flags & SENTINEL_SCRIPT_RUNNING) continue; 618 619 /* Skip if it's a retry, but not enough time has elapsed. */ 620 if (sj->start_time && sj->start_time > now) continue; 621 622 sj->flags |= SENTINEL_SCRIPT_RUNNING; 623 sj->start_time = mstime(); 624 sj->retry_num++; 625 pid = fork(); 626 627 if (pid == -1) { 628 /* Parent (fork error). 629 * We report fork errors as signal 99, in order to unify the 630 * reporting with other kind of errors. */ 631 sentinelEvent(REDIS_WARNING,"-script-error",NULL, 632 "%s %d %d", sj->argv[0], 99, 0); 633 sj->flags &= ~SENTINEL_SCRIPT_RUNNING; 634 sj->pid = 0; 635 } else if (pid == 0) { 636 /* Child */ 637 execve(sj->argv[0],sj->argv,environ); 638 /* If we are here an error occurred. */ 639 _exit(2); /* Don't retry execution. */ 640 } else { 641 sentinel.running_scripts++; 642 sj->pid = pid; 643 sentinelEvent(REDIS_DEBUG,"+script-child",NULL,"%ld",(long)pid); 644 } 645 } 646 } 647 648 /* How much to delay the execution of a script that we need to retry after 649 * an error? 650 * 651 * We double the retry delay for every further retry we do. So for instance 652 * if RETRY_DELAY is set to 30 seconds and the max number of retries is 10 653 * starting from the second attempt to execute the script the delays are: 654 * 30 sec, 60 sec, 2 min, 4 min, 8 min, 16 min, 32 min, 64 min, 128 min. */ 655 mstime_t sentinelScriptRetryDelay(int retry_num) { 656 mstime_t delay = SENTINEL_SCRIPT_RETRY_DELAY; 657 658 while (retry_num-- > 1) delay *= 2; 659 return delay; 660 } 661 662 /* Check for scripts that terminated, and remove them from the queue if the 663 * script terminated successfully. If instead the script was terminated by 664 * a signal, or returned exit code "1", it is scheduled to run again if 665 * the max number of retries did not already elapsed. */ 666 void sentinelCollectTerminatedScripts(void) { 667 int statloc; 668 pid_t pid; 669 670 while ((pid = wait3(&statloc,WNOHANG,NULL)) > 0) { 671 int exitcode = WEXITSTATUS(statloc); 672 int bysignal = 0; 673 listNode *ln; 674 sentinelScriptJob *sj; 675 676 if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc); 677 sentinelEvent(REDIS_DEBUG,"-script-child",NULL,"%ld %d %d", 678 (long)pid, exitcode, bysignal); 679 680 ln = sentinelGetScriptListNodeByPid(pid); 681 if (ln == NULL) { 682 redisLog(REDIS_WARNING,"wait3() returned a pid (%ld) we can't find in our scripts execution queue!", (long)pid); 683 continue; 684 } 685 sj = ln->value; 686 687 /* If the script was terminated by a signal or returns an 688 * exit code of "1" (that means: please retry), we reschedule it 689 * if the max number of retries is not already reached. */ 690 if ((bysignal || exitcode == 1) && 691 sj->retry_num != SENTINEL_SCRIPT_MAX_RETRY) 692 { 693 sj->flags &= ~SENTINEL_SCRIPT_RUNNING; 694 sj->pid = 0; 695 sj->start_time = mstime() + 696 sentinelScriptRetryDelay(sj->retry_num); 697 } else { 698 /* Otherwise let's remove the script, but log the event if the 699 * execution did not terminated in the best of the ways. */ 700 if (bysignal || exitcode != 0) { 701 sentinelEvent(REDIS_WARNING,"-script-error",NULL, 702 "%s %d %d", sj->argv[0], bysignal, exitcode); 703 } 704 listDelNode(sentinel.scripts_queue,ln); 705 sentinelReleaseScriptJob(sj); 706 sentinel.running_scripts--; 707 } 708 } 709 } 710 711 /* Kill scripts in timeout, they'll be collected by the 712 * sentinelCollectTerminatedScripts() function. */ 713 void sentinelKillTimedoutScripts(void) { 714 listNode *ln; 715 listIter li; 716 mstime_t now = mstime(); 717 718 listRewind(sentinel.scripts_queue,&li); 719 while ((ln = listNext(&li)) != NULL) { 720 sentinelScriptJob *sj = ln->value; 721 722 if (sj->flags & SENTINEL_SCRIPT_RUNNING && 723 (now - sj->start_time) > SENTINEL_SCRIPT_MAX_RUNTIME) 724 { 725 sentinelEvent(REDIS_WARNING,"-script-timeout",NULL,"%s %ld", 726 sj->argv[0], (long)sj->pid); 727 kill(sj->pid,SIGKILL); 728 } 729 } 730 } 731 732 /* Implements SENTINEL PENDING-SCRIPTS command. */ 733 void sentinelPendingScriptsCommand(redisClient *c) { 734 listNode *ln; 735 listIter li; 736 737 addReplyMultiBulkLen(c,listLength(sentinel.scripts_queue)); 738 listRewind(sentinel.scripts_queue,&li); 739 while ((ln = listNext(&li)) != NULL) { 740 sentinelScriptJob *sj = ln->value; 741 int j = 0; 742 743 addReplyMultiBulkLen(c,10); 744 745 addReplyBulkCString(c,"argv"); 746 while (sj->argv[j]) j++; 747 addReplyMultiBulkLen(c,j); 748 j = 0; 749 while (sj->argv[j]) addReplyBulkCString(c,sj->argv[j++]); 750 751 addReplyBulkCString(c,"flags"); 752 addReplyBulkCString(c, 753 (sj->flags & SENTINEL_SCRIPT_RUNNING) ? "running" : "scheduled"); 754 755 addReplyBulkCString(c,"pid"); 756 addReplyBulkLongLong(c,sj->pid); 757 758 if (sj->flags & SENTINEL_SCRIPT_RUNNING) { 759 addReplyBulkCString(c,"run-time"); 760 addReplyBulkLongLong(c,mstime() - sj->start_time); 761 } else { 762 mstime_t delay = sj->start_time ? (sj->start_time-mstime()) : 0; 763 if (delay < 0) delay = 0; 764 addReplyBulkCString(c,"run-delay"); 765 addReplyBulkLongLong(c,delay); 766 } 767 768 addReplyBulkCString(c,"retry-num"); 769 addReplyBulkLongLong(c,sj->retry_num); 770 } 771 } 772 773 /* This function calls, if any, the client reconfiguration script with the 774 * following parameters: 775 * 776 * <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port> 777 * 778 * It is called every time a failover starts, ends, or is aborted. 779 * 780 * <state> is "start", "end" or "abort". 781 * <role> is either "leader" or "observer". 782 * 783 * from/to fields are respectively master -> promoted slave addresses for 784 * "start" and "end", or the reverse (promoted slave -> master) in case of 785 * "abort". 786 */ 787 void sentinelCallClientReconfScript(sentinelRedisInstance *master, int role, char *state, sentinelAddr *from, sentinelAddr *to) { 788 char fromport[32], toport[32]; 789 790 if (master->client_reconfig_script == NULL) return; 791 ll2string(fromport,sizeof(fromport),from->port); 792 ll2string(toport,sizeof(toport),to->port); 793 sentinelScheduleScriptExecution(master->client_reconfig_script, 794 master->name, 795 (role == SENTINEL_LEADER) ? "leader" : "observer", 796 state, from->ip, fromport, to->ip, toport, NULL); 797 } 798 799 /* ========================== sentinelRedisInstance ========================= */ 800 801 /* Create a redis instance, the following fields must be populated by the 802 * caller if needed: 803 * runid: set to NULL but will be populated once INFO output is received. 804 * info_refresh: is set to 0 to mean that we never received INFO so far. 805 * 806 * If SRI_MASTER is set into initial flags the instance is added to 807 * sentinel.masters table. 808 * 809 * if SRI_SLAVE or SRI_SENTINEL is set then 'master' must be not NULL and the 810 * instance is added into master->slaves or master->sentinels table. 811 * 812 * If the instance is a slave or sentinel, the name parameter is ignored and 813 * is created automatically as hostname:port. 814 * 815 * The function fails if hostname can't be resolved or port is out of range. 816 * When this happens NULL is returned and errno is set accordingly to the 817 * createSentinelAddr() function. 818 * 819 * The function may also fail and return NULL with errno set to EBUSY if 820 * a master or slave with the same name already exists. */ 821 sentinelRedisInstance *createSentinelRedisInstance(char *name, int flags, char *hostname, int port, int quorum, sentinelRedisInstance *master) { 822 sentinelRedisInstance *ri; 823 sentinelAddr *addr; 824 dict *table = NULL; 825 char slavename[128], *sdsname; 826 827 redisAssert(flags & (SRI_MASTER|SRI_SLAVE|SRI_SENTINEL)); 828 redisAssert((flags & SRI_MASTER) || master != NULL); 829 830 /* Check address validity. */ 831 addr = createSentinelAddr(hostname,port); 832 if (addr == NULL) return NULL; 833 834 /* For slaves and sentinel we use ip:port as name. */ 835 if (flags & (SRI_SLAVE|SRI_SENTINEL)) { 836 snprintf(slavename,sizeof(slavename),"%s:%d",hostname,port); 837 name = slavename; 838 } 839 840 /* Make sure the entry is not duplicated. This may happen when the same 841 * name for a master is used multiple times inside the configuration or 842 * if we try to add multiple times a slave or sentinel with same ip/port 843 * to a master. */ 844 if (flags & SRI_MASTER) table = sentinel.masters; 845 else if (flags & SRI_SLAVE) table = master->slaves; 846 else if (flags & SRI_SENTINEL) table = master->sentinels; 847 sdsname = sdsnew(name); 848 if (dictFind(table,sdsname)) { 849 sdsfree(sdsname); 850 errno = EBUSY; 851 return NULL; 852 } 853 854 /* Create the instance object. */ 855 ri = zmalloc(sizeof(*ri)); 856 /* Note that all the instances are started in the disconnected state, 857 * the event loop will take care of connecting them. */ 858 ri->flags = flags | SRI_DISCONNECTED; 859 ri->name = sdsname; 860 ri->runid = NULL; 861 ri->addr = addr; 862 ri->cc = NULL; 863 ri->pc = NULL; 864 ri->pending_commands = 0; 865 ri->cc_conn_time = 0; 866 ri->pc_conn_time = 0; 867 ri->pc_last_activity = 0; 868 ri->last_avail_time = mstime(); 869 ri->last_pong_time = mstime(); 870 ri->last_pub_time = mstime(); 871 ri->last_hello_time = mstime(); 872 ri->last_master_down_reply_time = mstime(); 873 ri->s_down_since_time = 0; 874 ri->o_down_since_time = 0; 875 ri->down_after_period = master ? master->down_after_period : 876 SENTINEL_DOWN_AFTER_PERIOD; 877 ri->master_link_down_time = 0; 878 ri->auth_pass = NULL; 879 ri->slave_priority = SENTINEL_DEFAULT_SLAVE_PRIORITY; 880 ri->slave_reconf_sent_time = 0; 881 ri->slave_master_host = NULL; 882 ri->slave_master_port = 0; 883 ri->slave_master_link_status = SENTINEL_MASTER_LINK_STATUS_DOWN; 884 ri->sentinels = dictCreate(&instancesDictType,NULL); 885 ri->quorum = quorum; 886 ri->parallel_syncs = SENTINEL_DEFAULT_PARALLEL_SYNCS; 887 ri->master = master; 888 ri->slaves = dictCreate(&instancesDictType,NULL); 889 ri->info_refresh = 0; 890 891 /* Failover state. */ 892 ri->leader = NULL; 893 ri->failover_state = SENTINEL_FAILOVER_STATE_NONE; 894 ri->failover_state_change_time = 0; 895 ri->failover_start_time = 0; 896 ri->failover_timeout = SENTINEL_DEFAULT_FAILOVER_TIMEOUT; 897 ri->promoted_slave = NULL; 898 ri->notification_script = NULL; 899 ri->client_reconfig_script = NULL; 900 901 /* Add into the right table. */ 902 dictAdd(table, ri->name, ri); 903 return ri; 904 } 905 906 /* Release this instance and all its slaves, sentinels, hiredis connections. 907 * This function also takes care of unlinking the instance from the main 908 * masters table (if it is a master) or from its master sentinels/slaves table 909 * if it is a slave or sentinel. */ 910 void releaseSentinelRedisInstance(sentinelRedisInstance *ri) { 911 /* Release all its slaves or sentinels if any. */ 912 dictRelease(ri->sentinels); 913 dictRelease(ri->slaves); 914 915 /* Release hiredis connections. */ 916 if (ri->cc) sentinelKillLink(ri,ri->cc); 917 if (ri->pc) sentinelKillLink(ri,ri->pc); 918 919 /* Free other resources. */ 920 sdsfree(ri->name); 921 sdsfree(ri->runid); 922 sdsfree(ri->notification_script); 923 sdsfree(ri->client_reconfig_script); 924 sdsfree(ri->slave_master_host); 925 sdsfree(ri->leader); 926 sdsfree(ri->auth_pass); 927 releaseSentinelAddr(ri->addr); 928 929 /* Clear state into the master if needed. */ 930 if ((ri->flags & SRI_SLAVE) && (ri->flags & SRI_PROMOTED) && ri->master) 931 ri->master->promoted_slave = NULL; 932 933 zfree(ri); 934 } 935 936 /* Lookup a slave in a master Redis instance, by ip and port. */ 937 sentinelRedisInstance *sentinelRedisInstanceLookupSlave( 938 sentinelRedisInstance *ri, char *ip, int port) 939 { 940 sds key; 941 sentinelRedisInstance *slave; 942 943 redisAssert(ri->flags & SRI_MASTER); 944 key = sdscatprintf(sdsempty(),"%s:%d",ip,port); 945 slave = dictFetchValue(ri->slaves,key); 946 sdsfree(key); 947 return slave; 948 } 949 950 /* Return the name of the type of the instance as a string. */ 951 const char *sentinelRedisInstanceTypeStr(sentinelRedisInstance *ri) { 952 if (ri->flags & SRI_MASTER) return "master"; 953 else if (ri->flags & SRI_SLAVE) return "slave"; 954 else if (ri->flags & SRI_SENTINEL) return "sentinel"; 955 else return "unknown"; 956 } 957 958 /* This function removes all the instances found in the dictionary of instances 959 * 'd', having either: 960 * 961 * 1) The same ip/port as specified. 962 * 2) The same runid. 963 * 964 * "1" and "2" don't need to verify at the same time, just one is enough. 965 * If "runid" is NULL it is not checked. 966 * Similarly if "ip" is NULL it is not checked. 967 * 968 * This function is useful because every time we add a new Sentinel into 969 * a master's Sentinels dictionary, we want to be very sure about not 970 * having duplicated instances for any reason. This is so important because 971 * we use those other sentinels in order to run our quorum protocol to 972 * understand if it's time to proceed with the fail over. 973 * 974 * Making sure no duplication is possible we greatly improve the robustness 975 * of the quorum (otherwise we may end counting the same instance multiple 976 * times for some reason). 977 * 978 * The function returns the number of Sentinels removed. */ 979 int removeMatchingSentinelsFromMaster(sentinelRedisInstance *master, char *ip, int port, char *runid) { 980 dictIterator *di; 981 dictEntry *de; 982 int removed = 0; 983 984 di = dictGetSafeIterator(master->sentinels); 985 while((de = dictNext(di)) != NULL) { 986 sentinelRedisInstance *ri = dictGetVal(de); 987 988 if ((ri->runid && runid && strcmp(ri->runid,runid) == 0) || 989 (ip && strcmp(ri->addr->ip,ip) == 0 && port == ri->addr->port)) 990 { 991 dictDelete(master->sentinels,ri->name); 992 removed++; 993 } 994 } 995 dictReleaseIterator(di); 996 return removed; 997 } 998 999 /* Search an instance with the same runid, ip and port into a dictionary 1000 * of instances. Return NULL if not found, otherwise return the instance 1001 * pointer. 1002 * 1003 * runid or ip can be NULL. In such a case the search is performed only 1004 * by the non-NULL field. */ 1005 sentinelRedisInstance *getSentinelRedisInstanceByAddrAndRunID(dict *instances, char *ip, int port, char *runid) { 1006 dictIterator *di; 1007 dictEntry *de; 1008 sentinelRedisInstance *instance = NULL; 1009 1010 redisAssert(ip || runid); /* User must pass at least one search param. */ 1011 di = dictGetIterator(instances); 1012 while((de = dictNext(di)) != NULL) { 1013 sentinelRedisInstance *ri = dictGetVal(de); 1014 1015 if (runid && !ri->runid) continue; 1016 if ((runid == NULL || strcmp(ri->runid, runid) == 0) && 1017 (ip == NULL || (strcmp(ri->addr->ip, ip) == 0 && 1018 ri->addr->port == port))) 1019 { 1020 instance = ri; 1021 break; 1022 } 1023 } 1024 dictReleaseIterator(di); 1025 return instance; 1026 } 1027 1028 /* Simple master lookup by name */ 1029 sentinelRedisInstance *sentinelGetMasterByName(char *name) { 1030 sentinelRedisInstance *ri; 1031 sds sdsname = sdsnew(name); 1032 1033 ri = dictFetchValue(sentinel.masters,sdsname); 1034 sdsfree(sdsname); 1035 return ri; 1036 } 1037 1038 /* Add the specified flags to all the instances in the specified dictionary. */ 1039 void sentinelAddFlagsToDictOfRedisInstances(dict *instances, int flags) { 1040 dictIterator *di; 1041 dictEntry *de; 1042 1043 di = dictGetIterator(instances); 1044 while((de = dictNext(di)) != NULL) { 1045 sentinelRedisInstance *ri = dictGetVal(de); 1046 ri->flags |= flags; 1047 } 1048 dictReleaseIterator(di); 1049 } 1050 1051 /* Remove the specified flags to all the instances in the specified 1052 * dictionary. */ 1053 void sentinelDelFlagsToDictOfRedisInstances(dict *instances, int flags) { 1054 dictIterator *di; 1055 dictEntry *de; 1056 1057 di = dictGetIterator(instances); 1058 while((de = dictNext(di)) != NULL) { 1059 sentinelRedisInstance *ri = dictGetVal(de); 1060 ri->flags &= ~flags; 1061 } 1062 dictReleaseIterator(di); 1063 } 1064 1065 /* Reset the state of a monitored master: 1066 * 1) Remove all slaves. 1067 * 2) Remove all sentinels. 1068 * 3) Remove most of the flags resulting from runtime operations. 1069 * 4) Reset timers to their default value. 1070 * 5) In the process of doing this undo the failover if in progress. 1071 * 6) Disconnect the connections with the master (will reconnect automatically). 1072 */ 1073 void sentinelResetMaster(sentinelRedisInstance *ri, int flags) { 1074 redisAssert(ri->flags & SRI_MASTER); 1075 dictRelease(ri->slaves); 1076 dictRelease(ri->sentinels); 1077 ri->slaves = dictCreate(&instancesDictType,NULL); 1078 ri->sentinels = dictCreate(&instancesDictType,NULL); 1079 if (ri->cc) sentinelKillLink(ri,ri->cc); 1080 if (ri->pc) sentinelKillLink(ri,ri->pc); 1081 ri->flags &= SRI_MASTER|SRI_CAN_FAILOVER|SRI_DISCONNECTED; 1082 if (ri->leader) { 1083 sdsfree(ri->leader); 1084 ri->leader = NULL; 1085 } 1086 ri->failover_state = SENTINEL_FAILOVER_STATE_NONE; 1087 ri->failover_state_change_time = 0; 1088 ri->failover_start_time = 0; 1089 ri->promoted_slave = NULL; 1090 sdsfree(ri->runid); 1091 sdsfree(ri->slave_master_host); 1092 ri->runid = NULL; 1093 ri->slave_master_host = NULL; 1094 ri->last_avail_time = mstime(); 1095 ri->last_pong_time = mstime(); 1096 if (flags & SENTINEL_GENERATE_EVENT) 1097 sentinelEvent(REDIS_WARNING,"+reset-master",ri,"%@"); 1098 } 1099 1100 /* Call sentinelResetMaster() on every master with a name matching the specified 1101 * pattern. */ 1102 int sentinelResetMastersByPattern(char *pattern, int flags) { 1103 dictIterator *di; 1104 dictEntry *de; 1105 int reset = 0; 1106 1107 di = dictGetIterator(sentinel.masters); 1108 while((de = dictNext(di)) != NULL) { 1109 sentinelRedisInstance *ri = dictGetVal(de); 1110 1111 if (ri->name) { 1112 if (stringmatch(pattern,ri->name,0)) { 1113 sentinelResetMaster(ri,flags); 1114 reset++; 1115 } 1116 } 1117 } 1118 dictReleaseIterator(di); 1119 return reset; 1120 } 1121 1122 /* Reset the specified master with sentinelResetMaster(), and also change 1123 * the ip:port address, but take the name of the instance unmodified. 1124 * 1125 * This is used to handle the +switch-master and +redirect-to-master events. 1126 * 1127 * The function returns REDIS_ERR if the address can't be resolved for some 1128 * reason. Otherwise REDIS_OK is returned. 1129 * 1130 * TODO: make this reset so that original sentinels are re-added with 1131 * same ip / port / runid. 1132 */ 1133 1134 int sentinelResetMasterAndChangeAddress(sentinelRedisInstance *master, char *ip, int port) { 1135 sentinelAddr *oldaddr, *newaddr; 1136 1137 newaddr = createSentinelAddr(ip,port); 1138 if (newaddr == NULL) return REDIS_ERR; 1139 sentinelResetMaster(master,SENTINEL_NO_FLAGS); 1140 oldaddr = master->addr; 1141 master->addr = newaddr; 1142 /* Release the old address at the end so we are safe even if the function 1143 * gets the master->addr->ip and master->addr->port as arguments. */ 1144 releaseSentinelAddr(oldaddr); 1145 return REDIS_OK; 1146 } 1147 1148 /* ============================ Config handling ============================= */ 1149 char *sentinelHandleConfiguration(char **argv, int argc) { 1150 sentinelRedisInstance *ri; 1151 1152 if (!strcasecmp(argv[0],"monitor") && argc == 5) { 1153 /* monitor <name> <host> <port> <quorum> */ 1154 int quorum = atoi(argv[4]); 1155 1156 if (quorum <= 0) return "Quorum must be 1 or greater."; 1157 if (createSentinelRedisInstance(argv[1],SRI_MASTER,argv[2], 1158 atoi(argv[3]),quorum,NULL) == NULL) 1159 { 1160 switch(errno) { 1161 case EBUSY: return "Duplicated master name."; 1162 case ENOENT: return "Can't resolve master instance hostname."; 1163 case EINVAL: return "Invalid port number"; 1164 } 1165 } 1166 } else if (!strcasecmp(argv[0],"down-after-milliseconds") && argc == 3) { 1167 /* down-after-milliseconds <name> <milliseconds> */ 1168 ri = sentinelGetMasterByName(argv[1]); 1169 if (!ri) return "No such master with specified name."; 1170 ri->down_after_period = atoi(argv[2]); 1171 if (ri->down_after_period <= 0) 1172 return "negative or zero time parameter."; 1173 } else if (!strcasecmp(argv[0],"failover-timeout") && argc == 3) { 1174 /* failover-timeout <name> <milliseconds> */ 1175 ri = sentinelGetMasterByName(argv[1]); 1176 if (!ri) return "No such master with specified name."; 1177 ri->failover_timeout = atoi(argv[2]); 1178 if (ri->failover_timeout <= 0) 1179 return "negative or zero time parameter."; 1180 } else if (!strcasecmp(argv[0],"can-failover") && argc == 3) { 1181 /* can-failover <name> <yes/no> */ 1182 int yesno = yesnotoi(argv[2]); 1183 1184 ri = sentinelGetMasterByName(argv[1]); 1185 if (!ri) return "No such master with specified name."; 1186 if (yesno == -1) return "Argument must be either yes or no."; 1187 if (yesno) 1188 ri->flags |= SRI_CAN_FAILOVER; 1189 else 1190 ri->flags &= ~SRI_CAN_FAILOVER; 1191 } else if (!strcasecmp(argv[0],"parallel-syncs") && argc == 3) { 1192 /* parallel-syncs <name> <milliseconds> */ 1193 ri = sentinelGetMasterByName(argv[1]); 1194 if (!ri) return "No such master with specified name."; 1195 ri->parallel_syncs = atoi(argv[2]); 1196 } else if (!strcasecmp(argv[0],"notification-script") && argc == 3) { 1197 /* notification-script <name> <path> */ 1198 ri = sentinelGetMasterByName(argv[1]); 1199 if (!ri) return "No such master with specified name."; 1200 if (access(argv[2],X_OK) == -1) 1201 return "Notification script seems non existing or non executable."; 1202 ri->notification_script = sdsnew(argv[2]); 1203 } else if (!strcasecmp(argv[0],"client-reconfig-script") && argc == 3) { 1204 /* client-reconfig-script <name> <path> */ 1205 ri = sentinelGetMasterByName(argv[1]); 1206 if (!ri) return "No such master with specified name."; 1207 if (access(argv[2],X_OK) == -1) 1208 return "Client reconfiguration script seems non existing or " 1209 "non executable."; 1210 ri->client_reconfig_script = sdsnew(argv[2]); 1211 } else if (!strcasecmp(argv[0],"auth-pass") && argc == 3) { 1212 /* auth-pass <name> <password> */ 1213 ri = sentinelGetMasterByName(argv[1]); 1214 if (!ri) return "No such master with specified name."; 1215 ri->auth_pass = sdsnew(argv[2]); 1216 } else { 1217 return "Unrecognized sentinel configuration statement."; 1218 } 1219 return NULL; 1220 } 1221 1222 /* ====================== hiredis connection handling ======================= */ 1223 1224 /* Completely disconnect an hiredis link from an instance. */ 1225 void sentinelKillLink(sentinelRedisInstance *ri, redisAsyncContext *c) { 1226 if (ri->cc == c) { 1227 ri->cc = NULL; 1228 ri->pending_commands = 0; 1229 } 1230 if (ri->pc == c) ri->pc = NULL; 1231 c->data = NULL; 1232 ri->flags |= SRI_DISCONNECTED; 1233 redisAsyncFree(c); 1234 } 1235 1236 /* This function takes an hiredis context that is in an error condition 1237 * and make sure to mark the instance as disconnected performing the 1238 * cleanup needed. 1239 * 1240 * Note: we don't free the hiredis context as hiredis will do it for us 1241 * for async connections. */ 1242 void sentinelDisconnectInstanceFromContext(const redisAsyncContext *c) { 1243 sentinelRedisInstance *ri = c->data; 1244 int pubsub; 1245 1246 if (ri == NULL) return; /* The instance no longer exists. */ 1247 1248 pubsub = (ri->pc == c); 1249 sentinelEvent(REDIS_DEBUG, pubsub ? "-pubsub-link" : "-cmd-link", ri, 1250 "%@ #%s", c->errstr); 1251 if (pubsub) 1252 ri->pc = NULL; 1253 else 1254 ri->cc = NULL; 1255 ri->flags |= SRI_DISCONNECTED; 1256 } 1257 1258 void sentinelLinkEstablishedCallback(const redisAsyncContext *c, int status) { 1259 if (status != REDIS_OK) { 1260 sentinelDisconnectInstanceFromContext(c); 1261 } else { 1262 sentinelRedisInstance *ri = c->data; 1263 int pubsub = (ri->pc == c); 1264 1265 sentinelEvent(REDIS_DEBUG, pubsub ? "+pubsub-link" : "+cmd-link", ri, 1266 "%@"); 1267 } 1268 } 1269 1270 void sentinelDisconnectCallback(const redisAsyncContext *c, int status) { 1271 sentinelDisconnectInstanceFromContext(c); 1272 } 1273 1274 /* Send the AUTH command with the specified master password if needed. 1275 * Note that for slaves the password set for the master is used. 1276 * 1277 * We don't check at all if the command was successfully transmitted 1278 * to the instance as if it fails Sentinel will detect the instance down, 1279 * will disconnect and reconnect the link and so forth. */ 1280 void sentinelSendAuthIfNeeded(sentinelRedisInstance *ri, redisAsyncContext *c) { 1281 char *auth_pass = (ri->flags & SRI_MASTER) ? ri->auth_pass : 1282 ri->master->auth_pass; 1283 1284 if (auth_pass) 1285 redisAsyncCommand(c, sentinelDiscardReplyCallback, NULL, "AUTH %s", 1286 auth_pass); 1287 } 1288 1289 /* Create the async connections for the specified instance if the instance 1290 * is disconnected. Note that the SRI_DISCONNECTED flag is set even if just 1291 * one of the two links (commands and pub/sub) is missing. */ 1292 void sentinelReconnectInstance(sentinelRedisInstance *ri) { 1293 if (!(ri->flags & SRI_DISCONNECTED)) return; 1294 1295 /* Commands connection. */ 1296 if (ri->cc == NULL) { 1297 ri->cc = redisAsyncConnect(ri->addr->ip,ri->addr->port); 1298 if (ri->cc->err) { 1299 sentinelEvent(REDIS_DEBUG,"-cmd-link-reconnection",ri,"%@ #%s", 1300 ri->cc->errstr); 1301 sentinelKillLink(ri,ri->cc); 1302 } else { 1303 ri->cc_conn_time = mstime(); 1304 ri->cc->data = ri; 1305 redisAeAttach(server.el,ri->cc); 1306 redisAsyncSetConnectCallback(ri->cc, 1307 sentinelLinkEstablishedCallback); 1308 redisAsyncSetDisconnectCallback(ri->cc, 1309 sentinelDisconnectCallback); 1310 sentinelSendAuthIfNeeded(ri,ri->cc); 1311 } 1312 } 1313 /* Pub / Sub */ 1314 if ((ri->flags & SRI_MASTER) && ri->pc == NULL) { 1315 ri->pc = redisAsyncConnect(ri->addr->ip,ri->addr->port); 1316 if (ri->pc->err) { 1317 sentinelEvent(REDIS_DEBUG,"-pubsub-link-reconnection",ri,"%@ #%s", 1318 ri->pc->errstr); 1319 sentinelKillLink(ri,ri->pc); 1320 } else { 1321 int retval; 1322 1323 ri->pc_conn_time = mstime(); 1324 ri->pc->data = ri; 1325 redisAeAttach(server.el,ri->pc); 1326 redisAsyncSetConnectCallback(ri->pc, 1327 sentinelLinkEstablishedCallback); 1328 redisAsyncSetDisconnectCallback(ri->pc, 1329 sentinelDisconnectCallback); 1330 sentinelSendAuthIfNeeded(ri,ri->pc); 1331 /* Now we subscribe to the Sentinels "Hello" channel. */ 1332 retval = redisAsyncCommand(ri->pc, 1333 sentinelReceiveHelloMessages, NULL, "SUBSCRIBE %s", 1334 SENTINEL_HELLO_CHANNEL); 1335 if (retval != REDIS_OK) { 1336 /* If we can't subscribe, the Pub/Sub connection is useless 1337 * and we can simply disconnect it and try again. */ 1338 sentinelKillLink(ri,ri->pc); 1339 return; 1340 } 1341 } 1342 } 1343 /* Clear the DISCONNECTED flags only if we have both the connections 1344 * (or just the commands connection if this is a slave or a 1345 * sentinel instance). */ 1346 if (ri->cc && (ri->flags & (SRI_SLAVE|SRI_SENTINEL) || ri->pc)) 1347 ri->flags &= ~SRI_DISCONNECTED; 1348 } 1349 1350 /* ======================== Redis instances pinging ======================== */ 1351 1352 /* Process the INFO output from masters. */ 1353 void sentinelRefreshInstanceInfo(sentinelRedisInstance *ri, const char *info) { 1354 sds *lines; 1355 int numlines, j; 1356 int role = 0; 1357 int runid_changed = 0; /* true if runid changed. */ 1358 int first_runid = 0; /* true if this is the first runid we receive. */ 1359 1360 /* The following fields must be reset to a given value in the case they 1361 * are not found at all in the INFO output. */ 1362 ri->master_link_down_time = 0; 1363 1364 /* Process line by line. */ 1365 lines = sdssplitlen(info,strlen(info),"\r\n",2,&numlines); 1366 for (j = 0; j < numlines; j++) { 1367 sentinelRedisInstance *slave; 1368 sds l = lines[j]; 1369 1370 /* run_id:<40 hex chars>*/ 1371 if (sdslen(l) >= 47 && !memcmp(l,"run_id:",7)) { 1372 if (ri->runid == NULL) { 1373 ri->runid = sdsnewlen(l+7,40); 1374 first_runid = 1; 1375 } else { 1376 if (strncmp(ri->runid,l+7,40) != 0) { 1377 runid_changed = 1; 1378 sentinelEvent(REDIS_NOTICE,"+reboot",ri,"%@"); 1379 sdsfree(ri->runid); 1380 ri->runid = sdsnewlen(l+7,40); 1381 } 1382 } 1383 } 1384 1385 /* slave0:<ip>,<port>,<state> */ 1386 if ((ri->flags & SRI_MASTER) && 1387 sdslen(l) >= 7 && 1388 !memcmp(l,"slave",5) && isdigit(l[5])) 1389 { 1390 char *ip, *port, *end; 1391 1392 ip = strchr(l,':'); if (!ip) continue; 1393 ip++; /* Now ip points to start of ip address. */ 1394 port = strchr(ip,','); if (!port) continue; 1395 *port = '\0'; /* nul term for easy access. */ 1396 port++; /* Now port points to start of port number. */ 1397 end = strchr(port,','); if (!end) continue; 1398 *end = '\0'; /* nul term for easy access. */ 1399 1400 /* Check if we already have this slave into our table, 1401 * otherwise add it. */ 1402 if (sentinelRedisInstanceLookupSlave(ri,ip,atoi(port)) == NULL) { 1403 if ((slave = createSentinelRedisInstance(NULL,SRI_SLAVE,ip, 1404 atoi(port), ri->quorum,ri)) != NULL) 1405 { 1406 sentinelEvent(REDIS_NOTICE,"+slave",slave,"%@"); 1407 } 1408 } 1409 } 1410 1411 /* master_link_down_since_seconds:<seconds> */ 1412 if (sdslen(l) >= 32 && 1413 !memcmp(l,"master_link_down_since_seconds",30)) 1414 { 1415 ri->master_link_down_time = strtoll(l+31,NULL,10)*1000; 1416 } 1417 1418 /* role:<role> */ 1419 if (!memcmp(l,"role:master",11)) role = SRI_MASTER; 1420 else if (!memcmp(l,"role:slave",10)) role = SRI_SLAVE; 1421 1422 if (role == SRI_SLAVE) { 1423 /* master_host:<host> */ 1424 if (sdslen(l) >= 12 && !memcmp(l,"master_host:",12)) { 1425 sdsfree(ri->slave_master_host); 1426 ri->slave_master_host = sdsnew(l+12); 1427 } 1428 1429 /* master_port:<port> */ 1430 if (sdslen(l) >= 12 && !memcmp(l,"master_port:",12)) 1431 ri->slave_master_port = atoi(l+12); 1432 1433 /* master_link_status:<status> */ 1434 if (sdslen(l) >= 19 && !memcmp(l,"master_link_status:",19)) { 1435 ri->slave_master_link_status = 1436 (strcasecmp(l+19,"up") == 0) ? 1437 SENTINEL_MASTER_LINK_STATUS_UP : 1438 SENTINEL_MASTER_LINK_STATUS_DOWN; 1439 } 1440 1441 /* slave_priority:<priority> */ 1442 if (sdslen(l) >= 15 && !memcmp(l,"slave_priority:",15)) 1443 ri->slave_priority = atoi(l+15); 1444 } 1445 } 1446 ri->info_refresh = mstime(); 1447 sdsfreesplitres(lines,numlines); 1448 1449 /* ---------------------------- Acting half ----------------------------- */ 1450 if (sentinel.tilt) return; 1451 1452 /* Act if a master turned into a slave. */ 1453 if ((ri->flags & SRI_MASTER) && role == SRI_SLAVE) { 1454 if ((first_runid || runid_changed) && ri->slave_master_host) { 1455 /* If it is the first time we receive INFO from it, but it's 1456 * a slave while it was configured as a master, we want to monitor 1457 * its master instead. */ 1458 sentinelEvent(REDIS_WARNING,"+redirect-to-master",ri, 1459 "%s %s %d %s %d", 1460 ri->name, ri->addr->ip, ri->addr->port, 1461 ri->slave_master_host, ri->slave_master_port); 1462 sentinelResetMasterAndChangeAddress(ri,ri->slave_master_host, 1463 ri->slave_master_port); 1464 return; 1465 } 1466 } 1467 1468 /* Act if a slave turned into a master. */ 1469 if ((ri->flags & SRI_SLAVE) && role == SRI_MASTER) { 1470 if (!(ri->master->flags & SRI_FAILOVER_IN_PROGRESS) && 1471 (runid_changed || first_runid)) 1472 { 1473 /* If a slave turned into master but: 1474 * 1475 * 1) Failover not in progress. 1476 * 2) RunID hs changed, or its the first time we see an INFO output. 1477 * 1478 * We assume this is a reboot with a wrong configuration. 1479 * Log the event and remove the slave. */ 1480 int retval; 1481 1482 sentinelEvent(REDIS_WARNING,"-slave-restart-as-master",ri,"%@ #removing it from the attached slaves"); 1483 retval = dictDelete(ri->master->slaves,ri->name); 1484 redisAssert(retval == REDIS_OK); 1485 return; 1486 } else if (ri->flags & SRI_PROMOTED) { 1487 /* If this is a promoted slave we can change state to the 1488 * failover state machine. */ 1489 if ((ri->master->flags & SRI_FAILOVER_IN_PROGRESS) && 1490 (ri->master->flags & SRI_I_AM_THE_LEADER) && 1491 (ri->master->failover_state == 1492 SENTINEL_FAILOVER_STATE_WAIT_PROMOTION)) 1493 { 1494 ri->master->failover_state = SENTINEL_FAILOVER_STATE_RECONF_SLAVES; 1495 ri->master->failover_state_change_time = mstime(); 1496 sentinelEvent(REDIS_WARNING,"+promoted-slave",ri,"%@"); 1497 sentinelEvent(REDIS_WARNING,"+failover-state-reconf-slaves", 1498 ri->master,"%@"); 1499 sentinelCallClientReconfScript(ri->master,SENTINEL_LEADER, 1500 "start",ri->master->addr,ri->addr); 1501 } 1502 } else if (!(ri->master->flags & SRI_FAILOVER_IN_PROGRESS) || 1503 ((ri->master->flags & SRI_FAILOVER_IN_PROGRESS) && 1504 (ri->master->flags & SRI_I_AM_THE_LEADER) && 1505 ri->master->failover_state == 1506 SENTINEL_FAILOVER_STATE_WAIT_START)) 1507 { 1508 /* No failover in progress? Then it is the start of a failover 1509 * and we are an observer. 1510 * 1511 * We also do that if we are a leader doing a failover, in wait 1512 * start, but well, somebody else started before us. */ 1513 1514 if (ri->master->flags & SRI_FAILOVER_IN_PROGRESS) { 1515 sentinelEvent(REDIS_WARNING,"-failover-abort-race", 1516 ri->master, "%@"); 1517 sentinelAbortFailover(ri->master); 1518 } 1519 1520 ri->master->flags |= SRI_FAILOVER_IN_PROGRESS; 1521 sentinelEvent(REDIS_WARNING,"+failover-detected",ri->master,"%@"); 1522 ri->master->failover_state = SENTINEL_FAILOVER_STATE_DETECT_END; 1523 ri->master->failover_state_change_time = mstime(); 1524 ri->master->promoted_slave = ri; 1525 ri->flags |= SRI_PROMOTED; 1526 sentinelCallClientReconfScript(ri->master,SENTINEL_OBSERVER, 1527 "start", ri->master->addr,ri->addr); 1528 /* We are an observer, so we can only assume that the leader 1529 * is reconfiguring the slave instances. For this reason we 1530 * set all the instances as RECONF_SENT waiting for progresses 1531 * on this side. */ 1532 sentinelAddFlagsToDictOfRedisInstances(ri->master->slaves, 1533 SRI_RECONF_SENT); 1534 } 1535 } 1536 1537 /* Detect if the slave that is in the process of being reconfigured 1538 * changed state. */ 1539 if ((ri->flags & SRI_SLAVE) && role == SRI_SLAVE && 1540 (ri->flags & (SRI_RECONF_SENT|SRI_RECONF_INPROG))) 1541 { 1542 /* SRI_RECONF_SENT -> SRI_RECONF_INPROG. */ 1543 if ((ri->flags & SRI_RECONF_SENT) && 1544 ri->slave_master_host && 1545 strcmp(ri->slave_master_host, 1546 ri->master->promoted_slave->addr->ip) == 0 && 1547 ri->slave_master_port == ri->master->promoted_slave->addr->port) 1548 { 1549 ri->flags &= ~SRI_RECONF_SENT; 1550 ri->flags |= SRI_RECONF_INPROG; 1551 sentinelEvent(REDIS_NOTICE,"+slave-reconf-inprog",ri,"%@"); 1552 } 1553 1554 /* SRI_RECONF_INPROG -> SRI_RECONF_DONE */ 1555 if ((ri->flags & SRI_RECONF_INPROG) && 1556 ri->slave_master_link_status == SENTINEL_MASTER_LINK_STATUS_UP) 1557 { 1558 ri->flags &= ~SRI_RECONF_INPROG; 1559 ri->flags |= SRI_RECONF_DONE; 1560 sentinelEvent(REDIS_NOTICE,"+slave-reconf-done",ri,"%@"); 1561 /* If we are moving forward (a new slave is now configured) 1562 * we update the change_time as we are conceptually passing 1563 * to the next slave. */ 1564 ri->failover_state_change_time = mstime(); 1565 } 1566 } 1567 } 1568 1569 void sentinelInfoReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 1570 sentinelRedisInstance *ri = c->data; 1571 redisReply *r; 1572 1573 if (ri) ri->pending_commands--; 1574 if (!reply || !ri) return; 1575 r = reply; 1576 1577 if (r->type == REDIS_REPLY_STRING) { 1578 sentinelRefreshInstanceInfo(ri,r->str); 1579 } 1580 } 1581 1582 /* Just discard the reply. We use this when we are not monitoring the return 1583 * value of the command but its effects directly. */ 1584 void sentinelDiscardReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 1585 sentinelRedisInstance *ri = c->data; 1586 1587 if (ri) ri->pending_commands--; 1588 } 1589 1590 void sentinelPingReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 1591 sentinelRedisInstance *ri = c->data; 1592 redisReply *r; 1593 1594 if (ri) ri->pending_commands--; 1595 if (!reply || !ri) return; 1596 r = reply; 1597 1598 if (r->type == REDIS_REPLY_STATUS || 1599 r->type == REDIS_REPLY_ERROR) { 1600 /* Update the "instance available" field only if this is an 1601 * acceptable reply. */ 1602 if (strncmp(r->str,"PONG",4) == 0 || 1603 strncmp(r->str,"LOADING",7) == 0 || 1604 strncmp(r->str,"MASTERDOWN",10) == 0) 1605 { 1606 ri->last_avail_time = mstime(); 1607 } else { 1608 /* Send a SCRIPT KILL command if the instance appears to be 1609 * down because of a busy script. */ 1610 if (strncmp(r->str,"BUSY",4) == 0 && 1611 (ri->flags & SRI_S_DOWN) && 1612 !(ri->flags & SRI_SCRIPT_KILL_SENT)) 1613 { 1614 redisAsyncCommand(ri->cc, 1615 sentinelDiscardReplyCallback, NULL, "SCRIPT KILL"); 1616 ri->flags |= SRI_SCRIPT_KILL_SENT; 1617 } 1618 } 1619 } 1620 ri->last_pong_time = mstime(); 1621 } 1622 1623 /* This is called when we get the reply about the PUBLISH command we send 1624 * to the master to advertise this sentinel. */ 1625 void sentinelPublishReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 1626 sentinelRedisInstance *ri = c->data; 1627 redisReply *r; 1628 1629 if (ri) ri->pending_commands--; 1630 if (!reply || !ri) return; 1631 r = reply; 1632 1633 /* Only update pub_time if we actually published our message. Otherwise 1634 * we'll retry against in 100 milliseconds. */ 1635 if (r->type != REDIS_REPLY_ERROR) 1636 ri->last_pub_time = mstime(); 1637 } 1638 1639 /* This is our Pub/Sub callback for the Hello channel. It's useful in order 1640 * to discover other sentinels attached at the same master. */ 1641 void sentinelReceiveHelloMessages(redisAsyncContext *c, void *reply, void *privdata) { 1642 sentinelRedisInstance *ri = c->data; 1643 redisReply *r; 1644 1645 if (!reply || !ri) return; 1646 r = reply; 1647 1648 /* Update the last activity in the pubsub channel. Note that since we 1649 * receive our messages as well this timestamp can be used to detect 1650 * if the link is probably disconnected even if it seems otherwise. */ 1651 ri->pc_last_activity = mstime(); 1652 1653 /* Sanity check in the reply we expect, so that the code that follows 1654 * can avoid to check for details. */ 1655 if (r->type != REDIS_REPLY_ARRAY || 1656 r->elements != 3 || 1657 r->element[0]->type != REDIS_REPLY_STRING || 1658 r->element[1]->type != REDIS_REPLY_STRING || 1659 r->element[2]->type != REDIS_REPLY_STRING || 1660 strcmp(r->element[0]->str,"message") != 0) return; 1661 1662 /* We are not interested in meeting ourselves */ 1663 if (strstr(r->element[2]->str,server.runid) != NULL) return; 1664 1665 { 1666 int numtokens, port, removed, canfailover; 1667 char **token = sdssplitlen(r->element[2]->str, 1668 r->element[2]->len, 1669 ":",1,&numtokens); 1670 sentinelRedisInstance *sentinel; 1671 1672 if (numtokens == 4) { 1673 /* First, try to see if we already have this sentinel. */ 1674 port = atoi(token[1]); 1675 canfailover = atoi(token[3]); 1676 sentinel = getSentinelRedisInstanceByAddrAndRunID( 1677 ri->sentinels,token[0],port,token[2]); 1678 1679 if (!sentinel) { 1680 /* If not, remove all the sentinels that have the same runid 1681 * OR the same ip/port, because it's either a restart or a 1682 * network topology change. */ 1683 removed = removeMatchingSentinelsFromMaster(ri,token[0],port, 1684 token[2]); 1685 if (removed) { 1686 sentinelEvent(REDIS_NOTICE,"-dup-sentinel",ri, 1687 "%@ #duplicate of %s:%d or %s", 1688 token[0],port,token[2]); 1689 } 1690 1691 /* Add the new sentinel. */ 1692 sentinel = createSentinelRedisInstance(NULL,SRI_SENTINEL, 1693 token[0],port,ri->quorum,ri); 1694 if (sentinel) { 1695 sentinelEvent(REDIS_NOTICE,"+sentinel",sentinel,"%@"); 1696 /* The runid is NULL after a new instance creation and 1697 * for Sentinels we don't have a later chance to fill it, 1698 * so do it now. */ 1699 sentinel->runid = sdsnew(token[2]); 1700 } 1701 } 1702 1703 /* Update the state of the Sentinel. */ 1704 if (sentinel) { 1705 sentinel->last_hello_time = mstime(); 1706 if (canfailover) 1707 sentinel->flags |= SRI_CAN_FAILOVER; 1708 else 1709 sentinel->flags &= ~SRI_CAN_FAILOVER; 1710 } 1711 } 1712 sdsfreesplitres(token,numtokens); 1713 } 1714 } 1715 1716 void sentinelPingInstance(sentinelRedisInstance *ri) { 1717 mstime_t now = mstime(); 1718 mstime_t info_period; 1719 int retval; 1720 1721 /* Return ASAP if we have already a PING or INFO already pending, or 1722 * in the case the instance is not properly connected. */ 1723 if (ri->flags & SRI_DISCONNECTED) return; 1724 1725 /* For INFO, PING, PUBLISH that are not critical commands to send we 1726 * also have a limit of SENTINEL_MAX_PENDING_COMMANDS. We don't 1727 * want to use a lot of memory just because a link is not working 1728 * properly (note that anyway there is a redundant protection about this, 1729 * that is, the link will be disconnected and reconnected if a long 1730 * timeout condition is detected. */ 1731 if (ri->pending_commands >= SENTINEL_MAX_PENDING_COMMANDS) return; 1732 1733 /* If this is a slave of a master in O_DOWN condition we start sending 1734 * it INFO every second, instead of the usual SENTINEL_INFO_PERIOD 1735 * period. In this state we want to closely monitor slaves in case they 1736 * are turned into masters by another Sentinel, or by the sysadmin. */ 1737 if ((ri->flags & SRI_SLAVE) && 1738 (ri->master->flags & (SRI_O_DOWN|SRI_FAILOVER_IN_PROGRESS))) { 1739 info_period = 1000; 1740 } else { 1741 info_period = SENTINEL_INFO_PERIOD; 1742 } 1743 1744 if ((ri->flags & SRI_SENTINEL) == 0 && 1745 (ri->info_refresh == 0 || 1746 (now - ri->info_refresh) > info_period)) 1747 { 1748 /* Send INFO to masters and slaves, not sentinels. */ 1749 retval = redisAsyncCommand(ri->cc, 1750 sentinelInfoReplyCallback, NULL, "INFO"); 1751 if (retval != REDIS_OK) return; 1752 ri->pending_commands++; 1753 } else if ((now - ri->last_pong_time) > SENTINEL_PING_PERIOD) { 1754 /* Send PING to all the three kinds of instances. */ 1755 retval = redisAsyncCommand(ri->cc, 1756 sentinelPingReplyCallback, NULL, "PING"); 1757 if (retval != REDIS_OK) return; 1758 ri->pending_commands++; 1759 } else if ((ri->flags & SRI_MASTER) && 1760 (now - ri->last_pub_time) > SENTINEL_PUBLISH_PERIOD) 1761 { 1762 /* PUBLISH hello messages only to masters. */ 1763 struct sockaddr_in sa; 1764 socklen_t salen = sizeof(sa); 1765 1766 if (getsockname(ri->cc->c.fd,(struct sockaddr*)&sa,&salen) != -1) { 1767 char myaddr[128]; 1768 1769 snprintf(myaddr,sizeof(myaddr),"%s:%d:%s:%d", 1770 inet_ntoa(sa.sin_addr), server.port, server.runid, 1771 (ri->flags & SRI_CAN_FAILOVER) != 0); 1772 retval = redisAsyncCommand(ri->cc, 1773 sentinelPublishReplyCallback, NULL, "PUBLISH %s %s", 1774 SENTINEL_HELLO_CHANNEL,myaddr); 1775 if (retval != REDIS_OK) return; 1776 ri->pending_commands++; 1777 } 1778 } 1779 } 1780 1781 /* =========================== SENTINEL command ============================= */ 1782 1783 const char *sentinelFailoverStateStr(int state) { 1784 switch(state) { 1785 case SENTINEL_FAILOVER_STATE_NONE: return "none"; 1786 case SENTINEL_FAILOVER_STATE_WAIT_START: return "wait_start"; 1787 case SENTINEL_FAILOVER_STATE_SELECT_SLAVE: return "select_slave"; 1788 case SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE: return "send_slaveof_noone"; 1789 case SENTINEL_FAILOVER_STATE_WAIT_PROMOTION: return "wait_promotion"; 1790 case SENTINEL_FAILOVER_STATE_RECONF_SLAVES: return "reconf_slaves"; 1791 case SENTINEL_FAILOVER_STATE_ALERT_CLIENTS: return "alert_clients"; 1792 case SENTINEL_FAILOVER_STATE_DETECT_END: return "detect_end"; 1793 case SENTINEL_FAILOVER_STATE_UPDATE_CONFIG: return "update_config"; 1794 default: return "unknown"; 1795 } 1796 } 1797 1798 /* Redis instance to Redis protocol representation. */ 1799 void addReplySentinelRedisInstance(redisClient *c, sentinelRedisInstance *ri) { 1800 char *flags = sdsempty(); 1801 void *mbl; 1802 int fields = 0; 1803 1804 mbl = addDeferredMultiBulkLength(c); 1805 1806 addReplyBulkCString(c,"name"); 1807 addReplyBulkCString(c,ri->name); 1808 fields++; 1809 1810 addReplyBulkCString(c,"ip"); 1811 addReplyBulkCString(c,ri->addr->ip); 1812 fields++; 1813 1814 addReplyBulkCString(c,"port"); 1815 addReplyBulkLongLong(c,ri->addr->port); 1816 fields++; 1817 1818 addReplyBulkCString(c,"runid"); 1819 addReplyBulkCString(c,ri->runid ? ri->runid : ""); 1820 fields++; 1821 1822 addReplyBulkCString(c,"flags"); 1823 if (ri->flags & SRI_S_DOWN) flags = sdscat(flags,"s_down,"); 1824 if (ri->flags & SRI_O_DOWN) flags = sdscat(flags,"o_down,"); 1825 if (ri->flags & SRI_MASTER) flags = sdscat(flags,"master,"); 1826 if (ri->flags & SRI_SLAVE) flags = sdscat(flags,"slave,"); 1827 if (ri->flags & SRI_SENTINEL) flags = sdscat(flags,"sentinel,"); 1828 if (ri->flags & SRI_DISCONNECTED) flags = sdscat(flags,"disconnected,"); 1829 if (ri->flags & SRI_MASTER_DOWN) flags = sdscat(flags,"master_down,"); 1830 if (ri->flags & SRI_FAILOVER_IN_PROGRESS) 1831 flags = sdscat(flags,"failover_in_progress,"); 1832 if (ri->flags & SRI_I_AM_THE_LEADER) 1833 flags = sdscat(flags,"i_am_the_leader,"); 1834 if (ri->flags & SRI_PROMOTED) flags = sdscat(flags,"promoted,"); 1835 if (ri->flags & SRI_RECONF_SENT) flags = sdscat(flags,"reconf_sent,"); 1836 if (ri->flags & SRI_RECONF_INPROG) flags = sdscat(flags,"reconf_inprog,"); 1837 if (ri->flags & SRI_RECONF_DONE) flags = sdscat(flags,"reconf_done,"); 1838 1839 if (sdslen(flags) != 0) flags = sdsrange(flags,0,-2); /* remove last "," */ 1840 addReplyBulkCString(c,flags); 1841 sdsfree(flags); 1842 fields++; 1843 1844 addReplyBulkCString(c,"pending-commands"); 1845 addReplyBulkLongLong(c,ri->pending_commands); 1846 fields++; 1847 1848 if (ri->flags & SRI_FAILOVER_IN_PROGRESS) { 1849 addReplyBulkCString(c,"failover-state"); 1850 addReplyBulkCString(c,(char*)sentinelFailoverStateStr(ri->failover_state)); 1851 fields++; 1852 } 1853 1854 addReplyBulkCString(c,"last-ok-ping-reply"); 1855 addReplyBulkLongLong(c,mstime() - ri->last_avail_time); 1856 fields++; 1857 1858 addReplyBulkCString(c,"last-ping-reply"); 1859 addReplyBulkLongLong(c,mstime() - ri->last_pong_time); 1860 fields++; 1861 1862 if (ri->flags & SRI_S_DOWN) { 1863 addReplyBulkCString(c,"s-down-time"); 1864 addReplyBulkLongLong(c,mstime()-ri->s_down_since_time); 1865 fields++; 1866 } 1867 1868 if (ri->flags & SRI_O_DOWN) { 1869 addReplyBulkCString(c,"o-down-time"); 1870 addReplyBulkLongLong(c,mstime()-ri->o_down_since_time); 1871 fields++; 1872 } 1873 1874 /* Masters and Slaves */ 1875 if (ri->flags & (SRI_MASTER|SRI_SLAVE)) { 1876 addReplyBulkCString(c,"info-refresh"); 1877 addReplyBulkLongLong(c,mstime() - ri->info_refresh); 1878 fields++; 1879 } 1880 1881 /* Only masters */ 1882 if (ri->flags & SRI_MASTER) { 1883 addReplyBulkCString(c,"num-slaves"); 1884 addReplyBulkLongLong(c,dictSize(ri->slaves)); 1885 fields++; 1886 1887 addReplyBulkCString(c,"num-other-sentinels"); 1888 addReplyBulkLongLong(c,dictSize(ri->sentinels)); 1889 fields++; 1890 1891 addReplyBulkCString(c,"quorum"); 1892 addReplyBulkLongLong(c,ri->quorum); 1893 fields++; 1894 } 1895 1896 /* Only slaves */ 1897 if (ri->flags & SRI_SLAVE) { 1898 addReplyBulkCString(c,"master-link-down-time"); 1899 addReplyBulkLongLong(c,ri->master_link_down_time); 1900 fields++; 1901 1902 addReplyBulkCString(c,"master-link-status"); 1903 addReplyBulkCString(c, 1904 (ri->slave_master_link_status == SENTINEL_MASTER_LINK_STATUS_UP) ? 1905 "ok" : "err"); 1906 fields++; 1907 1908 addReplyBulkCString(c,"master-host"); 1909 addReplyBulkCString(c, 1910 ri->slave_master_host ? ri->slave_master_host : "?"); 1911 fields++; 1912 1913 addReplyBulkCString(c,"master-port"); 1914 addReplyBulkLongLong(c,ri->slave_master_port); 1915 fields++; 1916 1917 addReplyBulkCString(c,"slave-priority"); 1918 addReplyBulkLongLong(c,ri->slave_priority); 1919 fields++; 1920 } 1921 1922 /* Only sentinels */ 1923 if (ri->flags & SRI_SENTINEL) { 1924 addReplyBulkCString(c,"last-hello-message"); 1925 addReplyBulkLongLong(c,mstime() - ri->last_hello_time); 1926 fields++; 1927 1928 addReplyBulkCString(c,"can-failover-its-master"); 1929 addReplyBulkLongLong(c,(ri->flags & SRI_CAN_FAILOVER) != 0); 1930 fields++; 1931 1932 if (ri->flags & SRI_MASTER_DOWN) { 1933 addReplyBulkCString(c,"subjective-leader"); 1934 addReplyBulkCString(c,ri->leader ? ri->leader : "?"); 1935 fields++; 1936 } 1937 } 1938 1939 setDeferredMultiBulkLength(c,mbl,fields*2); 1940 } 1941 1942 /* Output a number of instances contained inside a dictionary as 1943 * Redis protocol. */ 1944 void addReplyDictOfRedisInstances(redisClient *c, dict *instances) { 1945 dictIterator *di; 1946 dictEntry *de; 1947 1948 di = dictGetIterator(instances); 1949 addReplyMultiBulkLen(c,dictSize(instances)); 1950 while((de = dictNext(di)) != NULL) { 1951 sentinelRedisInstance *ri = dictGetVal(de); 1952 1953 addReplySentinelRedisInstance(c,ri); 1954 } 1955 dictReleaseIterator(di); 1956 } 1957 1958 /* Lookup the named master into sentinel.masters. 1959 * If the master is not found reply to the client with an error and returns 1960 * NULL. */ 1961 sentinelRedisInstance *sentinelGetMasterByNameOrReplyError(redisClient *c, 1962 robj *name) 1963 { 1964 sentinelRedisInstance *ri; 1965 1966 ri = dictFetchValue(sentinel.masters,c->argv[2]->ptr); 1967 if (!ri) { 1968 addReplyError(c,"No such master with that name"); 1969 return NULL; 1970 } 1971 return ri; 1972 } 1973 1974 void sentinelCommand(redisClient *c) { 1975 if (!strcasecmp(c->argv[1]->ptr,"masters")) { 1976 /* SENTINEL MASTERS */ 1977 if (c->argc != 2) goto numargserr; 1978 1979 addReplyDictOfRedisInstances(c,sentinel.masters); 1980 } else if (!strcasecmp(c->argv[1]->ptr,"slaves")) { 1981 /* SENTINEL SLAVES <master-name> */ 1982 sentinelRedisInstance *ri; 1983 1984 if (c->argc != 3) goto numargserr; 1985 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) == NULL) 1986 return; 1987 addReplyDictOfRedisInstances(c,ri->slaves); 1988 } else if (!strcasecmp(c->argv[1]->ptr,"sentinels")) { 1989 /* SENTINEL SENTINELS <master-name> */ 1990 sentinelRedisInstance *ri; 1991 1992 if (c->argc != 3) goto numargserr; 1993 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) == NULL) 1994 return; 1995 addReplyDictOfRedisInstances(c,ri->sentinels); 1996 } else if (!strcasecmp(c->argv[1]->ptr,"is-master-down-by-addr")) { 1997 /* SENTINEL IS-MASTER-DOWN-BY-ADDR <ip> <port> */ 1998 sentinelRedisInstance *ri; 1999 char *leader = NULL; 2000 long port; 2001 int isdown = 0; 2002 2003 if (c->argc != 4) goto numargserr; 2004 if (getLongFromObjectOrReply(c,c->argv[3],&port,NULL) != REDIS_OK) 2005 return; 2006 ri = getSentinelRedisInstanceByAddrAndRunID(sentinel.masters, 2007 c->argv[2]->ptr,port,NULL); 2008 2009 /* It exists? Is actually a master? Is subjectively down? It's down. 2010 * Note: if we are in tilt mode we always reply with "0". */ 2011 if (!sentinel.tilt && ri && (ri->flags & SRI_S_DOWN) && 2012 (ri->flags & SRI_MASTER)) 2013 isdown = 1; 2014 if (ri) leader = sentinelGetSubjectiveLeader(ri); 2015 2016 /* Reply with a two-elements multi-bulk reply: down state, leader. */ 2017 addReplyMultiBulkLen(c,2); 2018 addReply(c, isdown ? shared.cone : shared.czero); 2019 addReplyBulkCString(c, leader ? leader : "?"); 2020 if (leader) sdsfree(leader); 2021 } else if (!strcasecmp(c->argv[1]->ptr,"reset")) { 2022 /* SENTINEL RESET <pattern> */ 2023 if (c->argc != 3) goto numargserr; 2024 addReplyLongLong(c,sentinelResetMastersByPattern(c->argv[2]->ptr,SENTINEL_GENERATE_EVENT)); 2025 } else if (!strcasecmp(c->argv[1]->ptr,"get-master-addr-by-name")) { 2026 /* SENTINEL GET-MASTER-ADDR-BY-NAME <master-name> */ 2027 sentinelRedisInstance *ri; 2028 2029 if (c->argc != 3) goto numargserr; 2030 ri = sentinelGetMasterByName(c->argv[2]->ptr); 2031 if (ri == NULL) { 2032 addReply(c,shared.nullmultibulk); 2033 } else if (ri->info_refresh == 0) { 2034 addReplySds(c,sdsnew("-IDONTKNOW I have not enough information to reply. Please ask another Sentinel.\r\n")); 2035 } else { 2036 sentinelAddr *addr = ri->addr; 2037 2038 /* If we are in the middle of a failover, and the slave was 2039 * already successfully switched to master role, we can advertise 2040 * the new address as slave in order to allow clients to talk 2041 * with the new master ASAP. */ 2042 if ((ri->flags & SRI_FAILOVER_IN_PROGRESS) && 2043 ri->promoted_slave && 2044 ri->failover_state >= SENTINEL_FAILOVER_STATE_RECONF_SLAVES) 2045 { 2046 addr = ri->promoted_slave->addr; 2047 } 2048 addReplyMultiBulkLen(c,2); 2049 addReplyBulkCString(c,addr->ip); 2050 addReplyBulkLongLong(c,addr->port); 2051 } 2052 } else if (!strcasecmp(c->argv[1]->ptr,"failover")) { 2053 /* SENTINEL FAILOVER <master-name> */ 2054 sentinelRedisInstance *ri; 2055 2056 if (c->argc != 3) goto numargserr; 2057 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) == NULL) 2058 return; 2059 if (ri->flags & SRI_FAILOVER_IN_PROGRESS) { 2060 addReplySds(c,sdsnew("-INPROG Failover already in progress\r\n")); 2061 return; 2062 } 2063 if (sentinelSelectSlave(ri) == NULL) { 2064 addReplySds(c,sdsnew("-NOGOODSLAVE No suitable slave to promote\r\n")); 2065 return; 2066 } 2067 sentinelStartFailover(ri,SENTINEL_FAILOVER_STATE_WAIT_START); 2068 ri->flags |= SRI_FORCE_FAILOVER; 2069 addReply(c,shared.ok); 2070 } else if (!strcasecmp(c->argv[1]->ptr,"pending-scripts")) { 2071 /* SENTINEL PENDING-SCRIPTS */ 2072 2073 if (c->argc != 2) goto numargserr; 2074 sentinelPendingScriptsCommand(c); 2075 } else { 2076 addReplyErrorFormat(c,"Unknown sentinel subcommand '%s'", 2077 (char*)c->argv[1]->ptr); 2078 } 2079 return; 2080 2081 numargserr: 2082 addReplyErrorFormat(c,"Wrong number of commands for 'sentinel %s'", 2083 (char*)c->argv[1]->ptr); 2084 } 2085 2086 void sentinelInfoCommand(redisClient *c) { 2087 char *section = c->argc == 2 ? c->argv[1]->ptr : "default"; 2088 sds info = sdsempty(); 2089 int defsections = !strcasecmp(section,"default"); 2090 int sections = 0; 2091 2092 if (c->argc > 2) { 2093 addReply(c,shared.syntaxerr); 2094 return; 2095 } 2096 2097 if (!strcasecmp(section,"server") || defsections) { 2098 if (sections++) info = sdscat(info,"\r\n"); 2099 sds serversection = genRedisInfoString("server"); 2100 info = sdscatlen(info,serversection,sdslen(serversection)); 2101 sdsfree(serversection); 2102 } 2103 2104 if (!strcasecmp(section,"sentinel") || defsections) { 2105 dictIterator *di; 2106 dictEntry *de; 2107 int master_id = 0; 2108 2109 if (sections++) info = sdscat(info,"\r\n"); 2110 info = sdscatprintf(info, 2111 "# Sentinel\r\n" 2112 "sentinel_masters:%lu\r\n" 2113 "sentinel_tilt:%d\r\n" 2114 "sentinel_running_scripts:%d\r\n" 2115 "sentinel_scripts_queue_length:%ld\r\n", 2116 dictSize(sentinel.masters), 2117 sentinel.tilt, 2118 sentinel.running_scripts, 2119 listLength(sentinel.scripts_queue)); 2120 2121 di = dictGetIterator(sentinel.masters); 2122 while((de = dictNext(di)) != NULL) { 2123 sentinelRedisInstance *ri = dictGetVal(de); 2124 char *status = "ok"; 2125 2126 if (ri->flags & SRI_O_DOWN) status = "odown"; 2127 else if (ri->flags & SRI_S_DOWN) status = "sdown"; 2128 info = sdscatprintf(info, 2129 "master%d:name=%s,status=%s,address=%s:%d," 2130 "slaves=%lu,sentinels=%lu\r\n", 2131 master_id++, ri->name, status, 2132 ri->addr->ip, ri->addr->port, 2133 dictSize(ri->slaves), 2134 dictSize(ri->sentinels)+1); 2135 } 2136 dictReleaseIterator(di); 2137 } 2138 2139 addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n", 2140 (unsigned long)sdslen(info))); 2141 addReplySds(c,info); 2142 addReply(c,shared.crlf); 2143 } 2144 2145 /* ===================== SENTINEL availability checks ======================= */ 2146 2147 /* Is this instance down from our point of view? */ 2148 void sentinelCheckSubjectivelyDown(sentinelRedisInstance *ri) { 2149 mstime_t elapsed = mstime() - ri->last_avail_time; 2150 2151 /* Check if we are in need for a reconnection of one of the 2152 * links, because we are detecting low activity. 2153 * 2154 * 1) Check if the command link seems connected, was connected not less 2155 * than SENTINEL_MIN_LINK_RECONNECT_PERIOD, but still we have an 2156 * idle time that is greater than down_after_period / 2 seconds. */ 2157 if (ri->cc && 2158 (mstime() - ri->cc_conn_time) > SENTINEL_MIN_LINK_RECONNECT_PERIOD && 2159 (mstime() - ri->last_pong_time) > (ri->down_after_period/2)) 2160 { 2161 sentinelKillLink(ri,ri->cc); 2162 } 2163 2164 /* 2) Check if the pubsub link seems connected, was connected not less 2165 * than SENTINEL_MIN_LINK_RECONNECT_PERIOD, but still we have no 2166 * activity in the Pub/Sub channel for more than 2167 * SENTINEL_PUBLISH_PERIOD * 3. 2168 */ 2169 if (ri->pc && 2170 (mstime() - ri->pc_conn_time) > SENTINEL_MIN_LINK_RECONNECT_PERIOD && 2171 (mstime() - ri->pc_last_activity) > (SENTINEL_PUBLISH_PERIOD*3)) 2172 { 2173 sentinelKillLink(ri,ri->pc); 2174 } 2175 2176 /* Update the subjectively down flag. */ 2177 if (elapsed > ri->down_after_period) { 2178 /* Is subjectively down */ 2179 if ((ri->flags & SRI_S_DOWN) == 0) { 2180 sentinelEvent(REDIS_WARNING,"+sdown",ri,"%@"); 2181 ri->s_down_since_time = mstime(); 2182 ri->flags |= SRI_S_DOWN; 2183 } 2184 } else { 2185 /* Is subjectively up */ 2186 if (ri->flags & SRI_S_DOWN) { 2187 sentinelEvent(REDIS_WARNING,"-sdown",ri,"%@"); 2188 ri->flags &= ~(SRI_S_DOWN|SRI_SCRIPT_KILL_SENT); 2189 } 2190 } 2191 } 2192 2193 /* Is this instance down accordingly to the configured quorum? */ 2194 void sentinelCheckObjectivelyDown(sentinelRedisInstance *master) { 2195 dictIterator *di; 2196 dictEntry *de; 2197 int quorum = 0, odown = 0; 2198 2199 if (master->flags & SRI_S_DOWN) { 2200 /* Is down for enough sentinels? */ 2201 quorum = 1; /* the current sentinel. */ 2202 /* Count all the other sentinels. */ 2203 di = dictGetIterator(master->sentinels); 2204 while((de = dictNext(di)) != NULL) { 2205 sentinelRedisInstance *ri = dictGetVal(de); 2206 2207 if (ri->flags & SRI_MASTER_DOWN) quorum++; 2208 } 2209 dictReleaseIterator(di); 2210 if (quorum >= master->quorum) odown = 1; 2211 } 2212 2213 /* Set the flag accordingly to the outcome. */ 2214 if (odown) { 2215 if ((master->flags & SRI_O_DOWN) == 0) { 2216 sentinelEvent(REDIS_WARNING,"+odown",master,"%@ #quorum %d/%d", 2217 quorum, master->quorum); 2218 master->flags |= SRI_O_DOWN; 2219 master->o_down_since_time = mstime(); 2220 } 2221 } else { 2222 if (master->flags & SRI_O_DOWN) { 2223 sentinelEvent(REDIS_WARNING,"-odown",master,"%@"); 2224 master->flags &= ~SRI_O_DOWN; 2225 } 2226 } 2227 } 2228 2229 /* Receive the SENTINEL is-master-down-by-addr reply, see the 2230 * sentinelAskMasterStateToOtherSentinels() function for more information. */ 2231 void sentinelReceiveIsMasterDownReply(redisAsyncContext *c, void *reply, void *privdata) { 2232 sentinelRedisInstance *ri = c->data; 2233 redisReply *r; 2234 2235 if (ri) ri->pending_commands--; 2236 if (!reply || !ri) return; 2237 r = reply; 2238 2239 /* Ignore every error or unexpected reply. 2240 * Note that if the command returns an error for any reason we'll 2241 * end clearing the SRI_MASTER_DOWN flag for timeout anyway. */ 2242 if (r->type == REDIS_REPLY_ARRAY && r->elements == 2 && 2243 r->element[0]->type == REDIS_REPLY_INTEGER && 2244 r->element[1]->type == REDIS_REPLY_STRING) 2245 { 2246 ri->last_master_down_reply_time = mstime(); 2247 if (r->element[0]->integer == 1) { 2248 ri->flags |= SRI_MASTER_DOWN; 2249 } else { 2250 ri->flags &= ~SRI_MASTER_DOWN; 2251 } 2252 sdsfree(ri->leader); 2253 ri->leader = sdsnew(r->element[1]->str); 2254 } 2255 } 2256 2257 /* If we think (subjectively) the master is down, we start sending 2258 * SENTINEL IS-MASTER-DOWN-BY-ADDR requests to other sentinels 2259 * in order to get the replies that allow to reach the quorum and 2260 * possibly also mark the master as objectively down. */ 2261 void sentinelAskMasterStateToOtherSentinels(sentinelRedisInstance *master) { 2262 dictIterator *di; 2263 dictEntry *de; 2264 2265 di = dictGetIterator(master->sentinels); 2266 while((de = dictNext(di)) != NULL) { 2267 sentinelRedisInstance *ri = dictGetVal(de); 2268 mstime_t elapsed = mstime() - ri->last_master_down_reply_time; 2269 char port[32]; 2270 int retval; 2271 2272 /* If the master state from other sentinel is too old, we clear it. */ 2273 if (elapsed > SENTINEL_INFO_VALIDITY_TIME) { 2274 ri->flags &= ~SRI_MASTER_DOWN; 2275 sdsfree(ri->leader); 2276 ri->leader = NULL; 2277 } 2278 2279 /* Only ask if master is down to other sentinels if: 2280 * 2281 * 1) We believe it is down, or there is a failover in progress. 2282 * 2) Sentinel is connected. 2283 * 3) We did not received the info within SENTINEL_ASK_PERIOD ms. */ 2284 if ((master->flags & (SRI_S_DOWN|SRI_FAILOVER_IN_PROGRESS)) == 0) 2285 continue; 2286 if (ri->flags & SRI_DISCONNECTED) continue; 2287 if (mstime() - ri->last_master_down_reply_time < SENTINEL_ASK_PERIOD) 2288 continue; 2289 2290 /* Ask */ 2291 ll2string(port,sizeof(port),master->addr->port); 2292 retval = redisAsyncCommand(ri->cc, 2293 sentinelReceiveIsMasterDownReply, NULL, 2294 "SENTINEL is-master-down-by-addr %s %s", 2295 master->addr->ip, port); 2296 if (retval == REDIS_OK) ri->pending_commands++; 2297 } 2298 dictReleaseIterator(di); 2299 } 2300 2301 /* =============================== FAILOVER ================================= */ 2302 2303 /* Given a master get the "subjective leader", that is, among all the sentinels 2304 * with given characteristics, the one with the lexicographically smaller 2305 * runid. The characteristics required are: 2306 * 2307 * 1) Has SRI_CAN_FAILOVER flag. 2308 * 2) Is not disconnected. 2309 * 3) Recently answered to our ping (no longer than 2310 * SENTINEL_INFO_VALIDITY_TIME milliseconds ago). 2311 * 2312 * The function returns a pointer to an sds string representing the runid of the 2313 * leader sentinel instance (from our point of view). Otherwise NULL is 2314 * returned if there are no suitable sentinels. 2315 */ 2316 2317 int compareRunID(const void *a, const void *b) { 2318 char **aptrptr = (char**)a, **bptrptr = (char**)b; 2319 return strcasecmp(*aptrptr, *bptrptr); 2320 } 2321 2322 char *sentinelGetSubjectiveLeader(sentinelRedisInstance *master) { 2323 dictIterator *di; 2324 dictEntry *de; 2325 char **instance = 2326 zmalloc(sizeof(char*)*(dictSize(master->sentinels)+1)); 2327 int instances = 0; 2328 char *leader = NULL; 2329 2330 if (master->flags & SRI_CAN_FAILOVER) { 2331 /* Add myself if I'm a Sentinel that can failover this master. */ 2332 instance[instances++] = server.runid; 2333 } 2334 2335 di = dictGetIterator(master->sentinels); 2336 while((de = dictNext(di)) != NULL) { 2337 sentinelRedisInstance *ri = dictGetVal(de); 2338 mstime_t lag = mstime() - ri->last_avail_time; 2339 2340 if (lag > SENTINEL_INFO_VALIDITY_TIME || 2341 !(ri->flags & SRI_CAN_FAILOVER) || 2342 (ri->flags & SRI_DISCONNECTED) || 2343 ri->runid == NULL) 2344 continue; 2345 instance[instances++] = ri->runid; 2346 } 2347 dictReleaseIterator(di); 2348 2349 /* If we have at least one instance passing our checks, order the array 2350 * by runid. */ 2351 if (instances) { 2352 qsort(instance,instances,sizeof(char*),compareRunID); 2353 leader = sdsnew(instance[0]); 2354 } 2355 zfree(instance); 2356 return leader; 2357 } 2358 2359 struct sentinelLeader { 2360 char *runid; 2361 unsigned long votes; 2362 }; 2363 2364 /* Helper function for sentinelGetObjectiveLeader, increment the counter 2365 * relative to the specified runid. */ 2366 void sentinelObjectiveLeaderIncr(dict *counters, char *runid) { 2367 dictEntry *de = dictFind(counters,runid); 2368 uint64_t oldval; 2369 2370 if (de) { 2371 oldval = dictGetUnsignedIntegerVal(de); 2372 dictSetUnsignedIntegerVal(de,oldval+1); 2373 } else { 2374 de = dictAddRaw(counters,runid); 2375 redisAssert(de != NULL); 2376 dictSetUnsignedIntegerVal(de,1); 2377 } 2378 } 2379 2380 /* Scan all the Sentinels attached to this master to check what is the 2381 * most voted leader among Sentinels. */ 2382 char *sentinelGetObjectiveLeader(sentinelRedisInstance *master) { 2383 dict *counters; 2384 dictIterator *di; 2385 dictEntry *de; 2386 unsigned int voters = 0, voters_quorum; 2387 char *myvote; 2388 char *winner = NULL; 2389 2390 redisAssert(master->flags & (SRI_O_DOWN|SRI_FAILOVER_IN_PROGRESS)); 2391 counters = dictCreate(&leaderVotesDictType,NULL); 2392 2393 /* Count my vote. */ 2394 myvote = sentinelGetSubjectiveLeader(master); 2395 if (myvote) { 2396 sentinelObjectiveLeaderIncr(counters,myvote); 2397 voters++; 2398 } 2399 2400 /* Count other sentinels votes */ 2401 di = dictGetIterator(master->sentinels); 2402 while((de = dictNext(di)) != NULL) { 2403 sentinelRedisInstance *ri = dictGetVal(de); 2404 if (ri->leader == NULL) continue; 2405 /* If the failover is not already in progress we are only interested 2406 * in Sentinels that believe the master is down. Otherwise the leader 2407 * selection is useful for the "failover-takedown" when the original 2408 * leader fails. In that case we consider all the voters. */ 2409 if (!(master->flags & SRI_FAILOVER_IN_PROGRESS) && 2410 !(ri->flags & SRI_MASTER_DOWN)) continue; 2411 sentinelObjectiveLeaderIncr(counters,ri->leader); 2412 voters++; 2413 } 2414 dictReleaseIterator(di); 2415 voters_quorum = voters/2+1; 2416 2417 /* Check what's the winner. For the winner to win, it needs two conditions: 2418 * 1) Absolute majority between voters (50% + 1). 2419 * 2) And anyway at least master->quorum votes. */ 2420 { 2421 uint64_t max_votes = 0; /* Max votes so far. */ 2422 2423 di = dictGetIterator(counters); 2424 while((de = dictNext(di)) != NULL) { 2425 uint64_t votes = dictGetUnsignedIntegerVal(de); 2426 2427 if (max_votes < votes) { 2428 max_votes = votes; 2429 winner = dictGetKey(de); 2430 } 2431 } 2432 dictReleaseIterator(di); 2433 if (winner && (max_votes < voters_quorum || max_votes < master->quorum)) 2434 winner = NULL; 2435 } 2436 winner = winner ? sdsnew(winner) : NULL; 2437 sdsfree(myvote); 2438 dictRelease(counters); 2439 return winner; 2440 } 2441 2442 /* Setup the master state to start a failover as a leader. 2443 * 2444 * State can be either: 2445 * 2446 * SENTINEL_FAILOVER_STATE_WAIT_START: starts a failover from scratch. 2447 * SENTINEL_FAILOVER_STATE_RECONF_SLAVES: takedown a failed failover. 2448 */ 2449 void sentinelStartFailover(sentinelRedisInstance *master, int state) { 2450 redisAssert(master->flags & SRI_MASTER); 2451 redisAssert(state == SENTINEL_FAILOVER_STATE_WAIT_START || 2452 state == SENTINEL_FAILOVER_STATE_RECONF_SLAVES); 2453 2454 master->failover_state = state; 2455 master->flags |= SRI_FAILOVER_IN_PROGRESS|SRI_I_AM_THE_LEADER; 2456 sentinelEvent(REDIS_WARNING,"+failover-triggered",master,"%@"); 2457 2458 /* Pick a random delay if it's a fresh failover (WAIT_START), and not 2459 * a recovery of a failover started by another sentinel. */ 2460 if (master->failover_state == SENTINEL_FAILOVER_STATE_WAIT_START) { 2461 master->failover_start_time = mstime() + 2462 SENTINEL_FAILOVER_FIXED_DELAY + 2463 (rand() % SENTINEL_FAILOVER_MAX_RANDOM_DELAY); 2464 sentinelEvent(REDIS_WARNING,"+failover-state-wait-start",master, 2465 "%@ #starting in %lld milliseconds", 2466 master->failover_start_time-mstime()); 2467 } 2468 master->failover_state_change_time = mstime(); 2469 } 2470 2471 /* This function checks if there are the conditions to start the failover, 2472 * that is: 2473 * 2474 * 1) Enough time has passed since O_DOWN. 2475 * 2) The master is marked as SRI_CAN_FAILOVER, so we can failover it. 2476 * 3) We are the objectively leader for this master. 2477 * 2478 * If the conditions are met we flag the master as SRI_FAILOVER_IN_PROGRESS 2479 * and SRI_I_AM_THE_LEADER. 2480 */ 2481 void sentinelStartFailoverIfNeeded(sentinelRedisInstance *master) { 2482 char *leader; 2483 int isleader; 2484 2485 /* We can't failover if the master is not in O_DOWN state or if 2486 * there is not already a failover in progress (to perform the 2487 * takedown if the leader died) or if this Sentinel is not allowed 2488 * to start a failover. */ 2489 if (!(master->flags & SRI_CAN_FAILOVER) || 2490 !(master->flags & (SRI_O_DOWN|SRI_FAILOVER_IN_PROGRESS))) return; 2491 2492 leader = sentinelGetObjectiveLeader(master); 2493 isleader = leader && strcasecmp(leader,server.runid) == 0; 2494 sdsfree(leader); 2495 2496 /* If I'm not the leader, I can't failover for sure. */ 2497 if (!isleader) return; 2498 2499 /* If the failover is already in progress there are two options... */ 2500 if (master->flags & SRI_FAILOVER_IN_PROGRESS) { 2501 if (master->flags & SRI_I_AM_THE_LEADER) { 2502 /* 1) I'm flagged as leader so I already started the failover. 2503 * Just return. */ 2504 return; 2505 } else { 2506 mstime_t elapsed = mstime() - master->failover_state_change_time; 2507 2508 /* 2) I'm the new leader, but I'm not flagged as leader in the 2509 * master: I did not started the failover, but the original 2510 * leader has no longer the leadership. 2511 * 2512 * In this case if the failover appears to be lagging 2513 * for at least 25% of the configured failover timeout, 2514 * I can assume I can take control. Otherwise 2515 * it's better to return and wait more. */ 2516 if (elapsed < (master->failover_timeout/4)) return; 2517 sentinelEvent(REDIS_WARNING,"+failover-takedown",master,"%@"); 2518 /* We have already an elected slave if we are in 2519 * FAILOVER_IN_PROGRESS state, that is, the slave that we 2520 * observed turning into a master. */ 2521 sentinelStartFailover(master,SENTINEL_FAILOVER_STATE_RECONF_SLAVES); 2522 /* As an observer we flagged all the slaves as RECONF_SENT but 2523 * now we are in charge of actually sending the reconfiguration 2524 * command so let's clear this flag for all the instances. */ 2525 sentinelDelFlagsToDictOfRedisInstances(master->slaves, 2526 SRI_RECONF_SENT); 2527 } 2528 } else { 2529 /* Brand new failover as SRI_FAILOVER_IN_PROGRESS was not set. 2530 * 2531 * Do we have a slave to promote? Otherwise don't start a failover 2532 * at all. */ 2533 if (sentinelSelectSlave(master) == NULL) return; 2534 sentinelStartFailover(master,SENTINEL_FAILOVER_STATE_WAIT_START); 2535 } 2536 } 2537 2538 /* Select a suitable slave to promote. The current algorithm only uses 2539 * the following parameters: 2540 * 2541 * 1) None of the following conditions: S_DOWN, O_DOWN, DISCONNECTED. 2542 * 2) last_avail_time more recent than SENTINEL_INFO_VALIDITY_TIME. 2543 * 3) info_refresh more recent than SENTINEL_INFO_VALIDITY_TIME. 2544 * 4) master_link_down_time no more than: 2545 * (now - master->s_down_since_time) + (master->down_after_period * 10). 2546 * 5) Slave priority can't be zero, otherwise the slave is discarded. 2547 * 2548 * Among all the slaves matching the above conditions we select the slave 2549 * with lower slave_priority. If priority is the same we select the slave 2550 * with lexicographically smaller runid. 2551 * 2552 * The function returns the pointer to the selected slave, otherwise 2553 * NULL if no suitable slave was found. 2554 */ 2555 2556 int compareSlavesForPromotion(const void *a, const void *b) { 2557 sentinelRedisInstance **sa = (sentinelRedisInstance **)a, 2558 **sb = (sentinelRedisInstance **)b; 2559 char *sa_runid, *sb_runid; 2560 2561 if ((*sa)->slave_priority != (*sb)->slave_priority) 2562 return (*sa)->slave_priority - (*sb)->slave_priority; 2563 2564 /* If priority is the same, select the slave with that has the 2565 * lexicographically smaller runid. Note that we try to handle runid 2566 * == NULL as there are old Redis versions that don't publish runid in 2567 * INFO. A NULL runid is considered bigger than any other runid. */ 2568 sa_runid = (*sa)->runid; 2569 sb_runid = (*sb)->runid; 2570 if (sa_runid == NULL && sb_runid == NULL) return 0; 2571 else if (sa_runid == NULL) return 1; /* a > b */ 2572 else if (sb_runid == NULL) return -1; /* a < b */ 2573 return strcasecmp(sa_runid, sb_runid); 2574 } 2575 2576 sentinelRedisInstance *sentinelSelectSlave(sentinelRedisInstance *master) { 2577 sentinelRedisInstance **instance = 2578 zmalloc(sizeof(instance[0])*dictSize(master->slaves)); 2579 sentinelRedisInstance *selected = NULL; 2580 int instances = 0; 2581 dictIterator *di; 2582 dictEntry *de; 2583 mstime_t max_master_down_time = 0; 2584 2585 if (master->flags & SRI_S_DOWN) 2586 max_master_down_time += mstime() - master->s_down_since_time; 2587 max_master_down_time += master->down_after_period * 10; 2588 2589 di = dictGetIterator(master->slaves); 2590 while((de = dictNext(di)) != NULL) { 2591 sentinelRedisInstance *slave = dictGetVal(de); 2592 mstime_t info_validity_time = mstime()-SENTINEL_INFO_VALIDITY_TIME; 2593 2594 if (slave->flags & (SRI_S_DOWN|SRI_O_DOWN|SRI_DISCONNECTED)) continue; 2595 if (slave->last_avail_time < info_validity_time) continue; 2596 if (slave->slave_priority == 0) continue; 2597 2598 /* If the master is in SDOWN state we get INFO for slaves every second. 2599 * Otherwise we get it with the usual period so we need to account for 2600 * a larger delay. */ 2601 if ((master->flags & SRI_S_DOWN) == 0) 2602 info_validity_time -= SENTINEL_INFO_PERIOD; 2603 if (slave->info_refresh < info_validity_time) continue; 2604 if (slave->master_link_down_time > max_master_down_time) continue; 2605 instance[instances++] = slave; 2606 } 2607 dictReleaseIterator(di); 2608 if (instances) { 2609 qsort(instance,instances,sizeof(sentinelRedisInstance*), 2610 compareSlavesForPromotion); 2611 selected = instance[0]; 2612 } 2613 zfree(instance); 2614 return selected; 2615 } 2616 2617 /* ---------------- Failover state machine implementation ------------------- */ 2618 void sentinelFailoverWaitStart(sentinelRedisInstance *ri) { 2619 /* If we in "wait start" but the master is no longer in ODOWN nor in 2620 * SDOWN condition we abort the failover. This is important as it 2621 * prevents a useless failover in a a notable case of netsplit, where 2622 * the sentinels are split from the redis instances. In this case 2623 * the failover will not start while there is the split because no 2624 * good slave can be reached. However when the split is resolved, we 2625 * can go to waitstart if the slave is back reachable a few milliseconds 2626 * before the master is. In that case when the master is back online 2627 * we cancel the failover. */ 2628 if ((ri->flags & (SRI_S_DOWN|SRI_O_DOWN|SRI_FORCE_FAILOVER)) == 0) { 2629 sentinelEvent(REDIS_WARNING,"-failover-abort-master-is-back", 2630 ri,"%@"); 2631 sentinelAbortFailover(ri); 2632 return; 2633 } 2634 2635 /* Start the failover going to the next state if enough time has 2636 * elapsed. */ 2637 if (mstime() >= ri->failover_start_time) { 2638 ri->failover_state = SENTINEL_FAILOVER_STATE_SELECT_SLAVE; 2639 ri->failover_state_change_time = mstime(); 2640 sentinelEvent(REDIS_WARNING,"+failover-state-select-slave",ri,"%@"); 2641 } 2642 } 2643 2644 void sentinelFailoverSelectSlave(sentinelRedisInstance *ri) { 2645 sentinelRedisInstance *slave = sentinelSelectSlave(ri); 2646 2647 if (slave == NULL) { 2648 sentinelEvent(REDIS_WARNING,"-failover-abort-no-good-slave",ri,"%@"); 2649 sentinelAbortFailover(ri); 2650 } else { 2651 sentinelEvent(REDIS_WARNING,"+selected-slave",slave,"%@"); 2652 slave->flags |= SRI_PROMOTED; 2653 ri->promoted_slave = slave; 2654 ri->failover_state = SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE; 2655 ri->failover_state_change_time = mstime(); 2656 sentinelEvent(REDIS_NOTICE,"+failover-state-send-slaveof-noone", 2657 slave, "%@"); 2658 } 2659 } 2660 2661 void sentinelFailoverSendSlaveOfNoOne(sentinelRedisInstance *ri) { 2662 int retval; 2663 2664 if (ri->promoted_slave->flags & SRI_DISCONNECTED) return; 2665 2666 /* Send SLAVEOF NO ONE command to turn the slave into a master. 2667 * We actually register a generic callback for this command as we don't 2668 * really care about the reply. We check if it worked indirectly observing 2669 * if INFO returns a different role (master instead of slave). */ 2670 retval = redisAsyncCommand(ri->promoted_slave->cc, 2671 sentinelDiscardReplyCallback, NULL, "SLAVEOF NO ONE"); 2672 if (retval != REDIS_OK) return; 2673 ri->promoted_slave->pending_commands++; 2674 sentinelEvent(REDIS_NOTICE, "+failover-state-wait-promotion", 2675 ri->promoted_slave,"%@"); 2676 ri->failover_state = SENTINEL_FAILOVER_STATE_WAIT_PROMOTION; 2677 ri->failover_state_change_time = mstime(); 2678 } 2679 2680 /* We actually wait for promotion indirectly checking with INFO when the 2681 * slave turns into a master. */ 2682 void sentinelFailoverWaitPromotion(sentinelRedisInstance *ri) { 2683 mstime_t elapsed = mstime() - ri->failover_state_change_time; 2684 2685 if (elapsed >= SENTINEL_PROMOTION_RETRY_PERIOD) { 2686 sentinelEvent(REDIS_WARNING,"-promotion-timeout",ri->promoted_slave, 2687 "%@"); 2688 sentinelEvent(REDIS_WARNING,"+failover-state-select-slave",ri,"%@"); 2689 ri->failover_state = SENTINEL_FAILOVER_STATE_SELECT_SLAVE; 2690 ri->failover_state_change_time = mstime(); 2691 ri->promoted_slave->flags &= ~SRI_PROMOTED; 2692 ri->promoted_slave = NULL; 2693 } 2694 } 2695 2696 void sentinelFailoverDetectEnd(sentinelRedisInstance *master) { 2697 int not_reconfigured = 0, timeout = 0; 2698 dictIterator *di; 2699 dictEntry *de; 2700 mstime_t elapsed = mstime() - master->failover_state_change_time; 2701 2702 /* We can't consider failover finished if the promoted slave is 2703 * not reachable. */ 2704 if (master->promoted_slave == NULL || 2705 master->promoted_slave->flags & SRI_S_DOWN) return; 2706 2707 /* The failover terminates once all the reachable slaves are properly 2708 * configured. */ 2709 di = dictGetIterator(master->slaves); 2710 while((de = dictNext(di)) != NULL) { 2711 sentinelRedisInstance *slave = dictGetVal(de); 2712 2713 if (slave->flags & (SRI_PROMOTED|SRI_RECONF_DONE)) continue; 2714 if (slave->flags & SRI_S_DOWN) continue; 2715 not_reconfigured++; 2716 } 2717 dictReleaseIterator(di); 2718 2719 /* Force end of failover on timeout. */ 2720 if (elapsed > master->failover_timeout) { 2721 not_reconfigured = 0; 2722 timeout = 1; 2723 sentinelEvent(REDIS_WARNING,"+failover-end-for-timeout",master,"%@"); 2724 } 2725 2726 if (not_reconfigured == 0) { 2727 int role = (master->flags & SRI_I_AM_THE_LEADER) ? SENTINEL_LEADER : 2728 SENTINEL_OBSERVER; 2729 2730 sentinelEvent(REDIS_WARNING,"+failover-end",master,"%@"); 2731 master->failover_state = SENTINEL_FAILOVER_STATE_UPDATE_CONFIG; 2732 master->failover_state_change_time = mstime(); 2733 sentinelCallClientReconfScript(master,role,"end",master->addr, 2734 master->promoted_slave->addr); 2735 } 2736 2737 /* If I'm the leader it is a good idea to send a best effort SLAVEOF 2738 * command to all the slaves still not reconfigured to replicate with 2739 * the new master. */ 2740 if (timeout && (master->flags & SRI_I_AM_THE_LEADER)) { 2741 dictIterator *di; 2742 dictEntry *de; 2743 char master_port[32]; 2744 2745 ll2string(master_port,sizeof(master_port), 2746 master->promoted_slave->addr->port); 2747 2748 di = dictGetIterator(master->slaves); 2749 while((de = dictNext(di)) != NULL) { 2750 sentinelRedisInstance *slave = dictGetVal(de); 2751 int retval; 2752 2753 if (slave->flags & 2754 (SRI_RECONF_DONE|SRI_RECONF_SENT|SRI_DISCONNECTED)) continue; 2755 2756 retval = redisAsyncCommand(slave->cc, 2757 sentinelDiscardReplyCallback, NULL, "SLAVEOF %s %s", 2758 master->promoted_slave->addr->ip, 2759 master_port); 2760 if (retval == REDIS_OK) { 2761 sentinelEvent(REDIS_NOTICE,"+slave-reconf-sent-be",slave,"%@"); 2762 slave->flags |= SRI_RECONF_SENT; 2763 } 2764 } 2765 dictReleaseIterator(di); 2766 } 2767 } 2768 2769 /* Send SLAVE OF <new master address> to all the remaining slaves that 2770 * still don't appear to have the configuration updated. */ 2771 void sentinelFailoverReconfNextSlave(sentinelRedisInstance *master) { 2772 dictIterator *di; 2773 dictEntry *de; 2774 int in_progress = 0; 2775 2776 di = dictGetIterator(master->slaves); 2777 while((de = dictNext(di)) != NULL) { 2778 sentinelRedisInstance *slave = dictGetVal(de); 2779 2780 if (slave->flags & (SRI_RECONF_SENT|SRI_RECONF_INPROG)) 2781 in_progress++; 2782 } 2783 dictReleaseIterator(di); 2784 2785 di = dictGetIterator(master->slaves); 2786 while(in_progress < master->parallel_syncs && 2787 (de = dictNext(di)) != NULL) 2788 { 2789 sentinelRedisInstance *slave = dictGetVal(de); 2790 int retval; 2791 char master_port[32]; 2792 2793 /* Skip the promoted slave, and already configured slaves. */ 2794 if (slave->flags & (SRI_PROMOTED|SRI_RECONF_DONE)) continue; 2795 2796 /* Clear the SRI_RECONF_SENT flag if too much time elapsed without 2797 * the slave moving forward to the next state. */ 2798 if ((slave->flags & SRI_RECONF_SENT) && 2799 (mstime() - slave->slave_reconf_sent_time) > 2800 SENTINEL_SLAVE_RECONF_RETRY_PERIOD) 2801 { 2802 sentinelEvent(REDIS_NOTICE,"-slave-reconf-sent-timeout",slave,"%@"); 2803 slave->flags &= ~SRI_RECONF_SENT; 2804 } 2805 2806 /* Nothing to do for instances that are disconnected or already 2807 * in RECONF_SENT state. */ 2808 if (slave->flags & (SRI_DISCONNECTED|SRI_RECONF_SENT|SRI_RECONF_INPROG)) 2809 continue; 2810 2811 /* Send SLAVEOF <new master>. */ 2812 ll2string(master_port,sizeof(master_port), 2813 master->promoted_slave->addr->port); 2814 retval = redisAsyncCommand(slave->cc, 2815 sentinelDiscardReplyCallback, NULL, "SLAVEOF %s %s", 2816 master->promoted_slave->addr->ip, 2817 master_port); 2818 if (retval == REDIS_OK) { 2819 slave->flags |= SRI_RECONF_SENT; 2820 slave->pending_commands++; 2821 slave->slave_reconf_sent_time = mstime(); 2822 sentinelEvent(REDIS_NOTICE,"+slave-reconf-sent",slave,"%@"); 2823 in_progress++; 2824 } 2825 } 2826 dictReleaseIterator(di); 2827 sentinelFailoverDetectEnd(master); 2828 } 2829 2830 /* This function is called when the slave is in 2831 * SENTINEL_FAILOVER_STATE_UPDATE_CONFIG state. In this state we need 2832 * to remove it from the master table and add the promoted slave instead. 2833 * 2834 * If there are no promoted slaves as this instance is unique, we remove 2835 * and re-add it with the same address to trigger a complete state 2836 * refresh. */ 2837 void sentinelFailoverSwitchToPromotedSlave(sentinelRedisInstance *master) { 2838 sentinelRedisInstance *ref = master->promoted_slave ? 2839 master->promoted_slave : master; 2840 2841 sentinelEvent(REDIS_WARNING,"+switch-master",master,"%s %s %d %s %d", 2842 master->name, master->addr->ip, master->addr->port, 2843 ref->addr->ip, ref->addr->port); 2844 2845 sentinelResetMasterAndChangeAddress(master,ref->addr->ip,ref->addr->port); 2846 } 2847 2848 void sentinelFailoverStateMachine(sentinelRedisInstance *ri) { 2849 redisAssert(ri->flags & SRI_MASTER); 2850 2851 if (!(ri->flags & SRI_FAILOVER_IN_PROGRESS)) return; 2852 2853 switch(ri->failover_state) { 2854 case SENTINEL_FAILOVER_STATE_WAIT_START: 2855 sentinelFailoverWaitStart(ri); 2856 break; 2857 case SENTINEL_FAILOVER_STATE_SELECT_SLAVE: 2858 sentinelFailoverSelectSlave(ri); 2859 break; 2860 case SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE: 2861 sentinelFailoverSendSlaveOfNoOne(ri); 2862 break; 2863 case SENTINEL_FAILOVER_STATE_WAIT_PROMOTION: 2864 sentinelFailoverWaitPromotion(ri); 2865 break; 2866 case SENTINEL_FAILOVER_STATE_RECONF_SLAVES: 2867 sentinelFailoverReconfNextSlave(ri); 2868 break; 2869 case SENTINEL_FAILOVER_STATE_DETECT_END: 2870 sentinelFailoverDetectEnd(ri); 2871 break; 2872 } 2873 } 2874 2875 /* Abort a failover in progress with the following steps: 2876 * 1) If this instance is the leaer send a SLAVEOF command to all the already 2877 * reconfigured slaves if any to configure them to replicate with the 2878 * original master. 2879 * 2) For both leaders and observers: clear the failover flags and state in 2880 * the master instance. 2881 * 3) If there is already a promoted slave and we are the leader, and this 2882 * slave is not DISCONNECTED, try to reconfigure it to replicate 2883 * back to the master as well, sending a best effort SLAVEOF command. 2884 */ 2885 void sentinelAbortFailover(sentinelRedisInstance *ri) { 2886 char master_port[32]; 2887 dictIterator *di; 2888 dictEntry *de; 2889 int sentinel_role; 2890 2891 redisAssert(ri->flags & SRI_FAILOVER_IN_PROGRESS); 2892 ll2string(master_port,sizeof(master_port),ri->addr->port); 2893 2894 /* Clear failover related flags from slaves. 2895 * Also if we are the leader make sure to send SLAVEOF commands to all the 2896 * already reconfigured slaves in order to turn them back into slaves of 2897 * the original master. */ 2898 di = dictGetIterator(ri->slaves); 2899 while((de = dictNext(di)) != NULL) { 2900 sentinelRedisInstance *slave = dictGetVal(de); 2901 if ((ri->flags & SRI_I_AM_THE_LEADER) && 2902 !(slave->flags & SRI_DISCONNECTED) && 2903 (slave->flags & (SRI_PROMOTED|SRI_RECONF_SENT|SRI_RECONF_INPROG| 2904 SRI_RECONF_DONE))) 2905 { 2906 int retval; 2907 2908 retval = redisAsyncCommand(slave->cc, 2909 sentinelDiscardReplyCallback, NULL, "SLAVEOF %s %s", 2910 ri->addr->ip, 2911 master_port); 2912 if (retval == REDIS_OK) 2913 sentinelEvent(REDIS_NOTICE,"-slave-reconf-undo",slave,"%@"); 2914 } 2915 slave->flags &= ~(SRI_RECONF_SENT|SRI_RECONF_INPROG|SRI_RECONF_DONE); 2916 } 2917 dictReleaseIterator(di); 2918 2919 sentinel_role = (ri->flags & SRI_I_AM_THE_LEADER) ? SENTINEL_LEADER : 2920 SENTINEL_OBSERVER; 2921 ri->flags &= ~(SRI_FAILOVER_IN_PROGRESS|SRI_I_AM_THE_LEADER|SRI_FORCE_FAILOVER); 2922 ri->failover_state = SENTINEL_FAILOVER_STATE_NONE; 2923 ri->failover_state_change_time = mstime(); 2924 if (ri->promoted_slave) { 2925 sentinelCallClientReconfScript(ri,sentinel_role,"abort", 2926 ri->promoted_slave->addr,ri->addr); 2927 ri->promoted_slave->flags &= ~SRI_PROMOTED; 2928 ri->promoted_slave = NULL; 2929 } 2930 } 2931 2932 /* The following is called only for master instances and will abort the 2933 * failover process if: 2934 * 2935 * 1) The failover is in progress. 2936 * 2) We already promoted a slave. 2937 * 3) The promoted slave is in extended SDOWN condition. 2938 */ 2939 void sentinelAbortFailoverIfNeeded(sentinelRedisInstance *ri) { 2940 /* Failover is in progress? Do we have a promoted slave? */ 2941 if (!(ri->flags & SRI_FAILOVER_IN_PROGRESS) || !ri->promoted_slave) return; 2942 2943 /* Is the promoted slave into an extended SDOWN state? */ 2944 if (!(ri->promoted_slave->flags & SRI_S_DOWN) || 2945 (mstime() - ri->promoted_slave->s_down_since_time) < 2946 (ri->down_after_period * SENTINEL_EXTENDED_SDOWN_MULTIPLIER)) return; 2947 2948 sentinelEvent(REDIS_WARNING,"-failover-abort-x-sdown",ri->promoted_slave,"%@"); 2949 sentinelAbortFailover(ri); 2950 } 2951 2952 /* ======================== SENTINEL timer handler ========================== 2953 * This is the "main" our Sentinel, being sentinel completely non blocking 2954 * in design. The function is called every second. 2955 * -------------------------------------------------------------------------- */ 2956 2957 /* Perform scheduled operations for the specified Redis instance. */ 2958 void sentinelHandleRedisInstance(sentinelRedisInstance *ri) { 2959 /* ========== MONITORING HALF ============ */ 2960 /* Every kind of instance */ 2961 sentinelReconnectInstance(ri); 2962 sentinelPingInstance(ri); 2963 2964 /* Masters and slaves */ 2965 if (ri->flags & (SRI_MASTER|SRI_SLAVE)) { 2966 /* Nothing so far. */ 2967 } 2968 2969 /* Only masters */ 2970 if (ri->flags & SRI_MASTER) { 2971 sentinelAskMasterStateToOtherSentinels(ri); 2972 } 2973 2974 /* ============== ACTING HALF ============= */ 2975 /* We don't proceed with the acting half if we are in TILT mode. 2976 * TILT happens when we find something odd with the time, like a 2977 * sudden change in the clock. */ 2978 if (sentinel.tilt) { 2979 if (mstime()-sentinel.tilt_start_time < SENTINEL_TILT_PERIOD) return; 2980 sentinel.tilt = 0; 2981 sentinelEvent(REDIS_WARNING,"-tilt",NULL,"#tilt mode exited"); 2982 } 2983 2984 /* Every kind of instance */ 2985 sentinelCheckSubjectivelyDown(ri); 2986 2987 /* Masters and slaves */ 2988 if (ri->flags & (SRI_MASTER|SRI_SLAVE)) { 2989 /* Nothing so far. */ 2990 } 2991 2992 /* Only masters */ 2993 if (ri->flags & SRI_MASTER) { 2994 sentinelCheckObjectivelyDown(ri); 2995 sentinelStartFailoverIfNeeded(ri); 2996 sentinelFailoverStateMachine(ri); 2997 sentinelAbortFailoverIfNeeded(ri); 2998 } 2999 } 3000 3001 /* Perform scheduled operations for all the instances in the dictionary. 3002 * Recursively call the function against dictionaries of slaves. */ 3003 void sentinelHandleDictOfRedisInstances(dict *instances) { 3004 dictIterator *di; 3005 dictEntry *de; 3006 sentinelRedisInstance *switch_to_promoted = NULL; 3007 3008 /* There are a number of things we need to perform against every master. */ 3009 di = dictGetIterator(instances); 3010 while((de = dictNext(di)) != NULL) { 3011 sentinelRedisInstance *ri = dictGetVal(de); 3012 3013 sentinelHandleRedisInstance(ri); 3014 if (ri->flags & SRI_MASTER) { 3015 sentinelHandleDictOfRedisInstances(ri->slaves); 3016 sentinelHandleDictOfRedisInstances(ri->sentinels); 3017 if (ri->failover_state == SENTINEL_FAILOVER_STATE_UPDATE_CONFIG) { 3018 switch_to_promoted = ri; 3019 } 3020 } 3021 } 3022 if (switch_to_promoted) 3023 sentinelFailoverSwitchToPromotedSlave(switch_to_promoted); 3024 dictReleaseIterator(di); 3025 } 3026 3027 /* This function checks if we need to enter the TITL mode. 3028 * 3029 * The TILT mode is entered if we detect that between two invocations of the 3030 * timer interrupt, a negative amount of time, or too much time has passed. 3031 * Note that we expect that more or less just 100 milliseconds will pass 3032 * if everything is fine. However we'll see a negative number or a 3033 * difference bigger than SENTINEL_TILT_TRIGGER milliseconds if one of the 3034 * following conditions happen: 3035 * 3036 * 1) The Sentiel process for some time is blocked, for every kind of 3037 * random reason: the load is huge, the computer was frozen for some time 3038 * in I/O or alike, the process was stopped by a signal. Everything. 3039 * 2) The system clock was altered significantly. 3040 * 3041 * Under both this conditions we'll see everything as timed out and failing 3042 * without good reasons. Instead we enter the TILT mode and wait 3043 * for SENTINEL_TILT_PERIOD to elapse before starting to act again. 3044 * 3045 * During TILT time we still collect information, we just do not act. */ 3046 void sentinelCheckTiltCondition(void) { 3047 mstime_t now = mstime(); 3048 mstime_t delta = now - sentinel.previous_time; 3049 3050 if (delta < 0 || delta > SENTINEL_TILT_TRIGGER) { 3051 sentinel.tilt = 1; 3052 sentinel.tilt_start_time = mstime(); 3053 sentinelEvent(REDIS_WARNING,"+tilt",NULL,"#tilt mode entered"); 3054 } 3055 sentinel.previous_time = mstime(); 3056 } 3057 3058 void sentinelTimer(void) { 3059 sentinelCheckTiltCondition(); 3060 sentinelHandleDictOfRedisInstances(sentinel.masters); 3061 sentinelRunPendingScripts(); 3062 sentinelCollectTerminatedScripts(); 3063 sentinelKillTimedoutScripts(); 3064 } 3065 3066