diff --git a/src/services/TelemetryStore.ts b/src/services/TelemetryStore.ts index dbae579..cb59a36 100644 --- a/src/services/TelemetryStore.ts +++ b/src/services/TelemetryStore.ts @@ -798,11 +798,16 @@ export const makeTelemetryStoreEffect = (db: TelemetryDatabase, opts: TelemetryS INSERT INTO span_operation_fts (trace_id, span_id, operation_name) SELECT json_extract(value, '$[0]'), json_extract(value, '$[1]'), json_extract(value, '$[2]') FROM json_each(?) `) - const updateSpanOperationSearchMany = (operations: ReadonlyArray) => { - if (operations.length === 0) return - const list = JSON.stringify(operations) - deleteSpanOperationSearch.run(list) - insertSpanOperationSearch.run(list) + // The operation index names its span only in UNINDEXED columns, so deleting by span scans + // the whole index. Only a span stored before has a row to replace; a new span, almost every + // one, is only inserted. Deleting for every span made each export cost time in proportion + // to everything already stored. + const spanExists = db.query(`SELECT 1 AS found FROM spans WHERE trace_id = ? AND span_id = ?`) + const deleteSpanOperationSearchMany = (spans: ReadonlyArray) => { + if (spans.length > 0) deleteSpanOperationSearch.run(JSON.stringify(spans)) + } + const insertSpanOperationSearchMany = (operations: ReadonlyArray) => { + if (operations.length > 0) insertSpanOperationSearch.run(JSON.stringify(operations)) } const insertLogAttributes = db.query(`INSERT INTO log_attributes (log_id, key, value) SELECT ?, key, value FROM json_each(?)`) const insertLogAttributesMany = (logId: number, attributes: Readonly>) => { @@ -1050,6 +1055,7 @@ export const makeTelemetryStoreEffect = (db: TelemetryDatabase, opts: TelemetryS const transaction = db.transaction((request: OtlpTraceExportRequest) => { const touchedTraceIds = new Set() const touchedOperations: Array = [] + const replacedSpans: Array = [] for (const resourceSpans of request.resourceSpans ?? []) { const resourceAttributes = attributeMap(resourceSpans.resource?.attributes) const serviceName = resourceAttributes["service.name"] || resourceAttributes["service_name"] || "unknown" @@ -1072,6 +1078,7 @@ export const makeTelemetryStoreEffect = (db: TelemetryDatabase, opts: TelemetryS attributes: attributeMap(event.attributes), })) + if (spanExists.get(traceId, spanId) !== null) replacedSpans.push([traceId, spanId]) insertSpan.run( traceId, spanId, @@ -1098,8 +1105,11 @@ export const makeTelemetryStoreEffect = (db: TelemetryDatabase, opts: TelemetryS } try { const BATCH_SIZE = 500 + for (let offset = 0; offset < replacedSpans.length; offset += BATCH_SIZE) { + deleteSpanOperationSearchMany(replacedSpans.slice(offset, offset + BATCH_SIZE)) + } for (let offset = 0; offset < touchedOperations.length; offset += BATCH_SIZE) { - updateSpanOperationSearchMany(touchedOperations.slice(offset, offset + BATCH_SIZE)) + insertSpanOperationSearchMany(touchedOperations.slice(offset, offset + BATCH_SIZE)) } } catch { // FTS is optional. diff --git a/src/telemetry.test.ts b/src/telemetry.test.ts index aa1e3a0..d94aec4 100644 --- a/src/telemetry.test.ts +++ b/src/telemetry.test.ts @@ -1125,4 +1125,31 @@ describe("motel telemetry store", () => { expect(services).toContain("active-child-only") }) + + it("replaces a re-exported span's operation in the search index once", async () => { + const span = (name: string) => ({ + resourceSpans: [{ + resource: { attributes: [{ key: "service.name", value: { stringValue: "reexport" } }] }, + scopeSpans: [{ spans: [{ traceId: "reexport-trace", spanId: "reexport-span", name, startTimeUnixNano: String(BigInt(Date.now()) * 1_000_000n), endTimeUnixNano: String(BigInt(Date.now() + 1) * 1_000_000n) }] }], + }], + }) + const search = (operation: string) => + Effect.flatMap(TelemetryStore, (store) => store.searchSpans({ serviceName: "reexport", operation })) + const [first, again, renamed, previous] = await storeRuntime.runPromise( + Effect.gen(function* () { + const store = yield* TelemetryStore + yield* store.ingestTraces(span("reexport.first")) + const first = yield* search("reexport.first") + // An exporter resends an export the collector stored but did not acknowledge. + yield* store.ingestTraces(span("reexport.first")) + const again = yield* search("reexport.first") + yield* store.ingestTraces(span("reexport.renamed")) + return [first, again, yield* search("reexport.renamed"), yield* search("reexport.first")] as const + }).pipe(Effect.provideService(References.MinimumLogLevel, "None")), + ) + expect(first).toHaveLength(1) + expect(again).toHaveLength(1) + expect(renamed.map((item) => item.span.operationName)).toEqual(["reexport.renamed"]) + expect(previous).toHaveLength(0) + }) })