Restructure suite run time log/job/ and work/

bin/cylc

@@ -225,6 +225,7 @@ admin_commands['test-db'         ] = [ 'test-db']
 admin_commands['test-battery'    ] = [ 'test-battery']
 admin_commands['import-examples' ] = [ 'import-examples']
 admin_commands['upgrade-db'      ] = [ 'upgrade-db']
+admin_commands['upgrade-run-dir' ] = [ 'upgrade-run-dir']
 admin_commands['check-software'  ] = [ 'check-software']
 
 license_commands = OrderedDict()
@@ -357,6 +358,7 @@ comsum[ 'test-db'    ] = 'Run an automated suite name database test'
 comsum[ 'test-battery' ] = 'Run a battery of self-diagnosing test suites'
 comsum[ 'import-examples' ] = 'Import example suites your suite name database'
 comsum[ 'upgrade-db'      ] = 'Upgrade a pre-cylc-5.4 suite name database'
+comsum[ 'upgrade-run-dir' ] = 'Upgrade a pre-cylc-6 suite run directory'
 comsum[ 'check-software'  ] = 'Check required software is installed.'
 # license
 comsum[ 'warranty'   ] = 'Print the GPLv3 disclaimer of warranty'

bin/cylc-cat-log

@@ -69,12 +69,12 @@ owner = options.db_owner
 if len(args) == 2:
     # task log requested
     taskid = args[1]
-    logname = taskid + '.' + options.tryext
+    task, point = cylc.TaskID.split(taskid)
+    log_dir = os.path.join(point, task, "%02d" % int(options.tryext), "job")
 
     suite_run_dir = sitecfg.get_derived_host_item(suite, 'suite run directory')
     suite_run_dao = CylcRuntimeDAO(suite_run_dir)
-    taskname, point_string = cylc.TaskID.split( taskid )
-    host = suite_run_dao.get_task_host(taskname, point_string)
+    host = suite_run_dao.get_task_host(task, point)
     suite_run_dao.close()
     owner = None
     if host and '@' in host:
@@ -84,11 +84,11 @@ if len(args) == 2:
     sjld = sitecfg.get_derived_host_item( suite, 'suite job log directory',
                                        host=host, owner=owner )
     if options.print_stdout:
-        fpath = os.path.join( sjld, logname + '.out' )
+        fpath = os.path.join( sjld, log_dir + '.out' )
     elif options.print_stderr:
-        fpath = os.path.join( sjld, logname + '.err' )
+        fpath = os.path.join( sjld, log_dir + '.err' )
     else:
-        fpath = os.path.join( sjld, logname )
+        fpath = os.path.join( sjld, log_dir )
 
     if host or owner:
         if host and owner:

bin/cylc-get-job-status

@@ -36,7 +36,7 @@ __PYTHON__
 for OPT in "$@"; do
     if [[ $OPT == '--help' || $OPT == '-h' ]]; then
         cat <<'__USAGE__'
-USAGE: cylc [control] get-job-status ST-FILE JOB-SYS JOB-ID
+USAGE: cylc [control] get-job-status TASK-ID ST-FILE JOB-SYS JOB-ID
 
 This command is normally invoked automatically by cylc, to poll for job
 status of a task. To determine the current or final state of a task
@@ -49,6 +49,7 @@ Options:
   -h, --help   Print this help message and exit.
 
 Arguments:
+  TASK-ID - the task's NAME.CYCLE
   ST-FILE - the task status file (written to the task log directory).
   JOB-SYS - the name of the job submission system, e.g. pbs.
   JOB-ID - the job ID in the job submission system.
@@ -57,36 +58,36 @@ __USAGE__
     fi
 done
 
-if (($# != 3)); then
+if (($# != 4)); then
     echo "ERROR: wrong number of args!" >&2
     exit 1
 fi
 
-ST_FILE=$1
-JOB_SYS=$2
-JOB_ID=$3
+TASK_ID=$1
+ST_FILE=$2
+JOB_SYS=$3
+JOB_ID=$4
 
-TASKID=$(basename ${ST_FILE%.*.status})
-if [[ -f $ST_FILE ]]; then
-    JOB_EXIT_TIME=$(awk -F= '$1=="CYLC_JOB_EXIT_TIME" {print $2}' $ST_FILE)
-    JOB_EXIT=$(awk -F= '$1=="CYLC_JOB_EXIT" {print $2}' $ST_FILE)
+if [[ -f "$ST_FILE" ]]; then
+    JOB_EXIT_TIME=$(awk -F= '$1=="CYLC_JOB_EXIT_TIME" {print $2}' "$ST_FILE")
+    JOB_EXIT=$(awk -F= '$1=="CYLC_JOB_EXIT" {print $2}' "$ST_FILE")
     if [[ $JOB_EXIT == 'SUCCEEDED' && -n $JOB_EXIT_TIME ]]; then
-        echo "polled $TASKID succeeded at $JOB_EXIT_TIME"
+        echo "polled $TASK_ID succeeded at $JOB_EXIT_TIME"
     elif [[ -n $JOB_EXIT && -n $JOB_EXIT_TIME ]]; then
-        echo "polled $TASKID failed at $JOB_EXIT_TIME"
+        echo "polled $TASK_ID failed at $JOB_EXIT_TIME"
     elif poll_job_sys $JOB_SYS $JOB_ID; then
         # TODO - checkout status file for internal outputs
-        JOB_INIT_TIME=$(awk -F= '$1=="CYLC_JOB_INIT_TIME" {print $2}' $ST_FILE)
-        echo "polled $TASKID started at $JOB_INIT_TIME"
+        JOB_INIT_TIME=$(awk -F= '$1=="CYLC_JOB_INIT_TIME" {print $2}' "$ST_FILE")
+        echo "polled $TASK_ID started at $JOB_INIT_TIME"
     else
         # Did not exit cleanly
-        echo "polled $TASKID failed at unknown-time"
+        echo "polled $TASK_ID failed at unknown-time"
     fi
 elif poll_job_sys $JOB_SYS $JOB_ID; then
-    echo "polled $TASKID submitted"
+    echo "polled $TASK_ID submitted"
 else
     # FIXME: Really a "submitted" but "disappeared"!
-    echo "polled $TASKID submission failed"
+    echo "polled $TASK_ID submission failed"
 fi
 
 exit

bin/cylc-upgrade-run-dir

@@ -0,0 +1,149 @@
+#!/usr/bin/env python
+
+#C: THIS FILE IS PART OF THE CYLC SUITE ENGINE.
+#C: Copyright (C) 2008-2014 Hilary Oliver, NIWA
+#C:
+#C: This program is free software: you can redistribute it and/or modify
+#C: it under the terms of the GNU General Public License as published by
+#C: the Free Software Foundation, either version 3 of the License, or
+#C: (at your option) any later version.
+#C:
+#C: This program is distributed in the hope that it will be useful,
+#C: but WITHOUT ANY WARRANTY; without even the implied warranty of
+#C: MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#C: GNU General Public License for more details.
+#C:
+#C: You should have received a copy of the GNU General Public License
+#C: along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import re
+import os
+import sys
+import shutil
+from optparse import OptionParser
+from collections import defaultdict
+
+from cylc.cfgspec.site import sitecfg
+from cylc.mkdir_p import mkdir_p
+
+OLD_LOGFILE_RE = re.compile("""
+            ^
+            ([\w]+)  # task name
+            \.
+            ([^.]+)  # any cycle time format
+            \.
+            (\d+)    # submit number
+            (\.
+              (.*)
+            )?       # optional extension
+            $
+            """,
+            re.VERBOSE
+        )
+
+OLD_WORKDIR_RE = re.compile("""
+            ^
+            ([\w]+)  # task name
+            \.
+            ([^.]+)  # any cycle time format
+            $
+            """,
+            re.VERBOSE
+        )
+
+def upgrade_logdir(jobdir):
+    """Upgrade a pre cylc-6 suite job log directory."""
+
+    if not os.path.isdir(jobdir):
+        return
+    os.chdir(jobdir)
+    print "Upgrading %s" % jobdir
+    max_subnums = defaultdict(lambda:defaultdict(int))
+    for old_jobfile in os.listdir("."):
+        m = re.match(OLD_LOGFILE_RE, old_jobfile)
+        if not m:
+            print >> sys.stderr, (
+                    "WARNING: skipping non-standard log file: %s" % old_jobfile
+                    )
+            continue
+        sys.stdout.write(".")
+        task_name, cycle_point, subnum, dot, extn = m.groups()
+        if len(subnum) == 1:
+            subnum = "0" + subnum
+        new_jobdir = os.path.join(cycle_point, task_name, subnum)
+        mkdir_p(new_jobdir)
+        if extn is None:
+            # The job script.
+            jfile = "job"
+        elif extn in ["out", "err", "status"]:
+            # Cylc-generated logs.
+            jfile = "job." + extn
+        else:
+            # User-generated file.
+            jfile = extn
+        new_jobfile = os.path.join(new_jobdir, jfile)
+        shutil.move(old_jobfile, new_jobfile)
+        # Record max submit number for each task.
+        if int(subnum) > int(max_subnums[cycle_point][task_name]):
+            max_subnums[cycle_point][task_name] = subnum
+
+    # Symlink "NN" to the latest submit numbers.
+    for cycle_point, task_names in max_subnums.items():
+        for task_name, max_subnum in task_names.items():
+            sys.stdout.write(".")
+            target = os.path.join(cycle_point, task_name, "NN")
+            try:
+                os.symlink(max_subnum, target)
+            except OSError as exc:
+                if not exc.filename:
+                    exc.filename = target
+                raise exc
+    sys.stdout.write("\n")
+
+def upgrade_workdir(workdir):
+    """Upgrade a pre cylc-6 suite work directory."""
+
+    if not os.path.isdir(workdir):
+        return
+    os.chdir(workdir)
+    print "Upgrading %s" % workdir
+    for old_workdir in os.listdir("."):
+        m = re.match(OLD_WORKDIR_RE, old_workdir)
+        if not m:
+            print >> sys.stderr, (
+                    "WARNING: skipping non-standard workdir %s" % old_workdir
+                    )
+            continue
+        sys.stdout.write(".")
+        task_name, cycle_point = m.groups()
+        mkdir_p(cycle_point)
+        new_workdir = os.path.join(cycle_point, task_name)
+        shutil.move(old_workdir, new_workdir)
+    sys.stdout.write("\n")
+
+
+parser = OptionParser(
+    usage = """cylc [admin] upgrade-run-dir SUITE
+
+For one-off conversion of a suite run directory to cylc-6 format.
+
+Arguments:
+     SUITE    suite name or run directory path""")
+
+(options, args) = parser.parse_args()
+
+arg0 = args[0]
+if os.path.isdir(arg0):
+    rundir = arg0
+else:
+    rundir = sitecfg.get_derived_host_item(arg0, "suite run directory")
+if not os.path.isdir(rundir):
+    sys.exit("ERROR: Directory not found: %s" % rundir)
+
+logdir = os.path.join( rundir, "log", "job" )
+upgrade_logdir(logdir)
+
+workdir = os.path.join( rundir, "work" )
+upgrade_workdir(workdir)
+
+print "Done"

lib/cylc/cfgspec/suite.py

@@ -257,7 +257,7 @@ def _coerce_interval_list( value, keys, args, back_comp_unit_factor=1 ):
             'manual completion'               : vdr( vtype='boolean', default=False ),
             'extra log files'                 : vdr( vtype='string_list', default=[] ),
             'enable resurrection'             : vdr( vtype='boolean', default=False ),
-            'work sub-directory'              : vdr( vtype='string', default='$CYLC_TASK_ID' ),
+            'work sub-directory'              : vdr( vtype='string', default='$CYLC_TASK_CYCLE_POINT/$CYLC_TASK_NAME' ),
             'submission polling intervals'    : vdr( vtype='interval_minutes_list', default=[] ),
             'execution polling intervals'     : vdr( vtype='interval_minutes_list', default=[] ),
             'environment filter' : {

lib/cylc/gui/SuiteControl.py

@@ -27,7 +27,7 @@
 from cylc.suite_host import is_remote_host
 from cylc.owner import is_remote_user
 from dbchooser import dbchooser
-from combo_logviewer import combo_logviewer
+from combo_logviewer import ComboLogViewer
 from warning_dialog import warning_dialog, info_dialog
 
 try:
@@ -2165,30 +2165,28 @@ def popup_logview( self, task_id, logfiles, choice=None ):
         err = []
         out = []
         extra = []
-        for f in logfiles:
-            if f.endswith('.err'):
-                err.append(f)
-            elif f.endswith('.out'):
-                out.append(f)
-            elif re.search( '.*' + task_id + '\.\d+$', f ): # /a/b/c/foo.1.2
-                js = f
+        for logfile in logfiles:
+            if logfile.endswith('/job.err'):
+                err.append(logfile)
+            elif logfile.endswith('/job.out'):
+                out.append(logfile)
+            elif logfile.endswith('/job'):
+                js = logfile
             else:
-                extra.append( f )
+                extra.append( logfile )
 
         # for re-tries this sorts in time order due to filename:
         # (TODO - does this still work, post secs-since-epoch file extensions?)
-        key_func = lambda x: [int(w) if w.isdigit() else w for w in
-                              re.split("(\d+)", x)]
-        err.sort(key=key_func, reverse=True)
-        out.sort(key=key_func, reverse=True)
+        err.sort(key=self._sort_key_func, reverse=True)
+        out.sort(key=self._sort_key_func, reverse=True)
         window.set_size_request(800, 400)
         if choice == 'job script':
             window.set_title( task_id + ": Job Script" )
             lv = textload( task_id, js )
         else:
             logs = out + err + extra
             window.set_title( task_id + ": Log Files" )
-            lv = combo_logviewer( task_id, logs )
+            lv = ComboLogViewer( task_id, logs )
         #print "ADDING to quitters: ", lv
         self.quitters.append( lv )
 
@@ -2206,6 +2204,9 @@ def popup_logview( self, task_id, logfiles, choice=None ):
         window.connect("delete_event", lv.quit_w_e)
         window.show_all()
 
+    def _sort_key_func(self, x):
+        return [int(w) if w.isdigit() else w for w in re.split("(\d+)", x)]
+
     def _set_tooltip( self, widget, tip_text ):
         tooltip = gtk.Tooltips()
         tooltip.enable()