diff --git a/dataflow-migrate-schedules/README.adoc b/dataflow-migrate-schedules/README.adoc index 8e3f3a2..eaf1e78 100644 --- a/dataflow-migrate-schedules/README.adoc +++ b/dataflow-migrate-schedules/README.adoc @@ -1,14 +1,14 @@ -= Cloud Scheduler Migration += Spring Cloud Data Flow Schedule Migration -The purpose of this project is to migrate existing schedules created with Spring -Cloud Data Flow 2.2.x and before to the new 2.3.0 format and stage the -SchedulerTaskLauncher. This is a Spring Boot application that utilizes Spring Batch to create a workflow +The purpose of this project is to migrate existing schedules that are using the SchedulerTaskLauncher +created with Spring Cloud Data Flow 2.3.0.RELEASE and 2.4.0.RELEASE to the new format. This new format allows the platform scheduler +to launch the task directly, returning it to the functionality that existing prior to 2.3.0.RELEASE. +This is a Spring Boot application that utilizes Spring Batch to create a workflow to migrate the schedules. This is a single step Spring Batch Job that does the following: * Read - Retrieves all schedules from scheduler. -* Process - Enriches the Schedule request with App and Deployer properties from the scheduler (or deployed app) -as well as data from the TaskDefinition. +* Process - Extracts the app properties from the SchedulerTaskLauncher instance and the associated platform job. * Write - Deploys artifacts if required and creates the new schedule. Once the migrated schedule has been created, the old schedule is destroyed. @@ -19,7 +19,7 @@ application and it will pick up where it left off. Thanks Spring Batch! :-) == Build the project === Services Required -In order to migrate the existing schedules to the 2.3.x Spring Cloud Data Flow Scheduling format, the Schedule Migrator requires the following services: +In order to migrate the existing schedules, the Schedule Migrator requires the following services: 1. Access to the database where Spring Cloud Data Flow stores its Task Definitions. For Cloud Foundry we need to bind the database to the Schedule Migrator. 2. Access to the Scheduling Agent. For Cloud Foundry we need to bind the PCF Scheduler to the Scheduler Migrator. @@ -34,7 +34,7 @@ In order to migrate the existing schedules to the 2.3.x Spring Cloud Data Flow S === Prerequisites -Spring Cloud Data Flow 2.3+ must be installed and running prior to launching the Cloud Scheduler Migration app. +Spring Cloud Data Flow 2.3+ must be installed and running prior to launching the migration app. === Launching your migration 1) Create a manifest.yml file in a work directory. @@ -57,9 +57,7 @@ applications: spring_cloud_deployer_cloudfoundry_skipSslValidation: spring_cloud_deployer_cloudfoundry_services: spring_cloud_scheduler_cloudfoundry_schedulerUrl: - spring_profiles_active: cf - spring.cloud.deployer.cloudfoundry.healthCheckTimeout: 300 - spring.cloud.deployer.cloudfoundry.apiTimeout: 300 + spring_profiles_active: cloudfoundry dataflowServerUri: spring_cloud_task_closecontextEnabled: true remoteRepositories_repo1_url: https://repo.spring.io/libs-snapshot @@ -87,12 +85,13 @@ From the `dataflow-migrate-schedules` directory launch the `runMigration.sh` usi chmod +x scripts/runMigration.sh ./scripts/runMigration.sh ``` + === Picking which schedules to migrate Use the `scheduleNamesToMigrate` property to specify a comma delimited list of the schedules you wish to migrate. If you don't specify this property all schedules will be migrated. For example: ``` -./scripts/runMigration.sh --scheduleNamesToMigrate=task-job3,task-job1 +./scripts/runMigration.sh --scheduleNamesToMigrate=schedulename-scdf-taskdefname,anotherschedulename-scdf-taskdefname ``` === Limiting one Scheduler to run at a time @@ -102,24 +101,88 @@ To enable this feature use the `runMigration.sh` script as follows. ./scripts/runMigration.sh --spring.cloud.task.single-instance-enabled=true ``` -=== Configuring Your Deployer Properties -The following deployer properties will affect all schedules to be migrated. -If a property is not set then the default will be used. - -==== Deployer properties to be applied to all migrated schedules: -* healthCheckTimeout -* apiTimeout -* statusTimeout -* stagingTimeout -* startupTimeout -* maximumConcurrentTasks -* javaOpts - -NOTE: Descriptions of these properties can be found : https://github.com/cppwfs/spring-cloud-dataflow-samples/blob/SCDF-121/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateProperties.java[here] - -=== Supported Databases -The database supported are enumerated https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-rdbms[here]. - === Previously Pushed Apps -The Cloud Schedule Migration app does not delete previously scheduled applications. -If these apps are no longer needed it is up to the user to delete them. +The Cloud Schedule Migration app deletes SchedulerTaskLauncher and its associated schedule. + +NOTE: If a single task definition has more than one schedule, the task definition +droplet created for first migrated schedule will be used for all scheduled tasks of that task definition. + +== Running The Project For Kubernetes + +=== Prerequisites + +* Spring Cloud Data Flow 2.3.+ must be installed and running prior to launching the migration app. +* Spring Cloud Data Flow must be using the `exec` entry point for scheduling apps. + +=== Launching your migration +1) Establish the following environment variables: +``` +export spring_datasource_url= +export spring_datasource_username= +export spring_datasource_password= +export spring_datasource_driverClassName= +export KUBERNETES_NAMESPACE=< The namespace for the scheduled apps > +export dbDriverClassName= +export dbUrl= +export dbUserName= +export dbPassword= +export dataflowUrl= +``` + +NOTE: There are two sets of datasource properties. This is because the `spring_datasource_*` +properties are used by the migration tool , while `db*` properties are used to +set the database connection information that is required by the migrated tasks. +The database connection URL used by the migration tool can be different than the +one used by the tasks. + +2) Configure environment to access the cluster where the schedules are located. + +NOTE: In some cases if `KUBECONFIG` has a list of kubeconfig files the application may not select the proper kubeconfig file. +In these cases set the `KUBECONFIG` so that it will use the proper config file in the `$HOME/.kube/` directory. + +3) Be sure to establish a port forward to the database that Spring Cloud Data Flow is using, +thus allowing the migration tool to gather information about task definitions. +For example if using mysql: `kubectl port-forward 3306:3306` + +4) To start the migration: +From the `dataflow-migrate-schedules` directory launch the `runKubernetesMigration.sh` using the commands below: +``` +chmod +x scripts/runKubernetesMigration.sh +./scripts/runKubernetesMigration.sh +``` + +=== Picking which schedules to migrate +By default the `runKubernetesMigration.sh` will migrate all schedules. +However if a specific set of schedules need to be migrated, then use the +`scheduleNamesToMigrate` property to specify a comma delimited list of +the schedules you wish to migrate. For example: +``` +./scripts/runKubernetesMigration.sh --scheduleNamesToMigrate=schedulename-scdf-taskdefname,anotherschedulename-scdf-taskdefname +``` + +=== Limiting one Scheduler to run at a time +If there is a requirement that only one `schedulemigrator` should run at a time you can set the `spring.cloud.task.single-instance-enabled` property to true. This will stop other executions of the schedulemigrator till the currently running instance completes. +To enable this feature use the `runMigration.sh` script as follows. +``` +./scripts/runKubernetesMigration.sh --spring.cloud.task.single-instance-enabled=true +``` + +== Configuring the Schedule Migration +The following properties configure how the scheduler migrator will migrate the schedules. + +* schedulerToken - The token (default `scdf-`) is used by SchedulerTaskLauncher as a delimiter +to separate the schedule name of each schedule into 2 components: `base schedule name` +and `task name`. This value will be used by the migration tool identify schedules to be migrated. +* taskLauncherPrefix - The prefix used by the SchedulerTaskLauncher to mark the properties for the launched apps. Default: `tasklauncher` +* scheduleNamesToMigrate - Comma delimited list of schedules to migrate. If empty then all schedules will be migrated. +* composedTaskRunnerRegisteredAppName - The registered application name for the composed task runner. Default: `composed-task-runner` +* dataflowUrl - The url of the Spring Cloud Data Flow Server that migrated composed task runners should execute task launch commands. + +=== Database Configuration for Kubernetes Migration +* dbUserName - The user name of the database that contains the task definitions for schedules to be migrated. +* dbPassword - The password of the database that contains the task definitions for schedules to be migrated. +* dbUrl - The url to the database that contains the task definitions for schedules to be migrated. +* dbDriverClassName - The driver class name to use for the database that contains the task definitions for schedules to be migrated. + +== Supported Databases +The database supported are enumerated https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-rdbms[here]. diff --git a/dataflow-migrate-schedules/pom.xml b/dataflow-migrate-schedules/pom.xml index 7af61b8..842bfc7 100644 --- a/dataflow-migrate-schedules/pom.xml +++ b/dataflow-migrate-schedules/pom.xml @@ -6,19 +6,19 @@ org.springframework.boot spring-boot-starter-parent - 2.2.1.RELEASE + 2.2.6.RELEASE io.spring migrateschedule 1.0.0.BUILD-SNAPSHOT - Schedule Migrator - Migrates SCDF Schedules to the 2.3 format + Schedule CD Migrator + Migrates SCDF Schedules so that SchedulerTaskLauncher is no longer required. 1.8 - 2.3.0.RC1 - 2.1.0.RELEASE + 2.5.0.BUILD-SNAPSHOT + 2.3.0.BUILD-SNAPSHOT @@ -81,6 +81,10 @@ org.mariadb.jdbc mariadb-java-client + + + org.springframework.boot + spring-boot-starter-jdbc org.springframework.cloud @@ -101,6 +105,11 @@ spring-cloud-dataflow-core ${spring-cloud-data-flow.version} + + org.springframework.cloud + spring-cloud-dataflow-server-core + ${spring-cloud-data-flow.version} + org.springframework.cloud spring-cloud-dataflow-registry @@ -119,6 +128,11 @@ org.springframework.integration spring-integration-jdbc + + org.springframework.cloud + spring-cloud-deployer-kubernetes + ${deployer.version} + diff --git a/dataflow-migrate-schedules/scripts/runKubernetesMigration.sh b/dataflow-migrate-schedules/scripts/runKubernetesMigration.sh new file mode 100755 index 0000000..7824423 --- /dev/null +++ b/dataflow-migrate-schedules/scripts/runKubernetesMigration.sh @@ -0,0 +1,4 @@ +export spring_profiles_active=kubernetes +export spring_autoconfigure_exclude=org.springframework.cloud.deployer.spi.cloudfoundry.CloudFoundryDeployerAutoConfiguration + +java -jar ./target/migrateschedule-1.0.0.BUILD-SNAPSHOT.jar $* diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/MigrateScheduleApplication.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/MigrateScheduleApplication.java index 6818f37..0484ee0 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/MigrateScheduleApplication.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/MigrateScheduleApplication.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -16,8 +16,11 @@ package io.spring.migrateschedule; +import org.springframework.boot.ApplicationArguments; +import org.springframework.boot.ApplicationRunner; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; @SpringBootApplication public class MigrateScheduleApplication { @@ -25,5 +28,4 @@ public class MigrateScheduleApplication { public static void main(String[] args) { SpringApplication.run(MigrateScheduleApplication.class, args); } - } diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerProcessor.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerProcessor.java index 2cb6e62..5b4c760 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerProcessor.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -28,8 +28,8 @@ import org.springframework.cloud.deployer.spi.scheduler.ScheduleInfo; /** * Enriches the {@link ConvertScheduleInfo} with information obtained from the platform. - * The new name for the schedule is established and the properties as well as commandline args - * so that the SchedulerTaskLauncher can process the entries. + * The name for the schedule and the properties for the new schedule is + * extracted from the ScheduleTaskLauncher properties and command line args.. * * @author Glenn Renfro */ @@ -48,7 +48,7 @@ public class SchedulerProcessor implements ItemProces @Override public ConvertScheduleInfo process(ConvertScheduleInfo scheduleInfo){ - if(scheduleInfo.getScheduleName().contains(migrateProperties.getSchedulerToken())) { + if(!scheduleInfo.getScheduleName().contains(migrateProperties.getSchedulerToken())) { throw new ScheduleProcessedException(scheduleInfo.getScheduleName()); } logger.info(String.format("Processing Schedule %s", scheduleInfo.getScheduleName())); diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerReader.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerReader.java index b673799..d628774 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerReader.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerReader.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerWriter.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerWriter.java index e480552..8548172 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerWriter.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/batch/SchedulerWriter.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -28,8 +28,7 @@ import org.springframework.cloud.deployer.spi.scheduler.ScheduleInfo; import org.springframework.cloud.deployer.spi.scheduler.Scheduler; /** - * Migrates the original schedule to the new scheduler format required for SCDF - * and stages the SchedulerTaskLauncher. + * Migrates the SchedulerTaskLauncher schedule to the new schedule. * * @author Glenn Renfro */ diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/BatchConfiguration.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/BatchConfiguration.java index 8c7754f..6e8ee77 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/BatchConfiguration.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/BatchConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -47,8 +47,6 @@ import org.springframework.context.annotation.Configuration; @EnableBatchProcessing @EnableTask public class BatchConfiguration { - - @Autowired public JobBuilderFactory jobBuilderFactory; diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/CFMigrateScheduleConfiguration.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/CFMigrateScheduleConfiguration.java index 9cc5780..44991d4 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/CFMigrateScheduleConfiguration.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/CFMigrateScheduleConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -18,16 +18,15 @@ package io.spring.migrateschedule.configuration; import io.pivotal.reactor.scheduler.ReactorSchedulerClient; import io.pivotal.scheduler.SchedulerClient; +import io.spring.migrateschedule.service.AppRegistrationRepository; import io.spring.migrateschedule.service.CFMigrateSchedulerService; import io.spring.migrateschedule.service.MigrateProperties; -import io.spring.migrateschedule.service.MigrateScheduleService; import io.spring.migrateschedule.service.TaskDefinitionRepository; import org.cloudfoundry.operations.CloudFoundryOperations; import org.cloudfoundry.reactor.ConnectionContext; import org.cloudfoundry.reactor.TokenProvider; import reactor.core.publisher.Mono; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.domain.EntityScan; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.cloud.deployer.resource.maven.MavenProperties; @@ -47,6 +46,7 @@ import org.springframework.context.annotation.Profile; @EntityScan({ "org.springframework.cloud.dataflow.core" }) +@Profile("cloudfoundry") public class CFMigrateScheduleConfiguration { @Bean @@ -69,9 +69,12 @@ public class CFMigrateScheduleConfiguration { public CFMigrateSchedulerService scheduleService(CloudFoundryOperations cloudFoundryOperations, SchedulerClient schedulerClient, CloudFoundryConnectionProperties properties, MigrateProperties migrateProperties, - TaskDefinitionRepository taskDefinitionRepository, MavenProperties mavenProperties) { + TaskDefinitionRepository taskDefinitionRepository, MavenProperties mavenProperties, + AppRegistrationRepository appRegistrationRepository, + TaskLauncher taskLauncher) { return new CFMigrateSchedulerService(cloudFoundryOperations, - schedulerClient, properties, migrateProperties, taskDefinitionRepository, mavenProperties); + schedulerClient, properties, migrateProperties, taskDefinitionRepository, + mavenProperties, appRegistrationRepository, taskLauncher); } @Bean diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/KubernetesMigrateScheduleConfiguration.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/KubernetesMigrateScheduleConfiguration.java new file mode 100644 index 0000000..b636f47 --- /dev/null +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/configuration/KubernetesMigrateScheduleConfiguration.java @@ -0,0 +1,68 @@ +/* + * Copyright 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 + * + * 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 io.spring.migrateschedule.configuration; + +import io.fabric8.kubernetes.client.KubernetesClient; +import io.spring.migrateschedule.service.AppRegistrationRepository; +import io.spring.migrateschedule.service.KubernetesMigrateSchedulerService; +import io.spring.migrateschedule.service.MigrateProperties; +import io.spring.migrateschedule.service.TaskDefinitionRepository; + +import org.springframework.boot.autoconfigure.domain.EntityScan; +import org.springframework.cloud.deployer.spi.kubernetes.KubernetesClientFactory; +import org.springframework.cloud.deployer.spi.kubernetes.KubernetesDeployerProperties; +import org.springframework.cloud.deployer.spi.kubernetes.KubernetesScheduler; +import org.springframework.cloud.deployer.spi.kubernetes.KubernetesSchedulerProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Primary; +import org.springframework.context.annotation.Profile; + +/** + * @author Glenn Renfro + */ +@Configuration +@EntityScan({ + "org.springframework.cloud.dataflow.core" +}) +@Profile("kubernetes") +public class KubernetesMigrateScheduleConfiguration { + + + @Bean + @Primary + public KubernetesSchedulerProperties kubernetesSchedulerProperties() { + return new KubernetesSchedulerProperties(); + } + + + @Bean + public KubernetesClient kubernetesClient(KubernetesDeployerProperties kubernetesDeployerProperties) { + return KubernetesClientFactory.getKubernetesClient(kubernetesDeployerProperties); + } + + @Bean + public KubernetesScheduler kubernetesScheduler(KubernetesClient kubernetesClient, KubernetesSchedulerProperties kubernetesSchedulerProperties) { + return new KubernetesScheduler( kubernetesClient, kubernetesSchedulerProperties); + } + + @Bean + public KubernetesMigrateSchedulerService scheduleService( TaskDefinitionRepository taskDefinitionRepository, AppRegistrationRepository appRegistrationRepository, KubernetesClient kubernetesClient, MigrateProperties migrateProperties) { + return new KubernetesMigrateSchedulerService(taskDefinitionRepository, appRegistrationRepository, kubernetesClient, migrateProperties); + } + +} diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/AbstractMigrateService.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/AbstractMigrateService.java index 7077f79..319d397 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/AbstractMigrateService.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/AbstractMigrateService.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -18,14 +18,15 @@ package io.spring.migrateschedule.service; import java.net.URI; import java.net.URISyntaxException; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.TreeMap; import java.util.stream.Collectors; +import org.springframework.cloud.dataflow.core.AppRegistration; +import org.springframework.cloud.dataflow.core.ApplicationType; import org.springframework.cloud.dataflow.core.TaskDefinition; +import org.springframework.cloud.dataflow.core.dsl.TaskNode; +import org.springframework.cloud.dataflow.core.dsl.TaskParser; import org.springframework.cloud.dataflow.registry.support.AppResourceCommon; import org.springframework.cloud.deployer.resource.maven.MavenProperties; import org.springframework.core.io.DefaultResourceLoader; @@ -40,24 +41,48 @@ import org.springframework.util.StringUtils; */ public abstract class AbstractMigrateService implements MigrateScheduleService { - private final static String DATA_FLOW_URI_KEY = "spring.cloud.dataflow.client.serverUri"; + public final static String COMMAND_ARGUMENT_PREFIX = "cmdarg."; - private final static String COMMAND_ARGUMENT_PREFIX = "cmdarg."; + public final static String APP_PREFIX = "app."; - protected final static String APP_PREFIX = "app."; - - protected final static String DEPLOYER_PREFIX = "deployer."; + public final static String DEPLOYER_PREFIX = "deployer."; protected MigrateProperties migrateProperties; - private TaskDefinitionRepository taskDefinitionRepository; + protected TaskDefinitionRepository taskDefinitionRepository; - private MavenProperties mavenProperties; + protected AppRegistrationRepository appRegistrationRepository; - public AbstractMigrateService(MigrateProperties migrateProperties, TaskDefinitionRepository taskDefinitionRepository, MavenProperties mavenProperties) { + protected MavenProperties mavenProperties; + + protected final String appArgPrefix; + + protected final String deployerArgPrefix; + + protected final String schedulerArgAppPrefix; + + protected final String commandArgPrefix; + + protected final int appArgPrefixLength; + + protected final int deployerArgPrefixLength; + + protected final int commandArgPrefixLength; + + public AbstractMigrateService(MigrateProperties migrateProperties, + TaskDefinitionRepository taskDefinitionRepository, + MavenProperties mavenProperties, AppRegistrationRepository appRegistrationRepository) { this.migrateProperties = migrateProperties; this.taskDefinitionRepository = taskDefinitionRepository; this.mavenProperties = mavenProperties; + this.appRegistrationRepository = appRegistrationRepository; + this.schedulerArgAppPrefix = String.format("--%s.", this.migrateProperties.getTaskLauncherPrefix()); + this.appArgPrefix = String.format("%s%s", this.schedulerArgAppPrefix, APP_PREFIX); + this.deployerArgPrefix = String.format("%s%s", this.schedulerArgAppPrefix, DEPLOYER_PREFIX); + this.appArgPrefixLength = this.appArgPrefix.length(); + this.deployerArgPrefixLength = this.deployerArgPrefix.length(); + this.commandArgPrefix = String.format("%s%s.", COMMAND_ARGUMENT_PREFIX, this.migrateProperties.getTaskLauncherPrefix()); + this.commandArgPrefixLength = this.commandArgPrefix.length(); } public TaskDefinition findTaskDefinitionByName(String taskDefinitionName) { @@ -76,7 +101,7 @@ public abstract class AbstractMigrateService implements MigrateScheduleService { * @param input the scheduler properties * @return scheduler properties for the task */ - protected static Map extractAndQualifySchedulerProperties(Map input) { + protected Map extractAndQualifySchedulerProperties(Map input) { final String prefix = "spring.cloud.scheduler."; Map result = new TreeMap<>(input).entrySet().stream() @@ -88,83 +113,75 @@ public abstract class AbstractMigrateService implements MigrateScheduleService { } /** - * Retrieve the resource for the SchedulerTaskLauncher and verify the URI. - * @return {@link Resource} for the SchedulerTaskLauncher. + * Retrieve the resource for the Scheduled task + * @return {@link Resource} for the scheduled task. */ - protected Resource getTaskLauncherResource() { - final URI url; + protected Resource getTaskLauncherResource(String resource) { try { - new URI(this.migrateProperties.getSchedulerTaskLauncherUrl()); //verify url + new URI(resource); //verify url } catch (URISyntaxException uriSyntaxException) { throw new IllegalArgumentException(uriSyntaxException); } AppResourceCommon appResourceCommon = new AppResourceCommon(this.mavenProperties, new DefaultResourceLoader()); - return appResourceCommon.getResource(this.migrateProperties.getSchedulerTaskLauncherUrl()); + return appResourceCommon.getResource(resource); } - /** - * Add the appropriate tags to the command line args so that the SchedulerTaskLauncher can - * extract them. - * @param args the command line args to be tagged. - * @return the tagged command line args. - */ - protected List tagCommandLineArgs(List args) { - List taggedArgs = new ArrayList<>(); + protected String extractDeployerKey(String arg, int deployerPrefixLength) { + return "spring.cloud.deployer." + extractAppKey(arg, deployerPrefixLength); + } - for(String arg : args) { - if(arg.contains("spring.cloud.task.name")) { - continue; - } - String updatedArg = arg; - if (!arg.startsWith(DATA_FLOW_URI_KEY) && !"--".concat(arg).startsWith(DATA_FLOW_URI_KEY)) { - updatedArg = COMMAND_ARGUMENT_PREFIX + - this.migrateProperties.getTaskLauncherPrefix() + "." + arg; - } - taggedArgs.add(updatedArg); + protected String extractAppKey(String arg, int prefixLength) { + int indexOfEqual = arg.indexOf("="); + arg = arg.substring(prefixLength, indexOfEqual); + int dotIndex = arg.indexOf("."); + String result = arg; + if (!arg.substring(0, dotIndex).equals("management")) { + result = arg.substring(dotIndex + 1); } - return taggedArgs; + return result; } - /** - * Add the appropriate tags to the command line args so that the SchedulerTaskLauncher can - * extract them. - * @param appName the name of the application to be associated with the property - * @param appProperties the properties to be tagged - * @param prefix the prefix to mark the property as to be used by the SchedulerTaskLauncher. - * @return the tagged command line args. - */ - protected Map tagProperties(String appName, Map appProperties, String prefix) { - Map taggedAppProperties = new HashMap<>(appProperties.size()); - for(String key : appProperties.keySet()) { - if(key.contains("spring.cloud.task.name")) { - continue; + protected String extractValue(String arg) { + int indexOfEqual = arg.indexOf("="); + return arg.substring(indexOfEqual + 1); + } + + protected ConvertScheduleInfo populateTaskDefinitionData(String appName, ConvertScheduleInfo scheduleInfo) { + TaskDefinition taskDefinition = this.taskDefinitionRepository.findByTaskName(appName); + + if (taskDefinition != null) { + TaskParser taskParser = new TaskParser(appName, taskDefinition.getDslText(), false, false); + String registeredAppName = taskDefinition.getRegisteredAppName(); + TaskNode taskNode = taskParser.parse(); + if (taskNode.isComposed()) { + registeredAppName = this.migrateProperties.getComposedTaskRunnerRegisteredAppName(); + scheduleInfo.setCtrDSL(taskNode.toExecutableDSL()); + scheduleInfo.setCTR(true); } - String updatedKey = key; - if (!key.startsWith(DATA_FLOW_URI_KEY)) { - if (StringUtils.hasText(appName)) { - updatedKey = this.migrateProperties.getTaskLauncherPrefix() + "." + - prefix + appName + "." + key; - } - else { - updatedKey = this.migrateProperties.getTaskLauncherPrefix() + "." + - prefix + key; - } - } - taggedAppProperties.put(updatedKey, appProperties.get(key)); + + AppRegistration appRegistration = this.appRegistrationRepository.findAppRegistrationByNameAndTypeAndDefaultVersionIsTrue(registeredAppName, ApplicationType.task); + + scheduleInfo.setTaskResource(getTaskLauncherResource(appRegistration.getUri().toString())); + scheduleInfo.setTaskDefinitionName(appName); } - return taggedAppProperties; + return scheduleInfo; } - /** - * Add the required SchedulerTaskLauncher properties. - * @param properties the map of properties in which to add the SchedulerTaskLauncher properties. - * @return updated properties. - */ - protected Map addSchedulerAppProps(Map properties) { - Map appProperties = new HashMap<>(properties); - appProperties.put("spring.cloud.dataflow.client.serverUri", this.migrateProperties.getDataflowServerUri()); - return appProperties; + protected ConvertScheduleInfo addDBInfoToScheduleInfo(ConvertScheduleInfo scheduleInfo) { + if(StringUtils.hasText(this.migrateProperties.getDbUrl())) { + scheduleInfo.getAppProperties().put("spring.datasource.url", this.migrateProperties.getDbUrl()); + } + if(StringUtils.hasText(this.migrateProperties.getDbDriverClassName())) { + scheduleInfo.getAppProperties().put("spring.datasource.driverClassName", this.migrateProperties.getDbDriverClassName()); + } + if(StringUtils.hasText(this.migrateProperties.getDbUserName())) { + scheduleInfo.getAppProperties().put("spring.datasource.username", this.migrateProperties.getDbUserName()); + } + if(StringUtils.hasText(this.migrateProperties.getDbPassword())) { + scheduleInfo.getAppProperties().put("spring.datasource.password", this.migrateProperties.getDbPassword()); + } + return scheduleInfo; } } diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/AppRegistrationRepository.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/AppRegistrationRepository.java new file mode 100644 index 0000000..c7052fc --- /dev/null +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/AppRegistrationRepository.java @@ -0,0 +1,35 @@ +/* + * Copyright 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 + * + * 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 io.spring.migrateschedule.service; + +import java.util.List; + +import org.springframework.cloud.dataflow.core.AppRegistration; +import org.springframework.cloud.dataflow.core.ApplicationType; +import org.springframework.data.keyvalue.repository.KeyValueRepository; +import org.springframework.transaction.annotation.Transactional; + +/** + * Repository interface for retrieving instances of the the {@link AppRegistration} class. + * @author Glenn Renfro + */ +@Transactional +public interface AppRegistrationRepository extends KeyValueRepository { + + AppRegistration findAppRegistrationByNameAndTypeAndDefaultVersionIsTrue(String name, ApplicationType type); + +} diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/CFMigrateSchedulerService.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/CFMigrateSchedulerService.java index 0495ceb..d904c73 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/CFMigrateSchedulerService.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/CFMigrateSchedulerService.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -41,7 +41,6 @@ import org.slf4j.LoggerFactory; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import org.springframework.cloud.dataflow.core.TaskDefinition; import org.springframework.cloud.deployer.resource.maven.MavenProperties; import org.springframework.cloud.deployer.spi.cloudfoundry.CloudFoundryConnectionProperties; import org.springframework.cloud.deployer.spi.core.AppDefinition; @@ -49,6 +48,7 @@ import org.springframework.cloud.deployer.spi.scheduler.ScheduleRequest; import org.springframework.cloud.deployer.spi.scheduler.Scheduler; import org.springframework.cloud.deployer.spi.scheduler.SchedulerException; import org.springframework.cloud.deployer.spi.scheduler.SchedulerPropertyKeys; +import org.springframework.cloud.deployer.spi.task.TaskLauncher; import org.springframework.util.StringUtils; /** @@ -61,6 +61,8 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { public static final String JAR_LAUNCHER = "org.springframework.boot.loader.JarLauncher"; + public static final String TASK_NAME_ARGUMENT_KEY = "--spring.cloud.scheduler.task.launcher.taskName="; + private static final int JAR_LAUNCHER_LENGTH = JAR_LAUNCHER.length(); private static final Logger logger = LoggerFactory.getLogger(CFMigrateSchedulerService.class); @@ -77,15 +79,20 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { private CloudFoundryConnectionProperties properties; + private TaskLauncher taskLauncher; + public CFMigrateSchedulerService(CloudFoundryOperations cloudFoundryOperations, SchedulerClient schedulerClient, CloudFoundryConnectionProperties properties, MigrateProperties migrateProperties, - TaskDefinitionRepository taskDefinitionRepository, MavenProperties mavenProperties) { - super(migrateProperties, taskDefinitionRepository, mavenProperties); + TaskDefinitionRepository taskDefinitionRepository, + MavenProperties mavenProperties, AppRegistrationRepository appRegistrationRepository, + TaskLauncher taskLauncher) { + super(migrateProperties, taskDefinitionRepository, mavenProperties, appRegistrationRepository); this.cloudFoundryOperations = cloudFoundryOperations; this.schedulerClient = schedulerClient; this.properties = properties; + this.taskLauncher = taskLauncher; } @Override @@ -93,9 +100,9 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { List result = new ArrayList<>(); int pageCount = getJobPageCount(); for (int i = PCF_PAGE_START_NUM; i <= pageCount; i++) { - logger.info(String.format("Reading Schedules Page %s of %s ", i, pageCount )); + logger.info(String.format("Reading Schedules Page %s of %s ", i, pageCount)); List scheduleInfoPage = getSchedules(i); - if(scheduleInfoPage == null) { + if (scheduleInfoPage == null) { throw new SchedulerException(SCHEDULER_SERVICE_ERROR_MESSAGE); } result.addAll(scheduleInfoPage); @@ -154,8 +161,8 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { private boolean isScheduleMigratable(String scheduleName) { boolean result; - if(migrateProperties.getScheduleNamesToMigrate().size() > 0) { - result = migrateProperties.getScheduleNamesToMigrate().contains(scheduleName); + if (this.migrateProperties.getScheduleNamesToMigrate().size() > 0) { + result = this.migrateProperties.getScheduleNamesToMigrate().contains(scheduleName); } else { result = true; @@ -174,7 +181,7 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { logger.info(String.format("Retrieving ApplicationManifest for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName())); ApplicationManifest applicationManifest = getApplicationManifest(scheduleInfo.getTaskDefinitionName()); - if(applicationManifest != null) { + if (applicationManifest != null) { addApplicationManifestPropsToConvertScheduleInfo(scheduleInfo, applicationManifest); } if (environment != null) { @@ -183,52 +190,64 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { } } logger.info(String.format("Tagging command line args for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName())); - List revisedCommandLineArgs = tagCommandLineArgs(scheduleInfo.getCommandLineArgs()); - revisedCommandLineArgs.add("--spring.cloud.scheduler.task.launcher.taskName=" + scheduleInfo.getTaskDefinitionName()); - scheduleInfo.setCommandLineArgs(revisedCommandLineArgs); - Map appProperties = null; + + String appName = getAppNameFromArgs(scheduleInfo); + + Map appProperties; try { logger.info(String.format("Extracting Spring App Properties for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName())); appProperties = getSpringAppProperties(scheduleInfo.getScheduleProperties()); - if(appProperties.size() > 0) { + if (appProperties.size() > 0) { scheduleInfo.setUseSpringApplicationJson(true); } } catch (Exception exception) { throw new IllegalArgumentException("Unable to parse SPRING_APPLICATION_JSON from USER VARIABLES", exception); } - logger.info(String.format("Retrieving Task Definition for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName())); - TaskDefinition taskDefinition = findTaskDefinitionByName(appProperties.get("spring.cloud.task.name")); - if (appProperties.size() > 0 && taskDefinition == null) { - throw new IllegalStateException(String.format("The schedule %s contains " + - "properties but the task definition %s does not exist and thus can't be migrated", - scheduleInfo.getScheduleName(), scheduleInfo.getTaskDefinitionName())); + populateTaskDefinitionData(appName, scheduleInfo); + + appProperties = cleanseAppProperties(scheduleInfo, appProperties); + if (scheduleInfo.isCTR()) { + appProperties.put("graph", scheduleInfo.getCtrDSL()); } + + logger.info(String.format("Retrieving Task Definition for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName())); + logger.info(String.format("Tagging app properties for application %s for schedule %s", scheduleInfo.getTaskDefinitionName(), scheduleInfo.getScheduleName())); - appProperties = tagProperties(taskDefinition.getRegisteredAppName(), appProperties, APP_PREFIX); - Map deployerProperties = tagProperties(taskDefinition.getRegisteredAppName(), - getDeployerProperties(scheduleInfo), DEPLOYER_PREFIX); - appProperties = addSchedulerAppProps(appProperties); + Map deployerProperties = getDeployerProperties(scheduleInfo); + appProperties.put("dataflow-server-uri", this.migrateProperties.getDataflowUrl()); appProperties.putAll(deployerProperties); scheduleInfo.setAppProperties(appProperties); + + scheduleInfo.setCommandLineArgs(cleanseCommandLineArgs(scheduleInfo, appName)); + addDBInfoToScheduleInfo(scheduleInfo); return scheduleInfo; } @Override public void migrateSchedule(Scheduler scheduler, ConvertScheduleInfo scheduleInfo) { - String scheduleName = scheduleInfo.getScheduleName() + "-" + getSchedulePrefixDefinitionName(scheduleInfo.getTaskDefinitionName()); - AppDefinition appDefinition = new AppDefinition(scheduleName, scheduleInfo.getAppProperties()); + logger.info("Migrating " + scheduleInfo); + if (scheduleInfo.getTaskResource() == null) { + logger.info(String.format("The task definition name for schedule %s does not exist in Data Flow", + scheduleInfo.getScheduleName())); + return; + } + String scheduleName = scheduleInfo.getScheduleName().substring(0, + scheduleInfo.getScheduleName().indexOf(this.migrateProperties.getSchedulerToken()) - 1); + AppDefinition appDefinition = new AppDefinition(scheduleInfo.getTaskDefinitionName(), scheduleInfo.getAppProperties()); logger.info(String.format("Extracting schedule specific properties for schedule %s", scheduleInfo.getScheduleName())); Map schedulerProperties = extractAndQualifySchedulerProperties(scheduleInfo.getScheduleProperties()); - ScheduleRequest scheduleRequest = new ScheduleRequest(appDefinition, schedulerProperties, new HashMap<>(), scheduleInfo.getCommandLineArgs(), scheduleName, getTaskLauncherResource()); + ScheduleRequest scheduleRequest = new ScheduleRequest(appDefinition, schedulerProperties, new HashMap<>(), scheduleInfo.getCommandLineArgs(), scheduleName, scheduleInfo.getTaskResource()); logger.info(String.format("Staging ScheduleTaskLauncher and scheduling %s", scheduleInfo.getScheduleName())); + scheduler.schedule(scheduleRequest); logger.info(String.format("Unscheduling original %s", scheduleInfo.getScheduleName())); - scheduler.unschedule(scheduleInfo.getScheduleName()); + this.taskLauncher.destroy(scheduleInfo.getScheduleName()); } /** * Retrieves the number of pages that can be returned when retrieving a list of jobs. + * * @return an int containing the number of available pages. */ private int getJobPageCount() { @@ -237,21 +256,18 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { .spaceId(requestSummary.getId()) .detailed(false).build()); }).block(); - if(response == null) { + if (response == null) { throw new SchedulerException(SCHEDULER_SERVICE_ERROR_MESSAGE); } return response.getPagination().getTotalPages(); } private Map getSpringAppProperties(Map properties) throws Exception { - Map result; - if(properties.containsKey("SPRING_APPLICATION_JSON")) { + Map result = new HashMap<>(); + if (properties.containsKey("SPRING_APPLICATION_JSON")) { result = new ObjectMapper() .readValue(properties.get("SPRING_APPLICATION_JSON"), Map.class); } - else { - result = new HashMap<>(); - } return result; } @@ -307,11 +323,12 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { .filter(application -> appId.equals(application.getId())) .singleOrEmpty(); } + private ApplicationManifest getApplicationManifest(String appName) { - return this.cloudFoundryOperations.applications() - .getApplicationManifest(GetApplicationManifestRequest - .builder().name(appName).build()) - .block(); + return this.cloudFoundryOperations.applications() + .getApplicationManifest(GetApplicationManifestRequest + .builder().name(appName).build()) + .block(); } private Map getDeployerProperties(ConvertScheduleInfo scheduleInfo) { @@ -320,10 +337,10 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { result.put(CLOUD_FOUNDRY_PREFIX + ".buildpack", scheduleInfo.getJavaBuildPack()); } if (scheduleInfo.getMemoryInMB() != null) { - result.put(CLOUD_FOUNDRY_PREFIX + ".memory", scheduleInfo.getMemoryInMB() + "m"); + result.put(CLOUD_FOUNDRY_PREFIX + ".memory", scheduleInfo.getMemoryInMB() + "m"); } if (scheduleInfo.getDiskInMB() != null) { - result.put(CLOUD_FOUNDRY_PREFIX + ".disk", scheduleInfo.getDiskInMB() + "m"); + result.put(CLOUD_FOUNDRY_PREFIX + ".disk", scheduleInfo.getDiskInMB() + "m"); } if (scheduleInfo.getApplicationHealthCheck() != null) { result.put(CLOUD_FOUNDRY_PREFIX + ".health-check", scheduleInfo.getApplicationHealthCheck().getValue()); @@ -344,45 +361,22 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { result.put(CLOUD_FOUNDRY_PREFIX + ".host", StringUtils.arrayToCommaDelimitedString(scheduleInfo.getHosts().toArray())); } - // Global deployer properties; - if (this.migrateProperties.getHealthCheckTimeout() != null) { - result.put(CLOUD_FOUNDRY_PREFIX + ".health-check-timeout", this.migrateProperties.getHealthCheckTimeout()); - } - if (this.migrateProperties.getJavaOptions() != null) { - result.put(CLOUD_FOUNDRY_PREFIX + ".javaOpts", this.migrateProperties.getJavaOptions()); - } - if (this.migrateProperties.getApiTimeout() != null) { - result.put(CLOUD_FOUNDRY_PREFIX + ".api-timeout", String.valueOf(this.migrateProperties.getApiTimeout())); - } - if (this.migrateProperties.getStatusTimeout() != null) { - result.put(CLOUD_FOUNDRY_PREFIX + ".status-timeout", String.valueOf(this.migrateProperties.getStatusTimeout())); - } - if (this.migrateProperties.getStagingTimeout() != null) { - result.put(CLOUD_FOUNDRY_PREFIX + ".staging-timeout", String.valueOf(this.migrateProperties.getStagingTimeout())); - } - if (this.migrateProperties.getStartupTimeout() != null) { - result.put(CLOUD_FOUNDRY_PREFIX + ".startup-timeout", String.valueOf(this.migrateProperties.getStartupTimeout())); - } - if (this.migrateProperties.getMaximumConcurrentTasks() != null) { - result.put(CLOUD_FOUNDRY_PREFIX + ".maximum-concurrent-tasks", String.valueOf(this.migrateProperties.getMaximumConcurrentTasks())); - } - return result; } - private ConvertScheduleInfo addApplicationManifestPropsToConvertScheduleInfo(ConvertScheduleInfo scheduleInfo, ApplicationManifest applicationManifest) { + private void addApplicationManifestPropsToConvertScheduleInfo(ConvertScheduleInfo scheduleInfo, ApplicationManifest applicationManifest) { scheduleInfo.setDiskInMB(applicationManifest.getDisk()); scheduleInfo.setMemoryInMB(applicationManifest.getMemory()); scheduleInfo.setApplicationHealthCheck(applicationManifest.getHealthCheckType()); scheduleInfo.setJavaBuildPack(applicationManifest.getBuildpack()); scheduleInfo.setHealthCheckEndPoint(applicationManifest.getHealthCheckHttpEndpoint()); - if(applicationManifest.getServices() != null && applicationManifest.getServices().size() > 0) { + if (applicationManifest.getServices() != null && applicationManifest.getServices().size() > 0) { scheduleInfo.setServices(applicationManifest.getServices()); } - if(applicationManifest.getDomains() != null && applicationManifest.getDomains().size() > 0) { + if (applicationManifest.getDomains() != null && applicationManifest.getDomains().size() > 0) { scheduleInfo.setDomains(applicationManifest.getDomains()); } - if(applicationManifest.getRoutes() != null && applicationManifest.getRoutes().size() > 0) { + if (applicationManifest.getRoutes() != null && applicationManifest.getRoutes().size() > 0) { List routes = new ArrayList<>(); for (Route route : applicationManifest.getRoutes()) { routes.add(route.getRoute()); @@ -390,13 +384,101 @@ public class CFMigrateSchedulerService extends AbstractMigrateService { scheduleInfo.setRoutes(routes); } if (applicationManifest.getHosts() != null && applicationManifest.getHosts().size() > 0) { - List hosts = new ArrayList<>(); - for (String host : applicationManifest.getHosts()) { - hosts.add(host); - } + List hosts = new ArrayList<>(applicationManifest.getHosts()); scheduleInfo.setHosts(hosts); } + } - return scheduleInfo; + private List cleanseCommandLineArgs(ConvertScheduleInfo scheduleInfo, String appName) { + List commandLineArgs = new ArrayList<>(); + + for (String arg : scheduleInfo.getCommandLineArgs()) { + String resultArg = (arg.startsWith("--")) ? "--" : ""; + if (arg.startsWith(COMMAND_ARGUMENT_PREFIX)) { + commandLineArgs.add(resultArg + extractAppKey(arg, this.appArgPrefixLength) + "=" + extractValue(arg)); + } + else if (arg.startsWith(this.deployerArgPrefix)) { + commandLineArgs.add(resultArg + extractDeployerKey(arg, this.appArgPrefixLength) + "=" + extractValue(arg)); + } + else if (arg.startsWith(this.commandArgPrefix)) { + commandLineArgs.add("--" + arg.substring(this.commandArgPrefixLength)); + } + else { + if (arg.startsWith("--spring.cloud.scheduler.task.launcher.taskName")) { + continue; + } + commandLineArgs.add(arg); + } + } + commandLineArgs.add("--spring.application.name=" + appName); + return commandLineArgs; + } + + private String cleanseAppPropertyKey(String key) { + String prefix = String.format("%s.%s", this.migrateProperties.getTaskLauncherPrefix(), APP_PREFIX); + key = key.substring(prefix.length()); + int dotIndex = key.indexOf("."); + String result = key; + if (!key.substring(0, dotIndex).equals("management")) { + result = key.substring(dotIndex + 1); + } + return result; + } + + private Map cleanseAppProperties(ConvertScheduleInfo scheduleInfo, Map properties) { + Map result = new HashMap<>(); + String ctrProperties = ""; + boolean isFirstCTREntry = true; + for (String key : properties.keySet()) { + if (scheduleInfo.isCTR()) { + String ctrPropertyCandidate = prepScheduleForCTR(scheduleInfo, key, properties.get(key)); + if (ctrPropertyCandidate != null) { + if (isFirstCTREntry) { + isFirstCTREntry = false; + ctrProperties += ctrPropertyCandidate; + } + else { + ctrProperties = ctrProperties + "," + ctrPropertyCandidate; + } + continue; + } + } + if (key.startsWith(this.migrateProperties.getTaskLauncherPrefix())) { + String cleansedKey = cleanseAppPropertyKey(key); + result.put(cleansedKey, properties.get(key)); + } + else { + result.put(key, properties.get(key)); + } + } + if (!ctrProperties.isEmpty()) { + result.put("composed-task-properties", ctrProperties); + } + return result; + } + + private String prepScheduleForCTR(ConvertScheduleInfo scheduleInfo, String key, String value) { + String ctrProperty = null; + String prefix = String.format("%s.%s%s.", this.migrateProperties.getTaskLauncherPrefix(), APP_PREFIX, scheduleInfo.getTaskDefinitionName()); + if (key.startsWith(prefix)) { + String newKey = key.substring(prefix.length()); + String appName = newKey.substring(0, newKey.indexOf(".")); + ctrProperty = String.format("%s%s-%s.%s%s=%s", APP_PREFIX, + scheduleInfo.getTaskDefinitionName(), appName, APP_PREFIX, + newKey, value); + } + return ctrProperty; + } + + private String getAppNameFromArgs(ConvertScheduleInfo scheduleInfo) { + String appName = null; + for (String command : scheduleInfo.getCommandLineArgs()) { + if (command.startsWith(TASK_NAME_ARGUMENT_KEY)) { + int appNameIndex = command.indexOf(TASK_NAME_ARGUMENT_KEY); + appName = command.substring(appNameIndex + TASK_NAME_ARGUMENT_KEY.length()); + break; + } + } + return appName; } } diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/ConvertScheduleInfo.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/ConvertScheduleInfo.java index b1c13a2..c866539 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/ConvertScheduleInfo.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/ConvertScheduleInfo.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -24,6 +24,7 @@ import java.util.Map; import org.cloudfoundry.operations.applications.ApplicationHealthCheck; import org.springframework.cloud.deployer.spi.scheduler.ScheduleInfo; +import org.springframework.core.io.Resource; /** * A child implementation of {@link ScheduleInfo} that adds additional attributes @@ -39,6 +40,8 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ private Map appProperties = new HashMap<>(); + private Map deployerProperties = new HashMap(); + private Integer diskInMB; private Integer memoryInMB; @@ -49,8 +52,6 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ private String healthCheckEndPoint; - private Integer maximumConcurrentTasks = 20; - private boolean useSpringApplicationJson; private List services; @@ -61,6 +62,12 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ private List hosts; + private Resource taskResource; + + private boolean isCTR; + + private String ctrDSL; + public List getCommandLineArgs() { return commandLineArgs; } @@ -70,7 +77,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ } public String getRegisteredAppName() { - return registeredAppName; + return this.registeredAppName; } public void setRegisteredAppName(String registeredAppName) { @@ -94,7 +101,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ } public Integer getDiskInMB() { - return diskInMB; + return this.diskInMB; } public void setDiskInMB(Integer diskInMB) { @@ -102,7 +109,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ } public Integer getMemoryInMB() { - return memoryInMB; + return this.memoryInMB; } public void setMemoryInMB(Integer memoryInMB) { @@ -110,7 +117,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ } public String getJavaBuildPack() { - return javaBuildPack; + return this.javaBuildPack; } public void setJavaBuildPack(String javaBuildPack) { @@ -118,7 +125,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ } public ApplicationHealthCheck getApplicationHealthCheck() { - return applicationHealthCheck; + return this.applicationHealthCheck; } public void setApplicationHealthCheck(ApplicationHealthCheck applicationHealthCheck) { @@ -126,7 +133,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ } public boolean isUseSpringApplicationJson() { - return useSpringApplicationJson; + return this.useSpringApplicationJson; } public void setUseSpringApplicationJson(boolean useSpringApplicationJson) { @@ -134,23 +141,15 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ } public String getHealthCheckEndPoint() { - return healthCheckEndPoint; + return this.healthCheckEndPoint; } public void setHealthCheckEndPoint(String healthCheckEndPoint) { this.healthCheckEndPoint = healthCheckEndPoint; } - public Integer getMaximumConcurrentTasks() { - return maximumConcurrentTasks; - } - - public void setMaximumConcurrentTasks(Integer maximumConcurrentTasks) { - this.maximumConcurrentTasks = maximumConcurrentTasks; - } - public List getServices() { - return services; + return this.services; } public void setServices(List services) { @@ -158,7 +157,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ } public List getDomains() { - return domains; + return this.domains; } public void setDomains(List domains) { @@ -166,7 +165,7 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ } public List getRoutes() { - return routes; + return this.routes; } public void setRoutes(List routes) { @@ -174,10 +173,42 @@ public class ConvertScheduleInfo extends ScheduleInfo implements Comparable{ } public List getHosts() { - return hosts; + return this.hosts; } public void setHosts(List hosts) { this.hosts = hosts; } + + public Resource getTaskResource() { + return this.taskResource; + } + + public void setTaskResource(Resource taskResource) { + this.taskResource = taskResource; + } + + public boolean isCTR() { + return this.isCTR; + } + + public void setCTR(boolean CTR) { + isCTR = CTR; + } + + public String getCtrDSL() { + return this.ctrDSL; + } + + public void setCtrDSL(String ctrDSL) { + this.ctrDSL = ctrDSL; + } + + public Map getDeployerProperties() { + return deployerProperties; + } + + public void setDeployerProperties(Map deployerProperties) { + this.deployerProperties = deployerProperties; + } } diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/KubernetesMigrateSchedulerService.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/KubernetesMigrateSchedulerService.java new file mode 100644 index 0000000..e3be0fb --- /dev/null +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/KubernetesMigrateSchedulerService.java @@ -0,0 +1,190 @@ +/* + * Copyright 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 + * + * 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 io.spring.migrateschedule.service; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.batch.CronJob; +import io.fabric8.kubernetes.api.model.batch.CronJobList; +import io.fabric8.kubernetes.client.KubernetesClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.cloud.dataflow.core.TaskDefinition; +import org.springframework.cloud.deployer.resource.maven.MavenProperties; +import org.springframework.cloud.deployer.spi.core.AppDefinition; +import org.springframework.cloud.deployer.spi.scheduler.ScheduleRequest; +import org.springframework.cloud.deployer.spi.scheduler.Scheduler; + + +/** + * Services required to migrate schedules to the 2.5.0 format in Kubernetes + * and stage the SchedulerTaskLauncher. + * + * @author Glenn Renfro + */ +public class KubernetesMigrateSchedulerService extends AbstractMigrateService { + + private static final Logger logger = LoggerFactory.getLogger(KubernetesMigrateSchedulerService.class); + + private KubernetesClient kubernetesClient; + + public KubernetesMigrateSchedulerService(TaskDefinitionRepository taskDefinitionRepository, + AppRegistrationRepository appRegistrationRepository, + KubernetesClient kubernetesClient, MigrateProperties migrateProperties) { + super(migrateProperties, taskDefinitionRepository, new MavenProperties(), appRegistrationRepository); + this.kubernetesClient = kubernetesClient; + this.migrateProperties = migrateProperties; + } + + /** + * Retrieves all the cronjobs from the kubernetes instance and extracts + * the schedule name, image and command line args from each cronjob and + * creates a {@link ConvertScheduleInfo}. + * @return a {@link List} of {@link ConvertScheduleInfo}s. + */ + @Override + public List scheduleInfoList() { + List result = new ArrayList<>(); + CronJobList cronJobs = this.kubernetesClient.batch().cronjobs().list(); + for (CronJob cronjob : cronJobs.getItems()) { + Container container = cronjob.getSpec().getJobTemplate().getSpec().getTemplate().getSpec().getContainers().get(0); + ConvertScheduleInfo scheduleInfo = new ConvertScheduleInfo(); + scheduleInfo.setScheduleName(cronjob.getMetadata().getName()); + if (!isScheduleMigratable(scheduleInfo.getScheduleName())) { + continue; + } + String appName = scheduleInfo.getScheduleName().substring( + scheduleInfo.getScheduleName().indexOf( + this.migrateProperties.getSchedulerToken()) + this.migrateProperties.getSchedulerToken().length()); + + scheduleInfo = populateTaskDefinitionData(appName, scheduleInfo); + + scheduleInfo.setCommandLineArgs(container.getArgs()); + scheduleInfo.setScheduleProperties(new HashMap<>()); + scheduleInfo.getScheduleProperties().put("spring.cloud.scheduler.cron.expression", cronjob.getSpec().getSchedule()); + result.add(scheduleInfo); + } + Collections.sort(result); + return result; + } + + private boolean isScheduleMigratable(String scheduleName) { + boolean result; + if (this.migrateProperties.getScheduleNamesToMigrate().size() > 0) { + result = this.migrateProperties.getScheduleNamesToMigrate().contains(scheduleName); + } + else { + result = true; + } + return result; + } + + /** + * Extracts the command line args from convertScheduleInfo that are tagged as properties + * and inserts them in the {@link ConvertScheduleInfo} app and deployer properties. + * Also adds the properties required for the launched task to interact with dataflow via the db. + * @param scheduleInfo A {@link ConvertScheduleInfo} to be enriched. + * @return the enriched {@link ConvertScheduleInfo}. + */ + @Override + public ConvertScheduleInfo enrichScheduleMetadata(ConvertScheduleInfo scheduleInfo) { + List commandLineArgs = new ArrayList<>(); + for (String arg : scheduleInfo.getCommandLineArgs()) { + if (arg.startsWith(this.appArgPrefix)) { + scheduleInfo.getAppProperties().put( + extractAppKey(arg, this.appArgPrefixLength), + extractValue(arg)); + } + else if (arg.startsWith(this.deployerArgPrefix)) { + scheduleInfo.getDeployerProperties().put( + extractDeployerKey(arg, this.deployerArgPrefixLength), + extractValue(arg)); + } + else if (arg.startsWith(this.commandArgPrefix)) { + commandLineArgs.add(arg.substring(this.commandArgPrefixLength)); + } + else { + if (arg.startsWith("--spring.cloud.scheduler.task.launcher.taskName")) { + continue; + } + commandLineArgs.add(arg); + } + } + if (scheduleInfo.isCTR()) { + scheduleInfo.getAppProperties().put("graph", scheduleInfo.getCtrDSL()); + } + + if (!scheduleInfo.getAppProperties().containsKey("spring.cloud.task.name")) { + scheduleInfo.getAppProperties().put("spring.cloud.task.name", scheduleInfo.getTaskDefinitionName()); + } + + addDBInfoToScheduleInfo(scheduleInfo); + + if (scheduleInfo.isCTR()) { + commandLineArgs.add("--dataflow-server-uri=" + this.migrateProperties.getDataflowUrl()); + } + scheduleInfo.setCommandLineArgs(commandLineArgs); + + return scheduleInfo; + } + + /** + * Creates a new schedule that launches the task and deletes the SchedulerTaskLauncher schedule. + * @param scheduler the deployer scheduler to build the new schedule. + * @param scheduleInfo the schedule info containing the existing schedule. + */ + @Override + public void migrateSchedule(Scheduler scheduler, ConvertScheduleInfo scheduleInfo) { + logger.info("Migrating " + scheduleInfo); + if(scheduleInfo.getTaskResource() == null) { + logger.info(String.format("The task definition name for schedule %s does not exist in Data Flow", + scheduleInfo.getScheduleName())); + return; + } + int scheduleTokenIndex = scheduleInfo.getScheduleName().indexOf( + this.migrateProperties.getSchedulerToken()) - 1; + String scheduleName = scheduleInfo.getScheduleName().substring(0, scheduleTokenIndex); + String appName = scheduleInfo.getScheduleName().substring( + scheduleInfo.getScheduleName().indexOf( + this.migrateProperties.getSchedulerToken()) + this.migrateProperties.getSchedulerToken().length()); + AppDefinition appDefinition = new AppDefinition(appName, scheduleInfo.getAppProperties()); + logger.info(String.format("Extracting schedule specific properties for schedule %s", scheduleInfo.getScheduleName())); + Map schedulerProperties = extractAndQualifySchedulerProperties(scheduleInfo.getScheduleProperties()); + ScheduleRequest scheduleRequest = new ScheduleRequest(appDefinition, schedulerProperties, scheduleInfo.getDeployerProperties(), scheduleInfo.getCommandLineArgs(), scheduleName, scheduleInfo.getTaskResource()); + logger.info(String.format("Staging Schedule %s", scheduleName)); + scheduler.schedule(scheduleRequest); + logger.info(String.format("Unscheduling original %s", scheduleInfo.getScheduleName())); + scheduler.unschedule(scheduleInfo.getScheduleName()); + } + + protected Map extractAndQualifySchedulerProperties(Map scheduleInfoProperties) { + scheduleInfoProperties.put("spring.cloud.scheduler.kubernetes.taskServiceAccountName", this.migrateProperties.getTaskServiceAccountName()); + //get properties from the spec + return scheduleInfoProperties; + } + + @Override + public TaskDefinition findTaskDefinitionByName(String taskDefinitionName) { + return null; + } +} diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateProperties.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateProperties.java index 8469169..1348749 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateProperties.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -25,8 +25,6 @@ import java.util.List; * @author Glenn Renfro */ public class MigrateProperties { - private String schedulerTaskLauncherUrl = "maven://org.springframework.cloud:spring-cloud-dataflow-scheduler-task-launcher:2.3.0.BUILD-SNAPSHOT"; - /** * The token for the updated schedules. */ @@ -37,61 +35,47 @@ public class MigrateProperties { */ private String taskLauncherPrefix = "tasklauncher"; - private String dataflowServerUri = "http://localhost:9393"; - - /** - * The global Java Options required for the applications to be launched by the schedulerTaskLauncher. - */ - private String javaOptions; - - /** - * The global timeout to be assigned to applications to be launched by the schedulerTaskLauncher. - */ - private String healthCheckTimeout; - - /** - * The global api timeout to be assigned to applications to be launched by scheduleTaskLauncher. - */ - private Long apiTimeout; - - /** - * Timeout for status API operations in milliseconds to be assigned to applications to be launched by scheduleTaskLauncher - */ - private Long statusTimeout; - - /** - * If set, the global override the timeout allocated for staging an app launched by the schedulefTaskLauncher. - */ - private Long stagingTimeout; - - /** - * If set, the global override the timeout allocated for starting an app launched by scheduleTaskLauncher. - */ - private Long startupTimeout; - - /** - * If set, the global override for the maximum number of concurrently running tasks. - */ - private Integer maximumConcurrentTasks; - - /** - * The number of seconds to wait for a schedule to complete. - * This excludes the time it takes to stage the application on Cloud Foundry. - */ - private int scheduleTimeoutInSeconds = 30; - /** * Comma delimited list of schedules to migrate. If empty then all schedules will be migrated. */ private List scheduleNamesToMigrate = new ArrayList<>(); - public String getSchedulerTaskLauncherUrl() { - return schedulerTaskLauncherUrl; - } - public void setSchedulerTaskLauncherUrl(String schedulerTaskLauncherUrl) { - this.schedulerTaskLauncherUrl = schedulerTaskLauncherUrl; - } + /** + * The registered application name for the composed task runner. + * Update this if the dataflow server use has a different composed task runner app name than the default. + */ + private String composedTaskRunnerRegisteredAppName = "composed-task-runner"; + + /** + * The user name of the database that the migrated task will use. + */ + private String dbUserName; + + /** + * The password of the database that the migrated task will use. + */ + private String dbPassword; + + /** + * The url to the database that the migrated task will use. + */ + private String dbUrl; + + /** + * The driver class name to use for the database that the migrated task will use. + */ + private String dbDriverClassName; + + /** + * The url of the Spring Cloud Data Flow Server that migrated composed task runners should execute task launch commands. + */ + private String dataflowUrl="http://localhost:9393"; + + /** + * Establish the name of the service account for the schedule. + */ + private String taskServiceAccountName = "default"; public String getSchedulerToken() { return schedulerToken; @@ -109,78 +93,6 @@ public class MigrateProperties { this.taskLauncherPrefix = taskLauncherPrefix; } - public String getDataflowServerUri() { - return dataflowServerUri; - } - - public void setDataflowServerUri(String dataflowServerUri) { - this.dataflowServerUri = dataflowServerUri; - } - - public int getScheduleTimeoutInSeconds() { - return scheduleTimeoutInSeconds; - } - - public void setScheduleTimeoutInSeconds(int scheduleTimeoutInSeconds) { - this.scheduleTimeoutInSeconds = scheduleTimeoutInSeconds; - } - - public String getJavaOptions() { - return javaOptions; - } - - public void setJavaOptions(String javaOptions) { - this.javaOptions = javaOptions; - } - - public String getHealthCheckTimeout() { - return healthCheckTimeout; - } - - public void setHealthCheckTimeout(String healthCheckTimeout) { - this.healthCheckTimeout = healthCheckTimeout; - } - - public Long getApiTimeout() { - return apiTimeout; - } - - public void setApiTimeout(Long apiTimeout) { - this.apiTimeout = apiTimeout; - } - - public Long getStatusTimeout() { - return statusTimeout; - } - - public void setStatusTimeout(Long statusTimeout) { - this.statusTimeout = statusTimeout; - } - - public Long getStagingTimeout() { - return stagingTimeout; - } - - public void setStagingTimeout(Long stagingTimeout) { - this.stagingTimeout = stagingTimeout; - } - - public Long getStartupTimeout() { - return startupTimeout; - } - - public void setStartupTimeout(Long startupTimeout) { - this.startupTimeout = startupTimeout; - } - - public Integer getMaximumConcurrentTasks() { - return maximumConcurrentTasks; - } - - public void setMaximumConcurrentTasks(Integer maximumConcurrentTasks) { - this.maximumConcurrentTasks = maximumConcurrentTasks; - } - public List getScheduleNamesToMigrate() { return scheduleNamesToMigrate; } @@ -188,4 +100,60 @@ public class MigrateProperties { public void setScheduleNamesToMigrate(List scheduleNamesToMigrate) { this.scheduleNamesToMigrate = scheduleNamesToMigrate; } + + public String getDbUserName() { + return dbUserName; + } + + public void setDbUserName(String dbUserName) { + this.dbUserName = dbUserName; + } + + public String getDbPassword() { + return dbPassword; + } + + public void setDbPassword(String dbPassword) { + this.dbPassword = dbPassword; + } + + public String getDbUrl() { + return dbUrl; + } + + public void setDbUrl(String dbUrl) { + this.dbUrl = dbUrl; + } + + public String getDbDriverClassName() { + return dbDriverClassName; + } + + public void setDbDriverClassName(String dbDriverClassName) { + this.dbDriverClassName = dbDriverClassName; + } + + public String getComposedTaskRunnerRegisteredAppName() { + return composedTaskRunnerRegisteredAppName; + } + + public void setComposedTaskRunnerRegisteredAppName(String composedTaskRunnerRegisteredAppName) { + this.composedTaskRunnerRegisteredAppName = composedTaskRunnerRegisteredAppName; + } + + public String getDataflowUrl() { + return dataflowUrl; + } + + public void setDataflowUrl(String dataflowUrl) { + this.dataflowUrl = dataflowUrl; + } + + public String getTaskServiceAccountName() { + return taskServiceAccountName; + } + + public void setTaskServiceAccountName(String taskServiceAccountName) { + this.taskServiceAccountName = taskServiceAccountName; + } } diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateScheduleService.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateScheduleService.java index 1da6923..8bf20d2 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateScheduleService.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/MigrateScheduleService.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/ScheduleProcessedException.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/ScheduleProcessedException.java index 669a747..8d66d0a 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/ScheduleProcessedException.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/ScheduleProcessedException.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/SchedulerSkipPolicy.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/SchedulerSkipPolicy.java index 3905566..504bf57 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/SchedulerSkipPolicy.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/SchedulerSkipPolicy.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. diff --git a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/TaskDefinitionRepository.java b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/TaskDefinitionRepository.java index 878951a..185fd33 100644 --- a/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/TaskDefinitionRepository.java +++ b/dataflow-migrate-schedules/src/main/java/io/spring/migrateschedule/service/TaskDefinitionRepository.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. diff --git a/dataflow-migrate-schedules/src/main/resources/application.properties b/dataflow-migrate-schedules/src/main/resources/application.properties index 398a1dc..5c34e4e 100644 --- a/dataflow-migrate-schedules/src/main/resources/application.properties +++ b/dataflow-migrate-schedules/src/main/resources/application.properties @@ -1 +1,2 @@ logging.level.org.springframework.cloud.task=debug +spring.main.web-application-type=none \ No newline at end of file diff --git a/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/AbstractApplications.java b/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/AbstractApplications.java index 10d0517..3f50b1d 100644 --- a/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/AbstractApplications.java +++ b/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/AbstractApplications.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -18,7 +18,6 @@ package io.spring.migrateschedule; import org.cloudfoundry.doppler.LogMessage; import org.cloudfoundry.operations.applications.ApplicationEvent; -import org.cloudfoundry.operations.applications.ApplicationManifest; import org.cloudfoundry.operations.applications.ApplicationSshEnabledRequest; import org.cloudfoundry.operations.applications.ApplicationSummary; import org.cloudfoundry.operations.applications.Applications; @@ -27,7 +26,6 @@ import org.cloudfoundry.operations.applications.DeleteApplicationRequest; import org.cloudfoundry.operations.applications.DisableApplicationSshRequest; import org.cloudfoundry.operations.applications.EnableApplicationSshRequest; import org.cloudfoundry.operations.applications.GetApplicationEventsRequest; -import org.cloudfoundry.operations.applications.GetApplicationManifestRequest; import org.cloudfoundry.operations.applications.ListApplicationTasksRequest; import org.cloudfoundry.operations.applications.LogsRequest; import org.cloudfoundry.operations.applications.PushApplicationManifestRequest; diff --git a/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/BatchIntegrationTests.java b/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/BatchIntegrationTests.java index 3c4ac1c..11aee9a 100644 --- a/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/BatchIntegrationTests.java +++ b/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/BatchIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -17,24 +17,20 @@ package io.spring.migrateschedule; import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; import java.util.HashMap; import java.util.List; -import io.spring.migrateschedule.service.ConvertScheduleInfo; import io.spring.migrateschedule.configuration.BatchConfiguration; +import io.spring.migrateschedule.service.AppRegistrationRepository; +import io.spring.migrateschedule.service.ConvertScheduleInfo; import io.spring.migrateschedule.service.MigrateScheduleService; import io.spring.migrateschedule.service.TaskDefinitionRepository; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.ArgumentCaptor; -import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; -import org.springframework.batch.core.JobParameters; -import org.springframework.batch.core.JobParametersBuilder; import org.springframework.batch.core.explore.JobExplorer; import org.springframework.batch.test.JobLauncherTestUtils; import org.springframework.batch.test.context.SpringBatchTest; @@ -70,7 +66,7 @@ import static org.mockito.Mockito.when; @DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) public class BatchIntegrationTests { - public static final String DEFAULT_SCHEDULE_NAME = "defaultScheduleName"; + public static final String DEFAULT_SCHEDULE_NAME = "defaultScheduleName-scdf-myapp"; public static final String DEFAULT_TASK_DEFINITION_NAME = "defaultTaskDefinitionName"; public static final String DEFAULT_APP_NAME = "defaultAppName"; @@ -86,6 +82,9 @@ public class BatchIntegrationTests { @MockBean private TaskDefinitionRepository taskDefinitionRepository; + @MockBean + private AppRegistrationRepository appRegistrationRepository; + @MockBean private Scheduler scheduler; diff --git a/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/CFGetSchedulesTests.java b/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/CFGetSchedulesTests.java index 17b5ec7..b895ec3 100644 --- a/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/CFGetSchedulesTests.java +++ b/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/CFGetSchedulesTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -42,6 +42,7 @@ import io.pivotal.scheduler.v1.jobs.ListJobsRequest; import io.pivotal.scheduler.v1.jobs.ListJobsResponse; import io.pivotal.scheduler.v1.jobs.ScheduleJobRequest; import io.pivotal.scheduler.v1.jobs.ScheduleJobResponse; +import io.spring.migrateschedule.service.AppRegistrationRepository; import io.spring.migrateschedule.service.CFMigrateSchedulerService; import io.spring.migrateschedule.service.ConvertScheduleInfo; import io.spring.migrateschedule.service.MigrateProperties; @@ -74,7 +75,7 @@ import reactor.core.publisher.Mono; import org.springframework.cloud.deployer.resource.maven.MavenProperties; import org.springframework.cloud.deployer.spi.cloudfoundry.CloudFoundryConnectionProperties; -import org.springframework.cloud.deployer.spi.scheduler.Scheduler; +import org.springframework.cloud.deployer.spi.task.TaskLauncher; import static org.assertj.core.api.Assertions.assertThat; @@ -92,7 +93,8 @@ public class CFGetSchedulesTests { private MigrateProperties migrateProperties; private TaskDefinitionRepository taskDefinitionRepository; private SchedulerClient schedulerClient; - private Scheduler scheduler; + private AppRegistrationRepository appRegistrationRepository; + private TaskLauncher taskLauncher; @BeforeEach public void setup() { @@ -102,11 +104,14 @@ public class CFGetSchedulesTests { this.cloudFoundryConnectionProperties.setSpace(DEFAULT_SPACE); this.migrateProperties = new MigrateProperties(); this.taskDefinitionRepository = Mockito.mock(TaskDefinitionRepository.class); - this.scheduler = Mockito.mock(Scheduler.class); + this.appRegistrationRepository = Mockito.mock(AppRegistrationRepository.class); + this.taskLauncher = Mockito.mock(TaskLauncher.class); + this.cfConvertSchedulerService = new CFMigrateSchedulerService(this.cloudFoundryOperations, this.schedulerClient, this.cloudFoundryConnectionProperties, this.migrateProperties, - this.taskDefinitionRepository, new MavenProperties()) ; + this.taskDefinitionRepository, new MavenProperties(), + this.appRegistrationRepository, this.taskLauncher) ; } diff --git a/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/CFMigrateScheduleConfigurationTests.java b/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/CFMigrateScheduleConfigurationTests.java index 248cd49..6cde31c 100644 --- a/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/CFMigrateScheduleConfigurationTests.java +++ b/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/CFMigrateScheduleConfigurationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 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. @@ -17,12 +17,15 @@ package io.spring.migrateschedule; +import java.net.URI; +import java.net.URISyntaxException; import java.util.HashMap; import io.pivotal.scheduler.SchedulerClient; +import io.spring.migrateschedule.service.AppRegistrationRepository; +import io.spring.migrateschedule.service.CFMigrateSchedulerService; import io.spring.migrateschedule.service.ConvertScheduleInfo; import io.spring.migrateschedule.service.MigrateProperties; -import io.spring.migrateschedule.service.CFMigrateSchedulerService; import io.spring.migrateschedule.service.TaskDefinitionRepository; import org.cloudfoundry.operations.CloudFoundryOperations; import org.cloudfoundry.operations.applications.ApplicationDetail; @@ -39,24 +42,27 @@ import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import reactor.core.publisher.Mono; +import org.springframework.cloud.dataflow.core.AppRegistration; import org.springframework.cloud.dataflow.core.TaskDefinition; import org.springframework.cloud.deployer.resource.maven.MavenProperties; import org.springframework.cloud.deployer.spi.cloudfoundry.CloudFoundryConnectionProperties; import org.springframework.cloud.deployer.spi.scheduler.ScheduleRequest; import org.springframework.cloud.deployer.spi.scheduler.Scheduler; +import org.springframework.cloud.deployer.spi.task.TaskLauncher; import static org.assertj.core.api.AssertionsForClassTypes.assertThat; -import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; public class CFMigrateScheduleConfigurationTests { - public static final String DEFAULT_SCHEDULE_NAME = "defaultScheduleName"; public static final String DEFAULT_TASK_DEFINITION_NAME = "defaultTaskDefinitionName"; + public static final String DEFAULT_SCHEDULE_NAME = "defaultScheduleName-scdf-" + DEFAULT_TASK_DEFINITION_NAME; public static final String DEFAULT_APP_NAME = "defaultAppName"; - public static final String DEFAULT_CMD_ARG = "defaultCmd=WOW"; + public static final String DEFAULT_CMD_ARG = "defaultCmd=MUCHWOW"; + public static final String TASK_LAUNCHER_TASK_NAME_ARG = "--spring.cloud.scheduler.task.launcher.taskName="+ DEFAULT_TASK_DEFINITION_NAME; public static final String DEFAULT_BUILD_PACK = "defaultBuildPack"; + public static final String DEFAULT_DATA_FLOW_URI = "http://localhost:9393"; private CFMigrateSchedulerService cfConvertSchedulerService; private CloudFoundryOperations cloudFoundryOperations; @@ -65,6 +71,8 @@ public class CFMigrateScheduleConfigurationTests { private TaskDefinitionRepository taskDefinitionRepository; private SchedulerClient schedulerClient; private Scheduler scheduler; + private AppRegistrationRepository appRegistrationRepository; + private TaskLauncher taskLauncher; @BeforeEach public void setup() { @@ -74,10 +82,16 @@ public class CFMigrateScheduleConfigurationTests { this.migrateProperties = new MigrateProperties(); this.taskDefinitionRepository = Mockito.mock(TaskDefinitionRepository.class); this.scheduler = Mockito.mock(Scheduler.class); + this.appRegistrationRepository = Mockito.mock(AppRegistrationRepository.class); + this.taskLauncher = Mockito.mock(TaskLauncher.class); + + this.cfConvertSchedulerService = new CFMigrateSchedulerService(this.cloudFoundryOperations, this.schedulerClient, this.cloudFoundryConnectionProperties, this.migrateProperties, - this.taskDefinitionRepository, new MavenProperties()) ; + this.taskDefinitionRepository, new MavenProperties(), + this.appRegistrationRepository, + this.taskLauncher) ; } @@ -85,15 +99,15 @@ public class CFMigrateScheduleConfigurationTests { public void testEnrichment() { ConvertScheduleInfo scheduleInfo = createFoundationConvertScheduleInfo(); assertThat(scheduleInfo.getAppProperties().keySet().size()).isEqualTo(5); - assertThat(scheduleInfo.getAppProperties().get("tasklauncher.app.defaultAppName.foo")).isEqualTo("bar"); - assertThat(scheduleInfo.getAppProperties().get("spring.cloud.dataflow.client.serverUri")).isEqualTo("http://localhost:9393"); - assertThat(scheduleInfo.getAppProperties().get("tasklauncher.deployer.defaultAppName.cloudfoundry.memory")).isEqualTo("1024m"); - assertThat(scheduleInfo.getAppProperties().get("tasklauncher.deployer.defaultAppName.cloudfoundry.health-check")).isEqualTo("port"); - assertThat(scheduleInfo.getAppProperties().get("tasklauncher.deployer.defaultAppName.cloudfoundry.disk")).isEqualTo("1024m"); + assertThat(scheduleInfo.getAppProperties().get("foo")).isEqualTo("bar"); + assertThat(scheduleInfo.getAppProperties().get("dataflow-server-uri")).isEqualTo(DEFAULT_DATA_FLOW_URI); + assertThat(scheduleInfo.getAppProperties().get("cloudfoundry.memory")).isEqualTo("1024m"); + assertThat(scheduleInfo.getAppProperties().get("cloudfoundry.health-check")).isEqualTo("port"); + assertThat(scheduleInfo.getAppProperties().get("cloudfoundry.disk")).isEqualTo("1024m"); assertThat(scheduleInfo.getCommandLineArgs().size()).isEqualTo(2); - assertThat(scheduleInfo.getCommandLineArgs().get(0)).isEqualTo("cmdarg.tasklauncher.defaultCmd=WOW"); - assertThat(scheduleInfo.getCommandLineArgs().get(1)).isEqualTo("--spring.cloud.scheduler.task.launcher.taskName=defaultTaskDefinitionName"); + assertThat(scheduleInfo.getCommandLineArgs().get(0)).isEqualTo(DEFAULT_CMD_ARG); + assertThat(scheduleInfo.getCommandLineArgs().get(1)).isEqualTo("--spring.application.name=defaultTaskDefinitionName"); } @Test @@ -103,18 +117,21 @@ public class CFMigrateScheduleConfigurationTests { scheduleInfo.setTaskDefinitionName(DEFAULT_TASK_DEFINITION_NAME); scheduleInfo.setScheduleProperties(new HashMap<>()); scheduleInfo.setRegisteredAppName(DEFAULT_APP_NAME); + scheduleInfo.getCommandLineArgs().add(TASK_LAUNCHER_TASK_NAME_ARG); Mockito.when(cloudFoundryOperations.applications()).thenReturn(new NoPropertyApplication()); + createMockAppRegistration(); TaskDefinition taskDefinition = TaskDefinition.TaskDefinitionBuilder .from(new TaskDefinition("fooTask", "foo")) .setTaskName(DEFAULT_TASK_DEFINITION_NAME) .setRegisteredAppName(DEFAULT_APP_NAME) + .setDslText("timestamp") .build(); Mockito.when(this.taskDefinitionRepository.findByTaskName(Mockito.any())).thenReturn(taskDefinition); scheduleInfo = this.cfConvertSchedulerService.enrichScheduleMetadata(scheduleInfo); assertThat(scheduleInfo.getAppProperties().keySet().size()).isEqualTo(1); - assertThat(scheduleInfo.getAppProperties().get("spring.cloud.dataflow.client.serverUri")).isEqualTo("http://localhost:9393"); + assertThat(scheduleInfo.getAppProperties().get("dataflow-server-uri")).isEqualTo("http://localhost:9393"); assertThat(scheduleInfo.getCommandLineArgs().size()).isEqualTo(1); - assertThat(scheduleInfo.getCommandLineArgs().get(0)).isEqualTo("--spring.cloud.scheduler.task.launcher.taskName=defaultTaskDefinitionName"); + assertThat(scheduleInfo.getCommandLineArgs().get(0)).isEqualTo("--spring.application.name=defaultTaskDefinitionName"); } @Test @@ -126,10 +143,10 @@ public class CFMigrateScheduleConfigurationTests { scheduleInfo.setRegisteredAppName(DEFAULT_APP_NAME); scheduleInfo.getCommandLineArgs().add(DEFAULT_CMD_ARG); Mockito.when(cloudFoundryOperations.applications()).thenReturn(new SinglePropertyApplication()); - assertThrows(IllegalStateException.class, () -> this.cfConvertSchedulerService.enrichScheduleMetadata(scheduleInfo)); + scheduleInfo = this.cfConvertSchedulerService.enrichScheduleMetadata(scheduleInfo); + assertThat(scheduleInfo.getTaskResource()).isNull(); } - @Test public void testMigrate() { ConvertScheduleInfo scheduleInfo = createFoundationConvertScheduleInfo(); @@ -137,8 +154,8 @@ public class CFMigrateScheduleConfigurationTests { final ArgumentCaptor scheduleRequestArgument = ArgumentCaptor.forClass(ScheduleRequest.class); final ArgumentCaptor scheduleNameArg = ArgumentCaptor.forClass(String.class); verify(this.scheduler, times(1)).schedule(scheduleRequestArgument.capture()); - verify(this.scheduler, times(1)).unschedule(scheduleNameArg.capture()); - assertThat(scheduleRequestArgument.getValue().getScheduleName()).isEqualTo("defaultScheduleName-scdf-defaultTaskDefinitionName"); + verify(this.taskLauncher, times(1)).destroy(scheduleNameArg.capture()); + assertThat(scheduleRequestArgument.getValue().getScheduleName()).isEqualTo("defaultScheduleName"); assertThat(scheduleNameArg.getValue()).isEqualTo(DEFAULT_SCHEDULE_NAME); } @@ -149,13 +166,17 @@ public class CFMigrateScheduleConfigurationTests { scheduleInfo.setScheduleProperties(new HashMap<>()); scheduleInfo.setRegisteredAppName(DEFAULT_APP_NAME); scheduleInfo.getCommandLineArgs().add(DEFAULT_CMD_ARG); + scheduleInfo.getCommandLineArgs().add(TASK_LAUNCHER_TASK_NAME_ARG); + Mockito.when(cloudFoundryOperations.applications()).thenReturn(new SinglePropertyApplication()); TaskDefinition taskDefinition = TaskDefinition.TaskDefinitionBuilder .from(new TaskDefinition("fooTask", "foo")) .setTaskName(DEFAULT_TASK_DEFINITION_NAME) .setRegisteredAppName(DEFAULT_APP_NAME) + .setDslText("timestamp") .build(); Mockito.when(this.taskDefinitionRepository.findByTaskName(Mockito.any())).thenReturn(taskDefinition); + createMockAppRegistration(); return this.cfConvertSchedulerService.enrichScheduleMetadata(scheduleInfo); } @@ -217,4 +238,15 @@ public class CFMigrateScheduleConfigurationTests { return Mono.empty(); } } + + private void createMockAppRegistration() { + AppRegistration appRegistration = new AppRegistration(); + try { + appRegistration.setUri(new URI("docker://fun:party")); + } + catch (URISyntaxException e) { + e.printStackTrace(); + } + Mockito.when(this.appRegistrationRepository.findAppRegistrationByNameAndTypeAndDefaultVersionIsTrue(Mockito.any(), Mockito.any())).thenReturn(appRegistration); + } } diff --git a/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/KubernetesMigrateScheduleConfigurationTests.java b/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/KubernetesMigrateScheduleConfigurationTests.java new file mode 100644 index 0000000..6410c20 --- /dev/null +++ b/dataflow-migrate-schedules/src/test/java/io/spring/migrateschedule/KubernetesMigrateScheduleConfigurationTests.java @@ -0,0 +1,219 @@ +/* + * Copyright 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 + * + * 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 io.spring.migrateschedule; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ObjectMeta; +import io.fabric8.kubernetes.api.model.PodSpec; +import io.fabric8.kubernetes.api.model.PodTemplateSpec; +import io.fabric8.kubernetes.api.model.batch.CronJob; +import io.fabric8.kubernetes.api.model.batch.CronJobList; +import io.fabric8.kubernetes.api.model.batch.CronJobSpec; +import io.fabric8.kubernetes.api.model.batch.DoneableCronJob; +import io.fabric8.kubernetes.api.model.batch.JobSpec; +import io.fabric8.kubernetes.api.model.batch.JobTemplateSpec; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.dsl.BatchAPIGroupDSL; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.spring.migrateschedule.service.AppRegistrationRepository; +import io.spring.migrateschedule.service.ConvertScheduleInfo; +import io.spring.migrateschedule.service.KubernetesMigrateSchedulerService; +import io.spring.migrateschedule.service.MigrateProperties; +import io.spring.migrateschedule.service.TaskDefinitionRepository; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +import org.springframework.cloud.dataflow.core.AppRegistration; +import org.springframework.cloud.dataflow.core.TaskDefinition; +import org.springframework.cloud.deployer.spi.scheduler.ScheduleRequest; +import org.springframework.cloud.deployer.spi.scheduler.Scheduler; +import org.springframework.core.io.Resource; + +import static org.assertj.core.api.AssertionsForClassTypes.assertThat; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +public class KubernetesMigrateScheduleConfigurationTests { + + public static final String DEFAULT_SCHEDULE_NAME = "defaultScheduleName-scdf-mydef"; + public static final String DEFAULT_TASK_DEFINITION_NAME = "defaultTaskDefinitionName"; + public static final String DEFAULT_APP_NAME = "defaultAppName"; + public static final String DEFAULT_CMD_ARG = "defaultCmd=WOW"; + public static final String DEFAULT_BUILD_PACK = "defaultBuildPack"; + public static final String DEFAULT_DB_URL = "jdbc:mydb://localhost:8888/task"; + public static final String DEFAULT_DRIVER_CLASS_NAME = "myDriverClassName"; + public static final String DEFAULT_DB_USER_NAME = "myuser"; + public static final String DEFAULT_DB_PASSWORD = "mypassword"; + + private KubernetesMigrateSchedulerService k8MigrateSchedulerService; + private KubernetesClient kubernetesClient; + private MigrateProperties migrateProperties; + private TaskDefinitionRepository taskDefinitionRepository; + private Scheduler scheduler; + private AppRegistrationRepository appRegistrationRepository; + private List sampleCronJobs; + + @BeforeEach + public void setup() { + this.kubernetesClient = Mockito.mock(KubernetesClient.class); + this.migrateProperties = new MigrateProperties(); + this.taskDefinitionRepository = Mockito.mock(TaskDefinitionRepository.class); + this.scheduler = Mockito.mock(Scheduler.class); + this.appRegistrationRepository = Mockito.mock(AppRegistrationRepository.class); + this.migrateProperties.setDbUrl(DEFAULT_DB_URL); + this.migrateProperties.setDbDriverClassName(DEFAULT_DRIVER_CLASS_NAME); + this.migrateProperties.setDbUserName(DEFAULT_DB_USER_NAME); + this.migrateProperties.setDbPassword(DEFAULT_DB_PASSWORD); + + this.k8MigrateSchedulerService = new KubernetesMigrateSchedulerService(this.taskDefinitionRepository, + this.appRegistrationRepository, + this.kubernetesClient, this.migrateProperties) ; + + BatchAPIGroupDSL batchAPIGroupDSL = Mockito.mock(BatchAPIGroupDSL.class); + MixedOperation> cronJobs = Mockito.mock(MixedOperation.class); + CronJobList cronJobList = Mockito.mock(CronJobList.class); + Mockito.when(this.kubernetesClient.batch()).thenReturn(batchAPIGroupDSL); + Mockito.when(batchAPIGroupDSL.cronjobs()).thenReturn(cronJobs); + Mockito.when(cronJobs.list()).thenReturn(cronJobList); + sampleCronJobs = new ArrayList<>(); + Mockito.when(cronJobList.getItems()).thenReturn(sampleCronJobs); + + //Setup kubernetesClient Mock to return a single schedule item in the result + CronJob cronJob = new CronJob(); + ObjectMeta objectMeta = new ObjectMeta(); + objectMeta.setName(DEFAULT_SCHEDULE_NAME); + cronJob.setMetadata(objectMeta); + this.sampleCronJobs.add(cronJob); + CronJobSpec cronJobSpec = new CronJobSpec(); + cronJob.setSpec(cronJobSpec); + JobTemplateSpec jobTemplateSpec = new JobTemplateSpec(); + cronJobSpec.setJobTemplate(jobTemplateSpec); + JobSpec jobSpec = new JobSpec(); + jobTemplateSpec.setSpec(jobSpec); + PodTemplateSpec podTemplateSpec = new PodTemplateSpec(); + jobSpec.setTemplate(podTemplateSpec); + PodSpec podSpec = new PodSpec(); + podTemplateSpec.setSpec(podSpec); + + Container container = new Container(); + podSpec.getContainers().add(container); + + } + + @Test + public void testEnrichment() { + ConvertScheduleInfo scheduleInfo = createFoundationConvertScheduleInfo(); + assertThat(scheduleInfo.getAppProperties().keySet().size()).isEqualTo(6); + assertThat(scheduleInfo.getAppProperties().get("foo")).isEqualTo("bar"); + assertThat(scheduleInfo.getDeployerProperties().get("que")).isEqualTo("qix"); + + validateDBProperties(scheduleInfo); + assertThat(scheduleInfo.getAppProperties().get("spring.cloud.task.name")).isEqualTo(DEFAULT_TASK_DEFINITION_NAME); + + assertThat(scheduleInfo.getCommandLineArgs().size()).isEqualTo(1); + assertThat(scheduleInfo.getCommandLineArgs().get(0)).isEqualTo("defaultCmd=WOW"); + } + + @Test + public void testEnrichmentNoProps() { + ConvertScheduleInfo scheduleInfo = new ConvertScheduleInfo(); + scheduleInfo.setScheduleName(DEFAULT_SCHEDULE_NAME); + scheduleInfo.setTaskDefinitionName(DEFAULT_TASK_DEFINITION_NAME); + scheduleInfo.setScheduleProperties(new HashMap<>()); + scheduleInfo.setRegisteredAppName(DEFAULT_APP_NAME); + TaskDefinition taskDefinition = TaskDefinition.TaskDefinitionBuilder + .from(new TaskDefinition("fooTask", "foo")) + .setTaskName(DEFAULT_TASK_DEFINITION_NAME) + .setRegisteredAppName(DEFAULT_APP_NAME) + .build(); + Mockito.when(this.taskDefinitionRepository.findByTaskName(Mockito.any())).thenReturn(taskDefinition); + scheduleInfo = this.k8MigrateSchedulerService.enrichScheduleMetadata(scheduleInfo); + assertThat(scheduleInfo.getAppProperties().keySet().size()).isEqualTo(5); + validateDBProperties(scheduleInfo); + assertThat(scheduleInfo.getAppProperties().get("spring.cloud.task.name")).isEqualTo(DEFAULT_TASK_DEFINITION_NAME); + assertThat(scheduleInfo.getCommandLineArgs().size()).isEqualTo(0); + } + + @Test + public void testListSchedules() { + + TaskDefinition taskDefinition = TaskDefinition.TaskDefinitionBuilder + .from(new TaskDefinition("fooTask", "foo")) + .setTaskName(DEFAULT_TASK_DEFINITION_NAME) + .setRegisteredAppName(DEFAULT_APP_NAME) + .setDslText("foo") + .build(); + AppRegistration appRegistration = new AppRegistration(); + try { + appRegistration.setUri(new URI("docker://fun:party")); + } + catch (URISyntaxException e) { + e.printStackTrace(); + } + Mockito.when(this.taskDefinitionRepository.findByTaskName(Mockito.any())).thenReturn(taskDefinition); + Mockito.when(this.appRegistrationRepository.findAppRegistrationByNameAndTypeAndDefaultVersionIsTrue(Mockito.any(), Mockito.any())).thenReturn(appRegistration); + List convertScheduleInfos = this.k8MigrateSchedulerService.scheduleInfoList(); + assertThat(convertScheduleInfos.size()).isEqualTo(1); + assertThat(convertScheduleInfos.get(0).getScheduleName()).isEqualTo(DEFAULT_SCHEDULE_NAME); + } + + @Test + public void testMigrate() { + ConvertScheduleInfo scheduleInfo = createFoundationConvertScheduleInfo(); + this.k8MigrateSchedulerService.migrateSchedule(this.scheduler, scheduleInfo); + final ArgumentCaptor scheduleRequestArgument = ArgumentCaptor.forClass(ScheduleRequest.class); + final ArgumentCaptor scheduleNameArg = ArgumentCaptor.forClass(String.class); + verify(this.scheduler, times(1)).schedule(scheduleRequestArgument.capture()); + verify(this.scheduler, times(1)).unschedule(scheduleNameArg.capture()); + assertThat(scheduleRequestArgument.getValue().getScheduleName()).isEqualTo("defaultScheduleName"); + assertThat(scheduleNameArg.getValue()).isEqualTo(DEFAULT_SCHEDULE_NAME); + } + + private ConvertScheduleInfo createFoundationConvertScheduleInfo() { + ConvertScheduleInfo scheduleInfo = new ConvertScheduleInfo(); + scheduleInfo.setScheduleName(DEFAULT_SCHEDULE_NAME); + scheduleInfo.setTaskDefinitionName(DEFAULT_TASK_DEFINITION_NAME); + scheduleInfo.setScheduleProperties(new HashMap<>()); + scheduleInfo.setRegisteredAppName(DEFAULT_APP_NAME); + scheduleInfo.getCommandLineArgs().add(DEFAULT_CMD_ARG); + scheduleInfo.setTaskResource(Mockito.mock(Resource.class)); + TaskDefinition taskDefinition = TaskDefinition.TaskDefinitionBuilder + .from(new TaskDefinition("fooTask", "foo")) + .setTaskName(DEFAULT_TASK_DEFINITION_NAME) + .setRegisteredAppName(DEFAULT_APP_NAME) + .build(); + Mockito.when(this.taskDefinitionRepository.findByTaskName(Mockito.any())).thenReturn(taskDefinition); + scheduleInfo.getAppProperties().put("foo", "bar"); + scheduleInfo.getDeployerProperties().put("que", "qix"); + return this.k8MigrateSchedulerService.enrichScheduleMetadata(scheduleInfo); + } + + private void validateDBProperties(ConvertScheduleInfo scheduleInfo) { + assertThat(scheduleInfo.getAppProperties().get("spring.datasource.url")).isEqualTo(DEFAULT_DB_URL); + assertThat(scheduleInfo.getAppProperties().get("spring.datasource.username")).isEqualTo(DEFAULT_DB_USER_NAME); + assertThat(scheduleInfo.getAppProperties().get("spring.datasource.password")).isEqualTo(DEFAULT_DB_PASSWORD); + assertThat(scheduleInfo.getAppProperties().get("spring.datasource.driverClassName")).isEqualTo(DEFAULT_DRIVER_CLASS_NAME); + } +}