diff --git a/src/commands/logs.rs b/src/commands/logs.rs index 818d2fcdc..24fd698d7 100644 --- a/src/commands/logs.rs +++ b/src/commands/logs.rs @@ -6,16 +6,17 @@ use is_terminal::IsTerminal; use crate::{ controllers::{ deployment::{ - FetchLogsParams, FetchNetworkFlowLogsParams, fetch_build_logs, fetch_deploy_logs, - fetch_http_logs, fetch_network_flow_logs, stream_build_logs, stream_deploy_logs, - stream_http_logs, stream_network_flow_logs, + FetchDnsQueryLogsParams, FetchLogsParams, FetchNetworkFlowLogsParams, fetch_build_logs, + fetch_deploy_logs, fetch_dns_query_logs, fetch_http_logs, fetch_network_flow_logs, + stream_build_logs, stream_deploy_logs, stream_dns_query_logs, stream_http_logs, + stream_network_flow_logs, }, project::resolve_service_context, }, util::{ logs::{ - LogFormat, format_network_flow_log_header, print_http_log, print_log, - print_network_flow_log, + LogFormat, format_dns_query_log_header, format_network_flow_log_header, + print_dns_query_log, print_http_log, print_log, print_network_flow_log, }, time::parse_time, }, @@ -173,6 +174,21 @@ impl fmt::Display for NetworkFlowPeerKind { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)] +pub enum DnsZone { + Internal, + External, +} + +impl fmt::Display for DnsZone { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Internal => write!(f, "internal"), + Self::External => write!(f, "external"), + } + } +} + fn build_http_filter(args: &Args) -> Result> { let status = args .status @@ -190,6 +206,38 @@ fn build_http_filter(args: &Args) -> Result> { )) } +fn build_dns_filter(args: &Args) -> Result> { + let mut filters = Vec::new(); + + if let Some(domain) = args.domain.as_deref() { + filters.push(format!("@domain:{domain}")); + } + if let Some(qname) = args.qname.as_deref() { + filters.push(format!("@qname:{qname}")); + } + if let Some(qtype) = args.qtype.as_deref() { + filters.push(format!("@qtype:{}", qtype.to_ascii_uppercase())); + } + if let Some(rcode) = args.rcode.as_deref() { + filters.push(format!("@rcode:{}", rcode.to_ascii_uppercase())); + } + if let Some(zone) = args.zone { + filters.push(format!("@zone:{zone}")); + } + if let Some(status) = args.status.as_deref() { + let status = status.to_ascii_lowercase(); + if status != "ok" && status != "failed" { + bail!("Invalid DNS status: {status}. Expected ok or failed."); + } + filters.push(format!("@status:{status}")); + } + if let Some(raw_filter) = args.filter.as_deref().filter(|filter| !filter.is_empty()) { + filters.push(raw_filter.to_string()); + } + + Ok((!filters.is_empty()).then(|| filters.join(" "))) +} + fn build_network_flow_filter(args: &Args) -> Result> { if args.method.is_some() || args.path.is_some() || args.request_id.is_some() { bail!("--method, --path, and --request-id can only be used with --http"); @@ -212,8 +260,8 @@ fn build_network_flow_filter(args: &Args) -> Result> { } fn validate_filter_modes(args: &Args) -> Result<()> { - if args.status.is_some() && !args.http && !args.network { - bail!("--status can only be used with --http or --network"); + if args.status.is_some() && !args.http && !args.network && !args.dns { + bail!("--status can only be used with --http, --network, or --dns"); } if !args.http && (args.method.is_some() || args.path.is_some() || args.request_id.is_some()) { @@ -226,9 +274,21 @@ fn validate_filter_modes(args: &Args) -> Result<()> { ); } + if !args.dns && has_dns_filter_args(args) { + bail!("--domain, --qname, --qtype, --rcode, and --zone can only be used with --dns"); + } + Ok(()) } +fn has_dns_filter_args(args: &Args) -> bool { + args.domain.is_some() + || args.qname.is_some() + || args.qtype.is_some() + || args.rcode.is_some() + || args.zone.is_some() +} + fn has_network_flow_filter_args(args: &Args) -> bool { args.protocol.is_some() || args.direction.is_some() @@ -371,8 +431,8 @@ pub fn compose_http_filter( #[derive(Parser)] #[clap( - about = "View build, deploy, HTTP, or network flow logs", - long_about = "View build, deploy, HTTP, or network flow logs. This will stream logs by default, or fetch historical logs if the --lines, --since, or --until flags are provided.", + about = "View build, deploy, HTTP, network flow, or DNS logs", + long_about = "View build, deploy, HTTP, network flow, or DNS logs. This will stream logs by default, or fetch historical logs if the --lines, --since, or --until flags are provided.", after_help = "Examples: Deployment logs: @@ -406,7 +466,15 @@ pub fn compose_http_filter( railway logs --network --lines 100 # Pull a snapshot and exit railway logs --network --direction egress --protocol tcp # Outbound TCP flows railway logs --network --peer postgres --port 5432 # Flows to a service peer - railway logs --network --status dropped # Dropped flows" + railway logs --network --status dropped # Dropped flows + + DNS logs: + railway logs --dns # Stream live DNS queries + railway logs --dns --status failed # Failed lookups only + railway logs --dns --rcode NXDOMAIN # Names that don't exist + railway logs --dns --zone internal # Private network lookups + railway logs --dns --domain example.com --qtype AAAA # IPv6 lookups for a domain + railway logs --dns --qname backend.railway.internal --lines 50 # History for an exact name" )] pub struct Args { /// Service to view logs from (defaults to linked service). Can be service name or service ID @@ -437,6 +505,10 @@ pub struct Args { #[clap(long, group = "log_type")] network: bool, + /// Show DNS query logs + #[clap(long, group = "log_type")] + dns: bool, + /// Deployment ID to view logs from. Defaults to most recent successful deployment, or latest deployment if none succeeded deployment_id: Option, @@ -473,6 +545,9 @@ For network flow logs (--network), all filterable fields: Numeric: @port Boolean: @dropped +For DNS logs (--dns), all filterable fields: + String: @qname, @domain, @qtype, @rcode, @zone, @status + Numeric operators: > >= < <= .. (range, e.g. @httpStatus:200..299) Logical operators: AND, OR, - (negation), parentheses for grouping @@ -491,7 +566,7 @@ Examples: #[clap(long, requires = "http", value_enum, ignore_case = true)] method: Option, - /// Filter HTTP logs by status code, or network flow logs by status (ok, dropped) + /// Filter HTTP logs by status code, network flow logs by status (ok, dropped), or DNS logs by resolution status (ok, failed) #[clap(long, value_name = "STATUS")] status: Option, @@ -548,6 +623,26 @@ Examples: #[clap(long = "drop-cause", requires = "network", value_name = "CAUSE")] drop_cause: Option, + /// Filter DNS logs by domain, including its subdomains + #[clap(long, requires = "dns", value_name = "DOMAIN")] + domain: Option, + + /// Filter DNS logs by the exact name looked up + #[clap(long, requires = "dns", value_name = "NAME")] + qname: Option, + + /// Filter DNS logs by record type (e.g. A, AAAA, CNAME, TXT) + #[clap(long, requires = "dns", value_name = "TYPE")] + qtype: Option, + + /// Filter DNS logs by response code (e.g. NOERROR, NXDOMAIN, SERVFAIL) + #[clap(long, requires = "dns", value_name = "RCODE")] + rcode: Option, + + /// Filter DNS logs by lookup zone + #[clap(long, requires = "dns", value_enum, ignore_case = true)] + zone: Option, + /// Always show logs from the latest deployment, even if it failed or is still building #[clap(long)] latest: bool, @@ -579,6 +674,11 @@ pub async fn command(args: Args) -> Result<()> { } else { None }; + let dns_filter = if args.dns { + build_dns_filter(&args)? + } else { + None + }; let start_date = args.since.as_ref().map(|s| parse_time(s)).transpose()?; let end_date = args.until.as_ref().map(|s| parse_time(s)).transpose()?; @@ -643,6 +743,50 @@ pub async fn command(args: Args) -> Result<()> { return Ok(()); } + if args.dns { + if args.deployment_id.is_some() || args.latest { + bail!( + "deployment IDs and --latest can only be used with deployment, build, or HTTP logs" + ); + } + + if !args.json { + println!("{}", format_dns_query_log_header()); + } + + if should_stream { + stream_dns_query_logs(environment_id, Some(service), dns_filter, |log| { + print_dns_query_log(log, args.json) + }) + .await?; + } else { + let mut has_logs = false; + fetch_dns_query_logs( + FetchDnsQueryLogsParams { + client: &client, + backboard: &backboard, + environment_id, + service_id: Some(service), + limit: args.lines.or(Some(500)), + filter: dns_filter, + start_date, + end_date, + }, + |log| { + has_logs = true; + print_dns_query_log(log, args.json) + }, + ) + .await?; + + if !has_logs && !args.json { + println!("No DNS queries found"); + } + } + + return Ok(()); + } + // Fetch all deployments so we can find a sensible default deployment id if // none is provided let vars = queries::deployments::Variables { @@ -767,6 +911,7 @@ mod tests { build: false, http: false, network: false, + dns: false, deployment_id: None, json: false, lines: None, @@ -785,6 +930,11 @@ mod tests { dst: None, host: None, drop_cause: None, + domain: None, + qname: None, + qtype: None, + rcode: None, + zone: None, latest: false, since: None, until: None, @@ -877,6 +1027,86 @@ mod tests { assert!(Args::try_parse_from(["logs", "--network", "--deployment"]).is_err()); } + #[test] + fn build_dns_filter_composes_typed_and_raw() { + let args = Args::parse_from(["logs", "--dns"]); + assert_eq!(build_dns_filter(&args).unwrap(), None); + + let args = Args::parse_from([ + "logs", + "--dns", + "--domain", + "example.com", + "--qname", + "api.example.com", + "--qtype", + "aaaa", + "--rcode", + "nxdomain", + "--zone", + "external", + "--status", + "FAILED", + "--filter", + "@answers:10.0.0.1", + ]); + assert_eq!( + build_dns_filter(&args).unwrap(), + Some( + "@domain:example.com @qname:api.example.com @qtype:AAAA @rcode:NXDOMAIN @zone:external @status:failed @answers:10.0.0.1" + .to_string() + ) + ); + } + + #[test] + fn build_dns_filter_rejects_invalid_status() { + let args = Args::parse_from(["logs", "--dns", "--status", "dropped"]); + assert!(build_dns_filter(&args).is_err()); + } + + #[test] + fn dns_logs_are_mutually_exclusive_with_other_log_modes() { + assert!(Args::try_parse_from(["logs", "--dns", "--http"]).is_err()); + assert!(Args::try_parse_from(["logs", "--dns", "--build"]).is_err()); + assert!(Args::try_parse_from(["logs", "--dns", "--deployment"]).is_err()); + assert!(Args::try_parse_from(["logs", "--dns", "--network"]).is_err()); + } + + #[test] + fn dns_typed_flags_require_dns_mode() { + assert!(Args::try_parse_from(["logs", "--rcode", "NXDOMAIN"]).is_err()); + + let mut args = base_args(); + args.http = true; + args.rcode = Some("NXDOMAIN".to_string()); + assert!(validate_filter_modes(&args).is_err()); + + let mut args = base_args(); + args.dns = true; + args.rcode = Some("NXDOMAIN".to_string()); + assert!(validate_filter_modes(&args).is_ok()); + } + + #[test] + fn build_dns_filter_accepts_uncommon_qtype_and_rcode_values() { + let args = Args::parse_from(["logs", "--dns", "--qtype", "ptr", "--rcode", "formerr"]); + + assert_eq!( + build_dns_filter(&args).unwrap(), + Some("@qtype:PTR @rcode:FORMERR".to_string()) + ); + } + + #[test] + fn status_accepts_dns_mode() { + let mut args = base_args(); + args.dns = true; + args.status = Some("failed".to_string()); + + assert!(validate_filter_modes(&args).is_ok()); + } + #[test] fn network_flow_typed_flags_require_network_mode() { let mut args = base_args(); diff --git a/src/controllers/deployment.rs b/src/controllers/deployment.rs index c131c2e15..eef3116a3 100644 --- a/src/controllers/deployment.rs +++ b/src/controllers/deployment.rs @@ -2,7 +2,8 @@ use crate::{ commands::{ queries::{self}, subscriptions::{ - self, build_logs, deployment, deployment_logs, http_logs, network_flow_logs, + self, build_logs, deployment, deployment_logs, dns_query_logs, http_logs, + network_flow_logs, }, }, post_graphql, @@ -27,10 +28,10 @@ const LOGS_RETRY_CONFIG: RetryConfig = RetryConfig { const HTTP_LOG_STREAM_AFTER_WINDOW: Duration = Duration::from_secs(60 * 60); const HTTP_LOG_STREAM_BATCH_SIZE: i64 = 500; -const HTTP_LOG_STREAM_STABLE_CONNECTION_DURATION: Duration = Duration::from_secs(30); -const NETWORK_FLOW_LOG_DEFAULT_LIMIT: i64 = 500; -const NETWORK_FLOW_STREAM_LOOKBACK_SECONDS: i64 = 30; -const NETWORK_FLOW_STREAM_DEDUPE_CACHE_SIZE: usize = 10_000; +const STREAM_STABLE_CONNECTION_DURATION: Duration = Duration::from_secs(30); +const ANCHORED_LOG_DEFAULT_LIMIT: i64 = 500; +const STREAM_LOOKBACK_SECONDS: i64 = 30; +const STREAM_DEDUPE_CACHE_SIZE: usize = 10_000; pub struct FetchLogsParams<'a> { pub client: &'a Client, @@ -53,6 +54,17 @@ pub struct FetchNetworkFlowLogsParams<'a> { pub end_date: Option>, } +pub struct FetchDnsQueryLogsParams<'a> { + pub client: &'a Client, + pub backboard: &'a str, + pub environment_id: String, + pub service_id: Option, + pub limit: Option, + pub filter: Option, + pub start_date: Option>, + pub end_date: Option>, +} + // Helper to handle the API's off-by-one bug where it returns limit+1 logs fn take_last_n_logs(mut logs: Vec, limit: Option) -> Vec { if let Some(l) = limit { @@ -66,7 +78,7 @@ fn take_last_n_logs(mut logs: Vec, limit: Option) -> Vec { } #[derive(Debug, PartialEq, Eq)] -struct NetworkFlowLogWindow { +struct AnchoredLogWindow { before_limit: Option, before_date: Option, anchor_date: Option, @@ -74,44 +86,44 @@ struct NetworkFlowLogWindow { after_limit: Option, } -fn format_network_flow_log_timestamp(date: DateTime) -> String { +fn format_anchored_log_timestamp(date: DateTime) -> String { date.to_rfc3339_opts(SecondsFormat::Nanos, true) } -fn network_flow_log_window( +fn anchored_log_window( limit: Option, start_date: Option>, end_date: Option>, now: DateTime, -) -> NetworkFlowLogWindow { - let before_limit = Some(limit.unwrap_or(NETWORK_FLOW_LOG_DEFAULT_LIMIT)); +) -> AnchoredLogWindow { + let before_limit = Some(limit.unwrap_or(ANCHORED_LOG_DEFAULT_LIMIT)); match (start_date, end_date) { - (Some(start), Some(end)) => NetworkFlowLogWindow { + (Some(start), Some(end)) => AnchoredLogWindow { before_limit, - before_date: Some(format_network_flow_log_timestamp(start)), - anchor_date: Some(format_network_flow_log_timestamp(end)), - after_date: Some(format_network_flow_log_timestamp(end)), + before_date: Some(format_anchored_log_timestamp(start)), + anchor_date: Some(format_anchored_log_timestamp(end)), + after_date: Some(format_anchored_log_timestamp(end)), after_limit: Some(0), }, - (Some(start), None) => NetworkFlowLogWindow { + (Some(start), None) => AnchoredLogWindow { before_limit, - before_date: Some(format_network_flow_log_timestamp(start)), - anchor_date: Some(format_network_flow_log_timestamp(now)), - after_date: Some(format_network_flow_log_timestamp(now)), + before_date: Some(format_anchored_log_timestamp(start)), + anchor_date: Some(format_anchored_log_timestamp(now)), + after_date: Some(format_anchored_log_timestamp(now)), after_limit: Some(0), }, - (None, Some(end)) => NetworkFlowLogWindow { + (None, Some(end)) => AnchoredLogWindow { before_limit, - before_date: Some(format_network_flow_log_timestamp( + before_date: Some(format_anchored_log_timestamp( DateTime::::from_timestamp(0, 0) .expect("Unix epoch should be a valid timestamp"), )), - anchor_date: Some(format_network_flow_log_timestamp(end)), - after_date: Some(format_network_flow_log_timestamp(end)), + anchor_date: Some(format_anchored_log_timestamp(end)), + after_date: Some(format_anchored_log_timestamp(end)), after_limit: Some(0), }, - (None, None) => NetworkFlowLogWindow { + (None, None) => AnchoredLogWindow { before_limit, before_date: None, anchor_date: None, @@ -205,8 +217,7 @@ pub async fn fetch_network_flow_logs( params: FetchNetworkFlowLogsParams<'_>, mut on_log: impl FnMut(queries::network_flow_logs::NetworkFlowLogFields), ) -> Result<()> { - let window = - network_flow_log_window(params.limit, params.start_date, params.end_date, Utc::now()); + let window = anchored_log_window(params.limit, params.start_date, params.end_date, Utc::now()); let vars = queries::network_flow_logs::Variables { environment_id: params.environment_id, service_id: params.service_id, @@ -230,6 +241,34 @@ pub async fn fetch_network_flow_logs( Ok(()) } +pub async fn fetch_dns_query_logs( + params: FetchDnsQueryLogsParams<'_>, + mut on_log: impl FnMut(queries::dns_query_logs::DnsQueryLogFields), +) -> Result<()> { + let window = anchored_log_window(params.limit, params.start_date, params.end_date, Utc::now()); + let vars = queries::dns_query_logs::Variables { + environment_id: params.environment_id, + service_id: params.service_id, + filter: params.filter, + before_limit: window.before_limit, + before_date: window.before_date, + anchor_date: window.anchor_date, + after_date: window.after_date, + after_limit: window.after_limit, + }; + + let response = + post_graphql::(params.client, params.backboard, vars).await?; + + let logs = take_last_n_logs(response.dns_query_logs, window.before_limit); + + for log in logs { + on_log(log); + } + + Ok(()) +} + pub async fn stream_build_logs( deployment_id: String, filter: Option, @@ -311,31 +350,43 @@ pub async fn stream_http_logs( } } -pub async fn stream_network_flow_logs( - environment_id: String, - service_id: Option, - filter: Option, - mut on_log: impl FnMut(network_flow_logs::NetworkFlowLogFields), -) -> Result<()> { - let mut max_capture_end: Option> = None; - let mut seen_flow_ids = NetworkFlowLogDedupe::new(NETWORK_FLOW_STREAM_DEDUPE_CACHE_SIZE); +struct StreamMessages { + stream_error: &'static str, + data_error: &'static str, + closed_without_events: &'static str, +} + +/// Drives an anchored log subscription (network flow, DNS) with the same +/// reconnect policy as `stream_http_logs_inner`: mid-stream errors are retried +/// with backoff, every reconnect waits at least the initial delay, and retry +/// state only resets once a connection proves stable. Each reconnect re-anchors +/// `beforeDate` to the newest timestamp seen minus a lookback window; rows at +/// or before the previous connection's watermark are flagged as replays via +/// `on_line`'s second argument. The callback returns whether the row was +/// emitted so replay-only connections do not reset retry state. +async fn stream_anchored_logs( + mut build_vars: impl FnMut(String) -> Q::Variables, + extract_lines: impl Fn(Q::ResponseData) -> Vec, + line_timestamp: impl Fn(&Line) -> &str, + mut on_line: impl FnMut(Line, Option>) -> bool, + messages: StreamMessages, +) -> Result<()> +where + Q: graphql_client::GraphQLQuery + Send + Sync + Unpin + 'static, + Q::Variables: Send + Sync + Unpin, + Q::ResponseData: std::fmt::Debug, +{ + let mut max_timestamp: Option> = None; + let mut replay_cutoff: Option> = None; let mut attempt = 0; let mut delay_ms = LOGS_RETRY_CONFIG.initial_delay_ms; loop { - let before_date = network_flow_stream_before_date(max_capture_end); - let vars = subscriptions::network_flow_logs::Variables { - environment_id: environment_id.clone(), - service_id: service_id.clone(), - filter: filter.clone(), - before_limit: Some(NETWORK_FLOW_LOG_DEFAULT_LIMIT), - before_date: Some(before_date), - anchor_date: None, - after_date: None, - after_limit: Some(0), - }; + let connected_at = Instant::now(); + let mut emitted_any_logs = false; + let vars = build_vars(stream_before_date(max_timestamp)); - let mut stream = match subscribe_graphql::(vars).await { + let mut stream = match subscribe_graphql::(vars).await { Ok(stream) => stream, Err(e) => { attempt += 1; @@ -351,35 +402,174 @@ pub async fn stream_network_flow_logs( } }; - attempt = 0; - delay_ms = LOGS_RETRY_CONFIG.initial_delay_ms; + let result = async { + while let Some(response) = stream.next().await { + let log = response + .context(messages.stream_error)? + .data + .context(messages.data_error)?; + + emitted_any_logs |= process_anchored_stream_lines( + extract_lines(log), + &line_timestamp, + replay_cutoff, + &mut max_timestamp, + &mut on_line, + ); + } + + Ok::<(), anyhow::Error>(()) + } + .await; + + replay_cutoff = max_timestamp; - while let Some(response) = stream.next().await { - let log = response - .context("Network flow log stream error")? - .data - .context("Failed to retrieve network flow logs")?; + let should_reset_retry_state = + should_reset_stream_retry_state(emitted_any_logs, connected_at.elapsed()); - for line in log.network_flow_logs { - update_max_network_flow_capture_end(&line.capture_end, &mut max_capture_end); + if should_reset_retry_state { + attempt = 0; + delay_ms = LOGS_RETRY_CONFIG.initial_delay_ms; + } else { + attempt += 1; - if !seen_flow_ids.insert(line.flow_id.clone()) { - continue; + match result { + Err(e) if attempt >= LOGS_RETRY_CONFIG.max_attempts => return Err(e), + Ok(()) if attempt >= LOGS_RETRY_CONFIG.max_attempts => { + return Err(anyhow!(messages.closed_without_events)); } - - on_log(line); + _ => {} } } + + sleep(Duration::from_millis(delay_ms)).await; + + if !should_reset_retry_state { + delay_ms = ((delay_ms as f64 * LOGS_RETRY_CONFIG.backoff_multiplier) as u64) + .min(LOGS_RETRY_CONFIG.max_delay_ms); + } + } +} + +fn process_anchored_stream_lines( + lines: Vec, + line_timestamp: &impl Fn(&Line) -> &str, + replay_cutoff: Option>, + max_timestamp: &mut Option>, + on_line: &mut impl FnMut(Line, Option>) -> bool, +) -> bool { + let mut emitted_any_logs = false; + + for line in lines { + update_max_stream_timestamp(line_timestamp(&line), max_timestamp); + emitted_any_logs |= on_line(line, replay_cutoff); } + + emitted_any_logs } -struct NetworkFlowLogDedupe { +pub async fn stream_network_flow_logs( + environment_id: String, + service_id: Option, + filter: Option, + mut on_log: impl FnMut(network_flow_logs::NetworkFlowLogFields), +) -> Result<()> { + let mut seen_flow_ids = StreamedLogDedupe::new(STREAM_DEDUPE_CACHE_SIZE); + + stream_anchored_logs::( + |before_date| subscriptions::network_flow_logs::Variables { + environment_id: environment_id.clone(), + service_id: service_id.clone(), + filter: filter.clone(), + before_limit: Some(ANCHORED_LOG_DEFAULT_LIMIT), + before_date: Some(before_date), + anchor_date: None, + after_date: None, + after_limit: Some(0), + }, + |data| data.network_flow_logs, + |line: &network_flow_logs::NetworkFlowLogFields| line.capture_end.as_str(), + |line, _replay_cutoff| { + // Flow IDs are unique, so any repeated ID is a replay regardless + // of when it arrives + if seen_flow_ids.insert(line.flow_id.clone()) { + on_log(line); + true + } else { + false + } + }, + StreamMessages { + stream_error: "Network flow log stream error", + data_error: "Failed to retrieve network flow logs", + closed_without_events: "Network flow log stream closed before receiving any events", + }, + ) + .await +} + +pub async fn stream_dns_query_logs( + environment_id: String, + service_id: Option, + filter: Option, + mut on_log: impl FnMut(dns_query_logs::DnsQueryLogFields), +) -> Result<()> { + // DNS query logs carry no unique row ID, so replays after a reconnect are + // recognized by their full serialized contents. Only rows at or before the + // reconnect watermark are eligible for deduplication: identical queries + // arriving live on a healthy connection are all emitted. + let mut seen_rows = StreamedLogDedupe::new(STREAM_DEDUPE_CACHE_SIZE); + + stream_anchored_logs::( + |before_date| subscriptions::dns_query_logs::Variables { + environment_id: environment_id.clone(), + service_id: service_id.clone(), + filter: filter.clone(), + before_limit: Some(ANCHORED_LOG_DEFAULT_LIMIT), + before_date: Some(before_date), + anchor_date: None, + after_date: None, + after_limit: Some(0), + }, + |data| data.dns_query_logs, + |line: &dns_query_logs::DnsQueryLogFields| line.queried_at.as_str(), + |line, replay_cutoff| { + if let Ok(row_key) = serde_json::to_string(&line) { + let already_seen = !seen_rows.insert(row_key); + if already_seen && is_replayed_row(&line.queried_at, replay_cutoff) { + return false; + } + } + + on_log(line); + true + }, + StreamMessages { + stream_error: "DNS query log stream error", + data_error: "Failed to retrieve DNS query logs", + closed_without_events: "DNS query log stream closed before receiving any events", + }, + ) + .await +} + +fn is_replayed_row(timestamp: &str, replay_cutoff: Option>) -> bool { + let Some(cutoff) = replay_cutoff else { + return false; + }; + + DateTime::parse_from_rfc3339(timestamp) + .map(|ts| ts.with_timezone(&Utc) <= cutoff) + .unwrap_or(false) +} + +struct StreamedLogDedupe { seen: HashSet, order: VecDeque, max_size: usize, } -impl NetworkFlowLogDedupe { +impl StreamedLogDedupe { fn new(max_size: usize) -> Self { Self { seen: HashSet::new(), @@ -406,25 +596,20 @@ impl NetworkFlowLogDedupe { } } -fn network_flow_stream_before_date(max_capture_end: Option>) -> String { - let anchor = max_capture_end.unwrap_or_else(Utc::now); - format_network_flow_log_timestamp( - anchor - ChronoDuration::seconds(NETWORK_FLOW_STREAM_LOOKBACK_SECONDS), - ) +fn stream_before_date(max_timestamp: Option>) -> String { + let anchor = max_timestamp.unwrap_or_else(Utc::now); + format_anchored_log_timestamp(anchor - ChronoDuration::seconds(STREAM_LOOKBACK_SECONDS)) } -fn update_max_network_flow_capture_end( - capture_end: &str, - max_capture_end: &mut Option>, -) { - let Ok(capture_end) = - DateTime::parse_from_rfc3339(capture_end).map(|date| date.with_timezone(&Utc)) +fn update_max_stream_timestamp(timestamp: &str, max_timestamp: &mut Option>) { + let Ok(timestamp) = + DateTime::parse_from_rfc3339(timestamp).map(|date| date.with_timezone(&Utc)) else { return; }; - if max_capture_end.is_none_or(|max| capture_end > max) { - *max_capture_end = Some(capture_end); + if max_timestamp.is_none_or(|max| timestamp > max) { + *max_timestamp = Some(timestamp); } } @@ -524,7 +709,7 @@ async fn stream_http_logs_inner( .await; let should_reset_retry_state = - should_reset_http_stream_retry_state(received_any_logs, connected_at.elapsed()); + should_reset_stream_retry_state(received_any_logs, connected_at.elapsed()); if should_reset_retry_state { attempt = 0; @@ -552,11 +737,8 @@ async fn stream_http_logs_inner( } } -fn should_reset_http_stream_retry_state( - received_any_logs: bool, - connection_duration: Duration, -) -> bool { - received_any_logs || connection_duration >= HTTP_LOG_STREAM_STABLE_CONNECTION_DURATION +fn should_reset_stream_retry_state(received_any_logs: bool, connection_duration: Duration) -> bool { + received_any_logs || connection_duration >= STREAM_STABLE_CONNECTION_DURATION } fn is_new_http_log( @@ -723,11 +905,11 @@ mod tests { let end = dt("2026-06-18T04:42:00Z"); let now = dt("2026-06-18T04:43:00Z"); - let window = network_flow_log_window(Some(100), Some(start), Some(end), now); + let window = anchored_log_window(Some(100), Some(start), Some(end), now); assert_eq!( window, - NetworkFlowLogWindow { + AnchoredLogWindow { before_limit: Some(100), before_date: Some("2026-06-18T04:41:00.000000000Z".to_string()), anchor_date: Some("2026-06-18T04:42:00.000000000Z".to_string()), @@ -743,11 +925,11 @@ mod tests { let end = dt("2026-06-18T04:42:00Z"); let now = dt("2026-06-18T04:43:00Z"); - let since_window = network_flow_log_window(None, Some(start), None, now); + let since_window = anchored_log_window(None, Some(start), None, now); assert_eq!( since_window, - NetworkFlowLogWindow { - before_limit: Some(NETWORK_FLOW_LOG_DEFAULT_LIMIT), + AnchoredLogWindow { + before_limit: Some(ANCHORED_LOG_DEFAULT_LIMIT), before_date: Some("2026-06-18T04:41:00.000000000Z".to_string()), anchor_date: Some("2026-06-18T04:43:00.000000000Z".to_string()), after_date: Some("2026-06-18T04:43:00.000000000Z".to_string()), @@ -755,11 +937,11 @@ mod tests { } ); - let until_window = network_flow_log_window(None, None, Some(end), now); + let until_window = anchored_log_window(None, None, Some(end), now); assert_eq!( until_window, - NetworkFlowLogWindow { - before_limit: Some(NETWORK_FLOW_LOG_DEFAULT_LIMIT), + AnchoredLogWindow { + before_limit: Some(ANCHORED_LOG_DEFAULT_LIMIT), before_date: Some("1970-01-01T00:00:00.000000000Z".to_string()), anchor_date: Some("2026-06-18T04:42:00.000000000Z".to_string()), after_date: Some("2026-06-18T04:42:00.000000000Z".to_string()), @@ -772,11 +954,11 @@ mod tests { fn test_network_flow_log_window_leaves_unbounded_snapshot_to_api_defaults() { let now = dt("2026-06-18T04:43:00Z"); - let window = network_flow_log_window(Some(20), None, None, now); + let window = anchored_log_window(Some(20), None, None, now); assert_eq!( window, - NetworkFlowLogWindow { + AnchoredLogWindow { before_limit: Some(20), before_date: None, anchor_date: None, @@ -788,7 +970,7 @@ mod tests { #[test] fn test_network_flow_dedupe_keeps_out_of_order_sibling_flows() { - let mut dedupe = NetworkFlowLogDedupe::new(10); + let mut dedupe = StreamedLogDedupe::new(10); assert!(dedupe.insert("newer-flow".to_string())); assert!(dedupe.insert("older-sibling-flow".to_string())); @@ -797,7 +979,7 @@ mod tests { #[test] fn test_network_flow_dedupe_bounds_cache_size() { - let mut dedupe = NetworkFlowLogDedupe::new(2); + let mut dedupe = StreamedLogDedupe::new(2); assert!(dedupe.insert("flow-1".to_string())); assert!(dedupe.insert("flow-2".to_string())); @@ -807,7 +989,7 @@ mod tests { #[test] fn test_network_flow_stream_before_date_uses_lookback() { - let before_date = network_flow_stream_before_date(Some(dt("2026-06-18T04:43:00Z"))); + let before_date = stream_before_date(Some(dt("2026-06-18T04:43:00Z"))); assert_eq!(before_date, "2026-06-18T04:42:30.000000000Z"); } @@ -816,10 +998,10 @@ mod tests { fn test_update_max_network_flow_capture_end_ignores_older_rows() { let mut max_capture_end = Some(dt("2026-06-18T04:43:00Z")); - update_max_network_flow_capture_end("2026-06-18T04:42:30Z", &mut max_capture_end); + update_max_stream_timestamp("2026-06-18T04:42:30Z", &mut max_capture_end); assert_eq!(max_capture_end, Some(dt("2026-06-18T04:43:00Z"))); - update_max_network_flow_capture_end("2026-06-18T04:43:30Z", &mut max_capture_end); + update_max_stream_timestamp("2026-06-18T04:43:30Z", &mut max_capture_end); assert_eq!(max_capture_end, Some(dt("2026-06-18T04:43:30Z"))); } @@ -864,26 +1046,71 @@ mod tests { } #[test] - fn test_should_reset_http_stream_retry_state_after_logs() { - assert!(should_reset_http_stream_retry_state( + fn test_should_reset_stream_retry_state_after_logs() { + assert!(should_reset_stream_retry_state( true, Duration::from_secs(1), )); } #[test] - fn test_should_reset_http_stream_retry_state_after_stable_connection() { - assert!(should_reset_http_stream_retry_state( + fn test_should_reset_stream_retry_state_after_stable_connection() { + assert!(should_reset_stream_retry_state( false, - HTTP_LOG_STREAM_STABLE_CONNECTION_DURATION, + STREAM_STABLE_CONNECTION_DURATION, )); } #[test] - fn test_should_not_reset_http_stream_retry_state_for_short_empty_connection() { - assert!(!should_reset_http_stream_retry_state( + fn test_should_not_reset_stream_retry_state_for_short_empty_connection() { + assert!(!should_reset_stream_retry_state( false, Duration::from_secs(1), )); } + + #[test] + fn test_replay_only_anchored_rows_do_not_reset_retry_state() { + let replay_cutoff = Some(dt("2026-06-18T04:43:00Z")); + let mut max_timestamp = replay_cutoff; + let mut seen_rows = HashSet::from(["replayed-row".to_string()]); + + let emitted_any_logs = process_anchored_stream_lines( + vec![( + "replayed-row".to_string(), + "2026-06-18T04:43:00Z".to_string(), + )], + &|line: &(String, String)| line.1.as_str(), + replay_cutoff, + &mut max_timestamp, + &mut |line, cutoff| { + let already_seen = !seen_rows.insert(line.0); + !(already_seen && is_replayed_row(&line.1, cutoff)) + }, + ); + + assert!(!emitted_any_logs); + assert!(!should_reset_stream_retry_state( + emitted_any_logs, + Duration::from_secs(1), + )); + } + + #[test] + fn test_is_replayed_row_only_flags_rows_at_or_before_cutoff() { + let cutoff = Some(dt("2026-06-18T04:43:00Z")); + + assert!(is_replayed_row("2026-06-18T04:42:59Z", cutoff)); + assert!(is_replayed_row("2026-06-18T04:43:00Z", cutoff)); + assert!(!is_replayed_row("2026-06-18T04:43:01Z", cutoff)); + } + + #[test] + fn test_is_replayed_row_fails_open_without_cutoff_or_valid_timestamp() { + assert!(!is_replayed_row("2026-06-18T04:42:59Z", None)); + assert!(!is_replayed_row( + "not-a-timestamp", + Some(dt("2026-06-18T04:43:00Z")) + )); + } } diff --git a/src/gql/queries/mod.rs b/src/gql/queries/mod.rs index 781e6deeb..91013606b 100644 --- a/src/gql/queries/mod.rs +++ b/src/gql/queries/mod.rs @@ -113,6 +113,14 @@ pub struct HttpLogs; )] pub struct NetworkFlowLogs; +#[derive(GraphQLQuery)] +#[graphql( + schema_path = "src/gql/schema.json", + query_path = "src/gql/queries/strings/DnsQueryLogs.graphql", + response_derives = "Debug, Serialize, Clone" +)] +pub struct DnsQueryLogs; + #[derive(GraphQLQuery)] #[graphql( schema_path = "src/gql/schema.json", diff --git a/src/gql/queries/strings/DnsQueryLogs.graphql b/src/gql/queries/strings/DnsQueryLogs.graphql new file mode 100644 index 000000000..bf9300f3e --- /dev/null +++ b/src/gql/queries/strings/DnsQueryLogs.graphql @@ -0,0 +1,36 @@ +query DnsQueryLogs( + $environmentId: String! + $serviceId: String + $filter: String + $beforeLimit: Int + $beforeDate: String + $anchorDate: String + $afterDate: String + $afterLimit: Int +) { + dnsQueryLogs( + environmentId: $environmentId + serviceId: $serviceId + filter: $filter + beforeDate: $beforeDate + anchorDate: $anchorDate + afterDate: $afterDate + beforeLimit: $beforeLimit + afterLimit: $afterLimit + ) { + ...DnsQueryLogFields + } +} + +fragment DnsQueryLogFields on DnsQueryLog { + queriedAt + qname + qtype + rcode + queryZone + answers + cnameChain + serviceId + deploymentId + deploymentInstanceId +} diff --git a/src/gql/schema.json b/src/gql/schema.json index 1e992ea01..10b0b314a 100644 --- a/src/gql/schema.json +++ b/src/gql/schema.json @@ -7145,6 +7145,216 @@ "name": "DisplayConfig", "possibleTypes": null }, + { + "description": "A single DNS query resolved for a service (one row per query, not aggregated)", + "enumValues": null, + "fields": [ + { + "args": [], + "deprecationReason": null, + "description": "The IP addresses the name resolved to", + "isDeprecated": false, + "name": "answers", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "LIST", + "name": null, + "ofType": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + } + } + }, + { + "args": [], + "deprecationReason": null, + "description": "Ordered CNAME targets the queried name aliased through before the final answer (empty when the response had no CNAME)", + "isDeprecated": false, + "name": "cnameChain", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "LIST", + "name": null, + "ofType": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + } + } + }, + { + "args": [], + "deprecationReason": null, + "description": "The deployment ID", + "isDeprecated": false, + "name": "deploymentId", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + }, + { + "args": [], + "deprecationReason": null, + "description": "The deployment instance ID", + "isDeprecated": false, + "name": "deploymentInstanceId", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + }, + { + "args": [], + "deprecationReason": null, + "description": "The domain name that was looked up", + "isDeprecated": false, + "name": "qname", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + }, + { + "args": [], + "deprecationReason": null, + "description": "The DNS record type queried (A, AAAA, TXT, ...)", + "isDeprecated": false, + "name": "qtype", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + }, + { + "args": [], + "deprecationReason": null, + "description": "When the query was resolved (ISO timestamp)", + "isDeprecated": false, + "name": "queriedAt", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + }, + { + "args": [], + "deprecationReason": null, + "description": "Whether the query targeted the internal zone or the public internet", + "isDeprecated": false, + "name": "queryZone", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "ENUM", + "name": "DnsQueryZone", + "ofType": null + } + } + }, + { + "args": [], + "deprecationReason": null, + "description": "The DNS response code (NOERROR, NXDOMAIN, SERVFAIL, ...) or synthetic TIMEOUT/ERROR when no upstream response was received", + "isDeprecated": false, + "name": "rcode", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + }, + { + "args": [], + "deprecationReason": null, + "description": "The service ID that made the queries", + "isDeprecated": false, + "name": "serviceId", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + } + ], + "inputFields": null, + "interfaces": [], + "kind": "OBJECT", + "name": "DnsQueryLog", + "possibleTypes": null + }, + { + "description": "Whether a DNS query targeted the internal zone or the public internet", + "enumValues": [ + { + "deprecationReason": null, + "description": null, + "isDeprecated": false, + "name": "external" + }, + { + "deprecationReason": null, + "description": null, + "isDeprecated": false, + "name": "internal" + } + ], + "fields": null, + "inputFields": null, + "interfaces": null, + "kind": "ENUM", + "name": "DnsQueryZone", + "possibleTypes": null + }, { "description": null, "enumValues": null, @@ -31354,6 +31564,125 @@ } } }, + { + "args": [ + { + "defaultValue": null, + "description": "Latest date to look for logs after the anchor", + "name": "afterDate", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Limit logs returned after the anchor", + "name": "afterLimit", + "type": { + "kind": "SCALAR", + "name": "Int", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Target date time to look for logs", + "name": "anchorDate", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Oldest date to look for logs before the anchor", + "name": "beforeDate", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Limit logs returned before the anchor", + "name": "beforeLimit", + "type": { + "kind": "SCALAR", + "name": "Int", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Filter by deployment instance / sandbox ID (optional)", + "name": "deploymentInstanceId", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + }, + { + "defaultValue": null, + "description": null, + "name": "environmentId", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + }, + { + "defaultValue": null, + "description": "Filter expression (e.g., @rcode:NXDOMAIN @qtype:A @domain:example.com @status:failed). Prefix a term with - to exclude it; repeat a key to match any of its values", + "name": "filter", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Filter by service ID (optional)", + "name": "serviceId", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + ], + "deprecationReason": null, + "description": "Fetch individual DNS query logs for an environment", + "isDeprecated": false, + "name": "dnsQueryLogs", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "LIST", + "name": null, + "ofType": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "OBJECT", + "name": "DnsQueryLog", + "ofType": null + } + } + } + } + }, { "args": [ { @@ -35038,7 +35367,7 @@ } ], "deprecationReason": null, - "description": "Resumable sessions inside a sandbox — interactive shells and one-off exec commands, live or recently exited. Null when the sandbox can't report them (e.g. its vm-init predates session listing).", + "description": "Resumable sessions inside a sandbox \u2014 interactive shells and one-off exec commands, live or recently exited. Null when the sandbox can't report them (e.g. its vm-init predates session listing).", "isDeprecated": false, "name": "sandboxSessions", "type": { @@ -44762,6 +45091,125 @@ } } }, + { + "args": [ + { + "defaultValue": null, + "description": "Latest date to look for logs after the anchor", + "name": "afterDate", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Limit logs returned after the anchor", + "name": "afterLimit", + "type": { + "kind": "SCALAR", + "name": "Int", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Target date time to look for logs", + "name": "anchorDate", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Oldest date to look for logs before the anchor", + "name": "beforeDate", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Limit logs returned before the anchor", + "name": "beforeLimit", + "type": { + "kind": "SCALAR", + "name": "Int", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Filter by deployment instance / sandbox ID (optional)", + "name": "deploymentInstanceId", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + }, + { + "defaultValue": null, + "description": null, + "name": "environmentId", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + }, + { + "defaultValue": null, + "description": "Filter expression (e.g., @rcode:NXDOMAIN @qtype:A @domain:example.com @status:failed). Prefix a term with - to exclude it; repeat a key to match any of its values", + "name": "filter", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + }, + { + "defaultValue": null, + "description": "Filter by service ID (optional)", + "name": "serviceId", + "type": { + "kind": "SCALAR", + "name": "String", + "ofType": null + } + } + ], + "deprecationReason": null, + "description": "Stream individual DNS query logs for an environment", + "isDeprecated": false, + "name": "dnsQueryLogs", + "type": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "LIST", + "name": null, + "ofType": { + "kind": "NON_NULL", + "name": null, + "ofType": { + "kind": "OBJECT", + "name": "DnsQueryLog", + "ofType": null + } + } + } + } + }, { "args": [ { @@ -55420,4 +55868,4 @@ ] } } -} \ No newline at end of file +} diff --git a/src/gql/subscriptions/mod.rs b/src/gql/subscriptions/mod.rs index fb29fc609..f6cadcdf3 100644 --- a/src/gql/subscriptions/mod.rs +++ b/src/gql/subscriptions/mod.rs @@ -32,6 +32,14 @@ pub struct HttpLogs; )] pub struct NetworkFlowLogs; +#[derive(GraphQLQuery)] +#[graphql( + schema_path = "src/gql/schema.json", + query_path = "src/gql/subscriptions/strings/DnsQueryLogs.graphql", + response_derives = "Debug, Serialize, Clone" +)] +pub struct DnsQueryLogs; + #[derive(GraphQLQuery)] #[graphql( schema_path = "src/gql/schema.json", diff --git a/src/gql/subscriptions/strings/DnsQueryLogs.graphql b/src/gql/subscriptions/strings/DnsQueryLogs.graphql new file mode 100644 index 000000000..d81fa6691 --- /dev/null +++ b/src/gql/subscriptions/strings/DnsQueryLogs.graphql @@ -0,0 +1,36 @@ +subscription DnsQueryLogs( + $environmentId: String! + $serviceId: String + $filter: String + $beforeLimit: Int + $beforeDate: String + $anchorDate: String + $afterDate: String + $afterLimit: Int +) { + dnsQueryLogs( + environmentId: $environmentId + serviceId: $serviceId + filter: $filter + beforeDate: $beforeDate + anchorDate: $anchorDate + afterDate: $afterDate + beforeLimit: $beforeLimit + afterLimit: $afterLimit + ) { + ...DnsQueryLogFields + } +} + +fragment DnsQueryLogFields on DnsQueryLog { + queriedAt + qname + qtype + rcode + queryZone + answers + cnameChain + serviceId + deploymentId + deploymentInstanceId +} diff --git a/src/util/logs.rs b/src/util/logs.rs index 1c929df55..17ee83d47 100644 --- a/src/util/logs.rs +++ b/src/util/logs.rs @@ -13,6 +13,12 @@ const NETWORK_FLOW_TRAFFIC_WIDTH: usize = 10; const NETWORK_FLOW_LATENCY_WIDTH: usize = 8; const NETWORK_FLOW_STATUS_WIDTH: usize = 16; +const DNS_QUERY_TIME_WIDTH: usize = 30; +const DNS_QUERY_ZONE_WIDTH: usize = 8; +const DNS_QUERY_TYPE_WIDTH: usize = 5; +const DNS_QUERY_RCODE_WIDTH: usize = 9; +const DNS_QUERY_NAME_WIDTH: usize = 47; + // Trait for common fields on log types pub trait LogLike { fn message(&self) -> &str; @@ -259,6 +265,82 @@ pub fn print_network_flow_log(log: T, json: bool) { println!("{}", format_network_flow_log_string(&log, json)); } +pub trait DnsQueryLogLike: Serialize { + fn queried_at(&self) -> &str; + fn qname(&self) -> &str; + fn qtype(&self) -> &str; + fn rcode(&self) -> &str; + fn query_zone_value(&self) -> String; + fn answers(&self) -> &[String]; +} + +pub fn format_dns_query_log_header() -> String { + format!( + "{:(log: &T, json: bool) -> String { + if json { + let mut value = serde_json::to_value(log).unwrap(); + if let Value::Object(map) = &mut value { + map.insert( + "timestamp".to_string(), + Value::String(log.queried_at().to_string()), + ); + } + return serde_json::to_string(&value).unwrap(); + } + + let zone = log.query_zone_value(); + let zone = if zone == "internal" { + zone.cyan() + } else { + zone.normal() + }; + let rcode = log.rcode(); + let rcode = match rcode { + "NOERROR" => rcode.green(), + "NXDOMAIN" => rcode.yellow(), + _ => rcode.red(), + }; + let answers = if log.answers().is_empty() { + "-".to_string() + } else { + log.answers().join(", ") + }; + + format!( + "{:(log: T, json: bool) { + println!("{}", format_dns_query_log_string(&log, json)); +} + fn endpoint(addr: &str, port: i64) -> String { if addr.contains(':') { format!("[{addr}]:{port}") @@ -508,6 +590,58 @@ impl NetworkFlowLogLike for queries::network_flow_logs::NetworkFlowLogFields { } } +impl DnsQueryLogLike for queries::dns_query_logs::DnsQueryLogFields { + fn queried_at(&self) -> &str { + &self.queried_at + } + + fn qname(&self) -> &str { + &self.qname + } + + fn qtype(&self) -> &str { + &self.qtype + } + + fn rcode(&self) -> &str { + &self.rcode + } + + fn query_zone_value(&self) -> String { + serialized_enum_value(&self.query_zone) + } + + fn answers(&self) -> &[String] { + &self.answers + } +} + +impl DnsQueryLogLike for subscriptions::dns_query_logs::DnsQueryLogFields { + fn queried_at(&self) -> &str { + &self.queried_at + } + + fn qname(&self) -> &str { + &self.qname + } + + fn qtype(&self) -> &str { + &self.qtype + } + + fn rcode(&self) -> &str { + &self.rcode + } + + fn query_zone_value(&self) -> String { + serialized_enum_value(&self.query_zone) + } + + fn answers(&self) -> &[String] { + &self.answers + } +} + impl NetworkFlowLogLike for subscriptions::network_flow_logs::NetworkFlowLogFields { fn capture_end(&self) -> &str { &self.capture_end