Support Task Operations and shell commands for task logs

spring-cloud-dataflow-rest-client/src/main/java/org/springframework/cloud/dataflow/rest/client/TaskOperations.java

@@ -105,6 +105,23 @@ public interface TaskOperations {
 	 */
 	TaskExecutionResource taskExecutionStatus(long id);
 
+	/**
+	 * Return the task execution log.  The platform to from which to retrieve the log will be set to {@code default}.
+	 *
+	 * @param externalExecutionId the external execution identifier of the task execution.
+	 * @return {@link String} containing the log.
+	 */
+	String taskExecutionLog(String externalExecutionId);
+
+	/**
+	 * Return the task execution log.
+	 *
+	 * @param externalExecutionId the external execution identifier of the task execution.
+	 * @param platform the platform from which to obtain the log.
+	 * @return {@link String} containing the log.
+	 */
+	String taskExecutionLog(String externalExecutionId, String platform);
+
 	/**
 	 * Return information including the count of currently executing tasks and task execution
 	 * limits.

spring-cloud-dataflow-rest-client/src/main/java/org/springframework/cloud/dataflow/rest/client/TaskTemplate.java

@@ -18,6 +18,7 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -71,6 +72,8 @@ public class TaskTemplate implements TaskOperations {
 
 	private static final String PLATFORM_LIST_RELATION = "tasks/platforms";
 
+	private static final String RETRIEVE_LOG = "tasks/logs";
+
 	private final RestTemplate restTemplate;
 
 	private final Link definitionsLink;
@@ -91,6 +94,8 @@ public class TaskTemplate implements TaskOperations {
 
 	private final String dataFlowServerVersion;
 
+	private final Link retrieveLogLink;
+
 	TaskTemplate(RestTemplate restTemplate, ResourceSupport resources, String dataFlowServerVersion) {
 		Assert.notNull(resources, "URI Resources must not be be null");
 		Assert.notNull(resources.getLink(EXECUTIONS_RELATION), "Executions relation is required");
@@ -101,6 +106,7 @@ public class TaskTemplate implements TaskOperations {
 		Assert.notNull(resources.getLink(EXECUTION_RELATION), "Execution relation is required");
 		Assert.notNull(resources.getLink(EXECUTION_RELATION_BY_NAME), "Execution by name relation is required");
 		Assert.notNull(dataFlowServerVersion, "dataFlowVersion must not be null");
+		Assert.notNull(resources.getLink(RETRIEVE_LOG), "Log relation is required");
 
 		this.dataFlowServerVersion = dataFlowServerVersion;
 
@@ -125,6 +131,7 @@ public class TaskTemplate implements TaskOperations {
 		this.executionsCurrentLink = resources.getLink(EXECUTIONS_CURRENT_RELATION);
 		this.validationLink = resources.getLink(VALIDATION_REL);
 		this.platformListLink = resources.getLink(PLATFORM_LIST_RELATION);
+		this.retrieveLogLink = resources.getLink(RETRIEVE_LOG);
 	}
 
 	@Override
@@ -194,6 +201,19 @@ public TaskExecutionResource taskExecutionStatus(long id) {
 		return restTemplate.getForObject(executionLink.expand(id).getHref(), TaskExecutionResource.class);
 	}
 
+	@Override
+	public String taskExecutionLog(String externalExecutionId) {
+		return taskExecutionLog(externalExecutionId, "default");
+	}
+
+	@Override
+	public String taskExecutionLog(String externalExecutionId, String platform) {
+		Map<String,String> map = new HashMap<>();
+		map.put("taskExternalExecutionId",externalExecutionId);
+		map.put("platformName", platform);
+		return restTemplate.getForObject(retrieveLogLink.expand(map).getHref(), String.class);
+	}
+
 	@Override
 	public Collection<CurrentTaskExecutionsResource> currentTaskExecutions() {
 		ParameterizedTypeReference<Collection<CurrentTaskExecutionsResource>> typeReference =

spring-cloud-dataflow-shell-core/src/main/java/org/springframework/cloud/dataflow/shell/command/TaskCommands.java

@@ -75,6 +75,7 @@ public class TaskCommands implements CommandMarker {
 
 	private static final String LAUNCH = "task launch";
 	private static final String STOP = "task execution stop";
+	private static final String LOG = "task execution log";
 
 	// Destroy Role
 
@@ -229,6 +230,20 @@ public String stop(@CliOption(key = { "", "ids" }, help = "the task execution id
 		return String.format("Request to stop the task execution with id(s): %s has been submitted", ids);
 	}
 
+	@CliCommand(value = LOG, help = "Retrieve task execution log")
+	public String retrieveTaskExecutionLog(@CliOption(key = { "", "id" }, help = "the task execution id", mandatory = true) long id,
+			@CliOption(key = { "", "platform" }, help = "the platform of the task execution", mandatory = false) String platform) {
+		TaskExecutionResource taskExecutionResource = taskOperations().taskExecutionStatus(id);
+		String result;
+		if(platform != null) {
+			result = taskOperations().taskExecutionLog(taskExecutionResource.getExternalExecutionId(), platform);
+		}
+		else {
+			result = taskOperations().taskExecutionLog(taskExecutionResource.getExternalExecutionId());
+		}
+		return result;
+	}
+
 	@CliCommand(value = DESTROY, help = "Destroy an existing task")
 	public String destroy(
 			@CliOption(key = { "", "name" }, help = "the name of the task to destroy", mandatory = true,

spring-cloud-dataflow-shell-core/src/test/java/org/springframework/cloud/dataflow/shell/command/TaskCommandTemplate.java

@@ -39,6 +39,10 @@
  */
 public class TaskCommandTemplate {
 
+	private final static int WAIT_INTERVAL = 500;
+
+	private final static int MAX_WAIT_TIME = 3000;
+
 	private final JLineShellComponent shell;
 
 	private List<String> tasks = new ArrayList<String>();
@@ -103,6 +107,65 @@ public long launchWithAlternateCTR(String taskName, String ctrAppName) {
 		return value;
 	}
 
+	/**
+	 * Launch a task and validate the result from shell on default platform.
+	 *
+	 * @param taskName the name of the task
+	 */
+	public String getTaskExecutionLog(String taskName) throws Exception{
+		long id = launchTaskExecutionForLog(taskName);
+		CommandResult cr = shell.executeCommand("task execution log --id " + id);
+		assertTrue(cr.toString().contains("Starting TimestampTaskApplication"));
+
+		return cr.toString();
+	}
+
+	/**
+	 * Launch a task with invalid platform.
+	 *
+	 * @param taskName the name of the task
+	 */
+	public void getTaskExecutionLogInvalidPlatform(String taskName) throws Exception{
+		long id = launchTaskExecutionForLog(taskName);
+		shell.executeCommand(String.format("task execution log --id %s --platform %s", id, "foo"));
+	}
+
+	/**
+	 * Launch a task with invalid task execution id
+
+	 */
+	public void getTaskExecutionLogInvalidId() throws Exception{
+		CommandResult cr = shell.executeCommand(String.format("task execution log --id %s", 88));
+	}
+
+	private long launchTaskExecutionForLog(String taskName) throws Exception{
+		// add the task name to the tasks list before assertion
+		tasks.add(taskName);
+		CommandResult cr = shell.executeCommand(String.format("task launch %s", taskName));
+		CommandResult idResult = shell.executeCommand("task execution list --name " + taskName);
+		Table taskExecutionResult = (Table) idResult.getResult();
+
+		long id = (long) taskExecutionResult.getModel().getValue(1, 1);
+		assertTrue(cr.toString().contains("with execution id " + id));
+		waitForDBToBePopulated(id);
+		return id;
+	}
+
+	private void waitForDBToBePopulated(long id) throws Exception {
+		for (int waitTime = 0; waitTime <= MAX_WAIT_TIME; waitTime += WAIT_INTERVAL) {
+			Thread.sleep(WAIT_INTERVAL);
+			if (isEndTime(id)) {
+				break;
+			}
+		}
+	}
+
+	private boolean isEndTime(long id) {
+		CommandResult cr = taskExecutionStatus(id);
+		Table table = (Table) cr.getResult();
+		return (table.getModel().getValue(6, 1) != null);
+
+	}
 	/**
 	 * Stop a task execution.
 	 *

spring-cloud-dataflow-shell-core/src/test/java/org/springframework/cloud/dataflow/shell/command/TaskCommandTests.java

@@ -128,6 +128,27 @@ public void testTaskLaunchCTRUsingAltCtrName() {
 		task().launchWithAlternateCTR(taskName, "timestamp");
 	}
 
+	@Test
+	public void testGetLog() throws Exception{
+		logger.info("Retrieving task execution log");
+		String taskName = generateUniqueStreamOrTaskName();
+		task().create(taskName, "timestamp");
+		task().getTaskExecutionLog(taskName);
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testGetLogInvalidPlatform() throws Exception{
+		logger.info("Retrieving task execution log");
+		String taskName = generateUniqueStreamOrTaskName();
+		task().create(taskName, "timestamp");
+		task().getTaskExecutionLogInvalidPlatform(taskName);
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testGetLogInvalidId() throws Exception{
+		task().getTaskExecutionLogInvalidId();
+	}
+
 	@Test
 	public void testTaskLaunchCTRUsingInvalidAltCtrAppName() {
 		testInvalidCTRLaunch("1: timestamp && 2: timestamp", "timesdaftamp",
@@ -245,8 +266,7 @@ public void testViewExecution() {
 
 		CommandResult idResult = task().taskExecutionList();
 		Table result = (Table) idResult.getResult();
-
-		long value = (long) result.getModel().getValue(1, 1);
+		long value = (long) result.getModel().getValue(findRowForExecutionId(result, TASK_EXECUTION_ID), 1);
 		logger.info("Looking up id " + value);
 		CommandResult cr = task().taskExecutionStatus(value);
 		assertTrue("task execution status command must be successful", cr.isSuccess());
@@ -339,4 +359,17 @@ private void verifyTableValue(Table table, int row, int col, Object expected) {
 		assertEquals(String.format("Row %d, Column %d should be: %s", row, col, expected),expected,
 			table.getModel().getValue(row, col));
 	}
+
+	private int findRowForExecutionId(Table table, long id) {
+		int result = -1;
+		for(int rowNum = 0; rowNum < table.getModel().getRowCount(); rowNum++) {
+			if(table.getModel().getValue(rowNum, 1).equals(id)) {
+				result = rowNum;
+				break;
+			}
+		}
+		assertTrue("Task Execution Id specified was not found in execution list", id > -1);
+		return result;
+	}
+
 }