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