plugins/plugins-event/plugins-event-carrier/plugins-event-carrier-restrequestor/src/main/java/org/onap/policy/apex/plugins/event/carrier/restrequestor/ApexRestRequestorConsumer.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2016-2018 Ericsson. All rights reserved.
   4  *  Modifications Copyright (C) 2019-2020 Nordix Foundation.
   5  * ================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  * SPDX-License-Identifier: Apache-2.0
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.policy.apex.plugins.event.carrier.restrequestor;
  23
  24 import java.net.URL;
  25 import java.util.ArrayList;
  26 import java.util.Arrays;
  27 import java.util.List;
  28 import java.util.Map;
  29 import java.util.Map.Entry;
  30 import java.util.Optional;
  31 import java.util.Properties;
  32 import java.util.Set;
  33 import java.util.concurrent.BlockingQueue;
  34 import java.util.concurrent.ConcurrentHashMap;
  35 import java.util.concurrent.LinkedBlockingQueue;
  36 import java.util.concurrent.TimeUnit;
  37 import java.util.regex.Matcher;
  38 import java.util.regex.Pattern;
  39 import javax.ws.rs.client.Client;
  40 import javax.ws.rs.client.ClientBuilder;
  41 import javax.ws.rs.client.Entity;
  42 import javax.ws.rs.client.Invocation.Builder;
  43 import javax.ws.rs.core.Response;
  44 import org.apache.commons.lang3.StringUtils;
  45 import org.onap.policy.apex.core.infrastructure.threading.ThreadUtilities;
  46 import org.onap.policy.apex.service.engine.event.ApexEventException;
  47 import org.onap.policy.apex.service.engine.event.ApexEventReceiver;
  48 import org.onap.policy.apex.service.engine.event.ApexEventRuntimeException;
  49 import org.onap.policy.apex.service.engine.event.ApexPluginsEventConsumer;
  50 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
  51 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerPeeredMode;
  52 import org.slf4j.Logger;
  53 import org.slf4j.LoggerFactory;
  54
  55 /**
  56  * This class implements an Apex event consumer that issues a REST request and returns the REST response to APEX as an
  57  * event.
  58  *
  59  * @author Liam Fallon (liam.fallon@ericsson.com)
  60  */
  61 public class ApexRestRequestorConsumer extends ApexPluginsEventConsumer {
  62     // Get a reference to the logger
  63     private static final Logger LOGGER = LoggerFactory.getLogger(ApexRestRequestorConsumer.class);
  64
  65     // The amount of time to wait in milliseconds between checks that the consumer thread has
  66     // stopped
  67     private static final long REST_REQUESTOR_WAIT_SLEEP_TIME = 50;
  68
  69     // The Key for property
  70     private static final String HTTP_CODE_STATUS = "HTTP_CODE_STATUS";
  71
  72     // The REST parameters read from the parameter service
  73     private RestRequestorCarrierTechnologyParameters restConsumerProperties;
  74
  75     // The timeout for REST requests
  76     private long restRequestTimeout = RestRequestorCarrierTechnologyParameters.DEFAULT_REST_REQUEST_TIMEOUT;
  77
  78     // The event receiver that will receive events from this consumer
  79     private ApexEventReceiver eventReceiver;
  80
  81     // The HTTP client that makes a REST call to get an input event for Apex
  82     private Client client;
  83
  84     // Temporary request holder for incoming REST requests
  85     private final BlockingQueue<ApexRestRequest> incomingRestRequestQueue = new LinkedBlockingQueue<>();
  86
  87     // Map of ongoing REST request threads indexed by the time they started at
  88     private final Map<ApexRestRequest, RestRequestRunner> ongoingRestRequestMap = new ConcurrentHashMap<>();
  89
  90     // The number of events received to date
  91     private Object eventsReceivedLock = new Object();
  92     private Integer eventsReceived = 0;
  93
  94     // The number of the next request runner thread
  95     private static long nextRequestRunnerThreadNo = 0;
  96
  97     private String untaggedUrl = null;
  98
  99     // The pattern for filtering status code
 100     private Pattern httpCodeFilterPattern = null;
 101
 102     @Override
 103     public void init(final String consumerName, final EventHandlerParameters consumerParameters,
 104         final ApexEventReceiver incomingEventReceiver) throws ApexEventException {
 105         this.eventReceiver = incomingEventReceiver;
 106         this.name = consumerName;
 107
 108         // Check and get the REST Properties
 109         if (!(consumerParameters
 110             .getCarrierTechnologyParameters() instanceof RestRequestorCarrierTechnologyParameters)) {
 111             final String errorMessage =
 112                 "specified consumer properties are not applicable to REST Requestor consumer (" + this.name + ")";
 113             throw new ApexEventException(errorMessage);
 114         }
 115         restConsumerProperties =
 116             (RestRequestorCarrierTechnologyParameters) consumerParameters.getCarrierTechnologyParameters();
 117
 118         // Check if we are in peered mode
 119         if (!consumerParameters.isPeeredMode(EventHandlerPeeredMode.REQUESTOR)) {
 120             final String errorMessage = "REST Requestor consumer (" + this.name
 121                 + ") must run in peered requestor mode with a REST Requestor producer";
 122             throw new ApexEventException(errorMessage);
 123         }
 124
 125         // Check if the HTTP method has been set
 126         if (restConsumerProperties.getHttpMethod() == null) {
 127             restConsumerProperties
 128                 .setHttpMethod(RestRequestorCarrierTechnologyParameters.DEFAULT_REQUESTOR_HTTP_METHOD);
 129         }
 130
 131         // Check if the HTTP URL has been set
 132         if (restConsumerProperties.getUrl() == null) {
 133             final String errorMessage = "no URL has been specified on REST Requestor consumer (" + this.name + ")";
 134             throw new ApexEventException(errorMessage);
 135         }
 136
 137         // Check if the HTTP URL is valid
 138         try {
 139             new URL(restConsumerProperties.getUrl());
 140         } catch (final Exception e) {
 141             final String errorMessage = "invalid URL has been specified on REST Requestor consumer (" + this.name + ")";
 142             throw new ApexEventException(errorMessage, e);
 143         }
 144
 145         this.httpCodeFilterPattern = Pattern.compile(restConsumerProperties.getHttpCodeFilter());
 146
 147         // Set the requestor timeout
 148         if (consumerParameters.getPeerTimeout(EventHandlerPeeredMode.REQUESTOR) != 0) {
 149             restRequestTimeout = consumerParameters.getPeerTimeout(EventHandlerPeeredMode.REQUESTOR);
 150         }
 151
 152         // Check if HTTP headers has been set
 153         if (restConsumerProperties.checkHttpHeadersSet()) {
 154             final String httpHeaderString = Arrays.deepToString(restConsumerProperties.getHttpHeaders());
 155             LOGGER.debug("REST Requestor consumer has http headers ({}): {}", this.name, httpHeaderString);
 156         }
 157
 158         // Initialize the HTTP client
 159         client = ClientBuilder.newClient();
 160     }
 161
 162     /**
 163      * Receive an incoming REST request from the peered REST Requestor producer and queue it.
 164      *
 165      * @param restRequest the incoming rest request to queue
 166      * @throws ApexEventRuntimeException on queueing errors
 167      */
 168     public void processRestRequest(final ApexRestRequest restRequest) {
 169         // Push the event onto the queue for handling
 170         try {
 171             incomingRestRequestQueue.add(restRequest);
 172         } catch (final Exception requestException) {
 173             final String errorMessage =
 174                 "could not queue request \"" + restRequest + "\" on REST Requestor consumer (" + this.name + ")";
 175             throw new ApexEventRuntimeException(errorMessage);
 176         }
 177     }
 178
 179     /**
 180      * Get the number of events received to date.
 181      *
 182      * @return the number of events received
 183      */
 184     public int getEventsReceived() {
 185         return eventsReceived;
 186     }
 187
 188     /**
 189      * {@inheritDoc}.
 190      */
 191     @Override
 192     public void run() {
 193         // The endless loop that receives events using REST calls
 194         while (consumerThread.isAlive() && !stopOrderedFlag) {
 195             try {
 196                 // Take the next event from the queue
 197                 final ApexRestRequest restRequest =
 198                     incomingRestRequestQueue.poll(REST_REQUESTOR_WAIT_SLEEP_TIME, TimeUnit.MILLISECONDS);
 199                 if (restRequest == null) {
 200                     // Poll timed out, check for request timeouts
 201                     timeoutExpiredRequests();
 202                     continue;
 203                 }
 204
 205                 Properties inputExecutionProperties = restRequest.getExecutionProperties();
 206                 untaggedUrl = restConsumerProperties.getUrl();
 207                 if (inputExecutionProperties != null) {
 208                     Set<String> names = restConsumerProperties.getKeysFromUrl();
 209                     Set<String> inputProperty = inputExecutionProperties.stringPropertyNames();
 210
 211                     names.stream().map(Optional::of)
 212                         .forEach(op -> op.filter(inputProperty::contains)
 213                             .orElseThrow(() -> new ApexEventRuntimeException(
 214                                 "key\"" + op.get() + "\"specified on url \"" + restConsumerProperties.getUrl()
 215                                     + "\"not found in execution properties passed by the current policy")));
 216
 217                     untaggedUrl = names.stream().reduce(untaggedUrl,
 218                         (acc, str) -> acc.replace("{" + str + "}", (String) inputExecutionProperties.get(str)));
 219                 }
 220
 221                 // Set the time stamp of the REST request
 222                 restRequest.setTimestamp(System.currentTimeMillis());
 223
 224                 // Create a thread to process the REST request and place it on the map of ongoing
 225                 // requests
 226                 final RestRequestRunner restRequestRunner = new RestRequestRunner(restRequest);
 227                 ongoingRestRequestMap.put(restRequest, restRequestRunner);
 228
 229                 // Start execution of the request
 230                 final Thread restRequestRunnerThread = new Thread(restRequestRunner);
 231                 restRequestRunnerThread.setName("RestRequestRunner_" + nextRequestRunnerThreadNo);
 232                 restRequestRunnerThread.start();
 233             } catch (final InterruptedException e) {
 234                 LOGGER.debug("Thread interrupted, Reason {}", e.getMessage());
 235                 Thread.currentThread().interrupt();
 236             }
 237         }
 238
 239         client.close();
 240     }
 241
 242     /**
 243      * This method times out REST requests that have expired.
 244      */
 245     private void timeoutExpiredRequests() {
 246         // Hold a list of timed out requests
 247         final List<ApexRestRequest> timedoutRequestList = new ArrayList<>();
 248
 249         // Check for timeouts
 250         for (final Entry<ApexRestRequest, RestRequestRunner> requestEntry : ongoingRestRequestMap.entrySet()) {
 251             if (System.currentTimeMillis() - requestEntry.getKey().getTimestamp() > restRequestTimeout) {
 252                 requestEntry.getValue().stop();
 253                 timedoutRequestList.add(requestEntry.getKey());
 254             }
 255         }
 256
 257         // Interrupt timed out requests and remove them from the ongoing map
 258         for (final ApexRestRequest timedoutRequest : timedoutRequestList) {
 259             final String errorMessage =
 260                 "REST Requestor consumer (" + this.name + "), REST request timed out: " + timedoutRequest;
 261             LOGGER.warn(errorMessage);
 262
 263             ongoingRestRequestMap.remove(timedoutRequest);
 264         }
 265     }
 266
 267     /**
 268      * {@inheritDoc}.
 269      */
 270     @Override
 271     public void stop() {
 272         stopOrderedFlag = true;
 273
 274         while (consumerThread.isAlive()) {
 275             ThreadUtilities.sleep(REST_REQUESTOR_WAIT_SLEEP_TIME);
 276         }
 277     }
 278
 279     /**
 280      * This class is used to start a thread for each request issued.
 281      *
 282      * @author Liam Fallon (liam.fallon@ericsson.com)
 283      */
 284     private class RestRequestRunner implements Runnable {
 285         private static final String APPLICATION_JSON = "application/json";
 286
 287         // The REST request being processed by this thread
 288         private final ApexRestRequest request;
 289
 290         // The thread executing the REST request
 291         private Thread restRequestThread;
 292
 293         /**
 294          * Constructor, initialise the request runner with the request.
 295          *
 296          * @param request the request this runner will issue
 297          */
 298         private RestRequestRunner(final ApexRestRequest request) {
 299             this.request = request;
 300         }
 301
 302         /**
 303          * {@inheritDoc}.
 304          */
 305         @Override
 306         public void run() {
 307             // Get the thread for the request
 308             restRequestThread = Thread.currentThread();
 309
 310             try {
 311                 // Execute the REST request
 312                 final Response response = sendEventAsRestRequest(untaggedUrl);
 313
 314                 // Match the return code
 315                 Matcher isPass = httpCodeFilterPattern.matcher(String.valueOf(response.getStatus()));
 316
 317                 // Check that the request worked
 318                 if (!isPass.matches()) {
 319                     final String errorMessage = "reception of event from URL \"" + restConsumerProperties.getUrl()
 320                         + "\" failed with status code " + response.getStatus() + " and message \""
 321                         + response.readEntity(String.class) + "\"";
 322                     throw new ApexEventRuntimeException(errorMessage);
 323                 }
 324
 325                 // Get the event we received
 326                 final String eventJsonString = response.readEntity(String.class);
 327
 328                 // Check there is content
 329                 if (StringUtils.isBlank(eventJsonString)) {
 330                     final String errorMessage =
 331                         "received an empty response to \"" + request + "\" from URL \"" + untaggedUrl + "\"";
 332                     throw new ApexEventRuntimeException(errorMessage);
 333                 }
 334
 335                 // build a key and value property in excutionProperties
 336                 Properties executionProperties = new Properties();
 337                 executionProperties.put(HTTP_CODE_STATUS, response.getStatus());
 338
 339                 // Send the event into Apex
 340                 eventReceiver.receiveEvent(request.getExecutionId(), executionProperties, eventJsonString);
 341
 342                 synchronized (eventsReceivedLock) {
 343                     eventsReceived++;
 344                 }
 345             } catch (final Exception e) {
 346                 LOGGER.warn("error receiving events on thread {}", consumerThread.getName(), e);
 347             } finally {
 348                 // Remove the request from the map of ongoing requests
 349                 ongoingRestRequestMap.remove(request);
 350             }
 351         }
 352
 353         /**
 354          * Stop the REST request.
 355          */
 356         private void stop() {
 357             restRequestThread.interrupt();
 358         }
 359
 360         /**
 361          * Execute the REST request.
 362          *
 363          *
 364          * @return the response to the REST request
 365          */
 366         public Response sendEventAsRestRequest(String untaggedUrl) {
 367             Builder headers = client.target(untaggedUrl).request(APPLICATION_JSON)
 368                 .headers(restConsumerProperties.getHttpHeadersAsMultivaluedMap());
 369             switch (restConsumerProperties.getHttpMethod()) {
 370                 case GET:
 371                     return headers.get();
 372
 373                 case PUT:
 374                     return headers.put(Entity.json(request.getEvent()));
 375
 376                 case POST:
 377                     return headers.post(Entity.json(request.getEvent()));
 378
 379                 case DELETE:
 380                     return headers.delete();
 381
 382                 default:
 383                     break;
 384             }
 385
 386             return null;
 387         }
 388     }
 389 }