Merge pull request #34 from guillermo-carrasco/master

pm storage cleanup and pm storage archive-to-swestore

.gitignore

@@ -3,6 +3,6 @@ pm.egg-info
 dist
 build
 _build
-*log
+*.log
 .DS*
 

clean.sh

@@ -0,0 +1,5 @@
+# Remove building leftovers
+rm -rf build dist *egg-info
+
+# Remove pyc files
+find . -type f -name *pyc -exec rm {} +

pm/controllers/__init__.py

@@ -0,0 +1,28 @@
+""" Core module.
+
+Place for controllers and other structural stuff.
+"""
+from cement.core import controller
+
+class BaseController(controller.CementBaseController):
+    """ Define an application BaseController
+
+    The most basic controller. To be used as a template for new and more complex
+    controllers.
+    """
+    class Meta:
+        label = 'base'
+        description = "Project Management - A tool for miscellaneous tasks at NGI"
+
+
+    @controller.expose(hide=True)
+    def default(self):
+        print "Execute pm --help to display available commands"
+
+    @controller.expose(hide=True, help="Prints a hello message")
+    def hello(self):
+        """ Testing method that just prints a hello message.
+
+        Will not be listed as an available option (--hide)
+        """
+        self.app.log.info("Welcome to Project Management tools!")

pm/controllers/storage.py

@@ -0,0 +1,102 @@
+""" PM controllers
+"""
+import os
+import re
+import shutil
+import time
+
+from cement.core import controller
+
+from pm.controllers import BaseController
+from pm.utils import filesystem, misc
+
+class StorageController(BaseController):
+    """ Storage Controller
+
+    Entry point for all functionalities related to storage
+    """
+    class Meta:
+        label = 'storage'
+        description = "Entry point for all functionalities related to storage"
+        stacked_on = 'base'
+        stacked_type = 'nested'
+        arguments = [
+            (['-r', '--run'], dict(type=str, help="Work with a specific run")),
+            (['-d', '--days'], dict(type=int, default=10, help="Days to consider a run \"old\""))
+        ]
+
+    #######################
+    # Storage subcommands #
+    #######################
+
+    @controller.expose(help="Move old runs to nosync directory so they're not synced to the processing server")
+    def cleanup(self):
+        for data_dir in self.app.config.get('storage', 'data_dirs'):
+            with filesystem.chdir(data_dir):
+                for run in [r for r in os.listdir(data_dir) if re.match(filesystem.RUN_RE, r)]:
+                    rta_file = os.path.join(run, 'RTAComplete.txt')
+                    if os.path.exists(rta_file):
+                        # 1 day == 60*60*24 seconds --> 86400
+                        if os.stat(rta_file).st_mtime < time.time() - 86400:
+                            self.app.log.info('Moving run {} to nosync directory'.format(os.path.basename(run)))
+                            shutil.move(run, 'nosync')
+                        else:
+                            self.app.log('RTAComplete.txt file exists but is not older than 1 day, skipping run {}'.format(run))
+
+
+    @controller.expose(help="Archive old runs to SWESTORE")
+    def archive_to_swestore(self):
+        # If the run is specified in the command line, check that exists and archive
+        if self.app.pargs.run:
+            if re.match(filesystem.RUN_RE, os.path.basename(self.app.pargs.run)):
+                if not os.path.exists(self.app.pargs.run):
+                    self.app.log.error(("Run {} not found. Please make sure to specify "
+                        "the absolute path or relative path being in the correct directory.".format(self.app.pargs.run)))
+                else:
+                    self._archive_run(self.pargs.run)
+            else:
+                self.app.log.error("The name {} doesn't look like an Illumina run".format(os.path.basename(run)))
+        # Otherwise find all runs in every data dir on the nosync partition
+        else:
+            self.app.log.info("Archiving old runs to SWESTORE")
+            for data_dir in self.app.config.get('storage', 'data_dirs'):
+                to_send_dir = os.path.join(data_dir, 'nosync')
+                self.app.log.info('Checking {} directory'.format(to_send_dir))
+                with filesystem.chdir(to_send_dir):
+                    for run in [r for r in os.listdir(to_send_dir) if re.match(filesystem.RUN_RE, r)]:
+                        self._archive_run(run)
+
+    #############################################################
+    # Class helper methods, not exposed as commands/subcommands #
+    #############################################################
+    def _archive_run(self, run):
+        """ Archive a specific run to swestore
+
+        :param str run: Run directory
+        """
+        def _send_to_swestore(f, dest, remove=True):
+            """ Send file to swestore checking adler32 on destination and eventually
+            removing the file from disk
+
+            :param str f: File to remove
+            :param str dest: Destination directory in Swestore
+            :param bool remove: If True, remove original file from source
+            """
+            self.app.log.info("Sending {} to swestore".format(f))
+            misc.call_external_command('iput -K -P {file} {dest}'.format(file=f, dest=dest),
+                    with_log_files=True)
+            self.app.log.info('Run {} sent correctly and checksum was okay.'.format(f))
+            if remove:
+                self.app.log.info('Removing run'.format(f))
+                os.remove(f)
+
+
+        if run.endswith('bz2'):
+            _send_to_swestore(run, self.app.config.get('storage', 'irods').get('irodsHome'))
+        else:
+            self.app.log.info("Compressing run {}".format(run))
+            # Compress with pbzip2
+            misc.call_external_command('tar --use-compress-program=pbzip2 -cf {run}.tar.bz2 {run}'.format(run=run))
+            self.app.log.info('Run {} successfully compressed! Removing from disk...'.format(run))
+            shutil.rmtree(run)
+            _send_to_swestore('{}.tar.bz2'.format(run), self.app.config.get('storage', 'irods').get('irodsHome'))

pm/log/__init__.py

@@ -1,2 +1,2 @@
-""" Project Management logging module
+""" PM logging module for external scripts
 """

pm/log/loggers.py

@@ -4,18 +4,14 @@
 import os
 import sys
 
-from ConfigParser import NoOptionError, NoSectionError
-
 from pm.utils import config as cl
 
 
 def minimal_logger(namespace, config_file=None, to_file=True, debug=False):
     """Make and return a minimal console logger. Optionally write to a file as well.
-
     :param str namespace: Namespace of logger
     :param bool to_file: Log to a file (location in configuration file)
     :param bool debug: Log in DEBUG level or not
-
     :return: A logging.Logger object
     :rtype: logging.Logger
     """
@@ -33,19 +29,17 @@ def minimal_logger(namespace, config_file=None, to_file=True, debug=False):
     # File logger
     if to_file:
         cwd = os.path.dirname(os.path.realpath('.'))
-        log_path = os.path.join(os.environ['HOME'], '.pm', 'pm.log')
+        log_path = os.path.join(os.environ['HOME'], 'pm.log')
         if config_file or os.environ.get('PM_CONFIG'):
             if os.environ.get('PM_CONFIG'):
-                config = cl.load_config(os.environ.get('PM_CONFIG'))
-            else:
-                config = cl.load_config(config_file)
-            try:
-                log_path = config.get('log', 'log_dir')
-            except (NoOptionError, NoSectionError) as e:
-                raise e("Section [log] or option 'log_dir' were not found in the configuration file.")
+                config = cl.load_yaml_config(os.environ.get('PM_CONFIG'))
             else:
-                fh = logging.FileHandler(log_path)
-                fh.setLevel(log_level)
-                fh.setFormatter(formatter)
-                log.addHandler(fh)
-    return log
+                config = cl.load_yaml_config(config_file)
+            log_path = config.get('log', {}).get('log_dir')
+            if not log_path:
+                raise RuntimeError("Section [log] or option 'log_dir' were not found in the configuration file.")
+        fh = logging.FileHandler(log_path)
+        fh.setLevel(log_level)
+        fh.setFormatter(formatter)
+        log.addHandler(fh)
+    return log

pm/utils/filesystem.py

@@ -3,6 +3,8 @@
 import contextlib
 import os
 
+RUN_RE = '\d{6}_[a-zA-Z\d\-]+_\d{4}_[AB0][A-Z\d]'
+
 @contextlib.contextmanager
 def chdir(new_dir):
     """Context manager to temporarily change to a new directory.

pm/utils/misc.py

@@ -0,0 +1,36 @@
+""" Miscellaneous or general-use methods
+"""
+import os
+import subprocess
+import sys
+
+from datetime import datetime
+
+def call_external_command(cl, with_log_files=False):
+    """ Executes an external command
+
+    :param string cl: Command line to be executed (command + options and parameters)
+    :param bool with_log_files: Create log files for stdout and stderr
+    """
+    if type(cl) == str:
+        cl = cl.split(' ')
+    command = os.path.basename(cl[0])
+    stdout = sys.stdout
+    stderr = sys.stderr
+
+    if with_log_files:
+        stdout = open(command + '.out', 'wa')
+        stderr = open(command + '.err', 'wa')
+        started = "Started command {} on {}".format(' '.join(cl), datetime.now())
+        stdout.write(started + '\n')
+        stdout.write(''.join(['=']*len(cl)) + '\n')
+
+    try:
+        subprocess.check_call(cl, stdout=stdout, stderr=stderr)
+    except subprocess.CalledProcessError, e:
+        e.message = "The command {} failed.".format(' '.join(cl))
+        raise e
+    finally:
+        if with_log_files:
+            stdout.close()
+            stderr.close()

scripts/pm

@@ -2,28 +2,36 @@
 
 """ This is the entry point for PM.
 """
+import os
+
 from cement.core import foundation, handler
-from pm.core import controllers
+from cement.ext.ext_yaml import YamlConfigHandler
+from pm.controllers import BaseController
+from pm.controllers import storage
+
+CONFIG_FLIE = os.path.join(os.getenv('HOME'), '.pm', 'pm.yaml')
 
 class PmApp(foundation.CementApp):
     class Meta:
         label = 'PM'
-        base_controller = controllers.BaseController
+        base_controller = BaseController
+        config_handler = YamlConfigHandler
 
 # Create the main application
 app = PmApp()
 
 try:
     # Register handlers
-    handler.register(controllers.StorageController)
+    handler.register(storage.StorageController)
 
     # Setup the application
     app.setup()
 
-    # Add any common arguments to all handlers
-    app.args.add_argument('-c', '--config', type=str, help="Path to the configuration file")
-
-    app.run()
+    # Run the application
+    if not app.config.parse_file(CONFIG_FLIE):
+        app.log.error('No config file {}; please create and set relevant config sections'.format(CONFIG_FLIE))
+    else:
+        app.run()
 finally:
     # close the app
     app.close()