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

Moved handler, adjust defaults

Browse Source
This commit is contained in:
Dominik Werder
2022-07-27 15:19:46 +02:00
parent 37204cdad8
commit 7c5ac697af
3 changed files with 67 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netfetch/src/ca.rs
+15 -19
View File
@@ -45,16 +45,14 @@ struct ChannelConfig {
backend: String,
channels: Vec<String>,
search: Vec<String>,
addr_bind: Ipv4Addr,
addr_conn: Ipv4Addr,
whitelist: String,
blacklist: String,
addr_bind: Option<Ipv4Addr>,
addr_conn: Option<Ipv4Addr>,
whitelist: Option<String>,
blacklist: Option<String>,
max_simul: Option<usize>,
timeout: Option<u64>,
#[serde(default)]
abort_after_search: u32,
pgconf: Database,
scyconf: ScyllaConfig,
postgresql: Database,
scylla: ScyllaConfig,
array_truncate: Option<usize>,
insert_worker_count: Option<usize>,
insert_scylla_sessions: Option<usize>,
@@ -74,8 +72,8 @@ pub async fn parse_config(config: PathBuf) -> Result<CaConnectOpts, Error> {
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<CaConnectOpts, Error> {
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,
netfetch/src/metrics.rs
+1 -166
View File
@@ -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<String>,
end: Option<String>,
//#[serde(rename = "match[]")]
//pattern: Option<Vec<String>>,
}
#[allow(unused)]
#[derive(Debug, Deserialize)]
struct PromLabelValues {
start: Option<String>,
end: Option<String>,
//#[serde(rename = "match[]")]
//pattern: Option<Vec<String>>,
}
async fn get_empty() -> String {
format!("")
}
@@ -140,60 +121,6 @@ async fn channel_remove(
}
}
async fn prom_query(
Query(params): Query<HashMap<String, String>>,
parts: Parts,
body: bytes::Bytes,
) -> axum::Json<serde_json::Value> {
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<HashMap<String, String>>,
parts: Parts,
body: bytes::Bytes,
) -> axum::Json<serde_json::Value> {
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<AtomicU64>,
@@ -201,8 +128,7 @@ pub async fn start_metrics_service(
command_queue_set: Arc<CommandQueueSet>,
ingest_commons: Arc<IngestCommons>,
) {
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<PromLabels>| 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<hyper::Body>| async move {
let bytes = hyper::body::to_bytes(body.0).await.unwrap();
readme.md
+51
View File
@@ -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.YML>
```
## 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"
```