airavata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From shame...@apache.org
Subject [12/15] git commit: Moved job restriction check logic to SimpleOrchestratorImpl
Date Fri, 26 Sep 2014 15:19:58 GMT
Moved job restriction check logic to SimpleOrchestratorImpl


Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/b8cacbc3
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/b8cacbc3
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/b8cacbc3

Branch: refs/heads/orchestratorJobThrottleFeature
Commit: b8cacbc3decd71085d2f01b1a81367d33139ff84
Parents: 459b7cb
Author: shamrath <shameerainfo@gmail.com>
Authored: Wed Sep 24 10:31:37 2014 -0400
Committer: shamrath <shameerainfo@gmail.com>
Committed: Wed Sep 24 10:31:37 2014 -0400

----------------------------------------------------------------------
 .../main/resources/airavata-server.properties   |   1 -
 .../main/resources/airavata-server.properties   |   1 +
 .../handlers/GridPullMonitorHandler.java        |  16 ---
 .../monitor/impl/pull/qstat/HPCPullMonitor.java |   7 +-
 .../airavata/gfac/monitor/util/CommonUtils.java |  30 +++---
 .../core/utils/OrchestratorUtils.java           | 100 ++++++++++++++++++
 .../core/validator/impl/JobCountValidator.java  | 104 +++++++++++--------
 .../cpi/impl/SimpleOrchestratorImpl.java        |  49 ++++++++-
 8 files changed, 223 insertions(+), 85 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/airavata/blob/b8cacbc3/modules/configuration/server/src/main/resources/airavata-server.properties
----------------------------------------------------------------------
diff --git a/modules/configuration/server/src/main/resources/airavata-server.properties b/modules/configuration/server/src/main/resources/airavata-server.properties
index 42b1bc8..ba938f2 100644
--- a/modules/configuration/server/src/main/resources/airavata-server.properties
+++ b/modules/configuration/server/src/main/resources/airavata-server.properties
@@ -200,7 +200,6 @@ start.submitter=true
 embedded.mode=true
 enable.validation=true
 enable.job.restriction.validation=true
-global.max.job.count=50
 orchestrator=org.apache.airavata.orchestrator.server.OrchestratorServer
 
 ###---------------------------API Server module Configurations---------------------------###

http://git-wip-us.apache.org/repos/asf/airavata/blob/b8cacbc3/modules/credential-store-service/credential-store-webapp/src/main/resources/airavata-server.properties
----------------------------------------------------------------------
diff --git a/modules/credential-store-service/credential-store-webapp/src/main/resources/airavata-server.properties
b/modules/credential-store-service/credential-store-webapp/src/main/resources/airavata-server.properties
index 91a8ae6..bfc0504 100644
--- a/modules/credential-store-service/credential-store-webapp/src/main/resources/airavata-server.properties
+++ b/modules/credential-store-service/credential-store-webapp/src/main/resources/airavata-server.properties
@@ -287,6 +287,7 @@ threadpool.size=10
 start.submitter=true
 embedded.mode=true
 enable.validation=true
+enable.job.restriction.validation=true
 orchestrator=org.apache.airavata.orchestrator.server.OrchestratorServer
 
 ###---------------------------API Server module Configurations---------------------------###

http://git-wip-us.apache.org/repos/asf/airavata/blob/b8cacbc3/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/handlers/GridPullMonitorHandler.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/handlers/GridPullMonitorHandler.java
b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/handlers/GridPullMonitorHandler.java
index 2376520..6caf553 100644
--- a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/handlers/GridPullMonitorHandler.java
+++ b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/handlers/GridPullMonitorHandler.java
@@ -102,22 +102,6 @@ public class GridPullMonitorHandler extends ThreadedHandler implements
Watcher{
                 e.printStackTrace();
             }
             CommonUtils.addMonitortoQueue(hpcPullMonitor.getQueue(), monitorID);
-            if (ServerSettings.getEnableJobRestrictionValidation().equals("true")) {
-                try {
-                    TaskDetails taskDetails = monitorID.getJobExecutionContext().getTaskData();
-
-                    ComputeResourceDescription computeResourceDescription =
-                            CommonUtils.getComputeResourceDescription(taskDetails);
-                    if (computeResourceDescription.getBatchQueues().size() > 0 &&
-                            computeResourceDescription.getBatchQueues().get(0).getMaxJobsInQueue()
> 0) {
-
-                    CommonUtils.increaseZkJobCount(monitorID); // update change job count
to zookeeper
-                    }
-                } catch (Exception e) {
-                    logger.error("Error reading max job count from Computer Resource Description,"
+
-                            " zookeeper job count update process failed");
-                }
-            }
         } catch (AiravataMonitorException e) {
             logger.error("Error adding monitorID object to the queue with experiment ", monitorID.getExperimentID());
         }

http://git-wip-us.apache.org/repos/asf/airavata/blob/b8cacbc3/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/pull/qstat/HPCPullMonitor.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/pull/qstat/HPCPullMonitor.java
b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/pull/qstat/HPCPullMonitor.java
index e9098ac..022b0a6 100644
--- a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/pull/qstat/HPCPullMonitor.java
+++ b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/impl/pull/qstat/HPCPullMonitor.java
@@ -42,6 +42,7 @@ import org.apache.airavata.gsi.ssh.api.SSHApiException;
 import org.apache.airavata.credential.store.util.AuthenticationInfo;
 import org.apache.airavata.model.appcatalog.computeresource.ComputeResourceDescription;
 import org.apache.airavata.model.workspace.experiment.JobState;
+import org.apache.airavata.model.workspace.experiment.TaskDetails;
 import org.apache.airavata.model.workspace.experiment.TaskState;
 import org.apache.airavata.schemas.gfac.GsisshHostType;
 import org.apache.airavata.schemas.gfac.SSHHostType;
@@ -266,9 +267,11 @@ public class HPCPullMonitor extends PullMonitor {
             for (MonitorID completedJob : completedJobs) {
                 CommonUtils.removeMonitorFromQueue(queue, completedJob);
                 if (ServerSettings.getEnableJobRestrictionValidation().equals("true")) {
// is job restriction available?
+                    TaskDetails taskDetails = completedJob.getJobExecutionContext().getTaskData();
                     ComputeResourceDescription computeResourceDesc = CommonUtils.getComputeResourceDescription(
-                            completedJob.getJobExecutionContext().getTaskData());
-                    if (computeResourceDesc.getBatchQueuesSize() > 0 && computeResourceDesc.getBatchQueues().get(0).getMaxJobsInQueue()
> 0) {
+                            taskDetails);
+                    if (computeResourceDesc.getBatchQueuesSize() > 0 &&
+                            taskDetails.getTaskScheduling().getQueueName() != null) {
                         if (zk == null) {
                             zk = completedJob.getJobExecutionContext().getZk();
                         }

http://git-wip-us.apache.org/repos/asf/airavata/blob/b8cacbc3/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/util/CommonUtils.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/util/CommonUtils.java
b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/util/CommonUtils.java
index d4e29ca..abc6b69 100644
--- a/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/util/CommonUtils.java
+++ b/modules/gfac/gfac-monitor/src/main/java/org/apache/airavata/gfac/monitor/util/CommonUtils.java
@@ -41,6 +41,7 @@ import org.apache.airavata.gfac.monitor.UserMonitorData;
 import org.apache.airavata.gfac.monitor.exception.AiravataMonitorException;
 import org.apache.airavata.model.appcatalog.appdeployment.ApplicationDeploymentDescription;
 import org.apache.airavata.model.appcatalog.appinterface.ApplicationInterfaceDescription;
+import org.apache.airavata.model.appcatalog.computeresource.BatchQueue;
 import org.apache.airavata.model.appcatalog.computeresource.ComputeResourceDescription;
 import org.apache.airavata.model.workspace.experiment.TaskDetails;
 import org.apache.airavata.schemas.gfac.GsisshHostType;
@@ -243,23 +244,6 @@ public class CommonUtils {
     public static void updateZkWithJobCount(ZooKeeper zk, final Map<String, Integer>
changeCountMap, boolean isAdd) {
         StringBuilder changeZNodePaths = new StringBuilder();
         try {
-            if (zk == null || !zk.getState().isConnected()) {
-                try {
-                    final CountDownLatch countDownLatch = new CountDownLatch(1);
-                    zk = new ZooKeeper(AiravataZKUtils.getZKhostPort(), 6000, new Watcher()
{
-                        @Override
-                        public void process(WatchedEvent event) {
-                            countDownLatch.countDown();
-                        }
-                    });
-                    countDownLatch.await();
-                } catch (ApplicationSettingsException e) {
-                    logger.error("Error while reading zookeeper hostport string");
-                } catch (IOException e) {
-                    logger.error("Error while reconnect attempt to zookeeper where zookeeper
connection loss state");
-                }
-            }
-
             for (String path : changeCountMap.keySet()) {
                 if (isAdd) {
                     CommonUtils.checkAndCreateZNode(zk, path);
@@ -325,7 +309,8 @@ public class CommonUtils {
      */
     public static String getJobCountUpdatePath(MonitorID monitorID){
         return new StringBuilder("/").append(Constants.STAT).append("/").append(monitorID.getUserName())
-                .append("/").append(monitorID.getHost().getType().getHostAddress()).append("/").append(Constants.JOB).toString();
+                .append("/").append(monitorID.getHost().getType().getHostAddress()).append("/").append(Constants.JOB).
+                        append("/").append(monitorID.getJobExecutionContext().getTaskData().getTaskScheduling().getQueueName()).toString();
     }
 
     /**
@@ -380,4 +365,13 @@ public class CommonUtils {
         }
     }
 
+    public static BatchQueue getBatchQueueByName(List<BatchQueue> batchQueues, String
queueName) {
+        for (BatchQueue bQueue : batchQueues) {
+            if (bQueue.getQueueName().equals(queueName)) {
+                return bQueue;
+            }
+        }
+        return null;
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/airavata/blob/b8cacbc3/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/utils/OrchestratorUtils.java
----------------------------------------------------------------------
diff --git a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/utils/OrchestratorUtils.java
b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/utils/OrchestratorUtils.java
index 8a2d574..d51cce7 100644
--- a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/utils/OrchestratorUtils.java
+++ b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/utils/OrchestratorUtils.java
@@ -22,13 +22,31 @@ package org.apache.airavata.orchestrator.core.utils;
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Map;
 
+import org.apache.aiaravata.application.catalog.data.impl.AppCatalogFactory;
+import org.apache.airavata.common.exception.AiravataException;
 import org.apache.airavata.common.exception.ApplicationSettingsException;
+import org.apache.airavata.common.utils.AiravataUtils;
+import org.apache.airavata.common.utils.Constants;
+import org.apache.airavata.common.utils.RequestData;
 import org.apache.airavata.common.utils.ServerSettings;
 import org.apache.airavata.commons.gfac.type.HostDescription;
+import org.apache.airavata.credential.store.credential.AuditInfo;
+import org.apache.airavata.credential.store.store.CredentialReader;
+import org.apache.airavata.credential.store.store.CredentialReaderFactory;
+import org.apache.airavata.gfac.monitor.util.CommonUtils;
+import org.apache.airavata.model.appcatalog.computeresource.BatchQueue;
+import org.apache.airavata.model.appcatalog.computeresource.ComputeResourceDescription;
+import org.apache.airavata.model.appcatalog.computeresource.JobSubmissionInterface;
+import org.apache.airavata.model.appcatalog.computeresource.SSHJobSubmission;
+import org.apache.airavata.model.error.ValidatorResult;
 import org.apache.airavata.model.workspace.experiment.ComputationalResourceScheduling;
+import org.apache.airavata.model.workspace.experiment.Experiment;
 import org.apache.airavata.model.workspace.experiment.TaskDetails;
+import org.apache.airavata.model.workspace.experiment.WorkflowNodeDetails;
 import org.apache.airavata.orchestrator.core.OrchestratorConfiguration;
+import org.apache.airavata.orchestrator.core.context.OrchestratorContext;
 import org.apache.airavata.orchestrator.core.exception.OrchestratorException;
 import org.apache.airavata.orchestrator.core.impl.GFACEmbeddedJobSubmitter;
 import org.apache.airavata.orchestrator.core.job.JobSubmitter;
@@ -68,4 +86,86 @@ public class OrchestratorUtils {
 //            throw new OrchestratorException(e);
 //        }
 //    }
+
+    public  static String getCommunityUserName(Experiment experiment,
+                                               ComputeResourceDescription computeResourceDes,
+                                               TaskDetails taskID,
+                                               String credStoreToken) throws AiravataException
{
+        ValidatorResult result;
+        try {
+            if (computeResourceDes.getBatchQueuesSize() > 0) {
+                BatchQueue batchQueue = CommonUtils.getBatchQueueByName(computeResourceDes.getBatchQueues(),
+                        taskID.getTaskScheduling().getQueueName());
+                if (batchQueue == null) {
+                    throw new IllegalArgumentException("Invalid queue name, There is no queue
with name :" +
+                            taskID.getTaskScheduling().getQueueName());
+                }
+                int resourceMaxJobCount = batchQueue.getMaxJobsInQueue();
+                if (resourceMaxJobCount > 0) {
+                    for (JobSubmissionInterface jobSubmissionInterface : computeResourceDes.getJobSubmissionInterfaces())
{
+                        switch (jobSubmissionInterface.getJobSubmissionProtocol()) {
+                            case LOCAL:
+                                // nothing to do
+                                return null;
+                            case SSH:
+                                SSHJobSubmission sshJobSubmission =
+                                        AppCatalogFactory.getAppCatalog().getComputeResource().getSSHJobSubmission(
+                                                jobSubmissionInterface.getJobSubmissionInterfaceId());
+                                switch (sshJobSubmission.getSecurityProtocol()) {
+                                    case GSI:
+                                        // gsi
+                                        RequestData requestData = new RequestData(ServerSettings.getDefaultUserGateway());
+                                        requestData.setTokenId(credStoreToken);
+                                        return requestData.getMyProxyUserName();
+                                    case SSH_KEYS:
+                                        CredentialReader credentialReader = CredentialReaderFactory.createCredentialStoreReader();
+                                        AuditInfo auditInfo = credentialReader.getAuditInfo(experiment.getUserName(),
credStoreToken);
+                                        return auditInfo.getCommunityUser().getUserName();
+                                    // ssh
+                                    default:
+                                        //nothing to do
+                                }
+                            default:
+                                //nothing to do
+                        }
+                    }
+                    return null;
+
+                }// end of inner if
+            }// end of outer if
+            return null;
+        } catch (Exception e) {
+            throw new AiravataException("Exception while getting community user name ", e);
+        }
+
+    }
+
+    public static boolean isJobSpaceAvailable(String communityUserName, String computeHostName,
String queueName, int resourceMaxJobCount)
+            throws ApplicationSettingsException {
+        if (communityUserName == null) {
+            throw new IllegalArgumentException("Community user name should not be null");
+        }
+        if (computeHostName == null) {
+            throw new IllegalArgumentException("Compute resource should not be null");
+        }
+        String keyPath = new StringBuilder("/" + Constants.STAT).append("/").append(communityUserName)
+                .append("/").toString();
+        String key = keyPath + computeHostName + "/" + Constants.JOB + "/" + queueName;
+        Map<String, Integer> jobCountMap = AiravataUtils.getJobCountMap(OrchestratorContext.getZk());
+        if (jobCountMap.containsKey(key)) {
+            int count = jobCountMap.get(key);
+            logger.info("Submitted job count = " + count + ", max job count = " + resourceMaxJobCount);
+            if (count < resourceMaxJobCount) {
+                return true;
+            }
+        } else {
+            logger.info("Job count map doesn't has key : " + key);
+            return true;
+        }
+        logger.info("Resource " + computeHostName + " doesn't has space to submit another
job, " +
+                "Configured resource max job count is " + resourceMaxJobCount + ".");
+        return false;
+    }
+
+
 }

http://git-wip-us.apache.org/repos/asf/airavata/blob/b8cacbc3/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/JobCountValidator.java
----------------------------------------------------------------------
diff --git a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/JobCountValidator.java
b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/JobCountValidator.java
index 0fb98ac..f1cc5f9 100644
--- a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/JobCountValidator.java
+++ b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/JobCountValidator.java
@@ -29,8 +29,12 @@ import org.apache.airavata.common.utils.AiravataUtils;
 import org.apache.airavata.common.utils.Constants;
 import org.apache.airavata.common.utils.RequestData;
 import org.apache.airavata.common.utils.ServerSettings;
+import org.apache.airavata.credential.store.credential.AuditInfo;
+import org.apache.airavata.credential.store.store.CredentialReader;
+import org.apache.airavata.credential.store.store.CredentialReaderFactory;
 import org.apache.airavata.gfac.monitor.util.CommonUtils;
 import org.apache.airavata.model.appcatalog.appdeployment.ApplicationDeploymentDescription;
+import org.apache.airavata.model.appcatalog.computeresource.BatchQueue;
 import org.apache.airavata.model.appcatalog.computeresource.ComputeResourceDescription;
 import org.apache.airavata.model.appcatalog.computeresource.JobSubmissionInterface;
 import org.apache.airavata.model.appcatalog.computeresource.SSHJobSubmission;
@@ -44,6 +48,7 @@ import org.apache.airavata.persistance.registry.jpa.model.TaskDetail;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -58,58 +63,67 @@ public class JobCountValidator implements JobMetadataValidator {
         ValidatorResult result;
         try {
             ComputeResourceDescription computeResourceDes = CommonUtils.getComputeResourceDescription(taskID);
-            if (computeResourceDes.getBatchQueuesSize() > 0 &&
-                    computeResourceDes.getBatchQueues().get(0).getMaxJobsInQueue() > 0)
{
-                int resourceMaxJobCount = computeResourceDes.getBatchQueues().get(0).getMaxJobsInQueue();
-                for (JobSubmissionInterface jobSubmissionInterface : computeResourceDes.getJobSubmissionInterfaces())
{
-                    switch (jobSubmissionInterface.getJobSubmissionProtocol()) {
-                        case LOCAL:
-                            // nothing to do
-                            return new ValidatorResult(true);
-                        case SSH:
-                            SSHJobSubmission sshJobSubmission =
-                                    AppCatalogFactory.getAppCatalog().getComputeResource().getSSHJobSubmission(
-                                            jobSubmissionInterface.getJobSubmissionInterfaceId());
-                            switch (sshJobSubmission.getSecurityProtocol()) {
-                                case GSI:
-                                    // gsi
-                                    RequestData requestData = new RequestData(ServerSettings.getDefaultUserGateway());
-                                    requestData.setTokenId(credStoreToken);
-                                    return isJobSpaceAvailable(requestData.getMyProxyUserName(),
-                                            computeResourceDes.getHostName(), resourceMaxJobCount);
-                                case SSH_KEYS:
-                                    result = new ValidatorResult(false);
-                                    result.setErrorDetails("SSH_KEY base job count validation
is not yet implemented");
-                                    return result;
-                                // ssh
-                                default:
-                                    result = new ValidatorResult(false);
-                                    result.setErrorDetails("Doesn't support " + sshJobSubmission.getSecurityProtocol()
+
-                                            " protocol yet");
-                                    return result;
-                            }
-                        default:
-                            result = new ValidatorResult(false);
-                            result.setErrorDetails("Doesn't support " +
-                                    jobSubmissionInterface.getJobSubmissionProtocol() + "
protocol yet");
-                            return result;
-                    }
+            if (computeResourceDes.getBatchQueuesSize() > 0) {
+                BatchQueue batchQueue = CommonUtils.getBatchQueueByName(computeResourceDes.getBatchQueues(),
+                        taskID.getTaskScheduling().getQueueName());
+                if (batchQueue == null) {
+                    throw new IllegalArgumentException("Invalid queue name, There is no queue
with name :" +
+                            taskID.getTaskScheduling().getQueueName());
                 }
-                result = new ValidatorResult(false);
-                result.setErrorDetails("No JobSubmission interface found");
-                return result;
-            } else {
-                return new ValidatorResult(true);
-            }
+                int resourceMaxJobCount = batchQueue.getMaxJobsInQueue();
+                if (resourceMaxJobCount > 0) {
+                    for (JobSubmissionInterface jobSubmissionInterface : computeResourceDes.getJobSubmissionInterfaces())
{
+                        switch (jobSubmissionInterface.getJobSubmissionProtocol()) {
+                            case LOCAL:
+                                // nothing to do
+                                return new ValidatorResult(true);
+                            case SSH:
+                                SSHJobSubmission sshJobSubmission =
+                                        AppCatalogFactory.getAppCatalog().getComputeResource().getSSHJobSubmission(
+                                                jobSubmissionInterface.getJobSubmissionInterfaceId());
+                                switch (sshJobSubmission.getSecurityProtocol()) {
+                                    case GSI:
+                                        // gsi
+                                        RequestData requestData = new RequestData(ServerSettings.getDefaultUserGateway());
+                                        requestData.setTokenId(credStoreToken);
+                                        return isJobSpaceAvailable(requestData.getMyProxyUserName(),
+                                                computeResourceDes.getHostName(), batchQueue.getQueueName(),
resourceMaxJobCount);
+                                    case SSH_KEYS:
+                                        CredentialReader credentialReader = CredentialReaderFactory.createCredentialStoreReader();
+                                        AuditInfo auditInfo = credentialReader.getAuditInfo(experiment.getUserName(),
credStoreToken);
+                                        return isJobSpaceAvailable(auditInfo.getCommunityUser().getUserName(),
+                                                computeResourceDes.getHostName(), batchQueue.getQueueName(),
resourceMaxJobCount);
+                                    // ssh
+                                    default:
+                                        result = new ValidatorResult(false);
+                                        result.setErrorDetails("Doesn't support " + sshJobSubmission.getSecurityProtocol()
+
+                                                " protocol yet");
+                                        return result;
+                                }
+                            default:
+                                result = new ValidatorResult(false);
+                                result.setErrorDetails("Doesn't support " +
+                                        jobSubmissionInterface.getJobSubmissionProtocol()
+ " protocol yet");
+                                return result;
+                        }
+                    }
+                    result = new ValidatorResult(false);
+                    result.setErrorDetails("No JobSubmission interface found");
+                    return result;
+
+                }// end of inner if
+            }// end of outer if
+            return new ValidatorResult(true);
         } catch (Exception e) {
+            logger.error("Exception occur while running job count validation process ", e);
             result = new ValidatorResult(false);
-            result.setErrorDetails("Exception occur while running validation process ");
+            result.setErrorDetails("Exception occur while running job count validation process
");
             return result;
         }
 
     }
 
-    private ValidatorResult isJobSpaceAvailable(String communityUserName, String computeHostName,
int resourceMaxJobCount)
+    private ValidatorResult isJobSpaceAvailable(String communityUserName, String computeHostName,
String queueName, int resourceMaxJobCount)
             throws ApplicationSettingsException {
         if (communityUserName == null) {
             throw new IllegalArgumentException("Community user name should not be null");
@@ -119,7 +133,7 @@ public class JobCountValidator implements JobMetadataValidator {
         }
         String keyPath = new StringBuilder("/" + Constants.STAT).append("/").append(communityUserName)
                 .append("/").toString();
-        String key = keyPath + computeHostName + "/" + Constants.JOB;
+        String key = keyPath + computeHostName + "/" + Constants.JOB + "/" + queueName;
         Map<String, Integer> jobCountMap = AiravataUtils.getJobCountMap(OrchestratorContext.getZk());
         if (jobCountMap.containsKey(key)) {
             int count = jobCountMap.get(key);

http://git-wip-us.apache.org/repos/asf/airavata/blob/b8cacbc3/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/impl/SimpleOrchestratorImpl.java
----------------------------------------------------------------------
diff --git a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/impl/SimpleOrchestratorImpl.java
b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/impl/SimpleOrchestratorImpl.java
index a950469..c0760dd 100644
--- a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/impl/SimpleOrchestratorImpl.java
+++ b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/impl/SimpleOrchestratorImpl.java
@@ -20,14 +20,23 @@
 */
 package org.apache.airavata.orchestrator.cpi.impl;
 
+import org.apache.airavata.common.exception.AiravataException;
+import org.apache.airavata.common.utils.Constants;
+import org.apache.airavata.common.utils.ServerSettings;
+import org.apache.airavata.gfac.monitor.util.CommonUtils;
+import org.apache.airavata.model.appcatalog.computeresource.BatchQueue;
+import org.apache.airavata.model.appcatalog.computeresource.ComputeResourceDescription;
 import org.apache.airavata.model.error.LaunchValidationException;
 import org.apache.airavata.model.error.ValidationResults;
 import org.apache.airavata.model.error.ValidatorResult;
 import org.apache.airavata.model.util.ExperimentModelUtil;
 import org.apache.airavata.model.workspace.experiment.*;
+import org.apache.airavata.orchestrator.core.context.OrchestratorContext;
 import org.apache.airavata.orchestrator.core.exception.OrchestratorException;
 import org.apache.airavata.orchestrator.core.job.JobSubmitter;
+import org.apache.airavata.orchestrator.core.utils.OrchestratorUtils;
 import org.apache.airavata.orchestrator.core.validator.JobMetadataValidator;
+import org.apache.airavata.orchestrator.core.validator.impl.JobCountValidator;
 import org.apache.airavata.registry.cpi.ChildDataType;
 import org.apache.airavata.registry.cpi.Registry;
 import org.apache.airavata.registry.cpi.RegistryModelType;
@@ -35,7 +44,9 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 
 public class SimpleOrchestratorImpl extends AbstractOrchestrator{
@@ -91,18 +102,50 @@ public class SimpleOrchestratorImpl extends AbstractOrchestrator{
         }
     }
 
-    public boolean launchExperiment(Experiment experiment, WorkflowNodeDetails workflowNode,
TaskDetails task,String tokenId) throws OrchestratorException {
+    public boolean launchExperiment(Experiment experiment, WorkflowNodeDetails workflowNode,
TaskDetails task,
+                                    String tokenId) throws OrchestratorException {
         // we give higher priority to userExperimentID
         String experimentId = experiment.getExperimentID();
         String taskId = task.getTaskID();
         // creating monitorID to register with monitoring queue
         // this is a special case because amqp has to be in place before submitting the job
         try {
-            return jobSubmitter.submit(experimentId, taskId,tokenId);
+            if (ServerSettings.getEnableJobRestrictionValidation().equals("true") &&
+                    task.getTaskScheduling().getQueueName() != null) {
+                ComputeResourceDescription computeResourceDes = CommonUtils.getComputeResourceDescription(task);
+                String communityUserName = OrchestratorUtils.getCommunityUserName(experiment,
computeResourceDes, task,
+                        tokenId);
+                BatchQueue batchQueue = CommonUtils.getBatchQueueByName(computeResourceDes.getBatchQueues(),
+                        task.getTaskScheduling().getQueueName());
+
+                synchronized (this) {
+                    boolean spaceAvaialble = OrchestratorUtils.isJobSpaceAvailable(communityUserName,
+                            computeResourceDes.getHostName(), batchQueue.getQueueName(),
batchQueue.getMaxJobsInQueue());
+                    if (spaceAvaialble) {
+                        if (jobSubmitter.submit(experimentId, taskId, tokenId)) {
+                            logger.info("Job submitted, experiment Id : " + experimentId
+ " , task Id : " + taskId);
+                            Map<String, Integer> jobUpdateMap = new HashMap<String,
Integer>();
+                            StringBuilder sb = new StringBuilder("/").append(Constants.STAT).append("/")
+                                    .append(communityUserName).append("/").append(computeResourceDes.getHostName())
+                                    .append("/").append(Constants.JOB).append("/").append(batchQueue.getQueueName());
+                            jobUpdateMap.put(sb.toString(), 1);
+                            CommonUtils.updateZkWithJobCount(OrchestratorContext.getZk(),
jobUpdateMap, true); // update change job count to zookeeper
+                            return true;
+                        } else {
+                            return false;
+                        }
+                    } else {
+                        throw new AiravataException("Please honour to the max job submission
restriction policy," +
+                                " max count is " + batchQueue.getMaxJobsInQueue());
+                    }
+                }// end of synchronized block
+            } else {
+                logger.info("Ignored job throttling");
+                return jobSubmitter.submit(experimentId, taskId, tokenId);
+            }
         } catch (Exception e) {
             throw new OrchestratorException("Error launching the job", e);
         }
-
     }
 
     /**


Mime
View raw message