Add MDC support to CBS client

Change-Id: Id1f1f9016b03658eca0afd0bd3bd724afc0bea96
Issue-ID: DCAEGEN2-1233
Signed-off-by: Piotr Jaszczyk <piotr.jaszczyk@nokia.com>

5 files changed, 46 insertions, 32 deletions

diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClient.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClient.java
index 73789268..b9a6e40d 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClient.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/api/CbsClient.java
@@ -22,6 +22,9 @@ package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api;
 import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import java.time.Duration;
+import java.util.UUID;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.ImmutableRequestDiagnosticContext;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import org.jetbrains.annotations.NotNull;
@@ -41,23 +44,24 @@ public interface CbsClient {
      * Returns a {@link Mono} that publishes new configuration after CBS client retrieves one.
      *
      * @return reactive stream of configuration
+     * @param diagnosticContext diagnostic context as defined in Logging Guideline
      * @since 1.1.2
      */
-    @NotNull Mono<JsonObject> get();
-
+    @NotNull Mono<JsonObject> get(RequestDiagnosticContext diagnosticContext);
 
     /**
      * Poll for configuration.
      *
-     * Will call {@link #get()} after {@code initialDelay} every {@code period}. Resulting entries may or may not be
+     * Will call {@link #get(RequestDiagnosticContext)} after {@code initialDelay} every {@code period}. Resulting entries may or may not be
      * changed, ie. items in the stream might be the same until change is made in CBS.
      *
      * @param initialDelay delay after first request attempt
      * @param period frequency of update checks
      * @return stream of configuration states
      */
-    default Flux<JsonElement> get(Duration initialDelay, Duration period) {
+    default Flux<JsonObject> get(RequestDiagnosticContext diagnosticContext, Duration initialDelay, Duration period) {
         return Flux.interval(initialDelay, period)
-                .flatMap(i -> get());
+                .map(i -> ImmutableRequestDiagnosticContext.copyOf(diagnosticContext).withInvocationId(UUID.randomUUID()))
+                .flatMap(this::get);
     }
 }
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImpl.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImpl.java
index 0d32320a..d11be24b 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImpl.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsClientImpl.java
@@ -25,6 +25,7 @@ import java.net.MalformedURLException;
 import java.net.URL;
 import org.jetbrains.annotations.NotNull;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.adapters.CloudHttpClient;
 import reactor.core.publisher.Mono;
 
@@ -55,7 +56,7 @@ public class CbsClientImpl implements CbsClient {
     }
 
     @Override
-    public @NotNull Mono<JsonObject> get() {
-        return Mono.defer(() -> httpClient.callHttpGet(fetchUrl, JsonObject.class));
+    public @NotNull Mono<JsonObject> get(RequestDiagnosticContext diagnosticContext) {
+        return Mono.defer(() -> httpClient.get(fetchUrl, diagnosticContext, JsonObject.class));
     }
 }
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsLookup.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsLookup.java
index ca7058f6..f5ec462d 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsLookup.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/CbsLookup.java
@@ -53,7 +53,7 @@ public class CbsLookup {
     }
 
     private Mono<JsonArray> fetchHttpData(String consulUrl) {
-        return httpClient.callHttpGet(consulUrl, JsonArray.class);
+        return httpClient.get(consulUrl, JsonArray.class);
     }
 
     private Mono<JsonObject> firstService(JsonArray services) {
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/adapters/CloudHttpClient.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/adapters/CloudHttpClient.java
index 438ff667..13347e10 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/adapters/CloudHttpClient.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/impl/adapters/CloudHttpClient.java
@@ -21,16 +21,12 @@
 package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.adapters;
 
 import com.google.gson.Gson;
-import com.google.gson.JsonSyntaxException;
 import io.netty.handler.codec.http.HttpStatusClass;
 import io.vavr.collection.Stream;
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.io.Reader;
-import java.net.URL;
 import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
@@ -51,9 +47,7 @@ public class CloudHttpClient {
     private final HttpClient httpClient;
 
     public CloudHttpClient() {
-        this(HttpClient.create()
-                .doOnRequest(CloudHttpClient::logRequest)
-                .doOnResponse(CloudHttpClient::logResponse));
+        this(HttpClient.create());
     }
 
 
@@ -61,9 +55,20 @@ public class CloudHttpClient {
         this.httpClient = httpClient;
     }
 
-    public <T> Mono<T> callHttpGet(String url, Class<T> bodyClass) {
-        return httpClient
-                .get()
+    public <T> Mono<T> get(String url, RequestDiagnosticContext context, Class<T> bodyClass) {
+        final HttpClient clientWithHeaders = httpClient
+                .doOnRequest((req, conn) -> logRequest(context, req))
+                .doOnResponse((rsp, conn) -> logResponse(context, rsp))
+                .headers(hdrs -> context.remoteCallHttpHeaders().forEach((BiConsumer<String, String>) hdrs::set));
+        return callHttpGet(clientWithHeaders, url, bodyClass);
+    }
+
+    public <T> Mono<T> get(String url, Class<T> bodyClass) {
+        return callHttpGet(httpClient, url, bodyClass);
+    }
+
+    private <T> Mono<T> callHttpGet(HttpClient client, String url, Class<T> bodyClass) {
+        return client.get()
                 .uri(url)
                 .responseSingle((resp, content) -> HttpStatusClass.SUCCESS.contains(resp.status().code())
                         ? content.asString()
@@ -81,18 +86,22 @@ public class CloudHttpClient {
         return gson.fromJson(body, bodyClass);
     }
 
-    private static void logRequest(HttpClientRequest httpClientRequest, Connection connection) {
-        LOGGER.debug("Request: {} {}", httpClientRequest.method(), httpClientRequest.uri());
-        if (LOGGER.isTraceEnabled()) {
-            final String headers = Stream.ofAll(httpClientRequest.requestHeaders())
-                    .map(entry -> entry.getKey() + "=" + entry.getValue())
-                    .collect(Collectors.joining("\n"));
-            LOGGER.trace(headers);
-        }
+    private void logRequest(RequestDiagnosticContext context, HttpClientRequest httpClientRequest) {
+        context.withSlf4jMdc(LOGGER.isDebugEnabled(), () -> {
+            LOGGER.debug("Request: {} {}", httpClientRequest.method(), httpClientRequest.uri());
+            if (LOGGER.isTraceEnabled()) {
+                final String headers = Stream.ofAll(httpClientRequest.requestHeaders())
+                        .map(entry -> entry.getKey() + "=" + entry.getValue())
+                        .collect(Collectors.joining("\n"));
+                LOGGER.trace(headers);
+            }
+        });
     }
 
-    private static void logResponse(HttpClientResponse httpClientResponse, Connection connection) {
-        LOGGER.debug("Response status: {}", httpClientResponse.status());
+    private void logResponse(RequestDiagnosticContext context, HttpClientResponse httpClientResponse) {
+        context.withSlf4jMdc(LOGGER.isDebugEnabled(), () -> {
+            LOGGER.debug("Response status: {}", httpClientResponse.status());
+        });
     }
-
 }
+
diff --git a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/ReactiveCloudConfigurationProvider.java b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/ReactiveCloudConfigurationProvider.java
index 5606a2d1..02e9b9cc 100644
--- a/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/ReactiveCloudConfigurationProvider.java
+++ b/rest-services/cbs-client/src/main/java/org/onap/dcaegen2/services/sdk/rest/services/cbs/client/providers/ReactiveCloudConfigurationProvider.java
@@ -73,7 +73,7 @@ public final class ReactiveCloudConfigurationProvider implements CloudConfigurat
 
     private Mono<String> callConsulForConfigBindingServiceEndpoint(EnvProperties envProperties) {
         LOGGER.info("Retrieving Config Binding Service endpoint from Consul");
-        return cloudHttpClient.callHttpGet(getConsulUrl(envProperties), JsonArray.class)
+        return cloudHttpClient.get(getConsulUrl(envProperties), JsonArray.class)
             .flatMap(jsonArray -> this.createConfigBindingServiceUrl(jsonArray, envProperties.appName()));
     }
 
@@ -84,7 +84,7 @@ public final class ReactiveCloudConfigurationProvider implements CloudConfigurat
 
     private Mono<JsonObject> callConfigBindingServiceForConfiguration(String configBindingServiceUri) {
         LOGGER.info("Retrieving configuration");
-        return cloudHttpClient.callHttpGet(configBindingServiceUri, JsonObject.class);
+        return cloudHttpClient.get(configBindingServiceUri, JsonObject.class);
     }