|
| 1 | +//! Tracing filter construction, including the hard floor that keeps |
| 2 | +//! value-bearing query plans out of the log/OTLP stream. |
| 3 | +//! |
| 4 | +//! `POST /query` deliberately never logs the raw statement (see |
| 5 | +//! [`crate::query_handlers`]) because callers may inline literal secrets or |
| 6 | +//! PII. Suppressing the handler's own `sql` field is not enough on its own: |
| 7 | +//! DataFusion reconstructs the *same literals* inside the plans it prints at |
| 8 | +//! DEBUG, e.g. |
| 9 | +//! |
| 10 | +//! ```text |
| 11 | +//! Projection: Utf8("TOP_SECRET") <- datafusion_optimizer::utils::log_plan |
| 12 | +//! ProjectionExec: expr=[TOP_SECRET as ...] <- datafusion-tracing span field |
| 13 | +//! ``` |
| 14 | +//! |
| 15 | +//! Those lines flow through the same subscriber — and therefore the same OTLP |
| 16 | +//! exporter — as everything else. So the confidentiality guarantee is enforced |
| 17 | +//! here, at the filter: every target known to print plans is pinned to INFO and |
| 18 | +//! cannot be lowered by `RUST_LOG`. |
| 19 | +//! |
| 20 | +//! An operator who is knowingly debugging a planning problem on non-sensitive |
| 21 | +//! data can lift the floor with `SKARDI_ALLOW_PLAN_VALUE_LOGGING=1`. It is an |
| 22 | +//! explicit, separate opt-in precisely because it re-enables value export. |
| 23 | +
|
| 24 | +use tracing_subscriber::EnvFilter; |
| 25 | + |
| 26 | +/// Env var that lifts the plan-value logging floor. Any value other than |
| 27 | +/// `0`/`false`/`no`/empty enables plan logging. |
| 28 | +pub const ALLOW_PLAN_VALUE_LOGGING_ENV: &str = "SKARDI_ALLOW_PLAN_VALUE_LOGGING"; |
| 29 | + |
| 30 | +/// Tracing target used for the datafusion-tracing execution/rule spans this |
| 31 | +/// server installs (see [`crate::server::setup_app_state`]). |
| 32 | +/// |
| 33 | +/// Those macros default their target to `module_path!()` of the *call site*, |
| 34 | +/// which would bury the spans under `skardi_server::server` where no filter |
| 35 | +/// could single them out. Naming the target explicitly is what makes the floor |
| 36 | +/// below able to cover them. |
| 37 | +pub const QUERY_PLAN_TARGET: &str = "skardi_query_plan"; |
| 38 | + |
| 39 | +/// Target prefixes whose DEBUG/TRACE records embed query plans, and with them |
| 40 | +/// the literal values from the statement. |
| 41 | +/// |
| 42 | +/// Matching is by prefix, mirroring `EnvFilter`'s own target matching, so |
| 43 | +/// `datafusion` also covers `datafusion_optimizer`, `datafusion_sql`, |
| 44 | +/// `datafusion_physical_optimizer`, `datafusion_federation` and |
| 45 | +/// `datafusion_tracing`. |
| 46 | +const PLAN_VALUE_TARGET_PREFIXES: &[&str] = &["datafusion", QUERY_PLAN_TARGET, "sqlparser"]; |
| 47 | + |
| 48 | +/// Build the tracing filter from the given `RUST_LOG` value (`None` when the |
| 49 | +/// variable is unset or invalid unicode), defaulting to `info`. |
| 50 | +/// |
| 51 | +/// Two hard caps are applied on top of whatever the operator asked for: |
| 52 | +/// |
| 53 | +/// * aws_config logs the AWS access key id in plaintext at INFO when resolving |
| 54 | +/// credentials, so it is capped at WARN unless `RUST_LOG` explicitly opts in. |
| 55 | +/// * Every target in [`PLAN_VALUE_TARGET_PREFIXES`] is pinned to INFO unless |
| 56 | +/// `allow_plan_value_logging` is set. Directives in `RUST_LOG` that would |
| 57 | +/// lower one of those targets are dropped before parsing, so neither a global |
| 58 | +/// `RUST_LOG=debug` nor a targeted `RUST_LOG=datafusion_optimizer=debug` can |
| 59 | +/// put query literals on the wire. |
| 60 | +pub fn build_env_filter(rust_log: Option<&str>, allow_plan_value_logging: bool) -> EnvFilter { |
| 61 | + let retained: Vec<&str> = match rust_log { |
| 62 | + Some(v) if !allow_plan_value_logging => v |
| 63 | + .split(',') |
| 64 | + .filter(|d| !directive_leaks_plan_values(d)) |
| 65 | + .collect(), |
| 66 | + Some(v) => v.split(',').collect(), |
| 67 | + None => Vec::new(), |
| 68 | + }; |
| 69 | + |
| 70 | + // Dropping every directive leaves an empty string, which `EnvFilter` reads |
| 71 | + // as "enable nothing" rather than "unset" — fall back to the default. |
| 72 | + let sanitized = retained.join(","); |
| 73 | + let mut env_filter = Some(sanitized.as_str()) |
| 74 | + .filter(|s| !s.trim().is_empty()) |
| 75 | + .and_then(|v| EnvFilter::try_new(v).ok()) |
| 76 | + .unwrap_or_else(|| "info".into()); |
| 77 | + |
| 78 | + if !allow_plan_value_logging { |
| 79 | + for prefix in PLAN_VALUE_TARGET_PREFIXES { |
| 80 | + // Anything the operator set for this exact target survived the |
| 81 | + // filter above, so it is already at or above the floor — adding the |
| 82 | + // floor would *raise* it (an explicit `datafusion=off` would come |
| 83 | + // back as `info`, since `add_directive` replaces by target). |
| 84 | + if retained.iter().any(|d| directive_target(d) == Some(prefix)) { |
| 85 | + continue; |
| 86 | + } |
| 87 | + env_filter = env_filter.add_directive( |
| 88 | + format!("{prefix}=info") |
| 89 | + .parse() |
| 90 | + .expect("valid plan-value floor directive"), |
| 91 | + ); |
| 92 | + } |
| 93 | + } |
| 94 | + |
| 95 | + if !rust_log.is_some_and(|v| v.contains("aws_config")) { |
| 96 | + env_filter = env_filter.add_directive("aws_config=warn".parse().expect("valid directive")); |
| 97 | + } |
| 98 | + env_filter |
| 99 | +} |
| 100 | + |
| 101 | +/// Read the plan-logging opt-out from the environment. |
| 102 | +pub fn allow_plan_value_logging_from_env() -> bool { |
| 103 | + std::env::var(ALLOW_PLAN_VALUE_LOGGING_ENV).is_ok_and(|v| { |
| 104 | + !matches!( |
| 105 | + v.trim().to_ascii_lowercase().as_str(), |
| 106 | + "" | "0" | "false" | "no" | "off" |
| 107 | + ) |
| 108 | + }) |
| 109 | +} |
| 110 | + |
| 111 | +/// Whether one comma-separated `RUST_LOG` directive would enable DEBUG/TRACE |
| 112 | +/// output for a plan-printing target. |
| 113 | +/// |
| 114 | +/// Bare level names (`debug`) set the *global default* and are left alone — the |
| 115 | +/// per-target floor directives out-specify them. A bare target (`datafusion`) |
| 116 | +/// means TRACE for that target in env_logger syntax, so it counts as a leak. |
| 117 | +fn directive_leaks_plan_values(directive: &str) -> bool { |
| 118 | + let Some(target) = directive_target(directive) else { |
| 119 | + return false; |
| 120 | + }; |
| 121 | + if !PLAN_VALUE_TARGET_PREFIXES |
| 122 | + .iter() |
| 123 | + .any(|prefix| target.starts_with(prefix)) |
| 124 | + { |
| 125 | + return false; |
| 126 | + } |
| 127 | + |
| 128 | + let level = directive |
| 129 | + .trim() |
| 130 | + .rsplit_once('=') |
| 131 | + .map(|(_, level)| level.trim()); |
| 132 | + match level { |
| 133 | + // `datafusion` / `datafusion=` both mean TRACE. |
| 134 | + None | Some("") => true, |
| 135 | + Some(level) => { |
| 136 | + let level = level.to_ascii_lowercase(); |
| 137 | + level == "debug" || level == "trace" || level.parse::<u8>().is_ok_and(|n| n >= 4) |
| 138 | + } |
| 139 | + } |
| 140 | +} |
| 141 | + |
| 142 | +/// The target a directive names, with any span filter stripped |
| 143 | +/// (`datafusion[span{k=v}]=debug` -> `datafusion`). `None` for a bare level |
| 144 | +/// name, which sets the global default rather than naming a target. |
| 145 | +fn directive_target(directive: &str) -> Option<&str> { |
| 146 | + let directive = directive.trim(); |
| 147 | + if directive.is_empty() { |
| 148 | + return None; |
| 149 | + } |
| 150 | + let target = match directive.rsplit_once('=') { |
| 151 | + Some((target, _)) => target, |
| 152 | + None if is_level_name(directive) => return None, |
| 153 | + None => directive, |
| 154 | + }; |
| 155 | + Some(target.split('[').next().unwrap_or(target).trim()) |
| 156 | +} |
| 157 | + |
| 158 | +/// Whether a bare `RUST_LOG` token is a level (global default) rather than a |
| 159 | +/// target name. |
| 160 | +fn is_level_name(token: &str) -> bool { |
| 161 | + matches!( |
| 162 | + token.to_ascii_lowercase().as_str(), |
| 163 | + "off" | "error" | "warn" | "info" | "debug" | "trace" |
| 164 | + ) || token.parse::<u8>().is_ok() |
| 165 | +} |
| 166 | + |
| 167 | +#[cfg(test)] |
| 168 | +mod tests { |
| 169 | + use super::*; |
| 170 | + |
| 171 | + #[test] |
| 172 | + fn env_filter_defaults_to_info_and_caps_aws_config() { |
| 173 | + let filter = build_env_filter(None, false).to_string(); |
| 174 | + assert!(filter.contains("info"), "got {filter}"); |
| 175 | + assert!(filter.contains("aws_config=warn"), "got {filter}"); |
| 176 | + } |
| 177 | + |
| 178 | + #[test] |
| 179 | + fn env_filter_caps_aws_config_for_unrelated_rust_log() { |
| 180 | + let filter = build_env_filter(Some("debug"), false).to_string(); |
| 181 | + assert!(filter.contains("debug"), "got {filter}"); |
| 182 | + assert!(filter.contains("aws_config=warn"), "got {filter}"); |
| 183 | + } |
| 184 | + |
| 185 | + #[test] |
| 186 | + fn env_filter_honors_explicit_aws_config_opt_in() { |
| 187 | + let filter = build_env_filter(Some("info,aws_config=debug"), false).to_string(); |
| 188 | + assert!(filter.contains("aws_config=debug"), "got {filter}"); |
| 189 | + assert!(!filter.contains("aws_config=warn"), "got {filter}"); |
| 190 | + } |
| 191 | + |
| 192 | + #[test] |
| 193 | + fn env_filter_falls_back_to_info_on_invalid_rust_log() { |
| 194 | + let filter = build_env_filter(Some("not a [valid directive"), false).to_string(); |
| 195 | + assert!(filter.contains("info"), "got {filter}"); |
| 196 | + assert!(filter.contains("aws_config=warn"), "got {filter}"); |
| 197 | + } |
| 198 | + |
| 199 | + #[test] |
| 200 | + fn global_debug_does_not_lower_plan_targets() { |
| 201 | + let filter = build_env_filter(Some("debug"), false).to_string(); |
| 202 | + for prefix in PLAN_VALUE_TARGET_PREFIXES { |
| 203 | + assert!(filter.contains(&format!("{prefix}=info")), "got {filter}"); |
| 204 | + } |
| 205 | + } |
| 206 | + |
| 207 | + #[test] |
| 208 | + fn targeted_plan_debug_directives_are_dropped() { |
| 209 | + let filter = build_env_filter( |
| 210 | + Some( |
| 211 | + "info,datafusion_optimizer=debug,datafusion_tracing=trace,skardi_query_plan=debug", |
| 212 | + ), |
| 213 | + false, |
| 214 | + ) |
| 215 | + .to_string(); |
| 216 | + assert!( |
| 217 | + !filter.contains("datafusion_optimizer=debug"), |
| 218 | + "got {filter}" |
| 219 | + ); |
| 220 | + assert!(!filter.contains("datafusion_tracing=trace"), "got {filter}"); |
| 221 | + assert!(!filter.contains("skardi_query_plan=debug"), "got {filter}"); |
| 222 | + assert!(filter.contains("datafusion=info"), "got {filter}"); |
| 223 | + } |
| 224 | + |
| 225 | + #[test] |
| 226 | + fn bare_plan_target_directive_is_dropped() { |
| 227 | + // `RUST_LOG=datafusion` means TRACE for that target. |
| 228 | + let filter = build_env_filter(Some("datafusion"), false).to_string(); |
| 229 | + assert!(filter.contains("datafusion=info"), "got {filter}"); |
| 230 | + assert!(filter.contains("info"), "got {filter}"); |
| 231 | + } |
| 232 | + |
| 233 | + #[test] |
| 234 | + fn plan_only_rust_log_falls_back_to_info() { |
| 235 | + let filter = build_env_filter(Some("datafusion=debug"), false).to_string(); |
| 236 | + assert!(filter.contains("info"), "got {filter}"); |
| 237 | + } |
| 238 | + |
| 239 | + #[test] |
| 240 | + fn non_plan_targets_keep_their_debug_level() { |
| 241 | + let filter = build_env_filter(Some("warn,skardi_server=debug"), false).to_string(); |
| 242 | + assert!(filter.contains("skardi_server=debug"), "got {filter}"); |
| 243 | + } |
| 244 | + |
| 245 | + #[test] |
| 246 | + fn raising_plan_targets_is_still_allowed() { |
| 247 | + // Only *lowering* is blocked; `off`/`warn` are honored as written. |
| 248 | + let filter = build_env_filter(Some("debug,datafusion=off"), false).to_string(); |
| 249 | + assert!(filter.contains("datafusion=off"), "got {filter}"); |
| 250 | + } |
| 251 | + |
| 252 | + #[test] |
| 253 | + fn explicit_opt_in_lifts_the_floor() { |
| 254 | + let filter = build_env_filter(Some("datafusion_optimizer=debug"), true).to_string(); |
| 255 | + assert!( |
| 256 | + filter.contains("datafusion_optimizer=debug"), |
| 257 | + "got {filter}" |
| 258 | + ); |
| 259 | + assert!(!filter.contains("datafusion=info"), "got {filter}"); |
| 260 | + } |
| 261 | + |
| 262 | + #[test] |
| 263 | + fn span_filter_syntax_is_recognised() { |
| 264 | + assert!(directive_leaks_plan_values("datafusion[span{k=v}]=debug")); |
| 265 | + assert!(!directive_leaks_plan_values("datafusion[span{k=v}]=info")); |
| 266 | + } |
| 267 | +} |
0 commit comments