Merge pull request #17 from mklkun/fix-lock

Fix ubuntu <20 versions apt lock issue in App deployments

build.gradle

@@ -45,6 +45,7 @@ dependencies {
     compile group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.12.1'
     compile group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-csv', version: '2.12.1'
     compile group: 'org.eclipse.emf', name: 'org.eclipse.emf.common', version: '2.20.0'
+    compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.11'
     annotationProcessor "org.projectlombok:lombok:1.18.12"
 }
 

pom.xml

@@ -95,6 +95,12 @@
       <version>2.12.1</version>
       <scope>compile</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+      <version>3.11</version>
+      <scope>compile</scope>
+    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

src/main/java/org/activeeon/morphemic/PAGateway.java

@@ -245,6 +245,12 @@ public class PAGateway {
                     commands.setPreStop(installation.optString("preStop"));
                     commands.setStop(installation.optString("stop"));
                     commands.setPostStop(installation.optString("postStop"));
+                    OperatingSystemType operatingSystemType = new OperatingSystemType();
+                    operatingSystemType.setOperatingSystemFamily(installation.optJSONObject("operatingSystem")
+                            .optString("operatingSystemFamily"));
+                    operatingSystemType.setOperatingSystemVersion(installation.optJSONObject("operatingSystem")
+                            .optFloat("operatingSystemVersion"));
+                    commands.setOperatingSystemType(operatingSystemType);
                     newTask.setInstallation(commands);
                     break;
                 case "spark":
@@ -633,7 +639,6 @@ public class PAGateway {
         } else {
             try {
                 nodeConfigJson += "\", \"portToOpens\": " + mapper.writeValueAsString(task.getPortsToOpen()) + "}";
-                ;
             } catch (IOException e) {
                 LOGGER.error(e.getStackTrace());
             }
@@ -774,6 +779,8 @@ public class PAGateway {
         if (task.getDeployments() == null || task.getDeployments().isEmpty()) {
             LOGGER.warn("The task " + task.getName() + " does not have a deployment. It will be scheduled on any free node.");
             scriptTasks.addAll(createAppTasks(task, "", ""));
+            task.setDeploymentFirstSubmittedTaskName(scriptTasks.get(0).getName());
+            task.setDeploymentLastSubmittedTaskName(scriptTasks.get(scriptTasks.size()-1).getName());
         }
         else {
             task.getDeployments().forEach(deployment -> {
@@ -787,15 +794,38 @@ public class PAGateway {
                 deployment.setIsDeployed(true);
             });
 
+            task.setDeploymentFirstSubmittedTaskName(scriptTasks.get(0).getName().substring(0, scriptTasks.get(0).getName().lastIndexOf("_0")));
+
             tasksTokens.forEach(taskToken -> {
                 String suffix = "_" + tasksTokens.indexOf(taskToken);
-                scriptTasks.addAll(createAppTasks(task, suffix, taskToken));
+                List<ScriptTask> appTasks = createAppTasks(task, suffix, taskToken);
+                task.setDeploymentLastSubmittedTaskName(appTasks.get(appTasks.size()-1).getName().substring(0, appTasks.get(appTasks.size()-1).getName().lastIndexOf(suffix)));
+                scriptTasks.addAll(0, appTasks);
+
+                if (task.getInstallation().getOperatingSystemType().getOperatingSystemFamily().toLowerCase(Locale.ROOT).equals("ubuntu") &&
+                        task.getInstallation().getOperatingSystemType().getOperatingSystemVersion() < 2000) {
+                    LOGGER.info("Adding apt lock handler task since task: " + task.getName() +
+                            " is meant to be executed in: " +
+                            task.getInstallation().getOperatingSystemType().getOperatingSystemFamily() +
+                            " version: " + task.getInstallation().getOperatingSystemType().getOperatingSystemVersion());
+                    scriptTasks.add(0, createLockHandlerTask(task.getName(), suffix, taskToken));
+                    scriptTasks.get(1).addDependence(scriptTasks.get(0));
+                }
             });
         }
 
+        scriptTasks.forEach(scriptTask -> task.addSubmittedTaskName(scriptTask.getName()));
+
         return scriptTasks;
     }
 
+    private ScriptTask createLockHandlerTask(String taskName, String suffix, String taskToken) {
+        String lockTaskName = "waitForLock_" + taskName + suffix;
+        ScriptTask lockHandlerTask = PAFactory.createBashScriptTaskFromFile(lockTaskName, "wait_for_lock_script.sh");
+        lockHandlerTask.addGenericInformation("NODE_ACCESS_TOKEN", taskToken);
+        return lockHandlerTask;
+    }
+
     private void setAllMandatoryDependencies(TaskFlowJob paJob, Job jobToSubmit) {
         jobToSubmit.getTasks().forEach(task -> {
             if (task.getParentTasks() != null && !task.getParentTasks().isEmpty()) {
@@ -803,8 +833,11 @@ public class PAGateway {
                     paJob.getTasks().forEach(paTask -> {
                         paJob.getTasks().forEach(paParentTask -> {
                             if (paTask.getName().contains(task.getName()) && paParentTask.getName().contains(parentTaskName)) {
+                                if (paTask.getName().contains(task.getDeploymentFirstSubmittedTaskName()) &&
+                                        paParentTask.getName().contains(jobToSubmit.findTask(parentTaskName).getDeploymentLastSubmittedTaskName())) {
                                     paTask.addDependence(paParentTask);
                                 }
+                            }
                         });
                     });
                 });
@@ -834,7 +867,6 @@ public class PAGateway {
             scriptTasks.forEach(scriptTask -> {
                 try {
                     paJob.addTask(scriptTask);
-                    task.addSubmittedTaskName(scriptTask.getName());
                 } catch (UserException e) {
                     LOGGER.error("Task " + task.getName() + " could not be added due to: " + e.toString());
                 }

src/main/java/org/activeeon/morphemic/model/CommandsInstallation.java

@@ -4,6 +4,7 @@ import lombok.*;
 
 import javax.persistence.Column;
 import javax.persistence.Embeddable;
+import javax.persistence.Embedded;
 
 @AllArgsConstructor
 @NoArgsConstructor
@@ -41,4 +42,7 @@ public class CommandsInstallation {
 
     @Column(name = "UPDATE_CMD", columnDefinition = "TEXT")
     private String updateCmd;
+
+    @Embedded
+    private OperatingSystemType operatingSystemType;
 }

src/main/java/org/activeeon/morphemic/model/OperatingSystemType.java

+package org.activeeon.morphemic.model;
+
+import lombok.*;
+
+import javax.persistence.Column;
+import javax.persistence.Embeddable;
+
+@AllArgsConstructor
+@NoArgsConstructor
+@ToString
+@Getter
+@Setter
+@Embeddable
+public class OperatingSystemType {
+    @Column(name = "OPERATING_SYSTEM_FAMILY")
+    private String operatingSystemFamily;
+
+    @Column(name = "OPERATING_SYSTEM_VERSION")
+    private float operatingSystemVersion;
+}

src/main/java/org/activeeon/morphemic/model/Task.java

@@ -53,6 +53,12 @@ public class Task implements Serializable {
     @ElementCollection(targetClass=String.class)
     private List<String> submittedTaskNames;
 
+    @Column(name = "DEPLOYMENT_FIRST_SUBMITTED_TASK_NAME")
+    private String deploymentFirstSubmittedTaskName;
+
+    @Column(name = "DEPLOYMENT_LAST_SUBMITTED_TASK_NAME")
+    private String deploymentLastSubmittedTaskName;
+
     public void addDeployment(Deployment deployment) {
         if (deployments==null){
             deployments = new LinkedList<>();

src/main/resources/wait_for_lock_script.sh

+i=0
+while [ `ps aux | grep [l]ock_is_held | wc -l` != 0 ]; do
+	echo "Lock_is_held $i"
+	ps aux | grep [l]ock_is_held
+	sleep 10
+	((i=i+10));
+done
+echo "Exited the while loop, time spent: $i"
+ps aux | grep [a]pt
\ No newline at end of file