OPENDJ-1602 (CR-5566) New pluggable storage based backend

Matthew Swift

17.18.2014 adfd690a621a8aa403ddcf6e58d15804c82de3a0

OPENDJ-1602 (CR-5566) New pluggable storage based backend

Add very primitive LDIFImport implementation which uses addEntry() to add each entry:

* no support for replacing existing entries
* assumes that storage is disk based and uses a single directory on the file system.

 opendj-sdk/opendj3-server-dev/src/server/org/opends/server/backends/pluggable/BackendImpl.java

@@ -31,7 +31,6 @@
import java.io.FilenameFilter;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
@@ -57,11 +56,10 @@
import org.opends.server.core.*;
import org.opends.server.extensions.DiskSpaceMonitor;
import org.opends.server.types.*;
import org.opends.server.util.RuntimeInformation;

import static org.opends.messages.BackendMessages.*;
import static org.opends.messages.JebMessages.*;
import static org.opends.server.backends.jeb.ConfigurableEnvironment.*;
import static org.opends.server.core.DirectoryServer.getServerErrorResultCode;
import static org.opends.server.util.ServerConstants.*;
import static org.opends.server.util.StaticUtils.*;

@@ -79,7 +77,7 @@
  private LocalDBBackendCfg cfg;
  /** The root JE container to use for this backend. */
  private RootContainer rootContainer;
  

  // FIXME: this is broken. Replace with read-write lock.
  /** A count of the total operation threads currently in the backend. */
  private final AtomicInteger threadTotalCount = new AtomicInteger(0);
@@ -738,97 +736,16 @@
  public LDIFImportResult importLDIF(LDIFImportConfig importConfig)
      throws DirectoryException
  {
    RuntimeInformation.logInfo();

    // If the backend already has the root container open, we must use the same
    // underlying root container
    boolean openRootContainer = rootContainer == null;

    // If the rootContainer is open, the backend is initialized by something else.
    // We can't do import while the backend is online.
    final ResultCode errorRC = DirectoryServer.getServerErrorResultCode();
    if(!openRootContainer)
    /*
     * If the rootContainer is open, the backend is initialized by something
     * else. We can't do import while the backend is online.
     */
    if (rootContainer != null)
    {
      throw new DirectoryException(errorRC, ERR_JEB_IMPORT_BACKEND_ONLINE.get());
      throw new DirectoryException(getServerErrorResultCode(),
          ERR_JEB_IMPORT_BACKEND_ONLINE.get());
    }

    try
    {
      final EnvironmentConfig envConfig = getEnvConfigForImport();

      if (!importConfig.appendToExistingData()
          && (importConfig.clearBackend() || cfg.getBaseDN().size() <= 1))
      {
        // We have the writer lock on the environment, now delete the
        // environment and re-open it. Only do this when we are
        // importing to all the base DNs in the backend or if the backend only
        // have one base DN.
        File parentDirectory = getFileForPath(cfg.getDBDirectory());
        File backendDirectory = new File(parentDirectory, cfg.getBackendId());
        // If the backend does not exist the import will create it.
        if (backendDirectory.exists())
        {
          EnvManager.removeFiles(backendDirectory.getPath());
        }
      }

      throw new NotImplementedException();
//      Importer importer = new Importer(importConfig, cfg, envConfig);
//      rootContainer = initializeRootContainer();
//      return importer.processImport(rootContainer);
    }
    catch (ExecutionException execEx)
    {
      logger.traceException(execEx);
      if (execEx.getCause() instanceof DirectoryException)
      {
        throw ((DirectoryException) execEx.getCause());
      }
      throw new DirectoryException(errorRC, ERR_EXECUTION_ERROR.get(execEx.getMessage()));
    }
    catch (InterruptedException intEx)
    {
      logger.traceException(intEx);
      throw new DirectoryException(errorRC, ERR_INTERRUPTED_ERROR.get(intEx.getMessage()));
    }
    catch (StorageRuntimeException e)
    {
      logger.traceException(e);
      throw new DirectoryException(errorRC, LocalizableMessage.raw(e.getMessage()));
    }
    catch (InitializationException ie)
    {
      logger.traceException(ie);
      throw new DirectoryException(errorRC, ie.getMessageObject());
    }
    catch (ConfigException ce)
    {
      logger.traceException(ce);
      throw new DirectoryException(errorRC, ce.getMessageObject());
    }
    finally
    {
      // leave the backend in the same state.
      try
      {
        if (rootContainer != null)
        {
          long startTime = System.currentTimeMillis();
          rootContainer.close();
          long finishTime = System.currentTimeMillis();
          long closeTime = (finishTime - startTime) / 1000;
          logger.info(NOTE_JEB_IMPORT_LDIF_ROOTCONTAINER_CLOSE, closeTime);
          rootContainer = null;
        }

        // Sync the environment to disk.
        logger.info(NOTE_JEB_IMPORT_CLOSING_DATABASE);
      }
      catch (StorageRuntimeException de)
      {
        logger.traceException(de);
      }
    }
    return new RootContainer(this, cfg).importLDIF(importConfig);
  }

  /**
@@ -987,7 +904,7 @@
              diskMonitor.setFullThreshold(newCfg.getDiskFullThreshold());
              diskMonitor.setLowThreshold(newCfg.getDiskLowThreshold());
            }
            

            // Put the new configuration in place.
            cfg = newCfg;
          }
@@ -1073,22 +990,6 @@
  }

  /**
   * Clears all the entries from the backend.  This method is for test cases
   * that use the JE backend.
   *
   * @throws  ConfigException  If an unrecoverable problem arises in the
   *                           process of performing the initialization.
   * @throws  StorageRuntimeException If an error occurs while removing the data.
   */
  public void clearBackend() throws ConfigException, StorageRuntimeException
  {
    // Determine the backend database directory.
    File parentDirectory = getFileForPath(cfg.getDBDirectory());
    File backendDirectory = new File(parentDirectory, cfg.getBackendId());
    EnvManager.removeFiles(backendDirectory.getPath());
  }

  /**
   * Creates a customized DirectoryException from the StorageRuntimeException
   * thrown by JE backend.
   *

 opendj-sdk/opendj3-server-dev/src/server/org/opends/server/backends/pluggable/EnvManager.java

File was deleted

 opendj-sdk/opendj3-server-dev/src/server/org/opends/server/backends/pluggable/RootContainer.java

@@ -34,6 +34,7 @@
import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.opendj.config.server.ConfigException;
import org.opends.messages.UtilityMessages;
import org.opends.server.admin.server.ConfigurationChangeListener;
import org.opends.server.admin.std.server.LocalDBBackendCfg;
import org.opends.server.api.Backend;
@@ -49,11 +50,22 @@
import org.opends.server.core.DirectoryServer;
import org.opends.server.types.ConfigChangeResult;
import org.opends.server.types.DN;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.Entry;
import org.opends.server.types.FilePermission;
import org.opends.server.types.InitializationException;
import org.opends.server.types.LDIFImportConfig;
import org.opends.server.types.LDIFImportResult;
import org.opends.server.types.OpenDsException;
import org.opends.server.util.LDIFException;
import org.opends.server.util.LDIFReader;
import org.opends.server.util.RuntimeInformation;

import static org.opends.messages.BackendMessages.*;
import static org.opends.messages.ConfigMessages.*;
import static org.opends.messages.JebMessages.*;
import static org.opends.messages.UtilityMessages.ERR_LDIF_SKIP;
import static org.opends.server.core.DirectoryServer.getServerErrorResultCode;
import static org.opends.server.util.StaticUtils.*;

/**
@@ -88,6 +100,8 @@
  /** The compressed schema manager for this backend. */
  private CompressedSchema compressedSchema;

  private File backendDirectory;



  /**
@@ -102,6 +116,8 @@
  {
    this.backend = backend;
    this.config = config;
    this.backendDirectory = new File(getFileForPath(config.getDBDirectory()),
        config.getBackendId());

    getMonitorProvider().enableFilterUseStats(config.isIndexFilterAnalyzerEnabled());
    getMonitorProvider().setMaxEntries(config.getIndexFilterAnalyzerMaxFilters());
@@ -114,23 +130,143 @@
    return storage;
  }

  /**
   * Opens the root container using the JE configuration object provided.
   *
   * @throws StorageRuntimeException       If a database error occurs when creating
   *                                 the environment.
   * @throws InitializationException If an initialization error occurs while
   *                                 creating the environment.
   * @throws ConfigException         If an configuration error occurs while
   *                                 creating the environment.
   */
  public void open()
      throws StorageRuntimeException, InitializationException, ConfigException
  LDIFImportResult importLDIF(LDIFImportConfig importConfig)
      throws DirectoryException
  {
    // Determine the backend database directory.
    File parentDirectory = getFileForPath(config.getDBDirectory());
    File backendDirectory = new File(parentDirectory, config.getBackendId());
    RuntimeInformation.logInfo();
    if (!importConfig.appendToExistingData()
        && (importConfig.clearBackend() || config.getBaseDN().size() <= 1))
    {
      removeFiles();
    }
    try
    {
      open();
      try
      {
        final LDIFReader reader;
        try
        {
          reader = new LDIFReader(importConfig);
        }
        catch (Exception e)
        {
          LocalizableMessage m = ERR_LDIF_BACKEND_CANNOT_CREATE_LDIF_READER.get(
                           stackTraceToSingleLineString(e));
          throw new DirectoryException(DirectoryServer.getServerErrorResultCode(),
                                       m, e);
        }

        while (true)
        {
          final Entry entry;
          try
          {
            entry = reader.readEntry();
            if (entry == null)
            {
              break;
            }
          }
          catch (LDIFException le)
          {
            if (!le.canContinueReading())
            {
              LocalizableMessage m = ERR_LDIF_BACKEND_ERROR_READING_LDIF
                  .get(stackTraceToSingleLineString(le));
              throw new DirectoryException(
                  DirectoryServer.getServerErrorResultCode(), m, le);
            }
            continue;
          }

          final DN dn = entry.getName();
          final EntryContainer ec = getEntryContainer(dn);
          if (ec == null)
          {
            final LocalizableMessage m = ERR_LDIF_SKIP.get(dn);
            logger.error(m);
            reader.rejectLastEntry(m);
            continue;
          }

          try
          {
            ec.addEntry(entry, null);
          }
          catch (DirectoryException e)
          {
            switch (e.getResultCode().asEnum())
            {
            case ENTRY_ALREADY_EXISTS:
              // TODO: support replace of existing entries.
              reader.rejectLastEntry(WARN_JEB_IMPORT_ENTRY_EXISTS.get());
              break;
            case NO_SUCH_OBJECT:
              reader.rejectLastEntry(ERR_JEB_IMPORT_PARENT_NOT_FOUND.get(dn
                  .parent()));
              break;
            default:
              // Not sure why it failed.
              reader.rejectLastEntry(e.getMessageObject());
              break;
            }
          }
        }
        return new LDIFImportResult(reader.getEntriesRead(),
            reader.getEntriesRejected(), reader.getEntriesIgnored());
      }
      finally
      {
        close();
      }
    }
    catch (DirectoryException e)
    {
      logger.traceException(e);
      throw e;
    }
    catch (OpenDsException e)
    {
      logger.traceException(e);
      throw new DirectoryException(getServerErrorResultCode(),
          e.getMessageObject());
    }
    catch (Exception e)
    {
      logger.traceException(e);
      throw new DirectoryException(getServerErrorResultCode(),
          LocalizableMessage.raw(e.getMessage()));
    }
  }

  private void removeFiles() throws StorageRuntimeException
  {
    if (!backendDirectory.isDirectory())
    {
      LocalizableMessage message = ERR_JEB_DIRECTORY_INVALID
          .get(backendDirectory.getPath());
      throw new StorageRuntimeException(message.toString());
    }

    try
    {
      File[] jdbFiles = backendDirectory.listFiles();
      for (File f : jdbFiles)
      {
        f.delete();
      }
    }
    catch (Exception e)
    {
      logger.traceException(e);
      LocalizableMessage message = ERR_JEB_REMOVE_FAIL.get(e.getMessage());
      throw new StorageRuntimeException(message.toString(), e);
    }
  }

  void open() throws StorageRuntimeException, ConfigException
  {
    // Create the directory if it doesn't exist.
    if (!backendDirectory.exists())
    {
@@ -186,9 +322,7 @@
      }
    }

    // Open the database environment
    storage = new PersistItStorage(backendDirectory, this.config);

    storage = new PersistItStorage(backendDirectory, config);
    compressedSchema = new DefaultCompressedSchema();
    try
    {
@@ -296,12 +430,11 @@
  private void openAndRegisterEntryContainers(WriteableStorage txn, Set<DN> baseDNs)
      throws StorageRuntimeException, InitializationException, ConfigException
  {
    EntryID id;
    EntryID highestID = null;
    for(DN baseDN : baseDNs)
    {
      EntryContainer ec = openEntryContainer(baseDN, null, txn);
      id = ec.getHighestEntryID(txn);
      EntryID id = ec.getHighestEntryID(txn);
      registerEntryContainer(baseDN, ec);
      if(highestID == null || id.compareTo(highestID) > 0)
      {