You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ke...@apache.org on 2020/11/30 01:28:04 UTC
[skywalking-nodejs] branch master updated: Make async stuff work
reliably (#2)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking-nodejs.git
The following commit(s) were added to refs/heads/master by this push:
new 0a277f6 Make async stuff work reliably (#2)
0a277f6 is described below
commit 0a277f6a166a1a3eb92fce3bb461c7563ee7bf8d
Author: Tomasz Pytel <to...@gmail.com>
AuthorDate: Sun Nov 29 22:27:55 2020 -0300
Make async stuff work reliably (#2)
---
src/agent/Buffer.ts | 2 +
.../protocol/grpc/clients/TraceReportClient.ts | 9 ++-
src/index.ts | 1 +
src/plugins/HttpPlugin.ts | 68 ++++++++++++--------
src/trace/context/Context.ts | 14 ++--
src/trace/context/ContextManager.ts | 68 ++++++++++++++++++--
src/trace/context/DummyContext.ts | 23 +++----
src/trace/context/SpanContext.ts | 74 +++++++++++++++-------
src/trace/span/EntrySpan.ts | 7 +-
src/trace/span/ExitSpan.ts | 8 +--
src/trace/span/Span.ts | 30 ++++-----
src/trace/span/StackedSpan.ts | 7 ++
tests/plugins/http/expected.data.yaml | 6 +-
13 files changed, 209 insertions(+), 108 deletions(-)
diff --git a/src/agent/Buffer.ts b/src/agent/Buffer.ts
index 6f0736c..a1cda3b 100644
--- a/src/agent/Buffer.ts
+++ b/src/agent/Buffer.ts
@@ -20,6 +20,7 @@
import { createLogger } from '../logging';
import Segment from '../trace/context/Segment';
import config from '../config/AgentConfig';
+import TraceReportClient from '../agent/protocol/grpc/clients/TraceReportClient';
const logger = createLogger(__filename);
@@ -42,6 +43,7 @@ class Buffer {
return this;
}
this.buffer.push(segment);
+ TraceReportClient.ref(); // this is currently hard-coded for grpc, if other protocols added need to change
return this;
}
diff --git a/src/agent/protocol/grpc/clients/TraceReportClient.ts b/src/agent/protocol/grpc/clients/TraceReportClient.ts
index a05eb82..13be3a2 100644
--- a/src/agent/protocol/grpc/clients/TraceReportClient.ts
+++ b/src/agent/protocol/grpc/clients/TraceReportClient.ts
@@ -31,6 +31,7 @@ const logger = createLogger(__filename);
class TraceReportClient implements Client {
reporterClient: TraceSegmentReportServiceClient;
+ timeout: any;
constructor() {
this.reporterClient = new TraceSegmentReportServiceClient(
@@ -44,6 +45,10 @@ class TraceReportClient implements Client {
return this.reporterClient.getChannel().getConnectivityState(true) === connectivityState.READY;
}
+ ref() {
+ this.timeout.ref();
+ }
+
start() {
const reportFunction = () => {
try {
@@ -72,11 +77,11 @@ class TraceReportClient implements Client {
stream.end();
} finally {
- setTimeout(reportFunction, 1000).unref();
+ this.timeout = setTimeout(reportFunction, 1000).unref();
}
};
- setTimeout(reportFunction, 1000).unref();
+ this.timeout = setTimeout(reportFunction, 1000).unref();
}
}
diff --git a/src/index.ts b/src/index.ts
index 413a8ac..87c35cd 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -52,3 +52,4 @@ class Agent {
}
export default new Agent();
+export { default as ContextManager } from './trace/context/ContextManager';
diff --git a/src/plugins/HttpPlugin.ts b/src/plugins/HttpPlugin.ts
index 99e5861..87c0891 100644
--- a/src/plugins/HttpPlugin.ts
+++ b/src/plugins/HttpPlugin.ts
@@ -23,6 +23,7 @@ import { ClientRequest, IncomingMessage, RequestOptions, ServerResponse } from '
import ContextManager from '../trace/context/ContextManager';
import { Component } from '../trace/Component';
import Tag from '../Tag';
+import Span from '../trace/span/Span';
import { SpanLayer } from '../proto/language-agent/Tracing_pb';
import { ContextCarrier } from '../trace/context/ContextCarrier';
import { createLogger } from '../logging';
@@ -59,26 +60,42 @@ class HttpPlugin implements SwPlugin {
};
const operation = pathname.replace(/\?.*$/g, '');
- const span = ContextManager.current.newExitSpan(operation, host).start();
- span.component = Component.HTTP;
- span.layer = SpanLayer.HTTP;
- span.tag(Tag.httpURL(host + pathname));
+ const [span, request]: [Span, ClientRequest] = ContextManager.withSpanNoStop(
+ ContextManager.current.newExitSpan(operation, host), (_span: Span, args) => {
- const request: ClientRequest = original.apply(this, arguments);
+ _span.component = Component.HTTP;
+ _span.layer = SpanLayer.HTTP;
+ _span.tag(Tag.httpURL(host + pathname));
- span.extract().items.forEach((item) => {
- request.setHeader(item.key, item.value);
- });
+ const _request = original.apply(this, args);
+
+ _span.extract().items.forEach((item) => {
+ _request.setHeader(item.key, item.value);
+ });
+
+ return [_span, _request];
+
+ }, arguments);
span.async();
+ let stopped = 0; // compensating if request aborted right after creation 'close' is not emitted
+ const stopIfNotStopped = () => !stopped++ ? span.stop() : null;
+ request.on('abort', stopIfNotStopped); // make sure we stop only once
+ request.on('close', stopIfNotStopped);
+ request.on('error', stopIfNotStopped);
+
request.prependListener('response', (res) => {
+ span.resync();
span.tag(Tag.httpStatusCode(res.statusCode));
if (res.statusCode && res.statusCode >= 400) {
span.errored = true;
}
+ if (res.statusMessage) {
+ span.tag(Tag.httpStatusMsg(res.statusMessage));
+ }
+ stopIfNotStopped();
});
- request.on('close', () => span.await().stop());
return request;
};
@@ -94,26 +111,26 @@ class HttpPlugin implements SwPlugin {
return original.apply(this, arguments);
}
- const args = arguments;
- const self = this;
+ const [req, res] = [arguments[1] as IncomingMessage, arguments[2] as ServerResponse];
- return ContextManager.withContext(() => {
- const [req, res] = [args[1] as IncomingMessage, args[2] as ServerResponse];
+ const headers = req.rawHeaders || [];
+ const headersMap: { [key: string]: string } = {};
- const headers = req.rawHeaders || [];
- const headersMap: { [key: string]: string } = {};
+ for (let i = 0; i < headers.length / 2; i += 2) {
+ headersMap[headers[i]] = headers[i + 1];
+ }
- for (let i = 0; i < headers.length / 2; i += 2) {
- headersMap[headers[i]] = headers[i + 1];
- }
+ const carrier = ContextCarrier.from(headersMap);
+ const operation = (req.url || '/').replace(/\?.*/g, '');
- const carrier = ContextCarrier.from(headersMap);
+ return ContextManager.withSpan(ContextManager.current.newEntrySpan(operation, carrier),
+ (span, self, args) => {
- const operation = (req.url || '/').replace(/\?.*/g, '');
- const span = ContextManager.current.newEntrySpan(operation, carrier).start();
span.component = Component.HTTP_SERVER;
span.layer = SpanLayer.HTTP;
- span.tag(Tag.httpURL(req.url));
+ span.tag(Tag.httpURL((req.headers.host || '') + req.url));
+
+ const ret = original.apply(self, args);
span.tag(Tag.httpStatusCode(res.statusCode));
if (res.statusCode && res.statusCode >= 400) {
@@ -123,12 +140,9 @@ class HttpPlugin implements SwPlugin {
span.tag(Tag.httpStatusMsg(res.statusMessage));
}
- res.on('close', () => {
- span.stop();
- });
+ return ret;
- return original.apply(self, args);
- });
+ }, this, arguments);
};
})(http.Server.prototype.emit);
}
diff --git a/src/trace/context/Context.ts b/src/trace/context/Context.ts
index 984033d..d1b234e 100644
--- a/src/trace/context/Context.ts
+++ b/src/trace/context/Context.ts
@@ -24,7 +24,6 @@ import { ContextCarrier } from './ContextCarrier';
export default interface Context {
segment: Segment;
- spans: Span[];
newLocalSpan(operation: string): Span;
@@ -36,13 +35,18 @@ export default interface Context {
stop(span: Span): boolean;
+ /* This should be called just before a span is passed to a different async context, like for example a callback from
+ an asynchronous operation the code belonging to the span initiated. After this is called a span should only call
+ .resync() or .stop(). See HttpPlugin.interceptClientRequest() in plugins/HttpPlugin.ts for example of usage. */
+ async(span: Span): void;
+
+ /* This should be called upon entering the new async context for a span that has previously executed .async(), it
+ should be the first thing the callback function belonging to the span does. */
+ resync(span: Span): void;
+
currentSpan(): Span | undefined;
capture(): Snapshot;
restore(snapshot: Snapshot): void;
-
- async(span: Span): void;
-
- await(span: Span): void;
}
diff --git a/src/trace/context/ContextManager.ts b/src/trace/context/ContextManager.ts
index 538a6d1..b22ab60 100644
--- a/src/trace/context/ContextManager.ts
+++ b/src/trace/context/ContextManager.ts
@@ -18,18 +18,76 @@
*/
import Context from '../../trace/context/Context';
+import Span from '../../trace/span/Span';
import SpanContext from '../../trace/context/SpanContext';
import { AsyncLocalStorage } from 'async_hooks';
-const store = new AsyncLocalStorage<Context>();
+type AsyncState = {context: Context, spans: Span[]};
+
+const store = new AsyncLocalStorage<AsyncState>();
class ContextManager {
- get current(): Context {
- return store.getStore() || new SpanContext();
+ get asyncState(): AsyncState {
+ let asyncState = store.getStore();
+
+ if (asyncState === undefined) {
+ asyncState = {context: new SpanContext(), spans: []};
+ store.enterWith(asyncState);
+ }
+
+ return asyncState;
+ }
+
+ get current(): Context { return this.asyncState.context; }
+ get spans(): Span[] { return this.asyncState.spans; }
+
+ spansDup(): Span[] {
+ let asyncState = store.getStore();
+
+ if (asyncState === undefined) {
+ asyncState = {context: new SpanContext(), spans: []};
+ } else {
+ asyncState = {context: asyncState.context, spans: [...asyncState.spans]};
+ }
+
+ store.enterWith(asyncState);
+
+ return asyncState.spans;
}
- withContext(callback: (...args: any[]) => void, ...args: any[]) {
- return store.run(new SpanContext(), callback);
+ clear(): void {
+ store.enterWith(undefined as unknown as AsyncState);
+ }
+
+ restore(context: Context, spans: Span[]): void {
+ store.enterWith({context, spans: spans || []});
+ }
+
+ withSpan(span: Span, callback: (...args: any[]) => any, ...args: any[]): any {
+ if(!span.startTime)
+ span.start();
+
+ try {
+ return callback(span, ...args);
+ } catch (e) {
+ span.error(e);
+ throw e;
+ } finally {
+ span.stop();
+ }
+ }
+
+ withSpanNoStop(span: Span, callback: (...args: any[]) => any, ...args: any[]): any {
+ if(!span.startTime)
+ span.start();
+
+ try {
+ return callback(span, ...args);
+ } catch (e) {
+ span.error(e);
+ span.stop();
+ throw e;
+ }
}
}
diff --git a/src/trace/context/DummyContext.ts b/src/trace/context/DummyContext.ts
index e7fc720..68c44cf 100644
--- a/src/trace/context/DummyContext.ts
+++ b/src/trace/context/DummyContext.ts
@@ -32,26 +32,23 @@ export default class DummyContext implements Context {
operation: '',
type: SpanType.LOCAL,
});
- spans: Span[] = [];
segment: Segment = new Segment();
depth = 0;
newEntrySpan(operation: string, carrier?: ContextCarrier): Span {
- this.depth++;
return this.span;
}
newExitSpan(operation: string, peer: string, carrier?: ContextCarrier): Span {
- this.depth++;
return this.span;
}
newLocalSpan(operation: string): Span {
- this.depth++;
return this.span;
}
start(): Context {
+ this.depth++;
return this;
}
@@ -59,8 +56,16 @@ export default class DummyContext implements Context {
return --this.depth === 0;
}
+ async(span: Span) {
+ return;
+ }
+
+ resync(span: Span) {
+ return;
+ }
+
currentSpan(): Span {
- return this.spans[this.spans.length - 1];
+ throw new Error('DummyContext.currentSpan() should never be called!');
}
capture(): Snapshot {
@@ -75,12 +80,4 @@ export default class DummyContext implements Context {
restore(snapshot: Snapshot) {
return;
}
-
- async(span: Span) {
- return;
- }
-
- await(span: Span) {
- return;
- }
}
diff --git a/src/trace/context/SpanContext.ts b/src/trace/context/SpanContext.ts
index 87b7443..fab8305 100644
--- a/src/trace/context/SpanContext.ts
+++ b/src/trace/context/SpanContext.ts
@@ -29,18 +29,18 @@ import { executionAsyncId } from 'async_hooks';
import Snapshot from '../../trace/context/Snapshot';
import SegmentRef from '../../trace/context/SegmentRef';
import { ContextCarrier } from './ContextCarrier';
+import ContextManager from './ContextManager';
const logger = createLogger(__filename);
export default class SpanContext implements Context {
spanId = 0;
- spans: Span[] = [];
+ nSpans = 0;
segment: Segment = new Segment();
- asyncCount: number = 0;
get parent(): Span | null {
- if (this.spans.length > 0) {
- return this.spans[this.spans.length - 1];
+ if (ContextManager.spans.length > 0) {
+ return ContextManager.spans[ContextManager.spans.length - 1];
}
return null;
}
@@ -57,6 +57,8 @@ export default class SpanContext implements Context {
});
}
+ ContextManager.spansDup();
+
const span = new EntrySpan({
id: this.spanId++,
parentId: this.parentId,
@@ -79,6 +81,8 @@ export default class SpanContext implements Context {
});
}
+ ContextManager.spansDup();
+
return new ExitSpan({
id: this.spanId++,
parentId: this.parentId,
@@ -96,6 +100,8 @@ export default class SpanContext implements Context {
});
}
+ ContextManager.spansDup();
+
return new LocalSpan({
id: this.spanId++,
parentId: this.parentId,
@@ -105,32 +111,61 @@ export default class SpanContext implements Context {
}
start(span: Span): Context {
- if (this.spans.every((s) => s.id !== span.id)) {
- this.spans.push(span);
+ logger.debug('Starting span', { span: span.operation, spans: ContextManager.spans, nSpans: this.nSpans });
+
+ this.nSpans += 1;
+ if (ContextManager.spans.every((s) => s.id !== span.id)) {
+ ContextManager.spans.push(span);
}
return this;
}
stop(span: Span): boolean {
- logger.info('Stopping span', { span: span.operation, spans: this.spans });
+ logger.debug('Stopping span', { span: span.operation, spans: ContextManager.spans, nSpans: this.nSpans });
if (span.finish(this.segment)) {
- const idx = this.spans.indexOf(span);
- if (idx >= 0) {
- this.spans.splice(idx, 1);
+ const idx = ContextManager.spans.indexOf(span);
+ if (idx !== -1) {
+ ContextManager.spans.splice(idx, 1);
}
}
- const finished = this.asyncCount === 0 && this.spans.length === 0;
- if (finished) {
+ if (--this.nSpans == 0) {
buffer.put(this.segment);
+ ContextManager.clear();
+ return true;
+ }
+
+ return false;
+ }
+
+ async(span: Span) {
+ logger.debug('Async span', { span: span.operation, spans: ContextManager.spans, nSpans: this.nSpans });
+
+ const idx = ContextManager.spans.indexOf(span);
+
+ if (idx !== -1) {
+ ContextManager.spans.splice(idx, 1);
+ }
+
+ if (this.nSpans === 1) { // this will pass the context to child async task so it doesn't mess with other tasks here
+ ContextManager.clear();
+ }
+ }
+
+ resync(span: Span) {
+ logger.debug('Resync span', { span: span.operation, spans: ContextManager.spans, nSpans: this.nSpans });
+
+ if ((span.context as SpanContext).nSpans === 1) {
+ ContextManager.restore(span.context, [span]);
+ } else if (ContextManager.spans.every((s) => s.id !== span.id)) {
+ ContextManager.spans.push(span);
}
- return finished;
}
currentSpan(): Span | undefined {
- return this.spans[this.spans.length - 1];
+ return ContextManager.spans[ContextManager.spans.length - 1];
}
capture(): Snapshot {
@@ -138,7 +173,7 @@ export default class SpanContext implements Context {
segmentId: this.segment.segmentId,
spanId: this.currentSpan()?.id ?? -1,
traceId: this.segment.relatedTraces[0],
- parentEndpoint: this.spans[0].operation,
+ parentEndpoint: ContextManager.spans[0].operation,
};
}
@@ -148,13 +183,4 @@ export default class SpanContext implements Context {
this.currentSpan()?.refer(ref);
this.segment.relate(ref.traceId);
}
-
- async(span: Span) {
- this.asyncCount++;
- this.spans.splice(this.spans.indexOf(span), 1);
- }
-
- await(span: Span) {
- this.asyncCount--;
- }
}
diff --git a/src/trace/span/EntrySpan.ts b/src/trace/span/EntrySpan.ts
index bede555..4b8126e 100644
--- a/src/trace/span/EntrySpan.ts
+++ b/src/trace/span/EntrySpan.ts
@@ -25,8 +25,6 @@ import { SpanLayer, SpanType } from '../../proto/language-agent/Tracing_pb';
import { ContextCarrier } from '../context/ContextCarrier';
export default class EntrySpan extends StackedSpan {
- maxDepth = 0;
-
constructor(options: SpanCtorOptions) {
super(
Object.assign(options, {
@@ -36,10 +34,7 @@ export default class EntrySpan extends StackedSpan {
}
start(): this {
- this.maxDepth = ++this.depth;
- if (this.maxDepth === 1) {
- super.start();
- }
+ super.start();
this.layer = SpanLayer.UNKNOWN;
this.component = Component.UNKNOWN;
this.logs.splice(0, this.logs.length);
diff --git a/src/trace/span/ExitSpan.ts b/src/trace/span/ExitSpan.ts
index 5b3fb8b..8c16db6 100644
--- a/src/trace/span/ExitSpan.ts
+++ b/src/trace/span/ExitSpan.ts
@@ -22,6 +22,7 @@ import { SpanCtorOptions } from './Span';
import config from '../../config/AgentConfig';
import { SpanType } from '../../proto/language-agent/Tracing_pb';
import { ContextCarrier } from '../context/ContextCarrier';
+import ContextManager from '../context/ContextManager';
export default class ExitSpan extends StackedSpan {
constructor(options: SpanCtorOptions) {
@@ -32,11 +33,6 @@ export default class ExitSpan extends StackedSpan {
);
}
- start(): this {
- this.depth++;
- return super.start();
- }
-
extract(): ContextCarrier {
return new ContextCarrier(
this.context.segment.relatedTraces[0],
@@ -44,7 +40,7 @@ export default class ExitSpan extends StackedSpan {
this.id,
config.serviceName,
config.serviceInstance,
- this.context.spans[0].operation,
+ ContextManager.spans[0].operation,
this.peer,
[],
);
diff --git a/src/trace/span/Span.ts b/src/trace/span/Span.ts
index 7ab5d8b..d854e9a 100644
--- a/src/trace/span/Span.ts
+++ b/src/trace/span/Span.ts
@@ -59,8 +59,6 @@ export default abstract class Span {
endTime = 0;
errored = false;
- _async = false;
-
constructor(options: SpanCtorOptions & { type: SpanType }) {
this.context = options.context;
this.operation = options.operation;
@@ -89,6 +87,18 @@ export default abstract class Span {
return this;
}
+ async(): this {
+ logger.debug(`Async span ${this.operation}`, this);
+ this.context.async(this);
+ return this;
+ }
+
+ resync(): this {
+ logger.debug(`Resync span ${this.operation}`, this);
+ this.context.resync(this);
+ return this;
+ }
+
finish(segment: Segment): boolean {
logger.debug('Finishing span', this);
this.endTime = new Date().getTime();
@@ -146,20 +156,4 @@ export default abstract class Span {
}
return this;
}
-
- async(): this {
- this._async = true;
- this.context.async(this);
- return this;
- }
-
- await(): this {
- this.context.await(this);
- this._async = false;
- return this;
- }
-
- get isAsync(): boolean {
- return this._async;
- }
}
diff --git a/src/trace/span/StackedSpan.ts b/src/trace/span/StackedSpan.ts
index e6d02a9..a06f1c2 100644
--- a/src/trace/span/StackedSpan.ts
+++ b/src/trace/span/StackedSpan.ts
@@ -31,6 +31,13 @@ export default class StackedSpan extends Span {
super(options);
}
+ start(): this {
+ if (++this.depth === 1) {
+ super.start();
+ }
+ return this;
+ }
+
finish(segment: Segment): boolean {
return --this.depth === 0 && super.finish(segment);
}
diff --git a/tests/plugins/http/expected.data.yaml b/tests/plugins/http/expected.data.yaml
index a0e03d3..288515b 100644
--- a/tests/plugins/http/expected.data.yaml
+++ b/tests/plugins/http/expected.data.yaml
@@ -34,7 +34,7 @@ segmentItems:
peer: ''
skipAnalysis: false
tags:
- - { key: http.url, value: /test }
+ - { key: http.url, value: server:5000/test }
- { key: http.status.code, value: '200' }
refs:
- { parentEndpoint: '', networkAddress: 'server:5000', refType: CrossProcess,
@@ -55,6 +55,7 @@ segmentItems:
tags:
- { key: http.url, value: httpbin.org/json }
- { key: http.status.code, value: '200' }
+ - { key: http.status.msg, value: OK }
- serviceName: client
segmentSize: 1
segments:
@@ -73,7 +74,7 @@ segmentItems:
peer: ''
skipAnalysis: false
tags:
- - { key: http.url, value: /test }
+ - { key: http.url, value: localhost:5001/test }
- { key: http.status.code, value: '200' }
- operationName: /test
operationId: 0
@@ -90,3 +91,4 @@ segmentItems:
tags:
- { key: http.url, value: 'server:5000/test' }
- { key: http.status.code, value: '200' }
+ - { key: http.status.msg, value: OK }