fix unix epoch parsing

Author: Paulo Remoli

src/filters/date_filter.rs

@@ -553,6 +553,14 @@ fn normalize_log_timestamp(ts: &str) -> Option<NormalizedTimestamp> {
         return normalize_nanos_ts(s);
     }
 
+    // Unix epoch: 10-digit seconds, 13-digit milliseconds, 16-digit microseconds
+    // (journalctl JSON __REALTIME_TIMESTAMP), or decimal seconds (journalctl short-unix)
+    if s.as_bytes().first().map(|b| b.is_ascii_digit()).unwrap_or(false) {
+        if let Some(n) = normalize_epoch_ts(s) {
+            return Some(n);
+        }
+    }
+
     // Handle bracket-prefixed timestamps
     if s.starts_with('[') {
         // dmesg: [  seconds.usecs] — digits/spaces/dots only inside brackets
@@ -717,8 +725,7 @@ fn normalize_bsd_ts(s: &str) -> Option<NormalizedTimestamp> {
     })
 }
 
-fn normalize_nanos_ts(s: &str) -> Option<NormalizedTimestamp> {
-    let nanos: i128 = s.parse().ok()?;
+fn nanos_to_normalized(nanos: i128) -> Option<NormalizedTimestamp> {
     let dt = time::OffsetDateTime::from_unix_timestamp_nanos(nanos).ok()?;
     let ms = ((nanos.unsigned_abs() % 1_000_000_000) / 1_000_000) as u16;
     let year = dt.year() as u32;
@@ -733,6 +740,46 @@ fn normalize_nanos_ts(s: &str) -> Option<NormalizedTimestamp> {
     })
 }
 
+fn normalize_nanos_ts(s: &str) -> Option<NormalizedTimestamp> {
+    let nanos: i128 = s.parse().ok()?;
+    nanos_to_normalized(nanos)
+}
+
+/// Handle Unix epoch timestamps in integer and decimal forms:
+/// - 10 digits: seconds          (e.g. `1436735381`)
+/// - 13 digits: milliseconds     (e.g. `1436735381000`)
+/// - 16 digits: microseconds     (e.g. `1699999999000000`, journalctl JSON)
+/// - decimal:   seconds.fraction (e.g. `1436735381.000000`, journalctl short-unix)
+fn normalize_epoch_ts(s: &str) -> Option<NormalizedTimestamp> {
+    if s.bytes().all(|b| b.is_ascii_digit()) {
+        let nanos = match s.len() {
+            10 => s.parse::<i128>().ok()? * 1_000_000_000,
+            13 => s.parse::<i128>().ok()? * 1_000_000,
+            16 => s.parse::<i128>().ok()? * 1_000,
+            _ => return None,
+        };
+        return nanos_to_normalized(nanos);
+    }
+    // Decimal seconds: integer part must be at least 9 digits (epoch ≥ ~2001)
+    if let Some(dot) = s.find('.') {
+        let int_part = &s[..dot];
+        let frac_part = &s[dot + 1..];
+        if int_part.len() >= 9
+            && int_part.bytes().all(|b| b.is_ascii_digit())
+            && !frac_part.is_empty()
+            && frac_part.bytes().all(|b| b.is_ascii_digit())
+        {
+            let secs: i64 = int_part.parse().ok()?;
+            let frac_len = frac_part.len().min(9);
+            let frac_digits: i128 = frac_part[..frac_len].parse().ok()?;
+            let frac_nanos = frac_digits * 10i128.pow((9 - frac_len) as u32);
+            let nanos = secs as i128 * 1_000_000_000 + frac_nanos;
+            return nanos_to_normalized(nanos);
+        }
+    }
+    None
+}
+
 fn normalize_apache_error_ts(s: &str) -> Option<NormalizedTimestamp> {
     // "[Mon Jan 15 10:30:00.123456 2024]" or "[Fri Dec 31 23:59:59 2024]"
     if !s.starts_with('[') || !s.ends_with(']') {
@@ -1471,6 +1518,49 @@ mod tests {
         assert!(normalize_log_timestamp("170004601023400000").is_none());
     }
 
+    // ── Unix epoch (seconds / milliseconds / microseconds / decimal) ──
+
+    #[test]
+    fn test_normalize_epoch_micros_journalctl_json() {
+        // journalctl JSON __REALTIME_TIMESTAMP (16 digits = microseconds)
+        let n = normalize_log_timestamp("1700046010234000").unwrap();
+        assert_eq!(buf_as_str(&n.canonical), "2023-11-15 11:00:10.234");
+    }
+
+    #[test]
+    fn test_normalize_epoch_millis() {
+        // 13-digit millisecond epoch (Pino, Bunyan, etc.)
+        let n = normalize_log_timestamp("1700046010234").unwrap();
+        assert_eq!(buf_as_str(&n.canonical), "2023-11-15 11:00:10.234");
+    }
+
+    #[test]
+    fn test_normalize_epoch_secs() {
+        // 10-digit second epoch
+        let n = normalize_log_timestamp("1700046010").unwrap();
+        assert_eq!(buf_as_str(&n.canonical), "2023-11-15 11:00:10.000");
+    }
+
+    #[test]
+    fn test_normalize_epoch_decimal_secs() {
+        // Decimal seconds (journalctl short-unix, syslog unix)
+        let n = normalize_log_timestamp("1700046010.234000").unwrap();
+        assert_eq!(buf_as_str(&n.canonical), "2023-11-15 11:00:10.234");
+    }
+
+    #[test]
+    fn test_normalize_epoch_decimal_short_frac() {
+        // Fewer fractional digits
+        let n = normalize_log_timestamp("1700046010.5").unwrap();
+        assert_eq!(buf_as_str(&n.canonical), "2023-11-15 11:00:10.500");
+    }
+
+    #[test]
+    fn test_normalize_epoch_decimal_rejects_short_integer() {
+        // Integer part < 9 digits → not an epoch
+        assert!(normalize_log_timestamp("12345678.000000").is_none());
+    }
+
     #[test]
     fn test_canonical_timestamp_nanos() {
         let result = canonical_timestamp("1700046010234000000", None);

src/parser/syslog.rs

@@ -12,6 +12,9 @@ enum SyslogFormat {
     /// rsyslog RSYSLOG_FileFormat: `ISO-timestamp hostname tag[pid]: message`
     /// (ISO 8601 timestamp with no `<PRI>` prefix).
     RsyslogIso,
+    /// rsyslog with Unix epoch timestamp: `1436735381.000000 hostname tag[pid]: message`
+    /// (no `<PRI>` prefix).
+    Unix,
 }
 
 impl SyslogFormat {
@@ -23,7 +26,8 @@ impl SyslogFormat {
         match i {
             0 => Self::Rfc3164,
             1 => Self::Rfc5424,
-            _ => Self::RsyslogIso,
+            2 => Self::RsyslogIso,
+            _ => Self::Unix,
         }
     }
 }
@@ -33,7 +37,7 @@ const MIN_SAMPLES: u32 = 50;
 #[derive(Debug, Default)]
 pub struct SyslogParser {
     format: OnceLock<SyslogFormat>,
-    fmt_counts: [AtomicU32; 3],
+    fmt_counts: [AtomicU32; 4],
     fmt_total: AtomicU32,
 }
 
@@ -455,6 +459,8 @@ fn detect_syslog_timestamp<'a>(s: &'a str, line: &'a [u8]) -> Option<(SyslogForm
             }
         } else if let Some((ts, _)) = super::timestamp::parse_iso_timestamp(body) {
             return Some((SyslogFormat::RsyslogIso, ts));
+        } else if let Some((ts, _)) = super::timestamp::parse_unix_timestamp(body) {
+            return Some((SyslogFormat::Unix, ts));
         }
     }
     None
@@ -489,12 +495,45 @@ fn extract_syslog_timestamp_rsyslog_iso(s: &str) -> Option<&str> {
     super::timestamp::parse_iso_timestamp(s).map(|(ts, _)| ts)
 }
 
+fn extract_syslog_timestamp_unix(s: &str) -> Option<&str> {
+    super::timestamp::parse_unix_timestamp(s).map(|(ts, _)| ts)
+}
+
+/// Parse rsyslog with Unix epoch timestamp: `1436735381.000000 hostname tag[pid]: message`.
+/// No `<PRI>` prefix.
+fn parse_rsyslog_unix_inner<'a>(s: &'a str) -> Option<DisplayParts<'a>> {
+    let (timestamp, ts_end) = super::timestamp::parse_unix_timestamp(s)?;
+    let rest = s[ts_end..].strip_prefix(' ')?;
+
+    let mut parts = DisplayParts {
+        timestamp: Some(timestamp),
+        ..Default::default()
+    };
+
+    if rest.is_empty() {
+        return Some(parts);
+    }
+
+    let (hostname, rest) = next_token(rest)?;
+    if !is_valid_syslog_hostname(hostname) {
+        return None;
+    }
+    push_field_as(&mut parts.extra_fields, FieldSemantic::Hostname, hostname);
+
+    if rest.is_empty() {
+        return Some(parts);
+    }
+
+    extract_tag_and_message(rest, &mut parts);
+    Some(parts)
+}
+
 impl SyslogParser {
     fn record_format(&self, fmt: SyslogFormat) {
         self.fmt_counts[fmt.index()].fetch_add(1, Ordering::Relaxed);
         let total = self.fmt_total.fetch_add(1, Ordering::Relaxed) + 1;
         if total >= MIN_SAMPLES && self.format.get().is_none() {
-            let winner = (0..3)
+            let winner = (0..4)
                 .max_by_key(|&i| self.fmt_counts[i].load(Ordering::Relaxed))
                 .unwrap_or(0);
             let _ = self.format.set(SyslogFormat::from_index(winner));
@@ -517,6 +556,7 @@ impl LogFormatParser for SyslogParser {
                 SyslogFormat::Rfc3164 => extract_syslog_timestamp_rfc3164(s, line),
                 SyslogFormat::Rfc5424 => extract_syslog_timestamp_rfc5424(s, line),
                 SyslogFormat::RsyslogIso => extract_syslog_timestamp_rsyslog_iso(s),
+                SyslogFormat::Unix => extract_syslog_timestamp_unix(s),
             };
         }
         let (fmt, ts) = detect_syslog_timestamp(s, line)?;
@@ -544,6 +584,7 @@ impl LogFormatParser for SyslogParser {
                     }
                 }
                 SyslogFormat::RsyslogIso => parse_rsyslog_iso_inner(s),
+                SyslogFormat::Unix => parse_rsyslog_unix_inner(s),
             };
             if result.is_some() {
                 return result;
@@ -572,6 +613,11 @@ impl LogFormatParser for SyslogParser {
             return Some(parts);
         }
 
+        if let Some(parts) = parse_rsyslog_unix_inner(s) {
+            self.record_format(SyslogFormat::Unix);
+            return Some(parts);
+        }
+
         None
     }
 
@@ -608,7 +654,7 @@ impl LogFormatParser for SyslogParser {
     ///   • ISO timestamp (`YYYY-MM-DDTHH:MM:SS…`) — rsyslog RSYSLOG_FileFormat
     ///
     /// Plain BSD lines without a priority prefix (`Oct 11 22:14:15 host tag: msg`)
-    /// are shared with journalctl `--output short` and are intentionally **not**
+    /// and Unix epoch lines are shared with journalctl and are intentionally **not**
     /// claimed here so that piped `journalctl` output is still detected as
     /// journalctl.  Those lines can still be *parsed* by `parse_line` once the
     /// format is locked by other lines in the sample.
@@ -1029,4 +1075,40 @@ mod tests {
         assert_eq!(before.level, after.level);
         assert_eq!(before.target, after.target);
     }
+
+    // ── Unix epoch (rsyslog custom template) ─────────────────────────
+
+    #[test]
+    fn test_unix_epoch_basic() {
+        let line = b"1436735381.000000 myhost sshd[1234]: Connection closed";
+        let parser = SyslogParser::default();
+        let parts = parser.parse_line(line).unwrap();
+        assert_eq!(parts.timestamp, Some("1436735381.000000"));
+        assert_eq!(parts.target, Some("sshd"));
+        assert_eq!(parts.message, Some("Connection closed"));
+        assert!(
+            parts
+                .extra_fields
+                .iter()
+                .any(|(_, k, v)| *k == "hostname" && *v == "myhost")
+        );
+    }
+
+    #[test]
+    fn test_unix_epoch_parse_timestamp() {
+        let line = b"1700000000.123456 myhost systemd[1]: Started service";
+        let parser = SyslogParser::default();
+        let ts = parser.parse_timestamp(line).unwrap();
+        assert_eq!(ts, "1700000000.123456");
+    }
+
+    #[test]
+    fn test_unix_epoch_timestamp_has_year() {
+        let parser = SyslogParser::default();
+        let line = b"1436735381.000000 myhost sshd[1234]: Connection closed";
+        for _ in 0..MIN_SAMPLES {
+            parser.parse_line(line).unwrap();
+        }
+        assert!(parser.timestamp_has_year());
+    }
 }