appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/openecomp/appc/adapter/messaging/dmaap/impl/DmaapConsumerImpl.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP : APPC
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Copyright (C) 2017 Amdocs
   8  * =============================================================================
   9  * Licensed under the Apache License, Version 2.0 (the "License");
  10  * you may not use this file except in compliance with the License.
  11  * You may obtain a copy of the License at
  12  *
  13  *      http://www.apache.org/licenses/LICENSE-2.0
  14  *
  15  * Unless required by applicable law or agreed to in writing, software
  16  * distributed under the License is distributed on an "AS IS" BASIS,
  17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18  * See the License for the specific language governing permissions and
  19  * limitations under the License.
  20  *
  21  * ECOMP is a trademark and service mark of AT&T Intellectual Property.
  22  * ============LICENSE_END=========================================================
  23  */
  24
  25 package org.onap.appc.adapter.messaging.dmaap.impl;
  26
  27 import java.io.IOException;
  28 import java.util.*;
  29
  30 import com.att.eelf.configuration.EELFLogger;
  31 import com.att.eelf.configuration.EELFManager;
  32 //import com.att.nsa.cambria.client.CambriaClientBuilders;
  33 //import com.att.nsa.cambria.client.CambriaClientBuilders.ConsumerBuilder;
  34 //import com.att.nsa.cambria.client.CambriaConsumer;
  35
  36 import com.att.nsa.mr.client.MRClientFactory;
  37 import com.att.nsa.mr.client.MRConsumer;
  38 import org.apache.commons.lang3.StringUtils;
  39 import org.onap.appc.adapter.message.Consumer;
  40 import org.onap.appc.configuration.Configuration;
  41 import org.onap.appc.configuration.ConfigurationFactory;
  42 import org.onap.appc.metricservice.MetricRegistry;
  43 import org.onap.appc.metricservice.MetricService;
  44 import org.onap.appc.metricservice.impl.MetricServiceImpl;
  45 import org.onap.appc.metricservice.metric.Metric;
  46 import org.onap.appc.metricservice.metric.MetricType;
  47 import org.onap.appc.metricservice.metric.DmaapRequestCounterMetric;
  48 import org.onap.appc.metricservice.policy.PublishingPolicy;
  49 import org.onap.appc.metricservice.publisher.LogPublisher;
  50 import org.osgi.framework.BundleContext;
  51 import org.osgi.framework.FrameworkUtil;
  52 import org.osgi.framework.ServiceReference;
  53
  54 public class DmaapConsumerImpl implements Consumer {
  55
  56     private static final EELFLogger LOG = EELFManager.getInstance().getLogger(DmaapConsumerImpl.class);
  57     private static final Configuration configuration = ConfigurationFactory.getConfiguration();
  58     // Default values
  59     private static final int DEFAULT_TIMEOUT_MS = 60000;
  60     private static final int DEFAULT_LIMIT = 1000;
  61     private static MetricRegistry metricRegistry;
  62     private String topic;
  63     private DmaapRequestCounterMetric dmaapKpiMetric;
  64     private boolean isMetricEnabled=false;
  65     private boolean useHttps = false;
  66     private MRConsumer client = null;
  67     private Properties props = null;
  68
  69
  70     public DmaapConsumerImpl(Collection<String> urls, String topicName, String consumerGroupName, String consumerId,String user, String password) {
  71         this(urls, topicName, consumerGroupName, consumerId,user, password,null);
  72
  73     }
  74
  75     public DmaapConsumerImpl(Collection<String> urls, String topicName, String consumerGroupName, String consumerId,String user, String password,String filter) {
  76         this.topic = topicName;
  77         this.props = new Properties();
  78         String urlsStr = StringUtils.join(urls, ',');
  79         props.setProperty("host",urlsStr);
  80         props.setProperty("group",consumerGroupName);
  81         props.setProperty("id",consumerId);
  82         props.setProperty("username",user);
  83         props.setProperty("password",password);
  84         if(filter != null) {
  85             props.setProperty("filter", filter);
  86         }
  87     }
  88
  89
  90     private void initMetric() {
  91         LOG.debug("Metric getting initialized");
  92         MetricService metricService = getMetricservice();
  93         metricRegistry = metricService.createRegistry("APPC");
  94         dmaapKpiMetric = metricRegistry.metricBuilderFactory().
  95                 dmaapRequestCounterBuilder().
  96                 withName("DMAAP_KPI").withType(MetricType.COUNTER).
  97                 withRecievedMessage(0)
  98                 .withPublishedMessage(0)
  99                 .build();
 100         if (metricRegistry.register(dmaapKpiMetric)) {
 101             Metric[] metrics = new Metric[]{dmaapKpiMetric};
 102             LogPublisher logPublisher = new LogPublisher(metricRegistry, metrics);
 103             LogPublisher[] logPublishers = new LogPublisher[1];
 104             logPublishers[0] = logPublisher;
 105             PublishingPolicy manuallyScheduledPublishingPolicy = metricRegistry.policyBuilderFactory().
 106                     scheduledPolicyBuilder().withPublishers(logPublishers).
 107                     withMetrics(metrics).
 108                     build();
 109             LOG.debug("Policy getting initialized");
 110             manuallyScheduledPublishingPolicy.init();
 111             LOG.debug("Metric initialized");
 112         }
 113     }
 114     private MRConsumer getClient() {
 115         return getClient(DEFAULT_TIMEOUT_MS, DEFAULT_LIMIT);
 116     }
 117
 118     /**
 119      * @return An instance of MRConsumer created from our class variables
 120      */
 121     private synchronized MRConsumer getClient(int waitMs, int limit) {
 122         try {
 123             props.setProperty("timeout",String.valueOf(waitMs));
 124             props.setProperty("limit",String.valueOf(limit));
 125             String topicProducerPropFileName = DmaapUtil.createConsumerPropFile(topic,props);
 126             return MRClientFactory.createConsumer ( topicProducerPropFileName);
 127         } catch (IOException e1) {
 128             LOG.error("failed to createConsumer",e1);
 129             return null;
 130         }
 131     }
 132
 133     @Override
 134     public synchronized void updateCredentials(String key, String secret) {
 135         LOG.info(String.format("Setting auth to %s for %s", key, this.toString()));
 136         String user = key;
 137         String password = secret;
 138         props.setProperty("user",String.valueOf(user));
 139         props.setProperty("password",String.valueOf(password));
 140         client = null;
 141     }
 142
 143     @Override
 144     public List<String> fetch(int waitMs, int limit) {
 145         Properties properties=configuration.getProperties();
 146         if(properties!=null && properties.getProperty("metric.enabled")!=null ){
 147           isMetricEnabled=Boolean.valueOf(properties.getProperty("metric.enabled"));
 148         }
 149         if(isMetricEnabled){
 150             initMetric();
 151         }
 152         LOG.debug(String.format("Fetching up to %d records with %dms wait on %s", limit, waitMs, this.toString()));
 153         List<String> out = new ArrayList<String>();
 154
 155         // Create client once and reuse it on subsequent fetches. This is
 156         // to support failover to other servers in the DMaaP cluster.
 157         if (client == null) {
 158                 LOG.info("Getting DMaaP Client ...");
 159                 client = getClient(waitMs, limit);
 160         }
 161         try {
 162             for (String s : client.fetch(waitMs, limit)) {
 163                 out.add(s);
 164                 if(isMetricEnabled){
 165                     ((DmaapRequestCounterMetric)metricRegistry.metric("DMAAP_KPI")).incrementRecievedMessage();
 166                 }
 167             }
 168             LOG.debug(String.format("Got %d records from %s", out.size(), this.toString()));
 169         } catch (Exception e) {
 170             // Connection exception
 171             LOG.error(String.format("Dmaap Connection Issue Detected. %s", e.getMessage()));
 172             e.printStackTrace();
 173             try {
 174                 LOG.warn(String.format("Sleeping for %dms to compensate for connection failure", waitMs));
 175                 Thread.sleep(waitMs);
 176             } catch (InterruptedException e2) {
 177                 LOG.warn(String.format("Failed to wait for %dms after bad fetch", waitMs));
 178             }
 179         }
 180
 181
 182         return out;
 183     }
 184
 185     /**
 186      * Close consumer Dmaap client
 187      */
 188     @Override
 189     public void close() {
 190         LOG.debug("Closing Dmaap consumer client....");
 191         if (client != null) {
 192                 client.close();
 193         }
 194     }
 195
 196     @Override
 197     public List<String> fetch() {
 198         return fetch(DEFAULT_TIMEOUT_MS, DEFAULT_LIMIT);
 199     }
 200
 201     @Override
 202     public String toString() {
 203         String hostStr = (props == null || props.getProperty("host") == null? "N/A" : props.getProperty("host"));
 204         String group = (props == null || props.getProperty("group") == null? "N/A" : props.getProperty("group"));
 205         String id = (props == null || props.getProperty("id") == null? "N/A" : props.getProperty("id"));
 206         return String.format("Consumer %s/%s listening to %s on [%s]", group, id, topic, hostStr);
 207     }
 208
 209     @Override
 210     public void useHttps(boolean yes) {
 211         useHttps = yes;
 212     }
 213
 214
 215     private MetricService getMetricservice() {
 216         BundleContext bctx = FrameworkUtil.getBundle(MetricService.class).getBundleContext();
 217         // Get AAIadapter reference
 218         ServiceReference sref = bctx.getServiceReference(MetricService.class.getName());
 219         if (sref != null) {
 220             LOG.info("Metric Service from bundlecontext");
 221             return (MetricServiceImpl) bctx.getService(sref);
 222
 223         } else {
 224             LOG.info("Metric Service error from bundlecontext");
 225             LOG.warn("Cannot find service reference for org.onap.appc.metricservice.MetricService");
 226             return null;
 227
 228         }
 229     }
 230
 231     public  Metric getMetric(String name){
 232         return metricRegistry.metric(name);
 233     }
 234 }