-
Notifications
You must be signed in to change notification settings - Fork 0
/
setup.rs
153 lines (128 loc) · 4.96 KB
/
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
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
// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
use std::borrow::ToOwned;
use std::sync::atomic::{AtomicBool, Ordering};
use chrono;
use clap::ArgMatches;
use tikv::config::{check_critical_config, persist_critical_config, MetricConfig, TiKvConfig};
use tikv_util::collections::HashMap;
use tikv_util::{self, logger};
// A workaround for checking if log is initialized.
pub static LOG_INITIALIZED: AtomicBool = AtomicBool::new(false);
#[macro_export]
macro_rules! fatal {
($lvl:expr $(, $arg:expr)*) => ({
if $crate::setup::LOG_INITIALIZED.load(::std::sync::atomic::Ordering::SeqCst) {
crit!($lvl $(, $arg)*);
} else {
eprintln!($lvl $(, $arg)*);
}
slog_global::clear_global();
::std::process::exit(1)
})
}
#[allow(dead_code)]
pub fn initial_logger(config: &TiKvConfig) {
let log_rotation_timespan = chrono::Duration::from_std(config.log_rotation_timespan.into())
.expect("config.log_rotation_timespan is an invalid duration.");
if config.log_file.is_empty() {
let drainer = logger::term_drainer();
// use async drainer and init std log.
logger::init_log(drainer, config.log_level, true, true, vec![]).unwrap_or_else(|e| {
fatal!("failed to initialize log: {}", e);
});
} else {
let drainer =
logger::file_drainer(&config.log_file, log_rotation_timespan).unwrap_or_else(|e| {
fatal!(
"failed to initialize log with file {}: {}",
config.log_file,
e
);
});
// use async drainer and init std log.
logger::init_log(drainer, config.log_level, true, true, vec![]).unwrap_or_else(|e| {
fatal!("failed to initialize log: {}", e);
});
};
LOG_INITIALIZED.store(true, Ordering::SeqCst);
}
#[allow(dead_code)]
pub fn initial_metric(cfg: &MetricConfig, node_id: Option<u64>) {
tikv_util::metrics::monitor_threads("tikv")
.unwrap_or_else(|e| fatal!("failed to start monitor thread: {}", e));
tikv_util::metrics::monitor_allocator_stats("tikv")
.unwrap_or_else(|e| fatal!("failed to monitor allocator stats: {}", e));
if cfg.interval.as_secs() == 0 || cfg.address.is_empty() {
return;
}
let mut push_job = cfg.job.clone();
if let Some(id) = node_id {
push_job.push_str(&format!("_{}", id));
}
info!("start prometheus client");
tikv_util::metrics::run_prometheus(cfg.interval.0, &cfg.address, &push_job);
}
#[allow(dead_code)]
pub fn overwrite_config_with_cmd_args(config: &mut TiKvConfig, matches: &ArgMatches<'_>) {
if let Some(level) = matches.value_of("log-level") {
config.log_level = logger::get_level_by_string(level).unwrap();
}
if let Some(file) = matches.value_of("log-file") {
config.log_file = file.to_owned();
}
if let Some(addr) = matches.value_of("addr") {
config.server.addr = addr.to_owned();
}
if let Some(advertise_addr) = matches.value_of("advertise-addr") {
config.server.advertise_addr = advertise_addr.to_owned();
}
if let Some(status_addr) = matches.value_of("status-addr") {
config.server.status_addr = status_addr.to_owned();
}
if let Some(data_dir) = matches.value_of("data-dir") {
config.storage.data_dir = data_dir.to_owned();
}
if let Some(endpoints) = matches.values_of("pd-endpoints") {
config.pd.endpoints = endpoints.map(ToOwned::to_owned).collect();
}
if let Some(labels_vec) = matches.values_of("labels") {
let mut labels = HashMap::default();
for label in labels_vec {
let mut parts = label.split('=');
let key = parts.next().unwrap().to_owned();
let value = match parts.next() {
None => fatal!("invalid label: {}", label),
Some(v) => v.to_owned(),
};
if parts.next().is_some() {
fatal!("invalid label: {}", label);
}
labels.insert(key, value);
}
config.server.labels = labels;
}
if let Some(capacity_str) = matches.value_of("capacity") {
let capacity = capacity_str.parse().unwrap_or_else(|e| {
fatal!("invalid capacity: {}", e);
});
config.raft_store.capacity = capacity;
}
if let Some(metrics_addr) = matches.value_of("metrics-addr") {
config.metric.address = metrics_addr.to_owned()
}
}
#[allow(dead_code)]
pub fn validate_and_persist_config(config: &mut TiKvConfig, persist: bool) {
if let Err(e) = check_critical_config(config) {
fatal!("critical config check failed: {}", e);
}
if persist {
if let Err(e) = persist_critical_config(&config) {
fatal!("persist critical config failed: {}", e);
}
}
config.compatible_adjust();
if let Err(e) = config.validate() {
fatal!("invalid configuration: {}", e.description());
}
}