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