diff options
Diffstat (limited to 'appc-event-listener/appc-event-listener-bundle/src/main/java/org')
40 files changed, 5400 insertions, 0 deletions
diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AbstractListener.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AbstractListener.java new file mode 100644 index 000000000..b36f4d7e6 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AbstractListener.java @@ -0,0 +1,159 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener; + +import java.security.SecureRandom; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.RejectedExecutionHandler; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.commons.lang3.concurrent.BasicThreadFactory; +import org.openecomp.appc.listener.impl.EventHandlerImpl; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; + +public abstract class AbstractListener implements Listener { + + private final EELFLogger LOG = EELFManager.getInstance().getLogger(AbstractListener.class); + + protected AtomicBoolean run = new AtomicBoolean(false); + + protected int QUEUED_MIN = 1; + protected int QUEUED_MAX = 10; + + protected int THREAD_MIN = 4; + protected int THREAD_MAX = THREAD_MIN; // Fixed thread pool + protected int THREAD_SCALE_DOWN_SEC = 10; // Number of seconds to wait until we remove idle threads + + protected ThreadPoolExecutor executor; + + protected EventHandler dmaap; + + protected ListenerProperties props; + + private String listenerId; + + public AbstractListener(ListenerProperties props) { + updateProperties(props); + + dmaap = new EventHandlerImpl(props); + if (dmaap.getClientId().equals("0")) { + dmaap.setClientId(String.valueOf(new SecureRandom().nextInt(1000))); + } + + BlockingQueue<Runnable> threadQueue = new ArrayBlockingQueue<Runnable>(QUEUED_MAX + QUEUED_MIN + 1); + executor = new ThreadPoolExecutor(THREAD_MIN, THREAD_MAX, THREAD_SCALE_DOWN_SEC, TimeUnit.SECONDS, threadQueue, + new JobRejectionHandler()); + + // Custom Named thread factory + BasicThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("DMaaP-Worker-%d").build(); + executor.setThreadFactory(threadFactory); + + run.set(true); + } + + /** + * Starts a loop that will only end after stop() or stopNow() are called. The loop will read messages off the DMaaP + * topic and perform some action on them while writing messages back to DMaaP at critical points in the execution. + * Inherited from Runnable. + * + * @see java.lang.Runnable#run() + */ + @Override + public void run() { + LOG.error("Listener.run() has not been implemented"); + } + + @Override + public void stop() { + run.set(false); + LOG.info(String.format("Stopping with %d messages in queue", executor.getQueue().size())); + executor.shutdown(); + try { + executor.awaitTermination(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + LOG.error("Listener graceful stop() failed", e); + } + + // close DMaaP clients + if (dmaap != null) { + dmaap.closeClients(); + } + LOG.info("Listener Thread Pool Finished"); + } + + @Override + public void stopNow() { + run.set(false); + LOG.info(String.format("StopNow called. Orphaning %d messages in the queue", executor.getQueue().size())); + executor.getQueue().clear(); + stop(); + } + + @Override + public String getBenchmark() { + return String.format("%s - No benchmarking implemented.", getListenerId()); + } + + @Override + public String getListenerId() { + return listenerId; + } + + // Sets the id of the listener in + @Override + public void setListenerId(String id) { + listenerId = id; + } + + private void updateProperties(ListenerProperties properties) { + this.props = properties; + QUEUED_MIN = + Integer.valueOf(props.getProperty(ListenerProperties.KEYS.THREADS_MIN_QUEUE, String.valueOf(QUEUED_MIN))); + QUEUED_MAX = + Integer.valueOf(props.getProperty(ListenerProperties.KEYS.THREADS_MAX_QUEUE, String.valueOf(QUEUED_MAX))); + THREAD_MIN = + Integer.valueOf(props.getProperty(ListenerProperties.KEYS.THREADS_MIN_POOL, String.valueOf(THREAD_MIN))); + THREAD_MAX = + Integer.valueOf(props.getProperty(ListenerProperties.KEYS.THREADS_MAX_POOL, String.valueOf(THREAD_MAX))); + + listenerId = props.getPrefix(); + } + + /** + * This class will be used to handle what happens when we cannot add a job because of a ThreadPool issue. It does + * not get invoked if there is any fault with the job. NOTE: So far, this has only been seen when doing a + * {@link Listener#stopNow} + * + */ + class JobRejectionHandler implements RejectedExecutionHandler { + @Override + public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { + LOG.error(String.format("A job was rejected. [%s]", r)); + } + + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AppcEventListenerActivator.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AppcEventListenerActivator.java new file mode 100644 index 000000000..4c2e28196 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/AppcEventListenerActivator.java @@ -0,0 +1,185 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener; + +import org.openecomp.appc.configuration.Configuration; +import org.openecomp.appc.configuration.ConfigurationFactory; +import org.openecomp.appc.listener.impl.ControllerImpl; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; +import org.osgi.framework.BundleActivator; +import org.osgi.framework.BundleContext; +import org.osgi.framework.ServiceRegistration; + +import java.util.HashSet; +import java.util.Properties; +import java.util.Set; + +/** + * This activator is used to initialize and terminate the dmaap listener controller and pool(s) + * <p> + * The DMaaP listener is responsible for listening to a topic on the Universal Event Bus and reading in messages that + * conform to the DCAE message format for APPC. These messages will then be parsed and passed along to the APPC Provider + * to take action on. The listener will also send messages out on DMaaP during critical phases. The messages sent out will + * have a status of: + * <ul> + * <li><i>PENDING</i> - The listener has read the message off of DMaaP and has put it in the queue to be processed</li> + * <li><i>ACTIVE</i> - The listener has begun actually processing the request and is waiting on the appc provider to + * complete the request</li> + * <li><i>SUCCESS</i> or <i>FAILURE</i> - The listener has gotten a response back from the appc provider. If it is a + * FAILURE, a message should also be included</li> + * </ul> + * </p> + * <p> + * Activation of the bundle will provision 1 controller that in turn will provision 1 (or in the future more) listener + * to interact with DMaaP. Each listener will have a queue of messages read off of DMaaP and a thread pool of workers to + * process them. This worker is responsible for contacting appc provider to perform the action + * </p> + * <p> + * When the bundle is deactivated, the stopNow() method is called and the thread pool is emptied and all remaining jobs + * are orphaned. Alternatively stop() could be called which would allow all remaining jobs in the queue to complete at + * the cost of longer run time. + * </p> + * + * @since Aug 30, 2015 + * @version $Id$ + */ +public class AppcEventListenerActivator implements BundleActivator { + + /** + * The bundle registration + */ + private ServiceRegistration registration = null; + + /** + * The configuration object + */ + private Configuration configuration; + + /** + * The bundle context + */ + private static BundleContext context; + + /** + * The reference to the actual implementation object that implements the services + */ + private Controller adapter; + + /** + * The logger to be used + */ + private final EELFLogger LOG = EELFManager.getInstance().getLogger(AppcEventListenerActivator.class); + + /** + * Called when this bundle is started so the Framework can perform the bundle-specific activities necessary to start + * this bundle. This method can be used to register services or to allocate any resources that this bundle needs. + * <p> + * This method must complete and return to its caller in a timely manner. + * </p> + * + * @param ctx + * The execution context of the bundle being started. + * @throws java.lang.Exception + * If this method throws an exception, this bundle is marked as stopped and the Framework will remove + * this bundle's listeners, unregister all services registered by this bundle, and release all services + * used by this bundle. + * @see org.osgi.framework.BundleActivator#start(org.osgi.framework.BundleContext) + */ + @Override + public void start(BundleContext ctx) throws Exception { + LOG.info("Starting Bundle " + getName()); + + context = ctx; + + configuration = ConfigurationFactory.getConfiguration(); + + Properties props = configuration.getProperties(); + + Set<ListenerProperties> listeners = new HashSet<ListenerProperties>(); + + // Configure App-C Closed Loop Listener +/* ListenerProperties clProps = new ListenerProperties("appc.ClosedLoop", props); + clProps.setListenerClass(org.openecomp.appc.listener.CL.impl.ListenerImpl.class); + listeners.add(clProps);*/ + + // Configure App-C 1607 Closed Loop Listener + ListenerProperties cl1607Props = new ListenerProperties("appc.ClosedLoop1607", props); + cl1607Props.setListenerClass(org.openecomp.appc.listener.CL1607.impl.ListenerImpl.class); + listeners.add(cl1607Props); + + +/* ListenerProperties clLCMProps = new ListenerProperties("appc.LCM", props); + clLCMProps.setListenerClass(org.openecomp.appc.listener.LCM.impl.ListenerImpl.class); + listeners.add(clLCMProps);*/ + +/* + ListenerProperties clLCMProps1607 = new ListenerProperties("appc.LCM1607", props); + clLCMProps1607.setListenerClass(org.openecomp.appc.listener.LCM1607.impl.ListenerImpl.class); + listeners.add(clLCMProps1607); +*/ + + adapter = new ControllerImpl(listeners); + if (ctx != null && registration == null) { + LOG.info("Registering service DMaaP Controller"); + registration = ctx.registerService(Controller.class, adapter, null); + } + adapter.start(); + + LOG.info("DMaaP Listener started successfully"); + } + + /** + * Called when this bundle is stopped so the Framework can perform the bundle-specific activities necessary to stop + * the bundle. In general, this method should undo the work that the BundleActivator.start method started. There + * should be no active threads that were started by this bundle when this bundle returns. A stopped bundle must not + * call any Framework objects. + * <p> + * This method must complete and return to its caller in a timely manner. + * </p> + * + * @param ctx + * The execution context of the bundle being stopped. + * @throws java.lang.Exception + * If this method throws an exception, the bundle is still marked as stopped, and the Framework will + * remove the bundle's listeners, unregister all services registered by the bundle, and release all + * services used by the bundle. * + * @see org.osgi.framework.BundleActivator#stop(org.osgi.framework.BundleContext) + */ + @Override + public void stop(BundleContext ctx) throws Exception { + boolean stopNow = true; + LOG.info("Stopping DMaaP Listener. StopNow=" + stopNow); + adapter.stop(stopNow); + if (registration != null) { + registration.unregister(); + registration = null; + } + LOG.info("DMaaP Listener stopped successfully"); + } + + public String getName() { + return "DMaaP Listener"; + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ListenerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ListenerImpl.java new file mode 100644 index 000000000..3f0f8a112 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ListenerImpl.java @@ -0,0 +1,128 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL.impl; + +import java.text.DateFormat; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.List; +import java.util.TimeZone; +import java.util.concurrent.RejectedExecutionException; + +import org.openecomp.appc.listener.AbstractListener; +import org.openecomp.appc.listener.ListenerProperties; +import org.openecomp.appc.listener.CL.model.IncomingMessage; +import org.openecomp.appc.listener.CL.model.Status; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; +import com.att.eelf.i18n.EELFResourceManager; + +public class ListenerImpl extends AbstractListener { + + private final EELFLogger LOG = EELFManager.getInstance().getLogger(ListenerImpl.class); + + private long startTime = 0; + + public ListenerImpl(ListenerProperties props) { + super(props); + String url = props.getProperty("provider.url"); + LOG.info("DMaaP Provider Endpoint: " + url); + ProviderOperations.setUrl(url); + + // Set Basic Auth + String user = props.getProperty("provider.user"); + String pass = props.getProperty("provider.pass"); + ProviderOperations.setAuthentication(user, pass); + } + + @Override + public void run() { + // Some vars for benchmarking + startTime = System.currentTimeMillis(); + + LOG.info("Running DMaaP Listener"); + + while (run.get()) { + // Only update if the queue is low. otherwise we read in more + // messages than we need + try { + if (executor.getQueue().size() <= QUEUED_MIN) { + LOG.debug("DMaaP queue running low. Querying for more jobs"); + List<IncomingMessage> messages = dmaap.getIncomingEvents(IncomingMessage.class, QUEUED_MAX); + LOG.debug(String.format("Read %d messages from dmaap", messages.size())); + for (IncomingMessage incoming : messages) { + // Acknowledge that we read the event + LOG.info("Acknowledging Message: " + incoming.getId()); + dmaap.postStatus(incoming.toOutgoing(Status.PENDING, null).toString()); + } + for (IncomingMessage incoming : messages) { + // Add to pool if still running + if (run.get()) { + LOG.info(String.format("Adding DMaaP message to pool queue [%s]", incoming.getId())); + if (incoming.isValid()) { + try { + executor.execute(new WorkerImpl(incoming, dmaap)); + } catch (RejectedExecutionException rejectEx) { + LOG.error("Task Rejected: ", rejectEx); + } + } else { + // Badly formed message + LOG.error("Message was not valid. Rejecting"); + } + } else { + LOG.info("Run stopped. Orphaning Message: " + incoming.getId()); + } + } + } + } catch (Exception e) { + LOG.error("Exception " + e.getClass().getSimpleName() + " caught in DMaaP listener"); + LOG.error(EELFResourceManager.format(e)); + LOG.error("DMaaP Listener logging and ignoring the exception, continue..."); + } + } + + LOG.info("Stopping DMaaP Listener thread"); + + // We've told the listener to stop + // TODO - Should we: + // 1) Put a message back on the queue indicating that APP-C never got to + // the message + // or + // 2) Let downstream figure it out after timeout between PENDING and + // ACTIVE messages + } + + @Override + public String getBenchmark() { + long time = System.currentTimeMillis(); + DateFormat df = new SimpleDateFormat("HH:mm:ss"); + df.setTimeZone(TimeZone.getTimeZone("UTC")); + String runningTime = df.format(new Date(time - startTime)); + + String out = String.format("Running for %s and completed %d jobs using %d threads.", runningTime, + executor.getCompletedTaskCount(), executor.getPoolSize()); + LOG.info("***BENCHMARK*** " + out); + return out; + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ProviderOperations.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ProviderOperations.java new file mode 100644 index 000000000..7a53bcdd9 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/ProviderOperations.java @@ -0,0 +1,296 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL.impl; + +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.net.MalformedURLException; +import java.net.Socket; +import java.net.URL; +import java.net.UnknownHostException; +import java.security.KeyManagementException; +import java.security.KeyStore; +import java.security.KeyStoreException; +import java.security.NoSuchAlgorithmException; +import java.security.UnrecoverableKeyException; +import java.security.cert.CertificateException; +import java.security.cert.X509Certificate; + +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; +import javax.net.ssl.X509TrustManager; + +import org.apache.commons.codec.binary.Base64; +import org.apache.commons.io.IOUtils; +import org.apache.http.HttpResponse; +import org.apache.http.HttpVersion; +import org.apache.http.client.HttpClient; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.conn.ClientConnectionManager; +import org.apache.http.conn.scheme.PlainSocketFactory; +import org.apache.http.conn.scheme.Scheme; +import org.apache.http.conn.scheme.SchemeRegistry; +import org.apache.http.conn.ssl.SSLSocketFactory; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.DefaultHttpClient; +import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager; +import org.apache.http.params.BasicHttpParams; +import org.apache.http.params.HttpParams; +import org.apache.http.params.HttpProtocolParams; +import org.apache.http.protocol.HTTP; +import org.json.JSONObject; +import org.openecomp.appc.exceptions.APPCException; +import org.openecomp.appc.listener.CL.model.IncomingMessage; +import org.openecomp.appc.listener.util.Mapper; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; + +public class ProviderOperations { + + private static final EELFLogger LOG = EELFManager.getInstance().getLogger(ProviderOperations.class); + + private static URL url; + + private static String basic_auth; + + //@formatter:off + @SuppressWarnings("nls") + private final static String TEMPLATE = "{\"input\": {\"common-request-header\": {\"service-request-id\": \"%s\"},\"vnf-resource\": {\"vm-id\": \"%s\"%s}}}"; + //@formatter:on + + /** + * Calls the AppcProvider to run a topology directed graph + * + * @param msg + * The incoming message to be run + * @return True if the result is success. Never returns false and throws an exception instead. + * @throws UnsupportedEncodingException + * @throws Exception + * if there was a failure processing the request. The exception message is the failure reason. + */ + @SuppressWarnings("nls") + public static boolean topologyDG(IncomingMessage msg) throws APPCException { + if (msg == null) { + throw new APPCException("Provided message was null"); + } + + HttpPost post = null; + try { + // Concatenate the "action" on the end of the URL + String path = url.getPath() + ":" + msg.getAction().getValue().toLowerCase(); + URL serviceUrl = new URL(url.getProtocol(), url.getHost(), url.getPort(), path); + + post = new HttpPost(serviceUrl.toExternalForm()); + post.setHeader("Content-Type", "application/json"); + post.setHeader("Accept", "application/json"); + + // Set Auth + if (basic_auth != null) { + post.setHeader("Authorization", "Basic " + basic_auth); + } + + String body = buildReqest(msg.getId(), msg.getUrl(), msg.getIdentityUrl()); + + + LOG.info(String.format("DMaaP ACTION PATH : %s", path)); + LOG.info(String.format("DMaaP ACTION BODY : %s", body)); + + StringEntity entity = new StringEntity(body); + entity.setContentType("application/json"); + post.setEntity(new StringEntity(body)); + } catch (UnsupportedEncodingException | MalformedURLException e) { + throw new APPCException(e); + } + + HttpClient client = getHttpClient(); + + int httpCode = 0; + String respBody = null; + try { + HttpResponse response = client.execute(post); + httpCode = response.getStatusLine().getStatusCode(); + respBody = IOUtils.toString(response.getEntity().getContent()); + } catch (IOException e) { + throw new APPCException(e); + } + + if (httpCode >= 200 && httpCode < 300 && respBody != null) { + JSONObject json; + try { + json = Mapper.toJsonObject(respBody); + } catch (Exception e) { + LOG.error("Error processing response from provider. Could not map response to json", e); + throw new APPCException("APPC has an unknown RPC error"); + } + boolean success; + String reason; + try { + JSONObject header = json.getJSONObject("output").getJSONObject("common-response-header"); + success = header.getBoolean("success"); + reason = header.getString("reason"); + } catch (Exception e) { + LOG.error("Unknown error prcoessing failed response from provider. Json not in expected format", e); + throw new APPCException("APPC has an unknown RPC error"); + } + if (success) { + return true; + } + String reasonStr = reason == null ? "Unknown" : reason; + LOG.warn(String.format("Topology Operation [%s] failed. Reason: %s", msg.getId(), reasonStr)); + throw new APPCException(reasonStr); + + } + throw new APPCException(String.format("Unexpected response from endpoint: [%d] - %s ", httpCode, respBody)); + } + + /** + * Updates the static var URL and returns the value; + * + * @return The new value of URL + */ + public static String getUrl() { + return url.toExternalForm(); + } + + public static void setUrl(String newUrl) { + try { + url = new URL(newUrl); + } catch (MalformedURLException e) { + e.printStackTrace(); + } + } + + /** + * Sets the basic authentication header for the given user and password. If either entry is null then set basic auth + * to null + * + * @param user + * The user with optional domain name + * @param password + * The password for the user + * @return The new value of the basic auth string that will be used in the request headers + */ + public static String setAuthentication(String user, String password) { + if (user != null && password != null) { + String authStr = user + ":" + password; + basic_auth = new String(Base64.encodeBase64(authStr.getBytes())); + } else { + basic_auth = null; + } + return basic_auth; + } + + /** + * Builds the request body for a topology operation + * + * @param id + * The request id + * @param action + * The action in lowercase + * @param url + * The vm's url + * @return A String containing the request body + */ + private static String buildReqest(String id, String url, String ident) { + String extraVmResource = ""; + if (ident != null) { + extraVmResource = String.format(", \"identity-url\": \"%s\"", ident); + } + return String.format(TEMPLATE, id, url, extraVmResource); + } + + @SuppressWarnings("deprecation") + private static HttpClient getHttpClient() throws APPCException { + HttpClient client; + if (url.getProtocol().equals("https")) { + try { + KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType()); + trustStore.load(null, null); + MySSLSocketFactory sf = new MySSLSocketFactory(trustStore); + sf.setHostnameVerifier(MySSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER); + + HttpParams params = new BasicHttpParams(); + HttpProtocolParams.setVersion(params, HttpVersion.HTTP_1_1); + HttpProtocolParams.setContentCharset(params, HTTP.UTF_8); + + SchemeRegistry registry = new SchemeRegistry(); + registry.register(new Scheme("http", PlainSocketFactory.getSocketFactory(), 80)); + registry.register(new Scheme("https", sf, 443)); + registry.register(new Scheme("https", sf, 8443)); + registry.register(new Scheme("http", sf, 8181)); + + ClientConnectionManager ccm = new ThreadSafeClientConnManager(params, registry); + client = new DefaultHttpClient(ccm, params); + } catch (Exception e) { + client = new DefaultHttpClient(); + } + } else if (url.getProtocol().equals("http")) { + client = new DefaultHttpClient(); + } else { + throw new APPCException( + "The provider.topology.url property is invalid. The url did not start with http[s]"); + } + return client; + } + + @SuppressWarnings("deprecation") + public static class MySSLSocketFactory extends SSLSocketFactory { + private SSLContext sslContext = SSLContext.getInstance("TLS"); + + public MySSLSocketFactory(KeyStore truststore) throws NoSuchAlgorithmException, KeyManagementException, + KeyStoreException, UnrecoverableKeyException { + super(truststore); + + TrustManager tm = new X509TrustManager() { + @Override + public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException { + } + + @Override + public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException { + } + + @Override + public X509Certificate[] getAcceptedIssuers() { + return null; + } + }; + + sslContext.init(null, new TrustManager[] { + tm + }, null); + } + + @Override + public Socket createSocket(Socket socket, String host, int port, boolean autoClose) + throws IOException, UnknownHostException { + return sslContext.getSocketFactory().createSocket(socket, host, port, autoClose); + } + + @Override + public Socket createSocket() throws IOException { + return sslContext.getSocketFactory().createSocket(); + } + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/WorkerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/WorkerImpl.java new file mode 100644 index 000000000..1eb13f736 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/impl/WorkerImpl.java @@ -0,0 +1,79 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL.impl; + +import org.openecomp.appc.exceptions.APPCException; +import org.openecomp.appc.listener.EventHandler; +import org.openecomp.appc.listener.CL.model.IncomingMessage; +import org.openecomp.appc.listener.CL.model.Status; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; + +public class WorkerImpl implements Runnable { + + private final EELFLogger LOG = EELFManager.getInstance().getLogger(WorkerImpl.class); + + // Should have all of the data we need for processing + private IncomingMessage event; + + // So we can post messages from inside the worker. + private EventHandler dmaap; + + public WorkerImpl(IncomingMessage message, EventHandler dmaap) { + this.event = message; + this.dmaap = dmaap; + } + + @Override + public void run() { + LOG.debug(String.format("Started working on %s", event.getId())); + dmaap.postStatus(event.toOutgoing(Status.ACTIVE, null)); + // Run the dg in a try catch to handle all exceptions and update the + // message at the end + try { + if (doDG(event)) { + dmaap.postStatus(event.toOutgoing(Status.SUCCESS, null)); + LOG.debug(String.format("Event %s finished successfully", event.getId())); + } else { + // Should never happen. Exception with message should be thrown instead. + LOG.error(String.format( + "We somehow returned false from doDG() instead of throwing exception. Incoming event [%s]", + event.toJson().toString())); + dmaap.postStatus(event.toOutgoing(Status.FAILURE)); + } + + } catch (Exception e) { + // Unknown exception from DG method. Fail and pass the exception + // along + String msg = "Exception: " + e.getMessage(); + LOG.warn(String.format("Event %s finished with failure. %s", event.getId(), msg)); + dmaap.postStatus(event.toOutgoing(Status.FAILURE, msg)); + } + + LOG.debug("Done working on " + event.getId()); + } + + private boolean doDG(IncomingMessage msg) throws APPCException { + return ProviderOperations.topologyDG(msg); + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Action.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Action.java new file mode 100644 index 000000000..a8fe8035b --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Action.java @@ -0,0 +1,55 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL.model; + +public enum Action { + Restart("Restart"), Rebuild("Rebuild"), Migrate("Migrate"), Evacuate("Evacuate"); + + /** + * Converts the string to an Action + * + * @param value + * The string to try and convert. Is case insensitive + * @return The action matching the string or null if no match was found. + */ + public static Action toAction(String value) { + if (value != null) { + for (Action e : values()) { + if (e.getValue().toUpperCase().equals(value.toUpperCase())) { + return e; + } + } + } + + return null; + } + + private String value; + + private Action(String valueToUse) { + value = valueToUse; + } + + public final String getValue() { + return value; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/CommonMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/CommonMessage.java new file mode 100644 index 000000000..76a112618 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/CommonMessage.java @@ -0,0 +1,181 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL.model; + +import java.io.Serializable; + +import org.json.JSONObject; +import org.openecomp.appc.listener.util.Mapper; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion; + +/** + * This class holds attributes that are common to DMaaP messages both coming in from DCAE and being sent out by APPC + * + */ +@JsonSerialize(include = Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class CommonMessage implements Serializable { + + private static final long serialVersionUID = 1L; + + /* + * The unique id of the event as of 1602 + */ + @JsonProperty("eventID") + private String id; + + /* + * The time that the request was sent out. + */ + @JsonProperty("requestTime") + private String requestTime; + + /* + * The originator of the event + */ + @JsonProperty("requestClient") + private String requestClient; + + /* + * The system that sent the message + */ + @JsonProperty("from") + private String fromSystem; + + /* + * The actual trap message + */ + @JsonProperty("message") + private String message; + + /* + * The vm name associated with the event + */ + @JsonProperty("VMName") + private String vmName; + + /* + * The policy name on the incoming event + */ + @JsonProperty("policyName") + private String policyName; + + /* + * The policy version on the incoming event + */ + @JsonProperty("policyVersion") + private String policyVersion; + + @JsonIgnore + private long startTime = System.currentTimeMillis(); + + /* + * Getters and Setters + */ + + public String getId() { + return id; + } + + public String getRequestTime() { + return requestTime; + } + + public String getRequestClient() { + return requestClient; + } + + public String getFromSystem() { + return fromSystem; + } + + public String getMessage() { + return message; + } + + public String getPolicyName() { + return policyName; + } + + public String getPolicyVersion() { + return policyVersion; + } + + public String getVmName() { + return vmName; + } + + public long getStartTime() { + return startTime; + } + + public void setId(String eventId) { + id = eventId; + } + + public void setRequestTime(String requestTime) { + this.requestTime = requestTime; + } + + public void setRequestClient(String requestClient) { + this.requestClient = requestClient; + } + + public void setFromSystem(String fromSystem) { + this.fromSystem = fromSystem; + } + + public void setMessage(String message) { + this.message = message; + } + + public void setPolicyName(String name) { + policyName = name; + } + + public void setPolicyVersion(String version) { + policyVersion = version; + } + + public void setVmName(String vmName) { + this.vmName = vmName; + } + + public void setStartTime(long startTime) { + this.startTime = startTime; + } + + /** + * Convenience method to return a json representation of this object. + * + * @return The json representation of this object + */ + public JSONObject toJson() { + return Mapper.toJsonObject(this); + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/IncomingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/IncomingMessage.java new file mode 100644 index 000000000..60d1d004b --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/IncomingMessage.java @@ -0,0 +1,140 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL.model; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion; + +/** + * This class reperesnts a message coming in from DCAE. + * + */ +@JsonSerialize(include = Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class IncomingMessage extends CommonMessage { + + private static final long serialVersionUID = 1L; + + /* + * The action being requested. Its presence signals that it is an incoming message and it is not present on outgoing + * messages + */ + @JsonProperty("request") + private String request; + + /* + * The url for the server used for auth. http://<compute>:<port>/<tenentId>/server/<serverID> + */ + @JsonProperty("VServerSelfLink") + private String Url; + + /* + * The tenant Id in OpenStack that the server belongs to + */ + @JsonProperty("TenantID") + private String TenantId; + + /* + * The VM's UUID in + */ + @JsonProperty("VMID") + private String VmId; + + @JsonProperty("Identity") + private String identityUrl; + + public String getRequest() { + return request; + } + + @JsonIgnore + public Action getAction() { + return Action.toAction(request); + } + + public String getUrl() { + return Url; + } + + public String getTenantId() { + return TenantId; + } + + public String getVmId() { + return VmId; + } + + public String getIdentityUrl() { + return identityUrl; + } + + public void setUrl(String Url) { + this.Url = Url; + } + + public void setTenantId(String TenantId) { + this.TenantId = TenantId; + } + + public void setVmId(String VmId) { + this.VmId = VmId; + } + + public void setRequest(String request) { + this.request = request; + } + + public void setIdentityUrl(String identityUrl) { + this.identityUrl = identityUrl; + } + + @Override + public String toString() { + String time = getRequestTime() != null ? getRequestTime() : "N/A"; + // String req = request != null ? request : "N/A"; + return String.format("[%s - %s]", time, getId()); + } + + public String toOutgoing(Status status) { + return toOutgoing(status, getMessage()); + } + + public String toOutgoing(Status status, String msg) { + OutgoingMessage out = new OutgoingMessage(this); + out.setResponse(status); + out.setMessage(msg); + return out.toResponse().toString(); + } + + /** + * Determines if this message should be parsed parsed. Will eventually check that the message is well formed, has + * all required fields, and had not exceeded any timing restrictions. + * + * @return True if the message should be parsed. False otherwise + */ + public boolean isValid() { + return true; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/OutgoingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/OutgoingMessage.java new file mode 100644 index 000000000..fa8560ed2 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/OutgoingMessage.java @@ -0,0 +1,150 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL.model; + +import java.net.InetAddress; +import java.security.SecureRandom; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.TimeZone; + +import org.json.JSONObject; +import org.openecomp.appc.listener.util.Mapper; +import org.openecomp.appc.util.Time; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion; + +/** + * This class represents a message being sent out to DMaaP by APPC to update listeners on the status of a request + * + */ +@JsonSerialize(include = Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class OutgoingMessage extends CommonMessage { + + private static final long serialVersionUID = -5447940920271469613L; + + @JsonProperty("response") + private Status response; + + @JsonProperty("responseTime") + private String responseTime; + + @JsonProperty("originalRequest") + private String originalRequest; + + public OutgoingMessage() { + + } + + public OutgoingMessage(IncomingMessage msg) { + setId(msg.getId()); + setOriginalRequest(msg.getRequest()); + setRequestClient(msg.getRequestClient()); + setRequestTime(msg.getRequestTime()); + setVmName(msg.getVmName()); + setFromSystem(generateFrom()); + setResponse(Status.PENDING); + setPolicyName(msg.getPolicyName()); + setPolicyVersion(msg.getPolicyVersion()); + setStartTime(msg.getStartTime()); + } + + @JsonProperty("duration") + public long getDuration() { + return System.currentTimeMillis() - getStartTime(); + } + + public Status getResponse() { + return response; + } + + public String getResponseTime() { + return responseTime; + } + + public String getOriginalRequest() { + return originalRequest; + } + + @JsonIgnore + public void setResponse(Status response) { + this.response = response; + } + + public void setResponse(String responseString) { + this.response = Status.valueOf(responseString); + } + + public void setResponseTime(String responseTime) { + this.responseTime = responseTime; + } + + public void setOriginalRequest(String originalRequest) { + this.originalRequest = originalRequest; + } + + public void updateResponseTime() { + SecureRandom rand = new SecureRandom(); + SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd kk:mm:ss.SSS"); + df.setTimeZone(TimeZone.getTimeZone("UTC")); + String date = df.format(new Date(Time.utcTime())); + this.responseTime = String.format("%s%03d", date, rand.nextInt(1000)); + } + + public String generateFrom() { + String name; + try { + InetAddress iAddress = InetAddress.getLocalHost(); + name = iAddress.getCanonicalHostName(); + } catch (Exception e) { + // Could not get anything from the InetAddress + name = "UnknownHost"; + } + return "appc@" + name; + } + + public JSONObject toResponse() { + updateResponseTime(); + JSONObject json = Mapper.toJsonObject(this); + + if (!json.has("message")) { + // If there is no message, parrot the status (response field) + // TODO - Can this be removed for 1602 making message truely optional? + json.put("message", this.getResponse().toString()); + } + + // Removed duplication of status from message for 1602 + // json.put("message", String.format("%s: %s", request, json.get("message"))); + + return json; + } + + @Override + public String toString() { + return String.format("%s - %s", getId(), getResponse()); + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Status.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Status.java new file mode 100644 index 000000000..d8e8b5c64 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL/model/Status.java @@ -0,0 +1,74 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL.model; + +public enum Status { + /* + * APP-C acknowledges that it has read the event off of the wire. This is the initial status of an OutgoingEvent + */ + PENDING("PENDING"), + + /* + * APP-C has started processing the event + */ + ACTIVE("ACTIVE"), + + /* + * APP-C has finished processing the event without errors + */ + SUCCESS("SUCCESS"), + + /* + * APP-C has finished processing the event with errors + */ + FAILURE("FAILURE"); + + /** + * Converts the string to an Status + * + * @param value + * The string to try and convert. Is case insensitive + * @return The status matching the string or null if no match was found. + */ + public static Status toStatus(String value) { + if (value != null) { + for (Status e : values()) { + if (e.getValue().toUpperCase().equals(value.toUpperCase())) { + return e; + } + } + } + + return null; + } + + private String value; + + private Status(String valueToUse) { + value = valueToUse; + } + + public final String getValue() { + return value; + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ListenerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ListenerImpl.java new file mode 100644 index 000000000..5d3b741ff --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ListenerImpl.java @@ -0,0 +1,130 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL1607.impl; + +import java.text.DateFormat; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.List; +import java.util.TimeZone; +import java.util.concurrent.RejectedExecutionException; + +import org.openecomp.appc.listener.AbstractListener; +import org.openecomp.appc.listener.ListenerProperties; +import org.openecomp.appc.listener.CL1607.model.IncomingMessage; +import org.openecomp.appc.listener.CL1607.model.Status; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; +import com.att.eelf.i18n.EELFResourceManager; + +public class ListenerImpl extends AbstractListener { + + private final EELFLogger LOG = EELFManager.getInstance().getLogger(ListenerImpl.class); + + private long startTime = 0; + + public ListenerImpl(ListenerProperties props) { + super(props); + String url = props.getProperty("provider.url"); + LOG.info("DMaaP Provider Endpoint: " + url); + ProviderOperations.setUrl(url); + + // Set Basic Auth + String user = props.getProperty("provider.user"); + String pass = props.getProperty("provider.pass"); + ProviderOperations.setAuthentication(user, pass); + } + + @Override + public void run() { + // Some vars for benchmarking + startTime = System.currentTimeMillis(); + + LOG.info("Running DMaaP Listener"); + + while (run.get()) { + // Only update if the queue is low. otherwise we read in more + // messages than we need + try { + if (executor.getQueue().size() <= QUEUED_MIN) { + LOG.debug("DMaaP queue running low. Querying for more jobs"); + List<IncomingMessage> messages = dmaap.getIncomingEvents(IncomingMessage.class, QUEUED_MAX); + LOG.debug(String.format("Read %d messages from dmaap", messages.size())); + for (IncomingMessage incoming : messages) { + // Acknowledge that we read the event + LOG.info("Acknowledging Message: " + incoming.getHeader().getRequestID()); + + //TODO: Should we post a pending status for 1607 + //dmaap.postStatus(incoming.toOutgoing(Status.PENDING, null).toString()); + } + for (IncomingMessage incoming : messages) { + // Add to pool if still running + if (run.get()) { + LOG.info(String.format("Adding DMaaP message to pool queue [%s]", incoming.getHeader().getRequestID())); + if (incoming.isValid()) { + try { + executor.execute(new WorkerImpl(incoming, dmaap)); + } catch (RejectedExecutionException rejectEx) { + LOG.error("Task Rejected: ", rejectEx); + } + } else { + // Badly formed message + LOG.error("Message was not valid. Rejecting"); + } + } else { + LOG.info("Run stopped. Orphaning Message: " + incoming.getHeader().getRequestID()); + } + } + } + } catch (Exception e) { + LOG.error("Exception " + e.getClass().getSimpleName() + " caught in DMaaP listener"); + LOG.error(EELFResourceManager.format(e)); + LOG.error("DMaaP Listener logging and ignoring the exception, continue..."); + } + } + + LOG.info("Stopping DMaaP Listener thread"); + + // We've told the listener to stop + // TODO - Should we: + // 1) Put a message back on the queue indicating that APP-C never got to + // the message + // or + // 2) Let downstream figure it out after timeout between PENDING and + // ACTIVE messages + } + + @Override + public String getBenchmark() { + long time = System.currentTimeMillis(); + DateFormat df = new SimpleDateFormat("HH:mm:ss"); + df.setTimeZone(TimeZone.getTimeZone("UTC")); + String runningTime = df.format(new Date(time - startTime)); + + String out = String.format("Running for %s and completed %d jobs using %d threads.", runningTime, + executor.getCompletedTaskCount(), executor.getPoolSize()); + LOG.info("***BENCHMARK*** " + out); + return out; + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ProviderOperations.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ProviderOperations.java new file mode 100644 index 000000000..c300df154 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/ProviderOperations.java @@ -0,0 +1,291 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL1607.impl; + +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.net.MalformedURLException; +import java.net.Socket; +import java.net.URL; +import java.net.UnknownHostException; +import java.security.KeyManagementException; +import java.security.KeyStore; +import java.security.KeyStoreException; +import java.security.NoSuchAlgorithmException; +import java.security.UnrecoverableKeyException; +import java.security.cert.CertificateException; +import java.security.cert.X509Certificate; + +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; +import javax.net.ssl.X509TrustManager; + +import org.apache.commons.codec.binary.Base64; +import org.apache.commons.io.IOUtils; +import org.apache.http.HttpResponse; +import org.apache.http.HttpVersion; +import org.apache.http.client.HttpClient; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.conn.ClientConnectionManager; +import org.apache.http.conn.scheme.PlainSocketFactory; +import org.apache.http.conn.scheme.Scheme; +import org.apache.http.conn.scheme.SchemeRegistry; +import org.apache.http.conn.ssl.SSLSocketFactory; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.DefaultHttpClient; +import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager; +import org.apache.http.params.BasicHttpParams; +import org.apache.http.params.HttpParams; +import org.apache.http.params.HttpProtocolParams; +import org.apache.http.protocol.HTTP; +import org.json.JSONObject; +import org.openecomp.appc.exceptions.APPCException; +import org.openecomp.appc.listener.CL1607.model.IncomingMessage; +import org.openecomp.appc.listener.util.Mapper; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; + +public class ProviderOperations { + + private static final EELFLogger LOG = EELFManager.getInstance().getLogger(ProviderOperations.class); + + private static URL url; + + private static String basic_auth; + + //@formatter:off + @SuppressWarnings("nls") + private final static String TEMPLATE = "{\"input\": {\"common-request-header\": {\"service-request-id\": \"%s\"},\"config-payload\": {\"config-url\": \"%s\",\"config-json\":\"%s\"}}}"; + //@formatter:on + + /** + * Calls the AppcProvider to run a topology directed graph + * + * @param msg + * The incoming message to be run + * @return True if the result is success. Never returns false and throws an exception instead. + * @throws UnsupportedEncodingException + * @throws Exception + * if there was a failure processing the request. The exception message is the failure reason. + */ + @SuppressWarnings("nls") + public static boolean topologyDG(IncomingMessage msg) throws APPCException { + if (msg == null) { + throw new APPCException("Provided message was null"); + } + + HttpPost post = null; + try { + // Concatenate the "action" on the end of the URL + String path = url.getPath() + ":" + msg.getAction().getValue(); + URL serviceUrl = new URL(url.getProtocol(), url.getHost(), url.getPort(), path); + + post = new HttpPost(serviceUrl.toExternalForm()); + post.setHeader("Content-Type", "application/json"); + post.setHeader("Accept", "application/json"); + + // Set Auth + if (basic_auth != null) { + post.setHeader("Authorization", "Basic " + basic_auth); + } + + //String body = buildReqest(msg.getId(), msg.getUrl(), msg.getIdentityUrl()); + String body = buildReqest(msg.getHeader().getRequestID(), msg.getPayload().getGenericVnfId(), msg.getPayload().getPgStreams()); + StringEntity entity = new StringEntity(body); + entity.setContentType("application/json"); + post.setEntity(new StringEntity(body)); + } catch (UnsupportedEncodingException | MalformedURLException e) { + throw new APPCException(e); + } + + HttpClient client = getHttpClient(); + + int httpCode = 0; + String respBody = null; + try { + HttpResponse response = client.execute(post); + httpCode = response.getStatusLine().getStatusCode(); + respBody = IOUtils.toString(response.getEntity().getContent()); + } catch (IOException e) { + throw new APPCException(e); + } + + if (httpCode == 200 && respBody != null) { + JSONObject json; + try { + json = Mapper.toJsonObject(respBody); + } catch (Exception e) { + LOG.error("Error prcoessing response from provider. Could not map response to json", e); + throw new APPCException("APPC has an unknown RPC error"); + } + boolean success; + String reason; + try { + JSONObject header = json.getJSONObject("output").getJSONObject("common-response-header"); + success = header.getBoolean("success"); + reason = header.getString("reason"); + } catch (Exception e) { + LOG.error("Unknown error prcoessing failed response from provider. Json not in expected format", e); + throw new APPCException("APPC has an unknown RPC error"); + } + if (success) { + return true; + } + String reasonStr = reason == null ? "Unknown" : reason; + LOG.warn(String.format("Topology Operation [%s] failed. Reason: %s", msg.getHeader().getRequestID(), reasonStr)); + throw new APPCException(reasonStr); + + } + throw new APPCException(String.format("Unexpected response from endpoint: [%d] - %s ", httpCode, respBody)); + } + + /** + * Updates the static var URL and returns the value; + * + * @return The new value of URL + */ + public static String getUrl() { + return url.toExternalForm(); + } + + public static void setUrl(String newUrl) { + try { + url = new URL(newUrl); + } catch (MalformedURLException e) { + e.printStackTrace(); + } + } + + /** + * Sets the basic authentication header for the given user and password. If either entry is null then set basic auth + * to null + * + * @param user + * The user with optional domain name + * @param password + * The password for the user + * @return The new value of the basic auth string that will be used in the request headers + */ + public static String setAuthentication(String user, String password) { + if (user != null && password != null) { + String authStr = user + ":" + password; + basic_auth = new String(Base64.encodeBase64(authStr.getBytes())); + } else { + basic_auth = null; + } + return basic_auth; + } + + /** + * Builds the request body for a topology operation + * + * @param id + * The request id + * @param url + * The vm's url + * + * @param pgstreams + * The streams to send to the traffic generator + * + * @return A String containing the request body + */ + private static String buildReqest(String id, String url, String pgstreams) { + + return String.format(TEMPLATE, id, url, pgstreams); + } + + @SuppressWarnings("deprecation") + private static HttpClient getHttpClient() throws APPCException { + HttpClient client; + if (url.getProtocol().equals("https")) { + try { + KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType()); + trustStore.load(null, null); + MySSLSocketFactory sf = new MySSLSocketFactory(trustStore); + sf.setHostnameVerifier(MySSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER); + + HttpParams params = new BasicHttpParams(); + HttpProtocolParams.setVersion(params, HttpVersion.HTTP_1_1); + HttpProtocolParams.setContentCharset(params, HTTP.UTF_8); + + SchemeRegistry registry = new SchemeRegistry(); + registry.register(new Scheme("http", PlainSocketFactory.getSocketFactory(), 80)); + registry.register(new Scheme("https", sf, 443)); + registry.register(new Scheme("https", sf, 8443)); + registry.register(new Scheme("http", sf, 8181)); + + ClientConnectionManager ccm = new ThreadSafeClientConnManager(params, registry); + client = new DefaultHttpClient(ccm, params); + } catch (Exception e) { + client = new DefaultHttpClient(); + } + } else if (url.getProtocol().equals("http")) { + client = new DefaultHttpClient(); + } else { + throw new APPCException( + "The provider.topology.url property is invalid. The url did not start with http[s]"); + } + return client; + } + + @SuppressWarnings("deprecation") + public static class MySSLSocketFactory extends SSLSocketFactory { + private SSLContext sslContext = SSLContext.getInstance("TLS"); + + public MySSLSocketFactory(KeyStore truststore) throws NoSuchAlgorithmException, KeyManagementException, + KeyStoreException, UnrecoverableKeyException { + super(truststore); + + TrustManager tm = new X509TrustManager() { + @Override + public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException { + } + + @Override + public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException { + } + + @Override + public X509Certificate[] getAcceptedIssuers() { + return null; + } + }; + + sslContext.init(null, new TrustManager[] { + tm + }, null); + } + + @Override + public Socket createSocket(Socket socket, String host, int port, boolean autoClose) + throws IOException, UnknownHostException { + return sslContext.getSocketFactory().createSocket(socket, host, port, autoClose); + } + + @Override + public Socket createSocket() throws IOException { + return sslContext.getSocketFactory().createSocket(); + } + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/WorkerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/WorkerImpl.java new file mode 100644 index 000000000..52bcd2f5e --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/impl/WorkerImpl.java @@ -0,0 +1,82 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL1607.impl; + +import org.openecomp.appc.exceptions.APPCException; +import org.openecomp.appc.listener.EventHandler; +import org.openecomp.appc.listener.CL1607.model.IncomingMessage; +import org.openecomp.appc.listener.CL1607.model.Status; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; + +public class WorkerImpl implements Runnable { + + private final EELFLogger LOG = EELFManager.getInstance().getLogger(WorkerImpl.class); + + // Should have all of the data we need for processing + private IncomingMessage event; + + // So we can post messages from inside the worker. + private EventHandler dmaap; + + public WorkerImpl(IncomingMessage message, EventHandler dmaap) { + this.event = message; + this.dmaap = dmaap; + } + + @Override + public void run() { + LOG.debug(String.format("Started working on %s", event.getHeader().getRequestID())); + + dmaap.postStatus(event.toOutgoing(Status.ACCEPTED)); + // Run the dg in a try catch to handle all exceptions and update the + // message at the end + try { + if (doDG(event)) { + dmaap.postStatus(event.toOutgoing(Status.SUCCESS)); + LOG.debug(String.format("Event %s finished successfully", event.getHeader().getRequestID())); + } else { + // Should never happen. Exception with message should be thrown instead. + LOG.error(String.format( + "We somehow returned false from doDG() instead of throwing exception. Incoming event [%s]", + event.toJson().toString())); + dmaap.postStatus(event.toOutgoing(Status.FAILURE)); + } + + } catch (Exception e) { + // Unknown exception from DG method. Fail and pass the exception + // along + String msg = "Exception: " + e.getMessage(); + LOG.warn(String.format("Event %s finished with failure. %s", event.getHeader().getRequestID(), msg)); + //TODO: should a message be included? there is nothing in the API spec for a msg? + //dmaap.postStatus(event.toOutgoing(Status.FAILURE, msg)); + dmaap.postStatus(event.toOutgoing(Status.FAILURE)); + } + + LOG.debug("Done working on " + event.getHeader().getRequestID()); + } + + private boolean doDG(IncomingMessage msg) throws APPCException { + return ProviderOperations.topologyDG(msg); + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Action.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Action.java new file mode 100644 index 000000000..9e7e923be --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Action.java @@ -0,0 +1,55 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL1607.model; + +public enum Action { + Restart("Restart"), Rebuild("Rebuild"), Migrate("Migrate"), Evacuate("Evacuate"), Snapshot("Snapshot"),modifyconfig("ModifyConfig"); + + /** + * Converts the string to an Action + * + * @param value + * The string to try and convert. Is case insensitive + * @return The action matching the string or null if no match was found. + */ + public static Action toAction(String value) { + if (value != null) { + for (Action e : values()) { + if (e.getValue().toUpperCase().equals(value.toUpperCase())) { + return e; + } + } + } + + return null; + } + + private String value; + + private Action(String valueToUse) { + value = valueToUse; + } + + public final String getValue() { + return value; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/CommonMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/CommonMessage.java new file mode 100644 index 000000000..80a78095b --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/CommonMessage.java @@ -0,0 +1,340 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL1607.model; + +import java.io.Serializable; +import java.util.Collection; + +import org.json.JSONObject; +import org.openecomp.appc.listener.util.Mapper; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import com.fasterxml.jackson.databind.node.ObjectNode; + +/** + * This class holds attributes that are common to DMaaP messages both coming in from DCAE and being sent out by APPC + * + */ +@JsonSerialize(include = Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class CommonMessage implements Serializable { + /* + * { "CommonHeader": { "TimeStamp": "0000-00-00T00:00:00.000Z", "APIver": "1.01", "OriginatorID": "policy.pdp01", + * "RequestID": "b74d13c5-bb26-4b04-992c-4679dfc8280e", "SubrequestID": "1" }, "Action": "RESTART", "Payload": { + * "VServerSelfLink": + * "http://192.168.1.2:8774/v2/abcde12345fghijk6789lmnopq123rst/servers/abc12345-1234-5678-890a-abcdefg12345", + * "VNF_NAME": "test", "VMID": "abc12345-1234-5678-890a-abcdefg12345", "TenantID": + * "abcde12345fghijk6789lmnopq123rst", "LOC_ID": "Test", "in-maint": "false", "Identity": + * "http://example.com:5000/v2.0", "Prov_status": "ACTIVE", "OAM_IPV4": "192.168.1.2", + * "is-closed-loop-disabled": "false", "VM_NAME": "basx0001vm034", "OAM_IPV6": "aaaa::bbbb:cccc:dddd:eeee/64" } } + */ + + private static final long serialVersionUID = 1L; + + /* + * The common header + */ + @JsonProperty("CommonHeader") + private CommonHeader header; + + /* + * The payload + */ + @JsonProperty("Payload") + private Payload payload; + + @JsonIgnore + private long startTime = System.currentTimeMillis(); + + /* + * Getters and Setters + */ + + public long getStartTime() { + return startTime; + } + + public void setStartTime(long startTime) { + this.startTime = startTime; + } + + /** + * @return the header + */ + public CommonHeader getHeader() { + return header; + } + + /** + * @param header + * the header to set + */ + public void setHeader(CommonHeader header) { + this.header = header; + } + + /** + * @return the payload + */ + public Payload getPayload() { + return payload; + } + + /** + * @param payload + * the payload to set + */ + public void setPayload(Payload payload) { + this.payload = payload; + } + + /** + * Convenience method to return a json representation of this object. + * + * @return The json representation of this object + */ + public JSONObject toJson() { + return Mapper.toJsonObject(this); + } + + @JsonIgnoreProperties(ignoreUnknown = true) + public static class CommonHeader { + /* + * "CommonHeader": { "TimeStamp": "2016-05-11T13:53:53.146Z", "APIver": "1.01", "OriginatorID": "policy.pdp01", + * "RequestID": "b74d13c5-bb26-4b04-992c-4679dfc8280e", "SubrequestID": "1" } + */ + + /* + * The timestamp of the message + */ + @JsonProperty("TimeStamp") + private String timeStamp; + + /* + * The API version of the message + */ + @JsonProperty("APIver") + private String apiVer; + + /* + * The Originator ID of the message + */ + @JsonProperty("OriginatorID") + private String originatorId; + + /* + * The Request Id of the message + */ + @JsonProperty("RequestID") + private String requestID; + + /* + * The Subrequest Id of the message + */ + @JsonProperty("SubrequestID") + private String subRequestId; + + /** + * @return the timeStamp + */ + public String getTimeStamp() { + return timeStamp; + } + + /** + * @param timeStamp + * the timeStamp to set + */ + public void setTimeStamp(String timeStamp) { + this.timeStamp = timeStamp; + } + + /** + * @return the apiVer + */ + public String getApiVer() { + return apiVer; + } + + /** + * @param apiVer + * the apiVer to set + */ + public void setApiVer(String apiVer) { + this.apiVer = apiVer; + } + + /** + * @return the originatorId + */ + public String getOriginatorId() { + return originatorId; + } + + /** + * @param originatorId + * the originatorId to set + */ + public void setOriginatorId(String originatorId) { + this.originatorId = originatorId; + } + + /** + * @return the requestID + */ + public String getRequestID() { + return requestID; + } + + /** + * @param requestID + * the requestID to set + */ + public void setRequestID(String requestID) { + this.requestID = requestID; + } + + /** + * @return the subRequestId + */ + public String getSubRequestId() { + return subRequestId; + } + + /** + * @param subRequestId + * the subRequestId to set + */ + public void setSubRequestId(String subRequestId) { + this.subRequestId = subRequestId; + } + }; + + @JsonIgnoreProperties(ignoreUnknown = true) + public static class Payload { + /* + * "Payload": { "VServerSelfLink": + * "http://192.168.1.2:8774/v2/abcde12345fghijk6789lmnopq123rst/servers/abc12345-1234-5678-890a-abcdefg12345", + * "VNF_NAME": "test", "VMID": "abc12345-1234-5678-890a-abcdefg12345", "TenantID": + * "abcde12345fghijk6789lmnopq123rst", "LOC_ID": "Test", "in-maint": "false", "Identity": + * "http://example.com:5000/v2.0", "Prov_status": "ACTIVE", "OAM_IPV4": "192.168.1.2", + * "is-closed-loop-disabled": "false", "VM_NAME": "test", "OAM_IPV6": "aaaa::bbbb:cccc:dddd:eeee/64" } + */ + + /* + * The TenantID of the message + */ + @JsonProperty("generic-vnf.vnf-id") + private String genericVnfId; + + /** + * @return the TenantID + */ + public String getGenericVnfId() { + return genericVnfId; + } + + /** + * @param TenantID + * the TenantID to set + */ + public void setGenericVnfId(String genericVnfId) { + this.genericVnfId = genericVnfId; + } + + @JsonProperty("pg-streams") + private pgStreams pgStreams; + + /** + * @return the TenantID + */ + + public String getPgStreams() { + String r = "{\\\"pg-streams\\\": {\\\"pg-stream\\\":["; + boolean first = true; + for(pgStream p : this.pgStreams.streams){ + String n = "{\\\"id\\\":\\\""+p.getId()+"\\\", \\\"is-enabled\\\":\\\""+p.getIsEnabled()+"\\\"}"; + if(!first){ + r = r.concat(","); + } + first = false; + r = r.concat(n); + } + r= r.concat("]}}"); + return r; + } + + /** + * @param TenantID + * the TenantID to set + */ + public void setPgStreams(pgStreams pgStreams) { + this.pgStreams = pgStreams; + } + + + + + + }; + @JsonIgnoreProperties(ignoreUnknown = true) + public static class pgStreams { + @JsonProperty("pg-stream") + private Collection<pgStream> streams; + + public Collection<pgStream> getStreams() { + return streams; + } + + public void setStreams(Collection<pgStream> streams) { + this.streams = streams; + } + + + }; + @JsonIgnoreProperties(ignoreUnknown = true) + public static class pgStream{ + public String getId() { + return id; + } + public void setId(String id) { + this.id = id; + } + public String getIsEnabled() { + return isEnabled; + } + public void setIsEnabled(String isEnabled) { + this.isEnabled = isEnabled; + } + @JsonProperty("id") + private String id; + @JsonProperty("is-enabled") + private String isEnabled; + }; + + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/IncomingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/IncomingMessage.java new file mode 100644 index 000000000..62d4d1b7d --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/IncomingMessage.java @@ -0,0 +1,88 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL1607.model; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion; + +/** + * This class represents a message coming in from DCAE. + * + */ +@JsonSerialize(include = Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class IncomingMessage extends CommonMessage { + + private static final long serialVersionUID = 1L; + + /* + * The action being requested. Its presence signals that it is an incoming message and it is not present on outgoing + * messages + */ + //TODO; use enum + @JsonProperty("Action") + private String action; + + + public String getRequest() { + return action; + } + + @JsonIgnore + public Action getAction() { + return Action.toAction(action); + } + + public void setRequest(String request) { + this.action = request; + } + +// @Override +// public String toString() { +// String time = getRequestTime() != null ? getRequestTime() : "N/A"; +// // String req = request != null ? request : "N/A"; +// return String.format("[%s - %s]", time, getId()); +// } + +// public String toOutgoing(Status status) { +// return toOutgoing(status); +// } + + public String toOutgoing(Status status) { + OutgoingMessage out = new OutgoingMessage(this); + out.setResponse(status); + return out.toResponse().toString(); + } + + /** + * Determines if this message should be parsed parsed. Will eventually check that the message is well formed, has + * all required fields, and had not exceeded any timing restrictions. + * + * @return True if the message should be parsed. False otherwise + */ + public boolean isValid() { + return true; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/OutgoingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/OutgoingMessage.java new file mode 100644 index 000000000..7b766212d --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/OutgoingMessage.java @@ -0,0 +1,192 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL1607.model; + +import java.net.InetAddress; +import java.security.SecureRandom; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.TimeZone; + +import org.json.JSONObject; +import org.openecomp.appc.listener.util.Mapper; +import org.openecomp.appc.util.Time; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize.Inclusion; + +/** + * This class represents a message being sent out to DMaaP by APPC to update listeners on the status of a request + * + */ +@JsonSerialize(include = Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class OutgoingMessage extends CommonMessage { + + public OutgoingMessage() { + + } + + public OutgoingMessage(IncomingMessage msg) { + setHeader(msg.getHeader()); + setPayload(msg.getPayload()); +// setId(msg.getId()); +// setOriginalRequest(msg.getRequest()); +// setRequestClient(msg.getRequestClient()); +// setRequestTime(msg.getRequestTime()); +// setVmName(msg.getVmName()); +// setFromSystem(generateFrom()); +// setResponse(Status.PENDING); +// setPolicyName(msg.getPolicyName()); +// setPolicyVersion(msg.getPolicyVersion()); +// setStartTime(msg.getStartTime()); + } + + private static final long serialVersionUID = -5447940920271469613L; + /* + * The status of the response + */ + @JsonProperty("Status") + private OutStatus status; + + /** + * @return the status + */ + public OutStatus getStatus() { + return status; + } + + /** + * @param status the status to set + */ + public void setStatus(OutStatus status) { + this.status = status; + } + + public void updateResponseTime() { + SecureRandom rand = new SecureRandom(); + SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd kk:mm:ss.SSS"); + df.setTimeZone(TimeZone.getTimeZone("UTC")); + String date = df.format(new Date(Time.utcTime())); + //this.responseTime = String.format("%s%03d", date, rand.nextInt(1000)); + } + + public String generateFrom() { + String name; + try { + InetAddress iAddress = InetAddress.getLocalHost(); + name = iAddress.getCanonicalHostName(); + } catch (Exception e) { + // Could not get anything from the InetAddress + name = "UnknownHost"; + } + return "appc@" + name; + } + + public JSONObject toResponse() { + updateResponseTime(); + JSONObject json = Mapper.toJsonObject(this); + + if (!json.has("message")) { + // If there is no message, parrot the status (response field) + // TODO - Can this be removed for 1602 making message truely optional? + //json.put("message", this.getResponse().toString()); + } + + // Removed duplication of status from message for 1602 + // json.put("message", String.format("%s: %s", request, json.get("message"))); + + return json; + } + +// @Override +// public String toString() { +// return String.format("%s - %s", getId(), getResponse()); +// } + + public static class OutStatus{ + @JsonProperty("Code") + private String code; + + @JsonProperty("Value") + private String value; + + /** + * @return the code + */ + public String getCode() { + return code; + } + + /** + * @param code the code to set + */ + public void setCode(String code) { + this.code = code; + } + + /** + * @return the value + */ + public String getValue() { + return value; + } + + /** + * @param value the value to set + */ + public void setValue(String value) { + this.value = value; + } + + } + + public void setResponse(Status newStatus) { + if(this.status == null){ + this.status = new OutStatus(); + } + + switch (newStatus){ + case ACCEPTED: + this.status.setValue(newStatus.getValue()); + this.status.setCode("100"); + break; + + case FAILURE: + this.status.setValue(newStatus.getValue()); + this.status.setCode("500"); + break; + + case SUCCESS: + this.status.setValue(newStatus.getValue()); + this.status.setCode("400"); + break; + default: + break; + + } + + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Status.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Status.java new file mode 100644 index 000000000..f7ec8febd --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/CL1607/model/Status.java @@ -0,0 +1,69 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.CL1607.model; + +public enum Status { + /* + * APP-C acknowledges that it has read the event off of the wire. This is the initial status of an OutgoingEvent + */ + ACCEPTED("ACCEPTED"), + + /* + * APP-C has finished processing the event without errors + */ + SUCCESS("SUCCESS"), + + /* + * APP-C has finished processing the event with errors + */ + FAILURE("FAILURE"); + + /** + * Converts the string to an Status + * + * @param value + * The string to try and convert. Is case insensitive + * @return The status matching the string or null if no match was found. + */ + public static Status toStatus(String value) { + if (value != null) { + for (Status e : values()) { + if (e.getValue().toUpperCase().equals(value.toUpperCase())) { + return e; + } + } + } + + return null; + } + + private String value; + + private Status(String valueToUse) { + value = valueToUse; + } + + public final String getValue() { + return value; + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Controller.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Controller.java new file mode 100644 index 000000000..66793e248 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Controller.java @@ -0,0 +1,55 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener; + +import java.util.Map; + +/** + * A controller is responsible for creating a listener for each ListenerProperties object that is passed in to it on + * instantiation. The controller will create a thread pool that will contain all of the listener threads so no listener + * can crash the controller. + * + */ +public interface Controller { + + /** + * Creates a new thread in the thread pool for an implementation of the {@see #Listener} class set in the + * ListenerProperties. This thread is run immediately after it is created. + */ + public void start(); + + /** + * Stops each of the listeners known by this controller. Takes an optional parameter that indicates the the listener + * should stop immediately rather than waiting for all threads to complete. + * + * @param stopNow + * Determines what method the listeners should use to shutdown. If true, listeners will use the stopNow() + * method. Otherwise they will use the stop() method. + */ + public void stop(boolean stopNow); + + /** + * @return A Map of ListenerProperties and the Listener object that is running in the controllers thread pool. + */ + public Map<ListenerProperties, Listener> getListeners(); + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/EventHandler.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/EventHandler.java new file mode 100644 index 000000000..5b2906f3c --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/EventHandler.java @@ -0,0 +1,185 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener; + +import java.util.Collection; +import java.util.List; +import java.util.Set; + +/** + * EventHandler defines a class that wraps DMaaP operations (most notably Get Message and Post Message) to make them + * easier to use. + * + */ +public interface EventHandler { + + /** + * Gets a list of messages as Strings on the read topic. + * + * @return A list of String messages. Never returns null. + */ + public List<String> getIncomingEvents(); + + /** + * Gets a list of messages as String on the read topic. + * + * @param limit + * The maximum amount of entries to return + * @return A list of String messages. Never returns null. + */ + public List<String> getIncomingEvents(int limit); + + /** + * Gets a list of messages Mapped to the given Class. If a message cannot be mapped to that class, it is discarded. + * + * @param cls + * The class to map the message to. + * @return A list of objects of the provided class. Never returns null. + */ + public <T> List<T> getIncomingEvents(Class<T> cls); + + /** + * Gets a list of messages Mapped to the given Class. If a message cannot be mapped to that class, it is discarded. + * + * @param cls + * The class to map the message to. + * @param limit + * The maximum amount of entries to return + * @return A list of objects of the provided class. Never returns null. + */ + public <T> List<T> getIncomingEvents(Class<T> cls, int limit); + + /** + * Posts the String message to the write topic(s). + * + * @param event + * The String to post. + */ + public void postStatus(String event); + + /** + * Posts the String message to the write topic(s) on the specified partition. Partitions are only used to guarantee + * ordering and do not impact if data is retreived. + * + * @param partition + * The partition to post to or null if no partition should be used. + * @param event + * The String to post. + */ + public void postStatus(String partition, String event); + + /** + * @return The client/group id used to read messages + */ + public String getClientId(); + + /** + * Set the client/group id used to read messages + * + * @param clientId + * The new clientId to use + */ + public void setClientId(String clientId); + + /** + * @return The client/group name to use. + */ + public String getClientName(); + + /** + * Set the client/group name used to read messages. + * + * @param clientName + * The new clientName to use + */ + public void setClientName(String clientName); + + /** + * @return The name of the topic to read from + */ + public String getReadTopic(); + + /** + * Set the name of the topic to read from. + * + * @param topic + * The new topic to read from + */ + public void setReadTopic(String topic); + + /** + * @return The name of the topic to write to + */ + public Set<String> getWriteTopics(); + + /** + * Set the name of the topic to write to + * + * @param topic + * The new topic to write to + */ + public void setWriteTopics(Set<String> topic); + + /** + * Adds a DMaaP host to the host pool + * + * @param host + * The host to add to the pool in <host>:<port> format + */ + public void addToPool(String host); + + /** + * Remove the host name from the pool if it exists + * + * @param host + * The host to add to the pool in <host>:<port> format + */ + public void removeFromPool(String host); + + /** + * Get all of the hosts in the DMaaP pool + * + * @return A collection of host in <host>:<port> format + */ + public Collection<String> getPool(); + + /** + * Clear any provided api credentials and make future requests as an unauthenticated user + */ + public void clearCredentials(); + + /** + * Set the api credentials and make future requests as an authenticated user + * + * @param access + * The access portion of the credentials (either user name or api key) + * @param secret + * The secret portion of the credentials (either password or api secret) + */ + public void setCredentials(String access, String secret); + + /** + * Close consumer/producer DMaaP clients + */ + public void closeClients(); + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/conv/Converter.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/conv/Converter.java new file mode 100644 index 000000000..021dfefc8 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/conv/Converter.java @@ -0,0 +1,85 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.conv; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.*; +import org.apache.commons.lang3.StringUtils; +import org.json.JSONObject; +import org.openecomp.appc.listener.LCM.model.DmaapMessage; +import org.openecomp.appc.listener.LCM.model.DmaapOutgoingMessage; +import org.openecomp.appc.listener.util.Mapper; + + +public class Converter { + + + public static DmaapOutgoingMessage convJsonNodeToDmaapOutgoingMessage(JsonNode inObj, String rpcName) { + DmaapOutgoingMessage outObj = new DmaapOutgoingMessage(); + outObj.setBody(inObj); + outObj.setRpcName(rpcName); + return outObj; + } + + public static String convDmaapOutgoingMessageToJsonString(DmaapMessage inObj) throws JsonProcessingException { +// return Mapper.toJsonString(inObj); + ObjectMapper objectMapper = new ObjectMapper(); + ObjectWriter writer = objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL).configure(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY,true) + .writer().withFeatures(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS); + return writer.writeValueAsString(inObj); + + } + + public static DmaapOutgoingMessage buildDmaapOutgoingMessageWithUnexpectedError(JsonNode dmaapInputBody, String rpcName,Exception inputException) { + DmaapOutgoingMessage dmaapOutgoingMessage = null; + String errMsg = StringUtils.isEmpty(inputException.getMessage())? inputException.toString() : inputException.getMessage(); + JSONObject commonHeaderJsonObject = Mapper.toJsonObject(dmaapInputBody.get("input").get("common-header")); + JSONObject jsonObjectOutput = new JSONObject().accumulate("common-header", commonHeaderJsonObject).accumulate("status", new JSONObject().accumulate("code",200).accumulate("value",errMsg)); + dmaapOutgoingMessage = new DmaapOutgoingMessage(); + dmaapOutgoingMessage.setRpcName(rpcName); + JSONObject jsonObjectBody = new JSONObject().accumulate("output",jsonObjectOutput); + JsonNode jsonNodeBody = Mapper.toJsonNodeFromJsonString(jsonObjectBody.toString()); + dmaapOutgoingMessage.setBody(jsonNodeBody); + return dmaapOutgoingMessage; + } + + public static String extractRequestIdWithSubId(JsonNode dmaapBody) { + String requestId; + //TODO: null pointer exception if dmaapBody is null, check if null or ensure is not null before calling + JsonNode commonHeaderJsonNode = dmaapBody.get("input").get("common-header"); + requestId = commonHeaderJsonNode.get("request-id").asText(); + requestId = requestId != null ? requestId : ""; + String subRequestId = commonHeaderJsonNode.get("sub-request-id").asText(); + if(!StringUtils.isEmpty(subRequestId)){ + requestId = requestId +"-"+subRequestId; + } + return requestId; + } + + public static Integer extractStatusCode(JsonNode event) { + Integer statusCode; + statusCode = event.get("output").get("status").get("code").asInt(); + return statusCode; + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/ListenerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/ListenerImpl.java new file mode 100644 index 000000000..dd877e98a --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/ListenerImpl.java @@ -0,0 +1,159 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.impl; + +import com.fasterxml.jackson.databind.JsonNode; +import org.apache.commons.lang3.StringUtils; +import org.openecomp.appc.listener.AbstractListener; +import org.openecomp.appc.listener.ListenerProperties; +import org.openecomp.appc.listener.LCM.conv.Converter; +import org.openecomp.appc.listener.LCM.model.DmaapIncomingMessage; +import org.openecomp.appc.listener.LCM.operation.ProviderOperations; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; +import com.att.eelf.i18n.EELFResourceManager; + +import java.text.DateFormat; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.List; +import java.util.TimeZone; +import java.util.concurrent.RejectedExecutionException; + +public class ListenerImpl extends AbstractListener { + + private final EELFLogger LOG = EELFManager.getInstance().getLogger(ListenerImpl.class); + + private long startTime = 0; + + public ListenerImpl(ListenerProperties props) { + super(props); + + String url = props.getProperty("provider.url"); + LOG.info("DMaaP Provider Endpoint: " + url); + ProviderOperations.setUrl(url); + + // Set Basic Auth + String user = props.getProperty("provider.user"); + String pass = props.getProperty("provider.pass"); + ProviderOperations.setAuthentication(user, pass); + } + + @Override + public void run() { + // Some vars for benchmarking + startTime = System.currentTimeMillis(); + + LOG.info("Running DMaaP Listener"); + + while (run.get()) { + // Only update if the queue is low. otherwise we read in more + // messages than we need + try { + if (executor.getQueue().size() <= QUEUED_MIN) { + LOG.debug("DMaaP queue running low. Querying for more jobs"); + + + List<DmaapIncomingMessage> messages = dmaap.getIncomingEvents(DmaapIncomingMessage.class, QUEUED_MAX); + LOG.debug(String.format("Read %d messages from dmaap", messages.size())); + for (DmaapIncomingMessage incoming : messages) { + // Acknowledge that we read the event + if (isValid(incoming)) { + String requestIdWithSubId = getRequestIdWithSubId(incoming.getBody()); + LOG.info("Acknowledging Message: " + requestIdWithSubId); +// dmaap.postStatus(incoming.toOutgoing(OperationStatus.PENDING)); + } + } + for (DmaapIncomingMessage incoming : messages) { + String requestIdWithSubId = getRequestIdWithSubId(incoming.getBody()); + // Add to pool if still running + if (run.get()) { + if (isValid(incoming)) { + LOG.info(String.format("Adding DMaaP message to pool queue [%s]", requestIdWithSubId)); + try { + executor.execute(new WorkerImpl(incoming.getRpcName(),incoming.getBody(), dmaap)); + } catch (RejectedExecutionException rejectEx) { + LOG.error("Task Rejected: ", rejectEx); + } + } else { + // Badly formed message + LOG.error("Message was not valid. Rejecting message: "+incoming); + } + } else { + if (isValid(incoming)) { + LOG.info("Run stopped. Orphaning Message: " + requestIdWithSubId); + } + else { + // Badly formed message + LOG.error("Message was not valid. Rejecting message: "+incoming); + } + } + } + } + } catch (Exception e) { + LOG.error("Exception " + e.getClass().getSimpleName() + " caught in DMaaP listener"); + LOG.error(EELFResourceManager.format(e)); + LOG.error("DMaaP Listener logging and ignoring the exception, continue..."); + } + } + + LOG.info("Stopping DMaaP Listener thread"); + + // We've told the listener to stop + // TODO - Should we: + // 1) Put a message back on the queue indicating that APP-C never got to + // the message + // or + // 2) Let downstream figure it out after timeout between PENDING and + // ACTIVE messages + } + + private boolean isValid(DmaapIncomingMessage incoming) { + return ((incoming != null) && + incoming.getBody() != null + && !StringUtils.isEmpty(incoming.getRpcName())); + } + + @Override + public String getBenchmark() { + long time = System.currentTimeMillis(); + DateFormat df = new SimpleDateFormat("HH:mm:ss"); + df.setTimeZone(TimeZone.getTimeZone("UTC")); + String runningTime = df.format(new Date(time - startTime)); + + String out = String.format("Running for %s and completed %d jobs using %d threads.", runningTime, + executor.getCompletedTaskCount(), executor.getPoolSize()); + LOG.info("***BENCHMARK*** " + out); + return out; + } + + private String getRequestIdWithSubId(JsonNode event){ + String requestId = ""; + try { + requestId = Converter.extractRequestIdWithSubId(event); + } catch (Exception e) { + LOG.error("failed to parse request-id and sub-request-id. Json not in expected format", e); + } + return requestId; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/WorkerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/WorkerImpl.java new file mode 100644 index 000000000..af8334ee4 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/impl/WorkerImpl.java @@ -0,0 +1,120 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.impl; + +import org.openecomp.appc.exceptions.APPCException; +import org.openecomp.appc.listener.EventHandler; +import org.openecomp.appc.listener.LCM.conv.Converter; +import org.openecomp.appc.listener.LCM.model.DmaapOutgoingMessage; +import org.openecomp.appc.listener.LCM.operation.ProviderOperations; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; + +public class WorkerImpl implements Runnable { + + private final EELFLogger LOG = EELFManager.getInstance().getLogger(WorkerImpl.class); + + // Should have all of the data we need for processing + private JsonNode event; + + // So we can post messages from inside the worker. + private EventHandler dmaap; + private String rpcName; + + public WorkerImpl(String rpcName,JsonNode message, EventHandler dmaap) { + this.rpcName = rpcName; + this.event = message; + this.dmaap = dmaap; + } + + @Override + public void run() { + String requestIdWithSubId = extractRequestIdWithSubId(event); + LOG.debug(String.format("Started working on %s", requestIdWithSubId)); + + // Run the dg in a try catch to handle all exceptions and update the + // message at the end + try { + JsonNode outputJsonNode = doDG(rpcName, event); + DmaapOutgoingMessage dmaapOutgoingMessage= Converter.convJsonNodeToDmaapOutgoingMessage(outputJsonNode,rpcName); + postMessageToDMaaP(dmaapOutgoingMessage,requestIdWithSubId); + Integer statusCode = extractStatusCode(dmaapOutgoingMessage.getBody()); + if (ProviderOperations.isSucceeded(statusCode)) { + LOG.debug(String.format("Event %s finished successfully", requestIdWithSubId)); + } else { + LOG.warn(String.format("Event %s failed", requestIdWithSubId)); + } + + } catch (Exception e) { + // Unknown exception from DG method. Fail and pass the exception + // along + String msg = "Exception: " + e.getMessage(); + LOG.error(String.format("Event %s finished with failure. %s", requestIdWithSubId, msg)); + DmaapOutgoingMessage dmaapOutgoingMessage= Converter.buildDmaapOutgoingMessageWithUnexpectedError(event,rpcName,e); + postMessageToDMaaP(dmaapOutgoingMessage,requestIdWithSubId); + } + + LOG.debug("Done working on " + requestIdWithSubId); + } + + + private Integer extractStatusCode(JsonNode event) { + Integer statusCode = null; + try { + statusCode = Converter.extractStatusCode(event); + } catch (Exception e) { + LOG.error("failed to parse statusCode. Json not in expected format", e); + } + return statusCode; + } + + + private String extractRequestIdWithSubId(JsonNode event){ + String requestId = ""; + try { + requestId = Converter.extractRequestIdWithSubId(event); + } catch (Exception e) { + LOG.error("failed to parse request-id and sub-request-id. Json not in expected format", e); + } + return requestId; + } + + + + private void postMessageToDMaaP(DmaapOutgoingMessage dmaapOutgoingMessage,String requestIdWithSubId) { + String dmaapOutgoingMessageJsonString; + try { + dmaapOutgoingMessageJsonString = Converter.convDmaapOutgoingMessageToJsonString(dmaapOutgoingMessage); + dmaap.postStatus(dmaapOutgoingMessage.getCambriaPartition(),dmaapOutgoingMessageJsonString); + } catch (JsonProcessingException e) { + LOG.error("failed to postMessageToDMaaP requestIdWithSubId: "+requestIdWithSubId+" dmaapOutgoingMessage: "+dmaapOutgoingMessage, e); + } + } + + private JsonNode doDG(String rpcName, JsonNode msg) throws APPCException { + return ProviderOperations.topologyDG(rpcName,msg); + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ActionIdentifiers.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ActionIdentifiers.java new file mode 100644 index 000000000..f17b18147 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ActionIdentifiers.java @@ -0,0 +1,81 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.model; + +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.io.Serializable; + + +public class ActionIdentifiers implements Serializable { + + @JsonProperty("service-instance-id") + private String serviceInstanceId; + @JsonProperty("vnf-id") + private String vnfID; + @JsonProperty("vnfc-name") + private String vnfcName; + @JsonProperty("vserver-id") + private String vserverId; + + public ActionIdentifiers() { + } + + public ActionIdentifiers(ActionIdentifiers actionIdentifiers) { + this.serviceInstanceId=actionIdentifiers.getServiceInstanceId(); + this.vnfID=actionIdentifiers.getVnfID(); + this.vnfcName=actionIdentifiers.getVnfcName(); + this.vserverId=actionIdentifiers.getVserverId(); + } + + public String getServiceInstanceId() { + return serviceInstanceId; + } + + public void setServiceInstanceId(String serviceInstanceId) { + this.serviceInstanceId = serviceInstanceId; + } + + public String getVnfID() { + return vnfID; + } + + public void setVnfID(String vnfID) { + this.vnfID = vnfID; + } + + public String getVnfcName() { + return vnfcName; + } + + public void setVnfcName(String vnfcName) { + this.vnfcName = vnfcName; + } + + public String getVserverId() { + return vserverId; + } + + public void setVserverId(String vserverId) { + this.vserverId = vserverId; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/CommonHeader.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/CommonHeader.java new file mode 100644 index 000000000..363c6407d --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/CommonHeader.java @@ -0,0 +1,112 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.model; + +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.io.Serializable; +import java.text.DateFormat; +import java.text.SimpleDateFormat; +import java.util.Map; + +import org.openecomp.appc.util.Time; + +public class CommonHeader implements Serializable { + @JsonProperty("timestamp") + private String timeStamp; + @JsonProperty("api-ver") + private String apiVer; + @JsonProperty("originator-id") + private String originatorId; + @JsonProperty("request-id") + private String requestID; + @JsonProperty("sub-request-id") + private String subRequestId; + @JsonProperty("flags") + private Map<String, String> flags; + + private static final DateFormat ZULU_FORMATTER = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SS'Z'"); + + public CommonHeader() { + } + + public CommonHeader(CommonHeader commonHeader) { + // changed to current system time + timeStamp = ZULU_FORMATTER.format(Time.utcDate()); + + apiVer = commonHeader.getApiVer(); + originatorId = commonHeader.getOriginatorId(); + requestID = commonHeader.getRequestID(); + subRequestId = commonHeader.getSubRequestId(); + flags = commonHeader.getFlags(); + } + + + + public String getTimeStamp() { + return timeStamp; + } + + public void setTimeStamp(String timeStamp) { + this.timeStamp = timeStamp; + } + + public String getApiVer() { + return apiVer; + } + + public void setApiVer(String apiVer) { + this.apiVer = apiVer; + } + + public String getRequestID() { + return requestID; + } + + public void setRequestID(String requestID) { + this.requestID = requestID; + } + + public String getOriginatorId() { + return originatorId; + } + + public void setOriginatorId(String originatorId) { + this.originatorId = originatorId; + } + + public String getSubRequestId() { + return subRequestId; + } + + public void setSubRequestId(String subRequestId) { + this.subRequestId = subRequestId; + } + + public Map<String, String> getFlags() { + return flags; + } + + public void setFlags(Map<String, String> flags) { + this.flags = flags; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapIncomingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapIncomingMessage.java new file mode 100644 index 000000000..ec1e9e767 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapIncomingMessage.java @@ -0,0 +1,50 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.model; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import org.json.JSONObject; +import org.openecomp.appc.listener.util.Mapper; + +/** + * This class represents a message being sent out to DMaaP by APPC to update listeners on the status of a request + * + */ +@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class DmaapIncomingMessage extends DmaapMessage{ + private final static String defaultCambriaPartition = "APP-C"; + + public DmaapIncomingMessage() { + super(); + setCambriaPartition(defaultCambriaPartition); + } + + + @Override + public String toString() { + return "DmaapIncomingMessage{"+super.toString()+"}"; + } +} + diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapMessage.java new file mode 100644 index 000000000..56099cb7e --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapMessage.java @@ -0,0 +1,79 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.model; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.JsonNode; + + +@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class DmaapMessage{ + + @JsonProperty("cambria.partition") + private String cambriaPartition; + + @JsonProperty("rpc-name") + private String rpcName; + + @JsonProperty("body") + private JsonNode body; + + public DmaapMessage() { + } + + public String getCambriaPartition() { + return cambriaPartition; + } + + public void setCambriaPartition(String cambriaPartition) { + this.cambriaPartition = cambriaPartition; + } + + public String getRpcName() { + return rpcName; + } + + public void setRpcName(String rpcName) { + this.rpcName = rpcName; + } + + public JsonNode getBody() { + return body; + } + + public void setBody(JsonNode body) { + this.body = body; + } + + @Override + public String toString() { + return "DmaapMessage{" + + "cambriaPartition='" + cambriaPartition + '\'' + + ", rpcName='" + rpcName + '\'' + + ", body=" + body + + '}'; + } +} + diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapOutgoingMessage.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapOutgoingMessage.java new file mode 100644 index 000000000..6d7cb997e --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/DmaapOutgoingMessage.java @@ -0,0 +1,50 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.model; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import org.json.JSONObject; +import org.openecomp.appc.listener.util.Mapper; + +/** + * This class represents a message being sent out to DMaaP by APPC to update listeners on the status of a request + * + */ +@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class DmaapOutgoingMessage extends DmaapMessage{ + private final static String defaultCambriaPartition = "MSO"; + + public DmaapOutgoingMessage() { + super(); + setCambriaPartition(defaultCambriaPartition); + } + + + @Override + public String toString() { + return "DmaapOutgoingMessage{"+super.toString()+"}"; + } +} + diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/InputBody.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/InputBody.java new file mode 100644 index 000000000..6fd0572f3 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/InputBody.java @@ -0,0 +1,102 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.model; + +import org.openecomp.appc.listener.LCM.operation.OperationStatus; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; + + +@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class InputBody { + + @JsonProperty("common-header") + private CommonHeader commonHeader; + + @JsonProperty("action-identifiers") + private ActionIdentifiers actionIdentifiers; + + @JsonProperty("action") + private String action; + + @JsonProperty("payload") + private Object payload; + + + public Object getPayload() { + return payload; + } + + public void setPayload(Object payload) { + this.payload = payload; + } + @JsonIgnore + public void setPayloadAsString(String payload) { + this.payload = payload; + } + + public String getAction() { + return action; + } + + public void setAction(String action) { + this.action = action; + } + + public ActionIdentifiers getActionIdentifiers() { + return actionIdentifiers; + } + + public void setActionIdentifiers(ActionIdentifiers actionIdentifiers) { + this.actionIdentifiers = actionIdentifiers; + } + + public CommonHeader getCommonHeader() { + return commonHeader; + } + + public void setCommonHeader(CommonHeader commonHeader) { + this.commonHeader = commonHeader; + } + +// public String toOutgoing(OperationStatus operationStatus) { +// OutputBody out = new OutputBody(this); +// out.setStatus(new ResponseStatus(operationStatus.getCode(), operationStatus.getValue())); +// return out.toResponse().toString(); +// } + +// public String toOutgoing(OperationStatus operationStatus,String islocked) { +// OutputBody out = new OutputBody(this); +// out.setStatus(new ResponseStatus(operationStatus.getCode(), operationStatus.getValue())); +// out.setLocked(islocked); +// return out.toResponse().toString(); +// } + + @JsonIgnore + public boolean isValid() { + return getCommonHeader() != null; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/OutputBody.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/OutputBody.java new file mode 100644 index 000000000..54b407b5f --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/OutputBody.java @@ -0,0 +1,98 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.model; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import org.json.JSONObject; +import org.openecomp.appc.listener.util.Mapper; + +/** + * This class represents a message being sent out to DMaaP by APPC to update listeners on the status of a request + * + */ +@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class OutputBody { + + @JsonProperty("common-header") + private CommonHeader header; + + @JsonProperty("status") + private ResponseStatus status; + + @JsonProperty("payload") + private String payload; + + @JsonProperty("locked") + private String locked; + + public OutputBody() { + } + + + public JSONObject toResponse() { + return Mapper.toJsonObject(this); + } + + public OutputBody(InputBody msg) { + this.header = new CommonHeader(msg.getCommonHeader()); + } + + public String getLocked() { + return locked; + } + + public void setLocked(String locked) { + this.locked = locked; + } + public String getPayload() { + return payload; + } + + public void setPayload(String payload) { + this.payload = payload; + } + + public CommonHeader getHeader() { + return header; + } + + public void setHeader(CommonHeader header) { + this.header = header; + } + + public ResponseStatus getStatus() { + return status; + } + + public void setStatus(ResponseStatus status) { + this.status = status; + } + + @Override + public String toString() { + return String.format("%s - %s(%s)", getHeader().getRequestID(), getStatus().getCode(), getStatus().getValue()); + } +} + diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ResponseStatus.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ResponseStatus.java new file mode 100644 index 000000000..9186c444a --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/model/ResponseStatus.java @@ -0,0 +1,70 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.model; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; + + +@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL) +@JsonIgnoreProperties(ignoreUnknown = true) +public class ResponseStatus { + @JsonProperty("code") + private Integer code; + @JsonProperty("message") + private String value; + + public ResponseStatus() { + } + + public ResponseStatus(Integer code, String value) { + this.code = code; + this.value = value; + } + + public Integer getCode() { + return code; + } + + public void setCode(Integer code) { + code = code; + } + + public String getValue() { + return value; + } + + + + public void setValue(String value) { + value = value; + } + + @Override + public String toString() { + return "ResponseStatus{" + + "code=" + code + + ", value='" + value + '\'' + + '}'; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/GenericProviderOperationRequestFormatter.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/GenericProviderOperationRequestFormatter.java new file mode 100644 index 000000000..5c54845cc --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/GenericProviderOperationRequestFormatter.java @@ -0,0 +1,115 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.operation; + +import com.fasterxml.jackson.databind.JsonNode; +import org.json.JSONObject; +import org.openecomp.appc.exceptions.APPCException; +import org.openecomp.appc.listener.LCM.model.InputBody; +import org.openecomp.appc.listener.LCM.model.ResponseStatus; +import org.openecomp.appc.listener.util.Mapper; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; + +import java.net.URL; + + + +public class GenericProviderOperationRequestFormatter implements ProviderOperationRequestFormatter { + + private final EELFLogger LOG = EELFManager.getInstance().getLogger(GenericProviderOperationRequestFormatter.class); + + //@formatter:off + @SuppressWarnings("nls") + private final static String TEMPLATE = "{\"input\": %s}"; + //@formatter:on + + @Override + public String buildPath(URL url, String rpcName) { + return url.getPath() + ":"+rpcName; + } + +// private String convertActionToUrl(String action) { +// String res = ""; +// switch (action) { +// case "SoftwareUpload": +// res = "software-upload"; +// break; +// case "LiveUpgrade": +// res = "live-upgrade"; +// break; +// case "HealthCheck": +// res = "health-check"; +// break; +// case "CheckLock": +// res = "check-lock"; +// break; +// default : +// res = action; +// } +// return res.toLowerCase(); +// } + + @Override + public String buildRequest(InputBody msg) { + JSONObject jsonObject = Mapper.toJsonObject(msg); + return String.format(TEMPLATE, jsonObject.toString()); + } + +/* @Override + public OperationStatus getOperationStatus(JSONObject responseBody) throws APPCException { + try { + JSONObject status = responseBody.getJSONObject("output").getJSONObject("status"); + return new OperationStatus(String.valueOf(status.getInt("code")), status.getString("message")); + } catch (Exception e) { + LOG.error("Unknown error processing failed response from provider. Json not in expected format", e); + throw new APPCException("APPC has an unknown RPC error"); + } + }*/ + + @Override + public ResponseStatus getResponseStatus(JsonNode responseBody)throws APPCException{ + try { + JsonNode status = responseBody.get("output").get("status"); + return new ResponseStatus(status.get("code").asInt(), status.get("message").asText()); + } catch (Exception e) { + LOG.error("Unknown error processing failed response from provider. Json not in expected format", e); + throw new APPCException("APPC has an unknown RPC error"); + } + } + + @Override + public String getLocked(JSONObject responseBody) throws APPCException { + try { + JSONObject outputObject=responseBody.getJSONObject("output"); + if(outputObject.has("locked")){ + return outputObject.getString("locked"); + }else{ + return null; + } + } catch (Exception e) { + LOG.error("Unknown error processing failed response from provider. Json not in expected format", e); + throw new APPCException("APPC has an unknown RPC error"); + } + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/OperationStatus.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/OperationStatus.java new file mode 100644 index 000000000..b2388af45 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/OperationStatus.java @@ -0,0 +1,66 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.operation; + + +public class OperationStatus { + public final static OperationStatus PENDING = new OperationStatus("PENDING", "PENDING"); + public final static OperationStatus ACTIVE = new OperationStatus("ACTIVE", "ACTIVE"); + public final static OperationStatus SUCCESS = new OperationStatus("SUCCESS", "SUCCESS"); + public final static OperationStatus FAILURE = new OperationStatus("FAILURE", "FAILURE"); + + private String code; + private String value; + + public OperationStatus() { + } + + + public OperationStatus(String code, String value) { + this.code = code; + this.value = value; + } + + public String getCode() { + return code; + } + + public void setCode(String code) { + this.code = code; + } + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = value; + } + + public boolean isSucceeded() { + if (code == null) { + return false; + } + int intCode = Integer.parseInt(code); + return (intCode >= 200) && (intCode < 300); // All 2xx statuses are success + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperationRequestFormatter.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperationRequestFormatter.java new file mode 100644 index 000000000..33aa5b673 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperationRequestFormatter.java @@ -0,0 +1,42 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.operation; + +import com.fasterxml.jackson.databind.JsonNode; +import org.json.JSONObject; +import org.openecomp.appc.exceptions.APPCException; +import org.openecomp.appc.listener.LCM.model.InputBody; +import org.openecomp.appc.listener.LCM.model.ResponseStatus; + +import java.net.URL; + + +public interface ProviderOperationRequestFormatter { + + String buildPath(URL url, String rpcName); + + String buildRequest(InputBody msg); + + ResponseStatus getResponseStatus(JsonNode json)throws APPCException; + + String getLocked(JSONObject json)throws APPCException; +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperations.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperations.java new file mode 100644 index 000000000..f5672a575 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/LCM/operation/ProviderOperations.java @@ -0,0 +1,260 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.LCM.operation; + +import com.fasterxml.jackson.databind.JsonNode; +import org.apache.commons.codec.binary.Base64; +import org.apache.commons.io.IOUtils; +import org.apache.http.HttpHeaders; +import org.apache.http.HttpResponse; +import org.apache.http.HttpVersion; +import org.apache.http.client.HttpClient; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.conn.ClientConnectionManager; +import org.apache.http.conn.scheme.PlainSocketFactory; +import org.apache.http.conn.scheme.Scheme; +import org.apache.http.conn.scheme.SchemeRegistry; +import org.apache.http.conn.ssl.SSLSocketFactory; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.DefaultHttpClient; +import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager; +import org.apache.http.params.BasicHttpParams; +import org.apache.http.params.HttpParams; +import org.apache.http.params.HttpProtocolParams; +import org.apache.http.protocol.HTTP; +import org.openecomp.appc.exceptions.APPCException; +import org.openecomp.appc.listener.LCM.model.ResponseStatus; +import org.openecomp.appc.listener.util.Mapper; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; + +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; +import javax.net.ssl.X509TrustManager; +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.net.MalformedURLException; +import java.net.Socket; +import java.net.URL; +import java.net.UnknownHostException; +import java.security.*; +import java.security.cert.CertificateException; +import java.security.cert.X509Certificate; + +public class ProviderOperations { + + private static final EELFLogger LOG = EELFManager.getInstance().getLogger(ProviderOperations.class); + + private static URL url; + private static String basic_auth; + + private static ProviderOperationRequestFormatter requestFormatter = new GenericProviderOperationRequestFormatter(); + + /** + * Calls the AppcProvider to run a topology directed graph + * + * @param msg The incoming message to be run + * @return True if the result is success. Never returns false and throws an exception instead. + * @throws UnsupportedEncodingException + * @throws Exception if there was a failure processing the request. The exception message is the failure reason. + */ + @SuppressWarnings("nls") + public static JsonNode topologyDG(String rpcName, JsonNode msg) throws APPCException { + if (msg == null) { + throw new APPCException("Provided message was null"); + } + + HttpPost post = null; + try { + + // Concatenate the "action" on the end of the URL + String path = requestFormatter.buildPath(url, rpcName); + URL serviceUrl = new URL(url.getProtocol(), url.getHost(), url.getPort(), path); + + post = new HttpPost(serviceUrl.toExternalForm()); + post.setHeader(HttpHeaders.CONTENT_TYPE, "application/json"); + post.setHeader(HttpHeaders.ACCEPT, "application/json"); + + // Set Auth + if (basic_auth != null) { + post.setHeader(HttpHeaders.AUTHORIZATION, "Basic " + basic_auth); + } + + String body = Mapper.toJsonString(msg); + StringEntity entity = new StringEntity(body); + entity.setContentType("application/json"); + post.setEntity(new StringEntity(body)); + } catch (UnsupportedEncodingException | MalformedURLException e) { + throw new APPCException(e); + } + + HttpClient client = getHttpClient(); + + int httpCode = 0; + String respBody = null; + try { + HttpResponse response = client.execute(post); + httpCode = response.getStatusLine().getStatusCode(); + respBody = IOUtils.toString(response.getEntity().getContent()); + } catch (IOException e) { + throw new APPCException(e); + } + + if (httpCode >= 200 && httpCode < 300 && respBody != null) { + JsonNode json; + try { + json = Mapper.toJsonNodeFromJsonString(respBody); + } catch (Exception e) { + LOG.error("Error processing response from provider. Could not map response to json", e); + throw new APPCException("APPC has an unknown RPC error"); + } + + ResponseStatus responseStatus = requestFormatter.getResponseStatus(json); + + if (!isSucceeded(responseStatus.getCode())) { + LOG.warn(String.format("Operation failed [%s]", msg.toString())); + } + + return json; + } + + throw new APPCException(String.format("Unexpected response from endpoint: [%d] - %s ", httpCode, respBody)); + } + + /** + * Updates the static var URL and returns the value; + * + * @return The new value of URL + */ + public static String getUrl() { + return url.toExternalForm(); + } + + public static void setUrl(String newUrl) { + try { + url = new URL(newUrl); + } catch (MalformedURLException e) { + e.printStackTrace(); + } + } + + /** + * Sets the basic authentication header for the given user and password. If either entry is null then set basic auth + * to null + * + * @param user The user with optional domain name + * @param password The password for the user + * @return The new value of the basic auth string that will be used in the request headers + */ + public static String setAuthentication(String user, String password) { + if (user != null && password != null) { + String authStr = user + ":" + password; + basic_auth = new String(Base64.encodeBase64(authStr.getBytes())); + } else { + basic_auth = null; + } + return basic_auth; + } + + @SuppressWarnings("deprecation") + private static HttpClient getHttpClient() throws APPCException { + HttpClient client; + if (url.getProtocol().equals("https")) { + try { + KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType()); + trustStore.load(null, null); + MySSLSocketFactory sf = new MySSLSocketFactory(trustStore); + sf.setHostnameVerifier(MySSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER); + + HttpParams params = new BasicHttpParams(); + HttpProtocolParams.setVersion(params, HttpVersion.HTTP_1_1); + HttpProtocolParams.setContentCharset(params, HTTP.UTF_8); + + SchemeRegistry registry = new SchemeRegistry(); + registry.register(new Scheme("http", PlainSocketFactory.getSocketFactory(), 80)); + registry.register(new Scheme("https", sf, 443)); + registry.register(new Scheme("https", sf, 8443)); + registry.register(new Scheme("http", sf, 8181)); + + ClientConnectionManager ccm = new ThreadSafeClientConnManager(params, registry); + client = new DefaultHttpClient(ccm, params); + } catch (Exception e) { + client = new DefaultHttpClient(); + } + } else if (url.getProtocol().equals("http")) { + client = new DefaultHttpClient(); + } else { + throw new APPCException( + "The provider.topology.url property is invalid. The url did not start with http[s]"); + } + return client; + } + + @SuppressWarnings("deprecation") + public static class MySSLSocketFactory extends SSLSocketFactory { + private SSLContext sslContext = SSLContext.getInstance("TLS"); + + public MySSLSocketFactory(KeyStore truststore) throws NoSuchAlgorithmException, KeyManagementException, + KeyStoreException, UnrecoverableKeyException { + super(truststore); + + TrustManager tm = new X509TrustManager() { + @Override + public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException { + } + + @Override + public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException { + } + + @Override + public X509Certificate[] getAcceptedIssuers() { + return null; + } + }; + + sslContext.init(null, new TrustManager[]{ + tm + }, null); + } + + @Override + public Socket createSocket(Socket socket, String host, int port, boolean autoClose) + throws IOException, UnknownHostException { + return sslContext.getSocketFactory().createSocket(socket, host, port, autoClose); + } + + @Override + public Socket createSocket() throws IOException { + return sslContext.getSocketFactory().createSocket(); + } + } + + public static boolean isSucceeded(Integer code) { + if (code == null) { + return false; + } + return ((code == 100) || (code == 400)); // only 100 & 400 statuses are success + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Listener.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Listener.java new file mode 100644 index 000000000..a8e180404 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/Listener.java @@ -0,0 +1,72 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener; + +/** + * This interface defines a listener that subscribes to a DMaaP topic and continually polls for messages. The + * listener does all operations in the run() method and long running operations should be created in a separate worker + * thread. + * + */ +public interface Listener extends Runnable { + + /** + * Should start a continuous poll to get messages from the message bus only ending when stop() or stopNow() are + * called. + * + * @see java.lang.Runnable#run() + */ + @Override + public void run(); + + /** + * Signals the listener to stop accepting new messages to the queue and to cleanly finish processing all remaining + * messages in the queue. This can take a significant amount of time to complete depending on the thread pool + * characteristics. Similar to {@link #stopNow()} + */ + public void stop(); + + /** + * Signals the listener to stop accepting new messages to the queue and to destroy all remaining messages in the + * queue. This will complete quicker than {@link #stop()} at the cost of discarded requests. Recovery of these + * requests would have to be caught downstream. Similar to {@link #stop()} + */ + public void stopNow(); + + /** + * @return A string that shows various benchmarking data. Can be used by humans to tune the thread pool. + */ + public String getBenchmark(); + + /** + * @return The listener's id when requesting messages from DMaaP. Also known as the group id. + */ + public String getListenerId(); + + /** + * Sets the listener's id to use when requesting messages from DMaaP. Also known as the group id. + * + * @param idString + * The new listener id + */ + public void setListenerId(String idString); +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/ListenerProperties.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/ListenerProperties.java new file mode 100644 index 000000000..c6c374eb0 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/ListenerProperties.java @@ -0,0 +1,319 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener; + +import java.util.Properties; + +/** + * A class for instantiating Listener objects. It is primarily used to hold properties that start with the given prefix. + * It also holds a class that implements {@see Listener} and will be used by the controller to spawn a new listener + * object. + * + * @since Apr 25, 2016 + * @version $Id$ + */ +public class ListenerProperties { + + private String prefix; + + private Class<? extends Listener> listenerClass; + + private Properties props; + + /** + * Creates a new listener object with the given prefix and properties. Any property starting with the prefix is + * added to the internal properties object with the prefix removed. All other properties are ignored. + * ListenerProperties constructor + * + * @param prefix + * The prefix of the properties to load + * @param allProps + * The properties object to load from. + */ + public ListenerProperties(String prefix, Properties allProps) { + this.prefix = prefix; + props = new Properties(); + + String dottedPrefix = String.format("%s.", prefix); + for (String key : allProps.stringPropertyNames()) { + if (key.startsWith(dottedPrefix) && key.length() > dottedPrefix.length()) { + props.put(key.substring(dottedPrefix.length()), allProps.get(key)); + } + } + } + + /** + * @return The prefix of these properties + */ + public String getPrefix() { + return prefix; + } + + /** + * Sets the listener class. Will be used by {@see Controller} to instantiate the Listener thread for this object + * + * @param cls + * The class to be created. Implements {@see Listener} + */ + public void setListenerClass(Class<? extends Listener> cls) { + this.listenerClass = cls; + } + + /** + * @return The class that will be used by {@see Controller} to instantiate the Listener thread for this object + */ + public Class<? extends Listener> getListenerClass() { + return listenerClass; + } + + /** + * Returns a property matching a given KEYS + * + * @param key + * The KEYS object who's value to return. + * @return The value of the property or null if none exists + */ + public String getProperty(KEYS key) { + return getProperty(key, null); + } + + /** + * Returns a property matching a given string. + * + * @param key + * The key who's value to return. + * @return The value of the property or null if none exists + */ + public String getProperty(String key) { + return getProperty(key, null); + } + + /** + * Returns a property matching a given KEYS + * + * @param key + * The KEYS object who's value to return. + * @param defaultValue + * The value to return if the property is not found + * @return The value of the property or null if none exists + */ + public String getProperty(KEYS key, String defaultValue) { + return getProperty(key.getPropertySuffix(), defaultValue); + } + + /** + * Returns a property matching a given string. + * + * @param key + * The key who's value to return. + * @param defaultValue + * The value to return if the property is not found + * @return The value of the property or null if none exists + */ + public String getProperty(String key, String defaultValue) { + return props.getProperty(key, defaultValue); + } + + /** + * @return The properties object containing all properties + */ + public Properties getProperties() { + return props; + } + + /** + * Reads the <i>prefix</i>.disabled property to determine if the listener is disabled and should not be run by the + * controller. Defaults to false if property not set or value cannot be parsed. + * + * @return true if the listener is disabled and should not be started. false if the listener should be start + * normally (default). + */ + public boolean isDisabled() { + return Boolean.valueOf(getProperty(KEYS.DISABLED, "false")); + } + + @Override + public String toString() { + return String.format("%s", prefix); + } + + /** + * The message service types that are available. Choices are DMaaP and DMaaP. + * + * @since Apr 25, 2016 + * @version $Id$ + */ + public enum MessageService { + DMaaP("dmaap"); + + private String val; + + private MessageService(String val) { + this.val = val; + } + + public String getValue() { + return val; + } + + /** + * Tries to match a string to a MessageService. If no match is found, returns the default (DMaaP) + * + * @param input + * the string to try and match + * @return A MessasgeService + */ + public static MessageService parse(String input) { + if (input != null) { + for (MessageService ms : MessageService.values()) { + if (ms.getValue().equals(input.toLowerCase())) { + return ms; + } + } + } + return MessageService.DMaaP; // Default + } + } + + /** + * Set of common properties that will be used by most systems. Primarily relating to DMaaP and ThreadPools + * + * @since Apr 25, 2016 + * @version $Id$ + */ + public enum KEYS { + /** + * Property to determine if the listener should be disabled. If not set, defaults to false + */ + DISABLED("disabled"), + + /** + * Property for the message service type. Should be a lower case string. See MessageService. + */ + MESSAGE_SERVICE("service"), + + /** + * A hostname or comma separated list (no spaces) of hostnames of servers in a cluster. Can have ports included + * as well.<br> + * Examples: + * <ul> + * <li>server1.appc.openecomp.org</li> + * <li>server1.appc.openecomp.org:3904</li> + * <li>server1.appc.openecomp.org,server2.appc.openecomp.org</li> + * </ul> + */ + HOSTS("poolMembers"), + + /** + * The topic that will be used for DMaaP read operations. Can only support a single topic. + */ + TOPIC_READ("topic.read"), + + /** + * The topic or topics that will be used to write to. If multiple topics are provided, should be in a comma + * seperated list with no spaces.<br> + * Examples: + * <ul> + * <li>TOPIC-1</li> + * <li>TOPIC-1,TOPIC-2,ANOTHER-TOPIC</li> + * </ul> + */ + TOPIC_WRITE("topic.write"), + + /** + * The highland park filter to use on read requests. If you are reading and writing to the same topic this must + * be provided. Filter should be in JSON format (not url escaped). + */ + TOPIC_READ_FILTER("topic.read.filter"), + + /** + * The amount of time in seconds that the DMaaP polling connection should stay open for. Recommended to be set + * high (around 60 seconds) as most clients will return immediately and not wait until the timeout is up to + * return if they have data. + */ + TOPIC_READ_TIMEOUT("topic.read.timeout"), + + /** + * The name of the client to use. Should be unique to the application. + */ + CLIENT_NAME("client.name"), + + /** + * The id of the client to use. Should be unique for each instance of the application in an environment. + */ + CLIENT_ID("client.name.id"), + + /** + * The User (DMaaP) to use for authentication. If a user is provided, you must include the + * domain name (e.g. example<b>@example.com</b>). + */ + AUTH_USER_KEY("client.key"), + + /** + * The password (DMaaP) to use for authentication. + */ + AUTH_SECRET_KEY("client.secret"), + + /** + * The minimum amount of size of the queue. A client should request new messages once the queue has dropped + * below this size. + */ + THREADS_MIN_QUEUE("threads.queuesize.min"), + + /** + * The maximum size of the queue. A client will request no new messages once this maximum size has been reached. + */ + THREADS_MAX_QUEUE("threads.queuesize.max"), + + /** + * The minimum size of the worker threads pool. This is the pool each listener will use to launch longer running + * operations. + */ + THREADS_MIN_POOL("threads.poolsize.min"), + + /** + * The maximum size of the worker threads pool. This is the pool each listener will use to launch longer running + * operations. + */ + THREADS_MAX_POOL("threads.poolsize.max"); + + private String suffix; + + private KEYS(String val) { + this.suffix = val; + } + + /** + * @param prefix + * The prefix to prepend + * @return a fully property name that corroponds to what is used in the properties file. Format is PREFIX.KEY + */ + public String getFullProp(String prefix) { + return String.format("%s.%s", prefix, suffix); + } + + public String getPropertySuffix() { + return suffix; + } + } + +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/ControllerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/ControllerImpl.java new file mode 100644 index 000000000..9e171915a --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/ControllerImpl.java @@ -0,0 +1,126 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.impl; + +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.lang3.concurrent.BasicThreadFactory; +import org.openecomp.appc.listener.Controller; +import org.openecomp.appc.listener.Listener; +import org.openecomp.appc.listener.ListenerProperties; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; + + +/** + * A common implementation of a controller. This controller should not need to be modified to implement new listeners + * + */ +public class ControllerImpl implements Controller { + + private final EELFLogger LOG = EELFManager.getInstance().getLogger(ControllerImpl.class); + + private int LISTENER_COUNT = 1; + + private Map<ListenerProperties, Listener> listeners = null; + + private ThreadPoolExecutor executor; + + /** + * Creates a Controller with the set of listener properties which will be used to start listener threads. + * + * @param properties + * A non null Set of ListenerProperties + */ + public ControllerImpl(Set<ListenerProperties> properties) { + listeners = new HashMap<ListenerProperties, Listener>(); + for (ListenerProperties props : properties) { + if (props.getClass() != null) { + listeners.put(props, null); + } else { + LOG.error(String.format( + "The ListenerProperties %s has no Listener class associated with it and will not run.", props)); + } + } + + LISTENER_COUNT = properties.size(); + + executor = new ThreadPoolExecutor(LISTENER_COUNT, LISTENER_COUNT, 1, TimeUnit.SECONDS, + new ArrayBlockingQueue<Runnable>(LISTENER_COUNT)); + + // Custom Named thread factory + BasicThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("Appc-Listener-%d").build(); + executor.setThreadFactory(threadFactory); + } + + @Override + public void start() { + LOG.info("Starting DMaaP Controller."); + for (ListenerProperties props : listeners.keySet()) { + try { + if (props.isDisabled()) { + LOG.warn(String.format("The listener %s is disabled and will not be run", props.getPrefix())); + } else { + Listener l = props.getListenerClass().getConstructor(ListenerProperties.class).newInstance(props); + l.setListenerId(props.getPrefix()); + listeners.put(props, l); + executor.execute(l); + } + } catch (Exception e) { + e.printStackTrace(); + LOG.error(String.format("Exception while starting listener %s.", props), e); + } + } + } + + @Override + public void stop(boolean stopNow) { + LOG.info("Stopping DMaaP Controller."); + Iterator<Listener> itr = listeners.values().iterator(); + while (itr.hasNext()) { + Listener l = itr.next(); + if (stopNow) { + l.stopNow(); + } else { + l.stop(); + } + itr.remove(); + } + executor.shutdown(); + try { + executor.awaitTermination(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + } + } + + @Override + public Map<ListenerProperties, Listener> getListeners() { + return listeners; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/EventHandlerImpl.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/EventHandlerImpl.java new file mode 100644 index 000000000..590afbe8b --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/impl/EventHandlerImpl.java @@ -0,0 +1,326 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.impl; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import org.openecomp.appc.adapter.dmaap.Consumer; +import org.openecomp.appc.adapter.dmaap.DmaapConsumer; +import org.openecomp.appc.adapter.dmaap.DmaapProducer; +import org.openecomp.appc.adapter.dmaap.Producer; +import org.openecomp.appc.adapter.dmaap.DmaapConsumer; +import org.openecomp.appc.adapter.dmaap.DmaapProducer; +import org.openecomp.appc.listener.EventHandler; +import org.openecomp.appc.listener.ListenerProperties; +import org.openecomp.appc.listener.ListenerProperties.MessageService; +import org.openecomp.appc.listener.util.Mapper; +import org.openecomp.appc.logging.LoggingConstants; + +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; +import org.slf4j.MDC; + +/** + * This class is a wrapper for the DMaaP client provided in appc-dmaap-adapter. Its aim is to ensure that only well formed + * messages are sent and received on DMaaP. + * + */ +public class EventHandlerImpl implements EventHandler { + + private final EELFLogger LOG = EELFManager.getInstance().getLogger(EventHandlerImpl.class); + + /* + * The amount of time in seconds to keep a connection to a topic open while waiting for data + */ + private int READ_TIMEOUT = 60; + + /* + * The pool of hosts to query against + */ + private Collection<String> pool; + + /* + * The topic to read messages from + */ + private String readTopic; + + /* + * The topic to write messages to + */ + private Set<String> writeTopics; + + /* + * The client (group) name to use for reading messages + */ + private String clientName; + + /* + * The id of the client (group) that is reading messages + */ + private String clientId; + + /* + * The api public key to use for authentication + */ + private String apiKey; + + /* + * The api secret key to use for authentication + */ + private String apiSecret; + + /* + * A json object containing filter arguments. + */ + private String filter_json; + + private MessageService messageService; + + private Consumer reader = null; + private Producer producer = null; + + public EventHandlerImpl(ListenerProperties props) { + pool = new HashSet<String>(); + writeTopics = new HashSet<String>(); + + if (props != null) { + readTopic = props.getProperty(ListenerProperties.KEYS.TOPIC_READ); + clientName = props.getProperty(ListenerProperties.KEYS.CLIENT_NAME, "APP-C"); + clientId = props.getProperty(ListenerProperties.KEYS.CLIENT_ID, "0"); + apiKey = props.getProperty(ListenerProperties.KEYS.AUTH_USER_KEY); + apiSecret = props.getProperty(ListenerProperties.KEYS.AUTH_SECRET_KEY); + + filter_json = props.getProperty(ListenerProperties.KEYS.TOPIC_READ_FILTER); + + READ_TIMEOUT = Integer + .valueOf(props.getProperty(ListenerProperties.KEYS.TOPIC_READ_TIMEOUT, String.valueOf(READ_TIMEOUT))); + + String hostnames = props.getProperty(ListenerProperties.KEYS.HOSTS); + if (hostnames != null && !hostnames.isEmpty()) { + for (String name : hostnames.split(",")) { + pool.add(name); + } + } + + String writeTopicStr = props.getProperty(ListenerProperties.KEYS.TOPIC_WRITE); + if (writeTopicStr != null) { + for (String topic : writeTopicStr.split(",")) { + writeTopics.add(topic); + } + } + + messageService = MessageService.parse(props.getProperty(ListenerProperties.KEYS.MESSAGE_SERVICE)); + + LOG.info(String.format( + "Configured to use %s client on host pool [%s]. Reading from [%s] filtered by %s. Wriring to [%s]. Authenticated using %s", + messageService, hostnames, readTopic, filter_json, writeTopics, apiKey)); + } + } + + @Override + public List<String> getIncomingEvents() { + return getIncomingEvents(1000); + } + + @Override + public List<String> getIncomingEvents(int limit) { + List<String> out = new ArrayList<String>(); + LOG.info(String.format("Getting up to %d incoming events", limit)); + // reuse the consumer object instead of creating a new one every time + if (reader == null) { + LOG.info("Getting Consumer..."); + reader = getConsumer(); + } + for (String item : reader.fetch(READ_TIMEOUT * 1000, limit)) { + out.add(item); + } + LOG.info(String.format("Read %d messages from %s as %s/%s.", out.size(), readTopic, clientName, clientId)); + return out; + } + + @Override + public <T> List<T> getIncomingEvents(Class<T> cls) { + return getIncomingEvents(cls, 1000); + } + + @Override + public <T> List<T> getIncomingEvents(Class<T> cls, int limit) { + List<String> incomingStrings = getIncomingEvents(limit); + return Mapper.mapList(incomingStrings, cls); + } + + @Override + public void postStatus(String event) { + postStatus(null, event); + } + + @Override + public void postStatus(String partition, String event) { + LOG.debug(String.format("Posting Message [%s]", event)); + if (producer == null) { + LOG.info("Getting Producer..."); + producer = getProducer(); + } + producer.post(partition, event); + } + + /** + * Returns a consumer object for direct access to our Cambria consumer interface + * + * @return An instance of the consumer interface + */ + protected Consumer getConsumer() { + LOG.debug(String.format("Getting Consumer: %s %s/%s/%s", pool, readTopic, clientName, clientId)); + if (filter_json == null && writeTopics.contains(readTopic)) { + LOG.error( + "*****We will be writing and reading to the same topic without a filter. This will cause an infinite loop.*****"); + } + Consumer out; + out = new DmaapConsumer(pool, readTopic, clientName, clientId, filter_json, apiKey, apiSecret); + for (String url : pool) { + if (url.contains("3905") || url.contains("https")) { + out.useHttps(true); + break; + } + } + return out; + } + + /** + * Returns a consumer object for direct access to our Cambria producer interface + * + * @return An instance of the producer interface + */ + protected Producer getProducer() { + LOG.debug(String.format("Getting Producer: %s %s", pool, readTopic)); + + Producer out; + out = new DmaapProducer(pool,writeTopics); + + if (apiKey != null && apiSecret != null) { + out.updateCredentials(apiKey, apiSecret); + } + + for (String url : pool) { + if (url.contains("3905") || url.contains("https")) { + out.useHttps(true); + break; + } + } + return out; + } + + @Override + public void closeClients() { + LOG.debug("Closing Consumer and Producer DMaaP clients"); + switch (messageService) { + case DMaaP: + if (reader != null) { + ((DmaapConsumer) reader).close(); + } + if (producer != null) { + ((DmaapProducer) producer).close(); + } + break; + default: + // close DMaaP clients + if (reader != null) { + ((DmaapConsumer) reader).close(); + } + if (producer != null) { + ((DmaapProducer) producer).close(); + } + } + } + + @Override + public String getClientId() { + return clientId; + } + + @Override + public void setClientId(String clientId) { + this.clientId = clientId; + } + + @Override + public String getClientName() { + return clientName; + } + + @Override + public void setClientName(String clientName) { + this.clientName = clientName; + MDC.put(LoggingConstants.MDCKeys.PARTNER_NAME, clientName); + } + + @Override + public void addToPool(String hostname) { + pool.add(hostname); + } + + @Override + public Collection<String> getPool() { + return pool; + } + + @Override + public void removeFromPool(String hostname) { + pool.remove(hostname); + } + + @Override + public String getReadTopic() { + return readTopic; + } + + @Override + public void setReadTopic(String readTopic) { + this.readTopic = readTopic; + } + + @Override + public Set<String> getWriteTopics() { + return writeTopics; + } + + @Override + public void setWriteTopics(Set<String> writeTopics) { + this.writeTopics = writeTopics; + } + + @Override + public void clearCredentials() { + apiKey = null; + apiSecret = null; + } + + @Override + public void setCredentials(String key, String secret) { + apiKey = key; + apiSecret = secret; + } +} diff --git a/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/util/Mapper.java b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/util/Mapper.java new file mode 100644 index 000000000..cdb0d14f0 --- /dev/null +++ b/appc-event-listener/appc-event-listener-bundle/src/main/java/org/openecomp/appc/listener/util/Mapper.java @@ -0,0 +1,134 @@ +/*- + * ============LICENSE_START======================================================= + * openECOMP : APP-C + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.appc.listener.util; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +import com.fasterxml.jackson.databind.JsonNode; +import org.json.JSONObject; +import com.att.eelf.configuration.EELFLogger; +import com.att.eelf.configuration.EELFManager; + +import com.fasterxml.jackson.databind.ObjectMapper; + +public class Mapper { + + private static final EELFLogger LOG = EELFManager.getInstance().getLogger(Mapper.class); + + private static ObjectMapper mapper = new ObjectMapper(); + + /** + * @return The object mapper that we are using. + */ + public static ObjectMapper getMapper() { + return mapper; + } + + /** + * Convert a String to a DcaeMessage + * + * @param data + * The json string to try and parse + * @return A DcaeMessage from the json string or null if it could not + */ + public static <T> T mapOne(String data, Class<T> cls) { + try { + return mapper.readValue(data, cls); + } catch (Exception e) { + LOG.warn(String.format("Could not map [ %s ] to %s", data, cls.getName()), e); + return null; + } + } + + public static <T> List<T> mapList(List<String> data, Class<T> cls) { + List<T> out = new ArrayList<T>(); + for (String s : data) { + T tmp = Mapper.mapOne(s, cls); + if (tmp != null) { + out.add(tmp); + } + } + return out; + } + + /** + * Convenience method to try and convert objects to json String + * + * @param obj + * The object to try and convert + * @return A json string representing the object or null if it could not be converted + */ + public static String toJsonString(Object obj) { + String jsonStr; + try { + if (obj instanceof JSONObject) { + jsonStr = obj.toString(); + }else { + jsonStr = mapper.writeValueAsString(obj); + } + return jsonStr; + } catch (Exception e) { + LOG.warn(String.format("Could not map %s to JSONObject.", obj), e); + return null; + } + } + + public static JSONObject toJsonObject(Object obj) { + String jsonStr; + try { + if (obj.getClass().equals(String.class)) { + jsonStr = (String) obj; + } else { + jsonStr = mapper.writeValueAsString(obj); + } + return new JSONObject(jsonStr); + } catch (Exception e) { + LOG.warn(String.format("Could not map %s to JSONObject.", obj), e); + return null; + } + } + public static JsonNode toJsonNodeFromJsonString(String jsonStr) { + JsonNode jsonNode = null; + if(jsonStr != null) { + try { + jsonNode = mapper.readTree(jsonStr); + } catch (IOException e) { + LOG.warn(String.format("Could not map %s to jsonNode.", jsonStr), e); + } + } + return jsonNode; + } + public static JsonNode toJsonNode(Object obj) { + JsonNode jsonNode = null; + String jsonStr = toJsonString(obj); + if(jsonStr != null) { + try { + jsonNode = mapper.readTree(jsonStr); + } catch (IOException e) { + LOG.warn(String.format("Could not map %s to JSONObject.", obj), e); + } + } + return jsonNode; + } +} |