Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/).

### Added

- Queue worker transactions are traced as coherent `<service> - tx` spans under the `cds.spawn - run task` root, instead of orphaned per-call spans

### Changed

### Fixed
Expand Down
19 changes: 19 additions & 0 deletions lib/tracing/cds.js
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,25 @@ module.exports = () => {
}
})

// Wrap `srv.tx(fn)` so the queue worker's two transactions (SELECT+UPDATE lock tx,
// then handle+DELETE dispatch tx) appear as child spans of the `cds.spawn - run task`
// root instead of each top-level CAP call inside them becoming an orphan root.
// Only wraps when there is no current `srv.context` (i.e. not already inside a request).
const _tx_proto = cds.Service.prototype.tx
cds.Service.prototype.tx = wrap(_tx_proto, {
wrapper: function tx() {
const fnIdx = typeof arguments[0] === 'function' ? 0 : typeof arguments[1] === 'function' ? 1 : -1
if (fnIdx < 0) return _tx_proto.apply(this, arguments)
// Skip if this service is already handling a request (has an active EventContext),
// or if this is a nested .tx() call (cds.Service.tx is a no-op when already in tx).
// Do NOT skip for a bare {} context set by processInboundMsg — that's the entry point
// for file-based messaging consumer delivery and should get a root span.
if (this.context instanceof cds.EventContext) return _tx_proto.apply(this, arguments)
const name = `${this.name || 'cds'} - tx`
return trace(name, _tx_proto, this, arguments, {})
}
})

const { spawn: _spawn } = cds
cds.spawn = wrap(_spawn, {
wrapper: function spawn() {
Expand Down
21 changes: 21 additions & 0 deletions test/bookshop/.cdsrc.json
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,18 @@
}
}
},
"[tracing-in-memory]": {
"requires": {
"telemetry": {
"tracing": {
"exporter": {
"module": "./lib/MyInMemorySpanExporter.js",
"class": "MyInMemorySpanExporter"
}
}
}
}
},
"[persistent-outbox]": {
"requires": {
"messaging": {
Expand All @@ -90,6 +102,15 @@
}
}
},
"[inboxed]": {
"requires": {
"messaging": {
"kind": "file-based-messaging",
"file": "../inboxed",
"inboxed": true
}
}
},
"[without-outbox]": {
"requires": {
"messaging": {
Expand Down
59 changes: 59 additions & 0 deletions test/bookshop/lib/MyInMemorySpanExporter.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
// In-memory span exporter for tests. Spans are accumulated in a module-level array that
// tests can import directly via `require('./lib/MyInMemorySpanExporter').captured`.
// Wired into the tracer provider via .cdsrc.json profile config (no provider-poking from tests).

const { ExportResultCode } = require('@opentelemetry/core')

const captured = []

class MyInMemorySpanExporter {
export(spans, resultCallback) {
captured.push(...spans)
resultCallback({ code: ExportResultCode.SUCCESS })
}

shutdown() {
return Promise.resolve()
}

forceFlush() {
return Promise.resolve()
}
}

// Returns the captured spans grouped by traceId, each group is a hierarchy:
// { traceId, root, all, byParent }
// `root` is the span with no parent inside the group (the visible root for the exporter's
// "elapsed times:" primer logic — i.e. spans whose parentSpanId is not present in this group).
function groupedByTrace() {
const byTrace = new Map()
for (const s of captured) {
const tid = s.spanContext().traceId
if (!byTrace.has(tid)) byTrace.set(tid, [])
byTrace.get(tid).push(s)
}

return [...byTrace.entries()].map(([traceId, all]) => {
const ids = new Set(all.map(s => s.spanContext().spanId))
const roots = all.filter(s => !s.parentSpanContext?.spanId || !ids.has(s.parentSpanContext.spanId))
const byParent = new Map()
for (const s of all) {
const pid = s.parentSpanContext?.spanId
if (!byParent.has(pid)) byParent.set(pid, [])
byParent.get(pid).push(s)
}
return { traceId, root: roots[0], roots, all, byParent }
})
}

// Returns just the visible "root" spans across all captured traces. These correspond 1:1 to
// "elapsed times:" primers our ConsoleSpanExporter would emit for the same data.
function rootSpans() {
return groupedByTrace().flatMap(g => g.roots)
}

function reset() {
captured.length = 0
}

module.exports = { MyInMemorySpanExporter, captured, groupedByTrace, rootSpans, reset }
3 changes: 3 additions & 0 deletions test/bookshop/srv/admin-service.cds
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,9 @@ service AdminService @(requires: 'admin') {

action test_spawn();
action test_emit();
action test_outboxed_send();
action test_outboxed_send_batch();
action test_scheduled();

event foo {
bar : String;
Expand Down
24 changes: 24 additions & 0 deletions test/bookshop/srv/admin-service.js
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,30 @@ module.exports = class AdminService extends cds.ApplicationService {
await messaging.emit('foo', { bar: 'baz' })
})

// test_outboxed_send: writes a task to the persistent outbox addressed to ExternalServiceOne,
// whose handler the test installs. Exercises the queue-worker path (scan, lock, dispatch).
this.on('test_outboxed_send', async () => {
const externalOne = await cds.connect.to('ExternalServiceOne')
await cds.queued(externalOne).send('call', {})
})

// test_outboxed_send_batch: writes multiple tasks to the persistent outbox to exercise chunkSize > 1 fan-out.
this.on('test_outboxed_send_batch', async () => {
const externalOne = await cds.connect.to('ExternalServiceOne')
const queued = cds.queued(externalOne)
await Promise.all([
queued.send('call', {}),
queued.send('call', {}),
queued.send('call', {})
])
})

// test_scheduled: schedules a one-shot task to fire after a short delay.
this.on('test_scheduled', async () => {
const externalOne = await cds.connect.to('ExternalServiceOne')
await cds.queued(externalOne).schedule('call', {}).after(10)
})

return super.init()
}
}
Expand Down
Loading