-
Bjarke Madsen authored
Process dashboards fast, but block until org is configured.
Bjarke Madsen authoredProcess dashboards fast, but block until org is configured.
helpers.py 6.70 KiB
import re
import logging
from itertools import product
from string import ascii_uppercase
from brian_dashboard_manager.templating.render import create_panel
PANEL_HEIGHT = 12
PANEL_WIDTH = 24
logger = logging.getLogger(__file__)
def get_description(interface):
return interface.get('description', '').strip()
def is_physical_interface(interface):
return re.match('^PHY', get_description(interface))
def is_aggregate_interface(interface):
return re.match('^LAG', get_description(interface))
def is_logical_interface(interface):
return re.match('^SRV_', get_description(interface))
def is_nren(interface):
regex = '(PHY|LAG|(SRV_(GLOBAL|LHCONE|MDVPN|IAS|CLS|L3VPN))) CUSTOMER'
return re.match(regex, get_description(interface))
def is_cls(interface):
return 'SRV_CLS' in get_description(interface)
def is_ias_public(interface):
return 'SRV_IAS PUBLIC' in get_description(interface)
def is_ias_private(interface):
return 'SRV_IAS PRIVATE' in get_description(interface)
def is_ias_customer(interface):
return 'SRV_IAS CUSTOMER' in get_description(interface)
def is_ias_upstream(interface):
return 'SRV_IAS UPSTREAM' in get_description(interface)
def is_re_peer(interface):
return 'SRV_GLOBAL RE_INTERCONNECT' in get_description(interface)
def is_re_customer(interface):
regex = '(PHY|LAG|SRV_GLOBAL) CUSTOMER'
return re.match(regex, get_description(interface))
def is_gcs(interface):
return re.match('^SRV_GCS', get_description(interface))
def is_geantopen(interface):
return 'GEANTOPEN' in get_description(interface)
def is_l2circuit(interface):
return 'SRV_L2CIRCUIT' in get_description(interface)
def is_lhcone_peer(interface):
description = get_description(interface)
return 'LHCONE' in description and 'SRV_L3VPN RE' in description
def is_lhcone_customer(interface):
description = get_description(interface)
return 'LHCONE' in description and 'SRV_L3VPN CUSTOMER' in description
def is_mdvpn(interface):
return re.match('^SRV_MDVPN CUSTOMER', get_description(interface))
def is_infrastructure_backbone(interface):
regex = '(SRV_GLOBAL|LAG|PHY) INFRASTRUCTURE BACKBONE'
return re.match(regex, get_description(interface))
def is_lag_backbone(interface):
is_lag = 'LAG' in get_description(interface)
return is_infrastructure_backbone(interface) and is_lag
def parse_backbone_name(description, *args, **kwargs):
link = description.split('|')[1].strip()
link = link.replace('( ', '(')
return link
def is_phy_upstream(interface):
return re.match('^PHY UPSTREAM', get_description(interface))
def parse_phy_upstream_name(description, host):
name = description.split(' ')[2].strip().upper()
location = host.split('.')[1].upper()
return f'{name} - {location}'
def num_generator(start=1):
num = start
while True:
yield num
num += 1
def gridPos_generator(id_generator, start=0):
num = start
while True:
yield {
"height": PANEL_HEIGHT,
"width": PANEL_WIDTH,
"x": 0,
"y": num * PANEL_HEIGHT,
"id": next(id_generator)
}
num += 1
def letter_generator():
i = 0
j = 0
num_letters = len(ascii_uppercase)
while True:
result = ascii_uppercase[i % num_letters]
# tack on an extra letter if we are out of them
if (i >= num_letters):
result += ascii_uppercase[j % num_letters]
j += 1
if (j != 0 and j % num_letters == 0):
i += 1
else:
i += 1
yield result
# peer_predicate is a function that is used to filter the interfaces
# parse_func receives interface information and returns a peer name.
def get_interface_data(interfaces, name_parse_func=None):
result = {}
for interface in interfaces:
if not name_parse_func:
# Most (but not all) descriptions use a format
# which has the peer name as the third element.
def name_parse_func(desc, *args, **kwargs):
return desc.split(' ')[2].upper()
description = interface.get('description', '').strip()
interface_name = interface.get('name')
host = interface.get('router', '')
dashboard_name = name_parse_func(description, host)
peer = result.get(dashboard_name, [])
router = host.replace('.geant.net', '')
panel_title = f"{router} - {{}} - {interface_name} - {description}"
peer.append({
'title': panel_title,
'interface': interface_name,
'hostname': host
})
result[dashboard_name] = peer
return result
# Helper used for generating all traffic/error panels
# with a single target field (ingress/egress or err/s)
def get_panel_fields(panel, panel_type, datasource):
letters = letter_generator()
def get_target_data(alias, field):
return {
**panel, # panel has target hostname and interface
'alias': alias,
'refId': next(letters),
'select_field': field,
'percentile': 'percentile' in alias.lower(),
}
error_fields = [('Ingress Errors', 'errorsIn'),
('Egress Errors', 'errorsOut')]
ingress = ['Ingress Traffic', 'Ingress 95th Percentile']
egress = ['Egress Traffic', 'Egress 95th Percentile']
is_v6 = panel_type == 'IPv6'
is_error = panel_type == 'errors'
in_field = 'ingressv6' if is_v6 else 'ingress'
out_field = 'egressv6' if is_v6 else 'egress'
fields = [*product(ingress, [in_field]), *product(egress, [out_field])]
targets = error_fields if is_error else fields
return create_panel({
**panel,
'datasource': datasource,
'title': panel['title'].format(panel_type),
'panel_targets': [get_target_data(*target) for target in targets],
'y_axis_type': 'errors' if is_error else 'bits',
})
def get_dashboard_data(data, datasource, tag, errors=False):
id_gen = num_generator()
gridPos = gridPos_generator(id_gen)
def get_panel_definitions(panels, datasource):
result = []
for panel in panels:
result.append(get_panel_fields(
{**panel, **next(gridPos)}, 'traffic', datasource))
result.append(get_panel_fields(
{**panel, **next(gridPos)}, 'IPv6', datasource))
if errors:
result.append(get_panel_fields(
{**panel, **next(gridPos)}, 'errors', datasource))
return result
for peer, panels in data.items():
yield {
'title': peer,
'datasource': datasource,
'panels': get_panel_definitions(panels, datasource),
'tag': tag
}