use latest time crate

Cargo.toml

@@ -26,7 +26,7 @@ pnet_macros  = "0.34.0"
 pcap         = "=0.7.0"
 rand         = "0.8.5"
 ring         = "0.17.7"
-time         = "0.1.42"
+time         = "0.3.34"
 url          = "2.5.0"
 
 [dependencies.kentik-api]

src/custom.rs

@@ -168,7 +168,7 @@ impl Customs {
             d if d >= min && d <= max => d,
             d if d >= min             => max,
             _                         => min,
-        }.num_milliseconds() as u32);
+        }.whole_milliseconds() as u32);
     }
 
     pub fn clear(&mut self) {

src/libkflow.rs

@@ -170,7 +170,7 @@ pub fn configure(cfg: &Config) -> Result<Device, Error> {
             ip:      opt(&cfg.capture.ip),
         },
         metrics: kflowConfigMetrics {
-            interval: cfg.metrics.interval.num_minutes() as c_int,
+            interval: cfg.metrics.interval.whole_minutes() as c_int,
             URL:      cfg.metrics.url.as_ptr(),
         },
         proxy: kflowConfigProxy {
@@ -182,7 +182,7 @@ pub fn configure(cfg: &Config) -> Result<Device, Error> {
         },
         dns: kflowConfigDNS {
             enable:   cfg.dns.enable as c_int,
-            interval: cfg.dns.interval.num_seconds() as c_int,
+            interval: cfg.dns.interval.whole_seconds() as c_int,
             URL:      cfg.dns.url.as_ptr(),
         },
         device_id:   cfg.device_id as c_int,
@@ -192,7 +192,7 @@ pub fn configure(cfg: &Config) -> Result<Device, Error> {
         device_plan: cfg.device_plan.unwrap_or(0) as c_int,
         device_site: cfg.device_site.unwrap_or(0) as c_int,
         sample:      cfg.sample as c_int,
-        timeout:     cfg.timeout.num_milliseconds() as c_int,
+        timeout:     cfg.timeout.whole_milliseconds() as c_int,
         verbose:     cfg.verbose as c_int,
         program:     cfg.program.as_ptr(),
         version:     cfg.version.as_ptr(),
@@ -297,7 +297,7 @@ pub fn send(key: &Key, ctr: &Counter, sr: u32, cs: Option<&[kflowCustom]>) -> Re
 
 pub fn stop(timeout: Duration) -> Result<(), Error> {
     unsafe {
-        match kflowStop(timeout.num_milliseconds() as c_int) {
+        match kflowStop(timeout.whole_milliseconds() as c_int) {
             0 => Ok(()),
             _ => Err(Error::Timeout),
         }

src/mode/dns.rs

@@ -161,7 +161,7 @@ impl Dns {
             let mut rs = Vec::with_capacity(self.buffer.len());
             swap(&mut self.buffer, &mut rs);
 
-            let timeout = Duration::milliseconds(10).to_std().unwrap();
+            let timeout = Duration::milliseconds(10).unsigned_abs();
             let len = rs.len();
             match self.client.send(rs, timeout) {
                 Ok(..) => debug!("DNS batch sent: {}", len),

src/mode/radius.rs

@@ -152,7 +152,7 @@ impl Radius {
                 deletes:     deletes,
             };
 
-            let timeout = Duration::milliseconds(10).to_std().unwrap();
+            let timeout = Duration::milliseconds(10).unsigned_abs();
             match self.client.send("kt_user", req, timeout) {
                 Ok(..) => (),
                 Err(e) => warn!("tag queue full: {:?}", e),

src/protocol/dhcp/decode.rs

@@ -48,7 +48,7 @@ impl Decoder {
             cs.add_str(self.chaddr, msg.chaddr.as_ref());
             msg.host.as_ref().map(|s| cs.add_str(self.host, s));
             msg.domain.as_ref().map(|s| cs.add_str(self.domain, s));
-            msg.lease.map(|d| cs.add_u32(self.lease, d.num_seconds() as u32));
+            msg.lease.map(|d| cs.add_u32(self.lease, d.whole_seconds() as u32));
             msg.latency.map(|d| cs.add_latency(self.latency, d));
             true
         }).unwrap_or(false)

src/protocol/dns/decode.rs

@@ -50,7 +50,7 @@ impl Decoder {
                     cs.add_u32(self.query_type, qq.qtype as u32);
                     cs.add_u32(self.reply_code, rc as u32);
                     cs.add_str(self.reply_data, self.data_str.as_ref().unwrap_or(&self.empty));
-                    cs.add_u32(self.latency, d.num_milliseconds() as u32);
+                    cs.add_u32(self.latency, d.whole_milliseconds() as u32);
                     true
                 },
             }

src/protocol/http/decode.rs

@@ -113,7 +113,7 @@ impl Decoder {
             cs.add_str(req_referer, res.referer.as_ref().unwrap_or(empty));
             cs.add_str(req_ua, res.ua.as_ref().unwrap_or(empty));
             cs.add_u32(res_status, res.status as u32);
-            cs.add_u32(latency, res.latency.num_milliseconds() as u32);
+            cs.add_u32(latency, res.latency.whole_milliseconds() as u32);
             true
         }).unwrap_or(false)
     }

src/reasm/mod.rs

@@ -48,7 +48,7 @@ impl Reassembler {
     }
 
     pub fn flush(&mut self, ts: Timestamp) {
-        if (ts - self.flushed).num_seconds() > 15 {
+        if (ts - self.flushed).whole_seconds() > 15 {
             self.ipv4.clear(ts, self.timeout);
             self.flushed = ts;
         }

src/tests/mod.rs

@@ -9,6 +9,7 @@ use std::borrow::Cow;
 use std::ffi::CStr;
 use std::mem::swap;
 use std::net::IpAddr;
+use std::time::{UNIX_EPOCH, SystemTime};
 use libc::{c_char, c_int};
 use pcap::Capture;
 use pnet::packet::{Packet as PacketExt, PacketSize};
@@ -50,6 +51,21 @@ fn timeout_range() {
     }
 }
 
+#[test]
+fn time_sanity_checks() {
+    let system    = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+    let timestamp = Timestamp::now().sec;
+    assert!(timestamp >= system);
+    assert!(timestamp - system < 2);
+
+    let timestamp = Timestamp::zero() + Duration::seconds(2);
+    assert_eq!(2, timestamp.sec);
+
+    let timestamp0 = Timestamp::zero() + Duration::seconds(2);
+    let timestamp1 = Timestamp::zero() + Duration::seconds(4);
+    assert_eq!(Duration::seconds(2), timestamp1 - timestamp0);
+}
+
 #[test]
 fn test_decode_wrong_ipv4_length() {
     let mut cap = Capture::from_file("pcaps/dns/sns-pb.isc.org.pcap").unwrap();
@@ -139,7 +155,7 @@ fn test_udp_first_exchange_latency() {
     let dst = Addr{addr: "8.8.4.4".parse().unwrap(),   port: 53   };
     let key = Key(Protocol::UDP, dst, src);
 
-    assert_eq!(Some(44), trk.latency(&key).map(|d| d.num_milliseconds()));
+    assert_eq!(Some(44), trk.latency(&key).map(|d| d.whole_milliseconds()));
 
     let mut customs = Customs::new(&CUSTOMS);
     trk.append(&key, &mut customs);
@@ -162,7 +178,7 @@ fn test_tcp_first_exchange_latency() {
     let dst = Addr{addr: "172.217.25.110".parse().unwrap(), port: 80   };
     let key = Key(Protocol::TCP, dst, src);
 
-    assert_eq!(Some(7), trk.latency(&key).map(|d| d.num_milliseconds()));
+    assert_eq!(Some(7), trk.latency(&key).map(|d| d.whole_milliseconds()));
 
     let mut customs = Customs::new(&CUSTOMS);
     trk.append(&key, &mut customs);

src/time.rs

@@ -1,6 +1,8 @@
+use std::ffi::CStr;
 use std::fmt;
+use std::mem::{MaybeUninit, zeroed};
 use std::ops::{Add, Sub};
-use libc::timeval;
+use libc::{self, CLOCK_REALTIME, timespec, timeval, time_t};
 use time::Duration;
 
 #[derive(Copy, Clone, Debug, Default, Eq, PartialEq, Ord, PartialOrd)]
@@ -11,9 +13,9 @@ pub struct Timestamp {
 
 impl Timestamp {
     pub fn now() -> Self {
-        let ts   = time::get_time();
-        let sec  = ts.sec  as u64;
-        let nsec = ts.nsec as u64;
+        let ts   = gettime();
+        let sec  = ts.tv_sec  as u64;
+        let nsec = ts.tv_nsec as u64;
         Self { sec, nsec }
     }
 
@@ -26,8 +28,8 @@ impl Add<Duration> for Timestamp {
     type Output = Self;
 
     fn add(self, d: Duration) -> Self::Output {
-        let sec  = d.num_seconds() as u64;
-        let nsec = d.num_nanoseconds().unwrap_or(0) as u64;
+        let sec  = d.whole_seconds() as u64;
+        let nsec = d.subsec_nanoseconds() as u64;
         Self {
             sec:  self.sec.saturating_add(sec),
             nsec: self.nsec.saturating_add(nsec),
@@ -56,14 +58,35 @@ impl From<timeval> for Timestamp {
 
 impl fmt::Display for Timestamp {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        let tm = time::at(time::Timespec{
-            sec:  self.sec  as i64,
-            nsec: self.nsec as i32,
-        });
+        write!(f, "{}", format("%F %T", self.sec as time_t))
+    }
+}
 
-        match time::strftime("%F %T", &tm) {
-            Ok(str) => write!(f, "{}", str),
-            Err(..) => Err(fmt::Error)
-        }
+fn gettime() -> timespec {
+    unsafe {
+        let mut ts = MaybeUninit::uninit();
+        match libc::clock_gettime(CLOCK_REALTIME, ts.as_mut_ptr()) {
+            0 => ts,
+            _ => MaybeUninit::zeroed()
+        }.assume_init()
+    }
+}
+
+fn format(fmt: &str, time: time_t) -> String {
+    unsafe {
+        let tm = match libc::localtime(&time) {
+            tm if !tm.is_null() => *tm,
+            _                   => zeroed(),
+        };
+
+        let mut str = [0i8; 32];
+        let ptr = str.as_mut_ptr();
+        let len = str.len() - 1;
+        let fmt = fmt.as_ptr() as *const _;
+
+        match libc::strftime(ptr, len, fmt, &tm) {
+            n if n > 0 => CStr::from_ptr(ptr),
+            _          => CStr::from_bytes_with_nul_unchecked(&[0]),
+        }.to_string_lossy().to_string()
     }
 }

src/timer.rs

@@ -35,7 +35,7 @@ pub struct Timeout {
 
 impl Timeout {
     pub fn new(delay: Duration) -> Self {
-        let skew = Uniform::new_inclusive(0, delay.num_seconds());
+        let skew = Uniform::new_inclusive(0, delay.whole_seconds());
         Timeout{
             delay: delay,
             skew:  skew,