gerrit.onap Code Review - appc.git/blob

   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) 2019 Ericsson
  10  * =============================================================================
  11  * Licensed under the Apache License, Version 2.0 (the "License");
  12  * you may not use this file except in compliance with the License.
  13  * You may obtain a copy of the License at
  14  *
  15  *      http://www.apache.org/licenses/LICENSE-2.0
  16  *
  17  * Unless required by applicable law or agreed to in writing, software
  18  * distributed under the License is distributed on an "AS IS" BASIS,
  19  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  20  * See the License for the specific language governing permissions and
  21  * limitations under the License.
  22  *
  23  * ============LICENSE_END=========================================================
  24  */
  25
  26 package org.onap.appc.adapter.messaging.dmaap.impl;
  27
  28 import com.att.eelf.configuration.EELFLogger;
  29 import com.att.eelf.configuration.EELFManager;
  30 import com.att.nsa.mr.client.MRBatchingPublisher;
  31 import com.att.nsa.mr.client.MRClientFactory;
  32 import java.io.IOException;
  33 import java.util.Collection;
  34 import java.util.Collections;
  35 import java.util.HashSet;
  36 import java.util.Properties;
  37 import java.util.Set;
  38 import java.util.UUID;
  39 import java.util.concurrent.TimeUnit;
  40 import org.apache.commons.lang3.StringUtils;
  41 import org.onap.appc.adapter.message.Producer;
  42 import org.onap.appc.adapter.messaging.dmaap.utils.DmaapUtil;
  43 import org.onap.appc.configuration.Configuration;
  44 import org.onap.appc.configuration.ConfigurationFactory;
  45 import org.onap.appc.metricservice.MetricRegistry;
  46 import org.onap.appc.metricservice.MetricService;
  47 import org.onap.appc.metricservice.metric.DmaapRequestCounterMetric;
  48 import org.onap.appc.metricservice.metric.Metric;
  49 import org.onap.appc.metricservice.metric.MetricType;
  50 import org.onap.appc.metricservice.policy.PublishingPolicy;
  51 import org.onap.appc.metricservice.publisher.LogPublisher;
  52 import org.osgi.framework.BundleContext;
  53 import org.osgi.framework.FrameworkUtil;
  54 import org.osgi.framework.ServiceReference;
  55
  56
  57 public class DmaapProducerImpl implements Producer {
  58
  59     private static final EELFLogger    LOG             = EELFManager.getInstance().getLogger(DmaapProducerImpl.class);
  60     private static       Configuration configuration   = ConfigurationFactory.getConfiguration();
  61
  62     private Set<String>                topics;
  63
  64     private Properties                 props           = null;
  65     private MetricRegistry             metricRegistry;
  66     private boolean                    useHttps        = false;
  67     private boolean                    isMetricEnabled = false;
  68
  69     private Set<MRBatchingPublisher>   clients;
  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             Collections.addAll(topics, topicName.split(","));
  76         }
  77     }
  78
  79     public DmaapProducerImpl(Collection<String> urls, Set<String> topicNames, String user, String password) {
  80         topics = topicNames;
  81         if (urls == null) {
  82             throw new IllegalArgumentException("Mandaory argument is null: urls");
  83         }
  84         this.props = new Properties();
  85         String urlsStr = StringUtils.join(urls, ',');
  86         props.setProperty("host", urlsStr);
  87         props.setProperty("id", UUID.randomUUID().toString());
  88         if (user != null && password != null) {
  89             props.setProperty("username", user);
  90             props.setProperty("password", password);
  91         } else {
  92             props.setProperty("TransportType", "HTTPNOAUTH");
  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().withPublishers(logPublishers).withMetrics(metrics).build();
 114
 115                 LOG.debug("Policy getting initialized");
 116                 manuallyScheduledPublishingPolicy.init();
 117                 LOG.debug("Metric initialized");
 118             }
 119         }
 120     }
 121
 122     private Set<MRBatchingPublisher> getClients() {
 123         Set<MRBatchingPublisher> out = new HashSet<>();
 124         for (String topic : topics) {
 125             try {
 126                 String topicProducerPropFileName = DmaapUtil.createProducerPropFile(topic, props);
 127                 final MRBatchingPublisher client = MRClientFactory.createBatchingPublisher(topicProducerPropFileName);
 128                 out.add(client);
 129             } catch (Exception e) {
 130                 LOG.error(e.getMessage(), e);
 131             }
 132         }
 133         return out;
 134     }
 135
 136     @Override
 137     public synchronized void updateCredentials(String key, String secret) {
 138         LOG.info(String.format("Setting auth to %s for %s", key, this.toString()));
 139         props.setProperty("username", String.valueOf(key));
 140         props.setProperty("password", String.valueOf(secret));
 141         clients = null;
 142     }
 143
 144     @Override
 145     public boolean post(String partition, String data) {
 146         LOG.debug("In DmaapProducerImpl.post()");
 147         boolean success = true;
 148         Properties properties = configuration.getProperties();
 149         if (properties != null && properties.getProperty("metric.enabled") != null) {
 150             isMetricEnabled = Boolean.valueOf(properties.getProperty("metric.enabled"));
 151         }
 152         if (isMetricEnabled) {
 153             initMetric();
 154         }
 155
 156         // Create clients once and reuse them on subsequent posts. This is
 157         // to support failover to other servers in the Dmaap cluster.
 158         if ((clients == null) || (clients.isEmpty())) {
 159             LOG.info("Getting CambriaBatchingPublisher Clients ...");
 160             clients = getClients();
 161         }
 162         LOG.debug("In DmaapProducerImpl.post()::: before sending to clients");
 163         for (MRBatchingPublisher client : clients) {
 164             try {
 165                 LOG.debug(String.format("Posting %s to %s", data, client));
 166                 client.send(partition, data);
 167             } catch (IOException e) {
 168                 LOG.error(e.getMessage(), e);
 169                 success = false;
 170             }
 171         }
 172         incrementPublishedMessage();
 173         return success;
 174     }
 175
 176     private void incrementPublishedMessage() {
 177         if (isMetricEnabled && metricRegistry != null) {
 178             ((DmaapRequestCounterMetric) metricRegistry.metric("DMAAP_KPI")).incrementPublishedMessage();
 179         }
 180     }
 181
 182     /**
 183      * Close producer Dmaap client.
 184      */
 185     @Override
 186     public void close() {
 187         if ((clients == null) || (clients.isEmpty())) {
 188             return;
 189         }
 190         LOG.debug("Closing Dmaap producer clients....");
 191         for (MRBatchingPublisher client : clients) {
 192             try {
 193                 client.close(1, TimeUnit.SECONDS);
 194             } catch (IOException | InterruptedException e) {
 195                 LOG.warn(String.format("Failed to cleanly close Dmaap connection for [%s]", client), e);
 196             }
 197         }
 198     }
 199
 200     @Override
 201     public void useHttps(boolean yes) {
 202         useHttps = yes;
 203     }
 204
 205     protected MetricService getMetricservice() {
 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         } else {
 212             LOG.info("Metric Service error from bundlecontext");
 213             LOG.warn("Cannot find service reference for org.onap.appc.metricservice.MetricService");
 214             return null;
 215         }
 216     }
 217
 218     public Properties getProperties() {
 219         return props;
 220     }
 221
 222     public boolean isHttps() {
 223         return useHttps;
 224     }
 225
 226 }