You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@unomi.apache.org by js...@apache.org on 2023/03/02 15:03:05 UTC
[unomi] branch master updated: Unomi 739 update purge system (#578)
This is an automated email from the ASF dual-hosted git repository.
jsinovassinnaik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new 82d540608 Unomi 739 update purge system (#578)
82d540608 is described below
commit 82d5406080eb39d5dbba39e77302f1e09832b7bb
Author: jsinovassin <58...@users.noreply.github.com>
AuthorDate: Thu Mar 2 16:02:59 2023 +0100
Unomi 739 update purge system (#578)
* UNOMI-739 : add session and event purge property
* UNOMI-739 : update purge mechanism
* UNOMI-739 : remove deletion in rollover policy
* UNOMI-739 : remove useless code and fix test
---
.../apache/unomi/api/services/ClusterService.java | 1 +
.../apache/unomi/api/services/ProfileService.java | 16 ++-
.../org/apache/unomi/itests/ProfileServiceIT.java | 21 ++-
.../main/resources/etc/custom.system.properties | 8 +-
.../ElasticSearchPersistenceServiceImpl.java | 78 +++++------
.../unomi/persistence/spi/PersistenceService.java | 20 +++
.../rest/endpoints/ClusterServiceEndPoint.java | 1 +
.../services/impl/profiles/ProfileServiceImpl.java | 148 ++++++++++++++-------
.../resources/OSGI-INF/blueprint/blueprint.xml | 4 +
.../main/resources/org.apache.unomi.services.cfg | 10 +-
.../2.2.0/create_rollover_policy_query.json | 6 -
11 files changed, 204 insertions(+), 109 deletions(-)
diff --git a/api/src/main/java/org/apache/unomi/api/services/ClusterService.java b/api/src/main/java/org/apache/unomi/api/services/ClusterService.java
index 4c89ba94b..299ac9098 100644
--- a/api/src/main/java/org/apache/unomi/api/services/ClusterService.java
+++ b/api/src/main/java/org/apache/unomi/api/services/ClusterService.java
@@ -41,6 +41,7 @@ public interface ClusterService {
*
* @param date the Date before which all data needs to be removed
*/
+ @Deprecated
void purge(final Date date);
/**
diff --git a/api/src/main/java/org/apache/unomi/api/services/ProfileService.java b/api/src/main/java/org/apache/unomi/api/services/ProfileService.java
index 527e77fec..03da6ce9b 100644
--- a/api/src/main/java/org/apache/unomi/api/services/ProfileService.java
+++ b/api/src/main/java/org/apache/unomi/api/services/ProfileService.java
@@ -417,7 +417,6 @@ public interface ProfileService {
* Purge (delete) profiles
* example: Purge profile inactive since 10 days only:
* purgeProfiles(10, 0);
- *
* example: Purge profile created since 30 days only:
* purgeProfiles(0, 30);
*
@@ -427,8 +426,21 @@ public interface ProfileService {
void purgeProfiles(int inactiveNumberOfDays, int existsNumberOfDays);
/**
- * Purge (delete) monthly indices by removing old indices
+ * Purge (delete) session items
+ * @param existsNumberOfDays will purge sessions created since this number of days (0 or negative value, will have no effect)
+ */
+ void purgeSessionItems(int existsNumberOfDays);
+
+ /**
+ * Purge (delete) event items
+ * @param existsNumberOfDays will purge events created since this number of days (0 or negative value, will have no effect)
+ */
+ void purgeEventItems(int existsNumberOfDays);
+
+ /**
+ * Use purgeSessionItems and purgeEventItems to remove rollover items instead
* @param existsNumberOfMonths used to remove monthly indices older than this number of months
*/
+ @Deprecated
void purgeMonthlyItems(int existsNumberOfMonths);
}
diff --git a/itests/src/test/java/org/apache/unomi/itests/ProfileServiceIT.java b/itests/src/test/java/org/apache/unomi/itests/ProfileServiceIT.java
index 79a3bf843..455ab9073 100644
--- a/itests/src/test/java/org/apache/unomi/itests/ProfileServiceIT.java
+++ b/itests/src/test/java/org/apache/unomi/itests/ProfileServiceIT.java
@@ -372,30 +372,37 @@ public class ProfileServiceIT extends BaseIT {
(count) -> count == (450 + originalEventsCount), 1000, 100);
// Should have no effect
- profileService.purgeMonthlyItems(0);
+ profileService.purgeSessionItems(0);
keepTrying("Sessions number should be 450", () -> persistenceService.getAllItemsCount(Session.ITEM_TYPE),
(count) -> count == (450 + originalSessionsCount), 1000, 100);
+ profileService.purgeEventItems(0);
keepTrying("Events number should be 450", () -> persistenceService.getAllItemsCount(Event.ITEM_TYPE),
(count) -> count == (450 + originalEventsCount), 1000, 100);
- // Should have no effect there is no monthly items older than 40 months
- profileService.purgeMonthlyItems(40);
+ // Should have no effect there is no sessions items older than 1200 days
+ profileService.purgeSessionItems(1200);
keepTrying("Sessions number should be 450", () -> persistenceService.getAllItemsCount(Session.ITEM_TYPE),
(count) -> count == (450 + originalSessionsCount), 1000, 100);
+ // Should have no effect there is no events items older than 1200 days
+ profileService.purgeEventItems(1200);
keepTrying("Events number should be 450", () -> persistenceService.getAllItemsCount(Event.ITEM_TYPE),
(count) -> count == (450 + originalEventsCount), 1000, 100);
- // Should purge monthly items older than 25 days
- profileService.purgeMonthlyItems(25);
+ // Should purge sessions older than 750 days
+ profileService.purgeSessionItems(750);
keepTrying("Sessions number should be 300", () -> persistenceService.getAllItemsCount(Session.ITEM_TYPE),
(count) -> count == (300 + originalSessionsCount), 1000, 100);
+ // Should purge events older than 750 days
+ profileService.purgeEventItems(750);
keepTrying("Events number should be 300", () -> persistenceService.getAllItemsCount(Event.ITEM_TYPE),
(count) -> count == (300 + originalEventsCount), 1000, 100);
- // Should purge monthly items older than 5 days
- profileService.purgeMonthlyItems(5);
+ // Should purge sessions older than 150 days
+ profileService.purgeSessionItems(150);
keepTrying("Sessions number should be 150", () -> persistenceService.getAllItemsCount(Session.ITEM_TYPE),
(count) -> count == (150 + originalSessionsCount), 1000, 100);
+ // Should purge events older than 150 days
+ profileService.purgeEventItems(150);
keepTrying("Events number should be 150", () -> persistenceService.getAllItemsCount(Event.ITEM_TYPE),
(count) -> count == (150 + originalEventsCount), 1000, 100);
}
diff --git a/package/src/main/resources/etc/custom.system.properties b/package/src/main/resources/etc/custom.system.properties
index e8b049d40..ab10c3a8e 100644
--- a/package/src/main/resources/etc/custom.system.properties
+++ b/package/src/main/resources/etc/custom.system.properties
@@ -162,8 +162,14 @@ org.apache.unomi.profile.purge.interval=${env:UNOMI_PROFILE_PURGE_INTERVAL:-1}
org.apache.unomi.profile.purge.inactiveTime=${env:UNOMI_PROFILE_PURGE_INACTIVETIME:-180}
# Purge profiles that have been created for a specific number of days
org.apache.unomi.profile.purge.existTime=${env:UNOMI_PROFILE_PURGE_EXISTTIME:--1}
-# Purge all monthly indexes (sessions/events) that have been created for a specific number of months
+# Deprecated: Use org.apache.unomi.sessions.purge.existTime and org.apache.unomi.events.purge.existTime instead.
+# If there is no value for org.apache.unomi.sessions.purge.existTime or org.apache.unomi.events.purge.existTime, the value
+# of this property will be use for the purge.
org.apache.unomi.monthly.index.purge.existTime=${env:UNOMI_MONTHLY_INDEX_PURGE_EXISTTIME:-12}
+# Purge sessions that have been created for a specific number of days
+org.apache.unomi.session.purge.existTime=${env:UNOMI_SESSION_PURGE_EXISTTIME:--1}
+# Purge events that have been created for a specific number of days
+org.apache.unomi.event.purge.existTime=${env:UNOMI_EVENT_PURGE_EXISTTIME:--1}
# Refresh Elasticsearch after saving a profile
org.apache.unomi.profile.forceRefreshOnSave=${env:UNOMI_PROFILE_REFRESH_ON_SAVE:-false}
# When performing segment updates, this controls the size of the scrolling query size used to iterate over all the
diff --git a/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java b/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java
index 13a802db0..d14d284d9 100644
--- a/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java
+++ b/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java
@@ -73,13 +73,7 @@ import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
-import org.elasticsearch.client.Node;
-import org.elasticsearch.client.RequestOptions;
-import org.elasticsearch.client.Requests;
-import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
-import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
-import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.client.*;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
@@ -154,8 +148,6 @@ import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.security.SecureRandom;
import java.security.cert.X509Certificate;
-import java.text.ParseException;
-import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -170,6 +162,8 @@ import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.elasticsearch.index.query.QueryBuilders.termQuery;
@@ -1409,10 +1403,6 @@ public class ElasticSearchPersistenceServiceImpl implements PersistenceService,
);
phases.put("hot", new Phase("hot", TimeValue.ZERO, hotActions));
- // TODO - Handle this with the purge https://issues.apache.org/jira/browse/UNOMI-726
- Map<String, LifecycleAction> deleteActions = Collections.singletonMap(DeleteAction.NAME, new DeleteAction());
- phases.put("delete", new Phase("delete", new TimeValue(90, TimeUnit.DAYS), deleteActions));
-
LifecyclePolicy policy = new LifecyclePolicy(indexPrefix + "-" + ROLLOVER_LIFECYCLE_NAME, phases);
PutLifecyclePolicyRequest request = new PutLifecyclePolicyRequest(policy);
org.elasticsearch.client.core.AcknowledgedResponse putLifecyclePolicy = client.indexLifecycle().putLifecyclePolicy(request, RequestOptions.DEFAULT);
@@ -1454,8 +1444,8 @@ public class ElasticSearchPersistenceServiceImpl implements PersistenceService,
}
}
- public boolean removeIndex(final String itemType) {
- String index = getIndex(itemType);
+ public boolean removeIndex(final String itemType, boolean addPrefix){
+ String index = addPrefix ? getIndex(itemType) : itemType;
Boolean result = new InClassLoaderExecute<Boolean>(metricsService, this.getClass().getName() + ".removeIndex", this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) {
protected Boolean execute(Object... args) throws IOException {
@@ -1475,6 +1465,9 @@ public class ElasticSearchPersistenceServiceImpl implements PersistenceService,
return result;
}
}
+ public boolean removeIndex(final String itemType) {
+ return removeIndex(itemType, true);
+ }
private void internalCreateRolloverTemplate(String itemName) throws IOException {
String rolloverAlias = indexPrefix + "-" + itemName;
@@ -1925,6 +1918,29 @@ public class ElasticSearchPersistenceServiceImpl implements PersistenceService,
}
}
+ @Override
+ public Map<String, Long> docCountPerIndex(String... indexes) {
+ return new InClassLoaderExecute<Map<String, Long>>(metricsService, this.getClass().getName() + ".docCountPerIndex", this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) {
+ @Override
+ protected Map<String, Long> execute(Object... args) throws IOException {
+ List<String> indexesForQuery = Stream.of(indexes).map(index -> getIndexNameForQuery(index)).collect(Collectors.toList());
+ String[] itemsArray = new String[indexesForQuery.size()];
+ itemsArray = indexesForQuery.toArray(itemsArray);
+ GetIndexRequest request = new GetIndexRequest(itemsArray);
+ GetIndexResponse getIndexResponse = client.indices().get(request, RequestOptions.DEFAULT);
+
+ Map<String, Long> countPerIndex = new HashMap<>();
+
+ for (String index : getIndexResponse.getIndices()) {
+ CountRequest countRequest = new CountRequest(index);
+ CountResponse response = client.count(countRequest, RequestOptions.DEFAULT);
+ countPerIndex.put(index, response.getCount());
+ }
+ return countPerIndex;
+ }
+ }.catchingExecuteInClassLoader(true);
+ }
+
private long queryCount(final QueryBuilder filter, final String itemType) {
return new InClassLoaderExecute<Long>(metricsService, this.getClass().getName() + ".queryCount", this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) {
@@ -2404,38 +2420,6 @@ public class ElasticSearchPersistenceServiceImpl implements PersistenceService,
@Override
public void purge(final Date date) {
- new InClassLoaderExecute<Object>(metricsService, this.getClass().getName() + ".purgeWithDate", this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) {
- @Override
- protected Object execute(Object... args) throws Exception {
-
- GetIndexRequest getIndexRequest = new GetIndexRequest(getAllIndexForQuery());
- GetIndexResponse getIndexResponse = client.indices().get(getIndexRequest, RequestOptions.DEFAULT);
- String[] indices = getIndexResponse.getIndices();
-
- SimpleDateFormat d = new SimpleDateFormat("yyyy-MM");
-
- List<String> toDelete = new ArrayList<>();
- for (String currentIndexName : indices) {
- int indexDatePrefixPos = currentIndexName.indexOf(INDEX_DATE_PREFIX);
- if (indexDatePrefixPos > -1) {
- try {
- Date indexDate = d.parse(currentIndexName.substring(indexDatePrefixPos + INDEX_DATE_PREFIX.length()));
-
- if (indexDate.before(date)) {
- toDelete.add(currentIndexName);
- }
- } catch (ParseException e) {
- throw new Exception("Cannot parse index name " + currentIndexName, e);
- }
- }
- }
- if (!toDelete.isEmpty()) {
- DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(toDelete.toArray(new String[toDelete.size()]));
- client.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
- }
- return null;
- }
- }.catchingExecuteInClassLoader(true);
}
@Override
diff --git a/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/PersistenceService.java b/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/PersistenceService.java
index d0911131d..751a34c85 100644
--- a/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/PersistenceService.java
+++ b/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/PersistenceService.java
@@ -680,8 +680,19 @@ public interface PersistenceService {
*
* @param date the date (not included) before which we want to erase all data
*/
+ @Deprecated
void purge(Date date);
+ /**
+ * Retrieves the number of document per indexes.
+ * If the index is a rollover index, each rollover index will be return with its own number of document
+ * For example: with "event" as parameter, the indexes named ...-event-000001, ...-event-000002 and so one will be returned
+ *
+ * @param indexes names of the indexes to count the documents
+ * @return Map where the key in the index name and the value is the number of document for this index
+ */
+ Map<String, Long> docCountPerIndex(String... indexes);
+
/**
* Retrieves all items of the specified Item subclass which specified ranged property is within the specified bounds, ordered according to the specified {@code sortBy} String
* and and paged: only {@code size} of them are retrieved, starting with the {@code offset}-th one.
@@ -733,6 +744,15 @@ public interface PersistenceService {
*/
boolean removeIndex(final String itemType);
+ /**
+ * Removes the index for the specified item type.
+ *
+ * @param itemType the item type
+ * @param addPrefix should add the index prefix to the itemType passed as parameter
+ * @return {@code true} if the operation was successful, {@code false} otherwise
+ */
+ boolean removeIndex(final String itemType, boolean addPrefix);
+
/**
* Removes all data associated with the provided scope.
*
diff --git a/rest/src/main/java/org/apache/unomi/rest/endpoints/ClusterServiceEndPoint.java b/rest/src/main/java/org/apache/unomi/rest/endpoints/ClusterServiceEndPoint.java
index 9cfeae9a2..d0c7cf29b 100644
--- a/rest/src/main/java/org/apache/unomi/rest/endpoints/ClusterServiceEndPoint.java
+++ b/rest/src/main/java/org/apache/unomi/rest/endpoints/ClusterServiceEndPoint.java
@@ -87,6 +87,7 @@ public class ClusterServiceEndPoint {
*/
@GET
@Path("/purge/{date}")
+ @Deprecated
public void purge(@PathParam("date") String date) {
try {
clusterService.purge(new SimpleDateFormat("yyyy-MM-dd").parse(date));
diff --git a/services/src/main/java/org/apache/unomi/services/impl/profiles/ProfileServiceImpl.java b/services/src/main/java/org/apache/unomi/services/impl/profiles/ProfileServiceImpl.java
index c12dc411f..7912cc6de 100644
--- a/services/src/main/java/org/apache/unomi/services/impl/profiles/ProfileServiceImpl.java
+++ b/services/src/main/java/org/apache/unomi/services/impl/profiles/ProfileServiceImpl.java
@@ -20,18 +20,7 @@ package org.apache.unomi.services.impl.profiles;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.lang3.StringUtils;
-import org.apache.unomi.api.BatchUpdate;
-import org.apache.unomi.api.Item;
-import org.apache.unomi.api.PartialList;
-import org.apache.unomi.api.Persona;
-import org.apache.unomi.api.PersonaSession;
-import org.apache.unomi.api.PersonaWithSessions;
-import org.apache.unomi.api.Profile;
-import org.apache.unomi.api.ProfileAlias;
-import org.apache.unomi.api.PropertyMergeStrategyExecutor;
-import org.apache.unomi.api.PropertyMergeStrategyType;
-import org.apache.unomi.api.PropertyType;
-import org.apache.unomi.api.Session;
+import org.apache.unomi.api.*;
import org.apache.unomi.api.conditions.Condition;
import org.apache.unomi.api.conditions.ConditionType;
import org.apache.unomi.api.query.Query;
@@ -56,25 +45,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.URL;
-import java.util.ArrayList;
-import java.util.Calendar;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.Date;
-import java.util.Enumeration;
-import java.util.GregorianCalendar;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.TimerTask;
-import java.util.TreeSet;
-import java.util.UUID;
+import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -203,7 +174,14 @@ public class ProfileServiceImpl implements ProfileService, SynchronousBundleList
private Integer purgeProfileExistTime = 0;
private Integer purgeProfileInactiveTime = 0;
+
+ /**
+ * Use purgeSessionExistTime and purgeEventExistTime instead
+ */
+ @Deprecated
private Integer purgeSessionsAndEventsTime = 0;
+ private Integer purgeSessionExistTime = 0;
+ private Integer purgeEventExistTime = 0;
private Integer purgeProfileInterval = 0;
private TimerTask purgeTask = null;
private long propertiesRefreshInterval = 10000;
@@ -256,6 +234,7 @@ public class ProfileServiceImpl implements ProfileService, SynchronousBundleList
}
}
bundleContext.addBundleListener(this);
+ initializeDefaultPurgeValuesIfNecessary();
initializePurge();
schedulePropertyTypeLoad();
logger.info("Profile service initialized.");
@@ -283,6 +262,22 @@ public class ProfileServiceImpl implements ProfileService, SynchronousBundleList
private void processBundleStop(BundleContext bundleContext) {
}
+ /**
+ * Fill purgeEventExistTime and purgeSessionExistTime with the old property purgeSessionsAndEventsTime
+ * if there is no value set for these properties. This is done to allow the using of the old property.
+ * This method should be removed once the purgeSessionsAndEventsTime property is deleted.
+ */
+ private void initializeDefaultPurgeValuesIfNecessary() {
+ if (purgeSessionsAndEventsTime > 0) {
+ if (purgeEventExistTime <= 0) {
+ purgeEventExistTime = purgeSessionsAndEventsTime * 30;
+ }
+ if (purgeSessionExistTime <= 0) {
+ purgeSessionExistTime = purgeSessionsAndEventsTime * 30;
+ }
+ }
+ }
+
public void setPurgeProfileExistTime(Integer purgeProfileExistTime) {
this.purgeProfileExistTime = purgeProfileExistTime;
}
@@ -291,6 +286,7 @@ public class ProfileServiceImpl implements ProfileService, SynchronousBundleList
this.purgeProfileInactiveTime = purgeProfileInactiveTime;
}
+ @Deprecated
public void setPurgeSessionsAndEventsTime(Integer purgeSessionsAndEventsTime) {
this.purgeSessionsAndEventsTime = purgeSessionsAndEventsTime;
}
@@ -299,6 +295,14 @@ public class ProfileServiceImpl implements ProfileService, SynchronousBundleList
this.purgeProfileInterval = purgeProfileInterval;
}
+ public void setPurgeSessionExistTime(Integer purgeSessionExistTime) {
+ this.purgeSessionExistTime = purgeSessionExistTime;
+ }
+
+ public void setPurgeEventExistTime(Integer purgeEventExistTime) {
+ this.purgeEventExistTime = purgeEventExistTime;
+ }
+
private void schedulePropertyTypeLoad() {
propertyTypeLoadTask = new TimerTask() {
@Override
@@ -366,26 +370,86 @@ public class ProfileServiceImpl implements ProfileService, SynchronousBundleList
}
}
+ private <T extends Item> void purgeRolloverItems(int existsNumberOfDays, Class<T> clazz) {
+ if (existsNumberOfDays > 0) {
+ String conditionType = null;
+ String itemType = null;
+
+ if (clazz.getName().equals(Event.class.getName())) {
+ conditionType = "eventPropertyCondition";
+ itemType = Event.ITEM_TYPE;
+ } else if (clazz.getName().equals(Session.class.getName())) {
+ conditionType = "sessionPropertyCondition";
+ itemType = Session.ITEM_TYPE;
+ }
+
+ ConditionType propertyConditionType = definitionsService.getConditionType(conditionType);
+ if (propertyConditionType == null) {
+ // definition service not yet fully instantiate
+ return;
+ }
+
+ Condition condition = new Condition(propertyConditionType);
+
+ condition.setParameter("propertyName", "timeStamp");
+ condition.setParameter("comparisonOperator", "lessThanOrEqualTo");
+ condition.setParameter("propertyValueDateExpr", "now-" + existsNumberOfDays + "d");
+ persistenceService.removeByQuery(condition, clazz);
+ deleteEmptyRolloverIndex(itemType);
+ }
+ }
+
+ @Override
+ public void purgeSessionItems(int existsNumberOfDays) {
+ if (existsNumberOfDays > 0) {
+ logger.info("Purging: Sessions created since more than {} days", existsNumberOfDays);
+ purgeRolloverItems(existsNumberOfDays, Session.class);
+ }
+ }
+
+ @Override
+ public void purgeEventItems(int existsNumberOfDays) {
+ if (existsNumberOfDays > 0) {
+ logger.info("Purging: Events created since more than {} days", existsNumberOfDays);
+ purgeRolloverItems(existsNumberOfDays, Event.class);
+ }
+ }
+
+ @Deprecated
@Override
public void purgeMonthlyItems(int existsNumberOfMonths) {
- if (existsNumberOfMonths > 0) {
- logger.info("Purging: Monthly items (sessions/events) created before {} months", existsNumberOfMonths);
- persistenceService.purge(getMonth(-existsNumberOfMonths).getTime());
+
+ }
+
+ public void deleteEmptyRolloverIndex(String indexName) {
+ TreeMap<String, Long> countsPerIndex = new TreeMap<>(persistenceService.docCountPerIndex(indexName));
+ if (countsPerIndex.size() >= 1) {
+ // do not check the last index, because it's the one used to write documents
+ countsPerIndex.pollLastEntry();
+ countsPerIndex.forEach((index, count) -> {
+ if (count == 0) {
+ persistenceService.removeIndex(index, false);
+ }
+ });
}
}
private void initializePurge() {
logger.info("Purge: Initializing");
- if (purgeProfileInactiveTime > 0 || purgeProfileExistTime > 0 || purgeSessionsAndEventsTime > 0) {
+ if (purgeProfileInactiveTime > 0 || purgeProfileExistTime > 0 || purgeSessionExistTime > 0 || purgeEventExistTime > 0) {
if (purgeProfileInactiveTime > 0) {
logger.info("Purge: Profile with no visits since more than {} days, will be purged", purgeProfileInactiveTime);
}
if (purgeProfileExistTime > 0) {
logger.info("Purge: Profile created since more than {} days, will be purged", purgeProfileExistTime);
}
- if (purgeSessionsAndEventsTime > 0) {
- logger.info("Purge: Monthly items (sessions/events) created since more than {} months, will be purged", purgeSessionsAndEventsTime);
+
+ if (purgeSessionExistTime > 0) {
+ logger.info("Purge: Session items created since more than {} days, will be purged", purgeSessionExistTime);
+ }
+ if (purgeEventExistTime > 0) {
+ logger.info("Purge: Event items created since more than {} days, will be purged", purgeEventExistTime);
}
purgeTask = new TimerTask() {
@@ -399,8 +463,8 @@ public class ProfileServiceImpl implements ProfileService, SynchronousBundleList
purgeProfiles(purgeProfileInactiveTime, purgeProfileExistTime);
// Monthly items purge
- purgeMonthlyItems(purgeSessionsAndEventsTime);
-
+ purgeSessionItems(purgeSessionExistTime);
+ purgeEventItems(purgeEventExistTime);
logger.info("Purge: executed in {} ms", System.currentTimeMillis() - purgeStartTime);
} catch (Throwable t) {
logger.error("Error while purging", t);
@@ -416,12 +480,6 @@ public class ProfileServiceImpl implements ProfileService, SynchronousBundleList
}
}
- private GregorianCalendar getMonth(int offset) {
- GregorianCalendar gc = new GregorianCalendar();
- gc = new GregorianCalendar(gc.get(Calendar.YEAR), gc.get(Calendar.MONTH), 1);
- gc.add(Calendar.MONTH, offset);
- return gc;
- }
public long getAllProfilesCount() {
return persistenceService.getAllItemsCount(Profile.ITEM_TYPE);
diff --git a/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml
index 3347588ab..f49f3c849 100644
--- a/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml
+++ b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml
@@ -29,6 +29,8 @@
<cm:property name="profile.purge.inactiveTime" value="180"/>
<cm:property name="profile.purge.existTime" value="-1"/>
<cm:property name="monthly.index.purge.existTime" value="12"/>
+ <cm:property name="session.purge.existTime" value="-1"/>
+ <cm:property name="event.purge.existTime" value="-1"/>
<cm:property name="segment.update.batchSize" value="1000"/>
<cm:property name="profile.forceRefreshOnSave" value="false"/>
<cm:property name="definitions.refresh.interval" value="10000"/>
@@ -235,6 +237,8 @@
<property name="purgeProfileInactiveTime" value="${services.profile.purge.inactiveTime}"/>
<property name="purgeProfileExistTime" value="${services.profile.purge.existTime}"/>
<property name="purgeSessionsAndEventsTime" value="${services.monthly.index.purge.existTime}"/>
+ <property name="purgeSessionExistTime" value="${services.session.purge.existTime}"/>
+ <property name="purgeEventExistTime" value="${services.event.purge.existTime}"/>
<property name="forceRefreshOnSave" value="${services.profile.forceRefreshOnSave}"/>
<property name="propertiesRefreshInterval" value="${services.properties.refresh.interval}"/>
<property name="schedulerService" ref="schedulerServiceImpl"/>
diff --git a/services/src/main/resources/org.apache.unomi.services.cfg b/services/src/main/resources/org.apache.unomi.services.cfg
index 239305529..818b9ca78 100644
--- a/services/src/main/resources/org.apache.unomi.services.cfg
+++ b/services/src/main/resources/org.apache.unomi.services.cfg
@@ -27,9 +27,17 @@ profile.purge.existTime=${org.apache.unomi.profile.purge.existTime:--1}
# Refresh Elasticsearch after saving a profile
profile.forceRefreshOnSave=${org.apache.unomi.profile.forceRefreshOnSave:-false}
-# Purge all monthly indexes (sessions/events) that have been created for a specific number of months
+# Deprecated: Use session.purge.existTime and event.purge.existTime instead.
+# If there is no value for session.purge.existTime or event.purge.existTime, the value
+# of this property will be use for the purge.
monthly.index.purge.existTime=${org.apache.unomi.monthly.index.purge.existTime:-12}
+# Purge sessions that have been created for a specific number of days
+session.purge.existTime=${org.apache.unomi.session.purge.existTime:--1}
+
+# Purge events that have been created for a specific number of days
+event.purge.existTime=${org.apache.unomi.event.purge.existTime:--1}
+
# When performing segment updates, this controls the size of the scrolling query size used to iterate over all the
# profiles that need updating
segment.update.batchSize=${org.apache.unomi.segment.update.batchSize:-1000}
diff --git a/tools/shell-commands/src/main/resources/requestBody/2.2.0/create_rollover_policy_query.json b/tools/shell-commands/src/main/resources/requestBody/2.2.0/create_rollover_policy_query.json
index c9bc94a29..2084db3e7 100644
--- a/tools/shell-commands/src/main/resources/requestBody/2.2.0/create_rollover_policy_query.json
+++ b/tools/shell-commands/src/main/resources/requestBody/2.2.0/create_rollover_policy_query.json
@@ -7,12 +7,6 @@
#rolloverHotActions
}
}
- },
- "delete": {
- "min_age": "90d",
- "actions": {
- "delete": {}
- }
}
}
}