|
| 1 | +use std::{borrow::Cow, collections::HashSet}; |
| 2 | + |
| 3 | +use schemars::JsonSchema; |
| 4 | +use serde::{Deserialize, Serialize, Serializer, de::Error}; |
| 5 | + |
| 6 | +use crate::config::merge::Merge; |
| 7 | + |
| 8 | +#[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Eq, Serialize)] |
| 9 | +#[serde(rename_all = "camelCase")] |
| 10 | +pub struct JvmArgumentOverrides { |
| 11 | + /// JVM arguments to be added |
| 12 | + #[serde(default)] |
| 13 | + add: Vec<String>, |
| 14 | + |
| 15 | + /// JVM arguments to be removed by exact match |
| 16 | + // |
| 17 | + // HashSet to be optimized for quick lookup |
| 18 | + #[serde(default)] |
| 19 | + remove: HashSet<String>, |
| 20 | + |
| 21 | + /// JVM arguments matching any of this regexes will be removed |
| 22 | + #[serde(default)] |
| 23 | + remove_regex: RegexSet, |
| 24 | + |
| 25 | + /// Sequence of [`JvmArgumentOverrides`] which must be applied before this one |
| 26 | + /// |
| 27 | + /// This field is used internally to combine the role and role group overrides. The fields of |
| 28 | + /// the role group cannot just be appended to the ones of the role because the fields `remove`, |
| 29 | + /// `remove_regex` and `add` of the role must be applied before the ones of the role group. |
| 30 | + #[serde(skip)] |
| 31 | + preceding_overrides: Vec<Self>, |
| 32 | +} |
| 33 | + |
| 34 | +impl Merge for JvmArgumentOverrides { |
| 35 | + fn merge(&mut self, defaults: &Self) { |
| 36 | + self.preceding_overrides.push(defaults.clone()); |
| 37 | + } |
| 38 | +} |
| 39 | + |
| 40 | +impl JvmArgumentOverrides { |
| 41 | + pub fn apply_to(&self, jvm_arguments: impl IntoIterator<Item = String>) -> Vec<String> { |
| 42 | + // 1. Apply the preceding overrides |
| 43 | + self.preceding_overrides |
| 44 | + .iter() |
| 45 | + // The vector should only contain one element, but if it contains more than one then |
| 46 | + // start with the one that was added last. |
| 47 | + .rev() |
| 48 | + .fold( |
| 49 | + jvm_arguments.into_iter().collect(), |
| 50 | + |jvm_arguments, overrides| overrides.apply_to(jvm_arguments), |
| 51 | + ) |
| 52 | + .into_iter() |
| 53 | + // 2. Remove exact matches |
| 54 | + .filter(|arg| !self.remove.contains(arg)) |
| 55 | + // 3. Remove arguments matching the regexes |
| 56 | + .filter(|arg| !self.remove_regex.0.is_match(arg)) |
| 57 | + // 4. Add arguments |
| 58 | + .chain(self.add.clone()) |
| 59 | + .collect() |
| 60 | + } |
| 61 | +} |
| 62 | + |
| 63 | +#[derive(Clone, Debug, Default)] |
| 64 | +struct RegexSet(regex::RegexSet); |
| 65 | + |
| 66 | +impl<'de> Deserialize<'de> for RegexSet { |
| 67 | + fn deserialize<D>(deserializer: D) -> Result<Self, D::Error> |
| 68 | + where |
| 69 | + D: serde::Deserializer<'de>, |
| 70 | + { |
| 71 | + let regexes = <Vec<Cow<'de, str>>>::deserialize(deserializer)?; |
| 72 | + |
| 73 | + let anchored_regexes = regexes |
| 74 | + .iter() |
| 75 | + .map(|maybe_anchored_regex| { |
| 76 | + maybe_anchored_regex |
| 77 | + .trim_start_matches('^') |
| 78 | + .trim_end_matches('$') |
| 79 | + }) |
| 80 | + .map(|unanchored_regex| format!("^{unanchored_regex}$")); |
| 81 | + |
| 82 | + match regex::RegexSet::new(anchored_regexes) { |
| 83 | + Ok(regexset) => Ok(Self(regexset)), |
| 84 | + Err(err) => Err(D::Error::custom(err)), |
| 85 | + } |
| 86 | + } |
| 87 | +} |
| 88 | + |
| 89 | +impl Serialize for RegexSet { |
| 90 | + fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> |
| 91 | + where |
| 92 | + S: Serializer, |
| 93 | + { |
| 94 | + self.0.patterns().serialize(serializer) |
| 95 | + } |
| 96 | +} |
| 97 | + |
| 98 | +impl Eq for RegexSet {} |
| 99 | + |
| 100 | +impl PartialEq for RegexSet { |
| 101 | + fn eq(&self, other: &Self) -> bool { |
| 102 | + self.0.patterns() == other.0.patterns() |
| 103 | + } |
| 104 | +} |
| 105 | + |
| 106 | +impl JsonSchema for RegexSet { |
| 107 | + fn schema_name() -> std::borrow::Cow<'static, str> { |
| 108 | + "RegexSet".into() |
| 109 | + } |
| 110 | + |
| 111 | + fn json_schema(_generator: &mut schemars::SchemaGenerator) -> schemars::Schema { |
| 112 | + schemars::json_schema!({ |
| 113 | + "type": "array", |
| 114 | + "items": { |
| 115 | + "type": "string" |
| 116 | + } |
| 117 | + }) |
| 118 | + } |
| 119 | +} |
| 120 | + |
| 121 | +#[cfg(test)] |
| 122 | +mod tests { |
| 123 | + use stackable_operator_derive::Fragment; |
| 124 | + |
| 125 | + use super::*; |
| 126 | + use crate::{ |
| 127 | + role_utils::{GenericRoleConfig, Role, RoleGroup}, |
| 128 | + v2::role_utils::{JavaCommonConfig, with_validated_config}, |
| 129 | + }; |
| 130 | + |
| 131 | + // #[derive( |
| 132 | + // Clone, Debug, Default, Deserialize, Fragment, JsonSchema, Merge, PartialEq, Serialize, |
| 133 | + // )] |
| 134 | + #[derive(Debug, Fragment, PartialEq)] |
| 135 | + #[fragment_attrs(derive(Clone, Debug, Default, Deserialize, Eq, PartialEq))] |
| 136 | + #[fragment(path_overrides(fragment = "crate::config::fragment",))] |
| 137 | + struct EmptyConfig {} |
| 138 | + |
| 139 | + impl Merge for EmptyConfigFragment { |
| 140 | + fn merge(&mut self, _defaults: &Self) {} |
| 141 | + } |
| 142 | + |
| 143 | + #[derive(Clone, Debug, Default, Deserialize, JsonSchema, Merge, PartialEq, Serialize)] |
| 144 | + #[merge(path_overrides(merge = "crate::config::merge"))] |
| 145 | + struct EmptyConfigOverrides {} |
| 146 | + |
| 147 | + #[test] |
| 148 | + fn test_merge_java_common_config() { |
| 149 | + // The operator generates some JVM arguments |
| 150 | + let operator_generated = [ |
| 151 | + "-Xms34406m".to_owned(), |
| 152 | + "-Xmx34406m".to_owned(), |
| 153 | + "-XX:+UseG1GC".to_owned(), |
| 154 | + "-XX:+ExitOnOutOfMemoryError".to_owned(), |
| 155 | + "-Djava.protocol.handler.pkgs=sun.net.www.protocol".to_owned(), |
| 156 | + "-Dsun.net.http.allowRestrictedHeaders=true".to_owned(), |
| 157 | + "-Djava.security.properties=/stackable/nifi/conf/security.properties".to_owned(), |
| 158 | + ]; |
| 159 | + |
| 160 | + let entire_role: Role< |
| 161 | + EmptyConfigFragment, |
| 162 | + EmptyConfigOverrides, |
| 163 | + GenericRoleConfig, |
| 164 | + JavaCommonConfig, |
| 165 | + > = serde_yaml::from_str( |
| 166 | + " |
| 167 | + # Let's say we want to set some additional HTTP Proxy and IPv4 settings |
| 168 | + # And we don't like the garbage collector for some reason... |
| 169 | + jvmArgumentOverrides: |
| 170 | + remove: |
| 171 | + - -XX:+UseG1GC |
| 172 | + add: # Add some networking arguments |
| 173 | + - -Dhttps.proxyHost=proxy.my.corp |
| 174 | + - -Dhttps.proxyPort=8080 |
| 175 | + - -Djava.net.preferIPv4Stack=true |
| 176 | + roleGroups: |
| 177 | + default: |
| 178 | + # For the roleGroup, let's say we need a different memory config. |
| 179 | + # For that to work we first remove the flags generated by the operator and add our own. |
| 180 | + # Also we override the proxy port to test that the roleGroup config takes precedence over the role config. |
| 181 | + jvmArgumentOverrides: |
| 182 | + removeRegex: |
| 183 | + - -Xmx.* |
| 184 | + - -Dhttps.proxyPort=.* |
| 185 | + add: |
| 186 | + - -Xmx40000m |
| 187 | + - -Dhttps.proxyPort=1234 |
| 188 | + ") |
| 189 | + .expect("Failed to parse role"); |
| 190 | + |
| 191 | + let role_group = entire_role |
| 192 | + .role_groups |
| 193 | + .get("default") |
| 194 | + .expect("role group should be defined"); |
| 195 | + |
| 196 | + let validated_config: RoleGroup<EmptyConfig, _, _> = |
| 197 | + with_validated_config(role_group, &entire_role, &EmptyConfigFragment {}) |
| 198 | + .expect("role spec should be valid"); |
| 199 | + |
| 200 | + let effective_jvm_config = validated_config |
| 201 | + .config |
| 202 | + .product_specific_common_config |
| 203 | + .jvm_argument_overrides |
| 204 | + .apply_to(operator_generated); |
| 205 | + |
| 206 | + let expected = vec![ |
| 207 | + "-Xms34406m".to_owned(), |
| 208 | + "-XX:+ExitOnOutOfMemoryError".to_owned(), |
| 209 | + "-Djava.protocol.handler.pkgs=sun.net.www.protocol".to_owned(), |
| 210 | + "-Dsun.net.http.allowRestrictedHeaders=true".to_owned(), |
| 211 | + "-Djava.security.properties=/stackable/nifi/conf/security.properties".to_owned(), |
| 212 | + "-Dhttps.proxyHost=proxy.my.corp".to_owned(), |
| 213 | + "-Djava.net.preferIPv4Stack=true".to_owned(), |
| 214 | + "-Xmx40000m".to_owned(), |
| 215 | + "-Dhttps.proxyPort=1234".to_owned(), |
| 216 | + ]; |
| 217 | + |
| 218 | + assert_eq!(effective_jvm_config, expected); |
| 219 | + } |
| 220 | + |
| 221 | + #[test] |
| 222 | + fn test_merge_java_common_config_keep_order() { |
| 223 | + let operator_generated = ["-Xms1m".to_owned()]; |
| 224 | + |
| 225 | + let entire_role: Role< |
| 226 | + EmptyConfigFragment, |
| 227 | + EmptyConfigOverrides, |
| 228 | + GenericRoleConfig, |
| 229 | + JavaCommonConfig, |
| 230 | + > = serde_yaml::from_str( |
| 231 | + " |
| 232 | + jvmArgumentOverrides: |
| 233 | + add: |
| 234 | + - -Xms2m |
| 235 | + roleGroups: |
| 236 | + default: |
| 237 | + jvmArgumentOverrides: |
| 238 | + add: |
| 239 | + - -Xms3m |
| 240 | + ", |
| 241 | + ) |
| 242 | + .expect("Failed to parse role"); |
| 243 | + |
| 244 | + let role_group = entire_role |
| 245 | + .role_groups |
| 246 | + .get("default") |
| 247 | + .expect("role group should be defined"); |
| 248 | + |
| 249 | + let validated_config: RoleGroup<EmptyConfig, _, _> = |
| 250 | + with_validated_config(role_group, &entire_role, &EmptyConfigFragment {}) |
| 251 | + .expect("role spec should be valid"); |
| 252 | + |
| 253 | + let effective_jvm_config = validated_config |
| 254 | + .config |
| 255 | + .product_specific_common_config |
| 256 | + .jvm_argument_overrides |
| 257 | + .apply_to(operator_generated); |
| 258 | + |
| 259 | + assert_eq!( |
| 260 | + effective_jvm_config, |
| 261 | + &[ |
| 262 | + "-Xms1m".to_owned(), |
| 263 | + "-Xms2m".to_owned(), |
| 264 | + "-Xms3m".to_owned() |
| 265 | + ] |
| 266 | + ); |
| 267 | + } |
| 268 | +} |
0 commit comments