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

temp remove command line arg parser

Browse Source
This commit is contained in:
Ara Sadoyan
2025-03-20 18:33:21 +01:00
parent 7501e367c6
commit 5524fd5011
9 changed files with 430 additions and 562 deletions
Download Patch File Download Diff File Expand all files Collapse all files

583
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

29
Cargo.toml
View File

@@ -4,25 +4,18 @@ version = "0.1.0"
edition = "2021"
[dependencies]
bytes = "1.9.0"
mini-redis = "0.4"
rand = "0.8.5"
tokio = { version = "1.44.1", features = ["full"] }
pingora = { version = "0.4.0", features = ["lb", "openssl"] } # openssl, rustls, boringssl
serde = { version = "1.0.219", features = ["derive"] }
dashmap = "6.1.0"
tokio = { version = "1", features = ["full"] }
#pingora = { version = "0.4", features = ["full"] }
pingora = { version = "0.4.0", features = ["lb", "openssl"] }
pingora-core = { version = "0.4.0" }
pingora-proxy = { version = "0.4.0" }
pingora-load-balancing = { version = "0.4.0" }
pingora-http = { version = "0.4.0" }
async-trait = { version = "0.1.85" }
env_logger = { version = "0.11.6" }
log = "0.4.25"
pingora-core = "0.4.0"
pingora-proxy = "0.4.0"
pingora-http = "0.4.0"
async-trait = "0.1.88"
env_logger = "0.11.7"
log = "0.4.26"
futures = "0.3.31"
notify = "8.0.0"
axum = "0.8.1"
serde = { version = "1.0.217", features = ["derive"] }
serde_json = "1.0.137"
reqwest = { version = "0.12" }
serde_yaml = "0.8.26"
clap = { version = "4.5.32", features = ["derive"] }
reqwest = "0.12.15"
serde_yaml = "0.8.26"

57
etc/upstreams.yaml
View File

@@ -1,49 +1,56 @@
upstreams:
myip.netangels.net:
paths:
"/draw":
protocol: "wsoc"
ssl: false
servers:
- "192.168.1.1:8000"
"/":
protocol: "http"
ssl: false
servers:
- "127.0.0.1:8000"
- "127.0.0.3:8000"
- "127.0.0.4:8000"
- "127.0.0.5:8000"
"/ping":
protocol: "http"
"/echo":
protocol: "X-Proxied-From:Fooooooooooooooo, X-DoSome-About:Yaaaaaaaaaaaaaaa"
ssl: false
servers:
- "127.0.0.1:8000"
- "127.0.0.2:8000"
- "127.0.0.3:8000"
- "127.0.0.4:8000"
"/ping":
protocol: "PLACEHOLDER"
ssl: false
servers:
- "127.0.0.1:8000"
- "127.0.0.2:8000"
"/draw":
protocol: "PLACEHOLDER"
ssl: false
servers:
- "192.168.1.1:8000"
polo.netangels.net:
paths:
"/":
protocol: "http"
protocol: "PLACEHOLDER"
ssl: false
servers:
- "192.168.1.10:8000"
"/ws":
protocol: "wsoc"
protocol: "PLACEHOLDER"
ssl: false
servers:
- "192.168.1.1:8000"
glop.netangels.net:
paths:
"/":
protocol: "http"
protocol: "PLACEHOLDER"
ssl: false
servers:
- "192.168.1.10:8000"
- "192.168.1.10:8000"
# apt.netangels.net:
# paths:
# "/":
# protocol: "http"
# ssl: true
# servers:
# - "apt.netangels.net:443"
apt.netangels.net:
paths:
"/":
protocol: "PLACEHOLDER"
ssl: true
servers:
- "apt.netangels.net:443"
127.0.0.1:
paths:
"/camerastatus":
protocol: "PLACEHOLDER"
ssl: false
servers:
- "192.168.1.5:8080"

47
src/utils/discovery.rs
View File

@@ -2,21 +2,22 @@ use crate::utils::parceyaml::load_yaml_to_dashmap;
use crate::utils::tools::*;
use crate::web::webserver;
use async_trait::async_trait;
use dashmap::DashMap;
use futures::channel::mpsc::Sender;
use futures::SinkExt;
use log::{error, info};
use notify::event::ModifyKind;
use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use std::fs;
use pingora::prelude::sleep;
use std::path::Path;
use std::sync::atomic::AtomicUsize;
use std::time::{Duration, Instant};
use tokio::task;
pub struct FromFileProvider {
pub path: String,
}
pub struct APIUpstreamProvider;
pub struct APIUpstreamProvider {
pub address: String,
}
#[async_trait]
pub trait Discovery {
@@ -26,7 +27,7 @@ pub trait Discovery {
#[async_trait]
impl Discovery for APIUpstreamProvider {
async fn start(&self, toreturn: Sender<UpstreamsDashMap>) {
webserver::run_server(toreturn).await;
webserver::run_server(self.address.clone(), toreturn).await;
}
}
@@ -37,16 +38,18 @@ impl Discovery for FromFileProvider {
}
}
pub async fn watch_file(fp: String, mut toreturn: Sender<UpstreamsDashMap>) {
sleep(Duration::from_millis(50)).await; // For having nice logs :-)
let file_path = fp.as_str();
let parent_dir = Path::new(file_path).parent().unwrap();
let (local_tx, mut local_rx) = tokio::sync::mpsc::channel::<notify::Result<Event>>(1);
println!("Watching for changes in {:?}", parent_dir);
let paths = fs::read_dir(parent_dir).unwrap();
for path in paths {
println!(" {}", path.unwrap().path().display())
}
info!("Watching for changes in {:?}", parent_dir);
let snd = load_yaml_to_dashmap(file_path, "filepath");
let _ = toreturn.send(snd).await.unwrap();
match snd {
Some(snd) => {
toreturn.send(snd).await.unwrap();
}
None => {}
}
let _watcher_handle = task::spawn_blocking({
let parent_dir = parent_dir.to_path_buf(); // Move directory path into the closure
@@ -72,29 +75,38 @@ pub async fn watch_file(fp: String, mut toreturn: Sender<UpstreamsDashMap>) {
if e.paths[0].to_str().unwrap().ends_with("yaml") {
if start.elapsed() > Duration::from_secs(2) {
start = Instant::now();
println!("Config File changed :=> {:?}", e);
info!("Config File changed :=> {:?}", e);
let snd = load_yaml_to_dashmap(file_path, "filepath");
let _ = toreturn.send(snd).await.unwrap();
match snd {
Some(snd) => {
toreturn.send(snd).await.unwrap();
}
None => {}
}
}
}
}
_ => (),
},
Err(e) => println!("Watch error: {:?}", e),
Err(e) => error!("Watch error: {:?}", e),
}
}
}
/*
#[allow(dead_code)]
pub fn build_upstreams(d: &str, kind: &str) -> UpstreamsDashMap {
let upstreams: UpstreamsDashMap = DashMap::new();
let mut contents = d.to_string();
match kind {
"filepath" => {
println!("Reading upstreams from {}", d);
let _ = match fs::read_to_string(d) {
Ok(data) => contents = data,
Ok(data) => {
println!("Reading upstreams from {}", d);
contents = data
}
Err(e) => {
eprintln!("Error reading file: {:?}", e);
error!("Reading upstreams file: {:?}", e);
return upstreams;
}
};
@@ -146,3 +158,4 @@ pub fn build_upstreams(d: &str, kind: &str) -> UpstreamsDashMap {
}
upstreams
}
*/

63
src/utils/parceyaml.rs
View File

@@ -1,6 +1,8 @@
use crate::utils::tools::*;
use dashmap::DashMap;
use log::{error, info, warn};
use serde::{Deserialize, Serialize};
use serde_yaml::Error;
use std::collections::HashMap;
use std::fs;
use std::sync::atomic::AtomicUsize;
@@ -22,40 +24,63 @@ struct PathConfig {
servers: Vec<String>,
}
pub fn load_yaml_to_dashmap(d: &str, kind: &str) -> UpstreamsDashMap {
pub fn load_yaml_to_dashmap(d: &str, kind: &str) -> Option<UpstreamsDashMap> {
let dashmap = UpstreamsDashMap::new();
let mut yaml_data = d.to_string();
match kind {
"filepath" => {
println!("Reading upstreams from {}", d);
let _ = match fs::read_to_string(d) {
Ok(data) => yaml_data = data,
Ok(data) => {
info!("Reading upstreams from {}", d);
yaml_data = data
}
Err(e) => {
eprintln!("Error reading file: {:?}", e);
return dashmap;
error!("Reading: {}: {:?}", d, e.to_string());
warn!("Running with empty upstreams list, chane it via API");
return None;
}
};
}
"content" => {
println!("Reading upstreams from API post body");
info!("Reading upstreams from API post body");
}
_ => println!("*******************> nothing <*******************"),
_ => error!("Mismatched parameter, only filepath|content is allowed "),
}
let parsed: Config = serde_yaml::from_str(&yaml_data).expect("Failed to parse YAML");
for (hostname, host_config) in parsed.upstreams {
let path_map = DashMap::new();
for (path, path_config) in host_config.paths {
let mut server_list = Vec::new();
for server in path_config.servers {
if let Some((ip, port_str)) = server.split_once(':') {
if let Ok(port) = port_str.parse::<u16>() {
server_list.push((ip.to_string(), port, path_config.ssl, path_config.protocol.clone()));
let p: Result<Config, Error> = serde_yaml::from_str(&yaml_data);
match p {
Ok(parsed) => {
for (hostname, host_config) in parsed.upstreams {
let path_map = DashMap::new();
for (path, path_config) in host_config.paths {
let mut server_list = Vec::new();
for server in path_config.servers {
if let Some((ip, port_str)) = server.split_once(':') {
if let Ok(port) = port_str.parse::<u16>() {
server_list.push((ip.to_string(), port, path_config.ssl, path_config.protocol.clone()));
}
}
}
path_map.insert(path, (server_list, AtomicUsize::new(0)));
}
dashmap.insert(hostname, path_map);
}
path_map.insert(path, (server_list, AtomicUsize::new(0)));
Some(dashmap)
}
Err(e) => {
error!("Failed to parse upstreams file: {}", e);
None
}
dashmap.insert(hostname, path_map);
}
dashmap
}
pub fn parce_main_config(path: &str) -> DashMap<String, String> {
info!("Parsing configuration");
let data = fs::read_to_string(path).unwrap();
let reply = DashMap::new();
let cfg: HashMap<String, String> = serde_yaml::from_str(&*data).expect("Failed to parse main config file");
for (k, v) in cfg {
reply.insert(k.to_string(), v.to_string());
}
reply
}

128
src/web/proxyhttp.rs
View File

@@ -2,38 +2,59 @@ use crate::utils::discovery::{APIUpstreamProvider, Discovery, FromFileProvider};
use crate::utils::tools::*;
use crate::utils::*;
use async_trait::async_trait;
use dashmap::DashMap;
use futures::channel::mpsc;
use futures::StreamExt;
use log::{info, warn};
use log::{debug, error, info, warn};
use pingora::prelude::*;
use pingora_core::prelude::HttpPeer;
use pingora_core::server::ShutdownWatch;
use pingora_core::services::background::BackgroundService;
use pingora_http::{RequestHeader, ResponseHeader};
use pingora_proxy::{ProxyHttp, Session};
use std::ops::Deref;
use std::sync::atomic::Ordering;
use std::sync::Arc;
pub struct LB {
pub ump_upst: Arc<UpstreamsDashMap>,
pub ump_full: Arc<UpstreamsDashMap>,
pub config: Arc<DashMap<String, String>>,
pub local: Arc<(String, u16)>,
}
#[async_trait]
impl BackgroundService for LB {
async fn start(&self, mut shutdown: ShutdownWatch) {
println!("Starting example background service");
let (tux, mut rux) = mpsc::channel::<UpstreamsDashMap>(0);
let file_load2 = FromFileProvider {
path: "etc/upstreams.yaml".to_string(),
};
info!("Starting background service");
let (tx, mut rx) = mpsc::channel::<UpstreamsDashMap>(0);
let api_load = APIUpstreamProvider;
let from_file = self.config.get("upstreams_conf");
match from_file {
Some(from_file) => {
let file_load = FromFileProvider { path: from_file.to_string() };
let tx_file = tx.clone();
let _ = tokio::spawn(async move { file_load.start(tx_file).await });
}
None => {
error!("Can't read config file");
}
}
let config_address = self.config.get("config_address");
match config_address {
Some(config_address) => {
let api_load = APIUpstreamProvider {
address: config_address.to_string(),
};
let tx_api = tx.clone();
let _ = tokio::spawn(async move { api_load.start(tx_api).await });
}
None => {
error!("Can't read config file");
}
}
let tux_file = tux.clone();
let tux_api = tux.clone();
let _ = tokio::spawn(async move { file_load2.start(tux_file).await });
let _ = tokio::spawn(async move { api_load.start(tux_api).await });
let uu = self.ump_upst.clone();
let ff = self.ump_full.clone();
let _ = tokio::spawn(async move { healthcheck::hc2(uu, ff).await });
@@ -43,7 +64,7 @@ impl BackgroundService for LB {
_ = shutdown.changed() => {
break;
}
val = rux.next() => {
val = rx.next() => {
match val {
Some(ss) => {
let foo = compare_dashmaps(&*self.ump_full, &ss);
@@ -67,26 +88,31 @@ pub trait GetHost {
}
#[async_trait]
impl GetHost for LB {
async fn get_host(&self, peer: &str, path: &str, upgrade: bool) -> Option<(String, u16, bool, String)> {
let mut _proto = "";
if upgrade {
_proto = "wsoc";
} else {
_proto = "http"
}
let host_entry = self.ump_upst.get(peer).unwrap();
let x = if let Some(entry) = host_entry.get(path) {
let (servers, index) = entry.value();
if servers.is_empty() {
return None;
async fn get_host(&self, peer: &str, path: &str, _upgrade: bool) -> Option<(String, u16, bool, String)> {
let _proto = "";
// if upgrade {
// _proto = "wsoc";
// } else {
// _proto = "http"
// }
let host_entry = self.ump_upst.get(peer);
match host_entry {
Some(host_entry) => {
let upstream = if let Some(entry) = host_entry.get(path) {
let (servers, index) = entry.value();
if servers.is_empty() {
return None;
}
let idx = index.fetch_add(1, Ordering::Relaxed) % servers.len();
// info!("Host: {}, Path: {}, Peer: {:?}, len: {}, idx: {}", peer, path, servers[idx], servers.len(), idx);
Some(servers[idx].clone())
} else {
None
};
upstream
}
let idx = index.fetch_add(1, Ordering::Relaxed) % servers.len();
println!("{} {:?} => len: {}, idx: {}", peer, servers[idx], servers.len(), idx);
Some(servers[idx].clone())
} else {
None
};
x
None => None,
}
}
}
@@ -95,31 +121,26 @@ impl ProxyHttp for LB {
type CTX = ();
fn new_ctx(&self) -> Self::CTX {}
async fn upstream_peer(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<Box<HttpPeer>> {
// let before = Instant::now();
let host_name = session.req_header().headers.get("host");
match host_name {
Some(host) => {
let header_host = host.to_str().unwrap().split(':').collect::<Vec<&str>>();
let ddr = self.get_host(header_host[0], session.req_header().uri.path(), session.is_upgrade_req());
match ddr.await {
Some((host, port, ssl, _proto)) => {
let peer = Box::new(HttpPeer::new((host, port), ssl, String::new()));
// info!("{:?}, Time => {:.2?}", session.request_summary(), before.elapsed());
Ok(peer)
}
None => {
warn!("Returning default list => {:?}, {:?}", host_name, session.req_header().uri);
let peer = Box::new(HttpPeer::new(("127.0.0.1", 3000), false, String::new()));
// info!("{:?}, Time => {:.2?}", session.request_summary(), before.elapsed());
warn!("Upstream not found. Host: {:?}, Path: {}", host, session.req_header().uri);
let peer = Box::new(HttpPeer::new(self.local.deref(), false, String::new()));
Ok(peer)
}
}
}
None => {
warn!("Returning default list => {:?}, {:?}", host_name, session.req_header().uri);
let peer = Box::new(HttpPeer::new(("127.0.0.1", 3000), false, String::new()));
// info!("{:?}, Time => {:.2?}", session.request_summary(), before.elapsed());
warn!("Upstream not found. Host: {:?}, Path: {}", host_name, session.req_header().uri);
let peer = Box::new(HttpPeer::new(self.local.deref(), false, String::new()));
Ok(peer)
}
}
@@ -159,15 +180,32 @@ impl ProxyHttp for LB {
}
Ok(())
}
async fn response_filter(&self, _session: &mut Session, _upstream_response: &mut ResponseHeader, _ctx: &mut Self::CTX) -> pingora_core::Result<()>
async fn response_filter(&self, _session: &mut Session, _upstream_response: &mut ResponseHeader, _ctx: &mut Self::CTX) -> Result<()>
where
Self::CTX: Send + Sync,
{
// -------------------------------------------------------------------------------------------- //
// let header_host = _session.req_header().headers.get("host");
// match header_host {
// Some(header_host) => {
// let data = self.ump_full.get(header_host.to_str().unwrap()).unwrap();
// let detail = data.get(_session.req_header().uri.path());
// let thelast = detail.unwrap().value().0[0].clone().3;
// println!(" Host ==> {}", thelast);
// }
// None => {}
// }
// _upstream_response.insert_header("X-DoSome-About", "Yaaaaaaaaaaaaaaa").unwrap();
// -------------------------------------------------------------------------------------------- //
_upstream_response.insert_header("X-Proxied-From", "Fooooooooooooooo").unwrap();
Ok(())
}
// async fn logging(&self, session: &mut Session, _e: Option<&pingora::Error>, ctx: &mut Self::CTX) {
// let response_code = session.response_written().map_or(0, |resp| resp.status.as_u16());
// info!("{}, response code: {response_code}", self.request_summary(session, ctx));
// }
async fn logging(&self, session: &mut Session, _e: Option<&pingora::Error>, ctx: &mut Self::CTX) {
let response_code = session.response_written().map_or(0, |resp| resp.status.as_u16());
debug!("{}, response code: {response_code}", self.request_summary(session, ctx));
}
}

59
src/web/start.rs
View File

@@ -1,47 +1,74 @@
use crate::utils::tools::*;
use crate::web::proxyhttp::LB;
// use clap::{arg, Parser};
use dashmap::DashMap;
use pingora_core::prelude::{background_service, Opt};
use pingora_core::server::Server;
use std::env;
use std::sync::Arc;
// #[derive(Parser, Debug)]
// #[command(version, about, long_about = None)]
// struct Args {
// #[arg(short, long)]
// address: String,
// #[arg(short, long)]
// port: String,
// }
pub fn run() {
env_logger::init();
let parameters = Some(Opt::parse_args()).unwrap();
let file = parameters.conf.clone().unwrap();
let maincfg = crate::utils::parceyaml::parce_main_config(file.as_str());
// let mut server = Server::new(None).unwrap();
let mut server = Server::new(Some(Opt::parse_args())).unwrap();
let mut local_conf: (String, u16) = ("0.0.0.0".to_string(), 0);
if let Some((ip, port_str)) = maincfg.get("config_address").unwrap().split_once(':') {
if let Ok(port) = port_str.parse::<u16>() {
local_conf = (ip.to_string(), port);
}
}
let mut server = Server::new(parameters).unwrap();
server.bootstrap();
let uf: UpstreamsDashMap = DashMap::new();
let ff: UpstreamsDashMap = DashMap::new();
let uf_config = Arc::new(uf);
let ff_config = Arc::new(ff);
let cfg = Arc::new(maincfg);
let local = Arc::new(local_conf);
let lb = LB {
ump_upst: uf_config.clone(),
ump_full: ff_config.clone(),
config: cfg.clone(),
local: local.clone(),
};
let bg = LB {
ump_upst: uf_config.clone(),
ump_full: ff_config.clone(),
config: cfg.clone(),
local: local.clone(),
};
// env_logger::Env::new();
// env_logger::init();
let log_level = cfg.get("log_level").unwrap();
match log_level.as_str() {
"info" => env::set_var("RUST_LOG", "info"),
"error" => env::set_var("RUST_LOG", "error"),
"warn" => env::set_var("RUST_LOG", "warn"),
"debug" => env::set_var("RUST_LOG", "debug"),
"trace" => env::set_var("RUST_LOG", "trace"),
"off" => env::set_var("RUST_LOG", "off"),
_ => {
println!("Error reading log level, defaulting to: INFO");
env::set_var("RUST_LOG", "info")
}
}
env_logger::builder()
// .format_timestamp(None)
// .format_module_path(false)
// .format_source_path(false)
// .format_target(false)
.init();
let bg_srvc = background_service("bgsrvc", bg);
let mut proxy = pingora_proxy::http_proxy_service(&server.configuration, lb);
let bindaddress = cfg.get("proxy_address_http").unwrap();
// let args = Args::parse();
// let addr = format!("{}:{}", args.address, args.port);
proxy.add_tcp("0.0.0.0:6193");
proxy.add_tcp(bindaddress.as_str());
server.add_service(proxy);
server.add_service(bg_srvc);

24
src/web/webserver.rs
View File

@@ -6,14 +6,11 @@ use axum::routing::{delete, get, head, post, put};
use axum::Router;
use futures::channel::mpsc::Sender;
use futures::SinkExt;
use log::info;
use tokio::net::TcpListener;
// struct UpstreamData {
// servers: UpstreamsDashMap,
// }
#[allow(unused_mut)]
pub async fn run_server(mut toreturn: Sender<UpstreamsDashMap>) {
pub async fn run_server(bindaddress: String, mut toreturn: Sender<UpstreamsDashMap>) {
let mut tr = toreturn.clone();
let app = Router::new()
.route("/{*wildcard}", get(getconfig))
@@ -25,13 +22,22 @@ pub async fn run_server(mut toreturn: Sender<UpstreamsDashMap>) {
"/conf",
post(|up: String| async move {
let serverlist = crate::utils::parceyaml::load_yaml_to_dashmap(up.as_str(), "content");
let _ = tr.send(serverlist).await.unwrap();
Response::builder().status(StatusCode::CREATED).body(Body::from("Config, conf file, updated!\n")).unwrap()
match serverlist {
Some(serverlist) => {
let _ = tr.send(serverlist).await.unwrap();
Response::builder().status(StatusCode::CREATED).body(Body::from("Config, conf file, updated!\n")).unwrap()
}
None => Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from("Failed to parce config file!\n"))
.unwrap(),
}
})
.with_state("state"),
);
let listener = TcpListener::bind("0.0.0.0:3000").await.unwrap();
println!("Axum API server running on port 3000");
let listener = TcpListener::bind(bindaddress.clone()).await.unwrap();
info!("Starting the API server on: {}", bindaddress);
axum::serve(listener, app).await.unwrap();
}

2
start.sh
View File

@@ -1,4 +1,4 @@
#!/bin/bash
export RUST_LOG=INFO
#export RUST_LOG=INFO
#reflex -d none -r 'src/' -s -- sh -c 'reset && cargo run -- --address 0.0.0.0 --port 6193'
reflex -d none -r 'src/' -s -- sh -c 'reset && cargo run -- -c etc/main.yaml'