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