feat: add HTTP Connection Manager drain_timeout support

- Add drain_timeout field with 5000ms default (Envoy compliant)
- Integrate with drain signaling and timeout enforcement
- Support protocol-specific draining (HTTP/1.1, HTTP/2, TCP)
- Add comprehensive test coverage for drain behaviors

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

Author: Eeshu-Yadav

Cargo.lock

@@ -37,10 +37,20 @@ use std::{
 
 use orion_interner::StringInterner;
 
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
+#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
+pub enum DrainType {
+    #[default]
+    Default,
+    ModifyOnly,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct Listener {
     pub name: CompactString,
     pub address: SocketAddr,
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub version_info: Option<String>,
     #[serde(with = "serde_filterchains")]
     pub filter_chains: HashMap<FilterChainMatch, FilterChain>,
     #[serde(skip_serializing_if = "Option::is_none", default = "Default::default")]
@@ -53,6 +63,8 @@ pub struct Listener {
     pub with_tlv_listener_filter: bool,
     #[serde(skip_serializing_if = "Option::is_none", default = "Default::default")]
     pub tlv_listener_filter_config: Option<super::listener_filters::TlvListenerFilterConfig>,
+    #[serde(default)]
+    pub drain_type: DrainType,
 }
 
 impl Listener {
@@ -320,7 +332,7 @@ mod envoy_conversions {
     use std::hash::{DefaultHasher, Hash, Hasher};
     use std::str::FromStr;
 
-    use super::{FilterChain, FilterChainMatch, Listener, MainFilter, ServerNameMatch, TlsConfig};
+    use super::{DrainType, FilterChain, FilterChainMatch, Listener, MainFilter, ServerNameMatch, TlsConfig};
     use crate::config::{
         common::*,
         core::{Address, CidrRange},
@@ -401,7 +413,7 @@ mod envoy_conversions {
                 per_connection_buffer_limit_bytes,
                 metadata,
                 deprecated_v1,
-                drain_type,
+                // drain_type,
                 // listener_filters,
                 listener_filters_timeout,
                 continue_on_listener_filters_timeout,
@@ -476,6 +488,7 @@ mod envoy_conversions {
                         .with_node("socket_options");
                 }
                 let bind_device = bind_device.into_iter().next();
+                let drain_type = DrainType::try_from(drain_type).unwrap_or_default();
                 Ok(Self {
                     name,
                     address,
@@ -485,12 +498,26 @@ mod envoy_conversions {
                     proxy_protocol_config,
                     with_tlv_listener_filter,
                     tlv_listener_filter_config,
+                    drain_type,
+                    version_info: None,
                 })
             }())
             .with_name(name)
         }
     }
 
+    impl TryFrom<i32> for DrainType {
+        type Error = GenericError;
+
+        fn try_from(value: i32) -> Result<Self, Self::Error> {
+            match value {
+                0 => Ok(DrainType::Default),
+                1 => Ok(DrainType::ModifyOnly),
+                _ => Err(GenericError::from_msg(format!("Unknown drain type: {}", value))),
+            }
+        }
+    }
+
     struct FilterChainWrapper((FilterChainMatch, FilterChain));
 
     impl TryFrom<EnvoyFilterChain> for FilterChainWrapper {

orion-configuration/src/config/listener.rs

@@ -51,6 +51,9 @@ pub struct HttpConnectionManager {
     #[serde(with = "humantime_serde")]
     #[serde(skip_serializing_if = "Option::is_none", default)]
     pub request_timeout: Option<Duration>,
+    #[serde(with = "humantime_serde")]
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub drain_timeout: Option<Duration>,
     #[serde(skip_serializing_if = "Vec::is_empty", default)]
     pub http_filters: Vec<HttpFilter>,
     #[serde(skip_serializing_if = "Vec::is_empty", default)]
@@ -564,6 +567,61 @@ mod tests {
         assert!(MatchHostScoreLPM::Wildcard < MatchHostScoreLPM::Suffix("foo.bar.test.com".len()));
         assert!(MatchHostScoreLPM::Wildcard == MatchHostScoreLPM::Wildcard);
     }
+
+    #[test]
+    fn test_drain_timeout_configuration() {
+        let config = HttpConnectionManager {
+            codec_type: CodecType::Auto,
+            route_specifier: RouteSpecifier::RouteConfig(RouteConfiguration {
+                name: "test_route".into(),
+                most_specific_header_mutations_wins: false,
+                response_header_modifier: Default::default(),
+                request_headers_to_add: vec![],
+                request_headers_to_remove: vec![],
+                virtual_hosts: vec![],
+            }),
+            http_filters: vec![],
+            enabled_upgrades: vec![],
+            access_log: vec![],
+            xff_settings: Default::default(),
+            generate_request_id: false,
+            preserve_external_request_id: false,
+            always_set_request_id_in_response: false,
+            tracing: None,
+            request_timeout: Some(Duration::from_secs(30)),
+            drain_timeout: Some(Duration::from_secs(10)),
+        };
+
+        assert_eq!(config.drain_timeout, Some(Duration::from_secs(10)));
+        assert_eq!(config.request_timeout, Some(Duration::from_secs(30)));
+    }
+
+    #[test]
+    fn test_drain_timeout_default() {
+        let config = HttpConnectionManager {
+            codec_type: CodecType::Http1,
+            route_specifier: RouteSpecifier::RouteConfig(RouteConfiguration {
+                name: "test_route_default".into(),
+                most_specific_header_mutations_wins: false,
+                response_header_modifier: Default::default(),
+                request_headers_to_add: vec![],
+                request_headers_to_remove: vec![],
+                virtual_hosts: vec![],
+            }),
+            http_filters: vec![],
+            enabled_upgrades: vec![],
+            access_log: vec![],
+            xff_settings: Default::default(),
+            generate_request_id: false,
+            preserve_external_request_id: false,
+            always_set_request_id_in_response: false,
+            tracing: None,
+            request_timeout: None,
+            drain_timeout: None,
+        };
+
+        assert_eq!(config.drain_timeout, None);
+    }
 }
 
 #[cfg(feature = "envoy-conversions")]
@@ -702,7 +760,7 @@ mod envoy_conversions {
                 stream_idle_timeout,
                 // request_timeout,
                 request_headers_timeout,
-                drain_timeout,
+                // drain_timeout,
                 delayed_close_timeout,
                 // access_log,
                 access_log_flush_interval,
@@ -753,6 +811,11 @@ mod envoy_conversions {
                 .transpose()
                 .map_err(|_| GenericError::from_msg("failed to convert into Duration"))
                 .with_node("request_timeout")?;
+            let drain_timeout = drain_timeout
+                .map(duration_from_envoy)
+                .transpose()
+                .map_err(|_| GenericError::from_msg("failed to convert into Duration"))
+                .with_node("drain_timeout")?;
             let enabled_upgrades = upgrade_configs
                 .iter()
                 .filter(|upgrade_config| upgrade_config.enabled.map(|enabled| enabled.value).unwrap_or(true))
@@ -819,6 +882,7 @@ mod envoy_conversions {
                 enabled_upgrades,
                 route_specifier,
                 request_timeout,
+                drain_timeout,
                 access_log,
                 xff_settings,
                 generate_request_id: generate_request_id.map(|v| v.value).unwrap_or(true),

orion-configuration/src/config/network_filters/http_connection_manager.rs

@@ -11,9 +11,11 @@ ahash = "0.8.11"
 arc-swap = "1.7.1"
 arrayvec = "0.7.6"
 async-stream = "0.3"
+async-trait = "0.1.77"
 atomic-time = "0.1.4"
 bytes.workspace = true
 compact_str.workspace = true
+dashmap = "6.0"
 enum_dispatch = "0.3.13"
 exponential-backoff.workspace = true
 futures.workspace = true