| 361 | } |
| 362 | |
| 363 | async fn try_redirect(input: &mut (impl AsyncRead + Unpin), to: String) -> Result<()> { |
| 364 | let dst_path = Path::new(&to); |
| 365 | let dst_path_buf = dst_path.to_path_buf(); |
| 366 | let (reopen_tx, mut reopen_rx) = mpsc::channel(1); |
| 367 | |
| 368 | // Set up file system watcher for logrotate detection |
| 369 | let mut watcher = |
| 370 | notify::recommended_watcher(move |res: Result<notify::Event, notify::Error>| { |
| 371 | if let Ok(event) = res { |
| 372 | // Check if the event affects our specific file |
| 373 | if (event.kind.is_remove() || event.kind.is_modify()) |
| 374 | && event.paths.iter().any(|p| p == &dst_path_buf) |
| 375 | { |
| 376 | let _ = reopen_tx.blocking_send(()); |
| 377 | } |
| 378 | } |
| 379 | })?; |
| 380 | // Watch the log file's parent directory |
| 381 | watcher.watch( |
| 382 | dst_path.parent().unwrap_or(Path::new(".")), |
| 383 | RecursiveMode::NonRecursive, |
| 384 | )?; |
| 385 | |
| 386 | let mut buffer = [0u8; 8192]; |
| 387 | loop { |
| 388 | // Open or reopen the log file in append mode |
| 389 | let mut file = fs::OpenOptions::new() |
| 390 | .create(true) |
| 391 | .append(true) |
| 392 | .open(dst_path)?; |
| 393 | |
| 394 | loop { |
| 395 | tokio::select! { |
| 396 | // Handle reopening signal |
| 397 | _ = reopen_rx.recv() => { |
| 398 | // Sync file to ensure all data is written |
| 399 | if let Err(e) = file.sync_all() { |
| 400 | error!("Failed to sync log file: {e}"); |
| 401 | break; |
| 402 | } |
| 403 | break; // Break inner loop to reopen file |
| 404 | } |
| 405 | // Read and write data |
| 406 | result = input.read(&mut buffer) => { |
| 407 | match result { |
| 408 | Ok(0) => return Ok(()), // EOF |
| 409 | Ok(n) => { |
| 410 | if let Err(e) = file.write_all(&buffer[..n]) { |
| 411 | error!("Failed to write to log file: {e}"); |
| 412 | break; |
| 413 | } |
| 414 | } |
| 415 | Err(e) => { |
| 416 | error!("Failed to read from process: {e}"); |
| 417 | return Err(e.into()); |
| 418 | } |
| 419 | } |
| 420 | } |