def analyze_app(
        app, env, account_id,
        bucket, prefix, store_dir,
        resources, ipdb, ipranges,
        start, end, tz,
        sink, period, sample_count,
        debug):
    """Analyze flow log records for application and generate metrics per period"""
    logging.basicConfig(level=logging.INFO)
    logging.getLogger('botocore').setLevel(logging.WARNING)

    executor = debug and MainThreadExecutor or ThreadPoolExecutor
    start, end = get_dates(start, end, tz)
    resolver = IPResolver(ipdb, ipdb, ipranges)

    for rtype_name in resources:
        rtype = Resource.get_type(rtype_name)
        resource_map = {
            rtype.id(r): r for r
            in rtype.get_resources(ipdb, start, end, app, env)}
        log.info("App:%s Env:%s Type:%s Found:%d",
                 app, env, rtype_name, len(resource_map))

        with sqlite3.connect(ipdb) as db:
            db.row_factory = row_factory
            cursor = db.cursor()
            cursor.execute(
                'select * from enis where resource_type in (%s)' % (
                    ", ".join(["'%s'" % r for r in resource_map.keys()])))
            enis = list(cursor)
            eni_map = {e['eni_id']: e for e in enis}

        # TODO: Download should be doing date bits here across the range of days.
        log_prefix = "%s/%s/flow-log/%s/%s" % (
            prefix.rstrip('/'),
            account_id,
            start.strftime('%Y/%m/%d'),
            "00000000-0000-0000-0000-000000000000")

        f_downloads = {}
        f_metrics = {}
        files = {}

        # should probably just queue this out to distributed worker pool
        with executor(max_workers=5) as w:
            client = boto3.client('s3')
            for e in enis:
                f_downloads[
                    w.submit(
                        eni_download_flows,
                        client, bucket,
                        log_prefix, start, end,
                        e['eni_id'], store_dir)] = e

            for f in as_completed(f_downloads):
                if f.exception():
                    log.warning(
                        "error processing eni %s download: %s",
                        eni_map[f_downloads[f]],
                        f.exception())
                    continue
                e = f_downloads[f]
                files[e['eni_id']] = f.result()

            ipset = {e['ip_address'] for e in enis}

            for eni_id, files in files.items():
                stream = eni_flow_stream(files, start, end)
                f_metrics[w.submit(
                    process_eni_metrics,
                    eni_map[eni_id], ipset,
                    stream,
                    start, end, period, sample_count,
                    resolver, sink)] = eni_id

            for f in as_completed(f_metrics):
                if f.exception():
                    log.warning(
                        "error processing eni %s download %s",
                        eni_map[f_metrics[f]],
                        f.exception())
                    continue
def analyze_app ( app , env , account_id , bucket , prefix , store_dir , resources , ipdb , ipranges , start , end , tz , sink , period , sample_count , debug ) : logging . basicConfig ( level = logging . INFO ) logging . getLogger ( 'botocore' ) . setLevel ( logging . WARNING ) executor = debug and MainThreadExecutor or ThreadPoolExecutor start , end = get_dates ( start , end , tz ) resolver = IPResolver ( ipdb , ipdb , ipranges ) for rtype_name in resources : rtype = Resource . get_type ( rtype_name ) resource_map = { rtype . id ( r ) : r for r in rtype . get_resources ( ipdb , start , end , app , env ) } log . info ( "App:%s Env:%s Type:%s Found:%d" , app , env , rtype_name , len ( resource_map ) ) with sqlite3 . connect ( ipdb ) as db : db . row_factory = row_factory cursor = db . cursor ( ) cursor . execute ( 'select * from enis where resource_type in (%s)' % ( ", " . join ( [ "'%s'" % r for r in resource_map . keys ( ) ] ) ) ) enis = list ( cursor ) eni_map = { e [ 'eni_id' ] : e for e in enis } log_prefix = "%s/%s/flow-log/%s/%s" % ( prefix . rstrip ( '/' ) , account_id , start . strftime ( '%Y/%m/%d' ) , "00000000-0000-0000-0000-000000000000" ) f_downloads = { } f_metrics = { } files = { } with executor ( max_workers = 5 ) as w : client = boto3 . client ( 's3' ) for e in enis : f_downloads [ w . submit ( eni_download_flows , client , bucket , log_prefix , start , end , e [ 'eni_id' ] , store_dir ) ] = e for f in as_completed ( f_downloads ) : if f . exception ( ) : log . warning ( "error processing eni %s download: %s" , eni_map [ f_downloads [ f ] ] , f . exception ( ) ) continue e = f_downloads [ f ] files [ e [ 'eni_id' ] ] = f . result ( ) ipset = { e [ 'ip_address' ] for e in enis } for eni_id , files in files . items ( ) : stream = eni_flow_stream ( files , start , end ) f_metrics [ w . submit ( process_eni_metrics , eni_map [ eni_id ] , ipset , stream , start , end , period , sample_count , resolver , sink ) ] = eni_id for f in as_completed ( f_metrics ) : if f . exception ( ) : log . warning ( "error processing eni %s download %s" , eni_map [ f_metrics [ f ] ] , f . exception ( ) ) continue
Analyze flow log records for application and generate metrics per period