| /* |
| * 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.services.impl.segments; |
| |
| import com.fasterxml.jackson.core.JsonProcessingException; |
| import net.jodah.failsafe.Failsafe; |
| import net.jodah.failsafe.RetryPolicy; |
| import org.apache.commons.io.FilenameUtils; |
| import org.apache.commons.io.IOUtils; |
| import org.apache.commons.lang3.CharEncoding; |
| import org.apache.unomi.api.Event; |
| import org.apache.unomi.api.Item; |
| import org.apache.unomi.api.Metadata; |
| import org.apache.unomi.api.PartialList; |
| import org.apache.unomi.api.Profile; |
| import org.apache.unomi.api.actions.Action; |
| import org.apache.unomi.api.conditions.Condition; |
| import org.apache.unomi.api.conditions.ConditionType; |
| import org.apache.unomi.api.query.Query; |
| import org.apache.unomi.api.rules.Rule; |
| import org.apache.unomi.api.segments.*; |
| import org.apache.unomi.api.services.EventService; |
| import org.apache.unomi.api.services.RulesService; |
| import org.apache.unomi.api.services.SchedulerService; |
| import org.apache.unomi.api.services.SegmentService; |
| import org.apache.unomi.persistence.spi.CustomObjectMapper; |
| import org.apache.unomi.persistence.spi.aggregate.TermsAggregate; |
| import org.apache.unomi.services.impl.AbstractServiceImpl; |
| import org.apache.unomi.services.impl.scheduler.SchedulerServiceImpl; |
| import org.apache.unomi.api.utils.ParserHelper; |
| import org.apache.unomi.api.exceptions.BadSegmentConditionException; |
| import org.osgi.framework.Bundle; |
| import org.osgi.framework.BundleContext; |
| import org.osgi.framework.BundleEvent; |
| import org.osgi.framework.SynchronousBundleListener; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.net.URL; |
| import java.nio.charset.StandardCharsets; |
| import java.security.MessageDigest; |
| import java.time.Duration; |
| import java.time.ZoneOffset; |
| import java.time.ZonedDateTime; |
| import java.util.*; |
| import java.util.concurrent.TimeUnit; |
| import java.util.stream.Collectors; |
| |
| public class SegmentServiceImpl extends AbstractServiceImpl implements SegmentService, SynchronousBundleListener { |
| |
| private static final Logger logger = LoggerFactory.getLogger(SegmentServiceImpl.class.getName()); |
| |
| private static final String VALIDATION_PROFILE_ID = "validation-profile-id"; |
| private static final String RESET_SCORING_SCRIPT = "resetScoringPlan"; |
| private static final String EVALUATE_SCORING_ELEMENT_SCRIPT = "evaluateScoringPlanElement"; |
| |
| private BundleContext bundleContext; |
| |
| private EventService eventService; |
| private RulesService rulesService; |
| private SchedulerService schedulerService; |
| |
| private long taskExecutionPeriod = 1; |
| private List<Segment> allSegments; |
| private List<Scoring> allScoring; |
| private int segmentUpdateBatchSize = 1000; |
| private long segmentRefreshInterval = 1000; |
| private int aggregateQueryBucketSize = 5000; |
| private int maxRetriesForUpdateProfileSegment = 0; |
| private long secondsDelayForRetryUpdateProfileSegment = 1; |
| private boolean batchSegmentProfileUpdate = false; |
| private boolean sendProfileUpdateEventForSegmentUpdate = true; |
| private int maximumIdsQueryCount = 5000; |
| private boolean pastEventsDisablePartitions = false; |
| private int dailyDateExprEvaluationHourUtc = 5; |
| |
| public SegmentServiceImpl() { |
| logger.info("Initializing segment service..."); |
| } |
| |
| public void setBundleContext(BundleContext bundleContext) { |
| this.bundleContext = bundleContext; |
| } |
| |
| public void setEventService(EventService eventService) { |
| this.eventService = eventService; |
| } |
| |
| public void setRulesService(RulesService rulesService) { |
| this.rulesService = rulesService; |
| } |
| |
| public void setSchedulerService(SchedulerService schedulerService) { |
| this.schedulerService = schedulerService; |
| } |
| |
| public void setSegmentUpdateBatchSize(int segmentUpdateBatchSize) { |
| this.segmentUpdateBatchSize = segmentUpdateBatchSize; |
| } |
| |
| public void setAggregateQueryBucketSize(int aggregateQueryBucketSize) { |
| this.aggregateQueryBucketSize = aggregateQueryBucketSize; |
| } |
| |
| public void setMaximumIdsQueryCount(int maximumIdsQueryCount) { |
| this.maximumIdsQueryCount = maximumIdsQueryCount; |
| } |
| |
| public void setPastEventsDisablePartitions(boolean pastEventsDisablePartitions) { |
| this.pastEventsDisablePartitions = pastEventsDisablePartitions; |
| } |
| |
| public void setSegmentRefreshInterval(long segmentRefreshInterval) { |
| this.segmentRefreshInterval = segmentRefreshInterval; |
| } |
| |
| public void setMaxRetriesForUpdateProfileSegment(int maxRetriesForUpdateProfileSegment) { |
| this.maxRetriesForUpdateProfileSegment = maxRetriesForUpdateProfileSegment; |
| } |
| |
| public void setSecondsDelayForRetryUpdateProfileSegment(long secondsDelayForRetryUpdateProfileSegment) { |
| this.secondsDelayForRetryUpdateProfileSegment = secondsDelayForRetryUpdateProfileSegment; |
| } |
| |
| public void setBatchSegmentProfileUpdate(boolean batchSegmentProfileUpdate) { |
| this.batchSegmentProfileUpdate = batchSegmentProfileUpdate; |
| } |
| |
| public void setSendProfileUpdateEventForSegmentUpdate(boolean sendProfileUpdateEventForSegmentUpdate) { |
| this.sendProfileUpdateEventForSegmentUpdate = sendProfileUpdateEventForSegmentUpdate; |
| } |
| |
| public void setDailyDateExprEvaluationHourUtc(int dailyDateExprEvaluationHourUtc) { |
| this.dailyDateExprEvaluationHourUtc = dailyDateExprEvaluationHourUtc; |
| } |
| |
| public void postConstruct() throws IOException { |
| logger.debug("postConstruct {" + bundleContext.getBundle() + "}"); |
| loadPredefinedSegments(bundleContext); |
| loadPredefinedScorings(bundleContext); |
| for (Bundle bundle : bundleContext.getBundles()) { |
| if (bundle.getBundleContext() != null && bundle.getBundleId() != bundleContext.getBundle().getBundleId()) { |
| loadPredefinedSegments(bundle.getBundleContext()); |
| loadPredefinedScorings(bundle.getBundleContext()); |
| } |
| } |
| bundleContext.addBundleListener(this); |
| initializeTimer(); |
| loadScripts(); |
| logger.info("Segment service initialized."); |
| } |
| |
| public void preDestroy() { |
| bundleContext.removeBundleListener(this); |
| logger.info("Segment service shutdown."); |
| } |
| |
| private void processBundleStartup(BundleContext bundleContext) { |
| if (bundleContext == null) { |
| return; |
| } |
| loadPredefinedSegments(bundleContext); |
| loadPredefinedScorings(bundleContext); |
| } |
| |
| private void processBundleStop(BundleContext bundleContext) { |
| if (bundleContext == null) { |
| return; |
| } |
| } |
| |
| private void loadPredefinedSegments(BundleContext bundleContext) { |
| Enumeration<URL> predefinedSegmentEntries = bundleContext.getBundle().findEntries("META-INF/cxs/segments", "*.json", true); |
| if (predefinedSegmentEntries == null) { |
| return; |
| } |
| |
| while (predefinedSegmentEntries.hasMoreElements()) { |
| URL predefinedSegmentURL = predefinedSegmentEntries.nextElement(); |
| logger.debug("Found predefined segment at " + predefinedSegmentURL + ", loading... "); |
| |
| try { |
| Segment segment = CustomObjectMapper.getObjectMapper().readValue(predefinedSegmentURL, Segment.class); |
| if (segment.getMetadata().getScope() == null) { |
| segment.getMetadata().setScope("systemscope"); |
| } |
| setSegmentDefinition(segment); |
| logger.info("Predefined segment with id {} registered", segment.getMetadata().getId()); |
| } catch (IOException e) { |
| logger.error("Error while loading segment definition " + predefinedSegmentURL, e); |
| } |
| } |
| } |
| |
| private void loadPredefinedScorings(BundleContext bundleContext) { |
| Enumeration<URL> predefinedScoringEntries = bundleContext.getBundle().findEntries("META-INF/cxs/scoring", "*.json", true); |
| if (predefinedScoringEntries == null) { |
| return; |
| } |
| |
| while (predefinedScoringEntries.hasMoreElements()) { |
| URL predefinedScoringURL = predefinedScoringEntries.nextElement(); |
| logger.debug("Found predefined scoring at " + predefinedScoringURL + ", loading... "); |
| |
| try { |
| Scoring scoring = CustomObjectMapper.getObjectMapper().readValue(predefinedScoringURL, Scoring.class); |
| if (scoring.getMetadata().getScope() == null) { |
| scoring.getMetadata().setScope("systemscope"); |
| } |
| setScoringDefinition(scoring); |
| logger.info("Predefined scoring with id {} registered", scoring.getMetadata().getId()); |
| } catch (IOException e) { |
| logger.error("Error while loading segment definition " + predefinedScoringURL, e); |
| } |
| } |
| } |
| |
| public PartialList<Metadata> getSegmentMetadatas(int offset, int size, String sortBy) { |
| return getMetadatas(offset, size, sortBy, Segment.class); |
| } |
| |
| public PartialList<Metadata> getSegmentMetadatas(String scope, int offset, int size, String sortBy) { |
| PartialList<Segment> segments = persistenceService.query("metadata.scope", scope, sortBy, Segment.class, offset, size); |
| List<Metadata> details = new LinkedList<>(); |
| for (Segment definition : segments.getList()) { |
| details.add(definition.getMetadata()); |
| } |
| return new PartialList<>(details, segments.getOffset(), segments.getPageSize(), segments.getTotalSize(), segments.getTotalSizeRelation()); |
| } |
| |
| public PartialList<Metadata> getSegmentMetadatas(Query query) { |
| return getMetadatas(query, Segment.class); |
| } |
| |
| private List<Segment> getAllSegmentDefinitions() { |
| List<Segment> allItems = persistenceService.getAllItems(Segment.class); |
| for (Segment segment : allItems) { |
| if (segment.getMetadata().isEnabled()) { |
| ParserHelper.resolveConditionType(definitionsService, segment.getCondition(), "segment " + segment.getItemId()); |
| } |
| } |
| return allItems; |
| } |
| |
| public Segment getSegmentDefinition(String segmentId) { |
| Segment definition = persistenceService.load(segmentId, Segment.class); |
| if (definition != null && definition.getMetadata().isEnabled()) { |
| ParserHelper.resolveConditionType(definitionsService, definition.getCondition(), "segment " + segmentId); |
| } |
| return definition; |
| } |
| |
| public void setSegmentDefinition(Segment segment) { |
| if (segment.getMetadata().isEnabled()) { |
| ParserHelper.resolveConditionType(definitionsService, segment.getCondition(), "segment " + segment.getItemId()); |
| if (!persistenceService.isValidCondition(segment.getCondition(), new Profile(VALIDATION_PROFILE_ID))) { |
| throw new BadSegmentConditionException(); |
| } |
| if (!segment.getMetadata().isMissingPlugins()) { |
| updateAutoGeneratedRules(segment.getMetadata(), segment.getCondition()); |
| } |
| } |
| |
| // make sure we update the name and description metadata that might not match, so first we remove the entry from the map |
| persistenceService.save(segment, null, true); |
| updateExistingProfilesForSegment(segment); |
| } |
| |
| private boolean checkSegmentDeletionImpact(Condition condition, String segmentToDeleteId) { |
| if (condition != null) { |
| @SuppressWarnings("unchecked") final List<Condition> subConditions = (List<Condition>) condition.getParameter("subConditions"); |
| if (subConditions != null) { |
| for (Condition subCondition : subConditions) { |
| if (checkSegmentDeletionImpact(subCondition, segmentToDeleteId)) { |
| return true; |
| } |
| } |
| } else if ("profileSegmentCondition".equals(condition.getConditionTypeId())) { |
| @SuppressWarnings("unchecked") final List<String> referencedSegmentIds = (List<String>) condition.getParameter("segments"); |
| |
| if (referencedSegmentIds.indexOf(segmentToDeleteId) >= 0) { |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Return an updated condition that do not contain a condition on the segmentId anymore |
| * it's remove the unnecessary boolean condition (if a condition is the only one of a boolean the boolean will be remove and the subcondition returned) |
| * it's return null when there is no more condition after (if the condition passed was only a segment condition on the segmentId) |
| * |
| * @param condition the condition to update |
| * @param segmentId the segment id to remove in the condition |
| * @return updated condition |
| */ |
| private Condition updateSegmentDependentCondition(Condition condition, String segmentId) { |
| if ("booleanCondition".equals(condition.getConditionTypeId())) { |
| @SuppressWarnings("unchecked") final List<Condition> subConditions = (List<Condition>) condition.getParameter("subConditions"); |
| List<Condition> updatedSubConditions = new LinkedList<>(); |
| for (Condition subCondition : subConditions) { |
| Condition updatedCondition = updateSegmentDependentCondition(subCondition, segmentId); |
| if (updatedCondition != null) { |
| updatedSubConditions.add(updatedCondition); |
| } |
| } |
| if (!updatedSubConditions.isEmpty()) { |
| if (updatedSubConditions.size() == 1) { |
| return updatedSubConditions.get(0); |
| } else { |
| condition.setParameter("subConditions", updatedSubConditions); |
| return condition; |
| } |
| } else { |
| return null; |
| } |
| } else if ("profileSegmentCondition".equals(condition.getConditionTypeId())) { |
| @SuppressWarnings("unchecked") final List<String> referencedSegmentIds = (List<String>) condition.getParameter("segments"); |
| if (referencedSegmentIds.indexOf(segmentId) >= 0) { |
| referencedSegmentIds.remove(segmentId); |
| if (referencedSegmentIds.isEmpty()) { |
| return null; |
| } else { |
| condition.setParameter("segments", referencedSegmentIds); |
| } |
| } |
| } |
| return condition; |
| } |
| |
| private Set<Segment> getSegmentDependentSegments(String segmentId) { |
| Set<Segment> impactedSegments = new HashSet<>(this.allSegments.size()); |
| for (Segment segment : this.allSegments) { |
| if (checkSegmentDeletionImpact(segment.getCondition(), segmentId)) { |
| impactedSegments.add(segment); |
| } |
| } |
| return impactedSegments; |
| } |
| |
| private Set<Scoring> getSegmentDependentScorings(String segmentId) { |
| Set<Scoring> impactedScoring = new HashSet<>(this.allScoring.size()); |
| for (Scoring scoring : this.allScoring) { |
| for (ScoringElement element : scoring.getElements()) { |
| if (checkSegmentDeletionImpact(element.getCondition(), segmentId)) { |
| impactedScoring.add(scoring); |
| break; |
| } |
| } |
| } |
| return impactedScoring; |
| } |
| |
| public DependentMetadata getSegmentDependentMetadata(String segmentId) { |
| List<Metadata> segments = new LinkedList<>(); |
| List<Metadata> scorings = new LinkedList<>(); |
| for (Segment definition : getSegmentDependentSegments(segmentId)) { |
| segments.add(definition.getMetadata()); |
| } |
| for (Scoring definition : getSegmentDependentScorings(segmentId)) { |
| scorings.add(definition.getMetadata()); |
| } |
| return new DependentMetadata(segments, scorings); |
| } |
| |
| public DependentMetadata removeSegmentDefinition(String segmentId, boolean validate) { |
| Set<Segment> impactedSegments = getSegmentDependentSegments(segmentId); |
| Set<Scoring> impactedScorings = getSegmentDependentScorings(segmentId); |
| if (!validate || (impactedSegments.isEmpty() && impactedScorings.isEmpty())) { |
| // update profiles |
| Condition segmentCondition = new Condition(); |
| segmentCondition.setConditionType(definitionsService.getConditionType("profilePropertyCondition")); |
| segmentCondition.setParameter("propertyName", "segments"); |
| segmentCondition.setParameter("comparisonOperator", "equals"); |
| segmentCondition.setParameter("propertyValue", segmentId); |
| updateProfilesSegment(segmentCondition, segmentId, false, false); |
| |
| // update impacted segments |
| for (Segment segment : impactedSegments) { |
| Condition updatedCondition = updateSegmentDependentCondition(segment.getCondition(), segmentId); |
| segment.setCondition(updatedCondition); |
| if (updatedCondition == null) { |
| clearAutoGeneratedRules(persistenceService.query("linkedItems", segment.getMetadata().getId(), null, Rule.class), segment.getMetadata().getId()); |
| segment.getMetadata().setEnabled(false); |
| } |
| setSegmentDefinition(segment); |
| } |
| |
| // update impacted scorings |
| for (Scoring scoring : impactedScorings) { |
| List<ScoringElement> updatedScoringElements = new ArrayList<>(); |
| for (ScoringElement scoringElement : scoring.getElements()) { |
| Condition updatedCondition = updateSegmentDependentCondition(scoringElement.getCondition(), segmentId); |
| if (updatedCondition != null) { |
| scoringElement.setCondition(updatedCondition); |
| updatedScoringElements.add(scoringElement); |
| } |
| } |
| scoring.setElements(updatedScoringElements); |
| if (updatedScoringElements.isEmpty()) { |
| clearAutoGeneratedRules(persistenceService.query("linkedItems", scoring.getMetadata().getId(), null, Rule.class), scoring.getMetadata().getId()); |
| scoring.getMetadata().setEnabled(false); |
| } |
| setScoringDefinition(scoring); |
| } |
| |
| persistenceService.remove(segmentId, Segment.class); |
| List<Rule> previousRules = persistenceService.query("linkedItems", segmentId, null, Rule.class); |
| clearAutoGeneratedRules(previousRules, segmentId); |
| } |
| |
| List<Metadata> segments = new LinkedList<>(); |
| List<Metadata> scorings = new LinkedList<>(); |
| for (Segment definition : impactedSegments) { |
| segments.add(definition.getMetadata()); |
| } |
| for (Scoring definition : impactedScorings) { |
| scorings.add(definition.getMetadata()); |
| } |
| return new DependentMetadata(segments, scorings); |
| } |
| |
| public PartialList<Profile> getMatchingIndividuals(String segmentID, int offset, int size, String sortBy) { |
| Segment segment = getSegmentDefinition(segmentID); |
| if (segment == null) { |
| return new PartialList<Profile>(); |
| } |
| Condition segmentCondition = new Condition(definitionsService.getConditionType("profilePropertyCondition")); |
| segmentCondition.setParameter("propertyName", "segments"); |
| segmentCondition.setParameter("comparisonOperator", "equals"); |
| segmentCondition.setParameter("propertyValue", segmentID); |
| |
| return persistenceService.query(segmentCondition, sortBy, Profile.class, offset, size); |
| } |
| |
| public long getMatchingIndividualsCount(String segmentID) { |
| if (getSegmentDefinition(segmentID) == null) { |
| return 0; |
| } |
| |
| Condition segmentCondition = new Condition(definitionsService.getConditionType("profilePropertyCondition")); |
| segmentCondition.setParameter("propertyName", "segments"); |
| segmentCondition.setParameter("comparisonOperator", "equals"); |
| segmentCondition.setParameter("propertyValue", segmentID); |
| |
| return persistenceService.queryCount(segmentCondition, Profile.ITEM_TYPE); |
| } |
| |
| public Boolean isProfileInSegment(Profile profile, String segmentId) { |
| Set<String> matchingSegments = getSegmentsAndScoresForProfile(profile).getSegments(); |
| |
| return matchingSegments.contains(segmentId); |
| } |
| |
| public SegmentsAndScores getSegmentsAndScoresForProfile(Profile profile) { |
| Set<String> segments = new HashSet<String>(); |
| Map<String, Integer> scores = new HashMap<String, Integer>(); |
| |
| List<Segment> allSegments = this.allSegments; |
| for (Segment segment : allSegments) { |
| if (segment.getMetadata().isEnabled() && persistenceService.testMatch(segment.getCondition(), profile)) { |
| segments.add(segment.getMetadata().getId()); |
| } |
| } |
| |
| List<Scoring> allScoring = this.allScoring; |
| Map<String, Integer> scoreModifiers = (Map<String, Integer>) profile.getSystemProperties().get("scoreModifiers"); |
| for (Scoring scoring : allScoring) { |
| if (scoring.getMetadata().isEnabled()) { |
| int score = 0; |
| for (ScoringElement scoringElement : scoring.getElements()) { |
| if (persistenceService.testMatch(scoringElement.getCondition(), profile)) { |
| score += scoringElement.getValue(); |
| } |
| } |
| String scoringId = scoring.getMetadata().getId(); |
| if (scoreModifiers != null && scoreModifiers.containsKey(scoringId) && scoreModifiers.get(scoringId) != null) { |
| score += scoreModifiers.get(scoringId); |
| } |
| scores.put(scoringId, score); |
| } |
| } |
| |
| return new SegmentsAndScores(segments, scores); |
| } |
| |
| public List<Metadata> getSegmentMetadatasForProfile(Profile profile) { |
| List<Metadata> metadatas = new ArrayList<>(); |
| |
| List<Segment> allSegments = this.allSegments; |
| for (Segment segment : allSegments) { |
| if (segment.getMetadata().isEnabled() && persistenceService.testMatch(segment.getCondition(), profile)) { |
| metadatas.add(segment.getMetadata()); |
| } |
| } |
| |
| return metadatas; |
| } |
| |
| public PartialList<Metadata> getScoringMetadatas(int offset, int size, String sortBy) { |
| return getMetadatas(offset, size, sortBy, Scoring.class); |
| } |
| |
| public PartialList<Metadata> getScoringMetadatas(Query query) { |
| return getMetadatas(query, Scoring.class); |
| } |
| |
| private List<Scoring> getAllScoringDefinitions() { |
| List<Scoring> allItems = persistenceService.getAllItems(Scoring.class); |
| for (Scoring scoring : allItems) { |
| if (scoring.getMetadata().isEnabled()) { |
| for (ScoringElement element : scoring.getElements()) { |
| ParserHelper.resolveConditionType(definitionsService, element.getCondition(), "scoring " + scoring.getItemId()); |
| } |
| } |
| } |
| return allItems; |
| } |
| |
| public Scoring getScoringDefinition(String scoringId) { |
| Scoring definition = persistenceService.load(scoringId, Scoring.class); |
| if (definition != null && definition.getMetadata().isEnabled()) { |
| for (ScoringElement element : definition.getElements()) { |
| ParserHelper.resolveConditionType(definitionsService, element.getCondition(), "scoring " + scoringId); |
| } |
| } |
| return definition; |
| } |
| |
| public void setScoringDefinition(Scoring scoring) { |
| if (scoring.getMetadata().isEnabled()) { |
| for (ScoringElement element : scoring.getElements()) { |
| ParserHelper.resolveConditionType(definitionsService, element.getCondition(), "scoring " + scoring.getItemId() + " element "); |
| if (!scoring.getMetadata().isMissingPlugins()) { |
| updateAutoGeneratedRules(scoring.getMetadata(), element.getCondition()); |
| } |
| } |
| } |
| // make sure we update the name and description metadata that might not match, so first we remove the entry from the map |
| persistenceService.save(scoring); |
| |
| persistenceService.createMapping(Profile.ITEM_TYPE, String.format( |
| "{\n" + |
| " \"properties\": {\n" + |
| " \"scores\": {\n" + |
| " \"properties\": {\n" + |
| " \"%s\": {\n" + |
| " \"type\":\"long\"\n" + |
| " }\n" + |
| " }\n" + |
| " }\n" + |
| " }\n" + |
| "}", scoring.getItemId())); |
| |
| updateExistingProfilesForScoring(scoring.getItemId(), scoring.getElements(), scoring.getMetadata().isEnabled()); |
| } |
| |
| public void createScoringDefinition(String scope, String scoringId, String name, String description) { |
| Metadata metadata = new Metadata(scope, scoringId, name, description); |
| Scoring scoring = new Scoring(metadata); |
| Condition rootCondition = new Condition(); |
| rootCondition.setConditionType(definitionsService.getConditionType("booleanCondition")); |
| rootCondition.setParameter("operator", "and"); |
| rootCondition.setParameter("subConditions", new ArrayList<Condition>()); |
| scoring.setElements(new ArrayList<ScoringElement>()); |
| |
| setScoringDefinition(scoring); |
| } |
| |
| private boolean checkScoringDeletionImpact(Condition condition, String scoringToDeleteId) { |
| if (condition != null) { |
| @SuppressWarnings("unchecked") final List<Condition> subConditions = (List<Condition>) condition.getParameter("subConditions"); |
| if (subConditions != null) { |
| for (Condition subCondition : subConditions) { |
| if (checkScoringDeletionImpact(subCondition, scoringToDeleteId)) { |
| return true; |
| } |
| } |
| } else if ("scoringCondition".equals(condition.getConditionTypeId())) { |
| if (scoringToDeleteId.equals(condition.getParameter("scoringPlanId"))) { |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Return an updated condition that do not contain a condition on the scoringId anymore |
| * it's remove the unnecessary boolean condition (if a condition is the only one of a boolean the boolean will be remove and the subcondition returned) |
| * it's return null when there is no more condition after (if the condition passed was only a scoring condition on the scoringId) |
| * |
| * @param condition the condition to update |
| * @param scoringId the scoring id to remove in the condition |
| * @return updated condition |
| */ |
| private Condition updateScoringDependentCondition(Condition condition, String scoringId) { |
| if ("booleanCondition".equals(condition.getConditionTypeId())) { |
| @SuppressWarnings("unchecked") final List<Condition> subConditions = (List<Condition>) condition.getParameter("subConditions"); |
| List<Condition> updatedSubConditions = new LinkedList<>(); |
| for (Condition subCondition : subConditions) { |
| Condition updatedCondition = updateScoringDependentCondition(subCondition, scoringId); |
| if (updatedCondition != null) { |
| updatedSubConditions.add(updatedCondition); |
| } |
| } |
| if (!updatedSubConditions.isEmpty()) { |
| if (updatedSubConditions.size() == 1) { |
| return updatedSubConditions.get(0); |
| } else { |
| condition.setParameter("subConditions", updatedSubConditions); |
| return condition; |
| } |
| } else { |
| return null; |
| } |
| } else if ("scoringCondition".equals(condition.getConditionTypeId()) |
| && scoringId.equals(condition.getParameter("scoringPlanId"))) { |
| return null; |
| } |
| return condition; |
| } |
| |
| private Set<Segment> getScoringDependentSegments(String scoringId) { |
| Set<Segment> impactedSegments = new HashSet<>(this.allSegments.size()); |
| for (Segment segment : this.allSegments) { |
| if (checkScoringDeletionImpact(segment.getCondition(), scoringId)) { |
| impactedSegments.add(segment); |
| } |
| } |
| return impactedSegments; |
| } |
| |
| private Set<Scoring> getScoringDependentScorings(String scoringId) { |
| Set<Scoring> impactedScoring = new HashSet<>(this.allScoring.size()); |
| for (Scoring scoring : this.allScoring) { |
| for (ScoringElement element : scoring.getElements()) { |
| if (checkScoringDeletionImpact(element.getCondition(), scoringId)) { |
| impactedScoring.add(scoring); |
| break; |
| } |
| } |
| } |
| return impactedScoring; |
| } |
| |
| public DependentMetadata getScoringDependentMetadata(String scoringId) { |
| List<Metadata> segments = new LinkedList<>(); |
| List<Metadata> scorings = new LinkedList<>(); |
| for (Segment definition : getScoringDependentSegments(scoringId)) { |
| segments.add(definition.getMetadata()); |
| } |
| for (Scoring definition : getScoringDependentScorings(scoringId)) { |
| scorings.add(definition.getMetadata()); |
| } |
| return new DependentMetadata(segments, scorings); |
| } |
| |
| public DependentMetadata removeScoringDefinition(String scoringId, boolean validate) { |
| Set<Segment> impactedSegments = getScoringDependentSegments(scoringId); |
| Set<Scoring> impactedScorings = getScoringDependentScorings(scoringId); |
| if (!validate || (impactedSegments.isEmpty() && impactedScorings.isEmpty())) { |
| // update profiles |
| updateExistingProfilesForScoring(scoringId, Collections.emptyList(), false); |
| |
| // update impacted segments |
| for (Segment segment : impactedSegments) { |
| Condition updatedCondition = updateScoringDependentCondition(segment.getCondition(), scoringId); |
| segment.setCondition(updatedCondition); |
| if (updatedCondition == null) { |
| clearAutoGeneratedRules(persistenceService.query("linkedItems", segment.getMetadata().getId(), null, Rule.class), segment.getMetadata().getId()); |
| segment.getMetadata().setEnabled(false); |
| } |
| setSegmentDefinition(segment); |
| } |
| |
| // update impacted scorings |
| for (Scoring scoring : impactedScorings) { |
| List<ScoringElement> updatedScoringElements = new ArrayList<>(); |
| for (ScoringElement scoringElement : scoring.getElements()) { |
| Condition updatedCondition = updateScoringDependentCondition(scoringElement.getCondition(), scoringId); |
| if (updatedCondition != null) { |
| scoringElement.setCondition(updatedCondition); |
| updatedScoringElements.add(scoringElement); |
| } |
| } |
| scoring.setElements(updatedScoringElements); |
| if (updatedScoringElements.isEmpty()) { |
| clearAutoGeneratedRules(persistenceService.query("linkedItems", scoring.getMetadata().getId(), null, Rule.class), scoring.getMetadata().getId()); |
| scoring.getMetadata().setEnabled(false); |
| } |
| setScoringDefinition(scoring); |
| } |
| |
| persistenceService.remove(scoringId, Scoring.class); |
| List<Rule> previousRules = persistenceService.query("linkedItems", scoringId, null, Rule.class); |
| clearAutoGeneratedRules(previousRules, scoringId); |
| } |
| |
| List<Metadata> segments = new LinkedList<>(); |
| List<Metadata> scorings = new LinkedList<>(); |
| for (Segment definition : impactedSegments) { |
| segments.add(definition.getMetadata()); |
| } |
| for (Scoring definition : impactedScorings) { |
| scorings.add(definition.getMetadata()); |
| } |
| return new DependentMetadata(segments, scorings); |
| } |
| |
| public void updateAutoGeneratedRules(Metadata metadata, Condition condition) { |
| List<Rule> previousRules = persistenceService.query("linkedItems", metadata.getId(), null, Rule.class); |
| List<Rule> rules = new ArrayList<Rule>(); |
| if (condition != null) { |
| getAutoGeneratedRules(metadata, condition, null, rules); |
| } |
| for (Rule rule : rules) { |
| rulesService.setRule(rule); |
| } |
| previousRules.removeAll(rules); |
| clearAutoGeneratedRules(previousRules, metadata.getId()); |
| } |
| |
| private void clearAutoGeneratedRules(List<Rule> rules, String idWithScope) { |
| for (Rule previousRule : rules) { |
| previousRule.getLinkedItems().removeAll(Collections.singleton(idWithScope)); |
| if (previousRule.getLinkedItems().isEmpty()) { |
| // todo remove profile properties ? |
| persistenceService.remove(previousRule.getItemId(), Rule.class); |
| } else { |
| persistenceService.update(previousRule, null, Rule.class, "linkedItems", previousRule.getLinkedItems()); |
| } |
| } |
| } |
| |
| private void getAutoGeneratedRules(Metadata metadata, Condition condition, Condition parentCondition, List<Rule> rules) { |
| Set<String> tags = condition.getConditionType().getMetadata().getSystemTags(); |
| if (tags.contains("eventCondition") && !tags.contains("profileCondition")) { |
| String key = getGeneratedPropertyKey(condition, parentCondition); |
| if (key != null) { |
| parentCondition.setParameter("generatedPropertyKey", key); |
| Rule rule = rulesService.getRule(key); |
| if (rule == null) { |
| rule = new Rule(new Metadata(metadata.getScope(), key, "Auto generated rule for " + metadata.getName(), "")); |
| rule.setCondition(condition); |
| rule.getMetadata().setHidden(true); |
| final Action action = new Action(); |
| action.setActionType(definitionsService.getActionType("setEventOccurenceCountAction")); |
| action.setParameter("pastEventCondition", parentCondition); |
| |
| rule.setActions(Arrays.asList(action)); |
| rule.setLinkedItems(Arrays.asList(metadata.getId())); |
| |
| // it's a new generated rules to keep track of the event count, we should update all the profile that match this past event |
| // it will update the count of event occurrence on the profile directly |
| recalculatePastEventOccurrencesOnProfiles(condition, parentCondition, true, false); |
| } else if (!rule.getLinkedItems().contains(metadata.getId())) { |
| rule.getLinkedItems().add(metadata.getId()); |
| } |
| rules.add(rule); |
| } |
| } else { |
| Collection<Object> values = new ArrayList<>(condition.getParameterValues().values()); |
| for (Object parameterValue : values) { |
| if (parameterValue instanceof Condition) { |
| getAutoGeneratedRules(metadata, (Condition) parameterValue, condition, rules); |
| } else if (parameterValue instanceof Collection) { |
| for (Object subCondition : (Collection<?>) parameterValue) { |
| if (subCondition instanceof Condition) { |
| getAutoGeneratedRules(metadata, (Condition) subCondition, condition, rules); |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| /** |
| * This will recalculate the event counts on the profiles that match the given past event condition |
| * |
| * @param eventCondition the real condition |
| * @param parentCondition the past event condition |
| * @param forceRefresh will refresh the Profile index in case it's true |
| * @param resetExistingProfilesNotMatching if true, will reset existing profiles having a count to 0, in case they do not have events matching anymore |
| * ("false" can be useful when you know that no existing profiles already exist because it's a new rule for example, |
| * in that case setting this to "false" allow to skip profiles queries and speedup this process. |
| * Otherwise use "true" here to be sure the count is reset to 0 on profiles that need to be reset) |
| */ |
| private void recalculatePastEventOccurrencesOnProfiles(Condition eventCondition, Condition parentCondition, |
| boolean forceRefresh, boolean resetExistingProfilesNotMatching) { |
| long t = System.currentTimeMillis(); |
| List<Condition> l = new ArrayList<Condition>(); |
| Condition andCondition = new Condition(); |
| andCondition.setConditionType(definitionsService.getConditionType("booleanCondition")); |
| andCondition.setParameter("operator", "and"); |
| andCondition.setParameter("subConditions", l); |
| |
| l.add(eventCondition); |
| |
| Integer numberOfDays = (Integer) parentCondition.getParameter("numberOfDays"); |
| String fromDate = (String) parentCondition.getParameter("fromDate"); |
| String toDate = (String) parentCondition.getParameter("toDate"); |
| |
| if (numberOfDays != null) { |
| Condition numberOfDaysCondition = new Condition(); |
| numberOfDaysCondition.setConditionType(definitionsService.getConditionType("sessionPropertyCondition")); |
| numberOfDaysCondition.setParameter("propertyName", "timeStamp"); |
| numberOfDaysCondition.setParameter("comparisonOperator", "greaterThan"); |
| numberOfDaysCondition.setParameter("propertyValue", "now-" + numberOfDays + "d"); |
| l.add(numberOfDaysCondition); |
| } |
| if (fromDate != null) { |
| Condition startDateCondition = new Condition(); |
| startDateCondition.setConditionType(definitionsService.getConditionType("sessionPropertyCondition")); |
| startDateCondition.setParameter("propertyName", "timeStamp"); |
| startDateCondition.setParameter("comparisonOperator", "greaterThanOrEqualTo"); |
| startDateCondition.setParameter("propertyValueDate", fromDate); |
| l.add(startDateCondition); |
| } |
| if (toDate != null) { |
| Condition endDateCondition = new Condition(); |
| endDateCondition.setConditionType(definitionsService.getConditionType("sessionPropertyCondition")); |
| endDateCondition.setParameter("propertyName", "timeStamp"); |
| endDateCondition.setParameter("comparisonOperator", "lessThanOrEqualTo"); |
| endDateCondition.setParameter("propertyValueDate", toDate); |
| l.add(endDateCondition); |
| } |
| |
| String propertyKey = (String) parentCondition.getParameter("generatedPropertyKey"); |
| Set<String> existingProfilesWithCounts = resetExistingProfilesNotMatching ? getExistingProfilesWithPastEventOccurrenceCount(propertyKey) : Collections.emptySet(); |
| |
| int updatedProfileCount = 0; |
| if (pastEventsDisablePartitions) { |
| Map<String, Long> eventCountByProfile = persistenceService.aggregateWithOptimizedQuery(eventCondition, new TermsAggregate("profileId"), Event.ITEM_TYPE, maximumIdsQueryCount); |
| Set<String> updatedProfiles = updatePastEventOccurrencesOnProfiles(eventCountByProfile, propertyKey); |
| existingProfilesWithCounts.removeAll(updatedProfiles); |
| updatedProfileCount = updatedProfiles.size(); |
| } else { |
| Map<String, Double> m = persistenceService.getSingleValuesMetrics(andCondition, 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<String, Long> eventCountByProfile = persistenceService.aggregateWithOptimizedQuery(andCondition, new TermsAggregate("profileId", i, numParts), Event.ITEM_TYPE); |
| Set<String> updatedProfiles = updatePastEventOccurrencesOnProfiles(eventCountByProfile, propertyKey); |
| existingProfilesWithCounts.removeAll(updatedProfiles); |
| updatedProfileCount += updatedProfiles.size(); |
| } |
| } |
| |
| // remaining existing profiles with counts should be reset to 0 since they have not been updated it means |
| // that they do not have matching events anymore in the time based condition |
| if (!existingProfilesWithCounts.isEmpty()) { |
| updatedProfileCount += updatePastEventOccurrencesOnProfiles( |
| existingProfilesWithCounts.stream().collect(Collectors.toMap(key -> key, value -> 0L)), propertyKey).size(); |
| } |
| |
| if (forceRefresh && updatedProfileCount > 0) { |
| persistenceService.refreshIndex(Profile.class, null); |
| } |
| |
| logger.info("{} profiles updated for past event condition in {}ms", updatedProfileCount, System.currentTimeMillis() - t); |
| } |
| |
| /** |
| * Return the list of profile ids, for profiles that already have an event count matching the generated property key |
| * |
| * @param generatedPropertyKey the generated property key of the generated rule for the given past event condition. |
| * @return the list of profile ids. |
| */ |
| private Set<String> getExistingProfilesWithPastEventOccurrenceCount(String generatedPropertyKey) { |
| Condition countExistsCondition = new Condition(); |
| countExistsCondition.setConditionType(definitionsService.getConditionType("profilePropertyCondition")); |
| countExistsCondition.setParameter("propertyName", "systemProperties.pastEvents." + generatedPropertyKey); |
| countExistsCondition.setParameter("comparisonOperator", "greaterThan"); |
| countExistsCondition.setParameter("propertyValueInteger", 0); |
| |
| Set<String> profileIds = new HashSet<>(); |
| if (pastEventsDisablePartitions) { |
| profileIds.addAll(persistenceService.aggregateWithOptimizedQuery(countExistsCondition, new TermsAggregate("itemId"), |
| Profile.ITEM_TYPE, maximumIdsQueryCount).keySet()); |
| } else { |
| Map<String, Double> m = persistenceService.getSingleValuesMetrics(countExistsCondition, new String[]{"card"}, "itemId.keyword", Profile.ITEM_TYPE); |
| long card = m.get("_card").longValue(); |
| int numParts = (int) (card / aggregateQueryBucketSize) + 2; |
| for (int i = 0; i < numParts; i++) { |
| profileIds.addAll(persistenceService.aggregateWithOptimizedQuery(countExistsCondition, new TermsAggregate("itemId", i, numParts), |
| Profile.ITEM_TYPE).keySet()); |
| } |
| } |
| return profileIds; |
| } |
| |
| public String getGeneratedPropertyKey(Condition condition, Condition parentCondition) { |
| try { |
| Map<String, Object> m = new HashMap<>(); |
| m.put("condition", condition); |
| m.put("numberOfDays", parentCondition.getParameter("numberOfDays")); |
| // Put fromDate and toDate only if exist - for backward compatibility |
| Object fromDate = parentCondition.getParameter("fromDate"); |
| if (fromDate != null) { |
| m.put("fromDate", parentCondition.getParameter("fromDate")); |
| } |
| Object toDate = parentCondition.getParameter("toDate"); |
| if (toDate != null) { |
| m.put("fromDate", parentCondition.getParameter("toDate")); |
| } |
| |
| String key = CustomObjectMapper.getObjectMapper().writeValueAsString(m); |
| return "eventTriggered" + getMD5(key); |
| } catch (JsonProcessingException e) { |
| logger.error("Cannot generate key", e); |
| return null; |
| } |
| } |
| |
| @Override |
| public void recalculatePastEventConditions() { |
| Set<String> segmentOrScoringIdsToReevaluate = new HashSet<>(); |
| // reevaluate auto generated rules used to store the event occurrence count on the profile |
| for (Metadata metadata : rulesService.getRuleMetadatas()) { |
| Rule rule = rulesService.getRule(metadata.getId()); |
| for (Action action : rule.getActions()) { |
| if (action.getActionTypeId().equals("setEventOccurenceCountAction")) { |
| Condition pastEventCondition = (Condition) action.getParameterValues().get("pastEventCondition"); |
| if (pastEventCondition.containsParameter("numberOfDays")) { |
| recalculatePastEventOccurrencesOnProfiles(rule.getCondition(), pastEventCondition, true, true); |
| logger.info("Event occurrence count on profiles updated for rule: {}", rule.getItemId()); |
| if (rule.getLinkedItems() != null && rule.getLinkedItems().size() > 0) { |
| segmentOrScoringIdsToReevaluate.addAll(rule.getLinkedItems()); |
| } |
| } |
| } |
| } |
| } |
| int pastEventSegmentsAndScoringsSize = segmentOrScoringIdsToReevaluate.size(); |
| logger.info("Found {} segments or scoring plans containing pastEventCondition conditions", pastEventSegmentsAndScoringsSize); |
| |
| // get Segments and Scoring that contains relative date expressions |
| segmentOrScoringIdsToReevaluate.addAll(allSegments.stream() |
| .filter(segment -> segment.getCondition() != null && segment.getCondition().toString().contains("propertyValueDateExpr")) |
| .map(Item::getItemId) |
| .collect(Collectors.toList())); |
| |
| segmentOrScoringIdsToReevaluate.addAll(allScoring.stream() |
| .filter(scoring -> scoring.getElements() != null && scoring.getElements().size() > 0 && scoring.getElements().stream() |
| .anyMatch(scoringElement -> scoringElement != null && scoringElement.getCondition() != null && scoringElement.getCondition().toString().contains("propertyValueDateExpr"))) |
| .map(Item::getItemId) |
| .collect(Collectors.toList())); |
| logger.info("Found {} segments or scoring plans containing date relative expressions", segmentOrScoringIdsToReevaluate.size() - pastEventSegmentsAndScoringsSize); |
| |
| // reevaluate segments and scoring. |
| if (segmentOrScoringIdsToReevaluate.size() > 0) { |
| persistenceService.refreshIndex(Profile.class, null); |
| for (String linkedItem : segmentOrScoringIdsToReevaluate) { |
| Segment linkedSegment = getSegmentDefinition(linkedItem); |
| if (linkedSegment != null) { |
| logger.info("Start segment recalculation for segment: {} - {}", linkedSegment.getItemId(), linkedSegment.getMetadata().getName()); |
| updateExistingProfilesForSegment(linkedSegment); |
| continue; |
| } |
| |
| Scoring linkedScoring = getScoringDefinition(linkedItem); |
| if (linkedScoring != null) { |
| logger.info("Start scoring plan recalculation for scoring plan: {} - {}", linkedScoring.getItemId(), linkedScoring.getMetadata().getName()); |
| updateExistingProfilesForScoring(linkedScoring.getItemId(), linkedScoring.getElements(), linkedScoring.getMetadata().isEnabled()); |
| } |
| } |
| } |
| } |
| |
| /** |
| * This will update all the profiles in the given map with the according new count occurrence for the given propertyKey |
| * |
| * @param eventCountByProfile the events count per profileId map |
| * @param propertyKey the generate property key for this past event condition, to keep track of the count in the profile |
| * @return the list of profiles for witch the count of event occurrences have been updated. |
| */ |
| private Set<String> updatePastEventOccurrencesOnProfiles(Map<String, Long> eventCountByProfile, String propertyKey) { |
| Set<String> profilesUpdated = new HashSet<>(); |
| Map<Item, Map> batch = new HashMap<>(); |
| Iterator<Map.Entry<String, Long>> entryIterator = eventCountByProfile.entrySet().iterator(); |
| while (entryIterator.hasNext()) { |
| Map.Entry<String, Long> entry = entryIterator.next(); |
| String profileId = entry.getKey(); |
| if (!profileId.startsWith("_")) { |
| Map<String, Long> pastEventCounts = new HashMap<>(); |
| pastEventCounts.put(propertyKey, entry.getValue()); |
| Map<String, Object> systemProperties = new HashMap<>(); |
| systemProperties.put("pastEvents", pastEventCounts); |
| systemProperties.put("lastUpdated", new Date()); |
| |
| Profile profile = new Profile(); |
| profile.setItemId(profileId); |
| batch.put(profile, Collections.singletonMap("systemProperties", systemProperties)); |
| profilesUpdated.add(profileId); |
| } |
| |
| if (batch.size() == segmentUpdateBatchSize || (!entryIterator.hasNext() && batch.size() > 0)) { |
| try { |
| persistenceService.update(batch, null, Profile.class); |
| } catch (Exception e) { |
| logger.error("Error updating {} profiles for past event system properties", batch.size(), e); |
| } finally { |
| batch.clear(); |
| } |
| } |
| } |
| return profilesUpdated; |
| } |
| |
| private String getMD5(String md5) { |
| try { |
| MessageDigest md = MessageDigest.getInstance("MD5"); |
| byte[] array = md.digest(md5.getBytes()); |
| StringBuffer sb = new StringBuffer(); |
| for (int i = 0; i < array.length; ++i) { |
| sb.append(Integer.toHexString((array[i] & 0xFF) | 0x100).substring(1, 3)); |
| } |
| return sb.toString(); |
| } catch (java.security.NoSuchAlgorithmException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| private void updateExistingProfilesForSegment(Segment segment) { |
| long updateProfilesForSegmentStartTime = System.currentTimeMillis(); |
| long updatedProfileCount = 0; |
| final String segmentId = segment.getItemId(); |
| |
| Condition segmentCondition = new Condition(); |
| segmentCondition.setConditionType(definitionsService.getConditionType("profilePropertyCondition")); |
| segmentCondition.setParameter("propertyName", "segments"); |
| segmentCondition.setParameter("comparisonOperator", "equals"); |
| segmentCondition.setParameter("propertyValue", segmentId); |
| |
| if (segment.getMetadata().isEnabled()) { |
| |
| ConditionType booleanConditionType = definitionsService.getConditionType("booleanCondition"); |
| ConditionType notConditionType = definitionsService.getConditionType("notCondition"); |
| |
| Condition profilesToAddCondition = new Condition(booleanConditionType); |
| profilesToAddCondition.setParameter("operator", "and"); |
| List<Condition> profilesToAddSubConditions = new ArrayList<>(); |
| profilesToAddSubConditions.add(segment.getCondition()); |
| Condition notOldSegmentCondition = new Condition(notConditionType); |
| notOldSegmentCondition.setParameter("subCondition", segmentCondition); |
| profilesToAddSubConditions.add(notOldSegmentCondition); |
| profilesToAddCondition.setParameter("subConditions", profilesToAddSubConditions); |
| |
| Condition profilesToRemoveCondition = new Condition(booleanConditionType); |
| profilesToRemoveCondition.setParameter("operator", "and"); |
| List<Condition> profilesToRemoveSubConditions = new ArrayList<>(); |
| profilesToRemoveSubConditions.add(segmentCondition); |
| Condition notNewSegmentCondition = new Condition(notConditionType); |
| notNewSegmentCondition.setParameter("subCondition", segment.getCondition()); |
| profilesToRemoveSubConditions.add(notNewSegmentCondition); |
| profilesToRemoveCondition.setParameter("subConditions", profilesToRemoveSubConditions); |
| |
| updatedProfileCount += updateProfilesSegment(profilesToAddCondition, segmentId, true, sendProfileUpdateEventForSegmentUpdate); |
| updatedProfileCount += updateProfilesSegment(profilesToRemoveCondition, segmentId, false, sendProfileUpdateEventForSegmentUpdate); |
| } else { |
| updatedProfileCount += updateProfilesSegment(segmentCondition, segmentId, false, sendProfileUpdateEventForSegmentUpdate); |
| } |
| logger.info("{} profiles updated in {}ms", updatedProfileCount, System.currentTimeMillis() - updateProfilesForSegmentStartTime); |
| } |
| |
| private long updateProfilesSegment(Condition profilesToUpdateCondition, String segmentId, boolean isAdd, boolean sendProfileUpdateEvent) { |
| long updatedProfileCount = 0; |
| PartialList<Profile> profiles = persistenceService.query(profilesToUpdateCondition, null, Profile.class, 0, segmentUpdateBatchSize, "10m"); |
| |
| while (profiles != null && profiles.getList().size() > 0) { |
| long startTime = System.currentTimeMillis(); |
| if (batchSegmentProfileUpdate) { |
| batchUpdateProfilesSegment(segmentId, profiles.getList(), isAdd); |
| } else { //send update profile one by one |
| for (Profile profileToUpdate : profiles.getList()) { |
| Map<String, Object> sourceMap = buildPropertiesMapForUpdateSegment(profileToUpdate, segmentId, isAdd); |
| persistenceService.update(profileToUpdate, null, Profile.class, sourceMap); |
| } |
| } |
| if (sendProfileUpdateEvent) |
| sendProfileUpdatedEvent(profiles.getList()); |
| |
| updatedProfileCount += profiles.size(); |
| logger.info("{} profiles {} to segment {} in {}ms", profiles.size(), isAdd ? "added" : "removed", segmentId, System.currentTimeMillis() - startTime); |
| |
| profiles = persistenceService.continueScrollQuery(Profile.class, profiles.getScrollIdentifier(), profiles.getScrollTimeValidity()); |
| } |
| |
| return updatedProfileCount; |
| } |
| |
| private void batchUpdateProfilesSegment(String segmentId, List<Profile> profiles, boolean isAdd) { |
| Map<Item, Map> profileToPropertiesMap = new HashMap<>(); |
| for (Profile profileToUpdate : profiles) { |
| Map<String, Object> propertiesToUpdate = buildPropertiesMapForUpdateSegment(profileToUpdate, segmentId, isAdd); |
| profileToPropertiesMap.put(profileToUpdate, propertiesToUpdate); |
| } |
| List<String> failedItemsIds = persistenceService.update(profileToPropertiesMap, null, Profile.class); |
| if (failedItemsIds != null) |
| failedItemsIds.forEach(s -> retryFailedSegmentUpdate(s, segmentId, isAdd)); |
| } |
| |
| private void retryFailedSegmentUpdate(String profileId, String segmentId, boolean isAdd) { |
| if (maxRetriesForUpdateProfileSegment > 0) { |
| RetryPolicy retryPolicy = new RetryPolicy() |
| .withDelay(Duration.ofSeconds(secondsDelayForRetryUpdateProfileSegment)) |
| .withMaxRetries(maxRetriesForUpdateProfileSegment); |
| |
| Failsafe.with(retryPolicy). |
| run(executionContext -> { |
| logger.warn("retry updating profile segment {}, profile {}, time {}", segmentId, profileId, new Date()); |
| Profile profileToAddUpdated = persistenceService.load(profileId, Profile.class); |
| Map<String, Object> sourceMapToUpdate = buildPropertiesMapForUpdateSegment(profileToAddUpdated, segmentId, isAdd); |
| boolean isUpdated = persistenceService.update(profileToAddUpdated, null, Profile.class, sourceMapToUpdate); |
| if (isUpdated == false) |
| throw new Exception(String.format("failed retry update profile segment {}, profile {}, time {}", segmentId, profileId, new Date())); |
| }); |
| } |
| } |
| |
| private void sendProfileUpdatedEvent(List<Profile> profiles) { |
| for (Profile profileToAdd : profiles) { |
| sendProfileUpdatedEvent(profileToAdd); |
| } |
| } |
| |
| private void sendProfileUpdatedEvent(Profile profile) { |
| Event profileUpdated = new Event("profileUpdated", null, profile, null, null, profile, new Date()); |
| profileUpdated.setPersistent(false); |
| eventService.send(profileUpdated); |
| } |
| |
| private Map<String, Object> buildPropertiesMapForUpdateSegment(Profile profile, String segmentId, boolean isAdd) { |
| if (isAdd) |
| profile.getSegments().add(segmentId); |
| else |
| profile.getSegments().remove(segmentId); |
| |
| Map<String, Object> sourceMap = new HashMap<>(); |
| sourceMap.put("segments", profile.getSegments()); |
| profile.setSystemProperty("lastUpdated", new Date()); |
| sourceMap.put("systemProperties", profile.getSystemProperties()); |
| return sourceMap; |
| } |
| |
| private void updateExistingProfilesForScoring(String scoringId, List<ScoringElement> scoringElements, boolean isEnabled) { |
| long startTime = System.currentTimeMillis(); |
| |
| String[] scripts = new String[scoringElements.size() + 1]; |
| Map<String, Object>[] scriptParams = new HashMap[scoringElements.size() + 1]; |
| Condition[] conditions = new Condition[scoringElements.size() + 1]; |
| |
| // reset Score |
| scriptParams[0] = new HashMap<>(); |
| scriptParams[0].put("scoringId", scoringId); |
| scripts[0] = RESET_SCORING_SCRIPT; |
| conditions[0] = new Condition(); |
| conditions[0].setConditionType(definitionsService.getConditionType("profilePropertyCondition")); |
| conditions[0].setParameter("propertyName", "scores." + scoringId); |
| conditions[0].setParameter("comparisonOperator", "exists"); |
| |
| // evaluate each elements of the scoring |
| if (isEnabled) { |
| int idx = 1; |
| for (ScoringElement element : scoringElements) { |
| scriptParams[idx] = new HashMap<>(); |
| scriptParams[idx].put("scoringId", scoringId); |
| scriptParams[idx].put("scoringValue", element.getValue()); |
| scripts[idx] = EVALUATE_SCORING_ELEMENT_SCRIPT; |
| conditions[idx] = element.getCondition(); |
| idx++; |
| } |
| } |
| persistenceService.updateWithQueryAndStoredScript(null, Profile.class, scripts, scriptParams, conditions); |
| logger.info("Updated scoring for profiles in {}ms", System.currentTimeMillis() - startTime); |
| } |
| |
| public void bundleChanged(BundleEvent event) { |
| switch (event.getType()) { |
| case BundleEvent.STARTED: |
| processBundleStartup(event.getBundle().getBundleContext()); |
| break; |
| case BundleEvent.STOPPING: |
| processBundleStop(event.getBundle().getBundleContext()); |
| break; |
| } |
| } |
| |
| private void initializeTimer() { |
| |
| TimerTask task = new TimerTask() { |
| @Override |
| public void run() { |
| try { |
| long currentTimeMillis = System.currentTimeMillis(); |
| logger.info("running scheduled task to recalculate segments and scoring that contains date relative conditions"); |
| recalculatePastEventConditions(); |
| logger.info("finished recalculate segments and scoring that contains date relative conditions in {}ms. ", System.currentTimeMillis() - currentTimeMillis); |
| } catch (Throwable t) { |
| logger.error("Error while updating profiles for segments and scoring that contains date relative conditions", t); |
| } |
| } |
| }; |
| long initialDelay = SchedulerServiceImpl.getTimeDiffInSeconds(dailyDateExprEvaluationHourUtc, ZonedDateTime.now(ZoneOffset.UTC)); |
| logger.info("daily recalculation job for segments and scoring that contains date relative conditions will run at fixed rate, initialDelay={}, taskExecutionPeriod={}", initialDelay, TimeUnit.DAYS.toSeconds(1)); |
| schedulerService.getScheduleExecutorService().scheduleAtFixedRate(task, initialDelay, taskExecutionPeriod, TimeUnit.DAYS); |
| |
| task = new TimerTask() { |
| @Override |
| public void run() { |
| try { |
| allSegments = getAllSegmentDefinitions(); |
| allScoring = getAllScoringDefinitions(); |
| } catch (Throwable t) { |
| logger.error("Error while loading segments and scoring definitions from persistence back-end", t); |
| } |
| } |
| }; |
| schedulerService.getScheduleExecutorService().scheduleAtFixedRate(task, 0, segmentRefreshInterval, TimeUnit.MILLISECONDS); |
| } |
| |
| private void loadScripts() throws IOException { |
| Enumeration<URL> scriptsURL = bundleContext.getBundle().findEntries("META-INF/cxs/painless", "*.painless", true); |
| if (scriptsURL == null) { |
| return; |
| } |
| |
| Map<String, String> scriptsById = new HashMap<>(); |
| while (scriptsURL.hasMoreElements()) { |
| URL scriptURL = scriptsURL.nextElement(); |
| logger.debug("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); |
| } |
| } |
| persistenceService.storeScripts(scriptsById); |
| } |
| |
| public void setTaskExecutionPeriod(long taskExecutionPeriod) { |
| this.taskExecutionPeriod = taskExecutionPeriod; |
| } |
| } |
