Spaces:
Running
Running
File size: 7,867 Bytes
aa89cb5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | import { context, trace, ROOT_CONTEXT } from '@opentelemetry/api';
import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks';
import { BasicTracerProvider, SimpleSpanProcessor, InMemorySpanExporter } from '@opentelemetry/sdk-trace-base';
import { LoggerProvider, SimpleLogRecordProcessor, InMemoryLogRecordExporter } from '@opentelemetry/sdk-logs';
import { MeterProvider } from '@opentelemetry/sdk-metrics';
import { OpenTelemetryRuntimeTelemetry } from '../src/telemetry/opentelemetry';
import { AbstractSQLTeaQLClient, EntitySchema } from '../src/sql/core';
import { SQLiteDriver } from '../src/sql/sqlite';
import { SelectQuery } from '../src/core/ast';
import { UserContext } from '../src/core/context';
test('a broken optional SQL-diagnostic capability cannot change a real query result or expose its error', async () => {
const warning = jest.spyOn(console, 'warn').mockImplementation(() => {});
const diagnostic = jest.fn();
class Client extends AbstractSQLTeaQLClient {
constructor() {super(new SQLiteDriver(':memory:'), {School: {table:'school_data',columns:{
id:{columnName:'id',logicalType:'integer',decode:'string'},version:{columnName:'version',logicalType:'integer',decode:'number'},
}}});}
}
const client = new Client().setDiagnosticSQLLogSink(undefined).setRuntimeTelemetry({
start: () => ({success() {},failure() {}}), recordDiagnostic: diagnostic,
recordsSQLDiagnostics: () => {throw new Error('PRIVATE-CUSTOM-CAPABILITY');},
});
try {
await new UserContext().insertResource('dataService',client).ensureSchema();
const rows = await client.executeQuery(new SelectQuery('School').limit(1)
.comment('inspect empty fixture').purpose('verify fail-open capability'));
expect(rows).toEqual([]);expect(diagnostic).not.toHaveBeenCalled();
expect(JSON.stringify(warning.mock.calls)).not.toContain('PRIVATE-CUSTOM-CAPABILITY');
} finally {await client.close();warning.mockRestore();}
});
test('out-of-order SQL completion retains its own span and does not replace global SDK context', async () => {
const beforeProvider = trace.getTracerProvider();
const beforeContext = context.active();
const manager = new AsyncLocalStorageContextManager().enable();
const spanExporter = new InMemorySpanExporter();
const logExporter = new InMemoryLogRecordExporter();
const tracerProvider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(spanExporter)] });
const loggerProvider = new LoggerProvider({ processors: [new SimpleLogRecordProcessor({ exporter: logExporter })] });
const meterProvider = new MeterProvider();
const telemetry = new OpenTelemetryRuntimeTelemetry(tracerProvider.getTracer('io.teaql.runtime'),
meterProvider.getMeter('io.teaql.runtime'), {}, loggerProvider.getLogger('io.teaql.runtime'), {
active: () => manager.active() === ROOT_CONTEXT ? context.active() : manager.active(),
with: (active, work) => manager.with(active, work),
});
const schemas: Record<string, EntitySchema> = { School: { table: 'school_data', columns: {
id: { columnName: 'id', logicalType: 'integer', decode: 'string' },
version: { columnName: 'version', logicalType: 'integer', decode: 'number' },
} } };
class Driver extends SQLiteDriver {
concurrent = false;
private calls = 0;
private releaseFirst!: () => void;
private readonly secondFinished = new Promise<void>(resolve => { this.releaseFirst = resolve; });
async query(sql: string, values: any[]) {
if (!this.concurrent) return super.query(sql, values);
const call = ++this.calls;
if (call === 1) await this.secondFinished;
const result = await super.query(sql, values);
if (call === 2) this.releaseFirst();
return result;
}
}
const driver = new Driver(':memory:');
class Client extends AbstractSQLTeaQLClient {
constructor() { super(driver, schemas); }
}
const client = new Client().setRuntimeTelemetry(telemetry).setDiagnosticSQLLogSink(undefined);
try {
await new UserContext().insertResource('dataService', client).ensureSchema();
driver.concurrent = true;
await Promise.all([1, 2].map(limit => client.executeQuery(new SelectQuery('School').limit(limit)
.comment(`concurrent ${limit}`).purpose('verify isolated SQL identity'))));
await Promise.all([tracerProvider.forceFlush(), loggerProvider.forceFlush()]);
const logs = logExporter.getFinishedLogRecords().filter(log => log.body === 'TeaQL SQL execution');
const spans = spanExporter.getFinishedSpans();
expect(logs).toHaveLength(2);
expect(logs.map(log => log.attributes['teaql.intent.comment'])).toEqual(['concurrent 2', 'concurrent 1']);
for (const log of logs) {
const provider = spans.find(span => span.spanContext().spanId === log.spanContext?.spanId);
expect(provider?.attributes['teaql.operation.family']).toBe('provider');
const query = spans.find(span => span.spanContext().spanId === provider?.parentSpanContext?.spanId);
expect(query?.attributes['teaql.operation.family']).toBe('query');
expect(provider?.spanContext().traceId).toBe(log.spanContext?.traceId);
}
expect(new Set(logs.map(log => log.spanContext?.traceId)).size).toBe(2);
expect(trace.getTracerProvider()).toBe(beforeProvider);
expect(context.active()).toBe(beforeContext);
} finally {
await client.close();
await Promise.all([tracerProvider.shutdown(), loggerProvider.shutdown(), meterProvider.shutdown()]);
manager.disable();
}
});
test('an existing customer SDK stays installed and receives no TeaQL-owned exports', async () => {
const externalManager = new AsyncLocalStorageContextManager().enable();
context.setGlobalContextManager(externalManager);
const externalExporter = new InMemorySpanExporter();
const externalProvider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(externalExporter)] });
trace.setGlobalTracerProvider(externalProvider);
const originalGlobalProvider = trace.getTracerProvider();
const manager = new AsyncLocalStorageContextManager().enable();
const ownExporter = new InMemorySpanExporter();
const ownProvider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(ownExporter)] });
const meterProvider = new MeterProvider();
const telemetry = new OpenTelemetryRuntimeTelemetry(ownProvider.getTracer('io.teaql.runtime'),
meterProvider.getMeter('io.teaql.runtime'), {}, undefined, {
active: () => manager.active() === ROOT_CONTEXT ? context.active() : manager.active(),
with: (active, work) => manager.with(active, work),
});
const externalSpan = externalProvider.getTracer('customer').startSpan('HTTP request');
try {
await context.with(trace.setSpan(context.active(), externalSpan), async () => {
const scope = telemetry.start({ family: 'query', name: 'Order.list' });
await scope.run!(async () => {}); scope.success();
expect(trace.getTracerProvider()).toBe(originalGlobalProvider);
expect(trace.getSpan(context.active())).toBe(externalSpan);
});
await ownProvider.forceFlush();
expect(ownExporter.getFinishedSpans()).toHaveLength(1);
expect(ownExporter.getFinishedSpans()[0].parentSpanContext?.spanId).toBe(externalSpan.spanContext().spanId);
expect(externalExporter.getFinishedSpans()).toHaveLength(0);
await ownProvider.shutdown(); manager.disable();
externalSpan.end(); await externalProvider.forceFlush();
expect(externalExporter.getFinishedSpans()).toHaveLength(1);
expect(trace.getTracerProvider()).toBe(originalGlobalProvider);
} finally {
externalSpan.end();
await ownProvider.shutdown(); manager.disable();
await externalProvider.shutdown(); await meterProvider.shutdown();
context.disable(); trace.disable(); externalManager.disable();
}
});
|