From d58009544ed35fd144057f60a0a4a4a61b5f4f08 Mon Sep 17 00:00:00 2001 From: Gianpaolo Coro Date: Tue, 24 Mar 2015 11:11:03 +0000 Subject: [PATCH] git-svn-id: https://svn.d4science.research-infrastructures.eu/gcube/trunk/data-analysis/EcologicalEngineSmartExecutor@113696 82a268e6-3cf1-43bd-a215-b396298e98cf --- .../job/management/QueueJobManager.java | 36 ++++++-- .../tests/RegressionTestBiOnym1Matcher.java | 87 +++++++++++++++++++ 2 files changed, 117 insertions(+), 6 deletions(-) create mode 100644 src/main/java/org/gcube/dataanalysis/executor/tests/RegressionTestBiOnym1Matcher.java diff --git a/src/main/java/org/gcube/dataanalysis/executor/job/management/QueueJobManager.java b/src/main/java/org/gcube/dataanalysis/executor/job/management/QueueJobManager.java index a784862..7ed53e8 100644 --- a/src/main/java/org/gcube/dataanalysis/executor/job/management/QueueJobManager.java +++ b/src/main/java/org/gcube/dataanalysis/executor/job/management/QueueJobManager.java @@ -16,6 +16,8 @@ import javax.jms.MessageListener; import org.apache.activemq.ActiveMQConnection; import org.gcube.common.clients.ProxyBuilderImpl; +import org.gcube.common.resources.gcore.ServiceEndpoint; +import org.gcube.common.resources.gcore.ServiceEndpoint.AccessPoint; import org.gcube.common.scope.api.ScopeProvider; import org.gcube.contentmanagement.blobstorage.resource.StorageObject; import org.gcube.contentmanagement.blobstorage.service.IClient; @@ -30,6 +32,8 @@ import org.gcube.dataanalysis.executor.messagequeue.Producer; import org.gcube.dataanalysis.executor.messagequeue.QCONSTANTS; import org.gcube.dataanalysis.executor.messagequeue.QueueManager; import org.gcube.dataanalysis.executor.scripts.ScriptIOWorker; +import org.gcube.resources.discovery.client.api.DiscoveryClient; +import org.gcube.resources.discovery.client.queries.api.SimpleQuery; import org.gcube.vremanagement.executor.api.SmartExecutor; import org.gcube.vremanagement.executor.api.types.LaunchParameter; import org.gcube.vremanagement.executor.client.plugins.ExecutorPlugin; @@ -37,6 +41,7 @@ import org.gcube.vremanagement.executor.client.plugins.query.SmartExecutorPlugin import org.gcube.vremanagement.executor.client.plugins.query.filter.ListEndpointDiscoveryFilter; import org.gcube.vremanagement.executor.client.plugins.query.filter.SpecificEndpointDiscoveryFilter; import org.gcube.vremanagement.executor.client.proxies.SmartExecutorProxy; +import static org.gcube.resources.discovery.icclient.ICFactory.*; public class QueueJobManager { @@ -273,11 +278,10 @@ public class QueueJobManager { AnalysisLogger.getLogger().info("Contacting " + actualNumberOfNodes + " Nodes"); // set globals setGlobalVars(serviceClass, serviceName, owner, localDir, remoteDir, outputDir, script, arguments, configuration, deletefiles); - // if not yet uploaded , upload required files - uploadFilesOnStorage(forceUpload); - // initializing queue setQueueVariables(); + // if not yet uploaded , upload required files + uploadFilesOnStorage(forceUpload); // broadcast a message to all executors for purging previous queues // purgeQueues(); createClientProducer(); @@ -423,7 +427,7 @@ public class QueueJobManager { AnalysisLogger.getLogger().info("Broadcasting to node " + (order + 1) + " on " + selectedEPR); - /* + /*OLD EXECUTOR CALL // run the executor script ExecutorCall call = new ExecutorCall(pluginName, gscope); call.setEndpointReference(selectedEPR); @@ -512,6 +516,26 @@ public class QueueJobManager { return numberOfEP; } */ + public String getQueueURL(String scope) throws Exception{ + //set the scope provider first! + //Service + //MessageBroker + + ScopeProvider.instance.set(scope); + SimpleQuery query = queryFor(ServiceEndpoint.class); + query.addCondition("$resource/Profile/Category/text() eq 'Service' and $resource/Profile/Name eq 'MessageBroker' "); + DiscoveryClient client = clientFor(ServiceEndpoint.class); + List resources = client.submit(query); + if (resources==null || resources.size()==0){ + throw new Exception("No Message-Queue available in scope "+scope); + } + else{ + AccessPoint ap = resources.get(0).profile().accessPoints().iterator().next(); + String queue = ap.address(); + AnalysisLogger.getLogger().debug("Found AMQ Url : "+queue); + return queue; + } + } private void setQueueVariables() throws Exception { queueName = "D4ScienceJob"; // + session; @@ -520,8 +544,8 @@ public class QueueJobManager { // TODO Check THIS - queueURL = gscope.getServiceMap().getEndpoints(GHNContext.MSGBROKER).iterator().next().getAddress().toString(); - +// queueURL = gscope.getServiceMap().getEndpoints(GHNContext.MSGBROKER).iterator().next().getAddress().toString(); + queueURL = getQueueURL(scope); //tests on ecosystem diff --git a/src/main/java/org/gcube/dataanalysis/executor/tests/RegressionTestBiOnym1Matcher.java b/src/main/java/org/gcube/dataanalysis/executor/tests/RegressionTestBiOnym1Matcher.java new file mode 100644 index 0000000..8d13174 --- /dev/null +++ b/src/main/java/org/gcube/dataanalysis/executor/tests/RegressionTestBiOnym1Matcher.java @@ -0,0 +1,87 @@ +package org.gcube.dataanalysis.executor.tests; + +import org.gcube.dataanalysis.ecoengine.configuration.AlgorithmConfiguration; +import org.gcube.dataanalysis.executor.generators.D4ScienceDistributedProcessing; +import org.gcube.dataanalysis.executor.nodes.transducers.bionym.BionymFlexibleWorkflowTransducer; +import org.gcube.dataanalysis.executor.nodes.transducers.bionym.utils.YasmeenGlobalParameters; + +public class RegressionTestBiOnym1Matcher { + + public static void executeWF(String[] args) throws Exception { + // Generate + AlgorithmConfiguration config = new AlgorithmConfiguration(); + config.setConfigPath("./cfg/"); + + config.setParam("DatabaseUserName","utente"); + config.setParam("DatabasePassword","d4science"); + config.setParam("DatabaseURL","jdbc:postgresql://statistical-manager.d.d4science.research-infrastructures.eu/testdb"); + + config.setParam(YasmeenGlobalParameters.parserNameParam,YasmeenGlobalParameters.BuiltinParsers.SIMPLE.name()); + config.setParam(YasmeenGlobalParameters.taxaAuthorityFileParam,YasmeenGlobalParameters.BuiltinDataSources.ASFIS.name()); + config.setParam(YasmeenGlobalParameters.activatePreParsingProcessing,"true"); + config.setParam(YasmeenGlobalParameters.useStemmedGenusAndSpecies,"false"); + + + config.setParam(BionymFlexibleWorkflowTransducer.matcherParamPrefix+"_"+1,YasmeenGlobalParameters.BuiltinMatchers.GSAy.name()); + config.setParam(BionymFlexibleWorkflowTransducer.thresholdParamPrefix+"_"+1,"0.6"); + config.setParam(BionymFlexibleWorkflowTransducer.maxresultsParamPrefix+"_"+1,"10"); + + config.setParam(BionymFlexibleWorkflowTransducer.destinationTableParam, "taxamatchoutputlocal"); + config.setParam(BionymFlexibleWorkflowTransducer.destinationTableLableParam, "taxamatchoutputlabel"); + //2 + config.setParam(BionymFlexibleWorkflowTransducer.originTableParam, "generic_id471e6d50_d243_4112_bc07_e22152438e5c"); + config.setParam(BionymFlexibleWorkflowTransducer.rawnamesColumnParam, "field0"); + + + config.setAgent("BIONYM"); + + config.setPersistencePath("./"); + config.setGcubeScope( "/gcube"); +// config.setGcubeScope( "/d4science.research-infrastructures.eu"); + config.setParam("ServiceUserName", "gianpaolo.coro"); + + config.setParam("DatabaseDriver", "org.postgresql.Driver"); + + generate(config); + + } + + public static void main(String []args) throws Exception{ + + for (int i=0;i<1;i++){ + executeWF(args); + } + } + + + public static void generate(AlgorithmConfiguration config) throws Exception { + + D4ScienceDistributedProcessing generator = new D4ScienceDistributedProcessing(config); + generator.init(); + + if (generator != null) { + long t0 = System.currentTimeMillis(); + TestGenerator tgs = new TestGenerator(generator); + Thread t = new Thread(tgs); + t.start(); + while (generator.getStatus() < 100) { + + String resLoad = generator.getResourceLoad(); + String ress = generator.getResources(); + String species = generator.getLoad(); + System.out.println("LOAD: " + resLoad); + System.out.println("RESOURCES: " + ress); + System.out.println("SPECIES: " + species); + System.out.println("STATUS: " + generator.getStatus()); + Thread.sleep(5000); + } + System.out.println("FINAL STATUS: " + generator.getStatus()+ " ELAPSED "+(System.currentTimeMillis()-t0)); + + } + else + System.out.println("Generator Algorithm Not Supported"); + +// generator.generate(); +// } + } +}