inteligr8/SearchServices
1
0
Fork 0
mirror of https://github.com/Alfresco/SearchServices.git synced 2026-09-16 18:12:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'feature/SEARCH-1752' into 'master'

Browse Source 
Feature/search 1752

See merge request search_discovery/insightengine!205
This commit is contained in:
Andrea Gazzarini
2019-10-31 05:28:31 +00:00
parent 337c62d0c1 ea691cba27
commit 0dc3ff2694
20 changed files with 615 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files
e2e-test/generator-alfresco-docker-compose/generators/app/templates/images/search/Dockerfile
+1 -3
View File
@@ -54,7 +54,6 @@ ENV MASTER_HOST $MASTER_HOST
# Set Master / Slave configuration for this Node
RUN if [ "$ENABLE_MASTER" == "true" ] ; then \
sed -i '/^bash.*/i echo "\nenable.master=${ENABLE_MASTER}\nenable.slave=${ENABLE_SLAVE}" >> ${DIST_DIR}/solrhome/templates/rerank/conf/solrcore.properties\n' \
${DIST_DIR}/solr/bin/search_config_setup.sh; \
sed -i "/^bash.*/i sed -i '/^\\\\\s*<requestHandler name=\"\\\\/replication\".*/a \
<lst name=\"master\">\
@@ -64,7 +63,6 @@ RUN if [ "$ENABLE_MASTER" == "true" ] ; then \
</lst>' ${DIST_DIR}/solrhome/templates/rerank/conf/solrconfig.xml\n" ${DIST_DIR}/solr/bin/search_config_setup.sh; \
fi
RUN if [ "$ENABLE_SLAVE" == "true" ] ; then \
sed -i '/^bash.*/i echo "\nenable.master=${ENABLE_MASTER}\nenable.slave=${ENABLE_SLAVE}" >> ${DIST_DIR}/solrhome/templates/rerank/conf/solrcore.properties\n' \
${DIST_DIR}/solr/bin/search_config_setup.sh; \
sed -i "/^bash.*/i sed -i '/^\\\\\s*<requestHandler name=\"\\\\/replication\".*/a \
<lst name=\"slave\">\
@@ -131,4 +129,4 @@ fi
RUN mkdir ${DIST_DIR}/keystore \
&& chown -R solr:solr ${DIST_DIR}/keystore
VOLUME ["${DIST_DIR}/keystore"]
VOLUME ["${DIST_DIR}/keystore"]
search-services/alfresco-search/doc/architecture/trackers/00001-content-tracker.md
-2
View File
@@ -150,8 +150,6 @@ The following table illustrates the configuration properties used by the Tracker
|alfresco.stores|workspace://SpacesStore|The reference to a node store| | |Y|Y| | |
|batch.count|5000|UpSert batch size (e.g. metadata docs, acls)| | |Y|Y| | |
|alfresco.maxLiveSearchers|2|Max allowed number of active searchers|Y| |Y|Y| | |
|enable.slave|false|Indicates if the hosting instance is a slave| | |Y|Y| | |
|enable.master|true|Indicates if the hosting instance is a master| | |Y|Y| | |
|shard.count|1|The total number of shards that compose the Solr infrastructure|| |Y|Y| | |
|shard.instance|0|The unique shard identifier assigned to this instance|| |Y|Y| | |
|shard.method|"DB_ID"|Data (Documents, ACLs) Routing criteria among shards| | |Y|Y| | |
search-services/alfresco-search/src/main/java/org/alfresco/solr/AlfrescoCoreAdminHandler.java
+1 -4
View File
@@ -224,10 +224,7 @@ public class AlfrescoCoreAdminHandler extends CoreAdminHandler
MultiThreadedHttpConnectionManager.shutdownAll();
//Remove any core trackers still hanging around
trackerRegistry.getCoreNames().forEach(coreName ->
{
trackerRegistry.removeTrackersForCore(coreName);
});
trackerRegistry.getCoreNames().forEach(coreName -> trackerRegistry.removeTrackersForCore(coreName));
//Remove any information servers
informationServers.clear();
search-services/alfresco-search/src/main/java/org/alfresco/solr/lifecycle/SolrCoreLoadListener.java
+196 -64
View File
@@ -18,6 +18,7 @@
*/
package org.alfresco.solr.lifecycle;
import static java.util.Arrays.asList;
import static java.util.Optional.ofNullable;
import org.alfresco.opencmis.dictionary.CMISStrictDictionaryService;
@@ -34,6 +35,7 @@ import org.alfresco.solr.tracker.CommitTracker;
import org.alfresco.solr.tracker.ContentTracker;
import org.alfresco.solr.tracker.MetadataTracker;
import org.alfresco.solr.tracker.ModelTracker;
import org.alfresco.solr.tracker.SlaveNodeStatePublisher;
import org.alfresco.solr.tracker.SolrTrackerScheduler;
import org.alfresco.solr.tracker.Tracker;
import org.alfresco.solr.tracker.TrackerRegistry;
@@ -82,6 +84,20 @@ public class SolrCoreLoadListener extends AbstractSolrEventListener
@Override
public void newSearcher(SolrIndexSearcher newSearcher, SolrIndexSearcher currentSearcher)
{
if (getCore().isReloaded())
{
LOGGER.info("Solr Core {}, instance {} has been reloaded. " +
"The previous tracking subsystem will be stopped and another set of trackers will be registered on this new instance.",
getCore().getName(),
getCore().hashCode());
}
else
{
LOGGER.info("Solr Core {}, instance {}, has been registered for the first time.",
getCore().getName(),
getCore().hashCode());
}
CoreContainer coreContainer = getCore().getCoreContainer();
AlfrescoCoreAdminHandler admin = (AlfrescoCoreAdminHandler) coreContainer.getMultiCoreHandler();
SolrCore core = getCore();
@@ -117,14 +133,62 @@ public class SolrCoreLoadListener extends AbstractSolrEventListener
scheduler);
}
/*
* The shutdown hook needs to be registered regardless we are slave or masters.
* This because if we are master all trackers will be scheduled, if we are slave the node state publisher
* will be scheduled.
*
* As consequence of that, regardless the node role, we will always have something to shutdown in the tracker
* registry.
*/
final List<Tracker> trackers = new ArrayList<>();
core.addCloseHook(new CloseHook()
{
@Override
public void preClose(SolrCore core)
{
LOGGER.info("Solr Core instance {} with name {} is going to be closed. Tracking Subsystem shutdown callback procedure has been started.", core.hashCode(), core.getName());
// IMPORTANT: the closure needs to be created with the trackers created in this method
shutdownTrackers(core, trackers, scheduler, false);
}
@Override
public void postClose(SolrCore core)
{
LOGGER.info("Solr Core instance {} with name {} has been closed. Tracking Subsystem shutdown callback procedure has been completed.", core.hashCode(), core.getName());
}
});
boolean trackersHaveBeenEnabled = Boolean.parseBoolean(coreProperties.getProperty("enable.alfresco.tracking", "true"));
boolean owningCoreIsSlave = isSlaveModeEnabledFor(core);
if (trackerRegistry.hasTrackersForCore(core.getName()))
{
LOGGER.info("Trackers (it could be only the node state publisher in case this node is a slave) for " + core.getName() + " are already registered, shutting them down.");
Collection<Tracker> alreadyRegisteredTrackers = trackerRegistry.getTrackersForCore(core.getName());
trackerRegistry.removeTrackersForCore(core.getName());
shutdownTrackers(core, alreadyRegisteredTrackers, scheduler, core.isReloaded());
admin.getInformationServers().remove(core.getName());
}
// Re-put the information server in the map because a Core reload (see above) could have removed the reference.
admin.getInformationServers().put(core.getName(), informationServer);
// Guard conditions: if trackers must be disabled then immediately return, we've done here.
// Case #1: trackers have been explicitly disabled.
if (!trackersHaveBeenEnabled)
{
LOGGER.info("SearchServices Core Trackers have been explicitly disabled on core \"{}\" through \"enable.alfresco.tracking\" configuration property.", core.getName());
SlaveNodeStatePublisher statePublisher = new SlaveNodeStatePublisher(false, coreProperties, repositoryClient, core.getName(), informationServer);
trackerRegistry.register(core.getName(), statePublisher);
scheduler.schedule(statePublisher, core.getName(), coreProperties);
trackers.add(statePublisher);
LOGGER.info("SearchServices Slave Node Provider have been created and scheduled for core \"{}\".", core.getName());
return;
}
@@ -132,81 +196,104 @@ public class SolrCoreLoadListener extends AbstractSolrEventListener
if (owningCoreIsSlave)
{
LOGGER.info("SearchServices Core Trackers have been disabled on core \"{}\" because it is a slave core.", core.getName());
SlaveNodeStatePublisher statePublisher = new SlaveNodeStatePublisher(false, coreProperties, repositoryClient, core.getName(), informationServer);
trackerRegistry.register(core.getName(), statePublisher);
scheduler.schedule(statePublisher, core.getName(), coreProperties);
trackers.add(statePublisher);
LOGGER.info("SearchServices Slave Node Provider have been created and scheduled for Core instance {} with name {}.", core.hashCode(), core.getName());
return;
}
LOGGER.info("SearchServices Tracking Subsystem starts on core {}", core.getName());
if (trackerRegistry.hasTrackersForCore(core.getName()))
{
LOGGER.info("Trackers for " + core.getName()+ " is already registered, shutting them down.");
shutdownTrackers(core.getName(), trackerRegistry.getTrackersForCore(core.getName()), scheduler);
trackerRegistry.removeTrackersForCore(core.getName());
admin.getInformationServers().remove(core.getName());
}
LOGGER.info("SearchServices Tracking Subsystem starts on Solr Core instance {} with name {}", core.hashCode(), core.getName());
final List<Tracker> trackers = createCoreTrackers(core.getName(), trackerRegistry, coreProperties, scheduler, repositoryClient, informationServer);
trackers.addAll(createAndScheduleCoreTrackers(core, trackerRegistry, coreProperties, scheduler, repositoryClient, informationServer));
CommitTracker commitTracker = new CommitTracker(coreProperties, repositoryClient, core.getName(), informationServer, trackers);
trackerRegistry.register(core.getName(), commitTracker);
scheduler.schedule(commitTracker, core.getName(), coreProperties);
LOGGER.info("SearchServices Core Trackers have been correctly registered and scheduled.");
LOGGER.info("Tracker {}, instance {}, belonging to Core {}, instance {} has been registered and scheduled.",
commitTracker.getClass().getSimpleName(),
commitTracker.hashCode(),
core.getName(),
core.hashCode());
//Add the commitTracker to the list of scheduled trackers that can be shutdown
trackers.add(commitTracker);
core.addCloseHook(new CloseHook()
{
@Override
public void preClose(SolrCore core)
{
LOGGER.info("Tracking Subsystem shutdown procedure for core {} has been started.", core.getName());
shutdownTrackers(core.getName(), trackers, scheduler);
}
@Override
public void postClose(SolrCore core)
{
LOGGER.info("Shutdown procedure for core {} has been completed.", core.getName());
}
});
}
List<Tracker> createCoreTrackers(String coreName,
TrackerRegistry trackerRegistry,
Properties props,
SolrTrackerScheduler scheduler,
SOLRAPIClient repositoryClient,
SolrInformationServer srv)
List<Tracker> createAndScheduleCoreTrackers(SolrCore core,
TrackerRegistry trackerRegistry,
Properties props,
SolrTrackerScheduler scheduler,
SOLRAPIClient repositoryClient,
SolrInformationServer srv)
{
List<Tracker> trackers = new ArrayList<>();
AclTracker aclTracker =
registerAndSchedule(
new AclTracker(props, repositoryClient, core.getName(), srv),
core,
props,
trackerRegistry,
scheduler);
AclTracker aclTracker = new AclTracker(props, repositoryClient, coreName, srv);
trackerRegistry.register(coreName, aclTracker);
scheduler.schedule(aclTracker, coreName, props);
ContentTracker contentTracker =
registerAndSchedule(
new ContentTracker(props, repositoryClient, core.getName(), srv),
core,
props,
trackerRegistry,
scheduler);
ContentTracker contentTrkr = new ContentTracker(props, repositoryClient, coreName, srv);
trackerRegistry.register(coreName, contentTrkr);
scheduler.schedule(contentTrkr, coreName, props);
MetadataTracker metadataTracker =
registerAndSchedule(
new MetadataTracker(true, props, repositoryClient, core.getName(), srv),
core,
props,
trackerRegistry,
scheduler);
MetadataTracker metaTrkr = new MetadataTracker(props, repositoryClient, coreName, srv);
trackerRegistry.register(coreName, metaTrkr);
scheduler.schedule(metaTrkr, coreName, props);
CascadeTracker cascadeTrkr = new CascadeTracker(props, repositoryClient, coreName, srv);
trackerRegistry.register(coreName, cascadeTrkr);
scheduler.schedule(cascadeTrkr, coreName, props);
CascadeTracker cascadeTracker =
registerAndSchedule(
new CascadeTracker(props, repositoryClient, core.getName(), srv),
core,
props,
trackerRegistry,
scheduler);
//The CommitTracker will acquire these locks in order
//The ContentTracker will likely have the longest runs so put it first to ensure the MetadataTracker is not paused while
//waiting for the ContentTracker to release it's lock.
//The aclTracker will likely have the shortest runs so put it last.
trackers.add(cascadeTrkr);
trackers.add(contentTrkr);
trackers.add(metaTrkr);
trackers.add(aclTracker);
return trackers;
return asList(cascadeTracker, contentTracker, metadataTracker, aclTracker);
}
/**
* Accepts a {@link Tracker} instance, registers and schedules it.
*
* @param tracker the tracker that will be scheduled and registered.
* @param core the owning core.
* @param properties configuration properties.
* @param registry the tracker registry instance.
* @param scheduler the tracker schedule instance.
* @param <T> the tracker instance.
* @return the registered and scheduled tracker instance.
*/
private <T extends Tracker> T registerAndSchedule(T tracker, SolrCore core, Properties properties, TrackerRegistry registry, SolrTrackerScheduler scheduler)
{
registry.register(core.getName(), tracker);
scheduler.schedule(tracker, core.getName(), properties);
LOGGER.info("Tracker {}, instance {}, belonging to Core {}, instance {} has been registered and scheduled.",
tracker.getClass().getSimpleName(),
tracker.hashCode(),
core.getName(),
core.hashCode());
return tracker;
}
private void createModelTracker(String coreName,
@@ -246,34 +333,79 @@ public class SolrCoreLoadListener extends AbstractSolrEventListener
* have multiple cores of the same name running. Left over trackers in the registry are cleaned up by the CoreContainer
* shutdown, that happens in the the AlfrescoCoreAdminHandler.shutdown().
*
* @param coreName The name of the core
* The coreHasBeenReloaded flag is used just for logging out meaningful messages about the owning core instance.
* If we are in a RELOAD scenario (coreHasBeenReloaded = true) we no longer have the reference of the closed core
* so we print only its name. Instead in case we are here because a core has been closed, we can print out the core
* reference in order to add meaningful information in the log.
*
* @param core The owning core name.
* @param coreTrackers A collection of trackers
* @param scheduler The scheduler
* @param coreHasBeenReloaded a flag indicating if we are on a Core RELOAD scenario.
*/
void shutdownTrackers(String coreName, Collection<Tracker> coreTrackers, SolrTrackerScheduler scheduler)
void shutdownTrackers(SolrCore core, Collection<Tracker> coreTrackers, SolrTrackerScheduler scheduler, boolean coreHasBeenReloaded)
{
coreTrackers.forEach(tracker -> shutdownTracker(core, tracker, scheduler, coreHasBeenReloaded));
}
/**
* Shutdown procedure for a single tracker.
* The coreHasBeenReloaded flag is used just for logging out meaningful messages about the owning core instance.
* If we are in a RELOAD scenario (coreHasBeenReloaded = true) we no longer have the reference of the closed core
* so we print only its name. Instead in case we are here because a core has been closed, we can print out the core
* reference in order to add meaningful information in the log.
*
* @param core the owning {@link SolrCore}
* @param tracker the {@link Tracker} instance we want to stop.
* @param scheduler the scheduler.
* @param coreHasBeenReloaded a flag indicating if we are on a Core RELOAD scenario.
*/
private void shutdownTracker(SolrCore core, Tracker tracker, SolrTrackerScheduler scheduler, boolean coreHasBeenReloaded)
{
// In case of reload the input core is not the owner: the owner is instead the previous (closed) core and we don't have its reference here.
String coreReference = core.getName() + (coreHasBeenReloaded ? "" : ", instance " + core.hashCode());
if (tracker.isAlreadyInShutDownMode())
{
LOGGER.info("Tracker {}, instance {} belonging to core {}, is already in shutdown mode.",
tracker.getClass().getSimpleName(),
tracker.hashCode(),
coreReference);
return;
}
LOGGER.info("Tracker {}, instance {} belonging to core {} shutdown procedure initiated.",
tracker.getClass().getSimpleName(),
tracker.hashCode(),
coreReference);
try
{
LOGGER.info("Shutting down Trackers Subsystem for core \"{}\" which contains {} core trackers.", coreName, coreTrackers.size());
// Sets the shutdown flag on the trackers to stop them from doing any more work
coreTrackers.forEach(tracker -> tracker.setShutdown(true));
tracker.setShutdown(true);
if (!scheduler.isShutdown())
{
coreTrackers.forEach(tracker -> scheduler.deleteJobForTrackerInstance(coreName, tracker) );
scheduler.deleteJobForTrackerInstance(core.getName(), tracker);
}
coreTrackers.forEach(Tracker::shutdown);
tracker.shutdown();
LOGGER.info("Tracker {}, instance {}, belonging to core {} shutdown procedure correctly terminated.",
tracker.getClass().getSimpleName(),
tracker.hashCode(),
coreReference);
}
catch (Exception e)
catch (Exception exception)
{
LOGGER.error("Tracking Subsystem shutdown procedure failed to shutdown trackers for core {}. See the stacktrace below for further details.", coreName, e);
LOGGER.error("Tracker {}, instance {} belonging to core {}, shutdown procedure failed. " +
"See the stacktrace below for further details.",
tracker.getClass().getSimpleName(),
tracker.hashCode(),
coreReference,
exception);
}
}
/**
* Checks if the content store belonging to the hosting Solr node must be set in read only mode.
* Checks if the configuration declares this node as a slave.
*
* @param core the hosting {@link SolrCore} instance.
* @return true if the content store must be set in read only mode, false otherwise.
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AbstractTracker.java
+45 -51
View File
@@ -18,6 +18,7 @@
*/
package org.alfresco.solr.tracker;
import java.lang.invoke.MethodHandles;
import java.net.ConnectException;
import java.net.SocketTimeoutException;
import java.util.Properties;
@@ -31,7 +32,6 @@ import org.alfresco.solr.client.SOLRAPIClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Abstract base class that provides common {@link Tracker} behaviour.
*
@@ -39,20 +39,18 @@ import org.slf4j.LoggerFactory;
*/
public abstract class AbstractTracker implements Tracker
{
public static final long TIME_STEP_32_DAYS_IN_MS = 1000 * 60 * 60 * 24 * 32L;
public static final long TIME_STEP_1_HR_IN_MS = 60 * 60 * 1000L;
public static final String SHARD_METHOD_ACLID = "ACL_ID";
public static final String SHARD_METHOD_DBID = "DB_ID";
protected final static Logger log = LoggerFactory.getLogger(AbstractTracker.class);
static final long TIME_STEP_32_DAYS_IN_MS = 1000 * 60 * 60 * 24 * 32L;
static final long TIME_STEP_1_HR_IN_MS = 60 * 60 * 1000L;
static final String SHARD_METHOD_DBID = "DB_ID";
protected final static Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
protected Properties props;
protected SOLRAPIClient client;
protected InformationServer infoSrv;
InformationServer infoSrv;
protected String coreName;
protected StoreRef storeRef;
protected long batchCount;
protected boolean isSlave = false;
protected boolean isMaster = true;
protected String alfrescoVersion;
protected TrackerStats trackerStats;
protected boolean runPostModelLoadInit = true;
@@ -71,14 +69,10 @@ public abstract class AbstractTracker implements Tracker
protected volatile boolean rollback;
protected final Type type;
/*
* A thread handler can be used by subclasses, but they have to intentionally instantiate it.
*/
protected ThreadHandler threadHandler;
ThreadHandler threadHandler;
/**
* Default constructor, strictly for testing.
@@ -98,8 +92,6 @@ public abstract class AbstractTracker implements Tracker
storeRef = new StoreRef(p.getProperty("alfresco.stores", "workspace://SpacesStore"));
batchCount = Integer.parseInt(p.getProperty("alfresco.batch.count", "5000"));
maxLiveSearchers = Integer.parseInt(p.getProperty("alfresco.maxLiveSearchers", "2"));
isSlave = Boolean.parseBoolean(p.getProperty("enable.slave", "false"));
isMaster = Boolean.parseBoolean(p.getProperty("enable.master", "true"));
shardCount = Integer.parseInt(p.getProperty("shard.count", "1"));
shardInstance = Integer.parseInt(p.getProperty("shard.instance", "0"));
@@ -115,19 +107,19 @@ public abstract class AbstractTracker implements Tracker
this.type = type;
log.info("Solr built for Alfresco version: " + alfrescoVersion);
LOGGER.info("Solr built for Alfresco version: {}", alfrescoVersion);
}
/**
* Subclasses must implement behaviour that completes the following steps, in order:
*
* <ol>
* <li>Purge</li>
* <li>Reindex</li>
* <li>Index</li>
* <li>Track repository</li>
* </ol>
* @throws Throwable
*/
protected abstract void doTrack() throws Throwable;
@@ -146,7 +138,7 @@ public abstract class AbstractTracker implements Tracker
}
catch(Exception e)
{
// Ignore
}
@@ -158,12 +150,7 @@ public abstract class AbstractTracker implements Tracker
getTrackerState();
if(state == null) {
return true;
} else {
return false;
}
return state == null;
}
/**
@@ -173,8 +160,9 @@ public abstract class AbstractTracker implements Tracker
@Override
public void track()
{
if(runLock.availablePermits() == 0) {
log.info("... " + this.getClass().getSimpleName() + " for core [" + coreName + "] is already in use "+ this.getClass());
if(runLock.availablePermits() == 0)
{
LOGGER.info("... {} for core [{}] is already in use {}", this.getClass().getSimpleName(), coreName, getClass());
return;
}
@@ -184,15 +172,14 @@ public abstract class AbstractTracker implements Tracker
* The runLock ensures that for each tracker type (metadata, content, commit, cascade) only one tracker will
* be running at a time.
*/
runLock.acquire();
if(state==null && Boolean.parseBoolean(System.getProperty("alfresco.test", "false")))
if (state==null && Boolean.parseBoolean(System.getProperty("alfresco.test", "false")))
{
assert(assertTrackerStateRemainsNull());
}
log.info("... Running " + this.getClass().getSimpleName() + " for core [" + coreName + "].");
LOGGER.info("... Running {} for core [{}]", this.getClass().getSimpleName(), coreName);
if(this.state == null)
{
@@ -200,8 +187,8 @@ public abstract class AbstractTracker implements Tracker
* Set the global state for the tracker here.
*/
this.state = getTrackerState();
log.debug("##### Setting tracker global state.");
log.debug("State set: " + this.state.toString());
LOGGER.debug("##### Setting tracker global state.");
LOGGER.debug("State set: {}", this.state.toString());
this.state.setRunning(true);
}
else
@@ -219,38 +206,39 @@ public abstract class AbstractTracker implements Tracker
catch(IndexTrackingShutdownException t)
{
setRollback(true);
log.info("Stopping index tracking for " + getClass().getSimpleName() + " - " + coreName);
LOGGER.info("Stopping index tracking for {} - {}", getClass().getSimpleName(), coreName);
}
catch(Throwable t)
{
setRollback(true);
if (t instanceof SocketTimeoutException || t instanceof ConnectException)
{
if (log.isDebugEnabled())
if (LOGGER.isDebugEnabled())
{
// DEBUG, so give the whole stack trace
log.warn("Tracking communication timed out for " + getClass().getSimpleName() + " - " + coreName, t);
LOGGER.warn("Tracking communication timed out for {} - {}", getClass().getSimpleName(), coreName, t);
}
else
{
// We don't need the stack trace. It timed out.
log.warn("Tracking communication timed out for " + getClass().getSimpleName() + " - " + coreName);
LOGGER.warn("Tracking communication timed out for for {} - {}", getClass().getSimpleName(), coreName);
}
}
else
{
log.error("Tracking failed for " + getClass().getSimpleName() + " - " + coreName, t);
LOGGER.error("Tracking failed for for {} - {}", getClass().getSimpleName(), coreName, t);
}
}
}
catch (InterruptedException e)
{
log.error("Semaphore interrupted for " + getClass().getSimpleName() + " - " + coreName, e);
LOGGER.error("Semaphore interrupted for for {} - {}", getClass().getSimpleName(), coreName, e);
}
finally
{
infoSrv.unregisterTrackerThread();
if(state != null) {
if(state != null)
{
//During a rollback state is set to null.
state.setRunning(false);
state.setCheck(false);
@@ -259,15 +247,18 @@ public abstract class AbstractTracker implements Tracker
}
}
public boolean getRollback() {
public boolean getRollback()
{
return this.rollback;
}
public void setRollback(boolean rollback) {
public void setRollback(boolean rollback)
{
this.rollback = rollback;
}
private void continueState() {
private void continueState()
{
infoSrv.continueState(state);
state.incrementTrackerCycles();
}
@@ -289,8 +280,6 @@ public abstract class AbstractTracker implements Tracker
return this.infoSrv.getTrackerInitialState();
}
}
/**
* Allows time for the scheduled asynchronous tasks to complete
@@ -309,6 +298,7 @@ public abstract class AbstractTracker implements Tracker
}
catch (InterruptedException e)
{
// Nothing to be done here
}
}
currentRunnable = this.threadHandler.peekHeadReindexWorker();
@@ -328,6 +318,12 @@ public abstract class AbstractTracker implements Tracker
}
}
@Override
public boolean isAlreadyInShutDownMode()
{
return shutdown;
}
@Override
public void setShutdown(boolean shutdown)
{
@@ -337,7 +333,6 @@ public abstract class AbstractTracker implements Tracker
@Override
public void shutdown()
{
log.warn("Core " + coreName + " shutdown called on tracker. " + getClass().getSimpleName() + " " + hashCode());
setShutdown(true);
if(this.threadHandler != null)
{
@@ -345,15 +340,16 @@ public abstract class AbstractTracker implements Tracker
}
}
public Semaphore getWriteLock() {
public Semaphore getWriteLock()
{
return this.writeLock;
}
public Semaphore getRunLock() {
public Semaphore getRunLock()
{
return this.runLock;
}
/**
* @return Alfresco version Solr was built for
*/
@@ -372,6 +368,4 @@ public abstract class AbstractTracker implements Tracker
{
return type;
}
}
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java
-5
View File
@@ -288,11 +288,6 @@ public class AclTracker extends AbstractTracker
protected void trackRepository() throws IOException, AuthenticationException, JSONException
{
checkShutdown();
if(!isMaster && isSlave)
{
return;
}
TrackerState state = super.getTrackerState();
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java
+2 -2
View File
@@ -50,14 +50,14 @@ public class CascadeTracker extends AbstractTracker implements Tracker
public CascadeTracker(Properties p, SOLRAPIClient client, String coreName,
InformationServer informationServer)
{
super(p, client, coreName, informationServer, Tracker.Type.Cascade);
super(p, client, coreName, informationServer, Tracker.Type.CASCADE);
threadHandler = new ThreadHandler(p, coreName, "CascadeTracker");
}
CascadeTracker()
{
super(Tracker.Type.Cascade);
super(Tracker.Type.CASCADE);
}
@Override
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CommitTracker.java
+2 -2
View File
@@ -51,7 +51,7 @@ public class CommitTracker extends AbstractTracker
**/
CommitTracker()
{
super(Tracker.Type.Commit);
super(Tracker.Type.COMMIT);
}
public CommitTracker(Properties p,
@@ -60,7 +60,7 @@ public class CommitTracker extends AbstractTracker
InformationServer informationServer,
List<Tracker> trackers)
{
super(p, client, coreName, informationServer, Tracker.Type.Commit);
super(p, client, coreName, informationServer, Tracker.Type.COMMIT);
//Set the trackers
for(Tracker tracker : trackers) {
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java
+2 -2
View File
@@ -45,7 +45,7 @@ public class ContentTracker extends AbstractTracker implements Tracker
public ContentTracker(Properties p, SOLRAPIClient client, String coreName,
InformationServer informationServer)
{
super(p, client, coreName, informationServer, Tracker.Type.Content);
super(p, client, coreName, informationServer, Tracker.Type.CONTENT);
contentReadBatchSize = Integer.parseInt(p.getProperty("alfresco.contentReadBatchSize", "100"));
contentUpdateBatchSize = Integer.parseInt(p.getProperty("alfresco.contentUpdateBatchSize", "1000"));
threadHandler = new ThreadHandler(p, coreName, "ContentTracker");
@@ -53,7 +53,7 @@ public class ContentTracker extends AbstractTracker implements Tracker
ContentTracker()
{
super(Tracker.Type.Content);
super(Tracker.Type.CONTENT);
}
@Override
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java
+15 -189
View File
@@ -20,27 +20,15 @@ package org.alfresco.solr.tracker;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.alfresco.error.AlfrescoRuntimeException;
import org.alfresco.httpclient.AuthenticationException;
import org.alfresco.opencmis.dictionary.CMISStrictDictionaryService;
import org.alfresco.repo.dictionary.NamespaceDAO;
import org.alfresco.repo.index.shard.ShardMethodEnum;
import org.alfresco.repo.index.shard.ShardState;
import org.alfresco.repo.index.shard.ShardStateBuilder;
import org.alfresco.repo.search.impl.QueryParserUtils;
import org.alfresco.service.cmr.dictionary.DictionaryService;
import org.alfresco.service.cmr.dictionary.PropertyDefinition;
import org.alfresco.service.namespace.QName;
import org.alfresco.solr.AlfrescoCoreAdminHandler;
import org.alfresco.solr.AlfrescoSolrDataModel;
import org.alfresco.solr.BoundedDeque;
import org.alfresco.solr.InformationServer;
import org.alfresco.solr.NodeReport;
@@ -53,95 +41,41 @@ import org.alfresco.solr.client.SOLRAPIClient;
import org.alfresco.solr.client.Transaction;
import org.alfresco.solr.client.Transactions;
import org.apache.commons.codec.EncoderException;
import org.apache.commons.lang3.StringUtils;
import org.json.JSONException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static java.util.Optional.of;
import static java.util.Optional.ofNullable;
import static org.alfresco.solr.tracker.DocRouterFactory.SHARD_KEY_KEY;
/*
* This tracks two things: transactions and metadata nodes
* @author Ahmed Owian
*/
public class MetadataTracker extends AbstractTracker implements Tracker
public class MetadataTracker extends NodeStatePublisher implements Tracker
{
protected final static Logger log = LoggerFactory.getLogger(MetadataTracker.class);
private static final int DEFAULT_TRANSACTION_DOCS_BATCH_SIZE = 100;
private static final int DEFAULT_NODE_BATCH_SIZE = 10;
private int transactionDocsBatchSize = DEFAULT_TRANSACTION_DOCS_BATCH_SIZE;
private int nodeBatchSize = DEFAULT_NODE_BATCH_SIZE;
private ConcurrentLinkedQueue<Long> transactionsToReindex = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> transactionsToIndex = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> transactionsToPurge = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> nodesToReindex = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> nodesToIndex = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<Long> nodesToPurge = new ConcurrentLinkedQueue<Long>();
private ConcurrentLinkedQueue<String> queriesToReindex = new ConcurrentLinkedQueue<String>();
private DocRouter docRouter;
/** The string representation of the shard key. */
private Optional<String> shardKey;
/** The property to use for determining the shard. */
private Optional<QName> shardProperty = Optional.empty();
private ConcurrentLinkedQueue<Long> transactionsToReindex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> transactionsToIndex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> transactionsToPurge = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> nodesToReindex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> nodesToIndex = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<Long> nodesToPurge = new ConcurrentLinkedQueue<>();
private ConcurrentLinkedQueue<String> queriesToReindex = new ConcurrentLinkedQueue<>();
public MetadataTracker(Properties p, SOLRAPIClient client, String coreName,
public MetadataTracker(final boolean isMaster, Properties p, SOLRAPIClient client, String coreName,
InformationServer informationServer)
{
super(p, client, coreName, informationServer, Tracker.Type.MetaData);
super(isMaster, p, client, coreName, informationServer, Tracker.Type.METADATA);
transactionDocsBatchSize = Integer.parseInt(p.getProperty("alfresco.transactionDocsBatchSize", "100"));
shardMethod = p.getProperty("shard.method", SHARD_METHOD_DBID);
shardKey = ofNullable(p.getProperty(SHARD_KEY_KEY));
firstUpdateShardProperty();
docRouter = DocRouterFactory.getRouter(p, ShardMethodEnum.getShardMethod(shardMethod));
nodeBatchSize = Integer.parseInt(p.getProperty("alfresco.nodeBatchSize", "10"));
threadHandler = new ThreadHandler(p, coreName, "MetadataTracker");
}
/**
* Set the shard property using the shard key.
*/
private void updateShardProperty()
{
shardKey.ifPresent(shardKeyName -> {
Optional<QName> updatedShardProperty = getShardProperty(shardKeyName);
if (!shardProperty.equals(updatedShardProperty))
{
if (updatedShardProperty.isEmpty())
{
log.warn("The model defining " + shardKeyName + " property has been disabled");
}
else
{
log.info("New " + SHARD_KEY_KEY + " property found for " + shardKeyName);
}
}
shardProperty = updatedShardProperty;
});
}
private void firstUpdateShardProperty()
{
shardKey.ifPresent( shardKeyName -> {
updateShardProperty();
if (shardProperty.isEmpty())
{
log.warn("Sharding property " + SHARD_KEY_KEY + " was set to " + shardKeyName + ", but no such property was found.");
}
});
}
MetadataTracker()
{
super(Tracker.Type.MetaData);
}
public DocRouter getDocRouter() {
return this.docRouter;
super(Tracker.Type.METADATA);
}
@Override
@@ -160,7 +94,8 @@ public class MetadataTracker extends AbstractTracker implements Tracker
}
}
public void maintenance() throws Exception {
public void maintenance() throws Exception
{
purgeTransactions();
purgeNodes();
reindexTransactions();
@@ -170,7 +105,8 @@ public class MetadataTracker extends AbstractTracker implements Tracker
indexNodes();
}
public boolean hasMaintenance() throws Exception {
public boolean hasMaintenance()
{
return transactionsToReindex.size() > 0 ||
transactionsToIndex.size() > 0 ||
transactionsToPurge.size() > 0 ||
@@ -180,30 +116,11 @@ public class MetadataTracker extends AbstractTracker implements Tracker
queriesToReindex.size() > 0;
}
private void trackRepository() throws IOException, AuthenticationException, JSONException, EncoderException
{
log.debug("####### MetadataTracker trackRepository Start #######");
checkShutdown();
if(!isMaster && isSlave)
{
// Dynamic registration
/*
* This section allows Solr's master/slave setup to be used with dynamic shard registration.
* In this scenario the slave is polling a "tracking" Solr node. The code below calls
* the repo to register the state of the node without pulling any real transactions from the repo.
*
* This allows the repo to register the replica so that it will be included in queries. But the slave Solr node
* will pull its data from a "tracking" Solr node using Solr's master/slave replication, rather then tracking the repository.
*
*/
ShardState shardstate = getShardState();
client.getTransactions(0L, null, 0L, null, 0, shardstate);
return;
}
// Check we are tracking the correct repository
TrackerState state = super.getTrackerState();
log.debug("####### MetadataTracker check CYCLE #######");
@@ -233,67 +150,6 @@ public class MetadataTracker extends AbstractTracker implements Tracker
trackTransactions();
}
/**
* The {@link ShardState}, as the name suggests, encapsulates/stores the state of the shard which hosts this
* {@link MetadataTracker} instance.
*
* The {@link ShardState} is primarily used in two places:
*
* <ul>
* <li>Transaction tracking: (see {@link #trackTransactions()}): for pulling/tracking transactions from Alfresco</li>
* <li>
* DynamicSharding: when the {@link MetadataTracker} is running on a slave instance it doesn't actually act
* as a tracker, it calls Alfresco to register the state of the node (the shard) without pulling any transactions.
* As consequence of that, Alfresco will be aware about the shard which will be included in subsequent queries.
* </li>
* </ul>
*
* @return the {@link ShardState} instance which stores the current state of the hosting shard.
*/
ShardState getShardState()
{
TrackerState transactionsTrackerState = super.getTrackerState();
TrackerState changeSetsTrackerState =
of(infoSrv.getAdminHandler())
.map(AlfrescoCoreAdminHandler::getTrackerRegistry)
.map(registry -> registry.getTrackerForCore(coreName, AclTracker.class))
.map(Tracker::getTrackerState)
.orElse(transactionsTrackerState);
HashMap<String, String> propertyBag = new HashMap<>();
propertyBag.put("coreName", coreName);
HashMap<String, String> extendedPropertyBag = new HashMap<>(propertyBag);
updateShardProperty();
shardProperty.ifPresent(p -> extendedPropertyBag.putAll(docRouter.getProperties(p)));
return ShardStateBuilder.shardState()
.withMaster(isMaster)
.withLastUpdated(System.currentTimeMillis())
.withLastIndexedChangeSetCommitTime(changeSetsTrackerState.getLastIndexedChangeSetCommitTime())
.withLastIndexedChangeSetId(changeSetsTrackerState.getLastIndexedChangeSetId())
.withLastIndexedTxCommitTime(transactionsTrackerState.getLastIndexedTxCommitTime())
.withLastIndexedTxId(transactionsTrackerState.getLastIndexedTxId())
.withPropertyBag(extendedPropertyBag)
.withShardInstance()
.withBaseUrl(infoSrv.getBaseUrl())
.withPort(infoSrv.getPort())
.withHostName(infoSrv.getHostName())
.withShard()
.withInstance(shardInstance)
.withFloc()
.withNumberOfShards(shardCount)
.withAddedStoreRef(storeRef)
.withTemplate(shardTemplate)
.withHasContent(transformContent)
.withShardMethod(ShardMethodEnum.getShardMethod(shardMethod))
.withPropertyBag(propertyBag)
.endFloc()
.endShard()
.endShardInstance()
.build();
}
/**
* Checks the first and last TX time
* @param state the state of this tracker
@@ -1230,34 +1086,4 @@ public class MetadataTracker extends AbstractTracker implements Tracker
{
this.queriesToReindex.offer(query);
}
/**
* Given the field name, returns the name of the property definition.
* If the property definition is not found, Empty optional is returned.
*
* @param field
*
* @return the name of the associated property definition if present, Optional.Empty() otherwise
*
*/
public static Optional<QName> getShardProperty(String field)
{
if (StringUtils.isBlank(field))
{
throw new IllegalArgumentException("Sharding property " + SHARD_KEY_KEY + " has not been set.");
}
AlfrescoSolrDataModel dataModel = AlfrescoSolrDataModel.getInstance();
NamespaceDAO namespaceDAO = dataModel.getNamespaceDAO();
DictionaryService dictionaryService = dataModel.getDictionaryService(CMISStrictDictionaryService.DEFAULT);
PropertyDefinition propertyDef = QueryParserUtils.matchPropertyDefinition("http://www.alfresco.org/model/content/1.0",
namespaceDAO,
dictionaryService,
field);
if (propertyDef == null)
{
return Optional.empty();
}
return of(propertyDef.getName());
}
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ModelTracker.java
+6 -6
View File
@@ -103,10 +103,10 @@ public class ModelTracker extends AbstractTracker implements Tracker
public ModelTracker(String solrHome, Properties p, SOLRAPIClient client, String coreName,
InformationServer informationServer)
{
super(p, client, coreName, informationServer, Tracker.Type.Model);
super(p, client, coreName, informationServer, Tracker.Type.MODEL);
String normalSolrHome = SolrResourceLoader.normalizeDir(solrHome);
alfrescoModelDir = new File(ConfigUtil.locateProperty("solr.model.dir", normalSolrHome+"alfrescoModels"));
log.info("Alfresco Model dir " + alfrescoModelDir);
LOGGER.info("Alfresco Model dir " + alfrescoModelDir);
if (!alfrescoModelDir.exists())
{
alfrescoModelDir.mkdir();
@@ -187,7 +187,7 @@ public class ModelTracker extends AbstractTracker implements Tracker
*/
ModelTracker()
{
super(Tracker.Type.Model);
super(Tracker.Type.MODEL);
}
@Override
@@ -197,7 +197,7 @@ public class ModelTracker extends AbstractTracker implements Tracker
int registeredSearcherCount = this.infoSrv.getRegisteredSearcherCount();
if (registeredSearcherCount >= getMaxLiveSearchers())
{
log.info(".... skipping tracking registered searcher count = " + registeredSearcherCount);
LOGGER.info(".... skipping tracking registered searcher count = " + registeredSearcherCount);
return;
}
@@ -268,7 +268,7 @@ public class ModelTracker extends AbstractTracker implements Tracker
}
catch (Throwable t)
{
log.error("Model tracking failed for core: "+ coreName, t);
LOGGER.error("Model tracking failed for core: "+ coreName, t);
}
}
@@ -534,7 +534,7 @@ public class ModelTracker extends AbstractTracker implements Tracker
{
loadedModels.add(modelName);
}
log.info("Loading model " + model.getName());
LOGGER.info("Loading model " + model.getName());
}
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/NodeStatePublisher.java
+225
View File
@@ -0,0 +1,225 @@
/*
* #%L
* Alfresco Solr Client
* %%
* Copyright (C) 2005 - 2016 Alfresco Software Limited
* %%
* This file is part of the Alfresco software.
* If the software was purchased under a paid Alfresco license, the terms of
* the paid license agreement will prevail. Otherwise, the software is
* provided under the following open source license terms:
*
* 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 <http://www.gnu.org/licenses/>.
* #L%
*/
package org.alfresco.solr.tracker;
import static java.util.Optional.of;
import static java.util.Optional.ofNullable;
import static org.alfresco.solr.tracker.DocRouterFactory.SHARD_KEY_KEY;
import org.alfresco.opencmis.dictionary.CMISStrictDictionaryService;
import org.alfresco.repo.dictionary.NamespaceDAO;
import org.alfresco.repo.index.shard.ShardMethodEnum;
import org.alfresco.repo.index.shard.ShardState;
import org.alfresco.repo.index.shard.ShardStateBuilder;
import org.alfresco.repo.search.impl.QueryParserUtils;
import org.alfresco.service.cmr.dictionary.DictionaryService;
import org.alfresco.service.cmr.dictionary.PropertyDefinition;
import org.alfresco.service.namespace.QName;
import org.alfresco.solr.AlfrescoCoreAdminHandler;
import org.alfresco.solr.AlfrescoSolrDataModel;
import org.alfresco.solr.InformationServer;
import org.alfresco.solr.TrackerState;
import org.alfresco.solr.client.SOLRAPIClient;
import org.apache.commons.lang3.StringUtils;
import java.util.HashMap;
import java.util.Optional;
import java.util.Properties;
/**
* Superclass for all components which are able to inform Alfresco about the hosting node state.
* This has been introduced in SEARCH-1752 for splitting the dual responsibility of the {@link MetadataTracker}.
* As consequence of that, this class contains all the members needed for obtaining a valid
* {@link org.alfresco.repo.index.shard.ShardState} that can be periodically communicated to Alfresco.
*
* @author Andrea Gazzarini
* @since 1.5
* @see <a href="https://issues.alfresco.com/jira/browse/SEARCH-1752">SEARCH-1752</a>
*/
public abstract class NodeStatePublisher extends AbstractTracker
{
DocRouter docRouter;
private final boolean isMaster;
/** The string representation of the shard key. */
private Optional<String> shardKey;
/** The property to use for determining the shard. */
protected Optional<QName> shardProperty = Optional.empty();
NodeStatePublisher(
boolean isMaster,
Properties p,
SOLRAPIClient client,
String coreName,
InformationServer informationServer,
Type type)
{
super(p, client, coreName, informationServer, type);
this.isMaster = isMaster;
shardMethod = p.getProperty("shard.method", SHARD_METHOD_DBID);
shardKey = ofNullable(p.getProperty(SHARD_KEY_KEY));
firstUpdateShardProperty();
docRouter = DocRouterFactory.getRouter(p, ShardMethodEnum.getShardMethod(shardMethod));
}
NodeStatePublisher(Type type)
{
super(type);
this.isMaster = false;
}
private void firstUpdateShardProperty()
{
shardKey.ifPresent( shardKeyName -> {
updateShardProperty();
if (shardProperty.isEmpty())
{
LOGGER.warn("Sharding property {} was set to {}, but no such property was found.", SHARD_KEY_KEY, shardKeyName);
}
});
}
/**
* Set the shard property using the shard key.
*/
void updateShardProperty()
{
shardKey.ifPresent(shardKeyName -> {
Optional<QName> updatedShardProperty = getShardProperty(shardKeyName);
if (!shardProperty.equals(updatedShardProperty))
{
if (updatedShardProperty.isEmpty())
{
LOGGER.warn("The model defining {} property has been disabled", shardKeyName);
}
else
{
LOGGER.info("New {} property found for {} ", SHARD_KEY_KEY, shardKeyName);
}
}
shardProperty = updatedShardProperty;
});
}
/**
* Given the field name, returns the name of the property definition.
* If the property definition is not found, Empty optional is returned.
*
* @param field the field name.
* @return the name of the associated property definition if present, Optional.Empty() otherwise
*/
static Optional<QName> getShardProperty(String field)
{
if (StringUtils.isBlank(field))
{
throw new IllegalArgumentException("Sharding property " + SHARD_KEY_KEY + " has not been set.");
}
AlfrescoSolrDataModel dataModel = AlfrescoSolrDataModel.getInstance();
NamespaceDAO namespaceDAO = dataModel.getNamespaceDAO();
DictionaryService dictionaryService = dataModel.getDictionaryService(CMISStrictDictionaryService.DEFAULT);
PropertyDefinition propertyDef = QueryParserUtils.matchPropertyDefinition("http://www.alfresco.org/model/content/1.0",
namespaceDAO,
dictionaryService,
field);
return ofNullable(propertyDef).map(PropertyDefinition::getName);
}
/**
* The {@link ShardState}, as the name suggests, encapsulates/stores the state of the shard which hosts this
* {@link MetadataTracker} instance.
*
* The {@link ShardState} is primarily used in two places:
*
* <ul>
* <li>Transaction tracking: (see {@link MetadataTracker#trackTransactions()}): for pulling/tracking transactions from Alfresco</li>
* <li>
* DynamicSharding: when the {@link MetadataTracker} is running on a slave instance it doesn't actually act
* as a tracker, it calls Alfresco to register the state of the node (the shard) without pulling any transactions.
* As consequence of that, Alfresco will be aware about the shard which will be included in subsequent queries.
* </li>
* </ul>
*
* @return the {@link ShardState} instance which stores the current state of the hosting shard.
*/
ShardState getShardState()
{
TrackerState transactionsTrackerState = getTrackerState();
TrackerState changeSetsTrackerState =
of(infoSrv.getAdminHandler())
.map(AlfrescoCoreAdminHandler::getTrackerRegistry)
.map(registry -> registry.getTrackerForCore(coreName, AclTracker.class))
.map(Tracker::getTrackerState)
.orElse(transactionsTrackerState);
HashMap<String, String> propertyBag = new HashMap<>();
propertyBag.put("coreName", coreName);
HashMap<String, String> extendedPropertyBag = new HashMap<>(propertyBag);
updateShardProperty();
shardProperty.ifPresent(p -> extendedPropertyBag.putAll(docRouter.getProperties(p)));
return ShardStateBuilder.shardState()
.withMaster(isMaster)
.withLastUpdated(System.currentTimeMillis())
.withLastIndexedChangeSetCommitTime(changeSetsTrackerState.getLastIndexedChangeSetCommitTime())
.withLastIndexedChangeSetId(changeSetsTrackerState.getLastIndexedChangeSetId())
.withLastIndexedTxCommitTime(transactionsTrackerState.getLastIndexedTxCommitTime())
.withLastIndexedTxId(transactionsTrackerState.getLastIndexedTxId())
.withPropertyBag(extendedPropertyBag)
.withShardInstance()
.withBaseUrl(infoSrv.getBaseUrl())
.withPort(infoSrv.getPort())
.withHostName(infoSrv.getHostName())
.withShard()
.withInstance(shardInstance)
.withFloc()
.withNumberOfShards(shardCount)
.withAddedStoreRef(storeRef)
.withTemplate(shardTemplate)
.withHasContent(transformContent)
.withShardMethod(ShardMethodEnum.getShardMethod(shardMethod))
.withPropertyBag(propertyBag)
.endFloc()
.endShard()
.endShardInstance()
.build();
}
/**
* Returns the {@link DocRouter} instance in use on this node.
*
* @return the {@link DocRouter} instance in use on this node.
*/
public DocRouter getDocRouter()
{
return this.docRouter;
}
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/SlaveNodeStatePublisher.java
+82
View File
@@ -0,0 +1,82 @@
package org.alfresco.solr.tracker;
import static org.alfresco.solr.tracker.Tracker.Type.NODE_STATE_PUBLISHER;
import org.alfresco.httpclient.AuthenticationException;
import org.alfresco.repo.index.shard.ShardState;
import org.alfresco.solr.SolrInformationServer;
import org.alfresco.solr.TrackerState;
import org.alfresco.solr.client.SOLRAPIClient;
import org.apache.commons.codec.EncoderException;
import java.io.IOException;
import java.util.Properties;
/**
* Despite belonging to the Tracker ecosystem, this component is actually a publisher, which periodically informs
* Alfresco about the state of the hosting slave node.
* As the name suggests, this worker is scheduled only when the hosting node acts as a slave.
* It allows Solr's master/slave setup to be used with dynamic shard registration.
*
* In this scenario the slave is polling a "tracking" Solr node. The tracker below calls
* the repo to register the state of the node without pulling any real transactions from the repo.
*
* This allows the repo to register the replica so that it will be included in queries. But the slave Solr node
* will pull its data from a "tracking" Solr node using Solr's master/slave replication, rather then tracking the repository.
*
* @author Andrea Gazzarini
* @since 1.5
*/
public class SlaveNodeStatePublisher extends NodeStatePublisher
{
public SlaveNodeStatePublisher(
boolean isMaster,
Properties coreProperties,
SOLRAPIClient repositoryClient,
String name,
SolrInformationServer informationServer)
{
super(isMaster, coreProperties, repositoryClient, name, informationServer, NODE_STATE_PUBLISHER);
}
@Override
protected void doTrack()
{
try
{
ShardState shardstate = getShardState();
client.getTransactions(0L, null, 0L, null, 0, shardstate);
}
catch (EncoderException | IOException | AuthenticationException exception )
{
LOGGER.error("Unable to publish this node state. " +
"A failure condition has been met during the outbound subscription message encoding process. " +
"See the stacktrace below for further details.", exception);
}
}
@Override
public void maintenance()
{
// Do nothing here
}
@Override
public boolean hasMaintenance()
{
return false;
}
/**
* When running in a slave mode, we need to recreate the tracker state every time.
* This because in that context we don't have any tracker updating the state (e.g. lastIndexedChangeSetCommitTime,
* lastIndexedChangeSetId)
*
* @return a new, fresh and up to date instance of {@link TrackerState}.
*/
@Override
public TrackerState getTrackerState()
{
return infoSrv.getTrackerInitialState();
}
}
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/SolrTrackerScheduler.java
+16 -11
View File
@@ -76,18 +76,20 @@ public class SolrTrackerScheduler
{
log.error("Failed to schedule " + jobType + " Job.", e);
}
private String getCron(Properties props, String cronType)
{
String cron = props.getProperty(cronType);
return cron == null ? props.getProperty("alfresco.cron",DEFAULT_CRON) : cron;
return cron == null ? props.getProperty("alfresco.cron", DEFAULT_CRON) : cron;
}
/**
* Schedules individual trackers based on the solrcore properties.
*
* @author Michael Suzuki
* @param tracker
* @param coreName
* @param props
* @param tracker the tracker to bo scheduled.
* @param coreName the owning core name.
* @param props the core properties.
*/
public void schedule(Tracker tracker, String coreName, Properties props)
{
@@ -98,27 +100,30 @@ public class SolrTrackerScheduler
Trigger trigger;
try
{
String cron = null;
String cron;
switch (tracker.getType())
{
case ACL:
cron = getCron(props,"alfresco.acl.tracker.cron");
break;
case Model:
case MODEL:
cron = getCron(props,"alfresco.model.tracker.cron");
break;
case Content:
case CONTENT:
cron = getCron(props,"alfresco.content.tracker.cron");
break;
case MetaData:
case METADATA:
cron = getCron(props,"alfresco.metadata.tracker.cron");
break;
case Cascade:
case CASCADE:
cron = getCron(props,"alfresco.cascade.tracker.cron");
break;
case Commit:
case COMMIT:
cron = getCron(props,"alfresco.commit.tracker.cron");
break;
case NODE_STATE_PUBLISHER:
cron = getCron(props,"alfresco.nodestate.tracker.cron");
break;
default:
cron = props.getProperty("alfresco.cron",DEFAULT_CRON);
break;
@@ -161,7 +166,7 @@ public class SolrTrackerScheduler
* identical to the instance that is passed in. If they are identical then the job is deleted.
* Otherwise, another core (of the same name) scheduled this job, so its left alone.
*
* @param coreName
* @param coreName the core name.
* @param tracker Specific instance of a tracker
*/
public void deleteJobForTrackerInstance(String coreName, Tracker tracker)
search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/Tracker.java
+11 -7
View File
@@ -36,7 +36,9 @@ public interface Tracker
String getAlfrescoVersion();
void setShutdown(boolean shutdown);
boolean isAlreadyInShutDownMode();
void shutdown();
boolean getRollback();
@@ -51,12 +53,14 @@ public interface Tracker
Type getType();
enum Type{
Model,
Content,
enum Type
{
MODEL,
CONTENT,
ACL,
Cascade,
Commit,
MetaData
CASCADE,
COMMIT,
METADATA,
NODE_STATE_PUBLISHER
}
}
search-services/alfresco-search/src/test/java/org/alfresco/solr/lifecycle/SolrCoreLoadListenerTest.java
+2 -2
View File
@@ -92,7 +92,7 @@ public class SolrCoreLoadListenerTest
@Test
public void coreTrackersRegistrationAndScheduling()
{
List<Tracker> coreTrackers = listener.createCoreTrackers(core.getName(), registry, coreProperties, scheduler, api, informationServer);
List<Tracker> coreTrackers = listener.createAndScheduleCoreTrackers(core, registry, coreProperties, scheduler, api, informationServer);
verify(registry).register(eq(coreName), any(AclTracker.class));
verify(registry).register(eq(coreName), any(ContentTracker.class));
@@ -113,7 +113,7 @@ public class SolrCoreLoadListenerTest
List<Tracker> coreTrackers =
asList(mock(AclTracker.class), mock(ContentTracker.class), mock(MetadataTracker.class), mock(CascadeTracker.class));
listener.shutdownTrackers(coreName, coreTrackers, scheduler);
listener.shutdownTrackers(core, coreTrackers, scheduler, false);
coreTrackers.forEach(tracker -> verify(tracker).setShutdown(true));
coreTrackers.forEach(tracker -> verify(scheduler).deleteJobForTrackerInstance(core.getName(), tracker));
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ContentTrackerTest.java
+1 -1
View File
@@ -137,7 +137,7 @@ public class ContentTrackerTest
@Test
public void typeCheck()
{
Assert.assertTrue(contentTracker.getType().equals(Tracker.Type.Content));
Assert.assertTrue(contentTracker.getType().equals(Tracker.Type.CONTENT));
}
}
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/MetadataTrackerTest.java
+7 -6
View File
@@ -47,7 +47,7 @@ import org.junit.runner.RunWith;
import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.runners.MockitoJUnitRunner;
import org.mockito.junit.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class MetadataTrackerTest
@@ -58,20 +58,23 @@ public class MetadataTrackerTest
@Mock
private SOLRAPIClient repositoryClient;
private String coreName = "theCoreName";
@Mock
private InformationServer srv;
@Spy
private Properties props;
@Mock
private TrackerStats trackerStats;
@Before
public void setUp() throws Exception
public void setUp()
{
doReturn("workspace://SpacesStore").when(props).getProperty("alfresco.stores");
when(srv.getTrackerStats()).thenReturn(trackerStats);
this.metadataTracker = spy(new MetadataTracker(props, repositoryClient, coreName, srv));
String coreName = "theCoreName";
this.metadataTracker = spy(new MetadataTracker(true, props, repositoryClient, coreName, srv));
ModelTracker modelTracker = mock(ModelTracker.class);
when(modelTracker.hasModels()).thenReturn(true);
@@ -197,6 +200,4 @@ public class MetadataTrackerTest
assertSame(nodes4Tx, nodes);
}
}
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ModelTrackerTest.java
-2
View File
@@ -102,8 +102,6 @@ public class ModelTrackerTest
when(props.getProperty("alfresco.stores", "workspace://SpacesStore")).thenReturn("workspace://SpacesStore");
when(props.getProperty("alfresco.batch.count", "5000")).thenReturn("5000");
when(props.getProperty("alfresco.maxLiveSearchers", "2")).thenReturn("2");
when(props.getProperty("enable.slave", "false")).thenReturn("false");
when(props.getProperty("enable.master", "true")).thenReturn("true");
when(props.getProperty("shard.count", "1")).thenReturn("1");
when(props.getProperty("shard.instance", "0")).thenReturn("0");
when(this.srv.getTrackerStats()).thenReturn(trackerStats);
search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/SolrTrackerSchedulerTest.java
+1 -3
View File
@@ -73,8 +73,6 @@ public class SolrTrackerSchedulerTest
props.put("alfresco.stores", "workspace://SpacesStore");
props.put("alfresco.batch.count", "5000");
props.put("alfresco.maxLiveSearchers", "2");
props.put("enable.slave", "false");
props.put("enable.master", "true");
props.put("shard.count", "1");
props.put("shard.instance", "0");
props.put("shard.method", "SHARD_METHOD_DBID");
@@ -128,7 +126,7 @@ public class SolrTrackerSchedulerTest
{
String exp = "0/4 * * * * ? *";
props.put("alfresco.metadata.tracker.cron", exp);
MetadataTracker metadataTracker = new MetadataTracker(props, client, exp, informationServer);
MetadataTracker metadataTracker = new MetadataTracker(true, props, client, exp, informationServer);
this.trackerScheduler.schedule(metadataTracker, CORE_NAME, props);
verify(spiedQuartzScheduler).scheduleJob(any(JobDetail.class), any(Trigger.class));
checkCronExpression(exp);