Allow more flexible database purge policy configuration (#268)

sensorhub-core/src/main/java/org/sensorhub/api/database/IObsSystemDbAutoPurgePolicy.java

@@ -16,6 +16,7 @@
 
 import org.slf4j.Logger;
 
+import java.util.Collection;
 
 /**
  * <p>
@@ -35,6 +36,7 @@ public interface IObsSystemDbAutoPurgePolicy
  * for this aging policy
  * @param db
  * @param log
+ * @param systemUIDs
  */
- public void trimStorage(IObsSystemDatabase db, Logger log); 
+ public void trimStorage(IObsSystemDatabase db, Logger log, Collection<String> systemUIDs);
 }

sensorhub-core/src/main/java/org/sensorhub/impl/database/system/HistoricalObsAutoPurgeConfig.java

@@ -17,6 +17,8 @@
 import org.sensorhub.api.config.DisplayInfo;
 import org.sensorhub.api.database.IObsSystemDbAutoPurgePolicy;
 
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * <p>
@@ -36,6 +38,10 @@ public abstract class HistoricalObsAutoPurgeConfig
  @DisplayInfo(label="Purge Execution Period", desc="Execution period of the purge policy (in seconds)")
  public double purgePeriod = 3600.0;
 
-
+ @DisplayInfo.Required
+ @DisplayInfo.FieldType(DisplayInfo.FieldType.Type.SYSTEM_UID)
+ @DisplayInfo(label="System UIDs", desc="Unique IDs of system drivers to purge")
+ public List<String> systemUIDs = new ArrayList<>(List.of("*"));
+
  public abstract IObsSystemDbAutoPurgePolicy getPolicy();
 }

sensorhub-core/src/main/java/org/sensorhub/impl/database/system/MaxAgeAutoPurgePolicy.java

@@ -25,6 +25,7 @@
 import org.sensorhub.api.datastore.system.SystemFilter;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
+import java.util.Collection;
 import org.sensorhub.api.database.IObsSystemDatabase;
 import org.slf4j.Logger;
 import org.vast.util.DateTimeFormat;
@@ -52,47 +53,62 @@ public class MaxAgeAutoPurgePolicy implements IObsSystemDbAutoPurgePolicy
 
 
  @Override
- public void trimStorage(IObsSystemDatabase db, Logger log)
+ public void trimStorage(IObsSystemDatabase db, Logger log, Collection<String> systemUIDs)
  {
  // remove all systems, datastreams, commandstreams and fois whose validity time period
  // ended before (now - max age) 
  var oldestRecordTime = Instant.now().minusSeconds((long)config.maxRecordAge);
- 
+
  long numProcRemoved = db.getSystemDescStore().removeEntries(new SystemFilter.Builder()
  .withValidTime(new TemporalFilter.Builder()
  .withOperator(RangeOp.CONTAINS)
  .withRange(Instant.MIN, oldestRecordTime)
  .build())
+ .withUniqueIDs(systemUIDs)
+ .includeMembers(true)
  .build());
 
  long numFoisRemoved = db.getFoiStore().removeEntries(new FoiFilter.Builder()
  .withValidTime(new TemporalFilter.Builder()
  .withOperator(RangeOp.CONTAINS)
  .withRange(Instant.MIN, oldestRecordTime)
  .build())
+ .withUniqueIDs(systemUIDs)
+ .includeMembers(true)
  .build());
 
  long numDsRemoved = db.getDataStreamStore().removeEntries(new DataStreamFilter.Builder()
  .withValidTime(new TemporalFilter.Builder()
  .withOperator(RangeOp.CONTAINS)
  .withRange(Instant.MIN, oldestRecordTime)
  .build())
+ .withSystems(new SystemFilter.Builder()
+ .withUniqueIDs(systemUIDs)
+ .includeMembers(true)
+ .build())
  .build());
 
  long numCsRemoved = db.getCommandStreamStore().removeEntries(new CommandStreamFilter.Builder()
  .withValidTime(new TemporalFilter.Builder()
  .withOperator(RangeOp.CONTAINS)
  .withRange(Instant.MIN, oldestRecordTime)
  .build())
+ .withSystems(new SystemFilter.Builder()
+ .withUniqueIDs(systemUIDs)
+ .includeMembers(true)
+ .build())
  .build());
 
  // for each remaining datastream, remove all obs with a timestamp older than
  // the latest result time minus the max age
  long numObsRemoved = 0;
- var allDataStreams = db.getDataStreamStore().selectEntries(db.getDataStreamStore().selectAllFilter()).iterator();
- while (allDataStreams.hasNext())
+ var dataStreams = db.getDataStreamStore()
+ .selectEntries(new DataStreamFilter.Builder()
+ .withSystems(new SystemFilter.Builder()
+ .withUniqueIDs(systemUIDs).includeMembers(true).build()).build()).iterator();
+ while (dataStreams.hasNext())
  {
- var dsEntry = allDataStreams.next();
+ var dsEntry = dataStreams.next();
  var dsID = dsEntry.getKey().getInternalID();
  var resultTimeRange = dsEntry.getValue().getResultTimeRange();
 
@@ -109,10 +125,13 @@ public void trimStorage(IObsSystemDatabase db, Logger log)
  // for each remaining command stream, remove all commands and status with a timestamp older than
  // the latest issue time minus the max age
  long numCmdRemoved = 0;
- var allCmdStreams = db.getCommandStreamStore().selectEntries(db.getCommandStreamStore().selectAllFilter()).iterator();
- while (allCmdStreams.hasNext())
+ var cmdStreams = db.getCommandStreamStore().selectEntries(
+ new CommandStreamFilter.Builder()
+ .withSystems(new SystemFilter.Builder()
+ .withUniqueIDs(systemUIDs).includeMembers(true).build()).build()).iterator();
+ while (cmdStreams.hasNext())
  {
- var dsEntry = allCmdStreams.next();
+ var dsEntry = cmdStreams.next();
  var dsID = dsEntry.getKey().getInternalID();
  var issueTimeRange = dsEntry.getValue().getIssueTimeRange();
 

sensorhub-core/src/main/java/org/sensorhub/impl/database/system/SystemDriverDatabase.java

@@ -80,21 +80,27 @@ protected void doStart() throws SensorHubException
  {
  throw new DataStoreException("Cannot instantiate underlying database " + config.dbConfig.moduleClass, e);
  }
+
+ if(!config.autoPurgeConfig.isEmpty())
+ autoPurgeTimer = new Timer();
 
  // start auto-purge timer thread if policy is specified and enabled
- if (config.autoPurgeConfig != null && config.autoPurgeConfig.enabled)
+ for(var autoPurgeConfig : config.autoPurgeConfig)
  {
- final IObsSystemDbAutoPurgePolicy policy = config.autoPurgeConfig.getPolicy();
- autoPurgeTimer = new Timer();
- TimerTask task = new TimerTask() {
- public void run()
- {
- if (!db.isReadOnly())
- policy.trimStorage(db, logger);
- }
- };
-
- autoPurgeTimer.schedule(task, 0, (long)(config.autoPurgeConfig.purgePeriod*1000));
+ if (autoPurgeConfig != null && autoPurgeConfig.enabled)
+ {
+ var uids = Collections.unmodifiableCollection(autoPurgeConfig.systemUIDs);
+ final IObsSystemDbAutoPurgePolicy policy = autoPurgeConfig.getPolicy();
+ TimerTask task = new TimerTask() {
+ public void run()
+ {
+ if (!db.isReadOnly())
+ policy.trimStorage(db, logger, uids);
+ }
+ };
+
+ autoPurgeTimer.schedule(task, 0, (long)(autoPurgeConfig.purgePeriod*1000));
+ }
  }
  }
 

sensorhub-core/src/main/java/org/sensorhub/impl/database/system/SystemDriverDatabaseConfig.java

@@ -14,6 +14,8 @@
 
 package org.sensorhub.impl.database.system;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Set;
 import java.util.TreeSet;
 import org.sensorhub.api.config.DisplayInfo;
@@ -46,7 +48,7 @@ public class SystemDriverDatabaseConfig extends DatabaseConfig
 
 
  @DisplayInfo(label="Automatic Purge Policy", desc="Policy for automatically purging historical data")
- public HistoricalObsAutoPurgeConfig autoPurgeConfig;
+ public List<HistoricalObsAutoPurgeConfig> autoPurgeConfig = new ArrayList<>();
 
 
  @DisplayInfo(desc="Minimum period between database commits (in ms)")