import org.openecomp.sdc.api.consumer.IConfiguration;
import java.net.URI;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
public class AsdcConfig implements IConfiguration {
- private String host;
- private String consumer;
- private String consumerId;
- private String env;
- private String keystorePath;
- private String keystorePass;
- private int pollingInterval; // Time between listening sessions
- private int pollingTimeout; // Time to listen for (dmaap timeout url param)/1000
- private List<String> types = new ArrayList<>();
- private String user;
- private String pass;
-
- private URI storeOp;
-
- Properties props;
-
- private final EELFLogger logger = EELFManager.getInstance().getLogger(AsdcConfig.class);
-
- public AsdcConfig(Properties props) throws Exception {
- this.props = props;
- init();
- }
-
- private void init() throws Exception {
- if (props != null) {
- // Keystore for ca cert
- keystorePath = props.getProperty("appc.asdc.keystore.path");
- keystorePass = props.getProperty("appc.asdc.keystore.pass");
-
- // ASDC host
- host = props.getProperty("appc.asdc.host");
- env = props.getProperty("appc.asdc.env");
- user = props.getProperty("appc.asdc.user");
- pass = props.getProperty("appc.asdc.pass");
-
- // DMaaP properties
- consumer = props.getProperty("appc.asdc.consumer");
- consumerId = props.getProperty("appc.asdc.consumer.id");
-
- pollingInterval = Integer.valueOf(props.getProperty("interval", "60"));
-
- // Client uses cambriaClient-0.2.4 which throws non relevant (wrong)
- // exceptions with times > 30s
- pollingTimeout = Integer.valueOf(props.getProperty("timeout", "25"));
-
- // Anything less than 60 and we risk 429 Too Many Requests
- if (pollingInterval < 60) {
- pollingInterval = 60;
- }
-
- if (pollingInterval > pollingInterval) {
- logger.warn(String.format(
- "Message acknowledgement may be delayed by %ds in the ADSC listener. [Listening Time: %s, Poll Period: %s]",
- pollingInterval - pollingTimeout, pollingTimeout, pollingInterval));
- }
-
- logParams();
-
- // Download type
- types.add("APPC_CONFIG");
- types.add("VF_LICENSE");
- types.add("TOSCA_CSAR");
- /*
- This types seems redundant, as it looks from the code that they are not being used anywhere
- */
-
- storeOp = new URI(props.getProperty("appc.asdc.provider.url"));
- }
- }
-
- @Override
- public boolean activateServerTLSAuth() {
- return false;
- }
-
- //@Override
- public boolean isFilterInEmptyResources() {
- return false;
- }
-
- @Override
- public String getAsdcAddress() {
- return host;
- }
-
- @Override
- public String getConsumerGroup() {
- return consumer;
- }
-
- @Override
- public String getConsumerID() {
- return consumerId;
- }
-
- @Override
- public String getEnvironmentName() {
- return env;
- }
-
- @Override
- public String getKeyStorePassword() {
- return keystorePass;
- }
-
- @Override
- public String getKeyStorePath() {
- return keystorePath;
- }
-
- @Override
- public String getPassword() {
- return pass;
- }
-
- @Override
- public int getPollingInterval() {
- return pollingInterval;
- }
-
- @Override
- public int getPollingTimeout() {
- return pollingTimeout;
- }
-
- @Override
- public List<String> getRelevantArtifactTypes() {
- return types;
- }
-
- @Override
- public String getUser() {
- return user;
- }
-
- public URI getStoreOpURI() {
- return storeOp;
- }
-
- /**
- * Logs the relevant parameters
- */
- public void logParams() {
- Map<String, String> params = new HashMap<String, String>();
- params.put("ASDC Host", getAsdcAddress());
- params.put("ASDC Environment", getEnvironmentName());
- params.put("Consumer Name", getConsumerGroup());
- params.put("Consumer ID", getConsumerID());
- params.put("Poll Active Wait", String.valueOf(getPollingInterval()));
- params.put("Poll Timeout", String.valueOf(getPollingTimeout()));
-
- logger.info(String.format("ASDC Params: %s", params));
- }
+ private String host;
+ private String consumer;
+ private String consumerId;
+ private String env;
+ private String keystorePath;
+ private String keystorePass;
+ /** Polling internal is time between listening sessions */
+ private int pollingInterval;
+ /** Polling timeout is the time to listen for (dmaap timeout url param)/1000 */
+ private int pollingTimeout;
+ private List<String> types = new ArrayList<>();
+ private String user;
+ private String pass;
+
+ private URI storeOp;
+
+ private Properties props;
+
+ private final EELFLogger logger = EELFManager.getInstance().getLogger(AsdcConfig.class);
+
+ AsdcConfig(Properties props) throws Exception {
+ this.props = props;
+ init();
+ }
+
+ private void init() throws Exception {
+ if (props == null) {
+ logger.error("SdcConfig init is skipped due to properties is null");
+ return;
+ }
+
+ // Keystore for ca cert
+ keystorePath = props.getProperty("appc.asdc.keystore.path");
+ keystorePass = props.getProperty("appc.asdc.keystore.pass");
+
+ // ASDC host
+ host = props.getProperty("appc.asdc.host");
+ env = props.getProperty("appc.asdc.env");
+ user = props.getProperty("appc.asdc.user");
+ pass = props.getProperty("appc.asdc.pass");
+
+ // DMaaP properties
+ consumer = props.getProperty("appc.asdc.consumer");
+ consumerId = props.getProperty("appc.asdc.consumer.id");
+
+ pollingInterval = Integer.valueOf(props.getProperty("interval", "60"));
+
+ // Client uses cambriaClient-0.2.4 which throws non relevant (wrong)
+ // exceptions with times > 30s
+ pollingTimeout = Integer.valueOf(props.getProperty("timeout", "25"));
+
+ // Anything less than 60 and we risk 429 Too Many Requests
+ if (pollingInterval < 60) {
+ pollingInterval = 60;
+ }
+
+ if (pollingInterval > pollingTimeout) {
+ logger.warn(String.format(
+ "Message acknowledgement may be delayed by %ds in the ADSC listener. [Listening Time: %s, Poll Period: %s]",
+ pollingInterval - pollingTimeout, pollingTimeout, pollingInterval));
+ }
+
+ logParams();
+
+ // Download type
+ /*
+ This types seems redundant, as it looks from the code that they are not being used anywhere
+ */
+ types.add("APPC_CONFIG");
+ types.add("VF_LICENSE");
+ types.add("TOSCA_CSAR");
+
+ storeOp = new URI(props.getProperty("appc.asdc.provider.url"));
+ }
+
+ @Override
+ public boolean activateServerTLSAuth() {
+ return false;
+ }
+
+ public boolean isFilterInEmptyResources() {
+ return false;
+ }
+
+ @Override
+ public String getAsdcAddress() {
+ return host;
+ }
+
+ @Override
+ public String getConsumerGroup() {
+ return consumer;
+ }
+
+ @Override
+ public String getConsumerID() {
+ return consumerId;
+ }
+
+ @Override
+ public String getEnvironmentName() {
+ return env;
+ }
+
+ @Override
+ public String getKeyStorePassword() {
+ return keystorePass;
+ }
+
+ @Override
+ public String getKeyStorePath() {
+ return keystorePath;
+ }
+
+ @Override
+ public String getPassword() {
+ return pass;
+ }
+
+ @Override
+ public int getPollingInterval() {
+ return pollingInterval;
+ }
+
+ @Override
+ public int getPollingTimeout() {
+ return pollingTimeout;
+ }
+
+ @Override
+ public List<String> getRelevantArtifactTypes() {
+ return types;
+ }
+
+ @Override
+ public String getUser() {
+ return user;
+ }
+
+ URI getStoreOpURI() {
+ return storeOp;
+ }
+
+ /**
+ * Logs the relevant parameters
+ */
+ private void logParams() {
+ Map<String, String> params = new HashMap<>();
+ params.put("ASDC Host", getAsdcAddress());
+ params.put("ASDC Environment", getEnvironmentName());
+ params.put("Consumer Name", getConsumerGroup());
+ params.put("Consumer ID", getConsumerID());
+ params.put("Poll Active Wait", String.valueOf(getPollingInterval()));
+ params.put("Poll Timeout", String.valueOf(getPollingTimeout()));
+
+ logger.info(String.format("ASDC Params: %s", params));
+ }
}
*/
package org.openecomp.appc.sdc.listener;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+import org.openecomp.appc.configuration.Configuration;
+import org.openecomp.appc.configuration.ConfigurationFactory;
+import org.openecomp.sdc.api.IDistributionClient;
+import org.openecomp.sdc.api.results.IDistributionClientResult;
import org.openecomp.sdc.impl.DistributionClientFactory;
import org.openecomp.sdc.utils.DistributionActionResultEnum;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-import org.openecomp.appc.configuration.Configuration;
-import org.openecomp.appc.configuration.ConfigurationFactory;
-import org.openecomp.sdc.api.IDistributionClient;
-import org.openecomp.sdc.api.results.IDistributionClientResult;
-import org.openecomp.sdc.impl.DistributionClientFactory;
-import org.openecomp.sdc.utils.DistributionActionResultEnum;
-import com.att.eelf.configuration.EELFLogger;
-import com.att.eelf.configuration.EELFManager;
-
+/**
+ * SDC listener handles bundle start and stop through start and stop method. <p>
+ * Register connection with SDC server based on properties file configuration when start,
+ * and disconnect with SDC server when stop.
+ */
public class AsdcListener {
+ private final EELFLogger logger = EELFManager.getInstance().getLogger(AsdcListener.class);
/**
* The bundle context
private AsdcConfig config;
private CountDownLatch latch;
-
- private final EELFLogger logger = EELFManager.getInstance().getLogger(AsdcListener.class);
+ private Thread startThread = null;
@SuppressWarnings("unused")
public void start() throws Exception {
- logger.info("Starting bundle ASDC Listener");
+ // Add timestamp to the log to differentiate the jmeter run testing calls.
+ final long timeStamp = System.currentTimeMillis();
+ logger.info(String.format("[%d] Starting SDC Listener", timeStamp));
+
Configuration configuration = ConfigurationFactory.getConfiguration();
Properties props = configuration.getProperties();
-
config = new AsdcConfig(props);
+ logger.debug(String.format("[%d] created SDC config", timeStamp));
client = DistributionClientFactory.createDistributionClient();
+ logger.debug(String.format("[%d] created SDC client", timeStamp));
+
callback = new AsdcCallback(config.getStoreOpURI(), client);
+ logger.debug(String.format("[%d] created SDC callback", timeStamp));
latch = new CountDownLatch(1);
- new Thread(new Runnable() {
- @Override
- public void run() {
- initialRegistration(config);
-
- IDistributionClientResult result = client.init(config, callback);
-
- if (result.getDistributionActionResult() == DistributionActionResultEnum.SUCCESS) {
- client.start();
- } else {
- logger.error(String.format("Could not register ASDC client. %s - %s", result.getDistributionActionResult(),
- result.getDistributionMessageResult()));
- }
-
- latch.countDown();
- }
- }).start();
+ startThread = new Thread(new StartRunnable(timeStamp));
+ startThread.setName(String.format("[%d] sdcListener start", timeStamp));
+ logger.debug(String.format("[%d] created SDC initialization thread", timeStamp));
+ startThread.start();
}
@SuppressWarnings("unused")
public void stop() throws InterruptedException {
- logger.info("Stopping ASDC Listener");
- latch.await(10, TimeUnit.SECONDS);
+ // Add timestamp to the log to differentiate the jmeter run testing calls.
+ final long timeStamp = System.currentTimeMillis();
+ logger.info(String.format("[%d] Stopping ASDC Listener", timeStamp));
+
+ stopStartThread(timeStamp);
+
+ if (latch != null) {
+ logger.debug(String.format("[%d] waiting ASDC latch count to 0 for 10 seconds", timeStamp));
+ latch.await(10, TimeUnit.SECONDS);
+ latch = null;
+ }
if (callback != null) {
+ logger.debug(String.format("[%d] stopping ASDC callback", timeStamp));
callback.stop();
+ callback = null;
}
if (client != null) {
+ logger.debug(String.format("[%d] stopping ASDC client", timeStamp));
client.stop();
+ client = null;
}
- logger.info("ASDC Listener stopped successfully");
+ logger.info(String.format("[%d] ASDC Listener stopped successfully", timeStamp));
}
- private boolean initialRegistration(AsdcConfig config) {
- try {
- final String jsonTemplate = "{\"consumerName\": \"%s\",\"consumerSalt\": \"%s\",\"consumerPassword\":\"%s\"}";
- String saltedPassStr = org.openecomp.tlv.sdc.security.Passwords.hashPassword(config.getPassword());
- if (saltedPassStr == null || !saltedPassStr.contains(":")) {
- return false;
+ void stopStartThread(long timeStamp) throws InterruptedException {
+ if (startThread == null) {
+ return;
+ }
+
+ if (startThread.getState() == Thread.State.TERMINATED) {
+ logger.debug(String.format("[%d] ASDC thread(%s) is already terminated.",
+ timeStamp, startThread.getName()));
+ } else {
+ logger.debug(String.format("[%d] ASDC thread(%s) is to be interrupted with state(%s)",
+ timeStamp, startThread.getName(), startThread.getState().toString()));
+
+ startThread.interrupt();
+
+ logger.debug(String.format("[%d] ASDC thread(%s) has been interrupted(%s) with state(%s)",
+ timeStamp, startThread.getName(), startThread.isInterrupted(),
+ startThread.getState().toString()));
+ }
+ startThread = null;
+ }
+
+ /**
+ * Runnable implementation for actual initialization during ASDC listener start
+ */
+ class StartRunnable implements Runnable {
+ private final long timeStamp;
+
+ StartRunnable(long theTimeStamp) {
+ timeStamp = theTimeStamp;
+ }
+
+ /**
+ * This run method calls ASDC client for init and start which are synchronized calls along with stop.
+ * To interrupt this thread at stop time, we added thread interrupted checking in each step
+ * for earlier interruption.
+ */
+ @Override
+ public void run() {
+ if (!initialRegistration()) {
+ logger.warn(String.format("[%d] ASDC thread initial registration failed.", timeStamp));
+ }
+
+ if (isThreadInterrupted("after initial registration")) {
+ return;
+ }
+
+ IDistributionClientResult result = client.init(config, callback);
+
+ if (isThreadInterrupted("after client init")) {
+ return;
}
- String[] saltedPass = saltedPassStr.split(":");
- String json = String.format(jsonTemplate, config.getUser(), saltedPass[0], saltedPass[1]);
+ if (result.getDistributionActionResult() == DistributionActionResultEnum.SUCCESS) {
+ client.start();
+ } else {
+ logger.error(String.format("[%d] Could not register ASDC client. %s - %s",
+ timeStamp, result.getDistributionActionResult(), result.getDistributionMessageResult()));
+ }
+
+ latch.countDown();
+ }
- Map<String, String> headers = new HashMap<>();
- // TODO - Replace the header below to sdc's requirements. What should the new value be
- headers.put("USER_ID", "test");
+ private boolean initialRegistration() {
+ try {
+ final String jsonTemplate =
+ "{\"consumerName\": \"%s\",\"consumerSalt\": \"%s\",\"consumerPassword\":\"%s\"}";
+ String saltedPassStr = org.openecomp.tlv.sdc.security.Passwords.hashPassword(config.getPassword());
+ if (saltedPassStr == null || !saltedPassStr.contains(":")) {
+ return false;
+ }
- // TODO - How to format the url. Always same endpoint or ports?
- String host = config.getAsdcAddress();
- URL url = new URL(
- String.format("http%s://%s/sdc2/rest/v1/consumers", host.contains("443") ? "s" : "", host));
+ String[] saltedPass = saltedPassStr.split(":");
+ String json = String.format(jsonTemplate, config.getUser(), saltedPass[0], saltedPass[1]);
- logger.info(String.format("Attempting to register user %s on %s with salted pass of %s", config.getUser(),
- url, saltedPass[1]));
+ Map<String, String> headers = new HashMap<>();
+ // TODO - Replace the header below to sdc's requirements. What should the new value be
+ headers.put("USER_ID", "test");
- ProviderResponse result = ProviderOperations.post(url, json, headers);
- return result.getStatus() == 200;
- } catch (Exception e) {
- logger.error("Error performing initial registration with ASDC server. User may not be able to connect", e);
+ // TODO - How to format the url. Always same endpoint or ports?
+ String host = config.getAsdcAddress();
+ URL url = new URL(String.format("http%s://%s/sdc2/rest/v1/consumers",
+ host.contains("443") ? "s" : "", host));
+
+ logger.info(String.format("Attempting to register user %s on %s with salted pass of %s",
+ config.getUser(), url, saltedPass[1]));
+
+ ProviderOperations providerOperations = new ProviderOperations();
+ ProviderResponse result = providerOperations.post(url, json, headers);
+ return result.getStatus() == 200;
+ } catch (Exception e) {
+ logger.error(
+ "Error performing initial registration with ASDC server. User may not be able to connect",
+ e);
+ return false;
+ }
+ }
+
+ private boolean isThreadInterrupted(String details) {
+ if (Thread.currentThread().isInterrupted()) {
+ logger.info(String.format("[%d] ASDC thread interrupted %s.", timeStamp, details));
+ return true;
+ }
return false;
}
}
package org.openecomp.appc.sdc.listener;
-
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
ArtifactStorageService.class,
ToscaCsarArtifactProcessor.class,
ArtifactProcessorFactory.class})
-public class TestAsdcListener {
+public class AsdcCallbackTest {
IDistributionClient client;
private EventSender eventSender;
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP : APPC
+ * ================================================================================
+ * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Copyright (C) 2017 Amdocs
+ * =============================================================================
+ * 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.
+ *
+ * ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ * ============LICENSE_END=========================================================
+ */
+package org.openecomp.appc.sdc.listener;
+
+import com.att.eelf.configuration.EELFLogger;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.openecomp.sdc.api.IDistributionClient;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.reflect.Whitebox;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(Thread.class)
+public class AsdcListenerTest {
+ private AsdcListener asdcListener;
+ private EELFLogger mockLogger = mock(EELFLogger.class);
+
+ @Before
+ public void setUp() throws Exception {
+ asdcListener = new AsdcListener();
+
+ // to avoid operation on logger fail, mock up the logger
+ Whitebox.setInternalState(asdcListener, "logger", mockLogger);
+ }
+
+ @Test
+ public void testStart() throws Exception {
+ asdcListener.start();
+ Assert.assertTrue("Should created startThread",
+ Whitebox.getInternalState(asdcListener, "startThread") != null);
+ }
+
+ @Test
+ public void testStop() throws Exception {
+ // test interrupt thread and other null case
+ MockThread mockThread = spy(new MockThread());
+ mockThread.setNewState(Thread.State.TIMED_WAITING);
+ Whitebox.setInternalState(asdcListener, "startThread", mockThread);
+
+ asdcListener.stop();
+ Mockito.verify(mockThread, times(1)).interrupt();
+ Assert.assertTrue("Should reset startThread",
+ Whitebox.getInternalState(asdcListener, "startThread") == null);
+
+ // test other non-null case and thread null case
+ IDistributionClient mockClient = mock(IDistributionClient.class);
+ Whitebox.setInternalState(asdcListener, "client", mockClient);
+ AsdcCallback mockCallback = mock(AsdcCallback.class);
+ Whitebox.setInternalState(asdcListener, "callback", mockCallback);
+ CountDownLatch mockLatch = mock(CountDownLatch.class);
+ Whitebox.setInternalState(asdcListener, "latch", mockLatch);
+
+ asdcListener.stop();
+
+ Mockito.verify(mockLatch, times(1)).await(10, TimeUnit.SECONDS);
+ Mockito.verify(mockCallback, times(1)).stop();
+ Mockito.verify(mockClient, times(1)).stop();
+ Assert.assertTrue("Should reset latch",
+ Whitebox.getInternalState(asdcListener, "latch") == null);
+ Assert.assertTrue("Should reset callback",
+ Whitebox.getInternalState(asdcListener, "callback") == null);
+ Assert.assertTrue("Should reset client",
+ Whitebox.getInternalState(asdcListener, "client") == null);
+ }
+
+ @Test
+ public void testStopStartThread() throws Exception {
+ // null case
+ asdcListener.stopStartThread(123);
+ Mockito.verify(mockLogger, times(0)).debug(String.valueOf(any()));
+
+ MockThread mockThread = spy(new MockThread());
+
+ // thread terminated case
+ Whitebox.setInternalState(asdcListener, "startThread", mockThread);
+ mockThread.setNewState(Thread.State.TERMINATED);
+ asdcListener.stopStartThread(123);
+ Mockito.verify(mockThread, times(0)).interrupt();
+ Mockito.verify(mockLogger, times(1)).debug(String.valueOf(any()));
+ Assert.assertTrue("Should reset startThread",
+ Whitebox.getInternalState(asdcListener, "startThread") == null);
+
+ // thread not termianted case
+ int timesCallThread = 0;
+ int timesCallLogger = 1;
+ for(Thread.State state : Thread.State.values()) {
+ if (state == Thread.State.TERMINATED) {
+ continue;
+ }
+ Whitebox.setInternalState(asdcListener, "startThread", mockThread);
+ mockThread.setNewState(state);
+ asdcListener.stopStartThread(123);
+ Mockito.verify(mockThread, times(++ timesCallThread)).interrupt();
+ Mockito.verify(mockLogger, times(timesCallLogger += 2)).debug(String.valueOf(any()));
+ Assert.assertTrue("Should reset startThread",
+ Whitebox.getInternalState(asdcListener, "startThread") == null);
+ }
+ }
+
+ /*
+ * I have used the following PowerMockito (due to Thread.getName() is a final method)
+ * try to mock up the thread behavior. But the mock Thread.getName() always returns null
+ * which works in intelliJ Junit test, but not Jenkins build:
+ * Thread mockThread = PowerMockito.mock(Thread.class);
+ * PowerMockito.doReturn(Thread.State.TERMINATED).when(mockThread).getState();
+ * PowerMockito.doReturn("testing").when(mockThread).getName();
+ * Hence, here goes the MockThread class to override Thread to my expected behavior.
+ */
+ class MockThread extends Thread {
+ private State state;
+
+ private MockThread() {
+ super.setName("testing");
+ }
+
+ void setNewState(State newState) {
+ state = newState;
+ }
+
+ @Override
+ public State getState() {
+ return state;
+ }
+ }
+}
\ No newline at end of file
Code Review / appc.git / commitdiff