daqbuf/daqbuffer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Read channel status from disk

Browse Source
This commit is contained in:
Dominik Werder
2022-12-16 06:27:35 +01:00
parent 3016da236b
commit 6c5ada63e7
15 changed files with 364 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
httpret/src/channel_status.rs
View File

@@ -13,7 +13,7 @@ pub struct ConnectionStatusEvents {}
impl ConnectionStatusEvents {
pub fn handler(req: &Request<Body>) -> Option<Self> {
if req.uri().path() == "/api/4/scylla/connection/status/events" {
if req.uri().path() == "/api/4/status/connection/events" {
Some(Self {})
} else {
None
@@ -63,7 +63,11 @@ impl ConnectionStatusEvents {
.as_ref()
.ok_or_else(|| Error::with_public_msg_no_trace(format!("No Scylla configured")))?;
let scy = scyllaconn::create_scy_session(scyco).await?;
let mut stream = scyllaconn::events::channel_state_events(q, scy).await?;
let chconf = dbconn::channelconfig::chconf_from_database(q.channel(), node_config).await?;
let series = chconf.series;
let do_one_before_range = true;
let mut stream =
scyllaconn::status::StatusStreamScylla::new(series, q.range().clone(), do_one_before_range, scy);
let mut ret = Vec::new();
while let Some(item) = stream.next().await {
let item = item?;
@@ -77,7 +81,7 @@ pub struct ChannelStatusEvents {}
impl ChannelStatusEvents {
pub fn handler(req: &Request<Body>) -> Option<Self> {
if req.uri().path() == "/api/4/scylla/channel/status/events" {
if req.uri().path() == "/api/4/status/channel/events" {
Some(Self {})
} else {
None
@@ -127,7 +131,11 @@ impl ChannelStatusEvents {
.as_ref()
.ok_or_else(|| Error::with_public_msg_no_trace(format!("No Scylla configured")))?;
let scy = scyllaconn::create_scy_session(scyco).await?;
let mut stream = scyllaconn::events::channel_state_events(q, scy).await?;
let chconf = dbconn::channelconfig::chconf_from_database(q.channel(), node_config).await?;
let series = chconf.series;
let do_one_before_range = true;
let mut stream =
scyllaconn::status::StatusStreamScylla::new(series, q.range().clone(), do_one_before_range, scy);
let mut ret = Vec::new();
while let Some(item) = stream.next().await {
let item = item?;

2
httpret/src/channelconfig.rs
View File

@@ -439,7 +439,7 @@ pub struct ScyllaChannelsActive {}
impl ScyllaChannelsActive {
pub fn handler(req: &Request<Body>) -> Option<Self> {
if req.uri().path() == "/api/4/scylla/channels/active" {
if req.uri().path() == "/api/4/channels/active" {
Some(Self {})
} else {
None

6
httpret/src/proxy.rs
View File

@@ -11,7 +11,7 @@ use hyper::service::{make_service_fn, service_fn};
use hyper::{Body, Request, Response, Server};
use itertools::Itertools;
use netpod::log::*;
use netpod::query::{BinnedQuery, PlainEventsQuery};
use netpod::query::{BinnedQuery, ChannelStateEventsQuery, PlainEventsQuery};
use netpod::{AppendToUrl, ChannelConfigQuery, FromUrl, HasBackend, HasTimeout, ProxyConfig};
use netpod::{ChannelSearchQuery, ChannelSearchResult, ChannelSearchSingleResult};
use netpod::{ACCEPT_ALL, APP_JSON};
@@ -121,6 +121,10 @@ async fn proxy_http_service_inner(
Ok(api4::channel_search(req, proxy_config).await?)
} else if path == "/api/4/events" {
Ok(proxy_single_backend_query::<PlainEventsQuery>(req, ctx, proxy_config).await?)
} else if path == "/api/4/status/connection/events" {
Ok(proxy_single_backend_query::<ChannelStateEventsQuery>(req, ctx, proxy_config).await?)
} else if path == "/api/4/status/channel/events" {
Ok(proxy_single_backend_query::<ChannelStateEventsQuery>(req, ctx, proxy_config).await?)
} else if path.starts_with("/api/4/map/pulse/") {
Ok(proxy_single_backend_query::<MapPulseQuery>(req, ctx, proxy_config).await?)
} else if path == "/api/4/binned" {

36
items_2/src/binsdim0.rs
View File

@@ -42,17 +42,31 @@ where
{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
let self_name = std::any::type_name::<Self>();
write!(
fmt,
"{self_name} count {} ts1s {:?} ts2s {:?} counts {:?} mins {:?} maxs {:?} avgs {:?}",
self.ts1s.len(),
self.ts1s.iter().map(|k| k / SEC).collect::<Vec<_>>(),
self.ts2s.iter().map(|k| k / SEC).collect::<Vec<_>>(),
self.counts,
self.mins,
self.maxs,
self.avgs,
)
if true {
write!(
fmt,
"{self_name} count {} ts1s {:?} ts2s {:?} counts {:?} mins {:?} maxs {:?} avgs {:?}",
self.ts1s.len(),
self.ts1s.iter().map(|k| k / SEC).collect::<Vec<_>>(),
self.ts2s.iter().map(|k| k / SEC).collect::<Vec<_>>(),
self.counts,
self.mins,
self.maxs,
self.avgs,
)
} else {
write!(
fmt,
"{self_name} count {} edges {:?} .. {:?} counts {:?} .. {:?} avgs {:?} .. {:?}",
self.ts1s.len(),
self.ts1s.front().map(|k| k / SEC),
self.ts2s.back().map(|k| k / SEC),
self.counts.front(),
self.counts.back(),
self.avgs.front(),
self.avgs.back(),
)
}
}
}

39
items_2/src/channelevents.rs
View File

@@ -19,6 +19,17 @@ pub enum ConnStatus {
Disconnect,
}
impl ConnStatus {
pub fn from_ca_ingest_status_kind(k: u32) -> Self {
match k {
1 => Self::Connect,
2 => Self::Disconnect,
3 => Self::Disconnect,
_ => Self::Disconnect,
}
}
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct ConnStatusEvent {
pub ts: u64,
@@ -121,10 +132,38 @@ mod serde_channel_events {
let e1: u32 = seq.next_element()?.ok_or(de::Error::missing_field("[1] nty"))?;
if e0 == EventsDim0::<u8>::serde_id() {
match e1 {
u8::SUB => {
let obj: EventsDim0<u8> = seq.next_element()?.ok_or(de::Error::missing_field("[2] obj"))?;
Ok(EvBox(Box::new(obj)))
}
u16::SUB => {
let obj: EventsDim0<u16> = seq.next_element()?.ok_or(de::Error::missing_field("[2] obj"))?;
Ok(EvBox(Box::new(obj)))
}
u32::SUB => {
let obj: EventsDim0<u32> = seq.next_element()?.ok_or(de::Error::missing_field("[2] obj"))?;
Ok(EvBox(Box::new(obj)))
}
u64::SUB => {
let obj: EventsDim0<u64> = seq.next_element()?.ok_or(de::Error::missing_field("[2] obj"))?;
Ok(EvBox(Box::new(obj)))
}
i8::SUB => {
let obj: EventsDim0<i8> = seq.next_element()?.ok_or(de::Error::missing_field("[2] obj"))?;
Ok(EvBox(Box::new(obj)))
}
i16::SUB => {
let obj: EventsDim0<i16> = seq.next_element()?.ok_or(de::Error::missing_field("[2] obj"))?;
Ok(EvBox(Box::new(obj)))
}
i32::SUB => {
let obj: EventsDim0<i32> = seq.next_element()?.ok_or(de::Error::missing_field("[2] obj"))?;
Ok(EvBox(Box::new(obj)))
}
i64::SUB => {
let obj: EventsDim0<i64> = seq.next_element()?.ok_or(de::Error::missing_field("[2] obj"))?;
Ok(EvBox(Box::new(obj)))
}
f32::SUB => {
let obj: EventsDim0<f32> = seq.next_element()?.ok_or(de::Error::missing_field("[2] obj"))?;
Ok(EvBox(Box::new(obj)))

2
items_2/src/eventsdim0.rs
View File

@@ -77,7 +77,7 @@ where
NTY: fmt::Debug,
{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
if true {
if false {
write!(
fmt,
"EventsDim0 {{ count {} ts {:?} vals {:?} }}",

24
netpod/src/netpod.rs
View File

@@ -28,6 +28,10 @@ pub const APP_JSON_LINES: &'static str = "application/jsonlines";
pub const APP_OCTET: &'static str = "application/octet-stream";
pub const ACCEPT_ALL: &'static str = "*/*";
pub const CONNECTION_STATUS_DIV: u64 = timeunits::DAY;
pub const TS_MSP_GRID_UNIT: u64 = timeunits::SEC * 10;
pub const TS_MSP_GRID_SPACING: u64 = 6 * 2;
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct AggQuerySingleChannel {
pub channel_config: ChannelConfig,
@@ -685,12 +689,20 @@ pub struct NanoRange {
impl fmt::Debug for NanoRange {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
let beg = chrono::Utc.timestamp_opt((self.beg / SEC) as i64, (self.beg % SEC) as u32);
let end = chrono::Utc.timestamp_opt((self.end / SEC) as i64, (self.end % SEC) as u32);
f.debug_struct("NanoRange")
.field("beg", &beg)
.field("end", &end)
.finish()
let beg = chrono::Utc
.timestamp_opt((self.beg / SEC) as i64, (self.beg % SEC) as u32)
.earliest();
let end = chrono::Utc
.timestamp_opt((self.end / SEC) as i64, (self.end % SEC) as u32)
.earliest();
if let (Some(a), Some(b)) = (beg, end) {
f.debug_struct("NanoRange").field("beg", &a).field("end", &b).finish()
} else {
f.debug_struct("NanoRange")
.field("beg", &beg)
.field("end", &end)
.finish()
}
}
}

14
nodenet/src/conn.rs
View File

@@ -174,20 +174,16 @@ async fn events_conn_handler_inner_try(
let scalar_type = f.scalar_type;
let shape = f.shape;
let do_test_stream_error = false;
let stream = match scyllaconn::events::make_scylla_stream(
&evq,
do_one_before_range,
debug!("Make EventsStreamScylla for {series} {scalar_type:?} {shape:?}");
let stream = scyllaconn::events::EventsStreamScylla::new(
series,
evq.range().clone(),
do_one_before_range,
scalar_type,
shape,
scy,
do_test_stream_error,
)
.await
{
Ok(k) => k,
Err(e) => return Err((e, netout))?,
};
);
let stream = stream.map(|item| {
let item = match item {
Ok(item) => match item {

1
scyllaconn/Cargo.toml
View File

@@ -13,6 +13,7 @@ serde_json = "1.0"
serde_cbor = "0.11.2"
erased-serde = "0.3"
tokio = { version = "1.23.0", default-features = false, features = ["time", "sync"] }
tracing = "0.1.37"
byteorder = "1.4.3"
bytes = "1.2.1"
num-traits = "0.2.15"

35
scyllaconn/src/events.rs
View File

@@ -5,7 +5,7 @@ use items_0::{Empty, Events, WithLen};
use items_2::channelevents::{ChannelEvents, ConnStatus, ConnStatusEvent};
use items_2::eventsdim0::EventsDim0;
use netpod::log::*;
use netpod::query::{ChannelStateEventsQuery, PlainEventsQuery};
use netpod::query::ChannelStateEventsQuery;
use netpod::timeunits::*;
use netpod::{NanoRange, ScalarType, Shape};
use scylla::Session as ScySession;
@@ -14,13 +14,13 @@ use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
async fn find_ts_msp(series: i64, range: NanoRange, scy: Arc<ScySession>) -> Result<VecDeque<u64>, Error> {
async fn find_ts_msp(series: u64, range: NanoRange, scy: Arc<ScySession>) -> Result<VecDeque<u64>, Error> {
info!("find_ts_msp series {} {:?}", series, range);
let mut ret = VecDeque::new();
// TODO use prepared statements
let cql = "select ts_msp from ts_msp where series = ? and ts_msp > ? and ts_msp < ?";
let res = scy
.query(cql, (series, range.beg as i64, range.end as i64))
.query(cql, (series as i64, range.beg as i64, range.end as i64))
.await
.err_conv()?;
for row in res.rows_typed_or_empty::<(i64,)>() {
@@ -28,13 +28,13 @@ async fn find_ts_msp(series: i64, range: NanoRange, scy: Arc<ScySession>) -> Res
ret.push_back(row.0 as u64);
}
let cql = "select ts_msp from ts_msp where series = ? and ts_msp >= ? limit 1";
let res = scy.query(cql, (series, range.end as i64)).await.err_conv()?;
let res = scy.query(cql, (series as i64, range.end as i64)).await.err_conv()?;
for row in res.rows_typed_or_empty::<(i64,)>() {
let row = row.err_conv()?;
ret.push_back(row.0 as u64);
}
let cql = "select ts_msp from ts_msp where series = ? and ts_msp <= ? order by ts_msp desc limit 2";
let res = scy.query(cql, (series, range.beg as i64)).await.err_conv()?;
let res = scy.query(cql, (series as i64, range.beg as i64)).await.err_conv()?;
for row in res.rows_typed_or_empty::<(i64,)>() {
let row = row.err_conv()?;
ret.push_front(row.0 as u64);
@@ -491,7 +491,7 @@ impl Stream for EventsStreamScylla {
loop {
break match self.state {
FrState::New => {
let fut = find_ts_msp(self.series as i64, self.range.clone(), self.scy.clone());
let fut = find_ts_msp(self.series, self.range.clone(), self.scy.clone());
let fut = Box::pin(fut);
self.state = FrState::FindMsp(fut);
continue;
@@ -558,28 +558,7 @@ impl Stream for EventsStreamScylla {
}
}
pub async fn make_scylla_stream(
evq: &PlainEventsQuery,
do_one_before_range: bool,
series: u64,
scalar_type: ScalarType,
shape: Shape,
scy: Arc<ScySession>,
do_test_stream_error: bool,
) -> Result<EventsStreamScylla, Error> {
let res = EventsStreamScylla::new(
series,
evq.range().clone(),
do_one_before_range,
scalar_type,
shape,
scy,
do_test_stream_error,
);
Ok(res)
}
pub async fn channel_state_events(
async fn _channel_state_events(
evq: &ChannelStateEventsQuery,
scy: Arc<ScySession>,
) -> Result<Pin<Box<dyn Stream<Item = Result<ConnStatusEvent, Error>> + Send>>, Error> {

1
scyllaconn/src/scyllaconn.rs
View File

@@ -2,6 +2,7 @@ pub mod bincache;
pub mod config;
pub mod errconv;
pub mod events;
pub mod status;
use err::Error;
use errconv::ErrConv;

237
scyllaconn/src/status.rs Normal file
View File

@@ -0,0 +1,237 @@
use crate::errconv::ErrConv;
use err::Error;
use futures_util::{Future, FutureExt, Stream};
use items_2::channelevents::{ConnStatus, ConnStatusEvent};
use netpod::log::*;
use netpod::NanoRange;
use netpod::CONNECTION_STATUS_DIV;
use scylla::Session as ScySession;
use std::collections::VecDeque;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
async fn read_next_status_events(
series: u64,
ts_msp: u64,
range: NanoRange,
fwd: bool,
do_one_before: bool,
scy: Arc<ScySession>,
) -> Result<VecDeque<ConnStatusEvent>, Error> {
if ts_msp >= range.end {
warn!(
"given ts_msp {} >= range.end {} not necessary to read this",
ts_msp, range.end
);
}
if range.end > i64::MAX as u64 {
return Err(Error::with_msg_no_trace(format!("range.end overflows i64")));
}
let res = if fwd {
let ts_lsp_min = if ts_msp < range.beg { range.beg - ts_msp } else { 0 };
let ts_lsp_max = if ts_msp < range.end { range.end - ts_msp } else { 0 };
trace!(
"FWD ts_msp {} ts_lsp_min {} ts_lsp_max {} beg {} end {}",
ts_msp,
ts_lsp_min,
ts_lsp_max,
range.beg,
range.end
);
// TODO use prepared!
let cql = concat!(
"select ts_lsp, pulse, kind from channel_status where series = ? and ts_msp = ? and ts_lsp >= ? and ts_lsp < ?"
);
scy.query(
cql,
(series as i64, ts_msp as i64, ts_lsp_min as i64, ts_lsp_max as i64),
)
.await
.err_conv()?
} else {
let ts_lsp_max = if ts_msp < range.beg { range.beg - ts_msp } else { 0 };
debug!(
"BCK ts_msp {} ts_lsp_max {} beg {} end {} {}",
ts_msp,
ts_lsp_max,
range.beg,
range.end,
stringify!($fname)
);
// TODO use prepared!
let cql = concat!(
"select ts_lsp, pulse, kind from channel_status where series = ? and ts_msp = ? and ts_lsp < ? order by ts_lsp desc limit 1"
);
scy.query(cql, (series as i64, ts_msp as i64, ts_lsp_max as i64))
.await
.err_conv()?
};
let mut last_before = None;
let mut ret = VecDeque::new();
for row in res.rows_typed_or_empty::<(i64, i64, i32)>() {
let row = row.err_conv()?;
let ts = ts_msp + row.0 as u64;
let _pulse = row.1 as u64;
let kind = row.2 as u32;
// from netfetch::store::ChannelStatus
let ev = ConnStatusEvent {
ts,
status: ConnStatus::from_ca_ingest_status_kind(kind),
};
if ts >= range.end {
} else if ts >= range.beg {
ret.push_back(ev);
} else {
last_before = Some(ev);
}
}
if do_one_before {
if let Some(ev) = last_before {
debug!("PREPENDING THE LAST BEFORE {ev:?}");
ret.push_front(ev);
}
}
trace!("found in total {} events ts_msp {}", ret.len(), ts_msp);
Ok(ret)
}
struct ReadValues {
series: u64,
range: NanoRange,
ts_msps: VecDeque<u64>,
fwd: bool,
do_one_before_range: bool,
fut: Pin<Box<dyn Future<Output = Result<VecDeque<ConnStatusEvent>, Error>> + Send>>,
scy: Arc<ScySession>,
}
impl ReadValues {
fn new(
series: u64,
range: NanoRange,
ts_msps: VecDeque<u64>,
fwd: bool,
do_one_before_range: bool,
scy: Arc<ScySession>,
) -> Self {
let mut ret = Self {
series,
range,
ts_msps,
fwd,
do_one_before_range,
fut: Box::pin(futures_util::future::ready(Err(Error::with_msg_no_trace(
"future not initialized",
)))),
scy,
};
ret.next();
ret
}
fn next(&mut self) -> bool {
if let Some(ts_msp) = self.ts_msps.pop_front() {
self.fut = self.make_fut(ts_msp, self.ts_msps.len() > 0);
true
} else {
false
}
}
fn make_fut(
&mut self,
ts_msp: u64,
_has_more_msp: bool,
) -> Pin<Box<dyn Future<Output = Result<VecDeque<ConnStatusEvent>, Error>> + Send>> {
let fut = read_next_status_events(
self.series,
ts_msp,
self.range.clone(),
self.fwd,
self.do_one_before_range,
self.scy.clone(),
);
Box::pin(fut)
}
}
enum FrState {
New,
ReadValues(ReadValues),
Done,
}
pub struct StatusStreamScylla {
state: FrState,
series: u64,
range: NanoRange,
do_one_before_range: bool,
scy: Arc<ScySession>,
ts_msps: VecDeque<u64>,
outbuf: VecDeque<ConnStatusEvent>,
}
impl StatusStreamScylla {
pub fn new(series: u64, range: NanoRange, do_one_before_range: bool, scy: Arc<ScySession>) -> Self {
Self {
state: FrState::New,
series,
range,
do_one_before_range,
scy,
ts_msps: VecDeque::new(),
outbuf: VecDeque::new(),
}
}
}
impl Stream for StatusStreamScylla {
type Item = Result<ConnStatusEvent, Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
let span = tracing::span!(tracing::Level::TRACE, "poll_next");
let _spg = span.enter();
loop {
if let Some(x) = self.outbuf.pop_front() {
break Ready(Some(Ok(x)));
}
break match self.state {
FrState::New => {
let mut ts_msps = VecDeque::new();
let mut ts = self.range.beg / CONNECTION_STATUS_DIV * CONNECTION_STATUS_DIV;
while ts < self.range.end {
ts_msps.push_back(ts);
ts += CONNECTION_STATUS_DIV;
}
let st = ReadValues::new(
self.series,
self.range.clone(),
self.ts_msps.clone(),
true,
self.do_one_before_range,
self.scy.clone(),
);
self.state = FrState::ReadValues(st);
continue;
}
FrState::ReadValues(ref mut st) => match st.fut.poll_unpin(cx) {
Ready(Ok(item)) => {
if !st.next() {
debug!("ReadValues exhausted");
self.state = FrState::Done;
}
for x in item {
self.outbuf.push_back(x);
}
continue;
}
Ready(Err(e)) => Ready(Some(Err(e))),
Pending => Pending,
},
FrState::Done => Ready(None),
};
}
}
}

2
streams/src/frames/inmem.rs
View File

@@ -194,7 +194,7 @@ where
}
}
Ready(Err(e)) => {
error!("poll_upstream {e:?}");
error!("poll_upstream need_min {} buf {:?} {:?}", self.need_min, self.buf, e);
self.done = true;
Ready(Some(Err(e)))
}

10
streams/src/slidebuf.rs
View File

@@ -399,3 +399,13 @@ impl SlideBuf {
}
}
}
impl fmt::Debug for SlideBuf {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
fmt.debug_struct("SlideBuf")
.field("cap", &self.cap())
.field("wp", &self.wp)
.field("rp", &self.rp)
.finish()
}
}

2
streams/src/tcprawclient.rs
View File

@@ -71,7 +71,7 @@ where
netout.flush().await?;
netout.forget();
// TODO for images, we need larger buffer capacity
let frames = InMemoryFrameAsyncReadStream::new(netin, 1024 * 128);
let frames = InMemoryFrameAsyncReadStream::new(netin, 1024 * 256);
let stream = EventsFromFrames::<_, T>::new(frames);
streams.push(Box::pin(stream) as _);
}