diff --git a/netfetch/Cargo.toml b/netfetch/Cargo.toml index 66e3199..ffacb0d 100644 --- a/netfetch/Cargo.toml +++ b/netfetch/Cargo.toml @@ -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" } diff --git a/netfetch/src/ca.rs b/netfetch/src/ca.rs index 5e4f6ec..3dd0090 100644 --- a/netfetch/src/ca.rs +++ b/netfetch/src/ca.rs @@ -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>> = None; @@ -40,6 +42,7 @@ pub fn get_metrics() -> &'static mut Option { #[derive(Debug, Serialize, Deserialize)] struct ChannelConfig { + backend: String, channels: Vec, search: Vec, addr_bind: Ipv4Addr, @@ -87,6 +90,7 @@ pub async fn parse_config(config: PathBuf) -> Result { }) .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 { } pub struct CaConnectOpts { + pub backend: String, pub channels: Vec, pub search: Vec, 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, insert_frac: Arc, pg_client: Arc, store_stats: Arc, @@ -232,42 +237,135 @@ async fn spawn_scylla_insert_workers( } pub struct CommandQueueSet { - queues: tokio::sync::Mutex>>, + queues: tokio::sync::Mutex>>, } impl CommandQueueSet { pub fn new() -> Self { Self { - queues: tokio::sync::Mutex::new(VecDeque::>::new()), + queues: tokio::sync::Mutex::new(BTreeMap::>::new()), } } - pub fn queues(&self) -> &tokio::sync::Mutex>> { + pub fn queues(&self) -> &tokio::sync::Mutex>> { &self.queues } } +pub struct IngestCommons { + pub pgconf: Arc, + pub local_epics_hostname: String, + pub insert_item_queue: Arc, + pub data_store: Arc, + pub insert_ivl_min: Arc, + pub conn_stats: Arc>>>, + pub command_queue_set: Arc, +} + +pub async fn find_channel_addr( + backend: String, + name: String, + pgconf: &Database, +) -> Result, 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::() { + 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, + data_store: Arc, + insert_ivl_min: Arc, + conn_stats: Arc>>>, + command_queue_set: Arc, +) -> Result>, 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>>> = 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(()) } diff --git a/netfetch/src/ca/conn.rs b/netfetch/src/ca/conn.rs index ac0a0b7..00ccc1a 100644 --- a/netfetch/src/ca/conn.rs +++ b/netfetch/src/ca/conn.rs @@ -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, + pub shape: Option, + // 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, + #[serde(skip_serializing_if = "Option::is_none", serialize_with = "ser_instant")] + pub ts_event_last: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub item_recv_ivl_ema: Option, + pub interest_score: f32, +} + +fn ser_instant(val: &Option, ser: S) -> Result { + 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> + Send>>), @@ -112,6 +200,10 @@ impl IdStore { #[derive(Debug)] pub enum ConnCommandKind { FindChannel(String, Sender<(SocketAddrV4, Vec)>), + ChannelState(String, Sender<(SocketAddrV4, Option)>), + ChannelStatesAll((), Sender<(SocketAddrV4, Vec)>), + ChannelAdd(String, Sender), + ChannelRemove(String, Sender), } #[derive(Debug)] @@ -127,6 +219,46 @@ impl ConnCommand { }; (cmd, rx) } + + pub fn channel_state( + name: String, + ) -> ( + ConnCommand, + async_channel::Receiver<(SocketAddrV4, Option)>, + ) { + 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)>, + ) { + 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) { + 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) { + 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, init_state_count: u64, cid_by_name: BTreeMap, @@ -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)); diff --git a/netfetch/src/ca/proto.rs b/netfetch/src/ca/proto.rs index 7f32501..5edacd4 100644 --- a/netfetch/src/ca/proto.rs +++ b/netfetch/src/ca/proto.rs @@ -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) => { diff --git a/netfetch/src/metrics.rs b/netfetch/src/metrics.rs index 68b41c4..e01e5d2 100644 --- a/netfetch/src/metrics.rs +++ b/netfetch/src/metrics.rs @@ -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, + #[allow(unused)] + timeout: Option, +} + +#[allow(unused)] +#[derive(Debug, Deserialize)] +struct PromLabels { + start: Option, + end: Option, + //#[serde(rename = "match[]")] + //pattern: Option>, +} + +#[allow(unused)] +#[derive(Debug, Deserialize)] +struct PromLabelValues { + start: Option, + end: Option, + //#[serde(rename = "match[]")] + //pattern: Option>, +} + async fn get_empty() -> String { format!("") } +async fn channel_add(params: HashMap, ingest_commons: Arc) -> 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, + ingest_commons: Arc, +) -> axum::Json { + use axum::Json; + use serde_json::Value; + let addr = if let Some(x) = params.get("addr") { + if let Ok(addr) = x.parse::() { + 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, insert_ivl_min: Arc, command_queue_set: Arc, + ingest_commons: Arc, ) { 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>| 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>| 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>| 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>| 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>| async move { channel_add(params, ingest_commons).await } + }), + ) + .route( + "/daqingest/channel/remove", + get({ + let ingest_commons = ingest_commons.clone(); + |Query(params): Query>| 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>, parts: Parts| async move { - info!("/api/v1/query params {params:?} {parts:?}"); + |Form(form): Form, Query(params): Query>, 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| 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| async move { let bytes = hyper::body::to_bytes(body.0).await.unwrap(); diff --git a/netfetch/src/series.rs b/netfetch/src/series.rs index 978fce6..793be1d 100644 --- a/netfetch/src/series.rs +++ b/netfetch/src/series.rs @@ -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 { Existing(T), } -#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)] +#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug, Serialize)] pub struct SeriesId(u64); impl SeriesId { diff --git a/stats/src/stats.rs b/stats/src/stats.rs index a56fbe0..dba0fef 100644 --- a/stats/src/stats.rs +++ b/stats/src/stats.rs @@ -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,