1
0
Fork 1
mirror of https://github.com/thatmattlove/hyperglass.git synced 2026-04-17 21:38:27 +00:00
thatmattlove-hyperglass/hyperglass/plugins/_builtin/bgp_route_frr.py
Tan Siewert 934e8f3146
bgp_route_frr: skip empty route query results
In case a route is not present in the RIB, FRR returns an empty JSON
object:

```
$ vtysh -c "show bgp ipv4 unicast 1.2.3.4 json"
{}
```

Skip these empty objects to prevent the parser from failing.

Signed-off-by: Tan Siewert <tan@siewert.io>
2025-10-04 21:35:17 +02:00

89 lines
2.6 KiB
Python

"""Parse FRR JSON Response to Structured Data."""
# Standard Library
import json
import typing as t
# Third Party
from pydantic import PrivateAttr, ValidationError
# Project
from hyperglass.log import log
from hyperglass.exceptions.private import ParsingError
from hyperglass.models.parsing.frr import FRRBGPTable
# Local
from .._output import OutputPlugin
if t.TYPE_CHECKING:
# Project
from hyperglass.models.data import OutputDataModel
from hyperglass.models.api.query import Query
# Local
from .._output import OutputType
def parse_frr(output: t.Sequence[str]) -> "OutputDataModel":
"""Parse a FRR BGP JSON response."""
result = None
_log = log.bind(plugin=BGPRoutePluginFrr.__name__)
for response in output:
try:
parsed: t.Dict = json.loads(response)
_log.debug("Pre-parsed data", data=parsed)
# If empty (i.e. no route found), skip
if not parsed:
continue
validated = FRRBGPTable(**parsed)
bgp_table = validated.bgp_table()
if result is None:
result = bgp_table
else:
result += bgp_table
except json.JSONDecodeError as err:
_log.bind(error=str(err)).critical("Failed to decode JSON")
raise ParsingError("Error parsing response data") from err
except KeyError as err:
_log.bind(key=str(err)).critical("Missing required key in response")
raise ParsingError("Error parsing response data") from err
except IndexError as err:
_log.critical(err)
raise ParsingError("Error parsing response data") from err
except ValidationError as err:
_log.critical(err)
raise ParsingError(err.errors()) from err
return result
class BGPRoutePluginFrr(OutputPlugin):
"""Coerce a FRR route table in JSON format to a standard BGP Table structure."""
_hyperglass_builtin: bool = PrivateAttr(True)
platforms: t.Sequence[str] = ("frr",)
directives: t.Sequence[str] = ("__hyperglass_frr_bgp_route_table__",)
def process(self, *, output: "OutputType", query: "Query") -> "OutputType":
"""Parse FRR response if data is a string (and is therefore unparsed)."""
should_process = all(
(
isinstance(output, (list, tuple)),
query.device.platform in self.platforms,
query.device.structured_output is True,
query.device.has_directives(*self.directives),
)
)
if should_process:
return parse_frr(output)
return output