From 8bdd18a34aa5f23ed24399c8ed0628e7fc8ccbc5 Mon Sep 17 00:00:00 2001 From: David Turanski Date: Wed, 10 Jun 2020 14:20:39 -0400 Subject: [PATCH] Implement Tasklauncher function Clean up test --- function/tasklauncher-function/README.adoc | 39 ++++ function/tasklauncher-function/pom.xml | 55 ++++++ .../cloud/fn/tasklauncher/LaunchRequest.java | 66 +++++++ .../fn/tasklauncher/TaskLauncherFunction.java | 167 ++++++++++++++++++ .../TaskLauncherFunctionConfiguration.java | 46 +++++ .../TaskLauncherFunctionProperties.java | 38 ++++ .../TaskLauncherFunctionApplicationTests.java | 151 ++++++++++++++++ pom.xml | 1 + 8 files changed, 563 insertions(+) create mode 100644 function/tasklauncher-function/README.adoc create mode 100644 function/tasklauncher-function/pom.xml create mode 100644 function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java create mode 100644 function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunction.java create mode 100644 function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionConfiguration.java create mode 100644 function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java create mode 100644 function/tasklauncher-function/src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java diff --git a/function/tasklauncher-function/README.adoc b/function/tasklauncher-function/README.adoc new file mode 100644 index 00000000..019be1b4 --- /dev/null +++ b/function/tasklauncher-function/README.adoc @@ -0,0 +1,39 @@ +# TaskLauncher Function + +This module provides a `Function` that uses the Data Flow REST client to launch a registered task on a configured https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-tasks[task platform]. +The client must be configured to connect to a remote Data Flow Server, including any required authentication (see Configuration Options below). + +## Beans for injection + +You can import the `TaskLauncherFunctionConfiguration` configuration in a Spring Boot application and then inject the following bean. + +`taskLauncherFunction` + +You may inject this as `TaskLauncherFunction` which implements `Function>`. + +You can use `taskLauncherFunction` as a qualifier when injecting. + +Once injected, you can use the `apply` method of the `Function` to launch a task. +The function takes a link:src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java[LaunchRequest] as input. +This is a simple value object that specifies, at a minimum, the name of the task registered in Data Flow. +Optionally, you can pass command line arguments as a `List` and deployment properties as a `Map`. +The return value is an `Optional` containing the unique task ID of the launched instance if the launch request is successful. + +NOTE: This version of the tasklauncher requires Spring Cloud Data Flow version 2.4.x or higher + +## Configuration Options + +Specific properties, including the task platform name are prefixed with `tasklauncher`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java[TaskLauncherFunctionProperties.java] + +Data Flow client configuration properties are prefixed with `spring.cloud.dataflow.client`. +Please see https://github.com/spring-cloud/spring-cloud-dataflow/blob/master/spring-cloud-dataflow-rest-client/src/main/java/org/springframework/cloud/dataflow/rest/client/config/DataFlowClientProperties.java[DataFlowClientProperties.java] for more details. + +## Examples + +See this link:src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java[test suite] for examples of how this function is used. + +## Other usage + +See this link:../../../applications/sink/tasklauncher-sink/README.adoc[README] where this function is used to create a Spring Cloud Stream application to submit task launch requests. \ No newline at end of file diff --git a/function/tasklauncher-function/pom.xml b/function/tasklauncher-function/pom.xml new file mode 100644 index 00000000..f78f3253 --- /dev/null +++ b/function/tasklauncher-function/pom.xml @@ -0,0 +1,55 @@ + + + 4.0.0 + tasklauncher-function + 1.0.0-SNAPSHOT + tasklauncher-function + Spring Native Function for applying filter SpEL expressions + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + 2.5.1.RELEASE + + + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.cloud + spring-cloud-dataflow-rest-client + ${spring.cloud.dataflow.version} + + + org.springframework.cloud + spring-cloud-scheduler-spi + + + org.springframework.cloud + spring-cloud-skipper + + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + + diff --git a/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java b/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java new file mode 100644 index 00000000..54590e44 --- /dev/null +++ b/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java @@ -0,0 +1,66 @@ +/* + * Copyright 2019-2020 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.fn.tasklauncher; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import com.fasterxml.jackson.annotation.JsonProperty; + +import org.springframework.util.Assert; + +/** + * @author David Turanski + **/ + +public class LaunchRequest { + @JsonProperty("args") + private List commandlineArguments = new ArrayList<>(); + @JsonProperty("deploymentProps") + private Map deploymentProperties = new HashMap<>(); + @JsonProperty("name") + private String taskName; + + public List getCommandlineArguments() { + return commandlineArguments; + } + + public void setCommandlineArguments(List commandlineArguments) { + Assert.notNull(commandlineArguments, "'commandLineArguments' cannot be null."); + this.commandlineArguments = commandlineArguments; + } + + public Map getDeploymentProperties() { + return deploymentProperties; + } + + public void setDeploymentProperties(Map deploymentProperties) { + Assert.notNull(commandlineArguments, "'deploymentProperties' cannot be null."); + this.deploymentProperties = deploymentProperties; + } + + public String getTaskName() { + return taskName; + } + + public void setTaskName(String taskName) { + Assert.hasText(taskName, "'taskName' cannot be blank."); + this.taskName = taskName; + } +} diff --git a/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunction.java b/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunction.java new file mode 100644 index 00000000..097de153 --- /dev/null +++ b/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunction.java @@ -0,0 +1,167 @@ +/* + * Copyright 2019-2020 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.fn.tasklauncher; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Function; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.cloud.dataflow.rest.client.TaskOperations; +import org.springframework.cloud.dataflow.rest.resource.CurrentTaskExecutionsResource; +import org.springframework.cloud.dataflow.rest.resource.LauncherResource; +import org.springframework.hateoas.PagedModel; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * + * A {@link Function} that submits a task {@link LaunchRequest} to a Data Flow server. + * This will check if the Data Flow task platform is at capacity. If not, will submit the + * task launch request, otherwise it will return and log a warning message. + * + * @author David Turanski + **/ +public class TaskLauncherFunction implements Function>, InitializingBean { + private static final Log log = LogFactory.getLog(TaskLauncherFunction.class); + + static final String TASK_PLATFORM_NAME = "spring.cloud.dataflow.task.platformName"; + + private final TaskOperations taskOperations; + + private String platformName = "default"; + + public TaskLauncherFunction(TaskOperations taskOperations) { + Assert.notNull(taskOperations, "`taskOperations` cannot be null."); + this.taskOperations = taskOperations; + } + + /** + * + * @param launchRequest the task launch request for the Data Flow server. + * @return an {@code Optional} containing the task Id if the request is accepted or + * empty otherwise. + */ + @Override + public Optional apply(LaunchRequest launchRequest) { + if (platformIsAcceptingNewTasks()) { + return Optional.of(launchTask(launchRequest)); + } + log.warn(String.format("Platform is at capacity. Did not submit task launch request for task %s.", + launchRequest.getTaskName())); + return Optional.empty(); + } + + public boolean platformIsAcceptingNewTasks() { + + boolean availableForNewTasks; + int maximumTaskExecutions = 0; + int runningExecutionCount = 0; + + List currentPlatforms = new ArrayList<>(); + + boolean validPlatform = false; + for (CurrentTaskExecutionsResource currentTaskExecutionsResource : taskOperations.currentTaskExecutions()) { + if (currentTaskExecutionsResource.getName().equals(platformName)) { + maximumTaskExecutions = currentTaskExecutionsResource.getMaximumTaskExecutions(); + runningExecutionCount = currentTaskExecutionsResource.getRunningExecutionCount(); + validPlatform = true; + } + currentPlatforms.add(currentTaskExecutionsResource.getName()); + } + + // Verify for each request as configuration may have changed on server. + assertValidPlatform(validPlatform, currentPlatforms); + + availableForNewTasks = runningExecutionCount < maximumTaskExecutions; + if (!availableForNewTasks) { + log.warn(String.format( + "The data Flow task platform %s has reached its concurrent task execution limit: (%d)", + platformName, + maximumTaskExecutions)); + } + + return availableForNewTasks; + + } + + private long launchTask(LaunchRequest request) { + String requestPlatformName = request.getDeploymentProperties().get(TASK_PLATFORM_NAME); + if (StringUtils.hasText(requestPlatformName) && !platformName.equals(requestPlatformName)) { + throw new IllegalStateException( + String.format( + "Task Launch request for Task %s contains deployment property '%s=%s' which does not " + + "match the platform configured for the Task Launcher: '%s'", + request.getTaskName(), + TASK_PLATFORM_NAME, + request.getDeploymentProperties().get(TASK_PLATFORM_NAME), + platformName)); + } + log.info(String.format("Launching Task %s on platform %s", request.getTaskName(), platformName)); + long taskId = taskOperations.launch(request.getTaskName(), + enrichDeploymentProperties(request.getDeploymentProperties()), + request.getCommandlineArguments(), null); + log.info(String.format("Launched Task %s - task ID is %d", request.getTaskName(), taskId)); + return taskId; + } + + private Map enrichDeploymentProperties(Map deploymentProperties) { + if (!deploymentProperties.containsKey(TASK_PLATFORM_NAME)) { + Map enrichedProperties = new HashMap<>(); + enrichedProperties.putAll(deploymentProperties); + enrichedProperties.put(TASK_PLATFORM_NAME, platformName); + return enrichedProperties; + } + return deploymentProperties; + } + + public void setPlatformName(String platformName) { + this.platformName = platformName; + } + + @Override + public void afterPropertiesSet() { + PagedModel launchers = taskOperations.listPlatforms(); + + boolean validPlatform = false; + List currentPlatforms = new ArrayList<>(); + + for (LauncherResource launcherResource : launchers) { + currentPlatforms.add(launcherResource.getName()); + if (launcherResource.getName().equals(platformName)) { + validPlatform = true; + } + } + + assertValidPlatform(validPlatform, currentPlatforms); + } + + private void assertValidPlatform(boolean validPlatform, List currentPlatforms) { + Assert.notEmpty(currentPlatforms, "The Data Flow Server has no task platforms configured"); + + Assert.isTrue(validPlatform, String.format( + "The task launcher's platform name '%s' does not match one of the Data Flow server's configured task " + + "platforms: [%s].", + platformName, StringUtils.collectionToCommaDelimitedString(currentPlatforms))); + } +} diff --git a/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionConfiguration.java b/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionConfiguration.java new file mode 100644 index 00000000..463eeeb3 --- /dev/null +++ b/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionConfiguration.java @@ -0,0 +1,46 @@ +/* + * Copyright 2019-2020 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.fn.tasklauncher; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.dataflow.rest.client.DataFlowOperations; +import org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * Configuration class for the TaskLauncher Data Flow Sink. + * + * @author David Turanski + * @author Gunnar Hillert + */ +@Configuration +@EnableConfigurationProperties({TaskLauncherFunctionProperties.class, DataFlowClientProperties.class}) +public class TaskLauncherFunctionConfiguration { + + @Bean + public TaskLauncherFunction taskLauncherFunction( + DataFlowOperations dataFlowOperations, TaskLauncherFunctionProperties functionProperties) { + + if (dataFlowOperations.taskOperations() == null) { + throw new IllegalArgumentException("The SCDF server does not support task operations"); + } + TaskLauncherFunction function = new TaskLauncherFunction(dataFlowOperations.taskOperations()); + function.setPlatformName(functionProperties.getPlatformName()); + return function; + } +} diff --git a/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java b/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java new file mode 100644 index 00000000..632c64f1 --- /dev/null +++ b/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java @@ -0,0 +1,38 @@ +/* + * Copyright 2019-2020 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.fn.tasklauncher; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * @author David Turanski + **/ +@ConfigurationProperties("tasklauncher") +public class TaskLauncherFunctionProperties { + /** + * The Spring Cloud Data Flow platform to use for launching tasks. + */ + private String platformName = "default"; + + public String getPlatformName() { + return platformName; + } + + public void setPlatformName(String platformName) { + this.platformName = platformName; + } +} diff --git a/function/tasklauncher-function/src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java b/function/tasklauncher-function/src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java new file mode 100644 index 00000000..9010dc99 --- /dev/null +++ b/function/tasklauncher-function/src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java @@ -0,0 +1,151 @@ +/* + * Copyright 2019-2020 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.fn.tasklauncher; + +import java.util.Collections; +import java.util.Optional; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.cloud.dataflow.rest.client.DataFlowOperations; +import org.springframework.cloud.dataflow.rest.client.TaskOperations; +import org.springframework.cloud.dataflow.rest.resource.CurrentTaskExecutionsResource; +import org.springframework.cloud.dataflow.rest.resource.LauncherResource; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.Profile; +import org.springframework.hateoas.PagedModel; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyMap; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@SpringBootTest +public class TaskLauncherFunctionApplicationTests { + + @Autowired + private TaskLauncherFunction taskLauncherFunction; + + @Autowired + private TaskOperations taskOperations; + + @Test + public void successfulLaunch() { + LaunchRequest launchRequest = new LaunchRequest(); + launchRequest.setTaskName("someTask"); + setCurrentExecutionState(0); + Optional taskId = taskLauncherFunction.apply(launchRequest); + assertThat(taskId.isPresent()).isTrue(); + assertThat(taskId.get()).isEqualTo(1L); + + verify(taskOperations).launch("someTask", + Collections.singletonMap(TaskLauncherFunction.TASK_PLATFORM_NAME, "default"), + Collections.emptyList(), + null); + } + + @Test + public void taskPlatformAtCapacity() { + LaunchRequest launchRequest = new LaunchRequest(); + launchRequest.setTaskName("someTask"); + setCurrentExecutionState(3); + Optional taskId = taskLauncherFunction.apply(launchRequest); + assertThat(taskId.isPresent()).isFalse(); + } + + @Test + public void platformMismatch() { + LaunchRequest launchRequest = new LaunchRequest(); + launchRequest.setTaskName("someTask"); + launchRequest + .setDeploymentProperties(Collections.singletonMap(TaskLauncherFunction.TASK_PLATFORM_NAME, "other")); + setCurrentExecutionState(0); + assertThatIllegalStateException().isThrownBy(() -> taskLauncherFunction.apply(launchRequest)) + .withStackTraceContaining("does not match the platform configured for the Task Launcher"); + } + + private void setCurrentExecutionState(int runningExecutions) { + CurrentTaskExecutionsResource currentTaskExecutionsResource = new CurrentTaskExecutionsResource(); + currentTaskExecutionsResource.setMaximumTaskExecutions(3); + currentTaskExecutionsResource.setRunningExecutionCount(runningExecutions); + currentTaskExecutionsResource.setName("default"); + when(taskOperations.currentTaskExecutions()) + .thenReturn(Collections.singletonList(currentTaskExecutionsResource)); + when(taskOperations.launch(anyString(), anyMap(), anyList(), isNull())).thenReturn(1L); + } + + @Test + public void noLaunchersConfigured() { + ApplicationContextRunner contextRunner = new ApplicationContextRunner().withUserConfiguration(TestConfig.class); + assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> contextRunner + .withPropertyValues("spring.profiles.active=nolaunchers") + .run(context -> context.start())) + .withCauseInstanceOf(BeanCreationException.class) + .withRootCauseInstanceOf(IllegalArgumentException.class) + .withStackTraceContaining("The Data Flow Server has no task platforms configured"); + } + + @Configuration + @Import(TaskLauncherFunctionConfiguration.class) + static class TestConfig { + + @Bean + @Profile("default") + TaskOperations taskOperations() { + TaskOperations taskOperations = mock(TaskOperations.class); + LauncherResource launcherResource = mock(LauncherResource.class); + when(launcherResource.getName()).thenReturn("default"); + + when(taskOperations.listPlatforms()).thenReturn(PagedModel.of( + Collections.singletonList(launcherResource), (PagedModel.PageMetadata) null)); + return taskOperations; + } + + @Bean + @Profile("nolaunchers") + TaskOperations taskOperationsNoLaunchers() { + TaskOperations taskOperations = mock(TaskOperations.class); + when(taskOperations.listPlatforms()).thenReturn(PagedModel.of( + Collections.emptyList(), (PagedModel.PageMetadata) null)); + return taskOperations; + } + + @Bean + DataFlowOperations dataFlowOperations(TaskOperations taskOperations) { + DataFlowOperations dataFlowOperations = mock(DataFlowOperations.class); + when(dataFlowOperations.taskOperations()).thenReturn(taskOperations); + return dataFlowOperations; + } + } + + @SpringBootApplication + static class TestApplication { + } +} diff --git a/pom.xml b/pom.xml index cad847d0..8459426c 100644 --- a/pom.xml +++ b/pom.xml @@ -72,6 +72,7 @@ function/spel-function function/payload-converter-function function/splitter-function + function/tasklauncher-function supplier/file-supplier supplier/ftp-supplier