jflow_server.py

#
# Copyright (C) 2012 INRA
# 
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# 
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
# 
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import cherrypy
import cgi
import tempfile
import json
import sys
from functools import wraps
import time
import os
import argparse
import logging

from argparse import ArgumentTypeError

try:
    import _preamble
except ImportError:
    sys.exc_clear()

import jflow
from jflow.workflows_manager import WorkflowsManager
from jflow.config_reader import JFlowConfigReader
from jflow.workflow import Workflow
from workflows.types import *
import jflow.utils as utils
from cctools.util import time_format

# function in charge to upload large files
class UploadFieldStorage(cgi.FieldStorage):
    """Our version uses a named temporary file instead of the default
    non-named file; keeping it visibile (named), allows us to create a
    2nd link after the upload is done, thus avoiding the overhead of
    making a copy to the destination filename."""

    def get_tmp_directory(self):
        jflowconf = JFlowConfigReader()
        return jflowconf.get_tmp_directory()

    def make_file(self, binary=None):
        jflowconf = JFlowConfigReader()
        tmp_folder = jflowconf.get_tmp_directory()
        if not os.path.exists( tmp_folder ):
            try : os.mkdir(tmp_folder)
            except : pass
        return tempfile.NamedTemporaryFile(dir=tmp_folder)


def noBodyProcess():
    """Sets cherrypy.request.process_request_body = False, giving
    us direct control of the file upload destination. By default
    cherrypy loads it to memory, we are directing it to disk."""
    cherrypy.request.process_request_body = False

cherrypy.tools.noBodyProcess = cherrypy.Tool('before_request_body', noBodyProcess)

# define functions in charge to handle cross domain calls
def CORS():
    cherrypy.response.headers['Access-Control-Allow-Origin'] = '*'
    cherrypy.response.headers['Access-Control-Allow-Methods'] = 'OPTIONS, GET, POST'
    cherrypy.response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Content-Range, Content-Disposition'
cherrypy.tools.CORS = cherrypy.Tool('before_finalize', CORS)


class JFlowServer (object):

    MULTIPLE_TYPE_SPLITER = "._."
    JFLOW_WDATA = "data"

    def __init__(self):
        # Create a workflow manager to get access to our workflows
        self.wfmanager = WorkflowsManager()
        self.jflow_config_reader = JFlowConfigReader()

    def jsonify(func):
        '''JSON and JSONP decorator for CherryPy'''
        @wraps(func)
        def wrapper(*args, **kw):
            value = func(*args, **kw)
            cherrypy.response.headers["Content-Type"] = "application/json"
            # if JSONP request
            if kw.has_key("callback"): return kw["callback"] + "(" + json.dumps(value) + ")"
            # else return the JSON
            else: return json.dumps(value)
        return wrapper

    def jsonify_workflow_status(self, workflow, init_to_zero=False):
        if workflow.start_time: start_time = time.asctime(time.localtime(workflow.start_time))
        else: start_time = "-"
        if workflow.start_time and workflow.end_time: elapsed_time = str(workflow.end_time-workflow.start_time)
        elif workflow.start_time: elapsed_time = str(time.time()-workflow.start_time)
        else: elapsed_time = "-"
        if workflow.end_time: end_time = time.asctime(time.localtime(workflow.end_time))
        else: end_time = "-"
        if init_to_zero:
            return {"id":utils.get_nb_string(workflow.id),
                    "name": workflow.name,
                    "status": Workflow.STATUS_STARTED,
                    "elapsed_time": str(elapsed_time),
                    "start_time": start_time,
                    "end_time": end_time,
                    "components": []}
        else :
            components = []
            for i, component in enumerate(workflow.get_components_nameid()):
                status_info = workflow.get_component_status(component)
                try: perc_waiting = (status_info["waiting"]*100.0)/status_info["tasks"]
                except: perc_waiting = 0
                try: perc_running = (status_info["running"]*100.0)/status_info["tasks"]
                except: perc_running = 0
                try: perc_failed = (status_info["failed"]*100.0)/status_info["tasks"]
                except: perc_failed = 0
                try: perc_aborted = (status_info["aborted"]*100.0)/status_info["tasks"]
                except: perc_aborted = 0
                try: perc_completed = (status_info["completed"]*100.0)/status_info["tasks"]
                except: perc_completed = 0
                components.append({"name": component,
                                  "elapsed_time": time_format(status_info["time"]),
                                  "total": status_info["tasks"],
                                  "waiting": status_info["waiting"],
                                  "failed": status_info["failed"],
                                  "running": status_info["running"],
                                  "aborted": status_info["aborted"],
                                  "completed": status_info["completed"]})
            status = {"id":utils.get_nb_string(workflow.id),
                      "name": workflow.name,
                      "metadata": workflow.metadata,
                      "status": workflow.get_status(),
                      "elapsed_time": str(elapsed_time),
                      "start_time": start_time,
                      "end_time": end_time,
                      "components": components}
         
            return status

    @cherrypy.expose
    @jsonify
    def get_available_workflows(self, **kwargs):
        workflows = []
        wf_instances = self.wfmanager.get_available_workflows()
        for instance in wf_instances:
            parameters, parameters_per_groups, groups = [], {}, ["default"]
            for param in instance.parameters:
                # if it's a multiple action change the action by the name
                if param.action == jflow.parameter.MiltipleAction:
                    action = "MiltipleAction"
                elif param.action == jflow.parameter.MiltipleAppendAction:
                    action = "MiltipleAppendAction"
                else:
                    action = param.action
                hash_param = {"help": param.help,
                              "required": param.required,
                              "default": param.default,
                              "choices": param.choices,
                              "action": action,
                              "type": param.get_type(),
                              "name": param.name,
                              "display_name": param.display_name,
                              "group": param.group}
                # if it's a multiple type add sub parameters
                if type(param.type) == jflow.parameter.MultipleParameters:
                    hash_param["sub_parameters"] = [] 
                    for sub_param in param.sub_parameters:
                        hash_param["sub_parameters"].append({"help": sub_param.help,
                              "required": sub_param.required,
                              "default": sub_param.default,
                              "choices": sub_param.choices,
                              "action": sub_param.action,
                              "type": sub_param.get_type(),
                              "name": param.name + JFlowServer.MULTIPLE_TYPE_SPLITER + sub_param.flag,
                              "display_name": sub_param.display_name,
                              "group": param.group})
                parameters.append(hash_param)
                if parameters_per_groups.has_key(param.group):
                    parameters_per_groups[param.group].append(hash_param)
                else: parameters_per_groups[param.group] = [hash_param]
                groups.append(param.group)
            workflows.append({"name": instance.name,
                              "help": instance.description,
                              "class": instance.__class__.__name__,
                              "parameters": parameters,
                              "parameters_per_groups": parameters_per_groups,
                              "groups": list(set(groups))})
        return workflows

    @cherrypy.expose
    @jsonify
    def run_workflow(self, **kwargs):
        kwargs_modified = {}
        for key in kwargs.keys():
            parts = key.split(JFlowServer.MULTIPLE_TYPE_SPLITER)
            if len(parts) == 1:
                kwargs_modified[key] = kwargs[key]
            elif len(parts) == 2:
                if kwargs_modified.has_key(parts[0]):
                    kwargs_modified[parts[0]].append((parts[1], kwargs[key]))
                else:
                    kwargs_modified[parts[0]] = [(parts[1], kwargs[key])]
        workflow = self.wfmanager.run_workflow(kwargs_modified["workflow_class"], kwargs_modified)
        return self.jsonify_workflow_status(workflow, True)

    @cherrypy.expose
    @jsonify
    def rerun_workflow(self, **kwargs):
        self.wfmanager.rerun_workflow(kwargs["workflow_id"])
        workflow = self.wfmanager.get_workflow(kwargs["workflow_id"])
        return self.jsonify_workflow_status(workflow)

    @cherrypy.expose
    @cherrypy.tools.noBodyProcess()
    @cherrypy.tools.CORS()
    def upload(self):
        # the file transfer can take a long time; by default cherrypy
        # limits responses to 300s; we increase it to 1h
        cherrypy.response.timeout = 3600

        # convert the header keys to lower case
        lcHDRS = {}
        for key, val in cherrypy.request.headers.iteritems():
            lcHDRS[key.lower()] = val

        # at this point we could limit the upload on content-length...
        # incomingBytes = int(lcHDRS['content-length'])

        # create our version of cgi.FieldStorage to parse the MIME encoded
        # form data where the file is contained
        formFields = UploadFieldStorage(fp=cherrypy.request.rfile,
                                        headers=lcHDRS,
                                        environ={'REQUEST_METHOD':'POST'},
                                        keep_blank_values=True)

        # we now create a link to the file, using the submitted
        # filename; if we renamed, there would be a failure because
        # the NamedTemporaryFile, used by our version of cgi.FieldStorage,
        # explicitly deletes the original filename
        for current in formFields.keys():
            if current != 'prefix':
                currentFile = formFields[current]
                os.link(
                        currentFile.file.name,
                        os.path.join(formFields.get_tmp_directory(), formFields.getvalue("prefix") + "_" + currentFile.filename.encode('ascii','ignore'))
                )

    @cherrypy.expose
    @jsonify
    def get_workflows_status(self, **kwargs):
        if kwargs.has_key("workflow_id"):
            workflow = self.wfmanager.get_workflow(kwargs["workflow_id"])
            return self.jsonify_workflow_status(workflow)
        else:
            status = []
            workflows = self.wfmanager.get_workflows()
            for workflow in workflows:
                if kwargs.has_key("metadata_filter"):
                    is_ok = False
                    for wf_meta in workflow.metadata:
                        for metadata in kwargs["metadata_filter"].split(","):
                            if wf_meta == metadata:
                                is_ok = True
                                break
                        if is_ok: break
                    if is_ok: status.append(self.jsonify_workflow_status(workflow))
                else:
                    status.append(self.jsonify_workflow_status(workflow))
            return status

    @cherrypy.expose
    @jsonify
    def get_workflow_outputs(self, **kwargs):
        return self.wfmanager.get_workflow_ouputs(JFlowServer.JFLOW_WDATA, kwargs["workflow_id"])

    @cherrypy.expose
    @jsonify
    def validate_field(self, **kwargs):
        try:
            value_key = None
            for key in kwargs.keys():
                if key != "type" and key != "callback" and key != "_": 
                    value_key = key
                    break
            eval(kwargs["type"])(kwargs[value_key])
            return True
        except Exception, e:
            return str(e)

if __name__ == '__main__':

    parser = argparse.ArgumentParser()
    parser.add_argument("--daemon", action="store_true", dest="daemon", default=False, help="Run the server as daemon")
    args = vars(parser.parse_args())

    # daemonize the server if asked to
    if args["daemon"]:
        from cherrypy.process.plugins import Daemonizer
        Daemonizer(cherrypy.engine).subscribe()
    # define the socket host and port
    jflowconf = JFlowConfigReader()
    socket_opts = jflowconf.get_socket_options()

    # remove any limit on the request body size; cherrypy's default is 100MB
    # (maybe we should just increase it ?)
    cherrypy.server.max_request_body_size = 0

    # increase server socket timeout to 60s; we are more tolerant of bad
    # quality client-server connections (cherrypy's defult is 10s)
    cherrypy.server.socket_timeout = 60

    app_conf = {'/': {'tools.staticdir.root': jflowconf.get_work_directory()},
                os.path.join('/', JFlowServer.JFLOW_WDATA): {'tools.staticdir.on'  : True,
                                                             'tools.staticdir.dir' : jflowconf.get_work_directory()}}

    cherrypy.config.update({'server.socket_host': socket_opts[0],
                            'server.socket_port': socket_opts[1]})
    # start the server
    cherrypy.quickstart(JFlowServer(), config=app_conf)