daqbuf/daqingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Add http apis to query status, add and remove channel

Browse Source
This commit is contained in:
Dominik Werder
2022-07-22 16:02:13 +02:00
parent dc89de73ce
commit f50a1513b3
7 changed files with 820 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netfetch/Cargo.toml
+2
View File
@@ -29,7 +29,9 @@ libc = "0.2"
regex = "1.5.5"
axum = "0.5"
http = "0.2"
url = "2.2"
hyper = "0.14"
chrono = "0.4"
log = { path = "../log" }
stats = { path = "../stats" }
err = { path = "../../daqbuffer/err" }
netfetch/src/ca.rs
+249 -94
View File
@@ -15,7 +15,7 @@ use netpod::{Database, ScyllaConfig};
use scylla::batch::Consistency;
use serde::{Deserialize, Serialize};
use stats::{CaConnStats, CaConnStatsAgg, CaConnStatsAggDiff};
use std::collections::{BTreeMap, VecDeque};
use std::collections::BTreeMap;
use std::net::{Ipv4Addr, SocketAddrV4};
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
@@ -23,6 +23,8 @@ use std::sync::{Arc, Mutex, Once};
use std::time::Duration;
use tokio::fs::OpenOptions;
use tokio::io::AsyncReadExt;
use tokio::sync::Mutex as TokMx;
use tokio::task::JoinHandle;
use tokio_postgres::Client as PgClient;
static mut METRICS: Option<Mutex<Option<CaConnStatsAgg>>> = None;
@@ -40,6 +42,7 @@ pub fn get_metrics() -> &'static mut Option<CaConnStatsAgg> {
#[derive(Debug, Serialize, Deserialize)]
struct ChannelConfig {
backend: String,
channels: Vec<String>,
search: Vec<String>,
addr_bind: Ipv4Addr,
@@ -87,6 +90,7 @@ pub async fn parse_config(config: PathBuf) -> Result<CaConnectOpts, Error> {
})
.collect();
Ok(CaConnectOpts {
backend: conf.backend,
channels: conf.channels,
search: conf.search,
addr_bind: conf.addr_bind,
@@ -106,6 +110,7 @@ pub async fn parse_config(config: PathBuf) -> Result<CaConnectOpts, Error> {
}
pub struct CaConnectOpts {
pub backend: String,
pub channels: Vec<String>,
pub search: Vec<String>,
pub addr_bind: Ipv4Addr,
@@ -127,7 +132,7 @@ async fn spawn_scylla_insert_workers(
scyconf: ScyllaConfig,
insert_scylla_sessions: usize,
insert_worker_count: usize,
insert_item_queue: &CommonInsertItemQueue,
insert_item_queue: Arc<CommonInsertItemQueue>,
insert_frac: Arc<AtomicU64>,
pg_client: Arc<PgClient>,
store_stats: Arc<stats::CaConnStats>,
@@ -232,42 +237,135 @@ async fn spawn_scylla_insert_workers(
}
pub struct CommandQueueSet {
queues: tokio::sync::Mutex<VecDeque<Sender<ConnCommand>>>,
queues: tokio::sync::Mutex<BTreeMap<SocketAddrV4, Sender<ConnCommand>>>,
}
impl CommandQueueSet {
pub fn new() -> Self {
Self {
queues: tokio::sync::Mutex::new(VecDeque::<Sender<ConnCommand>>::new()),
queues: tokio::sync::Mutex::new(BTreeMap::<SocketAddrV4, Sender<ConnCommand>>::new()),
}
}
pub fn queues(&self) -> &tokio::sync::Mutex<VecDeque<Sender<ConnCommand>>> {
pub fn queues(&self) -> &tokio::sync::Mutex<BTreeMap<SocketAddrV4, Sender<ConnCommand>>> {
&self.queues
}
}
pub struct IngestCommons {
pub pgconf: Arc<Database>,
pub local_epics_hostname: String,
pub insert_item_queue: Arc<CommonInsertItemQueue>,
pub data_store: Arc<DataStore>,
pub insert_ivl_min: Arc<AtomicU64>,
pub conn_stats: Arc<TokMx<Vec<Arc<CaConnStats>>>>,
pub command_queue_set: Arc<CommandQueueSet>,
}
pub async fn find_channel_addr(
backend: String,
name: String,
pgconf: &Database,
) -> Result<Option<SocketAddrV4>, Error> {
// TODO also here, provide a db pool.
let d = pgconf;
let (pg_client, pg_conn) = tokio_postgres::connect(
&format!("postgresql://{}:{}@{}:{}/{}", d.user, d.pass, d.host, d.port, d.name),
tokio_postgres::tls::NoTls,
)
.await
.unwrap();
// TODO allow clean shutdown on ctrl-c and join the pg_conn in the end:
tokio::spawn(pg_conn);
let pg_client = Arc::new(pg_client);
let qu_find_addr = pg_client
.prepare("with q1 as (select t1.facility, t1.channel, t1.addr from ioc_by_channel t1 where t1.facility = $1 and t1.channel in ($2) and t1.addr != '' order by t1.tsmod desc) select distinct on (q1.facility, q1.channel) q1.facility, q1.channel, q1.addr from q1")
.await
.map_err(|e| Error::with_msg_no_trace(format!("{e:?}")))?;
let rows = pg_client
.query(&qu_find_addr, &[&backend, &name])
.await
.map_err(|e| Error::with_msg_no_trace(format!("pg lookup error: {e:?}")))?;
if rows.is_empty() {
error!("can not find any addresses of channels {:?}", name);
} else {
for row in rows {
let addr: &str = row.get(2);
if addr == "" {
return Ok(None);
} else {
match addr.parse::<SocketAddrV4>() {
Ok(addr) => return Ok(Some(addr)),
Err(_) => return Ok(None),
}
}
}
}
Ok(None)
}
pub async fn create_ca_conn(
addr: SocketAddrV4,
local_epics_hostname: String,
array_truncate: usize,
insert_queue_max: usize,
insert_item_queue: Arc<CommonInsertItemQueue>,
data_store: Arc<DataStore>,
insert_ivl_min: Arc<AtomicU64>,
conn_stats: Arc<TokMx<Vec<Arc<CaConnStats>>>>,
command_queue_set: Arc<CommandQueueSet>,
) -> Result<JoinHandle<Result<(), Error>>, Error> {
info!("create new CaConn {:?}", addr);
let data_store = data_store.clone();
let conn = CaConn::new(
addr,
local_epics_hostname,
data_store.clone(),
insert_item_queue.sender(),
array_truncate,
insert_queue_max,
insert_ivl_min.clone(),
);
conn_stats.lock().await.push(conn.stats());
let stats2 = conn.stats();
let conn_command_tx = conn.conn_command_tx();
{
command_queue_set.queues().lock().await.insert(addr, conn_command_tx);
}
let conn_block = async move {
let mut conn = conn;
while let Some(item) = conn.next().await {
match item {
Ok(_) => {
stats2.conn_item_count_inc();
}
Err(e) => {
error!("CaConn gives error: {e:?}");
break;
}
}
}
Ok::<_, Error>(())
};
let jh = tokio::spawn(conn_block);
Ok(jh)
}
pub async fn ca_connect(opts: ListenFromFileOpts) -> Result<(), Error> {
let facility = "scylla";
let insert_frac = Arc::new(AtomicU64::new(1000));
let insert_ivl_min = Arc::new(AtomicU64::new(8800));
let opts = parse_config(opts.config).await?;
let scyconf = opts.scyconf.clone();
let command_queue_set = Arc::new(CommandQueueSet::new());
tokio::spawn(crate::metrics::start_metrics_service(
opts.api_bind.clone(),
insert_frac.clone(),
insert_ivl_min.clone(),
command_queue_set.clone(),
));
let d = Database {
let pgconf = Database {
name: opts.pgconf.name.clone(),
host: opts.pgconf.host.clone(),
port: opts.pgconf.port.clone(),
user: opts.pgconf.user.clone(),
pass: opts.pgconf.pass.clone(),
};
let d = &pgconf;
let (pg_client, pg_conn) = tokio_postgres::connect(
&format!("postgresql://{}:{}@{}:{}/{}", d.user, d.pass, d.host, d.port, d.name),
tokio_postgres::tls::NoTls,
@@ -294,6 +392,74 @@ pub async fn ca_connect(opts: ListenFromFileOpts) -> Result<(), Error> {
.await
.map_err(|e| Error::with_msg_no_trace(format!("{e:?}")))?;
let mut channels_by_host = BTreeMap::new();
let data_store = Arc::new(DataStore::new(pg_client.clone(), scy.clone()).await?);
let insert_item_queue = CommonInsertItemQueue::new(opts.insert_item_queue_cap);
let insert_item_queue = Arc::new(insert_item_queue);
// TODO use a new stats struct
let store_stats = Arc::new(CaConnStats::new());
spawn_scylla_insert_workers(
opts.scyconf.clone(),
opts.insert_scylla_sessions,
opts.insert_worker_count,
insert_item_queue.clone(),
insert_frac.clone(),
pg_client.clone(),
store_stats.clone(),
)
.await?;
let mut conn_jhs = vec![];
let conn_stats: Arc<TokMx<Vec<Arc<CaConnStats>>>> = Arc::new(TokMx::new(Vec::new()));
let command_queue_set = Arc::new(CommandQueueSet::new());
let ingest_commons = IngestCommons {
pgconf: Arc::new(pgconf.clone()),
local_epics_hostname: opts.local_epics_hostname.clone(),
insert_item_queue: insert_item_queue.clone(),
data_store: data_store.clone(),
insert_ivl_min: insert_ivl_min.clone(),
conn_stats: conn_stats.clone(),
command_queue_set: command_queue_set.clone(),
};
let ingest_commons = Arc::new(ingest_commons);
tokio::spawn(crate::metrics::start_metrics_service(
opts.api_bind.clone(),
insert_frac.clone(),
insert_ivl_min.clone(),
command_queue_set.clone(),
ingest_commons.clone(),
));
let metrics_agg_fut = {
let conn_stats = conn_stats.clone();
let local_stats = local_stats.clone();
async move {
let mut agg_last = CaConnStatsAgg::new();
loop {
tokio::time::sleep(Duration::from_millis(671)).await;
let agg = CaConnStatsAgg::new();
agg.push(&local_stats);
agg.push(&store_stats);
for g in conn_stats.lock().await.iter() {
agg.push(&g);
}
let m = get_metrics();
*m = Some(agg.clone());
if false {
let diff = CaConnStatsAggDiff::diff_from(&agg_last, &agg);
info!("{}", diff.display());
}
agg_last = agg;
if false {
break;
}
}
}
};
let metrics_agg_jh = tokio::spawn(metrics_agg_fut);
let mut chns_todo = &opts.channels[..];
let mut chstmp = ["__NONE__"; 8];
let mut ix = 0;
@@ -306,7 +472,14 @@ pub async fn ca_connect(opts: ListenFromFileOpts) -> Result<(), Error> {
.query(
&qu_find_addr,
&[
&facility, &chstmp[0], &chstmp[1], &chstmp[2], &chstmp[3], &chstmp[4], &chstmp[5], &chstmp[6],
&opts.backend,
&chstmp[0],
&chstmp[1],
&chstmp[2],
&chstmp[3],
&chstmp[4],
&chstmp[5],
&chstmp[6],
&chstmp[7],
],
)
@@ -340,90 +513,71 @@ pub async fn ca_connect(opts: ListenFromFileOpts) -> Result<(), Error> {
} else {
channels_by_host.get_mut(&addr).unwrap().push(ch.to_string());
}
{
let create_new = {
let g = command_queue_set.queues().lock().await;
if let Some(tx) = g.get(&addr) {
let (cmd, rx) = ConnCommand::channel_add(ch.to_string());
tx.send(cmd).await.unwrap();
if !rx.recv().await.unwrap() {
error!("Could not add channel: {}", ch);
}
false
} else {
true
}
};
if create_new {
info!("create new CaConn {:?} {:?}", addr, ch);
let data_store = data_store.clone();
let conn = CaConn::new(
addr,
opts.local_epics_hostname.clone(),
data_store.clone(),
insert_item_queue.sender(),
opts.array_truncate,
opts.insert_queue_max,
insert_ivl_min.clone(),
);
conn_stats.lock().await.push(conn.stats());
let stats2 = conn.stats();
let conn_command_tx = conn.conn_command_tx();
let tx = conn_command_tx.clone();
{
command_queue_set.queues().lock().await.insert(addr, conn_command_tx);
}
let conn_block = async move {
let mut conn = conn;
while let Some(item) = conn.next().await {
match item {
Ok(_) => {
stats2.conn_item_count_inc();
}
Err(e) => {
error!("CaConn gives error: {e:?}");
break;
}
}
}
Ok::<_, Error>(())
};
let jh = tokio::spawn(conn_block);
conn_jhs.push(jh);
{
let (cmd, rx) = ConnCommand::channel_add(ch.to_string());
tx.send(cmd).await.unwrap();
if !rx.recv().await.unwrap() {
error!("Could not add channel: {}", ch);
}
}
}
}
}
}
}
}
if opts.abort_after_search == 1 {
return Ok(());
}
let data_store = Arc::new(DataStore::new(pg_client.clone(), scy.clone()).await?);
let insert_item_queue = CommonInsertItemQueue::new(opts.insert_item_queue_cap);
// TODO use a new stats struct
let store_stats = Arc::new(CaConnStats::new());
spawn_scylla_insert_workers(
opts.scyconf.clone(),
opts.insert_scylla_sessions,
opts.insert_worker_count,
&insert_item_queue,
insert_frac.clone(),
pg_client.clone(),
store_stats.clone(),
)
.await?;
let mut conn_jhs = vec![];
let mut conn_stats = vec![];
info!("channels_by_host len {}", channels_by_host.len());
for (host, channels) in channels_by_host {
let data_store = data_store.clone();
let addr = SocketAddrV4::new(host.ip().clone(), host.port());
let mut conn = CaConn::new(
addr,
opts.local_epics_hostname.clone(),
data_store.clone(),
insert_item_queue.sender(),
opts.array_truncate,
opts.insert_queue_max,
insert_ivl_min.clone(),
);
conn_stats.push(conn.stats());
for c in channels {
conn.channel_add(c);
}
let stats2 = conn.stats();
let conn_command_tx = conn.conn_command_tx();
command_queue_set.queues().lock().await.push_back(conn_command_tx);
let conn_block = async move {
while let Some(item) = conn.next().await {
match item {
Ok(_) => {
stats2.conn_item_count_inc();
}
Err(e) => {
error!("CaConn gives error: {e:?}");
break;
}
}
}
Ok::<_, Error>(())
};
let jh = tokio::spawn(conn_block);
conn_jhs.push(jh);
}
let mut agg_last = CaConnStatsAgg::new();
loop {
tokio::time::sleep(Duration::from_millis(671)).await;
let agg = CaConnStatsAgg::new();
agg.push(&local_stats);
agg.push(&store_stats);
for g in &conn_stats {
agg.push(&g);
}
let m = get_metrics();
*m = Some(agg.clone());
if false {
let diff = CaConnStatsAggDiff::diff_from(&agg_last, &agg);
info!("{}", diff.display());
}
for _s1 in &conn_stats {}
agg_last = agg;
if false {
break;
}
}
for jh in conn_jhs {
match jh.await {
Ok(k) => match k {
@@ -437,5 +591,6 @@ pub async fn ca_connect(opts: ListenFromFileOpts) -> Result<(), Error> {
}
}
}
metrics_agg_jh.await.unwrap();
Ok(())
}
netfetch/src/ca/conn.rs
+216 -21
View File
@@ -13,6 +13,7 @@ use libc::c_int;
use log::*;
use netpod::timeunits::*;
use netpod::{ScalarType, Shape};
use serde::Serialize;
use stats::{CaConnStats, IntervalEma};
use std::collections::{BTreeMap, VecDeque};
use std::net::{Ipv4Addr, SocketAddrV4};
@@ -24,18 +25,60 @@ use std::time::{Duration, Instant, SystemTime};
use tokio::io::unix::AsyncFd;
use tokio::net::TcpStream;
#[derive(Debug)]
#[derive(Clone, Debug, Serialize)]
pub struct ChannelStateInfo {
pub name: String,
pub scalar_type: Option<ScalarType>,
pub shape: Option<Shape>,
// NOTE: this solution can yield to the same Instant serialize to different string representations.
#[serde(skip_serializing_if = "Option::is_none", serialize_with = "ser_instant")]
pub ts_created: Option<Instant>,
#[serde(skip_serializing_if = "Option::is_none", serialize_with = "ser_instant")]
pub ts_event_last: Option<Instant>,
#[serde(skip_serializing_if = "Option::is_none")]
pub item_recv_ivl_ema: Option<f32>,
pub interest_score: f32,
}
fn ser_instant<S: serde::Serializer>(val: &Option<Instant>, ser: S) -> Result<S::Ok, S::Error> {
match val {
Some(val) => {
let now = chrono::Utc::now();
let tsnow = Instant::now();
let t1 = if tsnow >= *val {
let dur = tsnow.duration_since(*val);
let dur2 = chrono::Duration::seconds(dur.as_secs() as i64)
.checked_add(&chrono::Duration::microseconds(dur.subsec_micros() as i64))
.unwrap();
now.checked_sub_signed(dur2).unwrap()
} else {
let dur = (*val).duration_since(tsnow);
let dur2 = chrono::Duration::seconds(dur.as_secs() as i64)
.checked_sub(&chrono::Duration::microseconds(dur.subsec_micros() as i64))
.unwrap();
now.checked_add_signed(dur2).unwrap()
};
//info!("formatting {:?}", t1);
let s = t1.format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string();
//info!("final string {:?}", s);
ser.serialize_str(&s)
}
None => ser.serialize_none(),
}
}
#[derive(Clone, Debug)]
enum ChannelError {
#[allow(unused)]
NoSuccess,
}
#[derive(Debug)]
#[derive(Clone, Debug)]
struct EventedState {
ts_last: Instant,
}
#[derive(Debug)]
#[derive(Clone, Debug)]
enum MonitoringState {
FetchSeriesId,
AddingEvent(SeriesId),
@@ -49,7 +92,7 @@ enum MonitoringState {
Muted,
}
#[derive(Debug)]
#[derive(Clone, Debug)]
struct CreatedState {
#[allow(unused)]
cid: u32,
@@ -71,7 +114,7 @@ struct CreatedState {
}
#[allow(unused)]
#[derive(Debug)]
#[derive(Clone, Debug)]
enum ChannelState {
Init,
Creating { cid: u32, ts_beg: Instant },
@@ -79,6 +122,51 @@ enum ChannelState {
Error(ChannelError),
}
impl ChannelState {
fn to_info(&self, name: String) -> ChannelStateInfo {
let scalar_type = match self {
ChannelState::Created(s) => Some(s.scalar_type.clone()),
_ => None,
};
let shape = match self {
ChannelState::Created(s) => Some(s.shape.clone()),
_ => None,
};
let ts_created = match self {
ChannelState::Created(s) => Some(s.ts_created.clone()),
_ => None,
};
let ts_event_last = match self {
ChannelState::Created(s) => match &s.state {
MonitoringState::Evented(_, s) => Some(s.ts_last),
_ => None,
},
_ => None,
};
let item_recv_ivl_ema = match self {
ChannelState::Created(s) => {
let ema = s.item_recv_ivl_ema.ema();
if ema.update_count() == 0 {
None
} else {
Some(ema.ema())
}
}
_ => None,
};
let interest_score = 1. / item_recv_ivl_ema.unwrap_or(1e10).max(1e-6).min(1e10);
ChannelStateInfo {
name,
scalar_type,
shape,
ts_created,
ts_event_last,
item_recv_ivl_ema,
interest_score,
}
}
}
enum CaConnState {
Unconnected,
Connecting(Pin<Box<dyn Future<Output = Result<TcpStream, Error>> + Send>>),
@@ -112,6 +200,10 @@ impl IdStore {
#[derive(Debug)]
pub enum ConnCommandKind {
FindChannel(String, Sender<(SocketAddrV4, Vec<String>)>),
ChannelState(String, Sender<(SocketAddrV4, Option<ChannelStateInfo>)>),
ChannelStatesAll((), Sender<(SocketAddrV4, Vec<ChannelStateInfo>)>),
ChannelAdd(String, Sender<bool>),
ChannelRemove(String, Sender<bool>),
}
#[derive(Debug)]
@@ -127,6 +219,46 @@ impl ConnCommand {
};
(cmd, rx)
}
pub fn channel_state(
name: String,
) -> (
ConnCommand,
async_channel::Receiver<(SocketAddrV4, Option<ChannelStateInfo>)>,
) {
let (tx, rx) = async_channel::bounded(1);
let cmd = Self {
kind: ConnCommandKind::ChannelState(name, tx),
};
(cmd, rx)
}
pub fn channel_states_all() -> (
ConnCommand,
async_channel::Receiver<(SocketAddrV4, Vec<ChannelStateInfo>)>,
) {
let (tx, rx) = async_channel::bounded(1);
let cmd = Self {
kind: ConnCommandKind::ChannelStatesAll((), tx),
};
(cmd, rx)
}
pub fn channel_add(name: String) -> (ConnCommand, async_channel::Receiver<bool>) {
let (tx, rx) = async_channel::bounded(1);
let cmd = Self {
kind: ConnCommandKind::ChannelAdd(name, tx),
};
(cmd, rx)
}
pub fn channel_remove(name: String) -> (ConnCommand, async_channel::Receiver<bool>) {
let (tx, rx) = async_channel::bounded(1);
let cmd = Self {
kind: ConnCommandKind::ChannelRemove(name, tx),
};
(cmd, rx)
}
}
#[allow(unused)]
@@ -136,6 +268,7 @@ pub struct CaConn {
cid_store: IdStore,
ioid_store: IdStore,
subid_store: IdStore,
// TODO use a Cid or so instead of u32.
channels: BTreeMap<u32, ChannelState>,
init_state_count: u64,
cid_by_name: BTreeMap<String, u32>,
@@ -210,7 +343,7 @@ impl CaConn {
match self.conn_command_rx.poll_next_unpin(cx) {
Ready(Some(a)) => match a.kind {
ConnCommandKind::FindChannel(pattern, tx) => {
info!("Search for {pattern:?}");
//info!("Search for {pattern:?}");
let mut res = Vec::new();
for name in self.name_by_cid.values() {
if !pattern.is_empty() && name.contains(&pattern) {
@@ -225,6 +358,65 @@ impl CaConn {
}
}
}
ConnCommandKind::ChannelState(name, tx) => {
//info!("State for {name:?}");
let res = match self.cid_by_name.get(&name) {
Some(cid) => match self.channels.get(cid) {
Some(state) => Some(state.to_info(name)),
None => None,
},
None => None,
};
let msg = (self.remote_addr_dbg.clone(), res);
if msg.1.is_some() {
info!("Sending back {msg:?}");
}
match tx.try_send(msg) {
Ok(_) => {}
Err(_) => {
error!("response channel full or closed");
}
}
}
ConnCommandKind::ChannelStatesAll((), tx) => {
let res = self
.channels
.iter()
.map(|(cid, state)| {
let name = self
.name_by_cid
.get(cid)
.map_or("--unknown--".into(), |x| x.to_string());
state.to_info(name)
})
.collect();
let msg = (self.remote_addr_dbg.clone(), res);
match tx.try_send(msg) {
Ok(_) => {}
Err(_) => {
error!("response channel full or closed");
}
}
}
ConnCommandKind::ChannelAdd(name, tx) => {
self.channel_add(name);
match tx.try_send(true) {
Ok(_) => {}
Err(_) => {
error!("response channel full or closed");
}
}
}
ConnCommandKind::ChannelRemove(name, tx) => {
info!("remove {}", name);
self.channel_remove(name);
match tx.try_send(true) {
Ok(_) => {}
Err(_) => {
error!("response channel full or closed");
}
}
}
},
Ready(None) => {
error!("Command queue closed");
@@ -250,6 +442,13 @@ impl CaConn {
}
}
pub fn channel_remove(&mut self, channel: String) {
let cid = self.cid_by_name(&channel);
if self.channels.contains_key(&cid) {
warn!("TODO actually cause the channel to get closed and removed {}", channel);
}
}
fn cid_by_name(&mut self, name: &str) -> u32 {
if let Some(cid) = self.cid_by_name.get(name) {
*cid
@@ -441,32 +640,25 @@ impl CaConn {
fn handle_event_add_res(&mut self, ev: proto::EventAddRes, tsnow: Instant) -> Result<(), Error> {
// TODO handle subid-not-found which can also be peer error:
let cid = *self.cid_by_subid.get(&ev.subid).unwrap();
//let name = self.name_by_cid(cid).unwrap().to_string();
let _name = self.name_by_cid(cid).unwrap().to_string();
// TODO get rid of the string clone when I don't want the log output any longer:
// TODO handle not-found error:
let mut series_2 = None;
let ch_s = self.channels.get_mut(&cid).unwrap();
match ch_s {
ChannelState::Created(st) => {
st.item_recv_ivl_ema.tick(Instant::now());
st.item_recv_ivl_ema.tick(tsnow);
let scalar_type = st.scalar_type.clone();
let shape = st.shape.clone();
match st.state {
MonitoringState::AddingEvent(ref series) => {
let series = series.clone();
series_2 = Some(series.clone());
// TODO get ts from faster common source:
st.state = MonitoringState::Evented(
series,
EventedState {
ts_last: Instant::now(),
},
);
st.state = MonitoringState::Evented(series, EventedState { ts_last: tsnow });
}
MonitoringState::Evented(ref series, ref mut st) => {
series_2 = Some(series.clone());
// TODO get ts from faster common source:
st.ts_last = Instant::now();
st.ts_last = tsnow;
}
_ => {
error!("unexpected state: EventAddRes while having {:?}", st.state);
@@ -811,11 +1003,13 @@ impl Stream for CaConn {
match tokio::time::timeout(Duration::from_millis(500), TcpStream::connect(addr)).await {
Ok(Ok(k)) => Ok(k),
Ok(Err(e)) => {
error!("Can not connect to {addr:?} {e:?}");
// TODO keep this in channel status field, or log when we have exponential backoff
trace!("Can not connect to {addr:?} {e:?}");
Err(e.into())
}
Err(e) => {
error!("Can not connect to {addr:?} {e:?}");
Err(_) => {
// TODO keep this in channel status field, or log when we have exponential backoff
trace!("Can not connect to {addr:?} timeout");
Err(Error::with_msg_no_trace(format!("timeout")))
}
}
@@ -832,7 +1026,8 @@ impl Stream for CaConn {
continue 'outer;
}
Ready(Err(e)) => {
error!("Connection error: {e:?}");
// TODO keep this in channel status field, or log when we have exponential backoff
trace!("Connection error: {e:?}");
// We can not connect to the remote.
// TODO do exponential backoff.
self.state = CaConnState::Wait(wait_fut(10000));
netfetch/src/ca/proto.rs
+22 -1
View File
@@ -601,7 +601,28 @@ impl CaMsg {
let ca_secs = u32::from_be_bytes(payload[4..8].try_into()?);
let ca_nanos = u32::from_be_bytes(payload[8..12].try_into()?);
let ca_sh = Shape::from_ca_count(hi.data_count)?;
let valbuf = &payload[12..];
let meta_padding = match ca_dbr_ty.meta {
CaDbrMetaType::Plain => 0,
CaDbrMetaType::Status => match ca_dbr_ty.scalar_type {
CaScalarType::I8 => 1,
CaScalarType::I16 => 0,
CaScalarType::I32 => 0,
CaScalarType::F32 => 0,
CaScalarType::F64 => 4,
CaScalarType::Enum => 0,
CaScalarType::String => 0,
},
CaDbrMetaType::Time => match ca_dbr_ty.scalar_type {
CaScalarType::I8 => 3,
CaScalarType::I16 => 2,
CaScalarType::I32 => 0,
CaScalarType::F32 => 0,
CaScalarType::F64 => 4,
CaScalarType::Enum => 2,
CaScalarType::String => 0,
},
};
let valbuf = &payload[12 + meta_padding..];
let value = match ca_sh {
Shape::Scalar => Self::ca_scalar_value(&ca_dbr_ty.scalar_type, valbuf)?,
Shape::Wave(n) => {
netfetch/src/metrics.rs
+320 -24
View File
@@ -1,22 +1,161 @@
use crate::ca::conn::ConnCommand;
use crate::ca::CommandQueueSet;
use crate::ca::{CommandQueueSet, IngestCommons};
use log::*;
use serde::Deserialize;
use std::collections::HashMap;
use std::net::SocketAddrV4;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
#[derive(Debug, Deserialize)]
struct QueryForm {
query: String,
time: Option<f64>,
#[allow(unused)]
timeout: Option<String>,
}
#[allow(unused)]
#[derive(Debug, Deserialize)]
struct PromLabels {
start: Option<String>,
end: Option<String>,
//#[serde(rename = "match[]")]
//pattern: Option<Vec<String>>,
}
#[allow(unused)]
#[derive(Debug, Deserialize)]
struct PromLabelValues {
start: Option<String>,
end: Option<String>,
//#[serde(rename = "match[]")]
//pattern: Option<Vec<String>>,
}
async fn get_empty() -> String {
format!("")
}
async fn channel_add(params: HashMap<String, String>, ingest_commons: Arc<IngestCommons>) -> String {
if let (Some(backend), Some(name)) = (params.get("backend"), params.get("name")) {
// TODO look up the address.
match crate::ca::find_channel_addr(backend.into(), name.into(), &ingest_commons.pgconf).await {
Ok(Some(addr)) => {
if ingest_commons
.command_queue_set
.queues()
.lock()
.await
.contains_key(&addr)
{
} else {
match crate::ca::create_ca_conn(
addr,
ingest_commons.local_epics_hostname.clone(),
256,
32,
ingest_commons.insert_item_queue.clone(),
ingest_commons.data_store.clone(),
ingest_commons.insert_ivl_min.clone(),
ingest_commons.conn_stats.clone(),
ingest_commons.command_queue_set.clone(),
)
.await
{
Ok(_) => {
// TODO keep the join handle.
}
Err(_) => {
error!("can not create CaConn");
}
}
}
if let Some(tx) = ingest_commons.command_queue_set.queues().lock().await.get(&addr) {
let (cmd, rx) = ConnCommand::channel_add(name.into());
if let Err(_) = tx.send(cmd).await {
error!("can not send command");
"false".into()
} else {
match rx.recv().await {
Ok(x) => {
if x {
"true".into()
} else {
"false".into()
}
}
Err(_) => "false".into(),
}
}
} else {
error!("Even after create, can not locate the connection.");
"false".into()
}
}
_ => {
error!("can not find addr for channel");
"false".into()
}
}
} else {
"false".into()
}
}
async fn channel_remove(
params: HashMap<String, String>,
ingest_commons: Arc<IngestCommons>,
) -> axum::Json<serde_json::Value> {
use axum::Json;
use serde_json::Value;
let addr = if let Some(x) = params.get("addr") {
if let Ok(addr) = x.parse::<SocketAddrV4>() {
addr
} else {
return Json(Value::Bool(false));
}
} else {
return Json(Value::Bool(false));
};
let backend = if let Some(x) = params.get("backend") {
x
} else {
return Json(Value::Bool(false));
};
let name = if let Some(x) = params.get("name") {
x
} else {
return Json(Value::Bool(false));
};
if let Some(tx) = ingest_commons.command_queue_set.queues().lock().await.get(&addr) {
// TODO any need to check the backend here?
let _ = backend;
let (cmd, rx) = ConnCommand::channel_remove(name.into());
if let Err(_) = tx.send(cmd).await {
error!("can not send command");
Json(Value::Bool(false))
} else {
match rx.recv().await {
Ok(x) => Json(Value::Bool(x)),
Err(_) => Json(Value::Bool(false)),
}
}
} else {
Json(Value::Bool(false))
}
}
pub async fn start_metrics_service(
bind_to: String,
insert_frac: Arc<AtomicU64>,
insert_ivl_min: Arc<AtomicU64>,
command_queue_set: Arc<CommandQueueSet>,
ingest_commons: Arc<IngestCommons>,
) {
use axum::extract::Query;
use axum::routing::{get, post, put};
use axum::Form;
use axum::{extract, Router};
use http::request::Parts;
let app = Router::new()
@@ -38,26 +177,97 @@ pub async fn start_metrics_service(
)
.route(
"/daqingest/find/channel",
get(|Query(params): Query<HashMap<String, String>>| async move {
let pattern = params.get("pattern").map_or(String::new(), |x| x.clone()).to_string();
let g = command_queue_set.queues().lock().await;
let mut rxs = Vec::new();
for tx in g.iter() {
let (cmd, rx) = ConnCommand::find_channel(pattern.clone());
rxs.push(rx);
if let Err(_) = tx.send(cmd).await {
error!("can not send command");
get({
let command_queue_set = command_queue_set.clone();
|Query(params): Query<HashMap<String, String>>| async move {
let pattern = params.get("pattern").map_or(String::new(), |x| x.clone()).to_string();
let g = command_queue_set.queues().lock().await;
let mut rxs = Vec::new();
for (_, tx) in g.iter() {
let (cmd, rx) = ConnCommand::find_channel(pattern.clone());
rxs.push(rx);
if let Err(_) = tx.send(cmd).await {
error!("can not send command");
}
}
}
let mut res = Vec::new();
for rx in rxs {
let item = rx.recv().await.unwrap();
let item = (item.0.to_string(), item.1);
if item.1.len() > 0 {
res.push(item);
let mut res = Vec::new();
for rx in rxs {
let item = rx.recv().await.unwrap();
if item.1.len() > 0 {
let item = (item.0.to_string(), item.1);
res.push(item);
}
}
serde_json::to_string(&res).unwrap()
}
serde_json::to_string(&res).unwrap()
}),
)
.route(
"/daqingest/channel/state",
get({
let command_queue_set = command_queue_set.clone();
|Query(params): Query<HashMap<String, String>>| async move {
let name = params.get("name").map_or(String::new(), |x| x.clone()).to_string();
let g = command_queue_set.queues().lock().await;
let mut rxs = Vec::new();
for (_, tx) in g.iter() {
let (cmd, rx) = ConnCommand::channel_state(name.clone());
rxs.push(rx);
if let Err(_) = tx.send(cmd).await {
error!("can not send command");
}
}
let mut res = Vec::new();
for rx in rxs {
let item = rx.recv().await.unwrap();
if let Some(st) = item.1 {
let item = (item.0.to_string(), st);
res.push(item);
}
}
serde_json::to_string(&res).unwrap()
}
}),
)
.route(
"/daqingest/channel/states",
get({
let command_queue_set = command_queue_set.clone();
|Query(_params): Query<HashMap<String, String>>| async move {
let g = command_queue_set.queues().lock().await;
let mut rxs = Vec::new();
for (_, tx) in g.iter() {
let (cmd, rx) = ConnCommand::channel_states_all();
rxs.push(rx);
if let Err(_) = tx.send(cmd).await {
error!("can not send command");
}
}
let mut res = Vec::new();
for rx in rxs {
let item = rx.recv().await.unwrap();
for h in item.1 {
res.push((item.0.clone(), h));
}
}
res.sort_unstable_by_key(|(_, v)| v.interest_score as u32);
let res: Vec<_> = res.into_iter().rev().take(10).collect();
serde_json::to_string(&res).unwrap()
}
}),
)
.route(
"/daqingest/channel/add",
get({
let ingest_commons = ingest_commons.clone();
|Query(params): Query<HashMap<String, String>>| async move { channel_add(params, ingest_commons).await }
}),
)
.route(
"/daqingest/channel/remove",
get({
let ingest_commons = ingest_commons.clone();
|Query(params): Query<HashMap<String, String>>| async move { channel_remove(params, ingest_commons).await }
}),
)
.route(
@@ -79,11 +289,11 @@ pub async fn start_metrics_service(
"status": "success",
"data": {
"version": "2.37",
"revision": "aaaaaaaaaaaaaaaaaaaaaaaaa",
"revision": "daqingest",
"branch": "dev",
"buildUser": "empty",
"buildDate": "2022-07-14",
"goVersion": "go1"
"buildUser": "dominik.werder",
"buildDate": "2022-07-21",
"goVersion": "nogo"
}
});
serde_json::to_string(&res).unwrap()
@@ -92,11 +302,97 @@ pub async fn start_metrics_service(
.route(
"/api/v1/query",
post(
|Query(params): Query<HashMap<String, String>>, parts: Parts| async move {
info!("/api/v1/query params {params:?} {parts:?}");
|Form(form): Form<QueryForm>, Query(params): Query<HashMap<String, String>>, parts: Parts| async move {
info!("/api/v1/query form {form:?} params {params:?} {parts:?}");
let res = if form.query == "1+1" {
serde_json::json!({
"status": "success",
"data": {
"resultType": "scalar",
"result": [form.time.unwrap_or(0.0), "2"]
}
})
} else {
serde_json::json!({
"status": "success"
})
};
serde_json::to_string(&res).unwrap()
},
),
)
.route(
"/api/v1/labels",
post(|Form(_form): Form<PromLabels>| async move {
let res = {
serde_json::json!({
"status": "success",
"data": ["__name__", "instance"]
})
};
serde_json::to_string(&res).unwrap()
}),
)
.route(
"/api/v1/label/__name__/values",
get(|| async move {
let res = {
serde_json::json!({
"status": "success",
"data": ["series1", "series2"]
})
};
serde_json::to_string(&res).unwrap()
}),
)
.route(
"/api/v1/label/instance/values",
get(|| async move {
let res = {
serde_json::json!({
"status": "success",
"data": ["node1", "node2"]
})
};
serde_json::to_string(&res).unwrap()
}),
)
.route(
"/api/v1/metadata",
get(|| async move {
let res = {
serde_json::json!({
"status": "success",
"data": {}
})
};
serde_json::to_string(&res).unwrap()
}),
)
.route(
"/api/v1/series",
post(|parts: Parts, body: bytes::Bytes| async move {
info!("Asked for series, form: {parts:?}");
let url = url::Url::parse(&format!("http://dummy{}", parts.uri))
.unwrap_or_else(|_| url::Url::parse("http://a/").unwrap());
info!("PARSED SERIES URL {:?}", url);
let bodyparams = url::Url::parse(&String::from_utf8_lossy(&body));
info!("BODY PARAMS: {:?}", bodyparams);
let res = {
serde_json::json!({
"status": "success",
"data": [
{
"__name__": "series1",
"job": "daqingest",
"instance": "node1"
}
]
})
};
serde_json::to_string(&res).unwrap()
}),
)
.fallback(
get(|parts: Parts, body: extract::RawBody<hyper::Body>| async move {
let bytes = hyper::body::to_bytes(body.0).await.unwrap();
netfetch/src/series.rs
+2 -1
View File
@@ -2,6 +2,7 @@ use crate::bsread::ChannelDescDecoded;
use crate::errconv::ErrConv;
use err::Error;
use log::*;
use serde::Serialize;
use std::time::{Duration, Instant};
use tokio_postgres::Client as PgClient;
@@ -11,7 +12,7 @@ pub enum Existence<T> {
Existing(T),
}
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug, Serialize)]
pub struct SeriesId(u64);
impl SeriesId {
stats/src/stats.rs
+9
View File
@@ -24,6 +24,15 @@ impl EMA {
}
}
pub fn with_ema(ema: f32) -> Self {
Self {
ema,
emv: 0.0,
k: 0.05,
update_count: 0,
}
}
pub fn default() -> Self {
Self {
ema: 0.0,