gerrit.onap Code Review - policy/drools-applications.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP
   4  * ================================================================================
   5  * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.controlloop.ophistory;
  22
  23 import java.util.Date;
  24 import java.util.Properties;
  25 import java.util.concurrent.BlockingQueue;
  26 import java.util.concurrent.LinkedBlockingQueue;
  27 import java.util.function.Consumer;
  28 import javax.persistence.EntityManager;
  29 import javax.persistence.EntityManagerFactory;
  30 import javax.persistence.Persistence;
  31 import lombok.AllArgsConstructor;
  32 import lombok.Getter;
  33 import lombok.NoArgsConstructor;
  34 import lombok.ToString;
  35 import org.eclipse.persistence.config.PersistenceUnitProperties;
  36 import org.onap.policy.common.parameters.ValidationResult;
  37 import org.onap.policy.common.utils.jpa.EntityMgrCloser;
  38 import org.onap.policy.common.utils.jpa.EntityTransCloser;
  39 import org.onap.policy.controlloop.ControlLoopOperation;
  40 import org.onap.policy.controlloop.VirtualControlLoopEvent;
  41 import org.onap.policy.database.operationshistory.Dbao;
  42 import org.onap.policy.guard.Util;
  43 import org.slf4j.Logger;
  44 import org.slf4j.LoggerFactory;
  45
  46 /**
  47  * Data manager that stores records in the DB, asynchronously, using a background thread.
  48  */
  49 public class OperationHistoryDataManagerImpl implements OperationHistoryDataManager {
  50     private static final Logger logger = LoggerFactory.getLogger(OperationHistoryDataManagerImpl.class);
  51
  52     /**
  53      * Added to the end of {@link #operations} when {@link #stop()} is called. This is
  54      * used to get the background thread out of a blocking wait for the next record.
  55      */
  56     private static final Record END_MARKER = new Record();
  57
  58     // copied from the parameters
  59     private final int maxQueueLength;
  60     private final int batchSize;
  61
  62     private final EntityManagerFactory emFactory;
  63
  64     /**
  65      * Thread that takes records from {@link #operations} and stores them in the DB.
  66      */
  67     private Thread thread;
  68
  69     /**
  70      * Set to {@code true} to stop the background thread.
  71      */
  72     private boolean stopped = false;
  73
  74     /**
  75      * Queue of operations waiting to be stored in the DB. When {@link #stop()} is called,
  76      * an {@link #END_MARKER} is added to the end of the queue.
  77      */
  78     private final BlockingQueue<Record> operations = new LinkedBlockingQueue<>();
  79
  80     /**
  81      * Number of records that have been added to the DB by this data manager instance.
  82      */
  83     @Getter
  84     private long recordsAdded = 0;
  85
  86
  87     /**
  88      * Constructs the object.
  89      *
  90      * @param params data manager parameters
  91      */
  92     public OperationHistoryDataManagerImpl(OperationHistoryDataManagerParams params) {
  93         ValidationResult result = params.validate("data-manager-properties");
  94         if (!result.isValid()) {
  95             throw new IllegalArgumentException(result.getResult());
  96         }
  97
  98         this.maxQueueLength = params.getMaxQueueLength();
  99         this.batchSize = params.getBatchSize();
 100
 101         // create the factory using the properties
 102         Properties props = toProperties(params);
 103         this.emFactory = makeEntityManagerFactory(params.getPersistenceUnit(), props);
 104     }
 105
 106     @Override
 107     public synchronized void start() {
 108         if (stopped || thread != null) {
 109             // already started
 110             return;
 111         }
 112
 113         thread = makeThread(emFactory, this::run);
 114         thread.setDaemon(true);
 115         thread.start();
 116     }
 117
 118     @Override
 119     public synchronized void stop() {
 120         stopped = true;
 121
 122         if (thread == null) {
 123             // no thread to close the factory - do it here
 124             emFactory.close();
 125
 126         } else {
 127             // the thread will close the factory when it sees the end marker
 128             operations.add(END_MARKER);
 129         }
 130     }
 131
 132     @Override
 133     public synchronized void store(String requestId, VirtualControlLoopEvent event, ControlLoopOperation operation) {
 134
 135         if (stopped) {
 136             logger.warn("operation history thread is stopped, discarding requestId={} event={} operation={}", requestId,
 137                             event, operation);
 138             return;
 139         }
 140
 141         operations.add(new Record(requestId, event, operation));
 142
 143         if (operations.size() > maxQueueLength) {
 144             Record discarded = operations.remove();
 145             logger.warn("too many items to store in the operation history table, discarding {}", discarded);
 146         }
 147     }
 148
 149     /**
 150      * Takes records from {@link #operations} and stores them in the queue. Continues to
 151      * run until {@link #stop()} is invoked, or the thread is interrupted.
 152      *
 153      * @param emfactory entity manager factory
 154      */
 155     private void run(EntityManagerFactory emfactory) {
 156         try {
 157             // store records until stopped, continuing if an exception occurs
 158             while (!stopped) {
 159                 try {
 160                     Record triple = operations.take();
 161                     storeBatch(emfactory.createEntityManager(), triple);
 162
 163                 } catch (RuntimeException e) {
 164                     logger.error("failed to save data to operation history table", e);
 165
 166                 } catch (InterruptedException e) {
 167                     logger.error("interrupted, discarding remaining operation history data", e);
 168                     Thread.currentThread().interrupt();
 169                     return;
 170                 }
 171             }
 172
 173             storeRemainingRecords(emfactory);
 174
 175         } finally {
 176             synchronized (this) {
 177                 stopped = true;
 178             }
 179
 180             emfactory.close();
 181         }
 182     }
 183
 184     /**
 185      * Store any remaining records, but stop at the first exception.
 186      *
 187      * @param emfactory entity manager factory
 188      */
 189     private void storeRemainingRecords(EntityManagerFactory emfactory) {
 190         try {
 191             while (!operations.isEmpty()) {
 192                 storeBatch(emfactory.createEntityManager(), operations.poll());
 193             }
 194
 195         } catch (RuntimeException e) {
 196             logger.error("failed to save remaining data to operation history table", e);
 197         }
 198     }
 199
 200     /**
 201      * Stores a batch of records.
 202      *
 203      * @param entityManager entity manager
 204      * @param firstRecord first record to be stored
 205      */
 206     private void storeBatch(EntityManager entityManager, Record firstRecord) {
 207
 208         try (EntityMgrCloser emc = new EntityMgrCloser(entityManager);
 209                         EntityTransCloser trans = new EntityTransCloser(entityManager.getTransaction())) {
 210
 211             int nrecords = 0;
 212             Record record = firstRecord;
 213
 214             while (record != null && record != END_MARKER) {
 215                 storeRecord(entityManager, record);
 216
 217                 if (++nrecords >= batchSize) {
 218                     break;
 219                 }
 220
 221                 record = operations.poll();
 222             }
 223
 224             trans.commit();
 225             recordsAdded += nrecords;
 226         }
 227     }
 228
 229     /**
 230      * Stores a record.
 231      *
 232      * @param entityManager entity manager
 233      * @param record record to be stored
 234      */
 235     private void storeRecord(EntityManager entityMgr, Record record) {
 236
 237         Dbao newEntry = new Dbao();
 238
 239         final VirtualControlLoopEvent event = record.getEvent();
 240         final ControlLoopOperation operation = record.getOperation();
 241
 242         newEntry.setClosedLoopName(event.getClosedLoopControlName());
 243         newEntry.setRequestId(record.getRequestId());
 244         newEntry.setActor(operation.getActor());
 245         newEntry.setOperation(operation.getOperation());
 246         newEntry.setTarget(operation.getTarget());
 247         newEntry.setSubrequestId(operation.getSubRequestId());
 248         newEntry.setMessage(operation.getMessage());
 249         newEntry.setOutcome(operation.getOutcome());
 250         if (operation.getStart() != null) {
 251             newEntry.setStarttime(new Date(operation.getStart().toEpochMilli()));
 252         }
 253         if (operation.getEnd() != null) {
 254             newEntry.setEndtime(new Date(operation.getEnd().toEpochMilli()));
 255         }
 256
 257         entityMgr.persist(newEntry);
 258     }
 259
 260     /**
 261      * Converts the parameters to Properties.
 262      *
 263      * @param params parameters to be converted
 264      * @return a new property set
 265      */
 266     private Properties toProperties(OperationHistoryDataManagerParams params) {
 267         Properties props = new Properties();
 268         props.put(Util.ECLIPSE_LINK_KEY_URL, params.getUrl());
 269         props.put(Util.ECLIPSE_LINK_KEY_USER, params.getUserName());
 270         props.put(Util.ECLIPSE_LINK_KEY_PASS, params.getPassword());
 271         props.put(PersistenceUnitProperties.CLASSLOADER, getClass().getClassLoader());
 272
 273         return props;
 274     }
 275
 276     @Getter
 277     @NoArgsConstructor
 278     @AllArgsConstructor
 279     @ToString
 280     private static class Record {
 281         private String requestId;
 282         private VirtualControlLoopEvent event;
 283         private ControlLoopOperation operation;
 284     }
 285
 286     // the following may be overridden by junit tests
 287
 288     protected EntityManagerFactory makeEntityManagerFactory(String opsHistPu, Properties props) {
 289         return Persistence.createEntityManagerFactory(opsHistPu, props);
 290     }
 291
 292     protected Thread makeThread(EntityManagerFactory emfactory, Consumer<EntityManagerFactory> command) {
 293         return new Thread(() -> command.accept(emfactory));
 294     }
 295 }