[Fix] Fixed pruning of messages in OperationsLog
This commit is contained in:
parent
cee1ac4f82
commit
2286752c57
|
@ -5,6 +5,7 @@ import java.util.concurrent.ExecutorService;
|
||||||
|
|
||||||
import li.strolch.agent.api.ComponentContainer;
|
import li.strolch.agent.api.ComponentContainer;
|
||||||
import li.strolch.agent.api.StrolchComponent;
|
import li.strolch.agent.api.StrolchComponent;
|
||||||
|
import li.strolch.agent.api.StrolchRealm;
|
||||||
import li.strolch.model.Locator;
|
import li.strolch.model.Locator;
|
||||||
import li.strolch.persistence.api.LogMessageDao;
|
import li.strolch.persistence.api.LogMessageDao;
|
||||||
import li.strolch.persistence.api.StrolchTransaction;
|
import li.strolch.persistence.api.StrolchTransaction;
|
||||||
|
@ -29,7 +30,7 @@ public class OperationsLog extends StrolchComponent {
|
||||||
this.logMessagesByRealmAndId = new HashMap<>();
|
this.logMessagesByRealmAndId = new HashMap<>();
|
||||||
this.logMessagesByLocator = new HashMap<>();
|
this.logMessagesByLocator = new HashMap<>();
|
||||||
|
|
||||||
this.executorService = getExecutorService("OperationsLog");
|
this.executorService = getSingleThreadExecutor("OperationsLog");
|
||||||
|
|
||||||
super.initialize(configuration);
|
super.initialize(configuration);
|
||||||
}
|
}
|
||||||
|
@ -65,28 +66,37 @@ public class OperationsLog extends StrolchComponent {
|
||||||
public synchronized void addMessage(LogMessage logMessage) {
|
public synchronized void addMessage(LogMessage logMessage) {
|
||||||
|
|
||||||
// store in global list
|
// store in global list
|
||||||
List<LogMessage> logMessages = this.logMessagesByRealmAndId
|
String realmName = logMessage.getRealm();
|
||||||
.computeIfAbsent(logMessage.getRealm(), r -> new ArrayList<>());
|
List<LogMessage> logMessages = this.logMessagesByRealmAndId.computeIfAbsent(realmName, r -> new ArrayList<>());
|
||||||
logMessages.add(logMessage);
|
logMessages.add(logMessage);
|
||||||
|
|
||||||
// store under locator
|
// store under locator
|
||||||
LinkedHashMap<Locator, LinkedHashSet<LogMessage>> logMessagesLocator = this.logMessagesByLocator
|
LinkedHashMap<Locator, LinkedHashSet<LogMessage>> logMessagesLocator = this.logMessagesByLocator
|
||||||
.computeIfAbsent(logMessage.getRealm(), this::newBoundedLocatorMap);
|
.computeIfAbsent(realmName, this::newBoundedLocatorMap);
|
||||||
LinkedHashSet<LogMessage> messages = logMessagesLocator
|
LinkedHashSet<LogMessage> messages = logMessagesLocator
|
||||||
.computeIfAbsent(logMessage.getLocator(), (l) -> new LinkedHashSet<>());
|
.computeIfAbsent(logMessage.getLocator(), (l) -> new LinkedHashSet<>());
|
||||||
messages.add(logMessage);
|
messages.add(logMessage);
|
||||||
|
|
||||||
this.executorService.submit(() -> this.persistAndPrune(logMessages, logMessage));
|
// prune if necessary
|
||||||
|
List<LogMessage> messagesToRemove = pruneMessages(logMessages);
|
||||||
|
|
||||||
|
// persist changes for non-transient realms
|
||||||
|
StrolchRealm realm = getContainer().getRealm(realmName);
|
||||||
|
if (!realm.getMode().isTransient())
|
||||||
|
this.executorService.submit(() -> this.persist(realm, logMessage, messagesToRemove));
|
||||||
}
|
}
|
||||||
|
|
||||||
private void persistAndPrune(List<LogMessage> logMessages, LogMessage logMessage) {
|
private List<LogMessage> pruneMessages(List<LogMessage> logMessages) {
|
||||||
runAsAgent(ctx -> {
|
if (logMessages.size() < this.maxMessages)
|
||||||
|
return Collections.emptyList();
|
||||||
|
|
||||||
|
List<LogMessage> messagesToRemove = new ArrayList<>();
|
||||||
|
|
||||||
List<LogMessage> messagesToRemove = null;
|
|
||||||
synchronized (this) {
|
|
||||||
if (logMessages.size() > this.maxMessages) {
|
|
||||||
messagesToRemove = new ArrayList<>();
|
|
||||||
int maxDelete = Math.max(1, (int) (this.maxMessages * 0.1));
|
int maxDelete = Math.max(1, (int) (this.maxMessages * 0.1));
|
||||||
|
int nrOfExcessMsgs = logMessages.size() - this.maxMessages;
|
||||||
|
if (nrOfExcessMsgs > 0)
|
||||||
|
maxDelete += nrOfExcessMsgs;
|
||||||
|
|
||||||
logger.info("Pruning " + maxDelete + " messages from OperationsLog...");
|
logger.info("Pruning " + maxDelete + " messages from OperationsLog...");
|
||||||
Iterator<LogMessage> iterator = logMessages.iterator();
|
Iterator<LogMessage> iterator = logMessages.iterator();
|
||||||
while (maxDelete > 0 && iterator.hasNext()) {
|
while (maxDelete > 0 && iterator.hasNext()) {
|
||||||
|
@ -95,19 +105,19 @@ public class OperationsLog extends StrolchComponent {
|
||||||
iterator.remove();
|
iterator.remove();
|
||||||
maxDelete--;
|
maxDelete--;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
return messagesToRemove;
|
||||||
}
|
}
|
||||||
|
|
||||||
// only for persisted realms
|
private void persist(StrolchRealm realm, LogMessage logMessage, List<LogMessage> messagesToRemove) {
|
||||||
if (!getContainer().getRealm(logMessage.getRealm()).getMode().isTransient()) {
|
runAsAgent(ctx -> {
|
||||||
try (StrolchTransaction tx = openTx(logMessage.getRealm(), ctx.getCertificate())) {
|
try (StrolchTransaction tx = realm.openTx(ctx.getCertificate(), getClass())) {
|
||||||
LogMessageDao logMessageDao = tx.getPersistenceHandler().getLogMessageDao(tx);
|
LogMessageDao logMessageDao = tx.getPersistenceHandler().getLogMessageDao(tx);
|
||||||
if (messagesToRemove != null && !messagesToRemove.isEmpty())
|
if (messagesToRemove != null && !messagesToRemove.isEmpty())
|
||||||
logMessageDao.removeAll(messagesToRemove);
|
logMessageDao.removeAll(messagesToRemove);
|
||||||
logMessageDao.save(logMessage);
|
logMessageDao.save(logMessage);
|
||||||
tx.commitOnClose();
|
tx.commitOnClose();
|
||||||
}
|
}
|
||||||
}
|
|
||||||
}
|
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -96,8 +96,8 @@ CREATE TABLE IF NOT EXISTS operations_log (
|
||||||
);
|
);
|
||||||
|
|
||||||
CREATE TABLE IF NOT EXISTS operations_log_values (
|
CREATE TABLE IF NOT EXISTS operations_log_values (
|
||||||
id varchar(255) PRIMARY KEY,
|
id varchar(255),
|
||||||
key VARCHAR(255),
|
key varchar(255),
|
||||||
value text
|
value text
|
||||||
);
|
);
|
||||||
|
|
||||||
|
|
|
@ -14,8 +14,8 @@ CREATE TABLE IF NOT EXISTS operations_log (
|
||||||
);
|
);
|
||||||
|
|
||||||
CREATE TABLE IF NOT EXISTS operations_log_values (
|
CREATE TABLE IF NOT EXISTS operations_log_values (
|
||||||
id varchar(255) PRIMARY KEY,
|
id varchar(255),
|
||||||
key VARCHAR(255),
|
key varchar(255),
|
||||||
value text
|
value text
|
||||||
);
|
);
|
||||||
|
|
||||||
|
|
|
@ -5,7 +5,6 @@ import static li.strolch.model.Tags.AGENT;
|
||||||
import static li.strolch.runtime.StrolchConstants.SYSTEM_USER_AGENT;
|
import static li.strolch.runtime.StrolchConstants.SYSTEM_USER_AGENT;
|
||||||
import static org.junit.Assert.assertEquals;
|
import static org.junit.Assert.assertEquals;
|
||||||
|
|
||||||
import java.util.ArrayList;
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.ResourceBundle;
|
import java.util.ResourceBundle;
|
||||||
|
|
||||||
|
@ -56,12 +55,12 @@ public class LogMessagesTestRunner {
|
||||||
|
|
||||||
if (realm.getMode().isTransient()) {
|
if (realm.getMode().isTransient()) {
|
||||||
List<LogMessage> messages = this.operationsLog.getMessages(realmName);
|
List<LogMessage> messages = this.operationsLog.getMessages(realmName);
|
||||||
assertEquals(1, messages.size());
|
assertEquals(2, messages.size());
|
||||||
} else {
|
} else {
|
||||||
try (StrolchTransaction tx = realm.openTx(this.certificate, "test");) {
|
try (StrolchTransaction tx = realm.openTx(this.certificate, "test");) {
|
||||||
LogMessageDao logMessageDao = tx.getPersistenceHandler().getLogMessageDao(tx);
|
LogMessageDao logMessageDao = tx.getPersistenceHandler().getLogMessageDao(tx);
|
||||||
List<LogMessage> logMessages = logMessageDao.queryLatest(this.realmName, Integer.MAX_VALUE);
|
List<LogMessage> logMessages = logMessageDao.queryLatest(this.realmName, Integer.MAX_VALUE);
|
||||||
assertEquals(1, logMessages.size());
|
assertEquals(2, logMessages.size());
|
||||||
|
|
||||||
LogMessage m = logMessages.get(0);
|
LogMessage m = logMessages.get(0);
|
||||||
assertEquals(logMessage.getId(), m.getId());
|
assertEquals(logMessage.getId(), m.getId());
|
||||||
|
@ -77,8 +76,10 @@ public class LogMessagesTestRunner {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
List<String> ids = new ArrayList<>();
|
// initialize with the existing message IDs
|
||||||
ids.add(logMessage.getId());
|
List<String> ids = this.operationsLog.getMessages(this.realmName).stream().map(LogMessage::getId)
|
||||||
|
.collect(toList());
|
||||||
|
|
||||||
for (int i = 0; i < MAX_MESSAGES * 2; i++) {
|
for (int i = 0; i < MAX_MESSAGES * 2; i++) {
|
||||||
LogMessage m = new LogMessage(this.realmName, SYSTEM_USER_AGENT,
|
LogMessage m = new LogMessage(this.realmName, SYSTEM_USER_AGENT,
|
||||||
Locator.valueOf(AGENT, "li.strolch.testbase", StrolchAgent.getUniqueId()),
|
Locator.valueOf(AGENT, "li.strolch.testbase", StrolchAgent.getUniqueId()),
|
||||||
|
@ -88,13 +89,13 @@ public class LogMessagesTestRunner {
|
||||||
}
|
}
|
||||||
|
|
||||||
// default is async persisting...
|
// default is async persisting...
|
||||||
Thread.sleep(100L);
|
Thread.sleep(1000L);
|
||||||
|
|
||||||
int trimSize = (int) (MAX_MESSAGES * 0.1);
|
int trimSize = (int) (MAX_MESSAGES * 0.1);
|
||||||
int expectedSize = MAX_MESSAGES - trimSize + 1;
|
int expectedSize = MAX_MESSAGES - trimSize + 2; // +2 => startup and first message
|
||||||
|
|
||||||
if (realm.getMode().isTransient()) {
|
if (realm.getMode().isTransient()) {
|
||||||
List<LogMessage> messages = this.operationsLog.getMessages(realmName);
|
List<LogMessage> messages = this.operationsLog.getMessages(this.realmName);
|
||||||
assertEquals(expectedSize, messages.size());
|
assertEquals(expectedSize, messages.size());
|
||||||
} else {
|
} else {
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue