diff --git a/.github/workflows/deploy-pr-preview.yml b/.github/workflows/deploy-pr-preview.yml index 2139d183a..0e2d0bbee 100644 --- a/.github/workflows/deploy-pr-preview.yml +++ b/.github/workflows/deploy-pr-preview.yml @@ -59,6 +59,13 @@ jobs: - name: Install dependencies run: bun install --frozen-lockfile + # Tinybird's TypeScript resource generator loads @maple/domain from source, + # whose span-display helpers import this package through its dist exports. + # A clean runner has no dist/ until the workspace dependency is built. + - name: Build Tinybird TypeScript workspace dependencies + if: ${{ github.event.action != 'closed' }} + run: bun turbo build --filter=@maple-dev/clickhouse-builder + # Needed for both branch create (deploy) and rm (close). `tb` is the # real Tinybird CLI (branch commands live there, not in `bunx tinybird`). - name: Install Tinybird CLI diff --git a/apps/api/src/lib/WarehouseQueryService.test.ts b/apps/api/src/lib/WarehouseQueryService.test.ts index 6f680dd2b..281244ee6 100644 --- a/apps/api/src/lib/WarehouseQueryService.test.ts +++ b/apps/api/src/lib/WarehouseQueryService.test.ts @@ -267,30 +267,23 @@ describe("WarehouseQueryService raw-SQL provider routing", () => { describe("bounded Tinybird response fetch", () => { it("accepts an exact-boundary response and aborts one byte over", async () => { - const realFetch = globalThis.fetch + const exact = await __testables.boundedResponseFetch( + MAX_RAW_SQL_RESULT_BYTES, + (async () => new Response(new Uint8Array(MAX_RAW_SQL_RESULT_BYTES))) as typeof fetch, + )("https://api.tinybird.example/v0/sql") + assert.strictEqual((await exact.arrayBuffer()).byteLength, MAX_RAW_SQL_RESULT_BYTES) + + let thrown: unknown try { - globalThis.fetch = (async () => - new Response(new Uint8Array(MAX_RAW_SQL_RESULT_BYTES))) as typeof fetch - const exact = await __testables.boundedResponseFetch(MAX_RAW_SQL_RESULT_BYTES)( - "https://api.tinybird.example/v0/sql", - ) - assert.strictEqual((await exact.arrayBuffer()).byteLength, MAX_RAW_SQL_RESULT_BYTES) - - globalThis.fetch = (async () => - new Response(new Uint8Array(MAX_RAW_SQL_RESULT_BYTES + 1))) as typeof fetch - let thrown: unknown - try { - await __testables.boundedResponseFetch(MAX_RAW_SQL_RESULT_BYTES)( - "https://api.tinybird.example/v0/sql", - ) - } catch (error) { - thrown = error - } - assert.instanceOf(thrown, Error) - assert.match((thrown as Error).message, /5000000 encoded bytes/) - } finally { - globalThis.fetch = realFetch + await __testables.boundedResponseFetch( + MAX_RAW_SQL_RESULT_BYTES, + (async () => new Response(new Uint8Array(MAX_RAW_SQL_RESULT_BYTES + 1))) as typeof fetch, + )("https://api.tinybird.example/v0/sql") + } catch (error) { + thrown = error } + assert.instanceOf(thrown, Error) + assert.match((thrown as Error).message, /5000000 encoded bytes/) }) }) @@ -652,8 +645,7 @@ describe("createTinybirdSdkSqlClient.insert wire framing (the production insert auth?: string body: string }> = [] - const realFetch = globalThis.fetch - globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const requestFetch = (async (input: RequestInfo | URL, init?: RequestInit) => { const headers = (init?.headers ?? {}) as Record captured.push({ url: String(input), @@ -665,12 +657,8 @@ describe("createTinybirdSdkSqlClient.insert wire framing (the production insert return new Response("", { status: 202 }) }) as typeof fetch - try { - const client = __testables.createTinybirdSdkSqlClient(tbConfig) - await client.insert("traces", [{ trace_id: "a" }, { trace_id: "b" }]) - } finally { - globalThis.fetch = realFetch - } + const client = __testables.createTinybirdSdkSqlClient(tbConfig, requestFetch) + await client.insert("traces", [{ trace_id: "a" }, { trace_id: "b" }]) assert.strictEqual(captured.length, 1) const req = captured[0]! @@ -685,18 +673,13 @@ describe("createTinybirdSdkSqlClient.insert wire framing (the production insert it("no-ops on an empty row set (no request issued)", async () => { let calls = 0 - const realFetch = globalThis.fetch - globalThis.fetch = (async () => { + const requestFetch = (async () => { calls++ return new Response("", { status: 202 }) }) as typeof fetch - try { - const client = __testables.createTinybirdSdkSqlClient(tbConfig) - await client.insert("traces", []) - } finally { - globalThis.fetch = realFetch - } + const client = __testables.createTinybirdSdkSqlClient(tbConfig, requestFetch) + await client.insert("traces", []) assert.strictEqual(calls, 0) }) @@ -715,24 +698,21 @@ describe("createTinybirdSdkSqlClient.sql FORMAT normalization", () => { const captureSql = async (sql: string): Promise => { const sent: string[] = [] - const realFetch = globalThis.fetch - globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { - const url = String(input) - const fromParam = new URL(url).searchParams.get("q") - const body = typeof init?.body === "string" ? init.body : "" - sent.push(fromParam ?? body) + const requestFetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = new URL(String(input)) + if (url.pathname.endsWith("/v0/sql")) { + const fromParam = url.searchParams.get("q") + const body = typeof init?.body === "string" ? init.body : "" + sent.push(fromParam ?? body) + } return new Response(JSON.stringify({ meta: [], data: [], rows: 0 }), { status: 200, headers: { "Content-Type": "application/json" }, }) }) as typeof fetch - try { - const client = __testables.createTinybirdSdkSqlClient(tbConfig) - await client.sql(sql, undefined) - } finally { - globalThis.fetch = realFetch - } + const client = __testables.createTinybirdSdkSqlClient(tbConfig, requestFetch) + await client.sql(sql, undefined) assert.strictEqual(sent.length, 1) return sent[0]! diff --git a/apps/api/src/lib/WarehouseQueryService.ts b/apps/api/src/lib/WarehouseQueryService.ts index b0bad7cb0..27c8614ff 100644 --- a/apps/api/src/lib/WarehouseQueryService.ts +++ b/apps/api/src/lib/WarehouseQueryService.ts @@ -106,9 +106,9 @@ const isEmptyJsonBodyError = (error: unknown): boolean => error instanceof SyntaxError && /unexpected end of json input/i.test(error.message) const boundedResponseFetch = - (maxBytes: number): typeof fetch => + (maxBytes: number, requestFetch: typeof fetch = fetch): typeof fetch => async (input, init) => { - const response = await fetch(input, init) + const response = await requestFetch(input, init) if (response.body === null) return response const reader = response.body.getReader() @@ -145,15 +145,18 @@ const boundedResponseFetch = }) } -const createTinybirdSdkSqlClient = (config: TinybirdBackendConfig): WarehouseSqlClient => { - const makeClient = (fetchAdapter?: typeof fetch) => +const createTinybirdSdkSqlClient = ( + config: TinybirdBackendConfig, + requestFetch: typeof fetch = fetch, +): WarehouseSqlClient => { + const makeClient = (fetchAdapter: typeof fetch = requestFetch) => new Tinybird({ baseUrl: config.host, token: config.token, datasources: {}, pipes: {}, devMode: false, - ...(fetchAdapter === undefined ? {} : { fetch: fetchAdapter }), + fetch: fetchAdapter, }) const client = makeClient() const boundedClients = new Map() @@ -176,7 +179,7 @@ const createTinybirdSdkSqlClient = (config: TinybirdBackendConfig): WarehouseSql if (limits !== undefined) { queryClient = boundedClients.get(limits.maxBytes) ?? - makeClient(boundedResponseFetch(limits.maxBytes)) + makeClient(boundedResponseFetch(limits.maxBytes, requestFetch)) boundedClients.set(limits.maxBytes, queryClient) } const result = await queryClient.sql>(jsonSql) @@ -198,7 +201,7 @@ const createTinybirdSdkSqlClient = (config: TinybirdBackendConfig): WarehouseSql if (rows.length === 0) return const ndjson = rows.map((row) => JSON.stringify(row)).join("\n") const url = `${config.host.replace(/\/$/, "")}/v0/events?name=${encodeURIComponent(datasource)}&wait=false` - const response = await fetch(url, { + const response = await requestFetch(url, { method: "POST", headers: { "Content-Type": "application/x-ndjson", @@ -261,7 +264,8 @@ export class WarehouseQueryService extends Context.Service< url: clickhouseUrl.toString().replace(/\/$/, ""), username: env.CLICKHOUSE_USER, password: Option.match(env.CLICKHOUSE_PASSWORD, { - onNone: () => (kind === "tinybird-gateway" ? Redacted.value(env.TINYBIRD_TOKEN) : ""), + onNone: () => + kind === "tinybird-gateway" ? Redacted.value(env.TINYBIRD_TOKEN) : "", onSome: Redacted.value, }), database: env.CLICKHOUSE_DATABASE, @@ -350,8 +354,7 @@ export class WarehouseQueryService extends Context.Service< password: override.value.password, database: override.value.database, }, - clientCacheKey: - purpose === "raw" ? `raw:${tenant.orgId}` : `read:${tenant.orgId}`, + clientCacheKey: purpose === "raw" ? `raw:${tenant.orgId}` : `read:${tenant.orgId}`, } } diff --git a/apps/cli/src/server/otlp/encode.test.ts b/apps/cli/src/server/otlp/encode.test.ts index 9b57de317..ea11f1b4f 100644 --- a/apps/cli/src/server/otlp/encode.test.ts +++ b/apps/cli/src/server/otlp/encode.test.ts @@ -41,6 +41,11 @@ const b64 = (h: string): string => Buffer.from(h, "hex").toString("base64") const javaAgentTracePayloadB64 = "CvULCqUICiUKG2RlcGxveW1lbnQuZW52aXJvbm1lbnQubmFtZRIGCgR0ZXN0ChYKCWhvc3QuYXJjaBIJCgdhYXJjaDY0CisKCWhvc3QubmFtZRIeChx6aGFvamlydWlkZU1hY0Jvb2stQWlyLmxvY2FsCiMKDm9zLmRlc2NyaXB0aW9uEhEKD01hYyBPUyBYIDI2LjQuMQoTCgdvcy50eXBlEggKBmRhcndpbgoWCgpvcy52ZXJzaW9uEggKBjI2LjQuMQqgAQoUcHJvY2Vzcy5jb21tYW5kX2FyZ3MShwEqhAEKVQpTL1VzZXJzL3poYW9qaXJ1aS9MaWJyYXJ5L0phdmEvSmF2YVZpcnR1YWxNYWNoaW5lcy9vcGVuamRrLTIzL0NvbnRlbnRzL0hvbWUvYmluL2phdmEKKwopL3RtcC9tYXBsZS1qYXZhLWFnZW50LTdLMHcvT3RlbFJlcHJvLmphdmEKcAoXcHJvY2Vzcy5leGVjdXRhYmxlLnBhdGgSVQpTL1VzZXJzL3poYW9qaXJ1aS9MaWJyYXJ5L0phdmEvSmF2YVZpcnR1YWxNYWNoaW5lcy9vcGVuamRrLTIzL0NvbnRlbnRzL0hvbWUvYmluL2phdmEKEwoLcHJvY2Vzcy5waWQSBBjtpQIKVwobcHJvY2Vzcy5ydW50aW1lLmRlc2NyaXB0aW9uEjgKNk9yYWNsZSBDb3Jwb3JhdGlvbiBPcGVuSkRLIDY0LUJpdCBTZXJ2ZXIgVk0gMjMrMzctMjM2OQo1ChRwcm9jZXNzLnJ1bnRpbWUubmFtZRIdChtPcGVuSkRLIFJ1bnRpbWUgRW52aXJvbm1lbnQKJwoXcHJvY2Vzcy5ydW50aW1lLnZlcnNpb24SDAoKMjMrMzctMjM2OQo9ChNzZXJ2aWNlLmluc3RhbmNlLmlkEiYKJGQxZTI5NjA0LTI5ODYtNDlmZC1iNWJiLWU0ZjFhN2Q0ZjkyYwohCgxzZXJ2aWNlLm5hbWUSEQoPd2lzY2hvaWNlci11c2VyCj0KFXRlbGVtZXRyeS5kaXN0cm8ubmFtZRIkCiJvcGVudGVsZW1ldHJ5LWphdmEtaW5zdHJ1bWVudGF0aW9uCiQKGHRlbGVtZXRyeS5kaXN0cm8udmVyc2lvbhIICgYyLjI4LjEKIAoWdGVsZW1ldHJ5LnNkay5sYW5ndWFnZRIGCgRqYXZhCiUKEnRlbGVtZXRyeS5zZGsubmFtZRIPCg1vcGVudGVsZW1ldHJ5CiEKFXRlbGVtZXRyeS5zZGsudmVyc2lvbhIICgYxLjYyLjAKTwoXdmNzLnJlcG9zaXRvcnkudXJsLmZ1bGwSNAoyaHR0cHM6Ly9naXRodWIuY29tL1dpc2Nob2ljZXItWGlhbi93aXNjaG9pY2VyLXVzZXISoQMKMQohaW8ub3BlbnRlbGVtZXRyeS5qYXZhLWh0dHAtY2xpZW50EgwyLjI4LjEtYWxwaGESwgIKEGmLshO9k6DTrmM7fx2w70cSCD/XzWWx87T5KgNHRVQwAzkgsMwtvzK3GEEJ4tgtvzK3GEoVCgt0aHJlYWQubmFtZRIGCgRtYWluSi0KCHVybC5mdWxsEiEKH2h0dHA6Ly8xMjcuMC4wLjE6MTgwODEvcGluZz9pPTJKEwoLc2VydmVyLnBvcnQSBBihjQFKEwoKZXJyb3IudHlwZRIFCgM0MDRKHQoOc2VydmVyLmFkZHJlc3MSCwoJMTI3LjAuMC4xSiEKGG5ldHdvcmsucHJvdG9jb2wudmVyc2lvbhIFCgMxLjFKIAoZaHR0cC5yZXNwb25zZS5zdGF0dXNfY29kZRIDGJQDShwKE2h0dHAucmVxdWVzdC5tZXRob2QSBQoDR0VUSg8KCXRocmVhZC5pZBICGAF6AhgChQEDAQAAGidodHRwczovL29wZW50ZWxlbWV0cnkuaW8vc2NoZW1hcy8xLjM3LjAaJ2h0dHBzOi8vb3BlbnRlbGVtZXRyeS5pby9zY2hlbWFzLzEuMjQuMA==" +// Minimal ExportTraceServiceRequest encoded with the upstream OTLP schema. +// Span.flags is field 16, so its fixed32 key is the two-byte sequence 0x85 0x01. +const pythonSpanFlagsPayloadHex = + "0a480a00124412420a1000000000000000000000000000000000120800000000000000002a0a776974682d666c6167733001390100000000000000410200000000000000850101000000" + const attr = (key: string, str: string) => ({ key, value: { stringValue: str } }) const sampleTraceReq = () => ({ @@ -272,6 +277,17 @@ describe("protobuf round-trip (proves vendored .proto field numbers)", () => { }) describe("external OTLP protobuf compatibility", () => { + it("decodes a non-zero fixed32 span flags field emitted by the Python SDK", () => { + const decoded = decodeTraceRequest(Buffer.from(pythonSpanFlagsPayloadHex, "hex")) as { + resourceSpans: Array<{ scopeSpans: Array<{ spans: Array<{ name: string; flags: number }> }> }> + } + const span = decoded.resourceSpans[0]?.scopeSpans[0]?.spans[0] + expect(span).toMatchObject({ name: "with-flags", flags: 1 }) + + const [batch] = encodeTraces(decoded) + expect(batch).toMatchObject({ datasource: "traces", rowCount: 1 }) + }) + it("decodes a Java agent OTLP/HTTP traces payload captured from the wire", () => { const decoded = decodeTraceRequest(Buffer.from(javaAgentTracePayloadB64, "base64")) const [batch] = encodeTraces(decoded) diff --git a/apps/cli/src/server/otlp/proto.ts b/apps/cli/src/server/otlp/proto.ts index c9f4884b7..1887b5225 100644 --- a/apps/cli/src/server/otlp/proto.ts +++ b/apps/cli/src/server/otlp/proto.ts @@ -119,13 +119,13 @@ message Span { string trace_state = 3; repeated KeyValue attributes = 4; uint32 dropped_attributes_count = 5; - uint32 flags = 6; + fixed32 flags = 6; } repeated Link links = 13; uint32 dropped_links_count = 14; Status status = 15; - uint32 flags = 16; + fixed32 flags = 16; } message Status {