relay/
setup.rs

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
#[cfg(feature = "processing")]
use anyhow::Context;
use anyhow::Result;
use relay_config::{Config, RelayMode};

pub fn check_config(config: &Config) -> Result<()> {
    if config.relay_mode() == RelayMode::Managed && config.credentials().is_none() {
        anyhow::bail!(
            "relay has no credentials, which are required in managed mode. \
             Generate some with \"relay credentials generate\" first.",
        );
    }

    if config.relay_mode() != RelayMode::Managed && config.processing_enabled() {
        anyhow::bail!("Processing can only be enabled in managed mode.");
    }

    #[cfg(feature = "processing")]
    if config.processing_enabled() {
        for (name, topic) in config.unused_topic_assignments() {
            relay_log::with_scope(
                |scope| scope.set_extra("topic", format!("{topic:?}").into()),
                || relay_log::error!("unused topic assignment '{name}'"),
            );
        }

        for topic in relay_kafka::KafkaTopic::iter() {
            let _ = config
                .kafka_config(*topic)
                .with_context(|| format!("invalid kafka configuration for topic '{topic:?}'"))?;
        }
    }

    Ok(())
}

/// Print spawn infos to the log.
pub fn dump_spawn_infos(config: &Config) {
    if config.path().as_os_str().is_empty() {
        relay_log::info!("launching relay without config folder");
    } else {
        relay_log::info!(
            "launching relay from config folder {}",
            config.path().display()
        );
    }
    relay_log::info!("  relay mode: {}", config.relay_mode());

    match config.relay_id() {
        Some(id) => relay_log::info!("  relay id: {id}"),
        None => relay_log::info!("  relay id: -"),
    };
    match config.public_key() {
        Some(key) => relay_log::info!("  public key: {key}"),
        None => relay_log::info!("  public key: -"),
    };
    relay_log::info!("  log level: {}", config.logging().level);
}

/// Dumps out credential info.
pub fn dump_credentials(config: &Config) {
    match config.relay_id() {
        Some(id) => println!("  relay id: {id}"),
        None => println!("  relay id: -"),
    };
    match config.public_key() {
        Some(key) => println!("  public key: {key}"),
        None => println!("  public key: -"),
    };
}

/// Initialize the metric system.
pub fn init_metrics(config: &Config) -> Result<()> {
    let addrs = config.statsd_addrs()?;
    if addrs.is_empty() {
        return Ok(());
    }

    let mut default_tags = config.metrics_default_tags().clone();
    if let Some(hostname_tag) = config.metrics_hostname_tag() {
        if let Some(hostname) = hostname::get().ok().and_then(|s| s.into_string().ok()) {
            default_tags.insert(hostname_tag.to_owned(), hostname);
        }
    }
    relay_statsd::init(
        config.metrics_prefix(),
        &addrs[..],
        default_tags,
        config.metrics_sample_rate(),
        config.metrics_aggregate(),
    );

    Ok(())
}