feat: Added instrumentation for ElasticSearch (#1785)

bin/docker-env-vars.sh

@@ -6,6 +6,7 @@
 IP=`docker-machine ip default 2>/dev/null`
 
 export NR_NODE_TEST_CASSANDRA_HOST=$IP
+export NR_NODE_TEST_ELASTIC_HOST=$IP
 export NR_NODE_TEST_MEMCACHED_HOST=$IP
 export NR_NODE_TEST_MONGODB_HOST=$IP
 export NR_NODE_TEST_MYSQL_HOST=$IP

bin/docker-services.sh

@@ -42,6 +42,16 @@ else
   docker run -d --name nr_node_cassandra -p 9042:9042 zmarcantel/cassandra;
 fi
 
+if docker ps -a | grep -q "nr_node_elastic"; then
+  docker start nr_node_elastic;
+else
+  docker run -d --name nr_node_elastic \
+  -p 9200:9200 \
+  -e "discovery.type=single-node" \
+  -e "xpack.security.enabled=false" \
+  docker.elastic.co/elasticsearch/elasticsearch:8.8.2;
+fi
+
 if docker ps -a | grep -q "nr_node_postgres"; then
   docker start nr_node_postgres;
 else

bin/run-versioned-tests.sh

@@ -56,7 +56,7 @@ echo "NPM7 = ${NPM7}"
 echo "CONTEXT MANAGER = ${CTX_MGR}"
 echo "C8 = ${C8}"
 
-# if $JOBS is not empy
+# if $JOBS is not empty
 if [ ! -z "$JOBS" ];
 then
   JOBS_ARGS="--jobs $JOBS"

lib/instrumentation/@elastic/elasticsearch.js

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2023 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+const logger = require('../../logger').child({ component: 'ElasticSearch' })
+
+/**
+ * Instruments the `@elastic/elasticsearch` module. This function is
+ * passed to `onRequire` when instantiating instrumentation.
+ *
+ * @param {object} _agent New Relic agent
+ * @param {object} elastic resolved module
+ * @param {string} _moduleName string representation of require/import path
+ * @param {object} shim New Relic shim
+ * @returns {void}
+ */
+module.exports = function initialize(_agent, elastic, _moduleName, shim) {
+  shim.setDatastore(shim.ELASTICSEARCH)
+  shim.setParser(queryParser)
+
+  shim.recordQuery(elastic.Transport.prototype, 'request', function wrapQuery(shim, _, __, args) {
+    const ctx = this
+    return {
+      query: JSON.stringify(args?.[0]),
+      promise: true,
+      opaque: true,
+      inContext: function inContext() {
+        getConnection.call(ctx, shim)
+      }
+    }
+  })
+}
+
+/**
+ * Parses the parameters sent to elasticsearch for collection,
+ * method, and query
+ *
+ * @param {object} params Query object received by the datashim.
+ * Required properties: path {string}, method {string}.
+ * Optional properties: querystring {string}, body {object}, and
+ * bulkBody {object}
+ * @returns {object} consisting of collection {string}, operation {string},
+ * and query {string}
+ */
+function queryParser(params) {
+  params = JSON.parse(params)
+
+  const { collection, operation } = parsePath(params.path, params.method)
+
+  // the substance of the query may be in querystring or in body.
+  let queryParam = {}
+  if (typeof params.querystring === 'object' && Object.keys(params.querystring).length > 0) {
+    queryParam = params.querystring
+  }
+  // let body or bulkBody override querystring, as some requests have both
+  if (typeof params.body === 'object' && Object.keys(params.body).length > 0) {
+    queryParam = params.body
+  } else if (typeof params.bulkBody === 'object' && Object.keys(params.bulkBody).length > 0) {
+    queryParam = params.bulkBody
+  }
+
+  const query = JSON.stringify(queryParam)
+
+  return {
+    collection,
+    operation,
+    query
+  }
+}
+
+/**
+ * Convenience function for parsing the params.path sent to the queryParser
+ * for normalized collection and operation
+ *
+ * @param {string} pathString params.path supplied to the query parser
+ * @param {string} method http method called by @elastic/elasticsearch
+ * @returns {object} consisting of collection {string} and operation {string}
+ */
+function parsePath(pathString, method) {
+  let collection
+  let operation
+  const defaultCollection = 'any'
+  const actions = {
+    GET: 'get',
+    PUT: 'create',
+    POST: 'create',
+    DELETE: 'delete',
+    HEAD: 'exists'
+  }
+  const suffix = actions[method]
+
+  try {
+    const path = pathString.split('/')
+    if (method === 'PUT' && path.length === 2) {
+      collection = path?.[1] || defaultCollection
+      operation = `index.create`
+      return { collection, operation }
+    }
+    path.forEach((segment, idx) => {
+      const prev = idx - 1
+      let opname
+      if (segment === '_search') {
+        collection = path?.[prev] || defaultCollection
+        operation = `search`
+      } else if (segment[0] === '_') {
+        opname = segment.substring(1)
+        collection = path?.[prev] || defaultCollection
+        operation = `${opname}.${suffix}`
+      }
+    })
+    if (!operation && !collection) {
+      // likely creating an index--no underscore segments
+      collection = path?.[1] || defaultCollection
+      operation = `index.${suffix}`
+    }
+  } catch (e) {
+    logger.warn('Failed to parse path for operation and collection. Using defaults')
+    logger.warn(e)
+    collection = defaultCollection
+    operation = 'unknown'
+  }
+
+  return { collection, operation }
+}
+
+/**
+ * Convenience function for deriving connection information from
+ * elasticsearch
+ *
+ * @param {object} shim The New Relic datastore-shim
+ * @returns {Function} captureInstanceAttributes method of shim
+ */
+function getConnection(shim) {
+  const connectionPool = this.connectionPool.connections[0]
+  const host = connectionPool.url.host.split(':')
+  const port = connectionPool.url.port || host?.[1]
+  return shim.captureInstanceAttributes(host[0], port)
+}
+
+module.exports.queryParser = queryParser
+module.exports.parsePath = parsePath
+module.exports.getConnection = getConnection

lib/instrumentations.js

@@ -17,6 +17,7 @@ module.exports = function instrumentations() {
     'bluebird': { type: MODULE_TYPE.PROMISE },
     'bunyan': { type: MODULE_TYPE.GENERIC },
     'director': { type: MODULE_TYPE.WEB_FRAMEWORK },
+    '@elastic/elasticsearch': { type: MODULE_TYPE.DATASTORE },
     'express': { type: MODULE_TYPE.WEB_FRAMEWORK },
     'fastify': { type: MODULE_TYPE.WEB_FRAMEWORK },
     'generic-pool': { type: MODULE_TYPE.GENERIC },

lib/shim/datastore-shim.js

@@ -29,6 +29,7 @@ const util = require('util')
 const DATASTORE_NAMES = {
   CASSANDRA: 'Cassandra',
   DYNAMODB: 'DynamoDB',
+  ELASTICSEARCH: 'ElasticSearch',
   MEMCACHED: 'Memcache',
   MONGODB: 'MongoDB',
   MYSQL: 'MySQL',

test/lib/params.js

@@ -26,6 +26,11 @@ module.exports = {
   cassandra_host: process.env.NR_NODE_TEST_CASSANDRA_HOST || 'localhost',
   cassandra_port: process.env.NR_NODE_TEST_CASSANDRA_PORT || 9042,
 
+  elastic_host: process.env.NR_NODE_TEST_ELASTIC_HOST || 'localhost',
+  elastic_port: process.env.NR_NODE_TEST_ELASTIC_PORT || 9200,
+  elastic_user: process.env.NR_NODE_TEST_ELASTIC_USER || 'elastic',
+  elastic_pass: process.env.NR_NODE_TEST_ELASTIC_PASS || 'changeme',
+
   postgres_host: process.env.NR_NODE_TEST_POSTGRES_HOST || 'localhost',
   postgres_port: process.env.NR_NODE_TEST_POSTGRES_PORT || 5432,
   postgres_prisma_port: process.env.NR_NODE_TEST_POSTGRES_PRISMA_PORT || 5434,

test/unit/instrumentation/elasticsearch.test.js

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2023 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+const tap = require('tap')
+const { parsePath, queryParser } = require('../../../lib/instrumentation/@elastic/elasticsearch')
+const methods = [
+  { name: 'GET', expected: 'get' },
+  { name: 'PUT', expected: 'create' },
+  { name: 'POST', expected: 'create' },
+  { name: 'DELETE', expected: 'delete' },
+  { name: 'HEAD', expected: 'exists' }
+]
+
+tap.test('parsePath should behave as expected', (t) => {
+  t.autoend()
+
+  t.test('indices', function (t) {
+    const path = '/indexName'
+    methods.forEach((m) => {
+      const { collection, operation } = parsePath(path, m.name)
+      const expectedOp = `index.${m.expected}`
+      t.equal(collection, 'indexName', `index should be 'indexName'`)
+      t.equal(operation, expectedOp, 'operation should include index and method')
+    })
+    t.end()
+  })
+  t.test('search of one index', function (t) {
+    const path = '/indexName/_search'
+    methods.forEach((m) => {
+      const { collection, operation } = parsePath(path, m.name)
+      const expectedOp = `search`
+      t.equal(collection, 'indexName', `index should be 'indexName'`)
+      t.equal(operation, expectedOp, `operation should be 'search'`)
+    })
+    t.end()
+  })
+  t.test('search of all indices', function (t) {
+    const path = '/_search/'
+    methods.forEach((m) => {
+      if (m.name === 'PUT') {
+        // skip PUT
+        return
+      }
+      const { collection, operation } = parsePath(path, m.name)
+      const expectedOp = `search`
+      t.equal(collection, 'any', 'index should be `any`')
+      t.equal(operation, expectedOp, `operation should match ${expectedOp}`)
+    })
+    t.end()
+  })
+  t.test('doc', function (t) {
+    const path = '/indexName/_doc/testKey'
+    methods.forEach((m) => {
+      const { collection, operation } = parsePath(path, m.name)
+      const expectedOp = `doc.${m.expected}`
+      t.equal(collection, 'indexName', `index should be 'indexName'`)
+      t.equal(operation, expectedOp, `operation should match ${expectedOp}`)
+    })
+    t.end()
+  })
+  t.test('path is /', function (t) {
+    const path = '/'
+    methods.forEach((m) => {
+      const { collection, operation } = parsePath(path, m.name)
+      const expectedOp = `index.${m.expected}`
+      t.equal(collection, 'any', 'index should be `any`')
+      t.equal(operation, expectedOp, `operation should match ${expectedOp}`)
+    })
+    t.end()
+  })
+  t.test(
+    'should provide sensible defaults when path is {} and parser encounters an error',
+    function (t) {
+      const path = {}
+      methods.forEach((m) => {
+        const { collection, operation } = parsePath(path, m.name)
+        const expectedOp = `unknown`
+        t.equal(collection, 'any', 'index should be `any`')
+        t.equal(operation, expectedOp, `operation should match '${expectedOp}'`)
+      })
+      t.end()
+    }
+  )
+})
+
+tap.test('queryParser should behave as expected', (t) => {
+  t.autoend()
+  t.test('given a querystring, it should use that for query', (t) => {
+    const params = JSON.stringify({
+      path: '/_search',
+      method: 'GET',
+      querystring: { q: 'searchterm' }
+    })
+    const expected = {
+      collection: 'any',
+      operation: 'search',
+      query: JSON.stringify({ q: 'searchterm' })
+    }
+    const parseParams = queryParser(params)
+    t.match(parseParams, expected, 'queryParser should handle query strings')
+    t.end()
+  })
+  t.test('given a body, it should use that for query', (t) => {
+    const params = JSON.stringify({
+      path: '/_search',
+      method: 'POST',
+      body: { match: { body: 'document' } }
+    })
+    const expected = {
+      collection: 'any',
+      operation: 'search',
+      query: JSON.stringify({ match: { body: 'document' } })
+    }
+    const parseParams = queryParser(params)
+    t.match(parseParams, expected, 'queryParser should handle query body')
+    t.end()
+  })
+  t.test('given a bulkBody, it should use that for query', (t) => {
+    const params = JSON.stringify({
+      path: '/_msearch',
+      method: 'POST',
+      bulkBody: [
+        {}, // cross-index searches have can have an empty metadata section
+        { query: { match: { body: 'sixth' } } },
+        {},
+        { query: { match: { body: 'bulk' } } }
+      ]
+    })
+    const expected = {
+      collection: 'any',
+      operation: 'msearch',
+      query: JSON.stringify([
+        {}, // cross-index searches have can have an empty metadata section
+        { query: { match: { body: 'sixth' } } },
+        {},
+        { query: { match: { body: 'bulk' } } }
+      ])
+    }
+    const parseParams = queryParser(params)
+    t.match(parseParams, expected, 'queryParser should handle query body')
+    t.end()
+  })
+})