Cherry-pick upgrade related changes to release 6.2

cdap-api/src/main/java/io/cdap/cdap/api/app/Application.java

@@ -32,4 +32,26 @@ public interface Application<T extends Config> {
    * @param context Used to access the environment, application configuration, and application (deployment) arguments
    */
   void configure(ApplicationConfigurer configurer, ApplicationContext<T> context);
+
+  /**
+   * Returns if application supports config update or not.
+   */
+  default boolean isUpdateSupported() {
+    return false;
+  }
+
+  /**
+   * Updates application configuration based on config and update actions inside applicationUpdateContext.
+   *
+   * @param applicationUpdateContext Used to access methods helpful for operations like upgrading plugin version for
+   * config.
+   * @return {@link ApplicationUpdateResult} object for the config update operation.
+   * @throws UnsupportedOperationException if application does not support config update operation.
+   * @throws Exception if there was an exception during update of app config. This exception will often wrap
+   *                   the actual exception.
+   */
+  default ApplicationUpdateResult<T> updateConfig(ApplicationUpdateContext applicationUpdateContext)
+    throws Exception {
+    throw new UnsupportedOperationException("Application config update operation is not supported.");
+  }
 }

cdap-api/src/main/java/io/cdap/cdap/api/app/ApplicationConfigUpdateAction.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * 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.cdap.cdap.api.app;
+
+/**
+ * Possible update actions for application config.
+ */
+public enum ApplicationConfigUpdateAction {
+  // Upgrade artifacts to latest available versions.
+  UPGRADE_ARTIFACT,
+}

cdap-api/src/main/java/io/cdap/cdap/api/app/ApplicationUpdateContext.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * 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.cdap.cdap.api.app;
+
+import io.cdap.cdap.api.Config;
+import io.cdap.cdap.api.artifact.ArtifactId;
+
+import io.cdap.cdap.api.artifact.ArtifactScope;
+import io.cdap.cdap.api.artifact.ArtifactVersionRange;
+import java.lang.reflect.Type;
+import java.util.List;
+import javax.annotation.Nullable;
+
+/**
+ * Context for updating Application configs. Provides helper methods for application to support operations like config
+ * upgrade.
+ */
+public interface ApplicationUpdateContext {
+
+  /**
+   * @return All update actions application should perform on the config.
+   */
+  List<ApplicationConfigUpdateAction> getUpdateActions();
+
+  /**
+   * Get the old config as an object of the given type. The platform would perform the json deserialization based on
+   * the provided type. This is for the case where an application has the same/compatible/old config class. Application
+   * should decide on how they want to convert config from old to current type.
+   *
+   * @param configType type of the config platform should deserialize to.
+   * @return application config serialized to an object of given configType.
+   */
+  <C extends Config> C getConfig(Type configType);
+
+  /**
+   * Get the application configuration as json string.
+   */
+  String getConfigAsString();
+
+  /**
+   * Gets list of plugin artifacts based on given parameters in sorted in ascending order by version.
+   *
+   * @param pluginType the plugin type.
+   * @param pluginName the plugin name.
+   * @param pluginScope the scope to search plugins in.
+   * @param pluginRange the range of the version candidate plugins should be in.
+   * @return artifact list of plugins which matches with given parameters, sorted in ascending order.
+   *         Returns empty list if no artifact for the plugin found.
+   */
+  default List<ArtifactId> getPluginArtifacts(String pluginType, String pluginName, ArtifactScope pluginScope,
+                                              @Nullable ArtifactVersionRange pluginRange) throws Exception {
+    return getPluginArtifacts(pluginType, pluginName, pluginScope, pluginRange, Integer.MAX_VALUE);
+  }
+
+  /**
+   * Gets list of plugin artifacts based on given parameters in sorted in ascending order by version.
+   * Returns plugin artifacts using given filters in ascending order.
+   *
+   * @param pluginType the plugin type.
+   * @param pluginName the plugin name.
+   * @param pluginScope the scope to search plugins in.
+   * @param pluginRange the range of the version candidate plugins should be in.
+   * @param limit number of results to return at max, if null, default will be INT_MAX.
+   * @return artifact list of plugins which matches with given parameters, sorted in ascending order.
+   *         Returns empty list if no artifact for the plugin found.
+   */
+  List<ArtifactId> getPluginArtifacts(String pluginType, String pluginName, ArtifactScope pluginScope,
+                                      @Nullable ArtifactVersionRange pluginRange, int limit) throws Exception;
+
+}
+

cdap-api/src/main/java/io/cdap/cdap/api/app/ApplicationUpdateResult.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * 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.cdap.cdap.api.app;
+
+import io.cdap.cdap.api.Config;
+
+/**
+ * Stores results of upgrading an application config.
+ *
+ * @param <T> {@link Config} config class that represents the configuration type of an Application.
+ */
+public class ApplicationUpdateResult<T extends Config> {
+
+  // Upgraded config.
+  private final T newConfig;
+
+  public ApplicationUpdateResult(T newConfig) {
+    this.newConfig = newConfig;
+  }
+
+  public T getNewConfig() {
+    return newConfig;
+  }
+}

cdap-api/src/main/java/io/cdap/cdap/api/artifact/ArtifactVersionRange.java

@@ -49,6 +49,14 @@ public ArtifactVersion getUpper() {
     return upper;
   }
 
+  public boolean isLowerInclusive() {
+    return isLowerInclusive;
+  }
+
+  public boolean isUpperInclusive() {
+    return isUpperInclusive;
+  }
+
   public boolean versionIsInRange(ArtifactVersion version) {
     int lowerCompare = version.compareTo(lower);
     boolean lowerSatisfied = isLowerInclusive ? lowerCompare >= 0 : lowerCompare > 0;
@@ -69,7 +77,7 @@ public String getVersionString() {
     }
   }
 
-  private boolean isExactVersion() {
+  public boolean isExactVersion() {
     return isLowerInclusive && isUpperInclusive && upper.equals(lower);
   }
 

cdap-api/src/main/java/io/cdap/cdap/internal/io/AbstractSchemaGenerator.java

@@ -69,7 +69,10 @@ public abstract class AbstractSchemaGenerator implements SchemaGenerator {
     simpleSchemas.put(byte[].class, Schema.of(Schema.Type.BYTES));
     simpleSchemas.put(ByteBuffer.class, Schema.of(Schema.Type.BYTES));
 
-        // Some extra ones for some common build-in types. Need corresponding handling in DatumReader/Writer
+    // TODO: (CDAP-16919) Convert Object class mapping to union of all simple schema types.
+    simpleSchemas.put(Object.class, Schema.of(Schema.Type.NULL));
+
+    // Some extra ones for some common build-in types. Need corresponding handling in DatumReader/Writer
     simpleSchemas.put(URI.class, Schema.of(Schema.Type.STRING));
     simpleSchemas.put(URL.class, Schema.of(Schema.Type.STRING));
     simpleSchemas.put(UUID.class, Schema.of(Schema.Type.BYTES));
@@ -105,6 +108,8 @@ protected final Schema doGenerate(TypeToken<?> typeToken, Set<String> knownRecor
     Type type = typeToken.getType();
     Class<?> rawType = typeToken.getRawType();
 
+    // Object type was introduced in SIMPLE_SCHEMAS to satisfy Java Object usage in ETLConfig.
+    // Do not consider Java Object as simple schema for schema generation purpose.
     if (SIMPLE_SCHEMAS.containsKey(rawType)) {
       return SIMPLE_SCHEMAS.get(rawType);
     }

cdap-app-fabric/src/main/java/io/cdap/cdap/app/guice/AppFabricServiceRuntimeModule.java

@@ -52,6 +52,7 @@
 import io.cdap.cdap.gateway.handlers.ConfigHandler;
 import io.cdap.cdap.gateway.handlers.ConsoleSettingsHttpHandler;
 import io.cdap.cdap.gateway.handlers.ImpersonationHandler;
+import io.cdap.cdap.gateway.handlers.InstanceOperationHttpHandler;
 import io.cdap.cdap.gateway.handlers.NamespaceHttpHandler;
 import io.cdap.cdap.gateway.handlers.OperationalStatsHttpHandler;
 import io.cdap.cdap.gateway.handlers.OperationsDashboardHttpHandler;
@@ -328,6 +329,7 @@ protected void configure() {
       handlerBinder.addBinding().to(ConfigHandler.class);
       handlerBinder.addBinding().to(VersionHandler.class);
       handlerBinder.addBinding().to(UsageHandler.class);
+      handlerBinder.addBinding().to(InstanceOperationHttpHandler.class);
       handlerBinder.addBinding().to(NamespaceHttpHandler.class);
       handlerBinder.addBinding().to(AppLifecycleHttpHandler.class);
       handlerBinder.addBinding().to(AppLifecycleHttpHandlerInternal.class);

cdap-app-fabric/src/main/java/io/cdap/cdap/app/store/Store.java

@@ -50,6 +50,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.BiConsumer;
 import java.util.function.Predicate;
 import javax.annotation.Nullable;
 
@@ -303,6 +304,15 @@ List<ProgramSpecification> getDeletedProgramSpecifications(ApplicationId id,
    */
   Collection<ApplicationSpecification> getAllApplications(NamespaceId id);
 
+  /**
+   * Scans for applications across all namespaces.
+   *
+   * @param txBatchSize maximum number of applications to scan in one transaction to
+   *                    prevent holding a single transaction for too long
+   * @param consumer a {@link BiConsumer} to consume each application being scanned
+   */
+  void scanApplications(int txBatchSize, BiConsumer<ApplicationId, ApplicationSpecification> consumer);
+
   /**
    * Returns a Map of {@link ApplicationSpecification} for the given set of {@link ApplicationId}.
    *

cdap-app-fabric/src/main/java/io/cdap/cdap/gateway/handlers/AppLifecycleHttpHandler.java

@@ -40,6 +40,7 @@
 import io.cdap.cdap.common.InvalidArtifactException;
 import io.cdap.cdap.common.NamespaceNotFoundException;
 import io.cdap.cdap.common.NotFoundException;
+import io.cdap.cdap.common.ServiceException;
 import io.cdap.cdap.common.conf.CConfiguration;
 import io.cdap.cdap.common.conf.Constants;
 import io.cdap.cdap.common.http.AbstractBodyConsumer;
@@ -57,6 +58,7 @@
 import io.cdap.cdap.internal.app.services.ApplicationLifecycleService;
 import io.cdap.cdap.proto.ApplicationDetail;
 import io.cdap.cdap.proto.ApplicationRecord;
+import io.cdap.cdap.proto.ApplicationUpdateDetail;
 import io.cdap.cdap.proto.BatchApplicationDetail;
 import io.cdap.cdap.proto.artifact.AppRequest;
 import io.cdap.cdap.proto.id.ApplicationId;
@@ -74,6 +76,7 @@
 import io.netty.handler.codec.http.HttpHeaderValues;
 import io.netty.handler.codec.http.HttpRequest;
 import io.netty.handler.codec.http.HttpResponseStatus;
+
 import org.apache.twill.filesystem.Location;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -378,6 +381,66 @@ public void updateApp(FullHttpRequest request, HttpResponder responder,
     }
   }
 
+  /**
+   * upgrades an existing application.
+   */
+  @POST
+  @Path("/apps/{app-id}/upgrade")
+  @AuditPolicy(AuditDetail.REQUEST_BODY)
+  public void upgradeApplication(HttpRequest request, HttpResponder responder,
+                                 @PathParam("namespace-id") String namespaceId,
+                                 @PathParam("app-id") String appName) throws Exception {
+    ApplicationId appId = validateApplicationId(namespaceId, appName);
+    applicationLifecycleService.upgradeApplication(appId, createProgramTerminator());
+    ApplicationUpdateDetail updateDetail = new ApplicationUpdateDetail(appId, "upgrade successful.");
+    responder.sendJson(HttpResponseStatus.OK, GSON.toJson(updateDetail));
+  }
+
+  /**
+   * Upgrades a lis of existing application to use latest version of application artifact and plugin artifacts.
+   *
+   * <pre>
+   * {@code
+   * [
+   *   {"appId":"XYZ"},
+   *   {"appId":"ABC"},
+   *   {"appId":"FOO"},
+   * ]
+   * }
+   * </pre>
+   * The response will be an array of {@link ApplicationUpdateDetail} object, which either indicates a success (200) or
+   * failure for each of the requested application in the same order as the request. The failure also indicates reason
+   * for the error.
+   */
+  @POST
+  @Path("/upgrade")
+  @AuditPolicy(AuditDetail.REQUEST_BODY)
+  public void upgradeApplications(FullHttpRequest request, HttpResponder responder,
+                                  @PathParam("namespace-id") String namespace) throws Exception {
+    // TODO: (CDAP-16910) Improve batch API performance as each application upgrade is an event independent of each
+    //  other.
+    List<ApplicationId> appIds = decodeAndValidateBatchApplication(validateNamespace(namespace), request);
+    List<ApplicationUpdateDetail> details = new ArrayList<>();
+    for (ApplicationId appId : appIds) {
+      ApplicationUpdateDetail updateDetail;
+      try {
+        applicationLifecycleService.upgradeApplication(appId, createProgramTerminator());
+        updateDetail = new ApplicationUpdateDetail(appId, "upgrade successful.");
+      } catch (UnsupportedOperationException e) {
+        String errorMessage = String.format("Application %s does not support upgrade.", appId);
+        updateDetail = new ApplicationUpdateDetail(appId, errorMessage);
+      } catch (InvalidArtifactException | NotFoundException e) {
+        updateDetail = new ApplicationUpdateDetail(appId, e.getMessage());
+      } catch (Exception e) {
+        updateDetail =
+          new ApplicationUpdateDetail(appId, new ServiceException("Upgrade failed due to internal error.", null,
+                                      HttpResponseStatus.INTERNAL_SERVER_ERROR));
+      }
+      details.add(updateDetail);
+    }
+    responder.sendJson(HttpResponseStatus.OK, GSON.toJson(details));
+  }
+
   /**
    * Gets {@link ApplicationDetail} for a set of applications. It expects a post body as a array of object, with each
    * object specifying the applciation id and an optional version. E.g.