feat: Added instrumentation for @opensearch-projects/opensearch v2.1.0+

docker-compose.yml

@@ -21,6 +21,28 @@ services:
       interval: 30s
       timeout: 10s
       retries: 5
+
+  opensearch:
+    container_name: nr_node_opensearch
+    image: opensearchproject/opensearch:2.1.0 
+    environment:
+      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
+      # Disable password
+      - "DISABLE_SECURITY_PLUGIN=true"
+      # Set cluster to single node
+      - "discovery.type=single-node"
+      # Disable high watermarks, used in CI as the runner is constrained on disk space
+      - "cluster.routing.allocation.disk.threshold_enabled=false"
+      - "network.host=_site_"
+      - "transport.host=127.0.0.1"
+      - "http.host=0.0.0.0"
+    ports:
+      - "9201:9200"
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://127.0.0.1:9200"]
+      interval: 30s
+      timeout: 10s
+      retries: 5
 
   # Kafka setup based on the e2e tests in node-rdkafka. Needs both the
   # `zookeeper` and `kafka` services.

lib/instrumentation/@opensearch-project/opensearch.js

@@ -0,0 +1,161 @@
+/*
+ * Copyright 2024 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+const { QuerySpec } = require('../../shim/specs')
+const semver = require('semver')
+const logger = require('../../logger').child({ component: 'OpenSearch' })
+const { isNotEmpty } = require('../../util/objects')
+
+/**
+ * Instruments the `@opensearch-project/opensearch` module. This function is
+ * passed to `onRequire` when instantiating instrumentation.
+ *
+ * @param {object} _agent New Relic agent
+ * @param {object} opensearch resolved module
+ * @param {string} _moduleName string representation of require/import path
+ * @param {object} shim New Relic shim
+ * @returns {void}
+ */
+module.exports = function initialize(_agent, opensearch, _moduleName, shim) {
+  const pkgVersion = shim.pkgVersion
+  if (semver.lt(pkgVersion, '2.1.0')) {
+    shim &&
+      shim.logger.debug(
+        `Opensearch support is for versions 2.1.0 and above. Not instrumenting ${pkgVersion}.`
+      )
+    return
+  }
+
+  shim.setDatastore(shim.OPENSEARCH)
+  shim.setParser(queryParser)
+
+  shim.recordQuery(
+    opensearch.Transport.prototype,
+    'request',
+    function wrapQuery(shim, _, __, args) {
+      const ctx = this
+      return new QuerySpec({
+        query: JSON.stringify(args?.[0]),
+        promise: true,
+        opaque: true,
+        inContext: function inContext() {
+          getConnection.call(ctx, shim)
+        }
+      })
+    }
+  )
+}
+
+/**
+ * Parses the parameters sent to opensearch 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 (isNotEmpty(params.querystring)) {
+    queryParam = params.querystring
+  }
+  // let body or bulkBody override querystring, as some requests have both
+  if (isNotEmpty(params.body)) {
+    queryParam = params.body
+  } else if (Array.isArray(params.bulkBody) && params.bulkBody.length) {
+    queryParam = params.bulkBody
+  }
+  // The helper interface provides a simpler API:
+
+  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 @opensearch-project/opensearch
+ * @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
+ * opensearch
+ *
+ * @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

@@ -11,6 +11,7 @@ const InstrumentationDescriptor = require('./instrumentation-descriptor')
 module.exports = function instrumentations() {
   return {
     '@elastic/elasticsearch': { type: InstrumentationDescriptor.TYPE_DATASTORE },
+    '@opensearch-project/opensearch': { type: InstrumentationDescriptor.TYPE_DATASTORE },
     '@grpc/grpc-js': { module: './instrumentation/grpc-js' },
     '@hapi/hapi': { type: InstrumentationDescriptor.TYPE_WEB_FRAMEWORK },
     '@hapi/vision': { type: InstrumentationDescriptor.TYPE_WEB_FRAMEWORK },

lib/shim/datastore-shim.js

@@ -38,6 +38,7 @@ const DATASTORE_NAMES = {
   MONGODB: 'MongoDB',
   MYSQL: 'MySQL',
   NEPTUNE: 'Neptune',
+  OPENSEARCH: 'OpenSearch',
   POSTGRES: 'Postgres',
   REDIS: 'Redis',
   PRISMA: 'Prisma'

test/lib/params.js

@@ -28,6 +28,8 @@ module.exports = {
 
   elastic_host: process.env.NR_NODE_TEST_ELASTIC_HOST || 'localhost',
   elastic_port: process.env.NR_NODE_TEST_ELASTIC_PORT || 9200,
+  opensearch_host: process.env.NR_NODE_TEST_OPENSEARCH_HOST || 'localhost',
+  opensearch_port: process.env.NR_NODE_TEST_OPENSEARCH_PORT || 9201,
 
   postgres_host: process.env.NR_NODE_TEST_POSTGRES_HOST || 'localhost',
   postgres_port: process.env.NR_NODE_TEST_POSTGRES_PORT || 5432,

test/unit/instrumentation/opensearch.test.js

@@ -0,0 +1,154 @@
+/*
+ * Copyright 2023 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+const test = require('node:test')
+const assert = require('node:assert')
+const {
+  parsePath,
+  queryParser
+} = require('../../../lib/instrumentation/@opensearch-project/opensearch')
+const instrumentation = require('../../../lib/instrumentation/@opensearch-project/opensearch')
+const methods = [
+  { name: 'GET', expected: 'get' },
+  { name: 'PUT', expected: 'create' },
+  { name: 'POST', expected: 'create' },
+  { name: 'DELETE', expected: 'delete' },
+  { name: 'HEAD', expected: 'exists' }
+]
+
+test('should log warning if version is not supported', async () => {
+  const shim = {
+    pkgVersion: '2.0.0',
+    logger: {
+      debug(msg) {
+        assert.equal(
+          msg,
+          'Opensearch support is for versions 2.1.0 and above. Not instrumenting 2.0.0.'
+        )
+      }
+    }
+  }
+  instrumentation({}, {}, '@opensearch-project/opensearch', shim)
+})
+test('parsePath should behave as expected', async (t) => {
+  await t.test('indices', async function () {
+    const path = '/indexName'
+    methods.forEach((m) => {
+      const { collection, operation } = parsePath(path, m.name)
+      const expectedOp = `index.${m.expected}`
+      assert.equal(collection, 'indexName', `index should be 'indexName'`)
+      assert.equal(operation, expectedOp, 'operation should include index and method')
+    })
+  })
+  await t.test('search of one index', async function () {
+    const path = '/indexName/_search'
+    methods.forEach((m) => {
+      const { collection, operation } = parsePath(path, m.name)
+      const expectedOp = `search`
+      assert.equal(collection, 'indexName', `index should be 'indexName'`)
+      assert.equal(operation, expectedOp, `operation should be 'search'`)
+    })
+  })
+  await t.test('search of all indices', async function () {
+    const path = '/_search/'
+    methods.forEach((m) => {
+      if (m.name === 'PUT') {
+        // skip PUT
+        return
+      }
+      const { collection, operation } = parsePath(path, m.name)
+      const expectedOp = `search`
+      assert.equal(collection, 'any', 'index should be `any`')
+      assert.equal(operation, expectedOp, `operation should match ${expectedOp}`)
+    })
+  })
+  await t.test('doc', async function () {
+    const path = '/indexName/_doc/testKey'
+    methods.forEach((m) => {
+      const { collection, operation } = parsePath(path, m.name)
+      const expectedOp = `doc.${m.expected}`
+      assert.equal(collection, 'indexName', `index should be 'indexName'`)
+      assert.equal(operation, expectedOp, `operation should match ${expectedOp}`)
+    })
+  })
+  await t.test('path is /', async function () {
+    const path = '/'
+    methods.forEach((m) => {
+      const { collection, operation } = parsePath(path, m.name)
+      const expectedOp = `index.${m.expected}`
+      assert.equal(collection, 'any', 'index should be `any`')
+      assert.equal(operation, expectedOp, `operation should match ${expectedOp}`)
+    })
+  })
+  await t.test(
+    'should provide sensible defaults when path is {} and parser encounters an error',
+    function () {
+      const path = {}
+      methods.forEach((m) => {
+        const { collection, operation } = parsePath(path, m.name)
+        const expectedOp = `unknown`
+        assert.equal(collection, 'any', 'index should be `any`')
+        assert.equal(operation, expectedOp, `operation should match '${expectedOp}'`)
+      })
+    }
+  )
+})
+
+test('queryParser should behave as expected', async (t) => {
+  await t.test('given a querystring, it should use that for query', () => {
+    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)
+    assert.deepEqual(parseParams, expected, 'queryParser should handle query strings')
+  })
+  await t.test('given a body, it should use that for query', () => {
+    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)
+    assert.deepEqual(parseParams, expected, 'queryParser should handle query body')
+  })
+  await t.test('given a bulkBody, it should use that for query', () => {
+    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.create',
+      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)
+    assert.deepEqual(parseParams, expected, 'queryParser should handle query body')
+  })
+})

test/versioned/opensearch/newrelic.js

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2020 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+exports.config = {
+  app_name: ['opensearch test'],
+  license_key: 'license key here',
+  utilization: {
+    detect_aws: false,
+    detect_pcf: false,
+    detect_azure: false,
+    detect_gcp: false,
+    detect_docker: false
+  },
+  logging: {
+    enabled: true
+  }
+}