Merge pull request #40 from indigo-dc/feat/chronos

[WIP] Implements [#3149] Job Submission from Tosca to Chronos/Mesos #34

pom.xml

@@ -26,6 +26,7 @@
 		<checkstyle-plugin.version>2.17</checkstyle-plugin.version>
 		<workflow-manager.version>0.0.3-SNAPSHOT</workflow-manager.version>
 		<alien4cloud.version>1.1.0-INDIGO2-SNAPSHOT</alien4cloud.version>
+		<chronos-client.version>0.0.1-SNAPSHOT</chronos-client.version>
 	</properties>
 
 	<distributionManagement>
@@ -234,6 +235,20 @@
 			<version>1.3.1</version>
 		</dependency>
 
+		<!-- Chronos Client -->
+		<dependency>
+			<groupId>it.infn.ba.indigo</groupId>
+			<artifactId>chronos-client</artifactId>
+			<version>${chronos-client.version}</version>
+		</dependency>
+
+		<!-- Graph library -->
+		<dependency>
+			<groupId>org.jgrapht</groupId>
+			<artifactId>jgrapht-core</artifactId>
+			<version>0.9.2</version>
+		</dependency>
+
 		<!-- test -->
 		<dependency>
 			<groupId>org.springframework</groupId>

src/main/java/it/reply/orchestrator/dal/entity/Deployment.java

@@ -1,9 +1,14 @@
 package it.reply.orchestrator.dal.entity;
 
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+
 import it.reply.orchestrator.enums.DeploymentProvider;
 import it.reply.orchestrator.enums.Status;
 import it.reply.orchestrator.enums.Task;
+import it.reply.utils.json.JsonUtility;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -50,10 +55,16 @@ public class Deployment extends AbstractResourceEntity {
   @Column(name = "template", columnDefinition = "LONGTEXT")
   private String template;
 
+  /**
+   * The user's inputs to the template.
+   */
+  @Transient
+  Map<String, Object> unserializedParameters = null;
+
   @ElementCollection(fetch = FetchType.EAGER)
   @MapKeyColumn(name = "name")
   @Column(name = "value")
-  Map<String, String> parameters = new HashMap<String, String>();
+  private Map<String, String> parameters = new HashMap<>();
 
   @ElementCollection(fetch = FetchType.EAGER)
   @MapKeyColumn(name = "name")
@@ -126,12 +137,58 @@ public void setTemplate(String template) {
     this.template = template;
   }
 
-  public Map<String, String> getParameters() {
-    return parameters;
-  }
-
-  public void setParameters(Map<String, String> parameters) {
-    this.parameters = parameters;
+  /**
+   * The user's inputs to the template.
+   *
+   * @throws IOException
+   * @throws JsonMappingException
+   * @throws JsonParseException
+   */
+  public synchronized Map<String, Object> getParameters() {
+
+    if (unserializedParameters != null) {
+      return unserializedParameters;
+    }
+
+    unserializedParameters = new HashMap<>();
+    for (Map.Entry<String, String> serializedParam : parameters.entrySet()) {
+      Object paramObject = null;
+      if (serializedParam.getValue() != null) {
+        try {
+          paramObject = JsonUtility.deserializeJson(serializedParam.getValue(), Object.class);
+        } catch (IOException ex) {
+          throw new RuntimeException("Failed to deserialize parameters in JSON", ex);
+        }
+      }
+
+      unserializedParameters.put(serializedParam.getKey(), paramObject);
+    }
+
+    return unserializedParameters;
+  }
+
+  /**
+   * The user's inputs to the template.
+   *
+   * @throws IOException
+   * @throws JsonMappingException
+   * @throws JsonParseException
+   */
+  public synchronized void setParameters(Map<String, Object> parameters) {
+    this.parameters = new HashMap<>();
+    for (Map.Entry<String, Object> parameter : parameters.entrySet()) {
+      String paramString = null;
+      if (parameter.getValue() != null) {
+        try {
+          paramString = JsonUtility.serializeJson(parameter.getValue());
+        } catch (IOException ex) {
+          throw new RuntimeException("Failed to serialize parameters in JSON", ex);
+        }
+      }
+
+      this.parameters.put(parameter.getKey(), paramString);
+    }
+    this.unserializedParameters = null;
   }
 
   public Map<String, String> getOutputs() {

src/main/java/it/reply/orchestrator/dal/repository/ResourceRepository.java

@@ -13,4 +13,6 @@ public interface ResourceRepository extends PagingAndSortingRepository<Resource,
   public Page<Resource> findByDeployment_id(String deploymentId, Pageable pageable);
 
   public Resource findByIdAndDeployment_id(String uuid, String deploymentId);
+
+  public Resource findByToscaNodeNameAndDeployment_id(String toscaNodeName, String deploymentId);
 }

src/main/java/it/reply/orchestrator/exception/http/BadRequestException.java

@@ -1,7 +1,8 @@
 package it.reply.orchestrator.exception.http;
 
 /**
- * Exception thrown when a resource doesn't exist.
+ * Exception thrown when the client request is bad (i.e. wrong information, wrong resource status,
+ * etc)
  * 
  * @author m.bassi
  *
@@ -17,5 +18,4 @@ public BadRequestException(String message) {
   public BadRequestException(String message, Throwable ex) {
     super(message, ex);
   }
-
-}
+}

src/main/java/it/reply/orchestrator/service/DeploymentService.java

@@ -8,13 +8,18 @@
 
 public interface DeploymentService {
 
-  public Page<Deployment> getDeployments(Pageable pageable);
+	public static final String WF_PARAM_DEPLOYMENT_ID = "DEPLOYMENT_ID";
+	public static final String WF_PARAM_DEPLOYMENT_TYPE = "DEPLOYMENT_TYPE";
+	public static final String DEPLOYMENT_TYPE_CHRONOS = "CHRONOS";
+	public static final String DEPLOYMENT_TYPE_TOSCA = "TOSCA";
 
-  public Deployment getDeployment(String id);
+	public Page<Deployment> getDeployments(Pageable pageable);
 
-  public Deployment createDeployment(DeploymentRequest request);
+	public Deployment getDeployment(String id);
 
-  public void updateDeployment(String id, DeploymentRequest request);
+	public Deployment createDeployment(DeploymentRequest request);
 
-  public void deleteDeployment(String id);
+	public void updateDeployment(String id, DeploymentRequest request);
+
+	public void deleteDeployment(String id);
 }

src/main/java/it/reply/orchestrator/service/DeploymentServiceImpl.java

@@ -5,7 +5,6 @@
 import alien4cloud.model.topology.NodeTemplate;
 import alien4cloud.tosca.model.ArchiveRoot;
 import alien4cloud.tosca.parser.ParsingException;
-import alien4cloud.tosca.parser.ParsingResult;
 
 import it.reply.orchestrator.config.WorkflowConfigProducerBean;
 import it.reply.orchestrator.dal.entity.Deployment;
@@ -14,14 +13,14 @@
 import it.reply.orchestrator.dal.repository.DeploymentRepository;
 import it.reply.orchestrator.dal.repository.ResourceRepository;
 import it.reply.orchestrator.dto.request.DeploymentRequest;
+import it.reply.orchestrator.enums.DeploymentProvider;
 import it.reply.orchestrator.enums.NodeStates;
 import it.reply.orchestrator.enums.Status;
 import it.reply.orchestrator.enums.Task;
 import it.reply.orchestrator.exception.OrchestratorException;
 import it.reply.orchestrator.exception.http.BadRequestException;
 import it.reply.orchestrator.exception.http.ConflictException;
 import it.reply.orchestrator.exception.http.NotFoundException;
-import it.reply.orchestrator.exception.http.OrchestratorApiException;
 import it.reply.orchestrator.exception.service.ToscaException;
 import it.reply.workflowmanager.exceptions.WorkflowException;
 import it.reply.workflowmanager.orchestrator.bpm.BusinessProcessManager;
@@ -38,7 +37,6 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 @Service
 public class DeploymentServiceImpl implements DeploymentService {
@@ -76,25 +74,44 @@ public Deployment getDeployment(String uuid) {
   public Deployment createDeployment(DeploymentRequest request) {
     Map<String, NodeTemplate> nodes;
     Deployment deployment;
+    boolean isChronosDeployment = false;
 
     try {
-      // Read the incoming template
-      nodes = toscaService.getArchiveRootFromTemplate(request.getTemplate()).getResult()
-          .getTopology().getNodeTemplates();
+      // Parse once, validate structure and user's inputs, replace user's input
+      ArchiveRoot parsingResult =
+          toscaService.prepareTemplate(request.getTemplate(), request.getParameters());
+
+      nodes = parsingResult.getTopology().getNodeTemplates();
+
       deployment = new Deployment();
       deployment.setStatus(Status.CREATE_IN_PROGRESS);
       deployment.setTask(Task.NONE);
-      deployment.setParameters(request.getParameters().entrySet().stream()
-          .collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue().toString())));
+      deployment.setParameters(request.getParameters());
 
       if (request.getCallback() != null) {
         deployment.setCallback(request.getCallback());
       }
-      createResources(deployment, nodes);
 
       deployment = deploymentRepository.save(deployment);
-      deployment
-          .setTemplate(toscaService.customizeTemplate(request.getTemplate(), deployment.getId()));
+
+      // FIXME: Define function to decide DeploymentProvider (Temporary - just for prototyping)
+      isChronosDeployment = isChronosDeployment(nodes);
+      if (!isChronosDeployment) {
+        // FIXME (BAD HACK) IM templates need some parameters to be added, but regenerating the
+        // template string with the current library is risky (loses some information!!)
+        // Re-parse and customize
+        String template = toscaService.customizeTemplate(request.getTemplate(), deployment.getId());
+        deployment.setTemplate(template);
+
+        // Re-parse with the updated nodes
+        parsingResult = toscaService.prepareTemplate(template, deployment.getParameters());
+        nodes = parsingResult.getTopology().getNodeTemplates();
+      } else {
+        deployment.setTemplate(request.getTemplate());
+      }
+
+      // Create internal resources representation (to store in DB)
+      createResources(deployment, nodes);
 
     } catch (IOException ex) {
       throw new OrchestratorException(ex.getMessage(), ex);
@@ -106,6 +123,11 @@ public Deployment createDeployment(DeploymentRequest request) {
 
     Map<String, Object> params = new HashMap<>();
     params.put("DEPLOYMENT_ID", deployment.getId());
+
+    // FIXME Put in deployment provider field
+    params.put(WF_PARAM_DEPLOYMENT_TYPE,
+        (isChronosDeployment ? DEPLOYMENT_TYPE_CHRONOS : DEPLOYMENT_TYPE_TOSCA));
+
     ProcessInstance pi = null;
     try {
       pi = wfService.startProcess(WorkflowConfigProducerBean.DEPLOY.getProcessId(), params,
@@ -120,6 +142,25 @@ public Deployment createDeployment(DeploymentRequest request) {
 
   }
 
+  /**
+   * Temporary method to decide whether a given deployment has to be deployed using Chronos (<b>just
+   * for experiments</b>). <br/>
+   * Currently, if there is at least one node whose name contains 'Chronos', the deployment is done
+   * with Chronos.
+   * 
+   * @param nodes
+   *          the template nodes.
+   * @return <tt>true</tt> if Chronos, <tt>false</tt> otherwise.
+   */
+  private static boolean isChronosDeployment(Map<String, NodeTemplate> nodes) {
+    for (Map.Entry<String, NodeTemplate> node : nodes.entrySet()) {
+      if (node.getValue().getType().contains("Chronos")) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   @Override
   @Transactional
   public void deleteDeployment(String uuid) {
@@ -130,19 +171,26 @@ public void deleteDeployment(String uuid) {
         throw new ConflictException(
             String.format("Deployment already in %s state.", deployment.getStatus().toString()));
       } else {
+        // Update deployment status
         deployment.setStatus(Status.DELETE_IN_PROGRESS);
         deployment.setStatusReason("");
         deployment.setTask(Task.NONE);
+        deployment = deploymentRepository.save(deployment);
+
+        // Abort all WF currently active on this deployment
         Iterator<WorkflowReference> wrIt = deployment.getWorkflowReferences().iterator();
         while (wrIt.hasNext()) {
           WorkflowReference wr = wrIt.next();
           wfService.abortProcess(wr.getProcessId(), wr.getRuntimeStrategy());
           wrIt.remove();
         }
-        deployment = deploymentRepository.save(deployment);
 
         Map<String, Object> params = new HashMap<>();
         params.put("DEPLOYMENT_ID", deployment.getId());
+
+        // FIXME: Temporary - just for test
+        params.put(WF_PARAM_DEPLOYMENT_TYPE, deployment.getDeploymentProvider().name());
+
         ProcessInstance pi = null;
         try {
           pi = wfService.startProcess(WorkflowConfigProducerBean.UNDEPLOY.getProcessId(), params,
@@ -164,22 +212,26 @@ public void deleteDeployment(String uuid) {
   public void updateDeployment(String id, DeploymentRequest request) {
     Deployment deployment = deploymentRepository.findOne(id);
     if (deployment != null) {
+
+      if (deployment.getDeploymentProvider() == DeploymentProvider.CHRONOS) {
+        // Chronos deployments cannot be updated
+        throw new BadRequestException("Chronos deployments cannot be updated.");
+      }
+
       if (deployment.getStatus() == Status.CREATE_COMPLETE
           || deployment.getStatus() == Status.UPDATE_COMPLETE
           || deployment.getStatus() == Status.UPDATE_FAILED) {
         try {
-          // Check if the new template is valid
-          ParsingResult<ArchiveRoot> parsingResult =
-              toscaService.getArchiveRootFromTemplate(request.getTemplate());
+          // Check if the new template is valid: parse, validate structure and user's inputs,
+          // replace user's inputs
+          toscaService.prepareTemplate(request.getTemplate(), deployment.getParameters());
 
         } catch (ParsingException | IOException ex) {
           throw new OrchestratorException(ex);
         }
         deployment.setStatus(Status.UPDATE_IN_PROGRESS);
         deployment.setTask(Task.NONE);
 
-        Iterator<WorkflowReference> wrIt = deployment.getWorkflowReferences().iterator();
-
         deployment = deploymentRepository.save(deployment);
 
         Map<String, Object> params = new HashMap<>();