diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/AlfrescoCoreAdminHandler.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/AlfrescoCoreAdminHandler.java index 73199ca57..2496b3134 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/AlfrescoCoreAdminHandler.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/AlfrescoCoreAdminHandler.java @@ -259,6 +259,12 @@ public class AlfrescoCoreAdminHandler extends CoreAdminHandler case "ACLTXREPORT": actionACLTXREPORT(rsp, params, cname); break; + case "RANGECHECK": + rangeCheck(rsp, cname); + break; + case "EXPAND": + expand(rsp, params, cname); + break; case "REPORT": actionREPORT(rsp, params, cname); break; @@ -368,7 +374,7 @@ public class AlfrescoCoreAdminHandler extends CoreAdminHandler String shardIds = params.get("shardIds"); Properties properties = extractCustomProperties(params); - return newCore(coreName,numShards,storeRef,templateName,replicationFactor,nodeInstance,numNodes,shardIds,properties,rsp); + return newCore(coreName, numShards, storeRef, templateName, replicationFactor, nodeInstance, numNodes, shardIds, properties, rsp); } private boolean newDefaultCore(SolrQueryRequest req, SolrQueryResponse rsp) { @@ -384,7 +390,7 @@ public class AlfrescoCoreAdminHandler extends CoreAdminHandler storeRef = new StoreRef(store); } - return newDefaultCore(coreName,storeRef,templateName,extraProperties,rsp); + return newDefaultCore(coreName, storeRef, templateName, extraProperties, rsp); } protected boolean newDefaultCore(String coreName, StoreRef storeRef, String templateName, Properties extraProperties, SolrQueryResponse rsp) @@ -502,7 +508,6 @@ public class AlfrescoCoreAdminHandler extends CoreAdminHandler /** * @param rsp - * @param params * @param storeRef * @param template * @param coreName @@ -781,7 +786,7 @@ public class AlfrescoCoreAdminHandler extends CoreAdminHandler AclTracker tracker = trackerRegistry.getTrackerForCore(cname, AclTracker.class); Long acltxid = Long.valueOf(params.get(ARG_ACLTXID)); NamedList report = new SimpleOrderedMap(); - report.add(cname, buildAclTxReport(getTrackerRegistry(), informationServers.get(cname),cname, tracker, acltxid)); + report.add(cname, buildAclTxReport(getTrackerRegistry(), informationServers.get(cname), cname, tracker, acltxid)); rsp.add("report", report); } else @@ -838,6 +843,170 @@ public class AlfrescoCoreAdminHandler extends CoreAdminHandler } } + private DocRouter getDocRouter(String cname) + { + Collection trackers = trackerRegistry.getTrackersForCore(cname); + MetadataTracker metadataTracker = null; + for(Tracker tracker : trackers) + { + if(tracker instanceof MetadataTracker) + { + metadataTracker = (MetadataTracker)tracker; + } + } + + DocRouter docRouter = metadataTracker.getDocRouter(); + return docRouter; + } + + + private void rangeCheck(SolrQueryResponse rsp,String cname) throws IOException + { + InformationServer informationServer = informationServers.get(cname); + + DocRouter docRouter = getDocRouter(cname); + + if(docRouter instanceof DBIDRangeRouter) { + + DBIDRangeRouter dbidRangeRouter = (DBIDRangeRouter) docRouter; + + if(!dbidRangeRouter.getInitialized()) + { + rsp.add("expand", 0); + rsp.add("exception", "DBIDRangeRouter not initialized yet."); + return; + } + + long startRange = dbidRangeRouter.getStartRange(); + long endRange = dbidRangeRouter.getEndRange(); + + long maxNodeId = informationServer.maxNodeId(); + long minNodeId = informationServer.minNodeId(); + long nodeCount = informationServer.nodeCount(); + + long bestGuess = -1; // -1 means expansion cannot be done. Either because expansion + // has already happened or we're above safe range + + long range = endRange - startRange; // We want this many nodes on the server + + long midpoint = startRange + ((long) (range * .5)); + + long safe = startRange + ((long) (range * .75)); + + long offset = maxNodeId-startRange; + + double density = 0; + + if(offset > 0) { + density = ((double)nodeCount) / ((double)offset); // This is how dense we are so far. + } + + if (!dbidRangeRouter.getExpanded()) + { + if(maxNodeId <= safe) + { + if (maxNodeId >= midpoint) + { + if(density >= 1) + { + //This is fully dense shard. I'm not sure if it's possible to have more nodes on the shards + //then the offset, but if it does happen don't expand. + bestGuess=0; + } + else + { + double multiplier = 1/density; + bestGuess = (long)(range*multiplier)-range; // This is how much to add + } + } + else + { + bestGuess = 0; // We're below the midpoint so it's to early to make a guess. + } + } + } + + rsp.add("start", startRange); + rsp.add("end", endRange); + rsp.add("nodeCount", nodeCount); + rsp.add("minDbid", minNodeId); + rsp.add("maxDbid", maxNodeId); + rsp.add("density", Math.abs(density)); + rsp.add("expand", bestGuess); + rsp.add("expanded", dbidRangeRouter.getExpanded()); + } else { + rsp.add("expand", -1); + rsp.add("exception", "ERROR: Wrong document router type:"+docRouter.getClass().getSimpleName()); + return; + } + } + + private synchronized void expand(SolrQueryResponse rsp, SolrParams params, String cname) + throws IOException + { + InformationServer informationServer = informationServers.get(cname); + DocRouter docRouter = getDocRouter(cname); + + if(docRouter instanceof DBIDRangeRouter) + { + long expansion = Long.parseLong(params.get("add")); + DBIDRangeRouter dbidRangeRouter = (DBIDRangeRouter)docRouter; + + if(!dbidRangeRouter.getInitialized()) + { + rsp.add("expand", -1); + rsp.add("exception", "DBIDRangeRouter not initialized yet."); + return; + } + + if(dbidRangeRouter.getExpanded()) + { + rsp.add("expand", -1); + rsp.add("exception", "dbid range has already been expanded."); + return; + } + + long currentEndRange = dbidRangeRouter.getEndRange(); + long startRange = dbidRangeRouter.getStartRange(); + long maxNodeId = informationServer.maxNodeId(); + + long range = currentEndRange-startRange; + long safe = startRange + ((long) (range * .75)); + + if(maxNodeId > safe) + { + rsp.add("expand", -1); + rsp.add("exception", "Expansion cannot occur if max DBID in the index is more then 75% of range."); + return; + } + + long newEndRange = expansion+dbidRangeRouter.getEndRange(); + try + { + informationServer.capIndex(newEndRange); + informationServer.hardCommit(); + dbidRangeRouter.setEndRange(newEndRange); + dbidRangeRouter.setExpanded(true); + assert newEndRange == dbidRangeRouter.getEndRange(); + rsp.add("expand", dbidRangeRouter.getEndRange()); + return; + } + catch(Throwable t) + { + rsp.add("expand", -1); + rsp.add("exception", t.getMessage()); + log.error("exception expanding", t); + return; + } + } + else + { + rsp.add("expand", -1); + rsp.add("exception", "Wrong document router type:"+docRouter.getClass().getSimpleName()); + return; + } + } + private void actionNODEREPORTS(SolrQueryResponse rsp, SolrParams params, String cname) throws IOException, JSONException { diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/InformationServer.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/InformationServer.java index b9813bcc9..13d4173f3 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/InformationServer.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/InformationServer.java @@ -58,6 +58,8 @@ public interface InformationServer extends InformationServerCollectionProvider void commit() throws IOException; + void hardCommit() throws IOException; + boolean commit(boolean openSearcher) throws IOException; void indexAclTransaction(AclChangeSet changeSet, boolean overwrite) throws IOException; @@ -72,6 +74,18 @@ public interface InformationServer extends InformationServerCollectionProvider void deleteByNodeId(Long nodeId) throws IOException; + void capIndex(long nodeId) throws IOException; + + long getIndexCap() throws IOException; + + long nodeCount() throws IOException; + + long maxNodeId() throws IOException; + + long minNodeId() throws IOException; + + void maintainCap(long nodeId) throws Exception; + void indexNode(Node node, boolean overwrite) throws IOException, AuthenticationException, JSONException; void indexNodes(List nodes, boolean overwrite, boolean cascade) throws IOException, AuthenticationException, JSONException; diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/SolrInformationServer.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/SolrInformationServer.java index c911330fe..04ee580be 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/SolrInformationServer.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/SolrInformationServer.java @@ -18,56 +18,6 @@ */ package org.alfresco.solr; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_ACLID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_ACLTXCOMMITTIME; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_ACLTXID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_ANAME; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_ANCESTOR; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_APATH; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_ASPECT; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_ASSOCTYPEQNAME; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_CASCADE_FLAG; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_DBID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_DENIED; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_DOC_TYPE; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_EXCEPTION_MESSAGE; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_EXCEPTION_STACK; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_FIELDS; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_FTSSTATUS; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_GEO; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_INACLTXID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_INTXID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_ISNODE; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_LID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_NPATH; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_NULLPROPERTIES; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_OWNER; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_PARENT; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_PARENT_ASSOC_CRC; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_PATH; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_PNAME; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_PRIMARYASSOCQNAME; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_PRIMARYASSOCTYPEQNAME; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_PRIMARYPARENT; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_PROPERTIES; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_QNAME; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_READER; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_SITE; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_SOLR4_ID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_S_ACLTXCOMMITTIME; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_S_ACLTXID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_S_INACLTXID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_S_INTXID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_S_TXCOMMITTIME; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_S_TXID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_TAG; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_TAG_SUGGEST; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_TENANT; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_TXCOMMITTIME; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_TXID; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_TYPE; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_VERSION; - import java.io.File; import java.io.IOException; import java.io.InputStream; @@ -199,6 +149,8 @@ import org.slf4j.LoggerFactory; import org.springframework.extensions.surf.util.I18NUtil; import org.springframework.util.FileCopyUtils; +import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.*; + /** * This is the Solr4 implementation of the information server (index). * @author Ahmed Owian @@ -232,6 +184,8 @@ public class SolrInformationServer implements InformationServer public static final String SOLR_PORT = "solr.port"; public static final String SOLR_BASEURL = "solr.baseurl"; + public static String indexCapId = "TRACKER!STATE!CAP"; + private static final Pattern CAPTURE_SITE = Pattern.compile("^/\\{http\\://www\\.alfresco\\.org/model/application/1\\.0\\}company\\_home/\\{http\\://www\\.alfresco\\.org/model/site/1\\.0\\}sites/\\{http\\://www\\.alfresco\\.org/model/content/1\\.0}([^/]*)/.*" ); private static final Pattern CAPTURE_TAG = Pattern.compile("^/\\{http\\://www\\.alfresco\\.org/model/content/1\\.0\\}taggable/\\{http\\://www\\.alfresco\\.org/model/content/1\\.0\\}([^/]*)/\\{\\}member"); @@ -956,6 +910,37 @@ public class SolrInformationServer implements InformationServer } } + @Override + public void hardCommit() throws IOException + { + // avoid multiple commits and warming searchers + commitAndRollbackLock.writeLock().lock(); + try + { + SolrQueryRequest request = null; + UpdateRequestProcessor processor = null; + try + { + request = getLocalSolrQueryRequest(); + processor = this.core.getUpdateProcessingChain(null).createProcessor(request, new SolrQueryResponse()); + CommitUpdateCommand commitUpdateCommand = new CommitUpdateCommand(request, false); + commitUpdateCommand.openSearcher = false; + commitUpdateCommand.softCommit = false; + commitUpdateCommand.waitSearcher = false; + processor.processCommit(commitUpdateCommand); + } + finally + { + if(processor != null) {processor.finish();} + if(request != null) {request.close();} + } + } + finally + { + commitAndRollbackLock.writeLock().unlock(); + } + } + public boolean commit(boolean openSearcher) throws IOException { canUpdate(); @@ -1674,6 +1659,106 @@ public class SolrInformationServer implements InformationServer } } + public void capIndex(long dbid) throws IOException + { + SolrQueryRequest request = null; + UpdateRequestProcessor processor = null; + + try + { + request = getLocalSolrQueryRequest(); + processor = this.core.getUpdateProcessingChain(null).createProcessor(request, new SolrQueryResponse()); + AddUpdateCommand cmd = new AddUpdateCommand(request); + cmd.overwrite = true; + SolrInputDocument input = new SolrInputDocument(); + input.addField(FIELD_SOLR4_ID, indexCapId); + input.addField(FIELD_VERSION, 0); + input.addField(FIELD_DBID, -dbid); //Making this negative to ensure it is never confused with node DBID + input.addField(FIELD_DOC_TYPE, DOC_TYPE_STATE); + cmd.solrDoc = input; + processor.processAdd(cmd); + } + finally + { + if(processor != null) {processor.finish();} + if(request != null) {request.close();} + } + } + + public void maintainCap(long dbid) throws IOException { + String deleteByQuery = FIELD_DBID+":{"+dbid+" TO *}"; + deleteByQuery(deleteByQuery); + } + + public long nodeCount() + { + return getDocListSize(FIELD_DOC_TYPE+":"+DOC_TYPE_NODE); + } + + public long maxNodeId() { + return topNodeId("desc"); + } + + public long minNodeId() { + return topNodeId("asc"); + } + + private long topNodeId(String sortDir) + { + SolrQueryRequest request = null; + try + { + request = this.getLocalSolrQueryRequest(); + ModifiableSolrParams params = new ModifiableSolrParams(request.getParams()); + params.set("q", FIELD_DOC_TYPE+":"+DOC_TYPE_NODE); + // Sets the rows to zero, because we actually just want the count + params.set("rows", 1); + params.set("sort", FIELD_DBID + " "+sortDir); + params.set("fl", FIELD_DBID); + SolrDocumentList docs = cloud.getSolrDocumentList(nativeRequestHandler, request, params); + Iterator it = docs.iterator(); + if(it.hasNext()) { + SolrDocument doc = it.next(); + long dbid = getFieldValueLong(doc, FIELD_DBID); + return dbid; + } else { + return 0; + } + } + finally + { + if (request != null) { request.close(); } + } + } + + public long getIndexCap() + { + SolrQueryRequest request = null; + try + { + request = this.getLocalSolrQueryRequest(); + ModifiableSolrParams params = new ModifiableSolrParams(request.getParams()); + params.set("q", FIELD_SOLR4_ID+":"+indexCapId); + params.set("rows", 1); + params.set("fl", FIELD_DBID); + SolrDocumentList docs = cloud.getSolrDocumentList(nativeRequestHandler, request, params); + Iterator it = docs.iterator(); + if(it.hasNext()) { + SolrDocument doc = it.next(); + long dbid = getFieldValueLong(doc, FIELD_DBID); + return Math.abs(dbid); + } else { + return -1; + } + } + finally + { + if (request != null) { request.close(); } + } + } + + + @Override public void indexNode(Node node, boolean overwrite) throws IOException, AuthenticationException, JSONException { diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CommitTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CommitTracker.java index b1569502b..ba0c2b6ff 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CommitTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CommitTracker.java @@ -80,7 +80,7 @@ public class CommitTracker extends AbstractTracker lastSearcherOpened = lastCommit = System.currentTimeMillis(); } - public boolean hasMaintenance() + public boolean hasMaintenance() throws Exception { return (metadataTracker.hasMaintenance() || aclTracker.hasMaintenance()); } diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/DBIDRangeRouter.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/DBIDRangeRouter.java index 3034164ff..0e42a423d 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/DBIDRangeRouter.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/DBIDRangeRouter.java @@ -22,6 +22,9 @@ import org.apache.solr.common.util.Hash; import org.alfresco.solr.client.Node; import org.alfresco.solr.client.Acl; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + /* * @author Joel @@ -29,13 +32,42 @@ import org.alfresco.solr.client.Acl; public class DBIDRangeRouter implements DocRouter { - private long startRange; - private long endRange; + private AtomicLong expandableRange; + private AtomicBoolean expanded = new AtomicBoolean(false); + private AtomicBoolean initialized = new AtomicBoolean(false); public DBIDRangeRouter(long startRange, long endRange) { this.startRange = startRange; - this.endRange = endRange; + this.expandableRange = new AtomicLong(endRange); + } + + public void setEndRange(long endRange) { + expandableRange.set(endRange); + } + + public void setExpanded(boolean expanded) { + this.expanded.set(expanded); + } + + public void setInitialized(boolean initialized) { + this.initialized.set(initialized); + } + + public boolean getInitialized() { + return this.initialized.get(); + } + + public long getEndRange() { + return expandableRange.longValue(); + } + + public long getStartRange() { + return this.startRange; + } + + public boolean getExpanded() { + return this.expanded.get(); } @Override @@ -47,7 +79,7 @@ public class DBIDRangeRouter implements DocRouter @Override public boolean routeNode(int shardCount, int shardInstance, Node node) { long dbid = node.getId(); - if(dbid >= startRange && dbid < endRange) { + if(dbid >= startRange && dbid < expandableRange.longValue()) { return true; } else { return false; diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/DocRouterFactory.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/DocRouterFactory.java index c852412ca..bfe9a28f2 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/DocRouterFactory.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/DocRouterFactory.java @@ -23,7 +23,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Properties; -import java.util.Set; /* * @author Joel @@ -40,12 +39,15 @@ public class DocRouterFactory log.info("Sharding via DB_ID"); return new DBIDRouter(); case DB_ID_RANGE: - String range = properties.getProperty("shard.range"); - String[] rangeParts = range.split("-"); - long startRange = Long.parseLong(rangeParts[0].trim()); - long endRange = Long.parseLong(rangeParts[1].trim()); - log.info("Sharding via DB_ID_RANGE"); - return new DBIDRangeRouter(startRange, endRange); + // + if(properties.containsKey("shard.range")) + { + log.info("Sharding via DB_ID_RANGE"); + String[] pair =properties.getProperty("shard.range").split("-"); + long start = Long.parseLong(pair[0]); + long end = Long.parseLong(pair[1]); + return new DBIDRangeRouter(start, end); + } case ACL_ID: log.info("Sharding via ACL_ID"); return new ACLIDMurmurRouter(); diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java index 091b1bd6e..2bf6812be 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java @@ -80,7 +80,6 @@ public class MetadataTracker extends AbstractTracker implements Tracker InformationServer informationServer) { super(p, client, coreName, informationServer, Tracker.Type.MetaData); - //System.out.println("####### MetadatTracker() ########"); transactionDocsBatchSize = Integer.parseInt(p.getProperty("alfresco.transactionDocsBatchSize", "100")); shardMethod = p.getProperty("shard.method", SHARD_METHOD_DBID); String shardKey = p.getProperty("shard.key"); @@ -98,6 +97,10 @@ public class MetadataTracker extends AbstractTracker implements Tracker super(Tracker.Type.MetaData); } + public DocRouter getDocRouter() { + return this.docRouter; + } + @Override protected void doTrack() throws AuthenticationException, IOException, JSONException, EncoderException { @@ -120,8 +123,8 @@ public class MetadataTracker extends AbstractTracker implements Tracker indexNodes(); } - public boolean hasMaintenance() { - return transactionsToReindex.size() > 0 || + public boolean hasMaintenance() throws Exception { + return transactionsToReindex.size() > 0 || transactionsToIndex.size() > 0 || transactionsToPurge.size() > 0 || nodesToReindex.size() > 0 || @@ -130,6 +133,8 @@ public class MetadataTracker extends AbstractTracker implements Tracker queriesToReindex.size() > 0; } + + private void trackRepository() throws IOException, AuthenticationException, JSONException, EncoderException { //System.out.println("########################### MetadataTracker.trackRepository ##########"); @@ -152,6 +157,21 @@ public class MetadataTracker extends AbstractTracker implements Tracker checkRepoAndIndexConsistency(state); } + if(docRouter instanceof DBIDRangeRouter) + { + DBIDRangeRouter dbidRangeRouter = (DBIDRangeRouter)docRouter; + long indexCap = infoSrv.getIndexCap(); + long endRange = dbidRangeRouter.getEndRange(); + assert(indexCap == -1 || indexCap >= endRange); + + if(indexCap > endRange) { + dbidRangeRouter.setExpanded(true); + dbidRangeRouter.setEndRange(indexCap); + } + + dbidRangeRouter.setInitialized(true); + } + checkShutdown(); trackTransactions(); } @@ -596,6 +616,10 @@ public class MetadataTracker extends AbstractTracker implements Tracker getWriteLock().acquire(); //System.out.println("######## Metadata Tracket Acquiring Write Lock ########"); + /* + * Check to see if we are using the capped router, and if the cap has already been set. + */ + TrackerState state = getTrackerState(); //System.out.println("######## Do Track Transactions ########:"+docCount); diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/Tracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/Tracker.java index 194bd1867..9f6654121 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/Tracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/Tracker.java @@ -29,7 +29,7 @@ public interface Tracker void maintenance() throws Exception; - boolean hasMaintenance(); + boolean hasMaintenance() throws Exception; Semaphore getWriteLock(); diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/AbstractAlfrescoDistributedTest.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/AbstractAlfrescoDistributedTest.java index c812ed435..d279f3315 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/AbstractAlfrescoDistributedTest.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/AbstractAlfrescoDistributedTest.java @@ -6,10 +6,9 @@ import org.alfresco.solr.client.NodeMetaData; import org.alfresco.solr.client.SOLRAPIQueueClient; import org.alfresco.solr.client.Transaction; import org.apache.commons.io.FileUtils; +import org.apache.lucene.index.IndexableField; import org.apache.lucene.index.Term; -import org.apache.lucene.search.Query; -import org.apache.lucene.search.TermQuery; -import org.apache.lucene.search.TopDocs; +import org.apache.lucene.search.*; import org.apache.solr.SolrTestCaseJ4; import org.apache.solr.client.solrj.SolrClient; import org.apache.solr.client.solrj.SolrRequest; @@ -33,6 +32,7 @@ import org.apache.solr.common.params.ModifiableSolrParams; import org.apache.solr.common.params.SolrParams; import org.apache.solr.common.util.ContentStream; import org.apache.solr.common.util.ContentStreamBase; +import org.apache.lucene.document.Document; import org.apache.solr.common.util.NamedList; import org.apache.solr.core.CoreContainer; import org.apache.solr.core.SolrCore; @@ -40,6 +40,8 @@ import org.apache.solr.request.LocalSolrQueryRequest; import org.apache.solr.request.SolrQueryRequest; import org.apache.solr.response.SolrQueryResponse; import org.apache.solr.search.SolrIndexSearcher; +import org.apache.solr.update.AddUpdateCommand; +import org.apache.solr.update.CommitUpdateCommand; import org.apache.solr.util.RefCounted; import org.eclipse.jetty.servlet.ServletHolder; import org.junit.Assert; @@ -58,7 +60,7 @@ import java.nio.file.Path; import java.util.*; import java.util.concurrent.atomic.AtomicInteger; -import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_DOC_TYPE; +import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.*; import static org.alfresco.solr.AlfrescoSolrUtils.createCoreUsingTemplate; /** @@ -261,6 +263,16 @@ public abstract class AbstractAlfrescoDistributedTest extends SolrTestCaseJ4 } } + public void waitForDocCountAllShards(Query query, int count, long waitMillis) throws Exception + { + List cores = getJettyCores(jettyShards); + long begin = System.currentTimeMillis(); + for (SolrCore core : cores) { + waitForDocCountCore(core, query, count, waitMillis, begin); + } + } + + /** * Delele by query on all Clients * @param q @@ -342,6 +354,8 @@ public abstract class AbstractAlfrescoDistributedTest extends SolrTestCaseJ4 SolrIndexSearcher searcher = refCounted.get(); TopDocs topDocs = searcher.search(query, 10); totalCount += topDocs.totalHits; + //System.out.println("####### shard count:"+core.getName()+":"+totalCount); + Thread.sleep(2000); } finally { refCounted.decref(); } @@ -353,6 +367,34 @@ public abstract class AbstractAlfrescoDistributedTest extends SolrTestCaseJ4 throw new Exception("Cluster:Wait error expected "+count+" found "+totalCount+" : "+query.toString()); } + protected void injectDocToShards(long txnId, long aclId, long dbId, String owner) throws Exception { + List cores = getJettyCores(jettyShards); + for(SolrCore core : cores) { + SolrInputDocument doc = new SolrInputDocument(); + String id = AlfrescoSolrDataModel.getNodeDocumentId(AlfrescoSolrDataModel.DEFAULT_TENANT, aclId, dbId); + doc.addField(FIELD_SOLR4_ID, id); + doc.addField(FIELD_VERSION, 0); + doc.addField(FIELD_DBID, "" + dbId); + doc.addField(FIELD_INTXID, "" + txnId); + doc.addField(FIELD_ACLID, "" + aclId); + doc.addField(FIELD_OWNER, owner); + + AbstractAlfrescoSolrTests.SolrServletRequest solrQueryRequest = null; + try + { + solrQueryRequest = new AbstractAlfrescoSolrTests.SolrServletRequest(core, null); + AddUpdateCommand addDocCmd = new AddUpdateCommand(solrQueryRequest); + addDocCmd.overwrite = true; + addDocCmd.solrDoc = doc; + core.getUpdateHandler().addDoc(addDocCmd); + } + finally + { + solrQueryRequest.close(); + } + } + } + /** * Gets the cores for the jetty instances * @return @@ -367,6 +409,18 @@ public abstract class AbstractAlfrescoDistributedTest extends SolrTestCaseJ4 return cores; } + protected List getAdminHandlers(Collection runners) + { + List coreAdminHandlers = new ArrayList(); + for (JettySolrRunner jettySolrRunner : runners) + { + CoreContainer coreContainer = jettySolrRunner.getCoreContainer(); + AlfrescoCoreAdminHandler coreAdminHandler = (AlfrescoCoreAdminHandler) coreContainer.getMultiCoreHandler(); + coreAdminHandlers.add(coreAdminHandler); + } + return coreAdminHandlers; + } + /** * Gets the Default test client. * @return @@ -638,6 +692,7 @@ public abstract class AbstractAlfrescoDistributedTest extends SolrTestCaseJ4 String[] ranges = {"0-100", "100-200", "200-300", "300-400"}; + for (int i = 0; i < numShards; i++) { Properties props = new Properties(); @@ -647,8 +702,8 @@ public abstract class AbstractAlfrescoDistributedTest extends SolrTestCaseJ4 props.put("shard.instance", Integer.toString(i)); props.put("shard.count", Integer.toString(numShards)); - if("DB_ID_RANGE".equalsIgnoreCase(props.getProperty("shard.method"))) { - //Add + if("DB_ID_RANGE".equalsIgnoreCase(props.getProperty("shard.method"))) + { props.put("shard.range", ranges[i]); } @@ -1627,7 +1682,7 @@ public abstract class AbstractAlfrescoDistributedTest extends SolrTestCaseJ4 Properties coreProperties = new Properties(); coreProperties.setProperty("name", coreName); coreProperties.setProperty("shard", "${shard:}"); - coreProperties.setProperty("collection", "${collection:"+coreName+"}"); + coreProperties.setProperty("collection", "${collection:" + coreName + "}"); coreProperties.setProperty("config", "${solrconfig:solrconfig.xml}"); coreProperties.setProperty("schema", "${schema:schema.xml}"); coreProperties.setProperty("coreNodeName", "${coreNodeName:}"); @@ -1682,6 +1737,166 @@ public abstract class AbstractAlfrescoDistributedTest extends SolrTestCaseJ4 return response; } + protected SolrQueryResponse callExpand(AlfrescoCoreAdminHandler coreAdminHandler, + SolrCore testingCore, + int value) { + SolrQueryRequest request = new LocalSolrQueryRequest(testingCore, + params(CoreAdminParams.ACTION, "EXPAND", + CoreAdminParams.CORE, testingCore.getName(), + "add", Integer.toString(value))); + SolrQueryResponse response = new SolrQueryResponse(); + coreAdminHandler.handleCustomAction(request, response); + return response; + } + + + + + protected List assertCapOnAllShards(String[] starts) throws Exception + { + List cores = getJettyCores(jettyShards); + List caps = new ArrayList(); + List alfrescoCoreAdminHandlers = getAdminHandlers(jettyShards); + for (int i=0; i cores = getJettyCores(jettyShards); + List alfrescoCoreAdminHandlers = getAdminHandlers(jettyShards); + SolrCore core = cores.get(shard); + AlfrescoCoreAdminHandler alfrescoCoreAdminHandler = alfrescoCoreAdminHandlers.get(shard); + SolrQueryResponse response = callHandler(alfrescoCoreAdminHandler, core, "RANGECHECK"); + NamedList values = response.getValues(); + String ex = (String)values.get("exception"); + if(ex == null || ex.indexOf("not initialized") == -1) { + return response; + } else { + Thread.sleep(1000); + } + } + } + + public SolrQueryResponse expand(int shard, int value) { + List cores = getJettyCores(jettyShards); + List alfrescoCoreAdminHandlers = getAdminHandlers(jettyShards); + SolrCore core = cores.get(shard); + AlfrescoCoreAdminHandler alfrescoCoreAdminHandler = alfrescoCoreAdminHandlers.get(shard); + SolrQueryResponse response = callExpand(alfrescoCoreAdminHandler, core, value); + return response; + } + + protected long assertCapForCore(AlfrescoCoreAdminHandler coreAdminHandler, + SolrCore core, + long start) + throws IOException + { + //Get the cap from the index + long cap = -1; + RefCounted refCounted = null; + Set fields = new HashSet(); + fields.add(FIELD_DBID); + try { + refCounted = core.getSearcher(); + SolrIndexSearcher searcher = refCounted.get(); + + TopDocs topDocs = searcher.search(new TermQuery(new Term(FIELD_SOLR4_ID, "TRACKER!STATE!CAP")), 10); + ScoreDoc doc = topDocs.scoreDocs[0]; + Document document = searcher.doc(doc.doc, fields); + cap = Math.abs(getFieldValueLong(document, FIELD_DBID)); + } finally { + refCounted.decref(); + } + + System.out.println("####### got cap:"+cap); + + //Check that the max DBID in the core matches the cap. + + long maxDBID = -1; + Query query = new TermQuery(new Term(FIELD_DOC_TYPE, SolrInformationServer.DOC_TYPE_NODE)); + try { + refCounted = core.getSearcher(); + SolrIndexSearcher searcher = refCounted.get(); + + TopDocs topDocs = searcher.search(query, + 1, + new Sort(new SortField(FIELD_DBID, SortField.Type.LONG, true))); + System.out.println("####### hits:"+topDocs.totalHits+":"+query); + + ScoreDoc doc = topDocs.scoreDocs[0]; + Document document = searcher.doc(doc.doc, fields); + maxDBID = getFieldValueLong(document, FIELD_DBID); + } finally { + refCounted.decref(); + } + + System.out.println("####### got max DBID:"+maxDBID); + + + if(maxDBID != cap) { + throw new IOException("Max DBID should equal cap:"+maxDBID+" != "+cap); + } + + long minDBID = -1; + try { + refCounted = core.getSearcher(); + SolrIndexSearcher searcher = refCounted.get(); + + TopDocs topDocs = searcher.search(query, + 1, + new Sort(new SortField(FIELD_DBID, SortField.Type.LONG, false))); + ScoreDoc doc = topDocs.scoreDocs[0]; + Document document = searcher.doc(doc.doc, fields); + minDBID = getFieldValueLong(document, FIELD_DBID); + } finally { + refCounted.decref(); + } + + System.out.println("####### got min DBID:"+minDBID); + + if(minDBID != start) { + throw new IOException("Min DBID should equal start:"+minDBID+" != "+start); + } + + + SolrQueryResponse response = callHandler(coreAdminHandler, core, "CHECKCAP"); + NamedList values = response.getValues(); + long checkCap = (long)values.get("CAP"); + if(checkCap != cap) + { + throw new IOException("The admin handler returned bad cap:"+checkCap+":"+cap); + } + + return cap; + + } + + private String getFieldValueString(Document doc, String fieldName) + { + IndexableField field = doc.getField(fieldName); + String value = null; + if (field != null) + { + value = field.stringValue(); + } + return value; + } + + + private long getFieldValueLong(Document doc, String fieldName) + { + return Long.parseLong(getFieldValueString(doc, fieldName)); + } + + protected void assertCHECKCAPAction(long[] caps) throws Exception + { + + } + /** * A JUnit Rule to setup Jetty */ diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedExpandDbidRangeAlfrescoSolrTrackerTest.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedExpandDbidRangeAlfrescoSolrTrackerTest.java new file mode 100644 index 000000000..4f2638adc --- /dev/null +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/DistributedExpandDbidRangeAlfrescoSolrTrackerTest.java @@ -0,0 +1,412 @@ +/* + * Copyright (C) 2005-2014 Alfresco Software Limited. + * + * This file is part of Alfresco + * + * Alfresco is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * Alfresco is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with Alfresco. If not, see . + */ +package org.alfresco.solr.tracker; + +import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_DOC_TYPE; +import static org.alfresco.repo.search.adaptor.lucene.QueryConstants.FIELD_SOLR4_ID; +import static org.alfresco.solr.AlfrescoSolrUtils.getAcl; +import static org.alfresco.solr.AlfrescoSolrUtils.getAclChangeSet; +import static org.alfresco.solr.AlfrescoSolrUtils.getAclReaders; +import static org.alfresco.solr.AlfrescoSolrUtils.getNode; +import static org.alfresco.solr.AlfrescoSolrUtils.getNodeMetaData; +import static org.alfresco.solr.AlfrescoSolrUtils.getTransaction; +import static org.alfresco.solr.AlfrescoSolrUtils.indexAclChangeSet; +import static org.alfresco.solr.AlfrescoSolrUtils.list; + +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + +import org.alfresco.solr.AbstractAlfrescoDistributedTest; +import org.alfresco.solr.SolrInformationServer; +import org.alfresco.solr.client.Acl; +import org.alfresco.solr.client.AclChangeSet; +import org.alfresco.solr.client.AclReaders; +import org.alfresco.solr.client.Node; +import org.alfresco.solr.client.NodeMetaData; +import org.alfresco.solr.client.Transaction; +import org.apache.lucene.index.Term; +import org.apache.lucene.search.TermQuery; +import org.apache.lucene.util.LuceneTestCase; +import org.apache.solr.SolrTestCaseJ4; +import org.apache.solr.common.util.NamedList; +import org.apache.solr.client.solrj.response.QueryResponse; +import org.apache.solr.response.SolrQueryResponse; +import org.junit.Rule; +import org.junit.Test; + +/** + * @author Joel + */ +@SolrTestCaseJ4.SuppressSSL +@LuceneTestCase.SuppressCodecs({"Appending","Lucene3x","Lucene40","Lucene41","Lucene42","Lucene43", "Lucene44", "Lucene45","Lucene46","Lucene47","Lucene48","Lucene49"}) +public class DistributedExpandDbidRangeAlfrescoSolrTrackerTest extends AbstractAlfrescoDistributedTest +{ + @Rule + public JettyServerRule jetty = new JettyServerRule(2,this, getShardMethod()); + + @Test + public void testDbIdRange() throws Exception + { + putHandleDefaults(); + + int numAcls = 250; + AclChangeSet bulkAclChangeSet = getAclChangeSet(numAcls); + + List bulkAcls = new ArrayList(); + List bulkAclReaders = new ArrayList(); + + for(int i=0; i nodes = new ArrayList(); + List nodeMetaDatas = new ArrayList(); + + Transaction bigTxn = getTransaction(0, numNodes); + + for(int i=0; i