Github/aralez
1
0
Fork 0
mirror of https://github.com/sadoyan/aralez.git synced 2026-04-29 22:38:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

Hatchecks kind of complete

Browse Source
This commit is contained in:
Ara Sadoyan
2025-02-25 21:22:29 +01:00
parent 1338d04963
commit c3bd2f143e
7 changed files with 905 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

867
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

7
Cargo.toml
View File

@@ -6,13 +6,8 @@ edition = "2021"
[dependencies]
bytes = "1.9.0"
mini-redis = "0.4"
#async-trait = "0.1"
hyper-reverse-proxy = "0.5"
hyper = { version = "0.14", features = ["full"] }
rand = "0.8.5"
dashmap = "6.1.0"
http-body-util = "0.1"
hyper-util = { version = "0.1", features = ["full"] }
tokio = { version = "1", features = ["full"] }
#pingora = { version = "0.4", features = ["full"] }
pingora = { version = "0.4.0", features = ["lb"] }
@@ -28,4 +23,4 @@ notify = "8.0.0"
axum = "0.8.1"
serde = { version = "1.0.217", features = ["derive"] }
serde_json = "1.0.137"
reqwest = { version = "0.12" }

23
src/utils/compare.rs
View File

@@ -1,6 +1,6 @@
use dashmap::DashMap;
use std::sync::atomic::AtomicUsize;
use tokio::sync::RwLockWriteGuard;
use tokio::sync::{RwLockReadGuard, RwLockWriteGuard};
#[allow(dead_code)]
pub fn dashmaps(map1: &RwLockWriteGuard<DashMap<String, (Vec<(String, u16)>, AtomicUsize)>>, map2: &DashMap<String, (Vec<(String, u16)>, AtomicUsize)>) -> bool {
@@ -22,3 +22,24 @@ pub fn dashmaps(map1: &RwLockWriteGuard<DashMap<String, (Vec<(String, u16)>, Ato
}
true
}
#[allow(dead_code)]
pub fn dm(map1: &RwLockReadGuard<DashMap<String, (Vec<(String, u16)>, AtomicUsize)>>, map2: &DashMap<String, (Vec<(String, u16)>, AtomicUsize)>) -> bool {
if map1.len() != map2.len() {
return false; // Different number of keys
}
for entry1 in map1.iter() {
let key = entry1.key();
let (vec1, _) = entry1.value(); // Extract value
if let Some(entry2) = map2.get(key) {
let (vec2, _) = entry2.value(); // Correctly extract value
if vec1 != vec2 {
return false;
}
} else {
return false;
}
}
true
}

30
src/utils/discovery.rs
View File

@@ -25,16 +25,10 @@ pub trait Discovery {
#[async_trait]
impl Discovery for APIUpstreamProvider {
async fn run(&self, toreturn: Sender<DashMap<String, (Vec<(String, u16)>, AtomicUsize)>>) {
let _ = tokio::spawn(async move { webserver::run_server(toreturn).await });
loop {
// let dm: DashMap<String, (Vec<(String, u16)>, AtomicUsize)> = DashMap::new();
// dm.insert(
// "popok.netangels.net".to_string(),
// (vec![("192.168.1.1".parse().unwrap(), 8000), ("192.168.1.10".parse().unwrap(), 8000)], AtomicUsize::new(0)),
// );
// let _ = toreturn.send(dm).await.unwrap();
tokio::time::sleep(Duration::from_secs(20)).await;
}
webserver::run_server(toreturn).await;
// let _ = tokio::spawn(async move { webserver::run_server(toreturn).await });
// let (_rtx, mut rrx) = tokio::sync::mpsc::channel::<bool>(1);
// let _ = rrx.blocking_recv();
}
}
@@ -55,7 +49,7 @@ pub async fn watch_file(fp: String, mut toreturn: Sender<DashMap<String, (Vec<(S
println!(" {}", path.unwrap().path().display())
}
let snd = read_upstreams_from_file(file_path, "filepath");
let snd = build_upstreams(file_path, "filepath");
let _ = toreturn.send(snd).await.unwrap();
let _watcher_handle = task::spawn_blocking({
@@ -69,10 +63,12 @@ pub async fn watch_file(fp: String, mut toreturn: Sender<DashMap<String, (Vec<(S
)
.unwrap();
watcher.watch(&parent_dir, RecursiveMode::Recursive).unwrap();
loop {
std::thread::sleep(Duration::from_secs(50));
}
// println!("========================");
let (_rtx, mut rrx) = tokio::sync::mpsc::channel::<bool>(1);
let _ = rrx.blocking_recv();
// loop {
// std::thread::sleep(Duration::from_secs(50));
// }
}
});
let mut start = Instant::now();
@@ -87,7 +83,7 @@ pub async fn watch_file(fp: String, mut toreturn: Sender<DashMap<String, (Vec<(S
start = Instant::now();
println!("Config File changed :=> {:?}", e);
let snd = read_upstreams_from_file(file_path, "filepath");
let snd = build_upstreams(file_path, "filepath");
let _ = toreturn.send(snd).await.unwrap();
}
}
@@ -98,7 +94,7 @@ pub async fn watch_file(fp: String, mut toreturn: Sender<DashMap<String, (Vec<(S
}
}
}
pub fn read_upstreams_from_file(d: &str, kind: &str) -> DashMap<String, (Vec<(String, u16)>, AtomicUsize)> {
pub fn build_upstreams(d: &str, kind: &str) -> DashMap<String, (Vec<(String, u16)>, AtomicUsize)> {
let upstreams = DashMap::new();
let mut contents = d.to_string();
match kind {

88
src/utils/healthcheck.rs
View File

@@ -6,25 +6,85 @@ use tokio::sync::RwLock;
use tokio::time::interval;
pub async fn hc(upslist: Arc<RwLock<DashMap<String, (Vec<(String, u16)>, AtomicUsize)>>>, fullist: Arc<RwLock<DashMap<String, (Vec<(String, u16)>, AtomicUsize)>>>) {
let mut period = interval(Duration::from_secs(20));
let mut period = interval(Duration::from_secs(2));
loop {
tokio::select! {
_ = period.tick() => {
let ups = upslist.write().await;
let full = fullist.write().await;
for val in full.iter_mut() {
// making some dummy ligic
match val.key().to_string().as_str() {
"polo.netangels.net" => ups.remove("polo.netangels.net"),
"glop.netangels.net" => ups.remove("glop.netangels.net"),
_ => ups.remove(""),
};
// let before = Instant::now();
let totest: DashMap<String, (Vec<(String, u16)>, AtomicUsize)> = DashMap::new();
let fclone: DashMap<String, (Vec<(String, u16)>, AtomicUsize)> = DashMap::new();
// println!("\nElapsed dash: {:.2?}", before.elapsed());
// let before = Instant::now();
{
let full = fullist.read().await;
for v in full.iter() {
fclone.insert(v.key().clone(), (v.value().0.clone(), AtomicUsize::new(0)));
}
} // lock releases when scope ends
// println!("Elapsed full: {:.2?}", before.elapsed());
for val in fclone.iter() {
let mut newvec = vec![];
for hostport in val.value().0.clone(){
let hostpart = hostport.0.split('/').last().unwrap(); // For later use
let url = format!("http://{}:{}", hostpart, hostport.1);
let resp = http_request(url.as_str(), "GET", "").await;
match resp{
true => {
newvec.push((hostpart.to_string(), hostport.1));
},
false => {
println!("Dead upstream. Host: {}, Upstream: {}:{} ",val.key(), hostpart.to_string(), hostport.1 );
}
}
}
totest.insert(val.key().clone(), (newvec, AtomicUsize::new(0)));
}
// println!("UPS: {:?}", ups);
drop(ups);
drop(full);
// let before = Instant::now();
{
let upsl = upslist.read().await;
if !crate::utils::compare::dm(&upsl, &totest) {
println!("Dashmaps not matched, synchronizing");
upsl.clear();
for (k, v) in totest { // loop takes the ownership
println!("Host: {}", k);
for vv in &v.0 {
println!(" :===> {:?}", vv);
}
upsl.insert(k, v);
}
}
}
// println!("Elapsed upsl: {:.2?}", before.elapsed());
}
}
}
}
async fn http_request(url: &str, method: &str, payload: &str) -> bool {
let client = reqwest::Client::new();
let to = Duration::from_secs(1);
match method {
"POST" => {
let response = client.post(url).body(payload.to_owned()).timeout(to).send().await;
match response {
Ok(r) => 100 <= r.status().as_u16() && r.status().as_u16() < 500,
Err(_) => false,
}
}
"GET" => {
let response = client.get(url).timeout(to).send().await;
match response {
Ok(r) => {
// println!("Response: {} : {}", r.status(), r.url());
100 <= r.status().as_u16() && r.status().as_u16() < 500
}
Err(_) => {
// println!("Error: {}", url);
false
}
}
}
_ => false,
}
}

46
src/web/proxyhttp.rs
View File

@@ -23,7 +23,6 @@ pub struct LB {
#[async_trait]
impl BackgroundService for LB {
async fn start(&self, mut shutdown: ShutdownWatch) {
tokio::spawn(healthcheck::hc(self.upstreams.clone(), self.umap_full.clone()));
println!("Starting example background service");
let (tx, mut rx) = mpsc::channel::<DashMap<String, (Vec<(String, u16)>, AtomicUsize)>>(0);
@@ -37,6 +36,9 @@ impl BackgroundService for LB {
let tx_api = tx.clone();
let _ = tokio::spawn(async move { api_load.run(tx_api).await });
let _ = tokio::spawn(async move { file_load.run(tx_file).await });
let up = self.upstreams.clone();
let fu = self.umap_full.clone();
let _ = tokio::spawn(async move { healthcheck::hc(up, fu).await });
loop {
tokio::select! {
@@ -46,22 +48,32 @@ impl BackgroundService for LB {
val = rx.next() => {
match val {
Some(newmap) => {
let umap_work = self.upstreams.write().await;
let umap_full = self.umap_full.write().await;
if !compare::dashmaps(&umap_full, &newmap) {
umap_work.clear();
umap_full.clear();
for (k,v) in newmap {
println!("Host: {}", k);
for vv in v.0.clone() {
println!(" ===> {:?}", vv);
let umap_work = self.upstreams.read().await;
let umap_full = self.umap_full.read().await;
match compare::dm(&umap_full, &newmap) {
false => {
drop(umap_full);
drop(umap_work);
let work = self.upstreams.write().await;
let full = self.umap_full.write().await;
work.clear();
full.clear();
for (k,v) in newmap {
println!("Host: {}", k);
for vv in v.0.clone() {
println!(" ===> {:?}", vv);
}
work.insert(k.clone(), (v.0.clone(), AtomicUsize::new(0))); // No need for extra vec!
full.insert(k, (v.0, AtomicUsize::new(0))); // Use `value.0` directly
}
umap_work.insert(k.clone(), (v.0.clone(), AtomicUsize::new(0))); // No need for extra vec!
umap_full.insert(k, (v.0, AtomicUsize::new(0))); // Use `value.0` directly
drop(full);
drop(work);
}
true => {
drop(umap_full);
drop(umap_work);
}
}
drop(umap_full);
drop(umap_work);
}
None => {}
}
@@ -86,7 +98,7 @@ impl GetHost for LB {
return None;
}
let idx = index.fetch_add(1, Ordering::Relaxed) % servers.len();
println!("{} {:?} => len: {}, idx: {}", peer, servers[idx], servers.len(), idx);
// println!("{} {:?} => len: {}, idx: {}", peer, servers[idx], servers.len(), idx);
Some(servers[idx].clone())
} else {
None
@@ -109,8 +121,8 @@ impl ProxyHttp for LB {
Ok(peer)
}
None => {
println!("Returning default list => {:?}", ("127.0.0.1", 8000));
let peer = Box::new(HttpPeer::new(("127.0.0.1", 8000), false, "".to_string()));
println!("Returning default list => {:?}", ("127.0.0.1", 3000));
let peer = Box::new(HttpPeer::new(("127.0.0.1", 3000), false, "".to_string()));
Ok(peer)
}
}

2
src/web/webserver.rs
View File

@@ -23,7 +23,7 @@ pub async fn run_server(mut toreturn: Sender<DashMap<String, (Vec<(String, u16)>
.route(
"/conf",
post(|up: String| async move {
let serverlist = crate::utils::discovery::read_upstreams_from_file(up.as_str(), "content");
let serverlist = crate::utils::discovery::build_upstreams(up.as_str(), "content");
let _ = tr.send(serverlist).await.unwrap();
Response::builder().status(StatusCode::CREATED).body(Body::from("Config, conf file, updated!\n")).unwrap()
})