appc-adapters/appc-dmaap-adapter/appc-dmaap-adapter-bundle/src/main/java/org/onap/appc/adapter/messaging/dmaap/impl/DmaapProducerImpl.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.MRBatchingPublisher;
  29 import com.att.nsa.mr.client.MRClientFactory;
  30 import java.io.IOException;
  31 import java.util.Collection;
  32 import java.util.Collections;
  33 import java.util.HashSet;
  34 import java.util.Properties;
  35 import java.util.Set;
  36 import java.util.UUID;
  37 import java.util.concurrent.TimeUnit;
  38 import org.apache.commons.lang3.StringUtils;
  39 import org.onap.appc.adapter.message.Producer;
  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.metric.DmaapRequestCounterMetric;
  45 import org.onap.appc.metricservice.metric.Metric;
  46 import org.onap.appc.metricservice.metric.MetricType;
  47 import org.onap.appc.metricservice.policy.PublishingPolicy;
  48 import org.onap.appc.metricservice.publisher.LogPublisher;
  49 import org.osgi.framework.BundleContext;
  50 import org.osgi.framework.FrameworkUtil;
  51 import org.osgi.framework.ServiceReference;
  52
  53 public class DmaapProducerImpl implements Producer {
  54
  55     private static final EELFLogger    LOG             = EELFManager.getInstance().getLogger(DmaapProducerImpl.class);
  56     private static final Configuration configuration   = ConfigurationFactory.getConfiguration();
  57
  58     private Set<String>                topics;
  59
  60     private Properties                 props           = null;
  61     private MetricRegistry             metricRegistry;
  62     private boolean                    useHttps        = false;
  63     private boolean                    isMetricEnabled = false;
  64
  65     private Set<MRBatchingPublisher>   clients;
  66
  67     public DmaapProducerImpl(Collection<String> urls, String topicName, String user, String password) {
  68         this(urls, (Set<String>) null, user, password);
  69         this.topics = new HashSet<>();
  70         if (topicName != null) {
  71             Collections.addAll(topics, topicName.split(","));
  72         }
  73     }
  74
  75     public DmaapProducerImpl(Collection<String> urls, Set<String> topicNames, String user, String password) {
  76         topics = topicNames;
  77         if (urls == null) {
  78             throw new IllegalArgumentException("Mandaory argument is null: urls");
  79         }
  80         this.props = new Properties();
  81         String urlsStr = StringUtils.join(urls, ',');
  82         props.setProperty("host", urlsStr);
  83         props.setProperty("id", UUID.randomUUID().toString());
  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
  92     private void initMetric() {
  93         LOG.debug("Metric getting initialized");
  94         MetricService metricService = getMetricservice();
  95         if (metricService != null) {
  96             metricRegistry = metricService.createRegistry("APPC");
  97
  98             DmaapRequestCounterMetric dmaapKpiMetric = metricRegistry.metricBuilderFactory()
  99                     .dmaapRequestCounterBuilder().withName("DMAAP_KPI").withType(MetricType.COUNTER)
 100                     .withRecievedMessage(0).withPublishedMessage(0).build();
 101
 102             if (metricRegistry.register(dmaapKpiMetric)) {
 103                 Metric[] metrics = new Metric[] { dmaapKpiMetric };
 104                 LogPublisher logPublisher = new LogPublisher(metricRegistry, metrics);
 105                 LogPublisher[] logPublishers = new LogPublisher[1];
 106                 logPublishers[0] = logPublisher;
 107
 108                 PublishingPolicy manuallyScheduledPublishingPolicy = metricRegistry.policyBuilderFactory()
 109                         .scheduledPolicyBuilder().withPublishers(logPublishers).withMetrics(metrics).build();
 110
 111                 LOG.debug("Policy getting initialized");
 112                 manuallyScheduledPublishingPolicy.init();
 113                 LOG.debug("Metric initialized");
 114             }
 115         }
 116     }
 117
 118     private Set<MRBatchingPublisher> getClients() {
 119         Set<MRBatchingPublisher> out = new HashSet<>();
 120         for (String topic : topics) {
 121             try {
 122                 String topicProducerPropFileName = DmaapUtil.createProducerPropFile(topic, props);
 123                 final MRBatchingPublisher client = MRClientFactory.createBatchingPublisher(topicProducerPropFileName);
 124                 out.add(client);
 125             } catch (Exception e) {
 126                 LOG.error(e.getMessage(), e);
 127             }
 128         }
 129         return out;
 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         clients = null;
 138     }
 139
 140     @Override
 141     public boolean post(String partition, String data) {
 142         boolean success = true;
 143         Properties properties = configuration.getProperties();
 144         if (properties != null && properties.getProperty("metric.enabled") != null) {
 145             isMetricEnabled = Boolean.valueOf(properties.getProperty("metric.enabled"));
 146         }
 147         if (isMetricEnabled) {
 148             initMetric();
 149         }
 150
 151         // Create clients once and reuse them on subsequent posts. This is
 152         // to support failover to other servers in the Dmaap cluster.
 153         if ((clients == null) || (clients.isEmpty())) {
 154             LOG.info("Getting CambriaBatchingPublisher Clients ...");
 155             clients = getClients();
 156         }
 157
 158         for (MRBatchingPublisher client : clients) {
 159             try {
 160                 LOG.debug(String.format("Posting %s to %s", data, client));
 161                 client.send(partition, data);
 162             } catch (IOException e) {
 163                 LOG.error(e.getMessage(), e);
 164                 success = false;
 165             }
 166         }
 167         incrementPublishedMessage();
 168         return success;
 169     }
 170
 171     private void incrementPublishedMessage() {
 172         if (isMetricEnabled && metricRegistry != null) {
 173             ((DmaapRequestCounterMetric) metricRegistry.metric("DMAAP_KPI")).incrementPublishedMessage();
 174         }
 175     }
 176
 177     /**
 178      * Close producer Dmaap client.
 179      */
 180     @Override
 181     public void close() {
 182         if ((clients == null) || (clients.isEmpty())) {
 183             return;
 184         }
 185         LOG.debug("Closing Dmaap producer clients....");
 186         for (MRBatchingPublisher client : clients) {
 187             try {
 188                 client.close(1, TimeUnit.SECONDS);
 189             } catch (IOException | InterruptedException e) {
 190                 LOG.warn(String.format("Failed to cleanly close Dmaap connection for [%s]", client), e);
 191             }
 192         }
 193     }
 194
 195     @Override
 196     public void useHttps(boolean yes) {
 197         useHttps = yes;
 198     }
 199
 200     private MetricService getMetricservice() {
 201         BundleContext bctx = FrameworkUtil.getBundle(MetricService.class).getBundleContext();
 202         ServiceReference sref = bctx.getServiceReference(MetricService.class.getName());
 203         if (sref != null) {
 204             LOG.info("Metric Service from bundlecontext");
 205             return (MetricService) bctx.getService(sref);
 206         } else {
 207             LOG.info("Metric Service error from bundlecontext");
 208             LOG.warn("Cannot find service reference for org.onap.appc.metricservice.MetricService");
 209             return null;
 210         }
 211     }
 212
 213     public Properties getProperties() {
 214         return props;
 215     }
 216
 217     public boolean isHttps() {
 218         return useHttps;
 219     }
 220 }