Replace linemux with reading output of 'tail' processes

Cargo.lock

@@ -22,7 +22,6 @@ name = "apache_prometheus_exporter"
 version = "0.1.0"
 dependencies = [
  "hyper",
- "linemux",
  "path-slash",
  "prometheus-client",
  "tokio",
@@ -73,44 +72,12 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
 
-[[package]]
-name = "crossbeam-channel"
-version = "0.5.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c2dd04ddaf88237dc3b8d8f9a3c1004b506b54b3313403944054d23c0870c521"
-dependencies = [
- "cfg-if",
- "crossbeam-utils",
-]
-
-[[package]]
-name = "crossbeam-utils"
-version = "0.8.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "51887d4adc7b564537b15adcfb307936f8075dfcd5f00dde9a9f1d29383682bc"
-dependencies = [
- "cfg-if",
- "once_cell",
-]
-
 [[package]]
 name = "dtoa"
 version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c6053ff46b5639ceb91756a85a4c8914668393a03170efd79c8884a529d80656"
 
-[[package]]
-name = "filetime"
-version = "0.2.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e94a7bbaa59354bc20dd75b67f23e2797b4490e9d6928203fb105c79e448c86c"
-dependencies = [
- "cfg-if",
- "libc",
- "redox_syscall",
- "windows-sys 0.36.1",
-]
-
 [[package]]
 name = "fnv"
 version = "1.0.7"
@@ -148,7 +115,6 @@ dependencies = [
  "futures-task",
  "pin-project-lite",
  "pin-utils",
- "slab",
 ]
 
 [[package]]
@@ -214,70 +180,18 @@ dependencies = [
  "want",
 ]
 
-[[package]]
-name = "inotify"
-version = "0.9.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f8069d3ec154eb856955c1c0fbffefbf5f3c40a104ec912d4797314c1801abff"
-dependencies = [
- "bitflags",
- "inotify-sys",
- "libc",
-]
-
-[[package]]
-name = "inotify-sys"
-version = "0.1.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e05c02b5e89bff3b946cedeca278abc628fe811e604f027c45a8aa3cf793d0eb"
-dependencies = [
- "libc",
-]
-
 [[package]]
 name = "itoa"
 version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6c8af84674fe1f223a982c933a0ee1086ac4d4052aa0fb8060c12c6ad838e754"
 
-[[package]]
-name = "kqueue"
-version = "1.0.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d6112e8f37b59803ac47a42d14f1f3a59bbf72fc6857ffc5be455e28a691f8e"
-dependencies = [
- "kqueue-sys",
- "libc",
-]
-
-[[package]]
-name = "kqueue-sys"
-version = "1.0.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8367585489f01bc55dd27404dcf56b95e6da061a256a666ab23be9ba96a2e587"
-dependencies = [
- "bitflags",
- "libc",
-]
-
 [[package]]
 name = "libc"
 version = "0.2.148"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9cdc71e17332e86d2e1d38c1f99edcb6288ee11b815fb1a4b049eaa2114d369b"
 
-[[package]]
-name = "linemux"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "feb035c7806bd7982a317d8d66815021e91f7ab14a5fbedee22b06f608f11b43"
-dependencies = [
- "futures-util",
- "notify",
- "pin-project-lite",
- "tokio",
-]
-
 [[package]]
 name = "lock_api"
 version = "0.4.8"
@@ -288,15 +202,6 @@ dependencies = [
  "scopeguard",
 ]
 
-[[package]]
-name = "log"
-version = "0.4.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "abb12e687cfb44aa40f41fc3978ef76448f9b6038cad6aef4259d3c095a2382e"
-dependencies = [
- "cfg-if",
-]
-
 [[package]]
 name = "memchr"
 version = "2.5.0"
@@ -319,28 +224,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "927a765cd3fc26206e66b296465fa9d3e5ab003e651c1b3c060e7956d96b19d2"
 dependencies = [
  "libc",
- "log",
  "wasi",
  "windows-sys 0.48.0",
 ]
 
-[[package]]
-name = "notify"
-version = "5.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "729f63e1ca555a43fe3efa4f3efdf4801c479da85b432242a7b726f353c88486"
-dependencies = [
- "bitflags",
- "crossbeam-channel",
- "filetime",
- "inotify",
- "kqueue",
- "libc",
- "mio",
- "walkdir",
- "windows-sys 0.45.0",
-]
-
 [[package]]
 name = "object"
 version = "0.32.1"
@@ -453,15 +340,6 @@ version = "0.1.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d626bb9dae77e28219937af045c257c28bfd3f69333c512553507f5f9798cb76"
 
-[[package]]
-name = "same-file"
-version = "1.0.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502"
-dependencies = [
- "winapi-util",
-]
-
 [[package]]
 name = "scopeguard"
 version = "1.1.0"
@@ -477,15 +355,6 @@ dependencies = [
  "libc",
 ]
 
-[[package]]
-name = "slab"
-version = "0.4.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4614a76b2a8be0058caa9dbbaf66d988527d86d003c11a94fbd335d7661edcef"
-dependencies = [
- "autocfg",
-]
-
 [[package]]
 name = "smallvec"
 version = "1.9.0"
@@ -589,17 +458,6 @@ version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c4f5b37a154999a8f3f98cc23a628d850e154479cd94decf3414696e12e31aaf"
 
-[[package]]
-name = "walkdir"
-version = "2.3.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "808cf2735cd4b6866113f648b791c6adc5714537bc222d9347bb203386ffda56"
-dependencies = [
- "same-file",
- "winapi",
- "winapi-util",
-]
-
 [[package]]
 name = "want"
 version = "0.3.1"
@@ -631,15 +489,6 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
 
-[[package]]
-name = "winapi-util"
-version = "0.1.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70ec6ce85bb158151cae5e5c87f95a8e97d2c0c4b001223f33a334e3ce5de178"
-dependencies = [
- "winapi",
-]
-
 [[package]]
 name = "winapi-x86_64-pc-windows-gnu"
 version = "0.4.0"
@@ -659,37 +508,13 @@ dependencies = [
  "windows_x86_64_msvc 0.36.1",
 ]
 
-[[package]]
-name = "windows-sys"
-version = "0.45.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "75283be5efb2831d37ea142365f009c02ec203cd29a3ebecbc093d52315b66d0"
-dependencies = [
- "windows-targets 0.42.2",
-]
-
 [[package]]
 name = "windows-sys"
 version = "0.48.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
 dependencies = [
- "windows-targets 0.48.5",
+ "windows-targets",
-]
-
-[[package]]
-name = "windows-targets"
-version = "0.42.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e5180c00cd44c9b1c88adb3693291f1cd93605ded80c250a75d472756b4d071"
-dependencies = [
- "windows_aarch64_gnullvm 0.42.2",
- "windows_aarch64_msvc 0.42.2",
- "windows_i686_gnu 0.42.2",
- "windows_i686_msvc 0.42.2",
- "windows_x86_64_gnu 0.42.2",
- "windows_x86_64_gnullvm 0.42.2",
- "windows_x86_64_msvc 0.42.2",
 ]
 
 [[package]]
@@ -698,21 +523,15 @@ version = "0.48.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c"
 dependencies = [
- "windows_aarch64_gnullvm 0.48.5",
+ "windows_aarch64_gnullvm",
  "windows_aarch64_msvc 0.48.5",
  "windows_i686_gnu 0.48.5",
  "windows_i686_msvc 0.48.5",
  "windows_x86_64_gnu 0.48.5",
- "windows_x86_64_gnullvm 0.48.5",
+ "windows_x86_64_gnullvm",
  "windows_x86_64_msvc 0.48.5",
 ]
 
-[[package]]
-name = "windows_aarch64_gnullvm"
-version = "0.42.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "597a5118570b68bc08d8d59125332c54f1ba9d9adeedeef5b99b02ba2b0698f8"
-
 [[package]]
 name = "windows_aarch64_gnullvm"
 version = "0.48.5"
@@ -725,12 +544,6 @@ version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9bb8c3fd39ade2d67e9874ac4f3db21f0d710bee00fe7cab16949ec184eeaa47"
 
-[[package]]
-name = "windows_aarch64_msvc"
-version = "0.42.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e08e8864a60f06ef0d0ff4ba04124db8b0fb3be5776a5cd47641e942e58c4d43"
-
 [[package]]
 name = "windows_aarch64_msvc"
 version = "0.48.5"
@@ -743,12 +556,6 @@ version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "180e6ccf01daf4c426b846dfc66db1fc518f074baa793aa7d9b9aaeffad6a3b6"
 
-[[package]]
-name = "windows_i686_gnu"
-version = "0.42.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c61d927d8da41da96a81f029489353e68739737d3beca43145c8afec9a31a84f"
-
 [[package]]
 name = "windows_i686_gnu"
 version = "0.48.5"
@@ -761,12 +568,6 @@ version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e2e7917148b2812d1eeafaeb22a97e4813dfa60a3f8f78ebe204bcc88f12f024"
 
-[[package]]
-name = "windows_i686_msvc"
-version = "0.42.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "44d840b6ec649f480a41c8d80f9c65108b92d89345dd94027bfe06ac444d1060"
-
 [[package]]
 name = "windows_i686_msvc"
 version = "0.48.5"
@@ -779,24 +580,12 @@ version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4dcd171b8776c41b97521e5da127a2d86ad280114807d0b2ab1e462bc764d9e1"
 
-[[package]]
-name = "windows_x86_64_gnu"
-version = "0.42.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8de912b8b8feb55c064867cf047dda097f92d51efad5b491dfb98f6bbb70cb36"
-
 [[package]]
 name = "windows_x86_64_gnu"
 version = "0.48.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e"
 
-[[package]]
-name = "windows_x86_64_gnullvm"
-version = "0.42.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "26d41b46a36d453748aedef1486d5c7a85db22e56aff34643984ea85514e94a3"
-
 [[package]]
 name = "windows_x86_64_gnullvm"
 version = "0.48.5"
@@ -809,12 +598,6 @@ version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c811ca4a8c853ef420abd8592ba53ddbbac90410fab6903b3e79972a631f7680"
 
-[[package]]
-name = "windows_x86_64_msvc"
-version = "0.42.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9aec5da331524158c6d1a4ac0ab1541149c0b9505fde06423b02f5ef0106b9f0"
-
 [[package]]
 name = "windows_x86_64_msvc"
 version = "0.48.5"

Cargo.toml

@@ -14,7 +14,6 @@ codegen-units = 1
 
 [dependencies]
 hyper = { version = "0.14.27", default-features = false, features = ["http1", "server", "runtime"] }
-linemux = "0.3.0"
 path-slash = "0.2.1"
 prometheus-client = "0.21.2"
-tokio = { version = "1.32.0", features = ["rt", "macros", "signal"] }
+tokio = { version = "1.32.0", features = ["io-util", "process", "rt", "macros", "signal"] }

src/log_watcher.rs

@@ -1,10 +1,8 @@
-use std::collections::HashMap;
-use std::io;
-use std::io::{Error, ErrorKind};
 use std::path::PathBuf;
+use std::process::Stdio;
 
-use linemux::{Line, MuxedLines};
+use tokio::io::{AsyncBufReadExt, BufReader};
-use tokio::sync::mpsc::UnboundedSender;
+use tokio::process::Command;
 
 use crate::ApacheMetrics;
 use crate::log_file_pattern::LogFilePath;
@@ -15,98 +13,126 @@ enum LogFileKind {
 	Error,
 }
 
-struct LogFileInfo<'a> {
+struct LogFile {
-	pub kind: LogFileKind,
+	pub path: PathBuf,
-	pub label: &'a String,
+	pub metadata: LogFileMetadata,
 }
 
-impl<'a> LogFileInfo<'a> {
+struct LogFileMetadata {
+	pub kind: LogFileKind,
+	pub label: String,
+}
+
+impl LogFileMetadata {
 	fn get_label_set(&self) -> [(&'static str, String); 1] {
 		[("file", self.label.clone())]
 	}
 }
 
-pub async fn watch_logs_task(access_log_files: Vec<LogFilePath>, error_log_files: Vec<LogFilePath>, metrics: ApacheMetrics, shutdown_send: UnboundedSender<()>) {
+pub async fn start_log_watcher(access_log_files: Vec<LogFilePath>, error_log_files: Vec<LogFilePath>, metrics: ApacheMetrics) -> bool {
-	if let Err(error) = watch_logs(access_log_files, error_log_files, metrics).await {
+	let mut watcher = LogWatcher::new();
-		println!("[LogWatcher] Error reading logs: {}", error);
+	
-		shutdown_send.send(()).unwrap();
+	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 LogWatcher<'a> {
+struct LogWatcher {
-	reader: MuxedLines,
+	files: Vec<LogFile>,
-	files: HashMap<PathBuf, LogFileInfo<'a>>,
 }
 
-impl<'a> LogWatcher<'a> {
+impl LogWatcher {
-	fn new() -> io::Result<LogWatcher<'a>> {
+	fn new() -> LogWatcher {
-		Ok(LogWatcher {
+		LogWatcher { files: Vec::new() }
-			reader: MuxedLines::new()?,
-			files: HashMap::new(),
-		})
 	}
 	
 	fn count_files_of_kind(&self, kind: LogFileKind) -> usize {
-		return self.files.values().filter(|info| info.kind == kind).count();
+		return self.files.iter().filter(|info| info.metadata.kind == kind).count();
 	}
 	
-	async fn add_file(&mut self, log_file: &'a LogFilePath, kind: LogFileKind) -> io::Result<()> {
+	fn add_file(&mut self, log_file: LogFilePath, kind: LogFileKind) {
-		let lookup_key = self.reader.add_file(&log_file.path).await?;
+		let path = log_file.path;
-		self.files.insert(lookup_key, LogFileInfo { kind, label: &log_file.label });
+		let label = log_file.label;
-		Ok(())
+		let metadata = LogFileMetadata { kind, label };
+		self.files.push(LogFile { path, metadata });
 	}
 	
-	async fn start_watching(&mut self, metrics: &ApacheMetrics) -> io::Result<()> {
+	async fn start(self, metrics: &ApacheMetrics) -> bool {
 		if self.files.is_empty() {
 			println!("[LogWatcher] No log files provided.");
-			return Err(Error::from(ErrorKind::Unsupported));
+			return false;
 		}
 		
 		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));
 		
-		for metadata in self.files.values() {
+		for file in self.files.into_iter() {
+			let metadata = file.metadata;
 			let label_set = metadata.get_label_set();
 			let _ = metrics.requests_total.get_or_create(&label_set);
 			let _ = metrics.errors_total.get_or_create(&label_set);
+			
+			let command = Command::new("tail")
+				.arg("-q") // Don't print file names.
+				.arg("-F") // Follow rotations.
+				.arg("-n").arg("0") // Start from end.
+				.arg(&file.path)
+				.env_clear()
+				.stdin(Stdio::null())
+				.stdout(Stdio::piped())
+				.stderr(Stdio::null())
+				.spawn();
+			
+			let mut process = match command {
+				Ok(process) => process,
+				Err(error) => {
+					println!("[LogWatcher] Error spawning tail process for file \"{}\": {}", file.path.to_string_lossy(), error);
+					return false;
+				}
+			};
+			
+			let stdout = match process.stdout.take() {
+				Some(stdout) => stdout,
+				None => {
+					println!("[LogWatcher] No output handle in tail process for file: {}", file.path.to_string_lossy());
+					return false;
+				}
+			};
+			
+			let mut output_reader = BufReader::new(stdout).lines();
+			let metrics = metrics.clone();
+			
+			tokio::spawn(async move {
+				loop {
+					match output_reader.next_line().await {
+						Ok(maybe_line) => match maybe_line {
+							Some(line) => handle_line(&metadata, line, &metrics),
+							None => break,
+						},
+						Err(e) => {
+							println!("[LogWatcher] Error reading from file \"{}\": {}", metadata.label, e);
+							break;
+						}
+					}
+				}
+			});
 		}
 		
-		loop {
+		true
-			if let Some(event) = self.reader.next_line().await? {
-				self.handle_line(event, metrics);
-			}
-		}
-	}
-	
-	fn handle_line(&mut self, event: Line, metrics: &ApacheMetrics) {
-		match self.files.get(event.source()) {
-			Some(metadata) => {
-				let label = metadata.label;
-				let (kind, family) = match metadata.kind {
-					LogFileKind::Access => ("access log", &metrics.requests_total),
-					LogFileKind::Error => ("error log", &metrics.errors_total),
-				};
-				
-				println!("[LogWatcher] Received {} line from \"{}\": {}", kind, label, event.line());
-				family.get_or_create(&metadata.get_label_set()).inc();
-			}
-			None => {
-				println!("[LogWatcher] Received line from unknown file: {}", event.source().display());
-			}
-		}
 	}
 }
 
-async fn watch_logs(access_log_files: Vec<LogFilePath>, error_log_files: Vec<LogFilePath>, metrics: ApacheMetrics) -> io::Result<()> {
+fn handle_line(metadata: &LogFileMetadata, line: String, metrics: &ApacheMetrics) {
-	let mut watcher = LogWatcher::new()?;
+	let (kind, family) = match metadata.kind {
+		LogFileKind::Access => ("access log", &metrics.requests_total),
+		LogFileKind::Error => ("error log", &metrics.errors_total),
+	};
 	
-	for log_file in &access_log_files {
+	println!("[LogWatcher] Received {} line from \"{}\": {}", kind, metadata.label, line);
-		watcher.add_file(log_file, LogFileKind::Access).await?;
+	family.get_or_create(&metadata.get_label_set()).inc();
-	}
-	
-	for log_file in &error_log_files {
-		watcher.add_file(log_file, LogFileKind::Error).await?;
-	}
-	
-	watcher.start_watching(&metrics).await?;
-	Ok(())
 }

src/main.rs

@@ -5,11 +5,10 @@ use std::str::FromStr;
 use std::sync::Mutex;
 
 use tokio::signal;
-use tokio::sync::mpsc;
 
 use crate::apache_metrics::ApacheMetrics;
 use crate::log_file_pattern::{LogFilePath, parse_log_file_pattern_from_env};
-use crate::log_watcher::watch_logs_task;
+use crate::log_watcher::start_log_watcher;
 use crate::web_server::WebServer;
 
 mod apache_metrics;
@@ -79,22 +78,21 @@ async fn main() -> ExitCode {
 	};
 	
 	let (metrics_registry, metrics) = ApacheMetrics::new();
-	let (shutdown_send, mut shutdown_recv) = mpsc::unbounded_channel();
 	
-	tokio::spawn(watch_logs_task(access_log_files, error_log_files, metrics.clone(), shutdown_send.clone()));
+	if !start_log_watcher(access_log_files, error_log_files, metrics).await {
-	tokio::spawn(server.serve(Mutex::new(metrics_registry)));
+		return ExitCode::FAILURE;
-	
-	drop(shutdown_send);
-	
-	tokio::select! {
-		_ = signal::ctrl_c() => {
-			println!("Received CTRL-C, shutting down...")
-		}
-		
-		_ = shutdown_recv.recv() => {
-			println!("Shutting down...");
-		}
 	}
 	
-	ExitCode::SUCCESS
+	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
+		}
+	}
 }