/* * 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.IOException; 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 java.util.concurrent.locks.ReentrantReadWriteLock; 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.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.LockManager; 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 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; /** * What the ack of a delivery whose replay ran out of memory says did not apply the * change. *

* A constant rather than a message built from the change, because this is read on the * road out of an {@code OutOfMemoryError}: formatting one asks the JVM for the memory * it has just refused, and what {@code processUpdateDone()} needs of this string is * that there is one - it sets {@code hasReplayError} on the ack and never sends the * text, which is why no ordinal is spent on it either. */ private static final String REPLAY_RAN_OUT_OF_MEMORY = "the replay of this change ran out of memory"; /** 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(); /** * Set by {@link #restartService()} when it left the session of this domain alone, so * that the configuration change which asked for the restart can say so. *

* Cleared by that change before it applies anything, and read by it once it is done - * both under {@link #serviceStateLock}, the lock every restart of the session runs * under, so what it reads is what its own steps asked for. A restart which is * suppressed outside a configuration change leaves it set for the next one to clear. */ @GuardedBy("serviceStateLock") private boolean sessionRestartSuppressed; /** * Held while a replay thread applies a change of this domain, and taken exclusively by * this domain on its way down. *

* A change which reached the backend has to be recorded in the ServerState which is saved * when the domain is disabled or shut down, or it ends up in the data and in no * ServerState: the replication server sends it again and a change which is already * applied is replayed a second time (issue #908). The flags which stop the replay - * {@link #disabled}, {@link #shutdown} - are read under this lock as well, so a domain on * its way down sets one of them and then takes this lock: what it waits for is the * changes which were already being applied, and no attempt starts after that. *

* What this closes is the window between an operation reaching the backend and its * {@code commit()}, which is the one issue #908 reports. It is not every road by which a * change can be in the data and in no saved ServerState: {@code commit()} only advances * the state over the changes which are committed from the head of the pending list, so a * change applied while an older one is still to be replayed is forgotten by the * {@code clear()} on the way down and sent again - the barrier of issue #889 seen from * the other side, and a thing to fix where that barrier is rather than here. A change * this replica made itself is a third road, and the one the ServerState recovery in * {@code PersistentServerState.loadState()} already repairs, since it only looks for the * CSNs of this server. *

* Deliberately not the fair kind. The read lock is taken for every attempt made on the * backend, and a queued writer blocks the readers which come after it even without * fairness; the readers which do barge past it are the ones which see the flag and leave * without applying anything. The flag is read once before the lock for that reason too - * the replay threads are a pool shared by every domain, and a thread which parks on the * lock of a domain going down is a thread no other domain gets its changes replayed by. *

* The lock is released at the end of every attempt, so a domain going down waits for one * attempt and the short backoff which follows it inside the loop, rather than for all * the attempts a delivery is given. The backoff between two attempts is a Thread.sleep() * of tens of milliseconds; the one between two session restarts, which is counted in * seconds, is outside every lock and stays there. */ private final ReentrantReadWriteLock replayLock = new ReentrantReadWriteLock(); private final ReentrantReadWriteLock.ReadLock replayReadLock = replayLock.readLock(); private final ReentrantReadWriteLock.WriteLock replayWriteLock = replayLock.writeLock(); /** * How long this domain waits for the replay threads which are applying one of its changes * before it saves its ServerState and goes down. *

* Derived from the ceiling the server itself puts on an operation which is waiting for an * entry rather than picked: {@link LockManager} gives the subtree lock and the entry lock * {@link LockManager#DEFAULT_LOCK_TIMEOUT} each, so a replayed change whose target is held * by a concurrent local operation - a client deleting the subtree above it, say - is * inside its attempt for twice that before it gives up with BUSY. A bound under that * ceiling would be spent by ordinary lock contention, and the change which is applied * after it would be the one this whole barrier exists to keep out of that window: no * import, no index rebuild and no wedged backend needed. *

* A ceiling rather than a guarantee: an operation also waits for the subtree lock of every * entry above its target, one timeout each, so a deep contended chain outlasts this. What * it buys is that the wait is not lost to the contention a serving backend has anyway. *

* It is paid in three places - held under serviceStateLock, inside BackendConfigManager's * write lock when a backend is being deregistered, and once per domain by a server going * down - which is why it is bounded at all. It is only ever spent in full by a replay * which is genuinely stuck: the wait ends the moment the attempt does. */ private static final long REPLAY_DRAIN_TIMEOUT_IN_MS = 2 * LockManager.DEFAULT_LOCK_TIMEOUT_UNITS.toMillis(LockManager.DEFAULT_LOCK_TIMEOUT) + 1000; /** * How long this domain waits for the replay of its changes on its way down. Only the * tests, which can not hold a replay thread for {@link #REPLAY_DRAIN_TIMEOUT_IN_MS}, set * another value. */ private volatile long replayDrainTimeoutInMs = REPLAY_DRAIN_TIMEOUT_IN_MS; /** * 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; /** Published by a configuration change, read by the replay threads without the lock. */ private volatile boolean solveConflictFlag = true; private final InternalClientConnection conn = getRootConnection(); private final AtomicBoolean shutdown = new AtomicBoolean(); private volatile boolean disabled; /** * Whether the data of this domain is being replaced by a total update into this replica: * the import streams over the session of this domain, on its listener thread, and the * backend the replay would apply a change to is deregistered for the length of it. *

* It is what {@link #disabled} is for an import or a restore run on this server, on the * one road which does not set that flag: {@link #preBackendImport(LocalBackend)} takes * the backend away without disabling the domain, because the domain can not stop the * session it is importing over. The replay threads read it where they read * {@link #disabled}: an attempt made meanwhile is made into no backend, and whatever it * decided is overwritten by the import, so the change is given back instead - without a * session restart, which {@link #sessionHasAnOwner()} refuses for the whole of the total * update. It is set for the length of {@link #importBackend(InputStream)}, so it covers * the reload of the ServerState and the reset of the pending changes which follow the * import. */ private volatile boolean importingData; /** * This list is used to temporary store operations that needs to be replayed * at session establishment time. */ private final SortedMap replayOperations = new TreeMap<>(); /** Published by a configuration change, read by the changelog threads without a lock. */ private volatile ExternalChangelogDomain eclDomain; private final ServerStateFlush flushThread; /** * How long {@link #shutdown()} waits for the state checkpointer to stop before it goes on * without it. The checkpointer has one last state to write when it is asked to stop, so it * normally stops within a modify: a checkpointer which is still writing after this went to * a backend which is not answering, and waiting for it any longer would hang the shutdown * of the whole server. This is the budget {@code ServerShutdownMonitor} gives a thread * before it starts interrupting them. It is spent by every call of {@link #shutdown()}: * once per domain when the server goes down, which shuts the domains down one after * another, and again by a second caller of a domain whose checkpointer is stuck. It bounds * the shutdown of this domain, not of the server: a write which ignores the interrupt * still holds the quiescence of a pluggable backend, so the server shutdown waits for it * again when it closes that backend; {@code SchemaBackend} has no such wait. */ private static final long FLUSH_THREAD_SHUTDOWN_TIMEOUT_IN_MS = 30000; /** 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() { while (!isShutdownInitiated()) { try { synchronized (this) { wait(1000); } if (!disabled && !ieRunning()) { saveState(); } } catch (InterruptedException e) { // Thread interrupted: check for shutdown. Thread.currentThread().interrupt(); } } /* * A disabled domain saved its ServerState and cleared it from memory, and an import * or a restore is about to replace the data: saving here would write that empty state * over the saved one, which is a REPLACE of ds-sync-state with no value at all - the * replica would come back with no ServerState rather than with the one it saved on * its way down. The disabled flag does not catch the total update this replica is the * target of: preBackendImport() sets ignoreBackendInitializationEvent, so disable() is * not called on that road and only the import says the data is being replaced. * * The direction is asked for rather than ieRunning(), which the save in the loop above * settles for: an export leaves the data and the ServerState of this domain alone, and * the replay of this domain keeps running for the whole of it - a remote-requested * export is dispatched to a thread pool for that very reason. Since the save in the * loop is skipped for either direction, this is the only one which persists the * changes replayed since the export began, and there is no repair for them afterwards: * checkAndUpdateServerState() only repairs the CSNs of this server. */ if (!disabled && !importInProgress()) { saveState(); } } /** * Writes the state of the domain to the backend, keeping a failure to itself. *

* A checkpoint which throws is not a reason to stop checkpointing: the state is still * marked as unsaved, so the next checkpoint writes it again. The exit save has no next * checkpoint: a domain whose last write failed comes back with the last state it did * write - its own CSNs repaired from ds-sync-hist by checkAndUpdateServerState(), those * of the other replicas as they were - and replays the changes since. Letting the * exception out would end this thread - and with it the checkpointing of this domain for * the rest of the life of the server, and the {@link LDAPReplicationDomain#shutdown()} * which waits for the thread to stop. *

* The write is run outside the monitor of this thread, which * {@link LDAPReplicationDomain#shutdown()} takes to wake it up: holding the monitor * across a write which does not come back would block a shutdown before it ever reaches * the bounded wait it does for this thread. */ private void saveState() { try { state.save(); } catch (RuntimeException e) { logger.error(ERR_CHECKPOINTING_STATE_FAILED, getBaseDN(), stackTraceToSingleLineString(e)); } } } /** * 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); // Checked before the ECL configuration, which reads the backend to create its default // entry: a domain on a backend of its own reports that rather than what it made of it. LocalBackend backend = getBackend(); if (backend == null) { throw new ConfigException(ERR_SEARCHING_DOMAIN_BACKEND.get(getBaseDN())); } // The ECL domain is created here rather than handed a change it could refuse, so its // result can only be a success. createECLConfigurationEntryIfMissing(configuration); applyECLConfiguration(requireECLConfiguration(configuration)); solveConflictFlag = isSolveConflict(configuration); 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; } /* * 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 * stop, the change and the start have to be atomic together, which taking the lock * inside each of disableService()/enableService() does not make them. */ synchronized (serviceStateLock) { // Disable service if configuration changed final boolean needRestart = needReconnection && allowReconnection; if (needRestart) { disableService(); } else if (needReconnection) { onSessionRestartSuppressed(); } // 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(); } } } /** * 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(); final DN replayedEntryDN; try { replayedEntryDN = findEntryDN(uuid); } catch (SearchFailedException e) { return searchDidNotRun(ctx.getCSN(), e); } if (replayedEntryDN != 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 final DN parentDnFromCtx; try { parentDnFromCtx = findEntryDN(ctx.getParentEntryUUID()); } catch (SearchFailedException e) { return searchDidNotRun(ctx.getCSN(), e); } 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(); } /** * Turns down an operation being replayed whose conflict resolution could not read the * data, so that the change is attempted again rather than applied on what a search * which did not run seemed to say. *

* The operation is reported as unavailable, which is what it is, and the search which * did not run is its error message: the change is retried in place, and left out of * the ServerState and asked for again if this server keeps failing to serve the * searches this phase reads the data with (issue #956). Nothing is logged here: this * hook runs on every attempt in place, and the error is reported once, with the * attempt which ended them, where a storage which failed to serve the operation is. * * @param csn the CSN of the change being replayed * @param e the search which did not run * @return the result which stops the operation */ private SynchronizationProviderResult searchDidNotRun(CSN csn, SearchFailedException e) { return new SynchronizationProviderResult.StopProcessing( ResultCode.UNAVAILABLE, e.report(csn, getBaseDN())); } /** * 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(); if (offlineCSN != null) { /* * Only a message which really was published is announced: the shutdown of a collocated * replication server waits for it to be forwarded, and would spend the whole grace * period waiting for one which never reached the wire. */ dsrsShutdownSync.replicaOfflineMsgSent(getBaseDN(), offlineCSN); } else if (logger.isTraceEnabled()) { logger.trace("Replica " + getServerId() + " of domain baseDN=" + getBaseDN() + " could not announce itself offline: the message was not published - a change which" + " is still in flight holds it back, or the broker had no session to write it to," + " and " + pendingChanges.size() + " change(s) are pending"); } } /** * 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)) { /* * Wait for the changes which are being applied before the ServerState is flushed for * the last time: the flush thread stopped below is the last thing which saves it, so a * change which reaches the backend after that save is in the data and in no * ServerState (issue #908). The flag was just set, so this waits for the attempts * which had started already and no new one begins. */ awaitReplayDrained(); final RSUpdater rsUpdater = this.rsUpdater.get(); if (rsUpdater != null) { rsUpdater.initiateShutdown(); } // stop the thread in charge of flushing the ServerState. 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(); } } /* * Wait for completion of the ServerStateFlush thread, but not for longer than the budget * it is given: a thread which is gone - killed by an Error on its way to the backend, say * - is never going to report that it is done, and a shutdown which waits for it forever * takes the shutdown of the server down with it. join() covers both, and a thread which * was never started as well. * * Every caller waits, the one which lost the race above included: returning at once would * let it go on while the checkpointer is still writing. What the loser gets is the budget * from its own arrival, which starts before the winner has asked the checkpointer to stop * - the winner may still be in awaitReplayDrained() - so its wait may end, and log the * whole budget as spent, while the winner is still waiting: a second shutdown() of a * domain whose checkpointer is stuck can return before the first one. */ try { flushThread.join(FLUSH_THREAD_SHUTDOWN_TIMEOUT_IN_MS); if (flushThread.isAlive()) { logger.error(ERR_STATE_CHECKPOINTER_NOT_STOPPED, getBaseDN(), FLUSH_THREAD_SHUTDOWN_TIMEOUT_IN_MS); } } 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. *

* The change is given back on the way out of a replay which was unwound before it * reached one of the roads which give it back: a change which stays owned by a thread * which is not replaying it anymore is refused as a duplicate on every later delivery, * so this domain's ServerState would never move past it (issue #922). * * @param msg * The UpdateMsg to be replayed. * @param replayThreadShutdown * whether the replay thread was asked to stop */ void replay(LDAPUpdateMsg msg, AtomicBoolean replayThreadShutdown) { try { replayChangeAndTheChangesWaitingForIt(msg, replayThreadShutdown); } catch (Throwable t) { /* * The roads which run to their end give the change back themselves, so what is left * to give back here is a change whose replay was unwound over them: by an * OutOfMemoryError, which is left to end this thread, or by a throw from what the * replay runs once the ack of the delivery has been published - the give-back of the * change and the hand-out of the changes which were waiting for it are on that road. * * It takes the road of a failed replay rather than being handed back on the spot, so * that a change which keeps unwinding the replays it is given to is counted as * failing and is eventually given up on: handing it back bare would have this domain * ask for it, and restart its session for it, for as long as the server is up. * * The changes this thread parked as waiting for another change are left alone: they * are handed to whichever thread clears the change they are waiting for, and that * thread takes them over. * * Which change this thread owns is read before anything is done with it, and that * read takes no lock and allocates nothing: everything below is gated on the answer, * so a lookup which threw in its turn - on the road out of a JVM which has just * refused an allocation - would leave the change listed, uncommitted and owned by a * thread which is about to end, which is the state this whole issue is about. */ CSN owned = null; try { owned = remotePendingChanges.getChangeOwnedByCurrentThread(); if (owned != null) { if (replayThreadShutdown.get() || shutdown.get() || disabled) { /* * The replay was not failing, it was being abandoned: this thread is stopping * because their number is being changed, or the domain is going away or is * being imported into. The change is handed back without being counted against * its give-up budget, which is the road a replay abandoned that way takes. */ abandonReplay(owned); } else { /* * A JVM which has run out of memory is told apart here rather than reported on * its own road: the change is given back counted, the way every other unwound * replay gives it back, but the line which says it is being asked for again is * not built - that asks the JVM for the memory it has just refused - and the * session is restarted without sitting through the backoff, since the thread * which is doing it is on its way out. * * A change whose budget is spent is still reported and still raises its alert * on this road: it is the one line which says this replica has diverged, and an * operator who is not told would be left with a replica which is silently * behind. That is the deliberate exception to the rule above. */ recoverFromReplayFailure(owned, replayThreadShutdown, t instanceof OutOfMemoryError); } } } catch (Throwable recoveryFailure) { /* * The give-back is the road which hands the change over, so a throw out of it - an * allocation which fails in its turn, where what unwound the replay was a JVM out * of memory - would leave the change owned by this thread after all. Hand it back * bare, without the failure count that road did not reach, and restart the session * so that it is delivered again: this is the last resort, the throwable is rethrown * whatever happens here, and the thread this runs on may well be ending on it - so * a request left for the next recovery of this domain to pick up is a request which * may never be run. */ if (owned != null) { remotePendingChanges.replayFailed(owned); sessionRestartRequested.set(true); /* * Reported and restarted under guards of their own, and in that order: the report * is the line an operator acts on, and the restart is what has the change * delivered again - so a report which can not be formatted, on a road an * OutOfMemoryError leads to, must not cost the restart. */ try { logger.error(ERR_REPLAY_GIVE_BACK_FAILED, owned, getBaseDN(), stackTraceToSingleLineString(recoveryFailure)); } catch (Throwable reportFailure) { suppress(recoveryFailure, reportFailure); } try { runRequestedSessionRestarts(false); } catch (Throwable restartFailure) { /* * Nothing is left to try: the change is listed, uncommitted and unowned, so any * later session restart of this domain delivers it again. This goes with the * throwable which is rethrown below rather than being reported on its own. */ suppress(recoveryFailure, restartFailure); } } // The error which unwound the replay is the one reported, whatever the give-back // ran into on top of it. suppress(t, recoveryFailure); } throw t; } } /** * Records the second throwable as one the first suppressed, unless the two are one and * the same. *

* A JVM which has run out of memory hands out the error it prepared before it ran out as * often as it is asked for one, so the roads out of a replay can carry the same instance * twice - and a throwable can not suppress itself. * * @param thrown the throwable which is reported * @param alsoThrown what was met on the way out of it */ private static void suppress(Throwable thrown, Throwable alsoThrown) { if (thrown != alsoThrown) { thrown.addSuppressed(alsoThrown); } } /** * Replays the change of the provided message, then the changes which were waiting for * it, for as long as there are some. * * @param msg * The UpdateMsg to be replayed. * @param replayThreadShutdown * whether the replay thread was asked to stop */ private void replayChangeAndTheChangesWaitingForIt( 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; /* * Read once, before anything of this delivery has run, so that the report of an ack * which could not be published names the change even when reading it off the message * is what threw: that report is written on the way out of a road which is already * failing, and it must not be the throw which unwinds the replay. */ final CSN delivered = msg.getCSN(); 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; /* * The search conflict resolution could not run on the last attempt, when it ended * on one: the result code of the operation says nothing of it - it is the conflict * the operation failed on - so the failure of the server below the loop is told * here, and reported there once the attempts are spent. */ SearchFailedException searchFailedResolvingConflict = null; int retryCount = IN_PLACE_REPLAY_ATTEMPTS; while (!dependency && !replayDone && retryCount-- > 0) { /* * What this attempt ends on, not what an earlier one did: an attempt the server * refuses before it reaches the data makes no search, and the verdict below the * loop reads the attempt which spent the last of them. */ searchFailedResolvingConflict = null; /* * The flags which say this domain is going down, or that its data is being * replaced, are read before the lock as well as under it. The replay threads are * a pool shared by every domain of this server, so a thread which took a change * of a domain which is going down should not queue behind the wait for that * domain: the changes of every other domain are behind it in the same pool. * * The read under the lock is the one which decides; the one above it is a * scheduling optimisation for the common case and nothing more. It cannot keep * this thread out of the queue: the flag can be set and the writer can queue * between the two reads, and a reader which arrives behind a queued writer blocks * even on a lock which is not the fair kind. */ boolean goingDown = replayThreadShutdown.get() || shutdown.get() || disabled || importingData; if (!goingDown) { /* * Every attempt made on the backend is under this lock, and so is the decision * to make one: a domain on its way down takes it exclusively once it has set * the flag read here, so a change which reaches the backend is recorded in the * ServerState which is saved on the way down, or is not applied at all * (issue #908). */ replayReadLock.lock(); try { goingDown = replayThreadShutdown.get() || shutdown.get() || disabled || importingData; if (!goingDown) { 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; /* * The DN this looks for is compared with what the operation reports rather * than the other way round: getEntryDN() returns null - and reports * INVALID_DN_SYNTAX - for an operation whose raw DN does not parse, and the * replay must report that once and step over the change rather than throw * before the CSN of the operation has been read, which would keep the change * out of the ServerState and have it asked for again for the whole give-up * window (issue #928). */ if (SET_PERMISSIVE_MODIFY_FOR_DN.equals(modifyOperation.getEntryDN())) { 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; try { 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 } catch (SearchFailedException e) { /* * Conflict resolution reads the data with a search of the entryUUID, and * that search did not run: nothing was decided here, and whether the entry * is still in the data is not known. Report the failure of the server it * is rather than let a caller read "no entry" out of a search which never * answered. It is logged once the attempts in place are spent rather than * on every one of them, as a storage which failed to serve the operation * is: a backend which is down for a while fails every attempt of every * change delivered meanwhile. */ searchFailedResolvingConflict = e; resolution = ConflictResolution.SEARCH_FAILED; } 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 SEARCH_FAILED: /* * Conflict resolution could not read the data, so the change is not in * it and nothing was concluded about the entry it targets. Give it the * in-place attempts a storage which failed gets - a storage busy for a * moment must not cost a session restart - and leave it out of the * ServerState once they are spent, which the failure of the server * below the loop reports and acts on. A change which is not in the data * must not advance the ServerState (issue #889). */ Thread.sleep(50); 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; } } } finally { replayReadLock.unlock(); } } if (goingDown) { /* * 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 (!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. So is an attempt whose * conflict resolution could not read the data, which the attempt itself says * rather than its result code: that one is the conflict the operation failed * on, not the search which did not run (issue #956). */ if (isServerFailure(lastResult, serverErrorResultCode) || ResultCode.BUSY.equals(lastResult) || serverErrorResultCode.equals(lastResult) || searchFailedResolvingConflict != null) { /* * 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. * * The error reported is the operation's, unless the attempt ended on a search * conflict resolution could not run: the operation then only says which conflict * it failed on, and the search is what failed. */ final Object error = searchFailedResolvingConflict != null ? searchFailedResolvingConflict.report(csn, getBaseDN()) : op.getErrorMessage(); final LocalizableMessage message = ERR_ERROR_REPLAYING_OPERATION.get( op, csn, lastResult, error); 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, including what runs * after the operation did - solveNamingConflict() calls addConflict() there - * 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; } } catch (OutOfMemoryError e) { /* * The JVM is out of memory, which is not something to carry on replaying from: the * error is left to unwind the replay thread, which ends on it (issue #923). It is * not turned into a report of its own here either - the stack trace of an error * which can be raised anywhere says little, and the uncaught exception handler of * DirectoryThread writes the one line this is worth, with an alert. * * The other errors of the JVM take the road below. A StackOverflowError is met by * the thread which recursed and is gone once the stack has unwound, and the entry * being replayed is what raises it rather than the state of this server: ending a * thread on it would have one change this replica can not replay cost it a replay * thread per delivery, and nothing creates a replay thread to replace one which * ends. * * The change is given back, counted as failing and asked for again on the way out * (issue #922). The one thing done here is to make the ack this delivery publishes * below say that the change was not applied: a replica which is asking for a change * again must not have told an assured write that it is in the data here. It is a * constant rather than a message built from the change, because building one asks * the JVM for the memory it has just refused. */ replayErrorMsg = REPLAY_RAN_OUT_OF_MEMORY; throw e; } catch (Error e) { /* * An Error out of the replay - a LinkageError met where a plugin or a backend class * is loaded, an AssertionError - unwinds every road this replay has out of here, the * ones which give the change back among them. The change is not in the data, so it * is reported and given back here, on the road every other failed replay takes: the * ack below says it was not applied, the failure counts against the give-up budget * of the change, the session is restarted for it to be delivered again (issue #922) * and the changes which were waiting for this one are replayed rather than left * waiting for a thread to hand them out. * * It is not given up on where no operation could be built from the message, which is * what an Exception at that point means: an Error says that this server could not run * the replay, not that the message is one no delivery could ever build an operation * from. */ final LocalizableMessage message = ERR_ERROR_REPLAYING_CHANGE.get( msg.getCSN(), getBaseDN(), 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. */ try { processUpdateDone(msg, replayErrorMsg); } catch (OutOfMemoryError e) { /* * The one throw from here which is not caught, on the same terms as the arm * above: a JVM which has run out of memory is not something to carry on * replaying from, the error is left to end this replay thread, and the uncaught * exception handler of DirectoryThread raises the alert #923 is about. Swallowed * here instead, it would have this thread go on to the roads below and to the * next change of the loop on an exhausted heap, with nothing reported anywhere. * * What the give-back on the way out of replay() then finds depends on the road * the replay took to get here. A replay which failed, or which was unwinding on * an OutOfMemoryError of its own, still owns its change: it is given back * counted, and the thread ends on this error rather than on the one it stepped * over. A replay which committed owns nothing anymore - commit() cleared the * owner, and the index the give-back reads, in the same step - so the give-back * is a no-op, and rightly so: a change which is in the data is not one to ask * for again. What that road steps over is getNextUpdate() below, so the changes * parked behind the committed change wait for the next replay of this domain to * hand them out. That is the trade #923 asks for: a thread which met this error * is not to carry on, not even for them. */ throw e; } catch (Throwable ackFailure) { /* * Publishing the ack says nothing about whether the change was applied, so a * throw here must not be a road out of the replay. It would step over the * give-back of the change and, where the change was committed, over the * getNextUpdate() below - the one drain of the changes this thread parked as * waiting for it - leaving them waiting for a thread which is not replaying * anything anymore. * * A master which is waiting for the ack waits out its assured timeout either * way - and there is one only for an assured write in safe-read mode, which is * the one delivery a replica acknowledges. processUpdateDone() runs for every * delivery all the same: it accounts for the delivery in the receive window and * in the processed-updates counter, and a throw from there is a throw from this * bookkeeping, with no ack owed to anybody. Either way it is reported and the * replay carries on to the road the change itself decided: applied, failed and * asked for again, or given up on. * * Every step of processUpdateDone() catches what it can meet - the broker keeps * a failure to publish to itself and retries it - so what reaches here is what * no code of the replication protocol expected: an Error, and the tests drive * it as one. */ try { logger.error(ERR_ACK_NOT_PUBLISHED, delivered, getBaseDN(), stackTraceToSingleLineString(ackFailure)); } catch (Throwable reportFailure) { /* * Guarded like the report of a give-back which failed: this one runs on the * same kind of road - the stack of the throwable is walked to build the line - * and a report which can not be built must not become the throw which unwinds * the replay past the give-back and past getNextUpdate(). * * The line is tried once more with the name of the error alone, which walks no * stack. Nothing rethrows what was caught here, so an error recorded as * suppressed on it would be recorded nowhere, and this is the road on which an * operator has the least to go on. A second refusal leaves nothing to say it * with, and the replay carries on with what the change decided. */ try { logger.error(ERR_ACK_NOT_PUBLISHED, delivered, getBaseDN(), ackFailure.getClass().getName()); } catch (Throwable secondReportFailure) { // Nothing is left to say it with. } } } } } 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 is given back: the replication server // delivers it again, now or - while a total update owns the session - after the // import restarts it. 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: the replay throws past the point where the operation ran as well - * addConflict() does, once solveNamingConflict() has seen the result - 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. *

* Package private for the tests, which pin what it answers for every registered result * code directly rather than through a running server. * * @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 */ @VisibleForTesting 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 the {@code replay-give-up-delay} of this domain 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) { return recoverFromReplayFailure(csn, replayThreadShutdown, false); } /** * Recovers from a change which could not be replayed, telling apart the replay which was * unwound by a JVM out of memory. * * @param csn * the CSN of the change which could not be replayed * @param replayThreadShutdown * whether the replay thread was asked to stop * @param outOfMemory * whether what unwound the replay was the JVM running out of memory, in which * case the line which says the change is being asked for again is not built - * the report of a change which is given up on is, since it is what says this * replica has diverged - and the session is restarted without the backoff * @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, boolean outOfMemory) { /* * 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; } /* * The budget is read from the configuration at every decision rather than kept in a * field of its own: an administrator who raises it because a maintenance window is * going to outlast it is not made to restart this server for that, and * applyConfigurationChange() replaces the configuration object as a whole. A negative * value is the "unlimited" of the duration syntax - this replica then keeps asking for * the change rather than ever recording one it did not apply, which is the choice of * an operator who would rather have the replication of this domain stop than have it * diverge. The generated getter yields the value in the base unit the property is * declared with, which is milliseconds here, so it is comparable to what the failure * reports as it is. */ final long giveUpDelayInMs = config.getReplayGiveUpDelay(); if (giveUpDelayInMs >= 0 && failure.getFailingForMs() >= giveUpDelayInMs) { 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 (sessionHasAnOwner()) { /* * 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; restarting the one an import streams over * would end the import on the entries which had arrived. The change given back here * is forgotten with the rest of the pending changes when the ServerState is loaded * again, from the backend or from the imported data, and the session started then * asks for everything that state does not cover - or, when the total update it was * given back for never begins, it is asked for by the next restart * (see sessionHasAnOwner()). */ return true; } if (!outOfMemory) { /* * Not on the road out of a JVM which has run out of memory: building this line asks * it for the memory it has just refused, and the ack of the delivery already says * that the change was not applied. The constant that ack carries exists for the same * reason. */ 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. Neither does * the thread an OutOfMemoryError is ending, for the same reason - and the restart is * run rather than left to be asked for again, because that thread will not be there to * run it, and a change nobody asks for again holds this domain's ServerState back. */ runRequestedSessionRestarts(!replayThreadShutdown.get() && !outOfMemory); 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)) { boolean restarted = false; try { restartSession(wait); restarted = true; } finally { if (!restarted) { /* * The request is put back where it was taken from. The flag is read and * cleared before the restart runs, so a restart which ends abruptly - the * session is stopped first, and starting it again creates a listener thread, * which the operating system can refuse - would otherwise leave this domain * with no session and with nothing left to ask for one. * * What a request left standing buys is bounded, and the bound is worth * stating. Its two readers are the roads out of a failed and of an abandoned * replay of this domain, and with no listener thread nothing is delivered * anymore: the replays left to run are the changes already taken off the * session - the ones waiting in the replay queue, and the ones parked as * dependencies. One of those failing finds the request standing and runs the * restart, which starts from a clean state, since disableService() drops the * listener thread which was never started. Once they are spent, the domain * stays down until it is disabled and enabled back, or the server is * restarted. That is said where it can be heard: a refused thread is an * OutOfMemoryError, and one which leaves recoverFromReplayFailure() or * abandonReplay() ends the replay thread it is met on, so the uncaught * exception handler of DirectoryThread writes the line and raises the alert, * with the start of the listener thread in the trace. */ sessionRestartRequested.set(true); } } } } 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 (sessionHasAnOwner()) { // The domain, or the import into it, owns its session, and the pending changes are // forgotten with the ServerState on its way down or at the end of the import. 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 (sessionHasAnOwner()) { // The domain is going away or is being imported into: the session is not this // thread's to stop. return; } disableService(); stoppedSession = getSessionGeneration(); } 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 (ownsItsSession() || getSessionGeneration() != stoppedSession) { /* * 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. Every stop and every start of a session is counted, so * the generation alone tells one session from another. * * An import is not asked about here: the session it would stream over is the one * this thread stopped, so none is streaming, and a total update which was asked * for meanwhile needs the session started back to be answered at all - its * request is what fails, and loudly, when no answer comes. */ return; } enableService(); } } /** * 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(); } } /** * 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. * @throws SearchFailedException if the search did not run, so that whether there is * such an entry is not known */ private DN findEntryDN(String uuid) throws SearchFailedException { if (uuid == null) { // A change which carries no entryUUID names no entry. return null; } /* * The entryUUID comes off the wire and nothing validates it as one, so it is looked up * as the value it is rather than read as part of a filter string: a value which * carried a wildcard would be read as a substring filter, and one which did not parse * - a dangling escape - would be a search which never runs, for as long as the change * is asked for. */ final SearchFilter filter = SearchFilter.createEqualityFilter( getServerContext().getSchema().getAttributeType(ENTRYUUID_ATTRIBUTE_NAME), ByteString.valueOfUtf8(uuid)); final InternalSearchOperation search = conn.processSearch(newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, filter)); if (search.getResultCode() != ResultCode.SUCCESS) { if (search.getResultCode() == ResultCode.NO_SUCH_OBJECT && baseEntryIsAbsentFromALiveBackend()) { /* * The backend serves the base DN and has no base entry yet: the search ran, and * nothing is below a base entry which is not there. This is the empty replica * about to receive it. */ return null; } /* * The search did not run - the backend is offline or being rebuilt, or the storage * failed to serve it - so it read nothing of the data: an entry it did not report * is not an entry which is not there. */ throw new SearchFailedException(uuid, search.getResultCode().getName() + " " + search.getErrorMessage()); } final SearchResultEntry resultEntry = getFirstResult(search); return resultEntry != null ? resultEntry.getName() : null; } /** * Returns whether the backend which serves the base DN of this domain is there and * holds no base entry, which is the state of an empty replica. *

* A search under the base DN of such a backend answers NO_SUCH_OBJECT, which is the * answer of a backend which is offline or being rebuilt as well - and no backend * answers SUCCESS with no entry for a base which is not there. The backend itself * tells the two apart: the search ran over a backend which is there and empty, and it * did not run over one which is gone or fails to answer. * * @return {@code true} if the backend is there and holds no base entry */ private boolean baseEntryIsAbsentFromALiveBackend() { final LocalBackend backend = getBackend(); if (backend == null) { // Nothing serves the base DN: the backend is offline or being rebuilt. return false; } try { return !backend.entryExists(getBaseDN()); } catch (DirectoryException e) { // The storage did not answer this any more than it answered the search. return false; } } /** 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 search conflict resolution reads the data with did not run: nothing was * decided, and the operation has to be attempted again. */ SEARCH_FAILED, /** The operation failed for a reason which is not a naming conflict. */ FAILED } /** * Reports a search conflict resolution reads the data with and which did not run: it * read nothing, so its result is no evidence about the data. *

* A search which failed and a search which found nothing look the same to a caller * which only reads the entries it returned, and every caller here reads "no entry" as * "the entry has been deleted": that answers {@link ConflictResolution#NOTHING_TO_DO}, * which records a change that was never applied as replayed, and the replication server * never sends a change again which this replica reports itself past (issue #956). */ private static final class SearchFailedException extends Exception { private static final long serialVersionUID = 1L; /** The entryUUID the search which did not run was looking for. */ private final String entryUUID; private SearchFailedException(String entryUUID, String cause) { super(cause); this.entryUUID = entryUUID; } /** * Describes the search which did not run for the change it was made for. * * @param csn the CSN of the change being replayed * @param baseDN the base DN of the domain the change belongs to * @return the error to report the change with */ private LocalizableMessage report(CSN csn, DN baseDN) { return ERR_REPLAY_ENTRYUUID_SEARCH_FAILED.get(csn, baseDN, entryUUID, getMessage()); } } /** * 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 * @throws SearchFailedException if the data could not be read */ private ConflictResolution solveNamingConflict(ModifyOperation op, ModifyMsg msg) throws SearchFailedException { 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 * @throws SearchFailedException if the data could not be read */ private ConflictResolution solveNamingConflict(DeleteOperation op, LDAPUpdateMsg msg) throws SearchFailedException { 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); 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. * * This is answered before the new superior is looked up, and before the branch which * marks the entry as conflicting: an entry which is not in the database can not be * marked, and the delete has already settled what this Modify DN was trying to do. */ numResolvedNamingConflicts.incrementAndGet(); return ConflictResolution.NOTHING_TO_DO; } // 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 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 IOException When the conflict marker could not be added to the msg. */ private void addConflict(AddMsg msg) throws IOException { 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) { /* * The replay is stopped before the ServerState is saved, and not the other way round: * a change a replay thread is applying has to be either recorded in the state which * is about to be saved or not applied at all, or it ends up in the data and in no * ServerState (issue #908). The flag keeps the attempts which have not started from * starting - it is read under the same lock as the attempt it guards - and the wait * below is for the ones which had started already. * * All of it stays under serviceStateLock, so that this and enable() remain the * mutually exclusive pair they have always been: an enable() which ran in the middle * of this would clear the flag and bring a session up, and this would then go on to * cut that session and clear a ServerState which the flush thread - reading a flag * which says the domain is enabled - would write back empty. That is what bounds * REPLAY_DRAIN_TIMEOUT_IN_MS: the wait is held under a lock which a session restart, * a configuration change and the shutdown of this domain take, and it runs inside * BackendConfigManager's write lock when a backend is being deregistered. */ disabled = true; disableService(); // This will cut the session and wake up the listener awaitReplayDrained(); state.save(); state.clearInMemory(); /* * 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); } } /** * Waits for the replay threads which are applying a change of this domain to be done * with it. *

* Called once {@link #disabled}, {@link #shutdown} or {@link #importingData} has been * set, which is what bounds the wait: a replay thread reads those under * {@link #replayReadLock}, the lock this takes exclusively, so no attempt starts once * this returns and what it waits for is the attempts which were running already. The * lock is released before returning for the same reason - what keeps the replay out is * the flag, not the lock. */ private void awaitReplayDrained() { boolean drained = false; boolean interrupted = false; try { drained = replayWriteLock.tryLock(replayDrainTimeoutInMs, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { /* * Give up waiting, and put the interrupt back rather than swallow it: whoever * interrupted this thread - the server going down, a thread pool taking its threads * away - is still waiting for it to stop, and this is not the last thing it does. * The cost is on the shutdown road, whose wait for the last ServerState flush is a * Thread.sleep() which ends on its first call once the flag is set: the flush thread * still runs that save, this one just stops waiting for it. */ interrupted = true; Thread.currentThread().interrupt(); } if (drained) { replayWriteLock.unlock(); return; } /* * The change which is being applied may reach the backend without being recorded in the * ServerState which is saved next, so the replication server sends it again and it is * replayed a second time. Better than holding an administrative task - an import, a * restore, a backend being taken offline - for as long as a backend which stopped * answering takes to answer. * * An interrupted wait is reported as what it is: it says nothing about how long the * replay of this domain takes, and the timeout it never spent would have an operator * reading a backend which is slow into it. */ if (interrupted) { logger.warn(WARN_REPLAY_DRAIN_INTERRUPTED, getBaseDN()); } else { logger.warn(WARN_REPLAY_NOT_DRAINED, getBaseDN(), replayDrainTimeoutInMs); } } /** * Returns how long this domain waits for the replay threads which are applying one of * its changes before it saves its ServerState and goes down. * * @return the timeout in milliseconds */ @VisibleForTesting public long getReplayDrainTimeout() { return replayDrainTimeoutInMs; } /** * Sets how long this domain waits for the replay threads which are applying one of its * changes before it saves its ServerState and goes down. *

* Only there for the tests which check what a domain does when that wait runs out: they * can not hold a replay thread for {@link #REPLAY_DRAIN_TIMEOUT_IN_MS}. * * @param timeoutInMs the timeout in milliseconds */ @VisibleForTesting public void setReplayDrainTimeout(long timeoutInMs) { replayDrainTimeoutInMs = timeoutInMs; } /** * 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; } /* * The flag is cleared before the session is started, where disable() sets it before * stopping one: enableService() ends with startListenService(), so the listener it * starts can list a delivery and hand it to a replay thread while this method is * still running. A replay thread which reads a flag that still says "disabled" * gives the change up at the top of its replay loop, and abandonReplay() does not * ask for it again - a domain on its way down owns its session - so the change is * left listed, uncommitted and owned by nobody. Nothing would replay it: the * replication server only sends it again over a session which is restarted, so this * domain's ServerState, and every change which depends on that one, would be held * back for as long as the session lives. */ disabled = false; boolean started = false; try { enableService(); started = true; } finally { if (!started) { /* * The other half of the same invariant: a domain whose session could not be * started owns that session the way a disabled one does, so the flag goes back * where it was rather than leave the replay threads believing there is a session * of theirs to restart. */ disabled = true; } } } } /** * 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. *

* A base entry which is not in the backend leaves the generationId unstored until the * entry appears, and is not an error - it is what a suffix waiting to be initialized by * an import looks like. The generationId used to be stored on the domain configuration * entry instead, and must not be again: that write reaches * {@link #applyConfigurationChange(ReplicationDomainCfg)} with the configuration * backend's update lock held, and so takes {@link #serviceStateLock} in the order * opposite to the one {@link #disable()} takes the two in. The generationId of a suffix * with no entry is a constant which {@code loadGenerationId()} computes again for free, * and a value a former version left on the configuration entry is still read back. * * @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) { 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 { /* * The replay of this domain is held off before the backend is taken away, the way * disable() holds it off before the backend is taken away for an import run on this * server: the flag keeps the attempts which have not started from starting, and the * wait is for the ones which had. What this road can not do is stop the session - * it is the session the entries are about to stream over, on this very thread. */ importingData = true; awaitReplayDrained(); 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)))); } finally { /* * The ServerState in memory is the one read from the data now, and the changes * listed as pending must not outlive the one they went with, as they do not when * the domain is disabled: a change given back while the import ran would stay * listed, uncommitted and owned by nobody, and a commit stops at the first * uncommitted change - the state would never move past it, and the replication * server does not send a change again which the state it is given covers. Nothing * listed a change meanwhile: the listener thread is the one running this import, * and the replay threads gave up every attempt while the flag was set. The restart * a replay thread may have asked for before the total update owned the session * goes with them: the caller starts the session again from the reloaded state. */ remotePendingChanges.clear(); sessionRestartRequested.set(false); consecutiveSessionRestarts.set(0); importingData = false; } } 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) { final ConfigChangeResult ccr = new ConfigChangeResult(); /* * The step which can fail comes first, so that none of the domain configuration is * published before it succeeded. It is not free of writes of its own: a domain * without an external changelog configuration is given the default entry here, and * that entry stays written whether or not the rest of this succeeds. Refusing a * change before anything at all is written is what isConfigurationChangeAcceptable() * is for, and it is where a change carrying an unreadable configuration is refused * for good - the modified entry is written to the server configuration between that * method and this one, and it is not rolled back when this one reports an error. What * this reads is read again under the lock, so that a change of that entry which lands * in between is applied rather than reverted by this snapshot of it. */ try { createECLConfigurationEntryIfMissing(configuration); // Read once here so that a configuration which cannot be read fails before any of // this change is published, and once more under the lock so that what is applied is // not an outdated snapshot of it. requireECLConfiguration(configuration); } catch (Exception e) { ccr.setResultCode(ResultCode.OTHER); ccr.addMessage(configChangeFailed(configuration, e)); return ccr; } /* * 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) { // Whatever a restart was suppressed for before this change is none of its business. sessionRestartSuppressed = false; /* * Reported rather than thrown, all of it: this listener runs on an entry which is * already written, and an exception leaving it would abort the listeners after it * as well. What is applied when one of these fails stays applied - which is what * the framework says of a listener of an entry written before it runs. */ try { this.config = configuration; changeConfig(configuration); // Read assured + fractional configuration and each time reconnect if needed. A // domain which owns its session gets none of those reconnections. final boolean allowReconnection = !ownsItsSession(); readAssuredConfig(configuration, allowReconnection); readFractionalConfig(configuration, allowReconnection); solveConflictFlag = isSolveConflict(configuration); /* * Applied last, and still under the lock: this one restarts the session as well * when the attributes published to the external changelog changed, and the * session it starts replays on everything set above. It is read again here, so * that a change of that entry which landed in between is applied rather than * reverted by an older snapshot of it. */ ccr.aggregate(applyECLConfiguration(requireECLConfiguration(configuration))); } catch (Exception e) { ccr.setResultCode(ResultCode.OTHER); ccr.addMessage(configChangeFailed(configuration, e)); } /* * Read here rather than reported by restartService() itself, which has no result to * report through and is called from the external changelog configuration above as * well: what a change needed a session for, and did not get, is one thing to the * administrator whichever step of it asked. Left out of the failure path on purpose * - a change which could not be applied has a reason of its own to carry, and the * session it did not restart is not what the administrator has to act on. */ if (sessionRestartSuppressed && ccr.getResultCode() == ResultCode.SUCCESS) { ccr.setAdminActionRequired(true); ccr.addMessage(NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED.get(getBaseDN())); } } return ccr; } /** * Whether this domain, rather than a configuration change, decides when its session * runs: it is shutting down, or it is disabled for the length of a total update. */ private boolean ownsItsSession() { return shutdown.get() || disabled; } /** * Whether the session of this domain has an owner other than the replay thread which * would restart it after a failed replay: the domain itself, when it is shutting down or * disabled ({@link #ownsItsSession()}), or a total update into this replica. *

* The total update owns the session from the moment it is asked for, not from the * moment its entries stream: the {@code InitializeTargetMsg} which answers the request * arrives over that session, so a restart made while it is on its way loses it, and the * import which follows reads its entries over the same session - stopping it ends the * import on the entries which had arrived. The import starts the next session itself, * from the state it loaded. A change given back while the total update owned the session * is not asked for again by anyone until then; if no import follows - the request was * refused, or gave up waiting - it stays listed until the next failed replay restarts * the session, which has the replication server send it again with everything after it. * Listed, it holds the ServerState back as well: a commit moves the state no further than * the oldest uncommitted change, so the state in memory, and the one persisted from it, * stop at the change until that restart. */ private boolean sessionHasAnOwner() { return ownsItsSession() || importInProgress(); } @Override protected void restartService() { synchronized (serviceStateLock) { if (ownsItsSession()) { /* * The domain is going away or is being imported into: a restart here would bring * a session, and the listener thread which goes with it, back up on a domain * whose ServerState is gone from memory. The session started when the domain is * enabled again reads the configuration this restart was asked for. * * Recorded rather than passed over in silence: the configuration a restart was * asked for is stored, and it is the session which is not brought up on it, so a * change which reports plain success would have the administrator believe the * domain is running on it already. A domain disabled for a total update comes up * on it when the total update ends; one which stays disabled - enable() gives up * when the data state it reads cannot be loaded, and nothing calls it again - * never does, and that is what the administrator is told to act on. */ onSessionRestartSuppressed(); return; } super.restartService(); } } /** * {@inheritDoc} *

* Every restart this domain leaves alone comes through here, whichever step of the * change asked for it: the broker properties which are renegotiated, the assured * configuration the replication server is told about as the session comes up, the * fractional configuration the session filters on, and the attributes the external * changelog publishes. */ @Override protected void onSessionRestartSuppressed() { synchronized (serviceStateLock) { sessionRestartSuppressed = true; } } /** * What the administrator is told a configuration change failed with. *

* The reason a {@link ConfigException} carries is passed on as it is: it names the step * which raised it, the external changelog configuration this listener reads included. * Anything else comes out of applying the domain configuration - the broker, the * assured and the fractional configuration - and is reported as such rather than as a * failure of the external changelog, which most of what this listener does has nothing * to do with. */ private LocalizableMessage configChangeFailed(ReplicationDomainCfg domCfg, Exception e) { if (e instanceof ConfigException) { return ((ConfigException) e).getMessageObject(); } return ERR_REPLICATION_DOMAIN_CONFIG_CHANGE_FAILED.get( domCfg.getBaseDN(), stackTraceToSingleLineString(e)); } @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 the external changelog configuration can be read. This is the one thing * applying the change can fail on, and this is where refusing it means something: the * modified entry is written to the server configuration between this method and * applyConfigurationChange(), and it is not rolled back when the latter reports an * error. Refused here, the entry is never written at all. *

* A domain whose external changelog configuration cannot be read therefore refuses * every change of its own entry until that configuration is repaired. What repairs * it is a change of the "cn=external changelog" entry, or its removal, neither of * which comes through here. */ try { readECLConfiguration(configuration); } catch (ConfigException e) { unacceptableReasons.add(e.getMessageObject()); 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 = eclConfigurationEntryDN(config); 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)); } } /** * Reads the ECL configuration of the domain, changing nothing. *

* This is what {@link #isConfigurationChangeAcceptable} checks, so it leaves the server * configuration as it found it - the entry the domain is missing is created by * {@link #createECLConfigurationEntryIfMissing} once the change is accepted. It is read * off the provided configuration rather than off {@link #config}, which the caller of * the latter has not published yet. * * @param domCfg The provided configuration. * @return The ECL configuration, or {@code null} when the domain has none yet. * @throws ConfigException When it exists but could not be read. */ private ExternalChangelogDomainCfg readECLConfiguration(ReplicationDomainCfg domCfg) throws ConfigException { try { return domCfg.getExternalChangelogDomain(); } catch (Exception e) { if (!Boolean.TRUE.equals(hasECLConfigurationEntry(domCfg))) { /* * There is none to read - a default one is created when the change is applied - * or whether there is one could not be told, and a domain is not held back from * every change of its own by a failure which never reached its entry. */ return null; } throw new ConfigException(NOTE_ERR_UNABLE_TO_ENABLE_ECL.get( "Replication Domain on " + domCfg.getBaseDN(), stackTraceToSingleLineString(e)), e); } } /** Reads the ECL configuration which must be there, as the caller has just created it. */ private ExternalChangelogDomainCfg requireECLConfiguration(ReplicationDomainCfg domCfg) throws ConfigException { final ExternalChangelogDomainCfg eclDomCfg = readECLConfiguration(domCfg); if (eclDomCfg == null) { throw new ConfigException(NOTE_ERR_UNABLE_TO_ENABLE_ECL.get( "Replication Domain on " + domCfg.getBaseDN(), "its external changelog configuration is gone")); } return eclDomCfg; } private static DN eclConfigurationEntryDN(ReplicationDomainCfg domCfg) { return DN.valueOf("cn=external changelog," + domCfg.dn()); } /** Whether the ECL configuration entry is there, or {@code null} when it cannot be told. */ private Boolean hasECLConfigurationEntry(ReplicationDomainCfg domCfg) { try { final ConfigurationHandler configHandler = getServerContext().getConfigurationHandler(); // There may not be any config entry related to this domain in some unit test cases return configHandler.hasEntry(domCfg.dn()) && configHandler.hasEntry(eclConfigurationEntryDN(domCfg)); } catch (Exception e) { logger.traceException(e); return null; } } /** * Creates the entry the ECL configuration of the domain is stored in, with its default * values, when the server configuration does not carry one yet. * * @param domCfg The provided configuration. * @throws ConfigException When an error occurred. */ private void createECLConfigurationEntryIfMissing(ReplicationDomainCfg domCfg) throws ConfigException { if (readECLConfiguration(domCfg) != null) { return; } // create the ecl config if it does not exist try { DN configDn = domCfg.dn(); ConfigurationHandler configHandler = getServerContext().getConfigurationHandler(); // domain with no config entry only when running unit tests if (configHandler.hasEntry(configDn)) { if (!configHandler.hasEntry(eclConfigurationEntryDN(domCfg))) { if (getBackend() == null) { // Read to tell a private backend from a public one just below. throw new ConfigException(ERR_SEARCHING_DOMAIN_BACKEND.get(domCfg.getBaseDN())); } // 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); try (LDIFReader reader = new LDIFReader(ldifImportConfig)) { Entry eclEntry = reader.readEntry(); configHandler.addEntry(Converters.from(eclEntry)); } } } } catch (ConfigException e) { throw e; } catch (Exception e) { throw new ConfigException(NOTE_ERR_UNABLE_TO_ENABLE_ECL.get( "Replication Domain on " + domCfg.getBaseDN(), stackTraceToSingleLineString(e)), e); } } /** * Applies the provided ECL configuration to this domain. *

* This restarts the session when the attributes published to the external changelog * changed, so the configuration it is applied along must be in place already. * * @param eclDomCfg The ECL configuration read by {@link #requireECLConfiguration}. * @return What the ECL domain made of the change: it reports a change it cannot apply * rather than throwing it. * @throws ConfigException When applying it failed, the session it restarts included. */ private ConfigChangeResult applyECLConfiguration(ExternalChangelogDomainCfg eclDomCfg) throws ConfigException { try { if (eclDomain != null) { return eclDomain.applyConfigurationChange(eclDomCfg); } // Create the ECL domain object eclDomain = new ExternalChangelogDomain(this, eclDomCfg); return new ConfigChangeResult(); } 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()); attributes.add("changes-with-failed-replay", remotePendingChanges.getFailingChangesSize()); } /** * 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); } }