Feature/k8s executor

pom.xml

@@ -153,6 +153,12 @@
             <artifactId>commons-cli</artifactId>
             <version>1.4</version>
         </dependency>
+
+        <dependency>
+	    	<groupId>io.kubernetes</groupId>
+	    	<artifactId>client-java</artifactId>
+	    	<version>8.0.2</version>
+	  	</dependency>
 
     </dependencies>
 

src/main/java/org/fogbowcloud/arrebol/ArrebolController.java

@@ -13,6 +13,7 @@
 import org.fogbowcloud.arrebol.execution.Worker;
 import org.fogbowcloud.arrebol.execution.WorkerTypes;
 import org.fogbowcloud.arrebol.execution.creator.DockerWorkerCreator;
+import org.fogbowcloud.arrebol.execution.creator.K8sWorkerCreator;
 import org.fogbowcloud.arrebol.execution.creator.RawWorkerCreator;
 import org.fogbowcloud.arrebol.execution.creator.WorkerCreator;
 import org.fogbowcloud.arrebol.models.command.CommandState;
@@ -32,6 +33,7 @@
 import org.fogbowcloud.arrebol.scheduler.DefaultScheduler;
 import org.fogbowcloud.arrebol.scheduler.FifoSchedulerPolicy;
 import org.fogbowcloud.arrebol.utils.ConfValidator;
+import org.hibernate.boot.registry.classloading.spi.ClassLoaderService.Work;
 import org.springframework.stereotype.Component;
 
 import java.io.BufferedReader;
@@ -188,7 +190,9 @@ private void buildWorkerCreator(Configuration configuration) throws Exception {
             this.workerCreator = new DockerWorkerCreator(configuration);
         } else if (poolType.equals(WorkerTypes.RAW.getType())) {
             this.workerCreator = new RawWorkerCreator(configuration);
-        } else {
+        } else if (poolType.equals(WorkerTypes.K8S.getType())) {
+        	this.workerCreator = new K8sWorkerCreator(configuration);
+    	}else {
             String poolTypeMsg = "Worker Pool Type configuration property wrong or missing. Please, verify your configuration file.";
             throw new IllegalArgumentException(poolTypeMsg);
         }

src/main/java/org/fogbowcloud/arrebol/execution/TaskExecutionResult.java

@@ -7,6 +7,7 @@ public class TaskExecutionResult {
     // this class is more like an struct, not very much OO but it ok
 
     public static final int UNDETERMINED_RESULT = Integer.MAX_VALUE;
+    public static final int SUCCESS_RESULT = 0;
     private final RESULT taskResult;
     private final int[] exitcodes;
 

src/main/java/org/fogbowcloud/arrebol/execution/WorkerTypes.java

@@ -2,7 +2,8 @@
 
 public enum WorkerTypes {
     DOCKER("docker"),
-    RAW("raw");
+    RAW("raw"),
+    K8S("k8s");
 
     private String type;
 

src/main/java/org/fogbowcloud/arrebol/execution/creator/K8sWorkerCreator.java

@@ -0,0 +1,79 @@
+package org.fogbowcloud.arrebol.execution.creator;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.UUID;
+
+import org.apache.log4j.Logger;
+import org.fogbowcloud.arrebol.execution.Worker;
+import org.fogbowcloud.arrebol.execution.k8s.K8sConfiguration;
+import org.fogbowcloud.arrebol.execution.k8s.K8sTaskExecutor;
+import org.fogbowcloud.arrebol.execution.k8s.client.DefaultK8sClient;
+import org.fogbowcloud.arrebol.execution.k8s.client.K8sClient;
+import org.fogbowcloud.arrebol.execution.k8s.resource.DefaultK8sClusterResource;
+import org.fogbowcloud.arrebol.execution.k8s.resource.K8sClusterResource;
+import org.fogbowcloud.arrebol.models.configuration.Configuration;
+import org.fogbowcloud.arrebol.models.specification.Specification;
+import org.fogbowcloud.arrebol.processor.spec.WorkerNode;
+import org.fogbowcloud.arrebol.resource.MatchAnyWorker;
+import org.fogbowcloud.arrebol.utils.AppUtil;
+
+public class K8sWorkerCreator implements WorkerCreator {
+
+	private final K8sConfiguration configuration;
+	private final Logger LOGGER = Logger.getLogger(K8sWorkerCreator.class);
+
+	public K8sWorkerCreator(Configuration configuration) throws Exception {
+		this.configuration = new K8sConfiguration(configuration);
+	}
+
+	@Override
+	public Collection<Worker> createWorkers(Integer poolId) {
+		Collection<Worker> workers = new LinkedList<>();
+		String address = configuration.getAddress();
+		int poolSize = configuration.getCapacity();
+		String namespace = configuration.getNamespace();
+		String volumeName = configuration.getVolumeName();
+		for (int i = 0; i < poolSize; i++) {
+			LOGGER.info("Creating k8s worker with address=" + address);
+			Worker worker = createK8sWorker(poolId, address, namespace, volumeName);
+			workers.add(worker);
+		}
+
+		return workers;
+	}
+
+	@Override
+	public Collection<Worker> createWorkers(Integer poolId, WorkerNode workerNode) {
+		Collection<Worker> workers = new LinkedList<>();
+		String namespace = configuration.getNamespace();
+		String volumeName = configuration.getVolumeName();
+        for(int i = 0; i < workerNode.getPoolSize(); i++){
+            LOGGER.info("Creating k8s worker with address=" + workerNode.getAddress());
+            Worker worker = createK8sWorker(poolId, workerNode.getAddress(), namespace, volumeName);
+            workers.add(worker);
+        }
+        return workers;
+	}
+
+	private Worker createK8sWorker(Integer poolId, String address, String namespace, String volumeName) {
+		String id = "k8s-executor-" + UUID.randomUUID().toString();
+		K8sClusterResource k8sClusterResource = new DefaultK8sClusterResource(id, address);
+		//TODO Add some specification here, something about the CPU and RAM requirements maybe
+		Specification resourceSpec = null;
+
+		K8sClient k8sClient = null;
+		try {
+			k8sClient = new DefaultK8sClient(address, namespace, volumeName);
+		} catch (IOException e) {
+			LOGGER.error("Error while create k8s client in worker [" + id + "]", e);
+		}
+
+		K8sTaskExecutor k8sTaskExecutor = new K8sTaskExecutor(k8sClusterResource, k8sClient);
+		Worker worker = new MatchAnyWorker(AppUtil.generateUniqueStringId(), resourceSpec, poolId, k8sTaskExecutor);
+		LOGGER.info("Created K8s Worker [" + worker.getId() + "]");
+		return worker;
+	}
+
+}

src/main/java/org/fogbowcloud/arrebol/execution/k8s/K8sConfiguration.java

@@ -0,0 +1,60 @@
+package org.fogbowcloud.arrebol.execution.k8s;
+
+import org.fogbowcloud.arrebol.execution.k8s.constants.K8sConstants;
+import org.fogbowcloud.arrebol.models.configuration.Configuration;
+import org.fogbowcloud.arrebol.models.configuration.Property;
+
+public class K8sConfiguration {
+	private final Integer capacity;
+	private final String address;
+	private final String namespace;
+	private final String volumeName;
+
+	public K8sConfiguration(Configuration configuration) throws Exception {
+		checkK8sConfigurationProperties(configuration);
+		Property<Double> capacity = configuration.getProperty(K8sConstants.K8S_CLUSTER_CAPACITY);
+		Property<String> address = configuration.getProperty(K8sConstants.K8S_CLUSTER_ADDRESS);
+		Property<String> namespace = configuration.getProperty(K8sConstants.K8S_CLUSTER_NAMESPACE);
+		Property<String> volumeName = configuration.getProperty(K8sConstants.K8S_CLUSTER_VOLUME_NAME);
+		this.capacity = capacity.getValue().intValue();
+		this.address = address.getValue();
+		this.namespace = namespace.getValue();
+		this.volumeName = volumeName.getValue();
+	}
+
+	private void checkK8sConfigurationProperties(Configuration configuration) throws Exception {
+		String verifyMsg = " Please, verify your configuration file.";
+		String capacityMsg = "K8s cluster capacity configuration property wrong or missing." + verifyMsg;
+		String addressMsg = "K8s cluster address configuration property wrong or missing." + verifyMsg;
+		String namespaceMsg = "K8s cluster namespace configuration property wrong or missing." + verifyMsg;
+
+		Property<Double> capacity = configuration.getProperty(K8sConstants.K8S_CLUSTER_CAPACITY);
+		Property<String> address = configuration.getProperty(K8sConstants.K8S_CLUSTER_ADDRESS);
+		Property<String> namespace = configuration.getProperty(K8sConstants.K8S_CLUSTER_NAMESPACE);
+
+		if (address.getValue() == null || address.getValue().trim().isEmpty()) {
+			throw new Exception(addressMsg);
+		} else if (capacity.getValue() == null || capacity.getValue().intValue() == 0) {
+			throw new Exception(capacityMsg);
+		} else if (namespace.getValue() == null || namespace.getValue().trim().isEmpty()) {
+			throw new Exception(namespaceMsg);
+		}
+	}
+
+	public Integer getCapacity() {
+		return capacity;
+	}
+
+	public String getAddress() {
+		return address;
+	}
+
+	public String getNamespace() {
+		return namespace;
+	}
+
+	public String getVolumeName() {
+		return volumeName;
+	}
+
+}

src/main/java/org/fogbowcloud/arrebol/execution/k8s/K8sTaskExecutor.java

@@ -0,0 +1,166 @@
+package org.fogbowcloud.arrebol.execution.k8s;
+
+import static java.lang.Thread.sleep;
+import static org.fogbowcloud.arrebol.execution.docker.constants.DockerConstants.ADDRESS_METADATA_KEY;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import javax.persistence.CascadeType;
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
+import javax.persistence.Id;
+import javax.persistence.OneToOne;
+import javax.persistence.Transient;
+
+import org.apache.log4j.Logger;
+import org.fogbowcloud.arrebol.execution.TaskExecutionResult;
+import org.fogbowcloud.arrebol.execution.TaskExecutor;
+import org.fogbowcloud.arrebol.execution.TaskExecutionResult.RESULT;
+import org.fogbowcloud.arrebol.execution.k8s.client.DefaultK8sClient;
+import org.fogbowcloud.arrebol.execution.k8s.client.K8sClient;
+import org.fogbowcloud.arrebol.execution.k8s.resource.DefaultK8sClusterResource;
+import org.fogbowcloud.arrebol.execution.k8s.resource.K8sClusterResource;
+import org.fogbowcloud.arrebol.models.command.Command;
+import org.fogbowcloud.arrebol.models.command.CommandState;
+import org.fogbowcloud.arrebol.models.task.RequirementsContants;
+import org.fogbowcloud.arrebol.models.task.Task;
+
+import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.models.V1Job;
+
+@Entity
+public class K8sTaskExecutor implements TaskExecutor {
+
+	@Id
+	@GeneratedValue(strategy = GenerationType.IDENTITY)
+	private Integer id;
+	@Transient
+	private final Logger LOGGER = Logger.getLogger(K8sTaskExecutor.class);
+	@OneToOne(cascade = CascadeType.ALL, orphanRemoval = true, targetEntity = DefaultK8sClusterResource.class)
+	private K8sClusterResource k8sClusterResource;
+	@OneToOne(cascade = CascadeType.ALL, orphanRemoval = true, targetEntity = DefaultK8sClient.class)
+	private K8sClient k8sClient;
+	@Transient
+	private static final long POOLING_PERIOD_TIME_MS = 5 * 1000;
+
+	public K8sTaskExecutor(K8sClusterResource k8sClusterResource, K8sClient k8sClient) {
+		this.k8sClusterResource = k8sClusterResource;
+		this.k8sClient = k8sClient;
+	}
+
+	public K8sTaskExecutor() {
+	}
+
+	@Override
+	public TaskExecutionResult execute(Task task) {
+		TaskExecutionResult taskExecutionResult = null;
+
+		String jobName = task.getId();
+
+		List<Command> commands = task.getTaskSpec().getCommands();
+		String command = joinCommands(commands);
+
+		Map<String, String> requirements = task.getTaskSpec().getRequirements();
+		String imageId = getImageId(requirements);
+
+		LOGGER.debug("Image id: " + imageId);
+		LOGGER.debug("Command: " + command);
+
+		int tasksListSize = task.getTaskSpec().getCommands().size();
+
+		try {
+			V1Job job = k8sClient.createJob(jobName, imageId, command);
+			boolean jobIsRunning = true;
+
+			while (jobIsRunning) {
+				try {
+					sleep(POOLING_PERIOD_TIME_MS);
+				} catch (InterruptedException e) {
+					LOGGER.error(e.getMessage(), e);
+				}
+
+				job = k8sClient.getJob(jobName);
+				jobIsRunning = job.getStatus().getActive() != null;
+			}
+
+			if (wasSuccessful(job)) {
+				finishSuccessfulExecution(task);
+				taskExecutionResult = getSuccessResultInstance(tasksListSize);
+			} else {
+				finishFailExecution(task);
+				taskExecutionResult = getFailResultInstance(tasksListSize);
+			}
+		} catch (ApiException e) {
+			LOGGER.error("Error while call K8s client API. " + e.getMessage(), e);
+			finishFailExecution(task);
+			taskExecutionResult = getFailResultInstance(tasksListSize);
+			return taskExecutionResult;
+		}
+
+		return taskExecutionResult;
+	}
+
+	@Override
+	public Map<String, String> getMetadata() {
+		Map<String, String> metadata = new HashMap<>();
+		String address = this.k8sClusterResource.getApiAddress();
+		metadata.put(ADDRESS_METADATA_KEY, address);
+		return metadata;
+	}
+
+	private String joinCommands(List<Command> commands) {
+		String result = "";
+		for (Command command : commands)
+			result += command.getCommand() + " && ";
+		result = result.substring(0, result.length() - 4);
+
+		return result;
+	}
+
+	private String getImageId(Map<String, String> requirements) throws IllegalArgumentException {
+		String imageId = null;
+		if (Objects.nonNull(requirements))
+			imageId = requirements.get(RequirementsContants.IMAGE_KEY);
+
+		return imageId;
+	}
+
+	private boolean wasSuccessful(V1Job job) {
+		Integer succeededAmount = job.getStatus().getSucceeded();
+		return succeededAmount != null && succeededAmount > 0;
+	}
+
+	private void finishSuccessfulExecution(Task task) {
+		for (Command c : task.getTaskSpec().getCommands()) {
+			c.setState(CommandState.FINISHED);
+			c.setExitcode(TaskExecutionResult.SUCCESS_RESULT);
+		}
+	}
+
+	private void finishFailExecution(Task task) {
+		for (Command c : task.getTaskSpec().getCommands()) {
+			c.setState(CommandState.FAILED);
+			c.setExitcode(TaskExecutionResult.UNDETERMINED_RESULT);
+		}
+	}
+
+	private TaskExecutionResult getFailResultInstance(int size) {
+		int[] exitCodes = new int[size];
+		Arrays.fill(exitCodes, TaskExecutionResult.UNDETERMINED_RESULT);
+		TaskExecutionResult taskExecutionResult = new TaskExecutionResult(RESULT.FAILURE, exitCodes);
+		return taskExecutionResult;
+	}
+
+	private TaskExecutionResult getSuccessResultInstance(int size) {
+		int[] exitCodes = new int[size];
+		Arrays.fill(exitCodes, TaskExecutionResult.SUCCESS_RESULT);
+		TaskExecutionResult taskExecutionResult = new TaskExecutionResult(RESULT.SUCCESS, exitCodes);
+		return taskExecutionResult;
+	}
+
+}