Skip to content

Commit abf6456

Browse files
sohankshirsagarsohankshirsagar
authored
fix preserve mysql2 stream queries and native tracing events (#140)
1 parent 5df98d6 commit abf6456

8 files changed

Lines changed: 400 additions & 103 deletions

File tree

‎.github/workflows/e2e.yml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ jobs:
1717
runs-on: ubuntu-latest
1818
timeout-minutes: 30
1919
strategy:
20+
fail-fast: false
2021
max-parallel: 6
2122
matrix:
2223
library:

‎src/core/tracing/SpanUtils.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -169,6 +169,11 @@ export class SpanUtils {
169169
logger.debug(
170170
`[SpanUtils] Stopping recording of child spans for span ${spanContext.spanId}, packageName: ${options.packageName}, instrumentationName: ${options.instrumentationName}`,
171171
);
172+
if (mode === TuskDriftMode.REPLAY) {
173+
throw new Error(
174+
`Unexpected child span in replay mode for span ${spanContext.spanId}, packageName: ${options.packageName}, instrumentationName: ${options.instrumentationName}`,
175+
);
176+
}
172177
return originalFunctionCall();
173178
}
174179
}

‎src/instrumentation/libraries/mysql2/Instrumentation.ts‎

Lines changed: 223 additions & 41 deletions
Large diffs are not rendered by default.

‎src/instrumentation/libraries/mysql2/e2e-tests/cjs-mysql2/src/index.ts‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1092,6 +1092,26 @@ const server = http.createServer(async (req, res) => {
10921092
return;
10931093
}
10941094

1095+
if (url === "/test/pool-execute-singleton-values" && method === "GET") {
1096+
pool.execute("SELECT * FROM test_users WHERE id = ?", [1], (error, results) => {
1097+
if (error) {
1098+
res.writeHead(500, { "Content-Type": "application/json" });
1099+
res.end(JSON.stringify({ success: false, error: error.message }));
1100+
return;
1101+
}
1102+
res.writeHead(200, { "Content-Type": "application/json" });
1103+
res.end(
1104+
JSON.stringify({
1105+
success: true,
1106+
data: results,
1107+
rowCount: Array.isArray(results) ? results.length : 0,
1108+
queryType: "pool-execute-singleton-values",
1109+
}),
1110+
);
1111+
});
1112+
return;
1113+
}
1114+
10951115
// 404 for unknown routes
10961116
res.writeHead(404, { "Content-Type": "application/json" });
10971117
res.end(JSON.stringify({ error: "Not found" }));

‎src/instrumentation/libraries/mysql2/e2e-tests/cjs-mysql2/src/test_requests.mjs‎

Lines changed: 32 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,34 +1,35 @@
1-
import { makeRequest, printRequestSummary } from '/app/test-utils.mjs';
1+
import { makeRequest, printRequestSummary } from "/app/test-utils.mjs";
22

3-
await makeRequest('GET', '/health');
4-
await makeRequest('GET', '/test/connection-query');
5-
await makeRequest('POST', '/test/connection-parameterized', { body: { userId: 1 } });
6-
await makeRequest('GET', '/test/connection-execute');
7-
await makeRequest('POST', '/test/connection-execute-params', { body: { userId: 2 } });
8-
await makeRequest('GET', '/test/pool-query');
9-
await makeRequest('POST', '/test/pool-parameterized', { body: { userId: 1 } });
10-
await makeRequest('GET', '/test/pool-execute');
11-
await makeRequest('POST', '/test/pool-execute-params', { body: { userId: 2 } });
12-
await makeRequest('GET', '/test/pool-getConnection');
13-
await makeRequest('GET', '/test/connection-connect');
14-
await makeRequest('GET', '/test/connection-ping');
15-
await makeRequest('GET', '/test/stream-query');
16-
await makeRequest('GET', '/test/sequelize-authenticate');
17-
await makeRequest('GET', '/test/sequelize-findall');
18-
await makeRequest('POST', '/test/sequelize-findone', { body: { userId: 1 } });
19-
await makeRequest('GET', '/test/sequelize-complex');
20-
await makeRequest('GET', '/test/sequelize-raw');
21-
await makeRequest('POST', '/test/sequelize-transaction');
22-
await makeRequest('GET', '/test/promise-connection-query');
23-
await makeRequest('GET', '/test/promise-pool-query');
24-
await makeRequest('GET', '/test/promise-pool-getconnection');
25-
await makeRequest('GET', '/test/transaction-methods');
26-
await makeRequest('GET', '/test/prepare-statement');
27-
await makeRequest('GET', '/test/change-user');
28-
await makeRequest('GET', '/test/nested-null-values');
29-
await makeRequest('GET', '/test/binary-data');
30-
await makeRequest('GET', '/test/knex-raw-query');
31-
await makeRequest('POST', '/test/knex-savepoint');
32-
await makeRequest('GET', '/test/knex-streaming');
3+
await makeRequest("GET", "/health");
4+
await makeRequest("GET", "/test/connection-query");
5+
await makeRequest("POST", "/test/connection-parameterized", { body: { userId: 1 } });
6+
await makeRequest("GET", "/test/connection-execute");
7+
await makeRequest("POST", "/test/connection-execute-params", { body: { userId: 2 } });
8+
await makeRequest("GET", "/test/pool-query");
9+
await makeRequest("POST", "/test/pool-parameterized", { body: { userId: 1 } });
10+
await makeRequest("GET", "/test/pool-execute");
11+
await makeRequest("POST", "/test/pool-execute-params", { body: { userId: 2 } });
12+
await makeRequest("GET", "/test/pool-getConnection");
13+
await makeRequest("GET", "/test/connection-connect");
14+
await makeRequest("GET", "/test/connection-ping");
15+
await makeRequest("GET", "/test/stream-query");
16+
await makeRequest("GET", "/test/sequelize-authenticate");
17+
await makeRequest("GET", "/test/sequelize-findall");
18+
await makeRequest("POST", "/test/sequelize-findone", { body: { userId: 1 } });
19+
await makeRequest("GET", "/test/sequelize-complex");
20+
await makeRequest("GET", "/test/sequelize-raw");
21+
await makeRequest("POST", "/test/sequelize-transaction");
22+
await makeRequest("GET", "/test/promise-connection-query");
23+
await makeRequest("GET", "/test/promise-pool-query");
24+
await makeRequest("GET", "/test/promise-pool-getconnection");
25+
await makeRequest("GET", "/test/transaction-methods");
26+
await makeRequest("GET", "/test/prepare-statement");
27+
await makeRequest("GET", "/test/change-user");
28+
await makeRequest("GET", "/test/nested-null-values");
29+
await makeRequest("GET", "/test/binary-data");
30+
await makeRequest("GET", "/test/knex-raw-query");
31+
await makeRequest("POST", "/test/knex-savepoint");
32+
await makeRequest("GET", "/test/knex-streaming");
33+
await makeRequest("GET", "/test/pool-execute-singleton-values");
3334

3435
printRequestSummary();

‎src/instrumentation/libraries/mysql2/e2e-tests/esm-mysql2/src/index.ts‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1052,6 +1052,26 @@ const server = http.createServer(async (req, res) => {
10521052
return;
10531053
}
10541054

1055+
if (url === "/test/pool-execute-singleton-values" && method === "GET") {
1056+
pool.execute("SELECT * FROM test_users WHERE id = ?", [1], (error, results) => {
1057+
if (error) {
1058+
res.writeHead(500, { "Content-Type": "application/json" });
1059+
res.end(JSON.stringify({ success: false, error: error.message }));
1060+
return;
1061+
}
1062+
res.writeHead(200, { "Content-Type": "application/json" });
1063+
res.end(
1064+
JSON.stringify({
1065+
success: true,
1066+
data: results,
1067+
rowCount: Array.isArray(results) ? results.length : 0,
1068+
queryType: "pool-execute-singleton-values",
1069+
}),
1070+
);
1071+
});
1072+
return;
1073+
}
1074+
10551075
// 404 for unknown routes
10561076
res.writeHead(404, { "Content-Type": "application/json" });
10571077
res.end(JSON.stringify({ error: "Not found" }));

‎src/instrumentation/libraries/mysql2/e2e-tests/esm-mysql2/src/test_requests.mjs‎

Lines changed: 32 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,34 +1,35 @@
1-
import { makeRequest, printRequestSummary } from '/app/test-utils.mjs';
1+
import { makeRequest, printRequestSummary } from "/app/test-utils.mjs";
22

3-
await makeRequest('GET', '/health');
4-
await makeRequest('GET', '/test/connection-query');
5-
await makeRequest('POST', '/test/connection-parameterized', { body: { userId: 1 } });
6-
await makeRequest('GET', '/test/connection-execute');
7-
await makeRequest('POST', '/test/connection-execute-params', { body: { userId: 2 } });
8-
await makeRequest('GET', '/test/pool-query');
9-
await makeRequest('POST', '/test/pool-parameterized', { body: { userId: 1 } });
10-
await makeRequest('GET', '/test/pool-execute');
11-
await makeRequest('POST', '/test/pool-execute-params', { body: { userId: 2 } });
12-
await makeRequest('GET', '/test/pool-getConnection');
13-
await makeRequest('GET', '/test/connection-connect');
14-
await makeRequest('GET', '/test/connection-ping');
15-
await makeRequest('GET', '/test/stream-query');
16-
await makeRequest('GET', '/test/sequelize-authenticate');
17-
await makeRequest('GET', '/test/sequelize-findall');
18-
await makeRequest('POST', '/test/sequelize-findone', { body: { userId: 1 } });
19-
await makeRequest('GET', '/test/sequelize-complex');
20-
await makeRequest('GET', '/test/sequelize-raw');
21-
await makeRequest('POST', '/test/sequelize-transaction');
22-
await makeRequest('GET', '/test/promise-connection-query');
23-
await makeRequest('GET', '/test/promise-pool-query');
24-
await makeRequest('GET', '/test/promise-pool-getconnection');
25-
await makeRequest('GET', '/test/transaction-methods');
26-
await makeRequest('GET', '/test/prepare-statement');
27-
await makeRequest('GET', '/test/change-user');
28-
await makeRequest('GET', '/test/nested-null-values');
29-
await makeRequest('GET', '/test/binary-data');
30-
await makeRequest('GET', '/test/knex-raw-query');
31-
await makeRequest('POST', '/test/knex-savepoint');
32-
await makeRequest('GET', '/test/knex-streaming');
3+
await makeRequest("GET", "/health");
4+
await makeRequest("GET", "/test/connection-query");
5+
await makeRequest("POST", "/test/connection-parameterized", { body: { userId: 1 } });
6+
await makeRequest("GET", "/test/connection-execute");
7+
await makeRequest("POST", "/test/connection-execute-params", { body: { userId: 2 } });
8+
await makeRequest("GET", "/test/pool-query");
9+
await makeRequest("POST", "/test/pool-parameterized", { body: { userId: 1 } });
10+
await makeRequest("GET", "/test/pool-execute");
11+
await makeRequest("POST", "/test/pool-execute-params", { body: { userId: 2 } });
12+
await makeRequest("GET", "/test/pool-getConnection");
13+
await makeRequest("GET", "/test/connection-connect");
14+
await makeRequest("GET", "/test/connection-ping");
15+
await makeRequest("GET", "/test/stream-query");
16+
await makeRequest("GET", "/test/sequelize-authenticate");
17+
await makeRequest("GET", "/test/sequelize-findall");
18+
await makeRequest("POST", "/test/sequelize-findone", { body: { userId: 1 } });
19+
await makeRequest("GET", "/test/sequelize-complex");
20+
await makeRequest("GET", "/test/sequelize-raw");
21+
await makeRequest("POST", "/test/sequelize-transaction");
22+
await makeRequest("GET", "/test/promise-connection-query");
23+
await makeRequest("GET", "/test/promise-pool-query");
24+
await makeRequest("GET", "/test/promise-pool-getconnection");
25+
await makeRequest("GET", "/test/transaction-methods");
26+
await makeRequest("GET", "/test/prepare-statement");
27+
await makeRequest("GET", "/test/change-user");
28+
await makeRequest("GET", "/test/nested-null-values");
29+
await makeRequest("GET", "/test/binary-data");
30+
await makeRequest("GET", "/test/knex-raw-query");
31+
await makeRequest("POST", "/test/knex-savepoint");
32+
await makeRequest("GET", "/test/knex-streaming");
33+
await makeRequest("GET", "/test/pool-execute-singleton-values");
3334

3435
printRequestSummary();

‎src/instrumentation/libraries/mysql2/integration-tests/mysql2.test.ts‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ TuskDrift.markAppAsReady();
1111

1212
import test from "ava";
1313
import { SpanKind } from "@opentelemetry/api";
14+
import diagnosticsChannel from "node:diagnostics_channel";
1415
import { SpanUtils } from "../../../../core/tracing/SpanUtils";
1516
import { TuskDriftMode } from "../../../../core/TuskDrift";
1617
import {
@@ -588,6 +589,72 @@ test.serial("should handle streaming queries", async (t) => {
588589
t.true(mysql2Spans.length > 0);
589590
});
590591

592+
test.serial("should emit native mysql2 tracing events for streaming queries", async (t) => {
593+
if (typeof diagnosticsChannel.tracingChannel !== "function") {
594+
t.pass();
595+
return;
596+
}
597+
598+
const events: Array<{ type: string; ctx: any }> = [];
599+
const queryChannel = diagnosticsChannel.tracingChannel("mysql2:query");
600+
const subscriber = {
601+
start(ctx: object) {
602+
events.push({ type: "start", ctx });
603+
},
604+
end(ctx: object) {
605+
events.push({ type: "end", ctx });
606+
},
607+
asyncStart(ctx: object) {
608+
events.push({ type: "asyncStart", ctx });
609+
},
610+
asyncEnd(ctx: object) {
611+
events.push({ type: "asyncEnd", ctx });
612+
},
613+
error(ctx: object) {
614+
events.push({ type: "error", ctx });
615+
},
616+
};
617+
618+
queryChannel.subscribe(subscriber);
619+
620+
try {
621+
const rows: any[] = [];
622+
623+
await new Promise<void>((resolve, reject) => {
624+
withRootSpan(() => {
625+
const query = connection.query("SELECT * FROM test_users ORDER BY id");
626+
627+
query
628+
.on("error", (err: any) => {
629+
reject(err);
630+
})
631+
.on("result", (row: any) => {
632+
rows.push(row);
633+
})
634+
.on("end", () => {
635+
resolve();
636+
});
637+
});
638+
});
639+
640+
t.is(rows.length, 2);
641+
642+
const eventTypes = events.map((event) => event.type);
643+
t.true(eventTypes.includes("start"));
644+
t.true(eventTypes.includes("asyncEnd"));
645+
t.false(eventTypes.includes("error"));
646+
647+
const startEvent = events.find((event) => event.type === "start");
648+
t.truthy(startEvent);
649+
t.true(String(startEvent?.ctx?.query || "").includes("SELECT * FROM test_users"));
650+
t.is(startEvent?.ctx?.database, TEST_MYSQL_CONFIG.database);
651+
t.is(startEvent?.ctx?.serverAddress, TEST_MYSQL_CONFIG.host);
652+
t.is(startEvent?.ctx?.serverPort, TEST_MYSQL_CONFIG.port);
653+
} finally {
654+
queryChannel.unsubscribe(subscriber);
655+
}
656+
});
657+
591658
test.serial("should handle connection.ping", async (t) => {
592659
await new Promise<void>((resolve, reject) => {
593660
withRootSpan(() => {

0 commit comments

Comments
 (0)