meta_srv/
discovery.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15pub mod lease;
16pub mod node_info;
17pub mod node_scheduling;
18pub mod utils;
19
20use api::v1::meta::heartbeat_request::NodeWorkloads;
21use common_error::ext::BoxedError;
22use common_meta::cluster::NodeInfo;
23use common_meta::distributed_time_constants::default_distributed_time_constants;
24use common_meta::error::Result;
25use common_meta::peer::{Peer, PeerResolver};
26use common_meta::{DatanodeId, FlownodeId};
27use common_time::util::DefaultSystemTimer;
28use snafu::ResultExt;
29
30use crate::cluster::MetaPeerClient;
31use crate::discovery::lease::{LeaseValueAccessor, LeaseValueType};
32
33// NOTE: these are inherent methods on purpose. `MetaPeerClient` does not implement
34// `PeerDiscovery`, so the only way to obtain a `PeerDiscoveryRef` inside metasrv is
35// `discovery::SchedulingAwarePeerDiscovery`, which stamps node scheduling states. Wrapping the raw
36// client directly would silently drop the drain fence.
37impl MetaPeerClient {
38    /// Returns all frontends that reported a heartbeat recently.
39    pub async fn active_frontends(&self) -> Result<Vec<NodeInfo>> {
40        utils::alive_frontend_infos(
41            &DefaultSystemTimer,
42            self,
43            default_distributed_time_constants().frontend_heartbeat_interval,
44        )
45        .await
46        .map_err(BoxedError::new)
47        .context(common_meta::error::ExternalSnafu)
48    }
49
50    /// Returns all datanodes that reported a heartbeat recently, without scheduling states.
51    pub async fn active_datanodes(
52        &self,
53        filter: Option<for<'a> fn(&'a NodeWorkloads) -> bool>,
54    ) -> Result<Vec<NodeInfo>> {
55        utils::alive_datanode_infos(
56            &DefaultSystemTimer,
57            self,
58            default_distributed_time_constants().datanode_lease,
59            filter,
60        )
61        .await
62        .map_err(BoxedError::new)
63        .context(common_meta::error::ExternalSnafu)
64    }
65
66    /// Returns all flownodes that reported a heartbeat recently.
67    pub async fn active_flownodes(
68        &self,
69        filter: Option<for<'a> fn(&'a NodeWorkloads) -> bool>,
70    ) -> Result<Vec<NodeInfo>> {
71        utils::alive_flownode_infos(
72            &DefaultSystemTimer,
73            self,
74            default_distributed_time_constants().flownode_lease,
75            filter,
76        )
77        .await
78        .map_err(BoxedError::new)
79        .context(common_meta::error::ExternalSnafu)
80    }
81}
82
83#[async_trait::async_trait]
84impl PeerResolver for MetaPeerClient {
85    async fn datanode(&self, id: DatanodeId) -> Result<Option<Peer>> {
86        let peer = self
87            .lease_value(LeaseValueType::Datanode, id)
88            .await
89            .map_err(BoxedError::new)
90            .context(common_meta::error::ExternalSnafu)?
91            .map(|(peer_id, lease)| Peer::new(peer_id, lease.node_addr));
92        Ok(peer)
93    }
94
95    async fn flownode(&self, id: FlownodeId) -> Result<Option<Peer>> {
96        let peer = self
97            .lease_value(LeaseValueType::Flownode, id)
98            .await
99            .map_err(BoxedError::new)
100            .context(common_meta::error::ExternalSnafu)?
101            .map(|(peer_id, lease)| Peer::new(peer_id, lease.node_addr));
102        Ok(peer)
103    }
104}
meta_srv/discovery.rs

meta_srv/
discovery.rs
