Event linked id introduced

dariol83 · dariol83 · commit f401386dafc9 · 2020-09-21T21:55:09.000+02:00
diff --git a/eu.dariolucia.reatmetric.api/src/main/java/eu/dariolucia/reatmetric/api/messages/OperationalMessage.java b/eu.dariolucia.reatmetric.api/src/main/java/eu/dariolucia/reatmetric/api/messages/OperationalMessage.java
@@ -43,12 +43,15 @@ public final class OperationalMessage extends AbstractDataItem implements Serial
 
     private final Severity severity;
 
-    public OperationalMessage(IUniqueId internalId, Instant generationTime, String id, String message, String source, Severity severity, Object extension) {
+    private final Integer linkedEntityId;
+
+    public OperationalMessage(IUniqueId internalId, Instant generationTime, String id, String message, String source, Severity severity, Integer linkedEntityId, Object extension) {
         super(internalId, generationTime, extension);
         this.id = id;
         this.message = message;
         this.source = source;
         this.severity = severity;
+        this.linkedEntityId = linkedEntityId;
     }
 
     public String getId() {
@@ -67,13 +70,18 @@ public Severity getSeverity() {
         return severity;
     }
 
+    public Integer getLinkedEntityId() {
+        return linkedEntityId;
+    }
+
     @Override
     public String toString() {
         return "OperationalMessage{" +
                 "id='" + id + '\'' +
                 ", message='" + message + '\'' +
                 ", source='" + source + '\'' +
                 ", severity=" + severity +
+                ", linkedEntityId=" + linkedEntityId +
                 ", generationTime=" + generationTime +
                 ", internalId=" + internalId +
                 '}';
diff --git a/eu.dariolucia.reatmetric.core/src/main/java/eu/dariolucia/reatmetric/core/api/IOperationalMessageBroker.java b/eu.dariolucia.reatmetric.core/src/main/java/eu/dariolucia/reatmetric/core/api/IOperationalMessageBroker.java
@@ -22,6 +22,6 @@
 
 public interface IOperationalMessageBroker {
 
-    OperationalMessage distribute(String id, String message, String source, Severity severity, Object[] additionalFields, boolean store) throws ReatmetricException;
+    OperationalMessage distribute(String id, String message, String source, Severity severity, Object[] additionalFields, Integer linkedEntityId, boolean store) throws ReatmetricException;
 
 }
diff --git a/eu.dariolucia.reatmetric.core/src/main/java/eu/dariolucia/reatmetric/core/impl/OperationalMessageBrokerImpl.java b/eu.dariolucia.reatmetric.core/src/main/java/eu/dariolucia/reatmetric/core/impl/OperationalMessageBrokerImpl.java
@@ -26,10 +26,7 @@
 
 import java.rmi.RemoteException;
 import java.time.Instant;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Predicate;
@@ -147,7 +144,22 @@ public void publish(LogRecord record) {
         } else if(record.getLevel().intValue() >= Level.WARNING.intValue()) {
             messageSeverity = Severity.WARN;
         }
-        OperationalMessage om = new OperationalMessage(idToAssign, record.getInstant(), REATMETRIC_ID, record.getMessage(), shortenLoggerName(record.getLoggerName()), messageSeverity, null);
+        // Check for parameters - Source
+        String source;
+        if(record.getParameters() != null && record.getParameters().length > 0 && record.getParameters()[0] != null) {
+            // Source override
+            source = Objects.toString(record.getParameters()[0]);
+        } else {
+            source = shortenLoggerName(record.getLoggerName());
+        }
+        // Check for parameters - Linked Entity ID
+        Integer entityId = null;
+        if(record.getParameters() != null && record.getParameters().length > 1 && record.getParameters()[1] instanceof Integer) {
+            // Entity
+            entityId = (Integer) record.getParameters()[1];
+        }
+
+        OperationalMessage om = new OperationalMessage(idToAssign, record.getInstant(), REATMETRIC_ID, record.getMessage(), source, messageSeverity, entityId,null);
         try {
             distribute(Collections.singletonList(om), true);
         } catch (ReatmetricException e) {
@@ -177,9 +189,9 @@ public void close() throws SecurityException {
     }
 
     @Override
-    public OperationalMessage distribute(String id, String message, String source, Severity severity, Object[] additionalFields, boolean store) throws ReatmetricException {
+    public OperationalMessage distribute(String id, String message, String source, Severity severity, Object[] additionalFields, Integer linkedId, boolean store) throws ReatmetricException {
         IUniqueId idToAssign = nextOperationalMessageId();
-        OperationalMessage om = new OperationalMessage(idToAssign, Instant.now(), id, message, source, severity, additionalFields);
+        OperationalMessage om = new OperationalMessage(idToAssign, Instant.now(), id, message, source, severity, linkedId, additionalFields);
         distribute(Collections.singletonList(om), store);
         return om;
     }
diff --git a/eu.dariolucia.reatmetric.driver.automation/src/main/java/eu/dariolucia/reatmetric/driver/automation/internal/ScriptExecutionManager.java b/eu.dariolucia.reatmetric.driver.automation/src/main/java/eu/dariolucia/reatmetric/driver/automation/internal/ScriptExecutionManager.java
@@ -275,7 +275,7 @@ public boolean initConnector(String connectorName, String[] keys, Object[] value
     private void logMessage(String message, Severity severity) {
         checkAborted();
         try {
-            context.getOperationalMessageBroker().distribute("Script", message, fileName, severity, null, true);
+            context.getOperationalMessageBroker().distribute("Script", message, fileName, severity, null, null, true);
         } catch (ReatmetricException e) {
             LOG.log(Level.SEVERE, "logMessage exception from automation " + fileName + ": " + e.getMessage(), e);
         }
diff --git a/eu.dariolucia.reatmetric.persist/src/main/java/eu/dariolucia/reatmetric/persist/services/AcknowledgedMessageArchive.java b/eu.dariolucia.reatmetric.persist/src/main/java/eu/dariolucia/reatmetric/persist/services/AcknowledgedMessageArchive.java
@@ -42,7 +42,7 @@ public class AcknowledgedMessageArchive extends AbstractDataItemArchive<Acknowle
             "WHEN NOT MATCHED THEN INSERT (UniqueId,GenerationTime,MessageId,State,UserName,AcknowledgementTime,AdditionalData) VALUES (?,?,?,?,?,?,?)";
     private static final String LAST_ID_QUERY = "SELECT UniqueId FROM ACK_MESSAGE_TABLE ORDER BY UniqueId DESC FETCH FIRST ROW ONLY";
     private static final String RETRIEVE_BY_ID_QUERY = "SELECT a.UniqueId, a.GenerationTime, a.State, a.UserName, a.AcknowledgementTime, a.AdditionalData, " +
-            "b.UniqueId, b.GenerationTime, b.Id, b.Text, b.Source, b.Severity, b.AdditionalData " +
+            "b.UniqueId, b.GenerationTime, b.Id, b.Text, b.Source, b.Severity, b.LinkedEntityId, b.AdditionalData " +
             "FROM ACK_MESSAGE_TABLE as a JOIN OPERATIONAL_MESSAGE_TABLE as b " +
             "ON (a.MessageId = b.UniqueId) " +
             "WHERE a.UniqueId=?";
@@ -104,7 +104,7 @@ protected String buildRetrieveByIdQuery() {
     @Override
     protected String buildRetrieveQuery(Instant startTime, int numRecords, RetrievalDirection direction, AcknowledgedMessageFilter filter) {
         StringBuilder query = new StringBuilder("SELECT a.UniqueId, a.GenerationTime, a.State, a.UserName, a.AcknowledgementTime, a.AdditionalData, " +
-                "b.UniqueId, b.GenerationTime, b.Id, b.Text, b.Source, b.Severity, b.AdditionalData " +
+                "b.UniqueId, b.GenerationTime, b.Id, b.Text, b.Source, b.Severity, b.LinkedEntityId, b.AdditionalData " +
                 "FROM ACK_MESSAGE_TABLE as a JOIN OPERATIONAL_MESSAGE_TABLE as b " +
                 "ON (a.MessageId = b.UniqueId) " +
                 "WHERE a.");
@@ -131,7 +131,7 @@ protected String buildRetrieveQuery(Instant startTime, int numRecords, Retrieval
     @Override
     protected String buildRetrieveQuery(Instant startTime, IUniqueId internalId, int numRecords, RetrievalDirection direction, AcknowledgedMessageFilter filter) {
         StringBuilder query = new StringBuilder("SELECT a.UniqueId, a.GenerationTime, a.State, a.UserName, a.AcknowledgementTime, a.AdditionalData, " +
-                "b.UniqueId, b.GenerationTime, b.Id, b.Text, b.Source, b.Severity, b.AdditionalData " +
+                "b.UniqueId, b.GenerationTime, b.Id, b.Text, b.Source, b.Severity, b.LinkedEntityId, b.AdditionalData " +
                 "FROM ACK_MESSAGE_TABLE as a JOIN OPERATIONAL_MESSAGE_TABLE as b " +
                 "ON (a.MessageId = b.UniqueId) " +
                 "WHERE a.");
diff --git a/eu.dariolucia.reatmetric.persist/src/main/java/eu/dariolucia/reatmetric/persist/services/OperationalMessageArchive.java b/eu.dariolucia.reatmetric.persist/src/main/java/eu/dariolucia/reatmetric/persist/services/OperationalMessageArchive.java
@@ -38,9 +38,9 @@ public class OperationalMessageArchive extends AbstractDataItemArchive<Operation
 
     private static final Logger LOG = Logger.getLogger(OperationalMessageArchive.class.getName());
 
-    private static final String STORE_STATEMENT = "INSERT INTO OPERATIONAL_MESSAGE_TABLE(UniqueId,GenerationTime,Id,Text,Source,Severity,AdditionalData) VALUES (?,?,?,?,?,?,?)";
+    private static final String STORE_STATEMENT = "INSERT INTO OPERATIONAL_MESSAGE_TABLE(UniqueId,GenerationTime,Id,Text,Source,Severity,LinkedEntityId,AdditionalData) VALUES (?,?,?,?,?,?,?,?)";
     private static final String LAST_ID_QUERY = "SELECT UniqueId FROM OPERATIONAL_MESSAGE_TABLE ORDER BY UniqueId DESC FETCH FIRST ROW ONLY";
-    private static final String RETRIEVE_BY_ID_QUERY = "SELECT UniqueId,GenerationTime,Id,Text,Source,Severity,AdditionalData FROM OPERATIONAL_MESSAGE_TABLE WHERE UniqueId=?";
+    private static final String RETRIEVE_BY_ID_QUERY = "SELECT UniqueId,GenerationTime,Id,Text,Source,Severity,LinkedEntityId,AdditionalData FROM OPERATIONAL_MESSAGE_TABLE WHERE UniqueId=?";
     private static final String LAST_GENERATION_TIME_QUERY = "SELECT MAX(GenerationTime) FROM OPERATIONAL_MESSAGE_TABLE";
 
     public OperationalMessageArchive(Archive controller) throws SQLException {
@@ -55,11 +55,16 @@ protected void setItemPropertiesToStatement(PreparedStatement storeStatement, Op
         storeStatement.setString(4, item.getMessage().length() > 255 ? item.getMessage().substring(0,255) : item.getMessage());
         storeStatement.setString(5, item.getSource());
         storeStatement.setShort(6, (short) item.getSeverity().ordinal());
+        if(item.getLinkedEntityId() == null) {
+            storeStatement.setNull(7, Types.INTEGER);
+        } else {
+            storeStatement.setInt(7, item.getLinkedEntityId());
+        }
         Object extension = item.getExtension();
         if(extension == null) {
-            storeStatement.setNull(7, Types.BLOB);
+            storeStatement.setNull(8, Types.BLOB);
         } else {
-            storeStatement.setBlob(7, toInputstream(item.getExtension()));
+            storeStatement.setBlob(8, toInputstream(item.getExtension()));
         }
     }
 
@@ -146,12 +151,16 @@ static OperationalMessage mapToItem(ResultSet rs, OperationalMessageFilter usedF
         String messageText = rs.getString(offset + 4);
         String messageSource = rs.getString(offset + 5);
         Severity severity = Severity.values()[rs.getShort(offset + 6)];
-        Blob extensionBlob = rs.getBlob(offset + 7);
+        Integer linkedEntityId = rs.getInt(offset + 7);
+        if(rs.wasNull()) {
+            linkedEntityId = null;
+        }
+        Blob extensionBlob = rs.getBlob(offset + 8);
         Object extension = null;
         if(extensionBlob != null && !rs.wasNull()) {
             extension = toObject(extensionBlob);
         }
-        return new OperationalMessage(new LongUniqueId(uniqueId), toInstant(genTime), messageId, messageText, messageSource, severity, extension);
+        return new OperationalMessage(new LongUniqueId(uniqueId), toInstant(genTime), messageId, messageText, messageSource, severity, linkedEntityId, extension);
     }
 
     @Override
diff --git a/eu.dariolucia.reatmetric.persist/src/main/resources/schema.ddl b/eu.dariolucia.reatmetric.persist/src/main/resources/schema.ddl
@@ -5,6 +5,7 @@ CREATE TABLE OPERATIONAL_MESSAGE_TABLE (
    Text VARCHAR(255) NOT NULL,
    Source VARCHAR(32),
    Severity SMALLINT NOT NULL,
+   LinkedEntityId INT,
    AdditionalData BLOB,
    PRIMARY KEY (UniqueId)
 )
diff --git a/eu.dariolucia.reatmetric.persist/src/test/java/eu/dariolucia/reatmetric/persist/AcknowledgedMessageArchiveTest.java b/eu.dariolucia.reatmetric.persist/src/test/java/eu/dariolucia/reatmetric/persist/AcknowledgedMessageArchiveTest.java
@@ -61,7 +61,7 @@ void testOperationalMessageStoreRetrieve() throws IOException, ArchiveException,
             IOperationalMessageArchive messageArchive = archive.getArchive(IOperationalMessageArchive.class);
             Instant t = Instant.now();
             // store one item
-            OperationalMessage toStore = new OperationalMessage(new LongUniqueId(0), t, "msgId1", "Text message", "Source1", Severity.ALARM, new Object[0]);
+            OperationalMessage toStore = new OperationalMessage(new LongUniqueId(0), t, "msgId1", "Text message", "Source1", Severity.ALARM, null, new Object[0]);
             messageArchive.store(toStore);
 
             IAcknowledgedMessageArchive ackArchive = archive.getArchive(IAcknowledgedMessageArchive.class);
diff --git a/eu.dariolucia.reatmetric.persist/src/test/java/eu/dariolucia/reatmetric/persist/OperationalMessageArchiveTest.java b/eu.dariolucia.reatmetric.persist/src/test/java/eu/dariolucia/reatmetric/persist/OperationalMessageArchiveTest.java
@@ -63,7 +63,7 @@ void testOperationalMessageStoreRetrieve() throws IOException, ArchiveException,
             IOperationalMessageArchive messageArchive = archive.getArchive(IOperationalMessageArchive.class);
             Instant t = Instant.now();
             // store one item
-            messageArchive.store(new OperationalMessage(new LongUniqueId(0), t, "msgId1", "Text message", "Source1", Severity.ALARM, new Object[0]));
+            messageArchive.store(new OperationalMessage(new LongUniqueId(0), t, "msgId1", "Text message", "Source1", Severity.ALARM, null, new Object[0]));
             Thread.sleep(2000);
             // retrieve: expected 1 item
             List<OperationalMessage> messages = messageArchive.retrieve(t.minusMillis(200), 10, RetrievalDirection.TO_FUTURE, null);
@@ -72,7 +72,7 @@ void testOperationalMessageStoreRetrieve() throws IOException, ArchiveException,
             assertEquals("Text message", messages.get(0).getMessage());
             assertEquals(Severity.ALARM, messages.get(0).getSeverity());
             // store one item
-            messageArchive.store(new OperationalMessage(new LongUniqueId(1), t, "msgId2", "Text message", "Source2", Severity.INFO, new Object[] { "test", 13, Instant.ofEpochMilli(1000)}));
+            messageArchive.store(new OperationalMessage(new LongUniqueId(1), t, "msgId2", "Text message", "Source2", Severity.INFO, 123, new Object[] { "test", 13, Instant.ofEpochMilli(1000)}));
             Thread.sleep(2000);
             // retrieve: expected 2 items
             messages = messageArchive.retrieve(t.minusMillis(200), 10, RetrievalDirection.TO_FUTURE, null);
@@ -97,7 +97,7 @@ void testOperationalMessageStoreRetrieve() throws IOException, ArchiveException,
             // last ID: expected 1
             assertEquals(1, messageArchive.retrieveLastId().asLong());
             // store one item
-            messageArchive.store(new OperationalMessage(new LongUniqueId(2), t.plusSeconds(1), "msgId3", "Text message", "Source2", Severity.ALARM, new Object[] { "test", 13, Instant.ofEpochMilli(1000)}));
+            messageArchive.store(new OperationalMessage(new LongUniqueId(2), t.plusSeconds(1), "msgId3", "Text message", "Source2", Severity.ALARM, null, new Object[] { "test", 13, Instant.ofEpochMilli(1000)}));
             Thread.sleep(2000);
             // retrieve: expected 3 items
             messages = messageArchive.retrieve(Instant.now(), 10, RetrievalDirection.TO_PAST, null);
@@ -107,9 +107,9 @@ void testOperationalMessageStoreRetrieve() throws IOException, ArchiveException,
             // store three items
             t = Instant.now().minusMillis(500);
             List<OperationalMessage> toStore = Arrays.asList(
-                    new OperationalMessage(new LongUniqueId(3), t, "msgId3", "Text message", "Source2", Severity.ALARM, new Object[] { }),
-                    new OperationalMessage(new LongUniqueId(4), t, "msgId4", "Text message special", "Source2", Severity.WARN, new Object[] { }),
-                    new OperationalMessage(new LongUniqueId(5), t, "msgId1", "Text message", "Source2", Severity.INFO, new Object[] { })
+                    new OperationalMessage(new LongUniqueId(3), t, "msgId3", "Text message", "Source2", Severity.ALARM, null, new Object[] { }),
+                    new OperationalMessage(new LongUniqueId(4), t, "msgId4", "Text message special", "Source2", Severity.WARN, null, new Object[] { }),
+                    new OperationalMessage(new LongUniqueId(5), t, "msgId1", "Text message", "Source2", Severity.INFO, null, new Object[] { })
             );
             messageArchive.store(toStore);
             Thread.sleep(2000);
diff --git a/eu.dariolucia.reatmetric.processing/src/main/java/eu/dariolucia/reatmetric/processing/impl/processors/EventProcessor.java b/eu.dariolucia.reatmetric.processing/src/main/java/eu/dariolucia/reatmetric/processing/impl/processors/EventProcessor.java
@@ -53,7 +53,7 @@ public class EventProcessor extends AbstractSystemEntityProcessor<EventProcessin
     private boolean conditionTriggerState = false;
 
     private boolean internallyTriggered = false;
-    private List<String> internalSource = new LinkedList<>(); // nulls allowed
+    private final List<String> internalSource = new LinkedList<>(); // nulls allowed
 
     private Instant lastReportedEventTime = null;
 
@@ -170,16 +170,20 @@ public List<AbstractDataItem> process(EventOccurrence newValue) {
                     generatedStates.add(this.state);
                     // Log the event if log is not suppressed
                     if(definition.isLogEnabled()) {
+                        String logSource = this.state.getSource();
+                        if(logSource == null) {
+                            logSource = getPath().asString();
+                        }
                         switch (definition.getSeverity()) {
                             case ALARM:
                             case ERROR:
-                                LOG.log(Level.SEVERE, "Event " + getPath() + " raised: " + getDefinition().getDescription() + " - Source: " + this.state.getSource());
+                                LOG.log(Level.SEVERE, "Event " + getPath() + " raised: " + getDefinition().getDescription(), new Object[] {logSource, getSystemEntityId()});
                                 break;
                             case WARN:
-                                LOG.log(Level.WARNING, "Event " + getPath() + " raised: " + getDefinition().getDescription() + " - Source: " + this.state.getSource());
+                                LOG.log(Level.WARNING, "Event " + getPath() + " raised: " + getDefinition().getDescription(), new Object[] {logSource, getSystemEntityId()});
                                 break;
                             case INFO:
-                                LOG.log(Level.INFO, "Event " + getPath() + " raised: " + getDefinition().getDescription() + " - Source: " + this.state.getSource());
+                                LOG.log(Level.INFO, "Event " + getPath() + " raised: " + getDefinition().getDescription(), new Object[] {logSource, getSystemEntityId()});
                                 break;
                         }
                     }
diff --git a/eu.dariolucia.reatmetric.processing/src/main/java/eu/dariolucia/reatmetric/processing/impl/processors/ParameterProcessor.java b/eu.dariolucia.reatmetric.processing/src/main/java/eu/dariolucia/reatmetric/processing/impl/processors/ParameterProcessor.java
@@ -291,6 +291,8 @@ public synchronized List<AbstractDataItem> process(ParameterSample newValue) thr
                 this.alarmBuilder.setCurrentValue(this.state.getAlarmState(), this.state.getEngValue(), this.state.getGenerationTime(), this.state.getReceptionTime());
                 if(this.alarmBuilder.isChangedSinceLastBuild()) {
                     alarmData = this.alarmBuilder.build(new LongUniqueId(processor.getNextId(AlarmParameterData.class)));
+                    // Generate alarm message
+                    generateAlarmMessage(alarmData);
                     if(LOG.isLoggable(Level.FINER)) {
                         LOG.log(Level.FINER, "Alarm Parameter Data generated: " + alarmData);
                     }
@@ -318,6 +320,23 @@ public synchronized List<AbstractDataItem> process(ParameterSample newValue) thr
         return generatedStates;
     }
 
+    private void generateAlarmMessage(AlarmParameterData alarmData) {
+        AlarmState state = alarmData.getCurrentAlarmState();
+        switch (state) {
+            case ALARM:
+            case ERROR:
+            {
+                LOG.log(Level.SEVERE, "Parameter " + getPath() + " in alarm, value " + alarmData.getCurrentValue(), new Object[] {getPath().asString(), getSystemEntityId()});
+            }
+            break;
+            case WARNING:
+            {
+                LOG.log(Level.WARNING, "Parameter " + getPath() + " in alarm, value " + alarmData.getCurrentValue(), new Object[] {getPath().asString(), getSystemEntityId()});
+            }
+            break;
+        }
+    }
+
     private void computeSystemEntityState(boolean stateChanged, List<AbstractDataItem> generatedStates) {
         if(stateChanged) {
             this.systemEntityBuilder.setAlarmState(this.state.getAlarmState());

Original file line number	Diff line number	Diff line change
`@@ -22,6 +22,6 @@`
`22`	`22`
`23`	`23`	`public interface IOperationalMessageBroker {`
`24`	`24`
`25`		`- OperationalMessage distribute(String id, String message, String source, Severity severity, Object[] additionalFields, boolean store) throws ReatmetricException;`
	`25`	`+ OperationalMessage distribute(String id, String message, String source, Severity severity, Object[] additionalFields, Integer linkedEntityId, boolean store) throws ReatmetricException;`
`26`	`26`
`27`	`27`	`}`
Original file line number	Diff line number	Diff line change
`@@ -275,7 +275,7 @@ public boolean initConnector(String connectorName, String[] keys, Object[] value`
`275`	`275`	`private void logMessage(String message, Severity severity) {`
`276`	`276`	`checkAborted();`
`277`	`277`	`try {`
`278`		`- context.getOperationalMessageBroker().distribute("Script", message, fileName, severity, null, true);`
	`278`	`+ context.getOperationalMessageBroker().distribute("Script", message, fileName, severity, null, null, true);`
`279`	`279`	`} catch (ReatmetricException e) {`
`280`	`280`	`LOG.log(Level.SEVERE, "logMessage exception from automation " + fileName + ": " + e.getMessage(), e);`
`281`	`281`	`}`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,7 @@ CREATE TABLE OPERATIONAL_MESSAGE_TABLE (`
`5`	`5`	`Text VARCHAR(255) NOT NULL,`
`6`	`6`	`Source VARCHAR(32),`
`7`	`7`	`Severity SMALLINT NOT NULL,`
	`8`	`+ LinkedEntityId INT,`
`8`	`9`	`AdditionalData BLOB,`
`9`	`10`	`PRIMARY KEY (UniqueId)`
`10`	`11`	`)`