crusty 0.2.0

Fast && scalable Broad Web Crawler developed on top of crusty-core
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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

#[allow(unused_imports)]
use crate::prelude::*;
use crate::{
    job_reader,
    types::*
};

use crusty_core::config as rc;

use std::{fs, env};

use serde::{Deserialize};
use once_cell::sync::Lazy;

pub static CONFIG: Lazy<Mutex<CrustyConfig>> = Lazy::new(|| Mutex::new(CrustyConfig::default() ));

#[derive(Clone, Debug, Deserialize)]
pub struct ClickhouseWriterConfig {
    pub table_name: String,
    pub label: String,
    pub buffer_capacity: usize,
    pub check_for_force_write_duration: rc::CDuration,
    pub force_write_duration: rc::CDuration,
}


#[derive(Clone, Debug, Deserialize)]
pub struct ClickhouseConfig {
    pub url: String,
    pub username: String,
    pub password: String,
    pub database: String,

    pub metrics_queue: ClickhouseWriterConfig,
    pub metrics_db: ClickhouseWriterConfig,
    pub metrics_task: ClickhouseWriterConfig,
    pub domain_discovery_insert: ClickhouseWriterConfig,
    pub domain_discovery_update: ClickhouseWriterConfig,
}

impl Default for ClickhouseConfig {
    fn default() -> Self {
        Self {
            url: String::from("http://localhost:8123"),
            username: String::from("default"),
            password: String::from(""),
            database: String::from("default"),

            metrics_queue: ClickhouseWriterConfig{
                table_name: String::from("metrics_queue"),
                label: String::from(""),
                buffer_capacity: 1000,
                check_for_force_write_duration: rc::CDuration::from_millis(100),
                force_write_duration: rc::CDuration::from_millis(500),
            },
            metrics_db: ClickhouseWriterConfig{
                table_name: String::from("metrics_db"),
                label: String::from(""),
                buffer_capacity: 1000,
                check_for_force_write_duration: rc::CDuration::from_millis(100),
                force_write_duration: rc::CDuration::from_millis(500),
            },
            metrics_task: ClickhouseWriterConfig{
                table_name: String::from("metrics_task"),
                label: String::from(""),
                buffer_capacity: 10000,
                check_for_force_write_duration: rc::CDuration::from_millis(100),
                force_write_duration: rc::CDuration::from_millis(500),
            },
            domain_discovery_insert: ClickhouseWriterConfig{
                table_name: String::from("domain_discovery"),
                label: String::from("insert"),
                buffer_capacity: 10000,
                check_for_force_write_duration: rc::CDuration::from_millis(500),
                force_write_duration: rc::CDuration::from_millis(2500),
            },
            domain_discovery_update: ClickhouseWriterConfig{
                table_name: String::from("domain_discovery"),
                label: String::from("update"),
                buffer_capacity: 10000,
                check_for_force_write_duration: rc::CDuration::from_millis(500),
                force_write_duration: rc::CDuration::from_millis(2500),
            },
        }
    }
}

#[derive(Clone, Debug, Deserialize)]
pub struct LogConfig {
    pub level: rc::CLevel,
    pub ansi: bool,
    pub filter: Option<Vec<String>>
}

impl Default for LogConfig {
    fn default() -> Self {
        Self {
            level: rc::CLevel(Level::INFO),
            ansi: true,
            filter: None,
        }
    }
}

#[derive(Clone, Debug, Deserialize)]
#[serde(default)]
pub struct CrustyConfig {
    pub host: String,
    pub app_id: String,
    pub log: LogConfig,
    pub clickhouse: ClickhouseConfig,

    pub ddc_cap: usize,
    pub ddc_lifetime: rc::CDuration,
    pub queue_monitor_interval: rc::CDuration,
    pub parser_processor_stack_size: rc::CBytes,

    pub networking_profile: rc::NetworkingProfile,
    pub concurrency_profile: rc::ConcurrencyProfile,
    pub job_reader: job_reader::JobReaderConfig,
}

impl Default for CrustyConfig {
    fn default() -> Self {
        Self {
            host: String::from("crawler-1"),
            app_id: String::from("rusty-spider"),
            log: LogConfig::default(),
            clickhouse: ClickhouseConfig::default(),

            ddc_cap: 25_000_000,
            ddc_lifetime: rc::CDuration::from_secs(60 * 60),
            queue_monitor_interval: rc::CDuration::from_secs(1),
            parser_processor_stack_size: rc::CBytes(1024 * 1024 * 32),

            networking_profile: rc::NetworkingProfile::default(),
            concurrency_profile: rc::ConcurrencyProfile::default(),
            job_reader: job_reader::JobReaderConfig::default()
        }
    }
}

pub fn load() -> Result<()> {
    let cfg_str = fs::read_to_string("./config.yaml")?;
    let mut config = CONFIG.lock().unwrap();

    let mut e = None;
    match serde_yaml::from_str(&cfg_str) {
        Ok(cfg) => {*config = cfg}
        Err(err) => {e = Some(err)}
    }

    if let Ok(seeds) = env::var("CRUSTY_SEEDS") {
        config.job_reader.seeds.extend(
            seeds.split(',').map(String::from).collect::<Vec<_>>()
        );
    }

    if let Some(err) = e {
        return Err(err.into())
    }
    Ok(())
}