Redis Cluster Gossip Protocol: PING, PONG, MEET

返回目录

PING / PONG / MEET 的发送

过程

计算freshNodes。freshNodes表示在消息中能携带的，在cluster节点字典中的节点总数，但需要减去myself和对端节点，因为myself的信息会存储在消息头中。实际上，并非所有在cluster节点字典中的节点都需要带出去，对于那些处于handshake，disconnected状态的节点是不用考虑的。
freshNodes = size(cluster nodes) - 2
计算wanted。wanted表示准备在消息中携带的gossip数。个数不能少于3，不能超过freshNodes。
wanted = floor(size(cluster nodes) / 10)
wanted = max(wanted, 3)
wanted = min(wanted, freshNodes)
获取pfailWanted。pfailWanted表示处于PFAIL状态的节点数。
如果当前link不是inbound，而且消息类型为PING，则更新对端实体的pingSent（发送ping的时间）为当前时间。
构建消息头部
计算maxIterations。maxIterations表示在遍历cluster节点字典时的最大遍历次数。
maxIterations = wanted * 3
编写gossip

gossipCount = 0
while freshNodes > 0  gossipCount < wanted && maxIterations:从cluster节点字典中随机抽取一个node /* 注意：是随机地抽取 */if node是myself || 处于PFAIL状态: /* myself会在消息头，PFAIL会在最后追加 */continueif (node处于handshake或者NOADDR(地址不可知)) ||(node还没有outbound link && node没有slots):freshNodes--	/* 节省CPU */continueif node已被添加到gossip:continue把node添加到gossipfreshNodes--gossipCount++

如果pfailCount > 0，则把PFAIL节点追加到gossip

while cluster节点字典还没遍历完 and pfailWanted > 0:获取一个nodeif (node处于handshake或者NOADDR) || node并非PFAIL:	/* 只追加PFAIL的节点 */continuegossipCount++pfailWanted--

如果配置了cluster-announce-hostname，则
- 给消息的mflags添加EXT_DATA标记，表示消息附带extension
- 把cluster-announce-hostname写入pingExtension
- 把pingExtension追加到消息末尾
计算消息的总长度
设置消息的count = gossipCount
设置消息的extensions = extension的个数
设置消息的totalLen = 上面计算出的总长度
发送消息
- PING和MEET通过outbound link发送
- PONG通过inbound link发送

gossip内容

每个要加入到gossip中的节点都会在其中生成一个对应条目，包含的信息：

节点ID
最近一次ping此节点的时间（如果已经收到此节点的PONG，会重置为0）
最近一次收到此节点PONG的时间
节点IP
节点的port，pport，cport
节点的flags

PING / PONG / MEET的接收处理

过程

第1 ~ 3步是涵盖所有类型的消息。

合法性检查
- 消息类型（type）的检查
- 消息长度检查。有的消息会携带gossip，有的会附带extension，所以消息的总长度是可变的，需要结合消息头部的元数据和消息的内容进行检查。
根据link和消息头，从cluster节点字典中查找实体sender

if link关联了node && node不是处于handshake:sender = node
else:sender = 根据消息头中的sender，从cluster节点字典中查找实体if sender存在 && link没有关联node:/* inbound link在创建时还不知道节点的真实ID，所以会找不到sender，因此要到达这里才能关联上sender */把link设置为sender的inbound link/* 关联后，下次就可以从link获取sender，而不需要再从cluster节点字典查找了 */把sender关联到link

通用处理，适用于所有消息类型

if sender存在:/* 更新数据接收时间，以免因为sender一直在发送数据而误认为它timeout */sender的dataReceived = 当前时间if sender没有处于handshake:/* 更新currentEpoch和configEpoch */if 消息头的currentEpoch > server的currentEpoch:server的currentEpoch = 消息头的currentEpochif 消息头的configEpoch > sender的configEpoch:sender的configEpoch = 消息头的configEpoch更新sender的复制偏移和相应时间if server正在做manual failover && myself是sender的slave &&消息头的mflags包含PAUSED && server的mf_master_offset == -1：更新server的mf_server_offset = sender的复制偏移

处理PING，MEET消息

if 消息类型是PING或者MEET:/* 获取myself的IP */if (消息类型是MEET || myself还没有IP) && 没有配置cluster-announce-ip:myself的IP = 从socket中获取自己的IP/* 在cluster节点字典中创建实体 */if sender不存在 && 消息类型为MEET:为sender创建node实体，生成随机ID/* 获取sender的IP和ports */if 消息头的myip不为全0:node的ip = myipelse:node的ip = 从socket获取对端的ip把消息头的port，pport, cport复制到node把node加入cluster节点字典/* 解析gossip */if sender不存在 && 消息类型为MEET：处理消息的gossip回复PONG消息给对端节点

处理PING，PONG，MEET消息

if 消息类型是PING或者PONG或者MEET:node = link关联的nodeif link是outbound:if node处于handshake:/* sender已经代表了对端节点，接下去node的ID也会被更新成跟sender的一样，同一个ID不能对应2个实体，所以需要把node删掉 */if sender存在: 检查ip和ports的变化，更新sender /* Point-1 */从cluster节点字典中删除node /* 会释放node的inbound和outbound link*/return更新node的ID = 消息头的sender /* 随机ID -> 真实的ID */从node中删除HANDSHAKE标记 /* 表示handshake过程结束 */根据消息头的flags，把node设置为master或slaveelse if node的ID != 消息头的sender:/* 可能对端变更了ID，也可能是由于网络的变化连错了节点 */标记node为NOADDR重置node的ip，port，pport，cport为0释放linkreturn/* 检查NOFAILOVER，更新sender */if sender存在:	/* 我们假定对端发来的信息是最新的，所以直接更新.NOFAILOVER对应的是cluster-replica-no-failover配置项 */if 消息头的flags带了NOFAILOVER:标记sender为NOFAILOVERelse:移除sender的NOFAILOVER/* 如果ip和各个端口发生了变化，则需要更新 */if sender存在 && 消息类型为PING && sender没有处于handshake:检查ip和ports的变化，更新sender/* 如果我们收到了PONG，需要清除PFAIL和FAIL状态 */if link是outbound && 消息类型为PONG:更新node的pongReceived = 当前时间重置node的pingSent = 0 if node处于PFAIL:清除PFAIL状态	/* 简单地移除PFAIL标记	*/else if node处于FAIL:清除node的FAIL状态 /* Point-2 *//* 检查是否发生了角色切换: slave -> master or master -> slave */if sender存在:if 消息头的slaveof为全0: /* 对端现在是master */把sender转换为master	 /* Point-3: 如果sender原本就是master，就无需转换 */else:	/* 对端现在是slave */if sender需要从master变为slave:把sender转换为slave	/* Point-4 */if sender的master发生了变化: 把sender从旧master迁移到新master /* Point-5 *//* 更新slots */if sender是master && sender的slots跟消息头的myslots存在不同:重新绑定消息头的myslots到sender /* Point-6 *//* 如果sender声称slot属于它，但事实并非如此，则需要通知sender */if sender存在 && sender的slots跟消息头的myslots存在不同:查找消息头的myslots中是否存在这样一个slot:1. 在myself看来，它不属于sender2. 它的configEpoch > 消息头的configEpochif 存在这样的slot:发送UPDATE消息给sender/* 解决configEpoch冲突 */		if sender是master && myself也是master &&消息头的configEpoch == myself的configEpoch:if myself的ID < sender的ID:	/* 字符串比较 */myself的configEpoch = (++server的currentEpoch)保存配置/* 处理gossip */		if sender存在:处理消息中的gossip  /* Point-7 *//* 处理消息中的pingExtension */if 消息头的extensions > 0: /* 消息带有extension *//* 当前只有一种extension: pingExtension */sender的hostname = pingExtension中的hostname

Points 解释

Point-1：检查ip和ports的变化，更新sender

/* 检查是否使用同一条连接，若是，则不可能有变化 */
if link跟sender的link一样:	return	
/* 检查IP和ports是否有变化 */
if sender的ports跟消息头的ports相同 &&: /* port, pport, cport */sender的ip跟消息头的myip相同: /* 如果myip为全0，则使用socket上的对端IP */return
更新sender的IP和ports
if sender的link存在: /* IP和ports变更了，需要重新建立连接 */释放掉sender的link	/* 释放后会使用新的IP和port自动重连 */
if sender是myself的master:更新server复制时使用的IP和port，重新开启复制

Point-2：清除node的FAIL状态

/* 对于slave节点，只要连接上，都会认为它恢复了;对于没有slot的master节点，只要连接上，就不需要在看它是否fail了足够长的时间 */
if sender是slave || sender没有slots:清除它的FAIL标记/* 对于master节点，如果它fail的时间足够长，而且还有slots属于它，说明它没有被failover，现在连接上了，则可以认为它恢复了 */
if sender是master && sender有slots &&当前时间 - sender的failTime > 2 * cluster-node-timeout:清除它的FAIL标记

Point-3：把sender转换为master

解除sender和oldMaster之间的关联
if oldMaster没有其他slave:清除oldMaster的MIGRATE_TO标记 
if sender != myself:添加MIGRATE_TO标记到sender
设置sender为master

Point-4：把sender转换为slave

把原本属于sender的slots，全部设置为没有owner
清除sender的MIGRATE_TO标记
设置sender为slave

Point-5：把sender从原本的master迁移到新的master

解除sender和oldMaster之间的关联
if oldMaster没有其他slave:清除oldMaster的MIGRATE_TO标记
建立sender和newMaster之间的关联
添加MIGRATE_TO标记到newMaster

Point-6：重新绑定消息头的myslots到sender

if sender == myself:  /* 不需要对自己进行更新 */return
if myself是master:currentMaster = myself
elsecurrentMaster = myself的master
newMaster = null
migratedOurSlots = 0
遍历消息头中的myslots:跳过那些已经属于sender的slot跳过处于importing状态的slot/* 如果slot没有owner，或者sender声称slot属于它 */if slot没有owner || slot的configEpoch < 消息头的configEpoch:if slot的owner == myself && slot上有key: /* slot是myself的，需要标识它是dirty */记录slot到dirtySlotsif slot的owner == currentMaster:/* 存在slot从currentMaster迁移到sender */newMaster = sender++migratedOurSlots /* 记录迁移的slot数 */把slot从旧的owner删除把slot添加到senderif newMaster != null &&		/* 存在slot从currentMaster迁移到sender */currentMaster没有slot了 && (cluster-allow-replica-migration ||	/* 配置了允许副本迁移 */消息头的myslots数量 == migratedOurSlots): /* myslots全部从currentMaster迁移到sender *//* 如果myself是master，当前没有slot，说明myself被failover，它需要成为newMaster的副本；如果myself是slave，它的master没有slot，说明它当前没有slot可以复制，myself需要成为myslots的副本 *//* sender成为myself的master */if myself是master:清除它的MIGRATE_TO标记设置它为slave清除它的那些处于importing和migrating的slotselse:解除它跟oldMaster之间的关联把myself的master设置为sender更新server复制时使用的IP和port，重新开启复制重置server的manual failover状态
else if dirtySlots数量 > 0:/* 消息表明这些slots跟我们没有关系了，但是我们还有key在上面，为了维持key和slot之间的一致性，需要清除掉这些key */遍历dirtySlots，删除每个slot上所有的key

Point-7：处理消息中的gossip

遍历消息中的每个gossip:根据gossip的nodename，从cluster节点字典中查找nodeif node存在:  /* 找到关联这个gossip的node */if sender是master && node != myself:if gossip表示node处于PFAIL或FAIL: /* 对端节点连接不上node */把sender发起的failureReport添加到node /* 如果已经存在，则更新时间 *//* 检查node是否需要标记为FAIL */neededQuorum = clusterSize / 2 + 1 /* 计算仲裁所需要的票数 */if node处于PFAIL状态 && /* 我们无法连接上它 */node没有处于FAIL状态: /* 没FAIL才需要处理 */清除那些已经失效的failureReport /* 时间超过了 2 * cluster-node-timeout */failures = 有效的failureReport数if myself是master:	/* 如果我们是master，我们也需要投票 */++failuresif falures >= neededQuorum: /* 多数通过, FAIL成立 */把node从PFAIL转换成FAIL，并设置failTime为当前时间广播FAIL消息到其他节点，迫使它们标记此node已FAILelse: /* sender能连接上node */从node上删除sender发起的failureReport /* 如果有的话 */清除那些已经失效的failureReport/* 检查是否需要更新node的pongReceived */if gossip表示node没有处于PFAIL或FAIL && /* 在对端节点看来，node没有PFAIL或FAIL */node的pingSent == 0 &&	/* 我们没有处于等待回复的ping */node上没有failureReport:	 /* 在我们看来，node没有FAIL */if gossip的pongReceived > node的pongReceived &&	/* 对端节点收到node的PONG比我们更新 */gossip的pongReceived < 当前时间 + 500ms:	/* 各个节点的时钟可能不同步，500ms是容错 */		node的pongReceived = gossip的pongReceived/* 检查IP和ports是否发生了变化 */	if node原本处于PFAIL或FAIL &&	/* 我们连接不上它 */但对端节点能连接上node &&	我们和对端节点拿到的node IP，port和cport不一样:释放掉我们对node的link更新node上的IP和ports，等待下次自动重连else:	/* node在我们这还不存在 */		/* 可以通过"cluster forget"命令把node从server的cluster节点字典中删掉，为了避免node的地址和端口被重用，错误连接到别的cluster，于是需要把删掉的node加入到server的黑名单中 */if sender存在 && gossip表示node是有地址的 &&gossip的nodename不在我们的黑名单中:创建node实体填入gossip的IP和ports加入cluster节点字典