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