etcds 0.16.0

An etcd v3 API server - light server version for queue management
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



#[macro_use]
extern crate slog;

use std::{fs::File, io::BufReader};
use std::process::ExitCode;
use clap::Parser;
use slog::Logger;
use sloggers::Build;
use sloggers::terminal::{Destination, TerminalLoggerBuilder};
use sloggers::types::{Severity, SourceLocation};

/// etcd standalone entry point
#[tokio::main]
async fn main() -> ExitCode {
    // Parse whole args with clap
    let mut args = EtcdCliArgs::parse();
   
    // Get config file
    let config = if let Ok(f) = File::open(&args.config_path) {
        match args.parse_from(BufReader::new(f)) {
            Ok(a) => a,
            Err(e) => {
                eprintln!("Error parsing config file {}: {}", args.config_path.display(), e);
                return ExitCode::from(11);
            }
        }
    } else {
        // If there is not config file return only config parsed from clap
        EtcdConfig::from(&mut args.config)
    };
    let log = logger(&config);
    match EtcdNode::init(
        config, log.clone(),
        #[cfg(feature = "tracer")] None,
    ).await {
        Ok(c) => {
            if let Err(e) = c.serve().await {
                crit!(log, "Error: {}", e);
                return ExitCode::from(12);
            }
        }
        Err(e) => {
            crit!(log, "Error: {}", e);
            return ExitCode::from(13);
        }
    }

    match wait_for_signal().await {
        Ok(msg) => {
            exit_with_msg(log, msg, 0)
        }
        Err(msg) => {
            exit_with_msg(log, msg, 10)
        }
    }
}

#[cfg(unix)]
use tokio::signal::unix::*;
use etcds::cli::{EtcdCliArgs, EtcdConfig};
use etcds::cluster::EtcdNode;

#[cfg(unix)]
async fn wait_for_signal() -> Result<String, String> {
    let mut int_stream = signal(SignalKind::interrupt()).map_err(|e| e.to_string())?;
    let mut term_stream = signal(SignalKind::terminate()).map_err(|e| e.to_string())?;
    let mut hangup_stream = signal(SignalKind::hangup()).map_err(|e| e.to_string())?;
    let sig = tokio::select! {
		_ = int_stream.recv() => {"interrupt"},
		_ = term_stream.recv() => {"terminate"},
		_ = hangup_stream.recv() => {"hangup"},
	};
    Ok(sig.to_string())
}

#[cfg(windows)]
async fn wait_for_signal() -> Result<String, String> {
    tokio::signal::ctrl_c().await
        .map(|_| format!("got CTRL-C."))
        .map_err(|err| format!("Unable to listen for shutdown signal: {}", err))
}

pub fn exit_with_msg(log: Logger, msg: String, code: u8) -> ExitCode {
    #[cfg(test)]
    assert!(false, "\n{}", msg);

    crit!(log, "\n{}", msg);
    // std::io::stdout().flush().unwrap();
    ExitCode::from(code)
}

pub fn logger(cfg: &EtcdConfig) -> Logger {

    let mut builder = TerminalLoggerBuilder::new();
    builder.channel_size(10240);
    builder.destination(Destination::Stdout);

    #[cfg(debug_assertions)]
    builder.source_location(SourceLocation::LocalFileAndLine);

    #[cfg(not(debug_assertions))]
    builder.source_location(SourceLocation::None);

    builder.level(severity_from_string(&cfg.log_level));
    builder.overflow_strategy(sloggers::types::OverflowStrategy::Drop);
    builder.build().unwrap()
}
#[inline]
fn severity_from_string(severity: &String) -> Severity {
    match severity.to_lowercase().as_str() {
        "error" => Severity::Error,
        "warning" => Severity::Warning,
        "info" => Severity::Info,
        "debug" => Severity::Debug,
        "trace" => Severity::Trace,
        _ => Severity::Critical,
    }
}