/* * The contents of this file are subject to the terms of the Common Development and * Distribution License (the License). You may not use this file except in compliance with the * License. * * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the * specific language governing permission and limitations under the License. * * When distributing Covered Software, include this CDDL Header Notice in each file and include * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL * Header, with the fields enclosed by brackets [] replaced by your own identifying * information: "Portions Copyright [year] [name of copyright owner]". * * Copyright 2006-2010 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.replication.plugin; import static org.forgerock.opendj.ldap.ResultCode.*; import static org.opends.messages.ReplicationMessages.*; import static org.opends.messages.ToolMessages.*; import static org.opends.server.config.ConfigConstants.*; import static org.opends.server.protocols.internal.InternalClientConnection.*; import static org.opends.server.protocols.internal.Requests.*; import static org.opends.server.replication.plugin.EntryHistorical.*; import static org.opends.server.replication.protocol.OperationContext.*; import static org.opends.server.util.CollectionUtils.*; import static org.opends.server.util.ServerConstants.*; import static org.opends.server.util.StaticUtils.*; import java.io.File; import java.io.InputStream; import java.io.OutputStream; import java.io.StringReader; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.Date; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.NoSuchElementException; import java.util.Set; import java.util.SortedMap; import java.util.StringTokenizer; import java.util.TreeMap; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import net.jcip.annotations.GuardedBy; import org.forgerock.i18n.LocalizableMessage; import org.forgerock.i18n.LocalizedIllegalArgumentException; import org.forgerock.i18n.slf4j.LocalizedLogger; import org.forgerock.opendj.adapter.server3x.Converters; import org.forgerock.opendj.config.server.ConfigChangeResult; import org.forgerock.opendj.config.server.ConfigException; import org.forgerock.opendj.config.server.ConfigurationChangeListener; import org.forgerock.opendj.ldap.AVA; import org.forgerock.opendj.ldap.ByteString; import org.forgerock.opendj.ldap.DN; import org.forgerock.opendj.ldap.DecodeException; import org.forgerock.opendj.ldap.ModificationType; import org.forgerock.opendj.ldap.RDN; import org.forgerock.opendj.ldap.ResultCode; import org.forgerock.opendj.ldap.SearchScope; import org.forgerock.opendj.ldap.schema.AttributeType; import org.forgerock.opendj.ldap.schema.ObjectClass; import org.forgerock.opendj.ldap.schema.Schema; import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.IsolationPolicy; import org.forgerock.opendj.server.config.server.ExternalChangelogDomainCfg; import org.forgerock.opendj.server.config.server.ReplicationDomainCfg; import org.forgerock.util.annotations.VisibleForTesting; import org.opends.server.api.AlertGenerator; import org.opends.server.api.DirectoryThread; import org.opends.server.api.LocalBackend; import org.opends.server.api.LocalBackend.BackendOperation; import org.opends.server.api.LocalBackendInitializationListener; import org.opends.server.api.MonitorData; import org.opends.server.api.ServerShutdownListener; import org.opends.server.api.SynchronizationProvider; import org.opends.server.backends.task.Task; import org.opends.server.config.ConfigConstants; import org.opends.server.config.ConfigurationHandler; import org.opends.server.controls.PagedResultsControl; import org.opends.server.core.AddOperation; import org.opends.server.core.DeleteOperation; import org.opends.server.core.DirectoryServer; import org.opends.server.core.LockFileManager; import org.opends.server.core.ModifyDNOperation; import org.opends.server.core.ModifyDNOperationBasis; import org.opends.server.core.ModifyOperation; import org.opends.server.core.ModifyOperationBasis; import org.opends.server.core.ServerContext; import org.opends.server.protocols.internal.InternalClientConnection; import org.opends.server.protocols.internal.InternalSearchListener; import org.opends.server.protocols.internal.InternalSearchOperation; import org.opends.server.protocols.internal.Requests; import org.opends.server.protocols.internal.SearchRequest; import org.opends.server.protocols.ldap.LDAPAttribute; import org.opends.server.protocols.ldap.LDAPControl; import org.opends.server.protocols.ldap.LDAPFilter; import org.opends.server.protocols.ldap.LDAPModification; import org.opends.server.replication.common.CSN; import org.opends.server.replication.common.ServerState; import org.opends.server.replication.common.ServerStatus; import org.opends.server.replication.common.StatusMachineEvent; import org.opends.server.replication.protocol.AddContext; import org.opends.server.replication.protocol.AddMsg; import org.opends.server.replication.protocol.DeleteContext; import org.opends.server.replication.protocol.DeleteMsg; import org.opends.server.replication.protocol.LDAPUpdateMsg; import org.opends.server.replication.protocol.ModifyContext; import org.opends.server.replication.protocol.ModifyDNMsg; import org.opends.server.replication.protocol.ModifyDnContext; import org.opends.server.replication.protocol.ModifyMsg; import org.opends.server.replication.protocol.OperationContext; import org.opends.server.replication.protocol.RoutableMsg; import org.opends.server.replication.protocol.UpdateMsg; import org.opends.server.replication.service.DSRSShutdownSync; import org.opends.server.replication.service.ReplicationBroker; import org.opends.server.replication.service.ReplicationDomain; import org.opends.server.tasks.PurgeConflictsHistoricalTask; import org.opends.server.tasks.TaskUtils; import org.opends.server.types.AdditionalLogItem; import org.opends.server.types.Attribute; import org.opends.server.types.AttributeBuilder; import org.opends.server.types.Attributes; import org.opends.server.types.Control; import org.opends.server.types.DirectoryException; import org.opends.server.types.Entry; import org.opends.server.types.ExistingFileBehavior; import org.opends.server.types.LDIFExportConfig; import org.opends.server.types.LDIFImportConfig; import org.opends.server.types.Modification; import org.opends.server.types.Operation; import org.opends.server.types.OperationType; import org.opends.server.types.RawModification; import org.opends.server.types.SearchFilter; import org.opends.server.types.SearchResultEntry; import org.opends.server.types.SearchResultReference; import org.opends.server.types.SynchronizationProviderResult; import org.opends.server.types.operation.PluginOperation; import org.opends.server.types.operation.PostOperationAddOperation; import org.opends.server.types.operation.PostOperationDeleteOperation; import org.opends.server.types.operation.PostOperationModifyDNOperation; import org.opends.server.types.operation.PostOperationModifyOperation; import org.opends.server.types.operation.PostOperationOperation; import org.opends.server.types.operation.PreOperationAddOperation; import org.opends.server.types.operation.PreOperationDeleteOperation; import org.opends.server.types.operation.PreOperationModifyDNOperation; import org.opends.server.types.operation.PreOperationModifyOperation; import org.opends.server.util.LDIFReader; import org.opends.server.util.TimeThread; import org.opends.server.workflowelement.localbackend.LocalBackendModifyOperation; /** * This class implements the bulk part of the Directory Server side * of the replication code. * It contains the root method for publishing a change, * processing a change received from the replicationServer service, * handle conflict resolution, * handle protocol messages from the replicationServer. *

* FIXME Move this class to org.opends.server.replication.service * or the equivalent package once this code is moved to a maven module. */ public final class LDAPReplicationDomain extends ReplicationDomain implements ConfigurationChangeListener, AlertGenerator, LocalBackendInitializationListener, ServerShutdownListener { /** * Set of attributes that will return all the user attributes and the * replication related operational attributes when used in a search operation. */ private static final Set USER_AND_REPL_OPERATIONAL_ATTRS = newHashSet(HISTORICAL_ATTRIBUTE_NAME, ENTRYUUID_ATTRIBUTE_NAME, "*"); /** * Initializing replication for the domain initiates backend finalization/initialization * This flag prevents the Replication Domain to disable/enable itself when * it is the event initiator */ private boolean ignoreBackendInitializationEvent; private volatile boolean serverShutdownRequested; @Override public String getShutdownListenerName() { return "LDAPReplicationDomain " + getBaseDN(); } @Override public void processServerShutdown(LocalizableMessage reason) { serverShutdownRequested = true; } /** * This class is used in the session establishment phase * when no Replication Server with all the local changes has been found * and we therefore need to recover them. * A search is then performed on the database using this * internalSearchListener. */ private class ScanSearchListener implements InternalSearchListener { private final CSN startCSN; private final CSN endCSN; public ScanSearchListener(CSN startCSN, CSN endCSN) { this.startCSN = startCSN; this.endCSN = endCSN; } @Override public void handleInternalSearchEntry( InternalSearchOperation searchOperation, SearchResultEntry searchEntry) throws DirectoryException { // Build the list of Operations that happened on this entry after startCSN // and before endCSN and add them to the replayOperations list Iterable updates = EntryHistorical.generateFakeOperations(searchEntry); for (FakeOperation op : updates) { CSN csn = op.getCSN(); if (csn.isNewerThan(startCSN) && csn.isOlderThan(endCSN)) { synchronized (replayOperations) { replayOperations.put(csn, op); } } } } @Override public void handleInternalSearchReference( InternalSearchOperation searchOperation, SearchResultReference searchReference) throws DirectoryException { // Nothing to do. } } @Override public void performBackendPreInitializationProcessing(LocalBackend backend) { // Nothing to do } @Override public void performBackendPostFinalizationProcessing(LocalBackend backend) { // Nothing to do } @Override public void performBackendPostInitializationProcessing(LocalBackend backend) { if (!ignoreBackendInitializationEvent && getBackend().getBackendID().equals(backend.getBackendID())) { enable(); } } @Override public void performBackendPreFinalizationProcessing(LocalBackend backend) { // Do not disable itself during a shutdown // And ignore the event if this replica is the event trigger (e.g. importing). if (!ignoreBackendInitializationEvent && !serverShutdownRequested && getBackend().getBackendID().equals(backend.getBackendID())) { disable(); } } /** The fully-qualified name of this class. */ private static final String CLASS_NAME = LDAPReplicationDomain.class.getName(); /** * The attribute used to mark conflicting entries. * The value of this attribute should be the dn that this entry was * supposed to have when it was marked as conflicting. */ public static final String DS_SYNC_CONFLICT = "ds-sync-conflict"; private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass(); private final DSRSShutdownSync dsrsShutdownSync; /** * The update to replay message queue where the listener thread is going to * push incoming update messages. */ private final BlockingQueue updateToReplayQueue; /** The number of naming conflicts successfully resolved. */ private final AtomicInteger numResolvedNamingConflicts = new AtomicInteger(); /** The number of modify conflicts successfully resolved. */ private final AtomicInteger numResolvedModifyConflicts = new AtomicInteger(); /** The number of unresolved naming conflicts. */ private final AtomicInteger numUnresolvedNamingConflicts = new AtomicInteger(); /** The number of updates replayed successfully by the replication. */ private final AtomicInteger numReplayedPostOpCalled = new AtomicInteger(); /** * How many times the replay of a change is attempted straight away, before the session * to the replication server is restarted and the change is asked for again: a lock or * a storage which is busy for a moment (OPENDJ-885) is waited out here. */ public static final int IN_PLACE_REPLAY_ATTEMPTS = 10; /** * How long the replay of a change is retried before this replica gives up on it and * moves on to the changes which follow it. *

* The budget is a duration rather than a number of attempts because what * {@link #isServerFailure(ResultCode, ResultCode)} reports is measured in minutes: a * backend which is being rebuilt, imported into or restored (OPENDJ-49) serves nothing * while it works, and a handful of attempts would have this replica give up on every * change of a maintenance window it only had to wait out. */ private static final long REPLAY_GIVE_UP_DELAY_IN_MS = 300000; /** * How long the session is left down before the change is asked for again, multiplied * by the number of attempts already made: a backend which keeps failing must not be * hammered with a session restart per failed change. */ private static final long REPLAY_RETRY_DELAY_IN_MS = 1000; /** * The longest the session is left down between two attempts. The wait runs on one of * the replay threads, which are shared by every domain of this server, so it is kept * short enough not to starve the domains which are healthy. */ private static final long MAX_REPLAY_RETRY_DELAY_IN_MS = 10000; /** * How long the alert telling that this replica diverges is not sent again. A single * cause - a schema which does not match, a backend which is gone - makes every change * in flight unreplayable, and one alert per change would be a storm. */ private static final long UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS = 60000; /** The number of updates this replica gave up replaying. */ private final AtomicInteger numFailedReplayedUpdates = new AtomicInteger(); /** Set while a replay thread is restarting the session after a failed replay. */ private final AtomicBoolean replayFailureRecovery = new AtomicBoolean(); /** * Set when a change whose replay failed has to be delivered again, and cleared by the * replay thread which restarts the session for it. A change released while the session * was being restarted has to be asked for over yet another session: the delivery this * one makes is turned down as a duplicate while a replay thread still owns it. */ private final AtomicBoolean sessionRestartRequested = new AtomicBoolean(); /** * How many times in a row the session was restarted without a change being replayed in * between. The backoff is computed from this rather than from the failures of the * change which happens to open the recovery: an outage fails every change in flight, * and the ones which are sent for the first time would otherwise keep the wait at its * shortest for as long as the outage lasts. */ private final AtomicInteger consecutiveSessionRestarts = new AtomicInteger(); /** * How long the replay of a change is retried before this replica gives up on it. Only * the tests, which can not wait out {@link #REPLAY_GIVE_UP_DELAY_IN_MS}, set another * value. */ private volatile long replayGiveUpDelayInMs = REPLAY_GIVE_UP_DELAY_IN_MS; /** * Serialises the session of this domain being stopped and started again: the replay * thread which restarts it after a failed replay must not race the domain being * disabled for an import or a restore, or it would bring a broker and a listener * thread back up on a domain which is supposed to be down. *

* Holding it costs something, and knowingly: {@code enableService()} connects to the * replication servers under this lock, so a shutdown, an import or a configuration * change which arrives while a replay thread is bringing the session back waits for * that connect - up to the configured connection timeout when the replication servers * are unreachable, which is the same outage that failed the replay. Every one of those * stops the session as its first act, so what they wait for is a session which is about * to be stopped again. The wait between the stop and the start is deliberately left * outside the lock, so the waiting is bounded by a connect rather than by the backoff. */ private final Object serviceStateLock = new Object(); /** * Bumped every time the session of this domain is stopped or started under * {@link #serviceStateLock}. A replay thread which stopped the session only starts it * back if this still is the session it stopped: a configuration change, or the end of * an import, may have started another one while it was waiting for the backend to * recover. *

* It does not count the sessions {@code changeConfig()} and {@code readAssuredConfig()} * stop and start, which they do without knowing about it: they run under the lock, so a * replay thread never observes one of theirs, but a session it stopped may well have * been replaced by one of theirs while it was waiting. That is why the guard in * {@link #restartSession(boolean)} reads {@code isListenerShuttingDown()} as well - a * session started outside this counter leaves it untouched, and only the listener says * that one is running. */ @GuardedBy("serviceStateLock") private long sessionGeneration; /** * Stands for "the alert about a change this replica gave up on was never sent". The * time it is compared with only moves forward from an origin which is arbitrary, so * zero is not far enough in the past to say it. */ private static final long UNREPLAYED_CHANGE_ALERT_NEVER_SENT = Long.MIN_VALUE / 2; /** When the alert about a change this replica gave up on was last sent. */ private final AtomicLong lastUnreplayedChangeAlertTime = new AtomicLong(UNREPLAYED_CHANGE_ALERT_NEVER_SENT); /** * The result codes conflict resolution knows how to solve. The result code the server * puts on an internal error is configurable and is not validated as a result code, so * it could be set to one of these: it must never take a change away from * {@code solveNamingConflict()}, which is the only thing which can solve them. */ private static final Set CONFLICT_RESULT_CODES = Collections.unmodifiableSet( newHashSet( ResultCode.NO_SUCH_OBJECT, ResultCode.ENTRY_ALREADY_EXISTS, ResultCode.NOT_ALLOWED_ON_RDN, ResultCode.NOT_ALLOWED_ON_NONLEAF, // solveNamingConflict(ModifyDNOperation) solves these two as well ResultCode.UNWILLING_TO_PERFORM, ResultCode.OBJECTCLASS_VIOLATION)); private final PersistentServerState state; private volatile boolean generationIdSavedStatus; /** * This object is used to store the list of update currently being done on the local database. * It is useful to make sure that the local operations are sent in a correct order to the * replication server and that the ServerState is not updated too early. */ private final PendingChanges pendingChanges; private final AtomicReference rsUpdater = new AtomicReference<>(null); /** * It contain the updates that were done on other servers, transmitted by the * replication server and that are currently replayed. *

* It is useful to make sure that dependencies between operations are * correctly fulfilled and to make sure that the ServerState is not updated * too early. */ private final RemotePendingChanges remotePendingChanges; private boolean solveConflictFlag = true; private final InternalClientConnection conn = getRootConnection(); private final AtomicBoolean shutdown = new AtomicBoolean(); private volatile boolean disabled; /** * This list is used to temporary store operations that needs to be replayed * at session establishment time. */ private final SortedMap replayOperations = new TreeMap<>(); private ExternalChangelogDomain eclDomain; /** A boolean indicating if the thread used to save the persistentServerState is terminated. */ private volatile boolean done = true; private final ServerStateFlush flushThread; /** The attribute name used to store the generation id in the backend. */ private static final String REPLICATION_GENERATION_ID = "ds-sync-generation-id"; /** The attribute name used to store the fractional include configuration in the backend. */ static final String REPLICATION_FRACTIONAL_INCLUDE = "ds-sync-fractional-include"; /** The attribute name used to store the fractional exclude configuration in the backend. */ static final String REPLICATION_FRACTIONAL_EXCLUDE = "ds-sync-fractional-exclude"; /** * Fractional replication variables. */ /** Holds the fractional configuration for this domain, if any. */ private final FractionalConfig fractionalConfig; /** The list of attributes that cannot be used in fractional replication configuration. */ private static final String[] FRACTIONAL_PROHIBITED_ATTRIBUTES = new String[] { "objectClass", "2.5.4.0" // objectClass OID }; private static final DN SET_PERMISSIVE_MODIFY_FOR_DN = DN.valueOf(DN_DEFAULT_SCHEMA_ROOT); /** * When true, this flag is used to force the domain status to be put in bad * data set just after the connection to the replication server. * This must be used when fractional replication is enabled with a * configuration different from the previous one (or at the very first * fractional usage time) : after connection, a ChangeStatusMsg is sent * requesting the bad data set status. Then none of the update messages * received from the replication server are taken into account until the * backend is synchronized with brand new data set compliant with the new * fractional configuration (i.e with compliant fractional configuration in * domain root entry). */ private boolean forceBadDataSet; /** * The message id to be used when an import is stopped with error by * the fractional replication ldif import plugin. */ private int importErrorMessageId = -1; /** LocalizableMessage type for ERR_FULL_UPDATE_IMPORT_FRACTIONAL_BAD_REMOTE. */ static final int IMPORT_ERROR_MESSAGE_BAD_REMOTE = 1; /** LocalizableMessage type for ERR_FULL_UPDATE_IMPORT_FRACTIONAL_REMOTE_IS_FRACTIONAL. */ static final int IMPORT_ERROR_MESSAGE_REMOTE_IS_FRACTIONAL = 2; /* * Definitions for the return codes of the * fractionalFilterOperation(PreOperationModifyOperation * modifyOperation, boolean performFiltering) method */ /** * The operation contains attributes subject to fractional filtering according * to the fractional configuration. */ private static final int FRACTIONAL_HAS_FRACTIONAL_FILTERED_ATTRIBUTES = 1; /** * The operation contains no attributes subject to fractional filtering * according to the fractional configuration. */ private static final int FRACTIONAL_HAS_NO_FRACTIONAL_FILTERED_ATTRIBUTES = 2; /** The operation should become a no-op. */ private static final int FRACTIONAL_BECOME_NO_OP = 3; /** * The last CSN purged in this domain. Allows to have a continuous purging * process from one purge processing (task run) to the next one. Values 0 when * the server starts. */ private CSN lastCSNPurgedFromHist = new CSN(0,0,0); /** * The thread that periodically saves the ServerState of this * LDAPReplicationDomain in the database. */ private class ServerStateFlush extends DirectoryThread { protected ServerStateFlush() { super("Replica DS(" + getServerId() + ") state checkpointer for domain \"" + getBaseDN() + "\""); } @Override public void run() { done = false; while (!isShutdownInitiated()) { try { synchronized (this) { wait(1000); if (!disabled && !ieRunning()) { state.save(); } } } catch (InterruptedException e) { // Thread interrupted: check for shutdown. Thread.currentThread().interrupt(); } } state.save(); done = true; } } /** * The thread that is responsible to update the RS to which this domain is * connected in case it is late and there is no RS which is up to date. */ private class RSUpdater extends DirectoryThread { private final CSN startCSN; protected RSUpdater(CSN replServerMaxCSN) { super("Replica DS(" + getServerId() + ") missing change publisher for domain \"" + getBaseDN() + "\""); this.startCSN = replServerMaxCSN; } @Override public void run() { // Replication server is missing some of our changes: // let's send them to him. logger.trace(DEBUG_GOING_TO_SEARCH_FOR_CHANGES); /* * Get all the changes that have not been seen by this * replication server and publish them. */ try { if (buildAndPublishMissingChanges(startCSN, broker)) { logger.trace(DEBUG_CHANGES_SENT); synchronized(replayOperations) { replayOperations.clear(); } } else { /* * An error happened trying to search for the updates * This server will start accepting again new updates but * some inconsistencies will stay between servers. * Log an error for the repair tool * that will need to re-synchronize the servers. */ logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN()); } } catch (Exception e) { /* * An error happened trying to search for the updates * This server will start accepting again new updates but * some inconsistencies will stay between servers. * Log an error for the repair tool * that will need to re-synchronize the servers. */ logger.error(ERR_CANNOT_RECOVER_CHANGES, getBaseDN()); } finally { broker.setRecoveryRequired(false); // RSUpdater thread has finished its work, let's remove it from memory // so another RSUpdater thread can be started if needed. rsUpdater.compareAndSet(this, null); } } } /** * Creates a new ReplicationDomain using configuration from configEntry. * * @param configuration The configuration of this ReplicationDomain. * @param updateToReplayQueue The queue for update messages to replay. * @param dsrsShutdownSync Synchronization object for shutdown of combined DS/RS instances. * @throws ConfigException In case of invalid configuration. */ LDAPReplicationDomain(ReplicationDomainCfg configuration, BlockingQueue updateToReplayQueue, DSRSShutdownSync dsrsShutdownSync) throws ConfigException { super(configuration, -1); this.updateToReplayQueue = updateToReplayQueue; this.dsrsShutdownSync = dsrsShutdownSync; // Get assured configuration readAssuredConfig(configuration, false); // Get fractional configuration fractionalConfig = new FractionalConfig(getBaseDN()); readFractionalConfig(configuration, false); storeECLConfiguration(configuration); solveConflictFlag = isSolveConflict(configuration); LocalBackend backend = getBackend(); if (backend == null) { throw new ConfigException(ERR_SEARCHING_DOMAIN_BACKEND.get(getBaseDN())); } try { generationId = loadGenerationId(); } catch (DirectoryException e) { logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), stackTraceToSingleLineString(e)); } /* * Create a new Persistent Server State that will be used to store * the last CSN seen from all LDAP servers in the topology. */ state = new PersistentServerState(getBaseDN(), getServerId(), getServerState()); flushThread = new ServerStateFlush(); /* * CSNGenerator is used to create new unique CSNs for each operation done on * this replication domain. * * The generator time is adjusted to the time of the last CSN received from * remote other servers. */ pendingChanges = new PendingChanges(getGenerator(), this); remotePendingChanges = new RemotePendingChanges(getServerState()); // listen for changes on the configuration configuration.addChangeListener(this); // register as an AlertGenerator DirectoryServer.registerAlertGenerator(this); getServerContext().getBackendConfigManager() .registerLocalBackendInitializationListener(this); DirectoryServer.registerShutdownListener(this); startPublishService(); } /** * Modify conflicts are solved for all suffixes but the schema suffix because * we don't want to store extra information in the schema ldif files. This has * no negative impact because the changes on schema should not produce * conflicts. */ private boolean isSolveConflict(ReplicationDomainCfg cfg) { return !getBaseDN().equals(DirectoryServer.getSchemaDN()) && cfg.isSolveConflicts(); } /** * Sets the error message id to be used when online import is stopped with * error by the fractional replication ldif import plugin. * @param importErrorMessageId The message to use. */ void setImportErrorMessageId(int importErrorMessageId) { this.importErrorMessageId = importErrorMessageId; } /** * This flag is used by the fractional replication ldif import plugin to stop * the (online) import process if a fractional configuration inconsistency is * detected by it. * * @return true if the online import currently in progress should continue, * false otherwise. */ private boolean isFollowImport() { return importErrorMessageId == -1; } /** * Gets and stores the fractional replication configuration parameters. * @param configuration The configuration object * @param allowReconnection Tells if one must reconnect if significant changes * occurred */ private void readFractionalConfig(ReplicationDomainCfg configuration, boolean allowReconnection) { // Read the configuration entry FractionalConfig newFractionalConfig; try { newFractionalConfig = FractionalConfig.toFractionalConfig(configuration); } catch(ConfigException e) { // Should not happen as normally already called without problem in // isConfigurationChangeAcceptable or isConfigurationAcceptable // if we come up to this method logger.info(NOTE_ERR_FRACTIONAL, getBaseDN(), stackTraceToSingleLineString(e)); return; } /** * Is there any change in fractional configuration ? */ // Compute current configuration boolean needReconnection; try { needReconnection = !FractionalConfig. isFractionalConfigEquivalent(fractionalConfig, newFractionalConfig); } catch (ConfigException e) { // Should not happen logger.info(NOTE_ERR_FRACTIONAL, getBaseDN(), stackTraceToSingleLineString(e)); return; } // Disable service if configuration changed final boolean needRestart = needReconnection && allowReconnection; /* * The session is stopped, the configuration it depends on is changed and the session * is started again under the lock which the replay thread restarting the session after * a failed replay holds too: a session brought up in the middle of this would be * reading a fractional configuration which is half way through being changed. The * pair has to be atomic, which the lock inside disableService()/enableService() does * not make it. */ synchronized (serviceStateLock) { if (needRestart) { disableService(); sessionGeneration++; } // Set new configuration int newFractionalMode = newFractionalConfig.fractionalConfigToInt(); fractionalConfig.setFractional(newFractionalMode != FractionalConfig.NOT_FRACTIONAL); if (fractionalConfig.isFractional()) { // Set new fractional configuration values fractionalConfig.setFractionalExclusive( newFractionalMode == FractionalConfig.EXCLUSIVE_FRACTIONAL); fractionalConfig.setFractionalSpecificClassesAttributes( newFractionalConfig.getFractionalSpecificClassesAttributes()); fractionalConfig.setFractionalAllClassesAttributes( newFractionalConfig.fractionalAllClassesAttributes); } else { // Reset default values fractionalConfig.setFractionalExclusive(true); fractionalConfig.setFractionalSpecificClassesAttributes( new HashMap>()); fractionalConfig.setFractionalAllClassesAttributes(new HashSet()); } // Reconnect if required if (needRestart) { enableService(); sessionGeneration++; } } } /** * Return true if the fractional configuration stored in the domain root * entry of the backend is equivalent to the fractional configuration stored * in the local variables. */ private boolean isBackendFractionalConfigConsistent() { // Read config stored in domain root entry if (logger.isTraceEnabled()) { logger.trace("Attempt to read the potential fractional config in domain root entry " + getBaseDN()); } // Search the domain root entry that is used to save the generation id SearchRequest request = newSearchRequest(getBaseDN(), SearchScope.BASE_OBJECT) .addAttribute(REPLICATION_GENERATION_ID, REPLICATION_FRACTIONAL_EXCLUDE, REPLICATION_FRACTIONAL_INCLUDE); InternalSearchOperation search = conn.processSearch(request); if (search.getResultCode() != ResultCode.SUCCESS && search.getResultCode() != ResultCode.NO_SUCH_OBJECT) { String errorMsg = search.getResultCode().getName() + " " + search.getErrorMessage(); logger.error(ERR_SEARCHING_GENERATION_ID, getBaseDN(), errorMsg); return false; } SearchResultEntry resultEntry = findReplicationSearchResultEntry(search); if (resultEntry == null) { /* * The backend is probably empty: if there is some fractional * configuration in memory, we do not let the domain being connected, * otherwise, it's ok */ return !fractionalConfig.isFractional(); } // Now extract fractional configuration if any Iterator exclIt = getAttributeValueIterator(resultEntry, REPLICATION_FRACTIONAL_EXCLUDE); Iterator inclIt = getAttributeValueIterator(resultEntry, REPLICATION_FRACTIONAL_INCLUDE); // Compare backend and local fractional configuration return isFractionalConfigConsistent(fractionalConfig, exclIt, inclIt); } private SearchResultEntry findReplicationSearchResultEntry( InternalSearchOperation searchOperation) { final SearchResultEntry resultEntry = getFirstResult(searchOperation); if (resultEntry != null) { Iterator attrs = resultEntry.getAllAttributes(REPLICATION_GENERATION_ID).iterator(); if (attrs.hasNext()) { Attribute attr = attrs.next(); if (attr.size() == 1) { return resultEntry; } if (attr.size() > 1) { String errorMsg = "#Values=" + attr.size() + " Must be exactly 1 in entry " + resultEntry.toLDIFString(); logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), errorMsg); } } } return null; } private static ServerContext getServerContext() { return DirectoryServer.getInstance().getServerContext(); } private Iterator getAttributeValueIterator(SearchResultEntry resultEntry, String attrName) { AttributeType attrType = getServerContext().getSchema().getAttributeType(attrName); List exclAttrs = resultEntry.getAllAttributes(attrType); if (!exclAttrs.isEmpty()) { Attribute exclAttr = exclAttrs.get(0); if (exclAttr != null) { return exclAttr.iterator(); } } return null; } /** * Return true if the fractional configuration passed as fractional * configuration attribute values is equivalent to the fractional * configuration stored in the local variables. * @param fractionalConfig The local fractional configuration * @param exclIt Fractional exclude mode configuration attribute values to * analyze. * @param inclIt Fractional include mode configuration attribute values to * analyze. * @return True if the fractional configuration passed as fractional * configuration attribute values is equivalent to the fractional * configuration stored in the local variables. */ static boolean isFractionalConfigConsistent( FractionalConfig fractionalConfig, Iterator exclIt, Iterator inclIt) { // Parse fractional configuration stored in passed fractional configuration attributes values Map> storedFractionalSpecificClassesAttributes = new HashMap<>(); Set storedFractionalAllClassesAttributes = new HashSet<>(); int storedFractionalMode; try { storedFractionalMode = FractionalConfig.parseFractionalConfig(exclIt, inclIt, storedFractionalSpecificClassesAttributes, storedFractionalAllClassesAttributes); } catch (ConfigException e) { // Should not happen as configuration in domain root entry is flushed // from valid configuration in local variables logger.info(NOTE_ERR_FRACTIONAL, fractionalConfig.getBaseDn(), stackTraceToSingleLineString(e)); return false; } FractionalConfig storedFractionalConfig = new FractionalConfig( fractionalConfig.getBaseDn()); storedFractionalConfig.setFractional(storedFractionalMode != FractionalConfig.NOT_FRACTIONAL); // Set stored fractional configuration values if (storedFractionalConfig.isFractional()) { storedFractionalConfig.setFractionalExclusive( storedFractionalMode == FractionalConfig.EXCLUSIVE_FRACTIONAL); } storedFractionalConfig.setFractionalSpecificClassesAttributes( storedFractionalSpecificClassesAttributes); storedFractionalConfig.setFractionalAllClassesAttributes( storedFractionalAllClassesAttributes); /* * Compare configuration stored in passed fractional configuration * attributes with local variable one */ try { return FractionalConfig. isFractionalConfigEquivalent(fractionalConfig, storedFractionalConfig); } catch (ConfigException e) { // Should not happen as configuration in domain root entry is flushed // from valid configuration in local variables so both should have already // been checked logger.info(NOTE_ERR_FRACTIONAL, fractionalConfig.getBaseDn(), stackTraceToSingleLineString(e)); return false; } } /** * Compare 2 attribute collections and returns true if they are equivalent. * * @param attributes1 * First attribute collection to compare. * @param attributes2 * Second attribute collection to compare. * @return True if both attribute collection are equivalent. * @throws ConfigException * If some attributes could not be retrieved from the schema. */ private static boolean areAttributesEquivalent( Collection attributes1, Collection attributes2) throws ConfigException { // Compare all classes attributes if (attributes1.size() != attributes2.size()) { return false; } // Check consistency of all classes attributes Schema schema = getServerContext().getSchema(); /* * For each attribute in attributes1, check there is the matching * one in attributes2. */ for (String attrName1 : attributes1) { // Get attribute from attributes1 AttributeType attributeType1 = schema.getAttributeType(attrName1); if (attributeType1.isPlaceHolder()) { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_ATTRIBUTE_TYPE.get(attrName1)); } // Look for matching one in attributes2 boolean foundAttribute = false; for (String attrName2 : attributes2) { AttributeType attributeType2 = schema.getAttributeType(attrName2); if (attributeType2.isPlaceHolder()) { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_ATTRIBUTE_TYPE.get(attrName2)); } if (attributeType1.equals(attributeType2)) { foundAttribute = true; break; } } // Found matching attribute ? if (!foundAttribute) { return false; } } return true; } /** * Check that the passed fractional configuration is acceptable * regarding configuration syntax, schema constraints... * Throws an exception if the configuration is not acceptable. * @param configuration The configuration to analyze. * @throws org.opends.server.config.ConfigException if the configuration is * not acceptable. */ private static void isFractionalConfigAcceptable( ReplicationDomainCfg configuration) throws ConfigException { /* * Parse fractional configuration */ // Read the configuration entry FractionalConfig newFractionalConfig = FractionalConfig.toFractionalConfig( configuration); if (!newFractionalConfig.isFractional()) { // Nothing to check return; } // Prepare variables to be filled with config Map> newFractionalSpecificClassesAttributes = newFractionalConfig.getFractionalSpecificClassesAttributes(); Set newFractionalAllClassesAttributes = newFractionalConfig.getFractionalAllClassesAttributes(); /* * Check attributes consistency : we only allow to filter MAY (optional) * attributes of a class : to be compliant with the schema, no MUST * (mandatory) attribute can be filtered by fractional replication. */ // Check consistency of specific classes attributes Schema schema = getServerContext().getSchema(); int fractionalMode = newFractionalConfig.fractionalConfigToInt(); for (Map.Entry> entry : newFractionalSpecificClassesAttributes.entrySet()) { String className = entry.getKey(); // Does the class exist ? ObjectClass fractionalClass = schema.getObjectClass(className); if (fractionalClass.isPlaceHolder()) { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_OBJECT_CLASS.get(className)); } boolean isExtensibleObjectClass = fractionalClass.isExtensible(); Set attributes = entry.getValue(); for (String attrName : attributes) { // Not a prohibited attribute ? if (isFractionalProhibitedAttr(attrName)) { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_PROHIBITED_ATTRIBUTE.get(attrName)); } // Does the attribute exist ? AttributeType attributeType = schema.getAttributeType(attrName); if (!attributeType.isPlaceHolder()) { // No more checking for the extensibleObject class if (!isExtensibleObjectClass && fractionalMode == FractionalConfig.EXCLUSIVE_FRACTIONAL // Exclusive mode : the attribute must be optional && !fractionalClass.isOptional(attributeType)) { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_NOT_OPTIONAL_ATTRIBUTE.get(attrName, className)); } } else { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_ATTRIBUTE_TYPE.get(attrName)); } } } // Check consistency of all classes attributes for (String attrName : newFractionalAllClassesAttributes) { // Not a prohibited attribute ? if (isFractionalProhibitedAttr(attrName)) { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_PROHIBITED_ATTRIBUTE.get(attrName)); } // Does the attribute exist ? if (schema.getAttributeType(attrName) == null) { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_ATTRIBUTE_TYPE.get(attrName)); } } } /** * Test if the passed attribute is not allowed to be used in configuration of * fractional replication. * @param attr Attribute to test. * @return true if the attribute is prohibited. */ private static boolean isFractionalProhibitedAttr(String attr) { for (String forbiddenAttr : FRACTIONAL_PROHIBITED_ATTRIBUTES) { if (forbiddenAttr.equalsIgnoreCase(attr)) { return true; } } return false; } /** * If fractional replication is enabled, this analyzes the operation and * suppresses the forbidden attributes in it so that they are not added in * the local backend. * * @param addOperation The operation to modify based on fractional * replication configuration * @param performFiltering Tells if the effective attribute filtering should * be performed or if the call is just to analyze if there are some * attributes filtered by fractional configuration * @return true if the operation contains some attributes subject to filtering * by the fractional configuration */ private boolean fractionalFilterOperation( PreOperationAddOperation addOperation, boolean performFiltering) { return fractionalRemoveAttributesFromEntry(fractionalConfig, addOperation.getEntryDN().rdn(), addOperation.getObjectClasses(), addOperation.getUserAttributes(), performFiltering); } /** * If fractional replication is enabled, this analyzes the operation and * suppresses the forbidden attributes in it so that they are not added in * the local backend. * * @param modifyDNOperation The operation to modify based on fractional * replication configuration * @param performFiltering Tells if the effective modifications should * be performed or if the call is just to analyze if there are some * inconsistency with fractional configuration * @return true if the operation is inconsistent with fractional * configuration */ private boolean fractionalFilterOperation( PreOperationModifyDNOperation modifyDNOperation, boolean performFiltering) { // Quick exit if not called for analyze and if (performFiltering && modifyDNOperation.deleteOldRDN()) { // The core will remove any occurrence of attribute that was part of the // old RDN, nothing more to do. return true; // Will not be used as analyze was not requested } // Create a list of filtered attributes for this entry Entry concernedEntry = modifyDNOperation.getOriginalEntry(); Set fractionalConcernedAttributes = createFractionalConcernedAttrList(fractionalConfig, concernedEntry.getObjectClasses().keySet()); boolean fractionalExclusive = fractionalConfig.isFractionalExclusive(); if (fractionalExclusive && fractionalConcernedAttributes.isEmpty()) { // No attributes to filter return false; } /* * Analyze the old and new rdn to see if they are some attributes to be * removed: if the oldRDN contains some forbidden attributes (for instance * it is possible if the entry was created with an add operation and the * RDN used contains a forbidden attribute: in this case the attribute value * has been kept to be consistent with the dn of the entry.) that are no * more part of the new RDN, we must remove any attribute of this type by * putting a modification to delete the attribute. */ boolean inconsistentOperation = false; RDN rdn = modifyDNOperation.getEntryDN().rdn(); RDN newRdn = modifyDNOperation.getNewRDN(); // Go through each attribute of the old RDN for (AVA ava : rdn) { AttributeType attributeType = ava.getAttributeType(); // Is it present in the fractional attributes established list ? boolean foundAttribute = fractionalConcernedAttributes.contains(attributeType); if (canRemoveAttribute(fractionalExclusive, foundAttribute) && !newRdn.hasAttributeType(attributeType) && !modifyDNOperation.deleteOldRDN()) { /* * A forbidden attribute is in the old RDN and no more in the new RDN, * and it has not been requested to remove attributes from old RDN: * let's remove the attribute from the entry to stay consistent with * fractional configuration */ Modification modification = new Modification(ModificationType.DELETE, Attributes.empty(attributeType)); modifyDNOperation.addModification(modification); inconsistentOperation = true; } } return inconsistentOperation; } /** * Remove attributes from an entry, according to the passed fractional * configuration. The entry is represented by the 2 passed parameters. * The attributes to be removed are removed using the remove method on the * passed iterator for the attributes in the entry. * @param fractionalConfig The fractional configuration to use * @param entryRdn The rdn of the entry to add * @param classes The object classes representing the entry to modify * @param attributesMap The map of attributes/values to be potentially removed * from the entry. * @param performFiltering Tells if the effective attribute filtering should * be performed or if the call is just an analyze to see if there are some * attributes filtered by fractional configuration * @return true if the operation contains some attributes subject to filtering * by the fractional configuration */ static boolean fractionalRemoveAttributesFromEntry( FractionalConfig fractionalConfig, RDN entryRdn, Map classes, Map> attributesMap, boolean performFiltering) { boolean hasSomeAttributesToFilter = false; /* * Prepare a list of attributes to be included/excluded according to the * fractional replication configuration */ Set fractionalConcernedAttributes = createFractionalConcernedAttrList(fractionalConfig, classes.keySet()); boolean fractionalExclusive = fractionalConfig.isFractionalExclusive(); if (fractionalExclusive && fractionalConcernedAttributes.isEmpty()) { return false; // No attributes to filter } // Prepare list of object classes of the added entry Set entryClasses = classes.keySet(); /* * Go through the user attributes and remove those that match filtered one * - exclude mode : remove only attributes that are in * fractionalConcernedAttributes * - include mode : remove any attribute that is not in * fractionalConcernedAttributes */ List> newRdnAttrLists = new ArrayList<>(); List rdnAttrTypes = new ArrayList<>(); final Set attrTypes = attributesMap.keySet(); for (Iterator iter = attrTypes.iterator(); iter.hasNext();) { AttributeType attributeType = iter.next(); // Only optional attributes may be removed if (isMandatoryAttribute(entryClasses, attributeType) // Do not remove an attribute if it is a prohibited one || isFractionalProhibited(attributeType) || !canRemoveAttribute(attributeType, fractionalExclusive, fractionalConcernedAttributes)) { continue; } if (!performFiltering) { // The call was just to check : at least one attribute to filter // found, return immediately the answer; return true; } // Do not remove an attribute/value that is part of the RDN of the // entry as it is forbidden if (entryRdn.hasAttributeType(attributeType)) { /* We must remove all values of the attributes map for this attribute type but the one that has the value which is in the RDN of the entry. In fact the (underlying )attribute list does not support remove so we have to create a new list, keeping only the attribute value which is the same as in the RDN */ ByteString rdnAttributeValue = entryRdn.getAttributeValue(attributeType); List attrList = attributesMap.get(attributeType); ByteString sameAttrValue = null; // Locate the attribute value identical to the one in the RDN for (Attribute attr : attrList) { if (attr.contains(rdnAttributeValue)) { for (ByteString attrValue : attr) { if (rdnAttributeValue.equals(attrValue)) { // Keep the value we want sameAttrValue = attrValue; } else { hasSomeAttributesToFilter = true; } } } else { hasSomeAttributesToFilter = true; } } // Recreate the attribute list with only the RDN attribute value if (sameAttrValue != null) // Paranoia check: should never be the case as we should always // find the attribute/value pair matching the pair in the RDN { // Construct and store new attribute list newRdnAttrLists.add(Attributes.createAsList(attributeType, sameAttrValue)); /* Store matching attribute type The mapping will be done using object from rdnAttrTypes as key and object from newRdnAttrLists (at same index) as value in the user attribute map to be modified */ rdnAttrTypes.add(attributeType); } } else { // Found an attribute to remove, remove it from the list. iter.remove(); hasSomeAttributesToFilter = true; } } // Now overwrite the attribute values for the attribute types present in the // RDN, if there are some filtered attributes in the RDN for (int index = 0 ; index < rdnAttrTypes.size() ; index++) { attributesMap.put(rdnAttrTypes.get(index), newRdnAttrLists.get(index)); } return hasSomeAttributesToFilter; } private static boolean isMandatoryAttribute(Set entryClasses, AttributeType attributeType) { for (ObjectClass objectClass : entryClasses) { if (objectClass.isRequired(attributeType)) { return true; } } return false; } private static boolean isFractionalProhibited(AttributeType attrType) { String attributeName = attrType.getNameOrOID(); return (attributeName != null && isFractionalProhibitedAttr(attributeName)) || isFractionalProhibitedAttr(attrType.getOID()); } private static boolean canRemoveAttribute(AttributeType attributeType, boolean fractionalExclusive, Set fractionalConcernedAttributes) { // Now remove the attribute or modification if: // - exclusive mode and attribute is in configuration // - inclusive mode and attribute is not in configuration return canRemoveAttribute(fractionalExclusive, fractionalConcernedAttributes.contains(attributeType)); } private static boolean canRemoveAttribute(boolean fractionalExclusive, boolean foundAttribute) { return (foundAttribute && fractionalExclusive) || (!foundAttribute && !fractionalExclusive); } /** * Prepares a list of attributes of interest for the fractional feature. * @param fractionalConfig The fractional configuration to use * @param entryObjectClasses The object classes of an entry on which an * operation is going to be performed. * @return The list of attributes of the entry to be excluded/included * when the operation will be performed. */ private static Set createFractionalConcernedAttrList( FractionalConfig fractionalConfig, Set entryObjectClasses) { /* * Is the concerned entry of a type concerned by fractional replication configuration ? * If yes, add the matching attribute names to a set of attributes * to take into account for filtering (inclusive or exclusive mode). * Using a Set to avoid duplicate attributes (from 2 inheriting classes for instance) */ Set fractionalConcernedAttributes = new HashSet<>(); // Get object classes the entry matches Set fractionalAllClassesAttributes = fractionalConfig.getFractionalAllClassesAttributes(); Map> fractionalSpecificClassesAttributes = fractionalConfig.getFractionalSpecificClassesAttributes(); Set fractionalClasses = fractionalSpecificClassesAttributes.keySet(); for (ObjectClass entryObjectClass : entryObjectClasses) { for(String fractionalClass : fractionalClasses) { if (entryObjectClass.hasNameOrOID(fractionalClass.toLowerCase())) { fractionalConcernedAttributes.addAll( fractionalSpecificClassesAttributes.get(fractionalClass)); } } } // Add to the set any attribute which is class independent fractionalConcernedAttributes.addAll(fractionalAllClassesAttributes); Set results = new HashSet<>(); for (String attrName : fractionalConcernedAttributes) { results.add(getServerContext().getSchema().getAttributeType(attrName)); } return results; } /** * If fractional replication is enabled, this analyzes the operation and * suppresses the forbidden attributes in it so that they are not added/ * deleted/modified in the local backend. * * @param modifyOperation The operation to modify based on fractional * replication configuration * @param performFiltering Tells if the effective attribute filtering should * be performed or if the call is just to analyze if there are some * attributes filtered by fractional configuration * @return FRACTIONAL_HAS_FRACTIONAL_FILTERED_ATTRIBUTES, * FRACTIONAL_HAS_NO_FRACTIONAL_FILTERED_ATTRIBUTES or FRACTIONAL_BECOME_NO_OP */ private int fractionalFilterOperation(PreOperationModifyOperation modifyOperation, boolean performFiltering) { /* * Prepare a list of attributes to be included/excluded according to the * fractional replication configuration */ Entry modifiedEntry = modifyOperation.getCurrentEntry(); Set fractionalConcernedAttributes = createFractionalConcernedAttrList(fractionalConfig, modifiedEntry.getObjectClasses().keySet()); boolean fractionalExclusive = fractionalConfig.isFractionalExclusive(); if (fractionalExclusive && fractionalConcernedAttributes.isEmpty()) { // No attributes to filter return FRACTIONAL_HAS_NO_FRACTIONAL_FILTERED_ATTRIBUTES; } // Prepare list of object classes of the modified entry DN entryToModifyDn = modifyOperation.getEntryDN(); Entry entryToModify; try { entryToModify = DirectoryServer.getEntry(entryToModifyDn); } catch(DirectoryException e) { logger.info(NOTE_ERR_FRACTIONAL, getBaseDN(), stackTraceToSingleLineString(e)); return FRACTIONAL_HAS_NO_FRACTIONAL_FILTERED_ATTRIBUTES; } Set entryClasses = entryToModify.getObjectClasses().keySet(); /* * Now go through the attribute modifications and filter the mods according * to the fractional configuration (using the just established concerned * attributes list): * - delete attributes: remove them if regarding a filtered attribute * - add attributes: remove them if regarding a filtered attribute * - modify attributes: remove them if regarding a filtered attribute */ int result = FRACTIONAL_HAS_NO_FRACTIONAL_FILTERED_ATTRIBUTES; List mods = modifyOperation.getModifications(); Iterator modsIt = mods.iterator(); while (modsIt.hasNext()) { Modification mod = modsIt.next(); Attribute attr = mod.getAttribute(); AttributeType attrType = attr.getAttributeDescription().getAttributeType(); // Fractional replication ignores operational attributes if (attrType.isOperational() || isMandatoryAttribute(entryClasses, attrType) || isFractionalProhibited(attrType) || !canRemoveAttribute(attrType, fractionalExclusive, fractionalConcernedAttributes)) { continue; } if (!performFiltering) { // The call was just to check : at least one attribute to filter // found, return immediately the answer; return FRACTIONAL_HAS_FRACTIONAL_FILTERED_ATTRIBUTES; } // Found a modification to remove, remove it from the list. modsIt.remove(); result = FRACTIONAL_HAS_FRACTIONAL_FILTERED_ATTRIBUTES; if (mods.isEmpty()) { // This operation must become a no-op as no more modification in it return FRACTIONAL_BECOME_NO_OP; } } return result; } /** * This is overwritten to allow stopping the (online) import process by the * fractional ldif import plugin when it detects that the (imported) remote * data set is not consistent with the local fractional configuration. * {@inheritDoc} */ @Override protected byte[] receiveEntryBytes() { if (isFollowImport()) { // Ok, next entry is allowed to be received return super.receiveEntryBytes(); } // Fractional ldif import plugin detected inconsistency between local and // remote server fractional configuration and is stopping the import // process: // This is an error termination during the import // The error is stored and the import is ended by returning null final ImportExportContext ieCtx = getImportExportContext(); LocalizableMessage msg = null; switch (importErrorMessageId) { case IMPORT_ERROR_MESSAGE_BAD_REMOTE: msg = NOTE_ERR_FULL_UPDATE_IMPORT_FRACTIONAL_BAD_REMOTE.get(getBaseDN(), ieCtx.getImportSource()); break; case IMPORT_ERROR_MESSAGE_REMOTE_IS_FRACTIONAL: msg = NOTE_ERR_FULL_UPDATE_IMPORT_FRACTIONAL_REMOTE_IS_FRACTIONAL.get(getBaseDN(), ieCtx.getImportSource()); break; } ieCtx.setException(new DirectoryException(UNWILLING_TO_PERFORM, msg)); return null; } /** * This is overwritten to allow stopping the (online) export process if the * local domain is fractional and the destination is all other servers: * This make no sense to have only fractional servers in a replicated * topology. This prevents from administrator manipulation error that would * lead to whole topology data corruption. * {@inheritDoc} */ @Override protected void initializeRemote(int target, int requestorID, Task initTask, int initWindow) throws DirectoryException { if (target == RoutableMsg.ALL_SERVERS && fractionalConfig.isFractional()) { LocalizableMessage msg = NOTE_ERR_FRACTIONAL_FORBIDDEN_FULL_UPDATE_FRACTIONAL.get(getBaseDN(), getServerId()); throw new DirectoryException(ResultCode.UNWILLING_TO_PERFORM, msg); } super.initializeRemote(target, requestorID, initTask, initWindow); } /** * Implement the handleConflictResolution phase of the deleteOperation. * * @param deleteOperation The deleteOperation. * @return A SynchronizationProviderResult indicating if the operation * can continue. */ SynchronizationProviderResult handleConflictResolution( PreOperationDeleteOperation deleteOperation) { if (!deleteOperation.isSynchronizationOperation() && !brokerIsConnected()) { LocalizableMessage msg = ERR_REPLICATION_COULD_NOT_CONNECT.get(getBaseDN()); return new SynchronizationProviderResult.StopProcessing( ResultCode.UNWILLING_TO_PERFORM, msg); } DeleteContext ctx = (DeleteContext) deleteOperation.getAttachment(SYNCHROCONTEXT); Entry deletedEntry = deleteOperation.getEntryToDelete(); if (ctx != null) { /* * This is a replication operation * Check that the modified entry has the same entryuuid * as it was in the original message. */ String operationEntryUUID = ctx.getEntryUUID(); String deletedEntryUUID = getEntryUUID(deletedEntry); if (!operationEntryUUID.equals(deletedEntryUUID)) { /* * The changes entry is not the same entry as the one on * the original change was performed. * Probably the original entry was renamed and replaced with * another entry. * We must not let the change proceed, return a negative * result and set the result code to NO_SUCH_OBJECT. * When the operation will return, the thread that started the operation * will try to find the correct entry and restart a new operation. */ return new SynchronizationProviderResult.StopProcessing( ResultCode.NO_SUCH_OBJECT, null); } } else { // There is no replication context attached to the operation // so this is not a replication operation. CSN csn = generateCSN(deleteOperation); String modifiedEntryUUID = getEntryUUID(deletedEntry); ctx = new DeleteContext(csn, modifiedEntryUUID); deleteOperation.setAttachment(SYNCHROCONTEXT, ctx); synchronized (replayOperations) { int size = replayOperations.size(); if (size >= 10000) { replayOperations.remove(replayOperations.firstKey()); } FakeOperation op = new FakeDelOperation( deleteOperation.getEntryDN(), csn, modifiedEntryUUID); replayOperations.put(csn, op); } } return new SynchronizationProviderResult.ContinueProcessing(); } /** * Implement the handleConflictResolution phase of the addOperation. * * @param addOperation The AddOperation. * @return A SynchronizationProviderResult indicating if the operation * can continue. */ SynchronizationProviderResult handleConflictResolution( PreOperationAddOperation addOperation) { if (!addOperation.isSynchronizationOperation() && !brokerIsConnected()) { LocalizableMessage msg = ERR_REPLICATION_COULD_NOT_CONNECT.get(getBaseDN()); return new SynchronizationProviderResult.StopProcessing( ResultCode.UNWILLING_TO_PERFORM, msg); } if (fractionalConfig.isFractional()) { if (addOperation.isSynchronizationOperation()) { /* * Filter attributes here for fractional replication. If fractional * replication is enabled, we analyze the operation to suppress the * forbidden attributes in it so that they are not added in the local * backend. This must be called before any other plugin is called, to * keep coherency across plugin calls. */ fractionalFilterOperation(addOperation, true); } else { /* * Direct access from an LDAP client : if some attributes are to be * removed according to the fractional configuration, simply forbid * the operation */ if (fractionalFilterOperation(addOperation, false)) { LocalizableMessage msg = NOTE_ERR_FRACTIONAL_FORBIDDEN_OPERATION.get(getBaseDN(), addOperation); return new SynchronizationProviderResult.StopProcessing( ResultCode.UNWILLING_TO_PERFORM, msg); } } } if (addOperation.isSynchronizationOperation()) { AddContext ctx = (AddContext) addOperation.getAttachment(SYNCHROCONTEXT); /* * If an entry with the same entry uniqueID already exist then * this operation has already been replayed in the past. */ String uuid = ctx.getEntryUUID(); if (findEntryDN(uuid) != null) { return new SynchronizationProviderResult.StopProcessing( ResultCode.NO_OPERATION, null); } /* The parent entry may have been renamed here since the change was done * on the first server, and another entry have taken the former dn * of the parent entry */ String parentEntryUUID = ctx.getParentEntryUUID(); // root entry have no parent, there is no need to check for it. if (parentEntryUUID != null) { // There is a potential of perfs improvement here // if we could avoid the following parent entry retrieval DN parentDnFromCtx = findEntryDN(ctx.getParentEntryUUID()); if (parentDnFromCtx == null) { // The parent does not exist with the specified unique id // stop the operation with NO_SUCH_OBJECT and let the // conflict resolution or the dependency resolution solve this. return new SynchronizationProviderResult.StopProcessing( ResultCode.NO_SUCH_OBJECT, null); } DN entryDN = addOperation.getEntryDN(); DN parentDnFromEntryDn = getServerContext().getBackendConfigManager() .getParentDNInSuffix(entryDN); if (parentDnFromEntryDn != null && !parentDnFromCtx.equals(parentDnFromEntryDn)) { // parentEntry has been renamed // replication name conflict resolution is expected to fix that // later in the flow return new SynchronizationProviderResult.StopProcessing( ResultCode.NO_SUCH_OBJECT, null); } } } return new SynchronizationProviderResult.ContinueProcessing(); } /** * Check that the broker associated to this ReplicationDomain has found * a Replication Server and that this LDAP server is therefore able to * process operations. * If not, set the ResultCode, the response message, * interrupt the operation, and return false * * @return true when it OK to process the Operation, false otherwise. * When false is returned the resultCode and the response message * is also set in the Operation. */ private boolean brokerIsConnected() { final IsolationPolicy isolationPolicy = config.getIsolationPolicy(); if (isolationPolicy.equals(IsolationPolicy.ACCEPT_ALL_UPDATES)) { // this policy imply that we always accept updates. return true; } if (isolationPolicy.equals(IsolationPolicy.REJECT_ALL_UPDATES)) { // this isolation policy specifies that the updates are denied // when the broker had problems during the connection phase // Updates are still accepted if the broker is currently connecting.. return !hasConnectionError(); } // we should never get there as the only possible policies are // ACCEPT_ALL_UPDATES and REJECT_ALL_UPDATES return true; } /** * Implement the handleConflictResolution phase of the ModifyDNOperation. * * @param modifyDNOperation The ModifyDNOperation. * @return A SynchronizationProviderResult indicating if the operation * can continue. */ SynchronizationProviderResult handleConflictResolution( PreOperationModifyDNOperation modifyDNOperation) { if (!modifyDNOperation.isSynchronizationOperation() && !brokerIsConnected()) { LocalizableMessage msg = ERR_REPLICATION_COULD_NOT_CONNECT.get(getBaseDN()); return new SynchronizationProviderResult.StopProcessing( ResultCode.UNWILLING_TO_PERFORM, msg); } if (fractionalConfig.isFractional()) { if (modifyDNOperation.isSynchronizationOperation()) { /* * Filter operation here for fractional replication. If fractional * replication is enabled, we analyze the operation and modify it if * necessary to stay consistent with what is defined in fractional * configuration. */ fractionalFilterOperation(modifyDNOperation, true); } else { /* * Direct access from an LDAP client : something is inconsistent with * the fractional configuration, forbid the operation. */ if (fractionalFilterOperation(modifyDNOperation, false)) { LocalizableMessage msg = NOTE_ERR_FRACTIONAL_FORBIDDEN_OPERATION.get(getBaseDN(), modifyDNOperation); return new SynchronizationProviderResult.StopProcessing( ResultCode.UNWILLING_TO_PERFORM, msg); } } } ModifyDnContext ctx = (ModifyDnContext) modifyDNOperation.getAttachment(SYNCHROCONTEXT); if (ctx != null) { /* * This is a replication operation * Check that the modified entry has the same entryuuid * as was in the original message. */ final String modifiedEntryUUID = getEntryUUID(modifyDNOperation.getOriginalEntry()); if (!modifiedEntryUUID.equals(ctx.getEntryUUID())) { /* * The modified entry is not the same entry as the one on * the original change was performed. * Probably the original entry was renamed and replaced with * another entry. * We must not let the change proceed, return a negative * result and set the result code to NO_SUCH_OBJECT. * When the operation will return, the thread that started the operation * will try to find the correct entry and restart a new operation. */ return new SynchronizationProviderResult.StopProcessing( ResultCode.NO_SUCH_OBJECT, null); } if (modifyDNOperation.getNewSuperior() != null) { /* * Also check that the current id of the * parent is the same as when the operation was performed. */ String newParentId = findEntryUUID(modifyDNOperation.getNewSuperior()); if (newParentId != null && ctx.getNewSuperiorEntryUUID() != null && !newParentId.equals(ctx.getNewSuperiorEntryUUID())) { return new SynchronizationProviderResult.StopProcessing( ResultCode.NO_SUCH_OBJECT, null); } } /* * If the object has been renamed more recently than this * operation, cancel the operation. */ EntryHistorical hist = EntryHistorical.newInstanceFromEntry( modifyDNOperation.getOriginalEntry()); if (hist.addedOrRenamedAfter(ctx.getCSN())) { return new SynchronizationProviderResult.StopProcessing( ResultCode.NO_OPERATION, null); } } else { // There is no replication context attached to the operation // so this is not a replication operation. CSN csn = generateCSN(modifyDNOperation); String newParentId = null; if (modifyDNOperation.getNewSuperior() != null) { newParentId = findEntryUUID(modifyDNOperation.getNewSuperior()); } Entry modifiedEntry = modifyDNOperation.getOriginalEntry(); String modifiedEntryUUID = getEntryUUID(modifiedEntry); ctx = new ModifyDnContext(csn, modifiedEntryUUID, newParentId); modifyDNOperation.setAttachment(SYNCHROCONTEXT, ctx); } return new SynchronizationProviderResult.ContinueProcessing(); } /** * Handle the conflict resolution. * Called by the core server after locking the entry and before * starting the actual modification. * @param modifyOperation the operation * @return code indicating is operation must proceed */ SynchronizationProviderResult handleConflictResolution( PreOperationModifyOperation modifyOperation) { if (!modifyOperation.isSynchronizationOperation() && !brokerIsConnected()) { LocalizableMessage msg = ERR_REPLICATION_COULD_NOT_CONNECT.get(getBaseDN()); return new SynchronizationProviderResult.StopProcessing( ResultCode.UNWILLING_TO_PERFORM, msg); } if (fractionalConfig.isFractional()) { if (modifyOperation.isSynchronizationOperation()) { /* * Filter attributes here for fractional replication. If fractional * replication is enabled, we analyze the operation and modify it so * that no forbidden attribute is added/modified/deleted in the local * backend. This must be called before any other plugin is called, to * keep coherency across plugin calls. */ if (fractionalFilterOperation(modifyOperation, true) == FRACTIONAL_BECOME_NO_OP) { // Every modifications filtered in this operation: the operation // becomes a no-op return new SynchronizationProviderResult.StopProcessing( ResultCode.NO_OPERATION, null); } } else { /* * Direct access from an LDAP client : if some attributes are to be * removed according to the fractional configuration, simply forbid * the operation */ switch(fractionalFilterOperation(modifyOperation, false)) { case FRACTIONAL_HAS_NO_FRACTIONAL_FILTERED_ATTRIBUTES: // Ok, let the operation happen break; case FRACTIONAL_HAS_FRACTIONAL_FILTERED_ATTRIBUTES: // Some attributes not compliant with fractional configuration : // forbid the operation LocalizableMessage msg = NOTE_ERR_FRACTIONAL_FORBIDDEN_OPERATION.get(getBaseDN(), modifyOperation); return new SynchronizationProviderResult.StopProcessing( ResultCode.UNWILLING_TO_PERFORM, msg); } } } ModifyContext ctx = (ModifyContext) modifyOperation.getAttachment(SYNCHROCONTEXT); Entry modifiedEntry = modifyOperation.getModifiedEntry(); if (ctx == null) { // No replication ctx attached => not a replicated operation // - create a ctx with : CSN, entryUUID // - attach the context to the op CSN csn = generateCSN(modifyOperation); ctx = new ModifyContext(csn, getEntryUUID(modifiedEntry)); modifyOperation.setAttachment(SYNCHROCONTEXT, ctx); } else { // Replication ctx attached => this is a replicated operation being // replayed here, it is necessary to // - check if the entry has been renamed // - check for conflicts String modifiedEntryUUID = ctx.getEntryUUID(); String currentEntryUUID = getEntryUUID(modifiedEntry); if (currentEntryUUID != null && !currentEntryUUID.equals(modifiedEntryUUID)) { /* * The current modified entry is not the same entry as the one on * the original modification was performed. * Probably the original entry was renamed and replaced with * another entry. * We must not let the modification proceed, return a negative * result and set the result code to NO_SUCH_OBJECT. * When the operation will return, the thread that started the * operation will try to find the correct entry and restart a new * operation. */ return new SynchronizationProviderResult.StopProcessing( ResultCode.NO_SUCH_OBJECT, null); } // Solve the conflicts between modify operations EntryHistorical historicalInformation = EntryHistorical.newInstanceFromEntry(modifiedEntry); modifyOperation.setAttachment(EntryHistorical.HISTORICAL, historicalInformation); if (historicalInformation.replayOperation(modifyOperation, modifiedEntry)) { numResolvedModifyConflicts.incrementAndGet(); } } return new SynchronizationProviderResult.ContinueProcessing(); } /** * The preOperation phase for the add Operation. * Its job is to generate the replication context associated to the * operation. It is necessary to do it in this phase because contrary to * the other operations, the entry UUID is not set when the handleConflict * phase is called. * * @param addOperation The Add Operation. */ void doPreOperation(PreOperationAddOperation addOperation) { final CSN csn = generateCSN(addOperation); final String entryUUID = getEntryUUID(addOperation); final AddContext ctx = new AddContext(csn, entryUUID, findEntryUUID(getServerContext().getBackendConfigManager() .getParentDNInSuffix(addOperation.getEntryDN()))); addOperation.setAttachment(SYNCHROCONTEXT, ctx); } @Override public void publishReplicaOfflineMsg() { final CSN offlineCSN = pendingChanges.putReplicaOfflineMsg(); dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), offlineCSN); } /** * Check if an operation must be synchronized. * Also update the list of pending changes and the server RUV * @param op the operation */ void synchronize(PostOperationOperation op) { ResultCode result = op.getResultCode(); // Note that a failed non-replication operation might not have a change // number. CSN curCSN = OperationContext.getCSN(op); if (curCSN != null && config.isLogChangenumber()) { op.addAdditionalLogItem(AdditionalLogItem.unquotedKeyValue(getClass(), "replicationCSN", curCSN)); } if (result == ResultCode.SUCCESS) { if (op.isSynchronizationOperation()) { // Replaying a sync operation numReplayedPostOpCalled.incrementAndGet(); try { remotePendingChanges.commit(curCSN); } catch (NoSuchElementException e) { logger.error(ERR_OPERATION_NOT_FOUND_IN_PENDING, op, curCSN); return; } resetSessionRestartBackoff(); } else { // Generate a replication message for a successful non-replication // operation. LDAPUpdateMsg msg = LDAPUpdateMsg.generateMsg(op); if (msg == null) { /* * This is an operation type that we do not know about * It should never happen. */ pendingChanges.remove(curCSN); logger.error(ERR_UNKNOWN_TYPE, op.getOperationType()); return; } addEntryAttributesForCL(msg,op); // If assured replication is configured, this will prepare blocking // mechanism. If assured replication is disabled, this returns // immediately prepareWaitForAckIfAssuredEnabled(msg); try { msg.encode(); pendingChanges.commitAndPushCommittedChanges(curCSN, msg); } catch (NoSuchElementException e) { logger.error(ERR_OPERATION_NOT_FOUND_IN_PENDING, op, curCSN); return; } // If assured replication is enabled, this will wait for the matching // ack or time out. If assured replication is disabled, this returns // immediately try { waitForAckIfAssuredEnabled(msg); } catch (TimeoutException ex) { // This exception may only be raised if assured replication is enabled logger.info(NOTE_DS_ACK_TIMEOUT, getBaseDN(), getAssuredTimeout(), msg); } } /* * If the operation is a DELETE on the base entry of the suffix * that is replicated, the generation is now lost because the * DB is empty. We need to save it again the next time we add an entry. */ if (OperationType.DELETE.equals(op.getOperationType()) && ((PostOperationDeleteOperation) op) .getEntryDN().equals(getBaseDN())) { generationIdSavedStatus = false; } if (!generationIdSavedStatus) { saveGenerationId(generationId); } } else if (!op.isSynchronizationOperation() && curCSN != null) { // Remove an unsuccessful non-replication operation from the pending // changes list. pendingChanges.remove(curCSN); pendingChanges.pushCommittedChanges(); } checkForClearedConflict(op); } /** * Check if the operation that just happened has cleared a conflict : * Clearing a conflict happens if the operation has free a DN that * for which an other entry was in conflict. * Steps: * - get the DN freed by a DELETE or MODRDN op * - search for entries put in the conflict space (dn=entryUUID'+'....) * because the expected DN was not available (ds-sync-conflict=expected DN) * - retain the entry with the oldest conflict * - rename this entry with the freedDN as it was expected originally */ private void checkForClearedConflict(PostOperationOperation op) { OperationType type = op.getOperationType(); if (op.getResultCode() != ResultCode.SUCCESS) { // those operations cannot have cleared a conflict return; } DN freedDN; if (type == OperationType.DELETE) { freedDN = ((PostOperationDeleteOperation) op).getEntryDN(); } else if (type == OperationType.MODIFY_DN) { freedDN = ((PostOperationModifyDNOperation) op).getEntryDN(); } else { return; } SearchFilter filter; try { filter = LDAPFilter.createEqualityFilter(DS_SYNC_CONFLICT, ByteString.valueOfUtf8(freedDN.toString())).toSearchFilter(); } catch (DirectoryException e) { // can not happen? logger.traceException(e); return; } SearchRequest request = newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, filter) .addAttribute(USER_AND_REPL_OPERATIONAL_ATTRS); InternalSearchOperation searchOp = conn.processSearch(request); Entry entryToRename = null; CSN entryToRenameCSN = null; for (SearchResultEntry entry : searchOp.getSearchEntries()) { EntryHistorical history = EntryHistorical.newInstanceFromEntry(entry); if (entryToRename == null) { entryToRename = entry; entryToRenameCSN = history.getDNDate(); } else if (!history.addedOrRenamedAfter(entryToRenameCSN)) { // this conflict is older than the previous, keep it. entryToRename = entry; entryToRenameCSN = history.getDNDate(); } } if (entryToRename != null) { DN entryDN = entryToRename.getName(); ModifyDNOperation newOp = renameEntry( entryDN, freedDN.rdn(), freedDN.parent(), false); ResultCode res = newOp.getResultCode(); if (res != ResultCode.SUCCESS) { logger.error(ERR_COULD_NOT_SOLVE_CONFLICT, entryDN, res); } } } /** * Rename an Entry Using a synchronization, non-replicated operation. * This method should be used instead of the InternalConnection methods * when the operation that need to be run must be local only and therefore * not replicated to the RS. * * @param targetDN The DN of the entry to rename. * @param newRDN The new RDN to be used. * @param parentDN The parentDN to be used. * @param markConflict A boolean indicating is this entry should be marked * as a conflicting entry. In such case the * DS_SYNC_CONFLICT attribute will be added to the entry * with the value of its original DN. * If false, the DS_SYNC_CONFLICT attribute will be * cleared. * * @return The operation that was run to rename the entry. */ private ModifyDNOperation renameEntry(DN targetDN, RDN newRDN, DN parentDN, boolean markConflict) { ModifyDNOperation newOp = new ModifyDNOperationBasis( conn, nextOperationID(), nextMessageID(), new ArrayList(0), targetDN, newRDN, false, parentDN); if (markConflict) { Attribute attr = Attributes.create(DS_SYNC_CONFLICT, targetDN.toString()); newOp.addModification(new Modification(ModificationType.REPLACE, attr)); } else { Attribute attr = Attributes.empty(DS_SYNC_CONFLICT); newOp.addModification(new Modification(ModificationType.DELETE, attr)); } runAsSynchronizedOperation(newOp); return newOp; } private void runAsSynchronizedOperation(Operation op) { op.setInternalOperation(true); op.setSynchronizationOperation(true); op.setDontSynchronize(true); op.run(); } /** Delete this ReplicationDomain. */ void delete() { shutdown(); removeECLDomainCfg(); } /** Shutdown this ReplicationDomain. */ public void shutdown() { if (shutdown.compareAndSet(false, true)) { final RSUpdater rsUpdater = this.rsUpdater.get(); if (rsUpdater != null) { rsUpdater.initiateShutdown(); } // stop the thread in charge of flushing the ServerState. if (flushThread != null) { flushThread.initiateShutdown(); synchronized (flushThread) { flushThread.notifyAll(); } } DirectoryServer.deregisterAlertGenerator(this); getServerContext().getBackendConfigManager() .deregisterLocalBackendInitializationListener(this); DirectoryServer.deregisterShutdownListener(this); // stop the ReplicationDomain, under the lock which the replay thread restarting the // session after a failed replay holds: it must not bring a broker and a listener // thread back up on a domain whose alert generator, flush thread and RSUpdater have // just been taken away. synchronized (serviceStateLock) { disableService(); sessionGeneration++; } } // wait for completion of the ServerStateFlush thread. try { while (!done) { Thread.sleep(50); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } /** * Marks the specified message as the one currently processed by a replay thread. * * @param msg the message being processed * @return {@code false} if the change is not pending anymore, which happens when the * session was restarted after a failed replay while this message was waiting * in the replay queue: it is sent again, so this copy must not be replayed */ boolean markInProgress(LDAPUpdateMsg msg) { if (remotePendingChanges.markInProgress(msg)) { return true; } /* * This delivery is not replayed, but it was taken off the replay queue all the same: * count it as processed, or replication-processed-updates would drift away from what * the session delivered - that attribute counts the deliveries this replica took off * the session, not the ones which reached the replay. No ack is owed for it either * way, though not always for the same reason. When the change was taken over, this * copy came over a session which has been torn down since and the delivery which took * over from it carries the ack. When a disabled domain forgot the change, no delivery * takes over and none is acked: the session that one came over is gone as well, so an * ack would reach nobody, and a server waiting on an assured write times it out the * way it does for every change in flight when a domain is taken out of the topology * for an import or a restore. */ incProcessedUpdates(); return false; } /** * Create and replay a synchronized Operation from an UpdateMsg. * * @param msg * The UpdateMsg to be replayed. * @param replayThreadShutdown * whether the replay thread was asked to stop */ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) { // Try replay the operation, then flush (replaying) any pending operation // whose dependency has been replayed until no more left. do { Operation op = null; // the last operation on which replay was attempted boolean dependency = false; boolean replayFailed = false; boolean replayAbandoned = false; String replayErrorMsg = null; CSN csn = null; try { // The next operation for which to attempt replay. // This local variable allow to keep error messages in the "op" local // variable until the next loop iteration starts. // "op" is already initialized to the next Operation because of the // error handling paths. Operation nextOp = op = msg.createOperation(conn); /* * The code this server puts on an internal error is a configuration knob: it is * read once here so that every attempt of this delivery, and the verdict which * follows them, are judged against the same one. Read inside the try - the ack of * a delivery is published in the finally below, whatever the delivery ran into - * and once the operation is built, so that a failure of the read is a failure of * an attempt rather than a message no operation could be built from. */ final ResultCode serverErrorResultCode = getServerContext().getCoreConfigManager().getServerErrorResultCode(); dependency = remotePendingChanges.checkDependencies(op, msg); boolean replayDone = false; boolean firstAttempt = true; int retryCount = IN_PLACE_REPLAY_ATTEMPTS; while (!dependency && !replayDone && retryCount-- > 0) { if (replayThreadShutdown.get() || shutdown.get() || disabled) { /* * Either this replay thread or this domain is going away, or the domain is * being imported into or restored, so let's leave. The change was never * applied, so the ack says so - an assured write must not be told that a * change this replica is asking for again is in the data here - and the change * is given back to the replication server rather than left listed as being * replayed by a thread which is gone. Handing it back stops and starts the * session, so it waits until the ack has been published on the session this * delivery came over. * * A disabled domain saved its ServerState, cleared it from memory and forgot * its pending changes: a thread which kept applying changes into the backend * being imported into would have every one of its commits fail on a map which * is empty, one ERR_OPERATION_NOT_FOUND_IN_PENDING per change in flight, and * would be writing into a backend the import owns. abandonReplay() knows there * is nothing left to hand back in that case. */ replayErrorMsg = NOTE_REPLAY_ABANDONED_CHANGE.get(msg.getCSN(), getBaseDN()).toString(); replayAbandoned = true; replayDone = true; break; } if (!firstAttempt) { /* * Every attempt runs an operation of its own. An Operation which already ran * carries the request controls and the access log items of that run, so * re-running the same one stacks one ManageDsaIT control - and one access log * record - per attempt. It also picks up the new state of the UpdateMsg when * conflict resolution rewrote it. * Note: When msg is a DeleteMsg, the DeleteOperation is properly created * with subtreeDelete request control when needed. */ nextOp = msg.createOperation(conn); } firstAttempt = false; // Try replay the operation op = nextOp; op.setInternalOperation(true); op.setSynchronizationOperation(true); // Always add the ManageDSAIT control so that updates to referrals // are processed locally. op.addRequestControl(new LDAPControl(OID_MANAGE_DSAIT_CONTROL)); // Warning: specific processing ahead. See OPENDJ-2792 if (op instanceof ModifyOperation) { ModifyOperation modifyOperation = (ModifyOperation) op; if (modifyOperation.getEntryDN().equals(SET_PERMISSIVE_MODIFY_FOR_DN)) { op.addRequestControl(new LDAPControl(OID_PERMISSIVE_MODIFY_CONTROL)); } } csn = OperationContext.getCSN(op); op.run(); ResultCode result = op.getResultCode(); if (result != ResultCode.SUCCESS) { if (result == ResultCode.NO_OPERATION) { // Pre-operation conflict resolution detected that the operation // was a no-op. For example, an add which has already been // replayed, or a modify DN operation on an entry which has been // renamed by a more recent modify DN. // The change is in the data: push it to the serverState. replayDone = true; recordChangeResolved(csn); } else if (result == ResultCode.BUSY) { /* * We probably could not get a lock (OPENDJ-885). Give the server * another chance to process this operation immediately. */ Thread.yield(); continue; } else if (isServerFailure(result, serverErrorResultCode)) { /* * It can happen when a rebuild is performed or the backend is * offline (OPENDJ-49), or when the storage failed to serve the * operation. Give the server another chance to process this * operation after some time. */ Thread.sleep(50); continue; } else { ConflictResolution resolution = ConflictResolution.NOTHING_TO_DO; if (op instanceof ModifyOperation) { ModifyOperation castOp = (ModifyOperation) op; dependency = remotePendingChanges.checkDependencies(castOp); ModifyMsg modifyMsg = (ModifyMsg) msg; resolution = dependency ? resolution : solveNamingConflict(castOp, modifyMsg); } else if (op instanceof DeleteOperation) { DeleteOperation castOp = (DeleteOperation) op; dependency = remotePendingChanges.checkDependencies(castOp); resolution = dependency ? resolution : solveNamingConflict(castOp, msg); } else if (op instanceof AddOperation) { AddOperation castOp = (AddOperation) op; AddMsg addMsg = (AddMsg) msg; dependency = remotePendingChanges.checkDependencies(castOp); resolution = dependency ? resolution : solveNamingConflict(castOp, addMsg); } else if (op instanceof ModifyDNOperation) { ModifyDNOperation castOp = (ModifyDNOperation) op; ModifyDNMsg modifyDNMsg = (ModifyDNMsg) msg; dependency = remotePendingChanges.checkDependencies(modifyDNMsg); resolution = dependency ? resolution : solveNamingConflict(castOp, modifyDNMsg); } // else: unknown type of operation ?! there is nothing to replay if (!dependency) { switch (resolution) { case NOTHING_TO_DO: // the update became a dummy update and the result // of the conflict resolution phase is to do nothing. // however we still need to push this change to the serverState replayDone = true; recordChangeResolved(csn); break; case FAILED: if (serverErrorResultCode.equals(result)) { /* * The result code is the one this server puts on an internal error and is * one conflict resolution knows how to solve, so the change was left to it * rather than treated as a failure of the server: it had its chance and * could not solve it, so the storage failing is what is left. Give it the * in-place attempts an UNAVAILABLE gets - a storage busy for a moment must * not cost a session restart - and leave the change out of the ServerState * once they are spent, which the failure of the server below the loop * reports and acts on, reading the result of the attempt which spent the * last of them. A change which is not in the data must not advance the * ServerState (issue #889). */ Thread.sleep(50); break; } /* * The operation did not fail on a naming conflict and not on the server * either: the change can not be applied on this replica. Skip it so that the * replica keeps replaying the changes which follow, but report the error in * the ack and tell the administrator that the data now diverge. */ final LocalizableMessage errorMsg = ERR_ERROR_REPLAYING_OPERATION.get( op, csn, result, op.getErrorMessage()); logger.error(errorMsg); replayErrorMsg = errorMsg.toString(); replayDone = true; skipUnreplayableChange(csn, errorMsg); break; default: /* * Try replaying the change again: the next attempt creates an operation * reflecting the new state of the UpdateMsg after conflict resolution * modified it, and dependencies might have been replayed by now. */ break; } } } } else { replayDone = true; } } if (!replayDone && !dependency) { final ResultCode lastResult = op.getResultCode(); /* * BUSY is a lock which could not be taken (OPENDJ-885): the in-place attempts * only yield to the thread which holds it, so a lock held for a while burns * every one of them in no time. It is as transient as a storage which failed, * and the change is just as absent from the data. So is a change whose last * attempt came back with the configured server-error-result-code, whether or * not conflict resolution owns that code: it had its chance, and what is left * is the storage failing to serve the operation. The result of that attempt is * what decides, so that this branch reports the failure it is acting on: an * attempt which ended on something conflict resolution kept rewriting is the * loop below, however the attempts before it ended. */ if (isServerFailure(lastResult, serverErrorResultCode) || ResultCode.BUSY.equals(lastResult) || serverErrorResultCode.equals(lastResult)) { /* * The server kept failing to apply the change, so the change is not in the data. * Leave it out of the ServerState, otherwise the replication server would never * send it again and this replica would silently diverge while reporting itself * up to date. */ final LocalizableMessage message = ERR_ERROR_REPLAYING_OPERATION.get( op, csn, lastResult, op.getErrorMessage()); logger.error(message); replayErrorMsg = message.toString(); replayFailed = true; } else { // Conflict resolution kept rewriting an operation which kept failing. // Continue with the next change but the servers could now become inconsistent. // Let the repair tool know about this. final LocalizableMessage message = ERR_LOOP_REPLAYING_OPERATION.get( op, op.getErrorMessage()); logger.error(message); numUnresolvedNamingConflicts.incrementAndGet(); replayErrorMsg = message.toString(); skipUnreplayableChange(csn, message); } } } catch (Exception e) { if (op == null) { /* * No operation could be built from this message: createOperation() threw, * whether it said so with a decoding exception or with an unchecked one. There * is nothing to retry and no delivery which would build one any better. * * The decoding exceptions are caught here rather than in a catch of their * own because such a catch would span the whole replay: addConflict(), which * solveNamingConflict() calls once the operation has run, declares one, and a * change whose operation ran must never be given up on where it failed. */ replayErrorMsg = giveUpOnUndecodableChange(msg, e); } else { /* * An Exception happened during the replay process: the change is not in the * data, so it must not be recorded as replayed. * Let the repair tool know about this. * * The operation was built, so whatever failed is a failure of this attempt * rather than a verdict on every delivery of the change - including a failure * before the CSN of the operation was read, such as the entry DN of a * ModifyMsg which does not parse leaving getEntryDN() null. Giving up on it * where it is reported would record a change which never reached the backend * in the ServerState, which is issue #889 by another route; it is left out of * the ServerState and asked for again instead, and the give-up budget bounds * how long this replica keeps asking. */ final LocalizableMessage message = ERR_EXCEPTION_REPLAYING_OPERATION.get(op, stackTraceToSingleLineString(e)); logger.error(message); replayErrorMsg = message.toString(); replayFailed = true; } } finally { if (!dependency) { /* * The ack is per delivery, and it says what this delivery did: a change which * failed is not in the data now, whether or not the delivery which follows * manages to apply it. Holding the ack back until the change is resolved would * not tell the truth any better - the session it came over is about to be torn * down, so nothing would reach the server which is waiting for it, and an * assured write would wait out its timeout rather than be told what happened. */ processUpdateDone(msg, replayErrorMsg); } } if (replayAbandoned) { /* * The ack has been published, so the change can be handed back now and the session * restarted for it: this thread is on its way out either way. */ abandonReplay(msg.getCSN()); return; } /* * The CSN of the change is read off the message rather than off the operation: the * two are the same - the operation carries the CSN of the message it was built from * - but a failure which happened before the operation was built, or before its CSN * was read, has a change to ask for again all the same. */ if (replayFailed && recoverFromReplayFailure(msg.getCSN(), replayThreadShutdown)) { // The ack has been published and the change, still owned by the replication // server, is being delivered again: there is nothing left to replay here. return; } // Now replay any pending update that had a dependency and whose // dependency has been replayed, do that until no more updates of that // type left... msg = remotePendingChanges.getNextUpdate(); } while (msg != null); } /** * Reports a message this replica can not turn into an operation, and gives up on the * change it carries. *

* There is no operation to retry and no delivery which would decode any better, so the * change is skipped rather than left out of the ServerState: a change which stays * listed and uncommitted is the barrier which holds this domain's ServerState - and * every change which follows it, from every master - back for good, since nothing asks * for it again and the delivery which would is turned down while a replay thread still * owns it. Skipping it says out loud what a wedged domain would only have implied: this * replica has diverged and must be reinitialized. *

* Only a message which no operation could be built from comes here, and it is the * {@code op == null} of its single caller which says so rather than the type of the * exception: a decoding exception is declared past the point where the operation ran * as well - by addConflict() - so a catch which read the type would give up on a * change the backend may well have applied. A failure of the replay of an operation * which was built, whenever it happens, keeps its change out of the ServerState and * has it delivered again instead: that one is a failure of an attempt, not of every * delivery of the change. * * @param msg the message which could not be decoded * @param e the failure to decode it * @return the error to report in the ack of this delivery */ private String giveUpOnUndecodableChange(LDAPUpdateMsg msg, Exception e) { LocalizableMessage message = ERR_EXCEPTION_DECODING_OPERATION.get(msg + " " + stackTraceToSingleLineString(e)); logger.error(message); skipUnreplayableChange(msg.getCSN(), message); return message.toString(); } /** * This method is called when an error happens while replaying * an operation. * It is necessary because the postOperation does not always get * called when error or Exceptions happen during the operation replay. * * @param csn the CSN of the operation with error. * @return {@code false} if the change was not listed in the pending changes anymore, * so that it has not been recorded as replayed: the replication server sends * it again. */ private boolean updateError(CSN csn) { try { remotePendingChanges.commit(csn); return true; } catch (NoSuchElementException e) { // A failure occurred after the change had been removed from the pending // changes table. if (logger.isTraceEnabled()) { logger.trace( "LDAPReplicationDomain.updateError: Unable to find remote " + "pending change for CSN %s", csn); } return false; } } /** * Returns whether the provided result code reports a failure of this server rather * than a change which can not be applied: the backend being offline or rebuilt * (OPENDJ-49), or the storage failing to serve the operation. * * @param result the result code of a replayed operation * @param serverErrorResultCode the result code this server puts on an internal error * @return {@code true} if the operation failed on the server itself */ private static boolean isServerFailure(ResultCode result, ResultCode serverErrorResultCode) { /* * The result code the server puts on an internal error is configurable and is not * validated as a result code, so it may well be one conflict resolution knows how to * solve: such a setting must not take a change away from solveNamingConflict(), which * is the only thing which can solve them. A change it could not solve either is a * failure of the server all the same, which replay() acts on once conflict resolution * has reported it. */ return ResultCode.UNAVAILABLE.equals(result) || (serverErrorResultCode.equals(result) && !CONFLICT_RESULT_CODES.contains(result)); } /** * Returns a time which only ever moves forward, in milliseconds. *

* How long a change has been failing and how long ago the last alert was sent are * durations rather than dates: the wall clock stepping backwards must not have this * replica retry a change for good, and stepping forwards must not have it give up on a * change it had only just started to retry. * * @return the number of milliseconds since an arbitrary origin */ private static long monotonicNowInMs() { return TimeUnit.NANOSECONDS.toMillis(System.nanoTime()); } /** * Records a change which conflict resolution found nothing left to do for: the change * is in the data, so it is recorded as replayed and the failures it went through - and * the session restarts they caused - are history. * * @param csn the CSN of the change */ private void recordChangeResolved(CSN csn) { updateError(csn); resetSessionRestartBackoff(); } /** * Has the change which fails next start the backoff between the session restarts over, * if this replica is not failing any change anymore. *

* A change made it and nothing is failing anymore, so the backend is serving again and * the session is not being restarted in a row. While something is still failing, a * change which was replayed says nothing of the kind - a change which can never be * applied here fails alone, among changes which replay perfectly well, and letting * those reset the wait would have this domain tear its session down every second for as * long as that one change takes to be given up on. */ private void resetSessionRestartBackoff() { if (!remotePendingChanges.hasFailingChanges()) { consecutiveSessionRestarts.set(0); } } /** * Records a change which could not be replayed as replayed anyway, so that this replica * keeps replaying the changes which follow it, and warns that the data now diverge. *

* The backoff between the session restarts is deliberately left alone: giving up on a * change is not a change being replayed, and whatever made this one unreplayable is * still failing the ones which are in flight with it. Restarting it from its shortest * wait would have a replica which gives up on a change now and then ask for every * change of an outage as fast as the replication server can send them, which is what * {@link #consecutiveSessionRestarts} is there to prevent. * * @param csn the CSN of the change which could not be replayed * @param cause the message describing why it could not be replayed */ private void skipUnreplayableChange(CSN csn, LocalizableMessage cause) { if (updateError(csn)) { numFailedReplayedUpdates.incrementAndGet(); sendUnreplayedChangeAlert(cause); } // Otherwise the change is not listed as pending anymore - the domain was disabled // while it was being replayed - so it has not been skipped: the replication server // sends it again and this replica gives up on it then. } /** * Tells the administrator that this replica gave up on a change and now diverges from * the rest of the topology. *

* Whatever makes a change unreplayable - a schema which does not match, a backend * which is gone - makes every change in flight unreplayable too, so the alert is not * sent again for {@link #UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS}: each skipped change * is logged, the alert is there to have the administrator look at the log. * * @param cause the message describing why the change could not be replayed */ private void sendUnreplayedChangeAlert(LocalizableMessage cause) { final long now = monotonicNowInMs(); final long lastSent = lastUnreplayedChangeAlertTime.get(); if (now - lastSent >= UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS && lastUnreplayedChangeAlertTime.compareAndSet(lastSent, now)) { DirectoryServer.sendAlertNotification( this, ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE, cause); } } /** * Lets the next change this replica gives up on raise its alert straight away. *

* Only there for the tests which check the alert: they must not be at the mercy of the * alert another test raised less than * {@link #UNREPLAYED_CHANGE_ALERT_INTERVAL_IN_MS} ago. */ @VisibleForTesting public void resetUnreplayedChangeAlertThrottle() { lastUnreplayedChangeAlertTime.set(UNREPLAYED_CHANGE_ALERT_NEVER_SENT); } /** * Recovers from a change which could not be replayed. *

* The change has deliberately been left out of the ServerState, so the replication * server still owns it: restart the session so that it is sent again and replayed on * a backend which has hopefully recovered in the meantime. Give up once its replay has * been failing for {@link #REPLAY_GIVE_UP_DELAY_IN_MS} and record it as replayed, so * that a change which can never be applied here does not stop this replica for good: * the administrator is told that this replica has diverged and must be reinitialized. * * @param csn * the CSN of the change which could not be replayed * @param replayThreadShutdown * whether the replay thread was asked to stop * @return {@code true} when the caller must stop replaying because the session is * being restarted or is going away, {@code false} when it may carry on with * the changes which follow */ private boolean recoverFromReplayFailure(CSN csn, AtomicBoolean replayThreadShutdown) { /* * The failure is recorded, and the change given up on, while this thread still owns * it: a change which is listed, uncommitted and unowned is what putRemoteUpdate() * takes over, so releasing it before the decision is made would let another delivery * be replayed by another thread while this one goes on to record the change as * skipped. */ final long now = monotonicNowInMs(); final RemotePendingChanges.ReplayFailure failure = remotePendingChanges.recordReplayFailure(csn, now); if (failure == null) { /* * There is no uncommitted change left to give up on or to ask for again: the domain * was disabled while this one was being replayed - its ServerState was saved and is * read again from the backend when the domain is enabled back - or the change had * already been recorded when this failure was reported. Carry on with the changes * which were waiting for it; a domain on its way down lists none. */ return false; } if (failure.getFailingForMs() >= replayGiveUpDelayInMs) { final LocalizableMessage message = ERR_REPLAY_SKIPPING_CHANGE.get( csn, getBaseDN(), failure.getFailingForMs(), failure.getAttempts()); logger.error(message); skipUnreplayableChange(csn, message); return false; } /* * The change stays listed as pending and uncommitted - it holds the ServerState back * so that the replication server sends it again - but this thread does not own it * anymore: the next delivery is the one which is replayed, and the copy which may * still wait in the shared replay queue is dropped when a replay thread takes it out, * because markInProgress() only accepts the delivery which is listed as pending. */ remotePendingChanges.replayFailed(csn); if (shutdown.get() || disabled) { /* * This whole domain is going away or is being imported into: there is no session of * this thread's to restart. Restarting the one which is being stopped would leave a * broker and a listener thread behind on a domain whose alert generator, flush * thread and RSUpdater are already gone. */ return true; } logger.warn(WARN_REPLAY_RETRYING_CHANGE, csn, getBaseDN(), failure.getAttempts()); /* * This change is not owned by anyone anymore, so the session has to be restarted for * the replication server to deliver it again. Ask for the restart before trying to * run it: a restart which is already under way may have started before this change * was released, and the delivery it asked for would then have been turned down as a * duplicate of a change a replay thread still owned. */ sessionRestartRequested.set(true); /* * A replay thread which is stopping - the number of them is being changed - restarts * the session all the same: nothing else would ask for the change it just released, * and the ServerState would stay behind it for good. It does not sit through the * backoff on its way out, though: the backend is not what is going away. */ runRequestedSessionRestarts(!replayThreadShutdown.get()); return true; } /** * Restarts the session as long as changes which could not be replayed are waiting to be * delivered again. * * @param wait whether to leave the backend some time to recover between two restarts */ private void runRequestedSessionRestarts(boolean wait) { /* * The outer loop is what makes a request which was made while this thread was giving * up the recovery its own: the thread which made it found the recovery taken and left * it to this one. */ while (sessionRestartRequested.get() && replayFailureRecovery.compareAndSet(false, true)) { try { while (sessionRestartRequested.getAndSet(false)) { restartSession(wait); } } finally { replayFailureRecovery.set(false); } } } /** * Gives a change back to the replication server when this replay thread stops before it * could apply it. *

* The change is not owned by anyone anymore and it was never applied, so the session is * restarted for it to be delivered again: it is left out of the ServerState, and the * changes which follow it are held back until it is replayed. * * @param csn the CSN of the change this thread was replaying */ private void abandonReplay(CSN csn) { remotePendingChanges.replayFailed(csn); if (shutdown.get() || disabled) { // The domain owns its session, and it forgets its pending changes on its way down. return; } /* * Logged here rather than where the change is abandoned: a server which is shutting * down abandons every change in flight, and none of them is asked for again before it * is started back - one line per change would say otherwise. */ logger.info(NOTE_REPLAY_ABANDONED_CHANGE, csn, getBaseDN()); sessionRestartRequested.set(true); runRequestedSessionRestarts(false); } /** * Stops the session to the replication server and starts it again, so that the changes * this replica could not replay are delivered again. */ private void restartSession(boolean wait) { final long stoppedSession; synchronized (serviceStateLock) { if (shutdown.get() || disabled) { // The domain is going away or is being imported into: it owns its session. return; } disableService(); stoppedSession = ++sessionGeneration; } if (wait) { /* * Leave the backend some time to recover rather than ask for the change straight * away: a session restart is not free for the replication server either. The wait * is not held under the lock, or a domain being disabled for an import would wait * it out. */ waitBeforeSessionRestart(consecutiveSessionRestarts.incrementAndGet()); } synchronized (serviceStateLock) { if (shutdown.get() || disabled || sessionGeneration != stoppedSession || !isListenerShuttingDown()) { /* * The domain went away while this thread was waiting, or the session was stopped * and started again by something else - a configuration change, the end of an * import - in the meantime: the session this thread stopped is gone, so it has * nothing left to start. The generation says a session was started under this * lock; the listener says one is running, which is what a restart made outside it * leaves behind. */ return; } enableService(); sessionGeneration++; } } /** * Waits for a while before the session to the replication server is started again, so * that a backend which keeps failing is not asked for every change it can not apply as * fast as the replication server can send them. * * @param restarts how many times in a row the session was restarted already */ private void waitBeforeSessionRestart(int restarts) { try { Thread.sleep(Math.min(REPLAY_RETRY_DELAY_IN_MS * restarts, MAX_REPLAY_RETRY_DELAY_IN_MS)); } catch (InterruptedException e) { /* * Do not wait, but do start the session again all the same: the session is down * because this thread stopped it, and leaving it down would take this domain out of * the topology until the server is restarted. The interrupt is left set for whoever * asked this thread to stop. */ Thread.currentThread().interrupt(); } } /** * Returns how long the replay of a change is retried before this replica gives up on * it. *

* Only there for the tests, which set another value and put this one back. * * @return how long a change is retried, in milliseconds */ @VisibleForTesting public long getReplayGiveUpDelay() { return replayGiveUpDelayInMs; } /** * Sets how long the replay of a change is retried before this replica gives up on it. *

* Only there for the tests, which can not wait out the {@link * #REPLAY_GIVE_UP_DELAY_IN_MS} a backend under maintenance is given. * * @param delayInMs how long a change is retried, in milliseconds */ @VisibleForTesting public void setReplayGiveUpDelay(long delayInMs) { this.replayGiveUpDelayInMs = delayInMs; } /** * Generate a new CSN and insert it in the pending list. * * @param operation * The operation for which the CSN must be generated. * @return The new CSN. */ private CSN generateCSN(PluginOperation operation) { return pendingChanges.putLocalOperation(operation); } /** * Find the Unique Id of the entry with the provided DN by doing a * search of the entry and extracting its entryUUID from its attributes. * * @param dn The dn of the entry for which the unique Id is searched. * * @return The unique Id of the entry with the provided DN. */ static String findEntryUUID(DN dn) { if (dn == null) { return null; } final SearchRequest request = newSearchRequest(dn, SearchScope.BASE_OBJECT) .addAttribute(ENTRYUUID_ATTRIBUTE_NAME); final InternalSearchOperation search = getRootConnection().processSearch(request); final SearchResultEntry resultEntry = getFirstResult(search); if (resultEntry != null) { return getEntryUUID(resultEntry); } return null; } private static SearchResultEntry getFirstResult(InternalSearchOperation search) { if (search.getResultCode() == ResultCode.SUCCESS) { final LinkedList results = search.getSearchEntries(); if (!results.isEmpty()) { return results.getFirst(); } } return null; } /** * Find the current DN of an entry from its entry UUID. * * @param uuid the Entry Unique ID. * @return The current DN of the entry or null if there is no entry with * the specified UUID. */ private DN findEntryDN(String uuid) { try { final SearchRequest request = newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, "entryuuid=" + uuid); InternalSearchOperation search = conn.processSearch(request); final SearchResultEntry resultEntry = getFirstResult(search); if (resultEntry != null) { return resultEntry.getName(); } } catch (DirectoryException e) { // never happens because the filter is always valid. } return null; } /** Outcome of the conflict resolution attempted after a replayed operation failed. */ private enum ConflictResolution { /** The update message was adjusted: the operation must be replayed again. */ REPLAY_AGAIN, /** The change is already reflected in the data: there is nothing left to replay. */ NOTHING_TO_DO, /** The operation failed for a reason which is not a naming conflict. */ FAILED } /** * Solve a conflict detected when replaying a modify operation. * * @param op The operation that triggered the conflict detection. * @param msg The operation that triggered the conflict detection. * @return the outcome of the conflict resolution */ private ConflictResolution solveNamingConflict(ModifyOperation op, ModifyMsg msg) { ResultCode result = op.getResultCode(); ModifyContext ctx = (ModifyContext) op.getAttachment(SYNCHROCONTEXT); String entryUUID = ctx.getEntryUUID(); if (result == ResultCode.NO_SUCH_OBJECT) { /* * The operation is a modification but * the entry has been renamed on a different master in the same time. * search if the entry has been renamed, and return the new dn * of the entry. */ DN newDN = findEntryDN(entryUUID); if (newDN != null) { // There is an entry with the same unique id as this modify operation // replay the modify using the current dn of this entry. msg.setDN(newDN); numResolvedNamingConflicts.incrementAndGet(); return ConflictResolution.REPLAY_AGAIN; } else { // This entry does not exist anymore. // It has probably been deleted, stop the processing of this operation numResolvedNamingConflicts.incrementAndGet(); return ConflictResolution.NOTHING_TO_DO; } } else if (result == ResultCode.NOT_ALLOWED_ON_RDN) { DN currentDN = findEntryDN(entryUUID); RDN currentRDN; if (currentDN != null) { currentRDN = currentDN.rdn(); } else { // The entry does not exist anymore. numResolvedNamingConflicts.incrementAndGet(); return ConflictResolution.NOTHING_TO_DO; } // The modify operation is trying to delete the value that is // currently used in the RDN. We need to alter the modify so that it does // not remove the current RDN value(s). List mods = op.getModifications(); for (Modification mod : mods) { AttributeType modAttrType = mod.getAttribute().getAttributeDescription().getAttributeType(); if ((mod.getModificationType() == ModificationType.DELETE || mod.getModificationType() == ModificationType.REPLACE) && currentRDN.hasAttributeType(modAttrType)) { // the attribute can't be deleted because it is used in the RDN, // turn this operation is a replace with the current RDN value(s); mod.setModificationType(ModificationType.REPLACE); Attribute newAttribute = mod.getAttribute(); AttributeBuilder attrBuilder = new AttributeBuilder(newAttribute); attrBuilder.add(currentRDN.getAttributeValue(modAttrType)); mod.setAttribute(attrBuilder.toAttribute()); } } msg.setMods(mods); numResolvedNamingConflicts.incrementAndGet(); return ConflictResolution.REPLAY_AGAIN; } else { // The other type of errors can not be caused by naming conflicts: // the operation simply failed, replay() reports it. return ConflictResolution.FAILED; } } /** * Solve a conflict detected when replaying a delete operation. * * @param op The operation that triggered the conflict detection. * @param msg The operation that triggered the conflict detection. * @return the outcome of the conflict resolution */ private ConflictResolution solveNamingConflict(DeleteOperation op, LDAPUpdateMsg msg) { ResultCode result = op.getResultCode(); DeleteContext ctx = (DeleteContext) op.getAttachment(SYNCHROCONTEXT); String entryUUID = ctx.getEntryUUID(); if (result == ResultCode.NO_SUCH_OBJECT) { /* Find if the entry is still in the database. */ DN currentDN = findEntryDN(entryUUID); if (currentDN == null) { /* * The entry has already been deleted, either because this delete * has already been replayed or because another concurrent delete * has already done the job. * In any case, there is nothing more to do. */ numResolvedNamingConflicts.incrementAndGet(); return ConflictResolution.NOTHING_TO_DO; } else { // This entry has been renamed, replay the delete using its new DN. msg.setDN(currentDN); numResolvedNamingConflicts.incrementAndGet(); return ConflictResolution.REPLAY_AGAIN; } } else if (result == ResultCode.NOT_ALLOWED_ON_NONLEAF) { /* * This may happen when we replay a DELETE done on a master * but children of this entry have been added on another master. * * Rename all the children by adding entryuuid in dn and delete this entry. * * The action taken here must be consistent with the actions * done in the solveNamingConflict(AddOperation) method * when we are adding an entry whose parent entry has already been deleted. */ if (findAndRenameChild(op.getEntryDN(), op)) { numUnresolvedNamingConflicts.incrementAndGet(); } return ConflictResolution.REPLAY_AGAIN; } else { // The other type of errors can not be caused by naming conflicts: // the operation simply failed, replay() reports it. return ConflictResolution.FAILED; } } /** * Solve a conflict detected when replaying a Modify DN operation. * * @param op The operation that triggered the conflict detection. * @param msg The operation that triggered the conflict detection. * @return the outcome of the conflict resolution * @throws Exception When the operation is not valid. */ private ConflictResolution solveNamingConflict(ModifyDNOperation op, LDAPUpdateMsg msg) throws Exception { ResultCode result = op.getResultCode(); ModifyDnContext ctx = (ModifyDnContext) op.getAttachment(SYNCHROCONTEXT); String entryUUID = ctx.getEntryUUID(); String newSuperiorID = ctx.getNewSuperiorEntryUUID(); /* * four possible cases : * - the modified entry has been renamed * - the new parent has been renamed * - the operation is replayed for the second time. * - the entry has been deleted * action : * - change the target dn and the new parent dn and * restart the operation, * - don't do anything if the operation is replayed. */ // get the current DN of this entry in the database. DN currentDN = findEntryDN(entryUUID); // Construct the new DN to use for the entry. DN entryDN = op.getEntryDN(); DN newSuperior; RDN newRDN = op.getNewRDN(); if (newSuperiorID != null) { newSuperior = findEntryDN(newSuperiorID); } else { newSuperior = entryDN.parent(); } //If we could not find the new parent entry, we missed this entry // earlier or it has disappeared from the database // Log this information for the repair tool and mark the entry // as conflicting. // stop the processing. if (newSuperior == null) { markConflictEntry(op, currentDN, currentDN.parent().child(newRDN)); numUnresolvedNamingConflicts.incrementAndGet(); return ConflictResolution.NOTHING_TO_DO; } DN newDN = newSuperior.child(newRDN); if (currentDN == null) { // The entry targeted by the Modify DN is not in the database // anymore. // This is a conflict between a delete and this modify DN. // The entry has been deleted, we can safely assume // that the operation is completed. numResolvedNamingConflicts.incrementAndGet(); return ConflictResolution.NOTHING_TO_DO; } // if the newDN and the current DN match then the operation // is a no-op (this was probably a second replay) // don't do anything. if (newDN.equals(currentDN)) { numResolvedNamingConflicts.incrementAndGet(); return ConflictResolution.NOTHING_TO_DO; } if (result == ResultCode.NO_SUCH_OBJECT || result == ResultCode.UNWILLING_TO_PERFORM || result == ResultCode.OBJECTCLASS_VIOLATION) { /* * The entry or it's new parent has not been found * reconstruct the operation with the DN we just built */ ModifyDNMsg modifyDnMsg = (ModifyDNMsg) msg; modifyDnMsg.setDN(currentDN); modifyDnMsg.setNewSuperior(newSuperior.toString()); numResolvedNamingConflicts.incrementAndGet(); return ConflictResolution.REPLAY_AGAIN; } else if (result == ResultCode.ENTRY_ALREADY_EXISTS) { /* * This may happen when two modifyDn operation * are done on different servers but with the same target DN * add the conflict object class to the entry * and rename it using its entryuuid. */ ModifyDNMsg modifyDnMsg = (ModifyDNMsg) msg; markConflictEntry(op, op.getEntryDN(), newDN); modifyDnMsg.setNewRDN(generateConflictRDN(entryUUID, modifyDnMsg.getNewRDN())); modifyDnMsg.setNewSuperior(newSuperior.toString()); numUnresolvedNamingConflicts.incrementAndGet(); return ConflictResolution.REPLAY_AGAIN; } else { // The other type of errors can not be caused by naming conflicts: // the operation simply failed, replay() reports it. return ConflictResolution.FAILED; } } /** * Solve a conflict detected when replaying a ADD operation. * * @param op The operation that triggered the conflict detection. * @param msg The message that triggered the conflict detection. * @return the outcome of the conflict resolution * @throws Exception When the operation is not valid. */ private ConflictResolution solveNamingConflict(AddOperation op, AddMsg msg) throws Exception { ResultCode result = op.getResultCode(); AddContext ctx = (AddContext) op.getAttachment(SYNCHROCONTEXT); String entryUUID = ctx.getEntryUUID(); String parentUniqueId = ctx.getParentEntryUUID(); if (result == ResultCode.NO_SUCH_OBJECT) { /* * This can happen if the parent has been renamed or deleted * find the parent dn and calculate a new dn for the entry */ if (parentUniqueId == null) { /* * This entry is the base dn of the backend. * It is quite surprising that the operation result be NO_SUCH_OBJECT. * There is nothing more we can do except log a * message for the repair tool to look at this problem. * TODO : Log the message */ return ConflictResolution.NOTHING_TO_DO; } DN parentDn = findEntryDN(parentUniqueId); if (parentDn == null) { /* * The parent has been deleted * rename the entry as a conflicting entry. * The action taken here must be consistent with the actions * done when in the solveNamingConflict(DeleteOperation) method * when we are deleting an entry that have some child entries. */ addConflict(msg); String conflictRDN = generateConflictRDN(entryUUID, op.getEntryDN().rdn().toString()); msg.setDN(DN.valueOf(conflictRDN + "," + getBaseDN())); // reset the parent entryUUID so that the check done is the // handleConflict phase does not fail. msg.setParentEntryUUID(null); numUnresolvedNamingConflicts.incrementAndGet(); } else { msg.setDN(DN.valueOf(msg.getDN().rdn() + "," + parentDn)); numResolvedNamingConflicts.incrementAndGet(); } return ConflictResolution.REPLAY_AGAIN; } else if (result == ResultCode.ENTRY_ALREADY_EXISTS) { /* * This can happen if * - two adds are done on different servers but with the * same target DN. * - the same ADD is being replayed for the second time on this server. * if the entryUUID already exist, assume this is a replay and * don't do anything * if the entry unique id do not exist, generate conflict. */ if (findEntryDN(entryUUID) != null) { // entry already exist : this is a replay return ConflictResolution.NOTHING_TO_DO; } else { addConflict(msg); String conflictRDN = generateConflictRDN(entryUUID, msg.getDN().toString()); msg.setDN(DN.valueOf(conflictRDN)); numUnresolvedNamingConflicts.incrementAndGet(); return ConflictResolution.REPLAY_AGAIN; } } else { // The other type of errors can not be caused by naming conflicts: // the operation simply failed, replay() reports it. return ConflictResolution.FAILED; } } /** * Find all the entries below the provided DN and rename them * so that they stay below the baseDN of this replicationDomain and * use the conflicting name and attribute. * * @param entryDN The DN of the entry whose child must be renamed. * @param conflictOp The Operation that generated the conflict. */ private boolean findAndRenameChild(DN entryDN, Operation conflictOp) { /* * TODO JNR Ludo thinks that: "Ideally, the operation should verify that the * entryUUID has not changed or try to use the entryUUID rather than the * DN.". entryUUID can be obtained from the caller of the current method. */ boolean conflict = false; // Find and rename child entries. final SearchRequest request = newSearchRequest(entryDN, SearchScope.SINGLE_LEVEL) .addAttribute(ENTRYUUID_ATTRIBUTE_NAME, HISTORICAL_ATTRIBUTE_NAME); InternalSearchOperation op = conn.processSearch(request); if (op.getResultCode() == ResultCode.SUCCESS) { for (SearchResultEntry entry : op.getSearchEntries()) { /* * Check the ADD and ModRDN date of the child entry * (All of them, not only the one that are newer than the DEL op) * and keep the entry as a conflicting entry. */ conflict = true; renameConflictEntry(conflictOp, entry.getName(), getEntryUUID(entry)); } } else { // log error and information for the REPAIR tool. logger.error(ERR_CANNOT_RENAME_CONFLICT_ENTRY, entryDN, conflictOp, op.getResultCode()); } return conflict; } /** * Rename an entry that was conflicting so that it stays below the * baseDN of the replicationDomain. * * @param conflictOp The Operation that caused the conflict. * @param dn The DN of the entry to be renamed. * @param entryUUID The uniqueID of the entry to be renamed. */ private void renameConflictEntry(Operation conflictOp, DN dn, String entryUUID) { LocalizableMessage alertMessage = NOTE_UNRESOLVED_CONFLICT.get(dn); DirectoryServer.sendAlertNotification(this, ALERT_TYPE_REPLICATION_UNRESOLVED_CONFLICT, alertMessage); RDN newRDN = generateDeleteConflictDn(entryUUID, dn); ModifyDNOperation newOp = renameEntry(dn, newRDN, getBaseDN(), true); if (newOp.getResultCode() != ResultCode.SUCCESS) { // log information for the repair tool. logger.error(ERR_CANNOT_RENAME_CONFLICT_ENTRY, dn, conflictOp, newOp.getResultCode()); } } /** * Generate a modification to add the conflict attribute to an entry * whose Dn is now conflicting with another entry. * * @param op The operation causing the conflict. * @param currentDN The current DN of the operation to mark as conflicting. * @param conflictDN The newDn on which the conflict happened. */ private void markConflictEntry(Operation op, DN currentDN, DN conflictDN) { // create new internal modify operation and run it. Attribute attr = Attributes.create(DS_SYNC_CONFLICT, conflictDN.toString()); List mods = newArrayList(new Modification(ModificationType.REPLACE, attr)); ModifyOperation newOp = new ModifyOperationBasis( conn, nextOperationID(), nextMessageID(), new ArrayList(0), currentDN, mods); runAsSynchronizedOperation(newOp); if (newOp.getResultCode() != ResultCode.SUCCESS) { // Log information for the repair tool. logger.error(ERR_CANNOT_ADD_CONFLICT_ATTRIBUTE, op, newOp.getResultCode()); } // Generate an alert to let the administration know that some // conflict could not be solved. LocalizableMessage alertMessage = NOTE_UNRESOLVED_CONFLICT.get(conflictDN); DirectoryServer.sendAlertNotification(this, ALERT_TYPE_REPLICATION_UNRESOLVED_CONFLICT, alertMessage); } /** * Add the conflict attribute to an entry that could * not be added because it is conflicting with another entry. * * @param msg The conflicting Add Operation. * * @throws DecodeException When an encoding error happened manipulating the * msg. */ private void addConflict(AddMsg msg) throws DecodeException { String normalizedDN = msg.getDN().toString(); // Generate an alert to let the administrator know that some // conflict could not be solved. LocalizableMessage alertMessage = NOTE_UNRESOLVED_CONFLICT.get(normalizedDN); DirectoryServer.sendAlertNotification(this, ALERT_TYPE_REPLICATION_UNRESOLVED_CONFLICT, alertMessage); // Add the conflict attribute msg.addAttribute(DS_SYNC_CONFLICT, normalizedDN); } /** * Generate the Dn to use for a conflicting entry. * * @param entryUUID The unique identifier of the entry involved in the * conflict. * @param rdn Original rdn. * @return The generated RDN for a conflicting entry. */ private String generateConflictRDN(String entryUUID, String rdn) { return "entryuuid=" + entryUUID + "+" + rdn; } /** * Generate the RDN to use for a conflicting entry whose father was deleted. * * @param entryUUID The unique identifier of the entry involved in the * conflict. * @param dn The original DN of the entry. * * @return The generated RDN for a conflicting entry. */ private RDN generateDeleteConflictDn(String entryUUID, DN dn) { String newRDN = "entryuuid=" + entryUUID + "+" + dn.rdn(); try { return RDN.valueOf(newRDN); } catch (LocalizedIllegalArgumentException e) { // cannot happen return null; } } /** * Check if the domain solve conflicts. * * @return a boolean indicating if the domain should solve conflicts. */ boolean solveConflict() { return solveConflictFlag; } /** * Disable the replication on this domain. * The session to the replication server will be stopped. * The domain will not be destroyed but call to the pre-operation * methods will result in failure. * The listener thread will be destroyed. * The monitor informations will still be accessible. */ public void disable() { synchronized (serviceStateLock) { state.save(); state.clearInMemory(); disabled = true; disableService(); // This will cut the session and wake up the listener sessionGeneration++; /* * The ServerState this bookkeeping goes with is now gone from memory and is loaded * again from the backend when the domain is enabled back, so the changes listed as * pending must not outlive it: one which stayed would be discarded as a duplicate * when the replication server sends it again, and nothing would ever replay it or * record it in the ServerState. The listener thread is stopped first, or a change * it lists after this would be the one left behind. */ remotePendingChanges.clear(); /* * The recovery from a failed replay is over as well: the change it was asking for * is gone with the pending changes, so a leftover request would have a replay thread * stop and start the session once for a delivery which can not come. */ sessionRestartRequested.set(false); consecutiveSessionRestarts.set(0); } } /** * Do what necessary when the data have changed : load state, load * generation Id. * If there is no such information check if there is a * ReplicaUpdateVector entry and translate it into a state * and generationId. * @exception DirectoryException Thrown when an error occurs. */ private void loadDataState() throws DirectoryException { state.clearInMemory(); state.loadState(); getGenerator().adjust(state.getMaxCSN(getServerId())); // Retrieves the generation ID associated with the data imported generationId = loadGenerationId(); } /** * Enable back the domain after a previous disable. * The domain will connect back to a replication Server and * will recreate threads to listen for messages from the Synchronization * server. * The generationId will be retrieved or computed if necessary. * The ServerState will also be read again from the local database. */ public void enable() { synchronized (serviceStateLock) { try { loadDataState(); } catch (Exception e) { /* TODO should mark that replicationServer service is * not available, log an error and retry upon timeout * should we stop the modifications ? */ logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), stackTraceToSingleLineString(e)); return; } enableService(); sessionGeneration++; disabled = false; } } /** * Compute the data generationId associated with the current data present * in the backend for this domain. * @return The computed generationId. * @throws DirectoryException When an error occurs. */ private long computeGenerationId() throws DirectoryException { final long genId = exportBackend(null, true); if (logger.isTraceEnabled()) { logger.trace("Computed generationId: generationId=" + genId); } return genId; } /** * Run a modify operation to update the entry whose DN is given as * a parameter with the generationID information. * * @param entryDN The DN of the entry to be updated. * @param generationId The value of the generationID to be saved. * * @return A ResultCode indicating if the operation was successful. */ private ResultCode runSaveGenerationId(DN entryDN, long generationId) { // The generationId is stored in the root entry of the domain. final ByteString asn1BaseDn = ByteString.valueOfUtf8(entryDN.toString()); LDAPAttribute attr = new LDAPAttribute(REPLICATION_GENERATION_ID, Long.toString(generationId)); List mods = new ArrayList<>(1); mods.add(new LDAPModification(ModificationType.REPLACE, attr)); ModifyOperation op = new ModifyOperationBasis( conn, nextOperationID(), nextMessageID(), new ArrayList(0), asn1BaseDn, mods); runAsSynchronizedOperation(op); return op.getResultCode(); } /** * Stores the value of the generationId. * @param generationId The value of the generationId. * @return a ResultCode indicating if the method was successful. */ private ResultCode saveGenerationId(long generationId) { ResultCode result = runSaveGenerationId(getBaseDN(), generationId); if (result != ResultCode.SUCCESS) { generationIdSavedStatus = false; if (result == ResultCode.NO_SUCH_OBJECT) { // If the base entry does not exist, save the generation // ID in the config entry result = runSaveGenerationId(config.dn(), generationId); } if (result != ResultCode.SUCCESS) { logger.error(ERR_UPDATING_GENERATION_ID, getBaseDN(), result.getName()); } } else { generationIdSavedStatus = true; } return result; } /** * Load the GenerationId from the root entry of the domain * from the REPLICATION_GENERATION_ID attribute in database * to memory, or compute it if not found. * * @return generationId The retrieved value of generationId * @throws DirectoryException When an error occurs. */ private long loadGenerationId() throws DirectoryException { if (logger.isTraceEnabled()) { logger.trace("Attempt to read generation ID from DB " + getBaseDN()); } // Search the database entry that is used to periodically save the generation id final SearchRequest request = newSearchRequest(getBaseDN(), SearchScope.BASE_OBJECT) .addAttribute(REPLICATION_GENERATION_ID); InternalSearchOperation search = conn.processSearch(request); if (search.getResultCode() == ResultCode.NO_SUCH_OBJECT) { // if the base entry does not exist look for the generationID // in the config entry. request.setName(config.dn()); search = conn.processSearch(request); } boolean found = false; long aGenerationId = -1; if (search.getResultCode() != ResultCode.SUCCESS) { if (search.getResultCode() != ResultCode.NO_SUCH_OBJECT) { String errorMsg = search.getResultCode().getName() + " " + search.getErrorMessage(); logger.error(ERR_SEARCHING_GENERATION_ID, getBaseDN(), errorMsg); } } else { List result = search.getSearchEntries(); SearchResultEntry resultEntry = result.get(0); if (resultEntry != null) { Iterator attrs = resultEntry.getAllAttributes(REPLICATION_GENERATION_ID).iterator(); if (attrs.hasNext()) { Attribute attr = attrs.next(); if (attr.size()>1) { String errorMsg = "#Values=" + attr.size() + " Must be exactly 1 in entry " + resultEntry.toLDIFString(); logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), errorMsg); } else if (attr.size() == 1) { found = true; try { aGenerationId = Long.decode(attr.iterator().next().toString()); } catch(Exception e) { logger.error(ERR_LOADING_GENERATION_ID, getBaseDN(), stackTraceToSingleLineString(e)); } } } } } if (!found) { aGenerationId = computeGenerationId(); saveGenerationId(aGenerationId); if (logger.isTraceEnabled()) { logger.trace("Generation ID created for domain baseDN=" + getBaseDN() + " generationId=" + aGenerationId); } } else { generationIdSavedStatus = true; if (logger.isTraceEnabled()) { logger.trace("Generation ID successfully read from domain baseDN=" + getBaseDN() + " generationId=" + aGenerationId); } } return aGenerationId; } /** * Do whatever is needed when a backup is started. * We need to make sure that the serverState is correctly save. */ void backupStart() { state.save(); } /** Do whatever is needed when a backup is finished. */ void backupEnd() { // Nothing is needed at the moment } /* * Total Update >> */ /** * This method trigger an export of the replicated data. * * @param output The OutputStream where the export should * be produced. * @throws DirectoryException When needed. */ @Override protected void exportBackend(OutputStream output) throws DirectoryException { exportBackend(output, false); } /** * Export the entries from the backend and/or compute the generation ID. * The ieContext must have been set before calling. * * @param output The OutputStream where the export should * be produced. * @param checksumOutput A boolean indicating if this export is * invoked to perform a checksum only * * @return The computed GenerationID. * * @throws DirectoryException when an error occurred */ private long exportBackend(OutputStream output, boolean checksumOutput) throws DirectoryException { LocalBackend backend = getBackend(); // Acquire a shared lock for the backend. try { String lockFile = LockFileManager.getBackendLockFileName(backend); StringBuilder failureReason = new StringBuilder(); if (! LockFileManager.acquireSharedLock(lockFile, failureReason)) { LocalizableMessage message = ERR_LDIFEXPORT_CANNOT_LOCK_BACKEND.get(backend.getBackendID(), failureReason); logger.error(message); throw new DirectoryException(ResultCode.OTHER, message); } } catch (Exception e) { LocalizableMessage message = ERR_LDIFEXPORT_CANNOT_LOCK_BACKEND.get(backend.getBackendID(), stackTraceToSingleLineString(e)); logger.error(message); throw new DirectoryException(ResultCode.OTHER, message); } long numberOfEntries = backend.getNumberOfEntriesInBaseDN(getBaseDN()); long entryCount = Math.min(numberOfEntries, 1000); OutputStream os; ReplLDIFOutputStream ros = null; if (checksumOutput) { ros = new ReplLDIFOutputStream(entryCount); os = ros; try { os.write(Long.toString(numberOfEntries).getBytes()); } catch(Exception e) { // Should never happen } } else { os = output; } // baseDN branch is the only one included in the export LDIFExportConfig exportConfig = new LDIFExportConfig(os); exportConfig.setIncludeBranches(newArrayList(getBaseDN())); // For the checksum computing mode, only consider the 'stable' attributes if (checksumOutput) { String includeAttributeStrings[] = { "objectclass", "sn", "cn", "entryuuid" }; Set includeAttributes = new HashSet<>(); for (String attrName : includeAttributeStrings) { includeAttributes.add(getServerContext().getSchema().getAttributeType(attrName)); } exportConfig.setIncludeAttributes(includeAttributes); } // Launch the export. long genID = 0; try { backend.exportLDIF(exportConfig); } catch (DirectoryException de) { if (ros == null || ros.getNumExportedEntries() < entryCount) { LocalizableMessage message = ERR_LDIFEXPORT_ERROR_DURING_EXPORT.get(de.getMessageObject()); logger.error(message); throw new DirectoryException(ResultCode.OTHER, message); } } catch (Exception e) { LocalizableMessage message = ERR_LDIFEXPORT_ERROR_DURING_EXPORT.get(stackTraceToSingleLineString(e)); logger.error(message); throw new DirectoryException(ResultCode.OTHER, message); } finally { // Clean up after the export by closing the export config. // Will also flush the export and export the remaining entries. exportConfig.close(); if (checksumOutput) { genID = ros.getChecksumValue(); } // Release the shared lock on the backend. try { String lockFile = LockFileManager.getBackendLockFileName(backend); StringBuilder failureReason = new StringBuilder(); if (! LockFileManager.releaseLock(lockFile, failureReason)) { LocalizableMessage message = WARN_LDIFEXPORT_CANNOT_UNLOCK_BACKEND.get(backend.getBackendID(), failureReason); logger.warn(message); throw new DirectoryException(ResultCode.OTHER, message); } } catch (Exception e) { LocalizableMessage message = WARN_LDIFEXPORT_CANNOT_UNLOCK_BACKEND.get(backend.getBackendID(), stackTraceToSingleLineString(e)); logger.warn(message); throw new DirectoryException(ResultCode.OTHER, message); } } return genID; } /** * Process backend before import. * * @param backend * The backend. * @throws DirectoryException * If the backend could not be disabled or locked exclusively. */ private void preBackendImport(LocalBackend backend) throws DirectoryException { // Prevent the processing of the backend finalisation event as the import will disable the attached backend ignoreBackendInitializationEvent = true; // FIXME setBackendEnabled should be part of TaskUtils ? TaskUtils.disableBackend(backend.getBackendID()); // Acquire an exclusive lock for the backend. String lockFile = LockFileManager.getBackendLockFileName(backend); StringBuilder failureReason = new StringBuilder(); if (! LockFileManager.acquireExclusiveLock(lockFile, failureReason)) { LocalizableMessage message = ERR_INIT_CANNOT_LOCK_BACKEND.get(backend.getBackendID(), failureReason); logger.error(message); throw new DirectoryException(ResultCode.OTHER, message); } } /** * This method triggers an import of the replicated data. * * @param input The InputStream from which the data are read. * @throws DirectoryException When needed. */ @Override protected void importBackend(InputStream input) throws DirectoryException { LocalBackend backend = getBackend(); LDIFImportConfig importConfig = null; ImportExportContext ieCtx = getImportExportContext(); try { if (!backend.supports(BackendOperation.LDIF_IMPORT)) { ieCtx.setExceptionIfNoneSet(new DirectoryException(OTHER, ERR_INIT_IMPORT_NOT_SUPPORTED.get(backend.getBackendID()))); return; } importConfig = new LDIFImportConfig(input); importConfig.setIncludeBranches(newLinkedHashSet(getBaseDN())); // We should not validate schema for replication importConfig.setValidateSchema(false); // Allow fractional replication ldif import plugin to be called importConfig.setInvokeImportPlugins(true); // Reset the follow import flag and message before starting the import importErrorMessageId = -1; // TODO How to deal with rejected entries during the import File rejectsFile = getFileForPath("logs" + File.separator + "replInitRejectedEntries"); importConfig.writeRejectedEntries(rejectsFile.getAbsolutePath(), ExistingFileBehavior.OVERWRITE); // Process import preBackendImport(backend); backend.importLDIF(importConfig, getServerContext()); } catch(Exception e) { ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER, ERR_INIT_IMPORT_FAILURE.get(stackTraceToSingleLineString(e)))); } finally { try { // Cleanup if (importConfig != null) { importConfig.close(); closeBackendImport(backend); // Re-enable backend backend = getBackend(); } loadDataState(); if (ieCtx.getException() != null) { // When an error occurred during an import, most of times // the generationId coming in the root entry of the imported data, // is not valid anymore (partial data in the backend). generationId = computeGenerationId(); saveGenerationId(generationId); } } catch (DirectoryException fe) { // If we already catch an Exception it's quite possible // that the loadDataState() and setGenerationId() fail // so we don't bother about the new Exception. // However if there was no Exception before we want // to return this Exception to the task creator. ieCtx.setExceptionIfNoneSet(new DirectoryException( ResultCode.OTHER, ERR_INIT_IMPORT_FAILURE.get(stackTraceToSingleLineString(fe)))); } } if (ieCtx.getException() != null) { throw ieCtx.getException(); } } /** * Make post import operations. * @param backend The backend implied in the import. * @exception DirectoryException Thrown when an error occurs. */ private void closeBackendImport(LocalBackend backend) throws DirectoryException { String lockFile = LockFileManager.getBackendLockFileName(backend); StringBuilder failureReason = new StringBuilder(); // Release lock if (!LockFileManager.releaseLock(lockFile, failureReason)) { LocalizableMessage message = WARN_LDIFIMPORT_CANNOT_UNLOCK_BACKEND.get(backend.getBackendID(), failureReason); logger.warn(message); throw new DirectoryException(ResultCode.OTHER, message); } TaskUtils.enableBackend(backend.getBackendID()); // Restore the processing of backend finalization events. ignoreBackendInitializationEvent = false; } /** * Retrieves a replication domain based on the baseDN. * * @param baseDN The baseDN of the domain to retrieve * @return The domain retrieved * @throws DirectoryException When an error occurred or no domain * match the provided baseDN. */ public static LDAPReplicationDomain retrievesReplicationDomain(DN baseDN) throws DirectoryException { LDAPReplicationDomain replicationDomain = null; // Retrieves the domain for (SynchronizationProvider provider : DirectoryServer.getSynchronizationProviders()) { if (!(provider instanceof MultimasterReplication)) { LocalizableMessage message = ERR_INVALID_PROVIDER.get(); throw new DirectoryException(ResultCode.OTHER, message); } // From the domainDN retrieves the replication domain LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null); if (domain == null) { break; } if (replicationDomain != null) { // Should never happen LocalizableMessage message = ERR_MULTIPLE_MATCHING_DOMAIN.get(); throw new DirectoryException(ResultCode.OTHER, message); } replicationDomain = domain; } if (replicationDomain == null) { throw new DirectoryException(ResultCode.OTHER, ERR_NO_MATCHING_DOMAIN.get(baseDN)); } return replicationDomain; } /** * Returns the backend associated to this domain. * @return The associated backend. */ private LocalBackend getBackend() { return getServerContext().getBackendConfigManager() .findLocalBackendForEntry(getBaseDN()); } /* * < modifications) { ModifyOperation op = new ModifyOperationBasis( conn, nextOperationID(), nextMessageID(), null, DirectoryServer.getSchemaDN(), modifications); final Entry schema; try { schema = DirectoryServer.getEntry(DirectoryServer.getSchemaDN()); } catch (DirectoryException e) { logger.traceException(e); logger.error(ERR_BACKEND_SEARCH_ENTRY.get(DirectoryServer.getSchemaDN().toString(), stackTraceToSingleLineString(e))); return; } LocalBackendModifyOperation localOp = new LocalBackendModifyOperation(op); CSN csn = generateCSN(localOp); OperationContext ctx = new ModifyContext(csn, getEntryUUID(schema)); localOp.setAttachment(SYNCHROCONTEXT, ctx); localOp.setResultCode(ResultCode.SUCCESS); synchronize(localOp); } /** * Check if the provided configuration is acceptable for add. * * @param configuration The configuration to check. * @param unacceptableReasons When the configuration is not acceptable, this * table is use to return the reasons why this * configuration is not acceptable. * * @return true if the configuration is acceptable, false other wise. */ static boolean isConfigurationAcceptable(ReplicationDomainCfg configuration, List unacceptableReasons) { // Check that there is not already a domain with the same DN final DN dn = configuration.getBaseDN(); LDAPReplicationDomain domain = MultimasterReplication.findDomain(dn, null); if (domain != null && domain.getBaseDN().equals(dn)) { unacceptableReasons.add(ERR_SYNC_INVALID_DN.get()); return false; } // Check that the base DN is configured as a base-dn of the directory server if (getServerContext().getBackendConfigManager().findLocalBackendForEntry(dn) == null) { unacceptableReasons.add(ERR_UNKNOWN_DN.get(dn)); return false; } // Check fractional configuration try { isFractionalConfigAcceptable(configuration); } catch (ConfigException e) { unacceptableReasons.add(e.getMessageObject()); return false; } return true; } @Override public ConfigChangeResult applyConfigurationChange( ReplicationDomainCfg configuration) { this.config = configuration; /* * Each of these stops and starts the session when what it changes calls for it, and * the configuration they change is read as the session comes up: hold the lock the * replay thread restarting the session after a failed replay takes, so that none of * them is interleaved with a session it did not start itself. */ synchronized (serviceStateLock) { changeConfig(configuration); // Read assured + fractional configuration and each time reconnect if needed readAssuredConfig(configuration, true); readFractionalConfig(configuration, true); } solveConflictFlag = isSolveConflict(configuration); final ConfigChangeResult ccr = new ConfigChangeResult(); try { storeECLConfiguration(configuration); } catch(Exception e) { ccr.setResultCode(ResultCode.OTHER); } return ccr; } @Override public boolean isConfigurationChangeAcceptable( ReplicationDomainCfg configuration, List unacceptableReasons) { // Check that a import/export is not in progress if (ieRunning()) { unacceptableReasons.add( NOTE_ERR_CANNOT_CHANGE_CONFIG_DURING_TOTAL_UPDATE.get()); return false; } // Check fractional configuration try { isFractionalConfigAcceptable(configuration); return true; } catch (ConfigException e) { unacceptableReasons.add(e.getMessageObject()); return false; } } @Override public Map getAlerts() { Map alerts = new LinkedHashMap<>(); alerts.put(ALERT_TYPE_REPLICATION_UNRESOLVED_CONFLICT, ALERT_DESCRIPTION_REPLICATION_UNRESOLVED_CONFLICT); alerts.put(ALERT_TYPE_REPLICATION_UNREPLAYED_CHANGE, ALERT_DESCRIPTION_REPLICATION_UNREPLAYED_CHANGE); return alerts; } @Override public String getClassName() { return CLASS_NAME; } @Override public DN getComponentEntryDN() { return config.dn(); } /** Starts the Replication Domain. */ public void start() { // Create the ServerStateFlush thread flushThread.start(); startListenService(); } /** Remove the configuration of the external changelog from this domain configuration. */ private void removeECLDomainCfg() { try { DN eclConfigEntryDN = DN.valueOf("cn=external changeLog," + config.dn()); if (getServerContext().getConfigurationHandler().hasEntry(eclConfigEntryDN)) { getServerContext().getConfigurationHandler().deleteEntry(eclConfigEntryDN); } } catch(Exception e) { logger.traceException(e); logger.error(ERR_CHECK_CREATE_REPL_BACKEND_FAILED, stackTraceToSingleLineString(e)); } } /** * Store the provided ECL configuration for the domain. * @param domCfg The provided configuration. * @throws ConfigException When an error occurred. */ private void storeECLConfiguration(ReplicationDomainCfg domCfg) throws ConfigException { ExternalChangelogDomainCfg eclDomCfg = null; // create the ecl config if it does not exist // There may not be any config entry related to this domain in some // unit test cases try { DN configDn = config.dn(); ConfigurationHandler configHandler = getServerContext().getConfigurationHandler(); if (configHandler.hasEntry(config.dn())) { try { eclDomCfg = domCfg.getExternalChangelogDomain(); } catch(Exception e) { /* do nothing */ } // domain with no config entry only when running unit tests if (eclDomCfg == null) { // no ECL config provided hence create a default one // create the default one DN eclConfigEntryDN = DN.valueOf("cn=external changelog," + configDn); if (!configHandler.hasEntry(eclConfigEntryDN)) { // no entry exist yet for the ECL config for this domain // create it String ldif = makeLdif( "dn: cn=external changelog," + configDn, "objectClass: top", "objectClass: ds-cfg-external-changelog-domain", "cn: external changelog", "ds-cfg-enabled: " + !getBackend().isPrivateBackend()); LDIFImportConfig ldifImportConfig = new LDIFImportConfig( new StringReader(ldif)); // No need to validate schema in replication ldifImportConfig.setValidateSchema(false); LDIFReader reader = new LDIFReader(ldifImportConfig); Entry eclEntry = reader.readEntry(); configHandler.addEntry(Converters.from(eclEntry)); ldifImportConfig.close(); } } } eclDomCfg = domCfg.getExternalChangelogDomain(); if (eclDomain != null) { eclDomain.applyConfigurationChange(eclDomCfg); } else { // Create the ECL domain object eclDomain = new ExternalChangelogDomain(this, eclDomCfg); } } catch (Exception e) { throw new ConfigException(NOTE_ERR_UNABLE_TO_ENABLE_ECL.get( "Replication Domain on " + getBaseDN(), stackTraceToSingleLineString(e)), e); } } private static String makeLdif(String... lines) { final StringBuilder buffer = new StringBuilder(); for (String line : lines) { buffer.append(line).append(EOL); } // Append an extra line so we can append LDIF Strings. buffer.append(EOL); return buffer.toString(); } @Override public void sessionInitiated(ServerStatus initStatus, ServerState rsState) { // Check domain fractional configuration consistency with local // configuration variables forceBadDataSet = !isBackendFractionalConfigConsistent(); super.sessionInitiated(initStatus, rsState); // Now for bad data set status if needed if (forceBadDataSet) { signalNewStatus(StatusMachineEvent.TO_BAD_GEN_ID_STATUS_EVENT); logger.info(NOTE_FRACTIONAL_BAD_DATA_SET_NEED_RESYNC, getBaseDN()); return; // Do not send changes to the replication server } try { /* * We must not publish changes to a replicationServer that has * not seen all our previous changes because this could cause * some other ldap servers to miss those changes. * Check that the ReplicationServer has seen all our previous * changes. */ CSN replServerMaxCSN = rsState.getCSN(getServerId()); // we don't want to update from here (a DS) an empty RS because // normally the RS should have been updated by other RSes except for // very last changes lost if the local connection was broken // ... hence the RS we are connected to should not be empty // ... or if it is empty, it is due to a voluntary reset // and we don't want to update it with our changes that could be huge. if (replServerMaxCSN != null && replServerMaxCSN.getSeqnum() != 0) { CSN ourMaxCSN = state.getMaxCSN(getServerId()); if (ourMaxCSN != null && !ourMaxCSN.isOlderThanOrEqualTo(replServerMaxCSN)) { pendingChanges.setRecovering(true); broker.setRecoveryRequired(true); final RSUpdater rsUpdater = new RSUpdater(replServerMaxCSN); if (this.rsUpdater.compareAndSet(null, rsUpdater)) { rsUpdater.start(); } } } } catch (Exception e) { logger.error(ERR_PUBLISHING_FAKE_OPS, getBaseDN(), stackTraceToSingleLineString(e)); } } /** * Build the list of changes that have been processed by this server after the * CSN given as a parameter and publish them using the given session. * * @param startCSN * The CSN where we need to start the search * @param session * The session to use to publish the changes * @return A boolean indicating he success of the operation. * @throws Exception * if an Exception happens during the search. */ boolean buildAndPublishMissingChanges(CSN startCSN, ReplicationBroker session) throws Exception { // Trim the changes in replayOperations that are older than the startCSN. synchronized (replayOperations) { Iterator it = replayOperations.keySet().iterator(); while (it.hasNext()) { if (shutdown.get()) { return false; } if (it.next().isNewerThan(startCSN)) { break; } it.remove(); } } CSN lastRetrievedChange; InternalSearchOperation op; CSN currentStartCSN = startCSN; do { if (shutdown.get()) { return false; } lastRetrievedChange = null; // We can't do the search in one go because we need to store the results // so that we are sure we send the operations in order and because the // list might be large. // So we search by interval of 10 seconds and store the results in the // replayOperations list so that they are sorted before sending them. long missingChangesDelta = currentStartCSN.getTime() + 10000; CSN endCSN = new CSN(missingChangesDelta, 0xffffffff, getServerId()); ScanSearchListener listener = new ScanSearchListener(currentStartCSN, endCSN); op = searchForChangedEntries(getBaseDN(), currentStartCSN, endCSN, listener); // Publish and remove all the changes from the replayOperations list // that are older than the endCSN. final List opsToSend = new LinkedList<>(); synchronized (replayOperations) { Iterator itOp = replayOperations.values().iterator(); while (itOp.hasNext()) { if (shutdown.get()) { return false; } FakeOperation fakeOp = itOp.next(); if (fakeOp.getCSN().isNewerThan(endCSN) // sanity check || !state.cover(fakeOp.getCSN()) // do not look for replay operations in the future || currentStartCSN.isNewerThan(now())) { break; } lastRetrievedChange = fakeOp.getCSN(); opsToSend.add(fakeOp); itOp.remove(); } } for (FakeOperation opToSend : opsToSend) { if (shutdown.get()) { return false; } session.publishRecovery(opToSend.generateMessage()); } if (lastRetrievedChange != null) { if (logger.isDebugEnabled()) { logger.debug(LocalizableMessage.raw("publish loop" + " >=" + currentStartCSN + " <=" + endCSN + " nentries=" + op.getEntriesSent() + " result=" + op.getResultCode() + " lastRetrievedChange=" + lastRetrievedChange)); } currentStartCSN = lastRetrievedChange; } else { if (logger.isDebugEnabled()) { logger.debug(LocalizableMessage.raw("publish loop" + " >=" + currentStartCSN + " <=" + endCSN + " nentries=" + op.getEntriesSent() + " result=" + op.getResultCode() + " no changes")); } currentStartCSN = endCSN; } } while (pendingChanges.recoveryUntil(currentStartCSN) && op.getResultCode().equals(ResultCode.SUCCESS)); return op.getResultCode().equals(ResultCode.SUCCESS); } private static CSN now() { // ensure now() will always come last with isNewerThan() test, // even though the timestamp, or the timestamp and seqnum would be the same return new CSN(TimeThread.getTime(), Integer.MAX_VALUE, Integer.MAX_VALUE); } /** * Search for the changes that happened since fromCSN based on the historical * attribute. The only changes that will be send will be the one generated on * the serverId provided in fromCSN. * * @param baseDN * the base DN * @param fromCSN * The CSN from which we want the changes * @param lastCSN * The max CSN that the search should return * @param resultListener * The listener that will process the entries returned * @return the internal search operation * @throws Exception * when raised. */ private static InternalSearchOperation searchForChangedEntries(DN baseDN, CSN fromCSN, CSN lastCSN, InternalSearchListener resultListener) throws Exception { String maxValueForId; if (lastCSN == null) { final Integer serverId = fromCSN.getServerId(); maxValueForId = "ffffffffffffffff" + String.format("%04x", serverId) + "ffffffff"; } else { maxValueForId = lastCSN.toString(); } String filter = "(&(" + HISTORICAL_ATTRIBUTE_NAME + ">=dummy:" + fromCSN + ")" + "(" + HISTORICAL_ATTRIBUTE_NAME + "<=dummy:" + maxValueForId + "))"; SearchRequest request = Requests.newSearchRequest(baseDN, SearchScope.WHOLE_SUBTREE, filter) .addAttribute(USER_AND_REPL_OPERATIONAL_ATTRS); return getRootConnection().processSearch(request, resultListener); } /** * Search for the changes that happened since fromCSN based on the historical * attribute. The only changes that will be send will be the one generated on * the serverId provided in fromCSN. * * @param baseDN * the base DN * @param fromCSN * The CSN from which we want the changes * @param resultListener * that will process the entries returned. * @return the internal search operation * @throws Exception * when raised. */ static InternalSearchOperation searchForChangedEntries(DN baseDN, CSN fromCSN, InternalSearchListener resultListener) throws Exception { return searchForChangedEntries(baseDN, fromCSN, null, resultListener); } /** * This method should return the total number of objects in the * replicated domain. * This count will be used for reporting. * * @throws DirectoryException when needed. * * @return The number of objects in the replication domain. */ @Override public long countEntries() throws DirectoryException { LocalBackend backend = getBackend(); if (!backend.supports(BackendOperation.LDIF_EXPORT)) { LocalizableMessage msg = ERR_INIT_EXPORT_NOT_SUPPORTED.get(backend.getBackendID()); logger.error(msg); throw new DirectoryException(ResultCode.OTHER, msg); } return backend.getNumberOfEntriesInBaseDN(getBaseDN()); } @Override public boolean processUpdate(UpdateMsg updateMsg) { // Ignore message if fractional configuration is inconsistent and // we have been passed into bad data set status if (forceBadDataSet) { return false; } if (updateMsg instanceof LDAPUpdateMsg) { LDAPUpdateMsg msg = (LDAPUpdateMsg) updateMsg; // Put the UpdateMsg in the RemotePendingChanges list. if (!remotePendingChanges.putRemoteUpdate(msg)) { /* * A replay thread already owns this change, so this delivery is a duplicate: * ignore it. This happens when there are uncommitted changes in the queue and * session failover occurs causing a recovery of all changes since the current * committed server state. See OPENDJ-1115. * * The copy which is already listed owns the change: it is the one which records * it in the ServerState once it really has been replayed. Report this delivery * as done - the window and the ack are per delivery - but as handled * asynchronously, so that the listener does not push the CSN to the ServerState * over a change which is still being replayed or is failing (issue #889). * * A change whose replay failed is not owned by anyone anymore, so this is not the * path it takes: putRemoteUpdate() takes this delivery over from the one which * failed and has it replayed again. */ if (logger.isTraceEnabled()) { logger.trace( "LDAPReplicationDomain.processUpdate: ignoring " + "duplicate change %s", msg.getCSN()); } processUpdateDone(msg, null); return false; } // Put update message into the replay queue // (block until some place in the queue is available) final UpdateToReplay updateToReplay = new UpdateToReplay(msg, this); while (!isListenerShuttingDown()) { // loop until we can offer to the queue or shutdown was initiated try { if (updateToReplayQueue.offer(updateToReplay, 1, TimeUnit.SECONDS)) { // successful offer to the queue, let's exit the loop break; } } catch (InterruptedException e) { // Thread interrupted: check for shutdown. Thread.currentThread().interrupt(); } } return false; } // unknown message type, this should not happen, just ignore it. return true; } @Override public void addAdditionalMonitoring(MonitorData attributes) { attributes.add("pending-updates", pendingChanges.size()); attributes.add("replayed-updates-ok", numReplayedPostOpCalled); attributes.add("replayed-updates-failed", numFailedReplayedUpdates); attributes.add("resolved-modify-conflicts", numResolvedModifyConflicts); attributes.add("resolved-naming-conflicts", numResolvedNamingConflicts); attributes.add("unresolved-naming-conflicts", numUnresolvedNamingConflicts); attributes.add("remote-pending-changes-size", remotePendingChanges.getQueueSize()); attributes.add("dependent-changes-size", remotePendingChanges.getDependentChangesSize()); attributes.add("changes-in-progress-size", remotePendingChanges.changesInProgressSize()); } /** * Verifies that the given string represents a valid source * from which this server can be initialized. * @param sourceString The string representing the source * @return The source as a integer value * @throws DirectoryException if the string is not valid */ public int decodeSource(String sourceString) throws DirectoryException { int source = 0; try { source = Integer.decode(sourceString); if (source >= -1 && source != getServerId()) { // TODO Verifies serverID is in the domain // We should check here that this is a server implied // in the current domain. return source; } } catch (Exception e) { LocalizableMessage message = ERR_INVALID_IMPORT_SOURCE.get( getBaseDN(), getServerId(), sourceString, stackTraceToSingleLineString(e)); throw new DirectoryException(ResultCode.OTHER, message, e); } LocalizableMessage message = ERR_INVALID_IMPORT_SOURCE.get(getBaseDN(), getServerId(), source, ""); throw new DirectoryException(ResultCode.OTHER, message); } /** * Called by synchronize post op plugin in order to add the entry historical * attributes to the UpdateMsg. * @param msg an replication update message * @param op the operation in progress */ private void addEntryAttributesForCL(UpdateMsg msg, PostOperationOperation op) { if (op instanceof PostOperationDeleteOperation) { PostOperationDeleteOperation delOp = (PostOperationDeleteOperation) op; final Set names = getEclIncludesForDeletes(); Entry entry = delOp.getEntryToDelete(); final DeleteMsg deleteMsg = (DeleteMsg) msg; deleteMsg.setEclIncludes(getIncludedAttributes(entry, names)); // For delete only, add the Authorized DN since it's required in the // ECL entry but is not part of rest of the message. DN deleterDN = delOp.getAuthorizationDN(); if (deleterDN != null) { deleteMsg.setInitiatorsName(deleterDN.toString()); } } else if (op instanceof PostOperationModifyOperation) { PostOperationModifyOperation modOp = (PostOperationModifyOperation) op; Set names = getEclIncludes(); Entry entry = modOp.getCurrentEntry(); ((ModifyMsg) msg).setEclIncludes(getIncludedAttributes(entry, names)); } else if (op instanceof PostOperationModifyDNOperation) { PostOperationModifyDNOperation modDNOp = (PostOperationModifyDNOperation) op; Set names = getEclIncludes(); Entry entry = modDNOp.getOriginalEntry(); ((ModifyDNMsg) msg).setEclIncludes(getIncludedAttributes(entry, names)); } else if (op instanceof PostOperationAddOperation) { PostOperationAddOperation addOp = (PostOperationAddOperation) op; Set names = getEclIncludes(); Entry entry = addOp.getEntryToAdd(); ((AddMsg) msg).setEclIncludes(getIncludedAttributes(entry, names)); } } private Collection getIncludedAttributes(Entry entry, Set names) { if (names.isEmpty()) { // Fast-path. return Collections.emptySet(); } else if (names.size() == 1 && names.contains("*")) { // Potential fast-path for delete operations. List attributes = new LinkedList<>(); for (List attributeList : entry.getUserAttributes().values()) { attributes.addAll(attributeList); } Attribute objectClassAttribute = entry.getObjectClassAttribute(); if (objectClassAttribute != null) { attributes.add(objectClassAttribute); } return attributes; } else { // Expand @objectclass references in attribute list if needed. // We do this now in order to take into account dynamic schema changes. final Set expandedNames = getExpandedNames(names); final Entry filteredEntry = entry.filterEntry(expandedNames, false, false, false); return collect(filteredEntry.getAllAttributes(), new ArrayList()); } } private Set getExpandedNames(Set names) { // Only rebuild the attribute set if necessary. if (!needsExpanding(names)) { return names; } final Set expandedNames = new HashSet<>(names.size()); for (String name : names) { if (name.startsWith("@")) { String ocName = name.substring(1); ObjectClass objectClass = getServerContext().getSchema().getObjectClass(ocName); if (!objectClass.isPlaceHolder()) { for (AttributeType at : objectClass.getRequiredAttributes()) { expandedNames.add(at.getNameOrOID()); } for (AttributeType at : objectClass.getOptionalAttributes()) { expandedNames.add(at.getNameOrOID()); } } } else { expandedNames.add(name); } } return expandedNames; } private boolean needsExpanding(Set names) { for (String name : names) { if (name.startsWith("@")) { return true; } } return false; } /** * Gets the fractional configuration of this domain. * @return The fractional configuration of this domain. */ FractionalConfig getFractionalConfig() { return fractionalConfig; } /** * This bean is a utility class used for holding the parsing * result of a fractional configuration. It also contains some facility * methods like fractional configuration comparison... */ static class FractionalConfig { /** * Tells if fractional replication is enabled or not (some fractional * constraints have been put in place). If this is true then * fractionalExclusive explains the configuration mode and either * fractionalSpecificClassesAttributes or fractionalAllClassesAttributes or * both should be filled with something. */ private boolean fractional; /** * - If true, tells that the configured fractional replication is exclusive: * Every attributes contained in fractionalSpecificClassesAttributes and * fractionalAllClassesAttributes should be ignored when replaying operation * in local backend. * - If false, tells that the configured fractional replication is * inclusive: * Only attributes contained in fractionalSpecificClassesAttributes and * fractionalAllClassesAttributes should be taken into account in local * backend. */ private boolean fractionalExclusive = true; /** * Used in fractional replication: holds attributes of a specific object class. * - key = object class (name or OID of the class) * - value = the attributes of that class that should be taken into account * (inclusive or exclusive fractional replication) (name or OID of the * attribute) * When an operation coming from the network is to be locally replayed, if * the concerned entry has an objectClass attribute equals to 'key': * - inclusive mode: only the attributes in 'value' will be added/deleted/modified * - exclusive mode: the attributes in 'value' will not be added/deleted/modified */ private Map> fractionalSpecificClassesAttributes = new HashMap<>(); /** * Used in fractional replication: holds attributes of any object class. * When an operation coming from the network is to be locally replayed: * - inclusive mode: only attributes of the matching entry not present in * fractionalAllClassesAttributes will be added/deleted/modified * - exclusive mode: attributes of the matching entry present in * fractionalAllClassesAttributes will not be added/deleted/modified * The attributes may be in human readable form of OID form. */ private Set fractionalAllClassesAttributes = new HashSet<>(); /** Base DN the fractional configuration is for. */ private final DN baseDN; /** * Constructs a new fractional configuration object. * @param baseDN The base DN the object is for. */ private FractionalConfig(DN baseDN) { this.baseDN = baseDN; } /** * Getter for fractional. * @return True if the configuration has fractional enabled */ boolean isFractional() { return fractional; } /** * Set the fractional parameter. * @param fractional The fractional parameter */ private void setFractional(boolean fractional) { this.fractional = fractional; } /** * Getter for fractionalExclusive. * @return True if the configuration has fractional exclusive enabled */ boolean isFractionalExclusive() { return fractionalExclusive; } /** * Set the fractionalExclusive parameter. * @param fractionalExclusive The fractionalExclusive parameter */ private void setFractionalExclusive(boolean fractionalExclusive) { this.fractionalExclusive = fractionalExclusive; } /** * Getter for fractionalSpecificClassesAttributes attribute. * @return The fractionalSpecificClassesAttributes attribute. */ Map> getFractionalSpecificClassesAttributes() { return fractionalSpecificClassesAttributes; } /** * Set the fractionalSpecificClassesAttributes parameter. * @param fractionalSpecificClassesAttributes The * fractionalSpecificClassesAttributes parameter to set. */ private void setFractionalSpecificClassesAttributes( Map> fractionalSpecificClassesAttributes) { this.fractionalSpecificClassesAttributes = fractionalSpecificClassesAttributes; } /** * Getter for fractionalSpecificClassesAttributes attribute. * @return The fractionalSpecificClassesAttributes attribute. */ Set getFractionalAllClassesAttributes() { return fractionalAllClassesAttributes; } /** * Set the fractionalAllClassesAttributes parameter. * @param fractionalAllClassesAttributes The * fractionalSpecificClassesAttributes parameter to set. */ private void setFractionalAllClassesAttributes( Set fractionalAllClassesAttributes) { this.fractionalAllClassesAttributes = fractionalAllClassesAttributes; } /** * Getter for the base baseDN. * @return The baseDN attribute. */ DN getBaseDn() { return baseDN; } /** * Extract the fractional configuration from the passed domain configuration * entry. * @param configuration The configuration object * @return The fractional replication configuration. * @throws ConfigException If an error occurred. */ static FractionalConfig toFractionalConfig( ReplicationDomainCfg configuration) throws ConfigException { // Prepare fractional configuration variables to parse Iterator exclIt = configuration.getFractionalExclude().iterator(); Iterator inclIt = configuration.getFractionalInclude().iterator(); // Get potentially new fractional configuration Map> newFractionalSpecificClassesAttributes = new HashMap<>(); Set newFractionalAllClassesAttributes = new HashSet<>(); int newFractionalMode = parseFractionalConfig(exclIt, inclIt, newFractionalSpecificClassesAttributes, newFractionalAllClassesAttributes); // Create matching parsed config object FractionalConfig result = new FractionalConfig(configuration.getBaseDN()); switch (newFractionalMode) { case NOT_FRACTIONAL: result.setFractional(false); result.setFractionalExclusive(true); break; case EXCLUSIVE_FRACTIONAL: case INCLUSIVE_FRACTIONAL: result.setFractional(true); result.setFractionalExclusive( newFractionalMode == EXCLUSIVE_FRACTIONAL); break; } result.setFractionalSpecificClassesAttributes( newFractionalSpecificClassesAttributes); result.setFractionalAllClassesAttributes( newFractionalAllClassesAttributes); return result; } /** * Parses a fractional replication configuration, filling the empty passed * variables and returning the used fractional mode. The 2 passed variables * to fill should be initialized (not null) and empty. * @param exclIt The list of fractional exclude configuration values (may be * null) * @param inclIt The list of fractional include configuration values (may be * null) * @param fractionalSpecificClassesAttributes An empty map to be filled with * what is read from the fractional configuration properties. * @param fractionalAllClassesAttributes An empty list to be filled with * what is read from the fractional configuration properties. * @return the fractional mode deduced from the passed configuration: * not fractional, exclusive fractional or inclusive fractional * modes */ private static int parseFractionalConfig( Iterator exclIt, Iterator inclIt, Map> fractionalSpecificClassesAttributes, Set fractionalAllClassesAttributes) throws ConfigException { // Determine if fractional-exclude or fractional-include property is used: // only one of them is allowed int fractionalMode; Iterator iterator; if (exclIt != null && exclIt.hasNext()) { if (inclIt != null && inclIt.hasNext()) { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_BOTH_MODES.get()); } fractionalMode = EXCLUSIVE_FRACTIONAL; iterator = exclIt; } else { if (inclIt != null && inclIt.hasNext()) { fractionalMode = INCLUSIVE_FRACTIONAL; iterator = inclIt; } else { return NOT_FRACTIONAL; } } while (iterator.hasNext()) { // Parse a value with the form class:attr1,attr2... // or *:attr1,attr2... String fractCfgStr = iterator.next().toString(); StringTokenizer st = new StringTokenizer(fractCfgStr, ":"); int nTokens = st.countTokens(); if (nTokens < 2) { throw new ConfigException(NOTE_ERR_FRACTIONAL_CONFIG_WRONG_FORMAT.get(fractCfgStr)); } // Get the class name String classNameLower = st.nextToken().toLowerCase(); boolean allClasses = "*".equals(classNameLower); // Get the attributes String attributes = st.nextToken(); st = new StringTokenizer(attributes, ","); while (st.hasMoreTokens()) { String attrNameLower = st.nextToken().toLowerCase(); // Store attribute in the appropriate variable if (allClasses) { fractionalAllClassesAttributes.add(attrNameLower); } else { Set attrList = fractionalSpecificClassesAttributes.get(classNameLower); if (attrList == null) { attrList = new LinkedHashSet<>(); fractionalSpecificClassesAttributes.put(classNameLower, attrList); } attrList.add(attrNameLower); } } } return fractionalMode; } /** Return type of the parseFractionalConfig method. */ private static final int NOT_FRACTIONAL = 0; private static final int EXCLUSIVE_FRACTIONAL = 1; private static final int INCLUSIVE_FRACTIONAL = 2; /** * Get an integer representation of the domain fractional configuration. * @return An integer representation of the domain fractional configuration. */ private int fractionalConfigToInt() { if (!fractional) { return NOT_FRACTIONAL; } else if (fractionalExclusive) { return EXCLUSIVE_FRACTIONAL; } return INCLUSIVE_FRACTIONAL; } /** * Compare 2 fractional replication configurations and returns true if they * are equivalent. * @param cfg1 First fractional configuration * @param cfg2 Second fractional configuration * @return True if both configurations are equivalent. * @throws ConfigException If some classes or attributes could not be * retrieved from the schema. */ private static boolean isFractionalConfigEquivalent(FractionalConfig cfg1, FractionalConfig cfg2) throws ConfigException { // Compare base DNs just to be consistent if (!cfg1.getBaseDn().equals(cfg2.getBaseDn())) { return false; } // Compare modes if (cfg1.isFractional() != cfg2.isFractional() || cfg1.isFractionalExclusive() != cfg2.isFractionalExclusive()) { return false; } // Compare all classes attributes Set allClassesAttrs1 = cfg1.getFractionalAllClassesAttributes(); Set allClassesAttrs2 = cfg2.getFractionalAllClassesAttributes(); if (!areAttributesEquivalent(allClassesAttrs1, allClassesAttrs2)) { return false; } // Compare specific classes attributes Map> specificClassesAttrs1 = cfg1.getFractionalSpecificClassesAttributes(); Map> specificClassesAttrs2 = cfg2.getFractionalSpecificClassesAttributes(); if (specificClassesAttrs1.size() != specificClassesAttrs2.size()) { return false; } /* * Check consistency of specific classes attributes * * For each class in specificClassesAttributes1, check that the attribute * list is equivalent to specificClassesAttributes2 attribute list */ Schema schema = getServerContext().getSchema(); for (Map.Entry> entry : specificClassesAttrs1.entrySet()) { String className1 = entry.getKey(); // Get class from specificClassesAttributes1 ObjectClass objectClass1 = schema.getObjectClass(className1); if (objectClass1.isPlaceHolder()) { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_OBJECT_CLASS.get(className1)); } // Look for matching one in specificClassesAttributes2 boolean foundClass = false; for (Map.Entry> mapEntry : specificClassesAttrs2.entrySet()) { String className2 = mapEntry.getKey(); ObjectClass objectClass2 = schema.getObjectClass(className2); if (objectClass2.isPlaceHolder()) { throw new ConfigException( NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_OBJECT_CLASS.get(className2)); } if (objectClass1.equals(objectClass2)) { foundClass = true; // Now compare the 2 attribute lists Set attributes1 = entry.getValue(); Set attributes2 = mapEntry.getValue(); if (!areAttributesEquivalent(attributes1, attributes2)) { return false; } break; } } // Found matching class ? if (!foundClass) { return false; } } return true; } } /** * Specifies whether this domain is enabled/disabled regarding the ECL. * @return enabled/disabled for the ECL. */ boolean isECLEnabled() { return this.eclDomain.isEnabled(); } /** * Return the minimum time (in ms) that the domain keeps the historical * information necessary to solve conflicts. * * @return the purge delay. */ long getHistoricalPurgeDelay() { return config.getConflictsHistoricalPurgeDelay() * 60 * 1000; } /** * Check and purge the historical attribute on all eligible entries under this domain. * * The purging logic is the same applied to individual entries during modify operations. This * task may be useful in scenarios where a large number of changes are made as a one-off occurrence. * Running a purge-historical after the 'ds-cfg-conflicts-historical-purge-delay' period has elapsed * would clear out obsolete historical data from all the modified entries reducing the overall * database size. * * @param task * the task raising this purge. * @param endDate * the date to stop this task whether the job is done or not. * @throws DirectoryException * when an exception happens. */ public void purgeConflictsHistorical(PurgeConflictsHistoricalTask task, long endDate) throws DirectoryException { logger.trace("[PURGE] purgeConflictsHistorical " + "on domain: " + getBaseDN() + "endDate:" + new Date(endDate) + "lastCSNPurgedFromHist: " + lastCSNPurgedFromHist.toStringUI()); // It would be nice to have an upper bound on this filter to eliminate results that don't have a purgeable // csn in them. However, historicalCsnOrderingMatch keys start with serverid rather than timestamp so this // isn't possible. String filter = "(" + HISTORICAL_ATTRIBUTE_NAME + ">=dummy:" + lastCSNPurgedFromHist + ")"; int count = 0; boolean finished = false; ByteString pagingCookie = null; while(!finished) { if (task != null) { task.setProgressStats(lastCSNPurgedFromHist, count); } finished = true; SearchRequest request = Requests.newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, filter) .addAttribute(USER_AND_REPL_OPERATIONAL_ATTRS) .addControl(new PagedResultsControl(false, ConfigConstants.DEFAULT_SIZE_LIMIT, pagingCookie)) .setSizeLimit(ConfigConstants.DEFAULT_SIZE_LIMIT + 1); InternalSearchOperation searchOp = conn.processSearch(request); for (Control c : searchOp.getResponseControls()) { if (c.getOID().equals(OID_PAGED_RESULTS_CONTROL)) { ByteString newPagingCookie = ((PagedResultsControl)c).getCookie(); if( newPagingCookie != null && newPagingCookie.length() > 0 && !newPagingCookie.equals(pagingCookie)) { pagingCookie = newPagingCookie; finished = false; } } } for (SearchResultEntry entry : searchOp.getSearchEntries()) { long maxTimeToRun = endDate - TimeThread.getTime(); if (maxTimeToRun < 0) { throw new DirectoryException(ResultCode.ADMIN_LIMIT_EXCEEDED, LocalizableMessage.raw(" end date reached")); } EntryHistorical entryHist = EntryHistorical.newInstanceFromEntry(entry); CSN latestOldCSN = entryHist.getOldestCSN(); entryHist.setPurgeDelay(getHistoricalPurgeDelay()); Attribute attr = entryHist.encodeAndPurge(); if(entryHist.getLastPurgedValuesCount() > 0) { lastCSNPurgedFromHist = latestOldCSN; List mods = newArrayList(new Modification(ModificationType.REPLACE, attr)); count += entryHist.getLastPurgedValuesCount(); ModifyOperation newOp = new ModifyOperationBasis( conn, nextOperationID(), nextMessageID(), new ArrayList(0), entry.getName(), mods); runAsSynchronizedOperation(newOp); if (newOp.getResultCode() != ResultCode.SUCCESS) { // Log information for the repair tool. logger.error(ERR_CANNOT_ADD_CONFLICT_ATTRIBUTE, newOp, newOp.getResultCode()); } else if (task != null) { task.setProgressStats(lastCSNPurgedFromHist, count); } } } } // If a full sweep was completed, the lastCSNPurgedFromHist must be reset so that the next // run-through starts from the beginning. Otherwise, subsequent runs of the task would only // pick up purgeable changes for the last server id. lastCSNPurgedFromHist = new CSN(0,0,0); } }