Create the DeployerPartitionHandler and related
DeployerStepExecutionHandler Created a PartitionHandler that delegates to a TaskLauncher from Spring Cloud Deployer to execute workers. Resolves spring-cloud/spring-cloud-task#109 Updates per code review
This commit is contained in:
Michael Minella
committed by
Glenn Renfro
Glenn Renfro
parent
726441dda3
commit
3d3b90812e
@@ -15,12 +15,16 @@
|
||||
*/
|
||||
package org.springframework.cloud.task.batch.configuration;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.batch.core.job.AbstractJob;
|
||||
import org.springframework.beans.BeansException;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.config.BeanPostProcessor;
|
||||
import org.springframework.cloud.task.batch.listener.TaskBatchExecutionListener;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Injects a configured {@link TaskBatchExecutionListener} into any batch jobs (beans
|
||||
@@ -35,10 +39,18 @@ public class TaskBatchExecutionListenerBeanPostProcessor implements BeanPostProc
|
||||
@Autowired
|
||||
private ApplicationContext applicationContext;
|
||||
|
||||
private List<String> jobNames = new ArrayList<>();
|
||||
|
||||
@Override
|
||||
public Object postProcessBeforeInitialization(Object bean, String beanName)
|
||||
throws BeansException {
|
||||
|
||||
if(jobNames.size() > 0) {
|
||||
if(!jobNames.contains(beanName)) {
|
||||
return bean;
|
||||
}
|
||||
}
|
||||
|
||||
int length = this.applicationContext
|
||||
.getBeanNamesForType(TaskBatchExecutionListener.class).length;
|
||||
|
||||
@@ -61,4 +73,10 @@ public class TaskBatchExecutionListenerBeanPostProcessor implements BeanPostProc
|
||||
throws BeansException {
|
||||
return bean;
|
||||
}
|
||||
|
||||
public void setJobNames(List<String> jobNames) {
|
||||
Assert.notNull(jobNames, "A list is required");
|
||||
|
||||
this.jobNames = jobNames;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,332 @@
|
||||
/*
|
||||
* Copyright 2016 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
|
||||
*
|
||||
* http://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.task.batch.partition;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.StepExecution;
|
||||
import org.springframework.batch.core.explore.JobExplorer;
|
||||
import org.springframework.batch.core.partition.PartitionHandler;
|
||||
import org.springframework.batch.core.partition.StepExecutionSplitter;
|
||||
import org.springframework.batch.poller.DirectPoller;
|
||||
import org.springframework.batch.poller.Poller;
|
||||
import org.springframework.cloud.deployer.spi.core.AppDefinition;
|
||||
import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
|
||||
import org.springframework.cloud.deployer.spi.task.TaskLauncher;
|
||||
import org.springframework.cloud.task.listener.annotation.BeforeTask;
|
||||
import org.springframework.cloud.task.repository.TaskExecution;
|
||||
import org.springframework.context.EnvironmentAware;
|
||||
import org.springframework.core.env.AbstractEnvironment;
|
||||
import org.springframework.core.env.Environment;
|
||||
import org.springframework.core.env.MapPropertySource;
|
||||
import org.springframework.core.env.PropertySource;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
|
||||
/**
|
||||
* <p>A {@link PartitionHandler} implementation that delegates to a {@link TaskLauncher} for
|
||||
* each of the workers. The id of the worker's StepExecution is passed as an environment
|
||||
* variable to the worker. The worker, bootstrapped by the
|
||||
* {@link DeployerStepExecutionHandler}, looks up the StepExecution in the JobRepository
|
||||
* and executes it. This PartitionHandler polls the JobRepository for the results.</p>
|
||||
*
|
||||
* <p>If the job fails, the partitions will be re-executed per normal batch rules (steps that
|
||||
* are complete should do nothing, failed steps should restart based on their
|
||||
* configurations).</p>
|
||||
*
|
||||
* <p>This PartitionHandler and all of the worker processes must share the same JobRepository
|
||||
* data store (aka point the same database).</p>
|
||||
*
|
||||
* @author Michael Minella
|
||||
*/
|
||||
public class DeployerPartitionHandler implements PartitionHandler, EnvironmentAware {
|
||||
|
||||
public static final String SPRING_CLOUD_TASK_JOB_EXECUTION_ID =
|
||||
"spring.cloud.task.job-execution-id";
|
||||
|
||||
public static final String SPRING_CLOUD_TASK_STEP_EXECUTION_ID =
|
||||
"spring.cloud.task.step-execution-id";
|
||||
|
||||
public static final String SPRING_CLOUD_TASK_STEP_NAME =
|
||||
"spring.cloud.task.step-name";
|
||||
|
||||
private int maxWorkers = -1;
|
||||
|
||||
private int gridSize = 1;
|
||||
|
||||
private int currentWorkers = 0;
|
||||
|
||||
private TaskLauncher taskLauncher;
|
||||
|
||||
private JobExplorer jobExplorer;
|
||||
|
||||
private TaskExecution taskExecution;
|
||||
|
||||
private Resource resource;
|
||||
|
||||
private Map<String, String> environmentProperties = new HashMap<>();
|
||||
|
||||
private String stepName;
|
||||
|
||||
private Log logger = LogFactory.getLog(DeployerPartitionHandler.class);
|
||||
|
||||
private long pollInterval = 10000;
|
||||
|
||||
private long timeout = -1;
|
||||
|
||||
private Environment environment;
|
||||
|
||||
public DeployerPartitionHandler(TaskLauncher taskLauncher,
|
||||
JobExplorer jobExplorer,
|
||||
Resource resource,
|
||||
String stepName) {
|
||||
Assert.notNull(taskLauncher, "A taskLauncher is required");
|
||||
Assert.notNull(jobExplorer, "A jobExplorer is required");
|
||||
Assert.notNull(resource, "A resource is required");
|
||||
Assert.hasText(stepName, "A step name is required");
|
||||
|
||||
this.taskLauncher = taskLauncher;
|
||||
this.jobExplorer = jobExplorer;
|
||||
this.resource = resource;
|
||||
this.stepName = stepName;
|
||||
}
|
||||
|
||||
/**
|
||||
* The maximum number of workers to be executing at once.
|
||||
*
|
||||
* @param maxWorkers number of workers. Defaults to -1 (unlimited)
|
||||
*/
|
||||
public void setMaxWorkers(int maxWorkers) {
|
||||
Assert.isTrue(maxWorkers != 0, "maxWorkers cannot be 0");
|
||||
this.maxWorkers = maxWorkers;
|
||||
}
|
||||
|
||||
/**
|
||||
* Approximate size of the pool of worker JVMs available. May be used by the
|
||||
* {@link StepExecutionSplitter} to determine how many partitions to create (at the
|
||||
* discretion of the {@link org.springframework.batch.core.partition.support.Partitioner}).
|
||||
*
|
||||
* @param gridSize size of grid. Defaults to 1
|
||||
*/
|
||||
public void setGridSize(int gridSize) {
|
||||
this.gridSize = gridSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* System properties to be made available for all workers.
|
||||
*
|
||||
* @param environmentProperties Map of properties
|
||||
*/
|
||||
public void setEnvironmentProperties(Map<String, String> environmentProperties) {
|
||||
this.environmentProperties = environmentProperties;
|
||||
}
|
||||
|
||||
/**
|
||||
* The interval to check the job repository for completed steps.
|
||||
*
|
||||
* @param pollInterval interval. Defaults to 10 seconds
|
||||
*/
|
||||
public void setPollInterval(long pollInterval) {
|
||||
this.pollInterval = pollInterval;
|
||||
}
|
||||
|
||||
/**
|
||||
* Timeout for the master step. This is a timeout for all workers to complete.
|
||||
*
|
||||
* @param timeout timeout. Defaults to none (-1).
|
||||
*/
|
||||
public void setTimeout(long timeout) {
|
||||
this.timeout = timeout;
|
||||
}
|
||||
|
||||
@BeforeTask
|
||||
public void beforeTask(TaskExecution taskExecution) {
|
||||
this.taskExecution = taskExecution;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Collection<StepExecution> handle(StepExecutionSplitter stepSplitter,
|
||||
StepExecution stepExecution) throws Exception {
|
||||
|
||||
final Set<StepExecution> tempCandidates =
|
||||
stepSplitter.split(stepExecution, this.gridSize);
|
||||
|
||||
// Following two lines due to https://jira.spring.io/browse/BATCH-2490
|
||||
final Set<StepExecution> candidates = new HashSet<>(tempCandidates.size());
|
||||
candidates.addAll(tempCandidates);
|
||||
|
||||
int partitions = candidates.size();
|
||||
|
||||
logger.debug(String.format("%s partitions were returned", partitions));
|
||||
|
||||
final Set<StepExecution> executed = new HashSet<>(candidates.size());
|
||||
|
||||
if(CollectionUtils.isEmpty(candidates)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
launchWorkers(candidates, executed);
|
||||
|
||||
candidates.removeAll(executed);
|
||||
|
||||
return pollReplies(stepExecution, executed, candidates, partitions);
|
||||
}
|
||||
|
||||
private void launchWorkers(Set<StepExecution> candidates, Set<StepExecution> executed) {
|
||||
for (StepExecution execution : candidates) {
|
||||
if(this.currentWorkers < this.maxWorkers || this.maxWorkers < 0) {
|
||||
launchWorker(execution);
|
||||
this.currentWorkers++;
|
||||
|
||||
executed.add(execution);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void launchWorker(StepExecution workerStepExecution) {
|
||||
//TODO: Refactor these to be passed as command line args once SCD-20 is complete
|
||||
// https://github.com/spring-cloud/spring-cloud-deployer/issues/20
|
||||
Map<String, String> parameters = getParameters(this.taskExecution.getParameters());
|
||||
parameters.put(SPRING_CLOUD_TASK_JOB_EXECUTION_ID,
|
||||
String.valueOf(workerStepExecution.getJobExecution().getId()));
|
||||
parameters.put(SPRING_CLOUD_TASK_STEP_EXECUTION_ID,
|
||||
String.valueOf(workerStepExecution.getId()));
|
||||
parameters.put(SPRING_CLOUD_TASK_STEP_NAME, this.stepName);
|
||||
|
||||
AppDefinition definition =
|
||||
new AppDefinition(String.format("%s:%s:%s",
|
||||
taskExecution.getTaskName(),
|
||||
workerStepExecution.getJobExecution().getJobInstance().getJobName(),
|
||||
workerStepExecution.getStepName()),
|
||||
parameters);
|
||||
|
||||
Map<String, String> environmentProperties = new HashMap<>(this.environmentProperties.size());
|
||||
environmentProperties.putAll(getCurrentEnvironmentProperties());
|
||||
environmentProperties.putAll(this.environmentProperties);
|
||||
|
||||
AppDeploymentRequest request =
|
||||
new AppDeploymentRequest(definition, this.resource, environmentProperties);
|
||||
|
||||
taskLauncher.launch(request);
|
||||
}
|
||||
|
||||
private Collection<StepExecution> pollReplies(final StepExecution masterStepExecution,
|
||||
final Set<StepExecution> executed,
|
||||
final Set<StepExecution> candidates,
|
||||
final int size) throws Exception {
|
||||
|
||||
final Collection<StepExecution> result = new ArrayList<>(executed.size());
|
||||
|
||||
Callable<Collection<StepExecution>> callback = new Callable<Collection<StepExecution>>() {
|
||||
@Override
|
||||
public Collection<StepExecution> call() throws Exception {
|
||||
Set<StepExecution> newExecuted = new HashSet<>();
|
||||
|
||||
for (StepExecution curStepExecution : executed) {
|
||||
if (!result.contains(curStepExecution)) {
|
||||
StepExecution partitionStepExecution =
|
||||
jobExplorer.getStepExecution(masterStepExecution.getJobExecutionId(), curStepExecution.getId());
|
||||
|
||||
if (isComplete(partitionStepExecution.getStatus())) {
|
||||
result.add(partitionStepExecution);
|
||||
currentWorkers--;
|
||||
|
||||
if (!candidates.isEmpty()) {
|
||||
|
||||
launchWorkers(candidates, newExecuted);
|
||||
candidates.removeAll(newExecuted);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
executed.addAll(newExecuted);
|
||||
|
||||
if(result.size() == size) {
|
||||
return result;
|
||||
}
|
||||
else {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
Poller<Collection<StepExecution>> poller = new DirectPoller<>(this.pollInterval);
|
||||
Future<Collection<StepExecution>> resultsFuture = poller.poll(callback);
|
||||
|
||||
if(timeout >= 0) {
|
||||
return resultsFuture.get(timeout, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
else {
|
||||
return resultsFuture.get();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isComplete(BatchStatus status) {
|
||||
return status.equals(BatchStatus.COMPLETED) || status.isGreaterThan(BatchStatus.STARTED);
|
||||
}
|
||||
|
||||
private Map<String, String> getParameters(List<String> parameters) {
|
||||
Map<String, String> parameterMap = new HashMap<>(parameters.size());
|
||||
|
||||
for (String parameter : parameters) {
|
||||
String[] pieces = parameter.split("=");
|
||||
parameterMap.put(pieces[0], pieces[1]);
|
||||
}
|
||||
|
||||
return parameterMap;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setEnvironment(Environment environment) {
|
||||
this.environment = environment;
|
||||
}
|
||||
|
||||
private Map<String, String> getCurrentEnvironmentProperties() {
|
||||
Map<String, String> currentEnvironment = new HashMap<>();
|
||||
|
||||
Set<String> keys = new HashSet<>();
|
||||
|
||||
for(Iterator it = ((AbstractEnvironment) this.environment).getPropertySources().iterator(); it.hasNext(); ) {
|
||||
PropertySource propertySource = (PropertySource) it.next();
|
||||
if (propertySource instanceof MapPropertySource) {
|
||||
keys.addAll(Arrays.asList(((MapPropertySource) propertySource).getPropertyNames()));
|
||||
}
|
||||
}
|
||||
|
||||
for (String key : keys) {
|
||||
currentEnvironment.put(key, this.environment.getProperty(key));
|
||||
}
|
||||
|
||||
return currentEnvironment;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
/*
|
||||
* Copyright 2016 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
|
||||
*
|
||||
* http://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.task.batch.partition;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.JobInterruptedException;
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.StepExecution;
|
||||
import org.springframework.batch.core.explore.JobExplorer;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.step.NoSuchStepException;
|
||||
import org.springframework.batch.core.step.StepLocator;
|
||||
import org.springframework.batch.integration.partition.BeanFactoryStepLocator;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.CommandLineRunner;
|
||||
import org.springframework.core.env.Environment;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* <p>A {@link CommandLineRunner} used to execute a {@link Step}. No result is provided
|
||||
* directly to the associated {@link DeployerPartitionHandler} as it will obtain the step
|
||||
* results directly from the shared job repository.</p>
|
||||
*
|
||||
* <p>The {@link StepExecution} is rehydrated based on the environment variables provided.
|
||||
* Specifically, the following variables are required:</p>
|
||||
* <ul>
|
||||
* <li>{@link DeployerPartitionHandler#SPRING_CLOUD_TASK_JOB_EXECUTION_ID}: The id of
|
||||
* the JobExecution.</li>
|
||||
* <li>{@link DeployerPartitionHandler#SPRING_CLOUD_TASK_STEP_EXECUTION_ID}: The id of
|
||||
* the StepExecution.</li>
|
||||
* <li>{@link DeployerPartitionHandler#SPRING_CLOUD_TASK_STEP_NAME}: The id of the
|
||||
* bean definition for the Step to execute. The id must be found within the provided
|
||||
* {@link BeanFactory}</li>
|
||||
* </ul>
|
||||
*
|
||||
* @author Michael Minella
|
||||
*/
|
||||
public class DeployerStepExecutionHandler implements CommandLineRunner {
|
||||
|
||||
private JobExplorer jobExplorer;
|
||||
|
||||
private JobRepository jobRepository;
|
||||
|
||||
private Log logger = LogFactory.getLog(DeployerStepExecutionHandler.class);
|
||||
|
||||
@Autowired
|
||||
private Environment environment;
|
||||
|
||||
private StepLocator stepLocator;
|
||||
|
||||
public DeployerStepExecutionHandler(BeanFactory beanFactory, JobExplorer jobExplorer, JobRepository jobRepository) {
|
||||
Assert.notNull(beanFactory, "A beanFactory is required");
|
||||
Assert.notNull(jobExplorer, "A jobExplorer is required");
|
||||
Assert.notNull(jobRepository, "A jobRepository is required");
|
||||
|
||||
this.stepLocator = new BeanFactoryStepLocator();
|
||||
((BeanFactoryStepLocator) this.stepLocator).setBeanFactory(beanFactory);
|
||||
|
||||
this.jobExplorer = jobExplorer;
|
||||
this.jobRepository = jobRepository;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run(String... args) throws Exception {
|
||||
|
||||
validateRequest();
|
||||
|
||||
Long jobExecutionId = Long.parseLong(environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID));
|
||||
Long stepExecutionId = Long.parseLong(environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID));
|
||||
StepExecution stepExecution = jobExplorer.getStepExecution(jobExecutionId, stepExecutionId);
|
||||
|
||||
if (stepExecution == null) {
|
||||
throw new NoSuchStepException(String.format("No StepExecution could be located for step execution id %s within job execution %s", stepExecutionId, jobExecutionId));
|
||||
}
|
||||
|
||||
String stepName = environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME);
|
||||
Step step = stepLocator.getStep(stepName);
|
||||
|
||||
try {
|
||||
logger.debug(String.format("Executing step %s with step execution id %s and job execution id %s", stepExecution.getStepName(), stepExecutionId, jobExecutionId));
|
||||
|
||||
step.execute(stepExecution);
|
||||
}
|
||||
catch (JobInterruptedException e) {
|
||||
stepExecution.setStatus(BatchStatus.STOPPED);
|
||||
jobRepository.update(stepExecution);
|
||||
}
|
||||
catch (Throwable e) {
|
||||
stepExecution.addFailureException(e);
|
||||
stepExecution.setStatus(BatchStatus.FAILED);
|
||||
jobRepository.update(stepExecution);
|
||||
}
|
||||
}
|
||||
|
||||
private void validateRequest() {
|
||||
Assert.isTrue(environment.containsProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_JOB_EXECUTION_ID), "A job execution id is required");
|
||||
Assert.isTrue(environment.containsProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_EXECUTION_ID), "A step execution id is required");
|
||||
Assert.isTrue(environment.containsProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME), "A step name is required");
|
||||
|
||||
Assert.isTrue(this.stepLocator.getStepNames().contains(environment.getProperty(DeployerPartitionHandler.SPRING_CLOUD_TASK_STEP_NAME)), "The step requested cannot be found in the provided BeanFactory");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user