src/dirbs/reports/stats_generator.py
def _refresh_data(config, conn, month, year, statsd, metrics_run_root, run_id, debug_query_performance):
    """Refreshes reporting stats from the DB and stores aggregated results into the various reporting tables."""
    logger = logging.getLogger('dirbs.report')
    nworkers = config.multiprocessing_config.max_db_connections
    db_config = config.db_config
    # Sort conditions by primary and secondary key
    condition_tuples = [ConditionTuple(x.label, x.blocking) for x in config.conditions]
    condition_tuples = _sort_conditions(condition_tuples)
    if len(condition_tuples) == 0:
        logger.warning('No conditions defined in config: No stats on classification will be generated in the report')

    # We need to get the list of operators out of the config
    operators = [op.id for op in config.region_config.operators]
    # Calculate days in month
    days_in_month = (datetime.date(year, month, 1) + relativedelta.relativedelta(months=1, days=-1)).day
    # Init variables for storing data
    per_operator_record_counts = defaultdict(int)
    per_operator_monthly_stats = defaultdict(lambda: defaultdict(int))
    per_operator_daily_stats = defaultdict(lambda: [defaultdict(int) for i in range(0, days_in_month)])
    per_operator_top_model_imei_counts = {}
    per_operator_top_model_gross_adds = {}
    per_operator_imei_imsi_overloading = {}
    per_operator_imsi_imei_overloading = {}
    per_operator_condition_counts = {}
    per_operator_tac_compliance_data = {}
    per_operator_compliance_data = {}
    per_operator_daily_imei_imsi_overloading = {}

    # We use the per-operator record counts to normalize performance numbers, so we need to do this first
    # in a separate executor
    with futures.ProcessPoolExecutor(max_workers=nworkers) as executor:
        logger.info('Simultaneously calculating data volume for each operator using {0:d} workers...'
                    .format(nworkers))
        logger.info('Queueing jobs to calculate monthly record counts...')
        futures_to_cb = {}
        _queue_record_count_jobs(executor, futures_to_cb, per_operator_record_counts, db_config,
                                 operators, month, year, statsd, metrics_run_root, debug_query_performance)

        # Process futures as they are completed, calling the associated callback passing the
        # future as the only argument (other arguments to the callback get partially applied
        # during the queue* functions above)
        for f in futures.as_completed(futures_to_cb):
            futures_to_cb[f](f)

        logger.info('Simultaneously calculating report data using {0:d} workers...'
                    .format(nworkers))
        logger.info('Queueing jobs to calculate stats...')
        futures_to_cb = {}
        _queue_compliance_jobs(executor, futures_to_cb, per_operator_condition_counts,
                               per_operator_tac_compliance_data, per_operator_compliance_data,
                               per_operator_monthly_stats, db_config, operators, month, year, condition_tuples,
                               per_operator_record_counts, statsd, metrics_run_root, debug_query_performance, run_id)
        _queue_imsi_imei_overloading_jobs(executor, futures_to_cb, per_operator_imsi_imei_overloading,
                                          db_config, operators, month, year, per_operator_record_counts,
                                          statsd, metrics_run_root, debug_query_performance)
        _queue_imei_imsi_overloading_jobs(executor, futures_to_cb, per_operator_imei_imsi_overloading,
                                          db_config, operators, month, year, per_operator_record_counts,
                                          statsd, metrics_run_root, debug_query_performance)
        _queue_daily_imei_imsi_overloading_jobs(executor, futures_to_cb, per_operator_daily_imei_imsi_overloading,
                                                db_config, operators, month, year, per_operator_record_counts,
                                                statsd, metrics_run_root, debug_query_performance)
        _queue_monthly_stats_jobs(executor, futures_to_cb, per_operator_monthly_stats, db_config, operators,
                                  month, year, per_operator_record_counts, statsd, metrics_run_root,
                                  debug_query_performance)
        _queue_top_model_gross_adds_jobs(executor, futures_to_cb, per_operator_top_model_gross_adds,
                                         db_config, operators, month, year, per_operator_record_counts, statsd,
                                         metrics_run_root, debug_query_performance)
        _queue_top_model_imei_jobs(executor, futures_to_cb, per_operator_top_model_imei_counts,
                                   db_config, operators, month, year, per_operator_record_counts,
                                   statsd, metrics_run_root, debug_query_performance)
        _queue_distinct_id_counts_jobs(executor, futures_to_cb, per_operator_monthly_stats, per_operator_daily_stats,
                                       db_config, operators, month, year, per_operator_record_counts, statsd,
                                       metrics_run_root, debug_query_performance)
        logger.info('Queued jobs to calculate stats. Processing will begin now...')

        # Process futures as they are completed, calling the associated callback passing the
        # future as the only argument (other arguments to the callback get partially applied
        # during the queue* functions above)
        for f in futures.as_completed(futures_to_cb):
            futures_to_cb[f](f)

    # The hll relative error is given by the expression ±1.04/√(2 ** log2m)
    # Check parameter log2m using SELECT hll_print(imei_hll) FROM daily_per_mno_hll_sketches LIMIT 1
    # or look up for the parameter in the file hll.c in postgres db container
    log2m = 11
    theoretical_error = 1.04 / 2 ** (log2m / 2) * 100
    all_ops = operators + [OperatorConfig.COUNTRY_OPERATOR_NAME]
    for op in all_ops:
        # Check whether compliance stats add up for each operator and log warning if not
        mc = per_operator_monthly_stats[op]
        ti = mc['num_imeis']
        ci = mc['num_compliant_imeis']
        nci = mc['num_noncompliant_imeis']
        percentage_error = abs(ci + nci - ti) / ti * 100 if ti > 0 else 0.0
        logger.info('Percentage error in monthly IMEI count (HLL vs. exact) for '
                    'operator {0}: {1:.3f}% (theoretical max error {2:.3f}%)'
                    .format(op, percentage_error, theoretical_error))
        mc['num_imeis'] = ci + nci
        if op == OperatorConfig.COUNTRY_OPERATOR_NAME:
            imei_stats_metric_root = '{0}monthly_stats.country.'.format(metrics_run_root)
        else:
            imei_stats_metric_root = '{0}monthly_stats.operators.{1}.'.format(metrics_run_root, op)

        # Store all monthly stat metrics in StatsD
        for k, v in mc.items():
            statsd.gauge('{0}{1}'.format(imei_stats_metric_root, k), v)

    logger.info('Finished calculating report data')
    logger.info('Storing report data in DB...')
    data_id, class_run_id = _store_report_data(conn,
                                               operators,
                                               month,
                                               year,
                                               condition_tuples,
                                               per_operator_record_counts,
                                               per_operator_daily_stats,
                                               per_operator_monthly_stats,
                                               per_operator_condition_counts,
                                               per_operator_top_model_imei_counts,
                                               per_operator_top_model_gross_adds,
                                               per_operator_imei_imsi_overloading,
                                               per_operator_imsi_imei_overloading,
                                               per_operator_compliance_data,
                                               per_operator_daily_imei_imsi_overloading,
                                               statsd,
                                               metrics_run_root)
    logger.info('Finished storing report data in DB')
    return data_id, class_run_id, per_operator_tac_compliance_data
Refreshes reporting stats from the DB and stores aggregated results into the various reporting tables.
Refreshes reporting stats from the DB and stores aggregated results into the various reporting tables.
Refreshes reporting stats from the DB and stores aggregated results into the various reporting tables .
def _refresh_data(config, conn, month, year, statsd, metrics_run_root, run_id, debug_query_performance):
    logger = logging.getLogger('dirbs.report')
    nworkers = config.multiprocessing_config.max_db_connections
    db_config = config.db_config
    condition_tuples = [ConditionTuple(x.label, x.blocking) for x in config.conditions]
    condition_tuples = _sort_conditions(condition_tuples)
    if len(condition_tuples) == 0:
        logger.warning('No conditions defined in config: No stats on classification will be generated in the report')
    operators = [op.id for op in config.region_config.operators]
    days_in_month = (datetime.date(year, month, 1) + relativedelta.relativedelta(months=1, days=-1)).day
    per_operator_record_counts = defaultdict(int)
    per_operator_monthly_stats = defaultdict(lambda: defaultdict(int))
    per_operator_daily_stats = defaultdict(lambda: [defaultdict(int) for i in range(0, days_in_month)])
    per_operator_top_model_imei_counts = {}
    per_operator_top_model_gross_adds = {}
    per_operator_imei_imsi_overloading = {}
    per_operator_imsi_imei_overloading = {}
    per_operator_condition_counts = {}
    per_operator_tac_compliance_data = {}
    per_operator_compliance_data = {}
    per_operator_daily_imei_imsi_overloading = {}
    with futures.ProcessPoolExecutor(max_workers=nworkers) as executor:
        logger.info('Simultaneously calculating data volume for each operator using {0:d} workers...'
                    .format(nworkers))
        logger.info('Queueing jobs to calculate monthly record counts...')
        futures_to_cb = {}
        _queue_record_count_jobs(executor, futures_to_cb, per_operator_record_counts, db_config,
                                 operators, month, year, statsd, metrics_run_root, debug_query_performance)
        for f in futures.as_completed(futures_to_cb):
            futures_to_cb[f](f)
        logger.info('Simultaneously calculating report data using {0:d} workers...'
                    .format(nworkers))
        logger.info('Queueing jobs to calculate stats...')
        futures_to_cb = {}
        _queue_compliance_jobs(executor, futures_to_cb, per_operator_condition_counts,
                               per_operator_tac_compliance_data, per_operator_compliance_data,
                               per_operator_monthly_stats, db_config, operators, month, year, condition_tuples,
                               per_operator_record_counts, statsd, metrics_run_root, debug_query_performance, run_id)
        _queue_imsi_imei_overloading_jobs(executor, futures_to_cb, per_operator_imsi_imei_overloading,
                                          db_config, operators, month, year, per_operator_record_counts,
                                          statsd, metrics_run_root, debug_query_performance)
        _queue_imei_imsi_overloading_jobs(executor, futures_to_cb, per_operator_imei_imsi_overloading,
                                          db_config, operators, month, year, per_operator_record_counts,
                                          statsd, metrics_run_root, debug_query_performance)
        _queue_daily_imei_imsi_overloading_jobs(executor, futures_to_cb, per_operator_daily_imei_imsi_overloading,
                                                db_config, operators, month, year, per_operator_record_counts,
                                                statsd, metrics_run_root, debug_query_performance)
        _queue_monthly_stats_jobs(executor, futures_to_cb, per_operator_monthly_stats, db_config, operators,
                                  month, year, per_operator_record_counts, statsd, metrics_run_root,
                                  debug_query_performance)
        _queue_top_model_gross_adds_jobs(executor, futures_to_cb, per_operator_top_model_gross_adds,
                                         db_config, operators, month, year, per_operator_record_counts, statsd,
                                         metrics_run_root, debug_query_performance)
        _queue_top_model_imei_jobs(executor, futures_to_cb, per_operator_top_model_imei_counts,
                                   db_config, operators, month, year, per_operator_record_counts,
                                   statsd, metrics_run_root, debug_query_performance)
        _queue_distinct_id_counts_jobs(executor, futures_to_cb, per_operator_monthly_stats, per_operator_daily_stats,
                                       db_config, operators, month, year, per_operator_record_counts, statsd,
                                       metrics_run_root, debug_query_performance)
        logger.info('Queued jobs to calculate stats. Processing will begin now...')
        for f in futures.as_completed(futures_to_cb):
            futures_to_cb[f](f)
    log2m = 11
    theoretical_error = 1.04 / 2 ** (log2m / 2) * 100
    all_ops = operators + [OperatorConfig.COUNTRY_OPERATOR_NAME]
    for op in all_ops:
        mc = per_operator_monthly_stats[op]
        ti = mc['num_imeis']
        ci = mc['num_compliant_imeis']
        nci = mc['num_noncompliant_imeis']
        percentage_error = abs(ci + nci - ti) / ti * 100 if ti > 0 else 0.0
        logger.info('Percentage error in monthly IMEI count (HLL vs. exact) for '
                    'operator {0}: {1:.3f}% (theoretical max error {2:.3f}%)'
                    .format(op, percentage_error, theoretical_error))
        mc['num_imeis'] = ci + nci
        if op == OperatorConfig.COUNTRY_OPERATOR_NAME:
            imei_stats_metric_root = '{0}monthly_stats.country.'.format(metrics_run_root)
        else:
            imei_stats_metric_root = '{0}monthly_stats.operators.{1}.'.format(metrics_run_root, op)
        for k, v in mc.items():
            statsd.gauge('{0}{1}'.format(imei_stats_metric_root, k), v)
    logger.info('Finished calculating report data')
    logger.info('Storing report data in DB...')
    data_id, class_run_id = _store_report_data(conn,
                                               operators,
                                               month,
                                               year,
                                               condition_tuples,
                                               per_operator_record_counts,
                                               per_operator_daily_stats,
                                               per_operator_monthly_stats,
                                               per_operator_condition_counts,
                                               per_operator_top_model_imei_counts,
                                               per_operator_top_model_gross_adds,
                                               per_operator_imei_imsi_overloading,
                                               per_operator_imsi_imei_overloading,
                                               per_operator_compliance_data,
                                               per_operator_daily_imei_imsi_overloading,
                                               statsd,
                                               metrics_run_root)
    logger.info('Finished storing report data in DB')
    return data_id, class_run_id, per_operator_tac_compliance_data
def _refresh_data ( config , conn , month , year , statsd , metrics_run_root , run_id , debug_query_performance ) : logger = logging . getLogger ( 'dirbs.report' ) nworkers = config . multiprocessing_config . max_db_connections db_config = config . db_config condition_tuples = [ ConditionTuple ( x . label , x . blocking ) for x in config . conditions ] condition_tuples = _sort_conditions ( condition_tuples ) if len ( condition_tuples ) == 0 : logger . warning ( 'No conditions defined in config: No stats on classification will be generated in the report' ) operators = [ op . id for op in config . region_config . operators ] days_in_month = ( datetime . date ( year , month , 1 ) + relativedelta . relativedelta ( months = 1 , days = - 1 ) ) . day per_operator_record_counts = defaultdict ( int ) per_operator_monthly_stats = defaultdict ( lambda : defaultdict ( int ) ) per_operator_daily_stats = defaultdict ( lambda : [ defaultdict ( int ) for i in range ( 0 , days_in_month ) ] ) per_operator_top_model_imei_counts = { } per_operator_top_model_gross_adds = { } per_operator_imei_imsi_overloading = { } per_operator_imsi_imei_overloading = { } per_operator_condition_counts = { } per_operator_tac_compliance_data = { } per_operator_compliance_data = { } per_operator_daily_imei_imsi_overloading = { } with futures . ProcessPoolExecutor ( max_workers = nworkers ) as executor : logger . info ( 'Simultaneously calculating data volume for each operator using {0:d} workers...' . format ( nworkers ) ) logger . info ( 'Queueing jobs to calculate monthly record counts...' ) futures_to_cb = { } _queue_record_count_jobs ( executor , futures_to_cb , per_operator_record_counts , db_config , operators , month , year , statsd , metrics_run_root , debug_query_performance ) for f in futures . as_completed ( futures_to_cb ) : futures_to_cb [ f ] ( f ) logger . info ( 'Simultaneously calculating report data using {0:d} workers...' . format ( nworkers ) ) logger . info ( 'Queueing jobs to calculate stats...' ) futures_to_cb = { } _queue_compliance_jobs ( executor , futures_to_cb , per_operator_condition_counts , per_operator_tac_compliance_data , per_operator_compliance_data , per_operator_monthly_stats , db_config , operators , month , year , condition_tuples , per_operator_record_counts , statsd , metrics_run_root , debug_query_performance , run_id ) _queue_imsi_imei_overloading_jobs ( executor , futures_to_cb , per_operator_imsi_imei_overloading , db_config , operators , month , year , per_operator_record_counts , statsd , metrics_run_root , debug_query_performance ) _queue_imei_imsi_overloading_jobs ( executor , futures_to_cb , per_operator_imei_imsi_overloading , db_config , operators , month , year , per_operator_record_counts , statsd , metrics_run_root , debug_query_performance ) _queue_daily_imei_imsi_overloading_jobs ( executor , futures_to_cb , per_operator_daily_imei_imsi_overloading , db_config , operators , month , year , per_operator_record_counts , statsd , metrics_run_root , debug_query_performance ) _queue_monthly_stats_jobs ( executor , futures_to_cb , per_operator_monthly_stats , db_config , operators , month , year , per_operator_record_counts , statsd , metrics_run_root , debug_query_performance ) _queue_top_model_gross_adds_jobs ( executor , futures_to_cb , per_operator_top_model_gross_adds , db_config , operators , month , year , per_operator_record_counts , statsd , metrics_run_root , debug_query_performance ) _queue_top_model_imei_jobs ( executor , futures_to_cb , per_operator_top_model_imei_counts , db_config , operators , month , year , per_operator_record_counts , statsd , metrics_run_root , debug_query_performance ) _queue_distinct_id_counts_jobs ( executor , futures_to_cb , per_operator_monthly_stats , per_operator_daily_stats , db_config , operators , month , year , per_operator_record_counts , statsd , metrics_run_root , debug_query_performance ) logger . info ( 'Queued jobs to calculate stats. Processing will begin now...' ) for f in futures . as_completed ( futures_to_cb ) : futures_to_cb [ f ] ( f ) log2m = 11 theoretical_error = 1.04 / 2 ** ( log2m / 2 ) * 100 all_ops = operators + [ OperatorConfig . COUNTRY_OPERATOR_NAME ] for op in all_ops : mc = per_operator_monthly_stats [ op ] ti = mc [ 'num_imeis' ] ci = mc [ 'num_compliant_imeis' ] nci = mc [ 'num_noncompliant_imeis' ] percentage_error = abs ( ci + nci - ti ) / ti * 100 if ti > 0 else 0.0 logger . info ( 'Percentage error in monthly IMEI count (HLL vs. exact) for ' 'operator {0}: {1:.3f}% (theoretical max error {2:.3f}%)' . format ( op , percentage_error , theoretical_error ) ) mc [ 'num_imeis' ] = ci + nci if op == OperatorConfig . COUNTRY_OPERATOR_NAME : imei_stats_metric_root = '{0}monthly_stats.country.' . format ( metrics_run_root ) else : imei_stats_metric_root = '{0}monthly_stats.operators.{1}.' . format ( metrics_run_root , op ) for k , v in mc . items ( ) : statsd . gauge ( '{0}{1}' . format ( imei_stats_metric_root , k ) , v ) logger . info ( 'Finished calculating report data' ) logger . info ( 'Storing report data in DB...' ) data_id , class_run_id = _store_report_data ( conn , operators , month , year , condition_tuples , per_operator_record_counts , per_operator_daily_stats , per_operator_monthly_stats , per_operator_condition_counts , per_operator_top_model_imei_counts , per_operator_top_model_gross_adds , per_operator_imei_imsi_overloading , per_operator_imsi_imei_overloading , per_operator_compliance_data , per_operator_daily_imei_imsi_overloading , statsd , metrics_run_root ) logger . info ( 'Finished storing report data in DB' ) return data_id , class_run_id , per_operator_tac_compliance_data
Refreshes reporting stats from the DB and stores aggregated results into the various reporting tables.
Refreshes reporting stats from the DB and stores aggregated results into the various reporting tables .
"""Refreshes reporting stats from the DB and stores aggregated results into the various reporting tables.""" # Sort conditions by primary and secondary key # We need to get the list of operators out of the config # Calculate days in month # Init variables for storing data # We use the per-operator record counts to normalize performance numbers, so we need to do this first # in a separate executor # Process futures as they are completed, calling the associated callback passing the # future as the only argument (other arguments to the callback get partially applied # during the queue* functions above) # Process futures as they are completed, calling the associated callback passing the # future as the only argument (other arguments to the callback get partially applied # during the queue* functions above) # The hll relative error is given by the expression ±1.04/√(2 ** log2m) # Check parameter log2m using SELECT hll_print(imei_hll) FROM daily_per_mno_hll_sketches LIMIT 1 # or look up for the parameter in the file hll.c in postgres db container # Check whether compliance stats add up for each operator and log warning if not # Store all monthly stat metrics in StatsD
config
conn
month
year
statsd
metrics_run_root
run_id
debug_query_performance