Merge "Add GRPC log tracing service."

5 files changed, 266 insertions, 0 deletions

diff --git a/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/BluePrintGrpcExtensions.kt b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/BluePrintGrpcExtensions.kt
new file mode 100644
index 000000000..55cf0941d
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/BluePrintGrpcExtensions.kt
@@ -0,0 +1,27 @@
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.grpc
+
+import io.grpc.Metadata
+
+fun Metadata.getStringKey(key: String): String? {
+    return this.get(Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER))
+}
+
+fun Metadata.putStringKeyValue(key: String, value: String) {
+    this.put(Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER), value)
+}
\ No newline at end of file
diff --git a/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/interceptor/GrpcClientLoggingInterceptor.kt b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/interceptor/GrpcClientLoggingInterceptor.kt
new file mode 100644
index 000000000..f3b14b59f
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/interceptor/GrpcClientLoggingInterceptor.kt
@@ -0,0 +1,47 @@
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.grpc.interceptor
+
+import io.grpc.*
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.service.GrpcLoggerService
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+
+
+class GrpcClientLoggingInterceptor : ClientInterceptor {
+    val log = logger(GrpcClientLoggingInterceptor::class)
+
+    val loggingService = GrpcLoggerService()
+
+    override fun <ReqT, RespT> interceptCall(method: MethodDescriptor<ReqT, RespT>,
+                                             callOptions: CallOptions, channel: Channel): ClientCall<ReqT, RespT> {
+
+        return object : ForwardingClientCall
+        .SimpleForwardingClientCall<ReqT, RespT>(channel.newCall(method, callOptions)) {
+
+            override fun start(responseListener: Listener<RespT>, headers: Metadata) {
+                val listener = object : ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(responseListener) {
+                    override fun onMessage(message: RespT) {
+                        loggingService.grpcInvoking(headers)
+                        super.onMessage(message)
+                    }
+                }
+                super.start(listener, headers)
+            }
+        }
+
+    }
+}
\ No newline at end of file
diff --git a/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/interceptor/GrpcServerLoggingInterceptor.kt b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/interceptor/GrpcServerLoggingInterceptor.kt
new file mode 100644
index 000000000..e21d5d3ce
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/interceptor/GrpcServerLoggingInterceptor.kt
@@ -0,0 +1,92 @@
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.grpc.interceptor
+
+import io.grpc.*
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.service.GrpcLoggerService
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.management.api.BluePrintDownloadInput
+import org.onap.ccsdk.cds.controllerblueprints.management.api.BluePrintRemoveInput
+import org.onap.ccsdk.cds.controllerblueprints.management.api.BluePrintUploadInput
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput
+import org.slf4j.MDC
+
+class GrpcServerLoggingInterceptor : ServerInterceptor {
+    val log = logger(GrpcServerLoggingInterceptor::class)
+    val loggingService = GrpcLoggerService()
+
+    override fun <ReqT : Any, RespT : Any> interceptCall(call: ServerCall<ReqT, RespT>,
+                                                         requestHeaders: Metadata, next: ServerCallHandler<ReqT, RespT>)
+            : ServerCall.Listener<ReqT> {
+
+        val forwardingServerCall = object : ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(call) {
+            override fun sendHeaders(responseHeaders: Metadata) {
+                loggingService.grpResponding(requestHeaders, responseHeaders)
+                super.sendHeaders(responseHeaders)
+            }
+        }
+
+        return object
+            : ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(
+                next.startCall(forwardingServerCall, requestHeaders)) {
+
+            override fun onMessage(message: ReqT) {
+                /** Get the requestId, SubRequestId and Originator Id and set in MDS context
+                 *  If you are using other GRPC services, Implement own Logging Interceptors to get tracing.
+                 * */
+                when (message) {
+                    is ExecutionServiceInput -> {
+                        val commonHeader = message.commonHeader
+                                ?: throw BluePrintProcessorException("missing common header in request")
+                        loggingService.grpcRequesting(call, commonHeader, next)
+                    }
+                    is BluePrintUploadInput -> {
+                        val commonHeader = message.commonHeader
+                                ?: throw BluePrintProcessorException("missing common header in request")
+                        loggingService.grpcRequesting(call, commonHeader, next)
+                    }
+                    is BluePrintDownloadInput -> {
+                        val commonHeader = message.commonHeader
+                                ?: throw BluePrintProcessorException("missing common header in request")
+                        loggingService.grpcRequesting(call, commonHeader, next)
+                    }
+                    is BluePrintRemoveInput -> {
+                        val commonHeader = message.commonHeader
+                                ?: throw BluePrintProcessorException("missing common header in request")
+                        loggingService.grpcRequesting(call, commonHeader, next)
+                    }
+                    else -> {
+                        loggingService.grpcRequesting(call, requestHeaders, next)
+                    }
+                }
+                super.onMessage(message)
+            }
+
+            override fun onComplete() {
+                MDC.clear()
+                super.onComplete()
+            }
+
+            override fun onCancel() {
+                MDC.clear()
+                super.onCancel()
+            }
+
+        }
+    }
+}
\ No newline at end of file
diff --git a/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/GrpcLoggerService.kt b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/GrpcLoggerService.kt
new file mode 100644
index 000000000..1b932480a
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/GrpcLoggerService.kt
@@ -0,0 +1,98 @@
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.grpc.service
+
+import io.grpc.Grpc
+import io.grpc.Metadata
+import io.grpc.ServerCall
+import io.grpc.ServerCallHandler
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.getStringKey
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.putStringKeyValue
+import org.onap.ccsdk.cds.controllerblueprints.common.api.CommonHeader
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants.ONAP_INVOCATION_ID
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants.ONAP_PARTNER_NAME
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants.ONAP_REQUEST_ID
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+import org.onap.ccsdk.cds.controllerblueprints.core.defaultToEmpty
+import org.onap.ccsdk.cds.controllerblueprints.core.defaultToUUID
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.slf4j.MDC
+import java.net.InetSocketAddress
+import java.time.ZoneOffset
+import java.time.ZonedDateTime
+import java.time.format.DateTimeFormatter
+import java.util.*
+
+class GrpcLoggerService {
+
+    private val log = logger(GrpcLoggerService::class)
+
+    /** Used when server receives request */
+    fun <ReqT : Any, RespT : Any> grpcRequesting(call: ServerCall<ReqT, RespT>,
+                                                 headers: Metadata, next: ServerCallHandler<ReqT, RespT>) {
+        val requestID = headers.getStringKey(ONAP_REQUEST_ID).defaultToUUID()
+        val invocationID = headers.getStringKey(ONAP_INVOCATION_ID).defaultToUUID()
+        val partnerName = headers.getStringKey(ONAP_PARTNER_NAME).defaultToUUID()
+        grpcRequesting(requestID, invocationID, partnerName, call)
+    }
+
+    fun <ReqT : Any, RespT : Any> grpcRequesting(call: ServerCall<ReqT, RespT>,
+                                                 headers: CommonHeader, next: ServerCallHandler<ReqT, RespT>) {
+        val requestID = headers.requestId.defaultToUUID()
+        val invocationID = headers.subRequestId.defaultToUUID()
+        val partnerName = headers.originatorId.defaultToUUID()
+        grpcRequesting(requestID, invocationID, partnerName, call)
+    }
+
+    fun <ReqT : Any, RespT : Any> grpcRequesting(requestID: String, invocationID: String, partnerName: String,
+                                                 call: ServerCall<ReqT, RespT>) {
+        val clientSocketAddress = call.attributes.get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR) as? InetSocketAddress
+                ?: throw BluePrintProcessorException("failed to get client address")
+        val serviceName = call.methodDescriptor.fullMethodName
+
+        MDC.put("InvokeTimestamp", ZonedDateTime.now(ZoneOffset.UTC).format(DateTimeFormatter.ISO_INSTANT))
+        MDC.put("RequestID", requestID)
+        MDC.put("InvocationID", invocationID)
+        MDC.put("PartnerName", partnerName)
+        MDC.put("ClientIPAddress", clientSocketAddress.address.defaultToEmpty())
+        MDC.put("ServerFQDN", clientSocketAddress.address.hostName.defaultToEmpty())
+        MDC.put("ServiceName", serviceName)
+        log.trace("MDC Properties : ${MDC.getCopyOfContextMap()}")
+    }
+
+
+    /** Used before invoking any GRPC outbound request, Inbound Invocation ID is used as request Id
+     * for outbound Request, If invocation Id is missing then default Request Id will be generated.
+     */
+    fun grpcInvoking(requestHeader: Metadata) {
+        requestHeader.putStringKeyValue(ONAP_REQUEST_ID, MDC.get("InvocationID").defaultToUUID())
+        requestHeader.putStringKeyValue(ONAP_INVOCATION_ID, UUID.randomUUID().toString())
+        val partnerName = System.getProperty("APPNAME") ?: "BlueprintsProcessor"
+        requestHeader.putStringKeyValue(ONAP_PARTNER_NAME, partnerName)
+    }
+
+    /** Used when server returns response */
+    fun grpResponding(requestHeaders: Metadata, responseHeaders: Metadata) {
+        try {
+            responseHeaders.putStringKeyValue(ONAP_REQUEST_ID, MDC.get("RequestID").defaultToEmpty())
+            responseHeaders.putStringKeyValue(ONAP_INVOCATION_ID, MDC.get("InvocationID").defaultToEmpty())
+            responseHeaders.putStringKeyValue(ONAP_PARTNER_NAME, MDC.get("PartnerName").defaultToEmpty())
+        } catch (e: Exception) {
+            log.warn("couldn't set grpc response headers", e)
+        }
+    }
+}
\ No newline at end of file
diff --git a/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/TokenAuthGrpcClientService.kt b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/TokenAuthGrpcClientService.kt
index dbff84211..601dc0e33 100644
--- a/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/TokenAuthGrpcClientService.kt
+++ b/ms/blueprintsprocessor/modules/commons/grpc-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/grpc/service/TokenAuthGrpcClientService.kt
@@ -21,6 +21,7 @@ import io.grpc.internal.DnsNameResolverProvider
 import io.grpc.internal.PickFirstLoadBalancerProvider
 import io.grpc.netty.NettyChannelBuilder
 import org.onap.ccsdk.cds.blueprintsprocessor.grpc.TokenAuthGrpcClientProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.interceptor.GrpcClientLoggingInterceptor
 
 class TokenAuthGrpcClientService(private val tokenAuthGrpcClientProperties: TokenAuthGrpcClientProperties)
     : BluePrintGrpcClientService {
@@ -30,6 +31,7 @@ class TokenAuthGrpcClientService(private val tokenAuthGrpcClientProperties: Toke
                 .forAddress(tokenAuthGrpcClientProperties.host, tokenAuthGrpcClientProperties.port)
                 .nameResolverFactory(DnsNameResolverProvider())
                 .loadBalancerFactory(PickFirstLoadBalancerProvider())
+                .intercept(GrpcClientLoggingInterceptor())
                 .intercept(TokenAuthClientInterceptor(tokenAuthGrpcClientProperties)).usePlaintext().build()
         return managedChannel
     }