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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * openECOMP : APP-C
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights
   6  *                                              reserved.
   7  * ================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *      http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.openecomp.appc.adapter.messaging.dmaap.impl;
  23
  24 import java.io.*;
  25 import java.util.*;
  26 import java.util.concurrent.TimeUnit;
  27
  28 import com.att.eelf.configuration.EELFLogger;
  29 import com.att.eelf.configuration.EELFManager;
  30 //import com.att.nsa.cambria.client.CambriaBatchingPublisher;
  31 //import com.att.nsa.cambria.client.CambriaClientBuilders;
  32 //import com.att.nsa.cambria.client.CambriaClientBuilders.PublisherBuilder;
  33
  34 import com.att.nsa.mr.client.MRBatchingPublisher;
  35 import com.att.nsa.mr.client.MRClientFactory;
  36 import org.apache.commons.lang3.StringUtils;
  37 import org.openecomp.appc.adapter.message.Producer;
  38 import org.openecomp.appc.adapter.messaging.dmaap.impl.DmaapUtil;
  39 import org.openecomp.appc.configuration.Configuration;
  40 import org.openecomp.appc.configuration.ConfigurationFactory;
  41 import org.openecomp.appc.metricservice.MetricRegistry;
  42 import org.openecomp.appc.metricservice.MetricService;
  43 import org.openecomp.appc.metricservice.metric.Metric;
  44 import org.openecomp.appc.metricservice.metric.MetricType;
  45 import org.openecomp.appc.metricservice.metric.DmaapRequestCounterMetric;
  46 import org.openecomp.appc.metricservice.policy.PublishingPolicy;
  47 import org.openecomp.appc.metricservice.publisher.LogPublisher;
  48 import org.osgi.framework.BundleContext;
  49 import org.osgi.framework.FrameworkUtil;
  50 import org.osgi.framework.ServiceReference;
  51
  52 public class DmaapProducerImpl implements Producer {
  53
  54     private static final EELFLogger LOG = EELFManager.getInstance().getLogger(DmaapProducerImpl.class);
  55     private static final Configuration configuration = ConfigurationFactory.getConfiguration();
  56
  57     private Set<String> topics = new HashSet<String>();
  58
  59     private Properties props = null;
  60     private static MetricRegistry metricRegistry;
  61     private boolean useHttps = false;
  62     private DmaapRequestCounterMetric dmaapKpiMetric;
  63     private boolean isMetricEnabled=false;
  64
  65     private Set<MRBatchingPublisher> clients;
  66
  67
  68     public DmaapProducerImpl(Collection<String> urls, String topicName, String user, String password) {
  69         this(urls, (Set<String>)null, user, password);
  70         this.topics = new HashSet<>();
  71         if (topicName != null) {
  72             for (String topic : topicName.split(",")) {
  73                 topics.add(topic);
  74             }
  75         }
  76     }
  77
  78     public DmaapProducerImpl(Collection<String> urls, Set<String> topicNames, String user, String password) {
  79         topics = topicNames;
  80         if(urls == null || user == null || password == null){
  81             throw new IllegalArgumentException("one of these mandaory argument is null: urls, user, password" );
  82         }
  83         this.props = new Properties();
  84         String urlsStr = StringUtils.join(urls, ',');
  85         props.setProperty("host",urlsStr);
  86         props.setProperty("id", UUID.randomUUID().toString());
  87         props.setProperty("username",user);
  88         props.setProperty("password",password);
  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     }
 115     private Set<MRBatchingPublisher> getClients() {
 116         Set<MRBatchingPublisher> out = new HashSet<MRBatchingPublisher>();
 117         for (String topic : topics) {
 118             try {
 119                 String topicProducerPropFileName = DmaapUtil.createProducerPropFile(topic,props);
 120                 final MRBatchingPublisher client = MRClientFactory.createBatchingPublisher (topicProducerPropFileName);
 121                 out.add(client);
 122             } catch (Exception e) {
 123                 e.printStackTrace();
 124             }
 125         }
 126
 127         return out;
 128     }
 129
 130     @Override
 131     public synchronized void updateCredentials(String key, String secret) {
 132         LOG.info(String.format("Setting auth to %s for %s", key, this.toString()));
 133         String user = key;
 134         String password = secret;
 135         props.setProperty("user",String.valueOf(user));
 136         props.setProperty("password",String.valueOf(password));
 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                 e.printStackTrace();
 164                 success = false;
 165             }
 166         }
 167         if(isMetricEnabled){
 168             ( (DmaapRequestCounterMetric) metricRegistry.metric("DMAAP_KPI")).incrementPublishedMessage();
 169         }
 170         return success;
 171     }
 172
 173     /**
 174      * Close producer Dmaap client
 175      */
 176     @Override
 177     public void close() {
 178         if ((clients == null) || (clients.isEmpty())) {
 179                 return;
 180         }
 181
 182         LOG.debug("Closing Dmaap producer clients....");
 183         for (MRBatchingPublisher client : clients) {
 184             try {
 185                 client.close(1, TimeUnit.SECONDS);
 186             }  catch (IOException | InterruptedException e) {
 187                 LOG.warn(String.format("Failed to cleanly close Dmaap connection for [%s]", client));
 188                 e.printStackTrace();
 189             }
 190         }
 191     }
 192
 193     @Override
 194     public void useHttps(boolean yes) {
 195         useHttps = yes;
 196     }
 197
 198     private MetricService getMetricservice() {
 199 /*
 200         return AppcDmaapAdapterActivator.getMetricService();
 201 */
 202
 203         BundleContext bctx = FrameworkUtil.getBundle(MetricService.class).getBundleContext();
 204         ServiceReference sref = bctx.getServiceReference(MetricService.class.getName());
 205         if (sref != null) {
 206             LOG.info("Metric Service from bundlecontext");
 207             return (MetricService) bctx.getService(sref);
 208
 209         } else {
 210             LOG.info("Metric Service error from bundlecontext");
 211             LOG.warn("Cannot find service reference for org.openecomp.appc.metricservice.MetricService");
 212             return null;
 213
 214         }
 215     }
 216
 217     public  Metric getMetric(String name){
 218         return metricRegistry.metric(name);
 219     }
 220 }