fix: resolve clippy warnings and improve code quality

Signed-off-by: Eeshu-Yadav <[email protected]>

Author: Eeshu-Yadav

orion-configuration/src/config/cluster.rs

@@ -155,7 +155,7 @@ impl From<LbEndpointVecDeser> for Vec<LbEndpoint> {
                     Address::Internal(internal_addr) => Some(LbEndpoint {
                         address: EndpointAddress::Internal(InternalEndpointAddress {
                             server_listener_name: internal_addr.server_listener_name.into(),
-                            endpoint_id: internal_addr.endpoint_id.map(|id| id.into()),
+                            endpoint_id: internal_addr.endpoint_id.map(std::convert::Into::into),
                         }),
                         health_status: HealthStatus::default(),
                         load_balancing_weight: NonZeroU32::MIN,
@@ -193,7 +193,7 @@ impl EndpointAddress {
     pub fn into_addr(self) -> Result<SocketAddr, String> {
         match self {
             EndpointAddress::Socket(addr) => Ok(addr),
-            EndpointAddress::Internal(_) => Err("Cannot convert internal address to socket address".to_string()),
+            EndpointAddress::Internal(_) => Err("Cannot convert internal address to socket address".to_owned()),
         }
     }
 }
@@ -785,7 +785,7 @@ mod envoy_conversions {
                         Address::Socket(socket_addr) => Ok(EndpointAddress::Socket(socket_addr)),
                         Address::Internal(internal_addr) => Ok(EndpointAddress::Internal(InternalEndpointAddress {
                             server_listener_name: internal_addr.server_listener_name.into(),
-                            endpoint_id: internal_addr.endpoint_id.map(|id| id.into()),
+                            endpoint_id: internal_addr.endpoint_id.map(std::convert::Into::into),
                         })),
                         Address::Pipe(_, _) => {
                             Err(GenericError::unsupported_variant("Pipe addresses are not supported for endpoints"))

orion-configuration/src/config/core.rs

@@ -305,7 +305,7 @@ pub mod envoy_conversions {
     impl std::fmt::Display for Address {
         fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
             match self {
-                Self::Socket(addr) => write!(f, "{}", addr),
+                Self::Socket(addr) => write!(f, "{addr}"),
                 Self::Internal(internal) => write!(f, "internal:{}", internal.server_listener_name),
                 Self::Pipe(path, _) => f.write_str(path),
             }

orion-configuration/src/config/runtime.rs

@@ -144,7 +144,7 @@ fn get_cgroup_v2_cpu_limit() -> crate::Result<usize> {
 }
 
 fn parse_cgroup_v2_cpu_max(content: &str) -> crate::Result<usize> {
-    let parts: Vec<&str> = content.trim().split_whitespace().collect();
+    let parts: Vec<&str> = content.split_whitespace().collect();
     if parts.len() == 2 && parts[0] != "max" {
         let quota: i64 = parts[0].parse()?;
         let period: i64 = parts[1].parse()?;
@@ -160,8 +160,8 @@ fn parse_cgroup_v2_cpu_max(content: &str) -> crate::Result<usize> {
 
 fn get_cgroup_v1_cpu_limit() -> crate::Result<usize> {
     let cgroup_path = get_cgroup_v1_cpu_path()?;
-    let quota_path = format!("{}/cpu.cfs_quota_us", cgroup_path);
-    let period_path = format!("{}/cpu.cfs_period_us", cgroup_path);
+    let quota_path = format!("{cgroup_path}/cpu.cfs_quota_us");
+    let period_path = format!("{cgroup_path}/cpu.cfs_period_us");
 
     let quota_content = std::fs::read_to_string(&quota_path)?;
     let period_content = std::fs::read_to_string(&period_path)?;
@@ -193,7 +193,7 @@ fn parse_cgroup_v1_cpu_path(cgroup_content: &str) -> crate::Result<String> {
         let mut parts = line.split(':');
         if let (Some(_), Some(controllers), Some(path)) = (parts.next(), parts.next(), parts.next()) {
             if controllers.split(',').any(|c| c == "cpu") {
-                return Some(format!("/sys/fs/cgroup/cpu{}", path));
+                return Some(format!("/sys/fs/cgroup/cpu{path}"));
             }
         }
         None
@@ -202,7 +202,7 @@ fn parse_cgroup_v1_cpu_path(cgroup_content: &str) -> crate::Result<String> {
     }
 
     if std::path::Path::new("/sys/fs/cgroup/cpu").exists() {
-        Ok("/sys/fs/cgroup/cpu".to_string())
+        Ok("/sys/fs/cgroup/cpu".to_owned())
     } else {
         Err("CPU cgroup path not found".into())
     }

orion-lib/src/clusters/load_assignment.rs

@@ -80,7 +80,7 @@ impl EndpointAddressType {
                 if let Ok(socket_addr) = addr_str.parse::<std::net::SocketAddr>() {
                     EndpointAddress::Socket(socket_addr)
                 } else {
-                    panic!("Cannot convert authority back to socket address: {}", addr_str);
+                    panic!("Cannot convert authority back to socket address: {addr_str}");
                 }
             },
             EndpointAddressType::Internal(internal_addr, _) => EndpointAddress::Internal(internal_addr.clone()),

orion-lib/src/listeners/filter_state.rs

@@ -45,8 +45,8 @@ pub enum DownstreamConnectionMetadata {
         proxy_local_address: SocketAddr,
     },
     /// Internal connections from in-process communication (e.g., waypoint proxy)
-    /// - listener_name: identifies which internal listener accepted this connection
-    /// - endpoint_id: optional identifier for the specific endpoint (used for load balancing)
+    /// - `listener_name`: identifies which internal listener accepted this connection
+    /// - `endpoint_id`: optional identifier for the specific endpoint (used for load balancing)
     FromInternal {
         listener_name: String,
         endpoint_id: Option<String>,

orion-lib/src/listeners/http_connection_manager.rs

@@ -1024,7 +1024,7 @@ impl Service<ExtendedRequest<Incoming>> for HttpRequestHandler {
                     add,
                     1,
                     trans_handler.thread_id(),
-                    &[KeyValue::new("listener", listener_name_for_route.to_string())]
+                    &[KeyValue::new("listener", listener_name_for_route.to_owned())]
                 );
 
                 if let Some(state) = trans_handler.span_state.as_ref() {
@@ -1049,7 +1049,7 @@ impl Service<ExtendedRequest<Incoming>> for HttpRequestHandler {
                             add,
                             nbytes + resp_head_size as u64,
                             trans_handler.thread_id(),
-                            &[KeyValue::new("listener", listener_name_for_response.to_string())]
+                            &[KeyValue::new("listener", listener_name_for_response.to_owned())]
                         );
 
                         let is_transaction_complete = if let Some(ctx) = trans_handler.access_log_ctx.as_ref() {

orion-lib/src/listeners/http_connection_manager/route.rs

@@ -194,7 +194,7 @@ impl<'a> RequestHandler<(MatchedRequest<'a>, &HttpConnectionManager)> for &Route
                         let err = err.into_inner();
                         let event_error = EventError::try_infer_from(&err);
                         let flags = event_error.clone().map(ResponseFlags::from).unwrap_or_default();
-                        let event_kind = event_error.map_or(EventKind::ViaUpstream, |e| EventKind::Error(e));
+                        let event_kind = event_error.map_or(EventKind::ViaUpstream, EventKind::Error);
                         debug!(
                             "HttpConnectionManager Error processing response {:?}: {}({})",
                             err,
@@ -211,7 +211,7 @@ impl<'a> RequestHandler<(MatchedRequest<'a>, &HttpConnectionManager)> for &Route
                 let err = err.into_inner();
                 let event_error = EventError::try_infer_from(&err);
                 let flags = event_error.clone().map(ResponseFlags::from).unwrap_or_default();
-                let event_kind = event_error.map_or(EventKind::ViaUpstream, |e| EventKind::Error(e));
+                let event_kind = event_error.map_or(EventKind::ViaUpstream, EventKind::Error);
                 debug!(
                     "Failed to get an HTTP connection: {:?}: {}({})",
                     err,

orion-lib/src/listeners/listener.rs

@@ -57,44 +57,6 @@ use tokio::{
 };
 use tracing::{debug, info, warn};
 
-async fn handle_internal_connection_static(
-    listener_name: String,
-    connection_pair: crate::transport::InternalConnectionPair,
-    filter_chains: Arc<HashMap<FilterChainMatch, FilterchainType>>,
-) -> Result<()> {
-    use crate::listeners::filter_state::DownstreamConnectionMetadata;
-
-    debug!("Handling new internal connection for listener '{}'", listener_name);
-
-    let downstream_metadata = DownstreamConnectionMetadata::FromInternal {
-        listener_name: listener_name.clone(),
-        endpoint_id: connection_pair.downstream.metadata().endpoint_id.clone(),
-    };
-
-    let filter_chain = match Listener::select_filterchain(&filter_chains, &downstream_metadata, None)? {
-        Some(fc) => fc,
-        None => {
-            warn!("No matching filter chain found for internal connection");
-            return Err(crate::Error::new("No matching filter chain"));
-        },
-    };
-
-    let _downstream_stream = connection_pair.downstream;
-
-    match &filter_chain.handler {
-        crate::listeners::filterchain::ConnectionHandler::Http(_http_manager) => {
-            info!("Processing internal connection through HTTP filter chain");
-            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-            Ok(())
-        },
-        crate::listeners::filterchain::ConnectionHandler::Tcp(_tcp_proxy) => {
-            info!("Processing internal connection through TCP filter chain");
-            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-            Ok(())
-        },
-    }
-}
-
 #[derive(Debug)]
 struct InternalConnectionWorkerPool {
     senders: Vec<mpsc::UnboundedSender<InternalConnectionTask>>,
@@ -402,7 +364,7 @@ impl Listener {
                 maybe_route_update = route_updates_receiver.recv() => {
                     //todo: add context to the error here once orion-error lands
                     match maybe_route_update {
-                        Ok(route_update) => {Self::process_route_update(&name, &filter_chains, route_update)},
+                        Ok(route_update) => {Self::process_route_update(name, &filter_chains, route_update)},
                         Err(e) => {return e.into();}
                     }
                 },
@@ -411,7 +373,7 @@ impl Listener {
                         Ok(secret_update) => {
                             // todo: possibly expensive clone - may need to rethink this structure
                             let mut filter_chains_clone = filter_chains.as_ref().clone();
-                            Self::process_secret_update(&name, &mut filter_chains_clone, secret_update);
+                            Self::process_secret_update(name, &mut filter_chains_clone, secret_update);
                             filter_chains = Arc::new(filter_chains_clone);
                         }
                         Err(e) => {return e.into();}
@@ -436,7 +398,7 @@ impl Listener {
         let filter_chains = Arc::new(filter_chains);
         let factory = global_internal_connection_factory();
 
-        let (_handle, mut connection_receiver, _listener_ref) = match factory.register_listener(name.to_string()).await
+        let (_handle, mut connection_receiver, _listener_ref) = match factory.register_listener(name.to_owned()).await
         {
             Ok(result) => result,
             Err(e) => {
@@ -450,35 +412,32 @@ impl Listener {
         loop {
             tokio::select! {
                 maybe_connection = connection_receiver.recv() => {
-                    match maybe_connection {
-                        Some(connection_pair) => {
-                            debug!("Internal listener '{}' received new connection", name);
-
-                            let filter_chains_clone = filter_chains.clone();
-                            let listener_name = name.to_string();
-
-                            // Use worker pool instead of tokio::spawn for better performance
-                            // with large numbers of short connections
-                            let task = InternalConnectionTask {
-                                listener_name,
-                                connection_pair,
-                                filter_chains: filter_chains_clone,
-                            };
-
-                            if let Err(e) = get_internal_worker_pool().submit_task(task) {
-                                warn!("Failed to submit internal connection task: {}", e);
-                            }
-                        }
-                        None => {
-                            warn!("Internal listener '{}' connection channel closed", name);
-                            break;
+                    if let Some(connection_pair) = maybe_connection {
+                        debug!("Internal listener '{}' received new connection", name);
+
+                        let filter_chains_clone = filter_chains.clone();
+                        let listener_name = name.to_owned();
+
+                        // Use worker pool instead of tokio::spawn for better performance
+                        // with large numbers of short connections
+                        let task = InternalConnectionTask {
+                            listener_name,
+                            connection_pair,
+                            filter_chains: filter_chains_clone,
+                        };
+
+                        if let Err(e) = get_internal_worker_pool().submit_task(task) {
+                            warn!("Failed to submit internal connection task: {}", e);
                         }
+                    } else {
+                        warn!("Internal listener '{}' connection channel closed", name);
+                        break;
                     }
                 },
                 maybe_route_update = route_updates_receiver.recv() => {
                     match maybe_route_update {
                         Ok(route_update) => {
-                            Self::process_route_update(&name, &filter_chains, route_update);
+                            Self::process_route_update(name, &filter_chains, route_update);
                         }
                         Err(e) => {
                             error!("Route update error for internal listener '{}': {}", name, e);
@@ -490,7 +449,7 @@ impl Listener {
                     match maybe_secret_update {
                         Ok(secret_update) => {
                             let mut filter_chains_clone = filter_chains.as_ref().clone();
-                            Self::process_secret_update(&name, &mut filter_chains_clone, secret_update);
+                            Self::process_secret_update(name, &mut filter_chains_clone, secret_update);
                             // TODO: Update the shared filter chains state for active connections
                         }
                         Err(e) => {
@@ -616,8 +575,8 @@ impl Listener {
 
         let ssl = AtomicBool::new(false);
         defer! {
-                            with_metric!(listeners::DOWNSTREAM_CX_DESTROY, add, 1, shard_id, &[KeyValue::new("listener", listener_name.to_string())]);
-            with_metric!(listeners::DOWNSTREAM_CX_ACTIVE, sub, 1, shard_id, &[KeyValue::new("listener", listener_name.to_string())]);
+                            with_metric!(listeners::DOWNSTREAM_CX_DESTROY, add, 1, shard_id, &[KeyValue::new("listener", listener_name.to_owned())]);
+            with_metric!(listeners::DOWNSTREAM_CX_ACTIVE, sub, 1, shard_id, &[KeyValue::new("listener", listener_name.to_owned())]);
             if ssl.load(Ordering::Relaxed) {
                 with_metric!(http::DOWNSTREAM_CX_SSL_ACTIVE, add, 1, shard_id, &[KeyValue::new("listener", listener_name)]);
             }
@@ -656,7 +615,7 @@ impl Listener {
                         add,
                         1,
                         shard_id,
-                        &[KeyValue::new("listener", listener_name.to_string())]
+                        &[KeyValue::new("listener", listener_name.to_owned())]
                     );
                     with_metric!(
                         http::DOWNSTREAM_CX_SSL_ACTIVE,
@@ -1014,3 +973,38 @@ filter_chains:
         assert_eq!(Listener::select_filterchain(&m, &metadata, Some("hello.world")).unwrap().copied(), Some(3));
     }
 }
+
+async fn handle_internal_connection_static(
+    listener_name: String,
+    connection_pair: crate::transport::InternalConnectionPair,
+    filter_chains: Arc<HashMap<FilterChainMatch, FilterchainType>>,
+) -> Result<()> {
+    use crate::listeners::filter_state::DownstreamConnectionMetadata;
+
+    debug!("Handling new internal connection for listener '{}'", listener_name);
+
+    let downstream_metadata = DownstreamConnectionMetadata::FromInternal {
+        listener_name: listener_name.clone(),
+        endpoint_id: connection_pair.downstream.metadata().endpoint_id.clone(),
+    };
+
+    let filter_chain = if let Some(fc) = Listener::select_filterchain(&filter_chains, &downstream_metadata, None)? { fc } else {
+        warn!("No matching filter chain found for internal connection");
+        return Err(crate::Error::new("No matching filter chain"));
+    };
+
+    let _downstream_stream = connection_pair.downstream;
+
+    match &filter_chain.handler {
+        crate::listeners::filterchain::ConnectionHandler::Http(_http_manager) => {
+            info!("Processing internal connection through HTTP filter chain");
+            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+            Ok(())
+        },
+        crate::listeners::filterchain::ConnectionHandler::Tcp(_tcp_proxy) => {
+            info!("Processing internal connection through TCP filter chain");
+            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+            Ok(())
+        },
+    }
+}

orion-lib/src/listeners/listeners_manager.rs

@@ -119,7 +119,7 @@ impl ListenersManager {
                         warn!("Internal problem when updating a route: {e}");
                     }
                 },
-                _ = ct.cancelled() => {
+                () = ct.cancelled() => {
                     warn!("Listener manager exiting");
                     return Ok(());
                 }
@@ -128,7 +128,7 @@ impl ListenersManager {
     }
 
     pub fn start_listener(&mut self, listener: Listener, listener_conf: ListenerConfig) -> Result<()> {
-        let listener_name = listener.get_name().to_string();
+        let listener_name = listener.get_name().to_owned();
         if let Some((addr, dev)) = listener.get_socket() {
             info!("Listener {} at {addr} (device bind:{})", listener_name, dev.is_some());
         } else {
@@ -148,7 +148,7 @@ impl ListenersManager {
         let listener_info = ListenerInfo::new(join_handle, listener_conf, version);
         self.listener_handles.insert(listener_name.clone(), listener_info);
 
-        let version_count = self.listener_handles.get_vec(&listener_name).map(|v| v.len()).unwrap_or(0);
+        let version_count = self.listener_handles.get_vec(&listener_name).map(std::vec::Vec::len).unwrap_or(0);
         info!("Started version {} of listener {} ({} total active version(s))", version, listener_name, version_count);
 
         Ok(())

orion-lib/src/transport/internal_cluster_connector.rs

@@ -83,8 +83,8 @@ impl InternalChannelConnector {
         Ok(InternalChannel {
             stream,
             cluster_name: self.cluster_name,
-            listener_name: self.connector.listener_name().to_string(),
-            endpoint_id: self.connector.endpoint_id().map(|s| s.to_string()),
+            listener_name: self.connector.listener_name().to_owned(),
+            endpoint_id: self.connector.endpoint_id().map(std::borrow::ToOwned::to_owned),
         })
     }
 
@@ -115,7 +115,7 @@ impl InternalChannel {
 }
 
 pub mod cluster_helpers {
-    use super::*;
+    use super::{InternalChannelConnector, global_internal_connection_factory};
     use orion_configuration::config::cluster::InternalEndpointAddress;
 
     pub fn create_internal_connector(
@@ -125,7 +125,7 @@ pub mod cluster_helpers {
         InternalChannelConnector::new(
             internal_addr.server_listener_name.to_string(),
             cluster_name,
-            internal_addr.endpoint_id.as_ref().map(|s| s.to_string()),
+            internal_addr.endpoint_id.as_ref().map(std::string::ToString::to_string),
         )
     }