You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sn...@apache.org on 2015/02/20 19:52:49 UTC
[1/2] incubator-usergrid git commit: Partial implementation of
re-index collection with option to do it in reverse.
Repository: incubator-usergrid
Updated Branches:
refs/heads/app-rebuild-fix 139984935 -> 2f077f1ba
Partial implementation of re-index collection with option to do it in reverse.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/a704ed73
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/a704ed73
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/a704ed73
Branch: refs/heads/app-rebuild-fix
Commit: a704ed73b8a2b17334ce87b8581d4fe09e63321f
Parents: 1399849
Author: Dave Johnson <dm...@apigee.com>
Authored: Fri Feb 20 12:10:38 2015 -0500
Committer: Dave Johnson <dm...@apigee.com>
Committed: Fri Feb 20 12:10:38 2015 -0500
----------------------------------------------------------------------
.../corepersistence/CpEntityManager.java | 41 ++++--
.../corepersistence/CpEntityManagerFactory.java | 18 ++-
.../usergrid/corepersistence/CpWalker.java | 88 ++++++------
.../usergrid/persistence/EntityManager.java | 3 +
.../persistence/EntityManagerFactory.java | 7 +-
.../cassandra/EntityManagerImpl.java | 5 +
.../PerformanceEntityRebuildIndexTest.java | 136 ++++++++++++++++++-
.../org/apache/usergrid/rest/IndexResource.java | 43 ++++--
.../apache/usergrid/rest/IndexResourceIT.java | 2 +-
9 files changed, 271 insertions(+), 72 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/a704ed73/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
index 7eb9f94..6af0539 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
@@ -32,8 +32,6 @@ import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -105,10 +103,6 @@ import org.apache.usergrid.utils.StringUtils;
import org.apache.usergrid.utils.UUIDUtils;
import com.google.common.base.Preconditions;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.CacheLoader;
-import com.google.common.cache.CacheLoader.InvalidCacheLoadException;
-import com.google.common.cache.LoadingCache;
import com.netflix.hystrix.exception.HystrixRuntimeException;
import com.yammer.metrics.annotation.Metered;
@@ -150,7 +144,6 @@ import static org.apache.usergrid.persistence.Schema.PROPERTY_TYPE;
import static org.apache.usergrid.persistence.Schema.PROPERTY_UUID;
import static org.apache.usergrid.persistence.Schema.TYPE_APPLICATION;
import static org.apache.usergrid.persistence.Schema.TYPE_ENTITY;
-import static org.apache.usergrid.persistence.SimpleEntityRef.getUuid;
import static org.apache.usergrid.persistence.cassandra.ApplicationCF.APPLICATION_AGGREGATE_COUNTERS;
import static org.apache.usergrid.persistence.cassandra.ApplicationCF.ENTITY_COMPOSITE_DICTIONARIES;
import static org.apache.usergrid.persistence.cassandra.ApplicationCF.ENTITY_COUNTERS;
@@ -2746,6 +2739,38 @@ public class CpEntityManager implements EntityManager {
}
+
+ /**
+ * Completely reindex the named collection in the application associated with this EntityManager.
+ */
+ @Override
+ public void reindexCollection(
+ final EntityManagerFactory.ProgressObserver po, String collectionName) throws Exception {
+
+ CpWalker walker = new CpWalker( );
+
+ walker.walkCollections( this, application, collectionName, new CpVisitor() {
+
+ @Override
+ public void visitCollectionEntry( EntityManager em, String collName, Entity entity ) {
+
+ try {
+ em.update( entity );
+ po.onProgress( entity );
+ }
+ catch ( WriteOptimisticVerifyException wo ) {
+ //swallow this, it just means this was already updated, which accomplishes our task. Just ignore.
+ logger.warn( "Someone beat us to updating entity {} in collection {}. Ignoring.", entity.getName(),
+ collName );
+ }
+ catch ( Exception ex ) {
+ logger.error( "Error repersisting entity", ex );
+ }
+ }
+ } );
+ }
+
+
/**
* Completely reindex the application associated with this EntityManager.
*/
@@ -2753,7 +2778,7 @@ public class CpEntityManager implements EntityManager {
CpWalker walker = new CpWalker( );
- walker.walkCollections( this, application, new CpVisitor() {
+ walker.walkCollections( this, application, null, new CpVisitor() {
@Override
public void visitCollectionEntry( EntityManager em, String collName, Entity entity ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/a704ed73/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
index bf85b45..01691d4 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManagerFactory.java
@@ -597,7 +597,8 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
public long performEntityCount() {
//TODO, this really needs to be a task that writes this data somewhere since this will get
//progressively slower as the system expands
- return AllEntitiesInSystemObservable.getAllEntitiesInSystem( managerCache, 1000 ).longCount().toBlocking().last();
+ return AllEntitiesInSystemObservable
+ .getAllEntitiesInSystem( managerCache, 1000 ).longCount().toBlocking().last();
}
@@ -744,8 +745,19 @@ public class CpEntityManagerFactory implements EntityManagerFactory, Application
}
@Override
- public void rebuildCollectionIndex(UUID appId, String collection, ProgressObserver po ) {
- throw new UnsupportedOperationException( "Not supported yet." );
+ public void rebuildCollectionIndex(
+ UUID appId, String collectionName, boolean reverse, ProgressObserver po ) throws Exception {
+
+ EntityManager em = getEntityManager( appId );
+
+ //explicitly invoke create index, we don't know if it exists or not in ES during a rebuild.
+ em.createIndex();
+ Application app = em.getApplication();
+
+ em.reindexCollection( po, collectionName, reverse );
+
+ logger.info("\n\nRebuilt index for application {} id {} collection {}\n",
+ new Object[] { app.getName(), appId, collectionName } );
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/a704ed73/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
index 8a9eed5..6bd90ec 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
@@ -55,15 +55,18 @@ public class CpWalker {
}
- public void walkCollections( final CpEntityManager em, final EntityRef start,
- final CpVisitor visitor ) throws Exception {
+ public void walkCollections(final CpEntityManager em, final EntityRef start,
+ String collectionName, final CpVisitor visitor) throws Exception {
- doWalkCollections( em, new SimpleId( start.getUuid(), start.getType() ), visitor );
+ doWalkCollections( em, collectionName, new SimpleId( start.getUuid(), start.getType() ), visitor );
}
private void doWalkCollections(
- final CpEntityManager em, final Id applicationId, final CpVisitor visitor ) {
+ final CpEntityManager em,
+ final String collectionName,
+ final Id applicationId,
+ final CpVisitor visitor ) {
final ApplicationScope applicationScope = em.getApplicationScope();
@@ -77,10 +80,17 @@ public class CpWalker {
applicationScope.getApplication().getUuid()
} );
- //only search edge types that start with collections
+ // only search edge types that start with collections
+
+ final String edgeType;
+ if ( collectionName != null ) {
+ edgeType = CpNamingUtils.EDGE_COLL_SUFFIX;
+ } else {
+ edgeType = CpNamingUtils.getEdgeTypeFromCollectionName( collectionName );
+ }
Observable<String> edgeTypes = gm.getEdgeTypesFromSource(
- new SimpleSearchEdgeType( applicationId, CpNamingUtils.EDGE_COLL_SUFFIX, null ) );
+ new SimpleSearchEdgeType( applicationId, edgeType, null ) );
edgeTypes.flatMap( new Func1<String, Observable<Edge>>() {
@Override
@@ -88,45 +98,45 @@ public class CpWalker {
logger.debug( "Loading edges of type {} from node {}", edgeType, applicationId );
- return gm.loadEdgesFromSource( new SimpleSearchByEdgeType( applicationId, edgeType, Long.MAX_VALUE,
- SearchByEdgeType.Order.DESCENDING, null ) );
- }
- } )
- //process our edges in parallel for as much efficiency as possible
- .parallel( new Func1<Observable<Edge>, Observable<Edge>>() {
- @Override
- public Observable<Edge> call( final Observable<Edge> edgeObservable ) {
- //visit and update the entity
- return edgeObservable.doOnNext( new Action1<Edge>() {
-
- @Override
- public void call( Edge edge ) {
-
- logger.info( "Re-indexing edge {}", edge );
-
- EntityRef targetNodeEntityRef =
- new SimpleEntityRef( edge.getTargetNode().getType(), edge.getTargetNode().getUuid() );
-
- Entity entity;
- try {
- entity = em.get( targetNodeEntityRef );
- }
- catch ( Exception ex ) {
- logger.error( "Error getting sourceEntity {}:{}, continuing", targetNodeEntityRef.getType(),
- targetNodeEntityRef.getUuid() );
- return;
- }
+ return gm.loadEdgesFromSource( new SimpleSearchByEdgeType(
+ applicationId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, null ) );
+ }
- String collName = CpNamingUtils.getCollectionName( edge.getType() );
+ } ).parallel( new Func1<Observable<Edge>, Observable<Edge>>() {
- visitor.visitCollectionEntry( em, collName, entity );
- }
- } );
+ @Override
+ public Observable<Edge> call( final Observable<Edge> edgeObservable ) { // process edges in parallel
+ return edgeObservable.doOnNext( new Action1<Edge>() { // visit and update then entity
+
+ @Override
+ public void call( Edge edge ) {
+
+ logger.info( "Re-indexing edge {}", edge );
+
+ EntityRef targetNodeEntityRef = new SimpleEntityRef(
+ edge.getTargetNode().getType(),
+ edge.getTargetNode().getUuid() );
+
+ Entity entity;
+ try {
+ entity = em.get( targetNodeEntityRef );
+ }
+ catch ( Exception ex ) {
+ logger.error( "Error getting sourceEntity {}:{}, continuing",
+ targetNodeEntityRef.getType(),
+ targetNodeEntityRef.getUuid() );
+ return;
+ }
+
+ String collName = CpNamingUtils.getCollectionName( edge.getType() );
+ visitor.visitCollectionEntry( em, collName, entity );
+ }
+ } );
}
}, Schedulers.io() )
- //wait for it to complete
+ // wait for it to complete
.toBlocking().lastOrDefault( null ); // end foreach on edges
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/a704ed73/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
index d425b7a..b7ca477 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
@@ -703,6 +703,9 @@ public interface EntityManager {
/** For testing purposes */
public void flushManagerCaches();
+ void reindexCollection(
+ EntityManagerFactory.ProgressObserver po, String collectionName) throws Exception;
+
public void reindex( final EntityManagerFactory.ProgressObserver po ) throws Exception;
/**
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/a704ed73/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java
index 7367eca..6a9095d 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManagerFactory.java
@@ -130,32 +130,29 @@ public interface EntityManagerFactory {
/**
* Return the migration status message
- * @return
*/
public String getMigrateDataStatus();
/**
* Return the current migration version of the system
- * @return
*/
public int getMigrateDataVersion();
/**
* Force the migration version to the specified version
- * @param version
*/
public void setMigrationVersion(int version);
/**
* Perform a realtime count of every entity in the system. This can be slow as it traverses the entire system graph
- * @return
*/
public long performEntityCount();
/** For testing purposes */
public void flushEntityManagerCaches();
- public void rebuildCollectionIndex(UUID appId, String collection, ProgressObserver object);
+ void rebuildCollectionIndex(
+ UUID appId, String collection, boolean reverse, ProgressObserver po) throws Exception;
/**
* Add a new index to the application for scale
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/a704ed73/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerImpl.java b/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerImpl.java
index 6c940a7..d0fe985 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerImpl.java
@@ -2932,6 +2932,11 @@ public class EntityManagerImpl implements EntityManager {
}
@Override
+ public void reindexCollection(EntityManagerFactory.ProgressObserver po, String collectionName) throws Exception {
+ throw new UnsupportedOperationException("Not supported.");
+ }
+
+ @Override
public void reindex(EntityManagerFactory.ProgressObserver po) throws Exception {
throw new UnsupportedOperationException("Not supported.");
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/a704ed73/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java b/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
index f8eec52..e853e94 100644
--- a/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
@@ -88,6 +88,139 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
@Test
+ public void rebuildOneCollectionIndex() throws Exception {
+
+ logger.info("Started rebuildIndex()");
+
+ final EntityManager em = app.getEntityManager();
+
+ // ----------------- create a bunch of entities
+
+ Map<String, Object> entityMap = new HashMap<String, Object>() {{
+ put("key1", 1000 );
+ put("key2", 2000 );
+ put("key3", "Some value");
+ }};
+ Map<String, Object> cat1map = new HashMap<String, Object>() {{
+ put("name", "enzo");
+ put("color", "orange");
+ }};
+ Map<String, Object> cat2map = new HashMap<String, Object>() {{
+ put("name", "marquee");
+ put("color", "grey");
+ }};
+ Map<String, Object> cat3map = new HashMap<String, Object>() {{
+ put("name", "bertha");
+ put("color", "tabby");
+ }};
+
+ Entity cat1 = em.create("cat", cat1map );
+ Entity cat2 = em.create("cat", cat2map );
+ Entity cat3 = em.create("cat", cat3map );
+
+ final long stopTime = System.currentTimeMillis() + RUNTIME_MS;
+
+ List<EntityRef> entityRefs = new ArrayList<EntityRef>();
+ int entityCount = 0;
+ while ( System.currentTimeMillis() < stopTime ) {
+
+ final Entity entity;
+
+ try {
+ entityMap.put("key", entityCount );
+ entity = em.create("testType", entityMap );
+
+ em.refreshIndex();
+
+ em.createConnection(entity, "herds", cat1);
+ em.createConnection(entity, "herds", cat2);
+ em.createConnection(entity, "herds", cat3);
+
+ } catch (Exception ex) {
+ throw new RuntimeException("Error creating entity", ex);
+ }
+
+ entityRefs.add(new SimpleEntityRef( entity.getType(), entity.getUuid() ) );
+ if ( entityCount % 10 == 0 ) {
+ logger.info("Created {} entities", entityCount );
+ }
+
+ entityCount++;
+ try { Thread.sleep( WRITE_DELAY_MS ); } catch (InterruptedException ignored ) {}
+ }
+
+ logger.info("Created {} entities", entityCount);
+ em.refreshIndex();
+
+ // ----------------- test that we can read them, should work fine
+
+ logger.debug("Read the data");
+ readData("testTypes", entityCount );
+
+ // ----------------- delete the system and application indexes
+
+ logger.debug("Deleting app index and system app index");
+ deleteIndex( CpNamingUtils.SYSTEM_APP_ID );
+ deleteIndex( em.getApplicationId() );
+
+ // ----------------- test that we can read them, should fail
+
+ logger.debug("Reading data, should fail this time ");
+ try {
+ readData( "testTypes", entityCount );
+ fail("should have failed to read data");
+
+ } catch (Exception expected) {}
+
+ // ----------------- rebuild index
+
+ logger.debug("Preparing to rebuild all indexes");;
+
+ final String meterName = this.getClass().getSimpleName() + ".rebuildIndex";
+ final Meter meter = registry.meter( meterName );
+
+ EntityManagerFactory.ProgressObserver po = new EntityManagerFactory.ProgressObserver() {
+ int counter = 0;
+
+ @Override
+ public void onProgress( final EntityRef entity ) {
+
+ meter.mark();
+ logger.debug("Indexing {}:{}", entity.getType(), entity.getUuid());
+ if ( !logger.isDebugEnabled() && counter % 100 == 0 ) {
+ logger.info("Reindexed {} entities", counter );
+ }
+ counter++;
+ }
+
+
+
+ @Override
+ public long getWriteDelayTime() {
+ return 0;
+ }
+ };
+
+ try {
+
+ setup.getEmf().rebuildApplicationIndexes( em.getApplicationId(), po );
+
+ reporter.report();
+ registry.remove( meterName );
+ logger.info("Rebuilt index");
+
+ } catch (Exception ex) {
+ logger.error("Error rebuilding index", ex);
+ fail();
+ }
+
+ // ----------------- test that we can read them
+
+ readData( "testTypes", entityCount );
+ }
+
+
+ @Test
public void rebuildIndex() throws Exception {
logger.info("Started rebuildIndex()");
@@ -252,7 +385,8 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
assertEquals( 2000, e.getProperty("key2"));
- Results catResults = em.searchConnectedEntities(e, Query.fromQL("select *").setConnectionType( "herds" ));
+ Results catResults = em.searchConnectedEntities(e,
+ Query.fromQL("select *").setConnectionType( "herds" ));
assertEquals( 3, catResults.size() );
if ( count % 100 == 0 ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/a704ed73/stack/rest/src/main/java/org/apache/usergrid/rest/IndexResource.java
----------------------------------------------------------------------
diff --git a/stack/rest/src/main/java/org/apache/usergrid/rest/IndexResource.java b/stack/rest/src/main/java/org/apache/usergrid/rest/IndexResource.java
index 7b391a1..251e25c 100644
--- a/stack/rest/src/main/java/org/apache/usergrid/rest/IndexResource.java
+++ b/stack/rest/src/main/java/org/apache/usergrid/rest/IndexResource.java
@@ -160,12 +160,13 @@ public class IndexResource extends AbstractContextResource {
@RequireSystemAccess
@PUT
@Path( "rebuild/" + RootResource.APPLICATION_ID_PATH + "/{collectionName}" )
- public JSONWithPadding rebuildIndexes( @Context UriInfo ui,
- @PathParam( "applicationId" ) final String applicationIdStr,
- @PathParam( "collectionName" ) final String collectionName,
- @QueryParam( "callback" ) @DefaultValue( "callback" ) String callback,
- @QueryParam( "delay" ) @DefaultValue( "10" ) final long delay )
- throws Exception {
+ public JSONWithPadding rebuildIndexes(
+ @Context UriInfo ui,
+ @PathParam( "applicationId" ) final String applicationIdStr,
+ @PathParam( "collectionName" ) final String collectionName,
+ @QueryParam( "reverse" ) @DefaultValue( "false" ) final Boolean reverse,
+ @QueryParam( "callback" ) @DefaultValue( "callback" ) String callback,
+ @QueryParam( "delay" ) @DefaultValue( "10" ) final long delay ) throws Exception {
final UUID appId = UUIDUtils.tryExtractUUID( applicationIdStr );
ApiResponse response = createApiResponse();
@@ -175,7 +176,13 @@ public class IndexResource extends AbstractContextResource {
public void run() {
- rebuildCollection( appId, collectionName, delay );
+ try {
+ rebuildCollection( appId, collectionName, reverse, delay );
+ } catch (Exception e) {
+
+ // TODO: handle this in rebuildCollection() instead
+ throw new RuntimeException("Error rebuilding collection");
+ }
}
};
@@ -192,13 +199,14 @@ public class IndexResource extends AbstractContextResource {
@POST
@Path( RootResource.APPLICATION_ID_PATH )
public JSONWithPadding addIndex(@Context UriInfo ui,
- @PathParam( "applicationId" ) final String applicationIdStr,
- Map<String, Object> config,
- @QueryParam( "callback" ) @DefaultValue( "callback" ) String callback) throws Exception{
+ @PathParam( "applicationId" ) final String applicationIdStr,
+ Map<String, Object> config,
+ @QueryParam( "callback" ) @DefaultValue( "callback" ) String callback) throws Exception{
+
ApiResponse response = createApiResponse();
final UUID appId = UUIDUtils.tryExtractUUID(applicationIdStr);
- if(!config.containsKey("replicas") || !config.containsKey("shards") ||
+ if (!config.containsKey("replicas") || !config.containsKey("shards") ||
!(config.get("replicas") instanceof Integer) || !(config.get("shards") instanceof Integer)){
throw new IllegalArgumentException("body must contains 'replicas' of type int and 'shards' of type int");
}
@@ -207,14 +215,20 @@ public class IndexResource extends AbstractContextResource {
throw new IllegalArgumentException("Please add an indexSuffix to your post");
}
- emf.addIndex(appId, config.get("indexSuffix").toString(), (int) config.get("shards"),(int) config.get("replicas"));
+ emf.addIndex(appId, config.get("indexSuffix").toString(),
+ (int) config.get("shards"),(int) config.get("replicas"));
response.setAction("Add index to alias");
return new JSONWithPadding(response, callback);
}
- private void rebuildCollection( final UUID applicationId, final String collectionName, final long delay ) {
+ private void rebuildCollection(
+ final UUID applicationId,
+ final String collectionName,
+ final boolean reverse,
+ final long delay ) throws Exception {
+
EntityManagerFactory.ProgressObserver po = new EntityManagerFactory.ProgressObserver() {
@Override
@@ -227,10 +241,9 @@ public class IndexResource extends AbstractContextResource {
}
};
-
logger.info( "Reindexing for app id: {} and collection {}", applicationId, collectionName );
- emf.rebuildCollectionIndex( applicationId, collectionName, po );
+ emf.rebuildCollectionIndex(applicationId, collectionName, reverse, po);
emf.refreshIndex();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/a704ed73/stack/rest/src/test/java/org/apache/usergrid/rest/IndexResourceIT.java
----------------------------------------------------------------------
diff --git a/stack/rest/src/test/java/org/apache/usergrid/rest/IndexResourceIT.java b/stack/rest/src/test/java/org/apache/usergrid/rest/IndexResourceIT.java
index 9bf0813..7f8d085 100644
--- a/stack/rest/src/test/java/org/apache/usergrid/rest/IndexResourceIT.java
+++ b/stack/rest/src/test/java/org/apache/usergrid/rest/IndexResourceIT.java
@@ -68,7 +68,7 @@ public class IndexResourceIT extends AbstractRestIT {
.queryParam("access_token", superToken)
.accept(MediaType.APPLICATION_JSON).type(MediaType.APPLICATION_JSON_TYPE)
.post(String.class, data));
- }catch (Exception e){
+ } catch (Exception e) {
LOG.error("failed", e);
fail(e.toString());
}
[2/2] incubator-usergrid git commit: reindexCollection now
implemented w/test in the Core module. Still needs a REST level test.
Posted by sn...@apache.org.
reindexCollection now implemented w/test in the Core module. Still needs a REST level test.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/2f077f1b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/2f077f1b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/2f077f1b
Branch: refs/heads/app-rebuild-fix
Commit: 2f077f1bada2cb79db5dc9af19f101b668f92b9c
Parents: a704ed7
Author: Dave Johnson <dm...@apigee.com>
Authored: Fri Feb 20 13:52:21 2015 -0500
Committer: Dave Johnson <dm...@apigee.com>
Committed: Fri Feb 20 13:52:21 2015 -0500
----------------------------------------------------------------------
.../corepersistence/CpEntityManager.java | 37 ++++----
.../usergrid/corepersistence/CpWalker.java | 20 +++-
.../usergrid/persistence/EntityManager.java | 2 +-
.../cassandra/EntityManagerFactoryImpl.java | 11 +--
.../cassandra/EntityManagerImpl.java | 4 +-
.../PerformanceEntityRebuildIndexTest.java | 97 ++++++++++++--------
.../management/importer/ImportCollectionIT.java | 6 +-
7 files changed, 105 insertions(+), 72 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2f077f1b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
index 6af0539..721ac80 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpEntityManager.java
@@ -2745,11 +2745,12 @@ public class CpEntityManager implements EntityManager {
*/
@Override
public void reindexCollection(
- final EntityManagerFactory.ProgressObserver po, String collectionName) throws Exception {
+ final EntityManagerFactory.ProgressObserver po, String collectionName, boolean reverse) throws Exception {
CpWalker walker = new CpWalker( );
- walker.walkCollections( this, application, collectionName, new CpVisitor() {
+ walker.walkCollections(
+ this, application, collectionName, reverse, new CpVisitor() {
@Override
public void visitCollectionEntry( EntityManager em, String collName, Entity entity ) {
@@ -2759,9 +2760,9 @@ public class CpEntityManager implements EntityManager {
po.onProgress( entity );
}
catch ( WriteOptimisticVerifyException wo ) {
- //swallow this, it just means this was already updated, which accomplishes our task. Just ignore.
- logger.warn( "Someone beat us to updating entity {} in collection {}. Ignoring.", entity.getName(),
- collName );
+ // swallow this, it just means this was already updated, which accomplishes our task
+ logger.warn( "Someone beat us to updating entity {} in collection {}. Ignoring.",
+ entity.getName(), collName );
}
catch ( Exception ex ) {
logger.error( "Error repersisting entity", ex );
@@ -2778,23 +2779,23 @@ public class CpEntityManager implements EntityManager {
CpWalker walker = new CpWalker( );
- walker.walkCollections( this, application, null, new CpVisitor() {
+ walker.walkCollections( this, application, null, false, new CpVisitor() {
@Override
public void visitCollectionEntry( EntityManager em, String collName, Entity entity ) {
- try {
- em.update( entity );
- po.onProgress( entity );
- }
- catch ( WriteOptimisticVerifyException wo ) {
- //swallow this, it just means this was already updated, which accomplishes our task. Just ignore.
- logger.warn( "Someone beat us to updating entity {} in collection {}. Ignoring.", entity.getName(),
- collName );
- }
- catch ( Exception ex ) {
- logger.error( "Error repersisting entity", ex );
- }
+ try {
+ em.update( entity );
+ po.onProgress( entity );
+ }
+ catch ( WriteOptimisticVerifyException wo ) {
+ //swallow this, it just means this was already updated, which accomplishes our task.
+ logger.warn( "Someone beat us to updating entity {} in collection {}. Ignoring.",
+ entity.getName(), collName );
+ }
+ catch ( Exception ex ) {
+ logger.error( "Error repersisting entity", ex );
+ }
}
} );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2f077f1b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
index 6bd90ec..fab7e16 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpWalker.java
@@ -56,15 +56,17 @@ public class CpWalker {
public void walkCollections(final CpEntityManager em, final EntityRef start,
- String collectionName, final CpVisitor visitor) throws Exception {
+ String collectionName, boolean reverse, final CpVisitor visitor) throws Exception {
- doWalkCollections( em, collectionName, new SimpleId( start.getUuid(), start.getType() ), visitor );
+ doWalkCollections(
+ em, collectionName, reverse, new SimpleId( start.getUuid(), start.getType() ), visitor );
}
private void doWalkCollections(
final CpEntityManager em,
final String collectionName,
+ final boolean reverse,
final Id applicationId,
final CpVisitor visitor ) {
@@ -80,12 +82,20 @@ public class CpWalker {
applicationScope.getApplication().getUuid()
} );
- // only search edge types that start with collections
+ final SearchByEdgeType.Order order;
+ if ( reverse ) {
+ order = SearchByEdgeType.Order.ASCENDING;
+ } else {
+ order = SearchByEdgeType.Order.DESCENDING;
+ }
final String edgeType;
- if ( collectionName != null ) {
+ if ( collectionName == null ) {
+ // only search edge types that end with collections suffix
edgeType = CpNamingUtils.EDGE_COLL_SUFFIX;
+
} else {
+ // only search edges to one collection
edgeType = CpNamingUtils.getEdgeTypeFromCollectionName( collectionName );
}
@@ -99,7 +109,7 @@ public class CpWalker {
logger.debug( "Loading edges of type {} from node {}", edgeType, applicationId );
return gm.loadEdgesFromSource( new SimpleSearchByEdgeType(
- applicationId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, null ) );
+ applicationId, edgeType, Long.MAX_VALUE, order , null ) );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2f077f1b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
index b7ca477..4054d77 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/EntityManager.java
@@ -704,7 +704,7 @@ public interface EntityManager {
public void flushManagerCaches();
void reindexCollection(
- EntityManagerFactory.ProgressObserver po, String collectionName) throws Exception;
+ EntityManagerFactory.ProgressObserver po, String collectionName, boolean reverse) throws Exception;
public void reindex( final EntityManagerFactory.ProgressObserver po ) throws Exception;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2f077f1b/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerFactoryImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerFactoryImpl.java b/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerFactoryImpl.java
index b7808d8..535a14e 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerFactoryImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerFactoryImpl.java
@@ -423,6 +423,11 @@ public class EntityManagerFactoryImpl implements EntityManagerFactory, Applicati
}
@Override
+ public void rebuildCollectionIndex(UUID appId, String collection, boolean reverse, ProgressObserver po) throws Exception {
+ throw new UnsupportedOperationException("Not supported.");
+ }
+
+ @Override
public void rebuildInternalIndexes(ProgressObserver po) throws Exception {
throw new UnsupportedOperationException("Not supported.");
}
@@ -461,12 +466,6 @@ public class EntityManagerFactoryImpl implements EntityManagerFactory, Applicati
throw new UnsupportedOperationException("Not supported in v1");
}
-
- @Override
- public void rebuildCollectionIndex(UUID appId, String collection, ProgressObserver po) {
- throw new UnsupportedOperationException("Not supported.");
- }
-
@Override
public void addIndex(UUID appId, String suffix,final int shards,final int replicas) {
throw new UnsupportedOperationException("Not supported in v1");
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2f077f1b/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerImpl.java b/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerImpl.java
index d0fe985..be43920 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/EntityManagerImpl.java
@@ -2932,7 +2932,9 @@ public class EntityManagerImpl implements EntityManager {
}
@Override
- public void reindexCollection(EntityManagerFactory.ProgressObserver po, String collectionName) throws Exception {
+ public void reindexCollection(
+ EntityManagerFactory.ProgressObserver po, String collectionName, boolean reverse) throws Exception {
+
throw new UnsupportedOperationException("Not supported.");
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2f077f1b/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java b/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
index e853e94..52b6fe4 100644
--- a/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang.RandomStringUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -60,7 +61,7 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
private static final MetricRegistry registry = new MetricRegistry();
private Slf4jReporter reporter;
- private static final long RUNTIME_MS = TimeUnit.SECONDS.toMillis( 3 );
+ private static final long RUNTIME_MS = TimeUnit.SECONDS.toMillis( 10 );
private static final long WRITE_DELAY_MS = 10;
@@ -92,7 +93,10 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
logger.info("Started rebuildIndex()");
- final EntityManager em = app.getEntityManager();
+ String rand = RandomStringUtils.randomAlphanumeric(5);
+ final UUID appId = setup.createApplication("org_" + rand, "app_" + rand);
+
+ final EntityManager em = setup.getEmf().getEntityManager( appId );
// ----------------- create a bunch of entities
@@ -101,40 +105,50 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
put("key2", 2000 );
put("key3", "Some value");
}};
- Map<String, Object> cat1map = new HashMap<String, Object>() {{
- put("name", "enzo");
- put("color", "orange");
- }};
- Map<String, Object> cat2map = new HashMap<String, Object>() {{
- put("name", "marquee");
- put("color", "grey");
- }};
- Map<String, Object> cat3map = new HashMap<String, Object>() {{
- put("name", "bertha");
- put("color", "tabby");
- }};
- Entity cat1 = em.create("cat", cat1map );
- Entity cat2 = em.create("cat", cat2map );
- Entity cat3 = em.create("cat", cat3map );
+// Map<String, Object> cat1map = new HashMap<String, Object>() {{
+// put("name", "enzo");
+// put("color", "orange");
+// }};
+// Map<String, Object> cat2map = new HashMap<String, Object>() {{
+// put("name", "marquee");
+// put("color", "grey");
+// }};
+// Map<String, Object> cat3map = new HashMap<String, Object>() {{
+// put("name", "bertha");
+// put("color", "tabby");
+// }};
+//
+// Entity cat1 = em.create("cat", cat1map );
+// Entity cat2 = em.create("cat", cat2map );
+// Entity cat3 = em.create("cat", cat3map );
final long stopTime = System.currentTimeMillis() + RUNTIME_MS;
List<EntityRef> entityRefs = new ArrayList<EntityRef>();
int entityCount = 0;
+ int herderCount = 0;
+ int shepardCount = 0;
while ( System.currentTimeMillis() < stopTime ) {
final Entity entity;
try {
entityMap.put("key", entityCount );
- entity = em.create("testType", entityMap );
+
+ if ( entityCount % 2 == 0 ) {
+ entity = em.create("catherder", entityMap);
+ herderCount++;
+ } else {
+ entity = em.create("catshepard", entityMap);
+ shepardCount++;
+ }
em.refreshIndex();
- em.createConnection(entity, "herds", cat1);
- em.createConnection(entity, "herds", cat2);
- em.createConnection(entity, "herds", cat3);
+// em.createConnection(entity, "herds", cat1);
+// em.createConnection(entity, "herds", cat2);
+// em.createConnection(entity, "herds", cat3);
} catch (Exception ex) {
throw new RuntimeException("Error creating entity", ex);
@@ -155,24 +169,25 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
// ----------------- test that we can read them, should work fine
logger.debug("Read the data");
- readData("testTypes", entityCount );
+ readData( em, "catherders", herderCount, 0);
+ readData( em, "catshepards", shepardCount, 0);
// ----------------- delete the system and application indexes
- logger.debug("Deleting app index and system app index");
- deleteIndex( CpNamingUtils.SYSTEM_APP_ID );
+ logger.debug("Deleting app index index");
+ //deleteIndex( CpNamingUtils.SYSTEM_APP_ID );
deleteIndex( em.getApplicationId() );
// ----------------- test that we can read them, should fail
logger.debug("Reading data, should fail this time ");
try {
- readData( "testTypes", entityCount );
+ readData( em, "testTypes", entityCount, 0 );
fail("should have failed to read data");
} catch (Exception expected) {}
- // ----------------- rebuild index
+ // ----------------- rebuild index for catherders only
logger.debug("Preparing to rebuild all indexes");;
@@ -203,7 +218,7 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
try {
- setup.getEmf().rebuildApplicationIndexes( em.getApplicationId(), po );
+ setup.getEmf().rebuildCollectionIndex( em.getApplicationId(), "catherders", false, po );
reporter.report();
registry.remove( meterName );
@@ -214,9 +229,10 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
fail();
}
- // ----------------- test that we can read them
+ // ----------------- test that we can read the catherder collection and not the catshepard
- readData( "testTypes", entityCount );
+ readData( em, "catherders", herderCount, 0 );
+ readData( em, "catshepards", 0, 0 );
}
@@ -225,7 +241,10 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
logger.info("Started rebuildIndex()");
- final EntityManager em = app.getEntityManager();
+ String rand = RandomStringUtils.randomAlphanumeric(5);
+ final UUID appId = setup.createApplication("org_" + rand, "app_" + rand);
+
+ final EntityManager em = setup.getEmf().getEntityManager(appId);
// ----------------- create a bunch of entities
@@ -288,19 +307,19 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
// ----------------- test that we can read them, should work fine
logger.debug("Read the data");
- readData("testTypes", entityCount );
+ readData( em, "testTypes", entityCount, 3 );
// ----------------- delete the system and application indexes
logger.debug("Deleting app index and system app index");
- deleteIndex( CpNamingUtils.SYSTEM_APP_ID );
+ //deleteIndex( CpNamingUtils.SYSTEM_APP_ID );
deleteIndex( em.getApplicationId() );
// ----------------- test that we can read them, should fail
logger.debug("Reading data, should fail this time ");
try {
- readData( "testTypes", entityCount );
+ readData( em, "testTypes", entityCount, 3 );
fail("should have failed to read data");
} catch (Exception expected) {}
@@ -349,7 +368,7 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
// ----------------- test that we can read them
- readData( "testTypes", entityCount );
+ readData( em, "testTypes", entityCount, 3 );
}
/**
@@ -369,9 +388,9 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
}
- private int readData( String collectionName, int expected ) throws Exception {
+ private int readData( EntityManager em,
+ String collectionName, int expectedEntities, int expectedConnections ) throws Exception {
- EntityManager em = app.getEntityManager();
em.refreshIndex();
Query q = Query.fromQL("select * where key1=1000");
@@ -387,7 +406,7 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
Results catResults = em.searchConnectedEntities(e,
Query.fromQL("select *").setConnectionType( "herds" ));
- assertEquals( 3, catResults.size() );
+ assertEquals( expectedConnections, catResults.size() );
if ( count % 100 == 0 ) {
logger.info( "read {} entities", count);
@@ -405,9 +424,9 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
}
}
- if ( expected != -1 && expected != count ) {
+ if ( expectedEntities != -1 && expectedEntities != count ) {
throw new RuntimeException("Did not get expected "
- + expected + " entities, instead got " + count );
+ + expectedEntities + " entities, instead got " + count );
}
return count;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2f077f1b/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportCollectionIT.java
----------------------------------------------------------------------
diff --git a/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportCollectionIT.java b/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportCollectionIT.java
index e566512..7fcfc57 100644
--- a/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportCollectionIT.java
+++ b/stack/services/src/test/java/org/apache/usergrid/management/importer/ImportCollectionIT.java
@@ -110,7 +110,8 @@ public class ImportCollectionIT {
bucketPrefix = System.getProperty( "bucketName" );
// start the scheduler after we're all set up
- JobSchedulerService jobScheduler = ConcurrentProcessSingleton.getInstance().getSpringResource().getBean( JobSchedulerService.class );
+ JobSchedulerService jobScheduler = ConcurrentProcessSingleton.getInstance()
+ .getSpringResource().getBean( JobSchedulerService.class );
if ( jobScheduler.state() != Service.State.RUNNING ) {
jobScheduler.startAsync();
jobScheduler.awaitRunning();
@@ -481,7 +482,8 @@ public class ImportCollectionIT {
ImportService importService = setup.getImportService();
- Import importEntity = importService.schedule(em.getApplication().getUuid(), new HashMap<String, Object>() {{
+ Import importEntity = importService.schedule(em.getApplication().getUuid(),
+ new HashMap<String, Object>() {{
put( "path", organization.getName() + em.getApplication().getName() );
put( "organizationId", organization.getUuid() );
put( "applicationId", em.getApplication().getUuid() );