"""CLI functions to run decider, worker and exec processes.""" import argparse from datetime import datetime from datetime import timedelta import importlib import json import logging import time import boto.swf.layer1 as swf_layer1 import boto.swf.layer2 as swf from garcon import activity from garcon import decider from analytics_aggregation import flows def execute_flow(flow, context, output_file=None, **kwargs): """Launch the workflow execution. Args: flow (module): Garcon flow module. context (str): Initial context parsed from JSON. output_file: Optional path to file to store all args as well as workflow_id and run_id. kwargs: Extensible call API. """ active_executions = swf_layer1.Layer1().list_open_workflow_executions( flow.domain, (datetime.now() - timedelta(days=1)).timestamp(), workflow_id=flow.name) if len(active_executions.get('executionInfos')): print( 'Workflow {flow_name} already started, exit'.format( flow_name=flow.name)) return start_kwargs = dict( workflow_id=flow.workflow_id(json.loads(context)), input=context) if hasattr(flow, 'timeout'): start_kwargs.update(execution_start_to_close_timeout=str(flow.timeout)) execution = swf.WorkflowType( name=flow.name, domain=flow.domain, version=flow.version, task_list=flow.name).start(**start_kwargs) if output_file: with open(output_file, 'w') as out_file: start_kwargs.update(run_id=execution.runId) json.dump(start_kwargs, out_file) return execution def run_decider(flow, **kwargs): """Launch the SWF decider process. Args: flow (module): Garcon flow module. kwargs (dict): Extensible call API. """ worker = decider.DeciderWorker(flow) while True: worker.run() time.sleep(1) def run_activity_worker(flow, **kwargs): """Launch the activity worker process. Args: flow (module): Garcon flow module. kwargs (dict): Extensible call API. """ worker = activity.ActivityWorker(flow) worker.run() _COMMANDS = {'exec': execute_flow, 'decider': run_decider, 'worker': run_activity_worker} def garcon(*args): """Main entry point for the Garcon command line integration. Args: args (tuple): Args for the flow. """ parser = argparse.ArgumentParser(description='Garcon command line util') parser.add_argument('cmd', choices=_COMMANDS.keys(), help='garcon command') parser.add_argument('flow', choices=flows.__all__, help='name of the flow') parser.add_argument('-c', '--context', help='initial context [json]') parser.add_argument('-l', '--log_level', help='logging level', default='error', choices=['critical', 'error', 'warning', 'info', 'debug']) parser.add_argument('-o', '--output_file', help='full path to output file, if applicable') # parse cl args (allows easy unit testing) args = parser.parse_args(args) if args else parser.parse_args() # import the flow module flow = importlib.import_module( '.{}.flow'.format(args.flow), flows.__name__) # look for the flow class, if it exists, otherwise fall back # on the flow module itself. flow_class = getattr(flow, 'Flow', None) if flow_class: flow = flow_class() # execute command args.context = args.context or '{}' logging.basicConfig(level=getattr(logging, args.log_level.upper())) _COMMANDS[args.cmd](flow=flow, context=args.context, output_file=args.output_file)