Skip to content
Snippets Groups Projects
Commit 94b805de authored by Jason Volk's avatar Jason Volk
Browse files

generalize log capture to all admin commands; simplify handler


Signed-off-by: default avatarJason Volk <jason@zemos.net>
parent eded585f
No related branches found
No related tags found
1 merge request!545Admin command tracing capture
use service::Services; use std::time::SystemTime;
use conduit_service::Services;
pub(crate) struct Command<'a> { pub(crate) struct Command<'a> {
pub(crate) services: &'a Services, pub(crate) services: &'a Services,
pub(crate) body: &'a [&'a str], pub(crate) body: &'a [&'a str],
pub(crate) timer: SystemTime,
} }
use std::{ use std::{
collections::{BTreeMap, HashMap}, collections::{BTreeMap, HashMap},
fmt::Write, fmt::Write,
sync::{Arc, Mutex}, sync::Arc,
time::{Instant, SystemTime}, time::{Instant, SystemTime},
}; };
use api::client::validate_and_add_event_id; use api::client::validate_and_add_event_id;
use conduit::{ use conduit::{debug, debug_error, err, info, trace, utils, warn, Error, PduEvent, Result};
debug, debug_error, err, info, log,
log::{capture, Capture},
utils, warn, Error, PduEvent, Result,
};
use ruma::{ use ruma::{
api::{client::error::ErrorKind, federation::event::get_room_state}, api::{client::error::ErrorKind, federation::event::get_room_state},
events::room::message::RoomMessageEventContent, events::room::message::RoomMessageEventContent,
...@@ -717,30 +713,14 @@ pub(super) async fn resolve_true_destination( ...@@ -717,30 +713,14 @@ pub(super) async fn resolve_true_destination(
)); ));
} }
let filter: &capture::Filter = &|data| {
data.level() <= log::Level::DEBUG
&& data.mod_name().starts_with("conduit")
&& matches!(data.span_name(), "actual" | "well-known" | "srv")
};
let state = &self.services.server.log.capture;
let logs = Arc::new(Mutex::new(String::new()));
let capture = Capture::new(state, Some(filter), capture::fmt_markdown(logs.clone()));
let capture_scope = capture.start();
let actual = self let actual = self
.services .services
.resolver .resolver
.resolve_actual_dest(&server_name, !no_cache) .resolve_actual_dest(&server_name, !no_cache)
.await?; .await?;
drop(capture_scope);
let msg = format!( let msg = format!("Destination: {}\nHostname URI: {}", actual.dest, actual.host,);
"{}\nDestination: {}\nHostname URI: {}",
logs.lock().expect("locked"),
actual.dest,
actual.host,
);
Ok(RoomMessageEventContent::text_markdown(msg)) Ok(RoomMessageEventContent::text_markdown(msg))
} }
......
use std::{panic::AssertUnwindSafe, sync::Arc, time::Instant}; use std::{
panic::AssertUnwindSafe,
sync::{Arc, Mutex},
time::SystemTime,
};
use clap::{CommandFactory, Parser}; use clap::{CommandFactory, Parser};
use conduit::{checked, error, trace, utils::string::common_prefix, Error, Result}; use conduit::{
debug, error,
log::{
capture,
capture::Capture,
fmt::{markdown_table, markdown_table_head},
},
trace,
utils::string::{collect_stream, common_prefix},
Error, Result,
};
use futures_util::future::FutureExt; use futures_util::future::FutureExt;
use ruma::{ use ruma::{
events::{ events::{
...@@ -14,6 +28,7 @@ ...@@ -14,6 +28,7 @@
admin::{CommandInput, CommandOutput, HandlerFuture, HandlerResult}, admin::{CommandInput, CommandOutput, HandlerFuture, HandlerResult},
Services, Services,
}; };
use tracing::Level;
use crate::{admin, admin::AdminCommand, Command}; use crate::{admin, admin::AdminCommand, Command};
...@@ -34,10 +49,21 @@ async fn handle_command(services: Arc<Services>, command: CommandInput) -> Handl ...@@ -34,10 +49,21 @@ async fn handle_command(services: Arc<Services>, command: CommandInput) -> Handl
.or_else(|error| handle_panic(&error, command)) .or_else(|error| handle_panic(&error, command))
} }
async fn process_command(services: Arc<Services>, command: &CommandInput) -> CommandOutput { async fn process_command(services: Arc<Services>, input: &CommandInput) -> CommandOutput {
process(services, &command.command) let (command, args, body) = match parse(&services, input) {
Err(error) => return error,
Ok(parsed) => parsed,
};
let context = Command {
services: &services,
body: &body,
timer: SystemTime::now(),
};
process(&context, command, &args)
.await .await
.and_then(|content| reply(content, command.reply_id.clone())) .and_then(|content| reply(content, input.reply_id.clone()))
} }
fn handle_panic(error: &Error, command: CommandInput) -> HandlerResult { fn handle_panic(error: &Error, command: CommandInput) -> HandlerResult {
...@@ -59,69 +85,61 @@ fn reply(mut content: RoomMessageEventContent, reply_id: Option<OwnedEventId>) - ...@@ -59,69 +85,61 @@ fn reply(mut content: RoomMessageEventContent, reply_id: Option<OwnedEventId>) -
} }
// Parse and process a message from the admin room // Parse and process a message from the admin room
async fn process(services: Arc<Services>, msg: &str) -> CommandOutput { async fn process(context: &Command<'_>, command: AdminCommand, args: &[String]) -> CommandOutput {
let lines = msg.lines().filter(|l| !l.trim().is_empty()); let filter: &capture::Filter =
let command = lines &|data| data.level() <= Level::DEBUG && data.mod_name().starts_with("conduit") && data.scope.contains(&"admin");
.clone() let logs = Arc::new(Mutex::new(
.next() collect_stream(|s| markdown_table_head(s)).expect("markdown table header"),
.expect("each string has at least one line"); ));
let (parsed, body) = match parse_command(command) {
Ok(parsed) => parsed, let capture = Capture::new(
Err(error) => { &context.services.server.log.capture,
let server_name = services.globals.server_name(); Some(filter),
let message = error.replace("server.name", server_name.as_str()); capture::fmt(markdown_table, logs.clone()),
return Some(RoomMessageEventContent::notice_markdown(message)); );
},
let capture_scope = capture.start();
let result = Box::pin(admin::process(command, context)).await;
drop(capture_scope);
debug!(
ok = result.is_ok(),
elapsed = ?context.timer.elapsed(),
command = ?args,
"command processed"
);
let logs = logs.lock().expect("locked");
let output = match result {
Err(error) => format!("{logs}\nEncountered an error while handling the command:\n```\n{error:#?}\n```"),
Ok(reply) => format!("{logs}\n{}", reply.body()), //TODO: content is recreated to add logs
}; };
let body = parse_body(AdminCommand::command(), &body, lines.skip(1).collect()).expect("trailing body parsed"); Some(RoomMessageEventContent::notice_markdown(output))
let context = Command {
services: &services,
body: &body,
};
let timer = Instant::now();
let result = Box::pin(admin::process(parsed, &context)).await;
let elapsed = timer.elapsed();
conduit::debug!(?command, ok = result.is_ok(), "command processed in {elapsed:?}");
match result {
Ok(reply) => Some(reply),
Err(error) => Some(RoomMessageEventContent::notice_markdown(format!(
"Encountered an error while handling the command:\n```\n{error:#?}\n```"
))),
}
} }
// Parse chat messages from the admin room into an AdminCommand object // Parse chat messages from the admin room into an AdminCommand object
fn parse_command(command_line: &str) -> Result<(AdminCommand, Vec<String>), String> { fn parse<'a>(
let argv = parse_line(command_line); services: &Arc<Services>, input: &'a CommandInput,
let com = AdminCommand::try_parse_from(&argv).map_err(|error| error.to_string())?; ) -> Result<(AdminCommand, Vec<String>, Vec<&'a str>), CommandOutput> {
Ok((com, argv)) let lines = input.command.lines().filter(|line| !line.trim().is_empty());
} let command_line = lines.clone().next().expect("command missing first line");
let body = lines.skip(1).collect();
fn parse_body<'a>(mut cmd: clap::Command, body: &'a [String], lines: Vec<&'a str>) -> Result<Vec<&'a str>> { match parse_command(command_line) {
let mut start = 1; Ok((command, args)) => Ok((command, args, body)),
'token: for token in body.iter().skip(1) { Err(error) => {
let cmd_ = cmd.clone(); let message = error
for sub in cmd_.get_subcommands() { .to_string()
if sub.get_name() == *token { .replace("server.name", services.globals.server_name().as_str());
start = checked!(start + 1)?; Err(Some(RoomMessageEventContent::notice_markdown(message)))
cmd = sub.clone(); },
continue 'token;
}
}
// positional arguments have to be skipped too
let num_posargs = cmd_.get_positionals().count();
start = checked!(start + num_posargs)?;
break;
} }
}
Ok(body fn parse_command(line: &str) -> Result<(AdminCommand, Vec<String>)> {
.iter() let argv = parse_line(line);
.skip(start) let command = AdminCommand::try_parse_from(&argv)?;
.map(String::as_str) Ok((command, argv))
.chain(lines)
.collect::<Vec<&'a str>>())
} }
fn complete_command(mut cmd: clap::Command, line: &str) -> String { fn complete_command(mut cmd: clap::Command, line: &str) -> String {
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment