gerrit.onap Code Review - policy/apex-pdp.git/blob

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