Compare commits
5
Commits
4158c63086
...
fc8de1a5b0
| Author | SHA1 | Date | |
|---|---|---|---|
 Hugh Kaznowski
|
fc8de1a5b0 | ||
|
Hugh KaznowskiandClaude Sonnet 5
|
c77404ba4e | ||
|
Hugh KaznowskiandClaude Sonnet 5
|
de9ba6f01b | ||
|
Hugh KaznowskiandClaude Sonnet 5
|
4ace654019 | ||
|
Hugh Kaznowski
|
a03ec1fb93 |
Generated
+291
@@ -70,6 +70,12 @@ dependencies = [
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "anyhow"
|
||||
version = "1.0.104"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470"
|
||||
|
||||
[[package]]
|
||||
name = "atomic-waker"
|
||||
version = "1.1.2"
|
||||
@@ -200,6 +206,7 @@ dependencies = [
|
||||
"iana-time-zone",
|
||||
"js-sys",
|
||||
"num-traits",
|
||||
"serde",
|
||||
"wasm-bindgen",
|
||||
"windows-link",
|
||||
]
|
||||
@@ -265,6 +272,40 @@ dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "darling"
|
||||
version = "0.24.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ed17f5901b6630b993ca003def43f2f8ef4014fc13b047b57aad617ff32bc2ec"
|
||||
dependencies = [
|
||||
"darling_core",
|
||||
"darling_macro",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "darling_core"
|
||||
version = "0.24.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6837e2cf7485aaae18f86181d2f0e9a7ed297a025e220aeabf63fdebd3a2ddff"
|
||||
dependencies = [
|
||||
"ident_case",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"strsim",
|
||||
"syn 3.0.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "darling_macro"
|
||||
version = "0.24.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2ac7135c3ef02b2f7833bbeb1be5ba7f966dcde8a87c6b87f65a778d71a02785"
|
||||
dependencies = [
|
||||
"darling_core",
|
||||
"quote",
|
||||
"syn 3.0.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "directories"
|
||||
version = "6.0.0"
|
||||
@@ -297,6 +338,18 @@ dependencies = [
|
||||
"syn 3.0.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dyn-clone"
|
||||
version = "1.0.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
|
||||
|
||||
[[package]]
|
||||
name = "equivalent"
|
||||
version = "1.0.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
|
||||
|
||||
[[package]]
|
||||
name = "errno"
|
||||
version = "0.3.14"
|
||||
@@ -355,6 +408,21 @@ dependencies = [
|
||||
"walkdir",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures"
|
||||
version = "0.3.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3"
|
||||
dependencies = [
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
"futures-executor",
|
||||
"futures-io",
|
||||
"futures-sink",
|
||||
"futures-task",
|
||||
"futures-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-channel"
|
||||
version = "0.3.34"
|
||||
@@ -362,6 +430,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"futures-sink",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -370,6 +439,40 @@ version = "0.3.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"
|
||||
|
||||
[[package]]
|
||||
name = "futures-executor"
|
||||
version = "0.3.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"futures-task",
|
||||
"futures-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-io"
|
||||
version = "0.3.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed"
|
||||
|
||||
[[package]]
|
||||
name = "futures-macro"
|
||||
version = "0.3.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-sink"
|
||||
version = "0.3.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d"
|
||||
|
||||
[[package]]
|
||||
name = "futures-task"
|
||||
version = "0.3.34"
|
||||
@@ -382,8 +485,13 @@ version = "0.3.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc"
|
||||
dependencies = [
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
"futures-io",
|
||||
"futures-macro",
|
||||
"futures-sink",
|
||||
"futures-task",
|
||||
"memchr",
|
||||
"pin-project-lite",
|
||||
"slab",
|
||||
]
|
||||
@@ -415,6 +523,12 @@ dependencies = [
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hashbrown"
|
||||
version = "0.17.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
|
||||
|
||||
[[package]]
|
||||
name = "heck"
|
||||
version = "0.5.0"
|
||||
@@ -639,6 +753,12 @@ dependencies = [
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ident_case"
|
||||
version = "1.0.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39"
|
||||
|
||||
[[package]]
|
||||
name = "idna"
|
||||
version = "1.1.0"
|
||||
@@ -679,6 +799,18 @@ dependencies = [
|
||||
"quote",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "indexmap"
|
||||
version = "2.14.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
|
||||
dependencies = [
|
||||
"equivalent",
|
||||
"hashbrown",
|
||||
"serde",
|
||||
"serde_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ipnet"
|
||||
version = "2.12.1"
|
||||
@@ -832,6 +964,7 @@ checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d"
|
||||
name = "ottyl"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"axum",
|
||||
"chrono",
|
||||
"clap",
|
||||
@@ -843,8 +976,10 @@ dependencies = [
|
||||
"nom",
|
||||
"pest",
|
||||
"pest_derive",
|
||||
"petname",
|
||||
"regex",
|
||||
"reqwest",
|
||||
"rmcp",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tempfile",
|
||||
@@ -875,6 +1010,12 @@ dependencies = [
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pastey"
|
||||
version = "0.2.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2ee67f1008b1ba2321834326597b8e186293b049a023cdef258527550b9935b4"
|
||||
|
||||
[[package]]
|
||||
name = "percent-encoding"
|
||||
version = "2.3.2"
|
||||
@@ -923,6 +1064,27 @@ dependencies = [
|
||||
"pest",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "petname"
|
||||
version = "3.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bfb4653a82ec48e6d324c788597aadf46449bfc7b8e76d2a03b513a944325c88"
|
||||
dependencies = [
|
||||
"petname-macros",
|
||||
"rand",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "petname-macros"
|
||||
version = "3.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "78c406fb544296f598ccf69b82bcf1b12440c8f23ebf75946105f86304f16eb2"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "phf"
|
||||
version = "0.11.3"
|
||||
@@ -1082,6 +1244,26 @@ dependencies = [
|
||||
"thiserror",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ref-cast"
|
||||
version = "1.0.27"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7e440fb4e4b4147295338efb76001ab9e4efc0e5839df2c47fc5ac2381d365c3"
|
||||
dependencies = [
|
||||
"ref-cast-impl",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ref-cast-impl"
|
||||
version = "1.0.27"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "regex"
|
||||
version = "1.13.1"
|
||||
@@ -1163,6 +1345,41 @@ dependencies = [
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rmcp"
|
||||
version = "3.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "42b6914fac0be956fe704a38239c3f44a9f841d1b06a5713d2f638065593f5b5"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"futures",
|
||||
"indexmap",
|
||||
"pastey",
|
||||
"pin-project-lite",
|
||||
"rmcp-macros",
|
||||
"schemars",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"thiserror",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"tracing",
|
||||
"uuid",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rmcp-macros"
|
||||
version = "3.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cdf1c49bd4d52014b94db0877410db273c2008f01628b0252a2e9460ad9b7fda"
|
||||
dependencies = [
|
||||
"darling",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"serde_json",
|
||||
"syn 3.0.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustc-hash"
|
||||
version = "2.1.3"
|
||||
@@ -1238,6 +1455,32 @@ dependencies = [
|
||||
"winapi-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "schemars"
|
||||
version = "1.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"dyn-clone",
|
||||
"ref-cast",
|
||||
"schemars_derive",
|
||||
"serde",
|
||||
"serde_json",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "schemars_derive"
|
||||
version = "1.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d98c67716b46af2f0b8cf752abc930f6f9aecfbf671ecfb531db8a31dbe4e2ba"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"serde_derive_internals",
|
||||
"syn 3.0.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "scopeguard"
|
||||
version = "1.2.0"
|
||||
@@ -1274,6 +1517,17 @@ dependencies = [
|
||||
"syn 3.0.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_derive_internals"
|
||||
version = "0.30.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f852137cce035d6a4df67ccce505ff6b3e9fd3a10e3e52b24dc71e650bb1a9bd"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_json"
|
||||
version = "1.0.151"
|
||||
@@ -1510,6 +1764,20 @@ dependencies = [
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-util"
|
||||
version = "0.7.19"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures-core",
|
||||
"futures-sink",
|
||||
"libc",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tower"
|
||||
version = "0.5.3"
|
||||
@@ -1564,9 +1832,21 @@ checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
|
||||
dependencies = [
|
||||
"log",
|
||||
"pin-project-lite",
|
||||
"tracing-attributes",
|
||||
"tracing-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tracing-attributes"
|
||||
version = "0.1.31"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.119",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tracing-core"
|
||||
version = "0.1.36"
|
||||
@@ -1624,6 +1904,17 @@ version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
|
||||
|
||||
[[package]]
|
||||
name = "uuid"
|
||||
version = "1.26.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812"
|
||||
dependencies = [
|
||||
"getrandom 0.4.3",
|
||||
"js-sys",
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "walkdir"
|
||||
version = "2.5.0"
|
||||
|
||||
+4
-1
@@ -18,7 +18,10 @@ humantime = "2"
|
||||
directories = "6.0.0"
|
||||
fs-embed = "0.2.4"
|
||||
include_dir = "0.7.4"
|
||||
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
|
||||
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] }
|
||||
petname = { version = "3.2.0", default-features = false, features = ["default-rng", "default-words"] }
|
||||
rmcp = { version = "3.2.0", default-features = false, features = ["server", "macros", "transport-io"] }
|
||||
anyhow = "1"
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile = "3.27.0"
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
---
|
||||
name: ottyl
|
||||
description: Query and configure a running ottyl observability instance (logs, params, metrics, dashboards) via its MCP server.
|
||||
---
|
||||
|
||||
# ottyl
|
||||
|
||||
ottyl turns a piped command's stdout into a searchable log timeline plus ad-hoc metrics and
|
||||
dashboards, served from a single process. This doc covers how an agent connects to and drives a
|
||||
*running* instance.
|
||||
|
||||
## Find a session
|
||||
|
||||
Each `ottyl` process registers itself under a name (random, e.g. `comic-gazelle`, unless started
|
||||
with `--name <name>`). List known sessions:
|
||||
|
||||
```
|
||||
ottyl sessions
|
||||
# comic-gazelle port=51423 pid=8421
|
||||
```
|
||||
|
||||
## Connect an MCP client
|
||||
|
||||
Start the MCP server (a separate short-lived process) pointed at that session, and register it
|
||||
with your MCP client (e.g. Claude Code's `.mcp.json`):
|
||||
|
||||
```
|
||||
ottyl mcp --session comic-gazelle
|
||||
# or, if you already know the port:
|
||||
ottyl mcp --port 51423
|
||||
```
|
||||
|
||||
This exposes six tools, each a thin proxy onto the target instance's HTTP API.
|
||||
|
||||
## Tools
|
||||
|
||||
### `query_logs`
|
||||
Count or read log lines matching a FilterBar query.
|
||||
|
||||
```json
|
||||
{ "filter": "regex(\"error\")", "count_only": true }
|
||||
```
|
||||
|
||||
### `declare_param`
|
||||
Extract a named value from matching log lines.
|
||||
|
||||
```json
|
||||
{
|
||||
"name": "status_code",
|
||||
"message_match_regex": "status=(\\d+)",
|
||||
"capture_group": "1",
|
||||
"kind": "number"
|
||||
}
|
||||
```
|
||||
|
||||
### `declare_metric`
|
||||
Bucket event counts/values by label, derived from matching log lines.
|
||||
|
||||
```json
|
||||
{
|
||||
"name": "errors_total",
|
||||
"message_match_regex": "level=error",
|
||||
"label_params": ["status_code"]
|
||||
}
|
||||
```
|
||||
|
||||
### `create_dashboard`
|
||||
Create an empty dashboard in a profile (defaults to the active profile).
|
||||
|
||||
```json
|
||||
{ "name": "Errors" }
|
||||
```
|
||||
|
||||
### `add_panel`
|
||||
Add a panel to a dashboard, or update one in place by `panel_id`.
|
||||
|
||||
```json
|
||||
{
|
||||
"dashboard_id": "<id from create_dashboard>",
|
||||
"title": "Errors over time",
|
||||
"promql": "rate(errors_total[5m])",
|
||||
"chart_type": "line"
|
||||
}
|
||||
```
|
||||
|
||||
### `get_dashboard`
|
||||
Read a dashboard's current panels.
|
||||
|
||||
```json
|
||||
{ "dashboard_id": "<id>" }
|
||||
```
|
||||
@@ -26,6 +26,10 @@ use tokio::sync::oneshot;
|
||||
|
||||
static WEBAPP_DIST: EmbedDir = fs_embed::fs_embed!("webapp/dist");
|
||||
|
||||
/// Embedded so an agent that has only discovered a port (no filesystem access into the ottyl
|
||||
/// repo/binary) can still fetch it — see `SKILL.md` at the repo root and ticket 069.
|
||||
const SKILL_MD: &str = include_str!("../SKILL.md");
|
||||
|
||||
#[derive(Clone)]
|
||||
struct ApiState {
|
||||
engine: EngineHandle,
|
||||
@@ -158,6 +162,7 @@ pub fn router(engine: EngineHandle, settings: SettingsHandle, settings_path: Pat
|
||||
};
|
||||
|
||||
Router::new()
|
||||
.route("/SKILL.md", get(get_skill_md))
|
||||
.route("/api/logs", get(get_logs))
|
||||
.route("/api/timeline", get(get_timeline))
|
||||
.route("/api/stats", get(get_stats))
|
||||
@@ -194,6 +199,10 @@ pub fn router(engine: EngineHandle, settings: SettingsHandle, settings_path: Pat
|
||||
.fallback(serve_webapp)
|
||||
}
|
||||
|
||||
async fn get_skill_md() -> impl IntoResponse {
|
||||
([(header::CONTENT_TYPE, "text/markdown; charset=utf-8")], SKILL_MD)
|
||||
}
|
||||
|
||||
/// Falls back to `index.html` for unmatched paths so client-side routing works.
|
||||
async fn serve_webapp(uri: axum::http::Uri) -> Response {
|
||||
let path = uri.path().trim_start_matches('/');
|
||||
|
||||
+400
-82
@@ -164,6 +164,13 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
|
||||
// every currently-ingested log, kept in lockstep with `log_count` — incremented
|
||||
// here on ingest, decremented in `evict_excess_logs` when a log is dropped.
|
||||
let mut approx_bytes_total: u64 = 0;
|
||||
// Ticket 060/061: one entry per param name that's ever been used as a `sort_by`, populated
|
||||
// lazily on first sorted query and kept incrementally up to date on every ingest thereafter;
|
||||
// swept for staleness on ingest too (see `purge_stale_sort_helper_indexes`).
|
||||
let mut sort_helper_indexes: HashMap<String, SortHelperIndexEntry> = HashMap::new();
|
||||
// Ticket 059: the most recently seen `log stream` header line's column names, cached across
|
||||
// ingested lines since a `log stream` pipe only prints its header once, before any data line.
|
||||
let mut log_stream_header: Option<crate::logfmt::LogStreamHeader> = None;
|
||||
|
||||
while let Some(message) = rx.recv().await {
|
||||
match message {
|
||||
@@ -175,9 +182,11 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
|
||||
DataValue::LogLiteral { text: text.clone() },
|
||||
);
|
||||
log_count += 1;
|
||||
auto_declare_log_stream_params(&mut index, &text);
|
||||
auto_declare_log_stream_params(&mut index, &mut log_stream_header, &text);
|
||||
auto_declare_params(&mut index, &text);
|
||||
apply_all_metrics(&mut index, sequence, timestamp, &text);
|
||||
update_sort_helper_indexes_on_ingest(&mut sort_helper_indexes, &index, sequence, timestamp, &text);
|
||||
purge_stale_sort_helper_indexes(&mut sort_helper_indexes);
|
||||
evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total);
|
||||
|
||||
// Evaluate alert rules against just this line, then hand any fired rule off to delivery via tokio::spawn so it stays off the hot path.
|
||||
@@ -212,16 +221,26 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
|
||||
evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total);
|
||||
}
|
||||
EngineMessage::DeclareParam { request, respond_to } => {
|
||||
// A manual re-declaration can change the regex/kind under the same name — drop any
|
||||
// helper index built under the old declaration so a sorted query rebuilds it fresh
|
||||
// rather than serving values derived from the now-stale regex.
|
||||
sort_helper_indexes.remove(&request.name);
|
||||
let result = declare_param(&mut index, request);
|
||||
let _ = respond_to.send(result);
|
||||
}
|
||||
EngineMessage::RemoveParamDeclaration { name, respond_to } => {
|
||||
let existed = index.remove(&IndexKey::ParamDeclaration { name }).is_some();
|
||||
let existed = index.remove(&IndexKey::ParamDeclaration { name: name.clone() }).is_some();
|
||||
// A helper index built under the old declaration (or a same-named one re-declared
|
||||
// with a different regex/kind later) would otherwise keep serving stale derived
|
||||
// values until ticket 061's sweep happens to purge it — dropping it here instead
|
||||
// means the next sorted query for this param rebuilds it fresh.
|
||||
sort_helper_indexes.remove(&name);
|
||||
let _ = respond_to.send(existed);
|
||||
}
|
||||
EngineMessage::ClearParams { respond_to } => {
|
||||
let removed = clear_all_param_declarations(&mut index);
|
||||
let (reextracted, logs_considered) = reextract_params_from_retained_logs(&mut index);
|
||||
sort_helper_indexes.clear();
|
||||
let _ = respond_to.send(ClearParamsResult { removed, reextracted, logs_considered });
|
||||
}
|
||||
EngineMessage::DeclareMetric { request, respond_to } => {
|
||||
@@ -237,7 +256,7 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
|
||||
let _ = respond_to.send(existed);
|
||||
}
|
||||
EngineMessage::Query { query, respond_to } => {
|
||||
let response = handle_query(&index, query, approx_bytes_total);
|
||||
let response = handle_query(&index, query, approx_bytes_total, &mut sort_helper_indexes);
|
||||
let _ = respond_to.send(response);
|
||||
}
|
||||
}
|
||||
@@ -438,43 +457,68 @@ fn auto_declare_params(index: &mut BTreeMap<IndexKey, DataValue>, text: &str) {
|
||||
}
|
||||
}
|
||||
|
||||
/// Declares `log_type`/`process`/`pid` as standalone params the first time a `log stream`-shaped line
|
||||
/// is seen (ticket 050/054) — `Type`, the process name, and `PID` are low-cardinality, genuinely
|
||||
/// useful columns derivable directly from `log stream`'s fixed-column shape
|
||||
/// (`crate::logfmt::is_log_stream_line`), with none of the `=`-scanning ambiguity `auto_declare_params`
|
||||
/// has to guard against. Gated on that shape so ordinary logfmt/JSON/prose input never gets these three
|
||||
/// params. Unlike `auto_declare_params`, adding these is never *skipped* on account of
|
||||
/// `MAX_AUTO_DECLARED_PARAMS` — it can only ever add exactly these three names, once each — though
|
||||
/// since they're stored as `auto_derived` too, once present they do count against that shared ceiling,
|
||||
/// tightening (slightly) the budget left for generic `key=value` pairs. Like `auto_declare_params`,
|
||||
/// never overwrites an existing declaration (manual or auto) for any of the three names.
|
||||
fn auto_declare_log_stream_params(index: &mut BTreeMap<IndexKey, DataValue>, text: &str) {
|
||||
/// Declares one param per column of the most-recently-seen `log stream` header line, plus a trailing
|
||||
/// unlabelled `message` param for whatever text follows the last named column (ticket 059) — replacing
|
||||
/// the old fixed three (`log_type`/`process`/`pid`) with whatever columns the header actually names.
|
||||
/// `header` is threaded through by the caller (`run`'s ingest loop) rather than owned here: a
|
||||
/// `log stream` pipe prints its header once, before any data line, and this function only ever sees
|
||||
/// one ingested line at a time, so the header has to be cached across calls somewhere. A later header
|
||||
/// line (e.g. a fresh `log stream` invocation piped into the same run) simply replaces the cached one.
|
||||
/// Unlike `auto_declare_params`, adding these is never *skipped* on account of
|
||||
/// `MAX_AUTO_DECLARED_PARAMS` — though since they're stored as `auto_derived` too, once present they do
|
||||
/// count against that shared ceiling, tightening (slightly) the budget left for generic `key=value`
|
||||
/// pairs. Like `auto_declare_params`, never overwrites an existing declaration (manual or auto) for
|
||||
/// any of these names.
|
||||
fn auto_declare_log_stream_params(
|
||||
index: &mut BTreeMap<IndexKey, DataValue>,
|
||||
header: &mut Option<crate::logfmt::LogStreamHeader>,
|
||||
text: &str,
|
||||
) {
|
||||
if let Some(detected) = crate::logfmt::detect_log_stream_header(text) {
|
||||
*header = Some(detected);
|
||||
return;
|
||||
}
|
||||
let Some(header) = header.as_ref() else {
|
||||
return;
|
||||
};
|
||||
if !crate::logfmt::is_log_stream_line(text) {
|
||||
return;
|
||||
}
|
||||
let candidates: [(&str, &str, &str, MetricPrimitiveKind); 3] = [
|
||||
("log_type", crate::logfmt::LOG_STREAM_TYPE_PATTERN, "log_type", MetricPrimitiveKind::String),
|
||||
("process", crate::logfmt::LOG_STREAM_PROCESS_PATTERN, "process", MetricPrimitiveKind::String),
|
||||
("pid", crate::logfmt::LOG_STREAM_PID_PATTERN, "pid", MetricPrimitiveKind::Number),
|
||||
];
|
||||
for (name, pattern, group, kind) in candidates {
|
||||
let index_key = IndexKey::ParamDeclaration { name: name.to_string() };
|
||||
let row = crate::logfmt::split_log_stream_row(text, header);
|
||||
|
||||
for (i, column) in header.columns.iter().enumerate() {
|
||||
let Some(value) = row.get(i) else { continue };
|
||||
let index_key = IndexKey::ParamDeclaration { name: column.name.clone() };
|
||||
if index.contains_key(&index_key) {
|
||||
continue;
|
||||
}
|
||||
index.insert(
|
||||
index_key,
|
||||
DataValue::ParamDeclaration(ParamDeclaration {
|
||||
name: name.to_string(),
|
||||
message_match_regex: pattern.to_string(),
|
||||
capture_group: group.to_string(),
|
||||
name: column.name.clone(),
|
||||
message_match_regex: crate::logfmt::log_stream_column_pattern(header, i, &column.name),
|
||||
capture_group: column.name.clone(),
|
||||
json_field_path: None,
|
||||
kind,
|
||||
kind: infer_kind(value.trim()),
|
||||
elapsed_since_last: false,
|
||||
auto_derived: true,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) {
|
||||
index.entry(IndexKey::ParamDeclaration { name: "message".to_string() }).or_insert_with(|| {
|
||||
DataValue::ParamDeclaration(ParamDeclaration {
|
||||
name: "message".to_string(),
|
||||
message_match_regex: crate::logfmt::log_stream_message_pattern(header, "message"),
|
||||
capture_group: "message".to_string(),
|
||||
json_field_path: None,
|
||||
kind: infer_kind(message.trim()),
|
||||
elapsed_since_last: false,
|
||||
auto_derived: true,
|
||||
})
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// Ticket 052: removes every `IndexKey::ParamDeclaration` entry — manual and auto-derived alike.
|
||||
@@ -1285,6 +1329,103 @@ fn param_declarations_range(
|
||||
/// meaningful reading as a standalone logs-table column (its value depends on which metric's label
|
||||
/// set it's grouped under, which only exists once it's referenced by a metric's `label_params`/
|
||||
/// `value_param`).
|
||||
/// Ticket 060: key type for a sort helper index — one such index exists per param name that's ever
|
||||
/// been used as a `sort_by`. Declaration order matters for the derived `Ord`: every `WithParam` key
|
||||
/// sorts before every `WithoutParam` key, and within each variant ordering falls out of the tuple
|
||||
/// fields (value then sequence; sequence alone), which is exactly `sort_missing_fallback`'s "sorted
|
||||
/// lines first, then lines missing the param ordered by `ottyl_seq`" semantics — expressed as part of
|
||||
/// the index's own ordering instead of a separate partition-and-append step per query.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
|
||||
enum SortHelperKey {
|
||||
WithParam(MetricPrimitive, u64),
|
||||
WithoutParam(u64),
|
||||
}
|
||||
|
||||
/// One param's helper index plus when a query last actually read from it — ticket 061 sweeps entries
|
||||
/// whose `last_accessed` is stale. The map's value is that line's `Timestamp`, not `()`, purely so a
|
||||
/// query can reconstruct the exact `IndexKey::IngestedLog { sequence, timestamp }` primary-index key
|
||||
/// and look its text up in O(log n) instead of re-scanning `logs_range` to find it by sequence alone.
|
||||
struct SortHelperIndexEntry {
|
||||
index: BTreeMap<SortHelperKey, Timestamp>,
|
||||
last_accessed: std::time::Instant,
|
||||
}
|
||||
|
||||
/// How long a sort helper index may sit unread before ticket 061's sweep drops it. Not exposed via
|
||||
/// `Settings` (ticket 061 explicitly scopes that out) — a dropped index is always transparently
|
||||
/// rebuildable on the next query for that sort key, so this is purely a memory-bound knob, not a
|
||||
/// user-visible behavior change.
|
||||
const SORT_HELPER_INDEX_PURGE_AFTER: std::time::Duration = std::time::Duration::from_secs(300);
|
||||
|
||||
/// Derives a single named param's value from `text`, for helper-index maintenance where deriving
|
||||
/// every declared param (`derive_log_params`) per ingested line would be wasted work when only one
|
||||
/// sort key's helper index actually needs updating.
|
||||
fn derive_single_param(
|
||||
text: &str,
|
||||
declarations: &[(String, Regex, String, Option<String>, MetricPrimitiveKind)],
|
||||
param_name: &str,
|
||||
) -> Option<MetricPrimitive> {
|
||||
let (_, regex, capture_group, json_field_path, kind) =
|
||||
declarations.iter().find(|(name, ..)| name == param_name)?;
|
||||
if *kind == MetricPrimitiveKind::Boolean {
|
||||
return Some(MetricPrimitive::Boolean(regex.is_match(text)));
|
||||
}
|
||||
let captures = regex.captures(text)?;
|
||||
let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) {
|
||||
Some(field_path) => crate::data_value::extract_json_field(text, field_path)?,
|
||||
None => captures.name(capture_group)?.as_str().to_string(),
|
||||
};
|
||||
parse_metric_primitive(&raw, *kind)
|
||||
}
|
||||
|
||||
/// One-time full scan building a sort helper index for `sort_by` from every currently-ingested log —
|
||||
/// used both the first time a param is sorted by (no helper index exists yet) and to transparently
|
||||
/// rebuild one ticket 061's sweep purged.
|
||||
fn build_sort_helper_index(index: &BTreeMap<IndexKey, DataValue>, sort_by: &str) -> BTreeMap<SortHelperKey, Timestamp> {
|
||||
let compiled_params = compiled_param_declarations(index);
|
||||
logs_range(index)
|
||||
.map(|(sequence, timestamp, text)| {
|
||||
let key = match derive_single_param(text, &compiled_params, sort_by) {
|
||||
Some(value) => SortHelperKey::WithParam(value, sequence),
|
||||
None => SortHelperKey::WithoutParam(sequence),
|
||||
};
|
||||
(key, timestamp)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Incrementally extends every currently-live helper index with one freshly-ingested line, called
|
||||
/// from `EngineMessage::Ingest` right alongside the primary index insert (ticket 060) — this is what
|
||||
/// lets a sorted query avoid re-deriving params for the whole timeline on every call once a helper
|
||||
/// index for that sort key exists. A no-op when no helper index has been built yet (the common case
|
||||
/// for a param nobody has sorted by), so ingest pays nothing for sort keys that are never queried.
|
||||
fn update_sort_helper_indexes_on_ingest(
|
||||
sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>,
|
||||
index: &BTreeMap<IndexKey, DataValue>,
|
||||
sequence: u64,
|
||||
timestamp: Timestamp,
|
||||
text: &str,
|
||||
) {
|
||||
if sort_helper_indexes.is_empty() {
|
||||
return;
|
||||
}
|
||||
let compiled_params = compiled_param_declarations(index);
|
||||
for (sort_by, entry) in sort_helper_indexes.iter_mut() {
|
||||
let key = match derive_single_param(text, &compiled_params, sort_by) {
|
||||
Some(value) => SortHelperKey::WithParam(value, sequence),
|
||||
None => SortHelperKey::WithoutParam(sequence),
|
||||
};
|
||||
entry.index.insert(key, timestamp);
|
||||
}
|
||||
}
|
||||
|
||||
/// Ticket 061: drops any helper index whose `last_accessed` exceeds `SORT_HELPER_INDEX_PURGE_AFTER`.
|
||||
/// Cheap to call opportunistically on every ingest (`HashMap::retain` is a no-op scan when nothing's
|
||||
/// stale) rather than needing a dedicated timer/task — a purged index rebuilds transparently via
|
||||
/// `build_sort_helper_index` the next time something sorts by it.
|
||||
fn purge_stale_sort_helper_indexes(sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>) {
|
||||
sort_helper_indexes.retain(|_, entry| entry.last_accessed.elapsed() < SORT_HELPER_INDEX_PURGE_AFTER);
|
||||
}
|
||||
|
||||
fn compiled_param_declarations(
|
||||
index: &BTreeMap<IndexKey, DataValue>,
|
||||
) -> Vec<(String, Regex, String, Option<String>, MetricPrimitiveKind)> {
|
||||
@@ -1409,6 +1550,7 @@ fn handle_query(
|
||||
index: &BTreeMap<IndexKey, DataValue>,
|
||||
query: EngineQuery,
|
||||
approx_bytes_total: u64,
|
||||
sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>,
|
||||
) -> EngineResponse {
|
||||
match query {
|
||||
EngineQuery::Logs(params) => {
|
||||
@@ -1453,34 +1595,64 @@ fn handle_query(
|
||||
params: derive_log_params(text, &compiled_params),
|
||||
})
|
||||
.collect(),
|
||||
// An alternate sort key needs every matching line's params derived up front (there's
|
||||
// no way to know which lines will end up in the first `limit` before sorting), then
|
||||
// reordered before the limit is finally applied.
|
||||
// Ticket 060: walk the (lazily built-or-reused) helper index for this sort key in
|
||||
// key order instead of materializing and sorting the full filtered set — the index's
|
||||
// own key ordering already expresses "sorted-by-value lines, then lines missing the
|
||||
// param ordered by ottyl_seq" (`sort_missing_fallback`'s semantics), so this just
|
||||
// filters/limits as it walks rather than partition-sort-concat afterward.
|
||||
Some(sort_by) => {
|
||||
let mut all: Vec<LogEntryDto> = filtered
|
||||
.map(|(sequence, timestamp, text)| LogEntryDto {
|
||||
let helper = sort_helper_indexes.entry(sort_by.to_string()).or_insert_with(|| {
|
||||
SortHelperIndexEntry {
|
||||
index: build_sort_helper_index(index, sort_by),
|
||||
last_accessed: std::time::Instant::now(),
|
||||
}
|
||||
});
|
||||
helper.last_accessed = std::time::Instant::now();
|
||||
|
||||
let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true);
|
||||
let mut entries = Vec::new();
|
||||
for (helper_key, timestamp) in helper.index.iter() {
|
||||
let sequence = match helper_key {
|
||||
SortHelperKey::WithParam(_, sequence) => *sequence,
|
||||
SortHelperKey::WithoutParam(sequence) => {
|
||||
if !sort_missing_fallback {
|
||||
continue;
|
||||
}
|
||||
*sequence
|
||||
}
|
||||
};
|
||||
if !(params.since_sequence.is_none_or(|since| sequence > since)
|
||||
&& params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
|
||||
&& params.end_ms.is_none_or(|end| timestamp.as_millis() <= end))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
// A line evicted (by `max_logs`) since this key was recorded is simply
|
||||
// skipped, same as buckets/other derived data aren't retroactively pruned
|
||||
// on eviction elsewhere in this file.
|
||||
let Some(DataValue::LogLiteral { text }) =
|
||||
index.get(&IndexKey::IngestedLog { sequence, timestamp: *timestamp })
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
if !compiled_filter
|
||||
.as_ref()
|
||||
.is_none_or(|filter| eval_compiled(filter, sequence, *timestamp, text))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
entries.push(LogEntryDto {
|
||||
sequence,
|
||||
timestamp_ms: timestamp.as_millis(),
|
||||
text: text.to_string(),
|
||||
text: text.clone(),
|
||||
params: derive_log_params(text, &compiled_params),
|
||||
})
|
||||
.collect();
|
||||
let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true);
|
||||
if sort_missing_fallback {
|
||||
let (mut with_param, mut without_param): (Vec<_>, Vec<_>) =
|
||||
all.drain(..).partition(|entry| entry.params.contains_key(sort_by));
|
||||
with_param.sort_by(|a, b| a.params[sort_by].cmp(&b.params[sort_by]));
|
||||
// Lines missing the sort param are only ordered relative to each other (by
|
||||
// `ottyl_seq`) and appended after the sorted lines, rather than interleaved
|
||||
// into the overall ordering — see ticket 045.
|
||||
without_param.sort_by_key(|entry| entry.sequence);
|
||||
with_param.into_iter().chain(without_param).collect()
|
||||
} else {
|
||||
all.retain(|entry| entry.params.contains_key(sort_by));
|
||||
all.sort_by(|a, b| a.params[sort_by].cmp(&b.params[sort_by]));
|
||||
all
|
||||
});
|
||||
if entries.len() >= limit {
|
||||
break;
|
||||
}
|
||||
}
|
||||
entries
|
||||
}
|
||||
};
|
||||
let entries: Vec<LogEntryDto> = entries.into_iter().take(limit).collect();
|
||||
EngineResponse::Logs(entries)
|
||||
@@ -2106,8 +2278,9 @@ mod tests {
|
||||
|
||||
/// Runs both auto-declaration passes over every fixture line, same order as `EngineMessage::Ingest`.
|
||||
fn ingest_log_stream_fixture(index: &mut BTreeMap<IndexKey, DataValue>) {
|
||||
let mut header = None;
|
||||
for line in LOG_STREAM_FIXTURE.lines() {
|
||||
auto_declare_log_stream_params(index, line);
|
||||
auto_declare_log_stream_params(index, &mut header, line);
|
||||
auto_declare_params(index, line);
|
||||
}
|
||||
}
|
||||
@@ -2115,13 +2288,13 @@ mod tests {
|
||||
#[test]
|
||||
fn ingesting_log_stream_fixture_does_not_blow_up_declared_param_count() {
|
||||
// Ticket 050: unbounded auto-declaration on this real `log stream` dump produced 220 distinct
|
||||
// params. The bound (MAX_AUTO_DECLARED_PARAMS + the 3 fixed log_stream params) must keep the
|
||||
// total in the tens, nowhere near "hundreds".
|
||||
// params. The bound (MAX_AUTO_DECLARED_PARAMS + the header-derived log_stream columns) must
|
||||
// keep the total in the tens, nowhere near "hundreds".
|
||||
let mut index = BTreeMap::new();
|
||||
ingest_log_stream_fixture(&mut index);
|
||||
|
||||
let declared = param_declarations_range(&index).count();
|
||||
// `log_type`/`process`/`pid` are stored as `auto_derived` too, so the shared
|
||||
// The header-derived columns (ticket 059) are stored as `auto_derived` too, so the shared
|
||||
// `MAX_AUTO_DECLARED_PARAMS` ceiling covers all of them together (see
|
||||
// `auto_declare_log_stream_params`'s doc comment).
|
||||
assert!(
|
||||
@@ -2144,29 +2317,38 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn log_stream_shaped_lines_expose_type_process_and_pid_as_queryable_params() {
|
||||
fn log_stream_shaped_lines_expose_header_named_columns_as_queryable_params() {
|
||||
let mut index = BTreeMap::new();
|
||||
ingest_log_stream_fixture(&mut index);
|
||||
|
||||
let declared: HashSet<&str> = param_declarations_range(&index).map(|d| d.0).collect();
|
||||
assert!(declared.contains("log_type"));
|
||||
assert!(declared.contains("process"));
|
||||
assert!(declared.contains("pid"));
|
||||
// Derived generically from the fixture's own header row (ticket 059), not a fixed set of three.
|
||||
for name in ["Timestamp", "Thread", "Type", "Activity", "PID", "TTL", "message"] {
|
||||
assert!(declared.contains(name), "expected {name} to be declared, got {declared:?}");
|
||||
}
|
||||
|
||||
let compiled = compiled_param_declarations(&index);
|
||||
let data_line = LOG_STREAM_FIXTURE.lines().nth(1).unwrap();
|
||||
let params = derive_log_params(data_line, &compiled);
|
||||
assert_eq!(params.get("log_type"), Some(&MetricPrimitive::String("Default".to_string())));
|
||||
assert_eq!(params.get("process"), Some(&MetricPrimitive::String("kitty".to_string())));
|
||||
assert_eq!(params.get("pid"), Some(&MetricPrimitive::Number(Number::Int(1343))));
|
||||
assert_eq!(params.get("Type"), Some(&MetricPrimitive::String("Default".to_string())));
|
||||
assert_eq!(params.get("PID"), Some(&MetricPrimitive::Number(Number::Int(1343))));
|
||||
assert_eq!(params.get("TTL"), Some(&MetricPrimitive::Number(Number::Int(7))));
|
||||
assert_eq!(
|
||||
params.get("message"),
|
||||
Some(&MetricPrimitive::String(
|
||||
"kitty: (SkyLight) [com.apple.SkyLight:default] SLSGetDisplaysWithOpenGLDisplayMask, 3614: Display 0x00000001: Mode 54, (0 x 0) [1728 x 1117], Depth 32 BPP, active".to_string()
|
||||
))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_log_stream_lines_do_not_get_log_type_process_or_pid_declared() {
|
||||
fn non_log_stream_lines_do_not_get_any_columns_declared_without_a_cached_header() {
|
||||
let mut index = BTreeMap::new();
|
||||
auto_declare_log_stream_params(&mut index, r#"ts=1788648105773 msg="entity_moved" x1=-4"#);
|
||||
let mut header = None;
|
||||
auto_declare_log_stream_params(&mut index, &mut header, r#"ts=1788648105773 msg="entity_moved" x1=-4"#);
|
||||
|
||||
assert!(param_declarations_range(&index).next().is_none());
|
||||
assert!(header.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -2176,8 +2358,9 @@ mod tests {
|
||||
let mut index = BTreeMap::new();
|
||||
let start = std::time::Instant::now();
|
||||
for _ in 0..repeats {
|
||||
let mut header = None;
|
||||
for line in LOG_STREAM_FIXTURE.lines() {
|
||||
auto_declare_log_stream_params(&mut index, line);
|
||||
auto_declare_log_stream_params(&mut index, &mut header, line);
|
||||
auto_declare_params(&mut index, line);
|
||||
}
|
||||
}
|
||||
@@ -3578,6 +3761,7 @@ mod tests {
|
||||
end_ms: None,
|
||||
},
|
||||
0,
|
||||
&mut HashMap::new(),
|
||||
);
|
||||
match response {
|
||||
EngineResponse::Promql(Err(message)) => assert!(!message.is_empty()),
|
||||
@@ -3599,6 +3783,7 @@ mod tests {
|
||||
end_ms: None,
|
||||
},
|
||||
0,
|
||||
&mut HashMap::new(),
|
||||
);
|
||||
match response {
|
||||
EngineResponse::Promql(Ok(series)) => {
|
||||
@@ -3691,6 +3876,7 @@ mod tests {
|
||||
&index,
|
||||
EngineQuery::Promql { query: "count(ottyl_seq)".to_string(), start_ms: None, end_ms: None },
|
||||
0,
|
||||
&mut HashMap::new(),
|
||||
);
|
||||
match response {
|
||||
EngineResponse::Promql(Ok(series)) => {
|
||||
@@ -3753,7 +3939,7 @@ mod tests {
|
||||
ingest_log(&mut index, 2, "still fine");
|
||||
ingest_log(&mut index, 3, "another error here");
|
||||
|
||||
match handle_query(&index, logs_query(Some(r#"regex("error")"#)), 0) {
|
||||
match handle_query(&index, logs_query(Some(r#"regex("error")"#)), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![1, 3]);
|
||||
@@ -3769,7 +3955,7 @@ mod tests {
|
||||
ingest_log(&mut index, i, &format!("line {i}"));
|
||||
}
|
||||
|
||||
match handle_query(&index, logs_query(Some("ottyl_seq<2")), 0) {
|
||||
match handle_query(&index, logs_query(Some("ottyl_seq<2")), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![0, 1]);
|
||||
@@ -3789,7 +3975,7 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
match handle_query(&index, logs_query(Some("ottyl_ts>2000")), 0) {
|
||||
match handle_query(&index, logs_query(Some("ottyl_ts>2000")), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![3, 4]);
|
||||
@@ -3816,7 +4002,7 @@ mod tests {
|
||||
ingest_log(&mut index, 1, "priority=1 boom");
|
||||
ingest_log(&mut index, 2, "priority=2 boom");
|
||||
|
||||
match handle_query(&index, logs_query_sorted("priority", None), 0) {
|
||||
match handle_query(&index, logs_query_sorted("priority", None), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![1, 2, 0]);
|
||||
@@ -3844,7 +4030,7 @@ mod tests {
|
||||
ingest_log(&mut index, 2, "priority=1 boom"); // has param
|
||||
ingest_log(&mut index, 3, "also missing"); // missing
|
||||
|
||||
match handle_query(&index, logs_query_sorted("priority", Some(true)), 0) {
|
||||
match handle_query(&index, logs_query_sorted("priority", Some(true)), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
// Lines with the param sorted ascending by value (2 then 0), followed by the
|
||||
@@ -3870,7 +4056,7 @@ mod tests {
|
||||
ingest_log(&mut index, 2, "priority=1 boom");
|
||||
ingest_log(&mut index, 3, "also missing");
|
||||
|
||||
match handle_query(&index, logs_query_sorted("priority", Some(false)), 0) {
|
||||
match handle_query(&index, logs_query_sorted("priority", Some(false)), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![2, 0]);
|
||||
@@ -3884,7 +4070,7 @@ mod tests {
|
||||
let mut index = BTreeMap::new();
|
||||
ingest_log(&mut index, 0, "hello");
|
||||
|
||||
match handle_query(&index, logs_query_sorted("does_not_exist", None), 0) {
|
||||
match handle_query(&index, logs_query_sorted("does_not_exist", None), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Error(err) => assert!(err.contains("does_not_exist")),
|
||||
_ => panic!("expected EngineResponse::Error"),
|
||||
}
|
||||
@@ -3908,7 +4094,7 @@ mod tests {
|
||||
ingest_log(&mut index, 2, "priority=1 boom");
|
||||
ingest_log(&mut index, 3, "also missing");
|
||||
|
||||
match handle_query(&index, logs_query(None), 0) {
|
||||
match handle_query(&index, logs_query(None), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![0, 1, 2, 3]);
|
||||
@@ -3917,6 +4103,136 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
// -- Ticket 060: sort helper index --
|
||||
|
||||
#[test]
|
||||
fn sort_helper_index_is_built_lazily_and_reused_across_queries() {
|
||||
let mut index = BTreeMap::new();
|
||||
declare_test_param(
|
||||
&mut index,
|
||||
"priority",
|
||||
r"priority=(?P<priority>\d+)",
|
||||
"priority",
|
||||
MetricPrimitiveKind::Number,
|
||||
);
|
||||
ingest_log(&mut index, 0, "priority=3 boom");
|
||||
ingest_log(&mut index, 1, "priority=1 boom");
|
||||
ingest_log(&mut index, 2, "priority=2 boom");
|
||||
|
||||
let mut sort_helper_indexes = HashMap::new();
|
||||
assert!(!sort_helper_indexes.contains_key("priority"));
|
||||
|
||||
// First query has no helper index yet — must lazily build one rather than erroring or
|
||||
// silently returning unsorted results.
|
||||
match handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![1, 2, 0]);
|
||||
}
|
||||
_ => panic!("expected Logs response"),
|
||||
}
|
||||
assert!(sort_helper_indexes.contains_key("priority"));
|
||||
assert_eq!(sort_helper_indexes["priority"].index.len(), 3);
|
||||
|
||||
// Ingest more lines directly into the already-built helper index (bypassing the full
|
||||
// `EngineMessage::Ingest` actor loop, same as every other test in this file exercises
|
||||
// `handle_query` directly) via the same incremental-update path `run()` calls on ingest,
|
||||
// then requery — the new lines must appear correctly ordered without the helper index
|
||||
// having been thrown away and rebuilt from scratch.
|
||||
ingest_log(&mut index, 3, "priority=0 boom");
|
||||
update_sort_helper_indexes_on_ingest(
|
||||
&mut sort_helper_indexes,
|
||||
&index,
|
||||
3,
|
||||
Timestamp::from_millis(3),
|
||||
"priority=0 boom",
|
||||
);
|
||||
ingest_log(&mut index, 4, "no priority here");
|
||||
update_sort_helper_indexes_on_ingest(
|
||||
&mut sort_helper_indexes,
|
||||
&index,
|
||||
4,
|
||||
Timestamp::from_millis(4),
|
||||
"no priority here",
|
||||
);
|
||||
assert_eq!(sort_helper_indexes["priority"].index.len(), 5);
|
||||
|
||||
match handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
// 3 (priority=0) sorts before the original three; 4 (missing the param) falls back
|
||||
// to ottyl_seq order, appended after every line that has the param.
|
||||
assert_eq!(sequences, vec![3, 1, 2, 0, 4]);
|
||||
}
|
||||
_ => panic!("expected Logs response"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sort_helper_index_incremental_update_is_a_noop_when_no_index_is_live() {
|
||||
// Guards `update_sort_helper_indexes_on_ingest`'s early return: ingest must not pay to
|
||||
// derive a param nobody has ever sorted by.
|
||||
let mut sort_helper_indexes: HashMap<String, SortHelperIndexEntry> = HashMap::new();
|
||||
let index = BTreeMap::new();
|
||||
update_sort_helper_indexes_on_ingest(&mut sort_helper_indexes, &index, 0, Timestamp::from_millis(0), "hello");
|
||||
assert!(sort_helper_indexes.is_empty());
|
||||
}
|
||||
|
||||
// -- Ticket 061: purging stale sort helper indexes --
|
||||
|
||||
#[test]
|
||||
fn purge_stale_sort_helper_indexes_drops_only_entries_past_the_purge_duration() {
|
||||
let mut sort_helper_indexes: HashMap<String, SortHelperIndexEntry> = HashMap::new();
|
||||
sort_helper_indexes.insert(
|
||||
"stale".to_string(),
|
||||
SortHelperIndexEntry {
|
||||
index: BTreeMap::new(),
|
||||
last_accessed: std::time::Instant::now() - SORT_HELPER_INDEX_PURGE_AFTER
|
||||
- std::time::Duration::from_secs(1),
|
||||
},
|
||||
);
|
||||
sort_helper_indexes.insert(
|
||||
"fresh".to_string(),
|
||||
SortHelperIndexEntry { index: BTreeMap::new(), last_accessed: std::time::Instant::now() },
|
||||
);
|
||||
|
||||
purge_stale_sort_helper_indexes(&mut sort_helper_indexes);
|
||||
|
||||
assert!(!sort_helper_indexes.contains_key("stale"));
|
||||
assert!(sort_helper_indexes.contains_key("fresh"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sorted_query_transparently_rebuilds_a_purged_helper_index() {
|
||||
let mut index = BTreeMap::new();
|
||||
declare_test_param(
|
||||
&mut index,
|
||||
"priority",
|
||||
r"priority=(?P<priority>\d+)",
|
||||
"priority",
|
||||
MetricPrimitiveKind::Number,
|
||||
);
|
||||
ingest_log(&mut index, 0, "priority=3 boom");
|
||||
ingest_log(&mut index, 1, "priority=1 boom");
|
||||
ingest_log(&mut index, 2, "priority=2 boom");
|
||||
|
||||
let mut sort_helper_indexes = HashMap::new();
|
||||
let _ = handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes);
|
||||
assert!(sort_helper_indexes.contains_key("priority"));
|
||||
|
||||
// Simulate ticket 061's sweep having purged this entry for staleness.
|
||||
sort_helper_indexes.remove("priority");
|
||||
|
||||
match handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![1, 2, 0]);
|
||||
}
|
||||
_ => panic!("expected Logs response"),
|
||||
}
|
||||
assert!(sort_helper_indexes.contains_key("priority"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn logs_query_with_ottyl_fmt_comparison_filters_by_detected_format_for_mixed_format_lines() {
|
||||
let mut index = BTreeMap::new();
|
||||
@@ -3926,7 +4242,7 @@ mod tests {
|
||||
ingest_log(&mut index, 2, "just some plain prose here");
|
||||
ingest_log(&mut index, 3, r#"{"level":"info","code":200}"#);
|
||||
|
||||
match handle_query(&index, logs_query(Some("ottyl_fmt==json")), 0) {
|
||||
match handle_query(&index, logs_query(Some("ottyl_fmt==json")), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![0, 3]);
|
||||
@@ -3934,7 +4250,7 @@ mod tests {
|
||||
_ => panic!("expected Logs response"),
|
||||
}
|
||||
|
||||
match handle_query(&index, logs_query(Some("ottyl_fmt==logfmt")), 0) {
|
||||
match handle_query(&index, logs_query(Some("ottyl_fmt==logfmt")), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![1]);
|
||||
@@ -3942,7 +4258,7 @@ mod tests {
|
||||
_ => panic!("expected Logs response"),
|
||||
}
|
||||
|
||||
match handle_query(&index, logs_query(Some("ottyl_fmt==none")), 0) {
|
||||
match handle_query(&index, logs_query(Some("ottyl_fmt==none")), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![2]);
|
||||
@@ -3958,7 +4274,7 @@ mod tests {
|
||||
ingest_log(&mut index, 1, r#"{"level":"info","code":200}"#);
|
||||
ingest_log(&mut index, 2, "not json at all");
|
||||
|
||||
match handle_query(&index, logs_query(Some(r#"json[level]==error"#)), 0) {
|
||||
match handle_query(&index, logs_query(Some(r#"json[level]==error"#)), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![0]);
|
||||
@@ -3966,7 +4282,7 @@ mod tests {
|
||||
_ => panic!("expected Logs response"),
|
||||
}
|
||||
|
||||
match handle_query(&index, logs_query(Some("json[code]>=300")), 0) {
|
||||
match handle_query(&index, logs_query(Some("json[code]>=300")), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![0]);
|
||||
@@ -3991,6 +4307,7 @@ mod tests {
|
||||
&index,
|
||||
logs_query(Some(r#"regex("boom") and requests[level]==error"#)),
|
||||
0,
|
||||
&mut HashMap::new(),
|
||||
) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
@@ -4004,6 +4321,7 @@ mod tests {
|
||||
&index,
|
||||
logs_query(Some(r#"regex("boom") or requests[level]==error"#)),
|
||||
0,
|
||||
&mut HashMap::new(),
|
||||
) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
@@ -4020,7 +4338,7 @@ mod tests {
|
||||
ingest_log(&mut index, 0, "level=info hello");
|
||||
ingest_log(&mut index, 1, "no match here");
|
||||
|
||||
let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0) else {
|
||||
let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0, &mut HashMap::new()) else {
|
||||
panic!("expected Logs response");
|
||||
};
|
||||
|
||||
@@ -4042,7 +4360,7 @@ mod tests {
|
||||
ingest_log(&mut index, 0, "an ERROR occurred");
|
||||
ingest_log(&mut index, 1, "everything is fine");
|
||||
|
||||
let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0) else {
|
||||
let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0, &mut HashMap::new()) else {
|
||||
panic!("expected Logs response");
|
||||
};
|
||||
|
||||
@@ -4058,7 +4376,7 @@ mod tests {
|
||||
ingest_log(&mut index, 1, "an error occurred");
|
||||
ingest_log(&mut index, 2, "another error here");
|
||||
|
||||
match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000), 0) {
|
||||
match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Timeline(buckets) => {
|
||||
let total: u64 = buckets.iter().map(|b| b.count).sum();
|
||||
assert_eq!(total, 2);
|
||||
@@ -4072,12 +4390,12 @@ mod tests {
|
||||
let mut index = BTreeMap::new();
|
||||
ingest_log(&mut index, 0, "hello");
|
||||
|
||||
match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@")), 0) {
|
||||
match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@")), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Error(message) => assert!(!message.is_empty()),
|
||||
_ => panic!("expected an Error response for a malformed filter string"),
|
||||
}
|
||||
|
||||
match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000), 0) {
|
||||
match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Error(message) => assert!(!message.is_empty()),
|
||||
_ => panic!("expected an Error response for a malformed filter string"),
|
||||
}
|
||||
@@ -4091,7 +4409,7 @@ mod tests {
|
||||
declare_metric(&mut index, metric_request("requests", pattern, vec!["level"])).unwrap();
|
||||
ingest_log(&mut index, 0, "level=error");
|
||||
|
||||
match handle_query(&index, logs_query(Some("requests==5")), 0) {
|
||||
match handle_query(&index, logs_query(Some("requests==5")), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Error(message) => assert!(message.contains("param")),
|
||||
_ => panic!("expected an Error response for a metric comparison without [param]"),
|
||||
}
|
||||
@@ -4106,7 +4424,7 @@ mod tests {
|
||||
declare_metric(&mut index, metric_request("requests", pattern, vec!["level", "duration"])).unwrap();
|
||||
ingest_log(&mut index, 0, "level=error duration=5");
|
||||
|
||||
match handle_query(&index, logs_query(Some("requests[nonexistent_param]==\"x\"")), 0) {
|
||||
match handle_query(&index, logs_query(Some("requests[nonexistent_param]==\"x\"")), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Error(message) => {
|
||||
assert!(
|
||||
message.contains("nonexistent_param"),
|
||||
@@ -4136,7 +4454,7 @@ mod tests {
|
||||
ingest_log(&mut index, 0, "level=error duration=5");
|
||||
ingest_log(&mut index, 1, "level=info duration=2");
|
||||
|
||||
match handle_query(&index, logs_query(Some("requests[level]==error")), 0) {
|
||||
match handle_query(&index, logs_query(Some("requests[level]==error")), 0, &mut HashMap::new()) {
|
||||
EngineResponse::Logs(entries) => {
|
||||
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
|
||||
assert_eq!(sequences, vec![0]);
|
||||
|
||||
+299
-17
@@ -101,16 +101,223 @@ pub fn is_log_stream_line(line: &str) -> bool {
|
||||
LOG_STREAM_RE.is_match(line)
|
||||
}
|
||||
|
||||
/// `message_match_regex` patterns `crate::engine::auto_declare_log_stream_params` uses to declare
|
||||
/// `log_type`/`process`/`pid` as standalone params the first time a `log stream`-shaped line is seen.
|
||||
/// Each isolates exactly one named capture group (matching `derive_log_params`'s one-group-per-param
|
||||
/// convention) over the same column layout `LOG_STREAM_RE` detects.
|
||||
pub const LOG_STREAM_TYPE_PATTERN: &str =
|
||||
r"^\d{4}-\d{2}-\d{2}\s+\S+\s+0x[0-9a-fA-F]+\s+(?P<log_type>\S+)\s+0x[0-9a-fA-F]+\s+\d+\s+\d+\s+[^\s:]+:\s";
|
||||
pub const LOG_STREAM_PROCESS_PATTERN: &str =
|
||||
r"^\d{4}-\d{2}-\d{2}\s+\S+\s+0x[0-9a-fA-F]+\s+\S+\s+0x[0-9a-fA-F]+\s+\d+\s+\d+\s+(?P<process>[^\s:]+):\s";
|
||||
pub const LOG_STREAM_PID_PATTERN: &str =
|
||||
r"^\d{4}-\d{2}-\d{2}\s+\S+\s+0x[0-9a-fA-F]+\s+\S+\s+0x[0-9a-fA-F]+\s+(?P<pid>\d+)\s+\d+\s+[^\s:]+:\s";
|
||||
/// Regex fragment for one column separator under the generic whitespace-columnar rule ticket 059
|
||||
/// asks for: a run of 2+ whitespace chars, or a lone tab, separates columns; a single plain space does
|
||||
/// not (so a value that legitimately contains one -- e.g. `log stream`'s `date time+tz` timestamp --
|
||||
/// stays intact within its own column). `is_separator_run` is the same rule applied at split time;
|
||||
/// this is that rule re-expressed as a regex fragment for the trailing message field's
|
||||
/// `message_match_regex`, which re-derives its value the same way whenever a caller queries by it
|
||||
/// later (declarations don't cache extracted values -- see `crate::engine::extract_param_raw`).
|
||||
const COLUMN_SEPARATOR_PATTERN: &str = r"(?:\t|[ \t]{2,})";
|
||||
|
||||
/// A column's value: one or more non-whitespace tokens joined by single spaces, with no upper bound
|
||||
/// on length. Used for the trailing (unbounded-width) message field and the last named column, where
|
||||
/// there's no next column's start position to bound against.
|
||||
const COLUMN_VALUE_PATTERN: &str = r"\S+(?: \S+)*";
|
||||
|
||||
/// Same idea as `COLUMN_VALUE_PATTERN`, but bounded to a known column `width` (in chars) -- needed for
|
||||
/// every column except the last, since two adjacent `log stream` values can legitimately be separated
|
||||
/// by only a single space when the first one nearly fills its column's on-screen width (e.g. the
|
||||
/// `Timestamp` column: its own `date time+tz` value already contains one internal space, so a
|
||||
/// content-only pattern can't tell that boundary apart from the one-space gap before the next column
|
||||
/// starts -- seed this from `LogStreamColumn::start` positions instead). Greedily matches up to
|
||||
/// `width` chars while requiring the match to end on a non-whitespace char, which is what excludes the
|
||||
/// column's own trailing padding spaces from the capture.
|
||||
fn bounded_column_value_pattern(width: usize) -> String {
|
||||
if width <= 1 {
|
||||
r"\S".to_string()
|
||||
} else {
|
||||
format!(r"\S(?:.{{0,{}}}\S)?", width - 2)
|
||||
}
|
||||
}
|
||||
|
||||
fn is_separator_run(run: &str) -> bool {
|
||||
run.len() >= 2 || run.contains('\t')
|
||||
}
|
||||
|
||||
static WHITESPACE_RUN: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"[ \t]+").expect("static regex must compile"));
|
||||
|
||||
/// Splits `line` at whitespace runs that qualify as a column separator (`is_separator_run`), returning
|
||||
/// each piece alongside its byte offset in `line`; a lone plain space never splits. Consumes at most
|
||||
/// `max_columns.map(|n| n - 1)` separators -- the final piece returned is the untouched remainder of
|
||||
/// the line, including any further whitespace runs inside it. That's what lets a trailing unlabelled
|
||||
/// `log stream` message tail (whose own internal spacing can be arbitrary) come back as one field
|
||||
/// instead of being chopped up by it.
|
||||
fn split_columns_with_offsets(line: &str, max_columns: Option<usize>) -> Vec<(usize, &str)> {
|
||||
let mut columns = Vec::new();
|
||||
let mut start = 0usize;
|
||||
let mut consumed = 0usize;
|
||||
for m in WHITESPACE_RUN.find_iter(line) {
|
||||
if max_columns.is_some_and(|max| consumed + 1 >= max) {
|
||||
break;
|
||||
}
|
||||
let run = &line[m.start()..m.end()];
|
||||
if is_separator_run(run) {
|
||||
columns.push((start, &line[start..m.start()]));
|
||||
start = m.end();
|
||||
consumed += 1;
|
||||
}
|
||||
}
|
||||
columns.push((start, &line[start..]));
|
||||
columns
|
||||
}
|
||||
|
||||
fn split_columns(line: &str, max_columns: Option<usize>) -> Vec<&str> {
|
||||
split_columns_with_offsets(line, max_columns).into_iter().map(|(_, s)| s).collect()
|
||||
}
|
||||
|
||||
/// A column-name identifier a generated `message_match_regex`'s named capture group can use as-is
|
||||
/// (the regex crate requires `[A-Za-z_][0-9A-Za-z_]*`).
|
||||
static COLUMN_NAME: LazyLock<Regex> =
|
||||
LazyLock::new(|| Regex::new(r"^[A-Za-z][A-Za-z0-9_]*$").expect("static regex must compile"));
|
||||
|
||||
/// One column named by a detected `log stream` header (ticket 059). `start` is that column's position
|
||||
/// in `char`s (not bytes -- see `char_slice`) within the header line -- meaningful only when
|
||||
/// `LogStreamHeader::tab_delimited` is `false` (see that field's doc comment).
|
||||
pub struct LogStreamColumn {
|
||||
pub name: String,
|
||||
start: usize,
|
||||
}
|
||||
|
||||
/// A detected `log stream` header: its named columns, plus whether they were tab- or space-separated.
|
||||
/// `log stream`'s real output right-pads each header name to a fixed on-screen width using runs of
|
||||
/// 2+ spaces, and every data line's values land in that same fixed alignment regardless of how much
|
||||
/// of that width they individually fill -- so for that shape, a data line's column boundaries are
|
||||
/// best found at the header's own fixed character positions, not by re-splitting the data line's own
|
||||
/// whitespace (see `split_log_stream_row`'s doc comment for why that alone isn't reliable there).
|
||||
/// A tab-separated header carries no such alignment guarantee (values aren't padded to a shared
|
||||
/// width), but a lone tab is itself an unambiguous separator, so that case instead re-splits each data
|
||||
/// line directly.
|
||||
pub struct LogStreamHeader {
|
||||
pub columns: Vec<LogStreamColumn>,
|
||||
tab_delimited: bool,
|
||||
}
|
||||
|
||||
/// Byte-indexed `str` slicing doesn't work directly with `LogStreamColumn::start` (a *char* offset),
|
||||
/// since `log stream` output is expected to stay ASCII in practice but callers shouldn't panic if a
|
||||
/// message contains multi-byte chars before the slice point. `end` of `None` slices to end of string.
|
||||
fn char_slice(s: &str, start: usize, end: Option<usize>) -> &str {
|
||||
if end.is_some_and(|end| end <= start) {
|
||||
return "";
|
||||
}
|
||||
let mut indices = s.char_indices();
|
||||
let byte_start = indices.nth(start).map(|(i, _)| i).unwrap_or(s.len());
|
||||
let byte_end = match end {
|
||||
Some(end) => s.char_indices().nth(end).map(|(i, _)| i).unwrap_or(s.len()),
|
||||
None => s.len(),
|
||||
};
|
||||
&s[byte_start..byte_end.max(byte_start)]
|
||||
}
|
||||
|
||||
/// Recognizes a `log stream`-style header row generically (ticket 059), instead of assuming the fixed
|
||||
/// `Timestamp Thread Type Activity PID TTL` names: a short run of bareword column names separated by
|
||||
/// qualifying whitespace runs (`split_columns_with_offsets`) that is, critically, *not itself* a
|
||||
/// `log stream` data line (`is_log_stream_line`) -- so a header is never mistaken for the data line
|
||||
/// immediately following it. `crate::engine::auto_declare_log_stream_params` caches the result across
|
||||
/// ingested lines, since `log stream` prints its header once, before any data line, and this only ever
|
||||
/// sees one line at a time.
|
||||
pub fn detect_log_stream_header(line: &str) -> Option<LogStreamHeader> {
|
||||
if is_log_stream_line(line) {
|
||||
return None;
|
||||
}
|
||||
let mut columns = Vec::new();
|
||||
for (byte_start, slice) in split_columns_with_offsets(line, None) {
|
||||
let trimmed = slice.trim();
|
||||
if trimmed.is_empty() {
|
||||
continue;
|
||||
}
|
||||
let leading_ws = slice.len() - slice.trim_start().len();
|
||||
let start = line[..byte_start + leading_ws].chars().count();
|
||||
columns.push(LogStreamColumn { name: trimmed.to_string(), start });
|
||||
}
|
||||
if columns.len() < 2 || !columns.iter().all(|c| COLUMN_NAME.is_match(&c.name)) {
|
||||
return None;
|
||||
}
|
||||
let tab_delimited = WHITESPACE_RUN.find_iter(line).any(|m| m.as_str().contains('\t'));
|
||||
Some(LogStreamHeader { columns, tab_delimited })
|
||||
}
|
||||
|
||||
/// Splits a `log stream` data line into `header`'s named columns plus, when present, one trailing
|
||||
/// unlabelled tail -- `crate::engine::auto_declare_log_stream_params` maps the first
|
||||
/// `header.columns.len()` pieces onto the header's names and the optional last piece onto the
|
||||
/// `message` param.
|
||||
///
|
||||
/// For a space-padded fixed-width header, every column but the last is sliced at the header's own
|
||||
/// fixed character positions (`LogStreamColumn::start`) rather than by re-splitting the data line's
|
||||
/// own whitespace, since two adjacent values can legitimately be joined by only a single space when
|
||||
/// the first nearly fills its column's width (e.g. `log stream`'s `Timestamp` column, whose value
|
||||
/// already contains one internal space). Only the last named column's end is whitespace-run-detected
|
||||
/// (`split_columns`), since that's what distinguishes its own value from the free-text message tail
|
||||
/// `log stream` appends after it.
|
||||
///
|
||||
/// For a tab-delimited header, there's no shared on-screen width to align against, but a lone tab is
|
||||
/// itself unambiguous, so this instead re-splits the data line directly on qualifying separators.
|
||||
pub fn split_log_stream_row<'a>(line: &'a str, header: &LogStreamHeader) -> Vec<&'a str> {
|
||||
if header.tab_delimited {
|
||||
return split_columns(line, Some(header.columns.len() + 1));
|
||||
}
|
||||
let mut result = Vec::with_capacity(header.columns.len() + 1);
|
||||
for (i, column) in header.columns.iter().enumerate() {
|
||||
match header.columns.get(i + 1) {
|
||||
Some(next) => result.push(char_slice(line, column.start, Some(next.start)).trim()),
|
||||
None => {
|
||||
let tail = char_slice(line, column.start, None);
|
||||
let pieces = split_columns(tail, Some(2));
|
||||
result.push(pieces[0].trim());
|
||||
if let Some(message) = pieces.get(1) {
|
||||
result.push(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
/// Builds the `message_match_regex` for the `index`-th (0-based) of `header`'s named columns.
|
||||
///
|
||||
/// Fixed-width case: skips exactly `header.columns[index].start` chars, then captures this column's
|
||||
/// own value bounded to its known width (`bounded_column_value_pattern`) -- unbounded for the last
|
||||
/// column, which has no next column to bound against.
|
||||
///
|
||||
/// Tab-delimited case: skips `index` preceding columns' own value+separator pairs instead, since
|
||||
/// there's no shared fixed width to anchor a character-position skip against.
|
||||
///
|
||||
/// Anchored at the start of the line since a `log stream` line never carries this fixed-per-run
|
||||
/// column prefix anywhere but the very start.
|
||||
pub fn log_stream_column_pattern(header: &LogStreamHeader, index: usize, capture_group: &str) -> String {
|
||||
if header.tab_delimited {
|
||||
let mut pattern = String::from("^");
|
||||
for _ in 0..index {
|
||||
pattern.push_str(COLUMN_VALUE_PATTERN);
|
||||
pattern.push_str(COLUMN_SEPARATOR_PATTERN);
|
||||
}
|
||||
pattern.push_str(&format!("(?P<{capture_group}>{COLUMN_VALUE_PATTERN})"));
|
||||
return pattern;
|
||||
}
|
||||
let start = header.columns[index].start;
|
||||
let value_pattern = match header.columns.get(index + 1) {
|
||||
Some(next) => bounded_column_value_pattern(next.start.saturating_sub(start)),
|
||||
None => COLUMN_VALUE_PATTERN.to_string(),
|
||||
};
|
||||
format!("^.{{{start}}}(?P<{capture_group}>{value_pattern})")
|
||||
}
|
||||
|
||||
/// Builds the `message_match_regex` for the trailing unlabelled tail after `header`'s last named
|
||||
/// column: everything from the end of that column's own separator to end of line, verbatim -- its own
|
||||
/// internal whitespace, however irregular, is never treated as a further separator.
|
||||
pub fn log_stream_message_pattern(header: &LogStreamHeader, capture_group: &str) -> String {
|
||||
if header.tab_delimited {
|
||||
let mut pattern = String::from("^");
|
||||
for _ in 0..header.columns.len() {
|
||||
pattern.push_str(COLUMN_VALUE_PATTERN);
|
||||
pattern.push_str(COLUMN_SEPARATOR_PATTERN);
|
||||
}
|
||||
pattern.push_str(&format!("(?P<{capture_group}>.*)$"));
|
||||
return pattern;
|
||||
}
|
||||
let start = header.columns.last().map(|c| c.start).unwrap_or(0);
|
||||
format!("^.{{{start}}}{COLUMN_VALUE_PATTERN}{COLUMN_SEPARATOR_PATTERN}(?P<{capture_group}>.*)$")
|
||||
}
|
||||
|
||||
/// Go `log/slog` `NewJSONHandler` output (the `go-log` label): a single-line JSON object whose first
|
||||
/// three fields are always `time`, `level`, `msg` in that fixed order (slog's `JSONHandler` writes the
|
||||
@@ -290,15 +497,90 @@ mod tests {
|
||||
assert!(!is_log_stream_line(header));
|
||||
}
|
||||
|
||||
fn column_names(header: &LogStreamHeader) -> Vec<&str> {
|
||||
header.columns.iter().map(|c| c.name.as_str()).collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn log_stream_patterns_capture_type_process_and_pid() {
|
||||
fn detects_log_stream_header_columns_generically() {
|
||||
let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap();
|
||||
let header = detect_log_stream_header(header_line).expect("fixture header must be detected");
|
||||
assert_eq!(column_names(&header), vec!["Timestamp", "Thread", "Type", "Activity", "PID", "TTL"]);
|
||||
assert!(!header.tab_delimited);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_log_stream_data_line_is_never_detected_as_its_own_header() {
|
||||
let data_line = LOG_STREAM_FIXTURE.lines().nth(1).unwrap();
|
||||
assert!(detect_log_stream_header(data_line).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plain_prose_is_not_detected_as_a_log_stream_header() {
|
||||
assert!(detect_log_stream_header("just some plain prose here").is_none());
|
||||
assert!(detect_log_stream_header("ts=1788648105773 msg=\"entity_moved\"").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn splits_log_stream_row_into_named_columns_plus_trailing_message() {
|
||||
let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap();
|
||||
let header = detect_log_stream_header(header_line).unwrap();
|
||||
let data_line = "2026-09-06 01:48:54.518117+0100 0x4dfed2 Default 0x0 606 0 WindowServer: (HSTouchHIDService) something happened";
|
||||
let row = split_log_stream_row(data_line, &header);
|
||||
assert_eq!(row.len(), header.columns.len() + 1);
|
||||
assert_eq!(row[0], "2026-09-06 01:48:54.518117+0100");
|
||||
assert_eq!(row[1], "0x4dfed2");
|
||||
assert_eq!(row[2], "Default");
|
||||
assert_eq!(row[3], "0x0");
|
||||
assert_eq!(row[4], "606");
|
||||
assert_eq!(row[5], "0");
|
||||
// The message tail is never chopped up by its own internal single/double spacing.
|
||||
assert_eq!(row[6], "WindowServer: (HSTouchHIDService) something happened");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_single_tab_separates_columns_exactly_like_a_run_of_spaces() {
|
||||
let header = detect_log_stream_header("Timestamp\tThread\tType").unwrap();
|
||||
assert_eq!(column_names(&header), vec!["Timestamp", "Thread", "Type"]);
|
||||
assert!(header.tab_delimited);
|
||||
|
||||
// Two tabs in a row still qualify as a single separator (like a run of 2+ spaces), so this
|
||||
// still cleanly separates Timestamp/Thread/Type despite the doubled tab before "Default".
|
||||
let spaced = split_log_stream_row("2026-09-06 01:48:54\t0x4dfed2\t\tDefault trailing message", &header);
|
||||
let tabbed = split_log_stream_row("2026-09-06 01:48:54\t0x4dfed2\tDefault trailing message", &header);
|
||||
assert_eq!(spaced[0], "2026-09-06 01:48:54");
|
||||
assert_eq!(spaced[1], "0x4dfed2");
|
||||
assert_eq!(spaced[2], "Default trailing message");
|
||||
assert_eq!(tabbed[0], "2026-09-06 01:48:54");
|
||||
assert_eq!(tabbed[1], "0x4dfed2");
|
||||
assert_eq!(tabbed[2], "Default trailing message");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn column_and_message_patterns_capture_the_same_fields_split_log_stream_row_finds() {
|
||||
let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap();
|
||||
let header = detect_log_stream_header(header_line).unwrap();
|
||||
let line = "2026-09-06 01:48:54.518117+0100 0x4dfed2 Default 0x0 606 0 WindowServer: (HSTouchHIDService) something happened";
|
||||
let type_re = Regex::new(LOG_STREAM_TYPE_PATTERN).unwrap();
|
||||
let process_re = Regex::new(LOG_STREAM_PROCESS_PATTERN).unwrap();
|
||||
let pid_re = Regex::new(LOG_STREAM_PID_PATTERN).unwrap();
|
||||
assert_eq!(type_re.captures(line).unwrap().name("log_type").unwrap().as_str(), "Default");
|
||||
assert_eq!(process_re.captures(line).unwrap().name("process").unwrap().as_str(), "WindowServer");
|
||||
assert_eq!(pid_re.captures(line).unwrap().name("pid").unwrap().as_str(), "606");
|
||||
|
||||
let type_re = Regex::new(&log_stream_column_pattern(&header, 2, "Type")).unwrap();
|
||||
assert_eq!(type_re.captures(line).unwrap().name("Type").unwrap().as_str(), "Default");
|
||||
|
||||
let pid_re = Regex::new(&log_stream_column_pattern(&header, 4, "PID")).unwrap();
|
||||
assert_eq!(pid_re.captures(line).unwrap().name("PID").unwrap().as_str(), "606");
|
||||
|
||||
let message_re = Regex::new(&log_stream_message_pattern(&header, "message")).unwrap();
|
||||
assert_eq!(
|
||||
message_re.captures(line).unwrap().name("message").unwrap().as_str(),
|
||||
"WindowServer: (HSTouchHIDService) something happened"
|
||||
);
|
||||
|
||||
// Also exercise the Timestamp column specifically -- its own value contains one internal
|
||||
// space, so a naive content-only pattern would previously over-consume into the next column.
|
||||
let timestamp_re = Regex::new(&log_stream_column_pattern(&header, 0, "Timestamp")).unwrap();
|
||||
assert_eq!(
|
||||
timestamp_re.captures(line).unwrap().name("Timestamp").unwrap().as_str(),
|
||||
"2026-09-06 01:48:54.518117+0100"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
+95
-6
@@ -8,17 +8,19 @@ mod filter_parser;
|
||||
mod index_key;
|
||||
mod ingest;
|
||||
mod logfmt;
|
||||
mod mcp_server;
|
||||
mod metric_primitive;
|
||||
mod profile;
|
||||
mod promql;
|
||||
mod rate;
|
||||
mod sessions;
|
||||
mod settings;
|
||||
mod timestamp;
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::sync::{Arc, LazyLock, Mutex};
|
||||
|
||||
use clap::Parser;
|
||||
use clap::{Parser, Subcommand};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
@@ -38,6 +40,9 @@ static SETTINGS_PATH_HELP: LazyLock<String> = LazyLock::new(|| {
|
||||
/// Observability dashboard from terminal output.
|
||||
#[derive(Parser, Debug)]
|
||||
struct Args {
|
||||
#[command(subcommand)]
|
||||
command: Option<Command>,
|
||||
|
||||
/// Port to serve the API and webapp on. Defaults to 0, letting the OS assign
|
||||
/// an ephemeral free port.
|
||||
#[arg(long, default_value_t = 0)]
|
||||
@@ -53,6 +58,30 @@ struct Args {
|
||||
/// URL of a Prometheus/OpenMetrics-compatible endpoint to poll for metrics, in addition to log ingestion.
|
||||
#[arg(long)]
|
||||
metrics_url: Option<String>,
|
||||
|
||||
/// Name to register this session under, so `ottyl mcp --session <name>` or `ottyl sessions`
|
||||
/// can find it later. Defaults to a random human-readable name (e.g. "comic-gazelle").
|
||||
/// Starting a second instance under the same name as one still running overwrites its
|
||||
/// registration (the older process keeps running, just without a discoverable session entry).
|
||||
#[arg(long)]
|
||||
name: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Subcommand, Debug)]
|
||||
enum Command {
|
||||
/// Start an MCP (Model Context Protocol) server over stdio, proxying tool calls to a running
|
||||
/// ottyl instance's HTTP API. Register the printed command with an MCP client (e.g. Claude
|
||||
/// Code's `.mcp.json`).
|
||||
Mcp {
|
||||
/// Name of a running ottyl session to connect to (see `ottyl sessions`).
|
||||
#[arg(long)]
|
||||
session: Option<String>,
|
||||
/// Port of a running ottyl instance to connect to, instead of resolving `--session` by name.
|
||||
#[arg(long)]
|
||||
port: Option<u16>,
|
||||
},
|
||||
/// List known ottyl sessions and the ports they're serving on.
|
||||
Sessions,
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
@@ -63,6 +92,19 @@ async fn main() {
|
||||
settings::default_settings_path()
|
||||
.expect("could not determine a default settings path; pass --settings-path")
|
||||
});
|
||||
|
||||
match args.command {
|
||||
Some(Command::Mcp { session, port }) => {
|
||||
run_mcp(&settings_path, session, port).await;
|
||||
return;
|
||||
}
|
||||
Some(Command::Sessions) => {
|
||||
print_sessions(&settings_path);
|
||||
return;
|
||||
}
|
||||
None => {}
|
||||
}
|
||||
|
||||
let settings = Arc::new(Mutex::new(Settings::load(&settings_path)));
|
||||
|
||||
let (engine_tx, engine_rx) = mpsc::channel(1024);
|
||||
@@ -113,7 +155,7 @@ async fn main() {
|
||||
tokio::spawn(ingest::run_metrics_poll(engine_tx.clone(), metrics_url));
|
||||
}
|
||||
|
||||
let app = api::router(engine_tx, settings, settings_path);
|
||||
let app = api::router(engine_tx, settings, settings_path.clone());
|
||||
let listener = TcpListener::bind(("0.0.0.0", args.port))
|
||||
.await
|
||||
.expect("failed to bind listener");
|
||||
@@ -126,16 +168,63 @@ async fn main() {
|
||||
Some(path) => format!("tailing {}", path.display()),
|
||||
None => "reading logs from stdin".to_string(),
|
||||
};
|
||||
eprintln!(
|
||||
"ottyl serving on http://localhost:{} ({})",
|
||||
bound_port, source
|
||||
);
|
||||
let session_name = args.name.clone().unwrap_or_else(sessions::generate_name);
|
||||
match sessions::register_session(&settings_path, &session_name, bound_port) {
|
||||
Ok(_) => eprintln!(
|
||||
"ottyl serving on http://localhost:{} ({}), session name: {}",
|
||||
bound_port, source, session_name
|
||||
),
|
||||
Err(err) => eprintln!(
|
||||
"ottyl serving on http://localhost:{} ({}); warning: failed to register session '{}': {}",
|
||||
bound_port, source, session_name, err
|
||||
),
|
||||
}
|
||||
|
||||
axum::serve(listener, app)
|
||||
.await
|
||||
.expect("server error");
|
||||
}
|
||||
|
||||
/// Resolves `--session`/`--port` into a target base URL and runs the MCP server over stdio until
|
||||
/// the client disconnects.
|
||||
async fn run_mcp(settings_path: &std::path::Path, session: Option<String>, port: Option<u16>) {
|
||||
use rmcp::ServiceExt;
|
||||
|
||||
let port = match (port, session) {
|
||||
(Some(port), _) => port,
|
||||
(None, Some(name)) => match sessions::resolve_session(settings_path, &name) {
|
||||
Some(info) => info.port,
|
||||
None => {
|
||||
eprintln!("error: no running ottyl session named '{name}'");
|
||||
std::process::exit(1);
|
||||
}
|
||||
},
|
||||
(None, None) => {
|
||||
eprintln!("error: `ottyl mcp` requires --session <name> or --port <port>");
|
||||
std::process::exit(1);
|
||||
}
|
||||
};
|
||||
|
||||
let base_url = format!("http://localhost:{port}");
|
||||
let service = mcp_server::OttylMcp::new(base_url)
|
||||
.serve(rmcp::transport::stdio())
|
||||
.await
|
||||
.expect("failed to start MCP server");
|
||||
service.waiting().await.expect("MCP server error");
|
||||
}
|
||||
|
||||
fn print_sessions(settings_path: &std::path::Path) {
|
||||
let dir = sessions::sessions_dir(settings_path);
|
||||
let sessions = sessions::list_sessions(&dir);
|
||||
if sessions.is_empty() {
|
||||
println!("no known ottyl sessions");
|
||||
return;
|
||||
}
|
||||
for session in sessions {
|
||||
println!("{}\tport={}\tpid={}", session.name, session.port, session.pid);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@@ -0,0 +1,409 @@
|
||||
//! MCP (Model Context Protocol) server for agents (ticket 067): a short-lived process, started
|
||||
//! via `ottyl mcp --session <name>` (see `crate::sessions`), that speaks MCP over stdio and
|
||||
//! proxies each tool call to a *running* ottyl instance's existing HTTP API. Every tool here is a
|
||||
//! thin translation onto `api.rs` routes/DTOs — no query logic is duplicated outside `engine.rs`.
|
||||
//!
|
||||
//! Uses the official `rmcp` SDK (https://github.com/modelcontextprotocol/rust-sdk) over the
|
||||
//! `transport-io` (stdio) transport rather than hand-rolling the JSON-RPC/MCP handshake, since a
|
||||
//! maintained implementation of the protocol's framing and lifecycle is worth the extra
|
||||
//! dependency.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use rmcp::handler::server::router::tool::ToolRouter;
|
||||
use rmcp::handler::server::wrapper::Parameters;
|
||||
use rmcp::model::{
|
||||
CallToolResult, ContentBlock, Implementation, ProtocolVersion, ServerCapabilities, ServerInfo,
|
||||
};
|
||||
use rmcp::{ErrorData as McpError, ServerHandler, schemars, tool, tool_handler, tool_router};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
pub struct QueryLogsParams {
|
||||
/// FilterBar query string, e.g. `regex("error") and ottyl_seq<5`. Omitted matches every log.
|
||||
#[serde(default)]
|
||||
pub filter: Option<String>,
|
||||
/// If true, return only how many logs matched instead of the log lines themselves.
|
||||
#[serde(default)]
|
||||
pub count_only: bool,
|
||||
/// Maximum number of log lines to return. Ignored when `count_only` is set.
|
||||
#[serde(default)]
|
||||
pub limit: Option<usize>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
pub struct DeclareParamParams {
|
||||
pub name: String,
|
||||
/// Regex matched against each log line; `capture_group` extracts this param's value from it.
|
||||
pub message_match_regex: String,
|
||||
pub capture_group: String,
|
||||
/// Dotted path into a JSON-structured log line to extract from instead of `capture_group`.
|
||||
#[serde(default)]
|
||||
pub json_field_path: Option<String>,
|
||||
/// One of: "string", "number", "timestamp", "duration", "boolean".
|
||||
pub kind: String,
|
||||
/// Extract the elapsed time since this param was last seen, instead of parsing captured text.
|
||||
#[serde(default)]
|
||||
pub elapsed_since_last: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, schemars::JsonSchema)]
|
||||
pub struct DeclareMetricParams {
|
||||
pub name: String,
|
||||
pub message_match_regex: String,
|
||||
/// Names of already-declared params whose values become this metric's bucket labels.
|
||||
#[serde(default)]
|
||||
pub label_params: Vec<String>,
|
||||
/// Name of an already-declared param whose value becomes this metric's per-event bucket value.
|
||||
#[serde(default)]
|
||||
pub value_param: Option<String>,
|
||||
/// Fixed labels layered on top of `label_params` for matching lines.
|
||||
#[serde(default)]
|
||||
pub matched_labels: Option<BTreeMap<String, String>>,
|
||||
/// Fixed labels recorded once for a non-matching line, instead of it being invisible to this metric.
|
||||
#[serde(default)]
|
||||
pub unmatched_labels: Option<BTreeMap<String, String>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
pub struct CreateDashboardParams {
|
||||
/// Profile to create the dashboard in. Defaults to the currently active profile.
|
||||
#[serde(default)]
|
||||
pub profile: Option<String>,
|
||||
pub name: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
pub struct AddPanelParams {
|
||||
/// Profile the dashboard lives in. Defaults to the currently active profile.
|
||||
#[serde(default)]
|
||||
pub profile: Option<String>,
|
||||
pub dashboard_id: String,
|
||||
/// Id of an existing panel to update in place; omit to add a new panel.
|
||||
#[serde(default)]
|
||||
pub panel_id: Option<String>,
|
||||
pub title: String,
|
||||
/// PromQL-style query backing this panel (see `crate::promql`).
|
||||
pub promql: String,
|
||||
/// One of: "bar", "line", "pie". Defaults to "line".
|
||||
#[serde(default = "default_chart_type")]
|
||||
pub chart_type: String,
|
||||
}
|
||||
|
||||
fn default_chart_type() -> String {
|
||||
"line".to_string()
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
pub struct GetDashboardParams {
|
||||
/// Profile the dashboard lives in. Defaults to the currently active profile.
|
||||
#[serde(default)]
|
||||
pub profile: Option<String>,
|
||||
pub dashboard_id: String,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct OttylMcp {
|
||||
client: reqwest::Client,
|
||||
base_url: String,
|
||||
// Constructed fresh per call by the `#[tool_handler]`-generated `ServerHandler` methods
|
||||
// (`Self::tool_router()`), not read through this field — kept only because `#[tool_router]`
|
||||
// requires a field of this name/type to exist on the struct.
|
||||
#[allow(dead_code)]
|
||||
tool_router: ToolRouter<OttylMcp>,
|
||||
}
|
||||
|
||||
fn mcp_error(context: &str, err: impl std::fmt::Display) -> McpError {
|
||||
McpError::internal_error(format!("{context}: {err}"), None)
|
||||
}
|
||||
|
||||
fn json_result(value: Value) -> Result<CallToolResult, McpError> {
|
||||
Ok(CallToolResult::success(vec![ContentBlock::text(
|
||||
serde_json::to_string_pretty(&value).unwrap_or_else(|_| value.to_string()),
|
||||
)]))
|
||||
}
|
||||
|
||||
#[tool_router]
|
||||
impl OttylMcp {
|
||||
pub fn new(base_url: String) -> Self {
|
||||
Self {
|
||||
client: reqwest::Client::new(),
|
||||
base_url,
|
||||
tool_router: Self::tool_router(),
|
||||
}
|
||||
}
|
||||
|
||||
fn url(&self, path: &str) -> String {
|
||||
format!("{}{path}", self.base_url)
|
||||
}
|
||||
|
||||
/// Resolves the profile a dashboard tool call should operate on: the explicit `profile`
|
||||
/// argument if given, otherwise the target instance's currently active profile.
|
||||
async fn resolve_profile(&self, explicit: Option<String>) -> Result<String, McpError> {
|
||||
if let Some(profile) = explicit {
|
||||
return Ok(profile);
|
||||
}
|
||||
let settings: Value = self
|
||||
.client
|
||||
.get(self.url("/api/settings"))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to reach ottyl", err))?
|
||||
.json()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to parse settings response", err))?;
|
||||
settings
|
||||
.get("active_profile")
|
||||
.and_then(Value::as_str)
|
||||
.map(str::to_string)
|
||||
.ok_or_else(|| {
|
||||
McpError::invalid_params(
|
||||
"no active profile on the target instance; pass `profile` explicitly",
|
||||
None,
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
#[tool(description = "Count or read log lines matching a FilterBar query against the running ottyl instance.")]
|
||||
async fn query_logs(
|
||||
&self,
|
||||
Parameters(params): Parameters<QueryLogsParams>,
|
||||
) -> Result<CallToolResult, McpError> {
|
||||
let mut query: Vec<(&str, String)> = Vec::new();
|
||||
if let Some(filter) = ¶ms.filter {
|
||||
query.push(("filter", filter.clone()));
|
||||
}
|
||||
if let Some(limit) = params.limit {
|
||||
query.push(("limit", limit.to_string()));
|
||||
}
|
||||
let response = self
|
||||
.client
|
||||
.get(self.url("/api/logs"))
|
||||
.query(&query)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to reach ottyl", err))?;
|
||||
if !response.status().is_success() {
|
||||
return Err(mcp_error("ottyl rejected the query", response.status()));
|
||||
}
|
||||
let logs: Value = response
|
||||
.json()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to parse logs response", err))?;
|
||||
if params.count_only {
|
||||
let count = logs.as_array().map(Vec::len).unwrap_or(0);
|
||||
json_result(serde_json::json!({ "count": count }))
|
||||
} else {
|
||||
json_result(logs)
|
||||
}
|
||||
}
|
||||
|
||||
#[tool(description = "Declare a param: a named value extracted from matching log lines.")]
|
||||
async fn declare_param(
|
||||
&self,
|
||||
Parameters(params): Parameters<DeclareParamParams>,
|
||||
) -> Result<CallToolResult, McpError> {
|
||||
let response = self
|
||||
.client
|
||||
.post(self.url("/api/params"))
|
||||
.json(¶ms)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to reach ottyl", err))?;
|
||||
let status = response.status();
|
||||
let body: Value = response
|
||||
.json()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to parse declare_param response", err))?;
|
||||
if !status.is_success() {
|
||||
return Err(mcp_error("ottyl rejected the param declaration", body));
|
||||
}
|
||||
json_result(body)
|
||||
}
|
||||
|
||||
#[tool(description = "Declare a metric: named event counts/values bucketed by label, derived from matching log lines.")]
|
||||
async fn declare_metric(
|
||||
&self,
|
||||
Parameters(params): Parameters<DeclareMetricParams>,
|
||||
) -> Result<CallToolResult, McpError> {
|
||||
let response = self
|
||||
.client
|
||||
.post(self.url("/api/metrics"))
|
||||
.json(¶ms)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to reach ottyl", err))?;
|
||||
let status = response.status();
|
||||
let body: Value = response
|
||||
.json()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to parse declare_metric response", err))?;
|
||||
if !status.is_success() {
|
||||
return Err(mcp_error("ottyl rejected the metric declaration", body));
|
||||
}
|
||||
json_result(body)
|
||||
}
|
||||
|
||||
#[tool(description = "Create a new, empty dashboard in a profile.")]
|
||||
async fn create_dashboard(
|
||||
&self,
|
||||
Parameters(params): Parameters<CreateDashboardParams>,
|
||||
) -> Result<CallToolResult, McpError> {
|
||||
let profile = self.resolve_profile(params.profile).await?;
|
||||
let response = self
|
||||
.client
|
||||
.post(self.url(&format!(
|
||||
"/api/settings/profiles/{}/dashboards",
|
||||
urlencoding_path(&profile)
|
||||
)))
|
||||
.json(&serde_json::json!({ "name": params.name }))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to reach ottyl", err))?;
|
||||
let status = response.status();
|
||||
let body: Value = response
|
||||
.json()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to parse create_dashboard response", err))?;
|
||||
if !status.is_success() {
|
||||
return Err(mcp_error("ottyl rejected the dashboard creation", body));
|
||||
}
|
||||
json_result(body)
|
||||
}
|
||||
|
||||
#[tool(description = "Read a dashboard's current panels.")]
|
||||
async fn get_dashboard(
|
||||
&self,
|
||||
Parameters(params): Parameters<GetDashboardParams>,
|
||||
) -> Result<CallToolResult, McpError> {
|
||||
let profile = self.resolve_profile(params.profile).await?;
|
||||
let body = self.fetch_dashboard(&profile, ¶ms.dashboard_id).await?;
|
||||
json_result(body)
|
||||
}
|
||||
|
||||
// Read-modify-write over HTTP (no dashboard-level locking): two concurrent `add_panel` calls
|
||||
// against the same dashboard can race and one's addition can be lost, same as two browser tabs
|
||||
// saving the same dashboard through the webapp today. Not addressed here — out of scope for a
|
||||
// thin proxy over the existing `save_dashboard` endpoint.
|
||||
#[tool(description = "Add a new panel to a dashboard, or update an existing one by panel_id.")]
|
||||
async fn add_panel(
|
||||
&self,
|
||||
Parameters(params): Parameters<AddPanelParams>,
|
||||
) -> Result<CallToolResult, McpError> {
|
||||
let profile = self.resolve_profile(params.profile).await?;
|
||||
let mut dashboard = self.fetch_dashboard(&profile, ¶ms.dashboard_id).await?;
|
||||
|
||||
let panel_id = params
|
||||
.panel_id
|
||||
.clone()
|
||||
.unwrap_or_else(|| generate_panel_id(¶ms.title, ¶ms.promql));
|
||||
let new_panel = serde_json::json!({
|
||||
"id": panel_id,
|
||||
"title": params.title,
|
||||
"promql": params.promql,
|
||||
"chart_type": params.chart_type,
|
||||
});
|
||||
|
||||
let panels = dashboard
|
||||
.get_mut("panels")
|
||||
.and_then(Value::as_array_mut)
|
||||
.ok_or_else(|| mcp_error("unexpected dashboard shape", "missing panels array"))?;
|
||||
match panels.iter().position(|panel| panel.get("id") == new_panel.get("id")) {
|
||||
Some(index) => panels[index] = new_panel,
|
||||
None => panels.push(new_panel),
|
||||
}
|
||||
|
||||
let response = self
|
||||
.client
|
||||
.put(self.url(&format!(
|
||||
"/api/settings/profiles/{}/dashboards/{}",
|
||||
urlencoding_path(&profile),
|
||||
urlencoding_path(¶ms.dashboard_id)
|
||||
)))
|
||||
.json(&serde_json::json!({ "name": dashboard.get("name"), "panels": dashboard.get("panels") }))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to reach ottyl", err))?;
|
||||
let status = response.status();
|
||||
let body: Value = response
|
||||
.json()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to parse add_panel response", err))?;
|
||||
if !status.is_success() {
|
||||
return Err(mcp_error("ottyl rejected the panel update", body));
|
||||
}
|
||||
json_result(body)
|
||||
}
|
||||
|
||||
async fn fetch_dashboard(&self, profile: &str, dashboard_id: &str) -> Result<Value, McpError> {
|
||||
let response = self
|
||||
.client
|
||||
.get(self.url(&format!(
|
||||
"/api/settings/profiles/{}/dashboards/{}",
|
||||
urlencoding_path(profile),
|
||||
urlencoding_path(dashboard_id)
|
||||
)))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to reach ottyl", err))?;
|
||||
let status = response.status();
|
||||
let body: Value = response
|
||||
.json()
|
||||
.await
|
||||
.map_err(|err| mcp_error("failed to parse dashboard response", err))?;
|
||||
if !status.is_success() {
|
||||
return Err(mcp_error("ottyl rejected the dashboard lookup", body));
|
||||
}
|
||||
Ok(body)
|
||||
}
|
||||
}
|
||||
|
||||
/// Percent-encodes a single path segment (profile/dashboard names/ids can contain characters that
|
||||
/// aren't valid unescaped in a URL path).
|
||||
fn urlencoding_path(segment: &str) -> String {
|
||||
segment
|
||||
.bytes()
|
||||
.map(|b| match b {
|
||||
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
|
||||
(b as char).to_string()
|
||||
}
|
||||
_ => format!("%{b:02X}"),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Generates a new panel id by hashing title/promql/time, avoiding a `uuid` dependency (matches
|
||||
/// `settings::generate_imported_css_id`'s approach for the same reason).
|
||||
fn generate_panel_id(title: &str, promql: &str) -> String {
|
||||
use std::collections::hash_map::DefaultHasher;
|
||||
use std::hash::{Hash, Hasher};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
let mut hasher = DefaultHasher::new();
|
||||
title.hash(&mut hasher);
|
||||
promql.hash(&mut hasher);
|
||||
let nanos = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_nanos())
|
||||
.unwrap_or(0);
|
||||
nanos.hash(&mut hasher);
|
||||
format!("panel-{:016x}", hasher.finish())
|
||||
}
|
||||
|
||||
#[tool_handler]
|
||||
impl ServerHandler for OttylMcp {
|
||||
fn get_info(&self) -> ServerInfo {
|
||||
ServerInfo::new(
|
||||
ServerCapabilities::builder().enable_tools().build(),
|
||||
)
|
||||
.with_server_info(Implementation::from_build_env())
|
||||
.with_protocol_version(ProtocolVersion::V_2024_11_05)
|
||||
.with_instructions(
|
||||
"Proxies to a running ottyl instance's HTTP API. Tools: query_logs, declare_param, \
|
||||
declare_metric, create_dashboard, add_panel, get_dashboard. See the embedded \
|
||||
SKILL.md (GET /SKILL.md on the target instance) for examples."
|
||||
.to_string(),
|
||||
)
|
||||
}
|
||||
}
|
||||
+188
@@ -0,0 +1,188 @@
|
||||
//! Named session registry (ticket 068): each running `ottyl` process registers itself under a
|
||||
//! name by writing a small file to `<settings dir>/sessions/<name>.json`, so a second CLI
|
||||
//! invocation (e.g. `ottyl mcp --session <name>`, ticket 067) can resolve a name to a port without
|
||||
//! the user copying one around by hand.
|
||||
|
||||
use std::fs;
|
||||
use std::io;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// On-disk shape of a `sessions/<name>.json` file.
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
pub struct SessionInfo {
|
||||
pub name: String,
|
||||
pub port: u16,
|
||||
pub pid: u32,
|
||||
}
|
||||
|
||||
/// The `sessions/` directory for a given settings file, following the same
|
||||
/// sibling-directory convention as `profiles/` (see `settings::migrate_legacy_profiles_file`).
|
||||
pub fn sessions_dir(settings_path: &Path) -> PathBuf {
|
||||
settings_path
|
||||
.parent()
|
||||
.map(|parent| parent.join("sessions"))
|
||||
.unwrap_or_else(|| PathBuf::from("sessions"))
|
||||
}
|
||||
|
||||
fn session_file_path(dir: &Path, name: &str) -> PathBuf {
|
||||
dir.join(format!("{name}.json"))
|
||||
}
|
||||
|
||||
/// Best-effort liveness check for a PID. Shells out to `kill -0` on unix (no signal is delivered,
|
||||
/// only permission/existence is checked); assumed alive on other platforms, which just means a
|
||||
/// stale session file there waits for an explicit name collision to be overwritten instead of
|
||||
/// being proactively cleaned up.
|
||||
#[cfg(unix)]
|
||||
fn is_pid_alive(pid: u32) -> bool {
|
||||
process::Command::new("kill")
|
||||
.args(["-0", &pid.to_string()])
|
||||
.stdout(process::Stdio::null())
|
||||
.stderr(process::Stdio::null())
|
||||
.status()
|
||||
.map(|status| status.success())
|
||||
.unwrap_or(true)
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
fn is_pid_alive(_pid: u32) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
/// Removes session files whose owning process is no longer running. Best-effort: there's no
|
||||
/// shutdown hook to unregister a session on graceful exit, so this runs on every startup instead.
|
||||
/// Any I/O error reading a single entry is ignored — a future call gets another chance.
|
||||
pub fn cleanup_stale(dir: &Path) {
|
||||
let Ok(entries) = fs::read_dir(dir) else {
|
||||
return;
|
||||
};
|
||||
for entry in entries.flatten() {
|
||||
let path = entry.path();
|
||||
if path.extension().and_then(|ext| ext.to_str()) != Some("json") {
|
||||
continue;
|
||||
}
|
||||
let Ok(contents) = fs::read_to_string(&path) else {
|
||||
continue;
|
||||
};
|
||||
let Ok(info) = serde_json::from_str::<SessionInfo>(&contents) else {
|
||||
continue;
|
||||
};
|
||||
if !is_pid_alive(info.pid) {
|
||||
let _ = fs::remove_file(&path);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Generates a random human-readable session name (e.g. `comic-gazelle`) for when `--name` isn't
|
||||
/// given explicitly.
|
||||
pub fn generate_name() -> String {
|
||||
petname::petname(2, "-").unwrap_or_else(|| format!("session-{}", process::id()))
|
||||
}
|
||||
|
||||
/// Registers this process as session `name` on `port`: cleans up stale entries, then writes
|
||||
/// `sessions/<name>.json`, overwriting any existing file for that name (documented behavior — a
|
||||
/// second instance started under the same explicit `--name` as a still-running one simply
|
||||
/// replaces its registration; the older process keeps running, just without a discoverable entry).
|
||||
pub fn register_session(settings_path: &Path, name: &str, port: u16) -> io::Result<PathBuf> {
|
||||
let dir = sessions_dir(settings_path);
|
||||
fs::create_dir_all(&dir)?;
|
||||
cleanup_stale(&dir);
|
||||
let info = SessionInfo {
|
||||
name: name.to_string(),
|
||||
port,
|
||||
pid: process::id(),
|
||||
};
|
||||
let path = session_file_path(&dir, name);
|
||||
fs::write(&path, serde_json::to_string_pretty(&info)?)?;
|
||||
Ok(path)
|
||||
}
|
||||
|
||||
/// Looks up a session by name. Returns `None` if no such file exists, it fails to parse, or its
|
||||
/// process is no longer alive (a stale entry left by a process that didn't exit through the normal
|
||||
/// next-`register_session` cleanup path).
|
||||
pub fn resolve_session(settings_path: &Path, name: &str) -> Option<SessionInfo> {
|
||||
let dir = sessions_dir(settings_path);
|
||||
let contents = fs::read_to_string(session_file_path(&dir, name)).ok()?;
|
||||
let info: SessionInfo = serde_json::from_str(&contents).ok()?;
|
||||
is_pid_alive(info.pid).then_some(info)
|
||||
}
|
||||
|
||||
/// Lists every session file under `dir`, live or stale (callers that only want live sessions
|
||||
/// should filter on `is_pid_alive`-backed `resolve_session` instead). Used by `ottyl sessions`.
|
||||
pub fn list_sessions(dir: &Path) -> Vec<SessionInfo> {
|
||||
let Ok(entries) = fs::read_dir(dir) else {
|
||||
return Vec::new();
|
||||
};
|
||||
let mut sessions: Vec<SessionInfo> = entries
|
||||
.flatten()
|
||||
.filter(|entry| entry.path().extension().and_then(|ext| ext.to_str()) == Some("json"))
|
||||
.filter_map(|entry| fs::read_to_string(entry.path()).ok())
|
||||
.filter_map(|contents| serde_json::from_str(&contents).ok())
|
||||
.collect();
|
||||
sessions.sort_by(|a, b| a.name.cmp(&b.name));
|
||||
sessions
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn register_session_writes_correct_port() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let settings_path = dir.path().join("settings.json");
|
||||
|
||||
let path = register_session(&settings_path, "foo", 4242).unwrap();
|
||||
assert!(path.exists());
|
||||
|
||||
let info = resolve_session(&settings_path, "foo").unwrap();
|
||||
assert_eq!(info.port, 4242);
|
||||
assert_eq!(info.name, "foo");
|
||||
assert_eq!(info.pid, process::id());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_session_returns_none_for_unknown_name() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let settings_path = dir.path().join("settings.json");
|
||||
assert!(resolve_session(&settings_path, "does-not-exist").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stale_session_does_not_block_reregistration_under_same_name() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let settings_path = dir.path().join("settings.json");
|
||||
let sessions_dir_path = sessions_dir(&settings_path);
|
||||
fs::create_dir_all(&sessions_dir_path).unwrap();
|
||||
|
||||
// A PID essentially guaranteed not to be alive (PID 1 exists but a huge, unlikely-assigned
|
||||
// PID stands in as "not alive" without depending on system PID-reuse specifics).
|
||||
let stale = SessionInfo { name: "foo".to_string(), port: 1111, pid: 999_999 };
|
||||
fs::write(
|
||||
session_file_path(&sessions_dir_path, "foo"),
|
||||
serde_json::to_string(&stale).unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
// Registering a live session under the same name succeeds and the stale cleanup pass (run
|
||||
// as part of registration) doesn't error out on the bogus pre-existing entry.
|
||||
register_session(&settings_path, "foo", 5555).unwrap();
|
||||
let info = resolve_session(&settings_path, "foo").unwrap();
|
||||
assert_eq!(info.port, 5555);
|
||||
assert_eq!(info.pid, process::id());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn list_sessions_returns_all_entries_sorted_by_name() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let settings_path = dir.path().join("settings.json");
|
||||
register_session(&settings_path, "zeta", 1).unwrap();
|
||||
register_session(&settings_path, "alpha", 2).unwrap();
|
||||
|
||||
let sessions = list_sessions(&sessions_dir(&settings_path));
|
||||
let names: Vec<&str> = sessions.iter().map(|s| s.name.as_str()).collect();
|
||||
assert_eq!(names, vec!["alpha", "zeta"]);
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -4,7 +4,7 @@
|
||||
<meta charset="UTF-8" />
|
||||
<link rel="icon" type="image/svg+xml" href="/favicon.svg" />
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
|
||||
<title>webapp</title>
|
||||
<title>ottyl</title>
|
||||
</head>
|
||||
<body>
|
||||
<div id="root"></div>
|
||||
|
||||
+37
-3
@@ -254,6 +254,28 @@ function StandaloneRanged({
|
||||
)
|
||||
}
|
||||
|
||||
/// Ticket 058: only replaces `params` state when the declaration list actually changed, so a param
|
||||
/// list refetched on every poll tick doesn't force `Params.tsx` to re-render when nothing changed.
|
||||
function updateParamsIfChanged(
|
||||
setParams: (updater: (current: ParamDeclaration[]) => ParamDeclaration[]) => void,
|
||||
next: ParamDeclaration[],
|
||||
) {
|
||||
setParams((current) => (JSON.stringify(current) === JSON.stringify(next) ? current : next))
|
||||
}
|
||||
|
||||
/// Ticket 066: text shown under the filter bar when the user hasn't typed a manual filter, so an active preset/sort order is still visible somewhere near the filter input instead of leaving it looking empty/stale.
|
||||
function describeActiveState(activePreset: TimeRangePresetId | null, sortSettings: LogSortSettings): string | null {
|
||||
const parts: string[] = []
|
||||
if (activePreset && activePreset !== 'all') {
|
||||
const preset = TIME_RANGE_PRESETS.find((p) => p.id === activePreset)
|
||||
parts.push(`last ${preset?.label ?? activePreset}`)
|
||||
}
|
||||
if (sortSettings.sortBy) {
|
||||
parts.push(`sorted by ${sortSettings.sortBy}`)
|
||||
}
|
||||
return parts.length > 0 ? parts.join(', ') : null
|
||||
}
|
||||
|
||||
/// The log viewer has its own `LogTimeline`, so it needs its own copy of App's log/timeline polling state instead of a carried-over range.
|
||||
function StandaloneLog() {
|
||||
const [logs, setLogs] = useState<LogEntry[]>([])
|
||||
@@ -311,7 +333,7 @@ function StandaloneLog() {
|
||||
)
|
||||
}
|
||||
try {
|
||||
const [nextBuckets, nextLogs] = await Promise.all([
|
||||
const [nextBuckets, nextLogs, nextParams] = await Promise.all([
|
||||
fetchTimeline({
|
||||
startMs: effectiveRange?.startMs,
|
||||
endMs: effectiveRange?.endMs,
|
||||
@@ -327,10 +349,13 @@ function StandaloneLog() {
|
||||
sortBy: sortSettings.sortBy ?? undefined,
|
||||
sortMissingFallback: sortSettings.sortBy ? sortSettings.sortMissingFallback : undefined,
|
||||
}),
|
||||
// Ticket 058: params can get auto-declared from freshly ingested lines mid-run, not just from a user's own edit.
|
||||
fetchParamDeclarations(),
|
||||
])
|
||||
if (!cancelled) {
|
||||
setBuckets(nextBuckets)
|
||||
setLogs(nextLogs)
|
||||
updateParamsIfChanged(setParams, nextParams)
|
||||
setFilterError(null)
|
||||
}
|
||||
} catch (err) {
|
||||
@@ -352,7 +377,12 @@ function StandaloneLog() {
|
||||
return (
|
||||
<StandalonePage>
|
||||
<div className="flex h-full min-h-0 flex-col">
|
||||
<FilterBar value={filter} onChange={setFilter} error={filterError} />
|
||||
<FilterBar
|
||||
value={filter}
|
||||
onChange={setFilter}
|
||||
error={filterError}
|
||||
activeSummary={describeActiveState(activePreset, sortSettings)}
|
||||
/>
|
||||
<div className="flex min-h-0 flex-1 flex-col">
|
||||
<LogTimeline
|
||||
buckets={buckets}
|
||||
@@ -458,7 +488,7 @@ function FullApp() {
|
||||
)
|
||||
}
|
||||
try {
|
||||
const [nextStats, nextBuckets, nextLogs] = await Promise.all([
|
||||
const [nextStats, nextBuckets, nextLogs, nextParams] = await Promise.all([
|
||||
fetchStats(),
|
||||
fetchTimeline({
|
||||
startMs: effectiveRange?.startMs,
|
||||
@@ -475,11 +505,14 @@ function FullApp() {
|
||||
sortBy: sortSettings.sortBy ?? undefined,
|
||||
sortMissingFallback: sortSettings.sortBy ? sortSettings.sortMissingFallback : undefined,
|
||||
}),
|
||||
// Ticket 058: params can get auto-declared from freshly ingested lines mid-run, not just from a user's own edit.
|
||||
fetchParamDeclarations(),
|
||||
])
|
||||
if (!cancelled) {
|
||||
setStats(nextStats)
|
||||
setBuckets(nextBuckets)
|
||||
setLogs(nextLogs)
|
||||
updateParamsIfChanged(setParams, nextParams)
|
||||
setFilterError(null)
|
||||
}
|
||||
} catch (err) {
|
||||
@@ -508,6 +541,7 @@ function FullApp() {
|
||||
value={filter}
|
||||
onChange={setFilter}
|
||||
error={filterError}
|
||||
activeSummary={describeActiveState(activePreset, sortSettings)}
|
||||
collapseToggle={
|
||||
<>
|
||||
<CollapseToggle collapsed={logViewerCollapsed} onToggle={() => setLogViewerCollapsed(!logViewerCollapsed)} />
|
||||
|
||||
@@ -153,6 +153,9 @@ export function Dashboards({ range, activePreset }: { range: TimeRange | null; a
|
||||
const [saveError, setSaveError] = useState<string | null>(null)
|
||||
// Last-persisted shape for the current draft, so the autosave effect can skip when nothing actually changed.
|
||||
const lastSavedRef = useRef<string | null>(null)
|
||||
// Ticket 064: latest `draft`, read inside the autosave timeout's async callback so a save response can tell
|
||||
// whether the user kept typing after the request went out, instead of unconditionally overwriting `draft`.
|
||||
const draftRef = useRef<Dashboard | null>(draft)
|
||||
const [collapsed, setCollapsed] = useSectionCollapsed('dashboards')
|
||||
|
||||
async function loadDashboards(profileName: string, selectAfterLoad?: string) {
|
||||
@@ -181,11 +184,21 @@ export function Dashboards({ range, activePreset }: { range: TimeRange | null; a
|
||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||
}, [])
|
||||
|
||||
useEffect(() => {
|
||||
draftRef.current = draft
|
||||
}, [draft])
|
||||
|
||||
// Deliberately keyed only on `selectedId`, not `dashboards` — this effect's job is to load a *newly selected*
|
||||
// dashboard's content into `draft`. Keying it on `dashboards` too meant every autosave's own `setDashboards`
|
||||
// call (below) re-ran this effect and clobbered whatever the user had typed since the save was kicked off
|
||||
// (ticket 064). Reading `dashboards` from the closure here still sees the just-loaded array, since
|
||||
// `setSelectedId`/`setDashboards` from `loadDashboards` land in the same render.
|
||||
useEffect(() => {
|
||||
const found = dashboards.find((d) => d.id === selectedId) ?? null
|
||||
setDraft(found)
|
||||
lastSavedRef.current = found ? JSON.stringify(dashboardSaveShape(found)) : null
|
||||
}, [selectedId, dashboards])
|
||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||
}, [selectedId])
|
||||
|
||||
// Skips entirely once the draft matches what's already persisted — covers both re-selecting an unchanged dashboard and this effect's own post-save re-render.
|
||||
useEffect(() => {
|
||||
@@ -198,9 +211,14 @@ export function Dashboards({ range, activePreset }: { range: TimeRange | null; a
|
||||
setSaveStatus('saving')
|
||||
try {
|
||||
const saved = await saveDashboard(activeProfile, dashboardId, dashboardSaveShape(draft))
|
||||
lastSavedRef.current = JSON.stringify(dashboardSaveShape(saved))
|
||||
lastSavedRef.current = snapshot
|
||||
setDashboards((prev) => prev.map((d) => (d.id === saved.id ? saved : d)))
|
||||
// Only reconcile `saved` into `draft` if nothing newer was typed while the request was in flight —
|
||||
// otherwise this stale response would overwrite those in-progress keystrokes (ticket 064).
|
||||
const current = draftRef.current
|
||||
if (current && current.id === dashboardId && JSON.stringify(dashboardSaveShape(current)) === snapshot) {
|
||||
setDraft(saved)
|
||||
}
|
||||
setSaveStatus('saved')
|
||||
setSaveError(null)
|
||||
} catch (err) {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { ReactNode } from 'react'
|
||||
import { useState } from 'react'
|
||||
import { useEffect, useState } from 'react'
|
||||
import { HelpIcon } from './HelpIcon'
|
||||
import { RegexHelpIcon } from './RegexHelpIcon'
|
||||
|
||||
@@ -13,14 +13,22 @@ export function FilterBar({
|
||||
onChange,
|
||||
error,
|
||||
collapseToggle,
|
||||
activeSummary,
|
||||
}: {
|
||||
value: string
|
||||
onChange: (value: string) => void
|
||||
error?: string | null
|
||||
collapseToggle?: ReactNode
|
||||
/** Ticket 066: a read-only textual summary of the active time-range preset / sort order, shown below the input when the user hasn't typed a manual filter — the filter text itself stays the single source of truth for what's actually sent to the server. */
|
||||
activeSummary?: string | null
|
||||
}) {
|
||||
const [draft, setDraft] = useState(value)
|
||||
|
||||
// Resync when `value` changes externally (e.g. a programmatic clear) rather than only reading it once on mount.
|
||||
useEffect(() => {
|
||||
setDraft(value)
|
||||
}, [value])
|
||||
|
||||
return (
|
||||
<form
|
||||
className="flex flex-col gap-1 border-b border-border px-3 py-2"
|
||||
@@ -30,15 +38,21 @@ export function FilterBar({
|
||||
}}
|
||||
>
|
||||
<div className="flex items-center gap-2">
|
||||
<HelpIcon sectionId="filtering" label="Help: Filtering" />
|
||||
<RegexHelpIcon label="Help: regex(...) syntax" />
|
||||
<span className="flex items-center gap-1">
|
||||
<HelpIcon sectionId="filtering" label="Help: filter syntax" />
|
||||
<span className="text-xs text-text-subtle">Filter syntax</span>
|
||||
</span>
|
||||
{collapseToggle}
|
||||
<input
|
||||
className={queryInputClassName}
|
||||
placeholder='regex("error") and ottyl_seq<5 or ottyl_ts>12:00'
|
||||
placeholder='ottyl_ts>-15m and requests[duration]>200 or regex("error")'
|
||||
value={draft}
|
||||
onChange={(e) => setDraft(e.target.value)}
|
||||
/>
|
||||
<span className="flex items-center gap-1" title="Cheat sheet for regex(...) clauses specifically">
|
||||
<RegexHelpIcon label="Help: regex(...) syntax" />
|
||||
<span className="text-xs text-text-faint">regex(...)</span>
|
||||
</span>
|
||||
<button
|
||||
type="submit"
|
||||
className="rounded border border-border-soft px-3 py-1 text-sm text-text-secondary hover:bg-surface-2"
|
||||
@@ -47,6 +61,7 @@ export function FilterBar({
|
||||
</button>
|
||||
</div>
|
||||
{error && <p className="text-xs text-danger-text">{error}</p>}
|
||||
{!error && !draft && activeSummary && <p className="text-xs text-text-subtle">Active: {activeSummary}</p>}
|
||||
</form>
|
||||
)
|
||||
}
|
||||
|
||||
@@ -333,7 +333,11 @@ export function Params({ params, onDeclared }: { params: ParamDeclaration[]; onD
|
||||
? ', elapsed since previous'
|
||||
: p.kind === 'boolean'
|
||||
? ''
|
||||
: `, ${p.json_field_path ? `json field ${p.json_field_path}` : `group ${p.capture_group}`}`}
|
||||
: p.json_field_path
|
||||
? `, json field ${p.json_field_path}`
|
||||
: p.capture_group !== p.name
|
||||
? `, group ${p.capture_group}`
|
||||
: ''}
|
||||
)
|
||||
</button>
|
||||
))}
|
||||
|
||||
Reference in New Issue
Block a user