| /* |
| * 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.usergrid.corepersistence.asyncevents; |
| |
| |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.UUID; |
| |
| import org.apache.usergrid.corepersistence.index.*; |
| import org.apache.usergrid.utils.UUIDUtils; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.usergrid.persistence.Schema; |
| import org.apache.usergrid.persistence.collection.EntityCollectionManager; |
| import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory; |
| import org.apache.usergrid.persistence.collection.MvccLogEntry; |
| import org.apache.usergrid.persistence.collection.serialization.SerializationFig; |
| import org.apache.usergrid.persistence.core.scope.ApplicationScope; |
| import org.apache.usergrid.persistence.graph.Edge; |
| import org.apache.usergrid.persistence.graph.GraphManager; |
| import org.apache.usergrid.persistence.graph.GraphManagerFactory; |
| import org.apache.usergrid.persistence.index.impl.IndexOperationMessage; |
| import org.apache.usergrid.persistence.model.entity.Entity; |
| import org.apache.usergrid.persistence.model.entity.Id; |
| import org.apache.usergrid.persistence.model.field.Field; |
| |
| import com.google.inject.Inject; |
| import com.google.inject.Singleton; |
| |
| import rx.Observable; |
| |
| |
| /** |
| * Service that executes event flows |
| */ |
| @Singleton |
| public class EventBuilderImpl implements EventBuilder { |
| |
| private static final Logger logger = LoggerFactory.getLogger( EventBuilderImpl.class ); |
| |
| private final IndexService indexService; |
| private final EntityCollectionManagerFactory entityCollectionManagerFactory; |
| private final GraphManagerFactory graphManagerFactory; |
| private final SerializationFig serializationFig; |
| |
| |
| @Inject |
| public EventBuilderImpl( final IndexService indexService, |
| final EntityCollectionManagerFactory entityCollectionManagerFactory, |
| final GraphManagerFactory graphManagerFactory, final SerializationFig serializationFig ) { |
| this.indexService = indexService; |
| this.entityCollectionManagerFactory = entityCollectionManagerFactory; |
| this.graphManagerFactory = graphManagerFactory; |
| this.serializationFig = serializationFig; |
| } |
| |
| |
| |
| @Override |
| public Observable<IndexOperationMessage> buildNewEdge( final ApplicationScope applicationScope, final Entity entity, |
| final Edge newEdge ) { |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("Indexing in app scope {} with entity {} and new edge {}", |
| applicationScope, entity, newEdge); |
| } |
| |
| return indexService.indexEdge( applicationScope, entity, newEdge ); |
| } |
| |
| |
| @Override |
| public IndexOperationMessage buildDeleteEdge( final ApplicationScope applicationScope, final Edge |
| edge ) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Deleting in app scope {} with edge {}", applicationScope, edge); |
| } |
| |
| final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope ); |
| final EntityCollectionManager ecm = entityCollectionManagerFactory.createCollectionManager( applicationScope ); |
| |
| final IndexOperationMessage combined = new IndexOperationMessage(); |
| |
| gm.deleteEdge( edge ) |
| .doOnNext( deletedEdge -> { |
| |
| logger.debug("Processing deleted edge for de-indexing {}", deletedEdge); |
| |
| // get ALL versions of the target node as any connection from this source node needs to be removed |
| ecm.getVersionsFromMaxToMin(deletedEdge.getTargetNode(), UUIDUtils.newTimeUUID()) |
| .doOnNext(mvccLogEntry -> { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Adding edge {} mvccLogEntry {} to de-index batch", deletedEdge.getTargetNode(), mvccLogEntry); |
| } |
| combined.ingest( |
| indexService |
| .deIndexEdge(applicationScope, deletedEdge, mvccLogEntry.getEntityId(), mvccLogEntry.getVersion()) |
| .toBlocking().lastOrDefault(new IndexOperationMessage())); |
| |
| }).toBlocking().lastOrDefault(null); |
| |
| }).toBlocking().lastOrDefault(null); |
| |
| return combined; |
| } |
| |
| |
| //Does the queue entityDelete mark the entity then immediately does to the deleteEntityIndex. seems like |
| //it'll need to be pushed up higher so we can do the marking that isn't async or does it not matter? |
| |
| @Override |
| public IndexOperationMessage buildEntityDelete(final ApplicationScope applicationScope, final Id entityId ) { |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("Deleting entity id (marked versions) from index in app scope {} with entityId {}", |
| applicationScope, entityId); |
| } |
| |
| final EntityCollectionManager ecm = entityCollectionManagerFactory.createCollectionManager( applicationScope ); |
| final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope ); |
| |
| MvccLogEntry mostRecentToDelete = |
| ecm.getVersionsFromMaxToMin( entityId, UUIDUtils.newTimeUUID() ) |
| .toBlocking() |
| .firstOrDefault( null, mvccLogEntry -> mvccLogEntry.getState() == MvccLogEntry.State.DELETED ); |
| |
| // logger.info("mostRecent stage={} entityId={} version={} state={}", |
| // mostRecentToDelete.getStage().name(), mostRecentToDelete.getEntityId(), |
| // mostRecentToDelete.getVersion().toString(), mostRecentToDelete.getState().name()); |
| |
| if (mostRecentToDelete == null) { |
| logger.info("No entity versions to delete for id {}", entityId.toString()); |
| } |
| // if nothing is marked, then abort |
| if(mostRecentToDelete == null){ |
| return new IndexOperationMessage(); |
| } |
| |
| final List<MvccLogEntry> logEntries = new ArrayList<>(); |
| Observable<MvccLogEntry> mvccLogEntryListObservable = |
| ecm.getVersionsFromMaxToMin( entityId, UUIDUtils.newTimeUUID() ); |
| mvccLogEntryListObservable |
| .filter( mvccLogEntry-> mvccLogEntry.getVersion().timestamp() <= mostRecentToDelete.getVersion().timestamp() ) |
| .buffer( serializationFig.getBufferSize() ) |
| .doOnNext( buffer -> ecm.delete( buffer ) ) |
| .doOnNext(mvccLogEntries -> { |
| logEntries.addAll(mvccLogEntries); |
| }).toBlocking().lastOrDefault(null); |
| |
| //logger.info("logEntries size={}", logEntries.size()); |
| |
| IndexOperationMessage combined = new IndexOperationMessage(); |
| |
| // do the edge deletes and build up de-index messages for each edge deleted |
| // assume we have "server1" and "region1" nodes in the graph with the following relationships (edges/connections): |
| // |
| // region1 -- zzzconnzzz|has --> server1 |
| // server1 -- zzzconnzzz|in --> region1 |
| // |
| // there will always be a relationship from the appId to each entity based on the entity type (collection): |
| // |
| // application -- zzzcollzzz|servers --> server1 |
| // application -- zzzcollzzz|regions --> region1 |
| // |
| // When deleting either "server1" or "region1" entity, the connections should get deleted and de-indexed along |
| // with the entry for the entity itself in the collection. The above example should have at minimum 3 things to |
| // be de-indexed. There may be more as either "server1" or "region1" could have multiple versions. |
| // |
| // Further comments using the example of deleting "server1" from the above example. |
| gm.compactNode(entityId).doOnNext(markedEdge -> { |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("Processing deleted edge for de-indexing {}", markedEdge); |
| } |
| |
| // if the edge was for a connection where the entity to be deleted is the source node, we need to load |
| // the target node's versions so that all versions of connections to that entity can be de-indexed |
| // server1 -- zzzconnzzz|in --> region1 |
| if(!markedEdge.getTargetNode().getType().equals(entityId.getType())){ |
| |
| // get ALL versions of the target node as any connection from this source node needs to be removed |
| ecm.getVersionsFromMaxToMin( markedEdge.getTargetNode(), UUIDUtils.newTimeUUID() ) |
| .doOnNext(mvccLogEntry -> { |
| logger.debug("Adding edge {} mvccLogEntry {} to de-index batch", markedEdge, mvccLogEntry); |
| combined.ingest( |
| indexService |
| .deIndexEdge(applicationScope, markedEdge, mvccLogEntry.getEntityId(), mvccLogEntry.getVersion()) |
| .toBlocking().lastOrDefault(new IndexOperationMessage())); |
| |
| }).toBlocking().lastOrDefault(null); |
| |
| }else { |
| |
| // for each version of the entity being deleted, de-index the connections where the entity is the target |
| // node ( application -- zzzcollzzz|servers --> server1 ) or (region1 -- zzzconnzzz|has --> server1) |
| logEntries.forEach(logEntry -> { |
| logger.debug("Adding edge {} mvccLogEntry {} to de-index batch", markedEdge, logEntry); |
| combined.ingest( |
| indexService |
| .deIndexEdge(applicationScope, markedEdge, logEntry.getEntityId(), logEntry.getVersion()) |
| .toBlocking().lastOrDefault(new IndexOperationMessage())); |
| }); |
| } |
| |
| }).toBlocking().lastOrDefault(null); |
| |
| return combined; |
| } |
| |
| @Override |
| public Observable<IndexOperationMessage> buildEntityIndex( final EntityIndexOperation entityIndexOperation ) { |
| |
| final ApplicationScope applicationScope = entityIndexOperation.getApplicationScope(); |
| |
| final Id entityId = entityIndexOperation.getId(); |
| |
| //load the entity |
| return entityCollectionManagerFactory.createCollectionManager( applicationScope ).load( entityId ).filter( |
| entity -> { |
| final Field<Long> modified = entity.getField( Schema.PROPERTY_MODIFIED ); |
| |
| /** |
| * We don't have a modified field, so we can't check, pass it through |
| */ |
| if ( modified == null ) { |
| return true; |
| } |
| |
| //entityIndexOperation.getUpdatedSince will always be 0 except for reindexing the application |
| //only re-index if it has been updated and been updated after our timestamp |
| return modified.getValue() >= entityIndexOperation.getUpdatedSince(); |
| } ) |
| //perform indexing on the task scheduler and start it |
| .flatMap( entity -> indexService.indexEntity( applicationScope, entity ) ); |
| } |
| |
| |
| @Override |
| public Observable<IndexOperationMessage> deIndexOldVersions( final ApplicationScope applicationScope, |
| final Id entityId, final UUID markedVersion ){ |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("Removing old versions of entity {} from index in app scope {}", entityId, applicationScope ); |
| } |
| |
| final EntityCollectionManager ecm = entityCollectionManagerFactory.createCollectionManager( applicationScope ); |
| |
| |
| return indexService.deIndexOldVersions( applicationScope, entityId, |
| getVersionsOlderThanOrEqualToMarked(ecm, entityId, markedVersion)); |
| |
| } |
| |
| |
| private List<UUID> getVersionsOlderThanOrEqualToMarked(final EntityCollectionManager ecm, |
| final Id entityId, final UUID markedVersion ){ |
| |
| final List<UUID> versions = new ArrayList<>(); |
| |
| // only take last 100 versions to avoid eating memory. a tool can be built for massive cleanups for old usergrid |
| // clusters that do not have this in-line cleanup |
| ecm.getVersionsFromMaxToMin( entityId, markedVersion) |
| .take(100) |
| .forEach( mvccLogEntry -> { |
| if ( mvccLogEntry.getVersion().timestamp() <= markedVersion.timestamp() ) { |
| versions.add(mvccLogEntry.getVersion()); |
| } |
| |
| }); |
| |
| |
| return versions; |
| } |
| |
| private List<UUID> getAllVersions( final EntityCollectionManager ecm, |
| final Id entityId ) { |
| |
| final List<UUID> versions = new ArrayList<>(); |
| |
| ecm.getVersionsFromMaxToMin(entityId, UUIDUtils.newTimeUUID()) |
| .forEach( mvccLogEntry -> { |
| versions.add(mvccLogEntry.getVersion()); |
| }); |
| |
| return versions; |
| } |
| |
| } |
