<commit_before>#!/usr/bin/env python

# Copyright 2016 University of Chicago
# Licensed under the APL 2.0 license

# Resync workflows with pegasus state if possible
import argparse
import sys
import os
import subprocess

import cStringIO
import psycopg2

import fsurfer
import fsurfer.log

PARAM_FILE_LOCATION = "/etc/freesurfer/db_info"
FREESURFER_BASE = '/local-scratch/fsurf/'
FREESURFER_SCRATCH = '/local-scratch/fsurf/scratch'
PEGASUSRC_PATH = '/etc/fsurf/pegasusconf/pegasusrc'
VERSION = fsurfer.__version__
DRY_RUN = False


def reset_workflow(workflow_id):
    """
    Reset workflow to QUEUED state if possible so that it can be retried

    :param workflow_id: workflow id
    :return: exit code for sys.exit
    """

    conn = fsurfer.helpers.get_db_client()
    cursor = conn.cursor()
    if DRY_RUN:
        sys.stdout.write("Resetting workflow {0}\n".format(workflow_id))
    job_run_delete = "DELETE FROM freesurfer_interface.job_run " \
                     "WHERE job_id = %s"
    job_reset = "UPDATE freesurfer_interface.jobs " \
                "SET state = 'QUEUED' " \
                "WHERE id = %s "
    try:
        cursor.execute(job_run_delete, [workflow_id])
        cursor.execute(job_reset, [workflow_id])
        if DRY_RUN:
            conn.rollback()
        else:
            conn.commit()
    finally:
        conn.close()


def resync_workflows():
    """
    Check all workflows running for more than a day and see if they
    need to be resynced

    :return: exit code
    """
    global DRY_RUN
    fsurfer.log.initialize_logging()
    logger = fsurfer.log.get_logger()
    parser = argparse.ArgumentParser(description="Generate and submit "
                                                 "workflows to process jobs")
    # version info
    parser.add_argument('--version', action='version', version='%(prog)s ' + VERSION)
    # Arguments for action
    parser.add_argument('--dry-run', dest='dry_run',
                        action='store_true', default=False,
                        help='Mock actions instead of carrying them out')
    parser.add_argument('--debug', dest='debug',
                        action='store_true', default=False,
                        help='Output debug messages')

    args = parser.parse_args(sys.argv[1:])
    if args.debug:
        fsurfer.log.set_debugging()
    if args.dry_run:
        sys.stdout.write("Doing a dry run, no changes will be made\n")
        DRY_RUN = True

    conn = fsurfer.helpers.get_db_client()
    cursor = conn.cursor()
    job_query = "SELECT id, username, pegasus_ts " \
                "FROM freesurfer_interface.jobs " \
                "WHERE state = 'RUNNING' AND " \
                "      age(job_date) >= '1 day'"
    update_workflow_state = "UPDATE freesurfer_interface.jobs " \
                            "SET state = %s " \
                            "WHERE id = %s "

    try:
        cursor.execute(job_query)
        for row in cursor.fetchall():
            logger.info("Examining workflow {0} started by user {1}".format(row[0], row[1]))
            if row[2] is None:
                logger.error("No workflow directory for running "
                             "workflow {0}".format(row[0]))
                return reset_workflow(row[0])
            workflow_id = row[0]
            scratch_directory = os.path.join(FREESURFER_SCRATCH,
                                             row[1],
                                             'workflows',
                                             'fsurf',
                                             'pegasus',
                                             'freesurfer',
                                             row[2])
            try:
                output = subprocess.check_output(['/usr/bin/pegasus-status',
                                                  '-l',
                                                  scratch_directory],
                                                 stderr=subprocess.STDOUT)
                if 'Failure' in cStringIO.StringIO(output).readlines()[2]:
                    if DRY_RUN:
                        sys.stdout.write("Would have failed "
                                         "workflow {0}\n".format(workflow_id))
                        continue
                    cursor.execute(update_workflow_state, [workflow_id,
                                                           'FAILED'])
                    subprocess.check_call(['/usr/bin/workflow_completed.py',
                                           '--failure',
                                           '--id',
                                           workflow_id])
                elif 'Success' in cStringIO.StringIO(output).readlines()[2]:
                    if DRY_RUN:
                        sys.stdout.write("Would have completed "
                                         "workflow {0}\n".format(workflow_id))
                        continue
                    cursor.execute(update_workflow_state, [workflow_id,
                                                           'COMPLETED'])
                    subprocess.check_call(['/usr/bin/workflow_completed.py',
                                           '--success',
                                           '--id',
                                           workflow_id])
            except subprocess.CalledProcessError as err:
                conn.rollback()
                logger.error("Couldn't run commands: {0}".format(err))
            else:
                conn.rollback()
                logger.error("Rolled back transaction")
        if DRY_RUN:
            conn.rollback()
        else:
            conn.commit()
    except psycopg2.Error as e:
        logger.exception("Got pgsql error: {0}".format(e))
        pass
    finally:
        conn.close()


if __name__ == '__main__':
    # workaround missing subprocess.check_output
    if "check_output" not in dir(subprocess):  # duck punch it in!
        def check_output(*popenargs, **kwargs):
            """
            Run command with arguments and return its output as a byte string.

            Backported from Python 2.7 as it's implemented as pure python
            on stdlib.

            """
            process = subprocess.Popen(stdout=subprocess.PIPE, *popenargs, **kwargs)
            output, unused_err = process.communicate()
            retcode = process.poll()
            if retcode:
                cmd = kwargs.get("args")
                if cmd is None:
                    cmd = popenargs[0]
                error = subprocess.CalledProcessError(retcode, cmd)
                error.output = output
                raise error
            return output

        subprocess.check_output = check_output
    sys.exit(resync_workflows())
<commit_msg>Fix arguments<commit_after>#!/usr/bin/env python

# Copyright 2016 University of Chicago
# Licensed under the APL 2.0 license

# Resync workflows with pegasus state if possible
import argparse
import sys
import os
import subprocess

import cStringIO
import psycopg2

import fsurfer
import fsurfer.log

PARAM_FILE_LOCATION = "/etc/freesurfer/db_info"
FREESURFER_BASE = '/local-scratch/fsurf/'
FREESURFER_SCRATCH = '/local-scratch/fsurf/scratch'
PEGASUSRC_PATH = '/etc/fsurf/pegasusconf/pegasusrc'
VERSION = fsurfer.__version__
DRY_RUN = False


def reset_workflow(workflow_id):
    """
    Reset workflow to QUEUED state if possible so that it can be retried

    :param workflow_id: workflow id
    :return: exit code for sys.exit
    """

    conn = fsurfer.helpers.get_db_client()
    cursor = conn.cursor()
    if DRY_RUN:
        sys.stdout.write("Resetting workflow {0}\n".format(workflow_id))
    job_run_delete = "DELETE FROM freesurfer_interface.job_run " \
                     "WHERE job_id = %s"
    job_reset = "UPDATE freesurfer_interface.jobs " \
                "SET state = 'QUEUED' " \
                "WHERE id = %s "
    try:
        cursor.execute(job_run_delete, [workflow_id])
        cursor.execute(job_reset, [workflow_id])
        if DRY_RUN:
            conn.rollback()
        else:
            conn.commit()
    finally:
        conn.close()


def resync_workflows():
    """
    Check all workflows running for more than a day and see if they
    need to be resynced

    :return: exit code
    """
    global DRY_RUN
    fsurfer.log.initialize_logging()
    logger = fsurfer.log.get_logger()
    parser = argparse.ArgumentParser(description="Generate and submit "
                                                 "workflows to process jobs")
    # version info
    parser.add_argument('--version', action='version', version='%(prog)s ' + VERSION)
    # Arguments for action
    parser.add_argument('--dry-run', dest='dry_run',
                        action='store_true', default=False,
                        help='Mock actions instead of carrying them out')
    parser.add_argument('--debug', dest='debug',
                        action='store_true', default=False,
                        help='Output debug messages')

    args = parser.parse_args(sys.argv[1:])
    if args.debug:
        fsurfer.log.set_debugging()
    if args.dry_run:
        sys.stdout.write("Doing a dry run, no changes will be made\n")
        DRY_RUN = True

    conn = fsurfer.helpers.get_db_client()
    cursor = conn.cursor()
    job_query = "SELECT id, username, pegasus_ts " \
                "FROM freesurfer_interface.jobs " \
                "WHERE state = 'RUNNING' AND " \
                "      age(job_date) >= '1 day'"
    update_workflow_state = "UPDATE freesurfer_interface.jobs " \
                            "SET state = %s " \
                            "WHERE id = %s "

    try:
        cursor.execute(job_query)
        for row in cursor.fetchall():
            logger.info("Examining workflow {0} started by user {1}".format(row[0], row[1]))
            if row[2] is None:
                logger.error("No workflow directory for running "
                             "workflow {0}".format(row[0]))
                return reset_workflow(row[0])
            workflow_id = row[0]
            scratch_directory = os.path.join(FREESURFER_SCRATCH,
                                             row[1],
                                             'workflows',
                                             'fsurf',
                                             'pegasus',
                                             'freesurfer',
                                             row[2])
            try:
                output = subprocess.check_output(['/usr/bin/pegasus-status',
                                                  '-l',
                                                  scratch_directory],
                                                 stderr=subprocess.STDOUT)
                if 'Failure' in cStringIO.StringIO(output).readlines()[2]:
                    if DRY_RUN:
                        sys.stdout.write("Would have failed "
                                         "workflow {0}\n".format(workflow_id))
                        continue
                    cursor.execute(update_workflow_state, ['FAILED',
                                                           workflow_id])
                    subprocess.check_call(['/usr/bin/workflow_completed.py',
                                           '--failure',
                                           '--id',
                                           workflow_id])
                elif 'Success' in cStringIO.StringIO(output).readlines()[2]:
                    if DRY_RUN:
                        sys.stdout.write("Would have completed "
                                         "workflow {0}\n".format(workflow_id))
                        continue
                    cursor.execute(update_workflow_state, ['COMPLETED',
                                                           workflow_id])
                    subprocess.check_call(['/usr/bin/workflow_completed.py',
                                           '--success',
                                           '--id',
                                           workflow_id])
            except subprocess.CalledProcessError as err:
                conn.rollback()
                logger.error("Couldn't run commands: {0}".format(err))
            else:
                conn.rollback()
                logger.error("Rolled back transaction")
        if DRY_RUN:
            conn.rollback()
        else:
            conn.commit()
    except psycopg2.Error as e:
        logger.exception("Got pgsql error: {0}".format(e))
        pass
    finally:
        conn.close()


if __name__ == '__main__':
    # workaround missing subprocess.check_output
    if "check_output" not in dir(subprocess):  # duck punch it in!
        def check_output(*popenargs, **kwargs):
            """
            Run command with arguments and return its output as a byte string.

            Backported from Python 2.7 as it's implemented as pure python
            on stdlib.

            """
            process = subprocess.Popen(stdout=subprocess.PIPE, *popenargs, **kwargs)
            output, unused_err = process.communicate()
            retcode = process.poll()
            if retcode:
                cmd = kwargs.get("args")
                if cmd is None:
                    cmd = popenargs[0]
                error = subprocess.CalledProcessError(retcode, cmd)
                error.output = output
                raise error
            return output

        subprocess.check_output = check_output
    sys.exit(resync_workflows())

{'fraction_non_alphanumeric': 0.09234167893961709, 'fraction_numerical': 0.004565537555228277, 'mean_word_length': 1.3685036623648412, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 80, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '90560', 'n_tokens_mistral': 3580, 'n_tokens_neox': 3423, 'n_words': 801}