ribeye 0.3.0

ribeye -- a Rust library for BGP RIB data processing.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

//! RIB data processors.
//!
//! This module contains the processors that are used to process RIB data.

mod as2rel;
mod meta;
mod peer_stats;
mod pfx2as;

pub use as2rel::As2relProcessor;
pub use meta::RibMeta;
pub use peer_stats::PeerStatsProcessor;
pub use pfx2as::Prefix2AsProcessor;

use anyhow::Result;
use bgpkit_parser::BgpElem;
use std::io::Write;
use tempfile::tempdir;
use tracing::info;

pub trait MessageProcessor {
    /// Get the name of the processor
    fn name(&self) -> String;

    /// Output paths of the processor. An output path can be a local file path or an S3 path.
    fn output_paths(&self) -> Option<Vec<String>>;

    fn reset_processor(&mut self, rib_meta: &RibMeta);

    /// Process a single entry in the RIB
    fn process_entry(&mut self, elem: &BgpElem) -> Result<()>;

    /// Generate final result in String to be written to output file
    fn to_result_string(&self) -> Option<String> {
        None
    }

    /// Finalize the processor, including producing the output and storing it
    fn output(&mut self) -> Result<()> {
        if self.output_paths().is_none() {
            // no output path, skip
            return Ok(());
        }

        let output_string = match self.to_result_string() {
            None => return Ok(()),
            Some(o) => o,
        };

        let output_paths = self.output_paths().unwrap();

        for output_path in output_paths {
            // if output_path starts with s3://, upload to S3
            if output_path.starts_with("s3://") {
                info!(
                    "finalizing {} processing, writing output to {}",
                    self.name(),
                    output_path.as_str(),
                );

                let temp_dir = tempfile::tempdir().unwrap();
                let file_path = temp_dir
                    .path()
                    .join("temp.bz2")
                    .to_str()
                    .unwrap()
                    .to_string();
                let mut writer = oneio::get_writer(file_path.as_str()).unwrap();
                writer.write_all(output_string.as_ref())?;
                drop(writer);

                let (bucket, p) = oneio::s3_url_parse(output_path.as_str())?;
                oneio::s3_upload(bucket.as_str(), p.as_str(), file_path.as_str()).unwrap();
                temp_dir.close().unwrap();
            } else {
                info!(
                    "finalizing {} processing, writing output to {}",
                    self.name(),
                    output_path.as_str()
                );

                let mut writer = oneio::get_writer(output_path.as_str())?;
                writer.write_all(output_string.as_ref())?;
                drop(writer);
            }
        }
        Ok(())
    }

    /// Summarize the latest RIBEye result files
    fn summarize_latest(&self, rib_metas: &[RibMeta], ignore_error: bool) -> Result<()>;

    fn to_boxed(self) -> Box<dyn MessageProcessor>
    where
        Self: Sized + 'static,
    {
        Box::new(self)
    }
}

pub(crate) fn write_output_file(
    output_file_dir: &str,
    output_content: &str,
    compress: bool,
) -> Result<()> {
    let output_file_path = match compress {
        true => format!("{}/latest.json.bz2", output_file_dir),
        false => format!("{}/latest.json", output_file_dir),
    };
    match output_file_dir.starts_with("s3://") {
        true => {
            // write to a temporary file first
            let tmp_dir = tempdir()?;
            let file_path = tmp_dir
                .path()
                .join("latest.json.bz2")
                .to_string_lossy()
                .to_string();
            let mut writer = oneio::get_writer(file_path.as_str())?;
            write!(writer, "{}", output_content)?;
            drop(writer);

            let (bucket, p) = oneio::s3_url_parse(output_file_path.as_str())?;
            oneio::s3_upload(bucket.as_str(), p.as_str(), file_path.as_str())?;
        }
        false => {
            let mut writer = oneio::get_writer(output_file_path.as_str())?;
            write!(writer, "{}", output_content)?;
            drop(writer);
        }
    }

    Ok(())
}