Resolve tickets 067-069: named sessions, MCP server, embedded SKILL.md
Adds a sessions/<name>.json registry (068) so a running ottyl instance can be discovered by name; an `ottyl mcp --session <name>` subcommand (067) that starts a short-lived MCP server over stdio (via the official rmcp SDK), proxying query_logs/declare_param/declare_metric/create_dashboard/ add_panel/get_dashboard tool calls onto the existing HTTP API; and a SKILL.md (069), embedded into the binary and served at GET /SKILL.md, that documents how an agent connects to and drives a running instance. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01A1ZJ3SrayzLzVndhVq9BmS
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
de9ba6f01b
commit
c77404ba4e
@@ -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('/');
|
||||
|
||||
+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"]);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user