Auto deploy feed version

src/main/java/com/conveyal/datatools/common/status/MonitorableJob.java

@@ -44,6 +44,7 @@ public abstract class MonitorableJob implements Runnable, Serializable {
     protected List<MonitorableJob> subJobs = new ArrayList<>();
 
     public enum JobType {
+        AUTO_DEPLOY_FEED_VERSION,
         UNKNOWN_TYPE,
         ARBITRARY_FEED_TRANSFORM,
         BUILD_TRANSPORT_NETWORK,

src/main/java/com/conveyal/datatools/manager/controllers/api/DeploymentController.java

@@ -18,6 +18,7 @@
 import com.conveyal.datatools.manager.models.Project;
 import com.conveyal.datatools.manager.persistence.Persistence;
 import com.conveyal.datatools.manager.utils.json.JsonManager;
+import com.mongodb.client.FindIterable;
 import org.bson.Document;
 import org.eclipse.jetty.http.HttpStatus;
 import org.slf4j.Logger;
@@ -49,7 +50,7 @@
  */
 public class DeploymentController {
     private static final Logger LOG = LoggerFactory.getLogger(DeploymentController.class);
-    private static Map<String, DeployJob> deploymentJobsByServer = new HashMap<>();
+    private static final Map<String, DeployJob> deploymentJobsByServer = new HashMap<>();
 
     /**
      * Gets the deployment specified by the request's id parameter and ensure that user has access to the
@@ -376,6 +377,51 @@ private static boolean terminateEC2InstanceForDeployment(Request req, Response r
         return true;
     }
 
+    /**
+     * Queue a new {@link DeployJob} if there are no conflicting jobs assigned to the specified server.
+     * @param deployJob new deploy job to queue
+     * @return whether the deploy job was successfully queued or not
+     */
+    public static boolean queueDeployJob(DeployJob deployJob) {
+        String serverId = deployJob.getServerId();
+        // Check that we can deploy to the specified target. (Any deploy job for the target that is presently active will
+        // cause a halt.)
+        if (deploymentJobsByServer.containsKey(serverId)) {
+            // There is a deploy job for the server. Check if it is active.
+            DeployJob conflictingDeployJob = deploymentJobsByServer.get(serverId);
+            if (conflictingDeployJob != null && !conflictingDeployJob.status.completed) {
+                // Another deploy job is actively being deployed to the server target.
+                LOG.error("New deploy job will not be queued due to active job in progress.");
+                return false;
+            }
+        }
+
+        // For any previous deployments sent to the server/router combination, set deployedTo to null because
+        // this new one will overwrite it. NOTE: deployedTo for the current deployment will only be updated after the
+        // successful completion of the deploy job.
+        FindIterable<Deployment> deploymentsWithSameTarget = Deployment.retrieveDeploymentForServerAndRouterId(
+            serverId,
+            deployJob.getDeployment().routerId
+        );
+        for (Deployment oldDeployment : deploymentsWithSameTarget) {
+            LOG.info("Setting deployment target to null id={}", oldDeployment.id);
+            Persistence.deployments.updateField(oldDeployment.id, "deployedTo", null);
+        }
+        // Finally, add deploy job to the heavy executor.
+        DataManager.heavyExecutor.execute(deployJob);
+        deploymentJobsByServer.put(serverId, deployJob);
+        return true;
+    }
+
+    public static DeployJob checkDeploymentInProgress(String deploymentId) {
+        for (DeployJob job : deploymentJobsByServer.values()) {
+            if (job.getDeploymentId().equals(deploymentId)) {
+                return job;
+            }
+        }
+        return null;
+    }
+
     /**
      * HTTP controller to fetch information about provided EC2 machines that power ELBs running a trip planner.
      */
@@ -396,8 +442,9 @@ private static String deploy (Request req, Response res) {
         String target = req.params("target");
         Deployment deployment = getDeploymentWithPermissions(req, res);
         Project project = Persistence.projects.getById(deployment.projectId);
-        if (project == null)
+        if (project == null) {
             logMessageAndHalt(req, 400, "Internal reference error. Deployment's project ID is invalid");
+        }
         // Get server by ID
         OtpServer otpServer = Persistence.servers.getById(target);
         if (otpServer == null) {
@@ -411,23 +458,6 @@ private static String deploy (Request req, Response res) {
             logMessageAndHalt(req, 401, "User not authorized to deploy to admin-only target OTP server.");
         }
 
-        // Check that we can deploy to the specified target. (Any deploy job for the target that is presently active will
-        // cause a halt.)
-        if (deploymentJobsByServer.containsKey(target)) {
-            // There is a deploy job for the server. Check if it is active.
-            DeployJob deployJob = deploymentJobsByServer.get(target);
-            if (deployJob != null && !deployJob.status.completed) {
-                // Job for the target is still active! Send a 202 to the requester to indicate that it is not possible
-                // to deploy to this target right now because someone else is deploying.
-                String message = String.format(
-                        "Will not process request to deploy %s. Deployment currently in progress for target: %s",
-                        deployment.name,
-                        target);
-                LOG.warn(message);
-                logMessageAndHalt(req, HttpStatus.ACCEPTED_202, message);
-            }
-        }
-
         // Get the URLs to deploy to.
         List<String> targetUrls = otpServer.internalUrl;
         if ((targetUrls == null || targetUrls.isEmpty()) && (otpServer.s3Bucket == null || otpServer.s3Bucket.isEmpty())) {
@@ -438,19 +468,17 @@ private static String deploy (Request req, Response res) {
             );
         }
 
-        // For any previous deployments sent to the server/router combination, set deployedTo to null because
-        // this new one will overwrite it. NOTE: deployedTo for the current deployment will only be updated after the
-        // successful completion of the deploy job.
-        for (Deployment oldDeployment : Deployment.retrieveDeploymentForServerAndRouterId(target, deployment.routerId)) {
-            LOG.info("Setting deployment target to null id={}", oldDeployment.id);
-            Persistence.deployments.updateField(oldDeployment.id, "deployedTo", null);
-        }
-
         // Execute the deployment job and keep track of it in the jobs for server map.
         DeployJob job = new DeployJob(deployment, userProfile, otpServer);
-        DataManager.heavyExecutor.execute(job);
-        deploymentJobsByServer.put(target, job);
-
+        if (!queueDeployJob(job)) {
+            // Job for the target is still active! Send a 202 to the requester to indicate that it is not possible
+            // to deploy to this target right now because someone else is deploying.
+            String message = String.format(
+                "Will not process request to deploy %s. Deployment currently in progress for target: %s",
+                deployment.name,
+                target);
+            logMessageAndHalt(req, HttpStatus.ACCEPTED_202, message);
+        }
         return SparkUtils.formatJobMessage(job.jobId, "Deployment initiating.");
     }
 

src/main/java/com/conveyal/datatools/manager/jobs/AutoDeployFeedJob.java

@@ -0,0 +1,93 @@
+package com.conveyal.datatools.manager.jobs;
+
+import com.conveyal.datatools.common.status.MonitorableJob;
+import com.conveyal.datatools.manager.DataManager;
+import com.conveyal.datatools.manager.auth.Auth0UserProfile;
+import com.conveyal.datatools.manager.controllers.api.DeploymentController;
+import com.conveyal.datatools.manager.models.*;
+import com.conveyal.datatools.manager.persistence.Persistence;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+
+/**
+ * Auto deploy new feed version to OTP server if {@link Project#autoDeploy} is enabled and other conditions are met
+ * (e.g., feed version has no critical errors, active deployment is not in progress, etc.). This job must run after
+ * {@Link ValidateFeedJob} as it has a dependency on the outcome of {@Link FeedVersion#hasCriticalErrors}.
+ */
+public class AutoDeployFeedJob extends MonitorableJob {
+    public static final Logger LOG = LoggerFactory.getLogger(ValidateFeedJob.class);
+
+    private FeedVersion feedVersion;
+    private final FeedSource feedSource;
+
+    AutoDeployFeedJob(FeedVersion version, Auth0UserProfile owner, FeedSource source) {
+        super(owner, "Auto Deploy Feed", JobType.AUTO_DEPLOY_FEED_VERSION);
+        feedVersion = version;
+        feedSource = source;
+    }
+
+    @Override
+    public void jobLogic () {
+        if (!DataManager.isModuleEnabled("deployment")) {
+            // Do not attempt to deploy if module is disabled.
+            return;
+        }
+        Project project = feedSource.retrieveProject();
+        if (feedSource.deployable && project.pinnedDeploymentId != null && project.autoDeploy) {
+            // TODO: Get deployment, update feed version for feed source, and kick off deployment to server that
+            //  deployment is currently pointed at.
+            Deployment deployment = Persistence.deployments.getById(project.pinnedDeploymentId);
+            if (deployment == null) {
+                LOG.error("Pinned deployment no longer exists.");
+                // FIXME: Remove pinned deployment?
+            } else {
+                // If the deployment is already in progress, do not attempt auto-deploy.
+                // FIXME: Should this be placed in the DeploymentController#queueDeployJob body? Perhaps, since things
+                //  could get jumbled when updating the deployment object in MongoDB if we're trying to simultaneously
+                //  deploy to two servers at once.
+                DeployJob activeJob = DeploymentController.checkDeploymentInProgress(deployment.id);
+                if (activeJob != null) {
+                    LOG.warn("Skipping deployment (deployment by {} in progress already)", activeJob.retrieveEmail());
+                    return;
+                }
+                // As long as the feed does not have any critical errors, move forward with auto-deployment.
+                if (!feedVersion.hasCriticalErrors()) {
+                    if (deployment.feedVersionIds == null) {
+                        // FIXME: is it possible that no previous versions have been deployed?
+                        deployment.feedVersionIds = new ArrayList<>();
+                    }
+                    // Remove previously defined version for this feed source.
+                    for (FeedVersion versionToReplace : deployment.retrieveFullFeedVersions()) {
+                        if (versionToReplace.feedSourceId.equals(feedSource.id)) {
+                            deployment.feedVersionIds.remove(versionToReplace.id);
+                        }
+                    }
+                    // Add new version ID TODO: Should we not do this if the feed source was not already applied?
+                    deployment.feedVersionIds.add(feedVersion.id);
+                    Persistence.deployments.replace(deployment.id, deployment);
+                    // Send deployment (with new feed version) to most recently used server.
+                    OtpServer server;
+                    if (deployment.latest() != null) {
+                        String latestServerId = deployment.latest().serverId;
+                        server = Persistence.servers.getById(latestServerId);
+                        if (server == null) {
+                            LOG.error("Server with id {} no longer exists. Skipping deployment.", latestServerId);
+                            return;
+                        }
+                    } else {
+                        // FIXME: Should we deploy some other server if deployment has not previously been deployed?
+                        LOG.error("Deployment {} has never been deployed. Skipping auto-deploy.", deployment.id);
+                        return;
+                    }
+                    // Finally, queue up the new deploy job.
+                    // FIXME: Assumption being that if this returns true the feed version has been deployed.
+                    feedVersion.autoDeployed = DeploymentController.queueDeployJob(new DeployJob(deployment, owner, server));
+                    if (feedVersion.autoDeployed) LOG.info("New deploy job initiated for {}", server.name);
+                    else LOG.warn("Could not auto-deploy to {} due to conflicting active deployment.", server.name);
+                }
+            }
+        }
+    }
+}

src/main/java/com/conveyal/datatools/manager/jobs/DeployJob.java

@@ -180,15 +180,24 @@ public OtpServer getOtpServer() {
         return otpServer;
     }
 
+    /**
+     * Primary constructor for kicking off a deployment of transit/OSM/config data to OpenTripPlanner.
+     *
+     * FIXME: It appears that DeployType#Replace is the only type used in the non-test code. Should the others be
+     *  removed?
+     * @param deployment the deployment (set of feed versions and configurations) to deploy
+     * @param owner the requesting user
+     * @param otpServer the server/ELB target for the deployment
+     */
     public DeployJob(Deployment deployment, Auth0UserProfile owner, OtpServer otpServer) {
         this(deployment, owner, otpServer, null, DeployType.REPLACE);
     }
 
-    public DeployJob(Deployment deployment, Auth0UserProfile owner, OtpServer otpServer, String bundlePath, DeployType deployType) {
+    protected DeployJob(Deployment deployment, Auth0UserProfile owner, OtpServer otpServer, String bundlePath, DeployType deployType) {
         this("Deploying " + deployment.name, deployment, owner, otpServer, bundlePath, deployType);
     }
 
-    public DeployJob(String jobName, Deployment deployment, Auth0UserProfile owner, OtpServer otpServer, String bundlePath, DeployType deployType) {
+    private DeployJob(String jobName, Deployment deployment, Auth0UserProfile owner, OtpServer otpServer, String bundlePath, DeployType deployType) {
         // TODO add new job type or get rid of enum in favor of just using class names
         super(owner, jobName, JobType.DEPLOY_TO_OTP);
         this.deployment = deployment;

src/main/java/com/conveyal/datatools/manager/jobs/ProcessSingleFeedJob.java

@@ -4,14 +4,19 @@
 import com.conveyal.datatools.editor.jobs.CreateSnapshotJob;
 import com.conveyal.datatools.manager.DataManager;
 import com.conveyal.datatools.manager.auth.Auth0UserProfile;
+import com.conveyal.datatools.manager.controllers.api.DeploymentController;
+import com.conveyal.datatools.manager.models.Deployment;
 import com.conveyal.datatools.manager.models.FeedSource;
 import com.conveyal.datatools.manager.models.FeedVersion;
+import com.conveyal.datatools.manager.models.OtpServer;
+import com.conveyal.datatools.manager.models.Project;
 import com.conveyal.datatools.manager.models.Snapshot;
 import com.conveyal.datatools.manager.models.transform.DbTransformation;
 import com.conveyal.datatools.manager.models.transform.FeedTransformDbTarget;
 import com.conveyal.datatools.manager.models.transform.FeedTransformRules;
 import com.conveyal.datatools.manager.models.transform.FeedTransformZipTarget;
 import com.conveyal.datatools.manager.models.transform.ZipTransformation;
+import com.conveyal.datatools.manager.persistence.Persistence;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.slf4j.Logger;
@@ -171,16 +176,9 @@ public void jobLogic() {
                 status.fail("Could not clone version.", e);
             }
         }
-
         // FIXME: Should we overwrite the input GTFS dataset if transforming in place?
-
-        // TODO: Any other activities that need to be run (e.g., module-specific activities).
-        if (DataManager.isModuleEnabled("deployment")) {
-//            Project project = feedSource.retrieveProject();
-//            if (project.autoDeployId)
-            // TODO: Get deployment, update feed version for feed source, and kick off deployment to server that
-            //  deployment is currently pointed at.
-        }
+        // Make the necessary checks and update the OTP deployment with this new dataset if required.
+        addNextJob(new AutoDeployFeedJob(feedVersion, owner, feedSource));
     }
 
     /**