Queue Splitting Feature

changelog.d/2066.added.md

@@ -0,0 +1 @@
+Client side support for the upcoming SQS queue splitting support in *mirrord for Teams*.

mirrord-schema.json

@@ -882,6 +882,18 @@
               "type": "null"
             }
           ]
+        },
+        "split_queues": {
+          "title": "feature.split_queues {#feature-split_queues}",
+          "description": "Define filters to split queues by, and make your local application consume only messages that match those filters. If you don't specify any filter for a queue that is however declared in the `MirrordWorkloadQueueRegistry` of the target you're using, a match-nothing filter will be used, and your local application will not receive any messages from that queue.",
+          "anyOf": [
+            {
+              "$ref": "#/definitions/SplitQueuesConfig"
+            },
+            {
+              "type": "null"
+            }
+          ]
         }
       },
       "additionalProperties": false
@@ -1431,6 +1443,33 @@
         }
       ]
     },
+    "QueueFilter": {
+      "description": "More queue types might be added in the future.",
+      "oneOf": [
+        {
+          "description": "Amazon Simple Queue Service.",
+          "type": "object",
+          "required": [
+            "message_filter",
+            "queue_type"
+          ],
+          "properties": {
+            "message_filter": {
+              "type": "object",
+              "additionalProperties": {
+                "type": "string"
+              }
+            },
+            "queue_type": {
+              "type": "string",
+              "enum": [
+                "SQS"
+              ]
+            }
+          }
+        }
+      ]
+    },
     "RolloutTarget": {
       "description": "<!--${internal}--> Mirror the rollout specified by [`RolloutTarget::rollout`].",
       "type": "object",
@@ -1451,6 +1490,16 @@
       },
       "additionalProperties": false
     },
+    "SplitQueuesConfig": {
+      "description": "```json { \"feature\": { \"split_queues\": { \"first-queue\": { \"queue_type\": \"SQS\", \"message_filter\": { \"wows\": \"so wows\", \"coolz\": \"^very .*\" } }, \"second-queue\": { \"queue_type\": \"SQS\", \"message_filter\": { \"who\": \"*you$\" } }, } } } ```",
+      "type": [
+        "object",
+        "null"
+      ],
+      "additionalProperties": {
+        "$ref": "#/definitions/QueueFilter"
+      }
+    },
     "StatefulSetTarget": {
       "type": "object",
       "required": [

mirrord/cli/src/error.rs

@@ -336,7 +336,7 @@ impl From<OperatorApiError> for CliError {
                 feature,
                 operator_version,
             } => Self::FeatureNotSupportedInOperatorError {
-                feature,
+                feature: feature.to_string(),
                 operator_version,
             },
             OperatorApiError::CreateKubeClient(e) => Self::CreateKubeApiFailed(e),

mirrord/cli/src/operator/session.rs

@@ -6,7 +6,7 @@ use mirrord_operator::{
         error::{OperatorApiError, OperatorOperation},
         MaybeClientCert, OperatorApi,
     },
-    crd::SessionCrd,
+    crd::{NewOperatorFeature, SessionCrd},
 };
 use mirrord_progress::{Progress, ProgressTracker};
 use tracing::Level;
@@ -99,7 +99,7 @@ impl SessionCommandHandler {
                 if code == 404 && reason.contains("parse") =>
             {
                 OperatorApiError::UnsupportedFeature {
-                    feature: "session management".to_string(),
+                    feature: NewOperatorFeature::SessionManagement,
                     operator_version: operator_api.operator().spec.operator_version.clone(),
                 }
             }
@@ -135,16 +135,16 @@ impl SessionCommandHandler {
                 )));
                 progress.success(Some("Session operation is completed."));
 
-                Ok(())
-            }
-        })
-        .transpose()?
-        // We might've gotten a `SessionCrd` instead of a `Status` (we have a `Left(T)`),
-        // meaning that the operation has started, but it might not be finished yet.
-        .unwrap_or_else(|| {
-            sub_progress.success(Some(&format!("No issues found when executing `{command}`, but the operation status could not be determined at this time.")));
-            progress.success(Some(&format!("`{command}` is done, but the operation might be pending.")));
-        });
+                    Ok(())
+                }
+            })
+            .transpose()?
+            // We might've gotten a `SessionCrd` instead of a `Status` (we have a `Left(T)`),
+            // meaning that the operation has started, but it might not be finished yet.
+            .unwrap_or_else(|| {
+                sub_progress.success(Some(&format!("No issues found when executing `{command}`, but the operation status could not be determined at this time.")));
+                progress.success(Some(&format!("`{command}` is done, but the operation might be pending.")));
+            });
 
         Ok(())
     }

mirrord/config/configuration.md

@@ -1196,6 +1196,36 @@ of regexes specified here. If there is a match, mirrord will connect your applic
 the target unix socket address on the target pod. Otherwise, it will leave the connection
 to happen locally on your machine.
 
+## feature.split_queues {#feature-split_queues}
+
+Define filters to split queues by, and make your local application consume only messages
+that match those filters.
+If you don't specify any filter for a queue that is however declared in the
+`MirrordWorkloadQueueRegistry` of the target you're using, a match-nothing filter
+will be used, and your local application will not receive any messages from that queue.
+
+```json
+{
+  "feature": {
+    "split_queues": {
+      "first-queue": {
+        "queue_type": "SQS",
+        "message_filter": {
+          "wows": "so wows",
+          "coolz": "^very .*"
+        }
+      },
+      "second-queue": {
+        "queue_type": "SQS",
+        "message_filter": {
+          "who": "*you$"
+        }
+      },
+    }
+  }
+}
+```
+
 ## internal_proxy {#root-internal_proxy}
 
 Configuration for the internal proxy mirrord spawns for each local mirrord session

mirrord/config/src/feature.rs

@@ -4,12 +4,13 @@ use schemars::JsonSchema;
 use serde::Serialize;
 
 use self::{copy_target::CopyTargetConfig, env::EnvConfig, fs::FsConfig, network::NetworkConfig};
-use crate::config::source::MirrordConfigSource;
+use crate::{config::source::MirrordConfigSource, feature::split_queues::SplitQueuesConfig};
 
 pub mod copy_target;
 pub mod env;
 pub mod fs;
 pub mod network;
+pub mod split_queues;
 
 /// Controls mirrord features.
 ///
@@ -96,6 +97,16 @@ pub struct FeatureConfig {
     /// Should mirrord return the hostname of the target pod when calling `gethostname`
     #[config(default = true)]
     pub hostname: bool,
+
+    /// ## feature.split_queues {#feature-split_queues}
+    ///
+    /// Define filters to split queues by, and make your local application consume only messages
+    /// that match those filters.
+    /// If you don't specify any filter for a queue that is however declared in the
+    /// `MirrordWorkloadQueueRegistry` of the target you're using, a match-nothing filter
+    /// will be used, and your local application will not receive any messages from that queue.
+    #[config(nested, unstable)]
+    pub split_queues: SplitQueuesConfig,
 }
 
 impl CollectAnalytics for &FeatureConfig {
@@ -105,5 +116,6 @@ impl CollectAnalytics for &FeatureConfig {
         analytics.add("network", &self.network);
         analytics.add("copy_target", &self.copy_target);
         analytics.add("hostname", self.hostname);
+        analytics.add("split_queues", &self.split_queues);
     }
 }

mirrord/config/src/feature/split_queues.rs

@@ -0,0 +1,113 @@
+use std::{
+    collections::{BTreeMap, HashMap},
+    ops::Not,
+};
+
+use mirrord_analytics::{Analytics, CollectAnalytics};
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::config::{ConfigContext, FromMirrordConfig, MirrordConfig};
+
+pub type QueueId = String;
+
+/// ```json
+/// {
+///   "feature": {
+///     "split_queues": {
+///       "first-queue": {
+///         "queue_type": "SQS",
+///         "message_filter": {
+///           "wows": "so wows",
+///           "coolz": "^very .*"
+///         }
+///       },
+///       "second-queue": {
+///         "queue_type": "SQS",
+///         "message_filter": {
+///           "who": "*you$"
+///         }
+///       },
+///     }
+///   }
+/// }
+/// ```
+#[derive(Clone, Debug, Eq, PartialEq, JsonSchema, Serialize, Deserialize, Default)]
+pub struct SplitQueuesConfig(pub Option<BTreeMap<QueueId, QueueFilter>>);
+
+impl SplitQueuesConfig {
+    pub fn is_set(&self) -> bool {
+        self.0.is_some()
+    }
+
+    /// Out of the whole queue splitting config, get only the sqs queues.
+    pub fn get_sqs_filter(&self) -> Option<HashMap<String, SqsMessageFilter>> {
+        self.0
+            .as_ref()
+            .map(BTreeMap::iter)
+            .map(|filters| {
+                filters
+                    // When there are more variants of QueueFilter, change this to a `filter_map`.
+                    .filter_map(|(queue_id, queue_filter)| match queue_filter {
+                        QueueFilter::Sqs(filter_mapping) => {
+                            Some((queue_id.clone(), filter_mapping.clone()))
+                        }
+                        _ => None,
+                    })
+                    .collect()
+            })
+            .and_then(|filters_map: HashMap<String, SqsMessageFilter>| {
+                filters_map.is_empty().not().then_some(filters_map)
+            })
+    }
+}
+
+impl MirrordConfig for SplitQueuesConfig {
+    type Generated = Self;
+
+    fn generate_config(
+        self,
+        _context: &mut ConfigContext,
+    ) -> crate::config::Result<Self::Generated> {
+        Ok(self)
+    }
+}
+
+impl FromMirrordConfig for SplitQueuesConfig {
+    type Generator = Self;
+}
+
+pub type MessageAttributeName = String;
+pub type AttributeValuePattern = String;
+
+/// A filter is a mapping between message attribute names and regexes they should match.
+/// The local application will only receive messages that match **all** of the given patterns.
+/// This means, only messages that have **all** the `MessageAttributeName`s in the filter,
+/// with values of those attributes matching the respective `AttributeValuePattern`.
+pub type SqsMessageFilter = BTreeMap<MessageAttributeName, AttributeValuePattern>;
+
+/// More queue types might be added in the future.
+#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, JsonSchema)]
+#[serde(tag = "queue_type", content = "message_filter")]
+pub enum QueueFilter {
+    /// Amazon Simple Queue Service.
+    #[serde(rename = "SQS")]
+    Sqs(SqsMessageFilter),
+    /// When a newer client sends a new filter kind to an older operator, that does not yet know
+    /// about that filter type, this is what that filter will be deserialized to.
+    #[schemars(skip)]
+    #[serde(other, skip_serializing)]
+    Unknown,
+}
+
+impl CollectAnalytics for &SplitQueuesConfig {
+    fn collect_analytics(&self, analytics: &mut Analytics) {
+        analytics.add(
+            "queue_count",
+            self.0
+                .as_ref()
+                .map(|mapping| mapping.len())
+                .unwrap_or_default(),
+        )
+    }
+}

mirrord/config/src/lib.rs

@@ -852,6 +852,7 @@ mod tests {
                 })),
                 copy_target: None,
                 hostname: None,
+                split_queues: None,
             }),
             connect_tcp: None,
             container: None,

mirrord/kube/src/api/kubernetes/rollout.rs

@@ -4,9 +4,9 @@ use k8s_openapi::{
     apimachinery::pkg::apis::meta::v1::ObjectMeta, ListableResource, Metadata,
     NamespaceResourceScope, Resource,
 };
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 
-#[derive(Clone, Debug, Deserialize)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Rollout {
     metadata: ObjectMeta,
     pub spec: serde_json::Value,

mirrord/operator/Cargo.toml

@@ -56,7 +56,7 @@ mirrord-progress = { path = "../progress", optional = true }
 mirrord-protocol = { path = "../protocol", optional = true }
 
 base64 = { workspace = true, optional = true }
-bincode =  { version = "2.0.0-rc.2", features = ["serde"], optional = true }
+bincode = { version = "2.0.0-rc.2", features = ["serde"], optional = true }
 chrono = { version = "0.4", features = ["clock", "serde"] }
 http = { version = "1", optional = true }
 http-body-util = { workspace = true, optional = true }