Add action to replace PipelineJob JobStore using serialized JSON

Author: bbimber

cluster/src/org/labkey/cluster/ClusterController.java

@@ -23,6 +23,7 @@
 import org.labkey.api.action.ConfirmAction;
 import org.labkey.api.action.SimpleRedirectAction;
 import org.labkey.api.action.SpringActionController;
+import org.labkey.api.collections.CaseInsensitiveHashMap;
 import org.labkey.api.data.DbSchema;
 import org.labkey.api.data.DbSchemaType;
 import org.labkey.api.data.Table;
@@ -352,6 +353,104 @@ public boolean handlePost(JobIdsForm form, BindException errors) throws Exceptio
         }
     }
 
+    @RequiresPermission(AdminPermission.class)
+    public static class ReplaceJobStoreAction extends ConfirmAction<JobIdsForm>
+    {
+        @Override
+        public void validateCommand(JobIdsForm form, Errors errors)
+        {
+
+        }
+
+        @Override
+        public URLHelper getSuccessURL(JobIdsForm form)
+        {
+            return PageFlowUtil.urlProvider(PipelineStatusUrls.class).urlBegin(getContainer());
+        }
+
+        @Override
+        public ModelAndView getConfirmView(JobIdsForm form, BindException errors) throws Exception
+        {
+            return new HtmlView(HtmlString.unsafe("This will read the serialized job JSON and replace the database JobStore it." +
+                    "To continue, enter a comma-delimited list of Job IDs and hit submit:<br><br>" +
+                    "<label>Enter Job ID(s): </label><input name=\"jobIds\" value=\"" + HtmlString.of(form.getJobIds()) + "\"><br>"));
+        }
+
+        @Override
+        public boolean handlePost(JobIdsForm form, BindException errors) throws Exception
+        {
+            String jobIDs = StringUtils.trimToNull(form.getJobIds());
+            if (jobIDs == null)
+            {
+                errors.reject(ERROR_MSG, "No JobIds provided");
+                return false;
+            }
+
+            Set<String> jobs = new HashSet<>(Arrays.asList(jobIDs.split(",")));
+            List<PipelineStatusFile> sfs = new ArrayList<>();
+            for (String id : jobs)
+            {
+                int jobId = Integer.parseInt(StringUtils.trimToNull(id));
+                PipelineStatusFile sf = PipelineService.get().getStatusFile(jobId);
+                if (sf == null)
+                {
+                    errors.reject(ERROR_MSG, "Unable to find job: " + id);
+                    return false;
+                }
+
+                if (PipelineJob.TaskStatus.running.name().equalsIgnoreCase(sf.getStatus()))
+                {
+                    errors.reject(ERROR_MSG, "This cannot be used on actively running jobs.  Status was: " + sf.getStatus());
+                    return false;
+                }
+
+                sfs.add(sf);
+            }
+
+            TableInfo ti = DbSchema.get("pipeline", DbSchemaType.Module).getTable("StatusFiles");
+            for (PipelineStatusFile sf : sfs)
+            {
+                File log = new File(sf.getFilePath());
+                File json = AbstractClusterExecutionEngine.getSerializedJobFile(log);
+                if (!json.exists())
+                {
+                    errors.reject(ERROR_MSG, "Unable to find pipeline JSON, expected: " + json.getPath());
+                    return false;
+                }
+
+                PipelineJob job = null;
+                try
+                {
+                    job = PipelineJob.readFromFile(json);
+                    job.getLogger().info("Replacing DB JobStore from JSON: " + job.getJobGUID() + ": " + job.getActiveTaskStatus());
+
+                    Map<String, Object> row = new CaseInsensitiveHashMap<>();
+                    row.put("rowid", sf.getRowId());
+                    row.put("jobStore", PipelineJob.serializeJob(job, false));
+                    row.put("activeTaskId", job.getActiveTaskId().toString());
+
+                    Table.update(getUser(), ti, row, sf.getRowId());
+                }
+                catch (Exception e)
+                {
+                    if (job != null)
+                    {
+                        job.getLogger().error("Unable to update JobStore", e);
+                    }
+                    else
+                    {
+                        _log.error("Unable to update JobStore", e);
+                    }
+
+                    errors.reject(ERROR_MSG, "Unable to update JobStore: " + e.getMessage());
+                    return false;
+                }
+            }
+
+            return true;
+        }
+    }
+
     @RequiresPermission(ReadPermission.class)
     public static class ViewJavaLogAction extends SimpleRedirectAction<ViewJavaLogForm>
     {