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

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