Merge pull request #6440 from opsmill/dga-20250511-jinja2-hash

Ensure Jinja2 computed attribute recalculate only when the template changes

Author: ogenstad

.gitignore

@@ -27,6 +27,7 @@ support_logs_*
 docs/build
 
 storage/*
+sandbox/*
 .coverage.*
 python_sdk/dist/*
 .benchmarks/*

backend/infrahub/computed_attribute/models.py

@@ -118,6 +118,10 @@ class PythonTransformTarget:
 class ComputedAttrJinja2TriggerDefinition(TriggerBranchDefinition):
     type: TriggerType = TriggerType.COMPUTED_ATTR_JINJA2
     computed_attribute: ComputedAttributeTarget
+    template_hash: str
+
+    def get_description(self) -> str:
+        return f"{super().get_description()} | hash:{self.template_hash}"
 
     @classmethod
     def from_computed_attribute(
@@ -139,6 +143,14 @@ def from_computed_attribute(
             # node creation events if the attribute is optional.
             event_trigger.events.add(NodeCreatedEvent.event_name)
 
+        if (
+            computed_attribute.attribute.computed_attribute
+            and computed_attribute.attribute.computed_attribute.jinja2_template is None
+        ) or not computed_attribute.attribute.computed_attribute:
+            raise ValueError("Jinja2 template is required for computed attribute")
+
+        template_hash = computed_attribute.attribute.computed_attribute.get_hash()
+
         event_trigger.match = {"infrahub.node.kind": trigger_node.kind}
         if branches_out_of_scope:
             event_trigger.match["infrahub.branch.name"] = [f"!{branch}" for branch in branches_out_of_scope]
@@ -177,6 +189,7 @@ def from_computed_attribute(
 
         definition = cls(
             name=f"{computed_attribute.key_name}{NAME_SEPARATOR}kind{NAME_SEPARATOR}{trigger_node.kind}",
+            template_hash=template_hash,
             branch=branch,
             computed_attribute=computed_attribute,
             trigger=event_trigger,

backend/infrahub/computed_attribute/tasks.py

@@ -14,7 +14,7 @@
 from infrahub.events import BranchDeletedEvent
 from infrahub.git.repository import get_initialized_repo
 from infrahub.services import InfrahubServices  # noqa: TC001  needed for prefect flow
-from infrahub.trigger.models import TriggerType
+from infrahub.trigger.models import TriggerSetupReport, TriggerType
 from infrahub.trigger.setup import setup_triggers
 from infrahub.workflows.catalogue import (
     COMPUTED_ATTRIBUTE_PROCESS_JINJA2,
@@ -25,7 +25,11 @@
 from infrahub.workflows.utils import add_tags, wait_for_schema_to_converge
 
 from .gather import gather_trigger_computed_attribute_jinja2, gather_trigger_computed_attribute_python
-from .models import ComputedAttrJinja2GraphQL, ComputedAttrJinja2GraphQLResponse, PythonTransformTarget
+from .models import (
+    ComputedAttrJinja2GraphQL,
+    ComputedAttrJinja2GraphQLResponse,
+    PythonTransformTarget,
+)
 
 if TYPE_CHECKING:
     from infrahub.core.schema.computed_attribute import ComputedAttribute
@@ -159,10 +163,10 @@ async def trigger_update_python_computed_attributes(
 
 
 @flow(
-    name="process_computed_attribute_value_jinja2",
-    flow_run_name="Update value for computed attribute {attribute_name}",
+    name="computed-attribute-jinja2-update-value",
+    flow_run_name="Update value for computed attribute {node_kind}:{attribute_name}",
 )
-async def update_computed_attribute_value_jinja2(
+async def computed_attribute_jinja2_update_value(
     branch_name: str,
     obj: ComputedAttrJinja2GraphQLResponse,
     node_kind: str,
@@ -246,7 +250,7 @@ async def process_jinja2(
         batch = await service.client.create_batch()
         for node in found:
             batch.add(
-                task=update_computed_attribute_value_jinja2,
+                task=computed_attribute_jinja2_update_value,
                 branch_name=branch_name,
                 obj=node,
                 node_kind=node_schema.kind,
@@ -302,36 +306,33 @@ async def computed_attribute_setup_jinja2(
 
         triggers = await gather_trigger_computed_attribute_jinja2()
 
-        # Since we can have multiple trigger per NodeKind
-        # we need to extract the list of unique node that should be processed
-        # also
-        # Because the automation in Prefect doesn't capture all information about the computed attribute
-        # we can't tell right now if a given computed attribute has changed and need to be updated
-        unique_nodes: set[tuple[str, str, str]] = {
-            (trigger.branch, trigger.computed_attribute.kind, trigger.computed_attribute.attribute.name)
-            for trigger in triggers
-        }
-        for branch, kind, attribute_name in unique_nodes:
-            if event_name != BranchDeletedEvent.event_name and branch == branch_name:
-                await service.workflow.submit_workflow(
-                    workflow=TRIGGER_UPDATE_JINJA_COMPUTED_ATTRIBUTES,
-                    context=context,
-                    parameters={
-                        "branch_name": branch,
-                        "computed_attribute_name": attribute_name,
-                        "computed_attribute_kind": kind,
-                    },
-                )
-
         # Configure all ComputedAttrJinja2Trigger in Prefect
         async with get_client(sync_client=False) as prefect_client:
-            await setup_triggers(
+            report: TriggerSetupReport = await setup_triggers(
                 client=prefect_client,
                 triggers=triggers,
                 trigger_type=TriggerType.COMPUTED_ATTR_JINJA2,
                 force_update=False,
             )  # type: ignore[misc]
 
+            # Since we can have multiple trigger per NodeKind
+            # we need to extract the list of unique node that should be processed
+            unique_nodes: set[tuple[str, str, str]] = {
+                (trigger.branch, trigger.computed_attribute.kind, trigger.computed_attribute.attribute.name)  # type: ignore[attr-defined]
+                for trigger in report.updated + report.created
+            }
+            for branch, kind, attribute_name in unique_nodes:
+                if event_name != BranchDeletedEvent.event_name and branch == branch_name:
+                    await service.workflow.submit_workflow(
+                        workflow=TRIGGER_UPDATE_JINJA_COMPUTED_ATTRIBUTES,
+                        context=context,
+                        parameters={
+                            "branch_name": branch,
+                            "computed_attribute_name": attribute_name,
+                            "computed_attribute_kind": kind,
+                        },
+                    )
+
         log.info(f"{len(triggers)} Computed Attribute for Jinja2 automation configuration completed")
 
 

backend/infrahub/trigger/setup.py

@@ -39,9 +39,9 @@ async def setup_triggers(
     report = TriggerSetupReport()
 
     if trigger_type:
-        log.info(f"Setting up triggers of type {trigger_type.value}")
+        log.debug(f"Setting up triggers of type {trigger_type.value}")
     else:
-        log.info("Setting up all triggers")
+        log.debug("Setting up all triggers")
 
     # -------------------------------------------------------------
     # Retrieve existing Deployments and Automation from the server
@@ -112,4 +112,15 @@ async def setup_triggers(
         await client.delete_automation(automation_id=existing_automation.id)
         log.info(f"{item_to_delete} Deleted")
 
+    if trigger_type:
+        log.info(
+            f"Processed triggers of type {trigger_type.value}: "
+            f"{len(report.created)} created, {len(report.updated)} updated, {len(report.unchanged)} unchanged, {len(report.deleted)} deleted"
+        )
+    else:
+        log.info(
+            f"Processed all triggers: "
+            f"{len(report.created)} created, {len(report.updated)} updated, {len(report.unchanged)} unchanged, {len(report.deleted)} deleted"
+        )
+
     return report

backend/infrahub/workflows/catalogue.py

@@ -251,6 +251,14 @@
     tags=[WorkflowTag.DATABASE_CHANGE],
 )
 
+COMPUTED_ATTRIBUTE_JINJA2_UPDATE_VALUE = WorkflowDefinition(
+    name="computed-attribute-jinja2-update-value",
+    type=WorkflowType.CORE,
+    module="infrahub.computed_attribute.tasks",
+    function="computed_attribute_jinja2_update_value",
+    tags=[WorkflowTag.DATABASE_CHANGE],
+)
+
 TRIGGER_UPDATE_JINJA_COMPUTED_ATTRIBUTES = WorkflowDefinition(
     name="trigger_update_jinja2_computed_attributes",
     type=WorkflowType.CORE,
@@ -443,6 +451,7 @@
     BRANCH_MERGE_POST_PROCESS,
     BRANCH_REBASE,
     BRANCH_VALIDATE,
+    COMPUTED_ATTRIBUTE_JINJA2_UPDATE_VALUE,
     COMPUTED_ATTRIBUTE_PROCESS_JINJA2,
     COMPUTED_ATTRIBUTE_PROCESS_TRANSFORM,
     COMPUTED_ATTRIBUTE_SETUP_JINJA2,

backend/tests/integration_docker/test_computed_attributes.py

@@ -4,23 +4,36 @@
 import pytest
 import yaml
 from infrahub_sdk import InfrahubClient
+from infrahub_sdk.task.models import TaskFilter, TaskState
 from infrahub_sdk.testing.docker import TestInfrahubDockerClient
 from infrahub_sdk.testing.repository import GitRepo
 
+from infrahub.workflows.catalogue import COMPUTED_ATTRIBUTE_JINJA2_UPDATE_VALUE
+
 CURRENT_DIRECTORY = Path(__file__).parent.resolve()
 
 
+async def wait_for_all_tasks_to_be_completed(client: InfrahubClient) -> None:
+    while (  # noqa: ASYNC110
+        await client.task.count(filters=TaskFilter(state=[TaskState.PENDING, TaskState.RUNNING, TaskState.SCHEDULED]))
+        > 0
+    ):
+        await sleep(1)
+
+
 class TestComputedAttributes(TestInfrahubDockerClient):
     @pytest.fixture(scope="class")
     def infrahub_version(self) -> str:
         return "local"
 
-    async def test_load_schema(self, client: InfrahubClient) -> None:
-        """Prepare the schema"""
+    @pytest.fixture(scope="class")
+    def schema_computed_tshirt(self) -> dict:
         with Path(CURRENT_DIRECTORY / "test_files/computed_tshirt.yml").open(encoding="utf-8") as file:  # noqa: ASYNC230
-            computed_tshirt = yaml.safe_load(file.read())
+            return yaml.safe_load(file.read())
 
-        tshirt_schema = await client.schema.load(schemas=[computed_tshirt], wait_until_converged=True)
+    async def test_load_schema(self, client: InfrahubClient, schema_computed_tshirt: dict) -> None:
+        """Prepare the schema"""
+        tshirt_schema = await client.schema.load(schemas=[schema_computed_tshirt], wait_until_converged=True)
         assert tshirt_schema.schema_updated
         # Validate that the schema is in sync after loading the device and interface schema
         assert await client.schema.in_sync()
@@ -190,3 +203,47 @@ async def test_transform_based_computed_attribute(self, client: InfrahubClient,
             await sleep(1)
 
         assert sth_router_1_swe.name.value == swe_name_router_1
+
+    async def test_update_schema_not_related_to_computed_attribute(
+        self, client: InfrahubClient, schema_computed_tshirt: dict
+    ) -> None:
+        nbr_task_before = await client.task.count(
+            filters=TaskFilter(workflow=[COMPUTED_ATTRIBUTE_JINJA2_UPDATE_VALUE.name])
+        )
+
+        # Update schema LocationSite with a change that IS NOT related to the computed attribute
+        # The computed attribute of type Jinja2 should not be updated neither on the sites nor on the continents
+        schema_computed_tshirt["nodes"][4]["description"] = "New Description that will trigger a new schema"
+        tshirt_schema = await client.schema.load(schemas=[schema_computed_tshirt], wait_until_converged=True)
+        assert tshirt_schema.schema_updated
+
+        # Validate that the schema is in sync after loading the device and interface schema
+        assert await client.schema.in_sync()
+
+        await sleep(1)
+        await wait_for_all_tasks_to_be_completed(client)
+
+        nbr_task_after_not_related = await client.task.count(
+            filters=TaskFilter(workflow=[COMPUTED_ATTRIBUTE_JINJA2_UPDATE_VALUE.name])
+        )
+        assert nbr_task_after_not_related == nbr_task_before
+
+        # Update schema LocationSite with a change that IS related to the computed attribute
+        # The computed attribute of type Jinja2 should be updated
+        schema_computed_tshirt["nodes"][4]["attributes"][3]["computed_attribute"]["jinja2_template"] = (
+            "WELCOME TO {{ name__value }}!"
+        )
+        tshirt_schema = await client.schema.load(schemas=[schema_computed_tshirt], wait_until_converged=True)
+        assert tshirt_schema.schema_updated
+
+        # Validate that the schema is in sync after loading the device and interface schema
+        assert await client.schema.in_sync()
+
+        await sleep(1)
+        await wait_for_all_tasks_to_be_completed(client)
+
+        # The computed attribute of type Jinja2 should be updated on the sites but NOT on the continents
+        nbr_task_after_related = await client.task.count(
+            filters=TaskFilter(workflow=[COMPUTED_ATTRIBUTE_JINJA2_UPDATE_VALUE.name])
+        )
+        assert nbr_task_after_related == nbr_task_after_not_related + 2

changelog/+comp_attr_jinja_hash.added.md

@@ -0,0 +1 @@
+Computed Attribute of kind Jinja will only be recalculated during a schema update if the template itself has been updated.