diff --git a/apps/cli/src/cli.ts b/apps/cli/src/cli.ts index 848baece8..2d8f8103b 100644 --- a/apps/cli/src/cli.ts +++ b/apps/cli/src/cli.ts @@ -12,6 +12,7 @@ import { use } from "./commands/config" import { start, stop, reset, checkpoint, restore } from "./commands/server" import { schema } from "./commands/schema" import { archive } from "./commands/archive" +import { deleteCommand } from "./commands/delete" import { update } from "./commands/update" import { OutputFormatSetting } from "./lib/output" @@ -48,6 +49,7 @@ export const cli = Command.make("maple").pipe( reset, checkpoint, restore, + deleteCommand, schema, // Parquet archives (local mode) archive, diff --git a/apps/cli/src/commands/delete.ts b/apps/cli/src/commands/delete.ts new file mode 100644 index 000000000..7840febe3 --- /dev/null +++ b/apps/cli/src/commands/delete.ts @@ -0,0 +1,236 @@ +import { Effect, Option, Schema } from "effect" +import * as Command from "effect/unstable/cli/Command" +import * as Flag from "effect/unstable/cli/Flag" +import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http" +import { readFileSync } from "node:fs" +import { Mode } from "../core/mode" +import { parseTimestampMs, sinceToMs } from "../core/time" +import { CliUsageError } from "../lib/errors" +import { bold, dim, green } from "../lib/style" +import { isLoopbackHostname } from "../lib/local-address" +import { maintenanceTokenPath } from "../server/archives/retention" +import { hasSubject, ScopedDeleteReport, ScopedDeleteRequest } from "../server/scoped-delete" +import { jsonFormatRequested, writeJson } from "./json-output" +import { serverDiscoveryPath } from "../server/store-version" +import { isProcessAlive, LocalStatus, prettyPath, ServerDiscovery } from "./server-args" + +const LOCAL_ONLY = "maple delete removes data from a local store only" + +/** Resolve the local server to delete from; remote mode is refused, never forwarded. */ +export const resolveDeleteBaseUrl = Effect.gen(function* () { + if (process.argv.includes("--remote")) + return yield* new CliUsageError({ + message: `${LOCAL_ONLY}; --remote is not supported`, + hint: "drop --remote and run it against a `maple start` server", + }) + const resolved = yield* (yield* Mode).resolve + if (resolved._tag !== "local") + return yield* new CliUsageError({ + message: `${LOCAL_ONLY}, and the active mode is remote`, + hint: "pass --local to target the running `maple start` server", + }) + return resolved.baseUrl.replace(/\/$/, "") +}) + +/** `--before` as a relative age (`7d`, `12h`) or an absolute UTC timestamp. */ +export const parseBefore = (raw: string, nowMs: number): number | null => { + const age = sinceToMs(raw) + return age === null ? parseTimestampMs(raw) : nowMs - age +} + +const failure = (message: string, hint?: string) => + hint === undefined ? new CliUsageError({ message }) : new CliUsageError({ message, hint }) + +/** Present flags as a request; a blank `--service` or `--namespace` counts as absent. */ +export const buildDeleteRequest = (flags: { + readonly service: string | undefined + readonly namespace: string | undefined + readonly env: string | undefined + readonly beforeMs: number | undefined +}): Effect.Effect => + Effect.gen(function* () { + const present = Object.entries({ + service: flags.service?.trim() || undefined, + namespace: flags.namespace?.trim() || undefined, + env: flags.env, + beforeMs: flags.beforeMs === undefined ? undefined : Math.floor(flags.beforeMs), + }).filter(([, value]) => value !== undefined) + const request = yield* Schema.decodeUnknownEffect(ScopedDeleteRequest)(Object.fromEntries(present)).pipe( + Effect.mapError(() => + failure( + "invalid delete flags", + "--service, --namespace and --env take at most 512 characters, and --before must be after 1970", + ), + ), + ) + if (!hasSubject(request)) + return yield* failure("pass --service, --namespace, or both", "e.g. maple delete --namespace pr-42") + return request + }) + +const decodeDiscovery = Schema.decodeUnknownOption(Schema.fromJsonString(ServerDiscovery)) + +/** + * The maintenance token goes only to the live loopback server its own store's + * discovery file names; any server could otherwise claim another store's dataDir. + */ +export const verifyStoreServer = ( + baseUrl: string, + status: LocalStatus, + deps: { + readonly readDiscovery: (path: string) => string + readonly isAlive: (pid: number) => boolean + } = { readDiscovery: (path) => readFileSync(path, "utf8"), isAlive: isProcessAlive }, +): Effect.Effect => + Effect.gen(function* () { + const target = URL.canParse(baseUrl) ? new URL(baseUrl) : undefined + if (target === undefined || !isLoopbackHostname(target.hostname.replace(/^\[(.*)\]$/, "$1"))) + return yield* failure( + `maple delete only talks to a loopback server, not ${baseUrl}`, + "run it on the machine that runs `maple start`, against 127.0.0.1", + ) + const path = serverDiscoveryPath(status.dataDir) + const discovery = Option.flatMap( + Option.liftThrowable(deps.readDiscovery)(path), + decodeDiscovery, + ) + const port = (url: string) => (URL.canParse(url) ? new URL(url).port : undefined) + const matches = Option.exists( + discovery, + (found) => + found.pid === status.pid && + found.dataDir === status.dataDir && + port(found.url) === target.port && + deps.isAlive(found.pid), + ) + if (!matches) + return yield* failure( + `the server at ${baseUrl} is not the live server ${prettyPath(path)} names`, + "point MAPLE_LOCAL_URL at the `maple start` that owns this store", + ) + }) + +const postDelete = (baseUrl: string, body: ScopedDeleteRequest & { readonly dryRun: boolean }) => + Effect.gen(function* () { + const client = yield* HttpClient.HttpClient + const status = yield* HttpClient.get(`${baseUrl}/local/status`).pipe( + Effect.flatMap(HttpClientResponse.filterStatusOk), + Effect.flatMap(HttpClientResponse.schemaBodyJson(LocalStatus)), + Effect.mapError(() => failure(`no Maple server answered at ${baseUrl}`, "start one with `maple start`")), + ) + yield* verifyStoreServer(baseUrl, status) + // The token sits beside the data dir, so only the machine that owns the store can delete. + const tokenPath = maintenanceTokenPath(status.dataDir) + const token = yield* Effect.try({ + try: () => readFileSync(tokenPath, "utf8").trim(), + catch: () => + failure( + `cannot read the maintenance token at ${prettyPath(tokenPath)}`, + "run maple delete as the same user, on the machine that runs `maple start`", + ), + }) + const response = yield* client + .execute( + HttpClientRequest.post(`${baseUrl}/local/maintenance/delete`).pipe( + HttpClientRequest.setHeader("x-maple-maintenance-token", token), + HttpClientRequest.bodyText(JSON.stringify(body), "application/json"), + ), + ) + .pipe(Effect.mapError((error) => failure(`delete request failed: ${error.message}`))) + const text = yield* response.text.pipe(Effect.orElseSucceed(() => "")) + if (response.status === 404) + return yield* failure("this server predates `maple delete`", "update with `maple update` and restart it") + if (response.status < 200 || response.status >= 300) + return yield* failure(text || `server answered HTTP ${response.status}`) + return yield* Schema.decodeUnknownEffect(Schema.fromJsonString(ScopedDeleteReport))(text).pipe( + Effect.mapError(() => failure("the server returned an unreadable delete report")), + ) + }) + +const describeRequest = (request: ScopedDeleteRequest): string => + [ + ...(request.service === undefined ? [] : [`service ${bold(request.service)}`]), + ...(request.namespace === undefined ? [] : [`namespace ${bold(request.namespace)}`]), + ...(request.env === undefined ? [] : [`env ${bold(request.env === "" ? '""' : request.env)}`]), + ...(request.beforeMs === undefined ? [] : [`before ${bold(new Date(request.beforeMs).toISOString())}`]), + ].join(", ") + +const renderReport = (report: ScopedDeleteReport): string => { + const lines = report.tables + .filter((t) => t.strategy !== "excluded" && (t.rows > 0 || (t.skippedHours ?? 0) > 0)) + .map((t) => + t.strategy === "filter" + ? ` ${t.table.padEnd(36)} ${t.rows} row(s)` + : ` ${t.table.padEnd(36)} ${t.rebuiltHours} hour(s) recomputed` + + ((t.skippedHours ?? 0) > 0 ? dim(`, ${t.skippedHours} past source retention kept`) : ""), + ) + return lines.length === 0 ? ` ${dim("nothing matched")}\n` : `${lines.join("\n")}\n` +} + +export const deleteCommand = Command.make("delete", { + service: Flag.optional( + Flag.String("service").pipe(Flag.withDescription("service.name whose telemetry to delete")), + ), + namespace: Flag.optional( + Flag.String("namespace").pipe( + Flag.withDescription("service.namespace whose telemetry to delete (with --service: both must match)"), + ), + ), + env: Flag.optional( + Flag.String("env").pipe( + Flag.withDescription("Only rows whose deployment.environment(.name) equals this value"), + ), + ), + before: Flag.optional( + Flag.String("before").pipe( + Flag.withDescription( + "Only rows older than this: an age (7d, 12h) or UTC timestamp; floored to the hour", + ), + ), + ), + yes: Flag.Boolean("yes").pipe( + Flag.withDescription("Delete; without it the command only previews what would go"), + Flag.withDefault(false), + ), +}).pipe( + Command.withDescription( + "Delete a service's or namespace's telemetry from the local store: raw tables and every derived rollup (local only)", + ), + Command.withExamples([ + { command: "maple delete --service checkout-pr-42" }, + { command: "maple delete --namespace pr-42 --yes" }, + { command: "maple delete --service api --env preview-17 --yes" }, + { command: "maple delete --service api --before 7d --yes" }, + ]), + Command.withHandler( + Effect.fnUntraced(function* (a) { + const baseUrl = yield* resolveDeleteBaseUrl + const rawBefore = Option.getOrUndefined(a.before) + const beforeMs = rawBefore === undefined ? undefined : parseBefore(rawBefore, Date.now()) + if (beforeMs === null) + return yield* failure(`invalid --before: ${rawBefore}`, "use an age like 7d or a timestamp like 2026-10-01 12:00") + const request = yield* buildDeleteRequest({ + service: Option.getOrUndefined(a.service), + namespace: Option.getOrUndefined(a.namespace), + env: Option.getOrUndefined(a.env), + beforeMs, + }) + const report = yield* postDelete(baseUrl, { ...request, dryRun: !a.yes }) + if (jsonFormatRequested()) yield* writeJson(report) + else + yield* Effect.sync(() => + process.stderr.write( + `${a.yes ? green("✓ deleted") : "would delete"} ${describeRequest(report.request)}\n` + + renderReport(report) + + (a.yes + ? dim( + " Checkpoints taken before now still hold these rows until the next refresh; `maple checkpoint` refreshes now.\n", + ) + : ""), + ), + ) + if (!a.yes && report.tables.some((t) => t.rows > 0)) + return yield* failure("nothing was deleted (preview)", "re-run with --yes to delete") + }), + ), +) diff --git a/apps/cli/src/commands/server-args.ts b/apps/cli/src/commands/server-args.ts index 807ffade1..e5c1f8bc0 100644 --- a/apps/cli/src/commands/server-args.ts +++ b/apps/cli/src/commands/server-args.ts @@ -99,6 +99,8 @@ export const LocalStatus = Schema.Struct({ url: Schema.String, dataDir: Schema.String, lastIngestAtMs: Schema.NullOr(Schema.Number), + /** Absent from binaries that predate `maple delete`. */ + lastDeleteAtMs: Schema.optionalKey(Schema.NullOr(Schema.Number)), }) export type LocalStatus = typeof LocalStatus.Type @@ -133,6 +135,12 @@ export const startedBeforeWrite = (startedAtMs: number, writtenAtMs: number): bo export const ingestedSince = (lastIngestAtMs: number | null | undefined, sinceMs: number): boolean => lastIngestAtMs === undefined || (lastIngestAtMs !== null && lastIngestAtMs >= sinceMs) +/** Did ingest or a scoped delete change the store since `sinceMs`? Unknown status always did. */ +export const storeChangedSince = (status: LocalStatus | undefined, sinceMs: number): boolean => + status === undefined || + ingestedSince(status.lastIngestAtMs, sinceMs) || + (status.lastDeleteAtMs != null && status.lastDeleteAtMs >= sinceMs) + export interface DetachedChildArgs { readonly entry: string | undefined readonly host: string diff --git a/apps/cli/src/commands/server.ts b/apps/cli/src/commands/server.ts index 17edfcf16..5a09e641c 100644 --- a/apps/cli/src/commands/server.ts +++ b/apps/cli/src/commands/server.ts @@ -76,7 +76,7 @@ import { type DirtyStorePolicy, hostedDashboardUrl, hostedUiOrigin, - ingestedSince, + storeChangedSince, isProcessAlive, LocalStatus, mapleCommand, @@ -852,8 +852,8 @@ const takeCheckpointQuietly = ( * * `BACKUP` runs on the server's own JS thread and stalls ingest and queries * while it runs, so an idle server skips it: once this process has a - * checkpoint, a tick with nothing ingested since that checkpoint started - * (`/local/status` `lastIngestAtMs`) has nothing new to protect. Consecutive + * checkpoint, a tick with nothing ingested or deleted since that checkpoint + * started (`/local/status` `lastIngestAtMs`/`lastDeleteAtMs`) has nothing new to protect. Consecutive * failures back the interval off (`checkpointRefreshBackoff`). */ const checkpointRefreshLoop = ( @@ -877,9 +877,8 @@ const checkpointRefreshLoop = ( yield* Effect.sleep(backoff.delay) if (lastTakenAtMs !== undefined) { const status = yield* fetchLocalStatus(statusUrl, "5 seconds") - const lastIngestAtMs = Option.isSome(status) ? status.value.lastIngestAtMs : undefined - if (!ingestedSince(lastIngestAtMs, lastTakenAtMs)) { - debugLog("checkpoint refresh skipped", "nothing ingested since the last checkpoint") + if (!storeChangedSince(Option.getOrUndefined(status), lastTakenAtMs)) { + debugLog("checkpoint refresh skipped", "nothing ingested or deleted since the last checkpoint") continue } } diff --git a/apps/cli/src/server/scoped-delete.ts b/apps/cli/src/server/scoped-delete.ts new file mode 100644 index 000000000..1848a8c0a --- /dev/null +++ b/apps/cli/src/server/scoped-delete.ts @@ -0,0 +1,810 @@ +// Scoped delete for the local store: one service's or one service.namespace's +// telemetry (optionally one deployment.environment and/or everything before an +// hour) leaves the raw tables and every table derived from them. See docs/local-mode.md. + +import { Effect, Option, Schema } from "effect" +import { existsSync } from "node:fs" +import type { Chdb } from "./chdb" +import { decodeJsonEachRow, decodeRowCounts } from "./chdb-rows" +import { durableJson, durableRemove } from "./durable-files" +import { readRealFile } from "./local-token" +import { + type LocalSchemaManifest, + type LocalSchemaObject, + ttlDaysFromDefinition, + viewBody, +} from "./schema-manifest" +import { serviceMapRollupInserts } from "./service-map-rollup" +import { dataDirSidecarPath } from "./store-version" + +const HOUR_MS = 3_600_000 +const DAY_MS = 24 * HOUR_MS + +type Db = Pick + +export class ScopedDeleteError extends Schema.TaggedError()( + "@maple/cli/ScopedDeleteError", + { message: Schema.String, cause: Schema.optionalKey(Schema.Defect()) }, +) {} + +/** The request dimensions a table can carry. */ +export type Dimension = "service" | "namespace" | "env" +const DIMENSIONS: ReadonlyArray = ["service", "namespace", "env"] + +/** Where a dimension lives in a table: a column, or the OTel resource map. */ +export type DimensionSource = { readonly column: string } | { readonly resourceAttributes: string } + +/** `scoped` tables filter on the dimensions they declare and recompute when asked for one they lack. */ +export type TableDeletePlan = + | { + readonly strategy: "scoped" + readonly time: string + /** Columns naming a service; a row matches when any equals `--service`. */ + readonly service: ReadonlyArray | null + readonly namespace: DimensionSource | null + readonly env: DimensionSource | null + /** Rebuild path when the request names an undeclared dimension. */ + readonly recompute: "views" | "service-map-rollup" | null + } + | { readonly strategy: "excluded"; readonly reason: string } + +const RESOURCE: DimensionSource = { resourceAttributes: "ResourceAttributes" } +const DEPLOYMENT_ENV: DimensionSource = { column: "DeploymentEnv" } +const SERVICE_NAMESPACE: DimensionSource = { column: "ServiceNamespace" } + +const raw = (time: string): TableDeletePlan => ({ + strategy: "scoped", + time, + service: ["ServiceName"], + namespace: RESOURCE, + env: RESOURCE, + recompute: null, +}) +const derived = ( + time: string, + dims: { readonly namespace?: DimensionSource; readonly env?: DimensionSource | null } = {}, +): TableDeletePlan => ({ + strategy: "scoped", + time, + service: ["ServiceName"], + namespace: dims.namespace ?? null, + env: dims.env === undefined ? DEPLOYMENT_ENV : dims.env, + recompute: "views", +}) +/** Rows carry no usable service key (no column, or an aggregate), so they are always recomputed. */ +const unkeyed = (time: string): TableDeletePlan => ({ + strategy: "scoped", + time, + service: null, + namespace: null, + env: null, + recompute: "views", +}) +/** Edges come from both services' spans, so a row matches on either end. */ +const edges = (target: string): TableDeletePlan => ({ + strategy: "scoped", + time: "Hour", + service: ["SourceService", target], + namespace: null, + env: DEPLOYMENT_ENV, + recompute: "service-map-rollup", +}) +const NOT_TELEMETRY = "control-plane history, not telemetry; local mode never writes it" +const SESSION_DATA = "browser session data; local OTLP ingest never writes it" + +/** Delete strategy per table name; keys are checked against the schema at runtime. */ +export type TableDeletePlans = Readonly> + +/** Every table in the local schema; `validateDeletePlan` fails when one is missing. */ +export const TABLE_DELETE_PLAN = { + ai_crawler_requests: derived("Timestamp", { env: null }), + ai_trace_index: derived("Timestamp"), + alert_checks: { strategy: "excluded", reason: NOT_TELEMETRY }, + attribute_keys_hourly: unkeyed("Hour"), + attribute_values_hourly: unkeyed("Hour"), + audit_log: { strategy: "excluded", reason: NOT_TELEMETRY }, + error_events: derived("Timestamp"), + error_events_by_time: derived("Timestamp"), + error_fingerprints_minutely: unkeyed("Minute"), + identity_links: { strategy: "excluded", reason: SESSION_DATA }, + logs: raw("Timestamp"), + logs_aggregates_hourly: derived("Hour", { namespace: SERVICE_NAMESPACE }), + metric_catalog: derived("Hour", { env: null }), + metrics_exponential_histogram: raw("TimeUnix"), + metrics_gauge: raw("TimeUnix"), + metrics_histogram: raw("TimeUnix"), + metrics_sum: raw("TimeUnix"), + product_events: derived("Timestamp", { env: null }), + service_address_resolutions_hourly: edges("ResolvedTargetService"), + service_external_edges_hourly: derived("Hour"), + service_map_children: derived("Timestamp"), + service_map_db_edges_hourly: derived("Hour"), + service_map_db_query_shapes_hourly: derived("Hour"), + service_map_edges_hourly: edges("TargetService"), + service_map_edges_hourly_ingest: { + strategy: "excluded", + reason: "Null engine: stores no rows; its view writes service_map_edges_hourly", + }, + service_map_spans: derived("Timestamp"), + service_operations_hourly: derived("Hour"), + service_operations_minutely: derived("Minute"), + service_overview_hourly: derived("Hour", { namespace: SERVICE_NAMESPACE }), + service_overview_minutely: derived("Minute", { namespace: SERVICE_NAMESPACE }), + service_overview_spans: derived("Timestamp", { namespace: SERVICE_NAMESPACE }), + service_platforms_hourly: derived("Hour"), + service_usage: derived("Hour", { env: null }), + session_events: { strategy: "excluded", reason: SESSION_DATA }, + session_replay_events: { strategy: "excluded", reason: SESSION_DATA }, + session_replays: { strategy: "excluded", reason: SESSION_DATA }, + span_metrics_calls_hourly: derived("Hour", { env: null }), + trace_detail_spans: derived("Timestamp", { namespace: RESOURCE, env: RESOURCE }), + trace_facets_hourly: derived("Hour", { namespace: SERVICE_NAMESPACE }), + trace_list_mv: derived("Timestamp", { namespace: SERVICE_NAMESPACE }), + traces: raw("Timestamp"), + traces_aggregates_hourly: derived("Hour"), +} satisfies TableDeletePlans + +const lookup = + (plans: TableDeletePlans) => + (table: string): TableDeletePlan | undefined => + plans[table] +/** The plan for any table name, or undefined when the schema has no such table. */ +export const tablePlan = lookup(TABLE_DELETE_PLAN) + +/** The service map rollup recomputes its two tables from raw spans only. */ +const ROLLUP_SOURCE = "traces" + +type ScopedPlan = Extract + +const declares = (entry: ScopedPlan, dimension: Dimension): boolean => entry[dimension] !== null + +// --------------------------------------------------------------------------- +// Request and SQL. +// --------------------------------------------------------------------------- + +export const ScopedDeleteRequest = Schema.Struct({ + service: Schema.optionalKey(Schema.NonEmptyString.check(Schema.isMaxLength(512))), + namespace: Schema.optionalKey(Schema.NonEmptyString.check(Schema.isMaxLength(512))), + env: Schema.optionalKey(Schema.String.check(Schema.isMaxLength(512))), + /** Exclusive cutoff, epoch ms; the server floors it to the UTC hour. */ + beforeMs: Schema.optionalKey(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))), +}) +export type ScopedDeleteRequest = typeof ScopedDeleteRequest.Type + +/** A delete must name a service or a namespace; env and time only narrow it. */ +export const hasSubject = (request: ScopedDeleteRequest): boolean => + request.service !== undefined || request.namespace !== undefined + +const requestedDimensions = (request: ScopedDeleteRequest): ReadonlyArray => + DIMENSIONS.filter((dimension) => request[dimension] !== undefined) + +/** Bucketed rollups cannot split an hour, so every table uses the same hour cutoff. */ +export const floorToHour = (ms: number): number => Math.floor(ms / HOUR_MS) * HOUR_MS + +const sqlString = (value: string): string => `'${value.replace(/\\/g, "\\\\").replace(/'/g, "\\'")}'` + +const sourceExpression = (source: DimensionSource, dimension: Dimension): string => { + if ("column" in source) return source.column + const map = source.resourceAttributes + return dimension === "namespace" + ? `${map}['service.namespace']` + : `coalesce(nullIf(${map}['deployment.environment.name'], ''), ${map}['deployment.environment'])` +} + +/** `dimension = value` on this table; the caller has checked the table declares it. */ +const dimensionPredicate = (entry: ScopedPlan, dimension: Dimension, value: string): string => { + if (dimension === "service") + return `(${(entry.service ?? []).map((column) => `${column} = ${sqlString(value)}`).join(" OR ")})` + const source = entry[dimension] + return source === null ? "0" : `${sourceExpression(source, dimension)} = ${sqlString(value)}` +} + +const scopePredicates = (entry: ScopedPlan, request: ScopedDeleteRequest, dims: ReadonlyArray) => + dims.map((dimension) => dimensionPredicate(entry, dimension, request[dimension] ?? "")) + +const hourOf = (column: string) => `toUnixTimestamp(toStartOfHour(toDateTime(${column})))` + +/** The rows a request names, or null when the table lacks a requested dimension. */ +const filterPredicate = (entry: ScopedPlan, request: ScopedDeleteRequest): string | null => { + const dims = requestedDimensions(request) + if (!dims.every((dimension) => declares(entry, dimension))) return null + const parts = scopePredicates(entry, request, dims) + if (request.beforeMs !== undefined) + parts.push(`${entry.time} < toDateTime(${Math.floor(request.beforeMs / 1000)})`) + return parts.join(" AND ") +} + +const hourList = (hours: ReadonlyArray) => hours.join(", ") + +// --------------------------------------------------------------------------- +// Schema graph and plan resolution. +// --------------------------------------------------------------------------- + +interface ViewEdge { + readonly name: string + readonly target: string + readonly sources: ReadonlyArray + readonly body: string +} + +const IDENTIFIER = /[A-Za-z_][A-Za-z0-9_]*/g + +const materializedViews = (manifest: LocalSchemaManifest): ReadonlyArray => { + const tables = new Set(manifest.objects.filter((o) => o.kind === "table").map((o) => o.name)) + return manifest.objects + .filter((o) => o.kind === "materialized_view") + .map((o) => { + const body = viewBody(o.definition) ?? "" + const sources = [...body.matchAll(/\bFROM\s+([A-Za-z_][A-Za-z0-9_]*)/gi)] + .map((match) => match[1]!) + .filter((name) => tables.has(name)) + return { + name: o.name, + target: o.definition.match(/\bTO\s+([A-Za-z_][A-Za-z0-9_]*)/i)?.[1] ?? "", + sources: [...new Set(sources)], + body, + } + }) +} + +const fromPattern = (source: string) => new RegExp(`\\bFROM\\s+${source}\\b`, "gi") +const sourceReferences = (body: string, source: string): number => + [...body.matchAll(fromPattern(source))].length + +interface Rebuild { + readonly table: string + readonly time: string + readonly method: "views" | "service-map-rollup" + /** Requested dimensions this table keys on; its clear and re-insert are limited to them. */ + readonly scope: ReadonlyArray + readonly views: ReadonlyArray +} + +interface ResolvedPlan { + readonly filters: ReadonlyArray<{ readonly table: string; readonly predicate: string }> + /** Topological: a rebuild never precedes the rebuild of a table it reads. */ + readonly rebuilds: ReadonlyArray + readonly excluded: ReadonlyArray<{ readonly table: string; readonly reason: string }> + /** Tables that can neither filter nor recompute this request. */ + readonly unsupported: ReadonlyArray +} + +const resolvePlan = ( + manifest: LocalSchemaManifest, + request: ScopedDeleteRequest, + plans: TableDeletePlans = TABLE_DELETE_PLAN, +): ResolvedPlan => { + const views = materializedViews(manifest) + const requested = requestedDimensions(request) + const filters: Array = [] + const pending: Rebuild[] = [] + const excluded: Array = [] + const unsupported: string[] = [] + for (const [table, entry] of Object.entries(plans)) { + if (entry.strategy === "excluded") { + excluded.push({ table, reason: entry.reason }) + continue + } + const predicate = filterPredicate(entry, request) + if (predicate !== null) { + filters.push({ table, predicate }) + continue + } + if (entry.recompute === null) { + unsupported.push(table) + continue + } + pending.push({ + table, + time: entry.time, + method: entry.recompute, + // The rollup recomputes whole hours, so its clear is never narrowed. + scope: + entry.recompute === "views" ? requested.filter((dimension) => declares(entry, dimension)) : [], + views: views + .filter((view) => view.target === table) + .flatMap((view) => { + const source = view.sources[0] + return source !== undefined && plans[source] !== undefined && plans[source]!.strategy !== "excluded" + ? [{ ...view, source }] + : [] + }), + }) + } + const rebuilds: Rebuild[] = [] + while (pending.length > 0) { + const ready = pending.findIndex( + (rebuild) => + !rebuild.views.some((view) => pending.some((other) => other !== rebuild && other.table === view.source)), + ) + rebuilds.push(...pending.splice(ready === -1 ? 0 : ready, 1)) + } + return { filters, rebuilds, excluded, unsupported } +} + +/** Every flag combination the CLI can send, for validation. */ +const DELETE_SCOPES: ReadonlyArray = [ + { service: "s" }, + { service: "s", env: "e" }, + { namespace: "n" }, + { namespace: "n", env: "e" }, + { service: "s", namespace: "n" }, + { service: "s", namespace: "n", env: "e" }, +] + +const scopeFlags = (request: ScopedDeleteRequest) => + requestedDimensions(request) + .map((dimension) => `--${dimension}`) + .join(" ") + +/** + * Everything that must hold for the plan to delete exactly the requested rows + * from the given schema, for every flag combination. Empty means sound. + */ +export const validateDeletePlan = ( + manifest: LocalSchemaManifest, + plans: TableDeletePlans = TABLE_DELETE_PLAN, +): ReadonlyArray => { + const problems: string[] = [] + const tables = new Map(manifest.objects.filter((o) => o.kind === "table").map((o) => [o.name, o])) + const views = materializedViews(manifest) + for (const name of tables.keys()) + if (plans[name] === undefined) problems.push(`${name}: no delete plan for this table`) + for (const name of Object.keys(plans)) + if (!tables.has(name)) problems.push(`${name}: planned but not in the schema`) + + for (const [name, entry] of Object.entries(plans)) { + const table = tables.get(name) + if (table !== undefined && entry.strategy === "scoped") problems.push(...columnProblems(name, table, entry)) + } + + for (const request of DELETE_SCOPES) { + const label = scopeFlags(request) + const plan = resolvePlan(manifest, request, plans) + for (const table of plan.unsupported) problems.push(`${table}: cannot delete ${label} and has no recompute`) + const rebuilt = new Map(plan.rebuilds.map((rebuild) => [rebuild.table, rebuild])) + for (const rebuild of plan.rebuilds) { + if (rebuild.method === "service-map-rollup") continue + if (rebuild.views.length === 0) problems.push(`${rebuild.table}: recomputed for ${label} but no view writes it`) + const target = plans[rebuild.table] + if (rebuild.scope.includes("service") && target?.strategy === "scoped" && target.service?.join() !== "ServiceName") + problems.push(`${rebuild.table}: a service-scoped recompute must key on ServiceName alone`) + for (const view of rebuild.views) { + if (view.sources.length !== 1) problems.push(`${view.name}: recomputing needs exactly one source table`) + if (sourceReferences(view.body, view.source) !== 1) + problems.push(`${view.name}: recomputing needs exactly one FROM ${view.source}`) + const source = plans[view.source] + for (const dimension of rebuild.scope) + if (source?.strategy !== "scoped" || !declares(source, dimension)) + problems.push(`${view.name}: source ${view.source} cannot scope ${dimension} for ${label}`) + } + // Re-inserting fires the views this table feeds; their own recompute must wipe that. + for (const view of views.filter((candidate) => candidate.sources.includes(rebuild.table))) { + const downstream = rebuilt.get(view.target) + if (downstream === undefined) + problems.push(`${view.name}: recomputing ${rebuild.table} for ${label} leaks into ${view.target}`) + else if (!downstream.scope.every((dimension) => rebuild.scope.includes(dimension))) + problems.push(`${view.target}: recompute scope for ${label} is wider than ${rebuild.table}'s`) + } + } + } + + // Rows derived from deletable telemetry must never land in an excluded table. + for (const view of views) { + if (plans[view.target]?.strategy !== "excluded") continue + for (const source of view.sources) + if (plans[source] !== undefined && plans[source]!.strategy !== "excluded") + problems.push(`${view.name}: writes excluded ${view.target} from ${source}`) + } + return [...new Set(problems)] +} + +const columnProblems = (name: string, table: LocalSchemaObject, entry: ScopedPlan): ReadonlyArray => { + const problems: string[] = [] + const columns = new Set(table.columns.map((c) => c.name)) + const keyColumns = new Set(table.orderBy?.match(IDENTIFIER) ?? []) + // Only a plain MergeTree row is one event; any other engine merges rows by key. + const merges = table.engine !== "MergeTree" + const needColumn = (column: string, role: string) => { + if (!columns.has(column)) problems.push(`${name}: ${role} column ${column} does not exist`) + else if (merges && !keyColumns.has(column)) + problems.push(`${name}: ${role} column ${column} is not in the sorting key of ${table.engine}`) + } + needColumn(entry.time, "time") + for (const column of entry.service ?? []) needColumn(column, "service") + for (const dimension of ["namespace", "env"] as const) { + const source = entry[dimension] + if (source === null) continue + if ("column" in source) needColumn(source.column, dimension) + else if (merges) problems.push(`${name}: a merging engine cannot filter ${dimension} from a map`) + else if (!columns.has(source.resourceAttributes)) + problems.push(`${name}: ${dimension} map ${source.resourceAttributes} does not exist`) + } + return problems +} + +// --------------------------------------------------------------------------- +// Execution. +// --------------------------------------------------------------------------- + +export const ScopedDeleteTableReport = Schema.Struct({ + table: Schema.String, + strategy: Schema.Literals(["filter", "rebuild", "excluded"]), + /** Filter: rows deleted. Rebuild: rows in the recomputed hours before recomputing. */ + rows: Schema.Number, + rebuiltHours: Schema.optionalKey(Schema.Number), + /** Hours older than a source's retention: left as-is because they cannot be recomputed. */ + skippedHours: Schema.optionalKey(Schema.Number), + reason: Schema.optionalKey(Schema.String), +}) +export type ScopedDeleteTableReport = typeof ScopedDeleteTableReport.Type + +export const ScopedDeleteReport = Schema.Struct({ + request: ScopedDeleteRequest, + dryRun: Schema.Boolean, + deletedRows: Schema.Number, + tables: Schema.Array(ScopedDeleteTableReport), +}) +export type ScopedDeleteReport = typeof ScopedDeleteReport.Type + +const run = (label: string, f: () => A) => + Effect.try({ + try: f, + catch: (cause) => + new ScopedDeleteError({ + message: `${label}: ${cause instanceof Error ? cause.message : String(cause)}`, + cause, + }), + }) + +const countRows = (db: Db, table: string, predicate: string) => + run(`count ${table}`, () => + Number( + decodeRowCounts( + db.query(`SELECT toString(count()) AS rowCount FROM ${table} WHERE ${predicate}`), + )[0]?.rowCount ?? 0, + ), + ) + +const decodeHours = decodeJsonEachRow(Schema.Struct({ hour: Schema.String })) +const decodeNames = decodeJsonEachRow(Schema.Struct({ name: Schema.String })) + +const distinctHours = (db: Db, table: string, time: string, predicate: string) => + run(`hours of ${table}`, () => + decodeHours( + db.query(`SELECT DISTINCT toString(${hourOf(time)}) AS hour FROM ${table} WHERE ${predicate}`), + ).map((row) => Number(row.hour)), + ) + +const scopedPlan = (table: string): Effect.Effect => { + const entry = tablePlan(table) + return entry?.strategy === "scoped" + ? Effect.succeed(entry) + : Effect.fail(new ScopedDeleteError({ message: `${table} has no scoped delete plan` })) +} + +/** Hours a rebuild must recompute: where its sources lose rows, or are themselves recomputed. */ +const affectedHours = ( + db: Db, + plan: ResolvedPlan, + rebuild: Rebuild, + request: ScopedDeleteRequest, + upstream: ReadonlyMap>, +) => + Effect.gen(function* () { + const sources = + rebuild.method === "service-map-rollup" ? [ROLLUP_SOURCE] : rebuild.views.map((view) => view.source) + const hours = new Set() + for (const source of new Set(sources)) { + const filter = plan.filters.find((candidate) => candidate.table === source) + const found = + filter === undefined + ? (upstream.get(source) ?? []) + : yield* distinctHours(db, source, (yield* scopedPlan(source)).time, filter.predicate) + for (const hour of found) hours.add(hour) + } + if (rebuild.method === "service-map-rollup" && hours.size > 0) { + // Only sealed hours: an open hour is still the rollup loop's to seal. + const sealed = new Set( + yield* distinctHours(db, "service_map_edges_hourly", "Hour", `${hourOf("Hour")} IN (${hourList([...hours])})`), + ) + for (const hour of hours) if (!sealed.has(hour)) hours.delete(hour) + } + return [...hours].sort((a, b) => a - b) + }) + +/** + * An hour is recomputable only while every source still holds all of it. TTL + * drops whole days at merge time, so keep a day of margin; the bundled TTL is + * the shortest a store can have, so this never recomputes from a partial source. + */ +const recomputable = ( + manifest: LocalSchemaManifest, + rebuild: Rebuild, + hours: ReadonlyArray, + nowMs: number, +) => { + const sources = + rebuild.method === "service-map-rollup" ? [ROLLUP_SOURCE] : rebuild.views.map((view) => view.source) + const ttls = sources.map((source) => + ttlDaysFromDefinition(manifest.objects.find((o) => o.name === source)?.definition ?? ""), + ) + const kept: number[] = [] + const skipped: number[] = [] + for (const hour of hours) { + const covered = ttls.every((days) => days === null || hour * 1000 + days * DAY_MS > nowMs + DAY_MS) + ;(covered ? kept : skipped).push(hour) + } + return { kept, skipped } +} + +const rebuildTargetPredicate = ( + entry: ScopedPlan, + rebuild: Rebuild, + request: ScopedDeleteRequest, + hours: ReadonlyArray, +) => [...scopePredicates(entry, request, rebuild.scope), `${hourOf(rebuild.time)} IN (${hourList(hours)})`].join(" AND ") + +const clearHours = (db: Db, rebuild: Rebuild, request: ScopedDeleteRequest, hours: ReadonlyArray) => + Effect.gen(function* () { + const entry = yield* scopedPlan(rebuild.table) + yield* run(`clear ${rebuild.table}`, () => + db.exec( + `ALTER TABLE ${rebuild.table} DELETE WHERE ${rebuildTargetPredicate(entry, rebuild, request, hours)} SETTINGS mutations_sync = 2`, + ), + ) + }) + +const recomputeViaViews = ( + db: Db, + manifest: LocalSchemaManifest, + rebuild: Rebuild, + request: ScopedDeleteRequest, + hours: ReadonlyArray, +) => + Effect.gen(function* () { + if (hours.length === 0) return + yield* clearHours(db, rebuild, request, hours) + // One UTC day per INSERT keeps each recompute's GROUP BY bounded. + const days = new Map() + for (const hour of hours) { + const day = Math.floor(hour / 86_400) + days.set(day, [...(days.get(day) ?? []), hour]) + } + for (const view of rebuild.views) { + const sourceColumns = manifest.objects + .find((o) => o.name === view.source) + ?.columns.map((c) => c.name) + .join(", ") + const source = yield* scopedPlan(view.source) + if (sourceColumns === undefined) + return yield* new ScopedDeleteError({ message: `${view.name}: unknown source ${view.source}` }) + // The view's own output columns, so the INSERT matches by name, not position. + const columns = yield* run(`columns of ${view.name}`, () => + decodeNames( + db.query( + `SELECT name FROM system.columns WHERE database = currentDatabase() AND table = ${sqlString(view.name)} ORDER BY position`, + ), + ).map((row) => row.name), + ) + if (columns.length === 0) + return yield* new ScopedDeleteError({ message: `${view.name}: no columns in system.columns` }) + for (const dayHours of days.values()) { + const scope = [ + ...scopePredicates(source, request, rebuild.scope), + `${hourOf(source.time)} IN (${hourList(dayHours)})`, + ].join(" AND ") + const body = view.body.replace( + fromPattern(view.source), + `FROM (SELECT ${sourceColumns} FROM ${view.source} WHERE ${scope}) AS ${view.source}`, + ) + const list = columns.join(", ") + yield* run(`recompute ${rebuild.table} via ${view.name}`, () => + db.exec(`INSERT INTO ${rebuild.table} (${list}) SELECT ${list} FROM (${body})`), + ) + } + } + }) + +/** Clear whole hours of the rollup's tables, then seal each hour again from surviving spans. */ +const recomputeServiceMap = ( + db: Db, + rebuilds: ReadonlyArray, + request: ScopedDeleteRequest, + hoursByTable: Readonly>>, +) => + Effect.gen(function* () { + const hours = [...new Set(rebuilds.flatMap((rebuild) => hoursByTable[rebuild.table] ?? []))].sort( + (a, b) => a - b, + ) + if (hours.length === 0) return + for (const rebuild of rebuilds) yield* clearHours(db, rebuild, request, hours) + for (const hour of hours) { + const statements = yield* serviceMapRollupInserts(hour * 1000).pipe( + Effect.mapError( + (cause) => new ScopedDeleteError({ message: `service map rollup: ${cause.message}`, cause }), + ), + ) + for (const statement of statements) yield* run("recompute service map", () => db.exec(statement)) + } + }) + +const applyFilters = (db: Db, filters: ResolvedPlan["filters"]) => + Effect.forEach( + filters, + ({ table, predicate }) => + Effect.gen(function* () { + yield* run(`delete from ${table}`, () => + db.exec(`ALTER TABLE ${table} DELETE WHERE ${predicate} SETTINGS mutations_sync = 2`), + ) + const left = yield* countRows(db, table, predicate) + if (left !== 0) + return yield* new ScopedDeleteError({ message: `${left} matching row(s) remain in ${table}` }) + }), + { discard: true }, + ) + +// --------------------------------------------------------------------------- +// Journal: written before the first mutation, removed after the last. +// --------------------------------------------------------------------------- + +const PendingDelete = Schema.Struct({ + formatVersion: Schema.Literal(1), + request: ScopedDeleteRequest, + rebuildHours: Schema.Record(Schema.String, Schema.Array(Schema.Int)), +}) +type PendingDelete = typeof PendingDelete.Type + +export const pendingDeletePath = (dataDir: string): string => + dataDirSidecarPath(dataDir, "maple-pending-delete.json") + +const readPending = (dataDir: string) => + Effect.gen(function* () { + const path = pendingDeletePath(dataDir) + if (!existsSync(path)) return Option.none() + const text = yield* run("read pending delete", () => readRealFile(path, "pending delete journal")) + return Option.some( + yield* Schema.decodeUnknownEffect(Schema.fromJsonString(PendingDelete))(text).pipe( + Effect.mapError( + (cause) => + new ScopedDeleteError({ message: `malformed pending delete journal ${path}`, cause }), + ), + ), + ) + }) + +const writeFile = (label: string, f: () => Promise) => + Effect.tryPromise({ + try: f, + catch: (cause) => new ScopedDeleteError({ message: `${label}: ${String(cause)}`, cause }), + }) + +const assertPlanSound = (manifest: LocalSchemaManifest) => { + const problems = validateDeletePlan(manifest) + return problems.length === 0 + ? Effect.void + : Effect.fail( + new ScopedDeleteError({ message: `scoped delete plan does not match the schema: ${problems.join("; ")}` }), + ) +} + +const assertSubject = (request: ScopedDeleteRequest) => + hasSubject(request) + ? Effect.void + : Effect.fail(new ScopedDeleteError({ message: "a scoped delete needs a service or a namespace" })) + +/** Hours per rebuild, in plan order so recomputed sources feed their dependants. */ +const planHours = ( + db: Db, + manifest: LocalSchemaManifest, + plan: ResolvedPlan, + request: ScopedDeleteRequest, + nowMs: number, +) => + Effect.gen(function* () { + const all = new Map>() + const kept: Record = {} + const skipped: Record = {} + for (const rebuild of plan.rebuilds) { + const hours = yield* affectedHours(db, plan, rebuild, request, all) + all.set(rebuild.table, hours) + const split = recomputable(manifest, rebuild, hours, nowMs) + kept[rebuild.table] = split.kept + skipped[rebuild.table] = split.skipped.length + } + return { kept, skipped } + }) + +/** Apply a journaled request: filters are idempotent, rebuilds take their hours from the journal. */ +const applyPending = (db: Db, manifest: LocalSchemaManifest, dataDir: string, pending: PendingDelete) => + Effect.gen(function* () { + const plan = resolvePlan(manifest, pending.request) + yield* applyFilters(db, plan.filters) + // Views in dependency order, so each recompute wipes what its source's re-insert cascaded. + for (const rebuild of plan.rebuilds) + if (rebuild.method === "views") + yield* recomputeViaViews(db, manifest, rebuild, pending.request, pending.rebuildHours[rebuild.table] ?? []) + yield* recomputeServiceMap( + db, + plan.rebuilds.filter((rebuild) => rebuild.method === "service-map-rollup"), + pending.request, + pending.rebuildHours, + ) + yield* writeFile("remove pending delete journal", () => durableRemove(pendingDeletePath(dataDir))) + }) + +/** Finish a delete an earlier failure or crash left journaled. Run before serving. */ +export const resumePendingDelete = (db: Db, manifest: LocalSchemaManifest, dataDir: string) => + Effect.gen(function* () { + const pending = yield* readPending(dataDir) + if (Option.isNone(pending)) return Option.none() + yield* assertPlanSound(manifest) + yield* assertSubject(pending.value.request) + // A restored checkpoint can bring the rows back; recompute their hours too. + const current = yield* planHours(db, manifest, resolvePlan(manifest, pending.value.request), pending.value.request, Date.now()) + const rebuildHours: Record = {} + for (const table of new Set([...Object.keys(pending.value.rebuildHours), ...Object.keys(current.kept)])) + rebuildHours[table] = [ + ...new Set([...(pending.value.rebuildHours[table] ?? []), ...(current.kept[table] ?? [])]), + ].sort((a, b) => a - b) + yield* applyPending(db, manifest, dataDir, { ...pending.value, rebuildHours }) + return Option.some(pending.value.request) + }) + +/** + * Delete what `request` names. The caller holds exclusive access to the store + * (the server's admission gate), so no ingest or rollup interleaves. + */ +export const runScopedDelete = ( + db: Db, + manifest: LocalSchemaManifest, + dataDir: string, + input: ScopedDeleteRequest, + options: { readonly dryRun: boolean; readonly nowMs?: number }, +): Effect.Effect => + Effect.gen(function* () { + yield* assertPlanSound(manifest) + yield* assertSubject(input) + const nowMs = options.nowMs ?? Date.now() + const request: ScopedDeleteRequest = + input.beforeMs === undefined ? input : { ...input, beforeMs: floorToHour(input.beforeMs) } + if (!options.dryRun) yield* resumePendingDelete(db, manifest, dataDir) + + const plan = resolvePlan(manifest, request) + if (plan.unsupported.length > 0) + return yield* new ScopedDeleteError({ + message: `cannot delete from ${plan.unsupported.join(", ")} for this request`, + }) + const tables: ScopedDeleteTableReport[] = [] + for (const { table, predicate } of plan.filters) + tables.push({ table, strategy: "filter", rows: yield* countRows(db, table, predicate) }) + const hours = yield* planHours(db, manifest, plan, request, nowMs) + for (const rebuild of plan.rebuilds) { + const kept = hours.kept[rebuild.table] ?? [] + const entry = yield* scopedPlan(rebuild.table) + tables.push({ + table: rebuild.table, + strategy: "rebuild", + rows: + kept.length === 0 + ? 0 + : yield* countRows(db, rebuild.table, rebuildTargetPredicate(entry, rebuild, request, kept)), + rebuiltHours: kept.length, + skippedHours: hours.skipped[rebuild.table] ?? 0, + }) + } + for (const { table, reason } of plan.excluded) tables.push({ table, strategy: "excluded", rows: 0, reason }) + tables.sort((a, b) => a.table.localeCompare(b.table)) + const report: ScopedDeleteReport = { + request, + dryRun: options.dryRun, + deletedRows: tables.reduce((sum, t) => sum + (t.strategy === "filter" ? t.rows : 0), 0), + tables, + } + if (options.dryRun) return report + + const pending: PendingDelete = { formatVersion: 1, request, rebuildHours: hours.kept } + yield* writeFile("write pending delete journal", () => durableJson(pendingDeletePath(dataDir), pending)) + yield* applyPending(db, manifest, dataDir, pending) + return report + }) diff --git a/apps/cli/src/server/serve.ts b/apps/cli/src/server/serve.ts index 886036f93..6cf5f9879 100644 --- a/apps/cli/src/server/serve.ts +++ b/apps/cli/src/server/serve.ts @@ -38,7 +38,8 @@ import { decodeTraceRequest, encodeExportResponse, } from "./otlp/proto" -import { CURRENT_LOCAL_SCHEMA, LOCAL_SCHEMA_SQL, SCHEMA_FINGERPRINT } from "./schema-identity" +import { CURRENT_LOCAL_SCHEMA, LOCAL_SCHEMA_MANIFEST, LOCAL_SCHEMA_SQL, SCHEMA_FINGERPRINT } from "./schema-identity" +import { hasSubject, resumePendingDelete, runScopedDelete, ScopedDeleteRequest } from "./scoped-delete" import { assertCurrentPhysicalSchema } from "./schema-physical" import { countArrayRows, @@ -610,6 +611,8 @@ interface ServerStatus { url: string readonly dataDir: string lastIngestAtMs: number | null + /** Last scoped delete that may have changed the store; checkpoint refresh reads it. */ + lastDeleteAtMs: number | null } /** OTLP-ingest request as a `Server`-kind span, mirroring the Rust gateway @@ -943,6 +946,45 @@ const MAX_CHECKPOINT_BODY_BYTES = 4 * 1024 const MAX_PROJECTION_BODY_BYTES = 512 * 1024 const MAX_CONSUMER_BODY_BYTES = 16 * 1024 const MAX_RETIREMENT_BODY_BYTES = 16 * 1024 +const MAX_DELETE_BODY_BYTES = 16 * 1024 + +const ScopedDeleteBody = Schema.Struct({ ...ScopedDeleteRequest.fields, dryRun: Schema.Boolean }) + +/** `maple delete`: removes a service's or namespace's rows from raw and derived tables while admission is closed. */ +const handleScopedDelete = ( + db: Chdb, + dataDir: string, + gate: RequestQuiescenceGate, + status: ServerStatus, + token: string, + req: Request, +): Effect.Effect => { + if (!maintenanceTokenMatches(token, req.headers.get("x-maple-maintenance-token"))) + return Effect.succeed(text("maintenance authorization required", 403)) + return Effect.gen(function* () { + const body = yield* readBoundedJson(req, MAX_DELETE_BODY_BYTES) + const decoded = Schema.decodeUnknownResult(ScopedDeleteBody, { onExcessProperty: "error" })(body) + if (Result.isFailure(decoded)) return text("invalid delete fields", 400) + const { dryRun, ...request } = decoded.success + if (!hasSubject(request)) return text("a delete needs service or namespace", 400) + const work = runScopedDelete(db, LOCAL_SCHEMA_MANIFEST, dataDir, request, { dryRun }) + const report = yield* gate.exclusiveEffect( + dryRun + ? work + : // Even a failed delete may have changed rows, so the next checkpoint refresh must run. + work.pipe(Effect.ensuring(Effect.sync(() => (status.lastDeleteAtMs = Date.now())))), + ) + return json(report) + }).pipe( + Effect.catchTags({ + "@maple/cli/MaintenanceInProgress": (error) => Effect.succeed(text(error.message, 409)), + "@maple/cli/RequestBodyTooLarge": (error) => Effect.succeed(text(error.message, 413)), + "@maple/cli/InvalidJsonBody": () => Effect.succeed(text("invalid JSON body", 400)), + "@maple/cli/ScopedDeleteError": (error) => + Effect.succeed(text(`scoped delete failed: ${error.message}`, 409)), + }), + ) +} /** The chDB half of a checkpoint backup failed. */ class CheckpointBackupFailed extends Schema.TaggedError()( @@ -1315,6 +1357,7 @@ const statusResponse = (status: ServerStatus): Response => url: status.url, dataDir: status.dataDir, lastIngestAtMs: status.lastIngestAtMs, + lastDeleteAtMs: status.lastDeleteAtMs, }) /** The `Bun.serve` fetch handler, closed over the chDB connection. Each ingest @@ -1377,6 +1420,12 @@ const makeFetch = return respond(await runSpan(handleConsumerClaim(gate, consumerToken, req))) if (url.pathname === "/local/eventing/acks") return respond(await runSpan(handleConsumerAcknowledgement(gate, consumerToken, req))) + if (url.pathname === "/local/maintenance/delete") + return respond( + await runSpan( + handleScopedDelete(db, options.dataDir, gate, context.status, maintenanceToken, req), + ), + ) if (url.pathname === "/local/retention/retire") return respond(await handleRetirement(db, authority, gate, maintenanceToken, req)) } @@ -1531,6 +1580,13 @@ export const startServer = ( message: `failed to enforce retired-day authority: ${describeThrown(error)}`, }), }) + // A delete that failed or crashed midway is journaled; finish it before + // serving. A failure keeps the journal for the next `maple delete` to resume. + yield* resumePendingDelete(db, LOCAL_SCHEMA_MANIFEST, options.dataDir).pipe( + Effect.catchTag("@maple/cli/ScopedDeleteError", (error) => + Effect.logWarning(`interrupted scoped delete is still pending: ${error.message}`), + ), + ) const maintenanceToken = yield* Effect.tryPromise({ try: () => ensureMaintenanceToken(options.dataDir), catch: (error) => @@ -1560,7 +1616,7 @@ export const startServer = ( ), ), ) - const status: ServerStatus = { url: "", dataDir: resolve(options.dataDir), lastIngestAtMs: null } + const status: ServerStatus = { url: "", dataDir: resolve(options.dataDir), lastIngestAtMs: null, lastDeleteAtMs: null } const server = yield* Effect.acquireRelease( Effect.try({ try: () => @@ -1622,6 +1678,7 @@ export const __testables = { handleConsumerRegistration, handleCheckpointBackup, handleEventingRead, + handleScopedDelete, handleProjectionActivation, ingest, readBoundedJson, diff --git a/apps/cli/src/server/service-map-rollup.ts b/apps/cli/src/server/service-map-rollup.ts index 43b970fa3..258ad1fa0 100644 --- a/apps/cli/src/server/service-map-rollup.ts +++ b/apps/cli/src/server/service-map-rollup.ts @@ -139,6 +139,18 @@ const sealedHours = (db: LocalServiceMapRollupDeps["db"], oldestHourMs: number, return CH.serviceMapHourSet(yield* decodeRows(decodeHourRows, text, "existing hours")) }) +/** The INSERTs that seal one hour from raw spans: resolutions, then edges (the seal). */ +export const serviceMapRollupInserts = (hourMs: number) => + Effect.gen(function* () { + const params = CH.serviceMapRollupHourParams(ORG_ID, hourMs) + const resolutions = yield* selectBody(CH.serviceMapResolutionsRollupSQL(params), "resolutions") + const edges = yield* selectBody(CH.serviceMapEdgesRollupSQL(params), "edges") + return [ + `INSERT INTO service_address_resolutions_hourly (${RESOLUTION_COLUMNS}) SELECT ${RESOLUTION_COLUMNS} FROM (${resolutions})`, + `INSERT INTO service_map_edges_hourly_ingest (${EDGE_COLUMNS}) SELECT ${EDGE_COLUMNS} FROM (${edges})`, + ] as const + }) + /** * Roll up one hour inside the admission gate. Resolutions are written first and * the edges (the seal) last, so a failure leaves the hour unsealed and the next @@ -151,22 +163,9 @@ const rollupHour = (deps: LocalServiceMapRollupDeps, hourMs: number) => leave === null ? Effect.succeed(false) : Effect.gen(function* () { - const params = CH.serviceMapRollupHourParams(ORG_ID, hourMs) - const resolutions = yield* selectBody( - CH.serviceMapResolutionsRollupSQL(params), - "resolutions", - ) - const edges = yield* selectBody(CH.serviceMapEdgesRollupSQL(params), "edges") - yield* runExec( - deps.db, - `INSERT INTO service_address_resolutions_hourly (${RESOLUTION_COLUMNS}) SELECT ${RESOLUTION_COLUMNS} FROM (${resolutions})`, - "insert address resolutions", - ) - yield* runExec( - deps.db, - `INSERT INTO service_map_edges_hourly_ingest (${EDGE_COLUMNS}) SELECT ${EDGE_COLUMNS} FROM (${edges})`, - "insert service map edges", - ) + const [resolutions, edges] = yield* serviceMapRollupInserts(hourMs) + yield* runExec(deps.db, resolutions, "insert address resolutions") + yield* runExec(deps.db, edges, "insert service map edges") return true }), (leave) => Effect.sync(() => leave?.()), diff --git a/apps/cli/test/delete-command.test.ts b/apps/cli/test/delete-command.test.ts new file mode 100644 index 000000000..c17a78941 --- /dev/null +++ b/apps/cli/test/delete-command.test.ts @@ -0,0 +1,117 @@ +import { afterEach, describe, expect, test } from "bun:test" +import { Effect, Layer } from "effect" +import { buildDeleteRequest, parseBefore, resolveDeleteBaseUrl, verifyStoreServer } from "../src/commands/delete" +import { Mode } from "../src/core/mode" +import { CliUsageError } from "../src/lib/errors" + +const withMode = (resolved: Effect.Success<(typeof Mode.Service)["resolve"]>) => + Effect.runPromise( + Effect.flip(resolveDeleteBaseUrl).pipe( + Effect.provide(Layer.succeed(Mode, { resolve: Effect.succeed(resolved) })), + ), + ) + +describe("maple delete is local-only", () => { + const argv = process.argv + afterEach(() => { + process.argv = argv + }) + + test("refuses --remote before resolving any backend", async () => { + process.argv = [...argv, "--remote"] + const error = await withMode({ _tag: "local", baseUrl: "http://127.0.0.1:4318" }) + expect(error).toBeInstanceOf(CliUsageError) + expect(error.message).toContain("--remote is not supported") + }) + + test("refuses when the active mode resolves to remote", async () => { + const error = await withMode({ _tag: "remote", apiUrl: "https://api.maple.dev", token: "t", orgId: undefined }) + expect(error).toBeInstanceOf(CliUsageError) + expect(error.message).toContain("the active mode is remote") + }) + + test("targets the resolved local server", async () => { + const url = await Effect.runPromise( + resolveDeleteBaseUrl.pipe( + Effect.provide( + Layer.succeed(Mode, { + resolve: Effect.succeed({ _tag: "local" as const, baseUrl: "http://127.0.0.1:4318/" }), + }), + ), + ), + ) + expect(url).toBe("http://127.0.0.1:4318") + }) +}) + +describe("--before", () => { + test("accepts an age or an absolute UTC timestamp", () => { + expect(parseBefore("2h", 10 * 3_600_000)).toBe(8 * 3_600_000) + expect(parseBefore("2026-10-01 12:00", 0)).toBe(Date.UTC(2026, 9, 1, 12)) + expect(parseBefore("yesterday", 0)).toBeNull() + }) +}) + +describe("delete request", () => { + const flags = { service: undefined, namespace: undefined, env: undefined, beforeMs: undefined } + const build = (input: Parameters[0]) => + Effect.runPromise( + Effect.match(buildDeleteRequest(input), { + onFailure: (error) => ({ error: error.message }), + onSuccess: (request) => ({ request }), + }), + ) + + test("needs a service or a namespace", async () => { + expect(await build(flags)).toEqual({ error: "pass --service, --namespace, or both" }) + expect(await build({ ...flags, service: " ", env: "prod" })).toEqual({ + error: "pass --service, --namespace, or both", + }) + }) + + test("reports an out-of-range value as invalid, not as a missing selector", async () => { + expect(await build({ ...flags, service: "a".repeat(513) })).toEqual({ error: "invalid delete flags" }) + expect(await build({ ...flags, namespace: "pr-42", beforeMs: -1 })).toEqual({ error: "invalid delete flags" }) + }) + + test("accepts a namespace alone or with a service, env and cutoff", async () => { + expect(await build({ ...flags, namespace: "pr-42" })).toEqual({ request: { namespace: "pr-42" } }) + expect(await build({ service: " api ", namespace: "pr-42", env: "", beforeMs: 1_000.7 })).toEqual({ + request: { service: "api", namespace: "pr-42", env: "", beforeMs: 1_000 }, + }) + }) +}) + +describe("token goes only to the store's own server", () => { + const status = { service: "maple-local", pid: 4242, version: "x", url: "", dataDir: "/s/data", lastIngestAtMs: null } as const + const discovery = JSON.stringify({ pid: 4242, url: "http://127.0.0.1:4318", dataDir: "/s/data", startedAt: "t" }) + const verify = (baseUrl: string, text: string | undefined, alive = true) => + Effect.runPromise( + Effect.match( + verifyStoreServer(baseUrl, status, { + readDiscovery: () => { + if (text === undefined) throw new Error("ENOENT") + return text + }, + isAlive: () => alive, + }), + { onFailure: (error) => error.message, onSuccess: () => "ok" }, + ), + ) + + test("accepts the live loopback server the discovery file names", async () => { + expect(await verify("http://127.0.0.1:4318", discovery)).toBe("ok") + expect(await verify("http://localhost:4318", discovery)).toBe("ok") + }) + + test("refuses a non-loopback target", async () => { + expect(await verify("http://maple.home.arpa:4318", discovery)).toContain("only talks to a loopback server") + }) + + test("refuses a server that merely claims the store", async () => { + expect(await verify("http://127.0.0.1:9999", discovery)).toContain("is not the live server") + expect(await verify("http://127.0.0.1:4318", discovery.replace("4242", "1"))).toContain("is not the live server") + expect(await verify("http://127.0.0.1:4318", undefined)).toContain("is not the live server") + expect(await verify("http://127.0.0.1:4318", discovery, false)).toContain("is not the live server") + }) +}) diff --git a/apps/cli/test/scoped-delete.test.ts b/apps/cli/test/scoped-delete.test.ts new file mode 100644 index 000000000..473ff1254 --- /dev/null +++ b/apps/cli/test/scoped-delete.test.ts @@ -0,0 +1,319 @@ +import { afterAll, beforeAll, describe, expect, test } from "bun:test" +import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs" +import { homedir, tmpdir } from "node:os" +import { join } from "node:path" +import { Effect, Schema } from "effect" +import schemaSql from "../src/server/schema/local-schema.sql" with { type: "text" } +import { Chdb } from "../src/server/chdb" +import { decodeJsonEachRow } from "../src/server/chdb-rows" +import { buildLocalSchemaManifest } from "../src/server/schema-manifest" +import { LOCAL_SCHEMA_MANIFEST } from "../src/server/schema-identity" +import { serviceMapRollupInserts } from "../src/server/service-map-rollup" +import { + pendingDeletePath, + resumePendingDelete, + runScopedDelete, + TABLE_DELETE_PLAN, + tablePlan, + validateDeletePlan, +} from "../src/server/scoped-delete" + +describe("scoped delete plan", () => { + // Fails when a table is added to the local schema without a delete strategy. + test("classifies every table in the bundled schema soundly", () => { + expect(validateDeletePlan(LOCAL_SCHEMA_MANIFEST)).toEqual([]) + const tables = LOCAL_SCHEMA_MANIFEST.objects.filter((o) => o.kind === "table").map((o) => o.name) + expect(Object.keys(TABLE_DELETE_PLAN).sort()).toEqual([...tables].sort()) + }) + + const withExtra = (ddl: string) => buildLocalSchemaManifest(`${schemaSql}\n\n${ddl}\n`) + + test("refuses a new derived table nobody classified", () => { + const manifest = withExtra( + "CREATE TABLE IF NOT EXISTS traces_by_route (OrgId String, Hour DateTime, ServiceName String, Count UInt64)\nENGINE = SummingMergeTree\nORDER BY (OrgId, Hour);\n\nCREATE MATERIALIZED VIEW IF NOT EXISTS traces_by_route_mv TO traces_by_route AS\nSELECT OrgId, toStartOfHour(toDateTime(Timestamp)) AS Hour, ServiceName, count() AS Count FROM traces GROUP BY OrgId, Hour, ServiceName;", + ) + expect(validateDeletePlan(manifest)).toContain("traces_by_route: no delete plan for this table") + }) + + test("refuses a filter on a merged column that is not a key", () => { + const problems = validateDeletePlan(LOCAL_SCHEMA_MANIFEST, { + ...TABLE_DELETE_PLAN, + error_fingerprints_minutely: { + strategy: "scoped", + time: "Minute", + service: ["ServiceName"], + namespace: null, + env: null, + recompute: "views", + }, + }) + expect(problems.some((p) => p.includes("ServiceName is not in the sorting key"))).toBe(true) + }) + + test("refuses a table that can neither filter nor recompute a namespace", () => { + const problems = validateDeletePlan(LOCAL_SCHEMA_MANIFEST, { + ...TABLE_DELETE_PLAN, + service_map_spans: { + strategy: "scoped", + time: "Timestamp", + service: ["ServiceName"], + namespace: null, + env: { column: "DeploymentEnv" }, + recompute: null, + }, + }) + expect(problems).toContain("service_map_spans: cannot delete --namespace and has no recompute") + }) + + test("refuses recomputing a view source whose dependant is filtered", () => { + const problems = validateDeletePlan(LOCAL_SCHEMA_MANIFEST, { + ...TABLE_DELETE_PLAN, + service_operations_hourly: { + strategy: "scoped", + time: "Hour", + service: ["ServiceName"], + namespace: { resourceAttributes: "ResourceAttributes" }, + env: { column: "DeploymentEnv" }, + recompute: "views", + }, + }) + expect(problems.some((p) => p.includes("leaks into service_operations_hourly"))).toBe(true) + }) + + test("refuses excluding a table that telemetry views write", () => { + const problems = validateDeletePlan(LOCAL_SCHEMA_MANIFEST, { + ...TABLE_DELETE_PLAN, + trace_list_mv: { strategy: "excluded", reason: "test" }, + }) + expect(problems).toContain("trace_list_mv_mv: writes excluded trace_list_mv from traces") + }) +}) + +// Needs a real libchdb; skipped where none is installed. +const libchdbAvailable = + (process.env.MAPLE_LIBCHDB !== undefined && existsSync(process.env.MAPLE_LIBCHDB)) || + existsSync(join(homedir(), ".maple", "bin", "libchdb.so")) || + existsSync(join(homedir(), ".maple", "bin", "libchdb.dylib")) + +const HOUR_MS = 3_600_000 +const iso = (ms: number) => new Date(ms).toISOString().replace("T", " ").replace("Z", "") +const decodeCount = decodeJsonEachRow(Schema.Struct({ n: Schema.String })) + +describe.skipIf(!libchdbAvailable)("scoped delete against chDB", () => { + let root = "" + let dataDir = "" + let db: Chdb + const now = Date.now() + const hourAgo = Math.floor(now / HOUR_MS) * HOUR_MS - HOUR_MS + 60_000 + const twoHoursAgo = hourAgo - HOUR_MS + + const count = (sql: string) => Number(decodeCount(db.query(sql))[0]?.n) + const serviceRows = (table: string, service: string) => { + const plan = tablePlan(table)! + const columns = plan.strategy === "scoped" && plan.service !== null ? plan.service : ["ServiceName"] + return count( + `SELECT toString(count()) AS n FROM ${table} WHERE ${columns.map((c) => `${c} = '${service}'`).join(" OR ")}`, + ) + } + const attributeKey = (key: string) => + count(`SELECT toString(sum(UsageCount)) AS n FROM attribute_keys_hourly WHERE AttributeKey = '${key}'`) + + const span = (at: number, id: string, service: string, env: string, parent = "", kind = "Server") => + `('local', '${iso(at)}', 't-${id}', 's-${id}', '${parent}', 'GET /x', '${kind}', '${service}', 1000000, 'Error', ` + + `map('deployment.environment', '${env}'), map('attr.${service}', 'v', 'shared.key', 'v', 'exception.type', 'E'), ` + + `[toDateTime64('${iso(at)}', 9)], ['exception'], [map('exception.type', 'Boom', 'exception.message', 'm')])` + const log = (at: number, service: string, env: string) => + `('local', '${iso(at)}', '${iso(at).slice(0, 19)}', '${service}', 'body', map('deployment.environment', '${env}'), map('log.${service}', 'v'))` + const metric = (at: number, service: string, env: string) => + `('local', map('deployment.environment', '${env}'), '${service}', 'm.${service}', map('k', 'v'), '${iso(at)}', '${iso(at)}', 1, 1)` + + beforeAll(() => { + root = mkdtempSync(join(tmpdir(), "maple-scoped-delete-")) + dataDir = join(root, "data") + db = Chdb.open({ dataDir, schemaSql }) + const spans = [ + span(twoHoursAgo, "k1", "keep", "prod"), + span(hourAgo, "k2", "keep", "prod"), + span(twoHoursAgo, "d1", "drop", "prod"), + span(hourAgo, "d2", "drop", "pr-1"), + span(hourAgo + 1, "d3", "drop", "prod", "", "Client"), + ] + db.exec( + `INSERT INTO traces (OrgId, Timestamp, TraceId, SpanId, ParentSpanId, SpanName, SpanKind, ServiceName, Duration, StatusCode, ResourceAttributes, SpanAttributes, EventsTimestamp, EventsName, EventsAttributes) VALUES ${spans.join(", ")}`, + ) + db.exec( + `INSERT INTO logs (OrgId, Timestamp, TimestampTime, ServiceName, Body, ResourceAttributes, LogAttributes) VALUES ${[ + log(twoHoursAgo, "keep", "prod"), + log(hourAgo, "drop", "prod"), + log(hourAgo, "drop", "pr-1"), + ].join(", ")}`, + ) + db.exec( + `INSERT INTO metrics_sum (OrgId, ResourceAttributes, ServiceName, MetricName, Attributes, StartTimeUnix, TimeUnix, Value, IsMonotonic) VALUES ${[ + metric(hourAgo, "keep", "prod"), + metric(hourAgo, "drop", "prod"), + metric(hourAgo, "drop", "pr-1"), + ].join(", ")}`, + ) + db.exec( + `INSERT INTO service_map_edges_hourly_ingest (OrgId, Hour, SourceService, TargetService, DeploymentEnv, CallCount) VALUES ('local', toStartOfHour(toDateTime('${iso(hourAgo).slice(0, 19)}')), 'keep', 'drop', 'prod', 3), ('local', toStartOfHour(toDateTime('${iso(hourAgo).slice(0, 19)}')), 'keep', 'db', 'prod', 2)`, + ) + }, 120_000) + + afterAll(() => { + db?.close() + if (root) rmSync(root, { recursive: true, force: true }) + }) + + const deleteNow = (request: Parameters[3], dryRun = false) => + Effect.runPromise(runScopedDelete(db, LOCAL_SCHEMA_MANIFEST, dataDir, request, { dryRun })) + + test("seed reached the derived tables", () => { + for (const table of ["trace_list_mv", "service_overview_hourly", "error_events", "service_usage", "metric_catalog"]) + expect(serviceRows(table, "drop")).toBeGreaterThan(0) + expect(attributeKey("attr.drop")).toBe(3) + }) + + test("dry run counts without deleting", async () => { + const report = await deleteNow({ service: "drop", env: "pr-1" }, true) + expect(report.dryRun).toBe(true) + expect(report.tables.find((t) => t.table === "traces")?.rows).toBe(1) + expect(serviceRows("traces", "drop")).toBe(3) + expect(existsSync(pendingDeletePath(dataDir))).toBe(false) + }) + + test("--env removes one environment and recomputes tables without one", async () => { + const usageBefore = count( + "SELECT toString(sum(TraceCount)) AS n FROM service_usage WHERE ServiceName = 'drop'", + ) + expect(usageBefore).toBe(3) + const report = await deleteNow({ service: "drop", env: "pr-1" }) + expect(report.tables.find((t) => t.table === "service_usage")?.strategy).toBe("rebuild") + expect(count("SELECT toString(count()) AS n FROM traces WHERE ServiceName = 'drop'")).toBe(2) + expect(count("SELECT toString(count()) AS n FROM trace_list_mv WHERE DeploymentEnv = 'pr-1'")).toBe(0) + expect( + count("SELECT toString(sum(TraceCount)) AS n FROM service_usage WHERE ServiceName = 'drop'"), + ).toBe(2) + expect(count("SELECT toString(sum(LogCount)) AS n FROM service_usage WHERE ServiceName = 'drop'")).toBe(1) + expect(attributeKey("attr.drop")).toBe(2) + expect(existsSync(pendingDeletePath(dataDir))).toBe(false) + }) + + test("--service removes every derived row and keeps other services intact", async () => { + const keepBefore = Object.fromEntries( + Object.entries(TABLE_DELETE_PLAN) + .filter(([, plan]) => plan.strategy === "scoped" && plan.service !== null) + .map(([table]) => [table, serviceRows(table, "keep")]), + ) + const sharedBefore = attributeKey("shared.key") + await deleteNow({ service: "drop" }) + for (const [table, plan] of Object.entries(TABLE_DELETE_PLAN)) { + if (plan.strategy !== "scoped" || plan.service === null) continue + expect({ table, rows: serviceRows(table, "drop") }).toEqual({ table, rows: 0 }) + // Edges from keep to drop go too: they were derived from drop's spans. + if (table !== "service_map_edges_hourly") + expect({ table, rows: serviceRows(table, "keep") }).toEqual({ table, rows: keepBefore[table]! }) + } + expect(serviceRows("service_map_edges_hourly", "keep")).toBe(1) + expect(serviceRows("error_fingerprints_minutely", "drop")).toBe(0) + expect(attributeKey("attr.drop")).toBe(0) + expect(attributeKey("log.drop")).toBe(0) + expect(attributeKey("attr.keep")).toBe(2) + expect(attributeKey("shared.key")).toBe(sharedBefore - 2) + }) + + test("--before only removes hours before the floored cutoff", async () => { + const report = await deleteNow({ service: "keep", beforeMs: hourAgo + 30 * 60_000 }) + expect(report.request.beforeMs).toBe(hourAgo - 60_000) + expect(count("SELECT toString(count()) AS n FROM traces WHERE ServiceName = 'keep'")).toBe(1) + expect(count("SELECT toString(count()) AS n FROM trace_list_mv WHERE ServiceName = 'keep'")).toBe(1) + expect(attributeKey("attr.keep")).toBe(1) + }) + + test("a journaled delete resumes on the next run", async () => { + db.exec( + `INSERT INTO traces (OrgId, Timestamp, TraceId, SpanId, ParentSpanId, SpanName, SpanKind, ServiceName, Duration, StatusCode, ResourceAttributes, SpanAttributes, EventsTimestamp, EventsName, EventsAttributes) VALUES ${span(hourAgo, "z1", "zombie", "prod")}`, + ) + writeFileSync( + pendingDeletePath(dataDir), + JSON.stringify({ formatVersion: 1, request: { service: "zombie" }, rebuildHours: {} }), + ) + await Effect.runPromise(resumePendingDelete(db, LOCAL_SCHEMA_MANIFEST, dataDir)) + expect(serviceRows("trace_list_mv", "zombie")).toBe(0) + expect(attributeKey("attr.zombie")).toBe(0) + expect(existsSync(pendingDeletePath(dataDir))).toBe(false) + }) +}) + +describe.skipIf(!libchdbAvailable)("namespace delete against chDB", () => { + let root = "" + let dataDir = "" + let db: Chdb + const hourMs = Math.floor(Date.now() / HOUR_MS) * HOUR_MS - 2 * HOUR_MS + const at = hourMs + 10 * 60_000 + + const count = (sql: string) => Number(decodeCount(db.query(sql))[0]?.n) + // A client span calling a child server span named "api", both in `namespace`. + const call = (id: string, client: string, namespace: string) => { + const resource = (service: string) => + `map('service.name', '${service}', 'service.namespace', '${namespace}', 'deployment.environment', 'prod')` + return ( + `('local', '${iso(at)}', 't-${id}', 'c-${id}', '', 'GET /api', 'Client', '${client}', 5000000, 'Ok', ${resource(client)}, map('server.address', 'api.internal')), ` + + `('local', '${iso(at + 1)}', 't-${id}', 's-${id}', 'c-${id}', 'GET /api', 'Server', 'api', 4000000, 'Error', ${resource("api")}, map('exception.type', 'E'))` + ) + } + const edge = (source: string) => + count( + `SELECT toString(sum(CallCount)) AS n FROM service_map_edges_hourly WHERE SourceService = '${source}' AND TargetService = 'api'`, + ) + + beforeAll(async () => { + root = mkdtempSync(join(tmpdir(), "maple-scoped-delete-ns-")) + dataDir = join(root, "data") + db = Chdb.open({ dataDir, schemaSql }) + db.exec( + `INSERT INTO traces (OrgId, Timestamp, TraceId, SpanId, ParentSpanId, SpanName, SpanKind, ServiceName, Duration, StatusCode, ResourceAttributes, SpanAttributes) VALUES ${call("a", "web", "pr-1")}, ${call("b", "caller", "pr-2")}`, + ) + for (const statement of await Effect.runPromise(serviceMapRollupInserts(hourMs))) db.exec(statement) + }, 120_000) + + afterAll(() => { + db?.close() + if (root) rmSync(root, { recursive: true, force: true }) + }) + + test("removes one namespace and keeps a same-named service in another", async () => { + expect(edge("web")).toBe(1) + expect(edge("caller")).toBe(1) + expect(count("SELECT toString(sum(SpanCount)) AS n FROM service_operations_hourly WHERE ServiceName = 'api'")).toBe(2) + + const report = await Effect.runPromise( + runScopedDelete(db, LOCAL_SCHEMA_MANIFEST, dataDir, { namespace: "pr-1" }, { dryRun: false }), + ) + const strategy = (table: string) => report.tables.find((t) => t.table === table)?.strategy + expect(strategy("trace_list_mv")).toBe("filter") + expect(strategy("error_events")).toBe("rebuild") + expect(strategy("service_map_edges_hourly")).toBe("rebuild") + + expect(count("SELECT toString(count()) AS n FROM traces WHERE ResourceAttributes['service.namespace'] = 'pr-1'")).toBe(0) + expect(count("SELECT toString(count()) AS n FROM traces WHERE ServiceName = 'api'")).toBe(1) + expect(count("SELECT toString(count()) AS n FROM trace_list_mv WHERE ServiceNamespace = 'pr-1'")).toBe(0) + for (const [table, plan] of Object.entries(TABLE_DELETE_PLAN)) { + if (plan.strategy !== "scoped" || plan.service === null) continue + const where = plan.service.map((c) => `${c} = 'web'`).join(" OR ") + expect({ table, rows: count(`SELECT toString(count()) AS n FROM ${table} WHERE ${where}`) }).toEqual({ + table, + rows: 0, + }) + } + // Recomputed from pr-2's surviving api span, including the cascaded hourly rollup. + expect(count("SELECT toString(count()) AS n FROM error_events WHERE ServiceName = 'api'")).toBe(1) + expect(count("SELECT toString(sum(OccurrenceCount)) AS n FROM error_fingerprints_minutely")).toBe(1) + expect(count("SELECT toString(sum(SpanCount)) AS n FROM service_operations_minutely WHERE ServiceName = 'api'")).toBe(1) + expect(count("SELECT toString(sum(SpanCount)) AS n FROM service_operations_hourly WHERE ServiceName = 'api'")).toBe(1) + expect(count("SELECT toString(sum(TraceCount)) AS n FROM service_usage WHERE ServiceName = 'api'")).toBe(1) + expect(edge("web")).toBe(0) + expect(edge("caller")).toBe(1) + expect(existsSync(pendingDeletePath(dataDir))).toBe(false) + }) +}) diff --git a/apps/cli/test/server-args.test.ts b/apps/cli/test/server-args.test.ts index 85f167d45..361c15c23 100644 --- a/apps/cli/test/server-args.test.ts +++ b/apps/cli/test/server-args.test.ts @@ -8,6 +8,7 @@ import { defaultDataDir, defaultLocalUrl, ingestedSince, + storeChangedSince, mapleCommand, parseElapsedSeconds, prettyPath, @@ -241,6 +242,14 @@ describe("checkpoint refresh skip", () => { strictEqual(ingestedSince(1_000, 1_000), true) strictEqual(ingestedSince(undefined, 1_000), true) // status unknown: take it }) + + it("also refreshes after a scoped delete, so restore points drop the deleted rows", () => { + const status = { service: "maple-local", pid: 1, version: "x", url: "", dataDir: "" } as const + strictEqual(storeChangedSince({ ...status, lastIngestAtMs: null }, 1_000), false) + strictEqual(storeChangedSince({ ...status, lastIngestAtMs: null, lastDeleteAtMs: 1_000 }, 1_000), true) + strictEqual(storeChangedSince({ ...status, lastIngestAtMs: 999, lastDeleteAtMs: 999 }, 1_000), false) + strictEqual(storeChangedSince(undefined, 1_000), true) + }) }) describe("prettyPath", () => { diff --git a/docs/local-mode.md b/docs/local-mode.md index 661761159..780fb00fc 100644 --- a/docs/local-mode.md +++ b/docs/local-mode.md @@ -235,6 +235,93 @@ Checkpoints and `maple stop` wait for at most one hour's join. Sealed hours and retired UTC days are skipped. A span that arrives after its hour is sealed is not counted in the map's complete hours. +### Deleting a service's or namespace's data + +`maple delete` removes one service's or one `service.namespace`'s telemetry from +the live store without a full `maple reset`, for example when one shared local +Maple serves many throwaway dev environments: + +```bash +maple delete --service checkout-pr-42 # preview: per-table counts, exits 1 +maple delete --service checkout-pr-42 --yes # delete everything for the service +maple delete --namespace pr-42 --yes # every service in one service.namespace +maple delete --service api --namespace pr-42 --yes # one service within one namespace +maple delete --service api --env preview-17 --yes # one deployment.environment only +maple delete --service api --before 7d --yes # rows older than an age or UTC timestamp +``` + +Pass `--service`, `--namespace`, or both (a row must match all given). `--env` +matches `deployment.environment.name`, falling back to `deployment.environment`, +the same expression the rollups store as `DeploymentEnv`. `--before` is floored +to the UTC hour, because hourly rollups cannot split an hour; the report echoes +the effective cutoff. A namespace delete keeps a same-named service in another +namespace: `api` in `pr-41` survives `--namespace pr-42`. + +The command is local-only: `--remote`, or a mode that resolves to remote, is +refused before anything is sent. It needs a running `maple start`, reads the +maintenance token beside the server's data directory (so only the machine and +user owning the store can delete), and posts to `POST /local/maintenance/delete`. +The read-only `/local/query` path is not involved. The server runs the delete +inside the admission gate's exclusive section, so ingest, queries, the service +map rollup and checkpoint backups wait for it. + +**Which tables.** Every table in the local schema is classified in +`TABLE_DELETE_PLAN` (`apps/cli/src/server/scoped-delete.ts`). A table either is +`excluded` or declares which of the three dimensions (service, namespace, env) +it can filter exactly, plus how to recompute it when a request names one it +lacks. `validateDeletePlan` checks the map against the bundled schema manifest +for every request shape. The server refuses to delete when they disagree, and +`apps/cli/test/scoped-delete.test.ts` fails when a table is added without a +classification. Per request, each table is handled one of three ways: + +- **filter**: the table declares every requested dimension, so + `ALTER TABLE ... DELETE` removes exactly the matching rows. The six raw tables + read namespace and env from `ResourceAttributes` and can always filter. + `trace_list_mv`, `service_overview_*`, `trace_facets_hourly`, + `logs_aggregates_hourly` and `trace_detail_spans` also carry the namespace; + most other rollups carry `ServiceName` and `DeploymentEnv` only. On a merging + engine (Aggregating, Summing, Replacing) every filter column must be in the + sorting key, so a row never mixes scopes. `service_map_edges_hourly` and + `service_address_resolutions_hourly` match the service on either end of the + edge, because an edge is derived from both services' spans. +- **rebuild**: the table lacks a requested dimension. The hours the deleted + source rows covered are cleared, limited to the requested dimensions the table + does key on, and recomputed from the surviving source rows. Most tables are + recomputed by re-running their own materialized-view bodies, one UTC day per + insert, in dependency order: a recomputed source fires its downstream views + (`service_operations_minutely` into `_hourly`, `error_events` into + `error_fingerprints_minutely`), and the downstream recompute that follows + wipes and redoes those hours. The two service map tables are recomputed by + re-running the hourly rollup from raw spans, only for hours it had already + sealed. `attribute_keys_hourly`, `attribute_values_hourly` and + `error_fingerprints_minutely` have no usable service key, so they are always + recomputed. An hour is recomputed only while every source still holds all of + it (bundled TTL plus a day of margin). Older hours are left as they are and + reported as skipped; their TTL removes them. +- **excluded**: tables no local telemetry reaches. `session_events`, + `session_replay_events`, `session_replays` and `identity_links` hold browser + session data that local OTLP ingest never writes. `alert_checks` and + `audit_log` are control-plane history, and `service_map_edges_hourly_ingest` is + a `Null` table that stores nothing. The validator also refuses any + materialized view that writes an excluded table from a deletable one. + +**Crash safety.** Before the first mutation the server durably writes +`maple-pending-delete.json` beside the data directory, holding the request and +the hours each rebuild will recompute. It is removed after the last step. +Filter deletes are idempotent, so a delete that fails midway is finished at the +next start (before the listener binds) or by the next `maple delete`. + +**Checkpoints and archives.** A delete changes only the live store. +`/local/status` reports `lastDeleteAtMs`, and the checkpoint refresh treats a +delete like ingest, so the next tick replaces `current`. Until two refreshes +have rotated `previous` as well, and for pinned or reset-preserved checkpoints, +`maple restore` brings the deleted rows back. Run `maple checkpoint` right after +a delete to refresh `current` at once. Parquet archives are immutable exports +of earlier checkpoints and are not touched; remove an archived day with +`maple archive expire`. Re-archiving a day after a delete exports fewer rows, +which `archive create` refuses without `--allow-shrink`. Retired days hold no +live raw rows, so a delete never recomputes them. + ### Versioned local-store migrations `maple start` never mutates a populated store in place when its schema identity @@ -359,7 +446,7 @@ and every table function that reaches the filesystem or network are refused, and the statement runs with `readonly = 1` plus caps (30s, 4 GB memory, 1M rows, 256 MiB result). A refusal is HTTP 400 with a body starting `read-only query endpoint: `. Internal writes (ingest, checkpoints, archives) -use chDB directly; the native test probes write through `/local/query` with the +and scoped deletes use chDB directly; the native test probes write through `/local/query` with the `x-maple-maintenance-token` header, a single-statement path reserved for them. The **server owns the output FORMAT**. `CH.compile(...)` appends @@ -368,8 +455,9 @@ the statement as JSON rows, and keeps a client's own SETTINGS only where they lower the caps. Clients therefore POST `compiled.sql` verbatim. `GET /local/status` returns `{service: "maple-local", pid, version, url, -dataDir, lastIngestAtMs}`, where `lastIngestAtMs` is when the server last -accepted a non-empty OTLP batch of any signal. The CLI probe and the UI's +dataDir, lastIngestAtMs, lastDeleteAtMs}`, where `lastIngestAtMs` is when the +server last accepted a non-empty OTLP batch of any signal and `lastDeleteAtMs` +is when a `maple delete` last ran (absent on older binaries). The CLI probe and the UI's connection pill use it; `/health` still answers `OK` for shell probes. Browser requests to `/local/*` are accepted only from the same origin, the @@ -545,6 +633,8 @@ maple query 'SELECT ServiceName, count() FROM traces WHERE $__orgFilter AND $__t - `maple slow-traces` and span search report error-or-not rather than the status code, and no span id. - v2 lists cap at 100 rows per page; `maple error` caps sample traces at 20. +- `maple delete` is local-only: it deletes from the local store (see + [Deleting a service's or namespace's data](#deleting-a-services-or-namespaces-data)) and refuses remote mode. ### Seeding data diff --git a/docs/local-telemetry-archives.md b/docs/local-telemetry-archives.md index c49ea537e..f444a7cec 100644 --- a/docs/local-telemetry-archives.md +++ b/docs/local-telemetry-archives.md @@ -29,6 +29,9 @@ and the directory and manifest layouts. live store is a separate, explicit `maple archive retire-live` step. - Archive rehydration into the Maple UI. Historical data is queried in DuckDB, not reloaded into the dashboard. +- Affected by `maple delete`. A scoped delete changes only the live store; + archived generations keep the deleted rows. Re-archiving an already + archived day after a delete needs `--allow-shrink`. - A second always-running database. Archives are files; DuckDB opens them on demand.