<commit_before>from builtins import str
from builtins import range
from builtins import object
import logging
import datetime
import os
import shutil
import tempfile
import re
import traceback
import json
import hashlib

from biomaj_core.utils import Utils
from biomaj_download.downloadclient import DownloadClient
from biomaj_download.message import message_pb2
from biomaj_download.download.http import HTTPParse
from biomaj_download.download.localcopy import LocalDownload

from biomaj.mongo_connector import MongoConnector
from biomaj.options import Options

from biomaj.process.processfactory import RemoveProcessFactory, PreProcessFactory, PostProcessFactory

from biomaj_zipkin.zipkin import Zipkin


class Workflow(object):
    """
    Bank update workflow
    """

    FLOW_INIT = 'init'
    FLOW_CHECK = 'check'
    FLOW_DEPENDS = 'depends'
    FLOW_PREPROCESS = 'preprocess'
    FLOW_RELEASE = 'release'
    FLOW_DOWNLOAD = 'download'
    FLOW_POSTPROCESS = 'postprocess'
    FLOW_REMOVEPROCESS = 'removeprocess'
    FLOW_PUBLISH = 'publish'
    FLOW_OVER = 'over'

    FLOW = [
        {'name': 'init', 'steps': []},
        {'name': 'check', 'steps': []},
        {'name': 'over', 'steps': []}
    ]

    def __init__(self, bank, session=None):
        """
        Instantiate a new workflow

        :param bank: bank on which to apply the workflow
        :type bank: :class:`biomaj.bank.Bank`
        """
        self.bank = bank
        if session is None:
            self.session = bank.session
        else:
            self.session = session
            self.bank.session = session
        self.options = bank.options
        self.name = bank.name
        # Skip all remaining tasks, no need to update
        self.skip_all = False

        self.session._session['update'] = False
        self.session._session['remove'] = False
        self.session.config.set('localrelease', '')
        self.session.config.set('remoterelease', '')
        # For micro services
        self.redis_client = None
        self.redis_prefix = None
        # Zipkin
        self.span = None

    def get_flow(self, task):
        for flow in Workflow.FLOW:
            if flow['name'] == task:
                return flow

    def start(self):
        """
        Start the workflow
        """
        logging.info('Workflow:Start')
        for flow in self.session.flow:
            if self.skip_all:
                logging.info('Workflow:Skip:' + flow['name'])
                self.session._session['status'][flow['name']] = None
                self.session._session['status'][Workflow.FLOW_OVER] = True
                continue

            if self.options.get_option(Options.STOP_BEFORE) == flow['name']:
                self.wf_over()
                break

            # Check for cancel request
            if self.redis_client and self.redis_client.get(self.redis_prefix + ':' + self.bank.name + ':action:cancel'):
                logging.warn('Cancel requested, stopping update')
                self.redis_client.delete(self.redis_prefix + ':' + self.bank.name + ':action:cancel')
                self.wf_over()
                return False

            # Always run INIT
            if flow['name'] != Workflow.FLOW_INIT and self.session.get_status(flow['name']):
                logging.info('Workflow:Skip:' + flow['name'])
            if flow['name'] == Workflow.FLOW_INIT or not self.session.get_status(flow['name']):
                logging.info('Workflow:Start:' + flow['name'])
                span = None
                if self.options.get_option('traceId'):
                    trace_id = self.options.get_option('traceId')
                    span_id = self.options.get_option('spanId')
                    span = Zipkin('biomaj-workflow', flow['name'], trace_id=trace_id, parent_id=span_id)
                    self.span = span
                    self.bank.config.set('zipkin_trace_id', span.get_trace_id())
                    self.bank.config.set('zipkin_span_id', span.get_span_id())

                try:
                    self.session._session['status'][flow['name']] = getattr(self, 'wf_' + flow['name'])()
                except Exception as e:
                    self.session._session['status'][flow['name']] = False
                    logging.exception('Workflow:' + flow['name'] + ':Exception:' + str(e))
                    logging.debug(traceback.format_exc())
                finally:
                    self.wf_progress(flow['name'], self.session._session['status'][flow['name']])

                if span:
                    span.add_binary_annotation('status', str(self.session._session['status'][flow['name']]))
                    span.trace()

                if flow['name'] != Workflow.FLOW_OVER and not self.session.get_status(flow['name']):
                    logging.error('Error during task ' + flow['name'])
                    if flow['name'] != Workflow.FLOW_INIT:
                        self.wf_over()
                    return False
                # Main task is over, execute sub tasks of main
                if not self.skip_all:
                    for step in flow['steps']:
                        span = None
                        try:
                            # Check for cancel request
                            if self.redis_client and self.redis_client.get(self.redis_prefix + ':' + self.bank.name + ':action:cancel'):
                                logging.warn('Cancel requested, stopping update')
                                self.redis_client.delete(self.redis_prefix + ':' + self.bank.name + ':action:cancel')
                                self.wf_over()
                                return False

                            if self.options.get_option('traceId'):
                                trace_id = self.options.get_option('traceId')
                                span_id = self.options.get_option('spanId')
                                span = Zipkin('biomaj-workflow', flow['name'] + ":wf_" + step, trace_id=trace_id, parent_id=span_id)
                                self.span = None
                            res = getattr(self, 'wf_' + step)()

                            if span:
                                span.add_binary_annotation('status', str(res))
                                span.trace()

                            if not res:
                                logging.error('Error during ' + flow['name'] + ' subtask: wf_' + step)
                                logging.error('Revert main task status ' + flow['name'] + ' to error status')
                                self.session._session['status'][flow['name']] = False
                                self.wf_over()
                                return False
                        except Exception as e:
                            logging.error('Workflow:' + flow['name'] + ' subtask: wf_' + step + ':Exception:' + str(e))
                            self.session._session['status'][flow['name']] = False
                            logging.debug(traceback.format_exc())
                            self.wf_over()
                            return False
            if self.options.get_option(Options.STOP_AFTER) == flow['name']:
                self.wf_over()
                break
        self.wf_progress_end()
        return True<commit_msg>set span for sub workflow step too<commit_after>from builtins import str
from builtins import range
from builtins import object
import logging
import datetime
import os
import shutil
import tempfile
import re
import traceback
import json
import hashlib

from biomaj_core.utils import Utils
from biomaj_download.downloadclient import DownloadClient
from biomaj_download.message import message_pb2
from biomaj_download.download.http import HTTPParse
from biomaj_download.download.localcopy import LocalDownload

from biomaj.mongo_connector import MongoConnector
from biomaj.options import Options

from biomaj.process.processfactory import RemoveProcessFactory, PreProcessFactory, PostProcessFactory

from biomaj_zipkin.zipkin import Zipkin


class Workflow(object):
    """
    Bank update workflow
    """

    FLOW_INIT = 'init'
    FLOW_CHECK = 'check'
    FLOW_DEPENDS = 'depends'
    FLOW_PREPROCESS = 'preprocess'
    FLOW_RELEASE = 'release'
    FLOW_DOWNLOAD = 'download'
    FLOW_POSTPROCESS = 'postprocess'
    FLOW_REMOVEPROCESS = 'removeprocess'
    FLOW_PUBLISH = 'publish'
    FLOW_OVER = 'over'

    FLOW = [
        {'name': 'init', 'steps': []},
        {'name': 'check', 'steps': []},
        {'name': 'over', 'steps': []}
    ]

    def __init__(self, bank, session=None):
        """
        Instantiate a new workflow

        :param bank: bank on which to apply the workflow
        :type bank: :class:`biomaj.bank.Bank`
        """
        self.bank = bank
        if session is None:
            self.session = bank.session
        else:
            self.session = session
            self.bank.session = session
        self.options = bank.options
        self.name = bank.name
        # Skip all remaining tasks, no need to update
        self.skip_all = False

        self.session._session['update'] = False
        self.session._session['remove'] = False
        self.session.config.set('localrelease', '')
        self.session.config.set('remoterelease', '')
        # For micro services
        self.redis_client = None
        self.redis_prefix = None
        # Zipkin
        self.span = None

    def get_flow(self, task):
        for flow in Workflow.FLOW:
            if flow['name'] == task:
                return flow

    def start(self):
        """
        Start the workflow
        """
        logging.info('Workflow:Start')
        for flow in self.session.flow:
            if self.skip_all:
                logging.info('Workflow:Skip:' + flow['name'])
                self.session._session['status'][flow['name']] = None
                self.session._session['status'][Workflow.FLOW_OVER] = True
                continue

            if self.options.get_option(Options.STOP_BEFORE) == flow['name']:
                self.wf_over()
                break

            # Check for cancel request
            if self.redis_client and self.redis_client.get(self.redis_prefix + ':' + self.bank.name + ':action:cancel'):
                logging.warn('Cancel requested, stopping update')
                self.redis_client.delete(self.redis_prefix + ':' + self.bank.name + ':action:cancel')
                self.wf_over()
                return False

            # Always run INIT
            if flow['name'] != Workflow.FLOW_INIT and self.session.get_status(flow['name']):
                logging.info('Workflow:Skip:' + flow['name'])
            if flow['name'] == Workflow.FLOW_INIT or not self.session.get_status(flow['name']):
                logging.info('Workflow:Start:' + flow['name'])
                span = None
                if self.options.get_option('traceId'):
                    trace_id = self.options.get_option('traceId')
                    span_id = self.options.get_option('spanId')
                    span = Zipkin('biomaj-workflow', flow['name'], trace_id=trace_id, parent_id=span_id)
                    self.span = span
                    self.bank.config.set('zipkin_trace_id', span.get_trace_id())
                    self.bank.config.set('zipkin_span_id', span.get_span_id())

                try:
                    self.session._session['status'][flow['name']] = getattr(self, 'wf_' + flow['name'])()
                except Exception as e:
                    self.session._session['status'][flow['name']] = False
                    logging.exception('Workflow:' + flow['name'] + ':Exception:' + str(e))
                    logging.debug(traceback.format_exc())
                finally:
                    self.wf_progress(flow['name'], self.session._session['status'][flow['name']])

                if span:
                    span.add_binary_annotation('status', str(self.session._session['status'][flow['name']]))
                    span.trace()

                if flow['name'] != Workflow.FLOW_OVER and not self.session.get_status(flow['name']):
                    logging.error('Error during task ' + flow['name'])
                    if flow['name'] != Workflow.FLOW_INIT:
                        self.wf_over()
                    return False
                # Main task is over, execute sub tasks of main
                if not self.skip_all:
                    for step in flow['steps']:
                        span = None
                        try:
                            # Check for cancel request
                            if self.redis_client and self.redis_client.get(self.redis_prefix + ':' + self.bank.name + ':action:cancel'):
                                logging.warn('Cancel requested, stopping update')
                                self.redis_client.delete(self.redis_prefix + ':' + self.bank.name + ':action:cancel')
                                self.wf_over()
                                return False

                            if self.options.get_option('traceId'):
                                trace_id = self.options.get_option('traceId')
                                span_id = self.options.get_option('spanId')
                                span = Zipkin('biomaj-workflow', flow['name'] + ":wf_" + step, trace_id=trace_id, parent_id=span_id)
                                self.span = span
                            res = getattr(self, 'wf_' + step)()

                            if span:
                                span.add_binary_annotation('status', str(res))
                                span.trace()

                            if not res:
                                logging.error('Error during ' + flow['name'] + ' subtask: wf_' + step)
                                logging.error('Revert main task status ' + flow['name'] + ' to error status')
                                self.session._session['status'][flow['name']] = False
                                self.wf_over()
                                return False
                        except Exception as e:
                            logging.error('Workflow:' + flow['name'] + ' subtask: wf_' + step + ':Exception:' + str(e))
                            self.session._session['status'][flow['name']] = False
                            logging.debug(traceback.format_exc())
                            self.wf_over()
                            return False
            if self.options.get_option(Options.STOP_AFTER) == flow['name']:
                self.wf_over()
                break
        self.wf_progress_end()
        return True
{'fraction_non_alphanumeric': 0.11019770364834568, 'fraction_numerical': 0.00013587879611386643, 'mean_word_length': 1.6536866774833243, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '66109', 'n_tokens_mistral': 3998, 'n_tokens_neox': 3766, 'n_words': 823}