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