| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| * |
| */ |
| package org.apache.directory.mitosis.service; |
| |
| |
| import org.apache.directory.mitosis.common.CSN; |
| import org.apache.directory.mitosis.common.Constants; |
| import org.apache.directory.mitosis.common.DefaultCSN; |
| import org.apache.directory.mitosis.common.ReplicaId; |
| import org.apache.directory.mitosis.configuration.ReplicationConfiguration; |
| import org.apache.directory.mitosis.operation.Operation; |
| import org.apache.directory.mitosis.operation.OperationFactory; |
| import org.apache.directory.mitosis.service.protocol.codec.ReplicationServerProtocolCodecFactory; |
| import org.apache.directory.mitosis.service.protocol.handler.ReplicationClientContextHandler; |
| import org.apache.directory.mitosis.service.protocol.handler.ReplicationServerContextHandler; |
| import org.apache.directory.mitosis.service.protocol.handler.ReplicationServerProtocolHandler; |
| import org.apache.directory.mitosis.store.ReplicationStore; |
| import org.apache.directory.server.core.DirectoryService; |
| import org.apache.directory.server.core.entry.ServerEntry; |
| import org.apache.directory.server.core.entry.ServerSearchResult; |
| import org.apache.directory.server.core.enumeration.SearchResultFilteringEnumeration; |
| import org.apache.directory.server.core.interceptor.BaseInterceptor; |
| import org.apache.directory.server.core.interceptor.Interceptor; |
| import org.apache.directory.server.core.interceptor.NextInterceptor; |
| import org.apache.directory.server.core.interceptor.context.AddOperationContext; |
| import org.apache.directory.server.core.interceptor.context.DeleteOperationContext; |
| import org.apache.directory.server.core.interceptor.context.EntryOperationContext; |
| import org.apache.directory.server.core.interceptor.context.GetMatchedNameOperationContext; |
| import org.apache.directory.server.core.interceptor.context.ListOperationContext; |
| import org.apache.directory.server.core.interceptor.context.LookupOperationContext; |
| import org.apache.directory.server.core.interceptor.context.ModifyOperationContext; |
| import org.apache.directory.server.core.interceptor.context.MoveAndRenameOperationContext; |
| import org.apache.directory.server.core.interceptor.context.MoveOperationContext; |
| import org.apache.directory.server.core.interceptor.context.RenameOperationContext; |
| import org.apache.directory.server.core.interceptor.context.SearchOperationContext; |
| import org.apache.directory.server.core.invocation.InvocationStack; |
| import org.apache.directory.server.core.partition.PartitionNexus; |
| import org.apache.directory.server.schema.registries.AttributeTypeRegistry; |
| import org.apache.directory.server.schema.registries.Registries; |
| import org.apache.directory.shared.ldap.constants.SchemaConstants; |
| import org.apache.directory.shared.ldap.entry.EntryAttribute; |
| import org.apache.directory.shared.ldap.entry.Value; |
| import org.apache.directory.shared.ldap.exception.LdapNameNotFoundException; |
| import org.apache.directory.shared.ldap.filter.ExprNode; |
| import org.apache.directory.shared.ldap.filter.FilterParser; |
| import org.apache.directory.shared.ldap.filter.PresenceNode; |
| import org.apache.directory.shared.ldap.message.AliasDerefMode; |
| import org.apache.directory.shared.ldap.name.LdapDN; |
| import org.apache.mina.common.IoAcceptor; |
| import org.apache.mina.filter.LoggingFilter; |
| import org.apache.mina.filter.codec.ProtocolCodecFilter; |
| import org.apache.mina.transport.socket.nio.SocketAcceptor; |
| import org.apache.mina.transport.socket.nio.SocketAcceptorConfig; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import javax.naming.NameNotFoundException; |
| import javax.naming.NamingEnumeration; |
| import javax.naming.NamingException; |
| import javax.naming.directory.SearchControls; |
| |
| import java.net.InetSocketAddress; |
| import java.text.ParseException; |
| import java.util.ArrayList; |
| import java.util.List; |
| |
| |
| /** |
| * An {@link Interceptor} that intercepts LDAP operations and propagates the |
| * changes occurred by the operations into other {@link ReplicaId}s so the DIT |
| * of each {@link ReplicaId} in the cluster has the same content without any |
| * conflict. |
| * <p> |
| * Once an operation is invoked, this interceptor transforms it into one or |
| * more operations that makes the requested operation more proper and robust |
| * for replication. The transformation process is actually just calling a |
| * respective factory method in {@link OperationFactory}. The methods in |
| * {@link OperationFactory} returns a new {@link Operation} instance. |
| * <p> |
| * The newly created {@link Operation} is used for three purposes. |
| * <ul> |
| * <li>To perform the requested operation to the local {@link PartitionNexus} |
| * <li>To store the created {@link Operation} itself to |
| * {@link ReplicationStore} so that it can be retrieved later by |
| * {@link ReplicationLogCleanJob} and {@link ReplicationClientContextHandler} |
| * <li>To transfer itself to other {@link ReplicaId}s via TCP/IP communication |
| * between {@link ReplicationClientContextHandler} and |
| * {@link ReplicationServerContextHandler} |
| * </ul> |
| * The first two actions (modifying the local DIT and storing the |
| * {@link Operation} to {@link ReplicationStore}) are performed automatically |
| * when |
| * {@link Operation#execute(PartitionNexus, ReplicationStore, Registries)} |
| * method is invoked. {@link ReplicationInterceptor} always call it instead of |
| * forwarding the requested operation to the next {@link Interceptor}. |
| * <p> |
| * The last action takes place by {@link ReplicationClientContextHandler}, |
| * which handles TCP/IP connection managed by {@link ClientConnectionManager}. |
| * <p> |
| * There are two special attributes in the entries to be replicated: |
| * <ul> |
| * <li><tt>entryCSN</tt> - stores {@link CSN} of the entry. This attribute is |
| * used to compare the incoming operation from other replica is still |
| * valid. If the local <tt>entryCSN</tt> value is bigger then that of the |
| * incoming operation, it means conflict, and therefore an appropriate |
| * conflict resolution mechanism should get engaged.</li> |
| * <li><tt>entryDeleted</tt> - is <tt>TRUE</tt> if and only if the entry is |
| * deleted. The entry is not deleted immediately by a delete operation |
| * because <tt>entryCSN</tt> attribute should be retained for certain |
| * amount of time to determine whether the incoming change LOG, which |
| * affects an entry with the same DN, is a conflict (modification on a |
| * deleted entry) or not (creation of a new entry). You can purge old |
| * deleted entries and related change logs in {@link ReplicationStore} by |
| * calling {@link #purgeAgedData()}, or they will be purged automatically |
| * by periodic manner as you configured with {@link ReplicationConfiguration}. |
| * by calling {@link ReplicationConfiguration#setLogMaxAge(int)}. |
| * Because of this attribute, <tt>lookup</tt> and <tt>search</tt> |
| * operations are overrided to ignore entries with <tt>entryDeleted</tt> |
| * set to <tt>TRUE</tt>.</li> |
| * </ul> |
| * |
| * @org.apache.xbean.XBean |
| * |
| * @author The Apache Directory Project (dev@directory.apache.org) |
| * @version $Rev$, $Date$ |
| */ |
| public class ReplicationInterceptor extends BaseInterceptor |
| { |
| private static final Logger LOG = LoggerFactory.getLogger( ReplicationInterceptor.class ); |
| |
| /** The service name */ |
| public static final String NAME = "replicationService"; |
| |
| |
| private static final String ENTRY_CSN_OID = "1.3.6.1.4.1.18060.0.4.1.2.30"; |
| private static final String ENTRY_DELETED_OID = "1.3.6.1.4.1.18060.0.4.1.2.31"; |
| |
| /** |
| * default name is the service name? |
| */ |
| private String name = NAME; |
| |
| private DirectoryService directoryService; |
| private ReplicationConfiguration configuration; |
| private PartitionNexus nexus; |
| private OperationFactory operationFactory; |
| private ReplicationStore store; |
| private IoAcceptor registry; |
| private final ClientConnectionManager clientConnectionManager = new ClientConnectionManager( this ); |
| private Registries registries; |
| |
| |
| public ReplicationInterceptor() |
| { |
| } |
| |
| /** |
| * this interceptor has configuration so it might be useful to allow several instances in a chain. |
| * @return configured name for this interceptor. |
| */ |
| public String getName() { |
| return name; |
| } |
| |
| public void setName(String name) { |
| this.name = name; |
| } |
| |
| public ReplicationConfiguration getConfiguration() |
| { |
| return configuration; |
| } |
| |
| |
| public void setConfiguration(ReplicationConfiguration configuration) { |
| this.configuration = configuration; |
| } |
| |
| |
| public void init( DirectoryService directoryService ) throws NamingException |
| { |
| configuration.validate(); |
| // and then preserve frequently used ones |
| this.directoryService = directoryService; |
| registries = directoryService.getRegistries(); |
| nexus = directoryService.getPartitionNexus(); |
| store = configuration.getStore(); |
| operationFactory = new OperationFactory( directoryService, configuration ); |
| |
| // Initialize store and service |
| store.open( directoryService, configuration ); |
| boolean serviceStarted = false; |
| try |
| { |
| startNetworking(); |
| serviceStarted = true; |
| } |
| catch ( Exception e ) |
| { |
| throw new ReplicationServiceException( "Failed to initialize MINA ServiceRegistry.", e ); |
| } |
| finally |
| { |
| if ( !serviceStarted ) |
| { |
| // roll back |
| store.close(); |
| } |
| } |
| |
| purgeAgedData(); |
| } |
| |
| |
| private void startNetworking() throws Exception |
| { |
| registry = new SocketAcceptor(); |
| SocketAcceptorConfig config = new SocketAcceptorConfig(); |
| config.setReuseAddress( true ); |
| |
| config.getFilterChain().addLast( "protocol", |
| new ProtocolCodecFilter( new ReplicationServerProtocolCodecFactory() ) ); |
| |
| config.getFilterChain().addLast( "logger", new LoggingFilter() ); |
| |
| // bind server protocol provider |
| registry.bind( new InetSocketAddress( configuration.getServerPort() ), new ReplicationServerProtocolHandler( |
| this ), config ); |
| |
| clientConnectionManager.start( configuration ); |
| } |
| |
| |
| public void destroy() |
| { |
| stopNetworking(); |
| store.close(); |
| } |
| |
| |
| private void stopNetworking() |
| { |
| // close all open connections, deactivate all filters and service registry |
| try |
| { |
| clientConnectionManager.stop(); |
| } |
| catch ( Exception e ) |
| { |
| LOG.error( "[Replica-{}] Failed to stop the client connection manager.", configuration.getReplicaId() ); |
| LOG.error( "Stop failure exception: ", e ); |
| } |
| registry.unbindAll(); |
| } |
| |
| |
| /** |
| * Forces this context to send replication data to the peer replica immediately. |
| */ |
| public void replicate() |
| { |
| LOG.info( "[Replica-{}] Forcing replication...", configuration.getReplicaId() ); |
| this.clientConnectionManager.replicate(); |
| } |
| |
| /** |
| * Wake the sleeping (unconnected) replicas. |
| */ |
| public void interruptConnectors() |
| { |
| LOG.info( "[Replica-{}] Waking sleeping replicas...", configuration.getReplicaId() ); |
| this.clientConnectionManager.interruptConnectors(); |
| } |
| |
| |
| /** |
| * Purges old replication logs and the old entries marked as 'deleted' |
| * (i.e. {@link Constants#ENTRY_DELETED} is <tt>TRUE</tt>). This method |
| * should be called periodically to make sure the size of the DIT and |
| * {@link ReplicationStore} increase limitlessly. |
| * |
| * @see ReplicationConfiguration#setLogMaxAge(int) |
| * @see ReplicationLogCleanJob |
| * @throws javax.naming.NamingException on error |
| */ |
| public void purgeAgedData() throws NamingException |
| { |
| ServerEntry rootDSE = nexus.getRootDSE( null ); |
| EntryAttribute namingContextsAttr = rootDSE.get( SchemaConstants.NAMING_CONTEXTS_AT ); |
| |
| if ( ( namingContextsAttr == null ) || ( namingContextsAttr.size() == 0 ) ) |
| { |
| throw new NamingException( "No namingContexts attributes in rootDSE." ); |
| } |
| |
| CSN purgeCSN = new DefaultCSN( System.currentTimeMillis() - configuration.getLogMaxAge() * 1000L * 60L * 60L |
| * 24L, // convert days to millis |
| new ReplicaId( "ZZZZZZZZZZZZZZZZ" ), Integer.MAX_VALUE ); |
| ExprNode filter; |
| |
| try |
| { |
| filter = FilterParser.parse( "(&(" + ENTRY_CSN_OID + "<=" + purgeCSN.toOctetString() + ")(" + ENTRY_DELETED_OID |
| + "=TRUE))" ); |
| } |
| catch ( ParseException e ) |
| { |
| throw ( NamingException ) new NamingException().initCause( e ); |
| } |
| |
| // Iterate all context partitions to send all entries of them. |
| for ( Value<?> namingContext:namingContextsAttr ) |
| { |
| // Convert attribute value to JNDI name. |
| LdapDN contextName; |
| |
| contextName = new LdapDN( (String)namingContext.get() ); |
| |
| contextName.normalize( registries.getAttributeTypeRegistry().getNormalizerMapping() ); |
| LOG.info( "[Replica-{}] Purging aged data under '{}'", configuration.getReplicaId(), contextName ); |
| purgeAgedData( contextName, filter ); |
| } |
| |
| store.removeLogs( purgeCSN, false ); |
| } |
| |
| |
| private void purgeAgedData( LdapDN contextName, ExprNode filter ) throws NamingException |
| { |
| SearchControls ctrl = new SearchControls(); |
| ctrl.setSearchScope( SearchControls.SUBTREE_SCOPE ); |
| ctrl.setReturningAttributes( new String[] { "entryCSN", "entryDeleted" } ); |
| |
| NamingEnumeration<ServerSearchResult> e = nexus.search( |
| new SearchOperationContext( registries, contextName, AliasDerefMode.DEREF_ALWAYS, filter, ctrl ) ); |
| |
| List<LdapDN> names = new ArrayList<LdapDN>(); |
| |
| try |
| { |
| while ( e.hasMore() ) |
| { |
| ServerSearchResult sr = e.next(); |
| LdapDN name = sr.getDn(); |
| |
| if ( name.size() > contextName.size() ) |
| { |
| names.add( name ); |
| } |
| } |
| } |
| finally |
| { |
| e.close(); |
| } |
| |
| for ( LdapDN name : names ) |
| { |
| try |
| { |
| name.normalize( registries.getAttributeTypeRegistry().getNormalizerMapping() ); |
| ServerEntry entry = nexus.lookup( new LookupOperationContext( registries, name ) ); |
| LOG.info( "[Replica-{}] Purge: " + name + " (" + entry + ')', configuration.getReplicaId() ); |
| nexus.delete( new DeleteOperationContext( registries, name ) ); |
| } |
| catch ( NamingException ex ) |
| { |
| LOG.error( "[Replica-{}] Failed to fetch/delete: " + name, configuration.getReplicaId(), ex ); |
| } |
| } |
| } |
| |
| |
| public void add( NextInterceptor nextInterceptor, AddOperationContext addContext ) throws NamingException |
| { |
| Operation op = operationFactory.newAdd( |
| addContext.getDn(), addContext.getEntry() ); |
| op.execute( nexus, store, registries ); |
| } |
| |
| |
| @Override |
| public void delete( NextInterceptor next, DeleteOperationContext opContext ) throws NamingException |
| { |
| Operation op = operationFactory.newDelete( opContext.getDn() ); |
| op.execute( nexus, store, registries ); |
| } |
| |
| |
| public void modify( NextInterceptor next, ModifyOperationContext modifyContext ) throws NamingException |
| { |
| Operation op = operationFactory.newModify( modifyContext ); |
| op.execute( nexus, store, registries ); |
| } |
| |
| |
| @Override |
| public void move( NextInterceptor next, MoveOperationContext moveOpContext ) throws NamingException |
| { |
| Operation op = operationFactory.newMove( moveOpContext.getDn(), moveOpContext.getParent() ); |
| op.execute( nexus, store, registries ); |
| } |
| |
| |
| @Override |
| public void moveAndRename( NextInterceptor next, MoveAndRenameOperationContext moveAndRenameOpContext ) throws NamingException |
| { |
| Operation op = operationFactory.newMove( moveAndRenameOpContext.getDn(), |
| moveAndRenameOpContext.getParent(), moveAndRenameOpContext.getNewRdn(), |
| moveAndRenameOpContext.getDelOldDn() ); |
| op.execute( nexus, store, registries ); |
| } |
| |
| |
| @Override |
| public void rename( NextInterceptor next, RenameOperationContext renameOpContext ) throws NamingException |
| { |
| Operation op = operationFactory.newModifyRn( renameOpContext.getDn(), renameOpContext.getNewRdn(), renameOpContext.getDelOldDn() ); |
| op.execute( nexus, store, registries ); |
| } |
| |
| |
| public boolean hasEntry( NextInterceptor nextInterceptor, EntryOperationContext entryContext ) throws NamingException |
| { |
| // Ask others first. |
| boolean hasEntry = nextInterceptor.hasEntry( entryContext ); |
| |
| // If the entry exists, |
| if ( hasEntry ) |
| { |
| // Check DELETED attribute. |
| try |
| { |
| ServerEntry entry = nextInterceptor.lookup( new LookupOperationContext( registries, entryContext.getDn() ) ); |
| hasEntry = !isDeleted( entry ); |
| } |
| catch ( NameNotFoundException e ) |
| { |
| hasEntry = false; |
| } |
| } |
| |
| return hasEntry; |
| } |
| |
| |
| public ServerEntry lookup( NextInterceptor nextInterceptor, LookupOperationContext lookupContext ) throws NamingException |
| { |
| if ( lookupContext.getAttrsId() != null ) |
| { |
| boolean found = false; |
| |
| String[] attrIds = lookupContext.getAttrsIdArray(); |
| |
| // Look for 'entryDeleted' attribute is in attrIds. |
| for ( String attrId:attrIds ) |
| { |
| if ( Constants.ENTRY_DELETED.equals( attrId ) ) |
| { |
| found = true; |
| break; |
| } |
| } |
| |
| // If not exists, add one. |
| if ( !found ) |
| { |
| String[] newAttrIds = new String[attrIds.length + 1]; |
| System.arraycopy( attrIds, 0, newAttrIds, 0, attrIds.length ); |
| newAttrIds[attrIds.length] = Constants.ENTRY_DELETED; |
| lookupContext.setAttrsId( newAttrIds ); |
| } |
| } |
| |
| ServerEntry result = nextInterceptor.lookup( lookupContext ); |
| ensureNotDeleted( lookupContext.getDn(), result ); |
| return result; |
| } |
| |
| |
| @Override |
| public NamingEnumeration<ServerSearchResult> list( NextInterceptor nextInterceptor, ListOperationContext opContext ) throws NamingException |
| { |
| NamingEnumeration<ServerSearchResult> result = nextInterceptor.search( |
| new SearchOperationContext( |
| registries, opContext.getDn(), opContext.getAliasDerefMode(), |
| new PresenceNode( SchemaConstants.OBJECT_CLASS_AT_OID ), |
| new SearchControls() ) ); |
| |
| return new SearchResultFilteringEnumeration( result, new SearchControls(), InvocationStack.getInstance().peek(), |
| Constants.DELETED_ENTRIES_FILTER, "List replication filter" ); |
| } |
| |
| |
| @Override |
| public NamingEnumeration<ServerSearchResult> search( NextInterceptor nextInterceptor, SearchOperationContext opContext ) throws NamingException |
| { |
| SearchControls searchControls = opContext.getSearchControls(); |
| |
| if ( searchControls.getReturningAttributes() != null ) |
| { |
| String[] oldAttrIds = searchControls.getReturningAttributes(); |
| String[] newAttrIds = new String[oldAttrIds.length + 1]; |
| System.arraycopy( oldAttrIds, 0, newAttrIds, 0, oldAttrIds.length ); |
| newAttrIds[oldAttrIds.length] = Constants.ENTRY_DELETED.toLowerCase(); |
| searchControls.setReturningAttributes( newAttrIds ); |
| } |
| |
| NamingEnumeration<ServerSearchResult> result = nextInterceptor.search( |
| new SearchOperationContext( registries, opContext.getDn(), opContext.getAliasDerefMode(), opContext.getFilter(), searchControls ) ); |
| return new SearchResultFilteringEnumeration( result, searchControls, InvocationStack.getInstance().peek(), |
| Constants.DELETED_ENTRIES_FILTER, "Search Replication filter" ); |
| } |
| |
| |
| private void ensureNotDeleted( LdapDN name, ServerEntry entry ) throws NamingException { |
| if ( isDeleted( entry ) ) |
| { |
| LdapNameNotFoundException e = new LdapNameNotFoundException( "Deleted entry: " + name.getUpName() ); |
| e.setResolvedName( nexus.getMatchedName( new GetMatchedNameOperationContext( registries, name ) ) ); |
| throw e; |
| } |
| } |
| |
| |
| private boolean isDeleted( ServerEntry entry ) throws NamingException |
| { |
| if ( entry == null ) |
| { |
| return true; |
| } |
| |
| return entry.contains( Constants.ENTRY_DELETED, "TRUE" ); |
| } |
| |
| |
| public DirectoryService getDirectoryService() |
| { |
| return directoryService; |
| } |
| } |
