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->master->flags & SRI_FAILOVER_IN_PROGRESS) && 1906 (ri->master->failover_state == 1907 SENTINEL_FAILOVER_STATE_WAIT_PROMOTION)) 1908 { 1909 /* Now that we are sure the slave was reconfigured as a master 1910 * set the master configuration epoch to the epoch we won the 1911 * election to perform this failover. This will force the other 1912 * Sentinels to update their config (assuming there is not 1913 * a newer one already available). */ 1914 ri->master->config_epoch = ri->master->failover_epoch; 1915 ri->master->failover_state = SENTINEL_FAILOVER_STATE_RECONF_SLAVES; 1916 ri->master->failover_state_change_time = mstime(); 1917 sentinelFlushConfig(); 1918 sentinelEvent(REDIS_WARNING,"+promoted-slave",ri,"%@"); 1919 sentinelEvent(REDIS_WARNING,"+failover-state-reconf-slaves", 1920 ri->master,"%@"); 1921 sentinelCallClientReconfScript(ri->master,SENTINEL_LEADER, 1922 "start",ri->master->addr,ri->addr); 1923 } else { 1924 /* A slave turned into a master. We want to force our view and 1925 * reconfigure as slave. Wait some time after the change before 1926 * going forward, to receive new configs if any. */ 1927 mstime_t wait_time = SENTINEL_PUBLISH_PERIOD*4; 1928 1929 if (sentinelMasterLooksSane(ri->master) && 1930 sentinelRedisInstanceNoDownFor(ri,wait_time) && 1931 mstime() - ri->role_reported_time > wait_time) 1932 { 1933 int retval = sentinelSendSlaveOf(ri, 1934 ri->master->addr->ip, 1935 ri->master->addr->port); 1936 if (retval == REDIS_OK) 1937 sentinelEvent(REDIS_NOTICE,"+convert-to-slave",ri,"%@"); 1938 } 1939 } 1940 } 1941 1942 /* Handle slaves replicating to a different master address. */ 1943 if ((ri->flags & SRI_SLAVE) && 1944 role == SRI_SLAVE && 1945 (ri->slave_master_port != ri->master->addr->port || 1946 strcasecmp(ri->slave_master_host,ri->master->addr->ip))) 1947 { 1948 mstime_t wait_time = ri->master->failover_timeout; 1949 1950 /* Make sure the master is sane before reconfiguring this instance 1951 * into a slave. */ 1952 if (sentinelMasterLooksSane(ri->master) && 1953 sentinelRedisInstanceNoDownFor(ri,wait_time) && 1954 mstime() - ri->slave_conf_change_time > wait_time) 1955 { 1956 int retval = sentinelSendSlaveOf(ri, 1957 ri->master->addr->ip, 1958 ri->master->addr->port); 1959 if (retval == REDIS_OK) 1960 sentinelEvent(REDIS_NOTICE,"+fix-slave-config",ri,"%@"); 1961 } 1962 } 1963 1964 /* Detect if the slave that is in the process of being reconfigured 1965 * changed state. */ 1966 if ((ri->flags & SRI_SLAVE) && role == SRI_SLAVE && 1967 (ri->flags & (SRI_RECONF_SENT|SRI_RECONF_INPROG))) 1968 { 1969 /* SRI_RECONF_SENT -> SRI_RECONF_INPROG. */ 1970 if ((ri->flags & SRI_RECONF_SENT) && 1971 ri->slave_master_host && 1972 strcmp(ri->slave_master_host, 1973 ri->master->promoted_slave->addr->ip) == 0 && 1974 ri->slave_master_port == ri->master->promoted_slave->addr->port) 1975 { 1976 ri->flags &= ~SRI_RECONF_SENT; 1977 ri->flags |= SRI_RECONF_INPROG; 1978 sentinelEvent(REDIS_NOTICE,"+slave-reconf-inprog",ri,"%@"); 1979 } 1980 1981 /* SRI_RECONF_INPROG -> SRI_RECONF_DONE */ 1982 if ((ri->flags & SRI_RECONF_INPROG) && 1983 ri->slave_master_link_status == SENTINEL_MASTER_LINK_STATUS_UP) 1984 { 1985 ri->flags &= ~SRI_RECONF_INPROG; 1986 ri->flags |= SRI_RECONF_DONE; 1987 sentinelEvent(REDIS_NOTICE,"+slave-reconf-done",ri,"%@"); 1988 } 1989 } 1990 } 1991 1992 void sentinelInfoReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 1993 sentinelRedisInstance *ri = c->data; 1994 redisReply *r; 1995 1996 if (ri) ri->pending_commands--; 1997 if (!reply || !ri) return; 1998 r = reply; 1999 2000 if (r->type == REDIS_REPLY_STRING) { 2001 sentinelRefreshInstanceInfo(ri,r->str); 2002 } 2003 } 2004 2005 /* Just discard the reply. We use this when we are not monitoring the return 2006 * value of the command but its effects directly. */ 2007 void sentinelDiscardReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 2008 sentinelRedisInstance *ri = c->data; 2009 2010 if (ri) ri->pending_commands--; 2011 } 2012 2013 void sentinelPingReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 2014 sentinelRedisInstance *ri = c->data; 2015 redisReply *r; 2016 2017 if (ri) ri->pending_commands--; 2018 if (!reply || !ri) return; 2019 r = reply; 2020 2021 if (r->type == REDIS_REPLY_STATUS || 2022 r->type == REDIS_REPLY_ERROR) { 2023 /* Update the "instance available" field only if this is an 2024 * acceptable reply. */ 2025 if (strncmp(r->str,"PONG",4) == 0 || 2026 strncmp(r->str,"LOADING",7) == 0 || 2027 strncmp(r->str,"MASTERDOWN",10) == 0) 2028 { 2029 ri->last_avail_time = mstime(); 2030 ri->last_ping_time = 0; /* Flag the pong as received. */ 2031 } else { 2032 /* Send a SCRIPT KILL command if the instance appears to be 2033 * down because of a busy script. */ 2034 if (strncmp(r->str,"BUSY",4) == 0 && 2035 (ri->flags & SRI_S_DOWN) && 2036 !(ri->flags & SRI_SCRIPT_KILL_SENT)) 2037 { 2038 if (redisAsyncCommand(ri->cc, 2039 sentinelDiscardReplyCallback, NULL, 2040 "SCRIPT KILL") == REDIS_OK) 2041 ri->pending_commands++; 2042 ri->flags |= SRI_SCRIPT_KILL_SENT; 2043 } 2044 } 2045 } 2046 ri->last_pong_time = mstime(); 2047 } 2048 2049 /* This is called when we get the reply about the PUBLISH command we send 2050 * to the master to advertise this sentinel. */ 2051 void sentinelPublishReplyCallback(redisAsyncContext *c, void *reply, void *privdata) { 2052 sentinelRedisInstance *ri = c->data; 2053 redisReply *r; 2054 2055 if (ri) ri->pending_commands--; 2056 if (!reply || !ri) return; 2057 r = reply; 2058 2059 /* Only update pub_time if we actually published our message. Otherwise 2060 * we'll retry against in 100 milliseconds. */ 2061 if (r->type != REDIS_REPLY_ERROR) 2062 ri->last_pub_time = mstime(); 2063 } 2064 2065 /* Process an hello message received via Pub/Sub in master or slave instance, 2066 * or sent directly to this sentinel via the (fake) PUBLISH command of Sentinel. 2067 * 2068 * If the master name specified in the message is not known, the message is 2069 * discareded. */ 2070 void sentinelProcessHelloMessage(char *hello, int hello_len) { 2071 /* Format is composed of 8 tokens: 2072 * 0=ip,1=port,2=runid,3=current_epoch,4=master_name, 2073 * 5=master_ip,6=master_port,7=master_config_epoch. */ 2074 int numtokens, port, removed, master_port; 2075 uint64_t current_epoch, master_config_epoch; 2076 char **token = sdssplitlen(hello, hello_len, ",", 1, &numtokens); 2077 sentinelRedisInstance *si, *master; 2078 2079 if (numtokens == 8) { 2080 /* Obtain a reference to the master this hello message is about */ 2081 master = sentinelGetMasterByName(token[4]); 2082 if (!master) goto cleanup; /* Unknown master, skip the message. */ 2083 2084 /* First, try to see if we already have this sentinel. */ 2085 port = atoi(token[1]); 2086 master_port = atoi(token[6]); 2087 si = getSentinelRedisInstanceByAddrAndRunID( 2088 master->sentinels,token[0],port,token[2]); 2089 current_epoch = strtoull(token[3],NULL,10); 2090 master_config_epoch = strtoull(token[7],NULL,10); 2091 2092 if (!si) { 2093 /* If not, remove all the sentinels that have the same runid 2094 * OR the same ip/port, because it's either a restart or a 2095 * network topology change. */ 2096 removed = removeMatchingSentinelsFromMaster(master,token[0],port, 2097 token[2]); 2098 if (removed) { 2099 sentinelEvent(REDIS_NOTICE,"-dup-sentinel",master, 2100 "%@ #duplicate of %s:%d or %s", 2101 token[0],port,token[2]); 2102 } 2103 2104 /* Add the new sentinel. */ 2105 si = createSentinelRedisInstance(NULL,SRI_SENTINEL, 2106 token[0],port,master->quorum,master); 2107 if (si) { 2108 sentinelEvent(REDIS_NOTICE,"+sentinel",si,"%@"); 2109 /* The runid is NULL after a new instance creation and 2110 * for Sentinels we don't have a later chance to fill it, 2111 * so do it now. */ 2112 si->runid = sdsnew(token[2]); 2113 sentinelFlushConfig(); 2114 } 2115 } 2116 2117 /* Update local current_epoch if received current_epoch is greater.*/ 2118 if (current_epoch > sentinel.current_epoch) { 2119 sentinel.current_epoch = current_epoch; 2120 sentinelFlushConfig(); 2121 sentinelEvent(REDIS_WARNING,"+new-epoch",master,"%llu", 2122 (unsigned long long) sentinel.current_epoch); 2123 } 2124 2125 /* Update master info if received configuration is newer. */ 2126 if (master->config_epoch < master_config_epoch) { 2127 master->config_epoch = master_config_epoch; 2128 if (master_port != master->addr->port || 2129 strcmp(master->addr->ip, token[5])) 2130 { 2131 sentinelAddr *old_addr; 2132 2133 sentinelEvent(REDIS_WARNING,"+config-update-from",si,"%@"); 2134 sentinelEvent(REDIS_WARNING,"+switch-master", 2135 master,"%s %s %d %s %d", 2136 master->name, 2137 master->addr->ip, master->addr->port, 2138 token[5], master_port); 2139 2140 old_addr = dupSentinelAddr(master->addr); 2141 sentinelResetMasterAndChangeAddress(master, token[5], master_port); 2142 sentinelCallClientReconfScript(master, 2143 SENTINEL_OBSERVER,"start", 2144 old_addr,master->addr); 2145 releaseSentinelAddr(old_addr); 2146 } 2147 } 2148 2149 /* Update the state of the Sentinel. */ 2150 if (si) si->last_hello_time = mstime(); 2151 } 2152 2153 cleanup: 2154 sdsfreesplitres(token,numtokens); 2155 } 2156 2157 2158 /* This is our Pub/Sub callback for the Hello channel. It's useful in order 2159 * to discover other sentinels attached at the same master. */ 2160 void sentinelReceiveHelloMessages(redisAsyncContext *c, void *reply, void *privdata) { 2161 sentinelRedisInstance *ri = c->data; 2162 redisReply *r; 2163 2164 if (!reply || !ri) return; 2165 r = reply; 2166 2167 /* Update the last activity in the pubsub channel. Note that since we 2168 * receive our messages as well this timestamp can be used to detect 2169 * if the link is probably disconnected even if it seems otherwise. */ 2170 ri->pc_last_activity = mstime(); 2171 2172 /* Sanity check in the reply we expect, so that the code that follows 2173 * can avoid to check for details. */ 2174 if (r->type != REDIS_REPLY_ARRAY || 2175 r->elements != 3 || 2176 r->element[0]->type != REDIS_REPLY_STRING || 2177 r->element[1]->type != REDIS_REPLY_STRING || 2178 r->element[2]->type != REDIS_REPLY_STRING || 2179 strcmp(r->element[0]->str,"message") != 0) return; 2180 2181 /* We are not interested in meeting ourselves */ 2182 if (strstr(r->element[2]->str,server.runid) != NULL) return; 2183 2184 sentinelProcessHelloMessage(r->element[2]->str, r->element[2]->len); 2185 } 2186 2187 /* Send an "Hello" message via Pub/Sub to the specified 'ri' Redis 2188 * instance in order to broadcast the current configuraiton for this 2189 * master, and to advertise the existence of this Sentinel at the same time. 2190 * 2191 * The message has the following format: 2192 * 2193 * sentinel_ip,sentinel_port,sentinel_runid,current_epoch, 2194 * master_name,master_ip,master_port,master_config_epoch. 2195 * 2196 * Returns REDIS_OK if the PUBLISH was queued correctly, otherwise 2197 * REDIS_ERR is returned. */ 2198 int sentinelSendHello(sentinelRedisInstance *ri) { 2199 char ip[REDIS_IP_STR_LEN]; 2200 char payload[REDIS_IP_STR_LEN+1024]; 2201 int retval; 2202 sentinelRedisInstance *master = (ri->flags & SRI_MASTER) ? ri : ri->master; 2203 sentinelAddr *master_addr = sentinelGetCurrentMasterAddress(master); 2204 2205 /* Try to obtain our own IP address. */ 2206 if (anetSockName(ri->cc->c.fd,ip,sizeof(ip),NULL) == -1) return REDIS_ERR; 2207 if (ri->flags & SRI_DISCONNECTED) return REDIS_ERR; 2208 2209 /* Format and send the Hello message. */ 2210 snprintf(payload,sizeof(payload), 2211 "%s,%d,%s,%llu," /* Info about this sentinel. */ 2212 "%s,%s,%d,%llu", /* Info about current master. */ 2213 ip, server.port, server.runid, 2214 (unsigned long long) sentinel.current_epoch, 2215 /* --- */ 2216 master->name,master_addr->ip,master_addr->port, 2217 (unsigned long long) master->config_epoch); 2218 retval = redisAsyncCommand(ri->cc, 2219 sentinelPublishReplyCallback, NULL, "PUBLISH %s %s", 2220 SENTINEL_HELLO_CHANNEL,payload); 2221 if (retval != REDIS_OK) return REDIS_ERR; 2222 ri->pending_commands++; 2223 return REDIS_OK; 2224 } 2225 2226 /* Send a PING to the specified instance and refresh the last_ping_time 2227 * if it is zero (that is, if we received a pong for the previous ping). 2228 * 2229 * On error zero is returned, and we can't consider the PING command 2230 * queued in the connection. */ 2231 int sentinelSendPing(sentinelRedisInstance *ri) { 2232 int retval = redisAsyncCommand(ri->cc, 2233 sentinelPingReplyCallback, NULL, "PING"); 2234 if (retval == REDIS_OK) { 2235 ri->pending_commands++; 2236 /* We update the ping time only if we received the pong for 2237 * the previous ping, otherwise we are technically waiting 2238 * since the first ping that did not received a reply. */ 2239 if (ri->last_ping_time == 0) ri->last_ping_time = mstime(); 2240 return 1; 2241 } else { 2242 return 0; 2243 } 2244 } 2245 2246 /* Send periodic PING, INFO, and PUBLISH to the Hello channel to 2247 * the specified master or slave instance. */ 2248 void sentinelSendPeriodicCommands(sentinelRedisInstance *ri) { 2249 mstime_t now = mstime(); 2250 mstime_t info_period, ping_period; 2251 int retval; 2252 2253 /* Return ASAP if we have already a PING or INFO already pending, or 2254 * in the case the instance is not properly connected. */ 2255 if (ri->flags & SRI_DISCONNECTED) return; 2256 2257 /* For INFO, PING, PUBLISH that are not critical commands to send we 2258 * also have a limit of SENTINEL_MAX_PENDING_COMMANDS. We don't 2259 * want to use a lot of memory just because a link is not working 2260 * properly (note that anyway there is a redundant protection about this, 2261 * that is, the link will be disconnected and reconnected if a long 2262 * timeout condition is detected. */ 2263 if (ri->pending_commands >= SENTINEL_MAX_PENDING_COMMANDS) return; 2264 2265 /* If this is a slave of a master in O_DOWN condition we start sending 2266 * it INFO every second, instead of the usual SENTINEL_INFO_PERIOD 2267 * period. In this state we want to closely monitor slaves in case they 2268 * are turned into masters by another Sentinel, or by the sysadmin. */ 2269 if ((ri->flags & SRI_SLAVE) && 2270 (ri->master->flags & (SRI_O_DOWN|SRI_FAILOVER_IN_PROGRESS))) { 2271 info_period = 1000; 2272 } else { 2273 info_period = SENTINEL_INFO_PERIOD; 2274 } 2275 2276 /* We ping instances every time the last received pong is older than 2277 * the configured 'down-after-milliseconds' time, but every second 2278 * anyway if 'down-after-milliseconds' is greater than 1 second. */ 2279 ping_period = ri->down_after_period; 2280 if (ping_period > SENTINEL_PING_PERIOD) ping_period = SENTINEL_PING_PERIOD; 2281 2282 if ((ri->flags & SRI_SENTINEL) == 0 && 2283 (ri->info_refresh == 0 || 2284 (now - ri->info_refresh) > info_period)) 2285 { 2286 /* Send INFO to masters and slaves, not sentinels. */ 2287 retval = redisAsyncCommand(ri->cc, 2288 sentinelInfoReplyCallback, NULL, "INFO"); 2289 if (retval == REDIS_OK) ri->pending_commands++; 2290 } else if ((now - ri->last_pong_time) > ping_period) { 2291 /* Send PING to all the three kinds of instances. */ 2292 sentinelSendPing(ri); 2293 } else if ((now - ri->last_pub_time) > SENTINEL_PUBLISH_PERIOD) { 2294 /* PUBLISH hello messages to all the three kinds of instances. */ 2295 sentinelSendHello(ri); 2296 } 2297 } 2298 2299 /* =========================== SENTINEL command ============================= */ 2300 2301 const char *sentinelFailoverStateStr(int state) { 2302 switch(state) { 2303 case SENTINEL_FAILOVER_STATE_NONE: return "none"; 2304 case SENTINEL_FAILOVER_STATE_WAIT_START: return "wait_start"; 2305 case SENTINEL_FAILOVER_STATE_SELECT_SLAVE: return "select_slave"; 2306 case SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE: return "send_slaveof_noone"; 2307 case SENTINEL_FAILOVER_STATE_WAIT_PROMOTION: return "wait_promotion"; 2308 case SENTINEL_FAILOVER_STATE_RECONF_SLAVES: return "reconf_slaves"; 2309 case SENTINEL_FAILOVER_STATE_UPDATE_CONFIG: return "update_config"; 2310 default: return "unknown"; 2311 } 2312 } 2313 2314 /* Redis instance to Redis protocol representation. */ 2315 void addReplySentinelRedisInstance(redisClient *c, sentinelRedisInstance *ri) { 2316 char *flags = sdsempty(); 2317 void *mbl; 2318 int fields = 0; 2319 2320 mbl = addDeferredMultiBulkLength(c); 2321 2322 addReplyBulkCString(c,"name"); 2323 addReplyBulkCString(c,ri->name); 2324 fields++; 2325 2326 addReplyBulkCString(c,"ip"); 2327 addReplyBulkCString(c,ri->addr->ip); 2328 fields++; 2329 2330 addReplyBulkCString(c,"port"); 2331 addReplyBulkLongLong(c,ri->addr->port); 2332 fields++; 2333 2334 addReplyBulkCString(c,"runid"); 2335 addReplyBulkCString(c,ri->runid ? ri->runid : ""); 2336 fields++; 2337 2338 addReplyBulkCString(c,"flags"); 2339 if (ri->flags & SRI_S_DOWN) flags = sdscat(flags,"s_down,"); 2340 if (ri->flags & SRI_O_DOWN) flags = sdscat(flags,"o_down,"); 2341 if (ri->flags & SRI_MASTER) flags = sdscat(flags,"master,"); 2342 if (ri->flags & SRI_SLAVE) flags = sdscat(flags,"slave,"); 2343 if (ri->flags & SRI_SENTINEL) flags = sdscat(flags,"sentinel,"); 2344 if (ri->flags & SRI_DISCONNECTED) flags = sdscat(flags,"disconnected,"); 2345 if (ri->flags & SRI_MASTER_DOWN) flags = sdscat(flags,"master_down,"); 2346 if (ri->flags & SRI_FAILOVER_IN_PROGRESS) 2347 flags = sdscat(flags,"failover_in_progress,"); 2348 if (ri->flags & SRI_PROMOTED) flags = sdscat(flags,"promoted,"); 2349 if (ri->flags & SRI_RECONF_SENT) flags = sdscat(flags,"reconf_sent,"); 2350 if (ri->flags & SRI_RECONF_INPROG) flags = sdscat(flags,"reconf_inprog,"); 2351 if (ri->flags & SRI_RECONF_DONE) flags = sdscat(flags,"reconf_done,"); 2352 2353 if (sdslen(flags) != 0) sdsrange(flags,0,-2); /* remove last "," */ 2354 addReplyBulkCString(c,flags); 2355 sdsfree(flags); 2356 fields++; 2357 2358 addReplyBulkCString(c,"pending-commands"); 2359 addReplyBulkLongLong(c,ri->pending_commands); 2360 fields++; 2361 2362 if (ri->flags & SRI_FAILOVER_IN_PROGRESS) { 2363 addReplyBulkCString(c,"failover-state"); 2364 addReplyBulkCString(c,(char*)sentinelFailoverStateStr(ri->failover_state)); 2365 fields++; 2366 } 2367 2368 addReplyBulkCString(c,"last-ping-sent"); 2369 addReplyBulkLongLong(c, 2370 ri->last_ping_time ? (mstime() - ri->last_ping_time) : 0); 2371 fields++; 2372 2373 addReplyBulkCString(c,"last-ok-ping-reply"); 2374 addReplyBulkLongLong(c,mstime() - ri->last_avail_time); 2375 fields++; 2376 2377 addReplyBulkCString(c,"last-ping-reply"); 2378 addReplyBulkLongLong(c,mstime() - ri->last_pong_time); 2379 fields++; 2380 2381 if (ri->flags & SRI_S_DOWN) { 2382 addReplyBulkCString(c,"s-down-time"); 2383 addReplyBulkLongLong(c,mstime()-ri->s_down_since_time); 2384 fields++; 2385 } 2386 2387 if (ri->flags & SRI_O_DOWN) { 2388 addReplyBulkCString(c,"o-down-time"); 2389 addReplyBulkLongLong(c,mstime()-ri->o_down_since_time); 2390 fields++; 2391 } 2392 2393 addReplyBulkCString(c,"down-after-milliseconds"); 2394 addReplyBulkLongLong(c,ri->down_after_period); 2395 fields++; 2396 2397 /* Masters and Slaves */ 2398 if (ri->flags & (SRI_MASTER|SRI_SLAVE)) { 2399 addReplyBulkCString(c,"info-refresh"); 2400 addReplyBulkLongLong(c,mstime() - ri->info_refresh); 2401 fields++; 2402 2403 addReplyBulkCString(c,"role-reported"); 2404 addReplyBulkCString(c, (ri->role_reported == SRI_MASTER) ? "master" : 2405 "slave"); 2406 fields++; 2407 2408 addReplyBulkCString(c,"role-reported-time"); 2409 addReplyBulkLongLong(c,mstime() - ri->role_reported_time); 2410 fields++; 2411 } 2412 2413 /* Only masters */ 2414 if (ri->flags & SRI_MASTER) { 2415 addReplyBulkCString(c,"config-epoch"); 2416 addReplyBulkLongLong(c,ri->config_epoch); 2417 fields++; 2418 2419 addReplyBulkCString(c,"num-slaves"); 2420 addReplyBulkLongLong(c,dictSize(ri->slaves)); 2421 fields++; 2422 2423 addReplyBulkCString(c,"num-other-sentinels"); 2424 addReplyBulkLongLong(c,dictSize(ri->sentinels)); 2425 fields++; 2426 2427 addReplyBulkCString(c,"quorum"); 2428 addReplyBulkLongLong(c,ri->quorum); 2429 fields++; 2430 2431 addReplyBulkCString(c,"failover-timeout"); 2432 addReplyBulkLongLong(c,ri->failover_timeout); 2433 fields++; 2434 2435 addReplyBulkCString(c,"parallel-syncs"); 2436 addReplyBulkLongLong(c,ri->parallel_syncs); 2437 fields++; 2438 2439 if (ri->notification_script) { 2440 addReplyBulkCString(c,"notification-script"); 2441 addReplyBulkCString(c,ri->notification_script); 2442 fields++; 2443 } 2444 2445 if (ri->client_reconfig_script) { 2446 addReplyBulkCString(c,"client-reconfig-script"); 2447 addReplyBulkCString(c,ri->client_reconfig_script); 2448 fields++; 2449 } 2450 } 2451 2452 /* Only slaves */ 2453 if (ri->flags & SRI_SLAVE) { 2454 addReplyBulkCString(c,"master-link-down-time"); 2455 addReplyBulkLongLong(c,ri->master_link_down_time); 2456 fields++; 2457 2458 addReplyBulkCString(c,"master-link-status"); 2459 addReplyBulkCString(c, 2460 (ri->slave_master_link_status == SENTINEL_MASTER_LINK_STATUS_UP) ? 2461 "ok" : "err"); 2462 fields++; 2463 2464 addReplyBulkCString(c,"master-host"); 2465 addReplyBulkCString(c, 2466 ri->slave_master_host ? ri->slave_master_host : "?"); 2467 fields++; 2468 2469 addReplyBulkCString(c,"master-port"); 2470 addReplyBulkLongLong(c,ri->slave_master_port); 2471 fields++; 2472 2473 addReplyBulkCString(c,"slave-priority"); 2474 addReplyBulkLongLong(c,ri->slave_priority); 2475 fields++; 2476 2477 addReplyBulkCString(c,"slave-repl-offset"); 2478 addReplyBulkLongLong(c,ri->slave_repl_offset); 2479 fields++; 2480 } 2481 2482 /* Only sentinels */ 2483 if (ri->flags & SRI_SENTINEL) { 2484 addReplyBulkCString(c,"last-hello-message"); 2485 addReplyBulkLongLong(c,mstime() - ri->last_hello_time); 2486 fields++; 2487 2488 addReplyBulkCString(c,"voted-leader"); 2489 addReplyBulkCString(c,ri->leader ? ri->leader : "?"); 2490 fields++; 2491 2492 addReplyBulkCString(c,"voted-leader-epoch"); 2493 addReplyBulkLongLong(c,ri->leader_epoch); 2494 fields++; 2495 } 2496 2497 setDeferredMultiBulkLength(c,mbl,fields*2); 2498 } 2499 2500 /* Output a number of instances contained inside a dictionary as 2501 * Redis protocol. */ 2502 void addReplyDictOfRedisInstances(redisClient *c, dict *instances) { 2503 dictIterator *di; 2504 dictEntry *de; 2505 2506 di = dictGetIterator(instances); 2507 addReplyMultiBulkLen(c,dictSize(instances)); 2508 while((de = dictNext(di)) != NULL) { 2509 sentinelRedisInstance *ri = dictGetVal(de); 2510 2511 addReplySentinelRedisInstance(c,ri); 2512 } 2513 dictReleaseIterator(di); 2514 } 2515 2516 /* Lookup the named master into sentinel.masters. 2517 * If the master is not found reply to the client with an error and returns 2518 * NULL. */ 2519 sentinelRedisInstance *sentinelGetMasterByNameOrReplyError(redisClient *c, 2520 robj *name) 2521 { 2522 sentinelRedisInstance *ri; 2523 2524 ri = dictFetchValue(sentinel.masters,c->argv[2]->ptr); 2525 if (!ri) { 2526 addReplyError(c,"No such master with that name"); 2527 return NULL; 2528 } 2529 return ri; 2530 } 2531 2532 void sentinelCommand(redisClient *c) { 2533 if (!strcasecmp(c->argv[1]->ptr,"masters")) { 2534 /* SENTINEL MASTERS */ 2535 if (c->argc != 2) goto numargserr; 2536 addReplyDictOfRedisInstances(c,sentinel.masters); 2537 } else if (!strcasecmp(c->argv[1]->ptr,"master")) { 2538 /* SENTINEL MASTER <name> */ 2539 sentinelRedisInstance *ri; 2540 2541 if (c->argc != 3) goto numargserr; 2542 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) 2543 == NULL) return; 2544 addReplySentinelRedisInstance(c,ri); 2545 } else if (!strcasecmp(c->argv[1]->ptr,"slaves")) { 2546 /* SENTINEL SLAVES <master-name> */ 2547 sentinelRedisInstance *ri; 2548 2549 if (c->argc != 3) goto numargserr; 2550 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) == NULL) 2551 return; 2552 addReplyDictOfRedisInstances(c,ri->slaves); 2553 } else if (!strcasecmp(c->argv[1]->ptr,"sentinels")) { 2554 /* SENTINEL SENTINELS <master-name> */ 2555 sentinelRedisInstance *ri; 2556 2557 if (c->argc != 3) goto numargserr; 2558 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) == NULL) 2559 return; 2560 addReplyDictOfRedisInstances(c,ri->sentinels); 2561 } else if (!strcasecmp(c->argv[1]->ptr,"is-master-down-by-addr")) { 2562 /* SENTINEL IS-MASTER-DOWN-BY-ADDR <ip> <port> <current-epoch> <runid>*/ 2563 sentinelRedisInstance *ri; 2564 long long req_epoch; 2565 uint64_t leader_epoch = 0; 2566 char *leader = NULL; 2567 long port; 2568 int isdown = 0; 2569 2570 if (c->argc != 6) goto numargserr; 2571 if (getLongFromObjectOrReply(c,c->argv[3],&port,NULL) != REDIS_OK || 2572 getLongLongFromObjectOrReply(c,c->argv[4],&req_epoch,NULL) 2573 != REDIS_OK) 2574 return; 2575 ri = getSentinelRedisInstanceByAddrAndRunID(sentinel.masters, 2576 c->argv[2]->ptr,port,NULL); 2577 2578 /* It exists? Is actually a master? Is subjectively down? It's down. 2579 * Note: if we are in tilt mode we always reply with "0". */ 2580 if (!sentinel.tilt && ri && (ri->flags & SRI_S_DOWN) && 2581 (ri->flags & SRI_MASTER)) 2582 isdown = 1; 2583 2584 /* Vote for the master (or fetch the previous vote) if the request 2585 * includes a runid, otherwise the sender is not seeking for a vote. */ 2586 if (ri && ri->flags & SRI_MASTER && strcasecmp(c->argv[5]->ptr,"*")) { 2587 leader = sentinelVoteLeader(ri,(uint64_t)req_epoch, 2588 c->argv[5]->ptr, 2589 &leader_epoch); 2590 } 2591 2592 /* Reply with a three-elements multi-bulk reply: 2593 * down state, leader, vote epoch. */ 2594 addReplyMultiBulkLen(c,3); 2595 addReply(c, isdown ? shared.cone : shared.czero); 2596 addReplyBulkCString(c, leader ? leader : "*"); 2597 addReplyLongLong(c, (long long)leader_epoch); 2598 if (leader) sdsfree(leader); 2599 } else if (!strcasecmp(c->argv[1]->ptr,"reset")) { 2600 /* SENTINEL RESET <pattern> */ 2601 if (c->argc != 3) goto numargserr; 2602 addReplyLongLong(c,sentinelResetMastersByPattern(c->argv[2]->ptr,SENTINEL_GENERATE_EVENT)); 2603 } else if (!strcasecmp(c->argv[1]->ptr,"get-master-addr-by-name")) { 2604 /* SENTINEL GET-MASTER-ADDR-BY-NAME <master-name> */ 2605 sentinelRedisInstance *ri; 2606 2607 if (c->argc != 3) goto numargserr; 2608 ri = sentinelGetMasterByName(c->argv[2]->ptr); 2609 if (ri == NULL) { 2610 addReply(c,shared.nullmultibulk); 2611 } else { 2612 sentinelAddr *addr = sentinelGetCurrentMasterAddress(ri); 2613 2614 addReplyMultiBulkLen(c,2); 2615 addReplyBulkCString(c,addr->ip); 2616 addReplyBulkLongLong(c,addr->port); 2617 } 2618 } else if (!strcasecmp(c->argv[1]->ptr,"failover")) { 2619 /* SENTINEL FAILOVER <master-name> */ 2620 sentinelRedisInstance *ri; 2621 2622 if (c->argc != 3) goto numargserr; 2623 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) == NULL) 2624 return; 2625 if (ri->flags & SRI_FAILOVER_IN_PROGRESS) { 2626 addReplySds(c,sdsnew("-INPROG Failover already in progress\r\n")); 2627 return; 2628 } 2629 if (sentinelSelectSlave(ri) == NULL) { 2630 addReplySds(c,sdsnew("-NOGOODSLAVE No suitable slave to promote\r\n")); 2631 return; 2632 } 2633 redisLog(REDIS_WARNING,"Executing user requested FAILOVER of '%s'", 2634 ri->name); 2635 sentinelStartFailover(ri); 2636 ri->flags |= SRI_FORCE_FAILOVER; 2637 addReply(c,shared.ok); 2638 } else if (!strcasecmp(c->argv[1]->ptr,"pending-scripts")) { 2639 /* SENTINEL PENDING-SCRIPTS */ 2640 2641 if (c->argc != 2) goto numargserr; 2642 sentinelPendingScriptsCommand(c); 2643 } else if (!strcasecmp(c->argv[1]->ptr,"monitor")) { 2644 /* SENTINEL MONITOR <name> <ip> <port> <quorum> */ 2645 sentinelRedisInstance *ri; 2646 long quorum, port; 2647 char buf[32]; 2648 2649 if (c->argc != 6) goto numargserr; 2650 if (getLongFromObjectOrReply(c,c->argv[5],&quorum,"Invalid quorum") 2651 != REDIS_OK) return; 2652 if (getLongFromObjectOrReply(c,c->argv[4],&port,"Invalid port") 2653 != REDIS_OK) return; 2654 /* Make sure the IP field is actually a valid IP before passing it 2655 * to createSentinelRedisInstance(), otherwise we may trigger a 2656 * DNS lookup at runtime. */ 2657 if (anetResolveIP(NULL,c->argv[3]->ptr,buf,sizeof(buf)) == ANET_ERR) { 2658 addReplyError(c,"Invalid IP address specified"); 2659 return; 2660 } 2661 2662 /* Parameters are valid. Try to create the master instance. */ 2663 ri = createSentinelRedisInstance(c->argv[2]->ptr,SRI_MASTER, 2664 c->argv[3]->ptr,port,quorum,NULL); 2665 if (ri == NULL) { 2666 switch(errno) { 2667 case EBUSY: 2668 addReplyError(c,"Duplicated master name"); 2669 break; 2670 case EINVAL: 2671 addReplyError(c,"Invalid port number"); 2672 break; 2673 default: 2674 addReplyError(c,"Unspecified error adding the instance"); 2675 break; 2676 } 2677 } else { 2678 sentinelFlushConfig(); 2679 sentinelEvent(REDIS_WARNING,"+monitor",ri,"%@ quorum %d",ri->quorum); 2680 addReply(c,shared.ok); 2681 } 2682 } else if (!strcasecmp(c->argv[1]->ptr,"remove")) { 2683 /* SENTINEL REMOVE <name> */ 2684 sentinelRedisInstance *ri; 2685 2686 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) 2687 == NULL) return; 2688 sentinelEvent(REDIS_WARNING,"-monitor",ri,"%@"); 2689 dictDelete(sentinel.masters,c->argv[2]->ptr); 2690 sentinelFlushConfig(); 2691 addReply(c,shared.ok); 2692 } else if (!strcasecmp(c->argv[1]->ptr,"set")) { 2693 if (c->argc < 3 || c->argc % 2 == 0) goto numargserr; 2694 sentinelSetCommand(c); 2695 } else { 2696 addReplyErrorFormat(c,"Unknown sentinel subcommand '%s'", 2697 (char*)c->argv[1]->ptr); 2698 } 2699 return; 2700 2701 numargserr: 2702 addReplyErrorFormat(c,"Wrong number of arguments for 'sentinel %s'", 2703 (char*)c->argv[1]->ptr); 2704 } 2705 2706 /* SENTINEL INFO [section] */ 2707 void sentinelInfoCommand(redisClient *c) { 2708 char *section = c->argc == 2 ? c->argv[1]->ptr : "default"; 2709 sds info = sdsempty(); 2710 int defsections = !strcasecmp(section,"default"); 2711 int sections = 0; 2712 2713 if (c->argc > 2) { 2714 addReply(c,shared.syntaxerr); 2715 return; 2716 } 2717 2718 if (!strcasecmp(section,"server") || defsections) { 2719 if (sections++) info = sdscat(info,"\r\n"); 2720 sds serversection = genRedisInfoString("server"); 2721 info = sdscatlen(info,serversection,sdslen(serversection)); 2722 sdsfree(serversection); 2723 } 2724 2725 if (!strcasecmp(section,"sentinel") || defsections) { 2726 dictIterator *di; 2727 dictEntry *de; 2728 int master_id = 0; 2729 2730 if (sections++) info = sdscat(info,"\r\n"); 2731 info = sdscatprintf(info, 2732 "# Sentinel\r\n" 2733 "sentinel_masters:%lu\r\n" 2734 "sentinel_tilt:%d\r\n" 2735 "sentinel_running_scripts:%d\r\n" 2736 "sentinel_scripts_queue_length:%ld\r\n", 2737 dictSize(sentinel.masters), 2738 sentinel.tilt, 2739 sentinel.running_scripts, 2740 listLength(sentinel.scripts_queue)); 2741 2742 di = dictGetIterator(sentinel.masters); 2743 while((de = dictNext(di)) != NULL) { 2744 sentinelRedisInstance *ri = dictGetVal(de); 2745 char *status = "ok"; 2746 2747 if (ri->flags & SRI_O_DOWN) status = "odown"; 2748 else if (ri->flags & SRI_S_DOWN) status = "sdown"; 2749 info = sdscatprintf(info, 2750 "master%d:name=%s,status=%s,address=%s:%d," 2751 "slaves=%lu,sentinels=%lu\r\n", 2752 master_id++, ri->name, status, 2753 ri->addr->ip, ri->addr->port, 2754 dictSize(ri->slaves), 2755 dictSize(ri->sentinels)+1); 2756 } 2757 dictReleaseIterator(di); 2758 } 2759 2760 addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n", 2761 (unsigned long)sdslen(info))); 2762 addReplySds(c,info); 2763 addReply(c,shared.crlf); 2764 } 2765 2766 /* SENTINEL SET <mastername> [<option> <value> ...] */ 2767 void sentinelSetCommand(redisClient *c) { 2768 sentinelRedisInstance *ri; 2769 int j, changes = 0; 2770 char *option, *value; 2771 2772 if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2])) 2773 == NULL) return; 2774 2775 /* Process option - value pairs. */ 2776 for (j = 3; j < c->argc; j += 2) { 2777 option = c->argv[j]->ptr; 2778 value = c->argv[j+1]->ptr; 2779 robj *o = c->argv[j+1]; 2780 long long ll; 2781 2782 if (!strcasecmp(option,"down-after-milliseconds")) { 2783 /* down-after-millisecodns <milliseconds> */ 2784 if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll <= 0) 2785 goto badfmt; 2786 ri->down_after_period = ll; 2787 sentinelPropagateDownAfterPeriod(ri); 2788 changes++; 2789 } else if (!strcasecmp(option,"failover-timeout")) { 2790 /* failover-timeout <milliseconds> */ 2791 if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll <= 0) 2792 goto badfmt; 2793 ri->failover_timeout = ll; 2794 changes++; 2795 } else if (!strcasecmp(option,"parallel-syncs")) { 2796 /* parallel-syncs <milliseconds> */ 2797 if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll <= 0) 2798 goto badfmt; 2799 ri->parallel_syncs = ll; 2800 changes++; 2801 } else if (!strcasecmp(option,"notification-script")) { 2802 /* notification-script <path> */ 2803 if (strlen(value) && access(value,X_OK) == -1) { 2804 addReplyError(c, 2805 "Notification script seems non existing or non executable"); 2806 if (changes) sentinelFlushConfig(); 2807 return; 2808 } 2809 sdsfree(ri->notification_script); 2810 ri->notification_script = strlen(value) ? sdsnew(value) : NULL; 2811 changes++; 2812 } else if (!strcasecmp(option,"client-reconfig-script")) { 2813 /* client-reconfig-script <path> */ 2814 if (strlen(value) && access(value,X_OK) == -1) { 2815 addReplyError(c, 2816 "Client reconfiguration script seems non existing or " 2817 "non executable"); 2818 if (changes) sentinelFlushConfig(); 2819 return; 2820 } 2821 sdsfree(ri->client_reconfig_script); 2822 ri->client_reconfig_script = strlen(value) ? sdsnew(value) : NULL; 2823 changes++; 2824 } else if (!strcasecmp(option,"auth-pass")) { 2825 /* auth-pass <password> */ 2826 sdsfree(ri->auth_pass); 2827 ri->auth_pass = strlen(value) ? sdsnew(value) : NULL; 2828 changes++; 2829 } else if (!strcasecmp(option,"quorum")) { 2830 /* quorum <count> */ 2831 if (getLongLongFromObject(o,&ll) == REDIS_ERR || ll <= 0) 2832 goto badfmt; 2833 ri->quorum = ll; 2834 changes++; 2835 } else { 2836 addReplyErrorFormat(c,"Unknown option '%s' for SENTINEL SET", 2837 option); 2838 if (changes) sentinelFlushConfig(); 2839 return; 2840 } 2841 sentinelEvent(REDIS_WARNING,"+set",ri,"%@ %s %s",option,value); 2842 } 2843 2844 if (changes) sentinelFlushConfig(); 2845 addReply(c,shared.ok); 2846 return; 2847 2848 badfmt: /* Bad format errors */ 2849 if (changes) sentinelFlushConfig(); 2850 addReplyErrorFormat(c,"Invalid argument '%s' for SENTINEL SET '%s'", 2851 value, option); 2852 } 2853 2854 /* Our fake PUBLISH command: it is actually useful only to receive hello messages 2855 * from the other sentinel instances, and publishing to a channel other than 2856 * SENTINEL_HELLO_CHANNEL is forbidden. 2857 * 2858 * Because we have a Sentinel PUBLISH, the code to send hello messages is the same 2859 * for all the three kind of instances: masters, slaves, sentinels. */ 2860 void sentinelPublishCommand(redisClient *c) { 2861 if (strcmp(c->argv[1]->ptr,SENTINEL_HELLO_CHANNEL)) { 2862 addReplyError(c, "Only HELLO messages are accepted by Sentinel instances."); 2863 return; 2864 } 2865 sentinelProcessHelloMessage(c->argv[2]->ptr,sdslen(c->argv[2]->ptr)); 2866 addReplyLongLong(c,1); 2867 } 2868 2869 /* ===================== SENTINEL availability checks ======================= */ 2870 2871 /* Is this instance down from our point of view? */ 2872 void sentinelCheckSubjectivelyDown(sentinelRedisInstance *ri) { 2873 mstime_t elapsed = 0; 2874 2875 if (ri->last_ping_time) 2876 elapsed = mstime() - ri->last_ping_time; 2877 2878 /* Check if we are in need for a reconnection of one of the 2879 * links, because we are detecting low activity. 2880 * 2881 * 1) Check if the command link seems connected, was connected not less 2882 * than SENTINEL_MIN_LINK_RECONNECT_PERIOD, but still we have a 2883 * pending ping for more than half the timeout. */ 2884 if (ri->cc && 2885 (mstime() - ri->cc_conn_time) > SENTINEL_MIN_LINK_RECONNECT_PERIOD && 2886 ri->last_ping_time != 0 && /* Ther is a pending ping... */ 2887 /* The pending ping is delayed, and we did not received 2888 * error replies as well. */ 2889 (mstime() - ri->last_ping_time) > (ri->down_after_period/2) && 2890 (mstime() - ri->last_pong_time) > (ri->down_after_period/2)) 2891 { 2892 sentinelKillLink(ri,ri->cc); 2893 } 2894 2895 /* 2) Check if the pubsub link seems connected, was connected not less 2896 * than SENTINEL_MIN_LINK_RECONNECT_PERIOD, but still we have no 2897 * activity in the Pub/Sub channel for more than 2898 * SENTINEL_PUBLISH_PERIOD * 3. 2899 */ 2900 if (ri->pc && 2901 (mstime() - ri->pc_conn_time) > SENTINEL_MIN_LINK_RECONNECT_PERIOD && 2902 (mstime() - ri->pc_last_activity) > (SENTINEL_PUBLISH_PERIOD*3)) 2903 { 2904 sentinelKillLink(ri,ri->pc); 2905 } 2906 2907 /* Update the SDOWN flag. We believe the instance is SDOWN if: 2908 * 2909 * 1) It is not replying. 2910 * 2) We believe it is a master, it reports to be a slave for enough time 2911 * to meet the down_after_period, plus enough time to get two times 2912 * INFO report from the instance. */ 2913 if (elapsed > ri->down_after_period || 2914 (ri->flags & SRI_MASTER && 2915 ri->role_reported == SRI_SLAVE && 2916 mstime() - ri->role_reported_time > 2917 (ri->down_after_period+SENTINEL_INFO_PERIOD*2))) 2918 { 2919 /* Is subjectively down */ 2920 if ((ri->flags & SRI_S_DOWN) == 0) { 2921 sentinelEvent(REDIS_WARNING,"+sdown",ri,"%@"); 2922 ri->s_down_since_time = mstime(); 2923 ri->flags |= SRI_S_DOWN; 2924 } 2925 } else { 2926 /* Is subjectively up */ 2927 if (ri->flags & SRI_S_DOWN) { 2928 sentinelEvent(REDIS_WARNING,"-sdown",ri,"%@"); 2929 ri->flags &= ~(SRI_S_DOWN|SRI_SCRIPT_KILL_SENT); 2930 } 2931 } 2932 } 2933 2934 /* Is this instance down according to the configured quorum? 2935 * 2936 * Note that ODOWN is a weak quorum, it only means that enough Sentinels 2937 * reported in a given time range that the instance was not reachable. 2938 * However messages can be delayed so there are no strong guarantees about 2939 * N instances agreeing at the same time about the down state. */ 2940 void sentinelCheckObjectivelyDown(sentinelRedisInstance *master) { 2941 dictIterator *di; 2942 dictEntry *de; 2943 int quorum = 0, odown = 0; 2944 2945 if (master->flags & SRI_S_DOWN) { 2946 /* Is down for enough sentinels? */ 2947 quorum = 1; /* the current sentinel. */ 2948 /* Count all the other sentinels. */ 2949 di = dictGetIterator(master->sentinels); 2950 while((de = dictNext(di)) != NULL) { 2951 sentinelRedisInstance *ri = dictGetVal(de); 2952 2953 if (ri->flags & SRI_MASTER_DOWN) quorum++; 2954 } 2955 dictReleaseIterator(di); 2956 if (quorum >= master->quorum) odown = 1; 2957 } 2958 2959 /* Set the flag accordingly to the outcome. */ 2960 if (odown) { 2961 if ((master->flags & SRI_O_DOWN) == 0) { 2962 sentinelEvent(REDIS_WARNING,"+odown",master,"%@ #quorum %d/%d", 2963 quorum, master->quorum); 2964 master->flags |= SRI_O_DOWN; 2965 master->o_down_since_time = mstime(); 2966 } 2967 } else { 2968 if (master->flags & SRI_O_DOWN) { 2969 sentinelEvent(REDIS_WARNING,"-odown",master,"%@"); 2970 master->flags &= ~SRI_O_DOWN; 2971 } 2972 } 2973 } 2974 2975 /* Receive the SENTINEL is-master-down-by-addr reply, see the 2976 * sentinelAskMasterStateToOtherSentinels() function for more information. */ 2977 void sentinelReceiveIsMasterDownReply(redisAsyncContext *c, void *reply, void *privdata) { 2978 sentinelRedisInstance *ri = c->data; 2979 redisReply *r; 2980 2981 if (ri) ri->pending_commands--; 2982 if (!reply || !ri) return; 2983 r = reply; 2984 2985 /* Ignore every error or unexpected reply. 2986 * Note that if the command returns an error for any reason we'll 2987 * end clearing the SRI_MASTER_DOWN flag for timeout anyway. */ 2988 if (r->type == REDIS_REPLY_ARRAY && r->elements == 3 && 2989 r->element[0]->type == REDIS_REPLY_INTEGER && 2990 r->element[1]->type == REDIS_REPLY_STRING && 2991 r->element[2]->type == REDIS_REPLY_INTEGER) 2992 { 2993 ri->last_master_down_reply_time = mstime(); 2994 if (r->element[0]->integer == 1) { 2995 ri->flags |= SRI_MASTER_DOWN; 2996 } else { 2997 ri->flags &= ~SRI_MASTER_DOWN; 2998 } 2999 if (strcmp(r->element[1]->str,"*")) { 3000 /* If the runid in the reply is not "*" the Sentinel actually 3001 * replied with a vote. */ 3002 sdsfree(ri->leader); 3003 if (ri->leader_epoch != r->element[2]->integer) 3004 redisLog(REDIS_WARNING, 3005 "%s voted for %s %llu", ri->name, 3006 r->element[1]->str, 3007 (unsigned long long) r->element[2]->integer); 3008 ri->leader = sdsnew(r->element[1]->str); 3009 ri->leader_epoch = r->element[2]->integer; 3010 } 3011 } 3012 } 3013 3014 /* If we think the master is down, we start sending 3015 * SENTINEL IS-MASTER-DOWN-BY-ADDR requests to other sentinels 3016 * in order to get the replies that allow to reach the quorum 3017 * needed to mark the master in ODOWN state and trigger a failover. */ 3018 #define SENTINEL_ASK_FORCED (1<<0) 3019 void sentinelAskMasterStateToOtherSentinels(sentinelRedisInstance *master, int flags) { 3020 dictIterator *di; 3021 dictEntry *de; 3022 3023 di = dictGetIterator(master->sentinels); 3024 while((de = dictNext(di)) != NULL) { 3025 sentinelRedisInstance *ri = dictGetVal(de); 3026 mstime_t elapsed = mstime() - ri->last_master_down_reply_time; 3027 char port[32]; 3028 int retval; 3029 3030 /* If the master state from other sentinel is too old, we clear it. */ 3031 if (elapsed > SENTINEL_ASK_PERIOD*5) { 3032 ri->flags &= ~SRI_MASTER_DOWN; 3033 sdsfree(ri->leader); 3034 ri->leader = NULL; 3035 } 3036 3037 /* Only ask if master is down to other sentinels if: 3038 * 3039 * 1) We believe it is down, or there is a failover in progress. 3040 * 2) Sentinel is connected. 3041 * 3) We did not received the info within SENTINEL_ASK_PERIOD ms. */ 3042 if ((master->flags & SRI_S_DOWN) == 0) continue; 3043 if (ri->flags & SRI_DISCONNECTED) continue; 3044 if (!(flags & SENTINEL_ASK_FORCED) && 3045 mstime() - ri->last_master_down_reply_time < SENTINEL_ASK_PERIOD) 3046 continue; 3047 3048 /* Ask */ 3049 ll2string(port,sizeof(port),master->addr->port); 3050 retval = redisAsyncCommand(ri->cc, 3051 sentinelReceiveIsMasterDownReply, NULL, 3052 "SENTINEL is-master-down-by-addr %s %s %llu %s", 3053 master->addr->ip, port, 3054 sentinel.current_epoch, 3055 (master->failover_state > SENTINEL_FAILOVER_STATE_NONE) ? 3056 server.runid : "*"); 3057 if (retval == REDIS_OK) ri->pending_commands++; 3058 } 3059 dictReleaseIterator(di); 3060 } 3061 3062 /* =============================== FAILOVER ================================= */ 3063 3064 /* Vote for the sentinel with 'req_runid' or return the old vote if already 3065 * voted for the specifed 'req_epoch' or one greater. 3066 * 3067 * If a vote is not available returns NULL, otherwise return the Sentinel 3068 * runid and populate the leader_epoch with the epoch of the vote. */ 3069 char *sentinelVoteLeader(sentinelRedisInstance *master, uint64_t req_epoch, char *req_runid, uint64_t *leader_epoch) { 3070 if (req_epoch > sentinel.current_epoch) { 3071 sentinel.current_epoch = req_epoch; 3072 sentinelFlushConfig(); 3073 sentinelEvent(REDIS_WARNING,"+new-epoch",master,"%llu", 3074 (unsigned long long) sentinel.current_epoch); 3075 } 3076 3077 if (master->leader_epoch < req_epoch && sentinel.current_epoch <= req_epoch) 3078 { 3079 sdsfree(master->leader); 3080 master->leader = sdsnew(req_runid); 3081 master->leader_epoch = sentinel.current_epoch; 3082 sentinelFlushConfig(); 3083 sentinelEvent(REDIS_WARNING,"+vote-for-leader",master,"%s %llu", 3084 master->leader, (unsigned long long) master->leader_epoch); 3085 /* If we did not voted for ourselves, set the master failover start 3086 * time to now, in order to force a delay before we can start a 3087 * failover for the same master. */ 3088 if (strcasecmp(master->leader,server.runid)) 3089 master->failover_start_time = mstime()+rand()%SENTINEL_MAX_DESYNC; 3090 } 3091 3092 *leader_epoch = master->leader_epoch; 3093 return master->leader ? sdsnew(master->leader) : NULL; 3094 } 3095 3096 struct sentinelLeader { 3097 char *runid; 3098 unsigned long votes; 3099 }; 3100 3101 /* Helper function for sentinelGetLeader, increment the counter 3102 * relative to the specified runid. */ 3103 int sentinelLeaderIncr(dict *counters, char *runid) { 3104 dictEntry *de = dictFind(counters,runid); 3105 uint64_t oldval; 3106 3107 if (de) { 3108 oldval = dictGetUnsignedIntegerVal(de); 3109 dictSetUnsignedIntegerVal(de,oldval+1); 3110 return oldval+1; 3111 } else { 3112 de = dictAddRaw(counters,runid); 3113 redisAssert(de != NULL); 3114 dictSetUnsignedIntegerVal(de,1); 3115 return 1; 3116 } 3117 } 3118 3119 /* Scan all the Sentinels attached to this master to check if there 3120 * is a leader for the specified epoch. 3121 * 3122 * To be a leader for a given epoch, we should have the majorify of 3123 * the Sentinels we know that reported the same instance as 3124 * leader for the same epoch. */ 3125 char *sentinelGetLeader(sentinelRedisInstance *master, uint64_t epoch) { 3126 dict *counters; 3127 dictIterator *di; 3128 dictEntry *de; 3129 unsigned int voters = 0, voters_quorum; 3130 char *myvote; 3131 char *winner = NULL; 3132 uint64_t leader_epoch; 3133 uint64_t max_votes = 0; 3134 3135 redisAssert(master->flags & (SRI_O_DOWN|SRI_FAILOVER_IN_PROGRESS)); 3136 counters = dictCreate(&leaderVotesDictType,NULL); 3137 3138 /* Count other sentinels votes */ 3139 di = dictGetIterator(master->sentinels); 3140 while((de = dictNext(di)) != NULL) { 3141 sentinelRedisInstance *ri = dictGetVal(de); 3142 if (ri->leader != NULL && ri->leader_epoch == sentinel.current_epoch) 3143 sentinelLeaderIncr(counters,ri->leader); 3144 voters++; 3145 } 3146 dictReleaseIterator(di); 3147 3148 /* Check what's the winner. For the winner to win, it needs two conditions: 3149 * 1) Absolute majority between voters (50% + 1). 3150 * 2) And anyway at least master->quorum votes. */ 3151 di = dictGetIterator(counters); 3152 while((de = dictNext(di)) != NULL) { 3153 uint64_t votes = dictGetUnsignedIntegerVal(de); 3154 3155 if (votes > max_votes) { 3156 max_votes = votes; 3157 winner = dictGetKey(de); 3158 } 3159 } 3160 dictReleaseIterator(di); 3161 3162 /* Count this Sentinel vote: 3163 * if this Sentinel did not voted yet, either vote for the most 3164 * common voted sentinel, or for itself if no vote exists at all. */ 3165 if (winner) 3166 myvote = sentinelVoteLeader(master,epoch,winner,&leader_epoch); 3167 else 3168 myvote = sentinelVoteLeader(master,epoch,server.runid,&leader_epoch); 3169 3170 if (myvote && leader_epoch == epoch) { 3171 uint64_t votes = sentinelLeaderIncr(counters,myvote); 3172 3173 if (votes > max_votes) { 3174 max_votes = votes; 3175 winner = myvote; 3176 } 3177 } 3178 voters++; /* Anyway, count me as one of the voters. */ 3179 3180 voters_quorum = voters/2+1; 3181 if (winner && (max_votes < voters_quorum || max_votes < master->quorum)) 3182 winner = NULL; 3183 3184 winner = winner ? sdsnew(winner) : NULL; 3185 sdsfree(myvote); 3186 dictRelease(counters); 3187 return winner; 3188 } 3189 3190 /* Send SLAVEOF to the specified instance, always followed by a 3191 * CONFIG REWRITE command in order to store the new configuration on disk 3192 * when possible (that is, if the Redis instance is recent enough to support 3193 * config rewriting, and if the server was started with a configuration file). 3194 * 3195 * If Host is NULL the function sends "SLAVEOF NO ONE". 3196 * 3197 * The command returns REDIS_OK if the SLAVEOF command was accepted for 3198 * (later) delivery otherwise REDIS_ERR. The command replies are just 3199 * discarded. */ 3200 int sentinelSendSlaveOf(sentinelRedisInstance *ri, char *host, int port) { 3201 char portstr[32]; 3202 int retval; 3203 3204 ll2string(portstr,sizeof(portstr),port); 3205 3206 /* If host is NULL we send SLAVEOF NO ONE that will turn the instance 3207 * into a master. */ 3208 if (host == NULL) { 3209 host = "NO"; 3210 memcpy(portstr,"ONE",4); 3211 } 3212 3213 /* In order to send SLAVEOF in a safe way, we send a transaction performing 3214 * the following tasks: 3215 * 1) Reconfigure the instance according to the specified host/port params. 3216 * 2) Rewrite the configuraiton. 3217 * 3) Disconnect all clients (but this one sending the commnad) in order 3218 * to trigger the ask-master-on-reconnection protocol for connected 3219 * clients. 3220 * 3221 * Note that we don't check the replies returned by commands, since we 3222 * will observe instead the effects in the next INFO output. */ 3223 retval = redisAsyncCommand(ri->cc, 3224 sentinelDiscardReplyCallback, NULL, "MULTI"); 3225 if (retval == REDIS_ERR) return retval; 3226 ri->pending_commands++; 3227 3228 retval = redisAsyncCommand(ri->cc, 3229 sentinelDiscardReplyCallback, NULL, "SLAVEOF %s %s", host, portstr); 3230 if (retval == REDIS_ERR) return retval; 3231 ri->pending_commands++; 3232 3233 retval = redisAsyncCommand(ri->cc, 3234 sentinelDiscardReplyCallback, NULL, "CONFIG REWRITE"); 3235 if (retval == REDIS_ERR) return retval; 3236 ri->pending_commands++; 3237 3238 /* CLIENT KILL TYPE <type> is only supported starting from Redis 2.8.12, 3239 * however sending it to an instance not understanding this command is not 3240 * an issue because CLIENT is variadic command, so Redis will not 3241 * recognized as a syntax error, and the transaction will not fail (but 3242 * only the unsupported command will fail). */ 3243 retval = redisAsyncCommand(ri->cc, 3244 sentinelDiscardReplyCallback, NULL, "CLIENT KILL TYPE normal"); 3245 if (retval == REDIS_ERR) return retval; 3246 ri->pending_commands++; 3247 3248 retval = redisAsyncCommand(ri->cc, 3249 sentinelDiscardReplyCallback, NULL, "EXEC"); 3250 if (retval == REDIS_ERR) return retval; 3251 ri->pending_commands++; 3252 3253 return REDIS_OK; 3254 } 3255 3256 /* Setup the master state to start a failover. */ 3257 void sentinelStartFailover(sentinelRedisInstance *master) { 3258 redisAssert(master->flags & SRI_MASTER); 3259 3260 master->failover_state = SENTINEL_FAILOVER_STATE_WAIT_START; 3261 master->flags |= SRI_FAILOVER_IN_PROGRESS; 3262 master->failover_epoch = ++sentinel.current_epoch; 3263 sentinelEvent(REDIS_WARNING,"+new-epoch",master,"%llu", 3264 (unsigned long long) sentinel.current_epoch); 3265 sentinelEvent(REDIS_WARNING,"+try-failover",master,"%@"); 3266 master->failover_start_time = mstime()+rand()%SENTINEL_MAX_DESYNC; 3267 master->failover_state_change_time = mstime(); 3268 } 3269 3270 /* This function checks if there are the conditions to start the failover, 3271 * that is: 3272 * 3273 * 1) Master must be in ODOWN condition. 3274 * 2) No failover already in progress. 3275 * 3) No failover already attempted recently. 3276 * 3277 * We still don't know if we'll win the election so it is possible that we 3278 * start the failover but that we'll not be able to act. 3279 * 3280 * Return non-zero if a failover was started. */ 3281 int sentinelStartFailoverIfNeeded(sentinelRedisInstance *master) { 3282 /* We can't failover if the master is not in O_DOWN state. */ 3283 if (!(master->flags & SRI_O_DOWN)) return 0; 3284 3285 /* Failover already in progress? */ 3286 if (master->flags & SRI_FAILOVER_IN_PROGRESS) return 0; 3287 3288 /* Last failover attempt started too little time ago? */ 3289 if (mstime() - master->failover_start_time < 3290 master->failover_timeout*2) 3291 { 3292 if (master->failover_delay_logged != master->failover_start_time) { 3293 time_t clock = (master->failover_start_time + 3294 master->failover_timeout*2) / 1000; 3295 char ctimebuf[26]; 3296 3297 ctime_r(&clock,ctimebuf); 3298 ctimebuf[24] = '\0'; /* Remove newline. */ 3299 master->failover_delay_logged = master->failover_start_time; 3300 redisLog(REDIS_WARNING, 3301 "Next failover delay: I will not start a failover before %s", 3302 ctimebuf); 3303 } 3304 return 0; 3305 } 3306 3307 sentinelStartFailover(master); 3308 return 1; 3309 } 3310 3311 /* Select a suitable slave to promote. The current algorithm only uses 3312 * the following parameters: 3313 * 3314 * 1) None of the following conditions: S_DOWN, O_DOWN, DISCONNECTED. 3315 * 2) Last time the slave replied to ping no more than 5 times the PING period. 3316 * 3) info_refresh not older than 3 times the INFO refresh period. 3317 * 4) master_link_down_time no more than: 3318 * (now - master->s_down_since_time) + (master->down_after_period * 10). 3319 * Basically since the master is down from our POV, the slave reports 3320 * to be disconnected no more than 10 times the configured down-after-period. 3321 * This is pretty much black magic but the idea is, the master was not 3322 * available so the slave may be lagging, but not over a certain time. 3323 * Anyway we'll select the best slave according to replication offset. 3324 * 5) Slave priority can't be zero, otherwise the slave is discarded. 3325 * 3326 * Among all the slaves matching the above conditions we select the slave 3327 * with, in order of sorting key: 3328 * 3329 * - lower slave_priority. 3330 * - bigger processed replication offset. 3331 * - lexicographically smaller runid. 3332 * 3333 * Basically if runid is the same, the slave that processed more commands 3334 * from the master is selected. 3335 * 3336 * The function returns the pointer to the selected slave, otherwise 3337 * NULL if no suitable slave was found. 3338 */ 3339 3340 /* Helper for sentinelSelectSlave(). This is used by qsort() in order to 3341 * sort suitable slaves in a "better first" order, to take the first of 3342 * the list. */ 3343 int compareSlavesForPromotion(const void *a, const void *b) { 3344 sentinelRedisInstance **sa = (sentinelRedisInstance **)a, 3345 **sb = (sentinelRedisInstance **)b; 3346 char *sa_runid, *sb_runid; 3347 3348 if ((*sa)->slave_priority != (*sb)->slave_priority) 3349 return (*sa)->slave_priority - (*sb)->slave_priority; 3350 3351 /* If priority is the same, select the slave with greater replication 3352 * offset (processed more data frmo the master). */ 3353 if ((*sa)->slave_repl_offset > (*sb)->slave_repl_offset) { 3354 return -1; /* a < b */ 3355 } else if ((*sa)->slave_repl_offset < (*sb)->slave_repl_offset) { 3356 return 1; /* b > a */ 3357 } 3358 3359 /* If the replication offset is the same select the slave with that has 3360 * the lexicographically smaller runid. Note that we try to handle runid 3361 * == NULL as there are old Redis versions that don't publish runid in 3362 * INFO. A NULL runid is considered bigger than any other runid. */ 3363 sa_runid = (*sa)->runid; 3364 sb_runid = (*sb)->runid; 3365 if (sa_runid == NULL && sb_runid == NULL) return 0; 3366 else if (sa_runid == NULL) return 1; /* a > b */ 3367 else if (sb_runid == NULL) return -1; /* a < b */ 3368 return strcasecmp(sa_runid, sb_runid); 3369 } 3370 3371 sentinelRedisInstance *sentinelSelectSlave(sentinelRedisInstance *master) { 3372 sentinelRedisInstance **instance = 3373 zmalloc(sizeof(instance[0])*dictSize(master->slaves)); 3374 sentinelRedisInstance *selected = NULL; 3375 int instances = 0; 3376 dictIterator *di; 3377 dictEntry *de; 3378 mstime_t max_master_down_time = 0; 3379 3380 if (master->flags & SRI_S_DOWN) 3381 max_master_down_time += mstime() - master->s_down_since_time; 3382 max_master_down_time += master->down_after_period * 10; 3383 3384 di = dictGetIterator(master->slaves); 3385 while((de = dictNext(di)) != NULL) { 3386 sentinelRedisInstance *slave = dictGetVal(de); 3387 mstime_t info_validity_time; 3388 3389 if (slave->flags & (SRI_S_DOWN|SRI_O_DOWN|SRI_DISCONNECTED)) continue; 3390 if (mstime() - slave->last_avail_time > SENTINEL_PING_PERIOD*5) continue; 3391 if (slave->slave_priority == 0) continue; 3392 3393 /* If the master is in SDOWN state we get INFO for slaves every second. 3394 * Otherwise we get it with the usual period so we need to account for 3395 * a larger delay. */ 3396 if (master->flags & SRI_S_DOWN) 3397 info_validity_time = SENTINEL_PING_PERIOD*5; 3398 else 3399 info_validity_time = SENTINEL_INFO_PERIOD*3; 3400 if (mstime() - slave->info_refresh > info_validity_time) continue; 3401 if (slave->master_link_down_time > max_master_down_time) continue; 3402 instance[instances++] = slave; 3403 } 3404 dictReleaseIterator(di); 3405 if (instances) { 3406 qsort(instance,instances,sizeof(sentinelRedisInstance*), 3407 compareSlavesForPromotion); 3408 selected = instance[0]; 3409 } 3410 zfree(instance); 3411 return selected; 3412 } 3413 3414 /* ---------------- Failover state machine implementation ------------------- */ 3415 void sentinelFailoverWaitStart(sentinelRedisInstance *ri) { 3416 char *leader; 3417 int isleader; 3418 3419 /* Check if we are the leader for the failover epoch. */ 3420 leader = sentinelGetLeader(ri, ri->failover_epoch); 3421 isleader = leader && strcasecmp(leader,server.runid) == 0; 3422 sdsfree(leader); 3423 3424 /* If I'm not the leader, and it is not a forced failover via 3425 * SENTINEL FAILOVER, then I can't continue with the failover. */ 3426 if (!isleader && !(ri->flags & SRI_FORCE_FAILOVER)) { 3427 int election_timeout = SENTINEL_ELECTION_TIMEOUT; 3428 3429 /* The election timeout is the MIN between SENTINEL_ELECTION_TIMEOUT 3430 * and the configured failover timeout. */ 3431 if (election_timeout > ri->failover_timeout) 3432 election_timeout = ri->failover_timeout; 3433 /* Abort the failover if I'm not the leader after some time. */ 3434 if (mstime() - ri->failover_start_time > election_timeout) { 3435 sentinelEvent(REDIS_WARNING,"-failover-abort-not-elected",ri,"%@"); 3436 sentinelAbortFailover(ri); 3437 } 3438 return; 3439 } 3440 sentinelEvent(REDIS_WARNING,"+elected-leader",ri,"%@"); 3441 ri->failover_state = SENTINEL_FAILOVER_STATE_SELECT_SLAVE; 3442 ri->failover_state_change_time = mstime(); 3443 sentinelEvent(REDIS_WARNING,"+failover-state-select-slave",ri,"%@"); 3444 } 3445 3446 void sentinelFailoverSelectSlave(sentinelRedisInstance *ri) { 3447 sentinelRedisInstance *slave = sentinelSelectSlave(ri); 3448 3449 /* We don't handle the timeout in this state as the function aborts 3450 * the failover or go forward in the next state. */ 3451 if (slave == NULL) { 3452 sentinelEvent(REDIS_WARNING,"-failover-abort-no-good-slave",ri,"%@"); 3453 sentinelAbortFailover(ri); 3454 } else { 3455 sentinelEvent(REDIS_WARNING,"+selected-slave",slave,"%@"); 3456 slave->flags |= SRI_PROMOTED; 3457 ri->promoted_slave = slave; 3458 ri->failover_state = SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE; 3459 ri->failover_state_change_time = mstime(); 3460 sentinelEvent(REDIS_NOTICE,"+failover-state-send-slaveof-noone", 3461 slave, "%@"); 3462 } 3463 } 3464 3465 void sentinelFailoverSendSlaveOfNoOne(sentinelRedisInstance *ri) { 3466 int retval; 3467 3468 /* We can't send the command to the promoted slave if it is now 3469 * disconnected. Retry again and again with this state until the timeout 3470 * is reached, then abort the failover. */ 3471 if (ri->promoted_slave->flags & SRI_DISCONNECTED) { 3472 if (mstime() - ri->failover_state_change_time > ri->failover_timeout) { 3473 sentinelEvent(REDIS_WARNING,"-failover-abort-slave-timeout",ri,"%@"); 3474 sentinelAbortFailover(ri); 3475 } 3476 return; 3477 } 3478 3479 /* Send SLAVEOF NO ONE command to turn the slave into a master. 3480 * We actually register a generic callback for this command as we don't 3481 * really care about the reply. We check if it worked indirectly observing 3482 * if INFO returns a different role (master instead of slave). */ 3483 retval = sentinelSendSlaveOf(ri->promoted_slave,NULL,0); 3484 if (retval != REDIS_OK) return; 3485 sentinelEvent(REDIS_NOTICE, "+failover-state-wait-promotion", 3486 ri->promoted_slave,"%@"); 3487 ri->failover_state = SENTINEL_FAILOVER_STATE_WAIT_PROMOTION; 3488 ri->failover_state_change_time = mstime(); 3489 } 3490 3491 /* We actually wait for promotion indirectly checking with INFO when the 3492 * slave turns into a master. */ 3493 void sentinelFailoverWaitPromotion(sentinelRedisInstance *ri) { 3494 /* Just handle the timeout. Switching to the next state is handled 3495 * by the function parsing the INFO command of the promoted slave. */ 3496 if (mstime() - ri->failover_state_change_time > ri->failover_timeout) { 3497 sentinelEvent(REDIS_WARNING,"-failover-abort-slave-timeout",ri,"%@"); 3498 sentinelAbortFailover(ri); 3499 } 3500 } 3501 3502 void sentinelFailoverDetectEnd(sentinelRedisInstance *master) { 3503 int not_reconfigured = 0, timeout = 0; 3504 dictIterator *di; 3505 dictEntry *de; 3506 mstime_t elapsed = mstime() - master->failover_state_change_time; 3507 3508 /* We can't consider failover finished if the promoted slave is 3509 * not reachable. */ 3510 if (master->promoted_slave == NULL || 3511 master->promoted_slave->flags & SRI_S_DOWN) return; 3512 3513 /* The failover terminates once all the reachable slaves are properly 3514 * configured. */ 3515 di = dictGetIterator(master->slaves); 3516 while((de = dictNext(di)) != NULL) { 3517 sentinelRedisInstance *slave = dictGetVal(de); 3518 3519 if (slave->flags & (SRI_PROMOTED|SRI_RECONF_DONE)) continue; 3520 if (slave->flags & SRI_S_DOWN) continue; 3521 not_reconfigured++; 3522 } 3523 dictReleaseIterator(di); 3524 3525 /* Force end of failover on timeout. */ 3526 if (elapsed > master->failover_timeout) { 3527 not_reconfigured = 0; 3528 timeout = 1; 3529 sentinelEvent(REDIS_WARNING,"+failover-end-for-timeout",master,"%@"); 3530 } 3531 3532 if (not_reconfigured == 0) { 3533 sentinelEvent(REDIS_WARNING,"+failover-end",master,"%@"); 3534 master->failover_state = SENTINEL_FAILOVER_STATE_UPDATE_CONFIG; 3535 master->failover_state_change_time = mstime(); 3536 } 3537 3538 /* If I'm the leader it is a good idea to send a best effort SLAVEOF 3539 * command to all the slaves still not reconfigured to replicate with 3540 * the new master. */ 3541 if (timeout) { 3542 dictIterator *di; 3543 dictEntry *de; 3544 3545 di = dictGetIterator(master->slaves); 3546 while((de = dictNext(di)) != NULL) { 3547 sentinelRedisInstance *slave = dictGetVal(de); 3548 int retval; 3549 3550 if (slave->flags & 3551 (SRI_RECONF_DONE|SRI_RECONF_SENT|SRI_DISCONNECTED)) continue; 3552 3553 retval = sentinelSendSlaveOf(slave, 3554 master->promoted_slave->addr->ip, 3555 master->promoted_slave->addr->port); 3556 if (retval == REDIS_OK) { 3557 sentinelEvent(REDIS_NOTICE,"+slave-reconf-sent-be",slave,"%@"); 3558 slave->flags |= SRI_RECONF_SENT; 3559 } 3560 } 3561 dictReleaseIterator(di); 3562 } 3563 } 3564 3565 /* Send SLAVE OF <new master address> to all the remaining slaves that 3566 * still don't appear to have the configuration updated. */ 3567 void sentinelFailoverReconfNextSlave(sentinelRedisInstance *master) { 3568 dictIterator *di; 3569 dictEntry *de; 3570 int in_progress = 0; 3571 3572 di = dictGetIterator(master->slaves); 3573 while((de = dictNext(di)) != NULL) { 3574 sentinelRedisInstance *slave = dictGetVal(de); 3575 3576 if (slave->flags & (SRI_RECONF_SENT|SRI_RECONF_INPROG)) 3577 in_progress++; 3578 } 3579 dictReleaseIterator(di); 3580 3581 di = dictGetIterator(master->slaves); 3582 while(in_progress < master->parallel_syncs && 3583 (de = dictNext(di)) != NULL) 3584 { 3585 sentinelRedisInstance *slave = dictGetVal(de); 3586 int retval; 3587 3588 /* Skip the promoted slave, and already configured slaves. */ 3589 if (slave->flags & (SRI_PROMOTED|SRI_RECONF_DONE)) continue; 3590 3591 /* If too much time elapsed without the slave moving forward to 3592 * the next state, consider it reconfigured even if it is not. 3593 * Sentinels will detect the slave as misconfigured and fix its 3594 * configuration later. */ 3595 if ((slave->flags & SRI_RECONF_SENT) && 3596 (mstime() - slave->slave_reconf_sent_time) > 3597 SENTINEL_SLAVE_RECONF_TIMEOUT) 3598 { 3599 sentinelEvent(REDIS_NOTICE,"-slave-reconf-sent-timeout",slave,"%@"); 3600 slave->flags &= ~SRI_RECONF_SENT; 3601 slave->flags |= SRI_RECONF_DONE; 3602 } 3603 3604 /* Nothing to do for instances that are disconnected or already 3605 * in RECONF_SENT state. */ 3606 if (slave->flags & (SRI_DISCONNECTED|SRI_RECONF_SENT|SRI_RECONF_INPROG)) 3607 continue; 3608 3609 /* Send SLAVEOF <new master>. */ 3610 retval = sentinelSendSlaveOf(slave, 3611 master->promoted_slave->addr->ip, 3612 master->promoted_slave->addr->port); 3613 if (retval == REDIS_OK) { 3614 slave->flags |= SRI_RECONF_SENT; 3615 slave->slave_reconf_sent_time = mstime(); 3616 sentinelEvent(REDIS_NOTICE,"+slave-reconf-sent",slave,"%@"); 3617 in_progress++; 3618 } 3619 } 3620 dictReleaseIterator(di); 3621 3622 /* Check if all the slaves are reconfigured and handle timeout. */ 3623 sentinelFailoverDetectEnd(master); 3624 } 3625 3626 /* This function is called when the slave is in 3627 * SENTINEL_FAILOVER_STATE_UPDATE_CONFIG state. In this state we need 3628 * to remove it from the master table and add the promoted slave instead. */ 3629 void sentinelFailoverSwitchToPromotedSlave(sentinelRedisInstance *master) { 3630 sentinelRedisInstance *ref = master->promoted_slave ? 3631 master->promoted_slave : master; 3632 3633 sentinelEvent(REDIS_WARNING,"+switch-master",master,"%s %s %d %s %d", 3634 master->name, master->addr->ip, master->addr->port, 3635 ref->addr->ip, ref->addr->port); 3636 3637 sentinelResetMasterAndChangeAddress(master,ref->addr->ip,ref->addr->port); 3638 } 3639 3640 void sentinelFailoverStateMachine(sentinelRedisInstance *ri) { 3641 redisAssert(ri->flags & SRI_MASTER); 3642 3643 if (!(ri->flags & SRI_FAILOVER_IN_PROGRESS)) return; 3644 3645 switch(ri->failover_state) { 3646 case SENTINEL_FAILOVER_STATE_WAIT_START: 3647 sentinelFailoverWaitStart(ri); 3648 break; 3649 case SENTINEL_FAILOVER_STATE_SELECT_SLAVE: 3650 sentinelFailoverSelectSlave(ri); 3651 break; 3652 case SENTINEL_FAILOVER_STATE_SEND_SLAVEOF_NOONE: 3653 sentinelFailoverSendSlaveOfNoOne(ri); 3654 break; 3655 case SENTINEL_FAILOVER_STATE_WAIT_PROMOTION: 3656 sentinelFailoverWaitPromotion(ri); 3657 break; 3658 case SENTINEL_FAILOVER_STATE_RECONF_SLAVES: 3659 sentinelFailoverReconfNextSlave(ri); 3660 break; 3661 } 3662 } 3663 3664 /* Abort a failover in progress: 3665 * 3666 * This function can only be called before the promoted slave acknowledged 3667 * the slave -> master switch. Otherwise the failover can't be aborted and 3668 * will reach its end (possibly by timeout). */ 3669 void sentinelAbortFailover(sentinelRedisInstance *ri) { 3670 redisAssert(ri->flags & SRI_FAILOVER_IN_PROGRESS); 3671 redisAssert(ri->failover_state <= SENTINEL_FAILOVER_STATE_WAIT_PROMOTION); 3672 3673 ri->flags &= ~(SRI_FAILOVER_IN_PROGRESS|SRI_FORCE_FAILOVER); 3674 ri->failover_state = SENTINEL_FAILOVER_STATE_NONE; 3675 ri->failover_state_change_time = mstime(); 3676 if (ri->promoted_slave) { 3677 ri->promoted_slave->flags &= ~SRI_PROMOTED; 3678 ri->promoted_slave = NULL; 3679 } 3680 } 3681 3682 /* ======================== SENTINEL timer handler ========================== 3683 * This is the "main" our Sentinel, being sentinel completely non blocking 3684 * in design. The function is called every second. 3685 * -------------------------------------------------------------------------- */ 3686 3687 /* Perform scheduled operations for the specified Redis instance. */ 3688 void sentinelHandleRedisInstance(sentinelRedisInstance *ri) { 3689 /* ========== MONITORING HALF ============ */ 3690 /* Every kind of instance */ 3691 sentinelReconnectInstance(ri); 3692 sentinelSendPeriodicCommands(ri); 3693 3694 /* ============== ACTING HALF ============= */ 3695 /* We don't proceed with the acting half if we are in TILT mode. 3696 * TILT happens when we find something odd with the time, like a 3697 * sudden change in the clock. */ 3698 if (sentinel.tilt) { 3699 if (mstime()-sentinel.tilt_start_time < SENTINEL_TILT_PERIOD) return; 3700 sentinel.tilt = 0; 3701 sentinelEvent(REDIS_WARNING,"-tilt",NULL,"#tilt mode exited"); 3702 } 3703 3704 /* Every kind of instance */ 3705 sentinelCheckSubjectivelyDown(ri); 3706 3707 /* Masters and slaves */ 3708 if (ri->flags & (SRI_MASTER|SRI_SLAVE)) { 3709 /* Nothing so far. */ 3710 } 3711 3712 /* Only masters */ 3713 if (ri->flags & SRI_MASTER) { 3714 sentinelCheckObjectivelyDown(ri); 3715 if (sentinelStartFailoverIfNeeded(ri)) 3716 sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_ASK_FORCED); 3717 sentinelFailoverStateMachine(ri); 3718 sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_NO_FLAGS); 3719 } 3720 } 3721 3722 /* Perform scheduled operations for all the instances in the dictionary. 3723 * Recursively call the function against dictionaries of slaves. */ 3724 void sentinelHandleDictOfRedisInstances(dict *instances) { 3725 dictIterator *di; 3726 dictEntry *de; 3727 sentinelRedisInstance *switch_to_promoted = NULL; 3728 3729 /* There are a number of things we need to perform against every master. */ 3730 di = dictGetIterator(instances); 3731 while((de = dictNext(di)) != NULL) { 3732 sentinelRedisInstance *ri = dictGetVal(de); 3733 3734 sentinelHandleRedisInstance(ri); 3735 if (ri->flags & SRI_MASTER) { 3736 sentinelHandleDictOfRedisInstances(ri->slaves); 3737 sentinelHandleDictOfRedisInstances(ri->sentinels); 3738 if (ri->failover_state == SENTINEL_FAILOVER_STATE_UPDATE_CONFIG) { 3739 switch_to_promoted = ri; 3740 } 3741 } 3742 } 3743 if (switch_to_promoted) 3744 sentinelFailoverSwitchToPromotedSlave(switch_to_promoted); 3745 dictReleaseIterator(di); 3746 } 3747 3748 /* This function checks if we need to enter the TITL mode. 3749 * 3750 * The TILT mode is entered if we detect that between two invocations of the 3751 * timer interrupt, a negative amount of time, or too much time has passed. 3752 * Note that we expect that more or less just 100 milliseconds will pass 3753 * if everything is fine. However we'll see a negative number or a 3754 * difference bigger than SENTINEL_TILT_TRIGGER milliseconds if one of the 3755 * following conditions happen: 3756 * 3757 * 1) The Sentiel process for some time is blocked, for every kind of 3758 * random reason: the load is huge, the computer was frozen for some time 3759 * in I/O or alike, the process was stopped by a signal. Everything. 3760 * 2) The system clock was altered significantly. 3761 * 3762 * Under both this conditions we'll see everything as timed out and failing 3763 * without good reasons. Instead we enter the TILT mode and wait 3764 * for SENTINEL_TILT_PERIOD to elapse before starting to act again. 3765 * 3766 * During TILT time we still collect information, we just do not act. */ 3767 void sentinelCheckTiltCondition(void) { 3768 mstime_t now = mstime(); 3769 mstime_t delta = now - sentinel.previous_time; 3770 3771 if (delta < 0 || delta > SENTINEL_TILT_TRIGGER) { 3772 sentinel.tilt = 1; 3773 sentinel.tilt_start_time = mstime(); 3774 sentinelEvent(REDIS_WARNING,"+tilt",NULL,"#tilt mode entered"); 3775 } 3776 sentinel.previous_time = mstime(); 3777 } 3778 3779 void sentinelTimer(void) { 3780 sentinelCheckTiltCondition(); 3781 sentinelHandleDictOfRedisInstances(sentinel.masters); 3782 sentinelRunPendingScripts(); 3783 sentinelCollectTerminatedScripts(); 3784 sentinelKillTimedoutScripts(); 3785 3786 /* We continuously change the frequency of the Redis "timer interrupt" 3787 * in order to desynchronize every Sentinel from every other. 3788 * This non-determinism avoids that Sentinels started at the same time 3789 * exactly continue to stay synchronized asking to be voted at the 3790 * same time again and again (resulting in nobody likely winning the 3791 * election because of split brain voting). */ 3792 server.hz = REDIS_DEFAULT_HZ + rand() % REDIS_DEFAULT_HZ; 3793 } 3794 3795