diff --git a/e2e-test/generator-alfresco-docker-compose/generators/app/templates/images/search/Dockerfile b/e2e-test/generator-alfresco-docker-compose/generators/app/templates/images/search/Dockerfile index d43b4f910..163f75682 100755 --- a/e2e-test/generator-alfresco-docker-compose/generators/app/templates/images/search/Dockerfile +++ b/e2e-test/generator-alfresco-docker-compose/generators/app/templates/images/search/Dockerfile @@ -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*\ @@ -64,7 +63,6 @@ RUN if [ "$ENABLE_MASTER" == "true" ] ; then \ ' ${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*\ @@ -131,4 +129,4 @@ fi RUN mkdir ${DIST_DIR}/keystore \ && chown -R solr:solr ${DIST_DIR}/keystore -VOLUME ["${DIST_DIR}/keystore"] \ No newline at end of file +VOLUME ["${DIST_DIR}/keystore"] diff --git a/search-services/alfresco-search/doc/architecture/trackers/00001-content-tracker.md b/search-services/alfresco-search/doc/architecture/trackers/00001-content-tracker.md index 77c0159cb..e267a79a9 100644 --- a/search-services/alfresco-search/doc/architecture/trackers/00001-content-tracker.md +++ b/search-services/alfresco-search/doc/architecture/trackers/00001-content-tracker.md @@ -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| | | 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 9255db2a0..f714307dc 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 @@ -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(); diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/lifecycle/SolrCoreLoadListener.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/lifecycle/SolrCoreLoadListener.java index c1ec3719d..0b41cba15 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/lifecycle/SolrCoreLoadListener.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/lifecycle/SolrCoreLoadListener.java @@ -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 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 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 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 createCoreTrackers(String coreName, - TrackerRegistry trackerRegistry, - Properties props, - SolrTrackerScheduler scheduler, - SOLRAPIClient repositoryClient, - SolrInformationServer srv) + List createAndScheduleCoreTrackers(SolrCore core, + TrackerRegistry trackerRegistry, + Properties props, + SolrTrackerScheduler scheduler, + SOLRAPIClient repositoryClient, + SolrInformationServer srv) { - List 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 the tracker instance. + * @return the registered and scheduled tracker instance. + */ + private 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 coreTrackers, SolrTrackerScheduler scheduler) + void shutdownTrackers(SolrCore core, Collection 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. diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AbstractTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AbstractTracker.java index ae944dd4b..d4429f92b 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AbstractTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AbstractTracker.java @@ -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: + * *
    *
  1. Purge
    2. *
  2. Reindex
    3. *
  3. Index
    4. *
  4. Track repository
    5. *
- * @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; } - - -} +} \ No newline at end of file diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java index f21ef9085..7851920cd 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/AclTracker.java @@ -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(); diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java index 363206404..56f0f6b3f 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/CascadeTracker.java @@ -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 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 3acbd4a51..d8a301cf9 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 @@ -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 trackers) { - super(p, client, coreName, informationServer, Tracker.Type.Commit); + super(p, client, coreName, informationServer, Tracker.Type.COMMIT); //Set the trackers for(Tracker tracker : trackers) { diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java index 28b4fc698..4c3ed15a8 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java @@ -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 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 58778d865..f118c523b 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 @@ -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 transactionsToReindex = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue transactionsToIndex = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue transactionsToPurge = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue nodesToReindex = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue nodesToIndex = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue nodesToPurge = new ConcurrentLinkedQueue(); - private ConcurrentLinkedQueue queriesToReindex = new ConcurrentLinkedQueue(); - private DocRouter docRouter; - /** The string representation of the shard key. */ - private Optional shardKey; - /** The property to use for determining the shard. */ - private Optional shardProperty = Optional.empty(); + private ConcurrentLinkedQueue transactionsToReindex = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue transactionsToIndex = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue transactionsToPurge = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue nodesToReindex = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue nodesToIndex = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue nodesToPurge = new ConcurrentLinkedQueue<>(); + private ConcurrentLinkedQueue 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 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: - * - *
    - *
  • Transaction tracking: (see {@link #trackTransactions()}): for pulling/tracking transactions from Alfresco
    • - *
  • - * 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. - *
    • - *
- * - * @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 propertyBag = new HashMap<>(); - propertyBag.put("coreName", coreName); - HashMap 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 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()); - } } diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ModelTracker.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ModelTracker.java index c19c31eaa..034d14228 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ModelTracker.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ModelTracker.java @@ -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()); } } diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/NodeStatePublisher.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/NodeStatePublisher.java new file mode 100644 index 000000000..d6fd2d78a --- /dev/null +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/NodeStatePublisher.java @@ -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 . + * #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 SEARCH-1752 + */ +public abstract class NodeStatePublisher extends AbstractTracker +{ + DocRouter docRouter; + private final boolean isMaster; + + /** The string representation of the shard key. */ + private Optional shardKey; + + /** The property to use for determining the shard. */ + protected Optional 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 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 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: + * + *
    + *
  • Transaction tracking: (see {@link MetadataTracker#trackTransactions()}): for pulling/tracking transactions from Alfresco
    • + *
  • + * 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. + *
    • + *
+ * + * @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 propertyBag = new HashMap<>(); + propertyBag.put("coreName", coreName); + HashMap 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; + } +} diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/SlaveNodeStatePublisher.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/SlaveNodeStatePublisher.java new file mode 100644 index 000000000..4e7ccf709 --- /dev/null +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/SlaveNodeStatePublisher.java @@ -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(); + } +} \ No newline at end of file diff --git a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/SolrTrackerScheduler.java b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/SolrTrackerScheduler.java index 78d2683ab..c584d80fb 100644 --- a/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/SolrTrackerScheduler.java +++ b/search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/SolrTrackerScheduler.java @@ -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) 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 9f6654121..05fe0766a 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 @@ -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 } } diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/lifecycle/SolrCoreLoadListenerTest.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/lifecycle/SolrCoreLoadListenerTest.java index 8462aefc0..e1ec88140 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/lifecycle/SolrCoreLoadListenerTest.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/lifecycle/SolrCoreLoadListenerTest.java @@ -92,7 +92,7 @@ public class SolrCoreLoadListenerTest @Test public void coreTrackersRegistrationAndScheduling() { - List coreTrackers = listener.createCoreTrackers(core.getName(), registry, coreProperties, scheduler, api, informationServer); + List 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 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)); diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ContentTrackerTest.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ContentTrackerTest.java index a2e74acfb..e3d6112da 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ContentTrackerTest.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ContentTrackerTest.java @@ -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)); } } diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/MetadataTrackerTest.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/MetadataTrackerTest.java index 56f444a9d..5c68f5974 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/MetadataTrackerTest.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/MetadataTrackerTest.java @@ -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); } - - } diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ModelTrackerTest.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ModelTrackerTest.java index 318eb27f5..71ee93280 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ModelTrackerTest.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/ModelTrackerTest.java @@ -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); diff --git a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/SolrTrackerSchedulerTest.java b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/SolrTrackerSchedulerTest.java index 822b01a32..e685b14be 100644 --- a/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/SolrTrackerSchedulerTest.java +++ b/search-services/alfresco-search/src/test/java/org/alfresco/solr/tracker/SolrTrackerSchedulerTest.java @@ -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);