Prevent un-registration of Apps used in deployed streams

markpollack · jvalkeal · commit eccc8e22dc83 · 2018-03-05T10:49:14.000Z
Resolves #2013
diff --git a/spring-cloud-dataflow-registry/src/main/java/org/springframework/cloud/dataflow/registry/service/DefaultAppRegistryService.java b/spring-cloud-dataflow-registry/src/main/java/org/springframework/cloud/dataflow/registry/service/DefaultAppRegistryService.java
@@ -166,7 +166,6 @@ public AppRegistration save(AppRegistration app) {
 	 */
 	public void delete(String name, ApplicationType type, String version) {
 		this.appRegistrationRepository.deleteAppRegistrationByNameAndTypeAndVersion(name, type, version);
-		// TODO select new default
 	}
 
 	@Override
diff --git a/spring-cloud-dataflow-rest-resource/src/main/java/org/springframework/cloud/dataflow/rest/SkipperStream.java b/spring-cloud-dataflow-rest-resource/src/main/java/org/springframework/cloud/dataflow/rest/SkipperStream.java
@@ -36,4 +36,8 @@ public abstract class SkipperStream {
 	public static final String SKIPPER_DEFAULT_KIND = "SpringCloudDeployerApplication";
 
 	public static final String SKIPPER_DEFAULT_MAINTAINER = "dataflow";
+
+	public static final String SKIPPER_SPEC_RESOURCE = "resource";
+
+	public static final String SKIPPER_SPEC_VERSION = "version";
 }
diff --git a/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/config/DataFlowControllerAutoConfiguration.java b/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/config/DataFlowControllerAutoConfiguration.java
@@ -75,14 +75,14 @@
 import org.springframework.cloud.dataflow.server.controller.RootController;
 import org.springframework.cloud.dataflow.server.controller.RuntimeAppsController;
 import org.springframework.cloud.dataflow.server.controller.RuntimeAppsController.AppInstanceController;
+import org.springframework.cloud.dataflow.server.controller.SkipperAppRegistryController;
 import org.springframework.cloud.dataflow.server.controller.SkipperStreamDeploymentController;
 import org.springframework.cloud.dataflow.server.controller.StreamDefinitionController;
 import org.springframework.cloud.dataflow.server.controller.StreamDeploymentController;
 import org.springframework.cloud.dataflow.server.controller.TaskDefinitionController;
 import org.springframework.cloud.dataflow.server.controller.TaskExecutionController;
 import org.springframework.cloud.dataflow.server.controller.ToolsController;
 import org.springframework.cloud.dataflow.server.controller.UiController;
-import org.springframework.cloud.dataflow.server.controller.VersionedAppRegistryController;
 import org.springframework.cloud.dataflow.server.controller.security.LoginController;
 import org.springframework.cloud.dataflow.server.controller.security.SecurityController;
 import org.springframework.cloud.dataflow.server.controller.support.MetricStore;
@@ -406,10 +406,15 @@ public AppRegistryService appRegistryService(AppRegistrationRepository appRegist
 		}
 
 		@Bean
-		public VersionedAppRegistryController appRegistryController2(AppRegistryService appRegistry,
-				ApplicationConfigurationMetadataResolver metadataResolver, ForkJoinPool appRegistryFJPFB,
-				MavenProperties mavenProperties) {
-			return new VersionedAppRegistryController(appRegistry, metadataResolver, appRegistryFJPFB, mavenProperties);
+		public SkipperAppRegistryController skipperAppRegistryController(
+				StreamDefinitionRepository streamDefinitionRepository,
+				StreamService streamService,
+				AppRegistryService appRegistry, ApplicationConfigurationMetadataResolver metadataResolver,
+				ForkJoinPool appRegistryFJPFB, MavenProperties mavenProperties) {
+			return new SkipperAppRegistryController(streamDefinitionRepository,
+					streamService,
+					appRegistry,
+					metadataResolver, appRegistryFJPFB, mavenProperties);
 		}
 	}
 
diff --git a/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/RestControllerAdvice.java b/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/RestControllerAdvice.java
@@ -91,7 +91,8 @@ public VndErrors onException(Exception e) {
 	 * application/vnd.error+json
 	 */
 	@ExceptionHandler({ AppAlreadyRegisteredException.class, DuplicateStreamDefinitionException.class,
-			DuplicateTaskException.class, StreamAlreadyDeployedException.class, StreamAlreadyDeployingException.class })
+			DuplicateTaskException.class, StreamAlreadyDeployedException.class, StreamAlreadyDeployingException.class,
+			UnregisterAppException.class})
 	@ResponseStatus(HttpStatus.CONFLICT)
 	@ResponseBody
 	public VndErrors onConflictException(Exception e) {
diff --git a/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/SkipperAppRegistryController.java b/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/SkipperAppRegistryController.java
@@ -24,22 +24,33 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.ForkJoinPool;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.springframework.boot.configurationmetadata.ConfigurationMetadataProperty;
 import org.springframework.cloud.dataflow.configuration.metadata.ApplicationConfigurationMetadataResolver;
 import org.springframework.cloud.dataflow.core.ApplicationType;
+import org.springframework.cloud.dataflow.core.StreamAppDefinition;
+import org.springframework.cloud.dataflow.core.StreamDefinition;
+import org.springframework.cloud.dataflow.core.StreamDeployment;
 import org.springframework.cloud.dataflow.registry.domain.AppRegistration;
 import org.springframework.cloud.dataflow.registry.service.AppRegistryService;
 import org.springframework.cloud.dataflow.registry.service.DefaultAppRegistryService;
 import org.springframework.cloud.dataflow.registry.support.NoSuchAppRegistrationException;
 import org.springframework.cloud.dataflow.registry.support.ResourceUtils;
+import org.springframework.cloud.dataflow.rest.SkipperStream;
 import org.springframework.cloud.dataflow.rest.resource.AppRegistrationResource;
 import org.springframework.cloud.dataflow.rest.resource.DetailedAppRegistrationResource;
+import org.springframework.cloud.dataflow.server.DataFlowServerUtil;
+import org.springframework.cloud.dataflow.server.repository.StreamDefinitionRepository;
+import org.springframework.cloud.dataflow.server.service.StreamService;
+import org.springframework.cloud.dataflow.server.support.CannotDetermineApplicationTypeException;
 import org.springframework.cloud.deployer.resource.maven.MavenProperties;
 import org.springframework.core.io.ByteArrayResource;
 import org.springframework.core.io.DefaultResourceLoader;
@@ -75,25 +86,33 @@
 @RestController
 @RequestMapping("/apps")
 @ExposesResourceFor(AppRegistrationResource.class)
-public class VersionedAppRegistryController {
+public class SkipperAppRegistryController {
 
-	private static final Logger logger = LoggerFactory.getLogger(VersionedAppRegistryController.class);
+	private static final Logger logger = LoggerFactory.getLogger(SkipperAppRegistryController.class);
 
 	private final Assembler assembler = new Assembler();
 
+	private final StreamDefinitionRepository streamDefinitionRepository;
+
 	private final AppRegistryService appRegistryService;
 
+	private final MavenProperties mavenProperties;
+
+	private final StreamService streamService;
+
 	private ApplicationConfigurationMetadataResolver metadataResolver;
 
 	private ForkJoinPool forkJoinPool;
 
-	private final MavenProperties mavenProperties;
-
 	private ResourceLoader resourceLoader = new DefaultResourceLoader();
 
-	public VersionedAppRegistryController(AppRegistryService appRegistryService,
+	public SkipperAppRegistryController(StreamDefinitionRepository streamDefinitionRepository,
+			StreamService streamService,
+			AppRegistryService appRegistryService,
 			ApplicationConfigurationMetadataResolver metadataResolver,
 			ForkJoinPool forkJoinPool, MavenProperties mavenProperties) {
+		this.streamDefinitionRepository = streamDefinitionRepository;
+		this.streamService = streamService;
 		this.appRegistryService = appRegistryService;
 		this.metadataResolver = metadataResolver;
 		this.forkJoinPool = forkJoinPool;
@@ -230,9 +249,70 @@ public void makeDefault(@PathVariable("type") ApplicationType type, @PathVariabl
 	@ResponseStatus(HttpStatus.OK)
 	public void unregister(@PathVariable("type") ApplicationType type, @PathVariable("name") String name,
 			@PathVariable("version") String version) {
+
+		if (type != ApplicationType.task) {
+			String streamWithApp = findStreamContainingAppOf(type, name, version);
+			if (streamWithApp != null) {
+				throw new UnregisterAppException(String.format("The app [%s:%s:%s] you're trying to unregister is " +
+						"currently used in stream '%s'.", name, type, version, streamWithApp));
+			}
+		}
+
 		appRegistryService.delete(name, type, version);
 	}
 
+	/**
+	 * Given the application type, name, and version, determine if it is being used in a deployed stream definition.
+	 *
+	 * @param appType the application type
+	 * @param appName the application name
+	 * @param appVersion application version
+	 * @return the name of the deployed stream where the app is being used.  If the app is not deployed in a stream,
+	 * return {@code null}.
+	 */
+	private String findStreamContainingAppOf(ApplicationType appType, String appName, String appVersion) {
+		Iterable<StreamDefinition> streamDefinitions = streamDefinitionRepository.findAll();
+		for (StreamDefinition streamDefinition : streamDefinitions) {
+			StreamDeployment streamDeployment = this.streamService.info(streamDefinition.getName());
+			for (StreamAppDefinition streamAppDefinition : streamDefinition.getAppDefinitions()) {
+				final String streamAppName = streamAppDefinition.getRegisteredAppName();
+				final ApplicationType streamAppType;
+				try {
+					streamAppType = DataFlowServerUtil.determineApplicationType(streamAppDefinition);
+				}
+				catch (CannotDetermineApplicationTypeException e) {
+					logger.warn("Can not determine ApplicationType for " + streamAppDefinition);
+					continue;
+				}
+				if (appType != streamAppType) {
+					continue;
+				}
+				Map<String, Map<String, String>> streamDeploymentPropertiesMap;
+				String streamDeploymentPropertiesString = streamDeployment.getDeploymentProperties();
+				ObjectMapper objectMapper = new ObjectMapper();
+				try {
+					streamDeploymentPropertiesMap = objectMapper.readValue(streamDeploymentPropertiesString,
+							new TypeReference<Map<String, Map<String, String>>>() {
+							});
+				}
+				catch (IOException e) {
+					throw new RuntimeException("Can not deserialize Stream Deployment Properties JSON '"
+							+ streamDeploymentPropertiesString + "'");
+				}
+				if (streamDeploymentPropertiesMap.containsKey(appName)) {
+					Map<String, String> appDeploymentProperties = streamDeploymentPropertiesMap.get(streamAppName);
+					if (appDeploymentProperties.containsKey(SkipperStream.SKIPPER_SPEC_VERSION)) {
+						String version = appDeploymentProperties.get(SkipperStream.SKIPPER_SPEC_VERSION);
+						if (version != null && version.equals(appVersion)) {
+							return streamDefinition.getName();
+						}
+					}
+				}
+			}
+		}
+		return null;
+	}
+
 	@Deprecated
 	@RequestMapping(value = "/{type}/{name}", method = RequestMethod.DELETE)
 	@ResponseStatus(HttpStatus.OK)
@@ -306,7 +386,7 @@ private void prefetchMetadata(List<AppRegistration> appRegistrations) {
 	class Assembler extends ResourceAssemblerSupport<AppRegistration, AppRegistrationResource> {
 
 		public Assembler() {
-			super(VersionedAppRegistryController.class, AppRegistrationResource.class);
+			super(SkipperAppRegistryController.class, AppRegistrationResource.class);
 		}
 
 		@Override
diff --git a/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/UnregisterAppException.java b/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/UnregisterAppException.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.cloud.dataflow.server.controller;
+
+/**
+ * Exception thrown when an application can not be unregistered.
+ *
+ * @author Christian Tzolov
+ */
+public class UnregisterAppException extends RuntimeException {
+
+	private static final long serialVersionUID = 1L;
+
+	public UnregisterAppException(String message) {
+		super(message);
+	}
+}
diff --git a/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/stream/SkipperStreamDeployer.java b/spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/stream/SkipperStreamDeployer.java
@@ -431,7 +431,8 @@ public StreamDeployment getStreamInfo(String streamName) {
 				SpringCloudDeployerApplicationSpec spec = applicationManifest.getSpec();
 				String applicationName = applicationManifest.getApplicationName();
 				versionAndDeploymentProperties.putAll(spec.getDeploymentProperties());
-				versionAndDeploymentProperties.put(spec.getResource(), spec.getVersion());
+				versionAndDeploymentProperties.put(SkipperStream.SKIPPER_SPEC_RESOURCE, spec.getResource());
+				versionAndDeploymentProperties.put(SkipperStream.SKIPPER_SPEC_VERSION, spec.getVersion());
 				streamPropertiesMap.put(applicationName, versionAndDeploymentProperties);
 			}
 			return new StreamDeployment(streamName, new JSONObject(streamPropertiesMap).toString());
diff --git a/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/configuration/TestDependencies.java b/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/configuration/TestDependencies.java
@@ -54,13 +54,13 @@
 import org.springframework.cloud.dataflow.server.controller.MetricsController;
 import org.springframework.cloud.dataflow.server.controller.RestControllerAdvice;
 import org.springframework.cloud.dataflow.server.controller.RuntimeAppsController;
+import org.springframework.cloud.dataflow.server.controller.SkipperAppRegistryController;
 import org.springframework.cloud.dataflow.server.controller.SkipperStreamDeploymentController;
 import org.springframework.cloud.dataflow.server.controller.StreamDefinitionController;
 import org.springframework.cloud.dataflow.server.controller.StreamDeploymentController;
 import org.springframework.cloud.dataflow.server.controller.TaskDefinitionController;
 import org.springframework.cloud.dataflow.server.controller.TaskExecutionController;
 import org.springframework.cloud.dataflow.server.controller.ToolsController;
-import org.springframework.cloud.dataflow.server.controller.VersionedAppRegistryController;
 import org.springframework.cloud.dataflow.server.controller.support.ApplicationsMetrics;
 import org.springframework.cloud.dataflow.server.controller.support.ApplicationsMetrics.Application;
 import org.springframework.cloud.dataflow.server.controller.support.ApplicationsMetrics.Instance;
@@ -274,9 +274,13 @@ public AppRegistryService appRegistryService(AppRegistrationRepository appRegist
 
 	@Bean
 	@ConditionalOnSkipperEnabled
-	public VersionedAppRegistryController versionedAppRegistryController(AppRegistryService appRegistry,
+	public SkipperAppRegistryController versionedAppRegistryController(
+			StreamDefinitionRepository streamDefinitionRepository,
+			StreamService streamService,
+			AppRegistryService appRegistry,
 			ApplicationConfigurationMetadataResolver metadataResolver, MavenProperties mavenProperties) {
-		return new VersionedAppRegistryController(appRegistry, metadataResolver, new ForkJoinPool(2), mavenProperties);
+		return new SkipperAppRegistryController(streamDefinitionRepository, streamService, appRegistry, metadataResolver,
+				new ForkJoinPool(2), mavenProperties);
 	}
 
 	@Bean
diff --git a/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/controller/SkipperAppRegistryControllerTests.java b/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/controller/SkipperAppRegistryControllerTests.java
diff --git a/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSkipperStreamServiceIntegrationTests.java b/spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/service/impl/DefaultSkipperStreamServiceIntegrationTests.java
diff --git a/spring-cloud-dataflow-server-core/src/test/resources/org/springframework/cloud/dataflow/server/service/impl/DefaultSkipperStreamServiceIntegrationTests-deploymentProps.json b/spring-cloud-dataflow-server-core/src/test/resources/org/springframework/cloud/dataflow/server/service/impl/DefaultSkipperStreamServiceIntegrationTests-deploymentProps.json

Original file line number	Diff line number	Diff line change
`@@ -166,7 +166,6 @@ public AppRegistration save(AppRegistration app) {`
`166`	`166`	`*/`
`167`	`167`	`public void delete(String name, ApplicationType type, String version) {`
`168`	`168`	`this.appRegistrationRepository.deleteAppRegistrationByNameAndTypeAndVersion(name, type, version);`
`169`		`- // TODO select new default`
`170`	`169`	`}`
`171`	`170`
`172`	`171`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -36,4 +36,8 @@ public abstract class SkipperStream {`
`36`	`36`	`public static final String SKIPPER_DEFAULT_KIND = "SpringCloudDeployerApplication";`
`37`	`37`
`38`	`38`	`public static final String SKIPPER_DEFAULT_MAINTAINER = "dataflow";`
	`39`	`+`
	`40`	`+ public static final String SKIPPER_SPEC_RESOURCE = "resource";`
	`41`	`+`
	`42`	`+ public static final String SKIPPER_SPEC_VERSION = "version";`
`39`	`43`	`}`