Skip to content
Merged
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
1 change: 0 additions & 1 deletion apps/alerting/tsconfig.json
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@
"strict": true,
"noFallthroughCasesInSwitch": true,
"noUncheckedSideEffectImports": true,
"baseUrl": ".",
"plugins": [
{
"name": "@effect/language-service",
Expand Down
2 changes: 1 addition & 1 deletion apps/api/src/mcp/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ import type { McpToolError, McpToolRegistrar, McpToolResult } from "./tools/type
interface ToolDefinition {
readonly name: string
readonly description: string
readonly schema: Schema.Top & { readonly DecodingServices: never }
readonly schema: Schema.Decoder<unknown, never>
readonly handler: (params: unknown) => Effect.Effect<McpToolResult, McpToolError, any>
}

Expand Down
95 changes: 55 additions & 40 deletions apps/api/src/mcp/tools/query-data.ts
Original file line number Diff line number Diff line change
Expand Up @@ -249,40 +249,31 @@ function formatQueryResult(
): McpToolResult {
const result = response.result
const metricLabel = metric ?? (source === "metrics" ? "avg" : "count")
const structuredResult = result.kind === "timeseries"
? {
kind: "timeseries" as const,
data: result.data.map((point) => ({
bucket: point.bucket,
series: { ...point.series },
})),
}
: {
kind: "breakdown" as const,
data: result.data.map((item) => ({
name: item.name,
value: item.value,
})),
}

const structuredData = {
tool: "query_data" as const,
data: {
timeRange: { start: startTime, end: endTime },
source,
kind,
metric: metricLabel,
groupBy,
result: structuredResult,
},
}

const lines: string[] = [
`=== ${capitalize(source)} ${capitalize(kind)}: ${metricLabel} ===`,
`Time range: ${startTime} — ${endTime}`,
]

if (result.kind === "timeseries") {
const structuredData = {
tool: "query_data" as const,
data: {
timeRange: { start: startTime, end: endTime },
source,
kind,
metric: metricLabel,
groupBy,
result: {
kind: "timeseries" as const,
data: result.data.map((point) => ({
bucket: point.bucket,
series: { ...point.series },
})),
},
},
}

if (result.data.length === 0) {
lines.push("", "No data points found.")
return { content: createDualContent(lines.join("\n"), structuredData) }
Expand All @@ -305,22 +296,46 @@ function formatQueryResult(
return { content: createDualContent(lines.join("\n"), structuredData) }
}

if (result.data.length === 0) {
lines.push("", "No data found.")
return { content: createDualContent(lines.join("\n"), structuredData) }
}
if (result.kind === "breakdown") {
const structuredData = {
tool: "query_data" as const,
data: {
timeRange: { start: startTime, end: endTime },
source,
kind,
metric: metricLabel,
groupBy,
result: {
kind: "breakdown" as const,
data: result.data.map((item) => ({
name: item.name,
value: item.value,
})),
},
},
}

if (groupBy) lines.push(`Grouped by: ${groupBy}`)
lines.push("")
if (result.data.length === 0) {
lines.push("", "No data found.")
return { content: createDualContent(lines.join("\n"), structuredData) }
}

const headers = ["Name", metricLabel]
const rows = result.data.map((item) => [
item.name,
formatMetricValue(metricLabel, item.value),
])
if (groupBy) lines.push(`Grouped by: ${groupBy}`)
lines.push("")

const headers = ["Name", metricLabel]
const rows = result.data.map((item) => [
item.name,
formatMetricValue(metricLabel, item.value),
])

lines.push(formatTable(headers, rows))
return { content: createDualContent(lines.join("\n"), structuredData) }
}

lines.push(formatTable(headers, rows))
return { content: createDualContent(lines.join("\n"), structuredData) }
// list results
lines.push(`Results: ${result.data.length}`)
return { content: [{ type: "text", text: lines.join("\n") }] }
}

function toInvalidQuerySpecMessage(error: unknown): string {
Expand Down
4 changes: 2 additions & 2 deletions apps/api/src/mcp/tools/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,11 @@ export interface McpToolResult {
}

export interface McpToolRegistrar {
tool<TSchema extends Schema.Top & { readonly DecodingServices: never }>(
tool<TSchema extends Schema.Decoder<unknown, never>>(
name: string,
description: string,
schema: TSchema,
handler: (params: Schema.Schema.Type<TSchema>) => Effect.Effect<McpToolResult, McpToolError, any>,
handler: (params: TSchema["Type"]) => Effect.Effect<McpToolResult, McpToolError, any>,
): void
}

Expand Down
161 changes: 102 additions & 59 deletions apps/api/src/services/QueryEngineService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,7 @@ import {
type QueryEngineSampleCountStrategy,
type QuerySpec,
type TimeseriesPoint,
buildTracesTimeseriesSQL,
buildTracesBreakdownSQL,
type TracesTimeseriesRow,
type TracesBreakdownRow,
CH,
} from "@maple/query-engine"
import {
QueryEngineExecutionError,
Expand Down Expand Up @@ -199,6 +196,7 @@ const validateTraceAttributeFilters = Effect.fn("QueryEngineService.validateTrac
query: QuerySpec,
): Effect.fn.Return<void, QueryEngineValidationError> {
if (query.source !== "traces") return
if (query.kind === "list") return

const details: string[] = []
if (query.groupBy?.includes("attribute") && !query.filters?.groupByAttributeKeys?.length) {
Expand Down Expand Up @@ -313,7 +311,7 @@ function groupTimeSeriesRows<T extends { bucket: string | Date; groupName: strin

function collapseMetricTimeseriesRows(
rows: ReadonlyArray<MetricTimeseriesRow>,
metric: QuerySpec["metric"],
metric: Extract<QuerySpec, { metric: string }>["metric"],
): Array<{ bucket: string; groupName: "all"; value: number }> {
const bucketMap = new Map<
string,
Expand Down Expand Up @@ -396,6 +394,20 @@ const mapTinybirdError = <A, R>(
),
)

/** Compile a CHQuery, execute it via Tinybird SQL, and return typed rows. */
const executeCHQuery = <Output extends Record<string, any>, Params extends Record<string, any>>(
tinybird: Pick<TinybirdServiceShape, "sqlQuery">,
tenant: TenantContext,
query: CH.CHQuery<any, Output, Params>,
params: Params,
context: string,
): Effect.Effect<ReadonlyArray<Output>, QueryEngineExecutionError> => {
const compiled = CH.compile(query, params)
return mapTinybirdError(tinybird.sqlQuery(tenant, compiled.sql), context).pipe(
Effect.map((rows) => compiled.castRows(rows)),
)
}

type QueryEngineTinybird = Pick<
TinybirdServiceShape,
| "sqlQuery"
Expand Down Expand Up @@ -425,7 +437,7 @@ const tracesMetricFieldMap = {
} as const

const tracesAggregateValueForMetric = (
metric: Extract<QuerySpec, { source: "traces" }>["metric"],
metric: Extract<QuerySpec, { source: "traces"; metric: string }>["metric"],
row: {
readonly count: number
readonly avgDuration: number
Expand Down Expand Up @@ -530,34 +542,29 @@ export const makeQueryEngineExecute = (tinybird: QueryEngineTinybird) =>
: undefined

if (request.query.source === "traces" && request.query.kind === "timeseries") {
const sql = buildTracesTimeseriesSQL({
orgId: tenant.orgId,
startTime: request.startTime,
endTime: request.endTime,
bucketSeconds: bucketSeconds!,
metric: request.query.metric,
needsSampling: false,
serviceName: request.query.filters?.serviceName,
spanName: request.query.filters?.spanName,
rootOnly: request.query.filters?.rootSpansOnly,
errorsOnly: request.query.filters?.errorsOnly,
groupBy: request.query.groupBy as string[] | undefined,
groupByAttributeKeys: request.query.filters?.groupByAttributeKeys as string[] | undefined,
environments: request.query.filters?.environments as string[] | undefined,
commitShas: request.query.filters?.commitShas as string[] | undefined,
attributeFilters: request.query.filters?.attributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
resourceAttributeFilters: request.query.filters?.resourceAttributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
apdexThresholdMs: request.query.metric === "apdex" ? request.query.apdexThresholdMs : undefined,
})

const result = yield* mapTinybirdError(
tinybird.sqlQuery(tenant, sql),
const rows = yield* executeCHQuery(
tinybird,
tenant,
CH.tracesTimeseriesQuery({
metric: request.query.metric,
needsSampling: false,
serviceName: request.query.filters?.serviceName,
spanName: request.query.filters?.spanName,
rootOnly: request.query.filters?.rootSpansOnly,
errorsOnly: request.query.filters?.errorsOnly,
groupBy: request.query.groupBy as string[] | undefined,
groupByAttributeKeys: request.query.filters?.groupByAttributeKeys as string[] | undefined,
environments: request.query.filters?.environments as string[] | undefined,
commitShas: request.query.filters?.commitShas as string[] | undefined,
attributeFilters: request.query.filters?.attributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
resourceAttributeFilters: request.query.filters?.resourceAttributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
apdexThresholdMs: request.query.metric === "apdex" ? request.query.apdexThresholdMs : undefined,
}),
{ orgId: tenant.orgId, startTime: request.startTime, endTime: request.endTime, bucketSeconds: bucketSeconds! },
"Failed to execute traces timeseries query",
)

const field = tracesMetricFieldMap[request.query.metric]
const rows = result as unknown as ReadonlyArray<TracesTimeseriesRow>

return new QueryEngineExecuteResponse({
result: {
kind: "timeseries",
Expand Down Expand Up @@ -644,36 +651,32 @@ export const makeQueryEngineExecute = (tinybird: QueryEngineTinybird) =>
}

if (request.query.source === "traces" && request.query.kind === "breakdown") {
const sql = buildTracesBreakdownSQL({
orgId: tenant.orgId,
startTime: request.startTime,
endTime: request.endTime,
metric: request.query.metric,
groupBy: request.query.groupBy,
groupByAttributeKey:
request.query.groupBy === "attribute"
? (request.query.filters?.groupByAttributeKeys as string[] | undefined)?.[0]
: undefined,
limit: request.query.limit,
serviceName: request.query.filters?.serviceName,
spanName: request.query.filters?.spanName,
rootOnly: request.query.filters?.rootSpansOnly,
errorsOnly: request.query.filters?.errorsOnly,
environments: request.query.filters?.environments as string[] | undefined,
commitShas: request.query.filters?.commitShas as string[] | undefined,
attributeFilters: request.query.filters?.attributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
resourceAttributeFilters: request.query.filters?.resourceAttributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
apdexThresholdMs: request.query.metric === "apdex" ? request.query.apdexThresholdMs : undefined,
})

const result = yield* mapTinybirdError(
tinybird.sqlQuery(tenant, sql),
const rows = yield* executeCHQuery(
tinybird,
tenant,
CH.tracesBreakdownQuery({
metric: request.query.metric,
groupBy: request.query.groupBy,
groupByAttributeKey:
request.query.groupBy === "attribute"
? (request.query.filters?.groupByAttributeKeys as string[] | undefined)?.[0]
: undefined,
limit: request.query.limit,
serviceName: request.query.filters?.serviceName,
spanName: request.query.filters?.spanName,
rootOnly: request.query.filters?.rootSpansOnly,
errorsOnly: request.query.filters?.errorsOnly,
environments: request.query.filters?.environments as string[] | undefined,
commitShas: request.query.filters?.commitShas as string[] | undefined,
attributeFilters: request.query.filters?.attributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
resourceAttributeFilters: request.query.filters?.resourceAttributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
apdexThresholdMs: request.query.metric === "apdex" ? request.query.apdexThresholdMs : undefined,
}),
{ orgId: tenant.orgId, startTime: request.startTime, endTime: request.endTime },
"Failed to execute traces breakdown query",
)

const field = tracesMetricFieldMap[request.query.metric]
const rows = result as unknown as ReadonlyArray<TracesBreakdownRow>

return new QueryEngineExecuteResponse({
result: {
kind: "breakdown",
Expand Down Expand Up @@ -743,6 +746,46 @@ export const makeQueryEngineExecute = (tinybird: QueryEngineTinybird) =>
})
}

if (request.query.source === "traces" && request.query.kind === "list") {
const rows = yield* executeCHQuery(
tinybird,
tenant,
CH.tracesListQuery({
limit: request.query.limit,
serviceName: request.query.filters?.serviceName,
spanName: request.query.filters?.spanName,
rootOnly: request.query.filters?.rootSpansOnly,
errorsOnly: request.query.filters?.errorsOnly,
environments: request.query.filters?.environments as string[] | undefined,
commitShas: request.query.filters?.commitShas as string[] | undefined,
attributeFilters: request.query.filters?.attributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" | "gt" | "gte" | "lt" | "lte" | "contains" }> | undefined,
resourceAttributeFilters: request.query.filters?.resourceAttributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" | "gt" | "gte" | "lt" | "lte" | "contains" }> | undefined,
}),
{ orgId: tenant.orgId, startTime: request.startTime, endTime: request.endTime },
"Failed to execute traces list query",
)

return new QueryEngineExecuteResponse({
result: {
kind: "list",
source: "traces",
data: rows.map((row) => ({
traceId: row.traceId,
timestamp: String(row.timestamp),
spanId: row.spanId,
serviceName: row.serviceName,
spanName: row.spanName,
durationMs: Number(row.durationMs),
statusCode: row.statusCode,
spanKind: row.spanKind,
hasError: Number(row.hasError) === 1,
spanAttributes: row.spanAttributes ?? {},
resourceAttributes: row.resourceAttributes ?? {},
})),
},
})
}

return yield* new QueryEngineValidationError({
message: "Unsupported query",
details: ["This source/kind combination is not supported"],
Expand Down Expand Up @@ -886,7 +929,7 @@ export const makeQueryEngineEvaluateGrouped = (tinybird: QueryEngineTinybird) =>
commit_shas: request.query.filters?.commitShas?.join(","),
...buildAttributeFilterParams(request.query.filters),
apdex_threshold_ms:
request.query.metric === "apdex" ? request.query.apdexThresholdMs : undefined,
(request.query as Extract<QuerySpec, { source: "traces"; metric: string }>).metric === "apdex" ? (request.query as Extract<QuerySpec, { source: "traces"; metric: string }>).apdexThresholdMs : undefined,
}),
"Failed to evaluate grouped traces alert query",
)
Expand All @@ -895,7 +938,7 @@ export const makeQueryEngineEvaluateGrouped = (tinybird: QueryEngineTinybird) =>
const sampleCount = Number(row.count ?? 0)
return {
groupKey: row.serviceName,
value: sampleCount > 0 ? tracesAggregateValueForMetric(request.query.metric as Extract<QuerySpec, { source: "traces" }>["metric"], row) : null,
value: sampleCount > 0 ? tracesAggregateValueForMetric((request.query as Extract<QuerySpec, { source: "traces"; metric: string }>).metric, row) : null,
sampleCount,
hasData: sampleCount > 0,
}
Expand Down Expand Up @@ -934,7 +977,7 @@ export const makeQueryEngineEvaluateGrouped = (tinybird: QueryEngineTinybird) =>
const sampleCount = Number(row.dataPointCount ?? 0)
return {
groupKey: row.serviceName,
value: sampleCount > 0 ? metricsAggregateValueForMetric(request.query.metric as Extract<QuerySpec, { source: "metrics" }>["metric"], row) : null,
value: sampleCount > 0 ? metricsAggregateValueForMetric((request.query as Extract<QuerySpec, { source: "metrics" }>).metric, row) : null,
sampleCount,
hasData: sampleCount > 0,
}
Expand Down
Loading
Loading