nokia.py

import logging
import pathlib
from brian_polling_manager.interface_stats.vendors.common import (
    netconf_connect,
    parse_interface_xml,
    remove_xml_namespaces,
)
from lxml import etree

logger = logging.getLogger(__name__)


NCCLIENT_PARAMS = {
    "device_params": {"name": "sros"},
    "nc_params": {"capabilities": ["urn:nokia.com:nc:pysros:pc"]},
    "timeout": 60,
}

INTERFACE_COUNTERS = {
    "__defaults__": {"transform": int, "required": False},
    "name": {
        "path": ["./lag-name", "./port-id"],
        "transform": lambda v: str(v).strip(),
    },
    "brian": {
        "ingressOctets": {"path": "./statistics/in-octets", "required": True},
        "ingressPackets": {
            "path": "./statistics/in-packets",
            "required": True,
        },
        "egressOctets": {"path": "./statistics/out-octets", "required": True},
        "egressPackets": {"path": "./statistics/out-packets", "required": True},
        "ingressErrors": {"path": "./statistics/in-errors"},
        "egressErrors": {"path": "./statistics/out-errors"},
        "ingressDiscards": {"path": "./statistics/in-discards"},
    },
    "errors": {
        "output_total_errors": {"path": "./statistics/out-errors"},
        "input_total_errors": {"path": "./statistics/in-errors"},
        "input_discards": {"path": "./statistics/in-discards"},
        "output_discards": {"path": "./statistics/out-discards"},
    },
}

INTERFACE_COUNTERS_ALT = {
    "__defaults__": {"transform": int, "required": False},
    "name": {
        "path": "./interface-name",
        "transform": lambda v: str(v).strip(),
    },
    "brian": {
        "ingressOctets": {"path": "./statistics/ip/in-octets", "required": True},
        "ingressPackets": {
            "path": "./statistics/ip/in-packets",
            "required": True,
        },
        "egressOctets": {"path": "./statistics/ip/out-octets", "required": True},
        "egressPackets": {"path": "./statistics/ip/out-packets", "required": True},
        "ingressOctetsv6": {"path": "./ipv6/statistics/in-octets"},
        "ingressPacketsv6": {"path": "./ipv6/statistics/in-packets"},
        "egressOctetsv6": {"path": "./ipv6/statistics/out-octets"},
        "egressPacketsv6": {"path": "./ipv6/statistics/out-packets"},
    },
    "errors": {"output_discards": {"path": "./statistics/ip/out-discard-packets"}},
}


def get_netconf_interface_info(router_name: str, ssh_params: dict):
    STATE_NS = "urn:nokia.com:sros:ns:yang:sr:state"

    def query(*path: str):
        # https://github.com/nokia/pysros/blob/main/examples/show_port_counters.py
        # https://netdevops.me/2020/nokia-yang-tree-and-path-browser/
        # https://github.com/hellt/nokia-yangtree

        root = etree.Element("filter")
        sub_elem = etree.SubElement(
            root, f"{{{STATE_NS}}}state", nsmap={"nokia-state": STATE_NS}
        )
        for p in path:
            sub_elem = etree.SubElement(sub_elem, f"{{{STATE_NS}}}{p}")
        return root

    def get_xml(conn, query_):
        response = conn.get(filter=query_)
        return remove_xml_namespaces(response.tostring.decode("utf-8"))

    queries = {
        "port": query("port"),
        "lag": query("lag"),
        "router-interface": query("router", "interface"),
    }
    with netconf_connect(
        hostname=router_name, ssh_params=ssh_params, **NCCLIENT_PARAMS
    ) as conn:
        return {ept: get_xml(conn, q) for ept, q in queries.items()}


def get_netconf_interface_info_from_source_dir(router_name: str, source_dir: str):
    def read_doc_or_raise(name: str):
        file = pathlib.Path(source_dir) / name
        if not file.is_file():
            raise ValueError(f"file {file} is not a valid file")

        return etree.fromstring(file.read_text())

    return {
        key: read_doc_or_raise(f"{router_name}-{key}s.xml")
        for key in ["port", "lag", "router-interface"]
    }


def _port_counters(state_doc: etree.Element):
    """
    :param state_doc: /nokia-state:state/port yang node
    :return: counters for all monitored ports
    """

    for elem in state_doc.xpath("//data/state/port"):
        yield parse_interface_xml(elem, INTERFACE_COUNTERS)


def _lag_counters(state_doc: etree.Element):
    """
    :param state_doc: /nokia-state:state/lag yang node
    :return: counters for all monitored lags
    """

    for elem in state_doc.xpath("//data/state/lag"):
        yield parse_interface_xml(elem, INTERFACE_COUNTERS)


def _router_interface_counters(state_doc: etree.Element):
    """
    :param state_doc: /nokia-state:state/router/interface yang node
    :return: counters for all monitored "router interfaces"
    """

    for elem in state_doc.xpath("//data/state/router/interface"):
        yield parse_interface_xml(elem, INTERFACE_COUNTERS_ALT)


def interface_counters(raw_counter_docs: dict):
    """

    :param raw_counter_docs: output of a call to get_netconf_interface_info
    :return: iterable of interface counters
    """
    yield from _port_counters(raw_counter_docs["port"])
    yield from _lag_counters(raw_counter_docs["lag"])
    yield from _router_interface_counters(raw_counter_docs["router-interface"])