Skip to content

Commit 8a3fafc

Browse files
dszakallasdszakallas
committed
feat(collector): force CS-CR, CS-NE events to be collected in pairs
1 parent 39aed94 commit 8a3fafc

3 files changed

Lines changed: 144 additions & 65 deletions

File tree

‎lib/agent/tracer/collector.js‎

Lines changed: 46 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -147,9 +147,8 @@ Collector.prototype.networkError = function (briefcase, error) {
147147
}
148148
}, levels.ERROR)
149149

150-
if (root) {
151-
this._flushCache(cacheId)
152-
}
150+
this._removeLockFromCache(cacheId)
151+
this._flushUnlockedCache(cacheId)
153152

154153
return { briefcase: briefcase }
155154
}
@@ -158,10 +157,10 @@ Collector.prototype.end = function (briefcase, options) {
158157
var communicationId = briefcase.communication.id
159158

160159
if (options && options.skip) {
161-
this._deleteCache(communicationId)
162-
} else {
163-
this._flushCache(communicationId)
160+
this._skipCache(communicationId)
164161
}
162+
this._removeLockFromCache(communicationId)
163+
this._flushUnlockedCache(communicationId)
165164
}
166165

167166
Collector.prototype.clientSend = function (payload, briefcase, options) {
@@ -207,9 +206,7 @@ Collector.prototype.clientSend = function (payload, briefcase, options) {
207206
d: payload.data
208207
}, payload.severity != null ? payload.severity : this.defaultSeverity)
209208

210-
if (root) {
211-
this._lockCache(cacheId)
212-
}
209+
this._addLockToCache(cacheId)
213210

214211
var severity = this._getSeverityHintFromCache(cacheId)
215212

@@ -228,7 +225,6 @@ Collector.prototype.clientSend = function (payload, briefcase, options) {
228225
}
229226

230227
Collector.prototype.serverRecv = function (payload, duffelBag, options) {
231-
briefcase = briefcase || {}
232228
var timestamp = microtime.now()
233229

234230
var communication = {
@@ -257,7 +253,7 @@ Collector.prototype.serverRecv = function (payload, duffelBag, options) {
257253
d: payload.data
258254
}, duffelBag.severity)
259255

260-
this._lockCache(communication.id)
256+
this._addLockToCache(communication.id)
261257

262258
return { briefcase: briefcase }
263259
}
@@ -276,7 +272,7 @@ Collector.prototype.serverSend = function (payload, briefcase, options) {
276272
var transactionId = briefcase.communication.transactionId
277273

278274
if (options && options.skip) {
279-
this._deleteCache(communicationId)
275+
this._skipCache(communicationId)
280276
} else {
281277
this._cache(communicationId, {
282278
t: EVENT_TYPE.SERVER_SEND,
@@ -287,9 +283,11 @@ Collector.prototype.serverSend = function (payload, briefcase, options) {
287283
s: payload.status,
288284
d: payload.data
289285
}, payload.severity != null ? payload.severity : this.defaultSeverity)
290-
this._flushCache(communicationId)
291286
}
292287

288+
this._removeLockFromCache(communicationId)
289+
this._flushUnlockedCache(communicationId)
290+
293291
var severity = this._getSeverityHintFromCache(communicationId)
294292

295293
var duffelBag = {
@@ -318,7 +316,7 @@ Collector.prototype.clientRecv = function (payload, duffelBag, briefcase) {
318316

319317
var root = !communication.parentId
320318

321-
var cacheId = root ? 'root-' + communication.id : communication.parentId
319+
var cacheId = root ? 'root-' + briefcase.csCtx.communicationId : communication.parentId
322320

323321
this._cache(cacheId, {
324322
t: EVENT_TYPE.CLIENT_RECV,
@@ -333,9 +331,8 @@ Collector.prototype.clientRecv = function (payload, duffelBag, briefcase) {
333331
d: payload.data
334332
}, duffelBag.severity)
335333

336-
if (root) {
337-
this._flushCache(cacheId)
338-
}
334+
this._removeLockFromCache(cacheId)
335+
this._flushUnlockedCache(cacheId)
339336

340337
return { briefcase: briefcase }
341338
}
@@ -347,41 +344,56 @@ Collector.prototype._cache = function (cacheId, data, severity) {
347344
}
348345
if (this._eventBuffers[cacheId] == null) {
349346
this._eventBuffers[cacheId] = {
350-
locked: false,
347+
skipped: false,
348+
locks: 0,
351349
severityBuf: new ExpiringBuffer(this._eventTtl),
352350
eventBuf: new ExpiringBuffer(this._eventTtl)
353351
}
352+
} else if (this._eventBuffers[cacheId].skipped) {
353+
return
354354
}
355355
this._eventBuffers[cacheId].eventBuf.push(data)
356356
this._eventBuffers[cacheId].severityBuf.push(severity)
357357
}
358358

359+
Collector.prototype._skipCache = function (cacheId) {
360+
cacheId = cacheId || 'root'
361+
if (this._eventBuffers[cacheId]) {
362+
this._eventBuffers[cacheId].severityBuf.clear()
363+
this._eventBuffers[cacheId].eventBuf.clear()
364+
this._eventBuffers[cacheId].skipped = true
365+
}
366+
}
367+
359368
Collector.prototype._deleteCache = function (cacheId) {
360369
cacheId = cacheId || 'root'
361370
delete this._eventBuffers[cacheId]
362371
}
363372

364-
Collector.prototype._lockCache = function (cacheId) {
373+
Collector.prototype._addLockToCache = function (cacheId) {
365374
cacheId = cacheId || 'root'
366375
if (this._eventBuffers[cacheId]) {
367-
this._eventBuffers[cacheId].locked = true
376+
++this._eventBuffers[cacheId].locks
368377
}
369378
}
370379

371-
Collector.prototype._unlockCache = function (cacheId) {
380+
Collector.prototype._removeLockFromCache = function (cacheId) {
372381
cacheId = cacheId || 'root'
373382
if (this._eventBuffers[cacheId]) {
374-
this._eventBuffers[cacheId].locked = false
383+
if (this._eventBuffers[cacheId].locks > 0) {
384+
--this._eventBuffers[cacheId].locks
385+
}
375386
}
376387
}
377388

378-
Collector.prototype._flushCache = function (cacheId) {
389+
Collector.prototype._flushUnlockedCache = function (cacheId) {
379390
cacheId = cacheId || 'root'
380391
var self = this
381-
if (this._eventBuffers[cacheId]) {
382-
if (some(this._eventBuffers[cacheId].severityBuf.elements(), function (severity) {
383-
return levels.gte(severity, self.mustCollectSeverity)
384-
})) {
392+
if (this._eventBuffers[cacheId] && this._eventBuffers[cacheId].locks === 0) {
393+
if (!this._eventBuffers[cacheId].skipped &&
394+
some(this._eventBuffers[cacheId].severityBuf.elements(), function (severity) {
395+
return levels.gte(severity, self.mustCollectSeverity)
396+
})) {
385397
var events = this._eventBuffers[cacheId].eventBuf.flush()
386398
if (events.length) {
387399
this._sampler.add(events)
@@ -404,12 +416,13 @@ Collector.prototype._getSeverityHintFromCache = function (cacheId) {
404416
Collector.prototype.collect = function () {
405417
var self = this
406418
Object.keys(this._eventBuffers).forEach(function (key) {
407-
if (self._eventBuffers[key].locked === true) {
419+
if (self._eventBuffers[key].locks > 0) {
408420
return
409421
} else {
410-
if (some(self._eventBuffers[key].severityBuf.elements(), function (severity) {
411-
return levels.gte(severity, self.mustCollectSeverity)
412-
})) {
422+
if (!self._eventBuffers[key].skipped &&
423+
some(self._eventBuffers[key].severityBuf.elements(), function (severity) {
424+
return levels.gte(severity, self.mustCollectSeverity)
425+
})) {
413426
var events = self._eventBuffers[key].eventBuf.flush()
414427
if (events.length) {
415428
self._sampler.add(events)
@@ -418,7 +431,8 @@ Collector.prototype.collect = function () {
418431
self._eventBuffers[key].eventBuf.expire()
419432
self._eventBuffers[key].severityBuf.expire()
420433

421-
if (self._eventBuffers[key].severityBuf.isEmpty()) {
434+
if (self._eventBuffers[key].skipped ||
435+
self._eventBuffers[key].severityBuf.isEmpty()) {
422436
delete self._eventBuffers[key]
423437
}
424438
}

‎lib/agent/tracer/collector.spec.js‎

Lines changed: 87 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -342,7 +342,7 @@ describe('Collector', function () {
342342
severity: tracer.defaultSeverity
343343
})
344344
})
345-
it('is collected', function () {
345+
it('is not collected on it\'s own', function () {
346346
var tracer = new Collector(options)
347347
var payload = {
348348
protocol: 'http',
@@ -368,18 +368,7 @@ describe('Collector', function () {
368368
return 2
369369
})
370370
tracer.clientSend(payload, briefcase)
371-
expect(tracer.collect()).to.eql([{
372-
t: 'cs',
373-
r: 'transactionId',
374-
i: 2,
375-
p: id,
376-
c: 'http',
377-
a: 'communicationId',
378-
ac: 'action',
379-
e: 'resource',
380-
h: 'host',
381-
d: {}
382-
}])
371+
expect(tracer.collect().length).to.eql(0)
383372
})
384373
it('is not collected (in itself) when not in a transaction', function () {
385374
var tracer = new Collector(options)
@@ -447,7 +436,7 @@ describe('Collector', function () {
447436
tracer.clientRecv(payload, duffelBag, briefcase)
448437
expect(tracer.collect()).to.eql([])
449438
})
450-
it('is collected', function () {
439+
it('is collected on it\'s own', function () {
451440
this.sandbox.stub(microtime, 'now', function () {
452441
return 2
453442
})
@@ -475,18 +464,7 @@ describe('Collector', function () {
475464
}
476465

477466
tracer.clientRecv(payload, duffelBag, briefcase)
478-
expect(tracer.collect()).to.eql([{
479-
t: 'cr',
480-
r: 'id',
481-
i: 2,
482-
p: 'child-id',
483-
o: 1,
484-
k: undefined,
485-
c: 'http',
486-
a: 'id',
487-
s: 'ok',
488-
d: { statusCode: 200 }
489-
}])
467+
expect(tracer.collect().length).to.eql(1)
490468
})
491469
it('is collected because duffelBag raises severity', function () {
492470
this.sandbox.stub(microtime, 'now', function () {
@@ -521,6 +499,31 @@ describe('Collector', function () {
521499
})
522500
describe('SR - CS', function () {
523501
it('SR mustCollect implies CS mustCollect', function () {
502+
var tracer = new Collector(options)
503+
var sr = {
504+
payload: {},
505+
duffelBag: {
506+
communicationId: 'communicationId',
507+
transactionId: 'transactionId',
508+
severity: tracer.mustCollectSeverity,
509+
parentServiceKey: 8,
510+
timestamp: 1
511+
}
512+
}
513+
var cs = {
514+
payload: {},
515+
briefcase: {
516+
communication: {
517+
id: sr.duffelBag.communicationId,
518+
transactionId: sr.duffelBag.transactionId
519+
}
520+
}
521+
}
522+
tracer.serverRecv(sr.payload, sr.duffelBag)
523+
var csResult = tracer.clientSend(cs.payload, cs.briefcase)
524+
expect(csResult.duffelBag.severity).to.eql(tracer.mustCollectSeverity)
525+
})
526+
it('is not collected', function () {
524527
var tracer = new Collector(options)
525528
var sr = {
526529
payload: {},
@@ -544,7 +547,7 @@ describe('Collector', function () {
544547
var srResult = tracer.serverRecv(sr.payload, sr.duffelBag)
545548
tracer.clientSend(cs.payload, cs.briefcase)
546549
tracer.end(srResult.briefcase)
547-
expect(tracer.collect().length).to.eql(2)
550+
expect(tracer.collect().length).to.eql(0)
548551
})
549552
})
550553

@@ -570,8 +573,64 @@ describe('Collector', function () {
570573
var srResult = tracer.serverRecv(sr.payload, sr.duffelBag)
571574
var csResult = tracer.clientSend(cs.payload, srResult.briefcase)
572575
tracer.networkError(csResult.briefcase, new Error())
576+
expect(tracer.collect().length).to.eql(0)
573577
tracer.serverSend(ss.payload, srResult.briefcase)
574-
tracer.end(srResult.briefcase)
578+
expect(tracer.collect().length).to.eql(4)
579+
})
580+
})
581+
582+
describe('SR - CS - SS - NE', function () {
583+
it('is collected', function () {
584+
var tracer = new Collector(options)
585+
var sr = {
586+
payload: {},
587+
duffelBag: {
588+
communicationId: 'communicationId',
589+
transactionId: 'transactionId',
590+
parentServiceKey: 8,
591+
timestamp: 1
592+
}
593+
}
594+
var cs = {
595+
payload: {}
596+
}
597+
var ss = {
598+
payload: {}
599+
}
600+
601+
var srResult = tracer.serverRecv(sr.payload, sr.duffelBag)
602+
var csResult = tracer.clientSend(cs.payload, srResult.briefcase)
603+
tracer.serverSend(ss.payload, srResult.briefcase)
604+
expect(tracer.collect().length).to.eql(0)
605+
tracer.networkError(csResult.briefcase, new Error())
606+
expect(tracer.collect().length).to.eql(4)
607+
})
608+
})
609+
610+
describe('SR - CS - SS - CR', function () {
611+
it('is collected', function () {
612+
var tracer = new Collector(options)
613+
var sr = {
614+
payload: {},
615+
duffelBag: {
616+
communicationId: 'communicationId',
617+
transactionId: 'transactionId',
618+
parentServiceKey: 8,
619+
timestamp: 1
620+
}
621+
}
622+
var cs = {
623+
payload: {}
624+
}
625+
var ss = {
626+
payload: {}
627+
}
628+
629+
var srResult = tracer.serverRecv(sr.payload, sr.duffelBag)
630+
var csResult = tracer.clientSend(cs.payload, srResult.briefcase)
631+
tracer.serverSend(ss.payload, srResult.briefcase)
632+
expect(tracer.collect().length).to.eql(0)
633+
tracer.clientRecv({}, { severity: tracer.mustCollectSeverity }, csResult.briefcase)
575634
expect(tracer.collect().length).to.eql(4)
576635
})
577636
})

‎lib/instrumentations/core/http/server.js‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -100,18 +100,25 @@ function httpServer (listener, agent) {
100100
transportDelay: timestamp != null ? srTime - timestamp : undefined
101101
})
102102

103-
var writeHeadWrapped = false
103+
var writeHeadAlreadyCalled = false
104+
var alreadyEnded = false // currently there is no start, SR acts as an end
104105

105106
response.on('finish', function () {
106-
collector.end(sr.briefcase)
107+
if (!alreadyEnded) {
108+
collector.end(sr.briefcase)
109+
}
107110
})
108111
response.on('close', function () {
109-
collector.end(sr.briefcase)
112+
if (!alreadyEnded) {
113+
collector.end(sr.briefcase)
114+
}
110115
})
111116

112117
Shimmer.wrap(response, 'writeHead', function wrapWriteHead (original, name) {
113118
return function onWriteHead () {
114-
if (writeHeadWrapped === false) {
119+
if (writeHeadAlreadyCalled === false) {
120+
alreadyEnded = true
121+
writeHeadAlreadyCalled = true
115122
var ssTime = microtime.now()
116123
var statusCode = response.statusCode || arguments[0]
117124
var tracerBriefcase = agent.storage.get('tracer.briefcase')
@@ -153,7 +160,6 @@ function httpServer (listener, agent) {
153160
}
154161
agent.rpmMetrics.addResponseTime(ssTime - srTime)
155162
agent.rpmMetrics.addStatusCode(response.statusCode)
156-
writeHeadWrapped = true
157163
}
158164
return original.apply(response, arguments)
159165
}

0 commit comments

Comments
 (0)