Update ES Database to include msearch, and mget calls as known DB type calls

CHANGELOG.md

@@ -1,10 +1,13 @@
 # Changelog
 
+## Unreleased
+- Instrument Elasticsearch operations msearch and mget ([#117](https://github.com/instana/nodejs-sensor/pull/117), thanks to @DtRWoS).
+
 ## 1.59.0
 - Add GRPC instrumentation ([grpc](https://www.npmjs.com/package/grpc)).
 
 ## 1.58.0
-- Support Node.js 11
+- Support Node.js 11.
 
 ## 1.57.0
 - Provide an API to set a logger after initialization to resolve init/logger cycle.
@@ -15,9 +18,9 @@
 - Left pad generated IDs with '0'.
 
 ## 1.55.2
-- Fix: Add HTTP query params for HTTP exits
-- Fix: Do not capture HTTP headers for HTTP exits (only for entries)
-- Fix: Capture erroneous span on synchronous exceptions in HTTP client (like malformed URL)
+- Fix: Add HTTP query params for HTTP exits.
+- Fix: Do not capture HTTP headers for HTTP exits (only for entries).
+- Fix: Capture erroneous span on synchronous exceptions in HTTP client (like malformed URL).
 
 ## 1.55.1
 - Fix method name of NoopSpanHandle.
@@ -26,7 +29,7 @@
 - Provide API to end spans manually to allow capturing of child exit spans for message consumers (RabbitMQ/amqplib, Kafka).
 
 ## 1.54.2
-- Fix broken trace context in some MongoDB usage scenarios
+- Fix broken trace context in some MongoDB usage scenarios.
 
 ## 1.54.1
 - RabbitMQ/amqplib: Small fix for TLS AMQP Urls (amqps).

package.json

@@ -97,6 +97,10 @@
     {
       "name": "Denis Fäcke",
       "email": "[email protected]"
+    },
+    {
+      "name": "DtRWoS",
+      "email": "[email protected]"
     }
   ],
   "license": "MIT",

src/tracing/instrumentation/database/elasticsearch.js

@@ -22,6 +22,8 @@ function instrument(es) {
     instrumentApi(client, 'search', info);
     instrumentApi(client, 'index', info);
     instrumentApi(client, 'get', info);
+    instrumentApi(client, 'msearch', info);
+    instrumentApi(client, 'mget', info);
 
     return client;
   };
@@ -53,15 +55,25 @@ function instrumentApi(client, action, info) {
     span.data = {
       elasticsearch: {
         action: action,
-        cluster: info.clusterName,
-        index: toStringEsMultiParameter(params.index),
-        type: toStringEsMultiParameter(params.type),
-        stats: toStringEsMultiParameter(params.stats),
-        id: action === 'get' ? params.id : undefined,
-        query: action === 'search' ? tracingUtil.shortenDatabaseStatement(JSON.stringify(params)) : undefined
+        cluster: info.clusterName
       }
     };
 
+    if (action === 'mget' && params.body && params.body.docs && Array.isArray(params.body.docs)) {
+      getSpanDataFromMget1(span, params.body.docs);
+    } else if (action === 'mget' && params.body && params.body.ids && Array.isArray(params.body.ids)) {
+      getSpanDataFromMget2(span, params);
+    } else if (action === 'msearch' && Array.isArray(params.body)) {
+      getSpanDataFromMsearch(span, params.body);
+    } else {
+      span.data.elasticsearch.index = toStringEsMultiParameter(params.index);
+      span.data.elasticsearch.type = toStringEsMultiParameter(params.type);
+      span.data.elasticsearch.stats = toStringEsMultiParameter(params.stats);
+      span.data.elasticsearch.id = action === 'get' ? params.id : undefined;
+      span.data.elasticsearch.query =
+        action === 'search' ? tracingUtil.shortenDatabaseStatement(JSON.stringify(params)) : undefined;
+    }
+
     cls.ns.bind(cb);
 
     if (arguments.length === 2) {
@@ -89,6 +101,10 @@ function instrumentApi(client, action, info) {
     function onSuccess(response) {
       if (response.hits != null && response.hits.total != null) {
         span.data.elasticsearch.hits = response.hits.total;
+      } else if (response.responses != null && Array.isArray(response.responses)) {
+        span.data.elasticsearch.hits = response.responses.reduce(function(hits, res) {
+          return hits + (res.hits && typeof res.hits.total === 'number' ? res.hits.total : 0);
+        }, 0);
       }
       span.d = Date.now() - span.ts;
       span.error = false;
@@ -123,6 +139,53 @@ function toStringEsMultiParameter(param) {
   return JSON.stringify(param);
 }
 
+function getSpanDataFromMget1(span, docs) {
+  var indices = [];
+  var types = [];
+  var stats = [];
+  var ids = [];
+  for (var i = 0; i < docs.length; i++) {
+    collectParamFrom(docs[i], '_index', indices);
+    collectParamFrom(docs[i], '_type', types);
+    collectParamFrom(docs[i], '_stats', stats);
+    collectParamFrom(docs[i], '_id', ids);
+  }
+  span.data.elasticsearch.index = indices.length > 0 ? indices.join(',') : undefined;
+  span.data.elasticsearch.type = types.length > 0 ? types.join(',') : undefined;
+  span.data.elasticsearch.stats = stats.length > 0 ? stats.join(',') : undefined;
+  span.data.elasticsearch.id = ids.length > 0 ? ids.join(',') : undefined;
+}
+
+function getSpanDataFromMget2(span, params) {
+  span.data.elasticsearch.index = params.index ? toStringEsMultiParameter(params.index) : undefined;
+  span.data.elasticsearch.type = params.index ? toStringEsMultiParameter(params.type) : undefined;
+  span.data.elasticsearch.stats = params.index ? toStringEsMultiParameter(params.stats) : undefined;
+  span.data.elasticsearch.id = params.body.ids.length > 0 ? params.body.ids.join(',') : undefined;
+}
+
+function getSpanDataFromMsearch(span, body) {
+  var indices = [];
+  var types = [];
+  var stats = [];
+  for (var i = 0; i < body.length; i++) {
+    collectParamFrom(body[i], 'index', indices);
+    collectParamFrom(body[i], 'type', types);
+    collectParamFrom(body[i], 'stats', stats);
+  }
+  span.data.elasticsearch.index = indices.length > 0 ? indices.join(',') : undefined;
+  span.data.elasticsearch.type = types.length > 0 ? types.join(',') : undefined;
+  span.data.elasticsearch.stats = stats.length > 0 ? stats.join(',') : undefined;
+}
+
+function collectParamFrom(bodyItem, key, accumulator) {
+  if (bodyItem && bodyItem[key]) {
+    var value = toStringEsMultiParameter(bodyItem[key]);
+    if (value != null && accumulator.indexOf(value) < 0) {
+      accumulator.push(value);
+    }
+  }
+}
+
 exports.activate = function() {
   isActive = true;
 };

test/tracing/database/elasticsearch/app.js

@@ -29,7 +29,6 @@ app.get('/', function(req, res) {
 });
 
 app.get('/get', function(req, res) {
-  log('Getting document', req.query.id);
   client.get(
     {
       index: req.query.index || 'myindex',
@@ -47,7 +46,6 @@ app.get('/get', function(req, res) {
 });
 
 app.get('/search', function(req, res) {
-  log('Searching document', req.query.q);
   client
     .search({
       index: req.query.index || 'myindex',
@@ -64,16 +62,84 @@ app.get('/search', function(req, res) {
     );
 });
 
+app.get('/mget1', function(req, res) {
+  var ids = req.query.id;
+  if (!Array.isArray(ids) || ids.length < 2) {
+    return res.status(400).json({ error: 'You need to provide an array of at least two document IDs.' });
+  }
+  client.mget(
+    {
+      body: {
+        docs: [
+          { _index: req.query.index || 'myindex', _type: 'mytype', _id: ids[0] },
+          { _index: req.query.index || 'myindex', _type: 'mytype', _id: ids[1] }
+        ]
+      }
+    },
+    function(error, response) {
+      if (error) {
+        res.status(500).json(error);
+      } else {
+        res.json(response);
+      }
+    }
+  );
+});
+
+app.get('/mget2', function(req, res) {
+  var ids = req.query.id;
+  if (!Array.isArray(ids) || ids.length < 2) {
+    return res.status(400).json({ error: 'You need to provide an array of at least two document IDs.' });
+  }
+  client.mget(
+    {
+      index: req.query.index || 'myindex',
+      type: 'mytype',
+      body: {
+        ids: ids
+      }
+    },
+    function(error, response) {
+      if (error) {
+        res.status(500).json(error);
+      } else {
+        res.json(response);
+      }
+    }
+  );
+});
+
+app.get('/msearch', function(req, res) {
+  var queryStrings = req.query.q;
+  if (!Array.isArray(queryStrings) || queryStrings.length < 2) {
+    return res.status(400).json({ error: 'You need to provide an array of at least two query params' });
+  }
+  var query = {
+    body: [
+      { index: req.query.index || 'myindex', type: 'mytype' },
+      { query: { query_string: { query: req.query.q[0] } } },
+      { index: req.query.index || 'myindex', type: 'mytype' },
+      { query: { query_string: { query: req.query.q[1] } } }
+    ]
+  };
+  client.msearch(query).then(
+    function(response) {
+      res.json(response);
+    },
+    function(error) {
+      res.status(500).json(error);
+    }
+  );
+});
+
 app.post('/index', function(req, res) {
-  log('Indexing document', req.body);
   client
     .index({
       index: req.query.index || 'myindex',
       type: 'mytype',
       body: req.body
     })
     .then(function(response) {
-      log('Refreshing index');
       return client.indices
         .refresh({
           index: '_all',
@@ -85,14 +151,13 @@ app.post('/index', function(req, res) {
             return response;
           },
           function(error) {
-            log('Index refresh failed.');
+            log('Index refresh failed.', error);
             throw error;
           }
         );
     })
     .then(
       function(response) {
-        log('Sending indexing response for document ' + response._id);
         res.json(response);
       },
       function(error) {

test/tracing/database/elasticsearch/controls.js

@@ -70,6 +70,18 @@ exports.search = function(opts) {
   return requestWithPath('GET', '/search', opts);
 };
 
+exports.mget1 = function(opts) {
+  return requestWithPath('GET', '/mget1', opts);
+};
+
+exports.mget2 = function(opts) {
+  return requestWithPath('GET', '/mget2', opts);
+};
+
+exports.msearch = function(opts) {
+  return requestWithPath('GET', '/msearch', opts);
+};
+
 exports.searchAndGet = function(opts) {
   return requestWithPath('GET', '/searchAndGet', opts);
 };