Introduce service naming framework (#2941)

* Add interaction with the `DD_TRACE_SPAN_ATTRIBUTE_SCHEMA` environment variable
* Provide the building blocks for 2 versions of an service and operation naming schema:
    * `v0` is the current version, i.e. a representation of the existing state of the plugin. One could checkout the commit with the introduction of this version, and have unchanged tests still pass.
    * `v1` is the new naming schema, made consistent across all tracers.
* Introduce a testing harness for easy tests of `v0` _and_ `v1` naming
* Apply this versioned naming for all messaging integrations - `rhea`, `amqp(10|lib)`, `kafkajs`, `google-cloud-pubsub`

packages/.eslintrc.json

@@ -9,6 +9,7 @@
     "expect": true,
     "sinon": true,
     "proxyquire": true,
+    "withNamingSchema": true,
     "withVersions": true,
     "withExports": true
   },

packages/datadog-plugin-amqp10/src/consumer.js

@@ -11,11 +11,9 @@ class Amqp10ConsumerPlugin extends ConsumerPlugin {
     const source = getShortName(link)
     const address = getAddress(link)
 
-    this.startSpan('amqp.receive', {
-      service: this.config.service || `${this.tracer._service}-amqp`,
+    this.startSpan({
       resource: ['receive', source].filter(v => v).join(' '),
       type: 'worker',
-      kind: 'consumer',
       meta: {
         'amqp.link.source.address': source,
         'amqp.link.role': 'receiver',

packages/datadog-plugin-amqp10/src/producer.js

@@ -13,10 +13,8 @@ class Amqp10ProducerPlugin extends ProducerPlugin {
     const address = getAddress(link)
     const target = getShortName(link)
 
-    this.startSpan('amqp.send', {
-      service: this.config.service || `${this.tracer._service}-amqp`,
+    this.startSpan({
       resource: ['send', target].filter(v => v).join(' '),
-      kind: 'producer',
       meta: {
         'amqp.link.target.address': target,
         'amqp.link.role': 'sender',

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

@@ -3,6 +3,8 @@
 const agent = require('../../dd-trace/test/plugins/agent')
 const { ERROR_MESSAGE, ERROR_STACK, ERROR_TYPE } = require('../../dd-trace/src/constants')
 
+const namingSchema = require('./naming')
+
 describe('Plugin', () => {
   let tracer
   let client
@@ -63,8 +65,8 @@ describe('Plugin', () => {
               .use(traces => {
                 const span = traces[0][0]
 
-                expect(span).to.have.property('name', 'amqp.send')
-                expect(span).to.have.property('service', 'test-amqp')
+                expect(span).to.have.property('name', namingSchema.send.opName)
+                expect(span).to.have.property('service', namingSchema.send.serviceName)
                 expect(span).to.have.property('resource', 'send amq.topic')
                 expect(span).to.not.have.property('type')
                 expect(span.meta).to.have.property('span.kind', 'producer')
@@ -84,7 +86,6 @@ describe('Plugin', () => {
 
             sender.send({ key: 'value' })
           })
-
           it('should handle errors', done => {
             let error
 
@@ -123,15 +124,21 @@ describe('Plugin', () => {
               expect(promise).to.have.property('value')
             })
           })
+
+          withNamingSchema(
+            () => sender.send({ key: 'value' }),
+            () => namingSchema.send.opName,
+            () => namingSchema.send.serviceName
+          )
         })
 
         describe('when consuming messages', () => {
           it('should do automatic instrumentation', done => {
             agent
               .use(traces => {
                 const span = traces[0][0]
-                expect(span).to.have.property('name', 'amqp.receive')
-                expect(span).to.have.property('service', 'test-amqp')
+                expect(span).to.have.property('name', namingSchema.receive.opName)
+                expect(span).to.have.property('service', namingSchema.receive.serviceName)
                 expect(span).to.have.property('resource', 'receive amq.topic')
                 expect(span).to.have.property('type', 'worker')
                 expect(span.meta).to.have.property('span.kind', 'consumer')
@@ -163,12 +170,18 @@ describe('Plugin', () => {
 
             sender.send({ key: 'value' })
           })
+
+          withNamingSchema(
+            () => sender.send({ key: 'value' }),
+            () => namingSchema.receive.opName,
+            () => namingSchema.receive.serviceName
+          )
         })
       })
 
       describe('with configuration', () => {
         beforeEach(() => {
-          agent.reload('amqp10', { service: 'test' })
+          agent.reload('amqp10', { service: 'test-custom-name' })
 
           const amqp = require(`../../../versions/amqp10@${version}`).get()
 
@@ -192,13 +205,19 @@ describe('Plugin', () => {
             .use(traces => {
               const span = traces[0][0]
 
-              expect(span).to.have.property('service', 'test')
+              expect(span).to.have.property('service', 'test-custom-name')
             }, 2)
             .then(done)
             .catch(done)
 
           sender.send({ key: 'value' })
         })
+
+        withNamingSchema(
+          () => sender.send({ key: 'value' }),
+          () => namingSchema.receive.opName,
+          () => 'test-custom-name'
+        )
       })
     })
   })

packages/datadog-plugin-amqp10/test/naming.js

@@ -0,0 +1,24 @@
+const { resolveNaming } = require('../../dd-trace/test/plugins/helpers')
+
+module.exports = resolveNaming({
+  send: {
+    v0: {
+      opName: 'amqp.send',
+      serviceName: 'test-amqp'
+    },
+    v1: {
+      opName: 'amqp.send',
+      serviceName: 'test'
+    }
+  },
+  receive: {
+    v0: {
+      opName: 'amqp.receive',
+      serviceName: 'test-amqp'
+    },
+    v1: {
+      opName: 'amqp.process',
+      serviceName: 'test'
+    }
+  }
+})

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

@@ -7,17 +7,18 @@ const { getResourceName } = require('./util')
 
 class AmqplibClientPlugin extends ClientPlugin {
   static get id () { return 'amqplib' }
+  static get type () { return 'messaging' }
   static get operation () { return 'command' }
 
   start ({ channel = {}, method, fields }) {
     if (method === 'basic.deliver' || method === 'basic.get') return
     if (method === 'basic.publish') return
 
     const stream = (channel.connection && channel.connection.stream) || {}
-    const span = this.startSpan('amqp.command', {
-      service: this.config.service || `${this.tracer._service}-amqp`,
+    const span = this.startSpan(this.operationName(), {
+      service: this.config.service || this.serviceName(),
       resource: getResourceName(method, fields),
-      kind: 'client',
+      kind: this.constructor.kind,
       meta: {
         'out.host': stream._host,
         [CLIENT_PORT_KEY]: stream.remotePort,

packages/datadog-plugin-amqplib/src/consumer.js

@@ -13,11 +13,9 @@ class AmqplibConsumerPlugin extends ConsumerPlugin {
 
     const childOf = extract(this.tracer, message)
 
-    this.startSpan('amqp.command', {
+    this.startSpan({
       childOf,
-      service: this.config.service || `${this.tracer._service}-amqp`,
       resource: getResourceName(method, fields),
-      kind: 'consumer',
       type: 'worker',
       meta: {
         'amqp.queue': fields.queue,

packages/datadog-plugin-amqplib/src/producer.js

@@ -13,10 +13,8 @@ class AmqplibProducerPlugin extends ProducerPlugin {
     if (method !== 'basic.publish') return
 
     const stream = (channel.connection && channel.connection.stream) || {}
-    const span = this.startSpan('amqp.command', {
-      service: this.config.service || `${this.tracer._service}-amqp`,
+    const span = this.startSpan({
       resource: getResourceName(method, fields),
-      kind: 'producer',
       meta: {
         'out.host': stream._host,
         [CLIENT_PORT_KEY]: stream.remotePort,

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

@@ -3,6 +3,8 @@
 const agent = require('../../dd-trace/test/plugins/agent')
 const { ERROR_MESSAGE, ERROR_STACK, ERROR_TYPE } = require('../../dd-trace/src/constants')
 
+const namingSchema = require('./naming')
+
 describe('Plugin', () => {
   let tracer
   let connection
@@ -55,8 +57,8 @@ describe('Plugin', () => {
               agent
                 .use(traces => {
                   const span = traces[0][0]
-                  expect(span).to.have.property('name', 'amqp.command')
-                  expect(span).to.have.property('service', 'test-amqp')
+                  expect(span).to.have.property('name', namingSchema.controlPlane.opName)
+                  expect(span).to.have.property('service', namingSchema.controlPlane.serviceName)
                   expect(span).to.have.property('resource', 'queue.declare test')
                   expect(span).to.not.have.property('type')
                   expect(span.meta).to.have.property('span.kind', 'client')
@@ -75,8 +77,8 @@ describe('Plugin', () => {
                 .use(traces => {
                   const span = traces[0][0]
 
-                  expect(span).to.have.property('name', 'amqp.command')
-                  expect(span).to.have.property('service', 'test-amqp')
+                  expect(span).to.have.property('name', namingSchema.controlPlane.opName)
+                  expect(span).to.have.property('service', namingSchema.controlPlane.serviceName)
                   expect(span).to.have.property('resource', 'queue.delete test')
                   expect(span).to.not.have.property('type')
                   expect(span.meta).to.have.property('span.kind', 'client')
@@ -113,6 +115,12 @@ describe('Plugin', () => {
                 error = e
               }
             })
+
+            withNamingSchema(
+              () => channel.assertQueue('test', {}, () => {}),
+              () => namingSchema.controlPlane.opName,
+              () => namingSchema.controlPlane.serviceName
+            )
           })
 
           describe('when publishing messages', () => {
@@ -121,8 +129,8 @@ describe('Plugin', () => {
                 .use(traces => {
                   const span = traces[0][0]
 
-                  expect(span).to.have.property('name', 'amqp.command')
-                  expect(span).to.have.property('service', 'test-amqp')
+                  expect(span).to.have.property('name', namingSchema.send.opName)
+                  expect(span).to.have.property('service', namingSchema.send.serviceName)
                   expect(span).to.have.property('resource', 'basic.publish exchange routingKey')
                   expect(span).to.not.have.property('type')
                   expect(span.meta).to.have.property('out.host', 'localhost')
@@ -160,6 +168,15 @@ describe('Plugin', () => {
                 error = e
               }
             })
+
+            withNamingSchema(
+              () => {
+                channel.assertExchange('exchange', 'direct', {}, () => {})
+                channel.publish('exchange', 'routingKey', Buffer.from('content'))
+              },
+              () => namingSchema.send.opName,
+              () => namingSchema.send.serviceName
+            )
           })
 
           describe('when consuming messages', () => {
@@ -170,8 +187,8 @@ describe('Plugin', () => {
               agent
                 .use(traces => {
                   const span = traces[0][0]
-                  expect(span).to.have.property('name', 'amqp.command')
-                  expect(span).to.have.property('service', 'test-amqp')
+                  expect(span).to.have.property('name', namingSchema.receive.opName)
+                  expect(span).to.have.property('service', namingSchema.receive.serviceName)
                   expect(span).to.have.property('resource', `basic.deliver ${queue}`)
                   expect(span).to.have.property('type', 'worker')
                   expect(span.meta).to.have.property('span.kind', 'consumer')
@@ -236,6 +253,18 @@ describe('Plugin', () => {
                 })
               })
             })
+
+            withNamingSchema(
+              () => {
+                channel.assertQueue('', {}, (err, ok) => {
+                  if (err) return
+                  channel.sendToQueue(ok.queue, Buffer.from('content'))
+                  channel.consume(ok.queue, () => {}, {}, (err, ok) => {})
+                })
+              },
+              () => namingSchema.receive.opName,
+              () => namingSchema.receive.serviceName
+            )
           })
         })
 
@@ -260,7 +289,7 @@ describe('Plugin', () => {
 
       describe('with configuration', () => {
         before(() => {
-          return agent.load('amqplib', { service: 'test' })
+          return agent.load('amqplib', { service: 'test-custom-service' })
         })
 
         after(() => {
@@ -286,14 +315,20 @@ describe('Plugin', () => {
         it('should be configured with the correct values', done => {
           agent
             .use(traces => {
-              expect(traces[0][0]).to.have.property('service', 'test')
+              expect(traces[0][0]).to.have.property('service', 'test-custom-service')
               expect(traces[0][0]).to.have.property('resource', 'queue.declare test')
             }, 2)
             .then(done)
             .catch(done)
 
           channel.assertQueue('test', {}, () => {})
         })
+
+        withNamingSchema(
+          () => channel.assertQueue('test', {}, () => {}),
+          () => namingSchema.controlPlane.opName,
+          () => 'test-custom-service'
+        )
       })
     })
   })

packages/datadog-plugin-amqplib/test/naming.js

@@ -0,0 +1,34 @@
+const { resolveNaming } = require('../../dd-trace/test/plugins/helpers')
+
+module.exports = resolveNaming({
+  send: {
+    v0: {
+      opName: 'amqp.command',
+      serviceName: 'test-amqp'
+    },
+    v1: {
+      opName: 'amqp.send',
+      serviceName: 'test'
+    }
+  },
+  receive: {
+    v0: {
+      opName: 'amqp.command',
+      serviceName: 'test-amqp'
+    },
+    v1: {
+      opName: 'amqp.process',
+      serviceName: 'test'
+    }
+  },
+  controlPlane: {
+    v0: {
+      opName: 'amqp.command',
+      serviceName: 'test-amqp'
+    },
+    v1: {
+      opName: 'amqp.command',
+      serviceName: 'test'
+    }
+  }
+})

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

@@ -4,15 +4,16 @@ const ClientPlugin = require('../../dd-trace/src/plugins/client')
 
 class GoogleCloudPubsubClientPlugin extends ClientPlugin {
   static get id () { return 'google-cloud-pubsub' }
+  static get type () { return 'messaging' }
   static get operation () { return 'request' }
 
   start ({ request, api, projectId }) {
     if (api === 'publish') return
 
-    this.startSpan('pubsub.request', {
-      service: this.config.service || `${this.tracer._service}-pubsub`,
+    this.startSpan(this.operationName(), {
+      service: this.config.service || this.serviceName(),
       resource: [api, request.name].filter(x => x).join(' '),
-      kind: 'client',
+      kind: this.constructor.kind,
       meta: {
         'pubsub.method': api,
         'gcloud.project_id': projectId

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

@@ -11,11 +11,9 @@ class GoogleCloudPubsubConsumerPlugin extends ConsumerPlugin {
     const topic = subscription.metadata && subscription.metadata.topic
     const childOf = this.tracer.extract('text_map', message.attributes) || null
 
-    this.startSpan('pubsub.receive', {
+    this.startSpan({
       childOf,
-      service: this.config.service,
       resource: topic,
-      kind: 'consumer',
       type: 'worker',
       meta: {
         'gcloud.project_id': subscription.pubsub.projectId,