Peer service computation (#3177)

* introduce peer service computation for outbound edges:
    * add DD_TRACE_SPAN_PEER_SERVICE environment variable
    * add peer service computation logic
    * TODO: add tests for config, outbound, tracer (?)
* provide a helper for testing peer service computation
* defer plugin `finish` to TracingPlugin in children

packages/.eslintrc.json

@@ -11,7 +11,8 @@
     "proxyquire": true,
     "withNamingSchema": true,
     "withVersions": true,
-    "withExports": true
+    "withExports": true,
+    "withPeerService": true
   },
   "rules": {
     "no-unused-expressions": 0,

packages/datadog-plugin-amqp10/test/index.spec.js

@@ -60,6 +60,13 @@ describe('Plugin', () => {
         })
 
         describe('when sending messages', () => {
+          withPeerService(
+            () => tracer,
+            () => sender.send({ key: 'value' }),
+            'localhost',
+            'out.host'
+          )
+
           it('should do automatic instrumentation', done => {
             agent
               .use(traces => {

packages/datadog-plugin-amqplib/test/index.spec.js

@@ -53,6 +53,13 @@ describe('Plugin', () => {
           })
 
           describe('when sending commands', () => {
+            withPeerService(
+              () => tracer,
+              () => channel.assertQueue('test', {}, () => {}),
+              'localhost',
+              'out.host'
+            )
+
             it('should do automatic instrumentation for immediate commands', done => {
               agent
                 .use(traces => {
@@ -124,6 +131,13 @@ describe('Plugin', () => {
           })
 
           describe('when publishing messages', () => {
+            withPeerService(
+              () => tracer,
+              () => channel.assertQueue('test', {}, () => {}),
+              'localhost',
+              'out.host'
+            )
+
             it('should do automatic instrumentation', done => {
               agent
                 .use(traces => {

packages/datadog-plugin-dns/src/lookup.js

@@ -33,7 +33,7 @@ class DNSLookupPlugin extends ClientPlugin {
       span.setTag('dns.address', result)
     }
 
-    span.finish()
+    super.finish()
   }
 }
 

packages/datadog-plugin-google-cloud-pubsub/src/consumer.js

@@ -32,7 +32,7 @@ class GoogleCloudPubsubConsumerPlugin extends ConsumerPlugin {
       span.setTag('pubsub.ack', 1)
     }
 
-    span.finish()
+    super.finish()
   }
 }
 

packages/datadog-plugin-graphql/src/execute.js

@@ -32,7 +32,7 @@ class GraphQLExecutePlugin extends TracingPlugin {
   finish ({ res, args }) {
     const span = this.activeSpan
     this.config.hooks.execute(span, args, res)
-    span.finish()
+    super.finish()
   }
 }
 

packages/datadog-plugin-graphql/src/parse.js

@@ -25,7 +25,7 @@ class GraphQLParsePlugin extends TracingPlugin {
 
     this.config.hooks.parse(span, source, document)
 
-    span.finish()
+    super.finish()
   }
 }
 

packages/datadog-plugin-graphql/src/resolve.js

@@ -56,11 +56,6 @@ class GraphQLResolvePlugin extends TracingPlugin {
     }
   }
 
-  finish (finishTime) {
-    const span = this.activeSpan
-    span.finish(finishTime)
-  }
-
   constructor (...args) {
     super(...args)
 

packages/datadog-plugin-graphql/src/validate.js

@@ -21,7 +21,7 @@ class GraphQLValidatePlugin extends TracingPlugin {
   finish ({ document, errors }) {
     const span = this.activeSpan
     this.config.hooks.validate(span, document, errors)
-    span.finish()
+    super.finish()
   }
 }
 

packages/datadog-plugin-grpc/src/client.js

@@ -57,7 +57,7 @@ class GrpcClientPlugin extends ClientPlugin {
       addMetadataTags(span, metadata, metadataFilter, 'response')
     }
 
-    span.finish()
+    super.finish()
   }
 
   configure (config) {

packages/datadog-plugin-grpc/src/server.js

@@ -68,7 +68,7 @@ class GrpcServerPlugin extends ServerPlugin {
       addMetadataTags(span, trailer, metadataFilter, 'response')
     }
 
-    span.finish()
+    super.finish()
   }
 
   configure (config) {

packages/datadog-plugin-http/src/client.js

@@ -87,7 +87,7 @@ class HttpClientPlugin extends ClientPlugin {
     addRequestHeaders(req, span, this.config)
 
     this.config.hooks.request(span, req, res)
-    span.finish()
+    super.finish()
   }
 
   error (err) {

packages/datadog-plugin-http/test/client.spec.js

@@ -58,6 +58,26 @@ describe('Plugin', () => {
             })
         })
 
+        withPeerService(
+          () => tracer,
+          () => {
+            const app = express()
+            app.get('/user', (req, res) => {
+              res.status(200).send()
+            })
+            getPort().then(port => {
+              appListener = server(app, port, () => {
+                const req = http.request(`${protocol}://localhost:${port}/user`, res => {
+                  res.on('data', () => {})
+                })
+                req.end()
+              })
+            })
+          },
+          'localhost',
+          'out.host'
+        )
+
         it('should do automatic instrumentation', done => {
           const app = express()
           app.get('/user', (req, res) => {

packages/datadog-plugin-http2/src/client.js

@@ -93,11 +93,6 @@ class Http2ClientPlugin extends ClientPlugin {
     this.enter(span, store)
   }
 
-  finish () {
-    const span = storage.getStore().span
-    span.finish()
-  }
-
   configure (config) {
     return super.configure(normalizeConfig(config))
   }

packages/datadog-plugin-http2/test/client.spec.js

@@ -54,6 +54,31 @@ describe('Plugin', () => {
             })
         })
 
+        withPeerService(
+          () => tracer,
+          done => {
+            getPort().then(port => {
+              const app = (stream, headers) => {
+                stream.respond({
+                  ':status': 200
+                })
+                stream.end()
+              }
+              appListener = server(app, port, () => {
+                const client = http2
+                  .connect(`${protocol}://localhost:${port}`)
+                  .on('error', done)
+
+                const req = client.request({ ':path': '/user', ':method': 'GET' })
+                req.on('error', done)
+
+                req.end()
+              })
+            })
+          },
+          'localhost', 'out.host'
+        )
+
         it('should do automatic instrumentation', done => {
           const app = (stream, headers) => {
             stream.respond({

packages/datadog-plugin-memcached/test/index.spec.js

@@ -25,6 +25,12 @@ describe('Plugin', () => {
           Memcached = proxyquire(`../../../versions/memcached@${version}/node_modules/memcached`, {})
         })
 
+        withPeerService(
+          () => tracer,
+          done => memcached.get('test', err => err && done(err)),
+          'localhost',
+          'out.host'
+        )
         it('should do automatic instrumentation when using callbacks', done => {
           memcached = new Memcached('localhost:11211', { retries: 0 })
 

packages/datadog-plugin-moleculer/src/client.js

@@ -29,7 +29,7 @@ class MoleculerClientPlugin extends ClientPlugin {
       span.addTags(moleculerTags(broker, ctx, this.config))
     }
 
-    span.finish()
+    super.finish()
   }
 }
 

packages/datadog-plugin-moleculer/test/index.spec.js

@@ -108,10 +108,28 @@ describe('Plugin', () => {
 
       describe('client', () => {
         describe('without configuration', () => {
-          before(() => agent.load('moleculer', { server: false }))
-          before(() => startBroker())
-          after(() => broker.stop())
-          after(() => agent.close({ ritmReset: false }))
+          const hostname = os.hostname()
+          let tracer
+
+          beforeEach(() => startBroker())
+          afterEach(() => broker.stop())
+
+          beforeEach(done => {
+            agent.load('moleculer', { server: false })
+              .then(() => { tracer = require('../../dd-trace') })
+              .then(done)
+              .catch(done)
+          })
+          afterEach(() => agent.close({ ritmReset: false }))
+
+          withPeerService(
+            () => tracer,
+            done => {
+              broker.call('math.add', { a: 5, b: 3 }).catch(done)
+            },
+            hostname,
+            'out.host'
+          )
 
           it('should do automatic instrumentation', done => {
             agent.use(traces => {
@@ -121,7 +139,7 @@ describe('Plugin', () => {
               expect(spans[0]).to.have.property('service', 'test')
               expect(spans[0]).to.have.property('resource', 'math.add')
               expect(spans[0].meta).to.have.property('span.kind', 'client')
-              expect(spans[0].meta).to.have.property('out.host', os.hostname())
+              expect(spans[0].meta).to.have.property('out.host', hostname)
               expect(spans[0].meta).to.have.property('moleculer.context.action', 'math.add')
               expect(spans[0].meta).to.have.property('moleculer.context.node_id', `server-${process.pid}`)
               expect(spans[0].meta).to.have.property('moleculer.context.request_id')

packages/datadog-plugin-net/test/index.spec.js

@@ -75,6 +75,16 @@ describe('Plugin', () => {
       })
     })
 
+    withPeerService(
+      () => tracer,
+      () => {
+        const socket = new net.Socket()
+        socket.connect(port, 'localhost')
+      },
+      'localhost',
+      'out.host'
+    )
+
     it('should instrument connect with a port', done => {
       const socket = new net.Socket()
       tracer.scope().activate(parent, () => {

packages/datadog-plugin-redis/test/legacy.spec.js

@@ -40,6 +40,13 @@ describe('Legacy Plugin', () => {
           sub = redis.createClient()
         })
 
+        withPeerService(
+          () => tracer,
+          () => client.get('foo'),
+          '127.0.0.1',
+          'out.host'
+        )
+
         it('should do automatic instrumentation when using callbacks', done => {
           client.on('error', done)
           agent

packages/datadog-plugin-rhea/test/index.spec.js

@@ -47,6 +47,13 @@ describe('Plugin', () => {
           })
 
           describe('sending a message', () => {
+            withPeerService(
+              () => tracer,
+              () => context.sender.send({ body: 'Hello World!' }),
+              'localhost',
+              'out.host'
+            )
+
             it('should automatically instrument', (done) => {
               agent.use(traces => {
                 const span = traces[0][0]

packages/datadog-plugin-sharedb/src/index.js

@@ -25,7 +25,7 @@ class SharedbPlugin extends ServerPlugin {
     if (this.config.hooks && this.config.hooks.reply) {
       this.config.hooks.reply(span, request, res)
     }
-    span.finish()
+    super.finish()
   }
 }
 

packages/dd-trace/src/config.js

@@ -306,6 +306,8 @@ class Config {
     const DD_TRACE_SPAN_ATTRIBUTE_SCHEMA = validateNamingVersion(
       process.env.DD_TRACE_SPAN_ATTRIBUTE_SCHEMA
     )
+    const DD_TRACE_PEER_SERVICE_DEFAULTS_ENABLED = process.env.DD_TRACE_PEER_SERVICE_DEFAULTS_ENABLED
+
     const DD_TRACE_X_DATADOG_TAGS_MAX_LENGTH = coalesce(
       process.env.DD_TRACE_X_DATADOG_TAGS_MAX_LENGTH,
       '512'
@@ -524,6 +526,10 @@ ken|consumer_?(?:id|key|secret)|sign(?:ed|ature)?|auth(?:entication|orization)?)
       exporters: DD_PROFILING_EXPORTERS
     }
     this.spanAttributeSchema = DD_TRACE_SPAN_ATTRIBUTE_SCHEMA
+    this.spanComputePeerService = (this.spanAttributeSchema === 'v0'
+      ? isTrue(DD_TRACE_PEER_SERVICE_DEFAULTS_ENABLED)
+      : true
+    )
     this.lookup = options.lookup
     this.startupLogs = isTrue(DD_TRACE_STARTUP_LOGS)
     // Disabled for CI Visibility's agentless

packages/dd-trace/src/constants.js

@@ -26,6 +26,8 @@ module.exports = {
   ERROR_STACK: 'error.stack',
   COMPONENT: 'component',
   CLIENT_PORT_KEY: 'network.destination.port',
+  PEER_SERVICE_KEY: 'peer.service',
+  PEER_SERVICE_SOURCE_KEY: '_dd.peer.service.source',
   SCI_REPOSITORY_URL: '_dd.git.repository_url',
   SCI_COMMIT_SHA: '_dd.git.commit.sha'
 }

packages/dd-trace/src/opentracing/tracer.js

@@ -26,6 +26,7 @@ class DatadogTracer {
     this._version = config.version
     this._env = config.env
     this._tags = config.tags
+    this._computePeerService = config.spanComputePeerService
     this._logInjection = config.logInjection
     this._debug = config.debug
     this._prioritySampler = new PrioritySampler(config.env, config.sampler)