fuel_telemetry/
file_watcher.rs

1use crate::{
2    daemonise, enforce_singleton, into_recoverable, telemetry_config, EnvSetting, Result,
3    TelemetryError, WatcherResult,
4};
5
6use base64::{engine::general_purpose::STANDARD, DecodeError, Engine};
7use chrono::NaiveDateTime;
8use influxdb_line_protocol::LineProtocolBuilder;
9use nix::{
10    fcntl::{Flock, FlockArg},
11    sys::signal::{kill, Signal::SIGKILL},
12    unistd::{getpid, Pid},
13};
14use regex::{Captures, Regex};
15use reqwest::blocking::{Client, Request, RequestBuilder};
16use std::fs::ReadDir;
17use std::{
18    collections::HashMap,
19    env::var,
20    fs::{read_dir, remove_file, File, OpenOptions},
21    io::{BufRead, BufReader, Read, Write},
22    path::{Path, PathBuf},
23    process::exit,
24    sync::{
25        atomic::{AtomicBool, AtomicI32, Ordering},
26        LazyLock,
27    },
28    thread::sleep,
29    time::Duration,
30};
31
32//
33// Module config
34//
35
36const PROCESS_NAME: &str = "telemetry-file-watcher";
37const INFLUXDB_ORG: &str = "Dev%20Team";
38const INFLUXDB_BUCKET: &str = "telemetry";
39const INFLUXDB_URL: &str = "https://us-east-1-1.aws.cloud2.influxdata.com";
40const INFLUXDB_TOKEN: &str =
41    "l7Sho-XGD9BfGLQrKWwoBub-hC0gqJ5xRS2zz4pkjb6cGyBJZUQpw7qpwTfXTFGLXufCh7ZmQWv4bUtAsT60Ag==";
42
43/// Configuration for `FileWatcher`
44#[derive(Debug, Clone)]
45struct FileWatcherConfig {
46    // The path to its lockfile
47    lockfile: PathBuf,
48    // The path to its logfile
49    logfile: PathBuf,
50    // The token to use to authenticate with InfluxDB
51    influxdb_token: String,
52    // The URL of the InfluxDB instance
53    influxdb_url: String,
54    // The interval to poll for aged-out telemetry files
55    poll_interval: Duration,
56}
57
58/// Get the `FileWatcher` configuration
59///
60/// This function returns the `'static` `FileWatcher` configuration.
61fn config() -> Result<&'static FileWatcherConfig> {
62    static FILEWATCHER_CONFIG: LazyLock<Result<FileWatcherConfig>> = LazyLock::new(|| {
63        let get_env = |key, default| EnvSetting::new(key, default).get();
64
65        // Since we use an hourly appender, we default to polling every hour
66        let poll_interval = get_env("FILEWATCHER_POLL_INTERVAL", "3600").parse()?;
67
68        // Format the InfluxDB URL (the org name needs to be URL-encoded)
69        let base_url = get_env("INFLUXDB_URL", INFLUXDB_URL);
70        let org = get_env("INFLUXDB_ORG", INFLUXDB_ORG);
71        let bucket = get_env("INFLUXDB_BUCKET", INFLUXDB_BUCKET);
72        let influxdb_url =
73            format!("{base_url}/api/v2/write?org={org}&bucket={bucket}&precision=ns");
74
75        Ok(FileWatcherConfig {
76            poll_interval: Duration::from_secs(poll_interval),
77            influxdb_token: get_env("INFLUXDB_TOKEN", INFLUXDB_TOKEN),
78            influxdb_url,
79            lockfile: Path::new(&telemetry_config()?.fuelup_tmp)
80                .join(format!("{PROCESS_NAME}.lock")),
81            logfile: Path::new(&telemetry_config()?.fuelup_log).join(format!("{PROCESS_NAME}.log")),
82        })
83    });
84
85    FILEWATCHER_CONFIG
86        .as_ref()
87        .map_err(|e| TelemetryError::InvalidConfig(e.to_string()))
88}
89
90//
91// FileWatcher implementation
92//
93
94/// A `FileWatcher` polls for aged-out telemetry files and sends them to InfluxDB
95#[derive(Default)]
96pub struct FileWatcher {
97    // The client used to send requests to InfluxDB
98    client: Option<Client>,
99    // The request to send to InfluxDB
100    request: Option<Request>,
101}
102
103// Prevent recursive calls to start()
104static STARTED: AtomicBool = AtomicBool::new(false);
105
106/// The PID of the currently running `FileWatcher` daemon
107pub static PID: AtomicI32 = AtomicI32::new(0);
108
109/// A regex to parse telemetry events
110static TELEMETRY_EVENT_REGEX: LazyLock<Result<Regex>> = LazyLock::new(|| {
111    Ok(Regex::new(
112        r"(?x)
113        ^                                                                         \s*
114        (?P<timestamp>[^\s]+)                                                     \s+
115        (?P<level>(TRACE|DEBUG|INFO|WARN|ERROR))                                  \s+
116        (?P<triple>.*?):(?P<os>.*?):(?P<os_version>.*?)                           \s+
117        (?P<crate_pkg_name>[^\s]+):(?P<crate_pkg_version>[^\s]+):(?P<file>[^\s]+) \s+
118        (?P<trace_id>[^\s]+)                                                      \s+
119        (?P<payload>.*)                                                           \s*
120        $
121        ",
122    )?)
123});
124
125/// A regex to parse telemetry payloads
126///
127/// Event payloads have the following formats:
128///
129/// ```text
130/// - <message_only>
131/// - [<span1>:<span2>:...:] [<message>] [<left1>= <right1> <left2>= <right2> ...]
132/// ```
133static TELEMETRY_PAYLOAD_REGEX: LazyLock<Result<Regex>> = LazyLock::new(|| {
134    Ok(Regex::new(
135        r"(?x)
136        (?P<span>[^\s\:][^\:]*?[^\s\:]*?) :
137        | (?:^|\s+) (?P<message_only>[^\=]+) \s* $
138        | (?:^|\s+) (?P<message>[^\=]+) \s+
139        | (?P<left>[^\s\=]+) = (?P<right>[^\=]+) (?:\s+|$)
140        ",
141    )?)
142});
143
144impl FileWatcher {
145    /// Create a new `FileWatcher`
146    ///
147    /// Warning: This function should only be called in applications and not
148    /// within libraries as it will clobber any set by the application.
149    ///
150    /// Warning: If you will be creating a `FileWatcher` and a `TelemetryLayer`,
151    /// you must create the `FileWatcher` before the `TelemetryLayer` as there is
152    /// a race condition in the thread runtime of `tracing` and the tokio runtime
153    /// of `Reqwest`. Swapping order of the two could lead to possible deadlocks.
154    pub fn new() -> Self {
155        Self::default()
156    }
157
158    /// Start the `FileWatcher`
159    ///
160    /// This function forks and has the parent immediately return. The forked
161    /// off child then daemonises to poll for aged-out telemetry files, sending
162    /// them to InfluxDB, and exits once there are no more files to process.
163    pub fn start(&mut self) -> WatcherResult<()> {
164        self.start_with_helpers(&DefaultStartHelpers)
165    }
166
167    fn start_with_helpers(&mut self, helpers: &impl StartHelpers) -> WatcherResult<()> {
168        if var("FUELUP_NO_TELEMETRY").is_ok() {
169            // If telemetry is disabled, immediately return
170            return Ok(());
171        }
172
173        let logfile = &config().map_err(into_recoverable)?.logfile;
174
175        if STARTED.load(Ordering::Relaxed) {
176            return Ok(());
177        } else {
178            STARTED.store(true, Ordering::Relaxed);
179        }
180
181        match helpers.daemonise(logfile) {
182            Ok(Some(pid)) => {
183                // We are the parent, so record the PID then immediately return
184                PID.store(pid.as_raw(), Ordering::Relaxed);
185                return Ok(());
186            }
187            Err(e) => {
188                // Couldn't daemonise, so clear globals before returning the error
189                STARTED.store(false, Ordering::Relaxed);
190                PID.store(0, Ordering::Relaxed);
191                return Err(e);
192            }
193            Ok(None) => {
194                // We are the child process, so continue as the `FileWatcher`...
195                PID.store(getpid().as_raw(), Ordering::Relaxed);
196            }
197        }
198
199        // From here on, we are no longer the original process, so the caller should
200        // treat errors as fatal. This means that on error the process should exit
201        // immediately as there should not be two identical flows of execution
202        // from the caller's perspective.
203
204        // Cache the client and request so we don't recreate them each time
205        self.client = Some(Client::new());
206        self.request = Some(helpers.build_request(self)?);
207
208        loop {
209            // Enforce a singleton to ensure we are the only process submitting
210            // telemetry to InfluxDB
211            let _lock = helpers.enforce_singleton(&config()?.lockfile)?;
212
213            // Poll for aged-out telemetry files
214            let directory_empty = helpers.poll_directory(self)?;
215
216            if directory_empty {
217                helpers.exit(0);
218            }
219
220            // Sleep for the configured interval before polling again
221            helpers.sleep(config()?.poll_interval);
222        }
223    }
224
225    /// Kill the `FileWatcher` daemon if one is running
226    pub fn kill() -> Result<bool> {
227        let pid = PID.load(Ordering::Relaxed);
228
229        if pid == 0 {
230            return Ok(false);
231        }
232
233        kill(Pid::from_raw(pid), SIGKILL)?;
234        PID.store(0, Ordering::Relaxed);
235        Ok(true)
236    }
237
238    /// Poll for aged-out telemetry files
239    fn poll_directory(&self) -> Result<bool> {
240        self.poll_directory_with_helpers(&mut DefaultPollDirectoryHelpers)
241    }
242
243    fn poll_directory_with_helpers(&self, helpers: &mut impl PollDirectoryHelpers) -> Result<bool> {
244        for file in helpers.find_telemetry_files(false)? {
245            // Lock the telemetry file to prevent being submitted twice
246            let locked_file = helpers
247                .flock(
248                    OpenOptions::new().read(true).open(&file)?,
249                    FlockArg::LockExclusiveNonblock,
250                )
251                .map_err(|(_, e)| e)?;
252
253            // Read the telemetry file line by line
254            let mut body = Vec::new();
255            let lines = helpers.buffered_reader(locked_file.try_clone()?)?;
256
257            for base64_line in lines {
258                // First, decode the Base64 line
259                let decoded_line = helpers.base64_decode(&base64_line)?;
260                let line = helpers.string_from_utf8(decoded_line)?;
261
262                let (event, spans_long, spans_short, fields) = helpers.parse_event(&line)?;
263                let line_protocol =
264                    helpers.build_line_protocol(&event, &spans_long, &spans_short, &fields);
265
266                body.push(helpers.string_from_utf8(line_protocol)?);
267            }
268
269            if body.is_empty() {
270                continue;
271            }
272
273            // Do not cache creating this request in the constructor then attempt
274            // to use build_from_parts() here, because the `Client` used to build
275            // the request is in another castle^Wprocess (we've since forked many
276            // times), and so a `Client::new()` here will deadlock as it will
277            // try to connect to a non-existant connection pool.
278            let request = helpers.build_request(self, &body)?;
279
280            if helpers.send_request(request)? {
281                // Only remove the telemetry file if successful
282                helpers.remove_file(&file)?;
283            }
284        }
285
286        // Return false if there are pending telemetry files to process
287        // regardless of age so that we keep going until there's no more work to
288        // be done
289        Ok(helpers.find_telemetry_files(true)?.is_empty())
290    }
291
292    /// Last resort logging of errors, only to be used by the caller when there
293    /// is no other way to report an error.
294    pub fn log_error(message: &str) -> Result<()> {
295        let mut file = OpenOptions::new()
296            .create(true)
297            .append(true)
298            .open(&config()?.logfile)?;
299
300        Ok(writeln!(file, "{message}")?)
301    }
302}
303
304trait StartHelpers {
305    fn daemonise(&self, logfile: &PathBuf) -> WatcherResult<Option<Pid>> {
306        daemonise(logfile)
307    }
308
309    fn build_request(&self, file_watcher: &FileWatcher) -> Result<Request> {
310        Ok(file_watcher
311            .client
312            .as_ref()
313            .ok_or(TelemetryError::InvalidClient)?
314            .post(&config()?.influxdb_url)
315            .header("Content-Type", "text/plain; charset=utf-8")
316            .header("Accept", "application/json")
317            .header(
318                "Authorization",
319                format!("Token {}", config()?.influxdb_token.clone()),
320            )
321            .build()?)
322    }
323
324    fn enforce_singleton(&self, lockfile_path: &Path) -> Result<Flock<File>> {
325        enforce_singleton(lockfile_path)
326    }
327
328    fn poll_directory(&self, file_watcher: &FileWatcher) -> Result<bool> {
329        file_watcher.poll_directory()
330    }
331
332    fn exit(&self, code: i32) {
333        exit(code)
334    }
335
336    fn sleep(&self, duration: Duration) {
337        sleep(duration)
338    }
339}
340
341struct DefaultStartHelpers;
342impl StartHelpers for DefaultStartHelpers {}
343
344trait PollDirectoryHelpers {
345    fn find_telemetry_files(&mut self, ignore_age: bool) -> Result<Vec<PathBuf>> {
346        find_telemetry_files(ignore_age)
347    }
348
349    fn flock(
350        &self,
351        file: File,
352        flags: FlockArg,
353    ) -> std::result::Result<Flock<File>, (File, nix::errno::Errno)> {
354        Flock::lock(file, flags)
355    }
356
357    fn buffered_reader<R: Read>(
358        &self,
359        file: R,
360    ) -> std::result::Result<Vec<String>, std::io::Error> {
361        BufReader::new(file)
362            .lines()
363            .collect::<std::result::Result<Vec<_>, _>>()
364    }
365
366    fn base64_decode(&self, line: &str) -> std::result::Result<Vec<u8>, DecodeError> {
367        STANDARD.decode(line.as_bytes())
368    }
369
370    fn string_from_utf8(
371        &mut self,
372        bytes: Vec<u8>,
373    ) -> std::result::Result<String, std::string::FromUtf8Error> {
374        String::from_utf8(bytes)
375    }
376
377    #[allow(clippy::type_complexity)]
378    fn parse_event<'a>(
379        &self,
380        line: &'a str,
381    ) -> Result<(
382        Captures<'a>,
383        Vec<String>,
384        Vec<String>,
385        HashMap<&'a str, String>,
386    )> {
387        let event = TELEMETRY_EVENT_REGEX
388            .as_ref()?
389            .captures(line)
390            .ok_or_else(|| TelemetryError::InvalidTracingEvent(line.to_string()))?;
391
392        let payload = event
393            .name("payload")
394            .ok_or_else(|| TelemetryError::InvalidTracingPayload(line.to_string()))?;
395
396        let (mut spans_long, mut spans_short, mut fields) = (vec![], vec![], HashMap::new());
397
398        for capture in TELEMETRY_PAYLOAD_REGEX
399            .as_ref()?
400            .captures_iter(payload.into())
401        {
402            if let Some(span) = capture.name("span") {
403                let span_str = span.as_str();
404                spans_long.push(span_str.to_string());
405                spans_short.push(span_str.split('{').next().unwrap_or(span_str).to_string());
406            } else if let Some(message) = capture.name("message") {
407                fields.insert("message", message.as_str().to_string());
408            } else if let Some(message_only) = capture.name("message_only") {
409                fields.insert("message", message_only.as_str().to_string());
410            } else if let (Some(left), Some(right)) = (capture.name("left"), capture.name("right"))
411            {
412                fields.insert(left.as_str(), right.as_str().to_string());
413            }
414        }
415
416        Ok((event, spans_long, spans_short, fields))
417    }
418
419    fn build_line_protocol(
420        &self,
421        event: &Captures<'_>,
422        spans_long: &[String],
423        spans_short: &[String],
424        fields: &HashMap<&str, String>,
425    ) -> Vec<u8> {
426        let mut line_protocol_builder = LineProtocolBuilder::new()
427            .measurement(event.name("crate_pkg_name").map_or("", |m| m.as_str()))
428            .tag(
429                "version",
430                event.name("crate_pkg_version").map_or("", |m| m.as_str()),
431            )
432            .tag("file", event.name("file").map_or("", |m| m.as_str()))
433            .tag("spans", &spans_short.join(":"))
434            .tag(
435                "trace_id",
436                event.name("trace_id").map_or("", |m| m.as_str()),
437            )
438            .field("spans_long", spans_long.join(":").as_str())
439            .field("triple", event.name("triple").map_or("", |m| m.as_str()));
440
441        for (key, value) in fields {
442            line_protocol_builder = line_protocol_builder.field(key, value.as_str());
443        }
444
445        // Set the timestamp of the LineProtocol. Here we force a `.9` decimal
446        // format rather than use the default nanoseconds format as InfluxDB seems
447        // to have issues parsing datetime data
448        let line_protocol_builder = line_protocol_builder.timestamp(
449            event
450                .name("timestamp")
451                .and_then(|m| {
452                    // Within `lib.rs:format_event()` we deliberately stick with
453                    // Zulu however we use `%Z` format here as external apps writing
454                    // telemetry files may use other timezone names
455                    NaiveDateTime::parse_from_str(m.as_str(), "%Y-%m-%dT%H:%M:%S%.9f%Z")
456                        .ok()
457                        .map(|dt| dt.and_utc().timestamp_nanos_opt())
458                })
459                .flatten()
460                .unwrap_or(0),
461        );
462
463        line_protocol_builder.close_line().build()
464    }
465
466    fn build_request(&self, file_watcher: &FileWatcher, body: &[String]) -> Result<RequestBuilder> {
467        Ok(RequestBuilder::from_parts(
468            file_watcher
469                .client
470                .as_ref()
471                .ok_or(TelemetryError::InvalidClient)?
472                .clone(),
473            file_watcher
474                .request
475                .as_ref()
476                .ok_or(TelemetryError::InvalidRequest)?
477                .try_clone()
478                .ok_or(TelemetryError::ReqwestCloneFailed)?,
479        )
480        .body(body.join("\n")))
481    }
482
483    fn send_request(&self, request: RequestBuilder) -> Result<bool> {
484        Ok(request.send().is_ok_and(|r| r.status().is_success()))
485    }
486
487    fn remove_file(&self, file: &Path) -> std::io::Result<()> {
488        remove_file(file)
489    }
490}
491
492struct DefaultPollDirectoryHelpers;
493impl PollDirectoryHelpers for DefaultPollDirectoryHelpers {}
494
495/// Find telemetry files
496///
497/// This function finds all telemetry files in the configured directory.
498///
499/// If `ignore_age` is true, return telemetry files regardless of age, otherwise
500/// only return files that are older than the configured poll interval.
501fn find_telemetry_files(ignore_age: bool) -> Result<Vec<PathBuf>> {
502    find_telemetry_files_with_read_dir_fn(ignore_age, |path| read_dir(path))
503}
504
505fn find_telemetry_files_with_read_dir_fn(
506    ignore_age: bool,
507    read_dir_fn: impl FnOnce(&str) -> std::result::Result<ReadDir, std::io::Error>,
508) -> Result<Vec<PathBuf>> {
509    let poll_interval = config()?.poll_interval;
510    let telemetry_dir = &telemetry_config()?.fuelup_tmp;
511
512    // Filter out files not containing `.telemetry.` in the filename
513    //
514    // Also, ignore file age if `ignore_age` is true
515    read_dir_fn(telemetry_dir)?
516        .filter_map(std::result::Result::ok)
517        .map(|e| e.path())
518        .filter(|p| p.is_file())
519        .filter(|p| p.to_string_lossy().contains(".telemetry."))
520        .filter_map(|path| {
521            if ignore_age
522                || poll_interval < path.metadata().ok()?.modified().ok()?.elapsed().ok()?
523            {
524                Some(Ok(path))
525            } else {
526                None
527            }
528        })
529        .collect::<Result<Vec<_>>>()
530}
531
532#[cfg(test)]
533mod config {
534    use super::*;
535    use crate::setup_fuelup_home;
536    use dirs::home_dir;
537    use rusty_fork::rusty_fork_test;
538    use std::env::set_var;
539
540    rusty_fork_test! {
541        #[test]
542        fn all_unset() {
543            let config = config().unwrap();
544
545            assert_eq!(
546                config.lockfile,
547                Path::new(
548                    &home_dir()
549                        .unwrap()
550                        .join(format!(".fuelup/tmp/{}.lock", PROCESS_NAME))
551                )
552            );
553            assert_eq!(
554                config.logfile,
555                Path::new(
556                    &home_dir()
557                        .unwrap()
558                        .join(format!(".fuelup/log/{}.log", PROCESS_NAME))
559                )
560            );
561            assert_eq!(config.poll_interval, Duration::from_secs(3600));
562            assert_eq!(config.influxdb_token, INFLUXDB_TOKEN);
563
564            assert_eq!(config.influxdb_url, format!(
565                "{}/api/v2/write?org={}&bucket={}&precision=ns",
566                INFLUXDB_URL,
567                INFLUXDB_ORG,
568                INFLUXDB_BUCKET
569            ));
570        }
571
572        #[test]
573        fn poll_interval_set() {
574            set_var("FILEWATCHER_POLL_INTERVAL", "2222");
575
576            let config = config().unwrap();
577            assert_eq!(config.poll_interval, Duration::from_secs(2222));
578        }
579
580        #[test]
581        fn poll_interval_invalid() {
582            set_var("FILEWATCHER_POLL_INTERVAL", "invalid");
583
584            let config = config();
585            assert_eq!(
586                config.err(),
587                Some(TelemetryError::InvalidConfig(
588                    TelemetryError::Parse("invalid digit found in string".to_string()).into()
589                ))
590            );
591        }
592
593        #[test]
594        fn influxdb_url_set() {
595            set_var("INFLUXDB_URL", "http://localhost:8000");
596
597            let config = config().unwrap();
598            assert_eq!(
599                config.influxdb_url,
600                format!("{}/api/v2/write?org={}&bucket={}&precision=ns",
601                "http://localhost:8000",
602                INFLUXDB_ORG,
603                INFLUXDB_BUCKET
604            ));
605        }
606
607        #[test]
608        fn influxdb_org_set() {
609            set_var("INFLUXDB_ORG", "org-name");
610
611            let config = config().unwrap();
612            assert_eq!(config.influxdb_url, format!(
613                "{}/api/v2/write?org={}&bucket={}&precision=ns",
614                INFLUXDB_URL,
615                "org-name",
616                INFLUXDB_BUCKET
617            ));
618        }
619
620        #[test]
621        fn influxdb_bucket_set() {
622            set_var("INFLUXDB_BUCKET", "bucket-name");
623
624            let config = config().unwrap();
625            assert_eq!(config.influxdb_url, format!(
626                "{}/api/v2/write?org={}&bucket={}&precision=ns",
627                INFLUXDB_URL,
628                INFLUXDB_ORG,
629                "bucket-name"
630            ));
631        }
632
633        #[test]
634        fn all_set() {
635            setup_fuelup_home();
636            let fuelup_home = var("FUELUP_HOME").unwrap();
637
638            set_var("FILEWATCHER_POLL_INTERVAL", "2222");
639            set_var("INFLUXDB_URL", "http://localhost:8000");
640            set_var("INFLUXDB_ORG", "org-name");
641            set_var("INFLUXDB_BUCKET", "bucket-name");
642
643            let config = config().unwrap();
644
645            assert_eq!(
646                config.lockfile,
647                Path::new(&format!("{}/tmp/{}.lock", &fuelup_home, PROCESS_NAME))
648            );
649
650            assert_eq!(
651                config.logfile,
652                Path::new(&format!("{}/log/{}.log", &fuelup_home, PROCESS_NAME))
653            );
654            assert_eq!(config.poll_interval, Duration::from_secs(2222));
655            assert_eq!(config.influxdb_token, INFLUXDB_TOKEN);
656            assert_eq!(
657                config.influxdb_url,
658                format!("{}/api/v2/write?org={}&bucket={}&precision=ns",
659                    "http://localhost:8000",
660                    "org-name",
661                    "bucket-name"
662                )
663            );
664        }
665    }
666}
667
668#[cfg(test)]
669mod new {
670    use super::*;
671    use rusty_fork::rusty_fork_test;
672
673    rusty_fork_test! {
674        #[test]
675        fn new() {
676            let file_watcher = FileWatcher::new();
677
678            assert!(file_watcher.client.is_none());
679            assert!(file_watcher.request.is_none());
680        }
681    }
682}
683
684#[cfg(test)]
685mod start {
686    use super::*;
687    use crate::{errors::into_fatal, setup_fuelup_home, WatcherError};
688    use nix::{
689        sys::wait::{waitpid, WaitStatus},
690        unistd::{dup2, fork, pipe, ForkResult},
691    };
692    use rusty_fork::rusty_fork_test;
693    use std::{
694        env::set_var,
695        fs::File,
696        io::{stdout, Read, Write},
697        os::fd::{AsRawFd, FromRawFd, IntoRawFd},
698    };
699
700    rusty_fork_test! {
701        #[test]
702        fn opted_out_is_true() {
703            set_var("FUELUP_NO_TELEMETRY", "true");
704
705            let mut file_watcher = FileWatcher::new();
706            let result = file_watcher.start();
707
708            // Make sure it didn't continue and init values
709            assert!(matches!(result, Ok(())));
710            assert!(file_watcher.client.is_none());
711            assert!(file_watcher.request.is_none());
712            assert!(!STARTED.load(Ordering::Relaxed));
713            assert_eq!(PID.load(Ordering::Relaxed), 0);
714        }
715
716        #[test]
717        fn opted_out_is_empty() {
718            // Even though it's empty, we only care if it's set
719            set_var("FUELUP_NO_TELEMETRY", "");
720
721            let mut file_watcher = FileWatcher::new();
722            let result = file_watcher.start();
723
724            // Make sure it didn't continue and init values
725            assert!(matches!(result, Ok(())));
726            assert!(file_watcher.client.is_none());
727            assert!(file_watcher.request.is_none());
728            assert!(!STARTED.load(Ordering::Relaxed));
729            assert_eq!(PID.load(Ordering::Relaxed), 0);
730        }
731
732        #[test]
733        fn already_started() {
734            STARTED.store(true, Ordering::Relaxed);
735            PID.store(1, Ordering::Relaxed);
736
737            let mut file_watcher = FileWatcher::new();
738            let result = file_watcher.start();
739
740            // Make sure it didn't continue and init values
741            assert!(matches!(result, Ok(())));
742            assert!(file_watcher.client.is_none());
743            assert!(file_watcher.request.is_none());
744            assert!(STARTED.load(Ordering::Relaxed));
745            assert_eq!(PID.load(Ordering::Relaxed), 1);
746
747            // Try to start it again to test re-entrance
748            let result = file_watcher.start();
749
750            assert!(matches!(result, Ok(())));
751            assert!(file_watcher.client.is_none());
752            assert!(file_watcher.request.is_none());
753            assert!(STARTED.load(Ordering::Relaxed));
754            assert_eq!(PID.load(Ordering::Relaxed), 1);
755        }
756
757        #[test]
758        fn daemonise_failed() {
759            struct DaemoniseFailed;
760
761            impl StartHelpers for DaemoniseFailed {
762                fn daemonise(&self, _logfile: &PathBuf) -> WatcherResult<Option<Pid>> {
763                    Err(into_fatal(TelemetryError::Mock))
764                }
765            }
766
767            let mut file_watcher = FileWatcher::new();
768            let result = file_watcher.start_with_helpers(&DaemoniseFailed);
769
770            assert_eq!(
771                result.err(),
772                Some(WatcherError::Fatal(TelemetryError::Mock))
773            );
774            assert!(!STARTED.load(Ordering::Relaxed));
775            assert_eq!(PID.load(Ordering::Relaxed), 0);
776        }
777
778        #[test]
779        fn daemonise_is_parent() {
780            struct DaemoniseIsParent;
781
782            impl StartHelpers for DaemoniseIsParent {
783                fn daemonise(&self, _logfile: &PathBuf) -> WatcherResult<Option<Pid>> {
784                    Ok(Some(Pid::from_raw(1337)))
785                }
786            }
787
788            let mut file_watcher = FileWatcher::new();
789            let result = file_watcher.start_with_helpers(&DaemoniseIsParent);
790
791            assert_eq!(result, Ok(()));
792            assert!(STARTED.load(Ordering::Relaxed));
793            assert_eq!(PID.load(Ordering::Relaxed), 1337);
794        }
795
796        #[test]
797        fn build_request_invalid_client() {
798            struct BuildRequestInvalidClient;
799
800            impl StartHelpers for BuildRequestInvalidClient {
801                fn daemonise(&self, _logfile: &PathBuf) -> WatcherResult<Option<Pid>> {
802                    Ok(None)
803                }
804
805                fn build_request(&self, _file_watcher: &FileWatcher) -> Result<Request> {
806                    Err(TelemetryError::InvalidClient)
807                }
808            }
809
810            let mut file_watcher = FileWatcher::new();
811            let result = file_watcher.start_with_helpers(&BuildRequestInvalidClient);
812
813            assert_eq!(
814                result,
815                Err(WatcherError::Fatal(TelemetryError::InvalidClient))
816            );
817        }
818
819        #[test]
820        fn build_request_build_failed() {
821            setup_fuelup_home();
822
823            struct BuildRequestBuildFailed;
824
825            impl StartHelpers for BuildRequestBuildFailed {
826                fn daemonise(&self, _logfile: &PathBuf) -> WatcherResult<Option<Pid>> {
827                    Ok(None)
828                }
829
830                fn build_request(&self, _file_watcher: &FileWatcher) -> Result<Request> {
831                    // Builder errors are private, so generate a real one
832                    Ok(Client::new()
833                        .post("http://localhost")
834                        .header("Invalid Header", "")
835                        .build()?)
836                }
837            }
838
839            let mut file_watcher = FileWatcher::new();
840            let result = file_watcher.start_with_helpers(&BuildRequestBuildFailed);
841
842            assert_eq!(
843                result,
844                Err(WatcherError::Fatal(TelemetryError::Reqwest(
845                    "builder error".to_string()
846                )))
847            );
848        }
849
850        #[test]
851        fn enforce_singleton_failed() {
852            struct EnforceSingletonFailed;
853
854            impl StartHelpers for EnforceSingletonFailed {
855                fn daemonise(&self, _logfile: &PathBuf) -> WatcherResult<Option<Pid>> {
856                    Ok(None)
857                }
858
859                fn enforce_singleton(&self, _lockfile_path: &Path) -> Result<Flock<File>> {
860                    Err(TelemetryError::Mock)
861                }
862            }
863
864            let mut file_watcher = FileWatcher::new();
865            let result = file_watcher.start_with_helpers(&EnforceSingletonFailed);
866
867            assert_eq!(result, Err(WatcherError::Fatal(TelemetryError::Mock)));
868        }
869
870        #[test]
871        fn poll_directory_failed() {
872            setup_fuelup_home();
873
874            struct PollDirectoryFailed;
875
876            impl StartHelpers for PollDirectoryFailed {
877                fn daemonise(&self, _logfile: &PathBuf) -> WatcherResult<Option<Pid>> {
878                    Ok(None)
879                }
880
881                fn poll_directory(&self, _file_watcher: &FileWatcher) -> Result<bool> {
882                    Err(TelemetryError::Mock)
883                }
884            }
885
886            let mut file_watcher = FileWatcher::new();
887            let result = file_watcher.start_with_helpers(&PollDirectoryFailed);
888
889            assert_eq!(result, Err(WatcherError::Fatal(TelemetryError::Mock)));
890        }
891
892        #[test]
893        fn poll() {
894            setup_fuelup_home();
895
896            struct Poll;
897
898            impl StartHelpers for Poll {
899                fn daemonise(&self, _logfile: &PathBuf) -> WatcherResult<Option<Pid>> {
900                    // Mock daemonising by not forking to become the child
901                    Ok(None)
902                }
903
904                fn poll_directory(&self, _file_watcher: &FileWatcher) -> Result<bool> {
905                    static HAS_POLLED: AtomicBool = AtomicBool::new(false);
906
907                    if !HAS_POLLED.load(Ordering::Relaxed) {
908                        HAS_POLLED.store(true, Ordering::Relaxed);
909
910                        // Send the PID so we can test it's as expected
911                        let file_watcher_pid = crate::file_watcher::PID
912                            .load(Ordering::Relaxed)
913                            .as_raw_fd()
914                            .to_ne_bytes();
915
916                        stdout().write_all(&file_watcher_pid).unwrap();
917                        stdout().flush().unwrap();
918
919                        Ok(false)
920                    } else {
921                        Ok(true)
922                    }
923                }
924
925                fn exit(&self, _code: i32) {
926                    // Test we actually exited via our expected code path
927                    exit(99)
928                }
929
930                fn sleep(&self, _duration: Duration) {
931                    // No Sleep Till Brooklyn
932                }
933            }
934
935            let (read_fd, write_fd) = pipe().unwrap();
936            let mut pipe_read = unsafe { File::from_raw_fd(read_fd.into_raw_fd()) };
937            let pipe_write = unsafe { File::from_raw_fd(write_fd.into_raw_fd()) };
938
939            match unsafe { fork() }.unwrap() {
940                ForkResult::Parent { child } => {
941                    drop(pipe_write);
942
943                    let mut pid_bytes = [0u8; std::mem::size_of::<Pid>()];
944                    pipe_read.read_exact(&mut pid_bytes).unwrap();
945
946                    let pid = Pid::from_raw(i32::from_ne_bytes(pid_bytes));
947                    assert_eq!(pid, child);
948
949                    match waitpid(child, None).unwrap() {
950                        WaitStatus::Exited(_, code) => {
951                            assert_eq!(code, 99);
952                        }
953                        _ => panic!("Child did not exit normally"),
954                    }
955                }
956                ForkResult::Child => {
957                    drop(pipe_read);
958                    dup2(pipe_write.as_raw_fd(), 1).unwrap();
959
960                    // Since we're mocking daemonising by not double forking in
961                    // the parent, we have to set the PID manually
962                    let pid_bytes = getpid().as_raw().to_ne_bytes();
963                    stdout().write_all(&pid_bytes).unwrap();
964                    stdout().flush().unwrap();
965
966                    let mut file_watcher = FileWatcher::new();
967                    let _ = file_watcher.start_with_helpers(&Poll);
968
969                    // Fallback status code
970                    exit(86);
971                }
972            }
973        }
974    }
975}
976
977#[cfg(test)]
978mod kill {
979    use super::*;
980    use nix::{
981        sys::{
982            signal::Signal,
983            wait::{waitpid, WaitPidFlag, WaitStatus},
984        },
985        unistd::{fork, ForkResult},
986    };
987    use rusty_fork::rusty_fork_test;
988    use std::thread::sleep;
989
990    rusty_fork_test! {
991        #[test]
992        fn kill_nobody() {
993            crate::file_watcher::PID.store(0, Ordering::Relaxed);
994            assert!(!FileWatcher::kill().unwrap());
995        }
996
997        #[test]
998        fn kill_file_watcher() {
999            let mut kill_called = false;
1000
1001            match unsafe { fork() }.unwrap() {
1002                ForkResult::Parent { child } => {
1003                    loop {
1004                        match waitpid(child, Some(WaitPidFlag::WNOHANG)) {
1005                            Ok(WaitStatus::StillAlive) => {
1006                                if !kill_called {
1007                                    // Since we're not daemonising, we have to set the PID manually
1008                                    crate::file_watcher::PID
1009                                        .store(child.as_raw(), Ordering::Relaxed);
1010                                    assert!(FileWatcher::kill().unwrap());
1011
1012                                    kill_called = true;
1013                                    continue;
1014                                }
1015                            }
1016                            Ok(WaitStatus::Signaled(child_pid, signal, _)) => {
1017                                assert_eq!(child_pid, child);
1018                                assert_eq!(signal, Signal::SIGKILL);
1019                                break;
1020                            }
1021                            _ => panic!("Child process terminated unexpectedly"),
1022                        }
1023                    }
1024                }
1025                ForkResult::Child => loop {
1026                    sleep(Duration::from_secs(10));
1027                },
1028            }
1029        }
1030    }
1031}
1032
1033#[cfg(test)]
1034mod poll_directory {
1035    use super::*;
1036    use crate::setup_fuelup_home;
1037    use rusty_fork::rusty_fork_test;
1038    use std::{fs::File, io::Write, sync::OnceLock};
1039
1040    fn setup_test_file() -> PathBuf {
1041        static TEST_FILE: OnceLock<PathBuf> = OnceLock::new();
1042
1043        let filename = TEST_FILE.get_or_init(|| {
1044            let filename = format!("{}/{}", telemetry_config().unwrap().fuelup_tmp, "test.telemetry.file");
1045
1046            let mut file = File::create(Path::new(filename.as_str())).unwrap();
1047            file.write_all(b"MjAyNS0wMy0xNlQxNDo0ODoxMC4wODQ4MDMwMDBaICBJTkZPIGFhcmNoNjQtYXBwbGUtZGFyd2luOkRhcndpbjoyMy42LjAgc2ltcGxlOjAuMS4wOmV4YW1wbGVzL3NpbXBsZS5ycyA0YjEyNDYyNy0wNjNiLTQwM2UtODEyZi02OWQxMzRlZjAxMzIgYXV0bzogZ0Z1ZWwh").unwrap();
1048
1049            PathBuf::from(filename)
1050        });
1051
1052        filename.clone()
1053    }
1054
1055    rusty_fork_test! {
1056        #[test]
1057        fn find_telemetry_files_failed() {
1058            setup_fuelup_home();
1059
1060            struct FindTelemetryFilesFailed;
1061
1062            impl PollDirectoryHelpers for FindTelemetryFilesFailed {
1063                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1064                    Err(TelemetryError::Mock)
1065                }
1066            }
1067
1068            let file_watcher = FileWatcher::new();
1069            let result = file_watcher.poll_directory_with_helpers(&mut FindTelemetryFilesFailed);
1070
1071            assert_eq!(result, Err(TelemetryError::Mock));
1072       }
1073
1074        #[test]
1075        fn find_telemetry_files_empty() {
1076            setup_fuelup_home();
1077
1078            struct FindTelemetryFilesEmpty;
1079
1080            impl PollDirectoryHelpers for FindTelemetryFilesEmpty {
1081                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1082                    Ok(vec![])
1083                }
1084            }
1085
1086            let file_watcher = FileWatcher::new();
1087            let result = file_watcher.poll_directory_with_helpers(&mut FindTelemetryFilesEmpty);
1088
1089            assert_eq!(result, Ok(true));
1090        }
1091
1092        #[test]
1093        fn flock_failed() {
1094            setup_fuelup_home();
1095
1096            struct FlockFailed;
1097
1098            impl PollDirectoryHelpers for FlockFailed {
1099                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1100                    Ok(vec![setup_test_file()])
1101                }
1102
1103                fn flock(&self, file: File, _flags: FlockArg)-> std::result::Result<Flock<File>, (File, nix::errno::Errno)> {
1104                    Err((file, nix::errno::Errno::EOWNERDEAD))
1105                }
1106            }
1107
1108            let file_watcher = FileWatcher::new();
1109            let result = file_watcher.poll_directory_with_helpers(&mut FlockFailed);
1110
1111            assert_eq!(
1112                result,
1113                Err(TelemetryError::Nix(nix::errno::Errno::EOWNERDEAD.to_string()))
1114            );
1115        }
1116
1117        #[test]
1118        fn buffered_reader_failed() {
1119            setup_fuelup_home();
1120
1121            struct BufferedReaderFailed;
1122
1123            impl PollDirectoryHelpers for BufferedReaderFailed {
1124                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1125                    Ok(vec![setup_test_file()])
1126                }
1127
1128                fn buffered_reader<R: Read>(&self, _file: R) -> std::result::Result<Vec<String>, std::io::Error> {
1129                    Err(std::io::Error::new(std::io::ErrorKind::Other, "Mock error"))
1130                }
1131            }
1132
1133            let file_watcher = FileWatcher::new();
1134            let result = file_watcher.poll_directory_with_helpers(&mut BufferedReaderFailed);
1135
1136            assert_eq!(
1137                result,
1138                Err(TelemetryError::IO("Mock error".to_string()))
1139            );
1140        }
1141
1142        #[test]
1143        fn empty_file() {
1144            setup_fuelup_home();
1145
1146            struct EmptyFile;
1147
1148            impl PollDirectoryHelpers for EmptyFile {
1149                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1150                    Ok(vec![setup_test_file()])
1151                }
1152
1153                fn buffered_reader<R: Read>(&self, _file: R) -> std::result::Result<Vec<String>, std::io::Error> {
1154                    Ok(vec![])
1155                }
1156            }
1157
1158            let file_watcher = FileWatcher::new();
1159            let result = file_watcher.poll_directory_with_helpers(&mut EmptyFile);
1160
1161            // The test file was simulated as empty, so it should stick around
1162            assert_eq!(result, Ok(false));
1163        }
1164
1165        #[test]
1166        fn base64_decode_failed() {
1167            setup_fuelup_home();
1168
1169            struct Base64DecodeFailed;
1170
1171            impl PollDirectoryHelpers for Base64DecodeFailed {
1172                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1173                    Ok(vec![setup_test_file()])
1174                }
1175
1176                fn base64_decode(&self, _line: &str) -> std::result::Result<Vec<u8>, DecodeError> {
1177                    Err(DecodeError::InvalidLength(111222333444555))
1178                }
1179            }
1180
1181            let file_watcher = FileWatcher::new();
1182            let result = file_watcher.poll_directory_with_helpers(&mut Base64DecodeFailed);
1183
1184            assert_eq!(
1185                result,
1186                Err(TelemetryError::Base64(DecodeError::InvalidLength(111222333444555)))
1187            );
1188        }
1189
1190        #[test]
1191        fn string_from_utf8_failed() {
1192            setup_fuelup_home();
1193
1194            struct StringFromUtf8Failed;
1195
1196            impl PollDirectoryHelpers for StringFromUtf8Failed {
1197                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1198                    Ok(vec![setup_test_file()])
1199                }
1200
1201                fn string_from_utf8(&mut self, _bytes: Vec<u8>) -> std::result::Result<String, std::string::FromUtf8Error> {
1202                    // Can't manually create an error, so simulate a failed UTF-8 conversion
1203                    String::from_utf8(vec![0x80])
1204                }
1205            }
1206
1207            let file_watcher = FileWatcher::new();
1208            let result = file_watcher.poll_directory_with_helpers(&mut StringFromUtf8Failed);
1209
1210            assert_eq!(
1211                result,
1212                Err(TelemetryError::Utf8("invalid utf-8 sequence of 1 bytes from index 0".to_string()))
1213            );
1214        }
1215
1216        #[test]
1217        fn parse_event_failed() {
1218            setup_fuelup_home();
1219
1220            struct ParseEventFailed;
1221
1222            impl PollDirectoryHelpers for ParseEventFailed {
1223                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1224                    Ok(vec![setup_test_file()])
1225                }
1226
1227                fn parse_event<'a>(&self, _line: &'a str) -> Result<(
1228                    Captures<'a>,
1229                    Vec<String>,
1230                    Vec<String>,
1231                    HashMap<&'a str, String>,
1232                )> {
1233                    Err(TelemetryError::Mock)
1234                }
1235            }
1236
1237            let file_watcher = FileWatcher::new();
1238            let result = file_watcher.poll_directory_with_helpers(&mut ParseEventFailed);
1239
1240            assert_eq!(result, Err(TelemetryError::Mock));
1241        }
1242
1243        #[test]
1244        fn line_protocol_from_utf8_failed() {
1245            setup_fuelup_home();
1246
1247            #[derive(Default)]
1248            struct LineProtocolFromUtf8Failed {
1249                call_counter: usize,
1250            }
1251
1252            impl PollDirectoryHelpers for LineProtocolFromUtf8Failed {
1253                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1254                    Ok(vec![setup_test_file()])
1255                }
1256
1257                fn string_from_utf8(&mut self, bytes: Vec<u8>) -> std::result::Result<String, std::string::FromUtf8Error> {
1258                    self.call_counter += 1;
1259
1260                    if self.call_counter == 2 {
1261                        String::from_utf8(vec![0x80])
1262                    } else {
1263                        String::from_utf8(bytes)
1264                    }
1265                }
1266            }
1267
1268            let file_watcher = FileWatcher::new();
1269            let result = file_watcher.poll_directory_with_helpers(&mut LineProtocolFromUtf8Failed::default());
1270
1271            assert_eq!(
1272                result,
1273                Err(TelemetryError::Utf8("invalid utf-8 sequence of 1 bytes from index 0".to_string()))
1274            );
1275        }
1276
1277        #[test]
1278        fn build_request_failed() {
1279            setup_fuelup_home();
1280
1281            struct BuildRequestFailed;
1282
1283            impl PollDirectoryHelpers for BuildRequestFailed {
1284                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1285                    Ok(vec![setup_test_file()])
1286                }
1287
1288                fn build_request(&self, _file_watcher: &FileWatcher, _body: &[String]) -> Result<RequestBuilder> {
1289                    Err(TelemetryError::Mock)
1290                }
1291            }
1292
1293            let file_watcher = FileWatcher::new();
1294            let result = file_watcher.poll_directory_with_helpers(&mut BuildRequestFailed);
1295
1296            assert_eq!(result, Err(TelemetryError::Mock));
1297        }
1298
1299        #[test]
1300        fn send_request_failed() {
1301            setup_fuelup_home();
1302
1303            struct SendRequestFailed;
1304
1305
1306            impl PollDirectoryHelpers for SendRequestFailed {
1307                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1308                    Ok(vec![setup_test_file()])
1309                }
1310
1311                fn send_request(&self, _request: RequestBuilder) -> Result<bool> {
1312                    Err(TelemetryError::Mock)
1313                }
1314            }
1315
1316            let mut file_watcher = FileWatcher::new();
1317            let start_helpers = DefaultStartHelpers;
1318
1319            file_watcher.client = Some(Client::new());
1320            file_watcher.request = Some(start_helpers.build_request(&file_watcher).unwrap());
1321
1322            let result = file_watcher.poll_directory_with_helpers(&mut SendRequestFailed);
1323            assert_eq!(result, Err(TelemetryError::Mock));
1324
1325            let test_file = setup_test_file();
1326            assert!(test_file.exists());
1327        }
1328
1329        #[test]
1330        fn remove_file_failed() {
1331            setup_fuelup_home();
1332
1333            struct RemoveFileFailed;
1334
1335            impl PollDirectoryHelpers for RemoveFileFailed {
1336                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1337                    Ok(vec![setup_test_file()])
1338                }
1339
1340                fn send_request(&self, _request: RequestBuilder) -> Result<bool> {
1341                    Ok(true)
1342                }
1343
1344                fn remove_file(&self, _file: &Path) -> std::io::Result<()> {
1345                    Err(std::io::Error::new(std::io::ErrorKind::Other, "Mock error"))
1346                }
1347            }
1348
1349            let mut file_watcher = FileWatcher::new();
1350            let start_helpers = DefaultStartHelpers;
1351
1352            file_watcher.client = Some(Client::new());
1353            file_watcher.request = Some(start_helpers.build_request(&file_watcher).unwrap());
1354
1355            let result = file_watcher.poll_directory_with_helpers(&mut RemoveFileFailed);
1356            assert_eq!(result, Err(TelemetryError::IO("Mock error".to_string())));
1357
1358            let test_file = setup_test_file();
1359            assert!(test_file.exists());
1360        }
1361
1362        #[test]
1363        fn ok() {
1364            setup_fuelup_home();
1365
1366            #[derive(Default)]
1367            struct AOk {
1368                call_count: usize,
1369            }
1370
1371            impl PollDirectoryHelpers for AOk {
1372                fn find_telemetry_files(&mut self, _ignore_age: bool) -> Result<Vec<PathBuf>> {
1373                    self.call_count += 1;
1374
1375                    if self.call_count == 2 {
1376                        Ok(vec![])
1377                    } else {
1378                        Ok(vec![setup_test_file()])
1379                    }
1380                }
1381
1382                fn send_request(&self, _request: RequestBuilder) -> Result<bool> {
1383                    Ok(true)
1384                }
1385            }
1386
1387            let mut file_watcher = FileWatcher::new();
1388            let start_helpers = DefaultStartHelpers;
1389
1390            file_watcher.client = Some(Client::new());
1391            file_watcher.request = Some(start_helpers.build_request(&file_watcher).unwrap());
1392
1393            let result = file_watcher.poll_directory_with_helpers(&mut AOk::default());
1394            assert_eq!(result, Ok(true));
1395
1396            let test_file = setup_test_file();
1397            assert!(!test_file.exists());
1398        }
1399    }
1400}
1401
1402#[cfg(test)]
1403mod find_telemetry_files {
1404    use super::*;
1405    use crate::setup_fuelup_home;
1406    use rusty_fork::rusty_fork_test;
1407    use std::{fs::File, io::Write, time::SystemTime};
1408
1409    fn setup_test_files() {
1410        for filename in [
1411            "old-file.telemetry.file",
1412            "new-file.telemetry.file",
1413            "invalid.file.name",
1414        ] {
1415            let filepath = format!("{}/{}", telemetry_config().unwrap().fuelup_tmp, filename);
1416            let mut file = File::create(Path::new(filepath.as_str())).unwrap();
1417            file.write_all(b"MjAyNS0wMy0xNlQxNDo0ODoxMC4wODQ4MDMwMDBaICBJTkZPIGFhcmNoNjQtYXBwbGUtZGFyd2luOkRhcndpbjoyMy42LjAgc2ltcGxlOjAuMS4wOmV4YW1wbGVzL3NpbXBsZS5ycyA0YjEyNDYyNy0wNjNiLTQwM2UtODEyZi02OWQxMzRlZjAxMzIgYXV0bzogZ0Z1ZWwh").unwrap();
1418        }
1419
1420        let old_file = File::open(Path::new(
1421            format!(
1422                "{}/old-file.telemetry.file",
1423                telemetry_config().unwrap().fuelup_tmp
1424            )
1425            .as_str(),
1426        ))
1427        .unwrap();
1428
1429        old_file
1430            .set_modified(
1431                SystemTime::now() - Duration::from_secs(config().unwrap().poll_interval.as_secs()),
1432            )
1433            .unwrap();
1434
1435        for dir in ["dir1", "dir2", "dir3"] {
1436            let dir = format!("{}/{}", telemetry_config().unwrap().fuelup_tmp, dir);
1437            std::fs::create_dir_all(dir).unwrap();
1438        }
1439    }
1440
1441    rusty_fork_test! {
1442        #[test]
1443        fn read_dir_failed() {
1444            setup_fuelup_home();
1445            setup_test_files();
1446
1447            let result = find_telemetry_files_with_read_dir_fn(
1448                false,
1449                |_| Err(std::io::Error::new(std::io::ErrorKind::Other, "Mock error")),
1450            );
1451
1452            assert_eq!(result, Err(TelemetryError::IO("Mock error".to_string())));
1453        }
1454
1455        #[test]
1456        fn ok() {
1457            setup_fuelup_home();
1458            setup_test_files();
1459
1460            let result = find_telemetry_files(false);
1461
1462            assert_eq!(result, Ok(vec![
1463                PathBuf::from(format!("{}/old-file.telemetry.file", telemetry_config().unwrap().fuelup_tmp)),
1464            ]));
1465        }
1466    }
1467}
fuel_telemetry/file_watcher.rs

fuel_telemetry/
file_watcher.rs
