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

Log fix

Browse Source
This commit is contained in:
Dominik Werder
2025-05-28 17:48:45 +02:00
parent c17032db23
commit 16aec2732b
13 changed files with 115 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netfetch/src/ca/beacons.rs
+2 -2
View File
@@ -72,7 +72,7 @@ pub async fn listen_beacons(
}
}?;
if n != 16 {
debug!("len recv {n}");
debug!("len recv {}", n);
}
if n >= 16 {
let mut cur = Cursor::new(bb);
@@ -84,7 +84,7 @@ pub async fn listen_beacons(
let addr_u32 = cur.get_u32();
let addr = Ipv4Addr::from(addr_u32);
if cmd == 0x0d {
debug!("beacon {remote} {ver} {addr} {port}");
debug!("beacon {} {} {} {}", remote, ver, addr, port);
let stnow = SystemTime::now();
let x = stnow.duration_since(SystemTime::UNIX_EPOCH).unwrap();
let ts = TsNano::from_ms(1000 * x.as_secs() + x.subsec_millis() as u64);
netfetch/src/ca/conn.rs
+36 -30
View File
@@ -1589,7 +1589,7 @@ impl CaConn {
}
pub fn channel_add(&mut self, conf: ChannelConfig, cssid: ChannelStatusSeriesId) -> Result<(), Error> {
debug!("channel_add {conf:?} {cssid:?}");
debug!("channel_add {:?} {:?}", conf, cssid);
if false {
if series::dbg::dbg_chn(&conf.name()) {
self.trace_channel_poll = true;
@@ -1598,7 +1598,7 @@ impl CaConn {
if self.cid_by_name(conf.name()).is_some() {
self.mett.channel_add_exists().inc();
if series::dbg::dbg_chn(&conf.name()) {
error!("logic error channel already exists {conf:?}");
error!("logic error channel already exists {:?}", conf);
}
Ok(())
} else {
@@ -1606,7 +1606,7 @@ impl CaConn {
if self.channels.contains_key(&cid) {
self.mett.channel_add_exists().inc();
if series::dbg::dbg_chn(&conf.name()) {
error!("logic error channel already exists {conf:?}");
error!("logic error channel already exists {:?}", conf);
}
Ok(())
} else {
@@ -1688,7 +1688,7 @@ impl CaConn {
if let Some(cid) = self.cid_by_name(&name) {
self.channel_remove_by_cid(cid);
} else {
warn!("channel_remove does not exist {name}");
warn!("channel_remove does not exist {}", name);
}
}
@@ -1803,7 +1803,7 @@ impl CaConn {
// as logic error.
// Close connection to the IOC. Cout as logic error.
let e = Error::UnknownCid(cid);
error!("{e}");
error!("transition_to_polling {}", e);
return Err(e);
};
if let ChannelState::Writable(st2) = ch_s {
@@ -1848,7 +1848,10 @@ impl CaConn {
(&mut x.state, &mut x.wrst, &x.conf)
} else {
// TODO return better as error and let caller decide (with more structured errors)
warn!("TODO handle_event_add_res can not find channel for {cid:?} {subid:?}");
warn!(
"TODO handle_event_add_res can not find channel for {:?} {:?}",
cid, subid
);
// TODO
// When removing a channel, keep it in "closed" btree for some time because messages can
// still arrive from all buffers.
@@ -1912,7 +1915,7 @@ impl CaConn {
match &mut st.reading {
ReadingState::EnableMonitoring(st2) => {
let dt = st2.tsbeg.elapsed().as_secs_f32();
trace!("change to Monitoring after dt {dt:.0} ms");
trace!("change to Monitoring after dt {:.0} ms", dt);
st.reading = ReadingState::Monitoring(MonitoringState {
tsbeg: tsnow,
subid: st2.subid,
@@ -2024,7 +2027,10 @@ impl CaConn {
&mut x.state
} else {
// TODO return better as error and let caller decide (with more structured errors)
warn!("TODO handle_event_add_res can not find channel for {cid:?} {subid:?}");
warn!(
"TODO handle_event_add_res can not find channel for {:?} {:?}",
cid, subid
);
// TODO
// When removing a channel, keep it in "closed" btree for some time because messages can
// still arrive from all buffers.
@@ -2050,7 +2056,7 @@ impl CaConn {
}
ReadingState::EnableMonitoring(..) => {
let name = self.name_by_cid(cid);
warn!("received event-cancel but channel {name:?} in wrong state");
warn!("received event-cancel but channel {:?} in wrong state", name);
}
ReadingState::Monitoring(st2) => {
match &mut st2.mon2state {
@@ -2061,16 +2067,16 @@ impl CaConn {
Monitoring2State::ReadPending(_) => {}
}
let name = self.name_by_cid(cid);
warn!("received event-cancel but channel {name:?} in wrong state");
warn!("received event-cancel but channel {:?} in wrong state", name);
}
ReadingState::Polling(..) => {
let name = self.name_by_cid(cid);
warn!("received event-cancel but channel {name:?} in wrong state");
warn!("received event-cancel but channel {:?} in wrong state", name);
}
},
_ => {
// TODO count instead of print
error!("unexpected state: EventAddRes while having {ch_s:?}");
error!("unexpected state: EventAddRes while having {:?}", ch_s);
}
}
Ok(())
@@ -2099,7 +2105,7 @@ impl CaConn {
let (ch_s, ch_wrst, ch_conf) = if let Some(x) = self.channels.get_mut(cid) {
(&mut x.state, &mut x.wrst, &x.conf)
} else {
warn!("handle_read_notify_res can not find channel for {cid:?} {ioid:?}");
warn!("handle_read_notify_res can not find channel for {:?} {:?}", cid, ioid);
return Ok(());
};
match ch_s {
@@ -2254,7 +2260,7 @@ impl CaConn {
}
_ => {
// TODO count instead of print
error!("unexpected state: ReadNotifyRes while having {ch_s:?}");
error!("unexpected state: ReadNotifyRes while having {:?}", ch_s);
}
}
} else {
@@ -2462,32 +2468,32 @@ impl CaConn {
CaMsgTy::VersionRes(n) => {
// debug!("see incoming {:?} {:?}", self.remote_addr_dbg, msg);
if n < 12 || n > 13 {
error!("see some unexpected version {n} channel search may not work.");
error!("see some unexpected version {} channel search may not work", n);
Ready(Some(Ok(())))
} else {
if n != 13 {
warn!("received peer version {n}");
warn!("received peer version {}", n);
}
self.state = CaConnState::PeerReady;
Ready(Some(Ok(())))
}
}
CaMsgTy::CreateChanRes(k) => {
warn!("got unexpected {k:?}",);
warn!("got unexpected {:?}", k);
Ready(Some(Ok(())))
}
CaMsgTy::AccessRightsRes(k) => {
warn!("got unexpected {k:?}",);
warn!("got unexpected {:?}", k);
Ready(Some(Ok(())))
}
k => {
warn!("got some other unhandled message: {k:?}");
warn!("got some other unhandled message: {:?}", k);
Ready(Some(Ok(())))
}
},
},
Err(e) => {
error!("got error item from CaProto {e:?}");
error!("got error item from CaProto {:?}", e);
Ready(Some(Err(e.into())))
}
},
@@ -2933,7 +2939,7 @@ impl CaConn {
Ready(Some(Ok(())))
}
Ready(Some(Err(e))) => {
error!("CaProto yields error: {e:?} remote {:?}", self.remote_addr_dbg);
error!("CaProto yields error: {} remote {:?}", e, self.remote_addr_dbg);
self.trigger_shutdown(ShutdownReason::Protocol);
Ready(Some(Err(e)))
}
@@ -3108,7 +3114,7 @@ impl CaConn {
}
Ok(Err(e)) => {
use std::io::ErrorKind;
debug!("error connect to {addr} {e}");
debug!("error connect to {} {}", addr, e);
let addr = addr.clone();
self.emit_connection_status_item(ConnectionStatusItem {
ts: self.tmp_ts_poll,
@@ -3124,7 +3130,7 @@ impl CaConn {
}
Err(e) => {
// TODO log with exponential backoff
debug!("timeout connect to {addr} {e}");
debug!("timeout connect to {} {}", addr, e);
let addr = addr.clone();
self.emit_connection_status_item(ConnectionStatusItem {
ts: self.tmp_ts_poll,
@@ -3204,7 +3210,7 @@ impl CaConn {
continue;
}
Ready(None) => {
error!("handle_conn_state yields {x:?}");
error!("handle_conn_state yields {:?}", x);
return Err(Error::LoopInnerLogicError);
}
Pending => {
@@ -3231,7 +3237,7 @@ impl CaConn {
Ready(()) => match self.as_mut().handle_own_ticker(cx) {
Ok(_) => Ok(Pending),
Err(e) => {
error!("handle_own_ticker {e}");
error!("handle_own_ticker {}", e);
self.trigger_shutdown(ShutdownReason::InternalError);
Err(e)
}
@@ -3267,7 +3273,7 @@ impl CaConn {
match self.tick_writers() {
Ok(()) => {}
Err(e) => {
error!("error in writers: {e}");
error!("error in writers: {}", e);
}
}
}
@@ -3486,7 +3492,7 @@ impl CaConn {
let self_name = "attempt_flush_queue";
use Poll::*;
if qu.len() != 0 {
trace_flush_queue!("{self_name} id {:10} len {}", id, qu.len());
trace_flush_queue!("{} id {:10} len {}", self_name, id, qu.len());
}
let mut have_progress = false;
let mut i = 0;
@@ -3509,7 +3515,7 @@ impl CaConn {
if sp.is_sending() {
match sp.poll_unpin(cx) {
Ready(Ok(())) => {
trace_flush_queue!("{self_name} id {:10} send done", id);
trace_flush_queue!("{} id {:10} send done", self_name, id);
have_progress = true;
}
Ready(Err(e)) => {
@@ -3517,7 +3523,7 @@ impl CaConn {
match e {
SpErr::NoSendInProgress => return Err(Error::NotSending),
SpErr::Closed(_) => {
error!("{self_name} queue closed id {:10}", id);
error!("{} queue closed id {:10}", self_name, id);
return Err(Error::ClosedSending);
}
}
@@ -3778,7 +3784,7 @@ impl Stream for CaConn {
have_pending = true;
}
Err(e) => {
error!("{e}");
error!("{}", e);
self.state = CaConnState::EndOfStream;
break Ready(Some(CaConnEvent::err_now(e)));
}
netfetch/src/ca/connset.rs
+11 -11
View File
@@ -655,7 +655,7 @@ impl CaConnSet {
}
}
for cmd in cmds {
debug!("call handle_remove_channel {cmd:?}");
debug!("call handle_remove_channel {:?}", cmd);
self.handle_remove_channel(cmd)?;
}
if let Err(_) = cmd.restx.try_send(Ok(())) {
@@ -727,7 +727,7 @@ impl CaConnSet {
}
}
Err(e) => {
warn!("TODO handle error {e}");
warn!("TODO handle error {}", e);
Ok(())
}
}
@@ -887,7 +887,7 @@ impl CaConnSet {
k.value = ChannelStateValue::ToRemove { addr: None };
}
WithStatusSeriesIdStateInner::WithAddress { addr, state: _ } => {
debug!("send remove {ch:?} to {addr}");
debug!("send remove {:?} to {}", ch, addr);
let conn_ress = self
.ca_conn_ress
.get_mut(&SocketAddr::V4(addr.clone()))
@@ -1140,7 +1140,7 @@ impl CaConnSet {
}
fn handle_ca_conn_eos(&mut self, addr: SocketAddr, reason: EndOfStreamReason) -> Result<(), Error> {
debug!("handle_ca_conn_eos {addr} {reason:?}");
debug!("handle_ca_conn_eos {} {:?}", addr, reason);
if let Some(e) = self.ca_conn_ress.remove(&addr) {
self.mett.ca_conn_eos_ok().inc();
self.await_ca_conn_jhs.push_back((addr, e.jh));
@@ -1411,7 +1411,7 @@ impl CaConnSet {
ActiveChannelState::WaitForStatusSeriesId { since } => {
let dt = stnow.duration_since(*since).unwrap_or(Duration::ZERO);
if dt > Duration::from_millis(20000) {
warn!("timeout can not get status series id for {ch:?}");
warn!("timeout can not get status series id for {:?}", ch);
*st2 = ActiveChannelState::Init { since: stnow };
} else {
// TODO
@@ -1675,7 +1675,7 @@ impl CaConnSet {
Ready(Err(e)) => match e {
scywr::senderpolling::Error::NoSendInProgress => {
let e = Error::PushCmdsNoSendInProgress(*addr);
error!("{e}");
error!("try_push_ca_conn_cmds {}", e);
return Some(Ready(Err(e)));
}
scywr::senderpolling::Error::Closed(_) => {
@@ -1816,7 +1816,7 @@ where
Some(Ready(Ok(())))
}
Ready(Err(e)) => {
error!("sender_polling_send {e}");
error!("sender_polling_send {}", e);
Some(Ready(Err(e.into())))
}
Pending => Some(Pending),
@@ -1871,7 +1871,7 @@ impl Stream for CaConnSet {
penpro.mark_progress();
}
Err(e) => {
error!("ticker {e}");
error!("ticker {}", e);
break Ready(Some(CaConnSetItem::Error(e)));
}
},
@@ -1893,15 +1893,15 @@ impl Stream for CaConnSet {
match x {
Ok(Ok(())) => {
self.mett.ca_conn_task_join_done_ok().inc();
debug!("CaConn {addr} finished well left {left}");
debug!("CaConn {} finished well left {}", addr, left);
}
Ok(Err(e)) => {
self.mett.ca_conn_task_join_done_err().inc();
error!("CaConn {addr} task error: {e} left {left}");
error!("CaConn {} task error: {} left {}", addr, e, left);
}
Err(e) => {
self.mett.ca_conn_task_join_err().inc();
error!("CaConn {addr} join error: {e} left {left}");
error!("CaConn {} join error: {} left {}", addr, e, left);
}
}
penpro.mark_progress();
netfetch/src/ca/finder.rs
+6 -6
View File
@@ -240,7 +240,7 @@ async fn finder_network_if_not_found(
let mut res = VecDeque::new();
let mut net = VecDeque::new();
for e in item {
trace!("{self_name} sees {e:?}");
trace!("{} sees {:?}", self_name, e);
if e.addr.is_none() {
net.push_back(e.channel);
} else {
@@ -248,12 +248,12 @@ async fn finder_network_if_not_found(
}
}
if let Err(_) = tx.send(res).await {
debug!("{self_name} res send error, break");
debug!("{} res send error, break", self_name);
break;
}
for ch in net {
if let Err(_) = net_tx.send(ch).await {
debug!("{self_name} net ch send error, break");
debug!("{} net ch send error, break", self_name);
break 'outer;
}
}
@@ -300,7 +300,7 @@ async fn process_net_result(
}
}
Err(e) => {
warn!("error during network search: {e}");
warn!("error during network search: {}", e);
break;
}
}
@@ -310,7 +310,7 @@ async fn process_net_result(
trace!("process_net_result dbtx closed");
for (i, jh) in ioc_search_index_worker_jhs.into_iter().enumerate() {
jh.await?;
trace!("process_net_result search index worker {i} awaited");
trace!("process_net_result search index worker {} awaited", i);
}
Ok(())
}
@@ -361,7 +361,7 @@ fn start_finder_ca(tx: Sender<DaemonEvent>, tgts: Vec<SocketAddrV4>) -> (Sender<
}
Err(e) => {
// TODO input is done... ignore from here on.
error!("Finder input channel error {e}");
error!("Finder input channel error {}",e);
qrx_more = false;
}
}
netfetch/src/ca/findioc.rs
+12 -12
View File
@@ -233,7 +233,7 @@ impl FindIocStream {
let mut addr_len = std::mem::size_of::<libc::sockaddr_in>();
let ec = unsafe { libc::getsockname(sock.0, &mut addr as *mut _ as _, &mut addr_len as *mut _ as _) };
if ec == -1 {
error!("getsockname {ec}");
error!("getsockname {}", ec);
return Err(Error::SocketConvertTokio);
} else {
if true {
@@ -303,7 +303,7 @@ impl FindIocStream {
}
} else if ec < 0 {
// stats.ca_udp_io_error().inc();
error!("unexpected received {ec}");
error!("unexpected received {}", ec);
Poll::Ready(Err(Error::ReadFailure))
} else if ec == 0 {
// stats.ca_udp_io_empty().inc();
@@ -318,7 +318,7 @@ impl FindIocStream {
for i in 0..(ec as usize) {
s1.extend(format!(" {:02x}", buf[i]).chars());
}
debug!("received answer {s1}");
debug!("received answer {}", s1);
debug!(
"received answer string {}",
String::from_utf8_lossy(buf[..ec as usize].into())
@@ -367,12 +367,12 @@ impl FindIocStream {
}
if msgs.len() == 1 {
// stats.ca_udp_warn().inc();
debug!("received answer with single message: {msgs:?}");
debug!("received answer with single message: {:?}", msgs);
}
let mut good = true;
if let CaMsgTy::VersionRes(v) = msgs[0].ty {
if v != 13 {
warn!("bad version in search response: {v}");
warn!("bad version in search response: {}", v);
good = false;
}
} else {
@@ -484,7 +484,7 @@ impl FindIocStream {
}
}
if !found_sid {
error!("can not find sid {sid:?} in batch {bid:?}");
error!("can not find sid {:?} in batch {:?}", sid, bid);
}
let all_done = batch.done.iter().all(|x| *x);
if all_done {
@@ -494,7 +494,7 @@ impl FindIocStream {
}
None => {
// TODO analyze reasons
error!("no batch for {bid:?}");
error!("no batch for {:?}", bid);
}
}
}
@@ -503,7 +503,7 @@ impl FindIocStream {
if self.sids_done.contains_key(&sid) {
self.result_for_done_sid_count += 1;
} else {
error!("no bid for {sid:?}");
error!("no bid for {:?}", sid);
}
}
}
@@ -608,7 +608,7 @@ impl Stream for FindIocStream {
have_progress = true;
}
Ready(Err(e)) => {
error!("{e}");
error!("FindIocStream {}", e);
}
Pending => {
g.clear_ready();
@@ -617,7 +617,7 @@ impl Stream for FindIocStream {
}
},
Ready(Err(e)) => {
error!("poll_write_ready {e}");
error!("poll_write_ready {}", e);
// TODO should we abort?
}
Pending => {}
@@ -693,7 +693,7 @@ impl Stream for FindIocStream {
continue;
}
Ready(Err(e)) => {
error!("Error from try_read {e:?}");
error!("try_read {}", e);
Ready(Some(Err(e)))
}
Pending => {
@@ -706,7 +706,7 @@ impl Stream for FindIocStream {
}
}
Ready(Err(e)) => {
error!("poll_read_ready {e:?}");
error!("poll_read_ready {}", e);
Ready(Some(Err(e.into())))
}
Pending => {
netfetch/src/ca/search.rs
+2 -2
View File
@@ -90,7 +90,7 @@ async fn search_tgts_from_opts(opts: &CaIngestOpts) -> Result<(Vec<SocketAddrV4>
}
}
Err(e) => {
error!("can not resolve {s} {e}");
error!("can not resolve {} {}", s, e);
}
}
}
@@ -110,7 +110,7 @@ async fn search_tgts_from_opts(opts: &CaIngestOpts) -> Result<(Vec<SocketAddrV4>
}
}
Err(e) => {
warn!("can not resolve {s} {e}");
warn!("can not resolve {} {}", s, e);
}
}
}