graphing: build a combined graph's expression per matched object

A combined graph substituted its cross-service fan-out at each metric
name, so a template drawing an arithmetic operation handed the operation
two fan-outs as operands - which the engine refuses. Build what a graph
draws once per matched object and combine the results instead, the way
the legacy combined path resolves a template per object.

An expression is now parsed against the one object it is built for, so a
nested threshold resolves against that object rather than the first
matched one, and a rule is no longer built where no object names it.

CMK-37904

Change-Id: If7801985660de86232fe12facc34606f918255df
JIRA-Ref: CMK-37904

Author: ottermata

packages/cmk-graphing-engine/cmk/graphing_engine/_from_api.py

@@ -37,12 +37,14 @@
 
 
 class QuantityBuilderProtocol(Protocol):
-    def __call__(self, metrics: Sequence[RRDMetric]) -> QuantityProtocol: ...
+    """Combines what a graph draws for each matched object into the one quantity drawing them."""
 
+    def __call__(self, per_object: Sequence[QuantityProtocol]) -> QuantityProtocol: ...
 
-def build_single_quantity(metrics: Sequence[RRDMetric]) -> QuantityProtocol:
-    (metric,) = metrics
-    return metric
+
+def build_single_quantity(per_object: Sequence[QuantityProtocol]) -> QuantityProtocol:
+    (quantity,) = per_object
+    return quantity
 
 
 def drawn_quantity(
@@ -53,21 +55,49 @@ def drawn_quantity(
     return quantity_builder([rrd_metric_of(service, metric_name) for service in services])
 
 
+@dataclass(frozen=True)
+class _ObjectContext:
+    """One of the objects a graph was matched on, to build what it draws for that object."""
+
+    service: Service
+    localizer: Callable[[str], str]
+    registered_metrics: Mapping[str, metrics_v1.Metric]
+
+    def metric(self, metric_name: str) -> RRDMetric:
+        return rrd_metric_of(self.service, metric_name)
+
+
 @dataclass(frozen=True)
 class _ParseContext:
     services: Sequence[Service]
     quantity_builder: QuantityBuilderProtocol
     localizer: Callable[[str], str]
     registered_metrics: Mapping[str, metrics_v1.Metric]
 
-    def drawn(self, metric_name: str) -> QuantityProtocol:
-        return drawn_quantity(metric_name, self.services, self.quantity_builder)
-
-    def scalar(self, metric_name: str) -> RRDMetric:
-        return rrd_metric_of(self.services[0], metric_name)
+    def _of(self, service: Service) -> _ObjectContext:
+        return _ObjectContext(service, self.localizer, self.registered_metrics)
+
+    def single_object(self) -> _ObjectContext | None:
+        # A rule names one object's thresholds; over several matched ones there is none to name them
+        # against, so no rule is built.
+        return self._of(self.services[0]) if len(self.services) == 1 else None
+
+    def object_for(self, bound: ApiQuantity) -> _ObjectContext | None:
+        # A bound naming a metric is read from one object; over several matched ones there is none, so
+        # that end is left to auto-scale. A bound of constants alone is read without one.
+        if any(metric_names_in_quantity(bound)):
+            return self.single_object()
+        return self._of(self.services[0])
+
+    def drawn(self, quantity: ApiQuantity) -> QuantityProtocol:
+        # Built once per matched object and combined, so an operation only ever takes that object's
+        # operands - never a quantity spanning the objects, which cannot be an operand.
+        return self.quantity_builder(
+            [_parse_quantity(quantity, self._of(service)) for service in self.services]
+        )
 
 
-def _curve_display(quantity: ApiQuantity, context: _ParseContext) -> CurveAttributes:
+def _curve_display(quantity: ApiQuantity, context: _ObjectContext) -> CurveAttributes:
     match quantity:
         case str():
             return metric_display_attributes(
@@ -109,39 +139,39 @@ def _curve_display(quantity: ApiQuantity, context: _ParseContext) -> CurveAttrib
             assert_never(quantity)
 
 
-def _parse_quantity(quantity: ApiQuantity, context: _ParseContext) -> QuantityProtocol:
+def _parse_quantity(quantity: ApiQuantity, context: _ObjectContext) -> QuantityProtocol:
     match quantity:
         case str():
-            return context.drawn(quantity)
+            return context.metric(quantity)
         case metrics_v1.Constant():
             return Constant(quantity.value, display=_curve_display(quantity, context))
         case metrics_v2_unstable.LowerWarningOf():
             return ScalarOf(
-                metric=context.scalar(quantity.metric_name),
+                metric=context.metric(quantity.metric_name),
                 scalar_kind=ScalarKind.LOWER_WARNING,
             )
         case metrics_v2_unstable.LowerCriticalOf():
             return ScalarOf(
-                metric=context.scalar(quantity.metric_name),
+                metric=context.metric(quantity.metric_name),
                 scalar_kind=ScalarKind.LOWER_CRITICAL,
             )
         case metrics_v1.WarningOf():
             return ScalarOf(
-                metric=context.scalar(quantity.metric_name), scalar_kind=ScalarKind.WARNING
+                metric=context.metric(quantity.metric_name), scalar_kind=ScalarKind.WARNING
             )
         case metrics_v1.CriticalOf():
             return ScalarOf(
-                metric=context.scalar(quantity.metric_name), scalar_kind=ScalarKind.CRITICAL
+                metric=context.metric(quantity.metric_name), scalar_kind=ScalarKind.CRITICAL
             )
         case metrics_v1.MinimumOf():
             return ScalarOf(
-                metric=context.scalar(quantity.metric_name),
+                metric=context.metric(quantity.metric_name),
                 scalar_kind=ScalarKind.MINIMUM,
                 color=parse_color(quantity.color),
             )
         case metrics_v1.MaximumOf():
             return ScalarOf(
-                metric=context.scalar(quantity.metric_name),
+                metric=context.metric(quantity.metric_name),
                 scalar_kind=ScalarKind.MAXIMUM,
                 color=parse_color(quantity.color),
             )
@@ -174,11 +204,8 @@ def _parse_quantity(quantity: ApiQuantity, context: _ParseContext) -> QuantityPr
 def _parse_bound(bound: int | float | ApiQuantity, context: _ParseContext) -> Bound | None:
     if isinstance(bound, int | float):
         return bound
-    # A bound naming a metric is read from one object; over several matched ones there is none, so
-    # that end is left to auto-scale. A bound of constants alone is read without one.
-    if len(context.services) > 1 and any(metric_names_in_quantity(bound)):
-        return None
-    return _parse_quantity(bound, context)
+    object_context = context.object_for(bound)
+    return None if object_context is None else _parse_quantity(bound, object_context)
 
 
 def _parse_range(
@@ -242,19 +269,26 @@ def _parse_lines(
     *,
     inverse: bool,
 ) -> tuple[Sequence[Stack], Sequence[Line], Sequence[Rule]]:
-    def _curve(q: ApiQuantity) -> Curve:
-        return build_curve(
-            _parse_quantity(q, context), context.localizer, context.registered_metrics
-        )
+    def _curve(quantity: QuantityProtocol) -> Curve:
+        return build_curve(quantity, context.localizer, context.registered_metrics)
 
-    stack_members = [_curve(q) for q in graph.compound_lines if not is_scalar(q)]
+    stack_members = [_curve(context.drawn(q)) for q in graph.compound_lines if not is_scalar(q)]
     stacks = [Stack(members=stack_members, inverse=inverse)] if stack_members else []
-    lines = [Line(curve=_curve(q), inverse=inverse) for q in graph.simple_lines if not is_scalar(q)]
-    rules = [
-        Rule(curve=_curve(q), inverse=inverse)
-        for q in (*graph.compound_lines, *graph.simple_lines)
-        if is_scalar(q)
+    lines = [
+        Line(curve=_curve(context.drawn(q)), inverse=inverse)
+        for q in graph.simple_lines
+        if not is_scalar(q)
     ]
+    single_object = context.single_object()
+    rules: Sequence[Rule] = (
+        ()
+        if single_object is None
+        else [
+            Rule(curve=_curve(_parse_quantity(q, single_object)), inverse=inverse)
+            for q in (*graph.compound_lines, *graph.simple_lines)
+            if is_scalar(q)
+        ]
+    )
     return stacks, lines, rules
 
 

packages/cmk-graphing-engine/cmk/graphing_engine/_matching.py

@@ -159,19 +159,10 @@ def _collect(base: Graph) -> None:
         # work, rather than building every graph and filtering afterwards.
         if graph_name is not None and not _matches_graph_name(base, graph_name):
             return
-        # Rules and predictive lines are single-service concepts; a graph over multiple services
-        # drops them.
+        # A predictive line is a single-service concept; a graph over multiple services has none (nor
+        # rules or a quantity range bound, which the parse already left out).
         if single_service is None:
-            matched_graphs.append(
-                Graph(
-                    name=base.name,
-                    title=base.title,
-                    kind=base.kind,
-                    vertical_range=base.vertical_range,
-                    stacks=base.stacks,
-                    lines=base.lines,
-                )
-            )
+            matched_graphs.append(base)
             return
         graph, predictive_names = _add_predictive_lines(
             base, single_service, available, localizer, registered_metrics
@@ -181,7 +172,7 @@ def _collect(base: Graph) -> None:
 
     def _fallback_rules(name: MetricName) -> Sequence[Rule]:
         if single_service is None:
-            return []
+            return ()
         metric = rrd_metric_of(single_service, name)
         return [
             Rule(

packages/cmk-graphing-engine/tests/test_matching.py

@@ -15,6 +15,7 @@
     build_matched_graphs,
     ConsolidationFunction,
     Constant,
+    Difference,
     evaluate_graphs,
     EvaluatedGraph,
     FetchedData,
@@ -645,10 +646,10 @@ def _rrd_on(service: Service, name: MetricName) -> RRDMetric:
 
 
 class _SumQuantityBuilder:
-    # Stand-in for a real aggregating QuantityBuilderProtocol (e.g. the pro _Aggregation): wraps the per-service
-    # RRDMetrics of one drawn metric in the engine's own Sum.
-    def __call__(self, metrics: Sequence[RRDMetric]) -> QuantityProtocol:
-        return Sum(summands=list(metrics))
+    # Stand-in for a real aggregating QuantityBuilderProtocol (e.g. the pro CombinedAggregation):
+    # wraps what one drawn quantity is per service in the engine's own Sum.
+    def __call__(self, per_object: Sequence[QuantityProtocol]) -> QuantityProtocol:
+        return Sum(summands=list(per_object))
 
 
 def _discover_combined(
@@ -688,6 +689,85 @@ def test_build_matched_graphs_aggregates_a_drawn_metric_across_services() -> Non
     assert [line.curve.value for line in _evaluate(discovered, fetch_data).lines] == [2.0]
 
 
+def test_build_matched_graphs_builds_a_drawn_operation_per_service() -> None:
+    h1, h2 = _services()
+    cpu_user = MetricName("cpu_user")
+    cpu_system = MetricName("cpu_system")
+    plugin = graphs_v1.Graph(
+        name="cpu",
+        title=Title("CPU"),
+        simple_lines=[
+            metrics_v1.Difference(
+                Title("Difference"),
+                metrics_v1.Color.GREEN,
+                minuend="cpu_user",
+                subtrahend="cpu_system",
+            )
+        ],
+    )
+    fetch_data = _FakeRRDFetchData(
+        performance_response={
+            h1: _perf_data(_perf(cpu_user), _perf(cpu_system)),
+            h2: _perf_data(_perf(cpu_user), _perf(cpu_system)),
+        }
+    )
+
+    [discovered] = _discover_combined([plugin], fetch_data=fetch_data)
+
+    # The builder combines one difference per service, rather than the difference taking two
+    # cross-service quantities as its operands - which no operation can have.
+    [line] = discovered.lines
+    quantity = line.curve.quantity
+    assert isinstance(quantity, Sum)
+    operands = []
+    for summand in quantity.summands:
+        assert isinstance(summand, Difference)
+        operands.append((summand.minuend, summand.subtrahend))
+    assert operands == [
+        (_rrd_on(h1, cpu_user), _rrd_on(h1, cpu_system)),
+        (_rrd_on(h2, cpu_user), _rrd_on(h2, cpu_system)),
+    ]
+    assert [line.curve.value for line in _evaluate(discovered, fetch_data).lines] == [0.0]
+
+
+def test_build_matched_graphs_resolves_a_nested_scalar_per_service() -> None:
+    h1, h2 = _services()
+    cpu_user = MetricName("cpu_user")
+    plugin = graphs_v1.Graph(
+        name="cpu",
+        title=Title("CPU"),
+        simple_lines=[
+            metrics_v1.Difference(
+                Title("Headroom"),
+                metrics_v1.Color.GREEN,
+                minuend=metrics_v1.MaximumOf("cpu_user", metrics_v1.Color.GREEN),
+                subtrahend="cpu_user",
+            )
+        ],
+    )
+    fetch_data = _FakeRRDFetchData(
+        performance_response={
+            h1: _perf_data(_perf(cpu_user)),
+            h2: _perf_data(_perf(cpu_user)),
+        }
+    )
+
+    [discovered] = _discover_combined([plugin], fetch_data=fetch_data)
+
+    # A threshold names the metric of the service its expression was built for, not of the first
+    # matched one.
+    [line] = discovered.lines
+    quantity = line.curve.quantity
+    assert isinstance(quantity, Sum)
+    thresholds = []
+    for summand in quantity.summands:
+        assert isinstance(summand, Difference)
+        assert isinstance(summand.minuend, ScalarOf)
+        assert summand.minuend.scalar_kind is ScalarKind.MAXIMUM
+        thresholds.append(summand.minuend.metric)
+    assert thresholds == [_rrd_on(h1, cpu_user), _rrd_on(h2, cpu_user)]
+
+
 def test_build_matched_graphs_drops_rules_and_predictive_for_multiple_services() -> None:
     h1, h2 = _services()
     cpu_user = MetricName("cpu_user")