Add support for ARP/ND

Author: wdesmedt

nornir_srl/connections/srlinux.py

@@ -3,6 +3,7 @@
 import json
 import re
 import copy
+import datetime
 
 from natsort import natsorted
 import jmespath
@@ -432,6 +433,13 @@ def get_mac_table(self, network_instance: Optional[str] = "*") -> Dict[str, Any]
                         Dest:destination, Type:type}}',
             "datatype": "state",
         }
+        if (
+            not "bridged"
+            in self.get(paths=["/system/features"], datatype="state")[0][
+                "system/features"
+            ]
+        ):
+            return {"mac_table": []}
         resp = self.get(
             paths=[path_spec.get("path", "")], datatype=path_spec["datatype"]
         )
@@ -653,6 +661,64 @@ def set_es_peers(resp):
         res = jmespath.search(path_spec["jmespath"], resp[0])
         return {"es": res}
 
+    def get_arp(self) -> Dict[str, Any]:
+        path_spec = {
+            "path": f"/interface[name=*]/subinterface[index=*]/ipv4/arp/neighbor",
+            "jmespath": '"interface"[*].subinterface[].{interface:"_subitf", entries:ipv4.arp.neighbor[].{IPv4:"ipv4-address",MAC:"link-layer-address",Type:origin,expiry:"_rel_expiry" }}',
+            "datatype": "state",
+        }
+        resp = self.get(
+            paths=[path_spec.get("path", "")], datatype=path_spec["datatype"]
+        )
+        for itf in resp[0].get("interface", []):
+            for subitf in itf.get("subinterface", []):
+                subitf["_subitf"] = f"{itf['name']}.{subitf['index']}"
+                for arp_entry in (
+                    subitf.get("ipv4", {}).get("arp", {}).get("neighbor", [])
+                ):
+                    try:
+                        ts = datetime.datetime.strptime(
+                            arp_entry["expiration-time"], "%Y-%m-%dT%H:%M:%S.%fZ"
+                        )
+                        arp_entry["_rel_expiry"] = (
+                            str(ts - datetime.datetime.now()).split(".")[0] + "s"
+                        )
+                    except:
+                        arp_entry["_rel_expiry"] = "-"
+
+        res = jmespath.search(path_spec["jmespath"], resp[0])
+        return {"arp": res}
+
+    def get_nd(self) -> Dict[str, Any]:
+        path_spec = {
+            "path": f"/interface[name=*]/subinterface[index=*]/ipv6/neighbor-discovery/neighbor",
+            "jmespath": '"interface"[*].subinterface[].{interface:"_subitf", entries:ipv6."neighbor-discovery".neighbor[].{IPv6:"ipv6-address",MAC:"link-layer-address",Type:origin,next_state:"_rel_expiry" }}',
+            "datatype": "state",
+        }
+        resp = self.get(
+            paths=[path_spec.get("path", "")], datatype=path_spec["datatype"]
+        )
+        for itf in resp[0].get("interface", []):
+            for subitf in itf.get("subinterface", []):
+                subitf["_subitf"] = f"{itf['name']}.{subitf['index']}"
+                for nd_entry in (
+                    subitf.get("ipv6", {})
+                    .get("neighbor-discovery", {})
+                    .get("neighbor", [])
+                ):
+                    try:
+                        ts = datetime.datetime.strptime(
+                            nd_entry["next-state-time"], "%Y-%m-%dT%H:%M:%S.%fZ"
+                        )
+                        nd_entry["_rel_expiry"] = (
+                            str(ts - datetime.datetime.now()).split(".")[0] + "s"
+                        )
+                    except:
+                        nd_entry["_rel_expiry"] = "-"
+
+        res = jmespath.search(path_spec["jmespath"], resp[0])
+        return {"nd": res}
+
     def get(
         self,
         paths: List[str],

nornir_srl/fsc.py

@@ -152,7 +152,7 @@ def pass_filter(row, filter):
                 if len(col_names) == 0:
                     col_names = get_fields(l)
                     for col in col_names:
-                        table.add_column(col, no_wrap=True)
+                        table.add_column(col, no_wrap=False)
                 common = {
                     x: y
                     for x, y in l.items()
@@ -727,5 +727,65 @@ def _es(task: Task) -> Result:
     )
 
 
+@cli.command()
+@click.pass_context
+@click.option(
+    "--field-filter",
+    "-f",
+    multiple=True,
+    help='filter fields with <field-name>=<glob-pattern>, e.g. -f name=ge-0/0/0 -f admin_state="ena*". Fieldnames correspond to column names of a report',
+)
+def arp(ctx: Context, field_filter: Optional[List] = None):
+    """Displays ARP table"""
+
+    def _arp(task: Task) -> Result:
+        device = task.host.get_connection(CONNECTION_NAME, task.nornir.config)
+        return Result(host=task.host, result=device.get_arp())
+
+    f_filter = (
+        {k: v for k, v in [f.split("=") for f in field_filter]} if field_filter else {}
+    )
+
+    result = ctx.obj["target"].run(task=_arp, name="arp", raise_on_error=False)
+    print_report(
+        result=result,
+        name="ARP table",
+        failed_hosts=result.failed_hosts,
+        box_type=ctx.obj["box_type"],
+        f_filter=f_filter,
+        i_filter=ctx.obj["i_filter"],
+    )
+
+
+@cli.command()
+@click.pass_context
+@click.option(
+    "--field-filter",
+    "-f",
+    multiple=True,
+    help='filter fields with <field-name>=<glob-pattern>, e.g. -f name=ge-0/0/0 -f admin_state="ena*". Fieldnames correspond to column names of a report',
+)
+def nd(ctx: Context, field_filter: Optional[List] = None):
+    """Displays IPv6 Neighbors"""
+
+    def _nd(task: Task) -> Result:
+        device = task.host.get_connection(CONNECTION_NAME, task.nornir.config)
+        return Result(host=task.host, result=device.get_nd())
+
+    f_filter = (
+        {k: v for k, v in [f.split("=") for f in field_filter]} if field_filter else {}
+    )
+
+    result = ctx.obj["target"].run(task=_nd, name="nd", raise_on_error=False)
+    print_report(
+        result=result,
+        name="IPv6 Neighbors",
+        failed_hosts=result.failed_hosts,
+        box_type=ctx.obj["box_type"],
+        f_filter=f_filter,
+        i_filter=ctx.obj["i_filter"],
+    )
+
+
 if __name__ == "__main__":
     cli(obj={})