-
Notifications
You must be signed in to change notification settings - Fork 109
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Splitter CRD Add new features enum and field for operator The old enum cannot be extended as it would break old clients. So we define a new extensible enum and a new field for it. We make all of those fields private so that users of that struct only access them via a getter that handles objects deserialized from either old or new versions. use copy target when splitting * working env * deletecollection for sqs sessions * change SqsMessageFilter to BTreeMap for iterating. * change queue_names to BTreeMap for iterating * change config_map_updates to BTreeMap for iterating * change queue_names to BTreeMap for iterating * change queues to BTreeMap for iterating * get output queue names from splitter status * splitter status * todo * set -> vec * tags * docs * remove active_sessions from splitter status * MirrordQueueSplitter -> MirrordWorkloadQueueRegistry, splitters. -> queues. * merge fix * lock * mergefix * remove pod spec from session status * remove podspec from registry crd status * save only updates instead of containers * docs * changelog * remove tests (moved to sqs-tests branch) * docs and improvements * cargo fmt * mirrord-schema.json * run medschool * fmt * oops * merge * display Q consumer * fix merge * fix feature use * rename sqs details and make optional * rename QueueDetails -> ActiveSqsSplits * move output_q_names to ActiveSqsSplits * fmt * CR: lambda var name * move method call to own .map call * CR: remove match arm for future variants. * CR: document SqsMessageFilter * CR: remove get_target_type * CR: Operator feature variant names. * CR: unused type * CR: docs: "pod spec" -> "pod template" Co-authored-by: Michał Smolarek <[email protected]> * CR: remove dead configmaps code * CR: make session status enum * CR: remove non_exauhstive * status enum stuff * CR: sqs session error - code + reason * just need to push a commit * pub sqs details * pub sqs details * impl Display for SqsSessionError * schema * CR: add container to queue consumer * QueueConsumer registry matching * QueueConsumer container getter * Add RegisteringFilters Status variant * get_split_details * CR: registry docs * CR: doc text formatting suggestion Co-authored-by: Michał Smolarek <[email protected]> * CR: individual tags per queue * missing pub * mark unstable * CR: CopyTargetCrd, send queues config * non non exauhstive * CR: CopyTargetCrd, send queues config * Added unknown variant to QueueFilter * CR: info log on implicit copy-target * Change QueueConsumer to have a valid k8s schema * CRD fixes: skip unknown variants in schema, change CleanupError status for k8s * starting status has to have items * docs * change start time to str because I remember k8s actually can't deal with unsigned integer of 64 bits and more * rule to patch pods * api group of pods is empty string * Can't patch pod container env :( * CR: out of place serde tag Co-authored-by: Michał Smolarek <[email protected]> * Support installing an SQS-enabled operator via `mirrord operator setup` command * incomplete flag docs --------- Co-authored-by: Michał Smolarek <[email protected]>
- Loading branch information
Showing
16 changed files
with
867 additions
and
117 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| Client side support for the upcoming SQS queue splitting support in *mirrord for Teams*. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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(), | ||
| ) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.