Skip to content

Implement page fetching or logs: manual cursor, plus automatic fetchi… #720

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
pieterbos wants to merge 1 commit into from
Closed

Implement page fetching or logs: manual cursor, plus automatic fetchi… #720

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
247 changes: 227 additions & 20 deletions src/commands/logs.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@ pub struct SearchArgs {
pub from: String,
pub to: String,
pub limit: i32,
pub cursor: Option<String>,
pub pages: u32,
pub sort: String,
pub storage: Option<String>,
pub index: Vec<String>,
Expand Down Expand Up @@ -204,16 +206,49 @@ fn parse_logs_sort(sort: &str) -> LogsSort {
}
}

fn append_log_page(aggregated: &mut Option<serde_json::Value>, page: serde_json::Value) {
if let Some(aggregated) = aggregated {
if let Some(page_data) = page.get("data").and_then(|value| value.as_array()) {
if let Some(data) = aggregated
.get_mut("data")
.and_then(|value| value.as_array_mut())
{
data.extend(page_data.iter().cloned());
} else {
aggregated["data"] = serde_json::Value::Array(page_data.to_vec());
}
}
if let Some(meta) = page.get("meta") {
aggregated["meta"] = meta.clone();
}
} else {
*aggregated = Some(page);
}
}

fn next_log_cursor(response: &serde_json::Value) -> Option<String> {
response
.pointer("/meta/page/after")
.and_then(|value| value.as_str())
.filter(|cursor| !cursor.is_empty())
.map(str::to_owned)
}

pub async fn search(cfg: &Config, args: SearchArgs) -> Result<()> {
let SearchArgs {
query,
from,
to,
limit,
cursor,
pages,
sort,
storage,
index,
} = args;
if pages == 0 {
bail!("--pages must be at least 1");
}
let api = crate::make_api!(LogsAPI, cfg);

let from_ms = util_ext::parse_time_to_unix_millis(&from)?;
Expand All @@ -232,39 +267,81 @@ pub async fn search(cfg: &Config, args: SearchArgs) -> Result<()> {
filter = filter.storage_tier(tier);
}

let body = LogsListRequest::new()
.filter(filter)
.page(LogsListRequestPage::new().limit(limit))
.sort(parse_logs_sort(&sort));
let mut request_cursor = cursor;
let mut response: Option<serde_json::Value> = None;
let mut next_cursor = None;

for _ in 0..pages {
let requested_cursor = request_cursor.clone();
let mut page = LogsListRequestPage::new().limit(limit);
if let Some(cursor) = requested_cursor.clone() {
page = page.cursor(cursor);
}

let params = ListLogsOptionalParams::default().body(body);
let body = LogsListRequest::new()
.filter(filter.clone())
.page(page)
.sort(parse_logs_sort(&sort));
let params = ListLogsOptionalParams::default().body(body);
let resp = api
.list_logs(params)
.await
.map_err(|e| anyhow::anyhow!("failed to search logs: {:?}", e))?;
let page_response = serde_json::to_value(&resp)?;
let has_results = page_response
.get("data")
.and_then(|data| data.as_array())
.is_some_and(|data| !data.is_empty());
let returned_cursor = next_log_cursor(&page_response);
append_log_page(&mut response, page_response);
next_cursor = returned_cursor.clone();

if !has_results {
next_cursor = None;
break;
}

let resp = api
.list_logs(params)
.await
.map_err(|e| anyhow::anyhow!("failed to search logs: {:?}", e))?;
match returned_cursor {
Some(after) if requested_cursor.as_deref() != Some(after.as_str()) => {
request_cursor = Some(after);
}
_ => {
next_cursor = None;
break;
}
}
}

let response = response.unwrap_or_else(|| serde_json::json!({}));

let meta = if cfg.agent_mode {
let count = resp.data.as_ref().map(|d| d.len());
let truncated = count.is_some_and(|c| c as i32 >= limit);
let count = response
.get("data")
.and_then(|data| data.as_array())
.map(|data| data.len());
let truncated = next_cursor.is_some() || count.is_some_and(|c| c as i32 >= limit);
Some(formatter::Metadata {
count,
truncated,
command: Some("logs search".into()),
next_action: if truncated {
Some(format!(
"Results may be truncated at {limit}. Use --limit={} or narrow the --query",
limit + 1
))
} else {
None
},
next_action: next_cursor
.map(|cursor| format!("More results available. Use --cursor=\"{cursor}\" to retrieve the next page."))
.or_else(|| {
if truncated {
Some(format!(
"Results may be truncated at {limit}. Use --limit={} or narrow the --query",
limit + 1
))
} else {
None
}
}),
})
} else {
None
};
formatter::format_and_print(
&resp,
&response,
&cfg.output_format,
cfg.agent_mode,
meta.as_ref(),
Expand Down Expand Up @@ -459,12 +536,41 @@ mod tests {
from: "1h".into(),
to: "now".into(),
limit: 10,
cursor: None,
pages: 1,
sort: "-timestamp".into(),
storage,
index,
}
}

#[test]
fn test_append_log_page_combines_data_and_keeps_latest_meta() {
let mut response = None;
append_log_page(
&mut response,
serde_json::json!({
"data": [{"id": "log-1"}],
"meta": {"page": {"after": "cursor-2"}}
}),
);
append_log_page(
&mut response,
serde_json::json!({
"data": [{"id": "log-2"}],
"meta": {"page": {}}
}),
);

assert_eq!(
response.unwrap(),
serde_json::json!({
"data": [{"id": "log-1"}, {"id": "log-2"}],
"meta": {"page": {}}
})
);
}

#[test]
fn test_normalize_storage_tier_alias() {
let tier = normalize_storage_tier(Some("online_archives".into())).unwrap();
Expand Down Expand Up @@ -817,6 +923,107 @@ mod tests {
cleanup_env();
}

#[tokio::test]
async fn test_logs_search_supports_all_output_formats() {
let _lock = lock_env().await;
let mut server = mockito::Server::new_async().await;
let mut cfg = test_config(&server.url());
let _mock = mock_any(
&mut server,
"POST",
r#"{"data":[{"id":"log-1","attributes":{"message":"error"}}],"meta":{"page":{}}}"#,
)
.await;

for format in [
OutputFormat::Json,
OutputFormat::Yaml,
OutputFormat::Table,
OutputFormat::Csv,
OutputFormat::Tsv,
] {
cfg.output_format = format.clone();
let result = super::search(&cfg, search_args("status:error", None, vec![])).await;
assert!(
result.is_ok(),
"logs search failed for {format}: {:?}",
result.err()
);
}
cleanup_env();
}

#[tokio::test]
async fn test_logs_search_with_cursor() {
let _lock = lock_env().await;
let mut server = mockito::Server::new_async().await;
let cfg = test_config(&server.url());
let _mock = server
.mock("POST", mockito::Matcher::Any)
.match_query(mockito::Matcher::Any)
.match_body(mockito::Matcher::Regex(
r#""cursor":"cursor-abc""#.to_string(),
))
.with_status(200)
.with_header("content-type", "application/json")
.with_body(r#"{"data": [], "meta": {"page": {}}}"#)
.create_async()
.await;

let mut args = search_args("status:error", None, vec![]);
args.cursor = Some("cursor-abc".into());
let result = super::search(&cfg, args).await;
assert!(
result.is_ok(),
"logs search with cursor failed: {:?}",
result.err()
);
cleanup_env();
}

#[tokio::test]
async fn test_logs_search_fetches_requested_pages() {
let _lock = lock_env().await;
let mut server = mockito::Server::new_async().await;
let cfg = test_config(&server.url());
let first_page = server
.mock("POST", mockito::Matcher::Any)
.match_query(mockito::Matcher::Any)
.match_body(mockito::Matcher::Regex(
r#""page":\{"limit":10\}"#.to_string(),
))
.with_status(200)
.with_header("content-type", "application/json")
.with_body(r#"{"data":[{"id":"log-1"}],"meta":{"page":{"after":"cursor-2"}}}"#)
.expect(1)
.create_async()
.await;
let second_page = server
.mock("POST", mockito::Matcher::Any)
.match_query(mockito::Matcher::Any)
.match_body(mockito::Matcher::Regex(
r#""cursor":"cursor-2""#.to_string(),
))
.with_status(200)
.with_header("content-type", "application/json")
.with_body(r#"{"data":[{"id":"log-2"}],"meta":{"page":{"after":"cursor-3"}}}"#)
.expect(1)
.create_async()
.await;

let mut args = search_args("status:error", None, vec![]);
args.pages = 2;
let result = super::search(&cfg, args).await;
assert!(
result.is_ok(),
"logs search pagination failed: {:?}",
result.err()
);
first_page.assert_async().await;
second_page.assert_async().await;
cleanup_env();
}

#[tokio::test]
async fn test_logs_search_with_indexes() {
let _lock = lock_env().await;
Expand Down
18 changes: 18 additions & 0 deletions src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3146,6 +3146,16 @@ enum LogActions {
to: String,
#[arg(long, default_value_t = 50, help = "Maximum number of logs (1-1000)")]
limit: i32,
#[arg(long, help = "Pagination cursor from a previous response")]
cursor: Option<String>,
#[arg(
long = "max-pages",
visible_alias = "pages",
default_value_t = 1,
value_parser = clap::value_parser!(u32).range(1..),
help = "Maximum pages to fetch automatically (default: 1)"
)]
pages: u32,
#[arg(long, help = "Sort order: asc or desc", default_value = "desc")]
sort: String,
#[arg(
Expand Down Expand Up @@ -12411,6 +12421,8 @@ async fn main_inner() -> anyhow::Result<()> {
from,
to,
limit,
cursor,
pages,
sort,
index,
storage,
Expand All @@ -12422,6 +12434,8 @@ async fn main_inner() -> anyhow::Result<()> {
from,
to,
limit,
cursor,
pages,
sort,
storage,
index,
Expand All @@ -12445,6 +12459,8 @@ async fn main_inner() -> anyhow::Result<()> {
from,
to,
limit,
cursor: None,
pages: 1,
sort,
storage,
index,
Expand All @@ -12469,6 +12485,8 @@ async fn main_inner() -> anyhow::Result<()> {
from,
to,
limit,
cursor: None,
pages: 1,
sort,
storage,
index,
Expand Down
Loading