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