meta_srv/selector/
lease_based.rs1use common_meta::affinity::AffinityFallback;
16use common_meta::peer::Peer;
17
18use crate::error::{self, Result};
19use crate::metasrv::SelectorContext;
20use crate::selector::common::{self, choose_items, filter_out_excluded_peers};
21use crate::selector::weighted_choose::{RandomWeightedChoose, WeightedItem};
22use crate::selector::{Selector, SelectorOptions};
23
24#[derive(Default)]
26pub struct LeaseBasedSelector;
27
28impl LeaseBasedSelector {
29 async fn select_once(
30 &self,
31 ctx: &SelectorContext,
32 opts: &SelectorOptions<'_>,
33 ) -> Result<Vec<Peer>> {
34 let alive_datanodes = common::active_schedulable_datanodes(ctx, opts.workload_filter)
36 .await?
37 .into_iter()
38 .filter(|node| {
39 opts.affinity
40 .is_none_or(|requirements| ctx.affinity.matches(requirements, node))
41 })
42 .map(|node| node.peer)
43 .collect::<Vec<_>>();
44
45 let mut weight_array = alive_datanodes
47 .into_iter()
48 .map(|p| WeightedItem {
49 item: p,
50 weight: 1.0,
51 })
52 .collect();
53
54 filter_out_excluded_peers(&mut weight_array, &opts.exclude_peer_ids);
56 let mut weighted_choose = RandomWeightedChoose::new(weight_array);
57 let selected = choose_items(opts, &mut weighted_choose)?;
58
59 Ok(selected)
60 }
61}
62
63#[async_trait::async_trait]
64impl Selector for LeaseBasedSelector {
65 type Context = SelectorContext;
66 type Output = Vec<Peer>;
67
68 async fn select(
69 &self,
70 ctx: &Self::Context,
71 mut opts: SelectorOptions<'_>,
72 ) -> Result<Self::Output> {
73 match self.select_once(ctx, &opts).await {
74 Err(error::Error::NoEnoughAvailableNode {
75 required,
76 available,
77 ..
78 }) if opts.affinity.is_some()
79 && ctx.affinity_fallback == AffinityFallback::Unconstrained
80 && available < required =>
81 {
82 opts.affinity = None;
83 self.select_once(ctx, &opts).await
84 }
85 result => result,
86 }
87 }
88}