champ-service/src/main/java/org/onap/champ/async/ChampAsyncRequestProcessor.java

   1 /**
   2  * ============LICENSE_START==========================================
   3  * org.onap.aai
   4  * ===================================================================
   5  * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
   6  * Copyright © 2017-2018 Amdocs
   7  * ===================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *     http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END============================================
  20  */
  21 package org.onap.champ.async;
  22
  23 import java.util.Optional;
  24 import java.util.TimerTask;
  25 import java.util.concurrent.ArrayBlockingQueue;
  26 import java.util.concurrent.BlockingQueue;
  27 import java.util.concurrent.Executors;
  28 import java.util.concurrent.ThreadPoolExecutor;
  29 import javax.naming.OperationNotSupportedException;
  30 import javax.ws.rs.core.Response.Status;
  31 import org.onap.aai.champcore.ChampTransaction;
  32 import org.onap.aai.cl.api.Logger;
  33 import org.onap.aai.cl.eelf.LoggerFactory;
  34 import org.onap.aai.event.api.EventConsumer;
  35 import org.onap.champ.ChampRESTAPI;
  36 import org.onap.champ.event.GraphEvent;
  37 import org.onap.champ.event.GraphEvent.GraphEventResult;
  38 import org.onap.champ.event.GraphEventEdge;
  39 import org.onap.champ.event.GraphEventVertex;
  40 import org.onap.champ.event.envelope.GraphEventEnvelope;
  41 import org.onap.champ.event.envelope.GraphEventHeader;
  42 import org.onap.champ.exception.ChampServiceException;
  43 import org.onap.champ.service.ChampDataService;
  44 import org.onap.champ.service.ChampThreadFactory;
  45 import org.onap.champ.service.logging.ChampMsgs;
  46
  47 /**
  48  * This Class polls the Graph events from request topic perform the necessary CRUD operation by calling champDAO and
  49  * queues up the response to be consumed by response handler.
  50  */
  51 public class ChampAsyncRequestProcessor extends TimerTask {
  52
  53     private Logger logger = LoggerFactory.getInstance().getLogger(ChampAsyncRequestProcessor.class);
  54
  55     private ChampDataService champDataService;
  56
  57     /**
  58      * Number of events that can be queued up.
  59      */
  60     private Integer requestProcesserQueueSize;
  61
  62     /**
  63      * Number of event publisher worker threads.
  64      */
  65     private Integer requestProcesserPoolSize;
  66
  67     /**
  68      * Number of event publisher worker threads.
  69      */
  70     private Integer requestPollingTimeSeconds;
  71
  72     /**
  73      * Internal queue where outgoing events will be buffered until they can be serviced by.
  74      **/
  75     private BlockingQueue<GraphEventEnvelope> requestProcesserEventQueue;
  76
  77     /**
  78      * Pool of worker threads that do the work of publishing the events to the event bus.
  79      */
  80     private ThreadPoolExecutor requestProcesserPool;
  81
  82     private ChampAsyncResponsePublisher champAsyncResponsePublisher;
  83
  84     private EventConsumer asyncRequestConsumer;
  85
  86     private static final Integer DEFAULT_ASYNC_REQUEST_PROCESS_QUEUE_CAPACITY = 10000;
  87
  88     private static final Integer DEFAULT_ASYNC_REQUEST_PROCESS_THREAD_POOL_SIZE = 10;
  89     private static final Integer DEFAULT_ASYNC_REQUEST_PROCESS_POLLING_SECOND = 30000;
  90     private static final String CHAMP_GRAPH_REQUEST_PROCESS_THREAD_NAME = "ChampAsyncGraphRequestEventProcessor";
  91     Logger auditLogger = LoggerFactory.getInstance().getAuditLogger(ChampRESTAPI.class.getName());
  92
  93     public ChampAsyncRequestProcessor(ChampDataService champDataService,
  94             ChampAsyncResponsePublisher champAsyncResponsePublisher, EventConsumer asyncRequestConsumer) {
  95
  96         this.requestProcesserQueueSize = DEFAULT_ASYNC_REQUEST_PROCESS_QUEUE_CAPACITY;
  97
  98         this.requestProcesserPoolSize = DEFAULT_ASYNC_REQUEST_PROCESS_THREAD_POOL_SIZE;
  99
 100         this.requestPollingTimeSeconds = DEFAULT_ASYNC_REQUEST_PROCESS_POLLING_SECOND;
 101         requestProcesserEventQueue = new ArrayBlockingQueue<>(requestProcesserQueueSize);
 102         requestProcesserPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(requestProcesserPoolSize,
 103                 new ChampThreadFactory(CHAMP_GRAPH_REQUEST_PROCESS_THREAD_NAME));
 104
 105         for (int i = 0; i < requestProcesserPoolSize; i++) {
 106             requestProcesserPool.submit(new ChampProcessorWorker());
 107         }
 108
 109         this.champDataService = champDataService;
 110         this.champAsyncResponsePublisher = champAsyncResponsePublisher;
 111         this.asyncRequestConsumer = asyncRequestConsumer;
 112         logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO,
 113                 "ChampAsyncRequestProcessor initialized SUCCESSFULLY! with event consumer "
 114                         + asyncRequestConsumer.getClass().getName());
 115     }
 116
 117     public ChampAsyncRequestProcessor(ChampDataService champDataService,
 118             ChampAsyncResponsePublisher champAsyncResponsePublisher, EventConsumer asyncRequestConsumer,
 119             Integer requestProcesserQueueSize, Integer requestProcesserPoolSize, Integer requestPollingTimeSeconds) {
 120
 121         this.requestProcesserQueueSize = requestProcesserQueueSize;
 122
 123         this.requestProcesserPoolSize = requestProcesserPoolSize;
 124
 125         this.requestPollingTimeSeconds = requestPollingTimeSeconds;
 126
 127         requestProcesserEventQueue = new ArrayBlockingQueue<>(requestProcesserQueueSize);
 128         requestProcesserPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(requestProcesserPoolSize,
 129                 new ChampThreadFactory(CHAMP_GRAPH_REQUEST_PROCESS_THREAD_NAME));
 130
 131         for (int i = 0; i < requestProcesserPoolSize; i++) {
 132             requestProcesserPool.submit(new ChampProcessorWorker());
 133         }
 134
 135         this.champDataService = champDataService;
 136         this.champAsyncResponsePublisher = champAsyncResponsePublisher;
 137         this.asyncRequestConsumer = asyncRequestConsumer;
 138         logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO,
 139                 "ChampAsyncRequestProcessor initialized SUCCESSFULLY! with event consumer "
 140                         + asyncRequestConsumer.getClass().getName());
 141     }
 142
 143     private class ChampProcessorWorker implements Runnable {
 144
 145         @Override
 146         public void run() {
 147
 148             while (true) {
 149
 150                 GraphEventEnvelope eventEnvelope = null;
 151                 GraphEvent event = null;
 152                 try {
 153                     // Get the next event to be published from the queue.
 154                     eventEnvelope = requestProcesserEventQueue.take();
 155                     event = eventEnvelope.getBody();
 156                 } catch (InterruptedException e) {
 157                     // Restore the interrupted status.
 158                     Thread.currentThread().interrupt();
 159                 }
 160
 161                 // Apply Champ Event header
 162                 eventEnvelope.setHeader(new GraphEventHeader.Builder().requestId(event.getTransactionId()).build());
 163
 164                 // Parse the event and call champ Dao to process , Create the
 165                 // response event and put it on response queue
 166                 event.setResult(GraphEventResult.SUCCESS);
 167
 168                 // Check if this request is part of an ongoing DB transaction
 169                 ChampTransaction transaction = champDataService.getTransaction(event.getDbTransactionId());
 170                 if ((event.getDbTransactionId() != null) && (transaction == null)) {
 171                     event.setResult(GraphEventResult.FAILURE);
 172                     event.setErrorMessage("Database transactionId " + event.getDbTransactionId() + " not found");
 173                     event.setHttpErrorStatus(Status.BAD_REQUEST);
 174                 }
 175
 176                 if (event.getResult() != GraphEventResult.FAILURE) {
 177                     try {
 178                         if (event.getVertex() != null) {
 179
 180                             switch (event.getOperation()) {
 181                                 case CREATE:
 182                                     event.setVertex(GraphEventVertex.fromChampObject(
 183                                             champDataService.storeObject(event.getVertex().toChampObject(),
 184                                                     Optional.ofNullable(transaction)),
 185                                             event.getVertex().getModelVersion()));
 186                                     break;
 187
 188                                 case UPDATE:
 189                                     event.setVertex(GraphEventVertex.fromChampObject(
 190                                             champDataService.replaceObject(event.getVertex().toChampObject(event.getVertex().toJson()),
 191                                                     event.getVertex().getId(), Optional.ofNullable(transaction)),
 192                                             event.getVertex().getModelVersion()));
 193                                     break;
 194                                 case DELETE:
 195                                     champDataService.deleteObject(event.getVertex().getId(),
 196                                             Optional.ofNullable(transaction));
 197                                     break;
 198                                 default:
 199                                     // log error
 200                             }
 201                         } else if (event.getEdge() != null) {
 202                             switch (event.getOperation()) {
 203                                 case CREATE:
 204                                     event.setEdge(GraphEventEdge.fromChampRelationship(
 205                                             champDataService.storeRelationship(event.getEdge().toChampRelationship(),
 206                                                     Optional.ofNullable(transaction)),
 207                                             event.getEdge().getModelVersion()));
 208                                     break;
 209
 210                                 case UPDATE:
 211                                     event.setEdge(GraphEventEdge.fromChampRelationship(
 212                                             champDataService.updateRelationship(event.getEdge().toChampRelationship(),
 213                                                     event.getEdge().getId(), Optional.ofNullable(transaction)),
 214                                             event.getEdge().getModelVersion()));
 215
 216                                     break;
 217                                 case DELETE:
 218                                     champDataService.deleteRelationship(event.getEdge().getId(),
 219                                             Optional.ofNullable(transaction));
 220                                     break;
 221                                 default:
 222                                     logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR,
 223                                             "Invalid operation for event transactionId: " + event.getTransactionId());
 224                             }
 225
 226                         } else {
 227                             logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR,
 228                                     "Invalid payload for event transactionId: " + event.getTransactionId());
 229                         }
 230                     } catch (ChampServiceException champException) {
 231                         logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, champException.getMessage());
 232                         event.setResult(GraphEventResult.FAILURE);
 233                         event.setErrorMessage(champException.getMessage());
 234                         event.setHttpErrorStatus(champException.getHttpStatus());
 235
 236                     } catch (Exception ex) {
 237                         logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, ex.getMessage());
 238                         event.setResult(GraphEventResult.FAILURE);
 239                         event.setErrorMessage(ex.getMessage());
 240                         event.setHttpErrorStatus(Status.INTERNAL_SERVER_ERROR);
 241                     }
 242                 }
 243
 244                 if (event.getResult().equals(GraphEventResult.SUCCESS)) {
 245                     logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO,
 246                             "Event processed of type: " + event.getObjectType() + " with key: " + event.getObjectKey()
 247                                     + " , transaction-id: " + event.getTransactionId() + " , operation: "
 248                                     + event.getOperation().toString() + " , result: " + event.getResult());
 249                 } else {
 250                     logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO,
 251                             "Event processed of type: " + event.getObjectType() + " with key: " + event.getObjectKey()
 252                                     + " , transaction-id: " + event.getTransactionId() + " , operation: "
 253                                     + event.getOperation().toString() + " , result: " + event.getResult() + " , error: "
 254                                     + event.getErrorMessage());
 255                 }
 256
 257                 champAsyncResponsePublisher.publishResponseEvent(eventEnvelope);
 258
 259             }
 260         }
 261     }
 262
 263     @Override
 264     public void run() {
 265
 266         logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, "Listening for graph events");
 267
 268         if (asyncRequestConsumer == null) {
 269             logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR,
 270                     "Unable to initialize ChampAsyncRequestProcessor");
 271         }
 272
 273         Iterable<String> events = null;
 274         try {
 275             events = asyncRequestConsumer.consume();
 276         } catch (Exception e) {
 277             logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, e.getMessage());
 278             return;
 279         }
 280
 281         if (events == null || !events.iterator().hasNext()) {
 282             logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, "No events recieved");
 283
 284         }
 285
 286         for (String event : events) {
 287             try {
 288                 GraphEventEnvelope requestEnvelope = GraphEventEnvelope.fromJson(event);
 289                 GraphEvent requestEvent = requestEnvelope.getBody();
 290                 auditLogger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO,
 291                         "Event received of type: " + requestEvent.getObjectType() + " with key: "
 292                                 + requestEvent.getObjectKey() + " , transaction-id: " + requestEvent.getTransactionId()
 293                                 + " , operation: " + requestEvent.getOperation().toString());
 294                 logger.info(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO,
 295                         "Event received of type: " + requestEvent.getObjectType() + " with key: "
 296                                 + requestEvent.getObjectKey() + " , transaction-id: " + requestEvent.getTransactionId()
 297                                 + " , operation: " + requestEvent.getOperation().toString());
 298                 logger.debug(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_INFO, "Event received with payload:" + event);
 299
 300                 // Try to submit the event to be published to the event bus.
 301                 if (!requestProcesserEventQueue.offer(requestEnvelope)) {
 302                     logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR,
 303                             "Event could not be published to the event bus due to: Internal buffer capacity exceeded.");
 304                 }
 305
 306             } catch (Exception e) {
 307                 logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_ERROR, e.getMessage());
 308             }
 309         }
 310
 311         try {
 312             asyncRequestConsumer.commitOffsets();
 313         } catch (OperationNotSupportedException e) {
 314             // Dmaap doesnt support commit with offset
 315             logger.debug(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_WARN, e.getMessage());
 316         } catch (Exception e) {
 317             logger.error(ChampMsgs.CHAMP_ASYNC_REQUEST_PROCESSOR_WARN, e.getMessage());
 318         }
 319
 320     }
 321
 322     public Integer getRequestPollingTimeSeconds() {
 323         return requestPollingTimeSeconds;
 324     }
 325
 326 }