worker.py

import concurrent.futures
import functools
import json
import logging
import os
import re
from typing import List

from celery import Task, states, chord
from celery.result import AsyncResult

from collections import defaultdict

from kombu.exceptions import KombuError
from lxml import etree

from inventory_provider.db import ims_data
from inventory_provider.db.ims import IMS
from inventory_provider.routes.poller import load_interfaces_to_poll
from inventory_provider.tasks.app import app
from inventory_provider.tasks.common \
    import get_next_redis, get_current_redis, \
    latch_db, get_latch, set_latch, update_latch_status, \
    ims_sorted_service_type_key, set_single_latch
from inventory_provider.tasks import monitor
from inventory_provider import config
from inventory_provider import environment
from inventory_provider import snmp
from inventory_provider import juniper
from redis import RedisError

FINALIZER_POLLING_FREQUENCY_S = 2.5
FINALIZER_TIMEOUT_S = 300

# TODO: error callback (cf. http://docs.celeryproject.org/en/latest/userguide/calling.html#linking-callbacks-errbacks)  # noqa: E501

environment.setup_logging()
logger = logging.getLogger(__name__)
log_task_entry_and_exit = functools.partial(
    environment.log_entry_and_exit, logger=logger)


class InventoryTaskError(Exception):
    pass


class InventoryTask(Task):

    config = None

    def __init__(self):

        self.args = []

        if InventoryTask.config:
            return

        assert os.path.isfile(
            os.environ['INVENTORY_PROVIDER_CONFIG_FILENAME']), (
                'config file %r not found' %
                os.environ['INVENTORY_PROVIDER_CONFIG_FILENAME'])

        with open(os.environ['INVENTORY_PROVIDER_CONFIG_FILENAME']) as f:
            logging.info(
                "Initializing worker with config from: %r" %
                os.environ['INVENTORY_PROVIDER_CONFIG_FILENAME'])
            InventoryTask.config = config.load(f)
            logging.debug("loaded config: %r" % InventoryTask.config)

    def log_info(self, message):
        logger.debug(message)
        self.send_event('task-info', message=message)

    def log_warning(self, message):
        logger.warning(message)
        self.send_event('task-warning', message=message)

    def log_error(self, message):
        logger.error(message)
        self.send_event('task-error', message=message)


def _unmanaged_interfaces():

    def _convert(d):
        # the config file keys are more readable than
        # the keys used in redis
        return {
            'name': d['address'],
            'interface address': d['network'],
            'interface name': d['interface'].lower(),
            'router': d['router'].lower()
        }

    yield from map(
        _convert,
        InventoryTask.config.get('unmanaged-interfaces', []))


@log_task_entry_and_exit
def refresh_juniper_bgp_peers(hostname, netconf):
    host_peerings = list(juniper.all_bgp_peers(netconf))
    r = get_next_redis(InventoryTask.config)
    r.set(f'juniper-peerings:hosts:{hostname}', json.dumps(host_peerings))


@log_task_entry_and_exit
def refresh_juniper_interface_list(hostname, netconf, lab=False):
    """
    load all interfaces from the netconf doc

    save under 'lab:...' if lab is true

    :param hostname:
    :param netconf:
    :param lab:
    :return:
    """
    logger.debug(
        'removing cached netconf-interfaces for %r' % hostname)

    r = get_next_redis(InventoryTask.config)

    interfaces_keybase = f'netconf-interfaces:{hostname}'
    bundles_keybase = f'netconf-interface-bundles:{hostname}'
    interfaces_all_key = f'netconf-interfaces-hosts:{hostname}'
    if lab:
        interfaces_keybase = f'lab:{interfaces_keybase}'
        interfaces_all_key = f'lab:{interfaces_all_key}'
        bundles_keybase = f'lab:{bundles_keybase}'

    rp = r.pipeline()
    rp.delete(interfaces_all_key)
    # scan with bigger batches, to mitigate network latency effects
    for k in r.scan_iter(f'{interfaces_keybase}:*', count=1000):
        rp.delete(k)
    for k in r.scan_iter(
            f'{bundles_keybase}:*', count=1000):
        rp.delete(k)
    rp.execute()

    all_bundles = defaultdict(list)

    rp = r.pipeline()

    rp.set(
        interfaces_all_key,
        json.dumps(list(juniper.interface_addresses(netconf))))

    for ifc in juniper.list_interfaces(netconf):

        bundles = ifc.get('bundle', None)
        for bundle in bundles:
            if bundle:
                all_bundles[bundle].append(ifc['name'])
        rp.set(
            f'{interfaces_keybase}:{ifc["name"]}',
            json.dumps(ifc))

    for k, v in all_bundles.items():
        rp.set(
            f'{bundles_keybase}:{k}',
            json.dumps(v))

    rp.execute()


def _erase_next_db(config):
    """
    flush next db, but first save latch and then restore afterwards

    TODO: handle the no latch scenario nicely
    :param config:
    :return:
    """
    r = get_next_redis(config)
    saved_latch = get_latch(r)

    if saved_latch:
        # execute as transaction to ensure that latch is always available in
        # db that is being flushed
        rp = r.pipeline()
        rp.multi()
        rp.flushdb()
        set_single_latch(
            rp,
            saved_latch['this'],
            saved_latch['current'],
            saved_latch['next'],
            saved_latch.get('timestamp', 0),
            saved_latch.get('update-started', 0)
        )
        rp.execute()

        # ensure latch is consistent in all dbs
        set_latch(
            config,
            new_current=saved_latch['current'],
            new_next=saved_latch['next'],
            timestamp=saved_latch.get('timestamp', 0),
            update_timestamp=saved_latch.get('update-started', 0))


def populate_poller_cache(interface_services, r):
    host_services = defaultdict(dict)
    for v in interface_services.values():
        # logger.debug(v)
        if v:
            h = v[0]['equipment']
            i = v[0]['port']
            host_services[h][i] = [{
                'id': s['id'],
                'name': s['name'],
                'type': s['service_type'],
                'status': s['status']
            } for s in v]
    # todo - delete from redis
    rp = r.pipeline()
    for k in r.scan_iter('poller_cache:*', count=1000):
        rp.delete(k)
    rp.execute()
    rp = r.pipeline()
    for host, interface_services in host_services.items():
        rp.set(f'poller_cache:{host}', json.dumps(interface_services))
    rp.execute()


def _build_subnet_db(update_callback=lambda s: None):

    r = get_next_redis(InventoryTask.config)

    update_callback('loading all network addresses')
    subnets = {}
    # scan with bigger batches, to mitigate network latency effects
    for k in r.scan_iter('netconf-interfaces-hosts:*', count=1000):
        k = k.decode('utf-8')
        hostname = k[len('netconf-interfaces-hosts:'):]
        host_interfaces = r.get(k).decode('utf-8')
        host_interfaces = json.loads(host_interfaces)
        for ifc in host_interfaces:
            ifc['router'] = hostname
            entry = subnets.setdefault(ifc['interface address'], [])
            entry.append(ifc)

    for ifc in _unmanaged_interfaces():
        entry = subnets.setdefault(ifc['interface address'], [])
        entry.append(ifc)

    update_callback('saving {} subnets'.format(len(subnets)))

    rp = r.pipeline()
    for k, v in subnets.items():
        rp.set(f'subnets:{k}', json.dumps(v))
    rp.execute()


def _build_juniper_peering_db(update_callback=lambda s: None):

    def _is_ix(peering_info):
        if peering_info.get('instance', '') != 'IAS':
            return False
        if not peering_info.get('group', '').startswith('GEANT-IX'):
            return False

        expected_keys = ('description', 'local-asn', 'remote-asn')
        if any(peering_info.get(x, None) is None for x in expected_keys):
            logger.error('internal data error, looks like ix peering but'
                         f'some expected keys are missing: {peering_info}')
            return False
        return True

    r = get_next_redis(InventoryTask.config)

    update_callback('loading all juniper network peerings')
    peerings_per_address = {}
    ix_peerings = []
    peerings_per_asn = {}
    peerings_per_logical_system = {}
    peerings_per_group = {}
    peerings_per_routing_instance = {}
    all_peerings = []

    # scan with bigger batches, to mitigate network latency effects
    key_prefix = 'juniper-peerings:hosts:'
    for k in r.scan_iter(f'{key_prefix}*', count=1000):
        key_name = k.decode('utf-8')
        hostname = key_name[len(key_prefix):]
        host_peerings = r.get(key_name).decode('utf-8')
        host_peerings = json.loads(host_peerings)
        for p in host_peerings:
            p['hostname'] = hostname
            peerings_per_address.setdefault(p['address'], []).append(p)
            if _is_ix(p):
                ix_peerings.append(p)
            asn = p.get('remote-asn', None)
            if asn:
                peerings_per_asn.setdefault(asn, []).append(p)
            logical_system = p.get('logical-system', None)
            if logical_system:
                peerings_per_logical_system.setdefault(
                    logical_system, []).append(p)
            group = p.get('group', None)
            if group:
                peerings_per_group.setdefault(group, []).append(p)
            routing_instance = p.get('instance', None)
            if routing_instance:
                peerings_per_routing_instance.setdefault(
                    routing_instance, []).append(p)
            all_peerings.append(p)

    # sort ix peerings by group
    ix_groups = {}
    for p in ix_peerings:
        description = p['description']
        keyword = description.split(' ')[0]  # regex needed??? (e.g. tabs???)
        ix_groups.setdefault(keyword, set()).add(p['address'])

    rp = r.pipeline()

    # for use with /msr/bgp
    rp.set('juniper-peerings:all', json.dumps(all_peerings))

    # create peering entries, keyed by remote addresses
    update_callback(f'saving {len(peerings_per_address)} remote peers')
    for k, v in peerings_per_address.items():
        rp.set(f'juniper-peerings:remote:{k}', json.dumps(v))

    # create pivoted ix group name lists
    update_callback(f'saving {len(ix_groups)} remote ix peering groups')
    for k, v in ix_groups.items():
        group_addresses = list(v)
        rp.set(f'juniper-peerings:ix-groups:{k}', json.dumps(group_addresses))

    # create pivoted asn peering lists
    update_callback(f'saving {len(peerings_per_asn)} asn peering lists')
    for k, v in peerings_per_asn.items():
        rp.set(f'juniper-peerings:peer-asn:{k}', json.dumps(v))

    # create pivoted logical-systems peering lists
    update_callback(
        f'saving {len(peerings_per_logical_system)}'
        ' logical-system peering lists')
    for k, v in peerings_per_logical_system.items():
        rp.set(f'juniper-peerings:logical-system:{k}', json.dumps(v))

    # create pivoted group peering lists
    update_callback(
        f'saving {len(peerings_per_group)} group peering lists')
    for k, v in peerings_per_group.items():
        rp.set(f'juniper-peerings:group:{k}', json.dumps(v))

    # create pivoted routing instance peering lists
    update_callback(
        f'saving {len(peerings_per_routing_instance)} group peering lists')
    for k, v in peerings_per_routing_instance.items():
        rp.set(f'juniper-peerings:routing-instance:{k}', json.dumps(v))

    rp.execute()


def _build_snmp_peering_db(update_callback=lambda s: None):

    r = get_next_redis(InventoryTask.config)

    update_callback('loading all snmp network peerings')
    peerings = {}

    # scan with bigger batches, to mitigate network latency effects
    key_prefix = 'snmp-peerings:hosts:'
    for k in r.scan_iter(f'{key_prefix}*', count=1000):
        key_name = k.decode('utf-8')
        hostname = key_name[len(key_prefix):]
        host_peerings = r.get(key_name).decode('utf-8')
        host_peerings = json.loads(host_peerings)
        for p in host_peerings:
            p['hostname'] = hostname
            peerings.setdefault(p['remote'], []).append(p)

    update_callback(f'saving {len(peerings)} remote peers')

    rp = r.pipeline()
    for k, v in peerings.items():
        rp.set(f'snmp-peerings:remote:{k}', json.dumps(v))
    rp.execute()


def check_task_status(task_id, parent=None, forget=False):
    r = AsyncResult(task_id, app=app)
    assert r.id == task_id  # sanity

    result = {
        'id': task_id,
        'status': r.status,
        'exception': r.status in states.EXCEPTION_STATES,
        'ready': r.status in states.READY_STATES,
        'success': r.status == states.SUCCESS,
        'parent': parent
    }

    # TODO: only discovered this case by testing, is this the only one?
    #       ... otherwise need to pre-test json serialization
    if isinstance(r.result, Exception):
        result['result'] = {
            'error type': type(r.result).__name__,
            'message': str(r.result)
        }
    else:
        result['result'] = r.result

    def child_taskids(children):
        # reverse-engineered, can't find documentation on this
        for child in children:
            if not child:
                continue
            if isinstance(child, list):
                logger.debug(f'list: {child}')
                yield from child_taskids(child)
                continue
            if isinstance(child, str):
                yield child
                continue
            assert isinstance(child, AsyncResult)
            yield child.id

    for child_id in child_taskids(getattr(r, 'children', []) or []):
        yield from check_task_status(child_id, parent=task_id)

    if forget and result['ready']:
        r.forget()

    yield result


@app.task(base=InventoryTask, bind=True, name='update_entry_point')
@log_task_entry_and_exit
def update_entry_point(self):

    try:
        _erase_next_db(InventoryTask.config)
        update_latch_status(InventoryTask.config, pending=True)
        monitor.clear_joblog(get_current_redis(InventoryTask.config))
        self.log_info("Starting update")

        routers = retrieve_and_persist_neteng_managed_device_list(
            info_callback=self.log_info,
            warning_callback=self.log_warning
        )
        lab_routers = InventoryTask.config.get('lab-routers', [])

        chord(
            (
                ims_task.s().on_error(task_error_handler.s()),
                chord(
                    (reload_router_config_chorded.s(r) for r in routers),
                    empty_task.si('router tasks complete')
                ),
                chord(
                    (reload_lab_router_config_chorded.s(r)
                     for r in lab_routers),
                    empty_task.si('lab router tasks complete')
                )
            ),
            final_task.si().on_error(task_error_handler.s())
        )()

        return self.request.id
    except (RedisError, KombuError):
        update_latch_status(InventoryTask.config, pending=False, failure=True)
        logger.exception('error launching refresh subtasks')
        raise


@app.task
def task_error_handler(request, exc, traceback):
    update_latch_status(InventoryTask.config, pending=False, failure=True)
    logger.warning('Task {0!r} raised error: {1!r}'.format(request.id, exc))


@app.task(base=InventoryTask, bind=True, name='empty_task')
def empty_task(self, message):
    logger.warning(f'message from empty task: {message}')


def retrieve_and_persist_neteng_managed_device_list(
        info_callback=lambda s: None,
        warning_callback=lambda s: None):
    netdash_equipment = None
    try:
        info_callback('querying netdash for managed routers')
        netdash_equipment = list(juniper.load_routers_from_netdash(
            InventoryTask.config['managed-routers']))
    except Exception as e:
        warning_callback(f'Error retrieving device list: {e}')

    if netdash_equipment:
        info_callback(f'found {len(netdash_equipment)} routers')
    else:
        warning_callback('No devices retrieved, using previous list')
        try:
            current_r = get_current_redis(InventoryTask.config)
            netdash_equipment = current_r.get('netdash')
            netdash_equipment = json.loads(netdash_equipment.decode('utf-8'))
            if not netdash_equipment:
                raise InventoryTaskError(
                    'No equipment retrieved from previous list')
        except Exception as e:
            warning_callback(str(e))
            update_latch_status(
                InventoryTask.config, pending=False, failure=True)
            raise e

    try:
        next_r = get_next_redis(InventoryTask.config)
        next_r.set('netdash', json.dumps(netdash_equipment))
        info_callback(f'saved {len(netdash_equipment)} managed routers')
    except Exception as e:
        warning_callback(str(e))
        update_latch_status(InventoryTask.config, pending=False, failure=True)
        raise e
    return netdash_equipment


@app.task(base=InventoryTask, bind=True, name='reload_lab_router_config')
@log_task_entry_and_exit
def reload_lab_router_config_chorded(self, hostname):
    try:
        self.log_info(f'loading netconf data for lab {hostname}')

        # load new netconf data, in this thread
        netconf_str = retrieve_and_persist_netconf_config(
            hostname, lab=True, update_callback=self.log_warning)
        netconf_doc = etree.fromstring(netconf_str)

        refresh_juniper_interface_list(hostname, netconf_doc, lab=True)

        # load snmp indexes
        community = juniper.snmp_community_string(netconf_doc)
        if not community:
            raise InventoryTaskError(
                f'error extracting community string for {hostname}')
        else:
            self.log_info(f'refreshing snmp interface indexes for {hostname}')
            logical_systems = juniper.logical_systems(netconf_doc)

            # load snmp data, in this thread
            snmp_refresh_interfaces_chorded(
                hostname, community, logical_systems, self.log_info)

        self.log_info(f'updated configuration for lab {hostname}')
    except Exception:
        errmsg = f'unhandled exception loading {hostname} info'
        logger.exception(errmsg)
        update_latch_status(InventoryTask.config, pending=True, failure=True)
        self.log_error(errmsg)
        # TODO: re-raise and handle in some common way for all tasks
        # raise


@app.task(base=InventoryTask, bind=True, name='reload_router_config')
@log_task_entry_and_exit
def reload_router_config_chorded(self, hostname):
    try:
        self.log_info(f'loading netconf data for {hostname}')
        netconf_str = retrieve_and_persist_netconf_config(
            hostname, update_callback=self.log_warning)

        netconf_doc = etree.fromstring(netconf_str)

        # clear cached classifier responses for this router, and
        # refresh peering data
        logger.info(f'refreshing peers & clearing cache for {hostname}')
        refresh_juniper_bgp_peers(hostname, netconf_doc)
        refresh_juniper_interface_list(hostname, netconf_doc)

        # load snmp indexes
        community = juniper.snmp_community_string(netconf_doc)
        if not community:
            raise InventoryTaskError(
                f'error extracting community string for {hostname}')
        else:
            self.log_info(f'refreshing snmp interface indexes for {hostname}')
            logical_systems = juniper.logical_systems(netconf_doc)

            # load snmp data, in this thread
            snmp_refresh_interfaces_chorded(
                hostname, community, logical_systems, self.log_info)
            snmp_refresh_peerings_chorded(hostname, community, logical_systems)

        logger.info(f'updated configuration for {hostname}')
    except Exception:
        errmsg = f'unhandled exception loading {hostname} info'
        logger.exception(errmsg)
        update_latch_status(InventoryTask.config, pending=True, failure=True)
        self.log_error(errmsg)
        # TODO: re-raise and handle in some common way for all tasks
        # raise


def retrieve_and_persist_netconf_config(
        hostname, lab=False, update_callback=lambda s: None):
    redis_key = f'netconf:{hostname}'
    if lab:
        redis_key = f'lab:{redis_key}'

    try:
        netconf_doc = juniper.load_config(
            hostname, InventoryTask.config["ssh"])
        netconf_str = etree.tostring(netconf_doc, encoding='unicode')
    except (ConnectionError, juniper.NetconfHandlingError, InventoryTaskError):
        msg = f'error loading netconf data from {hostname}'
        logger.exception(msg)
        update_callback(msg)
        r = get_current_redis(InventoryTask.config)

        netconf_str = r.get(redis_key)
        if not netconf_str:
            update_callback(f'no cached netconf for {redis_key}')
            raise InventoryTaskError(
                f'netconf error with {hostname}'
                f' and no cached netconf data found')
        logger.info(f'Returning cached netconf data for {hostname}')
        update_callback(f'Returning cached netconf data for {hostname}')

    r = get_next_redis(InventoryTask.config)
    r.set(redis_key, netconf_str)
    logger.info(f'netconf info loaded from {hostname}')
    return netconf_str


@log_task_entry_and_exit
def snmp_refresh_interfaces_chorded(
        hostname, community, logical_systems, update_callback=lambda s: None):
    try:
        interfaces = list(
            snmp.get_router_snmp_indexes(hostname, community, logical_systems))
    except ConnectionError:
        msg = f'error loading snmp interface data from {hostname}'
        logger.exception(msg)
        update_callback(msg)
        r = get_current_redis(InventoryTask.config)
        interfaces = r.get(f'snmp-interfaces:{hostname}')
        if not interfaces:
            raise InventoryTaskError(
                f'snmp error with {hostname}'
                f' and no cached snmp interface data found')
        # unnecessary json encode/decode here ... could be optimized
        interfaces = json.loads(interfaces.decode('utf-8'))
        update_callback(f'using cached snmp interface data for {hostname}')

    r = get_next_redis(InventoryTask.config)

    rp = r.pipeline()
    rp.set(f'snmp-interfaces:{hostname}', json.dumps(interfaces))

    # optimization for DBOARD3-372
    # interfaces is a list of dicts like: {'name': str, 'index': int}
    for ifc in interfaces:
        ifc['hostname'] = hostname
        rp.set(
            f'snmp-interfaces-single:{hostname}:{ifc["name"]}',
            json.dumps(ifc))

    rp.execute()

    update_callback(f'snmp interface info loaded from {hostname}')


@log_task_entry_and_exit
def snmp_refresh_peerings_chorded(
        hostname, community, logical_systems, update_callback=lambda S: None):
    try:
        peerings = list(
            snmp.get_peer_state_info(hostname, community, logical_systems))
    except ConnectionError:
        msg = f'error loading snmp peering data from {hostname}'
        logger.exception(msg)
        update_callback(msg)
        r = get_current_redis(InventoryTask.config)
        peerings = r.get(f'snmp-peerings:hosts:{hostname}')
        if peerings is None:
            raise InventoryTaskError(
                f'snmp error with {peerings}'
                f' and no cached peering data found')
        # unnecessary json encode/decode here ... could be optimized
        peerings = json.loads(peerings.decode('utf-8'))
        update_callback(f'using cached snmp peering data for {hostname}')

    r = get_next_redis(InventoryTask.config)
    r.set(f'snmp-peerings:hosts:{hostname}', json.dumps(peerings))

    update_callback(f'snmp peering info loaded from {hostname}')


def cache_extracted_ims_data(extracted_data, use_current=False):
    if use_current:
        r = get_current_redis(InventoryTask.config)
    else:
        r = get_next_redis(InventoryTask.config)

    for k, v in extracted_data.items():
        r.set(f'ims:cache:{k}', json.dumps(v))


@app.task(base=InventoryTask, bind=True, name='ims_task')
@log_task_entry_and_exit
def ims_task(self, use_current=False):
    try:
        extracted_data = extract_ims_data()
        cache_extracted_ims_data(extracted_data)
        transformed_data = transform_ims_data(extracted_data)
        transformed_data['locations'] = extracted_data['locations']
        transformed_data['lg_routers'] = extracted_data['lg_routers']
        persist_ims_data(transformed_data, use_current)
    except Exception as e:
        logger.error(e)
        update_latch_status(InventoryTask.config, pending=True, failure=True)


def extract_ims_data():
    c = InventoryTask.config["ims"]
    return _extract_ims_data(
        ims_api_url=c['api'],
        ims_username=c['username'],
        ims_password=c['password'])


def _extract_ims_data(ims_api_url, ims_username, ims_password):
    """
    convenient entry point for testing ...

    :param ims_api_url:
    :param ims_username:
    :param ims_password:
    :return:
    """
    def _ds() -> IMS:
        return IMS(ims_api_url, ims_username, ims_password)

    _ds().clear_dynamic_context_cache()

    locations = {}
    lg_routers = []
    geant_nodes = []
    customer_contacts = {}
    planned_work_contacts = {}
    circuit_ids_to_monitor = []
    circuit_ids_and_sids = {}
    additional_circuit_customers = {}
    flexils_data = {}
    customers = {}

    hierarchy = {}
    port_id_details = defaultdict(list)
    port_id_services = defaultdict(list)

    @log_task_entry_and_exit
    def _populate_locations():
        nonlocal locations
        locations = {k: v for k, v in ims_data.get_node_locations(ds=_ds())}

    @log_task_entry_and_exit
    def _populate_lg_routers():
        nonlocal lg_routers
        lg_routers = list(ims_data.lookup_lg_routers(ds=_ds()))

    @log_task_entry_and_exit
    def _populate_geant_nodes():
        nonlocal geant_nodes
        geant_nodes = list(ims_data.lookup_geant_nodes(ds=_ds()))

    @log_task_entry_and_exit
    def _populate_customer_contacts():
        nonlocal customer_contacts
        customer_contacts = \
            {k: v for k, v in ims_data.get_customer_tts_contacts(ds=_ds())}

    @log_task_entry_and_exit
    def _populate_customer_planned_work_contacts():
        nonlocal planned_work_contacts
        planned_work_contacts = \
            {k: v for k, v in
             ims_data.get_customer_planned_work_contacts(ds=_ds())}

    @log_task_entry_and_exit
    def _populate_circuit_ids_to_monitor():
        nonlocal circuit_ids_to_monitor
        circuit_ids_to_monitor = \
            list(ims_data.get_monitored_circuit_ids(ds=_ds()))

    @log_task_entry_and_exit
    def _populate_sids():
        nonlocal circuit_ids_and_sids
        circuit_ids_and_sids = \
            {cid: sid for cid, sid in ims_data.get_ids_and_sids(ds=_ds())}

    @log_task_entry_and_exit
    def _populate_additional_circuit_customers():
        nonlocal additional_circuit_customers
        additional_circuit_customers = \
            ims_data.get_circuit_related_customers(ds=_ds())

    exceptions = {}
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = {
            executor.submit(_populate_locations): 'locations',
            executor.submit(_populate_geant_nodes): 'geant_nodes',
            executor.submit(_populate_lg_routers): 'lg_routers',
            executor.submit(_populate_customer_contacts): 'customer_contacts',
            executor.submit(_populate_customer_planned_work_contacts):
                'planned_work_contacts',
            executor.submit(_populate_circuit_ids_to_monitor):
                'circuit_ids_to_monitor',
            executor.submit(_populate_sids): 'sids',
            executor.submit(_populate_additional_circuit_customers):
                'additional_circuit_customers'
        }

        for future in concurrent.futures.as_completed(futures):
            if future.exception():
                exceptions[futures[future]] = str(future.exception())

    if exceptions:
        raise InventoryTaskError(json.dumps(exceptions, indent=2))

    @log_task_entry_and_exit
    def _populate_customers():
        nonlocal customers
        customers = {c['id']: c for c in _ds().get_all_entities('customer')}

    @log_task_entry_and_exit
    def _populate_flexils_data():
        nonlocal flexils_data
        flexils_data = ims_data.get_flexils_by_circuitid(ds=_ds())

    @log_task_entry_and_exit
    def _populate_hierarchy():
        nonlocal hierarchy
        hierarchy = {
            d['id']: d for d in ims_data.get_circuit_hierarchy(ds=_ds())}
        logger.debug("hierarchy complete")

    @log_task_entry_and_exit
    def _populate_port_id_details():
        nonlocal port_id_details
        for x in ims_data.get_port_details(ds=_ds()):
            pd = port_id_details[x['port_id']]
            pd.append(x)
        logger.debug("Port details complete")

    @log_task_entry_and_exit
    def _populate_circuit_info():
        for x in ims_data.get_port_id_services(ds=_ds()):
            port_id_services[x['port_a_id']].append(x)
        logger.debug("port circuits complete")

    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = {
            executor.submit(_populate_hierarchy): 'hierarchy',
            executor.submit(_populate_port_id_details): 'port_id_details',
            executor.submit(_populate_circuit_info): 'circuit_info',
            executor.submit(_populate_flexils_data): 'flexils_data',
            executor.submit(_populate_customers): 'customers'
        }

        for future in concurrent.futures.as_completed(futures):
            if future.exception():
                exceptions[futures[future]] = str(future.exception())

    if exceptions:
        raise InventoryTaskError(json.dumps(exceptions, indent=2))

    return {
        'locations': locations,
        'lg_routers': lg_routers,
        'customer_contacts': customer_contacts,
        'planned_work_contacts': planned_work_contacts,
        'circuit_ids_to_monitor': circuit_ids_to_monitor,
        'circuit_ids_sids': circuit_ids_and_sids,
        'additional_circuit_customers': additional_circuit_customers,
        'hierarchy': hierarchy,
        'port_id_details': port_id_details,
        'port_id_services': port_id_services,
        'geant_nodes': geant_nodes,
        'flexils_data': flexils_data,
        'customers': customers
    }


def transform_ims_data(data):
    locations = data['locations']
    customer_contacts = data['customer_contacts']
    planned_work_contacts = data['planned_work_contacts']
    circuit_ids_to_monitor = data['circuit_ids_to_monitor']
    additional_circuit_customers = data['additional_circuit_customers']
    hierarchy = data['hierarchy']
    port_id_details = data['port_id_details']
    port_id_services = data['port_id_services']
    circuit_ids_and_sids = data['circuit_ids_sids']
    geant_nodes = data['geant_nodes']
    flexils_data = data['flexils_data']
    customers = data['customers']

    sid_services = defaultdict(list)

    def _get_circuit_contacts(c):
        customer_ids = {c['customerid']}
        customer_ids.update(
            [ac['id'] for ac in additional_circuit_customers.get(c['id'], [])]
        )
        tts_contacts = set().union(
            *[customer_contacts.get(cid, []) for cid in customer_ids])
        pw_contacts = set().union(
            *[planned_work_contacts.get(cid, []) for cid in customer_ids])
        return tts_contacts, pw_contacts

    for d in hierarchy.values():
        c, ttc = _get_circuit_contacts(d)
        d['contacts'] = sorted(list(c))
        d['planned_work_contacts'] = sorted(list(ttc))

        # add flexils data to port_id_details and port_id_services
        all_ils_details = flexils_data.get(d['id'])
        if all_ils_details:
            for ils_details in all_ils_details:
                pid = port_id_details.get(ils_details['key'], [])
                pid.append({
                    'port_id': ils_details['key'],
                    'equipment_name': ils_details['node_name'],
                    'interface_name': ils_details['full_port_name']
                })
                port_id_details[ils_details['key']] = pid

                pis = port_id_services.get(ils_details['key'], [])
                pis.append({
                    'id': d['id'],
                    'name': d['name'],
                    'project': d['project'],
                    'port_a_id': ils_details['key'],
                    'circuit_type': d['circuit-type'],
                    'status': d['status'],
                    'service_type': d['product'],
                    'customerid': d['customerid'],
                    'customer': customers.get(d['customerid'], ''),
                    'contacts': d['contacts'],
                    'planned_work_contacts': d['planned_work_contacts']
                })
                port_id_services[ils_details['key']] = pis

    def _convert_to_bits(value, unit):
        unit = unit.lower()
        conversions = {
            'm': 1 << 20,
            'mb': 1 << 20,
            'g': 1 << 30,
            'gbe': 1 << 30,
        }
        return int(value) * conversions[unit]

    def _get_speed(circuit_id):
        c = hierarchy[circuit_id]
        if c['status'] != 'operational':
            return 0
        pattern = re.compile(r'^(\d+)([a-zA-z]+)$')
        m = pattern.match(c['speed'])
        if m:
            try:
                return _convert_to_bits(m[1], m[2])
            except KeyError as e:
                logger.debug(f'Could not find key: {e} '
                             f'for circuit: {circuit_id}')
                return 0
        else:
            if c['circuit-type'] == 'service' \
                    or c['product'].lower() == 'ethernet':
                return sum(
                    (_get_speed(x) for x in c['carrier-circuits'])
                )
            else:
                return 0

    def _get_fibre_routes(c_id):
        _circ = hierarchy.get(c_id, None)
        if _circ is None:
            return
        if _circ['speed'].lower() == 'fibre_route':
            yield _circ['id']
        else:
            for cc in _circ['carrier-circuits']:
                yield from _get_fibre_routes(cc)

    def _get_related_services(circuit_id: str) -> List[dict]:
        rs = {}
        c = hierarchy.get(circuit_id, None)
        if c:

            if c['circuit-type'] == 'service':
                rs[c['id']] = {
                    'id': c['id'],
                    'name': c['name'],
                    'circuit_type': c['circuit-type'],
                    'service_type': c['product'],
                    'project': c['project'],
                    'contacts': c['contacts'],
                    'planned_work_contacts': c['planned_work_contacts']
                }
                if c['id'] in circuit_ids_to_monitor:
                    rs[c['id']]['status'] = c['status']
                else:
                    rs[c['id']]['status'] = 'non-monitored'

                if c['id'] in circuit_ids_and_sids:
                    rs[c['id']]['sid'] = circuit_ids_and_sids[c['id']]

            if c['sub-circuits']:
                for sub in c['sub-circuits']:
                    temp_parents = \
                        _get_related_services(sub)
                    rs.update({t['id']: t for t in temp_parents})
        return list(rs.values())

    def _format_service(s):
        s['additional_customers'] = \
            additional_circuit_customers.get(s['id'], [])
        s['original_status'] = s['status']
        s['monitored'] = True
        if s['circuit_type'] == 'service' \
                and s['id'] not in circuit_ids_to_monitor:
            s['monitored'] = False
            s['status'] = 'non-monitored'
        pd_a = port_id_details[s['port_a_id']][0]
        location_a = locations.get(pd_a['equipment_name'], None)
        if location_a:
            loc_a = location_a['pop']
        else:
            loc_a = locations['UNKNOWN_LOC']['pop']
            logger.warning(
                f'Unable to find location for {pd_a["equipment_name"]} - '
                f'Service ID {s["id"]}')
        s['pop_name'] = loc_a['name']
        s['pop_abbreviation'] = loc_a['abbreviation']
        s['equipment'] = pd_a['equipment_name']
        s['card_id'] = ''  # this is redundant I believe
        s['port'] = pd_a['interface_name']
        s['logical_unit'] = ''  # this is redundant I believe
        if 'port_b_id' in s:
            pd_b = port_id_details[s['port_b_id']][0]
            location_b = locations.get(pd_b['equipment_name'], None)
            if location_b:
                loc_b = location_b['pop']
            else:
                loc_b = locations['UNKNOWN_LOC']['pop']
                logger.warning(
                    f'Unable to find location for {pd_b["equipment_name"]} - '
                    f'Service ID {s["id"]}')

            s['other_end_pop_name'] = loc_b['name']
            s['other_end_pop_abbreviation'] = loc_b['abbreviation']
            s['other_end_equipment'] = pd_b['equipment_name']
            s['other_end_port'] = pd_b['interface_name']
        else:
            s['other_end_pop_name'] = ''
            s['other_end_pop_abbreviation'] = ''
            s['other_end_equipment'] = ''
            s['other_end_port'] = ''

        s.pop('port_a_id', None)
        s.pop('port_b_id', None)

    services_by_type = {}
    interface_services = defaultdict(list)
    # using a dict to ensure no duplicates
    node_pair_services = defaultdict(dict)
    pop_nodes = defaultdict(list)

    for value in locations.values():
        pop_nodes[value['pop']['name']].append(value['equipment-name'])

    for key, value in port_id_details.items():
        for details in value:
            k = f"{details['equipment_name']}:" \
                f"{details['interface_name']}"
            circuits = port_id_services.get(details['port_id'], [])

            for circ in circuits:
                # contacts, pw_contacts = _get_circuit_contacts(circ)
                # we only want to include the Related Services contacts
                contacts = set()
                pw_contacts = set()
                circ['fibre-routes'] = []
                for x in set(_get_fibre_routes(circ['id'])):
                    c = {
                        'id': hierarchy[x]['id'],
                        'name': hierarchy[x]['name'],
                        'status': hierarchy[x]['status']
                    }
                    circ['fibre-routes'].append(c)

                circ['related-services'] = \
                    _get_related_services(circ['id'])

                for tlc in circ['related-services']:
                    # why were these removed?
                    # contacts.update(tlc.pop('contacts'))
                    if circ['status'] == 'operational' \
                            and circ['id'] in circuit_ids_to_monitor \
                            and tlc['status'] == 'operational' \
                            and tlc['id'] in circuit_ids_to_monitor:
                        contacts.update(tlc.get('contacts'))
                        pw_contacts.update(
                            tlc.get('planned_work_contacts', []))
                circ['contacts'] = sorted(list(contacts))
                circ['planned_work_contacts'] = sorted(list(pw_contacts))

                circ['calculated-speed'] = _get_speed(circ['id'])
                _format_service(circ)

                type_services = services_by_type.setdefault(
                    ims_sorted_service_type_key(circ['service_type']), dict())
                type_services[circ['id']] = circ
                if circ['other_end_equipment']:
                    node_pair_services[
                        f"{circ['equipment']}/{circ['other_end_equipment']}"
                    ][circ['id']] = circ
                try:
                    # get the physical port circuit, if it exists
                    # https://jira.software.geant.org/browse/POL1-687
                    port_circuit = next(
                        c for c in circuits if c.get('port_type') == 'ports')
                except StopIteration:
                    port_circuit = None

                sid = None
                if circ['id'] in circuit_ids_and_sids:
                    sid = circuit_ids_and_sids[circ['id']]
                elif 'sid' in details:
                    if len(circuits) > 1:
                        if port_circuit != circ:
                            # if this is not the physical port circuit
                            # related to this port, then we don't want to
                            # assign the SID to this circuit, so skip.
                            continue

                    # assign the SID from the port to this circuit
                    sid = details['sid']

                if sid is None:
                    continue

                circ['sid'] = sid

                sid_info = {
                    'circuit_id': circ['id'],
                    'sid': sid,
                    'status': circ['original_status'],
                    'monitored': circ['monitored'],
                    'name': circ['name'],
                    'speed': circ['calculated-speed'],
                    'service_type': circ['service_type'],
                    'project': circ['project'],
                    'customer': circ['customer'],
                    'equipment': circ['equipment'],
                    'port': circ['port'],
                    'geant_equipment': circ['equipment'] in geant_nodes
                }
                if sid_info not in sid_services[sid]:
                    sid_services[sid].append(sid_info)

        interface_services[k].extend(circuits)

    return {
        'hierarchy': hierarchy,
        'interface_services': interface_services,
        'services_by_type': services_by_type,
        'node_pair_services': node_pair_services,
        'sid_services': sid_services,
        'pop_nodes': pop_nodes
    }


def persist_ims_data(data, use_current=False):
    hierarchy = data['hierarchy']
    locations = data['locations']
    lg_routers = data['lg_routers']
    interface_services = data['interface_services']
    services_by_type = data['services_by_type']
    node_pair_services = data['node_pair_services']
    sid_services = data['sid_services']
    pop_nodes = data['pop_nodes']

    def _get_pops():
        # de-dupe the sites (by abbreviation)
        pops = {
            equip['pop']['abbreviation']: equip['pop']
            for equip in locations.values()
            if equip['pop']['abbreviation']}
        return pops.values()

    if use_current:
        r = get_current_redis(InventoryTask.config)

        r.delete('ims:sid_services')

        # only need to delete the individual keys if it's just an IMS update
        # rather than a complete update (the db will have been flushed)
        for key_pattern in [
            'ims:pop:*',
            'ims:location:*',
            'ims:lg:*',
            'ims:circuit_hierarchy:*',
            'ims:interface_services:*',
            'ims:access_services:*',
            'ims:gws_indirect:*',
            'ims:node_pair_services:*',
            'ims:pop_nodes:*'
        ]:
            rp = r.pipeline()
            for k in r.scan_iter(key_pattern, count=1000):
                rp.delete(k)
    else:
        r = get_next_redis(InventoryTask.config)

    r.set('ims:sid_services', json.dumps(sid_services))
    rp = r.pipeline()
    for h, d in locations.items():
        rp.set(f'ims:location:{h}', json.dumps([d]))
    for pop in _get_pops():
        rp.set(f'ims:pop:{pop["abbreviation"]}', json.dumps(pop))
    rp.execute()
    rp = r.pipeline()
    for router in lg_routers:
        rp.set(f'ims:lg:{router["equipment name"]}', json.dumps(router))
    rp.execute()
    rp = r.pipeline()
    for circ in hierarchy.values():
        rp.set(f'ims:circuit_hierarchy:{circ["id"]}', json.dumps([circ]))
    rp.execute()
    rp = r.pipeline()
    for k, v in interface_services.items():
        rp.set(
            f'ims:interface_services:{k}',
            json.dumps(v))
    rp.execute()
    rp = r.pipeline()
    for k, v in node_pair_services.items():
        rp.set(
            f'ims:node_pair_services:{k}',
            json.dumps(list(v.values())))
    rp.execute()
    rp = r.pipeline()
    for k, v in pop_nodes.items():
        rp.set(
            f'ims:pop_nodes:{k}',
            json.dumps(sorted(v)))
    rp.execute()

    rp = r.pipeline()

    populate_poller_cache(interface_services, r)
    populate_mic_cache(interface_services, r)

    for service_type, services in services_by_type.items():
        for v in services.values():
            rp.set(
                f'ims:services:{service_type}:{v["name"]}',
                json.dumps({
                    'id': v['id'],
                    'name': v['name'],
                    'project': v['project'],
                    'here': {
                        'pop': {
                            'name': v['pop_name'],
                            'abbreviation': v['pop_abbreviation']
                        },
                        'equipment': v['equipment'],
                        'port': v['port'],
                    },
                    'there': {
                        'pop': {
                            'name': v['other_end_pop_name'],
                            'abbreviation': v['other_end_pop_abbreviation']
                        },
                        'equipment': v['other_end_equipment'],
                        'port': v['other_end_port'],
                    },
                    'speed_value': v['calculated-speed'],
                    'speed_unit': 'n/a',
                    'status': v['status'],
                    'type': v['service_type']
                }))

    rp.execute()


def populate_mic_cache(interface_services, r):
    cache_key = "mic:impact:all-data"
    all_data = defaultdict(lambda: defaultdict(dict))

    def _get_formatted_rs(_d):
        for rs in _d['related-services']:
            if rs['status'] == 'operational':
                yield {
                    'id': rs['id'],
                    'sid': rs.get('sid', ''),
                    'status': rs['status'],
                    'name': rs['name'],
                    'service_type': rs['service_type'],
                    'contacts': rs['contacts'],
                    'planned_work_contacts': rs['planned_work_contacts']
                }

    for services in interface_services.values():
        if services:
            current_interface_services = []
            for d in services:
                if d.get('related-services'):
                    current_interface_services.extend(
                        list(_get_formatted_rs(d)))
            if current_interface_services:
                site = f'{services[0]["pop_name"]} ' \
                       f'({services[0]["pop_abbreviation"]})'
                eq_name = services[0]['equipment']
                if_name = services[0]['port']
                all_data[site][eq_name][if_name] = current_interface_services

    result = json.dumps(all_data)
    r.set(cache_key, result.encode('utf-8'))


@app.task(base=InventoryTask, bind=True, name='final_task')
@log_task_entry_and_exit
def final_task(self):

    r = get_current_redis(InventoryTask.config)
    latch = get_latch(r)
    if latch['failure']:
        raise InventoryTaskError('Sub-task failed - check logs for details')

    _build_subnet_db(update_callback=self.log_info)
    _build_snmp_peering_db(update_callback=self.log_info)
    _build_juniper_peering_db(update_callback=self.log_info)
    populate_poller_interfaces_cache(warning_callback=self.log_warning)

    latch_db(InventoryTask.config)
    self.log_info('latched current/next dbs')


@log_task_entry_and_exit
def populate_poller_interfaces_cache(warning_callback=lambda s: None):
    no_lab_cache_key = 'classifier-cache:poller-interfaces:all:no_lab'
    all_cache_key = 'classifier-cache:poller-interfaces:all'
    non_lab_populated_interfaces = None
    all_populated_interfaces = None

    r = get_next_redis(InventoryTask.config)

    try:
        lab_keys_pattern = 'lab:netconf-interfaces-hosts:*'
        lab_equipment = [h.decode('utf-8')[len(lab_keys_pattern) - 1:]
                         for h in r.keys(lab_keys_pattern)]

        all_populated_interfaces = list(
            load_interfaces_to_poll(InventoryTask.config, use_next_redis=True))
        non_lab_populated_interfaces = [x for x in all_populated_interfaces
                                        if x['router'] not in lab_equipment]

    except Exception as e:
        warning_callback(f"Failed to retrieve all required data {e}")
        logger.exception(
            "Failed to retrieve all required data, logging exception")

    if not non_lab_populated_interfaces or not all_populated_interfaces:
        previous_r = get_current_redis(InventoryTask.config)

        def _load_previous(key):
            try:
                warning_callback(f"populating {key} "
                                 "from previously cached data")
                return json.loads(previous_r.get(key))
            except Exception as e:
                warning_callback(f"Failed to load {key} "
                                 f"from previously cached data: {e}")

        if not non_lab_populated_interfaces:
            non_lab_populated_interfaces = _load_previous(no_lab_cache_key)

        if not all_populated_interfaces:
            all_populated_interfaces = _load_previous(all_cache_key)

    r.set(no_lab_cache_key, json.dumps(non_lab_populated_interfaces))
    r.set(all_cache_key, json.dumps(all_populated_interfaces))