Skip to content

Commit

Permalink
fix: ports table in scheme
Browse files Browse the repository at this point in the history
  • Loading branch information
@DmitryTravyan
DmitryTravyan committed Jun 22, 2022
1 parent 3f197f4 commit 39e0f00
Show file tree
Hide file tree
Showing 4 changed files with 94 additions and 110 deletions.
2 changes: 1 addition & 1 deletion Cargo.lock
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

26 changes: 14 additions & 12 deletions src/libs/ins/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,6 @@ impl Instance {
roles: self.roles.clone(),
config: self.config.clone(),
})
.rev()
.collect()];
result.extend(match self.itype {
Type::Storage => (1..=self.count)
Expand All @@ -129,17 +128,20 @@ impl Instance {
config: self.config.clone(),
})
.rev()
.chain((1..=master_num).map(|num| Instance {
name: format!("dummy-{}", num),
parent: format!("dummy-{}", num),
count: 1,
replicas: 0,
itype: Type::Dummy,
weight: self.weight,
stateboard: false,
roles: self.roles.clone(),
config: self.config.clone(),
}))
.chain(
(1..=master_num)
.map(|num| Instance {
name: format!("dummy-{}", num),
parent: format!("dummy-{}", num),
count: 1,
replicas: 0,
itype: Type::Dummy,
weight: self.weight,
stateboard: false,
roles: self.roles.clone(),
config: self.config.clone(),
})
)
.collect::<Vec<Instance>>()
})
.collect(),
Expand Down
1 change: 1 addition & 0 deletions src/task/cluster/hosts.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ pub(in crate::task) struct FlatHost {
pub(in crate::task) instances: Vec<Instance>,
}

#[allow(unused)]
impl FlatHosts {
/// Recursively iterate over datacentres and inners.
/// Create list of hosts and return it.
Expand Down
175 changes: 78 additions & 97 deletions src/task/cluster/scheme.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,7 @@ use genin::libs::{
ins::{Instance, Role, Type},
vrs::Vars,
};
use indexmap::IndexMap;
use log::{debug, info, trace};
use log::{debug, info, trace, warn};
use prettytable::{color, Attr, Cell, Row, Table};
use serde_yaml::Value;

Expand All @@ -18,7 +17,7 @@ use super::{hosts::FlatHost, Cluster};
pub(in crate::task) struct Scheme {
pub(in crate::task) hosts: FlatHosts,
pub(in crate::task) vars: Vars,
pub(in crate::task) ports_map: IndexMap<String, String>,
pub(in crate::task) ports_vec: Vec<(u16, u16)>,
}

impl<'a> TryFrom<&'a Cluster> for Scheme {
Expand All @@ -44,7 +43,7 @@ impl<'a> TryFrom<&'a Cluster> for Scheme {

let mut hosts = FlatHosts::try_from(&cluster.hosts)?;
let mut ports = PortsVariants::None;
let mut ports_map = IndexMap::new();
ports.or_else(hosts[0].ports);

// Each iteration is Vec with non multiplied instance
// 1. multiply instance to `count()` and collect it as vector of vectors with Instance
Expand All @@ -55,105 +54,88 @@ impl<'a> TryFrom<&'a Cluster> for Scheme {
.instances
.iter()
.flat_map(|instance| instance.multiply())
.for_each(|mut multiplied_instances| {
debug!(
"mutliplied: {:?}",
&multiplied_instances
.rev()
.fold(
vec![Vec::new(), Vec::new()],
|mut acc: Vec<Vec<Instance>>, instances| {
trace!(
"{:?}",
instances
.iter()
.map(|ins| ins.name.as_str())
.collect::<Vec<&str>>()
);
match instances.last() {
Some(Instance {
itype: Type::Router | Type::Storage | Type::Dummy | Type::Replica,
..
}) => acc.push(instances),
_ => acc[0].extend(instances),
}
acc
},
)
.into_iter()
.rev()
.for_each(|mut instances| {
trace!(
"resulted instances: {:?}",
instances
.iter()
.map(|ins| ins.name.as_str())
.map(|instance| instance.name.as_str())
.collect::<Vec<&str>>()
);
debug!("starting port {:?}", &ports);
// Ports should upped after
// 1. new instance
// 2. hosts loop ended
ports.up();
let mut i = 0;
(0..hosts.len())
.cycle()
.scan((), |_, index| {
trace!("working with host with index {}", index);
multiplied_instances.pop().map(|mut instance| {
instances.pop().map(|instance| {
instance
.is_not_dummy()
.then(|| {
ports.or_else(hosts[index].ports);
i.eq(&hosts.len())
.then(|| {
ports.up();
i = 0;
})
.or_else(|| {
i += 1;
None
});
trace!(
"pushing {} to host with index {}",
instance.name,
index
);
ports_map.insert(
instance.name.to_string(),
format!(
"{}/{}",
ports.http_or_default(),
ports.binary_or_default()
),
);
instance.config.insert(
"advertise_uri".into(),
Value::String(format!(
"{}:{}",
hosts[index].ip.to_string(),
ports.binary_or_default()
)),
);
instance.config.insert(
"http_port".into(),
Value::String(ports.http_or_default().to_string()),
);
hosts.deref_mut()[index].instances.push(instance)
})
.or(None)
})
})
.for_each(|_| {});
});
let ports_vec = (1..=hosts[0].instances.len())
.map(|_| {
let (http, binary) = (ports.http_or_default(), ports.binary_or_default());
ports.up();
(http, binary)
})
.collect::<Vec<(u16, u16)>>();

(1..hosts.len()).for_each(|iteration| {
let (left, right) = hosts.split_at_mut(iteration);
left.last_mut()
.map(|left| {
right.iter_mut().rev().for_each(|right| {
left.instances
.last()
.map(|llast| {
llast.itype.eq(&Type::Custom)
&& right
.instances
.last()
.map(|rlast| !rlast.parent.eq(&llast.parent))
.unwrap_or_else(|| false)
&& left.instances.len() > right.instances.len()
&& left.instances.len() - right.instances.len() >= 2
})
.unwrap_or_else(|| false)
.then(|| {
trace!("moving instance from {} to {}", left.name(), right.name());
left.instances
.pop()
.map(|instance| right.instances.push(instance))
.unwrap_or_else(|| {});
})
.map(|_| {
debug!("instance moved from {} to {}", left.name(), right.name())
})
.unwrap_or_else(|| {});
});
})
.unwrap_or_else(|| {});
hosts.iter_mut().for_each(|flhosts| {
warn!("instances len: {}", flhosts.instances.len());
let ip = flhosts.ip.to_string();
flhosts
.instances
.iter_mut()
.enumerate()
.for_each(|(index, instance)| {
warn!("index: {} ports_vec: {:?}", index, ports_vec);
instance.config.insert(
"advertise_uri".into(),
Value::String(format!("{}:{}", ip, ports_vec[index].1)),
);
instance.config.insert(
"http_port".into(),
Value::String(ports_vec[index].0.to_string()),
);
});
});

// Add stateboard entity
//
// cartridge_failover_params:
Expand Down Expand Up @@ -205,7 +187,7 @@ impl<'a> TryFrom<&'a Cluster> for Scheme {
Ok(Scheme {
hosts,
vars: cluster.vars.clone(),
ports_map,
ports_vec,
})
}
}
Expand Down Expand Up @@ -241,32 +223,31 @@ impl Scheme {
.collect::<Vec<Cell>>(),
));

(0..self.hosts.max_len()).for_each(|pos| {
table.add_row(Row::new(
vec![Cell::new(format!("{}/{}", 8081, 3031).as_str())]
.into_iter()
.chain(self.hosts.iter().map(|host| {
host.instances
.get(pos)
.map(|instance| {
trace!("instance name {:?}", self.ports_map.get(&instance.name));
match (instance.itype, self.ports_map.get(&instance.name)) {
(_, None) => Cell::default(),
(Type::Router, Some(_)) => Cell::new(&instance.name)
self.ports_vec
.iter()
.enumerate()
.for_each(|(pos, (http, binary))| {
table.add_row(Row::new(
vec![Cell::new(format!("{}/{}", http, binary).as_str())]
.into_iter()
.chain(self.hosts.iter().map(|host| {
host.instances
.get(pos)
.map(|instance| match instance.itype {
Type::Router => Cell::new(&instance.name)
.with_style(Attr::ForegroundColor(color::BLUE)),
(Type::Storage, Some(_)) => Cell::new(&instance.name)
Type::Storage => Cell::new(&instance.name)
.with_style(Attr::ForegroundColor(color::BRIGHT_GREEN)),
(Type::Replica, Some(_)) => Cell::new(&instance.name)
Type::Replica => Cell::new(&instance.name)
.with_style(Attr::ForegroundColor(color::GREEN)),
_ => Cell::new(&instance.name)
.with_style(Attr::ForegroundColor(color::CYAN)),
}
})
.unwrap_or_else(Cell::default)
}))
.collect::<Vec<Cell>>(),
));
});
})
.unwrap_or_else(Cell::default)
}))
.collect::<Vec<Cell>>(),
));
});
table.printstd();
}

Expand Down

0 comments on commit 39e0f00

Please sign in to comment.