Skip to content
Snippets Groups Projects
Select Git revision
  • 9c37b26d44d72c41b4cc9cb3d95149998ea2bc99
  • master default protected
  • develop
  • featureXnymous
  • moveToL10woSimpleSaml
  • upgrateToL10
  • setupDoc
  • tomas.budvytis-master-patch-06795
  • v1.0.0
9 results

ServiceProviderController.php

Blame
  • cli.py 3.64 KiB
    import contextlib
    from functools import partial
    import json
    
    import click
    import jsonschema
    
    from brian_polling_manager.interface_stats import config, brian, juniper
    from brian_polling_manager import inventory, influx
    
    
    import logging.config
    import os
    
    
    LOGGING_DEFAULT_CONFIG = {
        'version': 1,
        'disable_existing_loggers': False,
        'formatters': {
            'simple': {
                'format': '%(asctime)s - %(name)s '
                          '(%(lineno)d) - %(levelname)s - %(message)s'
            }
        },
    
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'DEBUG',
                'formatter': 'simple',
                'stream': 'ext://sys.stdout'
            },
        },
    
        'loggers': {
            'brian_polling_manager': {
                'level': 'DEBUG',
                'handlers': ['console'],
                'propagate': False
            }
        },
    
        'root': {
            'level': 'INFO',
            'handlers': ['console']
        }
    }
    
    
    def setup_logging():
        """
        set up logging using the configured filename
    
        if LOGGING_CONFIG is defined in the environment, use this for
        the filename, otherwise use LOGGING_DEFAULT_CONFIG
        """
        logging_config = LOGGING_DEFAULT_CONFIG
        if 'LOGGING_CONFIG' in os.environ:
            filename = os.environ['LOGGING_CONFIG']
            with open(filename) as f:
                logging_config = json.loads(f.read())
    
        # # TODO: this mac workaround should be removed ...
        # import platform
        # if platform.system() == 'Darwin':
        #     logging_config['handlers']['syslog_handler']['address'] \
        #         = '/var/run/syslog'
    
        logging.config.dictConfig(logging_config)
    
    
    def _validate_config(_unused_ctx, _unused_param, file):
        try:
            return config.load(file)
        except json.JSONDecodeError:
            raise click.BadParameter('config file is not valid json')
        except jsonschema.ValidationError as e:
            raise click.BadParameter(e)
    
    
    def _managed_routers(inventory_base_urls):
        return {_ifc['router'] for _ifc in inventory.load_interfaces(inventory_base_urls)}
    
    
    def fqdn_netconf_doc_map(app_config_params):
        _result = {}
        for fqdn in _managed_routers(inventory_base_urls=app_config_params['inventory']):
            _result[fqdn] = juniper.get_interface_info_ncrpc(fqdn)
    
        return _result
    
    
    def _brian_points(router_fqdn, netconf_doc, measurement_name):
        interfaces = juniper.physical_interface_counters(netconf_doc)
        counters = brian.counters(router_fqdn=router_fqdn, interface_counters=interfaces)
        yield from map(partial(brian.ctr2point, measurement_name), counters)
    
        interfaces = juniper.logical_interface_counters(netconf_doc)
        counters = brian.counters(router_fqdn=router_fqdn, interface_counters=interfaces)
        yield from map(partial(brian.ctr2point, measurement_name), counters)
    
    
    def _main(app_config_params: dict):
        """
        callable entry point, without click
        ... tmp, for testing
    
        :param app_config_params:
        :return:
        """
    
        setup_logging()
    
        nc_doc_map = fqdn_netconf_doc_map(app_config_params)
        influx_params = app_config_params['influx']['brian-counters']
        for _fqdn, _ncdoc in nc_doc_map.items():
            points = _brian_points(
                router_fqdn=_fqdn,
                netconf_doc=_ncdoc,
                measurement_name=influx_params['measurement'])
    
            with contextlib.closing(influx.influx_client(influx_params)) as client:
                client.write_points(points)
    
    
    @click.command()
    @click.option(
        '--config', 'app_config_params',
        required=True,
        type=click.File('r'),
        help='Config filename',
        callback=_validate_config)
    def main(app_config_params: dict):
        _main(app_config_params)
    
    
    if __name__ == '__main__':
        main()