Modified code to insert a pending record when an operation starts.
Also modified it to update the existing record when the operation
completes, rather than adding a new record.
Note: the "outcome" for a "pending" record is left unset (i.e., it
is null).
Issue-ID: POLICY-2581
Change-Id: Ia1a02ed5a16b8af1328a49b22478fd57c4b9aca0
Signed-off-by: Jim Hahn <jrh3@att.com>
// operation started
++attempts;
state = State.OPERATION_STARTED;
- operationHistory.add(new Operation(outcome));
- break;
- }
- /*
- * Operation completed. If the last entry was a "start" (i.e., "end" field
- * is null), then replace it. Otherwise, just add the completion.
- */
- state = (outcome.getResult() == PolicyResult.SUCCESS ? State.OPERATION_SUCCESS
- : State.OPERATION_FAILURE);
- controlLoopResponse = outcome.getControlLoopResponse();
- if (!operationHistory.isEmpty() && operationHistory.peekLast().getClOperation().getEnd() == null) {
- operationHistory.removeLast();
+ } else {
+ /*
+ * Operation completed. If the last entry was a "start" (i.e., "end" field
+ * is null), then replace it. Otherwise, just add the completion.
+ */
+ state = (outcome.getResult() == PolicyResult.SUCCESS ? State.OPERATION_SUCCESS
+ : State.OPERATION_FAILURE);
+ controlLoopResponse = outcome.getControlLoopResponse();
+ if (!operationHistory.isEmpty() && operationHistory.peekLast().getClOperation().getEnd() == null) {
+ operationHistory.removeLast();
+ }
}
+
operationHistory.add(new Operation(outcome));
storeOperationInDataBase();
break;
package org.onap.policy.controlloop.ophistory;
import java.util.Date;
+import java.util.List;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
private final BlockingQueue<Record> operations = new LinkedBlockingQueue<>();
/**
- * Number of records that have been added to the DB by this data manager instance.
+ * Number of records that have been processed and committed into the DB by this data
+ * manager instance.
*/
@Getter
- private long recordsAdded = 0;
+ private long recordsCommitted = 0;
+
+ /**
+ * Number of records that have been inserted into the DB by this data manager
+ * instance, whether or not they were committed.
+ */
+ @Getter
+ private long recordsInserted = 0;
+
+ /**
+ * Number of records that have been updated within the DB by this data manager
+ * instance, whether or not they were committed.
+ */
+ @Getter
+ private long recordsUpdated = 0;
/**
}
trans.commit();
- recordsAdded += nrecords;
+ recordsCommitted += nrecords;
}
}
* @param record record to be stored
*/
private void storeRecord(EntityManager entityMgr, Record record) {
- Dbao newEntry = new Dbao();
final VirtualControlLoopEvent event = record.getEvent();
final ControlLoopOperation operation = record.getOperation();
logger.info("store operation history record for {}", event.getRequestId());
- newEntry.setClosedLoopName(event.getClosedLoopControlName());
- newEntry.setRequestId(record.getRequestId());
- newEntry.setActor(operation.getActor());
- newEntry.setOperation(operation.getOperation());
- newEntry.setTarget(record.getTargetEntity());
- newEntry.setSubrequestId(operation.getSubRequestId());
- newEntry.setMessage(operation.getMessage());
- newEntry.setOutcome(operation.getOutcome());
+ List<Dbao> results =
+ entityMgr.createQuery("select e from Dbao e"
+ + " where e.closedLoopName= ?1"
+ + " and e.requestId= ?2"
+ + " and e.subrequestId= ?3"
+ + " and e.actor= ?4"
+ + " and e.operation= ?5"
+ + " and e.target= ?6",
+ Dbao.class)
+ .setParameter(1, event.getClosedLoopControlName())
+ .setParameter(2, record.getRequestId())
+ .setParameter(3, operation.getSubRequestId())
+ .setParameter(4, operation.getActor())
+ .setParameter(5, operation.getOperation())
+ .setParameter(6, record.getTargetEntity())
+ .getResultList();
+
+ Dbao entry = (results.isEmpty() ? new Dbao() : results.get(0));
+
+ entry.setClosedLoopName(event.getClosedLoopControlName());
+ entry.setRequestId(record.getRequestId());
+ entry.setActor(operation.getActor());
+ entry.setOperation(operation.getOperation());
+ entry.setTarget(record.getTargetEntity());
+ entry.setSubrequestId(operation.getSubRequestId());
+ entry.setMessage(operation.getMessage());
+ entry.setOutcome(operation.getOutcome());
if (operation.getStart() != null) {
- newEntry.setStarttime(new Date(operation.getStart().toEpochMilli()));
+ entry.setStarttime(new Date(operation.getStart().toEpochMilli()));
+ } else {
+ entry.setStarttime(null);
}
if (operation.getEnd() != null) {
- newEntry.setEndtime(new Date(operation.getEnd().toEpochMilli()));
+ entry.setEndtime(new Date(operation.getEnd().toEpochMilli()));
+ } else {
+ entry.setEndtime(null);
}
- entityMgr.persist(newEntry);
+ if (results.isEmpty()) {
+ logger.info("insert operation history record for {}", event.getRequestId());
+ ++recordsInserted;
+ entityMgr.persist(entry);
+ } else {
+ logger.info("update operation history record for {}", event.getRequestId());
+ ++recordsUpdated;
+ entityMgr.merge(entry);
+ }
}
/**
assertFalse(mgr.nextStep());
verify(mgrctx, times(4)).updated(mgr);
- verifyDb(1, PolicyResult.SUCCESS, null);
+ verifyDb(2, PolicyResult.SUCCESS, null);
}
/**
assertTrue(mgr.nextStep());
assertEquals(ControlLoopOperationManager2.State.OPERATION_FAILURE, mgr.getState());
- verifyDb(1, PolicyResult.FAILURE, null);
+ verifyDb(2, PolicyResult.FAILURE, null);
assertThat(mgr.toString()).contains("attempts=1");
assertTrue(mgr.nextStep());
assertEquals(ControlLoopOperationManager2.State.OPERATION_FAILURE, mgr.getState());
- verifyDb(2, PolicyResult.FAILURE, null);
+ verifyDb(4, PolicyResult.FAILURE, null);
assertThat(mgr.toString()).contains("attempts=2");
assertTrue(mgr.nextStep());
assertEquals(ControlLoopOperationManager2.State.OPERATION_SUCCESS, mgr.getState());
- verifyDb(3, PolicyResult.SUCCESS, null);
+ verifyDb(6, PolicyResult.SUCCESS, null);
assertThat(mgr.toString()).contains("attempts=3");
public class OperationHistoryDataManagerImplTest {
private static final IllegalStateException EXPECTED_EXCEPTION = new IllegalStateException("expected exception");
+ private static final String MY_LOOP_NAME = "my-loop-name";
+ private static final String MY_ACTOR = "my-actor";
+ private static final String MY_OPERATION = "my-operation";
private static final String MY_TARGET = "my-target";
private static final String MY_ENTITY = "my-entity";
private static final String REQ_ID = "my-request-id";
MockitoAnnotations.initMocks(this);
event = new VirtualControlLoopEvent();
+ event.setClosedLoopControlName(MY_LOOP_NAME);
event.setRequestId(UUID.randomUUID());
operation = new ControlLoopOperation();
+ operation.setActor(MY_ACTOR);
+ operation.setOperation(MY_OPERATION);
operation.setTarget(MY_TARGET);
+ operation.setSubRequestId(UUID.randomUUID().toString());
threadFunction = null;
finished = new CountDownLatch(1);
runThread();
- assertEquals(1, mgr.getRecordsAdded());
+ assertEquals(1, mgr.getRecordsCommitted());
}
/**
// store
mgr.store(REQ_ID, event, MY_ENTITY, operation);
- assertEquals(0, mgr.getRecordsAdded());
+ assertEquals(0, mgr.getRecordsCommitted());
}
/**
runThread();
- assertEquals(MAX_QUEUE_LENGTH, mgr.getRecordsAdded());
+ assertEquals(MAX_QUEUE_LENGTH, mgr.getRecordsCommitted());
}
@Test
verify(emfSpy).close();
- assertEquals(3, mgr.getRecordsAdded());
+ assertEquals(3, mgr.getRecordsCommitted());
}
private void waitForThread() {
@Test
public void testStoreRecord() throws InterruptedException {
+ /*
+ * Note: we change sub-request ID each time to guarantee that the records are
+ * unique.
+ */
+
// no start time
mgr.store(REQ_ID, event, MY_ENTITY, operation);
- // no start time
+ // no end time
operation = new ControlLoopOperation(operation);
+ operation.setSubRequestId(UUID.randomUUID().toString());
operation.setStart(Instant.now());
mgr.store(REQ_ID, event, MY_ENTITY, operation);
// both start and end times
operation = new ControlLoopOperation(operation);
+ operation.setSubRequestId(UUID.randomUUID().toString());
operation.setEnd(Instant.now());
mgr.store(REQ_ID, event, MY_ENTITY, operation);
// only end time
operation = new ControlLoopOperation(operation);
+ operation.setSubRequestId(UUID.randomUUID().toString());
+ operation.setStart(null);
+ mgr.store(REQ_ID, event, MY_ENTITY, operation);
+
+ runThread();
+
+ // all of them should have been stored
+ assertEquals(4, mgr.getRecordsCommitted());
+
+ // each was unique
+ assertEquals(4, mgr.getRecordsInserted());
+ assertEquals(0, mgr.getRecordsUpdated());
+ }
+
+ /**
+ * Tests storeRecord() when records are updated.
+ */
+ @Test
+ public void testStoreRecordUpdate() throws InterruptedException {
+ /*
+ * Note: we do NOT change sub-request ID, so that records all refer to the same DB
+ * record.
+ */
+
+ // no start time
+ mgr.store(REQ_ID, event, MY_ENTITY, operation);
+
+ // no end time
+ operation.setStart(Instant.now());
+ mgr.store(REQ_ID, event, MY_ENTITY, operation);
+
+ // both start and end times
+ operation.setEnd(Instant.now());
+ mgr.store(REQ_ID, event, MY_ENTITY, operation);
+
+ // only end time
operation.setStart(null);
mgr.store(REQ_ID, event, MY_ENTITY, operation);
runThread();
// all of them should have been stored
- assertEquals(4, mgr.getRecordsAdded());
+ assertEquals(4, mgr.getRecordsCommitted());
+
+ // only one new record
+ assertEquals(1, mgr.getRecordsInserted());
+
+ // remainder were updates
+ assertEquals(3, mgr.getRecordsUpdated());
}
private void runThread() throws InterruptedException {
Code Review / policy / drools-applications.git / commitdiff