From 290aa864c6a7d02a4f310c5d1cf8106e02e9dd47 Mon Sep 17 00:00:00 2001 From: JoeOLeary Date: Wed, 13 Mar 2019 16:47:20 +0000 Subject: Add Dynamic Configuration *Implement Configurable in DataRouter subscriber *Add reconfigure.sh *Update Dockerfile to use reconfigure.sh Issue-ID: DCAEGEN2-1038 Change-Id: I9ce387819afff644c3dc76a94786fa1f95afd985 Signed-off-by: JoeOLeary --- .../org/onap/dcaegen2/services/pmmapper/App.java | 19 ++- .../services/pmmapper/config/Configurable.java | 31 ++++ .../pmmapper/config/DynamicConfiguration.java | 90 +++++++++++ .../pmmapper/datarouter/DataRouterSubscriber.java | 102 ++++++++++--- .../exceptions/ReconfigurationException.java | 30 ++++ .../services/pmmapper/model/MapperConfig.java | 9 ++ src/main/resources/Dockerfile | 3 +- src/main/resources/reconfigure.sh | 6 + .../pmmapper/config/DynamicConfigurationTest.java | 153 +++++++++++++++++++ .../datarouter/DataRouterSubscriberTest.java | 168 +++++++++++++++++++-- .../valid_bc_response.json | 15 ++ 11 files changed, 586 insertions(+), 40 deletions(-) create mode 100644 src/main/java/org/onap/dcaegen2/services/pmmapper/config/Configurable.java create mode 100644 src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java create mode 100644 src/main/java/org/onap/dcaegen2/services/pmmapper/exceptions/ReconfigurationException.java create mode 100644 src/main/resources/reconfigure.sh create mode 100644 src/test/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfigurationTest.java create mode 100644 src/test/resources/datarouter_subscriber_test/valid_bc_response.json (limited to 'src') diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java index 09d8975..6ebc61c 100644 --- a/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java +++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java @@ -26,6 +26,8 @@ import io.undertow.util.StatusCodes; import lombok.NonNull; import org.onap.dcaegen2.services.pmmapper.config.ConfigHandler; +import org.onap.dcaegen2.services.pmmapper.config.Configurable; +import org.onap.dcaegen2.services.pmmapper.config.DynamicConfiguration; import org.onap.dcaegen2.services.pmmapper.datarouter.DataRouterSubscriber; import org.onap.dcaegen2.services.pmmapper.exceptions.CBSConfigException; import org.onap.dcaegen2.services.pmmapper.exceptions.CBSServerError; @@ -47,6 +49,7 @@ import reactor.core.scheduler.Schedulers; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.ArrayList; public class App { private static final ONAPLogAdapter logger = new ONAPLogAdapter(LoggerFactory.getLogger(App.class)); @@ -57,9 +60,7 @@ public class App { public static void main(String[] args) throws InterruptedException, TooManyTriesException, CBSConfigException, EnvironmentConfigException, CBSServerError, MapperConfigException { Flux flux = Flux.create(eventFluxSink -> fluxSink = eventFluxSink); HealthCheckHandler healthCheckHandler = new HealthCheckHandler(); - MapperConfig mapperConfig = new ConfigHandler().getMapperConfig(); - MetadataFilter metadataFilter = new MetadataFilter(mapperConfig); Mapper mapper = new Mapper(mappingTemplate); XMLValidator validator = new XMLValidator(xmlSchema); @@ -73,14 +74,18 @@ public class App { .filter(validator::validate) .map(mapper::map) .subscribe(event -> logger.unwrap().info("Event Processed")); - - DataRouterSubscriber dataRouterSubscriber = new DataRouterSubscriber(fluxSink::next); - dataRouterSubscriber.start(mapperConfig); + DataRouterSubscriber dataRouterSubscriber = new DataRouterSubscriber(fluxSink::next, mapperConfig); + dataRouterSubscriber.start(); + ArrayList configurables = new ArrayList<>(); + configurables.add(dataRouterSubscriber); + DynamicConfiguration dynamicConfiguration = new DynamicConfiguration(configurables, mapperConfig); Undertow.builder() .addHttpListener(8081, "0.0.0.0") - .setHandler(Handlers.routing().add("put", "/delivery/{filename}", dataRouterSubscriber) - .add("get", "/healthcheck", healthCheckHandler)) + .setHandler(Handlers.routing() + .add("put", "/delivery/{filename}", dataRouterSubscriber) + .add("get", "/healthcheck", healthCheckHandler) + .add("get", "/reconfigure", dynamicConfiguration)) .build().start(); } diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/Configurable.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/Configurable.java new file mode 100644 index 0000000..ac2fe57 --- /dev/null +++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/Configurable.java @@ -0,0 +1,31 @@ +/* + * - + * * ============LICENSE_START======================================================= + * * Copyright (C) 2019 Nordix Foundation. + * * ================================================================================ + * * 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. + * * + * * SPDX-License-Identifier: Apache-2.0 + * * ============LICENSE_END========================================================= + * + */ + +package org.onap.dcaegen2.services.pmmapper.config; + +import org.onap.dcaegen2.services.pmmapper.exceptions.ReconfigurationException; +import org.onap.dcaegen2.services.pmmapper.model.MapperConfig; + +public interface Configurable { + + void reconfigure(MapperConfig mapperConfig) throws ReconfigurationException; +} diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java new file mode 100644 index 0000000..37fa8b5 --- /dev/null +++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfiguration.java @@ -0,0 +1,90 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 Nordix Foundation. + * ================================================================================ + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.dcaegen2.services.pmmapper.config; + +import io.undertow.server.HttpHandler; +import io.undertow.server.HttpServerExchange; +import io.undertow.util.StatusCodes; +import lombok.Data; +import org.onap.dcaegen2.services.pmmapper.exceptions.ReconfigurationException; +import org.onap.dcaegen2.services.pmmapper.model.MapperConfig; +import org.onap.dcaegen2.services.pmmapper.utils.HttpServerExchangeAdapter; +import org.onap.logging.ref.slf4j.ONAPLogAdapter; +import org.slf4j.LoggerFactory; + +import java.util.List; + +@Data +public class DynamicConfiguration implements HttpHandler{ + private static final ONAPLogAdapter logger = new ONAPLogAdapter(LoggerFactory.getLogger(DynamicConfiguration.class)); + private List configurables; + private MapperConfig originalConfig; + private ConfigHandler configHandler; + + /** + * Creates a Dynamic Configuration object with a list of configurable objects. + * @param configurables list of objects to reconfigure + * @param originalConfig original config to compare against. + */ + public DynamicConfiguration(List configurables, MapperConfig originalConfig){ + this.configurables = configurables; + this.originalConfig = originalConfig; + this.configHandler = new ConfigHandler(); + } + + private void applyConfiguration(MapperConfig updatedConfig) throws ReconfigurationException { + for (Configurable configurable : configurables) { + logger.unwrap().debug("Reconfiguring: {}", configurable); + configurable.reconfigure(updatedConfig); + } + } + + /** + * Receives requests to pull the latest configuration from CBS. + * @param httpServerExchange inbound http server exchange. + * @throws Exception + */ + @Override + public void handleRequest(HttpServerExchange httpServerExchange) throws Exception { + try { + logger.entering(new HttpServerExchangeAdapter(httpServerExchange)); + MapperConfig config = configHandler.getMapperConfig(); + int responseCode = StatusCodes.OK; + String responseMessage = StatusCodes.OK_STRING; + + if (!this.originalConfig.equals(config)) { + logger.unwrap().info("Configuration update detected."); + logger.unwrap().info("Reconfiguring configurables"); + try { + applyConfiguration(config); + this.originalConfig = config; + } catch (ReconfigurationException e) { + responseCode = StatusCodes.INTERNAL_SERVER_ERROR; + responseMessage = StatusCodes.INTERNAL_SERVER_ERROR_STRING; + applyConfiguration(this.originalConfig); + } + } + httpServerExchange.setStatusCode(responseCode).getResponseSender().send(responseMessage); + } finally { + logger.exiting(); + } + } +} diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/datarouter/DataRouterSubscriber.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/datarouter/DataRouterSubscriber.java index 2f2ab4d..f37bcd3 100644 --- a/src/main/java/org/onap/dcaegen2/services/pmmapper/datarouter/DataRouterSubscriber.java +++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/datarouter/DataRouterSubscriber.java @@ -24,11 +24,15 @@ import com.google.gson.Gson; import com.google.gson.GsonBuilder; import com.google.gson.JsonObject; import com.google.gson.JsonParseException; +import com.google.gson.JsonParser; +import com.google.gson.JsonSyntaxException; import io.undertow.util.HeaderValues; import lombok.Data; import lombok.NonNull; +import org.onap.dcaegen2.services.pmmapper.config.Configurable; import org.onap.dcaegen2.services.pmmapper.exceptions.NoMetadataException; +import org.onap.dcaegen2.services.pmmapper.exceptions.ReconfigurationException; import org.onap.dcaegen2.services.pmmapper.exceptions.TooManyTriesException; import org.onap.dcaegen2.services.pmmapper.model.EventMetadata; import org.onap.dcaegen2.services.pmmapper.model.MapperConfig; @@ -44,23 +48,28 @@ import org.onap.logging.ref.slf4j.ONAPLogConstants; import org.slf4j.LoggerFactory; import org.slf4j.MDC; +import java.io.BufferedReader; import java.io.IOException; +import java.io.InputStreamReader; import java.io.OutputStream; import java.io.OutputStreamWriter; import java.net.HttpURLConnection; +import java.net.MalformedURLException; +import java.net.URL; import java.nio.charset.StandardCharsets; import java.time.Instant; import java.util.Map; import java.util.Optional; import java.util.Random; import java.util.UUID; +import java.util.stream.Collectors; /** * Subscriber for events sent from data router * Provides an undertow HttpHandler to be used as an endpoint for data router to send events to. */ @Data -public class DataRouterSubscriber implements HttpHandler { +public class DataRouterSubscriber implements HttpHandler, Configurable { public static final String METADATA_HEADER = "X-DMAAP-DR-META"; public static final String PUB_ID_HEADER = "X-DMAAP-DR-PUBLISH-ID"; @@ -75,40 +84,41 @@ public class DataRouterSubscriber implements HttpHandler { private boolean limited = false; private Random jitterGenerator; private Gson metadataBuilder; + private MapperConfig config; + private String subscriberId; @NonNull private EventReceiver eventReceiver; /** * @param eventReceiver receiver for any inbound events. */ - public DataRouterSubscriber(EventReceiver eventReceiver) { + public DataRouterSubscriber(EventReceiver eventReceiver, MapperConfig config) { this.eventReceiver = eventReceiver; this.jitterGenerator = new Random(); this.metadataBuilder = new GsonBuilder().registerTypeAdapter(EventMetadata.class, new RequiredFieldDeserializer()) .create(); + this.config = config; + this.subscriberId=""; } /** * Starts data flow by subscribing to data router through bus controller. * - * @param config configuration object containing bus controller endpoint for subscription and - * all non constant configuration for subscription through this endpoint. * @throws TooManyTriesException in the event that timeout has occurred several times. */ - public void start(MapperConfig config) throws TooManyTriesException, InterruptedException { + public void start() throws TooManyTriesException, InterruptedException { try { logger.unwrap().info("Starting subscription to DataRouter {}", ONAPLogConstants.Markers.ENTRY); - subscribe(NUMBER_OF_ATTEMPTS, DEFAULT_TIMEOUT, config); + subscribe(); logger.unwrap().info("Successfully started DR Subscriber"); } finally { logger.unwrap().info("{}", ONAPLogConstants.Markers.EXIT); } } - private HttpURLConnection getBusControllerConnection(MapperConfig config, int timeout) throws IOException { - HttpURLConnection connection = (HttpURLConnection) config.getBusControllerSubscriptionUrl() - .openConnection(); - connection.setRequestMethod("POST"); + private HttpURLConnection getBusControllerConnection(String method, URL resource, int timeout) throws IOException { + HttpURLConnection connection = (HttpURLConnection) resource.openConnection(); + connection.setRequestMethod(method); connection.setConnectTimeout(timeout); connection.setReadTimeout(timeout); connection.setRequestProperty("Content-Type", "application/json"); @@ -135,26 +145,66 @@ public class DataRouterSubscriber implements HttpHandler { return subscriberObj; } - private void subscribe(int attempts, int timeout, MapperConfig config) throws TooManyTriesException, InterruptedException { + private void processResponse(HttpURLConnection connection) throws IOException { + try (BufferedReader responseBody = new BufferedReader(new InputStreamReader(connection.getInputStream()))) { + String body = responseBody.lines().collect(Collectors.joining("")); + updateSubscriberId(body); + } catch (IOException | JsonSyntaxException | IllegalStateException e) { + throw new IOException("Failed to process response", e); + } + } + + private void updateSubscriberId(String responseBody) { + JsonParser parser = new JsonParser(); + JsonObject responseObject = parser.parse(responseBody).getAsJsonObject(); + this.subscriberId = responseObject.get("subId").getAsString(); + } + + private void subscribe() throws TooManyTriesException, InterruptedException { + try { + URL subscribeResource = this.config.getBusControllerSubscriptionUrl(); + JsonObject subscribeBody = this.getBusControllerSubscribeBody(this.config); + request(NUMBER_OF_ATTEMPTS, DEFAULT_TIMEOUT, "POST", subscribeResource, subscribeBody); + } catch (MalformedURLException e) { + throw new IllegalStateException("Subscription URL is malformed", e); + } + + } + private void updateSubscriber() throws TooManyTriesException, InterruptedException { + try { + URL subscribeResource = this.config.getBusControllerSubscriptionUrl(); + URL updateResource = new URL(String.format("%s/%s", subscribeResource, subscriberId)); + JsonObject subscribeBody = this.getBusControllerSubscribeBody(this.config); + request(NUMBER_OF_ATTEMPTS, DEFAULT_TIMEOUT, "PUT", updateResource, subscribeBody); + } catch (MalformedURLException e) { + throw new IllegalStateException("Subscription URL is malformed", e); + } + } + + private void request(int attempts, int timeout, String method, URL resource, JsonObject subscribeBody) throws TooManyTriesException, InterruptedException { int subResponse = 504; String subMessage = ""; + boolean processFailure = false; try { - HttpURLConnection connection = getBusControllerConnection(config, timeout); - + HttpURLConnection connection = getBusControllerConnection(method, resource, timeout); try (OutputStream bodyStream = connection.getOutputStream(); OutputStreamWriter bodyWriter = new OutputStreamWriter(bodyStream, StandardCharsets.UTF_8)) { - bodyWriter.write(getBusControllerSubscribeBody(config).toString()); + bodyWriter.write(subscribeBody.toString()); } subResponse = connection.getResponseCode(); subMessage = connection.getResponseMessage(); + if (subResponse < 300) { + processResponse(connection); + } } catch (IOException e) { - logger.unwrap().error("Timeout Failure:", e); + logger.unwrap().error("Failure to process response", e); + processFailure = true; } logger.unwrap().info("Request to bus controller executed with Response Code: '{}' and Response Event: '{}'.", subResponse, subMessage); - if (subResponse >= 300 && attempts > 1) { + if ((subResponse >= 300 || processFailure) && attempts > 1 ) { Thread.sleep(timeout); - subscribe(--attempts, (timeout * 2) + jitterGenerator.nextInt(MAX_JITTER), config); - } else if (subResponse >= 300) { + request(--attempts, (timeout * 2) + jitterGenerator.nextInt(MAX_JITTER), method, resource, subscribeBody); + } else if (subResponse >= 300 || processFailure) { throw new TooManyTriesException("Failed to subscribe within appropriate amount of attempts"); } } @@ -209,4 +259,20 @@ public class DataRouterSubscriber implements HttpHandler { logger.exiting(); } } + + @Override + public void reconfigure(MapperConfig config) throws ReconfigurationException { + logger.unwrap().info("Checking new Configuration against existing."); + if(!this.config.dmaapInfoEquals(config) || !this.config.getDmaapDRFeedId().equals(config.getDmaapDRFeedId())){ + logger.unwrap().info("DMaaP Info changes found, reconfiguring."); + try { + this.config = config; + this.updateSubscriber(); + } catch (TooManyTriesException | InterruptedException e) { + Thread.currentThread().interrupt(); + throw new ReconfigurationException("Failed to reconfigure DataRouter subscriber.", e); + } + } + + } } diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/exceptions/ReconfigurationException.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/exceptions/ReconfigurationException.java new file mode 100644 index 0000000..66e3aee --- /dev/null +++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/exceptions/ReconfigurationException.java @@ -0,0 +1,30 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 Nordix Foundation. + * ================================================================================ + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.dcaegen2.services.pmmapper.exceptions; + +public class ReconfigurationException extends Exception { + public ReconfigurationException(String message) { + super(message); + } + public ReconfigurationException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/model/MapperConfig.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/model/MapperConfig.java index 2f13080..d28d850 100644 --- a/src/main/java/org/onap/dcaegen2/services/pmmapper/model/MapperConfig.java +++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/model/MapperConfig.java @@ -81,6 +81,15 @@ public class MapperConfig { public String getSubscriberIdentity(){ return this.getStreamsSubscribes().getDmaapSubscriber().getDmaapInfo().getSubscriberId(); } + + public boolean dmaapInfoEquals(MapperConfig mapperConfig){ + return this + .getStreamsSubscribes() + .getDmaapSubscriber() + .getDmaapInfo() + .equals(mapperConfig.getStreamsSubscribes().getDmaapSubscriber().getDmaapInfo()); + } + @Getter @EqualsAndHashCode private class StreamsSubscribes { diff --git a/src/main/resources/Dockerfile b/src/main/resources/Dockerfile index e2b6b89..2f4505c 100644 --- a/src/main/resources/Dockerfile +++ b/src/main/resources/Dockerfile @@ -19,7 +19,7 @@ # FROM openjdk:8-jre-alpine -ENTRYPOINT ["/usr/bin/java", "-jar", "/opt/app/pm-mapper/pm-mapper.jar"] +ENTRYPOINT ["/bin/sh", "-c", "nohup sh etc/reconfigure.sh & /usr/bin/java -jar /opt/app/pm-mapper/pm-mapper.jar"] ARG JAR WORKDIR /opt/app/pm-mapper @@ -32,5 +32,6 @@ RUN addgroup -S onap \ ADD target/${JAR} /opt/app/pm-mapper/pm-mapper.jar ADD target/classes/mapping.ftl /opt/app/pm-mapper/etc/mapping.ftl ADD target/classes/measCollec_plusString.xsd /opt/app/pm-mapper/etc/measCollec_plusString.xsd +ADD target/classes/reconfigure.sh /opt/app/pm-mapper/etc/reconfigure.sh USER pm-mapper diff --git a/src/main/resources/reconfigure.sh b/src/main/resources/reconfigure.sh new file mode 100644 index 0000000..ac6f940 --- /dev/null +++ b/src/main/resources/reconfigure.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env sh +while true +do + sleep 60 + echo $(wget -S --spider localhost:8081/reconfigure 2>&1) >> /var/log/ONAP/dcaegen2/services/pm-mapper/reconfigure.log +done diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfigurationTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfigurationTest.java new file mode 100644 index 0000000..67510da --- /dev/null +++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/config/DynamicConfigurationTest.java @@ -0,0 +1,153 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2019 Nordix Foundation. + * ================================================================================ + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.dcaegen2.services.pmmapper.config; + + +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import io.undertow.server.HttpServerExchange; +import io.undertow.util.StatusCodes; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.onap.dcaegen2.services.pmmapper.exceptions.ReconfigurationException; +import org.onap.dcaegen2.services.pmmapper.model.EnvironmentConfig; +import org.onap.dcaegen2.services.pmmapper.model.MapperConfig; +import org.onap.dcaegen2.services.pmmapper.utils.RequestSender; +import org.powermock.api.mockito.PowerMockito; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@RunWith(PowerMockRunner.class) +@PrepareForTest({DynamicConfiguration.class, EnvironmentConfig.class}) +public class DynamicConfigurationTest { + private static Path VALID_CONFIG_PATH = Paths.get("src/test/resources/valid_mapper_config.json"); + + private static ArrayList configurables; + private DynamicConfiguration objUnderTest; + private static String config; + private MapperConfig originalMapperConfig; + + @Mock + private RequestSender sender; + + @BeforeClass() + public static void setupBeforeClass() throws Exception { + config = new String(Files.readAllBytes(VALID_CONFIG_PATH)); + } + + @Before + public void setup() throws Exception { + configurables = new ArrayList<>(); + PowerMockito.mockStatic(EnvironmentConfig.class); + PowerMockito.when(EnvironmentConfig.getCBSHostName()).thenReturn(""); + PowerMockito.when(EnvironmentConfig.getCBSPort()).thenReturn(1); + PowerMockito.when(EnvironmentConfig.getServiceName()).thenReturn(""); + + when(sender.send(any())).thenReturn(config); + ConfigHandler configHandler = new ConfigHandler(sender); + originalMapperConfig = configHandler.getMapperConfig(); + objUnderTest = new DynamicConfiguration(configurables, originalMapperConfig); + } + + @Test + public void testNoChangeResponse() throws Exception { + ConfigHandler configHandler = new ConfigHandler(sender); + originalMapperConfig = configHandler.getMapperConfig(); + objUnderTest.setConfigHandler(configHandler); + + HttpServerExchange httpServerExchange = mock(HttpServerExchange.class, RETURNS_DEEP_STUBS); + objUnderTest.handleRequest(httpServerExchange); + assertEquals(originalMapperConfig, objUnderTest.getOriginalConfig()); + verify(httpServerExchange, times(1)).setStatusCode(StatusCodes.OK); + } + + @Test + public void testApplyOriginalUponFailure() throws Exception { + ConfigHandler configHandler = new ConfigHandler(sender); + Configurable configurable = mock(Configurable.class); + configurables.add(configurable); + JsonObject modifiedConfig = new JsonParser().parse(config).getAsJsonObject(); + modifiedConfig.addProperty("dmaap_dr_feed_id","3"); + when(sender.send(any())).thenReturn(modifiedConfig.toString()); + MapperConfig modifiedMapperConfig = configHandler.getMapperConfig(); + + objUnderTest.setConfigHandler(configHandler); + + doAnswer(new Answer() { + boolean failFirstReconfigure = true; + @Override + public Object answer(InvocationOnMock invocationOnMock) throws Throwable { + if (failFirstReconfigure) { + failFirstReconfigure = false; + throw new ReconfigurationException(""); + } + return null; + } + }).when(configurable).reconfigure(any()); + + HttpServerExchange httpServerExchange = mock(HttpServerExchange.class, RETURNS_DEEP_STUBS); + objUnderTest.handleRequest(httpServerExchange); + assertEquals(originalMapperConfig, objUnderTest.getOriginalConfig()); + verify(httpServerExchange, times(1)).setStatusCode(StatusCodes.INTERNAL_SERVER_ERROR); + verify(configurable, times(1)).reconfigure(modifiedMapperConfig); + verify(configurable, times(1)).reconfigure(originalMapperConfig); + } + + @Test + public void testSuccessfulReconfiguration() throws Exception { + ConfigHandler configHandler = new ConfigHandler(sender); + Configurable configurable = mock(Configurable.class); + configurables.add(configurable); + JsonObject modifiedConfig = new JsonParser().parse(config).getAsJsonObject(); + modifiedConfig.addProperty("dmaap_dr_feed_id","3"); + + when(sender.send(any())).thenReturn(modifiedConfig.toString()); + MapperConfig modifiedMapperConfig = configHandler.getMapperConfig(); + objUnderTest.setConfigHandler(configHandler); + + HttpServerExchange httpServerExchange = mock(HttpServerExchange.class, RETURNS_DEEP_STUBS); + objUnderTest.handleRequest(httpServerExchange); + assertEquals(modifiedMapperConfig, objUnderTest.getOriginalConfig()); + verify(httpServerExchange, times(1)).setStatusCode(StatusCodes.OK); + verify(configurable, times(1)).reconfigure(modifiedMapperConfig); + + } +} diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/datarouter/DataRouterSubscriberTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/datarouter/DataRouterSubscriberTest.java index fdc1bf6..dbb95a7 100644 --- a/src/test/java/org/onap/dcaegen2/services/pmmapper/datarouter/DataRouterSubscriberTest.java +++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/datarouter/DataRouterSubscriberTest.java @@ -43,10 +43,15 @@ import io.undertow.server.HttpServerExchange; import io.undertow.util.HeaderMap; import io.undertow.util.StatusCodes; +import java.io.ByteArrayInputStream; import java.io.IOException; +import java.io.InputStream; import java.net.HttpURLConnection; +import java.net.MalformedURLException; import java.net.URL; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.nio.file.Path; import java.nio.file.Paths; import org.junit.Before; import org.junit.Test; @@ -55,28 +60,38 @@ import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import org.onap.dcaegen2.services.pmmapper.config.ConfigHandler; +import org.onap.dcaegen2.services.pmmapper.exceptions.ReconfigurationException; import org.onap.dcaegen2.services.pmmapper.exceptions.TooManyTriesException; +import org.onap.dcaegen2.services.pmmapper.model.EnvironmentConfig; import org.onap.dcaegen2.services.pmmapper.model.Event; import org.onap.dcaegen2.services.pmmapper.model.MapperConfig; import org.onap.dcaegen2.services.pmmapper.utils.HttpServerExchangeAdapter; +import org.onap.dcaegen2.services.pmmapper.utils.RequestSender; import org.powermock.api.mockito.PowerMockito; import org.powermock.core.classloader.annotations.PrepareForTest; import org.powermock.modules.junit4.PowerMockRunner; import utils.LoggingUtils; @RunWith(PowerMockRunner.class) -@PrepareForTest(DataRouterSubscriber.class) +@PrepareForTest({DataRouterSubscriber.class, EnvironmentConfig.class}) public class DataRouterSubscriberTest { + private Path VALID_BC_RESPONSE_PATH = Paths.get("src/test/resources/datarouter_subscriber_test/valid_bc_response.json"); + private Path VALID_METADATA_PATH = Paths.get("src/test/resources/valid_metadata.json"); + private Path INVALID_METADATA_PATH = Paths.get("src/test/resources/invalid_metadata.json"); + private Path VALID_CONFIG_PATH = Paths.get("src/test/resources/valid_mapper_config.json"); @Mock private EventReceiver eventReceiver; + @Mock + private MapperConfig config; private DataRouterSubscriber objUnderTest; @Before public void setUp() { - objUnderTest = new DataRouterSubscriber(eventReceiver); + objUnderTest = new DataRouterSubscriber(eventReceiver, config); } @Test @@ -84,34 +99,37 @@ public class DataRouterSubscriberTest { PowerMockito.mockStatic(Thread.class); URL subEndpoint = mock(URL.class); - MapperConfig config = mock(MapperConfig.class); when(config.getBusControllerSubscriptionUrl()).thenReturn(subEndpoint); HttpURLConnection huc = mock(HttpURLConnection.class, RETURNS_DEEP_STUBS); when(subEndpoint.openConnection()).thenReturn(huc); when(huc.getResponseCode()).thenReturn(300); - Assertions.assertThrows(TooManyTriesException.class, () -> objUnderTest.start(config)); + Assertions.assertThrows(TooManyTriesException.class, () -> objUnderTest.start()); } @Test public void testStartImmediateSuccess() throws IOException, TooManyTriesException, InterruptedException { URL subEndpoint = mock(URL.class); - MapperConfig config = mock(MapperConfig.class); when(config.getBusControllerSubscriptionUrl()).thenReturn(subEndpoint); HttpURLConnection huc = mock(HttpURLConnection.class, RETURNS_DEEP_STUBS); + String bcResponse = new String(Files.readAllBytes(VALID_BC_RESPONSE_PATH)); + InputStream responseStream = new ByteArrayInputStream(bcResponse.getBytes(StandardCharsets.UTF_8)); + when(huc.getInputStream()).thenReturn(responseStream); when(subEndpoint.openConnection()).thenReturn(huc); when(huc.getResponseCode()).thenReturn(200); - objUnderTest.start(config); + objUnderTest.start(); verify(huc, times(1)).getResponseCode(); } @Test public void testStartDelayedSuccess() throws IOException, TooManyTriesException, InterruptedException { PowerMockito.mockStatic(Thread.class); - + HttpURLConnection huc = mock(HttpURLConnection.class, RETURNS_DEEP_STUBS); + String bcResponse = new String(Files.readAllBytes(VALID_BC_RESPONSE_PATH)); + InputStream responseStream = new ByteArrayInputStream(bcResponse.getBytes(StandardCharsets.UTF_8)); + when(huc.getInputStream()).thenReturn(responseStream); URL subEndpoint = mock(URL.class); - MapperConfig config = mock(MapperConfig.class); when(config.getBusControllerSubscriptionUrl()).thenReturn(subEndpoint); - HttpURLConnection huc = mock(HttpURLConnection.class, RETURNS_DEEP_STUBS); + when(subEndpoint.openConnection()).thenReturn(huc); doAnswer(new Answer() { boolean forceRetry = true; @@ -125,7 +143,7 @@ public class DataRouterSubscriberTest { return 200; } }).when(huc).getResponseCode(); - objUnderTest.start(config); + objUnderTest.start(); verify(huc, times(2)).getResponseCode(); } @@ -134,12 +152,11 @@ public class DataRouterSubscriberTest { PowerMockito.mockStatic(Thread.class); URL subEndpoint = mock(URL.class); - MapperConfig config = mock(MapperConfig.class); when(config.getBusControllerSubscriptionUrl()).thenReturn(subEndpoint); HttpURLConnection huc = mock(HttpURLConnection.class, RETURNS_DEEP_STUBS); when(subEndpoint.openConnection()).thenReturn(huc); doThrow(new IOException()).when(huc).getResponseCode(); - Assertions.assertThrows(TooManyTriesException.class, () -> objUnderTest.start(config)); + Assertions.assertThrows(TooManyTriesException.class, () -> objUnderTest.start()); } @Test @@ -170,11 +187,30 @@ public class DataRouterSubscriberTest { verify(eventReceiver, times(0)).receive(any()); } + @Test + public void testStartPositiveResponseCodeInvalidResponseBody() throws Exception{ + PowerMockito.mockStatic(EnvironmentConfig.class); + PowerMockito.mockStatic(Thread.class); + PowerMockito.when(EnvironmentConfig.getCBSHostName()).thenReturn(""); + PowerMockito.when(EnvironmentConfig.getCBSPort()).thenReturn(1); + PowerMockito.when(EnvironmentConfig.getServiceName()).thenReturn(""); + + URL mockURL = mock(URL.class); + when(config.getBusControllerSubscriptionUrl()).thenReturn(mockURL); + HttpURLConnection huc = mock(HttpURLConnection.class, RETURNS_DEEP_STUBS); + String bcResponse = "not a valid response"; + InputStream responseStream = new ByteArrayInputStream(bcResponse.getBytes(StandardCharsets.UTF_8)); + when(huc.getInputStream()).thenReturn(responseStream); + when(mockURL.openConnection()).thenReturn(huc); + when(huc.getResponseCode()).thenReturn(200); + Assertions.assertThrows(TooManyTriesException.class, () -> objUnderTest.start()); + } + @Test public void testRequestInboundInvalidMetadata() throws Exception { HttpServerExchange httpServerExchange = mock(HttpServerExchange.class, RETURNS_DEEP_STUBS); JsonObject metadata = new JsonParser().parse(new String(Files - .readAllBytes(Paths.get("src/test/resources/invalid_metadata.json")))).getAsJsonObject(); + .readAllBytes(INVALID_METADATA_PATH))).getAsJsonObject(); when(httpServerExchange.getRequestHeaders().get(any(String.class)).get(anyInt())) .thenReturn(metadata.toString()); when(httpServerExchange.setStatusCode(anyInt())).thenReturn(httpServerExchange); @@ -218,7 +254,7 @@ public class DataRouterSubscriberTest { when(httpServerExchange.getRequestReceiver()).thenReturn(receiver); String testString = "MESSAGE BODY"; JsonObject metadata = new JsonParser().parse( - new String(Files.readAllBytes(Paths.get("src/test/resources/valid_metadata.json")))).getAsJsonObject(); + new String(Files.readAllBytes(VALID_METADATA_PATH))).getAsJsonObject(); when(httpServerExchange.getRequestHeaders().get(DataRouterSubscriber.METADATA_HEADER).get(anyInt())) .thenReturn(metadata.toString()); when(httpServerExchange.getRequestHeaders().get(DataRouterSubscriber.PUB_ID_HEADER).getFirst()).thenReturn(""); @@ -244,4 +280,108 @@ public class DataRouterSubscriberTest { logAppender.stop(); } + @Test + public void testConfigThrowsMalformedURLException() throws MalformedURLException { + when(config.getBusControllerSubscriptionUrl()).thenThrow(MalformedURLException.class); + Assertions.assertThrows(IllegalStateException.class, () -> objUnderTest.start()); + } + @Test + public void testReconfigurationSameConfig() throws Exception { + PowerMockito.mockStatic(EnvironmentConfig.class); + PowerMockito.when(EnvironmentConfig.getCBSHostName()).thenReturn(""); + PowerMockito.when(EnvironmentConfig.getCBSPort()).thenReturn(1); + PowerMockito.when(EnvironmentConfig.getServiceName()).thenReturn(""); + + RequestSender sender = mock(RequestSender.class); + String mapperConfig = new String(Files.readAllBytes(VALID_CONFIG_PATH)); + when(sender.send(any())).thenReturn(mapperConfig); + MapperConfig originalMapperConfig = new ConfigHandler(sender).getMapperConfig(); + + DataRouterSubscriber objUnderTest = new DataRouterSubscriber(eventReceiver, originalMapperConfig); + objUnderTest.reconfigure(originalMapperConfig); + assertEquals(originalMapperConfig, objUnderTest.getConfig()); + } + + @Test + public void testReconfigurationModifiedFeedId() throws Exception { + PowerMockito.mockStatic(EnvironmentConfig.class); + PowerMockito.when(EnvironmentConfig.getCBSHostName()).thenReturn(""); + PowerMockito.when(EnvironmentConfig.getCBSPort()).thenReturn(1); + PowerMockito.when(EnvironmentConfig.getServiceName()).thenReturn(""); + + URL mockURL = mock(URL.class); + when(config.getBusControllerSubscriptionUrl()).thenReturn(mockURL); + HttpURLConnection huc = mock(HttpURLConnection.class, RETURNS_DEEP_STUBS); + String bcResponse = new String(Files.readAllBytes(VALID_BC_RESPONSE_PATH)); + InputStream responseStream = new ByteArrayInputStream(bcResponse.getBytes(StandardCharsets.UTF_8)); + when(huc.getInputStream()).thenReturn(responseStream); + when(mockURL.openConnection()).thenReturn(huc); + when(huc.getResponseCode()).thenReturn(200); + + PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL); + + RequestSender sender = mock(RequestSender.class); + String mapperConfig = new String(Files.readAllBytes(VALID_CONFIG_PATH)); + when(sender.send(any())).thenReturn(mapperConfig); + MapperConfig originalMapperConfig = new ConfigHandler(sender).getMapperConfig(); + JsonObject modifiedMapperConfigObj = new JsonParser().parse(mapperConfig).getAsJsonObject(); + modifiedMapperConfigObj.addProperty("dmaap_dr_feed_id", "3"); + when(sender.send(any())).thenReturn(modifiedMapperConfigObj.toString()); + MapperConfig modifiedMapperConfig = new ConfigHandler(sender).getMapperConfig(); + + DataRouterSubscriber objUnderTest = new DataRouterSubscriber(eventReceiver, originalMapperConfig); + objUnderTest.reconfigure(modifiedMapperConfig); + assertEquals(modifiedMapperConfig, objUnderTest.getConfig()); + } + + @Test + public void testReconfigurationModifiedUsername() throws Exception { + PowerMockito.mockStatic(EnvironmentConfig.class); + PowerMockito.when(EnvironmentConfig.getCBSHostName()).thenReturn(""); + PowerMockito.when(EnvironmentConfig.getCBSPort()).thenReturn(1); + PowerMockito.when(EnvironmentConfig.getServiceName()).thenReturn(""); + + URL mockURL = mock(URL.class); + when(config.getBusControllerSubscriptionUrl()).thenReturn(mockURL); + HttpURLConnection huc = mock(HttpURLConnection.class, RETURNS_DEEP_STUBS); + String bcResponse = new String(Files.readAllBytes(VALID_BC_RESPONSE_PATH)); + InputStream responseStream = new ByteArrayInputStream(bcResponse.getBytes(StandardCharsets.UTF_8)); + when(huc.getInputStream()).thenReturn(responseStream); + when(mockURL.openConnection()).thenReturn(huc); + when(huc.getResponseCode()).thenReturn(200); + + PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL); + + RequestSender sender = mock(RequestSender.class); + String mapperConfig = new String(Files.readAllBytes(VALID_CONFIG_PATH)); + when(sender.send(any())).thenReturn(mapperConfig); + MapperConfig originalMapperConfig = new ConfigHandler(sender).getMapperConfig(); + JsonObject modifiedMapperConfigObj = new JsonParser().parse(mapperConfig).getAsJsonObject(); + modifiedMapperConfigObj.get("streams_subscribes") + .getAsJsonObject().get("dmaap_subscriber") + .getAsJsonObject().get("dmaap_info") + .getAsJsonObject() + .addProperty("username", "bob"); + when(sender.send(any())).thenReturn(modifiedMapperConfigObj.toString()); + MapperConfig modifiedMapperConfig = new ConfigHandler(sender).getMapperConfig(); + + DataRouterSubscriber objUnderTest = new DataRouterSubscriber(eventReceiver, originalMapperConfig); + objUnderTest.reconfigure(modifiedMapperConfig); + assertEquals(modifiedMapperConfig, objUnderTest.getConfig()); + } + + @Test + public void testReconfigurationMalformedURL() throws Exception { + when(config.getBusControllerSubscriptionUrl()).thenThrow(MalformedURLException.class); + Assertions.assertThrows(IllegalStateException.class, () -> objUnderTest.reconfigure(config)); + } + @Test + public void testReconfigurationException() throws Exception { + PowerMockito.mockStatic(Thread.class); + URL url = mock(URL.class); + when(url.toString()).thenReturn("http://valid:8080/"); + when(url.openConnection()).thenThrow(IOException.class); + when(config.getBusControllerSubscriptionUrl()).thenReturn(url); + Assertions.assertThrows(ReconfigurationException.class, () -> objUnderTest.reconfigure(config)); + } } diff --git a/src/test/resources/datarouter_subscriber_test/valid_bc_response.json b/src/test/resources/datarouter_subscriber_test/valid_bc_response.json new file mode 100644 index 0000000..201b786 --- /dev/null +++ b/src/test/resources/datarouter_subscriber_test/valid_bc_response.json @@ -0,0 +1,15 @@ +{ + "type": "dr_Sub", + "lastMod": "2019-03-11T14:29:39.659", + "status": "VALID", + "dcaeLocationName": "location", + "deliveryURL": "delivery_url", + "feedId": "2", + "logURL": "https://dmaap-dr-prov/sublog/2", + "owner": "DGL", + "subId": "1", + "suspended": false, + "use100": false, + "username": "username", + "userpwd": "password" +} \ No newline at end of file -- cgit 1.2.3-korg