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

WIP better connection state handling

Browse Source
This commit is contained in:
Dominik Werder
2023-01-03 22:47:30 +01:00
parent 4a71678aba
commit 99357437c9
13 changed files with 613 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daqingest/Cargo.toml
+9 -3
View File
@@ -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" }
daqingest/src/bin/daqingest.rs
+7 -1
View File
@@ -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(())
daqingest/src/daemon.rs
+447
View File
@@ -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<SocketAddrV4> },
}
#[derive(Debug)]
pub enum DaemonEvent {
TimerTick,
ChannelAdd(Channel),
ChannelRemove(Channel),
SearchDone(Result<VecDeque<FindIocRes>, Error>),
}
pub struct DaemonOpts {
search_tgts: Vec<SocketAddrV4>,
search_excl: Vec<SocketAddrV4>,
}
struct OptFut<F> {
fut: Option<F>,
}
impl<F> OptFut<F> {
fn new(fut: Option<F>) -> Self {
Self { fut }
}
}
impl<F> futures_util::Future for OptFut<F>
where
F: futures_util::Future + std::marker::Unpin,
{
type Output = <F as futures_util::Future>::Output;
fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context) -> std::task::Poll<Self::Output> {
match self.fut.as_mut() {
Some(fut) => fut.poll_unpin(cx),
None => std::task::Poll::Pending,
}
}
}
pub struct Daemon {
opts: DaemonOpts,
channel_states: BTreeMap<Channel, ChannelState>,
tx: Sender<DaemonEvent>,
rx: Receiver<DaemonEvent>,
conns: BTreeMap<SocketAddrV4, CaConn>,
chan_check_next: Option<Channel>,
search_tx: Sender<String>,
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<String>) -> 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(())
}
daqingest/src/daqingest.rs
+2 -94
View File
@@ -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<String>,
#[command(subcommand)]
pub subcmd: SubCmd,
#[arg(long)]
pub nworkers: Option<usize>,
}
#[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<String>,
#[arg(long)]
pub source: Vec<String>,
#[arg(long)]
pub rcvbuf: Option<usize>,
#[arg(long)]
pub array_truncate: Option<usize>,
#[arg(long)]
pub do_pulse_id: bool,
#[arg(long)]
pub skip_insert: bool,
#[arg(long)]
pub process_channel_count_limit: Option<usize>,
}
impl From<Bsread> 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<String>,
#[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,
}
daqingest/src/opts.rs
+94
View File
@@ -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<String>,
#[command(subcommand)]
pub subcmd: SubCmd,
#[arg(long)]
pub nworkers: Option<usize>,
}
#[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<String>,
#[arg(long)]
pub source: Vec<String>,
#[arg(long)]
pub rcvbuf: Option<usize>,
#[arg(long)]
pub array_truncate: Option<usize>,
#[arg(long)]
pub do_pulse_id: bool,
#[arg(long)]
pub skip_insert: bool,
#[arg(long)]
pub process_channel_count_limit: Option<usize>,
}
impl From<Bsread> 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<String>,
#[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,
}
daqingest/src/query.rs
+3 -2
View File
@@ -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);
netfetch/Cargo.toml
+3 -3
View File
@@ -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"
netfetch/src/ca/findioc.rs
+10
View File
@@ -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
}
}
netfetch/src/ca/search.rs
-23
View File
@@ -154,29 +154,6 @@ pub async fn ca_search(opts: CaIngestOpts, channels: &Vec<String>) -> 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());
netfetch/src/channelwriter.rs
+3 -3
View File
@@ -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<Box<dyn Future<Output = Result<BatchResult, QueryError>> + 'a>>,
fut: Pin<Box<dyn Future<Output = Result<QueryResult, QueryError>> + '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<Box<dyn Future<Output = Result<BatchResult, QueryError>> + Send + 'a>>,
fut: Pin<Box<dyn Future<Output = Result<QueryResult, QueryError>> + Send + 'a>>,
polled: usize,
ts_create: Instant,
ts_poll_start: Instant,
netfetch/src/conf.rs
+4 -19
View File
@@ -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<String>,
#[serde(default)]
search_blacklist: Vec<String>,
#[serde(default)]
tmp_remove: Vec<String>,
addr_bind: Option<IpAddr>,
addr_conn: Option<IpAddr>,
whitelist: Option<String>,
blacklist: Option<String>,
max_simul: Option<usize>,
@@ -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<String>)
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
netfetch/src/metrics.rs
+30 -31
View File
@@ -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<String, String>, ingest_commons: Arc<Inge
}
async fn channel_states(
_params: HashMap<String, String>,
params: HashMap<String, String>,
ingest_commons: Arc<IngestCommons>,
) -> axum::Json<Vec<crate::ca::conn::ChannelStateInfo>> {
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<IngestC
axum::Json(true)
}
#[allow(unused)]
#[derive(Debug, Deserialize)]
struct DummyQuery {
name: String,
surname: Option<String>,
age: usize,
}
pub async fn start_metrics_service(bind_to: String, ingest_commons: Arc<IngestCommons>) {
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<axum::body::Body>| 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<DummyQuery>| 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<IngestCo
insert_ivl_min.store(v.0, Ordering::Release);
}
}),
)
.fallback(
get(|parts: Parts, body: extract::RawBody<hyper::Body>| 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<hyper::Body>| 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<hyper::Body>| 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<hyper::Body>| 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())
netfetch/src/store.rs
+1
View File
@@ -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()),
}
}
}