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

Remove unused archiver file format support

Browse Source
This commit is contained in:
Dominik Werder
2022-07-26 12:06:00 +02:00
parent f195a941f1
commit 41652fe3f9
37 changed files with 16 additions and 16545 deletions
Download Patch File Download Diff File Expand all files Collapse all files

427
httpret/src/channelarchiver.rs
View File

@@ -1,427 +0,0 @@
use crate::err::Error;
use crate::response;
use futures_core::Stream;
use futures_util::{StreamExt, TryStreamExt};
use http::{header, Method, Request, Response, StatusCode};
use hyper::Body;
use netpod::log::*;
use netpod::{get_url_query_pairs, Channel, NanoRange};
use netpod::{NodeConfigCached, APP_JSON_LINES};
use serde::Serialize;
use serde_json::Value as JsVal;
use url::Url;
fn json_lines_stream<S, I>(stream: S) -> impl Stream<Item = Result<Vec<u8>, Error>>
where
S: Stream<Item = Result<I, Error>>,
I: Serialize,
{
stream.map(|k| {
k.map(|k| {
let mut a = serde_json::to_vec(&k).unwrap();
a.push(0xa);
a
})
})
}
pub struct ListIndexFilesHttpFunction {}
impl ListIndexFilesHttpFunction {
pub fn prefix() -> &'static str {
"/api/4/channelarchiver/list/indexfiles"
}
pub fn name() -> &'static str {
"ListIndexFilesHttpFunction"
}
pub fn handler(path: &str) -> Option<Self> {
if path.starts_with(Self::prefix()) {
Some(Self {})
} else {
None
}
}
pub async fn handle(&self, req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
if req.method() != Method::GET {
return Ok(response(StatusCode::NOT_ACCEPTABLE).body(Body::empty())?);
}
info!("{} handle uri: {:?}", Self::name(), req.uri());
let conf = node_config
.node
.channel_archiver
.as_ref()
.ok_or(Error::with_msg_no_trace(
"this node is not configured as channel archiver",
))?;
let s = archapp_wrap::archapp::archeng::indexfiles::list_index_files(conf);
let s = futures_util::stream::unfold(s, |mut st| async move {
let x = st.next().await;
match x {
Some(x) => match x {
Ok(x) => {
let mut x = serde_json::to_vec(&x).unwrap();
x.push(b'\n');
Some((Ok::<_, Error>(x), st))
}
Err(e) => {
error!("{:?}", e);
None
}
},
None => None,
}
});
Ok(response(StatusCode::OK)
.header(header::CONTENT_TYPE, APP_JSON_LINES)
.body(Body::wrap_stream(s))?)
}
}
pub struct ScanIndexFiles {}
impl ScanIndexFiles {
pub fn prefix() -> &'static str {
"/api/4/channelarchiver/scan/indexfiles"
}
pub fn name() -> &'static str {
"ScanIndexFiles"
}
pub fn handler(path: &str) -> Option<Self> {
if path.starts_with(Self::prefix()) {
Some(Self {})
} else {
None
}
}
pub async fn handle(&self, req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
if req.method() != Method::GET {
return Ok(response(StatusCode::NOT_ACCEPTABLE).body(Body::empty())?);
}
info!("{} handle uri: {:?}", Self::name(), req.uri());
let conf = node_config
.node
.channel_archiver
.as_ref()
.ok_or(Error::with_msg_no_trace(
"this node is not configured as channel archiver",
))?;
let s = archapp_wrap::archapp::archeng::indexfiles::scan_index_files(conf.clone(), node_config.clone());
let s = s.map_err(Error::from);
let s = json_lines_stream(s);
Ok(response(StatusCode::OK)
.header(header::CONTENT_TYPE, APP_JSON_LINES)
.body(Body::wrap_stream(s))?)
}
}
pub struct ScanChannels {}
impl ScanChannels {
pub fn prefix() -> &'static str {
"/api/4/channelarchiver/scan/channels"
}
pub fn name() -> &'static str {
"ScanChannels"
}
pub fn handler(path: &str) -> Option<Self> {
if path.starts_with(Self::prefix()) {
Some(Self {})
} else {
None
}
}
pub async fn handle(&self, req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
if req.method() != Method::GET {
return Ok(response(StatusCode::NOT_ACCEPTABLE).body(Body::empty())?);
}
info!("{} handle uri: {:?}", Self::name(), req.uri());
let conf = node_config
.node
.channel_archiver
.as_ref()
.ok_or(Error::with_msg_no_trace(
"this node is not configured as channel archiver",
))?;
let s = archapp_wrap::archapp::archeng::indexfiles::scan_channels(node_config.clone(), conf.clone());
let s = s.map_err(Error::from);
let s = json_lines_stream(s);
Ok(response(StatusCode::OK)
.header(header::CONTENT_TYPE, APP_JSON_LINES)
.body(Body::wrap_stream(s))?)
}
}
pub struct ChannelNames {}
impl ChannelNames {
pub fn prefix() -> &'static str {
"/api/4/channelarchiver/channel/names"
}
pub fn name() -> &'static str {
"ChannelNames"
}
pub fn handler(path: &str) -> Option<Self> {
if path.starts_with(Self::prefix()) {
Some(Self {})
} else {
None
}
}
pub async fn handle(&self, req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
if req.method() != Method::GET {
return Ok(response(StatusCode::NOT_ACCEPTABLE).body(Body::empty())?);
}
info!("{} handle uri: {:?}", Self::name(), req.uri());
let database = &node_config.node_config.cluster.database;
use archapp_wrap::archapp::archeng;
let stream = archeng::configs::ChannelNameStream::new(database.clone());
let stream = stream.map_err(Error::from);
let stream = json_lines_stream(stream);
Ok(response(StatusCode::OK)
.header(header::CONTENT_TYPE, APP_JSON_LINES)
.body(Body::wrap_stream(stream))?)
}
}
pub struct ScanConfigs {}
impl ScanConfigs {
pub fn prefix() -> &'static str {
"/api/4/channelarchiver/scan/configs"
}
pub fn name() -> &'static str {
"ScanConfigs"
}
pub fn handler(path: &str) -> Option<Self> {
if path.starts_with(Self::prefix()) {
Some(Self {})
} else {
None
}
}
pub async fn handle(&self, req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
if req.method() != Method::GET {
return Ok(response(StatusCode::NOT_ACCEPTABLE).body(Body::empty())?);
}
info!("{} handle uri: {:?}", Self::name(), req.uri());
let database = &node_config.node_config.cluster.database;
let conf = node_config
.node
.channel_archiver
.as_ref()
.ok_or(Error::with_msg_no_trace(
"this node is not configured as channel archiver",
))?;
use archapp_wrap::archapp::archeng;
let stream = archeng::configs::ChannelNameStream::new(database.clone());
let stream = archeng::configs::ConfigStream::new(stream, node_config.clone(), conf.clone());
let stream = stream.map_err(Error::from);
let stream = json_lines_stream(stream);
Ok(response(StatusCode::OK)
.header(header::CONTENT_TYPE, APP_JSON_LINES)
.body(Body::wrap_stream(stream))?)
}
}
pub struct BlockRefStream {}
impl BlockRefStream {
pub fn prefix() -> &'static str {
"/api/4/channelarchiver/blockrefstream"
}
pub fn name() -> &'static str {
"BlockRefStream"
}
pub fn handler(path: &str) -> Option<Self> {
if path.starts_with(Self::prefix()) {
Some(Self {})
} else {
None
}
}
pub async fn handle(&self, req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
if req.method() != Method::GET {
return Ok(response(StatusCode::NOT_ACCEPTABLE).body(Body::empty())?);
}
info!("{} handle uri: {:?}", Self::name(), req.uri());
let database = &node_config.node_config.cluster.database;
let range = NanoRange { beg: 0, end: u64::MAX };
let url = Url::parse(&format!("dummy:{}", req.uri()))?;
let pairs = get_url_query_pairs(&url);
let channel_name = pairs.get("channelName").map(String::from).unwrap_or("NONE".into());
let channel = Channel {
backend: "".into(),
name: channel_name,
series: None,
//name: "ARIDI-PCT:CURRENT".into(),
};
use archapp_wrap::archapp::archeng;
let ixpaths = archeng::indexfiles::index_file_path_list(channel.clone(), database.clone()).await?;
info!("got categorized ixpaths: {:?}", ixpaths);
let ixpath = ixpaths.first().unwrap().clone();
let s = archeng::blockrefstream::blockref_stream(channel, range, true, ixpath);
let s = s.map(|item| match item {
Ok(item) => {
use archeng::blockrefstream::BlockrefItem::*;
match item {
Blockref(_k, jsval) => Ok(jsval),
JsVal(jsval) => Ok(jsval),
}
}
Err(e) => Err(e),
});
let s = s.map_err(Error::from);
let s = json_lines_stream(s);
let s = s.map(|item| match item {
Ok(k) => Ok(k),
Err(e) => {
error!("observe error: {}", e);
Err(e)
}
});
Ok(response(StatusCode::OK)
.header(header::CONTENT_TYPE, APP_JSON_LINES)
.body(Body::wrap_stream(s))?)
}
}
pub struct BlockStream {}
impl BlockStream {
pub fn prefix() -> &'static str {
"/api/4/channelarchiver/blockstream"
}
pub fn name() -> &'static str {
"BlockStream"
}
pub fn handler(path: &str) -> Option<Self> {
if path.starts_with(Self::prefix()) {
Some(Self {})
} else {
None
}
}
pub async fn handle(&self, req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
if req.method() != Method::GET {
return Ok(response(StatusCode::NOT_ACCEPTABLE).body(Body::empty())?);
}
info!("{} handle uri: {:?}", Self::name(), req.uri());
let database = &node_config.node_config.cluster.database;
let range = NanoRange { beg: 0, end: u64::MAX };
let url = Url::parse(&format!("dummy:{}", req.uri()))?;
let pairs = get_url_query_pairs(&url);
let read_queue = pairs.get("readQueue").unwrap_or(&"1".to_string()).parse()?;
let channel_name = pairs.get("channelName").map(String::from).unwrap_or("NONE".into());
let channel = Channel {
backend: node_config.node_config.cluster.backend.clone(),
name: channel_name,
series: None,
};
use archapp_wrap::archapp::archeng;
let ixpaths = archeng::indexfiles::index_file_path_list(channel.clone(), database.clone()).await?;
info!("got categorized ixpaths: {:?}", ixpaths);
let ixpath = ixpaths.first().unwrap().clone();
let s = archeng::blockrefstream::blockref_stream(channel, range.clone(), true, ixpath);
let s = Box::pin(s);
let s = archeng::blockstream::BlockStream::new(s, range.clone(), read_queue);
let s = s.map(|item| match item {
Ok(item) => {
use archeng::blockstream::BlockItem;
match item {
BlockItem::EventsItem(_item) => Ok(JsVal::String("EventsItem".into())),
BlockItem::JsVal(jsval) => Ok(jsval),
}
}
Err(e) => Err(e),
});
let s = s.map_err(Error::from);
let s = json_lines_stream(s);
let s = s.map(|item| match item {
Ok(k) => Ok(k),
Err(e) => {
error!("observe error: {}", e);
Err(e)
}
});
Ok(response(StatusCode::OK)
.header(header::CONTENT_TYPE, APP_JSON_LINES)
.body(Body::wrap_stream(s))?)
}
}
pub struct ListChannelsHttpFunction {}
impl ListChannelsHttpFunction {
pub fn prefix() -> &'static str {
"/api/4/channelarchiver/list/channels"
}
pub fn name() -> &'static str {
"ListChannelsHttpFunction"
}
pub fn handler(path: &str) -> Option<Self> {
if path.starts_with(Self::prefix()) {
Some(Self {})
} else {
None
}
}
pub async fn handle(&self, req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
if req.method() != Method::GET {
return Ok(response(StatusCode::NOT_ACCEPTABLE).body(Body::empty())?);
}
info!("{} handle uri: {:?}", Self::name(), req.uri());
let conf = node_config
.node
.channel_archiver
.as_ref()
.ok_or(Error::with_msg_no_trace(
"this node is not configured as channel archiver",
))?;
let s = archapp_wrap::archapp::archeng::list_all_channels(conf);
let s = futures_util::stream::unfold(s, |mut st| async move {
let x = st.next().await;
match x {
Some(x) => match x {
Ok(x) => {
let mut x = serde_json::to_vec(&x).unwrap();
x.push(b'\n');
Some((Ok::<_, Error>(x), st))
}
Err(e) => {
error!("{:?}", e);
None
}
},
None => None,
}
});
Ok(response(StatusCode::OK)
.header(header::CONTENT_TYPE, APP_JSON_LINES)
.body(Body::wrap_stream(s))?)
}
}

9
httpret/src/channelconfig.rs
View File

@@ -271,11 +271,10 @@ pub async fn channel_config(req: Request<Body>, node_config: &NodeConfigCached)
let conf = if let Some(scyco) = &node_config.node_config.cluster.scylla {
let pgconf = node_config.node_config.cluster.database.clone();
config_from_scylla(q, pgconf, scyco.clone(), node_config).await?
} else if let Some(conf) = &node_config.node.channel_archiver {
archapp_wrap::archapp::archeng::channel_config_from_db(&q, conf, &node_config.node_config.cluster.database)
.await?
} else if let Some(conf) = &node_config.node.archiver_appliance {
archapp_wrap::channel_config(&q, conf).await?
} else if let Some(_) = &node_config.node.channel_archiver {
return Err(Error::with_msg_no_trace("archapp not built"));
} else if let Some(_) = &node_config.node.archiver_appliance {
return Err(Error::with_msg_no_trace("archapp not built"));
} else {
parse::channelconfig::channel_config(&q, &node_config.node).await?
};

110
httpret/src/httpret.rs
View File

@@ -1,6 +1,5 @@
pub mod api1;
pub mod bodystream;
pub mod channelarchiver;
pub mod channelconfig;
pub mod download;
pub mod err;
@@ -29,7 +28,6 @@ use net::SocketAddr;
use netpod::log::*;
use netpod::query::BinnedQuery;
use netpod::timeunits::SEC;
use netpod::{get_url_query_pairs, Channel};
use netpod::{FromUrl, NodeConfigCached, NodeStatus, NodeStatusArchiverAppliance};
use netpod::{ACCEPT_ALL, APP_JSON, APP_JSON_LINES, APP_OCTET};
use nodenet::conn::events_service;
@@ -295,24 +293,6 @@ async fn http_service_try(req: Request<Body>, node_config: &NodeConfigCached) ->
} else {
Ok(response(StatusCode::METHOD_NOT_ALLOWED).body(Body::empty())?)
}
} else if path == "/api/4/archapp/files/scan/msgs" {
if req.method() == Method::GET {
Ok(archapp_scan_files(req, &node_config).await?)
} else {
Ok(response(StatusCode::METHOD_NOT_ALLOWED).body(Body::empty())?)
}
} else if path == "/api/4/archapp/files/scan/insert" {
if req.method() == Method::GET {
Ok(archapp_scan_files_insert(req, &node_config).await?)
} else {
Ok(response(StatusCode::METHOD_NOT_ALLOWED).body(Body::empty())?)
}
} else if path == "/api/4/archapp/channel/info" {
if req.method() == Method::GET {
Ok(archapp_channel_info(req, &node_config).await?)
} else {
Ok(response(StatusCode::METHOD_NOT_ALLOWED).body(Body::empty())?)
}
} else if let Some(h) = download::DownloadHandler::handler(&req) {
h.handle(req, &node_config).await
} else if let Some(h) = settings::SettingsThreadsMaxHandler::handler(&req) {
@@ -335,22 +315,6 @@ async fn http_service_try(req: Request<Body>, node_config: &NodeConfigCached) ->
h.handle(req, &node_config).await
} else if let Some(h) = pulsemap::Api4MapPulseHttpFunction::handler(&req) {
h.handle(req, &node_config).await
} else if let Some(h) = channelarchiver::ListIndexFilesHttpFunction::handler(path) {
h.handle(req, &node_config).await
} else if let Some(h) = channelarchiver::ListChannelsHttpFunction::handler(path) {
h.handle(req, &node_config).await
} else if let Some(h) = channelarchiver::ScanIndexFiles::handler(path) {
h.handle(req, &node_config).await
} else if let Some(h) = channelarchiver::ScanChannels::handler(path) {
h.handle(req, &node_config).await
} else if let Some(h) = channelarchiver::ScanConfigs::handler(path) {
h.handle(req, &node_config).await
} else if let Some(h) = channelarchiver::ChannelNames::handler(path) {
h.handle(req, &node_config).await
} else if let Some(h) = channelarchiver::BlockRefStream::handler(path) {
h.handle(req, &node_config).await
} else if let Some(h) = channelarchiver::BlockStream::handler(path) {
h.handle(req, &node_config).await
} else if let Some(h) = api1::RequestStatusHandler::handler(&req) {
h.handle(req, &node_config).await
} else if path.starts_with("/api/1/documentation/") {
@@ -826,77 +790,3 @@ pub fn status_board() -> Result<RwLockWriteGuard<'static, StatusBoard>, Error> {
Err(e) => Err(Error::with_msg(format!("{e:?}"))),
}
}
pub async fn archapp_scan_files(req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
let url = Url::parse(&format!("dummy:{}", req.uri()))?;
let pairs = get_url_query_pairs(&url);
let res = archapp_wrap::scan_files(pairs, node_config.clone()).await?;
let ret = response(StatusCode::OK)
.header(http::header::CONTENT_TYPE, APP_JSON_LINES)
.body(Body::wrap_stream(res.map(|k| match k {
Ok(k) => match k.serialize() {
Ok(mut item) => {
item.push(0xa);
Ok(item)
}
Err(e) => Err(e),
},
Err(e) => match serde_json::to_vec(&e) {
Ok(mut item) => {
item.push(0xa);
Ok(item)
}
Err(e) => Err(e.into()),
},
})))?;
Ok(ret)
}
pub async fn archapp_scan_files_insert(
req: Request<Body>,
node_config: &NodeConfigCached,
) -> Result<Response<Body>, Error> {
let url = Url::parse(&format!("dummy:{}", req.uri()))?;
let pairs = get_url_query_pairs(&url);
let res = archapp_wrap::scan_files_insert(pairs, node_config.clone()).await?;
let ret = response(StatusCode::OK)
.header(http::header::CONTENT_TYPE, APP_JSON_LINES)
.body(Body::wrap_stream(res.map(|k| match k {
Ok(k) => match k.serialize() {
Ok(mut item) => {
item.push(0xa);
Ok(item)
}
Err(e) => Err(e),
},
Err(e) => match serde_json::to_vec(&e) {
Ok(mut item) => {
item.push(0xa);
Ok(item)
}
Err(e) => Err(e.into()),
},
})))?;
Ok(ret)
}
pub async fn archapp_channel_info(req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
let url = Url::parse(&format!("dummy:{}", req.uri()))?;
let pairs = get_url_query_pairs(&url);
let channel = Channel::from_pairs(&pairs)?;
match archapp_wrap::channel_info(&channel, node_config).await {
Ok(res) => {
let buf = serde_json::to_vec(&res)?;
let ret = response(StatusCode::OK)
.header(http::header::CONTENT_TYPE, APP_JSON)
.body(Body::from(buf))?;
Ok(ret)
}
Err(e) => {
let ret = response(StatusCode::INTERNAL_SERVER_ERROR)
.header(http::header::CONTENT_TYPE, "text/text")
.body(Body::from(format!("{:?}", e)))?;
Ok(ret)
}
}
}