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;
* 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);
}
}
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;
}
@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));
}
}
}
private Mono<JsonArray> fetchHttpData(String consulUrl) {
- return httpClient.callHttpGet(consulUrl, JsonArray.class);
+ return httpClient.get(consulUrl, JsonArray.class);
}
private Mono<JsonObject> firstService(JsonArray services) {
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;
private final HttpClient httpClient;
public CloudHttpClient() {
- this(HttpClient.create()
- .doOnRequest(CloudHttpClient::logRequest)
- .doOnResponse(CloudHttpClient::logResponse));
+ this(HttpClient.create());
}
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()
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());
+ });
}
-
}
+
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()));
}
private Mono<JsonObject> callConfigBindingServiceForConfiguration(String configBindingServiceUri) {
LOGGER.info("Retrieving configuration");
- return cloudHttpClient.callHttpGet(configBindingServiceUri, JsonObject.class);
+ return cloudHttpClient.get(configBindingServiceUri, JsonObject.class);
}
import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.DummyHttpServer.sendString;
import com.google.gson.JsonObject;
+import io.vavr.collection.Stream;
import java.time.Duration;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.EnvProperties;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.ImmutableEnvProperties;
+import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
*/
class CbsClientImplIT {
- private static final String CONSUL_RESP = "[\n"
+ private static final String CONSUL_RESPONSE = "[\n"
+ " {\n"
+ " \"ServiceAddress\": \"HOST\",\n"
+ " \"ServiceName\": \"the_cbs\",\n"
+ " \"ServicePort\": PORT\n"
+ " }\n"
+ "]\n";
- private static final String RES_CONFIG = "/sample_config.json";
+ private static final String SAMPLE_CONFIG = "/sample_config.json";
+ private static final String SAMPLE_CONFIG_KEY = "keystore.path";
+ private static final String EXPECTED_CONFIG_VALUE = "/var/run/security/keystore.p12";
+ private static EnvProperties sampleEnvironment;
private static DummyHttpServer server;
@BeforeAll
static void setUp() {
server = DummyHttpServer.start(routes ->
routes.get("/v1/catalog/service/the_cbs", (req, resp) -> sendString(resp, lazyConsulResponse()))
- .get("/service_component/dcae-component", (req, resp) -> sendResource(resp, RES_CONFIG)));
+ .get("/service_component/dcae-component", (req, resp) -> sendResource(resp, SAMPLE_CONFIG)));
+ sampleEnvironment = ImmutableEnvProperties.builder()
+ .appName("dcae-component")
+ .cbsName("the_cbs")
+ .consulHost(server.host())
+ .consulPort(server.port())
+ .build();
}
@AfterAll
}
@Test
- void testCbsClient() {
+ void testCbsClientWithSingleCall() {
// given
- final EnvProperties env = ImmutableEnvProperties.builder()
- .appName("dcae-component")
- .cbsName("the_cbs")
- .consulHost(server.host())
- .consulPort(server.port())
- .build();
- final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(env);
+ final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
+ final RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
+
+ // when
+ final Mono<JsonObject> result = sut.flatMap(cbsClient -> cbsClient.get(diagnosticContext));
+
+ // then
+ StepVerifier.create(result.map(this::sampleConfigValue))
+ .expectNext(EXPECTED_CONFIG_VALUE)
+ .expectComplete()
+ .verify(Duration.ofSeconds(5));
+ }
+
+ @Test
+ void testCbsClientWithPeriodicCall() {
+ // given
+ final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
+ final RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
// when
- final Mono<JsonObject> result = sut.flatMap(CbsClient::get);
+ final Flux<JsonObject> result = sut.flatMapMany(cbsClient -> cbsClient.get(diagnosticContext, Duration.ZERO, Duration.ofMillis(10)));
// then
- StepVerifier.create(result.map(obj -> obj.get("keystore.path").getAsString()))
- .expectNext("/var/run/security/keystore.p12")
+ final int itemsToTake = 5;
+ StepVerifier.create(result.take(itemsToTake).map(this::sampleConfigValue))
+ .expectNextSequence(Stream.of(EXPECTED_CONFIG_VALUE).cycle(itemsToTake))
.expectComplete()
.verify(Duration.ofSeconds(5));
}
+ private String sampleConfigValue(JsonObject obj) {
+ return obj.get(SAMPLE_CONFIG_KEY).getAsString();
+ }
+
private static Mono<String> lazyConsulResponse() {
- return Mono.just(CONSUL_RESP)
+ return Mono.just(CONSUL_RESPONSE)
.map(CbsClientImplIT::processConsulResponseTemplate);
}
import com.google.gson.JsonObject;
import java.net.InetSocketAddress;
import org.junit.jupiter.api.Test;
+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;
String serviceName = "dcaegen2-ves-collector";
final CbsClientImpl cut = CbsClientImpl.create(httpClient, cbsAddress, serviceName);
final JsonObject httpResponse = new JsonObject();
- given(httpClient.callHttpGet(anyString(), any(Class.class))).willReturn(Mono.just(httpResponse));
+ given(httpClient.get(anyString(), any(RequestDiagnosticContext.class), any(Class.class))).willReturn(Mono.just(httpResponse));
+ RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
// when
- final JsonObject result = cut.get().block();
+ final JsonObject result = cut.get(diagnosticContext).block();
// then
- verify(httpClient).callHttpGet("http://cbshost:6969/service_component/dcaegen2-ves-collector", JsonObject.class);
+ final String expectedUrl = "http://cbshost:6969/service_component/dcaegen2-ves-collector";
+ verify(httpClient).get(expectedUrl, diagnosticContext, JsonObject.class);
assertThat(result).isSameAs(httpResponse);
}
}
\ No newline at end of file
+ env.consulPort()
+ "/v1/catalog/service/"
+ env.cbsName();
- given(httpClient.callHttpGet(url, JsonArray.class))
+ given(httpClient.get(url, JsonArray.class))
.willReturn(Mono.just(jsonArray));
}
// given
CloudHttpClient webClient = mock(CloudHttpClient.class);
when(
- webClient.callHttpGet("http://consul:8500/v1/catalog/service/config-binding-service", JsonArray.class))
+ webClient.get("http://consul:8500/v1/catalog/service/config-binding-service", JsonArray.class))
.thenReturn(Mono.just(configBindingServiceJson));
- when(webClient.callHttpGet("http://config-binding-service:10000/service_component/dcae-prh", JsonObject.class))
+ when(webClient.get("http://config-binding-service:10000/service_component/dcae-prh", JsonObject.class))
.thenReturn(Mono.just(configurationJsonMock));
ReactiveCloudConfigurationProvider provider = new ReactiveCloudConfigurationProvider(webClient);
// given
CloudHttpClient webClient = mock(CloudHttpClient.class);
when(
- webClient.callHttpGet("http://consul:8500/v1/catalog/service/config-binding-service", JsonArray.class))
+ webClient.get("http://consul:8500/v1/catalog/service/config-binding-service", JsonArray.class))
.thenReturn(Mono.just(emptyConfigBindingServiceJson));
ReactiveCloudConfigurationProvider provider = new ReactiveCloudConfigurationProvider(webClient);
~ ============LICENSE_END=========================================================
-->
<configuration>
- <property name="p_tim" value="%date{"yyyy-MM-dd'T'HH:mm:ss.SSSXXX", UTC}"/>
- <property name="p_lvl" value="%highlight(%-5level)"/>
- <property name="p_log" value="%50.50logger"/>
- <property name="SIMPLE_LOG_PATTERN" value="
+ <property name="p_log" value="%logger"/>
+ <property name="p_lor" value="%50.50logger"/>
+ <property name="p_tim" value="%date{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC}"/>
+ <property name="p_lvl" value="%level"/>
+ <property name="p_msg" value="%replace(%replace(%msg){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_mdc" value="%replace(%replace(%mdc){'\t', '\\\\t'}){'\n', '\\\\n'}"/>
+ <property name="p_exc" value="%replace(%replace(%rootException){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_mak" value="%replace(%replace(%marker){'\t', '\\\\t'}){'\n','\\\\n'}"/>
+ <property name="p_thr" value="%thread"/>
+
+ <property name="READABLE_LOG_PATTERN" value="
%nopexception
| ${p_tim}\t
-| ${p_log}\t
+| ${p_lor}\t
| ${p_lvl}\t
-| %msg%n"/>
+| %msg\t
+| ${p_mak}\t
+| %rootException\t
+| ${p_mdc}\t
+| ${p_thr}%n"/>
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
- <pattern>${SIMPLE_LOG_PATTERN}</pattern>
+ <pattern>${READABLE_LOG_PATTERN}</pattern>
</encoder>
</appender>
<groupId>org.immutables</groupId>
<artifactId>value</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.vavr</groupId>
+ <artifactId>vavr</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.jetbrains</groupId>
+ <artifactId>annotations</artifactId>
+ </dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * 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.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.rest.services.model.logging;
+
+import io.vavr.collection.HashMap;
+import io.vavr.collection.Map;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.UUID;
+import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
+import org.slf4j.MDC;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.2
+ */
+@Value.Immutable(singleton = true)
+public interface GlobalDiagnosticContext {
+
+ @Value.Default
+ default String instanceId() {
+ return UUID.randomUUID().toString();
+ }
+
+ @Value.Default
+ default String serverFqdn() {
+ try {
+ return InetAddress.getLocalHost().toString();
+ } catch (UnknownHostException ex) {
+ return InetAddress.getLoopbackAddress().toString();
+ }
+ }
+
+ @Value.Default
+ default String serviceName() {
+ return System.getenv().getOrDefault("HOSTNAME", "unknown_service");
+ }
+
+ @Value.Derived
+ default Map<String, String> asMap() {
+ return HashMap.of(
+ MdcVariables.INSTANCE_ID, instanceId(),
+ MdcVariables.SERVER_FQDN, serverFqdn(),
+ MdcVariables.SERVICE_NAME, serviceName());
+ }
+
+ static GlobalDiagnosticContext instance() {
+ return ImmutableGlobalDiagnosticContext.of();
+ }
+}
+
+
public final class MdcVariables {
+ @Deprecated
public static final String X_ONAP_REQUEST_ID = "X-ONAP-RequestID";
+ @Deprecated
public static final String X_INVOCATION_ID = "X-InvocationID";
- public static final String REQUEST_ID = "RequestID";
- public static final String INVOCATION_ID = "InvocationID";
+
public static final String INSTANCE_UUID = "InstanceUUID";
public static final String RESPONSE_CODE = "ResponseCode";
+ public static final String REQUEST_ID = "RequestID";
+ public static final String CLIENT_NAME = "PartnerName";
+ public static final String CLIENT_IP = "ClientIPAddress";
+ public static final String INVOCATION_ID = "InvocationID";
+ public static final String INVOCATION_TIMESTAMP = "InvokeTimestamp";
+ public static final String STATUS_CODE = "StatusCode";
+ public static final String INSTANCE_ID = "InstanceID";
+ public static final String SERVER_FQDN = "ServerFQDN";
public static final String SERVICE_NAME = "ServiceName";
+ private static final String HTTP_HEADER_PREFIX = "X-";
+
private MdcVariables() {
}
+ public static String httpHeader(String mdcName) {
+ return HTTP_HEADER_PREFIX + mdcName;
+ }
+
+ /**
+ * @deprecated use {@link RequestDiagnosticContext#withSlf4jMdc(Runnable)}.
+ * @param mdcContextMap
+ */
+ @Deprecated
public static void setMdcContextMap(Map<String, String> mdcContextMap) {
if (mdcContextMap != null) {
MDC.setContextMap(mdcContextMap);
--- /dev/null
+/*
+ * ============LICENSE_START====================================
+ * DCAEGEN2-SERVICES-SDK
+ * =========================================================
+ * Copyright (C) 2019 Nokia. All rights reserved.
+ * =========================================================
+ * 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.
+ * ============LICENSE_END=====================================
+ */
+
+package org.onap.dcaegen2.services.sdk.rest.services.model.logging;
+
+import io.vavr.collection.HashMap;
+import io.vavr.collection.Map;
+import java.util.UUID;
+import org.immutables.value.Value;
+import org.jetbrains.annotations.Nullable;
+import org.slf4j.MDC;
+
+/**
+ * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
+ * @since 1.1.2
+ */
+@Value.Immutable
+public interface RequestDiagnosticContext {
+
+ UUID requestId();
+
+ @Nullable UUID invocationId();
+
+ @Value.Default
+ default GlobalDiagnosticContext global() {
+ return GlobalDiagnosticContext.instance();
+ }
+
+ @Value.Derived
+ default Map<String, String> remoteCallHttpHeaders() {
+ java.util.Map<String, String> result = new java.util.HashMap<>();
+
+ result.put(MdcVariables.httpHeader(MdcVariables.REQUEST_ID), requestId().toString());
+
+ if (invocationId() != null) {
+ result.put(MdcVariables.httpHeader(MdcVariables.INVOCATION_ID), invocationId().toString());
+ }
+
+ return HashMap.ofAll(result);
+ }
+
+ @Value.Derived
+ default Map<String, String> asMap() {
+ java.util.Map<String, String> result = new java.util.HashMap<>();
+
+ if (requestId() != null) {
+ result.put(MdcVariables.REQUEST_ID, requestId().toString());
+ }
+
+ if (invocationId() != null) {
+ result.put(MdcVariables.INVOCATION_ID, invocationId().toString());
+ }
+
+ return global().asMap().merge(HashMap.ofAll(result));
+ }
+
+ default void withSlf4jMdc(Runnable runnable) {
+ withSlf4jMdc(true, runnable);
+ }
+
+ default void withSlf4jMdc(boolean loglevelEnabled, Runnable runnable) {
+ if (loglevelEnabled) {
+ final java.util.Map<String, String> ctxBefore = MDC.getCopyOfContextMap();
+ try {
+ MDC.setContextMap(asMap().toJavaMap());
+ runnable.run();
+ } finally {
+ if (ctxBefore == null) {
+ MDC.clear();
+ } else {
+ MDC.setContextMap(ctxBefore);
+ }
+ }
+ }
+ }
+
+ static ImmutableRequestDiagnosticContext create() {
+ return ImmutableRequestDiagnosticContext.builder()
+ .requestId(UUID.randomUUID())
+ .invocationId(UUID.randomUUID())
+ .build();
+ }
+}
+
+
Code Review / dcaegen2 / services / sdk.git / commitdiff