From 7c5ac697af8d43add54e026491c4c1b470859798 Mon Sep 17 00:00:00 2001 From: Dominik Werder Date: Wed, 27 Jul 2022 15:19:46 +0200 Subject: [PATCH] Moved handler, adjust defaults --- netfetch/src/ca.rs | 34 ++++---- netfetch/src/metrics.rs | 167 +--------------------------------------- readme.md | 51 ++++++++++++ 3 files changed, 67 insertions(+), 185 deletions(-) create mode 100644 readme.md diff --git a/netfetch/src/ca.rs b/netfetch/src/ca.rs index 3dd0090..c2ae97b 100644 --- a/netfetch/src/ca.rs +++ b/netfetch/src/ca.rs @@ -45,16 +45,14 @@ struct ChannelConfig { backend: String, channels: Vec, search: Vec, - addr_bind: Ipv4Addr, - addr_conn: Ipv4Addr, - whitelist: String, - blacklist: String, + addr_bind: Option, + addr_conn: Option, + whitelist: Option, + blacklist: Option, max_simul: Option, timeout: Option, - #[serde(default)] - abort_after_search: u32, - pgconf: Database, - scyconf: ScyllaConfig, + postgresql: Database, + scylla: ScyllaConfig, array_truncate: Option, insert_worker_count: Option, insert_scylla_sessions: Option, @@ -74,8 +72,8 @@ pub async fn parse_config(config: PathBuf) -> Result { file.read_to_end(&mut buf).await?; let mut conf: ChannelConfig = serde_yaml::from_slice(&buf).map_err(|e| Error::with_msg_no_trace(format!("{:?}", e)))?; - let re_p = regex::Regex::new(&conf.whitelist)?; - let re_n = regex::Regex::new(&conf.blacklist)?; + let re_p = regex::Regex::new(&conf.whitelist.unwrap_or("--nothing-whitelisted--".into()))?; + let re_n = regex::Regex::new(&conf.blacklist.unwrap_or("--nothing-blacklisted--".into()))?; conf.channels = conf .channels .into_iter() @@ -93,17 +91,16 @@ pub async fn parse_config(config: PathBuf) -> Result { backend: conf.backend, channels: conf.channels, search: conf.search, - addr_bind: conf.addr_bind, - addr_conn: conf.addr_conn, + addr_bind: conf.addr_bind.unwrap_or(Ipv4Addr::new(0, 0, 0, 0)), + addr_conn: conf.addr_conn.unwrap_or(Ipv4Addr::new(255, 255, 255, 255)), timeout: conf.timeout.unwrap_or(2000), - abort_after_search: conf.abort_after_search, - pgconf: conf.pgconf, - scyconf: conf.scyconf, + pgconf: conf.postgresql, + scyconf: conf.scylla, array_truncate: conf.array_truncate.unwrap_or(512), - insert_worker_count: conf.insert_worker_count.unwrap_or(8), + insert_worker_count: conf.insert_worker_count.unwrap_or(800), insert_scylla_sessions: conf.insert_scylla_sessions.unwrap_or(1), - insert_queue_max: conf.insert_queue_max.unwrap_or(32), - insert_item_queue_cap: conf.insert_item_queue_cap.unwrap_or(380000), + insert_queue_max: conf.insert_queue_max.unwrap_or(64), + insert_item_queue_cap: conf.insert_item_queue_cap.unwrap_or(200000), api_bind: conf.api_bind.unwrap_or_else(|| "0.0.0.0:3011".into()), local_epics_hostname: conf.local_epics_hostname, }) @@ -116,7 +113,6 @@ pub struct CaConnectOpts { pub addr_bind: Ipv4Addr, pub addr_conn: Ipv4Addr, pub timeout: u64, - pub abort_after_search: u32, pub pgconf: Database, pub scyconf: ScyllaConfig, pub array_truncate: usize, diff --git a/netfetch/src/metrics.rs b/netfetch/src/metrics.rs index 53241ec..2b7da36 100644 --- a/netfetch/src/metrics.rs +++ b/netfetch/src/metrics.rs @@ -3,30 +3,11 @@ use crate::ca::{CommandQueueSet, IngestCommons}; use axum::extract::Query; use http::request::Parts; use log::*; -use serde::Deserialize; use std::collections::HashMap; use std::net::SocketAddrV4; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; -#[allow(unused)] -#[derive(Debug, Deserialize)] -struct PromLabels { - start: Option, - end: Option, - //#[serde(rename = "match[]")] - //pattern: Option>, -} - -#[allow(unused)] -#[derive(Debug, Deserialize)] -struct PromLabelValues { - start: Option, - end: Option, - //#[serde(rename = "match[]")] - //pattern: Option>, -} - async fn get_empty() -> String { format!("") } @@ -140,60 +121,6 @@ async fn channel_remove( } } -async fn prom_query( - Query(params): Query>, - parts: Parts, - body: bytes::Bytes, -) -> axum::Json { - use axum::Json; - info!("/api/v1/query params {:?} {:?}", params, parts); - let url = url::Url::parse(&format!("dummy://{}", &parts.uri)); - info!("/api/v1/query parsed url: {:?}", url); - let body_str = String::from_utf8_lossy(&body); - info!("/api/v1/query body_str: {:?}", body_str); - let formurl = url::Url::parse(&format!("dummy:///?{}", body_str)); - info!("/api/v1/query formurl: {:?}", formurl); - let res = serde_json::json!({ - "status": "success", - "data": { - "resultType": "scalar", - "result": [40, "2"] - } - }); - Json(res) -} - -async fn prom_query_range( - Query(params): Query>, - parts: Parts, - body: bytes::Bytes, -) -> axum::Json { - use axum::Json; - info!("/api/v1/query_range {:?} Query(params) {:?}", parts, params); - let url = url::Url::parse(&format!("dummy://{}", &parts.uri)); - info!("/api/v1/query_range parsed url: {:?}", url); - let body_str = String::from_utf8_lossy(&body); - info!("/api/v1/query_range body_str: {:?}", body_str); - let formurl = url::Url::parse(&format!("dummy:///?{}", body_str)); - info!("/api/v1/query_range formurl: {:?}", formurl); - let res = serde_json::json!({ - "status": "success", - "data": { - "resultType": "matrix", - "result": [ - { - "metric": { - "__name__": "series1", - }, - "values": [ - ] - } - ] - } - }); - Json(res) -} - pub async fn start_metrics_service( bind_to: String, insert_frac: Arc, @@ -201,8 +128,7 @@ pub async fn start_metrics_service( command_queue_set: Arc, ingest_commons: Arc, ) { - use axum::routing::{get, post, put}; - use axum::Form; + use axum::routing::{get, put}; use axum::{extract, Router}; let app = Router::new() .route( @@ -328,97 +254,6 @@ pub async fn start_metrics_service( insert_ivl_min.store(v.0, Ordering::Release); }), ) - .route( - "/api/v1/status/buildinfo", - get(|| async { - let res = serde_json::json!({ - "status": "success", - "data": { - "version": "2.37", - "revision": "daqingest", - "branch": "dev", - "buildUser": "dominik.werder", - "buildDate": "2022-07-21", - "goVersion": "nogo" - } - }); - serde_json::to_string(&res).unwrap() - }), - ) - .route("/api/v1/query", post(prom_query)) - .route("/api/v1/query_range", post(prom_query_range)) - .route( - "/api/v1/labels", - post(|Form(_form): Form| async move { - let res = { - serde_json::json!({ - "status": "success", - "data": ["__name__", "instance"] - }) - }; - serde_json::to_string(&res).unwrap() - }), - ) - .route( - "/api/v1/label/__name__/values", - get(|| async move { - let res = { - serde_json::json!({ - "status": "success", - "data": ["series1", "series2"] - }) - }; - serde_json::to_string(&res).unwrap() - }), - ) - .route( - "/api/v1/label/instance/values", - get(|| async move { - let res = { - serde_json::json!({ - "status": "success", - "data": ["node1", "node2"] - }) - }; - serde_json::to_string(&res).unwrap() - }), - ) - .route( - "/api/v1/metadata", - get(|| async move { - let res = { - serde_json::json!({ - "status": "success", - "data": {} - }) - }; - serde_json::to_string(&res).unwrap() - }), - ) - .route( - "/api/v1/series", - post(|parts: Parts, body: bytes::Bytes| async move { - info!("Asked for series, form: {parts:?}"); - let url = url::Url::parse(&format!("http://dummy{}", parts.uri)) - .unwrap_or_else(|_| url::Url::parse("http://a/").unwrap()); - info!("PARSED SERIES URL {:?}", url); - let bodyparams = url::Url::parse(&String::from_utf8_lossy(&body)); - info!("BODY PARAMS: {:?}", bodyparams); - let res = { - serde_json::json!({ - "status": "success", - "data": [ - { - "__name__": "series1", - "job": "daqingest", - "instance": "node1" - } - ] - }) - }; - serde_json::to_string(&res).unwrap() - }), - ) .fallback( get(|parts: Parts, body: extract::RawBody| async move { let bytes = hyper::body::to_bytes(body.0).await.unwrap(); diff --git a/readme.md b/readme.md new file mode 100644 index 0000000..f6ceb76 --- /dev/null +++ b/readme.md @@ -0,0 +1,51 @@ +# DAQ Ingest + +## Build + +``` +cargo build --release +``` + +The resulting binary is found at `target/release/daqingest` and dynamically links only +to the most basic linux system libraries. + + +## Run + +``` +./daqingest channel-access ca-ingest +``` + + +## Config file example + +```yml +# Address to bind the HTTP API to, for runtime control and Prometheus metrics scrape: +api_bind: "0.0.0.0:3011" +# The hostname to send to channel access peers as our own hostname: +local_epics_hostname: sf-daqsync-02.psi.ch +# The backend name to use for the channels handled by this daqingest instance: +backend: scylla +# Hosts to use for channel access search: +search: + - "172.26.0.255" + - "172.26.2.255" + - "172.26.8.255" + - "..." +postgresql: + host: postgresql-host + port: 5432 + user: database-username + pass: the-password + name: the-database-name +scylla: + hosts: + - "sf-nube-11:19042" + - "sf-nube-12:19042" + - "sf-nube-13:19042" + - "sf-nube-14:19042" + keyspace: ks1 +channels: + - "SOME-CHANNEL:1" + - "OTHER-CHANNEL:2" +```