src/main/java/org/onap/aai/spike/service/SpikeEventProcessor.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.aai.spike.service;
  22
  23 import java.util.ArrayList;
  24 import java.util.TimerTask;
  25 import java.util.concurrent.BlockingQueue;
  26 import java.util.concurrent.ExecutionException;
  27 import java.util.concurrent.PriorityBlockingQueue;
  28 import javax.naming.OperationNotSupportedException;
  29 import com.google.gson.Gson;
  30 import com.google.gson.GsonBuilder;
  31 import org.onap.aai.cl.api.Logger;
  32 import org.onap.aai.cl.eelf.LoggerFactory;
  33 import org.onap.aai.event.api.EventConsumer;
  34 import org.onap.aai.event.api.EventPublisher;
  35 import org.onap.aai.event.api.MessageWithOffset;
  36 import org.onap.aai.spike.event.envelope.EventEnvelope;
  37 import org.onap.aai.spike.event.envelope.EventEnvelopeParser;
  38 import org.onap.aai.spike.event.incoming.GizmoGraphEvent;
  39 import org.onap.aai.spike.event.incoming.OffsetManager;
  40 import org.onap.aai.spike.event.outgoing.SpikeEventComparator;
  41 import org.onap.aai.spike.event.outgoing.SpikeEventExclusionStrategy;
  42 import org.onap.aai.spike.event.outgoing.SpikeGraphEvent;
  43 import org.onap.aai.spike.exception.SpikeException;
  44 import org.onap.aai.spike.logging.SpikeMsgs;
  45 import org.onap.aai.spike.util.SpikeConstants;
  46 import org.onap.aai.spike.util.SpikeProperties;
  47
  48 public class SpikeEventProcessor extends TimerTask {
  49
  50     /**
  51      * Client used for consuming events to the event bus.
  52      */
  53     private EventConsumer consumer;
  54     /**
  55      * Client used for publishing events to the event bus.
  56      */
  57     private EventPublisher publisher;
  58     /**
  59      * Internal queue where outgoing events will be buffered until they can be serviced by the event
  60      * publisher worker threads.
  61      */
  62     private BlockingQueue<SpikeGraphEvent> eventQueue;
  63
  64     private Integer eventQueueCapacity = DEFAULT_EVENT_QUEUE_CAPACITY;
  65     private Integer eventOffsetPeriod = DEFAULT_EVENT_OFFSET_COMMIT_PERIOD;
  66
  67     private OffsetManager offsetManager;
  68     private Long lastCommittedOffset = null;
  69     private EventEnvelopeParser eventEnvelopeParser;
  70
  71     /**
  72      * Number of events that can be queued up for publishing before it is dropped
  73      */
  74     private static final Integer DEFAULT_EVENT_QUEUE_CAPACITY = 10000;
  75     private static final Integer DEFAULT_EVENT_OFFSET_COMMIT_PERIOD = 10000;
  76
  77     private static Logger logger = LoggerFactory.getInstance().getLogger(SpikeEventProcessor.class.getName());
  78     private static Logger auditLogger = LoggerFactory.getInstance().getAuditLogger(SpikeEventProcessor.class.getName());
  79     private static final Gson gson =
  80             new GsonBuilder().setExclusionStrategies(new SpikeEventExclusionStrategy()).setPrettyPrinting().create();
  81
  82     public SpikeEventProcessor(EventConsumer consumer, EventPublisher publisher) {
  83         this.consumer = consumer;
  84         this.publisher = publisher;
  85
  86         try {
  87             eventQueueCapacity = Integer.parseInt(SpikeProperties.get(SpikeConstants.SPIKE_EVENT_QUEUE_CAPACITY));
  88             eventOffsetPeriod = Integer.parseInt(SpikeProperties.get(SpikeConstants.SPIKE_EVENT_OFFSET_CHECK_PERIOD));
  89
  90         } catch (Exception ex) {
  91         }
  92
  93         eventQueue = new PriorityBlockingQueue<SpikeGraphEvent>(eventQueueCapacity, new SpikeEventComparator());
  94         new Thread(new SpikeEventPublisher()).start();
  95
  96         // Instantiate the offset manager. This will run a background thread that
  97         // periodically updates the value of the most recent offset value that can
  98         // be safely committed with the event bus.
  99         offsetManager = new OffsetManager(eventQueueCapacity, eventOffsetPeriod);
 100         eventEnvelopeParser = new EventEnvelopeParser();
 101     }
 102
 103     @Override
 104     public void run() {
 105         logger.info(SpikeMsgs.SPIKE_QUERY_EVENT_SYSTEM);
 106
 107         if (consumer == null) {
 108             logger.error(SpikeMsgs.SPIKE_SERVICE_STARTED_FAILURE, SpikeConstants.SPIKE_SERVICE_NAME);
 109         }
 110
 111         Iterable<MessageWithOffset> events = null;
 112         try {
 113             events = consumer.consumeWithOffsets();
 114
 115         } catch (OperationNotSupportedException e) {
 116             // This means we are using DMaaP and can't use offsets
 117             try {
 118                 Iterable<String> tempEvents = consumer.consume();
 119                 ArrayList<MessageWithOffset> messages = new ArrayList<MessageWithOffset>();
 120                 for (String event : tempEvents) {
 121                     messages.add(new MessageWithOffset(0, event));
 122                 }
 123                 events = messages;
 124             } catch (Exception e1) {
 125                 logger.error(SpikeMsgs.SPIKE_EVENT_CONSUME_FAILURE, e1.getMessage());
 126                 return;
 127             }
 128         } catch (Exception e) {
 129             logger.error(SpikeMsgs.SPIKE_EVENT_CONSUME_FAILURE, e.getMessage());
 130             return;
 131         }
 132
 133         if (events == null || !events.iterator().hasNext()) {
 134             logger.info(SpikeMsgs.SPIKE_NO_EVENT_RECEIVED);
 135         }
 136
 137         for (MessageWithOffset event : events) {
 138             try {
 139                 logger.debug(SpikeMsgs.SPIKE_EVENT_RECEIVED, event.getMessage());
 140
 141                 GizmoGraphEvent modelEvent = eventEnvelopeParser.parseEvent(event.getMessage());
 142                 auditLogger.info(SpikeMsgs.SPIKE_EVENT_RECEIVED,
 143                         "of type: " + modelEvent.getObjectType() + " with key: " + modelEvent.getObjectKey()
 144                                 + " , transaction-id: " + modelEvent.getTransactionId());
 145                 logger.info(SpikeMsgs.SPIKE_EVENT_RECEIVED, "of type: " + modelEvent.getObjectType() + " with key: "
 146                         + modelEvent.getObjectKey() + " , transaction-id: " + modelEvent.getTransactionId());
 147
 148                 String modelEventJson = gson.toJson(modelEvent);
 149
 150                 // Log the current event as 'being processed' with the offset manager so that we know that it's
 151                 // associated offset is not yet save to be committed as 'done'.
 152                 offsetManager.cacheEvent(modelEvent.getTransactionId(), event.getOffset());
 153
 154                 while (eventQueue.size() >= eventQueueCapacity) {
 155                     // Wait until there's room in the queue
 156                     logger.error(SpikeMsgs.SPIKE_EVENT_PUBLISH_FAILURE,
 157                             "Event could not be published to the event bus due to: Internal buffer capacity exceeded. Waiting 10 seconds.");
 158                     Thread.sleep(10000);
 159                 }
 160
 161                 eventQueue.offer(modelEvent.toSpikeGraphEvent());
 162
 163                 logger.info(SpikeMsgs.SPIKE_EVENT_PROCESSED, "of type: " + modelEvent.getObjectType() + " with key: "
 164                         + modelEvent.getObjectKey() + " , transaction-id: " + modelEvent.getTransactionId());
 165                 logger.debug(SpikeMsgs.SPIKE_EVENT_PROCESSED, modelEventJson);
 166
 167             } catch (SpikeException | InterruptedException e) {
 168                 logger.error(SpikeMsgs.SPIKE_EVENT_CONSUME_FAILURE,
 169                         e.getMessage() + ".  Incoming event payload:\n" + event.getMessage());
 170             } catch (Exception e) {
 171                 logger.error(SpikeMsgs.SPIKE_EVENT_CONSUME_FAILURE,
 172                         e.getMessage() + ".  Incoming event payload:\n" + event.getMessage());
 173             }
 174         }
 175
 176         try {
 177
 178             // Get the next 'safe' offset to be committed from the offset manager.
 179             // We need to do this here istead of letting the offset manager just take care
 180             // of it for us because the event consumer is not thread safe. If we try to
 181             // commit the offsets from another thread, it gets unhappy...
 182             Long nextOffset = offsetManager.getNextOffsetToCommit();
 183
 184             // Make sure we actually have a real value...
 185             if (nextOffset != null) {
 186
 187                 // There is no point in continually committing the same offset value, so make sure
 188                 // that something has actually changed before we do anything...
 189                 if ((lastCommittedOffset == null) || (!lastCommittedOffset.equals(nextOffset))) {
 190
 191                     if (logger.isDebugEnabled()) {
 192                         logger.debug(
 193                                 "Committing offset: " + nextOffset + " to the event bus for Champ raw event topic.");
 194                     }
 195
 196                     // OK, let's commit the latest value...
 197                     consumer.commitOffsets(nextOffset);
 198                     lastCommittedOffset = nextOffset;
 199                 }
 200             }
 201         } catch (OperationNotSupportedException e) {
 202             // We must be working with a DMaap which doesn't support offset management. Swallow
 203             // the exception
 204         } catch (Exception e) {
 205             logger.error(SpikeMsgs.SPIKE_EVENT_CONSUME_FAILURE, e.getMessage());
 206         }
 207     }
 208
 209     /**
 210      * This class implements the threads which is responsible for buffering the events in memory and
 211      * ordering them before publishing it to topic
 212      * <p>
 213      * Each publish operation is performed synchronously, so that the thread will only move on to the
 214      * next available event once it has actually published the current event to the bus.
 215      */
 216     private class SpikeEventPublisher implements Runnable {
 217
 218         /**
 219          * Partition key to use when publishing events to the event stream. We WANT all events to go to a
 220          * single partition, so we are just using a hard-coded key for every event.
 221          */
 222         private static final String EVENTS_PARTITION_KEY = "SpikeEventKey";
 223         private static final int DEFAULT_EVENT_QUEUE_DELAY = 10000;
 224
 225         Integer eventQueueDelay = DEFAULT_EVENT_QUEUE_DELAY;
 226
 227         public SpikeEventPublisher() {
 228             try {
 229                 eventQueueDelay = Integer.parseInt(SpikeProperties.get(SpikeConstants.SPIKE_EVENT_QUEUE_DELAY));
 230             } catch (Exception ex) {
 231             }
 232         }
 233
 234         @Override
 235         public void run() {
 236             while (true) {
 237
 238                 SpikeGraphEvent nextEvent;
 239                 SpikeGraphEvent event = null;
 240                 try {
 241
 242                     // Get the next event to be published from the queue if it is old enough or we have too
 243                     // many items in the queue
 244                     if ((nextEvent = eventQueue.peek()) != null
 245                             && (System.currentTimeMillis() - nextEvent.getSpikeTimestamp() > eventQueueDelay
 246                                     || eventQueue.size() > eventQueueCapacity)) {
 247                         event = eventQueue.take();
 248                     } else {
 249                         // Small pause so that we aren't burning CPU
 250                         Thread.sleep(200);
 251                         continue;
 252                     }
 253
 254                 } catch (InterruptedException e) {
 255
 256                     // Restore the interrupted status.
 257                     Thread.currentThread().interrupt();
 258                 }
 259
 260                 // Try publishing the event to the event bus. This call will block
 261                 // until the event is published or times out.
 262                 try {
 263                     String eventJson = gson.toJson(new EventEnvelope(event));
 264                     int sentMessageCount = publisher.sendSync(EVENTS_PARTITION_KEY, eventJson);
 265                     if (sentMessageCount > 0) {
 266                         logger.info(SpikeMsgs.SPIKE_EVENT_PUBLISHED, "of type: " + event.getObjectType() + " with key: "
 267                                 + event.getObjectKey() + " , transaction-id: " + event.getTransactionId());
 268                         logger.debug(SpikeMsgs.SPIKE_EVENT_PUBLISHED, eventJson);
 269                     } else {
 270                         logger.warn(SpikeMsgs.SPIKE_PUBLISH_FAILED, "of type: " + event.getObjectType() + " with key: "
 271                                 + event.getObjectKey() + " , transaction-id: " + event.getTransactionId());
 272                         logger.debug(SpikeMsgs.SPIKE_PUBLISH_FAILED, eventJson);
 273                     }
 274
 275
 276                     // Inform the offset manager that this event has been published. It's offset
 277                     // can now, potentially, be safely committed to the event bus so that on a
 278                     // restart we won't reprocess it.
 279                     offsetManager.markAsPublished(event.getTransactionId());
 280
 281                 } catch (ExecutionException e) {
 282
 283                     // Publish timed out, queue it up to retry again. Since this message was pulled from the
 284                     // top of the queue, it will go back to the top.
 285                     logger.error(SpikeMsgs.SPIKE_EVENT_PUBLISH_FAILURE, "Retrying in 60 seconds. " + e.getMessage());
 286                     eventQueue.offer(event);
 287
 288                     try {
 289                         Thread.sleep(60000);
 290                     } catch (InterruptedException e1) {
 291                         e1.printStackTrace();
 292                     }
 293                 } catch (Exception e) {
 294                     logger.error(SpikeMsgs.SPIKE_EVENT_PUBLISH_FAILURE, e.getMessage());
 295                 }
 296             }
 297         }
 298     }
 299
 300 }