From 99357437c9ef5163ac494fce6672cbaab4916d94 Mon Sep 17 00:00:00 2001 From: Dominik Werder Date: Tue, 3 Jan 2023 22:47:30 +0100 Subject: [PATCH] WIP better connection state handling --- daqingest/Cargo.toml | 12 +- daqingest/src/bin/daqingest.rs | 8 +- daqingest/src/daemon.rs | 447 +++++++++++++++++++++++++++++++++ daqingest/src/daqingest.rs | 96 +------ daqingest/src/opts.rs | 94 +++++++ daqingest/src/query.rs | 5 +- netfetch/Cargo.toml | 6 +- netfetch/src/ca/findioc.rs | 10 + netfetch/src/ca/search.rs | 23 -- netfetch/src/channelwriter.rs | 6 +- netfetch/src/conf.rs | 23 +- netfetch/src/metrics.rs | 61 +++-- netfetch/src/store.rs | 1 + 13 files changed, 613 insertions(+), 179 deletions(-) create mode 100644 daqingest/src/daemon.rs create mode 100644 daqingest/src/opts.rs diff --git a/daqingest/Cargo.toml b/daqingest/Cargo.toml index 6e63ac3..3c2e4bc 100644 --- a/daqingest/Cargo.toml +++ b/daqingest/Cargo.toml @@ -13,10 +13,16 @@ path = "src/bin/daqingest.rs" [dependencies] clap = { version = "4.0.22", features = ["derive", "cargo"] } +tokio = { version = "1.23.0", features = ["rt-multi-thread", "io-util", "net", "time", "sync", "fs", "tracing"] } +tracing = "0.1.37" +futures-util = "0.3" +async-channel = "1.6" chrono = "0.4" bytes = "1.1" -scylla = "0.4" +scylla = "0.7" +serde = { version = "1.0", features = ["derive"] } err = { path = "../../daqbuffer/err" } -taskrun = { path = "../../daqbuffer/taskrun" } -netfetch = { path = "../netfetch" } log = { path = "../log" } +netpod = { path = "../../daqbuffer/netpod" } +netfetch = { path = "../netfetch" } +taskrun = { path = "../../daqbuffer/taskrun" } diff --git a/daqingest/src/bin/daqingest.rs b/daqingest/src/bin/daqingest.rs index d03c48e..4948cc8 100644 --- a/daqingest/src/bin/daqingest.rs +++ b/daqingest/src/bin/daqingest.rs @@ -1,5 +1,5 @@ use clap::Parser; -use daqingest::{ChannelAccess, DaqIngestOpts, SubCmd}; +use daqingest::opts::DaqIngestOpts; use err::Error; use log::*; use netfetch::conf::parse_config; @@ -11,6 +11,8 @@ pub fn main() -> Result<(), Error> { taskrun::tracing_init().unwrap(); info!("daqingest version {}", clap::crate_version!()); let res = runtime.block_on(async move { + use daqingest::opts::ChannelAccess; + use daqingest::opts::SubCmd; match opts.subcmd { SubCmd::Bsread(k) => netfetch::zmtp::zmtp_client(k.into()).await?, SubCmd::ListPkey => daqingest::query::list_pkey().await?, @@ -29,6 +31,10 @@ pub fn main() -> Result<(), Error> { let (conf, channels) = parse_config(k.config.into()).await?; netfetch::ca::ca_connect(conf, &channels).await? } + ChannelAccess::CaIngestNew(k) => { + let (conf, channels) = parse_config(k.config.into()).await?; + daqingest::daemon::run(conf, channels).await? + } }, } Ok(()) diff --git a/daqingest/src/daemon.rs b/daqingest/src/daemon.rs new file mode 100644 index 0000000..72bfce5 --- /dev/null +++ b/daqingest/src/daemon.rs @@ -0,0 +1,447 @@ +use async_channel::Receiver; +use async_channel::Sender; +use err::Error; +use futures_util::FutureExt; +use futures_util::StreamExt; +use log::*; +use netfetch::ca::conn::CaConn; +use netfetch::ca::conn::ConnCommand; +use netfetch::ca::findioc::FindIocRes; +use netfetch::ca::findioc::FindIocStream; +use netfetch::conf::CaIngestOpts; +use serde::Serialize; +use std::collections::BTreeMap; +use std::collections::VecDeque; +use std::net::IpAddr; +use std::net::SocketAddrV4; +use std::time::Duration; +use std::time::SystemTime; + +const CHECK_CHANS_PER_TICK: usize = 10; + +#[derive(Clone, Debug, Serialize, PartialEq, PartialOrd, Eq, Ord)] +pub struct Channel { + id: String, +} + +impl Channel { + pub fn new(id: String) -> Self { + Self { id } + } + + pub fn id(&self) -> &str { + &self.id + } +} + +#[derive(Clone, Debug, Serialize)] +pub enum ConnectionStateValue { + Unconnected, + Connected { since: SystemTime }, +} + +#[derive(Clone, Debug, Serialize)] +pub struct ConnectionState { + updated: SystemTime, + value: ConnectionStateValue, +} + +#[derive(Clone, Debug, Serialize)] +pub enum WithAddressState { + Unassigned { assign_at: SystemTime }, + Assigned(ConnectionState), +} + +#[derive(Clone, Debug, Serialize)] +pub enum ActiveChannelState { + UnknownAddress, + SearchPending { + since: SystemTime, + }, + WithAddress { + addr: SocketAddrV4, + state: WithAddressState, + }, + NoAddress, +} + +#[derive(Clone, Debug, Serialize)] +pub enum ChannelState { + Active(ActiveChannelState), + ToRemove { addr: Option }, +} + +#[derive(Debug)] +pub enum DaemonEvent { + TimerTick, + ChannelAdd(Channel), + ChannelRemove(Channel), + SearchDone(Result, Error>), +} + +pub struct DaemonOpts { + search_tgts: Vec, + search_excl: Vec, +} + +struct OptFut { + fut: Option, +} + +impl OptFut { + fn new(fut: Option) -> Self { + Self { fut } + } +} + +impl futures_util::Future for OptFut +where + F: futures_util::Future + std::marker::Unpin, +{ + type Output = ::Output; + + fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context) -> std::task::Poll { + match self.fut.as_mut() { + Some(fut) => fut.poll_unpin(cx), + None => std::task::Poll::Pending, + } + } +} + +pub struct Daemon { + opts: DaemonOpts, + channel_states: BTreeMap, + tx: Sender, + rx: Receiver, + conns: BTreeMap, + chan_check_next: Option, + search_tx: Sender, + ioc_finder_jh: tokio::task::JoinHandle<()>, +} + +impl Daemon { + pub fn new(opts: DaemonOpts) -> Self { + let (tx, rx) = async_channel::bounded(1); + let tgts = opts.search_tgts.clone(); + let (search_tx, ioc_finder_jh) = { + let (qtx, qrx) = async_channel::bounded(1); + let (atx, arx) = async_channel::bounded(1); + let ioc_finder_fut = async move { + const FINDER_JOB_QUEUE_LEN_MAX: usize = 1; + let mut finder = FindIocStream::new(tgts); + let mut fut1 = finder.next(); + let mut fut2 = qrx.recv().fuse(); + let mut fut_tick = Box::pin(tokio::time::sleep(Duration::from_millis(2000)).fuse()); + let mut asend = OptFut::new(None).fuse(); + loop { + tokio::time::sleep(Duration::from_millis(200)).await; + futures_util::select! { + _ = asend => { + info!("asend done"); + } + r1 = fut1 => { + match r1 { + Some(item) => { + asend = OptFut::new(Some(atx.send(item))).fuse(); + } + None => { + // TODO finder has stopped, do no longer poll on it + } + } + if finder.job_queue_len() < FINDER_JOB_QUEUE_LEN_MAX { + fut2 = qrx.recv().fuse(); + } + fut1 = finder.next(); + fut_tick = Box::pin(tokio::time::sleep(Duration::from_millis(2000)).fuse()); + } + r2 = fut2 => { + match r2 { + Ok(item) => { + info!("Push to finder: {item:?}"); + finder.push(item); + } + Err(e) => { + // TODO input is done... ignore from here on. + error!("{e}"); + break; + } + } + if finder.job_queue_len() < FINDER_JOB_QUEUE_LEN_MAX { + fut2 = qrx.recv().fuse(); + } + fut1 = finder.next(); + fut_tick = Box::pin(tokio::time::sleep(Duration::from_millis(2000)).fuse()); + } + _ = fut_tick => { + if finder.job_queue_len() < FINDER_JOB_QUEUE_LEN_MAX { + //fut2 = qrx.recv().fuse(); + } + fut1 = finder.next(); + fut_tick = Box::pin(tokio::time::sleep(Duration::from_millis(2000)).fuse()); + } + }; + } + }; + let ioc_finder_jh = taskrun::spawn(ioc_finder_fut); + taskrun::spawn({ + let tx = tx.clone(); + async move { + while let Ok(item) = arx.recv().await { + info!("forward search result item"); + match tx.send(DaemonEvent::SearchDone(item)).await { + Ok(_) => {} + Err(e) => { + error!("search res fwd {e}"); + } + } + } + warn!("search res fwd nput broken"); + } + }); + (qtx, ioc_finder_jh) + }; + Self { + opts, + channel_states: BTreeMap::new(), + tx, + rx, + conns: BTreeMap::new(), + chan_check_next: None, + search_tx, + ioc_finder_jh, + } + } + + fn check_chans(&mut self) -> Result<(), Error> { + let tsnow = SystemTime::now(); + let k = self.chan_check_next.take(); + info!("check_chans start at {:?}", k); + let it = if let Some(last) = k { + self.channel_states.range_mut(last..) + } else { + self.channel_states.range_mut(..) + }; + for (i, (ch, st)) in it.enumerate() { + info!("check chan {} {:?}", i, ch); + use ActiveChannelState::*; + use ChannelState::*; + match st { + Active(st2) => match st2 { + UnknownAddress => { + if self.search_tx.is_full() { + // TODO what to do if the queue is full? + } else { + match self.search_tx.try_send(ch.id().into()) { + Ok(_) => { + *st = Active(SearchPending { since: tsnow }); + } + Err(_) => { + error!("can not send search query"); + } + } + } + } + SearchPending { since } => { + // TODO handle Err + match tsnow.duration_since(*since) { + Ok(dt) => { + if dt >= Duration::from_millis(10000) { + warn!("Search timeout for {ch:?}"); + *st = Active(ActiveChannelState::NoAddress); + } + } + Err(e) => { + error!("{e}"); + } + } + } + WithAddress { addr, state } => { + use WithAddressState::*; + match state { + Unassigned { assign_at } => { + if *assign_at <= tsnow { + match self.conns.get(addr) { + Some(conn) => { + let tx = conn.conn_command_tx(); + let (cmd, rx) = ConnCommand::channel_add(ch.id().into()); + // TODO how to send the command from this non-async context? + //tx.send(cmd).await; + // TODO if the send can be assumed to be on its way (it may still fail) then update state + if true { + let cs = ConnectionState { + updated: tsnow, + value: ConnectionStateValue::Unconnected, + }; + *state = WithAddressState::Assigned(cs) + } + } + None => {} + } + } + } + Assigned(_) => { + // TODO check if channel is healthy and alive + } + } + } + NoAddress => { + // TODO try to find address again after some randomized timeout + } + }, + ToRemove { .. } => { + // TODO if assigned to some address, + } + } + if i >= CHECK_CHANS_PER_TICK { + self.chan_check_next = Some(ch.clone()); + break; + } + } + Ok(()) + } + + fn handle_timer_tick(&mut self) -> Result<(), Error> { + self.check_chans()?; + Ok(()) + } + + fn handle_channel_add(&mut self, ch: Channel) -> Result<(), Error> { + if !self.channel_states.contains_key(&ch) { + self.channel_states + .insert(ch, ChannelState::Active(ActiveChannelState::UnknownAddress)); + } + Ok(()) + } + + fn handle_channel_remove(&mut self, ch: Channel) -> Result<(), Error> { + if let Some(k) = self.channel_states.get_mut(&ch) { + match k { + ChannelState::Active(j) => match j { + ActiveChannelState::UnknownAddress => { + *k = ChannelState::ToRemove { addr: None }; + } + ActiveChannelState::SearchPending { .. } => { + *k = ChannelState::ToRemove { addr: None }; + } + ActiveChannelState::WithAddress { addr, .. } => { + *k = ChannelState::ToRemove { + addr: Some(addr.clone()), + }; + } + ActiveChannelState::NoAddress => { + *k = ChannelState::ToRemove { addr: None }; + } + }, + ChannelState::ToRemove { .. } => {} + } + } + Ok(()) + } + + fn handle_event(&mut self, item: DaemonEvent) -> Result<(), Error> { + use DaemonEvent::*; + match item { + TimerTick => self.handle_timer_tick(), + ChannelAdd(ch) => self.handle_channel_add(ch), + ChannelRemove(ch) => self.handle_channel_remove(ch), + SearchDone(res) => { + info!("handle SearchDone: {res:?}"); + match res { + Ok(a) => { + for res in a { + if let Some(addr) = &res.addr { + let addr = addr.clone(); + let ch = Channel::new(res.channel); + if let Some(st) = self.channel_states.get_mut(&ch) { + if let ChannelState::Active(ActiveChannelState::SearchPending { .. }) = st { + let stnew = ChannelState::Active(ActiveChannelState::WithAddress { + addr, + state: WithAddressState::Unassigned { + assign_at: SystemTime::now(), + }, + }); + self.channel_states.insert(ch, stnew); + } else { + warn!("state for {ch:?} is not SearchPending"); + } + } else { + warn!("can not find channel state for {ch:?}"); + } + } else { + warn!("no addr from search in {res:?}"); + } + } + } + Err(e) => { + error!("error from search: {e}"); + } + } + Ok(()) + } + } + } + + pub async fn daemon(&mut self) -> Result<(), Error> { + let ticker = { + let tx = self.tx.clone(); + async move { + let mut ticker = tokio::time::interval(Duration::from_millis(500)); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + ticker.tick().await; + if let Err(e) = tx.send(DaemonEvent::TimerTick).await { + error!("can not send TimerTick {e}"); + break; + } + } + } + }; + taskrun::spawn(ticker); + loop { + match self.rx.recv().await { + Ok(item) => { + info!("got daemon event {item:?}"); + match self.handle_event(item) { + Ok(_) => {} + Err(e) => { + error!("daemon: {e}"); + break; + } + } + } + Err(e) => { + error!("{e}"); + break; + } + } + } + Ok(()) + } +} + +pub async fn run(opts: CaIngestOpts, channels: Vec) -> Result<(), Error> { + info!("start up {opts:?}"); + let mut search_tgts = Vec::new(); + for s in opts.search() { + let addr: SocketAddrV4 = s.parse()?; + search_tgts.push(addr); + } + info!("parsed search_tgts {search_tgts:?}"); + let opts2 = DaemonOpts { + search_tgts, + search_excl: Vec::new(), + }; + let mut daemon = Daemon::new(opts2); + let tx = daemon.tx.clone(); + let daemon_jh = taskrun::spawn(async move { + // TODO handle Err + daemon.daemon().await.unwrap(); + }); + for s in &channels { + let ch = Channel::new(s.into()); + tx.send(DaemonEvent::ChannelAdd(ch)).await?; + } + info!("all channels sent to daemon"); + daemon_jh.await.unwrap(); + Ok(()) +} diff --git a/daqingest/src/daqingest.rs b/daqingest/src/daqingest.rs index c45223f..d6adc23 100644 --- a/daqingest/src/daqingest.rs +++ b/daqingest/src/daqingest.rs @@ -1,95 +1,3 @@ +pub mod daemon; +pub mod opts; pub mod query; - -use clap::ArgAction::Count; -use clap::Parser; -use netfetch::zmtp::ZmtpClientOpts; - -#[derive(Debug, Parser)] -#[command(author, version, about)] -pub struct DaqIngestOpts { - #[arg(long, action(Count))] - pub verbose: u32, - #[clap(long)] - pub tag: Option, - #[command(subcommand)] - pub subcmd: SubCmd, - #[arg(long)] - pub nworkers: Option, -} - -#[derive(Debug, Parser)] -pub enum SubCmd { - Bsread(Bsread), - ListPkey, - ListPulses, - FetchEvents(FetchEvents), - BsreadDump(BsreadDump), - #[command(subcommand)] - ChannelAccess(ChannelAccess), -} - -#[derive(Debug, Parser)] -pub struct Bsread { - #[arg(long)] - pub backend: String, - #[arg(long)] - pub scylla: Vec, - #[arg(long)] - pub source: Vec, - #[arg(long)] - pub rcvbuf: Option, - #[arg(long)] - pub array_truncate: Option, - #[arg(long)] - pub do_pulse_id: bool, - #[arg(long)] - pub skip_insert: bool, - #[arg(long)] - pub process_channel_count_limit: Option, -} - -impl From for ZmtpClientOpts { - fn from(k: Bsread) -> Self { - Self { - backend: k.backend, - scylla: k.scylla, - sources: k.source, - rcvbuf: k.rcvbuf, - array_truncate: k.array_truncate, - do_pulse_id: k.do_pulse_id, - process_channel_count_limit: k.process_channel_count_limit, - skip_insert: k.skip_insert, - } - } -} - -#[derive(Debug, Parser)] -pub struct FetchEvents { - #[arg(long, num_args(1..))] - pub scylla: Vec, - #[arg(long)] - pub channel: String, - #[arg(long)] - pub backend: String, -} - -#[derive(Debug, Parser)] -pub struct BsreadDump { - pub source: String, -} - -#[derive(Debug, Parser)] -pub enum ChannelAccess { - CaIngest(CaConfig), - CaSearch(CaSearch), -} - -#[derive(Debug, Parser)] -pub struct CaSearch { - pub config: String, -} - -#[derive(Debug, Parser)] -pub struct CaConfig { - pub config: String, -} diff --git a/daqingest/src/opts.rs b/daqingest/src/opts.rs new file mode 100644 index 0000000..11667d0 --- /dev/null +++ b/daqingest/src/opts.rs @@ -0,0 +1,94 @@ +use clap::ArgAction::Count; +use clap::Parser; +use netfetch::zmtp::ZmtpClientOpts; + +#[derive(Debug, Parser)] +#[command(author, version, about)] +pub struct DaqIngestOpts { + #[arg(long, action(Count))] + pub verbose: u32, + #[clap(long)] + pub tag: Option, + #[command(subcommand)] + pub subcmd: SubCmd, + #[arg(long)] + pub nworkers: Option, +} + +#[derive(Debug, Parser)] +pub enum SubCmd { + Bsread(Bsread), + ListPkey, + ListPulses, + FetchEvents(FetchEvents), + BsreadDump(BsreadDump), + #[command(subcommand)] + ChannelAccess(ChannelAccess), +} + +#[derive(Debug, Parser)] +pub struct Bsread { + #[arg(long)] + pub backend: String, + #[arg(long)] + pub scylla: Vec, + #[arg(long)] + pub source: Vec, + #[arg(long)] + pub rcvbuf: Option, + #[arg(long)] + pub array_truncate: Option, + #[arg(long)] + pub do_pulse_id: bool, + #[arg(long)] + pub skip_insert: bool, + #[arg(long)] + pub process_channel_count_limit: Option, +} + +impl From for ZmtpClientOpts { + fn from(k: Bsread) -> Self { + Self { + backend: k.backend, + scylla: k.scylla, + sources: k.source, + rcvbuf: k.rcvbuf, + array_truncate: k.array_truncate, + do_pulse_id: k.do_pulse_id, + process_channel_count_limit: k.process_channel_count_limit, + skip_insert: k.skip_insert, + } + } +} + +#[derive(Debug, Parser)] +pub struct FetchEvents { + #[arg(long, num_args(1..))] + pub scylla: Vec, + #[arg(long)] + pub channel: String, + #[arg(long)] + pub backend: String, +} + +#[derive(Debug, Parser)] +pub struct BsreadDump { + pub source: String, +} + +#[derive(Debug, Parser)] +pub enum ChannelAccess { + CaIngest(CaConfig), + CaIngestNew(CaConfig), + CaSearch(CaSearch), +} + +#[derive(Debug, Parser)] +pub struct CaSearch { + pub config: String, +} + +#[derive(Debug, Parser)] +pub struct CaConfig { + pub config: String, +} diff --git a/daqingest/src/query.rs b/daqingest/src/query.rs index 8c82ac5..ca2dd4d 100644 --- a/daqingest/src/query.rs +++ b/daqingest/src/query.rs @@ -1,7 +1,8 @@ -use crate::FetchEvents; +use crate::opts::FetchEvents; use log::*; use scylla::batch::Consistency; -use scylla::transport::errors::{NewSessionError, QueryError}; +use scylla::transport::errors::NewSessionError; +use scylla::transport::errors::QueryError; use scylla::SessionBuilder; pub struct Error(err::Error); diff --git a/netfetch/Cargo.toml b/netfetch/Cargo.toml index e1ae405..6b02d15 100644 --- a/netfetch/Cargo.toml +++ b/netfetch/Cargo.toml @@ -20,13 +20,13 @@ arrayref = "0.3" byteorder = "1.4" futures-util = "0.3" #pin-project-lite = "0.2" -scylla = "0.4" +scylla = "0.7" tokio-postgres = "0.7.6" -md-5 = "0.9" +md-5 = "0.10" hex = "0.4" libc = "0.2" regex = "1.7.0" -axum = "0.5" +axum = "0.6" http = "0.2" url = "2.2" hyper = "0.14" diff --git a/netfetch/src/ca/findioc.rs b/netfetch/src/ca/findioc.rs index 618d673..e74326f 100644 --- a/netfetch/src/ca/findioc.rs +++ b/netfetch/src/ca/findioc.rs @@ -109,6 +109,10 @@ impl FindIocStream { ) } + pub fn job_queue_len(&self) -> usize { + self.channels_input.len() + } + pub fn push(&mut self, x: String) { self.channels_input.push_back(x); } @@ -580,3 +584,9 @@ impl Stream for FindIocStream { } } } + +impl futures_util::stream::FusedStream for FindIocStream { + fn is_terminated(&self) -> bool { + false + } +} diff --git a/netfetch/src/ca/search.rs b/netfetch/src/ca/search.rs index 6cf1a2d..9aa79a4 100644 --- a/netfetch/src/ca/search.rs +++ b/netfetch/src/ca/search.rs @@ -154,29 +154,6 @@ pub async fn ca_search(opts: CaIngestOpts, channels: &Vec) -> Result<(), if do_block { info!("blacklisting {item:?}"); } else { - /* - let srcaddr = item.src.to_string(); - let addr = item.addr.map(|x| x.to_string()).unwrap_or(String::new()); - let rows = pg_client - .query(&qu_select, &[&facility, &item.channel, &srcaddr]) - .await - .unwrap(); - if true || rows.is_empty() { - //info!("insert {item:?}"); - pg_client - .execute(&qu_insert, &[&facility, &item.channel, &srcaddr, &addr]) - .await - .unwrap(); - } else { - //info!("update {item:?}"); - let addr2: &str = rows[0].get(0); - if addr2 != addr {} - pg_client - .execute(&qu_update, &[&facility, &item.channel, &srcaddr, &addr]) - .await - .unwrap(); - } - */ let queryaddr = item.query_addr.map(|x| x.to_string()); let responseaddr = item.response_addr.map(|x| x.to_string()); let addr = item.addr.map(|x| x.to_string()); diff --git a/netfetch/src/channelwriter.rs b/netfetch/src/channelwriter.rs index 55f1336..897a460 100644 --- a/netfetch/src/channelwriter.rs +++ b/netfetch/src/channelwriter.rs @@ -9,7 +9,7 @@ use scylla::batch::{Batch, BatchType}; use scylla::frame::value::{BatchValues, ValueList}; use scylla::prepared_statement::PreparedStatement; use scylla::transport::errors::QueryError; -use scylla::{BatchResult, QueryResult, Session as ScySession}; +use scylla::{QueryResult, Session as ScySession}; use std::mem; use std::pin::Pin; use std::sync::Arc; @@ -46,7 +46,7 @@ impl<'a> Future for ScyQueryFut<'a> { } pub struct ScyBatchFut<'a> { - fut: Pin> + 'a>>, + fut: Pin> + 'a>>, polled: usize, ts_create: Instant, ts_poll_start: Instant, @@ -101,7 +101,7 @@ impl<'a> Future for ScyBatchFut<'a> { } pub struct ScyBatchFutGen<'a> { - fut: Pin> + Send + 'a>>, + fut: Pin> + Send + 'a>>, polled: usize, ts_create: Instant, ts_poll_start: Instant, diff --git a/netfetch/src/conf.rs b/netfetch/src/conf.rs index abbdfef..e49b42c 100644 --- a/netfetch/src/conf.rs +++ b/netfetch/src/conf.rs @@ -5,8 +5,6 @@ use netpod::Database; use netpod::ScyllaConfig; use serde::Deserialize; use serde::Serialize; -use std::net::IpAddr; -use std::net::Ipv4Addr; use std::path::PathBuf; use std::time::Duration; use tokio::fs::OpenOptions; @@ -20,10 +18,6 @@ pub struct CaIngestOpts { search: Vec, #[serde(default)] search_blacklist: Vec, - #[serde(default)] - tmp_remove: Vec, - addr_bind: Option, - addr_conn: Option, whitelist: Option, blacklist: Option, max_simul: Option, @@ -53,18 +47,6 @@ impl CaIngestOpts { &self.backend } - pub fn addr_bind(&self) -> IpAddr { - self.addr_bind - .clone() - .unwrap_or_else(|| IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0))) - } - - pub fn addr_conn(&self) -> IpAddr { - self.addr_conn - .clone() - .unwrap_or_else(|| IpAddr::V4(Ipv4Addr::new(255, 255, 255, 255))) - } - pub fn api_bind(&self) -> String { self.api_bind.clone().unwrap_or_else(|| "0.0.0.0:3011".into()) } @@ -210,7 +192,10 @@ pub async fn parse_config(config: PathBuf) -> Result<(CaIngestOpts, Vec) let mut channels = Vec::new(); for line in lines { let line = String::from_utf8_lossy(line); - let use_line = if let Some(_cs) = re_p.captures(&line) { + let line = line.trim(); + let use_line = if line.is_empty() { + false + } else if let Some(_cs) = re_p.captures(&line) { true } else if re_n.is_match(&line) { false diff --git a/netfetch/src/metrics.rs b/netfetch/src/metrics.rs index a67be97..bc1364c 100644 --- a/netfetch/src/metrics.rs +++ b/netfetch/src/metrics.rs @@ -3,12 +3,13 @@ use crate::ca::IngestCommons; use crate::ca::METRICS; use axum::extract::Query; use err::Error; -use http::request::Parts; +use http::Request; use log::*; use serde::{Deserialize, Serialize}; use stats::{CaConnStats, CaConnStatsAgg, CaConnStatsAggDiff}; use std::collections::HashMap; -use std::net::{SocketAddr, SocketAddrV4}; +use std::net::SocketAddr; +use std::net::SocketAddrV4; use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::Duration; @@ -129,9 +130,10 @@ async fn channel_state(params: HashMap, ingest_commons: Arc, + params: HashMap, ingest_commons: Arc, ) -> axum::Json> { + let limit = params.get("limit").map(|x| x.parse()).unwrap_or(Ok(40)).unwrap_or(40); let vals = ingest_commons .ca_conn_set .send_command_to_all(|| ConnCommand::channel_states_all()) @@ -144,11 +146,7 @@ async fn channel_states( } } res.sort_unstable_by_key(|v| u32::MAX - v.interest_score as u32); - let res = if true { - res.into_iter().rev().take(10).collect() - } else { - res - }; + res.truncate(limit); axum::Json(res) } @@ -164,10 +162,33 @@ async fn extra_inserts_conf_set(v: ExtraInsertsConf, ingest_commons: Arc, + age: usize, +} + pub async fn start_metrics_service(bind_to: String, ingest_commons: Arc) { + use axum::extract; + use axum::http::StatusCode; use axum::routing::{get, put}; - use axum::{extract, Router}; + use axum::Router; let app = Router::new() + .fallback(|req: Request| async move { + info!("Fallback for {} {}", req.method(), req.uri()); + StatusCode::NOT_FOUND + }) + .nest( + "/some", + Router::new() + .route("/path1", get(|| async { (StatusCode::OK, format!("Hello there!")) })) + .route( + "/path2", + get(|qu: Query| async move { (StatusCode::OK, format!("{qu:?}")) }), + ), + ) .route( "/metrics", get(|| async { @@ -267,28 +288,6 @@ pub async fn start_metrics_service(bind_to: String, ingest_commons: Arc| async move { - let bytes = hyper::body::to_bytes(body.0).await.unwrap(); - let s = String::from_utf8_lossy(&bytes); - info!("GET {parts:?} body: {s:?}"); - }) - .post(|parts: Parts, body: extract::RawBody| async move { - let bytes = hyper::body::to_bytes(body.0).await.unwrap(); - let s = String::from_utf8_lossy(&bytes); - info!("POST {parts:?} body: {s:?}"); - }) - .put(|parts: Parts, body: extract::RawBody| async move { - let bytes = hyper::body::to_bytes(body.0).await.unwrap(); - let s = String::from_utf8_lossy(&bytes); - info!("PUT {parts:?} body: {s:?}"); - }) - .delete(|parts: Parts, body: extract::RawBody| async move { - let bytes = hyper::body::to_bytes(body.0).await.unwrap(); - let s = String::from_utf8_lossy(&bytes); - info!("DELETE {parts:?} body: {s:?}"); - }), ); axum::Server::bind(&bind_to.parse().unwrap()) .serve(app.into_make_service()) diff --git a/netfetch/src/store.rs b/netfetch/src/store.rs index 637027d..d51eec8 100644 --- a/netfetch/src/store.rs +++ b/netfetch/src/store.rs @@ -55,6 +55,7 @@ impl IntoSimplerError for QueryError { QueryError::TimeoutError => Error::DbTimeout, QueryError::TooManyOrphanedStreamIds(e) => Error::DbError(e.to_string()), QueryError::UnableToAllocStreamId => Error::DbError(e.to_string()), + QueryError::RequestTimeout(e) => Error::DbError(e.to_string()), } } }