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