Qdrant 集群下的共识服务(中)

qdrant 集群在启动的时候会进行健康检查 其中最重要是同步日志即保证commitLogIndex是一致。

1.健康检查

    // 集群模式会做健康检查
    let health_checker = Arc::new(common::health::HealthChecker::spawn(
        toc_arc.clone(),
        consensus_state.clone(),
        &runtime_handle,
        consensus_state.is_new_deployment() && bootstrap.is_some(),
    ));

    // 设置当前启动实例的一些状态，同时将检查的操作封装到Task中，调用exec() 来处理
    pub fn spawn(
        toc: Arc<TableOfContent>,
        consensus_state: ConsensusStateRef,
        runtime: &runtime::Handle,
        wait_for_bootstrap: bool,
    ) -> Self {
        let task = Task {
            toc,
            consensus_state,
            is_ready: Default::default(),
            is_ready_signal: Default::default(),
            check_ready_signal: Default::default(),
            cancel: Default::default(),
            wait_for_bootstrap,
        };

        let health_checker = Self {
            is_ready: task.is_ready.clone(),
            is_ready_signal: task.is_ready_signal.clone(),
            check_ready_signal: task.check_ready_signal.clone(),
            _cancel: task.cancel.clone().drop_guard(),
        };
        // 生成任务
        let task = runtime.spawn(task.exec());
    }
    // 具体实现
        async fn exec_impl(&self) {
        // 从集群中的响应中取出一个作为当前日志的索引 不一定是最终 
        let Some(mut cluster_commit_index) = self.cluster_commit_index(true).await else {
            self.set_ready();
            return;
        };

        // 根据raft的算法给所有的其他节点确保至少一半的正确回复，循环当前的实例与其他的节点的
        // commmit_log_index 达到相同
        loop {
   
            match self.cluster_commit_index(false).await {
                // 相同结束
                Some(new_index) if cluster_commit_index == new_index => break,
                // 更新当前实例的index
                Some(new_index) => cluster_commit_index = new_index,
                // 错误退出
                None => break,
            }
        }
    }



        async fn cluster_commit_index(&self, one_peer: bool) -> Option<u64> {
 
        // 获取集群对应的节点信息并发送请求
        let peer_address_by_id = self.consensus_state.peer_address_by_id();
        let transport_channel_pool = &self.toc.get_channel_service().channel_pool;
        let this_peer_id = self.toc.this_peer_id;
        let this_peer_uri = peer_address_by_id.get(&this_peer_id);
        let mut requests = peer_address_by_id
            .values()
            .filter(|&uri| Some(uri) != this_peer_uri)

            .unique()
            .map(
                |uri| 
                // 发送请求 最终是调用函数with_channel_timeout
                get_consensus_commit(transport_channel_pool, uri)
                )
            .collect::<FuturesUnordered<_>>()
            .inspect_err(|err| log::error!("GetConsensusCommit request failed: {err}"))
            .filter_map(|res| future::ready(res.ok()));      
        // Raft 共识操作要求至少达到大多数(所有节点/ 2 + 1)
        let sufficient_commit_indices_count = if !one_peer {
            // 当前节点算一个节点 所以 就是直接除以2
            peer_address_by_id.len() / 2
        } else {
            1
        };

        // 等待 `total nodex / 2` 成功响应
        let mut commit_indices: Vec<_> = (&mut requests)
            .take(sufficient_commit_indices_count)
            .collect()
            .await;
        // 找到所有所有响应中最大commit_index作为同步的commit_index
        let cluster_commit_index = commit_indices
            .into_iter()
            .map(|resp| resp.into_inner().commit)
            .max()
            .unwrap_or(0);

        Some(cluster_commit_index as _)
    }

    
    pub async fn with_channel_timeout<T, O: Future<Output = Result<T, Status>>>(
        &self,
        uri: &Uri,
        f: impl Fn(InterceptedService<Channel, AddTimeout>) -> O,
        timeout: Option<Duration>,
        retries: usize,
    ) -> Result<T, RequestError<Status>> {
        // 重试
        let mut retries_left = retries;
        let mut attempt = 0;
        let max_timeout = timeout.unwrap_or_else(|| self.grpc_timeout + self.connection_timeout);

        loop {
            let request_result: Result<T, _> = self.make_request(uri, &f, max_timeout).await;

            let error_result = match request_result {
                Ok(body) => return Ok(body),
                Err(err) => err,
            };

            let action = match error_result {
                RequestFailure::HealthCheck(healthcheck_error) => {
                    // 根据不同的错误类型来处理 
                }

            let (backoff_time, fallback_status) = match action {
                RetryAction::Fail(err) => return Err(RequestError::FromClosure(err)),
                RetryAction::RetryImmediately(fallback_status) => (Duration::ZERO, fallback_status),
                RetryAction::RetryWithBackoff(fallback_status) => {
                    //backoff  失败后会加上随机时间来重试 
                    let backoff = DEFAULT_BACKOFF * 2u32.pow(attempt as u32)
                        + Duration::from_millis(rng().random_range(0..100));

                    if backoff > max_timeout {
                        return Err(RequestError::FromClosure(fallback_status));
                    }
                    (backoff, fallback_status)
                }
                RetryAction::RetryOnce(fallback_status) => {
                    if retries_left > 1 {
                        retries_left = 1;
                    }
                    (Duration::ZERO, fallback_status)
                }
            };

            attempt += 1;
            // 多次重试失败后直接异常
            if retries_left == 0 {
                return Err(RequestError::FromClosure(fallback_status));
            }
            retries_left = retries_left.saturating_sub(1);
            // 休眠
            tokio::time::sleep(backoff_time).await;
        }
        }
    }