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/09/30 06:36:45 UTC
[skywalking-nodejs] 01/01: Try to stablize the first version
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch stablize
in repository https://gitbox.apache.org/repos/asf/skywalking-nodejs.git
commit 8ddad77b4699ea1aa8ac89f1bb50ab8044cb4175
Author: kezhenxu94 <ke...@163.com>
AuthorDate: Wed Sep 30 14:36:33 2020 +0800
Try to stablize the first version
---
package-lock.json | 20 ++-
package.json | 2 +
src/plugins/HttpPlugin.ts | 27 +--
src/plugins/MongodbPlugin.ts | 182 +++++++++++++++++++++
src/trace/context/Context.ts | 4 +
src/trace/context/ContextManager.ts | 23 +--
src/trace/context/DummyContext.ts | 8 +
src/trace/context/SpanContext.ts | 27 ++-
src/trace/span/Span.ts | 18 ++
tests/dev.ts | 59 +++++++
tests/plugins/http/client.ts | 57 +++++++
.../Context.ts => tests/plugins/http/server.ts | 31 ++--
12 files changed, 404 insertions(+), 54 deletions(-)
diff --git a/package-lock.json b/package-lock.json
index ca345e1..2b69f83 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1,6 +1,6 @@
{
- "name": "@ali/skywalking-nodejs",
- "version": "0.0.21-RC",
+ "name": "skywalking",
+ "version": "0.0.1",
"lockfileVersion": 1,
"requires": true,
"dependencies": {
@@ -60,6 +60,14 @@
"integrity": "sha512-rr+OQyAjxze7GgWrSaJwydHStIhHq2lvY3BOC2Mj7KnzI7XK0Uw1TOOdI9lDoajEbSWLiYgoo4f1R51erQfhPQ==",
"dev": true
},
+ "@types/end-of-stream": {
+ "version": "1.4.0",
+ "resolved": "https://registry.npmjs.org/@types/end-of-stream/-/end-of-stream-1.4.0.tgz",
+ "integrity": "sha512-d0FD2A4vpFI8wyQeQbr9VDVKtA1PmeGO3Ntn+6j626QTtAQ9HSqWFACP7rTHaV2cspVhLijl00Vvkf/U2UZGWA==",
+ "requires": {
+ "@types/node": "*"
+ }
+ },
"@types/google-protobuf": {
"version": "3.7.2",
"resolved": "https://registry.npmjs.org/@types/google-protobuf/-/google-protobuf-3.7.2.tgz",
@@ -1167,6 +1175,14 @@
"env-variable": "0.0.x"
}
},
+ "end-of-stream": {
+ "version": "1.4.4",
+ "resolved": "https://registry.npmjs.org/end-of-stream/-/end-of-stream-1.4.4.tgz",
+ "integrity": "sha512-+uw1inIHVPQoaVuHzRyXd21icM+cnt4CzD5rW+NC1wjOUSTOs+Te7FOv7AhN7vS9x/oIyhLP5PR1H+phQAHu5Q==",
+ "requires": {
+ "once": "^1.4.0"
+ }
+ },
"ent": {
"version": "2.2.0",
"resolved": "https://registry.npmjs.org/ent/-/ent-2.2.0.tgz",
diff --git a/package.json b/package.json
index 69656c5..1ceab2e 100644
--- a/package.json
+++ b/package.json
@@ -55,6 +55,8 @@
"worker-loader": "^2.0.0"
},
"dependencies": {
+ "@types/end-of-stream": "^1.4.0",
+ "end-of-stream": "^1.4.4",
"google-protobuf": "^3.12.2",
"grpc": "^1.10.1",
"semver": "^7.3.2",
diff --git a/src/plugins/HttpPlugin.ts b/src/plugins/HttpPlugin.ts
index a7caf18..5a7f39d 100644
--- a/src/plugins/HttpPlugin.ts
+++ b/src/plugins/HttpPlugin.ts
@@ -26,6 +26,7 @@ import Tag from '../Tag';
import { SpanLayer } from '../proto/language-agent/Tracing_pb';
import { ContextCarrier } from '../trace/context/ContextCarrier';
import { createLogger } from '../logging';
+import * as eos from 'end-of-stream';
const logger = createLogger(__filename);
@@ -46,7 +47,6 @@ class HttpPlugin implements SwPlugin {
const argc = arguments.length;
const url: URL | string | RequestOptions = arguments[0];
- const options = argc > 1 ? (typeof arguments[1] === 'function' ? {} : arguments[1]) : {};
const callback = typeof arguments[argc - 1] === 'function' ? arguments[argc - 1] : undefined;
const { host, pathname } =
@@ -65,32 +65,19 @@ class HttpPlugin implements SwPlugin {
span.layer = SpanLayer.HTTP;
span.tag(Tag.httpURL(host + pathname));
- const snapshot = ContextManager.current.capture();
-
const request: ClientRequest = original.apply(this, arguments);
span.extract().items.forEach((item) => {
request.setHeader(item.key, item.value);
});
- request.on('response', (res) => {
- res.prependListener('end', () => {
- span.tag(Tag.httpStatusCode(res.statusCode)).tag(Tag.httpStatusMsg(res.statusMessage));
-
- const callbackSpan = ContextManager.current.newLocalSpan('callback').start();
- callbackSpan.layer = SpanLayer.HTTP;
- callbackSpan.component = Component.HTTP;
+ span.async().stop();
- ContextManager.current.restore(snapshot);
-
- if (callback) {
- callback(res);
- }
+ request.on('response', (res) => {
+ span.tag(Tag.httpStatusCode(res.statusCode)).tag(Tag.httpStatusMsg(res.statusMessage));
- callbackSpan.stop();
- });
+ eos(res, () => span.await());
});
- span.stop();
return request;
};
@@ -123,7 +110,9 @@ class HttpPlugin implements SwPlugin {
span.layer = SpanLayer.HTTP;
span.tag(Tag.httpURL(req.url));
- span.tag(Tag.httpStatusCode(res.statusCode)).tag(Tag.httpStatusMsg(res.statusMessage)).stop();
+ span.tag(Tag.httpStatusCode(res.statusCode)).tag(Tag.httpStatusMsg(res.statusMessage)).async().stop();
+
+ eos(res, () => span.await());
return original.apply(this, arguments);
};
diff --git a/src/plugins/MongodbPlugin.ts b/src/plugins/MongodbPlugin.ts
new file mode 100644
index 0000000..44a045b
--- /dev/null
+++ b/src/plugins/MongodbPlugin.ts
@@ -0,0 +1,182 @@
+/*!
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+import SwPlugin from '../core/SwPlugin';
+import { createLogger } from '../logging';
+import ContextManager from '../trace/context/ContextManager';
+import Span from '../trace/span/Span';
+import { Component } from '../trace/Component';
+import { SpanLayer } from '../proto/language-agent/Tracing_pb';
+import { InsertWriteOpResult, MongoClient, MongoError, WithId } from 'mongodb';
+
+const logger = createLogger(__filename);
+
+class MongodbPlugin implements SwPlugin {
+ readonly module = 'mongodb';
+ readonly versions = '*'; // TODO: narrow down the supported version
+
+ install(): void {
+ const spanKeyedByRequestId: {
+ [requestId: string]: Span;
+ } = {};
+
+ const onStartEvent = (event: any) => {
+ if (logger.isDebugEnabled()) {
+ logger.debug('Mongodb start', event);
+ }
+ const span = ContextManager.current.newExitSpan(
+ `MongoDB/${event.databaseName}/${event.commandName}`,
+ event.address,
+ );
+ span.component = Component.MONGODB;
+ span.layer = SpanLayer.DATABASE;
+ spanKeyedByRequestId[event.requestId] = span.start();
+ };
+
+ const onSuccessEvent = async (event: any) => {
+ if (logger.isDebugEnabled()) {
+ logger.debug('Mongodb succeeded', event);
+ }
+
+ await (async () => setTimeout(() => {}, 1000))();
+
+ if (!spanKeyedByRequestId[event.requestId]) {
+ return;
+ }
+ const span = spanKeyedByRequestId[event.requestId];
+
+ delete spanKeyedByRequestId[event.requestId];
+
+ if (event.reply?.writeErrors) {
+ const errors = event.reply.writeErrors as [{ errmsg: string }];
+
+ span.errored = true;
+ span.logs.push({
+ items: [
+ {
+ key: 'ErrorMsg',
+ val: errors.map((it) => it.errmsg).join('\n'),
+ },
+ ],
+ timestamp: new Date().getTime(),
+ });
+ }
+
+ span.stop();
+ };
+
+ const onFailedEvent = (event: any) => {
+ if (logger.isDebugEnabled()) {
+ logger.debug('Mongodb failed', event);
+ }
+ if (!spanKeyedByRequestId[event.requestId]) {
+ return;
+ }
+ const span = spanKeyedByRequestId[event.requestId];
+
+ span.stop();
+
+ delete spanKeyedByRequestId[event.requestId];
+ };
+
+ const Client = require('mongodb/lib/mongo_client');
+
+ ((original) => {
+ Client.prototype.connect = function () {
+ const span = ContextManager.current.newExitSpan(`MongoDB/connect`, this.s.url).start();
+ span.component = Component.MONGODB;
+ span.layer = SpanLayer.DATABASE;
+
+ const snapshot = ContextManager.current.capture();
+
+ const callback = arguments[arguments.length - 1];
+ if (typeof callback === 'function') {
+ arguments[arguments.length - 1] = (error: MongoError, r: MongoClient) => {
+ ContextManager.current.restore(snapshot);
+
+ callback(error, r);
+ };
+ }
+
+ const result = original.apply(this, arguments);
+
+ span.stop();
+
+ return result;
+ };
+ })(Client.prototype.connect);
+
+ ((original) => {
+ Client.prototype.close = function () {
+ const span = ContextManager.current.newExitSpan(`MongoDB/close`, this.s.url).start();
+ span.component = Component.MONGODB;
+ span.layer = SpanLayer.DATABASE;
+
+ const snapshot = ContextManager.current.capture();
+
+ const callback = arguments[arguments.length - 1];
+ if (typeof callback === 'function') {
+ arguments[arguments.length - 1] = (error: MongoError, r: MongoClient) => {
+ ContextManager.current.restore(snapshot);
+
+ callback(error, r);
+ };
+ }
+
+ const result = original.apply(this, arguments);
+
+ span.stop();
+
+ return result;
+ };
+ })(Client.prototype.close);
+
+ const Collection = require('mongodb/lib/collection');
+
+ ((original) => {
+ Collection.prototype.insertMany = function (ns: string, ops: object[]) {
+ const { db, collection } = this.s.namespace;
+
+ const span = ContextManager.current.newExitSpan(`MongoDB/${db}/${collection}/insertMany`, this.s.url).start();
+ span.component = Component.MONGODB;
+ span.layer = SpanLayer.DATABASE;
+
+ const snapshot = ContextManager.current.capture();
+
+ const callback = arguments[arguments.length - 1];
+ if (typeof callback === 'function') {
+ arguments[arguments.length - 1] = (error: MongoError, r: InsertWriteOpResult<WithId<any>>) => {
+ ContextManager.current.restore(snapshot);
+
+ callback(error, r);
+ };
+ }
+
+ const result = original.apply(this, arguments);
+
+ span.stop();
+
+ return result;
+ };
+ })(Collection.prototype.insertMany);
+ }
+}
+
+// noinspection JSUnusedGlobalSymbols
+export default new MongodbPlugin();
diff --git a/src/trace/context/Context.ts b/src/trace/context/Context.ts
index 7393887..c17cdd9 100644
--- a/src/trace/context/Context.ts
+++ b/src/trace/context/Context.ts
@@ -36,6 +36,10 @@ export default interface Context {
stop(span: Span): boolean;
+ async(span: Span): Context;
+
+ await(span: Span): Context;
+
currentSpan(): Span | undefined;
capture(): Snapshot;
diff --git a/src/trace/context/ContextManager.ts b/src/trace/context/ContextManager.ts
index e49bbb0..de6dc34 100644
--- a/src/trace/context/ContextManager.ts
+++ b/src/trace/context/ContextManager.ts
@@ -25,19 +25,22 @@ class ContextManager {
contextKeyedByAsyncId: { [asyncId: number]: Context } = {};
constructor() {
- createHook({
- destroy: (asyncId: number) => {
- delete this.contextKeyedByAsyncId[asyncId];
- },
- }).enable();
+ // createHook({
+ // destroy: (asyncId: number) => {
+ // delete this.contextKeyedByAsyncId[asyncId];
+ // },
+ // }).enable();
}
- get current(): Context {
- const thisAsyncId = executionAsyncId();
-
- this.contextKeyedByAsyncId[thisAsyncId] = this.contextKeyedByAsyncId[thisAsyncId] || new SpanContext(thisAsyncId);
+ context = new SpanContext(1);
- return this.contextKeyedByAsyncId[thisAsyncId];
+ get current(): Context {
+ return this.context;
+ // const thisAsyncId = executionAsyncId();
+ //
+ // this.contextKeyedByAsyncId[thisAsyncId] = this.contextKeyedByAsyncId[thisAsyncId] || new SpanContext(thisAsyncId);
+ //
+ // return this.contextKeyedByAsyncId[thisAsyncId];
}
}
diff --git a/src/trace/context/DummyContext.ts b/src/trace/context/DummyContext.ts
index 54a98e8..52b7754 100644
--- a/src/trace/context/DummyContext.ts
+++ b/src/trace/context/DummyContext.ts
@@ -75,4 +75,12 @@ export default class DummyContext implements Context {
restore(snapshot: Snapshot) {
// Big Bang ~
}
+
+ async(span: Span): this {
+ return this;
+ }
+
+ await(span: Span): this {
+ return this;
+ }
}
diff --git a/src/trace/context/SpanContext.ts b/src/trace/context/SpanContext.ts
index d7c53b0..f7470f8 100644
--- a/src/trace/context/SpanContext.ts
+++ b/src/trace/context/SpanContext.ts
@@ -37,6 +37,7 @@ export default class SpanContext implements Context {
spanId = 0;
spans: Span[] = [];
segment: Segment = new Segment();
+ asyncSpans: Span[] = [];
constructor(public asyncId: number) {}
@@ -122,8 +123,12 @@ export default class SpanContext implements Context {
return true;
}
- if (this.tryFinish(span)) {
- this.spans.splice(0, 1);
+ const finished = this.tryFinish(span);
+ if (finished) {
+ this.spans.splice(this.spans.indexOf(span), 1);
+ }
+ if (finished && this.asyncSpans.length === 0) {
+ buffer.put(this.segment);
}
return this.spans.length === 0;
@@ -134,7 +139,6 @@ export default class SpanContext implements Context {
if (logger.isDebugEnabled()) {
logger.debug('Finishing span', { span });
}
- buffer.put(this.segment);
return true;
}
return false;
@@ -159,4 +163,21 @@ export default class SpanContext implements Context {
this.currentSpan()?.refer(ref);
this.segment.relate(ref.traceId);
}
+
+ async(span: Span): this {
+ if (!this.asyncSpans.includes(span)) {
+ this.asyncSpans.push(span);
+ }
+ return this;
+ }
+
+ await(span: Span): this {
+ if (this.asyncSpans.includes(span)) {
+ this.asyncSpans.splice(this.asyncSpans.indexOf(span), 1);
+ }
+ if (this.asyncSpans.length === 0) {
+ buffer.put(this.segment);
+ }
+ return this;
+ }
}
diff --git a/src/trace/span/Span.ts b/src/trace/span/Span.ts
index decfcf3..0a92868 100644
--- a/src/trace/span/Span.ts
+++ b/src/trace/span/Span.ts
@@ -58,6 +58,7 @@ export default abstract class Span {
startTime = 0;
endTime = 0;
errored = false;
+ _async = false;
constructor(options: SpanCtorOptions & { type: SpanType }) {
this.context = options.context;
@@ -141,4 +142,21 @@ export default abstract class Span {
}
return this;
}
+
+ get isAsync(): boolean {
+ return this._async;
+ }
+
+ async(): this {
+ this.context.async(this);
+ this._async = true;
+ return this;
+ }
+
+ await(): this {
+ this.endTime = new Date().getTime();
+ this.context.await(this);
+ this._async = false;
+ return this;
+ }
}
diff --git a/tests/dev.ts b/tests/dev.ts
new file mode 100644
index 0000000..0988a86
--- /dev/null
+++ b/tests/dev.ts
@@ -0,0 +1,59 @@
+/*!
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+import Agent from '../src/index';
+import { Db, MongoClient } from 'mongodb';
+
+import * as assert from 'assert';
+
+Agent.start({});
+// Connection URL
+const url = 'mongodb://localhost:27017';
+
+// Database Name
+const dbName = 'teambition';
+// Create a new MongoClient
+const client = new MongoClient(url, { useUnifiedTopology: true });
+const insertDocumentsOk = (db, callback) => {
+ const collection = db.collection('documents');
+ collection.insertMany([{ a: 1 }, { a: 2 }, { a: 3 }], (err, result) => {
+ callback(result);
+ });
+};
+const insertDocumentsFailed = (db: Db, callback) => {
+ const collection = db.collection('documents');
+ collection.insertMany([{ a: 1 }, { a: 2 }, { a: 3 }], (err, result) => {
+ callback(result);
+ });
+};
+// Use connect method to connect to the Server
+client.connect((err) => {
+ assert.equal(null, err);
+ const db = client.db(dbName);
+ insertDocumentsOk(db, () => {});
+ (async () => {
+ await (async () => setTimeout(() => {}, 1000))();
+ })();
+ insertDocumentsOk(db, () => {});
+ insertDocumentsFailed(db, () => {
+ client.close();
+ });
+});
+
+setTimeout(() => {}, 10000);
diff --git a/tests/plugins/http/client.ts b/tests/plugins/http/client.ts
new file mode 100644
index 0000000..ddd6785
--- /dev/null
+++ b/tests/plugins/http/client.ts
@@ -0,0 +1,57 @@
+/*!
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+// @ts-ignore
+import agent from '../../../src/index';
+
+agent.start({});
+
+import * as http from 'http';
+import { createServer, IncomingMessage, ServerResponse } from 'http';
+
+const providerHostPort = process.env.SW_PROVIDER || '127.0.0.1:5000';
+
+const port = 5001;
+
+const server = createServer((request: IncomingMessage, response: ServerResponse) => {
+ http
+ .request(
+ {
+ host: providerHostPort.split(':')[0],
+ port: providerHostPort.split(':')[1],
+ path: '/kids',
+ },
+ (res) => {
+ // http
+ // .request({
+ // host: providerHostPort.split(':')[0],
+ // port: providerHostPort.split(':')[1],
+ // path: '/inner-callback',
+ // })
+ // .end();
+ res.on('data', (chunk) => response.write(chunk));
+ res.on('end', () => response.end('\n'));
+ },
+ )
+ .end();
+});
+
+server.listen(port, () => {
+ console.log(`Server is listening on port ${port}`);
+});
diff --git a/src/trace/context/Context.ts b/tests/plugins/http/server.ts
similarity index 56%
copy from src/trace/context/Context.ts
copy to tests/plugins/http/server.ts
index 7393887..32b9292 100644
--- a/src/trace/context/Context.ts
+++ b/tests/plugins/http/server.ts
@@ -17,28 +17,19 @@
*
*/
-import Span from '../../trace/span/Span';
-import Segment from '../../trace/context/Segment';
-import Snapshot from '../../trace/context/Snapshot';
-import { ContextCarrier } from './ContextCarrier';
+// @ts-ignore
+import agent from '../../../src/index';
-export default interface Context {
- segment: Segment;
- spans: Span[];
+agent.start({});
- newLocalSpan(operation: string): Span;
+import { createServer, IncomingMessage, ServerResponse } from 'http';
- newEntrySpan(operation: string, carrier?: ContextCarrier): Span;
+const port = 5000;
- newExitSpan(operation: string, peer: string, carrier?: ContextCarrier): Span;
+const server = createServer((request: IncomingMessage, response: ServerResponse) => {
+ setTimeout(() => response.end('How you doin'), 1000);
+});
- start(span: Span): Context;
-
- stop(span: Span): boolean;
-
- currentSpan(): Span | undefined;
-
- capture(): Snapshot;
-
- restore(snapshot: Snapshot): void;
-}
+server.listen(port, () => {
+ console.log(`Server is listening on port ${port}`);
+});