This is an automated email from the ASF dual-hosted git repository.
isjarana pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/airavata.git
The following commit(s) were added to refs/heads/develop by this push:
new fe6a08210d support multiple compute resources
new 79394bfcb7 Merge pull request #390 from isururanawaka/metaschedular
fe6a08210d is described below
commit fe6a08210d7569bb268e845ca288fd05db1937e5
Author: Isuru Ranawaka <[email protected]>
AuthorDate: Tue Mar 7 22:02:53 2023 -0500
support multiple compute resources
---
.../api/server/handler/AiravataServerHandler.java | 27 ++++++++++-
.../selection/MultipleComputeResourcePolicy.java | 56 ++++++++++++++++------
2 files changed, 67 insertions(+), 16 deletions(-)
diff --git
a/airavata-api/airavata-api-server/src/main/java/org/apache/airavata/api/server/handler/AiravataServerHandler.java
b/airavata-api/airavata-api-server/src/main/java/org/apache/airavata/api/server/handler/AiravataServerHandler.java
index b957e5ab1c..04e1a86bcf 100644
---
a/airavata-api/airavata-api-server/src/main/java/org/apache/airavata/api/server/handler/AiravataServerHandler.java
+++
b/airavata-api/airavata-api-server/src/main/java/org/apache/airavata/api/server/handler/AiravataServerHandler.java
@@ -2185,12 +2185,18 @@ public class AiravataServerHandler implements
Airavata.Iface {
// Verify user has READ access to Application Deployment
final String appInterfaceId = experiment.getExecutionId();
- final String resourceHostId =
experiment.getUserConfigurationData().getComputationalResourceScheduling().getResourceHostId();
ApplicationInterfaceDescription applicationInterfaceDescription =
regClient.getApplicationInterface(appInterfaceId);
+
List<String> appModuleIds =
applicationInterfaceDescription.getApplicationModules();
// Assume that there is only one app module for this interface
(otherwise, how could we figure out the deployment)
String appModuleId = appModuleIds.get(0);
List<ApplicationDeploymentDescription>
applicationDeploymentDescriptions =
regClient.getApplicationDeployments(appModuleId);
+
+
+ if
(!experiment.getUserConfigurationData().isAiravataAutoSchedule()) {
+ final String resourceHostId =
experiment.getUserConfigurationData().getComputationalResourceScheduling().getResourceHostId();
+
+
Optional<ApplicationDeploymentDescription>
applicationDeploymentDescription = applicationDeploymentDescriptions
.stream()
.filter(dep ->
dep.getComputeHostId().equals(resourceHostId))
@@ -2200,8 +2206,25 @@ public class AiravataServerHandler implements
Airavata.Iface {
if (!sharingClient.userHasAccess(gatewayId, username + "@" +
gatewayId, appDeploymentId, gatewayId + ":READ")) {
throw new AuthorizationException("User " + username + " in
gateway " + gatewayId + " doesn't have access to app deployment " +
appDeploymentId);
}
- } else {
+ }
+ else {
throw new InvalidRequestException("Application deployment
doesn't exist for application interface " + appInterfaceId + " and host " +
resourceHostId + " in gateway " + gatewayId);
+ }} else if
(experiment.getUserConfigurationData().getAutoScheduledCompResourceSchedulingList()
!= null &&
+
!experiment.getUserConfigurationData().getAutoScheduledCompResourceSchedulingList().isEmpty())
{
+ List<ComputationalResourceSchedulingModel>
compResourceSchedulingList = experiment.
+
getUserConfigurationData().getAutoScheduledCompResourceSchedulingList();
+ for(ComputationalResourceSchedulingModel crScheduling:
compResourceSchedulingList){
+ Optional<ApplicationDeploymentDescription>
applicationDeploymentDescription = applicationDeploymentDescriptions
+ .stream()
+ .filter(dep ->
dep.getComputeHostId().equals(crScheduling.getResourceHostId()))
+ .findFirst();
+ if (applicationDeploymentDescription.isPresent()) {
+ final String appDeploymentId =
applicationDeploymentDescription.get().getAppDeploymentId();
+ if (!sharingClient.userHasAccess(gatewayId, username +
"@" + gatewayId, appDeploymentId, gatewayId + ":READ")) {
+ throw new AuthorizationException("User " + username
+ " in gateway " + gatewayId + " doesn't have access to app deployment " +
appDeploymentId);
+ }
+ }
+ }
}
submitExperiment(gatewayId, airavataExperimentId);
logger.info("Experiment with ExpId: " + airavataExperimentId + "
was submitted in gateway with gatewayID: " + gatewayId);
diff --git
a/modules/airavata-metascheduler/process-scheduler/src/main/java/org/apache/airavata/metascheduler/process/scheduling/engine/cr/selection/MultipleComputeResourcePolicy.java
b/modules/airavata-metascheduler/process-scheduler/src/main/java/org/apache/airavata/metascheduler/process/scheduling/engine/cr/selection/MultipleComputeResourcePolicy.java
index 1fdb9656d0..a0244aea9d 100644
---
a/modules/airavata-metascheduler/process-scheduler/src/main/java/org/apache/airavata/metascheduler/process/scheduling/engine/cr/selection/MultipleComputeResourcePolicy.java
+++
b/modules/airavata-metascheduler/process-scheduler/src/main/java/org/apache/airavata/metascheduler/process/scheduling/engine/cr/selection/MultipleComputeResourcePolicy.java
@@ -13,8 +13,10 @@ import org.slf4j.LoggerFactory;
import org.apache.airavata.registry.api.RegistryService;
import org.apache.airavata.registry.api.RegistryService.Client;
+import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
+import java.util.Random;
/**
* This class implements selecting one compute resource out of enabled
multiple compute resource polices.
@@ -47,29 +49,55 @@ public class MultipleComputeResourcePolicy extends
DefaultComputeResourceSelecti
ComputationalResourceSchedulingModel
computationalResourceSchedulingModel = userConfigurationDataModel
.getComputationalResourceScheduling();
- int crPoolFraction =
ServerSettings.getMetaschedulerMultipleCREnablingFactor();
- List<ComputeResourcePolicy> policyList = registryClient.
-
getGroupComputeResourcePolicyList(processModel.getGroupResourceProfileId());
+ List<ComputationalResourceSchedulingModel>
resourceSchedulingModels =
+
userConfigurationDataModel.getAutoScheduledCompResourceSchedulingList();
- int count = 0;
- int maxCount = (int) (policyList.size() * crPoolFraction);
+ List<String> retries = new ArrayList<>();
- while (count < maxCount) {
- ComputeResourcePolicy resourcePolicy =
policyList.get(count);
- List<String> queues =
resourcePolicy.getAllowedBatchQueues();
- String computeResourceId =
resourcePolicy.getComputeResourceId();
- ComputeResourceDescription comResourceDes =
registryClient.getComputeResource(computeResourceId);
- if (!queues.isEmpty()) {
- QueueStatusModel queueStatusModel =
registryClient.getQueueStatus(comResourceDes.getHostName(), queues.get(0));
+ while (retries.size()<resourceSchedulingModels.size()) {
+ Random rand = new Random();
+ int upperbound = resourceSchedulingModels.size();
+ int int_random = rand.nextInt(upperbound);
+ ComputationalResourceSchedulingModel
resourceSchedulingModel = resourceSchedulingModels.get(int_random);
+ String key =
resourceSchedulingModel.getResourceHostId()+"_"+resourceSchedulingModel.getQueueName();
+ if(!retries.contains(key)){
+ ComputeResourceDescription comResourceDes =
registryClient.getComputeResource(resourceSchedulingModel.getResourceHostId());
+ QueueStatusModel queueStatusModel =
registryClient.getQueueStatus(comResourceDes.getHostName(),
+ resourceSchedulingModel.getQueueName());
if (queueStatusModel.isQueueUp()) {
- return
Optional.of(computationalResourceSchedulingModel);
+ return Optional.of(resourceSchedulingModel);
+ }else{
+ retries.add(key);
}
}
- count++;
}
+
+// int crPoolFraction =
ServerSettings.getMetaschedulerMultipleCREnablingFactor();
+//
+// List<ComputeResourcePolicy> policyList = registryClient.
+//
getGroupComputeResourcePolicyList(processModel.getGroupResourceProfileId());
+//
+// int count = 0;
+// int maxCount = (int) (policyList.size() * crPoolFraction);
+//
+// while (count < maxCount) {
+// ComputeResourcePolicy resourcePolicy =
policyList.get(count);
+// List<String> queues =
resourcePolicy.getAllowedBatchQueues();
+//
+// String computeResourceId =
resourcePolicy.getComputeResourceId();
+// ComputeResourceDescription comResourceDes =
registryClient.getComputeResource(computeResourceId);
+//
+// if (!queues.isEmpty()) {
+// QueueStatusModel queueStatusModel =
registryClient.getQueueStatus(comResourceDes.getHostName(), queues.get(0));
+// if (queueStatusModel.isQueueUp()) {
+// return
Optional.of(computationalResourceSchedulingModel);
+// }
+// }
+// count++;
+// }
}
} catch (Exception exception) {
