[pinpoint-apm#223] commonStreamResponse's sequenceid per requestid of…

… Active Thread Count

* DisableTrace negative transactionId sequence generator
It's Java Agent use transactionId's sequence use active request map key

lib/client/grpc-data-sender.js

@@ -39,14 +39,6 @@ class GrpcDataSender {
     this.initializeProfilerClients(collectorIp, collectorTcpPort, config)
   }
 
-  setCommandEchoCallArguments(callArguments) {
-    this.commandEchoCallArguments = callArguments
-  }
-
-  setCommandStreamActiveThreadCountCallArguments(callArguments) {
-    this.commandStreamActiveThreadCountCallArguments = callArguments
-  }
-
   close() {
     this.closeScheduler()
     if (this.spanStream) {
@@ -145,10 +137,11 @@ class GrpcDataSender {
     this.profilerClient = new services.ProfilerCommandServiceClient(collectorIp + ":" + collectorTcpPort, grpc.credentials.createInsecure(), profilerBuilder.build())
   }
 
-  makeCommandStream() {
+  makeCommandStream(callArguments) {
     this.commandStream = new GrpcReadableStream(() => {
       const writable = this.profilerClient.handleCommandV2()
 
+      let activeThreadCountSequenceId = 0
       writable.on('data', (cmdRequest) => {
         const requestId = cmdRequest.getRequestid()
         const command = cmdRequest.getCommandCase()
@@ -166,12 +159,13 @@ class GrpcDataSender {
             const message = cmdRequest.getCommandecho().getMessage()
             cmdEchoResponse.setMessage(message)
 
-            this.sendCommandEcho(cmdEchoResponse, this.commandEchoCallArguments)
+            this.sendCommandEcho(cmdEchoResponse, callArguments)
           },
+          // ActiveThreadCountStreamSocket.java
           'ACTIVE_THREAD_COUNT': () => {
             const commonStreamResponse = new cmdMessages.PCmdStreamResponse()
             commonStreamResponse.setResponseid(requestId)
-            commonStreamResponse.setSequenceid(1)
+            commonStreamResponse.setSequenceid(++activeThreadCountSequenceId)
             const stringValue = new wrappers.StringValue()
             stringValue.setValue('')
             commonStreamResponse.setMessage(stringValue)
@@ -182,17 +176,17 @@ class GrpcDataSender {
             commandActiveThreadCountResponse.addActivethreadcount(1)
             commandActiveThreadCountResponse.setTimestamp(Date.now())
 
-            this.sendActiveThreadCount(commandActiveThreadCountResponse)
+            this.sendActiveThreadCount(commandActiveThreadCountResponse, callArguments)
           }
         })
       })
       return writable
     })
   }
 
-  makeActiveThreadCountStream() {
+  makeActiveThreadCountStream(callArguments) {
     this.activeThreadCountStream = new GrpcReadableStream(() => {
-      const callArguments = guardCallArguments(this.commandStreamActiveThreadCountCallArguments)
+      callArguments = guardCallArguments(callArguments)
       const metadata = callArguments.getMetadata()
       let options = callArguments.getOptions()
       const callback = callArguments.getCallback()
@@ -361,9 +355,9 @@ class GrpcDataSender {
     }
   }
 
-  sendSupportedServicesCommand() {
+  sendSupportedServicesCommand(callArguments) {
     if (!this.commandStream) {
-      this.makeCommandStream()
+      this.makeCommandStream(callArguments)
     }
 
     const pCmdMessage = CommandType.supportedServicesCommandMessage()
@@ -388,9 +382,9 @@ class GrpcDataSender {
     })
   }
 
-  sendActiveThreadCount(commandActiveThreadCountResponse) {
+  sendActiveThreadCount(commandActiveThreadCountResponse, callArguments) {
     if (!this.activeThreadCountStream) {
-      this.makeActiveThreadCountStream()
+      this.makeActiveThreadCountStream(callArguments)
     }
     this.activeThreadCountStream.push(commandActiveThreadCountResponse)
   }

lib/context/id-generator.js → lib/context/span-id.js

@@ -6,7 +6,8 @@
 
 'use strict'
 
-class IdGenerator {
+// SpanId.java in Java agent
+class SpanId {
   constructor () {
     this.MAX_NUM = Number.MAX_SAFE_INTEGER
   }
@@ -15,9 +16,9 @@ class IdGenerator {
     return Math.floor(Math.random() * this.MAX_NUM)
   }
 
-  stringValueOfNext() {
+  newSpanId() {
     return this.next.toString()
   }
 }
 
-module.exports = new IdGenerator()
+module.exports = new SpanId()

lib/context/trace-context.js

@@ -9,7 +9,7 @@
 const Trace = require('./trace')
 const TransactionId = require('./transaction-id')
 const TraceId = require('./trace-id')
-const IdGenerator = require('./id-generator')
+const SpanId = require('./span-id')
 const log = require('../utils/logger')
 const sampler = require('../sampler/sampler')
 const DisableTrace = require('./disable-trace')
@@ -65,7 +65,7 @@ class TraceContext {
       return new DisableTrace()
     }
     const transactionId = new TransactionId(this.agentInfo.agentId, this.agentInfo.agentStartTime.toString())
-    const spanId = IdGenerator.stringValueOfNext()
+    const spanId = SpanId.newSpanId()
     const traceId = new TraceId(transactionId, spanId)
     return this.createTraceObject(traceId)
   }

lib/instrumentation/http-shared.js

@@ -9,7 +9,7 @@
 const endOfStream = require('end-of-stream')
 const url = require('url')
 const log = require('../utils/logger')
-const IdGenerator = require('../context/id-generator')
+const SpanId = require('../context/span-id')
 const annotationKey = require('../constant/annotation-key')
 const RequestHeaderUtils = require('../instrumentation/request-header-utils')
 const AntPathMatcher = require('../utils/ant-path-matcher')
@@ -129,7 +129,7 @@ exports.traceOutgoingRequest = function (agent, moduleName) {
       asyncEventRecorder.recordAttribute(annotationKey.HTTP_URL, httpURL)
 
       const destinationId = host
-      const nextSpanId = IdGenerator.stringValueOfNext()
+      const nextSpanId = SpanId.newSpanId()
       RequestHeaderUtils.write(req, agent, nextSpanId, destinationId)
       asyncEventRecorder.recordNextSpanId(nextSpanId)
       asyncEventRecorder.recordDestinationId(destinationId)

lib/instrumentation/request-header-utils.js

@@ -50,6 +50,7 @@ class RequestHeaderUtils {
       return
     }
 
+    // RequestTraceReader.newTraceId(TraceHeader traceHeader) in Java agent
     const transactionId = TransactionId.toTransactionId(this.getHeader(request, PinpointHeader.HTTP_TRACE_ID))
     if (!transactionId) {
       return

lib/metric/active-thread-count.js

@@ -8,7 +8,6 @@
 
 const Scheduler = require('../utils/scheduler')
 const activeTrace = require('./active-trace')
-const log = require('../utils/logger')
 
 class ActiveThreadCount {
   constructor(dataSender, streamChannelManager, enabled) {

test/client/grpc-data-sender.test.js

@@ -47,7 +47,7 @@ class DataSource extends DataSourceCallCountable {
   initializeProfilerClients() { }
 }
 
-test.skip('Should send span ', function (t) {
+test('Should send span ', function (t) {
   const expectedSpan = {
     'traceId': {
       'transactionId': {
@@ -301,7 +301,7 @@ test.skip('sendSpanChunk redis.SET.end', function (t) {
   })
 })
 
-test.skip('sendSpanChunk redis.GET.end', (t) => {
+test('sendSpanChunk redis.GET.end', (t) => {
   let expectedSpanChunk = {
     'agentId': 'express-node-sample-id',
     'applicationName': 'express-node-sample-name',
@@ -420,7 +420,7 @@ test.skip('sendSpanChunk redis.GET.end', (t) => {
   })
 })
 
-test.skip('sendSpan', (t) => {
+test('sendSpan', (t) => {
   let expectedSpanChunk = {
     'traceId': {
       'transactionId': {
@@ -805,7 +805,6 @@ const emptyResponseService = (call, callback) => {
     }
   })
 
-
   const succeedOnRetryAttempt = call.metadata.get('succeed-on-retry-attempt')
   const previousAttempts = call.metadata.get('grpc-previous-rpc-attempts')
   const callRequests = getCallRequests()
@@ -859,8 +858,7 @@ test('sendSupportedServicesCommand and commandEcho', (t) => {
       t.equal(cmdEchoResponse.getMessage(), 'echo', 'echo message')
       afterOne(t)
     }).build()
-    dataSender.setCommandEchoCallArguments(callArguments)
-    dataSender.sendSupportedServicesCommand()
+    dataSender.sendSupportedServicesCommand(callArguments)
   })
 
   t.teardown(() => {
@@ -880,25 +878,30 @@ test('CommandStreamActiveThreadCount', (t) => {
   server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
     dataSender = beforeSpecificOne(port, ProfilerDataSource)
 
+    let callCount = 0
     dataCallbackOnServerCall = (data) => {
+      ++callCount
       const commonStreamResponse = data.getCommonstreamresponse()
       t.equal(commonStreamResponse.getResponseid(), requestId, 'response id matches request id')
-      t.equal(commonStreamResponse.getSequenceid(), 1, 'sequenceid is 1')
+      t.equal(commonStreamResponse.getSequenceid(), callCount, `sequenceid is ${callCount}`)
       t.equal(commonStreamResponse.getMessage().getValue(), '', 'message is empty')
 
       t.equal(data.getHistogramschematype(), 2, 'histogram schema type')
       t.equal(data.getActivethreadcountList()[0], 1, 'active thread count')
-      afterOne(t)
+
+      if (callCount == 2) {
+        afterOne(t)
+      }
     }
 
     const callArguments = new CallArgumentsBuilder(function (error, response) {
       serverCallWriter(CommandType.activeThreadCount)
+      serverCallWriter(CommandType.activeThreadCount)
     }).build()
-    dataSender.setCommandEchoCallArguments(callArguments)
-    dataSender.sendSupportedServicesCommand()
+    dataSender.sendSupportedServicesCommand(callArguments)
   })
   t.teardown(() => {
     dataSender.close()
     server.forceShutdown()
   })
-})
+})

test/client/grpc-fixture.js

@@ -102,9 +102,9 @@ class DataSourceCallCountable extends GrpcDataSender {
         super.sendSpan(span)
     }
 
-    sendSupportedServicesCommand() {
+    sendSupportedServicesCommand(callArguments) {
         increaseCallCount()
-        super.sendSupportedServicesCommand()
+        super.sendSupportedServicesCommand(callArguments)
     }
 }
 

test/context/trace-id.test.js

@@ -7,15 +7,15 @@
 const test = require('tape')
 const TransactionId = require('../../lib/context/transaction-id')
 const TraceId = require('../../lib/context/trace-id')
-const IdGenerator = require('../../lib/context/id-generator')
+const SpanId = require('../../lib/context/span-id')
 
 test('Should create', function (t) {
   t.plan(2)
 
   const agentId = 'agent-for-dev'
   const agentStartTime = Date.now()
   const transactionId = new TransactionId(agentId, agentStartTime.toString())
-  const spanId = IdGenerator.stringValueOfNext()
+  const spanId = SpanId.newSpanId()
   const traceId = new TraceId(transactionId, spanId)
 
   t.ok(traceId)

test/fixture.js

@@ -6,7 +6,7 @@
 
 const TransactionId = require('../lib/context/transaction-id')
 const TraceId = require('../lib/context/trace-id')
-const IdGenerator = require('../lib/context/id-generator')
+const SpanId = require('../lib/context/span-id')
 const shimmer = require('@pinpoint-apm/shimmer')
 const testConfig= require('./pinpoint-config-test')
 require('../lib/config').clear()
@@ -22,7 +22,7 @@ const getTransactionId = () => {
 }
 
 const getTraceId = (transactionId) => {
-  const spanId = IdGenerator.stringValueOfNext()
+  const spanId = SpanId.newSpanId()
   return new TraceId(transactionId || getTransactionId(), spanId)
 }