import csv
import functools
import json
import logging
import os
import re
import subprocess
import tempfile
import threading
import time
from datetime import datetime
from enum import IntFlag
from pathlib import Path
from typing import List
from uuid import uuid4
from redis.exceptions import RedisError
from kombu.exceptions import KombuError
from celery import Task, states
from celery.result import AsyncResult
from collections import defaultdict
from lxml import etree
import jsonschema
from inventory_provider.db import ims_data
from inventory_provider.db.ims import IMS
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
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
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__)
def log_task_entry_and_exit(f):
# cf. https://stackoverflow.com/a/47663642
# cf. https://stackoverflow.com/a/59098957/6708581
@functools.wraps(f)
def _w(*args, **kwargs):
logger.debug(f'>>> {f.__name__}{args}')
try:
return f(*args, **kwargs)
finally:
logger.debug(f'<<< {f.__name__}{args}')
return _w
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)
@app.task(base=InventoryTask, bind=True, name='snmp_refresh_peerings')
@log_task_entry_and_exit
def snmp_refresh_peerings(self, hostname, community, logical_systems):
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)
self.log_warning(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'))
self.log_warning(f'using cached snmp peering data for {hostname}')
r = get_next_redis(InventoryTask.config)
r.set(f'snmp-peerings:hosts:{hostname}', json.dumps(peerings))
self.log_info(f'snmp peering info loaded from {hostname}')
@app.task(base=InventoryTask, bind=True, name='snmp_refresh_interfaces')
@log_task_entry_and_exit
def snmp_refresh_interfaces(self, hostname, community, logical_systems):
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)
self.log_warning(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'))
self.log_warning(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()
self.log_info(f'snmp interface info loaded from {hostname}')
@app.task(base=InventoryTask, bind=True, name='netconf_refresh_config')
@log_task_entry_and_exit
def netconf_refresh_config(self, hostname):
try:
netconf_doc = juniper.load_config(
hostname, InventoryTask.config["ssh"])
netconf_str = etree.tostring(netconf_doc, encoding='unicode')
except (ConnectionError, juniper.NetconfHandlingError):
msg = f'error loading netconf data from {hostname}'
logger.exception(msg)
self.log_warning(msg)
r = get_current_redis(InventoryTask.config)
netconf_str = r.get(f'netconf:{hostname}')
if not netconf_str:
raise InventoryTaskError(
f'netconf error with {hostname}'
f' and no cached netconf data found')
self.log_warning(f'using cached netconf data for {hostname}')
r = get_next_redis(InventoryTask.config)
r.set(f'netconf:{hostname}', netconf_str)
self.log_info(f'netconf info loaded from {hostname}')
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', []))
@app.task(base=InventoryTask, bind=True,
name='update_neteng_managed_device_list')
@log_task_entry_and_exit
def update_neteng_managed_device_list(self):
self.log_info('querying netdash for managed routers')
routers = list(juniper.load_routers_from_netdash(
InventoryTask.config['managed-routers']))
self.log_info(f'found {len(routers)} routers, saving details')
r = get_next_redis(InventoryTask.config)
r.set('netdash', json.dumps(routers).encode('utf-8'))
self.log_info(f'saved {len(routers)} managed routers')
def load_netconf_data(hostname):
"""
this method should only be called from a task
:param hostname:
:return:
"""
r = get_next_redis(InventoryTask.config)
netconf = r.get('netconf:' + hostname)
if not netconf:
raise InventoryTaskError('no netconf data found for %r' % hostname)
return etree.fromstring(netconf.decode('utf-8'))
def clear_cached_classifier_responses(hostname=None):
if hostname:
logger.debug(
'removing cached classifier responses for %r' % hostname)
else:
logger.debug('removing all cached classifier responses')
r = get_next_redis(InventoryTask.config)
def _hostname_keys():
for k in r.keys('classifier-cache:juniper:%s:*' % hostname):
yield k
# TODO: very inefficient ... but logically simplest at this point
for k in r.keys('classifier-cache:peer:*'):
value = r.get(k.decode('utf-8'))
if not value:
# deleted in another thread
continue
value = json.loads(value.decode('utf-8'))
interfaces = value.get('interfaces', [])
if hostname in [i['interface']['router'] for i in interfaces]:
yield k
def _all_keys():
return r.keys('classifier-cache:*')
keys_to_delete = _hostname_keys() if hostname else _all_keys()
rp = r.pipeline()
for k in keys_to_delete:
rp.delete(k)
rp.execute()
@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):
logger.debug(
'removing cached netconf-interfaces for %r' % hostname)
r = get_next_redis(InventoryTask.config)
rp = r.pipeline()
rp.delete(f'netconf-interfaces-hosts:{hostname}')
# scan with bigger batches, to mitigate network latency effects
for k in r.scan_iter(f'netconf-interfaces:{hostname}:*', count=1000):
rp.delete(k)
for k in r.scan_iter(
f'netconf-interface-bundles:{hostname}:*', count=1000):
rp.delete(k)
rp.execute()
all_bundles = defaultdict(list)
rp = r.pipeline()
rp.set(
f'netconf-interfaces-hosts:{hostname}',
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'netconf-interfaces:{hostname}:{ifc["name"]}',
json.dumps(ifc))
for k, v in all_bundles.items():
rp.set(
f'netconf-interface-bundles:{hostname}:{k}',
json.dumps(v))
rp.execute()
@app.task(base=InventoryTask, bind=True, name='reload_router_config')
@log_task_entry_and_exit
def reload_router_config(self, hostname):
self.log_info(f'loading netconf data for {hostname}')
# get the timestamp for the current netconf data
current_netconf_timestamp = None
try:
netconf_doc = load_netconf_data(hostname)
current_netconf_timestamp \
= juniper.netconf_changed_timestamp(netconf_doc)
logger.debug(
'current netconf timestamp: %r' % current_netconf_timestamp)
except InventoryTaskError:
pass # ok at this point if not found
# load new netconf data, in this thread
netconf_refresh_config.apply(args=[hostname])
netconf_doc = load_netconf_data(hostname)
# return if new timestamp is the same as the original timestamp
new_netconf_timestamp = juniper.netconf_changed_timestamp(netconf_doc)
assert new_netconf_timestamp, \
'no timestamp available for new netconf data'
if new_netconf_timestamp == current_netconf_timestamp:
msg = f'no timestamp change for {hostname} netconf data'
logger.debug(msg)
self.log_info(msg)
return
# clear cached classifier responses for this router, and
# refresh peering data
self.log_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.apply(
args=[hostname, community, logical_systems])
snmp_refresh_peerings.apply(
args=[hostname, community, logical_systems])
clear_cached_classifier_responses(None)
self.log_info(f'updated configuration for {hostname}')
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)
r.flushdb()
if saved_latch:
set_latch(
config,
new_current=saved_latch['current'],
new_next=saved_latch['next'],
timestamp=saved_latch.get('timestamp', 0))
@log_task_entry_and_exit
def launch_refresh_cache_all(config):
"""
utility function intended to be called outside of the worker process
:param config: config structure as defined in config.py
:return:
"""
try:
_erase_next_db(config)
update_latch_status(config, pending=True)
monitor.clear_joblog(get_current_redis(config))
# first batch of subtasks: refresh cached IMS location data
subtasks = [
update_neteng_managed_device_list.apply_async(),
update_equipment_locations.apply_async(),
update_lg_routers.apply_async(),
]
[x.get() for x in subtasks]
# now launch the task whose only purpose is to
# act as a convenient parent for all of the remaining tasks
t = internal_refresh_phase_2.apply_async()
return t.id
except (RedisError, KombuError):
update_latch_status(config, pending=False, failure=True)
logger.exception('error launching refresh subtasks')
raise
@app.task(base=InventoryTask, bind=True, name='internal_refresh_phase_2')
@log_task_entry_and_exit
def internal_refresh_phase_2(self):
# second batch of subtasks:
# ims circuit information
try:
subtasks = [
update_circuit_hierarchy_and_port_id_services.apply_async()
]
r = get_next_redis(InventoryTask.config)
routers = r.get('netdash')
assert routers
netdash_equipment = json.loads(routers.decode('utf-8'))
for hostname in netdash_equipment:
logger.debug(f'queueing router refresh jobs for {hostname}')
subtasks.append(reload_router_config.apply_async(args=[hostname]))
pending_task_ids = [x.id for x in subtasks]
refresh_finalizer.apply_async(args=[json.dumps(pending_task_ids)])
except KombuError:
# TODO: possible race condition here
# e.g. if one of these tasks takes a long time and another
# update is started, we could end up with strange data
update_latch_status(config, pending=False, failure=True)
logger.exception('error launching refresh phase 2 subtasks')
raise
@app.task(
base=InventoryTask,
bind=True,
name='update_circuit_hierarchy_and_port_id_services')
@log_task_entry_and_exit
def update_circuit_hierarchy_and_port_id_services(self, use_current=False):
c = InventoryTask.config["ims"]
ds1 = IMS(c['api'], c['username'], c['password'])
ds2 = IMS(c['api'], c['username'], c['password'])
ds3 = IMS(c['api'], c['username'], c['password'])
locations = {k: v for k, v in ims_data.get_node_locations(ds1)}
tls_names = list(ims_data.get_service_types(ds1))
hierarchy = None
port_id_details = defaultdict(list)
port_id_services = defaultdict(list)
interface_services = defaultdict(list)
access_services = {}
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 _populate_hierarchy():
nonlocal hierarchy
hierarchy = {d['id']: d for d in ims_data.get_circuit_hierarchy(ds1)}
logger.debug("hierarchy complete")
def _populate_port_id_details():
nonlocal port_id_details
for x in ims_data.get_port_details(ds2):
pd = port_id_details[x['port_id']]
pd.append(x)
logger.debug("Port details complete")
def _populate_circuit_info():
for x in ims_data.get_port_id_services(ds3):
port_id_services[x['port_a_id']].append(x)
logger.debug("port circuits complete")
hierarchy_thread = threading.Thread(target=_populate_hierarchy)
hierarchy_thread.start()
port_id_details_thread = threading.Thread(target=_populate_port_id_details)
port_id_details_thread.start()
circuit_info_thread = threading.Thread(target=_populate_circuit_info)
circuit_info_thread.start()
hierarchy_thread.join()
circuit_info_thread.join()
port_id_details_thread.join()
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)
# whilst this is a service type the top level for reporting
# are the BGP services on top of it
tls_names.remove('IP PEERING - NON R&E (PUBLIC)')
def get_top_level_services(circuit_id: str) -> List[dict]:
tls = {}
c = hierarchy.get(circuit_id, None)
if c:
def _is_tls(candidate):
if candidate['product'] in tls_names:
return True
if candidate['speed'] == 'BGP':
return True
return False
if _is_tls(c):
tls[c['id']] = {
'id': c['id'],
'name': c['name'],
'status': c['status'],
'circuit_type': 'service',
'project': c['project']
}
elif c['sub-circuits']:
for sub in c['sub-circuits']:
temp_parents = \
get_top_level_services(sub)
tls.update({t['id']: t for t in temp_parents})
return list(tls.values())
def _format_service(s):
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)
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:
circ['fibre-routes'] = \
[{
'id': hierarchy[x]['id'],
'name': hierarchy[x]['name'],
'status': hierarchy[x]['status']
} for x in set(_get_fibre_routes(circ['id']))]
circ['top-level-services'] = \
get_top_level_services(circ['id'])
circ['calculated-speed'] = _get_speed(circ['id'])
_format_service(circ)
if circ['service_type'].lower() == 'geant ip':
access_services[circ['id']] = circ
interface_services[k].extend(circuits)
if use_current:
r = get_current_redis(InventoryTask.config)
else:
r = get_next_redis(InventoryTask.config)
rp = r.pipeline()
for k in r.scan_iter('ims:circuit_hierarchy:*', count=1000):
rp.delete(k)
rp.execute()
rp = r.pipeline()
for k in r.scan_iter('ims:interface_services:*', count=1000):
rp.delete(k)
rp.execute()
rp = r.pipeline()
for k in r.scan_iter('ims:access_services:*', count=1000):
rp.delete(k)
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 v in access_services.values():
rp.set(
f'ims:access_services:{v["name"]}',
json.dumps({
'id': v['id'],
'name': v['name'],
'pop_name': v['pop_name'],
'other_end_pop_name': v['other_end_pop_name'],
'equipment': v['equipment'],
'other_end_equipment': v['other_end_equipment'],
'speed_value': v['calculated-speed'],
'speed_unit': 'n/a'
}))
rp.execute()
@app.task(base=InventoryTask, bind=True, name='update_equipment_locations')
@log_task_entry_and_exit
def update_equipment_locations(self, use_current=False):
if use_current:
r = get_current_redis(InventoryTask.config)
else:
r = get_next_redis(InventoryTask.config)
rp = r.pipeline()
for k in r.scan_iter('ims:location:*', count=1000):
rp.delete(k)
rp.execute()
c = InventoryTask.config["ims"]
ds = IMS(c['api'], c['username'], c['password'])
rp = r.pipeline()
hostnames_found = set()
for h, d in ims_data.get_node_locations(ds):
# put into a list to match non-IMS version
rp.set(f'ims:location:{h}', json.dumps([d]))
if h in hostnames_found:
logger.debug(f'Multiple entries for {h}')
hostnames_found.add(h)
rp.execute()
@app.task(base=InventoryTask, bind=True, name='update_lg_routers')
@log_task_entry_and_exit
def update_lg_routers(self, use_current=False):
if use_current:
r = get_current_redis(InventoryTask.config)
for k in r.scan_iter('ims:lg:*'):
r.delete(k)
else:
r = get_next_redis(InventoryTask.config)
for k in r.scan_iter('ims:lg:*'):
r.delete(k)
c = InventoryTask.config["ims"]
ds = IMS(c['api'], c['username'], c['password'])
for router in ims_data.lookup_lg_routers(ds):
r.set(f'ims:lg:{router["equipment name"]}', json.dumps(router))
class OTRSFiles(IntFlag):
CUSTOMER_COMPANIES = 1
CUSTOMER_USERS = 2
@app.task(base=InventoryTask, bind=True, name='export_data_for_otrs')
@log_task_entry_and_exit
def export_data_for_otrs(self, files_to_export=None, export_duplicates=False):
debug_uuid = uuid4()
logger.debug(f'debug uuid: {debug_uuid}')
if files_to_export:
files_to_export = OTRSFiles(files_to_export)
else:
files_to_export = set(OTRSFiles)
ims_config = InventoryTask.config["ims"]
otrs_config = InventoryTask.config["otrs-export"]
command_template = 'rsync -aPq --no-perms --rsh="ssh -l {user} -p 22 -i {key_file} -o \'UserKnownHostsFile {known_hosts}\'" {source_dir}/* {destination}' # noqa
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
ds = IMS(
ims_config['api'],
ims_config['username'],
ims_config['password'])
prefix = datetime.now().strftime('%Y%m%d') + '_'
if OTRSFiles.CUSTOMER_COMPANIES in files_to_export:
cus_co_path = temp_path.joinpath(f'{prefix}customer_company.csv')
with open(cus_co_path, 'w+') as f:
writer = csv.writer(f, delimiter='^')
writer.writerows(ims_data.otrs_get_customer_company_rows(ds))
if OTRSFiles.CUSTOMER_USERS in files_to_export:
cus_usr_path = temp_path.joinpath(f'{prefix}customer_user.csv')
with open(cus_usr_path, 'w+') as f:
writer = csv.writer(f, delimiter='^')
writer.writerows(ims_data.otrs_get_customer_users_rows(
ds, return_duplicates=export_duplicates))
command = command_template.format(
user=otrs_config['username'],
key_file=otrs_config['private-key'],
known_hosts=otrs_config['known-hosts'],
source_dir=temp_dir,
destination=otrs_config['destination']
)
subprocess.run(command, shell=True, check=True)
return debug_uuid
def _wait_for_tasks(task_ids, update_callback=lambda s: None):
all_successful = True
start_time = time.time()
while task_ids and time.time() - start_time < FINALIZER_TIMEOUT_S:
update_callback('waiting for tasks to complete: %r' % task_ids)
time.sleep(FINALIZER_POLLING_FREQUENCY_S)
def _task_and_children_result(id):
tasks = list(check_task_status(id))
return {
'error': any([t['ready'] and not t['success'] for t in tasks]),
'ready': all([t['ready'] for t in tasks])
}
results = dict([
(id, _task_and_children_result(id))
for id in task_ids])
if any([t['error'] for t in results.values()]):
all_successful = False
task_ids = [
id for id, status in results.items()
if not status['ready']]
if task_ids:
raise InventoryTaskError(
'timeout waiting for pending tasks to complete')
if not all_successful:
raise InventoryTaskError(
'some tasks finished with an error')
update_callback('pending tasks completed in {} seconds'.format(
time.time() - start_time))
@app.task(base=InventoryTask, bind=True, name='refresh_finalizer')
@log_task_entry_and_exit
def refresh_finalizer(self, pending_task_ids_json):
# TODO: if more types of errors appear, use a finally block
input_schema = {
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "array",
"items": {"type": "string"}
}
try:
task_ids = json.loads(pending_task_ids_json)
logger.debug('task_ids: %r' % task_ids)
jsonschema.validate(task_ids, input_schema)
_wait_for_tasks(task_ids, update_callback=self.log_info)
_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)
except (jsonschema.ValidationError,
json.JSONDecodeError,
InventoryTaskError,
RedisError,
KombuError) as e:
update_latch_status(InventoryTask.config, failure=True)
raise e
latch_db(InventoryTask.config)
self.log_info('latched current/next dbs')
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 = {}
# 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)
# 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()
# 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