Add Global connection counter

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "layer4-proxy"
-version = "0.5.1"
+version = "0.6.1"
 edition = "2021"
 authors = ["Aleksandar Lazic <[email protected]>","Jacob Kiers <[email protected]>"]
 license = "Apache-2.0"

container-files/etc/l4p/config.yaml

@@ -17,7 +17,7 @@ servers:
       www.test1.com: proxy-via
     default: echo
     # note yet implemented
-    maxclients: 4
+    maxclients: 2
     via:
       *viaanchor
 

src/config/config_v1.rs

@@ -137,7 +137,6 @@ fn load_config(path: &str) -> Result<ParsedConfigV1, ConfigError> {
         env_logger::builder().default_format().init();
         //pretty_env_logger::init_custom_env("FOURTH_LOG");
         //pretty_env_logger::init_timed();
-
     }
 
     info!("Using config file: {}", &path);
@@ -154,7 +153,7 @@ fn load_config(path: &str) -> Result<ParsedConfigV1, ConfigError> {
         let ups = upstream.as_str().try_into()?;
         parsed_upstream.insert(name.to_string(), Upstream::Proxy(ups));
     }
-    
+
     let via: ViaUpstream = base.via.clone();
     debug!("via {:?}", via);
 

src/main.rs

@@ -6,7 +6,9 @@ use crate::config::ConfigV1;
 use crate::servers::Server;
 
 use log::{debug, error, info};
-use std::path::PathBuf;
+use std::{path::PathBuf, sync::atomic::AtomicUsize};
+
+static GLOBAL_THREAD_COUNT: AtomicUsize = AtomicUsize::new(0);
 
 fn main() {
     let config_path = match find_config() {
@@ -27,6 +29,8 @@ fn main() {
             std::process::exit(1);
         }
     };
+
+    debug!("GLOBAL_THREAD_COUNT :{:?}:", GLOBAL_THREAD_COUNT);
     debug!("{:?}", config);
 
     let mut server = Server::new_from_v1_config(config.base);

src/servers/mod.rs

@@ -2,7 +2,6 @@ use log::{debug, error, info};
 use std::collections::{HashMap, HashSet};
 use std::net::SocketAddr;
 use std::sync::Arc;
-use tokio::runtime::Handle;
 use tokio::sync::Semaphore;
 
 use tokio::signal::unix::{signal, SignalKind};
@@ -33,8 +32,8 @@ pub(crate) struct Proxy {
     pub default_action: String,
     pub upstream: HashMap<String, Upstream>,
     pub via: ViaUpstream,
-    //pub maxclients: Arc<Semaphore<>>,
-    pub maxclients: usize,
+    pub maxclients: Arc<Semaphore>,
+    //pub maxclients: usize,
 }
 
 impl Server {
@@ -76,8 +75,8 @@ impl Server {
                     default_action: default.clone(),
                     upstream: upstream.clone(),
                     via: proxy.via.clone(),
-                    //maxclients: Arc::new(Semaphore::new(proxy.maxclients)),
-                    maxclients: proxy.maxclients,
+                    maxclients: Arc::new(Semaphore::new(proxy.maxclients)),
+                    //maxclients: proxy.maxclients,
                 };
                 new_server.proxies.push(Arc::new(proxy));
             }

src/servers/protocol/tcp.rs

@@ -1,7 +1,9 @@
 use crate::servers::protocol::tls::get_sni;
 use crate::servers::Proxy;
+use crate::GLOBAL_THREAD_COUNT;
 use log::{debug, error, info, warn};
 use std::error::Error;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use tokio::net::{TcpListener, TcpStream};
 
@@ -16,8 +18,6 @@ pub(crate) async fn proxy(config: Arc<Proxy>) -> Result<(), Box<dyn Error>> {
 
     loop {
         let thread_proxy = config.clone();
-        //let permit = config.maxclients.clone().acquire_owned().await.unwrap();
-        //debug!("permit.num_permits {:?}",permit.num_permits());
         match listener.accept().await {
             Err(err) => {
                 error!("Failed to accept connection: {}", err);
@@ -40,10 +40,19 @@ pub(crate) async fn proxy(config: Arc<Proxy>) -> Result<(), Box<dyn Error>> {
 }
 
 async fn accept(inbound: TcpStream, proxy: Arc<Proxy>) -> Result<(), Box<dyn Error>> {
+    //let permit = proxy.maxclients.clone().acquire_owned().await.unwrap();
+    //info!("permit.num_permits {:?}", permit.num_permits());
+
     if proxy.default_action.contains("health") {
         debug!("Health check request")
     } else {
-        info!("New connection from {:?}", inbound.peer_addr()?);
+        let old = GLOBAL_THREAD_COUNT.fetch_add(1, Ordering::SeqCst);
+        info!(
+            "New connection from {:?} , num :{:?}: Current Connections :{:?}",
+            inbound.peer_addr()?,
+            old + 1,
+            GLOBAL_THREAD_COUNT
+        );
     }
 
     let upstream_name = match proxy.tls {
@@ -88,10 +97,21 @@ async fn accept(inbound: TcpStream, proxy: Arc<Proxy>) -> Result<(), Box<dyn Err
 
     match upstream.process(inbound, proxy.clone()).await {
         Ok(_) => {
-            info!("Connection closed for {:?}", upstream_name);
+            let old = GLOBAL_THREAD_COUNT.fetch_sub(1, Ordering::SeqCst);
+            info!(
+                "Connection closed for {:?}, num :{:?}: Current Connections :{:?}",
+                upstream_name, old, GLOBAL_THREAD_COUNT
+            );
+            //drop(permit);
             Ok(())
         }
         Err(e) => {
+            let old = GLOBAL_THREAD_COUNT.fetch_sub(1, Ordering::SeqCst);
+            info!(
+                "Connection closed for {:?}, num :{:?}: Current Connections :{:?}",
+                upstream_name, old, GLOBAL_THREAD_COUNT
+            );
+            //drop(permit);
             error!("my error {:?}", e);
             Ok(())
         }

src/upstreams/mod.rs

@@ -1,21 +1,20 @@
 mod proxy_to_upstream;
 
 use crate::servers::Proxy;
+use http_body_util::Full;
+use hyper::body::Bytes;
 use hyper::server::conn::http1;
 use hyper::service::service_fn;
-use hyper_util::rt::TokioIo;
 use hyper::{Request, Response};
-use http_body_util::Full;
-use hyper::body::Bytes;
+use hyper_util::rt::TokioIo;
 use log::debug;
 use serde::Deserialize;
+use std::convert::Infallible;
 use std::error::Error;
 use std::sync::Arc;
 use tokio::io;
 use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
 use tokio::net::TcpStream;
-use std::convert::Infallible;
-
 
 pub use crate::upstreams::proxy_to_upstream::ProxyToUpstream;