Add functionality for making DHCP graphs for a set of prefixes

The function fetch_graph_urls_for_prefixes takes a set of prefixes
(e.g. obtained from a models.manage.Prefix or from a models.manage.Vlan) and
returns one url per DHCP graph from Graphite related to one or more of the
prefixes. Each url returns a JSON with graph data.

Author: jorund1

python/nav/dhcpstats/common.py

@@ -18,11 +18,20 @@
 NAV that wants to make use of DHCP stats.
 """
 
+from collections import defaultdict
 from dataclasses import dataclass
-from typing import Iterable, Literal, Optional, Union
+from typing import Any, Iterable, Literal, Optional, Union
 
 import IPy
 
+from nav.metrics.graphs import (
+    aliased_series,
+    diffed_series,
+    json_graph_url,
+    nonempty_series,
+    summed_series,
+)
+from nav.metrics.names import get_expanded_nodes, safe_name
 from nav.metrics.templates import metric_path_for_dhcp
 
 
@@ -31,6 +40,133 @@
 GraphiteMetric = tuple[str, tuple[float, int]]
 
 
+def fetch_graph_urls_for_prefixes(prefixes: list[IPy.IP]) -> list[str]:
+    """
+    Takes a list of IP prefixes, queries Graphite for DHCP stat paths, and
+    returns a list of Graphite graph URLs; one URL for each group of DHCP stat
+    paths in Graphite where at least one path in the group represents a range of
+    IP addresses that intersect with one or more of the given prefixes.
+
+    Each returned Graphite URL points to JSON graph data and can be supplied
+    directly to NAV's JavaScript RickshawGraph function.
+    """
+    if not prefixes:
+        return []
+
+    all_paths = fetch_paths_from_graphite()
+    grouped_paths = group_paths(all_paths)
+    filtered_grouped_paths = drop_groups_not_in_prefixes(grouped_paths, prefixes)
+
+    graph_urls: list[str] = []
+    for paths_of_same_group in sorted(filtered_grouped_paths):
+        paths_of_same_group = sorted(paths_of_same_group)
+        graph_lines = []
+        for path in paths_of_same_group:
+            assigned_addresses = aliased_series(
+                nonempty_series(
+                    path.to_graphite_path("assigned"),
+                ),
+                name=f"Assigned addresses in {path.first_ip} - {path.last_ip}",
+                renderer="area",
+            )
+            graph_lines.append(assigned_addresses)
+
+        assert len(paths_of_same_group) > 0
+        path = paths_of_same_group[0]  # Just select an arbitrary path instance in the group
+        unassigned_addresses = aliased_series(
+            diffed_series(
+                summed_series(
+                    nonempty_series(
+                        path.to_graphite_path("total", wildcard_for_group=True)
+                    ),
+                ),
+                summed_series(
+                    nonempty_series(
+                        path.to_graphite_path("assigned", wildcard_for_group=True)
+                    ),
+                ),
+            ),
+            name="Unassigned addresses",
+            renderer="area",
+            color="#d9d9d9",
+        )
+        graph_lines.append(unassigned_addresses)
+
+        total_addresses = aliased_series(
+            summed_series(
+                nonempty_series(
+                    path.to_graphite_path("total", wildcard_for_group=True)
+                ),
+            ),
+            name="Total addresses",
+            color="#707070",
+        )
+        graph_lines.append(total_addresses)
+
+        type_human = path.allocation_type + "s"
+        title = f"DHCP {type_human} in '{path.group_name}' on server '{path.server_name}'"
+        graph_urls.append(json_graph_url(*graph_lines, title=title))
+    return graph_urls
+
+
+def fetch_paths_from_graphite():
+    """
+    Fetches and returns all unique DHCP stat paths in Graphite when their
+    trailing metric_name path segment has been removed.
+    """
+    wildcard = metric_path_for_dhcp(
+        ip_version=safe_name("{4,6}"),
+        server_name=safe_name("*"),
+        allocation_type=safe_name("{range,pool,subnet}"),
+        group_name_source=safe_name("{custom_groups,special_groups}"),
+        group_name=safe_name("*"),
+        first_ip=safe_name("*"),
+        last_ip=safe_name("*"),
+        metric_name="total",
+    )
+    graphite_paths = get_expanded_nodes(wildcard)
+
+    native_paths: list[DhcpPath] = []
+    for graphite_path in graphite_paths:
+        try:
+            native_path = DhcpPath.from_graphite_path(graphite_path)
+        except ValueError:
+            pass
+        else:
+            native_paths.append(native_path)
+    return native_paths
+
+
+def group_paths(paths: list["DhcpPath"]):
+    """
+    Takes a list of DhcpPath instances and partitions it into multiple lists,
+    such that two DhcpPath instances belong to the same list if and only if they
+    have the same group data.
+    """
+    grouped_paths: dict[Any, list[DhcpPath]] = defaultdict(list)
+    for path in paths:
+        group_total = path.to_graphite_path("total", wildcard_for_group=True)
+        group_assigned = path.to_graphite_path("assigned", wildcard_for_group=True)
+        grouped_paths[(group_total, group_assigned)].append(path)
+    return list(grouped_paths.values())
+
+
+def drop_groups_not_in_prefixes(grouped_paths: list[list["DhcpPath"]], prefixes: list[IPy.IP]):
+    """
+    Takes a list of grouped DhcpPath instances, and return only the groups that
+    has at least one DhcpPath instance that intersect with one or more of the
+    given prefixes.
+    """
+    grouped_paths_to_keep: list[list[DhcpPath]] = []
+    for paths_of_same_group in grouped_paths:
+        if any(
+            path.intersects(prefixes)
+            for path in paths_of_same_group
+        ):
+            grouped_paths_to_keep.append(paths_of_same_group)
+    return grouped_paths_to_keep
+
+
 @dataclass(frozen=True, order=True)
 class DhcpPath:
     """

python/nav/metrics/names.py

@@ -77,6 +77,23 @@ def get_all_leaves_below(top, ignored=None):
     return list(itertools.chain(*paths))
 
 
+def get_expanded_nodes(path):
+    """
+    Expands any wildcard in path and returns a list of all matching paths.
+
+    :param path: A search string, e.g. "nav.{a,b}.*.*"
+    :returns: A list of expanded metric paths,
+              e.g. ["nav.a.1.1", "nav.a.2.1", "nav.b.1.1", "nav.b.1.2"]
+    """
+    data = raw_metric_query(path, operation="expand")
+    if not isinstance(data, dict):
+        return []
+    result = data.get("results", [])
+    if not isinstance(result, list):
+        return []
+    return result
+
+
 def get_metric_leaf_children(path):
     """Returns a list of available graphite leaf nodes just below path.
 
@@ -132,15 +149,18 @@ def nodewalk(top, ignored=None):
             yield x
 
 
-def raw_metric_query(query):
+def raw_metric_query(query, operation="find"):
     """Runs a query for metric information against Graphite's REST API.
 
     :param query: A search string, e.g. "nav.devices.some-gw_example_org.*"
-    :returns: A list of matching metrics, each represented by a dict.
+    :param operation: One of "find" or "expand",
+                      see https://graphite.readthedocs.io/en/1.1.10/metrics_api.html
+    :returns: The JSON response from Graphite, or an empty list if response
+              could not be decoded.
 
     """
     base = CONFIG.get("graphiteweb", "base")
-    url = urljoin(base, "/metrics/find")
+    url = urljoin(base, "/metrics/" + operation)
     query = urlencode({'query': query})
     url = "%s?%s" % (url, query)
 

python/nav/web/static/js/src/plugins/rickshaw_graph.js

@@ -69,6 +69,9 @@ define([
             serie.key = name;
             serie.name = RickshawUtils.filterFunctionCalls(name);
             serie.renderer = meta.renderer ? meta.renderer : 'line';
+            if (meta.color !== undefined) {
+                serie.color = meta.color;
+            }
 
             // If this is a nav-metric, typically very long, display only the last two "parts"
             if (serie.name.substr(0, 4) === 'nav.') {

tests/unittests/metrics/names_test.py

@@ -1,6 +1,7 @@
 import pytest
 from unittest import TestCase
-from nav.metrics.names import join_series, escape_metric_name
+from unittest.mock import patch
+from nav.metrics.names import join_series, escape_metric_name, get_expanded_nodes, safe_name
 
 
 class MetricNamingTests(TestCase):
@@ -15,6 +16,28 @@ def test_join_single_series_should_return_same(self):
         self.assertEqual(join_series([series]), series)
 
 
+class TestGetExpandedNodes:
+    def test_when_valid_response_should_return_results(self):
+        raw_response = {
+            "results": [
+                "nav.foo.1",
+                "nav.foo.2",
+                "nav.foo.3",
+                "nav.bar.baz",
+            ]
+        }
+
+        with patch("nav.metrics.names.raw_metric_query", return_value=raw_response):
+            assert get_expanded_nodes("whatever.path") == raw_response["results"]
+
+    @pytest.mark.parametrize(
+        "raw_response", [[], {}, "foo", "", {"results": "foo"}]
+    )
+    def test_when_invalid_response_should_return_empty_list(self, raw_response):
+        with patch("nav.metrics.names.raw_metric_query", return_value=raw_response):
+            assert get_expanded_nodes("whatever.path") == []
+
+
 @pytest.mark.parametrize(
     "test_input,expected",
     [