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

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