/*
|
* CDDL HEADER START
|
*
|
* The contents of this file are subject to the terms of the
|
* Common Development and Distribution License, Version 1.0 only
|
* (the "License"). You may not use this file except in compliance
|
* with the License.
|
*
|
* You can obtain a copy of the license at
|
* trunk/opends/resource/legal-notices/OpenDS.LICENSE
|
* or https://OpenDS.dev.java.net/OpenDS.LICENSE.
|
* See the License for the specific language governing permissions
|
* and limitations under the License.
|
*
|
* When distributing Covered Code, include this CDDL HEADER in each
|
* file and include the License file at
|
* trunk/opends/resource/legal-notices/OpenDS.LICENSE. If applicable,
|
* add the following below this CDDL HEADER, with the fields enclosed
|
* by brackets "[]" replaced with your own identifying information:
|
* Portions Copyright [yyyy] [name of copyright owner]
|
*
|
* CDDL HEADER END
|
*
|
*
|
* Copyright 2006-2009 Sun Microsystems, Inc.
|
*/
|
package org.opends.server.replication.plugin;
|
|
import static org.opends.messages.ReplicationMessages.*;
|
import static org.opends.messages.ToolMessages.*;
|
import static org.opends.server.loggers.ErrorLogger.logError;
|
import static org.opends.server.loggers.debug.DebugLogger.debugEnabled;
|
import static org.opends.server.loggers.debug.DebugLogger.getTracer;
|
import static org.opends.server.replication.plugin.Historical.ENTRYUIDNAME;
|
import static org.opends.server.replication.protocol.OperationContext.*;
|
import static org.opends.server.util.ServerConstants.*;
|
import static org.opends.server.util.StaticUtils.createEntry;
|
import static org.opends.server.util.StaticUtils.getFileForPath;
|
import static org.opends.server.util.StaticUtils.stackTraceToSingleLineString;
|
|
import java.io.File;
|
import java.io.InputStream;
|
import java.io.OutputStream;
|
import java.io.StringReader;
|
import java.io.UnsupportedEncodingException;
|
import java.util.ArrayList;
|
import java.util.Collection;
|
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.SortedSet;
|
import java.util.StringTokenizer;
|
import java.util.TreeMap;
|
import java.util.concurrent.BlockingQueue;
|
import java.util.concurrent.TimeoutException;
|
import java.util.concurrent.atomic.AtomicInteger;
|
import java.util.zip.CheckedOutputStream;
|
import java.util.zip.DataFormatException;
|
|
import org.opends.messages.Message;
|
import org.opends.messages.MessageBuilder;
|
import org.opends.server.admin.server.ConfigurationChangeListener;
|
import org.opends.server.admin.std.meta.ReplicationDomainCfgDefn.AssuredType;
|
import org.opends.server.admin.std.meta.ReplicationDomainCfgDefn.*;
|
import org.opends.server.admin.std.server.ExternalChangelogDomainCfg;
|
import org.opends.server.admin.std.server.ReplicationDomainCfg;
|
import org.opends.server.api.AlertGenerator;
|
import org.opends.server.api.Backend;
|
import org.opends.server.api.DirectoryThread;
|
import org.opends.server.api.SynchronizationProvider;
|
import org.opends.server.backends.jeb.BackendImpl;
|
import org.opends.server.backends.task.Task;
|
import org.opends.server.config.ConfigException;
|
import org.opends.server.controls.SubtreeDeleteControl;
|
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.loggers.debug.DebugTracer;
|
import org.opends.server.protocols.asn1.ASN1Exception;
|
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.ldap.LDAPAttribute;
|
import org.opends.server.protocols.ldap.LDAPFilter;
|
import org.opends.server.protocols.ldap.LDAPModification;
|
import org.opends.server.replication.common.AssuredMode;
|
import org.opends.server.replication.common.ChangeNumber;
|
import org.opends.server.replication.common.ChangeNumberGenerator;
|
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.ProtocolSession;
|
import org.opends.server.replication.protocol.RoutableMsg;
|
import org.opends.server.replication.protocol.UpdateMsg;
|
import org.opends.server.replication.service.ReplicationBroker;
|
import org.opends.server.replication.service.ReplicationDomain;
|
import org.opends.server.replication.service.ReplicationMonitor;
|
import org.opends.server.tasks.TaskUtils;
|
import org.opends.server.types.AbstractOperation;
|
import org.opends.server.types.Attribute;
|
import org.opends.server.types.AttributeBuilder;
|
import org.opends.server.types.AttributeType;
|
import org.opends.server.types.AttributeValue;
|
import org.opends.server.types.AttributeValues;
|
import org.opends.server.types.Attributes;
|
import org.opends.server.types.ByteString;
|
import org.opends.server.types.ConfigChangeResult;
|
import org.opends.server.types.Control;
|
import org.opends.server.types.DN;
|
import org.opends.server.types.DebugLogLevel;
|
import org.opends.server.types.DereferencePolicy;
|
import org.opends.server.types.DirectoryException;
|
import org.opends.server.types.Entry;
|
import org.opends.server.types.ExistingFileBehavior;
|
import org.opends.server.types.LDAPException;
|
import org.opends.server.types.LDIFExportConfig;
|
import org.opends.server.types.LDIFImportConfig;
|
import org.opends.server.types.Modification;
|
import org.opends.server.types.ModificationType;
|
import org.opends.server.types.ObjectClass;
|
import org.opends.server.types.Operation;
|
import org.opends.server.types.RDN;
|
import org.opends.server.types.RawModification;
|
import org.opends.server.types.ResultCode;
|
import org.opends.server.types.Schema;
|
import org.opends.server.types.SearchFilter;
|
import org.opends.server.types.SearchResultEntry;
|
import org.opends.server.types.SearchResultReference;
|
import org.opends.server.types.SearchScope;
|
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.types.operation.PreOperationOperation;
|
import org.opends.server.util.LDIFReader;
|
import org.opends.server.workflowelement.externalchangelog.ECLWorkflowElement;
|
import org.opends.server.workflowelement.localbackend.*;
|
|
/**
|
* This class implements the bulk part of the.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.
|
*/
|
public class LDAPReplicationDomain extends ReplicationDomain
|
implements ConfigurationChangeListener<ReplicationDomainCfg>,
|
AlertGenerator
|
{
|
/**
|
* 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 ChangeNumber startingChangeNumber = null;
|
private ChangeNumber endChangeNumber = null;
|
|
public ScanSearchListener(
|
ChangeNumber startingChangeNumber,
|
ChangeNumber endChangeNumber)
|
{
|
this.startingChangeNumber = startingChangeNumber;
|
this.endChangeNumber = endChangeNumber;
|
}
|
|
public void handleInternalSearchEntry(
|
InternalSearchOperation searchOperation, SearchResultEntry searchEntry)
|
throws DirectoryException
|
{
|
// Build the list of Operations that happened on this entry
|
// after startingChangeNumber and before endChangeNumber and
|
// add them to the replayOperations list
|
Iterable<FakeOperation> updates =
|
Historical.generateFakeOperations(searchEntry);
|
|
for (FakeOperation op : updates)
|
{
|
ChangeNumber cn = op.getChangeNumber();
|
if ((cn.newer(startingChangeNumber)) && (cn.older(endChangeNumber)))
|
{
|
synchronized (replayOperations)
|
{
|
replayOperations.put(cn, op);
|
}
|
}
|
}
|
}
|
|
public void handleInternalSearchReference(
|
InternalSearchOperation searchOperation,
|
SearchResultReference searchReference) throws DirectoryException
|
{
|
// Nothing to do.
|
}
|
}
|
|
/**
|
* The fully-qualified name of this class.
|
*/
|
private static final String CLASS_NAME =
|
"org.opends.server.replication.plugin.LDAPReplicationDomain";
|
|
/**
|
* 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";
|
|
/**
|
* The tracer object for the debug logger.
|
*/
|
private static final DebugTracer TRACER = getTracer();
|
|
// The update to replay message queue where the listener thread is going to
|
// push incoming update messages.
|
private final BlockingQueue<UpdateToReplay> updateToReplayQueue;
|
private final AtomicInteger numResolvedNamingConflicts = new AtomicInteger();
|
private final AtomicInteger numResolvedModifyConflicts = new AtomicInteger();
|
private final AtomicInteger numUnresolvedNamingConflicts =
|
new AtomicInteger();
|
private final int debugCount = 0;
|
private final PersistentServerState state;
|
private int numReplayedPostOpCalled = 0;
|
|
private long generationId = -1;
|
private boolean generationIdSavedStatus = false;
|
|
private final ChangeNumberGenerator generator;
|
|
/**
|
* This object is used to store the list of update currently being
|
* done on the local database.
|
* Is 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;
|
|
/**
|
* 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 to make sure that the ServerState is
|
* not updated too early.
|
*/
|
private final RemotePendingChanges remotePendingChanges;
|
|
private final int serverId;
|
|
private final DN baseDn;
|
|
private boolean shutdown = false;
|
|
private final InternalClientConnection conn =
|
InternalClientConnection.getRootConnection();
|
|
private boolean solveConflictFlag = true;
|
|
private boolean disabled = false;
|
private boolean stateSavingDisabled = false;
|
|
// This list is used to temporary store operations that needs
|
// to be replayed at session establishment time.
|
private final TreeMap<ChangeNumber, FakeOperation> replayOperations =
|
new TreeMap<ChangeNumber, FakeOperation>();
|
|
/**
|
* The isolation policy that this domain is going to use.
|
* This field describes the behavior of the domain when an update is
|
* attempted and the domain could not connect to any Replication Server.
|
* Possible values are accept-updates or deny-updates, but other values
|
* may be added in the future.
|
*/
|
private IsolationPolicy isolationpolicy;
|
|
/**
|
* The DN of the configuration entry of this domain.
|
*/
|
private final DN configDn;
|
private ExternalChangelogDomain eclDomain;
|
|
/**
|
* A boolean indicating if the thread used to save the persistentServerState
|
* is terminated.
|
*/
|
private boolean done = true;
|
|
private ServerStateFlush flushThread;
|
|
/**
|
* The attribute name used to store the generation id in the backend.
|
*/
|
protected static final String REPLICATION_GENERATION_ID =
|
"ds-sync-generation-id";
|
/**
|
* The attribute name used to store the fractional include configuration in
|
* the backend.
|
*/
|
public static final String REPLICATION_FRACTIONAL_INCLUDE =
|
"ds-sync-fractional-include";
|
/**
|
* The attribute name used to store the fractional exclude configuration in
|
* the backend.
|
*/
|
public static final String REPLICATION_FRACTIONAL_EXCLUDE =
|
"ds-sync-fractional-exclude";
|
|
/**
|
* Fractional replication variables.
|
*/
|
|
// Holds the fractional configuration for this domain, if any.
|
private FractionalConfig fractionalConfig = null;
|
|
/**
|
* 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
|
};
|
|
/**
|
* 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 force_bad_data_set = false;
|
|
/**
|
* 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.
|
*/
|
private boolean followImport = true;
|
|
/**
|
* The message id to be used when an import is stopped with error by
|
* the fractional replication ldif import plugin.
|
*/
|
private int importErrorMessageId = -1;
|
/**
|
* Message type for ERR_FULL_UPDATE_IMPORT_FRACTIONAL_BAD_REMOTE.
|
*/
|
public static final int IMPORT_ERROR_MESSAGE_BAD_REMOTE = 1;
|
/**
|
* Message type for ERR_FULL_UPDATE_IMPORT_FRACTIONAL_REMOTE_IS_FRACTIONAL.
|
*/
|
public 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 thread that periodically saves the ServerState of this
|
* LDAPReplicationDomain in the database.
|
*/
|
private class ServerStateFlush extends DirectoryThread
|
{
|
protected ServerStateFlush()
|
{
|
super("Replication State Saver for server id " +
|
serverId + " and domain " + baseDn.toString());
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
@Override
|
public void run()
|
{
|
done = false;
|
|
while (shutdown == false)
|
{
|
try
|
{
|
synchronized (this)
|
{
|
this.wait(1000);
|
if (!disabled && !stateSavingDisabled )
|
{
|
// save the ServerState
|
state.save();
|
}
|
}
|
} catch (InterruptedException e)
|
{ }
|
}
|
state.save();
|
|
done = true;
|
}
|
}
|
|
/**
|
* The thread that is responsible to update the RS to which this domain is
|
* connected in case it is late and there is no RS which is up to date.
|
*/
|
private class RSUpdater extends DirectoryThread
|
{
|
private ChangeNumber startChangeNumber;
|
protected RSUpdater(ChangeNumber replServerMaxChangeNumber)
|
{
|
super("Replication Server Updater for server id " +
|
serverId + " and domain " + baseDn.toString());
|
this.startChangeNumber = replServerMaxChangeNumber;
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
@Override
|
public void run()
|
{
|
// Replication server is missing some of our changes: let's
|
// send them to him.
|
Message message = DEBUG_GOING_TO_SEARCH_FOR_CHANGES.get();
|
logError(message);
|
|
/*
|
* Get all the changes that have not been seen by this
|
* replication server and publish them.
|
*/
|
try
|
{
|
if (buildAndPublishMissingChanges(startChangeNumber, broker))
|
{
|
message = DEBUG_CHANGES_SENT.get();
|
logError(message);
|
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.
|
*/
|
message = ERR_CANNOT_RECOVER_CHANGES.get(
|
baseDn.toNormalizedString());
|
logError(message);
|
}
|
} 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.
|
*/
|
message = ERR_CANNOT_RECOVER_CHANGES.get(
|
baseDn.toNormalizedString());
|
logError(message);
|
}
|
finally
|
{
|
broker.setRecoveryRequired(false);
|
}
|
}
|
}
|
|
|
/**
|
* Creates a new ReplicationDomain using configuration from configEntry.
|
*
|
* @param configuration The configuration of this ReplicationDomain.
|
* @param updateToReplayQueue The queue for update messages to replay.
|
* @throws ConfigException In case of invalid configuration.
|
*/
|
public LDAPReplicationDomain(ReplicationDomainCfg configuration,
|
BlockingQueue<UpdateToReplay> updateToReplayQueue)
|
throws ConfigException
|
{
|
super(configuration.getBaseDN().toNormalizedString(),
|
configuration.getServerId());
|
|
/**
|
* The time in milliseconds between heartbeats from the replication
|
* server. Zero means heartbeats are off.
|
*/
|
long heartbeatInterval = 0;
|
|
// Read the configuration parameters.
|
Set<String> replicationServers = configuration.getReplicationServer();
|
serverId = configuration.getServerId();
|
baseDn = configuration.getBaseDN();
|
int window = configuration.getWindowSize();
|
heartbeatInterval = configuration.getHeartbeatInterval();
|
isolationpolicy = configuration.getIsolationPolicy();
|
configDn = configuration.dn();
|
this.updateToReplayQueue = updateToReplayQueue;
|
|
// Get assured configuration
|
readAssuredConfig(configuration, false);
|
|
// Get fractional configuration
|
fractionalConfig = new FractionalConfig(baseDn);
|
readFractionalConfig(configuration, false);
|
|
setGroupId((byte)configuration.getGroupId());
|
setURLs(configuration.getReferralsUrl());
|
|
createECLDomainCfg(configuration);
|
|
/*
|
* 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.
|
*/
|
if (baseDn.compareTo(DirectoryServer.getSchemaDN()) == 0)
|
{
|
solveConflictFlag = false;
|
}
|
else
|
{
|
solveConflictFlag = configuration.isSolveConflicts();
|
}
|
|
Backend backend = retrievesBackend(baseDn);
|
if (backend == null)
|
{
|
throw new ConfigException(ERR_SEARCHING_DOMAIN_BACKEND.get(
|
baseDn.toNormalizedString()));
|
}
|
|
try
|
{
|
generationId = loadGenerationId();
|
}
|
catch (DirectoryException e)
|
{
|
logError(ERR_LOADING_GENERATION_ID.get(
|
baseDn.toNormalizedString(), e.getLocalizedMessage()));
|
}
|
|
/*
|
* Create a new Persistent Server State that will be used to store
|
* the last ChangeNmber seen from all LDAP servers in the topology.
|
*/
|
state = new PersistentServerState(baseDn, serverId, getServerState());
|
|
/* Check if a ReplicaUpdateVector entry is present
|
* if so, and no state is already initialized
|
* translate the ruv into a serverState and
|
* a generationId
|
*/
|
Long compatGenId = state.checkRUVCompat();
|
if (compatGenId != null)
|
{
|
generationId = compatGenId;
|
saveGenerationId(generationId);
|
}
|
|
/*
|
* ChangeNumberGenerator is used to create new unique ChangeNumbers
|
* for each operation done on this replication domain.
|
*
|
* The generator time is adjusted to the time of the last CN received from
|
* remote other servers.
|
*/
|
generator = getGenerator();
|
|
pendingChanges =
|
new PendingChanges(generator, this);
|
|
startPublishService(replicationServers, window, heartbeatInterval,
|
configuration.getChangetimeHeartbeatInterval());
|
|
remotePendingChanges = new RemotePendingChanges(getServerState());
|
|
// listen for changes on the configuration
|
configuration.addChangeListener(this);
|
|
// register as an AltertGenerator
|
DirectoryServer.registerAlertGenerator(this);
|
}
|
|
/**
|
* Gets and stores the assured replication configuration parameters. Returns
|
* a boolean indicating if the passed configuration has changed compared to
|
* previous values and the changes require a reconnection.
|
* @param configuration The configuration object
|
* @param allowReconnection Tells if one must reconnect if significant changes
|
* occurred
|
*/
|
private void readAssuredConfig(ReplicationDomainCfg configuration,
|
boolean allowReconnection)
|
{
|
boolean needReconnection = false;
|
|
byte newSdLevel = (byte) configuration.getAssuredSdLevel();
|
if ((isAssured() && (getAssuredMode() == AssuredMode.SAFE_DATA_MODE)) &&
|
(newSdLevel != getAssuredSdLevel()))
|
{
|
needReconnection = true;
|
}
|
|
AssuredType newAssuredType = configuration.getAssuredType();
|
switch (newAssuredType)
|
{
|
case NOT_ASSURED:
|
if (isAssured())
|
{
|
needReconnection = true;
|
}
|
break;
|
case SAFE_DATA:
|
if (!isAssured() ||
|
(isAssured() && (getAssuredMode() == AssuredMode.SAFE_READ_MODE)))
|
{
|
needReconnection = true;
|
}
|
break;
|
case SAFE_READ:
|
if (!isAssured() ||
|
(isAssured() && (getAssuredMode() == AssuredMode.SAFE_DATA_MODE)))
|
{
|
needReconnection = true;
|
}
|
break;
|
}
|
|
// Disconnect if required: changing configuration values before
|
// disconnection would make assured replication used immediately and
|
// disconnection could cause some timeouts error.
|
if (needReconnection && allowReconnection)
|
disableService();
|
|
switch (newAssuredType)
|
{
|
case NOT_ASSURED:
|
setAssured(false);
|
break;
|
case SAFE_DATA:
|
setAssured(true);
|
setAssuredMode(AssuredMode.SAFE_DATA_MODE);
|
break;
|
case SAFE_READ:
|
setAssured(true);
|
setAssuredMode(AssuredMode.SAFE_READ_MODE);
|
break;
|
}
|
setAssuredSdLevel(newSdLevel);
|
setAssuredTimeout(configuration.getAssuredTimeout());
|
|
// Reconnect if required
|
if (needReconnection && allowReconnection)
|
enableService();
|
}
|
|
/**
|
* 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.
|
*/
|
public void setImportErrorMessageId(int importErrorMessageId)
|
{
|
this.importErrorMessageId = importErrorMessageId;
|
}
|
|
/**
|
* Sets the boolean telling if the online import currently in progress should
|
* continue.
|
* @param followImport The boolean telling if the online import currently in
|
* progress should continue.
|
*/
|
public void setFollowImport(boolean followImport)
|
{
|
this.followImport = followImport;
|
}
|
|
/**
|
* 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 = null;
|
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
|
Message message = NOTE_ERR_FRACTIONAL.get(baseDn.toString(),
|
e.getLocalizedMessage());
|
logError(message);
|
return;
|
}
|
|
/**
|
* Is there any change in fractional configuration ?
|
*/
|
|
// Compute current configuration
|
boolean needReconnection = false;
|
try
|
{
|
needReconnection = !FractionalConfig.
|
isFractionalConfigEquivalent(fractionalConfig, newFractionalConfig);
|
}
|
catch (ConfigException e)
|
{
|
// Should not happen
|
Message message = NOTE_ERR_FRACTIONAL.get(baseDn.toString(),
|
e.getLocalizedMessage());
|
logError(message);
|
return;
|
}
|
|
// Disable service if configuration changed
|
if (needReconnection && allowReconnection)
|
disableService();
|
|
// Set new configuration
|
int newFractionalMode = newFractionalConfig.fractionalConfigToInt();
|
fractionalConfig.setFractional(newFractionalMode !=
|
FractionalConfig.NOT_FRACTIONAL);
|
if (fractionalConfig.isFractional())
|
{
|
// Set new fractional configuration values
|
if (newFractionalMode == FractionalConfig.EXCLUSIVE_FRACTIONAL)
|
fractionalConfig.setFractionalExclusive(true);
|
else
|
fractionalConfig.setFractionalExclusive(false);
|
fractionalConfig.setFractionalSpecificClassesAttributes(
|
newFractionalConfig.getFractionalSpecificClassesAttributes());
|
fractionalConfig.setFractionalAllClassesAttributes(
|
newFractionalConfig.fractionalAllClassesAttributes);
|
} else
|
{
|
// Reset default values
|
fractionalConfig.setFractionalExclusive(true);
|
fractionalConfig.setFractionalSpecificClassesAttributes(
|
new HashMap<String, List<String>>());
|
fractionalConfig.setFractionalAllClassesAttributes(
|
new ArrayList<String>());
|
}
|
|
// Reconnect if required
|
if (needReconnection && allowReconnection)
|
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 (debugEnabled())
|
TRACER.debugInfo(
|
"Attempt to read the potential fractional config in domain root " +
|
"entry " + baseDn.toString());
|
|
ByteString asn1BaseDn = ByteString.valueOf(baseDn.toString());
|
boolean found = false;
|
LDAPFilter filter;
|
try
|
{
|
filter = LDAPFilter.decode("objectclass=*");
|
} catch (LDAPException e)
|
{
|
// Can not happen
|
return false;
|
}
|
|
/*
|
* Search the domain root entry that is used to save the generation id
|
*/
|
|
InternalSearchOperation search = null;
|
LinkedHashSet<String> attributes = new LinkedHashSet<String>(1);
|
attributes.add(REPLICATION_GENERATION_ID);
|
attributes.add(REPLICATION_FRACTIONAL_EXCLUDE);
|
attributes.add(REPLICATION_FRACTIONAL_INCLUDE);
|
search = conn.processSearch(asn1BaseDn,
|
SearchScope.BASE_OBJECT,
|
DereferencePolicy.DEREF_ALWAYS, 0, 0, false,
|
filter, attributes);
|
if (((search.getResultCode() != ResultCode.SUCCESS)) &&
|
((search.getResultCode() != ResultCode.NO_SUCH_OBJECT)))
|
{
|
Message message = ERR_SEARCHING_GENERATION_ID.get(
|
search.getResultCode().getResultCodeName() + " " +
|
search.getErrorMessage(),
|
baseDn.toString());
|
logError(message);
|
return false;
|
}
|
|
SearchResultEntry resultEntry = null;
|
if (search.getResultCode() == ResultCode.SUCCESS)
|
{
|
LinkedList<SearchResultEntry> result = search.getSearchEntries();
|
resultEntry = result.getFirst();
|
if (resultEntry != null)
|
{
|
AttributeType synchronizationGenIDType =
|
DirectoryServer.getAttributeType(REPLICATION_GENERATION_ID);
|
List<Attribute> attrs =
|
resultEntry.getAttribute(synchronizationGenIDType);
|
if (attrs != null)
|
{
|
Attribute attr = attrs.get(0);
|
if (attr.size() > 1)
|
{
|
Message message = ERR_LOADING_GENERATION_ID.get(
|
baseDn.toString(), "#Values=" + attr.size() +
|
" Must be exactly 1 in entry " +
|
resultEntry.toLDIFString());
|
logError(message);
|
} else if (attr.size() == 1)
|
{
|
found = true;
|
}
|
}
|
}
|
}
|
|
if (!found)
|
{
|
// 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
|
if (fractionalConfig.isFractional())
|
{
|
return false;
|
}
|
else
|
{
|
return true;
|
}
|
}
|
|
/*
|
* Now extract fractional configuration if any
|
*/
|
|
Iterator<String> exclIt = null;
|
AttributeType fractionalExcludeType =
|
DirectoryServer.getAttributeType(REPLICATION_FRACTIONAL_EXCLUDE);
|
List<Attribute> exclAttrs =
|
resultEntry.getAttribute(fractionalExcludeType);
|
if (exclAttrs != null)
|
{
|
Attribute exclAttr = exclAttrs.get(0);
|
if (exclAttr != null)
|
{
|
exclIt = new AttributeValueStringIterator(exclAttr.iterator());
|
}
|
}
|
|
Iterator<String> inclIt = null;
|
AttributeType fractionalIncludeType =
|
DirectoryServer.getAttributeType(REPLICATION_FRACTIONAL_INCLUDE);
|
List<Attribute> inclAttrs =
|
resultEntry.getAttribute(fractionalIncludeType);
|
if (inclAttrs != null)
|
{
|
Attribute inclAttr = inclAttrs.get(0);
|
if (inclAttr != null)
|
{
|
inclIt = new AttributeValueStringIterator(inclAttr.iterator());
|
}
|
}
|
|
// Compare backend and local fractional configuration
|
return isFractionalConfigConsistent(fractionalConfig, exclIt, inclIt);
|
}
|
|
/**
|
* 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<String> exclIt,
|
Iterator<String> inclIt)
|
{
|
/*
|
* Parse fractional configuration stored in passed fractional configuration
|
* attributes values
|
*/
|
|
Map<String, List<String>> storedFractionalSpecificClassesAttributes =
|
new HashMap<String, List<String>>();
|
List<String> storedFractionalAllClassesAttributes = new ArrayList<String>();
|
|
int storedFractionalMode = FractionalConfig.NOT_FRACTIONAL;
|
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
|
Message message = NOTE_ERR_FRACTIONAL.get(
|
fractionalConfig.getBaseDn().toString(), e.getLocalizedMessage());
|
logError(message);
|
return false;
|
}
|
|
FractionalConfig storedFractionalConfig = new FractionalConfig(
|
fractionalConfig.getBaseDn());
|
storedFractionalConfig.setFractional(storedFractionalMode !=
|
FractionalConfig.NOT_FRACTIONAL);
|
// Set stored fractional configuration values
|
if (storedFractionalConfig.isFractional())
|
{
|
if (storedFractionalMode == FractionalConfig.EXCLUSIVE_FRACTIONAL)
|
storedFractionalConfig.setFractionalExclusive(true);
|
else
|
storedFractionalConfig.setFractionalExclusive(false);
|
}
|
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
|
Message message = NOTE_ERR_FRACTIONAL.get(
|
fractionalConfig.getBaseDn().toString(), e.getLocalizedMessage());
|
logError(message);
|
return false;
|
}
|
}
|
|
/**
|
* Utility class to have get a string iterator from an AtributeValue iterator.
|
* Assuming the attribute values are strings.
|
*/
|
public static class AttributeValueStringIterator implements Iterator<String>
|
{
|
private Iterator<AttributeValue> attrValIt = null;
|
|
/**
|
* Creates a new AttributeValueStringIterator object.
|
* @param attrValIt The underlying attribute iterator to use, assuming
|
* internal values are strings.
|
*/
|
public AttributeValueStringIterator(Iterator<AttributeValue> attrValIt)
|
{
|
this.attrValIt = attrValIt;
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
public boolean hasNext()
|
{
|
return attrValIt.hasNext();
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
public String next()
|
{
|
return attrValIt.next().getValue().toString();
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
// Should not be needed anyway
|
public void remove()
|
{
|
attrValIt.remove();
|
}
|
}
|
|
/**
|
* Compare 2 attribute lists and returns true if they are equivalent.
|
* @param attributes1 First attribute list to compare.
|
* @param attributes2 Second attribute list to compare.
|
* @return True if both attribute lists are equivalent.
|
* @throws ConfigException If some attributes could not be retrieved from the
|
* schema.
|
*/
|
private static boolean isAttributeListEquivalent(
|
List<String> attributes1, List<String> attributes2) throws ConfigException
|
{
|
// Compare all classes attributes
|
if (attributes1.size() != attributes2.size())
|
return false;
|
|
// Check consistency of all classes attributes
|
Schema schema = DirectoryServer.getSchema();
|
/*
|
* For each attribute in attributes1, check there is the matching
|
* one in attributes2.
|
*/
|
for (String attributName1 : attributes1)
|
{
|
// Get attribute from attributes1
|
AttributeType attributeType1 = schema.getAttributeType(attributName1);
|
if (attributeType1 == null)
|
{
|
throw new ConfigException(
|
NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_ATTRIBUTE_TYPE.get(attributName1));
|
}
|
// Look for matching one in attributes2
|
boolean foundAttribute = false;
|
for (String attributName2 : attributes2)
|
{
|
AttributeType attributeType2 = schema.getAttributeType(attributName2);
|
if (attributeType2 == null)
|
{
|
throw new ConfigException(
|
NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_ATTRIBUTE_TYPE.get(
|
attributName2));
|
}
|
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<String, List<String>> newFractionalSpecificClassesAttributes =
|
newFractionalConfig.getFractionalSpecificClassesAttributes();
|
List<String> 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 = DirectoryServer.getSchema();
|
int fractionalMode = newFractionalConfig.fractionalConfigToInt();
|
for (String className : newFractionalSpecificClassesAttributes.keySet())
|
{
|
// Does the class exist ?
|
ObjectClass fractionalClass = schema.getObjectClass(
|
className.toLowerCase());
|
if (fractionalClass == null)
|
{
|
throw new ConfigException(
|
NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_OBJECT_CLASS.get(className));
|
}
|
|
boolean isExtensibleObjectClass = className.
|
equalsIgnoreCase("extensibleObject");
|
|
List<String> attributes =
|
newFractionalSpecificClassesAttributes.get(className);
|
|
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 != null)
|
{
|
// No more checking for the extensibleObject class
|
if (!isExtensibleObjectClass)
|
{
|
if (fractionalMode == FractionalConfig.EXCLUSIVE_FRACTIONAL)
|
{
|
// Exclusive mode : the attribute must be optional
|
if (!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
|
*/
|
public boolean fractionalFilterOperation(
|
PreOperationAddOperation addOperation, boolean performFiltering)
|
{
|
return fractionalRemoveAttributesFromEntry(fractionalConfig,
|
addOperation.getEntryDN().getRDN(), 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
|
*/
|
public boolean fractionalFilterOperation(
|
PreOperationModifyDNOperation modifyDNOperation, boolean performFiltering)
|
{
|
boolean inconsistentOperation = false;
|
|
// Quick exit if not called for analyze and
|
if (performFiltering)
|
{
|
if (modifyDNOperation.deleteOldRDN())
|
{
|
// The core will remove any occurence 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();
|
List<String> fractionalConcernedAttributes =
|
createFractionalConcernedAttrList(fractionalConfig,
|
concernedEntry.getObjectClasses().keySet());
|
|
boolean fractionalExclusive = fractionalConfig.isFractionalExclusive();
|
if ( fractionalExclusive && (fractionalConcernedAttributes.size() == 0) )
|
// No attributes to filter
|
return false;
|
|
/**
|
* Analyze the old and new rdn to see if they are some attributes to be
|
* removed: if the oldnRDN 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.
|
*/
|
|
RDN rdn = modifyDNOperation.getEntryDN().getRDN();
|
RDN newRdn = modifyDNOperation.getNewRDN();
|
|
// Go through each attribute of the old RDN
|
for (int i=0 ; i<rdn.getNumValues() ; i++)
|
{
|
AttributeType attributeType = rdn.getAttributeType(i);
|
boolean found = false;
|
// Is it present in the fractional attributes established list ?
|
for (String attrTypeStr : fractionalConcernedAttributes)
|
{
|
AttributeType attributeTypeFromList =
|
DirectoryServer.getAttributeType(attrTypeStr);
|
if (attributeTypeFromList.equals(attributeType))
|
{
|
found = true;
|
break;
|
}
|
}
|
boolean attributeToBeFiltered = ( (fractionalExclusive && found) ||
|
(!fractionalExclusive && !found) );
|
if (attributeToBeFiltered &&
|
!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:
|
// remove ourself 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<ObjectClass,String> classes, Map<AttributeType,
|
List<Attribute>> attributesMap, boolean performFiltering)
|
{
|
boolean hasSomeAttributesToFilter = false;
|
/*
|
* Prepare a list of attributes to be included/excluded according to the
|
* fractional replication configuration
|
*/
|
|
List<String> fractionalConcernedAttributes =
|
createFractionalConcernedAttrList(fractionalConfig, classes.keySet());
|
boolean fractionalExclusive = fractionalConfig.isFractionalExclusive();
|
if ( fractionalExclusive && (fractionalConcernedAttributes.size() == 0) )
|
return false; // No attributes to filter
|
|
// Prepare list of object classes of the added entry
|
Set<ObjectClass> 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
|
*/
|
Iterator<AttributeType> attributeTypes = attributesMap.keySet().iterator();
|
List<List<Attribute>> newRdnAttrLists = new ArrayList<List<Attribute>>();
|
List<AttributeType> rdnAttrTypes = new ArrayList<AttributeType>();
|
while (attributeTypes.hasNext())
|
{
|
AttributeType attributeType = attributeTypes.next();
|
|
// Only optional attributes may be removed
|
boolean isMandatoryAttribute = false;
|
for (ObjectClass objectClass : entryClasses)
|
{
|
if (objectClass.isRequired(attributeType))
|
{
|
isMandatoryAttribute = true;
|
break;
|
}
|
}
|
if (isMandatoryAttribute)
|
{
|
continue;
|
}
|
|
String attributeName = attributeType.getPrimaryName();
|
String attributeOid = attributeType.getOID();
|
// Do not remove an attribute if it is a prohibited one
|
if (((attributeName != null) &&
|
isFractionalProhibitedAttr(attributeName)) ||
|
isFractionalProhibitedAttr(attributeOid))
|
{
|
continue;
|
}
|
|
// Is the current attribute part of the established list ?
|
boolean foundAttribute =
|
fractionalConcernedAttributes.contains(attributeName.toLowerCase());
|
if (!foundAttribute)
|
{
|
foundAttribute =
|
fractionalConcernedAttributes.contains(attributeOid);
|
}
|
// Now remove the attribute if:
|
// - exclusive mode and attribute is in configuration
|
// - inclusive mode and attribute is not in configuration
|
if ((foundAttribute && fractionalExclusive) ||
|
(!foundAttribute && !fractionalExclusive))
|
{
|
if (performFiltering)
|
{
|
// 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
|
// suppot remove so we have to create a new list, keeping only the
|
// attribute value which is the same as in the RDN
|
AttributeValue rdnAttributeValue =
|
entryRdn.getAttributeValue(attributeType);
|
List<Attribute> attrList = attributesMap.get(attributeType);
|
Iterator<Attribute> attrIt = attrList.iterator();
|
AttributeValue sameAttrValue = null;
|
// Locate the attribute value identical to the one in the RDN
|
while(attrIt.hasNext())
|
{
|
Attribute attr = attrIt.next();
|
if (attr.contains(rdnAttributeValue))
|
{
|
Iterator<AttributeValue> attrValues = attr.iterator();
|
while(attrValues.hasNext())
|
{
|
AttributeValue attrValue = attrValues.next();
|
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 atribute list
|
List<Attribute> newRdnAttrList = new ArrayList<Attribute>();
|
AttributeBuilder attrBuilder =
|
new AttributeBuilder(attributeType);
|
attrBuilder.add(sameAttrValue);
|
newRdnAttrList.add(attrBuilder.toAttribute());
|
newRdnAttrLists.add(newRdnAttrList);
|
// 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.
|
attributeTypes.remove();
|
hasSomeAttributesToFilter = true;
|
}
|
}
|
else
|
{
|
// The call was just to check : at least one attribute to filter
|
// found, return immediatly the answer;
|
return true;
|
}
|
}
|
}
|
// Now overwrite the attribute values for the attribute types present in the
|
// RDN, if there are some filtered attributes in the RDN
|
int index = 0;
|
for (index = 0 ; index < rdnAttrTypes.size() ; index++)
|
{
|
attributesMap.put(rdnAttrTypes.get(index), newRdnAttrLists.get(index));
|
}
|
return hasSomeAttributesToFilter;
|
}
|
|
/**
|
* 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 List<String> createFractionalConcernedAttrList(
|
FractionalConfig fractionalConfig, Set<ObjectClass> entryObjectClasses)
|
{
|
/*
|
* Is the concerned entry of a type concerned by fractional replication
|
* configuration ? If yes, add the matching attribute names to a list of
|
* attributes to take into account for filtering
|
* (inclusive or exclusive mode)
|
*/
|
|
List<String> fractionalConcernedAttributes = new ArrayList<String>();
|
|
// Get object classes the entry matches
|
List<String> fractionalAllClassesAttributes =
|
fractionalConfig.getFractionalAllClassesAttributes();
|
Map<String, List<String>> fractionalSpecificClassesAttributes =
|
fractionalConfig.getFractionalSpecificClassesAttributes();
|
|
Set<String> fractionalClasses =
|
fractionalSpecificClassesAttributes.keySet();
|
for (ObjectClass entryObjectClass : entryObjectClasses)
|
{
|
for(String fractionalClass : fractionalClasses)
|
{
|
if (entryObjectClass.hasNameOrOID(fractionalClass.toLowerCase()))
|
{
|
List<String> attrList =
|
fractionalSpecificClassesAttributes.get(fractionalClass);
|
for(String attr : attrList)
|
{
|
// Avoid duplicate attributes (from 2 inheriting classes for
|
// instance)
|
if (!fractionalConcernedAttributes.contains(attr))
|
{
|
fractionalConcernedAttributes.add(attr);
|
}
|
}
|
}
|
}
|
}
|
|
/*
|
* Add to the list any attribute which is class independent
|
*/
|
for (String attr : fractionalAllClassesAttributes)
|
{
|
if (!fractionalConcernedAttributes.contains(attr))
|
{
|
fractionalConcernedAttributes.add(attr);
|
}
|
}
|
|
return fractionalConcernedAttributes;
|
}
|
|
/**
|
* 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
|
*/
|
public int fractionalFilterOperation(PreOperationModifyOperation
|
modifyOperation, boolean performFiltering)
|
{
|
int result = FRACTIONAL_HAS_NO_FRACTIONAL_FILTERED_ATTRIBUTES;
|
/*
|
* Prepare a list of attributes to be included/excluded according to the
|
* fractional replication configuration
|
*/
|
|
Entry modifiedEntry = modifyOperation.getCurrentEntry();
|
List<String> fractionalConcernedAttributes =
|
createFractionalConcernedAttrList(fractionalConfig,
|
modifiedEntry.getObjectClasses().keySet());
|
boolean fractionalExclusive = fractionalConfig.isFractionalExclusive();
|
if ( fractionalExclusive && (fractionalConcernedAttributes.size() == 0) )
|
// 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 = null;
|
try
|
{
|
entryToModify = DirectoryServer.getEntry(entryToModifyDn);
|
}
|
catch(DirectoryException e)
|
{
|
Message message = NOTE_ERR_FRACTIONAL.get(baseDn.toString(),
|
e.getLocalizedMessage());
|
logError(message);
|
return FRACTIONAL_HAS_NO_FRACTIONAL_FILTERED_ATTRIBUTES;
|
}
|
Set<ObjectClass> 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
|
*/
|
|
List<Modification> mods = modifyOperation.getModifications();
|
Iterator<Modification> modsIt = mods.iterator();
|
while (modsIt.hasNext())
|
{
|
Modification mod = modsIt.next();
|
Attribute attr = mod.getAttribute();
|
AttributeType attrType = attr.getAttributeType();
|
// Fractional replication ignores operational attributes
|
if (!attrType.isOperational())
|
{
|
// Only optional attributes may be removed
|
boolean isMandatoryAttribute = false;
|
for (ObjectClass objectClass : entryClasses)
|
{
|
if (objectClass.isRequired(attrType))
|
{
|
isMandatoryAttribute = true;
|
break;
|
}
|
}
|
if (isMandatoryAttribute)
|
{
|
continue;
|
}
|
|
String attributeName = attrType.getPrimaryName();
|
String attributeOid = attrType.getOID();
|
// Do not remove an attribute if it is a prohibited one
|
if (((attributeName != null) &&
|
isFractionalProhibitedAttr(attributeName)) ||
|
isFractionalProhibitedAttr(attributeOid))
|
{
|
continue;
|
}
|
// Is the current attribute part of the established list ?
|
boolean foundAttribute =
|
fractionalConcernedAttributes.contains(attributeName.toLowerCase());
|
if (!foundAttribute)
|
{
|
foundAttribute =
|
fractionalConcernedAttributes.contains(attributeOid);
|
}
|
|
// Now remove the modification if:
|
// - exclusive mode and the concerned attribute is in configuration
|
// - inclusive mode and the concerned attribute is not in configuration
|
if ( (foundAttribute && fractionalExclusive) ||
|
(!foundAttribute && !fractionalExclusive) )
|
{
|
if (performFiltering)
|
{
|
// Found a modification to remove, remove it from the list.
|
modsIt.remove();
|
result = FRACTIONAL_HAS_FRACTIONAL_FILTERED_ATTRIBUTES;
|
if (mods.size() == 0)
|
{
|
// This operation must become a no-op as no more modification in
|
// it
|
return FRACTIONAL_BECOME_NO_OP;
|
}
|
}
|
else
|
{
|
// The call was just to check : at least one attribute to filter
|
// found, return immediatly the answer;
|
return FRACTIONAL_HAS_FRACTIONAL_FILTERED_ATTRIBUTES;
|
}
|
}
|
}
|
}
|
|
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 (followImport)
|
{
|
// Ok, next entry is allowed to be received
|
return super.receiveEntryBytes();
|
} else
|
{
|
// 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
|
Message msg = null;
|
switch (importErrorMessageId)
|
{
|
case IMPORT_ERROR_MESSAGE_BAD_REMOTE:
|
msg = NOTE_ERR_FULL_UPDATE_IMPORT_FRACTIONAL_BAD_REMOTE.get(
|
baseDn.toString(), Integer.toString(ieContext.getImportSource()));
|
break;
|
case IMPORT_ERROR_MESSAGE_REMOTE_IS_FRACTIONAL:
|
msg = NOTE_ERR_FULL_UPDATE_IMPORT_FRACTIONAL_REMOTE_IS_FRACTIONAL.get(
|
baseDn.toString(), Integer.toString(ieContext.getImportSource()));
|
break;
|
}
|
ieContext.setException(
|
new DirectoryException(ResultCode.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) throws DirectoryException
|
{
|
if ((target == RoutableMsg.ALL_SERVERS) && fractionalConfig.isFractional())
|
{
|
Message msg = NOTE_ERR_FRACTIONAL_FORBIDDEN_FULL_UPDATE_FRACTIONAL.get(
|
baseDn.toString(), Integer.toString(getServerId()));
|
throw new DirectoryException(ResultCode.UNWILLING_TO_PERFORM, msg);
|
} else
|
{
|
super.initializeRemote(target, requestorID, initTask);
|
}
|
}
|
|
/**
|
* Returns the base DN of this ReplicationDomain.
|
*
|
* @return The base DN of this ReplicationDomain
|
*/
|
public DN getBaseDN()
|
{
|
return baseDn;
|
}
|
|
/**
|
* Implement the handleConflictResolution phase of the deleteOperation.
|
*
|
* @param deleteOperation The deleteOperation.
|
* @return A SynchronizationProviderResult indicating if the operation
|
* can continue.
|
*/
|
public SynchronizationProviderResult handleConflictResolution(
|
PreOperationDeleteOperation deleteOperation)
|
{
|
if ((!deleteOperation.isSynchronizationOperation())
|
&& (!brokerIsConnected(deleteOperation)))
|
{
|
Message msg = ERR_REPLICATION_COULD_NOT_CONNECT.get(baseDn.toString());
|
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.getEntryUid();
|
String modifiedEntryUUID = Historical.getEntryUuid(deletedEntry);
|
if (!operationEntryUUID.equals(modifiedEntryUUID))
|
{
|
/*
|
* 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_OBJET.
|
* 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.
|
ChangeNumber changeNumber = generateChangeNumber(deleteOperation);
|
String modifiedEntryUUID = Historical.getEntryUuid(deletedEntry);
|
ctx = new DeleteContext(changeNumber, modifiedEntryUUID);
|
deleteOperation.setAttachment(SYNCHROCONTEXT, ctx);
|
|
synchronized (replayOperations)
|
{
|
int size = replayOperations.size();
|
if (size >= 10000)
|
{
|
replayOperations.remove(replayOperations.firstKey());
|
}
|
replayOperations.put(
|
changeNumber,
|
new FakeDelOperation(
|
deleteOperation.getEntryDN().toString(),
|
changeNumber,modifiedEntryUUID ));
|
}
|
|
}
|
|
return new SynchronizationProviderResult.ContinueProcessing();
|
}
|
|
/**
|
* Implement the handleConflictResolution phase of the addOperation.
|
*
|
* @param addOperation The AddOperation.
|
* @return A SynchronizationProviderResult indicating if the operation
|
* can continue.
|
*/
|
public SynchronizationProviderResult handleConflictResolution(
|
PreOperationAddOperation addOperation)
|
{
|
if ((!addOperation.isSynchronizationOperation())
|
&& (!brokerIsConnected(addOperation)))
|
{
|
Message msg = ERR_REPLICATION_COULD_NOT_CONNECT.get(baseDn.toString());
|
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))
|
{
|
StringBuilder sb = new StringBuilder();
|
addOperation.toString(sb);
|
Message msg = NOTE_ERR_FRACTIONAL_FORBIDDEN_OPERATION.get(
|
baseDn.toString(), sb.toString());
|
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.getEntryUid();
|
if (findEntryDN(uuid) != null)
|
{
|
return new SynchronizationProviderResult.StopProcessing(
|
ResultCode.CANCELED, 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 parentUid = ctx.getParentUid();
|
// root entry have no parent,
|
// there is no need to check for it.
|
if (parentUid != null)
|
{
|
// There is a potential of perfs improvement here
|
// if we could avoid the following parent entry retrieval
|
DN parentDnFromCtx = findEntryDN(ctx.getParentUid());
|
|
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);
|
}
|
else
|
{
|
DN entryDN = addOperation.getEntryDN();
|
DN parentDnFromEntryDn = entryDN.getParentDNInSuffix();
|
if ((parentDnFromEntryDn != null)
|
&& (!parentDnFromCtx.equals(parentDnFromEntryDn)))
|
{
|
// parentEntry has been renamed
|
// replication name conflict resolution is expected to fix that
|
// later in the flow
|
return new SynchronizationProviderResult.StopProcessing(
|
ResultCode.NO_SUCH_OBJECT, null);
|
}
|
}
|
}
|
}
|
return new SynchronizationProviderResult.ContinueProcessing();
|
}
|
|
/**
|
* Check that the broker associated to this ReplicationDomain has found
|
* a Replication Server and that this LDAP server is therefore able to
|
* process operations.
|
* If not set the ResultCode and the response message,
|
* interrupt the operation, and return false
|
*
|
* @param op The Operation that needs to be checked.
|
*
|
* @return true when it OK to process the Operation, false otherwise.
|
* When false is returned the resultCode and the reponse message
|
* is also set in the Operation.
|
*/
|
private boolean brokerIsConnected(PreOperationOperation op)
|
{
|
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.
|
*/
|
public SynchronizationProviderResult handleConflictResolution(
|
PreOperationModifyDNOperation modifyDNOperation)
|
{
|
if ((!modifyDNOperation.isSynchronizationOperation())
|
&& (!brokerIsConnected(modifyDNOperation)))
|
{
|
Message msg = ERR_REPLICATION_COULD_NOT_CONNECT.get(baseDn.toString());
|
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))
|
{
|
StringBuilder sb = new StringBuilder();
|
modifyDNOperation.toString(sb);
|
Message msg = NOTE_ERR_FRACTIONAL_FORBIDDEN_OPERATION.get(
|
baseDn.toString(), sb.toString());
|
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.
|
*/
|
String modifiedEntryUUID =
|
Historical.getEntryUuid(modifyDNOperation.getOriginalEntry());
|
if (!modifiedEntryUUID.equals(ctx.getEntryUid()))
|
{
|
/*
|
* 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_OBJET.
|
* 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 = findEntryId(modifyDNOperation.getNewSuperior());
|
if ((newParentId != null) && (ctx.getNewParentId() != null) &&
|
(!newParentId.equals(ctx.getNewParentId())))
|
{
|
return new SynchronizationProviderResult.StopProcessing(
|
ResultCode.NO_SUCH_OBJECT, null);
|
}
|
}
|
/*
|
* If the object has been renamed more recently than this
|
* operation, cancel the operation.
|
*/
|
Historical hist = Historical.load(modifyDNOperation.getOriginalEntry());
|
if (hist.AddedOrRenamedAfter(ctx.getChangeNumber()))
|
{
|
return new SynchronizationProviderResult.StopProcessing(
|
ResultCode.SUCCESS, null);
|
}
|
}
|
else
|
{
|
// There is no replication context attached to the operation
|
// so this is not a replication operation.
|
ChangeNumber changeNumber = generateChangeNumber(modifyDNOperation);
|
String newParentId = null;
|
if (modifyDNOperation.getNewSuperior() != null)
|
{
|
newParentId = findEntryId(modifyDNOperation.getNewSuperior());
|
}
|
|
Entry modifiedEntry = modifyDNOperation.getOriginalEntry();
|
String modifiedEntryUUID = Historical.getEntryUuid(modifiedEntry);
|
ctx = new ModifyDnContext(changeNumber, 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
|
*/
|
public SynchronizationProviderResult handleConflictResolution(
|
PreOperationModifyOperation modifyOperation)
|
{
|
if ((!modifyOperation.isSynchronizationOperation())
|
&& (!brokerIsConnected(modifyOperation)))
|
{
|
Message msg = ERR_REPLICATION_COULD_NOT_CONNECT.get(baseDn.toString());
|
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.SUCCESS, 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
|
StringBuilder sb = new StringBuilder();
|
modifyOperation.toString(sb);
|
Message msg = NOTE_ERR_FRACTIONAL_FORBIDDEN_OPERATION.get(
|
baseDn.toString(), sb.toString());
|
return new SynchronizationProviderResult.StopProcessing(
|
ResultCode.UNWILLING_TO_PERFORM, msg);
|
}
|
}
|
}
|
|
ModifyContext ctx =
|
(ModifyContext) modifyOperation.getAttachment(SYNCHROCONTEXT);
|
|
Entry modifiedEntry = modifyOperation.getModifiedEntry();
|
if (ctx == null)
|
{
|
// There is no replication context attached to the operation
|
// so this is not a replication operation.
|
ChangeNumber changeNumber = generateChangeNumber(modifyOperation);
|
String modifiedEntryUUID = Historical.getEntryUuid(modifiedEntry);
|
if (modifiedEntryUUID == null)
|
modifiedEntryUUID = modifyOperation.getEntryDN().toString();
|
ctx = new ModifyContext(changeNumber, modifiedEntryUUID);
|
modifyOperation.setAttachment(SYNCHROCONTEXT, ctx);
|
}
|
else
|
{
|
// This is a replayed operation, it is necessary to
|
// - check if the entry has been renamed
|
// - check for conflicts
|
String modifiedEntryUUID = ctx.getEntryUid();
|
String currentEntryUUID = Historical.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_OBJET.
|
* 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
|
*/
|
Historical historicalInformation = Historical.load(modifiedEntry);
|
modifyOperation.setAttachment(Historical.HISTORICAL,
|
historicalInformation);
|
|
if (historicalInformation.replayOperation(modifyOperation, modifiedEntry))
|
{
|
numResolvedModifyConflicts.incrementAndGet();
|
}
|
|
if (modifyOperation.getModifications().isEmpty())
|
{
|
/*
|
* This operation becomes a no-op due to conflict resolution
|
* stop the processing and send an OK result
|
*/
|
return new SynchronizationProviderResult.StopProcessing(
|
ResultCode.SUCCESS, null);
|
}
|
}
|
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 uid is not set when the handleConflict
|
* phase is called.
|
*
|
* @param addOperation The Add Operation.
|
*/
|
public void doPreOperation(PreOperationAddOperation addOperation)
|
{
|
AddContext ctx = new AddContext(generateChangeNumber(addOperation),
|
Historical.getEntryUuid(addOperation),
|
findEntryId(addOperation.getEntryDN().getParentDNInSuffix()));
|
|
addOperation.setAttachment(SYNCHROCONTEXT, ctx);
|
Historical.generateState(addOperation);
|
}
|
|
/**
|
* Check if an operation must be synchronized.
|
* Also update the list of pending changes and the server RUV
|
* @param op the operation
|
*/
|
public void synchronize(PostOperationOperation op)
|
{
|
ResultCode result = op.getResultCode();
|
if ((result == ResultCode.SUCCESS) && op.isSynchronizationOperation())
|
{
|
numReplayedPostOpCalled++;
|
}
|
LDAPUpdateMsg msg = null;
|
|
// Note that a failed non-replication operation might not have a change
|
// number.
|
ChangeNumber curChangeNumber = OperationContext.getChangeNumber(op);
|
|
if ((result == ResultCode.SUCCESS) && (!op.isSynchronizationOperation()))
|
{
|
// Generate a replication message for a successful non-replication
|
// operation.
|
msg = LDAPUpdateMsg.generateMsg(op);
|
|
if (msg == null)
|
{
|
/*
|
* This is an operation type that we do not know about
|
* It should never happen.
|
*/
|
pendingChanges.remove(curChangeNumber);
|
Message message =
|
ERR_UNKNOWN_TYPE.get(op.getOperationType().toString());
|
logError(message);
|
return;
|
}
|
}
|
|
if (result == ResultCode.SUCCESS)
|
{
|
try
|
{
|
if (op.isSynchronizationOperation())
|
{
|
remotePendingChanges.commit(curChangeNumber);
|
}
|
else
|
{
|
try
|
{
|
addEntryAttributesForCL(msg,op);
|
}
|
catch(Exception e)
|
{
|
TRACER.debugCaught(DebugLogLevel.ERROR, e);
|
}
|
// If assured replication is configured, this will prepare blocking
|
// mechanism. If assured replication is disabled, this returns
|
// immediately
|
prepareWaitForAckIfAssuredEnabled(msg);
|
try
|
{
|
msg.encode();
|
} catch (UnsupportedEncodingException e)
|
{
|
// will be caught at publish time.
|
}
|
pendingChanges.commitAndPushCommittedChanges(curChangeNumber, msg);
|
}
|
}
|
catch (NoSuchElementException e)
|
{
|
Message message = ERR_OPERATION_NOT_FOUND_IN_PENDING.get(
|
curChangeNumber.toString(), op.toString());
|
logError(message);
|
return;
|
}
|
|
if (generationIdSavedStatus != true)
|
{
|
this.saveGenerationId(generationId);
|
}
|
|
if (!op.isSynchronizationOperation())
|
{
|
// 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
|
Message errorMsg = NOTE_DS_ACK_TIMEOUT.get(getServiceID(),
|
Long.toString(getAssuredTimeout()), msg.toString());
|
logError(errorMsg);
|
}
|
}
|
}
|
else if (!op.isSynchronizationOperation())
|
{
|
// Remove an unsuccessful non-replication operation from the pending
|
// changes list.
|
if (curChangeNumber != null)
|
{
|
pendingChanges.remove(curChangeNumber);
|
pendingChanges.pushCommittedChanges();
|
}
|
}
|
}
|
|
/**
|
* Get the number of updates in the pending list.
|
*
|
* @return The number of updates in the pending list
|
*/
|
public int getPendingUpdatesCount()
|
{
|
if (pendingChanges != null)
|
return pendingChanges.size();
|
else
|
return 0;
|
}
|
|
/**
|
* get the number of updates replayed successfully by the replication.
|
*
|
* @return The number of updates replayed successfully
|
*/
|
public int getNumReplayedPostOpCalled()
|
{
|
return numReplayedPostOpCalled;
|
}
|
|
/**
|
* Get the debugCount.
|
*
|
* @return Returns the debugCount.
|
*/
|
public int getDebugCount()
|
{
|
return debugCount;
|
}
|
|
/**
|
* Delete this ReplicationDomain.
|
*/
|
public void delete()
|
{
|
shutdown();
|
removeECLDomainCfg();
|
}
|
|
/**
|
* Shutdown this ReplicationDomain.
|
*/
|
public void shutdown()
|
{
|
// stop the flush thread
|
shutdown = true;
|
|
// stop the thread in charge of flushing the ServerState.
|
if (flushThread != null)
|
{
|
synchronized (flushThread)
|
{
|
flushThread.notify();
|
}
|
}
|
|
DirectoryServer.deregisterAlertGenerator(this);
|
|
// stop the ReplicationDomain
|
stopDomain();
|
|
// wait for completion of the persistentServerState thread.
|
try
|
{
|
while (!done)
|
{
|
Thread.sleep(50);
|
}
|
} catch (InterruptedException e)
|
{
|
// stop waiting when interrupted.
|
}
|
}
|
|
/**
|
* Create and replay a synchronized Operation from an UpdateMsg.
|
*
|
* @param msg The UpdateMsg to be replayed.
|
*/
|
public void replay(LDAPUpdateMsg msg)
|
{
|
Operation op = null;
|
boolean done = false;
|
boolean dependency = false;
|
ChangeNumber changeNumber = null;
|
int retryCount = 10;
|
|
// Try replay the operation, then flush (replaying) any pending operation
|
// whose dependency has been replayed until no more left.
|
do
|
{
|
String replayErrorMsg = null;
|
try
|
{
|
op = msg.createOperation(conn);
|
dependency = remotePendingChanges.checkDependencies(op, msg);
|
|
while ((!dependency) && (!done) && (retryCount-- > 0))
|
{
|
// Try replay the operation
|
op.setInternalOperation(true);
|
op.setSynchronizationOperation(true);
|
changeNumber = OperationContext.getChangeNumber(op);
|
((AbstractOperation) op).run();
|
|
ResultCode result = op.getResultCode();
|
|
if (result != ResultCode.SUCCESS)
|
{
|
if (op instanceof ModifyOperation)
|
{
|
ModifyOperation newOp = (ModifyOperation) op;
|
dependency = remotePendingChanges.checkDependencies(newOp);
|
ModifyMsg modifyMsg = (ModifyMsg) msg;
|
done = solveNamingConflict(newOp, modifyMsg);
|
} else if (op instanceof DeleteOperation)
|
{
|
DeleteOperation newOp = (DeleteOperation) op;
|
dependency = remotePendingChanges.checkDependencies(newOp);
|
done = solveNamingConflict(newOp, msg);
|
} else if (op instanceof AddOperation)
|
{
|
AddOperation newOp = (AddOperation) op;
|
AddMsg addMsg = (AddMsg) msg;
|
dependency = remotePendingChanges.checkDependencies(newOp);
|
done = solveNamingConflict(newOp, addMsg);
|
} else if (op instanceof ModifyDNOperationBasis)
|
{
|
ModifyDNOperationBasis newOp = (ModifyDNOperationBasis) op;
|
done = solveNamingConflict(newOp, msg);
|
} else
|
{
|
done = true; // unknown type of operation ?!
|
}
|
if (done)
|
{
|
// 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
|
updateError(changeNumber);
|
}
|
else
|
{
|
/*
|
* Create a new operation as the ConflictResolution
|
* different operation.
|
*/
|
op = msg.createOperation(conn);
|
if (op instanceof DeleteOperation) {
|
op.addRequestControl(new SubtreeDeleteControl(false));
|
}
|
}
|
}
|
else
|
{
|
done = true;
|
}
|
}
|
|
if (!done && !dependency)
|
{
|
// Continue with the next change but the servers could now become
|
// inconsistent.
|
// Let the repair tool know about this.
|
Message message = ERR_LOOP_REPLAYING_OPERATION.get(op.toString(),
|
op.getErrorMessage().toString());
|
logError(message);
|
numUnresolvedNamingConflicts.incrementAndGet();
|
replayErrorMsg = message.toString();
|
updateError(changeNumber);
|
}
|
} catch (ASN1Exception e)
|
{
|
Message message = ERR_EXCEPTION_DECODING_OPERATION.get(
|
String.valueOf(msg) + stackTraceToSingleLineString(e));
|
logError(message);
|
replayErrorMsg = message.toString();
|
} catch (LDAPException e)
|
{
|
Message message = ERR_EXCEPTION_DECODING_OPERATION.get(
|
String.valueOf(msg) + stackTraceToSingleLineString(e));
|
logError(message);
|
replayErrorMsg = message.toString();
|
} catch (DataFormatException e)
|
{
|
Message message = ERR_EXCEPTION_DECODING_OPERATION.get(
|
String.valueOf(msg) + stackTraceToSingleLineString(e));
|
logError(message);
|
replayErrorMsg = message.toString();
|
} catch (Exception e)
|
{
|
if (changeNumber != null)
|
{
|
/*
|
* An Exception happened during the replay process.
|
* Continue with the next change but the servers will now start
|
* to be inconsistent.
|
* Let the repair tool know about this.
|
*/
|
Message message = ERR_EXCEPTION_REPLAYING_OPERATION.get(
|
stackTraceToSingleLineString(e), op.toString());
|
logError(message);
|
replayErrorMsg = message.toString();
|
updateError(changeNumber);
|
} else
|
{
|
Message message = ERR_EXCEPTION_DECODING_OPERATION.get(
|
String.valueOf(msg) + stackTraceToSingleLineString(e));
|
logError(message);
|
replayErrorMsg = message.toString();
|
}
|
} finally
|
{
|
if (!dependency)
|
{
|
processUpdateDone(msg, replayErrorMsg);
|
}
|
}
|
|
// 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();
|
|
// Prepare restart of loop
|
done = false;
|
dependency = false;
|
changeNumber = null;
|
retryCount = 10;
|
|
} while (msg != null);
|
}
|
|
/**
|
* 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 changeNumber the ChangeNumber of the operation with error.
|
*/
|
public void updateError(ChangeNumber changeNumber)
|
{
|
remotePendingChanges.commit(changeNumber);
|
}
|
|
/**
|
* Generate a new change number and insert it in the pending list.
|
*
|
* @param operation The operation for which the change number must be
|
* generated.
|
* @return The new change number.
|
*/
|
private ChangeNumber generateChangeNumber(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 uniqueID 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 findEntryId(DN dn)
|
{
|
if (dn == null)
|
return null;
|
try
|
{
|
InternalClientConnection conn =
|
InternalClientConnection.getRootConnection();
|
LinkedHashSet<String> attrs = new LinkedHashSet<String>(1);
|
attrs.add(ENTRYUIDNAME);
|
InternalSearchOperation search = conn.processSearch(dn,
|
SearchScope.BASE_OBJECT, DereferencePolicy.NEVER_DEREF_ALIASES,
|
0, 0, false,
|
SearchFilter.createFilterFromString("objectclass=*"),
|
attrs);
|
|
if (search.getResultCode() == ResultCode.SUCCESS)
|
{
|
LinkedList<SearchResultEntry> result = search.getSearchEntries();
|
if (!result.isEmpty())
|
{
|
SearchResultEntry resultEntry = result.getFirst();
|
if (resultEntry != null)
|
{
|
return Historical.getEntryUuid(resultEntry);
|
}
|
}
|
}
|
} catch (DirectoryException e)
|
{
|
// never happens because the filter is always valid.
|
}
|
return null;
|
}
|
|
/**
|
* find the current DN of an entry from its entry UUID.
|
*
|
* @param uuid the Entry Unique ID.
|
* @return The current DN of the entry or null if there is no entry with
|
* the specified UUID.
|
*/
|
private DN findEntryDN(String uuid)
|
{
|
try
|
{
|
InternalSearchOperation search = conn.processSearch(baseDn,
|
SearchScope.WHOLE_SUBTREE,
|
SearchFilter.createFilterFromString("entryuuid="+uuid));
|
if (search.getResultCode() == ResultCode.SUCCESS)
|
{
|
LinkedList<SearchResultEntry> result = search.getSearchEntries();
|
if (!result.isEmpty())
|
{
|
SearchResultEntry resultEntry = result.getFirst();
|
if (resultEntry != null)
|
{
|
return resultEntry.getDN();
|
}
|
}
|
}
|
} catch (DirectoryException e)
|
{
|
// never happens because the filter is always valid.
|
}
|
return null;
|
}
|
|
/**
|
* 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 true if the process is completed, false if it must continue..
|
*/
|
private boolean solveNamingConflict(ModifyOperation op,
|
ModifyMsg msg)
|
{
|
ResultCode result = op.getResultCode();
|
ModifyContext ctx = (ModifyContext) op.getAttachment(SYNCHROCONTEXT);
|
String entryUid = ctx.getEntryUid();
|
|
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(entryUid);
|
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.toString());
|
numResolvedNamingConflicts.incrementAndGet();
|
return false;
|
}
|
else
|
{
|
// This entry does not exist anymore.
|
// It has probably been deleted, stop the processing of this operation
|
numResolvedNamingConflicts.incrementAndGet();
|
return true;
|
}
|
}
|
else if (result == ResultCode.NOT_ALLOWED_ON_RDN)
|
{
|
DN currentDN = findEntryDN(entryUid);
|
RDN currentRDN = null;
|
if (currentDN != null)
|
{
|
currentRDN = currentDN.getRDN();
|
}
|
else
|
{
|
// The entry does not exist anymore.
|
numResolvedNamingConflicts.incrementAndGet();
|
return true;
|
}
|
|
// 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<Modification> mods = op.getModifications();
|
for (Modification mod : mods)
|
{
|
AttributeType modAttrType = mod.getAttribute().getAttributeType();
|
if ((mod.getModificationType() == ModificationType.DELETE) ||
|
(mod.getModificationType() == ModificationType.REPLACE))
|
{
|
if (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;
|
if (newAttribute == null)
|
{
|
attrBuilder = new AttributeBuilder(modAttrType);
|
}
|
else
|
{
|
attrBuilder = new AttributeBuilder(newAttribute);
|
}
|
attrBuilder.add(currentRDN.getAttributeValue(modAttrType));
|
mod.setAttribute(attrBuilder.toAttribute());
|
}
|
}
|
}
|
msg.setMods(mods);
|
numResolvedNamingConflicts.incrementAndGet();
|
return false;
|
}
|
else
|
{
|
// The other type of errors can not be caused by naming conflicts.
|
// Log a message for the repair tool.
|
Message message = ERR_ERROR_REPLAYING_OPERATION.get(
|
op.toString(), ctx.getChangeNumber().toString(),
|
result.toString(), op.getErrorMessage().toString());
|
logError(message);
|
return true;
|
}
|
}
|
|
/**
|
* 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 true if the process is completed, false if it must continue..
|
*/
|
private boolean solveNamingConflict(DeleteOperation op,
|
LDAPUpdateMsg msg)
|
{
|
ResultCode result = op.getResultCode();
|
DeleteContext ctx = (DeleteContext) op.getAttachment(SYNCHROCONTEXT);
|
String entryUid = ctx.getEntryUid();
|
|
if (result == ResultCode.NO_SUCH_OBJECT)
|
{
|
/*
|
* Find if the entry is still in the database.
|
*/
|
DN currentDn = findEntryDN(entryUid);
|
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 is nothing more to do.
|
*/
|
numResolvedNamingConflicts.incrementAndGet();
|
return true;
|
}
|
else
|
{
|
/*
|
* This entry has been renamed, replay the delete using its new DN.
|
*/
|
msg.setDn(currentDn.toString());
|
numResolvedNamingConflicts.incrementAndGet();
|
return false;
|
}
|
}
|
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(entryUid, op.getEntryDN(), op))
|
numUnresolvedNamingConflicts.incrementAndGet();
|
|
return false;
|
}
|
else
|
{
|
// The other type of errors can not be caused by naming conflicts.
|
// Log a message for the repair tool.
|
Message message = ERR_ERROR_REPLAYING_OPERATION.get(
|
op.toString(), ctx.getChangeNumber().toString(),
|
result.toString(), op.getErrorMessage().toString());
|
logError(message);
|
return true;
|
}
|
}
|
|
/**
|
* 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 true if the process is completed, false if it must continue.
|
* @throws Exception When the operation is not valid.
|
*/
|
private boolean solveNamingConflict(ModifyDNOperation op,
|
LDAPUpdateMsg msg) throws Exception
|
{
|
ResultCode result = op.getResultCode();
|
ModifyDnContext ctx = (ModifyDnContext) op.getAttachment(SYNCHROCONTEXT);
|
String entryUid = ctx.getEntryUid();
|
String newSuperiorID = ctx.getNewParentId();
|
|
/*
|
* 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(entryUid);
|
|
// Construct the new DN to use for the entry.
|
DN entryDN = op.getEntryDN();
|
DN newSuperior = null;
|
RDN newRDN = op.getNewRDN();
|
|
if (newSuperiorID != null)
|
{
|
newSuperior = findEntryDN(newSuperiorID);
|
}
|
else
|
{
|
newSuperior = entryDN.getParent();
|
}
|
|
//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.getParent().concat(newRDN));
|
numUnresolvedNamingConflicts.incrementAndGet();
|
return true;
|
}
|
|
DN newDN = newSuperior.concat(newRDN);
|
|
if (currentDN == null)
|
{
|
// The entry targeted by the Modify DN is not in the database
|
// anymore.
|
// This is a conflict between a delete and this modify DN.
|
// The entry has been deleted, we can safely assume
|
// that the operation is completed.
|
numResolvedNamingConflicts.incrementAndGet();
|
return true;
|
}
|
|
// 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 true;
|
}
|
|
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;
|
msg.setDn(currentDN.toString());
|
modifyDnMsg.setNewSuperior(newSuperior.toString());
|
numResolvedNamingConflicts.incrementAndGet();
|
return false;
|
}
|
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(entryUid,
|
modifyDnMsg.getNewRDN()));
|
modifyDnMsg.setNewSuperior(newSuperior.toString());
|
numUnresolvedNamingConflicts.incrementAndGet();
|
return false;
|
}
|
else
|
{
|
// The other type of errors can not be caused by naming conflicts.
|
// Log a message for the repair tool.
|
Message message = ERR_ERROR_REPLAYING_OPERATION.get(
|
op.toString(), ctx.getChangeNumber().toString(),
|
result.toString(), op.getErrorMessage().toString());
|
logError(message);
|
return true;
|
}
|
}
|
|
|
/**
|
* 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 true if the process is completed, false if it must continue.
|
* @throws Exception When the operation is not valid.
|
*/
|
private boolean solveNamingConflict(AddOperation op,
|
AddMsg msg) throws Exception
|
{
|
ResultCode result = op.getResultCode();
|
AddContext ctx = (AddContext) op.getAttachment(SYNCHROCONTEXT);
|
String entryUid = ctx.getEntryUid();
|
String parentUniqueId = ctx.getParentUid();
|
|
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 TODO log a
|
* message for the repair tool to look at this problem.
|
*/
|
return true;
|
}
|
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);
|
|
msg.setDn(generateConflictRDN(entryUid,
|
op.getEntryDN().getRDN().toString()) + ","
|
+ baseDn);
|
// reset the parent uid so that the check done is the handleConflict
|
// phase does not fail.
|
msg.setParentUid(null);
|
numUnresolvedNamingConflicts.incrementAndGet();
|
return false;
|
}
|
else
|
{
|
RDN entryRdn = DN.decode(msg.getDn()).getRDN();
|
msg.setDn(entryRdn + "," + parentDn);
|
numResolvedNamingConflicts.incrementAndGet();
|
return false;
|
}
|
}
|
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 nsunique ID already exist, assume this is a replay and
|
* don't do anything
|
* if the entry unique id do not exist, generate conflict.
|
*/
|
if (findEntryDN(entryUid) != null)
|
{
|
// entry already exist : this is a replay
|
return true;
|
}
|
else
|
{
|
addConflict(msg);
|
msg.setDn(generateConflictRDN(entryUid, msg.getDn()));
|
numUnresolvedNamingConflicts.incrementAndGet();
|
return false;
|
}
|
}
|
else
|
{
|
// The other type of errors can not be caused by naming conflicts.
|
// log a message for the repair tool.
|
Message message = ERR_ERROR_REPLAYING_OPERATION.get(
|
op.toString(), ctx.getChangeNumber().toString(),
|
result.toString(), op.getErrorMessage().toString());
|
logError(message);
|
return true;
|
}
|
}
|
|
/**
|
* 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 entryUid The unique ID of the entry whose child must be renamed.
|
* @param entryDN The DN of the entry whose child must be renamed.
|
* @param conflictOp The Operation that generated the conflict.
|
*/
|
private boolean findAndRenameChild(
|
String entryUid, DN entryDN, Operation conflictOp)
|
{
|
boolean conflict = false;
|
|
// Find an rename child entries.
|
InternalClientConnection conn =
|
InternalClientConnection.getRootConnection();
|
DeleteContext ctx =
|
(DeleteContext) conflictOp.getAttachment(SYNCHROCONTEXT);
|
ChangeNumber cn = null;
|
if (ctx != null)
|
cn = ctx.getChangeNumber();
|
|
try
|
{
|
LinkedHashSet<String> attrs = new LinkedHashSet<String>(1);
|
attrs.add(ENTRYUIDNAME);
|
attrs.add(Historical.HISTORICALATTRIBUTENAME);
|
|
SearchFilter ALLMATCH;
|
ALLMATCH = SearchFilter.createFilterFromString("(objectClass=*)");
|
InternalSearchOperation op =
|
conn.processSearch(entryDN, SearchScope.SINGLE_LEVEL,
|
DereferencePolicy.NEVER_DEREF_ALIASES, 0, 0, false, ALLMATCH,
|
attrs);
|
|
if (op.getResultCode() == ResultCode.SUCCESS)
|
{
|
LinkedList<SearchResultEntry> entries = op.getSearchEntries();
|
if (entries != null)
|
{
|
for (SearchResultEntry entry : entries)
|
{
|
/*
|
* Check the ADD and ModRDN date of the child entry. If it is after
|
* the delete date then keep the entry as a conflicting entry,
|
* otherwise delete the entry with the operation.
|
*/
|
if (cn != null)
|
{
|
Historical hist = Historical.load(entry);
|
if (hist.AddedOrRenamedAfter(cn))
|
{
|
conflict = true;
|
markConflictEntry(conflictOp, entry.getDN(), entryDN);
|
renameConflictEntry(conflictOp, entry.getDN(),
|
Historical.getEntryUuid(entry));
|
}
|
}
|
}
|
}
|
}
|
else
|
{
|
// log error and information for the REPAIR tool.
|
MessageBuilder mb = new MessageBuilder();
|
mb.append(ERR_CANNOT_RENAME_CONFLICT_ENTRY.get());
|
mb.append(String.valueOf(entryDN));
|
mb.append(" ");
|
mb.append(String.valueOf(conflictOp));
|
mb.append(" ");
|
mb.append(String.valueOf(op.getResultCode()));
|
logError(mb.toMessage());
|
}
|
} catch (DirectoryException e)
|
{
|
// log error and information for the REPAIR tool.
|
MessageBuilder mb = new MessageBuilder();
|
mb.append(ERR_EXCEPTION_RENAME_CONFLICT_ENTRY.get());
|
mb.append(String.valueOf(entryDN));
|
mb.append(" ");
|
mb.append(String.valueOf(conflictOp));
|
mb.append(" ");
|
mb.append(e.getLocalizedMessage());
|
logError(mb.toMessage());
|
}
|
|
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 uid The uniqueID of the entry to be renamed.
|
*/
|
private void renameConflictEntry(Operation conflictOp, DN dn, String uid)
|
{
|
InternalClientConnection conn =
|
InternalClientConnection.getRootConnection();
|
|
ModifyDNOperation newOp = conn.processModifyDN(
|
dn, generateDeleteConflictDn(uid, dn),false, baseDn);
|
|
if (newOp.getResultCode() != ResultCode.SUCCESS)
|
{
|
// log information for the repair tool.
|
MessageBuilder mb = new MessageBuilder();
|
mb.append(ERR_CANNOT_RENAME_CONFLICT_ENTRY.get());
|
mb.append(String.valueOf(dn));
|
mb.append(" ");
|
mb.append(String.valueOf(conflictOp));
|
mb.append(" ");
|
mb.append(String.valueOf(newOp.getResultCode()));
|
logError(mb.toMessage());
|
}
|
}
|
|
|
/**
|
* 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.
|
InternalClientConnection conn =
|
InternalClientConnection.getRootConnection();
|
|
AttributeType attrType = DirectoryServer.getAttributeType(DS_SYNC_CONFLICT,
|
true);
|
Attribute attr = Attributes.create(attrType, AttributeValues.create(
|
attrType, conflictDN.toString()));
|
List<Modification> mods = new ArrayList<Modification>();
|
Modification mod = new Modification(ModificationType.REPLACE, attr);
|
mods.add(mod);
|
ModifyOperation newOp = conn.processModify(currentDN, mods);
|
if (newOp.getResultCode() != ResultCode.SUCCESS)
|
{
|
// Log information for the repair tool.
|
MessageBuilder mb = new MessageBuilder();
|
mb.append(ERR_CANNOT_ADD_CONFLICT_ATTRIBUTE.get());
|
mb.append(String.valueOf(op));
|
mb.append(" ");
|
mb.append(String.valueOf(newOp.getResultCode()));
|
logError(mb.toMessage());
|
}
|
|
// Generate an alert to let the administration know that some
|
// conflict could not be solved.
|
Message alertMessage = NOTE_UNRESOLVED_CONFLICT.get(conflictDN.toString());
|
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 ASN1Exception When an encoding error happened manipulating the
|
* msg.
|
*/
|
private void addConflict(AddMsg msg) throws ASN1Exception
|
{
|
// Generate an alert to let the administrator know that some
|
// conflict could not be solved.
|
Message alertMessage = NOTE_UNRESOLVED_CONFLICT.get(msg.getDn());
|
DirectoryServer.sendAlertNotification(this,
|
ALERT_TYPE_REPLICATION_UNRESOLVED_CONFLICT, alertMessage);
|
|
// Add the conflict attribute
|
msg.addAttribute(DS_SYNC_CONFLICT, msg.getDn());
|
}
|
|
/**
|
* Generate the Dn to use for a conflicting entry.
|
*
|
* @param entryUid 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 entryUid, String rdn)
|
{
|
return "entryuuid=" + entryUid + "+" + rdn;
|
}
|
|
/**
|
* Generate the RDN to use for a conflicting entry whose father was deleted.
|
*
|
* @param entryUid 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.
|
* @throws DirectoryException
|
*/
|
private RDN generateDeleteConflictDn(String entryUid, DN dn)
|
{
|
String newRDN = "entryuuid=" + entryUid + "+" + dn.getRDN();
|
RDN rdn = null;
|
try
|
{
|
rdn = RDN.decode(newRDN);
|
} catch (DirectoryException e)
|
{
|
// cannot happen
|
}
|
return rdn;
|
}
|
|
/**
|
* Get the number of modify conflicts successfully resolved.
|
* @return The number of modify conflicts successfully resolved.
|
*/
|
public int getNumResolvedModifyConflicts()
|
{
|
return numResolvedModifyConflicts.get();
|
}
|
|
/**
|
* Get the number of naming conflicts successfully resolved.
|
* @return The number of naming conflicts successfully resolved.
|
*/
|
public int getNumResolvedNamingConflicts()
|
{
|
return numResolvedNamingConflicts.get();
|
}
|
|
/**
|
* Get the number of unresolved conflicts.
|
* @return The number of unresolved conflicts.
|
*/
|
public int getNumUnresolvedNamingConflicts()
|
{
|
return numUnresolvedNamingConflicts.get();
|
}
|
|
/**
|
* Check if the domain solve conflicts.
|
*
|
* @return a boolean indicating if the domain should solve conflicts.
|
*/
|
public 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()
|
{
|
state.save();
|
state.clearInMemory();
|
disabled = true;
|
disableService(); // This will cut the session and wake up the listener
|
}
|
|
/**
|
* 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.
|
*/
|
protected void loadDataState()
|
throws DirectoryException
|
{
|
Long compatGenId = null;
|
|
state.clearInMemory();
|
state.loadState();
|
|
// Check to see if a Ruv needs to be translated
|
compatGenId = state.checkRUVCompat();
|
|
generator.adjust(state.getMaxChangeNumber(serverId));
|
// Retrieves the generation ID associated with the data imported
|
|
if (compatGenId != null)
|
{
|
generationId = compatGenId;
|
saveGenerationId(generationId);
|
}
|
else
|
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()
|
{
|
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 ?
|
*/
|
logError(ERR_LOADING_GENERATION_ID.get(
|
baseDn.toNormalizedString(), e.getLocalizedMessage()));
|
return;
|
}
|
|
enableService();
|
|
disabled = false;
|
}
|
|
/**
|
* Compute the data generationId associated with the current data present
|
* in the backend for this domain.
|
* @return The computed generationId.
|
* @throws DirectoryException When an error occurs.
|
*/
|
public long computeGenerationId() throws DirectoryException
|
{
|
long genId = exportBackend(null, true);
|
|
if (debugEnabled())
|
TRACER.debugInfo("Computed generationId: generationId=" + genId);
|
|
return genId;
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
@Override
|
public long getGenerationID()
|
{
|
return generationId;
|
}
|
|
/**
|
* Stores the value of the generationId.
|
* @param generationId The value of the generationId.
|
* @return a ResultCode indicating if the method was successful.
|
*/
|
public ResultCode saveGenerationId(long generationId)
|
{
|
// The generationId is stored in the root entry of the domain.
|
ByteString asn1BaseDn = ByteString.valueOf(baseDn.toString());
|
|
ArrayList<ByteString> values = new ArrayList<ByteString>();
|
ByteString value = ByteString.valueOf(Long.toString(generationId));
|
values.add(value);
|
|
LDAPAttribute attr =
|
new LDAPAttribute(REPLICATION_GENERATION_ID, values);
|
LDAPModification mod = new LDAPModification(ModificationType.REPLACE, attr);
|
ArrayList<RawModification> mods = new ArrayList<RawModification>(1);
|
mods.add(mod);
|
|
ModifyOperationBasis op =
|
new ModifyOperationBasis(conn, InternalClientConnection.nextOperationID(),
|
InternalClientConnection.nextMessageID(),
|
new ArrayList<Control>(0), asn1BaseDn,
|
mods);
|
op.setInternalOperation(true);
|
op.setSynchronizationOperation(true);
|
op.setDontSynchronize(true);
|
|
op.run();
|
|
ResultCode result = op.getResultCode();
|
if (result != ResultCode.SUCCESS)
|
{
|
generationIdSavedStatus = false;
|
if (result != ResultCode.NO_SUCH_OBJECT)
|
{
|
// The case where the backend is empty (NO_SUCH_OBJECT)
|
// is not an error case.
|
Message message = ERR_UPDATING_GENERATION_ID.get(
|
op.getResultCode().getResultCodeName() + " " +
|
op.getErrorMessage(),
|
baseDn.toString());
|
logError(message);
|
}
|
}
|
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.
|
*/
|
public long loadGenerationId()
|
throws DirectoryException
|
{
|
long generationId=-1;
|
|
if (debugEnabled())
|
TRACER.debugInfo(
|
"Attempt to read generation ID from DB " + baseDn.toString());
|
|
ByteString asn1BaseDn = ByteString.valueOf(baseDn.toString());
|
boolean found = false;
|
LDAPFilter filter;
|
try
|
{
|
filter = LDAPFilter.decode("objectclass=*");
|
}
|
catch (LDAPException e)
|
{
|
// can not happen
|
return -1;
|
}
|
|
/*
|
* Search the database entry that is used to periodically
|
* save the generation id
|
*/
|
InternalSearchOperation search = null;
|
LinkedHashSet<String> attributes = new LinkedHashSet<String>(1);
|
attributes.add(REPLICATION_GENERATION_ID);
|
search = conn.processSearch(asn1BaseDn,
|
SearchScope.BASE_OBJECT,
|
DereferencePolicy.DEREF_ALWAYS, 0, 0, false,
|
filter,attributes);
|
if (((search.getResultCode() != ResultCode.SUCCESS)) &&
|
((search.getResultCode() != ResultCode.NO_SUCH_OBJECT)))
|
{
|
Message message = ERR_SEARCHING_GENERATION_ID.get(
|
search.getResultCode().getResultCodeName() + " " +
|
search.getErrorMessage(),
|
baseDn.toString());
|
logError(message);
|
}
|
|
SearchResultEntry resultEntry = null;
|
if (search.getResultCode() == ResultCode.SUCCESS)
|
{
|
LinkedList<SearchResultEntry> result = search.getSearchEntries();
|
resultEntry = result.getFirst();
|
if (resultEntry != null)
|
{
|
AttributeType synchronizationGenIDType =
|
DirectoryServer.getAttributeType(REPLICATION_GENERATION_ID);
|
List<Attribute> attrs =
|
resultEntry.getAttribute(synchronizationGenIDType);
|
if (attrs != null)
|
{
|
Attribute attr = attrs.get(0);
|
if (attr.size()>1)
|
{
|
Message message = ERR_LOADING_GENERATION_ID.get(
|
baseDn.toString(), "#Values=" + attr.size() +
|
" Must be exactly 1 in entry " +
|
resultEntry.toLDIFString());
|
logError(message);
|
}
|
else if (attr.size() == 1)
|
{
|
found=true;
|
try
|
{
|
generationId = Long.decode(attr.iterator().next().toString());
|
}
|
catch(Exception e)
|
{
|
Message message = ERR_LOADING_GENERATION_ID.get(
|
baseDn.toString(), e.getLocalizedMessage());
|
logError(message);
|
}
|
}
|
}
|
}
|
}
|
|
if (!found)
|
{
|
generationId = computeGenerationId();
|
saveGenerationId(generationId);
|
|
if (debugEnabled())
|
TRACER.debugInfo("Generation ID created for domain base DN=" +
|
baseDn.toString() +
|
" generationId=" + generationId);
|
}
|
else
|
{
|
generationIdSavedStatus = true;
|
if (debugEnabled())
|
TRACER.debugInfo(
|
"Generation ID successfully read from domain base DN=" + baseDn +
|
" generationId=" + generationId);
|
}
|
return generationId;
|
}
|
|
/**
|
* Do whatever is needed when a backup is started.
|
* We need to make sure that the serverState is correctly save.
|
*/
|
public void backupStart()
|
{
|
state.save();
|
}
|
|
/**
|
* Do whatever is needed when a backup is finished.
|
*/
|
public void backupEnd()
|
{
|
// Nothing is needed at the moment
|
}
|
|
/*
|
* Total Update >>
|
*/
|
|
|
|
/**
|
* Clears all the entries from the JE backend determined by the
|
* be id passed into the method.
|
*
|
* @param createBaseEntry Indicate whether to automatically create the base
|
* entry and add it to the backend.
|
* @param beID The be id to clear.
|
* @param dn The suffix of the backend to create if the the createBaseEntry
|
* boolean is true.
|
* @throws Exception If an unexpected problem occurs.
|
*/
|
public static void clearJEBackend(boolean createBaseEntry, String beID,
|
String dn) throws Exception
|
{
|
BackendImpl backend = (BackendImpl)DirectoryServer.getBackend(beID);
|
|
// FIXME Should setBackendEnabled be part of TaskUtils ?
|
TaskUtils.disableBackend(beID);
|
|
try
|
{
|
String lockFile = LockFileManager.getBackendLockFileName(backend);
|
StringBuilder failureReason = new StringBuilder();
|
|
if (!LockFileManager.acquireExclusiveLock(lockFile, failureReason))
|
{
|
throw new RuntimeException(failureReason.toString());
|
}
|
|
try
|
{
|
backend.clearBackend();
|
}
|
finally
|
{
|
LockFileManager.releaseLock(lockFile, failureReason);
|
}
|
}
|
finally
|
{
|
TaskUtils.enableBackend(beID);
|
}
|
|
if (createBaseEntry)
|
{
|
DN baseDN = DN.decode(dn);
|
Entry e = createEntry(baseDN);
|
backend = (BackendImpl)DirectoryServer.getBackend(beID);
|
backend.addEntry(e, null);
|
}
|
}
|
|
/**
|
* 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
|
*/
|
protected long exportBackend(OutputStream output, boolean checksumOutput)
|
throws DirectoryException
|
{
|
long genID = 0;
|
Backend backend = retrievesBackend(this.baseDn);
|
long bec = backend.numSubordinates(baseDn, true) + 1;
|
long entryCount = (bec<1000?bec:1000);
|
|
// Acquire a shared lock for the backend.
|
try
|
{
|
String lockFile = LockFileManager.getBackendLockFileName(backend);
|
StringBuilder failureReason = new StringBuilder();
|
if (! LockFileManager.acquireSharedLock(lockFile, failureReason))
|
{
|
Message message = ERR_LDIFEXPORT_CANNOT_LOCK_BACKEND.get(
|
backend.getBackendID(), String.valueOf(failureReason));
|
logError(message);
|
throw new DirectoryException(
|
ResultCode.OTHER, message, null);
|
}
|
}
|
catch (Exception e)
|
{
|
Message message =
|
ERR_LDIFEXPORT_CANNOT_LOCK_BACKEND.get(
|
backend.getBackendID(), e.getLocalizedMessage());
|
logError(message);
|
throw new DirectoryException(
|
ResultCode.OTHER, message, null);
|
}
|
|
OutputStream os;
|
ReplLDIFOutputStream ros = null;
|
|
if (checksumOutput)
|
{
|
ros = new ReplLDIFOutputStream(this, entryCount);
|
os = new CheckedOutputStream(ros, new GenerationIdChecksum());
|
try
|
{
|
os.write((Long.toString(backend.numSubordinates(baseDn, true) + 1)).
|
getBytes());
|
}
|
catch(Exception e)
|
{
|
// Should never happen
|
}
|
}
|
else
|
{
|
os = output;
|
}
|
LDIFExportConfig exportConfig = new LDIFExportConfig(os);
|
|
// baseDn branch is the only one included in the export
|
List<DN> includeBranches = new ArrayList<DN>(1);
|
includeBranches.add(this.baseDn);
|
exportConfig.setIncludeBranches(includeBranches);
|
|
// For the checksum computing mode, only consider the 'stable' attributes
|
if (checksumOutput)
|
{
|
String includeAttributeStrings[] =
|
{"objectclass", "sn", "cn", "entryuuid"};
|
HashSet<AttributeType> includeAttributes;
|
includeAttributes = new HashSet<AttributeType>();
|
for (String attrName : includeAttributeStrings)
|
{
|
AttributeType attrType = DirectoryServer.getAttributeType(attrName);
|
if (attrType == null)
|
{
|
attrType = DirectoryServer.getDefaultAttributeType(attrName);
|
}
|
includeAttributes.add(attrType);
|
}
|
exportConfig.setIncludeAttributes(includeAttributes);
|
}
|
|
// Launch the export.
|
try
|
{
|
backend.exportLDIF(exportConfig);
|
}
|
catch (DirectoryException de)
|
{
|
if ((ros != null) &&
|
(ros.getNumExportedEntries() >= entryCount))
|
{
|
// This is the normal end when computing the generationId
|
// We can interrupt the export only by an IOException
|
}
|
else
|
{
|
Message message =
|
ERR_LDIFEXPORT_ERROR_DURING_EXPORT.get(de.getMessageObject());
|
logError(message);
|
throw new DirectoryException(
|
ResultCode.OTHER, message, null);
|
}
|
}
|
catch (Exception e)
|
{
|
Message message = ERR_LDIFEXPORT_ERROR_DURING_EXPORT.get(
|
stackTraceToSingleLineString(e));
|
logError(message);
|
throw new DirectoryException(
|
ResultCode.OTHER, message, null);
|
}
|
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 =
|
((CheckedOutputStream)os).getChecksum().getValue();
|
}
|
|
// Release the shared lock on the backend.
|
try
|
{
|
String lockFile = LockFileManager.getBackendLockFileName(backend);
|
StringBuilder failureReason = new StringBuilder();
|
if (! LockFileManager.releaseLock(lockFile, failureReason))
|
{
|
Message message = WARN_LDIFEXPORT_CANNOT_UNLOCK_BACKEND.get(
|
backend.getBackendID(), String.valueOf(failureReason));
|
logError(message);
|
throw new DirectoryException(
|
ResultCode.OTHER, message, null);
|
}
|
}
|
catch (Exception e)
|
{
|
Message message = WARN_LDIFEXPORT_CANNOT_UNLOCK_BACKEND.get(
|
backend.getBackendID(), stackTraceToSingleLineString(e));
|
logError(message);
|
throw new DirectoryException(
|
ResultCode.OTHER, message, null);
|
}
|
}
|
return genID;
|
}
|
|
/**
|
* Retrieves the backend related to the domain.
|
*
|
* @return The backend of that domain.
|
* @param baseDn The baseDn to retrieve the backend
|
*/
|
protected static Backend retrievesBackend(DN baseDn)
|
{
|
// Retrieves the backend related to this domain
|
return DirectoryServer.getBackend(baseDn);
|
}
|
|
/**
|
* Process backend before import.
|
* @param backend The backend.
|
* @throws Exception
|
*/
|
private void preBackendImport(Backend backend)
|
throws Exception
|
{
|
// Stop saving state
|
stateSavingDisabled = 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))
|
{
|
Message message = ERR_INIT_CANNOT_LOCK_BACKEND.get(
|
backend.getBackendID(),
|
String.valueOf(failureReason));
|
logError(message);
|
throw new DirectoryException(ResultCode.OTHER, message);
|
}
|
}
|
|
/**
|
* This method should trigger an import of the replicated data.
|
*
|
* @param input The InputStream from which
|
* @throws DirectoryException When needed.
|
*/
|
@Override
|
public void importBackend(InputStream input) throws DirectoryException
|
{
|
LDIFImportConfig importConfig = null;
|
DirectoryException de = null;
|
|
Backend backend = retrievesBackend(baseDn);
|
|
try
|
{
|
if (!backend.supportsLDIFImport())
|
{
|
Message message = ERR_INIT_IMPORT_NOT_SUPPORTED.get(
|
backend.getBackendID().toString());
|
logError(message);
|
de = new DirectoryException(ResultCode.OTHER, message);
|
}
|
else
|
{
|
importConfig =
|
new LDIFImportConfig(input);
|
List<DN> includeBranches = new ArrayList<DN>();
|
includeBranches.add(this.baseDn);
|
importConfig.setIncludeBranches(includeBranches);
|
importConfig.setAppendToExistingData(false);
|
importConfig.setSkipDNValidation(true);
|
// 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;
|
followImport = true;
|
|
// TODO How to deal with rejected entries during the import
|
importConfig.writeRejectedEntries(
|
getFileForPath("logs" + File.separator +
|
"replInitRejectedEntries").getAbsolutePath(),
|
ExistingFileBehavior.OVERWRITE);
|
|
// Process import
|
preBackendImport(backend);
|
backend.importLDIF(importConfig);
|
|
stateSavingDisabled = false;
|
}
|
}
|
catch(Exception e)
|
{
|
de = new DirectoryException(ResultCode.OTHER,
|
Message.raw(e.getLocalizedMessage()));
|
}
|
finally
|
{
|
// Cleanup
|
if (importConfig != null)
|
{
|
importConfig.close();
|
|
// Re-enable backend
|
closeBackendImport(backend);
|
|
backend = retrievesBackend(baseDn);
|
}
|
|
try
|
{
|
loadDataState();
|
|
if (debugEnabled())
|
TRACER.debugInfo(
|
"After import, the replication plugin restarts connections" +
|
" to all RSs to provide new generation ID=" + 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.
|
if (de == null)
|
de = fe;
|
}
|
}
|
// Sends up the root error.
|
if (de != null)
|
{
|
throw de;
|
}
|
}
|
|
/**
|
* Make post import operations.
|
* @param backend The backend implied in the import.
|
* @exception DirectoryException Thrown when an error occurs.
|
*/
|
protected void closeBackendImport(Backend backend)
|
throws DirectoryException
|
{
|
String lockFile = LockFileManager.getBackendLockFileName(backend);
|
StringBuilder failureReason = new StringBuilder();
|
|
// Release lock
|
if (!LockFileManager.releaseLock(lockFile, failureReason))
|
{
|
Message message = WARN_LDIFIMPORT_CANNOT_UNLOCK_BACKEND.get(
|
backend.getBackendID(), String.valueOf(failureReason));
|
logError(message);
|
throw new DirectoryException(ResultCode.OTHER, message);
|
}
|
|
TaskUtils.enableBackend(backend.getBackendID());
|
}
|
|
/**
|
* 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
|
DirectoryServer.getSynchronizationProviders();
|
for (SynchronizationProvider<?> provider :
|
DirectoryServer.getSynchronizationProviders())
|
{
|
if (!( provider instanceof MultimasterReplication))
|
{
|
Message message = ERR_INVALID_PROVIDER.get();
|
throw new DirectoryException(ResultCode.OTHER,
|
message);
|
}
|
|
// From the domainDN retrieves the replication domain
|
LDAPReplicationDomain sdomain =
|
MultimasterReplication.findDomain(baseDn, null);
|
if (sdomain == null)
|
{
|
break;
|
}
|
if (replicationDomain != null)
|
{
|
// Should never happen
|
Message message = ERR_MULTIPLE_MATCHING_DOMAIN.get();
|
throw new DirectoryException(ResultCode.OTHER,
|
message);
|
}
|
replicationDomain = sdomain;
|
}
|
|
if (replicationDomain == null)
|
{
|
throw new DirectoryException(ResultCode.OTHER,
|
ERR_NO_MATCHING_DOMAIN.get(String.valueOf(baseDn)));
|
}
|
return replicationDomain;
|
}
|
|
/**
|
* Returns the backend associated to this domain.
|
* @return The associated backend.
|
*/
|
public Backend getBackend()
|
{
|
return retrievesBackend(baseDn);
|
}
|
|
/*
|
* <<Total Update
|
*/
|
|
|
/**
|
* Push the modifications contained in the given parameter as
|
* a modification that would happen on a local server.
|
* The modifications are not applied to the local database,
|
* historical information is not updated but a ChangeNumber
|
* is generated and the ServerState associated to this domain is
|
* updated.
|
* @param modifications The modification to push
|
*/
|
public void synchronizeModifications(List<Modification> modifications)
|
{
|
ModifyOperation opBasis =
|
new ModifyOperationBasis(InternalClientConnection.getRootConnection(),
|
InternalClientConnection.nextOperationID(),
|
InternalClientConnection.nextMessageID(),
|
null, DirectoryServer.getSchemaDN(),
|
modifications);
|
LocalBackendModifyOperation op = new LocalBackendModifyOperation(opBasis);
|
|
ChangeNumber cn = generateChangeNumber(op);
|
OperationContext ctx = new ModifyContext(cn, "schema");
|
op.setAttachment(SYNCHROCONTEXT, ctx);
|
op.setResultCode(ResultCode.SUCCESS);
|
synchronize(op);
|
}
|
|
/**
|
* 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.
|
*/
|
public static boolean isConfigurationAcceptable(
|
ReplicationDomainCfg configuration, List<Message> unacceptableReasons)
|
{
|
// Check that there is not already a domain with the same DN
|
DN dn = configuration.getBaseDN();
|
LDAPReplicationDomain domain = MultimasterReplication.findDomain(dn, null);
|
if ((domain != null) && (domain.baseDn.equals(dn)))
|
{
|
Message message = ERR_SYNC_INVALID_DN.get();
|
unacceptableReasons.add(message);
|
return false;
|
}
|
|
// Check that the base DN is configured as a base-dn of the directory server
|
if (retrievesBackend(dn) == null)
|
{
|
Message message = ERR_UNKNOWN_DN.get(dn.toString());
|
unacceptableReasons.add(message);
|
return false;
|
}
|
|
// Check fractional configuration
|
try
|
{
|
isFractionalConfigAcceptable(configuration);
|
} catch (ConfigException e)
|
{
|
unacceptableReasons.add(e.getMessageObject());
|
return false;
|
}
|
|
return true;
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
public ConfigChangeResult applyConfigurationChange(
|
ReplicationDomainCfg configuration)
|
{
|
isolationpolicy = configuration.getIsolationPolicy();
|
|
changeConfig(
|
configuration.getReplicationServer(),
|
configuration.getWindowSize(),
|
configuration.getHeartbeatInterval(),
|
(byte)configuration.getGroupId());
|
|
// Read assured configuration and reconnect if needed
|
readAssuredConfig(configuration, true);
|
|
// Read fractional configuration and reconnect if needed
|
readFractionalConfig(configuration, true);
|
|
/*
|
* 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.
|
*/
|
if (baseDn.compareTo(DirectoryServer.getSchemaDN()) == 0)
|
{
|
solveConflictFlag = false;
|
}
|
else
|
{
|
solveConflictFlag = configuration.isSolveConflicts();
|
}
|
|
try
|
{
|
createECLDomainCfg(configuration);
|
}
|
catch(Exception e)
|
{
|
return new ConfigChangeResult(ResultCode.OTHER, false);
|
}
|
|
return new ConfigChangeResult(ResultCode.SUCCESS, false);
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
public boolean isConfigurationChangeAcceptable(
|
ReplicationDomainCfg configuration, List<Message> unacceptableReasons)
|
{
|
// Check that a import/export is not in progress
|
if (this.importInProgress() || this.exportInProgress())
|
{
|
unacceptableReasons.add(
|
NOTE_ERR_CANNOT_CHANGE_CONFIG_DURING_TOTAL_UPDATE.get());
|
return false;
|
}
|
|
// Check fractional configuration
|
try
|
{
|
isFractionalConfigAcceptable(configuration);
|
} catch (ConfigException e)
|
{
|
unacceptableReasons.add(e.getMessageObject());
|
return false;
|
}
|
|
return true;
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
public LinkedHashMap<String, String> getAlerts()
|
{
|
LinkedHashMap<String,String> alerts = new LinkedHashMap<String,String>();
|
|
alerts.put(ALERT_TYPE_REPLICATION_UNRESOLVED_CONFLICT,
|
ALERT_DESCRIPTION_REPLICATION_UNRESOLVED_CONFLICT);
|
return alerts;
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
public String getClassName()
|
{
|
return CLASS_NAME;
|
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
public DN getComponentEntryDN()
|
{
|
return configDn;
|
}
|
|
/**
|
* Starts the Replication Domain.
|
*/
|
public void start()
|
{
|
// Create the ServerStateFlush thread
|
flushThread = new ServerStateFlush();
|
flushThread.start();
|
|
startListenService();
|
}
|
|
|
/**
|
* Remove from this domain configuration, the configuration of the
|
* external change log.
|
*/
|
public void removeECLDomainCfg()
|
{
|
try
|
{
|
DN eclConfigEntryDN = DN.decode(
|
"cn=external changeLog," + configDn);
|
|
if (DirectoryServer.getConfigHandler().entryExists(eclConfigEntryDN))
|
{
|
DirectoryServer.getConfigHandler().deleteEntry(eclConfigEntryDN, null);
|
}
|
}
|
catch(Exception e)
|
{
|
TRACER.debugCaught(DebugLogLevel.ERROR, e);
|
MessageBuilder mb = new MessageBuilder();
|
mb.append(e.getMessage());
|
Message msg = ERR_CHECK_CREATE_REPL_BACKEND_FAILED.get(mb.toString());
|
logError(msg);
|
}
|
}
|
|
/**
|
* Create the ECl configuration.
|
* @param configuration The provided configuration.
|
* @throws ConfigException a.
|
*/
|
public void createECLDomainCfg(ReplicationDomainCfg configuration)
|
throws ConfigException
|
{
|
// create the ecl config if it does not exist
|
// There may ot be any config entry related to this domain in some
|
// unit test cases
|
try
|
{
|
ExternalChangelogDomainCfg eclDomCfg = null;
|
if (DirectoryServer.getConfigHandler().entryExists(configDn))
|
{
|
try
|
{ eclDomCfg = configuration.getExternalChangelogDomain();
|
}catch(Exception e) {}
|
if (eclDomCfg==null)
|
{
|
DN eclConfigEntryDN = DN.decode("cn=external changelog," + configDn);
|
if (!DirectoryServer.getConfigHandler().entryExists(eclConfigEntryDN))
|
{
|
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));
|
LDIFReader reader = new LDIFReader(ldifImportConfig);
|
Entry eclEntry = reader.readEntry();
|
DirectoryServer.getConfigHandler().addEntry(eclEntry, null);
|
ldifImportConfig.close();
|
}
|
}
|
}
|
eclDomCfg = configuration.getExternalChangelogDomain();
|
eclDomain = new ExternalChangelogDomain(this, eclDomCfg);
|
}
|
catch(Exception de)
|
{
|
throw new ConfigException(
|
NOTE_ERR_UNABLE_TO_ENABLE_ECL.get(
|
"Replication Domain on" + baseDn,
|
de.getMessage() + " " + de.getCause().getMessage()), de);
|
}
|
}
|
|
private static String makeLdif(String... lines)
|
{
|
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();
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
@Override
|
public void sessionInitiated(
|
ServerStatus initStatus,
|
ServerState replicationServerState,
|
long generationID,
|
ProtocolSession session)
|
{
|
// Check domain fractional configuration consistency with local
|
// configuration variables
|
force_bad_data_set = !isBackendFractionalConfigConsistent();
|
|
super.sessionInitiated(
|
initStatus, replicationServerState,generationID, session);
|
|
// Now that we are connected , we can enable ECL if :
|
// 1/ RS must in the same JVM and created an ECL_WORKFLOW_ELEMENT
|
// and 2/ this domain must NOT be private
|
if (!getBackend().isPrivateBackend())
|
{
|
try
|
{
|
ECLWorkflowElement wfe = (ECLWorkflowElement)
|
DirectoryServer.getWorkflowElement(
|
ECLWorkflowElement.ECL_WORKFLOW_ELEMENT);
|
if (wfe!=null)
|
wfe.getReplicationServer().enableECL();
|
}
|
catch(DirectoryException de)
|
{
|
Message message =
|
NOTE_ERR_UNABLE_TO_ENABLE_ECL.get(
|
"Replication Domain on" + baseDn,
|
de.getMessage() + " " + de.getCause().getMessage());
|
logError(message);
|
// and go on
|
}
|
}
|
|
// Now for bad data set status if needed
|
if (force_bad_data_set)
|
{
|
// Go into bad data set status
|
setNewStatus(StatusMachineEvent.TO_BAD_GEN_ID_STATUS_EVENT);
|
broker.signalStatusChange(status);
|
Message message = NOTE_FRACTIONAL_BAD_DATA_SET_NEED_RESYNC.get(
|
baseDn.toString());
|
logError(message);
|
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.
|
*/
|
ChangeNumber replServerMaxChangeNumber =
|
replicationServerState.getMaxChangeNumber(serverId);
|
|
// 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 volontary reset
|
// and we don't want to update it with our changes that could be huge.
|
if ((replServerMaxChangeNumber != null) &&
|
(replServerMaxChangeNumber.getSeqnum()!=0))
|
{
|
ChangeNumber ourMaxChangeNumber =
|
state.getMaxChangeNumber(serverId);
|
|
if ((ourMaxChangeNumber != null) &&
|
(!ourMaxChangeNumber.olderOrEqual(replServerMaxChangeNumber)))
|
{
|
pendingChanges.setRecovering(true);
|
broker.setRecoveryRequired(true);
|
new RSUpdater(replServerMaxChangeNumber).start();
|
}
|
}
|
} catch (Exception e)
|
{
|
Message message = ERR_PUBLISHING_FAKE_OPS.get(
|
baseDn.toNormalizedString(),
|
e.getLocalizedMessage() + stackTraceToSingleLineString(e));
|
logError(message);
|
}
|
|
}
|
|
/**
|
* Build the list of changes that have been processed by this server
|
* after the ChangeNumber given as a parameter and publish them
|
* using the given session.
|
*
|
* @param startingChangeNumber The ChangeNumber whe 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.
|
*/
|
public boolean buildAndPublishMissingChanges(
|
ChangeNumber startingChangeNumber,
|
ReplicationBroker session)
|
throws Exception
|
{
|
// Trim the changes in replayOperations that are older than
|
// the startingChangeNumber.
|
synchronized (replayOperations)
|
{
|
Iterator<ChangeNumber> it = replayOperations.keySet().iterator();
|
while (it.hasNext())
|
{
|
if (it.next().olderOrEqual(startingChangeNumber))
|
{
|
it.remove();
|
}
|
else
|
{
|
break;
|
}
|
}
|
}
|
|
ChangeNumber lastRetrievedChange = null;
|
long missingChangesDelta;
|
InternalSearchOperation op;
|
ChangeNumber currentStartChangeNumber = startingChangeNumber;
|
do
|
{
|
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.
|
missingChangesDelta = currentStartChangeNumber.getTime() + 10000;
|
ChangeNumber endChangeNumber =
|
new ChangeNumber(
|
missingChangesDelta, 0xffffffff, serverId);
|
|
ScanSearchListener listener =
|
new ScanSearchListener(currentStartChangeNumber, endChangeNumber);
|
op = searchForChangedEntries(
|
baseDn, currentStartChangeNumber, endChangeNumber, listener);
|
|
// Publish and remove all the changes from the replayOperations list
|
// that are older than the endChangeNumber.
|
LinkedList<FakeOperation> opsToSend = new LinkedList<FakeOperation>();
|
synchronized (replayOperations)
|
{
|
Iterator<FakeOperation> itOp = replayOperations.values().iterator();
|
while (itOp.hasNext())
|
{
|
FakeOperation fakeOp = itOp.next();
|
if ((fakeOp.getChangeNumber().olderOrEqual(endChangeNumber))
|
&& state.cover(fakeOp.getChangeNumber()))
|
{
|
lastRetrievedChange = fakeOp.getChangeNumber();
|
opsToSend.add(fakeOp);
|
itOp.remove();
|
}
|
else
|
{
|
break;
|
}
|
}
|
}
|
|
for (FakeOperation opToSend : opsToSend)
|
{
|
session.publishRecovery(opToSend.generateMessage());
|
}
|
opsToSend.clear();
|
if (lastRetrievedChange != null)
|
{
|
currentStartChangeNumber = lastRetrievedChange;
|
}
|
else
|
{
|
currentStartChangeNumber = endChangeNumber;
|
}
|
|
} while (pendingChanges.RecoveryUntil(lastRetrievedChange) &&
|
(op.getResultCode().equals(ResultCode.SUCCESS)));
|
|
return op.getResultCode().equals(ResultCode.SUCCESS);
|
}
|
|
|
/**
|
* Search for the changes that happened since fromChangeNumber
|
* based on the historical attribute. The only changes that will
|
* be send will be the one generated on the serverId provided in
|
* fromChangeNumber.
|
* @param baseDn the base DN
|
* @param fromChangeNumber The ChangeNumber from which we want the changes
|
* @param lastChangeNumber The max ChangeNumber that the search should return
|
* @param resultListener The listener that will process the entries returned
|
* @return the internal search operation
|
* @throws Exception when raised.
|
*/
|
public static InternalSearchOperation searchForChangedEntries(
|
DN baseDn,
|
ChangeNumber fromChangeNumber,
|
ChangeNumber lastChangeNumber,
|
InternalSearchListener resultListener)
|
throws Exception
|
{
|
InternalClientConnection conn =
|
InternalClientConnection.getRootConnection();
|
Integer serverId = fromChangeNumber.getServerId();
|
|
String maxValueForId;
|
if (lastChangeNumber == null)
|
{
|
maxValueForId = "ffffffffffffffff" + String.format("%04x", serverId)
|
+ "ffffffff";
|
}
|
else
|
{
|
maxValueForId = lastChangeNumber.toString();
|
}
|
|
LDAPFilter filter = LDAPFilter.decode(
|
"(&(" + Historical.HISTORICALATTRIBUTENAME + ">=dummy:"
|
+ fromChangeNumber + ")(" + Historical.HISTORICALATTRIBUTENAME +
|
"<=dummy:" + maxValueForId + "))");
|
|
LinkedHashSet<String> attrs = new LinkedHashSet<String>(1);
|
attrs.add(Historical.HISTORICALATTRIBUTENAME);
|
attrs.add(Historical.ENTRYUIDNAME);
|
attrs.add("*");
|
return conn.processSearch(
|
ByteString.valueOf(baseDn.toString()),
|
SearchScope.WHOLE_SUBTREE,
|
DereferencePolicy.NEVER_DEREF_ALIASES,
|
0, 0, false, filter,
|
attrs,
|
resultListener);
|
}
|
|
/**
|
* Search for the changes that happened since fromChangeNumber
|
* based on the historical attribute. The only changes that will
|
* be send will be the one generated on the serverId provided in
|
* fromChangeNumber.
|
* @param baseDn the base DN
|
* @param fromChangeNumber The change number from which we want the changes
|
* @param resultListener that will process the entries returned.
|
* @return the internal search operation
|
* @throws Exception when raised.
|
*/
|
public static InternalSearchOperation searchForChangedEntries(
|
DN baseDn,
|
ChangeNumber fromChangeNumber,
|
InternalSearchListener resultListener)
|
throws Exception
|
{
|
return searchForChangedEntries(
|
baseDn, fromChangeNumber, 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
|
{
|
Backend backend = retrievesBackend(baseDn);
|
|
if (!backend.supportsLDIFExport())
|
{
|
Message message = ERR_INIT_EXPORT_NOT_SUPPORTED.get(
|
backend.getBackendID().toString());
|
logError(message);
|
throw new DirectoryException(ResultCode.OTHER, message);
|
}
|
|
return backend.numSubordinates(baseDn, true) + 1;
|
}
|
|
/**
|
* {@inheritDoc}
|
*/
|
@Override
|
public boolean processUpdate(UpdateMsg updateMsg)
|
{
|
// Ignore message if fractional configuration is inconcsistent and
|
// we have been passed into bad data set status
|
if (force_bad_data_set)
|
{
|
return false;
|
}
|
|
if (updateMsg instanceof LDAPUpdateMsg)
|
{
|
LDAPUpdateMsg msg = (LDAPUpdateMsg) updateMsg;
|
|
// put the UpdateMsg in the RemotePendingChanges list.
|
remotePendingChanges.putRemoteUpdate(msg);
|
|
// Put update message into the replay queue
|
// (block until some place in the queue is available)
|
UpdateToReplay updateToReplay = new UpdateToReplay(msg, this);
|
updateToReplayQueue.offer(updateToReplay);
|
|
return false;
|
}
|
|
// unknown message type, this should not happen, just ignore it.
|
return true;
|
}
|
|
/**
|
* Monitoring information for the LDAPReplicationDomain.
|
*
|
* @return Monitoring attributes specific to the LDAPReplicationDomain.
|
*/
|
@Override
|
public Collection<Attribute> getAdditionalMonitoring()
|
{
|
ArrayList<Attribute> attributes = new ArrayList<Attribute>();
|
|
/* get number of changes in the pending list */
|
ReplicationMonitor.addMonitorData(
|
attributes, "pending-updates", getPendingUpdatesCount());
|
|
/* get number of changes successfully */
|
ReplicationMonitor.addMonitorData(attributes, "replayed-updates-ok",
|
getNumReplayedPostOpCalled());
|
|
/* get number of modify conflicts */
|
ReplicationMonitor.addMonitorData(attributes, "resolved-modify-conflicts",
|
getNumResolvedModifyConflicts());
|
|
/* get number of naming conflicts */
|
ReplicationMonitor.addMonitorData(attributes, "resolved-naming-conflicts",
|
getNumResolvedNamingConflicts());
|
|
/* get number of unresolved naming conflicts */
|
ReplicationMonitor.addMonitorData(attributes, "unresolved-naming-conflicts",
|
getNumUnresolvedNamingConflicts());
|
|
ReplicationMonitor.addMonitorData(attributes, "remote-pending-changes-size",
|
remotePendingChanges.getQueueSize());
|
|
return attributes;
|
}
|
|
/**
|
* 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;
|
Throwable cause = null;
|
try
|
{
|
source = Integer.decode(sourceString);
|
if ((source >= -1) && (source != serverId))
|
{
|
// 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)
|
{
|
cause = e;
|
}
|
|
ResultCode resultCode = ResultCode.OTHER;
|
Message message = ERR_INVALID_IMPORT_SOURCE.get();
|
if (cause != null)
|
{
|
throw new DirectoryException(
|
resultCode, message, cause);
|
}
|
else
|
{
|
throw new DirectoryException(
|
resultCode, message);
|
}
|
}
|
|
/**
|
* Called by synchronize post op plugin in order to add the entry historized
|
* attributes to the UpdateMsg.
|
* @param msg
|
* @param op
|
* @throws DirectoryException
|
*/
|
private void addEntryAttributesForCL(UpdateMsg msg,PostOperationOperation op)
|
throws DirectoryException
|
{
|
String[] entryAttributeNames =
|
getEclInclude().toArray(new String[0]);
|
ArrayList<Attribute> newattrs = new ArrayList<Attribute>();
|
|
if (op instanceof PostOperationDeleteOperation)
|
{
|
Entry entry = null;
|
PostOperationDeleteOperation delOp = (PostOperationDeleteOperation)op;
|
entry = delOp.getEntryToDelete();
|
for (String name : entryAttributeNames)
|
{
|
AttributeType atype = DirectoryServer.getAttributeType(name);
|
List<Attribute> attrs = entry.getAttribute(atype);
|
if (attrs != null)
|
for (Attribute a : attrs)
|
newattrs.add(a);
|
}
|
((DeleteMsg)msg).setEclIncludes(newattrs);
|
|
// For delete only, add the modifiersName since they are required in the
|
// ECL entry but are not part of other parts of the message
|
AttributeType atype = DirectoryServer.getAttributeType("modifiersname");
|
List<Attribute> attrs = entry.getAttribute(atype);
|
if (attrs != null)
|
{
|
for (Attribute a : attrs)
|
{
|
((DeleteMsg)msg).setInitiatorsName(a.iterator().next().toString());
|
break;
|
}
|
}
|
|
}
|
else if (op instanceof PostOperationModifyOperation)
|
{
|
Entry entry = null;
|
PostOperationModifyOperation modOp = (PostOperationModifyOperation)op;
|
entry = modOp.getCurrentEntry();
|
for (String name : entryAttributeNames)
|
{
|
AttributeType atype = DirectoryServer.getAttributeType(name);
|
List<Attribute> attrs = entry.getAttribute(atype);
|
for (Attribute a : attrs)
|
newattrs.add(a);
|
}
|
((ModifyMsg)msg).setEclIncludes(newattrs);
|
}
|
else if (op instanceof PostOperationModifyDNOperation)
|
{
|
Entry entry = null;
|
PostOperationModifyDNOperation modDNOp =
|
(PostOperationModifyDNOperation)op;
|
entry = modDNOp.getOriginalEntry();
|
for (String name : entryAttributeNames)
|
{
|
AttributeType atype = DirectoryServer.getAttributeType(name);
|
List<Attribute> attrs = entry.getAttribute(atype);
|
for (Attribute a : attrs)
|
newattrs.add(a);
|
}
|
((ModifyDNMsg)msg).setEclIncludes(newattrs);
|
}
|
else if (op instanceof PostOperationAddOperation)
|
{
|
Entry entry = null;
|
PostOperationAddOperation addOp = (PostOperationAddOperation)op;
|
entry = addOp.getEntryToAdd();
|
for (String name : entryAttributeNames)
|
{
|
AttributeType atype = DirectoryServer.getAttributeType(name);
|
List<Attribute> attrs = entry.getAttribute(atype);
|
if (attrs != null)
|
{
|
for (Attribute a : attrs)
|
newattrs.add(a);
|
}
|
else
|
{
|
// FIXME:ECL
|
}
|
}
|
((AddMsg)msg).setEclIncludes(newattrs);
|
}
|
}
|
|
/**
|
* 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 = false;
|
|
/**
|
* - 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<String, List<String>> fractionalSpecificClassesAttributes =
|
new HashMap<String, List<String>>();
|
|
/**
|
* 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 List<String> fractionalAllClassesAttributes =
|
new ArrayList<String>();
|
|
/**
|
* Base DN the fractional configuration is for.
|
*/
|
private DN baseDn = null;
|
|
/**
|
* Constructs a new fractional configuration object.
|
* @param baseDn The base dn the object is for.
|
*/
|
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
|
*/
|
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
|
*/
|
void setFractionalExclusive(boolean fractionalExclusive)
|
{
|
this.fractionalExclusive = fractionalExclusive;
|
}
|
|
/**
|
* Getter for fractionalSpecificClassesAttributes attribute.
|
* @return The fractionalSpecificClassesAttributes attribute.
|
*/
|
Map<String, List<String>> getFractionalSpecificClassesAttributes()
|
{
|
return fractionalSpecificClassesAttributes;
|
}
|
|
/**
|
* Set the fractionalSpecificClassesAttributes parameter.
|
* @param fractionalSpecificClassesAttributes The
|
* fractionalSpecificClassesAttributes parameter to set.
|
*/
|
void setFractionalSpecificClassesAttributes(Map<String,
|
List<String>> fractionalSpecificClassesAttributes)
|
{
|
this.fractionalSpecificClassesAttributes =
|
fractionalSpecificClassesAttributes;
|
}
|
|
/**
|
* Getter for fractionalSpecificClassesAttributes attribute.
|
* @return The fractionalSpecificClassesAttributes attribute.
|
*/
|
List<String> getFractionalAllClassesAttributes()
|
{
|
return fractionalAllClassesAttributes;
|
}
|
|
/**
|
* Set the fractionalAllClassesAttributes parameter.
|
* @param fractionalAllClassesAttributes The
|
* fractionalSpecificClassesAttributes parameter to set.
|
*/
|
void setFractionalAllClassesAttributes(
|
List<String> 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<String> exclIt = null;
|
SortedSet<String> fractionalExclude =
|
configuration.getFractionalExclude();
|
if (fractionalExclude != null)
|
{
|
exclIt = fractionalExclude.iterator();
|
}
|
|
Iterator<String> inclIt = null;
|
SortedSet<String> fractionalInclude =
|
configuration.getFractionalInclude();
|
if (fractionalInclude != null)
|
{
|
inclIt = fractionalInclude.iterator();
|
}
|
|
// Get potentially new fractional configuration
|
Map<String, List<String>> newFractionalSpecificClassesAttributes =
|
new HashMap<String, List<String>>();
|
List<String> newFractionalAllClassesAttributes = new ArrayList<String>();
|
|
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);
|
if (newFractionalMode == EXCLUSIVE_FRACTIONAL)
|
result.setFractionalExclusive(true);
|
else
|
result.setFractionalExclusive(false);
|
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<String> exclIt, Iterator<String> inclIt,
|
Map<String, List<String>> fractionalSpecificClassesAttributes,
|
List<String> fractionalAllClassesAttributes) throws ConfigException
|
{
|
int fractional_mode = NOT_FRACTIONAL;
|
|
// Determine if fractional-exclude or fractional-include property is used
|
// : only one of them is allowed
|
Iterator<String> fracConfIt = null;
|
|
// Deduce the wished fractional mode
|
if ((exclIt != null) && exclIt.hasNext())
|
{
|
if ((inclIt != null) && inclIt.hasNext())
|
{
|
throw new ConfigException(
|
NOTE_ERR_FRACTIONAL_CONFIG_BOTH_MODES.get());
|
}
|
else
|
{
|
fractional_mode = EXCLUSIVE_FRACTIONAL;
|
fracConfIt = exclIt;
|
}
|
}
|
else
|
{
|
if ((inclIt != null) && inclIt.hasNext())
|
{
|
fractional_mode = INCLUSIVE_FRACTIONAL;
|
fracConfIt = inclIt;
|
}
|
else
|
{
|
return NOT_FRACTIONAL;
|
}
|
}
|
|
while (fracConfIt.hasNext())
|
{
|
// Parse a value with the form class:attr1,attr2...
|
// or *:attr1,attr2...
|
String fractCfgStr = fracConfIt.next();
|
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 = classNameLower.equals("*");
|
// 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)
|
{
|
// Avoid duplicate attributes
|
if (!fractionalAllClassesAttributes.contains(attrNameLower))
|
{
|
fractionalAllClassesAttributes.add(attrNameLower);
|
}
|
}
|
else
|
{
|
List<String> attrList =
|
fractionalSpecificClassesAttributes.get(classNameLower);
|
if (attrList != null)
|
{
|
// Avoid duplicate attributes
|
if (!attrList.contains(attrNameLower))
|
{
|
attrList.add(attrNameLower);
|
}
|
} else
|
{
|
attrList = new ArrayList<String>();
|
attrList.add(attrNameLower);
|
fractionalSpecificClassesAttributes.put(classNameLower, attrList);
|
}
|
}
|
}
|
}
|
return fractional_mode;
|
}
|
|
// Return type of the parseFractionalConfig method
|
static final int NOT_FRACTIONAL = 0;
|
static final int EXCLUSIVE_FRACTIONAL = 1;
|
static final int INCLUSIVE_FRACTIONAL = 2;
|
|
/**
|
* Get an integer representation of the domain fractional configuration.
|
* @return An integer representation of the domain fractional configuration.
|
*/
|
int fractionalConfigToInt()
|
{
|
int fractionalMode = -1;
|
if (fractional)
|
{
|
if (fractionalExclusive)
|
fractionalMode = EXCLUSIVE_FRACTIONAL;
|
else
|
fractionalMode = INCLUSIVE_FRACTIONAL;
|
} else
|
{
|
fractionalMode = NOT_FRACTIONAL;
|
}
|
return fractionalMode;
|
}
|
|
/**
|
* Compare 2 fractional replication configurations and returns true if they
|
* are equivalent.
|
* @param fractionalConfig1 First fractional configuration
|
* @param fractionalConfig2 Second fractional configuration
|
* @return True if both configurations are equivalent.
|
* @throws ConfigException If some classes or attributes could not be
|
* retrieved from the schema.
|
*/
|
static boolean isFractionalConfigEquivalent(
|
FractionalConfig fractionalConfig1, FractionalConfig fractionalConfig2)
|
throws ConfigException
|
{
|
// Comapre base DNs just to be consistent
|
if (!fractionalConfig1.getBaseDn().equals(fractionalConfig2.getBaseDn()))
|
return false;
|
|
// Compare modes
|
if ( (fractionalConfig1.isFractional() !=
|
fractionalConfig2.isFractional()) ||
|
(fractionalConfig1.isFractionalExclusive() !=
|
fractionalConfig2.isFractionalExclusive()) )
|
return false;
|
|
// Compare all classes attributes
|
List<String> allClassesAttributes1 =
|
fractionalConfig1.getFractionalAllClassesAttributes();
|
List<String> allClassesAttributes2 =
|
fractionalConfig2.getFractionalAllClassesAttributes();
|
if (!isAttributeListEquivalent(allClassesAttributes1,
|
allClassesAttributes2))
|
return false;
|
|
// Compare specific classes attributes
|
Map<String, List<String>> specificClassesAttributes1 =
|
fractionalConfig1.getFractionalSpecificClassesAttributes();
|
Map<String, List<String>> specificClassesAttributes2 =
|
fractionalConfig2.getFractionalSpecificClassesAttributes();
|
if (specificClassesAttributes1.size() !=
|
specificClassesAttributes2.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 = DirectoryServer.getSchema();
|
for (String className1 : specificClassesAttributes1.keySet())
|
{
|
// Get class from specificClassesAttributes1
|
ObjectClass objectClass1 = schema.getObjectClass(className1);
|
if (objectClass1 == null)
|
{
|
throw new ConfigException(
|
NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_OBJECT_CLASS.get(className1));
|
}
|
|
// Look for matching one in specificClassesAttributes2
|
boolean foundClass = false;
|
for (String className2 : specificClassesAttributes2.keySet())
|
{
|
ObjectClass objectClass2 = schema.getObjectClass(className2);
|
if (objectClass2 == null)
|
{
|
throw new ConfigException(
|
NOTE_ERR_FRACTIONAL_CONFIG_UNKNOWN_OBJECT_CLASS.get(className2));
|
}
|
if (objectClass1.equals(objectClass2))
|
{
|
foundClass = true;
|
// Now compare the 2 attribute lists
|
List<String> attributes1 =
|
specificClassesAttributes1.get(className1);
|
List<String> attributes2 =
|
specificClassesAttributes2.get(className2);
|
if (!isAttributeListEquivalent(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.
|
*/
|
public boolean isECLEnabled()
|
{
|
return this.eclDomain.isEnabled();
|
}
|
}
|
