[pinpoint-apm#285] Client Side gRPC Stream deadline retry refactoring

   * The stream deadline configuration is loaded from pinpoint-config-default.json
   * Any value (null, undefined) should not terminate the agent because it needs to load JSON data and perform arithmetic

Author: feelform

lib/client/grpc-readable-stream-builder.js

@@ -74,6 +74,10 @@ class GrpcReadableStream {
             return
         }
 
+        if (this.ended) {
+            return
+        }
+
         try {
             const options = this.deadlineOptionsBuilder?.build() ?? {}
             const writableStream = this.makeWritableStream(options, (error, response) => {

test/client/active-request-grpc-readable-stream.test.js

@@ -18,11 +18,7 @@ const GrpcReadableStream = require('../../lib/client/grpc-readable-stream')
 const semver = require('semver')
 
 test('If you run the ActiveRequest function, and then Node agent send the fifth piece of data, and the Pinpoint server fails, the ActiveRequest gRPC Stream closes and the for statement stops', (t) => {
-    if (semver.satisfies(process.versions.node, '<19.0')) {
-        t.plan(56)
-    } else {
-        t.plan(55)
-    }
+    t.plan(56)
     const server = new grpc.Server()
     let requestId = 1
     let activeRequestCount = 0
@@ -180,8 +176,7 @@ test('If you run the ActiveRequest function, and then Node agent send the fifth
                     dataSender.close()
                     handleCommandCall.end()
                     shimmer.unwrap(GrpcReadableStream.prototype, 'pipeWritableStream')
-                    server2.tryShutdown(() => {
-                    })
+                    server2.forceShutdown()
                 })
             }
         })
@@ -235,11 +230,7 @@ test('If you run the ActiveRequest function, and deadline occurs, the ActiveRequ
                     const commandStream1st = this
                     t.equal(commandStream1st.name, '', '1st created stream is commandStream creating writable stream')
                     commandStream1st.writableStream.on('end', () => {
-                        if (semver.satisfies(process.versions.node, '<19.0')) {
-                            t.true(firedError, 'The first commandStream is ended by Functional Test is Ended')
-                        } else {
-                            t.false(firedError, 'The first commandStream is ended by deadline')
-                        }
+                        t.true(firedError, 'The first commandStream is ended by Functional Test is Ended')
                     })
                     let firedError
                     commandStream1st.writableStream.on('error', (error) => {
@@ -296,8 +287,7 @@ test('If you run the ActiveRequest function, and deadline occurs, the ActiveRequ
     t.teardown(() => {
         dataSender.close()
         handleCommandCall.end()
-        server.tryShutdown(() => {
-            shimmer.unwrap(GrpcReadableStream.prototype, 'pipeWritableStream')
-        })
+        shimmer.unwrap(GrpcReadableStream.prototype, 'pipeWritableStream')
+        server.forceShutdown()
     })
 })

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

@@ -8,7 +8,7 @@ const test = require('tape')
 const AsyncId = require('../../lib/context/async-id')
 const grpc = require('@grpc/grpc-js')
 const services = require('../../lib/data/v1/Service_grpc_pb')
-const { beforeSpecificOne, afterOne, getCallRequests, getMetadata, DataSourceCallCountable } = require('./grpc-fixture')
+const { beforeSpecificOne, afterOne, getCallRequests, getMetadata, DataSourceCallCountable, SpanOnlyFunctionalTestableDataSource } = require('./grpc-fixture')
 const cmdMessage = require('../../lib/data/v1/Cmd_pb')
 const CommandType = require('../../lib/client/command/command-type')
 const { Empty } = require('google-protobuf/google/protobuf/empty_pb')
@@ -56,19 +56,6 @@ function sendSpan(call) {
   callMetadata.push(call.metadata)
 }
 
-class DataSource extends DataSourceCallCountable {
-  constructor(collectorIp, collectorTcpPort, collectorStatPort, collectorSpanPort, agentInfo, config) {
-    super(collectorIp, collectorTcpPort, collectorStatPort, collectorSpanPort, agentInfo, config)
-  }
-
-  initializeClients() { }
-  initializeMetadataClients() { }
-  initializeStatStream() { }
-  initializePingStream() { }
-  initializeAgentInfoScheduler() { }
-  initializeProfilerClients() { }
-}
-
 test('Should send span', function (t) {
   agent.bindHttp()
   sendSpanMethodOnDataCallback = null
@@ -79,7 +66,7 @@ test('Should send span', function (t) {
   })
   let dataSender
   server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
-    const grpcDataSender = beforeSpecificOne(port, DataSource)
+    const grpcDataSender = beforeSpecificOne(port, SpanOnlyFunctionalTestableDataSource)
     const traceRoot = new RemoteTraceRootBuilder(agent.agentInfo, '5').build()
     dataSender = dataSenderMock(agent.config, grpcDataSender)
     const spanBuilder = new SpanBuilder(traceRoot)
@@ -175,7 +162,7 @@ test('sendSpanChunk redis.SET.end', function (t) {
 
   let dataSender
   server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
-    const grpcDataSender = beforeSpecificOne(port, DataSource)
+    const grpcDataSender = beforeSpecificOne(port, SpanOnlyFunctionalTestableDataSource)
     const traceRoot = new RemoteTraceRootBuilder(agent.agentInfo, '5').build()
     const asyncId = AsyncId.make()
     dataSender = dataSenderMock(agent.config, grpcDataSender)
@@ -253,7 +240,7 @@ test('sendSpanChunk redis.GET.end', (t) => {
 
   let dataSender
   server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
-    const grpcDataSender = beforeSpecificOne(port, DataSource)
+    const grpcDataSender = beforeSpecificOne(port, SpanOnlyFunctionalTestableDataSource)
     const traceRoot = new RemoteTraceRootBuilder(agent.agentInfo, '5').build()
     const asyncId = AsyncId.make()
     dataSender = dataSenderMock(agent.config, grpcDataSender)
@@ -324,7 +311,7 @@ test('sendSpan', (t) => {
 
   let dataSender
   server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
-    const grpcDataSender = beforeSpecificOne(port, DataSource)
+    const grpcDataSender = beforeSpecificOne(port, SpanOnlyFunctionalTestableDataSource)
     const traceRoot = new RemoteTraceRootBuilder(agent.agentInfo, '5').build()
     dataSender = dataSenderMock(agent.config, grpcDataSender)
     const spanBuilder = new SpanBuilder(traceRoot)
@@ -455,10 +442,9 @@ test('sendStat', (t) => {
         const actualHistogram = data.getAgentstat().getActivetrace().getHistogram().getActivetracecountList()
         t.deepEqual(originalHistogram, actualHistogram, 'active trace histogram')
         agent.dataSender.close()
-        collectorServer.tryShutdown(() => {
-          server.close(() => {
-            t.end()
-          })
+        collectorServer.forceShutdown()
+        server.close(() => {
+          t.end()
         })
       }
     })

test/client/grpc-fixture.js

@@ -29,14 +29,15 @@ function beforeSpecificOne(port, one, serviceConfig) {
     actualConfig.collectorStatPort = port
     actualConfig.collectorSpanPort = port
     actualConfig.enabledDataSending = true
-    return new one(
+    const dataSource = new one(
         actualConfig.collectorIp,
         actualConfig.collectorTcpPort,
         actualConfig.collectorStatPort,
         actualConfig.collectorSpanPort,
         agentInfo(),
         actualConfig
     )
+    return dataSource
 }
 
 function agentInfo() {

test/client/grpc-readable-stream.test.js

@@ -341,8 +341,6 @@ test('When gRPC server shutdown and then node agent grpcStream on error fired an
         if (loadCount === 2) {
             dataSender.close()
             server.forceShutdown()
-            server.tryShutdown(() => {
-            })
             t.end()
         }
     }