context, ConditionESQueryBuilderDispatcher dispatcher) {
throw new UnsupportedOperationException();
}
diff --git a/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ConditionESQueryBuilderDispatcher.java b/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ConditionESQueryBuilderDispatcher.java
index 39d796359..b7386e804 100644
--- a/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ConditionESQueryBuilderDispatcher.java
+++ b/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ConditionESQueryBuilderDispatcher.java
@@ -20,6 +20,7 @@
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import org.apache.unomi.api.conditions.Condition;
import org.apache.unomi.persistence.spi.conditions.ConditionContextHelper;
+import org.apache.unomi.persistence.spi.conditions.dispatcher.ConditionQueryBuilderDispatcher;
import org.apache.unomi.scripting.ScriptExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -28,7 +29,22 @@
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
-public class ConditionESQueryBuilderDispatcher {
+/**
+ * Dispatcher responsible for routing condition query building to the appropriate
+ * Elasticsearch-specific {@link ConditionESQueryBuilder} implementation.
+ *
+ * Responsibilities:
+ * - Maintain a registry of available query builders by their IDs
+ * - Resolve legacy queryBuilder IDs to the canonical IDs using centralized mapping in
+ * {@link org.apache.unomi.persistence.spi.conditions.dispatcher.ConditionQueryBuilderDispatcher}
+ * (with deprecation warnings)
+ * - Build query fragments (filters) and full queries from {@link org.apache.unomi.api.conditions.Condition}
+ *
+ * Notes:
+ * - Legacy mappings are centralized in SPI support; there is no runtime customization
+ * - New IDs are always preferred; legacy IDs trigger a warning and are mapped transparently
+ */
+public class ConditionESQueryBuilderDispatcher extends ConditionQueryBuilderDispatcher {
private static final Logger LOGGER = LoggerFactory.getLogger(ConditionESQueryBuilderDispatcher.class.getName());
private Map queryBuilders = new ConcurrentHashMap<>();
@@ -41,6 +57,12 @@ public void setScriptExecutor(ScriptExecutor scriptExecutor) {
this.scriptExecutor = scriptExecutor;
}
+ /**
+ * Registers a query builder implementation under the provided ID.
+ *
+ * @param name the queryBuilder ID (canonical, non-legacy)
+ * @param evaluator the query builder implementation
+ */
public void addQueryBuilder(String name, ConditionESQueryBuilder evaluator) {
queryBuilders.put(name, evaluator);
}
@@ -49,7 +71,6 @@ public void removeQueryBuilder(String name) {
queryBuilders.remove(name);
}
-
public String getQuery(Condition condition) {
return "{\"query\": " + getQueryBuilder(condition).toString() + "}";
}
@@ -57,7 +78,6 @@ public String getQuery(Condition condition) {
public Query getQueryBuilder(Condition condition) {
Query.Builder qb = new Query.Builder();
return qb.bool(b -> b.must(Query.of(q -> q.matchAll(m -> m))).filter(buildFilter(condition))).build();
-
}
public Query buildFilter(Condition condition) {
@@ -79,8 +99,14 @@ public Query buildFilter(Condition condition, Map context) {
throw new UnsupportedOperationException("No query builder defined for : " + condition.getConditionTypeId());
}
- if (queryBuilders.containsKey(queryBuilderKey)) {
- ConditionESQueryBuilder queryBuilder = queryBuilders.get(queryBuilderKey);
+ // Find the appropriate query builder key (new or legacy)
+ String finalQueryBuilderKey = findQueryBuilderKey(
+ queryBuilderKey,
+ condition.getConditionTypeId(),
+ queryBuilders::containsKey);
+
+ if (finalQueryBuilderKey != null) {
+ ConditionESQueryBuilder queryBuilder = queryBuilders.get(finalQueryBuilderKey);
Condition contextualCondition = ConditionContextHelper.getContextualCondition(condition, context, scriptExecutor);
if (contextualCondition != null) {
return queryBuilder.buildQuery(contextualCondition, context, this);
@@ -113,8 +139,14 @@ public long count(Condition condition, Map context) {
throw new UnsupportedOperationException("No query builder defined for : " + condition.getConditionTypeId());
}
- if (queryBuilders.containsKey(queryBuilderKey)) {
- ConditionESQueryBuilder queryBuilder = queryBuilders.get(queryBuilderKey);
+ // Find the appropriate query builder key (new or legacy)
+ String finalQueryBuilderKey = findQueryBuilderKey(
+ queryBuilderKey,
+ condition.getConditionTypeId(),
+ queryBuilders::containsKey);
+
+ if (finalQueryBuilderKey != null) {
+ ConditionESQueryBuilder queryBuilder = queryBuilders.get(finalQueryBuilderKey);
Condition contextualCondition = ConditionContextHelper.getContextualCondition(condition, context, scriptExecutor);
if (contextualCondition != null) {
return queryBuilder.count(contextualCondition, context, this);
@@ -126,4 +158,10 @@ public long count(Condition condition, Map context) {
LOGGER.debug("No matching query builder for condition {} and context {}", condition, context);
throw new UnsupportedOperationException();
}
+
+ @Override
+ protected Logger getLogger() {
+ return LOGGER;
+ }
+
}
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 20eb85a5d..ed3b744f0 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
@@ -1412,7 +1412,7 @@ protected Boolean execute(Object... args) throws IOException {
private void internalCreateRolloverTemplate(String itemName) throws IOException {
if (!mappings.containsKey(itemName)) {
- LOGGER.warn("Couldn't find mapping for item {}, won't create monthly index template", itemName);
+ LOGGER.warn("Couldn't find mapping for item {}, won't create rollover index template", itemName);
return;
}
diff --git a/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticsearchClientFactory.java b/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticsearchClientFactory.java
index 3873aca4c..6f2f82a79 100644
--- a/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticsearchClientFactory.java
+++ b/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticsearchClientFactory.java
@@ -109,4 +109,4 @@ public ElasticsearchClient build() {
return createClient(hosts, socketTimeout, sslContext, username, password);
}
}
-}
\ No newline at end of file
+}
diff --git a/persistence-elasticsearch/core/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/persistence-elasticsearch/core/src/main/resources/OSGI-INF/blueprint/blueprint.xml
index b94416a6a..77ebd264b 100644
--- a/persistence-elasticsearch/core/src/main/resources/OSGI-INF/blueprint/blueprint.xml
+++ b/persistence-elasticsearch/core/src/main/resources/OSGI-INF/blueprint/blueprint.xml
@@ -161,7 +161,7 @@
-
+
@@ -169,7 +169,7 @@
-
+
@@ -177,7 +177,7 @@
-
+
@@ -185,7 +185,7 @@
-
+
@@ -193,7 +193,7 @@
-
+
@@ -201,7 +201,7 @@
-
+
diff --git a/persistence-opensearch/conditions/pom.xml b/persistence-opensearch/conditions/pom.xml
new file mode 100644
index 000000000..b685ad7fa
--- /dev/null
+++ b/persistence-opensearch/conditions/pom.xml
@@ -0,0 +1,195 @@
+
+
+
+
+ 4.0.0
+
+
+ org.apache.unomi
+ unomi-persistence-opensearch
+ 3.1.0-SNAPSHOT
+
+
+ unomi-persistence-opensearch-conditions
+ Apache Unomi :: Persistence :: OpenSearch :: Conditions
+ Conditions OpenSearch persistence implementation for the Apache Unomi Context Server
+ bundle
+
+
+
+
+ org.apache.unomi
+ unomi-bom
+ ${project.version}
+ pom
+ import
+
+
+
+
+
+
+ org.osgi
+ osgi.core
+ provided
+
+
+ org.osgi
+ org.osgi.service.cm
+ provided
+
+
+
+ org.apache.unomi
+ unomi-api
+ ${project.version}
+ provided
+
+
+ org.apache.unomi
+ unomi-common
+ ${project.version}
+ provided
+
+
+ org.apache.unomi
+ unomi-persistence-spi
+ ${project.version}
+ provided
+
+
+ com.google.guava
+ guava
+ ${guava.version}
+
+
+
+ com.fasterxml.jackson.core
+ jackson-databind
+
+
+
+ org.apache.commons
+ commons-lang3
+
+
+ commons-collections
+ commons-collections
+
+
+ commons-io
+ commons-io
+
+
+
+ org.apache.unomi
+ unomi-metrics
+ ${project.version}
+ provided
+
+
+
+ junit
+ junit
+ test
+
+
+ com.hazelcast
+ hazelcast-all
+ 3.12.8
+ provided
+
+
+
+ org.apache.unomi
+ unomi-scripting
+ ${project.version}
+ provided
+
+
+
+ org.opensearch.client
+ opensearch-java
+ ${opensearch.version}
+ provided
+
+
+
+ org.apache.unomi
+ unomi-persistence-opensearch-core
+ ${project.version}
+ provided
+
+
+
+ joda-time
+ joda-time
+
+
+
+
+
+
+
+ org.apache.felix
+ maven-bundle-plugin
+ true
+
+
+
+ android.os;resolution:=optional,
+ com.conversantmedia.util.concurrent;resolution:=optional,
+ com.google.appengine.api;resolution:=optional,
+ com.google.appengine.api.utils;resolution:=optional,
+ com.google.apphosting.api;resolution:=optional,
+ jakarta.enterprise.context.spi;resolution:=optional,
+ jakarta.enterprise.inject.spi;resolution:=optional,
+ jdk.net;resolution:=optional,
+ org.apache.avalon.framework.logger;resolution:=optional,
+ org.apache.log;resolution:=optional,
+ org.apache.log4j,
+ org.brotli.dec;resolution:=optional,
+ org.conscrypt;resolution:=optional,
+ org.glassfish.hk2.osgiresourcelocator;resolution:=optional,
+ org.ietf.jgss;resolution:=optional,
+ org.joda.convert;resolution:=optional,
+ org.reactivestreams;resolution:=optional,
+ software.amazon.awssdk.auth.credentials;resolution:=optional,
+ software.amazon.awssdk.core.async;resolution:=optional,
+ software.amazon.awssdk.http;resolution:=optional,
+ software.amazon.awssdk.http.async;resolution:=optional,
+ software.amazon.awssdk.http.auth.aws.signer;resolution:=optional,
+ software.amazon.awssdk.http.auth.spi.signer;resolution:=optional,
+ software.amazon.awssdk.identity.spi;resolution:=optional,
+ software.amazon.awssdk.regions;resolution:=optional,
+ software.amazon.awssdk.utils;resolution:=optional,
+ software.amazon.awssdk.utils.builder;resolution:=optional,
+ sun.misc;resolution:=optional,
+ sun.nio.ch;resolution:=optional,
+ *
+
+ *;scope=compile|runtime
+ true
+
+
+
+
+
+
+
+
diff --git a/persistence-opensearch/conditions/src/main/java/org/apache/unomi/persistence/opensearch/querybuilders/advanced/GeoLocationByPointSessionConditionOSQueryBuilder.java b/persistence-opensearch/conditions/src/main/java/org/apache/unomi/persistence/opensearch/querybuilders/advanced/GeoLocationByPointSessionConditionOSQueryBuilder.java
new file mode 100644
index 000000000..4521f6e42
--- /dev/null
+++ b/persistence-opensearch/conditions/src/main/java/org/apache/unomi/persistence/opensearch/querybuilders/advanced/GeoLocationByPointSessionConditionOSQueryBuilder.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.unomi.persistence.opensearch.querybuilders.advanced;
+
+import org.apache.unomi.api.conditions.Condition;
+import org.apache.unomi.persistence.opensearch.ConditionOSQueryBuilder;
+import org.apache.unomi.persistence.opensearch.ConditionOSQueryBuilderDispatcher;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+
+import java.util.Map;
+
+public class GeoLocationByPointSessionConditionOSQueryBuilder implements ConditionOSQueryBuilder {
+ @Override
+ public Query buildQuery(Condition condition, Map context, ConditionOSQueryBuilderDispatcher dispatcher) {
+ String type = (String) condition.getParameter("type");
+ String name = condition.getParameter("name") == null ? "location" : (String) condition.getParameter("name");
+
+ if("circle".equals(type)) {
+ Double circleLatitude = ((Number) condition.getParameter("circleLatitude")).doubleValue();
+ Double circleLongitude = ((Number) condition.getParameter("circleLongitude")).doubleValue();
+ String distance = condition.getParameter("distance").toString();
+
+ if(circleLatitude != null && circleLongitude != null && distance != null) {
+ return Query.of(q -> q.geoDistance(g -> g.field(name).location(l -> l.latlon(latlong -> latlong.lat(circleLatitude).lon(circleLongitude))).distance(distance)));
+ }
+ } else if("rectangle".equals(type)) {
+ Double rectLatitudeNE = (Double) condition.getParameter("rectLatitudeNE");
+ Double rectLongitudeNE = (Double) condition.getParameter("rectLongitudeNE");
+ Double rectLatitudeSW = (Double) condition.getParameter("rectLatitudeSW");
+ Double rectLongitudeSW = (Double) condition.getParameter("rectLongitudeSW");
+
+ if(rectLatitudeNE != null && rectLongitudeNE != null && rectLatitudeSW != null && rectLongitudeSW != null) {
+ return Query.of(q -> q.geoBoundingBox(g -> g
+ .field(name)
+ .boundingBox(b -> b
+ .coords(c -> c
+ .top(rectLatitudeNE)
+ .left(rectLongitudeNE)
+ .bottom(rectLatitudeSW)
+ .right(rectLongitudeSW)
+ )
+ )
+ )
+ );
+ }
+ }
+
+ return null;
+ }
+
+}
diff --git a/persistence-opensearch/conditions/src/main/java/org/apache/unomi/persistence/opensearch/querybuilders/advanced/IdsConditionOSQueryBuilder.java b/persistence-opensearch/conditions/src/main/java/org/apache/unomi/persistence/opensearch/querybuilders/advanced/IdsConditionOSQueryBuilder.java
new file mode 100644
index 000000000..2ac25b63e
--- /dev/null
+++ b/persistence-opensearch/conditions/src/main/java/org/apache/unomi/persistence/opensearch/querybuilders/advanced/IdsConditionOSQueryBuilder.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.persistence.opensearch.querybuilders.advanced;
+
+import org.apache.unomi.api.conditions.Condition;
+import org.apache.unomi.persistence.opensearch.ConditionOSQueryBuilder;
+import org.apache.unomi.persistence.opensearch.ConditionOSQueryBuilderDispatcher;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+public class IdsConditionOSQueryBuilder implements ConditionOSQueryBuilder {
+
+ private int maximumIdsQueryCount = 5000;
+
+ public void setMaximumIdsQueryCount(int maximumIdsQueryCount) {
+ this.maximumIdsQueryCount = maximumIdsQueryCount;
+ }
+
+ @Override
+ public Query buildQuery(Condition condition, Map context, ConditionOSQueryBuilderDispatcher dispatcher) {
+ Collection ids = (Collection) condition.getParameter("ids");
+ Boolean match = (Boolean) condition.getParameter("match");
+
+ if (ids.size() > maximumIdsQueryCount) {
+ // Avoid building too big ids query - throw exception instead
+ throw new UnsupportedOperationException("Too many profiles, exceeding the maximum number of ids query count: " + maximumIdsQueryCount);
+ }
+
+ Query idsQuery = Query.of(q->q.ids(i->i.values(new ArrayList(ids))));
+ if (match) {
+ return idsQuery;
+ } else {
+ return Query.of(q->q.bool(b->b.mustNot(idsQuery)));
+ }
+ }
+}
diff --git a/persistence-opensearch/conditions/src/main/java/org/apache/unomi/persistence/opensearch/querybuilders/advanced/PastEventConditionOSQueryBuilder.java b/persistence-opensearch/conditions/src/main/java/org/apache/unomi/persistence/opensearch/querybuilders/advanced/PastEventConditionOSQueryBuilder.java
new file mode 100644
index 000000000..e1ebd4b30
--- /dev/null
+++ b/persistence-opensearch/conditions/src/main/java/org/apache/unomi/persistence/opensearch/querybuilders/advanced/PastEventConditionOSQueryBuilder.java
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.unomi.persistence.opensearch.querybuilders.advanced;
+
+import org.apache.unomi.api.Event;
+import org.apache.unomi.api.Profile;
+import org.apache.unomi.api.conditions.Condition;
+import org.apache.unomi.api.services.DefinitionsService;
+import org.apache.unomi.api.services.SegmentService;
+import org.apache.unomi.api.utils.ConditionBuilder;
+import org.apache.unomi.persistence.opensearch.ConditionOSQueryBuilder;
+import org.apache.unomi.persistence.opensearch.ConditionOSQueryBuilderDispatcher;
+import org.apache.unomi.persistence.spi.PersistenceService;
+import org.apache.unomi.persistence.spi.aggregate.TermsAggregate;
+import org.apache.unomi.persistence.spi.conditions.ConditionContextHelper;
+import org.apache.unomi.persistence.spi.conditions.PastEventConditionPersistenceQueryBuilder;
+import org.apache.unomi.scripting.ScriptExecutor;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+public class PastEventConditionOSQueryBuilder implements ConditionOSQueryBuilder, PastEventConditionPersistenceQueryBuilder {
+
+ private DefinitionsService definitionsService;
+ private PersistenceService persistenceService;
+ private SegmentService segmentService;
+ private ScriptExecutor scriptExecutor;
+
+ private int maximumIdsQueryCount = 5000;
+ private int aggregateQueryBucketSize = 5000;
+ private boolean pastEventsDisablePartitions = false;
+
+ public void setDefinitionsService(DefinitionsService definitionsService) {
+ this.definitionsService = definitionsService;
+ }
+
+ public void setPersistenceService(PersistenceService persistenceService) {
+ this.persistenceService = persistenceService;
+ }
+
+ public void setScriptExecutor(ScriptExecutor scriptExecutor) {
+ this.scriptExecutor = scriptExecutor;
+ }
+
+ public void setMaximumIdsQueryCount(int maximumIdsQueryCount) {
+ this.maximumIdsQueryCount = maximumIdsQueryCount;
+ }
+
+ public void setAggregateQueryBucketSize(int aggregateQueryBucketSize) {
+ this.aggregateQueryBucketSize = aggregateQueryBucketSize;
+ }
+
+ public void setPastEventsDisablePartitions(boolean pastEventsDisablePartitions) {
+ this.pastEventsDisablePartitions = pastEventsDisablePartitions;
+ }
+
+ public void setSegmentService(SegmentService segmentService) {
+ this.segmentService = segmentService;
+ }
+
+ @Override
+ public Query buildQuery(Condition condition, Map context, ConditionOSQueryBuilderDispatcher dispatcher) {
+ boolean eventsOccurred = getStrategyFromOperator((String) condition.getParameter("operator"));
+ int minimumEventCount = !eventsOccurred || condition.getParameter("minimumEventCount") == null ? 1 : (Integer) condition.getParameter("minimumEventCount");
+ int maximumEventCount = !eventsOccurred || condition.getParameter("maximumEventCount") == null ? Integer.MAX_VALUE : (Integer) condition.getParameter("maximumEventCount");
+ String generatedPropertyKey = (String) condition.getParameter("generatedPropertyKey");
+
+ if (generatedPropertyKey != null && generatedPropertyKey.equals(segmentService.getGeneratedPropertyKey((Condition) condition.getParameter("eventCondition"), condition))) {
+ // A property is already set on profiles matching the past event condition, use it to check the numbers of occurrences
+ return dispatcher.buildFilter(getProfileConditionForCounter(generatedPropertyKey, minimumEventCount, maximumEventCount, eventsOccurred), context);
+ } else {
+ // No property set - tries to build an idsQuery
+ // TODO see for deprecation, this should not happen anymore each past event condition should have a generatedPropertyKey
+ Condition eventCondition = getEventCondition(condition, context, null, definitionsService, scriptExecutor);
+ Set ids = getProfileIdsMatchingEventCount(eventCondition, minimumEventCount, maximumEventCount);
+ ConditionBuilder conditionBuilder = definitionsService.getConditionBuilder();
+ return dispatcher.buildFilter(conditionBuilder.condition("idsCondition").parameter("ids", ids).parameter("match", eventsOccurred).build(), context);
+ }
+ }
+
+ @Override
+ public long count(Condition condition, Map context, ConditionOSQueryBuilderDispatcher dispatcher) {
+ boolean eventsOccurred = getStrategyFromOperator((String) condition.getParameter("operator"));
+ int minimumEventCount = !eventsOccurred || condition.getParameter("minimumEventCount") == null ? 1 : (Integer) condition.getParameter("minimumEventCount");
+ int maximumEventCount = !eventsOccurred || condition.getParameter("maximumEventCount") == null ? Integer.MAX_VALUE : (Integer) condition.getParameter("maximumEventCount");
+ String generatedPropertyKey = (String) condition.getParameter("generatedPropertyKey");
+
+ if (generatedPropertyKey != null && generatedPropertyKey.equals(segmentService.getGeneratedPropertyKey((Condition) condition.getParameter("eventCondition"), condition))) {
+ // query profiles directly
+ return persistenceService.queryCount(getProfileConditionForCounter(generatedPropertyKey, minimumEventCount, maximumEventCount, eventsOccurred), Profile.ITEM_TYPE);
+ } else {
+ // No count filter - simply get the full number of distinct profiles
+ // TODO see for deprecation, this should not happen anymore each past event condition should have a generatedPropertyKey
+ Condition eventCondition = getEventCondition(condition, context, null, definitionsService, scriptExecutor);
+ if (eventsOccurred && minimumEventCount == 1 && maximumEventCount == Integer.MAX_VALUE) {
+ return persistenceService.getSingleValuesMetrics(eventCondition, new String[]{"card"}, "profileId.keyword", Event.ITEM_TYPE).get("_card").longValue();
+ }
+
+ Set profileIds = getProfileIdsMatchingEventCount(eventCondition, minimumEventCount, maximumEventCount);
+ ConditionBuilder conditionBuilder = definitionsService.getConditionBuilder();
+ return eventsOccurred ? profileIds.size() : persistenceService.queryCount(conditionBuilder.condition("idsCondition").parameter("ids", profileIds).parameter("match", false).build(), Profile.ITEM_TYPE);
+ }
+ }
+
+ public boolean getStrategyFromOperator(String operator) {
+ if (operator != null && !operator.equals("eventsOccurred") && !operator.equals("eventsNotOccurred")) {
+ throw new UnsupportedOperationException("Unsupported operator: " + operator + ", please use either 'eventsOccurred' or 'eventsNotOccurred'");
+ }
+ return operator == null || operator.equals("eventsOccurred");
+ }
+
+ private Condition getProfileConditionForCounter(String generatedPropertyKey, Integer minimumEventCount, Integer maximumEventCount, boolean eventsOccurred) {
+ if (eventsOccurred) {
+ return createEventOccurredCondition(generatedPropertyKey, minimumEventCount, maximumEventCount);
+ } else {
+ return createEventNotOccurredCondition(generatedPropertyKey);
+ }
+ }
+
+ private Condition createEventOccurredCondition(String generatedPropertyKey, Integer minimumEventCount, Integer maximumEventCount) {
+ ConditionBuilder conditionBuilder = definitionsService.getConditionBuilder();
+ ConditionBuilder.ConditionItem subConditionCount = conditionBuilder.profileProperty("systemProperties.pastEvents.count").between(minimumEventCount, maximumEventCount);
+ ConditionBuilder.ConditionItem subConditionKey = conditionBuilder.profileProperty("systemProperties.pastEvents.key").equalTo(generatedPropertyKey);
+ ConditionBuilder.ConditionItem booleanCondition = conditionBuilder.and(subConditionCount, subConditionKey);
+ return conditionBuilder.nested(booleanCondition, "systemProperties.pastEvents").build();
+ }
+
+ private Condition createEventNotOccurredCondition(String generatedPropertyKey) {
+ ConditionBuilder.ConditionItem counterMissing = createPastEventMustNotExistCondition(generatedPropertyKey);
+ ConditionBuilder conditionBuilder = definitionsService.getConditionBuilder();
+ ConditionBuilder.ConditionItem counterZero = conditionBuilder.profileProperty("systemProperties.pastEvents.count").equalTo(0);
+ ConditionBuilder.ConditionItem keyEquals = conditionBuilder.profileProperty("systemProperties.pastEvents.key").equalTo(generatedPropertyKey);
+ ConditionBuilder.ConditionItem keyExistsAndCounterZero = conditionBuilder.and(counterZero, keyEquals);
+ ConditionBuilder.ConditionItem nestedKeyExistsAndCounterZero = conditionBuilder.nested(keyExistsAndCounterZero, "systemProperties.pastEvents");
+ return conditionBuilder.or(counterMissing, nestedKeyExistsAndCounterZero).build();
+ }
+
+ private ConditionBuilder.ConditionItem createPastEventMustNotExistCondition(String generatedPropertyKey) {
+ ConditionBuilder conditionBuilder = definitionsService.getConditionBuilder();
+ ConditionBuilder.ConditionItem keyEquals = conditionBuilder.profileProperty("systemProperties.pastEvents.key").equalTo(generatedPropertyKey);
+ return conditionBuilder.not(keyEquals);
+ }
+
+ private Set getProfileIdsMatchingEventCount(Condition eventCondition, int minimumEventCount, int maximumEventCount) {
+ boolean noBoundaries = minimumEventCount == 1 && maximumEventCount == Integer.MAX_VALUE;
+ if (pastEventsDisablePartitions) {
+ Map eventCountByProfile = persistenceService.aggregateWithOptimizedQuery(eventCondition, new TermsAggregate("profileId"), Event.ITEM_TYPE, maximumIdsQueryCount);
+ eventCountByProfile.remove("_filtered");
+ return noBoundaries ?
+ eventCountByProfile.keySet() :
+ eventCountByProfile.entrySet()
+ .stream()
+ .filter(eventCountPerProfile -> (eventCountPerProfile.getValue() >= minimumEventCount && eventCountPerProfile.getValue() <= maximumEventCount))
+ .map(Map.Entry::getKey)
+ .collect(Collectors.toSet());
+ } else {
+ Set result = new HashSet<>();
+ // Get full cardinality to partition the terms aggregation
+ Map m = persistenceService.getSingleValuesMetrics(eventCondition, new String[]{"card"}, "profileId.keyword", Event.ITEM_TYPE);
+ long card = m.get("_card").longValue();
+
+ int numParts = (int) (card / aggregateQueryBucketSize) + 2;
+ for (int i = 0; i < numParts; i++) {
+ Map eventCountByProfile = persistenceService.aggregateWithOptimizedQuery(eventCondition, new TermsAggregate("profileId", i, numParts), Event.ITEM_TYPE);
+ if (eventCountByProfile != null) {
+ eventCountByProfile.remove("_filtered");
+ if (noBoundaries) {
+ result.addAll(eventCountByProfile.keySet());
+ } else {
+ for (Map.Entry entry : eventCountByProfile.entrySet()) {
+ if (entry.getValue() < minimumEventCount) {
+ // No more interesting buckets in this partition
+ break;
+ } else if (entry.getValue() <= maximumEventCount) {
+ result.add(entry.getKey());
+ }
+ }
+ }
+ }
+ }
+
+ return result;
+ }
+ }
+
+ public Condition getEventCondition(Condition condition, Map context, String profileId,
+ DefinitionsService definitionsService, ScriptExecutor scriptExecutor) {
+ Condition eventCondition;
+ try {
+ eventCondition = (Condition) condition.getParameter("eventCondition");
+ } catch (ClassCastException e) {
+ throw new IllegalArgumentException("Empty eventCondition");
+ }
+ if (eventCondition == null) {
+ throw new IllegalArgumentException("No eventCondition");
+ }
+ List l = new ArrayList();
+ Condition andCondition = new Condition();
+ andCondition.setConditionType(definitionsService.getConditionType("booleanCondition"));
+ andCondition.setParameter("operator", "and");
+ andCondition.setParameter("subConditions", l);
+
+ l.add(ConditionContextHelper.getContextualCondition(eventCondition, context, scriptExecutor));
+
+ if (profileId != null) {
+ Condition profileCondition = new Condition();
+ profileCondition.setConditionType(definitionsService.getConditionType("sessionPropertyCondition"));
+ profileCondition.setParameter("propertyName", "profileId");
+ profileCondition.setParameter("comparisonOperator", "equals");
+ profileCondition.setParameter("propertyValue", profileId);
+ l.add(profileCondition);
+ }
+
+ Integer numberOfDays = (Integer) condition.getParameter("numberOfDays");
+ String fromDate = (String) condition.getParameter("fromDate");
+ String toDate = (String) condition.getParameter("toDate");
+
+ if (numberOfDays != null) {
+ l.add(getTimeStampCondition("greaterThan", "propertyValueDateExpr", "now-" + numberOfDays + "d", definitionsService));
+ }
+ if (fromDate != null) {
+ l.add(getTimeStampCondition("greaterThanOrEqualTo", "propertyValueDate", fromDate, definitionsService));
+ }
+ if (toDate != null) {
+ l.add(getTimeStampCondition("lessThanOrEqualTo", "propertyValueDate", toDate, definitionsService));
+ }
+ return andCondition;
+ }
+
+ private static Condition getTimeStampCondition(String operator, String propertyValueParameter, Object propertyValue, DefinitionsService definitionsService) {
+ Condition endDateCondition = new Condition();
+ endDateCondition.setConditionType(definitionsService.getConditionType("sessionPropertyCondition"));
+ endDateCondition.setParameter("propertyName", "timeStamp");
+ endDateCondition.setParameter("comparisonOperator", operator);
+ endDateCondition.setParameter(propertyValueParameter, propertyValue);
+ return endDateCondition;
+ }
+}
diff --git a/persistence-opensearch/conditions/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/persistence-opensearch/conditions/src/main/resources/OSGI-INF/blueprint/blueprint.xml
new file mode 100644
index 000000000..6b4c88a67
--- /dev/null
+++ b/persistence-opensearch/conditions/src/main/resources/OSGI-INF/blueprint/blueprint.xml
@@ -0,0 +1,90 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ org.apache.unomi.persistence.opensearch.ConditionOSQueryBuilder
+ org.apache.unomi.persistence.spi.conditions.PastEventConditionPersistenceQueryBuilder
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/persistence-opensearch/core/pom.xml b/persistence-opensearch/core/pom.xml
new file mode 100644
index 000000000..092e998d2
--- /dev/null
+++ b/persistence-opensearch/core/pom.xml
@@ -0,0 +1,217 @@
+
+
+
+
+ 4.0.0
+
+
+ org.apache.unomi
+ unomi-persistence-opensearch
+ 3.1.0-SNAPSHOT
+
+
+ unomi-persistence-opensearch-core
+ Apache Unomi :: Persistence :: OpenSearch :: Core
+ Core OpenSearch persistence implementation for the Apache Unomi Context Server
+ bundle
+
+
+
+
+ org.apache.unomi
+ unomi-bom
+ ${project.version}
+ pom
+ import
+
+
+
+
+
+
+ org.osgi
+ osgi.core
+ provided
+
+
+ org.osgi
+ org.osgi.service.cm
+ provided
+
+
+
+ org.apache.unomi
+ unomi-api
+ ${project.version}
+ provided
+
+
+ org.apache.unomi
+ unomi-common
+ ${project.version}
+ provided
+
+
+ org.apache.unomi
+ unomi-persistence-spi
+ ${project.version}
+ provided
+
+
+ com.google.guava
+ guava
+ ${guava.version}
+
+
+
+ com.fasterxml.jackson.core
+ jackson-databind
+ provided
+
+
+ com.fasterxml.jackson.core
+ jackson-core
+ provided
+
+
+
+ org.apache.commons
+ commons-lang3
+ provided
+
+
+ commons-collections
+ commons-collections
+ provided
+
+
+ commons-io
+ commons-io
+ provided
+
+
+ joda-time
+ joda-time
+ provided
+
+
+
+ org.apache.unomi
+ unomi-metrics
+ ${project.version}
+ provided
+
+
+
+ junit
+ junit
+ test
+
+
+
+ org.apache.unomi
+ unomi-scripting
+ ${project.version}
+ provided
+
+
+
+ org.opensearch.client
+ opensearch-java
+ ${opensearch.version}
+
+
+
+ org.slf4j
+ slf4j-api
+ provided
+
+
+
+
+
+
+
+ org.apache.felix
+ maven-bundle-plugin
+ true
+
+
+
+ android.os;resolution:=optional,
+ com.google.appengine.*;resolution:=optional,
+ com.google.apphosting.*;resolution:=optional,
+ io.micrometer.*;resolution:=optional,
+ jakarta.enterprise.*;resolution:=optional,
+ jdk.net;resolution:=optional,
+ org.apache.avalon.framework.logger;resolution:=optional,
+ org.apache.log;resolution:=optional,
+ org.brotli.dec;resolution:=optional,
+ org.conscrypt;resolution:=optional,
+ org.glassfish.hk2.osgiresourcelocator;resolution:=optional,
+ org.ietf.jgss;resolution:=optional,
+ org.reactivestreams;resolution:=optional,
+ reactor.blockhound.*;resolution:=optional,
+ software.amazon.awssdk.*;resolution:=optional,
+ sun.misc;resolution:=optional,
+ sun.nio.ch;resolution:=optional,
+ *
+
+
+ org.opensearch.*;version="${opensearch.version}",
+ org.opensearch.index.query.*;version="${opensearch.version}",
+ org.apache.unomi.persistence.opensearch;version="${project.version}"
+
+ *;scope=compile|runtime
+ true
+
+ osgi.service;objectClass:List<String>="org.apache.unomi.persistence.spi.PersistenceService",
+ unomi.persistence;type=opensearch
+
+
+
+
+
+ org.codehaus.mojo
+ build-helper-maven-plugin
+
+
+ attach-artifacts
+ package
+
+ attach-artifact
+
+
+
+
+
+ src/main/resources/org.apache.unomi.persistence.opensearch.cfg
+
+ cfg
+ opensearchcfg
+
+
+
+
+
+
+
+
+
+
+
diff --git a/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/ConditionOSQueryBuilder.java b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/ConditionOSQueryBuilder.java
new file mode 100644
index 000000000..f0b1f048c
--- /dev/null
+++ b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/ConditionOSQueryBuilder.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.unomi.persistence.opensearch;
+
+import org.apache.unomi.api.conditions.Condition;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+
+import java.util.Map;
+
+/**
+ * SPI for building OpenSearch {@link Query} objects from Unomi {@link org.apache.unomi.api.conditions.Condition}
+ * instances. Implementations translate high-level conditions into OS queries and may optionally provide a
+ * {@link #count(Condition, Map, ConditionOSQueryBuilderDispatcher)} strategy when needed by callers.
+ */
+public interface ConditionOSQueryBuilder {
+
+ /**
+ * Builds an OpenSearch {@link Query} from the provided Unomi {@link Condition}.
+ * Implementations may use the {@code context} (e.g., resolved parameters) and delegate to
+ * the {@code dispatcher} for nested/child conditions.
+ *
+ * @param condition the condition to translate
+ * @param context additional context for parameter resolution and sub-builders
+ * @param dispatcher dispatcher to build sub-conditions when composing queries
+ * @return a concrete OpenSearch {@link Query}
+ */
+ Query buildQuery(Condition condition, Map context, ConditionOSQueryBuilderDispatcher dispatcher);
+
+ /**
+ * Optionally returns a fast count for the provided condition using an implementation-specific
+ * strategy. Default throws {@link UnsupportedOperationException}; override when a specialized
+ * count path exists.
+ *
+ * @param condition the condition to count
+ * @param context additional context for parameter resolution
+ * @param dispatcher dispatcher to count sub-conditions if needed
+ * @return the number of matching documents
+ */
+ default long count(Condition condition, Map context, ConditionOSQueryBuilderDispatcher dispatcher) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/ConditionOSQueryBuilderDispatcher.java b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/ConditionOSQueryBuilderDispatcher.java
new file mode 100644
index 000000000..01ed4f973
--- /dev/null
+++ b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/ConditionOSQueryBuilderDispatcher.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.unomi.persistence.opensearch;
+
+import org.apache.unomi.api.conditions.Condition;
+import org.apache.unomi.persistence.spi.conditions.ConditionContextHelper;
+import org.apache.unomi.persistence.spi.conditions.dispatcher.ConditionQueryBuilderDispatcher;
+import org.apache.unomi.scripting.ScriptExecutor;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Dispatcher responsible for routing condition query building to the appropriate
+ * OpenSearch-specific {@link ConditionOSQueryBuilder} implementation.
+ *
+ * Responsibilities:
+ * - Maintain a registry of available query builders by their IDs
+ * - Resolve legacy queryBuilder IDs to the canonical IDs using centralized mapping in
+ * {@link org.apache.unomi.persistence.spi.conditions.dispatcher.ConditionQueryBuilderDispatcher}
+ * (with deprecation warnings)
+ * - Build query fragments (filters) and full queries from {@link org.apache.unomi.api.conditions.Condition}
+ *
+ * Notes:
+ * - Legacy mappings are centralized in SPI support; there is no runtime customization
+ * - New IDs are always preferred; legacy IDs trigger a warning and are mapped transparently
+ */
+public class ConditionOSQueryBuilderDispatcher extends ConditionQueryBuilderDispatcher {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ConditionOSQueryBuilderDispatcher.class.getName());
+
+ private Map queryBuilders = new ConcurrentHashMap<>();
+ private ScriptExecutor scriptExecutor;
+
+ public ConditionOSQueryBuilderDispatcher() {
+ }
+
+ public void setScriptExecutor(ScriptExecutor scriptExecutor) {
+ this.scriptExecutor = scriptExecutor;
+ }
+
+ /**
+ * Registers a query builder implementation under the provided ID.
+ *
+ * @param name the queryBuilder ID (canonical, non-legacy)
+ * @param evaluator the query builder implementation
+ */
+ public void addQueryBuilder(String name, ConditionOSQueryBuilder evaluator) {
+ queryBuilders.put(name, evaluator);
+ }
+
+ public void removeQueryBuilder(String name) {
+ queryBuilders.remove(name);
+ }
+
+ public String getQuery(Condition condition) {
+ return "{\"query\": " + getQueryBuilder(condition).toString() + "}";
+ }
+
+ public Query getQueryBuilder(Condition condition) {
+ return Query.of(q->q.bool(b->b.must(Query.of(q2 -> q2.matchAll(t -> t))).filter(buildFilter(condition))));
+ }
+
+ public Query buildFilter(Condition condition) {
+ return buildFilter(condition, new HashMap<>());
+ }
+
+ public Query buildFilter(Condition condition, Map context) {
+ if (condition == null || condition.getConditionType() == null) {
+ throw new IllegalArgumentException("Condition is null or doesn't have type, impossible to build filter");
+ }
+
+ String queryBuilderKey = condition.getConditionType().getQueryBuilder();
+ if (queryBuilderKey == null && condition.getConditionType().getParentCondition() != null) {
+ context.putAll(condition.getParameterValues());
+ return buildFilter(condition.getConditionType().getParentCondition(), context);
+ }
+
+ if (queryBuilderKey == null) {
+ throw new UnsupportedOperationException("No query builder defined for : " + condition.getConditionTypeId());
+ }
+
+ // Find the appropriate query builder key (new or legacy)
+ String finalQueryBuilderKey = findQueryBuilderKey(
+ queryBuilderKey,
+ condition.getConditionTypeId(),
+ queryBuilders::containsKey);
+
+ if (finalQueryBuilderKey != null) {
+ ConditionOSQueryBuilder queryBuilder = queryBuilders.get(finalQueryBuilderKey);
+ Condition contextualCondition = ConditionContextHelper.getContextualCondition(condition, context, scriptExecutor);
+ if (contextualCondition != null) {
+ return queryBuilder.buildQuery(contextualCondition, context, this);
+ }
+ } else {
+ // if no matching
+ LOGGER.warn("No matching query builder. See debug log level for more information");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("No matching query builder for condition {} and context {}", condition, context);
+ }
+ }
+
+ return Query.of(q -> q.matchAll(t->t));
+ }
+
+ public long count(Condition condition) {
+ return count(condition, new HashMap<>());
+ }
+
+ public long count(Condition condition, Map context) {
+ if (condition == null || condition.getConditionType() == null) {
+ throw new IllegalArgumentException("Condition is null or doesn't have type, impossible to build filter");
+ }
+
+ String queryBuilderKey = condition.getConditionType().getQueryBuilder();
+ if (queryBuilderKey == null && condition.getConditionType().getParentCondition() != null) {
+ context.putAll(condition.getParameterValues());
+ return count(condition.getConditionType().getParentCondition(), context);
+ }
+
+ if (queryBuilderKey == null) {
+ throw new UnsupportedOperationException("No query builder defined for : " + condition.getConditionTypeId());
+ }
+
+ // Find the appropriate query builder key (new or legacy)
+ String finalQueryBuilderKey = findQueryBuilderKey(
+ queryBuilderKey,
+ condition.getConditionTypeId(),
+ queryBuilders::containsKey);
+
+ if (finalQueryBuilderKey != null) {
+ ConditionOSQueryBuilder queryBuilder = queryBuilders.get(finalQueryBuilderKey);
+ Condition contextualCondition = ConditionContextHelper.getContextualCondition(condition, context, scriptExecutor);
+ if (contextualCondition != null) {
+ return queryBuilder.count(contextualCondition, context, this);
+ }
+ }
+
+ // if no matching
+ LOGGER.warn("No matching query builder. See debug log level for more information");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("No matching query builder for condition {} and context {}", condition, context);
+ }
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ protected Logger getLogger() {
+ return LOGGER;
+ }
+
+}
diff --git a/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OSCustomObjectMapper.java b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OSCustomObjectMapper.java
new file mode 100644
index 000000000..e3fed7d36
--- /dev/null
+++ b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OSCustomObjectMapper.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.persistence.opensearch;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.unomi.api.Event;
+import org.apache.unomi.api.Item;
+import org.apache.unomi.persistence.spi.CustomObjectMapper;
+
+import java.util.Map;
+
+/**
+ * This CustomObjectMapper is used to avoid the version parameter to be registered in OpenSearch
+ */
+public class OSCustomObjectMapper extends CustomObjectMapper {
+
+ private static final long serialVersionUID = -5017620674440085575L;
+
+ public OSCustomObjectMapper() {
+ super();
+ this.addMixIn(Item.class, OSItemMixIn.class);
+ this.addMixIn(Event.class, OSEventMixIn.class);
+ this.configOverride(Map.class)
+ .setInclude(JsonInclude.Value.construct(JsonInclude.Include.ALWAYS, JsonInclude.Include.ALWAYS));
+ }
+
+ public static ObjectMapper getObjectMapper() {
+ return OSCustomObjectMapper.Holder.INSTANCE;
+ }
+
+ private static class Holder {
+ static final OSCustomObjectMapper INSTANCE = new OSCustomObjectMapper();
+ }
+}
diff --git a/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OSEventMixIn.java b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OSEventMixIn.java
new file mode 100644
index 000000000..28dd28cf4
--- /dev/null
+++ b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OSEventMixIn.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.persistence.opensearch;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
+/**
+ * This mixin is used in OSCustomObjectMapper to prevent the persistent parameter from being registered in ES
+ */
+public abstract class OSEventMixIn {
+
+ public OSEventMixIn() { }
+
+ @JsonIgnore abstract boolean isPersistent();
+}
diff --git a/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OSItemMixIn.java b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OSItemMixIn.java
new file mode 100644
index 000000000..dc00e5d01
--- /dev/null
+++ b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OSItemMixIn.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.persistence.opensearch;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
+/**
+ * This mixin is used in OSCustomObjectMapper to avoid the version parameter to be registered in ES
+ * @author dgaillard
+ */
+public abstract class OSItemMixIn {
+
+ public OSItemMixIn() { }
+
+ @JsonIgnore abstract Long getVersion();
+}
diff --git a/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OpenSearchPersistenceServiceImpl.java b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OpenSearchPersistenceServiceImpl.java
new file mode 100644
index 000000000..04b79493f
--- /dev/null
+++ b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OpenSearchPersistenceServiceImpl.java
@@ -0,0 +1,2719 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.unomi.persistence.opensearch;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.json.*;
+import jakarta.json.stream.JsonParser;
+import org.apache.commons.io.FilenameUtils;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hc.client5.http.auth.AuthScope;
+import org.apache.hc.client5.http.auth.UsernamePasswordCredentials;
+import org.apache.hc.client5.http.impl.auth.BasicCredentialsProvider;
+import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager;
+import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder;
+import org.apache.hc.client5.http.ssl.ClientTlsStrategyBuilder;
+import org.apache.hc.client5.http.ssl.NoopHostnameVerifier;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.nio.ssl.TlsStrategy;
+import org.apache.hc.core5.ssl.SSLContextBuilder;
+import org.apache.unomi.api.*;
+import org.apache.unomi.api.conditions.Condition;
+import org.apache.unomi.api.query.DateRange;
+import org.apache.unomi.api.query.IpRange;
+import org.apache.unomi.api.query.NumericRange;
+import org.apache.unomi.metrics.MetricAdapter;
+import org.apache.unomi.metrics.MetricsService;
+import org.apache.unomi.persistence.spi.PersistenceService;
+import org.apache.unomi.persistence.spi.aggregate.DateRangeAggregate;
+import org.apache.unomi.persistence.spi.aggregate.IpRangeAggregate;
+import org.apache.unomi.persistence.spi.aggregate.*;
+import org.apache.unomi.persistence.spi.conditions.ConditionContextHelper;
+import org.apache.unomi.persistence.spi.conditions.evaluator.ConditionEvaluatorDispatcher;
+import org.opensearch.client.*;
+import org.opensearch.client.json.JsonData;
+import org.opensearch.client.json.JsonpMapper;
+import org.opensearch.client.json.jackson.JacksonJsonpMapper;
+import org.opensearch.client.opensearch.OpenSearchClient;
+import org.opensearch.client.opensearch._types.*;
+import org.opensearch.client.opensearch._types.aggregations.*;
+import org.opensearch.client.opensearch._types.mapping.TypeMapping;
+import org.opensearch.client.opensearch._types.query_dsl.Query;
+import org.opensearch.client.opensearch.cluster.HealthRequest;
+import org.opensearch.client.opensearch.cluster.HealthResponse;
+import org.opensearch.client.opensearch.core.*;
+import org.opensearch.client.opensearch.core.bulk.BulkOperation;
+import org.opensearch.client.opensearch.core.bulk.UpdateOperation;
+import org.opensearch.client.opensearch.core.search.Hit;
+import org.opensearch.client.opensearch.core.search.HitsMetadata;
+import org.opensearch.client.opensearch.core.search.TotalHits;
+import org.opensearch.client.opensearch.core.search.TotalHitsRelation;
+import org.opensearch.client.opensearch.generic.Requests;
+import org.opensearch.client.opensearch.indices.*;
+import org.opensearch.client.opensearch.indices.get_alias.IndexAliases;
+import org.opensearch.client.opensearch.tasks.GetTasksResponse;
+import org.opensearch.client.transport.OpenSearchTransport;
+import org.opensearch.client.transport.endpoints.BooleanResponse;
+import org.opensearch.client.transport.httpclient5.ApacheHttpClient5TransportBuilder;
+import org.osgi.framework.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.util.*;
+import java.util.stream.Collectors;
+
+@SuppressWarnings("rawtypes")
+public class OpenSearchPersistenceServiceImpl implements PersistenceService, SynchronousBundleListener {
+
+ public static final String SEQ_NO = "seq_no";
+ public static final String PRIMARY_TERM = "primary_term";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(OpenSearchPersistenceServiceImpl.class.getName());
+ private static final String ROLLOVER_LIFECYCLE_NAME = "unomi-rollover-policy";
+
+ private boolean throwExceptions = false;
+
+ private OpenSearchClient client;
+
+ private final List openSearchAddressList = new ArrayList<>();
+ private String clusterName;
+ private String indexPrefix;
+ private String numberOfShards;
+ private String numberOfReplicas;
+ private String indexMappingTotalFieldsLimit;
+ private String indexMaxDocValueFieldsSearch;
+ private String[] fatalIllegalStateErrors;
+ private BundleContext bundleContext;
+ private final Map mappings = new HashMap<>();
+ private ConditionEvaluatorDispatcher conditionEvaluatorDispatcher;
+ private ConditionOSQueryBuilderDispatcher conditionOSQueryBuilderDispatcher;
+ private Map routingByType;
+
+ private Integer defaultQueryLimit = 10;
+ private Integer removeByQueryTimeoutInMinutes = 10;
+ private Integer taskWaitingTimeout = 3600000;
+ private Integer taskWaitingPollingInterval = 1000;
+
+ // Rollover configuration
+ private String sessionLatestIndex;
+ private List rolloverIndices;
+ private String rolloverMaxSize;
+ private String rolloverMaxAge;
+ private String rolloverMaxDocs;
+ private String rolloverIndexNumberOfShards;
+ private String rolloverIndexNumberOfReplicas;
+ private String rolloverIndexMappingTotalFieldsLimit;
+ private String rolloverIndexMaxDocValueFieldsSearch;
+
+ private String minimalOpenSearchVersion = "2.1.0";
+ private String maximalOpenSearchVersion = "4.0.0";
+
+ // authentication props
+ private String username;
+ private String password;
+ private boolean sslEnable = false;
+ private boolean sslTrustAllCertificates = false;
+
+ // AWS configuration
+ private boolean awsEnabled = false;
+ private String awsRegion;
+ private String awsServiceName = "opensearch"; // or "es" for Elasticsearch service
+
+ private int aggregateQueryBucketSize = 5000;
+
+ private MetricsService metricsService;
+ private boolean useBatchingForSave = false;
+ private boolean useBatchingForUpdate = true;
+ private String logLevelRestClient = "ERROR";
+ private boolean alwaysOverwrite = true;
+ private boolean aggQueryThrowOnMissingDocs = false;
+ private Integer aggQueryMaxResponseSizeHttp = null;
+ private Integer clientSocketTimeout = null;
+ private Map itemTypeToRefreshPolicy = new HashMap<>();
+
+ private final Map>> knownMappings = new HashMap<>();
+
+ private static final Map itemTypeIndexNameMap = new HashMap<>();
+ private static final Collection systemItems = Arrays.asList("actionType", "campaign", "campaignevent", "goal",
+ "userList", "propertyType", "scope", "conditionType", "rule", "scoring", "segment", "groovyAction", "topic",
+ "patch", "jsonSchema", "importConfig", "exportConfig", "rulestats");
+ static {
+ for (String systemItem : systemItems) {
+ itemTypeIndexNameMap.put(systemItem, "systemItems");
+ }
+
+ itemTypeIndexNameMap.put("profile", "profile");
+ itemTypeIndexNameMap.put("persona", "profile");
+ }
+
+ private final JsonpMapper jsonpMapper = new JacksonJsonpMapper();
+
+ private String minimalClusterState = "GREEN"; // Add this as a class field
+ private int clusterHealthTimeout = 30; // timeout in seconds
+ private int clusterHealthRetries = 3;
+
+ public void setBundleContext(BundleContext bundleContext) {
+ this.bundleContext = bundleContext;
+ }
+
+ public void setClusterName(String clusterName) {
+ this.clusterName = clusterName;
+ }
+
+ public void setOpenSearchAddresses(String openSearchAddresses) {
+ String[] openSearchAddressesArray = openSearchAddresses.split(",");
+ openSearchAddressList.clear();
+ for (String openSearchAddress : openSearchAddressesArray) {
+ openSearchAddressList.add(openSearchAddress.trim());
+ }
+ }
+
+ public void setItemTypeToRefreshPolicy(String itemTypeToRefreshPolicy) throws IOException {
+ if (!itemTypeToRefreshPolicy.isEmpty()) {
+ this.itemTypeToRefreshPolicy = new ObjectMapper().readValue(itemTypeToRefreshPolicy,
+ new TypeReference>() {
+ });
+ }
+ }
+
+ public void setFatalIllegalStateErrors(String fatalIllegalStateErrors) {
+ this.fatalIllegalStateErrors = Arrays.stream(fatalIllegalStateErrors.split(","))
+ .map(String::trim).filter(i -> !i.isEmpty()).toArray(String[]::new);
+ }
+
+ public void setAggQueryMaxResponseSizeHttp(String aggQueryMaxResponseSizeHttp) {
+ if (StringUtils.isNumeric(aggQueryMaxResponseSizeHttp)) {
+ this.aggQueryMaxResponseSizeHttp = Integer.parseInt(aggQueryMaxResponseSizeHttp);
+ }
+ }
+
+ public void setIndexPrefix(String indexPrefix) {
+ this.indexPrefix = indexPrefix;
+ }
+
+
+
+ public void setNumberOfShards(String numberOfShards) {
+ this.numberOfShards = numberOfShards;
+ }
+
+ public void setNumberOfReplicas(String numberOfReplicas) {
+ this.numberOfReplicas = numberOfReplicas;
+ }
+
+ public void setIndexMappingTotalFieldsLimit(String indexMappingTotalFieldsLimit) {
+ this.indexMappingTotalFieldsLimit = indexMappingTotalFieldsLimit;
+ }
+
+ public void setIndexMaxDocValueFieldsSearch(String indexMaxDocValueFieldsSearch) {
+ this.indexMaxDocValueFieldsSearch = indexMaxDocValueFieldsSearch;
+ }
+
+ public void setDefaultQueryLimit(Integer defaultQueryLimit) {
+ this.defaultQueryLimit = defaultQueryLimit;
+ }
+
+ public void setRoutingByType(Map routingByType) {
+ this.routingByType = routingByType;
+ }
+
+ public void setConditionEvaluatorDispatcher(ConditionEvaluatorDispatcher conditionEvaluatorDispatcher) {
+ this.conditionEvaluatorDispatcher = conditionEvaluatorDispatcher;
+ }
+
+ public void setConditionOSQueryBuilderDispatcher(ConditionOSQueryBuilderDispatcher conditionOSQueryBuilderDispatcher) {
+ this.conditionOSQueryBuilderDispatcher = conditionOSQueryBuilderDispatcher;
+ }
+
+ public void setRolloverIndices(String rolloverIndices) {
+ this.rolloverIndices = StringUtils.isNotEmpty(rolloverIndices) ? Arrays.asList(rolloverIndices.split(",").clone()) : null;
+ }
+
+ public void setRolloverMaxSize(String rolloverMaxSize) {
+ this.rolloverMaxSize = rolloverMaxSize;
+ }
+
+ public void setRolloverMaxAge(String rolloverMaxAge) {
+ this.rolloverMaxAge = rolloverMaxAge;
+ }
+
+ public void setRolloverMaxDocs(String rolloverMaxDocs) {
+ this.rolloverMaxDocs = rolloverMaxDocs;
+ }
+
+ public void setRolloverIndexNumberOfShards(String rolloverIndexNumberOfShards) {
+ this.rolloverIndexNumberOfShards = rolloverIndexNumberOfShards;
+ }
+
+ public void setRolloverIndexNumberOfReplicas(String rolloverIndexNumberOfReplicas) {
+ this.rolloverIndexNumberOfReplicas = rolloverIndexNumberOfReplicas;
+ }
+
+ public void setRolloverIndexMappingTotalFieldsLimit(String rolloverIndexMappingTotalFieldsLimit) {
+ this.rolloverIndexMappingTotalFieldsLimit = rolloverIndexMappingTotalFieldsLimit;
+ }
+
+ public void setRolloverIndexMaxDocValueFieldsSearch(String rolloverIndexMaxDocValueFieldsSearch) {
+ this.rolloverIndexMaxDocValueFieldsSearch = rolloverIndexMaxDocValueFieldsSearch;
+ }
+
+ public void setMinimalOpenSearchVersion(String minimalOpenSearchVersion) {
+ this.minimalOpenSearchVersion = minimalOpenSearchVersion;
+ }
+
+ public void setMaximalOpenSearchVersion(String maximalOpenSearchVersion) {
+ this.maximalOpenSearchVersion = maximalOpenSearchVersion;
+ }
+
+ public void setAggregateQueryBucketSize(int aggregateQueryBucketSize) {
+ this.aggregateQueryBucketSize = aggregateQueryBucketSize;
+ }
+
+ public void setClientSocketTimeout(String clientSocketTimeout) {
+ if (StringUtils.isNumeric(clientSocketTimeout)) {
+ this.clientSocketTimeout = Integer.parseInt(clientSocketTimeout);
+ }
+ }
+
+ public void setMetricsService(MetricsService metricsService) {
+ this.metricsService = metricsService;
+ }
+
+ public void setUseBatchingForSave(boolean useBatchingForSave) {
+ this.useBatchingForSave = useBatchingForSave;
+ }
+
+ public void setUseBatchingForUpdate(boolean useBatchingForUpdate) {
+ this.useBatchingForUpdate = useBatchingForUpdate;
+ }
+
+ public void setUsername(String username) {
+ this.username = username;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public void setSslEnable(boolean sslEnable) {
+ this.sslEnable = sslEnable;
+ }
+
+ public void setSslTrustAllCertificates(boolean sslTrustAllCertificates) {
+ this.sslTrustAllCertificates = sslTrustAllCertificates;
+ }
+
+
+ public void setAggQueryThrowOnMissingDocs(boolean aggQueryThrowOnMissingDocs) {
+ this.aggQueryThrowOnMissingDocs = aggQueryThrowOnMissingDocs;
+ }
+
+ public void setThrowExceptions(boolean throwExceptions) {
+ this.throwExceptions = throwExceptions;
+ }
+
+ public void setAlwaysOverwrite(boolean alwaysOverwrite) {
+ this.alwaysOverwrite = alwaysOverwrite;
+ }
+
+ public void setLogLevelRestClient(String logLevelRestClient) {
+ this.logLevelRestClient = logLevelRestClient;
+ }
+
+ public void setTaskWaitingTimeout(String taskWaitingTimeout) {
+ if (StringUtils.isNumeric(taskWaitingTimeout)) {
+ this.taskWaitingTimeout = Integer.parseInt(taskWaitingTimeout);
+ }
+ }
+
+ public void setTaskWaitingPollingInterval(String taskWaitingPollingInterval) {
+ if (StringUtils.isNumeric(taskWaitingPollingInterval)) {
+ this.taskWaitingPollingInterval = Integer.parseInt(taskWaitingPollingInterval);
+ }
+ }
+
+ public void setMinimalClusterState(String minimalClusterState) {
+ if ("GREEN".equalsIgnoreCase(minimalClusterState) || "YELLOW".equalsIgnoreCase(minimalClusterState)) {
+ this.minimalClusterState = minimalClusterState.toUpperCase();
+ } else {
+ LOGGER.warn("Invalid minimal cluster state: {}. Using default: GREEN", minimalClusterState);
+ }
+ }
+
+ public String getName() {
+ return "opensearch";
+ }
+
+ public void start() throws Exception {
+
+ // on startup
+ new InClassLoaderExecute<>(null, null, this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) {
+ public Object execute(Object... args) throws Exception {
+
+ buildClient();
+
+ InfoResponse response = client.info();
+ OpenSearchVersionInfo version = response.version();
+ Version clusterVersion = Version.parseVersion(version.number());
+ Version minimalVersion = Version.parseVersion(minimalOpenSearchVersion);
+ Version maximalVersion = Version.parseVersion(maximalOpenSearchVersion);
+ if (clusterVersion.compareTo(minimalVersion) < 0 ||
+ clusterVersion.equals(maximalVersion) ||
+ clusterVersion.compareTo(maximalVersion) > 0) {
+ throw new Exception("OpenSearch version is not within [" + minimalVersion + "," + maximalVersion + "), aborting startup !");
+ }
+
+ waitForClusterHealth();
+
+ registerRolloverLifecyclePolicy();
+
+ loadPredefinedMappings(bundleContext, false);
+ loadPainlessScripts(bundleContext);
+
+ // load predefined mappings and condition dispatchers of any bundles that were started before this one.
+ for (Bundle existingBundle : bundleContext.getBundles()) {
+ if (existingBundle.getBundleContext() != null) {
+ loadPredefinedMappings(existingBundle.getBundleContext(), false);
+ loadPainlessScripts(existingBundle.getBundleContext());
+ }
+ }
+
+ // Wait for minimal cluster state
+ LOGGER.info("Waiting for {} cluster status...", minimalClusterState);
+ client.cluster().health(new HealthRequest.Builder().waitForStatus(getHealthStatus(minimalClusterState)).build());
+ LOGGER.info("Cluster status is {}", minimalClusterState);
+
+ // We keep in memory the latest available session index to be able to load session using direct GET access on OpenSearch
+ if (isItemTypeRollingOver(Session.ITEM_TYPE)) {
+ LOGGER.info("Sessions are using rollover indices, loading latest session index available ...");
+ GetAliasResponse sessionAliasResponse = client.indices().getAlias(new GetAliasRequest.Builder().index(getIndex(Session.ITEM_TYPE)).build());
+ Map aliases = sessionAliasResponse.result();
+ if (!aliases.isEmpty()) {
+ sessionLatestIndex = new TreeSet<>(aliases.keySet()).last();
+ LOGGER.info("Latest available session index found is: {}", sessionLatestIndex);
+ } else {
+ throw new IllegalStateException("No index found for sessions");
+ }
+ }
+
+ return true;
+ }
+ }.executeInClassLoader();
+
+ bundleContext.addBundleListener(this);
+
+ LOGGER.info(this.getClass().getName() + " service started successfully.");
+ }
+
+ private void buildClient() throws NoSuchFieldException, IllegalAccessException, NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
+
+ List hosts = new ArrayList<>();
+ for (String openSearchAddress : openSearchAddressList) {
+ String[] openSearchAddressParts = openSearchAddress.split(":");
+ String openSearchHostName = openSearchAddressParts[0];
+ int openSearchPort = Integer.parseInt(openSearchAddressParts[1]);
+
+ hosts.add(new HttpHost(sslEnable ? "https" : "http", openSearchHostName, openSearchPort));
+ }
+
+ // Use Apache HttpClient 5 transport with proper configuration
+ ApacheHttpClient5TransportBuilder transportBuilder = ApacheHttpClient5TransportBuilder.builder(
+ hosts.toArray(new HttpHost[0])
+ );
+
+ // Configure JSON mapper
+ transportBuilder.setMapper(new JacksonJsonpMapper(OSCustomObjectMapper.getObjectMapper()));
+
+ // Configure socket timeout if specified
+ if (clientSocketTimeout != null) {
+ transportBuilder.setRequestConfigCallback(requestConfigBuilder -> {
+ requestConfigBuilder.setResponseTimeout(clientSocketTimeout, java.util.concurrent.TimeUnit.MILLISECONDS);
+ return requestConfigBuilder;
+ });
+ }
+
+ // Configure SSL and authentication
+ transportBuilder.setHttpClientConfigCallback(httpClientBuilder -> {
+ if (sslTrustAllCertificates) {
+ try {
+ TlsStrategy tlsStrategy = ClientTlsStrategyBuilder.create()
+ .setSslContext(SSLContextBuilder.create()
+ .loadTrustMaterial((chain, authType) -> true) // Trust all
+ .build())
+ .setHostnameVerifier(NoopHostnameVerifier.INSTANCE)
+ .build();
+
+ PoolingAsyncClientConnectionManager connectionManager =
+ PoolingAsyncClientConnectionManagerBuilder.create()
+ .setTlsStrategy(tlsStrategy)
+ .build();
+
+ httpClientBuilder.setConnectionManager(connectionManager);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to create SSL context", e);
+ }
+ }
+
+ if (StringUtils.isNotBlank(username)) {
+ final BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+ credentialsProvider.setCredentials(new AuthScope(null, -1),
+ new UsernamePasswordCredentials(username, password.toCharArray()));
+
+ httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
+ }
+
+ return httpClientBuilder;
+ });
+
+ OpenSearchTransport transport = transportBuilder.build();
+ client = new OpenSearchClient(transport);
+
+ LOGGER.info("Connecting to OpenSearch persistence backend using cluster name " + clusterName + " and index prefix " + indexPrefix + "...");
+ }
+
+
+ public void stop() {
+
+ new InClassLoaderExecute<>(null, null, this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) {
+ protected Object execute(Object... args) throws IOException {
+ LOGGER.info("Closing OpenSearch persistence backend...");
+ if (client != null) {
+ client._transport().close();
+ client = null;
+ }
+ return null;
+ }
+ }.catchingExecuteInClassLoader(true);
+
+ bundleContext.removeBundleListener(this);
+ }
+
+ public void bindConditionOSQueryBuilder(ServiceReference conditionOSQueryBuilderServiceReference) {
+ ConditionOSQueryBuilder conditionOSQueryBuilder = bundleContext.getService(conditionOSQueryBuilderServiceReference);
+ conditionOSQueryBuilderDispatcher.addQueryBuilder(conditionOSQueryBuilderServiceReference.getProperty("queryBuilderId").toString(), conditionOSQueryBuilder);
+ }
+
+ public void unbindConditionOSQueryBuilder(ServiceReference conditionOSQueryBuilderServiceReference) {
+ if (conditionOSQueryBuilderServiceReference == null) {
+ return;
+ }
+ conditionOSQueryBuilderDispatcher.removeQueryBuilder(conditionOSQueryBuilderServiceReference.getProperty("queryBuilderId").toString());
+ }
+
+ @Override
+ public void bundleChanged(BundleEvent event) {
+ if (event.getType() == BundleEvent.STARTING) {
+ loadPredefinedMappings(event.getBundle().getBundleContext(), true);
+ loadPainlessScripts(event.getBundle().getBundleContext());
+ }
+ }
+
+ private void loadPredefinedMappings(BundleContext bundleContext, boolean forceUpdateMapping) {
+ Enumeration predefinedMappings = bundleContext.getBundle().findEntries("META-INF/cxs/mappings", "*.json", true);
+ if (predefinedMappings == null) {
+ return;
+ }
+ while (predefinedMappings.hasMoreElements()) {
+ URL predefinedMappingURL = predefinedMappings.nextElement();
+ LOGGER.info("Found mapping at " + predefinedMappingURL + ", loading... ");
+ try {
+ final String path = predefinedMappingURL.getPath();
+ String name = path.substring(path.lastIndexOf('/') + 1, path.lastIndexOf('.'));
+ String mappingSource = loadMappingFile(predefinedMappingURL);
+
+ mappings.put(name, mappingSource);
+
+ if (!createIndex(name)) {
+ LOGGER.info("Found index for type {}", name);
+ if (forceUpdateMapping) {
+ LOGGER.info("Updating mapping for {}", name);
+ createMapping(name, mappingSource);
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error while loading mapping definition " + predefinedMappingURL, e);
+ }
+ }
+ }
+
+ private TypeMapping getTypeMapping(String mappingSource) {
+ JsonpMapper mapper = client._transport().jsonpMapper();
+ JsonParser parser = mapper
+ .jsonProvider()
+ .createParser(new StringReader(mappingSource));
+ return TypeMapping._DESERIALIZER.deserialize(parser, mapper);
+ }
+
+ private void loadPainlessScripts(BundleContext bundleContext) {
+ Enumeration scriptsURL = bundleContext.getBundle().findEntries("META-INF/cxs/painless", "*.painless", true);
+ if (scriptsURL == null) {
+ return;
+ }
+
+ Map scriptsById = new HashMap<>();
+ while (scriptsURL.hasMoreElements()) {
+ URL scriptURL = scriptsURL.nextElement();
+ LOGGER.info("Found painless script at " + scriptURL + ", loading... ");
+ try (InputStream in = scriptURL.openStream()) {
+ String script = IOUtils.toString(in, StandardCharsets.UTF_8);
+ String scriptId = FilenameUtils.getBaseName(scriptURL.getPath());
+ scriptsById.put(scriptId, script);
+ } catch (Exception e) {
+ LOGGER.error("Error while loading painless script " + scriptURL, e);
+ }
+
+ }
+
+ storeScripts(scriptsById);
+ }
+
+ private String loadMappingFile(URL predefinedMappingURL) throws IOException {
+ BufferedReader reader = new BufferedReader(new InputStreamReader(predefinedMappingURL.openStream()));
+
+ StringBuilder content = new StringBuilder();
+ String l;
+ while ((l = reader.readLine()) != null) {
+ content.append(l);
+ }
+ return content.toString();
+ }
+
+ @Override
+ public List getAllItems(final Class clazz) {
+ return getAllItems(clazz, 0, -1, null).getList();
+ }
+
+ @Override
+ public long getAllItemsCount(String itemType) {
+ return queryCount(Query.of(q -> q.matchAll(t -> t)), itemType);
+ }
+
+ @Override
+ public PartialList getAllItems(final Class clazz, int offset, int size, String sortBy) {
+ return getAllItems(clazz, offset, size, sortBy, null);
+ }
+
+ @Override
+ public PartialList getAllItems(final Class clazz, int offset, int size, String sortBy, String scrollTimeValidity) {
+ long startTime = System.currentTimeMillis();
+ try {
+ return query(Query.of(q -> q.matchAll(t -> t)), sortBy, clazz, offset, size, null, scrollTimeValidity);
+ } finally {
+ if (metricsService != null && metricsService.isActivated()) {
+ metricsService.updateTimer(this.getClass().getName() + ".getAllItems", startTime);
+ }
+ }
+ }
+
+ @Override
+ public T load(final String itemId, final Class clazz) {
+ return load(itemId, clazz, null);
+ }
+
+ @Override
+ @Deprecated
+ public T load(final String itemId, final Date dateHint, final Class clazz) {
+ return load(itemId, clazz, null);
+ }
+
+ @Override
+ @Deprecated
+ public CustomItem loadCustomItem(final String itemId, final Date dateHint, String customItemType) {
+ return load(itemId, CustomItem.class, customItemType);
+ }
+
+ @Override
+ public CustomItem loadCustomItem(final String itemId, String customItemType) {
+ return load(itemId, CustomItem.class, customItemType);
+ }
+
+ private T load(final String itemId, final Class clazz, final String customItemType) {
+ if (StringUtils.isEmpty(itemId)) {
+ return null;
+ }
+
+ return new InClassLoaderExecute(metricsService, this.getClass().getName() + ".loadItem", this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) {
+ protected T execute(Object... args) throws Exception {
+ try {
+ String itemType = Item.getItemType(clazz);
+ if (customItemType != null) {
+ itemType = customItemType;
+ }
+ String documentId = getDocumentIDForItemType(itemId, itemType);
+
+ boolean sessionSpecialDirectAccess = sessionLatestIndex != null && Session.ITEM_TYPE.equals(itemType) ;
+ if (!sessionSpecialDirectAccess && isItemTypeRollingOver(itemType)) {
+ return new MetricAdapter(metricsService, ".loadItemWithQuery") {
+ @Override
+ public T execute(Object... args) throws Exception {
+ if (customItemType == null) {
+ PartialList r = query(Query.of(q -> q.ids(i -> i.values(documentId))), null, clazz, 0, 1, null, null);
+ if (r.size() > 0) {
+ return r.get(0);
+ }
+ } else {
+ PartialList r = query(Query.of(q -> q.ids(i -> i.values(documentId))), null, customItemType, 0, 1, null, null);
+ if (r.size() > 0) {
+ return (T) r.get(0);
+ }
+ }
+ return null;
+ }
+ }.execute();
+ } else {
+ // Special handling for session we check the latest available index directly to speed up session loading
+ GetRequest.Builder getRequest = new GetRequest.Builder().index(sessionSpecialDirectAccess ? sessionLatestIndex : getIndex(itemType)).id(documentId);
+ GetResponse response = client.get(getRequest.build(), clazz);
+ if (response.found()) {
+ T value = response.source();
+ setMetadata(value, response.id(), response.version(), response.seqNo(), response.primaryTerm(), response.index());
+ return value;
+ } else {
+ return null;
+ }
+ }
+ } catch (OpenSearchException ose) {
+ if (ose.status() == 404) {
+ // this can happen if we are just testing the existence of the item, it is not always an error.
+ return null;
+ }
+ if ("IndexNotFound".equals(ose.error().type())) {
+ // this can happen if we are just testing the existence of the item, it is not always an error.
+ return null;
+ }
+ throw new Exception("Error loading itemType=" + clazz.getName() + " customItemType=" + customItemType + " itemId=" + itemId, ose);
+ } catch (Exception ex) {
+ throw new Exception("Error loading itemType=" + clazz.getName() + " customItemType=" + customItemType + " itemId=" + itemId, ex);
+ }
+ }
+ }.catchingExecuteInClassLoader(true);
+
+ }
+
+ private void setMetadata(Item item, String itemId, long version, long seqNo, long primaryTerm, String index) {
+ if (!systemItems.contains(item.getItemType()) && item.getItemId() == null) {
+ item.setItemId(itemId);
+ }
+ item.setVersion(version);
+ item.setSystemMetadata(SEQ_NO, seqNo);
+ item.setSystemMetadata(PRIMARY_TERM, primaryTerm);
+ item.setSystemMetadata("index", index);
+ }
+
+ @Override
+ public boolean isConsistent(Item item) {
+ return getRefreshPolicy(item.getItemType()) != Refresh.False;
+ }
+
+ @Override
+ public boolean save(final Item item) {
+ return save(item, useBatchingForSave, alwaysOverwrite);
+ }
+
+ @Override
+ public boolean save(final Item item, final boolean useBatching) {
+ return save(item, useBatching, alwaysOverwrite);
+ }
+
+ @Override
+ public boolean save(final Item item, final Boolean useBatchingOption, final Boolean alwaysOverwriteOption) {
+ final boolean useBatching = useBatchingOption == null ? this.useBatchingForSave : useBatchingOption;
+ final boolean alwaysOverwrite = alwaysOverwriteOption == null ? this.alwaysOverwrite : alwaysOverwriteOption;
+
+ Boolean result = new InClassLoaderExecute(metricsService, this.getClass().getName() + ".saveItem", this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) {
+ protected Boolean execute(Object... args) throws Exception {
+ try {
+ String itemType = item.getItemType();
+ if (item instanceof CustomItem) {
+ itemType = ((CustomItem) item).getCustomItemType();
+ }
+ String documentId = getDocumentIDForItemType(item.getItemId(), itemType);
+ String index = item.getSystemMetadata("index") != null ? (String) item.getSystemMetadata("index") : getIndex(itemType);
+
+ IndexRequest.Builder- indexRequest = new IndexRequest.Builder
- ().index(index);
+ indexRequest.id(documentId);
+ indexRequest.document(item);
+
+ if (!alwaysOverwrite) {
+ Long seqNo = (Long) item.getSystemMetadata(SEQ_NO);
+ Long primaryTerm = (Long) item.getSystemMetadata(PRIMARY_TERM);
+
+ if (seqNo != null && primaryTerm != null) {
+ indexRequest.ifSeqNo(seqNo);
+ indexRequest.ifPrimaryTerm(primaryTerm);
+ } else {
+ indexRequest.opType(OpType.Create);
+ }
+ }
+
+ if (routingByType.containsKey(itemType)) {
+ indexRequest.routing(routingByType.get(itemType));
+ }
+
+ try {
+ indexRequest.refresh(getRefreshPolicy(itemType));
+ IndexResponse response = client.index(indexRequest.build());
+ String responseIndex = response.index();
+ String itemId = response.id();
+ setMetadata(item, itemId, response.version(), response.seqNo(), response.primaryTerm(), responseIndex);
+
+ // Special handling for session, in case of new session we check that a rollover happen or not to update the latest available index
+ if (Session.ITEM_TYPE.equals(itemType) &&
+ sessionLatestIndex != null &&
+ response.result().equals(Result.Created) &&
+ !responseIndex.equals(sessionLatestIndex)) {
+ sessionLatestIndex = responseIndex;
+ }
+ logMetadataItemOperation("saved", item);
+ } catch (OpenSearchException ose) {
+ LOGGER.error("Could not find index {}, could not register item type {} with id {} ", index, itemType, item.getItemId(), ose);
+ return false;
+ }
+ return true;
+ } catch (IOException e) {
+ throw new Exception("Error saving item " + item, e);
+ }
+ }
+ }.catchingExecuteInClassLoader(true);
+ return Objects.requireNonNullElse(result, false);
+ }
+
+ @Override
+ public boolean update(final Item item, final Date dateHint, final Class clazz, final String propertyName, final Object propertyValue) {
+ return update(item, clazz, propertyName, propertyValue);
+ }
+
+ @Override
+ public boolean update(final Item item, final Date dateHint, final Class clazz, final Map source) {
+ return update(item, clazz, source);
+ }
+
+ @Override
+ public boolean update(final Item item, final Date dateHint, final Class clazz, final Map source, final boolean alwaysOverwrite) {
+ return update(item, clazz, source, alwaysOverwrite);
+ }
+
+ @Override
+ public boolean update(final Item item, final Class clazz, final String propertyName, final Object propertyValue) {
+ return update(item, clazz, Collections.singletonMap(propertyName, propertyValue), alwaysOverwrite);
+ }
+
+
+ @Override
+ public boolean update(final Item item, final Class clazz, final Map source) {
+ return update(item, clazz, source, alwaysOverwrite);
+ }
+
+ @Override
+ public boolean update(final Item item, final Class clazz, final Map source, final boolean alwaysOverwrite) {
+ Boolean result = new InClassLoaderExecute(metricsService, this.getClass().getName() + ".updateItem", this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) {
+ protected Boolean execute(Object... args) throws Exception {
+ try {
+ UpdateRequest updateRequest = createUpdateRequest(clazz, item, source, alwaysOverwrite);
+
+ UpdateResponse response = client.update(updateRequest, Item.class);
+ if (response.result().equals(Result.NoOp)) {
+ LOGGER.warn("Update of item {} with source {} returned NoOp", item.getItemId(), source);
+ }
+ setMetadata(item, response.id(), response.version(), response.seqNo(), response.primaryTerm(), response.index());
+ logMetadataItemOperation("updated", item);
+ return true;
+ } catch (OpenSearchException ose) {
+ throw new Exception("No index found for itemType=" + clazz.getName() + "itemId=" + item.getItemId(), ose);
+ }
+ }
+ }.catchingExecuteInClassLoader(true);
+ return Objects.requireNonNullElse(result, false);
+ }
+
+ private UpdateRequest createUpdateRequest(Class clazz, Item item, Map source, boolean alwaysOverwrite) {
+ String itemType = Item.getItemType(clazz);
+ String documentId = getDocumentIDForItemType(item.getItemId(), itemType);
+ String index = item.getSystemMetadata("index") != null ? (String) item.getSystemMetadata("index") : getIndex(itemType);
+
+ UpdateRequest.Builder updateRequest = new UpdateRequest.Builder().index(index).id(documentId);
+ updateRequest.doc(source);
+
+ if (!alwaysOverwrite) {
+ Long seqNo = (Long) item.getSystemMetadata(SEQ_NO);
+ Long primaryTerm = (Long) item.getSystemMetadata(PRIMARY_TERM);
+
+ if (seqNo != null && primaryTerm != null) {
+ updateRequest.ifSeqNo(seqNo);
+ updateRequest.ifPrimaryTerm(primaryTerm);
+ }
+ }
+ return updateRequest.build();
+ }
+
+ private UpdateOperation createUpdateOperation(Class clazz, Item item, Map source, boolean alwaysOverwrite) {
+ String itemType = Item.getItemType(clazz);
+ String documentId = getDocumentIDForItemType(item.getItemId(), itemType);
+ String index = item.getSystemMetadata("index") != null ? (String) item.getSystemMetadata("index") : getIndex(itemType);
+
+ UpdateOperation.Builder updateOperation = new UpdateOperation.Builder