Make TaskPlatform where SCDF is deployed primary and use with Scheduler

Fixes #3040

Author: David Turanski

spring-cloud-dataflow-core/src/main/java/org/springframework/cloud/dataflow/core/TaskPlatform.java

@@ -19,12 +19,17 @@
 import java.util.List;
 
 /**
+ * For a given platform, Cloud Foundry, Kubernetes, etc, associate a list of Task Launchers.
+ *
  * @author Donovan Muller
+ * @author David Turanski
  */
 public class TaskPlatform {
 
 	private String name;
 
+	private boolean primary;
+
 	private List<Launcher> launchers;
 
 	public TaskPlatform(String name, List<Launcher> launchers) {
@@ -47,5 +52,18 @@ public List<Launcher> getLaunchers() {
 	public void setLaunchers(List<Launcher> launchers) {
 		this.launchers = launchers;
 	}
+
+	/**
+	 * If true, identifies which Platform the Data Flow Server is running on.  This information is used to select
+	 * the TaskPlatform used by Scheduler.  @see SchedulerConfiguration.
+	 * @return true if the TaskPlatform is the same as where the Data Flow Server is running.
+	 */
+	public boolean isPrimary() {
+		return primary;
+	}
+
+	public void setPrimary(boolean primary) {
+		this.primary = primary;
+	}
 }
 

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

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.dataflow.server.config.cloudfoundry;
+
+import java.util.Optional;
+
+import io.pivotal.reactor.scheduler.ReactorSchedulerClient;
+import io.pivotal.scheduler.SchedulerClient;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.scheduler.spi.cloudfoundry.CloudFoundrySchedulerProperties;
+
+/**
+ * @author David Turanski
+ **/
+public class CloudFoundrySchedulerClientProvider {
+
+	private final CloudFoundryPlatformConnectionContextProvider connectionContextProvider;
+
+	private final CloudFoundryPlatformTokenProvider platformTokenProvider;
+
+	private final Optional<CloudFoundrySchedulerProperties> schedulerProperties;
+
+	public CloudFoundrySchedulerClientProvider(
+		CloudFoundryPlatformConnectionContextProvider connectionContextProvider,
+		CloudFoundryPlatformTokenProvider platformTokenProvider,
+		Optional<CloudFoundrySchedulerProperties> schedulerProperties) {
+
+
+		this.connectionContextProvider = connectionContextProvider;
+		this.platformTokenProvider = platformTokenProvider;
+		this.schedulerProperties = schedulerProperties;
+	}
+
+	public SchedulerClient cloudFoundrySchedulerClient(String account) {
+		return ReactorSchedulerClient.builder()
+				.connectionContext(connectionContextProvider.connectionContext(account))
+				.tokenProvider(platformTokenProvider.tokenProvider(account))
+				.root(Mono.just(schedulerProperties().getSchedulerUrl()))
+				.build();
+	}
+
+	public CloudFoundrySchedulerProperties schedulerProperties() {
+		return this.schedulerProperties.orElseGet(CloudFoundrySchedulerProperties::new);
+	}
+
+}

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

@@ -17,25 +17,38 @@
 
 import java.util.Optional;
 
-import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnCloudPlatform;
+import org.springframework.boot.cloud.CloudPlatform;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.dataflow.core.TaskPlatform;
+import org.springframework.cloud.dataflow.server.config.CloudProfileProvider;
+import org.springframework.cloud.dataflow.server.config.features.ConditionalOnTasksEnabled;
+import org.springframework.cloud.dataflow.server.config.features.SchedulerConfiguration.SchedulerConfigurationPropertyChecker;
 import org.springframework.cloud.scheduler.spi.cloudfoundry.CloudFoundrySchedulerProperties;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
 
 /**
  * Creates TaskPlatform implementations to launch/schedule tasks on Cloud Foundry.
  * @author Mark Pollack
  * @author David Turanski
  */
 @Configuration
+@ConditionalOnTasksEnabled
 @EnableConfigurationProperties(CloudFoundryPlatformProperties.class)
 public class CloudFoundryTaskPlatformAutoConfiguration {
 
 	@Bean
-	public TaskPlatform cloudFoundryPlatform(CloudFoundryTaskPlatformFactory cloudFoundryTaskPlatformFactory) {
-		return cloudFoundryTaskPlatformFactory.createTaskPlatform();
+	public TaskPlatform cloudFoundryPlatform(CloudFoundryTaskPlatformFactory cloudFoundryTaskPlatformFactory,
+		Environment environment) {
+		TaskPlatform taskPlatform = cloudFoundryTaskPlatformFactory.createTaskPlatform();
+		CloudProfileProvider cloudProfileProvider = new CloudFoundryCloudProfileProvider();
+		if (cloudProfileProvider.isCloudPlatform(environment)) {
+			taskPlatform.setPrimary(true);
+		}
+		return taskPlatform;
 	}
 
 	@Bean
@@ -59,21 +72,30 @@ public CloudFoundryPlatformClientProvider cloudFoundryClientProvider(
 				cloudFoundryPlatformProperties, connectionContextProvider, platformTokenProvider);
 	}
 
+	@Bean
+	@ConditionalOnCloudPlatform(CloudPlatform.CLOUD_FOUNDRY)
+	@Conditional(SchedulerConfigurationPropertyChecker.class)
+	public CloudFoundrySchedulerClientProvider schedulerClientProvider(
+			CloudFoundryPlatformConnectionContextProvider connectionContextProvider,
+			CloudFoundryPlatformTokenProvider platformTokenProvider,
+			Optional<CloudFoundrySchedulerProperties> schedulerProperties) {
+		return new CloudFoundrySchedulerClientProvider(
+				connectionContextProvider, platformTokenProvider, schedulerProperties);
+	}
+
 	@Bean
 	public CloudFoundryTaskPlatformFactory cloudFoundryTaskPlatformFactory(
 			CloudFoundryPlatformProperties cloudFoundryPlatformProperties,
 			CloudFoundryPlatformTokenProvider platformTokenProvider,
 			CloudFoundryPlatformConnectionContextProvider connectionContextProvider,
 			CloudFoundryPlatformClientProvider cloudFoundryClientProvider,
-			Optional<CloudFoundrySchedulerProperties> schedulerProperties,
-			@Value("${spring.cloud.dataflow.features.schedules-enabled:false}") boolean schedulesEnabled) {
+			Optional<CloudFoundrySchedulerClientProvider> cloudFoundrySchedulerClientProvider) {
 		return CloudFoundryTaskPlatformFactory.builder()
 				.platformProperties(cloudFoundryPlatformProperties)
 				.platformTokenProvider(platformTokenProvider)
 				.connectionContextProvider(connectionContextProvider)
 				.cloudFoundryClientProvider(cloudFoundryClientProvider)
-				.schedulesEnabled(schedulesEnabled)
-				.schedulerProperties(schedulerProperties)
+				.cloudFoundrySchedulerClientProvider(cloudFoundrySchedulerClientProvider)
 				.build();
 	}
 }

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

@@ -20,7 +20,7 @@
 
 import com.github.zafarkhaja.semver.Version;
 import io.jsonwebtoken.lang.Assert;
-import io.pivotal.reactor.scheduler.ReactorSchedulerClient;
+import io.pivotal.scheduler.SchedulerClient;
 import org.cloudfoundry.client.CloudFoundryClient;
 import org.cloudfoundry.client.v2.info.GetInfoRequest;
 import org.cloudfoundry.operations.CloudFoundryOperations;
@@ -29,7 +29,6 @@
 import org.cloudfoundry.reactor.TokenProvider;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import reactor.core.publisher.Mono;
 
 import org.springframework.cloud.dataflow.core.AbstractTaskPlatformFactory;
 import org.springframework.cloud.dataflow.core.Launcher;
@@ -49,33 +48,30 @@
  **/
 public class CloudFoundryTaskPlatformFactory extends AbstractTaskPlatformFactory<CloudFoundryPlatformProperties> {
 
-	private final boolean schedulerEnabled;
-
 	private final static String PLATFORM_TYPE = "Cloud Foundry";
 
 	private final static Logger logger = LoggerFactory.getLogger(CloudFoundryTaskPlatformFactory.class);
 
-	private final CloudFoundrySchedulerProperties schedulerProperties;
-
 	private final CloudFoundryPlatformTokenProvider platformTokenProvider;
 
 	private final CloudFoundryPlatformConnectionContextProvider connectionContextProvider;
 
 	private final CloudFoundryPlatformClientProvider cloudFoundryClientProvider;
 
+	private final Optional<CloudFoundrySchedulerClientProvider> cloudFoundrySchedulerClientProvider;
+
+
 	private CloudFoundryTaskPlatformFactory(CloudFoundryPlatformProperties cloudFoundryPlatformProperties,
 			CloudFoundryPlatformTokenProvider platformTokenProvider,
 			CloudFoundryPlatformConnectionContextProvider connectionContextProvider,
 			CloudFoundryPlatformClientProvider cloudFoundryClientProvider,
-			Optional<CloudFoundrySchedulerProperties> schedulerProperties,
-			boolean schedulerEnabled) {
-		super(cloudFoundryPlatformProperties, PLATFORM_TYPE);
+			Optional<CloudFoundrySchedulerClientProvider> cloudFoundrySchedulerClientProvider) {
 
-		this.schedulerEnabled = schedulerEnabled;
-		this.schedulerProperties = schedulerProperties.orElseGet(CloudFoundrySchedulerProperties::new);
+		super(cloudFoundryPlatformProperties, PLATFORM_TYPE);
 		this.platformTokenProvider = platformTokenProvider;
 		this.connectionContextProvider = connectionContextProvider;
 		this.cloudFoundryClientProvider = cloudFoundryClientProvider;
+		this.cloudFoundrySchedulerClientProvider = cloudFoundrySchedulerClientProvider;
 	}
 
 	@Override
@@ -89,36 +85,27 @@ public Launcher createLauncher(String account) {
 				deploymentProperties(account),
 				cloudFoundryOperations,
 				runtimeEnvironmentInfo(cloudFoundryClient, account));
-		Launcher launcher =  new Launcher(account, PLATFORM_TYPE, taskLauncher,
-				scheduler(
-						account,
-						taskLauncher,
-						cloudFoundryOperations,
-						connectionContext,
-						tokenProvider));
+		Launcher launcher = new Launcher(account, PLATFORM_TYPE, taskLauncher,
+				scheduler(account, taskLauncher, cloudFoundryOperations));
 		CloudFoundryConnectionProperties connectionProperties = connectionProperties(account);
 		launcher.setDescription(String.format("org = [%s], space = [%s], url = [%s]",
-			connectionProperties.getOrg(), connectionProperties.getSpace(),
-			connectionProperties.getUrl()));
+				connectionProperties.getOrg(), connectionProperties.getSpace(),
+				connectionProperties.getUrl()));
 		return launcher;
 	}
 
-	private Scheduler scheduler(String key, CloudFoundry2630AndLaterTaskLauncher taskLauncher,
-			CloudFoundryOperations cloudFoundryOperations, ConnectionContext connectionContext,
-			TokenProvider tokenProvider) {
+	private Scheduler scheduler(String account, CloudFoundry2630AndLaterTaskLauncher taskLauncher,
+			CloudFoundryOperations cloudFoundryOperations) {
 		Scheduler scheduler = null;
-		if (schedulerEnabled) {
-			ReactorSchedulerClient reactorSchedulerClient = ReactorSchedulerClient.builder()
-					.connectionContext(connectionContext)
-					.tokenProvider(tokenProvider)
-					.root(Mono.just(schedulerProperties.getSchedulerUrl()))
-					.build();
-
-			scheduler = new CloudFoundryAppScheduler(reactorSchedulerClient,
+		if (cloudFoundrySchedulerClientProvider.isPresent()) {
+			SchedulerClient schedulerClient =
+				cloudFoundrySchedulerClientProvider.get().cloudFoundrySchedulerClient(account);
+			scheduler = new CloudFoundryAppScheduler(
+					schedulerClient,
 					cloudFoundryOperations,
-					connectionProperties(key),
+					connectionProperties(account),
 					taskLauncher,
-					schedulerProperties);
+					cloudFoundrySchedulerClientProvider.get().schedulerProperties());
 		}
 		return scheduler;
 	}
@@ -196,6 +183,8 @@ public static class Builder {
 
 		private CloudFoundryPlatformClientProvider cloudFoundryClientProvider;
 
+		private Optional<CloudFoundrySchedulerClientProvider> cloudFoundrySchedulerClientProvider = Optional.empty();
+
 		public Builder platformProperties(CloudFoundryPlatformProperties platformProperties) {
 			this.platformProperties = platformProperties;
 			return this;
@@ -211,19 +200,25 @@ public Builder schedulerProperties(Optional<CloudFoundrySchedulerProperties> sch
 			return this;
 		}
 
+		public Builder cloudFoundrySchedulerClientProvider(Optional<CloudFoundrySchedulerClientProvider>
+			cloudFoundrySchedulerClientProvider) {
+			this.cloudFoundrySchedulerClientProvider = cloudFoundrySchedulerClientProvider;
+			return this;
+		}
+
 		public Builder platformTokenProvider(CloudFoundryPlatformTokenProvider platformTokenProvider) {
 			this.platformTokenProvider = platformTokenProvider;
 			return this;
 		}
 
 		public Builder connectionContextProvider(
-			CloudFoundryPlatformConnectionContextProvider connectionContextProvider) {
+				CloudFoundryPlatformConnectionContextProvider connectionContextProvider) {
 			this.connectionContextProvider = connectionContextProvider;
 			return this;
 		}
 
 		public Builder cloudFoundryClientProvider(
-			CloudFoundryPlatformClientProvider cloudFoundryClientProvider) {
+				CloudFoundryPlatformClientProvider cloudFoundryClientProvider) {
 			this.cloudFoundryClientProvider = cloudFoundryClientProvider;
 			return this;
 		}
@@ -235,12 +230,11 @@ public CloudFoundryTaskPlatformFactory build() {
 			Assert.notNull(cloudFoundryClientProvider, "'cloudFoundryClientProvider' is required.");
 
 			return new CloudFoundryTaskPlatformFactory(
-				platformProperties,
-				platformTokenProvider,
-				connectionContextProvider,
-				cloudFoundryClientProvider,
-				schedulerProperties,
-				schedulesEnabled);
+					platformProperties,
+					platformTokenProvider,
+					connectionContextProvider,
+					cloudFoundryClientProvider,
+					cloudFoundrySchedulerClientProvider);
 		}
 	}
 }

spring-cloud-dataflow-platform-cloudfoundry/src/test/java/org/springframework/cloud/dataflow/server/config/cloudfoundry/CloudFoundryTaskPlatformFactoryTests.java

@@ -19,6 +19,7 @@
 import java.util.Collections;
 import java.util.Optional;
 
+import io.pivotal.scheduler.SchedulerClient;
 import org.cloudfoundry.client.CloudFoundryClient;
 import org.cloudfoundry.client.v2.info.GetInfoResponse;
 import org.cloudfoundry.reactor.TokenProvider;
@@ -51,6 +52,9 @@ public class CloudFoundryTaskPlatformFactoryTests {
 	private CloudFoundryPlatformClientProvider cloudFoundryClientProvider = mock(
 			CloudFoundryPlatformClientProvider.class);
 
+	private CloudFoundrySchedulerClientProvider cloudFoundrySchedulerClientProvider = mock(
+			CloudFoundrySchedulerClientProvider.class);
+
 	private CloudFoundryClient cloudFoundryClient = mock(CloudFoundryClient.class);
 
 	private CloudFoundryPlatformProperties cloudFoundryPlatformProperties;
@@ -102,13 +106,18 @@ public void cloudFoundryTaskPlatformNoScheduler() {
 		assertThat(taskPlatform.getLaunchers().get(0).getTaskLauncher()).isInstanceOf(
 				CloudFoundry2630AndLaterTaskLauncher.class);
 		assertThat(taskPlatform.getLaunchers().get(0).getDescription()).isEqualTo(
-			"org = [org], space = [space], url = [https://localhost:9999]");
+				"org = [org], space = [space], url = [https://localhost:9999]");
 		assertThat(taskPlatform.getLaunchers().get(0).getScheduler()).isNull();
 	}
 
 	@Test
 	public void cloudFoundryTaskPlatformWithScheduler() {
 
+		when(cloudFoundrySchedulerClientProvider.cloudFoundrySchedulerClient(anyString())).thenReturn(
+				mock(SchedulerClient.class));
+		when(cloudFoundrySchedulerClientProvider.schedulerProperties())
+				.thenReturn(new CloudFoundrySchedulerProperties());
+
 		CloudFoundrySchedulerProperties schedulerProperties = new CloudFoundrySchedulerProperties();
 		schedulerProperties.setSchedulerUrl("https://localhost:9999");
 
@@ -118,6 +127,7 @@ public void cloudFoundryTaskPlatformWithScheduler() {
 				.platformTokenProvider(platformTokenProvider)
 				.connectionContextProvider(connectionContextProvider)
 				.cloudFoundryClientProvider(cloudFoundryClientProvider)
+				.cloudFoundrySchedulerClientProvider(Optional.of(cloudFoundrySchedulerClientProvider))
 				.schedulesEnabled(true)
 				.schedulerProperties(Optional.of(schedulerProperties))
 				.build();
@@ -130,7 +140,7 @@ public void cloudFoundryTaskPlatformWithScheduler() {
 		assertThat(taskPlatform.getLaunchers().get(0).getTaskLauncher()).isInstanceOf(
 				CloudFoundry2630AndLaterTaskLauncher.class);
 		assertThat(taskPlatform.getLaunchers().get(0).getDescription()).isEqualTo(
-			"org = [org], space = [space], url = [https://localhost:9999]");
+				"org = [org], space = [space], url = [https://localhost:9999]");
 		assertThat(taskPlatform.getLaunchers().get(0).getScheduler()).isNotNull();
 	}
 }