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