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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * policy-endpoints
   4  * ================================================================================
   5  * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.common.endpoints.event.comm.bus;
  22
  23 import java.util.ArrayList;
  24 import java.util.Arrays;
  25 import java.util.HashMap;
  26 import java.util.List;
  27 import java.util.Properties;
  28
  29 import org.onap.policy.common.endpoints.event.comm.bus.internal.SingleThreadedUebTopicSource;
  30 import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33
  34 /**
  35  * UEB Topic Source Factory
  36  */
  37 public interface UebTopicSourceFactory {
  38
  39     /**
  40      * Creates an UEB Topic Source based on properties files
  41      *
  42      * @param properties Properties containing initialization values
  43      *
  44      * @return an UEB Topic Source
  45      * @throws IllegalArgumentException if invalid parameters are present
  46      */
  47     public List<UebTopicSource> build(Properties properties);
  48
  49     /**
  50      * Instantiates a new UEB Topic Source
  51      *
  52      * @param servers list of servers
  53      * @param topic topic name
  54      * @param apiKey API Key
  55      * @param apiSecret API Secret
  56      * @param consumerGroup Consumer Group
  57      * @param consumerInstance Consumer Instance
  58      * @param fetchTimeout Read Fetch Timeout
  59      * @param fetchLimit Fetch Limit
  60      * @param managed is this source endpoint managed?
  61      *
  62      * @return an UEB Topic Source
  63      * @throws IllegalArgumentException if invalid parameters are present
  64      */
  65     public UebTopicSource build(List<String> servers, String topic, String apiKey, String apiSecret,
  66             String consumerGroup, String consumerInstance, int fetchTimeout, int fetchLimit, boolean managed,
  67             boolean useHttps, boolean allowSelfSignedCerts);
  68
  69     /**
  70      * Instantiates a new UEB Topic Source
  71      *
  72      * @param servers list of servers
  73      * @param topic topic name
  74      * @param apiKey API Key
  75      * @param apiSecret API Secret
  76      *
  77      * @return an UEB Topic Source
  78      * @throws IllegalArgumentException if invalid parameters are present
  79      */
  80     public UebTopicSource build(List<String> servers, String topic, String apiKey, String apiSecret);
  81
  82     /**
  83      * Instantiates a new UEB Topic Source
  84      *
  85      * @param servers list of servers
  86      * @param topic topic name
  87      *
  88      * @return an UEB Topic Source
  89      * @throws IllegalArgumentException if invalid parameters are present
  90      */
  91     public UebTopicSource build(List<String> servers, String topic);
  92
  93     /**
  94      * Destroys an UEB Topic Source based on a topic
  95      *
  96      * @param topic topic name
  97      * @throws IllegalArgumentException if invalid parameters are present
  98      */
  99     public void destroy(String topic);
 100
 101     /**
 102      * Destroys all UEB Topic Sources
 103      */
 104     public void destroy();
 105
 106     /**
 107      * gets an UEB Topic Source based on topic name
 108      *
 109      * @param topic the topic name
 110      * @return an UEB Topic Source with topic name
 111      * @throws IllegalArgumentException if an invalid topic is provided
 112      * @throws IllegalStateException if the UEB Topic Source is an incorrect state
 113      */
 114     public UebTopicSource get(String topic);
 115
 116     /**
 117      * Provides a snapshot of the UEB Topic Sources
 118      *
 119      * @return a list of the UEB Topic Sources
 120      */
 121     public List<UebTopicSource> inventory();
 122 }
 123
 124
 125 /* ------------- implementation ----------------- */
 126
 127 /**
 128  * Factory of UEB Source Topics indexed by topic name
 129  */
 130 class IndexedUebTopicSourceFactory implements UebTopicSourceFactory {
 131     private static final String MISSING_TOPIC = "A topic must be provided";
 132
 133     /**
 134      * Logger
 135      */
 136     private static Logger logger = LoggerFactory.getLogger(IndexedUebTopicSourceFactory.class);
 137
 138     /**
 139      * UEB Topic Name Index
 140      */
 141     protected HashMap<String, UebTopicSource> uebTopicSources = new HashMap<>();
 142
 143     /**
 144      * {@inheritDoc}
 145      */
 146     @Override
 147     public UebTopicSource build(List<String> servers, String topic, String apiKey, String apiSecret,
 148             String consumerGroup, String consumerInstance, int fetchTimeout, int fetchLimit, boolean managed,
 149             boolean useHttps, boolean allowSelfSignedCerts) {
 150         if (servers == null || servers.isEmpty()) {
 151             throw new IllegalArgumentException("UEB Server(s) must be provided");
 152         }
 153
 154         if (topic == null || topic.isEmpty()) {
 155             throw new IllegalArgumentException(MISSING_TOPIC);
 156         }
 157
 158         synchronized (this) {
 159             if (uebTopicSources.containsKey(topic)) {
 160                 return uebTopicSources.get(topic);
 161             }
 162
 163             UebTopicSource uebTopicSource = new SingleThreadedUebTopicSource(servers, topic, apiKey, apiSecret,
 164                     consumerGroup, consumerInstance, fetchTimeout, fetchLimit, useHttps, allowSelfSignedCerts);
 165
 166             if (managed) {
 167                 uebTopicSources.put(topic, uebTopicSource);
 168             }
 169
 170             return uebTopicSource;
 171         }
 172     }
 173
 174     /**
 175      * {@inheritDoc}
 176      */
 177     @Override
 178     public List<UebTopicSource> build(Properties properties) {
 179
 180         String readTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS);
 181         if (readTopics == null || readTopics.isEmpty()) {
 182             logger.info("{}: no topic for UEB Source", this);
 183             return new ArrayList<>();
 184         }
 185         List<String> readTopicList = new ArrayList<>(Arrays.asList(readTopics.split("\\s*,\\s*")));
 186
 187         List<UebTopicSource> newUebTopicSources = new ArrayList<>();
 188         synchronized (this) {
 189             for (String topic : readTopicList) {
 190                 if (this.uebTopicSources.containsKey(topic)) {
 191                     newUebTopicSources.add(this.uebTopicSources.get(topic));
 192                     continue;
 193                 }
 194
 195                 String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
 196                         + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
 197
 198                 if (servers == null || servers.isEmpty()) {
 199                     logger.error("{}: no UEB servers configured for sink {}", this, topic);
 200                     continue;
 201                 }
 202
 203                 List<String> serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
 204
 205                 String apiKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "." + topic
 206                         + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
 207
 208                 String apiSecret = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
 209                         + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
 210
 211                 String consumerGroup = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
 212                         + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX);
 213
 214                 String consumerInstance = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS
 215                         + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX);
 216
 217                 String fetchTimeoutString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS
 218                         + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX);
 219                 int fetchTimeout = UebTopicSource.DEFAULT_TIMEOUT_MS_FETCH;
 220                 if (fetchTimeoutString != null && !fetchTimeoutString.isEmpty()) {
 221                     try {
 222                         fetchTimeout = Integer.parseInt(fetchTimeoutString);
 223                     } catch (NumberFormatException nfe) {
 224                         logger.warn("{}: fetch timeout {} is in invalid format for topic {} ", this, fetchTimeoutString,
 225                                 topic);
 226                     }
 227                 }
 228
 229                 String fetchLimitString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS
 230                         + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX);
 231                 int fetchLimit = UebTopicSource.DEFAULT_LIMIT_FETCH;
 232                 if (fetchLimitString != null && !fetchLimitString.isEmpty()) {
 233                     try {
 234                         fetchLimit = Integer.parseInt(fetchLimitString);
 235                     } catch (NumberFormatException nfe) {
 236                         logger.warn("{}: fetch limit {} is in invalid format for topic {} ", this, fetchLimitString,
 237                                 topic);
 238                     }
 239                 }
 240
 241                 String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
 242                         + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
 243                 boolean managed = true;
 244                 if (managedString != null && !managedString.isEmpty()) {
 245                     managed = Boolean.parseBoolean(managedString);
 246                 }
 247
 248                 String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
 249                         + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
 250
 251                 // default is to use HTTP if no https property exists
 252                 boolean useHttps = false;
 253                 if (useHttpsString != null && !useHttpsString.isEmpty()) {
 254                     useHttps = Boolean.parseBoolean(useHttpsString);
 255                 }
 256
 257                 String allowSelfSignedCertsString =
 258                         properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "." + topic
 259                                 + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
 260
 261                 // default is to disallow self-signed certs
 262                 boolean allowSelfSignedCerts = false;
 263                 if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
 264                     allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
 265                 }
 266
 267                 UebTopicSource uebTopicSource = this.build(serverList, topic, apiKey, apiSecret, consumerGroup,
 268                         consumerInstance, fetchTimeout, fetchLimit, managed, useHttps, allowSelfSignedCerts);
 269                 newUebTopicSources.add(uebTopicSource);
 270             }
 271         }
 272         return newUebTopicSources;
 273     }
 274
 275     /**
 276      * {@inheritDoc}
 277      */
 278     @Override
 279     public UebTopicSource build(List<String> servers, String topic, String apiKey, String apiSecret) {
 280
 281         return this.build(servers, topic, apiKey, apiSecret, null, null, UebTopicSource.DEFAULT_TIMEOUT_MS_FETCH,
 282                 UebTopicSource.DEFAULT_LIMIT_FETCH, true, false, true);
 283     }
 284
 285     /**
 286      * {@inheritDoc}
 287      */
 288     @Override
 289     public UebTopicSource build(List<String> servers, String topic) {
 290         return this.build(servers, topic, null, null);
 291     }
 292
 293     /**
 294      * {@inheritDoc}
 295      */
 296     @Override
 297     public void destroy(String topic) {
 298
 299         if (topic == null || topic.isEmpty()) {
 300             throw new IllegalArgumentException(MISSING_TOPIC);
 301         }
 302
 303         UebTopicSource uebTopicSource;
 304
 305         synchronized (this) {
 306             if (!uebTopicSources.containsKey(topic)) {
 307                 return;
 308             }
 309
 310             uebTopicSource = uebTopicSources.remove(topic);
 311         }
 312
 313         uebTopicSource.shutdown();
 314     }
 315
 316     /**
 317      * {@inheritDoc}
 318      */
 319     @Override
 320     public UebTopicSource get(String topic) {
 321
 322         if (topic == null || topic.isEmpty()) {
 323             throw new IllegalArgumentException(MISSING_TOPIC);
 324         }
 325
 326         synchronized (this) {
 327             if (uebTopicSources.containsKey(topic)) {
 328                 return uebTopicSources.get(topic);
 329             } else {
 330                 throw new IllegalStateException("UebTopiceSource for " + topic + " not found");
 331             }
 332         }
 333     }
 334
 335     @Override
 336     public synchronized List<UebTopicSource> inventory() {
 337         return new ArrayList<>(this.uebTopicSources.values());
 338     }
 339
 340     @Override
 341     public void destroy() {
 342         List<UebTopicSource> readers = this.inventory();
 343         for (UebTopicSource reader : readers) {
 344             reader.shutdown();
 345         }
 346
 347         synchronized (this) {
 348             this.uebTopicSources.clear();
 349         }
 350     }
 351
 352     @Override
 353     public String toString() {
 354         StringBuilder builder = new StringBuilder();
 355         builder.append("IndexedUebTopicSourceFactory []");
 356         return builder.toString();
 357     }
 358
 359 }