Retrieve stream/task application logs

 - Add REST endpoints for stream apps log retrieval at Runtime apps controller
 - Add REST endpoint for task app log retrieval at TaskExecution controller
 - Add getLog() method SPI for TaskExecutionService and implemented it in DefaultTaskExecutionService
 - Add getLog() method in StreamDeployer and have it implemented in SkipperStreamDeployer
 - Add and update tests
 - Update CF task platform configuration to update doppler client to CF operations
 - Update Skipper client changes
 - Fix REST endpoint path for runtime apps log retrieval
 - Add separate controllers for stream/task logs retrieval
 - Change the return type of SCDF controllers that get logs
 - Fix API documentation and RootController
 - Remove runtime apps controller entries on log retrieval

Resolves #3306
Resolves #3307

Author: ilayaperumalg

pom.xml

@@ -61,11 +61,11 @@
 
 		<spring-cloud-dataflow-ui.version>2.1.1.BUILD-SNAPSHOT</spring-cloud-dataflow-ui.version>
 
-		<spring-cloud-deployer.version>2.0.2.BUILD-SNAPSHOT</spring-cloud-deployer.version>
+		<spring-cloud-deployer.version>2.0.3.BUILD-SNAPSHOT</spring-cloud-deployer.version>
 		<spring-cloud-deployer-local.version>2.0.4.BUILD-SNAPSHOT</spring-cloud-deployer-local.version>
 
 		<!-- note - check version of reactor at deployer-cf/cf-java-client uses -->
-		<spring-cloud-deployer-cloudfoundry.version>2.0.4.BUILD-SNAPSHOT</spring-cloud-deployer-cloudfoundry.version>
+		<spring-cloud-deployer-cloudfoundry.version>2.0.6.BUILD-SNAPSHOT</spring-cloud-deployer-cloudfoundry.version>
 		<reactor.version>3.2.0.RELEASE</reactor.version>
 		<spring-cloud-deployer-kubernetes.version>2.0.3.BUILD-SNAPSHOT</spring-cloud-deployer-kubernetes.version>
 		<kubernetes-client.version>4.1.0</kubernetes-client.version>

spring-cloud-dataflow-classic-docs/src/test/java/org/springframework/cloud/dataflow/server/rest/documentation/ApiDocumentation.java

@@ -106,6 +106,7 @@ public void index() throws Exception {
 				linkWithRel("tasks/executions/name").description("Returns all task executions for a given Task name"),
 				linkWithRel("tasks/executions/execution").description("Provides details for a specific task execution"),
 				linkWithRel("tasks/platforms").description("Provides platform accounts for launching tasks"),
+				linkWithRel("tasks/logs").description("Retrieve the task application log"),
 
 				linkWithRel("streams/definitions").description("Exposes the Streams resource"),
 				linkWithRel("streams/definitions/definition").description("Handle a specific Stream definition"),
@@ -118,6 +119,10 @@ public void index() throws Exception {
 				linkWithRel("streams/deployments/rollback/{name}/{version}").description("Rollback the stream to the previous or a specific version of the stream"),
 				linkWithRel("streams/deployments/update/{name}").description("Update the stream."),
 				linkWithRel("streams/deployments/platform/list").description("List of supported deployment platforms"),
+				linkWithRel("streams/logs").description("Retrieve application logs of the stream"),
+				linkWithRel("streams/logs/{streamName}").description("Retrieve application logs of the stream"),
+				linkWithRel("streams/logs/{streamName}/{appName}").description("Retrieve a specific application log of the stream"),
+
 
 				linkWithRel("tools/parseTaskTextToGraph").description("Parse a task definition into a graph structure"),
 				linkWithRel("tools/convertTaskGraphToText").description("Convert a graph format into " + "DSL text format")),
@@ -142,6 +147,12 @@ public void index() throws Exception {
 						fieldWithPath("_links.runtime/streams.href").description("Link to the runtime/streams"),
 						fieldWithPath("_links.runtime/streams.templated").type(JsonFieldType.BOOLEAN).optional().description("Link runtime/streams is templated"),
 
+						fieldWithPath("_links.streams/logs.href").description("Link to the streams/logs"),
+						fieldWithPath("_links.streams/logs/{streamName}.href").description("Link to the streams/logs/{streamName}"),
+						fieldWithPath("_links.streams/logs/{streamName}/{appName}.href").description("Link to the streams/logs/{streamName}/{appName}"),
+						fieldWithPath("_links.streams/logs/{streamName}.templated").type(JsonFieldType.BOOLEAN).optional().description("Link streams/logs/{streamName} is templated"),
+						fieldWithPath("_links.streams/logs/{streamName}/{appName}.templated").type(JsonFieldType.BOOLEAN).optional().description("Link streams/logs/{streamName}/{appName} is templated"),
+
 						fieldWithPath("_links.streams/deployments.href").description("Link to the streams/deployments"),
 						fieldWithPath("_links.streams/deployments/{name}.href").description("Link to the streams/deployments/{name}"),
 						fieldWithPath("_links.streams/deployments/{name}.templated").type(JsonFieldType.BOOLEAN).optional().description("Link streams/deployments/{name} is templated"),
@@ -173,6 +184,9 @@ public void index() throws Exception {
 						fieldWithPath("_links.tasks/executions/execution.href").description("Link to the tasks/executions/execution"),
 						fieldWithPath("_links.tasks/executions/execution.templated").type(JsonFieldType.BOOLEAN).optional().description("Link tasks/executions/execution is templated"),
 
+						fieldWithPath("_links.tasks/logs.href").description("Link to the tasks/logs"),
+						fieldWithPath("_links.tasks/logs.templated").type(JsonFieldType.BOOLEAN).optional().description("Link tasks/logs is templated"),
+
 						fieldWithPath("_links.tasks/schedules.href").description("Link to the tasks/executions/schedules"),
 						fieldWithPath("_links.tasks/schedules/instances.href").description("Link to the tasks/schedules/instances"),
 						fieldWithPath("_links.tasks/schedules/instances.templated").type(JsonFieldType.BOOLEAN).optional().description("Link tasks/schedules/instances is templated"),

spring-cloud-dataflow-platform-cloudfoundry/src/main/java/org/springframework/cloud/dataflow/server/config/cloudfoundry/CloudFoundryTaskPlatformFactory.java

@@ -28,6 +28,7 @@
 import org.cloudfoundry.operations.DefaultCloudFoundryOperations;
 import org.cloudfoundry.reactor.ConnectionContext;
 import org.cloudfoundry.reactor.TokenProvider;
+import org.cloudfoundry.reactor.doppler.ReactorDopplerClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -46,6 +47,7 @@
 
 /**
  * @author David Turanski
+ * @author Ilayaperumal Gopinathan
  **/
 public class CloudFoundryTaskPlatformFactory extends AbstractTaskPlatformFactory<CloudFoundryPlatformProperties> {
 
@@ -138,6 +140,9 @@ private CloudFoundryOperations cloudFoundryOperations(CloudFoundryClient cloudFo
 		return DefaultCloudFoundryOperations
 				.builder().cloudFoundryClient(cloudFoundryClient)
 				.organization(connectionProperties(account).getOrg())
+				.dopplerClient(ReactorDopplerClient.builder()
+						.connectionContext(connectionContext(account))
+						.tokenProvider(tokenProvider(account)).build())
 				.space(connectionProperties(account).getSpace()).build();
 	}
 

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/config/DataFlowControllerAutoConfiguration.java

@@ -69,9 +69,11 @@
 import org.springframework.cloud.dataflow.server.controller.RuntimeStreamsController;
 import org.springframework.cloud.dataflow.server.controller.StreamDefinitionController;
 import org.springframework.cloud.dataflow.server.controller.StreamDeploymentController;
+import org.springframework.cloud.dataflow.server.controller.StreamLogsController;
 import org.springframework.cloud.dataflow.server.controller.StreamValidationController;
 import org.springframework.cloud.dataflow.server.controller.TaskDefinitionController;
 import org.springframework.cloud.dataflow.server.controller.TaskExecutionController;
+import org.springframework.cloud.dataflow.server.controller.TaskLogsController;
 import org.springframework.cloud.dataflow.server.controller.TaskPlatformController;
 import org.springframework.cloud.dataflow.server.controller.TaskSchedulerController;
 import org.springframework.cloud.dataflow.server.controller.ToolsController;
@@ -290,6 +292,11 @@ public TaskValidationService taskValidationService(AppRegistryService appRegistr
 		public TaskValidationController taskValidationController(TaskValidationService taskValidationService) {
 			return new TaskValidationController(taskValidationService);
 		}
+
+		@Bean
+		public TaskLogsController taskLogsController(TaskExecutionService taskExecutionService) {
+			return new TaskLogsController(taskExecutionService);
+		}
 	}
 
 	@Configuration
@@ -332,6 +339,11 @@ public RuntimeStreamsController runtimeStreamsController(StreamDeployer streamDe
 			return new RuntimeStreamsController(streamDeployer);
 		}
 
+		@Bean
+		public StreamLogsController streamLogsController(StreamDeployer streamDeployer) {
+			return new StreamLogsController(streamDeployer);
+		}
+
 		@Bean
 		@ConditionalOnMissingBean(name = "runtimeAppsStatusFJPFB")
 		public ForkJoinPoolFactoryBean runtimeAppsStatusFJPFB() {

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/RootController.java

@@ -114,10 +114,12 @@ public RootResource info() {
 			root.add(ControllerLinkBuilder.linkTo(ControllerLinkBuilder.methodOn(StreamDeploymentController.class).platformList()).withRel("streams/deployments/platform/list"));
 			root.add(ControllerLinkBuilder.linkTo(ControllerLinkBuilder.methodOn(StreamDeploymentController.class).rollback(null, null)).withRel("streams/deployments/rollback/{name}/{version}"));
 			root.add(ControllerLinkBuilder.linkTo(ControllerLinkBuilder.methodOn(StreamDeploymentController.class).update(null, null)).withRel("streams/deployments/update/{name}"));
-
 			root.add(
 					unescapeTemplateVariables(entityLinks.linkToSingleResource(StreamDeploymentResource.class, "{name}")
 							.withRel("streams/deployments/deployment")));
+			root.add(ControllerLinkBuilder.linkTo(StreamLogsController.class).withRel("streams/logs"));
+			root.add(ControllerLinkBuilder.linkTo(ControllerLinkBuilder.methodOn(StreamLogsController.class).getLog(null)).withRel("streams/logs/{streamName}"));
+			root.add(ControllerLinkBuilder.linkTo(ControllerLinkBuilder.methodOn(StreamLogsController.class).getLog(null, null)).withRel("streams/logs/{streamName}/{appName}"));
 		}
 		if (featuresProperties.isTasksEnabled()) {
 			root.add(entityLinks.linkToCollectionResource(LauncherResource.class).withRel("tasks/platforms"));
@@ -134,6 +136,7 @@ public RootResource info() {
 					.withRel("tasks/executions/execution")));
 			root.add(unescapeTemplateVariables(entityLinks.linkToSingleResource(TaskAppStatusResource.class, "{name}")
 					.withRel("tasks/validation")));
+			root.add(ControllerLinkBuilder.linkTo(ControllerLinkBuilder.methodOn(TaskLogsController.class).getLog(null, null)).withRel("tasks/logs"));
 
 			if (featuresProperties.isSchedulesEnabled()) {
 				root.add(entityLinks.linkToCollectionResource(ScheduleInfoResource.class).withRel("tasks/schedules"));
@@ -163,6 +166,7 @@ public RootResource info() {
 			taskTemplated = entityLinks.linkToCollectionResource(JobExecutionThinResource.class).getHref() + "{?name}";
 			root.add(new Link(taskTemplated).withRel("jobs/thinexecutions/name"));
 
+
 		}
 		root.add(entityLinks.linkToCollectionResource(AppRegistrationResource.class).withRel("apps"));
 		root.add(entityLinks.linkToCollectionResource(AboutResource.class).withRel("about"));

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/StreamLogsController.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.dataflow.server.controller;
+
+import org.springframework.cloud.dataflow.server.stream.StreamDeployer;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.Assert;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * Retrieves logs of deployed stream applications.
+ *
+ * @author Ilayaperumal Gopinathan
+ */
+@RestController
+@RequestMapping("/streams/logs")
+public class StreamLogsController {
+
+	private final StreamDeployer streamDeployer;
+
+	/**
+	 * Construct Stream logs controller.
+	 *
+	 * @param streamDeployer the deployer this controller uses to get the logs of
+	 * deployed stream apps
+	 */
+	public StreamLogsController(StreamDeployer streamDeployer) {
+		Assert.notNull(streamDeployer, "StreamDeployer must not be null");
+		this.streamDeployer = streamDeployer;
+	}
+
+	@RequestMapping("{streamName}")
+	public ResponseEntity<String> getLog(@PathVariable String streamName) {
+		return new ResponseEntity<>(this.streamDeployer.getLog(streamName), HttpStatus.OK);
+	}
+
+	@RequestMapping("{streamName}/{appName}")
+	public ResponseEntity<String> getLog(@PathVariable String streamName, @PathVariable String appName) {
+		return new ResponseEntity<>(this.streamDeployer.getLog(streamName, appName), HttpStatus.OK);
+	}
+}

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/TaskLogsController.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.dataflow.server.controller;
+
+import org.springframework.cloud.dataflow.server.service.TaskExecutionService;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.Assert;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * Retrieves logs of task applications.
+ *
+ * @author Ilayaperumal Gopinathan
+ */
+@RestController
+@RequestMapping("/tasks/logs")
+public class TaskLogsController {
+
+	private final TaskExecutionService taskExecutionService;
+
+	/**
+	 * Construct Task logs controller.
+	 *
+	 * @param taskExecutionService the task execution service that this controller uses to get the logs of
+	 * launched task applications.
+	 */
+	public TaskLogsController(TaskExecutionService taskExecutionService) {
+		Assert.notNull(taskExecutionService, "TaskExecutionService must not be null");
+		this.taskExecutionService = taskExecutionService;
+	}
+
+	@RequestMapping(value = "", method = RequestMethod.GET, params = "name")
+	@ResponseStatus(HttpStatus.OK)
+	public ResponseEntity<String> getLog(@RequestParam(required = false, defaultValue = "default") String platformName,
+			@RequestParam("name") String taskName) {
+		return new ResponseEntity<>(this.taskExecutionService.getLog(platformName, taskName), HttpStatus.OK);
+	}
+}

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/service/TaskExecutionService.java

@@ -42,4 +42,13 @@ public interface TaskExecutionService {
 	 */
 	long executeTask(String taskName, Map<String, String> taskDeploymentProperties, List<String> commandLineArgs);
 
+	/**
+	 * Retrieve logs for the task application.
+	 *
+	 * @param platformName the name of the platform
+	 * @param taskId the ID that uniquely identifies the task
+	 * @return the logs of the task application.
+	 */
+	String getLog(String platformName, String taskId);
+
 }

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/service/impl/DefaultTaskExecutionService.java

@@ -22,6 +22,9 @@
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.cloud.dataflow.audit.service.AuditRecordService;
 import org.springframework.cloud.dataflow.core.AuditActionType;
 import org.springframework.cloud.dataflow.core.AuditOperationType;
@@ -92,6 +95,8 @@ public class DefaultTaskExecutionService implements TaskExecutionService {
 
 	public static final String TASK_PLATFORM_NAME = "spring.cloud.dataflow.task.platformName";
 
+	protected final Log logger = LogFactory.getLog(getClass().getName());
+
 	/**
 	 * Initializes the {@link DefaultTaskExecutionService}.
 	 *
@@ -182,6 +187,11 @@ public long executeTask(String taskName, Map<String, String> taskDeploymentPrope
 		return taskExecution.getExecutionId();
 	}
 
+	@Override
+	public String getLog(String platformName, String taskId) {
+		return findTaskLauncher(platformName).getLog(taskId);
+	}
+
 	private TaskLauncher findTaskLauncher(String platformName) {
 		Launcher launcher = this.launcherRepository.findByName(platformName);
 		if (launcher == null) {

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/stream/SkipperStreamDeployer.java

@@ -488,6 +488,16 @@ public List<AppStatus> getStreamStatuses(String streamName) {
 		return skipperStatus(streamName);
 	}
 
+	@Override
+	public String getLog(String streamName) {
+		return this.skipperClient.getLog(streamName);
+	}
+
+	@Override
+	public String getLog(String streamName, String appName) {
+		return this.skipperClient.getLog(streamName, appName);
+	}
+
 	private List<AppStatus> getStreamsStatuses(List<String> streamNames) {
 		try {
 			return this.forkJoinPool.submit(() -> streamNames.stream().parallel()