LMS-1966 Added JythonToplevelDataSetHandler.

SVN: 19406

datastore_server/source/java/ch/systemsx/cisd/etlserver/ThreadParameters.java

@@ -59,10 +59,12 @@ public final class ThreadParameters
     static final String GROUP_CODE_KEY = "group-code";
 
     @Private
+    public
     static final String INCOMING_DATA_COMPLETENESS_CONDITION =
             "incoming-data-completeness-condition";
 
     @Private
+    public
     static final String INCOMING_DATA_COMPLETENESS_CONDITION_MARKER_FILE = "marker-file";
 
     @Private
@@ -72,11 +74,13 @@ public final class ThreadParameters
             ThreadParameters.class);
 
     @Private
+    public
     static final String INCOMING_DIR = "incoming-dir";
 
     private static final String INCOMING_DIR_CREATE = "incoming-dir-create";
 
     @Private
+    public
     static final String DELETE_UNIDENTIFIED_KEY = "delete-unidentified";
 
     private static final String REPROCESS_FAULTY_DATASETS_NAME = "reprocess-faulty-datasets";

datastore_server/source/java/ch/systemsx/cisd/etlserver/registrator/AbstractOmniscientTopLevelDataSetRegistrator.java

@@ -46,6 +46,7 @@ import ch.systemsx.cisd.etlserver.TopLevelDataSetRegistratorGlobalState;
 import ch.systemsx.cisd.etlserver.utils.PostRegistrationExecutor;
 import ch.systemsx.cisd.etlserver.utils.PreRegistrationExecutor;
 import ch.systemsx.cisd.openbis.dss.generic.shared.dto.DataSetInformation;
+import ch.systemsx.cisd.openbis.generic.shared.basic.dto.DatabaseInstance;
 import ch.systemsx.cisd.openbis.generic.shared.dto.NewExternalData;
 
 /**
@@ -78,6 +79,8 @@ public abstract class AbstractOmniscientTopLevelDataSetRegistrator extends
 
         private final MarkerFileUtility markerFileUtility;
 
+        private final DatabaseInstance homeDatabaseInstance;
+
         private OmniscientTopLevelDataSetRegistratorState(
                 TopLevelDataSetRegistratorGlobalState globalState,
                 IStorageProcessor storageProcessor, ReentrantLock registrationLock,
@@ -97,6 +100,7 @@ public abstract class AbstractOmniscientTopLevelDataSetRegistrator extends
             this.markerFileUtility =
                     new MarkerFileUtility(operationLog, notificationLog, fileOperations,
                             storageProcessor);
+            this.homeDatabaseInstance = globalState.getOpenBisService().getHomeDatabaseInstance();
         }
 
         public TopLevelDataSetRegistratorGlobalState getGlobalState()
@@ -138,6 +142,11 @@ public abstract class AbstractOmniscientTopLevelDataSetRegistrator extends
         {
             return markerFileUtility;
         }
+
+        public DatabaseInstance getHomeDatabaseInstance()
+        {
+            return homeDatabaseInstance;
+        }
     }
 
     private final OmniscientTopLevelDataSetRegistratorState state;
@@ -207,7 +216,9 @@ public abstract class AbstractOmniscientTopLevelDataSetRegistrator extends
 
         DataSetRegistrationService service =
                 new DataSetRegistrationService(this, cleanAfterwardsAction);
+
         handleDataSet(incomingDataSetFile, service);
+        service.commit();
     }
 
     public boolean isStopped()

datastore_server/source/java/ch/systemsx/cisd/etlserver/registrator/DataSetRegistrationService.java

@@ -24,6 +24,7 @@ import ch.systemsx.cisd.etlserver.DataSetRegistrationAlgorithm;
 import ch.systemsx.cisd.etlserver.DataSetRegistrationAlgorithm.DataSetRegistrationAlgorithmState;
 import ch.systemsx.cisd.etlserver.DataSetRegistrationAlgorithm.IDataSetInApplicationServerRegistrator;
 import ch.systemsx.cisd.etlserver.DataSetRegistrationAlgorithm.IRollbackDelegate;
+import ch.systemsx.cisd.etlserver.DataSetRegistrationAlgorithmRunner;
 import ch.systemsx.cisd.etlserver.IDataStoreStrategy;
 import ch.systemsx.cisd.etlserver.TopLevelDataSetRegistratorGlobalState;
 import ch.systemsx.cisd.etlserver.registrator.AbstractOmniscientTopLevelDataSetRegistrator.OmniscientTopLevelDataSetRegistratorState;
@@ -74,22 +75,6 @@ public class DataSetRegistrationService implements IRollbackDelegate
         this.globalCleanAfterwardsAction = globalCleanAfterwardsAction;
     }
 
-    /**
-     * Factory method that creates a new registration details object.
-     */
-    public DataSetRegistrationDetails createRegistrationDetails()
-    {
-        return new DataSetRegistrationDetails();
-    }
-
-    /**
-     * Factory method that creates a new data set information object.
-     */
-    public DataSetInformation createDataSetInformation()
-    {
-        return new DataSetInformation();
-    }
-
     /**
      * Queue registration a data set and return a future for the data set that will be created.
      */
@@ -107,13 +92,16 @@ public class DataSetRegistrationService implements IRollbackDelegate
 
     public void commit()
     {
-
+        for (DataSetRegistrationAlgorithm registrationAlgorithm : dataSetRegistrations)
+        {
+            new DataSetRegistrationAlgorithmRunner(registrationAlgorithm).runAlgorithm();
+        }
         globalCleanAfterwardsAction.execute();
     }
 
     public void abort()
     {
-
+        dataSetRegistrations.clear();
     }
 
     private DataSetRegistrationAlgorithm createRegistrationAlgorithm(File incomingDataSetFile,
@@ -161,7 +149,8 @@ public class DataSetRegistrationService implements IRollbackDelegate
 
         private final DataSetInformation dataSetInformation;
 
-        DefaultApplicationServerRegistrator(AbstractOmniscientTopLevelDataSetRegistrator registrator,
+        DefaultApplicationServerRegistrator(
+                AbstractOmniscientTopLevelDataSetRegistrator registrator,
                 DataSetInformation dataSetInformation)
         {
             this.dataSetInformation = dataSetInformation;

datastore_server/source/java/ch/systemsx/cisd/etlserver/registrator/JythonTopLevelDataSetHandler.java

+/*
+ * Copyright 2011 ETH Zuerich, CISD
+ *
+ * Licensed 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 ch.systemsx.cisd.etlserver.registrator;
+
+import java.io.File;
+
+import org.python.util.PythonInterpreter;
+
+import ch.systemsx.cisd.common.filesystem.FileUtilities;
+import ch.systemsx.cisd.common.utilities.PropertyUtils;
+import ch.systemsx.cisd.etlserver.TopLevelDataSetRegistratorGlobalState;
+import ch.systemsx.cisd.openbis.dss.generic.shared.dto.DataSetInformation;
+
+/**
+ * @author Chandrasekhar Ramakrishnan
+ */
+public class JythonTopLevelDataSetHandler extends AbstractOmniscientTopLevelDataSetRegistrator
+{
+    // The key in the properties file
+    public static final String SCRIPT_PATH_KEY = "script-path";
+
+    private final String scriptPath;
+
+    /**
+     * @param globalState
+     */
+    protected JythonTopLevelDataSetHandler(TopLevelDataSetRegistratorGlobalState globalState)
+    {
+        super(globalState);
+
+        scriptPath =
+                PropertyUtils.getMandatoryProperty(globalState.getThreadParameters()
+                        .getThreadProperties(), SCRIPT_PATH_KEY);
+
+    }
+
+    @Override
+    protected void handleDataSet(File dataSetFile, DataSetRegistrationService service)
+    {
+        // Load the script
+        File scriptFile = new File(scriptPath);
+        String scriptString = FileUtilities.loadToString(scriptFile);
+
+        // Create an evaluator
+        PythonInterpreter interpreter = new PythonInterpreter();
+        interpreter.set("service", service);
+        interpreter.set("incoming", dataSetFile);
+        interpreter.set("state", getGlobalState());
+        setObjectFactory(interpreter);
+
+        interpreter.exec(scriptString);
+    }
+
+    /**
+     * Set the factory available to the python script. Subclasses may want to override.
+     */
+    protected void setObjectFactory(PythonInterpreter interpreter)
+    {
+        interpreter.set("factory", new JythonObjectFactory(getRegistratorState()));
+    }
+
+    public static class JythonObjectFactory
+    {
+        private final OmniscientTopLevelDataSetRegistratorState registratorState;
+
+        public JythonObjectFactory(OmniscientTopLevelDataSetRegistratorState registratorState)
+        {
+            this.registratorState = registratorState;
+        }
+
+        /**
+         * Factory method that creates a new registration details object.
+         */
+        public DataSetRegistrationDetails createRegistrationDetails()
+        {
+            DataSetRegistrationDetails registrationDetails = new DataSetRegistrationDetails();
+            registrationDetails.setDataSetInformation(createDataSetInformation());
+            return registrationDetails;
+        }
+
+        /**
+         * Factory method that creates a new data set information object.
+         */
+        public DataSetInformation createDataSetInformation()
+        {
+            DataSetInformation dataSetInfo = new DataSetInformation();
+            dataSetInfo.setInstanceCode(registratorState.getHomeDatabaseInstance().getCode());
+            dataSetInfo.setInstanceUUID(registratorState.getHomeDatabaseInstance().getUuid());
+
+            return dataSetInfo;
+        }
+    }
+}

datastore_server/sourceTest/java/ch/systemsx/cisd/etlserver/DataSetRegistrationAlgorithmTest.java

@@ -129,7 +129,6 @@ public class DataSetRegistrationAlgorithmTest extends AbstractFileSystemTestCase
         dataStoreStrategy = context.mock(IDataStoreStrategy.class);
         typeExtractor = context.mock(ITypeExtractor.class);
         storageProcessor = context.mock(IStorageProcessor.class);
-        // fileOperations = context.mock(IFileOperations.class);
         fileOperations = FileOperations.getInstance();
         dataSetValidator = context.mock(IDataSetValidator.class);
         mailClient = context.mock(IMailClient.class);

datastore_server/sourceTest/java/ch/systemsx/cisd/etlserver/registrator/JythonTopLevelDataSetRegistratorTest.java

+/*
+ * Copyright 2011 ETH Zuerich, CISD
+ *
+ * Licensed 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 ch.systemsx.cisd.etlserver.registrator;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Properties;
+
+import org.apache.commons.io.FileUtils;
+import org.jmock.Expectations;
+import org.jmock.Mockery;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import ch.systemsx.cisd.base.exceptions.IOExceptionUnchecked;
+import ch.systemsx.cisd.base.tests.AbstractFileSystemTestCase;
+import ch.systemsx.cisd.common.exceptions.ConfigurationFailureException;
+import ch.systemsx.cisd.common.filesystem.FileUtilities;
+import ch.systemsx.cisd.common.mail.From;
+import ch.systemsx.cisd.common.mail.IMailClient;
+import ch.systemsx.cisd.common.utilities.ExtendedProperties;
+import ch.systemsx.cisd.etlserver.IStorageProcessor;
+import ch.systemsx.cisd.etlserver.ITypeExtractor;
+import ch.systemsx.cisd.etlserver.ThreadParameters;
+import ch.systemsx.cisd.etlserver.TopLevelDataSetRegistratorGlobalState;
+import ch.systemsx.cisd.etlserver.validation.IDataSetValidator;
+import ch.systemsx.cisd.openbis.dss.generic.shared.IEncapsulatedOpenBISService;
+import ch.systemsx.cisd.openbis.dss.generic.shared.dto.DataSetInformation;
+import ch.systemsx.cisd.openbis.generic.shared.basic.dto.DataSetType;
+import ch.systemsx.cisd.openbis.generic.shared.basic.dto.DatabaseInstance;
+import ch.systemsx.cisd.openbis.generic.shared.basic.dto.Experiment;
+import ch.systemsx.cisd.openbis.generic.shared.basic.dto.Person;
+import ch.systemsx.cisd.openbis.generic.shared.dto.NewExternalData;
+import ch.systemsx.cisd.openbis.generic.shared.dto.StorageFormat;
+import ch.systemsx.cisd.openbis.generic.shared.dto.identifier.ExperimentIdentifierFactory;
+
+/**
+ * @author Chandrasekhar Ramakrishnan
+ */
+public class JythonTopLevelDataSetRegistratorTest extends AbstractFileSystemTestCase
+{
+    private static final String DATA_SET_CODE = "data-set-code";
+
+    private static final String DATABASE_INSTANCE_UUID = "db-uuid";
+
+    private static final DataSetType DATA_SET_TYPE = new DataSetType("O1");
+
+    private JythonTopLevelDataSetHandler handler;
+
+    private Mockery context;
+
+    private IEncapsulatedOpenBISService openBisService;
+
+    private IMailClient mailClient;
+
+    private IDataSetValidator dataSetValidator;
+
+    private File incomingDataSetFile;
+
+    private File subDataSet1;
+
+    private File subDataSet2;
+
+    @BeforeMethod
+    @Override
+    public void setUp() throws IOException
+    {
+        super.setUp();
+
+        context = new Mockery();
+        openBisService = context.mock(IEncapsulatedOpenBISService.class);
+        dataSetValidator = context.mock(IDataSetValidator.class);
+        mailClient = context.mock(IMailClient.class);
+    }
+
+    @Test
+    public void testDataSetRegistration()
+    {
+        setUpOpenBisExpectations();
+
+        createHandler();
+        incomingDataSetFile = createDirectory(workingDirectory, "data_set");
+
+        subDataSet1 = createDirectory(incomingDataSetFile, "sub_data_set_1");
+        subDataSet2 = createDirectory(incomingDataSetFile, "sub_data_set_2");
+
+        FileUtilities.writeToFile(new File(subDataSet1, "read1.me"), "hello world");
+        FileUtilities.writeToFile(new File(subDataSet2, "read2.me"), "hello world");
+
+        MockStorageProcessor.instance.storeRootDirectory = workingDirectory;
+
+        setUpDataSetValidatorExpectations();
+        setUpMailClientExpectations();
+
+        handler.handle(incomingDataSetFile);
+
+        assertEquals(2, MockStorageProcessor.instance.calledStoreDataCount);
+        assertEquals(2, MockStorageProcessor.instance.calledCommitCount);
+
+    }
+
+    @Test
+    public void testInvalidScriptPath()
+    {
+        setUpHomeDataBaseExpectations();
+
+        Properties threadProperties = new Properties();
+        threadProperties.put(ThreadParameters.INCOMING_DIR, "incoming");
+        threadProperties.put(ThreadParameters.INCOMING_DATA_COMPLETENESS_CONDITION,
+                ThreadParameters.INCOMING_DATA_COMPLETENESS_CONDITION_MARKER_FILE);
+        threadProperties.put(ThreadParameters.DELETE_UNIDENTIFIED_KEY, "false");
+        threadProperties.put(IStorageProcessor.STORAGE_PROCESSOR_KEY,
+                MockStorageProcessor.class.getName());
+        threadProperties.put(JythonTopLevelDataSetHandler.SCRIPT_PATH_KEY, "foo.py");
+
+        createHandler(threadProperties);
+        try
+        {
+            incomingDataSetFile = createDirectory(workingDirectory, "data_set");
+
+            subDataSet1 = createDirectory(incomingDataSetFile, "sub_data_set_1");
+            subDataSet2 = createDirectory(incomingDataSetFile, "sub_data_set_2");
+
+            FileUtilities.writeToFile(new File(subDataSet1, "read1.me"), "hello world");
+            FileUtilities.writeToFile(new File(subDataSet2, "read2.me"), "hello world");
+
+            handler.handle(incomingDataSetFile);
+
+            fail("The script should does not exist");
+        } catch (IOExceptionUnchecked ex)
+        {
+            assertTrue(ex.getCause().toString(), ex.getCause() instanceof FileNotFoundException);
+        }
+    }
+
+    @Test
+    public void testNoScriptPath()
+    {
+        setUpHomeDataBaseExpectations();
+
+        // omit the script path
+        Properties threadProperties = new Properties();
+        threadProperties.put(ThreadParameters.INCOMING_DIR, "incoming");
+        threadProperties.put(ThreadParameters.INCOMING_DATA_COMPLETENESS_CONDITION,
+                ThreadParameters.INCOMING_DATA_COMPLETENESS_CONDITION_MARKER_FILE);
+        threadProperties.put(ThreadParameters.DELETE_UNIDENTIFIED_KEY, "false");
+        threadProperties.put(IStorageProcessor.STORAGE_PROCESSOR_KEY,
+                MockStorageProcessor.class.getName());
+
+        try
+        {
+            createHandler(threadProperties);
+            fail("Should not be able to create the handler without specifiying a script");
+        } catch (ConfigurationFailureException ex)
+        {
+            assertEquals(
+                    "Given key 'script-path' not found in properties '[delete-unidentified, storage-processor, incoming-data-completeness-condition, incoming-dir]'",
+                    ex.getMessage());
+        }
+    }
+
+    public static final class MockStorageProcessor implements IStorageProcessor
+    {
+        static MockStorageProcessor instance;
+
+        int calledGetStoreRootDirectoryCount = 0;
+
+        int calledStoreDataCount = 0;
+
+        int calledCommitCount = 0;
+
+        File storeRootDirectory;
+
+        public MockStorageProcessor(ExtendedProperties props)
+        {
+            instance = this;
+        }
+
+        public File getStoreRootDirectory()
+        {
+            calledGetStoreRootDirectoryCount++;
+            return storeRootDirectory;
+        }
+
+        public void setStoreRootDirectory(File storeRootDirectory)
+        {
+            this.storeRootDirectory = storeRootDirectory;
+        }
+
+        public File storeData(DataSetInformation dataSetInformation, ITypeExtractor typeExtractor,
+                IMailClient mailClient, File incomingDataSetDirectory, File rootDir)
+        {
+            calledStoreDataCount++;
+            try
+            {
+                FileUtils.copyDirectory(incomingDataSetDirectory, rootDir);
+            } catch (IOException ex)
+            {
+                throw new IOExceptionUnchecked(ex);
+            }
+            return new File(rootDir, incomingDataSetDirectory.getName());
+        }
+
+        public void commit(File incomingDataSetDirectory, File storedDataDirectory)
+        {
+            calledCommitCount++;
+        }
+
+        public UnstoreDataAction rollback(File incomingDataSetDirectory, File storedDataDirectory,
+                Throwable exception)
+        {
+            return null;
+        }
+
+        public StorageFormat getStorageFormat()
+        {
+            return StorageFormat.PROPRIETARY;
+        }
+
+        public File tryGetProprietaryData(File storedDataDirectory)
+        {
+            return null;
+        }
+    }
+
+    private File createDirectory(File parentDir, String directoryName)
+    {
+        final File file = new File(parentDir, directoryName);
+        file.mkdir();
+        return file;
+    }
+
+    private void createHandler()
+    {
+        Properties threadProperties = new Properties();
+        threadProperties.put(ThreadParameters.INCOMING_DIR, "incoming");
+        threadProperties.put(ThreadParameters.INCOMING_DATA_COMPLETENESS_CONDITION,
+                ThreadParameters.INCOMING_DATA_COMPLETENESS_CONDITION_MARKER_FILE);
+        threadProperties.put(ThreadParameters.DELETE_UNIDENTIFIED_KEY, "false");
+        threadProperties.put(IStorageProcessor.STORAGE_PROCESSOR_KEY,
+                MockStorageProcessor.class.getName());
+        threadProperties.put(JythonTopLevelDataSetHandler.SCRIPT_PATH_KEY,
+                "sourceTest/java/ch/systemsx/cisd/etlserver/registrator/script.py");
+
+        createHandler(threadProperties);
+    }
+
+    private void createHandler(Properties threadProperties)
+    {
+        ThreadParameters threadParameters =
+                new ThreadParameters(threadProperties, "jython-handler-test");
+
+        TopLevelDataSetRegistratorGlobalState globalState =
+                new TopLevelDataSetRegistratorGlobalState("dss", openBisService, mailClient,
+                        dataSetValidator, true, threadParameters);
+
+        handler = new JythonTopLevelDataSetHandler(globalState);
+    }
+
+    private void setUpHomeDataBaseExpectations()
+    {
+        context.checking(new Expectations()
+            {
+                {
+
+                    DatabaseInstance databaseInstance = new DatabaseInstance();
+                    databaseInstance.setUuid(DATABASE_INSTANCE_UUID);
+                    one(openBisService).getHomeDatabaseInstance();
+                    will(returnValue(databaseInstance));
+                }
+            });
+    }
+
+    private void setUpOpenBisExpectations()
+    {
+        setUpHomeDataBaseExpectations();
+        context.checking(new Expectations()
+            {
+                {
+                    oneOf(openBisService).createDataSetCode();
+                    will(returnValue(DATA_SET_CODE + 1));
+
+                    oneOf(openBisService).createDataSetCode();
+                    will(returnValue(DATA_SET_CODE + 2));
+
+                    Experiment experiment = new Experiment();
+                    experiment.setIdentifier("/SPACE/PROJECT/EXP-CODE");
+                    experiment.setCode("EXP-CODE");
+                    Person registrator = new Person();
+                    registrator.setEmail("email@email.com");
+                    experiment.setRegistrator(registrator);
+
+                    exactly(2).of(openBisService).tryToGetExperiment(
+                            new ExperimentIdentifierFactory("/SPACE/PROJECT/EXP-CODE")
+                                    .createIdentifier());
+                    will(returnValue(experiment));
+
+                    exactly(2).of(openBisService).registerDataSet(
+                            with(any(DataSetInformation.class)), with(any(NewExternalData.class)));
+                }
+            });
+    }
+
+    private void setUpDataSetValidatorExpectations()
+    {
+        context.checking(new Expectations()
+            {
+                {
+                    one(dataSetValidator).assertValidDataSet(DATA_SET_TYPE, subDataSet1);
+                    one(dataSetValidator).assertValidDataSet(DATA_SET_TYPE, subDataSet2);
+                }
+            });
+    }
+
+    private void setUpMailClientExpectations()
+    {
+        context.checking(new Expectations()
+            {
+                {
+                    exactly(2).of(mailClient).sendMessage(with(any(String.class)),
+                            with(any(String.class)), with(aNull(String.class)),
+                            with(aNull(From.class)), with(any(String[].class)));
+                }
+            });
+    }
+}

datastore_server/sourceTest/java/ch/systemsx/cisd/etlserver/registrator/script.py

+import ch.systemsx.cisd.openbis.generic.shared.dto.identifier as identifier
+import java.io as io
+import ch.systemsx.cisd.openbis.generic.shared.basic.dto as dto
+
+# Create the Experiment Identifier
+identifier = identifier.ExperimentIdentifierFactory("/SPACE/PROJECT/EXP-CODE").createIdentifier()
+
+# Register data set 1
+registrationDetails = factory.createRegistrationDetails()
+dataSetInformation = registrationDetails.getDataSetInformation()
+dataSetInformation.setExperimentIdentifier(identifier)
+registrationDetails.setDataSetType(dto.DataSetType("O1"));
+service.queueDataSetRegistration(io.File(incoming, "sub_data_set_1"), registrationDetails)
+
+# Register data set 2
+registrationDetails = factory.createRegistrationDetails()
+dataSetInformation = registrationDetails.getDataSetInformation()
+dataSetInformation.setExperimentIdentifier(identifier)
+registrationDetails.setDataSetType(dto.DataSetType("O1"));
+service.queueDataSetRegistration(io.File(incoming, "sub_data_set_2"), registrationDetails)