add qos overwrite interceptor

Author: DenisBiryukov91

DEFAULT_CONFIG.json5

@@ -398,6 +398,39 @@
   //   ]
   //},
 
+  //  /// Configure QoS overwrite rules.
+  //  qos_overwrite: [
+  //     {
+  //       /// Id has to be unique within the rule set.
+  //       id: "rule1",
+  //       // Optional list of interfaces, if not specified, will be applied to all interfaces.
+  //       interfaces: [
+  //         "lo0",
+  //         "en0",
+  //       ],
+  //       messages: ["put", "delete"],
+  //       flows: ["egress"],
+  //       key_exprs: ["test/demo"],
+  //       overwrite: {
+  //          /// Optional new priority value, if not specified priority of the messages will stay unchanged.
+  //          priority: "real_time",
+  //          /// Optional new congestion control value, if not specified congestion control of the messages will stay unchanged.
+  //          congestion_control: "block",
+  //          /// Optional new priority value, if not specified priority of the messages will stay unchanged.
+  //          express: true
+  //        },
+  //     },
+  //     {
+  //        id: "rule2",
+  //        messages: ["query", "reply"],
+  //        flows: ["ingress"],
+  //        key_exprs: ["**"],
+  //        overwrite: {
+  //          priority: "interactive_high",
+  //        },
+  //     },
+  //   ],
+
   /// Configure internal transport parameters
   transport: {
     unicast: {

commons/zenoh-config/src/lib.rs

@@ -32,7 +32,7 @@ use std::{
 };
 
 use include::recursive_include;
-use qos::PublisherQoSConfList;
+use qos::{PublisherQoSConfList, QosOverwrites};
 use secrecy::{CloneableSecret, DebugSecret, Secret, SerializableSecret, Zeroize};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
@@ -201,6 +201,32 @@ pub enum Permission {
     Deny,
 }
 
+#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, Hash, PartialEq)]
+#[serde(rename_all = "snake_case")]
+pub enum QosOverwriteMessage {
+    Put,
+    Delete,
+    Query,
+    Reply,
+}
+
+#[derive(Default, Debug, Deserialize, Serialize, Clone)]
+pub struct QosOverwriteItemConf {
+    /// Optional identifier for the qos modification configuration item.
+    pub id: Option<String>,
+    /// A list of interfaces to which the qos will be applied.
+    /// QosOverwrite will be applied for all interfaces if the parameter is None.
+    pub interfaces: Option<Vec<String>>,
+    /// List of message types on which the qos ovewrite will be applied.
+    pub messages: Vec<QosOverwriteMessage>,
+    /// List of key expressions to apply qos ovewrite.
+    pub key_exprs: Vec<OwnedKeyExpr>,
+    // The qos value to ovewrite with.
+    pub overwrite: QosOverwrites,
+    /// QosOverwrite flow directions: egress and/or ingress.
+    pub flows: Option<Vec<InterceptorFlow>>,
+}
+
 /// Strategy for autoconnection, mainly to avoid nodes connecting to each other redundantly.
 #[derive(Default, Clone, Copy, Debug, Serialize, Deserialize, Eq, Hash, PartialEq)]
 #[serde(rename_all = "kebab-case")]
@@ -667,6 +693,9 @@ validated_struct::validator! {
             pub policies: Option<Vec<AclConfigPolicyEntry>>,
         },
 
+        /// Configuration of the qos overwrite rules.
+        pub qos_overwrite: Vec<QosOverwriteItemConf>,
+
         /// A list of directories where plugins may be searched for if no `__path__` was specified for them.
         /// The executable's current directory will be added to the search paths.
         pub plugins_loading: #[derive(Default)]

commons/zenoh-config/src/qos.rs

@@ -13,7 +13,7 @@
 //
 use serde::{Deserialize, Serialize};
 use zenoh_keyexpr::keyexpr_tree::{IKeyExprTreeMut, KeBoxTree};
-use zenoh_protocol::core::{key_expr::OwnedKeyExpr, CongestionControl, Reliability};
+use zenoh_protocol::core::{key_expr::OwnedKeyExpr, CongestionControl, Priority, Reliability};
 
 #[derive(Debug, Deserialize, Default, Serialize, Clone)]
 pub struct PublisherQoSConfList(pub(crate) Vec<PublisherQoSConf>);
@@ -39,32 +39,32 @@ pub(crate) struct PublisherQoSConf {
 
 #[derive(Debug, Default, Deserialize, Serialize, Clone)]
 pub struct PublisherQoSConfig {
-    pub congestion_control: Option<PublisherCongestionControlConf>,
-    pub priority: Option<PublisherPriorityConf>,
+    pub congestion_control: Option<CongestionControlConf>,
+    pub priority: Option<PriorityConf>,
     pub express: Option<bool>,
     #[cfg(feature = "unstable")]
-    pub reliability: Option<PublisherReliabilityConf>,
+    pub reliability: Option<ReliabilityConf>,
     #[cfg(feature = "unstable")]
     pub allowed_destination: Option<PublisherLocalityConf>,
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone, Copy)]
 #[serde(rename_all = "lowercase")]
-pub enum PublisherCongestionControlConf {
+pub enum CongestionControlConf {
     Drop,
     Block,
 }
 
-impl From<PublisherCongestionControlConf> for CongestionControl {
-    fn from(value: PublisherCongestionControlConf) -> Self {
+impl From<CongestionControlConf> for CongestionControl {
+    fn from(value: CongestionControlConf) -> Self {
         match value {
-            PublisherCongestionControlConf::Drop => Self::Drop,
-            PublisherCongestionControlConf::Block => Self::Block,
+            CongestionControlConf::Drop => Self::Drop,
+            CongestionControlConf::Block => Self::Block,
         }
     }
 }
 
-impl From<CongestionControl> for PublisherCongestionControlConf {
+impl From<CongestionControl> for CongestionControlConf {
     fn from(value: CongestionControl) -> Self {
         match value {
             CongestionControl::Drop => Self::Drop,
@@ -75,7 +75,7 @@ impl From<CongestionControl> for PublisherCongestionControlConf {
 
 #[derive(Debug, Deserialize, Serialize, Clone, Copy)]
 #[serde(rename_all = "snake_case")]
-pub enum PublisherPriorityConf {
+pub enum PriorityConf {
     RealTime = 1,
     InteractiveHigh = 2,
     InteractiveLow = 3,
@@ -85,23 +85,37 @@ pub enum PublisherPriorityConf {
     Background = 7,
 }
 
+impl From<PriorityConf> for Priority {
+    fn from(value: PriorityConf) -> Self {
+        match value {
+            PriorityConf::RealTime => Self::RealTime,
+            PriorityConf::InteractiveHigh => Self::InteractiveHigh,
+            PriorityConf::InteractiveLow => Self::InteractiveLow,
+            PriorityConf::DataHigh => Self::DataHigh,
+            PriorityConf::Data => Self::Data,
+            PriorityConf::DataLow => Self::DataLow,
+            PriorityConf::Background => Self::Background,
+        }
+    }
+}
+
 #[derive(Debug, Deserialize, Serialize, Clone, Copy)]
 #[serde(rename_all = "snake_case")]
-pub enum PublisherReliabilityConf {
+pub enum ReliabilityConf {
     BestEffort,
     Reliable,
 }
 
-impl From<PublisherReliabilityConf> for Reliability {
-    fn from(value: PublisherReliabilityConf) -> Self {
+impl From<ReliabilityConf> for Reliability {
+    fn from(value: ReliabilityConf) -> Self {
         match value {
-            PublisherReliabilityConf::BestEffort => Self::BestEffort,
-            PublisherReliabilityConf::Reliable => Self::Reliable,
+            ReliabilityConf::BestEffort => Self::BestEffort,
+            ReliabilityConf::Reliable => Self::Reliable,
         }
     }
 }
 
-impl From<Reliability> for PublisherReliabilityConf {
+impl From<Reliability> for ReliabilityConf {
     fn from(value: Reliability) -> Self {
         match value {
             Reliability::BestEffort => Self::BestEffort,
@@ -117,3 +131,13 @@ pub enum PublisherLocalityConf {
     Remote,
     Any,
 }
+
+#[derive(Debug, Default, Deserialize, Serialize, Clone)]
+pub struct QosOverwrites {
+    pub congestion_control: Option<CongestionControlConf>,
+    pub priority: Option<PriorityConf>,
+    pub express: Option<bool>,
+    // TODO: Add support for reliability overwrite (it is not possible right now, since reliability is not a part of RoutingContext, nor NetworkMessage)
+    // #[cfg(feature = "unstable")]
+    // pub reliability: Option<ReliabilityConf>,
+}

zenoh/src/api/publisher.rs

@@ -23,7 +23,7 @@ use std::{
 use futures::Sink;
 use serde::Deserialize;
 use tracing::error;
-use zenoh_config::qos::PublisherPriorityConf;
+use zenoh_config::qos::PriorityConf;
 use zenoh_core::{Resolvable, Resolve, Wait};
 use zenoh_protocol::core::CongestionControl;
 use zenoh_result::{Error, ZResult};
@@ -484,21 +484,21 @@ impl TryFrom<u8> for Priority {
     }
 }
 
-impl From<PublisherPriorityConf> for Priority {
-    fn from(value: PublisherPriorityConf) -> Self {
+impl From<PriorityConf> for Priority {
+    fn from(value: PriorityConf) -> Self {
         match value {
-            PublisherPriorityConf::RealTime => Self::RealTime,
-            PublisherPriorityConf::InteractiveHigh => Self::InteractiveHigh,
-            PublisherPriorityConf::InteractiveLow => Self::InteractiveLow,
-            PublisherPriorityConf::DataHigh => Self::DataHigh,
-            PublisherPriorityConf::Data => Self::Data,
-            PublisherPriorityConf::DataLow => Self::DataLow,
-            PublisherPriorityConf::Background => Self::Background,
+            PriorityConf::RealTime => Self::RealTime,
+            PriorityConf::InteractiveHigh => Self::InteractiveHigh,
+            PriorityConf::InteractiveLow => Self::InteractiveLow,
+            PriorityConf::DataHigh => Self::DataHigh,
+            PriorityConf::Data => Self::Data,
+            PriorityConf::DataLow => Self::DataLow,
+            PriorityConf::Background => Self::Background,
         }
     }
 }
 
-impl From<Priority> for PublisherPriorityConf {
+impl From<Priority> for PriorityConf {
     fn from(value: Priority) -> Self {
         match value {
             Priority::RealTime => Self::RealTime,

zenoh/src/net/routing/interceptor/mod.rs

@@ -35,6 +35,9 @@ use crate::api::key_expr::KeyExpr;
 pub mod downsampling;
 use crate::net::routing::interceptor::downsampling::downsampling_interceptor_factories;
 
+pub mod qos_overwrite;
+use crate::net::routing::interceptor::qos_overwrite::qos_overwrite_interceptor_factories;
+
 #[derive(Default, Debug)]
 pub struct InterfaceEnabled {
     pub ingress: bool,
@@ -88,6 +91,7 @@ pub(crate) fn interceptor_factories(config: &Config) -> ZResult<Vec<InterceptorF
     // res.push(Box::new(LoggerInterceptor {}));
     res.extend(downsampling_interceptor_factories(config.downsampling())?);
     res.extend(acl_interceptor_factories(config.access_control())?);
+    res.extend(qos_overwrite_interceptor_factories(config.qos_overwrite())?);
     Ok(res)
 }