gerrit.onap Code Review - policy/common.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * policy-endpoints
   4  * ================================================================================
   5  * Copyright (C) 2017-2021 AT&T Intellectual Property. All rights reserved.
   6  * Modifications Copyright (C) 2018 Samsung Electronics Co., Ltd.
   7  * Modifications Copyright (C) 2020,2023 Bell Canada. All rights reserved.
   8  * Modifications Copyright (C) 2022-2023 Nordix Foundation.
   9  * ================================================================================
  10  * Licensed under the Apache License, Version 2.0 (the "License");
  11  * you may not use this file except in compliance with the License.
  12  * You may obtain a copy of the License at
  13  *
  14  *      http://www.apache.org/licenses/LICENSE-2.0
  15  *
  16  * Unless required by applicable law or agreed to in writing, software
  17  * distributed under the License is distributed on an "AS IS" BASIS,
  18  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  19  * See the License for the specific language governing permissions and
  20  * limitations under the License.
  21  * ============LICENSE_END=========================================================
  22  */
  23
  24 package org.onap.policy.common.endpoints.event.comm.bus.internal;
  25
  26 import com.att.nsa.apiClient.http.HttpClient.ConnectionType;
  27 import com.att.nsa.cambria.client.CambriaBatchingPublisher;
  28 import com.att.nsa.cambria.client.CambriaClientBuilders;
  29 import java.net.MalformedURLException;
  30 import java.security.GeneralSecurityException;
  31 import java.util.ArrayList;
  32 import java.util.List;
  33 import java.util.Map;
  34 import java.util.Properties;
  35 import java.util.UUID;
  36 import java.util.concurrent.TimeUnit;
  37 import org.apache.commons.lang3.StringUtils;
  38 import org.apache.kafka.clients.producer.KafkaProducer;
  39 import org.apache.kafka.clients.producer.Producer;
  40 import org.apache.kafka.clients.producer.ProducerConfig;
  41 import org.apache.kafka.clients.producer.ProducerRecord;
  42 import org.onap.dmaap.mr.client.impl.MRSimplerBatchPublisher;
  43 import org.onap.dmaap.mr.client.response.MRPublisherResponse;
  44 import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
  45 import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
  46 import org.onap.policy.common.gson.annotation.GsonJsonIgnore;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49
  50 public interface BusPublisher {
  51
  52     String NO_MESSAGE_PROVIDED = "No message provided";
  53     String LOG_CLOSE = "{}: CLOSE";
  54     String LOG_CLOSE_FAILED = "{}: CLOSE FAILED";
  55
  56     /**
  57      * sends a message.
  58      *
  59      * @param partitionId id
  60      * @param message the message
  61      * @return true if success, false otherwise
  62      * @throws IllegalArgumentException if no message provided
  63      */
  64     boolean send(String partitionId, String message);
  65
  66     /**
  67      * closes the publisher.
  68      */
  69     void close();
  70
  71     /**
  72      * Cambria based library publisher.
  73      */
  74     class CambriaPublisherWrapper implements BusPublisher {
  75
  76         private static final Logger logger = LoggerFactory.getLogger(CambriaPublisherWrapper.class);
  77
  78         /**
  79          * The actual Cambria publisher.
  80          */
  81         @GsonJsonIgnore
  82         protected CambriaBatchingPublisher publisher;
  83
  84         /**
  85          * Constructor.
  86          *
  87          * @param busTopicParams topic parameters
  88          */
  89         public CambriaPublisherWrapper(BusTopicParams busTopicParams) {
  90
  91             var builder = new CambriaClientBuilders.PublisherBuilder();
  92
  93             builder.usingHosts(busTopicParams.getServers()).onTopic(busTopicParams.getTopic());
  94
  95             // Set read timeout to 30 seconds (TBD: this should be configurable)
  96             builder.withSocketTimeout(30000);
  97
  98             if (busTopicParams.isUseHttps()) {
  99                 if (busTopicParams.isAllowSelfSignedCerts()) {
 100                     builder.withConnectionType(ConnectionType.HTTPS_NO_VALIDATION);
 101                 } else {
 102                     builder.withConnectionType(ConnectionType.HTTPS);
 103                 }
 104             }
 105
 106             if (busTopicParams.isApiKeyValid() && busTopicParams.isApiSecretValid()) {
 107                 builder.authenticatedBy(busTopicParams.getApiKey(), busTopicParams.getApiSecret());
 108             }
 109
 110             if (busTopicParams.isUserNameValid() && busTopicParams.isPasswordValid()) {
 111                 builder.authenticatedByHttp(busTopicParams.getUserName(), busTopicParams.getPassword());
 112             }
 113
 114             try {
 115                 this.publisher = builder.build();
 116             } catch (MalformedURLException | GeneralSecurityException e) {
 117                 throw new IllegalArgumentException(e);
 118             }
 119         }
 120
 121         @Override
 122         public boolean send(String partitionId, String message) {
 123             if (message == null) {
 124                 throw new IllegalArgumentException(NO_MESSAGE_PROVIDED);
 125             }
 126
 127             try {
 128                 this.publisher.send(partitionId, message);
 129             } catch (Exception e) {
 130                 logger.warn("{}: SEND of {} cannot be performed because of {}", this, message, e.getMessage(), e);
 131                 return false;
 132             }
 133             return true;
 134         }
 135
 136         @Override
 137         public void close() {
 138             logger.info(LOG_CLOSE, this);
 139
 140             try {
 141                 this.publisher.close();
 142             } catch (Exception e) {
 143                 logger.warn("{}: CLOSE FAILED because of {}", this, e.getMessage(), e);
 144             }
 145         }
 146
 147         @Override
 148         public String toString() {
 149             return "CambriaPublisherWrapper []";
 150         }
 151
 152     }
 153
 154     /**
 155      * Kafka based library publisher.
 156      */
 157     class KafkaPublisherWrapper implements BusPublisher {
 158
 159         private static final Logger logger = LoggerFactory.getLogger(KafkaPublisherWrapper.class);
 160         private static final String KEY_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer";
 161
 162         private final String topic;
 163
 164         /**
 165          * Kafka publisher.
 166          */
 167         private final Producer<String, String> producer;
 168         protected Properties kafkaProps;
 169
 170         /**
 171          * Kafka Publisher Wrapper.
 172          *
 173          * @param busTopicParams topic parameters
 174          */
 175         protected KafkaPublisherWrapper(BusTopicParams busTopicParams) {
 176
 177             if (busTopicParams.isTopicInvalid()) {
 178                 throw new IllegalArgumentException("No topic for Kafka");
 179             }
 180
 181             this.topic = busTopicParams.getTopic();
 182
 183             // Setup Properties for consumer
 184             kafkaProps = new Properties();
 185             kafkaProps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, busTopicParams.getServers().get(0));
 186             if (busTopicParams.isAdditionalPropsValid()) {
 187                 kafkaProps.putAll(busTopicParams.getAdditionalProps());
 188             }
 189             if (kafkaProps.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG) == null) {
 190                 kafkaProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KEY_SERIALIZER);
 191             }
 192             if (kafkaProps.get(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG) == null) {
 193                 kafkaProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KEY_SERIALIZER);
 194             }
 195
 196             producer = new KafkaProducer<>(kafkaProps);
 197         }
 198
 199         @Override
 200         public boolean send(String partitionId, String message) {
 201             if (message == null) {
 202                 throw new IllegalArgumentException(NO_MESSAGE_PROVIDED);
 203             }
 204
 205             try {
 206                 // Create the record
 207                 ProducerRecord<String, String> producerRecord =
 208                         new ProducerRecord<>(topic, UUID.randomUUID().toString(), message);
 209
 210                 this.producer.send(producerRecord);
 211                 producer.flush();
 212             } catch (Exception e) {
 213                 logger.warn("{}: SEND of {} cannot be performed because of {}", this, message, e.getMessage(), e);
 214                 return false;
 215             }
 216             return true;
 217         }
 218
 219         @Override
 220         public void close() {
 221             logger.info(LOG_CLOSE, this);
 222
 223             try {
 224                 this.producer.close();
 225             } catch (Exception e) {
 226                 logger.warn("{}: CLOSE FAILED because of {}", this, e.getMessage(), e);
 227             }
 228         }
 229
 230         @Override
 231         public String toString() {
 232             return "KafkaPublisherWrapper []";
 233         }
 234
 235     }
 236
 237     /**
 238      * DmaapClient library wrapper.
 239      */
 240     abstract class DmaapPublisherWrapper implements BusPublisher {
 241
 242         private static final Logger logger = LoggerFactory.getLogger(DmaapPublisherWrapper.class);
 243
 244         /**
 245          * MR based Publisher.
 246          */
 247         protected MRSimplerBatchPublisher publisher;
 248         protected Properties props;
 249
 250         /**
 251          * MR Publisher Wrapper.
 252          *
 253          * @param servers messaging bus hosts
 254          * @param topic topic
 255          * @param username AAF or DME2 Login
 256          * @param password AAF or DME2 Password
 257          */
 258         protected DmaapPublisherWrapper(ProtocolTypeConstants protocol, List<String> servers, String topic,
 259                 String username, String password, boolean useHttps) {
 260
 261             if (StringUtils.isBlank(topic)) {
 262                 throw new IllegalArgumentException("No topic for DMaaP");
 263             }
 264
 265             configureProtocol(topic, protocol, servers, useHttps);
 266
 267             this.publisher.logTo(LoggerFactory.getLogger(MRSimplerBatchPublisher.class.getName()));
 268
 269             this.publisher.setUsername(username);
 270             this.publisher.setPassword(password);
 271
 272             props = new Properties();
 273
 274             props.setProperty("Protocol", (useHttps ? "https" : "http"));
 275             props.setProperty("contenttype", "application/json");
 276             props.setProperty("username", username);
 277             props.setProperty("password", password);
 278
 279             props.setProperty("topic", topic);
 280
 281             this.publisher.setProps(props);
 282
 283             if (protocol == ProtocolTypeConstants.AAF_AUTH) {
 284                 this.publisher.setHost(servers.get(0));
 285             }
 286
 287             logger.info("{}: CREATION: using protocol {}", this, protocol.getValue());
 288         }
 289
 290         private void configureProtocol(String topic, ProtocolTypeConstants protocol, List<String> servers,
 291                 boolean useHttps) {
 292
 293             if (protocol == ProtocolTypeConstants.AAF_AUTH) {
 294                 if (servers == null || servers.isEmpty()) {
 295                     throw new IllegalArgumentException("No DMaaP servers or DME2 partner provided");
 296                 }
 297
 298                 ArrayList<String> dmaapServers = new ArrayList<>();
 299                 String port = useHttps ? ":3905" : ":3904";
 300                 for (String server : servers) {
 301                     dmaapServers.add(server + port);
 302                 }
 303
 304                 this.publisher = new MRSimplerBatchPublisher.Builder().againstUrls(dmaapServers).onTopic(topic).build();
 305
 306                 this.publisher.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
 307
 308             } else if (protocol == ProtocolTypeConstants.DME2) {
 309                 ArrayList<String> dmaapServers = new ArrayList<>();
 310                 dmaapServers.add("0.0.0.0:3904");
 311
 312                 this.publisher = new MRSimplerBatchPublisher.Builder().againstUrls(dmaapServers).onTopic(topic).build();
 313
 314                 this.publisher.setProtocolFlag(ProtocolTypeConstants.DME2.getValue());
 315
 316             } else {
 317                 throw new IllegalArgumentException("Invalid DMaaP protocol " + protocol);
 318             }
 319         }
 320
 321         @Override
 322         public void close() {
 323             logger.info(LOG_CLOSE, this);
 324
 325             try {
 326                 this.publisher.close(1, TimeUnit.SECONDS);
 327
 328             } catch (InterruptedException e) {
 329                 logger.warn(LOG_CLOSE_FAILED, this, e);
 330                 Thread.currentThread().interrupt();
 331
 332             } catch (Exception e) {
 333                 logger.warn(LOG_CLOSE_FAILED, this, e);
 334             }
 335         }
 336
 337         @Override
 338         public boolean send(String partitionId, String message) {
 339             if (message == null) {
 340                 throw new IllegalArgumentException(NO_MESSAGE_PROVIDED);
 341             }
 342
 343             this.publisher.setPubResponse(new MRPublisherResponse());
 344             this.publisher.send(partitionId, message);
 345             MRPublisherResponse response = this.publisher.sendBatchWithResponse();
 346             if (response != null) {
 347                 logger.debug("DMaaP publisher received {} : {}", response.getResponseCode(),
 348                         response.getResponseMessage());
 349             }
 350
 351             return true;
 352         }
 353
 354         @Override
 355         public String toString() {
 356             return "DmaapPublisherWrapper [" + "publisher.getAuthDate()=" + publisher.getAuthDate()
 357                     + ", publisher.getAuthKey()=" + publisher.getAuthKey() + ", publisher.getHost()="
 358                     + publisher.getHost() + ", publisher.getProtocolFlag()=" + publisher.getProtocolFlag()
 359                     + ", publisher.getUsername()=" + publisher.getUsername() + "]";
 360         }
 361     }
 362
 363     /**
 364      * DmaapClient library wrapper.
 365      */
 366     class DmaapAafPublisherWrapper extends DmaapPublisherWrapper {
 367         /**
 368          * MR based Publisher.
 369          */
 370         public DmaapAafPublisherWrapper(List<String> servers, String topic, String aafLogin, String aafPassword,
 371                 boolean useHttps) {
 372
 373             super(ProtocolTypeConstants.AAF_AUTH, servers, topic, aafLogin, aafPassword, useHttps);
 374         }
 375     }
 376
 377     class DmaapDmePublisherWrapper extends DmaapPublisherWrapper {
 378
 379         /**
 380          * Constructor.
 381          *
 382          * @param busTopicParams topic parameters
 383          */
 384         public DmaapDmePublisherWrapper(BusTopicParams busTopicParams) {
 385
 386             super(ProtocolTypeConstants.DME2, busTopicParams.getServers(), busTopicParams.getTopic(),
 387                     busTopicParams.getUserName(), busTopicParams.getPassword(), busTopicParams.isUseHttps());
 388
 389             String dme2RouteOffer = busTopicParams.isAdditionalPropsValid()
 390                     ? busTopicParams.getAdditionalProps().get(PolicyEndPointProperties.DME2_ROUTE_OFFER_PROPERTY)
 391                     : null;
 392
 393             validateParams(busTopicParams, dme2RouteOffer);
 394
 395             String serviceName = busTopicParams.getServers().get(0);
 396
 397             /* These are required, no defaults */
 398             props.setProperty(PolicyEndPointProperties.DME2_SERVICE_NAME_PROPERTY, serviceName);
 399
 400             BusHelper.setCommonProperties(busTopicParams, dme2RouteOffer, props);
 401
 402             props.setProperty("MethodType", "POST");
 403
 404             if (busTopicParams.isAdditionalPropsValid()) {
 405                 addAdditionalProps(busTopicParams);
 406             }
 407
 408             this.publisher.setProps(props);
 409         }
 410
 411         private void validateParams(BusTopicParams busTopicParams, String dme2RouteOffer) {
 412             BusHelper.validateBusTopicParams(busTopicParams, PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS);
 413
 414             if ((busTopicParams.isPartnerInvalid()) && StringUtils.isBlank(dme2RouteOffer)) {
 415                 throw new IllegalArgumentException("Must provide at least "
 416                         + PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + busTopicParams.getTopic()
 417                         + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX + " or "
 418                         + PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "." + busTopicParams.getTopic()
 419                         + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX + " for DME2");
 420             }
 421         }
 422
 423         private void addAdditionalProps(BusTopicParams busTopicParams) {
 424             for (Map.Entry<String, String> entry : busTopicParams.getAdditionalProps().entrySet()) {
 425                 String key = entry.getKey();
 426                 String value = entry.getValue();
 427
 428                 if (value != null) {
 429                     props.setProperty(key, value);
 430                 }
 431             }
 432         }
 433     }
 434 }