Skip to content

feat(telemetry): add first-class OpenTelemetry tracing #23

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
psteinroe wants to merge 1 commit into from
Open

feat(telemetry): add first-class OpenTelemetry tracing #23

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 20 additions & 5 deletions bun.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

17 changes: 15 additions & 2 deletions migrations/0000000001_setup.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ create table pgconductor._private_executions (
failed_at timestamptz,
completed_at timestamptz,
payload jsonb,
trace_context jsonb,
run_at timestamptz default pgconductor._private_current_time() not null,
locked_at timestamptz,
locked_by uuid,
Expand Down Expand Up @@ -344,7 +345,8 @@ create type pgconductor.execution_spec as (
dedupe_next_slot boolean,
cron_expression text,
priority integer,
"group" text
"group" text,
trace_context jsonb
);

create type pgconductor.task_spec as (
Expand Down Expand Up @@ -603,6 +605,7 @@ begin
task_key,
queue,
payload,
trace_context,
run_at,
dedupe_key,
singleton_on,
Expand All @@ -615,6 +618,7 @@ begin
spec.task_key,
coalesce(spec.queue, 'default'),
spec.payload,
spec.trace_context,
coalesce(spec.run_at, v_now),
spec.dedupe_key,
case
Expand Down Expand Up @@ -652,7 +656,8 @@ create or replace function pgconductor.invoke(
p_dedupe_next_slot boolean default false,
p_cron_expression text default null,
p_priority integer default null,
p_group text default null
p_group text default null,
p_trace_context jsonb default null
)
returns table(id uuid)
language plpgsql
Expand Down Expand Up @@ -708,6 +713,7 @@ begin
task_key,
queue,
payload,
trace_context,
run_at,
dedupe_key,
singleton_on,
Expand All @@ -719,6 +725,7 @@ begin
p_task_key,
p_queue,
p_payload,
p_trace_context,
v_run_at,
p_dedupe_key,
v_singleton_on,
Expand All @@ -741,6 +748,7 @@ begin
task_key,
queue,
payload,
trace_context,
run_at,
dedupe_key,
singleton_on,
Expand All @@ -752,6 +760,7 @@ begin
p_task_key,
p_queue,
p_payload,
p_trace_context,
v_next_singleton_on,
p_dedupe_key,
v_next_singleton_on,
Expand All @@ -763,6 +772,7 @@ begin
where singleton_on is not null and completed_at is null and failed_at is null and cancelled = false
do update set
payload = excluded.payload,
trace_context = excluded.trace_context,
run_at = excluded.run_at,
priority = excluded.priority,
cron_expression = excluded.cron_expression,
Expand All @@ -778,6 +788,7 @@ begin
task_key,
queue,
payload,
trace_context,
run_at,
dedupe_key,
cron_expression,
Expand All @@ -788,6 +799,7 @@ begin
p_task_key,
p_queue,
p_payload,
p_trace_context,
v_run_at,
p_dedupe_key,
p_cron_expression,
Expand All @@ -796,6 +808,7 @@ begin
)
on conflict (task_key, dedupe_key, queue) do update set
payload = excluded.payload,
trace_context = excluded.trace_context,
run_at = excluded.run_at,
priority = excluded.priority,
cron_expression = excluded.cron_expression,
Expand Down
8 changes: 5 additions & 3 deletions package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,11 @@
"zod": "4.1.12",
"@standard-schema/spec": "1.0.0",
"cron-parser": "5.4.0",
"oxfmt": "latest",
"oxlint": "latest",
"oxlint-tsgolint": "latest"
"@opentelemetry/api": "1.9.0",
"@opentelemetry/sdk-trace-base": "1.30.1",
"oxfmt": "0.15.0",
"oxlint": "1.30.0",
"oxlint-tsgolint": "0.8.3"
},
"devDependencies": {
"oxfmt": "catalog:",
Expand Down
5 changes: 4 additions & 1 deletion packages/pgconductor-js/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,20 +33,23 @@
"provenance": true
},
"scripts": {
"build": "bun build src/index.ts --outdir dist --target node && bun build cli/index.ts --outdir dist/cli --target node",
"build": "bun build src/index.ts --outdir dist --target node --external @opentelemetry/api && bun build cli/index.ts --outdir dist/cli --target node --external @opentelemetry/api",
"test": "bun test",
"test:unit": "bun test tests/unit",
"test:integration": "bun test tests/integration",
"typecheck": "tsc --noEmit",
"perf": "bun perf/run.ts"
},
"devDependencies": {
"@opentelemetry/api": "catalog:",
"@opentelemetry/sdk-trace-base": "catalog:",
"@testcontainers/postgresql": "catalog:",
"@types/bun": "catalog:",
"expect-type": "catalog:",
"zod": "catalog:",
},
"peerDependencies": {
"@opentelemetry/api": ">=1.4 <2",
"typescript": "catalog:"
},
"dependencies": {
Expand Down
69 changes: 62 additions & 7 deletions packages/pgconductor-js/src/conductor.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,17 @@ import {
import { Worker, type WorkerConfig } from "./worker";
import { DefaultLogger, type Logger } from "./lib/logger";
import { SchemaManager } from "./schema-manager";
import {
carrierForContext,
contextForSpan,
endSpan,
messagingAttributes,
setSpanAttribute,
setSpanError,
startSpan,
runWithSpan,
} from "./telemetry";
import { SpanKind } from "@opentelemetry/api";
import type {
EventDefinition,
GenericDatabase,
Expand Down Expand Up @@ -80,6 +91,8 @@ export type ConductorOptions<
context: ExtraContext;

logger?: Logger;
/** Disable OpenTelemetry instrumentation. The default is enabled and uses the global API provider. */
telemetry?: false;
};

// similar to inngest client
Expand Down Expand Up @@ -142,6 +155,7 @@ export class Conductor<
database?: TDatabaseSchema;
context: TExtraContext;
logger?: Logger;
telemetry?: false;
},
): Conductor<TTaskSchemas, TEventSchemas, TDatabaseSchema, TExtraContext> {
return new Conductor<TTaskSchemas, TEventSchemas, TDatabaseSchema, TExtraContext>(options);
Expand Down Expand Up @@ -234,6 +248,7 @@ export class Conductor<
options.config,
this.options.context,
this.options.events?.definitions ?? [],
this.options.telemetry !== false,
);
}

Expand Down Expand Up @@ -303,6 +318,15 @@ export class Conductor<
const queue = task.queue || "default";

if (Array.isArray(payloadOrItems)) {
const producer =
this.options.telemetry === false
? null
: startSpan(
`send ${queue}`,
SpanKind.PRODUCER,
messagingAttributes(taskName, queue, "send"),
);
const carrier = producer ? carrierForContext(contextForSpan(producer)) : null;
const specs = payloadOrItems.map((item) => ({
task_key: taskName,
queue,
Expand All @@ -314,16 +338,47 @@ export class Conductor<
cron_expression: item.cron_expression,
priority: item.priority,
group: item.group,
trace_context: carrier,
}));
return this.db.invokeBatch(specs);
try {
const ids = await runWithSpan(producer, () => this.db.invokeBatch(specs));
setSpanAttribute(producer, "messaging.batch.message_count", payloadOrItems.length);
return ids;
} catch (error) {
setSpanError(producer, error);
throw error;
} finally {
endSpan(producer);
}
}

return this.db.invoke({
task_key: taskName,
queue,
payload: payloadOrItems,
...opts,
});
const producer =
this.options.telemetry === false
? null
: startSpan(
`send ${queue}`,
SpanKind.PRODUCER,
messagingAttributes(taskName, queue, "send"),
);
const carrier = producer ? carrierForContext(contextForSpan(producer)) : null;
try {
const id = await runWithSpan(producer, () =>
this.db.invoke({
task_key: taskName,
queue,
payload: payloadOrItems,
...opts,
trace_context: carrier,
}),
);
if (id) setSpanAttribute(producer, "messaging.message.id", id);
return id;
} catch (error) {
setSpanError(producer, error);
throw error;
} finally {
endSpan(producer);
}
}

/**
Expand Down
3 changes: 3 additions & 0 deletions packages/pgconductor-js/src/database-client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import {
type EmitEventArgs,
} from "./query-builder";
import { makeChildLogger, type Logger } from "./lib/logger";
import type { TraceContextCarrier } from "./internal-types";

export type JsonValue = string | number | boolean | null | Payload | JsonValue[];
export type Payload = { [key: string]: JsonValue };
Expand All @@ -39,6 +40,7 @@ export interface ExecutionSpec {
parent_execution_id?: string | null;
parent_step_key?: string | null;
parent_timeout_ms?: number | null;
trace_context?: TraceContextCarrier | null;
}

export interface TaskSpec {
Expand Down Expand Up @@ -82,6 +84,7 @@ export interface Execution {
dead_letter_error?: string | null;
dead_letter_attempts?: number | null;
dead_letter_failed_at?: Date | null;
trace_context?: TraceContextCarrier | null;
}

// todo: move all of this to query-builder too or create new types.ts file
Expand Down
Loading
Loading