Merge pull request #120 from LabKey/fb_merge_21.11_to_develop

Merge discvr-21.11 to develop

Author: bbimber

cluster/api-src/org/labkey/api/cluster/ClusterService.java

@@ -34,7 +34,11 @@ public static void setInstance(ClusterService instance)
      * This creates a barebones PipelineJob configured to run on the selected RemoteExecutionEngine.
      * You may wish to further configure this job.  This method does not submit the job, which you can do using PipelineService.queueJob()
      */
-    abstract public PipelineJob createClusterRemotePipelineJob(Container c, User u, String jobName, RemoteExecutionEngine engine, ClusterRemoteTask task, File logFile) throws PipelineValidationException;
+    abstract public PipelineJob createClusterRemotePipelineJob(Container c, User u, String jobName, RemoteExecutionEngine<?> engine, ClusterRemoteTask task, File logFile) throws PipelineValidationException;
+
+    abstract public File getSerializedJobFile(File jobLogFile);
+
+    abstract public File getExpectedSubmitScript(PipelineJob job);
 
     public interface ClusterRemoteTask extends Serializable
     {

cluster/resources/web/cluster/window/Buttons.js

@@ -0,0 +1,39 @@
+Ext4.namespace('Cluster.Utils');
+
+Cluster.Utils = new function() {
+    return {
+        buttonHandlerForLog: function (dataRegionName) {
+            const checked = LABKEY.DataRegions[dataRegionName].getChecked();
+            if (!checked.length){
+                Ext4.Msg.alert('Error', 'No rows selected');
+                return;
+            }
+            else if (checked.length > 1){
+                Ext4.Msg.alert('Error', 'Can only select one row at a time');
+                return;
+            }
+
+            window.open(LABKEY.ActionURL.buildURL('cluster', 'viewJavaLog', null, {jobId: checked[0]}), '_blank');
+        },
+
+        recoverCompletedJobs: function (dataRegionName) {
+            const checked = LABKEY.DataRegions[dataRegionName].getChecked();
+            if (!checked.length){
+                Ext4.Msg.alert('Error', 'No rows selected');
+                return;
+            }
+
+            window.location = LABKEY.ActionURL.buildURL('cluster', 'recoverCompletedJobs', null, {jobIds: checked.join(',')});
+        },
+
+        forcePipelineCancel: function (dataRegionName) {
+            const checked = LABKEY.DataRegions[dataRegionName].getChecked();
+            if (!checked.length){
+                Ext4.Msg.alert('Error', 'No rows selected');
+                return;
+            }
+
+            window.location = LABKEY.ActionURL.buildURL('cluster', 'forcePipelineCancel', null, {jobIds: checked.join(',')});
+        }
+    };
+};

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

@@ -16,17 +16,19 @@
 
 package org.labkey.cluster;
 
+import org.apache.commons.io.FilenameUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.labkey.api.action.ConfirmAction;
+import org.labkey.api.action.SimpleRedirectAction;
 import org.labkey.api.action.SpringActionController;
 import org.labkey.api.data.DbSchema;
 import org.labkey.api.data.DbSchemaType;
 import org.labkey.api.data.Table;
 import org.labkey.api.data.TableInfo;
+import org.labkey.api.pipeline.PipeRoot;
 import org.labkey.api.pipeline.PipelineJob;
-import org.labkey.api.pipeline.PipelineJobException;
 import org.labkey.api.pipeline.PipelineJobService;
 import org.labkey.api.pipeline.PipelineService;
 import org.labkey.api.pipeline.PipelineStatusFile;
@@ -35,17 +37,19 @@
 import org.labkey.api.security.RequiresPermission;
 import org.labkey.api.security.RequiresSiteAdmin;
 import org.labkey.api.security.permissions.AdminPermission;
+import org.labkey.api.security.permissions.ReadPermission;
+import org.labkey.api.settings.ResourceURL;
 import org.labkey.api.util.HtmlString;
 import org.labkey.api.util.PageFlowUtil;
 import org.labkey.api.util.URLHelper;
+import org.labkey.api.view.ActionURL;
 import org.labkey.api.view.HtmlView;
 import org.labkey.cluster.pipeline.AbstractClusterExecutionEngine;
 import org.springframework.validation.BindException;
 import org.springframework.validation.Errors;
 import org.springframework.web.servlet.ModelAndView;
 
 import java.io.File;
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -110,9 +114,10 @@ public URLHelper getSuccessURL(JobIdsForm form)
 
         public ModelAndView getConfirmView(JobIdsForm form, BindException errors) throws Exception
         {
+
             return new HtmlView(HtmlString.unsafe("This will change the status of the pipeline job with the provided ID to Cancelled.  It is intended to help the situation when the normal UI leave a job in a perpetual 'Cancelling' state." +
                     "To continue, enter a comma-delimited list of Job IDs and hit submit:<br><br>" +
-                    "<label>Enter Job ID(s): </label><input name=\"jobIds\"><br>"));
+                    "<label>Enter Job ID(s): </label><input name=\"jobIds\" value = \"" + form.getJobIds() + "\"><br>"));
         }
 
         public boolean handlePost(JobIdsForm form, BindException errors) throws Exception
@@ -274,7 +279,7 @@ public ModelAndView getConfirmView(JobIdsForm form, BindException errors) throws
         {
             return new HtmlView(HtmlString.unsafe("This will attempt to re-queue existing pipeline jobs using their serialized JSON text files.  It is intended as a workaround for the situation where a job has been marked complete." +
                     "To continue, enter a comma-delimited list of Job IDs and hit submit:<br><br>" +
-                    "<label>Enter Job ID(s): </label><input name=\"jobIds\"><br>"));
+                    "<label>Enter Job ID(s): </label><input name=\"jobIds\" value=\"" + form.getJobIds() + "\"><br>"));
         }
 
         public boolean handlePost(JobIdsForm form, BindException errors) throws Exception
@@ -306,28 +311,133 @@ public boolean handlePost(JobIdsForm form, BindException errors) throws Exceptio
                 sfs.add(sf);
             }
 
-            sfs.forEach(sf -> {
+            for (PipelineStatusFile sf : sfs)
+            {
                 File log = new File(sf.getFilePath());
                 File json = AbstractClusterExecutionEngine.getSerializedJobFile(log);
                 if (!json.exists())
                 {
-                    return;
+                    errors.reject(ERROR_MSG, "Unable to find pipeline JSON, expected: " + json.getPath());
+                    return false;
                 }
 
+                PipelineJob job = null;
                 try
                 {
-                    PipelineJob job = PipelineJob.readFromFile(json);
+                    job = PipelineJob.readFromFile(json);
 
-                    _log.info("Submitting job: " + job.getJobGUID() + ": " + job.getActiveTaskStatus());
+                    job.getLogger().info("Submitting job from JSON: " + job.getJobGUID() + ": " + job.getActiveTaskStatus());
                     PipelineService.get().setPipelineJobStatus(job, job.getActiveTaskStatus());
                 }
-                catch (PipelineJobException | IOException e)
+                catch (Exception e)
                 {
-                    _log.error(e);
+                    if (job != null)
+                    {
+                        job.getLogger().error("Unable to requeue job", e);
+                    }
+                    else
+                    {
+                        _log.error("Unable to requeue pipeline job", e);
+                    }
+
+                    errors.reject(ERROR_MSG, "Unable to requeue pipeline job: " + e.getMessage());
+                    return false;
                 }
-            });
+            }
 
             return true;
         }
     }
+
+    @RequiresPermission(ReadPermission.class)
+    public static class ViewJavaLogAction extends SimpleRedirectAction<ViewJavaLogForm>
+    {
+        @Override
+        public void validate(ViewJavaLogForm viewJavaLogForm, BindException errors)
+        {
+            super.validate(viewJavaLogForm, errors);
+
+            if (viewJavaLogForm.getJobId() == null)
+            {
+                errors.reject(ERROR_MSG, "Must provide JobId");
+            }
+
+            PipelineStatusFile sf = PipelineService.get().getStatusFile(viewJavaLogForm.getJobId());
+            if (sf == null)
+            {
+                errors.reject(ERROR_MSG, "Unknown job: " + viewJavaLogForm.getJobId());
+            }
+            else if (!sf.lookupContainer().hasPermission(getUser(), ReadPermission.class))
+            {
+                errors.reject(ERROR_MSG, "The current user does not have permission to view the folder: " + sf.lookupContainer().getPath());
+            }
+        }
+
+        @Override
+        public URLHelper getRedirectURL(ViewJavaLogForm viewJavaLogForm) throws Exception
+        {
+            PipelineStatusFile sf = PipelineService.get().getStatusFile(viewJavaLogForm.getJobId());
+            File parentDir = new File(sf.getFilePath()).getParentFile();
+            if (!parentDir.exists())
+            {
+                throw new IllegalArgumentException("Log directory doesnt exist: " + parentDir.getPath());
+            }
+
+            File[] javaLogs = parentDir.listFiles((dir, name) -> {
+                return name.endsWith(".java.log");
+            });
+
+            if (javaLogs == null || javaLogs.length == 0)
+            {
+                throw new IllegalArgumentException("No files ending with java.log found: " + parentDir.getPath());
+            }
+
+            long lastModifiedTime = Long.MIN_VALUE;
+            File chosenFile = null;
+            for (File file : javaLogs)
+            {
+                if (file.lastModified() > lastModifiedTime)
+                {
+                    chosenFile = file;
+                    lastModifiedTime = file.lastModified();
+                }
+            }
+
+            PipeRoot root = PipelineService.get().getPipelineRootSetting(sf.lookupContainer());
+            if (root == null)
+            {
+                throw new IllegalArgumentException("Unable to find pipeline root for folder: " + sf.lookupContainer().getPath());
+            }
+
+            if (!root.isUnderRoot(chosenFile))
+            {
+                throw new IllegalArgumentException("Log file is not under the pipeline root for folder: " + sf.lookupContainer().getPath());
+            }
+
+            String relPath = root.relativePath(chosenFile);
+            if (relPath == null)
+            {
+                throw new IllegalArgumentException("Unable to find log file path for folder: " + sf.lookupContainer().getPath());
+            }
+
+            relPath = org.labkey.api.util.Path.parse(FilenameUtils.separatorsToUnix(relPath)).encode();
+
+            return new URLHelper(root.getWebdavURL() + relPath);
+        }
+    }
+
+    public static class ViewJavaLogForm
+    {
+        private Integer _jobId;
+
+        public Integer getJobId()
+        {
+            return _jobId;
+        }
+
+        public void setJobId(Integer jobId)
+        {
+            _jobId = jobId;
+        }
+    }
 }

cluster/src/org/labkey/cluster/ClusterModule.java

@@ -24,6 +24,7 @@
 import org.labkey.api.data.DbSchema;
 import org.labkey.api.data.DbSchemaType;
 import org.labkey.api.ldk.ExtendedSimpleModule;
+import org.labkey.api.ldk.LDKService;
 import org.labkey.api.module.Module;
 import org.labkey.api.module.ModuleContext;
 import org.labkey.api.module.ModuleLoader;
@@ -36,7 +37,11 @@
 import org.labkey.api.view.WebPartFactory;
 import org.labkey.cluster.pipeline.ClusterPipelineJobNotificationProvider;
 import org.labkey.cluster.pipeline.ClusterPipelineProvider;
+import org.labkey.cluster.pipeline.SlurmExecutionEngine;
 import org.labkey.cluster.pipeline.TestCase;
+import org.labkey.cluster.query.ForceCancelJobsButton;
+import org.labkey.cluster.query.RecoverCompletedJobsButton;
+import org.labkey.cluster.query.ViewJavaLogButton;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -92,6 +97,10 @@ public void doStartupAfterSpringConfig(ModuleContext moduleContext)
         PipelineService.get().registerPipelineProvider(new ClusterPipelineProvider(this));
 
         PipelineService.get().registerPipelineJobNotificationProvider(new ClusterPipelineJobNotificationProvider());
+
+        LDKService.get().registerQueryButton(new ViewJavaLogButton(), "pipeline", "job");
+        LDKService.get().registerQueryButton(new RecoverCompletedJobsButton(), "pipeline", "job");
+        LDKService.get().registerQueryButton(new ForceCancelJobsButton(), "pipeline", "job");
     }
 
     @Override
@@ -120,6 +129,14 @@ public Set<Class> getIntegrationTests()
         return testClasses;
     }
 
+    @Override
+    public @NotNull Set<Class> getUnitTests()
+    {
+        return new HashSet<>(Arrays.asList(
+                SlurmExecutionEngine.TestCase.class
+        ));
+    }
+
     @Override
     protected void registerSchemas()
     {

cluster/src/org/labkey/cluster/ClusterServiceImpl.java

@@ -4,11 +4,15 @@
 import org.labkey.api.cluster.ClusterService;
 import org.labkey.api.data.Container;
 import org.labkey.api.pipeline.PipelineJob;
+import org.labkey.api.pipeline.PipelineJobService;
 import org.labkey.api.pipeline.PipelineValidationException;
 import org.labkey.api.pipeline.RemoteExecutionEngine;
 import org.labkey.api.pipeline.TaskId;
 import org.labkey.api.security.User;
+import org.labkey.cluster.pipeline.AbstractClusterExecutionEngine;
 import org.labkey.cluster.pipeline.ClusterPipelineJob;
+import org.labkey.cluster.pipeline.HTCondorExecutionEngine;
+import org.labkey.cluster.pipeline.SlurmExecutionEngine;
 
 import java.io.File;
 import java.util.ArrayList;
@@ -75,7 +79,52 @@ public String getClusterUser(Container c)
     }
 
     @Override
-    public PipelineJob createClusterRemotePipelineJob(Container c, User u, String jobName, RemoteExecutionEngine engine, ClusterRemoteTask task, File logFile) throws PipelineValidationException
+    public File getSerializedJobFile(File jobLogFile)
+    {
+        return AbstractClusterExecutionEngine.getSerializedJobFile(jobLogFile);
+    }
+
+    @Override
+    public File getExpectedSubmitScript(PipelineJob job)
+    {
+        if (job.isActiveTaskLocal())
+        {
+            //nothing to do:
+            return null;
+        }
+        else if (job.getActiveTaskFactory() != null)
+        {
+            String location = job.getActiveTaskFactory().getExecutionLocation();
+            for (RemoteExecutionEngine<?> engine : PipelineJobService.get().getRemoteExecutionEngines())
+            {
+                if (location.equals(engine.getConfig().getLocation()))
+                {
+                    if (SlurmExecutionEngine.TYPE.equals(engine.getType()))
+                    {
+                        return SlurmExecutionEngine.getExpectedSubmitScript(job);
+                    }
+                    else if (HTCondorExecutionEngine.TYPE.equals(engine.getType()))
+                    {
+                        return HTCondorExecutionEngine.getExpectedSubmitScript(job);
+                    }
+
+                    job.getLogger().error("Unknown execution engine type: " + engine.getType());
+                    return null;
+                }
+            }
+
+            job.getLogger().error("Unable to find remote execution engine for location: " + location);
+            return null;
+        }
+        else
+        {
+            job.getLogger().error("TaskFactory is null, cannot identify submit script");
+            return null;
+        }
+    }
+
+    @Override
+    public PipelineJob createClusterRemotePipelineJob(Container c, User u, String jobName, RemoteExecutionEngine<?> engine, ClusterRemoteTask task, File logFile) throws PipelineValidationException
     {
         return ClusterPipelineJob.createJob(c, u, jobName, task, engine, logFile);
     }