Compare commits

...

3 Commits

Author SHA1 Message Date
chylex 8a9a5b9aaa
Release 1.0.0 2023-10-04 15:38:35 +02:00
chylex 974f7f4035
Migrate error handling to anyhow crate 2023-10-04 15:34:08 +02:00
chylex bbc416b8d3
Refactor module structure and file names 2023-10-04 15:16:11 +02:00
12 changed files with 227 additions and 287 deletions

7
Cargo.lock generated
View File

@ -17,10 +17,17 @@ version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe"
[[package]]
name = "anyhow"
version = "1.0.75"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a4668cab20f66d8d020e1fbc0ebe47217433c1b6c8f2040faf858554e394ace6"
[[package]]
name = "apache_prometheus_exporter"
version = "0.1.0"
dependencies = [
"anyhow",
"hyper",
"notify",
"path-slash",

View File

@ -1,6 +1,6 @@
[package]
name = "apache_prometheus_exporter"
version = "0.1.0"
version = "1.0.0"
edition = "2021"
[[bin]]
@ -13,6 +13,7 @@ lto = true
codegen-units = 1
[dependencies]
anyhow = "1.0.75"
hyper = { version = "0.14.27", default-features = false, features = ["http1", "server", "runtime"] }
notify = { version = "6.1.1", default-features = false, features = ["macos_kqueue"] }
path-slash = "0.2.1"

View File

@ -1,9 +1,9 @@
use std::{env, io};
use std::env::VarError;
use std::fs::DirEntry;
use std::io;
use std::io::ErrorKind;
use std::path::{Path, PathBuf};
use anyhow::{anyhow, bail, Result};
use path_slash::PathExt;
/// Reads and parses an environment variable that determines the path and file name pattern of log files.
@ -13,34 +13,22 @@ use path_slash::PathExt;
/// 1. A simple path to a file.
/// 2. A path with a wildcard anywhere in the file name.
/// 3. A path with a standalone wildcard component (i.e. no prefix or suffix in the folder name).
pub fn parse_log_file_pattern_from_env(variable_name: &str) -> Result<LogFilePattern, String> {
match env::var(variable_name) {
Ok(str) => {
let pattern_str = Path::new(&str).to_slash().ok_or(format!("Environment variable {} contains an invalid path.", variable_name))?;
parse_log_file_pattern_from_str(&pattern_str)
}
Err(err) => match err {
VarError::NotPresent => Err(format!("Environment variable {} must be set.", variable_name)),
VarError::NotUnicode(_) => Err(format!("Environment variable {} contains invalid characters.", variable_name))
}
}
}
fn parse_log_file_pattern_from_str(pattern_str: &str) -> Result<LogFilePattern, String> {
if pattern_str.trim().is_empty() {
return Err(String::from("Path is empty."));
pub fn parse_log_file_pattern_from_str(pattern: &str) -> Result<LogFilePattern> {
let pattern = Path::new(pattern).to_slash().ok_or_else(|| anyhow!("Path is invalid"))?;
if pattern.trim().is_empty() {
bail!("Path is empty");
}
if let Some((left, right)) = pattern_str.split_once('*') {
if let Some((left, right)) = pattern.split_once('*') {
parse_log_file_pattern_split_on_wildcard(left, right)
} else {
Ok(LogFilePattern::WithoutWildcard(pattern_str.to_string()))
Ok(LogFilePattern::WithoutWildcard(pattern.to_string()))
}
}
fn parse_log_file_pattern_split_on_wildcard(left: &str, right: &str) -> Result<LogFilePattern, String> {
fn parse_log_file_pattern_split_on_wildcard(left: &str, right: &str) -> Result<LogFilePattern> {
if left.contains('*') || right.contains('*') {
return Err(String::from("Path has too many wildcards."));
bail!("Path has too many wildcards");
}
if left.ends_with('/') && right.starts_with('/') {
@ -51,7 +39,7 @@ fn parse_log_file_pattern_split_on_wildcard(left: &str, right: &str) -> Result<L
}
if right.contains('/') {
return Err(String::from("Path has a folder wildcard with a prefix or suffix."));
bail!("Path has a folder wildcard with a prefix or suffix");
}
if let Some((folder_path, file_name_prefix)) = left.rsplit_once('/') {
@ -122,10 +110,7 @@ impl LogFilePattern {
}
fn search_without_wildcard(path_str: &String) -> Result<Vec<LogFilePath>, io::Error> {
let path = Path::new(path_str);
let is_valid = path.is_file() || matches!(path.parent(), Some(parent) if parent.is_dir());
if is_valid {
if Path::new(path_str).is_file() {
Ok(vec![LogFilePath::with_empty_label(path_str)])
} else {
Err(io::Error::from(ErrorKind::NotFound))
@ -178,27 +163,27 @@ impl LogFilePath {
#[cfg(test)]
mod tests {
use crate::log_file_pattern::{LogFilePattern, parse_log_file_pattern_from_str};
use super::{LogFilePattern, parse_log_file_pattern_from_str};
#[test]
fn empty_path() {
assert!(matches!(parse_log_file_pattern_from_str(""), Err(err) if err == "Path is empty."));
assert!(matches!(parse_log_file_pattern_from_str(" "), Err(err) if err == "Path is empty."));
assert!(matches!(parse_log_file_pattern_from_str(""), Err(err) if err.to_string() == "Path is empty"));
assert!(matches!(parse_log_file_pattern_from_str(" "), Err(err) if err.to_string() == "Path is empty"));
}
#[test]
fn too_many_wildcards() {
assert!(matches!(parse_log_file_pattern_from_str("/path/*/to/files/*.log"), Err(err) if err == "Path has too many wildcards."));
assert!(matches!(parse_log_file_pattern_from_str("/path/*/to/files/*.log"), Err(err) if err.to_string() == "Path has too many wildcards"));
}
#[test]
fn folder_wildcard_with_prefix_not_supported() {
assert!(matches!(parse_log_file_pattern_from_str("/path/*abc/to/files/access.log"), Err(err) if err == "Path has a folder wildcard with a prefix or suffix."));
assert!(matches!(parse_log_file_pattern_from_str("/path/*abc/to/files/access.log"), Err(err) if err.to_string() == "Path has a folder wildcard with a prefix or suffix"));
}
#[test]
fn folder_wildcard_with_suffix_not_supported() {
assert!(matches!(parse_log_file_pattern_from_str("/path/abc*/to/files/access.log"), Err(err) if err == "Path has a folder wildcard with a prefix or suffix."));
assert!(matches!(parse_log_file_pattern_from_str("/path/abc*/to/files/access.log"), Err(err) if err.to_string() == "Path has a folder wildcard with a prefix or suffix"));
}
#[test]

View File

@ -2,6 +2,7 @@ use std::cmp::max;
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::{anyhow, bail, Context, Result};
use notify::{Event, EventKind};
use notify::event::{CreateKind, ModifyKind};
use tokio::fs::File;
@ -9,12 +10,12 @@ use tokio::io::{AsyncBufReadExt, BufReader, Lines};
use tokio::sync::mpsc;
use tokio::sync::mpsc::Receiver;
use crate::ApacheMetrics;
use crate::fs_watcher::{FsEventCallbacks, FsWatcher};
use crate::log_file_pattern::LogFilePath;
use crate::logs::filesystem_watcher::{FsEventCallbacks, FsWatcher};
use crate::logs::log_file_pattern::LogFilePath;
use crate::metrics::Metrics;
#[derive(Copy, Clone, PartialEq)]
enum LogFileKind {
pub enum LogFileKind {
Access,
Error,
}
@ -30,26 +31,12 @@ impl LogFileMetadata {
}
}
pub async fn start_log_watcher(access_log_files: Vec<LogFilePath>, error_log_files: Vec<LogFilePath>, metrics: ApacheMetrics) -> bool {
let mut watcher = LogWatcherConfiguration::new();
for log_file in access_log_files.into_iter() {
watcher.add_file(log_file, LogFileKind::Access);
}
for log_file in error_log_files.into_iter() {
watcher.add_file(log_file, LogFileKind::Error);
}
watcher.start(&metrics).await
}
struct LogWatcherConfiguration {
pub struct LogWatcherConfiguration {
files: Vec<(PathBuf, LogFileMetadata)>,
}
impl LogWatcherConfiguration {
fn new() -> LogWatcherConfiguration {
pub fn new() -> LogWatcherConfiguration {
LogWatcherConfiguration { files: Vec::new() }
}
@ -57,17 +44,16 @@ impl LogWatcherConfiguration {
return self.files.iter().filter(|(_, metadata)| metadata.kind == kind).count();
}
fn add_file(&mut self, log_file: LogFilePath, kind: LogFileKind) {
pub fn add_file(&mut self, log_file: LogFilePath, kind: LogFileKind) {
let path = log_file.path;
let label = log_file.label;
let metadata = LogFileMetadata { kind, label };
self.files.push((path, metadata));
}
async fn start(self, metrics: &ApacheMetrics) -> bool {
pub async fn start(self, metrics: &Metrics) -> Result<()> {
if self.files.is_empty() {
println!("[LogWatcher] No log files provided.");
return false;
bail!("No log files provided");
}
println!("[LogWatcher] Watching {} access log file(s) and {} error log file(s).", self.count_files_of_kind(LogFileKind::Access), self.count_files_of_kind(LogFileKind::Error));
@ -87,32 +73,16 @@ impl LogWatcherConfiguration {
prepared_files.push(PreparedFile { path, metadata, fs_event_receiver });
}
let fs_watcher = match FsWatcher::new(fs_callbacks) {
Ok(fs_watcher) => fs_watcher,
Err(e) => {
println!("[LogWatcher] Error creating filesystem watcher: {}", e);
return false;
}
};
let fs_watcher = FsWatcher::new(fs_callbacks).context("Could not create filesystem watcher")?;
for file in &prepared_files {
let file_path = &file.path;
if !file_path.is_absolute() {
println!("[LogWatcher] Error creating filesystem watcher, path is not absolute: {}", file_path.to_string_lossy());
return false;
bail!("Path is not absolute: {}", file_path.to_string_lossy());
}
let parent_path = if let Some(parent) = file_path.parent() {
parent
} else {
println!("[LogWatcher] Error creating filesystem watcher for parent directory of file \"{}\", parent directory does not exist", file_path.to_string_lossy());
return false;
};
if let Err(e) = fs_watcher.watch(parent_path).await {
println!("[LogWatcher] Error creating filesystem watcher for directory \"{}\": {}", parent_path.to_string_lossy(), e);
return false;
}
let parent_path = file_path.parent().ok_or_else(|| anyhow!("Path has no parent: {}", file_path.to_string_lossy()))?;
fs_watcher.watch(parent_path).await.with_context(|| format!("Could not create filesystem watcher for directory: {}", parent_path.to_string_lossy()))?;
}
let fs_watcher = Arc::new(fs_watcher);
@ -122,15 +92,13 @@ impl LogWatcherConfiguration {
let _ = metrics.requests_total.get_or_create(&label_set);
let _ = metrics.errors_total.get_or_create(&label_set);
let log_watcher = match LogWatcher::create(file.path, file.metadata, metrics.clone(), Arc::clone(&fs_watcher), file.fs_event_receiver).await {
Some(log_watcher) => log_watcher,
None => return false,
};
let log_watcher = LogWatcher::create(file.path.clone(), file.metadata, metrics.clone(), Arc::clone(&fs_watcher), file.fs_event_receiver);
let log_watcher = log_watcher.await.with_context(|| format!("Could not watch log file: {}", file.path.to_string_lossy()))?;
tokio::spawn(log_watcher.watch());
}
true
Ok(())
}
}
@ -141,14 +109,10 @@ struct LogWatcher {
}
impl LogWatcher {
async fn create(path: PathBuf, metadata: LogFileMetadata, metrics: ApacheMetrics, fs_watcher: Arc<FsWatcher>, fs_event_receiver: Receiver<Event>) -> Option<Self> {
let state = match LogWatchingState::initialize(path.clone(), fs_watcher).await {
Some(state) => state,
None => return None,
};
async fn create(path: PathBuf, metadata: LogFileMetadata, metrics: Metrics, fs_watcher: Arc<FsWatcher>, fs_event_receiver: Receiver<Event>) -> Result<Self> {
let state = LogWatchingState::initialize(path.clone(), fs_watcher).await?;
let processor = LogLineProcessor { path, metadata, metrics };
Some(LogWatcher { state, processor, fs_event_receiver })
Ok(LogWatcher { state, processor, fs_event_receiver })
}
async fn watch(mut self) {
@ -190,8 +154,11 @@ impl LogWatcher {
}
self.state = match self.state.reinitialize().await {
Some(state) => state,
None => break 'read_loop,
Ok(state) => state,
Err(e) => {
println!("Could not re-watch log file \"{}\": {}", path.to_string_lossy(), e);
break 'read_loop;
}
};
continue 'read_loop;
@ -236,24 +203,16 @@ struct LogWatchingState {
impl LogWatchingState {
const DEFAULT_BUFFER_CAPACITY: usize = 1024 * 4;
async fn initialize(path: PathBuf, fs_watcher: Arc<FsWatcher>) -> Option<LogWatchingState> {
if let Err(e) = fs_watcher.watch(&path).await {
println!("[LogWatcher] Error creating filesystem watcher for file \"{}\": {}", path.to_string_lossy(), e);
return None;
}
async fn initialize(path: PathBuf, fs_watcher: Arc<FsWatcher>) -> Result<LogWatchingState> {
fs_watcher.watch(&path).await.context("Could not create filesystem watcher")?;
let lines = match File::open(&path).await {
Ok(file) => BufReader::with_capacity(Self::DEFAULT_BUFFER_CAPACITY, file).lines(),
Err(e) => {
println!("[LogWatcher] Error opening file \"{}\": {}", path.to_string_lossy(), e);
return None;
}
};
let file = File::open(&path).await.context("Could not open file")?;
let lines = BufReader::with_capacity(Self::DEFAULT_BUFFER_CAPACITY, file).lines();
Some(LogWatchingState { path, lines, fs_watcher })
Ok(LogWatchingState { path, lines, fs_watcher })
}
async fn reinitialize(self) -> Option<LogWatchingState> {
async fn reinitialize(self) -> Result<LogWatchingState> {
LogWatchingState::initialize(self.path, self.fs_watcher).await
}
}
@ -261,7 +220,7 @@ impl LogWatchingState {
struct LogLineProcessor {
path: PathBuf,
metadata: LogFileMetadata,
metrics: ApacheMetrics,
metrics: Metrics,
}
impl LogLineProcessor {

48
src/logs/mod.rs Normal file
View File

@ -0,0 +1,48 @@
use std::env;
use std::env::VarError;
use anyhow::{anyhow, bail, Context, Result};
use log_file_watcher::{LogFileKind, LogWatcherConfiguration};
use crate::logs::log_file_pattern::{LogFilePath, parse_log_file_pattern_from_str};
use crate::metrics::Metrics;
mod access_log_parser;
mod filesystem_watcher;
mod log_file_pattern;
mod log_file_watcher;
pub fn find_log_files(environment_variable_name: &str, log_kind: &str) -> Result<Vec<LogFilePath>> {
let log_file_pattern_str = env::var(environment_variable_name).map_err(|err| match err {
VarError::NotPresent => anyhow!("Environment variable {} must be set", environment_variable_name),
VarError::NotUnicode(_) => anyhow!("Environment variable {} contains invalid characters", environment_variable_name)
})?;
let log_file_pattern = parse_log_file_pattern_from_str(&log_file_pattern_str).with_context(|| format!("Could not parse pattern: {}", log_file_pattern_str))?;
let log_files = log_file_pattern.search().with_context(|| format!("Could not search files: {}", log_file_pattern_str))?;
if log_files.is_empty() {
bail!("No files match pattern: {}", log_file_pattern_str);
}
for log_file in &log_files {
println!("Found {} file: {} (label \"{}\")", log_kind, log_file.path.display(), log_file.label);
}
Ok(log_files)
}
pub async fn start_log_watcher(access_log_files: Vec<LogFilePath>, error_log_files: Vec<LogFilePath>, metrics: Metrics) -> Result<()> {
let mut watcher = LogWatcherConfiguration::new();
for log_file in access_log_files.into_iter() {
watcher.add_file(log_file, LogFileKind::Access);
}
for log_file in error_log_files.into_iter() {
watcher.add_file(log_file, LogFileKind::Error);
}
watcher.start(&metrics).await
}

View File

@ -1,99 +1,38 @@
use std::env;
use std::net::{IpAddr, SocketAddr};
use std::process::ExitCode;
use std::str::FromStr;
use std::sync::Mutex;
use anyhow::{anyhow, Context};
use tokio::signal;
use crate::apache_metrics::ApacheMetrics;
use crate::log_file_pattern::{LogFilePath, parse_log_file_pattern_from_env};
use crate::log_watcher::start_log_watcher;
use crate::web_server::WebServer;
use crate::metrics::Metrics;
use crate::web::WebServer;
mod apache_metrics;
mod fs_watcher;
mod log_file_pattern;
mod log_parser;
mod log_watcher;
mod web_server;
mod logs;
mod metrics;
mod web;
const ACCESS_LOG_FILE_PATTERN: &str = "ACCESS_LOG_FILE_PATTERN";
const ERROR_LOG_FILE_PATTERN: &str = "ERROR_LOG_FILE_PATTERN";
fn find_log_files(environment_variable_name: &str, log_kind: &str) -> Option<Vec<LogFilePath>> {
let log_file_pattern = match parse_log_file_pattern_from_env(environment_variable_name) {
Ok(pattern) => pattern,
Err(error) => {
println!("Error: {}", error);
return None;
}
};
let log_files = match log_file_pattern.search() {
Ok(files) => files,
Err(error) => {
println!("Error searching {} files: {}", log_kind, error);
return None;
}
};
if log_files.is_empty() {
println!("Found no matching {} files.", log_kind);
return None;
}
for log_file in &log_files {
println!("Found {} file: {} (label \"{}\")", log_kind, log_file.path.display(), log_file.label);
}
Some(log_files)
}
#[tokio::main(flavor = "current_thread")]
async fn main() -> ExitCode {
async fn main() -> anyhow::Result<()> {
let host = env::var("HTTP_HOST").unwrap_or(String::from("127.0.0.1"));
let bind_ip = match IpAddr::from_str(&host) {
Ok(addr) => addr,
Err(_) => {
println!("Invalid HTTP host: {}", host);
return ExitCode::FAILURE;
}
};
let bind_ip = IpAddr::from_str(&host).map_err(|_| anyhow!("Invalid HTTP host: {}", host))?;
println!("Initializing exporter...");
let access_log_files = match find_log_files(ACCESS_LOG_FILE_PATTERN, "access log") {
Some(files) => files,
None => return ExitCode::FAILURE,
};
let access_log_files = logs::find_log_files(ACCESS_LOG_FILE_PATTERN, "access log").context("Could not find access log files")?;
let error_log_files = logs::find_log_files(ERROR_LOG_FILE_PATTERN, "error log").context("Could not find error log files")?;
let error_log_files = match find_log_files(ERROR_LOG_FILE_PATTERN, "error log") {
Some(files) => files,
None => return ExitCode::FAILURE,
};
let server = match WebServer::try_bind(SocketAddr::new(bind_ip, 9240)) {
Some(server) => server,
None => return ExitCode::FAILURE
};
let (metrics_registry, metrics) = ApacheMetrics::new();
if !start_log_watcher(access_log_files, error_log_files, metrics).await {
return ExitCode::FAILURE;
}
let server = WebServer::try_bind(SocketAddr::new(bind_ip, 9240)).context("Could not configure web server")?;
let (metrics_registry, metrics) = Metrics::new();
logs::start_log_watcher(access_log_files, error_log_files, metrics).await.context("Could not start watching logs")?;
tokio::spawn(server.serve(Mutex::new(metrics_registry)));
match signal::ctrl_c().await {
Ok(_) => {
println!("Received CTRL-C, shutting down...");
ExitCode::SUCCESS
}
Err(e) => {
println!("Error registering CTRL-C handler: {}", e);
ExitCode::FAILURE
}
}
signal::ctrl_c().await.with_context(|| "Could not register CTRL-C handler")?;
println!("Received CTRL-C, shutting down...");
Ok(())
}

View File

@ -4,20 +4,17 @@ use prometheus_client::registry::Registry;
type SingleLabel = [(&'static str, String); 1];
#[derive(Clone)]
pub struct ApacheMetrics {
#[derive(Clone, Default)]
pub struct Metrics {
pub requests_total: Family<SingleLabel, Counter>,
pub errors_total: Family<SingleLabel, Counter>
}
impl ApacheMetrics {
pub fn new() -> (Registry, ApacheMetrics) {
impl Metrics {
pub fn new() -> (Registry, Metrics) {
let mut registry = <Registry>::default();
let metrics = ApacheMetrics {
requests_total: Family::<SingleLabel, Counter>::default(),
errors_total: Family::<SingleLabel, Counter>::default()
};
let metrics = Metrics::default();
registry.register("apache_requests", "Number of received requests", metrics.requests_total.clone());
registry.register("apache_errors", "Number of logged errors", metrics.errors_total.clone());

View File

@ -0,0 +1,42 @@
use std::fmt;
use std::sync::{Arc, Mutex};
use hyper::{Body, http, Response, StatusCode};
use hyper::header::CONTENT_TYPE;
use prometheus_client::encoding::text::encode;
use prometheus_client::registry::Registry;
//noinspection SpellCheckingInspection
const METRICS_CONTENT_TYPE: &str = "application/openmetrics-text; version=1.0.0; charset=utf-8";
pub async fn handle(metrics_registry: Arc<Mutex<Registry>>) -> http::Result<Response<Body>> {
match try_encode(metrics_registry) {
MetricsEncodeResult::Ok(buf) => {
Response::builder().status(StatusCode::OK).header(CONTENT_TYPE, METRICS_CONTENT_TYPE).body(Body::from(buf))
}
MetricsEncodeResult::FailedAcquiringRegistryLock => {
println!("[WebServer] Failed acquiring lock on registry.");
Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::empty())
}
MetricsEncodeResult::FailedEncodingMetrics(e) => {
println!("[WebServer] Error encoding metrics: {}", e);
Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::empty())
}
}
}
enum MetricsEncodeResult {
Ok(String),
FailedAcquiringRegistryLock,
FailedEncodingMetrics(fmt::Error),
}
fn try_encode(metrics_registry: Arc<Mutex<Registry>>) -> MetricsEncodeResult {
let mut buf = String::new();
return if let Ok(metrics_registry) = metrics_registry.lock() {
encode(&mut buf, &metrics_registry).map_or_else(MetricsEncodeResult::FailedEncodingMetrics, |_| MetricsEncodeResult::Ok(buf))
} else {
MetricsEncodeResult::FailedAcquiringRegistryLock
};
}

57
src/web/mod.rs Normal file
View File

@ -0,0 +1,57 @@
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use anyhow::Context;
use hyper::{Body, Error, Method, Request, Response, Server, StatusCode};
use hyper::http::Result;
use hyper::server::Builder;
use hyper::server::conn::AddrIncoming;
use hyper::service::{make_service_fn, service_fn};
use prometheus_client::registry::Registry;
mod metrics_endpoint;
const MAX_BUFFER_SIZE: usize = 1024 * 32;
pub struct WebServer {
builder: Builder<AddrIncoming>,
}
impl WebServer {
//noinspection HttpUrlsUsage
pub fn try_bind(addr: SocketAddr) -> anyhow::Result<WebServer> {
println!("[WebServer] Starting web server on {0} with metrics endpoint: http://{0}/metrics", addr);
let builder = Server::try_bind(&addr).with_context(|| format!("Could not bind to {}", addr))?;
let builder = builder.tcp_keepalive(Some(Duration::from_secs(60)));
let builder = builder.http1_only(true);
let builder = builder.http1_keepalive(true);
let builder = builder.http1_max_buf_size(MAX_BUFFER_SIZE);
let builder = builder.http1_header_read_timeout(Duration::from_secs(10));
Ok(WebServer { builder })
}
pub async fn serve(self, metrics_registry: Mutex<Registry>) {
let metrics_registry = Arc::new(metrics_registry);
let service = make_service_fn(move |_| {
let metrics_registry = Arc::clone(&metrics_registry);
async move {
Ok::<_, Error>(service_fn(move |req| handle_request(req, Arc::clone(&metrics_registry))))
}
});
if let Err(e) = self.builder.serve(service).await {
println!("[WebServer] Error starting web server: {}", e);
}
}
}
async fn handle_request(req: Request<Body>, metrics_registry: Arc<Mutex<Registry>>) -> Result<Response<Body>> {
if req.method() == Method::GET && req.uri().path() == "/metrics" {
metrics_endpoint::handle(Arc::clone(&metrics_registry)).await
} else {
Response::builder().status(StatusCode::NOT_FOUND).body(Body::empty())
}
}

View File

@ -1,95 +0,0 @@
use std::fmt;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use hyper::{Body, Error, header, Method, Request, Response, Server, StatusCode};
use hyper::http::Result;
use hyper::server::Builder;
use hyper::server::conn::AddrIncoming;
use hyper::service::{make_service_fn, service_fn};
use prometheus_client::encoding::text::encode;
use prometheus_client::registry::Registry;
const MAX_BUFFER_SIZE: usize = 1024 * 32;
pub struct WebServer {
builder: Builder<AddrIncoming>,
}
impl WebServer {
//noinspection HttpUrlsUsage
pub fn try_bind(addr: SocketAddr) -> Option<WebServer> {
println!("[WebServer] Starting web server on {0} with metrics endpoint: http://{0}/metrics", addr);
let builder = match Server::try_bind(&addr) {
Ok(builder) => builder,
Err(e) => {
println!("[WebServer] Could not bind to {}: {}", addr, e);
return None;
}
};
let builder = builder.tcp_keepalive(Some(Duration::from_secs(60)));
let builder = builder.http1_only(true);
let builder = builder.http1_keepalive(true);
let builder = builder.http1_max_buf_size(MAX_BUFFER_SIZE);
let builder = builder.http1_header_read_timeout(Duration::from_secs(10));
Some(WebServer { builder })
}
pub async fn serve(self, metrics_registry: Mutex<Registry>) {
let metrics_registry = Arc::new(metrics_registry);
let service = make_service_fn(move |_| {
let metrics_registry = Arc::clone(&metrics_registry);
async move {
Ok::<_, Error>(service_fn(move |req| handle_request(req, Arc::clone(&metrics_registry))))
}
});
if let Err(e) = self.builder.serve(service).await {
println!("[WebServer] Error starting web server: {}", e);
}
}
}
async fn handle_request(req: Request<Body>, metrics_registry: Arc<Mutex<Registry>>) -> Result<Response<Body>> {
if req.method() == Method::GET && req.uri().path() == "/metrics" {
metrics_handler(Arc::clone(&metrics_registry)).await
} else {
Response::builder().status(StatusCode::NOT_FOUND).body(Body::empty())
}
}
//noinspection SpellCheckingInspection
async fn metrics_handler(metrics_registry: Arc<Mutex<Registry>>) -> Result<Response<Body>> {
match encode_metrics(metrics_registry) {
MetricsEncodeResult::Ok(buf) => {
Response::builder().status(StatusCode::OK).header(header::CONTENT_TYPE, "application/openmetrics-text; version=1.0.0; charset=utf-8").body(Body::from(buf))
}
MetricsEncodeResult::FailedAcquiringRegistryLock => {
println!("[WebServer] Failed acquiring lock on registry.");
Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::empty())
}
MetricsEncodeResult::FailedEncodingMetrics(e) => {
println!("[WebServer] Error encoding metrics: {}", e);
Response::builder().status(StatusCode::INTERNAL_SERVER_ERROR).body(Body::empty())
}
}
}
enum MetricsEncodeResult {
Ok(String),
FailedAcquiringRegistryLock,
FailedEncodingMetrics(fmt::Error),
}
fn encode_metrics(metrics_registry: Arc<Mutex<Registry>>) -> MetricsEncodeResult {
let mut buf = String::new();
return if let Ok(metrics_registry) = metrics_registry.lock() {
encode(&mut buf, &metrics_registry).map_or_else(MetricsEncodeResult::FailedEncodingMetrics, |_| MetricsEncodeResult::Ok(buf))
} else {
MetricsEncodeResult::FailedAcquiringRegistryLock
};
}