""" This script calculates flow timings for a given date. Only the runs which has data processing happened are included. It generates tab-separated output with the following columns: - start_time - close_time - duration (in seconds) - feed_name - context_date - dbt_triggered (DBT if DBT was triggered with the run) - url (link to SWF run in AWS console) """ import argparse from collections import namedtuple import datetime import itertools import json import logging from pathlib import Path import sys from feed_ingestion.bin import seat from feed_ingestion.util.aws import swf COLUMN_SEPARATOR = ',' logger = logging.getLogger(__name__) THIS_DIR = Path(__file__).parent CACHE_DIR = THIS_DIR / 'cache' SWF_DOMAIN_PROD = 'prod_swf_feed_ingestion' SWF_DOMAIN_QA = 'qa_swf_feed_ingestion' ExecutionInfo = namedtuple( typename='ExecutionInfo', field_names=[ 'start_time', 'close_time', 'duration', 'feed_name', 'context_date', 'dbt_triggered', 'url' ] ) COMPLETION_INDICATOR_TASK_NAMES = [ # 'set_overall_status_ingested', # apple # 'set_status_ingested', # spotify and others 'set_status_ingested_to_fact_analytics_report', # apple 'load_fact_tables', 'load_fact_analytics', 'load_fact_table', ] FLOWS = [ 'spotify', 'amazon_music', 'apple_music', # 'soundcloud', # 'rhapsody', 'pandora', 'deezer_daily', ] def list_executions(workflow_name, dates: list, swf_domain=SWF_DOMAIN_PROD): """List executions for a given workflow and date range.""" date_from, date_to = dates logger.info(f'Loading executions for ' f'{workflow_name} from {date_from} to {date_to}') executions = swf.list_closed_swf_executions( from_=date_from, to=date_to, swf_domain=swf_domain, execution_type=workflow_name, ) logger.info(f'Loaded {len(executions)} executions') return executions def list_events(execution, swf_domain=SWF_DOMAIN_PROD): """List events for a given execution.""" logger.info( f'Loading events for ' f'{execution["execution"]["workflowId"]} ' f'{execution["execution"]["runId"]}') events = swf.load_all_execution_events( swf_domain=swf_domain, workflow_id=execution['execution']['workflowId'], run_id=execution['execution']['runId'], ) logger.info(f'Loaded {len(events)} events') return events def datetime_to_utc_and_truncate(dt): """Convert datetime to UTC and truncate to minutes.""" dt = dt.astimezone(datetime.timezone.utc) dt = dt.replace(second=0, microsecond=0) return dt def first_not_empty(gen): """Return first not empty value from generator.""" for v in gen: if v: return v def get_feed_name_and_date(tasks): """Get feed name and date from tasks.""" bootstrap_task_names = [ 'bootstrap', 'bootstrap_feed' ] bootstrap = first_not_empty( find_tasks_by_name(task_name, tasks) for task_name in bootstrap_task_names)[0] bootstrap_result = get_task_result(bootstrap) feed_name_properties = [ 'feed_name', 'facts_feed_name', 'feed_name_for_fact_analytics' ] bootstrap_date = first_not_empty( bootstrap_result.get(f'{prefix}.date') for prefix in bootstrap_task_names) feed_name = first_not_empty( bootstrap_result.get(f'{prefix}.{feed_name}') for prefix, feed_name in itertools.product(bootstrap_task_names, feed_name_properties)) return feed_name, bootstrap_date def find_tasks_by_name(task_name, tasks): """Find tasks by name.""" return [task for task in tasks if task['name'].endswith(f'_{task_name}')] def get_task_result(task): """Extract result from task.""" bootstrap_result_str = ( task['closed'] ['activityTaskCompletedEventAttributes'] ['result']) bootstrap_result = json.loads(bootstrap_result_str) return bootstrap_result def workflow_name_from_feed_name(feed_name): """Convert feed name to workflow name.""" return f'{feed_name}_feed_ingestion' def find_completion_tasks(tasks): """Filter completed tasks.""" return any( find_tasks_by_name(task_name, tasks) for task_name in COMPLETION_INDICATOR_TASK_NAMES ) def main(swf_domain, date, flows): """Run main process.""" result = [] timestamp = datetime.datetime.combine(date, datetime.datetime.min.time()) for flow_name in flows: workflow_name = workflow_name_from_feed_name(flow_name) executions = list_executions( workflow_name=workflow_name, dates=[ timestamp, timestamp + seat.ONE_DAY, ], swf_domain=swf_domain, ) for execution in executions: events = list_events(execution, swf_domain=swf_domain) tasks = swf.tasks_from_events(events) if not find_completion_tasks(tasks): continue dbt_trigger = process_dbt_triggerer(date, execution, tasks) flow_name, context_date = get_feed_name_and_date(tasks) start_time = datetime_to_utc_and_truncate( execution['startTimestamp']) close_time = datetime_to_utc_and_truncate( execution['closeTimestamp']) url = swf.generate_execution_console_url( swf_domain=swf_domain, workflow_id=execution['execution']['workflowId'], run_id=execution['execution']['runId'], ) data = ExecutionInfo( start_time=start_time, close_time=close_time, duration=close_time - start_time, feed_name=flow_name, context_date=context_date, url=url, dbt_triggered='DBT' if dbt_trigger else '' ) result.append(data) logger.info(f'{close_time} {flow_name} {context_date} {url}') result.sort(key=lambda x: x.close_time) print(COLUMN_SEPARATOR.join(ExecutionInfo._fields)) for data in result: values = [ v.strftime('%Y-%m-%d %H:%M') if isinstance(v, datetime.datetime) else str(v) for v in data ] print(COLUMN_SEPARATOR.join(values)) def process_dbt_triggerer(date, execution, tasks): """Process DBT triggerer.""" dbt_trigger = find_tasks_by_name('build_jenkins_dbt', tasks) if dbt_trigger: assert len(dbt_trigger) == 1 dbt_trigger_result = get_task_result(dbt_trigger[0]) if date < datetime.date(2024, 1, 25): dbt_run_condition = not dbt_trigger_result.get( 'build_jenkins_dbt.stop') else: dbt_run_condition = dbt_trigger_result.get( 'build_jenkins_dbt.job_url') if not dbt_run_condition: logger.info(f"DBT trigger execution didn't happen, " f'reason: {dbt_trigger_result}') dbt_trigger = None else: logger.info(f'!!!DBT triggered by ' f'{execution["execution"]["workflowId"]}') return dbt_trigger def parseargs(argv): """Parse command line arguments.""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( '--swf-domain', default=SWF_DOMAIN_PROD, type=str, ) parser.add_argument( '--date', type=datetime.date.fromisoformat, required=True ) parser.add_argument( '--flows', default=FLOWS, type=lambda t: [s.strip() for s in t.split(',')], ) parser.add_argument('--cache', action='store_true') parser.add_argument('--no-cache', dest='cache', action='store_false') parser.set_defaults(cache=False) args = parser.parse_args(argv) return args if __name__ == '__main__': logging.basicConfig( level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s', ) args = parseargs(sys.argv[1:]) if args.cache: logger.info(f'Use cache for SWF calls. Dir: {CACHE_DIR}') from joblib import Memory CACHE_DIR.mkdir(exist_ok=True) cache = Memory(CACHE_DIR, verbose=0) # we can safely cache list_events, because we # request them only for COMPLETED executions, # so they cannot be changed list_events = cache.cache(list_events) if args.date < datetime.date.today() - seat.ONE_DAY: # do not cache list_executions for today and yesterday, # as they still can be updated list_executions = cache.cache(list_executions) main( swf_domain=args.swf_domain, date=args.date, flows=args.flows, )