initial commit

2026-04-29 22:38:36 +08:00 · 2025-02-07 15:12:28 +01:00
parent 320348c85f
commit b503f866b1
12 changed files with 2047 additions and 62 deletions
--- a/src/backup/handler.rs
+++ b/src/backup/handler.rs
@@ -0,0 +1,54 @@
+use crate::peers::Peers;
+use hyper::{Body, Request, Response, StatusCode};
+use std::convert::Infallible;
+use std::net::IpAddr;
+use std::thread::sleep;
+use std::time::Duration;
+
+pub async fn proxy_http(client_ip: IpAddr, req: Request<Body>, db: Peers) -> Result<Response<Body>, Infallible> {
+    let p = db.clone();
+    let rurl = req.uri().path();
+    let yoyo = req.uri().path().to_string(); // Bad thing, only for debug
+    let peer = crate::peers::return_peer(p, rurl);
+    match hyper_reverse_proxy::call(client_ip, peer.as_ref(), req).await {
+        Ok(response) => {
+            println!("Peer: {}, Client: {}, Path: {}, Status: {}", peer.as_ref(), client_ip, yoyo, response.status());
+            Ok(response)
+        }
+        Err(_error) => {
+            println!("Error: no live peers for: {}", yoyo);
+            Ok(Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::empty()).unwrap())
+        }
+    }
+}
+
+pub async fn healthcheck(peers: Peers, work: Peers) -> Result<Response<Body>, Infallible> {
+    loop {
+        println!("Main Peers -> {:?}", peers);
+        println!("Work Peers -> {:?}", work);
+        sleep(Duration::from_secs(10));
+
+        peers.clone().iter().for_each(|peer| {
+            work.insert(peer.key().clone(), peer.value().clone());
+            // work[peer.key()] = peer.value();
+        });
+    }
+}
+
+// #[tokio::main]
+// async fn client_check(url: &str) {
+//     let url = url.parse::<hyper::Uri>().unwrap();
+//     let host = url.host().expect("uri has no host");
+//     let port = url.port_u16().unwrap_or(80);
+//     let address = format!("{}:{}", host, port);
+//     let stream = TcpStream::connect(address).await.unwrap();
+//     let io = TokioIo::new(stream);
+//     let (mut sender, conn) = hyper::client::conn::http1::handshake(io).await.unwrap();
+//
+//     // Spawn a task to poll the connection, driving the HTTP state
+//     tokio::task::spawn(async move {
+//         if let Err(err) = conn.await {
+//             println!("Connection failed: {:?}", err);
+//         }
+//     });
+// }
--- a/src/backup/peers.rs
+++ b/src/backup/peers.rs
@@ -0,0 +1,28 @@
+use dashmap::DashMap;
+use rand::Rng;
+use std::sync::Arc;
+
+// type Db = Arc<RwLock<HashMap<Arc<str>, Arc<i32>>>>;
+pub type Peers = Arc<DashMap<Arc<str>, Vec<Arc<str>>>>;
+
+pub fn add_peers(peers: Peers, path: &str) {
+    if let Some(mut peers_list) = peers.get_mut(&Arc::from(path)) {
+        peers_list.push(Arc::from("http://192.168.1.1:8000"));
+        peers_list.push(Arc::from("http://192.168.1.10:8000"));
+    }
+    println!("Adding peers {} -> {:?}", peers.get(path).unwrap().key(), peers.get(path).unwrap().value());
+}
+pub fn return_peer(peers: Peers, path: &str) -> Arc<str> {
+    if let Some(peer_list) = peers.get(&Arc::from(path)) {
+        let mut rng = rand::thread_rng();
+        let r = rng.gen_range(0..peer_list.len());
+
+        if let Some(selected_peer) = peer_list.get(r) {
+            selected_peer.clone()
+        } else {
+            Arc::from("https://127.0.0.1:8443")
+        }
+    } else {
+        Arc::from("https://127.0.0.1:8443")
+    }
+}
--- a/src/backup/start.rs
+++ b/src/backup/start.rs
@@ -0,0 +1,45 @@
+use dashmap::DashMap;
+use hyper::server::conn::AddrStream;
+use hyper::service::{make_service_fn, service_fn};
+use hyper::Server;
+use web::handler;
+use web::peers;
+
+#[tokio::main]
+async fn main() {
+    let bind_addr = "0.0.0.0:6193";
+    let addr: SocketAddr = bind_addr.parse().expect("Could not parse ip:port.");
+
+    let main_peers: peers::Peers = Arc::new(DashMap::new());
+    let work_peers: peers::Peers = Arc::new(DashMap::new());
+
+    main_peers.insert(Arc::from("/first"), vec![]);
+    let p = main_peers.clone();
+    peers::add_peers(p, "/first");
+
+    main_peers.insert(Arc::from("/second"), vec![]);
+    let r = main_peers.clone();
+    peers::add_peers(r, "/second");
+
+    let main_prs = main_peers.clone();
+    let work_prs = work_peers.clone();
+    tokio::spawn(async move {
+        let h = handler::healthcheck(main_prs.clone(), work_prs.clone());
+        h.await.expect("health check failed");
+        // handler::healthcheck(main_prs.clone(), work_prs.clone()).await;
+    });
+
+    let make_svc = make_service_fn(|conn: &AddrStream| {
+        let remote_addr = conn.remote_addr().ip();
+        let value = work_peers.clone();
+        async move { Ok::<_, Infallible>(service_fn(move |req| handler::proxy_http(remote_addr, req, value.clone()))) }
+    });
+
+    let server = Server::bind(&addr).serve(make_svc);
+
+    println!("Running server on {:?}", addr);
+
+    if let Err(e) = server.await {
+        println!("server error: {}", e);
+    }
+}
--- a/src/backup/tcpcache.rs
+++ b/src/backup/tcpcache.rs
@@ -0,0 +1,113 @@
+/*
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+use tokio::sync::RwLock;
+
+type Db = Arc<RwLock<HashMap<Arc<[u8]>, Arc<[u8]>>>>;
+const DBG: bool = true;
+#[tokio::main]
+pub async fn run() {
+    println!("\n= = = = = = = = ASYNC TOKIO = = = = = = = = =\n");
+    if 1 == 1 {
+        return;
+    }
+    let listener = TcpListener::bind("0.0.0.0:6379").await.unwrap();
+    println!("Server is running on 0.0.0.0:6379 !\n");
+    let hashmap: Db = Arc::new(RwLock::new(HashMap::new()));
+    loop {
+        let (socket, _) = listener.accept().await.unwrap();
+        let hashmap = hashmap.clone();
+        tokio::spawn(async move {
+            process(socket, hashmap).await;
+        });
+    }
+}
+
+async fn process(mut socket: TcpStream, db: Db) {
+    let mut buf = vec![0; 1024];
+    loop {
+        let n = socket.read(&mut buf).await.expect("failed to read data from socket");
+        match n > 3 {
+            true => {
+                if let Some((action, key, value)) = process_data(&buf[..n - 1], b' ') {
+                    let mut map = db.write().await;
+                    match action {
+                        [115, 101, 116] => {
+                            // SET
+                            map.insert(Arc::from(key), Arc::from(value));
+                            socket.write_all("Done!\n".as_ref()).await.expect("failed to write");
+                        }
+                        [103, 101, 116] => {
+                            // GET
+                            let t = map.get(&Arc::from(key));
+                            match t {
+                                Some(t) => {
+                                    socket.write_all(t.as_ref()).await.expect("failed to write");
+                                }
+                                None => {
+                                    socket.write_all("Not Found !\n".as_ref()).await.expect("failed to read");
+                                }
+                            }
+                        }
+                        [100, 101, 108] => {
+                            // DEL
+                            let y = map.remove(&Arc::from(key));
+                            let mut _mssg = "";
+                            match y {
+                                Some(_) => {
+                                    _mssg = "Deleted !\n";
+                                }
+                                None => _mssg = "Not Found !\n",
+                            }
+                            socket.write_all(_mssg.as_ref()).await.expect("failed to write");
+                        }
+                        [98, 121, 101] => {
+                            //BYE
+                            socket.write_all("Bye!: Closing the connection\n".as_ref()).await.expect("failed");
+                            socket.shutdown().await.expect("shutdown socket error");
+                            return;
+                        }
+                        _ => socket.write_all("Unknown command: send (get/set/del)\n".as_ref()).await.expect("failed to read"),
+                    }
+                }
+            }
+            false => {
+                socket.write_all("Only get/set/del commands are accepted\n".as_ref()).await.expect("failed to respond");
+            }
+        }
+    }
+}
+
+fn process_data(data: &[u8], delim: u8) -> Option<(&[u8], &[u8], &[u8])> {
+    let action_bytes = &data[..3];
+
+    if DBG {
+        match data.get(4..) {
+            Some(_d) => {
+                println!(" DEBUG => {} : {} ", std::str::from_utf8(action_bytes).ok()?, std::str::from_utf8(&data[4..]).ok()?);
+            }
+            None => println!(" DEBUG => Goodbye"),
+        }
+    }
+
+    match action_bytes {
+        [103, 101, 116] | [100, 101, 108] => {
+            let key = &data[4..];
+            let val = &[];
+            return Some((action_bytes, key, val));
+        }
+        [115, 101, 116] => {
+            if let Some(pos) = data[4..].iter().position(|&b| b == delim) {
+                let (key_bytes, value_bytes) = data[4..].split_at(pos);
+                let value_bytes = &value_bytes[1..];
+                return Some((action_bytes, key_bytes, value_bytes));
+            }
+        }
+        _ => return Some((action_bytes, &[], &[])),
+    }
+
+    None
+}
+*/
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,5 +1,5 @@
 mod web;

 fn main() {
-    web::run::run();
+    web::start::run()
 }
--- a/src/web.rs
+++ b/src/web.rs
@@ -1 +1,5 @@
-pub mod run;
+pub mod proxyhttp;
+pub mod start;
+// pub mod tcpcache;
+// pub mod handler;
+// pub mod peers;
--- a/src/web/proxyhttp.rs
+++ b/src/web/proxyhttp.rs
@@ -0,0 +1,228 @@
+use async_trait::async_trait;
+use dashmap::DashMap;
+use log::{info, warn};
+use pingora::prelude::*;
+use pingora_core::prelude::HttpPeer;
+use pingora_http::{RequestHeader, ResponseHeader};
+use pingora_proxy::{ProxyHttp, Session};
+use std::any::type_name;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+#[allow(dead_code)]
+pub fn typeoff<T>(_: T) -> &'static str {
+    let to = type_name::<T>();
+    println!("{:?}", to);
+    to
+}
+
+// pub struct LB(pub Arc<LoadBalancer<RoundRobin>>);
+#[allow(dead_code)]
+pub struct LB {
+    // pub load_balancer: Arc<LoadBalancer<RoundRobin>>,
+    // pub upstreams_map: Arc<HashMap<String, Vec<(String, u16)>>>,
+    // pub upstreams_map: Arc<Mutex<HashMap<String, Vec<(String, u16)>>>>,
+    // upstreams: DashMap<String, (Vec<(&'static str, u16)>, AtomicUsize)>,
+    // pub upstreams_map: DashMap<String, Vec<(String, u16)>>,
+    pub upstreams_map: DashMap<String, (Vec<(String, u16)>, AtomicUsize)>,
+    // pub upstreams_maps: DashMap<String, Arc<LoadBalancer<RoundRobin>>>,
+}
+
+pub trait GetHost {
+    fn get_host(&self, peer: &str) -> Option<(String, u16)>;
+    fn set_host(&mut self, peer: &str, host: &str, port: u16);
+}
+
+impl GetHost for LB {
+    fn get_host(&self, peer: &str) -> Option<(String, u16)> {
+        // println!("{:?}", self.upstreams_map);
+        // let entry = self.upstreams_map.get(peer)?;
+        // let first = entry.value().first()?;
+        // println!("{:?}", entry.value());
+        // Some((first.0.clone(), first.1))
+
+        let entry = self.upstreams_map.get(peer)?;
+        let (servers, index) = entry.value();
+
+        if servers.is_empty() {
+            return None;
+        }
+
+        let idx = index.fetch_add(1, Ordering::Relaxed) % servers.len();
+        println!("{} {:?} => len: {}, idx: {}", peer, servers[idx], servers.len(), idx);
+        Some(servers[idx].clone())
+    }
+
+    fn set_host(&mut self, peer: &str, host: &str, port: u16) {
+        // let new_value = vec![(host.to_string(), port)];
+        // self.upstreams_map.insert(peer.to_string(), (new_value, AtomicUsize::new(0)));
+
+        let exists = self.upstreams_map.get(peer);
+        let mut toreturn = vec![];
+        match exists {
+            Some(e) => {
+                let (ko, _) = e.value();
+                let new_value = vec![(host.to_string(), port)];
+                for (k, v) in ko.clone().iter() {
+                    toreturn.push((k.to_string(), v.to_owned()));
+                }
+                toreturn.push(new_value[0].clone());
+            }
+            None => {
+                toreturn.push((host.to_string(), port));
+            }
+        }
+
+        println!(" ==> Updating peer list: name => {} | value => {:?}", peer.to_string(), toreturn);
+        self.upstreams_map.insert(peer.to_string(), (toreturn, AtomicUsize::new(0)));
+
+        // self.upstreams_map.insert(peer.to_string(), toreturn);
+
+        // use std::time::Instant;
+        // let now = Instant::now();
+        // self.get_host(peer);
+        // let elapsed = now.elapsed();
+        // println!("Elapsed: {:.2?}", elapsed);
+    }
+}
+
+#[async_trait]
+impl ProxyHttp for LB {
+    type CTX = ();
+    fn new_ctx(&self) -> Self::CTX {}
+    // async fn upstream_peer(&self, _session: &mut Session, _ctx: &mut ()) -> Result<Box<HttpPeer>> {
+    //     let upstream = self.load_balancer.select(b"", 256).ok_or_else(|| Error::explain(HTTPStatus(503), "no upstreams"))?;
+    //     let peer = HttpPeer::new(upstream.addr, false, "".to_string());
+    //
+    //     let host_name = _session.req_header().headers.get("host");
+    //     let fo = self.get_host(host_name.unwrap().to_str().unwrap());
+    //     println!("{:?}", fo);
+    //
+    //     Ok(Box::new(peer))
+    // }
+
+    async fn upstream_peer(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<Box<HttpPeer>> {
+        let host_name = session.req_header().headers.get("host");
+        let ddr = self.get_host(host_name.unwrap().to_str().unwrap());
+        match ddr {
+            Some((host, port)) => {
+                let peer = Box::new(HttpPeer::new((host, port), false, "".to_string()));
+                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()));
+                Ok(peer)
+            }
+        }
+    }
+
+    /*
+
+        async fn upstream_peer(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<Box<HttpPeer>> {
+            let host_name = session.req_header().headers.get("host").unwrap();
+            let addr = self.get_host(host_name.to_str().unwrap()).unwrap();
+            info!("connecting to {addr:?}");
+            let peer = Box::new(HttpPeer::new(addr, false, "".to_string()));
+            Ok(peer)
+        }
+    */
+    async fn request_filter(&self, _session: &mut Session, _ctx: &mut Self::CTX) -> pingora_core::Result<bool>
+    where
+        Self::CTX: Send + Sync,
+    {
+        // if !_session.req_header().uri.path().starts_with("/ping") {
+        if _session.req_header().uri.path().starts_with("/denied") {
+            let _ = _session.respond_error(403).await;
+            info!("Forbidded: {:?}, {}", _session.client_addr(), _session.req_header().uri.path().to_string());
+            return Ok(true);
+        };
+        Ok(false)
+    }
+    async fn upstream_request_filter(&self, _session: &mut Session, _upstream_request: &mut RequestHeader, _ctx: &mut Self::CTX) -> pingora_core::Result<()>
+    where
+        Self::CTX: Send + Sync,
+    {
+        let clientip = _session.client_addr();
+        match clientip {
+            Some(ip) => {
+                let inet = ip.as_inet();
+                match inet {
+                    Some(addr) => {
+                        _upstream_request
+                            .insert_header("X-Forwarded-For", addr.to_string().split(':').collect::<Vec<&str>>()[0])
+                            .unwrap();
+                    }
+                    None => warn!("Malformed Client IP: {:?}", inet),
+                }
+            }
+            None => {
+                warn!("Cannot detect client IP");
+            }
+        }
+        Ok(())
+    }
+    async fn response_filter(&self, _session: &mut Session, _upstream_response: &mut ResponseHeader, _ctx: &mut Self::CTX) -> pingora_core::Result<()>
+    where
+        Self::CTX: Send + Sync,
+    {
+        _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));
+    // }
+}
+
+/*
+pub struct SD;
+
+#[async_trait]
+impl ServiceDiscovery for SD {
+    async fn discover(&self) -> Result<(BTreeSet<Backend>, HashMap<u64, bool>)> {
+        let addrs = read_upstreams_from_file()?;
+
+        /*
+        let addr2 = read_upstreams_from_file_d()?;
+        let mut hmn = HashMap::new();
+        for (k, v) in addr2 {
+            let vl = v
+                .into_iter()
+                .map(|addr| Backend {
+                    addr,
+                    weight: 1,
+                    ext: Default::default(),
+                })
+                .collect::<BTreeSet<_>>();
+            hmn.insert(k, vl);
+        }
+        println!("{:?}", hmn);
+        println!("================== {:?}", addrs);
+        */
+
+        let backends = addrs
+            .into_iter()
+            .map(|addr| Backend {
+                addr,
+                weight: 1,
+                ext: Default::default(),
+            })
+            .collect::<BTreeSet<_>>();
+        Ok((backends, HashMap::new()))
+    }
+}
+
+fn read_upstreams_from_file() -> Result<Vec<SocketAddr>> {
+    let contents = std::fs::read_to_string("upstreams.txt").map_err(|e| Error::because(ErrorType::InternalError, "reading upstreams file", e))?;
+    let addrs = contents
+        .lines()
+        .filter(|line| !line.trim().is_empty())
+        .map(|line| {
+            line.trim()
+                .parse::<SocketAddr>()
+                .map_err(|e| Error::because(ErrorType::InternalError, "parsing upstream address", e))
+        })
+        .collect::<Result<Vec<_>>>()?;
+    Ok(addrs)
+}
+*/
--- a/src/web/run.rs
+++ b/src/web/run.rs
@@ -1,46 +0,0 @@
-use bytes::Bytes;
-use mini_redis::{Connection, Frame};
-use std::collections::HashMap;
-use std::sync::{Arc, Mutex};
-use tokio::net::{TcpListener, TcpStream};
-
-type Db = Arc<Mutex<HashMap<String, Bytes>>>;
-#[tokio::main]
-pub async fn run() {
-    println!("\n= = = = = = = = ASYNC TOKIO = = = = = = = = =\n");
-    let listener = TcpListener::bind("0.0.0.0:6379").await.unwrap();
-    println!("Server is running on 0.0.0.0:6379 !\n");
-    let db = Arc::new(Mutex::new(HashMap::new()));
-    loop {
-        let (socket, _) = listener.accept().await.unwrap();
-        let db = db.clone();
-        tokio::spawn(async move {
-            process(socket, db).await;
-        });
-    }
-}
-async fn process(socket: TcpStream, db: Db) {
-    use mini_redis::Command::{self, Get, Set};
-    let mut connection = Connection::new(socket);
-    while let Some(frame) = connection.read_frame().await.unwrap() {
-        let response = match Command::from_frame(frame).unwrap() {
-            Set(cmd) => {
-                // println!("{:?}", Command::from_frame(frame).unwrap());
-                let mut db = db.lock().unwrap();
-                db.insert(cmd.key().to_string(), cmd.value().clone());
-                Frame::Simple("OK".to_string())
-            }
-            Get(cmd) => {
-                let db = db.lock().unwrap();
-                if let Some(value) = db.get(cmd.key()) {
-                    println!("{:?}", db);
-                    Frame::Bulk(value.clone())
-                } else {
-                    Frame::Null
-                }
-            }
-            cmd => panic!("unimplemented {:?}", cmd),
-        };
-        connection.write_frame(&response).await.unwrap();
-    }
-}
--- a/src/web/start.rs
+++ b/src/web/start.rs
@@ -0,0 +1,66 @@
+use crate::web::proxyhttp::{GetHost, LB};
+use dashmap::DashMap;
+use pingora_core::server::Server;
+use std::sync::atomic::AtomicUsize;
+/*
+pub fn run1() {
+    let mut upstreams = LoadBalancer::try_from_iter(vec!["192.168.1.10:8000", "192.168.1.1:8000", "127.0.0.1:8000"]).unwrap();
+    env_logger::init();
+    let hc = TcpHealthCheck::new();
+    upstreams.set_health_check(hc);
+    upstreams.health_check_frequency = Some(Duration::from_secs(1));
+
+    let background = background_service("health check", upstreams);
+    let upstreams = background.task();
+    let mut server = Server::new(None).unwrap();
+    server.bootstrap();
+
+    let mut proxy = pingora_proxy::http_proxy_service(&server.configuration, proxyhttp::LB(upstreams));
+
+    proxy.add_tcp("0.0.0.0:6193");
+    server.add_service(background);
+    server.add_service(proxy);
+    server.run_forever();
+}
+*/
+
+pub fn run() {
+    env_logger::init();
+
+    let mut server = Server::new(None).unwrap();
+    server.bootstrap();
+
+    // let backends = Backends::new(Box::new(SD));
+    // let load_balancer = LoadBalancer::from_backends(backends);
+
+    // load_balancer.set_health_check(TcpHealthCheck::new());
+    // load_balancer.health_check_frequency = Some(Duration::from_secs(1));
+    // load_balancer.update_frequency = Some(Duration::from_secs(1));
+
+    // let background = background_service("load balancer", load_balancer);
+
+    let upstreams_map: DashMap<String, (Vec<(String, u16)>, AtomicUsize)> = DashMap::new();
+
+    let mut ll = LB {
+        upstreams_map,
+        // upstreams_maps: DashMap::new(),
+    };
+    add_hosts(&mut ll);
+
+    let mut lb = pingora_proxy::http_proxy_service(&server.configuration, ll);
+
+    lb.add_tcp("0.0.0.0:6193");
+    server.add_service(lb);
+    // server.add_service(background);
+
+    server.run_forever();
+}
+
+fn add_hosts(lb: &mut LB) {
+    lb.set_host("myip.netangels.net", "192.168.1.1", 8000);
+    lb.set_host("myip.netangels.net", "127.0.0.1", 8000);
+    lb.set_host("myip.netangels.net", "127.0.0.2", 8000);
+    lb.set_host("polo.netangels.net", "192.168.1.1", 8000);
+    lb.set_host("polo.netangels.net", "192.168.1.10", 8000);
+    lb.set_host("glop.netangels.net", "192.168.1.20", 8000);
+}