Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions docs/data-sources/builtin-toolsets/grafanadashboards.md
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,16 @@ A [Grafana service account token](https://grafana.com/docs/grafana/latest/admini
| grafana_get_home_dashboard | Get the home dashboard configuration |
| grafana_get_dashboard_tags | List all tags used across dashboards for categorization |

### Controlling response size

Large dashboards can produce very big JSON payloads. To keep tool outputs readable you can:

- **Limit depth:** `max_depth: 1` returns only the first level of keys, replacing deeper structures with a truncated marker.
- **Filter with jsonpath:** `jsonpath: $.dashboard.panels[*].title` extracts just the panel titles.
- **Filter with jq:** `jq: .dashboard.panels[].id` extracts panel IDs using familiar jq syntax.

Filters run before depth truncation, so you can combine them to precisely scope the response.

## How it Works

### Dashboard Query Extraction
Expand Down
132 changes: 129 additions & 3 deletions holmes/plugins/toolsets/grafana/toolset_grafana.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,11 @@
build_headers,
)
from holmes.plugins.toolsets.utils import toolset_name_for_one_liner
import jq # type: ignore
from jsonpath_ng import parse as jsonpath_parse # type: ignore
import logging

logger = logging.getLogger(__name__)


class GrafanaDashboardConfig(GrafanaConfig):
Expand Down Expand Up @@ -124,6 +129,53 @@ def _make_grafana_request(
)


def _truncate_to_depth(value: Any, max_depth: Optional[int], current_depth: int = 0):
"""Recursively truncate dictionaries/lists beyond the requested depth."""
if max_depth is None or max_depth < 0:
return value

if current_depth >= max_depth:
if isinstance(value, (dict, list)):
return f"...truncated at depth {max_depth}"
return value

if isinstance(value, dict):
return {
key: _truncate_to_depth(sub_value, max_depth, current_depth + 1)
for key, sub_value in value.items()
}
if isinstance(value, list):
return [
_truncate_to_depth(item, max_depth, current_depth + 1) for item in value
]

return value


def _apply_jsonpath_filter(data: Any, expression: str) -> Tuple[Optional[Any], Optional[str]]:
try:
jsonpath_expr = jsonpath_parse(expression)
matches = [match.value for match in jsonpath_expr.find(data)]
if len(matches) == 1:
return matches[0], None
return matches, None
except Exception as exc: # pragma: no cover - defensive
logger.debug("Failed to apply jsonpath filter", exc_info=exc)
return None, f"Invalid jsonpath expression: {exc}"


def _apply_jq_filter(data: Any, expression: str) -> Tuple[Optional[Any], Optional[str]]:
try:
compiled = jq.compile(expression)
results = compiled.input(data).all()
if len(results) == 1:
return results[0], None
return results, None
except Exception as exc: # pragma: no cover - defensive
logger.debug("Failed to apply jq filter", exc_info=exc)
return None, f"Invalid jq filter: {exc}"


class SearchDashboards(BaseGrafanaTool):
def __init__(self, toolset: GrafanaToolset):
super().__init__(
Expand Down Expand Up @@ -249,20 +301,57 @@ def __init__(self, toolset: GrafanaToolset):
type="string",
required=True,
),
"max_depth": ToolParameter(
description="Maximum nesting depth to return from the dashboard JSON (0 returns only top-level keys). Leave empty for full response.",
type="integer",
required=False,
),
"jsonpath": ToolParameter(
description="Optional jsonpath expression to extract specific parts of the dashboard JSON.",
type="string",
required=False,
),
"jq": ToolParameter(
description="Optional jq filter to apply to the dashboard JSON.",
type="string",
required=False,
),
},
)

def _invoke(self, params: dict, context: ToolInvokeContext) -> StructuredToolResult:
uid = params["uid"]
result = self._make_grafana_request(f"/api/dashboards/uid/{uid}", params)

data = result.data
if params.get("jsonpath"):
data, error = _apply_jsonpath_filter(data, params["jsonpath"])
if error:
return StructuredToolResult(
status=StructuredToolResultStatus.ERROR,
error=error,
params=params,
url=result.url,
)
if params.get("jq"):
data, error = _apply_jq_filter(data, params["jq"])
if error:
return StructuredToolResult(
status=StructuredToolResultStatus.ERROR,
error=error,
params=params,
url=result.url,
)

data = _truncate_to_depth(data, params.get("max_depth"))

dashboard_url = _build_grafana_dashboard_url(
self._toolset.grafana_config, uid=uid
)

return StructuredToolResult(
status=result.status,
data=result.data,
data=data,
params=result.params,
url=dashboard_url if dashboard_url else result.url,
)
Expand All @@ -277,11 +366,48 @@ def __init__(self, toolset: GrafanaToolset):
toolset=toolset,
name="grafana_get_home_dashboard",
description="Get the home dashboard using the /api/dashboards/home endpoint",
parameters={},
parameters={
"max_depth": ToolParameter(
description="Maximum nesting depth to return from the dashboard JSON (0 returns only top-level keys). Leave empty for full response.",
type="integer",
required=False,
),
"jsonpath": ToolParameter(
description="Optional jsonpath expression to extract specific parts of the dashboard JSON.",
type="string",
required=False,
),
"jq": ToolParameter(
description="Optional jq filter to apply to the dashboard JSON.",
type="string",
required=False,
),
},
)

def _invoke(self, params: dict, context: ToolInvokeContext) -> StructuredToolResult:
result = self._make_grafana_request("/api/dashboards/home", params)
data = result.data
if params.get("jsonpath"):
data, error = _apply_jsonpath_filter(data, params["jsonpath"])
if error:
return StructuredToolResult(
status=StructuredToolResultStatus.ERROR,
error=error,
params=params,
url=result.url,
)
if params.get("jq"):
data, error = _apply_jq_filter(data, params["jq"])
if error:
return StructuredToolResult(
status=StructuredToolResultStatus.ERROR,
error=error,
params=params,
url=result.url,
)

data = _truncate_to_depth(data, params.get("max_depth"))

config = self._toolset.grafana_config
dashboard_url = None
Expand All @@ -292,7 +418,7 @@ def _invoke(self, params: dict, context: ToolInvokeContext) -> StructuredToolRes

return StructuredToolResult(
status=result.status,
data=result.data,
data=data,
params=result.params,
url=dashboard_url if dashboard_url else None,
)
Expand Down
Loading
Loading