use crate::cli::EtcdConfig;
use crate::etcdpb::etcdserverpb::kv_server::{Kv, KvServer};
use crate::etcdpb::etcdserverpb::ResponseHeader;
use crate::queue::{Queue};
use crate::{EtcdEvents, EtcdMgmtEvent, KvEvent};
use rust_i18n::t;
use slog::{error, info, Logger};
use std::collections::{HashMap};
use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use tokio::sync::mpsc::Sender;
use tokio::sync::{mpsc, Mutex, RwLock};
use tonic::Status;
use tonic::transport::Server;
use tonic::transport::server::Router;
use uuid::Uuid;
use crate::etcdpb::etcdserverpb::auth_server::AuthServer;
use crate::etcdpb::etcdserverpb::cluster_server::ClusterServer;
use crate::etcdpb::etcdserverpb::lease_server::LeaseServer;
use crate::etcdpb::etcdserverpb::maintenance_server::MaintenanceServer;
use crate::etcdpb::etcdserverpb::watch_server::{ WatchServer};
use crate::etcdpb::etcdserverpb::WatchResponse;
use crate::etcdpb::v3electionpb::election_server::ElectionServer;
use crate::etcdpb::v3lockpb::lock_server::LockServer;
use crate::peer::EtcdCluster;
use crate::LP;
pub(crate) use crate::peer::EtcdPeerNode;
impl EtcdNode {
pub async fn init(cfg: EtcdConfig, log: Logger,
#[cfg(feature = "tracer")] tracer: Option<opentelemetry_sdk::trace::SdkTracer>,
) -> Result<Self, String> {
let node_id = parse_uuid(&cfg.node)?;
let cluster = EtcdCluster::connect(&cfg, node_id, parse_uuid(&cfg.cluster)?, &log).await?;
let (event, mut rsvr) = mpsc::channel(10);
let (watcher, watcher_rv) = mpsc::channel(10);
let term = Arc::new(AtomicU64::new(cfg.term));
let c = EtcdNode {
cfg: Arc::new(RwLock::new(cfg)),
vault: Arc::new(Default::default()),
queues: Arc::new(Default::default()),
observers: Arc::new(Default::default()),
watchers: Arc::new(Default::default()),
watch_notify: watcher,
peers: Arc::new(RwLock::new(cluster)),
node_id,
term,
event,
log: log.clone(),
#[cfg(feature = "tracer")] tracer: Arc::new(RwLock::new(tracer)),
};
{
let cfg = c.cfg.read().await;
let my_urls: Vec<String> = cfg.listen_client_urls
.split(',')
.filter(|u| !u.is_empty())
.map(|u| {
if u.starts_with("http") { u.to_string() }
else { format!("http://{}", u) }
})
.collect();
drop(cfg);
match c.peers.read().await.announce_and_sync(my_urls, &c.vault, &log).await {
Ok(n) => if n > 0 {
info!(log, "{}startup sync: loaded {} keys", LP, n);
},
Err(e) => info!(log, "{}startup sync skipped: {}", LP, e),
}
}
c.watch_notify(watcher_rv).await;
let grpc_client = c.clone();
tokio::spawn(async move {
while let Some(r) = rsvr.recv().await {
match r {
EtcdEvents::Data(kv) => match kv {
KvEvent::Put(kv) => {
let _ = grpc_client.put(tonic::Request::new(kv)).await;
},
KvEvent::Delete(kv) => {
let _ = grpc_client.delete_range(tonic::Request::new(kv)).await;
}
KvEvent::Txn(kv) => {
let _ = grpc_client.txn(tonic::Request::new(kv)).await;
}
}
EtcdEvents::Mgmt(e) => match e {
EtcdMgmtEvent::Config(c) => {
let _ = grpc_client.reconfigure(c).await;
}
#[cfg(feature = "tracer")]
EtcdMgmtEvent::Tracer(c) => {
*grpc_client.tracer.write().await = c;
}
EtcdMgmtEvent::Stop => {
info!(grpc_client.log, "{}received Stop event, shutting down event loop", LP);
break;
}
EtcdMgmtEvent::Restart => {
let new_term = grpc_client.term.fetch_add(1, Ordering::Relaxed) + 1;
info!(grpc_client.log, "{}restart: term incremented to {}", LP, new_term);
let cfg = grpc_client.cfg.read().await;
let my_urls: Vec<String> = cfg.listen_client_urls
.split(',')
.filter(|u| !u.is_empty())
.map(|u| {
if u.starts_with("http") { u.to_string() }
else { format!("http://{}", u) }
})
.collect();
drop(cfg);
match grpc_client.peers.read().await
.announce_and_sync(my_urls, &grpc_client.vault, &grpc_client.log).await {
Ok(n) => if n > 0 {
info!(grpc_client.log, "{}restart sync: loaded {} keys", LP, n);
},
Err(e) => info!(grpc_client.log, "{}restart sync skipped: {}", LP, e),
}
}
EtcdMgmtEvent::Pause(secs) => {
info!(grpc_client.log, "{}pausing for {} seconds", LP, secs);
tokio::time::sleep(tokio::time::Duration::from_secs(secs as u64)).await;
info!(grpc_client.log, "{}resumed after pause", LP);
}
}
}
}
});
Ok(c)
}
pub async fn serve(&self) -> Result<(), String> {
let mut srv = Server::builder();
let srv = self.add_services(srv.add_service(AuthServer::new(self.clone())), false);
let addrs = self.cfg.read().await.listen_client_urls.clone();
let addrsv: Vec<&str> = addrs.split(",").collect();
let adr: Vec<&str> = if addrsv[0].starts_with("http") {
let url: Vec<&str> = addrsv[0].split("//").collect();
url[1].split(":").collect()
} else {
addrsv[0].split(":").collect()
};
let bind = adr[0];
let adr = if bind.chars().next().unwrap_or(' ').is_numeric() {
addrsv[0].parse::<SocketAddr>().map_err(|e| format!("parsing {}: {}", addrs, e))?
} else {
let port = adr[1].parse::<u16>().map_err(|e| format!("expected port in {}: {}", addrs, e))?;
let ips: Vec<std::net::IpAddr> = dns_lookup::lookup_host(bind).expect(format!("Binding to {}", bind).as_str()).collect();
if ips.len() == 0 {
return Err(format!("No IpAddr found {}", bind));
}
SocketAddr::new(ips[0], port)
};
let name = self.cfg.read().await.name.clone();
info!(self.log, "{}Starting server [{}] at: {}", LP, name, addrs);
tokio::spawn(async move {
match srv.serve(adr) .await {
Ok(()) => {
println!("bye");
}
Err(e) => {
eprintln!("{} {}", t!("error"), e);
println!();
std::process::exit(10);
}
}
});
Ok(())
}
pub fn add_all_services(&self, srv: Router) -> Router {
self.add_services(srv, true)
}
fn add_services(&self, srv: Router, all: bool) -> Router {
let srv = if all {
srv.add_service(AuthServer::new(self.clone()))
} else {
srv
};
srv.add_service(KvServer::new(self.clone()))
.add_service(WatchServer::new(self.clone()))
.add_service(LockServer::new(self.clone()))
.add_service(ElectionServer::new(self.clone()))
.add_service(MaintenanceServer::new(self.clone()))
.add_service(ClusterServer::new(self.clone()))
.add_service(LeaseServer::new(self.clone()))
}
async fn reconfigure(&self, cfg: EtcdConfig) {
let peers = cfg.peers();
match self.peers.write().await.add_connections(peers).await {
Ok(cnt) => if cnt > 0 {
info!(self.log, "{}reconfigure: added {} peer(s)", LP, cnt);
},
Err(e) => {
error!(self.log, "{}Error adding peers: {}", LP, e);
}
}
if cfg.term > 0 {
self.term.store(cfg.term, Ordering::Relaxed);
}
let mut current = self.cfg.write().await;
current.initial_advertise_peer_urls = cfg.initial_advertise_peer_urls;
current.listen_client_urls = cfg.listen_client_urls;
current.listen_peer_urls = cfg.listen_peer_urls;
current.initial_cluster = cfg.initial_cluster;
current.initial_cluster_token = cfg.initial_cluster_token;
current.name = cfg.name;
current.max_txn_ops = cfg.max_txn_ops;
current.max_request_bytes = cfg.max_request_bytes;
current.log_level = cfg.log_level;
current.term = cfg.term;
}
pub(crate) fn response_header(&self) -> ResponseHeader {
ResponseHeader {
cluster_id: 0,
member_id: self.node_id,
revision: 0,
raft_term: self.term.load(Ordering::Relaxed),
}
}
pub async fn get_peer_urls(&self) -> String {
self.peers.read().await.peer_urls().await
}
}
pub type KvKey = Vec<u8>;
pub type NodeId = u64;
pub type ClientId = Uuid;
pub type WatcherId = i64;
#[derive(Clone)]
pub struct EtcdNode {
pub(crate) node_id: NodeId,
pub(crate) term: Arc<AtomicU64>,
pub(crate) cfg: Arc<RwLock<EtcdConfig>>,
pub(crate) vault: Arc<RwLock<HashMap<KvKey, crate::kv::Kv>>>,
pub(crate) queues: Arc<RwLock<HashMap<String, Queue>>>,
pub(crate) observers: Arc<RwLock<HashMap<KvKey, EtcdObserverType>>>,
pub(crate) watchers: Arc<RwLock<HashMap<ClientId, EtcdClientType>>>,
pub(crate) watch_notify: Sender<crate::kv::Kv>,
pub(crate) peers: Arc<RwLock<EtcdCluster>>,
pub event: Sender<EtcdEvents>,
pub(crate) log: Logger,
#[cfg(feature = "tracer")] pub(crate) tracer: Arc<RwLock<Option<opentelemetry_sdk::trace::SdkTracer>>>
}
pub(crate) type EtcdObserverType = Vec<(ClientId, WatcherId)>;
pub(crate) type EtcdPeerNodeType = Arc<Mutex<EtcdPeerNode>>;
pub(crate) type EtcdClientType = Arc<RwLock<EtcdClientNode>>;
pub struct EtcdClientNode {
#[allow(dead_code)]
pub(crate) client_id: ClientId,
pub(crate) watchers: HashMap<WatcherId, WatcherConsumer>,
}
pub struct WatcherConsumer {
pub(crate) key: KvKey,
pub(crate) client: Sender<Result<WatchResponse, Status>>,
}
#[inline]
fn parse_uuid(value: &String) -> Result<u64, String> {
let (a, b) = if value.len() > 0 {
Uuid::from_str(value.as_str())
.map_err(|e| format!("parsing to uuid: [{}]: {}", value, e))?
} else {
Uuid::new_v4()
}.as_u64_pair();
Ok(a^b)
}